//! The LISTEN forwarder (ADR-004): the wake substrate every later
//! mechanism task rides. This module carries the **skeleton** — the
//! dedicated non-pooled connection, the `poll_message` loop fanning
//! out into a bounded broadcast channel, the dynamic channel set the
//! notify-listen task's `listen` adds channels to — with the full
//! behavior (reconnect honesty contract, receiver bridging) the
//! notify-listen task's.
//!
//! # The two POC-pinned deadlock pitfalls, structurally excluded
//!
//! 1. **Query-vs-poll starvation deadlock** (POC failure mode 1): the
//!    `poll_message` loop must be *running before the first client
//!    query* on the listener connection. Every connection generation
//!    is structured the same way: the poll task is spawned first and
//!    owns the `Connection`; LISTEN/UNLISTEN queries ride the `Client`
//!    only afterwards — and even mid-connect the request cannot
//!    complete before the poll task drives the connection's response.
//!    The command select in the loop keeps the poll task forever
//!    co-resident with client queries on this connection (the
//!    exclusion is held permanently, not just at spawn).
//! 2. **Client-drop closes the session** (POC failure mode 2): the
//!    listener's `Client` is owned by the forwarder task and kept
//!    alive for the connection generation's lifetime — nobody else
//!    holds it, and it is only dropped at shutdown or once the session
//!    is already dead server-side (reconnect path).
//!
//! # Connection budget
//!
//! The listener connection is outside the pool (deadpool#360 — pooled
//! connections register LISTEN server-side but can never deliver): a
//! per-process budget line `max_size + 1` per LISTEN-ing process
//! (deployment.md). The listener sets `application_name` for
//! kill-targetability (the deployment.md ops note, POC-carried).
//!
//! # Fanout
//!
//! `poll_message`'s notifications fan into one bounded broadcast
//! channel (capacity 1024 per the POC's shape); lag is **surfaced,
//! not silent** — a `Lagged(n)` receiver resumes streaming (the POC's
//! bridge posture). The wake content (`Wake { channel }` only) and
//! per-receiver bridging are the notify-listen task's; the skeleton
//! broadcasts raw notification channel names.
//!
//! # The post-reconnect LISTEN-set invariant
//!
//! After every reconnect, the server-side LISTEN set matches the
//! registry snapshot and nothing queued from a dead generation can undo
//! it: the fresh generation's loop top re-issues LISTENs from the
//! snapshot (the re-issue's authority), and every queued command
//! predating the reconnect's generation is dropped rather than
//! replayed ([`reconcile_queued_commands`] — the generation tags, the
//! review of 002 Finding 4's fix; a stale UNLISTEN replayed after the
//! re-issued LISTEN would otherwise silently cancel a channel the
//! registry believes is listening, its wakes lost until the next
//! reconnect or a fresh registration). Queued commands carry the
//! generation they were queued under; the loop bumps the shared
//! counter at each reconnect.

use std::future::Future;
use std::sync::Arc;
use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};
use std::time::Duration;

use tokio::sync::{broadcast, mpsc, watch};
use tokio_postgres::NoTls;

use crate::seam::database_error;

/// The raw fanout sender's shared slot — the forwarder handle and the
/// loop both take it away on their exit paths, so receivers see the
/// terminal arm from *either* side.
pub(crate) type FanoutSlot = Arc<std::sync::Mutex<Option<broadcast::Sender<RawNotification>>>>;

/// The reconnect config's shared slot — the loop's connect step reads
/// it fresh per attempt; the `#[cfg(test)]` seam swaps it (the
/// unreachable-endpoint injection the failed-connect tests ride).
pub(crate) type ReconnectConfigSlot = Arc<std::sync::Mutex<tokio_postgres::Config>>;

/// A fresh connection generation the reconnect produces: the poll
/// loop's next `Connection` and the `Client` driving its queries.
pub(crate) type GenerationPair = (tokio_postgres::Client, ListenerConnection);

/// The listener connection's concrete type for the `NoTls` posture
/// (v1 TLS is unavailable on *every* connection path — pooled,
/// listener, and reconnect alike hardwire `NoTls`; the engine does
/// not ride the consumer's `Config` sslmode — TLS is a post-v1
/// deployment concern and deployment.md owns the statement; the
/// notify-listen task revisits if a task or ADR demands listener
/// TLS).
pub(crate) type ListenerConnection = tokio_postgres::Connection<
    tokio_postgres::Socket,
    <tokio_postgres::NoTls as tokio_postgres::tls::MakeTlsConnect<tokio_postgres::Socket>>::Stream,
