Skip to content
Sign up

flagon-io/g1t

public
00

Where people and agents ship software together. The open-source git platform for the whole job: issues, agents, checks and deploys to the edge.

gitagentscloudflarerustforgemcp
797 linesCodeBlameRaw
1//! The plan (`g1t_contracts::work::plan`): where each open work item of a
2//! workspace stands, its place there, and what to take next
3//! (migrations/0056_plan.sql). `plan_work_item` moves one, `get_plan`
4//! reads a workspace's within its bounds, `next_work_item` picks what is
5//! ready, and `configure_plan` sets Now's cap. `list_work_items` with a
6//! `horizon` reads one horizon in the plan's order (reading.rs).
7
8use g1t_contracts::access::{self, Capability};
9use g1t_contracts::events::WorkItemEvent;
10use g1t_contracts::time::rfc3339;
11use g1t_contracts::work::*;
12use g1t_contracts::{FailureCode, Outcome, User, Viewer};
13use g1t_kit::now_ms;
14use serde::Deserialize;
15use worker::Result;
16use worker::wasm_bindgen::JsValue;
17
18use super::rows::{Held, ITEM_COLUMNS, ItemRow, no_item, workspace_of};
19use crate::{Work, optional};
20
21/// An open work item's horizon, with NULL (rows from before the plan, and
22/// ones opened since) read as Triage.
23pub(crate) const HORIZON: &str = "COALESCE(work_items.horizon, 'triage')";
24
25/// The plan's order within a horizon: urgent first, then by rank, those
26/// with none last, newest first among them. Triage has no ranks, so it is
27/// newest first.
28pub(crate) const PLAN_ORDER: &str = "work_items.urgent DESC, work_items.rank IS NULL, work_items.rank, work_items.number DESC";
29
30/// The open, numbered work items of workspace `?1` a viewer may see: all
31/// of them for a member (`?2` = 1), else those filed in a public
32/// repository or one of `?3` (a JSON array of repository ids they have a
33/// role on). With `?4`, only those linked to that repository.
34const OPEN_VISIBLE: &str = "work_items.workspace = ?1 AND work_items.state = 'open' AND work_items.number IS NOT NULL
35 AND (?4 IS NULL OR EXISTS (SELECT 1 FROM work_item_repos r WHERE r.item_id = work_items.id AND r.repo_id = ?4))
36 AND (?2 = 1 OR EXISTS (SELECT 1 FROM work_item_repos o JOIN repo_workspaces rw ON rw.repo_id = o.repo_id
37 WHERE o.item_id = work_items.id AND o.role = 'origin'
38 AND (rw.private = 0 OR o.repo_id IN (SELECT value FROM json_each(?3)))))";
39
40/// What is ready to take, given `?5` (now, RFC 3339) and `?6` (the
41/// viewer's username, lowercase): nobody holds a live claim on it, nothing
42/// it is blocked by is open, and it is assigned to nobody or to them.
43const READY: &str = "NOT EXISTS (SELECT 1 FROM claims c WHERE c.item_id = work_items.id AND c.state = 'active' AND c.expires_at > ?5)
44 AND NOT EXISTS (SELECT 1 FROM json_each(work_items.blocked_by) j JOIN work_items b ON b.id = j.value WHERE b.state = 'open')
45 AND (json_array_length(work_items.assignees) = 0
46 OR EXISTS (SELECT 1 FROM json_each(work_items.assignees) WHERE lower(json_each.value) = ?6))";
47
48fn open_items(condition: &str, limit: u32) -> String {
49 format!("SELECT {ITEM_COLUMNS} FROM work_items WHERE {OPEN_VISIBLE} AND {condition} ORDER BY {PLAN_ORDER} LIMIT {limit}")
50}
51
52/// How many open work items each horizon holds.
53const COUNTS: &str = "SELECT COALESCE(work_items.horizon, 'triage') AS horizon, count(*) AS count FROM work_items
54 WHERE {OPEN_VISIBLE} GROUP BY 1";
55
56/// Later, by the repository each work item is filed in.
57const LATER_GROUPS: &str = "SELECT (SELECT COALESCE(rw.path, o.path) FROM work_item_repos o
58 LEFT JOIN repo_workspaces rw ON rw.repo_id = o.repo_id
59 WHERE o.item_id = work_items.id AND o.role = 'origin') AS repo, count(*) AS count
60 FROM work_items WHERE {OPEN_VISIBLE} AND COALESCE(work_items.horizon, 'triage') = 'later'
61 GROUP BY 1 ORDER BY 2 DESC, 1 LIMIT 50";
62
63/// The ranks on horizon `?2` of workspace `?1` around a place, leaving out
64/// work item `?3`: the last one ahead of `?4` (none: no bound) and the
65/// first one after `?5`.
66const NEIGHBOURS: &str = "SELECT
67 (SELECT max(rank) FROM work_items WHERE workspace = ?1 AND state = 'open' AND COALESCE(horizon, 'triage') = ?2
68 AND id <> ?3 AND rank IS NOT NULL AND (?4 IS NULL OR rank < ?4)) AS ahead,
69 (SELECT min(rank) FROM work_items WHERE workspace = ?1 AND state = 'open' AND COALESCE(horizon, 'triage') = ?2
70 AND id <> ?3 AND rank IS NOT NULL AND (?5 IS NULL OR rank > ?5)) AS behind";
71
72/// What the record says of each work item in the JSON array of ids `?1`:
73/// its hill's facts, when anything last happened on it, and where g1t
74/// last said it stands.
75pub(crate) const FACTS: &str = "SELECT w.number,
76 (SELECT count(*) FROM decisions d WHERE d.item_id = w.id
77 AND d.kind NOT IN ('claim', 'merge', 'integration_failure', 'integration_passed')) AS decisions,
78 (SELECT count(*) FROM pulls p WHERE p.item_id = w.id AND p.status = 'draft') AS draft_pulls,
79 (SELECT count(*) FROM session_touches t JOIN sessions s ON s.id = t.session_id
80 WHERE s.item_id = w.id AND t.kind = 'commit') AS commits,
81 (SELECT count(*) FROM pulls p WHERE p.item_id = w.id AND p.status = 'open') AS open_pulls,
82 (SELECT count(*) FROM pulls p WHERE p.item_id = w.id AND p.status = 'merged') AS merged_pulls,
83 (SELECT count(*) FROM work_item_repos r WHERE r.item_id = w.id AND r.landed_at IS NOT NULL) AS landed_targets,
84 max(w.updated_at,
85 COALESCE((SELECT max(s.last_seen_at) FROM sessions s WHERE s.item_id = w.id), ''),
86 COALESCE((SELECT max(d.created_at) FROM decisions d WHERE d.item_id = w.id), ''),
87 COALESCE((SELECT max(p.updated_at) FROM pulls p WHERE p.item_id = w.id), '')) AS active_at,
88 w.standing, w.standing_at
89 FROM work_items w WHERE w.id IN (SELECT value FROM json_each(?1))";
90
91#[derive(Deserialize)]
92pub(crate) struct FactsRow {
93 pub number: u32,
94 #[serde(default)]
95 pub decisions: u32,
96 #[serde(default)]
97 pub draft_pulls: u32,
98 #[serde(default)]
99 pub commits: u32,
100 #[serde(default)]
101 pub open_pulls: u32,
102 #[serde(default)]
103 pub merged_pulls: u32,
104 #[serde(default)]
105 pub landed_targets: u32,
106 #[serde(default)]
107 pub active_at: Option<String>,
108 #[serde(default)]
109 pub standing: Option<String>,
110 #[serde(default)]
111 pub standing_at: Option<String>,
112}
113
114impl FactsRow {
115 pub(crate) fn hill(&self) -> Hill {
116 hill_of(&HillFacts {
117 decisions: self.decisions,
118 draft_pulls: self.draft_pulls,
119 commits: self.commits,
120 open_pulls: self.open_pulls,
121 merged_pulls: self.merged_pulls,
122 landed_targets: self.landed_targets,
123 })
124 }
125
126 /// How it stands, quiet when nothing has happened since `quiet_before`
127 /// (RFC 3339).
128 pub(crate) fn progress(self, quiet_before: &str) -> Progress {
129 let hill = self.hill();
130 let active_at = self.active_at.filter(|at| !at.is_empty());
131 Progress {
132 number: self.number,
133 hill,
134 quiet: active_at.as_deref().is_some_and(|at| at < quiet_before),
135 active_at,
136 standing: self.standing,
137 standing_at: self.standing_at,
138 }
139 }
140}
141
142/// The RFC 3339 time `days` before `now_ms`.
143pub(crate) fn days_before(now_ms: u64, days: u32) -> String {
144 rfc3339(now_ms.saturating_sub(u64::from(days) * 86_400_000))
145}
146
147/// What closed from Now since `?5`, the most recently closed first.
148const LANDED: &str = "SELECT {ITEM_COLUMNS} FROM work_items
149 WHERE work_items.workspace = ?1 AND work_items.state = 'closed' AND work_items.number IS NOT NULL
150 AND work_items.horizon = 'now' AND work_items.closed_at >= ?5
151 AND (?4 IS NULL OR EXISTS (SELECT 1 FROM work_item_repos r WHERE r.item_id = work_items.id AND r.repo_id = ?4))
152 AND (?2 = 1 OR EXISTS (SELECT 1 FROM work_item_repos o JOIN repo_workspaces rw ON rw.repo_id = o.repo_id
153 WHERE o.item_id = work_items.id AND o.role = 'origin'
154 AND (rw.private = 0 OR o.repo_id IN (SELECT value FROM json_each(?3)))))
155 ORDER BY work_items.closed_at DESC LIMIT 10";
156
157/// How many have waited in Triage since before `?5`, and how many were set
158/// aside since `?6`.
159const WAITING: &str = "SELECT
160 (SELECT count(*) FROM work_items WHERE {OPEN_VISIBLE} AND COALESCE(work_items.horizon, 'triage') = 'triage'
161 AND work_items.created_at < ?5) AS triage_waiting,
162 (SELECT count(*) FROM work_items WHERE {OPEN_VISIBLE} AND work_items.horizon = 'set_aside'
163 AND work_items.planned_at >= ?6) AS set_aside_recent";
164
165#[derive(Deserialize)]
166struct Waiting {
167 #[serde(default)]
168 triage_waiting: u32,
169 #[serde(default)]
170 set_aside_recent: u32,
171}
172
173#[derive(Deserialize)]
174struct Neighbours {
175 ahead: Option<String>,
176 behind: Option<String>,
177}
178
179#[derive(Deserialize)]
180struct Count {
181 horizon: Option<String>,
182 count: u32,
183}
184
185#[derive(Deserialize)]
186struct Group {
187 repo: Option<String>,
188 count: u32,
189}
190
191/// Who a viewer is in a workspace, for the queries above: whether they
192/// are a member, and the repositories they were given a role on.
193fn standing(viewer: Option<&User>, workspace: &str) -> Result<(JsValue, JsValue)> {
194 let member = viewer.and_then(|user| access::workspace_role(user, workspace)).is_some();
195 let granted: Vec<&str> = viewer
196 .map(|user| user.grants.iter().filter(|grant| grant.workspace.eq_ignore_ascii_case(workspace)).map(|grant| grant.repo_id.as_str()).collect())
197 .unwrap_or_default();
198 Ok((u32::from(member).into(), serde_json::to_string(&granted)?.into()))
199}
200
201/// What placing a work item asks for, once read: its horizon after, and
202/// where on it.
203#[derive(Debug, PartialEq, Eq)]
204pub(crate) enum Place<'a> {
205 /// Leave its rank as it is.
206 Stay,
207 /// Last on the horizon.
208 Bottom,
209 Top,
210 /// Just before the work item of this rank.
211 Before(&'a str),
212 /// Just after the work item of this rank.
213 After(&'a str),
214}
215
216/// What a move says in the work item's conversation, if anything: a new
217/// horizon, and a change to urgent.
218pub(crate) fn move_note(from: Horizon, to: Horizon, urgent_before: bool, urgent: bool) -> Option<String> {
219 let mut parts = Vec::new();
220 if from != to {
221 parts.push(format!("moved this from {} to {}", from.label(), to.label()));
222 }
223 match (urgent_before, urgent) {
224 (false, true) => parts.push("marked this urgent".to_owned()),
225 (true, false) => parts.push("said this is no longer urgent".to_owned()),
226 _ => {}
227 }
228 (!parts.is_empty()).then(|| parts.join(" and "))
229}
230
231/// A budget as given: a positive number of dollars up to the most.
232pub(crate) fn valid_budget(budget: Option<f64>) -> std::result::Result<Option<f64>, String> {
233 match budget {
234 None => Ok(None),
235 Some(dollars) if dollars.is_finite() && dollars > 0.0 && dollars <= MAX_BUDGET_USD => Ok(Some((dollars * 100.0).round() / 100.0)),
236 Some(_) => Err(format!("A budget is a number of dollars above 0 and up to {MAX_BUDGET_USD}.")),
237 }
238}
239
240impl Work {
241 /// How many work items Now holds before workspace `workspace`'s plan
242 /// says it is over.
243 async fn now_cap(&self, workspace: &str) -> Result<u32> {
244 #[derive(Deserialize)]
245 struct Cap {
246 now_cap: u32,
247 }
248 Ok(self
249 .db
250 .prepare("SELECT now_cap FROM plan_settings WHERE workspace = ?")
251 .bind(&[workspace.into()])?
252 .first::<Cap>(None)
253 .await?
254 .map_or(DEFAULT_NOW_CAP, |row| row.now_cap))
255 }
256
257 /// The open work item numbered `number`, to place another next to.
258 async fn anchor(&self, held: &Held, number: u32) -> Result<Outcome<Held>> {
259 if number == held.item.number {
260 return Ok(Outcome::fail(FailureCode::Invalid, "A work item can't be placed next to itself."));
261 }
262 match self.item(&held.item.workspace, number).await? {
263 Some(anchor) if anchor.item.state == State::Open => Ok(Outcome::Ok(anchor)),
264 Some(_) => Ok(Outcome::fail(FailureCode::Invalid, format!("#{number} is closed, so it has no place in the plan."))),
265 None => Ok(Outcome::fail(FailureCode::NotFound, format!("#{number} is not a work item of {}.", held.item.workspace))),
266 }
267 }
268
269 /// The rank `place` gives work item `held` on `horizon`.
270 async fn rank_for(&self, held: &Held, horizon: Horizon, place: &Place<'_>) -> Result<Option<String>> {
271 if horizon == Horizon::Triage {
272 return Ok(None);
273 }
274 if *place == Place::Stay {
275 // A rank it already has there; one with none goes last.
276 if held.item.horizon == horizon {
277 if let Some(rank) = self.stored_rank(&held.item.id).await? {
278 return Ok(Some(rank));
279 }
280 }
281 return Box::pin(self.rank_for(held, horizon, &Place::Bottom)).await;
282 }
283 // With no bound given, `ahead` is the last rank there and `behind`
284 // the first.
285 let (upper, lower): (JsValue, JsValue) = match place {
286 Place::Before(rank) => ((*rank).into(), JsValue::NULL),
287 Place::After(rank) => (JsValue::NULL, (*rank).into()),
288 _ => (JsValue::NULL, JsValue::NULL),
289 };
290 let found = self
291 .db
292 .prepare(NEIGHBOURS)
293 .bind(&[held.item.workspace.as_str().into(), horizon.as_str().into(), held.item.id.as_str().into(), upper, lower])?
294 .first::<Neighbours>(None)
295 .await?
296 .unwrap_or(Neighbours { ahead: None, behind: None });
297 Ok(Some(match place {
298 Place::Before(rank) => rank_between(found.ahead.as_deref(), Some(rank)),
299 Place::After(rank) => rank_between(Some(rank), found.behind.as_deref()),
300 Place::Top => rank_between(None, found.behind.as_deref()),
301 Place::Bottom | Place::Stay => rank_between(found.ahead.as_deref(), None),
302 }))
303 }
304
305 async fn stored_rank(&self, id: &str) -> Result<Option<String>> {
306 #[derive(Deserialize)]
307 struct Rank {
308 rank: Option<String>,
309 }
310 Ok(self.db.prepare("SELECT rank FROM work_items WHERE id = ?").bind(&[id.into()])?.first::<Rank>(None).await?.and_then(|row| row.rank))
311 }
312
313 /// Gives `anchor` a rank, last on its horizon, if it has none, so
314 /// another can be placed next to it. Returns its rank.
315 async fn ranked(&self, anchor: &Held) -> Result<String> {
316 if let Some(rank) = self.stored_rank(&anchor.item.id).await? {
317 return Ok(rank);
318 }
319 let rank = self.rank_for(anchor, anchor.item.horizon, &Place::Bottom).await?.unwrap_or_else(|| rank_between(None, None));
320 self.db
321 .prepare("UPDATE work_items SET rank = ? WHERE id = ? AND rank IS NULL")
322 .bind(&[rank.as_str().into(), anchor.item.id.as_str().into()])?
323 .run()
324 .await?;
325 Ok(rank)
326 }
327
328 // --- RPC methods ----------------------------------------------------------
329
330 pub(crate) async fn plan_work_item(&self, a: PlanWorkItemArgs) -> Result<Outcome<WorkItem>> {
331 let workspace = a.workspace.trim().to_lowercase();
332 let held = check!(self.item_for(&Some(a.actor.clone()), &workspace, a.number, Capability::Read).await?);
333 if held.item.state == State::Closed {
334 return Ok(Outcome::fail(FailureCode::Conflict, "This work item is closed, so it has no place in the plan. Reopen it first."));
335 }
336 let positions = usize::from(a.before.is_some()) + usize::from(a.after.is_some()) + usize::from(a.top);
337 if positions > 1 {
338 return Ok(Outcome::fail(FailureCode::Invalid, "Say one of before, after or top."));
339 }
340 let budget = match a.budget_usd {
341 Some(budget) => match valid_budget(budget) {
342 Ok(budget) => Some(budget),
343 Err(message) => return Ok(Outcome::fail(FailureCode::Invalid, message)),
344 },
345 None => None,
346 };
347 // Next to another: on that one's horizon.
348 let anchor = match a.before.or(a.after) {
349 Some(number) => Some(check!(self.anchor(&held, number).await?)),
350 None => None,
351 };
352 let from = held.item.horizon;
353 // Urgent is first in Now.
354 let mut horizon = a.horizon.or(anchor.as_ref().map(|anchor| anchor.item.horizon)).unwrap_or(from);
355 if a.urgent == Some(true) {
356 if a.horizon.is_some_and(|asked| asked != Horizon::Now) {
357 return Ok(Outcome::fail(FailureCode::Invalid, "An urgent work item is in Now."));
358 }
359 horizon = Horizon::Now;
360 }
361 if let (Some(anchor), Some(asked)) = (&anchor, a.horizon) {
362 if anchor.item.horizon != asked {
363 return Ok(Outcome::fail(
364 FailureCode::Invalid,
365 format!("#{} is in {}, not {}.", anchor.item.number, anchor.item.horizon.label(), asked.label()),
366 ));
367 }
368 }
369 if horizon == Horizon::Triage && positions > 0 {
370 return Ok(Outcome::fail(FailureCode::Invalid, "Triage is in the order work items came in, so nothing is placed in it."));
371 }
372 // Who may: the Triage role to plan; the right to manage milestones
373 // for what the workspace commits to.
374 check!(held.allowed(Some(&a.actor), Capability::Triage));
375 let reranked = positions > 0 || horizon != from;
376 let commits = (horizon.committed() && reranked) || (from.committed() && horizon != from) || a.urgent.is_some_and(|urgent| urgent != held.item.urgent);
377 if commits && !access::can_item(Some(&a.actor), &workspace, None, Capability::ManageLabels) {
378 return Ok(Outcome::fail(
379 FailureCode::Forbidden,
380 format!("Deciding what is in Now takes the right to manage milestones. {}", access::needs_on_item(Capability::ManageLabels, &workspace)),
381 ));
382 }
383 let anchor_rank = match &anchor {
384 Some(anchor) => Some(self.ranked(anchor).await?),
385 None => None,
386 };
387 let place = match (&anchor_rank, a.before.is_some(), a.top) {
388 (Some(rank), true, _) => Place::Before(rank),
389 (Some(rank), false, _) => Place::After(rank),
390 (None, _, true) => Place::Top,
391 (None, _, false) if horizon != from => Place::Bottom,
392 _ => Place::Stay,
393 };
394 let rank = self.rank_for(&held, horizon, &place).await?;
395 let urgent = a.urgent.unwrap_or(held.item.urgent) && horizon == Horizon::Now;
396 let appetite = a.appetite.unwrap_or(held.item.appetite);
397 let budget = budget.unwrap_or(held.item.budget_usd);
398 let now = rfc3339(now_ms());
399 let planned_at = if reranked { Some(now.clone()) } else { held.item.planned_at.clone() };
400 self.db
401 .prepare(
402 "UPDATE work_items SET horizon = ?, rank = ?, appetite = ?, budget_usd = ?, urgent = ?, planned_at = ?
403 WHERE id = ? AND state = 'open'",
404 )
405 .bind(&[
406 horizon.as_str().into(),
407 optional(&rank),
408 optional(&appetite.map(|appetite| appetite.as_str().to_owned())),
409 budget.map_or(JsValue::NULL, JsValue::from),
410 u32::from(urgent).into(),
411 optional(&planned_at),
412 held.item.id.as_str().into(),
413 ])?
414 .run()
415 .await?;
416 let Some(after) = self.item_by_id(&held.item.id).await? else {
417 return Ok(no_item());
418 };
419 self.publish_item(
420 "work_item.planned",
421 &after,
422 Some(&a.actor),
423 WorkItemEvent {
424 horizon: Some(horizon),
425 from_horizon: (horizon != from).then_some(from),
426 appetite: after.item.appetite,
427 urgent: Some(urgent),
428 ..Self::item_event(&after)
429 },
430 )
431 .await?;
432 if let Some(note) = move_note(from, horizon, held.item.urgent, urgent) {
433 self.note_item(&after.item.id, &a.actor, &note).await?;
434 }
435 Ok(Outcome::Ok(after.shown(Some(&a.actor))))
436 }
437
438 pub(crate) async fn get_plan(&self, a: GetPlanArgs) -> Result<Outcome<Plan>> {
439 let workspace = workspace_of(&a.workspace, a.repo.as_ref());
440 if workspace.is_empty() {
441 return Ok(Outcome::fail(FailureCode::Invalid, "Say which workspace."));
442 }
443 let repo_id = match &a.repo {
444 Some(path) => {
445 let repo = check!(self.repo(path, &a.viewer).await?);
446 if !repo.namespace.eq_ignore_ascii_case(&workspace) {
447 return Ok(Outcome::Ok(Plan { workspace, now_cap: self.now_cap(&repo.namespace.to_lowercase()).await?, ..Plan::default() }));
448 }
449 Some(repo.id)
450 }
451 None => None,
452 };
453 let (member, granted) = standing(a.viewer.as_ref(), &workspace)?;
454 let bound = |extra: Option<JsValue>| {
455 let mut values = vec![workspace.as_str().into(), member.clone(), granted.clone(), optional(&repo_id)];
456 values.extend(extra);
457 values
458 };
459 let shown = if a.all { crate::LIST_PAGE } else { NEXT_SHOWN };
460 let now_ms = now_ms();
461 let week = days_before(now_ms, WEEK_DAYS);
462 let mut waiting_values = bound(Some(days_before(now_ms, TRIAGE_NUDGE_DAYS).into()));
463 waiting_values.push(days_before(now_ms, 30).into());
464 let statements = vec![
465 self.db.prepare(open_items(&format!("{HORIZON} = 'now'"), crate::LIST_PAGE)).bind(&bound(None))?,
466 self.db.prepare(open_items(&format!("{HORIZON} = 'next'"), shown)).bind(&bound(None))?,
467 self.db.prepare(COUNTS.replace("{OPEN_VISIBLE}", OPEN_VISIBLE)).bind(&bound(None))?,
468 self.db.prepare(LATER_GROUPS.replace("{OPEN_VISIBLE}", OPEN_VISIBLE)).bind(&bound(None))?,
469 self.db.prepare(LANDED.replace("{ITEM_COLUMNS}", ITEM_COLUMNS)).bind(&bound(Some(week.as_str().into())))?,
470 self.db.prepare(WAITING.replace("{OPEN_VISIBLE}", OPEN_VISIBLE)).bind(&waiting_values)?,
471 ];
472 let results = self.timing.db(6, self.db.batch(statements)).await?;
473 let rows = |index: usize| results.get(index).ok_or_else(|| worker::Error::RustError("short batch".into()));
474 let now = shown_rows(rows(0)?.results::<ItemRow>()?, &a.viewer);
475 let next = shown_rows(rows(1)?.results::<ItemRow>()?, &a.viewer);
476 let counts = rows(2)?.results::<Count>()?;
477 let total = |horizon: Horizon| {
478 counts.iter().filter(|row| row.horizon.as_deref().and_then(Horizon::parse).unwrap_or_default() == horizon).map(|row| row.count).sum()
479 };
480 let later = rows(3)?.results::<Group>()?.into_iter().map(|group| PlanGroup { repo: group.repo, count: group.count }).collect();
481 let landed = shown_rows(rows(4)?.results::<ItemRow>()?, &a.viewer);
482 let waiting = rows(5)?.results::<Waiting>()?.into_iter().next().unwrap_or(Waiting { triage_waiting: 0, set_aside_recent: 0 });
483 // How each in Now stands, in Now's order.
484 let ids: Vec<&str> = now.iter().map(|item| item.id.as_str()).collect();
485 let mut facts: Vec<FactsRow> = if ids.is_empty() {
486 Vec::new()
487 } else {
488 self.db.prepare(FACTS).bind(&[serde_json::to_string(&ids)?.into()])?.all().await?.results::<FactsRow>()?
489 };
490 let quiet_before = days_before(now_ms, QUIET_DAYS);
491 let progress = now
492 .iter()
493 .filter_map(|item| {
494 let at = facts.iter().position(|row| row.number == item.number)?;
495 Some(facts.swap_remove(at).progress(&quiet_before))
496 })
497 .collect();
498 Ok(Outcome::Ok(Plan {
499 progress,
500 landed,
501 triage_waiting: waiting.triage_waiting,
502 set_aside_recent: waiting.set_aside_recent,
503 now_cap: self.now_cap(&workspace).await?,
504 now,
505 next,
506 next_total: total(Horizon::Next),
507 triage_total: total(Horizon::Triage),
508 later_total: total(Horizon::Later),
509 later,
510 set_aside_total: total(Horizon::SetAside),
511 workspace,
512 }))
513 }
514
515 pub(crate) async fn next_work_item(&self, a: NextWorkItemArgs) -> Result<Outcome<Option<WorkItem>>> {
516 let workspace = workspace_of(&a.workspace, a.repo.as_ref());
517 let repo_id = match &a.repo {
518 Some(path) => {
519 let repo = check!(self.repo(path, &a.viewer).await?);
520 if !repo.namespace.eq_ignore_ascii_case(&workspace) {
521 return Ok(Outcome::Ok(None));
522 }
523 Some(repo.id)
524 }
525 None => None,
526 };
527 let (member, granted) = standing(a.viewer.as_ref(), &workspace)?;
528 let username = a.viewer.as_ref().map(|user| user.username.to_lowercase()).unwrap_or_default();
529 let sql = format!(
530 "SELECT {ITEM_COLUMNS} FROM work_items WHERE {OPEN_VISIBLE} AND {HORIZON} IN ('now', 'next') AND {READY}
531 ORDER BY CASE {HORIZON} WHEN 'now' THEN 0 ELSE 1 END, {PLAN_ORDER} LIMIT 20"
532 );
533 let rows = self
534 .db
535 .prepare(sql)
536 .bind(&[
537 workspace.as_str().into(),
538 member,
539 granted,
540 optional(&repo_id),
541 rfc3339(now_ms()).into(),
542 username.into(),
543 ])?
544 .all()
545 .await?
546 .results::<ItemRow>()?;
547 Ok(Outcome::Ok(shown_rows(rows, &a.viewer).into_iter().next()))
548 }
549
550 pub(crate) async fn configure_plan(&self, a: ConfigurePlanArgs) -> Result<Outcome<Plan>> {
551 let workspace = a.workspace.trim().to_lowercase();
552 if !access::can_item(Some(&a.actor), &workspace, None, Capability::ManageLabels) {
553 return Ok(Outcome::fail(FailureCode::Forbidden, access::needs_on_item(Capability::ManageLabels, &workspace)));
554 }
555 if !(1..=MAX_NOW_CAP).contains(&a.now_cap) {
556 return Ok(Outcome::fail(FailureCode::Invalid, format!("Now's cap is a number from 1 to {MAX_NOW_CAP}.")));
557 }
558 self.db
559 .prepare(
560 "INSERT INTO plan_settings (workspace, now_cap, updated_at, updated_by) VALUES (?1, ?2, ?3, ?4)
561 ON CONFLICT (workspace) DO UPDATE SET now_cap = excluded.now_cap, updated_at = excluded.updated_at,
562 updated_by = excluded.updated_by",
563 )
564 .bind(&[workspace.as_str().into(), a.now_cap.into(), rfc3339(now_ms()).into(), a.actor.id.as_str().into()])?
565 .run()
566 .await?;
567 Ok(Outcome::Ok(Plan { workspace, now_cap: a.now_cap, ..Plan::default() }))
568 }
569}
570
571/// The rows a viewer may see, as they are shown them.
572fn shown_rows(rows: Vec<ItemRow>, viewer: &Viewer) -> Vec<WorkItem> {
573 rows.into_iter()
574 .filter_map(ItemRow::held)
575 .filter(|held| held.may(viewer.as_ref(), Capability::Read))
576 .map(|held| held.shown(viewer.as_ref()))
577 .collect()
578}
579
580#[cfg(test)]
581mod tests {
582 use rusqlite::{Connection, ToSql};
583
584 use super::super::rows::ITEM_COLUMNS;
585 use super::*;
586 use crate::migrated::database;
587
588 /// acme's open work items 1 to 6 (2 private to rep_secret), 7 closed.
589 fn acme() -> Connection {
590 let db = database();
591 db.execute_batch(
592 "INSERT INTO repo_workspaces (repo_id, workspace, path, private)
593 VALUES ('rep_web', 'acme', 'acme/web', 0), ('rep_secret', 'acme', 'acme/secret', 1);
594 INSERT INTO work_items (id, workspace, number, title, body, labels, author_id, author_name, created_at, updated_at, state)
595 VALUES ('wki_1', 'acme', 1, 'One', '', '[]', 'usr_1', 'ana', '', '', 'open'),
596 ('wki_2', 'acme', 2, 'Two', '', '[]', 'usr_1', 'ana', '', '', 'open'),
597 ('wki_3', 'acme', 3, 'Three', '', '[]', 'usr_1', 'ana', '', '', 'open'),
598 ('wki_4', 'acme', 4, 'Four', '', '[]', 'usr_1', 'ana', '', '', 'open'),
599 ('wki_5', 'acme', 5, 'Five', '', '[]', 'usr_1', 'ana', '', '', 'open'),
600 ('wki_6', 'acme', 6, 'Six', '', '[]', 'usr_1', 'ana', '', '', 'open'),
601 ('wki_7', 'acme', 7, 'Seven', '', '[]', 'usr_1', 'ana', '', '', 'closed');
602 INSERT INTO work_item_repos (item_id, repo_id, role, path, added_at)
603 VALUES ('wki_1', 'rep_web', 'origin', 'acme/web', ''), ('wki_2', 'rep_secret', 'origin', 'acme/secret', ''),
604 ('wki_3', 'rep_web', 'origin', 'acme/web', ''), ('wki_4', 'rep_web', 'origin', 'acme/web', ''),
605 ('wki_5', 'rep_web', 'origin', 'acme/web', ''), ('wki_6', 'rep_web', 'origin', 'acme/web', ''),
606 ('wki_7', 'rep_web', 'origin', 'acme/web', '');",
607 )
608 .unwrap();
609 db
610 }
611
612 fn place(db: &Connection, number: u32, horizon: &str, rank: Option<&str>) {
613 db.execute("UPDATE work_items SET horizon = ?1, rank = ?2 WHERE number = ?3", rusqlite::params![horizon, rank, number]).unwrap();
614 }
615
616 /// The numbers on `horizon`, in the plan's order, as a member or not.
617 fn on(db: &Connection, horizon: &str, member: bool) -> Vec<u32> {
618 let sql = open_items(&format!("{HORIZON} = ?5"), 100).replace("{ITEM_COLUMNS}", ITEM_COLUMNS);
619 let values: [&dyn ToSql; 5] = [&"acme", &u32::from(member), &"[]", &None::<String>, &horizon];
620 let mut statement = db.prepare(&sql).unwrap();
621 statement.query_map(&values[..], |row| row.get::<_, u32>("number")).unwrap().map(Result::unwrap).collect()
622 }
623
624 #[test]
625 fn open_work_items_start_in_triage_newest_first() {
626 let db = acme();
627 assert_eq!(on(&db, "triage", true), [6, 5, 4, 3, 2, 1]);
628 // Someone outside sees only what is filed in a public repository.
629 assert_eq!(on(&db, "triage", false), [6, 5, 4, 3, 1]);
630 }
631
632 #[test]
633 fn a_horizon_reads_urgent_first_then_by_rank() {
634 let db = acme();
635 place(&db, 1, "next", Some("V"));
636 place(&db, 3, "next", Some("F"));
637 place(&db, 4, "next", None);
638 place(&db, 5, "now", Some("k"));
639 place(&db, 6, "now", Some("x"));
640 db.execute("UPDATE work_items SET urgent = 1 WHERE number = 6", []).unwrap();
641 assert_eq!(on(&db, "next", true), [3, 1, 4]);
642 assert_eq!(on(&db, "now", true), [6, 5]);
643 assert_eq!(on(&db, "triage", true), [2]);
644 }
645
646 /// The ranks around a place, as `rank_for` reads them.
647 fn neighbours(db: &Connection, horizon: &str, leaving_out: &str, upper: Option<&str>, lower: Option<&str>) -> (Option<String>, Option<String>) {
648 db.query_row(NEIGHBOURS, rusqlite::params!["acme", horizon, leaving_out, upper, lower], |row| Ok((row.get(0)?, row.get(1)?))).unwrap()
649 }
650
651 #[test]
652 fn a_place_is_found_between_its_neighbours() {
653 let db = acme();
654 place(&db, 1, "next", Some("F"));
655 place(&db, 3, "next", Some("V"));
656 place(&db, 4, "next", Some("k"));
657 // Before #3 (V): after F.
658 assert_eq!(neighbours(&db, "next", "wki_5", Some("V"), None), (Some("F".into()), Some("F".into())));
659 // After #3: ahead of k.
660 assert_eq!(neighbours(&db, "next", "wki_5", None, Some("V")).1, Some("k".into()));
661 // Moving #4 itself to the bottom: it is left out.
662 assert_eq!(neighbours(&db, "next", "wki_4", None, None).0, Some("V".into()));
663 let rank = rank_between(Some("F"), Some("V"));
664 place(&db, 5, "next", Some(&rank));
665 assert_eq!(on(&db, "next", true), [1, 5, 3, 4]);
666 }
667
668 #[test]
669 fn what_is_ready_leaves_out_claimed_blocked_and_others_work() {
670 let db = acme();
671 place(&db, 1, "next", Some("a"));
672 place(&db, 3, "next", Some("b"));
673 place(&db, 4, "now", Some("c"));
674 place(&db, 5, "next", Some("d"));
675 place(&db, 6, "next", Some("e"));
676 // #4 in Now is claimed; #1 is blocked by open #2; #3 is someone else's.
677 db.execute_batch(
678 "INSERT INTO claims (id, item_id, actor_id, actor_name, state, ttl_minutes, created_at, expires_at)
679 VALUES ('clm_1', 'wki_4', 'usr_2', 'bo', 'active', 30, '', '2999-01-01T00:00:00Z');
680 UPDATE work_items SET blocked_by = '[\"wki_2\"]' WHERE number = 1;
681 UPDATE work_items SET assignees = '[\"bo\"]' WHERE number = 3;",
682 )
683 .unwrap();
684 let sql = format!(
685 "SELECT number FROM work_items WHERE {OPEN_VISIBLE} AND {HORIZON} IN ('now', 'next') AND {READY}
686 ORDER BY CASE {HORIZON} WHEN 'now' THEN 0 ELSE 1 END, {PLAN_ORDER}"
687 );
688 let ready = |who: &str| -> Vec<u32> {
689 let values: [&dyn ToSql; 6] = [&"acme", &1, &"[]", &None::<String>, &"2026-10-19T00:00:00Z", &who];
690 let mut statement = db.prepare(&sql).unwrap();
691 statement.query_map(&values[..], |row| row.get::<_, u32>(0)).unwrap().map(Result::unwrap).collect()
692 };
693 assert_eq!(ready("ana"), [5, 6]);
694 assert_eq!(ready("bo"), [3, 5, 6]);
695 // The claim expires: #4, in Now, comes first.
696 db.execute("UPDATE claims SET expires_at = '2026-10-18T00:00:00Z'", []).unwrap();
697 assert_eq!(ready("ana"), [4, 5, 6]);
698 }
699
700 #[test]
701 fn counts_and_later_groups_cover_every_horizon() {
702 let db = acme();
703 place(&db, 1, "later", Some("a"));
704 place(&db, 3, "later", Some("b"));
705 place(&db, 4, "set_aside", Some("c"));
706 let values: [&dyn ToSql; 4] = [&"acme", &1, &"[]", &None::<String>];
707 let mut counts: Vec<(String, u32)> = db
708 .prepare(&COUNTS.replace("{OPEN_VISIBLE}", OPEN_VISIBLE))
709 .unwrap()
710 .query_map(&values[..], |row| Ok((row.get(0)?, row.get(1)?)))
711 .unwrap()
712 .map(Result::unwrap)
713 .collect();
714 counts.sort();
715 assert_eq!(counts, [("later".into(), 2), ("set_aside".into(), 1), ("triage".into(), 3)]);
716 let groups: Vec<(Option<String>, u32)> = db
717 .prepare(&LATER_GROUPS.replace("{OPEN_VISIBLE}", OPEN_VISIBLE))
718 .unwrap()
719 .query_map(&values[..], |row| Ok((row.get(0)?, row.get(1)?)))
720 .unwrap()
721 .map(Result::unwrap)
722 .collect();
723 assert_eq!(groups, [(Some("acme/web".into()), 2)]);
724 }
725
726 /// The facts `FACTS` reads for work item `number`.
727 fn facts(db: &Connection, number: u32) -> FactsRow {
728 let ids = format!("[\"wki_{number}\"]");
729 db.query_row(FACTS, [ids], |row| {
730 Ok(FactsRow {
731 number: row.get("number")?,
732 decisions: row.get("decisions")?,
733 draft_pulls: row.get("draft_pulls")?,
734 commits: row.get("commits")?,
735 open_pulls: row.get("open_pulls")?,
736 merged_pulls: row.get("merged_pulls")?,
737 landed_targets: row.get("landed_targets")?,
738 active_at: row.get("active_at")?,
739 standing: row.get("standing")?,
740 standing_at: row.get("standing_at")?,
741 })
742 })
743 .unwrap()
744 }
745
746 #[test]
747 fn the_record_says_how_far_along_each_is() {
748 let db = acme();
749 db.execute_batch(
750 "UPDATE work_items SET horizon = 'now', updated_at = '2026-10-01T00:00:00Z' WHERE number IN (1, 3, 4);
751 INSERT INTO decisions (id, repo_id, item_id, kind, confidence, summary, created_at)
752 VALUES ('dec_1', 'rep_web', 'wki_3', 'proposal', 'approved', 'Rank by fractional keys', '2026-10-10T00:00:00Z'),
753 ('dec_2', 'rep_web', 'wki_1', 'claim', 'approved', 'Scope: tests', '2026-10-11T00:00:00Z');
754 INSERT INTO sessions (id, repo_id, actor_id, actor_name, harness, state, started_at, last_seen_at, item_id)
755 VALUES ('ses_1', 'rep_web', 'usr_1', 'ana', 'claude_code', 'active', '2026-10-12T00:00:00Z', '2026-10-15T00:00:00Z', 'wki_4');
756 INSERT INTO session_touches (session_id, repo_id, kind, value, at) VALUES ('ses_1', 'rep_web', 'commit', 'abc1234', '2026-10-15T00:00:00Z');",
757 )
758 .unwrap();
759 // A claim's note is not a decision; nothing else has happened on #1.
760 let one = facts(&db, 1);
761 assert_eq!((one.hill(), one.active_at.as_deref()), (Hill::FiguringOut, Some("2026-10-11T00:00:00Z")));
762 assert_eq!(facts(&db, 3).hill(), Hill::Decided);
763 let four = facts(&db, 4);
764 assert_eq!((four.hill(), four.active_at.as_deref()), (Hill::Building, Some("2026-10-15T00:00:00Z")));
765 // Quiet: nothing since before the line.
766 assert!(facts(&db, 3).progress("2026-10-14T00:00:00Z").quiet);
767 assert!(!facts(&db, 4).progress("2026-10-14T00:00:00Z").quiet);
768 }
769
770 #[test]
771 fn the_plan_says_what_waits_too_long() {
772 let db = acme();
773 db.execute_batch(
774 "UPDATE work_items SET created_at = '2026-10-01T00:00:00Z' WHERE number IN (1, 3);
775 UPDATE work_items SET created_at = '2026-10-18T00:00:00Z' WHERE number = 4;
776 UPDATE work_items SET horizon = 'set_aside', planned_at = '2026-10-15T00:00:00Z' WHERE number = 5;
777 UPDATE work_items SET horizon = 'set_aside', planned_at = '2026-08-01T00:00:00Z' WHERE number = 6;",
778 )
779 .unwrap();
780 let values: [&dyn ToSql; 6] = [&"acme", &1, &"[]", &None::<String>, &"2026-10-12T00:00:00Z", &"2026-09-20T00:00:00Z"];
781 let (waiting, recent): (u32, u32) = db
782 .query_row(&WAITING.replace("{OPEN_VISIBLE}", OPEN_VISIBLE), &values[..], |row| Ok((row.get(0)?, row.get(1)?)))
783 .unwrap();
784 // #1 and #3 (and #2, made with no date) in Triage since before the line; #5 set aside lately.
785 assert_eq!((waiting, recent), (3, 1));
786 }
787
788 #[test]
789 fn a_move_says_what_changed() {
790 assert_eq!(move_note(Horizon::Triage, Horizon::Next, false, false).as_deref(), Some("moved this from Triage to Next"));
791 assert_eq!(move_note(Horizon::Next, Horizon::Now, false, true).as_deref(), Some("moved this from Next to Now and marked this urgent"));
792 assert_eq!(move_note(Horizon::Next, Horizon::Next, false, false), None);
793 assert_eq!(valid_budget(Some(12.345)), Ok(Some(12.35)));
794 assert!(valid_budget(Some(0.0)).is_err());
795 assert!(valid_budget(Some(f64::NAN)).is_err());
796 }
797}