package database import ( "context" "database/sql" "embed" "errors" "fmt" "io/fs" "sort" "strings" "time" _ "github.com/jackc/pgx/v5/stdlib" ) //go:embed migrations/*.sql var migrationFiles embed.FS func Open(ctx context.Context, databaseURL string) (*sql.DB, error) { db, err := sql.Open("pgx", databaseURL) if err != nil { return nil, fmt.Errorf("open database: %w", err) } db.SetMaxOpenConns(20) db.SetMaxIdleConns(5) db.SetConnMaxLifetime(30 * time.Minute) if err := db.PingContext(ctx); err != nil { db.Close() return nil, fmt.Errorf("ping database: %w", err) } return db, nil } // migrationLockKey serialises migration runs between instances of this service. // It is deliberately distinct from the advisory lock used to claim the first // system administrator. const migrationLockKey int64 = 554628171918 // querier is the subset of database/sql that migrations need, satisfied by both // *sql.DB and *sql.Conn. type querier interface { ExecContext(ctx context.Context, query string, args ...any) (sql.Result, error) QueryRowContext(ctx context.Context, query string, args ...any) *sql.Row BeginTx(ctx context.Context, opts *sql.TxOptions) (*sql.Tx, error) } // Migrate brings the schema up to date. // // The whole run holds a session-level advisory lock, because PostgreSQL's // CREATE TABLE IF NOT EXISTS is not race free: two instances starting against an // empty database at the same time can collide on the pg_type unique index and one // of them fails. The same race would let two instances apply the same migration // concurrently. Rolling a deployment with more than one replica is enough to // trigger it. func Migrate(ctx context.Context, db *sql.DB) error { conn, err := db.Conn(ctx) if err != nil { return fmt.Errorf("reserve migration connection: %w", err) } defer conn.Close() if _, err := conn.ExecContext(ctx, `SELECT pg_advisory_lock($1)`, migrationLockKey); err != nil { return fmt.Errorf("lock migrations: %w", err) } // The unlock runs on a context detached from the caller's, so an application // shutting down mid-migration still releases the lock for the next instance. defer func() { _, _ = conn.ExecContext(context.WithoutCancel(ctx), `SELECT pg_advisory_unlock($1)`, migrationLockKey) }() return applyMigrations(ctx, conn) } func applyMigrations(ctx context.Context, db querier) error { if _, err := db.ExecContext(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 migration table: %w", err) } entries, err := fs.ReadDir(migrationFiles, "migrations") if err != nil { return fmt.Errorf("read migrations: %w", err) } sort.Slice(entries, func(i, j int) bool { return entries[i].Name() < entries[j].Name() }) for _, entry := range entries { if entry.IsDir() || !strings.HasSuffix(entry.Name(), ".sql") { continue } var applied bool err := db.QueryRowContext(ctx, `SELECT EXISTS(SELECT 1 FROM schema_migrations WHERE version = $1)`, entry.Name()).Scan(&applied) if err != nil { return fmt.Errorf("check migration %s: %w", entry.Name(), err) } if applied { continue } content, err := migrationFiles.ReadFile("migrations/" + entry.Name()) if err != nil { return fmt.Errorf("read migration %s: %w", entry.Name(), err) } up, err := upSection(string(content)) if err != nil { return fmt.Errorf("parse migration %s: %w", entry.Name(), err) } tx, err := db.BeginTx(ctx, nil) if err != nil { return fmt.Errorf("begin migration %s: %w", entry.Name(), err) } if _, err = tx.ExecContext(ctx, up); err == nil { _, err = tx.ExecContext(ctx, `INSERT INTO schema_migrations(version) VALUES ($1)`, entry.Name()) } if err != nil { tx.Rollback() return fmt.Errorf("apply migration %s: %w", entry.Name(), err) } if err := tx.Commit(); err != nil { return fmt.Errorf("commit migration %s: %w", entry.Name(), err) } } return nil } func upSection(content string) (string, error) { const upMarker = "-- +goose Up" const downMarker = "-- +goose Down" start := strings.Index(content, upMarker) if start < 0 { return "", errors.New("missing up marker") } content = content[start+len(upMarker):] if end := strings.Index(content, downMarker); end >= 0 { content = content[:end] } if strings.TrimSpace(content) == "" { return "", errors.New("empty up migration") } return content, nil }