//! The stream mechanism's SQLite arm (ADR-015; ADR-019 §1/§6; ADR-021
//! §5; ADR-023 §2):
//!
//! - `stream(name)` — the validated constructor. Construction is free
//!   (validation + a closed-store check only — the table exists from
//!   bootstrap); the handle carries the connection machinery's arcs.
//! - Auto-commit publishes — an append to the durable log through the
//!   writer slot, returning the engine-assigned offset; keys are
//!   carried metadata (ADR-015 §1) with the non-empty-key rule (an
//!   empty `Some` key is `InvalidName`, matching the tx path).
//! - Cursor reads — reader-pool reads, `offset ASC` (global FIFO,
//!   ADR-015 §4), decoded into [`StreamEvent`] via the core
//!   constructor's `from_row` (the `topic` column carries the
//!   contract's `stream` field — the one name delta). The extent guard
//!   (ADR-023 §2) short-circuits `limit <= 0` to the empty `Vec` at
//!   the trait-impl entry, before the substrate — the `LIMIT -1`
//!   dialect artifact dies here.
//! - Checkpoints — explicit, monotone saves (a save below the stored
//!   checkpoint is a silent no-op, ADR-019 §6); an absent consumer
//!   reads 0.
//! - `trim_to` — the `DELETE … WHERE offset <= ?` bounded-growth op
//!   inside the writer-slot lease (ADR-015 §5); the horizon is a
//!   boundary argument, total per ADR-023 §2 (a negative horizon
//!   deletes nothing, idempotently — no guard, pass through).
//!   Surviving events keep their offsets.
//! - `subscribe` — the durable receiver: attach, replay from the
//!   stored offset to the current tail, then wake-driven re-reads off
//!   the watcher fanout (the same bridge shape as `listen`, yielding
//!   `Result<StreamEvent>` per `EventReceiver`). Watcher death closes
//!   it terminally (`recv() -> None`); recovery is a fresh
//!   `subscribe(consumer)` resuming from the saved checkpoint
//!   (ADR-021 §5).
//!
//! Subscribe specifics:
//!
//! - The watcher subscription is taken *before* any read, so no commit
//!   can fall between the attach read and the wake wait (a tick for a
//!   commit already re-read sits buffered; a drain that sees it costs
//!   an empty re-poll, never a missed event).
//! - The bridge drains pages ([`SUBSCRIBE_PAGE`]) to the tail, then
//!   parks on the wake feed. Wakes are coalesced hints (overtrigger on
//!   purpose, ADR-006): a wake may arrive for unrelated writes — the
//!   re-read from the in-memory cursor sees nothing new and parks
//!   again.
//! - A read failure while the store is open surfaces one
//!   `Err(Database)` on the receiver and waits for the next wake to
//!   re-read (transient arm); a failure after store close exits the
//!   loop (the consumer sees the close). An attach-read failure
//!   surfaces one `Err(Database)` and closes the receiver — recovery
//!   is a fresh `subscribe(consumer)`. Non-`Database` read errors
//!   (the decode-side `Codec` posture guard) are remapped into
//!   `Database` with the source chain preserved — `recv()`'s `Err` arm
//!   carries `Database` only (ADR-021 §5).
//! - The bridge pushes drained events into a bounded tokio channel
//!   and `blocking_send`s when full — a stalled consumer parks the
//!   bridge thread, and delivery resumes exactly where it stopped. The
//!   receiver's position is the last-*yielded* event's offset, so
//!   `save_offset()` never checkpoints an event the consumer did not
//!   observe.

use std::sync::Arc;
use std::sync::atomic::{AtomicBool, Ordering};

use parking_lot::Mutex;
use serde_json::Value;

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

use crate::seam::{
    ReaderLoopExit as LoopExit, blocking_read, closed_store_error, sqlite_error, with_reader,
    with_writer,
};
use crate::substrate::{
    Readers, SharedUpdateWatcher, Writer, stream_get_offset, stream_publish, stream_read_since,
    stream_save_offset,
};

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

