Monorepo for Tangled
appview/db: issue/pull state tables, lww resolver, & pending-records
Lewis: May this revision serve well! <lewis@tangled.org>
+308
+49
appview/db/db.go
+49
appview/db/db.go
···
2372
2372
return err
2373
2373
})
2374
2374
2375
+
orm.RunMigration(conn, logger, "add-issue-pull-state-tables", func(tx *sql.Tx) error {
2376
+
_, err := tx.Exec(`
2377
+
create table if not exists issue_states (
2378
+
id integer primary key autoincrement,
2379
+
did text not null,
2380
+
rkey text not null,
2381
+
at_uri text generated always as ('at://' || did || '/' || 'sh.tangled.repo.issue.state' || '/' || rkey) stored,
2382
+
2383
+
subject text not null,
2384
+
state text not null check (state in ('open', 'closed')),
2385
+
created_micros integer not null,
2386
+
2387
+
unique(did, rkey),
2388
+
foreign key (subject) references issues(at_uri) on delete cascade
2389
+
);
2390
+
create index if not exists idx_issue_states_subject on issue_states(subject);
2391
+
2392
+
create table if not exists pull_states (
2393
+
id integer primary key autoincrement,
2394
+
did text not null,
2395
+
rkey text not null,
2396
+
at_uri text generated always as ('at://' || did || '/' || 'sh.tangled.repo.pull.status' || '/' || rkey) stored,
2397
+
2398
+
subject text not null,
2399
+
status text not null check (status in ('open', 'closed', 'merged')),
2400
+
created_micros integer not null,
2401
+
2402
+
unique(did, rkey),
2403
+
foreign key (subject) references pulls(at_uri) on delete cascade
2404
+
);
2405
+
create index if not exists idx_pull_states_subject on pull_states(subject);
2406
+
2407
+
create table if not exists pending_state_records (
2408
+
id integer primary key autoincrement,
2409
+
did text not null,
2410
+
rkey text not null,
2411
+
nsid text not null,
2412
+
2413
+
subject text not null,
2414
+
record blob not null,
2415
+
created text not null default (strftime('%Y-%m-%dT%H:%M:%SZ', 'now')),
2416
+
2417
+
unique(did, rkey, nsid)
2418
+
);
2419
+
create index if not exists idx_pending_state_subject on pending_state_records(subject);
2420
+
`)
2421
+
return err
2422
+
})
2423
+
2375
2424
return &DB{
2376
2425
db,
2377
2426
logger,
+259
appview/db/entity_state.go
+259
appview/db/entity_state.go
···
1
+
package db
2
+
3
+
import (
4
+
"database/sql"
5
+
"errors"
6
+
"fmt"
7
+
8
+
"github.com/bluesky-social/indigo/atproto/syntax"
9
+
"tangled.org/core/appview/models"
10
+
)
11
+
12
+
type stateTable struct {
13
+
name string
14
+
valueCol string
15
+
}
16
+
17
+
var (
18
+
issueStateTable = stateTable{name: "issue_states", valueCol: "state"}
19
+
pullStateTable = stateTable{name: "pull_states", valueCol: "status"}
20
+
)
21
+
22
+
func putStateRecord(tx *sql.Tx, t stateTable, rec models.StateRecord) (syntax.ATURI, error) {
23
+
var priorSubject string
24
+
err := tx.QueryRow(
25
+
fmt.Sprintf(`select subject from %s where did = ? and rkey = ?`, t.name),
26
+
rec.Did, rec.Rkey,
27
+
).Scan(&priorSubject)
28
+
switch {
29
+
case errors.Is(err, sql.ErrNoRows):
30
+
priorSubject = ""
31
+
case err != nil:
32
+
return "", err
33
+
}
34
+
35
+
if _, err := tx.Exec(fmt.Sprintf(`
36
+
insert into %s (did, rkey, subject, %s, created_micros)
37
+
values (?, ?, ?, ?, ?)
38
+
on conflict(did, rkey) do update set
39
+
subject = excluded.subject,
40
+
%s = excluded.%s,
41
+
created_micros = excluded.created_micros
42
+
`, t.name, t.valueCol, t.valueCol, t.valueCol),
43
+
rec.Did, rec.Rkey, rec.Subject, string(rec.Value), rec.SortMicros); err != nil {
44
+
return "", err
45
+
}
46
+
47
+
if priorSubject != "" && syntax.ATURI(priorSubject) != rec.Subject {
48
+
return syntax.ATURI(priorSubject), nil
49
+
}
50
+
return "", nil
51
+
}
52
+
53
+
func deleteStateRecord(tx *sql.Tx, t stateTable, did, rkey string) (syntax.ATURI, error) {
54
+
var subject string
55
+
err := tx.QueryRow(
56
+
fmt.Sprintf(`select subject from %s where did = ? and rkey = ?`, t.name),
57
+
did, rkey,
58
+
).Scan(&subject)
59
+
switch {
60
+
case errors.Is(err, sql.ErrNoRows):
61
+
return "", nil
62
+
case err != nil:
63
+
return "", err
64
+
}
65
+
66
+
if _, err := tx.Exec(
67
+
fmt.Sprintf(`delete from %s where did = ? and rkey = ?`, t.name),
68
+
did, rkey,
69
+
); err != nil {
70
+
return "", err
71
+
}
72
+
return syntax.ATURI(subject), nil
73
+
}
74
+
75
+
func stateWinner(e Execer, t stateTable, subject syntax.ATURI) (models.StateValue, bool, error) {
76
+
var v string
77
+
err := e.QueryRow(fmt.Sprintf(`
78
+
select %s from %s
79
+
where subject = ?
80
+
order by created_micros desc, at_uri desc
81
+
limit 1
82
+
`, t.valueCol, t.name), subject).Scan(&v)
83
+
switch {
84
+
case errors.Is(err, sql.ErrNoRows):
85
+
return "", false, nil
86
+
case err != nil:
87
+
return "", false, err
88
+
}
89
+
return models.StateValue(v), true, nil
90
+
}
91
+
92
+
func PutIssueState(tx *sql.Tx, rec models.StateRecord) (syntax.ATURI, error) {
93
+
return putStateRecord(tx, issueStateTable, rec)
94
+
}
95
+
96
+
func PutPullStatus(tx *sql.Tx, rec models.StateRecord) (syntax.ATURI, error) {
97
+
return putStateRecord(tx, pullStateTable, rec)
98
+
}
99
+
100
+
type PendingStateRecord struct {
101
+
Did string
102
+
Rkey string
103
+
Nsid string
104
+
Subject syntax.ATURI
105
+
Record []byte
106
+
}
107
+
108
+
func ParkStateRecord(tx *sql.Tx, p PendingStateRecord) error {
109
+
_, err := tx.Exec(`
110
+
insert into pending_state_records (did, rkey, nsid, subject, record)
111
+
values (?, ?, ?, ?, ?)
112
+
on conflict(did, rkey, nsid) do update set
113
+
subject = excluded.subject,
114
+
record = excluded.record
115
+
`, p.Did, p.Rkey, p.Nsid, p.Subject, p.Record)
116
+
return err
117
+
}
118
+
119
+
func UnparkStateRecord(tx *sql.Tx, did, rkey, nsid string) error {
120
+
_, err := tx.Exec(
121
+
`delete from pending_state_records where did = ? and rkey = ? and nsid = ?`,
122
+
did, rkey, nsid,
123
+
)
124
+
return err
125
+
}
126
+
127
+
func DistinctPendingStateSubjects(e Execer) ([]syntax.ATURI, error) {
128
+
rows, err := e.Query(`select distinct subject from pending_state_records order by subject asc`)
129
+
if err != nil {
130
+
return nil, err
131
+
}
132
+
defer rows.Close()
133
+
134
+
var subjects []syntax.ATURI
135
+
for rows.Next() {
136
+
var s string
137
+
if err := rows.Scan(&s); err != nil {
138
+
return nil, err
139
+
}
140
+
subjects = append(subjects, syntax.ATURI(s))
141
+
}
142
+
return subjects, rows.Err()
143
+
}
144
+
145
+
func EvictStalePendingStateRecords(e Execer, before string) (int64, error) {
146
+
res, err := e.Exec(`
147
+
delete from pending_state_records
148
+
where created < ?
149
+
and not exists (select 1 from issues where at_uri = pending_state_records.subject)
150
+
and not exists (select 1 from pulls where at_uri = pending_state_records.subject)
151
+
`, before)
152
+
if err != nil {
153
+
return 0, err
154
+
}
155
+
return res.RowsAffected()
156
+
}
157
+
158
+
func PendingStateRecordsForSubject(e Execer, subject syntax.ATURI) ([]PendingStateRecord, error) {
159
+
rows, err := e.Query(
160
+
`select did, rkey, nsid, subject, record from pending_state_records where subject = ? order by id asc`,
161
+
subject,
162
+
)
163
+
if err != nil {
164
+
return nil, err
165
+
}
166
+
defer rows.Close()
167
+
168
+
var pending []PendingStateRecord
169
+
for rows.Next() {
170
+
var p PendingStateRecord
171
+
var subj string
172
+
if err := rows.Scan(&p.Did, &p.Rkey, &p.Nsid, &subj, &p.Record); err != nil {
173
+
return nil, err
174
+
}
175
+
p.Subject = syntax.ATURI(subj)
176
+
pending = append(pending, p)
177
+
}
178
+
return pending, rows.Err()
179
+
}
180
+
181
+
func DeleteIssueState(tx *sql.Tx, did, rkey string) (syntax.ATURI, error) {
182
+
return deleteStateRecord(tx, issueStateTable, did, rkey)
183
+
}
184
+
185
+
func DeletePullStatus(tx *sql.Tx, did, rkey string) (syntax.ATURI, error) {
186
+
return deleteStateRecord(tx, pullStateTable, did, rkey)
187
+
}
188
+
189
+
func setIssueOpen(tx *sql.Tx, subject syntax.ATURI, open bool) error {
190
+
v := 0
191
+
if open {
192
+
v = 1
193
+
}
194
+
_, err := tx.Exec(`update issues set open = ? where at_uri = ?`, v, subject)
195
+
return err
196
+
}
197
+
198
+
func applyIssueState(tx *sql.Tx, subject syntax.ATURI, resetWhenEmpty bool) error {
199
+
winner, ok, err := stateWinner(tx, issueStateTable, subject)
200
+
if err != nil {
201
+
return err
202
+
}
203
+
if !ok {
204
+
if resetWhenEmpty {
205
+
return setIssueOpen(tx, subject, true)
206
+
}
207
+
return nil
208
+
}
209
+
return setIssueOpen(tx, subject, winner == models.StateOpen)
210
+
}
211
+
212
+
func ResolveIssueState(tx *sql.Tx, subject syntax.ATURI) error {
213
+
return applyIssueState(tx, subject, false)
214
+
}
215
+
216
+
func RecomputeIssueState(tx *sql.Tx, subject syntax.ATURI) error {
217
+
return applyIssueState(tx, subject, true)
218
+
}
219
+
220
+
func pullStateFromValue(v models.StateValue) models.PullState {
221
+
switch v {
222
+
case models.StateMerged:
223
+
return models.PullMerged
224
+
case models.StateClosed:
225
+
return models.PullClosed
226
+
default:
227
+
return models.PullOpen
228
+
}
229
+
}
230
+
231
+
func setPullState(tx *sql.Tx, subject syntax.ATURI, st models.PullState) error {
232
+
_, err := tx.Exec(
233
+
`update pulls set state = ? where at_uri = ? and state <> ?`,
234
+
st, subject, models.PullAbandoned,
235
+
)
236
+
return err
237
+
}
238
+
239
+
func applyPullStatus(tx *sql.Tx, subject syntax.ATURI, resetWhenEmpty bool) error {
240
+
winner, ok, err := stateWinner(tx, pullStateTable, subject)
241
+
if err != nil {
242
+
return err
243
+
}
244
+
if !ok {
245
+
if resetWhenEmpty {
246
+
return setPullState(tx, subject, models.PullOpen)
247
+
}
248
+
return nil
249
+
}
250
+
return setPullState(tx, subject, pullStateFromValue(winner))
251
+
}
252
+
253
+
func ResolvePullStatus(tx *sql.Tx, subject syntax.ATURI) error {
254
+
return applyPullStatus(tx, subject, false)
255
+
}
256
+
257
+
func RecomputePullStatus(tx *sql.Tx, subject syntax.ATURI) error {
258
+
return applyPullStatus(tx, subject, true)
259
+
}