Menu
akurai-tasks
publicLatest change a3b219eee35afd2833ddd5655e5fb4889674a57f - feat: add strategy outcomes and delivery metrics by Ólafur Búi Ólafsson
use super::*;
use std::collections::{BTreeMap, BTreeSet, HashMap};
#[derive(Debug, Clone, Default)]
pub struct MetricsQuery {
pub projects: Vec<String>,
pub repos: Vec<String>,
pub owners: Vec<String>,
pub cycles: Vec<String>,
pub period: String,
pub as_of: Option<i64>,
}
impl Store {
pub fn metrics(&self, query: &MetricsQuery) -> Result<Value> {
let period_seconds = match query.period.as_str() {
"24h" => 24 * 60 * 60,
"7d" => 7 * 24 * 60 * 60,
"30d" => 30 * 24 * 60 * 60,
_ => {
return Err(Error::Invalid(
"metrics period must be 24h, 7d, or 30d".into(),
))
}
};
let as_of = query.as_of.unwrap_or_else(now);
let window_start = as_of.saturating_sub(period_seconds);
for project in &query.projects {
validate_key(project)?;
}
let mut page_query = Query {
projects: query.projects.clone(),
repos: query.repos.clone(),
owners: query.owners.clone(),
limit: 500,
byte_limit: 16 * 1024 * 1024,
..Query::default()
};
let mut items = Vec::new();
loop {
let page = self.query_items(&page_query)?;
if let Some(Value::Array(page_items)) = page.get("items") {
items.extend(page_items.iter().cloned());
}
let next = page
.get("nextCursor")
.and_then(Value::as_str)
.map(str::to_string);
let Some(next) = next else { break };
page_query.cursor = Some(next);
}
let scanned_items = items.len();
if !query.cycles.is_empty() {
items.retain(|item| {
field(item, "cycle")
.is_some_and(|cycle| query.cycles.iter().any(|wanted| wanted == cycle))
});
}
let events = match self.events(None)? {
Value::Array(events) => {
if query.projects.is_empty() {
events
} else {
events
.into_iter()
.filter(|event| {
field(event, "project").is_some_and(|project| {
query.projects.iter().any(|wanted| wanted == project)
})
})
.collect()
}
}
_ => Vec::new(),
};
let histories = build_histories(&events);
let mut dependencies = HashMap::new();
let mut db = self
.db
.lock()
.map_err(|_| Error::Storage("lock poisoned".into()))?;
for item in &items {
if let Some(id) = field(item, "id") {
dependencies.insert(
id.to_string(),
dependency_depth(&mut db, id, &mut BTreeSet::new())?,
);
}
}
drop(db);
let mut groups: BTreeMap<(String, String, String, String), Vec<Value>> = BTreeMap::new();
for item in items {
let key = (
field(&item, "project").unwrap_or("").to_string(),
field(&item, "repo").unwrap_or("").to_string(),
field(&item, "owner").unwrap_or("").to_string(),
field(&item, "cycle").unwrap_or("").to_string(),
);
groups.entry(key).or_default().push(item);
}
let definitions = object(vec![
("wip", string("current items not in Done or canceled; snapshot at asOf")),
("stateAge", string("asOf minus the latest explicit state-entry event; createdAt when no event exists")),
("leadTime", string("createdAt to the first Done transition inside the selected period")),
("cycleTime", string("In Progress transition to the first Done transition inside the selected period")),
("blockedAge", string("asOf minus blockedAt for currently blocked items; unknown when blockedAt is unavailable")),
("dependencyDepth", string("longest dependency chain for each source item; zero means no dependency")),
("handoffs", string("handoff commands created inside the selected period")),
("reopened", string("explicit backward state transitions inside the selected period")),
("terminal", string("Done is completed; canceled/canceledAt is excluded from WIP and retained only as terminal source data")),
]);
let mut group_values = Vec::new();
for ((project, repo, owner, cycle), source_items) in groups {
let ids: Vec<String> = source_items
.iter()
.filter_map(|i| field(i, "id").map(str::to_string))
.collect();
let wip: Vec<String> = source_items
.iter()
.filter(|i| !is_terminal(i))
.filter_map(|i| field(i, "id").map(str::to_string))
.collect();
let state_ages: Vec<(String, i64)> = source_items
.iter()
.filter_map(|item| {
let id = field(item, "id")?;
let entered = histories
.get(id)
.and_then(|h| h.last_state_at(field(item, "state").unwrap_or("")))
.unwrap_or_else(|| {
item.get("createdAt")
.and_then(Value::as_i64)
.unwrap_or(as_of)
});
Some((id.to_string(), as_of.saturating_sub(entered).max(0)))
})
.collect();
let blocked: Vec<(String, i64)> = source_items
.iter()
.filter(|i| i.get("blocked").and_then(Value::as_bool) == Some(true))
.filter_map(|item| {
let id = field(item, "id")?;
let at = item
.get("blockedAt")
.and_then(Value::as_i64)
.or_else(|| histories.get(id).and_then(|h| h.blocked_at));
Some((
id.to_string(),
at.map(|v| as_of.saturating_sub(v).max(0)).unwrap_or(-1),
))
})
.collect();
let lead = flow_durations(&source_items, &histories, window_start, as_of, Flow::Lead);
let cycle_time =
flow_durations(&source_items, &histories, window_start, as_of, Flow::Cycle);
let depths: Vec<(String, i64)> = source_items
.iter()
.filter_map(|item| {
field(item, "id")
.and_then(|id| dependencies.get(id).copied().map(|d| (id.to_string(), d)))
})
.collect();
let handoffs = event_ids(&events, window_start, as_of, "handoff", &ids);
let reopened = reopened_ids(&source_items, &histories, window_start, as_of);
let mut source = BTreeSet::new();
source.extend(ids.iter().cloned());
source.extend(handoffs.iter().cloned());
source.extend(reopened.iter().cloned());
let sample_size = source.len() as i64;
group_values.push(object(vec![
(
"project",
optional_string((!project.is_empty()).then_some(project.as_str())),
),
(
"repo",
optional_string((!repo.is_empty()).then_some(repo.as_str())),
),
(
"owner",
optional_string((!owner.is_empty()).then_some(owner.as_str())),
),
(
"cycle",
optional_string((!cycle.is_empty()).then_some(cycle.as_str())),
),
("sampleSize", Value::Int(sample_size)),
("smallSample", Value::Bool(sample_size < 5)),
(
"sourceItemIds",
Value::Array(source.into_iter().map(|id| string(&id)).collect()),
),
("wip", count_metric(wip.len() as i64, wip)),
("stateAge", duration_metric(&state_ages)),
("leadTime", duration_metric(&lead)),
("cycleTime", duration_metric(&cycle_time)),
(
"blockedAge",
duration_metric(
&blocked
.into_iter()
.filter(|(_, age)| *age >= 0)
.collect::<Vec<_>>(),
),
),
("dependencyDepth", depth_metric(&depths)),
("handoffs", count_metric(handoffs.len() as i64, handoffs)),
("reopened", count_metric(reopened.len() as i64, reopened)),
]));
}
let fresh_events = events
.iter()
.filter_map(|event| event.get("createdAt").and_then(Value::as_i64))
.max();
Ok(object(vec![
("period", string(&query.period)),
("asOf", Value::Int(as_of)),
("windowStart", Value::Int(window_start)),
("scannedItems", Value::Int(scanned_items as i64)),
("truncated", Value::Bool(false)),
(
"freshness",
object(vec![
("asOf", Value::Int(as_of)),
(
"latestEventAt",
fresh_events.map(Value::Int).unwrap_or(Value::Null),
),
(
"lagSeconds",
fresh_events
.map(|v| Value::Int(as_of.saturating_sub(v)))
.unwrap_or(Value::Null),
),
]),
),
("definitions", definitions),
("groups", Value::Array(group_values)),
]))
}
}
#[derive(Clone, Default)]
struct History {
states: Vec<(String, i64)>,
done_at: Option<i64>,
in_progress_at: Option<i64>,
blocked_at: Option<i64>,
transitions: Vec<(String, String, i64)>,
}
impl History {
fn last_state_at(&self, state: &str) -> Option<i64> {
self.states
.iter()
.rev()
.find(|(candidate, _)| candidate == state)
.map(|(_, at)| *at)
}
}
fn build_histories(events: &[Value]) -> HashMap<String, History> {
let mut out: HashMap<String, History> = HashMap::new();
for event in events {
let at = event.get("createdAt").and_then(Value::as_i64).unwrap_or(0);
let request = event.get("request");
let command = request.and_then(|r| field(r, "command")).unwrap_or("");
let id = request.and_then(|r| field(r, "id").or_else(|| field(r, "item")));
let Some(id) = id else { continue };
let history = out.entry(id.to_string()).or_default();
if command == "transition" {
let target = request
.and_then(|r| field(r, "target"))
.unwrap_or("")
.to_string();
let previous = history
.states
.last()
.map(|(state, _)| state.as_str())
.unwrap_or("Inbox")
.to_string();
history.transitions.push((previous, target.clone(), at));
history.states.push((target.clone(), at));
if target == "Done" && history.done_at.is_none() {
history.done_at = Some(at);
}
if target == "In Progress" && history.in_progress_at.is_none() {
history.in_progress_at = Some(at);
}
} else if command == "set_blocked"
&& request
.and_then(|r| r.get("reason"))
.and_then(Value::as_str)
.is_some()
{
history.blocked_at = Some(at);
}
}
out
}
#[derive(Clone, Copy)]
enum Flow {
Lead,
Cycle,
}
fn flow_durations(
items: &[Value],
histories: &HashMap<String, History>,
start: i64,
end: i64,
flow: Flow,
) -> Vec<(String, i64)> {
items
.iter()
.filter_map(|item| {
let id = field(item, "id")?;
let history = histories.get(id)?;
let done = history.done_at?;
if done < start || done > end {
return None;
}
let begin = match flow {
Flow::Lead => item.get("createdAt").and_then(Value::as_i64)?,
Flow::Cycle => history.in_progress_at?,
};
Some((id.to_string(), done.saturating_sub(begin).max(0)))
})
.collect()
}
fn event_ids(
events: &[Value],
start: i64,
end: i64,
command: &str,
allowed: &[String],
) -> Vec<String> {
events
.iter()
.filter_map(|event| {
let at = event.get("createdAt").and_then(Value::as_i64)?;
if at < start || at > end || field(event.get("request")?, "command") != Some(command) {
return None;
}
let id = event
.get("request")
.and_then(|r| field(r, "id").or_else(|| field(r, "item")))?;
allowed
.iter()
.any(|allowed_id| allowed_id == id)
.then_some(id.to_string())
})
.collect()
}
fn reopened_ids(
items: &[Value],
histories: &HashMap<String, History>,
start: i64,
end: i64,
) -> Vec<String> {
items
.iter()
.filter_map(|item| {
let id = field(item, "id")?;
let reopened = histories.get(id)?.transitions.iter().any(|(from, to, at)| {
*at >= start && *at <= end && state_number(to) < state_number(from)
});
reopened.then_some(id.to_string())
})
.collect()
}
fn state_number(state: &str) -> usize {
STATES
.iter()
.position(|candidate| *candidate == state)
.unwrap_or(usize::MAX)
}
fn is_terminal(item: &Value) -> bool {
field(item, "state") == Some("Done")
|| field(item, "state") == Some("Canceled")
|| item.get("canceledAt").is_some_and(|v| v != &Value::Null)
}
fn count_metric(value: i64, ids: Vec<String>) -> Value {
object(vec![
("value", Value::Int(value)),
("known", Value::Bool(true)),
(
"itemIds",
Value::Array(ids.into_iter().map(|id| string(&id)).collect()),
),
])
}
fn duration_metric(values: &[(String, i64)]) -> Value {
if values.is_empty() {
return object(vec![
("value", Value::Null),
("known", Value::Bool(false)),
("sampleSize", Value::Int(0)),
("itemIds", Value::Array(Vec::new())),
]);
}
let total: i64 = values.iter().map(|(_, value)| *value).sum();
object(vec![
("value", Value::Int(total / values.len() as i64)),
("known", Value::Bool(true)),
("sampleSize", Value::Int(values.len() as i64)),
(
"itemIds",
Value::Array(values.iter().map(|(id, _)| string(id)).collect()),
),
])
}
fn depth_metric(values: &[(String, i64)]) -> Value {
let max = values.iter().map(|(_, depth)| *depth).max();
object(vec![
("value", max.map(Value::Int).unwrap_or(Value::Null)),
("known", Value::Bool(max.is_some())),
("sampleSize", Value::Int(values.len() as i64)),
(
"itemIds",
Value::Array(values.iter().map(|(id, _)| string(id)).collect()),
),
])
}
fn dependency_depth(
db: &mut akurai_storage::BTree,
item: &str,
seen: &mut BTreeSet<String>,
) -> Result<i64> {
if !seen.insert(item.to_string()) {
return Ok(0);
}
let prefix = format!("dependency/{item}/");
let mut max = 0;
for (_, bytes) in db
.range(prefix.as_bytes(), &upper_bound(prefix.as_bytes()))
.map_err(storage)?
{
if let Some(depends) = field(&decode(&bytes)?, "dependsOn") {
max = max.max(1 + dependency_depth(db, depends, seen)?);
}
}
seen.remove(item);
Ok(max)
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn metrics_scan_beyond_default_query_page_without_truncation() {
let path = std::env::temp_dir().join(format!("tasks-metrics-{}.db", std::process::id()));
let _ = std::fs::remove_file(&path);
let store = Store::open(&path).unwrap();
store
.create_project(
"OPS",
"Operations",
vec!["owner/ops".into()],
"admin",
"project",
)
.unwrap();
for index in 0..101 {
store
.create_item(CreateItem {
project: "OPS",
title: "Metric item",
description: "",
repo: Some("owner/ops"),
priority: "normal",
actor: "admin",
idempotency_key: &format!("item-{index}"),
})
.unwrap();
}
let result = store
.metrics(&MetricsQuery {
projects: vec!["OPS".into()],
period: "24h".into(),
as_of: Some(now() + 1),
..MetricsQuery::default()
})
.unwrap();
assert_eq!(
result.get("scannedItems").and_then(Value::as_i64),
Some(101)
);
assert_eq!(
result.get("truncated").and_then(Value::as_bool),
Some(false)
);
let groups = match result.get("groups") {
Some(Value::Array(groups)) => groups,
_ => panic!("metrics groups missing"),
};
let group = groups
.iter()
.find(|group| field(group, "project") == Some("OPS"))
.expect("OPS metrics group missing");
assert_eq!(
group
.get("wip")
.and_then(|wip| wip.get("value"))
.and_then(Value::as_i64),
Some(101)
);
let _ = std::fs::remove_file(path);
}
}