use std::path::{Path, PathBuf};
use std::process::ExitCode;

use anyhow::{bail, Result};
use rusqlite::{params, Connection, OptionalExtension, Row};
use serde::Serialize;
use serde_json::{json, Value as JsonValue};

use crate::db;
use crate::display_safe::display_safe;
use crate::output::CommandReport;
use crate::paths::state::StateLayout;
use crate::profile;
use crate::repo::marker as repo_marker;
use crate::state::compiled::{self, ProjectionTarget};
use crate::state::session::{self as session_state, SessionMode};

const MAX_CHECKPOINT_REFS: usize = 32;
const MAX_LIFELOOP_RECEIPT_REFS: usize = 64;
const MAX_SERIALIZED_CURL_RECORD_BYTES: usize = 32 * 1024;

macro_rules! curl_columns {
    () => {
        "curl_id, thread_id, profile, project_id, locality_id, channel_ref, title, state,
         opened_at_epoch_s, closed_at_epoch_s, open_reason, close_reason, outcome,
         focus_key, focus_title, mode, summary, decisions_json, constraints_json,
         attempts_json, touched_paths_json, commit_refs_json, pr_refs_json,
         item_refs_json, handoff_before_digest, handoff_after_digest,
         checkpoint_refs_json, external_evidence_refs_json, lifeloop_receipt_refs_json,
         diagnostics_json, created_at_epoch_s, updated_at_epoch_s"
    };
}

const SELECT_CURLS_ORDERED_SQL: &str = concat!(
    "SELECT ",
    curl_columns!(),
    " FROM thread_curls
      WHERE profile = ?1 AND locality_id = ?2
      ORDER BY updated_at_epoch_s DESC, opened_at_epoch_s DESC, curl_id ASC"
);

const SELECT_CURL_BY_ID_SQL: &str = concat!(
    "SELECT ",
    curl_columns!(),
    " FROM thread_curls
      WHERE profile = ?1 AND locality_id = ?2 AND curl_id = ?3"
);

const SELECT_CURL_BY_INTERNAL_ID_SQL: &str = concat!(
    "SELECT ",
    curl_columns!(),
    " FROM thread_curls
      WHERE curl_id = ?1"
);

const SEARCH_CURLS_SQL: &str = concat!(
    "SELECT ",
    curl_columns!(),
    " FROM thread_curls
      WHERE profile = ?1 AND locality_id = ?2
        AND (
          instr(lower(title), ?3) > 0
          OR instr(lower(COALESCE(summary, '')), ?3) > 0
          OR instr(lower(COALESCE(focus_key, '')), ?3) > 0
          OR instr(lower(COALESCE(focus_title, '')), ?3) > 0
          OR instr(lower(decisions_json), ?3) > 0
          OR instr(lower(constraints_json), ?3) > 0
          OR instr(lower(attempts_json), ?3) > 0
          OR instr(lower(touched_paths_json), ?3) > 0
          OR instr(lower(commit_refs_json), ?3) > 0
          OR instr(lower(pr_refs_json), ?3) > 0
          OR instr(lower(item_refs_json), ?3) > 0
          OR instr(lower(checkpoint_refs_json), ?3) > 0
          OR instr(lower(external_evidence_refs_json), ?3) > 0
          OR instr(lower(lifeloop_receipt_refs_json), ?3) > 0
          OR instr(lower(diagnostics_json), ?3) > 0
        )
      ORDER BY updated_at_epoch_s DESC, opened_at_epoch_s DESC, curl_id ASC"
);

#[derive(Clone, Copy, Debug, Eq, PartialEq, Serialize)]
#[serde(rename_all = "snake_case")]
pub(crate) enum CurlState {
    Open,
    Closed,
}

impl CurlState {
    fn from_db_str(value: &str) -> Option<Self> {
        match value {
            "open" => Some(Self::Open),
            "closed" => Some(Self::Closed),
            _ => None,
        }
    }

    fn as_str(self) -> &'static str {
        match self {
            Self::Open => "open",
            Self::Closed => "closed",
        }
    }
}

#[derive(Clone, Debug, Serialize)]
pub(crate) struct CurlRecordView {
    curl_id: String,
    thread_id: String,
    profile: String,
    project_id: String,
    locality_id: String,
    channel_ref: String,
    title: String,
    state: CurlState,
    opened_at_epoch_s: u64,
    #[serde(skip_serializing_if = "Option::is_none")]
    closed_at_epoch_s: Option<u64>,
    #[serde(skip_serializing_if = "Option::is_none")]
    open_reason: Option<String>,
    #[serde(skip_serializing_if = "Option::is_none")]
    close_reason: Option<String>,
    #[serde(skip_serializing_if = "Option::is_none")]
    outcome: Option<String>,
    #[serde(skip_serializing_if = "Option::is_none")]
    focus_key: Option<String>,
    #[serde(skip_serializing_if = "Option::is_none")]
    focus_title: Option<String>,
    #[serde(skip_serializing_if = "Option::is_none")]
    mode: Option<String>,
    #[serde(skip_serializing_if = "Option::is_none")]
    summary: Option<String>,
    decisions: JsonValue,
    constraints: JsonValue,
    attempts: JsonValue,
    touched_paths: JsonValue,
    commit_refs: JsonValue,
    pr_refs: JsonValue,
    item_refs: JsonValue,
    #[serde(skip_serializing_if = "Option::is_none")]
    handoff_before_digest: Option<String>,
    #[serde(skip_serializing_if = "Option::is_none")]
    handoff_after_digest: Option<String>,
    checkpoint_refs: JsonValue,
    external_evidence_refs: JsonValue,
    lifeloop_receipt_refs: JsonValue,
    diagnostics: JsonValue,
    created_at_epoch_s: u64,
    updated_at_epoch_s: u64,
}

