//! `tasks` table operations.
//!
//! A committed task is a `task` node plus a `tasks` row. On creation it also
//! gets a canonical context `doc` and a `canonical-context` link (tech-spec §6).

use std::collections::HashMap;

use rusqlite::{Connection, OptionalExtension, Row};

use serde_json::json;

use super::{links, log, next_hlc, nodes, ops};
use crate::error::{Error, Result};
use crate::extract;
use crate::filter::ListFilter;
use crate::model::{
    Attention, Health, LinkType, NewTask, NodeKind, ProjectOverview, SchedulePatch, Task, TaskState,
};
use crate::oplog::op_type;
use crate::ranking::{self, RankedTask};
use crate::recurrence;

/// JSON payload of a task's scalar fields (for `task.create`/`task.set` ops).
fn scalar_payload(t: &Task) -> serde_json::Value {
    json!({
        "attention": t.attention.map(|a| a.as_str()),
        "do_date": t.do_date,
        "late_on": t.late_on,
        "state": t.state.as_str(),
        "recurrence": t.recurrence,
    })
}

/// Bump the task node's hlc/modified_at and record a `task.set` op snapshotting
/// the task's current scalars (LWW unit, tech-spec §12).
fn record_set(conn: &Connection, owner: &str, now: i64, node_id: &str) -> Result<()> {
    let task = require(conn, node_id)?;
    let hlc = next_hlc(conn, now)?;
    conn.execute(
        "UPDATE nodes SET modified_at = ?1, hlc = ?2 WHERE id = ?3",
        (now, &hlc, node_id),
    )?;
    ops::record(
        conn,
        owner,
        &hlc,
        op_type::TASK_SET,
        node_id,
        scalar_payload(&task),
    )?;
    Ok(())
}

fn from_row(row: &Row) -> rusqlite::Result<Task> {
    let attention = match row.get::<_, Option<String>>("attention")? {
        Some(s) => Some(
            Attention::parse(&s)
                .map_err(|e| rusqlite::Error::ToSqlConversionFailure(Box::new(e)))?,
        ),
        None => None,
    };
    Ok(Task {
        node_id: row.get("node_id")?,
        attention,
        do_date: row.get("do_date")?,
        late_on: row.get("late_on")?,
        state: TaskState::parse(&row.get::<_, String>("state")?)
            .map_err(|e| rusqlite::Error::ToSqlConversionFailure(Box::new(e)))?,
        recurrence: row.get("recurrence")?,
    })
}

const COLUMNS: &str = "node_id, attention, do_date, late_on, state, recurrence";

/// Create a committed task: its task node, the `tasks` row, the canonical
/// context doc, the `canonical-context` link, and (if given) an `in-project`
/// link — all in one transaction.
pub(super) fn create(conn: &mut Connection, owner: &str, now: i64, input: NewTask) -> Result<Task> {
    let task = Task {
        node_id: String::new(), // filled below
        attention: input.attention,
        do_date: input.do_date,
        late_on: input.late_on,
        state: TaskState::Outstanding,
        recurrence: input.recurrence,
    };

    let tx = conn.transaction()?;

    let task_hlc = next_hlc(&tx, now)?;
    let task_node = nodes::build(
        owner,
        now,
        &task_hlc,
        NodeKind::Task,
        input.title.clone(),
        None,
    );
    nodes::insert(&tx, &task_node)?;
    nodes::record_create(&tx, owner, &task_node)?;
    tx.execute(
        "INSERT INTO tasks (node_id, attention, do_date, late_on, state, recurrence)
         VALUES (?1, ?2, ?3, ?4, ?5, ?6)",
        (
            &task_node.id,
            task.attention.map(|a| a.as_str()),
            task.do_date,
            task.late_on,
            task.state.as_str(),
            &task.recurrence,
        ),
    )?;
    let task = Task {
        node_id: task_node.id.clone(),
        ..task
    };
    let task_create_hlc = next_hlc(&tx, now)?;
    ops::record(
        &tx,
        owner,
        &task_create_hlc,
        op_type::TASK_CREATE,
        &task.node_id,
        scalar_payload(&task),
    )?;

    // The canonical context doc (the task's jumping-off point / checklist body).
    let doc_hlc = next_hlc(&tx, now)?;
    let doc = nodes::build(
        owner,
        now,
        &doc_hlc,
        NodeKind::Doc,
        input.title.clone(),
        Some(String::new()),
    );
    nodes::insert(&tx, &doc)?;
    nodes::record_create(&tx, owner, &doc)?;
    links::add(
        &tx,
        owner,
        now,
        &task.node_id,
        &doc.id,
        LinkType::CanonicalContext,
    )?;

    if let Some(project_id) = &input.project_id {
        links::add(
            &tx,
            owner,
            now,
            &task.node_id,
            project_id,
            LinkType::InProject,
        )?;
    }

    tx.commit()?;

    Ok(task)
}

