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

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

use crate::output::CommandReport;
use crate::paths::state::StateLayout;
use crate::profile;
use crate::repo::marker as repo_marker;
use crate::state::continuity as runtime_state;
use crate::state::session as session_state;
use crate::state::thread_transfer;

const THREAD_SCHEMA_VERSION: u32 = 1;

macro_rules! thread_columns {
    () => {
        "thread_id, display_name, status, created_at_epoch_s, \
         updated_at_epoch_s, last_session_id, legacy_thread_ref"
    };
}

const SELECT_THREADS_ORDERED_SQL: &str = concat!(
    "SELECT ",
    thread_columns!(),
    " FROM threads ORDER BY updated_at_epoch_s DESC, created_at_epoch_s DESC, thread_id ASC"
);

const SELECT_THREAD_BY_ID_SQL: &str = concat!(
    "SELECT ",
    thread_columns!(),
    " FROM threads WHERE thread_id = ?1"
);

const SELECT_THREAD_BY_LEGACY_SQL: &str = concat!(
    "SELECT ",
    thread_columns!(),
    " FROM threads WHERE legacy_thread_ref = ?1"
);

const SELECT_THREADS_BY_DISPLAY_NAME_SQL: &str = concat!(
    "SELECT ",
    thread_columns!(),
    " FROM threads WHERE display_name = ?1 ORDER BY created_at_epoch_s ASC, thread_id ASC"
);

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

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

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

impl ThreadDerivedStatus {
    fn as_str(self) -> &'static str {
        match self {
            Self::Active => "active",
            Self::Dormant => "dormant",
            Self::Closed => "closed",
        }
    }
}

#[derive(Clone, Copy, Debug, Eq, PartialEq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub(crate) enum ThreadStateStatus {
    Bound,
    Unbound,
    Mismatch,
    DanglingBinding,
    Missing,
    Throttled,
}

#[derive(Clone, Copy, Debug, Eq, PartialEq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub(crate) enum ThreadBindingStatus {
    Bound,
    Missing,
    Throttled,
}

#[derive(Clone, Copy, Debug, Eq, PartialEq, Serialize)]
#[serde(rename_all = "snake_case")]
enum ThreadEventKind {
    Created,
    Shown,
    NameChanged,
    Resumed,
    Closed,
    SessionOpened,
    SessionClosed,
    BoundToChannel,
    UnboundFromChannel,
    Exported,
    Imported,
}

impl ThreadEventKind {
    fn as_str(self) -> &'static str {
        match self {
            Self::Created => "created",
            Self::Shown => "shown",
            Self::NameChanged => "name_changed",
            Self::Resumed => "resumed",
            Self::Closed => "closed",
            Self::SessionOpened => "session_opened",
            Self::SessionClosed => "session_closed",
            Self::BoundToChannel => "bound_to_channel",
            Self::UnboundFromChannel => "unbound_from_channel",
            Self::Exported => "exported",
            Self::Imported => "imported",
        }
    }
}

const THREAD_SCHEMA: &str = r#"
CREATE TABLE IF NOT EXISTS threads (
    thread_id TEXT PRIMARY KEY,
    display_name TEXT NOT NULL,
    status TEXT NOT NULL CHECK (status IN ('open', 'closed')),
    created_at_epoch_s INTEGER NOT NULL,
    updated_at_epoch_s INTEGER NOT NULL,
    last_session_id TEXT,
    legacy_thread_ref TEXT UNIQUE
);

CREATE INDEX IF NOT EXISTS threads_display_name_idx
    ON threads (display_name);

CREATE TABLE IF NOT EXISTS thread_events (
    id INTEGER PRIMARY KEY AUTOINCREMENT,
    thread_id TEXT NOT NULL,
    event_kind TEXT NOT NULL,
    at_epoch_s INTEGER NOT NULL,
    channel_ref TEXT,
    session_id TEXT,
    metadata_json TEXT NOT NULL DEFAULT '{}',
    FOREIGN KEY(thread_id) REFERENCES threads(thread_id)
);

CREATE INDEX IF NOT EXISTS thread_events_thread_idx
    ON thread_events (thread_id, at_epoch_s DESC, id DESC);

CREATE TABLE IF NOT EXISTS thread_channel_bindings (
    channel_ref TEXT PRIMARY KEY,
    thread_id TEXT NOT NULL,
    bound_at_epoch_s INTEGER NOT NULL,
    session_id TEXT,
    FOREIGN KEY(thread_id) REFERENCES threads(thread_id)
);

CREATE INDEX IF NOT EXISTS thread_channel_bindings_thread_idx
    ON thread_channel_bindings (thread_id);
"#;

