use anyhow::Result;
use rusqlite::{Connection, OptionalExtension};

use super::types::{Check, Status};
use crate::db;

const STALE_CAPTURE_HEARTBEAT_SECS: i64 = 7 * 24 * 60 * 60;
const SUMMARY_HEARTBEAT_GRACE_SECS: i64 = 60;

pub(super) fn check_capture_liveness(conn: Option<&Connection>, setup_checks: &[Check]) -> Check {
    let setup_findings = capture_setup_findings(setup_checks);
    let mut failures: Vec<String> = setup_findings
        .iter()
        .filter(|finding| matches!(finding.status, Status::Fail))
        .map(|finding| finding.detail.clone())
        .collect();
    let warnings: Vec<String> = setup_findings
        .iter()
        .filter(|finding| matches!(finding.status, Status::Warn))
        .map(|finding| finding.detail.clone())
        .collect();

    let Some(conn) = conn else {
        if failures.is_empty() {
            return Check::new("Capture liveness", Status::Warn, "cannot open database");
        }
        return Check::new("Capture liveness", Status::Fail, failures.join("; "));
    };

    let stats = match db::query_system_stats(conn) {
        Ok(stats) => stats,
        Err(err) => {
            if failures.is_empty() {
                return Check::new(
                    "Capture liveness",
                    Status::Warn,
                    format!("cannot load capture stats: {}", err),
                );
            }
            failures.push(format!("cannot load capture stats: {err}"));
            return Check::new("Capture liveness", Status::Fail, failures.join("; "));
        }
    };

    if stats.failed_pending_observations > 0 || stats.failed_extraction_tasks > 0 {
        failures.push(format!(
            "failed-observation backlog: {} failed pending observations, {} failed extraction tasks; run `remem pending list-failed --limit 20` and `remem worker --once`",
            stats.failed_pending_observations, stats.failed_extraction_tasks
        ));
    }
    if stats.actionable_capture_drops > 0 {
        failures.push(format!(
            "{} actionable capture drop(s); latest reason={}",
            stats.actionable_capture_drops,
            stats
                .latest_capture_drop_reason
                .as_deref()
                .unwrap_or("unknown")
        ));
    }

    let liveness = match query_liveness_rows(conn) {
        Ok(liveness) => liveness,
        Err(err) => {
            if failures.is_empty() {
                return Check::new(
                    "Capture liveness",
                    Status::Warn,
                    format!("cannot load capture liveness rows: {}", err),
                );
            }
            failures.push(format!("cannot load capture liveness rows: {err}"));
            return Check::new("Capture liveness", Status::Fail, failures.join("; "));
        }
    };
    if let Some(gap) = liveness.hosted_summary_without_capture {
        failures.push(format!(
            "{} hosted session summary row(s) have no captured_events heartbeat; latest session_row_id={} host={}",
            gap.count,
            gap.latest_session_row_id.unwrap_or_default(),
            gap.latest_host.unwrap_or_else(|| "unknown".to_string())
        ));
    }
    if stats.session_summaries > 0 && stats.latest_capture_activity_epoch.is_none() {
        failures.push(format!(
            "{} completed session summary row(s), but no captured_events/raw_messages/capture_drop_events heartbeat; hooks are not recording capture activity",
            stats.session_summaries
        ));
    }
    if let (Some(summary_epoch), Some(heartbeat_epoch)) = (
        liveness.latest_session_summary_epoch,
        stats.latest_capture_activity_epoch,
    ) {
        if summary_epoch.saturating_sub(heartbeat_epoch) > SUMMARY_HEARTBEAT_GRACE_SECS {
            failures.push(format!(
                "completed session summary is newer than latest capture heartbeat by {}s; hooks may be missing or stale",
                summary_epoch.saturating_sub(heartbeat_epoch)
            ));
        }
    }

    if !failures.is_empty() {
        return Check::new("Capture liveness", Status::Fail, failures.join("; "));
    }
    if stats.latest_capture_activity_epoch.is_none() {
        return Check::new(
            "Capture liveness",
            Status::Warn,
            join_detail(
                &warnings,
                "no capture heartbeat yet; run one host session and re-run doctor",
            ),
        );
    }

    let heartbeat_age_secs = chrono::Utc::now()
        .timestamp()
        .saturating_sub(stats.latest_capture_activity_epoch.unwrap_or_default());
    let detail = format!(
        "latest capture heartbeat {}s ago; captured_events={}, raw_messages={}, expected drops={}",
        heartbeat_age_secs, stats.captured_events, stats.raw_messages, stats.capture_drop_events
    );
    if heartbeat_age_secs > STALE_CAPTURE_HEARTBEAT_SECS {
        return Check::new(
            "Capture liveness",
            Status::Warn,
            join_detail(
                &warnings,
                &format!(
                    "{detail}; stale capture heartbeat exceeds {}s, run one host session and re-run doctor",
                    STALE_CAPTURE_HEARTBEAT_SECS
                ),
            ),
        );
    }
    if !warnings.is_empty() {
        return Check::new(
            "Capture liveness",
            Status::Warn,
            join_detail(&warnings, &detail),
        );
    }

    Check::new("Capture liveness", Status::Ok, detail)
}

