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

package jetstream

import (
	"context"
	"database/sql"
	"encoding/json"
	"fmt"
	"testing"
	"time"

	"atcr.io/pkg/appview/db"
	"atcr.io/pkg/atproto"
	"github.com/bluesky-social/indigo/atproto/identity"
	"github.com/bluesky-social/indigo/atproto/syntax"
	_ "github.com/tursodatabase/go-libsql"
)

// fakeDirectory is an in-memory identity.Directory for tests. Lookups of
// unregistered identifiers fail, mirroring unresolvable identities.
type fakeDirectory struct {
	byDID map[string]*identity.Identity
}

func (d *fakeDirectory) LookupDID(_ context.Context, did syntax.DID) (*identity.Identity, error) {
	if ident, ok := d.byDID[did.String()]; ok {
		return ident, nil
	}
	return nil, fmt.Errorf("%w: %s", identity.ErrDIDNotFound, did)
}

func (d *fakeDirectory) LookupHandle(_ context.Context, handle syntax.Handle) (*identity.Identity, error) {
	return nil, fmt.Errorf("%w: %s", identity.ErrHandleNotFound, handle)
}

func (d *fakeDirectory) Lookup(ctx context.Context, atid syntax.AtIdentifier) (*identity.Identity, error) {
	if did, err := atid.AsDID(); err == nil {
		return d.LookupDID(ctx, did)
	}
	return nil, fmt.Errorf("%w: %s", identity.ErrHandleResolutionFailed, atid)
}

func (d *fakeDirectory) Purge(_ context.Context, _ syntax.AtIdentifier) error {
	return nil
}

// holdIdentity builds an identity advertising the atcr_hold service, as real
// holds publish in their DID documents (see pds.HoldServices).
func holdIdentity(did, url string) *identity.Identity {
	return &identity.Identity{
		DID: syntax.DID(did),
		Services: map[string]identity.ServiceEndpoint{
			"atproto_pds": {Type: "AtprotoPersonalDataServer", URL: url},
			"atcr_hold":   {Type: "AtcrHoldService", URL: url},
		},
	}
}

// setupTestDB returns a database built from the real schema.
//
// This used to hand-maintain its own CREATE TABLE statements, which is the same
// drift problem TestSchemaMatchesMigrations exists to prevent, just moved into a
// test: the copy silently fell behind (it still had tags.id after that column
// was dropped, and lacked manifests.manifest_key) and only failed once a query
// happened to touch the difference. Using db.InitDB means schema changes cannot
// rot this file.
func setupTestDB(t *testing.T) *sql.DB {
	database, err := db.InitDB(":memory:", db.LibsqlConfig{})
	if err != nil {
		t.Fatalf("Failed to open test database: %v", err)
	}

	// Foreign keys off, matching the hand-rolled schema this replaced. These
	// tests insert manifests and tags for DIDs with no users row, exercising the
	// processor rather than referential integrity, and libSQL enables foreign
	// keys by default (unlike mattn). Exec, not QueryRow: unlike most libSQL
	// PRAGMAs this one returns no rows, the same way schema.go sets it.
	if _, err := database.Exec("PRAGMA foreign_keys = OFF"); err != nil {
		t.Fatalf("Failed to disable foreign keys: %v", err)
	}
	return database
}

func TestNewProcessor(t *testing.T) {
	database := setupTestDB(t)
	defer database.Close()

	tests := []struct {
		name     string
		useCache bool
	}{
		{"with cache", true},
		{"without cache", false},
	}

	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			p := NewProcessor(database, tt.useCache, nil)
			if p == nil {
				t.Fatal("NewProcessor returned nil")
			}
			if p.db != database {
				t.Error("Processor database not set correctly")
			}
			if p.useCache != tt.useCache {
				t.Errorf("useCache = %v, want %v", p.useCache, tt.useCache)
			}
			if tt.useCache && p.userCache == nil {
				t.Error("Cache enabled but userCache is nil")
			}
			if !tt.useCache && p.userCache != nil {
				t.Error("Cache disabled but userCache is not nil")
			}
		})
	}
}

