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 if err := orm.RunMigration(conn, logger, "switch-to-rbac-v2", func(tx *sql.Tx) error {
333 // NOTE: we are manually creating casbin table because DB migration will happen before rbac/v2 initialization.
334 if _, err := tx.Exec(`
335 CREATE TABLE acl_2(
336 p_type VARCHAR(32) DEFAULT '' NOT NULL,
337 v0 VARCHAR(255) DEFAULT '' NOT NULL,
338 v1 VARCHAR(255) DEFAULT '' NOT NULL,
339 v2 VARCHAR(255) DEFAULT '' NOT NULL,
340 v3 VARCHAR(255) DEFAULT '' NOT NULL,
341 v4 VARCHAR(255) DEFAULT '' NOT NULL,
342 v5 VARCHAR(255) DEFAULT '' NOT NULL,
343 CHECK (TYPEOF("p_type") = "text" AND
344 LENGTH("p_type") <= 32),
345 CHECK (TYPEOF("v0") = "text" AND
346 LENGTH("v0") <= 255),
347 CHECK (TYPEOF("v1") = "text" AND
348 LENGTH("v1") <= 255),
349 CHECK (TYPEOF("v2") = "text" AND
350 LENGTH("v2") <= 255),
351 CHECK (TYPEOF("v3") = "text" AND
352 LENGTH("v3") <= 255),
353 CHECK (TYPEOF("v4") = "text" AND
354 LENGTH("v4") <= 255),
355 CHECK (TYPEOF("v5") = "text" AND
356 LENGTH("v5") <= 255)
357 );
358 `); err != nil {
359 return err
360 }
361
362 // fresh spindles might not have acl table
363 hasAcl, err := tableExists(tx, "acl")
364 if err != nil {
365 return err
366 }
367 if !hasAcl {
368 return nil
369 }
370
371 for _, role := range []string{"repo:owner", "repo:collaborator"} {
372 if _, err := tx.Exec(`
373 insert into acl_2 (p_type, v0, v1, v2)
374 select distinct 'g', v0, v3, v2
375 from acl
376 where p_type = 'p' and v1 = 'thisserver' and v3 = ?
377 `, role); err != nil {
378 return err
379 }
380 }
381 return nil
382 }); err != nil {
383 return err
384 }
385
386 return nil
387}
388
389func tableExists(tx *sql.Tx, name string) (bool, error) {
390 var exists bool
391 err := tx.QueryRow(
392 `select exists (select 1 from sqlite_master where type = 'table' and name = ?)`,
393 name,
394 ).Scan(&exists)
395 return exists, err
396}
397
398func hasUniqueIndex(tx *sql.Tx, table string, cols []string) (bool, error) {
399 rows, err := tx.Query(
400 `select name from pragma_index_list(?) where "unique" = 1`,
401 table,
402 )
403 if err != nil {
404 return false, err
405 }
406 defer rows.Close()
407
408 var indexNames []string
409 for rows.Next() {
410 var name string
411 if err := rows.Scan(&name); err != nil {
412 return false, err
413 }
414 indexNames = append(indexNames, name)
415 }
416 if err := rows.Err(); err != nil {
417 return false, err
418 }
419
420 wantSorted := slices.Clone(cols)
421 slices.Sort(wantSorted)
422
423 for _, name := range indexNames {
424 colRows, err := tx.Query(
425 `select name from pragma_index_info(?) order by seqno`,
426 name,
427 )
428 if err != nil {
429 return false, err
430 }
431 var got []string
432 for colRows.Next() {
433 var c string
434 if err := colRows.Scan(&c); err != nil {
435 colRows.Close()
436 return false, err
437 }
438 got = append(got, c)
439 }
440 colRows.Close()
441 slices.Sort(got)
442 if slices.Equal(got, wantSorted) {
443 return true, nil
444 }
445 }
446 return false, nil
447}
448
449func (d *DB) SaveLastTimeUs(lastTimeUs int64) error {
450 _, err := d.Exec(`
451 insert into _jetstream (id, last_time_us)
452 values (1, ?)
453 on conflict(id) do update set last_time_us = excluded.last_time_us
454 `, lastTimeUs)
455 return err
456}
457
458func (d *DB) GetLastTimeUs() (int64, error) {
459 var lastTimeUs int64
460 row := d.QueryRow(`select last_time_us from _jetstream where id = 1;`)
461 err := row.Scan(&lastTimeUs)
462 return lastTimeUs, err
463}