//! Persist LHA session rollouts (.jsonl) so sessions can be replayed or inspected later.

use std::fs::File;
use std::fs::{self};
use std::io::Error as IoError;
use std::io::ErrorKind;
use std::path::Path;
use std::path::PathBuf;

use crate::product::protocol::ThreadId;
use crate::product::protocol::dynamic_tools::DynamicToolSpec;
use crate::product::protocol::models::BaseInstructions;
use serde_json::Value;
use time::OffsetDateTime;
use time::format_description::FormatItem;
use time::macros::format_description;
use tokio::io::AsyncWriteExt;
use tokio::sync::mpsc::Sender;
use tokio::sync::mpsc::{self};
use tokio::sync::oneshot;
use tracing::info;
use tracing::warn;

use super::ARCHIVED_SESSIONS_SUBDIR;
use super::SESSIONS_SUBDIR;
use super::list::Cursor;
use super::list::ThreadListConfig;
use super::list::ThreadListLayout;
use super::list::ThreadSortKey;
use super::list::ThreadsPage;
use super::list::get_threads;
use super::list::get_threads_in_root;
use super::metadata;
use super::policy::is_persisted_response_item;
use crate::product::agent::config::Config;
use crate::product::agent::default_client::originator;
use crate::product::agent::git_info::collect_git_info;
use crate::product::agent::state_db;
use crate::product::agent::state_db::StateDbHandle;
use crate::product::protocol::protocol::InitialHistory;
use crate::product::protocol::protocol::ROLLOUT_SCHEMA_VERSION_V3;
use crate::product::protocol::protocol::ResumedHistory;
use crate::product::protocol::protocol::RolloutItem;
use crate::product::protocol::protocol::RolloutLine;
use crate::product::protocol::protocol::SessionMeta;
use crate::product::protocol::protocol::SessionMetaLine;
use crate::product::protocol::protocol::SessionSource;
use crate::product::protocol::protocol::current_rollout_schema_version;
use crate::product::state::ThreadMetadataBuilder;

#[derive(Debug)]
pub(crate) struct UnsupportedRolloutSchema {
    found: Option<u32>,
    expected: u32,
}

impl std::fmt::Display for UnsupportedRolloutSchema {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        match self.found {
            Some(found) => write!(
                f,
                "unsupported rollout schema version {found}; expected {}. Please start a new thread.",
                self.expected
            ),
            None => write!(
                f,
                "unsupported rollout schema version missing or invalid; expected {}. Please start a new thread.",
                self.expected
            ),
        }
    }
}

impl std::error::Error for UnsupportedRolloutSchema {}

pub(crate) fn unsupported_rollout_schema_error(found: u32) -> IoError {
    IoError::new(
        ErrorKind::InvalidData,
        UnsupportedRolloutSchema {
            found: Some(found),
            expected: current_rollout_schema_version(),
        },
    )
}

pub(crate) fn unsupported_rollout_schema_missing_error() -> IoError {
    IoError::new(
        ErrorKind::InvalidData,
        UnsupportedRolloutSchema {
            found: None,
            expected: current_rollout_schema_version(),
        },
    )
}

pub(crate) fn is_supported_rollout_schema_version(version: u32) -> bool {
    version == ROLLOUT_SCHEMA_VERSION_V3 || version == current_rollout_schema_version()
}

pub fn is_unsupported_rollout_schema_error(err: &IoError) -> bool {
    err.get_ref()
        .and_then(|inner| inner.downcast_ref::<UnsupportedRolloutSchema>())
        .is_some()
}

pub(crate) fn is_unsupported_rollout_schema_anyhow(err: &anyhow::Error) -> bool {
    err.chain().any(|cause| {
        cause
            .downcast_ref::<IoError>()
            .is_some_and(is_unsupported_rollout_schema_error)
    })
}

