mire 0.1.1

//! Continuous projections: poll the event log and fold events into read models.
//!
//! Each registered subscription runs as its own tokio task inside a private
//! supervisor. The supervisor:
//!
//! - Cascades shutdown across all lease loops when the
//!   [`CancellationToken`] passed to [`run`] fires, or when any single
//!   subscription returns (success or poison). Peers get a configurable
//!   `shutdown_grace` to release their leases before being aborted.
//! - Holds each lease via the [`crate::lease`] primitives so at most one
//!   replica drives any given subscription at a time. Per-batch fenced
//!   checkpointing (with best-effort flushes on cancel and poison)
//!   guarantees a paused leader cannot advance the persisted cursor
//!   past events the new leader hasn't processed. Crash window is at
//!   most one batch of idempotent re-processing.
//!
//! ## Signals, supervisors, and the wrapper pattern
//!
//! The runner deliberately does **not** install `SIGTERM` / `Ctrl+C`
//! handlers or impose a supervisor dependency — that's a policy choice for
//! the host application. Callers wire those up themselves and pass a
//! `CancellationToken` that fires on the relevant signal.
//!
//! For applications that use [`notmad`](https://docs.rs/notmad), the
//! recommended pattern is a thin wrapper in the application crate so the
//! application controls its own `notmad` version:
//!
//! ```rust,ignore
//! use mire::ProjectionRunner;
//! use notmad::{Component, MadError};
//! use tokio_util::sync::CancellationToken;
//!
//! pub struct ManagedProjections(pub ProjectionRunner);
//!
//! impl Component for ManagedProjections {
//!     async fn run(&self, cancel: CancellationToken) -> Result<(), MadError> {
//!         self.0.run(cancel).await.map_err(MadError::Inner)
//!     }
//! }
//! ```
//!
//! [`run`]: ProjectionRunner::run

use std::collections::HashSet;
use std::sync::Arc;
use std::time::Duration;

use serde::de::DeserializeOwned;
use sqlx::PgPool;
use sqlx::postgres::Postgres;
use tokio::task::JoinSet;
use tokio_util::sync::CancellationToken;
use uuid::Uuid;

use crate::lease::{self, AcquireOutcome, LeaseStatus};
use crate::{Aggregate, EventData, EventStore, EventStoreError, RecordedEvent, Subscription};

/// Default grace period: when a lease loop returns (or is cancelled), peers
/// get this long to flush checkpoints and release their leases before the
/// supervisor force-cancels them. Override via
/// [`ProjectionRunnerBuilder::shutdown_grace`].
const DEFAULT_SHUTDOWN_GRACE: Duration = Duration::from_secs(10);

/// Envelope handed to an [`EventHandler`]: a typed, already-decoded event
/// plus the full [`RecordedEvent`] it came from.
///
/// Both fields are owned — the runner just deserialised the event from
/// `recorded.data`, so the handler is free to move parts of either into
/// downstream futures without cloning.
pub struct HandledEvent<E: EventData> {
    pub event: E,
    pub recorded: RecordedEvent,
}

impl<E: EventData> HandledEvent<E> {
    pub fn global_position(&self) -> i64 {
        self.recorded.global_position
    }

    pub fn stream_id(&self) -> &str {
        &self.recorded.stream_id
    }

    pub fn stream_version(&self) -> i64 {
        self.recorded.stream_version
    }

    pub fn event_type(&self) -> &str {
        &self.recorded.event_type
    }

    pub fn metadata(&self) -> &serde_json::Value {
        &self.recorded.metadata
    }

    pub fn transaction_id(&self) -> u64 {
        self.recorded.transaction_id
    }

    pub fn created_at(&self) -> chrono::DateTime<chrono::Utc> {
        self.recorded.created_at
    }
}