/// Promote a `- [ ]` context-item line in `container_id`'s body into a committed
/// task, rewriting that source line into a `[[link]]` to the new task (Fork A,
/// tech-spec §4.3, §6). `item_ref` is the **1-based index** of the item among
/// the container's context items in document order (code-fence-aware, matching
/// extraction).
pub(super) fn promote(
    conn: &mut Connection,
    owner: &str,
    now: i64,
    container_id: &str,
    item_ref: usize,
    attention: Option<Attention>,
    project_id: Option<String>,
) -> Result<Task> {
    let container =
        nodes::get(conn, container_id)?.ok_or_else(|| Error::NodeNotFound(container_id.into()))?;
    let body = container.body.unwrap_or_default();

    let idx = item_ref
        .checked_sub(1)
        .ok_or_else(|| Error::Integrity("item_ref is 1-based".into()))?;
    let item = extract::extract(&body)
        .context_items
        .into_iter()
        .nth(idx)
        .ok_or_else(|| Error::Integrity(format!("no context item #{item_ref} to promote")))?;
    let line = *extract::context_item_lines(&body)
        .get(idx)
        .ok_or_else(|| Error::Integrity(format!("no context item #{item_ref} to promote")))?;
    let title = item.text.trim().to_string();
    if title.is_empty() {
        return Err(Error::Integrity(
            "cannot promote an empty context item".into(),
        ));
    }

    // Mint the committed task (its own node + canonical context doc + link).
    let task = create(
        conn,
        owner,
        now,
        NewTask {
            title: title.clone(),
            attention,
            project_id,
            ..Default::default()
        },
    )?;

    // Rewrite the source line into a wiki-link to the new task. Updating the
    // container re-runs extraction, materializing the container→task `wiki` link
    // and dropping the now-promoted context item.
    let new_body = rewrite_line(&body, line, &format!("- [[{title}]]"));
    nodes::update(conn, owner, now, container_id, None, Some(new_body))?;

    Ok(task)
}

/// Replace body line `idx` (0-based) with `new_line`, preserving the original
/// line's leading whitespace. An out-of-range `idx` leaves the body unchanged.
fn rewrite_line(body: &str, idx: usize, new_line: &str) -> String {
    let mut lines: Vec<String> = body.split('\n').map(str::to_string).collect();
    if let Some(slot) = lines.get_mut(idx) {
        let indent: String = slot.chars().take_while(|c| c.is_whitespace()).collect();
        *slot = format!("{indent}{new_line}");
    }
    lines.join("\n")
}

/// Fetch a task by node id.
pub(super) fn get(conn: &Connection, node_id: &str) -> Result<Option<Task>> {
    let task = conn
        .query_row(
            &format!("SELECT {COLUMNS} FROM tasks WHERE node_id = ?1"),
            [node_id],
            from_row,
        )
        .optional()?;
    Ok(task)
}

fn require(conn: &Connection, node_id: &str) -> Result<Task> {
    get(conn, node_id)?.ok_or_else(|| Error::NodeNotFound(node_id.to_string()))
}

/// Set a task's lifecycle state. Completing a **recurring** task rolls it
/// forward in place (tech-spec §4.4) rather than marking it done.
pub(super) fn set_state(
    conn: &mut Connection,
    owner: &str,
    now: i64,
    node_id: &str,
    state: TaskState,
) -> Result<Task> {
    let task = require(conn, node_id)?;
    if state == TaskState::Done && task.recurrence.is_some() {
        return roll_forward(conn, owner, now, &task);
    }
    conn.execute(
        "UPDATE tasks SET state = ?1 WHERE node_id = ?2",
        (state.as_str(), node_id),
    )?;
    record_set(conn, owner, now, node_id)?;
    require(conn, node_id)
}

