verbs/
query.rs

1// SPDX-License-Identifier: Apache-2.0
2//! Structured query over the operation log.
3
4use std::{collections::BTreeMap, path::Path};
5
6use chrono::TimeZone;
7use objects::{
8    RecoveryDetails,
9    error::{HeddleError, Result},
10    object::{OperationId, StateId},
11    store::ObjectStore,
12};
13use oplog::{OpEntry, OpLog, OpLogBackend, OpRecord, RecordedHead};
14use refs::refs::{IndexedOperation, OperationLogIndex, OperationLogQuery};
15use schemars::JsonSchema;
16use serde::Serialize;
17
18use crate::{
19    ExecutionContext, HeddleReport, MachineOutputKind, OutputDiscriminator, ReportContract,
20    schema_for_report,
21};
22
23/// Query filters for the operation log facade.
24#[derive(Debug, Clone, Default, PartialEq, Eq)]
25pub struct QueryRequest {
26    pub actor: String,
27    pub symbol: String,
28    pub signal_kind: String,
29    pub thread: String,
30    pub verbs: Vec<String>,
31    pub since_secs: i64,
32    pub until_secs: i64,
33    pub limit: u32,
34    pub include_checkpoints: bool,
35}
36
37#[derive(Debug, Clone, Serialize, JsonSchema, PartialEq, Eq)]
38pub struct QueryReport {
39    pub output_kind: &'static str,
40    pub hits: Vec<QueryHit>,
41}
42
43impl QueryReport {
44    pub const CONTRACT: ReportContract = ReportContract {
45        schema_name: "query",
46        machine_output_kind: MachineOutputKind::Json,
47        output_discriminator: Some(OutputDiscriminator {
48            field: "output_kind",
49            value: "query",
50        }),
51        schema: schema_for_report::<QueryReport>,
52    };
53}
54
55impl HeddleReport for QueryReport {
56    const CONTRACT: ReportContract = QueryReport::CONTRACT;
57}
58
59#[derive(Debug, Clone, Serialize, JsonSchema, PartialEq, Eq)]
60pub struct QueryHit {
61    pub seq: u64,
62    pub timestamp_secs: i64,
63    pub verb: String,
64    pub actor_email: String,
65    pub operation_id: Option<String>,
66    pub thread: Option<String>,
67    pub symbols: Vec<String>,
68    pub signal_kinds: Vec<String>,
69    pub state_id: Option<String>,
70}
71
72/// Query is an operator-facing inspection command, so it should answer from
73/// the live oplog even before the rebuildable index sidecar has been warmed.
74/// Keep the scan bounded; long-tail history can use the index once populated.
75const OPLOG_FALLBACK_SCAN_WINDOW: usize = 100_000;
76
77pub fn query(ctx: &ExecutionContext, req: QueryRequest) -> Result<QueryReport> {
78    let repo = ctx.require_repo()?;
79    let mut q = build_query(&req)?;
80    let actor = q.actor.take();
81    let limit = q.limit.take();
82    let hits = query_combined(repo.heddle_dir(), &q)?;
83    q.actor = actor;
84    let mut selected = Vec::new();
85    for mut hit in hits {
86        fill_actor_email_from_state(repo, &mut hit)?;
87        if !hit.matches(&q) {
88            continue;
89        }
90        selected.push(hit);
91        if let Some(limit) = limit
92            && selected.len() >= limit
93        {
94            break;
95        }
96    }
97    Ok(QueryReport {
98        output_kind: "query",
99        hits: selected.into_iter().map(hit_to_report).collect(),
100    })
101}
102
103/// Map a user-facing query `--verb` onto the stored catalog name.
104/// `capture` is the CLI name for the durable `snapshot` verb. Unknown
105/// names return `None` so the filter can fail closed.
106fn canonicalize_query_verb(input: &str) -> Option<&'static str> {
107    let needle = input.trim();
108    if needle.is_empty() {
109        return None;
110    }
111    if needle.eq_ignore_ascii_case("capture") {
112        return Some("snapshot");
113    }
114    OpRecord::verbs(true)
115        .into_iter()
116        .find(|verb| verb.eq_ignore_ascii_case(needle))
117}
118
119fn resolve_query_verbs(verbs: &[String]) -> Result<Vec<String>> {
120    let mut out = Vec::new();
121    for verb in verbs {
122        let Some(canonical) = canonicalize_query_verb(verb) else {
123            return Err(HeddleError::recovery(RecoveryDetails::invalid_usage(
124                "unknown_query_verb",
125                format!("unknown query verb '{verb}'"),
126                "Use a catalog verb such as `capture` (stored as `snapshot`).",
127            )));
128        };
129        if !out.iter().any(|existing| existing == canonical) {
130            out.push(canonical.to_string());
131        }
132    }
133    Ok(out)
134}
135
136fn fill_actor_email_from_state(repo: &repo::Repository, hit: &mut IndexedOperation) -> Result<()> {
137    if !hit.actor_email.is_empty() {
138        return Ok(());
139    }
140    let Some(state_id) = hit.state_id else {
141        return Ok(());
142    };
143    let Some(state) = repo.store().get_state(&state_id)? else {
144        return Ok(());
145    };
146    let email = state.attribution.principal.email_lossy();
147    if !email.is_empty() {
148        hit.actor_email = email.into_owned();
149    }
150    Ok(())
151}
152
153fn build_query(req: &QueryRequest) -> Result<OperationLogQuery> {
154    let resolved = resolve_query_verbs(&req.verbs)?;
155    let mut q = OperationLogQuery {
156        actor: (!req.actor.is_empty()).then(|| req.actor.clone()),
157        symbol: (!req.symbol.is_empty()).then(|| req.symbol.clone()),
158        signal_kind: (!req.signal_kind.is_empty()).then(|| req.signal_kind.clone()),
159        thread: (!req.thread.is_empty()).then(|| req.thread.clone()),
160        verbs: (!resolved.is_empty()).then_some(resolved),
161        since: parse_unix_secs(req.since_secs),
162        until: parse_unix_secs(req.until_secs),
163        limit: (req.limit > 0).then_some(req.limit as usize),
164    };
165    if !req.include_checkpoints && q.verbs.is_none() {
166        q.verbs = Some(
167            OpRecord::verbs(false)
168                .iter()
169                .map(|s| s.to_string())
170                .collect(),
171        );
172    }
173    Ok(q)
174}
175
176fn parse_unix_secs(secs: i64) -> Option<chrono::DateTime<chrono::Utc>> {
177    if secs == 0 {
178        return None;
179    }
180    chrono::Utc.timestamp_opt(secs, 0).single()
181}
182
183fn query_combined(heddle_dir: &Path, query: &OperationLogQuery) -> Result<Vec<IndexedOperation>> {
184    let index = OperationLogIndex::new(heddle_dir);
185    let mut unbounded = query.clone();
186    unbounded.limit = None;
187
188    let mut by_seq = BTreeMap::new();
189    for hit in index.query(&unbounded)? {
190        by_seq.insert(hit.seq, hit);
191    }
192
193    if unbounded.symbol.is_none() && unbounded.signal_kind.is_none() {
194        for hit in query_oplog_fallback(heddle_dir, &unbounded)? {
195            by_seq.entry(hit.seq).or_insert(hit);
196        }
197    }
198
199    let mut hits: Vec<_> = by_seq.into_values().collect();
200    hits.sort_by_key(|hit| hit.seq);
201    if let Some(limit) = query.limit {
202        hits.truncate(limit);
203    }
204    Ok(hits)
205}
206
207fn query_oplog_fallback(
208    heddle_dir: &Path,
209    query: &OperationLogQuery,
210) -> Result<Vec<IndexedOperation>> {
211    let log = OpLog::new_unattributed(heddle_dir);
212    let mut entries = log.recent(OPLOG_FALLBACK_SCAN_WINDOW)?;
213    entries.reverse();
214    let mut hits = Vec::new();
215    for entry in entries {
216        let hit = indexed_from_oplog_entry(&entry);
217        if hit.matches(query) {
218            hits.push(hit);
219        }
220    }
221    Ok(hits)
222}
223
224fn indexed_from_oplog_entry(entry: &OpEntry) -> IndexedOperation {
225    IndexedOperation {
226        seq: entry.id,
227        timestamp_secs: entry.timestamp.timestamp(),
228        verb: entry.operation.verb().to_string(),
229        actor_email: entry.actor.email_lossy().into_owned(),
230        operation_id: entry.operation_id,
231        thread: thread_for(&entry.operation),
232        symbols: Vec::new(),
233        signal_kinds: Vec::new(),
234        state_id: primary_state_id(&entry.operation),
235    }
236}
237
238fn hit_to_report(hit: IndexedOperation) -> QueryHit {
239    QueryHit {
240        seq: hit.seq,
241        timestamp_secs: hit.timestamp_secs,
242        verb: hit.verb,
243        actor_email: hit.actor_email,
244        operation_id: hit.operation_id.map(operation_id_to_string),
245        thread: hit.thread,
246        symbols: hit.symbols,
247        signal_kinds: hit.signal_kinds,
248        state_id: hit.state_id.map(|id| id.to_string_full()),
249    }
250}
251
252fn operation_id_to_string(id: OperationId) -> String {
253    id.to_string()
254}
255
256fn thread_for(op: &OpRecord) -> Option<String> {
257    match op {
258        OpRecord::Snapshot { thread, .. } => thread.clone(),
259        OpRecord::ThreadCreate { name, .. } => Some(name.clone()),
260        OpRecord::ThreadDelete { name, .. } => Some(name.clone()),
261        OpRecord::ThreadUpdate { name, .. } => Some(name.clone()),
262        OpRecord::MarkerCreate { name, .. } => Some(name.clone()),
263        OpRecord::MarkerDelete { name, .. } => Some(name.clone()),
264        OpRecord::Checkpoint { thread, .. } => thread.clone(),
265        OpRecord::EphemeralThreadCollapse { thread, .. } => Some(thread.clone()),
266        OpRecord::FastForward { target_thread, .. } => Some(target_thread.clone()),
267        OpRecord::GitCheckpoint { branch, .. } => Some(branch.clone()),
268        OpRecord::RemoteThreadUpdate { thread, .. }
269        | OpRecord::RemoteThreadDelete { thread, .. } => Some(thread.clone()),
270        OpRecord::HeadUpdate {
271            new: RecordedHead::Attached { thread },
272            ..
273        } => Some(thread.clone()),
274        OpRecord::Goto { .. }
275        | OpRecord::Fork { .. }
276        | OpRecord::Collapse { .. }
277        | OpRecord::TransactionAbort { .. }
278        | OpRecord::TransactionCommit { .. }
279        | OpRecord::ConflictResolved { .. }
280        | OpRecord::Redact { .. }
281        | OpRecord::UndoRecoveryUpdate { .. }
282        | OpRecord::StateVisibilitySet { .. }
283        | OpRecord::StateVisibilityPromote { .. }
284        | OpRecord::HeadUpdate {
285            new: RecordedHead::Detached { .. },
286            ..
287        }
288        | OpRecord::Purge { .. } => None,
289    }
290}
291
292fn primary_state_id(op: &OpRecord) -> Option<StateId> {
293    match op {
294        OpRecord::Snapshot { new_state, .. } => Some(*new_state),
295        OpRecord::Goto { target, .. } => Some(*target),
296        OpRecord::ThreadCreate { state, .. } => Some(*state),
297        OpRecord::ThreadDelete { state, .. } => Some(*state),
298        OpRecord::ThreadUpdate { new_state, .. } => Some(*new_state),
299        OpRecord::Fork { new_state, .. } => Some(*new_state),
300        OpRecord::Collapse { result, .. } => Some(*result),
301        OpRecord::MarkerCreate { state, .. } => Some(*state),
302        OpRecord::MarkerDelete { state, .. } => Some(*state),
303        OpRecord::Checkpoint { state, .. } => Some(*state),
304        OpRecord::GitCheckpoint { state, .. } => Some(*state),
305        OpRecord::EphemeralThreadCollapse { final_state, .. } => Some(*final_state),
306        OpRecord::Redact { state, .. } => Some(*state),
307        OpRecord::StateVisibilitySet { state, .. }
308        | OpRecord::StateVisibilityPromote { state, .. } => Some(*state),
309        OpRecord::RemoteThreadUpdate { state, .. } | OpRecord::RemoteThreadDelete { state, .. } => {
310            Some(*state)
311        }
312        OpRecord::UndoRecoveryUpdate { state } => Some(*state),
313        OpRecord::HeadUpdate {
314            new: RecordedHead::Detached { state },
315            ..
316        } => Some(*state),
317        OpRecord::TransactionAbort { .. }
318        | OpRecord::TransactionCommit { .. }
319        | OpRecord::ConflictResolved { .. }
320        | OpRecord::Purge { .. }
321        | OpRecord::FastForward { .. }
322        | OpRecord::HeadUpdate {
323            new: RecordedHead::Attached { .. },
324            ..
325        } => None,
326    }
327}
328
329#[cfg(test)]
330mod tests {
331    use super::*;
332
333    #[test]
334    fn resolve_query_verbs_maps_capture_case_insensitively() {
335        assert_eq!(
336            resolve_query_verbs(&["capture".into(), "Capture".into()]).unwrap(),
337            vec!["snapshot".to_string()]
338        );
339        assert_eq!(
340            resolve_query_verbs(&["SNAPSHOT".into()]).unwrap(),
341            vec!["snapshot".to_string()]
342        );
343    }
344
345    #[test]
346    fn build_query_uses_resolved_capture_verb() {
347        let q = build_query(&QueryRequest {
348            verbs: vec!["capture".into()],
349            limit: 10,
350            ..QueryRequest::default()
351        })
352        .unwrap();
353        assert_eq!(q.verbs, Some(vec!["snapshot".to_string()]));
354    }
355
356    #[test]
357    fn unknown_query_verb_is_an_error() {
358        let err = resolve_query_verbs(&["captur".into()]).expect_err("unknown verb");
359        let objects::error::HeddleError::Recovery(details) = err else {
360            panic!("expected recovery error, got {err:?}");
361        };
362        assert_eq!(details.kind, "unknown_query_verb");
363        assert!(details.error.contains("captur"));
364    }
365
366    #[test]
367    fn actor_filter_returns_backfilled_capture() {
368        let temp = tempfile::tempdir().unwrap();
369        let repo = repo::Repository::init_default(temp.path()).unwrap();
370        let tree = repo
371            .store()
372            .put_tree(&objects::object::Tree::new())
373            .unwrap();
374        let state = objects::object::State::new_snapshot(
375            tree,
376            Vec::new(),
377            objects::object::Attribution::human(objects::object::Principal::new(
378                "Heddle Test",
379                "heddle@example.com",
380            )),
381        );
382        repo.store().put_state(&state).unwrap();
383
384        let stored_seq = oplog::OpLog::new_unattributed(repo.heddle_dir())
385            .record_batch(vec![oplog::OpRecord::Snapshot {
386                new_state: state.id(),
387                prev_head: None,
388                head: Some(state.id()),
389                thread: None,
390            }])
391            .unwrap()[0];
392
393        let ctx = ExecutionContext::builder().repo(repo).build();
394        let report = query(
395            &ctx,
396            QueryRequest {
397                actor: "heddle@example.com".into(),
398                verbs: vec!["capture".into()],
399                ..QueryRequest::default()
400            },
401        )
402        .unwrap();
403        assert!(
404            report.hits.iter().any(|hit| hit.seq == stored_seq
405                && hit.verb == "snapshot"
406                && hit.actor_email == "heddle@example.com"),
407            "actor filter must match the backfilled capture: {report:?}"
408        );
409    }
410}
verbs/query.rs

verbs/
query.rs
