use std::collections::{HashMap, HashSet};
use std::time::{SystemTime, UNIX_EPOCH};

use rusqlite::{named_params, Connection, ErrorCode, OptionalExtension, Row};
use serde::Serialize;
use serde_json::{Map as JsonMap, Number, Value};

use crate::error::GaalError;
use crate::model::{Fact, HandoffRecord};

const HERMES_ALIAS_SCHEME: &str = "hermes-fnv64-base32-8-v1";
const MAX_ALIAS_COLLISION_RETRIES: u32 = 1024;

/// Database-level session row, flattened to only SQLite-backed fields.
#[derive(Debug, Clone)]
pub struct SessionRow {
    pub id: String,
    pub engine: String,
    pub model: Option<String>,
    pub cwd: Option<String>,
    pub started_at: String,
    pub ended_at: Option<String>,
    pub exit_signal: Option<String>,
    pub last_event_at: Option<String>,
    pub parent_id: Option<String>,
    pub session_type: String,
    pub jsonl_path: String,
    pub total_input_tokens: i64,
    pub total_output_tokens: i64,
    pub cache_read_tokens: i64,
    pub cache_creation_tokens: i64,
    pub reasoning_tokens: i64,
    pub total_tools: i64,
    pub total_turns: i64,
    pub peak_context: i64,
    pub last_indexed_offset: i64,
    /// Subagent type from Agent tool_use input (e.g. "gsd-heavy", "Explore").
    pub subagent_type: Option<String>,
    /// Root-level session summary (e.g. Gemini `summary` field), used as headline fallback.
    pub gemini_summary: Option<String>,
}

/// Filters for listing sessions.
#[derive(Debug, Clone, Default)]
pub struct ListFilter {
    pub engine: Option<String>,
    pub session_type: Option<String>,
    pub since: Option<String>,
    pub before: Option<String>,
    pub cwd: Option<String>,
    pub tag: Option<String>,
    pub sort_by: Option<String>,
    pub limit: Option<i64>,
    pub include_subagents: bool,
    pub subagent_type: Option<String>,
}

/// Filters for source-rendered activity slices.
#[derive(Debug, Clone, Default)]
pub struct ActivityFilter {
    pub engine: Option<String>,
    pub since: String,
    pub before: String,
    pub cwd: Option<String>,
    pub include_subagents: bool,
    pub limit: Option<i64>,
}

/// Supported fact types for `who` queries and fact filtering.
#[derive(Debug, Clone, PartialEq, Eq, Hash)]
pub enum FactType {
    FileRead,
    FileWrite,
    Command,
    Error,
    GitOp,
    UserPrompt,
    AssistantReply,
    TaskSpawn,
}

impl FactType {
    fn as_str(&self) -> &'static str {
        match self {
            Self::FileRead => "file_read",
            Self::FileWrite => "file_write",
            Self::Command => "command",
            Self::Error => "error",
            Self::GitOp => "git_op",
            Self::UserPrompt => "user_prompt",
            Self::AssistantReply => "assistant_reply",
            Self::TaskSpawn => "task_spawn",
        }
    }
}

impl std::fmt::Display for FactType {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        write!(f, "{}", self.as_str())
    }
}

/// Filters for `query_who`.
#[derive(Debug, Clone, Default)]
pub struct WhoFilter {
    pub fact_types: Vec<FactType>,
    pub subject_pattern: Option<String>,
    pub since: Option<String>,
    pub before: Option<String>,
    pub cwd: Option<String>,
    pub engine: Option<String>,
    pub tag: Option<String>,
    pub failed_only: bool,
    pub limit: Option<i64>,
}

/// A single match row for `query_who`.
#[derive(Debug, Clone)]
pub struct WhoResult {
    pub session_id: String,
    pub engine: String,
    pub ts: String,
    pub fact_type: String,
    pub subject: Option<String>,
    pub detail: Option<String>,
    pub session_headline: Option<String>,
    pub session_type: String,
    pub parent_id: Option<String>,
}

/// Database index status snapshot.
#[derive(Debug, Clone, Default)]
pub struct IndexStatus {
    pub db_size_bytes: u64,
    pub sessions_total: i64,
    pub sessions_by_engine: HashMap<String, i64>,
    pub facts_total: i64,
    pub handoffs_total: i64,
    pub last_indexed_at: Option<String>,
    pub oldest_session: Option<String>,
    pub newest_session: Option<String>,
}

/// Aggregate counters for a filtered session set.
#[derive(Debug, Clone, Default)]
pub struct AggregateResult {
    pub sessions: i64,
    pub total_input_tokens: i64,
    pub total_output_tokens: i64,
    pub estimated_cost_usd: f64,
    pub by_engine: HashMap<String, i64>,
}

/// Insert or update a session row by primary key.
pub fn upsert_session(conn: &Connection, session: &SessionRow) -> Result<(), GaalError> {
    conn.execute(
        r#"
        INSERT INTO sessions (
            id, engine, model, cwd, started_at, ended_at, exit_signal, last_event_at,
            parent_id, session_type, jsonl_path, total_input_tokens, total_output_tokens,
            cache_read_tokens, cache_creation_tokens, reasoning_tokens,
            total_tools, total_turns, peak_context, last_indexed_offset, subagent_type,
            gemini_summary
        )
        VALUES (
            :id, :engine, :model, :cwd, :started_at, :ended_at, :exit_signal, :last_event_at,
            :parent_id, :session_type, :jsonl_path, :total_input_tokens, :total_output_tokens,
            :cache_read_tokens, :cache_creation_tokens, :reasoning_tokens,
            :total_tools, :total_turns, :peak_context, :last_indexed_offset, :subagent_type,
            :gemini_summary
        )
        ON CONFLICT(id) DO UPDATE SET
            engine = excluded.engine,
            model = excluded.model,
            cwd = excluded.cwd,
            started_at = excluded.started_at,
            ended_at = excluded.ended_at,
            exit_signal = excluded.exit_signal,
            last_event_at = excluded.last_event_at,
            parent_id = excluded.parent_id,
            session_type = excluded.session_type,
            jsonl_path = excluded.jsonl_path,
            total_input_tokens = excluded.total_input_tokens,
            total_output_tokens = excluded.total_output_tokens,
            cache_read_tokens = excluded.cache_read_tokens,
            cache_creation_tokens = excluded.cache_creation_tokens,
            reasoning_tokens = excluded.reasoning_tokens,
            total_tools = excluded.total_tools,
            total_turns = excluded.total_turns,
            peak_context = excluded.peak_context,
            last_indexed_offset = excluded.last_indexed_offset,
            subagent_type = excluded.subagent_type,
            gemini_summary = excluded.gemini_summary
        "#,
        named_params! {
            ":id": &session.id,
            ":engine": &session.engine,
            ":model": &session.model,
            ":cwd": &session.cwd,
            ":started_at": &session.started_at,
            ":ended_at": &session.ended_at,
            ":exit_signal": &session.exit_signal,
            ":last_event_at": &session.last_event_at,
            ":parent_id": &session.parent_id,
            ":session_type": &session.session_type,
            ":jsonl_path": &session.jsonl_path,
            ":total_input_tokens": session.total_input_tokens,
            ":total_output_tokens": session.total_output_tokens,
            ":cache_read_tokens": session.cache_read_tokens,
            ":cache_creation_tokens": session.cache_creation_tokens,
            ":reasoning_tokens": session.reasoning_tokens,
            ":total_tools": session.total_tools,
            ":total_turns": session.total_turns,
            ":peak_context": session.peak_context,
            ":last_indexed_offset": session.last_indexed_offset,
            ":subagent_type": &session.subagent_type,
            ":gemini_summary": &session.gemini_summary,
        },
    )
    .map_err(db_err)?;
    if session.engine == "hermes" {
        register_hermes_alias(conn, &session.id)?;
    }
    Ok(())
}