#[derive(Clone, Debug, Serialize, Deserialize)]
pub(crate) struct ThreadStateView {
    pub(crate) path: String,
    pub(crate) status: ThreadStateStatus,
    pub(crate) channel_ref: String,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub(crate) thread_id: Option<String>,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub(crate) display_name: Option<String>,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub(crate) record_status: Option<ThreadRecordStatus>,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub(crate) session_thread_id: Option<String>,
    pub(crate) binding: ThreadBindingView,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub(crate) active_session_matches: Option<bool>,
    #[serde(default, skip_serializing_if = "Vec::is_empty")]
    pub(crate) warnings: Vec<String>,
}

#[derive(Clone, Debug, Serialize, Deserialize)]
pub(crate) struct ThreadBindingView {
    pub(crate) status: ThreadBindingStatus,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub(crate) thread_id: Option<String>,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub(crate) bound_at_epoch_s: Option<u64>,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub(crate) session_id: Option<String>,
}

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

#[derive(Serialize)]
struct ThreadListItem {
    thread_id: String,
    display_name: String,
    status: ThreadDerivedStatus,
    record_status: ThreadRecordStatus,
    created_at_epoch_s: u64,
    updated_at_epoch_s: u64,
    #[serde(skip_serializing_if = "Option::is_none")]
    last_session_id: Option<String>,
    #[serde(skip_serializing_if = "Option::is_none")]
    legacy_thread_ref: Option<String>,
    bound_to_current_channel: bool,
}

#[derive(Serialize)]
pub struct ThreadShowReport {
    command: &'static str,
    ok: bool,
    path: String,
    profile: String,
    project_id: String,
    locality_id: String,
    store: String,
    channel_ref: String,
    thread: ThreadDetailView,
    events: Vec<ThreadEventView>,
}

#[derive(Serialize)]
struct ThreadDetailView {
    thread_id: String,
    display_name: String,
    status: ThreadDerivedStatus,
    record_status: ThreadRecordStatus,
    created_at_epoch_s: u64,
    updated_at_epoch_s: u64,
    #[serde(skip_serializing_if = "Option::is_none")]
    last_session_id: Option<String>,
    #[serde(skip_serializing_if = "Option::is_none")]
    legacy_thread_ref: Option<String>,
    binding: ThreadBindingView,
}

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

#[derive(Serialize)]
pub struct ThreadMutationReport {
    command: &'static str,
    ok: bool,
    path: String,
    profile: String,
    project_id: String,
    locality_id: String,
    store: String,
    channel_ref: String,
    thread: ThreadDetailView,
    action: &'static str,
}

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

#[derive(Clone)]
struct ThreadRecord {
    thread_id: String,
    display_name: String,
    status: ThreadRecordStatus,
    created_at_epoch_s: u64,
    updated_at_epoch_s: u64,
    last_session_id: Option<String>,
    legacy_thread_ref: Option<String>,
}

#[derive(Clone)]
struct ChannelBinding {
    thread_id: String,
    bound_at_epoch_s: u64,
    session_id: Option<String>,
}

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

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

        println!(
            "Threads for profile `{}` in linked project `{}`:",
            self.profile, self.project_id
        );
        for thread in &self.threads {
            let marker = if thread.bound_to_current_channel {
                "current"
            } else {
                thread.status.as_str()
            };
            println!(
                "- {} `{}` [{}]",
                thread.thread_id, thread.display_name, marker
            );
        }
    }
}

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

    fn render_text(&self) {
        println!(
            "Thread {} `{}` is {}.",
            self.thread.thread_id,
            self.thread.display_name,
            self.thread.status.as_str()
        );
        for event in self.events.iter().take(8) {
            println!("- {} at {}", event.event_kind, event.at_epoch_s);
        }
    }
}

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

    fn render_text(&self) {
        println!(
            "Thread {} `{}` {}.",
            self.thread.thread_id, self.thread.display_name, self.action
        );
    }
}

pub(crate) fn list(repo_root: &Path, explicit_profile: Option<&str>) -> Result<ThreadListReport> {
    let context = resolve_context(repo_root, explicit_profile, "list")?;
    let conn = open_thread_db(&context)?;
    ensure_legacy_thread(&conn, &context, None)?;
    let current_binding = read_binding(&conn, &context.channel_ref)?;
    let threads = read_threads(&conn)?
        .into_iter()
        .map(|record| list_item(&record, current_binding.as_ref()))
        .collect();

    Ok(ThreadListReport {
        command: "thread-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,
        threads,
    })
}