>;

/// The forwarder's reconnect backoff: exponential 50 ms → 2 s cap
/// (the POC's policy; ADR-004 — the notify-listen task owns the full
/// reconnect behavior on this skeleton's loop).
pub(crate) const RECONNECT_BASE_MS: u64 = 50;
pub(crate) const RECONNECT_MAX_MS: u64 = 2000;

/// The bounded broadcast capacity (the POC's 1024 shape; lag surfaced,
/// not silent).
const FANOUT_CAPACITY: usize = 1024;

/// Quote a channel name as a SQL identifier (the POC's `quote_ident`
/// shape — Postgres doubles embedded `"`). Channel names are arbitrary
/// legal names (colons etc.); quoting is what makes them work.
pub(crate) fn quote_ident(name: &str) -> String {
    format!("\"{}\"", name.replace('"', "\"\""))
}

/// The raw notification the poll loop fans out — the notify task's
/// receiver bridging maps it to the opaque [`alkstore::Wake`]
/// (`Wake { channel }` only — payloads never surface, ADR-008 §3).
#[derive(Debug, Clone)]
pub(crate) struct RawNotification {
    pub channel: String,
}

/// The dynamic channel set: the registry the forwarder re-issues
/// LISTENs from after every reconnect. `listen` (registration) and
/// receiver-drop (unregistration, the refcounted last-subscriber
/// rule) ride [`Forwarder`]'s register / unregister.
#[derive(Debug, Default)]
pub(crate) struct ChannelSet {
    /// BTreeMap for the deterministic snapshot; the value is the
    /// subscriber refcount (the last-subscriber UNLISTEN rule).
    channels: std::sync::Mutex<std::collections::BTreeMap<String, u64>>,
}

impl ChannelSet {
    /// Add a subscriber to `channel` (inserting the entry on the
    /// first); returns `true` if this is a brand-new channel entry
    /// (a LISTEN must be issued). Test-side: the refcount unit pin.
    pub(crate) fn add_subscriber(&self, channel: &str) -> bool {
        let mut map = self
            .channels
            .lock()
            .expect("channel registry mutex poisoned");
        let is_new = !map.contains_key(channel);
        *map.entry(channel.to_string()).or_insert(0) += 1;
        is_new
    }

    /// Remove a subscriber from `channel`; returns `true` if the last
    /// subscriber left (an UNLISTEN must be issued and the registry
    /// entry dropped). A channel not present is a no-op (`false`).
    /// Test-side: the refcount unit pin.
    pub(crate) fn remove_subscriber(&self, channel: &str) -> bool {
        let mut map = self
            .channels
            .lock()
            .expect("channel registry mutex poisoned");
        let Some(count) = map.get_mut(channel) else {
            return false;
        };
        if *count > 1 {
            *count -= 1;
            false
        } else {
            map.remove(channel);
            true
        }
    }

    /// The registry's live channel list — the reconnect re-issue
    /// source, and the tests' registry read.
    pub(crate) fn snapshot(&self) -> Vec<String> {
        self.channels
            .lock()
            .expect("channel registry mutex poisoned")
            .keys()
            .cloned()
            .collect()
    }
}

