admin/openhuman
1
0
Fork 0
mirror of https://github.com/tinyhumansai/openhuman.git synced 2026-07-27 21:08:00 +00:00
Code Issues Packages Projects Releases Wiki Activity

feat(notifications): bridge CEF providers into core ingest pipeline with dedup (#874)

Browse Source
This commit is contained in:
Mega Mind
2026-04-24 19:18:10 +05:30
committed by GitHub
parent 9ffc61a461
commit 208c276a3e
5 changed files with 195 additions and 48 deletions
Download Patch File Download Diff File Expand all files Collapse all files
app/src-tauri/Cargo.toml
-1
View File
@@ -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"
app/src/lib/webviewNotifications/service.ts
+42
View File
@@ -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. */
src/openhuman/notifications/rpc.rs
+27 -18
View File
@@ -31,10 +31,10 @@ pub async fn handle_ingest(params: Map<String, Value>) -> Result<Value, String>
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<String, Value>) -> Result<Value, String>
scored_at: None,
};
store::insert(&config, &notification)
.map_err(|e| format!("[notifications::rpc] insert failed: {e}"))?;
let inserted = store::insert_if_not_recent(&config, &notification)
.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<String, Value>) -> Result<Value, String>
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<String, Value>) -> Result<Value, String>
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<String, Value>) -> Result<Value, String>
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<String, Value>) -> Result<Value, String>
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<String, Value>) -> Result<Value, String> {
.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<String, Value>) -> Result<Value, Strin
let id = params
.get("id")
.and_then(|v| v.as_str())
.ok_or_else(|| "[notifications::rpc] missing required param 'id'".to_string())?
.ok_or_else(|| "[notification_intel] missing required param 'id'".to_string())?
.to_string();
store::mark_read(&config, &id)
.map_err(|e| format!("[notifications::rpc] mark_read failed: {e}"))?;
.map_err(|e| format!("[notification_intel] mark_read failed: {e}"))?;
tracing::debug!(id = %id, "[notifications::rpc] marked read");
tracing::debug!(id = %id, "[notification_intel] marked read");
let outcome = RpcOutcome::new(json!({ "ok": true }), vec![]);
outcome.into_cli_compatible_json()
@@ -215,9 +224,9 @@ pub async fn handle_settings_get(params: Map<String, Value>) -> Result<Value, St
let provider = params
.get("provider")
.and_then(|v| v.as_str())
.ok_or_else(|| "[notifications::rpc] missing required param 'provider'".to_string())?;
.ok_or_else(|| "[notification_intel] missing required param 'provider'".to_string())?;
let settings = store::get_settings(&config, provider)
.map_err(|e| format!("[notifications::rpc] settings_get failed: {e}"))?;
.map_err(|e| format!("[notification_intel] settings_get failed: {e}"))?;
let outcome = RpcOutcome::new(json!({ "settings": settings }), vec![]);
outcome.into_cli_compatible_json()
}
@@ -226,7 +235,7 @@ pub async fn handle_settings_get(params: Map<String, Value>) -> Result<Value, St
pub async fn handle_settings_set(params: Map<String, Value>) -> Result<Value, String> {
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<String, Value>) -> Result<Value, St
route_to_orchestrator: req.route_to_orchestrator,
};
store::upsert_settings(&config, &clamped)
.map_err(|e| format!("[notifications::rpc] settings_set failed: {e}"))?;
.map_err(|e| format!("[notification_intel] settings_set failed: {e}"))?;
let outcome = RpcOutcome::new(json!({ "ok": true, "settings": clamped }), vec![]);
outcome.into_cli_compatible_json()
}
src/openhuman/notifications/store.rs
+90
View File
@@ -33,6 +33,8 @@ CREATE INDEX IF NOT EXISTS idx_integration_notifications_provider
ON integration_notifications(provider);
CREATE INDEX IF NOT EXISTS idx_integration_notifications_status
ON integration_notifications(status);
CREATE INDEX IF NOT EXISTS idx_integration_notifications_dedup
ON integration_notifications(provider, account_id, title, body, received_at);
CREATE TABLE IF NOT EXISTS notification_settings (
provider TEXT PRIMARY KEY,
@@ -111,6 +113,70 @@ pub fn insert(config: &Config, n: &IntegrationNotification) -> 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<bool> {
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();
tests/webview_apis_bridge.rs
+36 -29
View File
@@ -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,
}
}
});