package repository import ( "context" "fmt" "os" "path/filepath" "sort" "strconv" "strings" "time" "github.com/jackc/pgx/v5/pgxpool" "go.uber.org/zap" ) type Postgres struct { Pool *pgxpool.Pool logger *zap.Logger } func NewPostgres(ctx context.Context, uri string, logger *zap.Logger) (*Postgres, error) { if uri == "" { return nil, fmt.Errorf("postgres URI is empty") } cfg, err := pgxpool.ParseConfig(uri) if err != nil { return nil, fmt.Errorf("postgres parse config: %w", err) } cfg.MaxConns = 20 cfg.MinConns = 2 cfg.MaxConnLifetime = 30 * time.Minute cfg.MaxConnIdleTime = 5 * time.Minute pingCtx, cancel := context.WithTimeout(ctx, 10*time.Second) defer cancel() pool, err := pgxpool.NewWithConfig(pingCtx, cfg) if err != nil { return nil, fmt.Errorf("postgres connect: %w", err) } if err := pool.Ping(pingCtx); err != nil { pool.Close() return nil, fmt.Errorf("postgres ping: %w", err) } logger.Info("connected to Postgres") return &Postgres{Pool: pool, logger: logger}, nil } func (p *Postgres) Close() { if p == nil || p.Pool == nil { return } p.Pool.Close() } // RunMigrations applies any *.up.sql files in dir that haven't been recorded // in schema_migrations yet, in lexical order. Tiny homegrown runner — no // external dependency. Down-migrations are not executed automatically. func (p *Postgres) RunMigrations(ctx context.Context, dir string) error { if _, err := p.Pool.Exec(ctx, ` CREATE TABLE IF NOT EXISTS schema_migrations ( version BIGINT PRIMARY KEY, applied_at TIMESTAMPTZ NOT NULL DEFAULT now() )`); err != nil { return fmt.Errorf("create schema_migrations: %w", err) } files, err := filepath.Glob(filepath.Join(dir, "*.up.sql")) if err != nil { return fmt.Errorf("glob migrations: %w", err) } sort.Strings(files) for _, f := range files { version, err := parseMigrationVersion(f) if err != nil { p.logger.Warn("skipping unparseable migration filename", zap.String("file", f), zap.Error(err)) continue } var exists bool if err := p.Pool.QueryRow(ctx, `SELECT EXISTS(SELECT 1 FROM schema_migrations WHERE version = $1)`, version, ).Scan(&exists); err != nil { return fmt.Errorf("check migration %d: %w", version, err) } if exists { continue } sqlBytes, err := os.ReadFile(f) if err != nil { return fmt.Errorf("read %s: %w", f, err) } tx, err := p.Pool.Begin(ctx) if err != nil { return fmt.Errorf("begin tx for %s: %w", f, err) } if _, err := tx.Exec(ctx, string(sqlBytes)); err != nil { _ = tx.Rollback(ctx) return fmt.Errorf("apply %s: %w", f, err) } if _, err := tx.Exec(ctx, `INSERT INTO schema_migrations(version) VALUES ($1)`, version); err != nil { _ = tx.Rollback(ctx) return fmt.Errorf("record %s: %w", f, err) } if err := tx.Commit(ctx); err != nil { return fmt.Errorf("commit %s: %w", f, err) } p.logger.Info("applied migration", zap.String("file", filepath.Base(f)), zap.Int64("version", version)) } return nil } func parseMigrationVersion(path string) (int64, error) { base := filepath.Base(path) idx := strings.Index(base, "_") if idx <= 0 { return 0, fmt.Errorf("filename %q must start with _", base) } v, err := strconv.ParseInt(base[:idx], 10, 64) if err != nil { return 0, fmt.Errorf("version prefix not an integer: %w", err) } return v, nil }