// Package db owns the SQLite connection, schema migrations, and all queries // used by the Control API. It is the only package in the system that talks // to the database directly — agents and probers never connect to it. package db import ( "context" "database/sql" _ "embed" "fmt" "time" _ "modernc.org/sqlite" ) //go:embed migrations/0001_init.sql var initSchema string //go:embed migrations/0002_dynamic_config.sql var dynamicConfigSchema string //go:embed migrations/0003_fip_settle_delay.sql var fipSettleDelaySchema string //go:embed migrations/0004_inbound_checks_admin.sql var inboundChecksAdminSchema string //go:embed migrations/0005_unbounded_sites.sql var unboundedSitesSchema string //go:embed migrations/0006_prober_heartbeat.sql var proberHeartbeatSchema string //go:embed migrations/0007_ip_registry.sql var ipRegistrySchema string //go:embed migrations/0008_auto_cycle.sql var autoCycleSchema string //go:embed migrations/0009_scale_indexes.sql var scaleIndexesSchema string // migrations is the ordered list of schema versions. Each entry's SQL is // applied, in order, for any version greater than the database's current // PRAGMA user_version — so a fresh database walks the whole list and an // existing one only picks up what's new. var migrations = []struct { version int sql string }{ {1, initSchema}, {2, dynamicConfigSchema}, {3, fipSettleDelaySchema}, {4, inboundChecksAdminSchema}, {5, unboundedSitesSchema}, {6, proberHeartbeatSchema}, {7, ipRegistrySchema}, {8, autoCycleSchema}, {9, scaleIndexesSchema}, } type DB struct { *sql.DB } // Open opens (creating if necessary) the SQLite database at path, applies // pragmas suited to a single-writer WAL workload, and runs any pending // schema migrations. func Open(ctx context.Context, path string) (*DB, error) { sqlDB, err := sql.Open("sqlite", path+"?_pragma=busy_timeout(5000)") if err != nil { return nil, fmt.Errorf("open sqlite: %w", err) } // Control API is the sole writer; one connection avoids SQLITE_BUSY // entirely for writes while still allowing concurrent reads via WAL. sqlDB.SetMaxOpenConns(1) for _, pragma := range []string{ "PRAGMA journal_mode=WAL", "PRAGMA synchronous=NORMAL", "PRAGMA foreign_keys=ON", "PRAGMA busy_timeout=5000", } { if _, err := sqlDB.ExecContext(ctx, pragma); err != nil { sqlDB.Close() return nil, fmt.Errorf("apply pragma %q: %w", pragma, err) } } d := &DB{DB: sqlDB} if err := d.migrate(ctx); err != nil { sqlDB.Close() return nil, fmt.Errorf("migrate: %w", err) } return d, nil } // migrate applies every pending migration in order, tracked via // PRAGMA user_version so repeated startups only apply what's new (and a // fresh database walks the whole list once). func (d *DB) migrate(ctx context.Context) error { var version int if err := d.QueryRowContext(ctx, "PRAGMA user_version").Scan(&version); err != nil { return fmt.Errorf("read user_version: %w", err) } for _, m := range migrations { if m.version <= version { continue } tx, err := d.BeginTx(ctx, nil) if err != nil { return err } if _, err := tx.ExecContext(ctx, m.sql); err != nil { tx.Rollback() return fmt.Errorf("apply migration %d: %w", m.version, err) } if _, err := tx.ExecContext(ctx, fmt.Sprintf("PRAGMA user_version=%d", m.version)); err != nil { tx.Rollback() return fmt.Errorf("set user_version=%d: %w", m.version, err) } if err := tx.Commit(); err != nil { return fmt.Errorf("commit migration %d: %w", m.version, err) } } return nil } // Now returns the current time truncated to millisecond precision, the // granularity used consistently for all timestamp columns. func Now() time.Time { return time.Now().UTC().Truncate(time.Millisecond) }