/// Roll a recurring task forward on completion (tech-spec §4.4): reset its
/// checklist to all-unchecked, log the occurrence, and advance the do-date to
/// the next RRULE instance strictly after `now` (skipping misses) — all in one
/// transaction. If the series is exhausted, the task is finally marked done.
fn roll_forward(conn: &mut Connection, owner: &str, now: i64, task: &Task) -> Result<Task> {
    let rrule = task
        .recurrence
        .as_deref()
        .expect("roll_forward called on a recurring task");
    let tx = conn.transaction()?;

    // 1. Fresh checklist — reset the canonical context doc's checkboxes.
    if let Some(doc_id) = links::first_dst(&tx, &task.node_id, LinkType::CanonicalContext)? {
        if let Some(doc) = nodes::get(&tx, &doc_id)? {
            let body = doc.body.unwrap_or_default();
            let reset = recurrence::reset_checkboxes(&body);
            if reset != body {
                nodes::rewrite_body_local(&tx, now, &doc_id, &reset)?;
                links::sync_wiki_links(&tx, owner, &doc_id, &reset, now)?;
            }
        }
    }

    // 2. Narrative history — append the completed occurrence to the log.
    let entry = match task.do_date {
        Some(d) => format!("- completed occurrence (do-date {d})"),
        None => "- completed occurrence".to_string(),
    };
    log::append(&tx, owner, now, &task.node_id, &entry)?;

    // 3. Advance the do-date (or finally finish a finite series).
    advance(&tx, now, &task.node_id, rrule, task.do_date)?;
    record_set(&tx, owner, now, &task.node_id)?;

    tx.commit()?;
    require(conn, &task.node_id)
}

/// Advance a recurring task to its next instance after `now`, or mark it `done`
/// if the series is exhausted. Shared by completion roll-forward and `skip`.
fn advance(
    conn: &Connection,
    now: i64,
    node_id: &str,
    rrule: &str,
    do_date: Option<i64>,
) -> Result<()> {
    let anchor = do_date.unwrap_or(now);
    match recurrence::next_occurrence(rrule, anchor, now)? {
        Some(next) => {
            conn.execute(
                "UPDATE tasks SET do_date = ?1, state = 'outstanding' WHERE node_id = ?2",
                (next, node_id),
            )?;
        }
        None => {
            conn.execute(
                "UPDATE tasks SET state = 'done' WHERE node_id = ?1",
                [node_id],
            )?;
        }
    }
    Ok(())
}

/// Skip the current occurrence of a recurring task: advance the do-date the same
/// way as completion but **without** logging a completion (tech-spec §4.4).
pub(super) fn skip(conn: &Connection, owner: &str, now: i64, node_id: &str) -> Result<Task> {
    let task = require(conn, node_id)?;
    let rrule = task
        .recurrence
        .as_deref()
        .ok_or_else(|| Error::Integrity(format!("skip on non-recurring task {node_id}")))?;
    advance(conn, now, node_id, rrule, task.do_date)?;
    record_set(conn, owner, now, node_id)?;
    require(conn, node_id)
}

/// The Tactical "what is next?" ranking for `owner` at `now` (tech-spec §7).
pub(super) fn next(
    conn: &Connection,
    owner: &str,
    now: i64,
    scope: Option<&str>,
    limit: usize,
) -> Result<Vec<RankedTask>> {
    let candidates = load_candidates(conn, owner)?;
    Ok(ranking::rank(candidates, now, scope, limit))
}

