//! WhatsApp Web scanner driven over the Chrome DevTools Protocol (CDP).
//!
//! Attaches to the embedded CEF webview via the in-process CDP transport
//! installed by `webview_accounts::open` (no TCP listener). Per tracked
//! WhatsApp-account webview, two interleaved loops run:
//!
//!   * **Fast tick** (`FAST_SCAN_INTERVAL`, 2s) — `dom_scan.js` scrapes
//!     rendered `[data-id]` message rows from the DOM. Emits only when
//!     the visible-set hash changes so idle windows stay silent.
//!   * **Full tick** (`FULL_SCAN_INTERVAL`, 30s) — `scanner.js` walks
//!     WhatsApp's IndexedDB stores (model-storage, signal-storage, …) to
//!     pull message metadata, chat names, contact names.
//!
//! Each scan groups messages by `(chatId, day)` and posts one
//! `openhuman.memory_doc_ingest` JSON-RPC call per group to the core, so
//! each day of a conversation upserts a single memory doc. We also emit
//! `webview:event` ingest events so any React UI listening can update
//! live when the main window is open.
//!
//! NOTE: only meaningful with the `cef` feature — the wry runtime does
//! not expose a remote debugging port. Compile-gated at the call site.

use std::sync::Arc;
use std::time::{Duration, Instant};

use parking_lot::Mutex;
use serde_json::{json, Value};
use tauri::{AppHandle, Emitter, Runtime};
use tokio::task::AbortHandle;
use tokio::time::sleep;

use crate::cdp::CdpConn;

mod dom_snapshot;
#[cfg(test)]
mod dom_snapshot_tests;
mod idb;

/// Cadence for the expensive full scan — pages the whole IDB via CDP and
/// captures a fresh DOM snapshot. Each pass serialises thousands of
/// message records, so we pay this cost infrequently.
const FULL_SCAN_INTERVAL: Duration = Duration::from_secs(30);
/// Cadence for the cheap fast scan (DOM `[data-id]` scrape only). Runs at
/// Franz-like 2s so the ingest stream feels live — each tick captures the
/// DOM via `DOMSnapshot.captureSnapshot` (pure CDP, no page-world JS).
const FAST_SCAN_INTERVAL: Duration = Duration::from_secs(2);

/// Product of one full scan — IDB walk (via `idb::walk`) joined with a
/// DOM snapshot (via `dom_snapshot::capture_messages`). `messages` carries
/// IDB-sourced metadata only; DOM-sourced bodies are merged in by id at
/// emit time (see `emit_snapshot`).
#[derive(Debug, Clone, Default)]
pub struct ScanSnapshot {
    pub ok: bool,
    pub error: Option<String>,
    /// `jid → display name`, drawn from chat/contact/group-metadata stores.
    pub chats: serde_json::Map<String, Value>,
    /// Normalised message metadata (no bodies — see note above).
    pub messages: Vec<Value>,
    /// DOM-scraped rendered bodies; merged into `messages` by id.
    pub dom_messages: Vec<Value>,
    /// Active chat's display name parsed from
    /// `header[data-testid="conversation-header"]`. Used by the merge step
    /// to reverse-look-up `chatId` for DOM rows that lack one (modern
    /// WhatsApp Web doesn't expose chat JID anywhere on the message rows).
    pub active_chat_name: Option<String>,
    /// Per-stage telemetry from the DOM-capture pass (issue #1376). `None`
    /// when the DOM call failed entirely (e.g. CDP disconnect mid-scan);
    /// otherwise carries the same `rows` as `dom_messages` plus counters
    /// disambiguating "0 rows" from "rows with no body".
    pub capture_report: Option<dom_snapshot::CaptureReport>,
}