#[derive(Clone, PartialEq, Eq)]
struct SetupFinding {
    status: Status,
    detail: String,
}

fn capture_setup_findings(checks: &[Check]) -> Vec<SetupFinding> {
    checks
        .iter()
        .filter_map(|check| {
            if check.name.starts_with("Hooks") {
                return hook_setup_finding(check);
            }
            if check.name == "Install paths" {
                return install_path_setup_finding(check);
            }
            None
        })
        .collect()
}

fn hook_setup_finding(check: &Check) -> Option<SetupFinding> {
    match check.status {
        Status::Fail => Some(SetupFinding {
            status: Status::Fail,
            detail: format!("{} failed: {}", check.name, check.detail),
        }),
        Status::Warn if hook_warning_blocks_capture(&check.detail) => Some(SetupFinding {
            status: Status::Fail,
            detail: format!("{} stale or incomplete: {}", check.name, check.detail),
        }),
        _ => None,
    }
}

fn hook_warning_blocks_capture(detail: &str) -> bool {
    detail.contains(" registered (run `remem install --target")
}

fn install_path_setup_finding(check: &Check) -> Option<SetupFinding> {
    match check.status {
        Status::Fail | Status::Warn => Some(SetupFinding {
            status: Status::Fail,
            detail: format!(
                "Install paths stale or mismatched: {}; hooks may execute a stale binary",
                check.detail
            ),
        }),
        Status::Ok => None,
    }
}

fn join_detail(prefixes: &[String], detail: &str) -> String {
    if prefixes.is_empty() {
        detail.to_string()
    } else {
        format!("{}; {detail}", prefixes.join("; "))
    }
}

#[derive(Debug, Clone, Default, PartialEq, Eq)]
struct CaptureLivenessRows {
    latest_session_summary_epoch: Option<i64>,
    hosted_summary_without_capture: Option<HostedSummaryGap>,
}

#[derive(Debug, Clone, PartialEq, Eq)]
struct HostedSummaryGap {
    count: i64,
    latest_session_row_id: Option<i64>,
    latest_host: Option<String>,
}

fn query_liveness_rows(conn: &Connection) -> Result<CaptureLivenessRows> {
    let latest_session_summary_epoch = conn.query_row(
        "SELECT MAX(created_at_epoch)
         FROM session_summaries
         WHERE created_at_epoch IS NOT NULL",
        [],
        |row| row.get::<_, Option<i64>>(0),
    )?;
    let hosted_summary_without_capture = query_hosted_summary_without_capture(conn)?;
    Ok(CaptureLivenessRows {
        latest_session_summary_epoch,
        hosted_summary_without_capture,
    })
}

fn query_hosted_summary_without_capture(conn: &Connection) -> Result<Option<HostedSummaryGap>> {
    let count = conn.query_row(
        "SELECT COUNT(*)
         FROM session_summaries ss
         WHERE ss.session_row_id IS NOT NULL
           AND NOT EXISTS (
               SELECT 1 FROM captured_events ce
               WHERE ce.session_row_id = ss.session_row_id
           )",
        [],
        |row| row.get::<_, i64>(0),
    )?;
    if count == 0 {
        return Ok(None);
    }
    let latest = conn
        .query_row(
            "SELECT ss.session_row_id, h.name
             FROM session_summaries ss
             LEFT JOIN hosts h ON h.id = ss.host_id
             WHERE ss.session_row_id IS NOT NULL
               AND NOT EXISTS (
                   SELECT 1 FROM captured_events ce
                   WHERE ce.session_row_id = ss.session_row_id
               )
             ORDER BY ss.created_at_epoch DESC, ss.id DESC
             LIMIT 1",
            [],
            |row| {
                Ok((
                    row.get::<_, Option<i64>>(0)?,
                    row.get::<_, Option<String>>(1)?,
                ))
            },
        )
        .optional()?;
    Ok(Some(HostedSummaryGap {
        count,
        latest_session_row_id: latest.as_ref().and_then(|row| row.0),
        latest_host: latest.and_then(|row| row.1),
    }))
}

#[cfg(test)]
mod tests {
    use rusqlite::params;

    use super::*;