#[derive(Clone, Debug, Serialize)]
struct CurlListItem {
    curl_id: String,
    thread_id: String,
    title: String,
    state: CurlState,
    opened_at_epoch_s: u64,
    #[serde(skip_serializing_if = "Option::is_none")]
    closed_at_epoch_s: Option<u64>,
    #[serde(skip_serializing_if = "Option::is_none")]
    outcome: Option<String>,
    #[serde(skip_serializing_if = "Option::is_none")]
    focus_key: Option<String>,
    #[serde(skip_serializing_if = "Option::is_none")]
    focus_title: Option<String>,
    updated_at_epoch_s: u64,
}

#[derive(Clone, Debug)]
struct CurlRecord {
    view: CurlRecordView,
}

#[derive(Clone, Debug, Serialize)]
struct CurlEventView {
    id: i64,
    curl_id: String,
    event_kind: String,
    at_epoch_s: u64,
    #[serde(skip_serializing_if = "Option::is_none")]
    session_id: Option<String>,
    metadata: JsonValue,
}

#[derive(Serialize)]
pub(crate) struct CurlListReport {
    command: &'static str,
    ok: bool,
    path: String,
    profile: String,
    project_id: String,
    locality_id: String,
    store: String,
    channel_ref: String,
    curls: Vec<CurlListItem>,
}

#[derive(Serialize)]
pub(crate) struct CurlShowReport {
    command: &'static str,
    ok: bool,
    path: String,
    profile: String,
    project_id: String,
    locality_id: String,
    store: String,
    channel_ref: String,
    curl: CurlRecordView,
    events: Vec<CurlEventView>,
}

#[derive(Serialize)]
pub(crate) struct CurlSearchReport {
    command: &'static str,
    ok: bool,
    path: String,
    profile: String,
    project_id: String,
    locality_id: String,
    store: String,
    channel_ref: String,
    query: String,
    total_matches: usize,
    curls: Vec<CurlListItem>,
}

#[derive(Serialize)]
pub(crate) struct CurlExportReport {
    command: &'static str,
    ok: bool,
    path: String,
    profile: String,
    project_id: String,
    locality_id: String,
    store: String,
    format: &'static str,
    records: Vec<CurlRecordView>,
}

#[derive(Clone)]
struct CurlContext {
    repo_root: PathBuf,
    layout: StateLayout,
    profile: String,
    locality_id: String,
    db_path: PathBuf,
    channel_ref: String,
}

impl CommandReport for CurlListReport {
    fn exit_code(&self) -> ExitCode {
        ExitCode::SUCCESS
    }

    fn render_text(&self) {
        if self.curls.is_empty() {
            println!(
                "No curls recorded for profile `{}` in linked project `{}`.",
                self.profile, self.project_id
            );
            return;
        }

        println!(
            "Curls for profile `{}` in linked project `{}`:",
            self.profile, self.project_id
        );
        for curl in &self.curls {
            let suffix = curl
                .outcome
                .as_deref()
                .map(|outcome| format!("{}/{}", curl.state.as_str(), outcome))
                .unwrap_or_else(|| curl.state.as_str().to_owned());
            println!("- {} `{}` [{}]", curl.curl_id, curl.title, suffix);
        }
    }
}

impl CommandReport for CurlShowReport {
    fn exit_code(&self) -> ExitCode {
        ExitCode::SUCCESS
    }

    fn render_text(&self) {
        println!(
            "Curl {} `{}` is {}.",
            self.curl.curl_id,
            self.curl.title,
            self.curl.state.as_str()
        );
        for event in self.events.iter().take(8) {
            println!("- {} at {}", event.event_kind, event.at_epoch_s);
        }
    }
}

impl CommandReport for CurlSearchReport {
    fn exit_code(&self) -> ExitCode {
        ExitCode::SUCCESS
    }

    fn render_text(&self) {
        if self.curls.is_empty() {
            println!("No curls matched `{}`.", self.query);
            return;
        }
        println!("{} curl(s) matched `{}`:", self.total_matches, self.query);
        for curl in &self.curls {
            println!(
                "- {} `{}` [{}]",
                curl.curl_id,
                curl.title,
                curl.state.as_str()
            );
        }
    }
}

impl CommandReport for CurlExportReport {
    fn exit_code(&self) -> ExitCode {
        ExitCode::SUCCESS
    }

    fn render_text(&self) {
        for record in &self.records {
            println!(
                "{}",
                serde_json::to_string(record).expect("curl record should serialize")
            );
        }
    }
}

