Stitch any CI into Tangled
jetstream stores data in database
+971
-30
+3
-2
go.mod
+3
-2
go.mod
···
4
4
5
5
require (
6
6
github.com/bluesky-social/jetstream v0.0.0-20260226214936-e0274250f654
7
+
github.com/charmbracelet/log v1.0.0
7
8
github.com/gorilla/websocket v1.5.4-0.20250319132907-e064f32e3674
9
+
github.com/mattn/go-sqlite3 v1.14.44
8
10
tangled.org/core v1.13.0-alpha
9
11
)
10
12
···
15
17
github.com/cespare/xxhash/v2 v2.3.0 // indirect
16
18
github.com/charmbracelet/colorprofile v0.2.3-0.20250311203215-f60798e515dc // indirect
17
19
github.com/charmbracelet/lipgloss v1.1.0 // indirect
18
-
github.com/charmbracelet/log v1.0.0 // indirect
19
20
github.com/charmbracelet/x/ansi v0.8.0 // indirect
20
21
github.com/charmbracelet/x/cellbuf v0.0.13-0.20250311204145-2c3ea96c31dd // indirect
21
22
github.com/charmbracelet/x/term v0.2.1 // indirect
···
49
50
golang.org/x/crypto v0.48.0 // indirect
50
51
golang.org/x/exp v0.0.0-20260112195511-716be5621a96 // indirect
51
52
golang.org/x/net v0.50.0 // indirect
52
-
golang.org/x/sys v0.41.0 // indirect
53
+
golang.org/x/sys v0.42.0 // indirect
53
54
golang.org/x/xerrors v0.0.0-20240903120638-7835f813f4da // indirect
54
55
google.golang.org/protobuf v1.36.11 // indirect
55
56
lukechampine.com/blake3 v1.4.1 // indirect
+4
-2
go.sum
+4
-2
go.sum
···
46
46
github.com/mattn/go-isatty v0.0.20/go.mod h1:W+V8PltTTMOvKvAeJH7IuucS94S2C6jfK/D7dTCTo3Y=
47
47
github.com/mattn/go-runewidth v0.0.16 h1:E5ScNMtiwvlvB5paMFdw9p4kSQzbXFikJ5SQO6TULQc=
48
48
github.com/mattn/go-runewidth v0.0.16/go.mod h1:Jdepj2loyihRzMpdS35Xk/zdY8IAYHsh153qUoGf23w=
49
+
github.com/mattn/go-sqlite3 v1.14.44 h1:3VSe+xafpbzsLbdr2AWlAZk9yRHiBhTBakioXaCKTF8=
50
+
github.com/mattn/go-sqlite3 v1.14.44/go.mod h1:pjEuOr8IwzLJP2MfGeTb0A35jauH+C2kbHKBr7yXKVQ=
49
51
github.com/minio/sha256-simd v1.0.1 h1:6kaan5IFmwTNynnKKpDHe6FWHohJOHhCPchzK49dzMM=
50
52
github.com/minio/sha256-simd v1.0.1/go.mod h1:Pz6AKMiUdngCLpeTL/RJY1M9rUuPMYujV5xJjtbRSN8=
51
53
github.com/mr-tron/base58 v1.2.0 h1:T/HDJBh4ZCPbU39/+c3rRvE0uKBQlU27+QI8LJ4t64o=
···
100
102
golang.org/x/net v0.50.0 h1:ucWh9eiCGyDR3vtzso0WMQinm2Dnt8cFMuQa9K33J60=
101
103
golang.org/x/net v0.50.0/go.mod h1:UgoSli3F/pBgdJBHCTc+tp3gmrU4XswgGRgtnwWTfyM=
102
104
golang.org/x/sys v0.6.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
103
-
golang.org/x/sys v0.41.0 h1:Ivj+2Cp/ylzLiEU89QhWblYnOE9zerudt9Ftecq2C6k=
104
-
golang.org/x/sys v0.41.0/go.mod h1:OgkHotnGiDImocRcuBABYBEXf8A9a87e/uXjp9XT3ks=
105
+
golang.org/x/sys v0.42.0 h1:omrd2nAlyT5ESRdCLYdm3+fMfNFE/+Rf4bDIQImRJeo=
106
+
golang.org/x/sys v0.42.0/go.mod h1:4GL1E5IUh+htKOUEOaiffhrAeqysfVGipDYzABqnCmw=
105
107
golang.org/x/xerrors v0.0.0-20240903120638-7835f813f4da h1:noIWHXmPHxILtqtCOPIhSt0ABwskkZKjD3bXGnZGpNY=
106
108
golang.org/x/xerrors v0.0.0-20240903120638-7835f813f4da/go.mod h1:NDW/Ps6MPRej6fsCIbMTohpP40sJ/P/vI1MoTEGwX90=
107
109
google.golang.org/protobuf v1.36.11 h1:fV6ZwhNocDyBLK0dj+fg8ektcVegBBuEolpbTQyBNVE=
+159
-25
jetstream.go
+159
-25
jetstream.go
···
19
19
20
20
import (
21
21
"context"
22
+
"encoding/json"
22
23
"fmt"
23
24
"time"
24
25
···
28
29
"tangled.org/core/api/tangled"
29
30
)
30
31
32
+
// jetstream operation strings. The jetstream protocol publishes these as
33
+
// the Commit.Operation field; pulling them out as constants keeps the
34
+
// switch in handleJetstreamEvent honest about typos.
35
+
const (
36
+
jsOpCreate = "create"
37
+
jsOpUpdate = "update"
38
+
jsOpDelete = "delete"
39
+
)
40
+
31
41
// startJetstream dials the configured jetstream endpoint and spawns a
32
42
// background goroutine that consumes events for the lifetime of ctx. It
33
43
// returns once the client is constructed; connection errors surface in
34
44
// logs, not return values, because the read loop is expected to reconnect
35
45
// on its own.
36
46
//
47
+
// The store is used for two things: loading the persisted cursor so we
48
+
// resume from the last seen event after a restart, and persisting
49
+
// observed records so the rest of tack can answer membership questions
50
+
// without re-reading the firehose.
51
+
//
37
52
// The logger is pulled from ctx (see log.go); falls back to slog.Default()
38
53
// if none is attached.
39
-
func startJetstream(ctx context.Context, cfg config) error {
54
+
func startJetstream(ctx context.Context, cfg config, st *store) error {
40
55
logger := loggerFrom(ctx).With("component", "jetstream")
41
56
42
57
// `wantedCollections` is a server-side filter: jetstream will only send
···
54
69
clientCfg.WebsocketURL = cfg.JetstreamURL
55
70
clientCfg.WantedCollections = collections
56
71
57
-
// Re-attach the component-scoped logger so handleJetstreamEvent — which
58
-
// the scheduler invokes with the ctx we pass to ConnectAndRead — can
72
+
// The handler closes over `st` and `logger` so the scheduler signature
73
+
// stays plain `func(ctx, *Event) error` — no need for a method
74
+
// receiver or a global.
75
+
handler := func(ctx context.Context, evt *jsmodels.Event) error {
76
+
return handleJetstreamEvent(ctx, st, evt)
77
+
}
78
+
79
+
// Re-attach the component-scoped logger so handler — which the
80
+
// scheduler invokes with the ctx we pass to ConnectAndRead — can
59
81
// pull it back out via loggerFrom.
60
82
ctx = loggerInto(ctx, logger)
61
83
···
66
88
c, err := client.NewClient(
67
89
clientCfg,
68
90
logger,
69
-
sequential.NewScheduler(
70
-
"tack",
71
-
logger,
72
-
handleJetstreamEvent),
91
+
sequential.NewScheduler("tack", logger, handler),
73
92
)
74
93
if err != nil {
75
94
return fmt.Errorf("new jetstream client: %w", err)
76
95
}
77
96
78
-
// Reconnect loop. ConnectAndRead blocks on the websocket and returns
79
-
// either when the connection drops (transient network error, server
80
-
// restart, etc.) or when ctx is cancelled. On error we sleep briefly
81
-
// and reconnect; on ctx cancellation we exit cleanly.
82
-
//
83
-
// TODO: pass a *cursor here once we persist one, so we resume from the
84
-
// last seen event instead of "now" after a restart.
85
97
go func() {
86
98
for {
87
-
if err := c.ConnectAndRead(ctx, nil); err != nil {
99
+
// We re-read the cursor from the store at every (re)connect so we
100
+
// pick up any progress the previous connection persisted before
101
+
// dying. nil means "start from now", which is the right default on
102
+
// a brand-new install or after a corrupt cursor read.
103
+
cur, err := st.LoadCursor(ctx)
104
+
if err != nil {
105
+
logger.Warn("ignoring unreadable cursor; resuming from now", "err", err)
106
+
cur = nil
107
+
}
108
+
if cur != nil {
109
+
logger.Info("connecting to jetstream", "cursor_us", *cur)
110
+
} else {
111
+
logger.Info("connecting to jetstream from now (no cursor)")
112
+
}
113
+
114
+
// Reconnect loop. ConnectAndRead blocks on the websocket and returns
115
+
// either when the connection drops (transient network error, server
116
+
// restart, etc.) or when ctx is cancelled. On error we sleep briefly
117
+
// and reconnect; on ctx cancellation we exit cleanly.
118
+
if err := c.ConnectAndRead(ctx, cur); err != nil {
88
119
if ctx.Err() != nil {
89
120
return
90
121
}
···
101
132
return nil
102
133
}
103
134
104
-
// handleJetstreamEvent is the per-event callback for the JetStream.
105
-
func handleJetstreamEvent(ctx context.Context, evt *jsmodels.Event) error {
106
-
// We only care about commits, which are the actual record CRUD operations
107
-
// on a user's PDS.
135
+
// handleJetstreamEvent is the per-event callback for the JetStream. It
136
+
// applies the event to the store and advances the persisted cursor. Any
137
+
// returned error is logged by the scheduler but does not tear down the
138
+
// connection — the next event will retry the cursor write implicitly.
139
+
func handleJetstreamEvent(ctx context.Context, st *store, evt *jsmodels.Event) error {
140
+
// We only care about commits, which are the actual record CRUD
141
+
// operations on a user's PDS. Account/identity events are ignored
142
+
// for now; if we ever care about handle changes we can add them.
108
143
if evt.Kind != jsmodels.EventKindCommit || evt.Commit == nil {
109
144
return nil
110
145
}
146
+
logger := loggerFrom(ctx)
111
147
112
-
loggerFrom(ctx).Debug("event",
113
-
"did", evt.Did,
114
-
"collection", evt.Commit.Collection,
115
-
"op", evt.Commit.Operation,
116
-
"rkey", evt.Commit.RKey,
117
-
)
148
+
// Dispatch on collection. Unknown collections shouldn't happen given
149
+
// our wantedCollections filter, but be defensive — jetstream may
150
+
// send schema changes ahead of us updating the filter.
151
+
if err := applyCommit(ctx, st, evt); err != nil {
152
+
logger.Error("apply commit",
153
+
"err", err,
154
+
"did", evt.Did,
155
+
"collection", evt.Commit.Collection,
156
+
"op", evt.Commit.Operation,
157
+
"rkey", evt.Commit.RKey,
158
+
)
159
+
160
+
// Fall through to cursor save: a single bad record shouldn't
161
+
// stall the cursor forever and force us to re-process every
162
+
// subsequent event after a restart.
163
+
}
164
+
165
+
// Advance the cursor. TimeUS is the jetstream-assigned microsecond
166
+
// timestamp; saving it after-apply means a crash mid-batch will at
167
+
// worst replay the failing event, never skip past it.
168
+
if err := st.SaveCursor(ctx, evt.TimeUS); err != nil {
169
+
// Returning the error logs it; it doesn't kill the scheduler.
170
+
return fmt.Errorf("save cursor: %w", err)
171
+
}
172
+
118
173
return nil
119
174
}
175
+
176
+
// applyCommit routes a commit to the right store mutation based on its
177
+
// collection NSID and operation.
178
+
func applyCommit(ctx context.Context, st *store, evt *jsmodels.Event) error {
179
+
c := evt.Commit
180
+
switch c.Collection {
181
+
case tangled.SpindleMemberNSID:
182
+
return applySpindleMember(ctx, st, evt.Did, c)
183
+
case tangled.RepoNSID:
184
+
return applyRepo(ctx, st, evt.Did, c)
185
+
case tangled.RepoCollaboratorNSID:
186
+
return applyRepoCollaborator(ctx, st, evt.Did, c)
187
+
default:
188
+
// Server-side filter should prevent this, but log so we notice
189
+
// if jetstream ever changes behavior.
190
+
loggerFrom(ctx).Debug("ignoring unexpected collection",
191
+
"collection", c.Collection)
192
+
return nil
193
+
}
194
+
}
195
+
196
+
func applySpindleMember(ctx context.Context, st *store, did string, c *jsmodels.Commit) error {
197
+
switch c.Operation {
198
+
case jsOpCreate, jsOpUpdate:
199
+
var rec tangled.SpindleMember
200
+
if err := json.Unmarshal(c.Record, &rec); err != nil {
201
+
return fmt.Errorf("decode spindle.member: %w", err)
202
+
}
203
+
return st.UpsertSpindleMember(ctx, did, c.RKey, rec.Instance, rec.Subject, rec.CreatedAt)
204
+
case jsOpDelete:
205
+
return st.DeleteSpindleMember(ctx, did, c.RKey)
206
+
}
207
+
return nil
208
+
}
209
+
210
+
func applyRepo(ctx context.Context, st *store, did string, c *jsmodels.Commit) error {
211
+
switch c.Operation {
212
+
case jsOpCreate, jsOpUpdate:
213
+
var rec tangled.Repo
214
+
if err := json.Unmarshal(c.Record, &rec); err != nil {
215
+
return fmt.Errorf("decode repo: %w", err)
216
+
}
217
+
return st.UpsertRepo(ctx, did, c.RKey,
218
+
rec.Knot, rec.Name,
219
+
deref(rec.Spindle), deref(rec.RepoDid),
220
+
rec.CreatedAt,
221
+
)
222
+
case jsOpDelete:
223
+
return st.DeleteRepo(ctx, did, c.RKey)
224
+
}
225
+
return nil
226
+
}
227
+
228
+
func applyRepoCollaborator(ctx context.Context, st *store, did string, c *jsmodels.Commit) error {
229
+
switch c.Operation {
230
+
case jsOpCreate, jsOpUpdate:
231
+
var rec tangled.RepoCollaborator
232
+
if err := json.Unmarshal(c.Record, &rec); err != nil {
233
+
return fmt.Errorf("decode repo.collaborator: %w", err)
234
+
}
235
+
return st.UpsertRepoCollaborator(ctx, did, c.RKey,
236
+
deref(rec.Repo), deref(rec.RepoDid),
237
+
rec.Subject, rec.CreatedAt,
238
+
)
239
+
case jsOpDelete:
240
+
return st.DeleteRepoCollaborator(ctx, did, c.RKey)
241
+
}
242
+
return nil
243
+
}
244
+
245
+
// deref returns the pointed-to string, or "" for nil. The lexicon types
246
+
// model optional fields as *string; the store schema treats absent and
247
+
// empty the same, so collapsing the two here keeps callers tidy.
248
+
func deref(s *string) string {
249
+
if s == nil {
250
+
return ""
251
+
}
252
+
return *s
253
+
}
+258
jetstream_test.go
+258
jetstream_test.go
···
1
+
package main
2
+
3
+
// Tests for handleJetstreamEvent. We exercise the full path — collection
4
+
// dispatch, record decoding, store mutation, and cursor advancement —
5
+
// using an in-memory store from store_test.go's newTestStore helper.
6
+
//
7
+
// Events are constructed by hand rather than recorded from a live
8
+
// jetstream so the tests don't need network access and stay fast.
9
+
10
+
import (
11
+
"context"
12
+
"encoding/json"
13
+
"testing"
14
+
15
+
jsmodels "github.com/bluesky-social/jetstream/pkg/models"
16
+
"tangled.org/core/api/tangled"
17
+
)
18
+
19
+
// commitEvent builds a jetstream commit event for tests. timeUS is the
20
+
// cursor value the handler should persist after applying. record can be
21
+
// nil for delete operations, which carry no body.
22
+
func commitEvent(timeUS int64, did, collection, op, rkey string, record any) *jsmodels.Event {
23
+
var raw json.RawMessage
24
+
if record != nil {
25
+
b, err := json.Marshal(record)
26
+
if err != nil {
27
+
panic(err) // test helper — callers control the input
28
+
}
29
+
raw = b
30
+
}
31
+
return &jsmodels.Event{
32
+
Did: did,
33
+
TimeUS: timeUS,
34
+
Kind: jsmodels.EventKindCommit,
35
+
Commit: &jsmodels.Commit{
36
+
Operation: op,
37
+
Collection: collection,
38
+
RKey: rkey,
39
+
Record: raw,
40
+
},
41
+
}
42
+
}
43
+
44
+
// requireCursor asserts the persisted cursor equals want. Pulled out
45
+
// because almost every test in this file checks it.
46
+
func requireCursor(t *testing.T, s *store, want int64) {
47
+
t.Helper()
48
+
got, err := s.LoadCursor(context.Background())
49
+
if err != nil {
50
+
t.Fatalf("load cursor: %v", err)
51
+
}
52
+
if got == nil || *got != want {
53
+
t.Fatalf("cursor = %v, want %d", got, want)
54
+
}
55
+
}
56
+
57
+
// TestHandleNonCommitEvent confirms account/identity events are ignored
58
+
// without error and without advancing the cursor — they don't have a
59
+
// TimeUS we want to commit to.
60
+
func TestHandleNonCommitEvent(t *testing.T) {
61
+
s := newTestStore(t)
62
+
ctx := context.Background()
63
+
64
+
evt := &jsmodels.Event{
65
+
Did: "did:plc:foo",
66
+
TimeUS: 100,
67
+
Kind: jsmodels.EventKindAccount,
68
+
}
69
+
if err := handleJetstreamEvent(ctx, s, evt); err != nil {
70
+
t.Fatalf("handle: %v", err)
71
+
}
72
+
got, err := s.LoadCursor(ctx)
73
+
if err != nil {
74
+
t.Fatalf("load cursor: %v", err)
75
+
}
76
+
if got != nil {
77
+
t.Fatalf("expected no cursor for non-commit, got %d", *got)
78
+
}
79
+
}
80
+
81
+
// TestHandleSpindleMemberCreate exercises the happy path: a create
82
+
// commit lands a row in spindle_members and advances the cursor.
83
+
func TestHandleSpindleMemberCreate(t *testing.T) {
84
+
s := newTestStore(t)
85
+
ctx := context.Background()
86
+
87
+
rec := tangled.SpindleMember{
88
+
Instance: "https://spindle.example",
89
+
Subject: "did:plc:alice",
90
+
CreatedAt: "2026-01-01T00:00:00Z",
91
+
}
92
+
evt := commitEvent(12345, "did:plc:owner", tangled.SpindleMemberNSID, jsOpCreate, "rk1", rec)
93
+
if err := handleJetstreamEvent(ctx, s, evt); err != nil {
94
+
t.Fatalf("handle: %v", err)
95
+
}
96
+
97
+
var instance, subject string
98
+
err := s.db.QueryRowContext(ctx,
99
+
`SELECT instance, subject FROM spindle_members WHERE did = ? AND rkey = ?`,
100
+
"did:plc:owner", "rk1",
101
+
).Scan(&instance, &subject)
102
+
if err != nil {
103
+
t.Fatalf("query: %v", err)
104
+
}
105
+
if instance != "https://spindle.example" || subject != "did:plc:alice" {
106
+
t.Fatalf("got (%q,%q)", instance, subject)
107
+
}
108
+
requireCursor(t, s, 12345)
109
+
}
110
+
111
+
// TestHandleSpindleMemberDelete confirms a delete commit removes the
112
+
// previously-persisted row.
113
+
func TestHandleSpindleMemberDelete(t *testing.T) {
114
+
s := newTestStore(t)
115
+
ctx := context.Background()
116
+
117
+
if err := s.UpsertSpindleMember(ctx, "did:plc:owner", "rk1", "i", "s", "t"); err != nil {
118
+
t.Fatalf("seed: %v", err)
119
+
}
120
+
evt := commitEvent(99, "did:plc:owner", tangled.SpindleMemberNSID, jsOpDelete, "rk1", nil)
121
+
if err := handleJetstreamEvent(ctx, s, evt); err != nil {
122
+
t.Fatalf("handle: %v", err)
123
+
}
124
+
if n := countRows(t, s, "spindle_members"); n != 0 {
125
+
t.Fatalf("after delete: %d rows, want 0", n)
126
+
}
127
+
requireCursor(t, s, 99)
128
+
}
129
+
130
+
// TestHandleRepoCreateOptionals verifies pointer-typed optional fields
131
+
// on tangled.Repo are derefed correctly (and nils become empty strings)
132
+
// when written to the store.
133
+
func TestHandleRepoCreateOptionals(t *testing.T) {
134
+
s := newTestStore(t)
135
+
ctx := context.Background()
136
+
137
+
spindle := "https://spindle.example"
138
+
repoDid := "did:plc:repo"
139
+
rec := tangled.Repo{
140
+
Knot: "knot.example",
141
+
Name: "myrepo",
142
+
Spindle: &spindle,
143
+
RepoDid: &repoDid,
144
+
CreatedAt: "2026-01-01T00:00:00Z",
145
+
}
146
+
evt := commitEvent(7, "did:plc:owner", tangled.RepoNSID, jsOpCreate, "repo1", rec)
147
+
if err := handleJetstreamEvent(ctx, s, evt); err != nil {
148
+
t.Fatalf("handle: %v", err)
149
+
}
150
+
151
+
var gotSpindle, gotRepoDid string
152
+
err := s.db.QueryRowContext(ctx,
153
+
`SELECT spindle, repo_did FROM repos WHERE did = ? AND rkey = ?`,
154
+
"did:plc:owner", "repo1",
155
+
).Scan(&gotSpindle, &gotRepoDid)
156
+
if err != nil {
157
+
t.Fatalf("query: %v", err)
158
+
}
159
+
if gotSpindle != spindle || gotRepoDid != repoDid {
160
+
t.Fatalf("got (%q,%q)", gotSpindle, gotRepoDid)
161
+
}
162
+
163
+
// Now a record with both optionals nil — should land as empty
164
+
// strings, not crash on a nil dereference in deref().
165
+
rec2 := tangled.Repo{
166
+
Knot: "knot.example",
167
+
Name: "other",
168
+
CreatedAt: "2026-01-01T00:00:00Z",
169
+
}
170
+
evt2 := commitEvent(8, "did:plc:owner", tangled.RepoNSID, jsOpCreate, "repo2", rec2)
171
+
if err := handleJetstreamEvent(ctx, s, evt2); err != nil {
172
+
t.Fatalf("handle nil-optionals: %v", err)
173
+
}
174
+
err = s.db.QueryRowContext(ctx,
175
+
`SELECT spindle, repo_did FROM repos WHERE did = ? AND rkey = ?`,
176
+
"did:plc:owner", "repo2",
177
+
).Scan(&gotSpindle, &gotRepoDid)
178
+
if err != nil {
179
+
t.Fatalf("query nil-optionals: %v", err)
180
+
}
181
+
if gotSpindle != "" || gotRepoDid != "" {
182
+
t.Fatalf("nil optionals: got (%q,%q), want both empty", gotSpindle, gotRepoDid)
183
+
}
184
+
requireCursor(t, s, 8)
185
+
}
186
+
187
+
// TestHandleRepoCollaboratorCreate covers the third dispatch arm so each
188
+
// collection has at least one apply test.
189
+
func TestHandleRepoCollaboratorCreate(t *testing.T) {
190
+
s := newTestStore(t)
191
+
ctx := context.Background()
192
+
193
+
repo := "myrepo"
194
+
rec := tangled.RepoCollaborator{
195
+
Repo: &repo,
196
+
Subject: "did:plc:carol",
197
+
CreatedAt: "2026-01-01T00:00:00Z",
198
+
}
199
+
evt := commitEvent(55, "did:plc:owner", tangled.RepoCollaboratorNSID, jsOpCreate, "c1", rec)
200
+
if err := handleJetstreamEvent(ctx, s, evt); err != nil {
201
+
t.Fatalf("handle: %v", err)
202
+
}
203
+
204
+
var subject string
205
+
err := s.db.QueryRowContext(ctx,
206
+
`SELECT subject FROM repo_collaborators WHERE did = ? AND rkey = ?`,
207
+
"did:plc:owner", "c1",
208
+
).Scan(&subject)
209
+
if err != nil {
210
+
t.Fatalf("query: %v", err)
211
+
}
212
+
if subject != "did:plc:carol" {
213
+
t.Fatalf("subject = %q", subject)
214
+
}
215
+
requireCursor(t, s, 55)
216
+
}
217
+
218
+
// TestHandleUnknownCollection makes sure a collection we didn't ask for
219
+
// (jetstream filter changes, schema drift) is silently dropped — no
220
+
// error, no row, but cursor still advances so we don't replay it.
221
+
func TestHandleUnknownCollection(t *testing.T) {
222
+
s := newTestStore(t)
223
+
ctx := context.Background()
224
+
225
+
evt := commitEvent(42, "did:plc:owner", "app.bsky.feed.post", jsOpCreate, "rk", map[string]string{"text": "hi"})
226
+
if err := handleJetstreamEvent(ctx, s, evt); err != nil {
227
+
t.Fatalf("handle: %v", err)
228
+
}
229
+
requireCursor(t, s, 42)
230
+
}
231
+
232
+
// TestHandleBadRecordAdvancesCursor is the failure-mode counterpart to
233
+
// the happy paths: a malformed record body must be logged-and-skipped
234
+
// (not returned as an error that pauses the scheduler) and the cursor
235
+
// must still advance so we don't loop on the same bad event forever.
236
+
func TestHandleBadRecordAdvancesCursor(t *testing.T) {
237
+
s := newTestStore(t)
238
+
ctx := context.Background()
239
+
240
+
evt := &jsmodels.Event{
241
+
Did: "did:plc:owner",
242
+
TimeUS: 1000,
243
+
Kind: jsmodels.EventKindCommit,
244
+
Commit: &jsmodels.Commit{
245
+
Operation: jsOpCreate,
246
+
Collection: tangled.SpindleMemberNSID,
247
+
RKey: "broken",
248
+
Record: json.RawMessage(`{not valid json`),
249
+
},
250
+
}
251
+
if err := handleJetstreamEvent(ctx, s, evt); err != nil {
252
+
t.Fatalf("handle should swallow decode error, got: %v", err)
253
+
}
254
+
if n := countRows(t, s, "spindle_members"); n != 0 {
255
+
t.Fatalf("bad record should not have inserted; got %d rows", n)
256
+
}
257
+
requireCursor(t, s, 1000)
258
+
}
+18
-1
main.go
+18
-1
main.go
···
23
23
Addr string
24
24
OwnerDID string
25
25
JetstreamURL string
26
+
DBPath string
26
27
}
27
28
28
29
func loadConfig() (config, error) {
···
30
31
Addr: envOr("TACK_LISTEN_ADDR", ":8080"),
31
32
OwnerDID: os.Getenv("TACK_OWNER_DID"),
32
33
JetstreamURL: envOr("TACK_JETSTREAM_URL", "wss://jetstream1.us-west.bsky.network/subscribe"),
34
+
DBPath: envOr("TACK_DB_PATH", "tack.db"),
33
35
}
34
36
addrFlag := flag.String("addr", cfg.Addr, "HTTP listen address (overrides TACK_LISTEN_ADDR)")
35
37
flag.Parse()
···
75
77
defer stop()
76
78
ctx = loggerInto(ctx, logger)
77
79
80
+
// Open (or create) the SQLite store. Holds jetstream cursor +
81
+
// observed Tangled membership records. Closed last during shutdown
82
+
// so anything writing to it during teardown still succeeds.
83
+
st, err := openStore(cfg.DBPath)
84
+
if err != nil {
85
+
logger.Error("failed to open store", "err", err, "path", cfg.DBPath)
86
+
os.Exit(1)
87
+
}
88
+
defer func() {
89
+
if err := st.Close(); err != nil {
90
+
logger.Error("close store", "err", err)
91
+
}
92
+
}()
93
+
logger.Info("store open", "path", cfg.DBPath)
94
+
78
95
// Start the JetStream listener in the background.
79
-
if err := startJetstream(ctx, cfg); err != nil {
96
+
if err := startJetstream(ctx, cfg, st); err != nil {
80
97
logger.Error("failed to start jetstream consumer", "err", err)
81
98
os.Exit(1)
82
99
}
+208
store.go
+208
store.go
···
1
+
package main
2
+
3
+
// SQLite-backed persistence for tack.
4
+
//
5
+
// Two responsibilities live here:
6
+
//
7
+
// 1. The jetstream cursor — a microsecond unix timestamp the AT Proto
8
+
// firehose uses to resume from a specific point. Without persistence
9
+
// every restart begins at "now" and we'd silently miss any record
10
+
// published while we were down.
11
+
//
12
+
// 2. Tangled membership state derived from jetstream commits:
13
+
// sh.tangled.spindle.member, sh.tangled.repo, and
14
+
// sh.tangled.repo.collaborator. We need this to later answer
15
+
// "is DID X allowed to trigger a build on this spindle for repo Y?"
16
+
//
17
+
// We use mattn/go-sqlite3, which is the most battle-tested SQLite driver
18
+
// for Go. It requires CGo, which is fine for tack — the project already
19
+
// builds under Nix where a C toolchain is readily available.
20
+
21
+
import (
22
+
"context"
23
+
"database/sql"
24
+
"errors"
25
+
"fmt"
26
+
"strconv"
27
+
28
+
_ "github.com/mattn/go-sqlite3"
29
+
)
30
+
31
+
// store wraps the SQLite handle and exposes the small set of operations
32
+
// the rest of tack needs. Keeping the surface narrow lets the persistence
33
+
// layer be swapped or mocked later without rewriting callers.
34
+
type store struct {
35
+
db *sql.DB
36
+
}
37
+
38
+
// openStore opens (or creates) the SQLite database at path and applies
39
+
// the schema. It also flips on WAL + NORMAL synchronous, which is the
40
+
// usual "this is a long-running server, not a one-shot script" config:
41
+
// concurrent reads don't block the single writer, and we trade a little
42
+
// crash-safety on power loss for a lot of write throughput.
43
+
func openStore(path string) (*store, error) {
44
+
// mattn/go-sqlite3 reads pragma-style query parameters (_journal_mode,
45
+
// _synchronous, _foreign_keys) and applies them on each new connection.
46
+
// journal_mode=WAL persists in the database file but setting it here
47
+
// also covers the first-ever open.
48
+
dsn := fmt.Sprintf("file:%s?_journal_mode=WAL&_synchronous=NORMAL&_foreign_keys=on", path)
49
+
db, err := sql.Open("sqlite3", dsn)
50
+
if err != nil {
51
+
return nil, fmt.Errorf("open sqlite %q: %w", path, err)
52
+
}
53
+
54
+
// SQLite only supports one writer at a time. Capping max open conns
55
+
// at 1 keeps database/sql from spinning up extra connections that
56
+
// will only ever serialize behind that writer.
57
+
db.SetMaxOpenConns(1)
58
+
59
+
s := &store{db: db}
60
+
if err := s.migrate(context.Background()); err != nil {
61
+
_ = db.Close()
62
+
return nil, err
63
+
}
64
+
return s, nil
65
+
}
66
+
67
+
// Close releases the underlying database handle.
68
+
func (s *store) Close() error {
69
+
return s.db.Close()
70
+
}
71
+
72
+
// metaCursorKey is the meta-table key under which we persist the
73
+
// jetstream cursor. Pulled out as a constant to avoid drift between
74
+
// load/save sites.
75
+
const metaCursorKey = "jetstream_cursor"
76
+
77
+
// LoadCursor returns the persisted jetstream cursor, or nil if none has
78
+
// been saved yet (signaling "start from now"). The cursor is a unix
79
+
// microsecond timestamp — see jetstream.Event.TimeUS.
80
+
func (s *store) LoadCursor(ctx context.Context) (*int64, error) {
81
+
var raw string
82
+
err := s.db.QueryRowContext(ctx,
83
+
`SELECT value FROM meta WHERE key = ?`, metaCursorKey,
84
+
).Scan(&raw)
85
+
if errors.Is(err, sql.ErrNoRows) {
86
+
return nil, nil
87
+
}
88
+
if err != nil {
89
+
return nil, fmt.Errorf("load cursor: %w", err)
90
+
}
91
+
v, err := strconv.ParseInt(raw, 10, 64)
92
+
if err != nil {
93
+
// A malformed cursor shouldn't wedge startup — log-and-ignore
94
+
// (return nil) so we just resume from "now". The caller has the
95
+
// logger; we surface the parse error so it can decide.
96
+
return nil, fmt.Errorf("parse cursor %q: %w", raw, err)
97
+
}
98
+
return &v, nil
99
+
}
100
+
101
+
// SaveCursor writes the jetstream cursor. It uses an UPSERT so the meta
102
+
// row is created on first call and updated thereafter.
103
+
func (s *store) SaveCursor(ctx context.Context, cursor int64) error {
104
+
_, err := s.db.ExecContext(ctx,
105
+
`INSERT INTO meta (key, value) VALUES (?, ?)
106
+
ON CONFLICT(key) DO UPDATE SET value = excluded.value`,
107
+
metaCursorKey, strconv.FormatInt(cursor, 10),
108
+
)
109
+
if err != nil {
110
+
return fmt.Errorf("save cursor: %w", err)
111
+
}
112
+
return nil
113
+
}
114
+
115
+
// UpsertSpindleMember records (or refreshes) a sh.tangled.spindle.member
116
+
// observation.
117
+
func (s *store) UpsertSpindleMember(ctx context.Context, did, rkey, instance, subject, createdAt string) error {
118
+
_, err := s.db.ExecContext(ctx,
119
+
`INSERT INTO spindle_members (did, rkey, instance, subject, created_at)
120
+
VALUES (?, ?, ?, ?, ?)
121
+
ON CONFLICT(did, rkey) DO UPDATE SET
122
+
instance = excluded.instance,
123
+
subject = excluded.subject,
124
+
created_at = excluded.created_at`,
125
+
did, rkey, instance, subject, createdAt,
126
+
)
127
+
if err != nil {
128
+
return fmt.Errorf("upsert spindle_member: %w", err)
129
+
}
130
+
return nil
131
+
}
132
+
133
+
// DeleteSpindleMember removes a member record by its ATProto identity.
134
+
func (s *store) DeleteSpindleMember(ctx context.Context, did, rkey string) error {
135
+
_, err := s.db.ExecContext(ctx,
136
+
`DELETE FROM spindle_members WHERE did = ? AND rkey = ?`,
137
+
did, rkey,
138
+
)
139
+
if err != nil {
140
+
return fmt.Errorf("delete spindle_member: %w", err)
141
+
}
142
+
return nil
143
+
}
144
+
145
+
// UpsertRepo records (or refreshes) a sh.tangled.repo observation.
146
+
// spindle and repoDid may be empty strings — we store them as such rather
147
+
// than as SQL NULLs to keep the upsert path uniform.
148
+
func (s *store) UpsertRepo(ctx context.Context, did, rkey, knot, name, spindle, repoDid, createdAt string) error {
149
+
_, err := s.db.ExecContext(ctx,
150
+
`INSERT INTO repos (did, rkey, knot, name, spindle, repo_did, created_at)
151
+
VALUES (?, ?, ?, ?, ?, ?, ?)
152
+
ON CONFLICT(did, rkey) DO UPDATE SET
153
+
knot = excluded.knot,
154
+
name = excluded.name,
155
+
spindle = excluded.spindle,
156
+
repo_did = excluded.repo_did,
157
+
created_at = excluded.created_at`,
158
+
did, rkey, knot, name, spindle, repoDid, createdAt,
159
+
)
160
+
if err != nil {
161
+
return fmt.Errorf("upsert repo: %w", err)
162
+
}
163
+
return nil
164
+
}
165
+
166
+
// DeleteRepo removes a repo record by its ATProto identity.
167
+
func (s *store) DeleteRepo(ctx context.Context, did, rkey string) error {
168
+
_, err := s.db.ExecContext(ctx,
169
+
`DELETE FROM repos WHERE did = ? AND rkey = ?`,
170
+
did, rkey,
171
+
)
172
+
if err != nil {
173
+
return fmt.Errorf("delete repo: %w", err)
174
+
}
175
+
return nil
176
+
}
177
+
178
+
// UpsertRepoCollaborator records (or refreshes) a
179
+
// sh.tangled.repo.collaborator observation.
180
+
func (s *store) UpsertRepoCollaborator(ctx context.Context, did, rkey, repo, repoDid, subject, createdAt string) error {
181
+
_, err := s.db.ExecContext(ctx,
182
+
`INSERT INTO repo_collaborators (did, rkey, repo, repo_did, subject, created_at)
183
+
VALUES (?, ?, ?, ?, ?, ?)
184
+
ON CONFLICT(did, rkey) DO UPDATE SET
185
+
repo = excluded.repo,
186
+
repo_did = excluded.repo_did,
187
+
subject = excluded.subject,
188
+
created_at = excluded.created_at`,
189
+
did, rkey, repo, repoDid, subject, createdAt,
190
+
)
191
+
if err != nil {
192
+
return fmt.Errorf("upsert repo_collaborator: %w", err)
193
+
}
194
+
return nil
195
+
}
196
+
197
+
// DeleteRepoCollaborator removes a collaborator record by its ATProto
198
+
// identity.
199
+
func (s *store) DeleteRepoCollaborator(ctx context.Context, did, rkey string) error {
200
+
_, err := s.db.ExecContext(ctx,
201
+
`DELETE FROM repo_collaborators WHERE did = ? AND rkey = ?`,
202
+
did, rkey,
203
+
)
204
+
if err != nil {
205
+
return fmt.Errorf("delete repo_collaborator: %w", err)
206
+
}
207
+
return nil
208
+
}
+68
store_migrate.go
+68
store_migrate.go
···
1
+
package main
2
+
3
+
// Schema definition and migration logic for the SQLite store. Pulled out
4
+
// of store.go so the big SQL block doesn't sit in the middle of the
5
+
// runtime API surface.
6
+
7
+
import (
8
+
"context"
9
+
"fmt"
10
+
)
11
+
12
+
// schema is the full set of CREATE statements applied at startup. It is
13
+
// idempotent and additive only — no `DROP`s — so future changes can be
14
+
// layered on as additional statements without needing a separate
15
+
// migration tool until the project actually outgrows that.
16
+
const schema = `
17
+
CREATE TABLE IF NOT EXISTS meta (
18
+
key TEXT PRIMARY KEY,
19
+
value TEXT NOT NULL
20
+
);
21
+
22
+
-- Records of sh.tangled.spindle.member. The owner of a spindle publishes
23
+
-- one of these per authorized member. (did, rkey) is the natural ATProto
24
+
-- key — did identifies the publisher's PDS, rkey identifies the record
25
+
-- within that PDS's collection.
26
+
CREATE TABLE IF NOT EXISTS spindle_members (
27
+
did TEXT NOT NULL,
28
+
rkey TEXT NOT NULL,
29
+
instance TEXT NOT NULL,
30
+
subject TEXT NOT NULL,
31
+
created_at TEXT NOT NULL,
32
+
PRIMARY KEY (did, rkey)
33
+
);
34
+
35
+
-- Records of sh.tangled.repo. We keep the full set so that when a
36
+
-- pipeline trigger arrives we can look up which knot/spindle/repo_did
37
+
-- it corresponds to without another round-trip.
38
+
CREATE TABLE IF NOT EXISTS repos (
39
+
did TEXT NOT NULL,
40
+
rkey TEXT NOT NULL,
41
+
knot TEXT NOT NULL,
42
+
name TEXT NOT NULL,
43
+
spindle TEXT,
44
+
repo_did TEXT,
45
+
created_at TEXT NOT NULL,
46
+
PRIMARY KEY (did, rkey)
47
+
);
48
+
49
+
-- Records of sh.tangled.repo.collaborator. Used together with repos to
50
+
-- decide whether a triggering DID is allowed to push builds to us.
51
+
CREATE TABLE IF NOT EXISTS repo_collaborators (
52
+
did TEXT NOT NULL,
53
+
rkey TEXT NOT NULL,
54
+
repo TEXT,
55
+
repo_did TEXT,
56
+
subject TEXT NOT NULL,
57
+
created_at TEXT NOT NULL,
58
+
PRIMARY KEY (did, rkey)
59
+
);
60
+
`
61
+
62
+
// migrate applies the schema. Safe to call repeatedly.
63
+
func (s *store) migrate(ctx context.Context) error {
64
+
if _, err := s.db.ExecContext(ctx, schema); err != nil {
65
+
return fmt.Errorf("apply schema: %w", err)
66
+
}
67
+
return nil
68
+
}
+253
store_test.go
+253
store_test.go
···
1
+
package main
2
+
3
+
// Tests for the SQLite store. We use t.TempDir() for an isolated database
4
+
// per test, which both keeps tests independent and exercises the open +
5
+
// migrate path on every run.
6
+
//
7
+
// Where the store doesn't expose a query method (it's intentionally
8
+
// write-mostly today) we drop down to raw SQL via s.db to verify the
9
+
// row is what we expect.
10
+
11
+
import (
12
+
"context"
13
+
"path/filepath"
14
+
"testing"
15
+
)
16
+
17
+
// newTestStore opens a fresh store in a per-test temp dir and registers
18
+
// cleanup. Centralized so test bodies stay focused on behavior.
19
+
func newTestStore(t *testing.T) *store {
20
+
t.Helper()
21
+
path := filepath.Join(t.TempDir(), "tack.db")
22
+
s, err := openStore(path)
23
+
if err != nil {
24
+
t.Fatalf("openStore: %v", err)
25
+
}
26
+
t.Cleanup(func() {
27
+
if err := s.Close(); err != nil {
28
+
t.Errorf("close store: %v", err)
29
+
}
30
+
})
31
+
return s
32
+
}
33
+
34
+
// TestOpenStoreIdempotent makes sure re-opening an existing database
35
+
// (which is what happens on every restart) succeeds and leaves the
36
+
// schema intact.
37
+
func TestOpenStoreIdempotent(t *testing.T) {
38
+
path := filepath.Join(t.TempDir(), "tack.db")
39
+
s1, err := openStore(path)
40
+
if err != nil {
41
+
t.Fatalf("first open: %v", err)
42
+
}
43
+
if err := s1.Close(); err != nil {
44
+
t.Fatalf("first close: %v", err)
45
+
}
46
+
s2, err := openStore(path)
47
+
if err != nil {
48
+
t.Fatalf("second open: %v", err)
49
+
}
50
+
defer s2.Close()
51
+
52
+
// Sanity check: the schema is in place by writing and reading back
53
+
// a cursor through the freshly re-opened handle.
54
+
ctx := context.Background()
55
+
if err := s2.SaveCursor(ctx, 42); err != nil {
56
+
t.Fatalf("save cursor: %v", err)
57
+
}
58
+
got, err := s2.LoadCursor(ctx)
59
+
if err != nil {
60
+
t.Fatalf("load cursor: %v", err)
61
+
}
62
+
if got == nil || *got != 42 {
63
+
t.Fatalf("cursor = %v, want 42", got)
64
+
}
65
+
}
66
+
67
+
// TestCursorRoundtrip covers the three states a cursor can be in:
68
+
// missing (nil), present, and overwritten. Together they exercise the
69
+
// INSERT and the ON CONFLICT branch of SaveCursor.
70
+
func TestCursorRoundtrip(t *testing.T) {
71
+
s := newTestStore(t)
72
+
ctx := context.Background()
73
+
74
+
got, err := s.LoadCursor(ctx)
75
+
if err != nil {
76
+
t.Fatalf("load cursor (empty): %v", err)
77
+
}
78
+
if got != nil {
79
+
t.Fatalf("expected nil cursor on fresh store, got %d", *got)
80
+
}
81
+
82
+
if err := s.SaveCursor(ctx, 1234567890); err != nil {
83
+
t.Fatalf("save cursor: %v", err)
84
+
}
85
+
got, err = s.LoadCursor(ctx)
86
+
if err != nil {
87
+
t.Fatalf("load cursor: %v", err)
88
+
}
89
+
if got == nil || *got != 1234567890 {
90
+
t.Fatalf("cursor after save = %v, want 1234567890", got)
91
+
}
92
+
93
+
// Overwrite path — exercises ON CONFLICT DO UPDATE.
94
+
if err := s.SaveCursor(ctx, 9999); err != nil {
95
+
t.Fatalf("overwrite cursor: %v", err)
96
+
}
97
+
got, err = s.LoadCursor(ctx)
98
+
if err != nil {
99
+
t.Fatalf("load cursor (overwrite): %v", err)
100
+
}
101
+
if got == nil || *got != 9999 {
102
+
t.Fatalf("cursor after overwrite = %v, want 9999", got)
103
+
}
104
+
}
105
+
106
+
// TestSpindleMemberLifecycle covers insert, update-on-conflict, and
107
+
// delete for spindle.member rows. We read back via raw SQL since the
108
+
// store doesn't expose a query helper.
109
+
func TestSpindleMemberLifecycle(t *testing.T) {
110
+
s := newTestStore(t)
111
+
ctx := context.Background()
112
+
113
+
const did, rkey = "did:plc:owner", "abc123"
114
+
115
+
if err := s.UpsertSpindleMember(ctx, did, rkey, "https://spindle.example", "did:plc:alice", "2026-01-01T00:00:00Z"); err != nil {
116
+
t.Fatalf("insert: %v", err)
117
+
}
118
+
119
+
var instance, subject, createdAt string
120
+
err := s.db.QueryRowContext(ctx,
121
+
`SELECT instance, subject, created_at FROM spindle_members WHERE did = ? AND rkey = ?`,
122
+
did, rkey,
123
+
).Scan(&instance, &subject, &createdAt)
124
+
if err != nil {
125
+
t.Fatalf("query after insert: %v", err)
126
+
}
127
+
if instance != "https://spindle.example" || subject != "did:plc:alice" || createdAt != "2026-01-01T00:00:00Z" {
128
+
t.Fatalf("after insert got (%q,%q,%q)", instance, subject, createdAt)
129
+
}
130
+
131
+
// Update path — same primary key, different fields.
132
+
if err := s.UpsertSpindleMember(ctx, did, rkey, "https://spindle.example", "did:plc:bob", "2026-02-02T00:00:00Z"); err != nil {
133
+
t.Fatalf("update: %v", err)
134
+
}
135
+
err = s.db.QueryRowContext(ctx,
136
+
`SELECT subject, created_at FROM spindle_members WHERE did = ? AND rkey = ?`,
137
+
did, rkey,
138
+
).Scan(&subject, &createdAt)
139
+
if err != nil {
140
+
t.Fatalf("query after update: %v", err)
141
+
}
142
+
if subject != "did:plc:bob" || createdAt != "2026-02-02T00:00:00Z" {
143
+
t.Fatalf("after update got (%q,%q)", subject, createdAt)
144
+
}
145
+
146
+
// Delete and verify the row is gone.
147
+
if err := s.DeleteSpindleMember(ctx, did, rkey); err != nil {
148
+
t.Fatalf("delete: %v", err)
149
+
}
150
+
if n := countRows(t, s, "spindle_members"); n != 0 {
151
+
t.Fatalf("after delete, spindle_members has %d rows, want 0", n)
152
+
}
153
+
154
+
// Deleting a row that doesn't exist must succeed silently — the
155
+
// jetstream stream can replay deletes after a restart.
156
+
if err := s.DeleteSpindleMember(ctx, did, rkey); err != nil {
157
+
t.Fatalf("delete missing: %v", err)
158
+
}
159
+
}
160
+
161
+
// TestRepoLifecycle exercises Repo upsert/delete and confirms that
162
+
// optional fields (spindle, repo_did) round-trip as the empty string
163
+
// rather than NULL — the store schema treats them uniformly.
164
+
func TestRepoLifecycle(t *testing.T) {
165
+
s := newTestStore(t)
166
+
ctx := context.Background()
167
+
168
+
const did, rkey = "did:plc:owner", "repo1"
169
+
170
+
if err := s.UpsertRepo(ctx, did, rkey, "knot.example", "myrepo", "https://spindle.example", "did:plc:repo", "2026-01-01T00:00:00Z"); err != nil {
171
+
t.Fatalf("insert: %v", err)
172
+
}
173
+
174
+
var knot, name, spindle, repoDid string
175
+
err := s.db.QueryRowContext(ctx,
176
+
`SELECT knot, name, spindle, repo_did FROM repos WHERE did = ? AND rkey = ?`,
177
+
did, rkey,
178
+
).Scan(&knot, &name, &spindle, &repoDid)
179
+
if err != nil {
180
+
t.Fatalf("query after insert: %v", err)
181
+
}
182
+
if knot != "knot.example" || name != "myrepo" || spindle != "https://spindle.example" || repoDid != "did:plc:repo" {
183
+
t.Fatalf("after insert got (%q,%q,%q,%q)", knot, name, spindle, repoDid)
184
+
}
185
+
186
+
// Upsert with cleared optional fields — verifies UPDATE actually
187
+
// overwrites them rather than preserving the old non-empty values.
188
+
if err := s.UpsertRepo(ctx, did, rkey, "knot.example", "myrepo", "", "", "2026-01-01T00:00:00Z"); err != nil {
189
+
t.Fatalf("update: %v", err)
190
+
}
191
+
err = s.db.QueryRowContext(ctx,
192
+
`SELECT spindle, repo_did FROM repos WHERE did = ? AND rkey = ?`,
193
+
did, rkey,
194
+
).Scan(&spindle, &repoDid)
195
+
if err != nil {
196
+
t.Fatalf("query after update: %v", err)
197
+
}
198
+
if spindle != "" || repoDid != "" {
199
+
t.Fatalf("after update got (%q,%q), want both empty", spindle, repoDid)
200
+
}
201
+
202
+
if err := s.DeleteRepo(ctx, did, rkey); err != nil {
203
+
t.Fatalf("delete: %v", err)
204
+
}
205
+
if n := countRows(t, s, "repos"); n != 0 {
206
+
t.Fatalf("after delete, repos has %d rows, want 0", n)
207
+
}
208
+
}
209
+
210
+
// TestRepoCollaboratorLifecycle mirrors the repo and member tests for
211
+
// the collaborator table.
212
+
func TestRepoCollaboratorLifecycle(t *testing.T) {
213
+
s := newTestStore(t)
214
+
ctx := context.Background()
215
+
216
+
const did, rkey = "did:plc:owner", "collab1"
217
+
218
+
if err := s.UpsertRepoCollaborator(ctx, did, rkey, "myrepo", "did:plc:repo", "did:plc:carol", "2026-01-01T00:00:00Z"); err != nil {
219
+
t.Fatalf("insert: %v", err)
220
+
}
221
+
222
+
var repo, repoDid, subject string
223
+
err := s.db.QueryRowContext(ctx,
224
+
`SELECT repo, repo_did, subject FROM repo_collaborators WHERE did = ? AND rkey = ?`,
225
+
did, rkey,
226
+
).Scan(&repo, &repoDid, &subject)
227
+
if err != nil {
228
+
t.Fatalf("query: %v", err)
229
+
}
230
+
if repo != "myrepo" || repoDid != "did:plc:repo" || subject != "did:plc:carol" {
231
+
t.Fatalf("got (%q,%q,%q)", repo, repoDid, subject)
232
+
}
233
+
234
+
if err := s.DeleteRepoCollaborator(ctx, did, rkey); err != nil {
235
+
t.Fatalf("delete: %v", err)
236
+
}
237
+
if n := countRows(t, s, "repo_collaborators"); n != 0 {
238
+
t.Fatalf("after delete, repo_collaborators has %d rows, want 0", n)
239
+
}
240
+
}
241
+
242
+
// countRows is a small SELECT COUNT(*) helper used by lifecycle tests
243
+
// to verify deletes actually removed the row. Table name is interpolated
244
+
// directly because callers pass a constant from the schema, not user
245
+
// input — SQLite doesn't allow parameterized table names anyway.
246
+
func countRows(t *testing.T, s *store, table string) int {
247
+
t.Helper()
248
+
var n int
249
+
if err := s.db.QueryRow(`SELECT COUNT(*) FROM ` + table).Scan(&n); err != nil {
250
+
t.Fatalf("count %s: %v", table, err)
251
+
}
252
+
return n
253
+
}