//! The gate live log: where a running gate's output goes.
//!
//! Every gate that shells out streams its bytes through here, so the TUI sees
//! the tail as it happens and the finished log is on disk under
//! [`super::GateCtx::log_path`] for the failure note to point at.

use super::GateCtx;
use crate::domain::{GateKind, GateRunId};
use crate::events::{self, Event, EventTx};
use anyhow::Result;
use ops_core::live_log::LiveLog;
use ops_core::remote::LogSink;
use std::path::PathBuf;
use std::sync::Arc;
use tokio::io::AsyncReadExt;
use tokio::process::Command;

/// The gate live-log callback: emit each chunk as a `GateLogChunk` event so the
/// TUI sees the tail stream in real time. `ops_core::live_log::LiveLog` owns the
/// disk append and the per-run sequence counter; this closure is the one
/// tool-specific bit.
pub(super) fn gate_chunk_cb(
    events: EventTx,
    run_id: GateRunId,
) -> ops_core::live_log::ChunkCallback {
    Box::new(move |seq, text| {
        events::emit(
            &events,
            Event::GateLogChunk {
                run_id,
                seq,
                text: text.to_owned(),
            },
        );
    })
}

/// Append raw bytes to a gate log outside the child-streaming path (target
/// banners). Best-effort, same as `LiveLog`: a broken log dir never turns a
/// passing gate red.
pub(super) async fn append_to_log(path: &std::path::Path, bytes: &[u8]) {
    use tokio::io::AsyncWriteExt;
    if let Some(parent) = path.parent()
        && tokio::fs::create_dir_all(parent).await.is_err()
    {
        return;
    }
    if let Ok(mut f) = tokio::fs::OpenOptions::new()
        .create(true)
        .append(true)
        .open(path)
        .await
    {
        let _ = f.write_all(bytes).await;
    }
}

/// Drain `stream` into the shared `LiveLog` (which forwards each chunk to
/// the on-disk log file AND broadcasts a `GateLogChunk` event), and return
/// the concatenated bytes so the classifier can still operate on the full
/// output post-hoc.
pub(super) async fn stream_into_log<R>(
    stream: Option<R>,
    log: std::sync::Arc<tokio::sync::Mutex<LiveLog>>,
) -> Vec<u8>
where
    R: tokio::io::AsyncRead + Unpin + Send + 'static,
{
    let mut total = Vec::new();
    let Some(mut s) = stream else { return total };
    let mut buf = [0u8; 4096];
    loop {
        match s.read(&mut buf).await {
            Ok(0) => break,
            Err(_) => break,
            Ok(n) => {
                total.extend_from_slice(&buf[..n]);
                log.lock().await.write_chunk(&buf[..n]).await;
            }
        }
    }
    total
}

/// Spawn a child, drain its stdout/stderr through a `LiveLog`, return the
/// combined buffers and exit status. Shared by `cargo_test` (no deadline)
/// and ad-hoc callers — `boot_smoke` rolls its own variant because of its
/// 3s kill window.
pub(super) async fn stream_child_to_live_log(
    child: &mut tokio::process::Child,
    events: EventTx,
    run_id: GateRunId,
    log_path: PathBuf,
) -> Result<(Vec<u8>, Vec<u8>, std::process::ExitStatus)> {
    let log = GateLog::new(LiveLog::open(log_path, gate_chunk_cb(events, run_id)).await);
    let out = log.stream_child(child).await;
    log.close().await;
    out
}

/// One gate's live log, held across every step of a multi-step gate.
///
/// Single-child gates call [`stream_child_to_live_log`] and are done. The
/// staged gates (`code_smoke`, `migration_dry_run`) instead run several
/// children plus banner lines between them, and they hold one `GateLog` across
/// the lot: a single sink means chunk sequence numbers stay monotonic for the
/// whole gate, and the on-disk log reads in the order things actually happened
/// rather than as stdout-then-stderr assembled at the end.
///
/// Every write is best-effort in the same way `LiveLog` is: a log directory
/// that cannot be written degrades to callback-only and never turns a passing
/// gate red.
pub(super) struct GateLog {
    sink: Arc<tokio::sync::Mutex<LiveLog>>,
}

impl GateLog {
    pub(super) fn new(sink: LiveLog) -> Self {
        Self {
            sink: Arc::new(tokio::sync::Mutex::new(sink)),
        }
    }

    /// Open the sink for `gate`'s log file, streaming to the TUI under `run_id`.
    pub(super) async fn open(ctx: &GateCtx, run_id: GateRunId, gate: GateKind) -> Self {
        Self::new(
            LiveLog::open(
                ctx.log_path(gate),
                gate_chunk_cb(ctx.events.clone(), run_id),
            )
            .await,
        )
    }

    /// Emit a banner (or any line the gate itself produces) through the same
    /// sink the children stream to, so it lands in sequence with their output.
    pub(super) async fn write(&self, bytes: &[u8]) {
        self.sink.lock().await.write_chunk(bytes).await;
    }

    /// Same, for the common `format!`-a-line case.
    pub(super) async fn line(&self, s: &str) {
        self.write(s.as_bytes()).await;
    }

    /// Spawn `cmd` with both pipes captured, stream them into the sink as they
    /// arrive, and return the buffers plus the exit status. The buffers are
    /// what the classifiers still operate on post-hoc.
    pub(super) async fn run(
        &self,
        cmd: &mut Command,
    ) -> std::io::Result<(Vec<u8>, Vec<u8>, std::process::ExitStatus)> {
        cmd.stdout(std::process::Stdio::piped())
            .stderr(std::process::Stdio::piped());
        let mut child = cmd.spawn()?;
        self.stream_child(&mut child)
            .await
            .map_err(std::io::Error::other)
    }

    /// Drain an already-spawned child's pipes into the sink and wait for it.
    pub(super) async fn stream_child(
        &self,
        child: &mut tokio::process::Child,
    ) -> Result<(Vec<u8>, Vec<u8>, std::process::ExitStatus)> {
        let (stdout_task, stderr_task) = self.drain_pipes(child);
        let status = child.wait().await?;
        let stdout_buf = stdout_task.await.unwrap_or_default();
        let stderr_buf = stderr_task.await.unwrap_or_default();
        Ok((stdout_buf, stderr_buf, status))
    }

    /// Start draining a child's pipes into the sink *without* waiting on the
    /// child. For `code_smoke`'s serve phase, which probes `/health` while the
    /// server is still up. The caller must await both handles for the buffers
    /// (and before [`Self::close`], so the flush isn't skipped).
    #[allow(clippy::type_complexity)]
    pub(super) fn drain_pipes(
        &self,
        child: &mut tokio::process::Child,
    ) -> (
        tokio::task::JoinHandle<Vec<u8>>,
        tokio::task::JoinHandle<Vec<u8>>,
    ) {
        (
            tokio::spawn(stream_into_log(child.stdout.take(), self.sink.clone())),
            tokio::spawn(stream_into_log(child.stderr.take(), self.sink.clone())),
        )
    }

    /// Flush the file. A still-outstanding streaming task (only possible if the
    /// gate returned without awaiting it) leaves the `Arc` shared, in which case
    /// the flush is skipped — `LiveLog` writes unbuffered to the OS either way,
    /// so nothing already written is lost.
    pub(super) async fn close(self) {
        if let Ok(mutex) = Arc::try_unwrap(self.sink) {
            mutex.into_inner().close().await;
        }
    }
}