Skip to main content

max / goingson

33.5 KB · 928 lines History Blame Raw
1 //! Dependency-graph repository methods for `SqliteTaskRepository`.
2 //!
3 //! `task_dependencies` holds the edges and is the only stored truth. Everything
4 //! a caller reads about blocking (readiness, depth, downstream count, cycles) is
5 //! derived from those edges plus task statuses, and cached back onto the task row
6 //! by [`recompute`].
7 //!
8 //! # Why the traversals are in Rust and not in SQL
9 //!
10 //! The obvious implementation is a recursive CTE per question. Three things
11 //! argue against it here. The depth we want is the LONGEST path, and a recursive
12 //! CTE naturally produces every path, so getting the longest means materialising
13 //! all of them and taking a max, which is exponential on a diamond-shaped graph.
14 //! Cycle handling in a recursive CTE is a depth guard, which silently truncates
15 //! rather than reporting the cycle we need to surface. And every write recomputes
16 //! the whole user graph anyway, so the work is one pass over a few thousand rows
17 //! either way, and the pass that is readable wins.
18 //!
19 //! # Why every write recomputes the whole user graph
20 //!
21 //! Completing a task changes the depth of everything downstream of it and the
22 //! downstream count of everything upstream, so the "affected set" of almost any
23 //! write is most of the connected component. Computing that set costs the same
24 //! traversal as just redoing all of it. A GoingsOn graph is thousands of tasks at
25 //! the outside, the pass is linear in edges, and being unconditionally correct is
26 //! worth more here than being clever: this cache is the input to what the user is
27 //! shown as available work.
28
29 use std::collections::{HashMap, HashSet, VecDeque};
30
31 use rusqlite::{Connection, params};
32
33 use goingson_core::{
34 CoreError, DependencyRejection, GraphPosition, LinkedTaskRef, ParseableEnum, Priority,
35 ProjectId, Result, Task, TaskDependencies, TaskDependency, TaskGraph, TaskGraphNode, TaskId,
36 TaskStatus, UserId, calculate_graph_urgency,
37 };
38
39 use crate::utils::{execute, format_datetime_now, parse_datetime, parse_uuid, query_all};
40
41 use super::task_repo::{
42 SqliteTaskRepository, TASK_SELECT_COLUMNS, TaskRowWithProject, rows_to_tasks,
43 };
44
45 /// A task as the graph pass needs it: identity, whether it still gates, and
46 /// where it is filed.
47 struct GraphTask {
48 id: TaskId,
49 title: String,
50 status: TaskStatus,
51 priority: Priority,
52 project_id: Option<ProjectId>,
53 /// Carried so a plan gate can name a blocker that lives in another project.
54 /// An unqualified title reads as if the blocker were local, which is exactly
55 /// the case where knowing otherwise matters.
56 project_name: Option<String>,
57 }
58
59 impl GraphTask {
60 /// Whether this task has stopped gating its dependents.
61 ///
62 /// `Deleted` counts alongside `Completed` on purpose. A soft-deleted task
63 /// will never be completed, so leaving its edges live would strand every
64 /// dependent at a depth that can never fall. See [`LinkedTaskRef::is_satisfied`],
65 /// which is the same rule on the read side.
66 fn is_satisfied(&self) -> bool {
67 matches!(self.status, TaskStatus::Completed | TaskStatus::Deleted)
68 }
69 }
70
71 /// The user's whole graph, loaded once.
72 struct LoadedGraph {
73 tasks: HashMap<TaskId, GraphTask>,
74 /// Every edge, satisfied or not: `blocked -> [blockers]`.
75 blockers: HashMap<TaskId, Vec<TaskId>>,
76 /// Every edge reversed: `blocker -> [blocked]`.
77 dependents: HashMap<TaskId, Vec<TaskId>>,
78 /// The edge rows themselves, for [`TaskGraph::edges`].
79 edges: Vec<TaskDependency>,
80 }
81
82 impl LoadedGraph {
83 /// Blockers of `id` that have not been satisfied, so still gate it.
84 fn live_blockers(&self, id: TaskId) -> impl Iterator<Item = TaskId> + '_ {
85 self.blockers
86 .get(&id)
87 .into_iter()
88 .flatten()
89 .copied()
90 .filter(|b| self.tasks.get(b).is_some_and(|t| !t.is_satisfied()))
91 }
92
93 /// Tasks `id` gates that are themselves still open. A satisfied dependent is
94 /// not work this task frees, so it does not count toward `unblocks_count`.
95 fn live_dependents(&self, id: TaskId) -> impl Iterator<Item = TaskId> + '_ {
96 self.dependents
97 .get(&id)
98 .into_iter()
99 .flatten()
100 .copied()
101 .filter(|d| self.tasks.get(d).is_some_and(|t| !t.is_satisfied()))
102 }
103 }
104
105 /// Load every task and edge belonging to a user.
106 ///
107 /// Tasks come in whole, `Deleted` included, because a deleted task is still a
108 /// legitimate endpoint of an edge and the satisfied-ness rule needs to see its
109 /// status to say so. Edges whose endpoints are not both the user's are excluded
110 /// by the join, which is also the ownership check: the table carries no
111 /// `user_id` of its own.
112 fn load(conn: &Connection, user_id: UserId) -> Result<LoadedGraph> {
113 struct TaskRow {
114 id: String,
115 title: String,
116 status: String,
117 priority: String,
118 project_id: Option<String>,
119 project_name: Option<String>,
120 }
121
122 let task_rows: Vec<TaskRow> = query_all(
123 conn,
124 "SELECT t.id, t.title, t.status, t.priority, t.project_id, p.name AS project_name
125 FROM tasks t
126 LEFT JOIN projects p ON p.id = t.project_id
127 WHERE t.user_id = ?",
128 params![user_id.to_string()],
129 |row| {
130 Ok(TaskRow {
131 id: row.get("id")?,
132 title: row.get("title")?,
133 status: row.get("status")?,
134 priority: row.get("priority")?,
135 project_id: row.get("project_id")?,
136 project_name: row.get("project_name")?,
137 })
138 },
139 )?;
140
141 let mut tasks = HashMap::with_capacity(task_rows.len());
142 for row in task_rows {
143 let id: TaskId = parse_uuid(&row.id)?.into();
144 tasks.insert(
145 id,
146 GraphTask {
147 id,
148 title: row.title,
149 status: TaskStatus::from_str_or_default(&row.status),
150 priority: Priority::from_str_or_default(&row.priority),
151 project_id: crate::utils::parse_uuid_opt(row.project_id.as_deref())?
152 .map(Into::into),
153 project_name: row.project_name,
154 },
155 );
156 }
157
158 struct EdgeRow {
159 id: String,
160 blocked_id: String,
161 blocker_id: String,
162 created_at: String,
163 }
164
165 let edge_rows: Vec<EdgeRow> = query_all(
166 conn,
167 "SELECT d.id, d.blocked_id, d.blocker_id, d.created_at
168 FROM task_dependencies d
169 JOIN tasks blocked ON blocked.id = d.blocked_id
170 JOIN tasks blocker ON blocker.id = d.blocker_id
171 WHERE blocked.user_id = ? AND blocker.user_id = ?
172 ORDER BY d.created_at, d.id",
173 params![user_id.to_string(), user_id.to_string()],
174 |row| {
175 Ok(EdgeRow {
176 id: row.get("id")?,
177 blocked_id: row.get("blocked_id")?,
178 blocker_id: row.get("blocker_id")?,
179 created_at: row.get("created_at")?,
180 })
181 },
182 )?;
183
184 let mut blockers: HashMap<TaskId, Vec<TaskId>> = HashMap::new();
185 let mut dependents: HashMap<TaskId, Vec<TaskId>> = HashMap::new();
186 let mut edges = Vec::with_capacity(edge_rows.len());
187
188 for row in edge_rows {
189 let blocked: TaskId = parse_uuid(&row.blocked_id)?.into();
190 let blocker: TaskId = parse_uuid(&row.blocker_id)?.into();
191 blockers.entry(blocked).or_default().push(blocker);
192 dependents.entry(blocker).or_default().push(blocked);
193 edges.push(TaskDependency {
194 id: parse_uuid(&row.id)?.into(),
195 blocked_id: blocked,
196 blocker_id: blocker,
197 created_at: parse_datetime(&row.created_at)?,
198 });
199 }
200
201 Ok(LoadedGraph {
202 tasks,
203 blockers,
204 dependents,
205 edges,
206 })
207 }
208
209 /// Every task on a cycle of live edges, plus the cycles themselves.
210 ///
211 /// Iterative depth-first search with the usual three colours, over live edges
212 /// only. Live edges are the right ones to walk: a cycle running through a
213 /// completed task is not currently stopping anything, and flagging it would
214 /// report a problem the user cannot act on and does not have.
215 fn find_cycles(graph: &LoadedGraph) -> (HashSet<TaskId>, Vec<Vec<TaskId>>) {
216 #[derive(Clone, Copy, PartialEq)]
217 enum Colour {
218 White,
219 Grey,
220 Black,
221 }
222
223 let mut colour: HashMap<TaskId, Colour> =
224 graph.tasks.keys().map(|id| (*id, Colour::White)).collect();
225 let mut on_cycle: HashSet<TaskId> = HashSet::new();
226 let mut cycles: Vec<Vec<TaskId>> = Vec::new();
227
228 // Stable iteration order so a cycle is reported the same way every run,
229 // which matters because the report is shown to a person and compared
230 // against the last one.
231 let mut roots: Vec<TaskId> = graph.tasks.keys().copied().collect();
232 roots.sort_by_key(ToString::to_string);
233
234 for root in roots {
235 if colour[&root] != Colour::White {
236 continue;
237 }
238
239 // Explicit stack of (node, blockers still to visit). Recursion would be
240 // bounded by graph depth, and a merged cycle is exactly the case where
241 // that bound is not something we control.
242 let mut path: Vec<TaskId> = Vec::new();
243 let mut stack: Vec<(TaskId, VecDeque<TaskId>)> =
244 vec![(root, graph.live_blockers(root).collect())];
245 colour.insert(root, Colour::Grey);
246 path.push(root);
247
248 while let Some((node, pending)) = stack.last_mut() {
249 let node = *node;
250 match pending.pop_front() {
251 Some(next) => match colour.get(&next).copied().unwrap_or(Colour::Black) {
252 Colour::White => {
253 colour.insert(next, Colour::Grey);
254 path.push(next);
255 stack.push((next, graph.live_blockers(next).collect()));
256 }
257 // Grey means we have walked back onto the current path, so
258 // everything from that point on is a cycle.
259 Colour::Grey => {
260 if let Some(start) = path.iter().position(|n| *n == next) {
261 let cycle: Vec<TaskId> = path[start..].to_vec();
262 on_cycle.extend(cycle.iter().copied());
263 cycles.push(cycle);
264 }
265 }
266 Colour::Black => {}
267 },
268 None => {
269 colour.insert(node, Colour::Black);
270 path.pop();
271 stack.pop();
272 }
273 }
274 }
275 }
276
277 (on_cycle, cycles)
278 }
279
280 /// Longest chain of live blockers ahead of each task.
281 ///
282 /// Memoised depth-first search. A task on a cycle has no meaningful depth, so it
283 /// is excluded here and scored at the cap by [`calculate_graph_urgency`], which
284 /// treats a cycle as the deepest possible block. Excluding them is also what
285 /// keeps this pass terminating without a visited set of its own.
286 fn block_depths(graph: &LoadedGraph, on_cycle: &HashSet<TaskId>) -> HashMap<TaskId, u32> {
287 let mut depth: HashMap<TaskId, u32> = HashMap::with_capacity(graph.tasks.len());
288
289 for id in graph.tasks.keys() {
290 if depth.contains_key(id) || on_cycle.contains(id) {
291 continue;
292 }
293 // Post-order walk: a node is resolved once every blocker it has is.
294 let mut stack = vec![(*id, false)];
295 while let Some((node, expanded)) = stack.pop() {
296 if depth.contains_key(&node) || on_cycle.contains(&node) {
297 continue;
298 }
299 if expanded {
300 let d = graph
301 .live_blockers(node)
302 .filter(|b| !on_cycle.contains(b))
303 .map(|b| depth.get(&b).copied().unwrap_or(0) + 1)
304 .max()
305 .unwrap_or(0);
306 // A task whose only live blockers are on a cycle is still
307 // blocked, even though none of them gave it a depth.
308 let d = if d == 0 && graph.live_blockers(node).next().is_some() {
309 1
310 } else {
311 d
312 };
313 depth.insert(node, d);
314 continue;
315 }
316 stack.push((node, true));
317 for blocker in graph.live_blockers(node) {
318 if !depth.contains_key(&blocker) && !on_cycle.contains(&blocker) {
319 stack.push((blocker, false));
320 }
321 }
322 }
323 }
324
325 depth
326 }
327
328 /// How many still-open tasks each task transitively frees.
329 ///
330 /// A breadth-first walk downstream per task, each with its own visited set, so a
331 /// cycle is traversed once and not forever. Quadratic in the worst case and
332 /// linear in practice, because dependency graphs are wide and shallow rather
333 /// than densely connected.
334 fn downstream_counts(graph: &LoadedGraph) -> HashMap<TaskId, u32> {
335 let mut counts = HashMap::with_capacity(graph.tasks.len());
336
337 for id in graph.tasks.keys() {
338 let mut seen: HashSet<TaskId> = HashSet::new();
339 let mut queue: VecDeque<TaskId> = graph.live_dependents(*id).collect();
340 seen.insert(*id);
341 while let Some(next) = queue.pop_front() {
342 if !seen.insert(next) {
343 continue;
344 }
345 queue.extend(graph.live_dependents(next));
346 }
347 // `seen` counted the task itself, which it does not free.
348 counts.insert(
349 *id,
350 u32::try_from(seen.len().saturating_sub(1)).unwrap_or(u32::MAX),
351 );
352 }
353
354 counts
355 }
356
357 /// Compute every task's graph position from a loaded graph.
358 fn positions(graph: &LoadedGraph) -> (HashMap<TaskId, GraphPosition>, Vec<Vec<TaskId>>) {
359 let (on_cycle, cycles) = find_cycles(graph);
360 let depths = block_depths(graph, &on_cycle);
361 let downstream = downstream_counts(graph);
362
363 let positions = graph
364 .tasks
365 .keys()
366 .map(|id| {
367 (
368 *id,
369 GraphPosition {
370 block_depth: depths.get(id).copied().unwrap_or(0),
371 unblocks_count: downstream.get(id).copied().unwrap_or(0),
372 in_cycle: on_cycle.contains(id),
373 },
374 )
375 })
376 .collect();
377
378 (positions, cycles)
379 }
380
381 /// Recompute and persist the cached graph columns for a user.
382 ///
383 /// Takes a connection so it can run inside a caller's transaction: completing a
384 /// task and re-scoring the graph it sits in have to land together or a reader
385 /// between them sees work that is neither done nor available.
386 ///
387 /// Returns the number of rows whose cached values actually moved. Writing only
388 /// the changed rows keeps this off the sync changelog for the common no-op case;
389 /// the columns are local, but the write itself would still churn `tasks`.
390 pub(crate) fn recompute(conn: &Connection, user_id: UserId) -> Result<usize> {
391 let graph = load(conn, user_id)?;
392 let (positions, _) = positions(&graph);
393
394 struct Cached {
395 id: String,
396 block_depth: i64,
397 unblocks_count: i64,
398 in_cycle: i32,
399 graph_urgency: f64,
400 }
401
402 let current: Vec<Cached> = query_all(
403 conn,
404 "SELECT id, block_depth, unblocks_count, in_cycle, graph_urgency
405 FROM tasks WHERE user_id = ?",
406 params![user_id.to_string()],
407 |row| {
408 Ok(Cached {
409 id: row.get("id")?,
410 block_depth: row.get("block_depth")?,
411 unblocks_count: row.get("unblocks_count")?,
412 in_cycle: row.get("in_cycle")?,
413 graph_urgency: row.get("graph_urgency")?,
414 })
415 },
416 )?;
417
418 let mut changed = 0usize;
419 for row in current {
420 let id: TaskId = parse_uuid(&row.id)?.into();
421 let want = positions.get(&id).copied().unwrap_or_default();
422 let want_urgency = calculate_graph_urgency(&want);
423
424 let same = i64::from(want.block_depth) == row.block_depth
425 && i64::from(want.unblocks_count) == row.unblocks_count
426 && i32::from(want.in_cycle) == row.in_cycle
427 // Exact comparison is right for a value both sides produce by the
428 // same rounding from the same integer inputs. A tolerance here would
429 // only hide a genuine drift in the scoring function.
430 && (want_urgency - row.graph_urgency).abs() < f64::EPSILON;
431 if same {
432 continue;
433 }
434
435 execute(
436 conn,
437 "UPDATE tasks SET block_depth = ?, unblocks_count = ?, in_cycle = ?, graph_urgency = ?
438 WHERE id = ? AND user_id = ?",
439 params![
440 i64::from(want.block_depth),
441 i64::from(want.unblocks_count),
442 i32::from(want.in_cycle),
443 want_urgency,
444 row.id,
445 user_id.to_string(),
446 ],
447 )?;
448 changed += 1;
449 }
450
451 Ok(changed)
452 }
453
454 /// Whether `blocked` is already reachable from `blocker` by following
455 /// dependencies, which is what makes the new edge a cycle.
456 ///
457 /// Walks ALL edges, not only live ones. A structural cycle that currently runs
458 /// through a completed task is still a cycle: reopening that task would close
459 /// it, and the write path is the one place we can refuse cheaply.
460 ///
461 /// Returns the offending path when there is one, so the refusal can name the
462 /// chain rather than just saying no.
463 fn path_between(graph: &LoadedGraph, from: TaskId, to: TaskId) -> Option<Vec<TaskId>> {
464 let mut seen: HashSet<TaskId> = HashSet::from([from]);
465 // Breadth-first, so the path reported is the shortest one, which is the one
466 // a person can most easily check.
467 let mut queue: VecDeque<TaskId> = VecDeque::from([from]);
468 let mut came_from: HashMap<TaskId, TaskId> = HashMap::new();
469
470 while let Some(node) = queue.pop_front() {
471 if node == to {
472 let mut path = vec![node];
473 let mut cursor = node;
474 while let Some(prev) = came_from.get(&cursor) {
475 path.push(*prev);
476 cursor = *prev;
477 }
478 path.reverse();
479 return Some(path);
480 }
481 for blocker in graph.blockers.get(&node).into_iter().flatten().copied() {
482 if seen.insert(blocker) {
483 came_from.insert(blocker, node);
484 queue.push_back(blocker);
485 }
486 }
487 }
488
489 None
490 }
491
492 /// Confirm a task exists and belongs to the user.
493 fn require_task(conn: &Connection, user_id: UserId, id: TaskId) -> Result<()> {
494 let found: Vec<i64> = query_all(
495 conn,
496 "SELECT 1 FROM tasks WHERE id = ? AND user_id = ?",
497 params![id.to_string(), user_id.to_string()],
498 |row| row.get(0),
499 )?;
500 if found.is_empty() {
501 return Err(CoreError::not_found("Task", id));
502 }
503 Ok(())
504 }
505
506 /// The tasks on one side of a task's edges, as [`LinkedTaskRef`]s.
507 ///
508 /// `column` selects the direction: matching on `blocked_id` yields the task's
509 /// blockers, matching on `blocker_id` yields its dependents.
510 fn linked(
511 conn: &Connection,
512 user_id: UserId,
513 task_id: TaskId,
514 match_column: &str,
515 other_column: &str,
516 ) -> Result<Vec<LinkedTaskRef>> {
517 struct Row {
518 id: String,
519 title: String,
520 status: String,
521 project_name: Option<String>,
522 }
523
524 // The columns are not caller-supplied; they come from the two call sites
525 // below as literals. Kept as parameters rather than duplicating the query.
526 let sql = format!(
527 "SELECT other.id, other.title, other.status, p.name AS project_name
528 FROM task_dependencies d
529 JOIN tasks other ON other.id = d.{other_column}
530 LEFT JOIN projects p ON p.id = other.project_id
531 WHERE d.{match_column} = ? AND other.user_id = ?
532 ORDER BY d.created_at, d.id"
533 );
534
535 let rows: Vec<Row> = query_all(
536 conn,
537 &sql,
538 params![task_id.to_string(), user_id.to_string()],
539 |row| {
540 Ok(Row {
541 id: row.get("id")?,
542 title: row.get("title")?,
543 status: row.get("status")?,
544 project_name: row.get("project_name")?,
545 })
546 },
547 )?;
548
549 rows.into_iter()
550 .map(|row| {
551 Ok(LinkedTaskRef {
552 id: parse_uuid(&row.id)?.into(),
553 title: row.title,
554 status: TaskStatus::from_str_or_default(&row.status),
555 project_name: row.project_name,
556 })
557 })
558 .collect()
559 }
560
561 impl TaskDependencies for SqliteTaskRepository {
562 #[tracing::instrument(skip(self))]
563 fn add_dependency(
564 &self,
565 user_id: UserId,
566 blocked_id: TaskId,
567 blocker_id: TaskId,
568 ) -> Result<TaskDependency> {
569 if blocked_id == blocker_id {
570 return Err(DependencyRejection::SelfEdge {
571 task_id: blocked_id,
572 }
573 .into());
574 }
575
576 let mut conn = self.db.conn()?;
577 let tx = conn.transaction().map_err(CoreError::database)?;
578
579 require_task(&tx, user_id, blocked_id)?;
580 require_task(&tx, user_id, blocker_id)?;
581
582 // The check and the insert share this transaction. Two sessions each
583 // adding one leg of a two-edge cycle would otherwise both read a clean
584 // graph and both commit.
585 let graph = load(&tx, user_id)?;
586 if let Some(path) = path_between(&graph, blocker_id, blocked_id) {
587 return Err(DependencyRejection::WouldCycle { path }.into());
588 }
589
590 let id = TaskDependency::deterministic_id(blocker_id, blocked_id);
591 let created_at = format_datetime_now();
592
593 // Idempotent on the deterministic id: drawing the same edge twice, here
594 // or on another device, converges on one row.
595 execute(
596 &tx,
597 "INSERT INTO task_dependencies (id, blocked_id, blocker_id, created_at, group_id)
598 VALUES (?, ?, ?, ?, (SELECT group_id FROM tasks WHERE id = ?))
599 ON CONFLICT(id) DO NOTHING",
600 params![
601 id.to_string(),
602 blocked_id.to_string(),
603 blocker_id.to_string(),
604 created_at,
605 blocked_id.to_string(),
606 ],
607 )?;
608
609 recompute(&tx, user_id)?;
610 tx.commit().map_err(CoreError::database)?;
611
612 Ok(TaskDependency {
613 id,
614 blocked_id,
615 blocker_id,
616 created_at: parse_datetime(&created_at)?,
617 })
618 }
619
620 #[tracing::instrument(skip(self))]
621 fn remove_dependency(
622 &self,
623 user_id: UserId,
624 blocked_id: TaskId,
625 blocker_id: TaskId,
626 ) -> Result<bool> {
627 let mut conn = self.db.conn()?;
628 let tx = conn.transaction().map_err(CoreError::database)?;
629
630 let removed = execute(
631 &tx,
632 "DELETE FROM task_dependencies
633 WHERE blocked_id = ? AND blocker_id = ?
634 AND EXISTS (SELECT 1 FROM tasks WHERE id = ? AND user_id = ?)",
635 params![
636 blocked_id.to_string(),
637 blocker_id.to_string(),
638 blocked_id.to_string(),
639 user_id.to_string(),
640 ],
641 )?;
642
643 if removed == 0 {
644 return Ok(false);
645 }
646
647 recompute(&tx, user_id)?;
648 tx.commit().map_err(CoreError::database)?;
649 Ok(true)
650 }
651
652 fn list_blockers(&self, user_id: UserId, task_id: TaskId) -> Result<Vec<LinkedTaskRef>> {
653 let conn = self.db.conn()?;
654 linked(&conn, user_id, task_id, "blocked_id", "blocker_id")
655 }
656
657 fn list_dependents(&self, user_id: UserId, task_id: TaskId) -> Result<Vec<LinkedTaskRef>> {
658 let conn = self.db.conn()?;
659 linked(&conn, user_id, task_id, "blocker_id", "blocked_id")
660 }
661
662 #[tracing::instrument(skip(self))]
663 fn task_graph(&self, user_id: UserId, project_id: Option<ProjectId>) -> Result<TaskGraph> {
664 let conn = self.db.conn()?;
665 let graph = load(&conn, user_id)?;
666 let (positions, cycles) = positions(&graph);
667
668 // Which tasks the caller asked about. A project scope selects the
669 // project's tasks, then pulls in whatever their edges reach: a blocker
670 // in another project is the reason a task in this one is not ready, and
671 // cutting it would show that task as available.
672 let in_scope: HashSet<TaskId> = match project_id {
673 None => graph.tasks.keys().copied().collect(),
674 Some(pid) => {
675 let seeds: Vec<TaskId> = graph
676 .tasks
677 .values()
678 .filter(|t| t.project_id == Some(pid))
679 .map(|t| t.id)
680 .collect();
681 let mut reached: HashSet<TaskId> = seeds.iter().copied().collect();
682 let mut queue: VecDeque<TaskId> = seeds.into_iter().collect();
683 while let Some(node) = queue.pop_front() {
684 let neighbours = graph
685 .blockers
686 .get(&node)
687 .into_iter()
688 .flatten()
689 .chain(graph.dependents.get(&node).into_iter().flatten())
690 .copied();
691 for n in neighbours {
692 if reached.insert(n) {
693 queue.push_back(n);
694 }
695 }
696 }
697 reached
698 }
699 };
700
701 // Only tasks that actually touch an edge. An isolated task is not part
702 // of any graph worth drawing, and including every unlinked task would
703 // bury the structure in noise.
704 let mut nodes: Vec<TaskGraphNode> = graph
705 .tasks
706 .values()
707 .filter(|t| in_scope.contains(&t.id))
708 .filter(|t| graph.blockers.contains_key(&t.id) || graph.dependents.contains_key(&t.id))
709 .map(|t| TaskGraphNode {
710 id: t.id,
711 title: t.title.clone(),
712 status: t.status.clone(),
713 priority: t.priority.clone(),
714 project_id: t.project_id,
715 position: positions.get(&t.id).copied().unwrap_or_default(),
716 })
717 .collect();
718 nodes.sort_by_key(|n| n.id.to_string());
719
720 let node_ids: HashSet<TaskId> = nodes.iter().map(|n| n.id).collect();
721 let edges = graph
722 .edges
723 .iter()
724 .filter(|e| node_ids.contains(&e.blocked_id) && node_ids.contains(&e.blocker_id))
725 .cloned()
726 .collect();
727 let cycles = cycles
728 .into_iter()
729 .filter(|c| c.iter().all(|id| node_ids.contains(id)))
730 .collect();
731
732 Ok(TaskGraph {
733 nodes,
734 edges,
735 cycles,
736 })
737 }
738
739 #[tracing::instrument(skip(self))]
740 fn list_ready(
741 &self,
742 user_id: UserId,
743 project_id: Option<ProjectId>,
744 max_depth: u32,
745 limit: Option<i64>,
746 ) -> Result<Vec<Task>> {
747 let conn = self.db.conn()?;
748
749 // Reads the cached column rather than recomputing, which is the whole
750 // point of caching it: this is the query the task list and every mode of
751 // /threads run, and it stays a plain indexed scan.
752 let mut sql = format!(
753 "SELECT {TASK_SELECT_COLUMNS}
754 FROM tasks t
755 LEFT JOIN projects p ON t.project_id = p.id AND p.user_id = ?
756 LEFT JOIN contacts ct ON ct.id = t.contact_id
757 WHERE t.user_id = ?
758 AND t.status IN ('Pending', 'Started')
759 AND t.block_depth <= ?
760 -- A task on a cycle carries depth 0 because a cycle has no
761 -- meaningful depth, and it is the one thing that must never read
762 -- as ready: nothing on it can ever open.
763 AND t.in_cycle = 0
764 AND (t.snoozed_until IS NULL OR t.snoozed_until <= ?)"
765 );
766
767 let mut binds: Vec<Box<dyn rusqlite::ToSql>> = vec![
768 Box::new(user_id.to_string()),
769 Box::new(user_id.to_string()),
770 Box::new(i64::from(max_depth)),
771 Box::new(format_datetime_now()),
772 ];
773
774 if let Some(pid) = project_id {
775 sql.push_str(" AND t.project_id = ?");
776 binds.push(Box::new(pid.to_string()));
777 }
778
779 // Shallowest first, then by effective urgency. Depth leads because a
780 // caller that asked for depth 2 still wants the work it can start today
781 // at the top; within one depth the score decides, and that score already
782 // carries the unblocks bonus.
783 sql.push_str(
784 " ORDER BY t.block_depth ASC, (t.urgency + t.graph_urgency) DESC, t.created_at DESC",
785 );
786
787 if let Some(limit) = limit {
788 sql.push_str(" LIMIT ?");
789 binds.push(Box::new(limit));
790 }
791
792 let rows: Vec<TaskRowWithProject> = query_all(
793 &conn,
794 &sql,
795 rusqlite::params_from_iter(binds.iter().map(std::convert::AsRef::as_ref)),
796 TaskRowWithProject::from_row,
797 )?;
798
799 rows_to_tasks(&conn, rows)
800 }
801
802 #[tracing::instrument(skip(self))]
803 fn plan_gates(
804 &self,
805 user_id: UserId,
806 window_start: chrono::DateTime<chrono::Utc>,
807 window_end: chrono::DateTime<chrono::Utc>,
808 ) -> Result<HashMap<TaskId, goingson_core::PlanGate>> {
809 let conn = self.db.conn()?;
810 let graph = load(&conn, user_id)?;
811
812 // What is already in the plan, and when. `scheduled_start` inside the
813 // window is the whole test: a blocker parked next Tuesday is not in
814 // today's plan and must not unlock anything in it, or the day becomes
815 // one nobody can actually execute.
816 struct Scheduled {
817 id: String,
818 scheduled_start: String,
819 }
820 let scheduled_rows: Vec<Scheduled> = query_all(
821 &conn,
822 "SELECT id, scheduled_start FROM tasks
823 WHERE user_id = ? AND scheduled_start IS NOT NULL
824 AND scheduled_start >= ? AND scheduled_start <= ?
825 AND status NOT IN ('Completed', 'Deleted')",
826 params![
827 user_id.to_string(),
828 crate::utils::format_datetime(&window_start),
829 crate::utils::format_datetime(&window_end),
830 ],
831 |row| {
832 Ok(Scheduled {
833 id: row.get("id")?,
834 scheduled_start: row.get("scheduled_start")?,
835 })
836 },
837 )?;
838
839 let mut in_plan: HashMap<TaskId, chrono::DateTime<chrono::Utc>> = HashMap::new();
840 for row in scheduled_rows {
841 in_plan.insert(
842 parse_uuid(&row.id)?.into(),
843 parse_datetime(&row.scheduled_start)?,
844 );
845 }
846
847 let mut gates = HashMap::new();
848 for id in graph.tasks.keys().copied() {
849 let live: Vec<TaskId> = graph.live_blockers(id).collect();
850 // No gate at all rather than an empty one: an absent entry is the
851 // "nothing is in the way" signal, so a caller never has to look
852 // inside to find out.
853 if live.is_empty() {
854 continue;
855 }
856
857 let after: Vec<LinkedTaskRef> = live
858 .iter()
859 .filter_map(|b| graph.tasks.get(b))
860 .map(|t| LinkedTaskRef {
861 id: t.id,
862 title: t.title.clone(),
863 status: t.status.clone(),
864 project_name: t.project_name.clone(),
865 })
866 .collect();
867
868 let unlocked_by_plan = live.iter().all(|b| in_plan.contains_key(b));
869 // Only meaningful once this task is itself in the plan; an
870 // unscheduled task has no start to be earlier than.
871 let out_of_order = in_plan.get(&id).is_some_and(|mine| {
872 live.iter()
873 .filter_map(|b| in_plan.get(b))
874 .any(|theirs| theirs > mine)
875 });
876
877 gates.insert(
878 id,
879 goingson_core::PlanGate {
880 after,
881 unlocked_by_plan,
882 out_of_order,
883 },
884 );
885 }
886
887 Ok(gates)
888 }
889
890 #[tracing::instrument(skip(self))]
891 fn recompute_graph(&self, user_id: UserId) -> Result<usize> {
892 let mut conn = self.db.conn()?;
893 let tx = conn.transaction().map_err(CoreError::database)?;
894 let changed = recompute(&tx, user_id)?;
895 tx.commit().map_err(CoreError::database)?;
896 Ok(changed)
897 }
898
899 fn list_all_dependencies(&self, user_id: UserId) -> Result<Vec<TaskDependency>> {
900 let conn = self.db.conn()?;
901 Ok(load(&conn, user_id)?.edges)
902 }
903
904 fn restore_dependency(&self, user_id: UserId, dependency: &TaskDependency) -> Result<()> {
905 let conn = self.db.conn()?;
906 execute(
907 &conn,
908 "INSERT INTO task_dependencies (id, blocked_id, blocker_id, created_at, group_id)
909 SELECT ?, ?, ?, ?, (SELECT group_id FROM tasks WHERE id = ?)
910 WHERE EXISTS (SELECT 1 FROM tasks WHERE id = ? AND user_id = ?)
911 AND EXISTS (SELECT 1 FROM tasks WHERE id = ? AND user_id = ?)
912 ON CONFLICT(id) DO NOTHING",
913 params![
914 dependency.id.to_string(),
915 dependency.blocked_id.to_string(),
916 dependency.blocker_id.to_string(),
917 crate::utils::format_datetime(&dependency.created_at),
918 dependency.blocked_id.to_string(),
919 dependency.blocked_id.to_string(),
920 user_id.to_string(),
921 dependency.blocker_id.to_string(),
922 user_id.to_string(),
923 ],
924 )?;
925 Ok(())
926 }
927 }
928