//! Read-only, bounded per-record inspection (`SMH-SPEC-SPEC0001`, Session inspection).
use smith::config::Config;
use smith::error::{Result, SmithError};
use smith::id::SessionId;
use smith_core::content_codec::EntryContentView;
use smith_core::frame::Frame;
use smith_core::session::EntryFrame;
use smith_core::store::SessionReader;
use std::fs::read_dir;
use std::io::Write;
use std::path::{Path, PathBuf};

/// Session files one incremental inspection may visit.
const SESSION_LIST_ENTRIES_MAX: usize = 1024;

fn invalid(path: &Path, reason: impl std::fmt::Display) -> SmithError {
    SmithError::Session {
        code: smith::error::SessionCode::SessionCorrupt,
        message: format!("{}: {reason}", path.display()),
    }
}

/// Summary of one named session without decoding its history.
#[derive(Debug)]
pub struct Summary {
    /// Identity from the validated file prologue.
    pub id: SessionId,
    /// On-disk file size observed during discovery.
    pub bytes: u64,
}

/// Compare a filename stem against a session identity without allocating:
/// the stem is canonical proquint text exactly when it decodes to the id.
fn stem_is(stem: Option<&std::ffi::OsStr>, id: SessionId) -> bool {
    stem.and_then(std::ffi::OsStr::to_str)
        .and_then(|text| smith::id::proquint::decode(text).ok())
        .is_some_and(|bits| bits == id.as_u128())
}

/// Visit sessions in ID filename order; only each header is decoded.
///
/// This repeatedly enumerates `dir` to select the next path after the prior
/// result. It uses O(1) retained paths and summaries, at O(n²) directory-entry
/// comparisons for n sessions, rather than materializing an index or result set.
///
/// # Errors
/// Fails explicitly if a session path has an invalid prologue or identity.
pub fn list(dir: &Path, mut visit: impl FnMut(Summary) -> Result<()>) -> Result<()> {
    let mut after = None;
    let mut visited = 0;
    while let Some(path) = next_session_path(dir, after.as_deref())? {
        if visited == SESSION_LIST_ENTRIES_MAX {
            return Err(SmithError::Session {
                code: smith::error::SessionCode::SessionListLimit,
                message: format!(
                    "{}: session listing exceeds {SESSION_LIST_ENTRIES_MAX} files; inspect sessions individually with `smith sessions get <id>`",
                    dir.display()
                ),
            });
        }
        // The previous path only bounds the search; release it before the
        // scan so one path stays live per visited session.
        drop(after.take());
        let reader = SessionReader::open(&path)?;
        let bytes = reader.file_size();
        let mut id = None;
        let scan = reader.scan_reporting(
            Config::default_valid().frame_bytes_max,
            |frame| match frame {
                Frame::Header { header } => {
                    id = Some(header.session_id);
                    Ok(false)
                }
                _ => Err(invalid(&path, "missing session header")),
            },
            |damage| Err(invalid(&path, damage)),
        )?;
        if scan.torn_tail_at.is_some() {
            return Err(invalid(&path, "invalid header frame"));
        }
        let Some(id) = id else {
            return Err(invalid(&path, "missing session header"));
        };
        if !stem_is(path.file_stem(), id) {
            return Err(invalid(&path, "filename does not match session identity"));
        }
        visit(Summary { id, bytes })?;
        visited += 1;
        after = Some(path);
    }
    Ok(())
}

fn next_session_path(dir: &Path, after: Option<&Path>) -> Result<Option<PathBuf>> {
    let entries = match read_dir(dir) {
        Ok(entries) => entries,
        Err(error) if error.kind() == std::io::ErrorKind::NotFound => return Ok(None),
        Err(error) => return Err(invalid(dir, error)),
    };
    let mut next = None;
    for entry in entries {
        let path = entry.map_err(|error| invalid(dir, error))?.path();
        if path.extension().is_none_or(|extension| extension != "smh")
            || after.is_some_and(|previous| path <= previous)
        {
            continue;
        }
        if next
            .as_ref()
            .is_none_or(|candidate: &PathBuf| path < *candidate)
        {
            next = Some(path);
        }
    }
    Ok(next)
}