/// The forwarder handle the store holds: the subscription point for
/// the fanout, the dynamic channel registry + LISTEN/UNLISTEN issuer,
/// and the shutdown switch.
///
/// Shutdown (`Forwarder::shutdown`, the store's `close`/`Drop`): the
/// loop exits, dropping the listener `Client` — the server session
/// ends and the listener connection is gone (the teardown's
/// "listener connection dropped" arm).
#[derive(Debug)]
pub(crate) struct Forwarder {
    /// The wake fanout — every receiver bridge holds a subscription;
    /// its release (the shutdown take, or the loop's exit-path release
    /// guard) is the receivers' terminal `Closed`/`None` arm. The
    /// shared [`FanoutSlot`] exactly because receiver-held
    /// `Arc<Forwarder>`s keep this struct alive — the release fires
    /// from both the handle's shutdown and the loop's exit
    /// (receivers-terminal-iff-loop-gone).
    fanout: FanoutSlot,
    /// The channel registry the reconnect re-issues from; read by the
    /// register/unregister surfaces below.
    channels: Arc<ChannelSet>,
    /// LISTEN/UNLISTEN commands to the live connection (the loop
    /// issues them; each carries the generation it was queued under,
    /// and a reconnect drops its predecessor's queued commands before
    /// replaying — a stale UNLISTEN must never undo the re-issue,
    /// review 002 Finding 4). Read by the register/unregister
    /// surfaces below.
    commands_tx: mpsc::UnboundedSender<ListenCommand>,
    /// The connection-generation counter the queued commands are
    /// tagged against (the loop bumps it at each reconnect; the
    /// register/unregister surfaces load it at send time).
    generation: GenerationSlot,
    /// Whether a listener connection is currently live — the
    /// register-path's honest transient state (registration during a
    /// gap fails with `Database`; the registry entry's recovery rides
    /// the reconnect re-issue). Read by `register` below.
    connected: Arc<AtomicBool>,
    /// The reconnect config's shared slot — test-observation surface
    /// for the failed-connect seam (the loop's own [`LoopCtx`] clone
    /// is the live read path; the field itself exists for the
    /// `#[cfg(test)]` swap accessors).
    #[cfg(test)]
    reconnect_config: ReconnectConfigSlot,
    /// The test-only generation-top LISTEN pause (review 003 Finding
    /// 2a's seam) — the loop's own [`LoopCtx`] clone is the live read
    /// path; the field exists for the test arm accessor below. Dead
    /// in production builds.
    #[cfg(test)]
    listen_pause: crate::seam::listen_pause::Flag,
    shutdown: watch::Sender<bool>,
}

/// Commands to the live listener connection (built by
/// `register`/`unregister` below). Every command carries the
/// generation it was queued under (the shared counter the loop bumps
/// at each reconnect): a reconnect drops queued commands predating
/// its generation before replaying — a stale UNLISTEN replayed after
/// the fresh generation's re-issued LISTENs would silently cancel a
/// channel the registry believes is listening (review 002 Finding 4).
///
/// `Listen` carries an ack the forwarder resolves after the server
/// processed the LISTEN — the registration is synchronous end-to-end,
/// so a `listen()` that returned sees every later commit (`listen`
/// **starts from now** — a notify racing the LISTEN cannot be lost;
/// delivery only starts at server-side registration).
#[derive(Debug)]
pub(crate) enum ListenCommand {
    Listen {
        generation: u64,
        channel: String,
        ack: tokio::sync::oneshot::Sender<alkstore::Result<()>>,
    },
    Unlisten {
        generation: u64,
        channel: String,
    },
}

/// The shared generation counter (the `Forwarder` handle tags queued
/// commands with its value at send time; the loop is the sole writer,
/// bumping it at each reconnect — a reader can never see a value the
/// loop has not yet re-issued from).
pub(crate) type GenerationSlot = Arc<AtomicU64>;

impl ListenCommand {
    /// The generation the command was queued under (the reconnect's
    /// drop decision reads this).
    fn generation(&self) -> u64 {
        match self {
            ListenCommand::Listen { generation, .. } => *generation,
            ListenCommand::Unlisten { generation, .. } => *generation,
        }
    }

    /// Resolve a dropped stale command: a queued-unprocessed LISTEN's
    /// caller awaits its ack — it receives the transient mid-LISTEN
    /// error (the connection it targeted is gone; the registration is
    /// recovered by the re-issue, `register`'s documented posture). A
    /// dropped UNLISTEN is best-effort — nothing to resolve.
    pub(crate) fn abandon_stale(self) {
        if let ListenCommand::Listen { ack, .. } = self {
            let _ = ack.send(Err(mid_listen_error()));
        }
    }
}

/// The stale-command reconcile (review 002 Finding 4's decision, the
/// pure helper the reconnect and the replay loop share): from the
/// commands queued when the previous connection generation ended and
/// the generation just re-connected into, `(to replay, dropped)`.
///
/// A command replays iff it was queued under the **current**
/// generation; everything older is dropped and never touches the
/// fresh connection. The invariant this enforces: after a reconnect,
/// the server-side LISTEN set matches the registry snapshot (the
/// re-issue) and nothing queued from a dead generation can undo it —
/// the snapshot re-issue covers every stale LISTEN (the registry is
/// the source of truth; registry-write-first ordering), and the stale
/// UNLISTENs are exactly what must not replay (the finding's silent
/// wake-loss). Note the decision rides the generation tags alone:
/// the snapshot's role is the re-issue's, not this filter's (the
/// generation shape of the review's two candidates).
pub(crate) fn reconcile_queued_commands(
    queued: Vec<ListenCommand>,
    generation: u64,
) -> (Vec<ListenCommand>, Vec<ListenCommand>) {
    let mut replay = Vec::new();
    let mut dropped = Vec::new();
    for command in queued {
        if command.generation() < generation {
            dropped.push(command);
        } else {
            replay.push(command);
        }
    }
    (replay, dropped)
}

