NextMyDay/internal/core/store/migrate.go
Claus Lohmar d11e14295b feat(core): add the store, identity plumbing and the door (SPEC-003/004)
- store: SQLite (WAL, foreign keys ON, CGO-free driver) + the embedded
  migration runner; each module's migrations are name-prefixed (D-012)
- audit: the append-only audit log; the actor threads via context (R-1)
- users: the user_id/GUID/email anchors (D-013)
- config: the YAML settings file (D-035) + validation
- httpx: cookies · loopback detection · trusted-proxy XFF · CSRF · security
  headers · redaction
- middleware: correlation id · audit · rate limiting
- mail: the invite/recovery mailer seam
- server: the core DOOR — the resolution switch, the dispatch registry
  (RegisterModule), the role gates, /setup · enroll/login · recovery flows,
  the reserved `static` workflow, and the loopback-only bootstrap (D-025)
2026-10-14 10:53:51 +00:00

144 lines
4.2 KiB
Go

package store
import (
"database/sql"
"fmt"
"io/fs"
"path"
"sort"
"strings"
"time"
)
// Migration is one ordered, named SQL migration — one file = one version. A
// migration may contain multiple statements; the driver executes the whole
// script atomically inside a transaction.
type Migration struct {
Version string // stable key, e.g. "0001_init" (the file name minus .sql)
SQL string
}
// LoadMigrations reads every <dir>/*.sql in fsys, sorted by file name so the
// order is deterministic (D-028 guardrail: "never map iteration").
func LoadMigrations(fsys fs.FS, dir string) ([]Migration, error) {
pattern := path.Join(dir, "*.sql")
names, err := fs.Glob(fsys, pattern)
if err != nil {
return nil, fmt.Errorf("store: glob %q: %w", pattern, err)
}
sort.Strings(names)
out := make([]Migration, 0, len(names))
for _, name := range names {
b, err := fs.ReadFile(fsys, name)
if err != nil {
return nil, fmt.Errorf("store: read migration %q: %w", name, err)
}
version := strings.TrimSuffix(path.Base(name), ".sql")
out = append(out, Migration{Version: version, SQL: string(b)})
}
return out, nil
}
// Migrate applies the core schema to the database. It is idempotent: already
// applied versions are skipped.
func (db *DB) Migrate() error {
migrations, err := LoadMigrations(coreMigrations, "migrations")
if err != nil {
return err
}
return RunMigrations(db.DB, migrations)
}
// RunMigrations applies each migration not yet recorded, in order, each inside
// its own transaction. It is the runner the core owns; modules will call it
// with their own embedded migration sets later (D-028).
func RunMigrations(db *sql.DB, migrations []Migration) error {
if err := ensureMigrationsTable(db); err != nil {
return err
}
applied, err := appliedVersions(db)
if err != nil {
return err
}
for _, m := range migrations {
if applied[m.Version] {
continue
}
if err := applyMigration(db, m); err != nil {
return err
}
}
return nil
}
// ensureMigrationsTable creates the bookkeeping table if it does not exist.
func ensureMigrationsTable(db *sql.DB) error {
const ddl = `CREATE TABLE IF NOT EXISTS schema_migrations (
version TEXT PRIMARY KEY,
applied_at TEXT NOT NULL
)`
if _, err := db.Exec(ddl); err != nil {
return fmt.Errorf("store: create schema_migrations: %w", err)
}
return nil
}
// appliedVersions returns the set of versions already recorded.
func appliedVersions(db *sql.DB) (map[string]bool, error) {
rows, err := db.Query(`SELECT version FROM schema_migrations`)
if err != nil {
return nil, fmt.Errorf("store: read schema_migrations: %w", err)
}
defer rows.Close()
applied := map[string]bool{}
for rows.Next() {
var v string
if err := rows.Scan(&v); err != nil {
return nil, fmt.Errorf("store: scan schema_migrations: %w", err)
}
applied[v] = true
}
if err := rows.Err(); err != nil {
return nil, fmt.Errorf("store: iterate schema_migrations: %w", err)
}
return applied, nil
}
// applyMigration runs one migration and records it in a single transaction —
// SQLite DDL is transactional, so a failure leaves no partial schema.
func applyMigration(db *sql.DB, m Migration) error {
tx, err := db.Begin()
if err != nil {
return fmt.Errorf("store: begin migration %q: %w", m.Version, err)
}
defer func() { _ = tx.Rollback() }()
if _, err := tx.Exec(m.SQL); err != nil {
return fmt.Errorf("store: apply migration %q: %w", m.Version, err)
}
if _, err := tx.Exec(
`INSERT INTO schema_migrations (version, applied_at) VALUES (?, ?)`,
m.Version, FormatTime(time.Now()),
); err != nil {
return fmt.Errorf("store: record migration %q: %w", m.Version, err)
}
if err := tx.Commit(); err != nil {
return fmt.Errorf("store: commit migration %q: %w", m.Version, err)
}
return nil
}
// FormatTime renders a timestamp in the canonical storage format: RFC 3339,
// UTC, second precision. Fixed width and UTC mean lexicographic order equals
// chronological order, so SQL range comparisons are sound.
func FormatTime(t time.Time) string {
return t.UTC().Truncate(time.Second).Format(time.RFC3339)
}
// ParseTime parses the canonical storage format produced by FormatTime.
func ParseTime(s string) (time.Time, error) {
return time.Parse(time.RFC3339, s)
}