/// Spawn a per-account CDP poller. Idempotent at call site (caller tracks
/// account → JoinHandle if it cares about cancellation).
///
/// The scanner runs two interleaved loops:
///   * **Fast tick** (`FAST_SCAN_INTERVAL`, 2s) — cheap DOM scrape. Only
///     emits an ingest event when the visible-row hash changes, so idle
///     windows don't spam the UI.
///   * **Full tick** (`FULL_SCAN_INTERVAL`, 30s) — the expensive IDB walk
///     + spy/keystore snapshot. Always emits.
///
/// Both ticks share the same `webview:event` ingest envelope so downstream
/// consumers don't need to care which one produced the event.
pub fn spawn_scanner<R: Runtime>(
    app: AppHandle<R>,
    account_id: String,
    url_prefix: String,
) -> Vec<AbortHandle> {
    let task = tokio::spawn(async move {
        let fragment = crate::cdp::target_url_fragment(&account_id);
        log::info!(
            "[wa] scanner up account={} url_prefix={} fragment={} fast={:?} full={:?}",
            account_id,
            url_prefix,
            fragment,
            FAST_SCAN_INTERVAL,
            FULL_SCAN_INTERVAL
        );
        // Wait a moment for the page to actually load + log in. We'd rather
        // miss the first cycle than thrash the CDP endpoint while the
        // target isn't even there yet.
        sleep(Duration::from_secs(5)).await;
        let mut last_dom_hash: Option<u64> = None;
        let mut last_full: Instant = Instant::now()
            .checked_sub(FULL_SCAN_INTERVAL)
            .unwrap_or_else(Instant::now);
        loop {
            // Gate: run a full IDB scan if enough time has elapsed,
            // otherwise run the cheap DOM-only scan.
            let do_full = last_full.elapsed() >= FULL_SCAN_INTERVAL;
            if !do_full {
                match scan_dom_once(&app, &account_id, &url_prefix, &fragment).await {
                    Ok(dom) => {
                        let changed =
                            last_dom_hash != Some(dom.hash) && !dom.dom_messages.is_empty();
                        if changed {
                            log::info!(
                                "[wa][{}] fast dom-scan rows={} hash={} (changed)",
                                account_id,
                                dom.dom_messages.len(),
                                dom.hash
                            );
                            emit_dom_only(&app, &account_id, &dom.dom_messages);
                            last_dom_hash = Some(dom.hash);
                        }
                    }
                    Err(e) => {
                        log::debug!("[wa][{}] dom-scan failed: {}", account_id, e);
                    }
                }
                sleep(FAST_SCAN_INTERVAL).await;
                continue;
            }
            last_full = Instant::now();
            match scan_once(&app, &account_id, &url_prefix, &fragment).await {
                Ok(snap) => {
                    // Per-stage telemetry from the DOM capture (#1376) —
                    // disambiguates "no rows seen" from "rows seen but body
                    // extraction returned empty" from "active chat header
                    // unresolved (downstream filter will drop the rows)".
                    let (seen, with_body, no_body, chat_resolved) = match &snap.capture_report {
                        Some(r) => (
                            r.rows_seen,
                            r.rows_with_body,
                            r.rows_dropped_no_body,
                            r.active_chat_resolved,
                        ),
                        None => (0, 0, 0, false),
                    };
                    log::info!(
                        "[wa][{}] full scan ok messages={} chats={} dom={} (seen={} with_body={} no_body={} chat_resolved={})",
                        account_id,
                        snap.messages.len(),
                        snap.chats.len(),
                        snap.dom_messages.len(),
                        seen,
                        with_body,
                        no_body,
                        chat_resolved,
                    );
                    // Preview a few DOM-scraped rows so it's obvious from the
                    // log whether the active chat produced fresh bodies.
                    for (i, dm) in snap.dom_messages.iter().take(5).enumerate() {
                        let chat = dm.get("chatId").and_then(|v| v.as_str()).unwrap_or("?");
                        let msg = dm.get("msgId").and_then(|v| v.as_str()).unwrap_or("?");
                        let from_me = dm.get("fromMe").and_then(|v| v.as_bool()).unwrap_or(false);
                        let author = dm.get("author").and_then(|v| v.as_str()).unwrap_or("");
                        let ts = dm
                            .get("preTimestamp")
                            .and_then(|v| v.as_str())
                            .unwrap_or("");
                        let body = dm.get("body").and_then(|v| v.as_str()).unwrap_or("");
                        let preview: String = body.chars().take(120).collect();
                        log::info!(
                            "[wa][{}] dom#{} chat={} msg={} fromMe={} [{}] {}: {:?}",
                            account_id,
                            i + 1,
                            chat,
                            msg,
                            from_me,
                            ts,
                            author,
                            preview
                        );
                    }
                    // TRACE-level structured row dump (first 3 rows, ≤120 char
                    // snippets) so a developer chasing #1376-style "dom=0 but
                    // bodies exist" can see exactly what the parser produced
                    // without re-instrumenting. Truncation lives in
                    // `dom_snapshot::text_snippet_preview` to honor the
                    // CLAUDE.md "no PII in trace dumps" rule.
                    if let Some(report) = snap.capture_report.as_ref() {
                        for (i, row) in report.rows.iter().take(3).enumerate() {
                            let pairs = dom_snapshot::text_snippet_preview(row, 120);
                            for (k, v) in &pairs {
                                log::trace!(
                                    "[wa][{}] dom-trace#{} {}={:?}",
                                    account_id,
                                    i + 1,
                                    k,
                                    v
                                );
                            }
                        }
                    }
                    emit_snapshot(&app, &account_id, &snap);
                }
                Err(e) => {
                    log::warn!("[wa][{}] scan failed: {}", account_id, e);
                }
            }
            // After a full scan, go back to fast-tick cadence until the
            // next `FULL_SCAN_INTERVAL` elapses.
            sleep(FAST_SCAN_INTERVAL).await;
        }
    });
    vec![task.abort_handle()]
}