/// Ensure every existing Hermes session has a durable alias row.
///
/// This is invoked during DB initialization so databases created before the
/// alias registry can resolve existing Hermes sessions by alias without a
/// forced reindex.
pub fn backfill_hermes_aliases(conn: &Connection) -> Result<(), GaalError> {
    let mut stmt = conn
        .prepare(
            r#"
            SELECT id
            FROM sessions
            WHERE engine = 'hermes'
            ORDER BY started_at ASC, id ASC
            "#,
        )
        .map_err(db_err)?;
    let mut rows = stmt.query([]).map_err(db_err)?;
    let mut ids = Vec::new();
    while let Some(row) = rows.next().map_err(db_err)? {
        ids.push(row.get::<_, String>(0).map_err(db_err)?);
    }
    drop(rows);
    drop(stmt);

    for id in ids {
        register_hermes_alias(conn, &id)?;
    }
    Ok(())
}

/// Register or fetch the deterministic 8-character alias for a Hermes session.
///
/// The first candidate hashes `hermes:` plus the full native session id. If the
/// alias is already assigned to a different session, subsequent candidates add a
/// deterministic counter salt until a free alias is found.
pub fn register_hermes_alias(conn: &Connection, session_id: &str) -> Result<String, GaalError> {
    if let Some(alias) = get_session_alias(conn, "hermes", session_id)? {
        return Ok(alias);
    }

    for counter in 0..=MAX_ALIAS_COLLISION_RETRIES {
        let alias = crate::util::hermes_alias_candidate(session_id, counter);
        if canonical_session_id_exists(conn, &alias)? {
            continue;
        }
        let existing = session_id_for_alias(conn, &alias, Some("hermes"))?;
        match existing.as_deref() {
            Some(existing_id) if existing_id == session_id => return Ok(alias),
            Some(_) => continue,
            None => {
                let inserted = insert_alias_row(conn, &alias, session_id, "hermes", counter);
                match inserted {
                    Ok(()) => return Ok(alias),
                    Err(err) if is_constraint_violation(&err) => {
                        if let Some(alias) = get_session_alias(conn, "hermes", session_id)? {
                            return Ok(alias);
                        }
                        continue;
                    }
                    Err(err) => return Err(db_err(err)),
                }
            }
        }
    }

    Err(GaalError::Internal(format!(
        "failed to allocate Hermes alias for {session_id} after {MAX_ALIAS_COLLISION_RETRIES} retries"
    )))
}

pub fn get_session_alias(
    conn: &Connection,
    engine: &str,
    session_id: &str,
) -> Result<Option<String>, GaalError> {
    conn.query_row(
        r#"
        SELECT alias
        FROM session_aliases
        WHERE session_id = :session_id
          AND engine = :engine
          AND scheme = :scheme
        LIMIT 1
        "#,
        named_params! {
            ":session_id": session_id,
            ":engine": engine,
            ":scheme": HERMES_ALIAS_SCHEME,
        },
        |row| row.get::<_, String>(0),
    )
    .optional()
    .map_err(db_err)
}

pub fn artifact_id_for_session(
    conn: &Connection,
    session: &SessionRow,
) -> Result<String, GaalError> {
    if session.engine == "hermes" {
        return Ok(get_session_alias(conn, &session.engine, &session.id)?
            .unwrap_or_else(|| crate::util::session_artifact_id(&session.engine, &session.id)));
    }

    Ok(crate::util::session_artifact_id(
        &session.engine,
        &session.id,
    ))
}

pub fn display_id_for_session(
    conn: &Connection,
    engine: &str,
    session_id: &str,
) -> Result<String, GaalError> {
    if engine == "hermes" {
        return Ok(get_session_alias(conn, engine, session_id)?
            .unwrap_or_else(|| crate::util::session_artifact_id(engine, session_id)));
    }

    Ok(crate::util::session_artifact_id(engine, session_id))
}

pub fn resolve_session_ids(
    conn: &Connection,
    reference: &str,
    engine: Option<&str>,
) -> Result<Vec<String>, GaalError> {
    let mut ids = Vec::new();

    let exact_ids = exact_session_ids(conn, reference, engine)?;
    let has_exact = !exact_ids.is_empty();
    for id in exact_ids {
        push_unique(&mut ids, id);
    }

    let alias_reference = reference.to_ascii_lowercase();
    if !has_exact && crate::util::is_hermes_alias(&alias_reference) {
        if let Some(id) = session_id_for_alias(conn, &alias_reference, engine)? {
            push_unique(&mut ids, id);
        }
    }

    if !has_exact {
        for id in prefix_session_ids(conn, reference, engine)? {
            push_unique(&mut ids, id);
        }
    }

    Ok(ids)
}

/// Return true when the alias registry is absent or missing rows for Hermes sessions.
pub fn hermes_alias_registry_needs_migration(conn: &Connection) -> Result<bool, GaalError> {
    if !table_exists(conn, "sessions")? {
        return Ok(true);
    }
    if !table_exists(conn, "session_aliases")? {
        return Ok(true);
    }

    conn.query_row(
        r#"
        SELECT EXISTS (
            SELECT 1
            FROM sessions s
            WHERE s.engine = 'hermes'
              AND NOT EXISTS (
                  SELECT 1
                  FROM session_aliases a
                  WHERE a.session_id = s.id
                    AND a.engine = 'hermes'
                    AND a.scheme = :scheme
              )
            LIMIT 1
        )
        "#,
        named_params! { ":scheme": HERMES_ALIAS_SCHEME },
        |row| row.get::<_, bool>(0),
    )
    .map_err(db_err)
}

/// Insert one fact and return the inserted row id.
pub fn insert_fact(conn: &Connection, fact: &Fact) -> Result<i64, GaalError> {
    let payload = fact_payload(fact)?;
    conn.execute(
        r#"
        INSERT INTO facts (
            session_id, ts, turn_number, fact_type, subject, detail, exit_code, success
        )
        VALUES (
            :session_id, :ts, :turn_number, :fact_type, :subject, :detail, :exit_code, :success
        )
        "#,
        named_params! {
            ":session_id": &payload.session_id,
            ":ts": &payload.ts,
            ":turn_number": &payload.turn_number,
            ":fact_type": &payload.fact_type,
            ":subject": &payload.subject,
            ":detail": &payload.detail,
            ":exit_code": &payload.exit_code,
            ":success": &payload.success,
        },
    )
    .map_err(db_err)?;
    Ok(conn.last_insert_rowid())
}

/// Insert facts using the provided connection (which may already be inside a transaction).
/// Callers are responsible for transaction management.
pub fn insert_facts_batch(conn: &Connection, facts: &[Fact]) -> Result<(), GaalError> {
    let mut stmt = conn
        .prepare(
            r#"
            INSERT INTO facts (
                session_id, ts, turn_number, fact_type, subject, detail, exit_code, success
            )
            VALUES (
                :session_id, :ts, :turn_number, :fact_type, :subject, :detail, :exit_code, :success
            )
            "#,
        )
        .map_err(db_err)?;

    for fact in facts {
        let payload = fact_payload(fact)?;
        stmt.execute(named_params! {
            ":session_id": &payload.session_id,
            ":ts": &payload.ts,
            ":turn_number": &payload.turn_number,
            ":fact_type": &payload.fact_type,
            ":subject": &payload.subject,
            ":detail": &payload.detail,
            ":exit_code": &payload.exit_code,
            ":success": &payload.success,
        })
        .map_err(db_err)?;
    }
    Ok(())
}

