#!/usr/bin/env bash
# ---------------------------------------------------------------------------
# run-migrations.sh — Idempotent Alembic migration runner with distributed lock
#
# Uses PostgreSQL advisory lock to prevent concurrent migration execution
# across multiple service instances (e.g. Railway scaling events).
#
# Safe for fresh and already Alembic-managed databases:
#   - If alembic_version table exists and is populated → just run upgrade
#   - If nothing exists → full upgrade from scratch
#   - If application tables exist without a revision → fail for reviewed
#     baseline recovery; never stamp an unknown schema to head
#
# Lock acquisition uses pg_try_advisory_xact_lock with bounded retries so a
# blocked deploy fails LOUDLY (stderr, non-zero exit) within
# MIGRATION_LOCK_TIMEOUT_S seconds instead of hanging silently past Railway's
# healthcheck window. Transaction-scoped locks are intentional here: Supavisor
# and other poolers can strand session-level advisory locks after a client
# disconnect, but PostgreSQL automatically releases transaction locks when the
# transaction ends or the backend dies. See incident 2026-04-17 through
# 2026-04-20.
#
# Usage:  MIGRATIONS_DATABASE_URL=postgresql://... ./scripts/run-migrations.sh
# ---------------------------------------------------------------------------
set -euo pipefail

# Advisory lock ID for migrations (distinct from scheduler lock 4294967295)
MIGRATION_LOCK_ID=4294967294

# Max seconds to wait for the migration lock before giving up with a hard error.
# Railway's default healthcheck retry window is 2 minutes, so we give up at 90 s
# to leave room for the actual migration + app boot after the lock is acquired.
MIGRATION_LOCK_TIMEOUT_S=${MIGRATION_LOCK_TIMEOUT_S:-90}

# Deployed DDL always uses the schema-owner credential.  Keep DATABASE_URL as
# the RLS-enforced application credential; never overwrite or reuse it here.
DEPLOYED_MIGRATION_RUN=false
case "${REGENGINE_ENV:-${ENVIRONMENT:-${ENV:-}}}" in
  production|staging) DEPLOYED_MIGRATION_RUN=true ;;
esac
for marker in RAILWAY_ENVIRONMENT RAILWAY_ENVIRONMENT_NAME RAILWAY_ENVIRONMENT_ID RAILWAY_PROJECT_NAME RAILWAY_PROJECT_ID RAILWAY_SERVICE_NAME RAILWAY_SERVICE_ID VERCEL_ENV VERCEL_URL RENDER FLY_APP_NAME KUBERNETES_SERVICE_HOST; do
  if [ -n "${!marker:-}" ]; then
    DEPLOYED_MIGRATION_RUN=true
    break
  fi
done

if [ -z "${MIGRATIONS_DATABASE_URL:-}" ] && [ "$DEPLOYED_MIGRATION_RUN" = true ]; then
  echo "!! MIGRATIONS_DATABASE_URL is required for deployed migrations." >&2
  echo "!! Refusing to use DATABASE_URL for schema-owner DDL." >&2
  exit 64
fi

MIGRATION_RUNNER_URL="${MIGRATIONS_DATABASE_URL:-${DATABASE_URL:-}}"
export MIGRATION_RUNNER_URL
if [ -z "$MIGRATION_RUNNER_URL" ]; then
  echo "!! MIGRATIONS_DATABASE_URL is required (DATABASE_URL is accepted only for local/CI scratch databases)." >&2
  echo "!! Refusing to fall back to development database defaults in deploy/runtime paths." >&2
  exit 64  # EX_USAGE
fi

# The supported runner is always a dedicated-owner path, including local/CI
# rehearsals. Export the selected URL so v111 can derive and prove the exact
# login identity even when a local caller used the legacy DATABASE_URL input.
export MIGRATIONS_DATABASE_URL="$MIGRATION_RUNNER_URL"
export REGENGINE_ENFORCE_MIGRATION_ROLE=1

MIGRATION_SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
RUNTIME_ROLE_CUTOVER_SCRIPT="${MIGRATION_SCRIPT_DIR}/apply_runtime_role_cutover.py"
export RUNTIME_ROLE_CUTOVER_SCRIPT
if [ ! -f "$RUNTIME_ROLE_CUTOVER_SCRIPT" ]; then
  echo "!! Runtime-role cutover helper is missing: $RUNTIME_ROLE_CUTOVER_SCRIPT" >&2
  exit 66  # EX_NOINPUT
