plumber/internal/store/migrate.go

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
		args []any
	}{
		{name: "create posts", sql: `
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 '',
  post_state TEXT NOT NULL,
  created_at TEXT NOT NULL,
  updated_at TEXT NOT NULL,
  CONSTRAINT posts_shape_check CHECK (
    (parent_id IS NULL AND title <> '' AND post_date <> '')
    OR
    (parent_id IS NOT NULL AND title = '' AND city = '' AND post_date = '')
  )
)`},
		{name: "index post replies", sql: `
CREATE INDEX IF NOT EXISTS idx_posts_parent_created
  ON posts(parent_id, created_at, id)`},
		{name: "index root posts", sql: `
CREATE INDEX IF NOT EXISTS idx_posts_root_date
  ON posts(post_date, post_state)
  WHERE parent_id IS NULL`},
		{name: "create post votes", sql: `
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)
)`},
		{name: "copy questions", sql: `
INSERT INTO posts (
  id, parent_id, author_id, title, body, city, post_date, post_state, created_at, updated_at
)
SELECT
  id, NULL, author_id, title, body, city, hunt_date,
  CASE WHEN hidden = 0 THEN $1 ELSE $2 END,
  created_at, created_at
FROM questions
ON CONFLICT (id) DO NOTHING`, args: []any{string(PostStateVisible), string(PostStateHidden)}},
		{name: "copy answers", sql: `
INSERT INTO posts (
  id, parent_id, author_id, title, body, city, post_date, post_state, created_at, updated_at
)
SELECT
  'answer:' || question_id, question_id, author_id, '', body, '', '',
  $1, created_at, updated_at
FROM answers
ON CONFLICT (id) DO NOTHING`, args: []any{string(PostStateVisible)}},
		{name: "copy votes", sql: `
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, step.args...); err != nil {
			return fmt.Errorf("%s: %w", step.name, err)
		}
	}
	return nil
}

func migratePostVoteIndex(ctx context.Context, exec execContext) error {
	if _, err := exec.ExecContext(ctx, `
CREATE INDEX IF NOT EXISTS idx_post_votes_post_id
  ON post_votes(post_id)`); err != nil {
		return fmt.Errorf("idx_post_votes_post_id: %w", err)
	}
	return nil
}

func migratePostDate(ctx context.Context, exec execContext) error {
	steps := []struct {
		name string
		sql  string
	}{
		{"rename post date", `
DO $migration$
BEGIN
  IF EXISTS (
    SELECT 1
    FROM information_schema.columns
    WHERE table_schema = current_schema()
      AND table_name = 'posts'
      AND column_name = 'hunt_date'
  ) AND NOT EXISTS (
    SELECT 1
    FROM information_schema.columns
    WHERE table_schema = current_schema()
      AND table_name = 'posts'
      AND column_name = 'post_date'
  ) THEN
    ALTER TABLE posts RENAME COLUMN hunt_date TO post_date;
  END IF;
END
$migration$`},
		{"drop legacy root date index", `
DROP INDEX IF EXISTS idx_posts_root_hunt`},
		{"create root date index", `
DO $migration$
BEGIN
  IF EXISTS (
    SELECT 1
    FROM information_schema.columns
    WHERE table_schema = current_schema()
      AND table_name = 'posts'
      AND column_name = 'post_state'
  ) THEN
    CREATE INDEX IF NOT EXISTS idx_posts_root_date
      ON posts(post_date, post_state)
      WHERE parent_id IS NULL;
  ELSE
    CREATE INDEX IF NOT EXISTS idx_posts_root_date
      ON posts(post_date, hidden)
      WHERE parent_id IS NULL;
  END IF;
END
$migration$`},
	}
	for _, step := range steps {
		if _, err := exec.ExecContext(ctx, step.sql); err != nil {
			return fmt.Errorf("%s: %w", step.name, err)
		}
	}
	return nil
}

func migratePostState(ctx context.Context, exec execContext) error {
	if _, err := exec.ExecContext(ctx, `
ALTER TABLE posts
  ADD COLUMN IF NOT EXISTS post_state TEXT`); err != nil {
		return fmt.Errorf("add post state: %w", err)
	}
	hasHidden, err := migrationColumnExists(ctx, exec, "posts", "hidden")
	if err != nil {
		return fmt.Errorf("check hidden column: %w", err)
	}
	if hasHidden {
		if _, err := exec.ExecContext(ctx, `
UPDATE posts
SET post_state = CASE
  WHEN hidden = 0 THEN $1
  ELSE $2
END`, string(PostStateVisible), string(PostStateHidden)); err != nil {
			return fmt.Errorf("copy hidden state: %w", err)
		}
	}

	steps := []struct {
		name string
		sql  string
	}{
		{"drop root date index", `
DROP INDEX IF EXISTS idx_posts_root_date`},
		{"drop legacy post shape constraint", `
ALTER TABLE posts DROP CONSTRAINT IF EXISTS posts_check`},
		{"drop post shape constraint", `
ALTER TABLE posts DROP CONSTRAINT IF EXISTS posts_shape_check`},
		{"drop hidden", `
ALTER TABLE posts DROP COLUMN IF EXISTS hidden`},
		{"require post state", `
ALTER TABLE posts ALTER COLUMN post_state SET NOT NULL`},
		{"create post shape constraint", `
ALTER TABLE posts
  ADD CONSTRAINT posts_shape_check CHECK (
    (parent_id IS NULL AND title <> '' AND post_date <> '')
    OR
    (parent_id IS NOT NULL AND title = '' AND city = '' AND post_date = '')
  )`},
		{"create root date index", `
CREATE INDEX idx_posts_root_date
  ON posts(post_date, post_state)
  WHERE parent_id IS NULL`},
	}
	for _, step := range steps {
		if _, err := exec.ExecContext(ctx, step.sql); err != nil {
			return fmt.Errorf("%s: %w", step.name, err)
		}
	}
	return nil
}

func migrationColumnExists(
	ctx context.Context,
	exec execContext,
	tableName string,
	columnName string,
) (bool, error) {
	rows, err := exec.QueryContext(ctx, `
SELECT EXISTS (
  SELECT 1
  FROM information_schema.columns
  WHERE table_schema = current_schema()
    AND table_name = $1
    AND column_name = $2
)`, tableName, columnName)
	if err != nil {
		return false, err
	}
	defer rows.Close()
	if !rows.Next() {
		return false, rows.Err()
	}
	var exists bool
	if err := rows.Scan(&exists); err != nil {
		return false, err
	}
	return exists, rows.Err()
}

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},
		{"005_post_vote_post_id_index", migratePostVoteIndex},
		{"006_post_date", migratePostDate},
		{"007_post_state", migratePostState},
	}
	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()
}