//! Per-harness adapters: discover sessions on disk, parse them into the IR, and (optionally)
//! emit the IR back into a harness's native format for cross-harness porting.

use crate::ir::{Harness, Session, SessionRef};
use crate::stream::{Flow, MessageSink, ParseOptions};
use anyhow::Result;
use chrono::{DateTime, Utc};
use serde_json::Value;
use std::io::BufRead;
use std::path::{Path, PathBuf};

pub mod chatgpt_app;
pub mod claude;
pub mod claude_app;
/// Claude Code `Workflow`-tool runs (phase tree + script), a first-class layer over the
/// `subagents/workflows/` agent tier that [`claude`] walks.
pub mod claude_workflow;
pub mod cline;
pub mod codex;
pub mod continuedev;
#[cfg(feature = "sqlite")]
pub mod cursor;
pub mod export;
pub mod gemini;
#[cfg(feature = "sqlite")]
pub mod goose;
pub mod grok;
#[cfg(feature = "sqlite")]
pub mod hermes;
pub mod kimi;
pub mod lmstudio;
pub mod openclaw;
pub mod opencode;
pub mod qwen;
pub mod roo;
#[cfg(feature = "sqlite")]
pub mod zed;

/// Parse an RFC3339 timestamp into UTC — the timestamp shape nearly every harness writes. Shared
/// here so adapters don't each carry a copy.
pub(crate) fn parse_ts(s: &str) -> Option<DateTime<Utc>> {
    DateTime::parse_from_rfc3339(s).ok().map(|d| d.with_timezone(&Utc))
}

/// Flexible timestamp from a JSON value: an RFC3339 string, epoch **milliseconds**, or epoch
/// **seconds**. The ms/s split is a heuristic — magnitudes above 10^12 are milliseconds (10^12
/// seconds is the year 33658; 10^12 ms is Sep 2001, so any modern harness timestamp disambiguates).
pub(crate) fn ts_from_value(v: &Value) -> Option<DateTime<Utc>> {
    if let Some(s) = v.as_str() {
        return parse_ts(s);
    }
    let n = v.as_i64().or_else(|| v.as_f64().map(|f| f as i64))?;
    if n.abs() > 1_000_000_000_000 {
        DateTime::from_timestamp_millis(n)
    } else {
        DateTime::from_timestamp(n, 0)
    }
}

/// Drive a tolerant JSONL pass over `reader`: blank and malformed lines are skipped (never fail a
/// session over one bad record), every parsed record goes to `f`, and `Flow::Stop` ends the pass
/// early. **Not** for offset-tracking paths (the claude/codex lazy-span readers need each line's
/// byte offset and keep their own `read_until`/mmap loops).
///
/// Returns how many non-blank lines were skipped as unreadable (corrupt JSON or an undecodable
/// read) before the pass ended — adapters surface a non-zero count via
/// [`note_skipped_lines`] so silent corruption is at least visible in the session metadata.
pub(crate) fn for_each_json_line<R: BufRead>(reader: R, mut f: impl FnMut(Value) -> Flow) -> u64 {
    let mut skipped = 0u64;
    for line in reader.lines() {
        // A read error on one line (rare: invalid UTF-8 chunk) shouldn't abort the whole session.
        let Ok(line) = line else {
            skipped += 1;
            continue;
        };
        if each_json_line_inner(&line, &mut skipped, &mut f) == Flow::Stop {
            break;
        }
    }
    skipped
}

/// [`for_each_json_line`] over already-read text (no per-line allocation). Returns the skip count.
pub(crate) fn for_each_json_line_str(text: &str, mut f: impl FnMut(Value) -> Flow) -> u64 {
    let mut skipped = 0u64;
    for line in text.lines() {
        if each_json_line_inner(line, &mut skipped, &mut f) == Flow::Stop {
            break;
        }
    }
    skipped
}