/// Enumerate outstanding committed tasks matching `filter` (tech-spec §8.2),
/// as **titled rows** ([`RankedTask`] shape — the same the plugin renders for
/// `next`, so the survey view needs no N+1 `node.get`). An empty
/// [`ListFilter`] yields the whole outstanding set (the Organizational survey,
/// tech-spec §6). `now` feeds the `actionable` do-date gate.
pub(super) fn list(
    conn: &Connection,
    owner: &str,
    now: i64,
    filter: &ListFilter,
) -> Result<Vec<RankedTask>> {
    let sql = "
        SELECT n.id, n.title, n.created_at, n.tombstoned,
               t.attention, t.do_date, t.late_on, t.state, t.recurrence,
               (SELECT dst_id FROM links
                  WHERE src_id = n.id AND type = 'in-project' AND tombstoned = 0
                  ORDER BY created_at, id LIMIT 1) AS project_id,
               (SELECT dst_id FROM links
                  WHERE src_id = n.id AND type = 'canonical-context' AND tombstoned = 0
                  ORDER BY created_at, id LIMIT 1) AS ctx_id
        FROM tasks t JOIN nodes n ON n.id = t.node_id
        WHERE n.owner_id = ?1 AND n.tombstoned = 0 AND t.state = 'outstanding'
        ORDER BY n.created_at, n.id";
    let mut stmt = conn.prepare(sql)?;
    let rows = stmt.query_map([owner], ranked_from_row)?;

    let mut out = Vec::new();
    for row in rows {
        let task = row?;
        if filter.matches(&task, now) {
            out.push(task);
        }
    }
    Ok(out)
}

/// Run a built-in filter view by name (tech-spec §8.2): resolve its project
/// names to ids (each subtree-expanded), build the [`ListFilter`], and list.
/// A view whose `scope_names` all fail to resolve yields no rows (the projects
/// don't exist), rather than silently widening to "any project".
pub(super) fn view(
    conn: &Connection,
    owner: &str,
    now: i64,
    name: &str,
) -> Result<Vec<RankedTask>> {
    let spec = crate::filter::builtin(name)
        .ok_or_else(|| Error::InvalidArg(format!("unknown view {name:?}")))?;

    let scope = resolve_project_names(conn, owner, spec.scope_names)?;
    if !spec.scope_names.is_empty() && scope.is_empty() {
        return Ok(Vec::new());
    }
    let exclude_projects = resolve_project_names(conn, owner, spec.exclude_names)?;

    let filter = ListFilter {
        attention_in: spec.attention_in.to_vec(),
        attention_not: spec.attention_not.to_vec(),
        scope,
        exclude_projects,
        actionable: spec.actionable,
        unfiled: spec.unfiled,
    };
    list(conn, owner, now, &filter)
}

/// Resolve project `names` to a deduped set of node ids, each expanded to its
/// project subtree. Names that don't resolve are skipped (a view tolerates a
/// store missing some of its projects).
fn resolve_project_names(conn: &Connection, owner: &str, names: &[&str]) -> Result<Vec<String>> {
    let mut ids = Vec::new();
    for name in names {
        if let Some(id) = links::resolve_project_id(conn, owner, name)? {
            for sub in links::project_subtree(conn, &id)? {
                if !ids.contains(&sub) {
                    ids.push(sub);
                }
            }
        }
    }
    Ok(ids)
}

/// Resolve a single project NAME to its scope: the project id plus its subtree
/// (parent→child). Errors if the name names no project, so `--project Foo` fails
/// loudly rather than silently widening to "everything".
pub(super) fn project_scope(conn: &Connection, owner: &str, name: &str) -> Result<Vec<String>> {
    let scope = resolve_project_names(conn, owner, &[name])?;
    if scope.is_empty() {
        return Err(Error::InvalidArg(format!("no project named {name:?}")));
    }
    Ok(scope)
}

/// Working-set health counts (tech-spec §7) — surfaced honestly.
pub(super) fn health(conn: &Connection, owner: &str) -> Result<Health> {
    let mut stmt = conn.prepare(
        "SELECT t.attention FROM tasks t JOIN nodes n ON n.id = t.node_id
         WHERE n.owner_id = ?1 AND n.tombstoned = 0 AND t.state = 'outstanding'",
    )?;
    let attentions: Vec<Option<String>> = stmt
        .query_map([owner], |r| r.get(0))?
        .collect::<rusqlite::Result<Vec<_>>>()?;

    let mut orange_count = 0;
    let mut active_count = 0;
    let mut on_deck_count = 0;
    for a in attentions.iter().flatten() {
        match a.as_str() {
            "orange" => {
                orange_count += 1;
                active_count += 1;
            }
            "red" | "white" => active_count += 1,
            "blue" => on_deck_count += 1,
            _ => {}
        }
    }
    Ok(Health {
        orange_count,
        active_count,
        on_deck_count,
        conflict_count: 0,
        sync_status: "local".to_string(),
    })
}