/// Handles a single event for an aggregate's stream, updating a read model.
/// The runner decodes `recorded.data` into the aggregate's event type before
/// invoking `handle`, so handlers see typed events directly. Returning `Err`
/// triggers a retry; persistent failure (past `max_attempts`) stops the
/// runner.
///
/// The aggregate is named once, as the associated type — call sites
/// (`.subscribe("sub-id", handler)`) need no turbofish, and handler
/// signatures don't carry a lifetime parameter.
pub trait EventHandler: Send + Sync + 'static {
    type Aggregate: Aggregate;

    fn handle(
        &self,
        event: HandledEvent<<Self::Aggregate as Aggregate>::Event>,
    ) -> impl std::future::Future<Output = anyhow::Result<()>> + Send;
}

/// Object-safe shim so handlers of different concrete types can live together
/// in one runner.
trait EventHandlerDyn: Send + Sync {
    fn handle_dyn<'a>(
        &'a self,
        event: &'a RecordedEvent,
    ) -> std::pin::Pin<Box<dyn std::future::Future<Output = anyhow::Result<()>> + Send + 'a>>;
}

struct TypedAdapter<H: EventHandler> {
    inner: H,
}

impl<H> EventHandlerDyn for TypedAdapter<H>
where
    H: EventHandler,
    <H::Aggregate as Aggregate>::Event: DeserializeOwned,
{
    fn handle_dyn<'a>(
        &'a self,
        recorded: &'a RecordedEvent,
    ) -> std::pin::Pin<Box<dyn std::future::Future<Output = anyhow::Result<()>> + Send + 'a>> {
        Box::pin(async move {
            let event =
                serde_json::from_value::<<H::Aggregate as Aggregate>::Event>(recorded.data.clone())
                    .map_err(|source| EventStoreError::Deserialization {
                        stream_id: recorded.stream_id.clone(),
                        global_position: recorded.global_position,
                        event_type: recorded.event_type.clone(),
                        source,
                    })?;
            self.inner
                .handle(HandledEvent {
                    event,
                    recorded: recorded.clone(),
                })
                .await
        })
    }
}

struct CategorySubscription {
    category: String,
    subscription_id: String,
    handler: Arc<dyn EventHandlerDyn>,
}

/// Builds a [`ProjectionRunner`].
pub struct ProjectionRunnerBuilder {
    event_store: EventStore,
    subscriptions: Vec<CategorySubscription>,
    poll_interval: Duration,
    batch_size: i64,
    max_attempts: u32,
    worker_id: String,
    lease_ttl: Duration,
    lease_recheck_interval: Duration,
    shutdown_grace: Duration,
}

impl ProjectionRunnerBuilder {
    /// Register `handler`, tracked under the durable checkpoint
    /// `subscription_id`. The stream category is derived from the handler's
    /// [`Aggregate::stream_category`] — there is no way to point the handler
    /// at the wrong category by mistake.
    pub fn subscribe<H>(mut self, subscription_id: impl Into<String>, handler: H) -> Self
    where
        H: EventHandler,
        <H::Aggregate as Aggregate>::Event: DeserializeOwned,
    {
        self.subscriptions.push(CategorySubscription {
            category: H::Aggregate::stream_category().to_string(),
            subscription_id: subscription_id.into(),
            handler: Arc::new(TypedAdapter { inner: handler }),
        });
        self
    }

    /// How long to sleep when every subscription is idle (default 100ms).
    pub fn poll_interval(mut self, interval: Duration) -> Self {
        self.poll_interval = interval;
        self
    }

    /// How many events to fetch per poll (default 100).
    pub fn batch_size(mut self, batch_size: i64) -> Self {
        self.batch_size = batch_size;
        self
    }

    /// How many times to attempt a handler before giving up (default 3).
    /// Once exceeded, the subscription's loop stops and the supervisor
    /// cascades shutdown to peers.
    pub fn max_attempts(mut self, max_attempts: u32) -> Self {
        self.max_attempts = max_attempts;
        self
    }

    /// Identity for this runner instance, recorded as the `worker_id` in
    /// the lease table. Defaults to a freshly generated UUIDv4 per builder
    /// construction. Override to a stable string (`POD_NAME`, hostname,
    /// etc.) when you want leadership to follow a specific replica
    /// identity across restarts.
    ///
    /// Two replicas configured with the same `worker_id` will both believe
    /// they are leaders for any lease they touch — treat the id as a
    /// strict per-replica identifier.
    pub fn worker_id(mut self, id: impl Into<String>) -> Self {
        self.worker_id = id.into();
        self
    }

