ttrpg-initiative-tracker/src/storage/ws.js

// 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.

'use strict';

// Native browser WebSocket if present, else ws pkg (Node/jest).
let WebSocketImpl;
if (typeof WebSocket !== 'undefined') {
  WebSocketImpl = WebSocket;
} else {
  WebSocketImpl = require('ws').WebSocket;
}

function createWsStorage({ baseUrl, wsUrl } = {}) {
  const API = (baseUrl || 'http://127.0.0.1:4001').replace(/\/$/, '');
  const WS = wsUrl || (API.replace(/^http/, 'ws') + '/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\//, '');
  }

  const docSubs = new Map();      // path -> Set<cb>
  const collSubs = new Map();     // collPath -> Set<cb>
  let ws = null;
  let wsReady = null;

  function ensureWs() {
    if (wsReady) return wsReady;
    wsReady = new Promise((resolve, reject) => {
      ws = new WebSocketImpl(WS);
      const onOpen = () => resolve(ws);
      const onError = (err) => { wsReady = null; reject(err instanceof Event ? new Error('ws error') : err); };
      const onClose = () => { wsReady = null; };
      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);
      };
      ws.onopen = onOpen;
      ws.onerror = onError;
      ws.onclose = onClose;
      ws.onmessage = onMessage;
      if (typeof ws.addEventListener === 'function') {
        ws.addEventListener('open', onOpen);
        ws.addEventListener('error', onError);
        ws.addEventListener('close', onClose);
        ws.addEventListener('message', 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 (doc belongs to this collection)
    if (c.parent) {
      const collCbs = collSubs.get(c.parent);
      if (collCbs) {
        const docs = await storage.getCollection(c.parent);
        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 });
      return res && res.data !== undefined ? res.data : null;
    },

    async setDoc(rawPath, data) {
      const p = norm(rawPath);
      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 });
    },

    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) {
      const p = norm(rawCollPath);
      return await api('GET', '/api/collection', { path: p });
    },

    async batchWrite(ops) {
      const normOps = ops.map(op => ({ ...op, path: norm(op.path) }));
      await api('POST', '/api/batch', null, { ops: normOps });
    },

    subscribeDoc(rawPath, cb) {
      const p = norm(rawPath);
      // Initial value via REST (independent of WS connect).
      storage.getDoc(p).then(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) {
      const p = norm(rawCollPath);
      // Initial value via REST (independent of WS connect).
      storage.getCollection(p).then(cb).catch(() => {});
      // 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() { if (ws) ws.close(); docSubs.clear(); collSubs.clear(); },

    _api: api,
    _test: {
      getWs: () => ws,
      forceDrop: () => { if (ws) ws.close(); },
      getReady: () => wsReady,
      docSubs, collSubs,
    },
  };

  return storage;
}

module.exports = { createWsStorage };