//! The stream mechanism's mem arm — the in-memory event log +
//! per-consumer offsets (engine-mem.md §Mapping the contract: global
//! FIFO by monotone offset; keyed publish rides the overlay).
//!
//! - Publishes append to the per-stream log under its section guard:
//!   engine-assigned offsets (the engine-wide allocator — monotone,
//!   immutable, never renumbered; the concrete numbering is engine-
//!   internal and never compared cross-engine — ADR-015 §4: mem's
//!   fresh-engine count starts at 1, the SQL engines' counters may
//!   not), keys
//!   are carried metadata (a present key must be non-empty —
//!   `InvalidName`, ADR-015 §1), and the commit boundary fires one
//!   wake per event on the stream's engine-internal wake channel.
//! - Cursor reads are `offset ASC` (global FIFO, ADR-015 §4) with the
//!   extent guard at the trait-impl entry (`limit <= 0` → empty
//!   `Vec`, ADR-023 §2).
//! - Checkpoints are explicit, monotone saves (a save below the
//!   stored checkpoint is a silent no-op — ADR-019 §6; an absent
//!   consumer reads 0, ADR-019 §1).
//! - `trim_to` deletes `offset <= horizon` exactly (a negative horizon
//!   deletes nothing, idempotently — the boundary argument is total,
//!   ADR-023 §2), **wakes nothing**, and surviving events keep their
//!   offsets (gaps legal, ADR-015 §5). Reads from a trimmed region
//!   resume at the horizon's first remaining row.
//! - `subscribe` attaches the receiver's wake feed **before** the
//!   stored-offset read (no commit can fall between — a buffered wake
//!   for a commit already re-read costs an empty re-poll, never a
//!   missed event), drains the log to the tail, then parks on the
//!   feed: wake-driven re-read with no polling requirement; explicit
//!   saves only. The one close path is engine drop — the bus close
//!   disconnects every feed and `recv()` returns the terminal `None`
//!   (ADR-021 §5).

use std::collections::VecDeque;
use std::ops::Bound;
use std::sync::Arc;

use serde_json::Value;

use alkstore::{
    BoxedFuture, Error, EventReceiver, Result, StreamEvent, StreamHandle, validate_local_name,
};

use crate::clock::Clock;
use crate::state::Guard;
use crate::state::{EventRow, State, StreamState};
use crate::wakes::WakeBus;

/// How many events a wake-driven re-read fetches per page while
/// draining to the tail.
const SUBSCRIBE_PAGE: i64 = 256;

/// The engine-internal wake channel a stream's subscribers ride:
/// engine-derived, under the reserved prefix — consumer entry points
/// reject the prefix, so a consumer `listen` can never attach to it
/// (and `notify` can never fire into it). The publish-side wake feed
/// is plumbing, not contract surface.
pub(crate) fn stream_wake_channel(stream: &str) -> String {
    format!("{}stream_wake:{}", alkstore::RESERVED_PREFIX, stream)
}

/// Insert one committed event under the streams guard (the auto-commit
/// publish and the commit-merge share this effect). Returns nothing —
/// the caller owns the wake firing at its commit boundary.
pub(crate) fn insert_event(s: &mut StreamState, stream: &str, row: EventRow) {
    s.logs
        .entry(stream.to_string())
        .or_default()
        .insert(row.offset, row);
}

/// Read `offset ASC`, `offset > cursor`, up to `limit` (the extent
/// guard is the caller's — this helper is contract-blind to it).
pub(crate) fn read_events(s: &StreamState, stream: &str, from: i64, limit: i64) -> Vec<EventRow> {
    s.logs
        .get(stream)
        .map(|log| {
            log.range((Bound::Excluded(from), Bound::Unbounded))
                .take(limit.max(0) as usize)
                .map(|(_, row)| row.clone())
                .collect()
        })
        .unwrap_or_default()
}

