//! [`SagaWorker`] — an autonomous background driver for a saga's
//! time-based maintenance.
//!
//! The host-driven entry points ([`PgSagaRunner::trigger`],
//! [`deliver`](PgSagaRunner::deliver), [`signal`](PgSagaRunner::signal)) are
//! the API for application code (HTTP handlers, projections feeding the
//! saga). But two jobs must happen on a *schedule*, with nobody calling in:
//!
//! - **Timeouts.** A step that suspended awaiting an external response with
//!   a deadline only fails (and compensates) when something ticks the clock.
//!   [`PgSagaRunner::tick_timeouts`] does the work; `SagaWorker` calls it on
//!   a poll interval so the host doesn't have to run its own timer.
//! - **Signal retention.** Consumed `es_saga_signals` rows are pruned on a
//!   slower cadence (SAGA-14) via
//!   [`PgSagaRunner::prune_consumed_signals`].
//!
//! Both underlying primitives are already replica-safe — `tick_timeouts`
//! leases each instance (skipping any held by a peer), and the prune is an
//! idempotent batched delete — so running a `SagaWorker` on every replica is
//! safe; instances are never double-processed.
//!
//! **Sharding (plan §P4.3).** When the runner is configured with
//! `num_shards > 1`, each worker owns a converging subset of shards (via
//! [`shard_lease`](crate::shard_lease)) and scans timeouts only for those —
//! so N workers split the scan work N ways instead of every worker scanning
//! everything. Ownership is *work distribution only*: the per-instance lease
//! still fences effects, so a shard transiently scanned by two workers (mid
//! rebalance) is safe. At the default `num_shards == 1` the worker keeps the
//! original whole-table scan with no shard-lease overhead.
//!
//! **Scope (v1):** this drives time + retention only. The lease *heartbeat*
//! (so a slow handler keeps exclusivity) already lives in the runner.
//! **Lazy recovery** (re-driving instances crashed mid-transition) and the
//! **outbox** (publishing Regime-B emits to external workers) are deferred —
//! see `tasks/phase-3-worker-loop.md`. Until recovery lands, an instance
//! that crashes between its claim and its commit is still healed only by a
//! later external `deliver`/`tick`, not by the worker itself.
//!
//! Like [`mire::ProjectionRunner`], the worker does not install signal
//! handlers — the host wires `SIGTERM`/`Ctrl+C` to the
//! [`CancellationToken`] it passes to [`run`](SagaWorker::run).

use std::time::Duration;

use tokio_util::sync::CancellationToken;
use tracing::instrument;

use crate::pg_runner::{PgRunnerError, PgSagaRunner};
use crate::saga::Saga;

/// Background driver for a saga's scheduled maintenance. Build one around a
/// [`PgSagaRunner`] and drive it with [`run`](Self::run).
pub struct SagaWorker<S: Saga> {
    runner: PgSagaRunner<S>,
    poll_interval: Duration,
    prune_interval: Duration,
    signal_retention: Option<Duration>,
    shard_lease_ttl: Duration,
    recovery_interval: Duration,
    drain_interval: Duration,
}

impl<S: Saga> SagaWorker<S> {
    /// Wrap `runner`. Defaults: poll timeouts every 1s, prune consumed
    /// signals every hour keeping 7 days of history, shard leases valid for
    /// 30s without renewal, recovery scan every 10s re-driving non-terminal
    /// instances whose lease has lapsed (phase 7 — lease-expiry-driven).
    pub fn new(runner: PgSagaRunner<S>) -> Self {
        Self {
            runner,
            poll_interval: Duration::from_secs(1),
            prune_interval: Duration::from_secs(3600),
            signal_retention: Some(Duration::from_secs(7 * 24 * 3600)),
            shard_lease_ttl: Duration::from_secs(30),
            recovery_interval: Duration::from_secs(10),
            drain_interval: Duration::from_millis(500),
        }
    }

    /// How often to drain the outbox of external emits (default 500ms), when
    /// the runner has an [`EmitPublisher`](crate::EmitPublisher). Lower bounds
    /// external-emit delivery latency. No effect without a publisher (the
    /// drain is skipped entirely).
    pub fn drain_interval(mut self, d: Duration) -> Self {
        self.drain_interval = d;
        self
    }

