//! the replay-then-tail loop every event stream runs.
//!
//! positions come from `db::sequencer`, so broadcasts arrive in position
//! order, and every committed row is either broadcast or covered by a later
//! persisted marker. a subscriber only has to take broadcasts in the order
//! they arrive, reading the db for a marker's rows, and one that falls behind
//! the broadcast channel can always resume from the db at its position.

use futures::future::{Either, select};
use std::collections::VecDeque;
use std::fmt;
use std::num::NonZeroUsize;
use std::sync::Arc;
use std::task::{Context, Poll, Wake, Waker};
use std::time::Instant;
use tokio::sync::broadcast::error::{RecvError, TryRecvError};
use tokio::sync::mpsc::error::TrySendError;
use tokio::sync::mpsc::{OwnedPermit, Permit};
use tokio::sync::{broadcast, mpsc};
use tracing::{debug, warn};

use super::replay_cap::{Pacer, Read};
use super::types::{ReplayChunk, ReplayFailed, StreamOptions, StreamTooSlow};

/// an event broadcast after its commit.
pub(crate) trait StreamBroadcast: Clone {
    /// where the event sits in commit order.
    type Position: Ord + Copy;

    fn position(&self) -> Self::Position;

    /// a persisted marker carries no event: every row through its position is
    /// committed, and the ones since the previous broadcast weren't broadcast.
    fn is_marker(&self) -> bool;
}

/// what one stream endpoint reads and sends.
pub(crate) trait StreamSource {
    type Broadcast: StreamBroadcast;
    type Output;

    /// up to `limit` outputs for the committed rows after `after`, through
    /// `through` when it is set.
    fn read(
        &mut self,
        after: Option<Position<Self>>,
        through: Option<Position<Self>>,
        limit: usize,
    ) -> Result<ReplayChunk<Self::Output, Position<Self>>, ReplayFailed>;

    /// the output for a live broadcast, or none when there's nothing to send.
    fn render(&mut self, event: Self::Broadcast) -> Option<Self::Output>;

    /// how many positions the stream has taken, which a capped replay reads
    /// as the live rate. an atomic load, so the commit path pays nothing.
    fn written(&self) -> u64;

    /// about how many bytes `output` takes on the wire, which a capped replay
    /// pays for.
    fn frame_bytes(output: &Self::Output) -> usize;

    /// whether a marker's rows are worth reading. a subscriber that filters
    /// them all out can skip the read.
    fn wants_marker(&self, _marker: &Self::Broadcast) -> bool {
        true
    }
}

pub(crate) type Position<S> = <<S as StreamSource>::Broadcast as StreamBroadcast>::Position;

/// where a subscriber begins.
pub(crate) struct Start<P> {
    /// the last position it has seen, none for before the first event
    after: Option<P>,
    /// replay the db through this position before tailing
    catch_up: Option<P>,
}

impl<P: Ord + Copy> Start<P> {
    /// tail live broadcasts after `head`.
    pub(crate) fn live(head: Option<P>) -> Self {
        Self {
            after: head,
            catch_up: None,
        }
    }

    /// replay the rows after `after` through `head`, then tail.
    pub(crate) fn replay(after: Option<P>, head: Option<P>) -> Self {
        Self {
            after,
            catch_up: head.filter(|head| after < Some(*head)),
        }
    }
}

enum CatchUp<P> {
    Through(P),
    ToEnd,
}

/// the stream is over: the subscriber hung up, was sent a closing error, or
/// the db couldn't be read.
struct Ended;

/// subscribe `event_rx` before reading the head `start` was built from, so
/// nothing committed in between is missed.
pub(crate) fn run_stream<S, E>(
    tx: mpsc::Sender<Result<S::Output, E>>,
    event_rx: broadcast::Receiver<S::Broadcast>,
    start: Start<Position<S>>,
    opts: StreamOptions,
    mut source: S,
) where
    S: StreamSource,
    E: From<StreamTooSlow> + fmt::Display,
{
    let mut out = Output::new(tx);
    let mut live = Live::new(event_rx, opts.pending_event_limit);
    let mut after = start.after;
    let mut catch_up = start.catch_up.map(CatchUp::Through);
    // only the replay the cursor asked for is capped. markers and lag catch ups are as big as
    // what got written meanwhile, so they're live work, and capping them would lag the tail
    let mut pacer = catch_up
        .as_ref()
        .and(opts.replay_cap)
        .map(|cap| Pacer::new(cap, source.written(), Instant::now()));

    loop {
        if out.tx.is_closed() {
            return;
        }

        if let Some(target) = catch_up.take() {
            let replayed = replay(
                &mut out,
                &mut source,
                &mut live,
                &mut after,
                target,
                opts,
                pacer.take(),
            );
            if replayed.is_err() {
                return;
            }
            continue;
        }

        let event = match live.next(&out.tx) {
            Next::Event(event) => event,
            Next::Lagged => {
                debug!("stream subscriber fell behind the broadcast, catching up from the db");
                catch_up = Some(CatchUp::ToEnd);
                continue;
            }
            Next::Closed => return,
        };

        let position = event.position();
        if after.is_some_and(|after| position <= after) {
            continue;
        }
        if event.is_marker() {
            if source.wants_marker(&event) {
                catch_up = Some(CatchUp::Through(position));
            } else {
                after = Some(position);
            }
            continue;
        }
        if let Some(output) = source.render(event)
            && send(&mut out, output, opts, &mut live).is_err()
        {
            return;
        }
        after = Some(position);
    }
}

