ahp
22d92c133d
Ship the product pass on the v0.1.3 schema with additive migrations, a replica migration lock, and bot accounts kept out of people lists.
79 lines
2.1 KiB
Go
79 lines
2.1 KiB
Go
package db
|
|
|
|
import (
|
|
"context"
|
|
"fmt"
|
|
"io/fs"
|
|
"sort"
|
|
"strings"
|
|
|
|
"github.com/jackc/pgx/v5"
|
|
"github.com/jackc/pgx/v5/pgxpool"
|
|
)
|
|
|
|
// migrateLockID is held for the whole migration pass so two replicas cannot
|
|
// apply the same file at once. The lock lives on one pooled connection and
|
|
// is released when that session ends.
|
|
const migrateLockID int64 = 764727013
|
|
|
|
// Migrate runs all *.sql files in the migrations fs in lexicographic order.
|
|
// Tracks applied migrations in schema_migrations.
|
|
func Migrate(ctx context.Context, pool *pgxpool.Pool, migrations fs.FS) error {
|
|
conn, err := pool.Acquire(ctx)
|
|
if err != nil {
|
|
return fmt.Errorf("migrate lock: %w", err)
|
|
}
|
|
defer conn.Release()
|
|
if _, err := conn.Exec(ctx, `SELECT pg_advisory_lock($1)`, migrateLockID); err != nil {
|
|
return fmt.Errorf("migrate lock: %w", err)
|
|
}
|
|
defer func() {
|
|
_, _ = conn.Exec(context.Background(), `SELECT pg_advisory_unlock($1)`, migrateLockID)
|
|
}()
|
|
|
|
if _, err := pool.Exec(ctx, `CREATE TABLE IF NOT EXISTS schema_migrations (version TEXT PRIMARY KEY, applied_at TIMESTAMPTZ NOT NULL DEFAULT now())`); err != nil {
|
|
return fmt.Errorf("create migrations table: %w", err)
|
|
}
|
|
entries, err := fs.ReadDir(migrations, ".")
|
|
if err != nil {
|
|
return err
|
|
}
|
|
var names []string
|
|
for _, e := range entries {
|
|
if !e.IsDir() && strings.HasSuffix(e.Name(), ".sql") {
|
|
names = append(names, e.Name())
|
|
}
|
|
}
|
|
sort.Strings(names)
|
|
|
|
for _, name := range names {
|
|
applied, err := pool.Exec(ctx, `SELECT 1 FROM schema_migrations WHERE version=$1`, name)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
if applied.RowsAffected() > 0 {
|
|
continue
|
|
}
|
|
body, err := fs.ReadFile(migrations, name)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
tx, err := pool.BeginTx(ctx, pgx.TxOptions{})
|
|
if err != nil {
|
|
return err
|
|
}
|
|
if _, err := tx.Exec(ctx, string(body)); err != nil {
|
|
_ = tx.Rollback(ctx)
|
|
return fmt.Errorf("apply %s: %w", name, err)
|
|
}
|
|
if _, err := tx.Exec(ctx, `INSERT INTO schema_migrations (version) VALUES ($1)`, name); err != nil {
|
|
_ = tx.Rollback(ctx)
|
|
return err
|
|
}
|
|
if err := tx.Commit(ctx); err != nil {
|
|
return err
|
|
}
|
|
}
|
|
return nil
|
|
}
|