//! Schema migration for the saga subsystem.
//!
//! The saga schema is additive on top of mire's event store — it adds
//! `es_saga_awaits` (routing), `es_saga_signals` (external signal queue),
//! `es_saga_leases` (per-instance processing leases),
//! `es_saga_shard_leases` (worker-fleet shard ownership), and
//! `es_saga_instances` (per-instance status projection for recovery/ops),
//! and does not modify any existing table.
//!
//! Saga instance state itself lives on mire aggregate streams; see
//! `impl Aggregate for SagaEnvelope<S>` in [`crate::envelope`]. The
//! tables here are projections / queues that complement the stream.
//!
//! ## Usage
//!
//! Call once at host startup, after `EventStore::migrate`:
//!
//! ```rust,no_run
//! # async fn run(store: &mire::EventStore) -> anyhow::Result<()> {
//! store.migrate().await?;            // mire's tables (events, streams, …)
//! mire_sagas::migrate(store).await?; // saga tables (awaits, signals)
//! # Ok(()) }
//! ```
//!
//! Concurrent calls from multiple replicas are safe — the migration
//! takes a transaction-scoped Postgres advisory lock for the duration,
//! same pattern as `EventStore::migrate`.

use mire::{EventStore, EventStoreError};

/// Transaction-scoped advisory lock key for saga schema migration.
/// Distinct from mire's own key (`0x5f6d_6972_655f_6d67`) so the two
/// migrations don't block each other if a host happens to run them in
/// parallel — they protect disjoint table sets.
const MIGRATE_LOCK_KEY: i64 = 0x5f73_6167_615f_6d67_u64 as i64;

/// Apply the saga schema migration. Idempotent — re-running against an
/// already-migrated database is a no-op.
pub async fn migrate(store: &EventStore) -> Result<(), EventStoreError> {
    let mut tx = store.pool().begin().await?;
    sqlx::query("SELECT pg_advisory_xact_lock($1)")
        .bind(MIGRATE_LOCK_KEY)
        .execute(&mut *tx)
        .await?;
    // Arm a bounded `lock_timeout` for the DDL below (transaction-local),
    // acquired AFTER the advisory lock so concurrent migrators still
    // serialise cleanly. A `CREATE INDEX` / `ALTER TABLE` that can't take
    // its table lock — because a live writer holds a conflicting row lock —
    // then fails fast with a retryable `DbErrorKind::LockTimeout` rather
    // than blocking forever behind a connection Postgres sees as idle.
    // Mirrors `mire::EventStore::migrate`; run once at boot regardless.
    sqlx::query("SELECT set_config('lock_timeout', '15000', true)")
        .execute(&mut *tx)
        .await?;
    sqlx::raw_sql(include_str!("../migrations/001_sagas.sql"))
        .execute(&mut *tx)
        .await?;
    // 002: per-instance processing leases (effects-vs-commit §11). Applied
    // under the same advisory lock; append-only, so re-running is a no-op.
    sqlx::raw_sql(include_str!("../migrations/002_saga_leases.sql"))
        .execute(&mut *tx)
        .await?;
    // 003: early-signal buffering — adds `consumed_at` to es_saga_signals
    // and the pending-signal index (review SAGA-10). Append-only.
    sqlx::raw_sql(include_str!("../migrations/003_saga_signal_consume.sql"))
        .execute(&mut *tx)
        .await?;
    // 004: shard-ownership leases for the worker fleet (plan §P4.2) — lets
    // workers partition scans across shards. Append-only; coordination only,
    // not a correctness fence (the instance lease + OCC remain those).
    sqlx::raw_sql(include_str!("../migrations/004_saga_shard_leases.sql"))
        .execute(&mut *tx)
        .await?;
    // 005: per-instance status projection (plan §P3.1) — the substrate for
    // lazy recovery + ops dashboards. Append-only; maintained transactionally
    // inside commit() alongside the envelope append.
    sqlx::raw_sql(include_str!("../migrations/005_saga_instances.sql"))
        .execute(&mut *tx)
        .await?;
    // 006: outbox for at-least-once external-emit delivery (plan §P3.2).
    // Rows written in the same tx as the StepEmitted/await; drained by a
    // SagaWorker's EmitPublisher. Append-only.
    sqlx::raw_sql(include_str!("../migrations/006_saga_outbox.sql"))
        .execute(&mut *tx)
        .await?;
    // 007: created_at on awaits so a timeout reports the configured budget
    // (plan §P5.4). Append-only idempotent ALTER.
    sqlx::raw_sql(include_str!("../migrations/007_saga_awaits_created_at.sql"))
        .execute(&mut *tx)
        .await?;
    // 008: fence token on the per-instance lease (phase 7). Append-only
    // idempotent ALTER; makes lease steals explicit so the TTL can be short.
    sqlx::raw_sql(include_str!("../migrations/008_saga_lease_fence.sql"))
        .execute(&mut *tx)
        .await?;
    // 009: partial index over live instances (phase 7) so lease-expiry
    // recovery's scan stays O(live) as terminal rows accumulate.
    sqlx::raw_sql(include_str!("../migrations/009_saga_instances_live.sql"))
        .execute(&mut *tx)
        .await?;
    tx.commit().await?;
    Ok(())
}
