//! Dependency-graph repository methods for `SqliteTaskRepository`.
//!
//! `task_dependencies` holds the edges and is the only stored truth. Everything
//! a caller reads about blocking (readiness, depth, downstream count, cycles) is
//! derived from those edges plus task statuses, and cached back onto the task row
//! by [`recompute`].
//!
//! # Why the traversals are in Rust and not in SQL
//!
//! The obvious implementation is a recursive CTE per question. Three things
//! argue against it here. The depth we want is the LONGEST path, and a recursive
//! CTE naturally produces every path, so getting the longest means materialising
//! all of them and taking a max, which is exponential on a diamond-shaped graph.
//! Cycle handling in a recursive CTE is a depth guard, which silently truncates
//! rather than reporting the cycle we need to surface. And every write recomputes
//! the whole user graph anyway, so the work is one pass over a few thousand rows
//! either way, and the pass that is readable wins.
//!
//! # Why every write recomputes the whole user graph
//!
//! Completing a task changes the depth of everything downstream of it and the
//! downstream count of everything upstream, so the "affected set" of almost any
//! write is most of the connected component. Computing that set costs the same
//! traversal as just redoing all of it. A GoingsOn graph is thousands of tasks at
//! the outside, the pass is linear in edges, and being unconditionally correct is
//! worth more here than being clever: this cache is the input to what the user is
//! shown as available work.

use std::collections::{HashMap, HashSet, VecDeque};

use rusqlite::{Connection, params};

use goingson_core::{
    CoreError, DependencyRejection, GraphPosition, LinkedTaskRef, ParseableEnum, Priority,
    ProjectId, Result, Task, TaskDependencies, TaskDependency, TaskGraph, TaskGraphNode, TaskId,
    TaskStatus, UserId, calculate_graph_urgency,
};

use crate::utils::{execute, format_datetime_now, parse_datetime, parse_uuid, query_all};

use super::task_repo::{
    SqliteTaskRepository, TASK_SELECT_COLUMNS, TaskRowWithProject, rows_to_tasks,
};

/// A task as the graph pass needs it: identity, whether it still gates, and
/// where it is filed.
struct GraphTask {
    id: TaskId,
    title: String,
    status: TaskStatus,
    priority: Priority,
    project_id: Option<ProjectId>,
    /// Carried so a plan gate can name a blocker that lives in another project.
    /// An unqualified title reads as if the blocker were local, which is exactly
    /// the case where knowing otherwise matters.
    project_name: Option<String>,
}

impl GraphTask {
    /// Whether this task has stopped gating its dependents.
    ///
    /// `Deleted` counts alongside `Completed` on purpose. A soft-deleted task
    /// will never be completed, so leaving its edges live would strand every
    /// dependent at a depth that can never fall. See [`LinkedTaskRef::is_satisfied`],
    /// which is the same rule on the read side.
    fn is_satisfied(&self) -> bool {
        matches!(self.status, TaskStatus::Completed | TaskStatus::Deleted)
    }
}

/// The user's whole graph, loaded once.
struct LoadedGraph {
    tasks: HashMap<TaskId, GraphTask>,
    /// Every edge, satisfied or not: `blocked -> [blockers]`.
    blockers: HashMap<TaskId, Vec<TaskId>>,
    /// Every edge reversed: `blocker -> [blocked]`.
    dependents: HashMap<TaskId, Vec<TaskId>>,
    /// The edge rows themselves, for [`TaskGraph::edges`].
    edges: Vec<TaskDependency>,
}

impl LoadedGraph {
    /// Blockers of `id` that have not been satisfied, so still gate it.
    fn live_blockers(&self, id: TaskId) -> impl Iterator<Item = TaskId> + '_ {
        self.blockers
            .get(&id)
            .into_iter()
            .flatten()
            .copied()
            .filter(|b| self.tasks.get(b).is_some_and(|t| !t.is_satisfied()))
    }

    /// Tasks `id` gates that are themselves still open. A satisfied dependent is
    /// not work this task frees, so it does not count toward `unblocks_count`.
    fn live_dependents(&self, id: TaskId) -> impl Iterator<Item = TaskId> + '_ {
        self.dependents
            .get(&id)
            .into_iter()
            .flatten()
            .copied()
            .filter(|d| self.tasks.get(d).is_some_and(|t| !t.is_satisfied()))
    }
}