pub(crate) fn list(repo_root: &Path, explicit_profile: Option<&str>) -> Result<CurlListReport> {
    let context = resolve_context(repo_root, explicit_profile, "list")?;
    let records = match open_existing_state_db(&context)? {
        Some(db) => read_curls(db.conn(), &context)?,
        None => Vec::new(),
    };
    Ok(CurlListReport {
        command: "curl-list",
        ok: true,
        path: context.repo_root.display().to_string(),
        profile: context.profile,
        project_id: context.locality_id.clone(),
        locality_id: context.locality_id,
        store: context.db_path.display().to_string(),
        channel_ref: context.channel_ref,
        curls: records.into_iter().map(list_item).collect(),
    })
}

pub(crate) fn show(
    repo_root: &Path,
    explicit_profile: Option<&str>,
    curl_id: &str,
) -> Result<CurlShowReport> {
    let context = resolve_context(repo_root, explicit_profile, "show")?;
    let Some(db) = open_existing_state_db(&context)? else {
        bail!("curl `{}` was not found", display_safe(curl_id));
    };
    let record = read_curl(db.conn(), &context, curl_id)?
        .ok_or_else(|| anyhow::anyhow!("curl `{}` was not found", display_safe(curl_id)))?;
    let events = read_events(db.conn(), curl_id)?;
    Ok(CurlShowReport {
        command: "curl-show",
        ok: true,
        path: context.repo_root.display().to_string(),
        profile: context.profile,
        project_id: context.locality_id.clone(),
        locality_id: context.locality_id,
        store: context.db_path.display().to_string(),
        channel_ref: context.channel_ref,
        curl: record.view,
        events,
    })
}

pub(crate) fn search(
    repo_root: &Path,
    explicit_profile: Option<&str>,
    query: &str,
) -> Result<CurlSearchReport> {
    let normalized = query.trim();
    if normalized.is_empty() {
        bail!("curl search query cannot be empty");
    }
    let context = resolve_context(repo_root, explicit_profile, "search")?;
    let records = match open_existing_state_db(&context)? {
        Some(db) => search_curls(db.conn(), &context, normalized)?,
        None => Vec::new(),
    };
    let total_matches = records.len();
    Ok(CurlSearchReport {
        command: "curl-search",
        ok: true,
        path: context.repo_root.display().to_string(),
        profile: context.profile,
        project_id: context.locality_id.clone(),
        locality_id: context.locality_id,
        store: context.db_path.display().to_string(),
        channel_ref: context.channel_ref,
        query: normalized.to_owned(),
        total_matches,
        curls: records.into_iter().map(list_item).collect(),
    })
}

pub(crate) fn export_jsonl(
    repo_root: &Path,
    explicit_profile: Option<&str>,
) -> Result<CurlExportReport> {
    let context = resolve_context(repo_root, explicit_profile, "export")?;
    let records = match open_existing_state_db(&context)? {
        Some(db) => read_curls(db.conn(), &context)?,
        None => Vec::new(),
    };
    Ok(CurlExportReport {
        command: "curl-export",
        ok: true,
        path: context.repo_root.display().to_string(),
        profile: context.profile,
        project_id: context.locality_id.clone(),
        locality_id: context.locality_id,
        store: context.db_path.display().to_string(),
        format: "jsonl",
        records: records.into_iter().map(|record| record.view).collect(),
    })
}

pub(crate) fn ensure_open_for_start_on_db(
    conn: &Connection,
    layout: &StateLayout,
    profile: &str,
    locality_id: &str,
    thread_id: &str,
    session_id: Option<&str>,
    mode: SessionMode,
    stale_reset: bool,
    now_epoch_s: u64,
) -> Result<()> {
    let channel_ref = channel_ref(layout);
    let handoff_before_digest = current_handoff_digest_on_db(conn)?;
    if let Some(open_curl_id) =
        find_open_curl_id(conn, profile, locality_id, &channel_ref, thread_id)?
    {
        if !stale_reset {
            return Ok(());
        }
        close_stale_open_curl_on_db(
            conn,
            &open_curl_id,
            session_id,
            handoff_before_digest.as_deref(),
            now_epoch_s,
        )?;
    }

    let curl_id = format!("curl_{}", crate::ids::new_ulid());
    let title = derive_start_title(conn, &curl_id)?;
    conn.execute(
        "INSERT INTO thread_curls
            (curl_id, thread_id, profile, project_id, locality_id, channel_ref, title,
             state, opened_at_epoch_s, open_reason, mode, handoff_before_digest,
             created_at_epoch_s, updated_at_epoch_s)
         VALUES (?1, ?2, ?3, ?4, ?4, ?5, ?6, 'open', ?7, 'start_activate',
                 ?8, ?9, ?7, ?7)",
        params![
            curl_id,
            thread_id,
            profile,
            locality_id,
            channel_ref,
            title,
            now_epoch_s,
            mode.as_str(),
            handoff_before_digest.as_deref(),
        ],
    )?;
    enforce_record_cap(conn, &curl_id)?;
    record_event_on_db(
        conn,
        &curl_id,
        "opened",
        now_epoch_s,
        session_id,
        json!({ "reason": "start_activate" }),
    )
}

