smith

//! SMH-SPEC-SPEC0001: Sessions, bounded storage requests.
//!
//! Bounded storage requests: filesystem reads run on a dedicated worker.
//!
//! Callers own immutable results, never an open reader or ambient I/O in a
//! core state transition.
use smith::config::Config;
use smith::error::{Result, SmithError};
use smith::id::EntryId;
use smith_alloc::{Scope, enter};
use smith_core::session::{EntryFrame, Session};
use smith_core::store::{SessionReader, SessionWriter, open_session};
use std::path::PathBuf;
use std::sync::mpsc::{self, SyncSender};

/// Serialized operations on one session file.
pub struct StorageWorker {
    tx: SyncSender<Request>,
    thread: Option<std::thread::JoinHandle<()>>,
}

enum Request {
    Open(Config, SyncSender<Result<Session>>),
    Append(EntryFrame, SyncSender<Result<()>>),
    Selected(usize, SyncSender<Result<Session>>),
    Verify(usize, SyncSender<Result<()>>),
    Fork(EntryId, PathBuf, Config, SyncSender<Result<Session>>),
    Window(
        EntryId,
        usize,
        usize,
        usize,
        SyncSender<Result<Vec<EntryFrame>>>,
    ),
}

fn read_window(
    path: &std::path::Path,
    id: EntryId,
    start: usize,
    limit: usize,
    max: usize,
) -> Result<Vec<EntryFrame>> {
    SessionReader::open(path).and_then(|reader| {
        let mut result = Vec::with_capacity(limit);
        let mut position = 0;
        reader.replay_at(id, max, |entry| {
            if position >= start && result.len() < limit {
                result.push(entry);
            }
            position += 1;
            Ok(())
        })?;
        Ok(result)
    })
}

fn stopped() -> SmithError {
    SmithError::Session {
        code: smith::error::SessionCode::StorageWorkerStopped,
        message: "session storage worker stopped".into(),
    }
}

impl StorageWorker {
    /// Start a dedicated filesystem worker for `path`.
    #[must_use]
    pub fn start(path: PathBuf) -> Self {
        let (tx, rx) = mpsc::sync_channel(1);
        let thread = std::thread::spawn(move || {
            let _session = enter(Scope::Session);
            let mut writer: Option<SessionWriter> = None;
            while let Ok(request) = rx.recv() {
                let _request = enter(Scope::IoRequest);
                match request {
                    Request::Open(config, reply) => {
                        let result = open_session(&path, config).map(|opened| {
                            writer = Some(opened.writer);
                            opened.session
                        });
                        // The requesting client may have stopped waiting.
                        let _reply_result: std::result::Result<
                            (),
                            std::sync::mpsc::SendError<Result<Session>>,
                        > = reply.send(result);
                    }
                    Request::Append(entry, reply) => {
                        let result = writer.as_mut().ok_or_else(stopped).and_then(|writer| {
                            let record = smith_core::session::Entry {
                                id: entry.id,
                                parent: entry.parent,
                                timestamp_ms: entry.timestamp_ms,
                                kind: entry.kind,
                                content: smith_core::session::Span { offset: 0, len: 0 },
                            };
                            writer.append_entry(&record, &entry.content)
                        });
                        // The requesting client may have stopped waiting.
                        let _reply_result: std::result::Result<
                            (),
                            std::sync::mpsc::SendError<Result<()>>,
                        > = reply.send(result);
                    }
                    Request::Selected(max, reply) => {
                        // The requesting client may have stopped waiting.
                        let _reply_result: std::result::Result<
                            (),
                            std::sync::mpsc::SendError<Result<Session>>,
                        > = reply.send(
                            SessionReader::open(&path)
                                .and_then(|reader| reader.selected_session(max)),
                        );
                    }
                    Request::Verify(max, reply) => {
                        // The requesting client may have stopped waiting.
                        let _reply_result: std::result::Result<
                            (),
                            std::sync::mpsc::SendError<Result<()>>,
                        > = reply.send(
                            SessionReader::open(&path).and_then(|reader| reader.verify_full(max)),
                        );
                    }
                    Request::Fork(id, target, config, reply) => {
                        // The requesting client may have stopped waiting.
                        let _reply_result: std::result::Result<
                            (),
                            std::sync::mpsc::SendError<Result<Session>>,
                        > = reply.send(
                            SessionReader::open(&path)
                                .and_then(|reader| reader.fork_at(id, &target, config))
                                .map(|opened| opened.session),
                        );
                    }
                    Request::Window(id, start, limit, max, reply) => {
                        // The requesting client may have stopped waiting.
                        let _reply_result: std::result::Result<
                            (),
                            std::sync::mpsc::SendError<Result<Vec<EntryFrame>>>,
                        > = reply.send(read_window(&path, id, start, limit, max));
                    }
                }
            }
        });
        Self {
            tx,
            thread: Some(thread),
        }
    }