    /// How often to run the lazy-recovery scan (default 10s). Recovery heals
    /// instances stranded by a crashed worker/host; it's not latency-critical
    /// (the deadline tick and live deliveries handle the common path), so a
    /// slower cadence than `poll_interval` is fine.
    pub fn recovery_interval(mut self, d: Duration) -> Self {
        self.recovery_interval = d;
        self
    }

    /// How long a shard-ownership lease (and fleet-membership heartbeat) stays
    /// valid without renewal (default 30s). Renewed every `poll_interval`, so
    /// keep this comfortably above it (≥ a few poll intervals) to tolerate a
    /// missed tick without shedding shards. **Fleet-wide constant** — every
    /// worker must agree, since liveness is judged per the querier's window
    /// (see [`shard_lease::rebalance`](crate::shard_lease::rebalance)). Only
    /// relevant when the runner's `num_shards > 1`.
    pub fn shard_lease_ttl(mut self, d: Duration) -> Self {
        self.shard_lease_ttl = d;
        self
    }

    /// How often to tick timeouts (default 1s). Lower bounds how late an
    /// expired step's compensation fires.
    pub fn poll_interval(mut self, d: Duration) -> Self {
        self.poll_interval = d;
        self
    }

    /// How often to prune consumed signal rows (default 1h).
    pub fn prune_interval(mut self, d: Duration) -> Self {
        self.prune_interval = d;
        self
    }

    /// Keep consumed signals this long before pruning (default 7 days).
    /// `None` disables retention (the table is never pruned).
    pub fn signal_retention(mut self, d: Option<Duration>) -> Self {
        self.signal_retention = d;
        self
    }

    /// The runner this worker drives — useful for inspection/tests.
    pub fn runner(&self) -> &PgSagaRunner<S> {
        &self.runner
    }