/// Load every task and edge belonging to a user.
///
/// Tasks come in whole, `Deleted` included, because a deleted task is still a
/// legitimate endpoint of an edge and the satisfied-ness rule needs to see its
/// status to say so. Edges whose endpoints are not both the user's are excluded
/// by the join, which is also the ownership check: the table carries no
/// `user_id` of its own.
fn load(conn: &Connection, user_id: UserId) -> Result<LoadedGraph> {
    struct TaskRow {
        id: String,
        title: String,
        status: String,
        priority: String,
        project_id: Option<String>,
        project_name: Option<String>,
    }

    let task_rows: Vec<TaskRow> = query_all(
        conn,
        "SELECT t.id, t.title, t.status, t.priority, t.project_id, p.name AS project_name
         FROM tasks t
         LEFT JOIN projects p ON p.id = t.project_id
         WHERE t.user_id = ?",
        params![user_id.to_string()],
        |row| {
            Ok(TaskRow {
                id: row.get("id")?,
                title: row.get("title")?,
                status: row.get("status")?,
                priority: row.get("priority")?,
                project_id: row.get("project_id")?,
                project_name: row.get("project_name")?,
            })
        },
    )?;

    let mut tasks = HashMap::with_capacity(task_rows.len());
    for row in task_rows {
        let id: TaskId = parse_uuid(&row.id)?.into();
        tasks.insert(
            id,
            GraphTask {
                id,
                title: row.title,
                status: TaskStatus::from_str_or_default(&row.status),
                priority: Priority::from_str_or_default(&row.priority),
                project_id: crate::utils::parse_uuid_opt(row.project_id.as_deref())?
                    .map(Into::into),
                project_name: row.project_name,
            },
        );
    }

    struct EdgeRow {
        id: String,
        blocked_id: String,
        blocker_id: String,
        created_at: String,
    }

    let edge_rows: Vec<EdgeRow> = query_all(
        conn,
        "SELECT d.id, d.blocked_id, d.blocker_id, d.created_at
         FROM task_dependencies d
         JOIN tasks blocked ON blocked.id = d.blocked_id
         JOIN tasks blocker ON blocker.id = d.blocker_id
         WHERE blocked.user_id = ? AND blocker.user_id = ?
         ORDER BY d.created_at, d.id",
        params![user_id.to_string(), user_id.to_string()],
        |row| {
            Ok(EdgeRow {
                id: row.get("id")?,
                blocked_id: row.get("blocked_id")?,
                blocker_id: row.get("blocker_id")?,
                created_at: row.get("created_at")?,
            })
        },
    )?;

    let mut blockers: HashMap<TaskId, Vec<TaskId>> = HashMap::new();
    let mut dependents: HashMap<TaskId, Vec<TaskId>> = HashMap::new();
    let mut edges = Vec::with_capacity(edge_rows.len());

    for row in edge_rows {
        let blocked: TaskId = parse_uuid(&row.blocked_id)?.into();
        let blocker: TaskId = parse_uuid(&row.blocker_id)?.into();
        blockers.entry(blocked).or_default().push(blocker);
        dependents.entry(blocker).or_default().push(blocked);
        edges.push(TaskDependency {
            id: parse_uuid(&row.id)?.into(),
            blocked_id: blocked,
            blocker_id: blocker,
            created_at: parse_datetime(&row.created_at)?,
        });
    }

    Ok(LoadedGraph {
        tasks,
        blockers,
        dependents,
        edges,
    })
}