fi

PYTHON_BIN=${PYTHON_BIN:-python3}
if ! command -v "${PYTHON_BIN}" >/dev/null 2>&1; then
  if command -v python >/dev/null 2>&1; then
    PYTHON_BIN=python
  else
    echo "!! Neither python3 nor python is available for migration runner." >&2
    exit 69  # EX_UNAVAILABLE
  fi
fi

echo "==> Acquiring migration advisory lock (id=${MIGRATION_LOCK_ID}, timeout=${MIGRATION_LOCK_TIMEOUT_S}s)..."

# Wrap entire migration in a Python script that holds an advisory transaction
# lock. pg_try_advisory_xact_lock returns immediately with true/false — we poll
# with exponential backoff so a busy holder causes a loud failure, not a silent
# hang.
${PYTHON_BIN} -c "
import os, sys, subprocess, time

from sqlalchemy import create_engine, text
from sqlalchemy.pool import NullPool

url = os.environ['MIGRATION_RUNNER_URL']
# psycopg (v3) is the only sync driver installed; a bare postgresql:// URL
# would make SQLAlchemy load psycopg2. Same rewrite as services/shared/db_url.py.
for prefix in ('postgresql+psycopg2://', 'postgresql+asyncpg://', 'postgresql://', 'postgres://'):
    if url.startswith(prefix):
        url = 'postgresql+psycopg://' + url[len(prefix):]
        break
timeout_s = int(os.environ.get('MIGRATION_LOCK_TIMEOUT_S', '${MIGRATION_LOCK_TIMEOUT_S}'))
# NullPool: closing a connection really disconnects it instead of parking an
# idle backend in the pool for the whole alembic run — the probe connection
# below must be physically gone, not just checked in, before migrations start.
engine = create_engine(url, pool_pre_ping=True, poolclass=NullPool)

conn = engine.connect()
tx = conn.begin()
lock_acquired = False

