package db
import (
"context"
"database/sql"
"encoding/json"
"log/slog"
"strings"
"time"
"github.com/bluesky-social/indigo/atproto/syntax"
_ "github.com/mattn/go-sqlite3"
"tangled.org/core/log"
"tangled.org/core/orm"
"tangled.org/core/spindle/models"
pipelinecodec "tangled.org/core/spindle/pipeline"
"tangled.org/core/sqlite"
"tangled.org/core/workflow"
)
type DB struct {
*sql.DB
}
type DBTX interface {
Query(query string, args ...any) (*sql.Rows, error)
QueryRow(query string, args ...any) *sql.Row
Exec(query string, args ...any) (sql.Result, error)
}
func queryDIDs(q DBTX, query string, args ...any) ([]syntax.DID, error) {
rows, err := q.Query(query, args...)
if err != nil {
return nil, err
}
defer rows.Close()
var out []syntax.DID
for rows.Next() {
var did syntax.DID
if err := rows.Scan(&did); err != nil {
return nil, err
}
out = append(out, did)
}
return out, rows.Err()
}
const maxIdleConns = 64
func Make(ctx context.Context, dbPath string) (*DB, error) {
logger := log.FromContext(ctx)
logger = log.SubLogger(logger, "db")
db, err := sqlite.Open(dbPath)
if err != nil {
return nil, err
}
// cancelled viewer queries would churn fds unless idle conns stick around
db.SetMaxIdleConns(maxIdleConns)
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 instance_allowlist (
did text primary key,
created_at text not null default (strftime('%Y-%m-%dT%H:%M:%SZ', 'now'))
);
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 nixos_toplevel_cache (
config_key text primary key,
toplevel text not null,
updated_at text not null
);
create table if not exists cache_entries (
id text primary key,
storage_key text unique not null,
owner_did text not null,
repo_did text not null,
engine text not null,
cache_key text not null,
cache_hash text not null,
checksum text not null default '',
size_bytes integer not null default 0,
restore_count integer not null default 0 check (restore_count >= 0),
state text not null check (state in ('pending', 'ready', 'deleting')),
created_at integer not null,
last_used_at integer not null
);
create index if not exists cache_entries_lookup
on cache_entries (repo_did, engine, cache_key, cache_hash, created_at desc)
where state = 'ready';
create index if not exists cache_entries_ready_expiry
on cache_entries (last_used_at) where state = 'ready';
create index if not exists cache_entries_pending_deadline
on cache_entries (last_used_at) where state in ('pending', 'deleting');
create index if not exists cache_entries_owner_state
on cache_entries (owner_did, state);
create table if not exists cache_object_deletions (
storage_key text primary key,
created_at integer 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,
traceparent text not null default '',
tracestate text not null default '',
created_at integer not null default (strftime('%s', 'now')),
created_at_ns integer not null default 0
);
create table if not exists schedule_runs (
repo_did text not null,
workflow text not null,
scheduled_at integer not null,
pipeline_id text,
primary key (repo_did, workflow, scheduled_at)
);
create table if not exists scheduled_repos (
repo_did text primary key,
branch text not null default '',
sha text not null default '',
refreshed_at integer not null
);
create table if not exists workflow_schedules (
repo_did text not null,
workflow text not null,
expression text not null,
timezone text not null default 'UTC',
primary key (repo_did, workflow, expression, timezone),
foreign key (repo_did) references scheduled_repos(repo_did) on delete cascade
);
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 mill_executor_tokens (
token_hash text primary key,
created_at text not null default (strftime('%Y-%m-%dT%H:%M:%SZ', 'now')),
expires_at text
);
create table if not exists mill_executors (
name text primary key,
token_hash text not null unique,
labels text,
quarantine_reason text,
quarantined_at text,
foreign key (token_hash) references mill_executor_tokens(token_hash) on delete cascade
);
create table if not exists mill_leases (
lease_id text primary key,
node_id text not null,
epoch text not null,
engine text not null,
knot text not null,
rkey text not null,
workflow text not null,
state text not null,
quota_reservation_id text,
owner_did text,
repo_did text,
mill_records_terminal_metrics integer not null default 0
);
create table if not exists mill_cache_capabilities (
lease_id text not null,
action text not null check (action in ('restore', 'save')),
cache_id text not null,
storage_key text not null,
primary key (lease_id, action, cache_id),
foreign key (lease_id) references mill_leases(lease_id) on delete cascade
);
create table if not exists mill_executor_cursors (
node_id text not null,
epoch text not null,
acked_seqno integer not null,
primary key (node_id, epoch)
);
create table if not exists mill_outbox_state (
epoch text not null,
next_seqno integer not null,
primary key (epoch)
);
create table if not exists mill_outbox_rows (
epoch text not null,
seqno integer not null,
payload blob not null,
byte_size integer not null,
control integer not null,
primary key (epoch, seqno),
foreign key (epoch) references mill_outbox_state(epoch) on delete cascade
);
create table if not exists mill_artifacts (
id integer primary key autoincrement,
lease_id text not null,
repo_did text not null,
knot text not null,
rkey text not null,
workflow text not null,
ref text not null,
hash text not null
);
create table if not exists executor_pending_artifacts (
lease_id text primary key,
knot text not null default '',
rkey text not null default '',
workflow text not null,
status text not null,
error text not null default '',
exit_code integer not null default 0,
ref text not null,
hash text not null,
failure_class text not null default '',
failure_reason text not null default '',
mill_records_terminal_metrics integer not null default 0
);
create table if not exists webhooks (
id integer primary key autoincrement,
repo_did text not null,
url text not null,
secret text,
active integer not null default 1,
events text not null, -- comma-separated event types
created_at text not null default (strftime('%Y-%m-%dT%H:%M:%SZ', 'now')),
updated_at text not null default (strftime('%Y-%m-%dT%H:%M:%SZ', 'now'))
);
create index if not exists idx_webhooks_repo_did on webhooks(repo_did);
create table if not exists webhook_deliveries (
id integer primary key autoincrement,
webhook_id integer not null references webhooks(id) on delete cascade,
event text not null,
delivery_id text not null,
url text not null,
request_body text,
response_code integer,
response_body text,
success integer not null default 0,
created_at text not null default (strftime('%Y-%m-%dT%H:%M:%SZ', 'now'))
);
create index if not exists idx_webhook_deliveries_webhook_id on webhook_deliveries(webhook_id);
create unique index if not exists idx_webhook_deliveries_delivery_id on webhook_deliveries(delivery_id);
-- what a pull's transitions are derived from, the stream only says that something changed
create table if not exists pull_tickets (
repo_did text not null,
rkey text not null,
author text not null,
state text not null,
primary key (repo_did, rkey)
);
create table if not exists pull_patches (
repo_did text not null,
rkey text not null,
ticket text not null,
head text not null,
created_at text not null,
primary key (repo_did, rkey)
);
create index if not exists idx_pull_patches_ticket on pull_patches(repo_did, ticket);
-- backfilled records that kept failing, retried on their own so the
-- backfill stream doesn't wait on them
create table if not exists backfill_parked (
uri text primary key,
rev text not null,
event blob not null,
attempts integer not null default 0,
retry_at integer not null,
error text not null default ''
);
create index if not exists idx_backfill_parked_retry_at on backfill_parked(retry_at);
-- repos whose refresh failed after a live event, so a restart still retries them
create table if not exists repo_retries (
repo_did text primary key,
attempts integer not null,
due integer not null
);
-- dids an interrupted or partly failed catch-up still has to backfill
create table if not exists backfill_pending (
did text primary key,
since integer not null
);
create table if not exists migrations (
id integer primary key autoincrement,
name text unique
);
`)
if err := runMigrations(ctx, conn, logger); err != nil {
return nil, err
}
return &DB{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, "events-pipeline-index", func(tx *sql.Tx) error {
_, err := tx.Exec(`
create table if not exists events (
rkey text not null,
nsid text not null,
event text not null,
created integer not null
);
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, "mill-executor-credentials", func(tx *sql.Tx) error {
var legacySchema int
if err := tx.QueryRow(
`select count(*) from pragma_table_info('mill_executors') where name = 'expires_at'`,
).Scan(&legacySchema); err != nil {
return err
}
if legacySchema == 0 {
return nil
}
_, err := tx.Exec(`
insert into mill_executor_tokens (token_hash, created_at, expires_at)
select token_hash, created_at, expires_at from mill_executors;
create table mill_executors_new (
name text primary key,
token_hash text not null unique,
labels text,
quarantine_reason text,
quarantined_at text,
foreign key (token_hash) references mill_executor_tokens(token_hash) on delete cascade
);
insert into mill_executors_new (
name, token_hash, labels, quarantine_reason, quarantined_at
)
select name, token_hash, labels, quarantine_reason, quarantined_at
from mill_executors;
drop table mill_executors;
alter table mill_executors_new rename to mill_executors;
`)
return err
}); err != nil {
return err
}
// older generations select these fields during authentication, so clear rather than drop them
if err := orm.RunMigration(conn, logger, "mill-clear-executor-quarantine", func(tx *sql.Tx) error {
_, err := tx.Exec(`update mill_executors set quarantine_reason = null, quarantined_at = null`)
return err
}); err != nil {
return err
}
if err := orm.RunMigration(conn, logger, "jobs-trace-context", func(tx *sql.Tx) error {
for _, column := range []string{"traceparent", "tracestate"} {
var present int
if err := tx.QueryRow(
`select count(*) from pragma_table_info('jobs') where name = ?`,
column,
).Scan(&present); err != nil {
return err
}
if present != 0 {
continue
}
if _, err := tx.Exec(`alter table jobs add column ` + column + ` text not null default ''`); err != nil {
return err
}
}
return nil
}); err != nil {
return err
}
if err := orm.RunMigration(conn, logger, "cache-entry-integrity", func(tx *sql.Tx) error {
for _, column := range []struct{ name, definition string }{
{"checksum", "text not null default ''"},
{"restore_count", "integer not null default 0"},
} {
var present int
if err := tx.QueryRow(`select count(*) from pragma_table_info('cache_entries') where name = ?`, column.name).Scan(&present); err != nil {
return err
}
if present == 0 {
if _, err := tx.Exec(`alter table cache_entries add column ` + column.name + ` ` + column.definition); err != nil {
return err
}
}
}
return nil
}); err != nil {
return err
}
if err := orm.RunMigration(conn, logger, "cache-quotas-schema", func(tx *sql.Tx) error {
_, err := tx.Exec(`
create table if not exists cache_objects (
repo_did text not null,
object_key text not null,
byte_size integer not null check (byte_size >= 0),
primary key (repo_did, object_key)
);
create index if not exists idx_cache_objects_key on cache_objects(object_key);
create table if not exists cache_repo_owners (
repo_did text primary key,
owner_did text not null
);
create index if not exists idx_cache_repo_owners_owner on cache_repo_owners(owner_did);
create table if not exists cache_reservations (
id text primary key,
object_key text not null,
byte_size integer not null check (byte_size >= 0),
repo_did text not null,
owner_did text not null,
phase text not null check (phase in ('reserved', 'publishing')),
created_at integer not null
);
create index if not exists idx_cache_reservations_repo on cache_reservations(repo_did);
create index if not exists idx_cache_reservations_owner on cache_reservations(owner_did);
create table if not exists cache_overrides (
scope text not null check (scope in ('user', 'repo')),
did text not null,
max_bytes integer check (max_bytes >= 0 or max_bytes is null),
primary key (scope, did)
);
create index if not exists idx_cache_overrides_scope on cache_overrides(scope);
`)
return err
}); err != nil {
return err
}
if err := orm.RunMigration(conn, logger, "quota-schema", func(tx *sql.Tx) error {
_, err := tx.Exec(`
create table if not exists quota_limits (
did text not null,
resource text not null,
max_amount integer check (max_amount >= 0 or max_amount is null),
primary key (did, resource)
);
create table if not exists quota_reservations (
id text not null,
resource text not null,
kind text not null,
key text not null,
amount integer not null check (amount >= 0),
repo_did text not null,
owner_did text not null,
phase text not null check (phase in ('reserved', 'publishing', 'active')),
created_at integer not null,
primary key (id, resource)
);
create index if not exists idx_quota_reservations_repo on quota_reservations(repo_did);
create index if not exists idx_quota_reservations_owner on quota_reservations(owner_did);
create table if not exists quota_allocations (
repo_did text not null,
resource text not null,
kind text not null,
key text not null,
amount integer not null check (amount >= 0),
primary key (repo_did, resource, kind, key)
);
create index if not exists idx_quota_allocations_key on quota_allocations(key);
create table if not exists quota_repo_owners (
repo_did text primary key,
owner_did text not null
);
create index if not exists idx_quota_repo_owners_owner on quota_repo_owners(owner_did);
`)
if err != nil {
return err
}
var hasOverrides int
err = tx.QueryRow(`select count(*) from sqlite_master where type='table' and name='cache_overrides'`).Scan(&hasOverrides)
if err != nil {
return err
}
if hasOverrides > 0 {
_, err = tx.Exec(`
insert or ignore into quota_limits (did, resource, max_amount)
select did, 'cache_storage_bytes', max_bytes from cache_overrides
`)
if err != nil {
return err
}
}
var hasReservations int
err = tx.QueryRow(`select count(*) from sqlite_master where type='table' and name='cache_reservations'`).Scan(&hasReservations)
if err != nil {
return err
}
if hasReservations > 0 {
_, err = tx.Exec(`
insert or ignore into quota_reservations (id, resource, kind, key, amount, repo_did, owner_did, phase, created_at)
select id, 'cache_storage_bytes', 'nix_cache', object_key, byte_size, repo_did, owner_did, phase, created_at from cache_reservations
`)
if err != nil {
return err
}
}
var hasObjects int
err = tx.QueryRow(`select count(*) from sqlite_master where type='table' and name='cache_objects'`).Scan(&hasObjects)
if err != nil {
return err
}
if hasObjects > 0 {
_, err = tx.Exec(`
insert or ignore into quota_allocations (repo_did, resource, kind, key, amount)
select repo_did, 'cache_storage_bytes', 'nix_cache', object_key, byte_size from cache_objects
`)
if err != nil {
return err
}
}
var hasRepoOwners int
err = tx.QueryRow(`select count(*) from sqlite_master where type='table' and name='cache_repo_owners'`).Scan(&hasRepoOwners)
if err != nil {
return err
}
if hasRepoOwners > 0 {
_, err = tx.Exec(`
insert or ignore into quota_repo_owners (repo_did, owner_did)
select repo_did, owner_did from cache_repo_owners
`)
if err != nil {
return err
}
}
_, err = tx.Exec(`
drop index if exists idx_cache_objects_key;
drop index if exists idx_cache_repo_owners_owner;
drop index if exists idx_cache_reservations_repo;
drop index if exists idx_cache_reservations_owner;
drop index if exists idx_cache_overrides_scope;
drop table if exists cache_objects;
drop table if exists cache_repo_owners;
drop table if exists cache_reservations;
drop table if exists cache_overrides;
`)
if err != nil {
return err
}
return nil
}); err != nil {
return err
}
if err := orm.RunMigration(conn, logger, "quota-zero-means-unlimited", func(tx *sql.Tx) error {
_, err := tx.Exec(`update quota_limits set max_amount = null where max_amount = 0`)
return err
}); err != nil {
return err
}
// persist charged subjects so restarts can distinguish live reservations
if err := orm.RunMigration(conn, logger, "mill-leases-quota-column", func(tx *sql.Tx) error {
for _, column := range []string{"quota_reservation_id", "owner_did", "repo_did"} {
var present int
if err := tx.QueryRow(
`select count(*) from pragma_table_info('mill_leases') where name = ?`,
column,
).Scan(&present); err != nil {
return err
}
if present != 0 {
continue
}
if _, err := tx.Exec(`alter table mill_leases add column ` + column + ` text`); err != nil {
return err
}
}
return nil
}); err != nil {
return err
}
if err := orm.RunMigration(conn, logger, "mill-artifacts-repo-did", func(tx *sql.Tx) error {
var present int
if err := tx.QueryRow(
`select count(*) from pragma_table_info('mill_artifacts') where name = 'repo_did'`,
).Scan(&present); err != nil {
return err
}
if present == 0 {
if _, err := tx.Exec(
`alter table mill_artifacts add column repo_did text not null default ''`,
); err != nil {
return err
}
}
_, err := tx.Exec(`
update mill_artifacts
set repo_did = coalesce((
select repo_did from mill_leases
where mill_leases.lease_id = mill_artifacts.lease_id
), repo_did, '')
where coalesce(repo_did, '') = '';
`)
return err
}); err != nil {
return err
}
if err := orm.RunMigration(conn, logger, "mill-artifacts-workflow-identity", func(tx *sql.Tx) error {
for _, column := range []string{"knot", "rkey"} {
var present int
if err := tx.QueryRow(
`select count(*) from pragma_table_info('mill_artifacts') where name = ?`,
column,
).Scan(&present); err != nil {
return err
}
if present != 0 {
continue
}
if _, err := tx.Exec(
`alter table mill_artifacts add column ` + column + ` text not null default ''`,
); err != nil {
return err
}
}
_, err := tx.Exec(`
create index if not exists idx_mill_artifacts_workflow_identity
on mill_artifacts (knot, rkey, workflow, id desc)
`)
return err
}); err != nil {
return err
}
if err := orm.RunMigration(conn, logger, "bans-schema", func(tx *sql.Tx) error {
_, err := tx.Exec(`
create table if not exists bans (
subject_did text primary key,
created_at text not null default (strftime('%Y-%m-%dT%H:%M:%SZ', 'now'))
);
`)
return err
}); err != nil {
return err
}
if err := orm.RunMigration(conn, logger, "jobs-created-at", func(tx *sql.Tx) error {
var present int
if err := tx.QueryRow(
`select count(*) from pragma_table_info('jobs') where name = 'created_at'`,
).Scan(&present); err != nil {
return err
}
if present != 0 {
return nil
}
_, err := tx.Exec(`alter table jobs add column created_at integer not null default 0`)
return err
}); err != nil {
return err
}
if err := orm.RunMigration(conn, logger, "jobs-created-at-ns", func(tx *sql.Tx) error {
var present int
if err := tx.QueryRow(
`select count(*) from pragma_table_info('jobs') where name = 'created_at_ns'`,
).Scan(&present); err != nil {
return err
}
if present == 0 {
if _, err := tx.Exec(`alter table jobs add column created_at_ns integer not null default 0`); err != nil {
return err
}
}
_, err := tx.Exec(`
update jobs
set created_at_ns = created_at * 1000000000
where created_at_ns = 0 and created_at between 1 and 9999999999
`)
return err
}); err != nil {
return err
}
if err := orm.RunMigration(conn, logger, "pending-artifact-failure-attribution", func(tx *sql.Tx) error {
for _, column := range []string{"failure_class", "failure_reason"} {
var present int
if err := tx.QueryRow(
`select count(*) from pragma_table_info('executor_pending_artifacts') where name = ?`,
column,
).Scan(&present); err != nil {
return err
}
if present != 0 {
continue
}
if _, err := tx.Exec(
`alter table executor_pending_artifacts add column ` + column + ` text not null default ''`,
); err != nil {
return err
}
}
return nil
}); err != nil {
return err
}
if err := orm.RunMigration(conn, logger, "pending-artifact-workflow-identity", func(tx *sql.Tx) error {
for _, column := range []string{"knot", "rkey"} {
var present int
if err := tx.QueryRow(
`select count(*) from pragma_table_info('executor_pending_artifacts') where name = ?`,
column,
).Scan(&present); err != nil {
return err
}
if present != 0 {
continue
}
if _, err := tx.Exec(
`alter table executor_pending_artifacts add column ` + column + ` text not null default ''`,
); err != nil {
return err
}
}
return nil
}); err != nil {
return err
}
if err := orm.RunMigration(conn, logger, "pending-artifact-terminal-metric-authority", func(tx *sql.Tx) error {
var present int
if err := tx.QueryRow(
`select count(*) from pragma_table_info('executor_pending_artifacts')
where name = 'mill_records_terminal_metrics'`,
).Scan(&present); err != nil {
return err
}
if present != 0 {
return nil
}
_, err := tx.Exec(`
alter table executor_pending_artifacts
add column mill_records_terminal_metrics integer not null default 0
`)
return err
}); err != nil {
return err
}
if err := orm.RunMigration(conn, logger, "mill-lease-terminal-metric-authority", func(tx *sql.Tx) error {
var present int
if err := tx.QueryRow(
`select count(*) from pragma_table_info('mill_leases')
where name = 'mill_records_terminal_metrics'`,
).Scan(&present); err != nil {
return err
}
if present != 0 {
return nil
}
_, err := tx.Exec(`
alter table mill_leases
add column mill_records_terminal_metrics integer not null default 0
`)
return err
}); err != nil {
return err
}
// re-introduce a repo display name, needed to build repository:renamed
// webhook payloads. the earlier repos-to-repo-did migration dropped the
// legacy name column; this adds it back as nullable metadata.
if err := orm.RunMigration(conn, logger, "repos-add-name-column", 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 {
if _, err := tx.Exec(`alter table repos add column name text`); err != nil {
return err
}
}
_, err := tx.Exec(`update repos set name = rkey where name is null`)
return err
}); err != nil {
return err
}
if err := orm.RunMigration(conn, logger, "pipelines-and-workflow-statuses", func(tx *sql.Tx) error {
if _, err := tx.Exec(`
drop table if exists workflows;
drop table if exists pipelines;
`); err != nil {
return err
}
if _, err := tx.Exec(`
create table pipelines (
id integer primary key autoincrement,
rkey text not null unique,
knot text not null,
repo_did text not null,
commit_sha text not null,
kind text not null,
payload text not null
);
create index idx_pipelines_repo_id on pipelines(repo_did, id);
create index idx_pipelines_repo_commit on pipelines(repo_did, commit_sha);
create table workflow_statuses (
id integer primary key autoincrement,
rkey text not null,
workflow text not null,
status text not null,
error text,
exit_code integer,
created_at text not null
);
create index idx_workflow_statuses_lookup on workflow_statuses(rkey, workflow, id);
`); err != nil {
return err
}
if err := migratePipelines(tx, logger); err != nil {
return err
}
if err := migrateWorkflowStatuses(tx, logger); err != nil {
return err
}
_, err := tx.Exec(`
drop index if exists idx_events_pipeline_lookup;
drop index if exists idx_events_pipeline_status;
alter table events rename to events_legacy;
`)
return err
}); err != nil {
return err
}
if err := orm.RunMigration(conn, logger, "pipeline-id-identity", func(tx *sql.Tx) error {
if _, err := tx.Exec(`
create table pipeline_log_renames (
knot text not null,
pipeline_id text not null,
workflow text not null,
primary key (knot, pipeline_id, workflow)
);
`); err != nil {
return err
}
if err := stagePipelineLogRenames(tx); err != nil {
return err
}
if err := migratePipelineIdentityColumns(tx); err != nil {
return err
}
if _, err := tx.Exec(`
alter table jobs drop column pipeline_id_knot;
alter table jobs rename column pipeline_id_rkey to pipeline_id;
alter table mill_leases drop column knot;
alter table mill_leases rename column rkey to pipeline_id;
drop index if exists idx_mill_artifacts_workflow_identity;
alter table mill_artifacts drop column knot;
alter table mill_artifacts rename column rkey to pipeline_id;
create index idx_mill_artifacts_workflow_identity
on mill_artifacts (pipeline_id, workflow, id desc);
alter table executor_pending_artifacts drop column knot;
alter table executor_pending_artifacts rename column rkey to pipeline_id;
`); err != nil {
return err
}
return nil
}); err != nil {
return err
}
if err := orm.RunMigration(conn, logger, "drop-legacy-pipeline-events", func(tx *sql.Tx) error {
_, err := tx.Exec(`
drop table if exists events;
drop table if exists events_legacy;
`)
return err
}); err != nil {
return err
}
if err := orm.RunMigration(conn, logger, "schedule-runs-prune-index", func(tx *sql.Tx) error {
_, err := tx.Exec(`create index if not exists idx_schedule_runs_prune on schedule_runs(scheduled_at)`)
return err
}); err != nil {
return err
}
if err := orm.RunMigration(conn, logger, "workflow-schedules-timezone", func(tx *sql.Tx) error {
var hasTimezone int
if err := tx.QueryRow(
`select count(*) from pragma_table_info('workflow_schedules') where name = 'timezone'`,
).Scan(&hasTimezone); err != nil {
return err
}
if hasTimezone > 0 {
return nil
}
_, err := tx.Exec(`
create table workflow_schedules_new (
repo_did text not null,
workflow text not null,
expression text not null,
timezone text not null default 'UTC',
primary key (repo_did, workflow, expression, timezone),
foreign key (repo_did) references scheduled_repos(repo_did) on delete cascade
);
insert into workflow_schedules_new (repo_did, workflow, expression, timezone)
select repo_did, workflow, expression, 'UTC' from workflow_schedules;
drop table workflow_schedules;
alter table workflow_schedules_new rename to workflow_schedules;
`)
return err
}); err != nil {
return err
}
if err := orm.RunMigration(conn, logger, "scheduled-repos-revision", func(tx *sql.Tx) error {
for _, column := range []string{"branch", "sha"} {
var present int
if err := tx.QueryRow(
`select count(*) from pragma_table_info('scheduled_repos') where name = ?`,
column,
).Scan(&present); err != nil {
return err
}
if present != 0 {
continue
}
if _, err := tx.Exec(
`alter table scheduled_repos add column ` + column + ` text not null default ''`,
); err != nil {
return err
}
}
return nil
}); err != nil {
return err
}
if err := orm.RunMigration(conn, logger, "drop-spindle-members-and-known-dids", func(tx *sql.Tx) error {
_, err := tx.Exec(`
drop table if exists spindle_members;
drop table if exists known_dids;
`)
return err
}); err != nil {
return err
}
if err := orm.RunMigration(conn, logger, "drop-unused-feed-tables", func(tx *sql.Tx) error {
_, err := tx.Exec(`
drop table if exists feed_cursors;
drop table if exists feed_refs;
drop table if exists record_revs;
`)
return err
}); err != nil {
return err
}
if err := orm.RunMigration(conn, logger, "pipelines-push-dedupe", func(tx *sql.Tx) error {
for _, column := range []string{"ref", "old_sha"} {
present, err := tableHasColumn(tx, "pipelines", column)
if err != nil {
return err
}
if present {
continue
}
if _, err := tx.Exec(`alter table pipelines add column ` + column + ` text`); err != nil {
return err
}
}
_, err := tx.Exec(`
create index if not exists idx_pipelines_push_dedupe
on pipelines(repo_did, commit_sha, ref, old_sha) where kind = 'push'
`)
return err
}); err != nil {
return err
}
if err := orm.RunMigration(conn, logger, "drop-pull-rounds", func(tx *sql.Tx) error {
_, err := tx.Exec(`drop table if exists pull_rounds`)
return err
}); err != nil {
return err
}
if err := orm.RunMigration(conn, logger, "pull-events-and-patch-sources", func(tx *sql.Tx) error {
_, err := tx.Exec(`
-- a derived pull request event until its webhook and pipeline ran, pipeline_id is
-- minted up front so a retry can tell its pipeline already exists
create table if not exists pull_events (
id integer primary key autoincrement,
repo_did text not null,
ticket text not null,
event text not null,
author text not null,
sender text not null,
pull blob not null,
rounds integer not null,
head text not null,
source_repo text not null,
source_branch text not null,
pipeline_id text not null,
webhook_fired integer not null default 0,
attempts integer not null default 0,
due integer not null
);
create index if not exists idx_pull_events_due on pull_events(due);
-- live pull request records whose handler failed before it stored anything
create table if not exists record_retries (
repo_did text not null,
collection text not null,
rkey text not null,
attempts integer not null,
due integer not null,
primary key (repo_did, collection, rkey)
);
`)
if err != nil {
return err
}
for _, column := range []string{"source_repo", "source_branch"} {
present, err := tableHasColumn(tx, "pull_patches", column)
if err != nil {
return err
}
if present {
continue
}
if _, err := tx.Exec(`alter table pull_patches add column ` + column + ` text not null default ''`); err != nil {
return err
}
}
return nil
}); err != nil {
return err
}
if err := orm.RunMigration(conn, logger, "repo-retries-epoch", func(tx *sql.Tx) error {
return addColumnIfMissing(tx, "repo_retries", "epoch", "integer not null default 0")
}); err != nil {
return err
}
if err := orm.RunMigration(conn, logger, "record-retries-epoch", func(tx *sql.Tx) error {
if err := addColumnIfMissing(tx, "record_retries", "epoch", "integer not null default 0"); err != nil {
return err
}
_, err := tx.Exec(`
create index if not exists idx_record_retries_due on record_retries(due);
create index if not exists idx_repo_retries_due on repo_retries(due);
`)
return err
}); err != nil {
return err
}
if err := orm.RunMigration(conn, logger, "pull-ticket-version", func(tx *sql.Tx) error {
return addColumnIfMissing(tx, "pull_tickets", "version", "text not null default ''")
}); err != nil {
return err
}
if err := orm.RunMigration(conn, logger, "repo-turns", func(tx *sql.Tx) error {
_, err := tx.Exec(`
create table if not exists repo_turns (
repo_did text primary key,
seq integer not null,
touched integer not null
);
create index if not exists idx_repo_turns_touched on repo_turns(touched);
`)
return err
}); err != nil {
return err
}
if err := orm.RunMigration(conn, logger, "repo-pushes", func(tx *sql.Tx) error {
_, err := tx.Exec(`
create table if not exists repo_pushes (
repo_did text not null,
ref text not null,
old_sha text not null,
new_sha text not null,
seen integer not null,
primary key (repo_did, ref, old_sha, new_sha)
);
create index if not exists idx_repo_pushes_seen on repo_pushes(seen);
`)
return err
}); err != nil {
return err
}
if err := orm.RunMigration(conn, logger, "wipes", func(tx *sql.Tx) error {
_, err := tx.Exec(`
-- a repo whose rows are gone but whose leftovers still need cleaning, steps is what is left
create table if not exists wipes (
repo_did text primary key,
reason text not null,
steps text not null,
attempts integer not null default 0,
due integer not null
);
create index if not exists idx_wipes_due on wipes(due);
`)
return err
}); err != nil {
return err
}
if err := orm.RunMigration(conn, logger, "invite-codes", func(tx *sql.Tx) error {
_, err := tx.Exec(`
create table if not exists invite_codes (
code text primary key,
created_by text not null,
expires_at integer not null
);
create index if not exists idx_invite_codes_expires on invite_codes(expires_at);
`)
return err
}); err != nil {
return err
}
if err := orm.RunMigration(conn, logger, "jobs-repo-index", func(tx *sql.Tx) error {
_, err := tx.Exec(`create index if not exists idx_jobs_repo_did_id on jobs(repo_did, id)`)
return err
}); err != nil {
return err
}
if err := orm.RunMigration(conn, logger, "pipelines-repo-kind-index", func(tx *sql.Tx) error {
_, err := tx.Exec(`create index if not exists idx_pipelines_repo_kind_id on pipelines(repo_did, kind, id)`)
return err
}); err != nil {
return err
}
if err := orm.RunMigration(conn, logger, "quota-allocations-store-path", func(tx *sql.Tx) error {
_, err := tx.Exec(`
alter table quota_allocations add column store_path text;
create index if not exists idx_quota_allocations_store_path on quota_allocations(store_path);
`)
return err
}); err != nil {
return err
}
// scope is only there so usage can be listed without joining repo ownership
if err := orm.RunMigration(conn, logger, "quota-usage-periods", func(tx *sql.Tx) error {
_, err := tx.Exec(`
create table if not exists quota_usage_periods (
did text not null,
resource text not null,
period_start integer not null,
scope text not null check (scope in ('user', 'repo')),
used integer not null check (used >= 0),
primary key (did, resource, period_start)
);
create index if not exists idx_quota_usage_periods_start on quota_usage_periods(period_start);
create table if not exists quota_usage_ledger (
usage_id text primary key,
recorded_at integer not null
);
create index if not exists idx_quota_usage_ledger_recorded on quota_usage_ledger(recorded_at);
`)
return err
}); err != nil {
return err
}
// spec and source_repo are what the job row needs, so a blocked or finished pipeline can
// be queued again. patch_uri links a pull request pipeline back to its round
if err := orm.RunMigration(conn, logger, "pipelines-spec-and-patch", func(tx *sql.Tx) error {
for _, column := range []string{"spec", "source_repo", "patch_uri"} {
if err := addColumnIfMissing(tx, "pipelines", column, "text"); err != nil {
return err
}
}
if err := addColumnIfMissing(tx, "pull_events", "patch", "text not null default ''"); err != nil {
return err
}
_, err := tx.Exec(`
create index if not exists idx_pipelines_patch
on pipelines(repo_did, patch_uri) where patch_uri is not null
`)
return err
}); err != nil {
return err
}
// last_used_at is unix seconds. seq is commit order.
if err := orm.RunMigration(conn, logger, "nix-cache-index", func(tx *sql.Tx) error {
if err := addColumnIfMissing(tx, "quota_reservations", "nix_object", "text"); err != nil {
return err
}
_, err := tx.Exec(`
create table if not exists nix_objects (
cache text not null,
store_hash text not null,
repo_did text not null,
owner_did text not null,
store_path text not null,
nar_size integer not null check (nar_size >= 0),
seq integer not null,
last_used_at integer not null,
state text not null check (state in ('live', 'evicting')),
gen integer not null,
primary key (cache, store_hash)
);
create unique index if not exists idx_nix_objects_seq on nix_objects(seq);
create index if not exists idx_nix_objects_repo on nix_objects(repo_did, state, last_used_at);
create index if not exists idx_nix_objects_owner on nix_objects(owner_did, state, last_used_at);
create index if not exists idx_nix_objects_state_used on nix_objects(state, last_used_at);
create table if not exists nix_refs (
cache text not null,
from_hash text not null,
to_hash text not null,
primary key (cache, from_hash, to_hash)
);
create index if not exists idx_nix_refs_to on nix_refs(cache, to_hash, from_hash);
`)
return err
}); err != nil {
return err
}
// in seconds. the seeded row stands in for tokens from older spindles, which lived up to 6h05m
if err := orm.RunMigration(conn, logger, "nix-grants", func(tx *sql.Tx) error {
_, err := tx.Exec(`
create table if not exists nix_grants (
expires_at integer not null,
life integer not null
);
create index if not exists idx_nix_grants_expires on nix_grants(expires_at);
insert into nix_grants (expires_at, life) values (unixepoch() + 21900, 21900);
`)
return err
}); err != nil {
return err
}
// deleting entries still hold their bytes until the delete lands
if err := orm.RunMigration(conn, logger, "generic-cache-allocations", func(tx *sql.Tx) error {
_, err := tx.Exec(`
insert or ignore into quota_repo_owners (repo_did, owner_did)
select repo_did, owner_did from cache_entries;
insert or ignore into quota_allocations (repo_did, resource, kind, key, amount)
select repo_did, 'cache_storage_bytes', 'generic_cache', id, size_bytes from cache_entries
where state in ('ready', 'deleting');
`)
return err
}); err != nil {
return err
}
return nil
}
func addColumnIfMissing(tx *sql.Tx, table, column, definition string) error {
present, err := tableHasColumn(tx, table, column)
if err != nil || present {
return err
}
_, err = tx.Exec("alter table " + table + " add column " + column + " " + definition)
return err
}
func tableHasColumn(tx *sql.Tx, table, column string) (bool, error) {
var count int
if err := tx.QueryRow(
`select count(*) from pragma_table_info(?) where name = ?`,
table, column,
).Scan(&count); err != nil {
return false, err
}
return count != 0, nil
}
func migratePipelineIdentityColumns(tx *sql.Tx) error {
hasRkey, err := tableHasColumn(tx, "pipelines", "rkey")
if err != nil {
return err
}
if hasRkey {
if _, err := tx.Exec(`
alter table pipelines drop column knot;
alter table pipelines rename column rkey to pipeline_id;
`); err != nil {
return err
}
}
hasStatusRkey, err := tableHasColumn(tx, "workflow_statuses", "rkey")
if err != nil {
return err
}
if hasStatusRkey {
if _, err := tx.Exec(`
drop index if exists idx_workflow_statuses_lookup;
alter table workflow_statuses rename column rkey to pipeline_id;
create index idx_workflow_statuses_lookup
on workflow_statuses (pipeline_id, workflow, id);
`); err != nil {
return err
}
}
return 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
}
func (d *DB) GetRepoOwnerAndDid(knot, rkey string) (owner string, repoDid string, err error) {
err = d.QueryRow("SELECT owner, repo_did FROM repos WHERE knot = ? AND rkey = ?", knot, rkey).Scan(&owner, &repoDid)
return
}
// the event log from before pipelines had their own table
const (
legacyPipelineNSID = "sh.tangled.pipeline"
legacyPipelineStatusNSID = "sh.tangled.pipeline.status"
)
type legacyPipelineStatus struct {
CreatedAt string `json:"createdAt"`
Error *string `json:"error,omitempty"`
ExitCode *int64 `json:"exitCode,omitempty"`
Pipeline string `json:"pipeline"`
Status string `json:"status"`
Workflow string `json:"workflow"`
}
// migratePipelines orders converted rows by their original creation time so
// pagination keeps the historical order.
func migratePipelines(tx *sql.Tx, logger *slog.Logger) error {
rows, err := tx.Query(
`select rkey, event, created from events where nsid = ? order by created asc`,
legacyPipelineNSID,
)
if err != nil {
return err
}
defer rows.Close()
type converted struct {
rkey, knot, repoDid, commitSha, kind string
payload []byte
}
var out []converted
var skipped int
for rows.Next() {
var rkey, eventJson string
var created int64
if err := rows.Scan(&rkey, &eventJson, &created); err != nil {
return err
}
var raw workflow.PipelineSpec
if err := json.Unmarshal([]byte(eventJson), &raw); err != nil {
skipped++
continue
}
if raw.TriggerMetadata == nil {
skipped++
continue
}
record, err := pipelinecodec.FromTangled(models.PipelineId(rkey), time.Unix(0, created), raw)
if err != nil {
skipped++
continue
}
payload, err := json.Marshal(record)
if err != nil {
skipped++
continue
}
var knot string
if raw.TriggerMetadata.Repo != nil {
knot = raw.TriggerMetadata.Repo.Knot
}
out = append(out, converted{rkey, knot, record.RepoDID, record.Commit, record.Trigger.Kind, payload})
}
if err := rows.Err(); err != nil {
return err
}
for _, c := range out {
if _, err := tx.Exec(
`insert into pipelines (rkey, knot, repo_did, commit_sha, kind, payload) values (?, ?, ?, ?, ?, ?)`,
c.rkey, c.knot, c.repoDid, c.commitSha, c.kind, string(c.payload),
); err != nil {
return err
}
}
logger.Info("backfilled pipelines", "converted", len(out), "skipped", skipped)
return nil
}
// migrateWorkflowStatuses flattens the legacy status event log.
func migrateWorkflowStatuses(tx *sql.Tx, logger *slog.Logger) error {
rows, err := tx.Query(
`select event from events where nsid = ? order by created asc`,
legacyPipelineStatusNSID,
)
if err != nil {
return err
}
defer rows.Close()
type converted struct {
rkey, workflow, status, createdAt string
wfError *string
exitCode *int64
}
var out []converted
var skipped int
for rows.Next() {
var eventJson string
if err := rows.Scan(&eventJson); err != nil {
return err
}
var st legacyPipelineStatus
if err := json.Unmarshal([]byte(eventJson), &st); err != nil {
skipped++
continue
}
idx := strings.LastIndex(st.Pipeline, "/")
if idx < 0 || idx == len(st.Pipeline)-1 {
skipped++
continue
}
rkey := st.Pipeline[idx+1:]
out = append(out, converted{rkey, st.Workflow, st.Status, st.CreatedAt, st.Error, st.ExitCode})
}
if err := rows.Err(); err != nil {
return err
}
convertedCount := 0
for _, c := range out {
result, err := tx.Exec(
`insert into workflow_statuses (rkey, workflow, status, error, exit_code, created_at)
select ?, ?, ?, ?, ?, ?
where exists (select 1 from pipelines where rkey = ?)`,
c.rkey, c.workflow, c.status, c.wfError, c.exitCode, c.createdAt, c.rkey,
)
if err != nil {
return err
}
inserted, err := result.RowsAffected()
if err != nil {
return err
}
if inserted == 0 {
skipped++
continue
}
convertedCount++
}
logger.Info("backfilled workflow statuses", "converted", convertedCount, "skipped", skipped)
return nil
}