pub(crate) fn record_checkpoint(
    repo_root: &Path,
    explicit_profile: Option<&str>,
    summary: &str,
    checkpoint_ref: &str,
) -> Result<()> {
    let context = resolve_context(repo_root, explicit_profile, "checkpoint")?;
    let Some(db) = open_existing_state_db(&context)? else {
        return Ok(());
    };
    let Some(session) = db::session::read(db.conn())? else {
        return Ok(());
    };
    let Some(thread_id) = session.thread_id.as_deref() else {
        return Ok(());
    };
    let now = session_state::now_epoch_s()?;
    record_checkpoint_on_db(
        db.conn(),
        &context,
        thread_id,
        session.session_id.as_deref(),
        summary,
        checkpoint_ref,
        now,
    )
}

pub(crate) fn record_handover(
    repo_root: &Path,
    explicit_profile: Option<&str>,
    summary: &str,
    outcome: &str,
    handoff_digest: &str,
) -> Result<()> {
    let context = resolve_context(repo_root, explicit_profile, "handover")?;
    let Some(db) = open_existing_state_db(&context)? else {
        return Ok(());
    };
    let Some(session) = db::session::read(db.conn())? else {
        return Ok(());
    };
    let Some(thread_id) = session.thread_id.as_deref() else {
        return Ok(());
    };
    let now = session_state::now_epoch_s()?;
    close_open_curl_on_db(
        db.conn(),
        &context,
        thread_id,
        session.session_id.as_deref(),
        summary,
        outcome,
        handoff_digest,
        now,
    )
}

pub(crate) fn record_renewal_prepared_on_db(
    conn: &Connection,
    layout: &StateLayout,
    profile: &str,
    locality_id: &str,
    thread_id: &str,
    previous_session_id: &str,
    renewal_lease_id: &str,
    lifeloop_receipt_id: Option<&str>,
    now_epoch_s: u64,
) -> Result<()> {
    let context = CurlContext {
        repo_root: PathBuf::new(),
        layout: layout.clone(),
        profile: profile.to_owned(),
        locality_id: locality_id.to_owned(),
        db_path: layout.state_db_path(),
        channel_ref: channel_ref(layout),
    };
    record_renewal_event_on_db(
        conn,
        &context,
        thread_id,
        previous_session_id,
        "renewal_prepared",
        renewal_lease_id,
        lifeloop_receipt_id,
        now_epoch_s,
    )
}

pub(crate) fn record_renewal_fulfilled_on_db(
    conn: &Connection,
    layout: &StateLayout,
    profile: &str,
    locality_id: &str,
    thread_id: &str,
    session_id: &str,
    renewal_lease_id: &str,
    lifeloop_receipt_id: Option<&str>,
    now_epoch_s: u64,
) -> Result<()> {
    let context = CurlContext {
        repo_root: PathBuf::new(),
        layout: layout.clone(),
        profile: profile.to_owned(),
        locality_id: locality_id.to_owned(),
        db_path: layout.state_db_path(),
        channel_ref: channel_ref(layout),
    };
    record_renewal_event_on_db(
        conn,
        &context,
        thread_id,
        session_id,
        "renewal_fulfilled",
        renewal_lease_id,
        lifeloop_receipt_id,
        now_epoch_s,
    )
}

fn resolve_context(
    repo_root: &Path,
    explicit_profile: Option<&str>,
    command: &str,
) -> Result<CurlContext> {
    let profile = profile::resolve(explicit_profile)?;
    let layout = StateLayout::resolve(repo_root, profile.clone())?;
    if !layout.profile_root().is_dir() {
        bail!(
            "profile `{}` does not exist at {}; bootstrap it with `ccd attach --path {}` before using `ccd curl {}`",
            layout.profile(),
            layout.profile_root().display(),
            repo_root.display(),
            command
        );
    }
    let Some(marker) = repo_marker::load(repo_root)? else {
        bail!(
            "repo is not linked: {} is missing; run `ccd attach --path {}` or `ccd link --path {}` first",
            repo_root.join(repo_marker::MARKER_FILE).display(),
            repo_root.display(),
            repo_root.display()
        );
    };
    Ok(CurlContext {
        repo_root: repo_root.to_path_buf(),
        db_path: layout.state_db_path(),
        channel_ref: channel_ref(&layout),
        profile: profile.to_string(),
        locality_id: marker.locality_id,
        layout,
    })
}

fn open_existing_state_db(context: &CurlContext) -> Result<Option<db::StateDb>> {
    if !context.db_path.exists() {
        return Ok(None);
    }
    db::StateDb::try_open_for_layout(&context.layout)
}

fn derive_start_title(conn: &Connection, curl_id: &str) -> Result<String> {
    let title = conn
        .query_row("SELECT title FROM handoff WHERE id = 1", [], |row| {
            row.get::<_, String>(0)
        })
        .optional()?
        .and_then(|title| normalize_title_candidate(&title))
        .unwrap_or_else(|| format!("Curl {}", short_id(curl_id)));
    Ok(truncate_utf8_bytes(&title, 120))
}