/// Every task on a cycle of live edges, plus the cycles themselves.
///
/// Iterative depth-first search with the usual three colours, over live edges
/// only. Live edges are the right ones to walk: a cycle running through a
/// completed task is not currently stopping anything, and flagging it would
/// report a problem the user cannot act on and does not have.
fn find_cycles(graph: &LoadedGraph) -> (HashSet<TaskId>, Vec<Vec<TaskId>>) {
    #[derive(Clone, Copy, PartialEq)]
    enum Colour {
        White,
        Grey,
        Black,
    }

    let mut colour: HashMap<TaskId, Colour> =
        graph.tasks.keys().map(|id| (*id, Colour::White)).collect();
    let mut on_cycle: HashSet<TaskId> = HashSet::new();
    let mut cycles: Vec<Vec<TaskId>> = Vec::new();

    // Stable iteration order so a cycle is reported the same way every run,
    // which matters because the report is shown to a person and compared
    // against the last one.
    let mut roots: Vec<TaskId> = graph.tasks.keys().copied().collect();
    roots.sort_by_key(ToString::to_string);

    for root in roots {
        if colour[&root] != Colour::White {
            continue;
        }

        // Explicit stack of (node, blockers still to visit). Recursion would be
        // bounded by graph depth, and a merged cycle is exactly the case where
        // that bound is not something we control.
        let mut path: Vec<TaskId> = Vec::new();
        let mut stack: Vec<(TaskId, VecDeque<TaskId>)> =
            vec![(root, graph.live_blockers(root).collect())];
        colour.insert(root, Colour::Grey);
        path.push(root);

        while let Some((node, pending)) = stack.last_mut() {
            let node = *node;
            match pending.pop_front() {
                Some(next) => match colour.get(&next).copied().unwrap_or(Colour::Black) {
                    Colour::White => {
                        colour.insert(next, Colour::Grey);
                        path.push(next);
                        stack.push((next, graph.live_blockers(next).collect()));
                    }
                    // Grey means we have walked back onto the current path, so
                    // everything from that point on is a cycle.
                    Colour::Grey => {
                        if let Some(start) = path.iter().position(|n| *n == next) {
                            let cycle: Vec<TaskId> = path[start..].to_vec();
                            on_cycle.extend(cycle.iter().copied());
                            cycles.push(cycle);
                        }
                    }
                    Colour::Black => {}
                },
                None => {
                    colour.insert(node, Colour::Black);
                    path.pop();
                    stack.pop();
                }
            }
        }
    }

    (on_cycle, cycles)
}

/// Longest chain of live blockers ahead of each task.
///
/// Memoised depth-first search. A task on a cycle has no meaningful depth, so it
/// is excluded here and scored at the cap by [`calculate_graph_urgency`], which
/// treats a cycle as the deepest possible block. Excluding them is also what
/// keeps this pass terminating without a visited set of its own.
fn block_depths(graph: &LoadedGraph, on_cycle: &HashSet<TaskId>) -> HashMap<TaskId, u32> {
    let mut depth: HashMap<TaskId, u32> = HashMap::with_capacity(graph.tasks.len());

    for id in graph.tasks.keys() {
        if depth.contains_key(id) || on_cycle.contains(id) {
            continue;
        }
        // Post-order walk: a node is resolved once every blocker it has is.
        let mut stack = vec![(*id, false)];
        while let Some((node, expanded)) = stack.pop() {
            if depth.contains_key(&node) || on_cycle.contains(&node) {
                continue;
            }
            if expanded {
                let d = graph
                    .live_blockers(node)
                    .filter(|b| !on_cycle.contains(b))
                    .map(|b| depth.get(&b).copied().unwrap_or(0) + 1)
                    .max()
                    .unwrap_or(0);
                // A task whose only live blockers are on a cycle is still
                // blocked, even though none of them gave it a depth.
                let d = if d == 0 && graph.live_blockers(node).next().is_some() {
                    1
                } else {
                    d
                };
                depth.insert(node, d);
                continue;
            }
            stack.push((node, true));
            for blocker in graph.live_blockers(node) {
                if !depth.contains_key(&blocker) && !on_cycle.contains(&blocker) {
                    stack.push((blocker, false));
                }
            }
        }
    }

    depth
}