    /// How long an acquired lease is valid without a heartbeat. Default
    /// 30s. The heartbeat fires at `lease_ttl / 3`.
    pub fn lease_ttl(mut self, ttl: Duration) -> Self {
        self.lease_ttl = ttl;
        self
    }

    /// How long a non-leader replica sleeps between lease acquisition
    /// attempts. Default 5s. Failover from a `SIGKILL`ed leader is
    /// bounded by `lease_ttl + lease_recheck_interval`.
    pub fn lease_recheck_interval(mut self, d: Duration) -> Self {
        self.lease_recheck_interval = d;
        self
    }

    /// Maximum time the supervisor waits for in-flight lease loops to
    /// release their leases on shutdown. Default 10s — beyond this the
    /// loops are force-cancelled and other replicas have to wait out the
    /// remaining lease TTL.
    pub fn shutdown_grace(mut self, d: Duration) -> Self {
        self.shutdown_grace = d;
        self
    }

    /// Panics if two subscriptions share a `subscription_id` — a
    /// checkpoint collision that would silently corrupt projection progress
    /// at runtime. Also emits a `tracing::warn` if the configured
    /// `sqlx` pool is too small for the registered subscription count
    /// (each subscription needs one held connection for heartbeat plus
    /// headroom for handler work).
    pub fn build(self) -> ProjectionRunner {
        let mut seen = HashSet::with_capacity(self.subscriptions.len());
        for sub in &self.subscriptions {
            if !seen.insert(sub.subscription_id.as_str()) {
                panic!(
                    "duplicate subscription_id `{}` registered with ProjectionRunner — \
                     two handlers would race on the same checkpoint row",
                    sub.subscription_id,
                );
            }
        }

        // Pool sizing sanity check. Each active lease holds one
        // connection for its heartbeat task; the remaining pool serves
        // polls + handler queries. Empirically, under-sized pools cap
        // throughput and amplify tail latency (B7/k=64 in the bench).
        // Recommend ≥ 2× subscription count.
        let n_subs = self.subscriptions.len();
        let pool_max = self.event_store.pool().options().get_max_connections() as usize;
        let recommended = n_subs.saturating_mul(2).max(n_subs + 4);
        if pool_max < recommended {
            tracing::warn!(
                pool_max_connections = pool_max,
                subscriptions = n_subs,
                recommended_minimum = recommended,
                "ProjectionRunner: connection pool may be undersized. Each active \
                 lease holds 1 connection for heartbeat; the rest serve polls and \
                 handler queries. Consider PgPoolOptions::new().max_connections(N) \
                 with N >= subscriptions × 2.",
            );
        }

        ProjectionRunner {
            event_store: self.event_store,
            subscriptions: self.subscriptions,
            poll_interval: self.poll_interval,
            batch_size: self.batch_size,
            max_attempts: self.max_attempts,
            worker_id: self.worker_id,
            lease_ttl: self.lease_ttl,
            lease_recheck_interval: self.lease_recheck_interval,
            shutdown_grace: self.shutdown_grace,
        }
    }
}

/// Long-running supervisor for projections. Drive with [`run`] from any
/// host supervisor; for `notmad`-based apps, see the module-level
/// wrapper example.
///
/// [`run`]: Self::run
pub struct ProjectionRunner {
    event_store: EventStore,
    subscriptions: Vec<CategorySubscription>,
    poll_interval: Duration,
    batch_size: i64,
    max_attempts: u32,
    worker_id: String,
    lease_ttl: Duration,
    lease_recheck_interval: Duration,
    shutdown_grace: Duration,
}

impl ProjectionRunner {
    /// Start building a runner bound to `event_store`.
    pub fn builder(event_store: EventStore) -> ProjectionRunnerBuilder {
        ProjectionRunnerBuilder {
            event_store,
            subscriptions: Vec::new(),
            poll_interval: Duration::from_millis(100),
            batch_size: 100,
            max_attempts: 3,
            worker_id: Uuid::new_v4().to_string(),
            lease_ttl: Duration::from_secs(30),
            lease_recheck_interval: Duration::from_secs(5),
            shutdown_grace: DEFAULT_SHUTDOWN_GRACE,
        }
    }