try:
    # Poll pg_try_advisory_xact_lock with exponential backoff up to timeout_s.
    # This is the key behavior change vs pg_advisory_lock: we never block
    # the container past the healthcheck window — we either get the lock
    # or exit non-zero with a diagnostic so Railway shows a real error.
    deadline = time.monotonic() + timeout_s
    delay = 0.5
    attempt = 0
    while True:
        attempt += 1
        got = conn.execute(text('SELECT pg_try_advisory_xact_lock(${MIGRATION_LOCK_ID})')).scalar()
        if got:
            lock_acquired = True
            print(f'==> Migration lock acquired (attempt {attempt})', flush=True)
            # This backend now sits idle-in-transaction while alembic runs in a
            # subprocess. A server-side idle_in_transaction_session_timeout (or
            # transaction_timeout, PG 17+) would terminate it and silently
            # release the advisory lock mid-migration — disable both for this
            # session. SET takes no relation locks.
            conn.execute(text(\"SET idle_in_transaction_session_timeout = 0\"))
            if int(conn.execute(text('SHOW server_version_num')).scalar()) >= 170000:
                conn.execute(text(\"SET transaction_timeout = 0\"))
            break
        remaining = deadline - time.monotonic()
        if remaining <= 0:
            # Surface the holder so the operator can terminate the stale backend.
            # A 1-arg bigint advisory lock shows its high 32 bits in classid and
            # low 32 bits in objid with objsubid=1; filter on all three plus
            # granted, or waiters and unrelated 2-arg locks get reported (and
            # the remediation below would terminate an innocent backend).
            holder = conn.execute(text(
                \"SELECT a.pid, a.application_name, a.state, a.state_change \"
                \"FROM pg_stat_activity a JOIN pg_locks l ON l.pid = a.pid \"
                \"WHERE l.locktype = 'advisory' AND l.granted \"
                \"  AND l.classid = ((${MIGRATION_LOCK_ID}::bigint >> 32) & 4294967295)::oid \"
                \"  AND l.objid = (${MIGRATION_LOCK_ID}::bigint & 4294967295)::oid \"
                \"  AND l.objsubid = 1\"
            )).fetchall()
            print(
                f'!! Could not acquire migration lock within {timeout_s}s after {attempt} attempts.',
                file=sys.stderr, flush=True,
            )
            print(f'!! Current holder(s): {holder!r}', file=sys.stderr, flush=True)
            print(
                '!! Remediation: connect to Postgres and run '
                'SELECT pg_terminate_backend(pid) for any stale holder shown above, '
                'then retry the Railway deployment.',
                file=sys.stderr, flush=True,
            )
            tx.rollback()
            sys.exit(75)  # EX_TEMPFAIL — Railway will mark deploy failed, ops can retry
        sleep_for = min(delay, remaining)
        print(f'    lock busy, retrying in {sleep_for:.1f}s (attempt {attempt}, {remaining:.0f}s left)', flush=True)
        time.sleep(sleep_for)
        delay = min(delay * 1.5, 5.0)

    # Keep this transaction open while Alembic runs in its own process. The
    # transaction exists only to own the advisory lock, and PostgreSQL releases
    # it automatically on commit, rollback, disconnect, or process death.
    #
    # The lock-holder transaction must never touch application relations:
    # any SELECT here holds ACCESS SHARE until the transaction ends, and a
    # migration that takes ACCESS EXCLUSIVE on the same relation (v092 runs
    # ALTER TABLE public.alembic_version ENABLE ROW LEVEL SECURITY) then
    # waits on this very connection forever — the runner deadlocks itself
    # (2026-07-15 deploy incident). State probes run on a separate
    # connection that is CLOSED before Alembic starts.
    # ---- Detect DB state (dedicated probe connection) ----
    probe = engine.connect()
    try:
        result = probe.execute(text(
            \"SELECT EXISTS (SELECT 1 FROM information_schema.tables WHERE table_name = 'alembic_version')\"
        ))
        has_alembic = result.scalar()

        if not has_alembic:
            result2 = probe.execute(text(
                \"SELECT EXISTS (SELECT 1 FROM information_schema.tables WHERE table_schema = 'fsma')\"
            ))
            state = 'NEEDS_BASELINE_REVIEW' if result2.scalar() else 'FRESH'
        else:
            result3 = probe.execute(text('SELECT count(*) FROM alembic_version'))
            if result3.scalar() > 0:
                state = 'ALREADY_MANAGED'
            else:
                result4 = probe.execute(text(
                    \"SELECT EXISTS (SELECT 1 FROM information_schema.tables WHERE table_schema = 'fsma')\"
                ))
                state = 'NEEDS_BASELINE_REVIEW' if result4.scalar() else 'FRESH'
    finally:
        probe.close()

    print(f'    DB state: {state}', flush=True)

    # ---- Run migrations ----
    if state == 'NEEDS_BASELINE_REVIEW':
        print(
            '!! Existing application schema has no trusted Alembic revision. '
            'Refusing to stamp an unknown schema to head. Restore migration '
            'metadata or execute an operator-reviewed baseline recovery before retrying.',
            file=sys.stderr,
            flush=True,
        )
        raise SystemExit(65)  # EX_DATAERR
    elif state == 'ALREADY_MANAGED':
        print('==> Alembic already managing this database — upgrading...', flush=True)
        subprocess.check_call(['alembic', 'upgrade', 'head'])
    else:
        print('==> Fresh database — running full migration...', flush=True)
        subprocess.check_call(['alembic', 'upgrade', 'head'])

    # Reapply the convergent v111 cutover after every managed migration run so
    # privilege/policy drift is repaired before startup. Pass the already-
    # selected schema-owner URL only to this child process; never replace
    # DATABASE_URL.
    cutover_env = os.environ.copy()
    cutover_env['MIGRATIONS_DATABASE_URL'] = url
    subprocess.check_call(
        [sys.executable, os.environ['RUNTIME_ROLE_CUTOVER_SCRIPT']],
        env=cutover_env,
    )

    print('==> Migrations and runtime-role cutover complete.', flush=True)
    tx.commit()

except BaseException:
    if tx.is_active:
        tx.rollback()
    raise
finally:
    conn.close()
    engine.dispose()
    if lock_acquired:
        print('==> Migration lock transaction closed', flush=True)
"