func TestProcessManifest_ImageManifest(t *testing.T) {
	database := setupTestDB(t)
	defer database.Close()

	// Insert test user (required for foreign key on repository_annotations)
	_, err := database.Exec(`INSERT INTO users (did, handle, pds_endpoint, last_seen) VALUES (?, ?, ?, ?)`,
		"did:plc:test123", "test.bsky.social", "https://pds.example.com", time.Now())
	if err != nil {
		t.Fatalf("Failed to insert test user: %v", err)
	}

	p := NewProcessor(database, false, nil)
	ctx := context.Background()

	// Create test manifest record
	manifestRecord := &atproto.ManifestRecord{
		Repository:    "test-app",
		Digest:        "sha256:abc123",
		MediaType:     "application/vnd.oci.image.manifest.v1+json",
		SchemaVersion: 2,
		HoldEndpoint:  "did:web:hold01.atcr.io",
		CreatedAt:     time.Now(),
		Config: &atproto.BlobReference{
			Digest: "sha256:config123",
			Size:   1234,
		},
		Layers: []atproto.BlobReference{
			{Digest: "sha256:layer1", Size: 5000, MediaType: "application/vnd.oci.image.layer.v1.tar+gzip"},
			{Digest: "sha256:layer2", Size: 3000, MediaType: "application/vnd.oci.image.layer.v1.tar+gzip"},
		},
		Annotations: map[string]string{
			"org.opencontainers.image.title":       "Test App",
			"org.opencontainers.image.description": "A test application",
			"org.opencontainers.image.source":      "https://github.com/test/app",
			"org.opencontainers.image.licenses":    "MIT",
			"io.atcr.icon":                         "https://example.com/icon.png",
		},
	}

	// Marshal to bytes for ProcessManifest
	recordBytes, err := json.Marshal(manifestRecord)
	if err != nil {
		t.Fatalf("Failed to marshal manifest: %v", err)
	}

	// Process manifest
	manifestID, err := p.ProcessManifest(ctx, "did:plc:test123", recordBytes)
	if err != nil {
		t.Fatalf("ProcessManifest failed: %v", err)
	}
	if manifestID == "" {
		t.Error("Expected non-zero manifest ID")
	}

	// Verify manifest was inserted
	var count int
	err = database.QueryRow("SELECT COUNT(*) FROM manifests WHERE did = ? AND repository = ? AND digest = ?",
		"did:plc:test123", "test-app", "sha256:abc123").Scan(&count)
	if err != nil {
		t.Fatalf("Failed to query manifests: %v", err)
	}
	if count != 1 {
		t.Errorf("Expected 1 manifest, got %d", count)
	}

	// Verify annotations were stored in repository_annotations table
	var title, source string
	err = database.QueryRow("SELECT value FROM repository_annotations WHERE did = ? AND repository = ? AND key = ?",
		"did:plc:test123", "test-app", "org.opencontainers.image.title").Scan(&title)
	if err != nil {
		t.Fatalf("Failed to query title annotation: %v", err)
	}
	if title != "Test App" {
		t.Errorf("title = %q, want %q", title, "Test App")
	}

	err = database.QueryRow("SELECT value FROM repository_annotations WHERE did = ? AND repository = ? AND key = ?",
		"did:plc:test123", "test-app", "org.opencontainers.image.source").Scan(&source)
	if err != nil {
		t.Fatalf("Failed to query source annotation: %v", err)
	}
	if source != "https://github.com/test/app" {
		t.Errorf("source = %q, want %q", source, "https://github.com/test/app")
	}

	// Verify layers were inserted
	var layerCount int
	err = database.QueryRow("SELECT COUNT(*) FROM layers WHERE manifest_key = ?", manifestID).Scan(&layerCount)
	if err != nil {
		t.Fatalf("Failed to query layers: %v", err)
	}
	if layerCount != 2 {
		t.Errorf("Expected 2 layers, got %d", layerCount)
	}

	// Verify no manifest references (this is an image, not a list)
	var refCount int
	err = database.QueryRow("SELECT COUNT(*) FROM manifest_references WHERE manifest_key = ?", manifestID).Scan(&refCount)
	if err != nil {
		t.Fatalf("Failed to query manifest_references: %v", err)
	}
	if refCount != 0 {
		t.Errorf("Expected 0 manifest references, got %d", refCount)
	}
}

func TestProcessManifest_ManifestList(t *testing.T) {
	database := setupTestDB(t)
	defer database.Close()

	p := NewProcessor(database, false, nil)
	ctx := context.Background()

	// Create test manifest list record
	manifestRecord := &atproto.ManifestRecord{
		Repository:    "test-app",
		Digest:        "sha256:list123",
		MediaType:     "application/vnd.oci.image.index.v1+json",
		SchemaVersion: 2,
		HoldEndpoint:  "did:web:hold01.atcr.io",
		CreatedAt:     time.Now(),
		Manifests: []atproto.ManifestReference{
			{
				Digest:    "sha256:amd64manifest",
				MediaType: "application/vnd.oci.image.manifest.v1+json",
				Size:      1000,
				Platform: &atproto.Platform{
					Architecture: "amd64",
					OS:           "linux",
				},
			},
			{
				Digest:    "sha256:arm64manifest",
				MediaType: "application/vnd.oci.image.manifest.v1+json",
				Size:      1100,
				Platform: &atproto.Platform{
					Architecture: "arm64",
					OS:           "linux",
					Variant:      "v8",
				},
			},
		},
	}

	// Marshal to bytes for ProcessManifest
	recordBytes, err := json.Marshal(manifestRecord)
	if err != nil {
		t.Fatalf("Failed to marshal manifest: %v", err)
	}

	// Process manifest list
	manifestID, err := p.ProcessManifest(ctx, "did:plc:test123", recordBytes)
	if err != nil {
		t.Fatalf("ProcessManifest failed: %v", err)
	}

	// Verify manifest references were inserted
	var refCount int
	err = database.QueryRow("SELECT COUNT(*) FROM manifest_references WHERE manifest_key = ?", manifestID).Scan(&refCount)
	if err != nil {
		t.Fatalf("Failed to query manifest_references: %v", err)
	}
	if refCount != 2 {
		t.Errorf("Expected 2 manifest references, got %d", refCount)
	}

	// Verify platform info was stored
	var arch, os string
	err = database.QueryRow("SELECT platform_architecture, platform_os FROM manifest_references WHERE manifest_key = ? AND reference_index = 0", manifestID).Scan(&arch, &os)
	if err != nil {
		t.Fatalf("Failed to query platform info: %v", err)
	}
	if arch != "amd64" {
		t.Errorf("platform_architecture = %q, want %q", arch, "amd64")
	}
	if os != "linux" {
		t.Errorf("platform_os = %q, want %q", os, "linux")
	}

	// Verify no layers (this is a list, not an image)
	var layerCount int
	err = database.QueryRow("SELECT COUNT(*) FROM layers WHERE manifest_key = ?", manifestID).Scan(&layerCount)
	if err != nil {
		t.Fatalf("Failed to query layers: %v", err)
	}
	if layerCount != 0 {
		t.Errorf("Expected 0 layers, got %d", layerCount)
	}
}