    /// Spawn each subscription as a tokio task and run them under a
    /// private supervisor until either:
    /// - `cancellation_token` fires (graceful shutdown), or
    /// - any subscription returns (poison or graceful) — cascades
    ///   cancellation to peers with `shutdown_grace` to release leases.
    ///
    /// Returns `Err` if any subscription failed; `Ok(())` on clean
    /// shutdown. Signal handling (`SIGTERM`, `Ctrl+C`) is the caller's
    /// responsibility — wire them to `cancellation_token` in the host
    /// application. See the module docs for a `notmad`-wrapper example.
    pub async fn run(&self, cancellation_token: CancellationToken) -> anyhow::Result<()> {
        // Internal token feeds every lease loop. Bridges to the external
        // token so callers don't need to think about peer cancellation:
        // their token fires → ours fires → all loops drain.
        let internal_cancel = CancellationToken::new();
        let bridge = {
            let internal = internal_cancel.clone();
            let external = cancellation_token.clone();
            tokio::spawn(async move {
                external.cancelled().await;
                internal.cancel();
            })
        };

        let mut joinset: JoinSet<LeaseLoopResult> = JoinSet::new();
        for cat_sub in &self.subscriptions {
            let work = LeaseLoopWork {
                event_store: self.event_store.clone(),
                category: cat_sub.category.clone(),
                subscription_id: cat_sub.subscription_id.clone(),
                handler: Arc::clone(&cat_sub.handler),
                worker_id: self.worker_id.clone(),
                lease_ttl: self.lease_ttl,
                lease_recheck_interval: self.lease_recheck_interval,
                poll_interval: self.poll_interval,
                batch_size: self.batch_size,
                max_attempts: self.max_attempts,
            };
            let cancel = internal_cancel.clone();
            joinset.spawn(work.run(cancel));
        }

        // Wait for the *first* completion. Any completion — graceful or
        // poisoned — cascades cancellation to peers. The aggregator below
        // then drains them with the configured grace period.
        let mut first_err: Option<anyhow::Error> = None;
        if let Some(joined) = joinset.join_next().await {
            match joined {
                Ok(LeaseLoopResult {
                    subscription_id,
                    outcome: Ok(()),
                }) => {
                    tracing::debug!(
                        subscription = %subscription_id,
                        "lease loop completed; cascading shutdown",
                    );
                }
                Ok(LeaseLoopResult {
                    subscription_id,
                    outcome: Err(e),
                }) => {
                    tracing::error!(
                        subscription = %subscription_id,
                        error = %e,
                        "lease loop failed; cascading shutdown",
                    );
                    first_err = Some(e);
                }
                Err(join_err) => {
                    tracing::error!(error = ?join_err, "lease loop task panicked");
                    first_err = Some(anyhow::anyhow!("lease loop task panicked: {join_err}"));
                }
            }
        }
        internal_cancel.cancel();

        // Drain remaining lease loops within shutdown_grace. The grace
        // period exists so workers can call lease::release before exiting,
        // which lets peer replicas take over without waiting out the TTL.
        let drain = async {
            while let Some(joined) = joinset.join_next().await {
                match joined {
                    Ok(LeaseLoopResult {
                        subscription_id,
                        outcome: Ok(()),
                    }) => {
                        tracing::debug!(
                            subscription = %subscription_id,
                            "lease loop drained",
                        );
                    }
                    Ok(LeaseLoopResult {
                        subscription_id,
                        outcome: Err(e),
                    }) => {
                        tracing::warn!(
                            subscription = %subscription_id,
                            error = %e,
                            "lease loop drained with error",
                        );
                        if first_err.is_none() {
                            first_err = Some(e);
                        }
                    }
                    Err(join_err) => {
                        tracing::error!(error = ?join_err, "lease loop drain panicked");
                        if first_err.is_none() {
                            first_err = Some(anyhow::anyhow!(
                                "lease loop task panicked during drain: {join_err}"
                            ));
                        }
                    }
                }
            }
        };

        if tokio::time::timeout(self.shutdown_grace, drain)
            .await
            .is_err()
        {
            tracing::warn!(
                grace = ?self.shutdown_grace,
                remaining = joinset.len(),
                "lease loops did not drain within grace period; aborting",
            );
            joinset.abort_all();
            while joinset.join_next().await.is_some() {}
        }

        bridge.abort();

        match first_err {
            Some(e) => Err(e),
            None => Ok(()),
        }
    }