/// Every project (owner-scoped, non-tombstoned) with its parent project (via a
/// `parent` link) and its direct outstanding-task count — the shape a sidebar
/// renders as a counted, indented tree (§8.1). Pure read-side: counts come from
/// a single `GROUP BY` over the `in-project` links, parents from the `parent`
/// links, both already in the store. Title-sorted for a stable sibling order.
pub(super) fn project_overview(conn: &Connection, owner: &str) -> Result<Vec<ProjectOverview>> {
    // Direct outstanding count per project: each task's project is its first
    // `in-project` link target (mirrors `list`/`load_candidates`).
    let mut count_stmt = conn.prepare(
        "SELECT (SELECT dst_id FROM links
                  WHERE src_id = n.id AND type = 'in-project' AND tombstoned = 0
                  ORDER BY created_at, id LIMIT 1) AS project_id,
                COUNT(*)
         FROM nodes n JOIN tasks t ON t.node_id = n.id
         WHERE n.owner_id = ?1 AND n.tombstoned = 0 AND t.state = 'outstanding'
         GROUP BY project_id",
    )?;
    let mut counts: HashMap<String, usize> = HashMap::new();
    let rows = count_stmt.query_map([owner], |r| {
        Ok((r.get::<_, Option<String>>(0)?, r.get::<_, i64>(1)?))
    })?;
    for row in rows {
        let (project_id, count) = row?;
        if let Some(pid) = project_id {
            counts.insert(pid, count as usize);
        }
    }

    // Parent of each project: the dst of its (first) `parent` link.
    let mut parent_stmt = conn.prepare(
        "SELECT dst_id FROM links
         WHERE src_id = ?1 AND type = 'parent' AND tombstoned = 0
         ORDER BY created_at, id LIMIT 1",
    )?;

    let mut out = Vec::new();
    for node in nodes::list(conn, owner, Some(NodeKind::Project))? {
        let parent_id = parent_stmt
            .query_row([&node.id], |r| r.get::<_, String>(0))
            .optional()?;
        out.push(ProjectOverview {
            outstanding: counts.get(&node.id).copied().unwrap_or(0),
            id: node.id,
            title: node.title,
            parent_id,
        });
    }
    out.sort_by(|a, b| a.title.cmp(&b.title));
    Ok(out)
}

/// Load every non-tombstoned committed task for `owner` as a ranking candidate,
/// joining in its project and canonical-context link targets.
fn load_candidates(conn: &Connection, owner: &str) -> Result<Vec<RankedTask>> {
    let sql = "
        SELECT n.id, n.title, n.created_at, n.tombstoned,
               t.attention, t.do_date, t.late_on, t.state, t.recurrence,
               (SELECT dst_id FROM links
                  WHERE src_id = n.id AND type = 'in-project' AND tombstoned = 0
                  ORDER BY created_at, id LIMIT 1) AS project_id,
               (SELECT dst_id FROM links
                  WHERE src_id = n.id AND type = 'canonical-context' AND tombstoned = 0
                  ORDER BY created_at, id LIMIT 1) AS ctx_id
        FROM tasks t JOIN nodes n ON n.id = t.node_id
        WHERE n.owner_id = ?1 AND n.tombstoned = 0";
    let mut stmt = conn.prepare(sql)?;
    let rows = stmt.query_map([owner], ranked_from_row)?;
    Ok(rows.collect::<rusqlite::Result<Vec<_>>>()?)
}

/// Map a row selected with the `id, title, created_at, tombstoned, attention,
/// do_date, late_on, state, project_id, ctx_id` shape into a [`RankedTask`].
/// Shared by `next`'s candidate load and the enriched `list`.
fn ranked_from_row(row: &Row) -> rusqlite::Result<RankedTask> {
    let attention = match row.get::<_, Option<String>>("attention")? {
        Some(s) => Some(
            Attention::parse(&s)
                .map_err(|e| rusqlite::Error::ToSqlConversionFailure(Box::new(e)))?,
        ),
        None => None,
    };
    Ok(RankedTask {
        node_id: row.get("id")?,
        title: row.get("title")?,
        attention,
        do_date: row.get("do_date")?,
        late_on: row.get("late_on")?,
        state: TaskState::parse(&row.get::<_, String>("state")?)
            .map_err(|e| rusqlite::Error::ToSqlConversionFailure(Box::new(e)))?,
        recurrence: row.get("recurrence")?,
        tombstoned: row.get::<_, i64>("tombstoned")? != 0,
        project_id: row.get("project_id")?,
        canonical_context_id: row.get("ctx_id")?,
        created_at: row.get("created_at")?,
    })
}

