alice.pds.demo.boltless.dev
This repository has no description
1package db
2
3import (
4 "testing"
5
6 "github.com/bluesky-social/indigo/atproto/syntax"
7 "tangled.org/core/appview/models"
8 "tangled.org/core/orm"
9)
10
11func seedIssue(t *testing.T, d *DB, repo *models.Repo, did, rkey string) *models.Issue {
12 t.Helper()
13 tx, err := d.Begin()
14 if err != nil {
15 t.Fatalf("Begin: %v", err)
16 }
17 issue := &models.Issue{
18 Did: did,
19 Rkey: rkey,
20 RepoDid: syntax.DID(repo.RepoDid),
21 Title: "title",
22 Body: "body",
23 Open: true,
24 }
25 if err := PutIssue(tx, issue); err != nil {
26 t.Fatalf("PutIssue: %v", err)
27 }
28 if err := tx.Commit(); err != nil {
29 t.Fatalf("Commit: %v", err)
30 }
31 return issue
32}
33
34func seedPull(t *testing.T, d *DB, repo *models.Repo, did, rkey string) *models.Pull {
35 t.Helper()
36 tx, err := d.Begin()
37 if err != nil {
38 t.Fatalf("Begin: %v", err)
39 }
40 pull := &models.Pull{
41 RepoDid: syntax.DID(repo.RepoDid),
42 OwnerDid: syntax.DID(did),
43 Rkey: syntax.RecordKey(rkey),
44 Title: "title",
45 Body: "body",
46 TargetBranch: "main",
47 State: models.PullOpen,
48 Versions: []models.PullVersion{
49 {Head: "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa"},
50 },
51 }
52 if err := PutPull(t.Context(), tx, pull, nil); err != nil {
53 t.Fatalf("PutPull: %v", err)
54 }
55 if err := tx.Commit(); err != nil {
56 t.Fatalf("Commit: %v", err)
57 }
58 return pull
59}
60
61func putIssueStateRec(t *testing.T, d *DB, rec models.StateRecord) {
62 t.Helper()
63 tx, err := d.Begin()
64 if err != nil {
65 t.Fatalf("Begin: %v", err)
66 }
67 if _, err := PutIssueState(tx, rec); err != nil {
68 t.Fatalf("PutIssueState: %v", err)
69 }
70 if err := ResolveIssueState(tx, rec.Subject); err != nil {
71 t.Fatalf("ResolveIssueState: %v", err)
72 }
73 if err := tx.Commit(); err != nil {
74 t.Fatalf("Commit: %v", err)
75 }
76}
77
78func deleteIssueStateRec(t *testing.T, d *DB, did, rkey string) {
79 t.Helper()
80 tx, err := d.Begin()
81 if err != nil {
82 t.Fatalf("Begin: %v", err)
83 }
84 subject, err := DeleteIssueState(tx, did, rkey)
85 if err != nil {
86 t.Fatalf("DeleteIssueState: %v", err)
87 }
88 if subject != "" {
89 if err := RecomputeIssueState(tx, subject); err != nil {
90 t.Fatalf("RecomputeIssueState: %v", err)
91 }
92 }
93 if err := tx.Commit(); err != nil {
94 t.Fatalf("Commit: %v", err)
95 }
96}
97
98func putPullStatusRec(t *testing.T, d *DB, rec models.StateRecord) {
99 t.Helper()
100 tx, err := d.Begin()
101 if err != nil {
102 t.Fatalf("Begin: %v", err)
103 }
104 if _, err := PutPullStatus(tx, rec); err != nil {
105 t.Fatalf("PutPullStatus: %v", err)
106 }
107 if err := ResolvePullStatus(tx, rec.Subject); err != nil {
108 t.Fatalf("ResolvePullStatus: %v", err)
109 }
110 if err := tx.Commit(); err != nil {
111 t.Fatalf("Commit: %v", err)
112 }
113}
114
115func issueOpen(t *testing.T, d *DB, subject syntax.ATURI) bool {
116 t.Helper()
117 issues, err := GetIssues(d, orm.FilterEq("at_uri", subject))
118 if err != nil || len(issues) != 1 {
119 t.Fatalf("GetIssues: %v len %d", err, len(issues))
120 }
121 return issues[0].Open
122}
123
124func pullStateOf(t *testing.T, d *DB, subject syntax.ATURI) models.PullState {
125 t.Helper()
126 pull, err := GetPull(t.Context(), d, orm.FilterEq("at_uri", subject))
127 if err != nil {
128 t.Fatalf("GetPulls: %v", err)
129 }
130 return pull.State
131}
132
133func issueRec(did, rkey string, subject syntax.ATURI, v models.StateValue, micros int64) models.StateRecord {
134 return models.StateRecord{Did: did, Rkey: rkey, Subject: subject, Value: v, SortMicros: micros}
135}
136
137func TestIssueStateLastWriterWins(t *testing.T) {
138 d := newTestDB(t)
139 repo := seedRepo(t, d, "did:plc:akshay", "knot.example", "anemone", "anemone", "did:plc:anemone")
140 issue := seedIssue(t, d, repo, "did:plc:akshay", "issue1")
141 subject := issue.AtUri()
142
143 if !issueOpen(t, d, subject) {
144 t.Fatal("new issue should be open")
145 }
146
147 putIssueStateRec(t, d, issueRec("did:plc:akshay", "s2", subject, models.StateClosed, 200))
148 putIssueStateRec(t, d, issueRec("did:plc:akshay", "s1", subject, models.StateOpen, 100))
149 if issueOpen(t, d, subject) {
150 t.Fatal("earlier open@100 must not beat closed@200")
151 }
152
153 putIssueStateRec(t, d, issueRec("did:plc:akshay", "s3", subject, models.StateOpen, 300))
154 if !issueOpen(t, d, subject) {
155 t.Fatal("open@300 should win")
156 }
157
158 putIssueStateRec(t, d, issueRec("did:plc:akshay", "zzz", subject, models.StateClosed, 300))
159 if issueOpen(t, d, subject) {
160 t.Fatal("a tie at 300 must break to the greater source uri zzz=closed")
161 }
162
163 putIssueStateRec(t, d, issueRec("did:plc:akshay", "zzz", subject, models.StateClosed, 300))
164 if issueOpen(t, d, subject) {
165 t.Fatal("replaying the winning record must not change the result")
166 }
167 var count int
168 if err := d.QueryRow(`select count(*) from issue_states where subject = ?`, subject).Scan(&count); err != nil {
169 t.Fatalf("count: %v", err)
170 }
171 if count != 4 {
172 t.Fatalf("replaying an existing record must not duplicate rows, got %d want 4", count)
173 }
174}
175
176func TestIssueStateOrderIndependent(t *testing.T) {
177 build := func(order []int) bool {
178 d := newTestDB(t)
179 repo := seedRepo(t, d, "did:plc:akshay", "knot.example", "anemone", "anemone", "did:plc:anemone")
180 issue := seedIssue(t, d, repo, "did:plc:akshay", "issue1")
181 subject := issue.AtUri()
182
183 recs := []models.StateRecord{
184 issueRec("did:plc:akshay", "a", subject, models.StateOpen, 100),
185 issueRec("did:plc:akshay", "b", subject, models.StateClosed, 300),
186 issueRec("did:plc:akshay", "c", subject, models.StateOpen, 200),
187 }
188 for _, idx := range order {
189 putIssueStateRec(t, d, recs[idx])
190 }
191 return issueOpen(t, d, subject)
192 }
193
194 forward := build([]int{0, 1, 2})
195 shuffled := build([]int{2, 0, 1})
196 if forward != shuffled {
197 t.Fatalf("order changed result: forward=%v shuffled=%v", forward, shuffled)
198 }
199 if forward {
200 t.Fatal("highest-micros record closed@300 must win regardless of order")
201 }
202}
203
204func TestIssueStateDeleteRecomputes(t *testing.T) {
205 d := newTestDB(t)
206 repo := seedRepo(t, d, "did:plc:akshay", "knot.example", "anemone", "anemone", "did:plc:anemone")
207 issue := seedIssue(t, d, repo, "did:plc:akshay", "issue1")
208 subject := issue.AtUri()
209
210 putIssueStateRec(t, d, issueRec("did:plc:akshay", "s1", subject, models.StateClosed, 100))
211 putIssueStateRec(t, d, issueRec("did:plc:akshay", "s2", subject, models.StateOpen, 200))
212 if !issueOpen(t, d, subject) {
213 t.Fatal("open@200 should win before deletion")
214 }
215
216 deleteIssueStateRec(t, d, "did:plc:akshay", "s2")
217 if issueOpen(t, d, subject) {
218 t.Fatal("deleting open@200 must fall back to closed@100")
219 }
220
221 deleteIssueStateRec(t, d, "did:plc:akshay", "s1")
222 if !issueOpen(t, d, subject) {
223 t.Fatal("deleting the last state record must revert to open")
224 }
225}
226
227func TestIssueStateSubjectChangeRecomputesPrior(t *testing.T) {
228 d := newTestDB(t)
229 repo := seedRepo(t, d, "did:plc:akshay", "knot.example", "anemone", "anemone", "did:plc:anemone")
230 issueA := seedIssue(t, d, repo, "did:plc:akshay", "issueA")
231 issueB := seedIssue(t, d, repo, "did:plc:akshay", "issueB")
232
233 putIssueStateRec(t, d, issueRec("did:plc:akshay", "s1", issueA.AtUri(), models.StateClosed, 100))
234 if issueOpen(t, d, issueA.AtUri()) {
235 t.Fatal("issueA should be closed after closed@100")
236 }
237
238 tx, err := d.Begin()
239 if err != nil {
240 t.Fatalf("Begin: %v", err)
241 }
242 prior, err := PutIssueState(tx, issueRec("did:plc:akshay", "s1", issueB.AtUri(), models.StateClosed, 200))
243 if err != nil {
244 t.Fatalf("PutIssueState: %v", err)
245 }
246 if prior != issueA.AtUri() {
247 t.Fatalf("put must report the prior subject %s, got %q", issueA.AtUri(), prior)
248 }
249 if err := ResolveIssueState(tx, issueB.AtUri()); err != nil {
250 t.Fatalf("ResolveIssueState B: %v", err)
251 }
252 if err := RecomputeIssueState(tx, prior); err != nil {
253 t.Fatalf("RecomputeIssueState A: %v", err)
254 }
255 if err := tx.Commit(); err != nil {
256 t.Fatalf("Commit: %v", err)
257 }
258
259 if !issueOpen(t, d, issueA.AtUri()) {
260 t.Fatal("issueA must revert to open once its only state record repoints to issueB")
261 }
262 if issueOpen(t, d, issueB.AtUri()) {
263 t.Fatal("issueB should be closed after the record repoints to it")
264 }
265}
266
267func TestPullStatusLastWriterWins(t *testing.T) {
268 d := newTestDB(t)
269 repo := seedRepo(t, d, "did:plc:akshay", "knot.example", "limpet", "limpet", "did:plc:limpet")
270 pull := seedPull(t, d, repo, "did:plc:akshay", "pull1")
271 subject := pull.AtUri()
272
273 if pullStateOf(t, d, subject) != models.PullOpen {
274 t.Fatal("new pull should be open")
275 }
276
277 putPullStatusRec(t, d, issueRec("did:plc:akshay", "s1", subject, models.StateClosed, 100))
278 if pullStateOf(t, d, subject) != models.PullClosed {
279 t.Fatal("closed@100 should win")
280 }
281
282 putPullStatusRec(t, d, issueRec("did:plc:akshay", "s2", subject, models.StateMerged, 200))
283 if pullStateOf(t, d, subject) != models.PullMerged {
284 t.Fatal("merged@200 should win over closed@100")
285 }
286
287 putPullStatusRec(t, d, issueRec("did:plc:akshay", "s3", subject, models.StateOpen, 300))
288 if pullStateOf(t, d, subject) != models.PullOpen {
289 t.Fatal("open@300 should win over merged@200")
290 }
291
292 if err := AbandonPulls(d, orm.FilterEq("at_uri", subject)); err != nil {
293 t.Fatalf("AbandonPulls: %v", err)
294 }
295 putPullStatusRec(t, d, issueRec("did:plc:akshay", "s4", subject, models.StateOpen, 400))
296 if pullStateOf(t, d, subject) != models.PullAbandoned {
297 t.Fatal("an abandoned pull must not be resurrected by a later status record")
298 }
299}
300
301func TestIssueStateForeignKey(t *testing.T) {
302 d := newTestDB(t)
303 repo := seedRepo(t, d, "did:plc:akshay", "knot.example", "anemone", "anemone", "did:plc:anemone")
304
305 tx, err := d.Begin()
306 if err != nil {
307 t.Fatalf("Begin: %v", err)
308 }
309 ghost := syntax.ATURI("at://did:plc:akshay/sh.tangled.repo.issue/ghost")
310 if _, err := PutIssueState(tx, issueRec("did:plc:akshay", "s1", ghost, models.StateClosed, 100)); err == nil {
311 t.Fatal("inserting state for a nonexistent issue must violate the foreign key")
312 }
313 tx.Rollback()
314
315 issue := seedIssue(t, d, repo, "did:plc:akshay", "issue1")
316 subject := issue.AtUri()
317 putIssueStateRec(t, d, issueRec("did:plc:akshay", "s1", subject, models.StateClosed, 100))
318
319 dtx, err := d.Begin()
320 if err != nil {
321 t.Fatalf("Begin: %v", err)
322 }
323 if err := DeleteIssues(dtx, "did:plc:akshay", "issue1"); err != nil {
324 t.Fatalf("DeleteIssues: %v", err)
325 }
326 if err := dtx.Commit(); err != nil {
327 t.Fatalf("Commit: %v", err)
328 }
329
330 var remaining int
331 if err := d.QueryRow(`select count(*) from issue_states where subject = ?`, subject).Scan(&remaining); err != nil {
332 t.Fatalf("count: %v", err)
333 }
334 if remaining != 0 {
335 t.Fatalf("deleting the issue must cascade-delete its state rows, got %d", remaining)
336 }
337}
338
339func TestPendingStateRecords(t *testing.T) {
340 d := newTestDB(t)
341 s1 := syntax.ATURI("at://did:plc:boltless/sh.tangled.repo.issue/i1")
342 s2 := syntax.ATURI("at://did:plc:akshay/sh.tangled.repo.pull/p1")
343 issueNsid := "sh.tangled.repo.issue.state"
344
345 park := func(did, rkey, nsid string, subject syntax.ATURI, record string) {
346 t.Helper()
347 tx, err := d.Begin()
348 if err != nil {
349 t.Fatalf("Begin: %v", err)
350 }
351 if err := ParkStateRecord(tx, PendingStateRecord{
352 Did: did, Rkey: rkey, Nsid: nsid, Subject: subject, Record: []byte(record),
353 }); err != nil {
354 t.Fatalf("ParkStateRecord: %v", err)
355 }
356 if err := tx.Commit(); err != nil {
357 t.Fatalf("Commit: %v", err)
358 }
359 }
360
361 park("did:plc:boltless", "s1", issueNsid, s1, `{"v":1}`)
362 park("did:plc:boltless", "s1", issueNsid, s1, `{"v":2}`)
363 park("did:plc:akshay", "p1", "sh.tangled.repo.pull.status", s2, `{}`)
364
365 pending, err := PendingStateRecordsForSubject(d, s1)
366 if err != nil {
367 t.Fatalf("query: %v", err)
368 }
369 if len(pending) != 1 || pending[0].Did != "did:plc:boltless" || string(pending[0].Record) != `{"v":2}` {
370 t.Fatalf("re-park must overwrite without duplicating, got %+v", pending)
371 }
372
373 subjects, err := DistinctPendingStateSubjects(d)
374 if err != nil {
375 t.Fatalf("DistinctPendingStateSubjects: %v", err)
376 }
377 if len(subjects) != 2 {
378 t.Fatalf("want 2 distinct subjects from 3 parked rows, got %d", len(subjects))
379 }
380
381 tx, err := d.Begin()
382 if err != nil {
383 t.Fatalf("Begin: %v", err)
384 }
385 if err := UnparkStateRecord(tx, "did:plc:boltless", "s1", issueNsid); err != nil {
386 t.Fatalf("UnparkStateRecord: %v", err)
387 }
388 if err := tx.Commit(); err != nil {
389 t.Fatalf("Commit: %v", err)
390 }
391 if pending, _ := PendingStateRecordsForSubject(d, s1); len(pending) != 0 {
392 t.Fatalf("want 0 after unpark, got %d", len(pending))
393 }
394}
395
396func TestEvictStalePendingStateRecords(t *testing.T) {
397 d := newTestDB(t)
398 repo := seedRepo(t, d, "did:plc:akshay", "knot.example", "anemone", "anemone", "did:plc:anemone")
399 issue := seedIssue(t, d, repo, "did:plc:akshay", "issue1")
400 live := issue.AtUri()
401 orphan := syntax.ATURI("at://did:plc:boltless/sh.tangled.repo.issue/ghost")
402
403 insert := func(rkey, created string, subject syntax.ATURI) {
404 t.Helper()
405 if _, err := d.Exec(
406 `insert into pending_state_records (did, rkey, nsid, subject, record, created) values (?, ?, ?, ?, ?, ?)`,
407 "did:plc:boltless", rkey, "sh.tangled.repo.issue.state", string(subject), []byte("{}"), created,
408 ); err != nil {
409 t.Fatalf("insert %s: %v", rkey, err)
410 }
411 }
412
413 insert("stale-orphan", "2000-01-01T00:00:00Z", orphan)
414 insert("fresh-orphan", "2999-01-01T00:00:00Z", orphan)
415 insert("stale-live", "2000-01-01T00:00:00Z", live)
416
417 evicted, err := EvictStalePendingStateRecords(d, "2026-01-01T00:00:00Z")
418 if err != nil {
419 t.Fatalf("EvictStalePendingStateRecords: %v", err)
420 }
421 if evicted != 2 {
422 t.Fatalf("both stale rows must be evicted regardless of subject presence, got %d want 2", evicted)
423 }
424
425 var remaining int
426 if err := d.QueryRow(`select count(*) from pending_state_records`).Scan(&remaining); err != nil {
427 t.Fatalf("count: %v", err)
428 }
429 if remaining != 1 {
430 t.Fatalf("only the fresh row must survive the TTL sweep, got %d want 1", remaining)
431 }
432}