/// Emit an ingest payload carrying only DOM-scraped rows, grouped by
/// (chatId, day) so React can upsert each day's transcript into memory.
fn emit_dom_only<R: Runtime>(app: &AppHandle<R>, account_id: &str, dom: &[Value]) {
    // Use the most recent contact-names snapshot from a full IDB scan so
    // DOM-only rows get resolved display names too.
    let names = contact_cache_get(account_id);
    emit_grouped_whatsapp(app, account_id, dom, &names, "cdp-dom");
}

/// Per-account snapshot of `{jid -> display name}`. Populated on every
/// full IDB scan (from chats / contacts / group-metadata stores) and read
/// by fast DOM-only ticks so the transcript lines show names instead of
/// raw JIDs even when the scrape comes from the DOM.
fn contact_cache(
) -> &'static std::sync::Mutex<std::collections::HashMap<String, serde_json::Map<String, Value>>> {
    use std::sync::OnceLock;
    static CACHE: OnceLock<
        std::sync::Mutex<std::collections::HashMap<String, serde_json::Map<String, Value>>>,
    > = OnceLock::new();
    CACHE.get_or_init(|| std::sync::Mutex::new(Default::default()))
}

fn contact_cache_put(account_id: &str, names: &serde_json::Map<String, Value>) {
    if names.is_empty() {
        return;
    }
    let mut g = contact_cache().lock().unwrap();
    g.insert(account_id.to_string(), names.clone());
}

fn contact_cache_get(account_id: &str) -> serde_json::Map<String, Value> {
    let g = contact_cache().lock().unwrap();
    g.get(account_id).cloned().unwrap_or_default()
}

fn chrono_now_millis() -> i64 {
    std::time::SystemTime::now()
        .duration_since(std::time::UNIX_EPOCH)
        .map(|d| d.as_millis() as i64)
        .unwrap_or(0)
}

/// Normalize a chat display name for active-chat → JID matching (issue #1376).
/// Lowercases, strips every non-ASCII-alphanumeric code point (drops emoji,
/// punctuation, dashes, separators), and collapses internal whitespace runs
/// to nothing — leaves `[a-z0-9]*`. Lets the matcher find a match when the
/// DOM-parsed header text drifts slightly from the IDB-stored chat name
/// (extra spaces, trailing emoji, hyphenation differences).
pub(crate) fn normalize_chat_name(s: &str) -> String {
    s.chars()
        .filter_map(|c| {
            if c.is_ascii_alphanumeric() {
                Some(c.to_ascii_lowercase())
            } else {
                None
            }
        })
        .collect()
}