/// How many still-open tasks each task transitively frees.
///
/// A breadth-first walk downstream per task, each with its own visited set, so a
/// cycle is traversed once and not forever. Quadratic in the worst case and
/// linear in practice, because dependency graphs are wide and shallow rather
/// than densely connected.
fn downstream_counts(graph: &LoadedGraph) -> HashMap<TaskId, u32> {
    let mut counts = HashMap::with_capacity(graph.tasks.len());

    for id in graph.tasks.keys() {
        let mut seen: HashSet<TaskId> = HashSet::new();
        let mut queue: VecDeque<TaskId> = graph.live_dependents(*id).collect();
        seen.insert(*id);
        while let Some(next) = queue.pop_front() {
            if !seen.insert(next) {
                continue;
            }
            queue.extend(graph.live_dependents(next));
        }
        // `seen` counted the task itself, which it does not free.
        counts.insert(
            *id,
            u32::try_from(seen.len().saturating_sub(1)).unwrap_or(u32::MAX),
        );
    }

    counts
}

/// Compute every task's graph position from a loaded graph.
fn positions(graph: &LoadedGraph) -> (HashMap<TaskId, GraphPosition>, Vec<Vec<TaskId>>) {
    let (on_cycle, cycles) = find_cycles(graph);
    let depths = block_depths(graph, &on_cycle);
    let downstream = downstream_counts(graph);

    let positions = graph
        .tasks
        .keys()
        .map(|id| {
            (
                *id,
                GraphPosition {
                    block_depth: depths.get(id).copied().unwrap_or(0),
                    unblocks_count: downstream.get(id).copied().unwrap_or(0),
                    in_cycle: on_cycle.contains(id),
                },
            )
        })
        .collect();

    (positions, cycles)
}

/// Recompute and persist the cached graph columns for a user.
///
/// Takes a connection so it can run inside a caller's transaction: completing a
/// task and re-scoring the graph it sits in have to land together or a reader
/// between them sees work that is neither done nor available.
///
/// Returns the number of rows whose cached values actually moved. Writing only
/// the changed rows keeps this off the sync changelog for the common no-op case;
/// the columns are local, but the write itself would still churn `tasks`.
pub(crate) fn recompute(conn: &Connection, user_id: UserId) -> Result<usize> {
    let graph = load(conn, user_id)?;
    let (positions, _) = positions(&graph);

    struct Cached {
        id: String,
        block_depth: i64,
        unblocks_count: i64,
        in_cycle: i32,
        graph_urgency: f64,
    }

    let current: Vec<Cached> = query_all(
        conn,
        "SELECT id, block_depth, unblocks_count, in_cycle, graph_urgency
         FROM tasks WHERE user_id = ?",
        params![user_id.to_string()],
        |row| {
            Ok(Cached {
                id: row.get("id")?,
                block_depth: row.get("block_depth")?,
                unblocks_count: row.get("unblocks_count")?,
                in_cycle: row.get("in_cycle")?,
                graph_urgency: row.get("graph_urgency")?,
            })
        },
    )?;

    let mut changed = 0usize;
    for row in current {
        let id: TaskId = parse_uuid(&row.id)?.into();
        let want = positions.get(&id).copied().unwrap_or_default();
        let want_urgency = calculate_graph_urgency(&want);

        let same = i64::from(want.block_depth) == row.block_depth
            && i64::from(want.unblocks_count) == row.unblocks_count
            && i32::from(want.in_cycle) == row.in_cycle
            // Exact comparison is right for a value both sides produce by the
            // same rounding from the same integer inputs. A tolerance here would
            // only hide a genuine drift in the scoring function.
            && (want_urgency - row.graph_urgency).abs() < f64::EPSILON;
        if same {
            continue;
        }

        execute(
            conn,
            "UPDATE tasks SET block_depth = ?, unblocks_count = ?, in_cycle = ?, graph_urgency = ?
             WHERE id = ? AND user_id = ?",
            params![
                i64::from(want.block_depth),
                i64::from(want.unblocks_count),
                i32::from(want.in_cycle),
                want_urgency,
                row.id,
                user_id.to_string(),
            ],
        )?;
        changed += 1;
    }

    Ok(changed)
}

