feat(db): SQLite 连接与迁移框架(embed 迁移/事务校验/PG 预留 M4)

This commit is contained in:
Aska committed 2026-09-06 04:26:02 +08:00
1 parent dd16176bb7
commit 34cac241c1
6 files changed
+726

No files matched your search

+130
View File
@@ -0,0 +1,130 @@
// Package db manages the SQLite connection and schema migrations for Cyrene v2.
//
// SQLite is the default embedded store. The pool is limited to a single open
// connection (SetMaxOpenConns(1) plus SetMaxIdleConns(1)) so all writes are
// serialized and can never contend on SQLite's single-writer file lock. A
// busy_timeout of 5000ms and WAL journal mode for file-backed databases
// further reduce lock contention.
//
// Migrations live in internal/db/migrations/*.sql, are embedded into the
// binary, and are applied in timestamp order inside transactions, tracked in a
// schema_migrations table. See MigrateUp and MigrateDown.
//
// PostgreSQL + pgvector is an optional enhancement that is intentionally out of
// scope for milestone M0. It will be added when the runtime reaches milestone M4
// (see DevDocs/开发时间线.md); until then the connection is SQLite only.
package db
import (
"context"
"database/sql"
"errors"
"fmt"
"io"
"log/slog"
"strings"
_ "modernc.org/sqlite" // pure-Go SQLite driver, registers driver name "sqlite"
)
// Package-level sentinel errors. Each is wrapped with %w at the call site so
// callers can use errors.Is.
var (
// ErrInvalidPath is returned when the database path is empty or blank.
ErrInvalidPath = errors.New("db: invalid database path")
// ErrNoEmbedded is returned when no migration files are embedded.
ErrNoEmbedded = errors.New("db: no migration files found in embed")
// ErrDownMissing is returned when a migration has an up file but no matching
// down file.
ErrDownMissing = errors.New("db: down migration file missing")
// ErrInvalidSteps is returned when MigrateDown receives a steps count < 1.
ErrInvalidSteps = errors.New("db: migration steps must be >= 1")
)
// Open opens (or creates) the SQLite database at path and returns a *sql.DB.
//
// path may be a filesystem path (relative or absolute) or ":memory:" for an
// isolated in-memory database, which tests use. Under ":memory:" each returned
// handle owns exactly one private, non-shared database, so parallel tests are
// isolated by construction.
//
// The pool is configured for write serialization: SetMaxOpenConns(1) and
// SetMaxIdleConns(1) prevent multiple writers from contending on the single-
// writer lock. busy_timeout is set to 5000ms; for file-backed databases the
// journal mode is switched to WAL. ":memory:" keeps the default memory journal,
// because WAL has no meaning there.
//
// log is a *slog.Logger; a nil logger is replaced with a discard logger so the
// function can always log safely. ctx is used for the connection ping and the
// WAL pragma. The caller must Close the returned DB when done.
func Open(ctx context.Context, path string, log *slog.Logger) (*sql.DB, error) {
if strings.TrimSpace(path) == "" {
return nil, fmt.Errorf("db: open: %w", ErrInvalidPath)
}
log = discardLogger(log)
db, err := sql.Open("sqlite", buildDSN(path))
if err != nil {
return nil, fmt.Errorf("db: open %q: %w", path, err)
}
// Write serialization: a single connection avoids writer-lock contention.
db.SetMaxOpenConns(1)
db.SetMaxIdleConns(1)
if !isMemory(path) {
if _, err := db.ExecContext(ctx, "PRAGMA journal_mode=WAL"); err != nil {
_ = db.Close()
return nil, fmt.Errorf("db: enable WAL on %q: %w", path, err)
}
}
if err := db.PingContext(ctx); err != nil {
_ = db.Close()
return nil, fmt.Errorf("db: ping %q: %w", path, err)
}
log.Info("sqlite opened", "path", path, "max_open_conns", 1)
return db, nil
}
// buildDSN converts a user-supplied path into a modernc.org/sqlite DSN.
//
// ":memory:" is returned as-is: the driver then gives that handle a private,
// non-shared in-memory database, which with SetMaxOpenConns(1) is stable for
// the lifetime of the handle. Any other path becomes a file: URI so the
// busy_timeout pragma is honored; pre-built file: URIs are preserved and only
// have the pragma appended.
func buildDSN(path string) string {
if path == ":memory:" || strings.Contains(path, "mode=memory") {
return path
}
dsn := path
if !strings.HasPrefix(path, "file:") {
dsn = "file:" + path
}
sep := "?"
if strings.Contains(dsn, "?") {
sep = "&"
}
return dsn + sep + "_pragma=busy_timeout(5000)"
}
// isMemory reports whether path refers to an in-memory SQLite database, in
// which case the WAL journal mode is skipped.
func isMemory(path string) bool {
return strings.Contains(strings.ToLower(path), ":memory:") ||
strings.Contains(strings.ToLower(path), "mode=memory")
}
// discardLogger returns log if non-nil, otherwise a logger that writes nowhere,
// so callers can always log safely.
func discardLogger(log *slog.Logger) *slog.Logger {
if log != nil {
return log
}
return slog.New(slog.NewTextHandler(io.Discard, nil))
}
+136
View File
@@ -0,0 +1,136 @@
package db
import (
"context"
"errors"
"io"
"log/slog"
"path/filepath"
"testing"
)
// testLogger returns a logger that writes nowhere, keeping test output clean.
func testLogger() *slog.Logger {
return slog.New(slog.NewTextHandler(io.Discard, nil))
}
func TestOpen_InMemory_正常(t *testing.T) {
t.Parallel()
ctx := context.Background()
db, err := Open(ctx, ":memory:", testLogger())
if err != nil {
t.Fatalf("Open(:memory:) error: %v", err)
}
defer db.Close()
if err := db.PingContext(ctx); err != nil {
t.Fatalf("ping in-memory db: %v", err)
}
var n int
if err := db.QueryRowContext(ctx, "SELECT count(*) FROM sqlite_master WHERE type='table'").Scan(&n); err != nil {
t.Fatalf("query sqlite_master: %v", err)
}
if n != 0 {
t.Fatalf("expected no tables before migrations, got %d", n)
}
}
func TestOpen_InMemory_NilLogger_正常(t *testing.T) {
t.Parallel()
ctx := context.Background()
// A nil logger must not panic and must produce a usable connection.
db, err := Open(ctx, ":memory:", nil)
if err != nil {
t.Fatalf("Open with nil logger error: %v", err)
}
defer db.Close()
if err := db.PingContext(ctx); err != nil {
t.Fatalf("ping with nil logger: %v", err)
}
}
func TestOpen_InMemory_Isolation_并发(t *testing.T) {
t.Parallel()
ctx := context.Background()
db1, err := Open(ctx, ":memory:", testLogger())
if err != nil {
t.Fatalf("open db1: %v", err)
}
defer db1.Close()
db2, err := Open(ctx, ":memory:", testLogger())
if err != nil {
t.Fatalf("open db2: %v", err)
}
defer db2.Close()
if _, err := db1.ExecContext(ctx, "CREATE TABLE t1(x)"); err != nil {
t.Fatalf("create table in db1: %v", err)
}
var n int
if err := db2.QueryRowContext(ctx, "SELECT count(*) FROM sqlite_master WHERE name='t1'").Scan(&n); err != nil {
t.Fatalf("query db2: %v", err)
}
if n != 0 {
t.Fatalf("db2 must not see db1's table (isolation broken), count=%d", n)
}
}
func TestOpen_File_正常(t *testing.T) {
t.Parallel()
ctx := context.Background()
path := filepath.Join(t.TempDir(), "cyrene.db")
db, err := Open(ctx, path, testLogger())
if err != nil {
t.Fatalf("Open(file) error: %v", err)
}
defer db.Close()
if err := db.PingContext(ctx); err != nil {
t.Fatalf("ping file db: %v", err)
}
// Persistence across a reopen with the same DSN.
db.ExecContext(ctx, "CREATE TABLE p(x)")
db.Close()
db2, err := Open(ctx, path, testLogger())
if err != nil {
t.Fatalf("reopen(file) error: %v", err)
}
defer db2.Close()
var n int
if err := db2.QueryRowContext(ctx, "SELECT count(*) FROM sqlite_master WHERE name='p'").Scan(&n); err != nil {
t.Fatalf("query reopened db: %v", err)
}
if n != 1 {
t.Fatalf("expected table p to persist across reopen, count=%d", n)
}
}
func TestOpen_BlankPath_异常(t *testing.T) {
t.Parallel()
ctx := context.Background()
for _, path := range []string{"", " "} {
if _, err := Open(ctx, path, testLogger()); !errors.Is(err, ErrInvalidPath) {
t.Fatalf("Open(%q) err = %v, want wrap of ErrInvalidPath", path, err)
}
}
}
func TestOpen_MissingParentDir_异常(t *testing.T) {
t.Parallel()
ctx := context.Background()
// Parent directory does not exist, so the connection cannot be established.
path := filepath.Join(t.TempDir(), "does-not-exist", "cyrene.db")
if _, err := Open(ctx, path, testLogger()); err == nil {
t.Fatalf("Open(%q) expected error for missing parent dir, got nil", path)
}
}
+219
View File
@@ -0,0 +1,219 @@
package db
import (
"context"
"database/sql"
"embed"
"fmt"
"log/slog"
"sort"
"strings"
)
//go:embed migrations/*.sql
var migrationsFS embed.FS
// migration is a single schema change with its forward (up) and reverse (down)
// SQL. id is the "{timestamp}_{desc}" derived from the file name, which is also
// the unique key recorded in schema_migrations.
type migration struct {
id string
upSQL string
downSQL string
}
// createMigrationsTable is a parameter-free DDL statement (no placeholders). It
// is executed with ExecContext to create the bookkeeping table on first use.
const createMigrationsTable = `
CREATE TABLE IF NOT EXISTS schema_migrations (
id TEXT PRIMARY KEY,
applied_at TEXT NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%fZ','now'))
)`
// MigrateUp applies every embedded migration that has not yet been recorded in
// schema_migrations. Migrations run in ascending timestamp order, each inside its
// own transaction; the schema_migrations row is inserted in that same
// transaction, so a failing migration leaves no half-applied state. Already
// applied migrations are skipped, making MigrateUp idempotent.
//
// The entire content of a migration's .up.sql is executed as one ExecContext
// inside the transaction. Keep migrations small and use several files for a
// larger schema change rather than packing many statements into one file.
func MigrateUp(ctx context.Context, db *sql.DB, log *slog.Logger) error {
if err := ensureMigrationTable(ctx, db); err != nil {
return err
}
applied, err := appliedMigrationIDs(ctx, db)
if err != nil {
return err
}
migs, err := loadMigrations()
if err != nil {
return err
}
for _, m := range migs {
if applied[m.id] {
continue
}
if err := applyUp(ctx, db, m, discardLogger(log)); err != nil {
return err
}
}
return nil
}
// MigrateDown rolls back the most recent steps migrations, in descending
// timestamp order. Each rollback runs the migration's .down.sql and deletes its
// schema_migrations row in the same transaction. steps must be >= 1; if steps
// exceeds the number of applied migrations, all applied migrations are rolled
// back.
func MigrateDown(ctx context.Context, db *sql.DB, steps int, log *slog.Logger) error {
if steps < 1 {
return fmt.Errorf("db: down: %w", ErrInvalidSteps)
}
log = discardLogger(log)
if err := ensureMigrationTable(ctx, db); err != nil {
return err
}
applied, err := appliedMigrationIDs(ctx, db)
if err != nil {
return err
}
migs, err := loadMigrations()
if err != nil {
return err
}
toRollback := make([]migration, 0, len(applied))
for _, m := range migs {
if applied[m.id] {
toRollback = append(toRollback, m)
}
}
// Reverse to descending order, then keep only the most recent `steps`.
for i, j := 0, len(toRollback)-1; i < j; i, j = i+1, j-1 {
toRollback[i], toRollback[j] = toRollback[j], toRollback[i]
}
if len(toRollback) > steps {
toRollback = toRollback[:steps]
}
for _, m := range toRollback {
if err := applyDown(ctx, db, m, log); err != nil {
return err
}
}
return nil
}
// ensureMigrationTable creates the schema_migrations bookkeeping table if it does
// not already exist.
func ensureMigrationTable(ctx context.Context, db *sql.DB) error {
if _, err := db.ExecContext(ctx, createMigrationsTable); err != nil {
return fmt.Errorf("db: ensure schema_migrations: %w", err)
}
return nil
}
// appliedMigrationIDs returns the set of migration ids already recorded in
// schema_migrations.
func appliedMigrationIDs(ctx context.Context, db *sql.DB) (map[string]bool, error) {
rows, err := db.QueryContext(ctx, "SELECT id FROM schema_migrations")
if err != nil {
return nil, fmt.Errorf("db: list applied migrations: %w", err)
}
defer rows.Close()
applied := make(map[string]bool)
for rows.Next() {
var id string
if err := rows.Scan(&id); err != nil {
return nil, fmt.Errorf("db: scan migration id: %w", err)
}
applied[id] = true
}
if err := rows.Err(); err != nil {
return nil, fmt.Errorf("db: iterate applied migrations: %w", err)
}
return applied, nil
}
// applyUp runs a single migration's up SQL in a transaction and records it.
func applyUp(ctx context.Context, db *sql.DB, m migration, log *slog.Logger) error {
tx, err := db.BeginTx(ctx, nil)
if err != nil {
return fmt.Errorf("db: begin up %s: %w", m.id, err)
}
defer tx.Rollback()
if _, err := tx.ExecContext(ctx, m.upSQL); err != nil {
return fmt.Errorf("db: apply up %s: %w", m.id, err)
}
if _, err := tx.ExecContext(ctx, "INSERT INTO schema_migrations (id) VALUES (?)", m.id); err != nil {
return fmt.Errorf("db: record up %s: %w", m.id, err)
}
if err := tx.Commit(); err != nil {
return fmt.Errorf("db: commit up %s: %w", m.id, err)
}
log.Info("migration applied", "migration", m.id)
return nil
}
// applyDown runs a single migration's down SQL in a transaction and unrecords it.
func applyDown(ctx context.Context, db *sql.DB, m migration, log *slog.Logger) error {
tx, err := db.BeginTx(ctx, nil)
if err != nil {
return fmt.Errorf("db: begin down %s: %w", m.id, err)
}
defer tx.Rollback()
if _, err := tx.ExecContext(ctx, m.downSQL); err != nil {
return fmt.Errorf("db: apply down %s: %w", m.id, err)
}
if _, err := tx.ExecContext(ctx, "DELETE FROM schema_migrations WHERE id = ?", m.id); err != nil {
return fmt.Errorf("db: unrecord down %s: %w", m.id, err)
}
if err := tx.Commit(); err != nil {
return fmt.Errorf("db: commit down %s: %w", m.id, err)
}
log.Info("migration rolled back", "migration", m.id)
return nil
}
// loadMigrations reads all embedded .up.sql files, pairs each with its matching
// .down.sql, and returns them sorted ascending by id (which places them in
// timestamp order). A .up.sql without a matching .down.sql is an error.
func loadMigrations() ([]migration, error) {
entries, err := migrationsFS.ReadDir("migrations")
if err != nil {
return nil, fmt.Errorf("db: read embedded migrations: %w", err)
}
migs := make([]migration, 0, len(entries)/2)
for _, e := range entries {
if e.IsDir() || !strings.HasSuffix(e.Name(), ".up.sql") {
continue
}
id := strings.TrimSuffix(e.Name(), ".up.sql")
upSQL, err := migrationsFS.ReadFile("migrations/" + e.Name())
if err != nil {
return nil, fmt.Errorf("db: read %s: %w", e.Name(), err)
}
downName := id + ".down.sql"
downSQL, err := migrationsFS.ReadFile("migrations/" + downName)
if err != nil {
return nil, fmt.Errorf("db: %w: %s", ErrDownMissing, downName)
}
migs = append(migs, migration{id: id, upSQL: string(upSQL), downSQL: string(downSQL)})
}
if len(migs) == 0 {
return nil, ErrNoEmbedded
}
sort.Slice(migs, func(i, j int) bool { return migs[i].id < migs[j].id })
return migs, nil
}
+223
View File
@@ -0,0 +1,223 @@
package db
import (
"context"
"database/sql"
"errors"
"testing"
)
// openMemory opens an isolated in-memory DB for a test.
func openMemory(t *testing.T) *sql.DB {
t.Helper()
ctx := context.Background()
db, err := Open(ctx, ":memory:", testLogger())
if err != nil {
t.Fatalf("Open(:memory:) failed: %v", err)
}
t.Cleanup(func() { db.Close() })
return db
}
// registeredMigrations returns the ids currently recorded in schema_migrations.
func registeredMigrations(t *testing.T, ctx context.Context, db *sql.DB) []string {
t.Helper()
rows, err := db.QueryContext(ctx, "SELECT id FROM schema_migrations")
if err != nil {
t.Fatalf("query schema_migrations: %v", err)
}
defer rows.Close()
var ids []string
for rows.Next() {
var id string
if err := rows.Scan(&id); err != nil {
t.Fatalf("scan migration id: %v", err)
}
ids = append(ids, id)
}
if err := rows.Err(); err != nil {
t.Fatalf("iterate schema_migrations: %v", err)
}
return ids
}
// appMetaExists reports whether the app_meta table is present.
func appMetaExists(t *testing.T, ctx context.Context, db *sql.DB) bool {
t.Helper()
var n int
if err := db.QueryRowContext(ctx, "SELECT count(*) FROM sqlite_master WHERE type='table' AND name='app_meta'").Scan(&n); err != nil {
t.Fatalf("check app_meta presence: %v", err)
}
return n == 1
}
func TestMigrateUp_FirstApply_正常(t *testing.T) {
t.Parallel()
ctx := context.Background()
db := openMemory(t)
if err := MigrateUp(ctx, db, testLogger()); err != nil {
t.Fatalf("MigrateUp first call: %v", err)
}
ids := registeredMigrations(t, ctx, db)
if len(ids) != 1 || ids[0] != "0001_init" {
t.Fatalf("schema_migrations = %v, want exactly [0001_init]", ids)
}
if !appMetaExists(t, ctx, db) {
t.Fatal("app_meta table should exist after MigrateUp")
}
// applied_at must be recorded on the single row.
var appliedAt string
if err := db.QueryRowContext(ctx, "SELECT applied_at FROM schema_migrations WHERE id = ?", "0001_init").Scan(&appliedAt); err != nil {
t.Fatalf("read applied_at: %v", err)
}
if appliedAt == "" {
t.Fatal("applied_at should not be empty")
}
}
func TestMigrateUp_Idempotent_正常(t *testing.T) {
t.Parallel()
ctx := context.Background()
db := openMemory(t)
if err := MigrateUp(ctx, db, testLogger()); err != nil {
t.Fatalf("MigrateUp first call: %v", err)
}
// Second call must not re-execute or duplicate the record.
if err := MigrateUp(ctx, db, testLogger()); err != nil {
t.Fatalf("MigrateUp second call: %v", err)
}
ids := registeredMigrations(t, ctx, db)
if len(ids) != 1 {
t.Fatalf("schema_migrations rows = %d, want 1 (idempotent), ids=%v", len(ids), ids)
}
if !appMetaExists(t, ctx, db) {
t.Fatal("app_meta should still exist after idempotent re-run")
}
}
func TestMigrateDown_ThenReUp_正常(t *testing.T) {
t.Parallel()
ctx := context.Background()
db := openMemory(t)
if err := MigrateUp(ctx, db, testLogger()); err != nil {
t.Fatalf("MigrateUp: %v", err)
}
// Roll back the single applied migration.
if err := MigrateDown(ctx, db, 1, testLogger()); err != nil {
t.Fatalf("MigrateDown: %v", err)
}
if appMetaExists(t, ctx, db) {
t.Fatal("app_meta should be dropped after MigrateDown")
}
if ids := registeredMigrations(t, ctx, db); len(ids) != 0 {
t.Fatalf("schema_migrations should be empty after rollback, got %v", ids)
}
// Re-applying must work end to end again.
if err := MigrateUp(ctx, db, testLogger()); err != nil {
t.Fatalf("re-MigrateUp: %v", err)
}
if !appMetaExists(t, ctx, db) {
t.Fatal("app_meta should exist again after re-MigrateUp")
}
if ids := registeredMigrations(t, ctx, db); len(ids) != 1 || ids[0] != "0001_init" {
t.Fatalf("schema_migrations = %v, want exactly [0001_init]", ids)
}
}
func TestMigrateDown_ZeroSteps_异常(t *testing.T) {
t.Parallel()
ctx := context.Background()
db := openMemory(t)
if err := MigrateUp(ctx, db, testLogger()); err != nil {
t.Fatalf("MigrateUp: %v", err)
}
for _, steps := range []int{0, -1} {
if err := MigrateDown(ctx, db, steps, testLogger()); !errors.Is(err, ErrInvalidSteps) {
t.Fatalf("MigrateDown(%d) err = %v, want wrap of ErrInvalidSteps", steps, err)
}
}
}
func TestMigrateDown_StepsBeyondApplied_边界(t *testing.T) {
t.Parallel()
ctx := context.Background()
db := openMemory(t)
if err := MigrateUp(ctx, db, testLogger()); err != nil {
t.Fatalf("MigrateUp: %v", err)
}
// More steps than applied must roll everything back without error.
if err := MigrateDown(ctx, db, 10, testLogger()); err != nil {
t.Fatalf("MigrateDown(10): %v", err)
}
if appMetaExists(t, ctx, db) {
t.Fatal("app_meta should be dropped when rolling back all")
}
if ids := registeredMigrations(t, ctx, db); len(ids) != 0 {
t.Fatalf("schema_migrations should be empty, got %v", ids)
}
}
func TestAppMeta_ReadWrite_正常(t *testing.T) {
t.Parallel()
ctx := context.Background()
db := openMemory(t)
if err := MigrateUp(ctx, db, testLogger()); err != nil {
t.Fatalf("MigrateUp: %v", err)
}
// Parameterized insert.
if _, err := db.ExecContext(ctx,
"INSERT INTO app_meta (key, value) VALUES (?, ?)", "schema_version", "1"); err != nil {
t.Fatalf("insert app_meta: %v", err)
}
// Parameterized read.
var value string
if err := db.QueryRowContext(ctx,
"SELECT value FROM app_meta WHERE key = ?", "schema_version").Scan(&value); err != nil {
t.Fatalf("read app_meta: %v", err)
}
if value != "1" {
t.Fatalf("value = %q, want %q", value, "1")
}
// Parameterized update.
if _, err := db.ExecContext(ctx,
"UPDATE app_meta SET value = ? WHERE key = ?", "2", "schema_version"); err != nil {
t.Fatalf("update app_meta: %v", err)
}
if err := db.QueryRowContext(ctx,
"SELECT value FROM app_meta WHERE key = ?", "schema_version").Scan(&value); err != nil {
t.Fatalf("read updated app_meta: %v", err)
}
if value != "2" {
t.Fatalf("value after update = %q, want %q", value, "2")
}
// Parameterized delete.
if _, err := db.ExecContext(ctx,
"DELETE FROM app_meta WHERE key = ?", "schema_version"); err != nil {
t.Fatalf("delete app_meta: %v", err)
}
var n int
if err := db.QueryRowContext(ctx,
"SELECT count(*) FROM app_meta WHERE key = ?", "schema_version").Scan(&n); err != nil {
t.Fatalf("count app_meta after delete: %v", err)
}
if n != 0 {
t.Fatalf("expected 0 rows after delete, got %d", n)
}
}
@@ -0,0 +1,2 @@
-- 0001_init: drops the placeholder table created by the up migration.
DROP TABLE IF EXISTS app_meta;
+16
View File
@@ -0,0 +1,16 @@
-- 0001_init: M0 migration-framework placeholder.
--
-- Creates a small, harmless key/value table so the migration plumbing
-- (embed -> load -> apply in a transaction -> record in schema_migrations)
-- can be exercised end to end. Real tables (messages, memories, sessions,
-- personas, ...) are added as their own timestamped migrations in later
-- milestones; this placeholder is NOT a domain table.
--
-- Note: key is the primary key on purpose — this is a key/value metadata
-- table, so it intentionally deviates from the "id primary key" convention
-- that applies to domain tables.
CREATE TABLE IF NOT EXISTS app_meta (
key TEXT PRIMARY KEY,
value TEXT NOT NULL,
updated_at TEXT NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%fZ','now'))
);