/// Insert or update a handoff row by session id.
pub fn upsert_handoff(conn: &Connection, handoff: &HandoffRecord) -> Result<(), GaalError> {
    let map = to_json_object(handoff)?;
    let session_id = required_json_string(&map, "session_id")?;
    let headline = optional_json_string(&map, "headline");
    let projects = optional_json_text(&map, "projects")?;
    let keywords = optional_json_text(&map, "keywords")?;
    let substance = optional_json_i64(&map, "substance").unwrap_or(0);
    let duration_minutes = optional_json_i64(&map, "duration_minutes").unwrap_or(0);
    let generated_at = optional_json_string(&map, "generated_at");
    let generated_by = optional_json_string(&map, "generated_by");
    let content_path = optional_json_string(&map, "content_path");

    conn.execute(
        r#"
        INSERT INTO handoffs (
            session_id, headline, projects, keywords, substance, duration_minutes,
            generated_at, generated_by, content_path
        )
        VALUES (
            :session_id, :headline, :projects, :keywords, :substance, :duration_minutes,
            :generated_at, :generated_by, :content_path
        )
        ON CONFLICT(session_id) DO UPDATE SET
            headline = excluded.headline,
            projects = excluded.projects,
            keywords = excluded.keywords,
            substance = excluded.substance,
            duration_minutes = excluded.duration_minutes,
            generated_at = excluded.generated_at,
            generated_by = excluded.generated_by,
            content_path = excluded.content_path
        "#,
        named_params! {
            ":session_id": &session_id,
            ":headline": &headline,
            ":projects": &projects,
            ":keywords": &keywords,
            ":substance": substance,
            ":duration_minutes": duration_minutes,
            ":generated_at": &generated_at,
            ":generated_by": &generated_by,
            ":content_path": &content_path,
        },
    )
    .map_err(db_err)?;
    Ok(())
}

/// Add a tag to a session.
pub fn add_tag(conn: &Connection, session_id: &str, tag: &str) -> Result<(), GaalError> {
    conn.execute(
        "INSERT OR IGNORE INTO session_tags (session_id, tag) VALUES (:session_id, :tag)",
        named_params! {
            ":session_id": session_id,
            ":tag": tag,
        },
    )
    .map_err(db_err)?;
    Ok(())
}

/// Remove a tag from a session.
pub fn remove_tag(conn: &Connection, session_id: &str, tag: &str) -> Result<(), GaalError> {
    conn.execute(
        "DELETE FROM session_tags WHERE session_id = :session_id AND tag = :tag",
        named_params! {
            ":session_id": session_id,
            ":tag": tag,
        },
    )
    .map_err(db_err)?;
    Ok(())
}

/// Delete a session and its associated facts by id.
pub fn delete_session(conn: &Connection, id: &str) -> Result<(), GaalError> {
    conn.execute(
        "DELETE FROM session_aliases WHERE session_id = :id",
        named_params! { ":id": id },
    )
    .map_err(db_err)?;
    conn.execute(
        "DELETE FROM facts WHERE session_id = :id",
        named_params! { ":id": id },
    )
    .map_err(db_err)?;
    conn.execute(
        "DELETE FROM sessions WHERE id = :id",
        named_params! { ":id": id },
    )
    .map_err(db_err)?;
    Ok(())
}

/// Fetch one session by id.
pub fn get_session(conn: &Connection, id: &str) -> Result<Option<SessionRow>, GaalError> {
    conn.query_row(
        r#"
        SELECT
            id, engine, model, cwd, started_at, ended_at, exit_signal, last_event_at,
            parent_id, session_type, jsonl_path, total_input_tokens, total_output_tokens,
            cache_read_tokens, cache_creation_tokens, reasoning_tokens,
            total_tools, total_turns, peak_context, last_indexed_offset, subagent_type,
            gemini_summary
        FROM sessions
        WHERE id = :id
        "#,
        named_params! { ":id": id },
        row_to_session,
    )
    .optional()
    .map_err(db_err)
}

/// Resolve sessions by an ID prefix, optionally filtering by engine.
pub fn resolve_by_prefix(
    conn: &Connection,
    prefix: &str,
    engine: Option<&str>,
) -> Result<Vec<SessionRow>, GaalError> {
    let ids = resolve_session_ids(conn, prefix, engine)?;
    let mut out = Vec::new();
    for id in ids {
        if let Some(session) = get_session(conn, &id)? {
            out.push(session);
        }
    }
    Ok(out)
}

/// List sessions with DB-level filtering.
pub fn list_sessions(conn: &Connection, filter: &ListFilter) -> Result<Vec<SessionRow>, GaalError> {
    let cwd_like = filter.cwd.as_ref().map(|value| format!("%{value}%"));
    let limit = filter.limit.unwrap_or(50).max(1);

    let sort_key = filter
        .sort_by
        .as_deref()
        .unwrap_or("started")
        .to_ascii_lowercase();
    let order_by = match sort_key.as_str() {
        "ended" => "s.ended_at DESC",
        "tokens" => "(s.total_input_tokens + s.total_output_tokens) DESC",
        "cost" => "(s.total_input_tokens + s.total_output_tokens) DESC",
        "duration" => "(strftime('%s', COALESCE(s.ended_at, CURRENT_TIMESTAMP)) - strftime('%s', s.started_at)) DESC",
        _ => "s.started_at DESC",
    };

    let sql = format!(
        r#"
        SELECT
            s.id, s.engine, s.model, s.cwd, s.started_at, s.ended_at, s.exit_signal, s.last_event_at,
            s.parent_id, s.session_type, s.jsonl_path, s.total_input_tokens, s.total_output_tokens,
            s.cache_read_tokens, s.cache_creation_tokens, s.reasoning_tokens,
            s.total_tools, s.total_turns, s.peak_context, s.last_indexed_offset, s.subagent_type,
            s.gemini_summary
        FROM sessions s
        WHERE (:engine IS NULL OR s.engine = :engine)
          AND (:session_type IS NULL OR s.session_type = :session_type)
          AND (:since IS NULL OR s.started_at >= :since)
          AND (:before IS NULL OR s.started_at <= :before)
          AND (:cwd_like IS NULL OR s.cwd LIKE :cwd_like)
          AND (
              :tag IS NULL OR EXISTS (
                  SELECT 1 FROM session_tags t
                  WHERE t.session_id = s.id AND t.tag = :tag
              )
          )
          AND (:include_subagents = 1 OR s.session_type != 'subagent')
          AND (:subagent_type IS NULL OR s.subagent_type = :subagent_type)
        ORDER BY {order_by}
        LIMIT :limit
        "#
    );

    let mut stmt = conn.prepare(&sql).map_err(db_err)?;
    let mut rows = stmt
        .query(named_params! {
            ":engine": filter.engine.as_deref(),
            ":session_type": filter.session_type.as_deref(),
            ":since": filter.since.as_deref(),
            ":before": filter.before.as_deref(),
            ":cwd_like": cwd_like.as_deref(),
            ":tag": filter.tag.as_deref(),
            ":include_subagents": filter.include_subagents as i64,
            ":subagent_type": filter.subagent_type.as_deref(),
            ":limit": limit,
        })
        .map_err(db_err)?;

    let mut out = Vec::new();
    while let Some(row) = rows.next().map_err(db_err)? {
        out.push(row_to_session(row).map_err(db_err)?);
    }

    Ok(out)
}

