whatsapp: persist chats and messages across restarts (v8 / 2.2.0)

WhatsApp delivers a history sync at login, not on reconnect, so an in-memory
store left the connector blind after every process restart (23 in two weeks).
The store is now mirrored to store/, next to auth/ and bind-mounted the same
way.

Not SQLite: node:sqlite needs Node 22 (unflagged only from 24) and the runtime
image ships Debian trixie's nodejs = 20.19.2; better-sqlite3 is native and the
slim image has no toolchain. So an append-only JSONL log for messages plus a
debounced JSON snapshot for chats/contacts, both written tmp+rename so a crash
cannot truncate them. Compaction on load and every 500 appends keeps the log
from creeping upward.

Also fixes a pre-existing duplication bug: pushMessage appended unconditionally,
re-adding every message a history re-sync redelivered. It now returns false on a
known id and the message is skipped in both the transcript and the log.

MAX_MSGS_PER_CHAT 200 -> 500. logout deletes store/ with auth/, so re-linking a
different phone cannot inherit the previous account's history.

Data at rest: message text is now written to the user's bind-mounted home.
Nothing but session keys was persisted before.

Verified on skald-runtime:v4 with a seeded store: 751 lines with 50 duplicates,
a 700-message chat and a torn trailing line -> 701 loaded, compacted to 501, cap
applied, second run 501 -> 501 unchanged.

connectors/whatsapp/index.js

@@ -125,22 +125,126 @@ const RECONNECT_BASE_MS = 1500;
 const RECONNECT_MAX_MS = 60_000;
 let backoffMs = RECONNECT_BASE_MS;
 
-// ── Lightweight in-memory store ───────────────────────────────────────────────
+// ── Store ─────────────────────────────────────────────────────────────────────
 // Baileys keeps no chat/contact store of its own; we build a minimal one from the
-// history-sync event and live upserts. It lives for the process lifetime — enough
-// for "what's going on now", not a full archive.
+// history-sync event and live upserts. It is mirrored to `store/` (bind-mounted next
+// to `auth/`, so it survives a container recreate) because WhatsApp only delivers a
+// history sync at *login* — without a mirror every process restart left the connector
+// blind until new messages happened to arrive.
+//
+// Deliberately not SQLite: `node:sqlite` needs Node 22 (the runtime image is 20.19.2)
+// and `better-sqlite3` is a native module the slim image cannot build. So: an
+// append-only JSONL log for messages (O(1) per message) plus a small debounced
+// snapshot for chats/contacts.
 const chats    = new Map();   // jid -> { id, name, unread, conversationTimestamp }
 const contacts = new Map();   // jid -> { id, name }
 const messages = new Map();   // jid -> [ { id, fromMe, ts, text, author } ]  (capped)
 
-const MAX_MSGS_PER_CHAT = 200;
+const MAX_MSGS_PER_CHAT = 500;
 
+const STORE_DIR  = path.join(__dirname, 'store');
+const MSG_LOG    = path.join(STORE_DIR, 'messages.jsonl');
+const META_FILE  = path.join(STORE_DIR, 'meta.json');
+const COMPACT_EVERY = 500;    // appends between rewrites of the log
+const META_DEBOUNCE_MS = 5000;
+
+let appendsSinceCompact = 0;
+let metaTimer = null;
+
+// Returns false when the message was already known: history sync re-delivers the same
+// ids on every login, and without this both the log and the transcript would grow a
+// duplicate per restart.
 function pushMessage(jid, m) {
-  if (!jid) return;
+  if (!jid) return false;
   let arr = messages.get(jid);
   if (!arr) { arr = []; messages.set(jid, arr); }
+  if (m.id && arr.some((x) => x.id === m.id)) return false;
   arr.push(m);
   if (arr.length > MAX_MSGS_PER_CHAT) arr.splice(0, arr.length - MAX_MSGS_PER_CHAT);
+  return true;
+}
+
+function writeFileAtomic(file, data) {
+  const tmp = `${file}.tmp`;
+  fs.writeFileSync(tmp, data);
+  fs.renameSync(tmp, file);   // rename is atomic: a crash mid-write cannot truncate the store
+}
+
+function loadStore() {
+  try {
+    const meta = JSON.parse(fs.readFileSync(META_FILE, 'utf8'));
+    for (const c of meta.chats || []) if (c?.id) chats.set(c.id, c);
+    for (const c of meta.contacts || []) if (c?.id) contacts.set(c.id, c);
+  } catch { /* first run, or a snapshot we cannot parse — start empty */ }
+
+  let restored = 0;
+  try {
+    for (const line of fs.readFileSync(MSG_LOG, 'utf8').split('\n')) {
+      if (!line) continue;
+      try {
+        const { jid, m } = JSON.parse(line);
+        if (pushMessage(jid, m)) restored++;
+      } catch { /* skip a torn trailing line */ }
+    }
+  } catch { /* no log yet */ }
+
+  if (restored || chats.size || contacts.size) {
+    log(`store loaded: ${chats.size} chats, ${contacts.size} contacts, ${restored} messages`);
+  }
+  // Rewrite immediately: replays the per-chat cap and drops duplicates, so the log
+  // cannot creep upward across restarts.
+  if (restored) compactMessageLog();
+}
+
+function compactMessageLog() {
+  try {
+    fs.mkdirSync(STORE_DIR, { recursive: true });
+    const out = [];
+    for (const [jid, arr] of messages) for (const m of arr) out.push(JSON.stringify({ jid, m }));
+    writeFileAtomic(MSG_LOG, out.length ? out.join('\n') + '\n' : '');
+    appendsSinceCompact = 0;
+  } catch (e) {
+    log(`store compaction failed: ${e.message}`);
+  }
+}
+
+function appendMessage(jid, m) {
+  try {
+    fs.mkdirSync(STORE_DIR, { recursive: true });
+    fs.appendFileSync(MSG_LOG, JSON.stringify({ jid, m }) + '\n');
+    if (++appendsSinceCompact >= COMPACT_EVERY) compactMessageLog();
+  } catch (e) {
+    log(`store append failed: ${e.message}`);
+  }
+}
+
+function saveMeta() {
+  try {
+    fs.mkdirSync(STORE_DIR, { recursive: true });
+    writeFileAtomic(META_FILE, JSON.stringify({
+      chats: [...chats.values()],
+      contacts: [...contacts.values()],
+    }));
+  } catch (e) {
+    log(`store meta save failed: ${e.message}`);
+  }
+}
+
+// chats/contacts churn in bursts during a history sync; one write per burst is enough.
+function scheduleMetaSave() {
+  if (metaTimer) return;
+  metaTimer = setTimeout(() => { metaTimer = null; saveMeta(); }, META_DEBOUNCE_MS);
+}
+
+function flushStore() {
+  if (metaTimer) { clearTimeout(metaTimer); metaTimer = null; }
+  saveMeta();
+}
+
+function clearStore() {
+  try { fs.rmSync(STORE_DIR, { recursive: true, force: true }); } catch { /* nothing to clear */ }
+  appendsSinceCompact = 0;
+  if (metaTimer) { clearTimeout(metaTimer); metaTimer = null; }
 }
 
 function contactName(jid) {
@@ -307,6 +411,7 @@ async function startSock() {
         contacts.set(c.id, { id: c.id, name: c.name || c.notify || c.verifiedName || null });
       }
       for (const m of hm || []) ingestMessage(m, false);
+      scheduleMetaSave();
       log(`history sync (type ${syncType ?? '?'}${progress != null ? `, ${progress}%` : ''}): ` +
           `+${(hc || []).length} chats, +${(hcs || []).length} contacts, +${(hm || []).length} messages ` +
           `→ ${chats.size} chats / ${contacts.size} contacts known`);
@@ -319,10 +424,12 @@ async function startSock() {
         unread: c.unreadCount || 0,
         conversationTimestamp: Number(c.conversationTimestamp) || 0,
       });
