// Package store owns the database: the pool, the migration runner and every // query. Nothing above it writes SQL. package store import ( "context" "crypto/rand" "embed" "encoding/base32" "fmt" "sort" "strings" "time" "github.com/jackc/pgx/v5/pgxpool" ) //go:embed all:migrations var migrationFS embed.FS type Store struct { pool *pgxpool.Pool // Budapest is where a campaign's closing day is decided. Loaded once // because time.LoadLocation parses the zone file on every call. Budapest *time.Location } func New(ctx context.Context, databaseURL string) (*Store, error) { pool, err := pgxpool.New(ctx, databaseURL) if err != nil { return nil, fmt.Errorf("connect: %w", err) } loc, err := time.LoadLocation("Europe/Budapest") if err != nil { return nil, fmt.Errorf("load Europe/Budapest: %w", err) } return &Store{pool: pool, Budapest: loc}, nil } func (s *Store) Close() { s.pool.Close() } // Ping is the readiness probe's only question: can we reach Postgres. func (s *Store) Ping(ctx context.Context) error { var one int return s.pool.QueryRow(ctx, "SELECT 1").Scan(&one) } // Migrate applies every embedded migration that has not run yet, in filename // order, each in its own transaction. An advisory lock keeps two containers // starting at the same time from racing. func (s *Store) Migrate(ctx context.Context) error { conn, err := s.pool.Acquire(ctx) if err != nil { return err } defer conn.Release() if _, err := conn.Exec(ctx, `SELECT pg_advisory_lock(4823001)`); err != nil { return fmt.Errorf("lock: %w", err) } defer conn.Exec(ctx, `SELECT pg_advisory_unlock(4823001)`) if _, err := conn.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 schema_migrations: %w", err) } applied := map[string]bool{} rows, err := conn.Query(ctx, `SELECT version FROM schema_migrations`) if err != nil { return err } for rows.Next() { var v string if err := rows.Scan(&v); err != nil { rows.Close() return err } applied[v] = true } rows.Close() if err := rows.Err(); err != nil { return err } entries, err := migrationFS.ReadDir("migrations") if err != nil { return err } var names []string for _, e := range entries { if strings.HasSuffix(e.Name(), ".sql") { names = append(names, e.Name()) } } sort.Strings(names) for _, name := range names { version := strings.TrimSuffix(name, ".sql") if applied[version] { continue } body, err := migrationFS.ReadFile("migrations/" + name) if err != nil { return err } tx, err := conn.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 (version) VALUES ($1)`, version); err != nil { tx.Rollback(ctx) return err } if err := tx.Commit(ctx); err != nil { return fmt.Errorf("migration %s commit: %w", name, err) } fmt.Printf("applied migration %s\n", version) } return nil } var idEncoding = base32.NewEncoding("0123456789abcdefghjkmnpqrstvwxyz").WithPadding(base32.NoPadding) // NewID replaces Prisma's cuid(). Random rather than sortable on purpose: // these ids appear in URLs, and a sortable id leaks how many campaigns exist. func NewID() string { b := make([]byte, 15) if _, err := rand.Read(b); err != nil { panic("crypto/rand failed: " + err.Error()) } return idEncoding.EncodeToString(b) } // Day formats a timestamp as the calendar day it falls on in Budapest. // The frontend receives days, never instants, so a backer in another timezone // sees the same deadline as the campaign owner. func (s *Store) Day(t time.Time) string { return t.In(s.Budapest).Format("2006-01-02") } // EndOfDay turns "2026-08-31" into the last instant of that day in Budapest. func (s *Store) EndOfDay(day string) (time.Time, error) { d, err := time.ParseInLocation("2006-01-02", day, s.Budapest) if err != nil { return time.Time{}, err } return d.Add(24*time.Hour - time.Second), nil }