alice.pds.demo.boltless.dev
This repository has no description
1package db
2
3import (
4 "context"
5 "database/sql"
6 "log/slog"
7 "slices"
8 "strings"
9
10 _ "github.com/mattn/go-sqlite3"
11 "tangled.org/core/log"
12 "tangled.org/core/orm"
13)
14
15type DB struct {
16 *sql.DB
17}
18
19type DBTX interface {
20 QueryRow(query string, args ...any) *sql.Row
21 Exec(query string, args ...any) (sql.Result, error)
22}
23
24func Make(ctx context.Context, dbPath string) (*DB, error) {
25 // https://github.com/mattn/go-sqlite3#connection-string
26 opts := []string{
27 "_foreign_keys=1",
28 "_journal_mode=WAL",
29 "_synchronous=NORMAL",
30 "_auto_vacuum=incremental",
31 "_busy_timeout=5000",
32 }
33
34 logger := log.FromContext(ctx)
35 logger = log.SubLogger(logger, "db")
36
37 db, err := sql.Open("sqlite3", dbPath+"?"+strings.Join(opts, "&"))
38 if err != nil {
39 return nil, err
40 }
41
42 conn, err := db.Conn(ctx)
43 if err != nil {
44 return nil, err
45 }
46 defer conn.Close()
47
48 _, err = conn.ExecContext(ctx, `
49 create table if not exists _jetstream (
50 id integer primary key autoincrement,
51 last_time_us integer not null
52 );
53
54 create table if not exists known_dids (
55 did text primary key
56 );
57
58 create table if not exists repos (
59 id integer primary key autoincrement,
60 knot text not null,
61 owner text not null,
62 rkey text not null,
63 repo_did text,
64 created_at text,
65 addedAt text not null default (strftime('%Y-%m-%dT%H:%M:%SZ', 'now')),
66
67 unique(owner, rkey)
68 );
69
70 create table if not exists repo_collaborators (
71 id integer primary key autoincrement,
72 owner_did text not null,
73 rkey text not null,
74 subject text not null,
75 repo_did text not null,
76 addedAt text not null default (strftime('%Y-%m-%dT%H:%M:%SZ', 'now')),
77
78 unique(owner_did, rkey)
79 );
80
81 create table if not exists spindle_members (
82 -- identifiers for the record
83 id integer primary key autoincrement,
84 did text not null,
85 rkey text not null,
86
87 -- data
88 instance text not null,
89 subject text not null,
90 created text not null default (strftime('%Y-%m-%dT%H:%M:%SZ', 'now')),
91
92 -- constraints
93 unique (did, rkey)
94 );
95
96 -- status event for a single workflow
97 create table if not exists events (
98 rkey text not null,
99 nsid text not null,
100 event text not null, -- json
101 created integer not null -- unix nanos
102 );
103
104 create table if not exists nixos_toplevel_cache (
105 config_key text primary key,
106 toplevel text not null,
107 updated_at text not null
108 );
109
110 create table if not exists pipelines (
111 id text primary key,
112 repo_did text not null,
113 commit_id text not null
114 );
115 create table if not exists jobs (
116 id integer primary key autoincrement,
117 repo_did text not null,
118 pipeline_id_knot text not null,
119 pipeline_id_rkey text not null,
120 source_repo text,
121 tpl text not null,
122 created_at integer not null default (strftime('%s', 'now'))
123 );
124
125 create table if not exists workflows (
126 id integer primary key autoincrement,
127 pipeline_id text not null,
128 name text not null,
129 status text not null default 'pending',
130
131 unique(pipeline_id, id),
132 foreign key (pipeline_id) references pipelines(id) on delete cascade
133 );
134
135 create table if not exists migrations (
136 id integer primary key autoincrement,
137 name text unique
138 );
139 `)
140 if err != nil {
141 return nil, err
142 }
143
144 if err := runMigrations(ctx, conn, logger); err != nil {
145 return nil, err
146 }
147
148 return &DB{db}, nil
149}
150
151func runMigrations(_ context.Context, conn *sql.Conn, logger *slog.Logger) error {
152 if err := orm.RunMigration(conn, logger, "repos-to-repo-did", func(tx *sql.Tx) error {
153 var hasName int
154 if err := tx.QueryRow(
155 `select count(*) from pragma_table_info('repos') where name = 'name'`,
156 ).Scan(&hasName); err != nil {
157 return err
158 }
159
160 if hasName > 0 {
161 var totalRows, copiedRows int
162 if err := tx.QueryRow(`select count(*) from repos`).Scan(&totalRows); err != nil {
163 return err
164 }
165 if err := tx.QueryRow(`select count(*) from repos where coalesce(name, '') <> ''`).Scan(&copiedRows); err != nil {
166 return err
167 }
168 if dropped := totalRows - copiedRows; dropped > 0 {
169 logger.Warn("dropping repo rows with empty name during migration", "dropped", dropped, "kept", copiedRows)
170 }
171
172 if _, err := tx.Exec(`
173 create table if not exists repos_new (
174 id integer primary key autoincrement,
175 knot text not null,
176 owner text not null,
177 rkey text not null,
178 repo_did text,
179 created_at text,
180 addedAt text not null default (strftime('%Y-%m-%dT%H:%M:%SZ', 'now')),
181
182 unique(owner, rkey)
183 );
184
185 insert into repos_new (id, knot, owner, rkey, addedAt)
186 select id, knot, owner, name, addedAt from repos where coalesce(name, '') <> '';
187
188 drop table repos;
189 alter table repos_new rename to repos;
190 `); err != nil {
191 return err
192 }
193 }
194
195 _, err := tx.Exec(`
196 create index if not exists idx_repos_repo_did on repos(repo_did);
197 create index if not exists idx_repos_owner_repo_did on repos(owner, repo_did);
198 create index if not exists idx_repo_collaborators_repo_did
199 on repo_collaborators(repo_did);
200 `)
201 return err
202 }); err != nil {
203 return err
204 }
205
206 if err := orm.RunMigration(conn, logger, "spindle-members-unique-on-rkey", func(tx *sql.Tx) error {
207 hasTarget, err := hasUniqueIndex(tx, "spindle_members", []string{"did", "rkey"})
208 if err != nil {
209 return err
210 }
211 if hasTarget {
212 return nil
213 }
214
215 var totalRows, distinctRows int
216 if err := tx.QueryRow(`select count(*) from spindle_members`).Scan(&totalRows); err != nil {
217 return err
218 }
219 if err := tx.QueryRow(`select count(*) from (select 1 from spindle_members group by did, rkey)`).Scan(&distinctRows); err != nil {
220 return err
221 }
222 if dropped := totalRows - distinctRows; dropped > 0 {
223 logger.Warn("dropping duplicate (did, rkey) rows during spindle_members rebuild", "dropped", dropped, "kept", distinctRows)
224 }
225
226 _, err = tx.Exec(`
227 create table spindle_members_new (
228 id integer primary key autoincrement,
229 did text not null,
230 rkey text not null,
231 instance text not null,
232 subject text not null,
233 created text not null default (strftime('%Y-%m-%dT%H:%M:%SZ', 'now')),
234 unique (did, rkey)
235 );
236
237 insert into spindle_members_new (id, did, rkey, instance, subject, created)
238 select id, did, rkey, instance, subject, created
239 from spindle_members sm
240 where id = (
241 select max(id) from spindle_members
242 where did = sm.did and rkey = sm.rkey
243 );
244
245 drop table spindle_members;
246 alter table spindle_members_new rename to spindle_members;
247 `)
248 return err
249 }); err != nil {
250 return err
251 }
252
253 if err := orm.RunMigration(conn, logger, "events-pipeline-index", func(tx *sql.Tx) error {
254 _, err := tx.Exec(`
255 create index if not exists idx_events_pipeline_lookup on events(
256 coalesce(json_extract(event, '$.triggerMetadata.repo.repoDid'),
257 json_extract(event, '$.triggerMetadata.repo.did')),
258 coalesce(json_extract(event, '$.triggerMetadata.push.newSha'),
259 json_extract(event, '$.triggerMetadata.pullRequest.sourceSha'),
260 json_extract(event, '$.triggerMetadata.manual.sha'))
261 ) where nsid = 'sh.tangled.pipeline';
262
263 create index if not exists idx_events_pipeline_status on events(
264 json_extract(event, '$.pipeline'),
265 json_extract(event, '$.workflow')
266 ) where nsid = 'sh.tangled.pipeline.status';
267 `)
268 return err
269 }); err != nil {
270 return err
271 }
272
273 if err := orm.RunMigration(conn, logger, "member-acl-table", func(tx *sql.Tx) error {
274 _, err := tx.Exec(`
275 CREATE TABLE members (
276 did TEXT PRIMARY KEY,
277 blocked INTEGER DEFAULT 0
278 );
279
280 insert or ignore into members (did) select subject from spindle_members;
281 insert or ignore into members (did) select did from spindle_members;
282
283 drop table spindle_members;
284 `)
285 return err
286 }); err != nil {
287 return err
288 }
289
290 // we use casbin instead
291 if err := orm.RunMigration(conn, logger, "drop-legacy-acl-tables", func(tx *sql.Tx) error {
292 _, err := tx.Exec(`
293 drop table if exists repo_collaborators;
294 drop table if exists known_dids;
295 `)
296 return err
297 }); err != nil {
298 return err
299 }
300
301 // repo_did is required
302 if err := orm.RunMigration(conn, logger, "enforce-repo_did", func(tx *sql.Tx) error {
303 _, err := tx.Exec(`
304 create table repos_new (
305 repo_did text primary key,
306 knot text not null,
307 owner text not null,
308 rkey text not null,
309 created_at text,
310 addedAt text not null default (strftime('%Y-%m-%dT%H:%M:%SZ', 'now')),
311
312 unique(owner, rkey)
313 );
314 insert into repos_new (repo_did, knot, owner, rkey, created_at, addedAt)
315 select repo_did, knot, owner, rkey, created_at, addedAt from repos
316 where coalesce(repo_did, '') <> ''
317 and rowid = (
318 select r2.rowid from repos r2
319 where r2.repo_did = repos.repo_did
320 order by r2.created_at is null, r2.created_at desc, r2.rowid desc
321 limit 1
322 );
323
324 drop table repos;
325 alter table repos_new rename to repos;
326 `)
327 return err
328 }); err != nil {
329 return err
330 }
331
332 return nil
333}
334
335func hasUniqueIndex(tx *sql.Tx, table string, cols []string) (bool, error) {
336 rows, err := tx.Query(
337 `select name from pragma_index_list(?) where "unique" = 1`,
338 table,
339 )
340 if err != nil {
341 return false, err
342 }
343 defer rows.Close()
344
345 var indexNames []string
346 for rows.Next() {
347 var name string
348 if err := rows.Scan(&name); err != nil {
349 return false, err
350 }
351 indexNames = append(indexNames, name)
352 }
353 if err := rows.Err(); err != nil {
354 return false, err
355 }
356
357 wantSorted := slices.Clone(cols)
358 slices.Sort(wantSorted)
359
360 for _, name := range indexNames {
361 colRows, err := tx.Query(
362 `select name from pragma_index_info(?) order by seqno`,
363 name,
364 )
365 if err != nil {
366 return false, err
367 }
368 var got []string
369 for colRows.Next() {
370 var c string
371 if err := colRows.Scan(&c); err != nil {
372 colRows.Close()
373 return false, err
374 }
375 got = append(got, c)
376 }
377 colRows.Close()
378 slices.Sort(got)
379 if slices.Equal(got, wantSorted) {
380 return true, nil
381 }
382 }
383 return false, nil
384}
385
386func (d *DB) SaveLastTimeUs(lastTimeUs int64) error {
387 _, err := d.Exec(`
388 insert into _jetstream (id, last_time_us)
389 values (1, ?)
390 on conflict(id) do update set last_time_us = excluded.last_time_us
391 `, lastTimeUs)
392 return err
393}
394
395func (d *DB) GetLastTimeUs() (int64, error) {
396 var lastTimeUs int64
397 row := d.QueryRow(`select last_time_us from _jetstream where id = 1;`)
398 err := row.Scan(&lastTimeUs)
399 return lastTimeUs, err
400}