internal/store/store.go

  1// Package store owns SQLite access and schema migrations.
  2package store
  3
  4import (
  5	"context"
  6	"database/sql"
  7	"embed"
  8	"errors"
  9	"fmt"
 10	"io/fs"
 11	"log/slog"
 12	"os"
 13	"sort"
 14	"strconv"
 15	"strings"
 16	"sync"
 17
 18	"gitbay.org/gitbay/internal/seal"
 19	"modernc.org/sqlite"
 20)
 21
 22//go:embed migrations/*.sql
 23var migrationFS embed.FS
 24
 25type Store struct {
 26	DB *sql.DB
 27
 28	// logWait holds one channel per build someone is following, closed
 29	// by the next change to that build's row (BuildLogWait).
 30	logMu   sync.Mutex
 31	logWait map[int64]chan struct{}
 32
 33	// onRevoke runs after each key revocation this process commits.
 34	revokeMu sync.Mutex
 35	onRevoke []func(Revoked)
 36
 37	// AuditJournal, when set, receives a copy of every audit row. The
 38	// daemon sets it to its own logger, whose output the service
 39	// journal keeps outside the database.
 40	AuditJournal *slog.Logger
 41	// secrets seals and opens the secret columns (secrets.go). Nil
 42	// stores values as given and refuses to open sealed ones.
 43	secrets *seal.Keyring
 44}
 45
 46// Open opens (creating if needed) the database at path with WAL mode and
 47// foreign keys enforced. Use ":memory:" in tests.
 48//
 49// _txlock=immediate is what serialises writers. Every transaction in this
 50// package writes, and a deferred one takes the write lock only when it
 51// reaches its first write — by which point another writer may hold it.
 52// SQLite answers that with SQLITE_BUSY and does not invoke the busy
 53// handler, because waiting would deadlock two transactions each holding a
 54// read lock the other needs; busy_timeout cannot help. Measured with
 55// eight concurrent read-then-write transactions, 44% of them failed.
 56// Beginning immediate takes the write lock up front, where busy_timeout
 57// does apply, so a second writer waits its turn: the same load runs with
 58// no failures, and readers, which WAL keeps out of the way, are
 59// unaffected (#121).
 60func Open(path string) (*Store, error) {
 61	// synchronous(FULL): a commit is durable before it returns, which
 62	// secret key rotation needs before it drops the old keys (#273).
 63	dsn := path + "?_txlock=immediate&_pragma=journal_mode(WAL)&_pragma=synchronous(FULL)&_pragma=foreign_keys(ON)&_pragma=busy_timeout(5000)"
 64	if path == ":memory:" {
 65		dsn = ":memory:?_txlock=immediate&_pragma=foreign_keys(ON)"
 66	}
 67	db, err := sql.Open("sqlite", dsn)
 68	if err != nil {
 69		return nil, err
 70	}
 71	if err := db.Ping(); err != nil {
 72		db.Close()
 73		return nil, err
 74	}
 75	// SQLite creates the file 0666&~umask, so it lands 0644 by default. The
 76	// directory above it is the real boundary, but the file holds token
 77	// hashes, addresses and private repo names and has no business being
 78	// world-readable on its own.
 79	if path != ":memory:" {
 80		if err := os.Chmod(path, 0o640); err != nil && !errors.Is(err, fs.ErrNotExist) {
 81			db.Close()
 82			return nil, err
 83		}
 84	}
 85	return &Store{DB: db}, nil
 86}
 87
 88func (s *Store) Close() error { return s.DB.Close() }
 89
 90type migration struct {
 91	version int
 92	name    string
 93	up      string
 94	down    string
 95	// upFKOff and downFKOff are true when the up/down script's first line
 96	// is the directive "-- foreign_keys: off".
 97	upFKOff   bool
 98	downFKOff bool
 99}