fn each_json_line_inner(line: &str, skipped: &mut u64, f: &mut impl FnMut(Value) -> Flow) -> Flow {
    let line = line.trim();
    if line.is_empty() {
        return Flow::Continue;
    }
    let Ok(v) = serde_json::from_str::<Value>(line) else {
        *skipped += 1;
        return Flow::Continue; // tolerate the occasional corrupt line
    };
    f(v)
}

/// Record `skipped` unreadable transcript lines in a session's `extra` (only when non-zero, so
/// clean sessions stay byte-identical). The count is a *lower bound* when a pass stopped early.
pub(crate) fn note_skipped_lines(s: &mut Session, skipped: u64) {
    if skipped > 0 {
        s.extra.insert("skipped_lines".into(), Value::Number(skipped.into()));
    }
}

/// Result of emitting an IR session into a harness's native on-disk format.
#[derive(Debug, Clone)]
pub struct EmitResult {
    /// Primary file (or directory) written.
    pub path: PathBuf,
    /// The new native session id in the target harness.
    pub new_id: String,
    /// A human hint for how to resume it, if known (e.g. `claude --resume <id>`).
    pub resume_hint: Option<String>,
}

/// A harness adapter. Implementors live in submodules and are registered in [`all`].
///
/// `Send + Sync` so [`crate::discover_all`] can fan discovery across adapters in parallel. Every
/// adapter holds only path data (connections are opened per call), so this bound is always satisfied.
pub trait Adapter: Send + Sync {
    fn harness(&self) -> Harness;

    /// The on-disk root this adapter reads from (already resolved against $HOME), if it exists.
    fn storage_root(&self) -> Option<PathBuf>;

    /// Cheaply enumerate sessions without fully parsing them.
    fn discover(&self) -> Result<Vec<SessionRef>>;

    /// Fully parse one discovered session into the IR.
    ///
    /// Adapters with a native [`stream`](Adapter::stream) implement this as
    /// `crate::stream::collect(self, r)`; adapters that don't yet stream implement `parse`
    /// directly and inherit the default [`stream`](Adapter::stream) that bridges through it.
    fn parse(&self, r: &SessionRef) -> Result<Session>;

    /// Stream a session's messages into `sink` one at a time, dropping each before the next, so
    /// peak memory is O(largest message) rather than O(transcript). Returns the session's metadata
    /// (id/cwd/title/model/git/timestamps) with an **empty** `messages` vec — the messages went to
    /// the sink. `opts` controls how much of each message is materialized (see [`ParseOptions`]).
    ///
    /// The default bridges through [`parse`](Adapter::parse): correct for any adapter, but with no
    /// memory savings (it materializes the whole `Session` first). Override with a native streaming
    /// parse to get the savings — see [`claude`](crate::harness::claude) for the reference.
    fn stream(&self, r: &SessionRef, _opts: &ParseOptions, sink: &mut dyn MessageSink) -> Result<Session> {
        let mut session = self.parse(r)?;
        let messages = std::mem::take(&mut session.messages);
        // The full session is already parsed, so its metadata (model/cwd/title) is known up front —
        // hand it to the sink before the body so header-rendering sinks have it.
        sink.meta(&session);
        for m in messages {
            if sink.message(m) == Flow::Stop {
                break;
            }
        }
        Ok(session)
    }

    /// Emit an IR session into this harness's native format under `out_dir`.
    /// Default: unsupported. Adapters override to enable being a *conversion target*.
    fn emit(&self, _session: &Session, _out_dir: &Path) -> Result<EmitResult> {
        anyhow::bail!("emitting to {} is not implemented yet", self.harness())
    }

    /// Whether [`emit`] is implemented (so the CLI can list valid `--to` targets).
    fn can_emit(&self) -> bool {
        false
    }
}

