Monorepo for Tangled
spindle: sync knot-managed collaborators from the acl firehose
Signed-off-by: dawn <dawn@tangled.org>
+174
+21
spindle/db/collaborators.go
+21
spindle/db/collaborators.go
···
26
26
return err
27
27
}
28
28
29
+
func (d *DB) AddKnotCollaborator(repoDid, subject syntax.DID) error {
30
+
_, err := d.Exec(
31
+
`insert into repo_collaborators (owner_did, rkey, subject, repo_did)
32
+
values (?, ?, ?, ?)
33
+
on conflict(owner_did, rkey) do nothing`,
34
+
repoDid.String(), subject.String(), subject.String(), repoDid.String(),
35
+
)
36
+
return err
37
+
}
38
+
29
39
func scanCollab(row interface{ Scan(...any) error }) (*RepoCollaborator, error) {
30
40
var owner, rkey, subject, repoDid string
31
41
if err := row.Scan(&owner, &rkey, &subject, &repoDid); err != nil {
···
57
67
}
58
68
if n == 0 {
59
69
return sql.ErrNoRows
70
+
}
71
+
return nil
72
+
}
73
+
74
+
func (d *DB) DeleteRepoCollaboratorBySubjectRepo(subject, repoDid syntax.DID) error {
75
+
_, err := d.Exec(
76
+
`delete from repo_collaborators where repo_did = ? and subject = ?`,
77
+
repoDid.String(), subject.String(),
78
+
)
79
+
if err != nil {
80
+
return fmt.Errorf("delete collaborator %s on %s: %w", subject, repoDid, err)
60
81
}
61
82
return nil
62
83
}
+94
spindle/db/collaborators_test.go
+94
spindle/db/collaborators_test.go
···
1
+
package db
2
+
3
+
import (
4
+
"testing"
5
+
6
+
"github.com/bluesky-social/indigo/atproto/syntax"
7
+
)
8
+
9
+
func subjectsOf(t *testing.T, d *DB, repoDid syntax.DID) []syntax.DID {
10
+
t.Helper()
11
+
rows, err := d.ListCollaboratorsByRepoDid(repoDid)
12
+
if err != nil {
13
+
t.Fatalf("ListCollaboratorsByRepoDid: %v", err)
14
+
}
15
+
out := make([]syntax.DID, 0, len(rows))
16
+
for _, r := range rows {
17
+
out = append(out, r.Subject)
18
+
}
19
+
return out
20
+
}
21
+
22
+
func TestAddKnotCollaborator_PersistsAndIsIdempotent(t *testing.T) {
23
+
d := newTestDB(t)
24
+
repo := syntax.DID("did:plc:repo")
25
+
bob := syntax.DID("did:plc:bob")
26
+
27
+
if err := d.AddKnotCollaborator(repo, bob); err != nil {
28
+
t.Fatalf("add: %v", err)
29
+
}
30
+
if err := d.AddKnotCollaborator(repo, bob); err != nil {
31
+
t.Fatalf("re-add: %v", err)
32
+
}
33
+
34
+
got := subjectsOf(t, d, repo)
35
+
if len(got) != 1 || got[0] != bob {
36
+
t.Fatalf("collaborators = %v, want exactly [bob]", got)
37
+
}
38
+
}
39
+
40
+
func TestDeleteRepoCollaboratorBySubjectRepo(t *testing.T) {
41
+
d := newTestDB(t)
42
+
repo := syntax.DID("did:plc:repo")
43
+
bob := syntax.DID("did:plc:bob")
44
+
carol := syntax.DID("did:plc:carol")
45
+
46
+
if err := d.AddKnotCollaborator(repo, bob); err != nil {
47
+
t.Fatalf("add bob: %v", err)
48
+
}
49
+
if err := d.AddKnotCollaborator(repo, carol); err != nil {
50
+
t.Fatalf("add carol: %v", err)
51
+
}
52
+
53
+
if err := d.DeleteRepoCollaboratorBySubjectRepo(bob, repo); err != nil {
54
+
t.Fatalf("delete bob: %v", err)
55
+
}
56
+
got := subjectsOf(t, d, repo)
57
+
if len(got) != 1 || got[0] != carol {
58
+
t.Fatalf("after removing bob, collaborators = %v, want [carol]", got)
59
+
}
60
+
61
+
if err := d.DeleteRepoCollaboratorBySubjectRepo(bob, repo); err != nil {
62
+
t.Fatalf("idempotent delete: %v", err)
63
+
}
64
+
}
65
+
66
+
func TestKnotCollaborator_NoCollisionAcrossReposAndSubjects(t *testing.T) {
67
+
d := newTestDB(t)
68
+
repoA := syntax.DID("did:plc:repoA")
69
+
repoB := syntax.DID("did:plc:repoB")
70
+
bob := syntax.DID("did:plc:bob")
71
+
carol := syntax.DID("did:plc:carol")
72
+
73
+
for _, c := range []struct{ repo, subj syntax.DID }{
74
+
{repoA, bob}, {repoB, bob}, {repoA, carol},
75
+
} {
76
+
if err := d.AddKnotCollaborator(c.repo, c.subj); err != nil {
77
+
t.Fatalf("add %s/%s: %v", c.repo, c.subj, err)
78
+
}
79
+
}
80
+
81
+
if got := subjectsOf(t, d, repoA); len(got) != 2 {
82
+
t.Errorf("repoA collaborators = %v, want bob+carol", got)
83
+
}
84
+
if got := subjectsOf(t, d, repoB); len(got) != 1 || got[0] != bob {
85
+
t.Errorf("repoB collaborators = %v, want [bob]", got)
86
+
}
87
+
88
+
if err := d.DeleteRepoCollaboratorBySubjectRepo(bob, repoA); err != nil {
89
+
t.Fatalf("delete bob@repoA: %v", err)
90
+
}
91
+
if got := subjectsOf(t, d, repoB); len(got) != 1 || got[0] != bob {
92
+
t.Errorf("repoB after removing bob@repoA = %v, want still [bob]", got)
93
+
}
94
+
}
+59
spindle/server.go
+59
spindle/server.go
···
2
2
3
3
import (
4
4
"context"
5
+
"database/sql"
5
6
_ "embed"
6
7
"encoding/json"
7
8
"errors"
···
24
25
"tangled.org/core/eventstream"
25
26
"tangled.org/core/idresolver"
26
27
"tangled.org/core/jetstream"
28
+
knotdb "tangled.org/core/knotserver/db"
27
29
kgit "tangled.org/core/knotserver/git"
28
30
"tangled.org/core/log"
29
31
"tangled.org/core/notifier"
···
408
410
func (s *Spindle) processKnotStream(ctx context.Context, src eventconsumer.Source, msg eventstream.Event) error {
409
411
l := log.FromContext(ctx).With("handler", "processKnotStream")
410
412
l = l.With("src", src.Key(), "msg.Nsid", msg.Nsid, "msg.Rkey", msg.Rkey)
413
+
if msg.Nsid == knotdb.RepoCollaboratorUpdateNSID {
414
+
return s.ingestKnotCollaborator(ctx, l, src, msg)
415
+
}
411
416
if msg.Nsid == tangled.GitRefUpdateNSID {
412
417
event := tangled.GitRefUpdate{}
413
418
if err := json.Unmarshal(msg.EventJson, &event); err != nil {
···
466
471
l.Info("pipeline triggered", "pipeline", pipelineId.AtUri())
467
472
}
468
473
474
+
return nil
475
+
}
476
+
477
+
func (s *Spindle) ingestKnotCollaborator(ctx context.Context, l *slog.Logger, src eventconsumer.Source, msg eventstream.Event) error {
478
+
var rec knotdb.RepoCollaboratorUpdate
479
+
if err := json.Unmarshal(msg.EventJson, &rec); err != nil {
480
+
l.Error("error unmarshalling collaboratorUpdate", "err", err)
481
+
return err
482
+
}
483
+
484
+
subject, err := syntax.ParseDID(rec.Subject)
485
+
if err != nil {
486
+
l.Info("skipping collaboratorUpdate with malformed subject", "subject", rec.Subject, "err", err)
487
+
return nil
488
+
}
489
+
repoDid, err := syntax.ParseDID(rec.Repo)
490
+
if err != nil {
491
+
l.Info("skipping collaboratorUpdate with malformed repo", "repo", rec.Repo, "err", err)
492
+
return nil
493
+
}
494
+
495
+
repo, err := s.db.GetRepoByDid(repoDid)
496
+
if errors.Is(err, sql.ErrNoRows) {
497
+
l.Info("skipping collaboratorUpdate for unknown repo", "repo", repoDid)
498
+
return nil
499
+
}
500
+
if err != nil {
501
+
return fmt.Errorf("lookup repo %s: %w", repoDid, err)
502
+
}
503
+
if src.Host != repo.Knot {
504
+
l.Warn("dropping collaboratorUpdate from non-owning knot", "src", src.Host, "repoKnot", repo.Knot)
505
+
return nil
506
+
}
507
+
508
+
switch rec.Op {
509
+
case knotdb.AclOpAdd:
510
+
if err := s.e.AddCollaborator(subject.String(), rbac.ThisServer, repoDid.String()); err != nil {
511
+
return fmt.Errorf("add collaborator policy: %w", err)
512
+
}
513
+
if err := s.db.AddKnotCollaborator(repoDid, subject); err != nil {
514
+
return fmt.Errorf("track collaborator: %w", err)
515
+
}
516
+
l.Info("added knot-managed collaborator", "subject", subject, "repo", repoDid)
517
+
case knotdb.AclOpRemove:
518
+
if err := s.e.RemoveCollaborator(subject.String(), rbac.ThisServer, repoDid.String()); err != nil {
519
+
return fmt.Errorf("remove collaborator policy: %w", err)
520
+
}
521
+
if err := s.db.DeleteRepoCollaboratorBySubjectRepo(subject, repoDid); err != nil {
522
+
return fmt.Errorf("delete collaborator row: %w", err)
523
+
}
524
+
l.Info("removed knot-managed collaborator", "subject", subject, "repo", repoDid)
525
+
default:
526
+
return fmt.Errorf("collaboratorUpdate unknown op %q", rec.Op)
527
+
}
469
528
return nil
470
529
}
471
530