/// List coarse session candidates whose indexed interval overlaps a half-open window.
///
/// Parsed source events are still authoritative for final inclusion; this query
/// only avoids rendering clearly unrelated sessions.
pub fn list_sessions_overlapping_window(
    conn: &Connection,
    filter: &ActivityFilter,
) -> Result<Vec<SessionRow>, GaalError> {
    let cwd_like = filter.cwd.as_ref().map(|value| format!("%{value}%"));
    let limit = filter.limit.unwrap_or(250).max(1);

    let mut stmt = conn
        .prepare(
            r#"
            SELECT
                s.id, s.engine, s.model, s.cwd, s.started_at, s.ended_at, s.exit_signal, s.last_event_at,
                s.parent_id, s.session_type, s.jsonl_path, s.total_input_tokens, s.total_output_tokens,
                s.cache_read_tokens, s.cache_creation_tokens, s.reasoning_tokens,
                s.total_tools, s.total_turns, s.peak_context, s.last_indexed_offset, s.subagent_type,
                s.gemini_summary
            FROM sessions s
            WHERE (:engine IS NULL OR s.engine = :engine)
              AND (:cwd_like IS NULL OR s.cwd LIKE :cwd_like)
              AND (:include_subagents = 1 OR s.session_type != 'subagent')
              AND s.started_at < :before
              AND COALESCE(s.ended_at, s.last_event_at, s.started_at) >= :since
            ORDER BY COALESCE(s.last_event_at, s.ended_at, s.started_at) DESC
            LIMIT :limit
            "#,
        )
        .map_err(db_err)?;

    let mut rows = stmt
        .query(named_params! {
            ":engine": filter.engine.as_deref(),
            ":cwd_like": cwd_like.as_deref(),
            ":include_subagents": filter.include_subagents as i64,
            ":since": &filter.since,
            ":before": &filter.before,
            ":limit": limit,
        })
        .map_err(db_err)?;

    let mut out = Vec::new();
    while let Some(row) = rows.next().map_err(db_err)? {
        out.push(row_to_session(row).map_err(db_err)?);
    }

    Ok(out)
}

/// Count sessions matching the filter (ignoring LIMIT) for "Showing N of M" messages.
pub fn count_sessions(conn: &Connection, filter: &ListFilter) -> Result<i64, GaalError> {
    let cwd_like = filter.cwd.as_ref().map(|value| format!("%{value}%"));

    let sql = r#"
        SELECT COUNT(*)
        FROM sessions s
        WHERE (:engine IS NULL OR s.engine = :engine)
          AND (:session_type IS NULL OR s.session_type = :session_type)
          AND (:since IS NULL OR s.started_at >= :since)
          AND (:before IS NULL OR s.started_at <= :before)
          AND (:cwd_like IS NULL OR s.cwd LIKE :cwd_like)
          AND (
              :tag IS NULL OR EXISTS (
                  SELECT 1 FROM session_tags t
                  WHERE t.session_id = s.id AND t.tag = :tag
              )
          )
          AND (:include_subagents = 1 OR s.session_type != 'subagent')
          AND (:subagent_type IS NULL OR s.subagent_type = :subagent_type)
    "#;

    let count: i64 = conn
        .query_row(
            sql,
            named_params! {
                ":engine": filter.engine.as_deref(),
                ":session_type": filter.session_type.as_deref(),
                ":since": filter.since.as_deref(),
                ":before": filter.before.as_deref(),
                ":cwd_like": cwd_like.as_deref(),
                ":tag": filter.tag.as_deref(),
                ":include_subagents": filter.include_subagents as i64,
                ":subagent_type": filter.subagent_type.as_deref(),
            },
            |row| row.get(0),
        )
        .map_err(db_err)?;

    Ok(count)
}

/// Get facts for a session, optionally filtered by fact type.
pub fn get_facts(
    conn: &Connection,
    session_id: &str,
    fact_type_filter: Option<FactType>,
) -> Result<Vec<Fact>, GaalError> {
    let fact_type = fact_type_filter.as_ref().map(FactType::as_str);

    let mut stmt = conn
        .prepare(
            r#"
            SELECT
                id, session_id, ts, turn_number, fact_type, subject, detail, exit_code, success
            FROM facts
            WHERE session_id = :session_id
              AND (:fact_type IS NULL OR fact_type = :fact_type)
            ORDER BY ts ASC, id ASC
            "#,
        )
        .map_err(db_err)?;

    let mut rows = stmt
        .query(named_params! {
            ":session_id": session_id,
            ":fact_type": fact_type,
        })
        .map_err(db_err)?;

    let mut out = Vec::new();
    while let Some(row) = rows.next().map_err(db_err)? {
        let id: i64 = row.get("id").map_err(db_err)?;
        let session_id_val: String = row.get("session_id").map_err(db_err)?;
        let ts: String = row.get("ts").map_err(db_err)?;
        let turn_number: Option<i64> = row.get("turn_number").map_err(db_err)?;
        let fact_type_val: String = row.get("fact_type").map_err(db_err)?;
        let subject: Option<String> = row.get("subject").map_err(db_err)?;
        let detail: Option<String> = row.get("detail").map_err(db_err)?;
        let exit_code: Option<i64> = row.get("exit_code").map_err(db_err)?;
        let success: Option<i64> = row.get("success").map_err(db_err)?;

        let mut base = JsonMap::new();
        base.insert("id".to_string(), Value::Number(Number::from(id)));
        base.insert("session_id".to_string(), Value::String(session_id_val));
        base.insert("ts".to_string(), Value::String(ts));
        insert_opt_i64(&mut base, "turn_number", turn_number);
        base.insert("fact_type".to_string(), Value::String(fact_type_val));
        insert_opt_string(&mut base, "subject", subject);
        insert_opt_string(&mut base, "detail", detail);
        insert_opt_i64(&mut base, "exit_code", exit_code);
        insert_opt_i64(&mut base, "success", success);

        let primary = Value::Object(base.clone());
        match serde_json::from_value::<Fact>(primary) {
            Ok(fact) => out.push(fact),
            Err(primary_err) => {
                if let Some(success_value) = success {
                    let mut fallback = base;
                    fallback.insert("success".to_string(), Value::Bool(success_value != 0));
                    let fallback_value = Value::Object(fallback);
                    match serde_json::from_value::<Fact>(fallback_value) {
                        Ok(fact) => out.push(fact),
                        Err(fallback_err) => {
                            return Err(GaalError::Internal(format!(
                                "failed to deserialize fact row: {primary_err}; fallback: {fallback_err}"
                            )));
                        }
                    }
                } else {
                    return Err(GaalError::Internal(format!(
                        "failed to deserialize fact row: {primary_err}"
                    )));
                }
            }
        }
    }

    Ok(out)
}