pub(crate) fn validate_rollout_line_schema_version(value: &Value) -> std::io::Result<()> {
    if value.get("type").and_then(Value::as_str) != Some("session_meta") {
        return Err(IoError::new(
            ErrorKind::InvalidData,
            "rollout does not start with session metadata",
        ));
    }

    let Some(version) = value
        .get("payload")
        .and_then(|payload| payload.get("rollout_schema_version"))
        .and_then(Value::as_u64)
        .and_then(|version| u32::try_from(version).ok())
    else {
        return Err(unsupported_rollout_schema_missing_error());
    };

    if !is_supported_rollout_schema_version(version) {
        return Err(unsupported_rollout_schema_error(version));
    }

    Ok(())
}

/// Records persisted rollout items for a session and flushes them to disk after every update.
///
/// Rollouts are recorded as JSONL and can be inspected with tools such as:
///
/// ```ignore
/// $ jq -C . ~/.lha/sessions/rollout-2025-05-07T17-24-21-5973b6c0-94b8-487b-a530-2aeb6098ae0e.jsonl
/// $ fx ~/.lha/sessions/rollout-2025-05-07T17-24-21-5973b6c0-94b8-487b-a530-2aeb6098ae0e.jsonl
/// ```
#[derive(Clone)]
pub struct RolloutRecorder {
    tx: Sender<RolloutCmd>,
    pub(crate) rollout_path: PathBuf,
    state_db: Option<StateDbHandle>,
}

#[derive(Clone)]
pub enum RolloutRecorderParams {
    Create {
        conversation_id: ThreadId,
        forked_from_id: Option<ThreadId>,
        source: SessionSource,
        base_instructions: BaseInstructions,
        dynamic_tools: Vec<DynamicToolSpec>,
        memory_mode: Option<String>,
    },
    Resume {
        path: PathBuf,
    },
}

enum RolloutCmd {
    AddItems(Vec<RolloutItem>),
    /// Ensure all prior writes are processed; respond when flushed.
    Flush {
        ack: oneshot::Sender<()>,
    },
    Shutdown {
        ack: oneshot::Sender<()>,
    },
}

impl RolloutRecorderParams {
    pub fn new(
        conversation_id: ThreadId,
        forked_from_id: Option<ThreadId>,
        source: SessionSource,
        base_instructions: BaseInstructions,
        dynamic_tools: Vec<DynamicToolSpec>,
        memory_mode: Option<String>,
    ) -> Self {
        Self::Create {
            conversation_id,
            forked_from_id,
            source,
            base_instructions,
            dynamic_tools,
            memory_mode,
        }
    }

    pub fn resume(path: PathBuf) -> Self {
        Self::Resume { path }
    }
}

impl RolloutRecorder {
    /// List threads (rollout files) under the provided LHA home directory.
    #[allow(clippy::too_many_arguments)]
    pub async fn list_threads(
        lha_home: &Path,
        page_size: usize,
        cursor: Option<&Cursor>,
        sort_key: ThreadSortKey,
        allowed_sources: &[SessionSource],
        model_providers: Option<&[String]>,
        default_provider: &str,
        cwd_filter: Option<&Path>,
    ) -> std::io::Result<ThreadsPage> {
        let stage = "list_threads";
        let page = get_threads(
            lha_home,
            page_size,
            cursor,
            sort_key,
            allowed_sources,
            model_providers,
            default_provider,
            cwd_filter,
        )
        .await?;

        // TODO(jif): drop after sqlite migration phase 1
        let state_db_ctx = state_db::open_if_present(lha_home, default_provider).await;
        if let Some(db_ids) = state_db::list_thread_ids_db(
            state_db_ctx.as_deref(),
            lha_home,
            page_size,
            cursor,
            sort_key,
            allowed_sources,
            model_providers,
            cwd_filter,
            false,
            stage,
        )
        .await
        {
            if page.items.len() != db_ids.len() {
                state_db::record_discrepancy(stage, "bad_len");
                return Ok(page);
            }
            for (id, item) in db_ids.iter().zip(page.items.iter()) {
                if !item.path.display().to_string().contains(&id.to_string()) {
                    state_db::record_discrepancy(stage, "bad_id");
                }
            }
        }
        Ok(page)
    }

