| 1 | // The local message store (#457): an IndexedDB cache that makes reloads and |
| 2 | // thread switches paint instantly, and (later) feeds message search. |
| 3 | // |
| 4 | // What it holds — and deliberately nothing more: the RAW server responses. |
| 5 | // Message rows are ciphertext blobs; the boot snapshot is the byte-equivalent |
| 6 | // of /api/threads, /api/roster, /api/nicknames, /api/tiers and |
| 7 | // /api/instance-key (wrapped keys, encrypted name/preview blobs). Nothing is |
| 8 | // stored decrypted, so this cache's at-rest posture equals the server's own — |
| 9 | // strictly tighter than the per-thread keys the service worker already keeps |
| 10 | // in the sibling "lchat" DB (push.js). Reading this DB gains an attacker |
| 11 | // nothing they couldn't already decrypt with those. |
| 12 | // |
| 13 | // The page hydrates by running the cached responses through the SAME apply + |
| 14 | // fold pipeline the network path uses (unwrap keys → titles → fold rows), so |
| 15 | // there is no parallel render path; the real fetches then reconcile — folds |
| 16 | // dedupe by message id, edits refold via edited_at, deletes replay via |
| 17 | // tombstone rows (#581/#583). |
| 18 | // |
| 19 | // The service worker will also write pushed rows here (phase 2); it can't |
| 20 | // import this module (classic worker), so keep DB_NAME/DB_VERSION/store names |
| 21 | // in sync with any inline writer added to sw.js. |
| 22 | // |
| 23 | // Every exported IDB call is best-effort: a quota error, a private-mode |
| 24 | // browser, or a torn write must degrade to "no cache" — never break chat. |
| 25 | |
| 26 | const DB_NAME = "lchat-store"; // NOT "lchat": three openers pin that at v1 for the SW's kv |
| 27 | const DB_VERSION = 1; |
| 28 | const SNAP_KEY = "boot"; |
| 29 | export const SNAP_VERSION = 1; // bump to shed incompatible old snapshots |
| 30 | |
| 31 | // ---- pure helpers (unit-tested in Node, no IndexedDB) --------------------- |
| 32 | |
| 33 | // A snapshot is usable only when it carries the current format version and the |
| 34 | // two responses hydration cannot start without. Everything else is optional — |
| 35 | // a partial snapshot (e.g. tiers never fetched) still paints. |
| 36 | export function snapshotUsable(snap) { |
| 37 | return !!( |
| 38 | snap && |
| 39 | snap.v === SNAP_VERSION && |
| 40 | snap.threads && |
| 41 | Array.isArray(snap.threads.threads) && |
| 42 | snap.roster && |
| 43 | Array.isArray(snap.roster.roster) |
| 44 | ); |
| 45 | } |
| 46 | |
| 47 | // Split a fetched page into rows to keep and the ids its tombstones unsend. |
| 48 | // The tombstone row itself is stored too: it keeps the delete durable locally |
| 49 | // (a later hydrate re-folds it as a no-op) while the target's ciphertext is |
| 50 | // actually dropped — mirroring the server's own posture. |
| 51 | export function partitionRows(rows) { |
| 52 | const puts = []; |
| 53 | const unsend = []; |
| 54 | for (const m of rows || []) { |
| 55 | if (!m || !m.id) continue; |
| 56 | puts.push(m); |
| 57 | if (m.tombstone_of) unsend.push(m.tombstone_of); |
| 58 | } |
| 59 | return { puts, unsend }; |
| 60 | } |
| 61 | |
| 62 | // Which cached thread ids are no longer in the live list (left, evicted, or |
| 63 | // deleted while this device was away) — their rows must drop. Only trust a |
| 64 | // well-formed live list: an error body must never wipe the cache. |
| 65 | export function staleThreadIds(cachedIds, liveList) { |
| 66 | if (!liveList || !Array.isArray(liveList.threads)) return []; |
| 67 | const live = new Set(liveList.threads.map((t) => t.id)); |
| 68 | return (cachedIds || []).filter((id) => !live.has(id)); |
| 69 | } |
| 70 | |
| 71 | // ---- IndexedDB plumbing --------------------------------------------------- |
| 72 | |
| 73 | let dbPromise = null; |
| 74 | function openStore() { |
| 75 | if (dbPromise) return dbPromise; |
| 76 | dbPromise = new Promise((resolve, reject) => { |
| 77 | let req; |
| 78 | try { |
| 79 | req = indexedDB.open(DB_NAME, DB_VERSION); |
| 80 | } catch (e) { |
| 81 | reject(e); // e.g. storage disabled entirely |
| 82 | return; |
| 83 | } |
| 84 | req.onupgradeneeded = () => { |
| 85 | const db = req.result; |
| 86 | if (!db.objectStoreNames.contains("kv")) db.createObjectStore("kv"); |
| 87 | if (!db.objectStoreNames.contains("msgs")) { |
| 88 | // One row per server message row, newest read by id order within a |
| 89 | // thread. keyPath mirrors the server's identity: (thread, message id). |
| 90 | db.createObjectStore("msgs", { keyPath: ["t", "id"] }); |
| 91 | } |
| 92 | }; |
| 93 | req.onsuccess = () => { |
| 94 | // If another tab upgrades the schema later, close so it isn't blocked. |
| 95 | req.result.onversionchange = () => { |
| 96 | req.result.close(); |
| 97 | dbPromise = null; |
| 98 | }; |
| 99 | resolve(req.result); |
| 100 | }; |
| 101 | req.onerror = () => reject(req.error); |
| 102 | req.onblocked = () => reject(new Error("blocked")); |
| 103 | }); |
| 104 | dbPromise.catch(() => (dbPromise = null)); // let a later call retry |
| 105 | return dbPromise; |
| 106 | } |
| 107 | |
| 108 | // Run one transaction; resolves when it completes. `fn(t, done)` does the |
| 109 | // requests — a read calls `done(value)` from a request's onsuccess to set what |
| 110 | // the transaction resolves to; a write just issues its requests. Any failure |
| 111 | // resolves `fallback` (best-effort contract — see the header). |
| 112 | async function tx(storeNames, mode, fn, fallback = undefined) { |
| 113 | try { |
| 114 | const db = await openStore(); |
| 115 | return await new Promise((resolve) => { |
| 116 | const t = db.transaction(storeNames, mode); |
| 117 | let out = fallback; |
| 118 | const done = (v) => (out = v); |
| 119 | try { |
| 120 | fn(t, done); |
| 121 | } catch { |
| 122 | try { |
| 123 | t.abort(); |
| 124 | } catch {} |
| 125 | resolve(fallback); |
| 126 | return; |
| 127 | } |
| 128 | t.oncomplete = () => resolve(out); |
| 129 | t.onabort = () => resolve(fallback); |
| 130 | t.onerror = () => resolve(fallback); |
| 131 | }); |
| 132 | } catch { |
| 133 | return fallback; |
| 134 | } |
| 135 | } |
| 136 | |
| 137 | // ---- boot snapshot -------------------------------------------------------- |
| 138 | |
| 139 | export function getSnapshot() { |
| 140 | return tx(["kv"], "readonly", (t, done) => { |
| 141 | const req = t.objectStore("kv").get(SNAP_KEY); |
| 142 | req.onsuccess = () => done(req.result || null); |
| 143 | }, null); |
| 144 | } |
| 145 | |
| 146 | export function putSnapshot(snap) { |
| 147 | return tx(["kv"], "readwrite", (t) => { |
| 148 | t.objectStore("kv").put(snap, SNAP_KEY); |
| 149 | }); |
| 150 | } |
| 151 | |
| 152 | // ---- message rows --------------------------------------------------------- |
| 153 | |
| 154 | // The newest `limit` raw rows for a thread, ascending by id — the same shape |
| 155 | // and order /api/messages returns, so folds can't tell cache from network. |
| 156 | export function getRecentMessages(threadId, limit = 100) { |
| 157 | return tx(["msgs"], "readonly", (t, done) => { |
| 158 | const rows = []; |
| 159 | const range = IDBKeyRange.bound([threadId, 0], [threadId, Infinity]); |
| 160 | const cur = t.objectStore("msgs").openCursor(range, "prev"); |
| 161 | cur.onsuccess = () => { |
| 162 | const c = cur.result; |
| 163 | if (c && rows.length < limit) { |
| 164 | rows.push(c.value); |
| 165 | c.continue(); |
| 166 | } else { |
| 167 | done(rows.reverse()); |
| 168 | } |
| 169 | }; |
| 170 | }, []); |
| 171 | } |
| 172 | |
| 173 | // Store fetched rows; apply any tombstones in the batch (drop their targets' |
| 174 | // ciphertext). Rows are stored verbatim plus the keyPath's thread tag. |
| 175 | export function putMessages(threadId, rows) { |
| 176 | const { puts, unsend } = partitionRows(rows); |
| 177 | if (!puts.length) return Promise.resolve(); |
| 178 | return tx(["msgs"], "readwrite", (t) => { |
| 179 | const s = t.objectStore("msgs"); |
| 180 | for (const m of puts) s.put({ ...m, t: threadId }); |
| 181 | for (const id of unsend) s.delete([threadId, id]); |
| 182 | }); |
| 183 | } |
| 184 | |
| 185 | // A live edit frame rewrote the ciphertext in place — mirror it. |
| 186 | export function applyEditRow(threadId, id, blob, editedAt) { |
| 187 | return tx(["msgs"], "readwrite", (t) => { |
| 188 | const s = t.objectStore("msgs"); |
| 189 | const g = s.get([threadId, id]); |
| 190 | g.onsuccess = () => { |
| 191 | if (g.result) s.put({ ...g.result, blob, edited_at: editedAt || Math.floor(Date.now() / 1000) }); |
| 192 | }; |
| 193 | }); |
| 194 | } |
| 195 | |
| 196 | // A live delete frame — drop the ciphertext now; the tombstone row arrives |
| 197 | // with the next fetched page and is stored then. |
| 198 | export function deleteMessageRow(threadId, id) { |
| 199 | return tx(["msgs"], "readwrite", (t) => { |
| 200 | t.objectStore("msgs").delete([threadId, id]); |
| 201 | }); |
| 202 | } |
| 203 | |
| 204 | export function clearThread(threadId) { |
| 205 | return tx(["msgs", "kv"], "readwrite", (t) => { |
| 206 | t.objectStore("msgs").delete(IDBKeyRange.bound([threadId, 0], [threadId, Infinity])); |
| 207 | // Any per-thread bookkeeping (e.g. a future backfill watermark) goes too. |
| 208 | t.objectStore("kv").delete("thread:" + threadId); |
| 209 | }); |
| 210 | } |
| 211 | |
| 212 | // Thread ids present in the msgs store — the prune sweep's input. |
| 213 | export function cachedThreadIds() { |
| 214 | return tx(["msgs"], "readonly", (t, done) => { |
| 215 | const ids = []; |
| 216 | // Jump cursor: visit one key per thread, then skip past its rows. |
| 217 | const cur = t.objectStore("msgs").openKeyCursor(); |
| 218 | cur.onsuccess = () => { |
| 219 | const c = cur.result; |
| 220 | if (c) { |
| 221 | ids.push(c.key[0]); |
| 222 | c.continue([c.key[0], Infinity]); |
| 223 | } else { |
| 224 | done(ids); |
| 225 | } |
| 226 | }; |
| 227 | }, []); |
| 228 | } |
| 229 | |
| 230 | // ---- full-history backfill (#457) ----------------------------------------- |
| 231 | // Page a thread's history backwards into the store with the server's `before` |
| 232 | // cursor, persisting an oldest-fetched watermark per thread so an interrupted |
| 233 | // run RESUMES rather than restarts. A thread that reaches its beginning (an |
| 234 | // empty page) is marked complete and never re-walked — from then on the |
| 235 | // ordinary forward sync keeps coverage total. Rows are stored, never folded: |
| 236 | // this is the search substrate (#586), not a display path. |
| 237 | // |
| 238 | // Known limit, accepted for now: the watermark only ever descends from the |
| 239 | // oldest row present when the walk started. A gap ABOVE that point (a thread |
| 240 | // that outran the cached window between sessions) isn't detected — those |
| 241 | // messages exist server-side but not locally until a fresh walk. Fine for |
| 242 | // search-completeness at this product's scale; revisit with interval tracking |
| 243 | // if it ever bites. |
| 244 | |
| 245 | // Where a thread's backfill has reached. null = never started. |
| 246 | export function getBackfillMark(threadId) { |
| 247 | return tx(["kv"], "readonly", (t, done) => { |
| 248 | const req = t.objectStore("kv").get("thread:" + threadId); |
| 249 | req.onsuccess = () => done(req.result || null); |
| 250 | }, null); |
| 251 | } |
| 252 | |
| 253 | function setBackfillMark(threadId, mark) { |
| 254 | return tx(["kv"], "readwrite", (t) => { |
| 255 | t.objectStore("kv").put(mark, "thread:" + threadId); |
| 256 | }); |
| 257 | } |
| 258 | |
| 259 | // The lowest stored id for a thread — where a first walk starts from. |
| 260 | export function oldestStoredId(threadId) { |
| 261 | return tx(["msgs"], "readonly", (t, done) => { |
| 262 | const range = IDBKeyRange.bound([threadId, 0], [threadId, Infinity]); |
| 263 | const cur = t.objectStore("msgs").openKeyCursor(range, "next"); |
| 264 | cur.onsuccess = () => done(cur.result ? cur.result.key[1] : 0); |
| 265 | }, 0); |
| 266 | } |
| 267 | |
| 268 | // Pure: the `before` cursor a run should continue from. -1 = already complete; |
| 269 | // 0 = nothing local yet (seed a recent page first). |
| 270 | export function backfillCursor(mark, oldestStored) { |
| 271 | if (mark && mark.complete) return -1; |
| 272 | if (mark && mark.oldest) return mark.oldest; |
| 273 | return oldestStored || 0; |
| 274 | } |
| 275 | |
| 276 | // Drive a thread's backfill to completion (or cancellation). Transport is |
| 277 | // injected — fetchPage(before) resolves the raw rows of one page (before=0 ⇒ |
| 278 | // the most recent page) — so this stays free of tokens and testable end to |
| 279 | // end by importing the module. onPage({oldest, oldestAt, count, rows}) is |
| 280 | // AWAITED per page — it narrates progress ("searching back to 2024…") and |
| 281 | // receives the page's rows for incremental indexing; an AbortSignal cancels |
| 282 | // between pages. |
| 283 | export async function runBackfill(threadId, fetchPage, { onPage, signal } = {}) { |
| 284 | let cursor = backfillCursor(await getBackfillMark(threadId), await oldestStoredId(threadId)); |
| 285 | if (cursor === -1) return { complete: true }; |
| 286 | if (!cursor) { |
| 287 | const rows = await fetchPage(0); |
| 288 | if (!rows.length) { |
| 289 | // A thread with no history at all is trivially complete. |
| 290 | await setBackfillMark(threadId, { oldest: 0, complete: true }); |
| 291 | return { complete: true }; |
| 292 | } |
| 293 | await putMessages(threadId, rows); |
| 294 | cursor = rows[0].id; |
| 295 | await setBackfillMark(threadId, { oldest: cursor, complete: false }); |
| 296 | } |
| 297 | while (!(signal && signal.aborted)) { |
| 298 | const rows = await fetchPage(cursor); |
| 299 | if (!rows.length) { |
| 300 | await setBackfillMark(threadId, { oldest: cursor, complete: true }); |
| 301 | return { complete: true }; |
| 302 | } |
| 303 | const next = rows[0].id; // pages come ascending; the first row is the oldest |
| 304 | if (!(next < cursor)) { |
| 305 | // The walk only terminates because each page's oldest id is strictly |
| 306 | // older than the cursor. A page that fails to advance (duplicate or |
| 307 | // non-descending ids — a server bug) would replay forever; stop |
| 308 | // honestly instead. The mark is untouched, so a later run retries. |
| 309 | return { complete: false }; |
| 310 | } |
| 311 | await putMessages(threadId, rows); |
| 312 | cursor = next; |
| 313 | await setBackfillMark(threadId, { oldest: cursor, complete: false }); |
| 314 | // Hand the page's rows over for incremental indexing, and AWAIT: a slow |
| 315 | // consumer backpressures the walk rather than racing it. |
| 316 | if (onPage) await onPage({ oldest: cursor, oldestAt: rows[0].created_at || 0, count: rows.length, rows }); |
| 317 | } |
| 318 | return { complete: false }; |
| 319 | } |
| 320 | |
| 321 | // Every stored row for a thread, ascending — the search scan's input. Bounded |
| 322 | // by cap as a runaway guard; search streams per thread, so a giant thread |
| 323 | // still yields in slices. |
| 324 | export function allMessages(threadId, cap = 50000) { |
| 325 | return tx(["msgs"], "readonly", (t, done) => { |
| 326 | const rows = []; |
| 327 | const range = IDBKeyRange.bound([threadId, 0], [threadId, Infinity]); |
| 328 | const cur = t.objectStore("msgs").openCursor(range, "next"); |
| 329 | cur.onsuccess = () => { |
| 330 | const c = cur.result; |
| 331 | if (c && rows.length < cap) { |
| 332 | rows.push(c.value); |
| 333 | c.continue(); |
| 334 | } else { |
| 335 | done(rows); |
| 336 | } |
| 337 | }; |
| 338 | }, []); |
| 339 | } |
| 340 | |
| 341 | // Sign-out: everything goes, snapshot included — the next visitor to this |
| 342 | // browser profile must find nothing. |
| 343 | export function wipeStore() { |
| 344 | return tx(["msgs", "kv"], "readwrite", (t) => { |
| 345 | t.objectStore("msgs").clear(); |
| 346 | t.objectStore("kv").clear(); |
| 347 | }); |
| 348 | } |
| 349 | |
amadan