diff --git a/crates/openhuman-cli/Cargo.toml b/crates/openhuman-cli/Cargo.toml index 5341e1c4296..ec37de0f7b2 100644 --- a/crates/openhuman-cli/Cargo.toml +++ b/crates/openhuman-cli/Cargo.toml @@ -122,6 +122,12 @@ path = "../../tests/cost_cap_removed.rs" name = "storage_approvals_e2e" path = "../../tests/storage_approvals_e2e.rs" +[[test]] +# Its own binary for the same reason as storage_approvals_e2e: it installs a +# storage backend into the process-wide slot. +name = "storage_domains_e2e" +path = "../../tests/storage_domains_e2e.rs" + [[test]] name = "embedded_server_shutdown_e2e" path = "../../tests/embedded_server_shutdown_e2e.rs" diff --git a/crates/openhuman-core/src/desktop/notifications/README.md b/crates/openhuman-core/src/desktop/notifications/README.md index 6c5eb04e182..688a5b55ec4 100644 --- a/crates/openhuman-core/src/desktop/notifications/README.md +++ b/crates/openhuman-core/src/desktop/notifications/README.md @@ -85,6 +85,22 @@ SQLite DB at `{workspace_dir}/notifications/notifications.db`, opened per-call v `insert_if_not_recent` runs a `BEGIN IMMEDIATE` transaction so concurrent duplicate ingests collapse to a single insert. +### On a storage backend + +When the host configured a storage backend (`OPENHUMAN_STORAGE_URL` / +`[storage] url`, see `crate::storage`), every `store` function uses +`store_documents.rs` instead of `notifications.db`: the same operations on the +`tinystoragedrivers` document port, under the current call's storage scope +(the acting agent; `local` on a single-user host; refused in SaaS mode with +no acting agent). Collections `integration_notifications`, `notification_dedup`, +`notification_settings` and `core_notifications`. Every insert first +advances a `notification_dedup` document (a hash of provider, account, +title and body) under compare-and-swap, which replaces the SQL store's +`BEGIN IMMEDIATE`: two processes ingesting the same content in the same +minute insert it once. `stats` folds the counts in the store, since the +port has no `GROUP BY`. With no backend configured (the desktop default) +`notifications.db` is used as described above. + ## Dependencies - `crate::core::bus::BUS` and `crate::core::events::DomainEvent` (`BUS.subscribe` for the bridge, `BUS.publish` for triage results); the `EventHandler` trait comes from `tinybus`. diff --git a/crates/openhuman-core/src/desktop/notifications/mod.rs b/crates/openhuman-core/src/desktop/notifications/mod.rs index 26e05c3d13b..320ff862517 100644 --- a/crates/openhuman-core/src/desktop/notifications/mod.rs +++ b/crates/openhuman-core/src/desktop/notifications/mod.rs @@ -26,6 +26,7 @@ pub mod bus; pub mod rpc; pub mod schemas; pub mod store; +mod store_documents; pub mod types; pub use bus::{ diff --git a/crates/openhuman-core/src/desktop/notifications/rpc.rs b/crates/openhuman-core/src/desktop/notifications/rpc.rs index f7092dae9b3..bc7eb655355 100644 --- a/crates/openhuman-core/src/desktop/notifications/rpc.rs +++ b/crates/openhuman-core/src/desktop/notifications/rpc.rs @@ -88,7 +88,7 @@ pub async fn handle_ingest(params: Map) -> Result // Spawn background triage — the ingest RPC returns immediately. let id_for_triage = id.clone(); let config_for_triage = config.clone(); - tokio::spawn(async move { + crate::core::runtime::spawn_scoped(async move { let envelope = TriggerEnvelope { source: TriggerSource::WebviewIntegration { provider: req.provider.clone(), diff --git a/crates/openhuman-core/src/desktop/notifications/store.rs b/crates/openhuman-core/src/desktop/notifications/store.rs index 36733b83a82..48edf771cf6 100644 --- a/crates/openhuman-core/src/desktop/notifications/store.rs +++ b/crates/openhuman-core/src/desktop/notifications/store.rs @@ -1,10 +1,13 @@ //! SQLite persistence for `IntegrationNotification` records. //! +//! With a storage backend configured ([`crate::storage`]) every function +//! here is served from the document port instead (`store_documents.rs`). +//! //! Uses a synchronous `rusqlite::Connection` opened per call, following the //! same `with_connection` pattern as the cron domain. use anyhow::{Context, Result}; -use chrono::{DateTime, Utc}; +use chrono::Utc; use rusqlite::{params, Connection}; use crate::config::Config; @@ -106,6 +109,9 @@ fn with_connection(config: &Config, f: impl FnOnce(&Connection) -> Result) /// Persist a new notification to the store. pub fn insert(config: &Config, n: &IntegrationNotification) -> Result<()> { + if let Some(docs) = super::store_documents::current()? { + return docs.insert(n, false).map(|_| ()); + } with_connection(config, |conn| { conn.execute( "INSERT INTO integration_notifications @@ -144,6 +150,9 @@ pub fn insert(config: &Config, n: &IntegrationNotification) -> Result<()> { /// event is ignored (no duplicates). Returns `true` when a new row was written, /// `false` when an event with the same id already existed. pub fn insert_core_notification(config: &Config, event: &CoreNotificationEvent) -> Result { + if let Some(docs) = super::store_documents::current()? { + return docs.insert_core_notification(&config.workspace_dir.to_string_lossy(), event); + } with_connection(config, |conn| { let payload = serde_json::to_string(event) .context("[notifications::store] serialize core notification failed")?; @@ -172,6 +181,13 @@ pub fn list_core_notifications( only_unread: bool, limit: usize, ) -> Result> { + if let Some(docs) = super::store_documents::current()? { + return docs.list_core_notifications( + &config.workspace_dir.to_string_lossy(), + only_unread, + limit, + ); + } with_connection(config, |conn| { let sql = if only_unread { "SELECT payload FROM core_notifications WHERE read = 0 @@ -207,6 +223,9 @@ pub fn list_core_notifications( /// Mark a persisted core notification as read so it isn't re-surfaced on the /// next sync-down. Returns `true` when a row was updated. pub fn mark_core_notification_read(config: &Config, id: &str) -> Result { + if let Some(docs) = super::store_documents::current()? { + return docs.mark_core_notification_read(&config.workspace_dir.to_string_lossy(), id); + } with_connection(config, |conn| { let affected = conn .execute( @@ -220,6 +239,9 @@ pub fn mark_core_notification_read(config: &Config, id: &str) -> Result { /// Count unread persisted core notifications. pub fn unread_core_notification_count(config: &Config) -> Result { + if let Some(docs) = super::store_documents::current()? { + return docs.unread_core_notification_count(&config.workspace_dir.to_string_lossy()); + } with_connection(config, |conn| { let count: i64 = conn .query_row( @@ -236,6 +258,9 @@ pub fn unread_core_notification_count(config: &Config) -> Result { /// /// Returns `true` when inserted, `false` when skipped as duplicate. pub fn insert_if_not_recent(config: &Config, n: &IntegrationNotification) -> Result { + if let Some(docs) = super::store_documents::current()? { + return docs.insert(n, true); + } with_connection(config, |conn| { conn.execute_batch("BEGIN IMMEDIATE") .context("[notifications::store] begin insert_if_not_recent tx failed")?; @@ -313,6 +338,9 @@ pub fn list( provider_filter: Option<&str>, min_score: Option, ) -> Result> { + if let Some(docs) = super::store_documents::current()? { + return docs.list(limit, offset, provider_filter, min_score); + } with_connection(config, |conn| { // Build a dynamic query instead of relying on nullable-aware WHERE // logic so the SQL stays readable for future contributors. @@ -360,6 +388,13 @@ pub fn update_triage( action: &str, reason: &str, ) -> Result<()> { + if let Some(docs) = super::store_documents::current()? { + return docs.update_triage(id, score, action, reason).map(|found| { + if !found { + tracing::warn!(id = %id, action = %action, "[notifications::store] update_triage matched no rows"); + } + }); + } with_connection(config, |conn| { let now = Utc::now().to_rfc3339(); let updated = conn @@ -393,6 +428,13 @@ pub fn update_triage( /// Transition a notification from `unread` to `read`. pub fn mark_read(config: &Config, id: &str) -> Result<()> { + if let Some(docs) = super::store_documents::current()? { + return docs.set_status(id, NotificationStatus::Read).map(|found| { + if !found { + tracing::warn!(id = %id, "[notifications::store] mark_read matched no rows"); + } + }); + } with_connection(config, |conn| { let updated = conn .execute( @@ -414,6 +456,9 @@ pub fn mark_read(config: &Config, id: &str) -> Result<()> { /// Count unread notifications. pub fn unread_count(config: &Config) -> Result { + if let Some(docs) = super::store_documents::current()? { + return docs.unread_count(); + } with_connection(config, |conn| { let count: i64 = conn .query_row( @@ -435,6 +480,9 @@ pub fn exists_recent( title: &str, body: &str, ) -> Result { + if let Some(docs) = super::store_documents::current()? { + return docs.exists_recent(provider, account_id, title, body); + } with_connection(config, |conn| { let count: i64 = match account_id { Some(aid) => conn.query_row( @@ -463,6 +511,9 @@ pub fn exists_recent( /// /// Returns `true` when at least one row matched and was updated. pub fn mark_dismissed(config: &Config, id: &str) -> Result { + if let Some(docs) = super::store_documents::current()? { + return docs.set_status(id, NotificationStatus::Dismissed); + } with_connection(config, |conn| { let updated = conn .execute( @@ -484,6 +535,9 @@ pub fn mark_dismissed(config: &Config, id: &str) -> Result { /// /// Returns `true` when at least one row matched and was updated. pub fn mark_acted(config: &Config, id: &str) -> Result { + if let Some(docs) = super::store_documents::current()? { + return docs.set_status(id, NotificationStatus::Acted); + } with_connection(config, |conn| { let updated = conn .execute( @@ -503,6 +557,9 @@ pub fn mark_acted(config: &Config, id: &str) -> Result { /// Return aggregate statistics for the notification intelligence pipeline. pub fn stats(config: &Config) -> Result { + if let Some(docs) = super::store_documents::current()? { + return docs.stats(); + } use std::collections::HashMap; with_connection(config, |conn| { let total: i64 = conn @@ -587,6 +644,9 @@ pub fn stats(config: &Config) -> Result { /// Upsert provider-level notification settings. pub fn upsert_settings(config: &Config, settings: &NotificationSettings) -> Result<()> { + if let Some(docs) = super::store_documents::current()? { + return docs.upsert_settings(settings); + } with_connection(config, |conn| { conn.execute( "INSERT INTO notification_settings (provider, enabled, importance_threshold, route_to_orchestrator) @@ -609,6 +669,9 @@ pub fn upsert_settings(config: &Config, settings: &NotificationSettings) -> Resu /// Read provider-level notification settings with defaults when missing. pub fn get_settings(config: &Config, provider: &str) -> Result { + if let Some(docs) = super::store_documents::current()? { + return docs.get_settings(provider); + } with_connection(config, |conn| { let mut stmt = conn .prepare( @@ -638,72 +701,8 @@ pub fn get_settings(config: &Config, provider: &str) -> Result) -> Result> { - let mut out = Vec::new(); - while let Some(row) = rows - .next() - .context("[notifications::store] row iteration failed")? - { - out.push(row_to_notification(row)?); - } - Ok(out) -} - -fn row_to_notification(row: &rusqlite::Row<'_>) -> Result { - let raw_payload_str: String = row.get(5)?; - let raw_payload: serde_json::Value = serde_json::from_str(&raw_payload_str) - .unwrap_or(serde_json::Value::String(raw_payload_str)); - - let status_str: String = row.get(9)?; - let status = match status_str.as_str() { - "read" => NotificationStatus::Read, - "acted" => NotificationStatus::Acted, - "dismissed" => NotificationStatus::Dismissed, - _ => NotificationStatus::Unread, - }; - - let received_at_str: String = row.get(10)?; - let received_at: DateTime = received_at_str.parse().unwrap_or_else(|e| { - tracing::warn!( - raw = %received_at_str, - error = %e, - "[notifications::store] invalid received_at, using now" - ); - Utc::now() - }); - - let scored_at_str: Option = row.get(11)?; - let scored_at: Option> = scored_at_str.and_then(|s| match s.parse() { - Ok(t) => Some(t), - Err(e) => { - tracing::warn!( - raw = %s, - error = %e, - "[notifications::store] invalid scored_at, treating as unscored" - ); - None - } - }); - - Ok(IntegrationNotification { - id: row.get(0)?, - provider: row.get(1)?, - account_id: row.get(2)?, - title: row.get(3)?, - body: row.get(4)?, - raw_payload, - importance_score: row.get(6)?, - triage_action: row.get(7)?, - triage_reason: row.get(8)?, - status, - received_at, - scored_at, - }) -} +mod store_rows; +use store_rows::rows_to_notifications; #[cfg(test)] #[path = "store_tests.rs"] diff --git a/crates/openhuman-core/src/desktop/notifications/store/store_rows.rs b/crates/openhuman-core/src/desktop/notifications/store/store_rows.rs new file mode 100644 index 00000000000..abafb02a1fe --- /dev/null +++ b/crates/openhuman-core/src/desktop/notifications/store/store_rows.rs @@ -0,0 +1,71 @@ +//! Row conversion for the SQLite notification store. + +use anyhow::{Context, Result}; +use chrono::{DateTime, Utc}; + +use super::super::types::{IntegrationNotification, NotificationStatus}; + +pub(super) fn rows_to_notifications( + mut rows: rusqlite::Rows<'_>, +) -> Result> { + let mut out = Vec::new(); + while let Some(row) = rows + .next() + .context("[notifications::store] row iteration failed")? + { + out.push(row_to_notification(row)?); + } + Ok(out) +} + +fn row_to_notification(row: &rusqlite::Row<'_>) -> Result { + let raw_payload_str: String = row.get(5)?; + let raw_payload: serde_json::Value = serde_json::from_str(&raw_payload_str) + .unwrap_or(serde_json::Value::String(raw_payload_str)); + + let status_str: String = row.get(9)?; + let status = match status_str.as_str() { + "read" => NotificationStatus::Read, + "acted" => NotificationStatus::Acted, + "dismissed" => NotificationStatus::Dismissed, + _ => NotificationStatus::Unread, + }; + + let received_at_str: String = row.get(10)?; + let received_at: DateTime = received_at_str.parse().unwrap_or_else(|e| { + tracing::warn!( + raw = %received_at_str, + error = %e, + "[notifications::store] invalid received_at, using now" + ); + Utc::now() + }); + + let scored_at_str: Option = row.get(11)?; + let scored_at: Option> = scored_at_str.and_then(|s| match s.parse() { + Ok(t) => Some(t), + Err(e) => { + tracing::warn!( + raw = %s, + error = %e, + "[notifications::store] invalid scored_at, treating as unscored" + ); + None + } + }); + + Ok(IntegrationNotification { + id: row.get(0)?, + provider: row.get(1)?, + account_id: row.get(2)?, + title: row.get(3)?, + body: row.get(4)?, + raw_payload, + importance_score: row.get(6)?, + triage_action: row.get(7)?, + triage_reason: row.get(8)?, + status, + received_at, + scored_at, + }) +} diff --git a/crates/openhuman-core/src/desktop/notifications/store_documents.rs b/crates/openhuman-core/src/desktop/notifications/store_documents.rs new file mode 100644 index 00000000000..f03983a6eb9 --- /dev/null +++ b/crates/openhuman-core/src/desktop/notifications/store_documents.rs @@ -0,0 +1,521 @@ +//! Notifications on the `tinystoragedrivers` document port. +//! +//! Used instead of `notifications/notifications.db` when the host configured +//! a storage backend ([`crate::storage`]); the functions in `store.rs` pick it +//! per call. Records live under the current call's storage scope (the acting +//! agent, `local` on a single-user host). +//! +//! # Layout +//! +//! | Collection | Document id | Holds | +//! | --- | --- | --- | +//! | `integration_notifications` | notification id | one ingested notification | +//! | `notification_dedup` | hash of provider, account, title, body | when that content last arrived | +//! | `notification_settings` | provider | one provider's settings | +//! | `core_notifications` | event id | one persisted core notification event | +//! +//! Optional columns are left out of a document rather than stored as `null`, +//! so "is unscored" is a missing field on every driver. `received_ms` (epoch +//! milliseconds) orders the list, since RFC 3339 strings with differing +//! fractional digits do not sort as instants. +//! +//! The SQL store makes "insert unless the same content arrived in the last +//! minute" atomic with `BEGIN IMMEDIATE`. Here every insert advances a +//! `notification_dedup` document under compare-and-swap first, so two +//! processes ingesting the same notification at once insert it once. + +use std::collections::HashMap; +use std::sync::Arc; + +use anyhow::{Context, Result}; +use chrono::{DateTime, Duration, Utc}; +use serde_json::{json, Map, Value}; +use sha2::{Digest, Sha256}; +use tinystoragedrivers::{ + CollectionSpec, ErrorKind, Filter, IndexSpec, Precondition, Query, Sort, Versioned, +}; + +use super::types::{ + CoreNotificationEvent, IntegrationNotification, NotificationSettings, NotificationStats, + NotificationStatus, +}; +use crate::storage::documents::{compare_and_swap, text, Repo, CAS_ATTEMPTS}; +use crate::storage::{DocumentStore, DocumentStoreExt, StorageError}; + +const NOTIFICATIONS: &str = "integration_notifications"; +const DEDUP: &str = "notification_dedup"; +const SETTINGS: &str = "notification_settings"; +const CORE: &str = "core_notifications"; +const DOMAIN: &str = "notifications::store"; + +/// How long identical content counts as a duplicate. +const DEDUP_WINDOW_SECS: i64 = 60; + +fn collections() -> Vec { + vec![ + CollectionSpec::new(NOTIFICATIONS) + .index(IndexSpec::new("by_provider", ["provider", "received_ms"])) + .index(IndexSpec::new("by_status", ["status"])), + CollectionSpec::new(DEDUP), + CollectionSpec::new(SETTINGS), + CollectionSpec::new(CORE).index(IndexSpec::new( + "by_read", + ["workspace", "read", "timestamp_ms"], + )), + ] +} + +/// The document store for this call, when the host configured one. +pub(super) fn current() -> Result> { + Ok(Repo::current(DOMAIN, collections)?.map(Docs)) +} + +fn status_of(raw: Option<&str>) -> NotificationStatus { + match raw { + Some("read") => NotificationStatus::Read, + Some("acted") => NotificationStatus::Acted, + Some("dismissed") => NotificationStatus::Dismissed, + _ => NotificationStatus::Unread, + } +} + +fn parse_time(raw: Option<&str>) -> Option> { + raw.and_then(|value| value.parse().ok()) +} + +fn to_doc(n: &IntegrationNotification) -> Value { + let mut doc = Map::new(); + doc.insert("provider".into(), json!(n.provider)); + doc.insert("title".into(), json!(n.title)); + doc.insert("body".into(), json!(n.body)); + doc.insert("raw_payload".into(), json!(n.raw_payload.to_string())); + doc.insert("status".into(), json!(n.status.as_str())); + doc.insert("received_at".into(), json!(n.received_at.to_rfc3339())); + doc.insert( + "received_ms".into(), + json!(n.received_at.timestamp_millis()), + ); + let optional = [ + ("account_id", n.account_id.as_ref().map(|v| json!(v))), + ("importance_score", n.importance_score.map(|v| json!(v))), + ("triage_action", n.triage_action.as_ref().map(|v| json!(v))), + ("triage_reason", n.triage_reason.as_ref().map(|v| json!(v))), + ("scored_at", n.scored_at.map(|v| json!(v.to_rfc3339()))), + ]; + for (field, value) in optional { + if let Some(value) = value { + doc.insert(field.into(), value); + } + } + Value::Object(doc) +} + +fn to_notification(stored: &Versioned) -> IntegrationNotification { + let doc = &stored.doc; + let raw_payload = text(doc, "raw_payload").map_or(Value::Null, |raw| { + serde_json::from_str(raw).unwrap_or_else(|_| Value::String(raw.to_string())) + }); + IntegrationNotification { + id: stored.id.clone(), + provider: text(doc, "provider").unwrap_or_default().to_string(), + account_id: text(doc, "account_id").map(str::to_string), + title: text(doc, "title").unwrap_or_default().to_string(), + body: text(doc, "body").unwrap_or_default().to_string(), + raw_payload, + importance_score: doc + .get("importance_score") + .and_then(Value::as_f64) + .map(|score| score as f32), + triage_action: text(doc, "triage_action").map(str::to_string), + triage_reason: text(doc, "triage_reason").map(str::to_string), + status: status_of(text(doc, "status")), + received_at: parse_time(text(doc, "received_at")).unwrap_or_else(Utc::now), + scored_at: parse_time(text(doc, "scored_at")), + } +} + +/// The dedup document id for one piece of content. Each part is length +/// prefixed, so no two distinct tuples hash the same input. +fn dedup_id(provider: &str, account_id: Option<&str>, title: &str, body: &str) -> String { + let mut hasher = Sha256::new(); + for part in [Some(provider), account_id, Some(title), Some(body)] { + match part { + Some(part) => hasher.update(format!("{}:{part};", part.len())), + None => hasher.update("-;"), + } + } + hex::encode(hasher.finalize()) +} + +/// Records that the content arrived at `now_ms` (the wall clock, never the +/// notification's own timestamp, which a provider controls). With +/// `skip_recent`, does nothing and returns `None` when the same content +/// already arrived within [`DEDUP_WINDOW_SECS`] of `now_ms`. Otherwise +/// returns the previous arrival time (`Some(None)` when there was none), so a +/// failed insert can restore it with [`release_content`]. +async fn claim_content( + docs: &Arc, + id: &str, + now_ms: i64, + skip_recent: bool, +) -> Result>, StorageError> { + let window_start = now_ms - DEDUP_WINDOW_SECS * 1000; + for _ in 0..CAS_ATTEMPTS { + let stored = docs.get(DEDUP, id).await?; + let last = stored + .as_ref() + .and_then(|stored| stored.doc.get("last_ms")) + .and_then(Value::as_i64); + if skip_recent && last.is_some_and(|last| last >= window_start) { + return Ok(None); + } + let precondition = stored + .as_ref() + .map_or(Precondition::Absent, Versioned::unchanged); + match docs + .put(DEDUP, id, json!({ "last_ms": now_ms }), precondition) + .await + { + Ok(_) => return Ok(Some(last)), + Err(error) if error.kind() == ErrorKind::Conflict => {} + Err(error) => return Err(error), + } + } + Err(StorageError::conflict(format!( + "notification dedup {id} kept changing under {CAS_ATTEMPTS} attempts" + ))) +} + +/// Undoes a [`claim_content`] whose notification was never stored, so a retry +/// is not reported as a duplicate. Best effort: it only restores the claim if +/// it is still the one made at `claimed_ms`; a failure is logged, since the +/// caller is already returning the insert's own error. +async fn release_content( + docs: &Arc, + id: &str, + claimed_ms: i64, + previous: Option, +) { + let outcome: Result<(), StorageError> = async { + let Some(stored) = docs.get(DEDUP, id).await? else { + return Ok(()); + }; + if stored.doc.get("last_ms").and_then(Value::as_i64) != Some(claimed_ms) { + return Ok(()); + } + match previous { + Some(last) => { + docs.put(DEDUP, id, json!({ "last_ms": last }), stored.unchanged()) + .await?; + } + None => { + docs.delete(DEDUP, id, stored.unchanged()).await?; + } + } + Ok(()) + } + .await; + if let Err(error) = outcome { + tracing::warn!(%error, "[notifications::store] could not release dedup claim"); + } +} + +/// The core-notification document id: the workspace is part of the key, since +/// each workspace's events are persisted separately in the SQL store (one +/// database per workspace) and event ids repeat across them. +fn core_id(workspace: &str, event_id: &str) -> String { + format!("{}:{workspace}/{event_id}", workspace.len()) +} + +/// The notification store over one scoped document handle. +#[derive(Clone)] +pub(super) struct Docs(Repo); + +impl Docs { + #[cfg(test)] + pub(super) fn over(scoped: &crate::storage::ScopedStorage) -> Self { + Self(Repo::over(scoped, DOMAIN, collections)) + } + + /// Inserts `n`, failing if its id exists. With `skip_recent`, returns + /// `false` instead when the same content arrived in the last minute. + pub(super) fn insert(&self, n: &IntegrationNotification, skip_recent: bool) -> Result { + let id = n.id.clone(); + let doc = to_doc(n); + let dedup = dedup_id(&n.provider, n.account_id.as_deref(), &n.title, &n.body); + let now_ms = Utc::now().timestamp_millis(); + self.0.run(|docs| async move { + let Some(previous) = claim_content(&docs, &dedup, now_ms, skip_recent).await? else { + return Ok(false); + }; + if let Err(error) = docs + .put(NOTIFICATIONS, &id, doc, Precondition::Absent) + .await + { + release_content(&docs, &dedup, now_ms, previous).await; + return Err(error); + } + Ok(true) + }) + } + + pub(super) fn exists_recent( + &self, + provider: &str, + account_id: Option<&str>, + title: &str, + body: &str, + ) -> Result { + let id = dedup_id(provider, account_id, title, body); + let window_start = (Utc::now() - Duration::seconds(DEDUP_WINDOW_SECS)).timestamp_millis(); + self.0.run(|docs| async move { + Ok(docs + .get(DEDUP, &id) + .await? + .and_then(|stored| stored.doc.get("last_ms").and_then(Value::as_i64)) + .is_some_and(|last| last >= window_start)) + }) + } + + pub(super) fn list( + &self, + limit: usize, + offset: usize, + provider: Option<&str>, + min_score: Option, + ) -> Result> { + if limit == 0 { + return Ok(Vec::new()); + } + let filter = [ + provider.map(|provider| Filter::eq("provider", provider)), + min_score.map(|score| { + Filter::exists("importance_score", false) + .or(Filter::gte("importance_score", f64::from(score))) + }), + ] + .into_iter() + .flatten() + .reduce(Filter::and) + .unwrap_or(Filter::All); + let query = Query::filter(filter) + .sort(Sort::desc("received_ms")) + .limit(offset.saturating_add(limit)); + self.0.run(|docs| async move { + let page = docs.query(NOTIFICATIONS, &query).await?; + Ok(page + .items + .iter() + .skip(offset) + .take(limit) + .map(to_notification) + .collect()) + }) + } + + pub(super) fn update_triage( + &self, + id: &str, + score: f32, + action: &str, + reason: &str, + ) -> Result { + let (score, action, reason) = (f64::from(score), action.to_string(), reason.to_string()); + let scored_at = Utc::now().to_rfc3339(); + self.set(id, move |next| { + next["importance_score"] = json!(score); + next["triage_action"] = json!(action); + next["triage_reason"] = json!(reason); + next["scored_at"] = json!(scored_at); + }) + } + + pub(super) fn set_status(&self, id: &str, status: NotificationStatus) -> Result { + self.set(id, move |next| next["status"] = json!(status.as_str())) + } + + /// Applies `edit` to notification `id`; `false` when it does not exist. + fn set(&self, id: &str, edit: impl Fn(&mut Value) + Send + Sync + 'static) -> Result { + let id = id.to_string(); + self.0.run(|docs| async move { + let changed = compare_and_swap(&docs, NOTIFICATIONS, &id, |doc| { + let mut next = doc.clone(); + edit(&mut next); + Some(next) + }) + .await?; + Ok(changed.is_some()) + }) + } + + pub(super) fn unread_count(&self) -> Result { + self.0.run(|docs| async move { + let count = docs + .count(NOTIFICATIONS, &Filter::eq("status", "unread")) + .await?; + Ok(i64::try_from(count).unwrap_or(i64::MAX)) + }) + } + + /// Aggregates over every notification in the scope. The port has no + /// `GROUP BY`, so the counts are folded here; the notification center + /// holds one user's notifications, which keeps this bounded in practice. + pub(super) fn stats(&self) -> Result { + self.0.run(|docs| async move { + let mut stats = NotificationStats { + total: 0, + unread: 0, + unscored: 0, + by_provider: HashMap::new(), + by_action: HashMap::new(), + }; + for stored in docs.query_all(NOTIFICATIONS, &Query::all()).await? { + let doc = &stored.doc; + stats.total += 1; + if text(doc, "status") == Some("unread") { + stats.unread += 1; + } + if doc.get("importance_score").is_none() { + stats.unscored += 1; + } + let provider = text(doc, "provider").unwrap_or_default().to_string(); + *stats.by_provider.entry(provider).or_default() += 1; + if let Some(action) = text(doc, "triage_action") { + *stats.by_action.entry(action.to_string()).or_default() += 1; + } + } + Ok(stats) + }) + } + + pub(super) fn upsert_settings(&self, settings: &NotificationSettings) -> Result<()> { + let id = settings.provider.clone(); + let doc = json!({ + "enabled": settings.enabled, + "importance_threshold": f64::from(settings.importance_threshold), + "route_to_orchestrator": settings.route_to_orchestrator, + }); + self.0.run(|docs| async move { + docs.put(SETTINGS, &id, doc, Precondition::None) + .await + .map(|_| ()) + }) + } + + pub(super) fn get_settings(&self, provider: &str) -> Result { + let id = provider.to_string(); + self.0.run(|docs| async move { + let defaults = NotificationSettings { + provider: id.clone(), + ..NotificationSettings::default() + }; + let Some(stored) = docs.get(SETTINGS, &id).await? else { + return Ok(defaults); + }; + let doc = &stored.doc; + Ok(NotificationSettings { + enabled: doc + .get("enabled") + .and_then(Value::as_bool) + .unwrap_or(defaults.enabled), + importance_threshold: doc + .get("importance_threshold") + .and_then(Value::as_f64) + .map_or(defaults.importance_threshold, |value| value as f32), + route_to_orchestrator: doc + .get("route_to_orchestrator") + .and_then(Value::as_bool) + .unwrap_or(defaults.route_to_orchestrator), + provider: id, + }) + }) + } + + pub(super) fn insert_core_notification( + &self, + workspace: &str, + event: &CoreNotificationEvent, + ) -> Result { + let id = core_id(workspace, &event.id); + let doc = json!({ + "workspace": workspace, + "payload": serde_json::to_string(event) + .context("[notifications::store] serialize core notification failed")?, + "timestamp_ms": event.timestamp_ms, + "read": false, + "created_at": Utc::now().to_rfc3339(), + }); + self.0.run(|docs| async move { + match docs.put(CORE, &id, doc, Precondition::Absent).await { + Ok(_) => Ok(true), + // A re-publish of the same event id is ignored. + Err(error) if error.kind() == ErrorKind::Conflict => Ok(false), + Err(error) => Err(error), + } + }) + } + + pub(super) fn list_core_notifications( + &self, + workspace: &str, + only_unread: bool, + limit: usize, + ) -> Result> { + if limit == 0 { + return Ok(Vec::new()); + } + let in_workspace = Filter::eq("workspace", workspace); + let filter = if only_unread { + in_workspace.and(Filter::eq("read", false)) + } else { + in_workspace + }; + let query = Query::filter(filter) + .sort(Sort::desc("timestamp_ms")) + .limit(limit); + self.0.run(|docs| async move { + let page = docs.query(CORE, &query).await?; + Ok(page + .items + .iter() + .filter_map(|stored| { + let payload = text(&stored.doc, "payload")?; + serde_json::from_str(payload) + .map_err(|error| { + // A single corrupt record must not break the sync-down. + tracing::warn!( + %error, + "[notifications::store] skipping undeserializable core notification" + ); + }) + .ok() + }) + .collect()) + }) + } + + pub(super) fn mark_core_notification_read(&self, workspace: &str, id: &str) -> Result { + let id = core_id(workspace, id); + self.0.run(|docs| async move { + let marked = compare_and_swap(&docs, CORE, &id, |doc| { + let mut next = doc.clone(); + next["read"] = json!(true); + Some(next) + }) + .await?; + Ok(marked.is_some()) + }) + } + + pub(super) fn unread_core_notification_count(&self, workspace: &str) -> Result { + let filter = Filter::eq("workspace", workspace).and(Filter::eq("read", false)); + self.0.run(|docs| async move { + let count = docs.count(CORE, &filter).await?; + Ok(i64::try_from(count).unwrap_or(i64::MAX)) + }) + } +} + +#[cfg(test)] +#[path = "store_documents_tests.rs"] +mod tests; diff --git a/crates/openhuman-core/src/desktop/notifications/store_documents_tests.rs b/crates/openhuman-core/src/desktop/notifications/store_documents_tests.rs new file mode 100644 index 00000000000..5de61b2832d --- /dev/null +++ b/crates/openhuman-core/src/desktop/notifications/store_documents_tests.rs @@ -0,0 +1,347 @@ +use super::*; +use crate::desktop::notifications::types::CoreNotificationCategory; +use crate::storage::{MemoryStorage, Scope, StorageBackend}; + +const WS: &str = "/workspace/a"; + +fn docs_in(storage: &MemoryStorage, scope: &str) -> Docs { + Docs::over(&storage.for_scope(&Scope::new(scope).unwrap()).unwrap()) +} + +fn docs() -> Docs { + docs_in(&MemoryStorage::new(), "local") +} + +fn note(id: &str, provider: &str, body: &str, secs_ago: i64) -> IntegrationNotification { + IntegrationNotification { + id: id.to_string(), + provider: provider.to_string(), + account_id: Some("acct".to_string()), + title: "New message".to_string(), + body: body.to_string(), + raw_payload: json!({ "k": 1 }), + importance_score: None, + triage_action: None, + triage_reason: None, + status: NotificationStatus::Unread, + received_at: Utc::now() - Duration::seconds(secs_ago), + scored_at: None, + } +} + +fn event(id: &str, ts: u64) -> CoreNotificationEvent { + CoreNotificationEvent { + id: id.to_string(), + category: CoreNotificationCategory::Agents, + title: "Cron job completed".to_string(), + body: "done".to_string(), + deep_link: None, + timestamp_ms: ts, + actions: None, + workspace: None, + workspace_revision: None, + } +} + +fn ids(list: &[IntegrationNotification]) -> Vec<&str> { + list.iter().map(|n| n.id.as_str()).collect() +} + +#[test] +fn notifications_round_trip_newest_first() { + let store = docs(); + assert!(store.insert(¬e("old", "slack", "a", 30), false).unwrap()); + assert!(store.insert(¬e("new", "gmail", "b", 0), false).unwrap()); + let listed = store.list(10, 0, None, None).unwrap(); + assert_eq!(ids(&listed), ["new", "old"]); + assert_eq!(listed[1].raw_payload, json!({ "k": 1 })); + assert_eq!(listed[1].account_id.as_deref(), Some("acct")); + assert!(listed[1].importance_score.is_none()); + assert!( + store.insert(¬e("old", "slack", "z", 0), false).is_err(), + "an id is inserted once" + ); + assert_eq!(ids(&store.list(1, 1, None, None).unwrap()), ["old"]); + assert!(store.list(0, 0, None, None).unwrap().is_empty()); + assert_eq!( + ids(&store.list(10, 0, Some("slack"), None).unwrap()), + ["old"] + ); +} + +#[test] +fn identical_content_within_a_minute_is_skipped() { + let store = docs(); + assert!(store.insert(¬e("a", "slack", "hi", 0), true).unwrap()); + assert!(!store.insert(¬e("b", "slack", "hi", 0), true).unwrap()); + assert!(store.insert(¬e("c", "slack", "other", 0), true).unwrap()); + assert!(store + .exists_recent("slack", Some("acct"), "New message", "hi") + .unwrap()); + assert!(!store + .exists_recent("slack", None, "New message", "hi") + .unwrap()); + assert!(!store + .exists_recent("gmail", Some("acct"), "New message", "hi") + .unwrap()); + assert_eq!(ids(&store.list(10, 0, None, None).unwrap()).len(), 2); +} + +#[test] +fn content_older_than_the_window_is_not_a_duplicate() { + let store = docs(); + let dedup = dedup_id("slack", Some("acct"), "New message", "hi"); + let stale = (Utc::now() - Duration::seconds(120)).timestamp_millis(); + store + .0 + .run(|docs| async move { + docs.put( + DEDUP, + &dedup, + json!({ "last_ms": stale }), + Precondition::Absent, + ) + .await + .map(|_| ()) + }) + .unwrap(); + assert!(!store + .exists_recent("slack", Some("acct"), "New message", "hi") + .unwrap()); + assert!(store.insert(¬e("b", "slack", "hi", 0), true).unwrap()); +} + +#[test] +fn dedup_follows_arrival_not_the_notifications_own_timestamp() { + let store = docs(); + // A delayed event (stamped two minutes ago) and a future-dated one both + // arrived now, so identical content right after is a duplicate either way. + assert!(store + .insert(¬e("a", "slack", "late", 120), true) + .unwrap()); + assert!(!store.insert(¬e("b", "slack", "late", 0), true).unwrap()); + assert!(store + .insert(¬e("c", "slack", "future", -3600), true) + .unwrap()); + assert!(!store + .insert(¬e("d", "slack", "future", 0), true) + .unwrap()); +} + +#[test] +fn a_failed_insert_releases_its_dedup_claim() { + let store = docs(); + store + .insert(¬e("taken", "slack", "first", 0), false) + .unwrap(); + // Same id, new content: the claim succeeds, the notification put fails. + assert!(store + .insert(¬e("taken", "slack", "second", 0), true) + .is_err()); + assert!( + !store + .exists_recent("slack", Some("acct"), "New message", "second") + .unwrap(), + "the claim was rolled back" + ); + assert!(store + .insert(¬e("fresh", "slack", "second", 0), true) + .unwrap()); +} + +#[test] +fn workspaces_keep_core_notifications_apart() { + let store = docs(); + assert!(store + .insert_core_notification("/workspace/a", &event("e", 1)) + .unwrap()); + assert!(store + .insert_core_notification("/workspace/b", &event("e", 2)) + .unwrap()); + let a = store + .list_core_notifications("/workspace/a", false, 10) + .unwrap(); + assert_eq!(a.len(), 1); + assert_eq!(a[0].timestamp_ms, 1); + assert!(store + .mark_core_notification_read("/workspace/a", "e") + .unwrap()); + assert_eq!( + store + .unread_core_notification_count("/workspace/a") + .unwrap(), + 0 + ); + assert_eq!( + store + .unread_core_notification_count("/workspace/b") + .unwrap(), + 1 + ); + assert!(!store + .mark_core_notification_read("/workspace/c", "e") + .unwrap()); +} + +#[test] +fn concurrent_duplicates_insert_once() { + let store = docs(); + let inserted: usize = (0..8) + .map(|i| { + let store = store.clone(); + std::thread::spawn(move || { + store + .insert(¬e(&format!("n{i}"), "slack", "same", 0), true) + .unwrap() + }) + }) + .collect::>() + .into_iter() + .map(|handle| usize::from(handle.join().unwrap())) + .sum(); + assert_eq!(inserted, 1); +} + +#[test] +fn triage_status_and_score_filter() { + let store = docs(); + store.insert(¬e("low", "slack", "a", 2), false).unwrap(); + store.insert(¬e("high", "slack", "b", 1), false).unwrap(); + store + .insert(¬e("unscored", "slack", "c", 0), false) + .unwrap(); + assert!(store.update_triage("low", 0.2, "drop", "noise").unwrap()); + assert!(store.update_triage("high", 0.9, "react", "urgent").unwrap()); + assert!(!store.update_triage("missing", 0.5, "drop", "x").unwrap()); + let filtered = store.list(10, 0, None, Some(0.5)).unwrap(); + assert_eq!(ids(&filtered), ["unscored", "high"]); + assert_eq!(filtered[1].triage_action.as_deref(), Some("react")); + assert!(filtered[1].scored_at.is_some()); + + assert_eq!(store.unread_count().unwrap(), 3); + assert!(store.set_status("low", NotificationStatus::Read).unwrap()); + assert!(store.set_status("high", NotificationStatus::Acted).unwrap()); + assert!(!store + .set_status("missing", NotificationStatus::Dismissed) + .unwrap()); + assert_eq!(store.unread_count().unwrap(), 1); + + let stats = store.stats().unwrap(); + assert_eq!((stats.total, stats.unread, stats.unscored), (3, 1, 1)); + assert_eq!(stats.by_provider.get("slack"), Some(&3)); + assert_eq!(stats.by_action.get("drop"), Some(&1)); + assert_eq!(stats.by_action.get("react"), Some(&1)); +} + +#[test] +fn settings_default_then_upsert() { + let store = docs(); + let defaults = store.get_settings("slack").unwrap(); + assert_eq!(defaults.provider, "slack"); + let mut settings = defaults.clone(); + settings.enabled = !defaults.enabled; + settings.importance_threshold = 0.25; + settings.route_to_orchestrator = !defaults.route_to_orchestrator; + store.upsert_settings(&settings).unwrap(); + let read = store.get_settings("slack").unwrap(); + assert_eq!(read.enabled, settings.enabled); + assert_eq!(read.importance_threshold, 0.25); + assert_eq!(read.route_to_orchestrator, settings.route_to_orchestrator); + assert_eq!( + store.get_settings("gmail").unwrap().enabled, + defaults.enabled + ); +} + +#[test] +fn core_notifications_persist_once_and_mark_read() { + let store = docs(); + assert!(store.insert_core_notification(WS, &event("a", 1)).unwrap()); + assert!(!store.insert_core_notification(WS, &event("a", 1)).unwrap()); + assert!(store.insert_core_notification(WS, &event("b", 2)).unwrap()); + let all = store.list_core_notifications(WS, false, 10).unwrap(); + assert_eq!( + all.iter().map(|e| e.id.as_str()).collect::>(), + ["b", "a"] + ); + assert_eq!(all[0], event("b", 2)); + assert_eq!(store.unread_core_notification_count(WS).unwrap(), 2); + assert!(store.mark_core_notification_read(WS, "b").unwrap()); + assert!( + store.mark_core_notification_read(WS, "b").unwrap(), + "still exists" + ); + assert!(!store.mark_core_notification_read(WS, "missing").unwrap()); + let unread = store.list_core_notifications(WS, true, 10).unwrap(); + assert_eq!( + unread.iter().map(|e| e.id.as_str()).collect::>(), + ["a"] + ); + assert_eq!(store.unread_core_notification_count(WS).unwrap(), 1); + assert!(store + .list_core_notifications(WS, false, 0) + .unwrap() + .is_empty()); +} + +#[test] +fn a_corrupt_core_payload_is_skipped() { + let store = docs(); + store + .insert_core_notification(WS, &event("good", 1)) + .unwrap(); + store + .0 + .run(|docs| async move { + docs.put( + CORE, + "bad", + json!({ "workspace": WS, "payload": "not json", "timestamp_ms": 2, "read": false }), + Precondition::Absent, + ) + .await + .map(|_| ()) + }) + .unwrap(); + let listed = store.list_core_notifications(WS, false, 10).unwrap(); + assert_eq!(listed.len(), 1); + assert_eq!(listed[0].id, "good"); +} + +#[test] +fn an_unparseable_raw_payload_reads_as_a_string() { + let stored = Versioned { + id: "x".to_string(), + version: tinystoragedrivers::Version::FIRST, + doc: json!({ "raw_payload": "plain text", "status": "weird" }), + }; + let read = to_notification(&stored); + assert_eq!(read.raw_payload, json!("plain text")); + assert_eq!(read.status, NotificationStatus::Unread); +} + +#[test] +fn scopes_keep_notifications_apart() { + let storage = MemoryStorage::new(); + docs_in(&storage, "alice") + .insert(¬e("a", "slack", "hi", 0), true) + .unwrap(); + let bob = docs_in(&storage, "bob"); + assert!(bob.list(10, 0, None, None).unwrap().is_empty()); + assert!(!bob + .exists_recent("slack", Some("acct"), "New message", "hi") + .unwrap()); + assert!(bob.insert(¬e("a", "slack", "hi", 0), true).unwrap()); +} + +#[test] +fn dedup_ids_do_not_collide_across_fields() { + assert_ne!( + dedup_id("a", None, "b", "c"), + dedup_id("a", Some(""), "b", "c") + ); + assert_ne!( + dedup_id("ab", None, "c", "d"), + dedup_id("a", None, "bc", "d") + ); +} diff --git a/crates/openhuman-core/src/integrations/task_sources/README.md b/crates/openhuman-core/src/integrations/task_sources/README.md index fd1e7c3862a..88dcffacfca 100644 --- a/crates/openhuman-core/src/integrations/task_sources/README.md +++ b/crates/openhuman-core/src/integrations/task_sources/README.md @@ -124,6 +124,19 @@ SQLite at `/task_sources/sources.db` (WAL, 5s busy timeout, migra The additive idempotent `ingested_tasks.card_id` migration preserves older databases. App-level defaults (enabled flag, default interval, per-fetch cap, auto_proactive) live in config (`TaskSourcesConfig`), not the store. +### On a storage backend + +When the host configured a storage backend (`OPENHUMAN_STORAGE_URL` / +`[storage] url`, see `crate::storage`), every `store` function uses +`store_documents.rs` instead of `sources.db`: the same operations on the +`tinystoragedrivers` document port, under the current call's storage scope +(the acting agent; `local` on a single-user host; refused in SaaS mode with +no acting agent). Collections `task_sources` (one per source) and `ingested_tasks` (one +per `(source_id, external_id)`). `update_source` applies the patch under +compare-and-swap, and removing a source removes its ledger entries (the +SQL cascade). With no backend configured (the desktop default) +`sources.db` is used as described above. + ## Dependencies - `crate::core::all`: `ControllerFuture`, `RegisteredController` for the RPC registry. diff --git a/crates/openhuman-core/src/integrations/task_sources/mod.rs b/crates/openhuman-core/src/integrations/task_sources/mod.rs index b582d4cbccb..fc16132eb09 100644 --- a/crates/openhuman-core/src/integrations/task_sources/mod.rs +++ b/crates/openhuman-core/src/integrations/task_sources/mod.rs @@ -24,6 +24,7 @@ pub mod pipeline; pub mod route; mod schemas; pub mod store; +mod store_documents; pub mod tools; pub mod types; diff --git a/crates/openhuman-core/src/integrations/task_sources/store.rs b/crates/openhuman-core/src/integrations/task_sources/store.rs index ce8fd614e7a..866040a52cc 100644 --- a/crates/openhuman-core/src/integrations/task_sources/store.rs +++ b/crates/openhuman-core/src/integrations/task_sources/store.rs @@ -8,7 +8,9 @@ //! the UI can list recently ingested items. //! //! Mirrors the `cron` domain's `with_connection` + migrate-on-open -//! pattern. +//! pattern. With a storage backend configured ([`crate::storage`]) every +//! function here is served from the document port instead +//! (`store_documents.rs`). use std::collections::HashSet; use std::path::{Path, PathBuf}; @@ -87,9 +89,27 @@ pub fn add_source( let now = Utc::now(); let filter_json = serde_json::to_string(&filter).context("serialize task source filter")?; let target_json = serde_json::to_string(&target).context("serialize task source target")?; + + if let Some(docs) = super::store_documents::current()? { + return docs.add_source(&TaskSource { + id, + provider, + connection_id, + name, + enabled: true, + filter, + interval_secs, + target, + max_tasks_per_fetch, + created_at: now, + last_fetch_at: None, + last_status: None, + }); + } + + // SQLite-only bound: the document store holds the full `u64`. let interval_i64 = i64::try_from(interval_secs) .context("task source interval_secs exceeds SQLite INTEGER range")?; - with_connection(config, |conn| { conn.execute( "INSERT INTO task_sources ( @@ -116,6 +136,9 @@ pub fn add_source( } pub fn get_source(config: &Config, id: &str) -> Result { + if let Some(docs) = super::store_documents::current()? { + return docs.get_source(id); + } with_connection(config, |conn| { let mut stmt = conn.prepare(&format!("{SELECT_SOURCE_COLUMNS} WHERE id = ?1"))?; let mut rows = stmt.query(params![id])?; @@ -128,6 +151,9 @@ pub fn get_source(config: &Config, id: &str) -> Result { } pub fn list_sources(config: &Config) -> Result> { + if let Some(docs) = super::store_documents::current()? { + return docs.list_sources(); + } with_connection(config, |conn| { let mut stmt = conn.prepare(&format!( "{SELECT_SOURCE_COLUMNS} ORDER BY created_at ASC, id ASC" @@ -141,17 +167,12 @@ pub fn list_sources(config: &Config) -> Result> { }) } -/// Apply a partial patch to a task source. +/// Applies `patch` to `source`, as both stores do. /// -/// **Implementation note:** this function opens three separate SQLite -/// connections (read-modify-write + read-back). At settings-panel scale the -/// overhead is acceptable, but there is a theoretical TOCTOU window between -/// the initial `get_source` and the subsequent `UPDATE`. A future refactor -/// could fold all three operations into a single `with_connection` call using -/// a SQL `UPDATE … RETURNING` pattern. -pub fn update_source(config: &Config, id: &str, patch: TaskSourcePatch) -> Result { - let mut source = get_source(config, id)?; - +/// # Errors +/// +/// When the patch's filter is for a different provider than the source. +pub(super) fn apply_patch(source: &mut TaskSource, patch: TaskSourcePatch) -> Result<()> { if let Some(name) = patch.name { source.name = Some(name).filter(|s| !s.trim().is_empty()); } @@ -180,6 +201,24 @@ pub fn update_source(config: &Config, id: &str, patch: TaskSourcePatch) -> Resul if let Some(connection_id) = patch.connection_id { source.connection_id = Some(connection_id).filter(|s| !s.trim().is_empty()); } + Ok(()) +} + +/// Apply a partial patch to a task source. +/// +/// **Implementation note:** this function opens three separate SQLite +/// connections (read-modify-write + read-back). At settings-panel scale the +/// overhead is acceptable, but there is a theoretical TOCTOU window between +/// the initial `get_source` and the subsequent `UPDATE`. A future refactor +/// could fold all three operations into a single `with_connection` call using +/// a SQL `UPDATE … RETURNING` pattern. The document store applies the patch +/// under compare-and-swap, so it has no such window. +pub fn update_source(config: &Config, id: &str, patch: TaskSourcePatch) -> Result { + if let Some(docs) = super::store_documents::current()? { + return docs.update_source(id, patch); + } + let mut source = get_source(config, id)?; + apply_patch(&mut source, patch)?; let filter_json = serde_json::to_string(&source.filter).context("serialize filter")?; let target_json = serde_json::to_string(&source.target).context("serialize target")?; let interval_i64 = i64::try_from(source.interval_secs) @@ -211,6 +250,9 @@ pub fn update_source(config: &Config, id: &str, patch: TaskSourcePatch) -> Resul } pub fn remove_source(config: &Config, id: &str) -> Result<()> { + if let Some(docs) = super::store_documents::current()? { + return docs.remove_source(id); + } let changed = with_connection(config, |conn| { conn.execute("DELETE FROM task_sources WHERE id = ?1", params![id]) .context("Failed to delete task source") @@ -229,6 +271,9 @@ pub fn record_fetch( reason: FetchReason, status: &str, ) -> Result<()> { + if let Some(docs) = super::store_documents::current()? { + return docs.record_fetch(id, finished_at, reason, status); + } let line = format!("{}: {status}", reason.as_str()); with_connection(config, |conn| { conn.execute( @@ -249,6 +294,9 @@ pub fn is_ingested( external_id: &str, hash: &str, ) -> Result { + if let Some(docs) = super::store_documents::current()? { + return docs.is_ingested(source_id, external_id, hash); + } with_connection(config, |conn| { let mut stmt = conn.prepare( "SELECT content_hash FROM ingested_tasks WHERE source_id = ?1 AND external_id = ?2", @@ -270,6 +318,9 @@ pub fn is_ingested( /// todo board, the ledger row itself is the record. The column stays so /// older databases open unchanged. pub fn mark_ingested(config: &Config, source_id: &str, task: &NormalizedTask) -> Result<()> { + if let Some(docs) = super::store_documents::current()? { + return docs.mark_ingested(source_id, task); + } let hash = content_hash(task); let payload = serde_json::to_string(task).context("serialize ingested task payload")?; let now = Utc::now().to_rfc3339(); @@ -294,6 +345,9 @@ pub fn mark_ingested(config: &Config, source_id: &str, task: &NormalizedTask) -> /// content hash. The pipeline uses it to tell an edited upstream task from a /// brand-new one in its logs. pub fn was_ingested(config: &Config, source_id: &str, external_id: &str) -> Result { + if let Some(docs) = super::store_documents::current()? { + return docs.was_ingested(source_id, external_id); + } with_connection(config, |conn| { let mut stmt = conn.prepare("SELECT 1 FROM ingested_tasks WHERE source_id = ?1 AND external_id = ?2")?; @@ -305,6 +359,9 @@ pub fn was_ingested(config: &Config, source_id: &str, external_id: &str) -> Resu /// Return ingested task ids for one source. Used by reconciliation to prune /// ledger rows that no longer match the upstream source/filter. pub fn list_ingested_refs(config: &Config, source_id: &str) -> Result> { + if let Some(docs) = super::store_documents::current()? { + return docs.list_ingested_refs(source_id); + } with_connection(config, |conn| { let mut stmt = conn.prepare( "SELECT external_id FROM ingested_tasks @@ -326,6 +383,9 @@ pub fn list_ingested_refs(config: &Config, source_id: &str) -> Result Result { + if let Some(docs) = super::store_documents::current()? { + return docs.remove_ingested(source_id, external_id); + } let changed = with_connection(config, |conn| { conn.execute( "DELETE FROM ingested_tasks WHERE source_id = ?1 AND external_id = ?2", @@ -342,6 +402,9 @@ pub fn list_ingested( source_id: &str, limit: usize, ) -> Result> { + if let Some(docs) = super::store_documents::current()? { + return docs.list_ingested(source_id, limit); + } // Floor of 1: a caller passing `limit = 0` still gets at least one row // rather than a confusing empty result; `unwrap_or(50)` is the fallback // in the unlikely event that `limit` exceeds `i64::MAX`. @@ -367,6 +430,9 @@ pub fn list_ingested( /// Delete every task source (+ cascade ingested rows). Used by the E2E /// `test_reset` RPC. pub fn clear_all(config: &Config) -> Result { + if let Some(docs) = super::store_documents::current()? { + return docs.clear_all(); + } with_connection(config, |conn| { let removed = conn .execute("DELETE FROM task_sources", params![]) @@ -376,55 +442,8 @@ pub fn clear_all(config: &Config) -> Result { }) } -const SELECT_SOURCE_COLUMNS: &str = "SELECT id, provider, connection_id, name, enabled, filter, \ - interval_secs, target, max_tasks_per_fetch, created_at, last_fetch_at, last_status \ - FROM task_sources"; - -fn map_source_row(row: &rusqlite::Row<'_>) -> rusqlite::Result { - let provider_raw: String = row.get(1)?; - let provider = ProviderSlug::parse(&provider_raw).map_err(sql_conv)?; - - let filter_raw: String = row.get(5)?; - let filter: FilterSpec = serde_json::from_str(&filter_raw) - .map_err(|e| sql_conv(format!("invalid filter json: {e}")))?; - - let target_raw: String = row.get(7)?; - let target: SourceTarget = serde_json::from_str(&target_raw) - .map_err(|e| sql_conv(format!("invalid target json: {e}")))?; - - let created_at_raw: String = row.get(9)?; - let last_fetch_raw: Option = row.get(10)?; - - Ok(TaskSource { - id: row.get(0)?, - provider, - connection_id: row.get(2)?, - name: row.get(3)?, - enabled: row.get::<_, i64>(4)? != 0, - filter, - interval_secs: u64::try_from(row.get::<_, i64>(6)?) - .map_err(|_| sql_conv("invalid negative interval_secs in task_sources DB"))?, - target, - max_tasks_per_fetch: u32::try_from(row.get::<_, i64>(8)?) - .map_err(|_| sql_conv("invalid max_tasks_per_fetch in task_sources DB"))?, - created_at: parse_rfc3339(&created_at_raw).map_err(sql_conv)?, - last_fetch_at: match last_fetch_raw { - Some(raw) => Some(parse_rfc3339(&raw).map_err(sql_conv)?), - None => None, - }, - last_status: row.get(11)?, - }) -} - -fn parse_rfc3339(raw: &str) -> Result> { - let parsed = DateTime::parse_from_rfc3339(raw) - .with_context(|| format!("Invalid RFC3339 timestamp in task_sources DB: {raw}"))?; - Ok(parsed.with_timezone(&Utc)) -} - -fn sql_conv(err: E) -> rusqlite::Error { - rusqlite::Error::ToSqlConversionFailure(anyhow::anyhow!("{err}").into()) -} +mod store_rows; +use store_rows::{map_source_row, SELECT_SOURCE_COLUMNS}; /// Tracks which task_sources database files have already had their schema DDL /// (the `CREATE TABLE`/`CREATE INDEX` batch plus the `add_column_if_missing` diff --git a/crates/openhuman-core/src/integrations/task_sources/store/store_rows.rs b/crates/openhuman-core/src/integrations/task_sources/store/store_rows.rs new file mode 100644 index 00000000000..a6f43d97600 --- /dev/null +++ b/crates/openhuman-core/src/integrations/task_sources/store/store_rows.rs @@ -0,0 +1,57 @@ +//! Row conversion for the SQLite task-source store. + +use anyhow::{Context, Result}; +use chrono::{DateTime, Utc}; + +use super::super::types::{FilterSpec, ProviderSlug, SourceTarget, TaskSource}; + +pub(super) const SELECT_SOURCE_COLUMNS: &str = + "SELECT id, provider, connection_id, name, enabled, filter, \ + interval_secs, target, max_tasks_per_fetch, created_at, last_fetch_at, last_status \ + FROM task_sources"; + +pub(super) fn map_source_row(row: &rusqlite::Row<'_>) -> rusqlite::Result { + let provider_raw: String = row.get(1)?; + let provider = ProviderSlug::parse(&provider_raw).map_err(sql_conv)?; + + let filter_raw: String = row.get(5)?; + let filter: FilterSpec = serde_json::from_str(&filter_raw) + .map_err(|e| sql_conv(format!("invalid filter json: {e}")))?; + + let target_raw: String = row.get(7)?; + let target: SourceTarget = serde_json::from_str(&target_raw) + .map_err(|e| sql_conv(format!("invalid target json: {e}")))?; + + let created_at_raw: String = row.get(9)?; + let last_fetch_raw: Option = row.get(10)?; + + Ok(TaskSource { + id: row.get(0)?, + provider, + connection_id: row.get(2)?, + name: row.get(3)?, + enabled: row.get::<_, i64>(4)? != 0, + filter, + interval_secs: u64::try_from(row.get::<_, i64>(6)?) + .map_err(|_| sql_conv("invalid negative interval_secs in task_sources DB"))?, + target, + max_tasks_per_fetch: u32::try_from(row.get::<_, i64>(8)?) + .map_err(|_| sql_conv("invalid max_tasks_per_fetch in task_sources DB"))?, + created_at: parse_rfc3339(&created_at_raw).map_err(sql_conv)?, + last_fetch_at: match last_fetch_raw { + Some(raw) => Some(parse_rfc3339(&raw).map_err(sql_conv)?), + None => None, + }, + last_status: row.get(11)?, + }) +} + +fn parse_rfc3339(raw: &str) -> Result> { + let parsed = DateTime::parse_from_rfc3339(raw) + .with_context(|| format!("Invalid RFC3339 timestamp in task_sources DB: {raw}"))?; + Ok(parsed.with_timezone(&Utc)) +} + +fn sql_conv(err: E) -> rusqlite::Error { + rusqlite::Error::ToSqlConversionFailure(anyhow::anyhow!("{err}").into()) +} diff --git a/crates/openhuman-core/src/integrations/task_sources/store_documents.rs b/crates/openhuman-core/src/integrations/task_sources/store_documents.rs new file mode 100644 index 00000000000..ab2f476856b --- /dev/null +++ b/crates/openhuman-core/src/integrations/task_sources/store_documents.rs @@ -0,0 +1,350 @@ +//! Task sources on the `tinystoragedrivers` document port. +//! +//! Used instead of `task_sources/sources.db` when the host configured a +//! storage backend ([`crate::storage`]); the functions in `store.rs` pick it +//! per call. Records live under the current call's storage scope (the acting +//! agent, `local` on a single-user host). +//! +//! # Layout +//! +//! | Collection | Document id | Holds | +//! | --- | --- | --- | +//! | `task_sources` | source id | one configured source | +//! | `ingested_tasks` | `(source_id, external_id)` | one dedup ledger entry | +//! +//! `filter`, `target` and the task `payload` are JSON strings, as in SQLite, +//! so a MongoDB-backed port never meets an arbitrary key. Epoch-millisecond +//! fields (`created_ms`, `ingested_ms`) carry the ordering. Removing a source +//! removes its ledger entries, which the SQL schema did with a cascade. + +use anyhow::{anyhow, Context, Result}; +use chrono::{DateTime, Utc}; +use serde_json::{json, Map, Value}; +use tinystoragedrivers::{CollectionSpec, Filter, IndexSpec, Precondition, Query, Sort, Versioned}; + +use super::store::{apply_patch, content_hash, IngestedTaskRef}; +use super::types::{ + FetchReason, FilterSpec, ProviderSlug, SourceTarget, TaskSource, TaskSourcePatch, +}; +use crate::integrations::composio::providers::NormalizedTask; +use crate::storage::documents::{compare_and_swap, text, Repo}; +use crate::storage::{DocumentStoreExt, StorageError}; + +const SOURCES: &str = "task_sources"; +const INGESTED: &str = "ingested_tasks"; +const DOMAIN: &str = "task_sources::store"; + +fn collections() -> Vec { + vec![ + CollectionSpec::new(SOURCES).index(IndexSpec::new("by_created", ["created_ms"])), + CollectionSpec::new(INGESTED) + .index(IndexSpec::new("by_source", ["source_id", "ingested_ms"])), + ] +} + +/// The document store for this call, when the host configured one. +pub(super) fn current() -> Result> { + Ok(Repo::current(DOMAIN, collections)?.map(Docs)) +} + +/// The ledger id for one `(source, external task)`; the length prefix keeps +/// `("a/b", "c")` and `("a", "b/c")` apart. +fn ingested_id(source_id: &str, external_id: &str) -> String { + format!("{}:{source_id}/{external_id}", source_id.len()) +} + +fn to_doc(source: &TaskSource) -> Result { + let mut doc = Map::new(); + doc.insert("provider".into(), json!(source.provider.as_str())); + doc.insert("enabled".into(), json!(source.enabled)); + doc.insert( + "filter".into(), + json!(serde_json::to_string(&source.filter).context("serialize task source filter")?), + ); + doc.insert("interval_secs".into(), json!(source.interval_secs)); + doc.insert( + "target".into(), + json!(serde_json::to_string(&source.target).context("serialize task source target")?), + ); + doc.insert( + "max_tasks_per_fetch".into(), + json!(source.max_tasks_per_fetch), + ); + doc.insert("created_at".into(), json!(source.created_at.to_rfc3339())); + doc.insert( + "created_ms".into(), + json!(source.created_at.timestamp_millis()), + ); + let optional = [ + ("connection_id", source.connection_id.clone()), + ("name", source.name.clone()), + ( + "last_fetch_at", + source.last_fetch_at.map(|at| at.to_rfc3339()), + ), + ("last_status", source.last_status.clone()), + ]; + for (field, value) in optional { + if let Some(value) = value { + doc.insert(field.into(), json!(value)); + } + } + Ok(Value::Object(doc)) +} + +fn parse_time(raw: &str) -> Result> { + Ok(DateTime::parse_from_rfc3339(raw) + .with_context(|| format!("invalid RFC3339 timestamp in task source: {raw}"))? + .with_timezone(&Utc)) +} + +fn to_source(stored: &Versioned) -> Result { + let doc = &stored.doc; + let field = |name: &str| text(doc, name).ok_or_else(|| anyhow!("task source missing {name}")); + let number = |name: &str| { + doc.get(name) + .and_then(Value::as_u64) + .ok_or_else(|| anyhow!("task source missing {name}")) + }; + Ok(TaskSource { + id: stored.id.clone(), + provider: ProviderSlug::parse(field("provider")?).map_err(|error| anyhow!(error))?, + connection_id: text(doc, "connection_id").map(str::to_string), + name: text(doc, "name").map(str::to_string), + enabled: doc.get("enabled").and_then(Value::as_bool).unwrap_or(true), + filter: serde_json::from_str::(field("filter")?) + .context("invalid filter json")?, + interval_secs: number("interval_secs")?, + target: serde_json::from_str::(field("target")?) + .context("invalid target json")?, + max_tasks_per_fetch: u32::try_from(number("max_tasks_per_fetch")?) + .context("invalid max_tasks_per_fetch")?, + created_at: parse_time(field("created_at")?)?, + last_fetch_at: text(doc, "last_fetch_at").map(parse_time).transpose()?, + last_status: text(doc, "last_status").map(str::to_string), + }) +} + +fn not_found(id: &str) -> anyhow::Error { + anyhow!("Task source '{id}' not found") +} + +/// The task-source store over one scoped document handle. +#[derive(Clone)] +pub(super) struct Docs(Repo); + +impl Docs { + #[cfg(test)] + pub(super) fn over(scoped: &crate::storage::ScopedStorage) -> Self { + Self(Repo::over(scoped, DOMAIN, collections)) + } + + pub(super) fn add_source(&self, source: &TaskSource) -> Result { + let id = source.id.clone(); + let doc = to_doc(source)?; + self.0.run(|docs| async move { + docs.put(SOURCES, &id, doc, Precondition::Absent) + .await + .map(|_| ()) + })?; + self.get_source(&source.id) + } + + pub(super) fn get_source(&self, id: &str) -> Result { + let key = id.to_string(); + let stored = self + .0 + .run(|docs| async move { docs.get(SOURCES, &key).await })? + .ok_or_else(|| not_found(id))?; + to_source(&stored) + } + + pub(super) fn list_sources(&self) -> Result> { + let stored = self.0.run(|docs| async move { + let query = Query::all() + .sort(Sort::asc("created_ms")) + .sort(Sort::asc("_id")); + docs.query_all(SOURCES, &query).await + })?; + stored.iter().map(to_source).collect() + } + + pub(super) fn update_source(&self, id: &str, patch: TaskSourcePatch) -> Result { + // Validate against the current source first, so a mismatched filter + // is reported as the SQL store reports it. + apply_patch(&mut self.get_source(id)?, patch.clone())?; + let key = id.to_string(); + let updated = self.0.run(|docs| async move { + compare_and_swap(&docs, SOURCES, &key, |doc| { + let stored = Versioned { + id: key.clone(), + version: tinystoragedrivers::Version::FIRST, + doc: doc.clone(), + }; + let mut source = to_source(&stored).ok()?; + apply_patch(&mut source, patch.clone()).ok()?; + to_doc(&source).ok() + }) + .await + })?; + to_source(&updated.ok_or_else(|| not_found(id))?) + } + + pub(super) fn remove_source(&self, id: &str) -> Result<()> { + let key = id.to_string(); + let removed = self.0.run(|docs| async move { + // Ledger first: if the source delete then fails, the source is + // still there (and merely re-ingests), instead of a ledger with no + // owner that a later source with the same id would inherit. + docs.delete_where(INGESTED, &Filter::eq("source_id", key.clone())) + .await?; + docs.delete(SOURCES, &key, Precondition::None).await + })?; + if removed { + Ok(()) + } else { + Err(not_found(id)) + } + } + + pub(super) fn record_fetch( + &self, + id: &str, + finished_at: DateTime, + reason: FetchReason, + status: &str, + ) -> Result<()> { + let key = id.to_string(); + let line = format!("{}: {status}", reason.as_str()); + let finished = finished_at.to_rfc3339(); + self.0.run(|docs| async move { + compare_and_swap(&docs, SOURCES, &key, |doc| { + let mut next = doc.clone(); + next["last_fetch_at"] = json!(finished); + next["last_status"] = json!(line); + Some(next) + }) + .await + .map(|_| ()) + }) + } + + fn ledger_entry(&self, source_id: &str, external_id: &str) -> Result>> { + let key = ingested_id(source_id, external_id); + self.0 + .run(|docs| async move { docs.get(INGESTED, &key).await }) + } + + pub(super) fn is_ingested( + &self, + source_id: &str, + external_id: &str, + hash: &str, + ) -> Result { + Ok(self + .ledger_entry(source_id, external_id)? + .is_some_and(|stored| text(&stored.doc, "content_hash") == Some(hash))) + } + + pub(super) fn was_ingested(&self, source_id: &str, external_id: &str) -> Result { + Ok(self.ledger_entry(source_id, external_id)?.is_some()) + } + + pub(super) fn mark_ingested(&self, source_id: &str, task: &NormalizedTask) -> Result<()> { + self.mark_ingested_at(source_id, task, Utc::now()) + } + + /// [`Self::mark_ingested`] with the ingestion time supplied. + fn mark_ingested_at( + &self, + source_id: &str, + task: &NormalizedTask, + now: DateTime, + ) -> Result<()> { + // The SQL ledger has a foreign key to the source; keep that parent + // check so a fetch racing a removal does not leave an orphan entry. + self.get_source(source_id)?; + let key = ingested_id(source_id, &task.external_id); + let doc = json!({ + "source_id": source_id, + "external_id": task.external_id, + "content_hash": content_hash(task), + "title": task.title, + "payload": serde_json::to_string(task).context("serialize ingested task payload")?, + "ingested_at": now.to_rfc3339(), + "ingested_ms": now.timestamp_millis(), + }); + self.0.run(|docs| async move { + docs.put(INGESTED, &key, doc, Precondition::None) + .await + .map(|_| ()) + }) + } + + pub(super) fn list_ingested_refs(&self, source_id: &str) -> Result> { + let query = Query::filter(Filter::eq("source_id", source_id)) + .sort(Sort::asc("ingested_ms")) + .sort(Sort::asc("external_id")); + self.0.run(|docs| async move { + Ok(docs + .query_all(INGESTED, &query) + .await? + .iter() + .filter_map(|stored| { + Some(IngestedTaskRef { + external_id: text(&stored.doc, "external_id")?.to_string(), + }) + }) + .collect()) + }) + } + + pub(super) fn remove_ingested(&self, source_id: &str, external_id: &str) -> Result { + let key = ingested_id(source_id, external_id); + self.0 + .run(|docs| async move { docs.delete(INGESTED, &key, Precondition::None).await }) + } + + pub(super) fn list_ingested( + &self, + source_id: &str, + limit: usize, + ) -> Result> { + let query = + Query::filter(Filter::eq("source_id", source_id).and(Filter::exists("payload", true))) + .sort(Sort::desc("ingested_ms")) + .limit(limit.max(1)); + self.0.run(|docs| async move { + let page = docs.query(INGESTED, &query).await?; + page.items + .iter() + .map(|stored| { + let raw = text(&stored.doc, "payload").ok_or_else(|| { + StorageError::serialization(format!( + "ingested task {} has no payload", + stored.id + )) + })?; + serde_json::from_str(raw).map_err(|error| { + StorageError::serialization(format!( + "ingested task {} payload is not valid: {error}", + stored.id + )) + }) + }) + .collect::, StorageError>>() + }) + } + + pub(super) fn clear_all(&self) -> Result { + self.0.run(|docs| async move { + let removed = docs.delete_where(SOURCES, &Filter::All).await?; + docs.delete_where(INGESTED, &Filter::All).await?; + Ok(usize::try_from(removed).unwrap_or(usize::MAX)) + }) + } +} + +#[cfg(test)] +#[path = "store_documents_tests.rs"] +mod tests; diff --git a/crates/openhuman-core/src/integrations/task_sources/store_documents_tests.rs b/crates/openhuman-core/src/integrations/task_sources/store_documents_tests.rs new file mode 100644 index 00000000000..55dcb6f5731 --- /dev/null +++ b/crates/openhuman-core/src/integrations/task_sources/store_documents_tests.rs @@ -0,0 +1,263 @@ +use super::*; +use crate::storage::{MemoryStorage, Scope, StorageBackend}; + +fn docs_in(storage: &MemoryStorage, scope: &str) -> Docs { + Docs::over(&storage.for_scope(&Scope::new(scope).unwrap()).unwrap()) +} + +fn docs() -> Docs { + docs_in(&MemoryStorage::new(), "local") +} + +fn github_filter() -> FilterSpec { + FilterSpec::Github { + repo: Some("tinyhumansai/openhuman".into()), + labels: vec!["bug".into()], + assignee_is_me: true, + state: Some("open".into()), + fetch_mode: Default::default(), + extra: json!({}), + } +} + +fn source(id: &str) -> TaskSource { + TaskSource { + id: id.to_string(), + provider: ProviderSlug::Github, + connection_id: None, + name: Some("My issues".into()), + enabled: true, + filter: github_filter(), + interval_secs: 1800, + target: SourceTarget::AgentTodoProactive, + max_tasks_per_fetch: 25, + created_at: Utc::now(), + last_fetch_at: None, + last_status: None, + } +} + +fn task(external_id: &str, title: &str) -> NormalizedTask { + NormalizedTask { + external_id: external_id.into(), + provider: "github".into(), + title: title.into(), + ..Default::default() + } +} + +#[test] +fn sources_round_trip_oldest_first() { + let store = docs(); + let mut older = source("b"); + older.created_at = Utc::now() - chrono::Duration::seconds(10); + let first = store.add_source(&older).unwrap(); + assert_eq!(first, source_with_time("b", first.created_at)); + store.add_source(&source("a")).unwrap(); + let ids: Vec = store + .list_sources() + .unwrap() + .into_iter() + .map(|s| s.id) + .collect(); + assert_eq!(ids, ["b", "a"]); + assert!( + store.add_source(&source("a")).is_err(), + "an id is added once" + ); + let missing = store.get_source("nope").unwrap_err(); + assert!(missing.to_string().contains("not found")); +} + +fn source_with_time(id: &str, created_at: DateTime) -> TaskSource { + TaskSource { + created_at, + ..source(id) + } +} + +#[test] +fn update_applies_the_patch_and_validates_it() { + let store = docs(); + store.add_source(&source("s")).unwrap(); + let updated = store + .update_source( + "s", + TaskSourcePatch { + name: Some(" ".into()), + enabled: Some(false), + interval_secs: Some(60), + connection_id: Some("conn-1".into()), + ..Default::default() + }, + ) + .unwrap(); + assert!(updated.name.is_none(), "a blank name clears it"); + assert!(!updated.enabled); + assert_eq!(updated.interval_secs, 60); + assert_eq!(updated.connection_id.as_deref(), Some("conn-1")); + assert_eq!(store.get_source("s").unwrap(), updated); + + let notion = FilterSpec::Notion { + database_id: None, + assigned_to_me: false, + status: None, + extra: json!({}), + }; + let error = store + .update_source( + "s", + TaskSourcePatch { + filter: Some(notion), + ..Default::default() + }, + ) + .unwrap_err(); + assert!(error.to_string().contains("does not match"), "{error}"); + assert!(store + .update_source("nope", TaskSourcePatch::default()) + .is_err()); +} + +#[test] +fn record_fetch_stamps_the_source() { + let store = docs(); + store.add_source(&source("s")).unwrap(); + let at = Utc::now(); + store + .record_fetch("s", at, FetchReason::Manual, "ok: 3 new") + .unwrap(); + store + .record_fetch("missing", at, FetchReason::Manual, "ok") + .unwrap(); + let read = store.get_source("s").unwrap(); + assert_eq!(read.last_fetch_at, Some(at)); + assert_eq!( + read.last_status.as_deref(), + Some(format!("{}: ok: 3 new", FetchReason::Manual.as_str()).as_str()) + ); +} + +#[test] +fn the_ingest_ledger_is_edit_aware() { + let store = docs(); + store.add_source(&source("s")).unwrap(); + let original = task("1", "Fix it"); + assert!(!store.was_ingested("s", "1").unwrap()); + store.mark_ingested("s", &original).unwrap(); + assert!(store.was_ingested("s", "1").unwrap()); + assert!(store + .is_ingested("s", "1", &content_hash(&original)) + .unwrap()); + let edited = task("1", "Fix it properly"); + assert!(!store.is_ingested("s", "1", &content_hash(&edited)).unwrap()); + store.mark_ingested("s", &edited).unwrap(); + assert!(store.is_ingested("s", "1", &content_hash(&edited)).unwrap()); + assert_eq!(store.list_ingested("s", 10).unwrap().len(), 1, "an upsert"); +} + +#[test] +fn ledger_lists_refs_oldest_first_and_tasks_newest_first() { + let store = docs(); + store.add_source(&source("s")).unwrap(); + let base = Utc::now(); + store + .mark_ingested_at("s", &task("1", "one"), base - chrono::Duration::seconds(10)) + .unwrap(); + store + .mark_ingested_at("s", &task("2", "two"), base) + .unwrap(); + store.add_source(&source("other")).unwrap(); + store.mark_ingested("other", &task("9", "nine")).unwrap(); + let refs: Vec = store + .list_ingested_refs("s") + .unwrap() + .into_iter() + .map(|r| r.external_id) + .collect(); + assert_eq!(refs, ["1", "2"]); + let titles: Vec = store + .list_ingested("s", 0) + .unwrap() + .into_iter() + .map(|t| t.title) + .collect(); + assert_eq!(titles, ["two"], "a zero limit still returns one"); + assert!(store.remove_ingested("s", "1").unwrap()); + assert!(!store.remove_ingested("s", "1").unwrap()); +} + +#[test] +fn removing_a_source_drops_its_ledger() { + let store = docs(); + store.add_source(&source("s")).unwrap(); + store.add_source(&source("t")).unwrap(); + store.mark_ingested("s", &task("1", "one")).unwrap(); + store.mark_ingested("t", &task("1", "one")).unwrap(); + store.remove_source("s").unwrap(); + assert!(store.remove_source("s").is_err(), "already gone"); + assert!(!store.was_ingested("s", "1").unwrap()); + assert!(store.was_ingested("t", "1").unwrap()); + assert_eq!(store.clear_all().unwrap(), 1); + assert!(store.list_sources().unwrap().is_empty()); + assert!(!store.was_ingested("t", "1").unwrap()); +} + +#[test] +fn a_corrupt_source_is_an_error_not_a_panic() { + let store = docs(); + store + .0 + .run(|docs| async move { + docs.put( + SOURCES, + "bad", + json!({ "provider": "github" }), + Precondition::Absent, + ) + .await + .map(|_| ()) + }) + .unwrap(); + assert!(store.get_source("bad").is_err()); + assert!(store.list_sources().is_err()); +} + +#[test] +fn scopes_keep_sources_apart() { + let storage = MemoryStorage::new(); + let alice = docs_in(&storage, "alice"); + alice.add_source(&source("s")).unwrap(); + alice.mark_ingested("s", &task("1", "one")).unwrap(); + let bob = docs_in(&storage, "bob"); + assert!(bob.list_sources().unwrap().is_empty()); + assert!(!bob.was_ingested("s", "1").unwrap()); + assert_ne!(ingested_id("a/b", "c"), ingested_id("a", "b/c")); +} + +#[test] +fn a_ledger_write_needs_its_source() { + let store = docs(); + assert!(store.mark_ingested("ghost", &task("1", "one")).is_err()); + assert!(!store.was_ingested("ghost", "1").unwrap()); +} + +#[test] +fn a_corrupt_ledger_payload_is_an_error() { + let store = docs(); + store.add_source(&source("s")).unwrap(); + store + .0 + .run(|docs| async move { + docs.put( + INGESTED, + "bad", + json!({ "source_id": "s", "payload": "not json", "ingested_ms": 1 }), + Precondition::Absent, + ) + .await + .map(|_| ()) + }) + .unwrap(); + assert!(store.list_ingested("s", 10).is_err()); +} diff --git a/crates/openhuman-core/src/security/devices/README.md b/crates/openhuman-core/src/security/devices/README.md index be1b061e5e1..da398ea8a39 100644 --- a/crates/openhuman-core/src/security/devices/README.md +++ b/crates/openhuman-core/src/security/devices/README.md @@ -83,6 +83,18 @@ DDL is created idempotently on every connection open (`with_connection`). `peer_ Separately, encrypted X25519 private keys are persisted as `enc2:` strings (via `keyring::SecretStore`, ChaCha20-Poly1305) keyed by `channel_id` in the in-memory `PERSISTED_KEYPAIRS` map, allowing keypair reconstruction for reconnect handshakes. +### On a storage backend + +When the host configured a storage backend (`OPENHUMAN_STORAGE_URL` / +`[storage] url`, see `crate::storage`), every `store` function uses +`store_documents.rs` instead of `devices.db`: the same operations on the +`tinystoragedrivers` document port, under the current call's storage scope +(the acting agent; `local` on a single-user host; refused in SaaS mode with +no acting agent). One `paired_devices` document per `channel_id`. Pairing replaces the +document; touching and revoking are compare-and-swap, so a touch never +revives a device another process revoked. With no backend configured (the desktop default) +`devices.db` is used as described above. + ## Dependencies - `crate::config` (`Config`, `config::rpc::load_config_with_timeout`): workspace paths and config loading for handlers. diff --git a/crates/openhuman-core/src/security/devices/mod.rs b/crates/openhuman-core/src/security/devices/mod.rs index 4f4f3b977e9..e9127e206a7 100644 --- a/crates/openhuman-core/src/security/devices/mod.rs +++ b/crates/openhuman-core/src/security/devices/mod.rs @@ -8,6 +8,7 @@ pub mod crypto; pub mod rpc; pub mod schemas; pub mod store; +mod store_documents; pub mod tunnel_client; pub mod types; diff --git a/crates/openhuman-core/src/security/devices/store.rs b/crates/openhuman-core/src/security/devices/store.rs index 40eb0f37459..e5bba9dbd1a 100644 --- a/crates/openhuman-core/src/security/devices/store.rs +++ b/crates/openhuman-core/src/security/devices/store.rs @@ -1,5 +1,8 @@ //! SQLite persistence for paired devices. //! +//! With a storage backend configured ([`crate::storage`]) every function +//! here is served from the document port instead (`store_documents.rs`). +//! //! Follows the same `with_connection` pattern as `cron/store.rs`: //! open a per-call connection to a domain-scoped `.db` file inside the //! workspace directory, execute DDL on each open (idempotent), then run @@ -24,6 +27,9 @@ pub fn insert_device( device_pubkey: &str, core_session_token_hash: &str, ) -> Result { + if let Some(docs) = super::store_documents::current()? { + return docs.insert_device(channel_id, label, device_pubkey, core_session_token_hash); + } let now = Utc::now().to_rfc3339(); with_connection(config, |conn| { conn.execute( @@ -47,6 +53,9 @@ pub fn insert_device( /// Update `last_seen_at` for a device (called on `tunnel:peer-status` online events). pub fn touch_device(config: &Config, channel_id: &str) -> Result<()> { + if let Some(docs) = super::store_documents::current()? { + return docs.touch_device(channel_id); + } let now = Utc::now().to_rfc3339(); with_connection(config, |conn| { conn.execute( @@ -60,6 +69,9 @@ pub fn touch_device(config: &Config, channel_id: &str) -> Result<()> { /// Mark a device as revoked (soft delete). pub fn revoke_device(config: &Config, channel_id: &str) -> Result { + if let Some(docs) = super::store_documents::current()? { + return docs.revoke_device(channel_id); + } let rows = with_connection(config, |conn| { conn.execute( "UPDATE paired_devices SET revoked = 1 WHERE channel_id = ?1", @@ -72,6 +84,9 @@ pub fn revoke_device(config: &Config, channel_id: &str) -> Result { /// Load a single paired device by channel_id (returns None if not found). pub fn get_device(config: &Config, channel_id: &str) -> Result> { + if let Some(docs) = super::store_documents::current()? { + return docs.get_device(channel_id); + } with_connection(config, |conn| { let mut stmt = conn.prepare( "SELECT channel_id, label, device_pubkey, created_at, last_seen_at, revoked \ @@ -84,6 +99,9 @@ pub fn get_device(config: &Config, channel_id: &str) -> Result Result> { + if let Some(docs) = super::store_documents::current()? { + return docs.list_devices(); + } with_connection(config, |conn| { let mut stmt = conn.prepare( "SELECT channel_id, label, device_pubkey, created_at, last_seen_at, revoked \ diff --git a/crates/openhuman-core/src/security/devices/store_documents.rs b/crates/openhuman-core/src/security/devices/store_documents.rs new file mode 100644 index 00000000000..3f58ba80aab --- /dev/null +++ b/crates/openhuman-core/src/security/devices/store_documents.rs @@ -0,0 +1,134 @@ +//! Paired devices on the `tinystoragedrivers` document port. +//! +//! Used instead of `devices/devices.db` when the host configured a storage +//! backend ([`crate::storage`]); the functions in `store.rs` pick it per call. +//! Devices live under the current call's storage scope (the acting agent, +//! `local` on a single-user host). +//! +//! One `paired_devices` document per `channel_id`, with the SQLite columns as +//! fields. Pairing replaces a document outright (`INSERT OR REPLACE`); +//! touching and revoking are compare-and-swap, so a touch never revives a +//! device another process revoked in between. + +use anyhow::Result; +use chrono::Utc; +use serde_json::{json, Value}; +use tinystoragedrivers::{CollectionSpec, Filter, IndexSpec, Precondition, Query, Sort, Versioned}; + +use crate::security::devices::types::PairedDevice; +use crate::storage::documents::{compare_and_swap, text, Repo}; +use crate::storage::DocumentStoreExt; + +const DEVICES: &str = "paired_devices"; +const DOMAIN: &str = "devices::store"; + +fn collections() -> Vec { + vec![CollectionSpec::new(DEVICES).index(IndexSpec::new("by_active", ["revoked", "created_at"]))] +} + +/// The document store for this call, when the host configured one. +pub(super) fn current() -> Result> { + Ok(Repo::current(DOMAIN, collections)?.map(Docs)) +} + +fn to_device(stored: &Versioned) -> PairedDevice { + let doc = &stored.doc; + PairedDevice { + channel_id: stored.id.clone(), + label: text(doc, "label").unwrap_or_default().to_string(), + device_pubkey: text(doc, "device_pubkey").unwrap_or_default().to_string(), + created_at: text(doc, "created_at").unwrap_or_default().to_string(), + last_seen_at: text(doc, "last_seen_at").map(str::to_string), + peer_online: None, + revoked: doc.get("revoked").and_then(Value::as_bool).unwrap_or(false), + } +} + +/// The paired-device store over one scoped document handle. +#[derive(Clone)] +pub(super) struct Docs(Repo); + +impl Docs { + #[cfg(test)] + pub(super) fn over(scoped: &crate::storage::ScopedStorage) -> Self { + Self(Repo::over(scoped, DOMAIN, collections)) + } + + pub(super) fn insert_device( + &self, + channel_id: &str, + label: &str, + device_pubkey: &str, + core_session_token_hash: &str, + ) -> Result { + let id = channel_id.to_string(); + let doc = json!({ + "label": label, + "device_pubkey": device_pubkey, + "core_session_token_hash": core_session_token_hash, + "created_at": Utc::now().to_rfc3339(), + "last_seen_at": Value::Null, + "revoked": false, + }); + self.0.run(|docs| async move { + let version = docs + .put(DEVICES, &id, doc.clone(), Precondition::None) + .await?; + Ok(to_device(&Versioned { id, version, doc })) + }) + } + + pub(super) fn touch_device(&self, channel_id: &str) -> Result<()> { + let id = channel_id.to_string(); + let now = Utc::now().to_rfc3339(); + self.0.run(|docs| async move { + compare_and_swap(&docs, DEVICES, &id, |doc| { + if doc.get("revoked") == Some(&json!(true)) { + return None; + } + let mut next = doc.clone(); + next["last_seen_at"] = json!(now); + Some(next) + }) + .await + .map(|_| ()) + }) + } + + pub(super) fn revoke_device(&self, channel_id: &str) -> Result { + let id = channel_id.to_string(); + self.0.run(|docs| async move { + // Like the SQL `UPDATE`, revoking an already revoked device still + // reports that the device exists. + let revoked = compare_and_swap(&docs, DEVICES, &id, |doc| { + let mut next = doc.clone(); + next["revoked"] = json!(true); + Some(next) + }) + .await?; + Ok(revoked.is_some()) + }) + } + + pub(super) fn get_device(&self, channel_id: &str) -> Result> { + let id = channel_id.to_string(); + self.0 + .run(|docs| async move { Ok(docs.get(DEVICES, &id).await?.as_ref().map(to_device)) }) + } + + pub(super) fn list_devices(&self) -> Result> { + self.0.run(|docs| async move { + let query = Query::filter(Filter::eq("revoked", false)).sort(Sort::asc("created_at")); + Ok(docs + .query_all(DEVICES, &query) + .await? + .iter() + .map(to_device) + .collect()) + }) + } +} + +#[cfg(test)] +#[path = "store_documents_tests.rs"] +mod tests; diff --git a/crates/openhuman-core/src/security/devices/store_documents_tests.rs b/crates/openhuman-core/src/security/devices/store_documents_tests.rs new file mode 100644 index 00000000000..db486efcb8d --- /dev/null +++ b/crates/openhuman-core/src/security/devices/store_documents_tests.rs @@ -0,0 +1,94 @@ +use super::*; +use crate::storage::{MemoryStorage, Scope, StorageBackend}; + +fn docs_in(storage: &MemoryStorage, scope: &str) -> Docs { + Docs::over(&storage.for_scope(&Scope::new(scope).unwrap()).unwrap()) +} + +fn docs() -> Docs { + docs_in(&MemoryStorage::new(), "local") +} + +#[test] +fn a_paired_device_round_trips() { + let store = docs(); + let device = store.insert_device("ch-1", "iPhone", "pk", "hash").unwrap(); + assert_eq!(device.channel_id, "ch-1"); + assert_eq!(device.label, "iPhone"); + assert!(!device.revoked); + assert!(device.last_seen_at.is_none()); + let read = store.get_device("ch-1").unwrap().expect("stored"); + assert_eq!(read.device_pubkey, "pk"); + assert_eq!(read.created_at, device.created_at); + assert!(store.get_device("missing").unwrap().is_none()); +} + +#[test] +fn pairing_again_replaces_the_device() { + let store = docs(); + store.insert_device("ch-1", "old", "pk1", "h").unwrap(); + store.revoke_device("ch-1").unwrap(); + let again = store.insert_device("ch-1", "new", "pk2", "h").unwrap(); + assert!(!again.revoked, "a re-pair clears the revocation"); + assert_eq!(store.list_devices().unwrap()[0].label, "new"); +} + +#[test] +fn touch_marks_live_devices_only() { + let store = docs(); + store.insert_device("live", "a", "pk", "h").unwrap(); + store.insert_device("gone", "b", "pk", "h").unwrap(); + store.revoke_device("gone").unwrap(); + store.touch_device("live").unwrap(); + store.touch_device("gone").unwrap(); + store.touch_device("missing").unwrap(); + assert!(store + .get_device("live") + .unwrap() + .unwrap() + .last_seen_at + .is_some()); + assert!(store + .get_device("gone") + .unwrap() + .unwrap() + .last_seen_at + .is_none()); +} + +#[test] +fn revoke_hides_the_device_and_reports_existence() { + let store = docs(); + store.insert_device("ch-1", "a", "pk", "h").unwrap(); + assert!(store.revoke_device("ch-1").unwrap()); + assert!(store.revoke_device("ch-1").unwrap(), "still exists"); + assert!(!store.revoke_device("missing").unwrap()); + assert!(store.list_devices().unwrap().is_empty()); + assert!(store.get_device("ch-1").unwrap().unwrap().revoked); +} + +#[test] +fn devices_list_oldest_first() { + let store = docs(); + store.insert_device("first", "a", "pk", "h").unwrap(); + std::thread::sleep(std::time::Duration::from_millis(5)); + store.insert_device("second", "b", "pk", "h").unwrap(); + let ids: Vec = store + .list_devices() + .unwrap() + .into_iter() + .map(|device| device.channel_id) + .collect(); + assert_eq!(ids, ["first", "second"]); +} + +#[test] +fn scopes_keep_devices_apart() { + let storage = MemoryStorage::new(); + docs_in(&storage, "alice") + .insert_device("ch-1", "a", "pk", "h") + .unwrap(); + let bob = docs_in(&storage, "bob"); + assert!(bob.list_devices().unwrap().is_empty()); + assert!(!bob.revoke_device("ch-1").unwrap()); +} diff --git a/crates/openhuman-core/src/storage/README.md b/crates/openhuman-core/src/storage/README.md index 5a98b3884af..af7526a7034 100644 --- a/crates/openhuman-core/src/storage/README.md +++ b/crates/openhuman-core/src/storage/README.md @@ -32,12 +32,26 @@ on `storage-mongodb`. - `driver_is_shared(driver)` / `installed_is_shared()`: whether other processes may write the same backend (MongoDB). Boot-time recovery, such as the orphaned-run sweep, is skipped on a shared backend. +- `current_scope()` / `current_scoped()`: the acting agent's scope (`local` + on a single-user host; an error in SaaS mode with no acting agent) and the + installed backend under it. +- `block_on(future)`: runs a storage future from synchronous store code on + one shared runtime thread. +- `documents::Repo` and `documents::compare_and_swap`: the base the domain + stores build on. A `Repo` holds one domain's scoped document handle, + declares its collections and runs each call; `compare_and_swap` is the + guarded-`UPDATE` loop. ## Consumers - The session store: `openhuman_rpc::session_store::install_for_host` opens the configured backend before boot and installs `DriverSessionStores` over it. See that module's README. +- Domain stores that switch to the document port when a backend is + installed, each in a `store_documents.rs` beside its SQLite `store.rs`: + approvals (`security::approval`), paired devices (`security::devices`), + notifications (`desktop::notifications`) and task sources + (`integrations::task_sources`). ## Boundaries diff --git a/crates/openhuman-core/src/storage/documents.rs b/crates/openhuman-core/src/storage/documents.rs new file mode 100644 index 00000000000..2343d6d901a --- /dev/null +++ b/crates/openhuman-core/src/storage/documents.rs @@ -0,0 +1,135 @@ +//! A small base for domain stores on the document port. +//! +//! Most core stores have a synchronous API and a fixed set of collections. +//! [`Repo`] holds one scoped document handle, declares the domain's +//! collections before each call, and runs the call on the shared blocking +//! bridge ([`super::block_on`]). [`compare_and_swap`] is the read, change, +//! conditional-write loop every guarded SQL `UPDATE … WHERE` becomes, so two +//! processes on one database never both apply a change that requires a +//! particular prior state. + +use std::future::Future; +use std::sync::Arc; + +use anyhow::{anyhow, Context, Result}; +use serde_json::Value; +use tinystoragedrivers::{CollectionSpec, ErrorKind, Versioned}; + +use super::{block_on, current_scoped, DocumentStore, ScopedStorage, StorageError}; + +/// Compare-and-swap attempts before a contended update gives up. +pub const CAS_ATTEMPTS: usize = 32; + +/// One domain's document store under one storage scope. +#[derive(Clone)] +pub struct Repo { + docs: Arc, + domain: &'static str, + collections: fn() -> Vec, +} + +impl Repo { + /// The repo for this call when the host configured a backend, scoped to + /// the acting agent; `None` keeps the domain on its classic store. + /// + /// # Errors + /// + /// When the scope cannot be resolved — in SaaS mode with no acting agent + /// — so the call fails instead of reading a shared bucket. + pub fn current( + domain: &'static str, + collections: fn() -> Vec, + ) -> Result> { + Ok(current_scoped() + .with_context(|| format!("[{domain}] resolve the storage scope"))? + .map(|scoped| Self::over(&scoped, domain, collections))) + } + + /// The repo over an already scoped handle (tests, explicit scopes). + pub fn over( + scoped: &ScopedStorage, + domain: &'static str, + collections: fn() -> Vec, + ) -> Self { + Self { + docs: Arc::clone(scoped.documents()), + domain, + collections, + } + } + + /// Runs `op` against the store from synchronous code, after declaring the + /// domain's collections (idempotent). + /// + /// # Errors + /// + /// Any storage error, prefixed with the domain name. + pub fn run(&self, op: F) -> Result + where + F: FnOnce(Arc) -> Fut, + Fut: Future> + Send + 'static, + T: Send + 'static, + { + let docs = Arc::clone(&self.docs); + let specs = (self.collections)(); + let future = op(Arc::clone(&docs)); + let domain = self.domain; + block_on(async move { + for spec in &specs { + docs.ensure_collection(spec).await?; + } + future.await + }) + .map_err(|error| anyhow!("[{domain}] storage: {error}")) + } +} + +/// Applies `change` to document `id` in `collection` under compare-and-swap +/// and returns what was stored, or `None` when the document is missing or +/// `change` declines (it is no longer in the state the change requires). +/// +/// # Errors +/// +/// A storage error, or a conflict when the document kept changing for +/// [`CAS_ATTEMPTS`] attempts. +pub async fn compare_and_swap( + docs: &Arc, + collection: &str, + id: &str, + change: impl Fn(&Value) -> Option, +) -> Result>, StorageError> { + for _ in 0..CAS_ATTEMPTS { + let Some(stored) = docs.get(collection, id).await? else { + return Ok(None); + }; + let Some(next) = change(&stored.doc) else { + return Ok(None); + }; + match docs + .put(collection, id, next.clone(), stored.unchanged()) + .await + { + Ok(version) => { + return Ok(Some(Versioned { + id: id.to_string(), + version, + doc: next, + })); + } + Err(error) if error.kind() == ErrorKind::Conflict => {} + Err(error) => return Err(error), + } + } + Err(StorageError::conflict(format!( + "{collection}/{id} kept changing under {CAS_ATTEMPTS} attempts" + ))) +} + +/// The string field `field` of `doc`, when present. +pub fn text<'a>(doc: &'a Value, field: &str) -> Option<&'a str> { + doc.get(field).and_then(Value::as_str) +} + +#[cfg(test)] +#[path = "documents_tests.rs"] +mod tests; diff --git a/crates/openhuman-core/src/storage/documents_tests.rs b/crates/openhuman-core/src/storage/documents_tests.rs new file mode 100644 index 00000000000..25e8c2b401c --- /dev/null +++ b/crates/openhuman-core/src/storage/documents_tests.rs @@ -0,0 +1,129 @@ +use super::*; +use serde_json::json; +use tinystoragedrivers::{IndexSpec, Precondition}; + +use crate::storage::{MemoryStorage, Scope, StorageBackend}; + +const THINGS: &str = "things"; + +fn specs() -> Vec { + vec![CollectionSpec::new(THINGS).index(IndexSpec::new("by_state", ["state"]))] +} + +fn repo(storage: &MemoryStorage, scope: &str) -> Repo { + let scoped = storage.for_scope(&Scope::new(scope).unwrap()).unwrap(); + Repo::over(&scoped, "test", specs) +} + +#[test] +fn run_declares_the_collections_and_returns_the_result() { + let storage = MemoryStorage::new(); + let repo = repo(&storage, "local"); + repo.run(|docs| async move { + docs.put( + THINGS, + "a", + json!({ "state": "open" }), + Precondition::Absent, + ) + .await + .map(|_| ()) + }) + .unwrap(); + let state = repo + .run(|docs| async move { docs.get(THINGS, "a").await }) + .unwrap() + .map(|stored| text(&stored.doc, "state").map(str::to_string)); + assert_eq!(state, Some(Some("open".to_string()))); +} + +#[test] +fn storage_errors_name_the_domain() { + let storage = MemoryStorage::new(); + let error = repo(&storage, "local") + .run(|_| async { Err::<(), _>(StorageError::conflict("raced")) }) + .unwrap_err(); + assert!(error.to_string().starts_with("[test] storage:"), "{error}"); +} + +#[test] +fn compare_and_swap_applies_declines_and_skips_missing() { + let storage = MemoryStorage::new(); + let repo = repo(&storage, "local"); + let (closed, declined, missing) = repo + .run(|docs| async move { + docs.put( + THINGS, + "a", + json!({ "state": "open" }), + Precondition::Absent, + ) + .await?; + let close = |doc: &Value| { + (text(doc, "state") == Some("open")).then(|| json!({ "state": "closed" })) + }; + let closed = compare_and_swap(&docs, THINGS, "a", close).await?; + let declined = compare_and_swap(&docs, THINGS, "a", close).await?; + let missing = compare_and_swap(&docs, THINGS, "nope", close).await?; + Ok((closed, declined, missing)) + }) + .unwrap(); + assert_eq!( + closed.map(|stored| stored.doc), + Some(json!({ "state": "closed" })) + ); + assert!(declined.is_none(), "already closed"); + assert!(missing.is_none()); +} + +#[test] +fn concurrent_swaps_apply_once() { + let storage = MemoryStorage::new(); + let repo = repo(&storage, "local"); + repo.run(|docs| async move { + docs.put( + THINGS, + "a", + json!({ "state": "open" }), + Precondition::Absent, + ) + .await + .map(|_| ()) + }) + .unwrap(); + let winners: usize = (0..8) + .map(|_| { + let repo = repo.clone(); + std::thread::spawn(move || { + repo.run(|docs| async move { + compare_and_swap(&docs, THINGS, "a", |doc| { + (text(doc, "state") == Some("open")).then(|| json!({ "state": "closed" })) + }) + .await + }) + .unwrap() + .is_some() + }) + }) + .collect::>() + .into_iter() + .map(|handle| usize::from(handle.join().unwrap())) + .sum(); + assert_eq!(winners, 1); +} + +#[test] +fn scopes_do_not_see_each_other() { + let storage = MemoryStorage::new(); + repo(&storage, "alice") + .run(|docs| async move { + docs.put(THINGS, "a", json!({}), Precondition::Absent) + .await + .map(|_| ()) + }) + .unwrap(); + let seen = repo(&storage, "bob") + .run(|docs| async move { docs.get(THINGS, "a").await }) + .unwrap(); + assert!(seen.is_none()); +} diff --git a/crates/openhuman-core/src/storage/mod.rs b/crates/openhuman-core/src/storage/mod.rs index c4299297bc5..52109954df1 100644 --- a/crates/openhuman-core/src/storage/mod.rs +++ b/crates/openhuman-core/src/storage/mod.rs @@ -18,6 +18,8 @@ //! driver the build does not carry fails at [`open`] naming the feature, so a //! misconfigured deployment stops at boot instead of at its first write. +pub mod documents; + use std::future::Future; use std::sync::{Arc, LazyLock, OnceLock, RwLock}; diff --git a/tests/storage_domains_e2e.rs b/tests/storage_domains_e2e.rs new file mode 100644 index 00000000000..388e449980d --- /dev/null +++ b/tests/storage_domains_e2e.rs @@ -0,0 +1,97 @@ +//! Paired devices, notifications and task sources on a configured storage +//! backend, end to end through their public store functions. +//! +//! Its own test binary because it installs a backend into the process-wide +//! storage slot, which would reroute every other suite's stores in a shared +//! process. One test, so nothing in this binary races the slot either. + +use std::sync::Arc; + +use chrono::Utc; +use openhuman_core::config::Config; +use openhuman_core::desktop::notifications::store as notifications; +use openhuman_core::desktop::notifications::types::{IntegrationNotification, NotificationStatus}; +use openhuman_core::integrations::composio::providers::NormalizedTask; +use openhuman_core::integrations::task_sources::store as task_sources; +use openhuman_core::integrations::task_sources::types::{FilterSpec, ProviderSlug, SourceTarget}; +use openhuman_core::security::devices::store as devices; + +fn notification(id: &str) -> IntegrationNotification { + IntegrationNotification { + id: id.to_string(), + provider: "slack".to_string(), + account_id: None, + title: "New message".to_string(), + body: "hello".to_string(), + raw_payload: serde_json::json!({}), + importance_score: None, + triage_action: None, + triage_reason: None, + status: NotificationStatus::Unread, + received_at: Utc::now(), + scored_at: None, + } +} + +#[test] +fn a_configured_backend_holds_devices_notifications_and_task_sources() { + let workspace = tempfile::tempdir().unwrap(); + let config = Config { + workspace_dir: workspace.path().to_path_buf(), + ..Config::default() + }; + openhuman_core::storage::install(Arc::new(openhuman_core::storage::MemoryStorage::new())); + + devices::insert_device(&config, "ch-1", "iPhone", "pk", "hash").unwrap(); + assert_eq!(devices::list_devices(&config).unwrap().len(), 1); + assert!(devices::revoke_device(&config, "ch-1").unwrap()); + + assert!(notifications::insert_if_not_recent(&config, ¬ification("n1")).unwrap()); + assert!(!notifications::insert_if_not_recent(&config, ¬ification("n2")).unwrap()); + assert_eq!(notifications::unread_count(&config).unwrap(), 1); + notifications::mark_read(&config, "n1").unwrap(); + assert_eq!(notifications::stats(&config).unwrap().total, 1); + + let filter = FilterSpec::Github { + repo: None, + labels: Vec::new(), + assignee_is_me: true, + state: None, + fetch_mode: Default::default(), + extra: serde_json::json!({}), + }; + let source = task_sources::add_source( + &config, + ProviderSlug::Github, + None, + None, + filter, + 600, + SourceTarget::TodoOnly, + 10, + ) + .unwrap(); + let task = NormalizedTask { + external_id: "1".into(), + title: "Fix it".into(), + ..Default::default() + }; + task_sources::mark_ingested(&config, &source.id, &task).unwrap(); + assert!(task_sources::was_ingested(&config, &source.id, "1").unwrap()); + assert_eq!(task_sources::list_sources(&config).unwrap().len(), 1); + + for db in [ + "devices/devices.db", + "notifications/notifications.db", + "task_sources/sources.db", + ] { + assert!(!workspace.path().join(db).exists(), "{db} was not written"); + } + + // Without a backend the classic databases are back in use. + assert!(openhuman_core::storage::clear()); + assert!(devices::list_devices(&config).unwrap().is_empty()); + assert_eq!(notifications::unread_count(&config).unwrap(), 0); + assert!(task_sources::list_sources(&config).unwrap().is_empty()); + assert!(workspace.path().join("devices/devices.db").exists()); +}