    /// List archived threads (rollout files) under the archived sessions directory.
    #[allow(clippy::too_many_arguments)]
    pub async fn list_archived_threads(
        lha_home: &Path,
        page_size: usize,
        cursor: Option<&Cursor>,
        sort_key: ThreadSortKey,
        allowed_sources: &[SessionSource],
        model_providers: Option<&[String]>,
        default_provider: &str,
        cwd_filter: Option<&Path>,
    ) -> std::io::Result<ThreadsPage> {
        let stage = "list_archived_threads";
        let root = lha_home.join(ARCHIVED_SESSIONS_SUBDIR);
        let page = get_threads_in_root(
            root,
            page_size,
            cursor,
            sort_key,
            ThreadListConfig {
                allowed_sources,
                model_providers,
                cwd_filter,
                default_provider,
                layout: ThreadListLayout::Flat,
            },
        )
        .await?;

        // TODO(jif): drop after sqlite migration phase 1
        let state_db_ctx = state_db::open_if_present(lha_home, default_provider).await;
        if let Some(db_ids) = state_db::list_thread_ids_db(
            state_db_ctx.as_deref(),
            lha_home,
            page_size,
            cursor,
            sort_key,
            allowed_sources,
            model_providers,
            cwd_filter,
            true,
            stage,
        )
        .await
        {
            if page.items.len() != db_ids.len() {
                state_db::record_discrepancy(stage, "bad_len");
                return Ok(page);
            }
            for (id, item) in db_ids.iter().zip(page.items.iter()) {
                if !item.path.display().to_string().contains(&id.to_string()) {
                    state_db::record_discrepancy(stage, "bad_id");
                }
            }
        }
        Ok(page)
    }

    /// Find the newest recorded thread path, optionally filtering to a matching cwd.
    #[allow(clippy::too_many_arguments)]
    pub async fn find_latest_thread_path(
        lha_home: &Path,
        page_size: usize,
        cursor: Option<&Cursor>,
        sort_key: ThreadSortKey,
        allowed_sources: &[SessionSource],
        model_providers: Option<&[String]>,
        default_provider: &str,
        filter_cwd: Option<&Path>,
    ) -> std::io::Result<Option<PathBuf>> {
        let mut cursor = cursor.cloned();
        loop {
            let page = Self::list_threads(
                lha_home,
                page_size,
                cursor.as_ref(),
                sort_key,
                allowed_sources,
                model_providers,
                default_provider,
                filter_cwd,
            )
            .await?;
            if let Some(path) = page.items.first().map(|item| item.path.clone()) {
                return Ok(Some(path));
            }
            cursor = page.next_cursor;
            if cursor.is_none() {
                return Ok(None);
            }
        }
    }

    /// Attempt to create a new [`RolloutRecorder`]. If the sessions directory
    /// cannot be created or the rollout file cannot be opened we return the
    /// error so the caller can decide whether to disable persistence.
    pub async fn new(
        config: &Config,
        params: RolloutRecorderParams,
        state_db_ctx: Option<StateDbHandle>,
        state_builder: Option<ThreadMetadataBuilder>,
    ) -> std::io::Result<Self> {
        let (file, rollout_path, meta) = match params {
            RolloutRecorderParams::Create {
                conversation_id,
                forked_from_id,
                source,
                base_instructions,
                dynamic_tools,
                memory_mode,
            } => {
                let LogFileInfo {
                    file,
                    path,
                    conversation_id: session_id,
                    timestamp,
                } = create_log_file(config, conversation_id)?;

                let timestamp_format: &[FormatItem] = format_description!(
                    "[year]-[month]-[day]T[hour]:[minute]:[second].[subsecond digits:3]Z"
                );
                let timestamp = timestamp
                    .to_offset(time::UtcOffset::UTC)
                    .format(timestamp_format)
                    .map_err(|e| IoError::other(format!("failed to format timestamp: {e}")))?;

                (
                    tokio::fs::File::from_std(file),
                    path,
                    Some(SessionMeta {
                        id: session_id,
                        forked_from_id,
                        timestamp,
                        cwd: config.cwd.clone(),
                        originator: originator().value,
                        cli_version: env!("CARGO_PKG_VERSION").to_string(),
                        rollout_schema_version: current_rollout_schema_version(),
                        source,
                        model_provider: Some(config.model_provider_id.clone()),
                        base_instructions: Some(base_instructions),
                        dynamic_tools: if dynamic_tools.is_empty() {
                            None
                        } else {
                            Some(dynamic_tools)
                        },
                        memory_mode,
                    }),
                )
            }
            RolloutRecorderParams::Resume { path } => (
                tokio::fs::OpenOptions::new()
                    .append(true)
                    .open(&path)
                    .await?,
                path,
                None,
            ),
        };

        // Clone the cwd for the spawned task to collect git info asynchronously
        let cwd = config.cwd.clone();

        // A reasonably-sized bounded channel. If the buffer fills up the send
        // future will yield, which is fine – we only need to ensure we do not
        // perform *blocking* I/O on the caller's thread.
        let (tx, rx) = mpsc::channel::<RolloutCmd>(256);

        // Spawn a Tokio task that owns the file handle and performs async
        // writes. Using `tokio::fs::File` keeps everything on the async I/O
        // driver instead of blocking the runtime.
        tokio::task::spawn(rollout_writer(
            file,
            rx,
            meta,
            cwd,
            rollout_path.clone(),
            state_db_ctx.clone(),
            state_builder,
            config.model_provider_id.clone(),
        ));

        Ok(Self {
            tx,
            rollout_path,
            state_db: state_db_ctx,
        })
    }

