talon_http/
store.rs

1//! The slice of `talon-db` the HTTP channel uses, behind a trait so the
2//! server can be tested with an in-memory fake. [`talon_db::Db`] implements
3//! it by delegation.
4//!
5//! All methods are blocking (SQLite); the server calls them through
6//! [`run_blocking`] so they never stall the async runtime.
7
8use std::collections::BTreeMap;
9use std::sync::Arc;
10
11use talon_core::types::{ScheduledTask, SessionRow};
12use talon_core::{Error, Result};
13use talon_db::{Db, Investigation};
14
15/// Persistence operations used by the HTTP channel.
16pub trait HttpStore: Send + Sync {
17    /// Scheduled task by id.
18    fn get_task_by_id(&self, id: &str) -> Result<Option<ScheduledTask>>;
19    /// Insert a scheduled task.
20    fn create_task(&self, task: &ScheduledTask) -> Result<()>;
21    /// Delete a scheduled task (and its run logs).
22    fn delete_task(&self, id: &str) -> Result<()>;
23    /// Tasks belonging to a group folder.
24    fn get_tasks_for_group(&self, group_folder: &str) -> Result<Vec<ScheduledTask>>;
25    /// Stored sessions, keyed by lane session key.
26    fn get_session_rows(&self) -> Result<Vec<SessionRow>>;
27    /// Insert or update an investigation job.
28    fn upsert_investigation(&self, inv: &Investigation) -> Result<()>;
29    /// Investigation by id.
30    fn get_investigation(&self, id: &str) -> Result<Option<Investigation>>;
31    /// Most recent investigations first.
32    fn list_investigations(&self, limit: usize) -> Result<Vec<Investigation>>;
33    /// Latest investigation for an alert id.
34    fn latest_investigation_for_alert(&self, alert_id: &str) -> Result<Option<Investigation>>;
35    /// Number of investigations per status (`queued`, `running`, ...).
36    fn investigation_status_counts(&self) -> Result<BTreeMap<String, u64>> {
37        let mut counts = BTreeMap::new();
38        for inv in self.list_investigations(usize::MAX)? {
39            *counts.entry(inv.status).or_insert(0) += 1;
40        }
41        Ok(counts)
42    }
43}
44
45impl HttpStore for Db {
46    fn get_task_by_id(&self, id: &str) -> Result<Option<ScheduledTask>> {
47        Db::get_task_by_id(self, id)
48    }
49    fn create_task(&self, task: &ScheduledTask) -> Result<()> {
50        Db::create_task(self, task)
51    }
52    fn delete_task(&self, id: &str) -> Result<()> {
53        Db::delete_task(self, id)
54    }
55    fn get_tasks_for_group(&self, group_folder: &str) -> Result<Vec<ScheduledTask>> {
56        Db::get_tasks_for_group(self, group_folder)
57    }
58    fn get_session_rows(&self) -> Result<Vec<SessionRow>> {
59        Db::get_session_rows(self)
60    }
61    fn upsert_investigation(&self, inv: &Investigation) -> Result<()> {
62        Db::upsert_investigation(self, inv)
63    }
64    fn get_investigation(&self, id: &str) -> Result<Option<Investigation>> {
65        Db::get_investigation(self, id)
66    }
67    fn list_investigations(&self, limit: usize) -> Result<Vec<Investigation>> {
68        Db::list_investigations(self, limit)
69    }
70    fn latest_investigation_for_alert(&self, alert_id: &str) -> Result<Option<Investigation>> {
71        Db::latest_investigation_for_alert(self, alert_id)
72    }
73    fn investigation_status_counts(&self) -> Result<BTreeMap<String, u64>> {
74        Db::investigation_status_counts(self)
75    }
76}
77
78/// Run a store operation on the blocking pool. A panic inside the store is
79/// contained and reported as an error instead of tearing down the request
80/// task.
81pub(crate) async fn run_blocking<T, F>(store: &Arc<dyn HttpStore>, f: F) -> Result<T>
82where
83    T: Send + 'static,
84    F: FnOnce(&dyn HttpStore) -> Result<T> + Send + 'static,
85{
86    let store = Arc::clone(store);
87    tokio::task::spawn_blocking(move || f(store.as_ref()))
88        .await
89        .map_err(|e| Error::Db(format!("store task failed: {e}")))?
90}
talon_http/store.rs

talon_http/
store.rs