fn normalize_title_candidate(title: &str) -> Option<String> {
    let mut trimmed = title.trim();
    for prefix in ["Next Session:", "Next:"] {
        if let Some(rest) = trimmed.strip_prefix(prefix) {
            trimmed = rest.trim();
            break;
        }
    }
    (!trimmed.is_empty()).then(|| trimmed.to_owned())
}

fn short_id(id: &str) -> String {
    id.rsplit_once('_')
        .map(|(_, suffix)| suffix)
        .unwrap_or(id)
        .chars()
        .take(8)
        .collect()
}

fn read_curls(conn: &Connection, context: &CurlContext) -> Result<Vec<CurlRecord>> {
    let mut stmt = conn.prepare(SELECT_CURLS_ORDERED_SQL)?;
    let rows = stmt.query_map(params![context.profile, context.locality_id], curl_from_row)?;
    rows.collect::<std::result::Result<Vec<_>, _>>()
        .map_err(Into::into)
}

fn read_curl(
    conn: &Connection,
    context: &CurlContext,
    curl_id: &str,
) -> Result<Option<CurlRecord>> {
    conn.query_row(
        SELECT_CURL_BY_ID_SQL,
        params![context.profile, context.locality_id, curl_id],
        curl_from_row,
    )
    .optional()
    .map_err(Into::into)
}

fn search_curls(conn: &Connection, context: &CurlContext, query: &str) -> Result<Vec<CurlRecord>> {
    let pattern = query.to_lowercase();
    let mut stmt = conn.prepare(SEARCH_CURLS_SQL)?;
    let rows = stmt.query_map(
        params![context.profile, context.locality_id, pattern],
        curl_from_row,
    )?;
    rows.collect::<std::result::Result<Vec<_>, _>>()
        .map_err(Into::into)
}

fn read_events(conn: &Connection, curl_id: &str) -> Result<Vec<CurlEventView>> {
    let mut stmt = conn.prepare(
        "SELECT id, curl_id, event_kind, at_epoch_s, session_id, metadata_json
         FROM thread_curl_events
         WHERE curl_id = ?1
         ORDER BY at_epoch_s DESC, id DESC",
    )?;
    let rows = stmt.query_map(params![curl_id], |row| {
        let metadata_json: String = row.get(5)?;
        Ok(CurlEventView {
            id: row.get(0)?,
            curl_id: row.get(1)?,
            event_kind: row.get(2)?,
            at_epoch_s: row.get(3)?,
            session_id: row.get(4)?,
            metadata: parse_json_object_for_row(&metadata_json, 5, "metadata_json")?,
        })
    })?;
    rows.collect::<std::result::Result<Vec<_>, _>>()
        .map_err(Into::into)
}

fn record_checkpoint_on_db(
    conn: &Connection,
    context: &CurlContext,
    thread_id: &str,
    session_id: Option<&str>,
    summary: &str,
    checkpoint_ref: &str,
    now_epoch_s: u64,
) -> Result<()> {
    let Some(curl_id) = find_open_curl_id(
        conn,
        &context.profile,
        &context.locality_id,
        &context.channel_ref,
        thread_id,
    )?
    else {
        return Ok(());
    };
    let refs_json: String = conn.query_row(
        "SELECT checkpoint_refs_json FROM thread_curls WHERE curl_id = ?1",
        params![curl_id],
        |row| row.get(0),
    )?;
    let refs_json = append_bounded_json_string(&refs_json, checkpoint_ref, MAX_CHECKPOINT_REFS)?;
    let summary = truncate_utf8_bytes(summary.trim(), 2048);
    conn.execute(
        "UPDATE thread_curls
         SET summary = CASE
                WHEN summary IS NULL OR summary = '' THEN ?1
                ELSE summary
             END,
             checkpoint_refs_json = ?2,
             updated_at_epoch_s = ?3
        WHERE curl_id = ?4",
        params![summary, refs_json, now_epoch_s, curl_id],
    )?;
    enforce_record_cap(conn, &curl_id)?;
    record_event_on_db(
        conn,
        &curl_id,
        "checkpointed",
        now_epoch_s,
        session_id,
        json!({ "checkpoint_ref": checkpoint_ref }),
    )
}

fn close_open_curl_on_db(
    conn: &Connection,
    context: &CurlContext,
    thread_id: &str,
    session_id: Option<&str>,
    summary: &str,
    outcome: &str,
    handoff_digest: &str,
    now_epoch_s: u64,
) -> Result<()> {
    let Some(curl_id) = find_open_curl_id(
        conn,
        &context.profile,
        &context.locality_id,
        &context.channel_ref,
        thread_id,
    )?
    else {
        return Ok(());
    };
    let summary = truncate_utf8_bytes(summary.trim(), 2048);
    conn.execute(
        "UPDATE thread_curls
         SET state = 'closed',
             closed_at_epoch_s = ?1,
             close_reason = 'handover',
             outcome = ?2,
             summary = CASE
                WHEN ?3 = '' THEN summary
                ELSE ?3
             END,
             handoff_after_digest = ?5,
             updated_at_epoch_s = ?1
         WHERE curl_id = ?4",
        params![now_epoch_s, outcome, summary, curl_id, handoff_digest],
    )?;
    enforce_record_cap(conn, &curl_id)?;
    record_event_on_db(
        conn,
        &curl_id,
        "closed",
        now_epoch_s,
        session_id,
        json!({ "reason": "handover", "outcome": outcome }),
    )
}