/// Whether `blocked` is already reachable from `blocker` by following
/// dependencies, which is what makes the new edge a cycle.
///
/// Walks ALL edges, not only live ones. A structural cycle that currently runs
/// through a completed task is still a cycle: reopening that task would close
/// it, and the write path is the one place we can refuse cheaply.
///
/// Returns the offending path when there is one, so the refusal can name the
/// chain rather than just saying no.
fn path_between(graph: &LoadedGraph, from: TaskId, to: TaskId) -> Option<Vec<TaskId>> {
    let mut seen: HashSet<TaskId> = HashSet::from([from]);
    // Breadth-first, so the path reported is the shortest one, which is the one
    // a person can most easily check.
    let mut queue: VecDeque<TaskId> = VecDeque::from([from]);
    let mut came_from: HashMap<TaskId, TaskId> = HashMap::new();

    while let Some(node) = queue.pop_front() {
        if node == to {
            let mut path = vec![node];
            let mut cursor = node;
            while let Some(prev) = came_from.get(&cursor) {
                path.push(*prev);
                cursor = *prev;
            }
            path.reverse();
            return Some(path);
        }
        for blocker in graph.blockers.get(&node).into_iter().flatten().copied() {
            if seen.insert(blocker) {
                came_from.insert(blocker, node);
                queue.push_back(blocker);
            }
        }
    }

    None
}

/// Confirm a task exists and belongs to the user.
fn require_task(conn: &Connection, user_id: UserId, id: TaskId) -> Result<()> {
    let found: Vec<i64> = query_all(
        conn,
        "SELECT 1 FROM tasks WHERE id = ? AND user_id = ?",
        params![id.to_string(), user_id.to_string()],
        |row| row.get(0),
    )?;
    if found.is_empty() {
        return Err(CoreError::not_found("Task", id));
    }
    Ok(())
}

/// The tasks on one side of a task's edges, as [`LinkedTaskRef`]s.
///
/// `column` selects the direction: matching on `blocked_id` yields the task's
/// blockers, matching on `blocker_id` yields its dependents.
fn linked(
    conn: &Connection,
    user_id: UserId,
    task_id: TaskId,
    match_column: &str,
    other_column: &str,
) -> Result<Vec<LinkedTaskRef>> {
    struct Row {
        id: String,
        title: String,
        status: String,
        project_name: Option<String>,
    }

    // The columns are not caller-supplied; they come from the two call sites
    // below as literals. Kept as parameters rather than duplicating the query.
    let sql = format!(
        "SELECT other.id, other.title, other.status, p.name AS project_name
         FROM task_dependencies d
         JOIN tasks other ON other.id = d.{other_column}
         LEFT JOIN projects p ON p.id = other.project_id
         WHERE d.{match_column} = ? AND other.user_id = ?
         ORDER BY d.created_at, d.id"
    );

    let rows: Vec<Row> = query_all(
        conn,
        &sql,
        params![task_id.to_string(), user_id.to_string()],
        |row| {
            Ok(Row {
                id: row.get("id")?,
                title: row.get("title")?,
                status: row.get("status")?,
                project_name: row.get("project_name")?,
            })
        },
    )?;

    rows.into_iter()
        .map(|row| {
            Ok(LinkedTaskRef {
                id: parse_uuid(&row.id)?.into(),
                title: row.title,
                status: TaskStatus::from_str_or_default(&row.status),
                project_name: row.project_name,
            })
        })
        .collect()
}

