/** * @file sseAutomat.js * @description Central Server-Sent Events hub managing the single persistent * `GET /api/events` connection and fanning mutation signals into active automats. * * Upstream Triggers: * - `authAutomat`: Connects stream when `hasAuth` is true (and reconnects when * `selectedCompanyId` changes); disconnects when `hasAuth` is false (logout). * * Downstream Triggers: * Dispatches real-time SSE mutation events into active automats without eagerly * importing or instantiating unmounted stores: * - `companyAutomat`: Refreshes warehouses, carriers, attributes, properties, * suppliers, and brands when `companyAutomat` is active. * - `categoryAutomat`: Refreshes category tree and open category views on `'category'`. * - `articleIndexAutomat`: Refreshes derived article index on `'articles'` / `'warehouse'`. * - `artikelAutomat` & `articleDetailAutomat`: Refreshes watched SKU on `article-`. * - `orderAutomat`, `quoteAutomat`, `shipmentAutomat`, `deliveryAutomat`, `retoureAutomat`: * Reloads when those automats are active. */ import { Automat } from 'react-automat'; import authAutomat from './authAutomat'; /** * @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. * * @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; } } 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. * * @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); }); } // #region connect // MARK: opens GET /api/events EventSource when hasAuth is true function connect() { const auth = authAutomat.state; const hasAuth = auth.hasAuth ?? (auth.status === 'authed' && Boolean(auth.user)); if (!hasAuth) { disconnect(); return; } const companyId = auth.user?.selectedCompanyId != null ? String(auth.user.selectedCompanyId) : null; 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 }); const es = new EventSource(url.href, { withCredentials: true }); addTrackedListener(es, 'company', () => { const comp = Automat.get('company'); if (comp?.actions?.reload) { void comp.actions.reload().catch((err) => { console.error('[sse] company refresh failed', err); }); } }); // Category tree changed: refresh the tree and every open category view. addTrackedListener(es, 'category', () => { const cat = Automat.get('category'); if (cat?.actions?.loadTree) { void cat.actions.loadTree().catch((err) => { console.error('[sse] category tree refresh failed', err); }); } cat?.actions?.invalidateAllCategoryViews?.(); }); // Article index changed: refresh the index and open category views. addTrackedListener(es, 'articles', () => { Automat.get('category')?.actions?.invalidateAllCategoryViews?.(); const idx = Automat.get('articleIndex'); if (idx?.actions?.reload) { void idx.actions.reload().catch((err) => { console.error('[sse] article index refresh failed', err); }); } }); addTrackedListener(es, 'warehouse', () => { const comp = Automat.get('company'); if (comp?.actions?.reloadWarehouses) { void comp.actions.reloadWarehouses().catch((err) => { console.error('[sse] warehouse refresh failed', err); }); } Automat.get('category')?.actions?.invalidateAllCategoryViews?.(); const idx = Automat.get('articleIndex'); if (idx?.actions?.reload) { void idx.actions.reload().catch((err) => { console.error('[sse] article index refresh on warehouse event failed', err); }); } }); addTrackedListener(es, 'carrier', () => { const comp = Automat.get('company'); if (comp?.actions?.reloadCarriers) { void comp.actions.reloadCarriers().catch((err) => { console.error('[sse] carrier refresh failed', err); }); } }); addTrackedListener(es, 'attributename', () => { const comp = Automat.get('company'); if (comp?.actions?.loadAttributes) { void comp.actions.loadAttributes().catch((err) => { console.error('[sse] attributename refresh failed', err); }); } }); addTrackedListener(es, 'propertyname', () => { const comp = Automat.get('company'); if (comp?.actions?.loadProperties) { void comp.actions.loadProperties().catch((err) => { console.error('[sse] propertyname refresh failed', err); }); } }); addTrackedListener(es, 'supplier', () => { const comp = Automat.get('company'); if (comp?.actions?.reloadSuppliers) { void comp.actions.reloadSuppliers().catch((err) => { console.error('[sse] supplier refresh failed', err); }); } }); addTrackedListener(es, 'brand', (event) => { try { const payload = JSON.parse(event?.data ?? '{}'); if (payload?.sender && payload.sender === getSseClientId()) return; } catch {} const comp = Automat.get('company'); if (comp?.actions?.loadBrands) { void comp.actions.loadBrands(true).catch(() => {}); } }); addTrackedListener(es, 'order', () => { const ord = Automat.get('order'); if (ord?.actions?.reload) { void ord.actions.reload().catch((err) => { console.error('[sse] order refresh failed', err); }); } const shp = Automat.get('shipment'); if (shp?.actions?.reload) { void shp.actions.reload().catch((err) => { console.error('[sse] shipment refresh failed', err); }); } }); addTrackedListener(es, 'quote', () => { const q = Automat.get('quote'); if (q?.actions?.reload) { void q.actions.reload().catch((err) => { console.error('[sse] quote refresh failed', err); }); } }); addTrackedListener(es, 'shipment', () => { const shp = Automat.get('shipment'); if (shp?.actions?.reload) { void shp.actions.reload().catch((err) => { console.error('[sse] shipment refresh failed', err); }); } const ord = Automat.get('order'); if (ord?.actions?.reload) { void ord.actions.reload().catch((err) => { console.error('[sse] order refresh failed', err); }); } }); addTrackedListener(es, 'delivery', () => { const del = Automat.get('delivery'); if (del?.actions?.reload) { void del.actions.reload().catch((err) => { console.error('[sse] delivery refresh failed', err); }); } }); addTrackedListener(es, 'retoure', () => { const ret = Automat.get('retoure'); if (ret?.actions?.reload) { void ret.actions.reload().catch((err) => { console.error('[sse] retoure refresh failed', err); }); } const shp = Automat.get('shipment'); if (shp?.actions?.reload) { void shp.actions.reload().catch((err) => { console.error('[sse] shipment refresh failed', err); }); } }); addTrackedListener(es, 'open', () => { sseAutomat.setState({ status: 'connected' }); }); addTrackedListener(es, 'message', () => {}); es.onerror = (e) => { console.error('[sse] event stream error', e); recordEvent(e || { type: 'error', data: '', lastEventId: '' }); sseAutomat.setState({ status: 'error' }); }; eventSource = es; attachHandlers(es); } // #endregion let wasAuthed = false; function handleAuthSync(auth) { const isAuthed = Boolean(auth?.hasAuth ?? (auth?.status === 'authed' && auth?.user)); const companyId = isAuthed && auth?.user?.selectedCompanyId != null ? String(auth.user.selectedCompanyId) : null; const relogin = isAuthed && !wasAuthed; wasAuthed = isAuthed; if (!isAuthed) { disconnect(); sseAutomat.setState({ companyId: null }); return; } if (!relogin && eventSource && companyId === sseAutomat.state.companyId) { return; } disconnect(); sseAutomat.setState({ companyId }); connect(); } const actions = { /** * Closes and reopens the event stream. * * @returns {void} */ reconnect() { disconnect(); connect(); }, /** * Synchronizes the SSE connection with `authAutomat` state. * * @param {import('./authAutomat').AuthState} authState */ syncWithAuth(authState) { handleAuthSync(authState ?? authAutomat.state); }, clearEvents() { sseAutomat.setState({ events: [] }); }, /** * Registers a handler for a statically named backend event. * * @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. * * @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. * * @type {Automat} */ sseAutomat = new Automat(initialState, actions, { name: 'sse' }); /** * Refreshes the cached article plus any open history/supplier view. * * @param {string} sku Article SKU. * @returns {void} */ export function invalidateArticle(sku) { const key = String(sku); const art = Automat.get('artikel'); if (art?.actions?.reloadArticle) { void art.actions.reloadArticle(key).catch(() => {}); } const hist = Automat.get('articleHistory'); if (hist?.has?.(key)) { void hist.reload(key).catch(() => {}); } const sup = Automat.get('articleSupplier'); if (sup?.has?.(key)) { void sup.reload(key).catch(() => {}); } } authAutomat.subscribe(handleAuthSync); handleAuthSync(authAutomat.state); export default sseAutomat;