/// Get the handoff record for a session.
pub fn get_handoff(
    conn: &Connection,
    session_id: &str,
) -> Result<Option<HandoffRecord>, GaalError> {
    let row = conn
        .query_row(
            r#"
            SELECT
                session_id, headline, projects, keywords, substance, duration_minutes,
                generated_at, generated_by, content_path
            FROM handoffs
            WHERE session_id = :session_id
            "#,
            named_params! { ":session_id": session_id },
            |row| {
                Ok((
                    row.get::<_, String>("session_id")?,
                    row.get::<_, Option<String>>("headline")?,
                    row.get::<_, Option<String>>("projects")?,
                    row.get::<_, Option<String>>("keywords")?,
                    row.get::<_, i64>("substance")?,
                    row.get::<_, i64>("duration_minutes")?,
                    row.get::<_, Option<String>>("generated_at")?,
                    row.get::<_, Option<String>>("generated_by")?,
                    row.get::<_, Option<String>>("content_path")?,
                ))
            },
        )
        .optional()
        .map_err(db_err)?;

    let Some((
        session_id_val,
        headline,
        projects_text,
        keywords_text,
        substance,
        duration_minutes,
        generated_at,
        generated_by,
        content_path,
    )) = row
    else {
        return Ok(None);
    };

    let mut map = JsonMap::new();
    map.insert("session_id".to_string(), Value::String(session_id_val));
    insert_opt_string(&mut map, "headline", headline);
    map.insert("projects".to_string(), parse_embedded_json(projects_text));
    map.insert("keywords".to_string(), parse_embedded_json(keywords_text));
    map.insert(
        "substance".to_string(),
        Value::Number(Number::from(substance)),
    );
    map.insert(
        "duration_minutes".to_string(),
        Value::Number(Number::from(duration_minutes)),
    );
    insert_opt_string(&mut map, "generated_at", generated_at);
    insert_opt_string(&mut map, "generated_by", generated_by);
    insert_opt_string(&mut map, "content_path", content_path);

    let handoff = serde_json::from_value::<HandoffRecord>(Value::Object(map))
        .map_err(|e| GaalError::Internal(format!("failed to deserialize handoff row: {e}")))?;
    Ok(Some(handoff))
}

/// Get all tags for a session.
pub fn get_tags(conn: &Connection, session_id: &str) -> Result<Vec<String>, GaalError> {
    let mut stmt = conn
        .prepare("SELECT tag FROM session_tags WHERE session_id = :session_id ORDER BY tag ASC")
        .map_err(db_err)?;
    let mut rows = stmt
        .query(named_params! { ":session_id": session_id })
        .map_err(db_err)?;

    let mut out = Vec::new();
    while let Some(row) = rows.next().map_err(db_err)? {
        out.push(row.get::<_, String>(0).map_err(db_err)?);
    }
    Ok(out)
}

/// Run an inverted "who did what" query over facts joined with sessions/handoffs.
pub fn query_who(conn: &Connection, filter: &WhoFilter) -> Result<Vec<WhoResult>, GaalError> {
    let subject_pattern = filter
        .subject_pattern
        .as_ref()
        .map(|pattern| format!("%{pattern}%"));
    let cwd_like = filter.cwd.as_ref().map(|value| format!("%{value}%"));
    let limit = filter.limit.unwrap_or(10).max(1);
    let failed_only = if filter.failed_only { 1_i64 } else { 0_i64 };

    // Build a fact_type IN (...) clause from the enum-derived strings.
    // Values come from FactType::as_str() (hardcoded &'static str), so
    // embedding them directly is safe from injection.
    let fact_type_clause = if filter.fact_types.is_empty() {
        "1 = 1".to_string()
    } else {
        let quoted: Vec<String> = filter
            .fact_types
            .iter()
            .map(|ft| format!("'{}'", ft.as_str()))
            .collect();
        format!("f.fact_type IN ({})", quoted.join(", "))
    };

    let sql = format!(
        r#"
            SELECT
                f.session_id,
                s.engine,
                f.ts,
                f.fact_type,
                f.subject,
                f.detail,
                h.headline,
                COALESCE(s.session_type, 'standalone') AS session_type,
                s.parent_id
            FROM facts f
            INNER JOIN sessions s ON s.id = f.session_id
            LEFT JOIN handoffs h ON h.session_id = f.session_id
            WHERE (:since IS NULL OR f.ts >= :since)
              AND (:before IS NULL OR f.ts <= :before)
              AND (:engine IS NULL OR s.engine = :engine)
              AND (:cwd_like IS NULL OR s.cwd LIKE :cwd_like)
              AND (
                    :subject_pattern IS NULL
                    OR COALESCE(f.subject, '') LIKE :subject_pattern
                    OR COALESCE(f.detail, '') LIKE :subject_pattern
                  )
              AND (
                    :tag IS NULL OR EXISTS (
                        SELECT 1
                        FROM session_tags t
                        WHERE t.session_id = s.id AND t.tag = :tag
                    )
                  )
              AND (
                    :failed_only = 0
                    OR (f.exit_code IS NOT NULL AND f.exit_code != 0)
                    OR f.success = 0
                  )
              AND {fact_type_clause}
            ORDER BY f.ts DESC, f.id DESC
            LIMIT :limit
            "#,
    );

    let mut stmt = conn.prepare(&sql).map_err(db_err)?;

    let mut rows = stmt
        .query(named_params! {
            ":since": filter.since.as_deref(),
            ":before": filter.before.as_deref(),
            ":engine": filter.engine.as_deref(),
            ":cwd_like": cwd_like.as_deref(),
            ":subject_pattern": subject_pattern.as_deref(),
            ":tag": filter.tag.as_deref(),
            ":failed_only": failed_only,
            ":limit": limit,
        })
        .map_err(db_err)?;

    // Safety-net post-filter: keeps the allowed_types check in case the SQL
    // clause and the caller's intent ever drift apart.
    let allowed_types: Option<HashSet<String>> = if filter.fact_types.is_empty() {
        None
    } else {
        Some(
            filter
                .fact_types
                .iter()
                .map(|fact_type| fact_type.as_str().to_string())
                .collect(),
        )
    };

    let mut out = Vec::new();
    while let Some(row) = rows.next().map_err(db_err)? {
        let result = WhoResult {
            session_id: row.get("session_id").map_err(db_err)?,
            engine: row.get("engine").map_err(db_err)?,
            ts: row.get("ts").map_err(db_err)?,
            fact_type: row.get("fact_type").map_err(db_err)?,
            subject: row.get("subject").map_err(db_err)?,
            detail: row.get("detail").map_err(db_err)?,
            session_headline: row.get("headline").map_err(db_err)?,
            session_type: row
                .get::<_, Option<String>>("session_type")
                .map_err(db_err)?
                .unwrap_or_else(|| "standalone".to_string()),
            parent_id: row.get("parent_id").map_err(db_err)?,
        };

        if let Some(types) = &allowed_types {
            if !types.contains(&result.fact_type) {
                continue;
            }
        }

        out.push(result);
        if out.len() as i64 >= limit {
            break;
        }
    }

    Ok(out)
}

/// Return high-level index status and row counts.
pub fn get_index_status(conn: &Connection) -> Result<IndexStatus, GaalError> {
    let db_size_bytes = match std::fs::metadata(crate::db::db_path()) {
        Ok(metadata) => metadata.len(),
        Err(_) => 0,
    };

    let sessions_total: i64 = conn
        .query_row("SELECT COUNT(*) FROM sessions", [], |row| row.get(0))
        .map_err(db_err)?;
    let facts_total: i64 = conn
        .query_row("SELECT COUNT(*) FROM facts", [], |row| row.get(0))
        .map_err(db_err)?;
    let handoffs_total: i64 = conn
        .query_row("SELECT COUNT(*) FROM handoffs", [], |row| row.get(0))
        .map_err(db_err)?;

    let mut sessions_by_engine = HashMap::new();
    {
        let mut stmt = conn
            .prepare("SELECT engine, COUNT(*) as count FROM sessions GROUP BY engine")
            .map_err(db_err)?;
        let mut rows = stmt.query([]).map_err(db_err)?;
        while let Some(row) = rows.next().map_err(db_err)? {
            let engine: String = row.get(0).map_err(db_err)?;
            let count: i64 = row.get(1).map_err(db_err)?;
            sessions_by_engine.insert(engine, count);
        }
    }

    let last_indexed_at: Option<String> = conn
        .query_row("SELECT MAX(last_event_at) FROM sessions", [], |row| {
            row.get(0)
        })
        .map_err(db_err)?;
    let oldest_session: Option<String> = conn
        .query_row("SELECT MIN(started_at) FROM sessions", [], |row| row.get(0))
        .map_err(db_err)?;
    let newest_session: Option<String> = conn
        .query_row("SELECT MAX(started_at) FROM sessions", [], |row| row.get(0))
        .map_err(db_err)?;

    Ok(IndexStatus {
        db_size_bytes,
        sessions_total,
        sessions_by_engine,
        facts_total,
        handoffs_total,
        last_indexed_at,
        oldest_session,
        newest_session,
    })
}