    /// Run until `cancel` fires. Ticks timeouts every `poll_interval` and
    /// prunes consumed signals every `prune_interval`. Transient errors are
    /// logged and the loop continues — a single failed tick must not take
    /// the worker down (mirrors `ProjectionRunner`'s resilience). Returns
    /// `Ok(())` on graceful shutdown.
    #[instrument(level = "debug", skip_all, fields(saga = S::name(), num_shards = self.runner.num_shards()))]
    pub async fn run(self, cancel: CancellationToken) -> Result<(), PgRunnerError> {
        // Fail loudly before doing any work if the fleet disagrees on
        // num_shards (a row stamped under more shards than we're configured
        // for): scanning under a too-small shard count would silently strand
        // the out-of-range shards' timeouts/signals (plan §P4.1 startup
        // guard, principle 9 — fail fast, never silently drop work).
        self.runner.verify_shard_config().await?;

        // Sharding is active only when the deployment configured >1 shard.
        // At num_shards == 1 (the host-driven / single-worker default) we keep
        // the original behavior exactly: scan all awaits, no shard-lease or
        // membership writes (plan §P4.1 deviation note). With >1 shard the
        // worker owns a converging subset and scans only those (plan §P4.3).
        let sharded = self.runner.num_shards() > 1;
        // `owned` is refreshed by a rebalance tick before each timeout scan.
        let mut owned: Vec<i32> = Vec::new();

        let mut timeout_tick = tokio::time::interval(self.poll_interval);
        timeout_tick.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip);
        let mut prune_tick = tokio::time::interval(self.prune_interval);
        prune_tick.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip);
        let mut recovery_tick = tokio::time::interval(self.recovery_interval);
        recovery_tick.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip);
        // Only run the outbox drain when a publisher is configured (otherwise
        // outbox rows are intentionally left to accumulate / be polled).
        let drain_enabled = self.runner.has_publisher();
        let mut drain_tick = tokio::time::interval(self.drain_interval);
        drain_tick.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip);
        // Consume the immediate-fire ticks so we don't prune/tick/recover/drain
        // at t=0 before the host has done anything.
        timeout_tick.tick().await;
        prune_tick.tick().await;
        recovery_tick.tick().await;
        drain_tick.tick().await;

        tracing::info!(
            saga = S::name(),
            poll_interval = ?self.poll_interval,
            num_shards = self.runner.num_shards(),
            sharded,
            "saga worker started",
        );

        let result = loop {
            tokio::select! {
                _ = cancel.cancelled() => {
                    tracing::info!(saga = S::name(), "saga worker shutting down");
                    break Ok(());
                }
                _ = timeout_tick.tick() => {
                    // Converge shard ownership first, then scan only what we
                    // own. release_excess = true: shed surplus shards every
                    // tick so a newly-joined worker gets its share within a
                    // tick or two. (Bounded hysteresis to damp membership
                    // flapping is a documented follow-up; over/under-ownership
                    // is always safe — the instance lease is the fence.)
                    let outcome = if sharded {
                        match self.runner.rebalance_shards(
                            self.shard_lease_ttl, self.shard_lease_ttl, true,
                        ).await {
                            Ok(shards) => {
                                owned = shards;
                                self.runner.tick_timeouts_in_shards(chrono::Utc::now(), &owned).await
                            }
                            Err(e) => {
                                // A failed rebalance must not drop the tick: fall
                                // back to scanning the shards we last owned.
                                tracing::warn!(saga = S::name(), error = %e, "shard rebalance failed; scanning last-owned shards");
                                self.runner.tick_timeouts_in_shards(chrono::Utc::now(), &owned).await
                            }
                        }
                    } else {
                        self.runner.tick_timeouts(chrono::Utc::now()).await
                    };
                    match outcome {
                        Ok(n) if n > 0 => tracing::debug!(saga = S::name(), timed_out = n, "ticked timeouts"),
                        Ok(_) => {}
                        Err(e) => tracing::warn!(saga = S::name(), error = %e, "tick_timeouts failed; continuing"),
                    }
                }
                _ = prune_tick.tick() => {
                    if let Some(retention) = self.signal_retention {
                        match self.runner.prune_consumed_signals(retention).await {
                            Ok(n) if n > 0 => tracing::debug!(saga = S::name(), pruned = n, "pruned consumed signals"),
                            Ok(_) => {}
                            Err(e) => tracing::warn!(saga = S::name(), error = %e, "prune_consumed_signals failed; continuing"),
                        }
                    }
                }
                _ = recovery_tick.tick() => {
                    // Re-drive instances stranded by a crashed worker/host
                    // (P3.3). Recovery is lease-expiry-driven (phase 7): a
                    // candidate is a non-terminal, no-await instance with no
                    // live lease, so latency tracks the (short) lease TTL, not
                    // a fixed staleness clock. Scan only owned shards when
                    // sharded so the fleet splits recovery work too; `owned` is
                    // kept fresh by the (more frequent) timeout tick's rebalance.
                    let outcome = if sharded {
                        self.runner.recover_in_shards(&owned).await
                    } else {
                        self.runner.recover().await
                    };
                    match outcome {
                        Ok(n) if n > 0 => tracing::info!(saga = S::name(), recovered = n, "re-drove stranded instances"),
                        Ok(_) => {}
                        Err(e) => tracing::warn!(saga = S::name(), error = %e, "recovery scan failed; continuing"),
                    }
                }
                _ = drain_tick.tick(), if drain_enabled => {
                    // Deliver external (Regime-B) emits via the publisher
                    // (P3.2), only the owned shards when sharded so the fleet
                    // splits the drain too.
                    let outcome = if sharded {
                        self.runner.drain_outbox_in_shards(&owned).await
                    } else {
                        self.runner.drain_outbox().await
                    };
                    match outcome {
                        Ok(n) if n > 0 => tracing::debug!(saga = S::name(), published = n, "drained outbox emits"),
                        Ok(_) => {}
                        Err(e) => tracing::warn!(saga = S::name(), error = %e, "outbox drain failed; continuing"),
                    }
                }
            }
        };

        // Graceful shutdown: relinquish shards + membership so survivors
        // reabsorb them on their next tick rather than waiting out the TTL.
        if sharded && let Err(e) = self.runner.release_shards().await {
            tracing::warn!(saga = S::name(), error = %e, "failed to release shards on shutdown");
        }
        result
    }
}