/// send the db's rows after `after` until `target`, moving `after` along.
fn replay<S, E>(
    out: &mut Output<S::Output, E>,
    source: &mut S,
    live: &mut Live<S::Broadcast>,
    after: &mut Option<Position<S>>,
    target: CatchUp<Position<S>>,
    opts: StreamOptions,
    mut pacer: Option<Pacer>,
) -> Result<(), Ended>
where
    S: StreamSource,
    E: From<StreamTooSlow> + fmt::Display,
{
    let through = match target {
        CatchUp::Through(position) => Some(position),
        CatchUp::ToEnd => None,
    };
    let mut blocked_since: Option<Instant> = None;

    loop {
        // skip the db read while the output channel is saturated
        let Some(limit) = replay_chunk_size_for(&out.tx, opts.replay_chunk_size) else {
            let deadline = *blocked_since.get_or_insert_with(Instant::now) + opts.send_timeout;
            match live.wait_for_room(&out.tx, deadline) {
                Room::Ready(_) => continue,
                Room::Closed => return Err(Ended),
                Room::TimedOut => {}
            }
            return Err(out.close(StreamTooSlow::send_timeout(opts.send_timeout).into()));
        };
        blocked_since = None;

        let chunk = source
            .read(*after, through, limit)
            .map_err(|ReplayFailed| Ended)?;
        let mut read = Read {
            scanned: chunk.scanned,
            sent: chunk.events.len(),
            sent_bytes: 0,
        };
        for output in chunk.events {
            read.sent_bytes += S::frame_bytes(&output);
            send(out, output, opts, live)?;
        }
        *after = (*after).max(chunk.last_seen);

        if chunk.exhausted {
            // every row through the target is committed, so none are left
            *after = (*after).max(through);
            return Ok(());
        }
        if let Some(pacer) = &mut pacer {
            let wait = pacer.charge(read, source.written(), Instant::now());
            // a reader that hangs up mid wait ends the thread right away
            if !wait.is_zero() && block_on_until(out.tx.closed(), Instant::now() + wait).is_some() {
                return Err(Ended);
            }
        }
    }
}

/// send one output, waiting out a full channel for up to the send timeout.
fn send<O, E, B>(
    out: &mut Output<O, E>,
    output: O,
    opts: StreamOptions,
    live: &mut Live<B>,
) -> Result<(), Ended>
where
    B: Clone,
    E: From<StreamTooSlow> + fmt::Display,
{
    match live.wait_for_room(&out.tx, Instant::now() + opts.send_timeout) {
        Room::Ready(permit) => {
            permit.send(Ok(output));
            return Ok(());
        }
        Room::Closed => return Err(Ended),
        Room::TimedOut => {}
    }
    Err(out.close(StreamTooSlow::send_timeout(opts.send_timeout).into()))
}

/// what waiting for a slot in the subscriber's channel came to
enum Room<'a, T> {
    Ready(Permit<'a, T>),
    /// the subscriber hung up
    Closed,
    /// still full at the deadline
    TimedOut,
}

/// runs `fut` on this thread until it finishes or `deadline` passes. the thread sleeps until
/// something wakes the future, so a wait costs nothing however long it is
fn block_on_until<F: Future>(fut: F, deadline: Instant) -> Option<F::Output> {
    struct Unpark(std::thread::Thread);

    impl Wake for Unpark {
        fn wake(self: Arc<Self>) {
            self.0.unpark();
        }
    }

    let waker = Waker::from(Arc::new(Unpark(std::thread::current())));
    let mut cx = Context::from_waker(&waker);
    let mut fut = std::pin::pin!(fut);
    loop {
        if let Poll::Ready(output) = fut.as_mut().poll(&mut cx) {
            return Some(output);
        }
        let left = deadline.saturating_duration_since(Instant::now());
        if left.is_zero() {
            return None;
        }
        std::thread::park_timeout(left);
    }
}

/// the subscriber's channel, with one slot held from the start so a closing
/// error always fits behind the outputs already queued.
struct Output<O, E> {
    tx: mpsc::Sender<Result<O, E>>,
    closing: Option<OwnedPermit<Result<O, E>>>,
}

