92 lines
2.0 KiB
Go
92 lines
2.0 KiB
Go
package db
|
|
|
|
import (
|
|
"context"
|
|
"fmt"
|
|
"os"
|
|
"path/filepath"
|
|
"sort"
|
|
"strings"
|
|
"time"
|
|
|
|
"github.com/jackc/pgx/v5/pgxpool"
|
|
)
|
|
|
|
func Connect(ctx context.Context, databaseURL string) (*pgxpool.Pool, error) {
|
|
cfg, err := pgxpool.ParseConfig(databaseURL)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("parse database url: %w", err)
|
|
}
|
|
cfg.MaxConns = 20
|
|
cfg.MinConns = 2
|
|
cfg.MaxConnLifetime = time.Hour
|
|
cfg.MaxConnIdleTime = 15 * time.Minute
|
|
cfg.HealthCheckPeriod = time.Minute
|
|
|
|
pool, err := pgxpool.NewWithConfig(ctx, cfg)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("connect: %w", err)
|
|
}
|
|
pingCtx, cancel := context.WithTimeout(ctx, 10*time.Second)
|
|
defer cancel()
|
|
if err := pool.Ping(pingCtx); err != nil {
|
|
pool.Close()
|
|
return nil, fmt.Errorf("ping: %w", err)
|
|
}
|
|
return pool, nil
|
|
}
|
|
|
|
func Migrate(ctx context.Context, pool *pgxpool.Pool, dir string) error {
|
|
entries, err := os.ReadDir(dir)
|
|
if err != nil {
|
|
return fmt.Errorf("read migrations: %w", err)
|
|
}
|
|
var files []string
|
|
for _, e := range entries {
|
|
if e.IsDir() || !strings.HasSuffix(e.Name(), ".sql") {
|
|
continue
|
|
}
|
|
files = append(files, e.Name())
|
|
}
|
|
sort.Strings(files)
|
|
|
|
_, err = pool.Exec(ctx, `
|
|
CREATE TABLE IF NOT EXISTS schema_migrations (
|
|
filename TEXT PRIMARY KEY,
|
|
applied_at TIMESTAMPTZ NOT NULL DEFAULT NOW()
|
|
)`)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
|
|
for _, name := range files {
|
|
var exists bool
|
|
if err := pool.QueryRow(ctx, `SELECT EXISTS(SELECT 1 FROM schema_migrations WHERE filename=$1)`, name).Scan(&exists); err != nil {
|
|
return err
|
|
}
|
|
if exists {
|
|
continue
|
|
}
|
|
body, err := os.ReadFile(filepath.Join(dir, name))
|
|
if err != nil {
|
|
return err
|
|
}
|
|
tx, err := pool.Begin(ctx)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
if _, err := tx.Exec(ctx, string(body)); err != nil {
|
|
_ = tx.Rollback(ctx)
|
|
return fmt.Errorf("migration %s: %w", name, err)
|
|
}
|
|
if _, err := tx.Exec(ctx, `INSERT INTO schema_migrations (filename) VALUES ($1)`, name); err != nil {
|
|
_ = tx.Rollback(ctx)
|
|
return err
|
|
}
|
|
if err := tx.Commit(ctx); err != nil {
|
|
return err
|
|
}
|
|
}
|
|
return nil
|
|
}
|