impl TaskDependencies for SqliteTaskRepository {
    #[tracing::instrument(skip(self))]
    fn add_dependency(
        &self,
        user_id: UserId,
        blocked_id: TaskId,
        blocker_id: TaskId,
    ) -> Result<TaskDependency> {
        if blocked_id == blocker_id {
            return Err(DependencyRejection::SelfEdge {
                task_id: blocked_id,
            }
            .into());
        }

        let mut conn = self.db.conn()?;
        let tx = conn.transaction().map_err(CoreError::database)?;

        require_task(&tx, user_id, blocked_id)?;
        require_task(&tx, user_id, blocker_id)?;

        // The check and the insert share this transaction. Two sessions each
        // adding one leg of a two-edge cycle would otherwise both read a clean
        // graph and both commit.
        let graph = load(&tx, user_id)?;
        if let Some(path) = path_between(&graph, blocker_id, blocked_id) {
            return Err(DependencyRejection::WouldCycle { path }.into());
        }

        let id = TaskDependency::deterministic_id(blocker_id, blocked_id);
        let created_at = format_datetime_now();

        // Idempotent on the deterministic id: drawing the same edge twice, here
        // or on another device, converges on one row.
        execute(
            &tx,
            "INSERT INTO task_dependencies (id, blocked_id, blocker_id, created_at, group_id)
             VALUES (?, ?, ?, ?, (SELECT group_id FROM tasks WHERE id = ?))
             ON CONFLICT(id) DO NOTHING",
            params![
                id.to_string(),
                blocked_id.to_string(),
                blocker_id.to_string(),
                created_at,
                blocked_id.to_string(),
            ],
        )?;

        recompute(&tx, user_id)?;
        tx.commit().map_err(CoreError::database)?;

        Ok(TaskDependency {
            id,
            blocked_id,
            blocker_id,
            created_at: parse_datetime(&created_at)?,
        })
    }

    #[tracing::instrument(skip(self))]
    fn remove_dependency(
        &self,
        user_id: UserId,
        blocked_id: TaskId,
        blocker_id: TaskId,
    ) -> Result<bool> {
        let mut conn = self.db.conn()?;
        let tx = conn.transaction().map_err(CoreError::database)?;

        let removed = execute(
            &tx,
            "DELETE FROM task_dependencies
             WHERE blocked_id = ? AND blocker_id = ?
               AND EXISTS (SELECT 1 FROM tasks WHERE id = ? AND user_id = ?)",
            params![
                blocked_id.to_string(),
                blocker_id.to_string(),
                blocked_id.to_string(),
                user_id.to_string(),
            ],
        )?;

        if removed == 0 {
            return Ok(false);
        }

        recompute(&tx, user_id)?;
        tx.commit().map_err(CoreError::database)?;
        Ok(true)
    }

    fn list_blockers(&self, user_id: UserId, task_id: TaskId) -> Result<Vec<LinkedTaskRef>> {
        let conn = self.db.conn()?;
        linked(&conn, user_id, task_id, "blocked_id", "blocker_id")
    }

    fn list_dependents(&self, user_id: UserId, task_id: TaskId) -> Result<Vec<LinkedTaskRef>> {
        let conn = self.db.conn()?;
        linked(&conn, user_id, task_id, "blocker_id", "blocked_id")
    }

    #[tracing::instrument(skip(self))]
    fn task_graph(&self, user_id: UserId, project_id: Option<ProjectId>) -> Result<TaskGraph> {
        let conn = self.db.conn()?;
        let graph = load(&conn, user_id)?;
        let (positions, cycles) = positions(&graph);

        // Which tasks the caller asked about. A project scope selects the
        // project's tasks, then pulls in whatever their edges reach: a blocker
        // in another project is the reason a task in this one is not ready, and
        // cutting it would show that task as available.
        let in_scope: HashSet<TaskId> = match project_id {
            None => graph.tasks.keys().copied().collect(),
            Some(pid) => {
                let seeds: Vec<TaskId> = graph
                    .tasks
                    .values()
                    .filter(|t| t.project_id == Some(pid))
                    .map(|t| t.id)
                    .collect();
                let mut reached: HashSet<TaskId> = seeds.iter().copied().collect();
                let mut queue: VecDeque<TaskId> = seeds.into_iter().collect();
                while let Some(node) = queue.pop_front() {
                    let neighbours = graph
                        .blockers
                        .get(&node)
                        .into_iter()
                        .flatten()
                        .chain(graph.dependents.get(&node).into_iter().flatten())
                        .copied();
                    for n in neighbours {
                        if reached.insert(n) {
                            queue.push_back(n);
                        }
                    }
                }
                reached
            }
        };

        // Only tasks that actually touch an edge. An isolated task is not part
        // of any graph worth drawing, and including every unlinked task would
        // bury the structure in noise.
        let mut nodes: Vec<TaskGraphNode> = graph
            .tasks
            .values()
            .filter(|t| in_scope.contains(&t.id))
            .filter(|t| graph.blockers.contains_key(&t.id) || graph.dependents.contains_key(&t.id))
            .map(|t| TaskGraphNode {
                id: t.id,
                title: t.title.clone(),
                status: t.status.clone(),
                priority: t.priority.clone(),
                project_id: t.project_id,
                position: positions.get(&t.id).copied().unwrap_or_default(),
            })
            .collect();
        nodes.sort_by_key(|n| n.id.to_string());

        let node_ids: HashSet<TaskId> = nodes.iter().map(|n| n.id).collect();
        let edges = graph
            .edges
            .iter()
            .filter(|e| node_ids.contains(&e.blocked_id) && node_ids.contains(&e.blocker_id))
            .cloned()
            .collect();
        let cycles = cycles
            .into_iter()
            .filter(|c| c.iter().all(|id| node_ids.contains(id)))
            .collect();

        Ok(TaskGraph {
            nodes,
            edges,
            cycles,
        })
    }