impl<O, E: fmt::Display> Output<O, E> {
    fn new(tx: mpsc::Sender<Result<O, E>>) -> Self {
        let closing = tx.clone().try_reserve_owned().ok();
        Self { tx, closing }
    }

    /// queue `err` as the last item, ending the stream.
    fn close(&mut self, err: E) -> Ended {
        warn!(%err, "closing stream subscriber");
        if let Some(closing) = self.closing.take() {
            closing.send(Err(err));
        }
        Ended
    }
}

/// the broadcasts a subscriber hasn't taken yet: the broadcast channel, plus
/// a bounded backlog drained from it while a send waits, so a briefly slow
/// subscriber doesn't lag the channel.
struct Live<B> {
    rx: broadcast::Receiver<B>,
    backlog: VecDeque<B>,
    limit: usize,
    /// broadcasts after the backlog were lost
    lagged: bool,
}

enum Next<B> {
    Event(B),
    Lagged,
    /// the broadcast channel closed, or the subscriber hung up
    Closed,
}

impl<B: Clone> Live<B> {
    fn new(rx: broadcast::Receiver<B>, limit: usize) -> Self {
        Self {
            rx,
            backlog: VecDeque::new(),
            limit,
            lagged: false,
        }
    }

    /// move waiting broadcasts into the backlog, up to its limit. past it the
    /// channel holds them, until it lags.
    fn drain(&mut self) {
        while !self.lagged && self.backlog.len() < self.limit {
            match self.rx.try_recv() {
                Ok(event) => self.backlog.push_back(event),
                Err(TryRecvError::Lagged(_)) => self.lagged = true,
                Err(TryRecvError::Empty | TryRecvError::Closed) => return,
            }
        }
    }

    /// waits for a slot in `tx` until `deadline`, taking broadcasts into the backlog meanwhile
    /// so a subscriber blocked for a moment doesn't lag the channel
    fn wait_for_room<'a, T>(&mut self, tx: &'a mpsc::Sender<T>, deadline: Instant) -> Room<'a, T> {
        match tx.try_reserve() {
            Ok(permit) => return Room::Ready(permit),
            Err(TrySendError::Closed(())) => return Room::Closed,
            Err(TrySendError::Full(())) => {}
        }
        loop {
            self.drain();
            let taking = !self.lagged && self.backlog.len() < self.limit;
            let rx = &mut self.rx;
            let woke = block_on_until(
                async {
                    tokio::select! {
                        biased;
                        permit = tx.reserve() => Ok(permit),
                        event = rx.recv(), if taking => Err(event),
                    }
                },
                deadline,
            );
            match woke {
                None => return Room::TimedOut,
                Some(Ok(Ok(permit))) => return Room::Ready(permit),
                Some(Ok(Err(_))) => return Room::Closed,
                Some(Err(Ok(event))) => self.backlog.push_back(event),
                Some(Err(Err(RecvError::Lagged(_)))) => self.lagged = true,
                // a closed broadcast ends the stream once the backlog is sent
                Some(Err(Err(RecvError::Closed))) => {
                    return match block_on_until(tx.reserve(), deadline) {
                        None => Room::TimedOut,
                        Some(Ok(permit)) => Room::Ready(permit),
                        Some(Err(_)) => Room::Closed,
                    };
                }
            }
        }
    }

    /// the next broadcast, waiting for one unless the subscriber on `tx`
    /// hangs up first.
    fn next<T>(&mut self, tx: &mpsc::Sender<T>) -> Next<B> {
        if let Some(event) = self.backlog.pop_front() {
            return Next::Event(event);
        }
        if std::mem::take(&mut self.lagged) {
            return Next::Lagged;
        }
        let recv = std::pin::pin!(self.rx.recv());
        let hung_up = std::pin::pin!(tx.closed());
        match futures::executor::block_on(select(recv, hung_up)) {
            Either::Left((Ok(event), _)) => Next::Event(event),
            Either::Left((Err(RecvError::Lagged(_)), _)) => Next::Lagged,
            Either::Left((Err(RecvError::Closed), _)) | Either::Right(_) => Next::Closed,
        }
    }
}

fn replay_chunk_size_for<T>(
    tx: &mpsc::Sender<T>,
    configured: Option<NonZeroUsize>,
) -> Option<usize> {
    let available = tx.capacity();

    if available == 0 {
        return None;
    }

    let max_cap = tx.max_capacity().max(1);
    let desired = configured
        .map(NonZeroUsize::get)
        .unwrap_or_else(|| (max_cap / 2).max(1))
        .min(max_cap);

    Some(desired.min(available).max(1))
}

#[cfg(all(test, any(feature = "indexer_stream", feature = "relay")))]
mod tests {
    use super::*;
    use crate::control::stream::replay_cap::ReplayCap;
    use std::collections::BTreeSet;
    use std::num::NonZeroU64;
    use std::sync::Mutex;
    use std::thread::JoinHandle;
    use std::time::Duration;

