Something went wrong. Try again.
Monorepo for Tangled
Something went wrong. Try again.
Go
at sl/no-verify
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464package db
import ( "context" "database/sql" "log/slog" "slices" "strings"
_ "github.com/mattn/go-sqlite3" "tangled.org/core/log" "tangled.org/core/orm")
type DB struct { *sql.DB}
type DBTX interface { QueryRow(query string, args ...any) *sql.Row Exec(query string, args ...any) (sql.Result, error)}
func Make(ctx context.Context, dbPath string) (*DB, error) { // https://github.com/mattn/go-sqlite3#connection-string opts := []string{ "_foreign_keys=1", "_journal_mode=WAL", "_synchronous=NORMAL", "_auto_vacuum=incremental", "_busy_timeout=5000", }
logger := log.FromContext(ctx) logger = log.SubLogger(logger, "db")
db, err := sql.Open("sqlite3", dbPath+"?"+strings.Join(opts, "&")) if err != nil { return nil, err }
conn, err := db.Conn(ctx) if err != nil { return nil, err } defer conn.Close()
_, err = conn.ExecContext(ctx, ` create table if not exists _jetstream ( id integer primary key autoincrement, last_time_us integer not null );
create table if not exists known_dids ( did text primary key );
create table if not exists repos ( id integer primary key autoincrement, knot text not null, owner text not null, rkey text not null, repo_did text, created_at text, addedAt text not null default (strftime('%Y-%m-%dT%H:%M:%SZ', 'now')),
unique(owner, rkey) );
create table if not exists repo_collaborators ( id integer primary key autoincrement, owner_did text not null, rkey text not null, subject text not null, repo_did text not null, addedAt text not null default (strftime('%Y-%m-%dT%H:%M:%SZ', 'now')),
unique(owner_did, rkey) );
create table if not exists spindle_members ( -- identifiers for the record id integer primary key autoincrement, did text not null, rkey text not null,
-- data instance text not null, subject text not null, created text not null default (strftime('%Y-%m-%dT%H:%M:%SZ', 'now')),
-- constraints unique (did, rkey) );
-- status event for a single workflow create table if not exists events ( rkey text not null, nsid text not null, event text not null, -- json created integer not null -- unix nanos );
create table if not exists nixos_toplevel_cache ( config_key text primary key, toplevel text not null, updated_at text not null );
create table if not exists pipelines ( id text primary key, repo_did text not null, commit_id text not null ); create table if not exists jobs ( id integer primary key autoincrement, repo_did text not null, pipeline_id_knot text not null, pipeline_id_rkey text not null, source_repo text, tpl text not null, created_at integer not null default (strftime('%s', 'now')) );
create table if not exists workflows ( id integer primary key autoincrement, pipeline_id text not null, name text not null, status text not null default 'pending',
unique(pipeline_id, id), foreign key (pipeline_id) references pipelines(id) on delete cascade );
create table if not exists migrations ( id integer primary key autoincrement, name text unique ); `) if err != nil { return nil, err }
if err := runMigrations(ctx, conn, logger); err != nil { return nil, err }
return &DB{db}, nil}
func runMigrations(_ context.Context, conn *sql.Conn, logger *slog.Logger) error { if err := orm.RunMigration(conn, logger, "repos-to-repo-did", func(tx *sql.Tx) error { var hasName int if err := tx.QueryRow( `select count(*) from pragma_table_info('repos') where name = 'name'`, ).Scan(&hasName); err != nil { return err }
if hasName > 0 { var totalRows, copiedRows int if err := tx.QueryRow(`select count(*) from repos`).Scan(&totalRows); err != nil { return err } if err := tx.QueryRow(`select count(*) from repos where coalesce(name, '') <> ''`).Scan(&copiedRows); err != nil { return err } if dropped := totalRows - copiedRows; dropped > 0 { logger.Warn("dropping repo rows with empty name during migration", "dropped", dropped, "kept", copiedRows) }
if _, err := tx.Exec(` create table if not exists repos_new ( id integer primary key autoincrement, knot text not null, owner text not null, rkey text not null, repo_did text, created_at text, addedAt text not null default (strftime('%Y-%m-%dT%H:%M:%SZ', 'now')),
unique(owner, rkey) );
insert into repos_new (id, knot, owner, rkey, addedAt) select id, knot, owner, name, addedAt from repos where coalesce(name, '') <> '';
drop table repos; alter table repos_new rename to repos; `); err != nil { return err } }
_, err := tx.Exec(` create index if not exists idx_repos_repo_did on repos(repo_did); create index if not exists idx_repos_owner_repo_did on repos(owner, repo_did); create index if not exists idx_repo_collaborators_repo_did on repo_collaborators(repo_did); `) return err }); err != nil { return err }
if err := orm.RunMigration(conn, logger, "spindle-members-unique-on-rkey", func(tx *sql.Tx) error { hasTarget, err := hasUniqueIndex(tx, "spindle_members", []string{"did", "rkey"}) if err != nil { return err } if hasTarget { return nil }
var totalRows, distinctRows int if err := tx.QueryRow(`select count(*) from spindle_members`).Scan(&totalRows); err != nil { return err } if err := tx.QueryRow(`select count(*) from (select 1 from spindle_members group by did, rkey)`).Scan(&distinctRows); err != nil { return err } if dropped := totalRows - distinctRows; dropped > 0 { logger.Warn("dropping duplicate (did, rkey) rows during spindle_members rebuild", "dropped", dropped, "kept", distinctRows) }
_, err = tx.Exec(` create table spindle_members_new ( id integer primary key autoincrement, did text not null, rkey text not null, instance text not null, subject text not null, created text not null default (strftime('%Y-%m-%dT%H:%M:%SZ', 'now')), unique (did, rkey) );
insert into spindle_members_new (id, did, rkey, instance, subject, created) select id, did, rkey, instance, subject, created from spindle_members sm where id = ( select max(id) from spindle_members where did = sm.did and rkey = sm.rkey );
drop table spindle_members; alter table spindle_members_new rename to spindle_members; `) return err }); err != nil { return err }
if err := orm.RunMigration(conn, logger, "events-pipeline-index", func(tx *sql.Tx) error { _, err := tx.Exec(` create index if not exists idx_events_pipeline_lookup on events( coalesce(json_extract(event, '$.triggerMetadata.repo.repoDid'), json_extract(event, '$.triggerMetadata.repo.did')), coalesce(json_extract(event, '$.triggerMetadata.push.newSha'), json_extract(event, '$.triggerMetadata.pullRequest.sourceSha'), json_extract(event, '$.triggerMetadata.manual.sha')) ) where nsid = 'sh.tangled.pipeline';
create index if not exists idx_events_pipeline_status on events( json_extract(event, '$.pipeline'), json_extract(event, '$.workflow') ) where nsid = 'sh.tangled.pipeline.status'; `) return err }); err != nil { return err }
if err := orm.RunMigration(conn, logger, "member-acl-table", func(tx *sql.Tx) error { _, err := tx.Exec(` CREATE TABLE members ( did TEXT PRIMARY KEY, blocked INTEGER DEFAULT 0 );
insert or ignore into members (did) select subject from spindle_members; insert or ignore into members (did) select did from spindle_members;
drop table spindle_members; `) return err }); err != nil { return err }
// we use casbin instead if err := orm.RunMigration(conn, logger, "drop-legacy-acl-tables", func(tx *sql.Tx) error { _, err := tx.Exec(` drop table if exists repo_collaborators; drop table if exists known_dids; `) return err }); err != nil { return err }
// repo_did is required if err := orm.RunMigration(conn, logger, "enforce-repo_did", func(tx *sql.Tx) error { _, err := tx.Exec(` create table repos_new ( repo_did text primary key, knot text not null, owner text not null, rkey text not null, created_at text, addedAt text not null default (strftime('%Y-%m-%dT%H:%M:%SZ', 'now')),
unique(owner, rkey) ); insert into repos_new (repo_did, knot, owner, rkey, created_at, addedAt) select repo_did, knot, owner, rkey, created_at, addedAt from repos where coalesce(repo_did, '') <> '' and rowid = ( select r2.rowid from repos r2 where r2.repo_did = repos.repo_did order by r2.created_at is null, r2.created_at desc, r2.rowid desc limit 1 );
drop table repos; alter table repos_new rename to repos; `) return err }); err != nil { return err }
if err := orm.RunMigration(conn, logger, "switch-to-rbac-v2", func(tx *sql.Tx) error { // NOTE: we are manually creating casbin table because DB migration will happen before rbac/v2 initialization. if _, err := tx.Exec(` CREATE TABLE acl_2( p_type VARCHAR(32) DEFAULT '' NOT NULL, v0 VARCHAR(255) DEFAULT '' NOT NULL, v1 VARCHAR(255) DEFAULT '' NOT NULL, v2 VARCHAR(255) DEFAULT '' NOT NULL, v3 VARCHAR(255) DEFAULT '' NOT NULL, v4 VARCHAR(255) DEFAULT '' NOT NULL, v5 VARCHAR(255) DEFAULT '' NOT NULL, CHECK (TYPEOF("p_type") = "text" AND LENGTH("p_type") <= 32), CHECK (TYPEOF("v0") = "text" AND LENGTH("v0") <= 255), CHECK (TYPEOF("v1") = "text" AND LENGTH("v1") <= 255), CHECK (TYPEOF("v2") = "text" AND LENGTH("v2") <= 255), CHECK (TYPEOF("v3") = "text" AND LENGTH("v3") <= 255), CHECK (TYPEOF("v4") = "text" AND LENGTH("v4") <= 255), CHECK (TYPEOF("v5") = "text" AND LENGTH("v5") <= 255) ); `); err != nil { return err }
// fresh spindles might not have acl table hasAcl, err := tableExists(tx, "acl") if err != nil { return err } if !hasAcl { return nil }
for _, role := range []string{"repo:owner", "repo:collaborator"} { if _, err := tx.Exec(` insert into acl_2 (p_type, v0, v1, v2) select distinct 'g', v0, v3, v2 from acl where p_type = 'p' and v1 = 'thisserver' and v3 = ? `, role); err != nil { return err } } return nil }); err != nil { return err }
return nil}
func tableExists(tx *sql.Tx, name string) (bool, error) { var exists bool err := tx.QueryRow( `select exists (select 1 from sqlite_master where type = 'table' and name = ?)`, name, ).Scan(&exists) return exists, err}
func hasUniqueIndex(tx *sql.Tx, table string, cols []string) (bool, error) { rows, err := tx.Query( `select name from pragma_index_list(?) where "unique" = 1`, table, ) if err != nil { return false, err } defer rows.Close()
var indexNames []string for rows.Next() { var name string if err := rows.Scan(&name); err != nil { return false, err } indexNames = append(indexNames, name) } if err := rows.Err(); err != nil { return false, err }
wantSorted := slices.Clone(cols) slices.Sort(wantSorted)
for _, name := range indexNames { colRows, err := tx.Query( `select name from pragma_index_info(?) order by seqno`, name, ) if err != nil { return false, err } var got []string for colRows.Next() { var c string if err := colRows.Scan(&c); err != nil { colRows.Close() return false, err } got = append(got, c) } colRows.Close() slices.Sort(got) if slices.Equal(got, wantSorted) { return true, nil } } return false, nil}
func (d *DB) SaveLastTimeUs(lastTimeUs int64) error { _, err := d.Exec(` insert into _jetstream (id, last_time_us) values (1, ?) on conflict(id) do update set last_time_us = excluded.last_time_us `, lastTimeUs) return err}
func (d *DB) GetLastTimeUs() (int64, error) { var lastTimeUs int64 row := d.QueryRow(`select last_time_us from _jetstream where id = 1;`) err := row.Scan(&lastTimeUs) return lastTimeUs, err}