    /// The `worker_id` baked into this runner's lease acquisitions.
    pub fn worker_id(&self) -> &str {
        &self.worker_id
    }

    /// Inspect the current lease row for `subscription_id`. Returns
    /// `None` if the subscription has never been acquired (no row in
    /// `es_projection_leases`).
    pub async fn lease_status(
        &self,
        subscription_id: &str,
    ) -> Result<Option<LeaseStatus>, EventStoreError> {
        lease::status(self.event_store.pool(), subscription_id).await
    }
}

// ============================================================
// Per-subscription lease loop.
// ============================================================

struct LeaseLoopWork {
    event_store: EventStore,
    category: String,
    subscription_id: String,
    handler: Arc<dyn EventHandlerDyn>,
    worker_id: String,
    lease_ttl: Duration,
    lease_recheck_interval: Duration,
    poll_interval: Duration,
    batch_size: i64,
    max_attempts: u32,
}

struct LeaseLoopResult {
    subscription_id: String,
    outcome: anyhow::Result<()>,
}

impl LeaseLoopWork {
    async fn run(self, cancel: CancellationToken) -> LeaseLoopResult {
        let outcome = lease_loop(
            &self.event_store,
            &self.category,
            &self.subscription_id,
            self.handler.as_ref(),
            &self.worker_id,
            self.lease_ttl,
            self.lease_recheck_interval,
            self.poll_interval,
            self.batch_size,
            self.max_attempts,
            cancel,
        )
        .await;
        LeaseLoopResult {
            subscription_id: self.subscription_id,
            outcome,
        }
    }
}

/// How a single inner-loop pass ended. Drives the outer loop's next move.
enum InnerExit {
    /// Outer cancel fired. Graceful shutdown.
    Cancelled,
    /// Heartbeat or checkpoint detected we'd lost the lease.
    Fenced,
    /// Handler failed `max_attempts` consecutive times — the subscription
    /// is poisoned and we cascade up.
    PoisonHandler(anyhow::Error),
    /// Transient DB / sqlx error during poll or checkpoint. Outer loop
    /// releases the lease, sleeps, and re-acquires.
    Transient(EventStoreError),
}