    #[derive(Clone, Debug)]
    enum Ev {
        /// a committed row, broadcast live
        Row(u64),
        /// a broadcast that has no row
        Ephemeral(u64),
        Marker(u64),
    }

    impl StreamBroadcast for Ev {
        type Position = u64;

        fn position(&self) -> u64 {
            match self {
                Self::Row(id) | Self::Ephemeral(id) | Self::Marker(id) => *id,
            }
        }

        fn is_marker(&self) -> bool {
            matches!(self, Self::Marker(_))
        }
    }

    /// the db and broadcast channel the way the sequencer drives them:
    /// commit, then broadcast, in position order.
    struct Log {
        db: Mutex<BTreeSet<u64>>,
        tx: Mutex<Option<broadcast::Sender<Ev>>>,
        next: Mutex<u64>,
        reads: Mutex<usize>,
        reads_fail: Mutex<bool>,
    }

    impl Log {
        fn new(capacity: usize) -> (Arc<Self>, broadcast::Receiver<Ev>) {
            let (tx, rx) = broadcast::channel(capacity);
            let log = Self {
                db: Mutex::new(BTreeSet::new()),
                tx: Mutex::new(Some(tx)),
                next: Mutex::new(1),
                reads: Mutex::new(0),
                reads_fail: Mutex::new(false),
            };
            (Arc::new(log), rx)
        }

        fn take(&self) -> u64 {
            let mut next = self.next.lock().unwrap();
            *next += 1;
            *next - 1
        }

        fn broadcast(&self, ev: Ev) {
            if let Some(tx) = &*self.tx.lock().unwrap() {
                let _ = tx.send(ev);
            }
        }

        fn live(&self) -> u64 {
            let id = self.take();
            self.db.lock().unwrap().insert(id);
            self.broadcast(Ev::Row(id));
            id
        }

        fn ephemeral(&self) -> u64 {
            let id = self.take();
            self.broadcast(Ev::Ephemeral(id));
            id
        }

        /// rows written without a broadcast, like a backfill
        fn backfill(&self, n: usize) -> Vec<u64> {
            let ids: Vec<_> = (0..n).map(|_| self.take()).collect();
            self.db.lock().unwrap().extend(&ids);
            self.broadcast(Ev::Marker(*ids.last().unwrap()));
            ids
        }

        fn head(&self) -> Option<u64> {
            self.next
                .lock()
                .unwrap()
                .checked_sub(1)
                .filter(|id| *id > 0)
        }

        fn close(&self) {
            self.tx.lock().unwrap().take();
        }

        fn fail_reads(&self) {
            *self.reads_fail.lock().unwrap() = true;
        }

        /// wait until the subscriber has taken every broadcast off the channel
        fn wait_taken(&self) {
            let deadline = Instant::now() + Duration::from_secs(5);
            while self
                .tx
                .lock()
                .unwrap()
                .as_ref()
                .is_some_and(|tx| !tx.is_empty())
            {
                assert!(
                    Instant::now() < deadline,
                    "the subscriber never took a broadcast"
                );
                std::thread::sleep(Duration::from_millis(1));
            }
        }
    }

    /// runs before each db read, to commit at exact points of a replay
    type OnRead = Box<dyn FnMut(&Log, usize) + Send>;

    struct Source {
        log: Arc<Log>,
        wants_markers: bool,
        /// the rows the subscriber wants, the rest are scanned and filtered out
        sends: fn(u64) -> bool,
        on_read: OnRead,
    }

    impl StreamSource for Source {
        type Broadcast = Ev;
        type Output = u64;

        fn read(
            &mut self,
            after: Option<u64>,
            through: Option<u64>,
            limit: usize,
        ) -> Result<ReplayChunk<u64, u64>, ReplayFailed> {
            let call = {
                let mut reads = self.log.reads.lock().unwrap();
                *reads += 1;
                *reads - 1
            };
            (self.on_read)(&self.log, call);
            if *self.log.reads_fail.lock().unwrap() {
                return Err(ReplayFailed);
            }
            let start = after.map_or(0, |after| after + 1);
            let end = through.unwrap_or(u64::MAX);
            let db = self.log.db.lock().unwrap();
            let mut rows = db.range(start..=end).copied().peekable();
            let mut chunk = ReplayChunk {
                events: Vec::new(),
                last_seen: after,
                exhausted: false,
                scanned: 0,
            };
            // up to four rows scanned per output, like the real reads
            while chunk.events.len() < limit && chunk.scanned < limit * 4 {
                let Some(row) = rows.next() else { break };
                chunk.scanned += 1;
                chunk.last_seen = Some(row);
                chunk
                    .events
                    .extend(Some(row).filter(|row| (self.sends)(*row)));
            }
            chunk.exhausted = rows.peek().is_none();
            Ok(chunk)
        }

