pulpod 0.1.0

use std::sync::Arc;

use pulpo_common::event::PulpoEvent;
use pulpo_common::session::{InterventionCode, Session, SessionStatus};
use tracing::{debug, info};

use super::{
    IdleAction, IdleConfig, ReadyContext, build_session_event, detect_agent_exited,
    detect_and_store_output_metadata, detect_waiting_for_input, resolve_backend_id,
};
use crate::backend::Backend;
use crate::store::Store;

pub(super) async fn check_idle_sessions(
    backend: &Arc<dyn Backend>,
    store: &Store,
    idle_config: &IdleConfig,
    ready_ctx: &ReadyContext,
    extra_waiting_patterns: &[String],
) {
    let sessions = super::list_sessions_or_warn(store, "Idle check").await;

    let live: Vec<_> = sessions
        .into_iter()
        .filter(|session| {
            session.status == SessionStatus::Active || session.status == SessionStatus::Idle
        })
        .collect();

    let now = chrono::Utc::now();
    let timeout =
        chrono::Duration::seconds(idle_config.timeout_secs.try_into().unwrap_or(i64::MAX));

    for session in &live {
        check_session_idle(
            backend,
            store,
            idle_config,
            session,
            now,
            timeout,
            ready_ctx,
            extra_waiting_patterns,
        )
        .await;
    }
}

pub(super) async fn check_session_idle(
    backend: &Arc<dyn Backend>,
    store: &Store,
    idle_config: &IdleConfig,
    session: &Session,
    now: chrono::DateTime<chrono::Utc>,
    timeout: chrono::Duration,
    ready_ctx: &ReadyContext,
    extra_waiting_patterns: &[String],
) {
    let backend_id = resolve_backend_id(session, backend.as_ref());
    let current_output = match backend.capture_output(&backend_id, 500) {
        Ok(output) => output,
        #[allow(unused_variables)]
        Err(error) => {
            debug!(
                "Idle check: failed to capture output for {}: {error}",
                session.name
            );
            return;
        }
    };

    // The `.code` exit marker (written by `wrap_command` immediately after the
    // wrapped agent command exits) is a deterministic, race-free alternative to the
    // historical text-scrape below. The text scrape stays in place for compatibility
    // — e.g. adopted external tmux sessions have no `wrap_command` wrapper and thus
    // no marker, ever, and must keep relying on it (see `watchdog::adopt`).
    let marker_exit_code =
        crate::session::utils::read_exit_code_marker(store.data_dir(), &session.id.to_string());
    if marker_exit_code.is_some() || detect_agent_exited(&current_output) {
        handle_session_ready(store, session, ready_ctx, marker_exit_code).await;
        return;
    }

    #[allow(unused_variables)]
    if let Err(error) = store
        .update_session_output_snapshot(&session.id.to_string(), &current_output)
        .await
    {
        coverage_warn!(
            "Idle check: failed to update output snapshot for {}: {error}",
            session.name
        );
        return;
    }

    let exact_usage = crate::usage::read_exact_usage_for_session(session);
    detect_and_store_output_metadata(store, session, &current_output, exact_usage).await;

    let output_changed = session.output_snapshot.as_deref() != Some(current_output.as_str());
    if output_changed {
        handle_active_session(store, session, ready_ctx).await;
        return;
    }

    if session.status == SessionStatus::Active {
        let immediate = detect_waiting_for_input(&current_output, extra_waiting_patterns);
        let last_change = session.last_output_at.unwrap_or(session.created_at);
        let sustained = (now - last_change).num_seconds()
            >= i64::try_from(idle_config.threshold_secs).unwrap_or(i64::MAX);
        if immediate || sustained {
            info!(
                "Session {} idle ({}), transitioning to idle",
                session.name,
                if immediate {
                    "waiting pattern"
                } else {
                    "output unchanged"
                }
            );
            #[allow(unused_variables)]
            if let Err(error) = store
                .update_session_status(&session.id.to_string(), SessionStatus::Idle)
                .await
            {
                coverage_warn!(
                    "Idle check: failed to transition {} to idle: {error}",
                    session.name
                );
            } else if let Some(tx) = &ready_ctx.event_tx {
                let event = build_session_event(
                    session,
                    SessionStatus::Idle,
                    Some(SessionStatus::Active),
                    &ready_ctx.node_name,
                    Some(current_output.clone()),
                );
                let _ = tx.send(PulpoEvent::Session(event));
            }
            return;
        }
    }

    handle_idle_session(
        backend,
        store,
        idle_config,
        session,
        now,
        timeout,
        ready_ctx,
    )
    .await;
}