async fn scan_once<R: Runtime>(
    app: &AppHandle<R>,
    account_id: &str,
    url_prefix: &str,
    url_fragment: &str,
) -> Result<ScanSnapshot, String> {
    // One CDP attach per tick — we attach to the WhatsApp page session via
    // the account's in-process CDP transport, run the IDB walk + DOM
    // snapshot, then detach (which frees every RemoteObject the IDB walk
    // materialised, so no per-object releases).
    let url_prefix_owned = url_prefix.to_string();
    let url_fragment_owned = url_fragment.to_string();
    let pred = move |t: &crate::cdp::target::CdpTarget| -> bool {
        t.url.starts_with(&url_prefix_owned) && t.url.ends_with(&url_fragment_owned)
    };
    let (mut cdp, page_session) =
        crate::cdp::target::connect_and_attach_matching_in_process::<R, _>(app, account_id, pred)
            .await
            .map_err(|e| format!("attach: {e} (prefix={url_prefix} fragment={url_fragment})"))?;

    // IDB + DOM are independent — run IDB first (the heavier of the two)
    // so a DOM failure doesn't mask IDB errors. Errors are captured on
    // `snap.error` instead of bubbling so the caller can still act on
    // whatever partial data came back.
    let mut snap = ScanSnapshot {
        ok: true,
        ..Default::default()
    };
    match idb::walk(&mut cdp, &page_session, url_prefix).await {
        Ok((messages, chat_names)) => {
            snap.messages = messages.iter().map(idb::IdbMessage::to_json).collect();
            snap.chats = chat_names
                .into_iter()
                .map(|(k, v)| (k, Value::String(v)))
                .collect();
        }
        Err(e) => {
            snap.ok = false;
            snap.error = Some(format!("idb walk: {e}"));
            log::warn!("[wa][{}] idb walk failed: {}", account_id, e);
        }
    }
    let mut capture_report: Option<dom_snapshot::CaptureReport> = None;
    match dom_snapshot::capture_messages(&mut cdp, &page_session).await {
        Ok(report) => {
            snap.dom_messages = report
                .rows
                .iter()
                .map(dom_snapshot::DomMessage::to_json)
                .collect();
            snap.active_chat_name = report.active_chat_name.clone();
            capture_report = Some(report);
        }
        Err(e) => {
            // Fast-tick DOM scans will retry every 2s, so degrade gracefully.
            log::warn!("[wa][{}] dom snapshot failed: {}", account_id, e);
        }
    }

    let _ = cdp
        .call(
            "Target.detachFromTarget",
            json!({ "sessionId": page_session }),
            None,
        )
        .await;
    let _ = app;
    snap.capture_report = capture_report;
    Ok(snap)
}

/// Result of a fast DOM-only scan. Small enough to bounce back every 2s.
#[derive(Debug, Default)]
pub struct DomScanResult {
    pub dom_messages: Vec<Value>,
    pub hash: u64,
}

/// Fast tick: open a CDP session, attach to the WhatsApp page, snapshot
/// the DOM via `DOMSnapshot.captureSnapshot`, detach. No IDB, no worker
/// enumeration, no JavaScript runs in the page — the snapshot is produced
/// at the browser's C++ layer. The flat-array response is parsed in Rust
/// (see `dom_snapshot.rs`).
async fn scan_dom_once<R: Runtime>(
    app: &AppHandle<R>,
    account_id: &str,
    url_prefix: &str,
    url_fragment: &str,
) -> Result<DomScanResult, String> {
    let url_prefix_owned = url_prefix.to_string();
    let url_fragment_owned = url_fragment.to_string();
    let pred = move |t: &crate::cdp::target::CdpTarget| -> bool {
        t.url.starts_with(&url_prefix_owned) && t.url.ends_with(&url_fragment_owned)
    };
    let (mut cdp, page_session) =
        crate::cdp::target::connect_and_attach_matching_in_process::<R, _>(app, account_id, pred)
            .await
            .map_err(|e| format!("attach: {e} (prefix={url_prefix} fragment={url_fragment})"))?;
    let captured = dom_snapshot::capture_messages(&mut cdp, &page_session).await;
    // Detach no matter what — otherwise dangling sessions pile up on long
    // runs and eventually the CDP endpoint refuses new attachments.
    let _ = cdp
        .call(
            "Target.detachFromTarget",
            json!({ "sessionId": page_session }),
            None,
        )
        .await;
    let report = captured?;
    let dom_messages: Vec<Value> = report
        .rows
        .iter()
        .map(dom_snapshot::DomMessage::to_json)
        .collect();
    log::debug!(
        "[wa][{}] fast dom-scan rows={} hash={} (seen={} with_body={} no_body={} chat_resolved={})",
        account_id,
        dom_messages.len(),
        report.hash,
        report.rows_seen,
        report.rows_with_body,
        report.rows_dropped_no_body,
        report.active_chat_resolved,
    );
    Ok(DomScanResult {
        dom_messages,
        hash: report.hash,
    })
}