        fn written(&self) -> u64 {
            *self.log.next.lock().unwrap()
        }

        fn frame_bytes(_: &u64) -> usize {
            FRAME_BYTES
        }

        fn render(&mut self, event: Ev) -> Option<u64> {
            match event {
                Ev::Row(id) | Ev::Ephemeral(id) => Some(id),
                Ev::Marker(_) => unreachable!("markers are handled by the engine"),
            }
        }

        fn wants_marker(&self, _marker: &Ev) -> bool {
            self.wants_markers
        }
    }

    type Out = mpsc::Receiver<Result<u64, StreamTooSlow>>;

    const FRAME_BYTES: usize = 100;

    fn opts() -> StreamOptions {
        StreamOptions {
            replay_chunk_size: NonZeroUsize::new(2),
            replay_cap: None,
            pending_event_limit: 4,
            send_timeout: Duration::from_secs(5),
        }
    }

    /// run a subscriber whose channel holds `out_capacity` outputs, plus the
    /// slot the engine keeps for a closing error.
    fn spawn(
        log: &Arc<Log>,
        event_rx: broadcast::Receiver<Ev>,
        start: Start<u64>,
        out_capacity: usize,
        opts: StreamOptions,
        on_read: impl FnMut(&Log, usize) + Send + 'static,
    ) -> (JoinHandle<()>, Out) {
        let (tx, out) = mpsc::channel(out_capacity + 1);
        let source = Source {
            log: log.clone(),
            wants_markers: true,
            sends: |_| true,
            on_read: Box::new(on_read),
        };
        let handle = std::thread::spawn(move || run_stream(tx, event_rx, start, opts, source));
        (handle, out)
    }

    fn take(out: &mut Out, n: usize) -> Vec<u64> {
        (0..n)
            .map(|_| {
                let deadline = Instant::now() + Duration::from_secs(5);
                loop {
                    match out.try_recv() {
                        Ok(item) => break item.expect("stream error"),
                        Err(mpsc::error::TryRecvError::Empty) if Instant::now() < deadline => {
                            std::thread::sleep(Duration::from_millis(5));
                        }
                        Err(err) => panic!("no output: {err}"),
                    }
                }
            })
            .collect()
    }

    fn finish(log: &Log, handle: JoinHandle<()>, mut out: Out) {
        log.close();
        handle.join().unwrap();
        assert!(out.try_recv().is_err(), "unexpected extra output");
    }

    /// wait for the stream thread to end by itself
    fn ends(handle: JoinHandle<()>) {
        let deadline = Instant::now() + Duration::from_secs(5);
        while !handle.is_finished() {
            assert!(Instant::now() < deadline, "the stream never ended");
            std::thread::sleep(Duration::from_millis(5));
        }
        handle.join().unwrap();
    }

    fn nothing_on_read(_: &Log, _: usize) {}

    #[test]
    fn replays_in_chunks_then_tails() {
        let (log, rx) = Log::new(16);
        let rows = log.backfill(5);
        let (handle, mut out) = spawn(
            &log,
            rx,
            Start::replay(None, log.head()),
            16,
            opts(),
            nothing_on_read,
        );

        assert_eq!(take(&mut out, 5), rows);
        let live = log.live();
        assert_eq!(take(&mut out, 1), [live]);
        finish(&log, handle, out);
    }

    #[test]
    fn a_live_start_skips_everything_through_its_head() {
        let (log, rx) = Log::new(16);
        // already broadcast, so they wait in the subscriber's channel
        log.backfill(3);
        log.live();
        let (handle, mut out) = spawn(&log, rx, Start::live(log.head()), 16, opts(), |_, _| {
            panic!("rows through the head need no db read");
        });

        let live = log.live();
        assert_eq!(take(&mut out, 1), [live]);
        finish(&log, handle, out);
    }

    #[test]
    fn a_replay_start_reads_only_past_its_cursor() {
        let (log, rx) = Log::new(16);
        let rows = log.backfill(4);
        let (handle, mut out) = spawn(
            &log,
            rx,
            Start::replay(Some(rows[1]), log.head()),
            16,
            opts(),
            nothing_on_read,
        );
        assert_eq!(take(&mut out, 2), rows[2..]);
        let live = log.live();
        assert_eq!(take(&mut out, 1), [live]);
        finish(&log, handle, out);

        let (log, rx) = Log::new(16);
        log.backfill(2);
        let (handle, mut out) = spawn(
            &log,
            rx,
            Start::replay(log.head(), log.head()),
            16,
            opts(),
            |_, _| panic!("a cursor at the head has nothing to replay"),
        );
        let live = log.live();
        assert_eq!(take(&mut out, 1), [live]);
        finish(&log, handle, out);
    }

