// ws.js — thin storage adapter over generic KV backend (HTTP + WebSocket). // Passthrough: no shape translation. Backend = firebase mirror. // Implements same interface as memory.js. Tested by storage contract vs running server. // Native browser WebSocket if present, else ws pkg (Node/jest). // Lazy load ws pkg so CRA prod build (ESM) doesn't choke on require(). let WebSocketImpl; if (typeof WebSocket !== 'undefined') { WebSocketImpl = WebSocket; } function createServerStorage({ baseUrl, realtimeUrl } = {}) { // Same-origin by default: empty baseUrl = relative fetch (Caddy/proxy). // Fallback to localhost for bare `npm start` dev without proxy. const API = (baseUrl || (typeof window !== 'undefined' && window.location ? '' : 'http://127.0.0.1:4001')).replace(/\/$/, ''); let WS; if (realtimeUrl) { WS = realtimeUrl; } else if (typeof window !== 'undefined' && window.location) { // derive from current origin (http→ws, https→wss), same host/port. const proto = window.location.protocol === 'https:' ? 'wss:' : 'ws:'; WS = `${proto}//${window.location.host}/ws`; } else { WS = 'ws://127.0.0.1:4001/ws'; } // App passes firebase-prefixed paths: artifacts/{APP_ID}/public/data/campaigns/... // Backend uses canonical paths. Strip prefix. function norm(p) { if (!p) return p; return p.replace(/^[\s\S]*\/public\/data\//, ''); } // Inject id (last path segment) into doc — matches main firebase truth: // { id: docSnap.id, ...snap.data() } // Backend stores docs WITHOUT id; client adds it so all adapters share shape. function withId(rawPath, data) { if (data === null || data === undefined) return data; const id = norm(rawPath).split('/').pop(); return { id, ...data }; } const docSubs = new Map(); // path -> Set const collSubs = new Map(); // collPath -> Set const collConstraints = new Map(); // collPath -> constraints[] (per collection) // Last data emitted per path. Dedupe identical re-emits so WS-open // re-fetch (authoritative catch-up) doesn't double-fire when REST already // delivered the same data. Race: REST empty → WS-open populated fires. const lastEmitted = new Map(); // path -> serialized data (or null for empty) let ws = null; let wsReady = null; let disposed = false; let reconnectTimer = null; let everConnected = false; const RECONNECT_DELAY = 500; // Emit helper: dedupe identical re-emits per path. WS-open re-fetch // (authoritative catch-up) + initial REST both may fire; skip dup. function emit(kind, path, data, cbs) { const key = `${kind}:${path}`; const ser = data === undefined ? 'undef' : JSON.stringify(data); if (lastEmitted.get(key) === ser) return; // identical, skip lastEmitted.set(key, ser); cbs.forEach(cb => cb(data)); } function ensureWs() { if (wsReady) return wsReady; wsReady = new Promise((resolve, reject) => { (async () => { // Node/jest only: load ws pkg via dynamic import. Browser uses global // WebSocket. Avoids require() in CRA prod ESM bundle (webpack crash). let WsClass = WebSocketImpl; if (!WsClass) { const wsPkg = await import('ws'); WsClass = wsPkg.WebSocket; } ws = new WsClass(WS); const onOpen = () => { const isReconnect = everConnected; everConnected = true; // resubscribe all existing subscribers after (re)connect for (const p of docSubs.keys()) { ws.send(JSON.stringify({ type: 'subscribe', kind: 'doc', path: p })); } for (const p of collSubs.keys()) { ws.send(JSON.stringify({ type: 'subscribe', kind: 'collection', path: p })); } // Reconnect only: re-fetch current values — catches writes that // happened while disconnected (broadcast missed). Skip on first connect // (initial REST fetch in subscribeDoc/subscribeCollection already did). if (isReconnect) { for (const [p, cbs] of docSubs) { storage.getDoc(p).then(doc => { cbs.forEach(cb => cb(doc)); }).catch(() => {}); } for (const [p, cbs] of collSubs) { const cstr = collConstraints.get(p) || []; storage.getCollection(p, cstr).then(docs => { cbs.forEach(cb => cb(docs)); }).catch(() => {}); } } resolve(ws); }; const onError = (err) => { wsReady = null; reject(err instanceof Event ? new Error('ws error') : err); }; const onClose = () => { wsReady = null; ws = null; if (disposed) return; // auto-reconnect (BUG-8): try again after delay. ensureWs() re-arms. if (reconnectTimer) clearTimeout(reconnectTimer); reconnectTimer = setTimeout(() => { reconnectTimer = null; if (!disposed) ensureWs().catch(() => {}); }, RECONNECT_DELAY); if (reconnectTimer && typeof reconnectTimer.unref === 'function') reconnectTimer.unref(); }; const onMessage = (ev) => { const raw = typeof ev === 'string' ? ev : (ev.data !== undefined ? ev.data : ev); let msg; try { msg = JSON.parse(typeof raw === 'string' ? raw : raw.toString()); } catch { return; } handleMessage(msg); }; // Bind ONCE. Browser WS + ws pkg are EventTargets: assigning onX AND // addEventListener('x') registers the handler twice -> every event // fires 2x (double onOpen -> spurious reconnect re-fetch, double // onMessage -> emit dedupe races under write load -> UI looks stale). ws.onopen = onOpen; ws.onerror = onError; ws.onclose = onClose; ws.onmessage = onMessage; })(); }); return wsReady; } // Backend pushes change notices keyed by path. Re-fetch affected subscribers. async function handleMessage(msg) { if (msg.type !== 'change' || !msg.change) { return; } const c = msg.change; // doc subscriber at exact changed path const docCbs = docSubs.get(c.path); if (docCbs) { const doc = await storage.getDoc(c.path); docCbs.forEach(cb => cb(doc)); } // collection subscribers at parent path (apply their stored constraints) if (c.parent) { const collCbs = collSubs.get(c.parent); if (collCbs) { const constraints = collConstraints.get(c.parent) || []; // Server honors orderBy + limit in SQL now. const docs = await storage.getCollection(c.parent, constraints); collCbs.forEach(cb => cb(docs)); } } } async function api(method, path, query, body) { let url = `${API}${path}`; if (query) { const qs = new URLSearchParams(query).toString(); url += `?${qs}`; } const res = await fetch(url, { method, headers: body ? { 'Content-Type': 'application/json' } : undefined, body: body ? JSON.stringify(body) : undefined, }); if (!res.ok) { const t = await res.text().catch(() => ''); throw new Error(`API ${method} ${path} ${res.status}: ${t}`); } const text = await res.text(); return text ? JSON.parse(text) : null; } const storage = { async getDoc(rawPath) { const p = norm(rawPath); const res = await api('GET', '/api/doc', { path: p }); const data = res && res.data !== undefined ? res.data : null; return withId(rawPath, data); }, async setDoc(rawPath, data, opts = {}) { const p = norm(rawPath); // merge flag -> PATCH (shallow merge, create-on-miss). Matches firebase // setDoc(path, data, {merge:true}) semantics. if (opts.merge) { await api('PATCH', '/api/doc', null, { path: p, patch: data }); } else { await api('PUT', '/api/doc', null, { path: p, data }); } }, async updateDoc(rawPath, patch) { const p = norm(rawPath); await api('PATCH', '/api/doc', null, { path: p, patch }); }, async deleteDoc(rawPath) { const p = norm(rawPath); await api('DELETE', '/api/doc', { path: p }); }, // Bulk delete collection (optionally where-filtered). No fetch. SQL DELETE. // DEV ONLY. Method ships in prod build but throws if not dev. async deleteCollection(rawCollPath, whereField, whereValue) { if (process.env.NODE_ENV !== 'development' && process.env.NODE_ENV !== 'test') { throw new Error('deleteCollection: dev only'); } const p = norm(rawCollPath); const query = { path: p }; if (whereField) { query.whereField = whereField; query.whereValue = whereValue; } const res = await api('DELETE', '/api/collection', query); return res.deleted; }, async addDoc(rawCollPath, data) { const p = norm(rawCollPath); const res = await api('POST', '/api/collection', null, { path: p, data }); return { id: res.id, path: res.path }; }, async getCollection(rawCollPath, queryConstraints = []) { const p = norm(rawCollPath); const query = { path: p }; // Push where + orderBy + limit + offset to server SQL when present. // Avoids client-side full-scan + re-sort on every WS change notification. for (const c of queryConstraints || []) { if (!c || !c.__type) continue; if (c.__type === 'where') { query.whereField = c.field; query.whereOp = c.op; query.whereValue = c.value; } else if (c.__type === 'orderBy') { query.orderBy = c.field; query.dir = c.dir || 'asc'; } else if (c.__type === 'limit') { query.limit = c.n; } else if (c.__type === 'offset') { query.offset = c.offset; } } const docs = await api('GET', '/api/collection', query); if (!Array.isArray(docs)) return []; return docs.map(d => { if (d && typeof d === 'object' && 'id' in d && 'data' in d) { return { id: d.id, ...d.data }; } return { ...d }; }); }, async countCollection(rawCollPath) { const p = norm(rawCollPath); const res = await api('GET', '/api/collection/count', { path: p }); return (res && typeof res.count === 'number') ? res.count : 0; }, async batchWrite(ops) { const normOps = ops.map(op => ({ ...op, path: norm(op.path) })); await api('POST', '/api/batch', null, { ops: normOps }); }, // Transactional undo (server mode only). Firebase falls back to 2-write // in App.js (via store.txUndo check). undo = { encounterPath, updates, redo }. // redo=true applies redo patch + flips undone:false. async undo({ logPath, undo, redo = false }) { const normUndo = { ...undo, encounterPath: norm(undo.encounterPath) }; const normLog = norm(logPath); await api('POST', '/api/undo', null, { logPath: normLog, undo: normUndo, redo }); }, subscribeDoc(rawPath, cb, errCb) { const p = norm(rawPath); lastEmitted.delete('doc:'+p); // fresh sub → allow first emit // Initial value via REST (independent of WS connect). getDoc injects id. storage.getDoc(p).then(doc => emit('doc', p, doc, docSubs.get(p) || new Set([cb]))).catch(() => {}); // WS only for subsequent change notifications. ensureWs().then(() => { ws.send(JSON.stringify({ type: 'subscribe', kind: 'doc', path: p })); }).catch(() => {}); if (!docSubs.has(p)) docSubs.set(p, new Set()); docSubs.get(p).add(cb); return () => { docSubs.get(p)?.delete(cb); }; }, subscribeCollection(rawCollPath, cb, queryConstraints = [], errCb) { const p = norm(rawCollPath); collConstraints.set(p, queryConstraints || []); lastEmitted.delete('coll:'+p); // fresh sub → allow first emit // Server honors orderBy + limit in SQL now. No client re-sort. storage.getCollection(p, queryConstraints || []) .then(docs => emit('coll', p, docs, collSubs.get(p) || new Set([cb]))) .catch(err => { if (errCb) errCb(err); }); // WS only for subsequent change notifications. ensureWs().then(() => { ws.send(JSON.stringify({ type: 'subscribe', kind: 'collection', path: p })); }).catch(() => {}); if (!collSubs.has(p)) collSubs.set(p, new Set()); collSubs.get(p).add(cb); return () => { collSubs.get(p)?.delete(cb); }; }, dispose(cb) { disposed = true; if (reconnectTimer) { clearTimeout(reconnectTimer); reconnectTimer = null; } if (ws) ws.close(); docSubs.clear(); collSubs.clear(); collConstraints.clear(); if (typeof cb === 'function') cb(); }, _api: api, _test: { getWs: () => ws, forceDrop: () => { if (ws) ws.close(); }, getReady: () => wsReady, docSubs, collSubs, }, }; return storage; } export { createServerStorage };