/// Forward the snapshot to React via the same `webview:event` channel
/// recipe ingest already uses. UI code can listen for kind == "ingest".
fn emit_snapshot<R: Runtime>(app: &AppHandle<R>, account_id: &str, snap: &ScanSnapshot) {
    if !snap.ok {
        log::warn!(
            "[wa][{}] snapshot not ok (idb walk failed: {:?}) — falling through with dom-only data",
            account_id,
            snap.error
        );
        // 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
    // anywhere on individual message rows or in the URL, so this is the
    // only signal we have. The IDB-side `chats` map has `name → jid` (we
    // store it as `jid → {name, …}`, so iterate). Match prefers exact
    // case-sensitive equality and falls back to case-insensitive; ignore
    // ambiguous matches (multiple chats with the same display name) so we
    // don't mis-attribute messages.
    let active_chat_jid: Option<String> = snap.active_chat_name.as_deref().and_then(|name| {
        let name_lc = name.to_ascii_lowercase();
        let name_norm = normalize_chat_name(name);
        let mut exact: Vec<&str> = Vec::new();
        let mut ci: Vec<&str> = Vec::new();
        let mut normalized: Vec<&str> = Vec::new();
        let mut substring: Vec<&str> = Vec::new();
        for (jid, chat) in snap.chats.iter() {
            let chat_name = chat.get("name").and_then(|v| v.as_str()).unwrap_or("");
            if chat_name == name {
                exact.push(jid);
            } else if !chat_name.is_empty() && chat_name.to_ascii_lowercase() == name_lc {
                ci.push(jid);
            } else if !chat_name.is_empty() {
                // Normalized tier (issue #1376): strip non-alphanumeric +
                // collapse whitespace + lowercase, then compare equality.
                // Catches drift introduced by emoji, punctuation, or
                // double-spaces between the DOM header text and the IDB
                // chat name (e.g. group titles like "17-18-19 July samagam"
                // vs "17 18 19 July samagam ✨"). Substring stays as a final
                // fallback so loose matches are tried last.
                let chat_norm = normalize_chat_name(chat_name);
                if !name_norm.is_empty() && chat_norm == name_norm {
                    normalized.push(jid);
                } else if chat_name.to_ascii_lowercase().contains(&name_lc)
                    || name_lc.contains(&chat_name.to_ascii_lowercase())
                {
                    substring.push(jid);
                }
            }
        }
        // Prefer exact > case-insensitive > normalized > substring. Each
        // tier only wins when it has exactly one candidate (avoids
        // cross-attribution when many chats share a token like a common
        // first name). Tier counts feed into the warn log so ambiguous
        // resolutions are visible in the scanner output.
        match (exact.len(), ci.len(), normalized.len(), substring.len()) {
            (1, _, _, _) => Some(exact[0].to_string()),
            (0, 1, _, _) => Some(ci[0].to_string()),
            (0, 0, 1, _) => Some(normalized[0].to_string()),
            (0, 0, 0, 1) => Some(substring[0].to_string()),
            (0, 0, 0, 0) => {
                // No IDB chat matches the active-chat header — happens for
                // 1:1 chats where IDB stores the peer JID but the
                // human-readable contact name lives in the device address
                // book (never reaches IDB). Synthesize a stable
                // `dom:<normalized-name>` backfill key so DOM rows still
                // persist instead of being filtered at the structured-store
                // empty-`chat_id` guard. Distinct from real WhatsApp JIDs
                // (which always contain `@`), so downstream consumers can
                // tell DOM-only chat ids apart.
                let synth = normalize_chat_name(name);
                if synth.is_empty() {
                    None
                } else {
                    Some(format!("dom:{synth}"))
                }
            }
            (e, c, n, s) => {
                log::warn!(
                    "[whatsapp_scanner] ambiguous active-chat resolution: {} candidates for '{}' — skipping backfill",
                    e + c + n + s,
                    name
                );
                None
            }
        }
    });
    log::info!(
        "[wa][{}] active chat resolution: name={:?} → jid={:?} chats_in_map={}",
        account_id,
        snap.active_chat_name,
        active_chat_jid,
        snap.chats.len()
    );
    // Join DOM-scraped bodies into the messages list by msgId. WhatsApp
    // 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.
    // 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,
            patched,
            appended,
            messages.len()
        );
    }
    // Cache the contact/chat name map so the next fast DOM-only tick can
    // resolve sender JIDs → display names without re-walking IDB.
    contact_cache_put(account_id, &snap.chats);
    // Also emit one grouped `whatsapp` ingest event per (chatId, day) so
    // the React listener can call `openhuman.memory_doc_ingest` with a
    // stable namespace/key that upserts cleanly.
    emit_grouped_whatsapp(app, account_id, &messages, &snap.chats, "cdp-indexeddb");
}

