From 208c276a3e85f0980204b844e59a484a8da5c5f7 Mon Sep 17 00:00:00 2001 From: Mega Mind <146339422+M3gA-Mind@users.noreply.github.com> Date: Fri, 24 Apr 2026 19:18:10 +0530 Subject: [PATCH] feat(notifications): bridge CEF providers into core ingest pipeline with dedup (#874) --- app/src-tauri/Cargo.toml | 1 - app/src/lib/webviewNotifications/service.ts | 42 ++++++++++ src/openhuman/notifications/rpc.rs | 45 ++++++----- src/openhuman/notifications/store.rs | 90 +++++++++++++++++++++ tests/webview_apis_bridge.rs | 65 ++++++++------- 5 files changed, 195 insertions(+), 48 deletions(-) diff --git a/app/src-tauri/Cargo.toml b/app/src-tauri/Cargo.toml index 8e149f9ec..dff5effcb 100644 --- a/app/src-tauri/Cargo.toml +++ b/app/src-tauri/Cargo.toml @@ -69,7 +69,6 @@ semver = "1" log = "0.4" env_logger = "0.11" -base64 = "0.22" # Used by the imessage_scanner module. anyhow = "1.0" diff --git a/app/src/lib/webviewNotifications/service.ts b/app/src/lib/webviewNotifications/service.ts index c23108dc3..6acc52f8c 100644 --- a/app/src/lib/webviewNotifications/service.ts +++ b/app/src/lib/webviewNotifications/service.ts @@ -2,12 +2,14 @@ import { isTauri } from '@tauri-apps/api/core'; import { listen, type UnlistenFn } from '@tauri-apps/api/event'; import debug from 'debug'; +import { ingestNotification } from '../../services/notificationService'; import { store } from '../../store'; import { focusAccountFromNotification, noteWebviewNotificationFired, } from '../../store/accountsSlice'; import { notificationReceived } from '../../store/notificationSlice'; +import { addNotification } from '../../store/notificationsSlice'; import { WEBVIEW_NOTIFICATION_FIRED_EVENT, type WebviewNotificationFired } from './types'; const log = debug('webview-notifications'); @@ -83,6 +85,46 @@ function handleFired(payload: WebviewNotificationFired): void { deepLink: `/accounts/${accountId}`, }) ); + + // Mirror into the core triage pipeline — fire-and-forget. + log( + '[notification_intel] forwarding to core ingest provider=%s account_present=%s', + provider, + accountId ? 'yes' : 'no' + ); + void ingestNotification({ + provider, + account_id: accountId, + title, + body, + raw_payload: { tag, provider, account_id: accountId, title, body }, + }) + .then(result => { + if (!result.skipped) { + log('[notification_intel] ingest created id=%s', result.id); + store.dispatch( + addNotification({ + id: result.id, + provider, + account_id: accountId, + title, + body, + raw_payload: { tag, provider, account_id: accountId, title, body }, + status: 'unread', + received_at: new Date().toISOString(), + importance_score: undefined, + triage_action: undefined, + triage_reason: undefined, + scored_at: undefined, + }) + ); + } else { + log('[notification_intel] ingest skipped reason=%s', result.reason); + } + }) + .catch(err => { + errLog('[notification_intel] ingest failed provider=%s: %O', provider, err); + }); } /** Exposed for tests — resets module singletons between runs. */ diff --git a/src/openhuman/notifications/rpc.rs b/src/openhuman/notifications/rpc.rs index 2a87ac224..db43ad15b 100644 --- a/src/openhuman/notifications/rpc.rs +++ b/src/openhuman/notifications/rpc.rs @@ -31,10 +31,10 @@ pub async fn handle_ingest(params: Map) -> Result let config = config_rpc::load_config_with_timeout().await?; let req: NotificationIngestRequest = serde_json::from_value(Value::Object(params.clone())) - .map_err(|e| format!("[notifications::rpc] invalid ingest params: {e}"))?; + .map_err(|e| format!("[notification_intel] invalid ingest params: {e}"))?; let provider_settings = store::get_settings(&config, &req.provider) - .map_err(|e| format!("[notifications::rpc] get_settings failed: {e}"))?; + .map_err(|e| format!("[notification_intel] get_settings failed: {e}"))?; if !provider_settings.enabled { let outcome = RpcOutcome::new( json!({ "skipped": true, "reason": "provider_disabled" }), @@ -58,13 +58,22 @@ pub async fn handle_ingest(params: Map) -> Result scored_at: None, }; - store::insert(&config, ¬ification) - .map_err(|e| format!("[notifications::rpc] insert failed: {e}"))?; + let inserted = store::insert_if_not_recent(&config, ¬ification) + .map_err(|e| format!("[notification_intel] insert_if_not_recent failed: {e}"))?; + if !inserted { + tracing::debug!( + provider = %req.provider, + title_chars = req.title.chars().count(), + "[notification_intel] skipping duplicate notification" + ); + let outcome = RpcOutcome::new(json!({ "skipped": true, "reason": "duplicate" }), vec![]); + return outcome.into_cli_compatible_json(); + } tracing::debug!( id = %id, provider = %req.provider, - "[notifications::rpc] ingested notification, spawning triage" + "[notification_intel] ingested notification, spawning triage" ); // Spawn background triage — the ingest RPC returns immediately. @@ -101,7 +110,7 @@ pub async fn handle_ingest(params: Map) -> Result id = %id_for_triage, action = %action, score = score, - "[notifications::rpc] triage complete" + "[notification_intel] triage complete" ); if let Err(e) = store::update_triage( @@ -114,7 +123,7 @@ pub async fn handle_ingest(params: Map) -> Result tracing::warn!( id = %id_for_triage, error = %e, - "[notifications::rpc] failed to persist triage result" + "[notification_intel] failed to persist triage result" ); } @@ -127,7 +136,7 @@ pub async fn handle_ingest(params: Map) -> Result tracing::warn!( id = %id_for_triage, error = %e, - "[notifications::rpc] apply_decision failed" + "[notification_intel] apply_decision failed" ); } } @@ -136,7 +145,7 @@ pub async fn handle_ingest(params: Map) -> Result tracing::warn!( id = %id_for_triage, error = %e, - "[notifications::rpc] triage pipeline failed" + "[notification_intel] triage pipeline failed" ); } } @@ -177,10 +186,10 @@ pub async fn handle_list(params: Map) -> Result { .map(|v| v as f32); let items = store::list(&config, limit, offset, provider.as_deref(), min_score) - .map_err(|e| format!("[notifications::rpc] list failed: {e}"))?; + .map_err(|e| format!("[notification_intel] list failed: {e}"))?; let unread = store::unread_count(&config) - .map_err(|e| format!("[notifications::rpc] unread_count failed: {e}"))?; + .map_err(|e| format!("[notification_intel] unread_count failed: {e}"))?; let outcome = RpcOutcome::new(json!({ "items": items, "unread_count": unread }), vec![]); outcome.into_cli_compatible_json() @@ -197,13 +206,13 @@ pub async fn handle_mark_read(params: Map) -> Result) -> Result) -> Result) -> Result { let config = config_rpc::load_config_with_timeout().await?; let req: NotificationSettingsUpsertRequest = serde_json::from_value(Value::Object(params)) - .map_err(|e| format!("[notifications::rpc] invalid settings_set params: {e}"))?; + .map_err(|e| format!("[notification_intel] invalid settings_set params: {e}"))?; let clamped = NotificationSettings { provider: req.provider, enabled: req.enabled, @@ -234,7 +243,7 @@ pub async fn handle_settings_set(params: Map) -> Result Result<()> { }) } +/// Atomically insert a notification unless a matching one arrived recently. +/// +/// Returns `true` when inserted, `false` when skipped as duplicate. +pub fn insert_if_not_recent(config: &Config, n: &IntegrationNotification) -> Result { + with_connection(config, |conn| { + let tx = conn + .unchecked_transaction() + .context("[notifications::store] begin insert_if_not_recent tx failed")?; + + let count: i64 = match n.account_id.as_deref() { + Some(aid) => tx.query_row( + "SELECT COUNT(*) FROM integration_notifications + WHERE provider = ?1 AND account_id = ?2 + AND title = ?3 AND body = ?4 + AND unixepoch(received_at) >= unixepoch('now', '-60 seconds')", + params![&n.provider, aid, &n.title, &n.body], + |row| row.get(0), + ), + None => tx.query_row( + "SELECT COUNT(*) FROM integration_notifications + WHERE provider = ?1 AND account_id IS NULL + AND title = ?2 AND body = ?3 + AND unixepoch(received_at) >= unixepoch('now', '-60 seconds')", + params![&n.provider, &n.title, &n.body], + |row| row.get(0), + ), + } + .context("[notifications::store] insert_if_not_recent dedup query failed")?; + + if count > 0 { + tx.commit() + .context("[notifications::store] commit duplicate tx failed")?; + return Ok(false); + } + + tx.execute( + "INSERT INTO integration_notifications + (id, provider, account_id, title, body, raw_payload, + importance_score, triage_action, triage_reason, status, + received_at, scored_at) + VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, ?11, ?12)", + params![ + n.id, + n.provider, + n.account_id, + n.title, + n.body, + n.raw_payload.to_string(), + n.importance_score, + n.triage_action, + n.triage_reason, + n.status.as_str(), + n.received_at.to_rfc3339(), + n.scored_at.map(|t| t.to_rfc3339()), + ], + ) + .context("[notifications::store] insert_if_not_recent insert failed")?; + + tx.commit() + .context("[notifications::store] commit insert_if_not_recent tx failed")?; + Ok(true) + }) +} + /// List notifications with optional filtering. pub fn list( config: &Config, @@ -434,6 +500,30 @@ mod tests { assert_eq!(gmail[0].provider, "gmail"); } + #[test] + fn insert_if_not_recent_skips_duplicate() { + let dir = TempDir::new().unwrap(); + let config = test_config(&dir); + let n = sample_notification("dup-a", "slack"); + assert!(insert_if_not_recent(&config, &n).unwrap()); + + let n2 = sample_notification("dup-b", "slack"); + assert!(!insert_if_not_recent(&config, &n2).unwrap()); + } + + #[test] + fn insert_if_not_recent_rejects_expired_window_only() { + let dir = TempDir::new().unwrap(); + let config = test_config(&dir); + + let mut old = sample_notification("old1", "slack"); + old.received_at = Utc::now() - chrono::Duration::seconds(120); + insert(&config, &old).unwrap(); + + let fresh_same_content = sample_notification("fresh1", "slack"); + assert!(insert_if_not_recent(&config, &fresh_same_content).unwrap()); + } + #[test] fn settings_roundtrip_defaults_and_upsert() { let dir = TempDir::new().unwrap(); diff --git a/tests/webview_apis_bridge.rs b/tests/webview_apis_bridge.rs index 35f5321f8..f06f6bfe5 100644 --- a/tests/webview_apis_bridge.rs +++ b/tests/webview_apis_bridge.rs @@ -56,35 +56,42 @@ async fn ensure_mock_server() -> u16 { Err(_) => return, }; let (mut sink, mut stream) = ws.split(); - while let Some(Ok(Message::Text(text))) = stream.next().await { - let req: Value = serde_json::from_str(&text).unwrap(); - let id = req["id"].as_str().unwrap().to_string(); - let method = req["method"].as_str().unwrap().to_string(); - let resp = match method.as_str() { - "gmail.list_labels" => json!({ - "kind": "response", - "id": id, - "ok": true, - "result": [ - {"id": "INBOX", "name": "Inbox", "kind": "system", "unread": 3}, - {"id": "Receipts", "name": "Receipts", "kind": "user", "unread": null} - ], - }), - "gmail.trash" => json!({ - "kind": "response", - "id": id, - "ok": false, - "error": "simulated failure from mock bridge", - }), - _ => json!({ - "kind": "response", - "id": id, - "ok": false, - "error": format!("mock bridge: unhandled method '{method}'"), - }), - }; - if sink.send(Message::Text(resp.to_string())).await.is_err() { - break; + while let Some(msg) = stream.next().await { + match msg { + Ok(Message::Text(text)) => { + let req: Value = serde_json::from_str(&text).unwrap(); + let id = req["id"].as_str().unwrap().to_string(); + let method = req["method"].as_str().unwrap().to_string(); + let resp = match method.as_str() { + "gmail.list_labels" => json!({ + "kind": "response", + "id": id, + "ok": true, + "result": [ + {"id": "INBOX", "name": "Inbox", "kind": "system", "unread": 3}, + {"id": "Receipts", "name": "Receipts", "kind": "user", "unread": null} + ], + }), + "gmail.trash" => json!({ + "kind": "response", + "id": id, + "ok": false, + "error": "simulated failure from mock bridge", + }), + _ => json!({ + "kind": "response", + "id": id, + "ok": false, + "error": format!("mock bridge: unhandled method '{method}'"), + }), + }; + if sink.send(Message::Text(resp.to_string())).await.is_err() { + break; + } + } + Ok(Message::Close(_)) => break, + Ok(_) => continue, + Err(_) => break, } } });