    #[test]
    fn a_marker_reads_its_rows_from_the_db() {
        let (log, rx) = Log::new(16);
        let (handle, mut out) = spawn(
            &log,
            rx,
            Start::live(log.head()),
            16,
            opts(),
            nothing_on_read,
        );

        let first = log.live();
        let rows = log.backfill(3);
        let last = log.live();
        let expected: Vec<_> = [first].into_iter().chain(rows).chain([last]).collect();
        assert_eq!(take(&mut out, 5), expected);
        finish(&log, handle, out);
    }

    #[test]
    fn broadcast_only_events_keep_their_place() {
        let (log, rx) = Log::new(16);
        let (handle, mut out) = spawn(
            &log,
            rx,
            Start::live(log.head()),
            16,
            opts(),
            nothing_on_read,
        );

        let rows = log.backfill(2);
        let ephemeral = log.ephemeral();
        let more = log.backfill(2);
        let expected: Vec<_> = rows.into_iter().chain([ephemeral]).chain(more).collect();
        assert_eq!(take(&mut out, 5), expected);
        finish(&log, handle, out);
    }

    #[test]
    fn commits_during_a_replay_are_sent_once() {
        let (log, rx) = Log::new(16);
        let rows = log.backfill(3);
        let (handle, mut out) = spawn(
            &log,
            rx,
            Start::replay(None, log.head()),
            16,
            opts(),
            |log, call| {
                if call == 0 {
                    // lands past the head the replay started with, broadcast
                    // before the replay reaches it
                    log.live();
                    log.backfill(1);
                }
            },
        );

        let sent = take(&mut out, 5);
        assert_eq!(sent[..3], rows[..]);
        assert_eq!(sent, (rows[0]..rows[0] + 5).collect::<Vec<_>>());
        finish(&log, handle, out);
    }

    #[test]
    fn a_lagging_subscriber_catches_up_from_the_db() {
        let (log, rx) = Log::new(2);
        let opts = StreamOptions {
            pending_event_limit: 1,
            ..opts()
        };
        let (handle, mut out) = spawn(&log, rx, Start::live(log.head()), 1, opts, nothing_on_read);

        let first = log.live();
        std::thread::sleep(Duration::from_millis(50));
        // the subscriber is stuck sending while these overflow the channel
        // and its backlog
        let rest: Vec<_> = (0..8).map(|_| log.live()).collect();
        let expected: Vec<_> = [first].into_iter().chain(rest).collect();
        assert_eq!(take(&mut out, 9), expected);

        let live = log.live();
        assert_eq!(take(&mut out, 1), [live]);
        finish(&log, handle, out);
    }

    #[test]
    fn the_backlog_holds_broadcasts_while_a_send_waits() {
        let (log, rx) = Log::new(2);
        let (handle, mut out) = spawn(&log, rx, Start::live(log.head()), 1, opts(), |_, _| {
            panic!("a backlog within its limit needs no db read");
        });

        // fill the output, then leave the subscriber waiting to send the next
        let mut sent = vec![log.live()];
        log.wait_taken();
        sent.push(log.ephemeral());
        log.wait_taken();
        // more than the broadcast channel holds, taken into the backlog while
        // that send waits. they have no rows, so a lag would lose them
        for _ in 0..3 {
            sent.push(log.ephemeral());
            log.wait_taken();
        }
        assert_eq!(take(&mut out, 5), sent);
        finish(&log, handle, out);
    }

    #[test]
    fn unwanted_markers_are_skipped_without_a_read() {
        let (log, rx) = Log::new(16);
        let (tx, mut out) = mpsc::channel(16);
        let source = Source {
            log: log.clone(),
            wants_markers: false,
            sends: |_| true,
            on_read: Box::new(|_, _| panic!("an unwanted marker needs no db read")),
        };
        let handle =
            std::thread::spawn(move || run_stream(tx, rx, Start::live(None), opts(), source));

        log.backfill(3);
        let live = log.live();
        assert_eq!(take(&mut out, 1), [live]);
        finish(&log, handle, out);
    }

    #[test]
    fn a_stuck_subscriber_is_closed() {
        let (log, rx) = Log::new(16);
        let rows = log.backfill(2);
        let opts = StreamOptions {
            send_timeout: Duration::ZERO,
            ..opts()
        };
        let (handle, mut out) = spawn(
            &log,
            rx,
            Start::replay(None, log.head()),
            1,
            opts,
            nothing_on_read,
        );
        handle.join().unwrap();

        // the error comes after what was queued, even though that filled the channel
        assert_eq!(out.try_recv().unwrap().unwrap(), rows[0]);
        assert!(matches!(out.try_recv(), Ok(Err(StreamTooSlow { .. }))));
        assert!(out.try_recv().is_err());
    }