    #[tracing::instrument(skip(self))]
    fn list_ready(
        &self,
        user_id: UserId,
        project_id: Option<ProjectId>,
        max_depth: u32,
        limit: Option<i64>,
    ) -> Result<Vec<Task>> {
        let conn = self.db.conn()?;

        // Reads the cached column rather than recomputing, which is the whole
        // point of caching it: this is the query the task list and every mode of
        // /threads run, and it stays a plain indexed scan.
        let mut sql = format!(
            "SELECT {TASK_SELECT_COLUMNS}
             FROM tasks t
             LEFT JOIN projects p ON t.project_id = p.id AND p.user_id = ?
             LEFT JOIN contacts ct ON ct.id = t.contact_id
             WHERE t.user_id = ?
               AND t.status IN ('Pending', 'Started')
               AND t.block_depth <= ?
               -- A task on a cycle carries depth 0 because a cycle has no
               -- meaningful depth, and it is the one thing that must never read
               -- as ready: nothing on it can ever open.
               AND t.in_cycle = 0
               AND (t.snoozed_until IS NULL OR t.snoozed_until <= ?)"
        );

        let mut binds: Vec<Box<dyn rusqlite::ToSql>> = vec![
            Box::new(user_id.to_string()),
            Box::new(user_id.to_string()),
            Box::new(i64::from(max_depth)),
            Box::new(format_datetime_now()),
        ];

        if let Some(pid) = project_id {
            sql.push_str(" AND t.project_id = ?");
            binds.push(Box::new(pid.to_string()));
        }

        // Shallowest first, then by effective urgency. Depth leads because a
        // caller that asked for depth 2 still wants the work it can start today
        // at the top; within one depth the score decides, and that score already
        // carries the unblocks bonus.
        sql.push_str(
            " ORDER BY t.block_depth ASC, (t.urgency + t.graph_urgency) DESC, t.created_at DESC",
        );

        if let Some(limit) = limit {
            sql.push_str(" LIMIT ?");
            binds.push(Box::new(limit));
        }

        let rows: Vec<TaskRowWithProject> = query_all(
            &conn,
            &sql,
            rusqlite::params_from_iter(binds.iter().map(std::convert::AsRef::as_ref)),
            TaskRowWithProject::from_row,
        )?;

        rows_to_tasks(&conn, rows)
    }