/// Set a task's attention-state.
pub(super) fn set_attention(
    conn: &Connection,
    owner: &str,
    now: i64,
    node_id: &str,
    attention: Attention,
) -> Result<Task> {
    let updated = conn.execute(
        "UPDATE tasks SET attention = ?1 WHERE node_id = ?2",
        (attention.as_str(), node_id),
    )?;
    if updated == 0 {
        return Err(Error::NodeNotFound(node_id.to_string()));
    }
    record_set(conn, owner, now, node_id)?;
    require(conn, node_id)
}

/// Re-file a task under `project_id`, or unfile it entirely when `None`
/// (tech-spec §8.1 move-to-project). OR-set link semantics: tombstone the
/// task's existing `in-project` links, then add a fresh one if a project is
/// given. A given `project_id` must name a live `project`-kind node. Records
/// only link ops (no task-scalar change), all in one transaction.
pub(super) fn set_project(
    conn: &mut Connection,
    owner: &str,
    now: i64,
    node_id: &str,
    project_id: Option<&str>,
) -> Result<Task> {
    require(conn, node_id)?; // task must exist
    if let Some(pid) = project_id {
        let project = nodes::get(conn, pid)?.ok_or_else(|| Error::NodeNotFound(pid.into()))?;
        if project.tombstoned || project.kind != NodeKind::Project {
            return Err(Error::InvalidArg(format!("{pid} is not a project node")));
        }
    }

    let tx = conn.transaction()?;
    for link in links::outgoing(&tx, node_id)? {
        if link.link_type == LinkType::InProject {
            links::tombstone(&tx, owner, now, &link.id)?;
        }
    }
    if let Some(pid) = project_id {
        links::add(&tx, owner, now, node_id, pid, LinkType::InProject)?;
    }
    tx.commit()?;

    require(conn, node_id)
}

/// Delete a project: **unfile every task currently filed under it** (tombstone
/// the `in-project` links, so those tasks fall to the Inbox), then tombstone the
/// project node itself — atomically. Tasks are preserved, never deleted.
pub(super) fn delete_project(
    conn: &mut Connection,
    owner: &str,
    now: i64,
    project_id: &str,
) -> Result<()> {
    let project =
        nodes::get(conn, project_id)?.ok_or_else(|| Error::NodeNotFound(project_id.into()))?;
    if project.kind != NodeKind::Project {
        return Err(Error::InvalidArg(format!(
            "{project_id} is not a project node"
        )));
    }

    let tx = conn.transaction()?;
    for link in links::backlinks(&tx, project_id)? {
        if link.link_type == LinkType::InProject {
            links::tombstone(&tx, owner, now, &link.id)?;
        }
    }
    nodes::tombstone(&tx, owner, now, project_id)?;
    tx.commit()?;
    Ok(())
}

/// Apply a partial schedule update (do-date / late-on / recurrence) — the
/// "reschedule" path (tech-spec §6). Reads the current row, overlays the
/// present `patch` fields (a double-option per field: absent = leave, `null` =
/// clear, value = set), writes all three columns, and records the LWW op.
pub(super) fn set_schedule(
    conn: &Connection,
    owner: &str,
    now: i64,
    node_id: &str,
    patch: SchedulePatch,
) -> Result<Task> {
    let mut task = require(conn, node_id)?;
    if let Some(v) = patch.do_date {
        task.do_date = v;
    }
    if let Some(v) = patch.late_on {
        task.late_on = v;
    }
    if let Some(v) = patch.recurrence {
        task.recurrence = v;
    }
    conn.execute(
        "UPDATE tasks SET do_date = ?1, late_on = ?2, recurrence = ?3 WHERE node_id = ?4",
        (&task.do_date, &task.late_on, &task.recurrence, node_id),
    )?;
    record_set(conn, owner, now, node_id)?;
    require(conn, node_id)
}