/// Parse a unix-seconds timestamp to a UTC `YYYY-MM-DD` string. Uses the
/// Howard Hinnant civil-from-days algorithm — no external deps.
fn seconds_to_ymd(secs: i64) -> String {
    let days = secs.div_euclid(86_400);
    let z = days + 719_468;
    let era = if z >= 0 { z } else { z - 146_096 } / 146_097;
    let doe = (z - era * 146_097) as u64;
    let yoe = (doe - doe / 1460 + doe / 36_524 - doe / 146_096) / 365;
    let y = yoe as i64 + era * 400;
    let doy = doe - (365 * yoe + yoe / 4 - yoe / 100);
    let mp = (5 * doy + 2) / 153;
    let d = (doy - (153 * mp + 2) / 5 + 1) as u32;
    let m = (if mp < 10 { mp + 3 } else { mp - 9 }) as u32;
    let y_real = (if m <= 2 { y + 1 } else { y }) as i32;
    format!("{:04}-{:02}-{:02}", y_real, m, d)
}

/// Parse WA's `data-pre-plain-text` timestamp (e.g. `"4:53 AM, 7/5/2025"`)
/// to `YYYY-MM-DD`. Returns None if the format doesn't match.
fn parse_pre_timestamp_ymd(s: &str) -> Option<String> {
    // Everything after the first comma is the date: "4:53 AM, 7/5/2025"
    let (_, date_part) = s.split_once(',')?;
    let date_part = date_part.trim();
    let parts: Vec<&str> = date_part.split('/').collect();
    if parts.len() != 3 {
        return None;
    }
    let m: u32 = parts[0].trim().parse().ok()?;
    let d: u32 = parts[1].trim().parse().ok()?;
    let y: i32 = parts[2].trim().parse().ok()?;
    if !(1..=12).contains(&m) || !(1..=31).contains(&d) || !(1900..=3000).contains(&y) {
        return None;
    }
    Some(format!("{:04}-{:02}-{:02}", y, m, d))
}

