|
| 1 | +package migrate |
| 2 | + |
| 3 | +import ( |
| 4 | + "context" |
| 5 | + "database/sql" |
| 6 | + "fmt" |
| 7 | + "log/slog" |
| 8 | + "time" |
| 9 | +) |
| 10 | + |
| 11 | +type Migration struct { |
| 12 | + Name string |
| 13 | + SQL string |
| 14 | +} |
| 15 | + |
| 16 | +type Migrations struct { |
| 17 | + ModuleName string |
| 18 | + Migrations []Migration |
| 19 | +} |
| 20 | + |
| 21 | +type Dialect struct { |
| 22 | + createSQL string |
| 23 | + claimSQL string |
| 24 | +} |
| 25 | + |
| 26 | +var ( |
| 27 | + SQLite = Dialect{ |
| 28 | + createSQL: `CREATE TABLE if NOT EXISTS schema_migrations ( |
| 29 | + module_name TEXT NOT NULL, |
| 30 | + migration TEXT NOT NULL, |
| 31 | + applied_at TEXT NOT NULL, |
| 32 | + PRIMARY KEY (module_name, migration) |
| 33 | + )`, |
| 34 | + claimSQL: `INSERT INTO schema_migrations (module_name, migration, applied_at) VALUES (?, ?, ?) ON CONFLICT(module_name, migration) DO NOTHING`, |
| 35 | + } |
| 36 | + Postgres = Dialect{ |
| 37 | + createSQL: `DO $$ BEGIN |
| 38 | + CREATE TABLE IF NOT EXISTS schema_migrations ( |
| 39 | + module_name TEXT NOT NULL, |
| 40 | + migration TEXT NOT NULL, |
| 41 | + applied_at TIMESTAMPTZ NOT NULL, |
| 42 | + PRIMARY KEY (module_name, migration) |
| 43 | + ); |
| 44 | + EXCEPTION |
| 45 | + WHEN duplicate_table THEN NULL; |
| 46 | + WHEN duplicate_object THEN NULL; |
| 47 | + WHEN unique_violation THEN NULL; |
| 48 | + END $$`, |
| 49 | + claimSQL: `INSERT INTO schema_migrations (module_name, migration, applied_at) VALUES ($1, $2, $3) ON CONFLICT(module_name, migration) DO NOTHING`, |
| 50 | + } |
| 51 | + MySQL = Dialect{ |
| 52 | + createSQL: `CREATE TABLE if NOT EXISTS schema_migrations ( |
| 53 | + module_name VARCHAR(255) NOT NULL, |
| 54 | + migration VARCHAR(255) NOT NULL, |
| 55 | + applied_at DATETIME NOT NULL, |
| 56 | + PRIMARY KEY (module_name, migration) |
| 57 | + )`, |
| 58 | + claimSQL: `INSERT IGNORE INTO schema_migrations (module_name, migration, applied_at) VALUES (?, ?, ?)`, |
| 59 | + } |
| 60 | +) |
| 61 | + |
| 62 | +type Option interface { |
| 63 | + apply(*config) |
| 64 | +} |
| 65 | + |
| 66 | +type config struct { |
| 67 | + migrations []Migrations |
| 68 | + logger *slog.Logger |
| 69 | +} |
| 70 | + |
| 71 | +func (m Migrations) apply(c *config) { |
| 72 | + c.migrations = append(c.migrations, m) |
| 73 | +} |
| 74 | + |
| 75 | +type loggerOption struct{ logger *slog.Logger } |
| 76 | + |
| 77 | +func (o loggerOption) apply(c *config) { c.logger = o.logger } |
| 78 | + |
| 79 | +func WithLogger(logger *slog.Logger) Option { return loggerOption{logger} } |
| 80 | + |
| 81 | +// Migrate runs database migrations in a safe and idempotent way. |
| 82 | +// |
| 83 | +// It creates a schema_migrations table if it doesn't exist, then applies each migration |
| 84 | +// exactly once by claiming it first. Migrations are grouped by module name and executed |
| 85 | +// in order. Each migration runs in its own transaction - if it fails, the transaction |
| 86 | +// is rolled back and an error is returned. |
| 87 | +// |
| 88 | +// The function is safe for concurrent execution: only one instance will successfully |
| 89 | +// claim and apply each migration. Already applied migrations are skipped automatically. |
| 90 | +// |
| 91 | +// Parameters: |
| 92 | +// - ctx: context for cancellation and timeout control |
| 93 | +// - db: database connection to run migrations on |
| 94 | +// - dialect: database-specific SQL syntax (SQLite, Postgres, or MySQL) |
| 95 | +// - opts: optional configuration (migrations to run, logger for output) |
| 96 | +// |
| 97 | +// Returns an error if table creation fails, transaction handling fails, or any |
| 98 | +// migration SQL fails to execute. |
| 99 | +func Migrate( |
| 100 | + ctx context.Context, |
| 101 | + db *sql.DB, |
| 102 | + dialect Dialect, |
| 103 | + opts ...Option, |
| 104 | +) error { |
| 105 | + var cfg config |
| 106 | + for _, o := range opts { |
| 107 | + o.apply(&cfg) |
| 108 | + } |
| 109 | + |
| 110 | + _, err := db.ExecContext(ctx, dialect.createSQL) |
| 111 | + if err != nil { |
| 112 | + return fmt.Errorf("create schema_migrations: %w", err) |
| 113 | + } |
| 114 | + |
| 115 | + for _, group := range cfg.migrations { |
| 116 | + for _, m := range group.Migrations { |
| 117 | + tx, err := db.BeginTx(ctx, nil) |
| 118 | + if err != nil { |
| 119 | + return fmt.Errorf("begin tx for %s/%s: %w", group.ModuleName, m.Name, err) |
| 120 | + } |
| 121 | + |
| 122 | + res, err := tx.ExecContext(ctx, dialect.claimSQL, |
| 123 | + group.ModuleName, m.Name, time.Now().UTC(), |
| 124 | + ) |
| 125 | + if err != nil { |
| 126 | + _ = tx.Rollback() |
| 127 | + return fmt.Errorf("claim migration %s/%s: %w", group.ModuleName, m.Name, err) |
| 128 | + } |
| 129 | + |
| 130 | + n, _ := res.RowsAffected() |
| 131 | + if n == 0 { |
| 132 | + _ = tx.Rollback() |
| 133 | + continue |
| 134 | + } |
| 135 | + |
| 136 | + if _, err := tx.ExecContext(ctx, m.SQL); err != nil { |
| 137 | + _ = tx.Rollback() |
| 138 | + return fmt.Errorf("apply migration %s/%s: %w", group.ModuleName, m.Name, err) |
| 139 | + } |
| 140 | + |
| 141 | + if err := tx.Commit(); err != nil { |
| 142 | + return fmt.Errorf("commit migration %s/%s: %w", group.ModuleName, m.Name, err) |
| 143 | + } |
| 144 | + |
| 145 | + if cfg.logger != nil { |
| 146 | + cfg.logger.InfoContext(ctx, "applied migration", |
| 147 | + "module", group.ModuleName, |
| 148 | + "name", m.Name, |
| 149 | + ) |
| 150 | + } |
| 151 | + } |
| 152 | + } |
| 153 | + |
| 154 | + return nil |
| 155 | +} |
0 commit comments