pub(crate) fn show(
    repo_root: &Path,
    explicit_profile: Option<&str>,
    selector: &str,
) -> Result<ThreadShowReport> {
    let context = resolve_context(repo_root, explicit_profile, "show")?;
    let conn = open_thread_db(&context)?;
    ensure_legacy_thread(&conn, &context, None)?;
    let record = resolve_thread(&conn, selector)?;
    record_event(
        &conn,
        &record.thread_id,
        ThreadEventKind::Shown,
        Some(&context.channel_ref),
        None,
        json!({}),
    )?;
    let current_binding = read_binding(&conn, &context.channel_ref)?;
    let events = read_events(&conn, &record.thread_id)?;

    Ok(ThreadShowReport {
        command: "thread-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,
        thread: detail_view(&record, current_binding.as_ref()),
        events,
    })
}

pub(crate) fn name(
    repo_root: &Path,
    explicit_profile: Option<&str>,
    thread_id: &str,
    display_name: &str,
) -> Result<ThreadMutationReport> {
    let context = resolve_context(repo_root, explicit_profile, "name")?;
    let conn = open_thread_db(&context)?;
    ensure_legacy_thread(&conn, &context, None)?;
    let mut record = read_thread(&conn, thread_id)?
        .ok_or_else(|| anyhow::anyhow!("thread `{thread_id}` was not found"))?;
    let normalized = normalize_display_name(display_name)?;
    ensure_display_name_available(&conn, &normalized, Some(&record.thread_id))?;
    let previous = record.display_name.clone();
    let now = session_state::now_epoch_s()?;
    conn.execute(
        "UPDATE threads SET display_name = ?1, updated_at_epoch_s = ?2 WHERE thread_id = ?3",
        params![normalized, now, record.thread_id],
    )?;
    record.display_name = normalized;
    record.updated_at_epoch_s = now;
    record_event(
        &conn,
        &record.thread_id,
        ThreadEventKind::NameChanged,
        Some(&context.channel_ref),
        None,
        json!({ "previous": previous, "display_name": record.display_name }),
    )?;
    let current_binding = read_binding(&conn, &context.channel_ref)?;

    Ok(mutation_report(
        "thread-name",
        "renamed",
        context,
        detail_view(&record, current_binding.as_ref()),
    ))
}

pub(crate) fn resume(
    repo_root: &Path,
    explicit_profile: Option<&str>,
    selector: &str,
) -> Result<ThreadMutationReport> {
    let context = resolve_context(repo_root, explicit_profile, "resume")?;
    let conn = open_thread_db(&context)?;
    ensure_legacy_thread(&conn, &context, None)?;
    let mut record = resolve_thread(&conn, selector)?;
    let active_session = load_active_session(&context.layout)?;
    if let Some(session) = active_session.as_ref() {
        if let Some(session_thread_id) = session.thread_id.as_deref() {
            if session_thread_id != record.thread_id {
                bail!(
                    "active session is bound to thread `{session_thread_id}`; clear or close that session before resuming `{}`",
                    record.thread_id
                );
            }
        }
    }

    let session_id = active_session
        .as_ref()
        .and_then(|session| session.session_id.as_deref());
    bind_thread(
        &conn,
        &context,
        &mut record,
        session_id,
        ThreadEventKind::Resumed,
    )?;

    Ok(mutation_report(
        "thread-resume",
        "resumed in the current channel",
        context.clone(),
        detail_view(&record, read_binding(&conn, &context.channel_ref)?.as_ref()),
    ))
}

pub(crate) fn close(
    repo_root: &Path,
    explicit_profile: Option<&str>,
    selector: &str,
) -> Result<ThreadMutationReport> {
    let context = resolve_context(repo_root, explicit_profile, "close")?;
    let conn = open_thread_db(&context)?;
    ensure_legacy_thread(&conn, &context, None)?;
    let mut record = resolve_thread(&conn, selector)?;
    let active_session = load_active_session(&context.layout)?;
    if active_session
        .as_ref()
        .and_then(|session| session.thread_id.as_deref())
        == Some(record.thread_id.as_str())
    {
        bail!(
            "thread `{}` is bound to the active session; clear the session before closing the thread",
            record.thread_id
        );
    }

    let now = session_state::now_epoch_s()?;
    let current_binding = read_binding(&conn, &context.channel_ref)?;
    conn.execute(
        "DELETE FROM thread_channel_bindings WHERE thread_id = ?1",
        params![record.thread_id],
    )?;
    if current_binding
        .as_ref()
        .is_some_and(|binding| binding.thread_id == record.thread_id)
    {
        record_event(
            &conn,
            &record.thread_id,
            ThreadEventKind::UnboundFromChannel,
            Some(&context.channel_ref),
            None,
            json!({ "reason": "thread_close" }),
        )?;
    }
    conn.execute(
        "UPDATE threads SET status = 'closed', updated_at_epoch_s = ?1 WHERE thread_id = ?2",
        params![now, record.thread_id],
    )?;
    record.status = ThreadRecordStatus::Closed;
    record.updated_at_epoch_s = now;
    record_event(
        &conn,
        &record.thread_id,
        ThreadEventKind::Closed,
        Some(&context.channel_ref),
        None,
        json!({}),
    )?;

    Ok(mutation_report(
        "thread-close",
        "closed",
        context.clone(),
        detail_view(&record, read_binding(&conn, &context.channel_ref)?.as_ref()),
    ))
}