func TestProcessTag(t *testing.T) {
	database := setupTestDB(t)
	defer database.Close()

	p := NewProcessor(database, false, nil)
	ctx := context.Background()

	// Create test tag record (using ManifestDigest field for simplicity)
	tagRecord := &atproto.TagRecord{
		Repository:     "test-app",
		Tag:            "latest",
		ManifestDigest: "sha256:abc123",
		UpdatedAt:      time.Now(),
	}

	// Marshal to bytes for ProcessTag
	recordBytes, err := json.Marshal(tagRecord)
	if err != nil {
		t.Fatalf("Failed to marshal tag: %v", err)
	}

	// Process tag
	err = p.ProcessTag(ctx, "did:plc:test123", recordBytes)
	if err != nil {
		t.Fatalf("ProcessTag failed: %v", err)
	}

	// Verify tag was inserted
	var count int
	err = database.QueryRow("SELECT COUNT(*) FROM tags WHERE did = ? AND repository = ? AND tag = ?",
		"did:plc:test123", "test-app", "latest").Scan(&count)
	if err != nil {
		t.Fatalf("Failed to query tags: %v", err)
	}
	if count != 1 {
		t.Errorf("Expected 1 tag, got %d", count)
	}

	// Verify digest was stored
	var digest string
	err = database.QueryRow("SELECT digest FROM tags WHERE did = ? AND repository = ? AND tag = ?",
		"did:plc:test123", "test-app", "latest").Scan(&digest)
	if err != nil {
		t.Fatalf("Failed to query tag digest: %v", err)
	}
	if digest != "sha256:abc123" {
		t.Errorf("digest = %q, want %q", digest, "sha256:abc123")
	}

	// Test upserting same tag with new digest
	tagRecord.ManifestDigest = "sha256:newdigest"
	recordBytes, err = json.Marshal(tagRecord)
	if err != nil {
		t.Fatalf("Failed to marshal tag: %v", err)
	}
	err = p.ProcessTag(ctx, "did:plc:test123", recordBytes)
	if err != nil {
		t.Fatalf("ProcessTag (upsert) failed: %v", err)
	}

	// Verify tag was updated
	err = database.QueryRow("SELECT digest FROM tags WHERE did = ? AND repository = ? AND tag = ?",
		"did:plc:test123", "test-app", "latest").Scan(&digest)
	if err != nil {
		t.Fatalf("Failed to query updated tag: %v", err)
	}
	if digest != "sha256:newdigest" {
		t.Errorf("digest = %q, want %q", digest, "sha256:newdigest")
	}

	// Verify still only one tag (upsert, not insert)
	err = database.QueryRow("SELECT COUNT(*) FROM tags WHERE did = ? AND repository = ? AND tag = ?",
		"did:plc:test123", "test-app", "latest").Scan(&count)
	if err != nil {
		t.Fatalf("Failed to query tags after upsert: %v", err)
	}
	if count != 1 {
		t.Errorf("Expected 1 tag after upsert, got %d", count)
	}
}

func TestProcessStar(t *testing.T) {
	database := setupTestDB(t)
	defer database.Close()

	// Insert test users (starrer + owner) so EnsureUser finds them without network calls
	for _, did := range []string{"did:plc:starrer123", "did:plc:owner123"} {
		_, err := database.Exec(
			`INSERT INTO users (did, handle, pds_endpoint, last_seen) VALUES (?, ?, ?, ?)`,
			did, did+".test", "https://pds.example.com", time.Now())
		if err != nil {
			t.Fatalf("Failed to insert test user %s: %v", did, err)
		}
	}

	p := NewProcessor(database, false, nil)
	ctx := context.Background()

	// Create test star record (new AT URI format)
	starRecord := atproto.NewStarRecord("did:plc:owner123", "test-app")

	// Marshal to bytes for ProcessStar
	recordBytes, err := json.Marshal(starRecord)
	if err != nil {
		t.Fatalf("Failed to marshal star: %v", err)
	}

	// Process star
	err = p.ProcessStar(ctx, "did:plc:starrer123", recordBytes)
	if err != nil {
		t.Fatalf("ProcessStar failed: %v", err)
	}

	// Verify star was inserted
	var count int
	err = database.QueryRow("SELECT COUNT(*) FROM stars WHERE starrer_did = ? AND owner_did = ? AND repository = ?",
		"did:plc:starrer123", "did:plc:owner123", "test-app").Scan(&count)
	if err != nil {
		t.Fatalf("Failed to query stars: %v", err)
	}
	if count != 1 {
		t.Errorf("Expected 1 star, got %d", count)
	}

	// Test upserting same star (should be idempotent)
	recordBytes, err = json.Marshal(starRecord)
	if err != nil {
		t.Fatalf("Failed to marshal star: %v", err)
	}
	err = p.ProcessStar(ctx, "did:plc:starrer123", recordBytes)
	if err != nil {
		t.Fatalf("ProcessStar (upsert) failed: %v", err)
	}

	// Verify still only one star
	err = database.QueryRow("SELECT COUNT(*) FROM stars WHERE starrer_did = ? AND owner_did = ? AND repository = ?",
		"did:plc:starrer123", "did:plc:owner123", "test-app").Scan(&count)
	if err != nil {
		t.Fatalf("Failed to query stars after upsert: %v", err)
	}
	if count != 1 {
		t.Errorf("Expected 1 star after upsert, got %d", count)
	}
}