#[allow(clippy::too_many_arguments)]
async fn lease_loop(
    event_store: &EventStore,
    category: &str,
    subscription_id: &str,
    handler: &dyn EventHandlerDyn,
    worker_id: &str,
    lease_ttl: Duration,
    lease_recheck_interval: Duration,
    poll_interval: Duration,
    batch_size: i64,
    max_attempts: u32,
    cancel: CancellationToken,
) -> anyhow::Result<()> {
    let pool = event_store.pool().clone();
    let heartbeat_interval = lease_ttl / 3;

    loop {
        if cancel.is_cancelled() {
            return Ok(());
        }

        let fence_token =
            match lease::try_acquire(&pool, subscription_id, worker_id, lease_ttl).await {
                Ok(AcquireOutcome::Acquired { fence_token }) => fence_token,
                Ok(AcquireOutcome::Held) => {
                    tokio::select! {
                        _ = tokio::time::sleep(lease_recheck_interval) => continue,
                        _ = cancel.cancelled() => return Ok(()),
                    }
                }
                Err(e) => {
                    tracing::warn!(
                        subscription = %subscription_id,
                        worker = %worker_id,
                        error = %e,
                        "lease acquire failed; backing off",
                    );
                    tokio::select! {
                        _ = tokio::time::sleep(lease_recheck_interval) => continue,
                        _ = cancel.cancelled() => return Ok(()),
                    }
                }
            };

        tracing::info!(
            subscription = %subscription_id,
            worker = %worker_id,
            fence_token,
            "acquired projection lease",
        );

        // Held heartbeat connection — never returned to the pool while we
        // own the lease, so a saturated pool can't starve us into a
        // self-fence.
        let hb_conn = match pool.acquire().await {
            Ok(c) => c,
            Err(e) => {
                tracing::warn!(
                    subscription = %subscription_id,
                    error = %e,
                    "could not acquire heartbeat connection; releasing lease",
                );
                let _ = lease::release(&pool, subscription_id, worker_id, fence_token).await;
                tokio::select! {
                    _ = tokio::time::sleep(lease_recheck_interval) => continue,
                    _ = cancel.cancelled() => return Ok(()),
                }
            }
        };

        let fence_signal = CancellationToken::new();
        let hb_handle = {
            let sub_id = subscription_id.to_string();
            let worker = worker_id.to_string();
            let fence_signal = fence_signal.clone();
            let cancel = cancel.clone();
            tokio::spawn(heartbeat_task(
                hb_conn,
                sub_id,
                worker,
                fence_token,
                lease_ttl,
                heartbeat_interval,
                fence_signal,
                cancel,
            ))
        };

        // Fresh Subscription per acquisition — reads the cursor from DB so
        // we resume from the previous leader's last successful checkpoint.
        let subscription_result = Subscription::create(
            event_store.clone(),
            pool.clone(),
            subscription_id,
            batch_size,
        )
        .await;

        let exit = match subscription_result {
            Ok(mut subscription) => {
                inner_loop(
                    &pool,
                    subscription_id,
                    category,
                    handler,
                    &mut subscription,
                    fence_token,
                    poll_interval,
                    max_attempts,
                    cancel.clone(),
                    fence_signal.clone(),
                )
                .await
            }
            Err(e) => InnerExit::Transient(e),
        };

        // Stop the heartbeat and let it drop its connection.
        fence_signal.cancel();
        let _ = hb_handle.await;

        match exit {
            InnerExit::Cancelled => {
                let released = lease::release(&pool, subscription_id, worker_id, fence_token)
                    .await
                    .unwrap_or(false);
                tracing::info!(
                    subscription = %subscription_id,
                    fence_token,
                    released,
                    "released projection lease on shutdown",
                );
                return Ok(());
            }
            InnerExit::Fenced => {
                tracing::info!(
                    subscription = %subscription_id,
                    fence_token,
                    "projection lease fenced; re-acquiring",
                );
                continue;
            }
            InnerExit::PoisonHandler(err) => {
                let _ = lease::release(&pool, subscription_id, worker_id, fence_token).await;
                tracing::error!(
                    subscription = %subscription_id,
                    error = %err,
                    "subscription stopped: handler failed past max_attempts",
                );
                return Err(err);
            }
            InnerExit::Transient(e) => {
                let _ = lease::release(&pool, subscription_id, worker_id, fence_token).await;
                tracing::warn!(
                    subscription = %subscription_id,
                    error = %e,
                    "transient lease-loop error; backing off then re-acquiring",
                );
                tokio::select! {
                    _ = tokio::time::sleep(lease_recheck_interval) => continue,
                    _ = cancel.cancelled() => return Ok(()),
                }
            }
        }
    }
}