    fn setup_liveness_conn() -> anyhow::Result<Connection> {
        let conn = Connection::open_in_memory()?;
        crate::migrate::run_migrations(&conn)?;
        Ok(conn)
    }

    fn record_liveness_capture(conn: &Connection) -> anyhow::Result<()> {
        crate::db::record_captured_event(
            conn,
            &crate::db::CaptureEventInput {
                host: "codex-cli",
                session_id: "sess-doctor",
                project: "/tmp/remem-doctor",
                cwd: Some("/tmp/remem-doctor"),
                event_type: "message",
                role: Some("user"),
                tool_name: None,
                content: "captured event",
                task_kind: None,
            },
        )?;
        Ok(())
    }

    #[test]
    fn capture_liveness_fails_on_failed_observation_backlog() -> anyhow::Result<()> {
        let conn = setup_liveness_conn()?;
        let id = crate::db::enqueue_pending(
            &conn,
            "codex-cli",
            "sess-failed",
            "/tmp/remem",
            "Bash",
            Some("{}"),
            Some("{}"),
            Some("/tmp/remem"),
        )?;
        conn.execute(
            "UPDATE pending_observations SET status = 'failed' WHERE id = ?1",
            [id],
        )?;

        let check = check_capture_liveness(Some(&conn), &[]);

        assert!(matches!(check.status, Status::Fail));
        assert!(check.detail.contains("failed-observation backlog"));
        assert!(check.detail.contains("failed pending observations"));
        Ok(())
    }

    #[test]
    fn capture_liveness_fails_when_summary_is_newer_than_heartbeat() -> anyhow::Result<()> {
        let conn = setup_liveness_conn()?;
        record_liveness_capture(&conn)?;
        conn.execute(
            "INSERT INTO session_summaries
             (memory_session_id, project, request, completed, created_at_epoch)
             VALUES ('legacy-newer-summary', '/tmp/remem-doctor', 'done', 'done', strftime('%s', 'now') + 120)",
            [],
        )?;

        let check = check_capture_liveness(Some(&conn), &[]);

        assert!(matches!(check.status, Status::Fail));
        assert!(check
            .detail
            .contains("summary is newer than latest capture heartbeat"));
        Ok(())
    }

    #[test]
    fn capture_liveness_fails_when_hosted_summary_has_no_capture() -> anyhow::Result<()> {
        let conn = setup_liveness_conn()?;
        conn.execute(
            "INSERT OR IGNORE INTO hosts(name, enabled, created_at_epoch)
             VALUES ('codex-cli', 1, strftime('%s', 'now'))",
            [],
        )?;
        let host_id: i64 =
            conn.query_row("SELECT id FROM hosts WHERE name = 'codex-cli'", [], |row| {
                row.get(0)
            })?;
        conn.execute(
            "INSERT INTO session_summaries
             (memory_session_id, project, request, completed, created_at_epoch, host_id, session_row_id)
             VALUES ('hosted-without-capture', '/tmp/remem-doctor', 'done', 'done',
                     strftime('%s', 'now'), ?1, 9876)",
            params![host_id],
        )?;

        let check = check_capture_liveness(Some(&conn), &[]);

        assert!(matches!(check.status, Status::Fail));
        assert!(check.detail.contains("hosted session summary"));
        assert!(check.detail.contains("session_row_id=9876"));
        Ok(())
    }

    #[test]
    fn capture_liveness_fails_on_missing_hook_setup() {
        let setup = vec![Check::new(
            "Hooks (codex)",
            Status::Fail,
            "no remem hooks (run `remem install --target codex`)",
        )];

        let check = check_capture_liveness(None, &setup);

        assert!(matches!(check.status, Status::Fail));
        assert!(check.detail.contains("Hooks (codex) failed"));
    }

    #[test]
    fn capture_liveness_fails_on_partial_hook_setup() {
        let setup = vec![Check::new(
            "Hooks (codex)",
            Status::Warn,
            "1/2 registered (run `remem install --target codex` to fix)",
        )];

        let check = check_capture_liveness(None, &setup);

        assert!(matches!(check.status, Status::Fail));
        assert!(check.detail.contains("stale or incomplete"));
    }

    #[test]
    fn capture_liveness_fails_on_stale_install_path_setup() {
        let setup = vec![Check::new(
            "Install paths",
            Status::Warn,
            "2 remem executable(s) found; configured /opt/remem; candidates: /usr/local/bin/remem (0.5.1); fix: remove or upgrade stale installs",
        )];

        let check = check_capture_liveness(None, &setup);

        assert!(matches!(check.status, Status::Fail));
        assert!(check.detail.contains("Install paths stale or mismatched"));
        assert!(check.detail.contains("stale binary"));
    }
}