/// Aggregate token/cost and bucket counts over a filtered session set.
pub fn get_aggregate(conn: &Connection, filter: &ListFilter) -> Result<AggregateResult, GaalError> {
    let mut aggregate_filter = filter.clone();
    // Override limit to fetch all matching sessions — list_sessions defaults
    // None to 50, which silently truncated aggregate results.
    aggregate_filter.limit = Some(i64::MAX);
    let sessions = list_sessions(conn, &aggregate_filter)?;

    let mut by_engine: HashMap<String, i64> = HashMap::new();
    let mut total_input_tokens = 0_i64;
    let mut total_output_tokens = 0_i64;
    let mut total_cost = 0.0_f64;

    for session in &sessions {
        total_input_tokens += session.total_input_tokens;
        total_output_tokens += session.total_output_tokens;
        total_cost += estimate_session_cost(session);
        *by_engine.entry(session.engine.clone()).or_insert(0) += 1;
    }

    Ok(AggregateResult {
        sessions: sessions.len() as i64,
        total_input_tokens,
        total_output_tokens,
        estimated_cost_usd: (total_cost * 100.0).round() / 100.0,
        by_engine,
    })
}

fn row_to_session(row: &Row<'_>) -> rusqlite::Result<SessionRow> {
    Ok(SessionRow {
        id: row.get("id")?,
        engine: row.get("engine")?,
        model: row.get("model")?,
        cwd: row.get("cwd")?,
        started_at: row.get("started_at")?,
        ended_at: row.get("ended_at")?,
        exit_signal: row.get("exit_signal")?,
        last_event_at: row.get("last_event_at")?,
        parent_id: row.get("parent_id")?,
        session_type: row
            .get::<_, Option<String>>("session_type")?
            .unwrap_or_else(|| "standalone".to_string()),
        jsonl_path: row.get("jsonl_path")?,
        total_input_tokens: row.get("total_input_tokens")?,
        total_output_tokens: row.get("total_output_tokens")?,
        cache_read_tokens: row.get::<_, Option<i64>>("cache_read_tokens")?.unwrap_or(0),
        cache_creation_tokens: row
            .get::<_, Option<i64>>("cache_creation_tokens")?
            .unwrap_or(0),
        reasoning_tokens: row.get::<_, Option<i64>>("reasoning_tokens")?.unwrap_or(0),
        total_tools: row.get("total_tools")?,
        total_turns: row.get("total_turns")?,
        peak_context: row.get::<_, Option<i64>>("peak_context")?.unwrap_or(0),
        last_indexed_offset: row.get("last_indexed_offset")?,
        subagent_type: row
            .get::<_, Option<String>>("subagent_type")?
            .filter(|s| !s.is_empty()),
        gemini_summary: row
            .get::<_, Option<String>>("gemini_summary")
            .ok()
            .flatten()
            .filter(|s| !s.is_empty()),
    })
}

/// Model-aware cost rates: (input, output, cache_read, cache_write) per MTok.
fn cost_rates(model: &str) -> (f64, f64, f64, f64) {
    let m = model.to_lowercase();
    // gpt-5.4-mini must be checked before gpt-5.4
    if m.contains("gpt-5.4-mini") {
        (0.15, 0.60, 0.015, 0.0)
    } else if m.contains("gpt-5.4") {
        (2.50, 10.0, 0.25, 0.0)
    } else if m.contains("opus") {
        (15.0, 75.0, 1.5, 18.75)
    } else if m.contains("haiku") {
        (0.25, 1.25, 0.025, 0.3)
    } else {
        (3.0, 15.0, 0.3, 3.75) // default sonnet
    }
}

/// Estimate cost for a single session using model-aware rates.
/// Reasoning tokens are charged at the output rate.
pub fn estimate_session_cost(row: &SessionRow) -> f64 {
    let model = row.model.as_deref().unwrap_or("sonnet");
    let (rate_in, rate_out, rate_cr, rate_cw) = cost_rates(model);
    let mtok = 1_000_000.0_f64;
    let cost = (row.total_input_tokens as f64 / mtok) * rate_in
        + (row.total_output_tokens as f64 / mtok) * rate_out
        + (row.cache_read_tokens as f64 / mtok) * rate_cr
        + (row.cache_creation_tokens as f64 / mtok) * rate_cw
        + (row.reasoning_tokens as f64 / mtok) * rate_out;
    (cost * 100.0).round() / 100.0
}

struct FactPayload {
    session_id: String,
    ts: String,
    turn_number: Option<i64>,
    fact_type: String,
    subject: Option<String>,
    detail: Option<String>,
    exit_code: Option<i64>,
    success: Option<i64>,
}

fn fact_payload(fact: &Fact) -> Result<FactPayload, GaalError> {
    let map = to_json_object(fact)?;
    Ok(FactPayload {
        session_id: required_json_string(&map, "session_id")?,
        ts: required_json_string(&map, "ts")?,
        turn_number: optional_json_i64(&map, "turn_number"),
        fact_type: required_json_string(&map, "fact_type")?,
        subject: optional_json_string(&map, "subject"),
        detail: optional_json_string(&map, "detail"),
        exit_code: optional_json_i64(&map, "exit_code"),
        success: optional_json_i64_or_bool(&map, "success"),
    })
}

fn to_json_object<T: Serialize>(value: &T) -> Result<JsonMap<String, Value>, GaalError> {
    let serialized = serde_json::to_value(value)
        .map_err(|e| GaalError::Internal(format!("serialization error: {e}")))?;
    match serialized {
        Value::Object(map) => Ok(map),
        _ => Err(GaalError::Internal(
            "expected serialized value to be a JSON object".to_string(),
        )),
    }
}

fn required_json_string(map: &JsonMap<String, Value>, key: &str) -> Result<String, GaalError> {
    match map.get(key).and_then(Value::as_str) {
        Some(value) => Ok(value.to_string()),
        None => Err(GaalError::Internal(format!(
            "missing required string field `{key}`"
        ))),
    }
}

fn optional_json_string(map: &JsonMap<String, Value>, key: &str) -> Option<String> {
    let value = map.get(key)?;
    if value.is_null() {
        return None;
    }

    match value.as_str() {
        Some(text) => Some(text.to_string()),
        None => Some(value.to_string()),
    }
}

fn optional_json_i64(map: &JsonMap<String, Value>, key: &str) -> Option<i64> {
    let value = map.get(key)?;
    if value.is_null() {
        return None;
    }

    if let Some(int_value) = value.as_i64() {
        return Some(int_value);
    }

    value.as_str().and_then(|text| text.parse::<i64>().ok())
}

fn optional_json_i64_or_bool(map: &JsonMap<String, Value>, key: &str) -> Option<i64> {
    let value = map.get(key)?;
    if value.is_null() {
        return None;
    }

    if let Some(bool_value) = value.as_bool() {
        return Some(if bool_value { 1 } else { 0 });
    }

    optional_json_i64(map, key)
}

fn optional_json_text(
    map: &JsonMap<String, Value>,
    key: &str,
) -> Result<Option<String>, GaalError> {
    let Some(value) = map.get(key) else {
        return Ok(None);
    };
    if value.is_null() {
        return Ok(None);
    }
    serde_json::to_string(value)
        .map(Some)
        .map_err(|e| GaalError::Internal(format!("failed to serialize `{key}`: {e}")))
}