/// The mid-LISTEN transient error — a `register` whose LISTEN did not
/// land (died mid-command, or dropped stale at a reconnect) acks/
/// returns this; the retry recovers it (the registry entry stays
/// either way and the re-issue covers the channel on reconnect).
fn mid_listen_error() -> alkstore::Error {
    database_error(
        "the listener connection died mid-LISTEN; \
         the channel is registered and re-issued on reconnect",
    )
}

/// Execute one command against the live connection (the replay's and
/// the serve loop's shared arm). A `Listen` acks its caller
/// synchronously; an `Unlisten` is best-effort.
async fn run_command(client: &tokio_postgres::Client, command: ListenCommand) {
    match command {
        ListenCommand::Listen { channel, ack, .. } => {
            let result = if client
                .batch_execute(&format!("LISTEN {};", quote_ident(&channel)))
                .await
                .is_ok()
            {
                Ok(())
            } else {
                Err(mid_listen_error())
            };
            let _ = ack.send(result);
        }
        ListenCommand::Unlisten { channel, .. } => {
            let _ = client
                .batch_execute(&format!("UNLISTEN {};", quote_ident(&channel)))
                .await;
        }
    }
}

impl Forwarder {
    /// Take over an established listener connection (the `open`
    /// constructor connects so `open` fails `Database` synchronously
    /// on an unreachable server) and spawn the forwarder loop.
    pub(crate) fn spawn(
        client: tokio_postgres::Client,
        connection: ListenerConnection,
        reconnect_config: tokio_postgres::Config,
    ) -> Forwarder {
        let (fanout, _) = broadcast::channel(FANOUT_CAPACITY);
        let fanout_slot: FanoutSlot = Arc::new(std::sync::Mutex::new(Some(fanout.clone())));
        let channels = Arc::new(ChannelSet::default());
        let (commands_tx, commands_rx) = mpsc::unbounded_channel();
        let (shutdown_tx, shutdown_rx) = watch::channel(false);
        let generation: GenerationSlot = Arc::new(AtomicU64::new(0));
        let reconnect_config_slot: ReconnectConfigSlot =
            Arc::new(std::sync::Mutex::new(reconnect_config));
        #[cfg(test)]
        let listen_pause = crate::seam::listen_pause::disarmed();
        // The connection handed in was just established — the liveness
        // flag starts true and the loop clears it at each generation's
        // death (the register path's honest transient state).
        let connected = Arc::new(AtomicBool::new(true));

        // The loop owns both the Client and the Connection — the
        // pitfall-2 exclusion (kept alive for the connection's
        // lifetime).
        tokio::spawn(forwarder_loop(LoopCtx {
            client,
            connection,
            reconnect_config: reconnect_config_slot.clone(),
            #[cfg(test)]
            listen_pause: listen_pause.clone(),
            fanout: fanout.clone(),
            fanout_slot: fanout_slot.clone(),
            channels: channels.clone(),
            commands: commands_rx,
            generation: generation.clone(),
            connected: connected.clone(),
            shutdown: shutdown_rx,
        }));

        Forwarder {
            fanout: fanout_slot.clone(),
            channels,
            commands_tx,
            generation,
            connected,
            #[cfg(test)]
            reconnect_config: reconnect_config_slot,
            #[cfg(test)]
            listen_pause,
            shutdown: shutdown_tx,
        }
    }

    /// Subscribe to the raw fanout (the notify task bridges broadcast
    /// receivers into `WakeReceiver`s). `None` after shutdown — or
    /// after *any* forwarder-loop exit (the loop's release guard takes
    /// the shared sender too, the defense-in-depth arm: a subscribe
    /// racing a dead loop fails closed rather than parking on a
    /// forever-silent broadcast).
    pub(crate) fn subscribe(&self) -> Option<broadcast::Receiver<RawNotification>> {
        self.fanout
            .lock()
            .expect("fanout mutex poisoned")
            .as_ref()
            .map(|s| s.subscribe())
    }

