//! Per-instance processing leases (effects-vs-commit §11).
//!
//! Guarantees a saga instance's effects run **at-least-once but never
//! concurrently**: a worker must hold the `(saga_name, instance_id)` lease to
//! process the instance. A second worker that finds a live lease held by
//! someone else backs off ([`AcquireOutcome::Contended`]); when the holder
//! dies its `leased_until` lapses and another worker takes over and re-drives.
//!
//! The lease gives mutual exclusion only. Stale-commit fencing is the event
//! store's existing OCC: a takeover worker advances the instance's stream
//! version, so a preempted holder's later commit conflicts and is discarded —
//! "if a worker no longer holds the lease, its effect is invalid." No separate
//! fence token is needed (unlike `mire`'s projection leases) because saga
//! state IS a version-checked event stream.

use std::time::Duration;

use mire::EventStoreError;
use sqlx::{PgPool, Postgres};
use tracing::trace;

/// Outcome of [`acquire`].
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum AcquireOutcome {
    /// This worker now holds the lease (freshly acquired or stolen from an
    /// expired holder). Carries the lease's **fence** — bumped on every steal,
    /// so the holder can detect (via [`heartbeat`]) that a peer has taken over,
    /// and so a same-`worker_id` restart can't masquerade as its own stale
    /// lease. Pass it to [`heartbeat`] / [`release`].
    Acquired { fence: i64 },
    /// A different, still-live worker holds it. The caller must not process
    /// the instance; back off and let the holder (or a later takeover) do it.
    Contended,
}

/// Try to acquire (or steal-if-expired) the processing lease for
/// `(saga_name, instance_id)`, valid for `ttl`. A single statement: the
/// `ON CONFLICT ... WHERE` clause is the mutual-exclusion fence — two racing
/// acquirers serialize on the PK row lock and the loser re-evaluates
/// `leased_until < now()` against the winner's fresh row, returning 0 rows.
///
/// **Strict exclusion (no same-worker bypass):** acquisition succeeds *only*
/// when the existing lease has expired (`leased_until < now()`) or there is no
/// row. Unlike `mire`'s projection lease, we do **not** let the same
/// `worker_id` re-acquire a still-live lease — otherwise two concurrent calls
/// on one runner (which share a worker_id) would both "acquire" and process
/// the instance simultaneously, defeating the whole point. Each
/// `trigger`/`deliver`/`tick` call acquires once and releases; there is no
/// in-call re-acquire, so strictness costs nothing. A crashed worker reusing
/// the same id after restart simply waits out its own stale lease's TTL.
pub async fn acquire(
    pool: &PgPool,
    saga_name: &str,
    instance_id: &str,
    worker_id: &str,
    ttl: Duration,
) -> Result<AcquireOutcome, EventStoreError> {
    let ttl_secs = ttl.as_secs_f64().to_string();
    // A fresh lease starts at fence 1; stealing an expired one bumps the
    // existing fence (`es_saga_leases.fence + 1`). `RETURNING fence` hands the
    // acquirer the value it must heartbeat/release under.
    let row = sqlx::query_scalar::<_, i64>(
        "INSERT INTO es_saga_leases
            (saga_name, instance_id, worker_id, leased_until, heartbeat_at, fence)
         VALUES ($1, $2, $3, now() + ($4 || ' seconds')::interval, now(), 1)
         ON CONFLICT (saga_name, instance_id) DO UPDATE
            SET worker_id    = EXCLUDED.worker_id,
                leased_until = EXCLUDED.leased_until,
                heartbeat_at = now(),
                fence        = es_saga_leases.fence + 1
          WHERE es_saga_leases.leased_until < now()
         RETURNING fence",
    )
    .bind(saga_name)
    .bind(instance_id)
    .bind(worker_id)
    .bind(ttl_secs)
    .fetch_optional(pool)
    .await?;
    Ok(match row {
        Some(fence) => {
            trace!(
                saga_name,
                instance_id, worker_id, fence, "instance lease acquired (fresh or stolen)",
            );
            AcquireOutcome::Acquired { fence }
        }
        None => {
            trace!(
                saga_name,
                instance_id, worker_id, "instance lease contended (live holder)",
            );
            AcquireOutcome::Contended
        }
    })
}

/// Extend our lease deadline. Returns `Ok(true)` if the lease is still ours
/// (1 row updated), `Ok(false)` if we've been preempted — either a peer stole
/// it (the row now has a higher `fence`) or took it over under a different
/// `worker_id`. The `fence` guard is what makes this robust to a crashed
/// worker that restarts under the same `worker_id`: its old in-flight task
/// heartbeats the *old* fence and correctly sees it has lost the lease.
pub async fn heartbeat<'e, E>(
    executor: E,
    saga_name: &str,
    instance_id: &str,
    worker_id: &str,
    fence: i64,
    ttl: Duration,
) -> Result<bool, EventStoreError>
where
    E: sqlx::Executor<'e, Database = Postgres>,
{
    let ttl_secs = ttl.as_secs_f64().to_string();
    let affected = sqlx::query(
        "UPDATE es_saga_leases
            SET leased_until = now() + ($1 || ' seconds')::interval,
                heartbeat_at = now()
          WHERE saga_name = $2 AND instance_id = $3 AND worker_id = $4 AND fence = $5",
    )
    .bind(ttl_secs)
    .bind(saga_name)
    .bind(instance_id)
    .bind(worker_id)
    .bind(fence)
    .execute(executor)
    .await?
    .rows_affected();
    let ok = affected == 1;
    if ok {
        trace!(
            saga_name,
            instance_id, worker_id, fence, "instance lease heartbeat ok",
        );
    } else {
        trace!(
            saga_name,
            instance_id,
            worker_id,
            fence,
            "instance lease heartbeat: preempted (fence no longer matches)",
        );
    }
    Ok(ok)
}

/// Release the lease on graceful completion so a peer can process the instance
/// immediately rather than waiting out the TTL. Best-effort: only frees the
/// row if we still hold it *at this fence* (guards against releasing a lease a
/// takeover worker now owns — including a same-`worker_id` restart, caught by
/// the fence).
pub async fn release<'e, E>(
    executor: E,
    saga_name: &str,
    instance_id: &str,
    worker_id: &str,
    fence: i64,
) -> Result<(), EventStoreError>
where
    E: sqlx::Executor<'e, Database = Postgres>,
{
    sqlx::query(
        "UPDATE es_saga_leases
            SET leased_until = '-infinity'
          WHERE saga_name = $1 AND instance_id = $2 AND worker_id = $3 AND fence = $4",
    )
    .bind(saga_name)
    .bind(instance_id)
    .bind(worker_id)
    .bind(fence)
    .execute(executor)
    .await?;
    trace!(
        saga_name,
        instance_id, worker_id, fence, "instance lease released",
    );
    Ok(())
}