// TestProcessStar_InvalidRecord verifies that star records that don't match
// the current schema (bad JSON or unparseable subject AT URI) are skipped
// with a warning rather than failing the firehose event. Regression guard
// for the behavior change from "return error" to "log warn + return nil".
func TestProcessStar_InvalidRecord(t *testing.T) {
	database := setupTestDB(t)
	defer database.Close()

	if _, err := database.Exec(
		`INSERT INTO users (did, handle, pds_endpoint, last_seen) VALUES (?, ?, ?, ?)`,
		"did:plc:starrer123", "starrer.test", "https://pds.example.com", time.Now()); err != nil {
		t.Fatalf("Failed to insert starrer: %v", err)
	}

	p := NewProcessor(database, false, nil)
	ctx := context.Background()

	cases := []struct {
		name string
		body []byte
	}{
		{name: "garbage JSON", body: []byte("not json at all")},
		{name: "missing subject", body: []byte(`{"$type":"io.atcr.sailor.star","createdAt":"2025-01-01T00:00:00Z"}`)},
		{name: "non-AT-URI subject", body: []byte(`{"$type":"io.atcr.sailor.star","subject":"https://example.com/notaturi","createdAt":"2025-01-01T00:00:00Z"}`)},
	}

	for _, tc := range cases {
		t.Run(tc.name, func(t *testing.T) {
			if err := p.ProcessStar(ctx, "did:plc:starrer123", tc.body); err != nil {
				t.Errorf("ProcessStar with %s: expected nil error (skip-with-warning), got %v", tc.name, err)
			}
		})
	}

	// No rows should have been inserted.
	var count int
	if err := database.QueryRow("SELECT COUNT(*) FROM stars WHERE starrer_did = ?", "did:plc:starrer123").Scan(&count); err != nil {
		t.Fatalf("query stars: %v", err)
	}
	if count != 0 {
		t.Errorf("expected 0 stars after invalid records, got %d", count)
	}
}

func TestProcessSailorProfile(t *testing.T) {
	database := setupTestDB(t)
	defer database.Close()

	if _, err := database.Exec(
		`INSERT INTO users (did, handle, pds_endpoint, last_seen) VALUES (?, ?, ?, ?)`,
		"did:plc:profile123", "profile.test", "https://pds.example.com", time.Now()); err != nil {
		t.Fatalf("Failed to insert user: %v", err)
	}

	p := NewProcessor(database, false, nil)
	ctx := context.Background()

	readDomain := func() string {
		var d string
		if err := database.QueryRow("SELECT registry_domain FROM users WHERE did = ?", "did:plc:profile123").Scan(&d); err != nil {
			t.Fatalf("query registry_domain: %v", err)
		}
		return d
	}

	// A profile with registryDomain set caches it on the user row.
	rec := []byte(`{"$type":"io.atcr.sailor.profile","registryDomain":"buoy.cr","createdAt":"2025-01-01T00:00:00Z"}`)
	if err := p.ProcessSailorProfile(ctx, "did:plc:profile123", rec, nil); err != nil {
		t.Fatalf("ProcessSailorProfile failed: %v", err)
	}
	if got := readDomain(); got != "buoy.cr" {
		t.Errorf("expected cached registry domain 'buoy.cr', got %q", got)
	}

	// A profile with an empty registryDomain clears the cached value.
	recEmpty := []byte(`{"$type":"io.atcr.sailor.profile","createdAt":"2025-01-01T00:00:00Z"}`)
	if err := p.ProcessSailorProfile(ctx, "did:plc:profile123", recEmpty, nil); err != nil {
		t.Fatalf("ProcessSailorProfile (empty) failed: %v", err)
	}
	if got := readDomain(); got != "" {
		t.Errorf("expected cleared registry domain, got %q", got)
	}
}

func TestProcessManifest_Duplicate(t *testing.T) {
	database := setupTestDB(t)
	defer database.Close()

	p := NewProcessor(database, false, nil)
	ctx := context.Background()

	manifestRecord := &atproto.ManifestRecord{
		Repository:    "test-app",
		Digest:        "sha256:abc123",
		MediaType:     "application/vnd.oci.image.manifest.v1+json",
		SchemaVersion: 2,
		HoldEndpoint:  "did:web:hold01.atcr.io",
		CreatedAt:     time.Now(),
	}

	// Marshal to bytes for ProcessManifest
	recordBytes, err := json.Marshal(manifestRecord)
	if err != nil {
		t.Fatalf("Failed to marshal manifest: %v", err)
	}

	// Insert first time
	id1, err := p.ProcessManifest(ctx, "did:plc:test123", recordBytes)
	if err != nil {
		t.Fatalf("First ProcessManifest failed: %v", err)
	}

	// Insert duplicate
	id2, err := p.ProcessManifest(ctx, "did:plc:test123", recordBytes)
	if err != nil {
		t.Fatalf("Duplicate ProcessManifest failed: %v", err)
	}

	// Should return existing ID
	if id1 != id2 {
		t.Errorf("Duplicate manifest got different key: %s vs %s", id1, id2)
	}

	// Verify only one manifest exists
	var count int
	err = database.QueryRow("SELECT COUNT(*) FROM manifests WHERE did = ? AND digest = ?",
		"did:plc:test123", "sha256:abc123").Scan(&count)
	if err != nil {
		t.Fatalf("Failed to query manifests: %v", err)
	}
	if count != 1 {
		t.Errorf("Expected 1 manifest, got %d", count)
	}
}