+      scheduleMetaSave();
     });
     s.ev.on('contacts.upsert', (cs) => {
       if (myGen !== generation) return;
       for (const c of cs) contacts.set(c.id, { id: c.id, name: c.name || c.notify || c.verifiedName || null });
+      scheduleMetaSave();
     });
     s.ev.on('contacts.update', (cs) => {
       if (myGen !== generation) return;
@@ -330,6 +437,7 @@ async function startSock() {
         const prev = contacts.get(c.id) || { id: c.id };
         contacts.set(c.id, { ...prev, name: c.name || c.notify || prev.name || null });
       }
+      scheduleMetaSave();
     });
 
     s.ev.on('messages.upsert', ({ messages: ms, type }) => {
@@ -363,19 +471,24 @@ function ingestMessage(m, live) {
     // and are pure noise in a transcript.
     if (!text) return;
 
-    pushMessage(jid, {
+    const stored = {
       id: m.key?.id,
       fromMe: !!m.key?.fromMe,
       ts: Number(m.messageTimestamp) || 0,
       text,
       author: m.key?.participant || (m.key?.fromMe ? meJid : jid),
-    });
+    };
+    if (!pushMessage(jid, stored)) return;   // already known — do not log it twice
+    appendMessage(jid, stored);
+
     if (live && !chats.has(jid)) {
       chats.set(jid, { id: jid, name: m.pushName || null, unread: 0, conversationTimestamp: Number(m.messageTimestamp) || 0 });
+      scheduleMetaSave();
     } else if (live) {
       const ch = chats.get(jid);
       ch.conversationTimestamp = Number(m.messageTimestamp) || ch.conversationTimestamp;
       if (m.pushName && !ch.name) ch.name = m.pushName;
+      scheduleMetaSave();
     }
   } catch { /* one malformed frame must not stop the stream */ }
 }
@@ -435,6 +548,7 @@ async function toolLogout() {
   if (reconnectTimer) { clearTimeout(reconnectTimer); reconnectTimer = null; }
   try { fs.rmSync(AUTH_DIR, { recursive: true, force: true }); } catch { /* nothing to clear */ }
   chats.clear(); contacts.clear(); messages.clear();
+  clearStore();
   curQr = null; state = 'connecting'; starting = false; meJid = null;
   backoffMs = RECONNECT_BASE_MS;
   scheduleReconnect(500);
@@ -634,6 +748,7 @@ async function handleRequest(msg) {
 async function main() {
   log('Starting WhatsApp MCP server (Baileys)');
   fs.mkdirSync(MEDIA_DIR, { recursive: true });
+  loadStore();
 
   if (await loadBaileys()) {
     startSock().catch((e) => log(`initial startSock failed: ${e.message}`));
@@ -658,6 +773,7 @@ function shutdown(code) {
   generation++;
   teardownSock(sock);
   sock = null;
+  flushStore();
   process.exit(code);
 }