/// All registered adapters.
pub fn all() -> Vec<Box<dyn Adapter>> {
    let mut adapters: Vec<Box<dyn Adapter>> = vec![
        Box::new(claude::Claude::new()),
        Box::new(codex::Codex::new()),
        Box::new(grok::Grok::new()),
        Box::new(opencode::OpenCode::new()),
        Box::new(gemini::Gemini::new()),
        Box::new(openclaw::OpenClaw::new()),
        Box::new(claude_app::ClaudeApp::new()),
        Box::new(chatgpt_app::ChatGptApp::new()),
        Box::new(kimi::Kimi::new()),
        Box::new(qwen::Qwen::new()),
        Box::new(lmstudio::LmStudio::new()),
        Box::new(cline::Cline::new()),
        Box::new(roo::Roo::new()),
        Box::new(continuedev::Continue::new()),
        Box::new(export::ChatGptExport::new()),
        Box::new(export::ClaudeExport::new()),
    ];
    #[cfg(feature = "sqlite")]
    {
        adapters.push(Box::new(hermes::Hermes::new()));
        adapters.push(Box::new(cursor::Cursor::new()));
        adapters.push(Box::new(goose::Goose::new()));
        adapters.push(Box::new(zed::Zed::new()));
    }
    adapters
}

/// The adapter for a specific harness, if registered.
pub fn for_harness(h: Harness) -> Option<Box<dyn Adapter>> {
    all().into_iter().find(|a| a.harness() == h)
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn ts_from_value_handles_all_shapes() {
        // RFC3339 string.
        let s = ts_from_value(&serde_json::json!("2026-01-02T03:04:05Z")).unwrap();
        assert_eq!(s.timestamp(), 1_767_323_045);
        // Epoch milliseconds (> 10^12).
        let ms = ts_from_value(&serde_json::json!(1_767_323_045_000i64)).unwrap();
        assert_eq!(ms, s);
        // Epoch seconds (<= 10^12), integer and float.
        assert_eq!(ts_from_value(&serde_json::json!(1_767_323_045i64)).unwrap(), s);
        assert_eq!(ts_from_value(&serde_json::json!(1_767_323_045.9f64)).unwrap(), s);
        // Junk.
        assert!(ts_from_value(&serde_json::json!("not a time")).is_none());
        assert!(ts_from_value(&serde_json::json!(null)).is_none());
        assert!(ts_from_value(&serde_json::json!({"t": 1})).is_none());
    }

    #[test]
    fn for_each_json_line_skips_junk_and_stops() {
        let text = "{\"n\":1}\n\n   \nnot json\n{\"n\":2}\n{\"n\":3}";
        let mut seen = Vec::new();
        let skipped = for_each_json_line_str(text, |v| {
            let n = v.get("n").and_then(Value::as_i64).unwrap_or(0);
            seen.push(n);
            if n == 2 {
                return Flow::Stop;
            }
            Flow::Continue
        });
        assert_eq!(seen, vec![1, 2], "blank/junk skipped, Stop honored");
        assert_eq!(skipped, 1, "the `not json` line counts; blank lines don't");

        // The reader-driven variant sees the same records and the same skip count.
        let mut seen2 = Vec::new();
        let skipped2 = for_each_json_line(std::io::Cursor::new(text.as_bytes()), |v| {
            seen2.push(v.get("n").and_then(Value::as_i64).unwrap_or(0));
            Flow::Continue
        });
        assert_eq!(seen2, vec![1, 2, 3]);
        assert_eq!(skipped2, 1);

        // note_skipped_lines: zero leaves `extra` untouched (clean sessions stay byte-identical).
        let mut s = Session {
            id: "x".into(),
            harness: Harness::Codex,
            cwd: None,
            title: None,
            created_at: None,
            updated_at: None,
            model: None,
            git: None,
            messages: Vec::new(),
            source_path: None,
            extra: serde_json::Map::new(),
        };
        note_skipped_lines(&mut s, 0);
        assert!(s.extra.is_empty());
        note_skipped_lines(&mut s, 3);
        assert_eq!(s.extra.get("skipped_lines").and_then(Value::as_u64), Some(3));
    }
}