/// The bridge's tokio-channel capacity — the smoothing buffer between
/// the bridge thread's re-read loop and the consumer (the substrate's
/// 1-slot sync feed stays the coalescing point).
const EVENT_CHANNEL_CAPACITY: usize = 16;

/// The auto-commit publish paths (see the module docs). Stream-name
/// validation fires at the entry point, before any round trip and
/// before the payload is touched (ADR-008 §4); a present key must be
/// non-empty.
fn publish_with_key(
    writer: Arc<Writer>,
    stream: &str,
    key: Option<String>,
    payload: Value,
) -> BoxedFuture<'static, Result<i64>> {
    if let Err(e) = validate_shared_name(stream) {
        return Box::pin(async move { Err(e) });
    }
    let key = match key {
        Some(k) if k.is_empty() || k.trim().is_empty() => {
            return Box::pin(async move { Err(Error::InvalidName { name: k }) });
        }
        other => other,
    };
    let stream = stream.to_string();
    Box::pin(async move {
        with_writer(writer, move |conn| {
            let bytes = alkstore::encode_payload(&payload)?;
            stream_publish(
                conn,
                &stream,
                key.as_deref(),
                std::str::from_utf8(&bytes)
                    .map_err(|e| Error::Codec(format!("row payload must be utf-8: {e}")))?,
            )
            .map_err(sqlite_error)
        })
        .await
    })
}

/// One decoded page from the reader pool: `offset > from`, ASC, up to
/// `limit` (the extent guard is the caller's — this helper is
/// contract-blind to it).
fn read_page(
    readers: Arc<Readers>,
    stream: String,
    from: i64,
    limit: i64,
) -> BoxedFuture<'static, Result<Vec<StreamEvent>>> {
    let read = move |conn: &rusqlite::Connection| {
        let page = stream_read_since(conn, &stream, from, limit).map_err(sqlite_error)?;
        stream_events_from_json_page(&page, stream.clone())
    };
    Box::pin(with_reader(readers, read))
}

/// Decode the substrate's stream-page JSON (the fork's inherited
/// return shape) into the contract's [`StreamEvent`] values via the
/// core constructor's `from_row`. Shared by the auto-commit reads
/// (this module) and the tx reads (`tx.rs`) — one decode owner. The
/// `topic` column carries the contract's `stream` field (the one name
/// delta, mapped at this layer); malformed rows map to `Error::Codec`
/// (decode-side failures is exactly what `Codec` is for — ADR-021 §2's
/// posture guard; the fork's writer always wrote well-formed rows).
pub(crate) fn stream_events_from_json_page(page: &str, stream: String) -> Result<Vec<StreamEvent>> {
    let array: serde_json::Value = serde_json::from_str(page)
        .map_err(|e| Error::Codec(format!("stream page must be json: {e}")))?;
    let rows = array
        .as_array()
        .ok_or_else(|| Error::Codec("stream page must be a json array".to_string()))?;
    let mut out = Vec::with_capacity(rows.len());
    for row in rows {
        out.push(StreamEvent::from_row(
            serde_json::from_value(row["offset"].clone())
                .map_err(|e| Error::Codec(format!("stream row offset must be i64: {e}")))?,
            stream.clone(),
            if row["key"].is_null() {
                None
            } else {
                serde_json::from_value(row["key"].clone())
                    .map_err(|e| Error::Codec(format!("stream row key must be a string: {e}")))?
            },
            row["payload"]
                .as_str()
                .ok_or_else(|| Error::Codec("stream row payload must be a string".to_string()))?
                .as_bytes()
                .to_vec(),
            serde_json::from_value(row["created_at"].clone())
                .map_err(|e| Error::Codec(format!("stream row created_at must be i64: {e}")))?,
        ));
    }
    Ok(out)
}