/// Scan all complete entries in physical order within one file-size snapshot.
/// Callback receives one decoded entry at a time; no history is retained.
///
/// # Errors
/// Reports damaged frames, incomplete tails and invalid content explicitly.
pub fn entries(
    path: &Path,
    mut visit: impl FnMut(&EntryFrame, &EntryContentView<'_>) -> Result<()>,
) -> Result<()> {
    let reader = SessionReader::open(path)?;
    let mut header = false;
    let mut count = 0;
    let scan = reader.scan_reporting(
        Config::default_valid().frame_bytes_max,
        |frame| {
            match frame {
                Frame::Header { header: found } if count == 0 => {
                    header = true;
                    if !stem_is(path.file_stem(), found.session_id) {
                        return Err(invalid(path, "filename does not match session identity"));
                    }
                }
                Frame::Known { entry } if header => {
                    let content =
                        EntryContentView::decode(&entry.content).map_err(|e| invalid(path, e))?;
                    if content.kind() != entry.kind {
                        return Err(invalid(path, format!("entry {} kind mismatch", entry.id)));
                    }
                    visit(&entry, &content)?;
                }
                Frame::Unknown { .. }
                | Frame::Index { .. }
                | Frame::Selection { .. }
                | Frame::Checkpoint { .. }
                    if header => {}
                _ => return Err(invalid(path, "invalid or duplicate session header")),
            }
            count += 1;
            Ok(true)
        },
        |damage| Err(invalid(path, damage)),
    )?;
    if !header {
        return Err(invalid(path, "missing session header"));
    }
    if scan.torn_tail_at.is_some() && !reader.writer_active() {
        return Err(invalid(path, "incomplete tail"));
    }
    Ok(())
}

/// Stream one inspected record to the output boundary without building an entry JSON tree.
///
/// # Errors
/// Returns an I/O error for a broken pipe or failed serialization.
pub fn output(
    out: &mut impl Write,
    id: SessionId,
    entry: &EntryFrame,
    content: &EntryContentView<'_>,
    json: bool,
    get: bool,
) -> std::io::Result<()> {
    if json {
        write!(out, "{{\"session\":\"{id}\",\"entry\":\"{}\",", entry.id)?;
        if get {
            match entry.parent {
                Some(parent) => write!(out, "\"parent\":\"{parent}\",")?,
                None => out.write_all(b"\"parent\":null,")?,
            }
            write!(out, "\"timestamp_ms\":{},", entry.timestamp_ms)?;
        }
        out.write_all(b"\"content\":")?;
    } else if get {
        write!(out, "{id} {} {} ", entry.id, entry.timestamp_ms)?;
    } else {
        write!(out, "{id}:{}: ", entry.id)?;
    }
    serde_json::to_writer(&mut *out, content).map_err(std::io::Error::other)?;
    if json {
        out.write_all(b"}")?;
    }
    out.write_all(b"\n")
}

/// Search literal UTF-8 substring among recorded string values, not JSON field names.
#[must_use]
pub fn contains(content: &EntryContentView<'_>, pattern: &str) -> bool {
    content.contains(pattern)
}

#[cfg(test)]
mod tests {
    use super::*;
    use smith_alloc::{Counting, Scope, measure};
    use smith_core::session::{EntryContent, EntryFrame, Session};
    use smith_core::store::SessionWriter;
    use std::io::Write;

    #[global_allocator]
    static ALLOC: Counting = Counting;

    const FILES: usize = 4;
    const SMALL_ENTRIES: usize = 4;
    const LARGE_ENTRIES: usize = 128;
    const PAYLOAD_BYTES: usize = 32 * 1024;
    // Retain the original ceilings, expressed with their 164-byte reference
    // path capacity. Account only for the required live paths (one for list,
    // two for grep), whose capacity is determined outside the measured code.
    // Longer TMPDIR paths cannot consume the fixed inspection budget, and
    // extra path copies or payload allocations still exceed it (SMH-ISSUE-MUTTMPDR).
    const REFERENCE_PATH_BYTES: u64 = 164;
    const LIST_PEAK_BYTES: u64 = 65_776;
    const GREP_PEAK_BYTES: u64 = 131_676;
    const PER_ENTRY_VARIANCE_BYTES: u64 = 16;

    struct BoundedSink;

    impl Write for BoundedSink {
        fn write(&mut self, bytes: &[u8]) -> std::io::Result<usize> {
            Ok(bytes.len())
        }

        fn flush(&mut self) -> std::io::Result<()> {
            Ok(())
        }
    }

    fn fixture(dir: &Path, sessions: usize, entries: usize, tag: &str) {
        std::fs::create_dir_all(dir).unwrap();
        for file in 0..sessions {
            let session = Session::new();
            let path = dir.join(format!("{}.smh", session.id()));
            let mut writer = SessionWriter::open(&path, Config::default_valid()).unwrap();
            writer
                .append_frame(&Frame::Header {
                    header: session.header(),
                })
                .unwrap();
            for index in 0..entries {
                let content = if index % 2 == 0 {
                    EntryContent::Message(smith::message::Message::with_text(
                        smith::message::Role::User,
                        format!("needle {tag} {file} {index} {}", "x".repeat(PAYLOAD_BYTES)),
                    ))
                } else {
                    EntryContent::Meta {
                        kind: "event".into(),
                        detail: smith::StructuredData::from_serializable(&serde_json::json!({
                            "needle": tag,
                            "index": index,
                            "payload": "x".repeat(PAYLOAD_BYTES),
                            "nested": [{"kind": "structured", "values": [true, 0, null]}],
                        }))
                        .unwrap(),
                    }
                };
                writer
                    .append_frame(&Frame::Known {
                        entry: EntryFrame::new(None, &content).unwrap(),
                    })
                    .unwrap();
            }
        }
    }

    fn list_cost(dir: &Path) -> smith_alloc::Stats {
        let (_, cost) = measure(Scope::IoRequest, || {
            let mut seen = 0;
            list(dir, |_| {
                seen += 1;
                Ok(())
            })
            .unwrap();
            assert_eq!(seen, FILES);
        });
        cost
    }

    fn grep_cost(dir: &Path, entries_per_file: usize) -> smith_alloc::Stats {
        let (_, cost) = measure(Scope::IoRequest, || {
            let mut sink = BoundedSink;
            let mut matches = 0;
            list(dir, |summary| {
                entries(
                    &dir.join(format!("{}.smh", summary.id)),
                    |entry, content| {
                        if contains(content, "needle") {
                            matches += 1;
                            output(&mut sink, summary.id, entry, content, true, false).unwrap();
                        }
                        Ok(())
                    },
                )
            })
            .unwrap();
            assert_eq!(matches, FILES * (entries_per_file / 2));
        });
        cost
    }

    #[test]
    fn stem_comparison_decodes_without_allocating() {
        // SMH-SPEC-SPEC0001: filename identity is checked by value, not text.
        let id = SessionId::from_u128(0x5E55);
        let matching = PathBuf::from(format!("{id}.smh"));
        let mismatched = PathBuf::from(format!("{}.smh", SessionId::from_u128(0x5E56)));
        let unparsable = PathBuf::from("not-an-id.smh");
        let uppercase = PathBuf::from(format!("{}.smh", id.to_string().to_uppercase()));
        let rootless = PathBuf::from("/");
        let (verdicts, cost) = measure(Scope::IoRequest, || {
            [&matching, &mismatched, &unparsable, &uppercase, &rootless]
                .map(|path| stem_is(path.file_stem(), id))
        });
        assert_eq!(verdicts, [true, false, false, false, false]);
        assert_eq!(cost.allocations, 0, "stem comparison allocated");
    }

    #[test]
    fn list_visits_stable_order_incrementally() {
        let dir = std::env::temp_dir().join(format!("smith-inspect-list-{}", std::process::id()));
        let _ = std::fs::remove_dir_all(&dir);
        std::fs::create_dir_all(&dir).unwrap();
        let mut expected = Vec::new();
        for _ in 0..3 {
            let session = Session::new();
            let path = dir.join(format!("{}.smh", session.id()));
            let mut writer = SessionWriter::open(&path, Config::default_valid()).unwrap();
            writer
                .append_frame(&Frame::Header {
                    header: session.header(),
                })
                .unwrap();
            expected.push(session.id());
        }
        expected.sort();
        let mut visited = Vec::new();
        list(&dir, |summary| {
            visited.push(summary.id);
            Ok(())
        })
        .unwrap();
        assert_eq!(visited, expected);
        let _ = std::fs::remove_dir_all(dir);
    }

    #[test]
    fn listing_stops_at_the_named_session_file_limit() {
        let dir = std::env::temp_dir().join(format!("smith-inspect-limit-{}", std::process::id()));
        let _ = std::fs::remove_dir_all(&dir);
        std::fs::create_dir_all(&dir).unwrap();
        for _ in 0..=SESSION_LIST_ENTRIES_MAX {
            let session = Session::new();
            let path = dir.join(format!("{}.smh", session.id()));
            let mut writer = SessionWriter::open(&path, Config::default_valid()).unwrap();
            writer
                .append_frame(&Frame::Header {
                    header: session.header(),
                })
                .unwrap();
        }
        let error = list(&dir, |_| Ok(())).unwrap_err();
        assert_eq!(error.code(), "SESSION_LIST_LIMIT");
        assert!(error.to_string().contains("inspect sessions individually"));
        let _ = std::fs::remove_dir_all(dir);
    }

    #[test]
    fn inspection_large_history_peak_is_independent_of_payload_bytes() {
        // SMH-SPEC-SPEC0001: list and grep stream bounded records, including
        // JSON-like structured content, rather than retaining history or output.
        let root = std::env::temp_dir().join(format!("smith-inspect-{}", std::process::id()));
        for prefix_bytes in [0, 64, 128] {
            let temporary = root.join("p".repeat(prefix_bytes));
            std::fs::create_dir_all(&temporary).unwrap();
            let temporary = std::fs::canonicalize(&temporary).unwrap();
            let small = temporary.join("small");
            let large = temporary.join("large");
            let path_bytes = u64::try_from(
                small
                    .join(format!("{}.smh", SessionId::from_u128(0)))
                    .capacity(),
            )
            .unwrap();
            fixture(&small, FILES, SMALL_ENTRIES, "history");
            fixture(&large, FILES, LARGE_ENTRIES, "history");
            let listed_small = list_cost(&small);
            let listed_large = list_cost(&large);
            assert_eq!(
                listed_small.peak, listed_large.peak,
                "listing peak is per file, not history payload bytes"
            );
            let listed_peak =
                listed_large.peak.checked_sub(path_bytes).unwrap() + REFERENCE_PATH_BYTES;
            assert!(
                listed_peak <= LIST_PEAK_BYTES,
                "path-normalized listing peak {listed_peak} bytes exceeds ceiling {LIST_PEAK_BYTES} bytes"
            );

            let grepped_small = grep_cost(&small, SMALL_ENTRIES);
            let grepped_large = grep_cost(&large, LARGE_ENTRIES);
            assert!(
                grepped_large.peak <= grepped_small.peak + PER_ENTRY_VARIANCE_BYTES,
                "grep peak grew from {} to {} bytes",
                grepped_small.peak,
                grepped_large.peak
            );
            let grepped_peak =
                grepped_large.peak.checked_sub(2 * path_bytes).unwrap() + 2 * REFERENCE_PATH_BYTES;
            assert!(
                grepped_peak <= GREP_PEAK_BYTES,
                "path-normalized grep peak {grepped_peak} bytes exceeds ceiling {GREP_PEAK_BYTES} bytes"
            );
            assert!(
                grepped_large.peak < 256 * 1024,
                "grep peak {} bytes for {} bytes of session payload",
                grepped_large.peak,
                std::fs::read_dir(&large)
                    .unwrap()
                    .filter_map(std::result::Result::ok)
                    .filter_map(|entry| entry.metadata().ok())
                    .map(|metadata| metadata.len())
                    .sum::<u64>()
            );
        }
        std::fs::remove_dir_all(root).unwrap();
    }

    #[test]
    fn live_incomplete_tail_is_pending_until_writer_exits() {
        // SMH-SPEC-SPEC0001: Sessions, publication and observer snapshots.
        let dir = std::env::temp_dir().join(format!("smith-inspect-tail-{}", std::process::id()));
        std::fs::create_dir_all(&dir).unwrap();
        let session = Session::new();
        let path = dir.join(format!("{}.smh", session.id()));
        let mut writer = SessionWriter::open(&path, Config::default_valid()).unwrap();
        writer
            .append_frame(&Frame::Header {
                header: session.header(),
            })
            .unwrap();
        let mut tail = std::fs::OpenOptions::new()
            .append(true)
            .open(&path)
            .unwrap();
        tail.write_all(&[4, 0]).unwrap();
        assert!(entries(&path, |_, _| Ok(())).is_ok());
        drop(writer);
        assert_eq!(
            entries(&path, |_, _| Ok(())).unwrap_err().code(),
            "SESSION_CORRUPT"
        );
        let _ = std::fs::remove_dir_all(dir);
    }
}