    /// Register a channel AND issue `LISTEN` on the live connection,
    /// waiting for the server-side registration (the ack — a `listen`
    /// that returned sees every later commit; `listen` starts from
    /// now).
    ///
    /// The registry write happens first: if the connection dies
    /// between the write and the LISTEN delivery, the reconnect path
    /// re-issues from the snapshot — a registration is never lost
    /// (the ordering is the registration-recovery discipline).
    ///
    /// Refcounted: N subscribers on one channel share one LISTEN.
    ///
    /// Fails `Database` when no listener connection is currently live
    /// (mid-reconnect) or the LISTEN itself failed — transient; a
    /// retry recovers it (the registry entry stays either way and the
    /// re-issue covers the channel on reconnect).
    pub(crate) async fn register(&self, channel: &str) -> Result<(), alkstore::Error> {
        self.channels.add_subscriber(channel);
        if !self.connected.load(Ordering::Acquire) {
            return Err(database_error(
                "the listener connection is not currently up (mid-reconnect); \
                 the channel is registered and re-issued on reconnect",
            ));
        }
        let (ack_tx, ack_rx) = tokio::sync::oneshot::channel();
        let sent = self
            .commands_tx
            .send(ListenCommand::Listen {
                generation: self.generation.load(Ordering::Acquire),
                channel: channel.to_string(),
                ack: ack_tx,
            })
            .is_ok();
        if !sent {
            return Err(database_error("the forwarder loop is gone (shutting down)"));
        }
        ack_rx
            .await
            .map_err(|_| database_error("the forwarder loop is gone (shutting down)"))?
    }

    /// Unregister one subscriber's reference on `channel` AND issue
    /// `UNLISTEN` when the **last** subscriber for the channel drops
    /// (the refcount; zero subscribers ⇒ no LISTEN needed). Registry
    /// write first (ordering symmetric with [`Forwarder::register`]'s
    /// recovery logic); the UNLISTEN is best-effort — a lost one
    /// re-issues nothing (UNLISTEN is not recovery-relevant: the
    /// snapshot won't carry the removed channel). The queued UNLISTEN
    /// rides its send-time generation: if the connection died before
    /// processing it, the reconnect drops it rather than replaying —
    /// a stale UNLISTEN after the re-issued LISTENs would otherwise
    /// silently cancel a re-registered channel (review 002 Finding 4;
    /// the fresh connection's LISTEN set is the snapshot's).
    pub(crate) fn unregister(&self, channel: &str) {
        if self.channels.remove_subscriber(channel) {
            let _ = self.commands_tx.send(ListenCommand::Unlisten {
                generation: self.generation.load(Ordering::Acquire),
                channel: channel.to_string(),
            });
        }
    }

    /// Shutdown: flip the switch — the loop exits, dropping the
    /// listener `Client` and its session (the listener connection is
    /// gone) — and release this handle's fanout sender. The broadcast
    /// closes once every sender is dropped (the loop's exit-path
    /// release guard empties the shared slot too; this handle's take
    /// fires the terminal arm at the flip — receivers unaffected by
    /// `Arc<Forwarder>` lifetimes see it). Idempotent.
    pub(crate) fn shutdown(&self) {
        let _ = self.shutdown.send(true);
        // Take the sender even if a receiver holds this Arc: the
        // receivers' terminal `None` must fire at engine shutdown
        // whatever lifetime stands on the handle.
        *self.fanout.lock().expect("fanout mutex poisoned") = None;
    }

    /// Aim the reconnect arm's connect step at `config` (the failed-
    /// connect test seam — the unreachable-endpoint injection). The
    /// loop reads the slot fresh per attempt, so a swap lands on the
    /// next attempt.
    #[cfg(test)]
    pub(crate) fn swap_reconnect_config(&self, config: tokio_postgres::Config) {
        *self
            .reconnect_config
            .lock()
            .expect("reconnect config mutex poisoned") = config;
    }

    /// The reconnect config's current content (the failed-connect
    /// test's restore point — the swap-back carries the original).
    #[cfg(test)]
    pub(crate) fn reconnect_config_snapshot(&self) -> tokio_postgres::Config {
        self.reconnect_config
            .lock()
            .expect("reconnect config mutex poisoned")
            .clone()
    }