    fn request<T>(&self, build: impl FnOnce(SyncSender<Result<T>>) -> Request) -> Result<T> {
        let (tx, rx) = mpsc::sync_channel(1);
        self.tx.send(build(tx)).map_err(|_| stopped())?;
        rx.recv().map_err(|_| stopped())?
    }

    /// Open or resume a writable session; keep writer ownership on worker.
    ///
    /// # Errors
    /// Returns a storage or worker fault.
    pub fn open(&self, config: Config) -> Result<Session> {
        self.request(|reply| Request::Open(config, reply))
    }

    /// Persist an entry before it becomes visible in the in-memory session.
    ///
    /// # Errors
    /// Returns a storage or worker fault.
    pub fn append(&self, entry: EntryFrame) -> Result<()> {
        self.request(|reply| Request::Append(entry, reply))
    }

    /// Refresh the selected-context snapshot without retaining other branches.
    ///
    /// # Errors
    /// Returns a storage or worker fault.
    pub fn selected(&self, max: usize) -> Result<Session> {
        self.request(|reply| Request::Selected(max, reply))
    }

    /// Verify the full file, not just the selected context.
    ///
    /// # Errors
    /// Returns an integrity or worker fault.
    pub fn verify(&self, max: usize) -> Result<()> {
        self.request(|reply| Request::Verify(max, reply))
    }

    /// Fork a requested historical tip without reading on the caller thread.
    ///
    /// # Errors
    /// Returns a storage or worker fault.
    pub fn fork(&self, id: EntryId, target: PathBuf, config: Config) -> Result<Session> {
        self.request(|reply| Request::Fork(id, target, config, reply))
    }

    /// Own a selected branch's bounded visible window, without retaining
    /// preceding or subsequent payloads. At most 64 entries per request.
    ///
    /// # Errors
    /// Returns a limit, ancestry, storage, or worker fault.
    pub fn window(
        &self,
        id: EntryId,
        start: usize,
        limit: usize,
        max: usize,
    ) -> Result<Vec<EntryFrame>> {
        if limit == 0 || limit > 64 {
            return Err(SmithError::Session {
                code: smith::error::SessionCode::StorageWindowLimit,
                message: "window limit must be 1..=64".into(),
            });
        }
        self.request(|reply| Request::Window(id, start, limit, max, reply))
    }
}

