Menu
akurai-tasks
publicLatest change 350a37d39889e09d093509a69e7f06106958ffa2 - Harden task coordination and recovery by Ólafur Búi Ólafsson
use super::*;
use std::cmp::Ordering;
#[derive(Clone, Debug)]
pub struct Query {
pub projects: Vec<String>,
pub repos: Vec<String>,
pub states: Vec<String>,
pub priorities: Vec<String>,
pub owners: Vec<String>,
pub types: Vec<String>,
pub labels: Vec<String>,
pub parent: Option<String>,
pub cycle: Option<String>,
pub text: Option<String>,
pub blocked: Option<bool>,
pub created_after: Option<i64>,
pub created_before: Option<i64>,
pub updated_after: Option<i64>,
pub updated_before: Option<i64>,
pub sort: String,
pub descending: bool,
pub limit: usize,
pub byte_limit: usize,
pub cursor: Option<String>,
}
impl Default for Query {
fn default() -> Self {
Self {
projects: Vec::new(),
repos: Vec::new(),
states: Vec::new(),
priorities: Vec::new(),
owners: Vec::new(),
types: Vec::new(),
labels: Vec::new(),
parent: None,
cycle: None,
text: None,
blocked: None,
created_after: None,
created_before: None,
updated_after: None,
updated_before: None,
sort: "id".into(),
descending: false,
limit: 100,
byte_limit: 1_048_576,
cursor: None,
}
}
}
impl Store {
pub fn query_items(&self, query: &Query) -> Result<Value> {
if query.limit == 0 || query.limit > 500 {
return Err(Error::Invalid("query limit must be 1..=500".into()));
}
if query.byte_limit < 1024 || query.byte_limit > 16 * 1024 * 1024 {
return Err(Error::Invalid(
"query byte limit must be 1024..=16777216".into(),
));
}
if !matches!(
query.sort.as_str(),
"id" | "created" | "updated" | "priority" | "state" | "title"
) {
return Err(Error::Invalid(format!("unsupported sort {}", query.sort)));
}
for project in &query.projects {
validate_key(project)?;
}
for state in &query.states {
state_index(state)?;
}
let fingerprint = query_fingerprint(query);
let cursor_id = match &query.cursor {
Some(cursor) => {
let (scope, id) = decode_cursor(cursor)?;
if scope != fingerprint {
return Err(Error::Invalid("query cursor scope mismatch".into()));
}
Some(id)
}
None => None,
};
let mut db = self
.db
.lock()
.map_err(|_| Error::Storage("lock poisoned".into()))?;
let entries = db
.range(b"item/", &upper_bound(b"item/"))
.map_err(storage)?;
let scanned = entries.len();
let mut items = Vec::new();
for (_, bytes) in entries {
let item = decode(&bytes)?;
if matches_query(&item, query) {
items.push(item);
}
}
items.sort_by(|left, right| compare_items(left, right, &query.sort));
if query.descending {
items.reverse();
}
let start = match cursor_id {
Some(id) => {
items
.iter()
.position(|item| field(item, "id") == Some(&id))
.ok_or_else(|| Error::Invalid("query cursor is no longer valid".into()))?
+ 1
}
None => 0,
};
let mut page = Vec::with_capacity(query.limit.min(items.len().saturating_sub(start)));
let mut bytes = 0usize;
for item in items.iter().skip(start).take(query.limit) {
let item_bytes = item.to_json().len();
if !page.is_empty() && bytes.saturating_add(item_bytes) > query.byte_limit {
break;
}
bytes = bytes.saturating_add(item_bytes);
page.push(item.clone());
}
let consumed = start + page.len();
let next = if consumed < items.len() {
page.last()
.and_then(|item| field(item, "id").map(|id| encode_cursor(&fingerprint, id)))
} else {
None
};
Ok(object(vec![
("items", Value::Array(page)),
(
"nextCursor",
next.as_deref().map(string).unwrap_or(Value::Null),
),
("scanned", Value::Int(scanned as i64)),
("limit", Value::Int(query.limit as i64)),
("sort", string(&query.sort)),
("descending", Value::Bool(query.descending)),
]))
}
}
fn matches_query(item: &Value, query: &Query) -> bool {
if !query.projects.is_empty() && !contains(&query.projects, field(item, "project")) {
return false;
}
if !query.repos.is_empty() && !contains(&query.repos, field(item, "repo")) {
return false;
}
if !query.states.is_empty() && !contains(&query.states, field(item, "state")) {
return false;
}
if !query.priorities.is_empty() && !contains(&query.priorities, field(item, "priority")) {
return false;
}
if !query.owners.is_empty() && !contains(&query.owners, field(item, "owner")) {
return false;
}
if !query.types.is_empty() && !contains(&query.types, field(item, "type")) {
return false;
}
if let Some(parent) = &query.parent {
if field(item, "parent") != Some(parent) {
return false;
}
}
if let Some(cycle) = &query.cycle {
if field(item, "cycle") != Some(cycle) {
return false;
}
}
if let Some(blocked) = query.blocked {
if item.get("blocked").and_then(Value::as_bool) != Some(blocked) {
return false;
}
}
if !query.labels.is_empty() {
let labels = match item.get("labels") {
Some(Value::Array(values)) => values,
_ => return false,
};
if !query
.labels
.iter()
.all(|wanted| labels.iter().any(|label| label.as_str() == Some(wanted)))
{
return false;
}
}
let created = item.get("createdAt").and_then(Value::as_i64).unwrap_or(0);
let updated = item
.get("updatedAt")
.and_then(Value::as_i64)
.unwrap_or(created);
if query.created_after.is_some_and(|value| created < value)
|| query.created_before.is_some_and(|value| created > value)
|| query.updated_after.is_some_and(|value| updated < value)
|| query.updated_before.is_some_and(|value| updated > value)
{
return false;
}
if let Some(text) = &query.text {
let needle = text.to_ascii_lowercase();
let haystack = format!(
"{}\n{}\n{}",
field(item, "id").unwrap_or(""),
field(item, "title").unwrap_or(""),
field(item, "description").unwrap_or("")
)
.to_ascii_lowercase();
if !haystack.contains(&needle) {
return false;
}
}
true
}
fn contains(values: &[String], actual: Option<&str>) -> bool {
actual.is_some_and(|actual| values.iter().any(|value| value == actual))
}
fn compare_items(left: &Value, right: &Value, sort: &str) -> Ordering {
let primary = match sort {
"created" => number(left, "createdAt").cmp(&number(right, "createdAt")),
"updated" => number(left, "updatedAt").cmp(&number(right, "updatedAt")),
"priority" => priority(field(left, "priority")).cmp(&priority(field(right, "priority"))),
"state" => state_index(field(left, "state").unwrap_or("Inbox"))
.unwrap_or(0)
.cmp(&state_index(field(right, "state").unwrap_or("Inbox")).unwrap_or(0)),
"title" => field(left, "title")
.unwrap_or("")
.cmp(field(right, "title").unwrap_or("")),
_ => Ordering::Equal,
};
primary.then_with(|| {
field(left, "id")
.unwrap_or("")
.cmp(field(right, "id").unwrap_or(""))
})
}
fn number(value: &Value, key: &str) -> i64 {
value.get(key).and_then(Value::as_i64).unwrap_or(0)
}
fn priority(value: Option<&str>) -> usize {
match value {
Some("critical") => 0,
Some("high") => 1,
Some("normal") => 2,
Some("low") => 3,
_ => 4,
}
}
fn query_fingerprint(query: &Query) -> String {
let mut scope = query.clone();
scope.cursor = None;
scope.limit = 0;
scope.byte_limit = 0;
auth::digest(&format!("{scope:?}"))[..32].to_string()
}
fn encode_cursor(scope: &str, id: &str) -> String {
let value = format!("query/{scope}/{id}");
let mut out = String::with_capacity(value.len() * 2);
for byte in value.as_bytes() {
out.push_str(&format!("{byte:02x}"));
}
out
}
fn decode_cursor(value: &str) -> Result<(String, String)> {
if !value.len().is_multiple_of(2) || value.len() > 1024 {
return Err(Error::Invalid("invalid query cursor".into()));
}
let mut out = Vec::with_capacity(value.len() / 2);
for pair in value.as_bytes().chunks(2) {
let text =
std::str::from_utf8(pair).map_err(|_| Error::Invalid("invalid query cursor".into()))?;
out.push(
u8::from_str_radix(text, 16)
.map_err(|_| Error::Invalid("invalid query cursor".into()))?,
);
}
let decoded =
std::str::from_utf8(&out).map_err(|_| Error::Invalid("invalid query cursor".into()))?;
let mut parts = decoded.splitn(3, '/');
if parts.next() != Some("query") {
return Err(Error::Invalid("query cursor scope mismatch".into()));
}
let scope = parts
.next()
.filter(|part| !part.is_empty())
.ok_or_else(|| Error::Invalid("invalid query cursor".into()))?;
let id = parts
.next()
.filter(|part| !part.is_empty())
.ok_or_else(|| Error::Invalid("invalid query cursor".into()))?;
Ok((scope.to_string(), id.to_string()))
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn structured_query_spans_projects_repositories_and_filters() {
let path = std::env::temp_dir().join(format!(
"akurai-tasks-query-{}-{}.db",
std::process::id(),
now()
));
let _ = std::fs::remove_file(&path);
let store = Store::open(&path).unwrap();
for (project, repo) in [("ONE", "owner/api"), ("TWO", "owner/web")] {
store
.create_project(
project,
project,
vec![repo.into()],
"admin",
&format!("p-{project}"),
)
.unwrap();
store
.create_item(CreateItem {
project,
title: &format!("{project} searchable"),
description: "needle",
repo: Some(repo),
priority: if project == "ONE" { "high" } else { "low" },
actor: "admin",
idempotency_key: &format!("i-{project}"),
})
.unwrap();
}
let query = Query {
projects: vec!["ONE".into(), "TWO".into()],
repos: vec!["owner/web".into()],
text: Some("needle".into()),
priorities: vec!["low".into()],
..Query::default()
};
let result = store.query_items(&query).unwrap();
let items = match result.get("items") {
Some(Value::Array(items)) => items,
_ => panic!("missing items"),
};
assert_eq!(items.len(), 1);
assert_eq!(field(&items[0], "project"), Some("TWO"));
drop(store);
let _ = std::fs::remove_file(path);
}
#[test]
fn pagination_follows_global_sort_order_and_binds_cursor_scope() {
let path = std::env::temp_dir().join(format!(
"akurai-tasks-query-page-{}-{}.db",
std::process::id(),
now()
));
let _ = std::fs::remove_file(&path);
let store = Store::open(&path).unwrap();
store
.create_project(
"SORT",
"Sort",
vec!["owner/repo".into()],
"admin",
"project",
)
.unwrap();
for (index, title) in ["Zulu", "Alpha"].iter().enumerate() {
store
.create_item(CreateItem {
project: "SORT",
title,
description: "",
repo: Some("owner/repo"),
priority: "normal",
actor: "admin",
idempotency_key: &format!("item-{index}"),
})
.unwrap();
}
let first_query = Query {
projects: vec!["SORT".into()],
sort: "title".into(),
limit: 1,
..Query::default()
};
let first = store.query_items(&first_query).unwrap();
let Some(Value::Array(first_items)) = first.get("items") else {
panic!("missing first page");
};
assert_eq!(field(&first_items[0], "title"), Some("Alpha"));
let cursor = field(&first, "nextCursor").unwrap().to_string();
let second = store
.query_items(&Query {
cursor: Some(cursor.clone()),
..first_query.clone()
})
.unwrap();
let Some(Value::Array(second_items)) = second.get("items") else {
panic!("missing second page");
};
assert_eq!(field(&second_items[0], "title"), Some("Zulu"));
assert!(store
.query_items(&Query {
cursor: Some(cursor),
sort: "priority".into(),
..first_query
})
.is_err());
drop(store);
let _ = std::fs::remove_file(path);
}
}