Compare commits
4 Commits
feat/upaas
...
0c8de65be8
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
0c8de65be8 | ||
|
|
8b6aa74951 | ||
|
|
3d92f2625b | ||
| 9c29cb57df |
@@ -21,6 +21,10 @@ import (
|
||||
//go:embed schema/*.sql
|
||||
var schemaFS embed.FS
|
||||
|
||||
// bootstrapVersion is the migration that creates the schema_migrations
|
||||
// table itself. It is applied before the normal migration loop.
|
||||
const bootstrapVersion = "000"
|
||||
|
||||
// Params defines dependencies for Database.
|
||||
type Params struct {
|
||||
fx.In
|
||||
@@ -35,6 +39,41 @@ type Database struct {
|
||||
config *config.Config
|
||||
}
|
||||
|
||||
// ParseMigrationVersion extracts the numeric version prefix from a migration
|
||||
// filename. Filenames must follow the pattern "<version>.sql" or
|
||||
// "<version>_<description>.sql", where version is a zero-padded numeric
|
||||
// string (e.g. "001", "002"). Returns the version string and an error if
|
||||
// the filename does not match the expected pattern.
|
||||
func ParseMigrationVersion(filename string) (string, error) {
|
||||
name := strings.TrimSuffix(filename, filepath.Ext(filename))
|
||||
if name == "" {
|
||||
return "", fmt.Errorf("invalid migration filename %q: empty name", filename)
|
||||
}
|
||||
|
||||
// Split on underscore to separate version from description.
|
||||
// If there's no underscore, the entire stem is the version.
|
||||
version := name
|
||||
if idx := strings.IndexByte(name, '_'); idx >= 0 {
|
||||
version = name[:idx]
|
||||
}
|
||||
|
||||
if version == "" {
|
||||
return "", fmt.Errorf("invalid migration filename %q: empty version prefix", filename)
|
||||
}
|
||||
|
||||
// Validate the version is purely numeric.
|
||||
for _, ch := range version {
|
||||
if ch < '0' || ch > '9' {
|
||||
return "", fmt.Errorf(
|
||||
"invalid migration filename %q: version %q contains non-numeric character %q",
|
||||
filename, version, string(ch),
|
||||
)
|
||||
}
|
||||
}
|
||||
|
||||
return version, nil
|
||||
}
|
||||
|
||||
// New creates a new Database instance.
|
||||
func New(lc fx.Lifecycle, params Params) (*Database, error) {
|
||||
s := &Database{
|
||||
@@ -84,127 +123,86 @@ func (s *Database) connect(ctx context.Context) error {
|
||||
s.db = db
|
||||
s.log.Info("database connected")
|
||||
|
||||
return s.runMigrations(ctx)
|
||||
return ApplyMigrations(ctx, s.db, s.log)
|
||||
}
|
||||
|
||||
func (s *Database) runMigrations(ctx context.Context) error {
|
||||
// Create migrations tracking table
|
||||
_, err := s.db.ExecContext(ctx, `
|
||||
CREATE TABLE IF NOT EXISTS schema_migrations (
|
||||
version TEXT PRIMARY KEY,
|
||||
applied_at DATETIME DEFAULT CURRENT_TIMESTAMP
|
||||
)
|
||||
`)
|
||||
if err != nil {
|
||||
return fmt.Errorf("failed to create migrations table: %w", err)
|
||||
}
|
||||
|
||||
// Get list of migration files
|
||||
// collectMigrations reads the embedded schema directory and returns
|
||||
// migration filenames sorted lexicographically.
|
||||
func collectMigrations() ([]string, error) {
|
||||
entries, err := schemaFS.ReadDir("schema")
|
||||
if err != nil {
|
||||
return fmt.Errorf("failed to read schema directory: %w", err)
|
||||
return nil, fmt.Errorf("failed to read schema directory: %w", err)
|
||||
}
|
||||
|
||||
// Sort migration files by name (001.sql, 002.sql, etc.)
|
||||
var migrations []string
|
||||
|
||||
for _, entry := range entries {
|
||||
if !entry.IsDir() && strings.HasSuffix(entry.Name(), ".sql") {
|
||||
migrations = append(migrations, entry.Name())
|
||||
}
|
||||
}
|
||||
|
||||
sort.Strings(migrations)
|
||||
|
||||
// Apply each migration that hasn't been applied yet
|
||||
for _, migration := range migrations {
|
||||
version := strings.TrimSuffix(migration, filepath.Ext(migration))
|
||||
return migrations, nil
|
||||
}
|
||||
|
||||
// Check if already applied
|
||||
var count int
|
||||
err := s.db.QueryRowContext(ctx,
|
||||
"SELECT COUNT(*) FROM schema_migrations WHERE version = ?",
|
||||
version,
|
||||
).Scan(&count)
|
||||
if err != nil {
|
||||
return fmt.Errorf("failed to check migration status: %w", err)
|
||||
}
|
||||
// bootstrapMigrationsTable ensures the schema_migrations table exists
|
||||
// by applying 000.sql if the table is missing.
|
||||
func bootstrapMigrationsTable(ctx context.Context, db *sql.DB, log *slog.Logger) error {
|
||||
var tableExists int
|
||||
|
||||
if count > 0 {
|
||||
s.log.Debug("migration already applied", "version", version)
|
||||
err := db.QueryRowContext(ctx,
|
||||
"SELECT COUNT(*) FROM sqlite_master WHERE type='table' AND name='schema_migrations'",
|
||||
).Scan(&tableExists)
|
||||
if err != nil {
|
||||
return fmt.Errorf("failed to check for migrations table: %w", err)
|
||||
}
|
||||
|
||||
continue
|
||||
}
|
||||
if tableExists > 0 {
|
||||
return nil
|
||||
}
|
||||
|
||||
// Read and apply migration
|
||||
content, err := schemaFS.ReadFile(filepath.Join("schema", migration))
|
||||
if err != nil {
|
||||
return fmt.Errorf("failed to read migration %s: %w", migration, err)
|
||||
}
|
||||
content, err := schemaFS.ReadFile("schema/000.sql")
|
||||
if err != nil {
|
||||
return fmt.Errorf("failed to read bootstrap migration 000.sql: %w", err)
|
||||
}
|
||||
|
||||
s.log.Info("applying migration", "version", version)
|
||||
if log != nil {
|
||||
log.Info("applying bootstrap migration", "version", bootstrapVersion)
|
||||
}
|
||||
|
||||
_, err = s.db.ExecContext(ctx, string(content))
|
||||
if err != nil {
|
||||
return fmt.Errorf("failed to apply migration %s: %w", migration, err)
|
||||
}
|
||||
|
||||
// Record migration as applied
|
||||
_, err = s.db.ExecContext(ctx,
|
||||
"INSERT INTO schema_migrations (version) VALUES (?)",
|
||||
version,
|
||||
)
|
||||
if err != nil {
|
||||
return fmt.Errorf("failed to record migration %s: %w", migration, err)
|
||||
}
|
||||
|
||||
s.log.Info("migration applied successfully", "version", version)
|
||||
_, err = db.ExecContext(ctx, string(content))
|
||||
if err != nil {
|
||||
return fmt.Errorf("failed to apply bootstrap migration: %w", err)
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
// DB returns the underlying sql.DB.
|
||||
func (s *Database) DB() *sql.DB {
|
||||
return s.db
|
||||
}
|
||||
// ApplyMigrations applies all pending migrations to db. An optional logger
|
||||
// may be provided for informational output; pass nil for silent operation.
|
||||
// This is exported so tests can apply the real schema without the full fx
|
||||
// lifecycle.
|
||||
func ApplyMigrations(ctx context.Context, db *sql.DB, log *slog.Logger) error {
|
||||
if err := bootstrapMigrationsTable(ctx, db, log); err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
// ApplyMigrations applies all migrations to the given database.
|
||||
// This is useful for testing where you want to use the real schema
|
||||
// without the full fx lifecycle.
|
||||
func ApplyMigrations(db *sql.DB) error {
|
||||
ctx := context.Background()
|
||||
|
||||
// Create migrations tracking table
|
||||
_, err := db.ExecContext(ctx, `
|
||||
CREATE TABLE IF NOT EXISTS schema_migrations (
|
||||
version TEXT PRIMARY KEY,
|
||||
applied_at DATETIME DEFAULT CURRENT_TIMESTAMP
|
||||
)
|
||||
`)
|
||||
migrations, err := collectMigrations()
|
||||
if err != nil {
|
||||
return fmt.Errorf("failed to create migrations table: %w", err)
|
||||
return err
|
||||
}
|
||||
|
||||
// Get list of migration files
|
||||
entries, err := schemaFS.ReadDir("schema")
|
||||
if err != nil {
|
||||
return fmt.Errorf("failed to read schema directory: %w", err)
|
||||
}
|
||||
|
||||
// Sort migration files by name (001.sql, 002.sql, etc.)
|
||||
var migrations []string
|
||||
for _, entry := range entries {
|
||||
if !entry.IsDir() && strings.HasSuffix(entry.Name(), ".sql") {
|
||||
migrations = append(migrations, entry.Name())
|
||||
}
|
||||
}
|
||||
sort.Strings(migrations)
|
||||
|
||||
// Apply each migration that hasn't been applied yet
|
||||
for _, migration := range migrations {
|
||||
version := strings.TrimSuffix(migration, filepath.Ext(migration))
|
||||
version, parseErr := ParseMigrationVersion(migration)
|
||||
if parseErr != nil {
|
||||
return parseErr
|
||||
}
|
||||
|
||||
// Check if already applied
|
||||
// Check if already applied.
|
||||
var count int
|
||||
|
||||
err := db.QueryRowContext(ctx,
|
||||
"SELECT COUNT(*) FROM schema_migrations WHERE version = ?",
|
||||
version,
|
||||
@@ -214,29 +212,46 @@ func ApplyMigrations(db *sql.DB) error {
|
||||
}
|
||||
|
||||
if count > 0 {
|
||||
if log != nil {
|
||||
log.Debug("migration already applied", "version", version)
|
||||
}
|
||||
|
||||
continue
|
||||
}
|
||||
|
||||
// Read and apply migration
|
||||
content, err := schemaFS.ReadFile(filepath.Join("schema", migration))
|
||||
if err != nil {
|
||||
return fmt.Errorf("failed to read migration %s: %w", migration, err)
|
||||
// Read and apply migration.
|
||||
content, readErr := schemaFS.ReadFile(filepath.Join("schema", migration))
|
||||
if readErr != nil {
|
||||
return fmt.Errorf("failed to read migration %s: %w", migration, readErr)
|
||||
}
|
||||
|
||||
_, err = db.ExecContext(ctx, string(content))
|
||||
if err != nil {
|
||||
return fmt.Errorf("failed to apply migration %s: %w", migration, err)
|
||||
if log != nil {
|
||||
log.Info("applying migration", "version", version)
|
||||
}
|
||||
|
||||
// Record migration as applied
|
||||
_, err = db.ExecContext(ctx,
|
||||
_, execErr := db.ExecContext(ctx, string(content))
|
||||
if execErr != nil {
|
||||
return fmt.Errorf("failed to apply migration %s: %w", migration, execErr)
|
||||
}
|
||||
|
||||
// Record migration as applied.
|
||||
_, recErr := db.ExecContext(ctx,
|
||||
"INSERT INTO schema_migrations (version) VALUES (?)",
|
||||
version,
|
||||
)
|
||||
if err != nil {
|
||||
return fmt.Errorf("failed to record migration %s: %w", migration, err)
|
||||
if recErr != nil {
|
||||
return fmt.Errorf("failed to record migration %s: %w", migration, recErr)
|
||||
}
|
||||
|
||||
if log != nil {
|
||||
log.Info("migration applied successfully", "version", version)
|
||||
}
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
// DB returns the underlying sql.DB.
|
||||
func (s *Database) DB() *sql.DB {
|
||||
return s.db
|
||||
}
|
||||
|
||||
224
internal/database/database_test.go
Normal file
224
internal/database/database_test.go
Normal file
@@ -0,0 +1,224 @@
|
||||
package database
|
||||
|
||||
import (
|
||||
"context"
|
||||
"database/sql"
|
||||
"testing"
|
||||
|
||||
_ "modernc.org/sqlite" // SQLite driver registration
|
||||
)
|
||||
|
||||
// openTestDB returns a fresh in-memory SQLite database.
|
||||
func openTestDB(t *testing.T) *sql.DB {
|
||||
t.Helper()
|
||||
|
||||
db, err := sql.Open("sqlite", ":memory:")
|
||||
if err != nil {
|
||||
t.Fatalf("failed to open test db: %v", err)
|
||||
}
|
||||
|
||||
t.Cleanup(func() { db.Close() })
|
||||
|
||||
return db
|
||||
}
|
||||
|
||||
func TestParseMigrationVersion(t *testing.T) {
|
||||
tests := []struct {
|
||||
name string
|
||||
filename string
|
||||
want string
|
||||
wantErr bool
|
||||
}{
|
||||
{
|
||||
name: "version only",
|
||||
filename: "001.sql",
|
||||
want: "001",
|
||||
},
|
||||
{
|
||||
name: "version with description",
|
||||
filename: "001_initial_schema.sql",
|
||||
want: "001",
|
||||
},
|
||||
{
|
||||
name: "multi-digit version",
|
||||
filename: "042_add_indexes.sql",
|
||||
want: "042",
|
||||
},
|
||||
{
|
||||
name: "long version number",
|
||||
filename: "00001_long_prefix.sql",
|
||||
want: "00001",
|
||||
},
|
||||
{
|
||||
name: "description with multiple underscores",
|
||||
filename: "003_add_user_auth_tables.sql",
|
||||
want: "003",
|
||||
},
|
||||
{
|
||||
name: "empty filename",
|
||||
filename: ".sql",
|
||||
wantErr: true,
|
||||
},
|
||||
{
|
||||
name: "leading underscore",
|
||||
filename: "_description.sql",
|
||||
wantErr: true,
|
||||
},
|
||||
{
|
||||
name: "non-numeric version",
|
||||
filename: "abc_migration.sql",
|
||||
wantErr: true,
|
||||
},
|
||||
{
|
||||
name: "mixed alphanumeric version",
|
||||
filename: "001a_migration.sql",
|
||||
wantErr: true,
|
||||
},
|
||||
}
|
||||
|
||||
for _, tt := range tests {
|
||||
t.Run(tt.name, func(t *testing.T) {
|
||||
got, err := ParseMigrationVersion(tt.filename)
|
||||
if tt.wantErr {
|
||||
if err == nil {
|
||||
t.Errorf("ParseMigrationVersion(%q) expected error, got %q", tt.filename, got)
|
||||
}
|
||||
|
||||
return
|
||||
}
|
||||
|
||||
if err != nil {
|
||||
t.Errorf("ParseMigrationVersion(%q) unexpected error: %v", tt.filename, err)
|
||||
|
||||
return
|
||||
}
|
||||
|
||||
if got != tt.want {
|
||||
t.Errorf("ParseMigrationVersion(%q) = %q, want %q", tt.filename, got, tt.want)
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
func TestApplyMigrations_CreatesSchemaAndTables(t *testing.T) {
|
||||
db := openTestDB(t)
|
||||
ctx := context.Background()
|
||||
|
||||
if err := ApplyMigrations(ctx, db, nil); err != nil {
|
||||
t.Fatalf("ApplyMigrations failed: %v", err)
|
||||
}
|
||||
|
||||
// The schema_migrations table must exist and contain at least
|
||||
// version "000" (the bootstrap) and "001" (the initial schema).
|
||||
rows, err := db.Query("SELECT version FROM schema_migrations ORDER BY version")
|
||||
if err != nil {
|
||||
t.Fatalf("failed to query schema_migrations: %v", err)
|
||||
}
|
||||
defer rows.Close()
|
||||
|
||||
var versions []string
|
||||
for rows.Next() {
|
||||
var v string
|
||||
if err := rows.Scan(&v); err != nil {
|
||||
t.Fatalf("failed to scan version: %v", err)
|
||||
}
|
||||
|
||||
versions = append(versions, v)
|
||||
}
|
||||
|
||||
if err := rows.Err(); err != nil {
|
||||
t.Fatalf("row iteration error: %v", err)
|
||||
}
|
||||
|
||||
if len(versions) < 2 {
|
||||
t.Fatalf("expected at least 2 migrations recorded, got %d: %v", len(versions), versions)
|
||||
}
|
||||
|
||||
if versions[0] != "000" {
|
||||
t.Errorf("first recorded migration = %q, want %q", versions[0], "000")
|
||||
}
|
||||
|
||||
if versions[1] != "001" {
|
||||
t.Errorf("second recorded migration = %q, want %q", versions[1], "001")
|
||||
}
|
||||
|
||||
// Verify that the application tables created by 001.sql exist.
|
||||
for _, table := range []string{"source_content", "source_metadata", "output_content", "request_cache", "negative_cache", "cache_stats"} {
|
||||
var count int
|
||||
|
||||
err := db.QueryRow(
|
||||
"SELECT COUNT(*) FROM sqlite_master WHERE type='table' AND name=?",
|
||||
table,
|
||||
).Scan(&count)
|
||||
if err != nil {
|
||||
t.Fatalf("failed to check for table %s: %v", table, err)
|
||||
}
|
||||
|
||||
if count != 1 {
|
||||
t.Errorf("table %s does not exist after migrations", table)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func TestApplyMigrations_Idempotent(t *testing.T) {
|
||||
db := openTestDB(t)
|
||||
ctx := context.Background()
|
||||
|
||||
if err := ApplyMigrations(ctx, db, nil); err != nil {
|
||||
t.Fatalf("first ApplyMigrations failed: %v", err)
|
||||
}
|
||||
|
||||
// Running a second time must succeed without errors.
|
||||
if err := ApplyMigrations(ctx, db, nil); err != nil {
|
||||
t.Fatalf("second ApplyMigrations failed: %v", err)
|
||||
}
|
||||
|
||||
// Verify no duplicate rows in schema_migrations.
|
||||
var count int
|
||||
|
||||
err := db.QueryRow("SELECT COUNT(*) FROM schema_migrations WHERE version = '000'").Scan(&count)
|
||||
if err != nil {
|
||||
t.Fatalf("failed to count 000 rows: %v", err)
|
||||
}
|
||||
|
||||
if count != 1 {
|
||||
t.Errorf("expected exactly 1 row for version 000, got %d", count)
|
||||
}
|
||||
}
|
||||
|
||||
func TestBootstrapMigrationsTable_FreshDatabase(t *testing.T) {
|
||||
db := openTestDB(t)
|
||||
ctx := context.Background()
|
||||
|
||||
if err := bootstrapMigrationsTable(ctx, db, nil); err != nil {
|
||||
t.Fatalf("bootstrapMigrationsTable failed: %v", err)
|
||||
}
|
||||
|
||||
// schema_migrations table must exist.
|
||||
var tableCount int
|
||||
|
||||
err := db.QueryRow(
|
||||
"SELECT COUNT(*) FROM sqlite_master WHERE type='table' AND name='schema_migrations'",
|
||||
).Scan(&tableCount)
|
||||
if err != nil {
|
||||
t.Fatalf("failed to check for table: %v", err)
|
||||
}
|
||||
|
||||
if tableCount != 1 {
|
||||
t.Fatalf("schema_migrations table not created")
|
||||
}
|
||||
|
||||
// Version "000" must be recorded.
|
||||
var recorded int
|
||||
|
||||
err = db.QueryRow(
|
||||
"SELECT COUNT(*) FROM schema_migrations WHERE version = '000'",
|
||||
).Scan(&recorded)
|
||||
if err != nil {
|
||||
t.Fatalf("failed to check version: %v", err)
|
||||
}
|
||||
|
||||
if recorded != 1 {
|
||||
t.Errorf("expected version 000 to be recorded, got count %d", recorded)
|
||||
}
|
||||
}
|
||||
9
internal/database/schema/000.sql
Normal file
9
internal/database/schema/000.sql
Normal file
@@ -0,0 +1,9 @@
|
||||
-- Migration 000: Schema migrations tracking table
|
||||
-- Applied as a bootstrap step before the normal migration loop.
|
||||
|
||||
CREATE TABLE IF NOT EXISTS schema_migrations (
|
||||
version TEXT PRIMARY KEY,
|
||||
applied_at DATETIME DEFAULT CURRENT_TIMESTAMP
|
||||
);
|
||||
|
||||
INSERT OR IGNORE INTO schema_migrations (version) VALUES ('000');
|
||||
@@ -82,7 +82,7 @@ func setupTestDB(t *testing.T) *sql.DB {
|
||||
t.Fatalf("failed to open test db: %v", err)
|
||||
}
|
||||
|
||||
if err := database.ApplyMigrations(db); err != nil {
|
||||
if err := database.ApplyMigrations(context.Background(), db, nil); err != nil {
|
||||
t.Fatalf("failed to apply migrations: %v", err)
|
||||
}
|
||||
|
||||
|
||||
@@ -16,7 +16,7 @@ func setupStatsTestDB(t *testing.T) *sql.DB {
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if err := database.ApplyMigrations(db); err != nil {
|
||||
if err := database.ApplyMigrations(context.Background(), db, nil); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
t.Cleanup(func() { db.Close() })
|
||||
|
||||
@@ -2,6 +2,7 @@ package imgcache
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"context"
|
||||
"database/sql"
|
||||
"image"
|
||||
"image/color"
|
||||
@@ -193,7 +194,7 @@ func setupServiceTestDB(t *testing.T) *sql.DB {
|
||||
}
|
||||
|
||||
// Use the real production schema via migrations
|
||||
if err := database.ApplyMigrations(db); err != nil {
|
||||
if err := database.ApplyMigrations(context.Background(), db, nil); err != nil {
|
||||
t.Fatalf("failed to apply migrations: %v", err)
|
||||
}
|
||||
|
||||
|
||||
Reference in New Issue
Block a user