    /// Arm the test-only generation-top LISTEN pause (one-shot) — the
    /// connection-death-during-LISTEN seam (review 003 Finding 2a):
    /// the loop sleeps a short window at the next generation's top,
    /// *before* issuing its snapshot LISTEN, so a test can kill the
    /// fresh generation's backend inside the window and make the
    /// LISTEN fail honestly (the real fall-through arm runs).
    #[cfg(test)]
    pub(crate) fn arm_generation_listen_pause(&self) {
        crate::seam::listen_pause::arm(&self.listen_pause)
    }
}

/// The reconnect arm's retry core (review 002 Finding 1's fix): sleep
/// → attempt the connect — repeat until an attempt succeeds or the
/// shutdown flip, **never returning with nothing connected** (a failed
/// connect retries inside this arm; the skeleton's fall-back-into-the-
/// loop-with-an-empty-slot shape — one failed connect killing the
/// loop forever — is structurally gone).
///
/// `backoff_ms` carries the live exponential backoff (doubled per
/// attempt, capped at [`RECONNECT_MAX_MS`]; the caller resets it on
/// the next *verified-alive* generation). The shutdown flip aborts
/// with `None` — the only exit besides `Some`'s fresh generation.
pub(crate) async fn reconnect_with_retry<F, Fut>(
    mut attempt: F,
    backoff_ms: &mut u64,
    shutdown: &mut watch::Receiver<bool>,
) -> Option<GenerationPair>
where
    F: FnMut() -> Fut,
    Fut: Future<Output = Option<GenerationPair>>,
{
    loop {
        tokio::select! {
            biased;

            // Shutdown (or the sender's drop) exits mid-backoff — the
            // subscribers' terminal arm is not held hostage by a
            // backoff cycle.
            _ = shutdown.changed() => return None,

            _ = tokio::time::sleep(Duration::from_millis(*backoff_ms)) => {}
        }
        *backoff_ms = backoff_ms.saturating_mul(2).min(RECONNECT_MAX_MS);
        if let Some(generation) = attempt().await {
            return Some(generation);
        }
    }
}

/// Releases the forwarder handle's fanout sender at the loop's exit —
/// *every* exit path: Drop runs on return, unwind, and task abort
/// alike. This is the receivers-terminal-iff-loop-gone invariant's
/// enforcement side (the shutdown take covers the flip; this guard
/// covers every other exit the defense-in-depth posture guards
/// against) — a dead loop can never leave subscribers parked on a
/// forever-silent broadcast.
pub(crate) struct FanoutRelease {
    pub(crate) slot: FanoutSlot,
}

impl Drop for FanoutRelease {
    fn drop(&mut self) {
        *self.slot.lock().expect("fanout mutex poisoned") = None;
    }
}

/// The forwarder loop (the POC's ~90-line shape, restructured for the
/// dynamic channel set + command issuer + shutdown + the reconnect
/// retry arm).
///
/// Per connection generation: spawn the poll task (it owns the
/// `Connection` and fans out `poll_message`'s notifications) → issue
/// LISTENs from the live channel-set snapshot → serve commands and
/// wait for the connection's death → reconnect via
/// [`reconnect_with_retry`] (exponential backoff 50 ms → 2 s cap;
/// failed connects never leave the arm). The loop-carried connection
/// is *always present* at the generation's top by construction: the
/// first iteration holds the handed-in connection, and only a
/// successful connect re-binds it — there is no empty-slot arm left to
/// fall back into.
///
/// The reconnect bumps the shared generation counter (before the
/// snapshot re-issue, so commands queued past the bump — a healthy
/// `register`'s among them — tag as current-generation and replay);
/// at each fresh generation's top the predecessor's queued commands
/// are drained through [`reconcile_queued_commands`] — everything
/// tagged under a dead generation is dropped, never replayed onto the
/// new connection (the stale-UNLISTEN hazard, review 002 Finding 4).
/// The invariant: **after a reconnect, the server-side LISTEN set
/// matches the registry snapshot and nothing queued from a dead
/// generation can undo it.**
///
/// The receiver-facing close shape (ADR-021 §5's pg arm): the fanout
/// sender clones live here (loop) and on the shared slot (taken by the
/// shutdown flip or the exit-path release guard — see
/// [`FanoutRelease`]); a connection death and its reconnect are
/// invisible to subscribers (the fanout never drops on a generation
/// boundary — receivers stay open across reconnects, they receive the
/// reserved reconnect-wake instead); the senders drop only when the
/// loop is gone — the invariant's shape.
struct LoopCtx {
    client: tokio_postgres::Client,
    connection: ListenerConnection,
    reconnect_config: ReconnectConfigSlot,
    /// The test-only generation-top LISTEN pause (review 003 Finding
    /// 2a's seam); dead in production builds.
    #[cfg(test)]
    listen_pause: crate::seam::listen_pause::Flag,
    fanout: broadcast::Sender<RawNotification>,
    fanout_slot: FanoutSlot,
    channels: Arc<ChannelSet>,
    commands: mpsc::UnboundedReceiver<ListenCommand>,
    generation: GenerationSlot,
    connected: Arc<AtomicBool>,
    shutdown: watch::Receiver<bool>,
}