fn insert_opt_string(map: &mut JsonMap<String, Value>, key: &str, value: Option<String>) {
    match value {
        Some(text) => {
            map.insert(key.to_string(), Value::String(text));
        }
        None => {
            map.insert(key.to_string(), Value::Null);
        }
    }
}

fn insert_opt_i64(map: &mut JsonMap<String, Value>, key: &str, value: Option<i64>) {
    match value {
        Some(int_value) => {
            map.insert(key.to_string(), Value::Number(Number::from(int_value)));
        }
        None => {
            map.insert(key.to_string(), Value::Null);
        }
    }
}

fn parse_embedded_json(value: Option<String>) -> Value {
    let Some(raw) = value else {
        return Value::Null;
    };

    match serde_json::from_str::<Value>(&raw) {
        Ok(parsed) => parsed,
        Err(_) => Value::String(raw),
    }
}

fn insert_alias_row(
    conn: &Connection,
    alias: &str,
    session_id: &str,
    engine: &str,
    counter: u32,
) -> rusqlite::Result<()> {
    conn.execute(
        r#"
        INSERT INTO session_aliases (alias, session_id, engine, scheme, counter, created_at)
        VALUES (:alias, :session_id, :engine, :scheme, :counter, :created_at)
        "#,
        named_params! {
            ":alias": alias,
            ":session_id": session_id,
            ":engine": engine,
            ":scheme": HERMES_ALIAS_SCHEME,
            ":counter": i64::from(counter),
            ":created_at": unix_now(),
        },
    )?;
    Ok(())
}

fn session_id_for_alias(
    conn: &Connection,
    alias: &str,
    engine: Option<&str>,
) -> Result<Option<String>, GaalError> {
    conn.query_row(
        r#"
        SELECT session_id
        FROM session_aliases
        WHERE alias = :alias
          AND (:engine IS NULL OR engine = :engine)
          AND scheme = :scheme
        "#,
        named_params! {
            ":alias": alias,
            ":engine": engine,
            ":scheme": HERMES_ALIAS_SCHEME,
        },
        |row| row.get::<_, String>(0),
    )
    .optional()
    .map_err(db_err)
}

fn canonical_session_id_exists(conn: &Connection, id: &str) -> Result<bool, GaalError> {
    conn.query_row(
        "SELECT EXISTS (SELECT 1 FROM sessions WHERE id = :id LIMIT 1)",
        named_params! { ":id": id },
        |row| row.get::<_, bool>(0),
    )
    .map_err(db_err)
}

fn table_exists(conn: &Connection, table_name: &str) -> Result<bool, GaalError> {
    conn.query_row(
        "SELECT COUNT(*) FROM sqlite_master WHERE type='table' AND name=?1",
        [table_name],
        |row| row.get::<_, i64>(0),
    )
    .map(|count| count > 0)
    .map_err(db_err)
}

fn exact_session_ids(
    conn: &Connection,
    reference: &str,
    engine: Option<&str>,
) -> Result<Vec<String>, GaalError> {
    let mut stmt = conn
        .prepare(
            r#"
            SELECT id
            FROM sessions
            WHERE id = :id
              AND (:engine IS NULL OR engine = :engine)
            ORDER BY started_at DESC
            "#,
        )
        .map_err(db_err)?;
    let mut rows = stmt
        .query(named_params! {
            ":id": reference,
            ":engine": engine,
        })
        .map_err(db_err)?;

    let mut out = Vec::new();
    while let Some(row) = rows.next().map_err(db_err)? {
        out.push(row.get::<_, String>(0).map_err(db_err)?);
    }
    Ok(out)
}

fn prefix_session_ids(
    conn: &Connection,
    reference: &str,
    engine: Option<&str>,
) -> Result<Vec<String>, GaalError> {
    let like = format!("{reference}%");
    let mut stmt = conn
        .prepare(
            r#"
            SELECT id
            FROM sessions
            WHERE id LIKE :prefix
              AND (:engine IS NULL OR engine = :engine)
            ORDER BY started_at DESC
            "#,
        )
        .map_err(db_err)?;
    let mut rows = stmt
        .query(named_params! {
            ":prefix": like,
            ":engine": engine,
        })
        .map_err(db_err)?;

    let mut out = Vec::new();
    while let Some(row) = rows.next().map_err(db_err)? {
        out.push(row.get::<_, String>(0).map_err(db_err)?);
    }
    Ok(out)
}

fn push_unique(ids: &mut Vec<String>, id: String) {
    if !ids.iter().any(|existing| existing == &id) {
        ids.push(id);
    }
}

fn unix_now() -> i64 {
    SystemTime::now()
        .duration_since(UNIX_EPOCH)
        .map(|d| d.as_secs() as i64)
        .unwrap_or(0)
}

fn is_constraint_violation(err: &rusqlite::Error) -> bool {
    err.sqlite_error_code() == Some(ErrorCode::ConstraintViolation)
}

fn db_err(err: rusqlite::Error) -> GaalError {
    GaalError::Db(err)
}

/// Fetch a value from the `meta` key/value table.
///
/// Returns `Ok(None)` when the key is absent, `Ok(Some(value))` otherwise.
/// Used for durable cursors such as incremental-backfill timestamps.
pub fn get_meta(conn: &Connection, key: &str) -> rusqlite::Result<Option<String>> {
    conn.query_row(
        "SELECT value FROM meta WHERE key = :key",
        named_params! { ":key": key },
        |row| row.get::<_, String>(0),
    )
    .optional()
}

/// Upsert a value into the `meta` key/value table.
///
/// Stamps `updated_at` with the current unix-second timestamp.  Callers should
/// treat this as authoritative persistence — failure propagates to the caller
/// so the cursor does not silently skip forward on a half-written row.
pub fn set_meta(conn: &Connection, key: &str, value: &str) -> rusqlite::Result<()> {
    let now = std::time::SystemTime::now()
        .duration_since(std::time::UNIX_EPOCH)
        .map(|d| d.as_secs() as i64)
        .unwrap_or(0);
    conn.execute(
        "INSERT OR REPLACE INTO meta (key, value, updated_at) VALUES (:key, :value, :updated_at)",
        named_params! {
            ":key": key,
            ":value": value,
            ":updated_at": now,
        },
    )?;
    Ok(())
}

#[cfg(test)]
mod meta_tests {
    use super::*;
    use crate::db::init_db;
    use rusqlite::Connection;

    fn fresh_conn() -> Connection {
        let conn = Connection::open_in_memory().expect("open in-memory db");
        init_db(&conn).expect("init schema");
        conn
    }

    #[test]
    fn meta_roundtrip_get_set() {
        let conn = fresh_conn();
        assert_eq!(get_meta(&conn, "backfill:claude").unwrap(), None);
        set_meta(&conn, "backfill:claude", "1234567890").unwrap();
        assert_eq!(
            get_meta(&conn, "backfill:claude").unwrap(),
            Some("1234567890".to_string())
        );
        set_meta(&conn, "backfill:claude", "9999999999").unwrap();
        assert_eq!(
            get_meta(&conn, "backfill:claude").unwrap(),
            Some("9999999999".to_string())
        );
    }