100
101// fkOffDirective, as the first line of a migration script, opts that
102// direction out of foreign-key enforcement for its step.
103const fkOffDirective = "-- foreign_keys: off"
104
105func loadMigrations() ([]migration, error) {
106	entries, err := fs.ReadDir(migrationFS, "migrations")
107	if err != nil {
108		return nil, err
109	}
110	byVersion := map[int]*migration{}
111	for _, e := range entries {
112		name := e.Name()
113		// <version>_<name>.<up|down>.sql
114		base, ok := strings.CutSuffix(name, ".sql")
115		if !ok {
116			return nil, fmt.Errorf("migration %q: not .sql", name)
117		}
118		var dir string
119		if b, ok := strings.CutSuffix(base, ".up"); ok {
120			base, dir = b, "up"
121		} else if b, ok := strings.CutSuffix(base, ".down"); ok {
122			base, dir = b, "down"
123		} else {
124			return nil, fmt.Errorf("migration %q: missing .up/.down", name)
125		}
126		verStr, rest, ok := strings.Cut(base, "_")
127		if !ok {
128			return nil, fmt.Errorf("migration %q: missing version prefix", name)
129		}
130		ver, err := strconv.Atoi(verStr)
131		if err != nil {
132			return nil, fmt.Errorf("migration %q: bad version: %w", name, err)
133		}
134		m := byVersion[ver]
135		if m == nil {
136			m = &migration{version: ver, name: rest}
137			byVersion[ver] = m
138		}
139		sqlBytes, err := migrationFS.ReadFile("migrations/" + name)
140		if err != nil {
141			return nil, err
142		}
143		text := string(sqlBytes)
144		firstLine, _, _ := strings.Cut(text, "\n")
145		fkOff := strings.TrimSpace(firstLine) == fkOffDirective
146		if dir == "up" {
147			m.up = text
148			m.upFKOff = fkOff
149		} else {
150			m.down = text
151			m.downFKOff = fkOff
152		}
153	}
154	var ms []migration
155	for _, m := range byVersion {
156		if m.up == "" || m.down == "" {
157			return nil, fmt.Errorf("migration %d %q: missing up or down file", m.version, m.name)
158		}
159		ms = append(ms, *m)
160	}
161	sort.Slice(ms, func(i, j int) bool { return ms[i].version < ms[j].version })
162	for i, m := range ms {
163		if m.version != i+1 {
164			return nil, fmt.Errorf("migration versions not contiguous at %d", m.version)
165		}
166	}
167	return ms, nil
168}
169
170// Version returns the current schema version (0 = empty database).
171func (s *Store) Version() (int, error) {
172	var v int
173	err := s.DB.QueryRow("PRAGMA user_version").Scan(&v)
174	return v, err
175}
176
177// TakeMigrationNote returns and clears what a migration left to be
178// logged with it, "" for nothing.
179func (s *Store) TakeMigrationNote() (string, error) {
180	var note string
181	err := s.DB.QueryRow("DELETE FROM settings WHERE key = 'migration_note' RETURNING value").Scan(&note)
182	if errors.Is(err, sql.ErrNoRows) {
183		return "", nil
184	}
185	return note, err
186}
187
188// MigrateUp applies all pending migrations.
189func (s *Store) MigrateUp() error { return s.migrateTo(-1) }
190
191// MigrateTo migrates up or down to the given version. 0 empties the schema.
192func (s *Store) MigrateTo(target int) error { return s.migrateTo(target) }
193
194func (s *Store) migrateTo(target int) error {
195	ms, err := loadMigrations()
196	if err != nil {
197		return err
198	}
199	if target < 0 {
200		target = len(ms)
201	}
202	if target > len(ms) {
203		return fmt.Errorf("no such schema version %d (max %d)", target, len(ms))
204	}
205	cur, err := s.Version()
206	if err != nil {
207		return err
208	}
209	for cur < target {
210		m := ms[cur]
211		if err := s.migrateStep(m.up, m.version, m.upFKOff); err != nil {
212			return fmt.Errorf("migration %d up: %w", m.version, err)
213		}
214		cur = m.version
215	}
216	for cur > target {
217		m := ms[cur-1]
218		if err := s.migrateStep(m.down, m.version-1, m.downFKOff); err != nil {
219			return fmt.Errorf("migration %d down: %w", m.version, err)
220		}
221		cur = m.version - 1
222	}
223	return nil
224}
225
226func (s *Store) migrateStep(sqlText string, newVersion int, fkOff bool) (retErr error) {
227	if !fkOff {
228		tx, err := s.DB.Begin()
229		if err != nil {
230			return err
231		}
232		defer tx.Rollback()
233		if _, err := tx.Exec(sqlText); err != nil {
234			return err
235		}
236		if _, err := tx.Exec(fmt.Sprintf("PRAGMA user_version = %d", newVersion)); err != nil {
237			return err
238		}
239		return tx.Commit()
240	}
241
242	// A script whose first line is "-- foreign_keys: off" rebuilds a
243	// table that other tables reference (labels, milestones): with
244	// foreign keys on, the rebuild-by-rename loses the children's
245	// rows. PRAGMA foreign_keys is a no-op inside a transaction, and
246	// the pool gives no guarantee that a pragma set on one connection
247	// is seen by the connection Begin() draws next, so the whole step
248	// — pragma off, transaction, foreign_key_check, commit, pragma on —
249	// runs on a single pinned connection. The check runs before commit:
250	// checking after would report a violation once the bad schema and
251	// user_version were already persisted.
252	ctx := context.Background()
253	conn, err := s.DB.Conn(ctx)
254	if err != nil {
255		return err
256	}
257	defer conn.Close()
258	if _, err := conn.ExecContext(ctx, "PRAGMA foreign_keys = OFF"); err != nil {
259		return err
260	}
261	// The connection goes back to the pool when this returns, so every
262	// path out of here has to put foreign keys back on first.
263	defer func() {
264		if _, err := conn.ExecContext(ctx, "PRAGMA foreign_keys = ON"); err != nil && retErr == nil {
265			retErr = err
266		}
267	}()
268	tx, err := conn.BeginTx(ctx, nil)
269	if err != nil {
270		return err
271	}
272	defer tx.Rollback()
273	if _, err := tx.Exec(sqlText); err != nil {
274		return err
275	}
276	if _, err := tx.Exec(fmt.Sprintf("PRAGMA user_version = %d", newVersion)); err != nil {
277		return err
278	}
279	// foreign_key_check works with enforcement off: it inspects the data
280	// directly rather than consulting the pragma. Running it here, inside
281	// the transaction, means a violation rolls back the whole rebuild
282	// (the deferred tx.Rollback fires) instead of leaving the bad schema
283	// and version committed.
284	rows, err := tx.QueryContext(ctx, "PRAGMA foreign_key_check")
285	if err != nil {
286		return err
287	}
288	if rows.Next() {
289		var table string
290		var rowid sql.NullInt64
291		var referredTable string
292		var fkid int
293		if err := rows.Scan(&table, &rowid, &referredTable, &fkid); err != nil {
294			rows.Close()
295			return err
296		}
297		rows.Close()
298		return fmt.Errorf("foreign_key_check failed after migration: %s row %d", table, rowid.Int64)
299	}
300	if err := rows.Err(); err != nil {
301		rows.Close()
302		return err
303	}
304	rows.Close()
305	return tx.Commit()
306}
307
308// IsInternal reports whether err is the database or the I/O beneath it
309// failing, as opposed to a sentinel or a message about the caller's
310// input. Callers map it to a failure exit rather than a usage error.
311func IsInternal(err error) bool {
312	var sqlErr *sqlite.Error
313	var pathErr *fs.PathError
314	return errors.As(err, &sqlErr) || errors.As(err, &pathErr) ||
315		errors.Is(err, sql.ErrTxDone) || errors.Is(err, sql.ErrConnDone) ||
316		errors.Is(err, context.DeadlineExceeded) || errors.Is(err, context.Canceled)
317}