    pub fn rollout_path(&self) -> &Path {
        self.rollout_path.as_path()
    }

    pub fn state_db(&self) -> Option<StateDbHandle> {
        self.state_db.clone()
    }

    pub(crate) async fn record_items(&self, items: &[RolloutItem]) -> std::io::Result<()> {
        let mut filtered = Vec::new();
        for item in items {
            // Note that function calls may look a bit strange if they are
            // "fully qualified MCP tool calls," so we could consider
            // reformatting them in that case.
            if is_persisted_response_item(item) {
                filtered.push(item.clone());
            }
        }
        if filtered.is_empty() {
            return Ok(());
        }
        self.tx
            .send(RolloutCmd::AddItems(filtered))
            .await
            .map_err(|e| IoError::other(format!("failed to queue rollout items: {e}")))
    }

    /// Flush all queued writes and wait until they are committed by the writer task.
    pub async fn flush(&self) -> std::io::Result<()> {
        let (tx, rx) = oneshot::channel();
        self.tx
            .send(RolloutCmd::Flush { ack: tx })
            .await
            .map_err(|e| IoError::other(format!("failed to queue rollout flush: {e}")))?;
        rx.await
            .map_err(|e| IoError::other(format!("failed waiting for rollout flush: {e}")))
    }

    pub(crate) async fn load_rollout_items(
        path: &Path,
    ) -> std::io::Result<(Vec<RolloutItem>, Option<ThreadId>, usize)> {
        info!("Resuming rollout from {path:?}");
        let text = tokio::fs::read_to_string(path).await?;
        if text.trim().is_empty() {
            return Err(IoError::other("empty session file"));
        }

        let mut items: Vec<RolloutItem> = Vec::new();
        let mut thread_id: Option<ThreadId> = None;
        let mut parse_errors = 0usize;
        let mut checked_schema_version = false;
        for line in text.lines() {
            if line.trim().is_empty() {
                continue;
            }
            let v: Value = match serde_json::from_str(line) {
                Ok(v) => v,
                Err(e) => {
                    warn!("failed to parse line as JSON: {line:?}, error: {e}");
                    parse_errors = parse_errors.saturating_add(1);
                    continue;
                }
            };
            if !checked_schema_version {
                validate_rollout_line_schema_version(&v)?;
                checked_schema_version = true;
            }

            // Parse the rollout line structure
            match serde_json::from_value::<RolloutLine>(v.clone()) {
                Ok(rollout_line) => match rollout_line.item {
                    RolloutItem::SessionMeta(session_meta_line) => {
                        if !is_supported_rollout_schema_version(
                            session_meta_line.meta.rollout_schema_version,
                        ) {
                            return Err(unsupported_rollout_schema_error(
                                session_meta_line.meta.rollout_schema_version,
                            ));
                        }
                        // Use the FIRST SessionMeta encountered in the file as the canonical
                        // thread id and main session information. Keep all items intact.
                        if thread_id.is_none() {
                            thread_id = Some(session_meta_line.meta.id);
                        }
                        items.push(RolloutItem::SessionMeta(session_meta_line));
                    }
                    RolloutItem::TranscriptItem(item) => {
                        items.push(RolloutItem::TranscriptItem(item));
                    }
                    RolloutItem::GhostSnapshot(item) => {
                        items.push(RolloutItem::GhostSnapshot(item));
                    }
                    RolloutItem::Compacted(item) => {
                        items.push(RolloutItem::Compacted(item));
                    }
                    RolloutItem::TurnContext(item) => {
                        items.push(RolloutItem::TurnContext(item));
                    }
                    RolloutItem::Workflow(item) => {
                        items.push(RolloutItem::Workflow(item));
                    }
                    RolloutItem::EventMsg(_ev) => {
                        items.push(RolloutItem::EventMsg(_ev));
                    }
                },
                Err(e) => {
                    warn!("failed to parse rollout line: {e}");
                    parse_errors = parse_errors.saturating_add(1);
                }
            }
        }

        tracing::debug!(
            "Resumed rollout with {} items, thread ID: {:?}, parse errors: {}",
            items.len(),
            thread_id,
            parse_errors,
        );
        Ok((items, thread_id, parse_errors))
    }