/// Group messages by (chatId, day) and emit one `webview:event` per group
/// matching the shape `persistWhatsappChatDay` (React) consumes. React in
/// turn calls `openhuman.memory_doc_ingest` to upsert each day's transcript
/// into the memory layer.
fn emit_grouped_whatsapp<R: Runtime>(
    app: &AppHandle<R>,
    account_id: &str,
    messages: &[Value],
    chats: &serde_json::Map<String, Value>,
    source: &str,
) {
    use std::collections::HashMap;
    let now_secs = std::time::SystemTime::now()
        .duration_since(std::time::UNIX_EPOCH)
        .map(|d| d.as_secs() as i64)
        .unwrap_or(0);

    // Group: (chatId, day) -> Vec<normalized message>
    let mut groups: HashMap<(String, String), Vec<Value>> = 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,
        };
        // Require body — memory docs without content are noise.
        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;
        }

        // Derive day + canonical timestamp (seconds).
        let (day, ts_secs): (String, i64) =
            if let Some(t) = m.get("timestamp").and_then(|v| v.as_i64()) {
                (seconds_to_ymd(t), t)
            } else if let Some(pre) = m.get("preTimestamp").and_then(|v| v.as_str()) {
                match parse_pre_timestamp_ymd(pre) {
                    Some(d) => (d, now_secs),
                    None => (seconds_to_ymd(now_secs), now_secs),
                }
            } else {
                (seconds_to_ymd(now_secs), now_secs)
            };

        // React expects `fromMe`, `from`, `body`, `timestamp` (sec), `type`.
        let from_me = m.get("fromMe").and_then(|v| v.as_bool()).unwrap_or(false);
        let raw_from: Option<String> = m
            .get("from")
            .and_then(|v| v.as_str())
            .map(|s| s.to_string());
        // Prefer: chats[from].name → DOM `author` (parsed from data-pre-plain-text)
        //       → chats[chatId].name (1:1 chats where chatId == sender)
        //       → raw JID as last resort.
        let author_from_dom = m
            .get("author")
            .and_then(|v| v.as_str())
            .filter(|s| !s.is_empty())
            .map(|s| s.to_string());
        let resolved_name: Option<String> = raw_from
            .as_ref()
            .and_then(|jid| {
                chats
                    .get(jid)
                    .and_then(|c| c.get("name"))
                    .and_then(|v| v.as_str())
                    .map(|s| s.to_string())
            })
            .or(author_from_dom)
            .or_else(|| {
                chats
                    .get(&chat_id)
                    .and_then(|c| c.get("name"))
                    .and_then(|v| v.as_str())
                    .map(|s| s.to_string())
            });
        // `from` field keeps the JID so downstream code can key by it;
        // `fromName` carries the human-readable label for the transcript.
        let from_value = raw_from
            .clone()
            .or_else(|| resolved_name.clone())
            .unwrap_or_default();
        let id = m
            .get("id")
            .cloned()
            .or_else(|| m.get("dataId").cloned())
            .unwrap_or(Value::Null);
        let type_ = m.get("type").cloned().unwrap_or(Value::Null);
        let normalized = json!({
            "id": id,
            "chatId": chat_id.clone(),
            "fromMe": from_me,
            "from": from_value,
            "fromName": resolved_name,
            "body": body,
            "timestamp": ts_secs,
            "type": type_,
        });
        groups.entry((chat_id, day)).or_default().push(normalized);
    }

    // Emit one event per (chatId, day). Match envelope shape React expects
    // so when the main window IS open the UI updates live. In parallel we
    // POST the same payload directly to the core RPC so the memory write
    // happens regardless of whether the React listener is attached.
    let mut emitted = 0usize;
    for ((chat_id, day), msgs) in groups {
        let chat_name = chats
            .get(&chat_id)
            .and_then(|c| c.get("name"))
            .and_then(|v| v.as_str())
            .unwrap_or(&chat_id)
            .to_string();
        let payload = json!({
            "provider": "whatsapp",
            "source": source,
            "chatId": chat_id,
            "chatName": chat_name,
            "day": day,
            "messages": msgs,
        });
        let envelope = json!({
            "account_id": account_id,
            "provider": "whatsapp",
            "kind": "ingest",
            "payload": payload.clone(),
            "ts": chrono_now_millis(),
        });
        if let Err(e) = app.emit("webview:event", &envelope) {
            log::warn!("[wa][{}] ingest emit failed: {}", account_id, e);
        } else {
            emitted += 1;
        }
        // Direct memory write via core RPC — fire-and-forget so the
        // scanner tick doesn't block on HTTP.
        let acct = account_id.to_string();
        tokio::spawn(async move {
            if let Err(e) = post_memory_doc_ingest(&acct, &payload).await {
                log::warn!("[wa][{}] memory write failed: {}", acct, e);
            }
        });
    }
    if emitted > 0 {
        log::info!(
            "[wa][{}] emitted {} ingest group(s) source={}",
            account_id,
            emitted,
            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<Value> = 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);
                // Per-row source tag (issue #1376): rows that picked up their
                // body from the DOM scrape get tagged `cdp-dom` so the
                // structured store can distinguish DOM-sourced text from
                // IDB-sourced metadata. `merge_dom_into_snapshot` stamps
                // `bodySource = "dom"` (IDB row patched with DOM body) and
                // `"dom-only"` (DOM row with no IDB peer); both cases fall
                // back to `cdp-dom`. Everything else inherits the caller's
                // tag (full-scan = `cdp-indexeddb`, fast-tick = `cdp-dom`).
                let row_source = m
                    .get("bodySource")
                    .and_then(|v| v.as_str())
                    .filter(|s| matches!(*s, "dom" | "dom-only"))
                    .map(|_| "cdp-dom")
                    .unwrap_or(source);
                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": row_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 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<Value> {
    let chat_id = ingest
        .get("chatId")
        .and_then(|v| v.as_str())
        .unwrap_or_default();
    let day = ingest
        .get("day")
        .and_then(|v| v.as_str())
        .unwrap_or_default();
    let chat_name = ingest
        .get("chatName")
        .and_then(|v| v.as_str())
        .unwrap_or(chat_id);
    let empty: Vec<Value> = Vec::new();
    let msgs: &Vec<Value> = ingest
        .get("messages")
        .and_then(|v| v.as_array())
        .unwrap_or(&empty);
    if chat_id.is_empty() || day.is_empty() || msgs.is_empty() {
        return None;
    }

    // Build a stable transcript — sorted by timestamp, one line per msg.
    let mut sorted: Vec<&Value> = msgs.iter().collect();
    sorted.sort_by_key(|m| m.get("timestamp").and_then(|v| v.as_i64()).unwrap_or(0));
    let transcript: String = sorted
        .iter()
        .map(|m| {
            let ts = m.get("timestamp").and_then(|v| v.as_i64()).unwrap_or(0);
            let hhmm = if ts > 0 {
                let secs_of_day = (ts.rem_euclid(86_400)) as u32;
                format!("{:02}:{:02}Z", secs_of_day / 3600, (secs_of_day / 60) % 60)
            } else {
                "--:--".to_string()
            };
            let who = if m.get("fromMe").and_then(|v| v.as_bool()).unwrap_or(false) {
                "me".to_string()
            } else {
                // Prefer the resolved display name; fall back to raw JID
                // (the "from" field), then "?".
                m.get("fromName")
                    .and_then(|v| v.as_str())
                    .filter(|s| !s.is_empty())
                    .or_else(|| m.get("from").and_then(|v| v.as_str()))
                    .filter(|s| !s.is_empty())
                    .unwrap_or("?")
                    .to_string()
            };
            let body = m
                .get("body")
                .and_then(|v| v.as_str())
                .unwrap_or("")
                .replace(['\r', '\n'], " ");
            let type_ = m
                .get("type")
                .and_then(|v| v.as_str())
                .filter(|t| *t != "chat" && !t.is_empty())
                .map(|t| format!(" [{t}]"))
                .unwrap_or_default();
            format!("[{hhmm}] {who}{type_}: {body}")
        })
        .collect::<Vec<_>>()
        .join("\n");

    let header = format!(
        "# WhatsApp — {chat_name} — {day}\nchat_id: {chat_id}\naccount_id: {account_id}\nmessages: {n}\n\n",
        n = sorted.len()
    );
    let content = format!("{header}{transcript}");

    let namespace = format!("whatsapp-web:{account_id}");
    let key = format!("{chat_id}:{day}");
    let title = format!("WhatsApp · {chat_name} · {day}");

    Some(json!({
        "namespace": namespace,
        "key": key,
        "title": title,
        "content": content,
        "source_type": "whatsapp-web",
        "priority": "medium",
        "tags": ["whatsapp", "chat-transcript", day],
        "metadata": {
            "provider": "whatsapp",
            "account_id": account_id,
            "chat_id": chat_id,
            "chat_name": chat_name,
            "day": day,
            "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,
        "method": "openhuman.memory_doc_ingest",
        "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(());
    }

    // 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
    let chats_param: serde_json::Map<String, Value> = chats
        .as_object()
        .map(|o| {
            o.iter()
                .map(|(jid, v)| {
                    let name = if let Some(s) = v.as_str() {
                        if s.is_empty() {
                            Value::Null
                        } else {
                            Value::String(s.to_string())
                        }
                    } else {
                        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();

    // 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();

    // 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 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(),
        total_batches,
    );
    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<Value>, 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<String, (String, Value)> = 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<String> = 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<String> = 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.
#[derive(Default)]
pub struct ScannerRegistry {
    started: Mutex<std::collections::HashMap<String, Vec<AbortHandle>>>,
}

impl ScannerRegistry {
    pub fn new() -> Arc<Self> {
        Arc::new(Self::default())
    }

    pub fn ensure_scanner<R: Runtime>(
        &self,
        app: AppHandle<R>,
        account_id: String,
        url_prefix: String,
    ) {
        let mut g = self.started.lock();
        if g.contains_key(&account_id) {
            log::debug!("[wa] scanner already running for {}", account_id);
            return;
        }
        let handles = spawn_scanner(app, account_id.clone(), url_prefix);
        g.insert(account_id, handles);
    }

    pub fn forget(&self, account_id: &str) {
        let handles = self.started.lock().remove(account_id);
        if let Some(handles) = handles {
            let count = handles.len();
            for handle in handles {
                handle.abort();
            }
            log::info!("[wa] aborted {} scanner task(s) for {}", count, account_id);
        }
    }

    pub fn forget_all(&self) -> usize {
        let entries: Vec<_> = self.started.lock().drain().collect();
        let task_count = entries.iter().map(|(_, handles)| handles.len()).sum();
        for (account_id, handles) in entries {
            for handle in handles {
                handle.abort();
            }
            log::debug!("[wa] aborted scanner tasks for {}", account_id);
        }
        if task_count > 0 {
            log::info!("[wa] aborted {} scanner task(s)", task_count);
        }
        task_count
    }
}

#[cfg(test)]
#[path = "mod_tests.rs"]
mod tests;
