Monorepo for Tangled
appview/db: entity-state backfill queries
Lewis: May this revision serve well! <lewis@tangled.org>
+266
+85
appview/db/entity_state_backfill.go
+85
appview/db/entity_state_backfill.go
···
1
+
package db
2
+
3
+
import (
4
+
"context"
5
+
6
+
"github.com/bluesky-social/indigo/atproto/syntax"
7
+
8
+
"tangled.org/core/appview/models"
9
+
)
10
+
11
+
const EntityStateBackfillName = "backfill-entity-state"
12
+
13
+
type BackfillSubject struct {
14
+
Subject syntax.ATURI
15
+
Value models.StateValue
16
+
CreatedAt string
17
+
}
18
+
19
+
func EnqueueEntityStateBackfill(ctx context.Context, e Execer) (int64, error) {
20
+
res, err := e.ExecContext(ctx, `
21
+
insert into pds_migration (name, did, collection, rkey)
22
+
select distinct ?, r.did, '', ''
23
+
from repos r
24
+
where r.did like 'did:%'
25
+
and (
26
+
exists (
27
+
select 1 from issues i
28
+
where i.repo_did = r.repo_did and i.open = 0 and i.deleted is null and i.rkey != ''
29
+
and not exists (select 1 from issue_states s where s.subject = i.at_uri)
30
+
)
31
+
or exists (
32
+
select 1 from pulls p
33
+
where p.repo_did = r.repo_did and p.state in (0, 2) and p.rkey != ''
34
+
and not exists (select 1 from pull_states s where s.subject = p.at_uri)
35
+
)
36
+
)
37
+
on conflict(name, did, collection, rkey) do update set
38
+
status = 'pending',
39
+
retry_count = 0,
40
+
retry_after = 0,
41
+
error_msg = null
42
+
where pds_migration.status in ('done', 'failed')
43
+
`, EntityStateBackfillName)
44
+
if err != nil {
45
+
return 0, err
46
+
}
47
+
return res.RowsAffected()
48
+
}
49
+
50
+
func ColumnOnlyClosedSubjectsForOwner(ctx context.Context, e Execer, owner syntax.DID) ([]BackfillSubject, error) {
51
+
rows, err := e.QueryContext(ctx, `
52
+
select i.at_uri, ?, i.created
53
+
from issues i join repos r on i.repo_did = r.repo_did
54
+
where r.did = ? and i.open = 0 and i.deleted is null and i.rkey != ''
55
+
and not exists (select 1 from issue_states s where s.subject = i.at_uri)
56
+
union all
57
+
select p.at_uri, case p.state when 2 then ? else ? end, p.created
58
+
from pulls p join repos r on p.repo_did = r.repo_did
59
+
where r.did = ? and p.state in (0, 2) and p.rkey != ''
60
+
and not exists (select 1 from pull_states s where s.subject = p.at_uri)
61
+
`,
62
+
string(models.StateClosed),
63
+
owner,
64
+
string(models.StateMerged), string(models.StateClosed),
65
+
owner,
66
+
)
67
+
if err != nil {
68
+
return nil, err
69
+
}
70
+
defer rows.Close()
71
+
72
+
var subjects []BackfillSubject
73
+
for rows.Next() {
74
+
var subject, value, created string
75
+
if err := rows.Scan(&subject, &value, &created); err != nil {
76
+
return nil, err
77
+
}
78
+
subjects = append(subjects, BackfillSubject{
79
+
Subject: syntax.ATURI(subject),
80
+
Value: models.StateValue(value),
81
+
CreatedAt: created,
82
+
})
83
+
}
84
+
return subjects, rows.Err()
85
+
}
+181
appview/db/entity_state_backfill_test.go
+181
appview/db/entity_state_backfill_test.go
···
1
+
package db
2
+
3
+
import (
4
+
"context"
5
+
"testing"
6
+
7
+
"github.com/bluesky-social/indigo/atproto/syntax"
8
+
"github.com/samber/lo"
9
+
"tangled.org/core/appview/models"
10
+
"tangled.org/core/orm"
11
+
)
12
+
13
+
func TestColumnOnlyClosedSubjectsForOwner(t *testing.T) {
14
+
d := newTestDB(t)
15
+
owner := syntax.DID("did:plc:akshay")
16
+
repo := seedRepo(t, d, string(owner), "knot.example", "anemone", "anemone", "did:plc:anemone")
17
+
18
+
closedIssue := seedIssue(t, d, repo, "did:plc:boltless", "issueClosed")
19
+
seedIssue(t, d, repo, "did:plc:boltless", "issueOpen")
20
+
recordedIssue := seedIssue(t, d, repo, "did:plc:boltless", "issueRecorded")
21
+
22
+
if err := CloseIssues(d, orm.FilterEq("at_uri", closedIssue.AtUri())); err != nil {
23
+
t.Fatalf("CloseIssues closed: %v", err)
24
+
}
25
+
if err := CloseIssues(d, orm.FilterEq("at_uri", recordedIssue.AtUri())); err != nil {
26
+
t.Fatalf("CloseIssues recorded: %v", err)
27
+
}
28
+
putIssueStateRec(t, d, issueRec("did:plc:boltless", "srec", recordedIssue.AtUri(), models.StateClosed, 100))
29
+
30
+
emptyRkey := seedIssue(t, d, repo, "did:plc:boltless", "")
31
+
if err := CloseIssues(d, orm.FilterEq("at_uri", emptyRkey.AtUri())); err != nil {
32
+
t.Fatalf("CloseIssues empty rkey: %v", err)
33
+
}
34
+
35
+
mergedPull := seedPull(t, d, repo, "did:plc:boltless", "pullMerged")
36
+
closedPull := seedPull(t, d, repo, "did:plc:boltless", "pullClosed")
37
+
seedPull(t, d, repo, "did:plc:boltless", "pullOpen")
38
+
39
+
if err := MergePulls(d, orm.FilterEq("at_uri", mergedPull.AtUri())); err != nil {
40
+
t.Fatalf("MergePulls: %v", err)
41
+
}
42
+
if err := ClosePulls(d, orm.FilterEq("at_uri", closedPull.AtUri())); err != nil {
43
+
t.Fatalf("ClosePulls: %v", err)
44
+
}
45
+
46
+
subjects, err := ColumnOnlyClosedSubjectsForOwner(context.Background(), d, owner)
47
+
if err != nil {
48
+
t.Fatalf("ColumnOnlyClosedSubjectsForOwner: %v", err)
49
+
}
50
+
51
+
got := lo.SliceToMap(subjects, func(s BackfillSubject) (syntax.ATURI, models.StateValue) {
52
+
return s.Subject, s.Value
53
+
})
54
+
55
+
if len(got) != 3 {
56
+
t.Fatalf("got %d subjects, want 3: %+v", len(got), got)
57
+
}
58
+
if got[closedIssue.AtUri()] != models.StateClosed {
59
+
t.Fatalf("closed issue: got %q want closed", got[closedIssue.AtUri()])
60
+
}
61
+
if got[mergedPull.AtUri()] != models.StateMerged {
62
+
t.Fatalf("merged pull: got %q want merged", got[mergedPull.AtUri()])
63
+
}
64
+
if got[closedPull.AtUri()] != models.StateClosed {
65
+
t.Fatalf("closed pull: got %q want closed", got[closedPull.AtUri()])
66
+
}
67
+
if _, ok := got[emptyRkey.AtUri()]; ok {
68
+
t.Fatal("closed issue with empty rkey must be excluded")
69
+
}
70
+
}
71
+
72
+
func mustEnqueueBackfill(t *testing.T, d *DB) {
73
+
t.Helper()
74
+
if _, err := EnqueueEntityStateBackfill(context.Background(), d); err != nil {
75
+
t.Fatalf("EnqueueEntityStateBackfill: %v", err)
76
+
}
77
+
}
78
+
79
+
func markBackfillDone(t *testing.T, d *DB) {
80
+
t.Helper()
81
+
if _, err := d.Exec(
82
+
`update pds_migration set status = 'done' where name = ?`, EntityStateBackfillName,
83
+
); err != nil {
84
+
t.Fatalf("mark done: %v", err)
85
+
}
86
+
}
87
+
88
+
func TestEnqueueEntityStateBackfill(t *testing.T) {
89
+
owner := syntax.DID("did:plc:akshay")
90
+
for _, tc := range []struct {
91
+
name string
92
+
arrange func(t *testing.T, d *DB, issue *models.Issue)
93
+
wantRows int64
94
+
wantNoRow bool
95
+
wantStatus models.PDSMigrationStatus
96
+
}{
97
+
{
98
+
name: "enqueues owner with column-only closed work",
99
+
arrange: func(t *testing.T, d *DB, issue *models.Issue) {},
100
+
wantRows: 1,
101
+
wantStatus: models.PDSMigrationStatusPending,
102
+
},
103
+
{
104
+
name: "idempotent while still pending",
105
+
arrange: func(t *testing.T, d *DB, issue *models.Issue) { mustEnqueueBackfill(t, d) },
106
+
wantRows: 0,
107
+
wantStatus: models.PDSMigrationStatusPending,
108
+
},
109
+
{
110
+
name: "re-arms completed owner with fresh work",
111
+
arrange: func(t *testing.T, d *DB, issue *models.Issue) {
112
+
mustEnqueueBackfill(t, d)
113
+
markBackfillDone(t, d)
114
+
},
115
+
wantRows: 1,
116
+
wantStatus: models.PDSMigrationStatusPending,
117
+
},
118
+
{
119
+
name: "leaves settled owner done",
120
+
arrange: func(t *testing.T, d *DB, issue *models.Issue) {
121
+
mustEnqueueBackfill(t, d)
122
+
putIssueStateRec(t, d, issueRec(string(owner), "srec", issue.AtUri(), models.StateClosed, 100))
123
+
markBackfillDone(t, d)
124
+
},
125
+
wantRows: 0,
126
+
wantStatus: models.PDSMigrationStatusDone,
127
+
},
128
+
{
129
+
name: "skips owner whose closed work is already recorded",
130
+
arrange: func(t *testing.T, d *DB, issue *models.Issue) {
131
+
putIssueStateRec(t, d, issueRec(string(owner), "srec", issue.AtUri(), models.StateClosed, 100))
132
+
},
133
+
wantRows: 0,
134
+
wantNoRow: true,
135
+
},
136
+
} {
137
+
t.Run(tc.name, func(t *testing.T) {
138
+
d := newTestDB(t)
139
+
repo := seedRepo(t, d, string(owner), "knot.example", "anemone", "anemone", "did:plc:anemone")
140
+
issue := seedIssue(t, d, repo, "did:plc:boltless", "issue1")
141
+
if err := CloseIssues(d, orm.FilterEq("at_uri", issue.AtUri())); err != nil {
142
+
t.Fatalf("CloseIssues: %v", err)
143
+
}
144
+
tc.arrange(t, d, issue)
145
+
146
+
n, err := EnqueueEntityStateBackfill(context.Background(), d)
147
+
if err != nil {
148
+
t.Fatalf("EnqueueEntityStateBackfill: %v", err)
149
+
}
150
+
if n != tc.wantRows {
151
+
t.Fatalf("enqueue affected %d rows, want %d", n, tc.wantRows)
152
+
}
153
+
154
+
if tc.wantNoRow {
155
+
var count int
156
+
if err := d.QueryRow(
157
+
`select count(*) from pds_migration where name = ?`, EntityStateBackfillName,
158
+
).Scan(&count); err != nil {
159
+
t.Fatalf("query: %v", err)
160
+
}
161
+
if count != 0 {
162
+
t.Fatalf("owner with no column-only closed work produced %d rows, want 0", count)
163
+
}
164
+
return
165
+
}
166
+
167
+
var did, status string
168
+
if err := d.QueryRow(
169
+
`select did, status from pds_migration where name = ?`, EntityStateBackfillName,
170
+
).Scan(&did, &status); err != nil {
171
+
t.Fatalf("query: %v", err)
172
+
}
173
+
if syntax.DID(did) != owner {
174
+
t.Fatalf("row did = %s, want owner %s", did, owner)
175
+
}
176
+
if status != string(tc.wantStatus) {
177
+
t.Fatalf("row status = %s, want %s", status, tc.wantStatus)
178
+
}
179
+
})
180
+
}
181
+
}