// Package store owns SQLite access and schema migrations. package store import ( "context" "database/sql" "embed" "errors" "fmt" "io/fs" "log/slog" "os" "sort" "strconv" "strings" "sync" "modernc.org/sqlite" ) //go:embed migrations/*.sql var migrationFS embed.FS type Store struct { DB *sql.DB // logWait holds one channel per build someone is following, closed // by the next change to that build's row (BuildLogWait). logMu sync.Mutex logWait map[int64]chan struct{} // onRevoke runs after each key revocation this process commits. revokeMu sync.Mutex onRevoke []func(Revoked) // AuditJournal, when set, receives a copy of every audit row. The // daemon sets it to its own logger, whose output the service // journal keeps outside the database. AuditJournal *slog.Logger } // Open opens (creating if needed) the database at path with WAL mode and // foreign keys enforced. Use ":memory:" in tests. // // _txlock=immediate is what serialises writers. Every transaction in this // package writes, and a deferred one takes the write lock only when it // reaches its first write — by which point another writer may hold it. // SQLite answers that with SQLITE_BUSY and does not invoke the busy // handler, because waiting would deadlock two transactions each holding a // read lock the other needs; busy_timeout cannot help. Measured with // eight concurrent read-then-write transactions, 44% of them failed. // Beginning immediate takes the write lock up front, where busy_timeout // does apply, so a second writer waits its turn: the same load runs with // no failures, and readers, which WAL keeps out of the way, are // unaffected (#121). func Open(path string) (*Store, error) { dsn := path + "?_txlock=immediate&_pragma=journal_mode(WAL)&_pragma=foreign_keys(ON)&_pragma=busy_timeout(5000)" if path == ":memory:" { dsn = ":memory:?_txlock=immediate&_pragma=foreign_keys(ON)" } db, err := sql.Open("sqlite", dsn) if err != nil { return nil, err } if err := db.Ping(); err != nil { db.Close() return nil, err } // SQLite creates the file 0666&~umask, so it lands 0644 by default. The // directory above it is the real boundary, but the file holds token // hashes, addresses and private repo names and has no business being // world-readable on its own. if path != ":memory:" { if err := os.Chmod(path, 0o640); err != nil && !errors.Is(err, fs.ErrNotExist) { db.Close() return nil, err } } return &Store{DB: db}, nil } func (s *Store) Close() error { return s.DB.Close() } type migration struct { version int name string up string down string // upFKOff and downFKOff are true when the up/down script's first line // is the directive "-- foreign_keys: off". upFKOff bool downFKOff bool } // fkOffDirective, as the first line of a migration script, opts that // direction out of foreign-key enforcement for its step. const fkOffDirective = "-- foreign_keys: off" func loadMigrations() ([]migration, error) { entries, err := fs.ReadDir(migrationFS, "migrations") if err != nil { return nil, err } byVersion := map[int]*migration{} for _, e := range entries { name := e.Name() // _..sql base, ok := strings.CutSuffix(name, ".sql") if !ok { return nil, fmt.Errorf("migration %q: not .sql", name) } var dir string if b, ok := strings.CutSuffix(base, ".up"); ok { base, dir = b, "up" } else if b, ok := strings.CutSuffix(base, ".down"); ok { base, dir = b, "down" } else { return nil, fmt.Errorf("migration %q: missing .up/.down", name) } verStr, rest, ok := strings.Cut(base, "_") if !ok { return nil, fmt.Errorf("migration %q: missing version prefix", name) } ver, err := strconv.Atoi(verStr) if err != nil { return nil, fmt.Errorf("migration %q: bad version: %w", name, err) } m := byVersion[ver] if m == nil { m = &migration{version: ver, name: rest} byVersion[ver] = m } sqlBytes, err := migrationFS.ReadFile("migrations/" + name) if err != nil { return nil, err } text := string(sqlBytes) firstLine, _, _ := strings.Cut(text, "\n") fkOff := strings.TrimSpace(firstLine) == fkOffDirective if dir == "up" { m.up = text m.upFKOff = fkOff } else { m.down = text m.downFKOff = fkOff } } var ms []migration for _, m := range byVersion { if m.up == "" || m.down == "" { return nil, fmt.Errorf("migration %d %q: missing up or down file", m.version, m.name) } ms = append(ms, *m) } sort.Slice(ms, func(i, j int) bool { return ms[i].version < ms[j].version }) for i, m := range ms { if m.version != i+1 { return nil, fmt.Errorf("migration versions not contiguous at %d", m.version) } } return ms, nil } // Version returns the current schema version (0 = empty database). func (s *Store) Version() (int, error) { var v int err := s.DB.QueryRow("PRAGMA user_version").Scan(&v) return v, err } // MigrateUp applies all pending migrations. func (s *Store) MigrateUp() error { return s.migrateTo(-1) } // MigrateTo migrates up or down to the given version. 0 empties the schema. func (s *Store) MigrateTo(target int) error { return s.migrateTo(target) } func (s *Store) migrateTo(target int) error { ms, err := loadMigrations() if err != nil { return err } if target < 0 { target = len(ms) } if target > len(ms) { return fmt.Errorf("no such schema version %d (max %d)", target, len(ms)) } cur, err := s.Version() if err != nil { return err } for cur < target { m := ms[cur] if err := s.migrateStep(m.up, m.version, m.upFKOff); err != nil { return fmt.Errorf("migration %d up: %w", m.version, err) } cur = m.version } for cur > target { m := ms[cur-1] if err := s.migrateStep(m.down, m.version-1, m.downFKOff); err != nil { return fmt.Errorf("migration %d down: %w", m.version, err) } cur = m.version - 1 } return nil } func (s *Store) migrateStep(sqlText string, newVersion int, fkOff bool) (retErr error) { if !fkOff { tx, err := s.DB.Begin() if err != nil { return err } defer tx.Rollback() if _, err := tx.Exec(sqlText); err != nil { return err } if _, err := tx.Exec(fmt.Sprintf("PRAGMA user_version = %d", newVersion)); err != nil { return err } return tx.Commit() } // A script whose first line is "-- foreign_keys: off" rebuilds a // table that other tables reference (labels, milestones): with // foreign keys on, the rebuild-by-rename loses the children's // rows. PRAGMA foreign_keys is a no-op inside a transaction, and // the pool gives no guarantee that a pragma set on one connection // is seen by the connection Begin() draws next, so the whole step // — pragma off, transaction, foreign_key_check, commit, pragma on — // runs on a single pinned connection. The check runs before commit: // checking after would report a violation once the bad schema and // user_version were already persisted. ctx := context.Background() conn, err := s.DB.Conn(ctx) if err != nil { return err } defer conn.Close() if _, err := conn.ExecContext(ctx, "PRAGMA foreign_keys = OFF"); err != nil { return err } // The connection goes back to the pool when this returns, so every // path out of here has to put foreign keys back on first. defer func() { if _, err := conn.ExecContext(ctx, "PRAGMA foreign_keys = ON"); err != nil && retErr == nil { retErr = err } }() tx, err := conn.BeginTx(ctx, nil) if err != nil { return err } defer tx.Rollback() if _, err := tx.Exec(sqlText); err != nil { return err } if _, err := tx.Exec(fmt.Sprintf("PRAGMA user_version = %d", newVersion)); err != nil { return err } // foreign_key_check works with enforcement off: it inspects the data // directly rather than consulting the pragma. Running it here, inside // the transaction, means a violation rolls back the whole rebuild // (the deferred tx.Rollback fires) instead of leaving the bad schema // and version committed. rows, err := tx.QueryContext(ctx, "PRAGMA foreign_key_check") if err != nil { return err } if rows.Next() { var table string var rowid sql.NullInt64 var referredTable string var fkid int if err := rows.Scan(&table, &rowid, &referredTable, &fkid); err != nil { rows.Close() return err } rows.Close() return fmt.Errorf("foreign_key_check failed after migration: %s row %v", table, rowid) } if err := rows.Err(); err != nil { rows.Close() return err } rows.Close() return tx.Commit() } // IsInternal reports whether err is the database or the I/O beneath it // failing, as opposed to a sentinel or a message about the caller's // input. Callers map it to a failure exit rather than a usage error. func IsInternal(err error) bool { var sqlErr *sqlite.Error var pathErr *fs.PathError return errors.As(err, &sqlErr) || errors.As(err, &pathErr) || errors.Is(err, sql.ErrTxDone) || errors.Is(err, sql.ErrConnDone) || errors.Is(err, context.DeadlineExceeded) || errors.Is(err, context.Canceled) }