From 37cd17889c4b28dd5676676e43ecda0a7ad52a31 Mon Sep 17 00:00:00 2001 From: Mega Mind <146339422+M3gA-Mind@users.noreply.github.com> Date: Fri, 8 May 2026 00:55:26 +0530 Subject: [PATCH] fix(channels): fix WhatsApp structured ingest pipeline + Memory page sync status (#1326) --- app/src-tauri/src/whatsapp_scanner/mod.rs | 138 ++++++++++------ app/src-tauri/vendor/tauri-cef | 2 +- .../intelligence/MemoryWorkspace.tsx | 2 + .../WhatsAppMemorySection.test.tsx | 150 ++++++++++++++++++ .../intelligence/WhatsAppMemorySection.tsx | 110 +++++++++++++ app/src/utils/tauriCommands/memory.test.ts | 92 ++++++++++- app/src/utils/tauriCommands/memory.ts | 59 +++++++ 7 files changed, 505 insertions(+), 48 deletions(-) create mode 100644 app/src/components/intelligence/WhatsAppMemorySection.test.tsx create mode 100644 app/src/components/intelligence/WhatsAppMemorySection.tsx diff --git a/app/src-tauri/src/whatsapp_scanner/mod.rs b/app/src-tauri/src/whatsapp_scanner/mod.rs index d55ac56c8..18417abb5 100644 --- a/app/src-tauri/src/whatsapp_scanner/mod.rs +++ b/app/src-tauri/src/whatsapp_scanner/mod.rs @@ -490,11 +490,11 @@ impl CdpConn { fn emit_snapshot(app: &AppHandle, account_id: &str, snap: &ScanSnapshot) { if !snap.ok { log::warn!( - "[wa][{}] snapshot not ok, error={:?}", + "[wa][{}] snapshot not ok (idb walk failed: {:?}) — falling through with dom-only data", account_id, snap.error ); - return; + // Fall through so DOM messages still reach the structured store. } // Resolve the active chat's JID from its display name (parsed from the // conversation header). Modern WhatsApp Web doesn't put the chat JID @@ -1055,15 +1055,8 @@ async fn post_whatsapp_data_ingest( 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}. + // Convert chats map values to {name: string|null} once, before batching. // The scanner passes chats as either: // - Value::String(display_name) — contact-cache format // - Value::Object({name: ..., ...}) — full IDB scan format @@ -1073,14 +1066,12 @@ async fn post_whatsapp_data_ingest( 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()) @@ -1093,44 +1084,99 @@ async fn post_whatsapp_data_ingest( }) .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, - }); - + // Split messages into chunks to stay well under the HTTP body size limit. + // Chats are sent only with the first batch (upserts are idempotent). + const BATCH_SIZE: usize = 500; + let empty_chats = Value::Object(serde_json::Map::new()); let url = crate::core_rpc::core_rpc_url_value(); - 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 resp = req - .json(&body) - .send() - .await - .map_err(|e| format!("POST {url}: {e}"))?; - 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}")); - } + + // Build at least one batch even when messages is empty (chats-only upsert). + let chunks: Vec<&[Value]> = if messages.is_empty() { + vec![&[]] + } else { + messages.chunks(BATCH_SIZE).collect() + }; + + let total_batches = chunks.len(); log::debug!( - "[wa][{}] whatsapp_data_ingest ok account={} messages={}", + "[wa][{}] whatsapp_data_ingest chats={} messages={} batches={} source={}", account_id, + chats_param.len(), + messages.len(), + total_batches, + source + ); + + for (batch_idx, chunk) in chunks.iter().enumerate() { + let batch_chats = if batch_idx == 0 { + Value::Object(chats_param.clone()) + } else { + empty_chats.clone() + }; + let params = json!({ + "account_id": account_id, + "chats": batch_chats, + "messages": chunk, + }); + let body = json!({ + "jsonrpc": "2.0", + "id": 1, + "method": "openhuman.whatsapp_data_ingest", + "params": params, + }); + + let mut last_err = String::new(); + let mut succeeded = false; + 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][{}] whatsapp_data_ingest connect error batch={}/{} attempt {}: {}", + account_id, + batch_idx + 1, + total_batches, + attempt, + last_err + ); + tokio::time::sleep(Duration::from_millis(500)).await; + } + continue; + } + 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}")); + } + succeeded = true; + break; + } + } + } + if !succeeded { + return Err(last_err); + } + } + + log::debug!( + "[wa][{}] whatsapp_data_ingest ok messages={} batches={}", account_id, - messages.len() + messages.len(), + total_batches, ); Ok(()) } diff --git a/app/src-tauri/vendor/tauri-cef b/app/src-tauri/vendor/tauri-cef index f1ee9554f..c10bc0f72 160000 --- a/app/src-tauri/vendor/tauri-cef +++ b/app/src-tauri/vendor/tauri-cef @@ -1 +1 @@ -Subproject commit f1ee9554ffc414524ed6c2013dd286f0fe38907f +Subproject commit c10bc0f7293f6070478fc8af1a157004f5916f10 diff --git a/app/src/components/intelligence/MemoryWorkspace.tsx b/app/src/components/intelligence/MemoryWorkspace.tsx index 5880047b6..cefb674f6 100644 --- a/app/src/components/intelligence/MemoryWorkspace.tsx +++ b/app/src/components/intelligence/MemoryWorkspace.tsx @@ -39,6 +39,7 @@ import { } from '../../utils/tauriCommands'; import { MemoryGraph } from './MemoryGraph'; import { MemorySources } from './MemorySources'; +import { WhatsAppMemorySection } from './WhatsAppMemorySection'; interface MemoryWorkspaceProps { onToast?: (toast: Omit) => void; @@ -241,6 +242,7 @@ export function MemoryWorkspace({ onToast }: MemoryWorkspaceProps) { return (
+
({ + whatsappListChats: (...args: unknown[]) => mockWhatsappListChats(...args), +})); + +function makeChat(overrides: Record = {}) { + return { + chat_id: 'chat-1', + display_name: 'Test Chat', + is_group: false, + account_id: 'acc-1', + last_message_ts: 1_700_000_000, + message_count: 5, + updated_at: 1_700_000_000, + ...overrides, + }; +} + +describe('', () => { + beforeEach(() => { + mockWhatsappListChats.mockReset(); + }); + + afterEach(() => { + vi.restoreAllMocks(); + }); + + it('renders nothing when load returns an empty array', async () => { + mockWhatsappListChats.mockResolvedValueOnce([]); + const { container } = render(); + await waitFor(() => expect(mockWhatsappListChats).toHaveBeenCalled()); + expect(container.firstChild).toBeNull(); + }); + + it('stays hidden when load throws (error is swallowed silently)', async () => { + mockWhatsappListChats.mockRejectedValueOnce(new Error('scanner not ready')); + const { container } = render(); + await waitFor(() => expect(mockWhatsappListChats).toHaveBeenCalled()); + expect(container.firstChild).toBeNull(); + }); + + it('calls whatsappListChats with limit:200', async () => { + mockWhatsappListChats.mockResolvedValueOnce([]); + render(); + await waitFor(() => expect(mockWhatsappListChats).toHaveBeenCalledWith({ limit: 200 })); + }); + + it('renders section and plural "chats" when multiple chats load', async () => { + mockWhatsappListChats.mockResolvedValueOnce([ + makeChat({ chat_id: 'c1' }), + makeChat({ chat_id: 'c2' }), + ]); + render(); + await waitFor(() => screen.getByTestId('whatsapp-memory-section')); + expect(screen.getByText(/2 chats synced/)).toBeTruthy(); + }); + + it('renders singular "chat" for exactly 1 chat', async () => { + mockWhatsappListChats.mockResolvedValueOnce([makeChat()]); + render(); + await waitFor(() => screen.getByTestId('whatsapp-memory-section')); + const el = screen.getByText(/chat synced/); + expect(el.textContent).toMatch(/^1 chat synced/); + }); + + it('shows "just now" when delta < 60s', async () => { + const updatedAt = 1_700_000_000; + vi.spyOn(Date, 'now').mockReturnValue((updatedAt + 30) * 1000); + mockWhatsappListChats.mockResolvedValueOnce([makeChat({ updated_at: updatedAt })]); + render(); + await waitFor(() => screen.getByTestId('whatsapp-memory-section')); + expect(screen.getByText(/just now/)).toBeTruthy(); + }); + + it('shows "Xm ago" for 60-3599s delta', async () => { + const updatedAt = 1_700_000_000; + vi.spyOn(Date, 'now').mockReturnValue((updatedAt + 120) * 1000); + mockWhatsappListChats.mockResolvedValueOnce([makeChat({ updated_at: updatedAt })]); + render(); + await waitFor(() => screen.getByTestId('whatsapp-memory-section')); + expect(screen.getByText(/2m ago/)).toBeTruthy(); + }); + + it('shows "Xh ago" for 3600-86399s delta', async () => { + const updatedAt = 1_700_000_000; + vi.spyOn(Date, 'now').mockReturnValue((updatedAt + 7200) * 1000); + mockWhatsappListChats.mockResolvedValueOnce([makeChat({ updated_at: updatedAt })]); + render(); + await waitFor(() => screen.getByTestId('whatsapp-memory-section')); + expect(screen.getByText(/2h ago/)).toBeTruthy(); + }); + + it('shows "Xd ago" for delta >= 86400s', async () => { + const updatedAt = 1_700_000_000; + vi.spyOn(Date, 'now').mockReturnValue((updatedAt + 86400 * 3) * 1000); + mockWhatsappListChats.mockResolvedValueOnce([makeChat({ updated_at: updatedAt })]); + render(); + await waitFor(() => screen.getByTestId('whatsapp-memory-section')); + expect(screen.getByText(/3d ago/)).toBeTruthy(); + }); + + it('omits timestamp when all chats have updated_at = 0', async () => { + mockWhatsappListChats.mockResolvedValueOnce([makeChat({ updated_at: 0 })]); + render(); + await waitFor(() => screen.getByTestId('whatsapp-memory-section')); + const el = screen.getByText(/chat synced/); + expect(el.textContent).not.toMatch(/ago|just now/); + }); + + it('handleSync: clicking Sync reloads data and updates count', async () => { + mockWhatsappListChats + .mockResolvedValueOnce([makeChat()]) + .mockResolvedValueOnce([makeChat({ chat_id: 'c1' }), makeChat({ chat_id: 'c2' })]); + render(); + await waitFor(() => screen.getByTestId('whatsapp-memory-section')); + + fireEvent.click(screen.getByRole('button')); + await waitFor(() => expect(mockWhatsappListChats).toHaveBeenCalledTimes(2)); + await waitFor(() => screen.getByText(/2 chats synced/)); + }); + + it('handleSync: button shows "Syncing…" and is disabled while load is in flight', async () => { + mockWhatsappListChats.mockResolvedValueOnce([makeChat()]); + render(); + await waitFor(() => screen.getByTestId('whatsapp-memory-section')); + + // eslint-disable-next-line @typescript-eslint/no-explicit-any + let resolveSync!: (v: any) => void; + mockWhatsappListChats.mockReturnValueOnce( + new Promise(r => { + resolveSync = r; + }) + ); + fireEvent.click(screen.getByRole('button')); + + await waitFor(() => screen.getByText('Syncing…')); + expect(screen.getByRole('button')).toBeDisabled(); + + await act(async () => { + resolveSync([makeChat()]); + }); + await waitFor(() => screen.getByText('Sync')); + }); +}); diff --git a/app/src/components/intelligence/WhatsAppMemorySection.tsx b/app/src/components/intelligence/WhatsAppMemorySection.tsx new file mode 100644 index 000000000..874c258d1 --- /dev/null +++ b/app/src/components/intelligence/WhatsAppMemorySection.tsx @@ -0,0 +1,110 @@ +import { useCallback, useEffect, useState } from 'react'; + +import { whatsappListChats } from '../../utils/tauriCommands/memory'; + +interface WhatsAppMemorySectionProps { + pollIntervalMs?: number; +} + +export function WhatsAppMemorySection({ pollIntervalMs = 30000 }: WhatsAppMemorySectionProps) { + const [chatCount, setChatCount] = useState(null); + const [lastSyncTs, setLastSyncTs] = useState(null); + const [syncing, setSyncing] = useState(false); + + const load = useCallback(async () => { + try { + const chats = await whatsappListChats({ limit: 200 }); + setChatCount(chats.length); + const latest = chats.reduce((max, c) => Math.max(max, c.updated_at), 0); + setLastSyncTs(latest > 0 ? latest : null); + } catch { + // Scanner may not have data yet — stay hidden. + } + }, []); + + useEffect(() => { + void load(); + }, [load]); + + useEffect(() => { + if (!pollIntervalMs) return undefined; + const id = setInterval(() => void load(), pollIntervalMs); + return () => clearInterval(id); + }, [pollIntervalMs, load]); + + const handleSync = useCallback(async () => { + setSyncing(true); + try { + await load(); + } finally { + setSyncing(false); + } + }, [load]); + + if (chatCount === null || chatCount === 0) return null; + + return ( +
+
+
+ + WhatsApp + + {chatCount.toLocaleString()} chat{chatCount !== 1 ? 's' : ''} synced + {lastSyncTs !== null && <> · {relativeTime(lastSyncTs)}} + +
+ +
+
+ ); +} + +function relativeTime(secs: number): string { + const delta = Date.now() / 1000 - secs; + if (delta < 60) return 'just now'; + if (delta < 3600) return `${Math.floor(delta / 60)}m ago`; + if (delta < 86400) return `${Math.floor(delta / 3600)}h ago`; + return `${Math.floor(delta / 86400)}d ago`; +} + +function WhatsAppIcon() { + return ( + + ); +} + +function RefreshIcon({ spinning }: { spinning: boolean }) { + return ( + + ); +} diff --git a/app/src/utils/tauriCommands/memory.test.ts b/app/src/utils/tauriCommands/memory.test.ts index 84d564737..acd6afe81 100644 --- a/app/src/utils/tauriCommands/memory.test.ts +++ b/app/src/utils/tauriCommands/memory.test.ts @@ -5,7 +5,14 @@ import { beforeEach, describe, expect, type Mock, test, vi } from 'vitest'; import { callCoreRpc } from '../../services/coreRpcClient'; import { isTauri } from './common'; -import { aiListMemoryFiles, memoryLearnAll, memorySyncAll, memorySyncChannel } from './memory'; +import { + aiListMemoryFiles, + memoryLearnAll, + memorySyncAll, + memorySyncChannel, + whatsappListChats, + whatsappListMessages, +} from './memory'; vi.mock('../../services/coreRpcClient', () => ({ callCoreRpc: vi.fn() })); vi.mock('./common', () => ({ isTauri: vi.fn(() => true) })); @@ -150,3 +157,86 @@ describe('aiListMemoryFiles', () => { await expect(aiListMemoryFiles()).rejects.toThrow(/Not running in Tauri/); }); }); + +describe('whatsappListChats', () => { + test('throws when not in Tauri', async () => { + mockIsTauri.mockReturnValue(false); + await expect(whatsappListChats()).rejects.toThrow('Not running in Tauri'); + expect(mockCallCoreRpc).not.toHaveBeenCalled(); + }); + + test('calls correct RPC method with provided params', async () => { + mockCallCoreRpc.mockResolvedValueOnce([]); + await whatsappListChats({ limit: 10 }); + expect(mockCallCoreRpc).toHaveBeenCalledWith({ + method: 'openhuman.whatsapp_data_list_chats', + params: { limit: 10 }, + }); + }); + + test('uses empty params object when none provided', async () => { + mockCallCoreRpc.mockResolvedValueOnce([]); + await whatsappListChats(); + expect(mockCallCoreRpc).toHaveBeenCalledWith({ + method: 'openhuman.whatsapp_data_list_chats', + params: {}, + }); + }); + + test('returns array directly when response is already an array', async () => { + const chats = [{ chat_id: 'c1', display_name: 'Direct' }]; + mockCallCoreRpc.mockResolvedValueOnce(chats); + const result = await whatsappListChats(); + expect(result).toBe(chats); + }); + + test('extracts result field from wrapped response envelope', async () => { + const chats = [{ chat_id: 'c2', display_name: 'Wrapped' }]; + mockCallCoreRpc.mockResolvedValueOnce({ result: chats, logs: [] }); + const result = await whatsappListChats(); + expect(result).toEqual(chats); + }); + + test('returns empty array when wrapped response has no result field', async () => { + mockCallCoreRpc.mockResolvedValueOnce({ logs: [] }); + const result = await whatsappListChats(); + expect(result).toEqual([]); + }); +}); + +describe('whatsappListMessages', () => { + test('throws when not in Tauri', async () => { + mockIsTauri.mockReturnValue(false); + await expect(whatsappListMessages({ chat_id: 'c1' })).rejects.toThrow('Not running in Tauri'); + expect(mockCallCoreRpc).not.toHaveBeenCalled(); + }); + + test('calls correct RPC method with required chat_id param', async () => { + mockCallCoreRpc.mockResolvedValueOnce([]); + await whatsappListMessages({ chat_id: 'c1', limit: 50 }); + expect(mockCallCoreRpc).toHaveBeenCalledWith({ + method: 'openhuman.whatsapp_data_list_messages', + params: { chat_id: 'c1', limit: 50 }, + }); + }); + + test('returns array directly when response is already an array', async () => { + const msgs = [{ message_id: 'm1', body: 'hello' }]; + mockCallCoreRpc.mockResolvedValueOnce(msgs); + const result = await whatsappListMessages({ chat_id: 'c1' }); + expect(result).toBe(msgs); + }); + + test('extracts result field from wrapped response envelope', async () => { + const msgs = [{ message_id: 'm2', body: 'world' }]; + mockCallCoreRpc.mockResolvedValueOnce({ result: msgs, logs: [] }); + const result = await whatsappListMessages({ chat_id: 'c1' }); + expect(result).toEqual(msgs); + }); + + test('returns empty array when wrapped response has no result field', async () => { + mockCallCoreRpc.mockResolvedValueOnce({ logs: [] }); + const result = await whatsappListMessages({ chat_id: 'c1' }); + expect(result).toEqual([]); + }); +}); diff --git a/app/src/utils/tauriCommands/memory.ts b/app/src/utils/tauriCommands/memory.ts index 2ce269d3d..5a4f9e8bd 100644 --- a/app/src/utils/tauriCommands/memory.ts +++ b/app/src/utils/tauriCommands/memory.ts @@ -351,3 +351,62 @@ export async function memoryLearnAll(namespaces?: string[]): Promise { + if (!isTauri()) { + throw new Error('Not running in Tauri'); + } + const resp = await callCoreRpc<{ result?: WhatsAppChat[]; logs?: string[] } | WhatsAppChat[]>({ + method: 'openhuman.whatsapp_data_list_chats', + params: params ?? {}, + }); + if (Array.isArray(resp)) return resp; + return (resp as { result?: WhatsAppChat[] }).result ?? []; +} + +/** List messages for a chat from the local store. */ +export async function whatsappListMessages(params: { + chat_id: string; + account_id?: string; + limit?: number; + offset?: number; +}): Promise { + if (!isTauri()) { + throw new Error('Not running in Tauri'); + } + const resp = await callCoreRpc< + { result?: WhatsAppMessage[]; logs?: string[] } | WhatsAppMessage[] + >({ method: 'openhuman.whatsapp_data_list_messages', params }); + if (Array.isArray(resp)) return resp; + return (resp as { result?: WhatsAppMessage[] }).result ?? []; +}