fn record_renewal_event_on_db(
    conn: &Connection,
    context: &CurlContext,
    thread_id: &str,
    session_id: &str,
    event_kind: &'static str,
    renewal_lease_id: &str,
    lifeloop_receipt_id: Option<&str>,
    now_epoch_s: u64,
) -> Result<()> {
    let Some(curl_id) = find_open_curl_id(
        conn,
        &context.profile,
        &context.locality_id,
        &context.channel_ref,
        thread_id,
    )?
    else {
        return Ok(());
    };
    let compact_receipt_id =
        lifeloop_receipt_id.map(|receipt_id| truncate_utf8_bytes(receipt_id, 160));
    if let Some(receipt_id) = compact_receipt_id.as_deref() {
        let refs_json: String = conn.query_row(
            "SELECT lifeloop_receipt_refs_json FROM thread_curls WHERE curl_id = ?1",
            params![curl_id],
            |row| row.get(0),
        )?;
        let refs_json =
            append_bounded_json_string(&refs_json, receipt_id, MAX_LIFELOOP_RECEIPT_REFS)?;
        conn.execute(
            "UPDATE thread_curls
             SET lifeloop_receipt_refs_json = ?1, updated_at_epoch_s = ?2
             WHERE curl_id = ?3",
            params![refs_json, now_epoch_s, curl_id],
        )?;
    } else {
        conn.execute(
            "UPDATE thread_curls SET updated_at_epoch_s = ?1 WHERE curl_id = ?2",
            params![now_epoch_s, curl_id],
        )?;
    }
    enforce_record_cap(conn, &curl_id)?;
    record_event_on_db(
        conn,
        &curl_id,
        event_kind,
        now_epoch_s,
        Some(session_id),
        json!({
            "renewal_lease_id": truncate_utf8_bytes(renewal_lease_id, 160),
            "lifeloop_receipt_id": compact_receipt_id,
        }),
    )
}

fn find_open_curl_id(
    conn: &Connection,
    profile: &str,
    locality_id: &str,
    channel_ref: &str,
    thread_id: &str,
) -> Result<Option<String>> {
    conn.query_row(
        "SELECT curl_id FROM thread_curls
         WHERE profile = ?1
           AND locality_id = ?2
           AND channel_ref = ?3
           AND thread_id = ?4
           AND state = 'open'",
        params![profile, locality_id, channel_ref, thread_id],
        |row| row.get(0),
    )
    .optional()
    .map_err(Into::into)
}

fn record_event_on_db(
    conn: &Connection,
    curl_id: &str,
    event_kind: &str,
    at_epoch_s: u64,
    session_id: Option<&str>,
    metadata: JsonValue,
) -> Result<()> {
    conn.execute(
        "INSERT INTO thread_curl_events
            (curl_id, event_kind, at_epoch_s, session_id, metadata_json)
         VALUES (?1, ?2, ?3, ?4, ?5)",
        params![
            curl_id,
            event_kind,
            at_epoch_s,
            session_id,
            serde_json::to_string(&metadata)?,
        ],
    )?;
    Ok(())
}

fn append_bounded_json_string(raw: &str, item: &str, max_items: usize) -> Result<String> {
    let mut value: JsonValue = serde_json::from_str(raw)?;
    let Some(items) = value.as_array_mut() else {
        bail!("curl reference field must contain a JSON array");
    };
    let bounded_item = truncate_utf8_bytes(item, 160);
    if !items
        .iter()
        .any(|existing| existing.as_str() == Some(bounded_item.as_str()))
    {
        items.push(JsonValue::String(bounded_item));
    }
    if items.len() > max_items {
        let trim_count = items.len() - max_items;
        items.drain(0..trim_count);
    }
    Ok(serde_json::to_string(&value)?)
}

fn current_handoff_digest_on_db(conn: &Connection) -> Result<Option<String>> {
    db::handoff::read(conn).map(|handoff| {
        handoff.map(|state| {
            compiled::compute_handoff_digest_for_runtime_handoff(&state, ProjectionTarget::Session)
        })
    })
}

fn close_stale_open_curl_on_db(
    conn: &Connection,
    curl_id: &str,
    session_id: Option<&str>,
    handoff_digest: Option<&str>,
    now_epoch_s: u64,
) -> Result<()> {
    conn.execute(
        "UPDATE thread_curls
         SET state = 'closed',
             closed_at_epoch_s = ?1,
             close_reason = 'stale_recovered',
             outcome = 'continued',
             handoff_after_digest = COALESCE(?2, handoff_after_digest),
             updated_at_epoch_s = ?1
         WHERE curl_id = ?3",
        params![now_epoch_s, handoff_digest, curl_id],
    )?;
    enforce_record_cap(conn, curl_id)?;
    record_event_on_db(
        conn,
        curl_id,
        "closed",
        now_epoch_s,
        session_id,
        json!({ "reason": "stale_recovered", "outcome": "continued" }),
    )
}

