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