#[allow(clippy::too_many_arguments)]
async fn inner_loop(
    pool: &PgPool,
    subscription_id: &str,
    category: &str,
    handler: &dyn EventHandlerDyn,
    subscription: &mut Subscription,
    fence_token: i64,
    poll_interval: Duration,
    max_attempts: u32,
    cancel: CancellationToken,
    fence_signal: CancellationToken,
) -> InnerExit {
    loop {
        if cancel.is_cancelled() {
            return InnerExit::Cancelled;
        }
        if fence_signal.is_cancelled() {
            return InnerExit::Fenced;
        }

        let events = match subscription.poll_category(category).await {
            Ok(e) => e,
            Err(e) => return InnerExit::Transient(e),
        };

        if events.is_empty() {
            tokio::select! {
                _ = tokio::time::sleep(poll_interval) => continue,
                _ = cancel.cancelled() => return InnerExit::Cancelled,
                _ = fence_signal.cancelled() => return InnerExit::Fenced,
            }
        }

        // Per-batch (not per-event) fenced checkpoint. `last_done` tracks
        // the position of the last successfully-dispatched event so any
        // mid-batch exit can best-effort-flush progress before returning.
        // Crash window: at most one batch of idempotent re-processing,
        // which the at-least-once handler contract already covers.
        let mut last_done: Option<(i64, u64)> = None;

        for event in &events {
            if let Err(e) =
                dispatch_with_retries(handler, event, max_attempts, subscription_id).await
            {
                // Flush progress through the last successful event before
                // poisoning, so the next leader doesn't redo work we
                // already committed.
                if let Some((pos, txid)) = last_done {
                    let _ = lease::checkpoint(pool, subscription_id, fence_token, pos, txid).await;
                }
                return InnerExit::PoisonHandler(e);
            }

            last_done = Some((event.global_position, event.transaction_id));

            if cancel.is_cancelled() {
                if let Some((pos, txid)) = last_done {
                    let _ = lease::checkpoint(pool, subscription_id, fence_token, pos, txid).await;
                }
                return InnerExit::Cancelled;
            }
            if fence_signal.is_cancelled() {
                // Can't checkpoint — our fence token is stale by definition.
                return InnerExit::Fenced;
            }
        }

        // End-of-batch fenced checkpoint.
        if let Some((pos, txid)) = last_done {
            let ok = match lease::checkpoint(pool, subscription_id, fence_token, pos, txid).await {
                Ok(b) => b,
                Err(e) => return InnerExit::Transient(e),
            };
            if !ok {
                return InnerExit::Fenced;
            }
        }
    }
}

#[allow(clippy::too_many_arguments)]
async fn heartbeat_task(
    mut conn: sqlx::pool::PoolConnection<Postgres>,
    subscription_id: String,
    worker_id: String,
    fence_token: i64,
    lease_ttl: Duration,
    heartbeat_interval: Duration,
    fence_signal: CancellationToken,
    cancel: CancellationToken,
) {
    let mut interval = tokio::time::interval(heartbeat_interval);
    interval.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip);
    interval.tick().await; // consume the immediate-fire tick

    loop {
        tokio::select! {
            _ = interval.tick() => {
                match lease::heartbeat(
                    &mut conn, &subscription_id, &worker_id, fence_token, lease_ttl,
                ).await {
                    Ok(true) => {}
                    Ok(false) => {
                        tracing::info!(
                            subscription = %subscription_id,
                            fence_token,
                            "heartbeat fenced; signalling inner loop",
                        );
                        fence_signal.cancel();
                        return;
                    }
                    Err(e) => {
                        tracing::warn!(
                            subscription = %subscription_id,
                            error = %e,
                            "heartbeat error; treating as fenced to re-acquire",
                        );
                        fence_signal.cancel();
                        return;
                    }
                }
            }
            _ = cancel.cancelled() => return,
            _ = fence_signal.cancelled() => return,
        }
    }
}

async fn dispatch_with_retries(
    handler: &dyn EventHandlerDyn,
    event: &RecordedEvent,
    max_attempts: u32,
    subscription_id: &str,
) -> anyhow::Result<()> {
    let mut last_err = None;
    for attempt in 0..max_attempts {
        match handler.handle_dyn(event).await {
            Ok(()) => return Ok(()),
            Err(e) => {
                let delay = match attempt {
                    0 => Duration::from_millis(100),
                    1 => Duration::from_millis(500),
                    _ => Duration::from_secs(2),
                };
                tracing::warn!(
                    subscription = %subscription_id,
                    event_type = %event.event_type,
                    global_position = event.global_position,
                    attempt = attempt + 1,
                    "projection handler failed, retrying in {delay:?}: {e:#}",
                );
                last_err = Some(e);
                tokio::time::sleep(delay).await;
            }
        }
    }

    let err = last_err.expect("max_attempts > 0");
    tracing::error!(
        subscription = %subscription_id,
        event_type = %event.event_type,
        global_position = event.global_position,
        "projection handler failed after {} attempts: {err:#}",
        max_attempts,
    );
    Err(err)
}