func TestProcessManifest_EmptyAnnotations(t *testing.T) {
	database := setupTestDB(t)
	defer database.Close()

	p := NewProcessor(database, false, nil)
	ctx := context.Background()

	// Manifest with nil annotations
	manifestRecord := &atproto.ManifestRecord{
		Repository:    "test-app",
		Digest:        "sha256:abc123",
		MediaType:     "application/vnd.oci.image.manifest.v1+json",
		SchemaVersion: 2,
		HoldEndpoint:  "did:web:hold01.atcr.io",
		CreatedAt:     time.Now(),
		Annotations:   nil,
	}

	// Marshal to bytes for ProcessManifest
	recordBytes, err := json.Marshal(manifestRecord)
	if err != nil {
		t.Fatalf("Failed to marshal manifest: %v", err)
	}

	_, err = p.ProcessManifest(ctx, "did:plc:test123", recordBytes)
	if err != nil {
		t.Fatalf("ProcessManifest failed: %v", err)
	}

	// Verify no annotations were stored (nil annotations should not create entries)
	var annotationCount int
	err = database.QueryRow("SELECT COUNT(*) FROM repository_annotations WHERE did = ? AND repository = ?",
		"did:plc:test123", "test-app").Scan(&annotationCount)
	if err != nil {
		t.Fatalf("Failed to query annotations: %v", err)
	}
	if annotationCount != 0 {
		t.Errorf("Expected 0 annotations for nil annotations, got %d", annotationCount)
	}
}

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

	processor := NewProcessor(db, false, nil)

	// Setup: Create test user
	testDID := "did:plc:alice123"
	testHandle := "alice.bsky.social"
	testPDS := "https://bsky.social"
	_, err := db.Exec(`
		INSERT INTO users (did, handle, pds_endpoint, last_seen)
		VALUES (?, ?, ?, ?)
	`, testDID, testHandle, testPDS, time.Now())
	if err != nil {
		t.Fatalf("Failed to insert test user: %v", err)
	}

	// Test 1: Process identity change event
	newHandle := "alice-new.bsky.social"
	err = processor.ProcessIdentity(context.Background(), testDID, newHandle)
	// Note: This will fail to invalidate cache since we don't have a real identity directory,
	// but we can still verify the database update happened
	if err != nil {
		t.Logf("Expected cache invalidation error (no real directory): %v", err)
	}

	// Verify handle was updated in database
	var retrievedHandle string
	err = db.QueryRow(`
		SELECT handle FROM users WHERE did = ?
	`, testDID).Scan(&retrievedHandle)
	if err != nil {
		t.Fatalf("Failed to query updated user: %v", err)
	}
	if retrievedHandle != newHandle {
		t.Errorf("Expected handle '%s', got '%s'", newHandle, retrievedHandle)
	}

	// Test 2: Process identity change for non-existent user
	// Should not error (UPDATE just affects 0 rows)
	err = processor.ProcessIdentity(context.Background(), "did:plc:nonexistent", "new.handle")
	if err != nil {
		t.Logf("Expected cache invalidation error: %v", err)
	}

	// Test 3: Process multiple identity changes
	handles := []string{"alice1.bsky.social", "alice2.bsky.social", "alice3.bsky.social"}
	for _, handle := range handles {
		err = processor.ProcessIdentity(context.Background(), testDID, handle)
		if err != nil {
			t.Logf("Expected cache invalidation error: %v", err)
		}

		err = db.QueryRow(`
			SELECT handle FROM users WHERE did = ?
		`, testDID).Scan(&retrievedHandle)
		if err != nil {
			t.Fatalf("Failed to query user after handle update: %v", err)
		}
		if retrievedHandle != handle {
			t.Errorf("Expected handle '%s', got '%s'", handle, retrievedHandle)
		}
	}
}

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

	// repo_pages, hold_captain_records and hold_crew_members come from the real
	// schema now; setupTestDB creates them.

	// Register the hold DID so captain verification resolves locally instead
	// of hitting the network.
	atproto.SetDirectory(&fakeDirectory{byDID: map[string]*identity.Identity{
		"did:web:hold.example.com": holdIdentity("did:web:hold.example.com", "https://hold.example.com"),
	}})
	defer atproto.SetDirectory(nil)

	processor := NewProcessor(db, false, nil)
	ctx := context.Background()
	var err error

	// Test 1: ProcessRecord routes manifest correctly
	// Note: Schema validation may fail for io.atcr.manifest since we can't resolve the schema,
	// but this tests the routing logic
	manifestRecord := map[string]any{
		"$type":         "io.atcr.manifest",
		"repository":    "test-app",
		"digest":        "sha256:route123",
		"mediaType":     "application/vnd.oci.image.manifest.v1+json",
		"schemaVersion": 2,
		"holdDid":       "did:web:hold01.atcr.io",
		"createdAt":     time.Now().Format(time.RFC3339),
	}
	recordBytes, _ := json.Marshal(manifestRecord)

	// Note: ProcessRecord will skip validation if lexicon can't be resolved (expected in tests)
	// and will skip EnsureUser since we don't have a real PDS to resolve
	// Just verify the record is processed without panic
	err = processor.ProcessRecord(ctx, "did:plc:test123", atproto.ManifestCollection, "route123", recordBytes, false, nil)
	// Error expected since we can't resolve identity - that's fine for this test
	if err != nil {
		t.Logf("Expected error (can't resolve identity): %v", err)
	}

	// Test 2: ProcessRecord handles captain record without creating user
	captainRecord := map[string]any{
		"$type":              "io.atcr.hold.captain",
		"owner":              "did:plc:owner123",
		"public":             true,
		"allowAllCrew":       false,
		"enableBlueskyPosts": false,
		"deployedAt":         time.Now().Format(time.RFC3339),
	}
	captainBytes, _ := json.Marshal(captainRecord)

	// This should NOT call EnsureUser (captain is a hold collection)
	err = processor.ProcessRecord(ctx, "did:web:hold.example.com", atproto.CaptainCollection, "self", captainBytes, false, nil)
	if err != nil {
		t.Logf("Error processing captain (validation may fail in test): %v", err)
	}

	// Verify no user was created for the hold DID
	var userCount int
	err = db.QueryRow(`SELECT COUNT(*) FROM users WHERE did = ?`, "did:web:hold.example.com").Scan(&userCount)
	if err != nil {
		t.Fatalf("Failed to query users: %v", err)
	}
	if userCount != 0 {
		t.Error("Captain record processing should NOT create a user entry for holds")
	}

	// Test 3: ProcessRecord handles delete operations
	err = processor.ProcessRecord(ctx, "did:plc:test123", atproto.ManifestCollection, "sha256:todelete", nil, true, nil)
	if err != nil {
		t.Errorf("Delete should not error: %v", err)
	}
}

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

	holdDID := "did:web:realhold.example.com"
	userDID := "did:plc:notahold"
	atproto.SetDirectory(&fakeDirectory{byDID: map[string]*identity.Identity{
		holdDID: holdIdentity(holdDID, "https://realhold.example.com"),
		userDID: {
			DID: syntax.DID(userDID),
			Services: map[string]identity.ServiceEndpoint{
				"atproto_pds": {Type: "AtprotoPersonalDataServer", URL: "https://pds.example.com"},
			},
		},
	}})
	defer atproto.SetDirectory(nil)

	processor := NewProcessor(db, false, nil)
	ctx := context.Background()

	captainData, _ := 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),
	})

	captainCount := func(did string) int {
		t.Helper()
		var n int
		if err := db.QueryRow(`SELECT COUNT(*) FROM hold_captain_records WHERE hold_did = ?`, did).Scan(&n); err != nil {
			t.Fatalf("Failed to count captain records: %v", err)
		}
		return n
	}

	// A DID advertising the atcr_hold service is cached.
	if err := processor.ProcessCaptain(ctx, holdDID, captainData); err != nil {
		t.Fatalf("ProcessCaptain failed for real hold: %v", err)
	}
	if captainCount(holdDID) != 1 {
		t.Error("Captain record from a real hold should be cached")
	}

	// A resolvable DID without the atcr_hold service is skipped silently.
	if err := processor.ProcessCaptain(ctx, userDID, captainData); err != nil {
		t.Fatalf("ProcessCaptain should skip non-hold DIDs without error: %v", err)
	}
	if captainCount(userDID) != 0 {
		t.Error("Captain record from a non-hold DID should not be cached")
	}

	// An unresolvable DID is skipped silently (periodic backfill retries).
	if err := processor.ProcessCaptain(ctx, "did:plc:unresolvable", captainData); err != nil {
		t.Fatalf("ProcessCaptain should skip unresolvable DIDs without error: %v", err)
	}
	if captainCount("did:plc:unresolvable") != 0 {
		t.Error("Captain record from an unresolvable DID should not be cached")
	}
}

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

	processor := NewProcessor(db, false, nil)
	ctx := context.Background()

	// Test: Invalid JSON should be skipped silently (no error returned)
	invalidJSON := []byte(`{invalid json}`)
	err := processor.ProcessRecord(ctx, "did:plc:test123", atproto.ManifestCollection, "test", invalidJSON, false, nil)
	// Should return nil (skipped silently) not an error
	if err != nil {
		t.Errorf("Invalid record should be skipped silently, got error: %v", err)
	}
}

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

	processor := NewProcessor(db, false, nil)
	ctx := context.Background()

	// Test 1: Manifest passes (no strict validation)
	manifestJSON := []byte(`{"$type": "io.atcr.manifest", "repository": "test"}`)
	err := processor.ValidateRecord(ctx, atproto.ManifestCollection, manifestJSON)
	if err != nil {
		t.Errorf("Manifest should pass validation: %v", err)
	}

	// Test 2: Invalid JSON returns error
	invalidJSON := []byte(`{invalid}`)
	err = processor.ValidateRecord(ctx, atproto.ManifestCollection, invalidJSON)
	if err == nil {
		t.Error("Invalid JSON should return error")
	}

	// Test 3: Captain with valid owner passes
	captainValid := []byte(`{"owner": "did:plc:owner123", "public": true}`)
	err = processor.ValidateRecord(ctx, atproto.CaptainCollection, captainValid)
	if err != nil {
		t.Errorf("Valid captain should pass: %v", err)
	}

	// Test 4: Captain with empty owner is rejected
	captainEmpty := []byte(`{"owner": "", "public": true}`)
	err = processor.ValidateRecord(ctx, atproto.CaptainCollection, captainEmpty)
	if err == nil {
		t.Error("Captain with empty owner should be rejected")
	}

	// Test 5: Captain with invalid owner (not a DID) is rejected
	captainInvalid := []byte(`{"owner": "notadid", "public": true}`)
	err = processor.ValidateRecord(ctx, atproto.CaptainCollection, captainInvalid)
	if err == nil {
		t.Error("Captain with invalid owner should be rejected")
	}

	// Test 6: Crew with valid member passes
	crewValid := []byte(`{"member": "did:plc:member123", "role": "write"}`)
	err = processor.ValidateRecord(ctx, atproto.CrewCollection, crewValid)
	if err != nil {
		t.Errorf("Valid crew should pass: %v", err)
	}

	// Test 7: Crew with empty member is rejected
	crewEmpty := []byte(`{"member": "", "role": "write"}`)
	err = processor.ValidateRecord(ctx, atproto.CrewCollection, crewEmpty)
	if err == nil {
		t.Error("Crew with empty member should be rejected")
	}
}

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

	processor := NewProcessor(db, false, nil)

	// Setup: Create test user
	testDID := "did:plc:bob456"
	testHandle := "bob.bsky.social"
	testPDS := "https://bsky.social"
	_, err := db.Exec(`
		INSERT INTO users (did, handle, pds_endpoint, last_seen)
		VALUES (?, ?, ?, ?)
	`, testDID, testHandle, testPDS, time.Now())
	if err != nil {
		t.Fatalf("Failed to insert test user: %v", err)
	}

	// Test 1: Process account deactivation event
	err = processor.ProcessAccount(context.Background(), testDID, false, "deactivated")
	// Note: Cache invalidation will fail without real directory, but that's expected
	if err != nil {
		t.Logf("Expected cache invalidation error (no real directory): %v", err)
	}

	// Verify user still exists in database (we don't delete on deactivation)
	var exists bool
	err = db.QueryRow(`
		SELECT EXISTS(SELECT 1 FROM users WHERE did = ?)
	`, testDID).Scan(&exists)
	if err != nil {
		t.Fatalf("Failed to check if user exists: %v", err)
	}
	if !exists {
		t.Error("User should still exist after deactivation event (no deletion)")
	}

	// Test 2: Process account with active=true (should be ignored)
	err = processor.ProcessAccount(context.Background(), testDID, true, "active")
	if err != nil {
		t.Errorf("Expected no error for active account, got: %v", err)
	}

	// Test 3: Process account with status != "deactivated" (should be ignored)
	err = processor.ProcessAccount(context.Background(), testDID, false, "suspended")
	if err != nil {
		t.Errorf("Expected no error for non-deactivated status, got: %v", err)
	}

	// Test 4: Process account deactivation for non-existent user
	err = processor.ProcessAccount(context.Background(), "did:plc:nonexistent", false, "deactivated")
	// Cache invalidation will fail, but that's expected
	if err != nil {
		t.Logf("Expected cache invalidation error: %v", err)
	}

	// Test 5: Process multiple deactivation events (idempotent)
	for i := range 3 {
		err = processor.ProcessAccount(context.Background(), testDID, false, "deactivated")
		if err != nil {
			t.Logf("Expected cache invalidation error on iteration %d: %v", i, err)
		}
	}

	// User should still exist after multiple deactivations
	err = db.QueryRow(`
		SELECT EXISTS(SELECT 1 FROM users WHERE did = ?)
	`, testDID).Scan(&exists)
	if err != nil {
		t.Fatalf("Failed to check if user exists after multiple deactivations: %v", err)
	}
	if !exists {
		t.Error("User should still exist after multiple deactivation events")
	}

	// Test 6: Process account deletion - should delete user data
	err = processor.ProcessAccount(context.Background(), testDID, false, "deleted")
	if err != nil {
		t.Logf("Cache invalidation error during deletion (expected): %v", err)
	}

	// User should be deleted after "deleted" status
	err = db.QueryRow(`
		SELECT EXISTS(SELECT 1 FROM users WHERE did = ?)
	`, testDID).Scan(&exists)
	if err != nil {
		t.Fatalf("Failed to check if user exists after deletion: %v", err)
	}
	if exists {
		t.Error("User should NOT exist after deletion event")
	}

	// Test 7: Process deletion for already-deleted user (idempotent)
	err = processor.ProcessAccount(context.Background(), testDID, false, "deleted")
	if err != nil {
		t.Errorf("Deletion of non-existent user should not error, got: %v", err)
	}
}