pub(crate) fn current_state(
    layout: &StateLayout,
    locality_id: &str,
    tracked_session: Option<&session_state::SessionStateFile>,
) -> Result<ThreadStateView> {
    let context = context_from_layout(layout, locality_id)?;
    let session_thread_id = tracked_session.and_then(|state| state.thread_id.clone());
    if !context.db_path.exists() {
        return Ok(missing_state(context, session_thread_id));
    }
    let conn = open_thread_db(&context)?;
    let binding = read_binding(&conn, &context.channel_ref)?;
    let record = match binding.as_ref() {
        Some(binding) => read_thread(&conn, &binding.thread_id)?,
        None => match session_thread_id.as_deref() {
            Some(thread_id) => read_thread(&conn, thread_id)?,
            None => None,
        },
    };
    Ok(state_view(
        context,
        record.as_ref(),
        binding.as_ref(),
        session_thread_id,
    ))
}

pub(crate) fn ensure_active_thread_for_session(
    repo_root: &Path,
    layout: &StateLayout,
    profile: &str,
    locality_id: &str,
    session_id: Option<&str>,
    existing_session_thread_id: Option<&str>,
) -> Result<String> {
    let context = ThreadContext {
        repo_root: repo_root.to_path_buf(),
        layout: layout.clone(),
        profile: profile.to_owned(),
        locality_id: locality_id.to_owned(),
        db_path: layout.repo_threads_db_path(locality_id)?,
        channel_ref: channel_ref(layout),
    };
    let conn = open_thread_db(&context)?;
    let binding = read_binding(&conn, &context.channel_ref)?;
    ensure_no_thread_contention(existing_session_thread_id, binding.as_ref())?;
    let mut record = resolve_target_record(
        &conn,
        &context,
        existing_session_thread_id,
        binding.as_ref(),
        session_id,
    )?;
    bind_thread(
        &conn,
        &context,
        &mut record,
        session_id,
        ThreadEventKind::BoundToChannel,
    )?;
    Ok(record.thread_id)
}

fn ensure_no_thread_contention(
    existing_session_thread_id: Option<&str>,
    binding: Option<&ChannelBinding>,
) -> Result<()> {
    if let (Some(session_thread_id), Some(binding)) = (existing_session_thread_id, binding) {
        if session_thread_id != binding.thread_id {
            bail!(
                "active session thread `{session_thread_id}` does not match current channel binding `{}`; use `ccd thread resume {session_thread_id}` or clear the session before switching threads",
                binding.thread_id
            );
        }
    }
    Ok(())
}

fn resolve_target_record(
    conn: &Connection,
    context: &ThreadContext,
    explicit_session_thread_id: Option<&str>,
    binding: Option<&ChannelBinding>,
    session_id: Option<&str>,
) -> Result<ThreadRecord> {
    if let Some(thread_id) = explicit_session_thread_id {
        return read_thread(conn, thread_id)?.ok_or_else(|| {
            anyhow::anyhow!("active session references missing thread `{thread_id}`")
        });
    }
    if let Some(binding) = binding {
        return read_thread(conn, &binding.thread_id)?.ok_or_else(|| {
            anyhow::anyhow!(
                "current channel binding references missing thread `{}`",
                binding.thread_id
            )
        });
    }
    ensure_legacy_thread(conn, context, session_id)
}

pub(crate) fn record_exported(
    layout: &StateLayout,
    locality_id: &str,
    thread_id: &str,
    metadata: JsonValue,
) -> Result<()> {
    let context = context_from_layout(layout, locality_id)?;
    let conn = open_thread_db(&context)?;
    record_event(
        &conn,
        thread_id,
        ThreadEventKind::Exported,
        Some(&context.channel_ref),
        None,
        metadata,
    )
}

pub(crate) fn record_imported(
    layout: &StateLayout,
    locality_id: &str,
    thread_id: &str,
    metadata: JsonValue,
) -> Result<()> {
    let context = context_from_layout(layout, locality_id)?;
    let conn = open_thread_db(&context)?;
    record_event(
        &conn,
        thread_id,
        ThreadEventKind::Imported,
        Some(&context.channel_ref),
        None,
        metadata,
    )
}

