From ba857c7c1bcda26a318aeb2f15f12873a97bd928 Mon Sep 17 00:00:00 2001 From: Lewis Date: Tue, 02 Jun 2026 11:09:18 +0000 Subject: [PATCH] knotserver: backfill members & collaborators from casbin Lewis: May this revision serve well! --- knotserver/backfill.go | 142 ++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++-------------------------------------------------------------------- knotserver/backfill_test.go | 266 ++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++ knotserver/migrate.go | 9 +++++---- knotserver/server.go | 14 +++++++++++--- knotserver/db/member.go | 23 ----------------------- 5 file(s) changed, 356 insertion(s)(+), 98 deletion(s)(-) diff --git a/knotserver/backfill.go b/knotserver/backfill.go --- a/knotserver/backfill.go +++ b/knotserver/backfill.go @@ -4,34 +4,29 @@ "context" "fmt" "log/slog" - "time" + "slices" - comatproto "github.com/bluesky-social/indigo/api/atproto" "github.com/bluesky-social/indigo/atproto/syntax" - "github.com/bluesky-social/indigo/xrpc" - "tangled.org/core/api/tangled" - "tangled.org/core/idresolver" "tangled.org/core/knotserver/db" "tangled.org/core/rbac" ) const ( - knotMembersBackfillMigration = "backfill-knot-members-from-pds-v2" - knotMembersBackfillPerOwner = 30 * time.Second + collaboratorBackfillMigration = "backfill-collaborators-from-casbin-v1" + knotMemberBackfillMigration = "backfill-knot-members-from-casbin-v1" ) func BackfillKnotMembers( ctx context.Context, d *db.DB, e *rbac.Enforcer, - resolver *idresolver.Resolver, - hostname string, + ownerDid string, logger *slog.Logger, ) error { - l := logger.With("migration", knotMembersBackfillMigration) + l := logger.With("migration", knotMemberBackfillMigration) - applied, err := d.IsMigrationApplied(knotMembersBackfillMigration) + applied, err := d.IsMigrationApplied(knotMemberBackfillMigration) if err != nil { return fmt.Errorf("check migration applied: %w", err) } @@ -39,94 +34,105 @@ return nil } + owner, err := syntax.ParseDID(ownerDid) + if err != nil { + return fmt.Errorf("invalid knot owner DID %q: %w", ownerDid, err) + } + + members, err := e.GetKnotUsersByRole("server:member", rbac.ThisServer) + if err != nil { + return fmt.Errorf("list members: %w", err) + } owners, err := e.GetKnotUsersByRole("server:owner", rbac.ThisServer) if err != nil { return fmt.Errorf("list owners: %w", err) } var rows []db.KnotMember - for _, owner := range owners { - ownerCtx, cancel := context.WithTimeout(ctx, knotMembersBackfillPerOwner) - ownerRows, err := fetchOwnerKnotMembers(ownerCtx, resolver, hostname, owner, l) - cancel() - if err != nil { - l.Warn("skipping owner during backfill", "owner", owner, "err", err) + for _, candidate := range members { + if slices.Contains(owners, candidate) { continue } - rows = append(rows, ownerRows...) - } - - for _, m := range rows { - if err := e.AddKnotMember(rbac.ThisServer, m.Subject.String()); err != nil { - return fmt.Errorf("grant ACL for %s: %w", m.Subject, err) + subject, err := syntax.ParseDID(candidate) + if err != nil { + l.Warn("skipping member with invalid DID", "candidate", candidate, "err", err) + continue } + rows = append(rows, db.KnotMember{Did: owner, Subject: subject}) } - if err := d.ApplyKnotMembersBackfill(ctx, rows, knotMembersBackfillMigration); err != nil { + if err := d.ApplyKnotMemberBackfill(ctx, rows, knotMemberBackfillMigration); err != nil { return fmt.Errorf("apply backfill: %w", err) } - l.Info("backfilled knot members", "count", len(rows), "owners", len(owners)) + l.Info("backfilled knot members from casbin", "count", len(rows)) return nil } -func fetchOwnerKnotMembers( +func BackfillCollaborators( ctx context.Context, - resolver *idresolver.Resolver, - hostname string, - owner string, - l *slog.Logger, -) ([]db.KnotMember, error) { - ownerDid, err := syntax.ParseDID(owner) + d *db.DB, + e *rbac.Enforcer, + logger *slog.Logger, + markApplied bool, +) error { + l := logger.With("migration", collaboratorBackfillMigration) + + applied, err := d.IsMigrationApplied(collaboratorBackfillMigration) if err != nil { - return nil, fmt.Errorf("invalid owner DID %q: %w", owner, err) + return fmt.Errorf("check migration applied: %w", err) + } + if applied { + return nil } - ident, err := resolver.ResolveIdent(ctx, owner) + repoDids, err := d.ListRepoDids() if err != nil { - return nil, fmt.Errorf("resolve %s: %w", owner, err) + return fmt.Errorf("list repos: %w", err) } - client := &xrpc.Client{Host: ident.PDSEndpoint()} - var ( - rows []db.KnotMember - cursor string - ) - for { - out, err := comatproto.RepoListRecords(ctx, client, tangled.KnotMemberNSID, cursor, 100, owner, false) + var rows []db.Collaborator + for _, repoDid := range repoDids { + ownerDid, _, err := d.GetRepoKeyOwner(repoDid) if err != nil { - return nil, fmt.Errorf("list records: %w", err) + l.Warn("skipping repo during collaborator backfill", "repoDid", repoDid, "err", err) + continue } - for _, rec := range out.Records { - m, ok := rec.Value.Val.(*tangled.KnotMember) - if !ok || m.Domain != hostname { - continue - } - subject, err := syntax.ParseDID(m.Subject) + + repo, err := syntax.ParseDID(repoDid) + if err != nil { + l.Warn("skipping repo with invalid DID", "repoDid", repoDid, "err", err) + continue + } + owner, err := syntax.ParseDID(ownerDid) + if err != nil { + l.Warn("skipping repo with invalid owner DID", "repoDid", repoDid, "owner", ownerDid, "err", err) + continue + } + + collaborators, err := e.GetUserByRoleInRepo("repo:collaborator", rbac.ThisServer, repoDid) + if err != nil { + return fmt.Errorf("list collaborators for %s: %w", repoDid, err) + } + + for _, candidate := range collaborators { + subject, err := syntax.ParseDID(candidate) if err != nil { - l.Warn("invalid subject DID in record, skipping", "uri", rec.Uri, "err", err) + l.Warn("skipping collaborator with invalid DID", "repoDid", repoDid, "candidate", candidate, "err", err) continue } - uri, err := syntax.ParseATURI(rec.Uri) - if err != nil { - l.Warn("invalid AT URI in record, skipping", "uri", rec.Uri, "err", err) - continue - } - rkey := uri.RecordKey().String() - if rkey == "" { - l.Warn("empty rkey in AT URI, skipping", "uri", rec.Uri) - continue - } - rows = append(rows, db.KnotMember{ - Did: ownerDid, - Rkey: rkey, + rows = append(rows, db.Collaborator{ + RepoDid: repo, Subject: subject, + AddedBy: owner, }) } - if out.Cursor == nil || *out.Cursor == "" || *out.Cursor == cursor { - break - } - cursor = *out.Cursor } - return rows, nil + + if err := d.ApplyCollaboratorBackfill(ctx, rows, collaboratorBackfillMigration, markApplied); err != nil { + return fmt.Errorf("apply backfill: %w", err) + } + + l.Info("backfilled collaborators from casbin", "count", len(rows), "repos", len(repoDids), "marked", markApplied) + return nil } diff --git a/knotserver/backfill_test.go b/knotserver/backfill_test.go new file mode 100644 --- /dev/null +++ b/knotserver/backfill_test.go @@ -0,0 +1,266 @@ +package knotserver + +import ( + "context" + "io" + "log/slog" + "path/filepath" + "testing" + + "github.com/bluesky-social/indigo/atproto/syntax" + + "tangled.org/core/knotserver/db" + "tangled.org/core/rbac" +) + +const ( + bfOwner = "did:plc:akshay" + bfCollab = "did:plc:boltless" + bfRepo = "did:plc:limpet" +) + +func newBackfillEnv(t *testing.T) (*db.DB, *rbac.Enforcer) { + t.Helper() + dir := t.TempDir() + d, err := db.Setup(context.Background(), filepath.Join(dir, "knot.db")) + if err != nil { + t.Fatalf("db.Setup: %v", err) + } + e, err := rbac.NewEnforcer(filepath.Join(dir, "rbac.db")) + if err != nil { + t.Fatalf("NewEnforcer: %v", err) + } + if err := e.AddKnot(rbac.ThisServer); err != nil { + t.Fatalf("AddKnot: %v", err) + } + if err := e.AddKnotOwner(rbac.ThisServer, bfOwner); err != nil { + t.Fatalf("AddKnotOwner: %v", err) + } + return d, e +} + +func seedCasbinRepo(t *testing.T, d *db.DB, e *rbac.Enforcer, repoDid string, collaborators ...string) { + t.Helper() + if err := d.StoreRepoKey(repoDid, []byte("signing"), bfOwner, "reponame"); err != nil { + t.Fatalf("StoreRepoKey: %v", err) + } + if err := e.AddRepo(bfOwner, rbac.ThisServer, repoDid); err != nil { + t.Fatalf("AddRepo: %v", err) + } + for _, c := range collaborators { + if err := e.AddCollaborator(c, rbac.ThisServer, repoDid); err != nil { + t.Fatalf("AddCollaborator %s: %v", c, err) + } + } +} + +func runBackfill(t *testing.T, d *db.DB, e *rbac.Enforcer) { + t.Helper() + logger := slog.New(slog.NewTextHandler(io.Discard, nil)) + if err := BackfillCollaborators(context.Background(), d, e, logger, true); err != nil { + t.Fatalf("BackfillCollaborators: %v", err) + } +} + +func runMemberBackfill(t *testing.T, d *db.DB, e *rbac.Enforcer) { + t.Helper() + logger := slog.New(slog.NewTextHandler(io.Discard, nil)) + if err := BackfillKnotMembers(context.Background(), d, e, bfOwner, logger); err != nil { + t.Fatalf("BackfillKnotMembers: %v", err) + } +} + +func TestBackfillCollaborators_FoldsCasbinAndExcludesOwner(t *testing.T) { + d, e := newBackfillEnv(t) + seedCasbinRepo(t, d, e, bfRepo, bfCollab) + + runBackfill(t, d, e) + + list, _, err := db.ListCollaborators(d, syntax.DID(bfRepo), db.ListPage{Limit: db.ListMaxLimit}) + if err != nil { + t.Fatalf("ListCollaborators: %v", err) + } + if len(list) != 1 { + t.Fatalf("collaborators = %+v, want exactly one (owner must be excluded)", list) + } + if list[0].Subject != syntax.DID(bfCollab) { + t.Errorf("subject = %s, want %s", list[0].Subject, bfCollab) + } + if list[0].AddedBy != syntax.DID(bfOwner) { + t.Errorf("addedBy = %s, want owner %s", list[0].AddedBy, bfOwner) + } +} + +func TestBackfillCollaborators_OneTimeAndNonDestructive(t *testing.T) { + d, e := newBackfillEnv(t) + seedCasbinRepo(t, d, e, bfRepo, bfCollab) + + runBackfill(t, d, e) + + if err := e.AddCollaborator("did:plc:scallop", rbac.ThisServer, bfRepo); err != nil { + t.Fatalf("post-migration casbin add: %v", err) + } + runBackfill(t, d, e) + + list, _, err := db.ListCollaborators(d, syntax.DID(bfRepo), db.ListPage{Limit: db.ListMaxLimit}) + if err != nil { + t.Fatalf("ListCollaborators: %v", err) + } + if len(list) != 1 { + t.Fatalf("collaborators = %d, want 1; backfill must run once and never resurrect later casbin state", len(list)) + } + if list[0].Subject != syntax.DID(bfCollab) { + t.Errorf("subject = %s, want original %s preserved", list[0].Subject, bfCollab) + } +} + +func TestBackfillCollaborators_LeavesMembersUntouched(t *testing.T) { + d, e := newBackfillEnv(t) + seedCasbinRepo(t, d, e, bfRepo, bfCollab) + + owner := syntax.DID(bfOwner) + member := syntax.DID("did:plc:whelk") + if err := db.AddKnotMemberDirect(d, owner, member); err != nil { + t.Fatalf("seed member: %v", err) + } + if err := e.AddKnotMember(rbac.ThisServer, member.String()); err != nil { + t.Fatalf("seed member acl: %v", err) + } + + runBackfill(t, d, e) + + members, _, err := db.ListKnotMembers(d, db.ListPage{Limit: db.ListMaxLimit}) + if err != nil { + t.Fatalf("ListKnotMembers: %v", err) + } + if len(members) != 1 || members[0].Subject != member { + t.Fatalf("members = %+v, want the seeded member preserved", members) + } + + collabs, _, err := db.ListCollaborators(d, syntax.DID(bfRepo), db.ListPage{Limit: db.ListMaxLimit}) + if err != nil { + t.Fatalf("ListCollaborators: %v", err) + } + if len(collabs) != 1 || collabs[0].Subject != syntax.DID(bfCollab) { + t.Errorf("collaborators = %+v, want only the casbin collaborator", collabs) + } +} + +func TestBackfillCollaborators_UnmarkedRunDefersMarker(t *testing.T) { + d, e := newBackfillEnv(t) + seedCasbinRepo(t, d, e, bfRepo, bfCollab) + + logger := slog.New(slog.NewTextHandler(io.Discard, nil)) + if err := BackfillCollaborators(context.Background(), d, e, logger, false); err != nil { + t.Fatalf("unmarked backfill: %v", err) + } + + list, _, err := db.ListCollaborators(d, syntax.DID(bfRepo), db.ListPage{Limit: db.ListMaxLimit}) + if err != nil { + t.Fatalf("ListCollaborators: %v", err) + } + if len(list) != 1 { + t.Fatalf("collaborators = %d, want 1 after unmarked run", len(list)) + } + applied, err := d.IsMigrationApplied(collaboratorBackfillMigration) + if err != nil { + t.Fatalf("IsMigrationApplied: %v", err) + } + if applied { + t.Fatal("unmarked run must not write the migration marker") + } + + if err := e.AddCollaborator("did:plc:scallop", rbac.ThisServer, bfRepo); err != nil { + t.Fatalf("late casbin add: %v", err) + } + runBackfill(t, d, e) + + list, _, err = db.ListCollaborators(d, syntax.DID(bfRepo), db.ListPage{Limit: db.ListMaxLimit}) + if err != nil { + t.Fatalf("ListCollaborators after marked run: %v", err) + } + if len(list) != 2 { + t.Errorf("collaborators = %d, want 2; the marked rerun must fold late casbin state", len(list)) + } + if applied, _ := d.IsMigrationApplied(collaboratorBackfillMigration); !applied { + t.Error("marked run did not write the migration marker") + } +} + +func TestBackfillKnotMembers_FoldsCasbinAndExcludesOwner(t *testing.T) { + d, e := newBackfillEnv(t) + member := syntax.DID("did:plc:whelk") + if err := e.AddKnotMember(rbac.ThisServer, member.String()); err != nil { + t.Fatalf("seed casbin member: %v", err) + } + + runMemberBackfill(t, d, e) + + members, _, err := db.ListKnotMembers(d, db.ListPage{Limit: db.ListMaxLimit}) + if err != nil { + t.Fatalf("ListKnotMembers: %v", err) + } + if len(members) != 1 { + t.Fatalf("members = %+v, want exactly one; the owner must be excluded", members) + } + if members[0].Subject != member { + t.Errorf("subject = %s, want %s", members[0].Subject, member) + } + if members[0].Did != syntax.DID(bfOwner) { + t.Errorf("added by = %s, want owner %s", members[0].Did, bfOwner) + } +} + +func TestBackfillKnotMembers_OneTimeAndNonDestructive(t *testing.T) { + d, e := newBackfillEnv(t) + member := syntax.DID("did:plc:whelk") + if err := e.AddKnotMember(rbac.ThisServer, member.String()); err != nil { + t.Fatalf("seed casbin member: %v", err) + } + + runMemberBackfill(t, d, e) + + if err := e.AddKnotMember(rbac.ThisServer, "did:plc:scallop"); err != nil { + t.Fatalf("post-migration casbin add: %v", err) + } + runMemberBackfill(t, d, e) + + members, _, err := db.ListKnotMembers(d, db.ListPage{Limit: db.ListMaxLimit}) + if err != nil { + t.Fatalf("ListKnotMembers: %v", err) + } + if len(members) != 1 || members[0].Subject != member { + t.Fatalf("members = %+v, want only the original member; backfill must run once", members) + } + if applied, _ := d.IsMigrationApplied(knotMemberBackfillMigration); !applied { + t.Error("member backfill did not write its migration marker") + } +} + +func TestBackfillCollaborators_EmptyMarksApplied(t *testing.T) { + d, e := newBackfillEnv(t) + seedCasbinRepo(t, d, e, bfRepo) + + runBackfill(t, d, e) + + applied, err := d.IsMigrationApplied(collaboratorBackfillMigration) + if err != nil { + t.Fatalf("IsMigrationApplied: %v", err) + } + if !applied { + t.Fatal("migration not marked applied after a zero-collaborator backfill; it would re-scan every boot") + } + + if err := e.AddCollaborator(bfCollab, rbac.ThisServer, bfRepo); err != nil { + t.Fatalf("post-migration casbin add: %v", err) + } + runBackfill(t, d, e) + + list, _, err := db.ListCollaborators(d, syntax.DID(bfRepo), db.ListPage{Limit: db.ListMaxLimit}) + if err != nil { + t.Fatalf("ListCollaborators: %v", err) + } + if len(list) != 0 { + t.Errorf("collaborators = %d, want 0; an applied migration must not fold later casbin state", len(list)) + } +} diff --git a/knotserver/migrate.go b/knotserver/migrate.go --- a/knotserver/migrate.go +++ b/knotserver/migrate.go @@ -70,11 +70,11 @@ return repos } -func migrateReposOnStartup(ctx context.Context, c *config.Config, d *db.DB, e *rbac.Enforcer, n *notifier.Notifier, logger *slog.Logger) { +func migrateReposOnStartup(ctx context.Context, c *config.Config, d *db.DB, e *rbac.Enforcer, n *notifier.Notifier, logger *slog.Logger) bool { repos := scanLegacyRepos(c.Repo.ScanPath, logger) if len(repos) == 0 { logger.Info("no legacy repos found, migration complete") - return + return true } logger.Info("starting legacy repo migration", "count", len(repos)) @@ -90,7 +90,7 @@ select { case <-ctx.Done(): logger.Info("migration interrupted by shutdown", "migrated", migrated, "remaining", len(repos)-migrated) - return + return false default: } @@ -103,6 +103,7 @@ } logger.Info("legacy repo migration complete", "migrated", migrated, "total", len(repos), "duration", time.Since(start)) + return migrated == len(repos) } func migrateOneRepo( @@ -131,7 +132,7 @@ } if err := rewriteRBACPolicies(e, repo.ownerDid, repo.repoName, repoDid, l); err != nil { - l.Error("RBAC rewrite failed (non-fatal)", "error", err) + return fmt.Errorf("rewriting RBAC policies: %w", err) } newPath := filepath.Join(c.Repo.ScanPath, repoDid) diff --git a/knotserver/server.go b/knotserver/server.go --- a/knotserver/server.go +++ b/knotserver/server.go @@ -108,8 +108,11 @@ resolver := idresolver.DefaultResolver(c.Server.PlcUrl) - if err := BackfillKnotMembers(ctx, db, e, resolver, c.Server.Hostname, logger); err != nil { - logger.Warn("knot members backfill failed, continuing", "err", err) + if err := BackfillKnotMembers(ctx, db, e, c.Server.Owner, logger); err != nil { + logger.Warn("knot member backfill failed, continuing", "err", err) + } + if err := BackfillCollaborators(ctx, db, e, logger, false); err != nil { + logger.Warn("collaborator backfill failed, continuing", "err", err) } // probe and initialise the sandbox backend. @@ -140,7 +143,12 @@ sb = &sandbox.NoopBackend{} } - go migrateReposOnStartup(ctx, c, db, e, ¬ifier, log.SubLogger(logger, "migrate")) + go func() { + migrated := migrateReposOnStartup(ctx, c, db, e, ¬ifier, log.SubLogger(logger, "migrate")) + if err := BackfillCollaborators(ctx, db, e, logger, migrated); err != nil { + logger.Warn("collaborator backfill failed, continuing", "err", err) + } + }() mux, err := Setup(ctx, c, db, e, jc, ¬ifier, resolver, sb) if err != nil { diff --git a/knotserver/db/member.go b/knotserver/db/member.go --- a/knotserver/db/member.go +++ b/knotserver/db/member.go @@ -94,29 +94,6 @@ ) } -func (d *DB) ApplyKnotMembersBackfill(ctx context.Context, rows []KnotMember, migrationName string) error { - conn, err := d.db.Conn(ctx) - if err != nil { - return err - } - defer conn.Close() - - return orm.RunMigration(conn, d.logger, migrationName, func(tx *sql.Tx) error { - for _, m := range rows { - if err := AddDid(tx, m.Subject.String()); err != nil { - return err - } - if _, err := tx.ExecContext(ctx, - `insert or ignore into knot_members (did, rkey, subject) values (?, ?, ?)`, - m.Did, m.Rkey, m.Subject, - ); err != nil { - return err - } - } - return nil - }) -} - func AddKnotMember(q DBTX, member KnotMember) error { _, err := q.Exec( `insert or ignore into knot_members (did, rkey, subject) values (?, ?, ?)`, -- tangled.sh