mirror of
https://github.com/tinyhumansai/openhuman.git
synced 2026-07-30 23:14:37 +00:00
feat(gmessages): scaffold Google Messages Web scanner (Windows, read-only IDB) (#735)
Co-authored-by: Jwalin Shah <jshah1331@gmail.com>
This commit is contained in:
Jwalin Shah
co-authored by
Jwalin Shah
Jwalin Shah
parent
5542acd1cd
commit
a3b0a78f2b
@@ -0,0 +1,205 @@
|
||||
//! CDP-driven walk of the Google Messages Web `bugle_db` IndexedDB.
|
||||
//!
|
||||
//! Pairs with `idb.rs` (schema + normalization). This module does the
|
||||
//! `IndexedDB.requestData` paging + `Runtime.callFunctionOn` serialisation
|
||||
//! dance, then hands the raw JSON rows to `idb::normalize_*` for shape
|
||||
//! checking.
|
||||
|
||||
use serde_json::{json, Value};
|
||||
|
||||
use super::idb::{
|
||||
self, Conversation, Message, ParticipantMap, DATABASE_NAME, STORE_CONVERSATIONS,
|
||||
STORE_MESSAGES, STORE_PARTICIPANTS,
|
||||
};
|
||||
use crate::cdp::CdpConn;
|
||||
|
||||
/// IndexedDB security origin for the Google Messages Web app.
|
||||
const ORIGIN: &str = "https://messages.google.com";
|
||||
/// Rows per `IndexedDB.requestData` call — matches the WhatsApp scanner.
|
||||
const PAGE_SIZE: i64 = 500;
|
||||
/// Hard cap per store to bound full-scan cost on huge histories.
|
||||
const MAX_RECORDS_PER_STORE: usize = 20_000;
|
||||
/// `Runtime.callFunctionOn` batch size for RemoteObject serialisation.
|
||||
const SERIALIZE_BATCH: usize = 100;
|
||||
|
||||
pub struct WalkResult {
|
||||
pub messages: Vec<Message>,
|
||||
pub conversations: Vec<Conversation>,
|
||||
pub participants: ParticipantMap,
|
||||
}
|
||||
|
||||
/// Walk `bugle_db`: messages, conversations, participants. Per-store
|
||||
/// failures are logged and swallowed so one bad store doesn't nuke the
|
||||
/// cycle — the caller still gets whatever did come back.
|
||||
pub async fn walk(cdp: &mut CdpConn, session: &str) -> Result<WalkResult, String> {
|
||||
// `IndexedDB.enable` is a no-op on modern Chromium but older CEF
|
||||
// builds refuse `requestData` without it. Cost is trivial.
|
||||
if let Err(e) = cdp.call("IndexedDB.enable", json!({}), Some(session)).await {
|
||||
log::debug!("[gmessages][idb] enable: {}", e);
|
||||
}
|
||||
|
||||
let messages_raw = match read_store(cdp, session, STORE_MESSAGES).await {
|
||||
Ok(v) => v,
|
||||
Err(e) => {
|
||||
log::warn!("[gmessages][idb] read {} failed: {}", STORE_MESSAGES, e);
|
||||
Vec::new()
|
||||
}
|
||||
};
|
||||
let convos_raw = match read_store(cdp, session, STORE_CONVERSATIONS).await {
|
||||
Ok(v) => v,
|
||||
Err(e) => {
|
||||
log::warn!("[gmessages][idb] read {} failed: {}", STORE_CONVERSATIONS, e);
|
||||
Vec::new()
|
||||
}
|
||||
};
|
||||
let parts_raw = match read_store(cdp, session, STORE_PARTICIPANTS).await {
|
||||
Ok(v) => v,
|
||||
Err(e) => {
|
||||
log::warn!("[gmessages][idb] read {} failed: {}", STORE_PARTICIPANTS, e);
|
||||
Vec::new()
|
||||
}
|
||||
};
|
||||
|
||||
let messages: Vec<Message> = messages_raw
|
||||
.iter()
|
||||
.filter_map(idb::normalize_message)
|
||||
.collect();
|
||||
let conversations: Vec<Conversation> = convos_raw
|
||||
.iter()
|
||||
.filter_map(idb::normalize_conversation)
|
||||
.collect();
|
||||
let mut participants = ParticipantMap::default();
|
||||
for raw in &parts_raw {
|
||||
if let Some((id, name)) = idb::normalize_participant(raw) {
|
||||
participants.insert(id, name);
|
||||
}
|
||||
}
|
||||
|
||||
log::info!(
|
||||
"[gmessages][idb] walk messages={} conversations={} participants={}",
|
||||
messages.len(),
|
||||
conversations.len(),
|
||||
participants.len()
|
||||
);
|
||||
Ok(WalkResult {
|
||||
messages,
|
||||
conversations,
|
||||
participants,
|
||||
})
|
||||
}
|
||||
|
||||
async fn read_store(cdp: &mut CdpConn, session: &str, store: &str) -> Result<Vec<Value>, String> {
|
||||
let mut out: Vec<Value> = Vec::new();
|
||||
let mut skip: i64 = 0;
|
||||
loop {
|
||||
let remaining = MAX_RECORDS_PER_STORE.saturating_sub(out.len());
|
||||
if remaining == 0 {
|
||||
break;
|
||||
}
|
||||
let page = (remaining as i64).min(PAGE_SIZE);
|
||||
let resp = cdp
|
||||
.call(
|
||||
"IndexedDB.requestData",
|
||||
json!({
|
||||
"securityOrigin": ORIGIN,
|
||||
"databaseName": DATABASE_NAME,
|
||||
"objectStoreName": store,
|
||||
"indexName": "",
|
||||
"skipCount": skip,
|
||||
"pageSize": page,
|
||||
}),
|
||||
Some(session),
|
||||
)
|
||||
.await?;
|
||||
let entries = resp
|
||||
.get("objectStoreDataEntries")
|
||||
.and_then(|x| x.as_array())
|
||||
.cloned()
|
||||
.unwrap_or_default();
|
||||
if entries.is_empty() {
|
||||
break;
|
||||
}
|
||||
let value_refs: Vec<&Value> = entries
|
||||
.iter()
|
||||
.map(|e| e.get("value").unwrap_or(&Value::Null))
|
||||
.collect();
|
||||
let materialised = serialize_values(cdp, session, &value_refs).await?;
|
||||
out.extend(materialised);
|
||||
|
||||
let has_more = resp
|
||||
.get("hasMore")
|
||||
.and_then(|x| x.as_bool())
|
||||
.unwrap_or(false);
|
||||
skip += entries.len() as i64;
|
||||
if !has_more {
|
||||
break;
|
||||
}
|
||||
}
|
||||
log::debug!("[gmessages][idb] store={} records={}", store, out.len());
|
||||
Ok(out)
|
||||
}
|
||||
|
||||
async fn serialize_values(
|
||||
cdp: &mut CdpConn,
|
||||
session: &str,
|
||||
values: &[&Value],
|
||||
) -> Result<Vec<Value>, String> {
|
||||
let mut result: Vec<Value> = vec![Value::Null; values.len()];
|
||||
let mut pending: Vec<(usize, String)> = Vec::new();
|
||||
for (i, v) in values.iter().enumerate() {
|
||||
if let Some(inline) = v.get("value") {
|
||||
result[i] = inline.clone();
|
||||
continue;
|
||||
}
|
||||
if let Some(oid) = v.get("objectId").and_then(|x| x.as_str()) {
|
||||
pending.push((i, oid.to_string()));
|
||||
}
|
||||
}
|
||||
for chunk in pending.chunks(SERIALIZE_BATCH) {
|
||||
let oids: Vec<&str> = chunk.iter().map(|(_, oid)| oid.as_str()).collect();
|
||||
let serialised = call_function_batch(cdp, session, &oids).await?;
|
||||
if serialised.len() != chunk.len() {
|
||||
return Err(format!(
|
||||
"serialise batch length mismatch: got {}, expected {}",
|
||||
serialised.len(),
|
||||
chunk.len()
|
||||
));
|
||||
}
|
||||
for ((idx, _), val) in chunk.iter().zip(serialised.into_iter()) {
|
||||
result[*idx] = val;
|
||||
}
|
||||
}
|
||||
Ok(result)
|
||||
}
|
||||
|
||||
async fn call_function_batch(
|
||||
cdp: &mut CdpConn,
|
||||
session: &str,
|
||||
object_ids: &[&str],
|
||||
) -> Result<Vec<Value>, String> {
|
||||
if object_ids.is_empty() {
|
||||
return Ok(Vec::new());
|
||||
}
|
||||
let (first, rest) = object_ids.split_first().unwrap();
|
||||
let args: Vec<Value> = rest.iter().map(|oid| json!({ "objectId": oid })).collect();
|
||||
let resp = cdp
|
||||
.call(
|
||||
"Runtime.callFunctionOn",
|
||||
json!({
|
||||
"objectId": first,
|
||||
"functionDeclaration": "function(){return [this].concat(Array.prototype.slice.call(arguments));}",
|
||||
"arguments": args,
|
||||
"returnByValue": true,
|
||||
"silent": true,
|
||||
}),
|
||||
Some(session),
|
||||
)
|
||||
.await?;
|
||||
if let Some(exc) = resp.get("exceptionDetails") {
|
||||
return Err(format!("callFunctionOn threw: {exc}"));
|
||||
}
|
||||
resp.pointer("/result/value")
|
||||
.and_then(|v| v.as_array())
|
||||
.cloned()
|
||||
.ok_or_else(|| format!("callFunctionOn result not array: {resp}"))
|
||||
}
|
||||
@@ -0,0 +1,313 @@
|
||||
//! Google Messages Web `bugle_db` IndexedDB schema + normalization.
|
||||
//!
|
||||
//! Schema knowledge is taken from publicly documented reverse-engineering
|
||||
//! of the Google Messages Web client (the `mautrix-gmessages` project and
|
||||
//! the Google Messages Web source itself). No code is copied —
|
||||
//! only the factual store / key shape, which is not copyrightable.
|
||||
//!
|
||||
//! Stores we care about:
|
||||
//! * `conversations` — thread metadata (id, participant ids, name)
|
||||
//! * `messages` — individual SMS/RCS rows
|
||||
//! * `participants` — participant id → contact name resolution
|
||||
//!
|
||||
//! Stores we deliberately skip:
|
||||
//! * `settings`, `drafts`, `attachments-cache` — not needed for recall.
|
||||
//!
|
||||
//! This module only holds schema + normalization. The CDP walk that
|
||||
//! actually calls `IndexedDB.requestData` will live alongside the WhatsApp
|
||||
//! scanner's CDP plumbing once we lift a shared `cdp` module — see the
|
||||
//! TODO in `mod.rs`.
|
||||
|
||||
use std::collections::HashMap;
|
||||
|
||||
use serde_json::Value;
|
||||
|
||||
/// `bugle_db` database name. Stable since ~2022 per mautrix-gmessages
|
||||
/// history; Google has not shipped a schema rename in the tracked window.
|
||||
pub const DATABASE_NAME: &str = "bugle_db";
|
||||
pub const STORE_CONVERSATIONS: &str = "conversations";
|
||||
pub const STORE_MESSAGES: &str = "messages";
|
||||
pub const STORE_PARTICIPANTS: &str = "participants";
|
||||
|
||||
/// Normalized message row emitted to the memory-doc pipeline.
|
||||
#[derive(Debug, Clone, Default, PartialEq, Eq)]
|
||||
pub struct Message {
|
||||
pub id: String,
|
||||
pub thread_id: Option<String>,
|
||||
/// `None` when the message is outbound (sent by the user).
|
||||
pub sender_id: Option<String>,
|
||||
pub from_me: bool,
|
||||
/// Plain UTF-8 body. Attachments / reactions collapse to empty string
|
||||
/// at normalization — callers render them as `[non-text]`.
|
||||
pub text: String,
|
||||
pub timestamp_unix: i64,
|
||||
/// "sms", "rcs", "mms", etc. Preserved for downstream filters.
|
||||
pub message_type: Option<String>,
|
||||
}
|
||||
|
||||
/// Normalized conversation (thread) metadata.
|
||||
#[derive(Debug, Clone, Default, PartialEq, Eq)]
|
||||
pub struct Conversation {
|
||||
pub thread_id: String,
|
||||
pub display_name: Option<String>,
|
||||
pub participant_ids: Vec<String>,
|
||||
}
|
||||
|
||||
/// Participant-id → display-name map. Populated from the `participants`
|
||||
/// store; used by `format_transcript` to render human-readable senders.
|
||||
#[derive(Debug, Default, Clone)]
|
||||
pub struct ParticipantMap {
|
||||
inner: HashMap<String, String>,
|
||||
}
|
||||
|
||||
impl ParticipantMap {
|
||||
pub fn insert(&mut self, id: String, name: String) {
|
||||
self.inner.insert(id, name);
|
||||
}
|
||||
|
||||
pub fn display_name(&self, id: &str) -> Option<String> {
|
||||
self.inner.get(id).cloned()
|
||||
}
|
||||
|
||||
pub fn len(&self) -> usize {
|
||||
self.inner.len()
|
||||
}
|
||||
|
||||
pub fn is_empty(&self) -> bool {
|
||||
self.inner.is_empty()
|
||||
}
|
||||
}
|
||||
|
||||
/// Convert a raw JSON row from the `messages` object store into our
|
||||
/// normalized shape. Returns `None` if required fields are missing or
|
||||
/// malformed — we log + skip rather than failing the entire walk.
|
||||
///
|
||||
/// Expected bugle_db fields (observed, documented in mautrix-gmessages):
|
||||
/// * `messageId` (string) — primary key
|
||||
/// * `conversationId` (string)
|
||||
/// * `senderId` (string, absent for outgoing)
|
||||
/// * `messageStatus` (object with `status` int; outgoing statuses 2/4/6)
|
||||
/// * `text` (string; may be absent for attachment-only)
|
||||
/// * `timestamp` (int; microseconds since unix epoch)
|
||||
/// * `messageType` (string: "SMS", "RCS", etc.)
|
||||
pub fn normalize_message(raw: &Value) -> Option<Message> {
|
||||
let id = raw.get("messageId")?.as_str()?.to_string();
|
||||
let thread_id = raw
|
||||
.get("conversationId")
|
||||
.and_then(|v| v.as_str())
|
||||
.map(str::to_string);
|
||||
let sender_id = raw
|
||||
.get("senderId")
|
||||
.and_then(|v| v.as_str())
|
||||
.map(str::to_string);
|
||||
let from_me = is_outgoing(raw);
|
||||
let text = raw
|
||||
.get("text")
|
||||
.and_then(|v| v.as_str())
|
||||
.unwrap_or("")
|
||||
.to_string();
|
||||
// bugle_db timestamps are microseconds since unix epoch. Guard against
|
||||
// the legacy-seconds form (< 10^12 = before year 33700 in micros,
|
||||
// practically any real timestamp is in the 10^15 range).
|
||||
let timestamp_unix = raw.get("timestamp").and_then(|v| v.as_i64()).map(|t| {
|
||||
if t > 1_000_000_000_000 {
|
||||
t / 1_000_000
|
||||
} else {
|
||||
t
|
||||
}
|
||||
})?;
|
||||
let message_type = raw
|
||||
.get("messageType")
|
||||
.and_then(|v| v.as_str())
|
||||
.map(|s| s.to_ascii_lowercase());
|
||||
|
||||
Some(Message {
|
||||
id,
|
||||
thread_id,
|
||||
sender_id: if from_me { None } else { sender_id },
|
||||
from_me,
|
||||
text,
|
||||
timestamp_unix,
|
||||
message_type,
|
||||
})
|
||||
}
|
||||
|
||||
/// Heuristic: bugle_db marks outgoing messages with a `messageStatus`
|
||||
/// object whose `status` is in {2 (OUTGOING_DELIVERED), 4 (OUTGOING_READ),
|
||||
/// 6 (OUTGOING_FAILED)} or an explicit boolean `isOutgoing` on newer
|
||||
/// schemas. Fall back to `senderId == null` which is also a reliable
|
||||
/// signal on older writes.
|
||||
fn is_outgoing(raw: &Value) -> bool {
|
||||
if let Some(b) = raw.get("isOutgoing").and_then(|v| v.as_bool()) {
|
||||
return b;
|
||||
}
|
||||
if let Some(status) = raw
|
||||
.get("messageStatus")
|
||||
.and_then(|s| s.get("status"))
|
||||
.and_then(|v| v.as_i64())
|
||||
{
|
||||
// Status codes 1-9 are outgoing; 10+ are incoming (OUTGOING_* vs
|
||||
// INCOMING_* in the bugle_db protobuf enum). Exact values per
|
||||
// mautrix-gmessages' `libgm/events/types.go`.
|
||||
return (1..=9).contains(&status);
|
||||
}
|
||||
raw.get("senderId")
|
||||
.map(|v| v.is_null() || v.as_str().is_some_and(str::is_empty))
|
||||
.unwrap_or(false)
|
||||
}
|
||||
|
||||
/// Normalize a `conversations` store row.
|
||||
pub fn normalize_conversation(raw: &Value) -> Option<Conversation> {
|
||||
let thread_id = raw.get("conversationId")?.as_str()?.to_string();
|
||||
let display_name = raw.get("name").and_then(|v| v.as_str()).map(str::to_string);
|
||||
let participant_ids = raw
|
||||
.get("participantIds")
|
||||
.and_then(|v| v.as_array())
|
||||
.map(|arr| {
|
||||
arr.iter()
|
||||
.filter_map(|v| v.as_str().map(str::to_string))
|
||||
.collect()
|
||||
})
|
||||
.unwrap_or_default();
|
||||
Some(Conversation {
|
||||
thread_id,
|
||||
display_name,
|
||||
participant_ids,
|
||||
})
|
||||
}
|
||||
|
||||
/// Normalize a `participants` store row into `(id, name)`.
|
||||
pub fn normalize_participant(raw: &Value) -> Option<(String, String)> {
|
||||
let id = raw.get("participantId")?.as_str()?.to_string();
|
||||
let name = raw
|
||||
.get("fullName")
|
||||
.or_else(|| raw.get("firstName"))
|
||||
.or_else(|| raw.get("displayName"))
|
||||
.and_then(|v| v.as_str())
|
||||
.unwrap_or("")
|
||||
.to_string();
|
||||
if name.is_empty() {
|
||||
None
|
||||
} else {
|
||||
Some((id, name))
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
use serde_json::json;
|
||||
|
||||
#[test]
|
||||
fn normalize_incoming_sms_row() {
|
||||
let raw = json!({
|
||||
"messageId": "msg-1",
|
||||
"conversationId": "thread-1",
|
||||
"senderId": "+15551234567",
|
||||
"text": "hello",
|
||||
"timestamp": 1_700_000_000_000_000i64,
|
||||
"messageType": "SMS",
|
||||
"messageStatus": { "status": 100 },
|
||||
});
|
||||
let m = normalize_message(&raw).expect("normalize ok");
|
||||
assert_eq!(m.id, "msg-1");
|
||||
assert_eq!(m.thread_id.as_deref(), Some("thread-1"));
|
||||
assert_eq!(m.sender_id.as_deref(), Some("+15551234567"));
|
||||
assert!(!m.from_me);
|
||||
assert_eq!(m.text, "hello");
|
||||
assert_eq!(m.timestamp_unix, 1_700_000_000);
|
||||
assert_eq!(m.message_type.as_deref(), Some("sms"));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn normalize_outgoing_row_sets_from_me_and_blanks_sender() {
|
||||
let raw = json!({
|
||||
"messageId": "msg-2",
|
||||
"conversationId": "thread-1",
|
||||
"senderId": "+15559998888",
|
||||
"text": "yo",
|
||||
"timestamp": 1_700_000_005_000_000i64,
|
||||
"messageStatus": { "status": 4 },
|
||||
});
|
||||
let m = normalize_message(&raw).expect("normalize ok");
|
||||
assert!(m.from_me, "status=4 is OUTGOING_READ");
|
||||
assert!(m.sender_id.is_none(), "outgoing rows blank the sender");
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn normalize_accepts_legacy_second_precision_timestamp() {
|
||||
let raw = json!({
|
||||
"messageId": "msg-3",
|
||||
"conversationId": "thread-1",
|
||||
"senderId": "+15551234567",
|
||||
"text": "hi",
|
||||
"timestamp": 1_700_000_000i64,
|
||||
"messageStatus": { "status": 100 },
|
||||
});
|
||||
let m = normalize_message(&raw).expect("normalize ok");
|
||||
assert_eq!(m.timestamp_unix, 1_700_000_000);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn normalize_skips_row_missing_required_fields() {
|
||||
let raw = json!({
|
||||
"conversationId": "thread-1",
|
||||
"text": "no id",
|
||||
"timestamp": 1_700_000_000_000_000i64,
|
||||
});
|
||||
assert!(normalize_message(&raw).is_none());
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn normalize_conversation_row_with_participants() {
|
||||
let raw = json!({
|
||||
"conversationId": "thread-1",
|
||||
"name": "Family Group",
|
||||
"participantIds": ["+15551234567", "+15559998888"],
|
||||
});
|
||||
let c = normalize_conversation(&raw).expect("normalize ok");
|
||||
assert_eq!(c.thread_id, "thread-1");
|
||||
assert_eq!(c.display_name.as_deref(), Some("Family Group"));
|
||||
assert_eq!(c.participant_ids.len(), 2);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn normalize_participant_row_prefers_full_name() {
|
||||
let raw = json!({
|
||||
"participantId": "+15551234567",
|
||||
"fullName": "Alice Example",
|
||||
"firstName": "Alice",
|
||||
});
|
||||
let (id, name) = normalize_participant(&raw).expect("normalize ok");
|
||||
assert_eq!(id, "+15551234567");
|
||||
assert_eq!(name, "Alice Example");
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn normalize_participant_falls_back_to_first_name() {
|
||||
let raw = json!({
|
||||
"participantId": "+15551234567",
|
||||
"firstName": "Alice",
|
||||
});
|
||||
let (_, name) = normalize_participant(&raw).expect("normalize ok");
|
||||
assert_eq!(name, "Alice");
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn normalize_participant_returns_none_for_empty_name() {
|
||||
let raw = json!({
|
||||
"participantId": "+15551234567",
|
||||
});
|
||||
assert!(normalize_participant(&raw).is_none());
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn participant_map_roundtrip() {
|
||||
let mut pm = ParticipantMap::default();
|
||||
assert!(pm.is_empty());
|
||||
pm.insert("+15551234567".into(), "Alice".into());
|
||||
assert_eq!(pm.len(), 1);
|
||||
assert_eq!(pm.display_name("+15551234567").as_deref(), Some("Alice"));
|
||||
assert!(pm.display_name("unknown").is_none());
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,239 @@
|
||||
//! Google Messages Web scanner — Windows-focused, read-only IndexedDB walk.
|
||||
//!
|
||||
//! Scope for Stage 1:
|
||||
//! * Read-only scan of `bugle_db` (the IndexedDB database used by
|
||||
//! `messages.google.com/web`) via CDP on the embedded CEF webview.
|
||||
//! * One ingest call per `(thread_id, day)` group — same
|
||||
//! `openhuman.memory_doc_ingest` shape the iMessage and WhatsApp
|
||||
//! scanners already use.
|
||||
//! * No DOM automation. No send path. Send is deferred to a separate
|
||||
//! PR that will use OS Accessibility APIs (macOS AX / Windows UIA) —
|
||||
//! indistinguishable from a screen reader, so ToS-clean.
|
||||
//!
|
||||
//! Targeted at Windows + Android (the only practical combo for Google
|
||||
//! Messages — iPhone owners use iMessage, mac users typically use
|
||||
//! Messages Web in a browser tab that the CEF shell doesn't own). The
|
||||
//! code is windows-gated at module-scope; on other targets the public
|
||||
//! surface compiles to no-op stubs so `lib.rs` stays clean.
|
||||
//!
|
||||
//! History model differs from iMessage:
|
||||
//! * iMessage (#724) reads `chat.db` which holds FULL history locally.
|
||||
//! * Google Messages Web only caches in `bugle_db` what the web client
|
||||
//! has already synced. If the user never scrolled to older
|
||||
//! conversations, those pages aren't in IDB. Document this behavior
|
||||
//! in the UI — "scroll to backfill older history."
|
||||
//!
|
||||
//! CDP wiring TODO:
|
||||
//! * The CEF remote-debugging port + per-account target selection lives
|
||||
//! in `whatsapp_scanner::mod` today (`CDP_HOST`/`CDP_PORT`,
|
||||
//! `Target.getTargets` filter). When this module is promoted from
|
||||
//! scaffold to running scanner, lift that plumbing into a shared
|
||||
//! `cdp` module and point this scanner at the Google Messages Web
|
||||
//! target (`messages.google.com/web`). Until then `run_scanner` is a
|
||||
//! stub that logs and exits — the PR ships the normalization +
|
||||
//! memory-doc shape so downstream can iterate without the full CDP
|
||||
//! loop landed.
|
||||
|
||||
// Scaffold PR — orchestrator loop is a stub pending the shared CDP lift
|
||||
// from `whatsapp_scanner`. Once that lands and this module actually
|
||||
// drives `idb::walk` + `memory_doc_ingest`, drop the blanket allow below.
|
||||
#![allow(dead_code)]
|
||||
|
||||
#[cfg(target_os = "windows")]
|
||||
use std::sync::Arc;
|
||||
#[cfg(target_os = "windows")]
|
||||
use std::time::Duration;
|
||||
|
||||
#[cfg(target_os = "windows")]
|
||||
use parking_lot::Mutex;
|
||||
#[cfg(target_os = "windows")]
|
||||
use tauri::{AppHandle, Runtime};
|
||||
|
||||
pub mod idb;
|
||||
|
||||
#[cfg(target_os = "windows")]
|
||||
const SCAN_INTERVAL: Duration = Duration::from_secs(60);
|
||||
|
||||
/// Per-account scanner registry. Google Messages Web supports one paired
|
||||
/// phone per browser session; the registry shape is kept symmetric with
|
||||
/// the iMessage / WhatsApp scanners for future multi-account expansion.
|
||||
#[cfg(target_os = "windows")]
|
||||
pub struct ScannerRegistry {
|
||||
inner: Mutex<Option<tokio::task::JoinHandle<()>>>,
|
||||
}
|
||||
|
||||
#[cfg(target_os = "windows")]
|
||||
impl ScannerRegistry {
|
||||
pub fn new() -> Self {
|
||||
Self {
|
||||
inner: Mutex::new(None),
|
||||
}
|
||||
}
|
||||
|
||||
pub fn ensure_scanner<R: Runtime>(self: Arc<Self>, app: AppHandle<R>, account_id: String) {
|
||||
let mut guard = self.inner.lock();
|
||||
if guard.as_ref().map_or(false, |h| !h.is_finished()) {
|
||||
return;
|
||||
}
|
||||
let handle = tokio::spawn(run_scanner(app, account_id));
|
||||
*guard = Some(handle);
|
||||
}
|
||||
}
|
||||
|
||||
/// Stub loop — logs and exits. Wire CDP target discovery + `idb::walk`
|
||||
/// here once the shared `cdp` module is lifted from `whatsapp_scanner`.
|
||||
/// See module-level TODO.
|
||||
#[cfg(target_os = "windows")]
|
||||
async fn run_scanner<R: Runtime>(_app: AppHandle<R>, account_id: String) {
|
||||
log::info!(
|
||||
"[gmessages] scanner scaffold loaded account={} interval={:?} — CDP wiring pending",
|
||||
account_id,
|
||||
SCAN_INTERVAL
|
||||
);
|
||||
}
|
||||
|
||||
// Non-Windows stub so the rest of the app compiles unchanged on mac/linux.
|
||||
#[cfg(not(target_os = "windows"))]
|
||||
pub struct ScannerRegistry;
|
||||
|
||||
#[cfg(not(target_os = "windows"))]
|
||||
impl ScannerRegistry {
|
||||
pub fn new() -> Self {
|
||||
Self
|
||||
}
|
||||
pub fn ensure_scanner<R: tauri::Runtime>(
|
||||
self: std::sync::Arc<Self>,
|
||||
_app: tauri::AppHandle<R>,
|
||||
_account_id: String,
|
||||
) {
|
||||
}
|
||||
}
|
||||
|
||||
/// Format a list of normalized messages into a transcript string suitable
|
||||
/// for `memory_doc_ingest.content`. Matches the iMessage scanner output
|
||||
/// shape so Neocortex sees a uniform format across channels.
|
||||
pub fn format_transcript(messages: &[idb::Message], participants: &idb::ParticipantMap) -> String {
|
||||
let mut out = String::new();
|
||||
for m in messages {
|
||||
let sender = if m.from_me {
|
||||
"me".to_string()
|
||||
} else {
|
||||
m.sender_id
|
||||
.as_deref()
|
||||
.and_then(|sid| participants.display_name(sid))
|
||||
.unwrap_or_else(|| m.sender_id.clone().unwrap_or_else(|| "unknown".into()))
|
||||
};
|
||||
let text = m.text.replace('\n', " ");
|
||||
let body = if text.is_empty() {
|
||||
"[non-text]".to_string()
|
||||
} else {
|
||||
text
|
||||
};
|
||||
out.push_str(&format!("[{}] {}: {}\n", m.timestamp_unix, sender, body));
|
||||
}
|
||||
out
|
||||
}
|
||||
|
||||
/// Group a flat list of messages into `(thread_id, YYYY-MM-DD) -> Vec<Message>`.
|
||||
/// Day bucketing uses the local timezone — users inspect memory docs by
|
||||
/// their calendar day, not UTC (same policy as iMessage #724 after the
|
||||
/// CodeRabbit local-TZ fix).
|
||||
pub fn group_by_thread_day(
|
||||
messages: Vec<idb::Message>,
|
||||
) -> Vec<((String, String), Vec<idb::Message>)> {
|
||||
use std::collections::BTreeMap;
|
||||
let mut groups: BTreeMap<(String, String), Vec<idb::Message>> = BTreeMap::new();
|
||||
for m in messages {
|
||||
let Some(thread_id) = m.thread_id.clone() else {
|
||||
continue;
|
||||
};
|
||||
let day = seconds_to_ymd(m.timestamp_unix);
|
||||
groups.entry((thread_id, day)).or_default().push(m);
|
||||
}
|
||||
groups.into_iter().collect()
|
||||
}
|
||||
|
||||
/// Local-timezone day bucket for a unix-second timestamp. Returns
|
||||
/// "YYYY-MM-DD" or "unknown" for values that fall outside chrono's range.
|
||||
pub fn seconds_to_ymd(secs: i64) -> String {
|
||||
use chrono::{Local, TimeZone};
|
||||
Local
|
||||
.timestamp_opt(secs, 0)
|
||||
.single()
|
||||
.map(|dt| dt.format("%Y-%m-%d").to_string())
|
||||
.unwrap_or_else(|| "unknown".into())
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
|
||||
fn msg(id: &str, thread: &str, ts: i64, text: &str, from_me: bool) -> idb::Message {
|
||||
idb::Message {
|
||||
id: id.into(),
|
||||
thread_id: Some(thread.into()),
|
||||
sender_id: if from_me {
|
||||
None
|
||||
} else {
|
||||
Some("+15551234567".into())
|
||||
},
|
||||
from_me,
|
||||
text: text.into(),
|
||||
timestamp_unix: ts,
|
||||
message_type: Some("sms".into()),
|
||||
}
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn group_by_thread_day_buckets_messages_correctly() {
|
||||
// Two messages ~5s apart in the same thread should fall into one
|
||||
// group; a third in a different thread into its own group.
|
||||
let base = 1_700_000_000;
|
||||
let msgs = vec![
|
||||
msg("1", "t1", base, "hi", false),
|
||||
msg("2", "t1", base + 5, "yo", true),
|
||||
msg("3", "t2", base, "other", false),
|
||||
];
|
||||
let groups = group_by_thread_day(msgs);
|
||||
assert_eq!(groups.len(), 2);
|
||||
let t1 = groups.iter().find(|((t, _), _)| t == "t1").unwrap();
|
||||
assert_eq!(t1.1.len(), 2);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn format_transcript_includes_sender_and_body() {
|
||||
let msgs = vec![
|
||||
msg("1", "t1", 1_700_000_000, "hi", false),
|
||||
msg("2", "t1", 1_700_000_005, "yo", true),
|
||||
];
|
||||
let participants = idb::ParticipantMap::default();
|
||||
let t = format_transcript(&msgs, &participants);
|
||||
assert!(t.contains("hi"));
|
||||
assert!(t.contains("me: yo"));
|
||||
assert!(t.contains("+15551234567: hi"));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn format_transcript_resolves_display_name_from_participants() {
|
||||
let mut participants = idb::ParticipantMap::default();
|
||||
participants.insert("+15551234567".into(), "Alice".into());
|
||||
let msgs = vec![msg("1", "t1", 1_700_000_000, "hi", false)];
|
||||
let t = format_transcript(&msgs, &participants);
|
||||
assert!(t.contains("Alice: hi"), "got {:?}", t);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn format_transcript_marks_empty_body_as_non_text() {
|
||||
let msgs = vec![msg("1", "t1", 1_700_000_000, "", false)];
|
||||
let t = format_transcript(&msgs, &idb::ParticipantMap::default());
|
||||
assert!(t.contains("[non-text]"), "got {:?}", t);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn seconds_to_ymd_shape() {
|
||||
let out = seconds_to_ymd(1_700_000_000);
|
||||
assert_eq!(out.len(), 10);
|
||||
assert_eq!(&out[4..5], "-");
|
||||
assert_eq!(&out[7..8], "-");
|
||||
}
|
||||
}
|
||||
@@ -9,6 +9,8 @@ mod core_update;
|
||||
mod discord_scanner;
|
||||
mod gmail;
|
||||
#[cfg(feature = "cef")]
|
||||
mod gmessages_scanner;
|
||||
#[cfg(feature = "cef")]
|
||||
mod imessage_scanner;
|
||||
mod notification_settings;
|
||||
#[cfg(feature = "cef")]
|
||||
@@ -626,6 +628,10 @@ pub fn run() {
|
||||
.manage(notification_settings::NotificationSettingsState::new());
|
||||
#[cfg(feature = "cef")]
|
||||
let builder = builder.manage(std::sync::Arc::new(imessage_scanner::ScannerRegistry::new()));
|
||||
#[cfg(feature = "cef")]
|
||||
let builder = builder.manage(std::sync::Arc::new(
|
||||
gmessages_scanner::ScannerRegistry::new(),
|
||||
));
|
||||
let builder = builder.manage(whatsapp_scanner::ScannerRegistry::new());
|
||||
#[cfg(feature = "cef")]
|
||||
let builder = builder.manage(std::sync::Arc::new(slack_scanner::ScannerRegistry::new()));
|
||||
|
||||
Reference in New Issue
Block a user