pub(crate) fn record_session_opened(
    layout: &StateLayout,
    locality_id: &str,
    thread_id: &str,
    session_id: &str,
    metadata: JsonValue,
) -> Result<()> {
    let context = context_from_layout(layout, locality_id)?;
    let conn = open_thread_db(&context)?;
    record_event(
        &conn,
        thread_id,
        ThreadEventKind::SessionOpened,
        Some(&context.channel_ref),
        Some(session_id),
        metadata,
    )
}

pub(crate) fn record_session_closed(
    layout: &StateLayout,
    locality_id: &str,
    thread_id: &str,
    session_id: &str,
    metadata: JsonValue,
) -> Result<()> {
    let context = context_from_layout(layout, locality_id)?;
    let conn = open_thread_db(&context)?;
    record_event(
        &conn,
        thread_id,
        ThreadEventKind::SessionClosed,
        Some(&context.channel_ref),
        Some(session_id),
        metadata,
    )
}

pub(crate) fn migrated_legacy_thread_id(
    repo_root: &Path,
    layout: &StateLayout,
    profile: &str,
    locality_id: &str,
) -> Result<String> {
    let context = ThreadContext {
        repo_root: repo_root.to_path_buf(),
        layout: layout.clone(),
        profile: profile.to_owned(),
        locality_id: locality_id.to_owned(),
        db_path: layout.repo_threads_db_path(locality_id)?,
        channel_ref: channel_ref(layout),
    };
    let conn = open_thread_db(&context)?;
    let record = ensure_legacy_thread(&conn, &context, None)?;
    Ok(record.thread_id)
}

fn mutation_report(
    command: &'static str,
    action: &'static str,
    context: ThreadContext,
    thread: ThreadDetailView,
) -> ThreadMutationReport {
    ThreadMutationReport {
        command,
        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,
        thread,
        action,
    }
}