/// The stream-scoped handle (see the module docs).
pub struct SqliteStreamHandle {
    name: String,
    writer: Arc<Writer>,
    readers: Arc<Readers>,
    watcher: Arc<SharedUpdateWatcher>,
    closed: Arc<AtomicBool>,
}

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

impl SqliteStreamHandle {
    pub(crate) fn new(
        name: String,
        writer: Arc<Writer>,
        readers: Arc<Readers>,
        watcher: Arc<SharedUpdateWatcher>,
        closed: Arc<AtomicBool>,
    ) -> Self {
        Self {
            name,
            writer,
            readers,
            watcher,
            closed,
        }
    }
}

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

    fn publish<'a>(&'a self, payload: Value) -> BoxedFuture<'a, Result<i64>> {
        publish_with_key(self.writer.clone(), &self.name, None, payload)
    }

    fn publish_with_key<'a>(
        &'a self,
        key: Option<String>,
        payload: Value,
    ) -> BoxedFuture<'a, Result<i64>> {
        publish_with_key(self.writer.clone(), &self.name, key, payload)
    }

    fn read_since<'a>(
        &'a self,
        offset: i64,
        limit: i64,
    ) -> BoxedFuture<'a, Result<Vec<StreamEvent>>> {
        if self.closed.load(Ordering::Acquire) {
            return Box::pin(async { Err(closed_store_error()) });
        }
        if limit <= 0 {
            return Box::pin(async { Ok(Vec::new()) });
        }
        read_page(self.readers.clone(), self.name.clone(), offset, limit)
    }

    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 self.closed.load(Ordering::Acquire) {
            return Box::pin(async { Err(closed_store_error()) });
        }
        if limit <= 0 {
            return Box::pin(async { Ok(Vec::new()) });
        }
        let readers = self.readers.clone();
        let stream = self.name.clone();
        let consumer = consumer.to_string();
        let read = move |conn: &rusqlite::Connection| {
            let from = stream_get_offset(conn, &consumer, &stream).map_err(sqlite_error)?;
            let page = stream_read_since(conn, &stream, from, limit).map_err(sqlite_error)?;
            stream_events_from_json_page(&page, stream.clone())
        };
        Box::pin(with_reader(readers, read))
    }

    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) });
        }
        if self.closed.load(Ordering::Acquire) {
            return Box::pin(async { Err(closed_store_error()) });
        }
        let writer = self.writer.clone();
        let stream = self.name.clone();
        let consumer = consumer.to_string();
        Box::pin(async move {
            with_writer(writer, move |conn| {
                stream_save_offset(conn, &consumer, &stream, offset)
                    .map(|_| ())
                    .map_err(sqlite_error)
            })
            .await
        })
    }

    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) });
        }
        if self.closed.load(Ordering::Acquire) {
            return Box::pin(async { Err(closed_store_error()) });
        }
        let readers = self.readers.clone();
        let stream = self.name.clone();
        let consumer = consumer.to_string();
        let read = move |conn: &rusqlite::Connection| {
            stream_get_offset(conn, &consumer, &stream).map_err(sqlite_error)
        };
        Box::pin(with_reader(readers, read))
    }

    fn trim_to<'a>(&'a self, horizon: i64) -> BoxedFuture<'a, Result<i64>> {
        if self.closed.load(Ordering::Acquire) {
            return Box::pin(async { Err(closed_store_error()) });
        }
        let writer = self.writer.clone();
        let stream = self.name.clone();
        Box::pin(async move {
            with_writer(writer, move |conn| {
                let deleted = conn
                    .execute(
                        "DELETE FROM __alkstore_stream WHERE topic = ?1 AND offset <= ?2",
                        rusqlite::params![stream, horizon],
                    )
                    .map_err(sqlite_error)?;
                Ok(deleted as i64)
            })
            .await
        })
    }

    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) });
        }
        if self.closed.load(Ordering::Acquire) {
            return Box::pin(async { Err(closed_store_error()) });
        }
        let name = self.name.clone();
        let consumer = consumer.to_string();
        let writer = self.writer.clone();
        let readers = self.readers.clone();
        let watcher = self.watcher.clone();
        let closed = self.closed.clone();
        Box::pin(async move {
            spawn_bridge(name, consumer, writer, readers, watcher, closed)
                .map(|rx| Box::new(rx) as Box<dyn EventReceiver>)
        })
    }
}

