Menu
akurai-tasks
publicLatest change 2c9c32aa6fc1bd2551a09c6a1fc1e71b2dd98c79 - feat: complete and deploy AkurAI Tasks 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 start = match &query.cursor {
Some(cursor) => decode_cursor(cursor)?,
None => b"item/".to_vec(),
};
let mut db = self
.db
.lock()
.map_err(|_| Error::Storage("lock poisoned".into()))?;
let scan_limit = query.limit.saturating_mul(20).clamp(query.limit, 10_000);
let page = db
.range_bounded(&start, &upper_bound(b"item/"), scan_limit, query.byte_limit)
.map_err(storage)?;
let mut items = Vec::new();
let mut next = None;
for (key, bytes) in &page.items {
let item = decode(bytes)?;
if matches_query(&item, query) {
if items.len() == query.limit {
next = Some(encode_cursor(key));
break;
}
items.push(item);
}
}
if next.is_none() {
next = page.continuation.as_ref().map(|key| encode_cursor(key));
}
items.sort_by(|left, right| compare_items(left, right, &query.sort));
if query.descending {
items.reverse();
}
Ok(object(vec![
("items", Value::Array(items)),
(
"nextCursor",
next.as_deref().map(string).unwrap_or(Value::Null),
),
("scanned", Value::Int(page.items.len() 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 {
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("")),
_ => 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 encode_cursor(bytes: &[u8]) -> String {
let mut out = String::with_capacity(bytes.len() * 2);
for byte in bytes {
out.push_str(&format!("{byte:02x}"));
}
out
}
fn decode_cursor(value: &str) -> Result<Vec<u8>> {
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()))?,
);
}
if !out.starts_with(b"item/") {
return Err(Error::Invalid("query cursor scope mismatch".into()));
}
Ok(out)
}
#[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);
}
}