    #[test]
    fn a_hung_up_subscriber_ends_without_a_broadcast() {
        let (log, rx) = Log::new(16);
        let (handle, out) = spawn(
            &log,
            rx,
            Start::live(log.head()),
            16,
            opts(),
            nothing_on_read,
        );

        // most likely waiting for a broadcast by now
        std::thread::sleep(Duration::from_millis(50));
        drop(out);
        ends(handle);
    }

    #[test]
    fn a_failed_read_ends_the_stream() {
        let (log, rx) = Log::new(16);
        let rows = log.backfill(5);
        let (handle, mut out) = spawn(
            &log,
            rx,
            Start::replay(None, log.head()),
            16,
            opts(),
            |log, call| {
                if call == 1 {
                    log.fail_reads();
                }
            },
        );

        // rather than skip the rows it couldn't read and go on tailing
        assert_eq!(take(&mut out, 2), rows[..2]);
        ends(handle);
        log.live();
        assert!(out.try_recv().is_err());
    }

    /// a subscriber that reads as fast as it can never waits on a timer: the thread sends again
    /// the moment there's room, so many times the channel's size goes through in no time
    #[test]
    fn a_fast_subscriber_is_never_held_back_by_a_full_channel() {
        const N: usize = 20_000;
        const CAPACITY: usize = 16;
        let (log, rx) = Log::new(16);
        let rows = log.backfill(N);
        let opts = StreamOptions {
            replay_chunk_size: NonZeroUsize::new(4),
            ..opts()
        };
        let started = Instant::now();
        let (handle, mut out) = spawn(
            &log,
            rx,
            Start::replay(None, log.head()),
            CAPACITY,
            opts,
            nothing_on_read,
        );
        let sent: Vec<_> = (0..N)
            .map(|_| out.blocking_recv().unwrap().unwrap())
            .collect();
        let took = started.elapsed();
        assert_eq!(sent, rows);
        // sleeping 10 ms whenever the channel filled took N / CAPACITY x 10 ms, 12.5 s here
        assert!(took < Duration::from_secs(2), "{took:?}");
        finish(&log, handle, out);
    }

    /// whether it fills up mid chunk or between chunks, a subscriber that stops reading is
    /// closed once it has been full for the send timeout, not before
    #[test]
    fn a_subscriber_that_stops_reading_is_closed_after_the_send_timeout() {
        for chunk in [1, 2] {
            let (log, rx) = Log::new(16);
            let rows = log.backfill(4);
            let opts = StreamOptions {
                replay_chunk_size: NonZeroUsize::new(chunk),
                send_timeout: Duration::from_millis(300),
                ..opts()
            };
            let started = Instant::now();
            let (handle, mut out) = spawn(
                &log,
                rx,
                Start::replay(None, log.head()),
                1,
                opts,
                nothing_on_read,
            );
            ends(handle);
            let took = started.elapsed();
            assert!(
                took >= Duration::from_millis(300),
                "chunk {chunk}: {took:?}"
            );
            assert_eq!(out.try_recv().unwrap().unwrap(), rows[0]);
            assert!(matches!(out.try_recv(), Ok(Err(StreamTooSlow { .. }))));
            assert!(out.try_recv().is_err());
        }
    }

    /// a reader whose replay is capped at a floor of `floor` bytes a second, on a box whose
    /// head never moves
    fn capped(floor: u64) -> StreamOptions {
        StreamOptions {
            replay_chunk_size: NonZeroUsize::new(250),
            replay_cap: Some(ReplayCap::new(10, NonZeroU64::new(floor).unwrap())),
            ..opts()
        }
    }

    fn spawn_capped(
        log: &Arc<Log>,
        event_rx: broadcast::Receiver<Ev>,
        start: Start<u64>,
        opts: StreamOptions,
        sends: fn(u64) -> bool,
    ) -> (JoinHandle<()>, Out) {
        let (tx, out) = mpsc::channel(1024);
        let source = Source {
            log: log.clone(),
            wants_markers: true,
            sends,
            on_read: Box::new(nothing_on_read),
        };
        let handle = std::thread::spawn(move || run_stream(tx, event_rx, start, opts, source));
        (handle, out)
    }

    fn receive(out: &mut Out, n: usize) -> Vec<u64> {
        (0..n)
            .map(|_| out.blocking_recv().unwrap().unwrap())
            .collect()
    }

    #[test]
    fn a_cursor_replay_is_held_to_the_cap() {
        let (log, rx) = Log::new(16);
        let rows = log.backfill(3000);
        let started = Instant::now();
        let start = Start::replay(None, log.head());
        // 1000 rows a second
        let (handle, mut out) = spawn_capped(&log, rx, start, capped(100_000), |_| true);
        assert_eq!(receive(&mut out, 3000), rows);
        // a second banked, then the last chunk is read once 2750 rows are paid for
        let took = started.elapsed();
        assert!(took >= Duration::from_millis(1500), "{took:?}");
        finish(&log, handle, out);
    }