/// The `subscribe` bridge construction: the watcher subscription is
/// taken before any read, the bridge thread then replays and parks on
/// the wake feed. A thread-spawn failure unsubscribes and fails with
/// `Database` (no panics cross the seam, the W-2 posture).
fn spawn_bridge(
    name: String,
    consumer: String,
    writer: Arc<Writer>,
    readers: Arc<Readers>,
    watcher: Arc<SharedUpdateWatcher>,
    store_closed: Arc<AtomicBool>,
) -> Result<SqliteEventReceiver> {
    let (id, wake_rx) = watcher.subscribe();
    let wake_rx = Arc::new(Mutex::new(wake_rx));
    let (event_tx, event_rx) =
        tokio::sync::mpsc::channel::<Result<StreamEvent>>(EVENT_CHANNEL_CAPACITY);
    let thread_name = name.clone();
    let thread_consumer = consumer.clone();
    let thread_readers = readers.clone();
    let thread_closed = store_closed.clone();
    let thread_tx = event_tx.clone();
    let spawned = std::thread::Builder::new()
        .name(format!("alkstore-stream-bridge-{thread_name}"))
        .spawn(move || {
            run_subscribe_loop(
                thread_name,
                thread_consumer,
                thread_readers,
                thread_closed,
                wake_rx,
                thread_tx,
            );
        });
    if let Err(e) = spawned {
        watcher.unsubscribe(id);
        return Err(Error::database(std::io::Error::other(format!(
            "failed to spawn stream bridge thread: {e}"
        ))));
    }
    Ok(SqliteEventReceiver {
        name,
        consumer,
        rx: event_rx,
        watcher,
        id,
        position: 0,
        writer,
        readers,
        store_closed,
    })
}

/// The bridge loop body (runs on its own thread — this file's honest
/// sync shape, mirroring `listen`'s bridge thread):
///
/// 1. read the consumer's stored offset (the attach read),
/// 2. drain pages until the tail (`offset ASC`),
/// 3. park on the wake feed (any commit ticks it), re-drain on wake,
/// 4. a read failure while the store is open surfaces one
///    `Err(Database)` and waits for the next wake; a read failure
///    after store close exits (the death path — watcher death via the
///    cleared feed, or store close, disconnects the wake feed and ends
///    the loop; the receiver sees the channel close).
fn run_subscribe_loop(
    name: String,
    consumer: String,
    readers: Arc<Readers>,
    store_closed: Arc<AtomicBool>,
    wake_rx: Arc<Mutex<std::sync::mpsc::Receiver<()>>>,
    event_tx: tokio::sync::mpsc::Sender<Result<StreamEvent>>,
) {
    let mut cursor = match blocking_read(&readers, &store_closed, |conn| {
        stream_get_offset(conn, &consumer, &name).map_err(sqlite_error)
    }) {
        Ok(offset) => offset,
        Err(LoopExit::Closed) => return,
        Err(LoopExit::Failed(e)) => {
            let _ = event_tx.blocking_send(Err(database_only(e)));
            return;
        }
    };

    loop {
        loop {
            match blocking_read(&readers, &store_closed, |conn| {
                let page =
                    stream_read_since(conn, &name, cursor, SUBSCRIBE_PAGE).map_err(sqlite_error)?;
                stream_events_from_json_page(&page, name.clone())
            }) {
                Err(LoopExit::Closed) => return,
                Err(LoopExit::Failed(e)) => {
                    if event_tx.blocking_send(Err(database_only(e))).is_err() {
                        return;
                    }
                    break;
                }
                Ok(events) if events.is_empty() => break,
                Ok(events) => {
                    for event in events {
                        cursor = event.offset;
                        if event_tx.blocking_send(Ok(event)).is_err() {
                            return;
                        }
                    }
                }
            }
        }
        match wake_rx.lock().recv() {
            Ok(()) => {}
            // Watcher death: the death-guard cleared the feed. Close.
            Err(_) => return,
        }
    }
}

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