fn enforce_record_cap(conn: &Connection, curl_id: &str) -> Result<()> {
    if curl_record_len(conn, curl_id)? <= MAX_SERIALIZED_CURL_RECORD_BYTES {
        return Ok(());
    }

    for column in [
        "diagnostics_json",
        "attempts_json",
        "constraints_json",
        "decisions_json",
        "external_evidence_refs_json",
        "lifeloop_receipt_refs_json",
        "checkpoint_refs_json",
        "touched_paths_json",
        "item_refs_json",
        "pr_refs_json",
        "commit_refs_json",
    ] {
        while curl_record_len(conn, curl_id)? > MAX_SERIALIZED_CURL_RECORD_BYTES
            && trim_json_array_head(conn, curl_id, column)?
        {}
    }

    if curl_record_len(conn, curl_id)? > MAX_SERIALIZED_CURL_RECORD_BYTES {
        conn.execute(
            "UPDATE thread_curls
             SET summary = NULL
             WHERE curl_id = ?1",
            params![curl_id],
        )?;
    }

    let len = curl_record_len(conn, curl_id)?;
    if len > MAX_SERIALIZED_CURL_RECORD_BYTES {
        bail!(
            "thread curl `{}` exceeds the {}-byte serialized record cap after trimming ({len} bytes)",
            display_safe(curl_id),
            MAX_SERIALIZED_CURL_RECORD_BYTES
        );
    }
    Ok(())
}

fn curl_record_len(conn: &Connection, curl_id: &str) -> Result<usize> {
    let record = conn
        .query_row(
            SELECT_CURL_BY_INTERNAL_ID_SQL,
            params![curl_id],
            curl_from_row,
        )
        .optional()?
        .ok_or_else(|| anyhow::anyhow!("curl `{}` was not found", display_safe(curl_id)))?;
    Ok(serde_json::to_vec(&record.view)?.len())
}

fn trim_json_array_head(conn: &Connection, curl_id: &str, column: &'static str) -> Result<bool> {
    let raw: String = conn.query_row(
        &format!("SELECT {column} FROM thread_curls WHERE curl_id = ?1"),
        params![curl_id],
        |row| row.get(0),
    )?;
    let mut value: JsonValue = serde_json::from_str(&raw)?;
    let Some(items) = value.as_array_mut() else {
        bail!("{column} must contain a JSON array");
    };
    if items.is_empty() {
        return Ok(false);
    }
    items.remove(0);
    conn.execute(
        &format!("UPDATE thread_curls SET {column} = ?1 WHERE curl_id = ?2"),
        params![serde_json::to_string(&value)?, curl_id],
    )?;
    Ok(true)
}

fn curl_from_row(row: &Row<'_>) -> rusqlite::Result<CurlRecord> {
    let state_str: String = row.get(7)?;
    let state = CurlState::from_db_str(&state_str).ok_or_else(|| {
        rusqlite::Error::FromSqlConversionFailure(
            7,
            rusqlite::types::Type::Text,
            Box::new(std::io::Error::new(
                std::io::ErrorKind::InvalidData,
                format!("invalid curl state `{state_str}`"),
            )),
        )
    })?;

    Ok(CurlRecord {
        view: CurlRecordView {
            curl_id: row.get(0)?,
            thread_id: row.get(1)?,
            profile: row.get(2)?,
            project_id: row.get(3)?,
            locality_id: row.get(4)?,
            channel_ref: row.get(5)?,
            title: row.get(6)?,
            state,
            opened_at_epoch_s: row.get(8)?,
            closed_at_epoch_s: row.get(9)?,
            open_reason: row.get(10)?,
            close_reason: row.get(11)?,
            outcome: row.get(12)?,
            focus_key: row.get(13)?,
            focus_title: row.get(14)?,
            mode: row.get(15)?,
            summary: row.get(16)?,
            decisions: parse_json_array(row, 17, "decisions_json")?,
            constraints: parse_json_array(row, 18, "constraints_json")?,
            attempts: parse_json_array(row, 19, "attempts_json")?,
            touched_paths: parse_json_array(row, 20, "touched_paths_json")?,
            commit_refs: parse_json_array(row, 21, "commit_refs_json")?,
            pr_refs: parse_json_array(row, 22, "pr_refs_json")?,
            item_refs: parse_json_array(row, 23, "item_refs_json")?,
            handoff_before_digest: row.get(24)?,
            handoff_after_digest: row.get(25)?,
            checkpoint_refs: parse_json_array(row, 26, "checkpoint_refs_json")?,
            external_evidence_refs: parse_json_array(row, 27, "external_evidence_refs_json")?,
            lifeloop_receipt_refs: parse_json_array(row, 28, "lifeloop_receipt_refs_json")?,
            diagnostics: parse_json_array(row, 29, "diagnostics_json")?,
            created_at_epoch_s: row.get(30)?,
            updated_at_epoch_s: row.get(31)?,
        },
    })
}