    #[test]
    fn a_filtered_replay_pays_for_the_rows_it_scans() {
        let (log, rx) = Log::new(16);
        let rows = log.backfill(4000);
        let wanted: Vec<_> = rows.into_iter().filter(|row| row % 50 == 0).collect();
        let started = Instant::now();
        let start = Start::replay(None, log.head());
        let (handle, mut out) = spawn_capped(&log, rx, start, capped(32_000), |row| row % 50 == 0);
        assert_eq!(receive(&mut out, wanted.len()), wanted);
        // reads of 1000 rows for 20 sends cost 33 kb each, so the middle two wait a second, where
        // the 2 kb they send would never wait
        let took = started.elapsed();
        assert!(took >= Duration::from_millis(1500), "{took:?}");
        finish(&log, handle, out);
    }

    #[test]
    fn a_quiet_box_still_replays_at_the_floor() {
        let (log, rx) = Log::new(16);
        let rows = log.backfill(1500);
        let start = Start::replay(None, log.head());
        let (handle, mut out) = spawn_capped(&log, rx, start, capped(100_000), |_| true);
        // the head never moves, so without the floor the replay would never get past its first second
        assert_eq!(take(&mut out, 1500), rows);
        finish(&log, handle, out);
    }

    #[test]
    fn a_capped_replay_leaves_the_live_tail_alone() {
        let (log, rx) = Log::new(16);
        let rows = log.backfill(2000);
        let start = Start::replay(None, log.head());
        let (handle, mut out) = spawn_capped(&log, rx, start, capped(100_000), |_| true);
        assert_eq!(take(&mut out, 2000), rows);

        // a backfill's marker on the tail costs as much as a replay but isn't one, so it reads
        // its rows at once. even a fresh second of budget would leave 1500 of them to wait out
        let started = Instant::now();
        let backfilled = log.backfill(2500);
        let live = log.live();
        let expected: Vec<_> = backfilled.into_iter().chain([live]).collect();
        assert_eq!(take(&mut out, 2501), expected);
        let took = started.elapsed();
        assert!(took < Duration::from_millis(500), "{took:?}");
        finish(&log, handle, out);
    }

    #[test]
    fn a_reader_that_hangs_up_mid_wait_ends_at_once() {
        let (log, rx) = Log::new(16);
        log.backfill(1000);
        // the first chunk leaves 200 rows owed at 50 a second, a 4 s wait
        let start = Start::replay(None, log.head());
        let (handle, mut out) = spawn_capped(&log, rx, start, capped(5000), |_| true);
        take(&mut out, 250);
        drop(out);
        let started = Instant::now();
        ends(handle);
        let took = started.elapsed();
        assert!(took < Duration::from_secs(2), "{took:?}");
    }

    fn chunk_size(n: usize) -> Option<NonZeroUsize> {
        NonZeroUsize::new(n)
    }

    #[test]
    fn replay_chunk_auto_uses_half_max_capacity() {
        let (tx, _rx) = mpsc::channel::<()>(500);
        assert_eq!(replay_chunk_size_for(&tx, None), Some(250));
    }

    #[test]
    fn replay_chunk_manual_is_respected() {
        let (tx, _rx) = mpsc::channel::<()>(500);
        assert_eq!(replay_chunk_size_for(&tx, chunk_size(64)), Some(64));
    }

    #[test]
    fn replay_chunk_manual_is_capped_to_max_capacity() {
        let (tx, _rx) = mpsc::channel::<()>(500);
        assert_eq!(replay_chunk_size_for(&tx, chunk_size(1000)), Some(500));
    }

    #[test]
    fn replay_chunk_is_capped_to_current_available_capacity() {
        let (tx, mut rx) = mpsc::channel::<()>(4);

        tx.try_send(()).unwrap();
        tx.try_send(()).unwrap();
        tx.try_send(()).unwrap();

        assert_eq!(replay_chunk_size_for(&tx, None), Some(1));

        let _ = rx.try_recv();
    }

    #[test]
    fn replay_chunk_none_when_channel_full() {
        let (tx, _rx) = mpsc::channel::<()>(2);

        tx.try_send(()).unwrap();
        tx.try_send(()).unwrap();

        assert_eq!(replay_chunk_size_for(&tx, None), None);
    }

    #[test]
    fn replay_chunk_manual_none_when_channel_full() {
        let (tx, _rx) = mpsc::channel::<()>(2);

        tx.try_send(()).unwrap();
        tx.try_send(()).unwrap();

        assert_eq!(replay_chunk_size_for(&tx, chunk_size(1)), None);
    }

    #[test]
    fn replay_chunk_auto_min_one_for_small_channel() {
        let (tx, _rx) = mpsc::channel::<()>(1);
        assert_eq!(replay_chunk_size_for(&tx, None), Some(1));
    }
}