at-container-registry/pkg/appview/jetstream/backfill_batch_test.go

package jetstream

import (
	"context"
	"encoding/json"
	"testing"
	"time"

	"atcr.io/pkg/atproto"
	"github.com/bluesky-social/indigo/atproto/identity"
	"github.com/bluesky-social/indigo/atproto/syntax"
)

// countingDirectory wraps fakeDirectory and counts LookupDID calls per DID,
// so tests can assert verification results are memoized within a batch.
type countingDirectory struct {
	fakeDirectory
	lookups map[string]int
}

func (d *countingDirectory) LookupDID(ctx context.Context, did syntax.DID) (*identity.Identity, error) {
	d.lookups[did.String()]++
	return d.fakeDirectory.LookupDID(ctx, did)
}

func TestBatchCaptains_VerifiesHoldService(t *testing.T) {
	db := setupTestDB(t)
	defer db.Close()

	execStatements(t, db, `
		CREATE TABLE hold_captain_records (
			hold_did TEXT PRIMARY KEY,
			owner_did TEXT NOT NULL,
			public BOOLEAN NOT NULL,
			allow_all_crew BOOLEAN NOT NULL,
			deployed_at TEXT,
			region TEXT,
			successor TEXT,
			updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
		);
	`)

	realHold := "did:web:realhold.example.com"
	notAHold := "did:plc:notahold"
	unresolvable := "did:plc:unresolvable"

	dir := &countingDirectory{
		fakeDirectory: fakeDirectory{byDID: map[string]*identity.Identity{
			realHold: holdIdentity(realHold, "https://realhold.example.com"),
			notAHold: {
				DID: syntax.DID(notAHold),
				Services: map[string]identity.ServiceEndpoint{
					"atproto_pds": {Type: "AtprotoPersonalDataServer", URL: "https://pds.example.com"},
				},
			},
		}},
		lookups: map[string]int{},
	}
	atproto.SetDirectory(dir)
	defer atproto.SetDirectory(nil)

	worker := &BackfillWorker{db: db}

	captainValue, _ := json.Marshal(map[string]any{
		"$type":              "io.atcr.hold.captain",
		"owner":              "did:plc:owner123",
		"public":             true,
		"allowAllCrew":       true,
		"enableBlueskyPosts": false,
		"deployedAt":         time.Now().Format(time.RFC3339),
	})
	captainRecord := func(holdDID string) atproto.Record {
		return atproto.Record{
			URI:   "at://" + holdDID + "/io.atcr.hold.captain/self",
			Value: captainValue,
		}
	}

	records := []atproto.Record{
		captainRecord(realHold),
		captainRecord(realHold), // duplicate DID: exercises the verification memo
		captainRecord(notAHold),
		captainRecord(unresolvable),
	}

	count, err := worker.batchCaptains(context.Background(), realHold, records)
	if err != nil {
		t.Fatalf("batchCaptains failed: %v", err)
	}
	if count != 2 {
		t.Errorf("batchCaptains processed %d records, want 2 (both from the real hold)", count)
	}

	rows, err := db.Query(`SELECT hold_did FROM hold_captain_records`)
	if err != nil {
		t.Fatalf("Failed to query captain records: %v", err)
	}
	defer rows.Close()

	var cached []string
	for rows.Next() {
		var did string
		if err := rows.Scan(&did); err != nil {
			t.Fatalf("Failed to scan captain record: %v", err)
		}
		cached = append(cached, did)
	}
	if len(cached) != 1 || cached[0] != realHold {
		t.Errorf("cached captain records = %v, want only %q", cached, realHold)
	}

	// Each DID should be verified exactly once per batch, regardless of how
	// many of its records appear.
	for _, did := range []string{realHold, notAHold, unresolvable} {
		if got := dir.lookups[did]; got != 1 {
			t.Errorf("LookupDID(%s) called %d times, want 1 (memoized)", did, got)
		}
	}
}