From 2b00f54292c54681e0c4687fe9a950c7337f7e46 Mon Sep 17 00:00:00 2001 From: Lewis Date: Tue, 12 May 2026 12:51:40 +0000 Subject: [PATCH] spindle: legacy secret migration via tap resync nudge Lewis: May this revision serve well! --- spindle/secret_copy.go | 59 +++++++++++++++++++++++++++++++++++++++++++++++++++++++++++ spindle/server.go | 2 +- spindle/startup_migrations.go | 140 +++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++--------------------------------------------------------- spindle/startup_migrations_test.go | 542 ++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++-------------------------------------------------------------------------- spindle/tapclient.go | 12 ++++++++++++ spindle/db/repos.go | 16 ++++++++++++---- spindle/db/repos_test.go | 143 +++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++ 7 file(s) changed, 778 insertion(s)(+), 136 deletion(s)(-) diff --git a/spindle/secret_copy.go b/spindle/secret_copy.go --- a/spindle/secret_copy.go +++ b/spindle/secret_copy.go @@ -4,7 +4,10 @@ "context" "errors" "fmt" + "log/slog" + "github.com/bluesky-social/indigo/atproto/syntax" + "tangled.org/core/spindle/db" "tangled.org/core/spindle/secrets" ) @@ -36,4 +39,60 @@ } } return step(cur, 0) +} + +func legacyKeyCandidates(owner syntax.DID, name string, rkey syntax.RecordKey) []string { + o := owner.String() + r := rkey.String() + switch { + case name == "" && r == "": + return nil + case name == "": + return []string{o + "/" + r} + case r == "" || name == r: + return []string{o + "/" + name} + default: + return []string{o + "/" + name, o + "/" + r} + } +} + +func migrateLegacyRepoSecrets(ctx context.Context, d *db.DB, vault secrets.Manager, logger *slog.Logger, owner syntax.DID, name string, rkey syntax.RecordKey, repoDid syntax.DID) { + candidates := legacyKeyCandidates(owner, name, rkey) + if len(candidates) == 0 { + return + } + flag := "legacy-secret-copy:" + repoDid.String() + ":" + rkey.String() + var exists bool + if err := d.QueryRowContext(ctx, `select exists (select 1 from migrations where name = ?)`, flag).Scan(&exists); err != nil { + logger.Warn("legacy secret copy: check migration flag", "err", err) + return + } + if exists { + return + } + + newID := secrets.RepoIdentifier(repoDid.String()) + var step func(remaining []string, copied int) (int, error) + step = func(remaining []string, copied int) (int, error) { + if len(remaining) == 0 { + return copied, nil + } + oldID := secrets.RepoIdentifier(remaining[0]) + n, err := copyRepoSecrets(ctx, vault, oldID, newID) + if err != nil { + return copied, fmt.Errorf("copy %s -> %s: %w", oldID, newID, err) + } + return step(remaining[1:], copied+n) + } + total, err := step(candidates, 0) + if err != nil { + logger.Warn("legacy secret copy failed", "err", err) + return + } + + if _, err := d.ExecContext(ctx, `insert or ignore into migrations (name) values (?)`, flag); err != nil { + logger.Warn("legacy secret copy: mark flag failed", "err", err) + return + } + logger.Info("legacy secret migrate done", "owner", owner, "name", name, "rkey", rkey, "repoDid", repoDid, "candidates", candidates, "copied", total) } diff --git a/spindle/server.go b/spindle/server.go --- a/spindle/server.go +++ b/spindle/server.go @@ -100,7 +100,7 @@ return nil, fmt.Errorf("unknown secrets provider: %s", cfg.Server.Secrets.Provider) } - if err := runStartupMigrations(ctx, d, vault, logger); err != nil { + if err := runStartupMigrations(ctx, d, cfg.Server.Tap.Embed, cfg.Server.Tap.DBPath, logger); err != nil { return nil, fmt.Errorf("failed to run startup migrations: %w", err) } diff --git a/spindle/startup_migrations.go b/spindle/startup_migrations.go --- a/spindle/startup_migrations.go +++ b/spindle/startup_migrations.go @@ -3,78 +3,104 @@ import ( "context" "database/sql" + "errors" "fmt" "log/slog" + "os" - "tangled.org/core/orm" + _ "github.com/mattn/go-sqlite3" "tangled.org/core/spindle/db" - "tangled.org/core/spindle/secrets" ) -func runStartupMigrations(ctx context.Context, d *db.DB, vault secrets.Manager, logger *slog.Logger) error { - conn, err := d.DB.Conn(ctx) - if err != nil { - return fmt.Errorf("acquire spindle conn: %w", err) +const forceTapResyncFlag = "force-tap-repo-resync-v1" + +func runStartupMigrations(ctx context.Context, d *db.DB, tapEmbed bool, tapDBPath string, logger *slog.Logger) error { + if err := cleanupOrphanRepos(ctx, d, logger); err != nil { + return fmt.Errorf("cleanup orphan repos: %w", err) } - defer conn.Close() - - return orm.RunMigration(conn, logger, "copy-owner-rkey-secrets-to-repo-did", func(tx *sql.Tx) error { - return copyOwnerRkeySecretsToRepoDid(ctx, tx, vault, logger) - }) + if !tapEmbed { + logger.Warn("tap not embedded: legacy repos won't auto-resync; trigger external tap resync to migrate secrets/casbin") + return nil + } + if err := nudgeTapForResync(ctx, d, tapDBPath, logger); err != nil { + return fmt.Errorf("nudge tap for resync: %w", err) + } + return nil } -type repoSecretPair struct { - oldID, newID secrets.RepoIdentifier +func cleanupOrphanRepos(ctx context.Context, d *db.DB, logger *slog.Logger) error { + res, err := d.ExecContext(ctx, ` + delete from repos + where coalesce(repo_did, '') = '' + and exists ( + select 1 from repos r2 + where r2.owner = repos.owner + and coalesce(r2.repo_did, '') <> '' + ) + `) + if err != nil { + return fmt.Errorf("delete orphan repos: %w", err) + } + n, _ := res.RowsAffected() + if n > 0 { + logger.Info("cleaned up orphan repos missing repo_did", "deleted", n) + } + return nil } -func loadRepoSecretPairs(ctx context.Context, tx *sql.Tx) ([]repoSecretPair, error) { - rows, err := tx.QueryContext(ctx, - `select owner, rkey, repo_did from repos - where repo_did is not null and repo_did <> ''`, +func nudgeTapForResync(ctx context.Context, d *db.DB, tapDBPath string, logger *slog.Logger) error { + if tapDBPath == "" { + return fmt.Errorf("tap db path empty in embed mode") + } + var exists bool + if err := d.QueryRowContext(ctx, + `select exists (select 1 from migrations where name = ?)`, + forceTapResyncFlag, + ).Scan(&exists); err != nil { + return fmt.Errorf("check %s flag: %w", forceTapResyncFlag, err) + } + if exists { + logger.Warn("skipped migration, already applied", "migration", forceTapResyncFlag) + return nil + } + + markDone := func() error { + if _, err := d.ExecContext(ctx, + `insert or ignore into migrations (name) values (?)`, + forceTapResyncFlag, + ); err != nil { + return fmt.Errorf("mark %s done: %w", forceTapResyncFlag, err) + } + return nil + } + + if _, err := os.Stat(tapDBPath); errors.Is(err, os.ErrNotExist) { + logger.Info("tap db not yet created, marking resync nudge done", "migration", forceTapResyncFlag, "path", tapDBPath) + return markDone() + } else if err != nil { + return fmt.Errorf("stat tap db: %w", err) + } + + tdb, err := sql.Open("sqlite3", tapDBPath+"?_busy_timeout=5000") + if err != nil { + return fmt.Errorf("open tap db: %w", err) + } + defer tdb.Close() + + if _, err := tdb.ExecContext(ctx, `delete from repo_records`); err != nil { + return fmt.Errorf("clear tap repo_records: %w", err) + } + res, err := tdb.ExecContext(ctx, + `update repos set state = 'desynchronized', retry_after = 0 where state in ('active','error')`, ) if err != nil { - return nil, fmt.Errorf("select repos: %w", err) + return fmt.Errorf("desync tap repos: %w", err) } - defer rows.Close() + n, _ := res.RowsAffected() - var collect func(acc []repoSecretPair) ([]repoSecretPair, error) - collect = func(acc []repoSecretPair) ([]repoSecretPair, error) { - if !rows.Next() { - return acc, rows.Err() - } - var owner, rkey, repoDid string - if err := rows.Scan(&owner, &rkey, &repoDid); err != nil { - return acc, fmt.Errorf("scan repos row: %w", err) - } - return collect(append(acc, repoSecretPair{ - oldID: secrets.RepoIdentifier(owner + "/" + rkey), - newID: secrets.RepoIdentifier(repoDid), - })) - } - return collect(nil) -} - -func copyOwnerRkeySecretsToRepoDid(ctx context.Context, tx *sql.Tx, vault secrets.Manager, logger *slog.Logger) error { - pairs, err := loadRepoSecretPairs(ctx, tx) - if err != nil { + if err := markDone(); err != nil { return err } - - var step func(remaining []repoSecretPair, totalCopied int) error - step = func(remaining []repoSecretPair, totalCopied int) error { - if len(remaining) == 0 { - logger.Info("secret copy migration complete", "rows", len(pairs), "copied", totalCopied) - return nil - } - p := remaining[0] - n, err := copyRepoSecrets(ctx, vault, p.oldID, p.newID) - if err != nil { - return fmt.Errorf("copy %s -> %s: %w", p.oldID, p.newID, err) - } - if n > 0 { - logger.Info("secrets copied", "old", p.oldID, "new", p.newID, "count", n) - } - return step(remaining[1:], totalCopied+n) - } - return step(pairs, 0) + logger.Info("nudged tap to resync", "migration", forceTapResyncFlag, "repos_desynced", n) + return nil } diff --git a/spindle/startup_migrations_test.go b/spindle/startup_migrations_test.go --- a/spindle/startup_migrations_test.go +++ b/spindle/startup_migrations_test.go @@ -2,6 +2,7 @@ import ( "context" + "database/sql" "io" "log/slog" "path/filepath" @@ -13,6 +14,65 @@ "tangled.org/core/spindle/db" "tangled.org/core/spindle/secrets" ) + +func seedTapDB(t *testing.T, path string) { + t.Helper() + tdb, err := sql.Open("sqlite3", path) + if err != nil { + t.Fatalf("open tap db: %v", err) + } + defer tdb.Close() + if _, err := tdb.Exec(` + create table repos ( + did text primary key, + state text not null default 'pending', + status text not null default 'active', + handle text default '', + rev text default '', + prev_data text default '', + error_msg text default '', + retry_count integer not null default 0, + retry_after integer not null default 0 + ); + create table repo_records ( + did text not null, + collection text not null, + rkey text not null, + cid text not null, + primary key (did, collection, rkey) + ); + `); err != nil { + t.Fatalf("create tap tables: %v", err) + } +} + +func tapRepoState(t *testing.T, path, did string) string { + t.Helper() + tdb, err := sql.Open("sqlite3", path) + if err != nil { + t.Fatalf("open tap db: %v", err) + } + defer tdb.Close() + var state string + if err := tdb.QueryRow(`select state from repos where did = ?`, did).Scan(&state); err != nil { + t.Fatalf("query state for %s: %v", did, err) + } + return state +} + +func tapRecordCount(t *testing.T, path string) int { + t.Helper() + tdb, err := sql.Open("sqlite3", path) + if err != nil { + t.Fatalf("open tap db: %v", err) + } + defer tdb.Close() + var n int + if err := tdb.QueryRow(`select count(*) from repo_records`).Scan(&n); err != nil { + t.Fatalf("count repo_records: %v", err) + } + return n +} func newTestSpindleDB(t *testing.T) *db.DB { t.Helper() @@ -33,18 +93,6 @@ return vault } -func mustAddRepo(t *testing.T, d *db.DB, knot, owner, rkey, repoDid string) { - t.Helper() - if err := d.AddRepo(db.Repo{ - Knot: knot, - Owner: syntax.DID(owner), - Rkey: syntax.RecordKey(rkey), - RepoDid: syntax.DID(repoDid), - }); err != nil { - t.Fatalf("AddRepo(%s): %v", rkey, err) - } -} - func mustAddSecret(t *testing.T, vault secrets.Manager, repo, key, value string, createdAt time.Time, by string) { t.Helper() err := vault.AddSecret(context.Background(), secrets.UnlockedSecret{ @@ -59,39 +107,31 @@ } } -func TestStartupMigrations_CopyOwnerRkeySecretsToRepoDid(t *testing.T) { +func TestMigrateLegacyRepoSecrets_NameCandidate(t *testing.T) { ctx := context.Background() logger := slog.New(slog.NewTextHandler(io.Discard, nil)) - d := newTestSpindleDB(t) vault := newTestVault(t) - owner := "did:plc:akshay" - migratedRepoDid := "did:plc:boltless" - skippedRkey := "3kspindlerkey00b" - migratedRkey := "3kspindlerkey00a" - - mustAddRepo(t, d, "knot.test", owner, migratedRkey, migratedRepoDid) - mustAddRepo(t, d, "knot.test", owner, skippedRkey, "") + owner := syntax.DID("did:plc:akshay") + repoDid := syntax.DID("did:plc:boltless") + displayName := "myrepo" + rkey := syntax.RecordKey("3kspindlerkey00a") created := time.Date(2024, 6, 1, 12, 0, 0, 0, time.UTC) - oldRepoKey := owner + "/" + migratedRkey - skippedKey := owner + "/" + skippedRkey + oldNameKey := owner.String() + "/" + displayName - mustAddSecret(t, vault, oldRepoKey, "API_KEY", "alpha", created, owner) - mustAddSecret(t, vault, oldRepoKey, "DB_PASSWORD", "bravo", created.Add(1*time.Hour), owner) - mustAddSecret(t, vault, skippedKey, "STRAY", "delta", created, owner) + mustAddSecret(t, vault, oldNameKey, "API_KEY", "alpha", created, owner.String()) + mustAddSecret(t, vault, oldNameKey, "DB_PASSWORD", "bravo", created.Add(1*time.Hour), owner.String()) - if err := runStartupMigrations(ctx, d, vault, logger); err != nil { - t.Fatalf("first migration run: %v", err) - } + migrateLegacyRepoSecrets(ctx, d, vault, logger, owner, displayName, rkey, repoDid) - copied, err := vault.GetSecretsUnlocked(ctx, secrets.RepoIdentifier(migratedRepoDid)) + copied, err := vault.GetSecretsUnlocked(ctx, secrets.RepoIdentifier(repoDid)) if err != nil { t.Fatalf("GetSecretsUnlocked(new): %v", err) } if len(copied) != 2 { - t.Fatalf("expected 2 secrets under new repo_did key, got %d", len(copied)) + t.Fatalf("expected 2 secrets under repo_did key, got %d", len(copied)) } want := map[string]struct { @@ -104,7 +144,7 @@ for _, s := range copied { w, ok := want[s.Key] if !ok { - t.Errorf("unexpected key %q under %s", s.Key, migratedRepoDid) + t.Errorf("unexpected key %q under %s", s.Key, repoDid) continue } if s.Value != w.value { @@ -113,12 +153,12 @@ if !s.CreatedAt.Equal(w.createdAt) { t.Errorf("%s: CreatedAt got %s, want %s", s.Key, s.CreatedAt, w.createdAt) } - if string(s.Repo) != migratedRepoDid { - t.Errorf("%s: Repo got %s, want %s", s.Key, s.Repo, migratedRepoDid) + if string(s.Repo) != repoDid.String() { + t.Errorf("%s: Repo got %s, want %s", s.Key, s.Repo, repoDid) } } - orig, err := vault.GetSecretsUnlocked(ctx, secrets.RepoIdentifier(oldRepoKey)) + orig, err := vault.GetSecretsUnlocked(ctx, secrets.RepoIdentifier(oldNameKey)) if err != nil { t.Fatalf("GetSecretsUnlocked(old): %v", err) } @@ -126,19 +166,8 @@ t.Errorf("expected old-key secrets preserved, got %d", len(orig)) } - stray, err := vault.GetSecretsUnlocked(ctx, secrets.RepoIdentifier(skippedKey)) - if err != nil { - t.Fatalf("GetSecretsUnlocked(skipped): %v", err) - } - if len(stray) != 1 { - t.Errorf("expected skipped repo's old-key secret untouched, got %d", len(stray)) - } - - if err := runStartupMigrations(ctx, d, vault, logger); err != nil { - t.Fatalf("second migration run: %v", err) - } - - again, err := vault.GetSecretsUnlocked(ctx, secrets.RepoIdentifier(migratedRepoDid)) + migrateLegacyRepoSecrets(ctx, d, vault, logger, owner, displayName, rkey, repoDid) + again, err := vault.GetSecretsUnlocked(ctx, secrets.RepoIdentifier(repoDid)) if err != nil { t.Fatalf("GetSecretsUnlocked(new) after re-run: %v", err) } @@ -149,54 +178,108 @@ var marked int if err := d.QueryRow( `select count(*) from migrations where name = ?`, - "copy-owner-rkey-secrets-to-repo-did", + "legacy-secret-copy:"+repoDid.String()+":"+rkey.String(), ).Scan(&marked); err != nil { t.Fatalf("query migrations: %v", err) } if marked != 1 { - t.Errorf("expected migration recorded exactly once, got %d", marked) + t.Errorf("expected per-repo flag recorded exactly once, got %d", marked) } } -func TestStartupMigrations_NoRepos(t *testing.T) { +func TestMigrateLegacyRepoSecrets_RkeyCandidate(t *testing.T) { ctx := context.Background() logger := slog.New(slog.NewTextHandler(io.Discard, nil)) d := newTestSpindleDB(t) vault := newTestVault(t) - if err := runStartupMigrations(ctx, d, vault, logger); err != nil { - t.Fatalf("migration on empty db: %v", err) - } -} - -func TestStartupMigrations_PartialPreExisting(t *testing.T) { - ctx := context.Background() - logger := slog.New(slog.NewTextHandler(io.Discard, nil)) - d := newTestSpindleDB(t) - vault := newTestVault(t) - - owner := "did:plc:akshay" - repoDid := "did:plc:boltless" - rkey := "3kspindlerkey00a" - mustAddRepo(t, d, "knot.test", owner, rkey, repoDid) + owner := syntax.DID("did:plc:akshay") + repoDid := syntax.DID("did:plc:boltless") + displayName := "myrepo" + rkey := syntax.RecordKey("3kspindlerkey00a") created := time.Date(2024, 6, 1, 12, 0, 0, 0, time.UTC) - oldKey := owner + "/" + rkey - mustAddSecret(t, vault, oldKey, "API_KEY", "alpha", created, owner) - mustAddSecret(t, vault, oldKey, "DB_PASSWORD", "bravo", created, owner) + oldRkeyKey := owner.String() + "/" + rkey.String() - mustAddSecret(t, vault, repoDid, "API_KEY", "pre-existing", created.Add(-24*time.Hour), owner) + mustAddSecret(t, vault, oldRkeyKey, "API_KEY", "alpha", created, owner.String()) + mustAddSecret(t, vault, oldRkeyKey, "DB_PASSWORD", "bravo", created, owner.String()) - if err := runStartupMigrations(ctx, d, vault, logger); err != nil { - t.Fatalf("migration: %v", err) - } + migrateLegacyRepoSecrets(ctx, d, vault, logger, owner, displayName, rkey, repoDid) got, err := vault.GetSecretsUnlocked(ctx, secrets.RepoIdentifier(repoDid)) if err != nil { t.Fatalf("GetSecretsUnlocked: %v", err) } if len(got) != 2 { - t.Fatalf("expected 2 secrets under new key, got %d", len(got)) + t.Fatalf("expected 2 secrets copied via rkey candidate, got %d", len(got)) + } +} + +func TestMigrateLegacyRepoSecrets_BothCandidates(t *testing.T) { + ctx := context.Background() + logger := slog.New(slog.NewTextHandler(io.Discard, nil)) + d := newTestSpindleDB(t) + vault := newTestVault(t) + + owner := syntax.DID("did:plc:akshay") + repoDid := syntax.DID("did:plc:boltless") + displayName := "myrepo" + rkey := syntax.RecordKey("3kspindlerkey00a") + + created := time.Date(2024, 6, 1, 12, 0, 0, 0, time.UTC) + oldNameKey := owner.String() + "/" + displayName + oldRkeyKey := owner.String() + "/" + rkey.String() + + mustAddSecret(t, vault, oldNameKey, "FROM_NAME", "n", created, owner.String()) + mustAddSecret(t, vault, oldRkeyKey, "FROM_RKEY", "r", created, owner.String()) + + migrateLegacyRepoSecrets(ctx, d, vault, logger, owner, displayName, rkey, repoDid) + + got, err := vault.GetSecretsUnlocked(ctx, secrets.RepoIdentifier(repoDid)) + if err != nil { + t.Fatalf("GetSecretsUnlocked: %v", err) + } + if len(got) != 2 { + t.Fatalf("expected 2 secrets merged from both candidates, got %d", len(got)) + } + seen := map[string]string{} + for _, s := range got { + seen[s.Key] = s.Value + } + if seen["FROM_NAME"] != "n" { + t.Errorf("FROM_NAME missing or wrong value: %q", seen["FROM_NAME"]) + } + if seen["FROM_RKEY"] != "r" { + t.Errorf("FROM_RKEY missing or wrong value: %q", seen["FROM_RKEY"]) + } +} + +func TestMigrateLegacyRepoSecrets_PreExistingTakesPriority(t *testing.T) { + ctx := context.Background() + logger := slog.New(slog.NewTextHandler(io.Discard, nil)) + d := newTestSpindleDB(t) + vault := newTestVault(t) + + owner := syntax.DID("did:plc:akshay") + repoDid := syntax.DID("did:plc:boltless") + displayName := "myrepo" + rkey := syntax.RecordKey("3kspindlerkey00a") + + created := time.Date(2024, 6, 1, 12, 0, 0, 0, time.UTC) + oldKey := owner.String() + "/" + displayName + + mustAddSecret(t, vault, oldKey, "API_KEY", "alpha", created, owner.String()) + mustAddSecret(t, vault, oldKey, "DB_PASSWORD", "bravo", created, owner.String()) + mustAddSecret(t, vault, repoDid.String(), "API_KEY", "pre-existing", created.Add(-24*time.Hour), owner.String()) + + migrateLegacyRepoSecrets(ctx, d, vault, logger, owner, displayName, rkey, repoDid) + + got, err := vault.GetSecretsUnlocked(ctx, secrets.RepoIdentifier(repoDid)) + if err != nil { + t.Fatalf("GetSecretsUnlocked: %v", err) + } + if len(got) != 2 { + t.Fatalf("expected 2 secrets under repo_did key, got %d", len(got)) } for _, s := range got { if s.Key == "API_KEY" && s.Value != "pre-existing" { @@ -205,5 +288,316 @@ if s.Key == "DB_PASSWORD" && s.Value != "bravo" { t.Errorf("DB_PASSWORD should be copied, got %q", s.Value) } + } +} + +func TestMigrateLegacyRepoSecrets_EmptyName(t *testing.T) { + ctx := context.Background() + logger := slog.New(slog.NewTextHandler(io.Discard, nil)) + d := newTestSpindleDB(t) + vault := newTestVault(t) + + owner := syntax.DID("did:plc:akshay") + repoDid := syntax.DID("did:plc:boltless") + rkey := syntax.RecordKey("3kspindlerkey00a") + + created := time.Date(2024, 6, 1, 12, 0, 0, 0, time.UTC) + oldRkeyKey := owner.String() + "/" + rkey.String() + mustAddSecret(t, vault, oldRkeyKey, "API_KEY", "alpha", created, owner.String()) + + migrateLegacyRepoSecrets(ctx, d, vault, logger, owner, "", rkey, repoDid) + + got, err := vault.GetSecretsUnlocked(ctx, secrets.RepoIdentifier(repoDid)) + if err != nil { + t.Fatalf("GetSecretsUnlocked: %v", err) + } + if len(got) != 1 { + t.Errorf("expected 1 secret via rkey candidate when name empty, got %d", len(got)) + } +} + +func TestMigrateLegacyRepoSecrets_BothEmpty(t *testing.T) { + ctx := context.Background() + logger := slog.New(slog.NewTextHandler(io.Discard, nil)) + d := newTestSpindleDB(t) + vault := newTestVault(t) + + owner := syntax.DID("did:plc:akshay") + repoDid := syntax.DID("did:plc:boltless") + + migrateLegacyRepoSecrets(ctx, d, vault, logger, owner, "", "", repoDid) + + got, err := vault.GetSecretsUnlocked(ctx, secrets.RepoIdentifier(repoDid)) + if err != nil { + t.Fatalf("GetSecretsUnlocked: %v", err) + } + if len(got) != 0 { + t.Errorf("expected no work when both name and rkey empty, got %d secrets", len(got)) + } + + var marked int + if err := d.QueryRow( + `select count(*) from migrations where name like ?`, + "legacy-secret-copy:"+repoDid.String()+":%", + ).Scan(&marked); err != nil { + t.Fatalf("query migrations: %v", err) + } + if marked != 0 { + t.Errorf("empty inputs should not record flag, got %d", marked) + } +} + +func TestNudgeTapForResync(t *testing.T) { + ctx := context.Background() + logger := slog.New(slog.NewTextHandler(io.Discard, nil)) + d := newTestSpindleDB(t) + + tapPath := filepath.Join(t.TempDir(), "tap.db") + seedTapDB(t, tapPath) + + tdb, err := sql.Open("sqlite3", tapPath) + if err != nil { + t.Fatalf("open tap db: %v", err) + } + if _, err := tdb.Exec(`insert into repos (did, state) values + ('did:plc:akshay', 'active'), + ('did:plc:boltless', 'error'), + ('did:plc:limpet', 'pending') + `); err != nil { + t.Fatalf("seed repos: %v", err) + } + if _, err := tdb.Exec(`insert into repo_records (did, collection, rkey, cid) values + ('did:plc:akshay', 'sh.tangled.repo', '3kspindlerkey00a', 'bafyone'), + ('did:plc:boltless', 'sh.tangled.repo', '3kspindlerkey00b', 'bafytwo') + `); err != nil { + t.Fatalf("seed records: %v", err) + } + tdb.Close() + + if err := nudgeTapForResync(ctx, d, tapPath, logger); err != nil { + t.Fatalf("nudgeTapForResync: %v", err) + } + + if got := tapRecordCount(t, tapPath); got != 0 { + t.Errorf("expected repo_records cleared, got %d", got) + } + if got := tapRepoState(t, tapPath, "did:plc:akshay"); got != "desynchronized" { + t.Errorf("active should flip to desynchronized, got %s", got) + } + if got := tapRepoState(t, tapPath, "did:plc:boltless"); got != "desynchronized" { + t.Errorf("error should flip to desynchronized, got %s", got) + } + if got := tapRepoState(t, tapPath, "did:plc:limpet"); got != "pending" { + t.Errorf("pending should not be touched, got %s", got) + } + + tdb2, err := sql.Open("sqlite3", tapPath) + if err != nil { + t.Fatalf("reopen tap db: %v", err) + } + if _, err := tdb2.Exec(`update repos set state = 'active' where did = 'did:plc:akshay'`); err != nil { + t.Fatalf("reseed: %v", err) + } + tdb2.Close() + + if err := nudgeTapForResync(ctx, d, tapPath, logger); err != nil { + t.Fatalf("nudgeTapForResync second run: %v", err) + } + if got := tapRepoState(t, tapPath, "did:plc:akshay"); got != "active" { + t.Errorf("idempotent re-run should not touch state, got %s", got) + } + + var marked int + if err := d.QueryRow( + `select count(*) from migrations where name = ?`, + "force-tap-repo-resync-v1", + ).Scan(&marked); err != nil { + t.Fatalf("query migrations: %v", err) + } + if marked != 1 { + t.Errorf("expected flag recorded exactly once, got %d", marked) + } +} + +func TestNudgeTapForResync_MissingDB(t *testing.T) { + ctx := context.Background() + logger := slog.New(slog.NewTextHandler(io.Discard, nil)) + d := newTestSpindleDB(t) + + missing := filepath.Join(t.TempDir(), "absent.db") + + if err := nudgeTapForResync(ctx, d, missing, logger); err != nil { + t.Fatalf("missing tap db should succeed: %v", err) + } + + var marked int + if err := d.QueryRow( + `select count(*) from migrations where name = ?`, + "force-tap-repo-resync-v1", + ).Scan(&marked); err != nil { + t.Fatalf("query migrations: %v", err) + } + if marked != 1 { + t.Errorf("expected flag recorded even when tap db absent, got %d", marked) + } +} + +func TestNudgeTapForResync_EmptyPath(t *testing.T) { + ctx := context.Background() + logger := slog.New(slog.NewTextHandler(io.Discard, nil)) + d := newTestSpindleDB(t) + + if err := nudgeTapForResync(ctx, d, "", logger); err == nil { + t.Errorf("expected error for empty tap db path") + } + + var marked int + if err := d.QueryRow( + `select count(*) from migrations where name = ?`, + "force-tap-repo-resync-v1", + ).Scan(&marked); err != nil { + t.Fatalf("query migrations: %v", err) + } + if marked != 0 { + t.Errorf("empty path should not mark flag, got %d", marked) + } +} + +func TestRunStartupMigrations_NonEmbedSkipsTapNudge(t *testing.T) { + ctx := context.Background() + logger := slog.New(slog.NewTextHandler(io.Discard, nil)) + d := newTestSpindleDB(t) + + if err := runStartupMigrations(ctx, d, false, "", logger); err != nil { + t.Fatalf("non-embed should not error on empty path: %v", err) + } + + var marked int + if err := d.QueryRow( + `select count(*) from migrations where name = ?`, + "force-tap-repo-resync-v1", + ).Scan(&marked); err != nil { + t.Fatalf("query migrations: %v", err) + } + if marked != 0 { + t.Errorf("non-embed mode should skip tap nudge flag, got %d", marked) + } +} + +func TestCleanupOrphanRepos_DeletesWhenSiblingExists(t *testing.T) { + ctx := context.Background() + logger := slog.New(slog.NewTextHandler(io.Discard, nil)) + d := newTestSpindleDB(t) + + owner := "did:plc:akshay" + if _, err := d.Exec(`insert into repos (knot, owner, rkey, repo_did, created_at) values + ('k', ?, 'legacy_name', null, null), + ('k', ?, '3kspindlerkey00a', 'did:plc:boltless', '2024-01-01T00:00:00Z')`, + owner, owner); err != nil { + t.Fatalf("seed: %v", err) + } + + if err := cleanupOrphanRepos(ctx, d, logger); err != nil { + t.Fatalf("cleanupOrphanRepos: %v", err) + } + + var nullCount int + if err := d.QueryRow(`select count(*) from repos where repo_did is null`).Scan(&nullCount); err != nil { + t.Fatalf("null count: %v", err) + } + if nullCount != 0 { + t.Errorf("orphan should be deleted when sibling exists, got %d remaining", nullCount) + } + + var sibCount int + if err := d.QueryRow(`select count(*) from repos where repo_did is not null`).Scan(&sibCount); err != nil { + t.Fatalf("sibling count: %v", err) + } + if sibCount != 1 { + t.Errorf("sibling row should be preserved, got %d", sibCount) + } +} + +func TestCleanupOrphanRepos_KeepsWhenAlone(t *testing.T) { + ctx := context.Background() + logger := slog.New(slog.NewTextHandler(io.Discard, nil)) + d := newTestSpindleDB(t) + + owner := "did:plc:akshay" + if _, err := d.Exec(`insert into repos (knot, owner, rkey, repo_did, created_at) values + ('k', ?, 'legacy_name', null, null)`, owner); err != nil { + t.Fatalf("seed: %v", err) + } + + if err := cleanupOrphanRepos(ctx, d, logger); err != nil { + t.Fatalf("cleanupOrphanRepos: %v", err) + } + + var remaining int + if err := d.QueryRow(`select count(*) from repos where owner = ?`, owner).Scan(&remaining); err != nil { + t.Fatalf("count: %v", err) + } + if remaining != 1 { + t.Errorf("orphan with no sibling should be kept (preserves owner registration), got %d", remaining) + } +} + +func TestCleanupOrphanRepos_PerOwnerScope(t *testing.T) { + ctx := context.Background() + logger := slog.New(slog.NewTextHandler(io.Discard, nil)) + d := newTestSpindleDB(t) + + ownerA := "did:plc:akshay" + ownerB := "did:plc:limpet" + if _, err := d.Exec(`insert into repos (knot, owner, rkey, repo_did, created_at) values + ('k', ?, 'legacy_a', null, null), + ('k', ?, '3krealkey', 'did:plc:boltless', '2024-01-01T00:00:00Z'), + ('k', ?, 'legacy_b', null, null)`, + ownerA, ownerA, ownerB); err != nil { + t.Fatalf("seed: %v", err) + } + + if err := cleanupOrphanRepos(ctx, d, logger); err != nil { + t.Fatalf("cleanupOrphanRepos: %v", err) + } + + var ownerARows, ownerBRows int + if err := d.QueryRow(`select count(*) from repos where owner = ?`, ownerA).Scan(&ownerARows); err != nil { + t.Fatalf("count A: %v", err) + } + if ownerARows != 1 { + t.Errorf("ownerA: orphan should be deleted (sibling exists), expected 1 row, got %d", ownerARows) + } + if err := d.QueryRow(`select count(*) from repos where owner = ?`, ownerB).Scan(&ownerBRows); err != nil { + t.Fatalf("count B: %v", err) + } + if ownerBRows != 1 { + t.Errorf("ownerB: orphan should be kept (no sibling), expected 1 row, got %d", ownerBRows) + } +} + +func TestCleanupOrphanRepos_EmptyStringRepoDid(t *testing.T) { + ctx := context.Background() + logger := slog.New(slog.NewTextHandler(io.Discard, nil)) + d := newTestSpindleDB(t) + + owner := "did:plc:akshay" + if _, err := d.Exec(`insert into repos (knot, owner, rkey, repo_did, created_at) values + ('k', ?, 'legacy_empty', '', null), + ('k', ?, '3krealkey', 'did:plc:boltless', '2024-01-01T00:00:00Z')`, + owner, owner); err != nil { + t.Fatalf("seed: %v", err) + } + + if err := cleanupOrphanRepos(ctx, d, logger); err != nil { + t.Fatalf("cleanupOrphanRepos: %v", err) + } + + var emptyCount int + if err := d.QueryRow(`select count(*) from repos where coalesce(repo_did, '') = ''`).Scan(&emptyCount); err != nil { + t.Fatalf("empty count: %v", err) + } + if emptyCount != 0 { + t.Errorf("empty-string repo_did orphan should be deleted when sibling exists, got %d remaining", emptyCount) } } diff --git a/spindle/tapclient.go b/spindle/tapclient.go --- a/spindle/tapclient.go +++ b/spindle/tapclient.go @@ -133,6 +133,12 @@ return fmt.Errorf("add repo: %w", err) } + legacyName := "" + if record.Name != nil { + legacyName = *record.Name + } + migrateLegacyRepoSecrets(ctx, t.spindle.db, t.spindle.vault, l, ownerDid, legacyName, rkey, repoDid) + if removed, err := t.spindle.db.CollapseRepoSiblings(ownerDid, repoDid); err != nil { l.Warn("collapse rename siblings failed", "err", err) } else if removed > 0 { @@ -332,11 +338,14 @@ cutoff := time.Now().Add(-pendingCollabTTL) t.pendingMu.Lock() defer t.pendingMu.Unlock() + expired := 0 for did, list := range t.pendingCollabs { kept := list[:0] for _, p := range list { if !p.at.Before(cutoff) { kept = append(kept, p) + } else { + expired++ } } if len(kept) == 0 { @@ -344,5 +353,8 @@ } else { t.pendingCollabs[did] = kept } + } + if expired > 0 { + t.logger.Warn("expired buffered collaborator events without matching repo arrival", "count", expired, "ttl", pendingCollabTTL) } } diff --git a/spindle/db/repos.go b/spindle/db/repos.go --- a/spindle/db/repos.go +++ b/spindle/db/repos.go @@ -36,10 +36,18 @@ `delete from repos where owner = ? and repo_did = ? - and created_at is not null - and created_at < ( - select max(created_at) from repos - where owner = ? and repo_did = ? and created_at is not null + and ( + (created_at is null and exists ( + select 1 from repos r2 + where r2.owner = repos.owner + and r2.repo_did = repos.repo_did + and r2.created_at is not null + and r2.rkey <> repos.rkey + )) + or (created_at is not null and created_at < ( + select max(created_at) from repos + where owner = ? and repo_did = ? and created_at is not null + )) )`, owner.String(), repoDid.String(), owner.String(), repoDid.String(), ) diff --git a/spindle/db/repos_test.go b/spindle/db/repos_test.go new file mode 100644 --- /dev/null +++ b/spindle/db/repos_test.go @@ -0,0 +1,143 @@ +package db + +import ( + "context" + "path/filepath" + "testing" + + "github.com/bluesky-social/indigo/atproto/syntax" +) + +func newTestDB(t *testing.T) *DB { + t.Helper() + d, err := Make(context.Background(), filepath.Join(t.TempDir(), "spindle.db")) + if err != nil { + t.Fatalf("Make: %v", err) + } + t.Cleanup(func() { d.Close() }) + return d +} + +func TestCollapseRepoSiblings_DeletesStaleNullCreatedAtWithDifferentRkey(t *testing.T) { + d := newTestDB(t) + owner := syntax.DID("did:plc:akshay") + repoDid := syntax.DID("did:plc:boltless") + + if _, err := d.Exec(`insert into repos (knot, owner, rkey, repo_did, created_at) values + ('k', ?, 'stale-bogus-rkey', ?, null), + ('k', ?, 'fresh-pds-rkey', ?, '2024-06-01T00:00:00Z')`, + owner.String(), repoDid.String(), + owner.String(), repoDid.String()); err != nil { + t.Fatalf("seed: %v", err) + } + + n, err := d.CollapseRepoSiblings(owner, repoDid) + if err != nil { + t.Fatalf("CollapseRepoSiblings: %v", err) + } + if n != 1 { + t.Errorf("expected 1 stale row deleted, got %d", n) + } + + var rkey string + if err := d.QueryRow(`select rkey from repos where owner = ? and repo_did = ?`, + owner.String(), repoDid.String()).Scan(&rkey); err != nil { + t.Fatalf("query: %v", err) + } + if rkey != "fresh-pds-rkey" { + t.Errorf("expected fresh row preserved, got rkey=%q", rkey) + } +} + +func TestCollapseRepoSiblings_KeepsNullCreatedAtWhenAlone(t *testing.T) { + d := newTestDB(t) + owner := syntax.DID("did:plc:akshay") + repoDid := syntax.DID("did:plc:boltless") + + if _, err := d.Exec(`insert into repos (knot, owner, rkey, repo_did, created_at) values + ('k', ?, 'sole-row', ?, null)`, + owner.String(), repoDid.String()); err != nil { + t.Fatalf("seed: %v", err) + } + + n, err := d.CollapseRepoSiblings(owner, repoDid) + if err != nil { + t.Fatalf("CollapseRepoSiblings: %v", err) + } + if n != 0 { + t.Errorf("expected 0 deletions when only NULL row exists, got %d", n) + } + + var count int + if err := d.QueryRow(`select count(*) from repos where owner = ?`, owner.String()).Scan(&count); err != nil { + t.Fatalf("count: %v", err) + } + if count != 1 { + t.Errorf("sole NULL row should survive, got %d remaining", count) + } +} + +func TestCollapseRepoSiblings_OlderTimestampLoses(t *testing.T) { + d := newTestDB(t) + owner := syntax.DID("did:plc:akshay") + repoDid := syntax.DID("did:plc:boltless") + + if _, err := d.Exec(`insert into repos (knot, owner, rkey, repo_did, created_at) values + ('k', ?, 'older-rkey', ?, '2024-01-01T00:00:00Z'), + ('k', ?, 'newer-rkey', ?, '2024-06-01T00:00:00Z')`, + owner.String(), repoDid.String(), + owner.String(), repoDid.String()); err != nil { + t.Fatalf("seed: %v", err) + } + + n, err := d.CollapseRepoSiblings(owner, repoDid) + if err != nil { + t.Fatalf("CollapseRepoSiblings: %v", err) + } + if n != 1 { + t.Errorf("expected older row collapsed, got %d", n) + } + + var rkey string + if err := d.QueryRow(`select rkey from repos where owner = ? and repo_did = ?`, + owner.String(), repoDid.String()).Scan(&rkey); err != nil { + t.Fatalf("query: %v", err) + } + if rkey != "newer-rkey" { + t.Errorf("expected newer row preserved, got rkey=%q", rkey) + } +} + +func TestCollapseRepoSiblings_KeepsNullRowWithMatchingRkey(t *testing.T) { + d := newTestDB(t) + owner := syntax.DID("did:plc:akshay") + repoDid := syntax.DID("did:plc:boltless") + + if _, err := d.Exec(`insert into repos (knot, owner, rkey, repo_did, created_at) values + ('k', ?, 'matched-rkey', ?, null)`, + owner.String(), repoDid.String()); err != nil { + t.Fatalf("seed: %v", err) + } + + if err := d.AddRepo(Repo{ + Knot: "k", + Owner: owner, + Rkey: "matched-rkey", + RepoDid: repoDid, + CreatedAt: "2024-06-01T00:00:00Z", + }); err != nil { + t.Fatalf("AddRepo upsert: %v", err) + } + + if _, err := d.CollapseRepoSiblings(owner, repoDid); err != nil { + t.Fatalf("CollapseRepoSiblings: %v", err) + } + + var count int + if err := d.QueryRow(`select count(*) from repos where owner = ?`, owner.String()).Scan(&count); err != nil { + t.Fatalf("count: %v", err) + } + if count != 1 { + t.Errorf("upserted row should be the single survivor, got %d", count) + } +} -- tangled.sh