use anyhow::{bail, Result};
use rusqlite::{params, Connection};
use serde::Serialize;

use crate::db::{self, CaptureEventInput, ExtractionTaskKind};

#[derive(Debug, Clone, Serialize, PartialEq, Eq)]
pub struct LegacyPendingMigration {
    pub pending_id: i64,
    pub event_id: String,
    pub captured_event_id: i64,
    pub extraction_task_id: i64,
    pub host: String,
    pub project: String,
    pub session_id: String,
}

struct LegacyPendingRow {
    id: i64,
    host: String,
    session_id: String,
    project: String,
    tool_name: String,
    tool_input: Option<String>,
    tool_response: Option<String>,
    cwd: Option<String>,
    created_at_epoch: i64,
}

pub fn count_legacy_migration_candidates(
    conn: &Connection,
    project: Option<&str>,
    limit: i64,
) -> Result<usize> {
    let limit = limit.max(1);
    let now = chrono::Utc::now().timestamp();
    let count: i64 = if let Some(project) = project {
        conn.query_row(
            "SELECT COUNT(*) FROM (
                 SELECT id FROM pending_observations
                 WHERE project = ?1
                   AND (status = 'pending'
                        OR (status = 'processing'
                            AND (lease_expires_epoch IS NULL OR lease_expires_epoch < ?3)))
                 ORDER BY created_at_epoch ASC, id ASC
                 LIMIT ?2
             )",
            params![project, limit, now],
            |row| row.get(0),
        )?
    } else {
        conn.query_row(
            "SELECT COUNT(*) FROM (
                 SELECT id FROM pending_observations
                 WHERE status = 'pending'
                    OR (status = 'processing'
                        AND (lease_expires_epoch IS NULL OR lease_expires_epoch < ?2))
                 ORDER BY created_at_epoch ASC, id ASC
                 LIMIT ?1
             )",
            params![limit, now],
            |row| row.get(0),
        )?
    };
    Ok(count.max(0) as usize)
}

pub fn migrate_legacy_pending(
    conn: &mut Connection,
    project: Option<&str>,
    fallback_host: Option<&str>,
    limit: i64,
) -> Result<Vec<LegacyPendingMigration>> {
    let fallback_host = fallback_host.map(normalize_capture_host).transpose()?;
    let tx = conn.transaction()?;
    let rows = select_legacy_pending_rows(&tx, project, limit)?;
    let mut migrated = Vec::new();

    for row in rows {
        let host = capture_host_for_row(&row.host, fallback_host)?;
        let event_id = legacy_event_id(row.id);
        let content = legacy_capture_content(&row);
        let outcome = db::record_captured_event_with_id_and_created_at(
            &tx,
            &CaptureEventInput {
                host,
                session_id: &row.session_id,
                project: &row.project,
                cwd: row.cwd.as_deref(),
                event_type: "tool_result",
                role: None,
                tool_name: Some(&row.tool_name),
                content: &content,
                task_kind: Some(ExtractionTaskKind::ObservationExtract),
            },
            Some(&event_id),
            row.created_at_epoch,
        )?;
        let extraction_task_id = outcome.extraction_task_id.ok_or_else(|| {
            anyhow::anyhow!("legacy pending migration did not enqueue extraction")
        })?;
        let now = chrono::Utc::now().timestamp();
        let changed = tx.execute(
            "UPDATE pending_observations
             SET status = 'migrated',
                 lease_owner = NULL,
                 lease_expires_epoch = NULL,
                 next_retry_epoch = NULL,
                 last_error = NULL,
                 updated_at_epoch = ?2
             WHERE id = ?1
               AND (status = 'pending'
                    OR (status = 'processing'
                        AND (lease_expires_epoch IS NULL OR lease_expires_epoch < ?3)))",
            params![row.id, now, now],
        )?;
        if changed != 1 {
            bail!("legacy pending row {} changed while migrating", row.id);
        }
        migrated.push(LegacyPendingMigration {
            pending_id: row.id,
            event_id,
            captured_event_id: outcome.event_row_id,
            extraction_task_id,
            host: host.to_string(),
            project: row.project,
            session_id: row.session_id,
        });
    }

    tx.commit()?;
    Ok(migrated)
}