impl Drop for StorageWorker {
    fn drop(&mut self) {
        let (replacement, _rx) = mpsc::sync_channel(1);
        drop(std::mem::replace(&mut self.tx, replacement));
        if let Some(thread) = self.thread.take() {
            // Drop cannot report a worker panic; closed requests report worker failure.
            let _join_result: std::thread::Result<()> = thread.join();
        }
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use smith::message::{Message, Role};
    use smith_core::session::{EntryContent, EntryFrame};

    fn history(path: &std::path::Path, entries: usize) -> EntryId {
        use smith_core::frame::Frame;
        use smith_core::store::SessionWriter;
        let session = Session::new();
        let mut writer = SessionWriter::open(path, Config::default_valid()).unwrap();
        writer
            .append_frame(&Frame::Header {
                header: session.header(),
            })
            .unwrap();
        let mut tip = None;
        for index in 0..entries {
            let entry = EntryFrame::new(
                tip,
                &EntryContent::Message(Message::with_text(
                    Role::User,
                    format!("{index} {}", "x".repeat(32 * 1024)),
                )),
            )
            .unwrap();
            tip = Some(entry.id);
            writer.append_frame(&Frame::Known { entry }).unwrap();
        }
        writer.select_entry(tip.unwrap()).unwrap();
        tip.unwrap()
    }

    #[test]
    fn worker_resumes_selected_context_and_reads_bounded_windows() {
        // SMH-SPEC-SPEC0001: Sessions, bounded working sets and replay.
        let dir = std::env::temp_dir().join(format!("smith-storage-worker-{}", std::process::id()));
        let _ = std::fs::remove_dir_all(&dir);
        std::fs::create_dir_all(&dir).unwrap();
        let worker = StorageWorker::start(dir.join("session.smh"));
        let cfg = Config::default_valid();
        let session = worker.open(cfg.clone()).unwrap();
        assert!(session.selected_id().is_none());
        // Writer lock remains on the worker after open, including idle periods.
        assert_eq!(
            smith_core::store::SessionWriter::open(dir.join("session.smh"), cfg.clone())
                .unwrap_err()
                .code(),
            "STORE_BUSY"
        );
        let root = EntryFrame::new(
            None,
            &EntryContent::Message(Message::with_text(Role::User, "root")),
        )
        .unwrap();
        let tip = EntryFrame::new(
            Some(root.id),
            &EntryContent::Message(Message::with_text(Role::Assistant, "tip")),
        )
        .unwrap();
        for entry in [&root, &tip] {
            worker.append(entry.clone()).unwrap();
        }
        assert_eq!(
            worker.selected(cfg.frame_bytes_max).unwrap().selected_id(),
            Some(tip.id)
        );
        worker.verify(cfg.frame_bytes_max).unwrap();
        assert_eq!(
            worker
                .window(tip.id, 0, 2, cfg.frame_bytes_max)
                .unwrap()
                .into_iter()
                .map(|entry| entry.id)
                .collect::<Vec<_>>(),
            [root.id, tip.id]
        );
        assert_eq!(
            worker.window(tip.id, 1, 1, cfg.frame_bytes_max).unwrap()[0].id,
            tip.id
        );
        assert_eq!(
            worker
                .window(tip.id, 0, 65, cfg.frame_bytes_max)
                .unwrap_err()
                .code(),
            "STORAGE_WINDOW_LIMIT"
        );
        drop(worker);
        let _ = std::fs::remove_dir_all(dir);
    }

    #[test]
    fn worker_window_peak_does_not_grow_with_history() {
        let root =
            std::env::temp_dir().join(format!("smith-storage-bounds-{}", std::process::id()));
        let _ = std::fs::remove_dir_all(&root);
        std::fs::create_dir_all(&root).unwrap();
        let small_path = root.join("small.smh");
        let large_path = root.join("large.smh");
        let small_tip = history(&small_path, 4);
        let large_tip = history(&large_path, 128);
        let small_worker = StorageWorker::start(small_path);
        let large_worker = StorageWorker::start(large_path);
        let before = smith_alloc::snapshot(smith_alloc::Scope::IoRequest);
        let small_rows = small_worker
            .window(small_tip, 3, 1, Config::default_valid().frame_bytes_max)
            .unwrap();
        let after_small = smith_alloc::snapshot(smith_alloc::Scope::IoRequest);
        let large_rows = large_worker
            .window(large_tip, 127, 1, Config::default_valid().frame_bytes_max)
            .unwrap();
        let after_large = smith_alloc::snapshot(smith_alloc::Scope::IoRequest);
        assert_eq!((small_rows.len(), large_rows.len()), (1, 1));
        assert!(
            after_small.allocations > before.allocations
                && after_large.allocations > after_small.allocations,
            "production worker requests must charge allocations to IoRequest"
        );
        let retained = after_large.live;
        drop((small_rows, large_rows, small_worker, large_worker));
        let released = smith_alloc::snapshot(smith_alloc::Scope::IoRequest).live;
        assert!(
            retained != released,
            "freeing rows outside worker scope must release their original IoRequest charge"
        );
        let _ = std::fs::remove_dir_all(root);
    }
}