    pub async fn get_rollout_history(path: &Path) -> std::io::Result<InitialHistory> {
        let (items, thread_id, _parse_errors) = Self::load_rollout_items(path).await?;
        let conversation_id = thread_id
            .ok_or_else(|| IoError::other("failed to parse thread ID from rollout file"))?;

        if items.is_empty() {
            return Ok(InitialHistory::New);
        }

        info!("Resumed rollout successfully from {path:?}");
        Ok(InitialHistory::Resumed(ResumedHistory {
            conversation_id,
            history: items,
            rollout_path: path.to_path_buf(),
        }))
    }

    pub async fn shutdown(&self) -> std::io::Result<()> {
        let (tx_done, rx_done) = oneshot::channel();
        match self.tx.send(RolloutCmd::Shutdown { ack: tx_done }).await {
            Ok(_) => rx_done
                .await
                .map_err(|e| IoError::other(format!("failed waiting for rollout shutdown: {e}"))),
            Err(e) => {
                warn!("failed to send rollout shutdown command: {e}");
                Err(IoError::other(format!(
                    "failed to send rollout shutdown command: {e}"
                )))
            }
        }
    }
}

struct LogFileInfo {
    /// Opened file handle to the rollout file.
    file: File,

    /// Full path to the rollout file.
    path: PathBuf,

    /// Session ID (also embedded in filename).
    conversation_id: ThreadId,

    /// Timestamp for the start of the session.
    timestamp: OffsetDateTime,
}

fn create_log_file(config: &Config, conversation_id: ThreadId) -> std::io::Result<LogFileInfo> {
    // Resolve ~/.lha/sessions/YYYY/MM/DD and create it if missing.
    let timestamp = OffsetDateTime::now_local()
        .map_err(|e| IoError::other(format!("failed to get local time: {e}")))?;
    let mut dir = config.lha_home.clone();
    dir.push(SESSIONS_SUBDIR);
    dir.push(timestamp.year().to_string());
    dir.push(format!("{:02}", u8::from(timestamp.month())));
    dir.push(format!("{:02}", timestamp.day()));
    fs::create_dir_all(&dir)?;

    // Custom format for YYYY-MM-DDThh-mm-ss. Use `-` instead of `:` for
    // compatibility with filesystems that do not allow colons in filenames.
    let format: &[FormatItem] =
        format_description!("[year]-[month]-[day]T[hour]-[minute]-[second]");
    let date_str = timestamp
        .format(format)
        .map_err(|e| IoError::other(format!("failed to format timestamp: {e}")))?;

    let filename = format!("rollout-{date_str}-{conversation_id}.jsonl");

    let path = dir.join(filename);
    let file = std::fs::OpenOptions::new()
        .append(true)
        .create(true)
        .open(&path)?;

    Ok(LogFileInfo {
        file,
        path,
        conversation_id,
        timestamp,
    })
}

