/** * @file sseAutomat.js * @description Central Server-Sent Events hub managing the single persistent * `GET /api/events` connection and fanning mutation signals into domain automats. * * Upstream Triggers: * - `authAutomat`: Subscribes to auth state changes — reconnects stream when * `selectedCompanyId` changes/becomes available; disconnects on `'anon'` (logout). * * Downstream Triggers: * Dispatches real-time SSE mutation events into specific automats: * - `artikelAutomat`: Re-fetches article via `reloadArticle(sku)` on `article-` signals. * - `articleDetailAutomat` (`articleHistoryAutomat`, `articleSupplierAutomat`): * Reloads SKU-keyed views on `article-` signals. * - `categoryAutomat`: Triggers `actions.loadTree()` and `invalidateAllCategoryViews()` * on `'category'` SSE events. * - `articleIndexAutomat`: Triggers `actions.reload()` on `'articles'` SSE events; * the index is derived from the reloaded category views, so no extra request. * - `warehouseAutomat`: Triggers `actions.reload()` on `'warehouse'` SSE events. * - `supplierAutomat`: Triggers `actions.reload()` on `'supplier'` SSE events. * - `attributeAutomat`: Triggers `actions.load()` on `'attributename'` SSE events. * - `propertyAutomat`: Triggers `actions.load()` on `'propertyname'` SSE events. */ import { Automat } from 'react-automat'; import authAutomat from './authAutomat'; import artikelAutomat from './artikelAutomat'; import { articleHistoryAutomat, articleSupplierAutomat } from './articleDetailAutomat'; import { articleIndexAutomat } from './articleIndexAutomat'; import categoryAutomat, { invalidateAllCategoryViews } from './categoryAutomat'; import warehouseAutomat from './warehouseAutomat'; import attributeAutomat from './attributeAutomat'; import propertyAutomat from './propertyAutomat'; import supplierAutomat from './supplierAutomat'; import orderAutomat from './orderAutomat'; import quoteAutomat from './quoteAutomat'; import shipmentAutomat from './shipmentAutomat'; import { handleDraftEvent } from './draftSync'; /** * @typedef {Object} SseState * @property {'disconnected'|'connecting'|'connected'|'error'} status Connection state. * @property {string|null} companyId Company the stream is bound to. * @property {string|null} clientId Tab-scoped random session id sent as `?cid=` to `/api/events`. * @property {Array<{id: number, type: string, data: string, lastEventId: string, time: number}>} events Recent stream events. */ const EVENT_HISTORY_LIMIT = 200; const SSE_CLIENT_ID_KEY = 'sse.cid'; function createClientId() { try { if (typeof crypto !== 'undefined' && typeof crypto.randomUUID === 'function') { return crypto.randomUUID(); } } catch {} return `${Date.now().toString(36)}-${Math.random().toString(36).slice(2, 10)}${Math.random().toString(36).slice(2, 10)}`; } /** * Returns the tab-scoped SSE client id, creating and persisting it on first use. * Created when the automat module is evaluated so every `EventSource` in this * tab shares the same `cid`. * * @returns {string} Client id. */ export function getSseClientId() { try { const existing = window.sessionStorage.getItem(SSE_CLIENT_ID_KEY); if (existing) return existing; const id = createClientId(); window.sessionStorage.setItem(SSE_CLIENT_ID_KEY, id); return id; } catch { if (!getSseClientId.fallback) getSseClientId.fallback = createClientId(); return getSseClientId.fallback; } } // Created once per tab at automat creation time and stored to sessionStorage. const sseClientId = typeof window !== 'undefined' ? getSseClientId() : null; /** @type {SseState} */ const initialState = { status: 'disconnected', companyId: null, clientId: sseClientId, events: [], }; /** Event name prefix for per-article signals. */ const ARTICLE_EVENT_PREFIX = 'article-'; let sseAutomat; /** @type {EventSource | null} */ let eventSource = null; /** * Handlers for statically named events (`category`, `articles`, …). * @type {Map void, detach: (() => void)|null}>>} */ const handlers = new Map(); let nextEventId = 0; const seenEvents = new WeakSet(); /** * Handlers per watched SKU, so one `article-` listener is attached while * at least one view is open on that SKU. * @type {Map void>>} */ const articleWatchers = new Map(); function recordEvent(event) { if (seenEvents.has(event)) return; seenEvents.add(event); const entry = { id: ++nextEventId, type: event.type, data: typeof event.data === 'string' ? event.data : '', lastEventId: event.lastEventId ?? '', time: Date.now(), }; sseAutomat.setState((state) => ({ events: [...state.events.slice(-(EVENT_HISTORY_LIMIT - 1)), entry], })); } function addTrackedListener(es, type, handler) { const listener = (event) => { recordEvent(event); handler(event); }; es.addEventListener(type, listener); return () => es.removeEventListener(type, listener); } function disconnect() { if (eventSource) { console.log('[sse] disconnect'); try { eventSource.close(); } catch {} eventSource = null; } if (sseAutomat.state.status !== 'disconnected') { sseAutomat.setState({ status: 'disconnected' }); } } /** * Attaches every registered handler to a freshly opened stream. * * @param {EventSource} es Open event stream. * @returns {void} */ function attachHandlers(es) { for (const [event, set] of handlers) { for (const entry of set) { entry.detach = addTrackedListener(es, event, entry.handler); } } for (const [sku] of articleWatchers) { attachArticleListener(sku); } } /** * Binds the single `article-` listener for a watched SKU. * * The listener resolves the handler set at call time, so a SKU that lost all * watchers turns into a cheap no-op instead of needing removal. Listeners are * reclaimed whenever the stream is rebuilt on a company switch. * * @param {string} sku Article SKU. * @returns {void} */ function attachArticleListener(sku) { if (!eventSource) return; addTrackedListener(eventSource, `${ARTICLE_EVENT_PREFIX}${sku}`, (event) => { const set = articleWatchers.get(sku); if (!set || set.size === 0) return; const changed = String(event.type).slice(ARTICLE_EVENT_PREFIX.length); for (const handler of set) handler(changed); }); } function connect() { const companyId = authAutomat.state.user?.selectedCompanyId; if (companyId == null) { disconnect(); return; } const url = new URL('/api/events', window.location.origin); if (sseClientId) url.searchParams.set('cid', sseClientId); console.log(`[sse] connecting`); sseAutomat.setState({ status: 'connecting', companyId: String(companyId) }); const es = new EventSource(url.href, { withCredentials: true }); // Category tree changed: refresh the tree and every open category view. addTrackedListener(es, 'category', () => { void categoryAutomat.actions.loadTree().catch((err) => { console.error('[sse] category tree refresh failed', err); }); invalidateAllCategoryViews(); }); // Article index changed: refresh the index and open category views. addTrackedListener(es, 'articles', () => { invalidateAllCategoryViews(); void articleIndexAutomat.actions.reload().catch((err) => { console.error('[sse] article index refresh failed', err); }); }); addTrackedListener(es, 'warehouse', () => { void warehouseAutomat.actions.reload().catch((err) => { console.error('[sse] warehouse refresh failed', err); }); }); addTrackedListener(es, 'attributename', () => { void attributeAutomat.actions.load().catch((err) => { console.error('[sse] attributename refresh failed', err); }); }); addTrackedListener(es, 'propertyname', () => { void propertyAutomat.actions.load().catch((err) => { console.error('[sse] propertyname refresh failed', err); }); }); addTrackedListener(es, 'supplier', () => { void supplierAutomat.actions.reload().catch((err) => { console.error('[sse] supplier refresh failed', err); }); }); addTrackedListener(es, 'order', () => { void orderAutomat.actions.reload().catch((err) => { console.error('[sse] order refresh failed', err); }); void shipmentAutomat.actions.reload().catch((err) => { console.error('[sse] shipment refresh failed', err); }); }); addTrackedListener(es, 'quote', () => { void quoteAutomat.actions.reload().catch((err) => { console.error('[sse] quote refresh failed', err); }); }); addTrackedListener(es, 'shipment', () => { void shipmentAutomat.actions.reload().catch((err) => { console.error('[sse] shipment refresh failed', err); }); void orderAutomat.actions.reload().catch((err) => { console.error('[sse] order refresh failed', err); }); }); addTrackedListener(es, 'draft', (event) => { try { const payload = typeof event.data === 'string' ? JSON.parse(event.data) : event.data; handleDraftEvent(payload); } catch (err) { console.error('[sse] draft event handling failed', err); } }); addTrackedListener(es, 'open', () => { sseAutomat.setState({ status: 'connected' }); }); addTrackedListener(es, 'message', () => {}); // EventSource reconnects on its own; only surface the state. es.onerror = () => { console.error('[sse] event stream error'); recordEvent({ type: 'error', data: '', lastEventId: '' }); sseAutomat.setState({ status: 'error' }); }; eventSource = es; attachHandlers(es); } const actions = { /** * Closes and reopens the event stream. Required after a company switch, * because the stream is scoped to the selected company at connect time. * * @returns {void} */ reconnect() { disconnect(); connect(); }, clearEvents() { sseAutomat.setState({ events: [] }); }, /** * Registers a handler for a statically named backend event. * * Registration order does not matter: handlers added while disconnected are * attached on the next connect. * * @param {string} event Event name (e.g. `category`, `articles`). * @param {(event: MessageEvent) => void} handler Signal handler. * @returns {() => void} Unsubscribe callback. */ on(event, handler) { let set = handlers.get(event); if (!set) { set = new Set(); handlers.set(event, set); } let entry = Array.from(set).find((candidate) => candidate.handler === handler); if (!entry) { entry = { handler, detach: null }; set.add(entry); if (eventSource) entry.detach = addTrackedListener(eventSource, event, handler); } return () => { entry.detach?.(); set.delete(entry); if (set.size === 0) handlers.delete(event); }; }, /** * Subscribes to `article-` signals for one SKU. * * The backend emits this event whenever that article changes. Exactly one * listener is attached per SKU while at least one caller is subscribed. * * @param {string} sku Article SKU to watch. * @param {(sku: string) => void} handler Called with the changed SKU. * @returns {() => void} Unsubscribe callback. */ watchArticle(sku, handler) { const key = String(sku); let set = articleWatchers.get(key); if (!set) { set = new Set(); articleWatchers.set(key, set); attachArticleListener(key); } set.add(handler); return () => { const current = articleWatchers.get(key); if (!current) return; current.delete(handler); if (current.size === 0) articleWatchers.delete(key); }; }, }; /** * Global `GET /api/events` connection. * * Owns the single `EventSource` for the tab and dispatches signals into the * domain automats. Reconnects whenever the selected company changes. * * @type {Automat} */ sseAutomat = new Automat(initialState, actions, { name: 'sse' }); /** * Refreshes the cached article plus any open history/supplier view. * * Background refreshes stay silent: an article that is not on screen has no * visible subscriber to notify, so failures are swallowed. * * @param {string} sku Article SKU. * @returns {void} */ export function invalidateArticle(sku) { const key = String(sku); void artikelAutomat.actions.reloadArticle(key).catch(() => {}); if (articleHistoryAutomat.has(key)) { void articleHistoryAutomat.reload(key).catch(() => {}); } if (articleSupplierAutomat.has(key)) { void articleSupplierAutomat.reload(key).catch(() => {}); } } // Reconnect when the selected company changes; disconnect on logout. // On logout the stream must close; on relogin (anon -> authed with a company) // the stream is re-established even if the company id matches the previous one. let wasAuthed = false; authAutomat.subscribe((auth) => { const companyId = auth.status === 'authed' && auth.user?.selectedCompanyId != null ? String(auth.user.selectedCompanyId) : null; const isAuthed = companyId != null; const relogin = isAuthed && !wasAuthed; wasAuthed = isAuthed; if (!relogin && companyId === sseAutomat.state.companyId) return; if (companyId == null) { disconnect(); sseAutomat.setState({ companyId: null }); return; } // A different company (or a fresh login) means a different stream scope. disconnect(); sseAutomat.setState({ companyId }); connect(); }); export default sseAutomat;