/// The monotone checkpoint save (ADR-019 §6): a save at-or-below the
/// stored checkpoint is a silent no-op; the direct form, the tx form,
/// and the receiver's no-arg form all land here.
pub(crate) fn save_offset_row(s: &mut StreamState, stream: &str, consumer: &str, offset: i64) {
    let key = (stream.to_string(), consumer.to_string());
    match s.offsets.get(&key) {
        Some(current) if offset <= *current => {}
        _ => {
            s.offsets.insert(key, offset);
        }
    }
}

/// The stored checkpoint; absent consumer = 0.
pub(crate) fn get_offset_row(s: &StreamState, stream: &str, consumer: &str) -> i64 {
    s.offsets
        .get(&(stream.to_string(), consumer.to_string()))
        .copied()
        .unwrap_or(0)
}

/// `trim_to`: delete `offset <= horizon` (exact boundary); surviving
/// events keep their offsets. Returns the deleted count. Wakes nothing.
pub(crate) fn trim_events(s: &mut StreamState, stream: &str, horizon: i64) -> i64 {
    let Some(log) = s.logs.get_mut(stream) else {
        return 0;
    };
    let removed: Vec<i64> = log
        .range((Bound::Unbounded, Bound::Included(horizon)))
        .map(|(k, _)| *k)
        .collect();
    let count = removed.len();
    for offset in removed {
        log.remove(&offset);
    }
    count as i64
}

fn event_from_row(row: EventRow, stream: &str) -> StreamEvent {
    StreamEvent::from_row(
        row.offset,
        stream.to_string(),
        row.key,
        row.payload,
        row.created_at,
    )
}

/// The stream-scoped handle (see the module docs) for one validated
/// stream name.
pub struct MemStreamHandle {
    pub(crate) name: String,
    pub(crate) state: Arc<State>,
    pub(crate) clock: Arc<Clock>,
    pub(crate) wakes: Arc<WakeBus>,
}

impl std::fmt::Debug for MemStreamHandle {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("MemStreamHandle")
            .field("name", &self.name)
            .finish_non_exhaustive()
    }
}

impl MemStreamHandle {
    pub(crate) fn new(
        name: String,
        state: Arc<State>,
        clock: Arc<Clock>,
        wakes: Arc<WakeBus>,
    ) -> Self {
        Self {
            name,
            state,
            clock,
            wakes,
        }
    }

    /// The keyed publish is the one publish shape (the plain form is
    /// `key = None`).
    fn stage_publish(&self, key: Option<String>, payload: Value) -> Result<i64> {
        if let Some(k) = &key
            && (k.is_empty() || k.trim().is_empty())
        {
            return Err(Error::InvalidName { name: k.clone() });
        }
        let bytes = alkstore::encode_payload(&payload)?;
        let now = self.clock.unix_now();
        let offset = self.state.allocate_event_offset();
        let row = EventRow {
            offset,
            key,
            payload: bytes,
            created_at: now,
        };
        {
            let mut s = self.state.streams.guard();
            insert_event(&mut s, &self.name, row);
        }
        // The auto-commit publish's commit boundary is its own
        // completion: one wake per event, per-subscriber FIFO.
        let channel = stream_wake_channel(&self.name);
        self.wakes.fire_commit_wakes([channel]);
        Ok(offset)
    }
}

impl StreamHandle for MemStreamHandle {
    fn name(&self) -> &str {
        &self.name
    }