/// Transition a session to `Ready`: the agent process has exited but the wrapper's
/// fallback shell is still lingering in tmux (still resumable, still alive). `exit_code`
/// is `Some` when the `.code` exit marker was found and parsed (see `check_session_idle`);
/// it is persisted alongside the status update so the exit code is visible even before
/// the session eventually resolves to `Stopped`/`Lost`.
pub(super) async fn handle_session_ready(
    store: &Store,
    session: &Session,
    ctx: &ReadyContext,
    exit_code: Option<i32>,
) {
    let previous = session.status;
    info!(
        session_name = %session.name,
        "Agent exited, transitioning to ready"
    );
    if let Some(code) = exit_code {
        #[allow(unused_variables)]
        if let Err(error) = store
            .update_session_exit_code(&session.id.to_string(), code)
            .await
        {
            coverage_warn!(
                session_name = %session.name,
                "Failed to record exit code: {error}"
            );
        }
    }
    #[allow(unused_variables)]
    if let Err(error) = store
        .update_session_status(&session.id.to_string(), SessionStatus::Ready)
        .await
    {
        coverage_warn!(
            session_name = %session.name,
            "Failed to transition to ready: {error}"
        );
        return;
    }

    if let Some(tx) = &ctx.event_tx {
        let event = build_session_event(
            session,
            SessionStatus::Ready,
            Some(previous),
            &ctx.node_name,
            session.output_snapshot.clone(),
        );
        let _ = tx.send(PulpoEvent::Session(event));
    }
}

pub(super) async fn handle_active_session(
    store: &Store,
    session: &Session,
    ready_ctx: &ReadyContext,
) {
    if session.status == SessionStatus::Idle {
        info!(
            "Session {} has new output, transitioning back to active",
            session.name
        );
        #[allow(unused_variables)]
        if let Err(error) = store
            .update_session_status(&session.id.to_string(), SessionStatus::Active)
            .await
        {
            coverage_warn!(
                "Idle check: failed to transition {} back to active: {error}",
                session.name
            );
        } else if let Some(tx) = &ready_ctx.event_tx {
            let event = build_session_event(
                session,
                SessionStatus::Active,
                Some(SessionStatus::Idle),
                &ready_ctx.node_name,
                session.output_snapshot.clone(),
            );
            let _ = tx.send(PulpoEvent::Session(event));
        }
    }

    if session.idle_since.is_none() {
        return;
    }

    info!(
        "Idle check: session {} active again, clearing idle status",
        session.name
    );
    #[allow(unused_variables)]
    if let Err(error) = store
        .clear_session_idle_since(&session.id.to_string())
        .await
    {
        coverage_warn!(
            "Idle check: failed to clear idle_since for {}: {error}",
            session.name
        );
    }
}

pub(super) async fn handle_idle_session(
    backend: &Arc<dyn Backend>,
    store: &Store,
    idle_config: &IdleConfig,
    session: &Session,
    now: chrono::DateTime<chrono::Utc>,
    timeout: chrono::Duration,
    ready_ctx: &ReadyContext,
) {
    let last_activity = session.last_output_at.unwrap_or(session.created_at);
    let idle_duration = now - last_activity;

    if idle_duration <= timeout {
        return;
    }

    let minutes = idle_duration.num_minutes();
    match idle_config.action {
        IdleAction::Alert => {
            if session.idle_since.is_none() {
                coverage_warn!(
                    "Idle check: session {} idle for {minutes} minutes, marking as idle",
                    session.name
                );
                #[allow(unused_variables)]
                if let Err(error) = store
                    .update_session_idle_since(&session.id.to_string())
                    .await
                {
                    coverage_warn!(
                        "Idle check: failed to set idle_since for {}: {error}",
                        session.name
                    );
                }
            }
        }
        IdleAction::Kill => {
            let reason = format!("Idle for {minutes} minutes");

            // Shared breaker path: captures a final output snapshot before the
            // kill, so idle-killed sessions keep their last output.
            if !super::intervention::stop_and_record(
                backend,
                store,
                session,
                InterventionCode::IdleTimeout,
                &reason,
                ready_ctx,
                "Idle check: failed to kill idle session",
                "Idle check: failed to record intervention",
            )
            .await
            {
                return;
            }
            coverage_warn!(
                "Idle check: stopped idle session {} after {minutes} minutes",
                session.name
            );
        }
    }
}

pub(super) async fn cleanup_ready_sessions(
    backend: &Arc<dyn Backend>,
    store: &Store,
    ready_ttl_secs: u64,
) {
    let sessions = super::list_sessions_or_warn(store, "Ready cleanup").await;

    let now = chrono::Utc::now();
    let ttl = chrono::Duration::seconds(ready_ttl_secs.try_into().unwrap_or(i64::MAX));

    for session in sessions
        .iter()
        .filter(|session| session.status == SessionStatus::Ready)
    {
        let age = now - session.updated_at;
        if age <= ttl {
            continue;
        }

        let backend_id = resolve_backend_id(session, backend.as_ref());
        #[allow(unused_variables)]
        if let Err(error) = backend.kill_session(&backend_id) {
            debug!(
                session_name = %session.name,
                "Ready cleanup: tmux already gone: {error}"
            );
        }
        #[allow(unused_variables)]
        if let Err(error) = store
            .update_session_status(&session.id.to_string(), SessionStatus::Stopped)
            .await
        {
            coverage_warn!(
                session_name = %session.name,
                "Ready cleanup: failed to mark stopped: {error}"
            );
        } else {
            info!(
                session_name = %session.name,
                age_secs = age.num_seconds(),
                "Ready cleanup: stopped tmux shell after TTL"
            );
        }
    }
}