Feature 02 task 1: SQLite connection and embedded migration runner (#10)

Co-authored-by: dev_agent <dev_agent@cubecraftlabs.com>
Co-committed-by: dev_agent <dev_agent@cubecraftlabs.com>

internal/store/migrate.go

@@ -0,0 +1,105 @@
+package store
+
+import (
+	"context"
+	"database/sql"
+	"fmt"
+	"io/fs"
+	"strings"
+	"time"
+)
+
+const createSchemaMigrations = `CREATE TABLE IF NOT EXISTS schema_migrations (
+	version    TEXT PRIMARY KEY,
+	applied_at TEXT NOT NULL
+)`
+
+// Migrate applies every *.sql file at the root of migrations that is not yet
+// recorded in schema_migrations, in filename order. Each migration runs in its
+// own transaction together with the insert that records it, so a failing
+// migration leaves the database at the previous version. A version is the
+// file name without its .sql extension.
+//
+// The database must not hold a version unknown to migrations: that means it
+// was migrated by a newer build, and Migrate refuses to continue.
+func Migrate(ctx context.Context, db *sql.DB, migrations fs.FS) error {
+	files, err := fs.Glob(migrations, "*.sql")
+	if err != nil {
+		return fmt.Errorf("migrate: list migrations: %w", err)
+	}
+	if len(files) == 0 {
+		return fmt.Errorf("migrate: no migrations found")
+	}
+
+	if _, err := db.ExecContext(ctx, createSchemaMigrations); err != nil {
+		return fmt.Errorf("migrate: create schema_migrations: %w", err)
+	}
+	applied, err := appliedVersions(ctx, db)
+	if err != nil {
+		return err
+	}
+
+	known := make(map[string]bool, len(files))
+	for _, name := range files {
+		known[strings.TrimSuffix(name, ".sql")] = true
+	}
+	for version := range applied {
+		if !known[version] {
+			return fmt.Errorf("migrate: database has version %q, which this build does not know", version)
+		}
+	}
+
+	for _, name := range files { // fs.Glob returns names in lexical order
+		version := strings.TrimSuffix(name, ".sql")
+		if applied[version] {
+			continue
+		}
+		body, err := fs.ReadFile(migrations, name)
+		if err != nil {
+			return fmt.Errorf("migrate: read %s: %w", name, err)
+		}
+		if err := apply(ctx, db, version, string(body)); err != nil {
+			return fmt.Errorf("migrate: apply %s: %w", name, err)
+		}
+	}
+	return nil
+}
+
+func appliedVersions(ctx context.Context, db *sql.DB) (map[string]bool, error) {
+	rows, err := db.QueryContext(ctx, `SELECT version FROM schema_migrations`)
+	if err != nil {
+		return nil, fmt.Errorf("migrate: read schema_migrations: %w", err)
+	}
+	defer rows.Close()
+	applied := make(map[string]bool)
+	for rows.Next() {
+		var v string
+		if err := rows.Scan(&v); err != nil {
+			return nil, fmt.Errorf("migrate: read schema_migrations: %w", err)
+		}
+		applied[v] = true
+	}
+	if err := rows.Err(); err != nil {
+		return nil, fmt.Errorf("migrate: read schema_migrations: %w", err)
+	}
+	return applied, nil
+}
+
+func apply(ctx context.Context, db *sql.DB, version, body string) error {
+	tx, err := db.BeginTx(ctx, nil)
+	if err != nil {
+		return err
+	}
+	defer tx.Rollback() // no-op once committed
+
+	if _, err := tx.ExecContext(ctx, body); err != nil {
+		return err
+	}
+	if _, err := tx.ExecContext(ctx,
+		`INSERT INTO schema_migrations (version, applied_at) VALUES (?, ?)`,
+		version, time.Now().UTC().Format(time.RFC3339),
+	); err != nil {
+		return err
+	}
+	return tx.Commit()
+}

internal/store/migrate_test.go

@@ -0,0 +1,206 @@
+package store
+
+import (
+	"context"
+	"io/fs"
+	"strings"
+	"testing"
+	"testing/fstest"
+	"time"
+
+	"code.cubecraftlabs.com/Maniacs_Mansion/Meal-Quest/migrations"
+)
+
+var testMigrations = fstest.MapFS{
+	"0001_first.sql":  {Data: []byte("CREATE TABLE first (id INTEGER PRIMARY KEY);")},
+	"0002_second.sql": {Data: []byte("CREATE TABLE second (id INTEGER PRIMARY KEY);\nCREATE TABLE third (id INTEGER PRIMARY KEY);")},
+	"README.md":       {Data: []byte("not a migration")},
+}
+
+type migrationRow struct {
+	version   string
+	appliedAt string
+}
+
+func migrationRows(t *testing.T, s *Store) []migrationRow {
+	t.Helper()
+	rows, err := s.Reader.Query("SELECT version, applied_at FROM schema_migrations ORDER BY version")
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer rows.Close()
+	var out []migrationRow
+	for rows.Next() {
+		var r migrationRow
+		if err := rows.Scan(&r.version, &r.appliedAt); err != nil {
+			t.Fatal(err)
+		}
+		out = append(out, r)
+	}
+	if err := rows.Err(); err != nil {
+		t.Fatal(err)
+	}
+	return out
+}
+
+func versions(rows []migrationRow) string {
+	vs := make([]string, len(rows))
+	for i, r := range rows {
+		vs[i] = r.version
+	}
+	return strings.Join(vs, ",")
+}
+
+func tableExists(t *testing.T, s *Store, name string) bool {
+	t.Helper()
+	var n int
+	if err := s.Reader.QueryRow("SELECT count(*) FROM sqlite_schema WHERE type = 'table' AND name = ?", name).Scan(&n); err != nil {
+		t.Fatal(err)
+	}
+	return n == 1
+}
+
+func TestMigrateFreshDatabaseReachesLatest(t *testing.T) {
+	s := openTestStore(t)
+	if err := Migrate(context.Background(), s.Writer, testMigrations); err != nil {
+		t.Fatal(err)
+	}
+	rows := migrationRows(t, s)
+	if got := versions(rows); got != "0001_first,0002_second" {
+		t.Fatalf("versions = %q", got)
+	}
+	for _, r := range rows {
+		if _, err := time.Parse(time.RFC3339, r.appliedAt); err != nil || !strings.HasSuffix(r.appliedAt, "Z") {
+			t.Errorf("applied_at %q is not UTC RFC 3339", r.appliedAt)
+		}
+	}
+	for _, table := range []string{"first", "second", "third"} {
+		if !tableExists(t, s, table) {
+			t.Errorf("table %s missing", table)
+		}
+	}
+}
+
+func TestMigrateSecondRunIsNoOp(t *testing.T) {
+	ctx := context.Background()
+	s := openTestStore(t)
+	if err := Migrate(ctx, s.Writer, testMigrations); err != nil {
+		t.Fatal(err)
+	}
+	before := migrationRows(t, s)
+	if err := Migrate(ctx, s.Writer, testMigrations); err != nil {
+		t.Fatalf("second run: %v", err)
+	}
+	after := migrationRows(t, s)
+	if len(after) != len(before) {
+		t.Fatalf("rows after second run = %v, want %v", after, before)
+	}
+	for i := range before {
+		if after[i] != before[i] {
+			t.Errorf("row %d changed: %v -> %v", i, before[i], after[i])
+		}
+	}
+}
+
+func TestMigrateBrokenMigrationRollsBack(t *testing.T) {
+	ctx := context.Background()
+	s := openTestStore(t)
+	if err := Migrate(ctx, s.Writer, testMigrations); err != nil {
+		t.Fatal(err)
+	}
+
+	broken := fstest.MapFS{"0003_broken.sql": {Data: []byte(
+		// The first statement succeeds; the second is invalid, so the
+		// table created by the first must be rolled back with it.
+		"CREATE TABLE fourth (id INTEGER PRIMARY KEY);\nCREATE TABLE oops (;",
+	)}}
+	for name, f := range testMigrations {
+		broken[name] = f
+	}
+	err := Migrate(ctx, s.Writer, broken)
+	if err == nil {
+		t.Fatal("broken migration returned no error")
+	}
+	if !strings.Contains(err.Error(), "0003_broken.sql") {
+		t.Errorf("error %q does not name the failing migration", err)
+	}
+	if got := versions(migrationRows(t, s)); got != "0001_first,0002_second" {
+		t.Errorf("versions after failure = %q, want previous version intact", got)
+	}
+	if tableExists(t, s, "fourth") {
+		t.Error("table from the failed migration was not rolled back")
+	}
+	for _, table := range []string{"first", "second", "third"} {
+		if !tableExists(t, s, table) {
+			t.Errorf("previous table %s missing after failed migration", table)
+		}
+	}
+}
+
+func TestMigrateAppliesInFilenameOrder(t *testing.T) {
+	s := openTestStore(t)
+	// 0002 depends on 0001; MapFS iteration order is random, so this only
+	// passes if migrations are sorted by name.
+	ordered := fstest.MapFS{
+		"0002_child.sql":  {Data: []byte("CREATE TABLE child (parent_id INTEGER REFERENCES parent(id)); INSERT INTO child VALUES (1);")},
+		"0001_parent.sql": {Data: []byte("CREATE TABLE parent (id INTEGER PRIMARY KEY); INSERT INTO parent VALUES (1);")},
+	}
+	if err := Migrate(context.Background(), s.Writer, ordered); err != nil {
+		t.Fatal(err)
+	}
+}
+
+func TestMigrateUnknownAppliedVersionFails(t *testing.T) {
+	ctx := context.Background()
+	s := openTestStore(t)
+	if err := Migrate(ctx, s.Writer, testMigrations); err != nil {
+		t.Fatal(err)
+	}
+	older := fstest.MapFS{"0001_first.sql": testMigrations["0001_first.sql"]}
+	if err := Migrate(ctx, s.Writer, older); err == nil {
+		t.Fatal("migrating a database newer than the build returned no error")
+	}
+}
+
+func TestMigrateNoMigrationsFails(t *testing.T) {
+	s := openTestStore(t)
+	if err := Migrate(context.Background(), s.Writer, fstest.MapFS{}); err == nil {
+		t.Fatal("empty migrations returned no error")
+	}
+}
+
+func TestMigrateEmbeddedMigrations(t *testing.T) {
+	s := openTestStore(t)
+	if err := Migrate(context.Background(), s.Writer, migrations.FS); err != nil {
+		t.Fatal(err)
+	}
+	if got := versions(migrationRows(t, s)); got != "0000_bootstrap" {
+		t.Errorf("versions = %q, want 0000_bootstrap", got)
+	}
+}
+
+func TestMigrateEmbeddedThenNewMigrationApplies(t *testing.T) {
+	ctx := context.Background()
+	s := openTestStore(t)
+	if err := Migrate(ctx, s.Writer, migrations.FS); err != nil {
+		t.Fatal(err)
+	}
+	bootstrap, err := fs.ReadFile(migrations.FS, "0000_bootstrap.sql")
+	if err != nil {
+		t.Fatal(err)
+	}
+	// A later build adds 0001_init.sql alongside the unchanged bootstrap.
+	upgraded := fstest.MapFS{
+		"0000_bootstrap.sql": {Data: bootstrap},
+		"0001_init.sql":      {Data: []byte("CREATE TABLE later (id INTEGER PRIMARY KEY);")},
+	}
+	if err := Migrate(ctx, s.Writer, upgraded); err != nil {
+		t.Fatal(err)
+	}
+	if got := versions(migrationRows(t, s)); got != "0000_bootstrap,0001_init" {
+		t.Errorf("versions = %q, want 0000_bootstrap,0001_init", got)
+	}
+	if !tableExists(t, s, "later") {
+		t.Error("table from 0001_init was not created")
+	}
+}

internal/store/store.go

@@ -0,0 +1,79 @@
+// Package store owns the MealQuest SQLite database: one writer connection,
+// a read pool, and the embedded schema migrations.
+package store
+
+import (
+	"context"
+	"database/sql"
+	"errors"
+	"fmt"
+	"net/url"
+	"path/filepath"
+
+	_ "modernc.org/sqlite" // registers the "sqlite" database/sql driver
+)
+
+// FileName is the database file inside the data directory.
+const FileName = "mealquest.db"
+
+// readPoolSize is the maximum number of open read connections.
+const readPoolSize = 4
+
+// connPragmas are applied by the driver to every new connection.
+var connPragmas = []string{
+	"busy_timeout(5000)",
+	"journal_mode(WAL)",
+	"foreign_keys(1)",
+}
+
+// Store holds the database handles. All writes go through Writer, which has
+// exactly one connection; Reader is a pool of read-only connections.
+type Store struct {
+	Writer *sql.DB
+	Reader *sql.DB
+}
+
+// Open opens <dataDir>/mealquest.db, creating it if absent, and verifies both
+// handles can connect. It does not apply migrations; call Migrate for that.
+func Open(ctx context.Context, dataDir string) (*Store, error) {
+	path, err := filepath.Abs(filepath.Join(dataDir, FileName))
+	if err != nil {
+		return nil, fmt.Errorf("store: resolve database path: %w", err)
+	}
+
+	// The writer is opened and pinged first so the file exists and is in WAL
+	// mode before any read-only connection touches it.
+	writer, err := openDB(ctx, path, 1, url.Values{"_txlock": {"immediate"}})
+	if err != nil {
+		return nil, fmt.Errorf("store: open writer: %w", err)
+	}
+	reader, err := openDB(ctx, path, readPoolSize, url.Values{"_pragma": {"query_only(1)"}})
+	if err != nil {
+		return nil, errors.Join(fmt.Errorf("store: open reader: %w", err), writer.Close())
+	}
+	return &Store{Writer: writer, Reader: reader}, nil
+}
+
+// Close closes both handles.
+func (s *Store) Close() error {
+	return errors.Join(s.Reader.Close(), s.Writer.Close())
+}
+
+func openDB(ctx context.Context, path string, maxConns int, extra url.Values) (*sql.DB, error) {
+	q := url.Values{"_pragma": connPragmas}
+	for k, vs := range extra {
+		q[k] = append(q[k], vs...)
+	}
+	dsn := (&url.URL{Scheme: "file", Path: path, RawQuery: q.Encode()}).String()
+
+	db, err := sql.Open("sqlite", dsn)
+	if err != nil {
+		return nil, err
+	}
+	db.SetMaxOpenConns(maxConns)
+	db.SetMaxIdleConns(maxConns)
+	if err := db.PingContext(ctx); err != nil {
+		return nil, errors.Join(err, db.Close())
+	}
+	return db, nil
+}

internal/store/store_test.go

@@ -0,0 +1,116 @@
+package store
+
+import (
+	"context"
+	"database/sql"
+	"os"
+	"path/filepath"
+	"testing"
+)
+
+func openTestStore(t *testing.T) *Store {
+	t.Helper()
+	s, err := Open(context.Background(), t.TempDir())
+	if err != nil {
+		t.Fatal(err)
+	}
+	t.Cleanup(func() {
+		if err := s.Close(); err != nil {
+			t.Error(err)
+		}
+	})
+	return s
+}
+
+func TestOpenCreatesDatabaseFile(t *testing.T) {
+	dir := t.TempDir()
+	s, err := Open(context.Background(), dir)
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer s.Close()
+	if _, err := os.Stat(filepath.Join(dir, FileName)); err != nil {
+		t.Fatalf("database file: %v", err)
+	}
+}
+
+func TestOpenMissingDirFails(t *testing.T) {
+	_, err := Open(context.Background(), filepath.Join(t.TempDir(), "missing"))
+	if err == nil {
+		t.Fatal("Open succeeded on a missing data directory")
+	}
+}
+
+func TestPoolSizes(t *testing.T) {
+	s := openTestStore(t)
+	if got := s.Writer.Stats().MaxOpenConnections; got != 1 {
+		t.Errorf("writer max open connections = %d, want 1", got)
+	}
+	if got := s.Reader.Stats().MaxOpenConnections; got != readPoolSize {
+		t.Errorf("reader max open connections = %d, want %d", got, readPoolSize)
+	}
+}
+
+// TestPragmasOnEveryConnection holds every reader connection open at once so
+// each one is a distinct connection, then checks the pragmas on each and on
+// the writer.
+func TestPragmasOnEveryConnection(t *testing.T) {
+	ctx := context.Background()
+	s := openTestStore(t)
+
+	conns := []*sql.Conn{}
+	for range readPoolSize {
+		c, err := s.Reader.Conn(ctx)
+		if err != nil {
+			t.Fatal(err)
+		}
+		defer c.Close()
+		conns = append(conns, c)
+	}
+	w, err := s.Writer.Conn(ctx)
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer w.Close()
+	conns = append(conns, w)
+
+	for i, c := range conns {
+		var fk, busy int
+		var journal string
+		if err := c.QueryRowContext(ctx, "PRAGMA foreign_keys").Scan(&fk); err != nil {
+			t.Fatal(err)
+		}
+		if err := c.QueryRowContext(ctx, "PRAGMA busy_timeout").Scan(&busy); err != nil {
+			t.Fatal(err)
+		}
+		if err := c.QueryRowContext(ctx, "PRAGMA journal_mode").Scan(&journal); err != nil {
+			t.Fatal(err)
+		}
+		if fk != 1 || busy != 5000 || journal != "wal" {
+			t.Errorf("connection %d: foreign_keys=%d busy_timeout=%d journal_mode=%q, want 1, 5000, wal", i, fk, busy, journal)
+		}
+	}
+}
+
+func TestReaderIsReadOnly(t *testing.T) {
+	s := openTestStore(t)
+	if _, err := s.Reader.Exec("CREATE TABLE t (id INTEGER)"); err == nil {
+		t.Fatal("write through the reader succeeded")
+	}
+	if _, err := s.Writer.Exec("CREATE TABLE t (id INTEGER)"); err != nil {
+		t.Fatalf("write through the writer: %v", err)
+	}
+}
+
+func TestForeignKeysEnforced(t *testing.T) {
+	s := openTestStore(t)
+	if _, err := s.Writer.Exec(`
+		CREATE TABLE parent (id INTEGER PRIMARY KEY);
+		CREATE TABLE child (parent_id INTEGER NOT NULL REFERENCES parent(id));
+	`); err != nil {
+		t.Fatal(err)
+	}
+	if _, err := s.Writer.Exec("INSERT INTO child (parent_id) VALUES (42)"); err == nil {
+		t.Fatal("insert violating a foreign key succeeded")
+	}
+}