    fn publish<'a>(&'a self, payload: Value) -> BoxedFuture<'a, Result<i64>> {
        match self.stage_publish(None, payload) {
            Ok(offset) => Box::pin(async move { Ok(offset) }),
            Err(e) => Box::pin(async move { Err(e) }),
        }
    }

    fn publish_with_key<'a>(
        &'a self,
        key: Option<String>,
        payload: Value,
    ) -> BoxedFuture<'a, Result<i64>> {
        match self.stage_publish(key, payload) {
            Ok(offset) => Box::pin(async move { Ok(offset) }),
            Err(e) => Box::pin(async move { Err(e) }),
        }
    }

    fn read_since<'a>(
        &'a self,
        offset: i64,
        limit: i64,
    ) -> BoxedFuture<'a, Result<Vec<StreamEvent>>> {
        if limit <= 0 {
            return Box::pin(async { Ok(Vec::new()) });
        }
        let events = {
            let s = self.state.streams.guard();
            read_events(&s, &self.name, offset, limit)
        };
        let name = self.name.clone();
        Box::pin(async move {
            Ok(events
                .into_iter()
                .map(|r| event_from_row(r, &name))
                .collect())
        })
    }

    fn read_from_consumer<'a>(
        &'a self,
        consumer: &str,
        limit: i64,
    ) -> BoxedFuture<'a, Result<Vec<StreamEvent>>> {
        if let Err(e) = validate_local_name(consumer) {
            return Box::pin(async move { Err(e) });
        }
        if limit <= 0 {
            return Box::pin(async { Ok(Vec::new()) });
        }
        let name = self.name.clone();
        let state = self.state.clone();
        let consumer = consumer.to_string();
        Box::pin(async move {
            let events = {
                let s = state.streams.guard();
                let from = get_offset_row(&s, &name, &consumer);
                read_events(&s, &name, from, limit)
            };
            Ok(events
                .into_iter()
                .map(|r| event_from_row(r, &name))
                .collect())
        })
    }

    fn save_offset<'a>(&'a self, consumer: &str, offset: i64) -> BoxedFuture<'a, Result<()>> {
        if let Err(e) = validate_local_name(consumer) {
            return Box::pin(async move { Err(e) });
        }
        let name = self.name.clone();
        let state = self.state.clone();
        let consumer = consumer.to_string();
        Box::pin(async move {
            let mut s = state.streams.guard();
            save_offset_row(&mut s, &name, &consumer, offset);
            Ok(())
        })
    }

    fn get_offset<'a>(&'a self, consumer: &str) -> BoxedFuture<'a, Result<i64>> {
        if let Err(e) = validate_local_name(consumer) {
            return Box::pin(async move { Err(e) });
        }
        let name = self.name.clone();
        let state = self.state.clone();
        let consumer = consumer.to_string();
        Box::pin(async move {
            let s = state.streams.guard();
            Ok(get_offset_row(&s, &name, &consumer))
        })
    }

    fn trim_to<'a>(&'a self, horizon: i64) -> BoxedFuture<'a, Result<i64>> {
        let name = self.name.clone();
        let state = self.state.clone();
        Box::pin(async move {
            let mut s = state.streams.guard();
            Ok(trim_events(&mut s, &name, horizon))
        })
    }

    fn subscribe<'a>(&'a self, consumer: &str) -> BoxedFuture<'a, Result<Box<dyn EventReceiver>>> {
        if let Err(e) = validate_local_name(consumer) {
            return Box::pin(async move { Err(e) });
        }
        let name = self.name.clone();
        let consumer = consumer.to_string();
        let state = self.state.clone();
        let wakes = self.wakes.clone();
        Box::pin(async move {
            // Attach first (starts from "now"; attach-after-commit
            // no-replay is structural), then read the stored offset and
            // drain to the tail.
            let feed = wakes.attach(&stream_wake_channel(&name));
            let mut cursor;
            let mut pending = VecDeque::new();
            {
                let s = state.streams.guard();
                cursor = get_offset_row(&s, &name, &consumer);
                loop {
                    let rows = read_events(&s, &name, cursor, SUBSCRIBE_PAGE);
                    let got = rows.len();
                    for row in rows {
                        cursor = row.offset;
                        pending.push_back(event_from_row(row, &name));
                    }
                    if got < SUBSCRIBE_PAGE as usize {
                        break;
                    }
                }
            }
            Ok(Box::new(MemEventReceiver {
                stream: name,
                consumer,
                state,
                feed,
                pending,
                cursor,
                position: 0,
            }) as Box<dyn EventReceiver>)
        })
    }
}

/// The durable subscription receiver (see the module docs). Its
/// position is the last-*yielded* event's offset, so `save_offset()`
/// never checkpoints an event the consumer did not observe.
pub struct MemEventReceiver {
    stream: String,
    consumer: String,
    state: Arc<State>,
    feed: crate::wakes::WakeFeed,
    pending: VecDeque<StreamEvent>,
    cursor: i64,
    position: i64,
}