#[allow(clippy::too_many_arguments)]
async fn rollout_writer(
    file: tokio::fs::File,
    mut rx: mpsc::Receiver<RolloutCmd>,
    mut meta: Option<SessionMeta>,
    cwd: std::path::PathBuf,
    rollout_path: PathBuf,
    state_db_ctx: Option<StateDbHandle>,
    mut state_builder: Option<ThreadMetadataBuilder>,
    default_provider: String,
) -> std::io::Result<()> {
    let mut writer = JsonlWriter { file };
    if let Some(builder) = state_builder.as_mut() {
        builder.rollout_path = rollout_path.clone();
    }

    // If we have a meta, collect git info asynchronously and write meta first
    if let Some(session_meta) = meta.take() {
        let git_info = collect_git_info(&cwd).await;
        let session_meta_line = SessionMetaLine {
            meta: session_meta,
            git: git_info,
        };
        if state_db_ctx.is_some() {
            state_builder =
                metadata::builder_from_session_meta(&session_meta_line, rollout_path.as_path());
        }

        // Write the SessionMeta as the first item in the file, wrapped in a rollout line
        let rollout_item = RolloutItem::SessionMeta(session_meta_line);
        writer.write_rollout_item(&rollout_item).await?;
        state_db::reconcile_rollout(
            state_db_ctx.as_deref(),
            rollout_path.as_path(),
            default_provider.as_str(),
            state_builder.as_ref(),
            std::slice::from_ref(&rollout_item),
        )
        .await;
    }

    // Process rollout commands
    while let Some(cmd) = rx.recv().await {
        match cmd {
            RolloutCmd::AddItems(items) => {
                let mut persisted_items = Vec::new();
                for item in items {
                    if is_persisted_response_item(&item) {
                        writer.write_rollout_item(&item).await?;
                        persisted_items.push(item);
                    }
                }
                if persisted_items.is_empty() {
                    continue;
                }
                if let Some(builder) = state_builder.as_mut() {
                    builder.rollout_path = rollout_path.clone();
                }
                state_db::apply_rollout_items(
                    state_db_ctx.as_deref(),
                    rollout_path.as_path(),
                    default_provider.as_str(),
                    state_builder.as_ref(),
                    persisted_items.as_slice(),
                    "rollout_writer",
                )
                .await;
            }
            RolloutCmd::Flush { ack } => {
                // Ensure underlying file is flushed and then ack.
                if let Err(e) = writer.file.flush().await {
                    let _ = ack.send(());
                    return Err(e);
                }
                let _ = ack.send(());
            }
            RolloutCmd::Shutdown { ack } => {
                let _ = ack.send(());
            }
        }
    }

    Ok(())
}

struct JsonlWriter {
    file: tokio::fs::File,
}

#[derive(serde::Serialize)]
struct RolloutLineRef<'a> {
    timestamp: String,
    #[serde(flatten)]
    item: &'a RolloutItem,
}

impl JsonlWriter {
    async fn write_rollout_item(&mut self, rollout_item: &RolloutItem) -> std::io::Result<()> {
        let timestamp_format: &[FormatItem] = format_description!(
            "[year]-[month]-[day]T[hour]:[minute]:[second].[subsecond digits:3]Z"
        );
        let timestamp = OffsetDateTime::now_utc()
            .format(timestamp_format)
            .map_err(|e| IoError::other(format!("failed to format timestamp: {e}")))?;

        let line = RolloutLineRef {
            timestamp,
            item: rollout_item,
        };
        self.write_line(&line).await
    }
    async fn write_line(&mut self, item: &impl serde::Serialize) -> std::io::Result<()> {
        let mut json = serde_json::to_string(item)?;
        json.push('\n');
        self.file.write_all(json.as_bytes()).await?;
        self.file.flush().await?;
        Ok(())
    }
}
