package store import ( "context" "database/sql" "fmt" "log" ) const migrateLockKey int64 = 0x706c756d5f6d6967 // "plum_mig" // migrateUserProfileColumns adds avatar_url and state when missing (existing DBs). func migrateUserProfileColumns(ctx context.Context, exec execContext) error { cols := []string{"avatar_url", "state"} for _, col := range cols { stmt := fmt.Sprintf(`ALTER TABLE users ADD COLUMN IF NOT EXISTS %s TEXT NOT NULL DEFAULT ''`, col) if _, err := exec.ExecContext(ctx, stmt); err != nil { return fmt.Errorf("add column %s: %w", col, err) } } return nil } // migrateUserEmail adds email and a partial unique index on lower(email). func migrateUserEmail(ctx context.Context, exec execContext) error { if _, err := exec.ExecContext(ctx, `ALTER TABLE users ADD COLUMN IF NOT EXISTS email TEXT NOT NULL DEFAULT ''`); err != nil { return fmt.Errorf("add column email: %w", err) } if _, err := exec.ExecContext(ctx, ` CREATE UNIQUE INDEX IF NOT EXISTS users_email_lower_uidx ON users (lower(email)) WHERE email <> ''`); err != nil { return fmt.Errorf("users_email_lower_uidx: %w", err) } return nil } // migratePosts creates the unified post model and snapshots legacy content. // Legacy tables remain in place until the application cutover is complete. func migratePosts(ctx context.Context, exec execContext) error { steps := []struct { name string sql string }{ {"create posts", ` CREATE TABLE IF NOT EXISTS posts ( id TEXT PRIMARY KEY, parent_id TEXT REFERENCES posts(id) ON DELETE CASCADE, author_id TEXT NOT NULL REFERENCES users(id), title TEXT NOT NULL DEFAULT '', body TEXT NOT NULL, city TEXT NOT NULL DEFAULT '', post_date TEXT NOT NULL DEFAULT '', hidden INTEGER NOT NULL DEFAULT 0, created_at TEXT NOT NULL, updated_at TEXT NOT NULL, CHECK ( (parent_id IS NULL AND title <> '' AND post_date <> '') OR (parent_id IS NOT NULL AND title = '' AND city = '' AND post_date = '' AND hidden = 0) ) )`}, {"index post replies", ` CREATE INDEX IF NOT EXISTS idx_posts_parent_created ON posts(parent_id, created_at, id)`}, {"index root posts", ` CREATE INDEX IF NOT EXISTS idx_posts_root_date ON posts(post_date, hidden) WHERE parent_id IS NULL`}, {"create post votes", ` CREATE TABLE IF NOT EXISTS post_votes ( user_id TEXT NOT NULL REFERENCES users(id), post_id TEXT NOT NULL REFERENCES posts(id) ON DELETE CASCADE, value INTEGER NOT NULL CHECK (value IN (-1, 1)), PRIMARY KEY (user_id, post_id) )`}, {"copy questions", ` INSERT INTO posts ( id, parent_id, author_id, title, body, city, post_date, hidden, created_at, updated_at ) SELECT id, NULL, author_id, title, body, city, hunt_date, hidden, created_at, created_at FROM questions ON CONFLICT (id) DO NOTHING`}, {"copy answers", ` INSERT INTO posts ( id, parent_id, author_id, title, body, city, post_date, hidden, created_at, updated_at ) SELECT 'answer:' || question_id, question_id, author_id, '', body, '', '', 0, created_at, updated_at FROM answers ON CONFLICT (id) DO NOTHING`}, {"copy votes", ` INSERT INTO post_votes (user_id, post_id, value) SELECT user_id, question_id, value FROM votes ON CONFLICT (user_id, post_id) DO NOTHING`}, } for _, step := range steps { if _, err := exec.ExecContext(ctx, step.sql); err != nil { return fmt.Errorf("%s: %w", step.name, err) } } return nil } type execContext interface { ExecContext(ctx context.Context, query string, args ...any) (sql.Result, error) QueryContext(ctx context.Context, query string, args ...any) (*sql.Rows, error) } // applyMigrations runs versioned migrations under a session-level advisory lock // held for the entire process (check versions → apply → record). func applyMigrations(db *sql.DB, schemaSQL string) error { ctx := context.Background() conn, err := db.Conn(ctx) if err != nil { return err } defer conn.Close() if _, err := conn.ExecContext(ctx, `SELECT pg_advisory_lock($1)`, migrateLockKey); err != nil { return fmt.Errorf("migrate lock: %w", err) } defer func() { if _, unlockErr := conn.ExecContext(ctx, `SELECT pg_advisory_unlock($1)`, migrateLockKey); unlockErr != nil { log.Printf("migrate unlock: %v", unlockErr) } }() if _, err := conn.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("schema_migrations: %w", err) } applied, err := appliedVersions(ctx, conn) if err != nil { return err } migrations := []struct { version string run func(context.Context, execContext) error }{ {"001_schema", func(ctx context.Context, exec execContext) error { return applySchema(ctx, exec, schemaSQL) }}, {"002_user_profile_columns", migrateUserProfileColumns}, {"003_user_email", migrateUserEmail}, {"004_posts", migratePosts}, } for _, m := range migrations { if applied[m.version] { continue } log.Printf("migrate: applying %s", m.version) if err := m.run(ctx, conn); err != nil { return fmt.Errorf("migrate %s: %w", m.version, err) } if _, err := conn.ExecContext(ctx, `INSERT INTO schema_migrations (version) VALUES ($1)`, m.version); err != nil { return fmt.Errorf("record %s: %w", m.version, err) } } return nil } func appliedVersions(ctx context.Context, exec execContext) (map[string]bool, error) { rows, err := exec.QueryContext(ctx, `SELECT version FROM schema_migrations`) if err != nil { return nil, err } defer rows.Close() out := map[string]bool{} for rows.Next() { var v string if err := rows.Scan(&v); err != nil { return nil, err } out[v] = true } return out, rows.Err() }