fn resolve_context(
    repo_root: &Path,
    explicit_profile: Option<&str>,
    command: &str,
) -> Result<ThreadContext> {
    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 thread {}`",
            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()
        )
    };
    let db_path = layout.repo_threads_db_path(&marker.locality_id)?;
    Ok(ThreadContext {
        repo_root: repo_root.to_path_buf(),
        layout,
        profile: profile.to_string(),
        locality_id: marker.locality_id,
        db_path,
        channel_ref: channel_ref_for_path(repo_root, explicit_profile),
    })
}

fn context_from_layout(layout: &StateLayout, locality_id: &str) -> Result<ThreadContext> {
    Ok(ThreadContext {
        repo_root: PathBuf::new(),
        layout: layout.clone(),
        profile: layout.profile().to_string(),
        locality_id: locality_id.to_owned(),
        db_path: layout.repo_threads_db_path(locality_id)?,
        channel_ref: channel_ref(layout),
    })
}

fn open_thread_db(context: &ThreadContext) -> Result<Connection> {
    if let Some(parent) = context.db_path.parent() {
        std::fs::create_dir_all(parent)
            .with_context(|| format!("failed to create {}", parent.display()))?;
    }
    let conn = Connection::open(&context.db_path)
        .with_context(|| format!("failed to open {}", context.db_path.display()))?;
    let version: u32 = conn.pragma_query_value(None, "user_version", |row| row.get(0))?;
    if version > THREAD_SCHEMA_VERSION {
        bail!(
            "thread state database {} was created by a newer ccd schema version {}; this ccd supports up to {}",
            context.db_path.display(),
            version,
            THREAD_SCHEMA_VERSION
        );
    }
    conn.execute_batch("PRAGMA foreign_keys=ON;")?;
    conn.execute_batch(THREAD_SCHEMA)?;
    if version < THREAD_SCHEMA_VERSION {
        conn.pragma_update(None, "user_version", THREAD_SCHEMA_VERSION)?;
    }
    Ok(conn)
}

fn ensure_legacy_thread(
    conn: &Connection,
    context: &ThreadContext,
    session_id: Option<&str>,
) -> Result<ThreadRecord> {
    let legacy_ref = thread_transfer::stable_thread_id(&context.profile, &context.locality_id);
    if let Some(record) = read_thread_by_legacy(conn, &legacy_ref)? {
        if session_id.is_some() && record.last_session_id.as_deref() != session_id {
            update_last_session(conn, &record.thread_id, session_id)?;
        }
        return Ok(record);
    }

    let thread_id = format!("thread_{}", crate::ids::new_ulid());
    let display_name = unique_display_name(
        conn,
        &initial_display_name(context, &thread_id),
        Some(&thread_id),
    )?;
    let now = session_state::now_epoch_s()?;
    conn.execute(
        "INSERT INTO threads
            (thread_id, display_name, status, created_at_epoch_s, updated_at_epoch_s,
             last_session_id, legacy_thread_ref)
         VALUES (?1, ?2, 'open', ?3, ?3, ?4, ?5)",
        params![thread_id, display_name, now, session_id, legacy_ref],
    )?;
    record_event(
        conn,
        &thread_id,
        ThreadEventKind::Created,
        Some(&context.channel_ref),
        session_id,
        json!({ "legacy_thread_ref": legacy_ref }),
    )?;
    read_thread(conn, &thread_id)?.ok_or_else(|| anyhow::anyhow!("created thread vanished"))
}

fn bind_thread(
    conn: &Connection,
    context: &ThreadContext,
    record: &mut ThreadRecord,
    session_id: Option<&str>,
    event_kind: ThreadEventKind,
) -> Result<()> {
    let previous = read_binding(conn, &context.channel_ref)?;
    if previous
        .as_ref()
        .is_some_and(|binding| binding.thread_id != record.thread_id)
    {
        let previous = previous.expect("checked Some above");
        record_event(
            conn,
            &previous.thread_id,
            ThreadEventKind::UnboundFromChannel,
            Some(&context.channel_ref),
            previous.session_id.as_deref(),
            json!({ "next_thread_id": record.thread_id }),
        )?;
    }

    let now = session_state::now_epoch_s()?;
    conn.execute(
        "INSERT OR REPLACE INTO thread_channel_bindings
            (channel_ref, thread_id, bound_at_epoch_s, session_id)
         VALUES (?1, ?2, ?3, ?4)",
        params![context.channel_ref, record.thread_id, now, session_id],
    )?;
    conn.execute(
        "UPDATE threads
         SET status = 'open', updated_at_epoch_s = ?1, last_session_id = COALESCE(?2, last_session_id)
         WHERE thread_id = ?3",
        params![now, session_id, record.thread_id],
    )?;
    record.status = ThreadRecordStatus::Open;
    record.updated_at_epoch_s = now;
    if let Some(session_id) = session_id {
        record.last_session_id = Some(session_id.to_owned());
    }
    record_event(
        conn,
        &record.thread_id,
        event_kind,
        Some(&context.channel_ref),
        session_id,
        json!({}),
    )
}

fn resolve_thread(conn: &Connection, selector: &str) -> Result<ThreadRecord> {
    if let Some(record) = read_thread(conn, selector)? {
        return Ok(record);
    }
    let matches = read_threads_by_display_name(conn, selector)?;
    match matches.len() {
        0 => bail!("thread `{selector}` was not found"),
        1 => Ok(matches.into_iter().next().expect("one item")),
        _ => bail!("thread name `{selector}` is ambiguous; use the canonical thread_id instead"),
    }
}

fn read_threads(conn: &Connection) -> Result<Vec<ThreadRecord>> {
    let mut stmt = conn.prepare(SELECT_THREADS_ORDERED_SQL)?;
    let rows = stmt.query_map([], thread_record_from_row)?;
    rows.collect::<std::result::Result<Vec<_>, _>>()
        .map_err(Into::into)
}

fn read_thread(conn: &Connection, thread_id: &str) -> Result<Option<ThreadRecord>> {
    conn.query_row(
        SELECT_THREAD_BY_ID_SQL,
        params![thread_id],
        thread_record_from_row,
    )
    .optional()
    .map_err(Into::into)
}

fn read_thread_by_legacy(conn: &Connection, legacy_ref: &str) -> Result<Option<ThreadRecord>> {
    conn.query_row(
        SELECT_THREAD_BY_LEGACY_SQL,
        params![legacy_ref],
        thread_record_from_row,
    )
    .optional()
    .map_err(Into::into)
}

fn read_threads_by_display_name(conn: &Connection, name: &str) -> Result<Vec<ThreadRecord>> {
    let mut stmt = conn.prepare(SELECT_THREADS_BY_DISPLAY_NAME_SQL)?;
    let rows = stmt.query_map(params![name], thread_record_from_row)?;
    rows.collect::<std::result::Result<Vec<_>, _>>()
        .map_err(Into::into)
}

fn thread_record_from_row(row: &rusqlite::Row<'_>) -> rusqlite::Result<ThreadRecord> {
    let status_text: String = row.get(2)?;
    let status = ThreadRecordStatus::from_db_str(&status_text).ok_or_else(|| {
        rusqlite::Error::FromSqlConversionFailure(
            2,
            rusqlite::types::Type::Text,
            Box::new(std::io::Error::new(
                std::io::ErrorKind::InvalidData,
                format!("unexpected thread record status `{status_text}`"),
            )),
        )
    })?;
    Ok(ThreadRecord {
        thread_id: row.get(0)?,
        display_name: row.get(1)?,
        status,
        created_at_epoch_s: row.get(3)?,
        updated_at_epoch_s: row.get(4)?,
        last_session_id: row.get(5)?,
        legacy_thread_ref: row.get(6)?,
    })
}

fn read_binding(conn: &Connection, channel_ref: &str) -> Result<Option<ChannelBinding>> {
    conn.query_row(
        "SELECT channel_ref, thread_id, bound_at_epoch_s, session_id
         FROM thread_channel_bindings
         WHERE channel_ref = ?1",
        params![channel_ref],
        |row| {
            Ok(ChannelBinding {
                thread_id: row.get(1)?,
                bound_at_epoch_s: row.get(2)?,
                session_id: row.get(3)?,
            })
        },
    )
    .optional()
    .map_err(Into::into)
}

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

fn record_event(
    conn: &Connection,
    thread_id: &str,
    event_kind: ThreadEventKind,
    channel_ref: Option<&str>,
    session_id: Option<&str>,
    metadata: JsonValue,
) -> Result<()> {
    let at = session_state::now_epoch_s()?;
    conn.execute(
        "INSERT INTO thread_events
            (thread_id, event_kind, at_epoch_s, channel_ref, session_id, metadata_json)
         VALUES (?1, ?2, ?3, ?4, ?5, ?6)",
        params![
            thread_id,
            event_kind.as_str(),
            at,
            channel_ref,
            session_id,
            serde_json::to_string(&metadata)?
        ],
    )?;
    Ok(())
}

fn update_last_session(conn: &Connection, thread_id: &str, session_id: Option<&str>) -> Result<()> {
    let Some(session_id) = session_id else {
        return Ok(());
    };
    let now = session_state::now_epoch_s()?;
    conn.execute(
        "UPDATE threads SET last_session_id = ?1, updated_at_epoch_s = ?2 WHERE thread_id = ?3",
        params![session_id, now, thread_id],
    )?;
    Ok(())
}

fn normalize_display_name(value: &str) -> Result<String> {
    let normalized = value.trim();
    if normalized.is_empty() {
        bail!("thread display name cannot be empty");
    }
    if normalized.contains('\0') {
        bail!("thread display name cannot contain NUL bytes");
    }
    Ok(normalized.to_owned())
}

fn ensure_display_name_available(
    conn: &Connection,
    display_name: &str,
    excluding_thread_id: Option<&str>,
) -> Result<()> {
    let matches = read_threads_by_display_name(conn, display_name)?;
    let conflicting = matches.into_iter().any(|record| {
        excluding_thread_id.is_none_or(|thread_id| record.thread_id != thread_id)
            && record.status != ThreadRecordStatus::Closed
    });
    if conflicting {
        bail!("thread display name `{display_name}` is already in use");
    }
    Ok(())
}

fn unique_display_name(
    conn: &Connection,
    base: &str,
    excluding_thread_id: Option<&str>,
) -> Result<String> {
    let base = normalize_display_name(base).unwrap_or_else(|_| "Thread".to_owned());
    let mut candidate = base.clone();
    let mut suffix = 2_u32;
    loop {
        if ensure_display_name_available(conn, &candidate, excluding_thread_id).is_ok() {
            return Ok(candidate);
        }
        candidate = format!("{base} ({suffix})");
        suffix += 1;
    }
}

fn initial_display_name(context: &ThreadContext, thread_id: &str) -> String {
    runtime_state::load_runtime_state(&context.repo_root, &context.layout, &context.locality_id)
        .ok()
        .and_then(|runtime| {
            let title = runtime.state.handoff.title.trim();
            (!title.is_empty()).then(|| title.to_owned())
        })
        .or_else(|| {
            context
                .repo_root
                .file_name()
                .and_then(|name| name.to_str())
                .map(|name| format!("{name} thread"))
        })
        .unwrap_or_else(|| {
            let short = thread_id
                .strip_prefix("thread_")
                .unwrap_or(thread_id)
                .chars()
                .take(8)
                .collect::<String>();
            format!("Thread {short}")
        })
}

fn load_active_session(layout: &StateLayout) -> Result<Option<session_state::SessionStateFile>> {
    let Some(state) = session_state::load_for_layout(layout)? else {
        return Ok(None);
    };
    let now = session_state::now_epoch_s()?;
    if session_state::is_stale(&state, now) || state.session_id.is_none() {
        return Ok(None);
    }
    Ok(Some(state))
}

fn list_item(record: &ThreadRecord, binding: Option<&ChannelBinding>) -> ThreadListItem {
    let bound_to_current_channel =
        binding.is_some_and(|binding| binding.thread_id == record.thread_id);
    ThreadListItem {
        thread_id: record.thread_id.clone(),
        display_name: record.display_name.clone(),
        status: derived_status(record, bound_to_current_channel),
        record_status: record.status,
        created_at_epoch_s: record.created_at_epoch_s,
        updated_at_epoch_s: record.updated_at_epoch_s,
        last_session_id: record.last_session_id.clone(),
        legacy_thread_ref: record.legacy_thread_ref.clone(),
        bound_to_current_channel,
    }
}

fn detail_view(record: &ThreadRecord, binding: Option<&ChannelBinding>) -> ThreadDetailView {
    let bound_to_current_channel =
        binding.is_some_and(|binding| binding.thread_id == record.thread_id);
    ThreadDetailView {
        thread_id: record.thread_id.clone(),
        display_name: record.display_name.clone(),
        status: derived_status(record, bound_to_current_channel),
        record_status: record.status,
        created_at_epoch_s: record.created_at_epoch_s,
        updated_at_epoch_s: record.updated_at_epoch_s,
        last_session_id: record.last_session_id.clone(),
        legacy_thread_ref: record.legacy_thread_ref.clone(),
        binding: binding_view(binding),
    }
}

fn state_view(
    context: ThreadContext,
    record: Option<&ThreadRecord>,
    binding: Option<&ChannelBinding>,
    session_thread_id: Option<String>,
) -> ThreadStateView {
    let binding_thread_id = binding.map(|value| value.thread_id.clone());
    let active_session_matches = match (session_thread_id.as_ref(), binding_thread_id.as_ref()) {
        (Some(session), Some(binding)) => Some(session == binding),
        (Some(_), None) => Some(false),
        _ => None,
    };
    let status = match (record, binding, active_session_matches) {
        (_, _, Some(false)) => ThreadStateStatus::Mismatch,
        (Some(record), Some(binding), _) if record.thread_id == binding.thread_id => {
            ThreadStateStatus::Bound
        }
        (Some(_), Some(_), _) => ThreadStateStatus::Mismatch,
        (Some(_), None, _) => ThreadStateStatus::Unbound,
        (None, Some(_), _) => ThreadStateStatus::DanglingBinding,
        (None, None, _) => ThreadStateStatus::Missing,
    };
    let mut warnings = Vec::new();
    match status {
        ThreadStateStatus::Mismatch => warnings
            .push("active session thread_id and current channel binding disagree".to_owned()),
        ThreadStateStatus::DanglingBinding => {
            warnings.push("current channel binding references a missing thread record".to_owned())
        }
        _ => {}
    }
    ThreadStateView {
        path: context.db_path.display().to_string(),
        status,
        channel_ref: context.channel_ref,
        thread_id: record
            .map(|record| record.thread_id.clone())
            .or(binding_thread_id),
        display_name: record.map(|record| record.display_name.clone()),
        record_status: record.map(|record| record.status),
        session_thread_id,
        binding: binding_view(binding),
        active_session_matches,
        warnings,
    }
}

fn missing_state(context: ThreadContext, session_thread_id: Option<String>) -> ThreadStateView {
    ThreadStateView {
        path: context.db_path.display().to_string(),
        status: ThreadStateStatus::Missing,
        channel_ref: context.channel_ref,
        thread_id: None,
        display_name: None,
        record_status: None,
        session_thread_id,
        binding: ThreadBindingView {
            status: ThreadBindingStatus::Missing,
            thread_id: None,
            bound_at_epoch_s: None,
            session_id: None,
        },
        active_session_matches: None,
        warnings: Vec::new(),
    }
}

fn binding_view(binding: Option<&ChannelBinding>) -> ThreadBindingView {
    match binding {
        Some(binding) => ThreadBindingView {
            status: ThreadBindingStatus::Bound,
            thread_id: Some(binding.thread_id.clone()),
            bound_at_epoch_s: Some(binding.bound_at_epoch_s),
            session_id: binding.session_id.clone(),
        },
        None => ThreadBindingView {
            status: ThreadBindingStatus::Missing,
            thread_id: None,
            bound_at_epoch_s: None,
            session_id: None,
        },
    }
}

fn derived_status(record: &ThreadRecord, bound_to_current_channel: bool) -> ThreadDerivedStatus {
    if record.status == ThreadRecordStatus::Closed {
        ThreadDerivedStatus::Closed
    } else if bound_to_current_channel {
        ThreadDerivedStatus::Active
    } else {
        ThreadDerivedStatus::Dormant
    }
}

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

fn channel_ref_for_path(repo_root: &Path, explicit_profile: Option<&str>) -> String {
    match profile::resolve(explicit_profile)
        .and_then(|profile| StateLayout::resolve(repo_root, profile))
    {
        Ok(layout) => channel_ref(&layout),
        Err(_) => format!("path:{}", repo_root.display()),
    }
}