/// The durable subscription receiver: the tokio side of the bridge,
/// owning the consumer's position state. Its [`Drop`] unsubscribes the
/// substrate feed eagerly (before the watcher's next fanout would) —
/// the bridge thread exits on the failed send or the disconnected wake
/// feed, and the consumer sees the channel close (`None`, terminal).
///
/// `position` is the last-yielded event's offset (0 before any yield —
/// `save_offset()` is then a no-op, the stored checkpoint untouched,
/// ADR-021 §5; ADR-019 §6).
struct SqliteEventReceiver {
    name: String,
    consumer: String,
    rx: tokio::sync::mpsc::Receiver<Result<StreamEvent>>,
    watcher: Arc<SharedUpdateWatcher>,
    id: u64,
    position: i64,
    writer: Arc<Writer>,
    readers: Arc<Readers>,
    store_closed: Arc<AtomicBool>,
}

impl Drop for SqliteEventReceiver {
    fn drop(&mut self) {
        self.watcher.unsubscribe(self.id);
    }
}

impl EventReceiver for SqliteEventReceiver {
    fn recv<'a>(&'a mut self) -> BoxedFuture<'a, Option<Result<StreamEvent>>> {
        Box::pin(async move {
            match self.rx.recv().await {
                Some(Ok(event)) => {
                    self.position = event.offset;
                    Some(Ok(event))
                }
                // A transient read failure — the bridge keeps running
                // and re-reads on the next wake.
                Some(Err(e)) => Some(Err(e)),
                // The bridge exited: watcher death or store close —
                // `None` = closed, terminal (it never reopens).
                None => None,
            }
        })
    }

    fn try_recv(&mut self) -> Result<Option<StreamEvent>> {
        match self.rx.try_recv() {
            Ok(Ok(event)) => {
                self.position = event.offset;
                Ok(Some(event))
            }
            Ok(Err(e)) => Err(e),
            Err(tokio::sync::mpsc::error::TryRecvError::Empty) => Ok(None),
            Err(tokio::sync::mpsc::error::TryRecvError::Disconnected) => Err(Error::Closed),
        }
    }

    fn read_since<'a>(
        &'a mut self,
        offset: i64,
        limit: i64,
    ) -> BoxedFuture<'a, Result<Vec<StreamEvent>>> {
        if self.store_closed.load(Ordering::Acquire) {
            return Box::pin(async { Err(closed_store_error()) });
        }
        if limit <= 0 {
            return Box::pin(async { Ok(Vec::new()) });
        }
        read_page(self.readers.clone(), self.name.clone(), offset, limit)
    }

    fn save_offset(&mut self) -> Result<()> {
        let position = self.position;
        if position == 0 {
            return Ok(());
        }
        // The save runs through the same monotone op the direct form
        // uses — the forms interleave and cannot regress (ADR-019 §6).
        // The trait's signature is sync, so the storage round trip
        // takes the writer slot on this thread (the slot is nearly
        // always free between ops; a long open transaction parks this
        // call — the honest lease posture surfaced, ADR-007).
        if self.store_closed.load(Ordering::Acquire) {
            return Err(closed_store_error());
        }
        let conn = match self.writer.acquire() {
            Some(conn) => conn,
            None => return Err(closed_store_error()),
        };
        let out = stream_save_offset(&conn, &self.consumer, &self.name, position)
            .map(|_| ())
            .map_err(sqlite_error);
        self.writer.release(conn);
        out
    }

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