fn parse_json_array(
    row: &Row<'_>,
    index: usize,
    column: &'static str,
) -> rusqlite::Result<JsonValue> {
    let raw: String = row.get(index)?;
    let value: JsonValue = serde_json::from_str(&raw).map_err(|error| {
        rusqlite::Error::FromSqlConversionFailure(
            index,
            rusqlite::types::Type::Text,
            Box::new(error),
        )
    })?;
    if value.is_array() {
        Ok(value)
    } else {
        Err(rusqlite::Error::FromSqlConversionFailure(
            index,
            rusqlite::types::Type::Text,
            Box::new(std::io::Error::new(
                std::io::ErrorKind::InvalidData,
                format!("{column} must contain a JSON array"),
            )),
        ))
    }
}

fn parse_json_object_for_row(
    raw: &str,
    index: usize,
    column: &'static str,
) -> rusqlite::Result<JsonValue> {
    let value: JsonValue = serde_json::from_str(raw).map_err(|error| {
        rusqlite::Error::FromSqlConversionFailure(
            index,
            rusqlite::types::Type::Text,
            Box::new(error),
        )
    })?;
    if value.is_object() {
        Ok(value)
    } else {
        Err(rusqlite::Error::FromSqlConversionFailure(
            index,
            rusqlite::types::Type::Text,
            Box::new(std::io::Error::new(
                std::io::ErrorKind::InvalidData,
                format!("{column} must contain a JSON object"),
            )),
        ))
    }
}

fn list_item(record: CurlRecord) -> CurlListItem {
    let view = record.view;
    CurlListItem {
        curl_id: view.curl_id,
        thread_id: view.thread_id,
        title: view.title,
        state: view.state,
        opened_at_epoch_s: view.opened_at_epoch_s,
        closed_at_epoch_s: view.closed_at_epoch_s,
        outcome: view.outcome,
        focus_key: view.focus_key,
        focus_title: view.focus_title,
        updated_at_epoch_s: view.updated_at_epoch_s,
    }
}

fn channel_ref(layout: &StateLayout) -> String {
    format!(
        "profile:{}|state:{}",
        layout.profile(),
        layout.clone_profile_root().display()
    )
}

fn truncate_utf8_bytes(value: &str, max_bytes: usize) -> String {
    if value.len() <= max_bytes {
        return value.to_owned();
    }
    let mut end = max_bytes;
    while !value.is_char_boundary(end) {
        end -= 1;
    }
    value[..end].to_owned()
}

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

    fn test_conn() -> Connection {
        let conn = Connection::open_in_memory().unwrap();
        crate::db::schema::initialize(&conn).unwrap();
        conn
    }

    fn insert_curl(conn: &Connection, title: &str, summary: Option<&str>, diagnostics_json: &str) {
        conn.execute(
            "INSERT INTO thread_curls
                (curl_id, thread_id, profile, project_id, locality_id, channel_ref, title,
                 state, opened_at_epoch_s, open_reason, summary, diagnostics_json,
                 created_at_epoch_s, updated_at_epoch_s)
             VALUES
                ('curl_test', 'thread_test', 'main', 'project_test', 'project_test',
                 'profile:main|state:/tmp/test', ?1, 'open', 1, 'start_activate',
                 ?2, ?3, 1, 1)",
            params![title, summary, diagnostics_json],
        )
        .unwrap();
    }

    #[test]
    fn append_bounded_json_string_deduplicates_after_truncation() {
        let long_ref = "r".repeat(240);
        let once = append_bounded_json_string("[]", &long_ref, MAX_LIFELOOP_RECEIPT_REFS).unwrap();
        let twice =
            append_bounded_json_string(&once, &long_ref, MAX_LIFELOOP_RECEIPT_REFS).unwrap();
        let refs: Vec<String> = serde_json::from_str(&twice).unwrap();

        assert_eq!(refs.len(), 1);
        assert_eq!(refs[0].len(), 160);
    }

    #[test]
    fn enforce_record_cap_trims_bounded_arrays_before_summary() {
        let conn = test_conn();
        let diagnostics = serde_json::to_string(
            &(0..100)
                .map(|index| format!("diagnostic-{index}-{}", "x".repeat(512)))
                .collect::<Vec<_>>(),
        )
        .unwrap();
        insert_curl(
            &conn,
            "Compact curl",
            Some("summary survives"),
            &diagnostics,
        );

        enforce_record_cap(&conn, "curl_test").unwrap();

        assert!(curl_record_len(&conn, "curl_test").unwrap() <= MAX_SERIALIZED_CURL_RECORD_BYTES);
        let (summary, diagnostics_json): (Option<String>, String) = conn
            .query_row(
                "SELECT summary, diagnostics_json FROM thread_curls WHERE curl_id = 'curl_test'",
                [],
                |row| Ok((row.get(0)?, row.get(1)?)),
            )
            .unwrap();
        let trimmed: Vec<String> = serde_json::from_str(&diagnostics_json).unwrap();
        assert_eq!(summary.as_deref(), Some("summary survives"));
        assert!(trimmed.len() < 100);
    }

    #[test]
    fn enforce_record_cap_refuses_when_fixed_fields_alone_are_too_large() {
        let conn = test_conn();
        insert_curl(&conn, &"t".repeat(40_000), None, "[]");

        let error = enforce_record_cap(&conn, "curl_test")
            .expect_err("oversized fixed fields should fail closed");
        assert!(error
            .to_string()
            .contains("serialized record cap after trimming"));
    }
}
