100 lines
2.7 KiB
Go
100 lines
2.7 KiB
Go
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
|
|
}
|
|
|
|
func Migrate(ctx context.Context, db *sql.DB) 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
|
|
}
|