package database import ( "context" "crypto/sha256" "embed" "fmt" "sort" "github.com/google/uuid" "github.com/jackc/pgx/v5" "github.com/jackc/pgx/v5/pgxpool" "veyra/services/api/internal/platform/config" ) //go:embed migrations/*.sql var migrations embed.FS func Open(ctx context.Context, c config.Config) (*pgxpool.Pool, error) { if err := c.ValidateDatabase(); err != nil { return nil, err } p, err := pgxpool.New(ctx, c.DatabaseURL(c.PGDatabase)) if err != nil { return nil, fmt.Errorf("configure PostgreSQL pool") } if err := p.Ping(ctx); err != nil { p.Close() return nil, fmt.Errorf("PostgreSQL connection unavailable") } return p, nil } // Create creates only the configured database and never drops or resets data. func Create(ctx context.Context, c config.Config) (bool, error) { if err := c.ValidateDatabase(); err != nil { return false, err } conn, err := pgx.Connect(ctx, c.DatabaseURL(c.PGAdminDatabase)) if err != nil { return false, fmt.Errorf("administrative PostgreSQL connection unavailable") } defer conn.Close(context.Background()) if _, err := conn.Exec(ctx, "SELECT pg_advisory_lock(hashtext($1)::bigint)", "veyra:create:"+c.PGDatabase); err != nil { return false, fmt.Errorf("lock database creation: %w", err) } var exists bool if err := conn.QueryRow(ctx, "SELECT EXISTS(SELECT 1 FROM pg_database WHERE datname = $1)", c.PGDatabase).Scan(&exists); err != nil { return false, fmt.Errorf("inspect database: %w", err) } if exists { return false, nil } if _, err := conn.Exec(ctx, "CREATE DATABASE "+pgx.Identifier{c.PGDatabase}.Sanitize()); err != nil { return false, fmt.Errorf("create database failed; verify role CREATEDB permission") } return true, nil } // Migrate serializes concurrent runners and atomically records each migration. // Even the migration ledger follows the veyra_ prefix and string UUID ID rules. func Migrate(ctx context.Context, pool *pgxpool.Pool) error { conn, err := pool.Acquire(ctx) if err != nil { return err } defer conn.Release() tx, err := conn.Begin(ctx) if err != nil { return err } defer tx.Rollback(context.Background()) if _, err := tx.Exec(ctx, "SELECT pg_advisory_xact_lock(727198321)"); err != nil { return err } if _, err := tx.Exec(ctx, `CREATE TABLE IF NOT EXISTS veyra_schema_migration ( id varchar(36) PRIMARY KEY CHECK (id ~ '^[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$'), version varchar(120) NOT NULL UNIQUE, checksum varchar(64) NOT NULL, applied_at timestamptz NOT NULL DEFAULT now())`); err != nil { return fmt.Errorf("create migration ledger: %w", err) } entries, err := migrations.ReadDir("migrations") if err != nil { return err } sort.Slice(entries, func(i, j int) bool { return entries[i].Name() < entries[j].Name() }) for _, entry := range entries { sql, err := migrations.ReadFile("migrations/" + entry.Name()) if err != nil { return err } checksum := fmt.Sprintf("%x", sha256.Sum256(sql)) var stored string err = tx.QueryRow(ctx, "SELECT checksum FROM veyra_schema_migration WHERE version=$1", entry.Name()).Scan(&stored) if err == nil { if stored != checksum { return fmt.Errorf("migration %s checksum mismatch", entry.Name()) } continue } if err != pgx.ErrNoRows { return err } if _, err := tx.Exec(ctx, string(sql)); err != nil { return fmt.Errorf("apply %s: %w", entry.Name(), err) } if _, err := tx.Exec(ctx, "INSERT INTO veyra_schema_migration(id,version,checksum) VALUES($1,$2,$3)", uuid.NewString(), entry.Name(), checksum); err != nil { return err } } return tx.Commit(ctx) }