impl std::fmt::Debug for MemEventReceiver {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("MemEventReceiver")
            .field("stream", &self.stream)
            .finish_non_exhaustive()
    }
}

impl MemEventReceiver {
    /// One forward read from the cursor into the pending buffer.
    fn drain_a_page(&mut self) -> Result<()> {
        let events = {
            let s = self.state.streams.guard();
            read_events(&s, &self.stream, self.cursor, SUBSCRIBE_PAGE)
        };
        for row in events {
            self.cursor = row.offset;
            self.pending.push_back(event_from_row(row, &self.stream));
        }
        Ok(())
    }

    fn pop_positional(&mut self) -> Option<StreamEvent> {
        let event = self.pending.pop_front()?;
        self.position = event.offset;
        Some(event)
    }
}

impl EventReceiver for MemEventReceiver {
    fn recv<'a>(&'a mut self) -> BoxedFuture<'a, Option<Result<StreamEvent>>> {
        Box::pin(async move {
            loop {
                if let Some(event) = self.pop_positional() {
                    return Some(Ok(event));
                }
                // Park on the wake feed: a commit that lands after this
                // park fires a wake that releases it; a commit that
                // landed before it already buffered a wake (attach-first).
                let _wake = match self.feed.rx().recv().await {
                    Some(wake) => wake,
                    None => return None,
                };
                if let Err(e) = self.drain_a_page() {
                    return Some(Err(database_only(e)));
                }
            }
        })
    }

    fn try_recv(&mut self) -> Result<Option<StreamEvent>> {
        if let Some(event) = self.pop_positional() {
            return Ok(Some(event));
        }
        // A buffered wake is the hint that a commit landed which the
        // drain has not seen yet; the poll drains it and re-reads once.
        let closed = matches!(
            self.feed.rx().try_recv(),
            Err(tokio::sync::mpsc::error::TryRecvError::Disconnected)
        );
        self.drain_a_page()?;
        if let Some(event) = self.pop_positional() {
            return Ok(Some(event));
        }
        if closed {
            // The bus close (engine drop) is the one close path: once
            // the feed is disconnected and nothing is buffered or
            // readable, Closed is the terminal poll arm (never reopens).
            return Err(Error::Closed);
        }
        Ok(None)
    }

    fn read_since<'a>(
        &'a mut self,
        offset: i64,
        limit: i64,
    ) -> BoxedFuture<'a, Result<Vec<StreamEvent>>> {
        if limit <= 0 {
            return Box::pin(async { Ok(Vec::new()) });
        }
        let events = {
            let s = self.state.streams.guard();
            read_events(&s, &self.stream, offset, limit)
        };
        let stream = self.stream.clone();
        Box::pin(async move {
            Ok(events
                .into_iter()
                .map(|r| event_from_row(r, &stream))
                .collect())
        })
    }

    fn save_offset(&mut self) -> Result<()> {
        let position = self.position;
        if position == 0 {
            return Ok(());
        }
        let mut s = self.state.streams.guard();
        save_offset_row(&mut s, &self.stream, &self.consumer, position);
        Ok(())
    }

    fn offset(&self) -> i64 {
        self.position
    }
}