// TestBackfillDoesNotStampLastSeen: last_seen means "this user did something
// recently", and the backfill is not the user doing something.
//
// It walks every historical record in the network, so stamping there records
// when the backfill ran rather than when the user was last active — and does it
// for every user at once, on every run. That makes the column useless as an
// activity signal, which is the only thing it is for.
func TestBackfillDoesNotStampLastSeen(t *testing.T) {
	database := setupTestDB(t)
	defer database.Close()

	const did = "did:plc:lastseen"
	atproto.SetDirectory(&fakeDirectory{byDID: map[string]*identity.Identity{
		did: {
			DID:    syntax.DID(did),
			Handle: syntax.Handle("lastseen.example.com"),
			Services: map[string]identity.ServiceEndpoint{
				"atproto_pds": {Type: "AtprotoPersonalDataServer", URL: "https://pds.example.com"},
			},
		},
	}})

	stale := time.Now().Add(-90 * 24 * time.Hour).UTC().Truncate(time.Second)
	if _, err := database.Exec(`
		INSERT INTO users (did, handle, pds_endpoint, last_seen)
		VALUES (?, 'lastseen.example.com', 'https://pds.example.com', ?)
	`, did, stale); err != nil {
		t.Fatalf("seed user: %v", err)
	}

	// Backfill: useCache=false.
	backfill := NewProcessor(database, false, nil)
	if err := backfill.EnsureUser(context.Background(), did); err != nil {
		t.Fatalf("backfill EnsureUser: %v", err)
	}

	var after time.Time
	if err := database.QueryRow(`SELECT last_seen FROM users WHERE did = ?`, did).Scan(&after); err != nil {
		t.Fatalf("read last_seen: %v", err)
	}
	if !after.UTC().Truncate(time.Second).Equal(stale) {
		t.Errorf("backfill moved last_seen from %v to %v; it now records when the backfill ran, "+
			"not when the user was active", stale, after)
	}
}