fn select_legacy_pending_rows(
    conn: &Connection,
    project: Option<&str>,
    limit: i64,
) -> Result<Vec<LegacyPendingRow>> {
    let limit = limit.max(1);
    let now = chrono::Utc::now().timestamp();
    let sql = if project.is_some() {
        "SELECT id, host, session_id, project, tool_name, tool_input, tool_response, cwd, created_at_epoch
         FROM pending_observations
         WHERE project = ?1
           AND (status = 'pending'
                OR (status = 'processing'
                    AND (lease_expires_epoch IS NULL OR lease_expires_epoch < ?3)))
         ORDER BY created_at_epoch ASC, id ASC
         LIMIT ?2"
    } else {
        "SELECT id, host, session_id, project, tool_name, tool_input, tool_response, cwd, created_at_epoch
         FROM pending_observations
         WHERE status = 'pending'
            OR (status = 'processing'
                AND (lease_expires_epoch IS NULL OR lease_expires_epoch < ?2))
         ORDER BY created_at_epoch ASC, id ASC
         LIMIT ?1"
    };
    let mut stmt = conn.prepare(sql)?;
    let rows = if let Some(project) = project {
        stmt.query_map(params![project, limit, now], row_from_db)?
    } else {
        stmt.query_map(params![limit, now], row_from_db)?
    };
    rows.collect::<std::result::Result<Vec<_>, _>>()
        .map_err(Into::into)
}

fn row_from_db(row: &rusqlite::Row<'_>) -> rusqlite::Result<LegacyPendingRow> {
    Ok(LegacyPendingRow {
        id: row.get(0)?,
        host: row.get(1)?,
        session_id: row.get(2)?,
        project: row.get(3)?,
        tool_name: row.get(4)?,
        tool_input: row.get(5)?,
        tool_response: row.get(6)?,
        cwd: row.get(7)?,
        created_at_epoch: row.get(8)?,
    })
}

fn capture_host_for_row<'a>(row_host: &'a str, fallback_host: Option<&'a str>) -> Result<&'a str> {
    match normalize_capture_host(row_host) {
        Ok(host) => Ok(host),
        Err(_) => fallback_host
            .ok_or_else(|| anyhow::anyhow!("legacy pending row has host='{row_host}'; pass --host claude-code or --host codex-cli")),
    }
}

fn normalize_capture_host(host: &str) -> Result<&str> {
    match host {
        crate::runtime_config::CLAUDE_HOST | crate::runtime_config::CODEX_HOST => Ok(host),
        _ => bail!("invalid capture host '{host}'"),
    }
}

fn legacy_event_id(id: i64) -> String {
    format!("legacy-pending-{id}")
}

fn legacy_capture_content(row: &LegacyPendingRow) -> String {
    let git_branch = row.cwd.as_deref().and_then(db::detect_git_branch);
    serde_json::json!({
        "summary": format!("Recovered legacy {} event", row.tool_name),
        "event_type": "legacy_pending_observation",
        "detail": format!(
            "Recovered from pending_observations id={} created_at_epoch={}",
            row.id, row.created_at_epoch
        ),
        "files": serde_json::Value::Null,
        "exit_code": serde_json::Value::Null,
        "tool_name": row.tool_name,
        "tool_input": parse_jsonish(row.tool_input.as_deref()),
        "tool_response": parse_jsonish(row.tool_response.as_deref()),
        "git_branch": git_branch,
    })
    .to_string()
}

fn parse_jsonish(value: Option<&str>) -> serde_json::Value {
    match value {
        Some(value) => serde_json::from_str(value)
            .unwrap_or_else(|_| serde_json::Value::String(value.to_string())),
        None => serde_json::Value::Null,
    }
}

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

    #[test]
    fn unknown_host_count_is_queryable() -> Result<()> {
        let conn = Connection::open_in_memory()?;
        conn.execute_batch(
            "CREATE TABLE pending_observations (
                id INTEGER PRIMARY KEY,
                host TEXT NOT NULL,
                session_id TEXT NOT NULL,
                project TEXT NOT NULL,
                tool_name TEXT NOT NULL,
                tool_input TEXT,
                tool_response TEXT,
                cwd TEXT,
                created_at_epoch INTEGER NOT NULL,
                updated_at_epoch INTEGER NOT NULL,
                status TEXT NOT NULL,
                attempt_count INTEGER NOT NULL,
                next_retry_epoch INTEGER,
                last_error TEXT,
                lease_owner TEXT,
                lease_expires_epoch INTEGER
            );",
        )?;
        conn.execute(
            "INSERT INTO pending_observations
             (host, session_id, project, tool_name, created_at_epoch, updated_at_epoch, status, attempt_count)
             VALUES ('unknown', 's', 'p', 'Edit', 1, 1, 'pending', 0)",
            [],
        )?;

        assert_eq!(count_legacy_migration_candidates(&conn, Some("p"), 10)?, 1);
        assert_eq!(
            count_legacy_migration_candidates(&conn, Some("other"), 10)?,
            0
        );
        Ok(())
    }

    #[test]
    fn legacy_event_id_is_stable() {
        assert_eq!(legacy_event_id(42), "legacy-pending-42");
    }
}
