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::sync::Arc;
9
10use talon_core::types::{ScheduledTask, SessionRow};
11use talon_core::{Error, Result};
12use talon_db::{Db, Investigation};
13
14/// Persistence operations used by the HTTP channel.
15pub trait HttpStore: Send + Sync {
16    /// Scheduled task by id.
17    fn get_task_by_id(&self, id: &str) -> Result<Option<ScheduledTask>>;
18    /// Insert a scheduled task.
19    fn create_task(&self, task: &ScheduledTask) -> Result<()>;
20    /// Delete a scheduled task (and its run logs).
21    fn delete_task(&self, id: &str) -> Result<()>;
22    /// Tasks belonging to a group folder.
23    fn get_tasks_for_group(&self, group_folder: &str) -> Result<Vec<ScheduledTask>>;
24    /// Stored sessions, keyed by lane session key.
25    fn get_session_rows(&self) -> Result<Vec<SessionRow>>;
26    /// Insert or update an investigation job.
27    fn upsert_investigation(&self, inv: &Investigation) -> Result<()>;
28    /// Investigation by id.
29    fn get_investigation(&self, id: &str) -> Result<Option<Investigation>>;
30    /// Most recent investigations first.
31    fn list_investigations(&self, limit: usize) -> Result<Vec<Investigation>>;
32    /// Latest investigation for an alert id.
33    fn latest_investigation_for_alert(&self, alert_id: &str) -> Result<Option<Investigation>>;
34}
35
36impl HttpStore for Db {
37    fn get_task_by_id(&self, id: &str) -> Result<Option<ScheduledTask>> {
38        Db::get_task_by_id(self, id)
39    }
40    fn create_task(&self, task: &ScheduledTask) -> Result<()> {
41        Db::create_task(self, task)
42    }
43    fn delete_task(&self, id: &str) -> Result<()> {
44        Db::delete_task(self, id)
45    }
46    fn get_tasks_for_group(&self, group_folder: &str) -> Result<Vec<ScheduledTask>> {
47        Db::get_tasks_for_group(self, group_folder)
48    }
49    fn get_session_rows(&self) -> Result<Vec<SessionRow>> {
50        Db::get_session_rows(self)
51    }
52    fn upsert_investigation(&self, inv: &Investigation) -> Result<()> {
53        Db::upsert_investigation(self, inv)
54    }
55    fn get_investigation(&self, id: &str) -> Result<Option<Investigation>> {
56        Db::get_investigation(self, id)
57    }
58    fn list_investigations(&self, limit: usize) -> Result<Vec<Investigation>> {
59        Db::list_investigations(self, limit)
60    }
61    fn latest_investigation_for_alert(&self, alert_id: &str) -> Result<Option<Investigation>> {
62        Db::latest_investigation_for_alert(self, alert_id)
63    }
64}
65
66/// Run a store operation on the blocking pool. A panic inside the store is
67/// contained and reported as an error instead of tearing down the request
68/// task.
69pub(crate) async fn run_blocking<T, F>(store: &Arc<dyn HttpStore>, f: F) -> Result<T>
70where
71    T: Send + 'static,
72    F: FnOnce(&dyn HttpStore) -> Result<T> + Send + 'static,
73{
74    let store = Arc::clone(store);
75    tokio::task::spawn_blocking(move || f(store.as_ref()))
76        .await
77        .map_err(|e| Error::Db(format!("store task failed: {e}")))?
78}
talon_http/store.rs

talon_http/
store.rs