// TestLiveEventStampsLastSeen is the other half: a live commit means this user
// just wrote a record, which is activity worth recording.
func TestLiveEventStampsLastSeen(t *testing.T) {
	database := setupTestDB(t)
	defer database.Close()

	const did = "did:plc:lastseenlive"
	atproto.SetDirectory(&fakeDirectory{byDID: map[string]*identity.Identity{
		did: {
			DID:    syntax.DID(did),
			Handle: syntax.Handle("live.example.com"),
			Services: map[string]identity.ServiceEndpoint{
				"atproto_pds": {Type: "AtprotoPersonalDataServer", URL: "https://pds.example.com"},
			},
		},
	}})

	stale := time.Now().Add(-90 * 24 * time.Hour).UTC().Truncate(time.Second)
	if _, err := database.Exec(`
		INSERT INTO users (did, handle, pds_endpoint, last_seen)
		VALUES (?, 'live.example.com', 'https://pds.example.com', ?)
	`, did, stale); err != nil {
		t.Fatalf("seed user: %v", err)
	}

	// Worker: useCache=true.
	live := NewProcessor(database, true, nil)
	if err := live.EnsureUser(context.Background(), did); err != nil {
		t.Fatalf("live EnsureUser: %v", err)
	}

	var after time.Time
	if err := database.QueryRow(`SELECT last_seen FROM users WHERE did = ?`, did).Scan(&after); err != nil {
		t.Fatalf("read last_seen: %v", err)
	}
	if !after.After(stale) {
		t.Errorf("live event did not stamp last_seen: still %v", after)
	}
}

