From a35ba10bb2e654a9eca3ff2d006c9c1a7447739a Mon Sep 17 00:00:00 2001 From: Mega Mind <146339422+M3gA-Mind@users.noreply.github.com> Date: Thu, 7 May 2026 01:41:52 +0530 Subject: [PATCH] feat(channels): expose WhatsApp Web data to agent via structured RPC API (#1308) Co-authored-by: Cursor --- app/src-tauri/src/whatsapp_scanner/mod.rs | 973 +++++++++++++++++++--- src/core/all.rs | 50 +- src/core/jsonrpc.rs | 9 + src/openhuman/about_app/catalog.rs | 10 + src/openhuman/mod.rs | 1 + src/openhuman/whatsapp_data/global.rs | 57 ++ src/openhuman/whatsapp_data/mod.rs | 29 + src/openhuman/whatsapp_data/ops.rs | 200 +++++ src/openhuman/whatsapp_data/rpc.rs | 122 +++ src/openhuman/whatsapp_data/schemas.rs | 263 ++++++ src/openhuman/whatsapp_data/store.rs | 662 +++++++++++++++ src/openhuman/whatsapp_data/types.rs | 134 +++ tests/json_rpc_e2e.rs | 377 +++++++++ 13 files changed, 2763 insertions(+), 124 deletions(-) create mode 100644 src/openhuman/whatsapp_data/global.rs create mode 100644 src/openhuman/whatsapp_data/mod.rs create mode 100644 src/openhuman/whatsapp_data/ops.rs create mode 100644 src/openhuman/whatsapp_data/rpc.rs create mode 100644 src/openhuman/whatsapp_data/schemas.rs create mode 100644 src/openhuman/whatsapp_data/store.rs create mode 100644 src/openhuman/whatsapp_data/types.rs diff --git a/app/src-tauri/src/whatsapp_scanner/mod.rs b/app/src-tauri/src/whatsapp_scanner/mod.rs index bd2d467cc..d55ac56c8 100644 --- a/app/src-tauri/src/whatsapp_scanner/mod.rs +++ b/app/src-tauri/src/whatsapp_scanner/mod.rs @@ -529,7 +529,17 @@ fn emit_snapshot(app: &AppHandle, account_id: &str, snap: &ScanSn (1, _, _) => Some(exact[0].to_string()), (0, 1, _) => Some(ci[0].to_string()), (0, 0, 1) => Some(substring[0].to_string()), - _ => None, + (e, c, s) => { + let count = e + c + s; + if count > 1 { + log::warn!( + "[whatsapp_scanner] ambiguous active-chat resolution: {} candidates for '{}' — skipping backfill", + count, + name + ); + } + None + } } }); log::info!( @@ -543,113 +553,14 @@ fn emit_snapshot(app: &AppHandle, account_id: &str, snap: &ScanSn // caches decrypted bodies in memory, so IndexedDB gives us metadata and // the DOM gives us text for currently-rendered chats — unioning them // here gives downstream consumers a single message list. - let mut messages = snap.messages.clone(); - if !snap.dom_messages.is_empty() { - use std::collections::{HashMap, HashSet}; - // Index DOM rows by BOTH full `data-id` ("true_chatId_msgId") AND - // bare msgId — IDB's `_serialized` matches data-id but some paths - // use just the msgId. Each map entry keeps its source dataId so a - // patch via either key only consumes the row once. - let mut by_msg_id: HashMap = HashMap::new(); - for dm in &snap.dom_messages { - let did = dm - .get("dataId") - .and_then(|v| v.as_str()) - .unwrap_or("") - .to_string(); - if did.is_empty() { - continue; - } - by_msg_id.insert(did.clone(), (did.clone(), dm.clone())); - if let Some(mid) = dm.get("msgId").and_then(|v| v.as_str()) { - by_msg_id - .entry(mid.to_string()) - .or_insert_with(|| (did.clone(), dm.clone())); - } - } - let mut consumed: HashSet = HashSet::new(); - let mut patched = 0usize; - for m in messages.iter_mut() { - let mid_opt = m.get("id").and_then(|v| v.as_str()).map(|s| s.to_string()); - let has_body = m - .get("body") - .and_then(|v| v.as_str()) - .map(|s| !s.is_empty()) - .unwrap_or(false); - if has_body { - continue; - } - if let Some(mid) = mid_opt { - // Try the full IDB id first (legacy compound `"true_chat_msg"`), - // then fall back to the bare msgId tail. Current WhatsApp Web - // (observed 2026-04-30) emits only the bare msgId on DOM rows - // while IDB still keys by the compound `_serialized` form, so - // the tail is the only segment that matches across both - // sources. `splitn` keeps the msgId intact when it contains - // underscores (legacy edge case). - let bare_mid = mid.rsplitn(2, '_').next().map(str::to_string); - let lookup = by_msg_id - .get(&mid) - .cloned() - .or_else(|| bare_mid.as_deref().and_then(|b| by_msg_id.get(b).cloned())); - if let Some((did, dm)) = lookup { - if consumed.contains(&did) { - continue; - } - if let Some(body) = dm.get("body").and_then(|v| v.as_str()) { - if let Some(obj) = m.as_object_mut() { - obj.insert("body".to_string(), json!(body)); - obj.insert("bodySource".to_string(), json!("dom")); - patched += 1; - consumed.insert(did); - } - } - } - } - } - // Unmatched DOM rows with a body get appended (dedup by dataId). - let mut appended = 0usize; - let mut appended_dids: HashSet = HashSet::new(); - for (_key, (did, dm)) in by_msg_id { - if consumed.contains(&did) || appended_dids.contains(&did) { - continue; - } - if dm - .get("body") - .and_then(|v| v.as_str()) - .map(|s| !s.is_empty()) - .unwrap_or(false) - { - // Modern WhatsApp Web's DOM rows don't carry chatId. Fall - // back to the active conversation's JID (resolved above - // from the conversation header). Without this stamp, - // emit_grouped_whatsapp drops the row at its non-empty - // chat_id check (line ~643), so memory ingest stays empty - // even though we have decrypted body text in hand. Group - // chats are still tricky here — `from_me`/`author` come - // from the per-row data-pre-plain-text — but the chatId - // (group JID) is whatever the user has open, which is the - // common case for an active scanner tick. - let dom_chat_id = dm - .get("chatId") - .and_then(|v| v.as_str()) - .filter(|s| !s.is_empty()) - .map(|s| Value::String(s.to_string())) - .or_else(|| active_chat_jid.clone().map(Value::String)) - .unwrap_or(Value::Null); - messages.push(json!({ - "id": dm.get("dataId").cloned().unwrap_or(Value::Null), - "chatId": dom_chat_id, - "fromMe": dm.get("fromMe").cloned().unwrap_or(Value::Null), - "body": dm.get("body").cloned().unwrap_or(Value::Null), - "author": dm.get("author").cloned().unwrap_or(Value::Null), - "preTimestamp": dm.get("preTimestamp").cloned().unwrap_or(Value::Null), - "bodySource": "dom-only", - })); - appended += 1; - appended_dids.insert(did); - } - } + // The merge logic lives in `merge_dom_into_snapshot` so it can be + // exercised independently in unit tests. + let (messages, patched, appended) = merge_dom_into_snapshot( + &snap.messages, + &snap.dom_messages, + active_chat_jid.as_deref(), + ); + if patched > 0 || appended > 0 { log::info!( "[wa][{}] dom-merge patched={} appended={} total={}", account_id, @@ -855,13 +766,96 @@ fn emit_grouped_whatsapp( source ); } + + // Dual-write: also persist structured chat+message data via the + // dedicated whatsapp_data store. Fire-and-forget alongside the existing + // memory doc ingest path — does not affect scanner tick timing. + { + let acct = account_id.to_string(); + let chats_value = Value::Object(chats.clone()); + // Build normalized message array for the structured ingest. + // Handles both full IDB-scan shape (chatId, timestamp, from/fromName, + // type) and fast DOM-only rows (author, preTimestamp, dataId). + let msgs_for_ingest: Vec = messages + .iter() + .filter_map(|m| { + // Accept chatId from full-scan or chat/chat_id fallbacks on DOM rows. + let chat_id = m + .get("chatId") + .or_else(|| m.get("chat")) + .or_else(|| m.get("chat_id")) + .and_then(|v| v.as_str()) + .filter(|s| !s.is_empty())? + .to_string(); + let body = m + .get("body") + .and_then(|v| v.as_str()) + .map(|s| s.trim()) + .unwrap_or(""); + // Include non-text messages (stickers/images) so message_count + // and last_message_ts stay accurate. Empty body is allowed. + let msg_id = m + .get("id") + .cloned() + .or_else(|| m.get("dataId").cloned()) + .and_then(|v| v.as_str().map(|s| s.to_string())) + .unwrap_or_default(); + if msg_id.is_empty() { + return None; + } + // Resolve sender: full-scan uses fromName/from; DOM rows use author. + let sender = m + .get("fromName") + .cloned() + .or_else(|| m.get("from").cloned()) + .or_else(|| m.get("author").cloned()); + let sender_jid = m + .get("from") + .cloned() + .or_else(|| m.get("author").cloned()) + .or_else(|| m.get("participant").cloned()); + // Resolve timestamp: full-scan has numeric timestamp; + // DOM rows may carry a string preTimestamp that needs parsing. + let timestamp = m + .get("timestamp") + .and_then(|v| v.as_i64()) + .or_else(|| m.get("preTimestamp").and_then(|v| v.as_i64())) + .unwrap_or(0); + Some(json!({ + "message_id": msg_id, + "chat_id": chat_id, + "sender": sender.unwrap_or(Value::Null), + "sender_jid": sender_jid.unwrap_or(Value::Null), + "from_me": m.get("fromMe").and_then(|v| v.as_bool()).unwrap_or(false), + "body": body, + "timestamp": timestamp, + "message_type": m.get("type").cloned().unwrap_or(Value::Null), + "source": source, + })) + }) + .collect(); + let src = source.to_string(); + tokio::spawn(async move { + if let Err(e) = + post_whatsapp_data_ingest(&acct, &chats_value, &msgs_for_ingest, &src).await + { + log::warn!( + "[wa][{}] whatsapp_data structured ingest failed: {}", + acct, + e + ); + } + }); + } } -/// Build the `openhuman.memory_doc_ingest` payload for a single -/// (chatId, day) group and POST it directly to the core. The shape -/// mirrors `persistWhatsappChatDay` on the React side so the memory docs -/// line up whether the scanner or the UI drove the ingest. -async fn post_memory_doc_ingest(account_id: &str, ingest: &Value) -> Result<(), String> { +/// Build the JSON-RPC `params` object for `openhuman.memory_doc_ingest` +/// from a single (chatId, day) ingest payload. Extracted as a pure +/// function so it can be tested independently of the HTTP layer. +/// +/// Returns `None` when the payload is missing required fields (chatId, day, +/// or a non-empty messages array) — callers should skip the HTTP call. +fn build_doc_ingest_params(account_id: &str, ingest: &Value) -> Option { let chat_id = ingest .get("chatId") .and_then(|v| v.as_str()) @@ -880,7 +874,7 @@ async fn post_memory_doc_ingest(account_id: &str, ingest: &Value) -> Result<(), .and_then(|v| v.as_array()) .unwrap_or(&empty); if chat_id.is_empty() || day.is_empty() || msgs.is_empty() { - return Ok(()); + return None; } // Build a stable transcript — sorted by timestamp, one line per msg. @@ -935,7 +929,7 @@ async fn post_memory_doc_ingest(account_id: &str, ingest: &Value) -> Result<(), let key = format!("{chat_id}:{day}"); let title = format!("WhatsApp · {chat_name} · {day}"); - let params = json!({ + Some(json!({ "namespace": namespace, "key": key, "title": title, @@ -952,7 +946,39 @@ async fn post_memory_doc_ingest(account_id: &str, ingest: &Value) -> Result<(), "message_count": sorted.len(), }, "category": "core", - }); + })) +} + +/// Build the `openhuman.memory_doc_ingest` payload for a single +/// (chatId, day) group and POST it directly to the core. The shape +/// mirrors `persistWhatsappChatDay` on the React side so the memory docs +/// line up whether the scanner or the UI drove the ingest. +/// +/// Retries once (after 500ms) on connection errors so the scanner isn't +/// silently dropped when the core sidecar isn't ready yet at startup. +async fn post_memory_doc_ingest(account_id: &str, ingest: &Value) -> Result<(), String> { + let params = match build_doc_ingest_params(account_id, ingest) { + Some(p) => p, + None => return Ok(()), + }; + + // Extract namespace/key for the success log from the built params. + let namespace = params + .get("namespace") + .and_then(|v| v.as_str()) + .unwrap_or("") + .to_string(); + let key = params + .get("key") + .and_then(|v| v.as_str()) + .unwrap_or("") + .to_string(); + let msg_count = params + .get("metadata") + .and_then(|m| m.get("message_count")) + .and_then(|v| v.as_u64()) + .unwrap_or(0); + let body = json!({ "jsonrpc": "2.0", "id": 1, @@ -960,6 +986,125 @@ async fn post_memory_doc_ingest(account_id: &str, ingest: &Value) -> Result<(), "params": params, }); + let url = crate::core_rpc::core_rpc_url_value(); + + // Retry up to 2 attempts with 500ms delay on connection errors (e.g. + // core sidecar not yet ready at scanner startup). HTTP-level errors + // (non-2xx responses, JSON-RPC errors) are not retried — they indicate + // a real problem rather than a startup race. + let mut last_err = String::new(); + for attempt in 1u8..=2 { + let client = reqwest::Client::builder() + .timeout(Duration::from_secs(15)) + .build() + .map_err(|e| format!("http client: {e}"))?; + let req = crate::core_rpc::apply_auth(client.post(&url)) + .map_err(|e| format!("prepare {url}: {e}"))?; + let send_result = req.json(&body).send().await; + match send_result { + Err(e) if e.is_connect() || e.is_timeout() => { + last_err = format!("POST {url}: {e}"); + if attempt < 2 { + log::debug!( + "[wa][{}] memory ingest connect error (attempt {}), retrying in 500ms: {}", + account_id, + attempt, + e + ); + sleep(Duration::from_millis(500)).await; + continue; + } + return Err(last_err); + } + Err(e) => return Err(format!("POST {url}: {e}")), + Ok(resp) => { + let status = resp.status(); + if !status.is_success() { + let body_text = resp.text().await.unwrap_or_default(); + return Err(format!("{status}: {body_text}")); + } + let v: Value = resp.json().await.map_err(|e| format!("decode: {e}"))?; + if let Some(err) = v.get("error") { + return Err(format!("rpc error: {err}")); + } + log::info!( + "[wa][{}] memory upsert ok namespace={} key={} msgs={}", + account_id, + namespace, + key, + msg_count + ); + return Ok(()); + } + } + } + Err(last_err) +} + +/// POST a structured `openhuman.whatsapp_data_ingest` payload to the core. +/// +/// This is the dual-write path alongside `post_memory_doc_ingest`. It +/// persists chats and messages into the dedicated `whatsapp_data.db` SQLite +/// store so the agent can query them via structured RPC tools. +async fn post_whatsapp_data_ingest( + account_id: &str, + chats: &Value, + messages: &[Value], + source: &str, +) -> Result<(), String> { + if messages.is_empty() && chats.as_object().map(|o| o.is_empty()).unwrap_or(true) { + return Ok(()); + } + log::debug!( + "[wa][{}] whatsapp_data_ingest chats={} messages={} source={}", + account_id, + chats.as_object().map(|o| o.len()).unwrap_or(0), + messages.len(), + source + ); + + // Convert chats map values to {name: string|null}. + // The scanner passes chats as either: + // - Value::String(display_name) — contact-cache format + // - Value::Object({name: ..., ...}) — full IDB scan format + let chats_param: serde_json::Map = chats + .as_object() + .map(|o| { + o.iter() + .map(|(jid, v)| { + let name = if let Some(s) = v.as_str() { + // String-valued entry from contact cache: value IS the name. + if s.is_empty() { + Value::Null + } else { + Value::String(s.to_string()) + } + } else { + // Object entry: look for a "name" field. + v.get("name") + .and_then(|n| n.as_str()) + .filter(|s| !s.is_empty()) + .map(|s| Value::String(s.to_string())) + .unwrap_or(Value::Null) + }; + (jid.clone(), json!({ "name": name })) + }) + .collect() + }) + .unwrap_or_default(); + + let params = json!({ + "account_id": account_id, + "chats": Value::Object(chats_param), + "messages": messages, + }); + let body = json!({ + "jsonrpc": "2.0", + "id": 1, + "method": "openhuman.whatsapp_data_ingest", + "params": params, + }); + let url = crate::core_rpc::core_rpc_url_value(); let client = reqwest::Client::builder() .timeout(Duration::from_secs(15)) @@ -974,23 +1119,140 @@ async fn post_memory_doc_ingest(account_id: &str, ingest: &Value) -> Result<(), .map_err(|e| format!("POST {url}: {e}"))?; let status = resp.status(); if !status.is_success() { - let body = resp.text().await.unwrap_or_default(); - return Err(format!("{status}: {body}")); + let body_text = resp.text().await.unwrap_or_default(); + return Err(format!("{status}: {body_text}")); } let v: Value = resp.json().await.map_err(|e| format!("decode: {e}"))?; if let Some(err) = v.get("error") { return Err(format!("rpc error: {err}")); } - log::info!( - "[wa][{}] memory upsert ok namespace={} key={} msgs={}", + log::debug!( + "[wa][{}] whatsapp_data_ingest ok account={} messages={}", account_id, - namespace, - key, - sorted.len() + account_id, + messages.len() ); Ok(()) } +/// Merge DOM-scraped rows into an IDB-sourced message list. +/// +/// Extracted from `emit_snapshot` so the merge logic can be tested +/// independently of the Tauri `AppHandle`. Behaviour: +/// +/// 1. Build an index of DOM rows keyed by both their full `dataId` and bare +/// `msgId` (the current WA Web format emits only the bare hex id). +/// 2. Patch IDB messages that have an empty `body` with the DOM row's body; +/// mark the DOM row as consumed. +/// 3. Append unmatched DOM rows that have a non-empty body, stamping +/// `chatId` from `active_chat_jid` when the row lacks one. +/// +/// Returns the merged message list along with patch/append counts for +/// diagnostic logging. +fn merge_dom_into_snapshot( + idb_messages: &[Value], + dom_messages: &[Value], + active_chat_jid: Option<&str>, +) -> (Vec, usize, usize) { + use std::collections::{HashMap, HashSet}; + + let mut messages = idb_messages.to_vec(); + + if dom_messages.is_empty() { + return (messages, 0, 0); + } + + // Index DOM rows by full dataId and bare msgId. + let mut by_msg_id: HashMap = HashMap::new(); + for dm in dom_messages { + let did = dm + .get("dataId") + .and_then(|v| v.as_str()) + .unwrap_or("") + .to_string(); + if did.is_empty() { + continue; + } + by_msg_id.insert(did.clone(), (did.clone(), dm.clone())); + if let Some(mid) = dm.get("msgId").and_then(|v| v.as_str()) { + by_msg_id + .entry(mid.to_string()) + .or_insert_with(|| (did.clone(), dm.clone())); + } + } + + let mut consumed: HashSet = HashSet::new(); + let mut patched = 0usize; + + for m in messages.iter_mut() { + let mid_opt = m.get("id").and_then(|v| v.as_str()).map(|s| s.to_string()); + let has_body = m + .get("body") + .and_then(|v| v.as_str()) + .map(|s| !s.is_empty()) + .unwrap_or(false); + if has_body { + continue; + } + if let Some(mid) = mid_opt { + let bare_mid = mid.rsplitn(2, '_').next().map(str::to_string); + let lookup = by_msg_id + .get(&mid) + .cloned() + .or_else(|| bare_mid.as_deref().and_then(|b| by_msg_id.get(b).cloned())); + if let Some((did, dm)) = lookup { + if consumed.contains(&did) { + continue; + } + if let Some(body) = dm.get("body").and_then(|v| v.as_str()) { + if let Some(obj) = m.as_object_mut() { + obj.insert("body".to_string(), json!(body)); + obj.insert("bodySource".to_string(), json!("dom")); + patched += 1; + consumed.insert(did); + } + } + } + } + } + + // Append unmatched DOM rows that have a body. + let mut appended = 0usize; + let mut appended_dids: HashSet = HashSet::new(); + for (_key, (did, dm)) in by_msg_id { + if consumed.contains(&did) || appended_dids.contains(&did) { + continue; + } + if dm + .get("body") + .and_then(|v| v.as_str()) + .map(|s| !s.is_empty()) + .unwrap_or(false) + { + let dom_chat_id = dm + .get("chatId") + .and_then(|v| v.as_str()) + .filter(|s| !s.is_empty()) + .map(|s| Value::String(s.to_string())) + .or_else(|| active_chat_jid.map(|j| Value::String(j.to_string()))) + .unwrap_or(Value::Null); + messages.push(json!({ + "id": dm.get("dataId").cloned().unwrap_or(Value::Null), + "chatId": dom_chat_id, + "fromMe": dm.get("fromMe").cloned().unwrap_or(Value::Null), + "body": dm.get("body").cloned().unwrap_or(Value::Null), + "author": dm.get("author").cloned().unwrap_or(Value::Null), + "preTimestamp": dm.get("preTimestamp").cloned().unwrap_or(Value::Null), + "bodySource": "dom-only", + })); + appended += 1; + appended_dids.insert(did); + } + } + + (messages, patched, appended) +} + /// Track which (account_id, provider) pairs we've already started a scanner /// for. The webview lifecycle can call `ensure_scanner` repeatedly without /// double-spawning. @@ -1154,4 +1416,469 @@ mod tests { assert!(registry.started.lock().is_empty()); assert_all_cancelled(tasks).await; } + + // ── seconds_to_ymd ──────────────────────────────────────────────────────── + + #[test] + fn seconds_to_ymd_known_timestamp() { + // Unix timestamp 1_700_000_000 = 2023-11-14 (UTC). + assert_eq!(seconds_to_ymd(1_700_000_000), "2023-11-14"); + } + + #[test] + fn seconds_to_ymd_epoch_zero() { + // Unix epoch origin = 1970-01-01. + assert_eq!(seconds_to_ymd(0), "1970-01-01"); + } + + #[test] + fn seconds_to_ymd_output_format_is_yyyy_mm_dd() { + let s = seconds_to_ymd(1_700_000_000); + // Must match YYYY-MM-DD: 10 chars, digit/digit/digit/digit-...-... + assert_eq!(s.len(), 10, "expected 10-char date string, got: {s}"); + let parts: Vec<&str> = s.split('-').collect(); + assert_eq!(parts.len(), 3, "expected 3 dash-separated parts: {s}"); + assert_eq!(parts[0].len(), 4, "year must be 4 digits: {s}"); + assert_eq!(parts[1].len(), 2, "month must be 2 digits: {s}"); + assert_eq!(parts[2].len(), 2, "day must be 2 digits: {s}"); + assert!( + parts.iter().all(|p| p.chars().all(|c| c.is_ascii_digit())), + "all parts must be numeric: {s}" + ); + } + + // ── parse_pre_timestamp_ymd ─────────────────────────────────────────────── + + #[test] + fn parse_pre_timestamp_ymd_valid_wa_format() { + // WhatsApp Web format: "4:53 AM, 7/5/2025" + let result = parse_pre_timestamp_ymd("4:53 AM, 7/5/2025"); + assert_eq!(result.as_deref(), Some("2025-07-05")); + } + + #[test] + fn parse_pre_timestamp_ymd_another_valid_date() { + // "10:01 PM, 11/14/2023" — matches our known ts + let result = parse_pre_timestamp_ymd("10:01 PM, 11/14/2023"); + assert_eq!(result.as_deref(), Some("2023-11-14")); + } + + #[test] + fn parse_pre_timestamp_ymd_empty_string_returns_none() { + assert!(parse_pre_timestamp_ymd("").is_none()); + } + + #[test] + fn parse_pre_timestamp_ymd_no_comma_returns_none() { + assert!(parse_pre_timestamp_ymd("4:53 AM 7/5/2025").is_none()); + } + + #[test] + fn parse_pre_timestamp_ymd_invalid_date_parts_return_none() { + // Month 13 is out of range. + assert!(parse_pre_timestamp_ymd("10:00 AM, 13/5/2025").is_none()); + // Day 32 is out of range. + assert!(parse_pre_timestamp_ymd("10:00 AM, 1/32/2025").is_none()); + } + + #[test] + fn parse_pre_timestamp_ymd_garbage_returns_none() { + assert!(parse_pre_timestamp_ymd("not a timestamp at all").is_none()); + } + + // ── emit_grouped_whatsapp grouping ──────────────────────────────────────── + + /// Build a minimal message Value that `emit_grouped_whatsapp` will accept. + fn make_msg(chat_id: &str, ts: i64, body: &str, from_me: bool) -> Value { + json!({ + "chatId": chat_id, + "body": body, + "timestamp": ts, + "fromMe": from_me, + "from": if from_me { "me" } else { chat_id }, + }) + } + + #[test] + fn grouping_produces_correct_group_count_and_keys() { + use std::collections::HashMap; + + // 3 messages in alice@c.us on day 2023-11-14 (ts ≈ 1_700_000_000). + // 2 messages in group@g.us on a different day (ts ≈ 1_700_100_000 = + // 2023-11-15 UTC). + let day1_ts = 1_700_000_000i64; // 2023-11-14 + let day2_ts = 1_700_100_000i64; // 2023-11-15 + + let messages = vec![ + make_msg("alice@c.us", day1_ts, "Hello", false), + make_msg("alice@c.us", day1_ts + 60, "How are you?", false), + make_msg("alice@c.us", day1_ts + 120, "Fine thanks", true), + make_msg("group@g.us", day2_ts, "Meeting at 3pm", false), + make_msg("group@g.us", day2_ts + 30, "Got it", true), + ]; + + // Collect groups the same way emit_grouped_whatsapp does it. + let empty_chats = serde_json::Map::new(); + let now_secs = std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .map(|d| d.as_secs() as i64) + .unwrap_or(0); + + let mut groups: HashMap<(String, String), Vec> = HashMap::new(); + for m in &messages { + let chat_id = match m.get("chatId").and_then(|v| v.as_str()) { + Some(s) if !s.is_empty() => s.to_string(), + _ => continue, + }; + let body = m + .get("body") + .and_then(|v| v.as_str()) + .map(|s| s.trim().to_string()) + .unwrap_or_default(); + if body.is_empty() { + continue; + } + let day: String = if let Some(t) = m.get("timestamp").and_then(|v| v.as_i64()) { + seconds_to_ymd(t) + } else { + seconds_to_ymd(now_secs) + }; + let _ = &empty_chats; + groups.entry((chat_id, day)).or_default().push(m.clone()); + } + + assert_eq!(groups.len(), 2, "expected exactly 2 (chatId, day) groups"); + + let alice_day = seconds_to_ymd(day1_ts); + let group_day = seconds_to_ymd(day2_ts); + + let alice_key = ("alice@c.us".to_string(), alice_day.clone()); + let group_key = ("group@g.us".to_string(), group_day.clone()); + + assert!( + groups.contains_key(&alice_key), + "alice group missing; groups: {groups:?}" + ); + assert!( + groups.contains_key(&group_key), + "group@g.us group missing; groups: {groups:?}" + ); + + assert_eq!( + groups[&alice_key].len(), + 3, + "alice chat should have 3 messages" + ); + assert_eq!( + groups[&group_key].len(), + 2, + "group chat should have 2 messages" + ); + } + + // ── transcript format ───────────────────────────────────────────────────── + + #[test] + fn build_doc_ingest_params_transcript_contains_senders_and_bodies() { + let day_ts = 1_700_000_000i64; // 2023-11-14 + let ingest = json!({ + "chatId": "alice@c.us", + "chatName": "Alice", + "day": seconds_to_ymd(day_ts), + "messages": [ + { + "chatId": "alice@c.us", + "fromMe": false, + "from": "alice@c.us", + "fromName": "Alice", + "body": "Hey there!", + "timestamp": day_ts, + }, + { + "chatId": "alice@c.us", + "fromMe": true, + "from": "me", + "fromName": null, + "body": "Hi Alice!", + "timestamp": day_ts + 60, + }, + { + "chatId": "alice@c.us", + "fromMe": false, + "from": "alice@c.us", + "fromName": "Alice", + "body": "How are you?", + "timestamp": day_ts + 120, + }, + ], + }); + + let params = build_doc_ingest_params("test-acct@c.us", &ingest) + .expect("should build params for valid ingest"); + + let content = params + .get("content") + .and_then(|v| v.as_str()) + .expect("content must be present"); + + // Senders should appear in the transcript. + assert!( + content.contains("Alice"), + "transcript must contain sender name 'Alice'; content:\n{content}" + ); + assert!( + content.contains("me"), + "transcript must contain 'me' for self-sent messages; content:\n{content}" + ); + + // Bodies must be present. + assert!( + content.contains("Hey there!"), + "transcript must contain first message body; content:\n{content}" + ); + assert!( + content.contains("Hi Alice!"), + "transcript must contain second message body; content:\n{content}" + ); + assert!( + content.contains("How are you?"), + "transcript must contain third message body; content:\n{content}" + ); + + // Lines must appear in ascending timestamp order — verify by position. + let pos_hey = content.find("Hey there!").expect("Hey there not found"); + let pos_hi = content.find("Hi Alice!").expect("Hi Alice not found"); + let pos_how = content.find("How are you?").expect("How are you not found"); + assert!( + pos_hey < pos_hi && pos_hi < pos_how, + "transcript lines must be in timestamp order" + ); + } + + // ── build_doc_ingest_params payload shape ───────────────────────────────── + + #[test] + fn build_doc_ingest_params_namespace_and_key_format() { + let day = "2023-11-14"; + let ingest = json!({ + "chatId": "alice@c.us", + "chatName": "Alice", + "day": day, + "messages": [ + { "chatId": "alice@c.us", "fromMe": false, "from": "alice@c.us", + "fromName": "Alice", "body": "Hello", "timestamp": 1_700_000_000i64 } + ], + }); + + let params = + build_doc_ingest_params("test-acct@c.us", &ingest).expect("should build params"); + + assert_eq!( + params.get("namespace").and_then(|v| v.as_str()), + Some("whatsapp-web:test-acct@c.us"), + "namespace must be 'whatsapp-web:'" + ); + assert_eq!( + params.get("key").and_then(|v| v.as_str()), + Some("alice@c.us:2023-11-14"), + "key must be ':'" + ); + assert_eq!( + params.get("source_type").and_then(|v| v.as_str()), + Some("whatsapp-web"), + "source_type must be 'whatsapp-web'" + ); + + // Content must be non-empty and contain the body. + let content = params + .get("content") + .and_then(|v| v.as_str()) + .expect("content must be present"); + assert!(!content.is_empty(), "content must not be empty"); + assert!( + content.contains("Hello"), + "content must contain message body; got:\n{content}" + ); + } + + #[test] + fn build_doc_ingest_params_missing_chat_id_returns_none() { + let ingest = json!({ + "chatName": "Alice", + "day": "2023-11-14", + "messages": [ + { "chatId": "alice@c.us", "fromMe": false, "body": "Hello", "timestamp": 1i64 } + ], + }); + assert!( + build_doc_ingest_params("acct", &ingest).is_none(), + "missing chatId must return None" + ); + } + + #[test] + fn build_doc_ingest_params_empty_messages_returns_none() { + let ingest = json!({ + "chatId": "alice@c.us", + "chatName": "Alice", + "day": "2023-11-14", + "messages": [], + }); + assert!( + build_doc_ingest_params("acct", &ingest).is_none(), + "empty messages must return None" + ); + } + + // ── DOM-IDB merge ───────────────────────────────────────────────────────── + + #[test] + fn merge_dom_patches_empty_body_from_idb_message() { + // IDB message with empty body; matching DOM row has the decrypted body. + let idb = vec![json!({ + "id": "abc123", + "chatId": "alice@c.us", + "fromMe": false, + "body": "", + })]; + let dom = vec![json!({ + "dataId": "abc123", + "msgId": "abc123", + "chatId": "alice@c.us", + "fromMe": false, + "body": "Hello", + "author": "Alice", + "preTimestamp": null, + })]; + + let (merged, patched, appended) = merge_dom_into_snapshot(&idb, &dom, None); + + assert_eq!(patched, 1, "one message should be patched"); + assert_eq!(appended, 0, "no messages should be appended"); + assert_eq!(merged.len(), 1, "still one message in merged list"); + + let body = merged[0] + .get("body") + .and_then(|v| v.as_str()) + .expect("body must be present"); + assert_eq!(body, "Hello", "patched body must equal DOM body"); + + let source = merged[0] + .get("bodySource") + .and_then(|v| v.as_str()) + .expect("bodySource must be present"); + assert_eq!(source, "dom", "bodySource must be 'dom' after patching"); + } + + #[test] + fn merge_dom_appends_unmatched_row_with_active_chat_backfill() { + // No IDB messages; DOM has a row with no chatId. active_chat_jid + // should be stamped onto the appended message. + let idb: Vec = vec![]; + let dom = vec![json!({ + "dataId": "newrow1", + "msgId": "newrow1", + "chatId": "", // empty — needs backfill + "fromMe": false, + "body": "Hey from active chat", + "author": "Bob", + "preTimestamp": null, + })]; + + let (merged, patched, appended) = merge_dom_into_snapshot(&idb, &dom, Some("bob@c.us")); + + assert_eq!(patched, 0, "nothing to patch"); + assert_eq!(appended, 1, "one row should be appended"); + assert_eq!(merged.len(), 1, "merged list should have 1 entry"); + + let chat_id = merged[0] + .get("chatId") + .and_then(|v| v.as_str()) + .expect("chatId must be present"); + assert_eq!( + chat_id, "bob@c.us", + "chatId should be backfilled from active_chat_jid" + ); + + let body_source = merged[0] + .get("bodySource") + .and_then(|v| v.as_str()) + .expect("bodySource must be present"); + assert_eq!(body_source, "dom-only"); + } + + #[test] + fn merge_dom_does_not_append_row_without_body() { + // DOM rows without a body should be silently skipped. + let idb: Vec = vec![]; + let dom = vec![json!({ + "dataId": "empty1", + "msgId": "empty1", + "chatId": "alice@c.us", + "fromMe": false, + "body": "", + })]; + + let (merged, patched, appended) = merge_dom_into_snapshot(&idb, &dom, None); + + assert_eq!(patched, 0); + assert_eq!(appended, 0, "empty-body DOM rows must not be appended"); + assert!( + merged.is_empty(), + "no messages should appear in merged list" + ); + } + + #[test] + fn merge_dom_does_not_consume_row_twice() { + // Two IDB messages with the same bare msgId; only the first match + // should consume the DOM row. + let idb = vec![ + json!({ "id": "chat_abc", "chatId": "alice@c.us", "fromMe": false, "body": "" }), + json!({ "id": "chat_abc_2", "chatId": "alice@c.us", "fromMe": true, "body": "" }), + ]; + // DOM row keyed only by bare msgId "abc". + let dom = vec![json!({ + "dataId": "abc", + "msgId": "abc", + "chatId": "alice@c.us", + "fromMe": false, + "body": "Only once", + })]; + + let (merged, patched, _appended) = merge_dom_into_snapshot(&idb, &dom, None); + + // Exactly one of the two IDB messages should be patched. + assert_eq!(patched, 1, "DOM row must be consumed at most once"); + assert_eq!(merged.len(), 2, "both IDB messages must survive merge"); + let patched_bodies: Vec<&str> = merged + .iter() + .filter_map(|m| m.get("body").and_then(|v| v.as_str())) + .filter(|b| *b == "Only once") + .collect(); + assert_eq!( + patched_bodies.len(), + 1, + "body 'Only once' must appear exactly once in merged list" + ); + } + + #[test] + fn merge_dom_empty_dom_returns_idb_messages_unchanged() { + let idb = vec![ + json!({ "id": "m1", "chatId": "a@c.us", "body": "hello" }), + json!({ "id": "m2", "chatId": "a@c.us", "body": "" }), + ]; + let dom: Vec = vec![]; + + let (merged, patched, appended) = merge_dom_into_snapshot(&idb, &dom, None); + + assert_eq!(patched, 0); + assert_eq!(appended, 0); + assert_eq!(merged.len(), 2, "IDB messages must be returned unchanged"); + assert_eq!( + merged[0].get("body").and_then(|v| v.as_str()), + Some("hello") + ); + } } diff --git a/src/core/all.rs b/src/core/all.rs index 439ae87cf..3a56e4ae0 100644 --- a/src/core/all.rs +++ b/src/core/all.rs @@ -49,6 +49,11 @@ impl RegisteredController { /// The global static registry of all controllers, initialized once on first access. static REGISTRY: OnceLock> = OnceLock::new(); +/// Internal-only controllers: registered for RPC dispatch but NOT in the agent-facing +/// schema catalog. These handlers are callable by trusted callers (e.g. the Tauri scanner) +/// but should not be advertised to agents via tool listings or schema discovery. +static INTERNAL_REGISTRY: OnceLock> = OnceLock::new(); + /// The global static registry of standalone CLI adapters. static CLI_ADAPTERS: OnceLock> = OnceLock::new(); @@ -69,6 +74,16 @@ fn registry() -> &'static [RegisteredController] { .as_slice() } +/// Returns a reference to the internal-only controller registry. +/// +/// These controllers are callable over RPC but are NOT included in agent tool listings +/// or schema discovery endpoints. +fn internal_registry() -> &'static [RegisteredController] { + INTERNAL_REGISTRY + .get_or_init(build_internal_only_controllers) + .as_slice() +} + /// Returns a reference to the global CLI adapter registry. fn cli_adapters() -> &'static [RegisteredCliAdapter] { CLI_ADAPTERS.get_or_init(|| { @@ -192,6 +207,21 @@ fn build_registered_controllers() -> Vec { ); // Integration notification ingest, triage, and per-provider settings controllers.extend(crate::openhuman::notifications::all_notifications_registered_controllers()); + // Structured WhatsApp Web data — agent-facing read-only controllers (list/search). + // The write-path ingest controller is registered separately in build_internal_only_controllers. + controllers.extend(crate::openhuman::whatsapp_data::all_whatsapp_data_registered_controllers()); + controllers +} + +/// Aggregates controllers that are registered for RPC routing but NOT exposed to agents. +/// +/// These are write-path or internal-only handlers callable by trusted callers +/// (e.g. the Tauri scanner ingest path) that should not appear in agent tool listings. +fn build_internal_only_controllers() -> Vec { + let mut controllers = Vec::new(); + // whatsapp_data ingest: scanner-side write path. Callable over RPC by the + // Tauri scanner but excluded from agent-facing schema discovery. + controllers.extend(crate::openhuman::whatsapp_data::all_whatsapp_data_internal_controllers()); controllers } @@ -255,6 +285,8 @@ fn build_declared_controller_schemas() -> Vec { ); // Integration notification ingest, triage, and per-provider settings schemas.extend(crate::openhuman::notifications::all_notifications_controller_schemas()); + // Structured WhatsApp Web data — local SQLite store, agent-queryable + schemas.extend(crate::openhuman::whatsapp_data::all_whatsapp_data_controller_schemas()); schemas } @@ -340,6 +372,9 @@ pub fn namespace_description(namespace: &str) -> Option<&'static str> { "Integration notification ingest, triage scoring, listing, read-state, \ and per-provider routing settings.", ), + "whatsapp_data" => Some( + "Structured WhatsApp conversation and message store — list chats, read messages, and search across WhatsApp Web data.", + ), _ => None, } } @@ -361,9 +396,13 @@ pub fn rpc_method_from_parts(namespace: &str, function: &str) -> Option } /// Retrieves the schema for a specific RPC method. +/// +/// Checks both the agent-facing registry and the internal registry so that +/// parameter validation still applies to internal-only methods (e.g. ingest). pub fn schema_for_rpc_method(method: &str) -> Option { registry() .iter() + .chain(internal_registry().iter()) .find(|r| r.rpc_method_name() == method) .map(|r| r.schema.clone()) } @@ -400,7 +439,11 @@ pub fn validate_params( /// Attempts to invoke a registered RPC method by name. /// -/// Returns `None` if the method is not found in the registry. +/// Checks both the agent-facing controller registry and the internal-only registry, +/// so scanner-side write paths (e.g. `openhuman.whatsapp_data_ingest`) are routable +/// even though they are not included in agent tool listings. +/// +/// Returns `None` if the method is not found in either registry. pub async fn try_invoke_registered_rpc( method: &str, params: Map, @@ -410,6 +453,11 @@ pub async fn try_invoke_registered_rpc( return Some((controller.handler)(params).await); } } + for controller in internal_registry() { + if controller.rpc_method_name() == method { + return Some((controller.handler)(params).await); + } + } None } diff --git a/src/core/jsonrpc.rs b/src/core/jsonrpc.rs index 1d8e40398..acdf2a533 100644 --- a/src/core/jsonrpc.rs +++ b/src/core/jsonrpc.rs @@ -674,6 +674,15 @@ async fn run_server_inner( ), Err(e) => log::warn!("[boot] memory::global init failed: {e}"), } + // Initialize the WhatsApp data store so scanner ingest calls + // can write data without requiring a lazy-init fallback. + match crate::openhuman::whatsapp_data::global::init(cfg.workspace_dir.clone()) { + Ok(_) => log::info!( + "[boot] whatsapp_data::global initialized (workspace={})", + cfg.workspace_dir.display() + ), + Err(e) => log::warn!("[boot] whatsapp_data::global init failed: {e}"), + } } let (resolved_port, port_source) = match port { diff --git a/src/openhuman/about_app/catalog.rs b/src/openhuman/about_app/catalog.rs index 909222fac..5ac6719b9 100644 --- a/src/openhuman/about_app/catalog.rs +++ b/src/openhuman/about_app/catalog.rs @@ -686,6 +686,16 @@ const CAPABILITIES: &[Capability] = &[ status: CapabilityStatus::Beta, privacy: None, }, + Capability { + id: "channels.whatsapp_read_messages", + name: "Read WhatsApp Messages", + domain: "channels", + category: CapabilityCategory::Channels, + description: "Read and search WhatsApp Web conversations and messages after connecting WhatsApp in OpenHuman. Data is stored locally only and never transmitted.", + how_to: "Connect WhatsApp Web via Channels, then ask the agent to read or summarise your messages.", + status: CapabilityStatus::Beta, + privacy: LOCAL_RAW, + }, Capability { id: "settings.configure_ai", name: "Configure AI", diff --git a/src/openhuman/mod.rs b/src/openhuman/mod.rs index 319bfbe4d..95704fc2b 100644 --- a/src/openhuman/mod.rs +++ b/src/openhuman/mod.rs @@ -70,4 +70,5 @@ pub mod webhooks; pub mod webview_accounts; pub mod webview_apis; pub mod webview_notifications; +pub mod whatsapp_data; pub mod workspace; diff --git a/src/openhuman/whatsapp_data/global.rs b/src/openhuman/whatsapp_data/global.rs new file mode 100644 index 000000000..52767586d --- /dev/null +++ b/src/openhuman/whatsapp_data/global.rs @@ -0,0 +1,57 @@ +//! Process-global WhatsApp data store singleton. +//! +//! One `WhatsAppDataStore` lives for the entire core process, shared by RPC +//! handlers and any other subsystem that needs it. +//! +//! # Usage +//! +//! ```ignore +//! // At startup: +//! whatsapp_data::global::init(workspace_dir)?; +//! +//! // In RPC handlers: +//! let store = whatsapp_data::global::store()?; +//! ``` + +use std::path::PathBuf; +use std::sync::{Arc, OnceLock}; + +use crate::openhuman::whatsapp_data::store::WhatsAppDataStore; + +/// Shared, thread-safe reference to the store. +pub type WhatsAppDataStoreRef = Arc; + +static GLOBAL_STORE: OnceLock = OnceLock::new(); + +/// Initialise the global store from a workspace directory. Idempotent — +/// only the first call has any effect; subsequent calls return the existing +/// instance. +pub fn init(workspace_dir: PathBuf) -> Result { + if let Some(existing) = GLOBAL_STORE.get() { + log::debug!("[whatsapp_data:global] already initialised"); + return Ok(Arc::clone(existing)); + } + log::info!( + "[whatsapp_data:global] initialising store workspace={}", + workspace_dir.display() + ); + let store = Arc::new( + WhatsAppDataStore::new(&workspace_dir) + .map_err(|e| format!("[whatsapp_data] store init failed: {e}"))?, + ); + let _ = GLOBAL_STORE.set(Arc::clone(&store)); + Ok(GLOBAL_STORE.get().cloned().unwrap_or(store)) +} + +/// Return the global store. Errors if [`init`] has not been called yet. +pub fn store() -> Result { + GLOBAL_STORE.get().cloned().ok_or_else(|| { + "whatsapp_data global store accessed before init — call init(workspace) at startup" + .to_string() + }) +} + +/// Return the global store if already initialised, without error. +pub fn store_if_ready() -> Option { + GLOBAL_STORE.get().cloned() +} diff --git a/src/openhuman/whatsapp_data/mod.rs b/src/openhuman/whatsapp_data/mod.rs new file mode 100644 index 000000000..8ca2589bb --- /dev/null +++ b/src/openhuman/whatsapp_data/mod.rs @@ -0,0 +1,29 @@ +//! Structured WhatsApp Web data — local-only SQLite persistence and agent API. +//! +//! This domain stores WhatsApp chats and messages scraped by the Tauri +//! `whatsapp_scanner` via CDP, making them queryable by the agent through +//! the JSON-RPC controller surface. +//! +//! **Data locality**: all data remains on-device in `whatsapp_data.db`; it is +//! never transmitted to any external service. +//! +//! ## Agent-facing RPC methods (read-only) +//! - `openhuman.whatsapp_data_list_chats` +//! - `openhuman.whatsapp_data_list_messages` +//! - `openhuman.whatsapp_data_search_messages` +//! +//! ## Internal-only RPC method (write, scanner-side) +//! - `openhuman.whatsapp_data_ingest` — NOT exposed via agent tool listings + +pub mod global; +pub mod ops; +pub mod rpc; +mod schemas; +pub mod store; +pub mod types; + +pub use schemas::{ + all_controller_schemas as all_whatsapp_data_controller_schemas, + all_internal_controllers as all_whatsapp_data_internal_controllers, + all_registered_controllers as all_whatsapp_data_registered_controllers, +}; diff --git a/src/openhuman/whatsapp_data/ops.rs b/src/openhuman/whatsapp_data/ops.rs new file mode 100644 index 000000000..99c087b81 --- /dev/null +++ b/src/openhuman/whatsapp_data/ops.rs @@ -0,0 +1,200 @@ +//! Business logic for WhatsApp data ingestion and retrieval. +//! +//! All operations take a `&WhatsAppDataStore` so callers control the store +//! lifetime (shared `Arc` at runtime, fresh instance in tests). + +use anyhow::Result; + +use crate::openhuman::whatsapp_data::{ + store::WhatsAppDataStore, + types::{ + IngestRequest, IngestResult, ListChatsRequest, ListMessagesRequest, SearchMessagesRequest, + WhatsAppChat, WhatsAppMessage, + }, +}; + +/// Number of seconds in 90 days — the auto-prune horizon. +const PRUNE_HORIZON_SECS: i64 = 90 * 24 * 60 * 60; + +/// Ingest a scanner snapshot: upsert chats and messages, then prune messages +/// older than 90 days. +/// +/// Returns counts for observability / logging at the RPC layer. +pub fn ingest(store: &WhatsAppDataStore, req: IngestRequest) -> Result { + log::debug!( + "[whatsapp_data] ingest start chats={} messages={} (account redacted)", + req.chats.len(), + req.messages.len() + ); + + let chats_upserted = store.upsert_chats(&req.account_id, &req.chats)?; + let messages_upserted = store.upsert_messages(&req.account_id, &req.messages)?; + + let now = std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .map(|d| d.as_secs() as i64) + .unwrap_or(0); + let cutoff_ts = now - PRUNE_HORIZON_SECS; + let messages_pruned = store.prune_old_messages(cutoff_ts)?; + + let result = IngestResult { + chats_upserted, + messages_upserted, + messages_pruned, + }; + log::debug!( + "[whatsapp_data] ingest done chats_upserted={} messages_upserted={} pruned={} (account redacted)", + result.chats_upserted, + result.messages_upserted, + result.messages_pruned + ); + Ok(result) +} + +/// Return chats from the local store, optionally filtered by account. +pub fn list_chats(store: &WhatsAppDataStore, req: ListChatsRequest) -> Result> { + log::debug!( + "[whatsapp_data] list_chats has_account={} limit={:?} offset={:?}", + req.account_id.is_some(), + req.limit, + req.offset + ); + store.list_chats(&req) +} + +/// Return messages for a chat, with optional time range and pagination. +pub fn list_messages( + store: &WhatsAppDataStore, + req: ListMessagesRequest, +) -> Result> { + log::debug!( + "[whatsapp_data] list_messages has_account={} (chat/account redacted)", + req.account_id.is_some() + ); + store.list_messages(&req) +} + +/// Full-text search over message bodies. +pub fn search_messages( + store: &WhatsAppDataStore, + req: SearchMessagesRequest, +) -> Result> { + log::debug!( + "[whatsapp_data] search_messages has_account={} has_chat={} (query/identifiers redacted)", + req.account_id.is_some(), + req.chat_id.is_some() + ); + store.search_messages(&req) +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::openhuman::whatsapp_data::types::{ChatMeta, IngestMessage}; + use std::collections::HashMap; + use tempfile::tempdir; + + fn make_store() -> (WhatsAppDataStore, tempfile::TempDir) { + let tmp = tempdir().expect("tempdir"); + let store = WhatsAppDataStore::new(tmp.path()).expect("store"); + (store, tmp) + } + + fn sample_request() -> IngestRequest { + // Use a timestamp close to "now" so messages are not pruned by the + // 90-day auto-prune horizon. We derive it from the system clock + // minus one hour so even on slow CI boxes the message is comfortably + // within the retention window. + let recent_ts = std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .map(|d| d.as_secs() as i64 - 3600) + .unwrap_or(1_750_000_000); + let mut chats = HashMap::new(); + chats.insert( + "alice@c.us".to_string(), + ChatMeta { + name: Some("Alice".to_string()), + }, + ); + IngestRequest { + account_id: "acct1".to_string(), + chats, + messages: vec![IngestMessage { + message_id: "msg1".to_string(), + chat_id: "alice@c.us".to_string(), + sender: Some("Alice".to_string()), + sender_jid: None, + from_me: Some(false), + body: Some("Hello!".to_string()), + timestamp: Some(recent_ts), + message_type: Some("chat".to_string()), + source: Some("cdp-dom".to_string()), + }], + } + } + + #[test] + fn ingest_returns_correct_counts() { + let (store, _tmp) = make_store(); + let result = ingest(&store, sample_request()).unwrap(); + assert_eq!(result.chats_upserted, 1); + assert_eq!(result.messages_upserted, 1); + } + + #[test] + fn list_chats_after_ingest() { + let (store, _tmp) = make_store(); + ingest(&store, sample_request()).unwrap(); + + let chats = list_chats( + &store, + ListChatsRequest { + account_id: None, + limit: None, + offset: None, + }, + ) + .unwrap(); + assert_eq!(chats.len(), 1); + assert_eq!(chats[0].chat_id, "alice@c.us"); + } + + #[test] + fn list_messages_after_ingest() { + let (store, _tmp) = make_store(); + ingest(&store, sample_request()).unwrap(); + + let msgs = list_messages( + &store, + ListMessagesRequest { + chat_id: "alice@c.us".to_string(), + account_id: None, + since_ts: None, + until_ts: None, + limit: None, + offset: None, + }, + ) + .unwrap(); + assert_eq!(msgs.len(), 1); + assert_eq!(msgs[0].body, "Hello!"); + } + + #[test] + fn search_messages_after_ingest() { + let (store, _tmp) = make_store(); + ingest(&store, sample_request()).unwrap(); + + let results = search_messages( + &store, + SearchMessagesRequest { + query: "Hello".to_string(), + chat_id: None, + account_id: None, + limit: None, + }, + ) + .unwrap(); + assert_eq!(results.len(), 1); + } +} diff --git a/src/openhuman/whatsapp_data/rpc.rs b/src/openhuman/whatsapp_data/rpc.rs new file mode 100644 index 000000000..b6a2d18d5 --- /dev/null +++ b/src/openhuman/whatsapp_data/rpc.rs @@ -0,0 +1,122 @@ +//! RPC handler functions for WhatsApp data domain. +//! +//! Each function: +//! 1. Acquires the global `WhatsAppDataStore`. +//! 2. Delegates to `ops::*` for business logic. +//! 3. Returns an `RpcOutcome`. +//! +//! When no WhatsApp session is active (store not yet initialised), the +//! handlers return an actionable "not connected" error so the agent can +//! surface a useful message instead of a crash. + +use anyhow::Result; + +use crate::openhuman::whatsapp_data::{ + global, ops, + types::{ + IngestRequest, IngestResult, ListChatsRequest, ListMessagesRequest, SearchMessagesRequest, + WhatsAppChat, WhatsAppMessage, + }, +}; +use crate::rpc::RpcOutcome; + +/// Ensure the global store is initialised. +/// +/// On first call after core startup this may lazily initialise using the +/// default workspace path. For the scanner-side ingest path the store is +/// already warm from the `core_server` startup sequence. +fn require_store() -> Result { + global::store() +} + +/// Ingest a WhatsApp scanner snapshot. +/// +/// Called by the Tauri whatsapp_scanner after each full CDP scan tick. +pub async fn whatsapp_data_ingest(req: IngestRequest) -> Result, String> { + log::debug!( + "[whatsapp_data][rpc] ingest enter chats={} messages={} (account redacted)", + req.chats.len(), + req.messages.len() + ); + let store = require_store()?; + let result = ops::ingest(&store, req).map_err(|e| { + log::warn!("[whatsapp_data][rpc] ingest error: {e}"); + format!("[whatsapp_data] ingest failed: {e}") + })?; + log::debug!( + "[whatsapp_data][rpc] ingest ok chats={} messages={} pruned={}", + result.chats_upserted, + result.messages_upserted, + result.messages_pruned + ); + Ok(RpcOutcome::single_log( + result, + "whatsapp_data ingest complete", + )) +} + +/// List WhatsApp chats, optionally filtered by account. +pub async fn whatsapp_data_list_chats( + req: ListChatsRequest, +) -> Result>, String> { + log::debug!( + "[whatsapp_data][rpc] list_chats enter has_account={} limit={:?} offset={:?}", + req.account_id.is_some(), + req.limit, + req.offset + ); + let store = require_store()?; + let chats = ops::list_chats(&store, req).map_err(|e| { + log::warn!("[whatsapp_data][rpc] list_chats error: {e}"); + format!("[whatsapp_data] list_chats failed: {e}") + })?; + log::debug!("[whatsapp_data][rpc] list_chats ok count={}", chats.len()); + Ok(RpcOutcome::single_log( + chats, + "whatsapp_data list_chats complete", + )) +} + +/// List messages for a chat, with optional time range and pagination. +pub async fn whatsapp_data_list_messages( + req: ListMessagesRequest, +) -> Result>, String> { + log::debug!( + "[whatsapp_data][rpc] list_messages enter has_account={} (chat redacted)", + req.account_id.is_some() + ); + let store = require_store()?; + let msgs = ops::list_messages(&store, req).map_err(|e| { + log::warn!("[whatsapp_data][rpc] list_messages error: {e}"); + format!("[whatsapp_data] list_messages failed: {e}") + })?; + log::debug!("[whatsapp_data][rpc] list_messages ok count={}", msgs.len()); + Ok(RpcOutcome::single_log( + msgs, + "whatsapp_data list_messages complete", + )) +} + +/// Full-text search over message bodies. +pub async fn whatsapp_data_search_messages( + req: SearchMessagesRequest, +) -> Result>, String> { + log::debug!( + "[whatsapp_data][rpc] search_messages enter has_account={} has_chat={} (query/identifiers redacted)", + req.account_id.is_some(), + req.chat_id.is_some() + ); + let store = require_store()?; + let results = ops::search_messages(&store, req).map_err(|e| { + log::warn!("[whatsapp_data][rpc] search_messages error: {e}"); + format!("[whatsapp_data] search_messages failed: {e}") + })?; + log::debug!( + "[whatsapp_data][rpc] search_messages ok count={}", + results.len() + ); + Ok(RpcOutcome::single_log( + results, + "whatsapp_data search_messages complete", + )) +} diff --git a/src/openhuman/whatsapp_data/schemas.rs b/src/openhuman/whatsapp_data/schemas.rs new file mode 100644 index 000000000..9caede624 --- /dev/null +++ b/src/openhuman/whatsapp_data/schemas.rs @@ -0,0 +1,263 @@ +//! Controller schemas and handler dispatch for the `whatsapp_data` namespace. +//! +//! Agent-facing (read-only) RPC methods: +//! - `openhuman.whatsapp_data_list_chats` +//! - `openhuman.whatsapp_data_list_messages` +//! - `openhuman.whatsapp_data_search_messages` +//! +//! Internal write path (NOT exposed to the agent controller registry): +//! - `openhuman.whatsapp_data_ingest` — called by the Tauri scanner only +//! +//! Keeping ingest off the agent-facing registry prevents an agent from +//! mutating or poisoning the local WhatsApp store directly. + +use serde::de::DeserializeOwned; +use serde_json::{Map, Value}; + +use crate::core::all::{ControllerFuture, RegisteredController}; +use crate::core::{ControllerSchema, FieldSchema, TypeSchema}; +use crate::rpc::RpcOutcome; + +/// Returns controller schemas advertised to the agent (read-only subset). +/// The ingest schema is intentionally excluded — it is an internal write path +/// called by the scanner, not something the agent should be able to invoke. +pub fn all_controller_schemas() -> Vec { + vec![ + schemas("list_chats"), + schemas("list_messages"), + schemas("search_messages"), + ] +} + +/// Returns registered controllers for the agent-facing dispatcher (read-only). +/// The ingest handler is registered separately via `all_internal_controllers()` +/// and wired by the scanner — not through the agent controller registry. +pub fn all_registered_controllers() -> Vec { + vec![ + RegisteredController { + schema: schemas("list_chats"), + handler: handle_list_chats, + }, + RegisteredController { + schema: schemas("list_messages"), + handler: handle_list_messages, + }, + RegisteredController { + schema: schemas("search_messages"), + handler: handle_search_messages, + }, + ] +} + +/// Returns the full controller set including the internal ingest handler. +/// Used by the core RPC dispatcher so the scanner can call +/// `openhuman.whatsapp_data_ingest` over JSON-RPC without exposing it to agents. +pub fn all_internal_controllers() -> Vec { + let mut controllers = all_registered_controllers(); + controllers.insert( + 0, + RegisteredController { + schema: schemas("ingest"), + handler: handle_ingest, + }, + ); + controllers +} + +pub fn schemas(function: &str) -> ControllerSchema { + match function { + "ingest" => ControllerSchema { + namespace: "whatsapp_data", + function: "ingest", + description: "Ingest a WhatsApp Web scanner snapshot (chats + messages). \ + Called by the Tauri scanner after each CDP tick. Data is stored \ + locally only — never transmitted externally.", + inputs: vec![ + required_string("account_id", "WhatsApp account identifier (phone JID)."), + FieldSchema { + name: "chats", + ty: TypeSchema::Json, + comment: "Map of chat JID → {name: string | null}.", + required: true, + }, + FieldSchema { + name: "messages", + ty: TypeSchema::Array(Box::new(TypeSchema::Json)), + comment: "Array of message objects to upsert.", + required: true, + }, + ], + outputs: vec![FieldSchema { + name: "result", + ty: TypeSchema::Json, + comment: "Ingest summary: chats_upserted, messages_upserted, messages_pruned.", + required: true, + }], + }, + "list_chats" => ControllerSchema { + namespace: "whatsapp_data", + function: "list_chats", + description: "List locally-stored WhatsApp chats, ordered by most recent message. \ + When account_id is omitted, chats from all accounts are returned.", + inputs: vec![ + optional_string("account_id", "Filter to a specific WhatsApp account."), + FieldSchema { + name: "limit", + ty: TypeSchema::Option(Box::new(TypeSchema::U64)), + comment: "Maximum results (default 50).", + required: false, + }, + FieldSchema { + name: "offset", + ty: TypeSchema::Option(Box::new(TypeSchema::U64)), + comment: "Pagination offset (default 0).", + required: false, + }, + ], + outputs: vec![FieldSchema { + name: "chats", + ty: TypeSchema::Array(Box::new(TypeSchema::Json)), + comment: "Array of WhatsAppChat objects.", + required: true, + }], + }, + "list_messages" => ControllerSchema { + namespace: "whatsapp_data", + function: "list_messages", + description: "List messages for a WhatsApp chat, ordered by timestamp ascending. \ + When account_id is omitted, messages from all accounts are included.", + inputs: vec![ + required_string("chat_id", "JID of the chat to retrieve messages for."), + optional_string("account_id", "Filter to a specific WhatsApp account."), + FieldSchema { + name: "since_ts", + ty: TypeSchema::Option(Box::new(TypeSchema::I64)), + comment: "Only return messages at or after this Unix timestamp (seconds).", + required: false, + }, + FieldSchema { + name: "until_ts", + ty: TypeSchema::Option(Box::new(TypeSchema::I64)), + comment: "Only return messages at or before this Unix timestamp (seconds).", + required: false, + }, + FieldSchema { + name: "limit", + ty: TypeSchema::Option(Box::new(TypeSchema::U64)), + comment: "Maximum results (default 100).", + required: false, + }, + FieldSchema { + name: "offset", + ty: TypeSchema::Option(Box::new(TypeSchema::U64)), + comment: "Pagination offset (default 0).", + required: false, + }, + ], + outputs: vec![FieldSchema { + name: "messages", + ty: TypeSchema::Array(Box::new(TypeSchema::Json)), + comment: "Array of WhatsAppMessage objects.", + required: true, + }], + }, + "search_messages" => ControllerSchema { + namespace: "whatsapp_data", + function: "search_messages", + description: + "Search locally-stored WhatsApp message bodies. \ + Case-insensitive substring match. \ + When account_id / chat_id are omitted, all accounts / chats are searched.", + inputs: vec![ + required_string("query", "Search query matched against message bodies."), + optional_string("chat_id", "Restrict search to a specific chat JID."), + optional_string( + "account_id", + "Restrict search to a specific WhatsApp account.", + ), + FieldSchema { + name: "limit", + ty: TypeSchema::Option(Box::new(TypeSchema::U64)), + comment: "Maximum results (default 20).", + required: false, + }, + ], + outputs: vec![FieldSchema { + name: "messages", + ty: TypeSchema::Array(Box::new(TypeSchema::Json)), + comment: "Array of WhatsAppMessage objects matching the query.", + required: true, + }], + }, + _ => ControllerSchema { + namespace: "whatsapp_data", + function: "unknown", + description: "Unknown whatsapp_data controller function.", + inputs: vec![], + outputs: vec![FieldSchema { + name: "error", + ty: TypeSchema::String, + comment: "Lookup error details.", + required: true, + }], + }, + } +} + +// ── Handlers ──────────────────────────────────────────────────────────────── + +fn handle_ingest(params: Map) -> ControllerFuture { + Box::pin(async move { + let req = deserialize_params(params)?; + to_json(crate::openhuman::whatsapp_data::rpc::whatsapp_data_ingest(req).await?) + }) +} + +fn handle_list_chats(params: Map) -> ControllerFuture { + Box::pin(async move { + let req = deserialize_params(params)?; + to_json(crate::openhuman::whatsapp_data::rpc::whatsapp_data_list_chats(req).await?) + }) +} + +fn handle_list_messages(params: Map) -> ControllerFuture { + Box::pin(async move { + let req = deserialize_params(params)?; + to_json(crate::openhuman::whatsapp_data::rpc::whatsapp_data_list_messages(req).await?) + }) +} + +fn handle_search_messages(params: Map) -> ControllerFuture { + Box::pin(async move { + let req = deserialize_params(params)?; + to_json(crate::openhuman::whatsapp_data::rpc::whatsapp_data_search_messages(req).await?) + }) +} + +// ── Helpers ────────────────────────────────────────────────────────────────── + +fn deserialize_params(params: Map) -> Result { + serde_json::from_value(Value::Object(params)).map_err(|e| format!("invalid params: {e}")) +} + +fn to_json(outcome: RpcOutcome) -> Result { + outcome.into_cli_compatible_json() +} + +fn required_string(name: &'static str, comment: &'static str) -> FieldSchema { + FieldSchema { + name, + ty: TypeSchema::String, + comment, + required: true, + } +} + +fn optional_string(name: &'static str, comment: &'static str) -> FieldSchema { + FieldSchema { + name, + ty: TypeSchema::Option(Box::new(TypeSchema::String)), + comment, + required: false, + } +} diff --git a/src/openhuman/whatsapp_data/store.rs b/src/openhuman/whatsapp_data/store.rs new file mode 100644 index 000000000..cea314b46 --- /dev/null +++ b/src/openhuman/whatsapp_data/store.rs @@ -0,0 +1,662 @@ +//! SQLite-backed persistence for structured WhatsApp Web data. +//! +//! Data is stored in a dedicated `whatsapp_data.db` file inside the +//! workspace directory. Tables: `wa_chats` and `wa_messages`. +//! +//! This store is local-only; no data is transmitted to external services. + +use std::collections::HashMap; +use std::path::Path; + +use anyhow::{Context, Result}; +use rusqlite::{params, Connection}; + +use crate::openhuman::whatsapp_data::types::{ + ChatMeta, IngestMessage, ListChatsRequest, ListMessagesRequest, SearchMessagesRequest, + WhatsAppChat, WhatsAppMessage, +}; + +/// SQLite-backed store for WhatsApp chats and messages. +pub struct WhatsAppDataStore { + db_path: std::path::PathBuf, +} + +impl WhatsAppDataStore { + /// Open or create the `whatsapp_data.db` SQLite database in `workspace_dir`. + /// The directory (and any parents) are created if they do not exist. + pub fn new(workspace_dir: &Path) -> Result { + let db_path = workspace_dir.join("whatsapp_data").join("whatsapp_data.db"); + if let Some(parent) = db_path.parent() { + std::fs::create_dir_all(parent) + .with_context(|| format!("create whatsapp_data dir: {}", parent.display()))?; + } + log::debug!("[whatsapp_data] opening store at {}", db_path.display()); + let store = Self { db_path }; + store.init_schema()?; + Ok(store) + } + + /// Initialize the schema. Idempotent — safe to call on every startup. + fn init_schema(&self) -> Result<()> { + let conn = self.open_conn()?; + conn.execute_batch( + "PRAGMA journal_mode = WAL; + PRAGMA foreign_keys = ON; + + CREATE TABLE IF NOT EXISTS wa_chats ( + account_id TEXT NOT NULL, + chat_id TEXT NOT NULL, + display_name TEXT NOT NULL DEFAULT '', + is_group INTEGER NOT NULL DEFAULT 0, + last_message_ts INTEGER NOT NULL DEFAULT 0, + message_count INTEGER NOT NULL DEFAULT 0, + updated_at INTEGER NOT NULL DEFAULT 0, + PRIMARY KEY (account_id, chat_id) + ); + + CREATE TABLE IF NOT EXISTS wa_messages ( + account_id TEXT NOT NULL, + chat_id TEXT NOT NULL, + message_id TEXT NOT NULL, + sender TEXT NOT NULL DEFAULT '', + sender_jid TEXT, + from_me INTEGER NOT NULL DEFAULT 0, + body TEXT NOT NULL DEFAULT '', + timestamp INTEGER NOT NULL DEFAULT 0, + message_type TEXT, + source TEXT NOT NULL DEFAULT '', + ingested_at INTEGER NOT NULL DEFAULT 0, + PRIMARY KEY (account_id, chat_id, message_id) + ); + CREATE INDEX IF NOT EXISTS idx_wa_msg_ts ON wa_messages(account_id, chat_id, timestamp); + CREATE INDEX IF NOT EXISTS idx_wa_msg_body ON wa_messages(account_id, body);", + ) + .context("init whatsapp_data schema")?; + log::debug!("[whatsapp_data] schema ready"); + Ok(()) + } + + fn open_conn(&self) -> Result { + Connection::open(&self.db_path) + .with_context(|| format!("open whatsapp_data db: {}", self.db_path.display())) + } + + fn now_secs() -> i64 { + std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .map(|d| d.as_secs() as i64) + .unwrap_or(0) + } + + /// Upsert chat metadata rows. Returns the number of rows inserted or updated. + pub fn upsert_chats( + &self, + account_id: &str, + chats: &HashMap, + ) -> Result { + if chats.is_empty() { + return Ok(0); + } + let conn = self.open_conn()?; + let now = Self::now_secs(); + let mut count = 0usize; + for (chat_id, meta) in chats { + let name = meta.name.as_deref().unwrap_or(""); + let is_group = chat_id.ends_with("@g.us") as i64; + conn.execute( + "INSERT INTO wa_chats (account_id, chat_id, display_name, is_group, updated_at) + VALUES (?1, ?2, ?3, ?4, ?5) + ON CONFLICT(account_id, chat_id) DO UPDATE SET + display_name = CASE WHEN excluded.display_name != '' THEN excluded.display_name ELSE display_name END, + is_group = excluded.is_group, + updated_at = excluded.updated_at", + params![account_id, chat_id, name, is_group, now], + ) + .with_context(|| format!("upsert wa_chat {chat_id}"))?; + count += 1; + } + log::debug!( + "[whatsapp_data] upserted {} chats (account redacted)", + count + ); + Ok(count) + } + + /// Upsert message rows. Returns the number of rows inserted or updated. + pub fn upsert_messages(&self, account_id: &str, msgs: &[IngestMessage]) -> Result { + if msgs.is_empty() { + return Ok(0); + } + let conn = self.open_conn()?; + let now = Self::now_secs(); + let mut count = 0usize; + for m in msgs { + if m.message_id.is_empty() || m.chat_id.is_empty() { + continue; + } + // Persist all messages, including non-text ones (stickers, images, + // system events). Dropping empty-body rows biases message_count + // and last_message_ts to text-only messages, making active chats + // look stale whenever the latest event has no body. + let body = m.body.as_deref().unwrap_or(""); + let ts = m.timestamp.unwrap_or(0); + let from_me = m.from_me.unwrap_or(false) as i64; + conn.execute( + "INSERT INTO wa_messages + (account_id, chat_id, message_id, sender, sender_jid, from_me, + body, timestamp, message_type, source, ingested_at) + VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, ?11) + ON CONFLICT(account_id, chat_id, message_id) DO UPDATE SET + sender = CASE WHEN excluded.sender != '' THEN excluded.sender ELSE sender END, + sender_jid = COALESCE(excluded.sender_jid, sender_jid), + from_me = excluded.from_me, + body = CASE WHEN excluded.body != '' THEN excluded.body ELSE body END, + timestamp = excluded.timestamp, + message_type = COALESCE(excluded.message_type, message_type), + source = excluded.source, + ingested_at = excluded.ingested_at", + params![ + account_id, + m.chat_id, + m.message_id, + m.sender.as_deref().unwrap_or(""), + m.sender_jid.as_deref(), + from_me, + body, + ts, + m.message_type.as_deref(), + m.source.as_deref().unwrap_or(""), + now, + ], + ) + .with_context(|| { + format!( + "upsert wa_message chat={} msg={}", + m.chat_id, m.message_id + ) + })?; + count += 1; + } + + // Refresh chat stats after message upsert. + if count > 0 { + conn.execute( + "UPDATE wa_chats + SET message_count = (SELECT COUNT(*) FROM wa_messages + WHERE wa_messages.account_id = wa_chats.account_id + AND wa_messages.chat_id = wa_chats.chat_id), + last_message_ts = COALESCE( + (SELECT MAX(timestamp) FROM wa_messages + WHERE wa_messages.account_id = wa_chats.account_id + AND wa_messages.chat_id = wa_chats.chat_id), + last_message_ts), + updated_at = ?1 + WHERE account_id = ?2", + rusqlite::params![now, account_id], + ) + .context("refresh wa_chats stats")?; + } + + log::debug!( + "[whatsapp_data] upserted {} messages (account redacted)", + count + ); + Ok(count) + } + + /// Delete messages older than `cutoff_ts` (Unix seconds). Returns the count removed. + /// + /// After the delete, refreshes `wa_chats.message_count` and + /// `last_message_ts` for every chat that lost rows, so `list_chats` + /// returns accurate counts and ordering immediately. + pub fn prune_old_messages(&self, cutoff_ts: i64) -> Result { + let conn = self.open_conn()?; + let now = Self::now_secs(); + + // Collect affected (account_id, chat_id) pairs before deleting. + let mut stmt = conn.prepare( + "SELECT DISTINCT account_id, chat_id FROM wa_messages + WHERE timestamp > 0 AND timestamp < ?1", + )?; + let affected: Vec<(String, String)> = stmt + .query_map(params![cutoff_ts], |row| Ok((row.get(0)?, row.get(1)?)))? + .collect::>() + .context("collect affected chats for prune")?; + + let changed = conn + .execute( + "DELETE FROM wa_messages WHERE timestamp > 0 AND timestamp < ?1", + params![cutoff_ts], + ) + .context("prune old wa_messages")?; + + // Refresh aggregate stats for every affected chat so list_chats + // reflects the post-prune state immediately. + if changed > 0 { + for (acct, chat_id) in &affected { + conn.execute( + "UPDATE wa_chats + SET message_count = (SELECT COUNT(*) FROM wa_messages + WHERE account_id = wa_chats.account_id + AND chat_id = wa_chats.chat_id), + last_message_ts = COALESCE( + (SELECT MAX(timestamp) FROM wa_messages + WHERE account_id = wa_chats.account_id + AND chat_id = wa_chats.chat_id), + last_message_ts), + updated_at = ?3 + WHERE account_id = ?1 AND chat_id = ?2", + params![acct, chat_id, now], + ) + .with_context(|| format!("refresh chat stats after prune: {chat_id}"))?; + } + log::debug!( + "[whatsapp_data] pruned {} messages (affected {} chats)", + changed, + affected.len() + ); + } + Ok(changed as u64) + } + + /// List chats, optionally filtered by account. Ordered by `last_message_ts` DESC. + pub fn list_chats(&self, req: &ListChatsRequest) -> Result> { + let conn = self.open_conn()?; + let limit = req.limit.unwrap_or(50) as i64; + let offset = req.offset.unwrap_or(0) as i64; + + let chats = if let Some(ref acct) = req.account_id { + let mut stmt = conn.prepare( + "SELECT account_id, chat_id, display_name, is_group, last_message_ts, + message_count, updated_at + FROM wa_chats + WHERE account_id = ?1 + ORDER BY last_message_ts DESC + LIMIT ?2 OFFSET ?3", + )?; + let rows = stmt + .query_map(params![acct, limit, offset], map_chat_row)? + .collect::>>() + .context("list chats (filtered)")?; + rows + } else { + let mut stmt = conn.prepare( + "SELECT account_id, chat_id, display_name, is_group, last_message_ts, + message_count, updated_at + FROM wa_chats + ORDER BY last_message_ts DESC + LIMIT ?1 OFFSET ?2", + )?; + let rows = stmt + .query_map(params![limit, offset], map_chat_row)? + .collect::>>() + .context("list chats (all)")?; + rows + }; + log::debug!("[whatsapp_data] list_chats returned {} rows", chats.len()); + Ok(chats) + } + + /// List messages for a chat, with optional time range and pagination. + pub fn list_messages(&self, req: &ListMessagesRequest) -> Result> { + let conn = self.open_conn()?; + let limit = req.limit.unwrap_or(100) as i64; + let offset = req.offset.unwrap_or(0) as i64; + let since_ts = req.since_ts.unwrap_or(0); + let until_ts = req.until_ts.unwrap_or(i64::MAX); + + let msgs = if let Some(ref acct) = req.account_id { + let mut stmt = conn.prepare( + "SELECT account_id, chat_id, message_id, sender, sender_jid, from_me, + body, timestamp, message_type, source + FROM wa_messages + WHERE account_id = ?1 + AND chat_id = ?2 + AND timestamp >= ?3 + AND timestamp <= ?4 + ORDER BY timestamp ASC + LIMIT ?5 OFFSET ?6", + )?; + let rows = stmt + .query_map( + params![acct, req.chat_id, since_ts, until_ts, limit, offset], + map_message_row, + )? + .collect::>>() + .context("list messages (filtered by account)")?; + rows + } else { + let mut stmt = conn.prepare( + "SELECT account_id, chat_id, message_id, sender, sender_jid, from_me, + body, timestamp, message_type, source + FROM wa_messages + WHERE chat_id = ?1 + AND timestamp >= ?2 + AND timestamp <= ?3 + ORDER BY timestamp ASC + LIMIT ?4 OFFSET ?5", + )?; + let rows = stmt + .query_map( + params![req.chat_id, since_ts, until_ts, limit, offset], + map_message_row, + )? + .collect::>>() + .context("list messages (all accounts)")?; + rows + }; + log::debug!( + "[whatsapp_data] list_messages returned {} rows (chat/account redacted)", + msgs.len() + ); + Ok(msgs) + } + + /// Full-text search over message bodies (case-insensitive LIKE). + pub fn search_messages(&self, req: &SearchMessagesRequest) -> Result> { + if req.query.trim().is_empty() { + return Ok(vec![]); + } + let conn = self.open_conn()?; + let limit = req.limit.unwrap_or(20) as i64; + let pattern = format!("%{}%", req.query.replace('%', "\\%").replace('_', "\\_")); + + // Build the query dynamically depending on optional filters. + // Each branch binds to a local `rows` variable so `stmt` is dropped + // before the result is returned (fixes E0597 borrow lifetimes). + let msgs: Vec = match (&req.account_id, &req.chat_id) { + (Some(acct), Some(chat_id)) => { + let mut stmt = conn.prepare( + "SELECT account_id, chat_id, message_id, sender, sender_jid, from_me, + body, timestamp, message_type, source + FROM wa_messages + WHERE account_id = ?1 + AND chat_id = ?2 + AND body LIKE ?3 ESCAPE '\\' + ORDER BY timestamp DESC + LIMIT ?4", + )?; + let rows = stmt + .query_map(params![acct, chat_id, pattern, limit], map_message_row)? + .collect::>>() + .context("search messages (account+chat)")?; + rows + } + (Some(acct), None) => { + let mut stmt = conn.prepare( + "SELECT account_id, chat_id, message_id, sender, sender_jid, from_me, + body, timestamp, message_type, source + FROM wa_messages + WHERE account_id = ?1 + AND body LIKE ?2 ESCAPE '\\' + ORDER BY timestamp DESC + LIMIT ?3", + )?; + let rows = stmt + .query_map(params![acct, pattern, limit], map_message_row)? + .collect::>>() + .context("search messages (account)")?; + rows + } + (None, Some(chat_id)) => { + let mut stmt = conn.prepare( + "SELECT account_id, chat_id, message_id, sender, sender_jid, from_me, + body, timestamp, message_type, source + FROM wa_messages + WHERE chat_id = ?1 + AND body LIKE ?2 ESCAPE '\\' + ORDER BY timestamp DESC + LIMIT ?3", + )?; + let rows = stmt + .query_map(params![chat_id, pattern, limit], map_message_row)? + .collect::>>() + .context("search messages (chat)")?; + rows + } + (None, None) => { + let mut stmt = conn.prepare( + "SELECT account_id, chat_id, message_id, sender, sender_jid, from_me, + body, timestamp, message_type, source + FROM wa_messages + WHERE body LIKE ?1 ESCAPE '\\' + ORDER BY timestamp DESC + LIMIT ?2", + )?; + let rows = stmt + .query_map(params![pattern, limit], map_message_row)? + .collect::>>() + .context("search messages (all)")?; + rows + } + }; + log::debug!( + "[whatsapp_data] search_messages returned {} rows (query/account redacted)", + msgs.len() + ); + Ok(msgs) + } +} + +fn map_chat_row(row: &rusqlite::Row<'_>) -> rusqlite::Result { + Ok(WhatsAppChat { + account_id: row.get(0)?, + chat_id: row.get(1)?, + display_name: row.get(2)?, + is_group: row.get::<_, i64>(3)? != 0, + last_message_ts: row.get(4)?, + message_count: row.get::<_, i64>(5)? as u32, + updated_at: row.get(6)?, + }) +} + +fn map_message_row(row: &rusqlite::Row<'_>) -> rusqlite::Result { + Ok(WhatsAppMessage { + account_id: row.get(0)?, + chat_id: row.get(1)?, + message_id: row.get(2)?, + sender: row.get(3)?, + sender_jid: row.get(4)?, + from_me: row.get::<_, i64>(5)? != 0, + body: row.get(6)?, + timestamp: row.get(7)?, + message_type: row.get(8)?, + source: row.get(9)?, + }) +} + +#[cfg(test)] +mod tests { + use super::*; + use tempfile::tempdir; + + fn make_store() -> (WhatsAppDataStore, tempfile::TempDir) { + let tmp = tempdir().expect("tempdir"); + let store = WhatsAppDataStore::new(tmp.path()).expect("store"); + (store, tmp) + } + + #[test] + fn upsert_and_list_chats() { + let (store, _tmp) = make_store(); + let mut chats = HashMap::new(); + chats.insert( + "chat1@c.us".to_string(), + ChatMeta { + name: Some("Alice".to_string()), + }, + ); + chats.insert( + "group1@g.us".to_string(), + ChatMeta { + name: Some("My Group".to_string()), + }, + ); + let count = store.upsert_chats("acct1", &chats).unwrap(); + assert_eq!(count, 2); + + let req = ListChatsRequest { + account_id: Some("acct1".to_string()), + limit: None, + offset: None, + }; + let rows = store.list_chats(&req).unwrap(); + assert_eq!(rows.len(), 2); + + let group = rows.iter().find(|c| c.chat_id == "group1@g.us").unwrap(); + assert!(group.is_group); + let dm = rows.iter().find(|c| c.chat_id == "chat1@c.us").unwrap(); + assert!(!dm.is_group); + } + + #[test] + fn upsert_and_list_messages() { + let (store, _tmp) = make_store(); + let mut chats = HashMap::new(); + chats.insert( + "chat1@c.us".to_string(), + ChatMeta { + name: Some("Alice".to_string()), + }, + ); + store.upsert_chats("acct1", &chats).unwrap(); + + let msgs = vec![ + IngestMessage { + message_id: "msg1".to_string(), + chat_id: "chat1@c.us".to_string(), + sender: Some("Alice".to_string()), + sender_jid: None, + from_me: Some(false), + body: Some("Hello there".to_string()), + timestamp: Some(1_700_000_000), + message_type: Some("chat".to_string()), + source: Some("cdp-dom".to_string()), + }, + IngestMessage { + message_id: "msg2".to_string(), + chat_id: "chat1@c.us".to_string(), + sender: Some("me".to_string()), + sender_jid: None, + from_me: Some(true), + body: Some("Hey!".to_string()), + timestamp: Some(1_700_000_100), + message_type: Some("chat".to_string()), + source: Some("cdp-indexeddb".to_string()), + }, + ]; + let count = store.upsert_messages("acct1", &msgs).unwrap(); + assert_eq!(count, 2); + + let req = ListMessagesRequest { + chat_id: "chat1@c.us".to_string(), + account_id: Some("acct1".to_string()), + since_ts: None, + until_ts: None, + limit: None, + offset: None, + }; + let rows = store.list_messages(&req).unwrap(); + assert_eq!(rows.len(), 2); + assert_eq!(rows[0].body, "Hello there"); + assert_eq!(rows[1].body, "Hey!"); + } + + #[test] + fn search_messages_finds_match() { + let (store, _tmp) = make_store(); + let mut chats = HashMap::new(); + chats.insert( + "chat1@c.us".to_string(), + ChatMeta { + name: Some("Alice".to_string()), + }, + ); + store.upsert_chats("acct1", &chats).unwrap(); + + let msgs = vec![ + IngestMessage { + message_id: "m1".to_string(), + chat_id: "chat1@c.us".to_string(), + sender: Some("Alice".to_string()), + sender_jid: None, + from_me: Some(false), + body: Some("Can you bring the umbrella?".to_string()), + timestamp: Some(1_700_000_000), + message_type: None, + source: Some("cdp-dom".to_string()), + }, + IngestMessage { + message_id: "m2".to_string(), + chat_id: "chat1@c.us".to_string(), + sender: Some("me".to_string()), + sender_jid: None, + from_me: Some(true), + body: Some("Sure, no problem".to_string()), + timestamp: Some(1_700_000_200), + message_type: None, + source: Some("cdp-dom".to_string()), + }, + ]; + store.upsert_messages("acct1", &msgs).unwrap(); + + let req = SearchMessagesRequest { + query: "umbrella".to_string(), + chat_id: None, + account_id: None, + limit: None, + }; + let results = store.search_messages(&req).unwrap(); + assert_eq!(results.len(), 1); + assert!(results[0].body.contains("umbrella")); + } + + #[test] + fn prune_removes_old_messages() { + let (store, _tmp) = make_store(); + let mut chats = HashMap::new(); + chats.insert("chat1@c.us".to_string(), ChatMeta { name: None }); + store.upsert_chats("acct1", &chats).unwrap(); + + let msgs = vec![ + IngestMessage { + message_id: "old".to_string(), + chat_id: "chat1@c.us".to_string(), + sender: None, + sender_jid: None, + from_me: Some(false), + body: Some("Old message".to_string()), + timestamp: Some(1_000_000), + message_type: None, + source: None, + }, + IngestMessage { + message_id: "new".to_string(), + chat_id: "chat1@c.us".to_string(), + sender: None, + sender_jid: None, + from_me: Some(false), + body: Some("New message".to_string()), + timestamp: Some(2_000_000_000), + message_type: None, + source: None, + }, + ]; + store.upsert_messages("acct1", &msgs).unwrap(); + + let pruned = store.prune_old_messages(1_500_000_000).unwrap(); + assert_eq!(pruned, 1); + + let req = ListMessagesRequest { + chat_id: "chat1@c.us".to_string(), + account_id: None, + since_ts: None, + until_ts: None, + limit: None, + offset: None, + }; + let remaining = store.list_messages(&req).unwrap(); + assert_eq!(remaining.len(), 1); + assert_eq!(remaining[0].message_id, "new"); + } +} diff --git a/src/openhuman/whatsapp_data/types.rs b/src/openhuman/whatsapp_data/types.rs new file mode 100644 index 000000000..a8b500396 --- /dev/null +++ b/src/openhuman/whatsapp_data/types.rs @@ -0,0 +1,134 @@ +//! Normalized WhatsApp data structures — local-only, never transmitted externally. +//! +//! These types represent the structured data extracted from WhatsApp Web via CDP and +//! persisted in a local SQLite database. All data remains local; nothing is sent to +//! any remote service. + +use std::collections::HashMap; + +use serde::{Deserialize, Serialize}; + +/// A WhatsApp chat (conversation) record stored locally. +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct WhatsAppChat { + /// JID e.g. "123456@c.us" or "group@g.us" + pub chat_id: String, + /// Human-readable display name from WhatsApp contacts/group metadata. + pub display_name: String, + /// True if this chat is a group conversation. + pub is_group: bool, + /// The connected WhatsApp account identifier. + pub account_id: String, + /// Unix timestamp (seconds) of the most recent message stored. + pub last_message_ts: i64, + /// Number of messages stored for this chat. + pub message_count: u32, + /// Unix timestamp (seconds) when this record was last updated. + pub updated_at: i64, +} + +/// A single WhatsApp message record stored locally. +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct WhatsAppMessage { + /// WhatsApp message identifier (compound or bare form). + pub message_id: String, + /// JID of the chat this message belongs to. + pub chat_id: String, + /// Display name of the sender (stored as-is from WhatsApp). + pub sender: String, + /// JID of the sender, when available from IDB metadata. + pub sender_jid: Option, + /// True if the message was sent by the account owner. + pub from_me: bool, + /// Decrypted message body text. + pub body: String, + /// Unix timestamp (seconds) of the message. + pub timestamp: i64, + /// WhatsApp message type (e.g. "chat", "image", "sticker"). + pub message_type: Option, + /// The connected WhatsApp account identifier. + pub account_id: String, + /// Data source: "cdp-dom" or "cdp-indexeddb". + pub source: String, +} + +/// Metadata about a single chat in an ingest payload. +#[derive(Debug, Deserialize)] +pub struct ChatMeta { + /// Display name for the chat, if available. + pub name: Option, +} + +/// A single message entry in an ingest payload. +#[derive(Debug, Deserialize)] +pub struct IngestMessage { + pub message_id: String, + pub chat_id: String, + pub sender: Option, + pub sender_jid: Option, + pub from_me: Option, + pub body: Option, + pub timestamp: Option, + pub message_type: Option, + pub source: Option, +} + +/// Request payload for `openhuman.whatsapp_data_ingest`. +#[derive(Debug, Deserialize)] +pub struct IngestRequest { + /// The WhatsApp account identifier (usually the phone JID). + pub account_id: String, + /// Map of chat JID → chat metadata (display name, etc.). + pub chats: HashMap, + /// Messages to upsert into the local store. + pub messages: Vec, +} + +/// Summary result returned after an ingest operation. +#[derive(Debug, Serialize)] +pub struct IngestResult { + pub chats_upserted: usize, + pub messages_upserted: usize, + pub messages_pruned: u64, +} + +/// Request payload for `openhuman.whatsapp_data_list_chats`. +#[derive(Debug, Deserialize)] +pub struct ListChatsRequest { + /// Optional filter by account. When absent, all accounts are returned. + pub account_id: Option, + /// Maximum number of results (default: 50). + pub limit: Option, + /// Pagination offset (default: 0). + pub offset: Option, +} + +/// Request payload for `openhuman.whatsapp_data_list_messages`. +#[derive(Debug, Deserialize)] +pub struct ListMessagesRequest { + /// JID of the chat to retrieve messages for. + pub chat_id: String, + /// Optional filter by account. When absent, all accounts are searched. + pub account_id: Option, + /// Only return messages at or after this Unix timestamp (seconds). + pub since_ts: Option, + /// Only return messages at or before this Unix timestamp (seconds). + pub until_ts: Option, + /// Maximum number of results (default: 100). + pub limit: Option, + /// Pagination offset (default: 0). + pub offset: Option, +} + +/// Request payload for `openhuman.whatsapp_data_search_messages`. +#[derive(Debug, Deserialize)] +pub struct SearchMessagesRequest { + /// Full-text search query matched against message bodies (case-insensitive LIKE). + pub query: String, + /// Optional filter by chat JID. + pub chat_id: Option, + /// Optional filter by account. When absent, all accounts are searched. + pub account_id: Option, + /// Maximum number of results (default: 20). + pub limit: Option, +} diff --git a/tests/json_rpc_e2e.rs b/tests/json_rpc_e2e.rs index ca5cb6c8e..6b3fe246f 100644 --- a/tests/json_rpc_e2e.rs +++ b/tests/json_rpc_e2e.rs @@ -3381,3 +3381,380 @@ async fn channels_status_reflects_managed_dm_credential_e2e() { mock_join.abort(); rpc_join.abort(); } + +/// WhatsApp data: ingest → list_chats → list_messages → search_messages +/// +/// Validates the full structured data pipeline: +/// 1. Ingest two chats with five messages. +/// 2. list_chats returns both chats. +/// 3. list_messages for one chat returns the correct messages. +/// 4. search_messages finds the one matching message body. +#[tokio::test] +async fn whatsapp_data_ingest_and_query_e2e() { + let _env_lock = json_rpc_e2e_env_lock(); + let tmp = tempdir().expect("tempdir"); + let home = tmp.path(); + let openhuman_home = home.join(".openhuman"); + + let _home_guard = EnvVarGuard::set_to_path("HOME", home); + let _workspace_guard = EnvVarGuard::unset("OPENHUMAN_WORKSPACE"); + let _backend_url_guard = EnvVarGuard::unset("BACKEND_URL"); + let _vite_backend_guard = EnvVarGuard::unset("VITE_BACKEND_URL"); + + let (mock_addr, mock_join) = serve_on_ephemeral(mock_upstream_router()).await; + let mock_origin = format!("http://{}", mock_addr); + write_min_config(&openhuman_home, &mock_origin); + + // Init the whatsapp_data global before the router handles any requests. + openhuman_core::openhuman::whatsapp_data::global::init(openhuman_home.clone()) + .expect("whatsapp_data global init"); + + let (rpc_addr, rpc_join) = serve_on_ephemeral(build_core_http_router(false)).await; + let rpc_base = format!("http://{}", rpc_addr); + tokio::time::sleep(Duration::from_millis(100)).await; + + // ── 1. Ingest: 2 chats, 5 messages ────────────────────────────────────── + // Use timestamps relative to now so the 90-day auto-prune never removes them. + let now_ts = chrono::Utc::now().timestamp(); + let ingest = post_json_rpc( + &rpc_base, + 9001, + "openhuman.whatsapp_data_ingest", + json!({ + "account_id": "e2e-acct@c.us", + "chats": { + "alice@c.us": { "name": "Alice" }, + "group1@g.us": { "name": "Friends Group" } + }, + "messages": [ + { + "message_id": "msg-1", + "chat_id": "alice@c.us", + "sender": "Alice", + "sender_jid": "alice@c.us", + "from_me": false, + "body": "Hey, how are you?", + "timestamp": now_ts - 3600, + "message_type": "chat", + "source": "cdp-dom" + }, + { + "message_id": "msg-2", + "chat_id": "alice@c.us", + "sender": "me", + "sender_jid": null, + "from_me": true, + "body": "Doing great, thanks!", + "timestamp": now_ts - 3540, + "message_type": "chat", + "source": "cdp-dom" + }, + { + "message_id": "msg-3", + "chat_id": "alice@c.us", + "sender": "Alice", + "sender_jid": "alice@c.us", + "from_me": false, + "body": "Can you send me the umbrella report?", + "timestamp": now_ts - 3480, + "message_type": "chat", + "source": "cdp-dom" + }, + { + "message_id": "msg-4", + "chat_id": "group1@g.us", + "sender": "Bob", + "sender_jid": "bob@c.us", + "from_me": false, + "body": "Meeting rescheduled to 3pm", + "timestamp": now_ts - 2600, + "message_type": "chat", + "source": "cdp-indexeddb" + }, + { + "message_id": "msg-5", + "chat_id": "group1@g.us", + "sender": "me", + "sender_jid": null, + "from_me": true, + "body": "Got it, I'll be there", + "timestamp": now_ts - 2540, + "message_type": "chat", + "source": "cdp-indexeddb" + } + ] + }), + ) + .await; + let ingest_result = assert_no_jsonrpc_error(&ingest, "whatsapp_data_ingest"); + // The result may be wrapped in a logs envelope {result: ..., logs: [...]} + // or returned bare depending on whether logs are present. + let ingest_inner = ingest_result.get("result").unwrap_or(ingest_result); + let chats_upserted = ingest_inner + .get("chats_upserted") + .and_then(Value::as_u64) + .unwrap_or_else(|| panic!("missing chats_upserted in: {ingest_result}")); + assert_eq!( + chats_upserted, 2, + "expected 2 chats upserted: {ingest_result}" + ); + + // ── 2. list_chats — both chats should appear ───────────────────────────── + let list_chats = post_json_rpc( + &rpc_base, + 9002, + "openhuman.whatsapp_data_list_chats", + json!({ "account_id": "e2e-acct@c.us" }), + ) + .await; + let list_chats_result = assert_no_jsonrpc_error(&list_chats, "whatsapp_data_list_chats"); + // Unwrap the result/logs envelope if present, then find the chats array. + let list_chats_inner = list_chats_result.get("result").unwrap_or(list_chats_result); + let chats_arr = list_chats_inner + .as_array() + .or_else(|| list_chats_inner.get("chats").and_then(Value::as_array)) + .unwrap_or_else(|| panic!("expected chats array: {list_chats_result}")); + assert_eq!(chats_arr.len(), 2, "expected 2 chats: {list_chats_result}"); + + let chat_ids: Vec<&str> = chats_arr + .iter() + .filter_map(|c| c.get("chat_id").and_then(Value::as_str)) + .collect(); + assert!( + chat_ids.contains(&"alice@c.us"), + "alice chat missing: {chat_ids:?}" + ); + assert!( + chat_ids.contains(&"group1@g.us"), + "group chat missing: {chat_ids:?}" + ); + + // ── 3. list_messages — alice's chat should have 3 messages ─────────────── + let list_msgs = post_json_rpc( + &rpc_base, + 9003, + "openhuman.whatsapp_data_list_messages", + json!({ + "chat_id": "alice@c.us", + "account_id": "e2e-acct@c.us" + }), + ) + .await; + let list_msgs_result = assert_no_jsonrpc_error(&list_msgs, "whatsapp_data_list_messages"); + let list_msgs_inner = list_msgs_result.get("result").unwrap_or(list_msgs_result); + let msgs_arr = list_msgs_inner + .as_array() + .or_else(|| list_msgs_inner.get("messages").and_then(Value::as_array)) + .unwrap_or_else(|| panic!("expected messages array: {list_msgs_result}")); + assert_eq!( + msgs_arr.len(), + 3, + "expected 3 messages for alice: {list_msgs_result}" + ); + + // Messages should be ordered by timestamp ascending. + let bodies: Vec<&str> = msgs_arr + .iter() + .filter_map(|m| m.get("body").and_then(Value::as_str)) + .collect(); + assert_eq!(bodies[0], "Hey, how are you?"); + assert_eq!(bodies[1], "Doing great, thanks!"); + assert_eq!(bodies[2], "Can you send me the umbrella report?"); + + // ── 4. search_messages — "umbrella" should match exactly 1 message ─────── + let search = post_json_rpc( + &rpc_base, + 9004, + "openhuman.whatsapp_data_search_messages", + json!({ "query": "umbrella" }), + ) + .await; + let search_result = assert_no_jsonrpc_error(&search, "whatsapp_data_search_messages"); + let search_inner = search_result.get("result").unwrap_or(search_result); + let search_arr = search_inner + .as_array() + .or_else(|| search_inner.get("messages").and_then(Value::as_array)) + .unwrap_or_else(|| panic!("expected messages array from search: {search_result}")); + assert_eq!( + search_arr.len(), + 1, + "expected exactly 1 message matching 'umbrella': {search_result}" + ); + let found_body = search_arr[0] + .get("body") + .and_then(Value::as_str) + .unwrap_or(""); + assert!( + found_body.contains("umbrella"), + "search result body should contain 'umbrella': {found_body}" + ); + + // ── 5. account isolation — search scoped to first account only ──────────── + // Ingest a second account with a message that also contains "umbrella" to + // verify that account_id filtering prevents cross-account leakage. + let second_ingest = post_json_rpc( + &rpc_base, + 9005, + "openhuman.whatsapp_data_ingest", + json!({ + "account_id": "other-acct@c.us", + "chats": { + "contact@c.us": { "name": "Other Contact" } + }, + "messages": [ + { + "message_id": "other-msg-1", + "chat_id": "contact@c.us", + "sender": "Other Contact", + "sender_jid": "contact@c.us", + "from_me": false, + "body": "Can you bring the umbrella?", + "timestamp": now_ts - 1000, + "message_type": "chat", + "source": "cdp-dom" + } + ] + }), + ) + .await; + assert_no_jsonrpc_error(&second_ingest, "whatsapp_data_ingest (second account)"); + + // search scoped to first account should still return exactly 1 message and + // that message's account_id must be from the first account. + let scoped_search = post_json_rpc( + &rpc_base, + 9006, + "openhuman.whatsapp_data_search_messages", + json!({ + "query": "umbrella", + "account_id": "e2e-acct@c.us" + }), + ) + .await; + let scoped_result = + assert_no_jsonrpc_error(&scoped_search, "whatsapp_data_search_messages (scoped)"); + let scoped_inner = scoped_result.get("result").unwrap_or(scoped_result); + let scoped_arr = scoped_inner + .as_array() + .or_else(|| scoped_inner.get("messages").and_then(Value::as_array)) + .unwrap_or_else(|| panic!("expected messages array from scoped search: {scoped_result}")); + assert_eq!( + scoped_arr.len(), + 1, + "account-scoped search should return exactly 1 umbrella message: {scoped_result}" + ); + // Every result must belong to the queried account. + for msg in scoped_arr { + let msg_acct = msg.get("account_id").and_then(Value::as_str).unwrap_or(""); + assert_eq!( + msg_acct, "e2e-acct@c.us", + "scoped search returned message from wrong account: {msg}" + ); + } + + mock_join.abort(); + rpc_join.abort(); +} + +#[tokio::test] +async fn whatsapp_memory_doc_ingest_e2e() { + let _env_lock = json_rpc_e2e_env_lock(); + let tmp = tempdir().expect("tempdir"); + let home = tmp.path(); + let openhuman_home = home.join(".openhuman"); + + let _home_guard = EnvVarGuard::set_to_path("HOME", home); + let _workspace_guard = EnvVarGuard::unset("OPENHUMAN_WORKSPACE"); + let _backend_url_guard = EnvVarGuard::unset("BACKEND_URL"); + let _vite_backend_guard = EnvVarGuard::unset("VITE_BACKEND_URL"); + // Disable strict embedding so ingest falls back to the Inert + // (zero-vector) embedder when no Ollama endpoint is reachable. CI + // has no local Ollama; without this the memory_doc_ingest call + // would fail at the chunk-embedding step. + let _embed_strict_guard = EnvVarGuard::set("OPENHUMAN_MEMORY_EMBED_STRICT", "false"); + let _embed_endpoint_guard = EnvVarGuard::set("OPENHUMAN_MEMORY_EMBED_ENDPOINT", ""); + let _embed_model_guard = EnvVarGuard::set("OPENHUMAN_MEMORY_EMBED_MODEL", ""); + + let (mock_addr, mock_join) = serve_on_ephemeral(mock_upstream_router()).await; + let mock_origin = format!("http://{}", mock_addr); + write_min_config(&openhuman_home, &mock_origin); + + let (rpc_addr, rpc_join) = serve_on_ephemeral(build_core_http_router(false)).await; + let rpc_base = format!("http://{}", rpc_addr); + tokio::time::sleep(Duration::from_millis(100)).await; + + // ── 1. Ingest a WhatsApp-shaped memory document ─────────────────────────── + let ingest = post_json_rpc( + &rpc_base, + 9101, + "openhuman.memory_doc_ingest", + json!({ + "namespace": "whatsapp-web:test-acct@c.us", + "key": "alice@c.us:2026-05-07", + "title": "WhatsApp: Alice (2026-05-07)", + "content": "[10:00] Alice: Hey!\n[10:01] me: Hi there!\n[10:02] Alice: How are you?", + "source_type": "whatsapp-web", + "tags": ["whatsapp", "chat"], + "metadata": { + "chat_id": "alice@c.us", + "account_id": "test-acct@c.us" + } + }), + ) + .await; + assert_no_jsonrpc_error(&ingest, "memory_doc_ingest"); + + // ── 2. List documents scoped to the WhatsApp namespace ─────────────────── + let doc_list = post_json_rpc( + &rpc_base, + 9102, + "openhuman.memory_doc_list", + json!({ "namespace": "whatsapp-web:test-acct@c.us" }), + ) + .await; + let doc_list_result = assert_no_jsonrpc_error(&doc_list, "memory_doc_list"); + + // The result may be wrapped in a logs envelope {result: ..., logs: [...]} + // or returned bare depending on whether logs are present. + let doc_list_inner = doc_list_result.get("result").unwrap_or(doc_list_result); + + // The doc_list response can be: + // - an array directly + // - { documents: [...], count: N } + // - { result: [...] } + let docs_arr = doc_list_inner + .as_array() + .or_else(|| doc_list_inner.get("documents").and_then(Value::as_array)) + .or_else(|| doc_list_inner.get("items").and_then(Value::as_array)) + .unwrap_or_else(|| { + panic!("memory_doc_list: expected documents array in result: {doc_list_result}") + }); + + assert!( + !docs_arr.is_empty(), + "memory_doc_list should return at least 1 document after ingest: {doc_list_result}" + ); + + // ── 3. Verify the ingested document has the correct key and namespace ───── + let found = docs_arr.iter().find(|doc| { + let key_match = doc + .get("key") + .and_then(Value::as_str) + .map(|k| k == "alice@c.us:2026-05-07") + .unwrap_or(false); + let ns_match = doc + .get("namespace") + .and_then(Value::as_str) + .map(|n| n == "whatsapp-web:test-acct@c.us") + .unwrap_or(false); + key_match || ns_match + }); + assert!( + found.is_some(), + "ingested document with key 'alice@c.us:2026-05-07' not found in doc_list; \ + docs: {docs_arr:?}" + ); + + mock_join.abort(); + rpc_join.abort(); +}