/// `recv()`'s `Err` arm carries `Database` only (ADR-021 §5): a read
/// error mapped elsewhere is remapped into the opaque fallback with
/// the original preserved as the source (the decode-side posture
/// guard — mem reads cannot fail in practice).
fn database_only(e: Error) -> Error {
    match e {
        already @ Error::Database(_) => already,
        other => Error::database(other),
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::store::MemStore;
    use alkstore::{EventReceiver, Store, StreamHandle};

    fn handle(store: &MemStore, name: &str) -> MemStreamHandle {
        MemStreamHandle::new(
            name.to_string(),
            store.state.clone(),
            store.clock.clone(),
            store.wakes.clone(),
        )
    }

    #[tokio::test(flavor = "current_thread")]
    async fn publish_read_round_trip_is_global_fifo_by_offset() {
        let store = MemStore::new();
        let handle = handle(&store, "orders");
        let a = handle.publish(serde_json::json!(1)).await.unwrap();
        let b = handle.publish(serde_json::json!(2)).await.unwrap();
        let keyed = handle
            .publish_with_key(Some("cust-7".into()), serde_json::json!(3))
            .await
            .unwrap();
        assert!(a < b && b < keyed, "offsets are monotone per stream");
        let events = handle.read_since(0, 10).await.unwrap();
        assert_eq!(events.len(), 3);
        assert_eq!(
            events.iter().map(|e| e.offset).collect::<Vec<_>>(),
            vec![a, b, keyed],
            "offset ASC — global FIFO regardless of keys"
        );
        assert_eq!(events[0].stream, "orders");
        assert_eq!(events[0].key, None);
        assert_eq!(events[2].key.as_deref(), Some("cust-7"));
        assert_eq!(events[2].payload_as::<i64>().unwrap(), 3);
        assert!(
            events
                .iter()
                .all(|e| e.created_at == store.clock.unix_now())
        );
    }

    #[tokio::test(flavor = "current_thread")]
    async fn empty_some_key_is_invalid_at_every_entry() {
        let store = MemStore::new();
        let handle = handle(&store, "orders");
        let err = handle
            .publish_with_key(Some(String::new()), serde_json::json!(1))
            .await
            .unwrap_err();
        assert!(matches!(err, alkstore::Error::InvalidName { .. }));
        let err = handle
            .publish_with_key(Some("   ".into()), serde_json::json!(1))
            .await
            .unwrap_err();
        assert!(matches!(err, alkstore::Error::InvalidName { .. }));
        let mut tx = store.begin_tx().await.unwrap();
        let err = tx
            .publish_with_key_tx("orders", Some(String::new()), serde_json::json!(1))
            .await
            .unwrap_err();
        assert!(matches!(err, alkstore::Error::InvalidName { .. }));
    }

    #[tokio::test(flavor = "current_thread")]
    async fn read_extents_clamp_empty() {
        let store = MemStore::new();
        let handle = handle(&store, "orders");
        handle.publish(serde_json::json!(1)).await.unwrap();
        assert!(handle.read_since(0, 0).await.unwrap().is_empty());
        assert!(
            handle.read_since(0, -3).await.unwrap().is_empty(),
            "the LIMIT -1 dialect artifact is dead"
        );
    }

    #[tokio::test(flavor = "current_thread")]
    async fn saves_are_monotone_and_absent_consumers_read_zero() {
        let store = MemStore::new();
        let handle = handle(&store, "orders");
        assert_eq!(
            handle.get_offset("c1").await.unwrap(),
            0,
            "an absent consumer reads 0"
        );
        let tail = handle.publish(serde_json::json!(1)).await.unwrap();
        handle.save_offset("c1", tail).await.unwrap();
        assert_eq!(handle.get_offset("c1").await.unwrap(), tail);
        handle.save_offset("c1", tail - 1).await.unwrap();
        assert_eq!(
            handle.get_offset("c1").await.unwrap(),
            tail,
            "a below-checkpoint save is a silent no-op (monotone)"
        );
        handle.save_offset("c1", tail).await.unwrap();
        assert_eq!(
            handle.get_offset("c1").await.unwrap(),
            tail,
            "an equal save is also a no-op"
        );
        handle.save_offset("c1", tail + 5).await.unwrap();
        assert_eq!(handle.get_offset("c1").await.unwrap(), tail + 5);
    }

    #[tokio::test(flavor = "current_thread")]
    async fn trim_is_exact_boundary_and_wakes_nothing() {
        let store = MemStore::new();
        let handle = handle(&store, "orders");
        let a = handle.publish(serde_json::json!(1)).await.unwrap();
        let b = handle.publish(serde_json::json!(2)).await.unwrap();
        let c = handle.publish(serde_json::json!(3)).await.unwrap();
        assert_eq!(
            handle.trim_to(b - 1).await.unwrap(),
            1,
            "exact boundary: offset <= horizon"
        );
        let events = handle.read_since(0, 10).await.unwrap();
        assert_eq!(
            events.iter().map(|e| e.offset).collect::<Vec<_>>(),
            vec![b, c],
            "surviving rows keep their offsets, never renumbered"
        );
        assert_eq!(
            handle.trim_to(-100).await.unwrap(),
            0,
            "a negative horizon deletes nothing"
        );
        assert_eq!(handle.trim_to(c).await.unwrap(), 2);
        assert_eq!(handle.trim_to(c).await.unwrap(), 0, "idempotent");
        // Reads from a trimmed region resume at the horizon's first
        // remaining row.
        assert!(handle.read_since(0, 10).await.unwrap().is_empty());
        let _ = a;
        assert_eq!(handle.get_offset("late").await.unwrap(), 0);
    }

    #[tokio::test(flavor = "current_thread")]
    async fn subscriber_replays_from_the_saved_offset_and_ride_wakes() {
        let store = MemStore::new();
        let handle = handle(&store, "orders");
        let _a = handle.publish(serde_json::json!(1)).await.unwrap();
        let b = handle.publish(serde_json::json!(2)).await.unwrap();
        handle.save_offset("c1", b).await.unwrap();
        let mut receiver = handle.subscribe("c1").await.unwrap();
        // Everything after the checkpoint replays on attach.
        let c = handle.publish(serde_json::json!(3)).await.unwrap();
        let first = receiver.recv().await.unwrap().unwrap();
        assert_eq!(
            first.offset, c,
            "replay starts at the stored checkpoint's forward tail"
        );
        assert_eq!(first.payload_as::<i64>().unwrap(), 3);
        // Post-attach publishes deliver without polling.
        let d = handle.publish(serde_json::json!(4)).await.unwrap();
        let second = receiver.recv().await.unwrap().unwrap();
        assert_eq!(second.offset, d);
        // The explicit-save receiver form and the direct form compose.
        receiver.save_offset().unwrap();
        assert_eq!(receiver.offset(), d);
        assert_eq!(
            handle.get_offset("c1").await.unwrap(),
            d,
            "the receiver saves the last-yielded offset"
        );
        // A save before any yield is a no-op on the stored checkpoint.
        let mut fresh = handle.subscribe("c2").await.unwrap();
        assert_eq!(fresh.offset(), 0);
        fresh.save_offset().unwrap();
        assert_eq!(handle.get_offset("c2").await.unwrap(), 0);
        let _ = _a;
    }

    #[tokio::test(flavor = "current_thread")]
    async fn read_from_consumer_reads_the_cursor_and_clamps_the_extent() {
        let store = MemStore::new();
        let handle = handle(&store, "orders");
        let a = handle.publish(serde_json::json!(1)).await.unwrap();
        let b = handle.publish(serde_json::json!(2)).await.unwrap();
        let c = handle.publish(serde_json::json!(3)).await.unwrap();
        let head = handle.read_from_consumer("c1", 10).await.unwrap();
        assert_eq!(
            head.iter().map(|e| e.offset).collect::<Vec<_>>(),
            vec![a, b, c],
            "an absent consumer reads from the head (checkpoint 0)"
        );
        handle.save_offset("c1", b).await.unwrap();
        let cursor = handle.read_from_consumer("c1", 10).await.unwrap();
        assert_eq!(
            cursor.iter().map(|e| e.offset).collect::<Vec<_>>(),
            vec![c],
            "the cursor read resumes at the saved checkpoint's forward tail"
        );
        assert!(handle.read_from_consumer("c1", 0).await.unwrap().is_empty());
        assert!(
            handle
                .read_from_consumer("c1", -2)
                .await
                .unwrap()
                .is_empty(),
            "the extent guard clamps non-positive limits to empty"
        );
        handle.save_offset("c1", c).await.unwrap();
        assert!(
            handle
                .read_from_consumer("c1", 10)
                .await
                .unwrap()
                .is_empty(),
            "a cursor at the tail reads empty"
        );
        match handle.read_from_consumer("", 10).await {
            Err(alkstore::Error::InvalidName { .. }) => {}
            _other => panic!("an empty consumer name rejects"),
        }
    }

    #[tokio::test(flavor = "current_thread")]
    async fn saved_sub_horizon_offsets_survive_trim_and_consumer_reads_resume_at_the_horizon() {
        let store = MemStore::new();
        let handle = handle(&store, "orders");
        let a = handle.publish(serde_json::json!(1)).await.unwrap();
        let b = handle.publish(serde_json::json!(2)).await.unwrap();
        let _c = handle.publish(serde_json::json!(3)).await.unwrap();
        handle.save_offset("c1", a).await.unwrap();
        assert_eq!(handle.trim_to(b - 1).await.unwrap(), 1);
        assert_eq!(
            handle.get_offset("c1").await.unwrap(),
            a,
            "the saved sub-horizon checkpoint is untouched by trim — offsets are never renumbered"
        );
        let resumed = handle.read_from_consumer("c1", 10).await.unwrap();
        assert_eq!(
            resumed.iter().map(|e| e.offset).collect::<Vec<_>>(),
            vec![b, _c],
            "reads from a trimmed-away region resume at the horizon's first remaining row"
        );
    }

    #[tokio::test(flavor = "current_thread")]
    async fn receiver_try_recv_reads_without_await() {
        let store = MemStore::new();
        let handle = handle(&store, "orders");
        let mut receiver = handle.subscribe("c1").await.unwrap();
        assert!(try_none(receiver.as_mut()), "idle reads as Ok(None)");
        let a = handle.publish(serde_json::json!(1)).await.unwrap();
        let got = receiver.try_recv().unwrap().unwrap();
        assert_eq!(got.offset, a);
        assert!(try_none(receiver.as_mut()));
    }

    fn try_none(receiver: &mut dyn EventReceiver) -> bool {
        matches!(receiver.try_recv(), Ok(None))
    }

    #[tokio::test(flavor = "current_thread")]
    async fn engine_drop_closes_the_receiver_on_both_arms() {
        let (mut receiver, offset) = {
            let store = MemStore::new();
            let handle = handle(&store, "orders");
            let offset = handle.publish(serde_json::json!(1)).await.unwrap();
            (handle.subscribe("c1").await.unwrap(), offset)
        };
        // The engine is gone; the replayed pre-drop event still
        // delivers (it was committed and visible), then the receiver
        // is closed — terminal on both arms (never reopens).
        let delivered = receiver
            .try_recv()
            .unwrap()
            .expect("the pre-drop event replays");
        assert_eq!(delivered.offset, offset);
        assert!(matches!(receiver.try_recv(), Err(alkstore::Error::Closed)));
        let closed = tokio::time::timeout(std::time::Duration::from_millis(200), receiver.recv())
            .await
            .expect("the closed feed parks no receiver");
        assert!(
            closed.is_none(),
            "the terminal None arm (engine drop is the close cause)"
        );
    }

    #[tokio::test(flavor = "current_thread")]
    async fn trim_wakes_nothing_for_an_open_subscriber() {
        let store = MemStore::new();
        let handle = handle(&store, "orders");
        let a = handle.publish(serde_json::json!(1)).await.unwrap();
        let b = handle.publish(serde_json::json!(2)).await.unwrap();
        let mut receiver = handle.subscribe("c1").await.unwrap();
        assert_eq!(receiver.recv().await.unwrap().unwrap().offset, a);
        assert_eq!(receiver.recv().await.unwrap().unwrap().offset, b);
        handle.trim_to(b).await.unwrap();
        let quiet =
            tokio::time::timeout(std::time::Duration::from_millis(50), receiver.recv()).await;
        assert!(quiet.is_err(), "trim emits no wake and delivers no event");
        let _ = receiver.read_since(0, 10).await.unwrap();
    }
}