    #[test]
    fn hermes_upsert_registers_alias_and_resolves_it() {
        let conn = fresh_conn();
        let session = hermes_session("20260504_141414_childbb", "2026-05-04T14:14:14Z");

        upsert_session(&conn, &session).unwrap();

        let alias = get_session_alias(&conn, "hermes", &session.id)
            .unwrap()
            .expect("alias should exist");
        assert_eq!(alias.len(), 8);
        assert!(crate::util::is_hermes_alias(&alias));
        assert_eq!(
            resolve_session_ids(&conn, &alias, None).unwrap(),
            vec![session.id.clone()]
        );
        assert_eq!(
            resolve_session_ids(&conn, &session.id, None).unwrap(),
            vec![session.id.clone()]
        );
    }

    #[test]
    fn hermes_alias_collision_uses_next_deterministic_counter() {
        let conn = fresh_conn();
        let existing = hermes_session("20260504_111111_existing", "2026-05-04T11:11:11Z");
        let collided = hermes_session("20260504_121212_collided", "2026-05-04T12:12:12Z");
        upsert_session(&conn, &existing).unwrap();
        upsert_session(&conn, &collided).unwrap();

        let candidate = crate::util::hermes_alias_candidate(&collided.id, 0);
        conn.execute(
            "DELETE FROM session_aliases WHERE session_id = :session_id",
            named_params! { ":session_id": &collided.id },
        )
        .unwrap();
        conn.execute(
            "DELETE FROM session_aliases WHERE session_id = :session_id",
            named_params! { ":session_id": &existing.id },
        )
        .unwrap();
        conn.execute(
            "DELETE FROM session_aliases WHERE alias = :alias",
            named_params! { ":alias": &candidate },
        )
        .unwrap();
        insert_alias_row(&conn, &candidate, &existing.id, "hermes", 0).unwrap();

        let alias = register_hermes_alias(&conn, &collided.id).unwrap();

        assert_ne!(alias, candidate);
        assert_eq!(alias, crate::util::hermes_alias_candidate(&collided.id, 1));
        assert_eq!(
            resolve_session_ids(&conn, &alias, Some("hermes")).unwrap(),
            vec![collided.id.clone()]
        );
    }

    #[test]
    fn hermes_alias_collision_with_canonical_session_id_uses_next_counter() {
        let conn = fresh_conn();
        let hermes = hermes_session("20260504_121212_collided", "2026-05-04T12:12:12Z");
        let candidate = crate::util::hermes_alias_candidate(&hermes.id, 0);
        upsert_session(
            &conn,
            &SessionRow {
                id: candidate.clone(),
                engine: "codex".to_string(),
                model: None,
                cwd: None,
                started_at: "2026-05-04T12:00:00Z".to_string(),
                ended_at: None,
                exit_signal: None,
                last_event_at: None,
                parent_id: None,
                session_type: "standalone".to_string(),
                jsonl_path: "/tmp/codex.jsonl".to_string(),
                total_input_tokens: 0,
                total_output_tokens: 0,
                cache_read_tokens: 0,
                cache_creation_tokens: 0,
                reasoning_tokens: 0,
                total_tools: 0,
                total_turns: 1,
                peak_context: 0,
                last_indexed_offset: 0,
                subagent_type: None,
                gemini_summary: None,
            },
        )
        .unwrap();
        upsert_session(&conn, &hermes).unwrap();

        let alias = get_session_alias(&conn, "hermes", &hermes.id)
            .unwrap()
            .expect("alias should exist");
        assert_ne!(alias, candidate);
        assert_eq!(alias, crate::util::hermes_alias_candidate(&hermes.id, 1));
        assert_eq!(
            resolve_session_ids(&conn, &candidate, None).unwrap(),
            vec![candidate]
        );
        assert_eq!(
            resolve_session_ids(&conn, &alias, Some("hermes")).unwrap(),
            vec![hermes.id.clone()]
        );
    }

    #[test]
    fn hermes_alias_registry_reports_pre_alias_schema_needs_migration() {
        let conn = Connection::open_in_memory().expect("open in-memory db");
        conn.execute_batch(
            r#"
            CREATE TABLE sessions (
                id TEXT PRIMARY KEY,
                engine TEXT NOT NULL,
                started_at TEXT NOT NULL,
                jsonl_path TEXT NOT NULL
            );
            INSERT INTO sessions (id, engine, started_at, jsonl_path)
            VALUES ('20260504_121212_oldschema', 'hermes', '2026-05-04T12:12:12Z', '/tmp/hermes.db');
            "#,
        )
        .expect("seed old schema");

        assert!(hermes_alias_registry_needs_migration(&conn).unwrap());
    }

    #[test]
    fn hermes_alias_registry_reports_missing_alias_rows_need_migration() {
        let conn = fresh_conn();
        let session = hermes_session("20260504_131313_missingalias", "2026-05-04T13:13:13Z");
        upsert_session(&conn, &session).unwrap();

        assert!(!hermes_alias_registry_needs_migration(&conn).unwrap());

        conn.execute(
            "DELETE FROM session_aliases WHERE session_id = :session_id",
            named_params! { ":session_id": &session.id },
        )
        .unwrap();

        assert!(hermes_alias_registry_needs_migration(&conn).unwrap());
    }

    #[test]
    fn artifact_id_for_hermes_prefers_registered_alias() {
        let conn = fresh_conn();
        let session = hermes_session("20260504_131313_aliaspath", "2026-05-04T13:13:13Z");
        upsert_session(&conn, &session).unwrap();

        let alias = get_session_alias(&conn, "hermes", &session.id)
            .unwrap()
            .expect("alias should exist");
        assert_eq!(artifact_id_for_session(&conn, &session).unwrap(), alias);
    }

    #[test]
    fn activity_overlap_query_does_not_let_stale_open_sessions_starve_limit() {
        let conn = fresh_conn();
        upsert_session(
            &conn,
            &test_session("staleopen", "2026-01-01T00:00:00Z", None, None),
        )
        .unwrap();
        upsert_session(
            &conn,
            &test_session(
                "realmay25",
                "2026-05-20T00:00:00Z",
                None,
                Some("2026-05-25T12:00:00Z"),
            ),
        )
        .unwrap();

        let rows = list_sessions_overlapping_window(
            &conn,
            &ActivityFilter {
                since: "2026-05-25T00:00:00Z".to_string(),
                before: "2026-05-26T00:00:00Z".to_string(),
                include_subagents: true,
                limit: Some(1),
                ..ActivityFilter::default()
            },
        )
        .unwrap();

        assert_eq!(rows.len(), 1);
        assert_eq!(rows[0].id, "realmay25");
    }

    fn hermes_session(id: &str, started_at: &str) -> SessionRow {
        let mut session = test_session(id, started_at, None, Some(started_at));
        session.engine = "hermes".to_string();
        session.jsonl_path = "/tmp/hermes-state.db".to_string();
        session
    }

    fn test_session(
        id: &str,
        started_at: &str,
        ended_at: Option<&str>,
        last_event_at: Option<&str>,
    ) -> SessionRow {
        SessionRow {
            id: id.to_string(),
            engine: "codex".to_string(),
            model: Some("gpt-5.5".to_string()),
            cwd: Some("/tmp/gaal-test".to_string()),
            started_at: started_at.to_string(),
            ended_at: ended_at.map(str::to_string),
            exit_signal: None,
            last_event_at: last_event_at.map(str::to_string),
            parent_id: None,
            session_type: "standalone".to_string(),
            jsonl_path: "/tmp/missing.jsonl".to_string(),
            total_input_tokens: 0,
            total_output_tokens: 0,
            cache_read_tokens: 0,
            cache_creation_tokens: 0,
            reasoning_tokens: 0,
            total_tools: 0,
            total_turns: 1,
            peak_context: 0,
            last_indexed_offset: 0,
            subagent_type: None,
            gemini_summary: None,
        }
    }
}