async fn forwarder_loop(
    LoopCtx {
        client,
        connection,
        reconnect_config,
        #[cfg(test)]
        listen_pause,
        fanout,
        fanout_slot,
        channels,
        mut commands,
        generation: generation_slot,
        connected,
        mut shutdown,
    }: LoopCtx,
) {
    let _fanout_release = FanoutRelease { slot: fanout_slot };
    let mut client = client;
    let mut backoff_ms = RECONNECT_BASE_MS;
    let mut first_listen = true;
    let mut connection = connection;
    let mut generation = generation_slot.load(Ordering::Acquire);

    loop {
        // The dedicated poll loop, spawned BEFORE any client query can
        // ride this connection (pitfall 1's structural exclusion —
        // held every generation, and the command select below keeps
        // the poll task co-resident with client queries forever). The
        // poll task owns the Connection; the loop drives queries
        // through the Client.
        let mut poll_task = tokio::spawn(poll_loop(connection, fanout.clone()));

        // The test-only LISTEN-window seam (review 003 Finding 2a):
        // one-shot armed, this generation's top pauses here for the
        // window a backend-kill test can land in — the LISTEN below
        // then runs against a dead session and fails honestly, the
        // exact "connection died before/during the LISTEN" shape the
        // fall-through arm exists for. Production builds compile a
        // plain `false`.
        #[cfg(test)]
        let pre_listen_pause = crate::seam::listen_pause::take(&listen_pause);
        #[cfg(not(test))]
        let pre_listen_pause = false;
        if pre_listen_pause {
            #[cfg(test)]
            tokio::time::sleep(crate::seam::listen_pause::WINDOW).await;
        }

        // Issue LISTENs from the dynamic channel set's snapshot (on
        // first connect it carries pre-registered channels; on
        // reconnect it carries everything `listen` ever registered).
        // With no channels yet, a `SELECT 1` liveness probe stands in
        // — `connected` must reflect a verified-alive session, not an
        // assumed one.
        let listen_sql = channels
            .snapshot()
            .iter()
            .map(|ch| format!("LISTEN {};\n", quote_ident(ch)))
            .collect::<String>();
        let listen_ok = if listen_sql.is_empty() {
            client.simple_query("SELECT 1").await.is_ok()
        } else {
            client.batch_execute(&listen_sql).await.is_ok()
        };

        if listen_ok {
            backoff_ms = RECONNECT_BASE_MS;
            connected.store(true, Ordering::Release);
            // The stale command drain (review 002 Finding 4): the
            // re-issue above is the server-side LISTEN set's authority
            // — the predecessor generation's queued commands are
            // reconciled before the serve loop starts, so a stale
            // UNLISTEN can never replay after it and silently cancel a
            // live channel. Commands queued past the bump (a
            // register/unregister that raced the reconnect) replay
            // normally.
            let mut queued = Vec::new();
            while let Ok(cmd) = commands.try_recv() {
                queued.push(cmd);
            }
            let (replay, dropped) = reconcile_queued_commands(queued, generation);
            for cmd in dropped {
                cmd.abandon_stale();
            }
            for cmd in replay {
                run_command(&client, cmd).await;
            }
            if !first_listen {
                // The synthetic reconnect-wake (ADR-008 §4's reserved
                // channel; ADR-006's recovery semantics) broadcast to
                // every subscriber's receiver — the no-replay hole's
                // honest communication. The wake-content mapping is
                // the notify-listen task's; the skeleton broadcasts
                // the reserved channel's name so the fanout carries it.
                let _ = fanout.send(RawNotification {
                    channel: alkstore::RESERVED_LISTENER_RECONNECTED.to_string(),
                });
            }
            first_listen = false;
        } else {
            // The connection died before/during the LISTEN — fall
            // through to the reconnect path with the poll task's
            // completion below.
            connected.store(false, Ordering::Release);
        }

        // Drive the connection until it dies (or shutdown). Client
        // queries (LISTEN/UNLISTEN commands) run while the poll task
        // polls the same session — the co-residency that excludes
        // pitfall 1's starvation permanently.
        if listen_ok {
            loop {
                tokio::select! {
                    biased;

                    // Shutdown: exit the whole forwarder (dropping the
                    // client closes the listener session).
                    _ = shutdown.changed() => {
                        poll_task.abort();
                        return;
                    }

                    cmd = commands.recv() => {
                        match cmd {
                            Some(command) if command.generation() < generation => {
                                // A stale command queued under a dead
                                // generation (a send racing the
                                // reconnect's drain): dropped here for
                                // the same reason the generation-top
                                // drain drops its batch — it must not
                                // touch this connection (review 002
                                // Finding 4).
                                command.abandon_stale();
                            }
                            Some(command) => {
                                // Acked after the server processed the
                                // LISTEN: a `listen()` return stamps
                                // "registered from now" end-to-end. A
                                // failed LISTEN (connection died
                                // mid-command) acks the failure — the
                                // caller sees it; the registry entry
                                // rides the reconnect re-issue.
                                run_command(&client, command).await;
                            }
                            // The command sender is gone (the store —
                            // and its forwarder handle — went away
                            // without a shutdown flip): serve no more,
                            // exit (the release guard fires; the
                            // receivers' terminal arm).
                            None => {
                                poll_task.abort();
                                return;
                            }
                        }
                    }

                    // The poll task ended = the connection died (a
                    // fatal poll error or server close). The session
                    // is already gone server-side; reconnect below.
                    _ = &mut poll_task => {
                        break;
                    }
                }
            }
        } else {
            // A failed probe/LISTEN means the connection died: wait
            // the poll task out (it owns the connection and is its
            // reaper).
            let _ = poll_task.await;
        }

        // Reconnect: the Client is dead with the session (dropping it
        // here closes nothing server-side — pitfall 2 was satisfied
        // for the connection's whole lifetime). The retry core sleeps
        // and attempts the connect — with the *fresh* slot content per
        // attempt (the test seam swaps it) — until a connect succeeds
        // or the shutdown flip; a failed connect never falls back into
        // the loop above without a connection (the loop's `None` arm
        // is restructured away entirely).
        connected.store(false, Ordering::Release);
        let attempt = {
            let slot = reconnect_config.clone();
            move || {
                let guard = slot.lock().expect("reconnect config mutex poisoned");
                let config = guard.clone();
                drop(guard);
                async move { config.connect(NoTls).await.ok() }
            }
        };
        let Some((next_client, next_connection)) =
            reconnect_with_retry(attempt, &mut backoff_ms, &mut shutdown).await
        else {
            // Shutdown mid-backoff/mid-retry: the poll task is already
            // done (the generation's end preceded this arm); plain
            // return — the guard and the client drop carry the rest.
            return;
        };
        // The generation bump happens here — *before* the fresh
        // generation's loop top re-issues the snapshot — so every
        // command queued under the dead connection is stale-tagged
        // against the new one and the generation-top drain drops it
        // (review 002 Finding 4); commands queued from here on
        // (register's honest error path, the re-issue) tag current.
        generation += 1;
        generation_slot.store(generation, Ordering::Release);
        client = next_client;
        connection = next_connection;
    }
}

/// The poll loop over one connection generation: `poll_message`
/// forever, fanning notifications into the bounded broadcast. Returns
/// when the connection dies (a fatal poll error or server close) —
/// the forwarder loop reconnects.
async fn poll_loop(mut connection: ListenerConnection, fanout: broadcast::Sender<RawNotification>) {
    loop {
        match std::future::poll_fn(|cx| connection.poll_message(cx)).await {
            Some(Ok(tokio_postgres::AsyncMessage::Notification(n))) => {
                // Bounded fanout — lag is surfaced, not silent: the
                // POC's posture is a `Lagged(n)` receiver keeps
                // streaming (the receiver bridge's `Lagged` arm logs /
                // recovers; capacity pressure must not kill the
                // fanout). The send-error arm is "no receivers
                // subscribed" — the normal idle posture.
                let _ = fanout.send(RawNotification {
                    channel: n.channel().to_string(),
                });
            }
            Some(Ok(_)) => {}
            Some(Err(_)) | None => return,
        }
    }
}