// TestTouchLastSeenIsThrottled covers the other half of 13edb71.
//
// That commit throttles two writes: DeviceStore.UpdateLastUsed and this one.
// Only the device side got a test. touchLastSeen is the hotter of the two — it
// ran once per indexed record, so a busy firehose meant a database round trip
// per event for a timestamp read in hours or days. Deleting the throttle here
// leaves every existing test green.
func TestTouchLastSeenIsThrottled(t *testing.T) {
	database := setupTestDB(t)
	defer database.Close()

	const did = "did:plc:touchthrottle"
	stale := time.Now().Add(-90 * 24 * time.Hour).UTC().Truncate(time.Second)
	if _, err := database.Exec(`
		INSERT INTO users (did, handle, pds_endpoint, last_seen)
		VALUES (?, 'touch.example.com', 'https://pds.example.com', ?)
	`, did, stale); err != nil {
		t.Fatalf("seed user: %v", err)
	}

	// useCache=true is the live worker, and the only configuration that has a
	// cache to throttle against.
	live := NewProcessor(database, true, nil)

	if err := live.touchLastSeen(did); err != nil {
		t.Fatalf("first touchLastSeen: %v", err)
	}
	var first time.Time
	if err := database.QueryRow(`SELECT last_seen FROM users WHERE did = ?`, did).Scan(&first); err != nil {
		t.Fatalf("read last_seen: %v", err)
	}
	if first.UTC().Truncate(time.Second).Equal(stale) {
		t.Fatal("first call did not write last_seen")
	}

	// Put a value there that any further write would visibly change, then
	// hammer it the way a firehose burst would.
	marker := first.Add(-time.Hour).UTC().Truncate(time.Second)
	if _, err := database.Exec(`UPDATE users SET last_seen = ? WHERE did = ?`, marker, did); err != nil {
		t.Fatalf("set marker: %v", err)
	}
	for range 50 {
		if err := live.touchLastSeen(did); err != nil {
			t.Fatalf("throttled touchLastSeen: %v", err)
		}
	}

	var after time.Time
	if err := database.QueryRow(`SELECT last_seen FROM users WHERE did = ?`, did).Scan(&after); err != nil {
		t.Fatalf("read last_seen: %v", err)
	}
	if !after.UTC().Truncate(time.Second).Equal(marker) {
		t.Error("last_seen was rewritten during 50 back-to-back events; the throttle is not holding")
	}
}

// TestTouchLastSeenThrottlesPerUser: one busy account must not suppress
// another's first write, which is what a shared timestamp would do.
func TestTouchLastSeenThrottlesPerUser(t *testing.T) {
	database := setupTestDB(t)
	defer database.Close()

	const didA = "did:plc:busyuser"
	const didB = "did:plc:quietuser"
	stale := time.Now().Add(-90 * 24 * time.Hour).UTC().Truncate(time.Second)
	for _, did := range []string{didA, didB} {
		if _, err := database.Exec(`
			INSERT INTO users (did, handle, pds_endpoint, last_seen)
			VALUES (?, ?, 'https://pds.example.com', ?)
		`, did, did+".example.com", stale); err != nil {
			t.Fatalf("seed %s: %v", did, err)
		}
	}

	live := NewProcessor(database, true, nil)
	for range 10 {
		if err := live.touchLastSeen(didA); err != nil {
			t.Fatalf("touchLastSeen A: %v", err)
		}
	}
	if err := live.touchLastSeen(didB); err != nil {
		t.Fatalf("touchLastSeen B: %v", err)
	}

	var b time.Time
	if err := database.QueryRow(`SELECT last_seen FROM users WHERE did = ?`, didB).Scan(&b); err != nil {
		t.Fatalf("read last_seen: %v", err)
	}
	if b.UTC().Truncate(time.Second).Equal(stale) {
		t.Error("a busy account's writes suppressed a quiet account's first write")
	}
}