    #[tracing::instrument(skip(self))]
    fn plan_gates(
        &self,
        user_id: UserId,
        window_start: chrono::DateTime<chrono::Utc>,
        window_end: chrono::DateTime<chrono::Utc>,
    ) -> Result<HashMap<TaskId, goingson_core::PlanGate>> {
        let conn = self.db.conn()?;
        let graph = load(&conn, user_id)?;

        // What is already in the plan, and when. `scheduled_start` inside the
        // window is the whole test: a blocker parked next Tuesday is not in
        // today's plan and must not unlock anything in it, or the day becomes
        // one nobody can actually execute.
        struct Scheduled {
            id: String,
            scheduled_start: String,
        }
        let scheduled_rows: Vec<Scheduled> = query_all(
            &conn,
            "SELECT id, scheduled_start FROM tasks
             WHERE user_id = ? AND scheduled_start IS NOT NULL
               AND scheduled_start >= ? AND scheduled_start <= ?
               AND status NOT IN ('Completed', 'Deleted')",
            params![
                user_id.to_string(),
                crate::utils::format_datetime(&window_start),
                crate::utils::format_datetime(&window_end),
            ],
            |row| {
                Ok(Scheduled {
                    id: row.get("id")?,
                    scheduled_start: row.get("scheduled_start")?,
                })
            },
        )?;

        let mut in_plan: HashMap<TaskId, chrono::DateTime<chrono::Utc>> = HashMap::new();
        for row in scheduled_rows {
            in_plan.insert(
                parse_uuid(&row.id)?.into(),
                parse_datetime(&row.scheduled_start)?,
            );
        }

        let mut gates = HashMap::new();
        for id in graph.tasks.keys().copied() {
            let live: Vec<TaskId> = graph.live_blockers(id).collect();
            // No gate at all rather than an empty one: an absent entry is the
            // "nothing is in the way" signal, so a caller never has to look
            // inside to find out.
            if live.is_empty() {
                continue;
            }

            let after: Vec<LinkedTaskRef> = live
                .iter()
                .filter_map(|b| graph.tasks.get(b))
                .map(|t| LinkedTaskRef {
                    id: t.id,
                    title: t.title.clone(),
                    status: t.status.clone(),
                    project_name: t.project_name.clone(),
                })
                .collect();

            let unlocked_by_plan = live.iter().all(|b| in_plan.contains_key(b));
            // Only meaningful once this task is itself in the plan; an
            // unscheduled task has no start to be earlier than.
            let out_of_order = in_plan.get(&id).is_some_and(|mine| {
                live.iter()
                    .filter_map(|b| in_plan.get(b))
                    .any(|theirs| theirs > mine)
            });

            gates.insert(
                id,
                goingson_core::PlanGate {
                    after,
                    unlocked_by_plan,
                    out_of_order,
                },
            );
        }

        Ok(gates)
    }

    #[tracing::instrument(skip(self))]
    fn recompute_graph(&self, user_id: UserId) -> Result<usize> {
        let mut conn = self.db.conn()?;
        let tx = conn.transaction().map_err(CoreError::database)?;
        let changed = recompute(&tx, user_id)?;
        tx.commit().map_err(CoreError::database)?;
        Ok(changed)
    }

    fn list_all_dependencies(&self, user_id: UserId) -> Result<Vec<TaskDependency>> {
        let conn = self.db.conn()?;
        Ok(load(&conn, user_id)?.edges)
    }

    fn restore_dependency(&self, user_id: UserId, dependency: &TaskDependency) -> Result<()> {
        let conn = self.db.conn()?;
        execute(
            &conn,
            "INSERT INTO task_dependencies (id, blocked_id, blocker_id, created_at, group_id)
             SELECT ?, ?, ?, ?, (SELECT group_id FROM tasks WHERE id = ?)
             WHERE EXISTS (SELECT 1 FROM tasks WHERE id = ? AND user_id = ?)
               AND EXISTS (SELECT 1 FROM tasks WHERE id = ? AND user_id = ?)
             ON CONFLICT(id) DO NOTHING",
            params![
                dependency.id.to_string(),
                dependency.blocked_id.to_string(),
                dependency.blocker_id.to_string(),
                crate::utils::format_datetime(&dependency.created_at),
                dependency.blocked_id.to_string(),
                dependency.blocked_id.to_string(),
                user_id.to_string(),
                dependency.blocker_id.to_string(),
                user_id.to_string(),
            ],
        )?;
        Ok(())
    }
}