/** * @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'; import { articleHistoryAutomat, articleSupplierAutomat } from './articleDetailAutomat'; /** * @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. * @property {{id: number, type: string, data: string, lastEventId: string, time: number}|null} lastEvent Most recent event. */ 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: [], lastEvent: null, }; /** Event name prefix for per-article signals. */ const ARTICLE_EVENT_PREFIX = 'article-'; let sseAutomat; /** @type {EventSource | null} */ let eventSource = null; /** @type {AbortController | null} */ let streamAbortController = null; let reconnectTimer = null; /** * Handlers for event subscriptions (`category`, `articles`, `article-*`, …). * @type {Map void)|null}>>} */ const handlers = new Map(); let nextEventId = 0; const seenEvents = new WeakSet(); /** * Handlers per watched 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 : (event.data ? JSON.stringify(event.data) : ''), lastEventId: event.lastEventId ?? '', time: Date.now(), }; sseAutomat.setState((state) => ({ events: [...state.events.slice(-(EVENT_HISTORY_LIMIT - 1)), entry], lastEvent: entry, })); } function clearReconnectTimer() { if (reconnectTimer) { clearTimeout(reconnectTimer); reconnectTimer = null; } } function scheduleReconnect() { clearReconnectTimer(); const auth = authAutomat.state; const hasAuth = auth.hasAuth ?? (auth.status === 'authed' && Boolean(auth.user)); if (!hasAuth) return; reconnectTimer = setTimeout(() => { reconnectTimer = null; connect(); }, 1500); } function disconnect() { clearReconnectTimer(); if (streamAbortController) { streamAbortController.abort(); streamAbortController = null; } if (eventSource) { console.log('[sse] disconnect'); try { eventSource.close(); } catch {} eventSource = null; } if (sseAutomat && sseAutomat.state.status !== 'disconnected') { sseAutomat.setState({ status: 'disconnected' }); } } /** * Dispatches an incoming event through registered handlers and article watchers. * * @param {{ type: string, data: any, lastEventId?: string }} event */ function dispatchSseEvent(event) { recordEvent(event); const eventType = String(event.type); // 1. Built-in system handling (e.g. user / company switch sync) if (eventType === 'user') { try { const payload = typeof event.data === 'string' ? JSON.parse(event.data || '{}') : (event.data || {}); const nextCompanyId = payload.selectedCompanyId ?? null; if (authAutomat.state.user?.selectedCompanyId !== nextCompanyId) { authAutomat.setState((prev) => ({ ...prev, user: prev.user ? { ...prev.user, selectedCompanyId: nextCompanyId } : prev.user, hasCompany: Boolean((prev.hasAuth ?? (prev.status === 'authed' && prev.user)) && nextCompanyId != null), })); } } catch (err) { console.error('[sse] failed to handle user event', err); } } // 2. Dispatch to exact handlers registered via on(eventType, handler) const exactSet = handlers.get(eventType); if (exactSet && exactSet.size > 0) { for (const entry of exactSet) { try { entry.handler(event); } catch (err) { console.error(`[sse] error in handler for ${eventType}:`, err); } } } else { // Fallback direct dispatch if store didn't subscribe yet if (eventType === 'articles') { Automat.get('category')?.actions?.invalidateAllCategoryViews?.(); void Automat.get('articleIndex')?.actions?.reload?.().catch(() => {}); } else if (eventType === 'category') { void Automat.get('category')?.actions?.loadTree?.().catch(() => {}); Automat.get('category')?.actions?.invalidateAllCategoryViews?.(); } else if (eventType === 'company') { void Automat.get('company')?.actions?.reload?.().catch(() => {}); } else if (eventType === 'warehouse') { void Automat.get('company')?.actions?.reloadWarehouses?.().catch(() => {}); Automat.get('category')?.actions?.invalidateAllCategoryViews?.(); void Automat.get('articleIndex')?.actions?.reload?.().catch(() => {}); } else if (eventType === 'supplier') { void Automat.get('company')?.actions?.reloadSuppliers?.().catch(() => {}); } else if (eventType === 'brand') { void Automat.get('company')?.actions?.loadBrands?.(true).catch(() => {}); } else if (eventType === 'unit') { void Automat.get('company')?.actions?.loadUnits?.().catch(() => {}); } else if (eventType === 'carrier') { void Automat.get('company')?.actions?.reloadCarriers?.().catch(() => {}); } else if (eventType === 'posterminal') { void Automat.get('company')?.actions?.loadTerminals?.().catch(() => {}); } else if (eventType === 'customergroup') { void Automat.get('company')?.actions?.reloadCustomergroups?.().catch(() => {}); } else if (eventType === 'propertyname') { void Automat.get('company')?.actions?.loadProperties?.().catch(() => {}); } else if (eventType === 'attributename') { void Automat.get('company')?.actions?.loadAttributes?.().catch(() => {}); } else if (eventType === 'account') { void Automat.get('company')?.actions?.reloadAccounts?.().catch(() => {}); } else if (eventType === 'order') { void Automat.get('order')?.actions?.reload?.().catch(() => {}); void Automat.get('shipment')?.actions?.reload?.().catch(() => {}); } else if (eventType === 'delivery') { void Automat.get('delivery')?.actions?.reload?.().catch(() => {}); } else if (eventType === 'retoure') { void Automat.get('retoure')?.actions?.reload?.().catch(() => {}); } else if (eventType === 'quote') { void Automat.get('quote')?.actions?.reload?.().catch(() => {}); } else if (eventType === 'shipment') { void Automat.get('shipment')?.actions?.reload?.().catch(() => {}); } } // 3. Dispatch to article-specific handlers if it starts with article- if (eventType.startsWith(ARTICLE_EVENT_PREFIX)) { const sku = eventType.slice(ARTICLE_EVENT_PREFIX.length); const lowerSku = sku.trim().toLowerCase(); // Watchers registered via watchArticle(sku, handler) const watchers = articleWatchers.get(lowerSku) || articleWatchers.get(sku); if (watchers) { for (const handler of watchers) { try { handler(sku); } catch (err) { console.error(`[sse] error in article watcher for ${sku}:`, err); } } } // Article wildcard handlers (registered via on('article-*', ...) or onArticle(...)) const articleWildcard = handlers.get('article-*'); if (articleWildcard) { for (const entry of articleWildcard) { try { entry.handler(sku, event); } catch (err) { console.error('[sse] error in article-* handler:', err); } } } // Single article detail & history/supplier caches invalidation invalidateArticle(sku); } // 4. Wildcard handlers registered via on('*', handler) const allWildcard = handlers.get('*'); if (allWildcard) { for (const entry of allWildcard) { try { entry.handler(event); } catch (err) { console.error('[sse] error in * handler:', err); } } } } function addTrackedListener(es, type, handler) { const listener = (event) => { dispatchSseEvent(event); handler(event); }; es.addEventListener(type, listener); return () => es.removeEventListener(type, listener); } function attachHandlers(es) { for (const [event, set] of handlers) { if (event === '*' || event === 'article-*') continue; for (const entry of set) { entry.detach = addTrackedListener(es, event, (ev) => dispatchSseEvent(ev)); } } for (const [sku] of articleWatchers) { attachArticleListener(sku); } } function attachArticleListener(sku) { if (!eventSource) return; addTrackedListener(eventSource, `${ARTICLE_EVENT_PREFIX}${sku}`, (event) => { const changed = String(event.type).slice(ARTICLE_EVENT_PREFIX.length); const set = articleWatchers.get(sku); if (set) { for (const handler of set) handler(changed); } }); } async function startFetchStream(url, companyId) { if (streamAbortController) { streamAbortController.abort(); streamAbortController = null; } const controller = new AbortController(); streamAbortController = controller; const signal = controller.signal; try { const res = await fetch(url.href, { credentials: 'include', headers: { Accept: 'text/event-stream' }, signal, }); if (signal.aborted) return; if (!res.ok) { if (res.status === 401) { sseAutomat.setState({ status: 'disconnected' }); return; } throw new Error(`SSE HTTP ${res.status}`); } if (!res.body) { throw new Error('Response body is null'); } sseAutomat.setState({ status: 'connected', companyId }); const reader = res.body.getReader(); const decoder = new TextDecoder(); let buffer = ''; let eventType = 'message'; let eventData = ''; let lastEventId = ''; while (!signal.aborted) { const { done, value } = await reader.read(); if (done) break; buffer += decoder.decode(value, { stream: true }); const lines = buffer.split(/\r\n|\r|\n/); buffer = lines.pop() ?? ''; for (const line of lines) { if (line === '') { if (eventData || eventType !== 'message') { dispatchSseEvent({ type: eventType, data: eventData, lastEventId, }); } eventType = 'message'; eventData = ''; } else if (line.startsWith(':')) { continue; } else if (line.startsWith('event:')) { eventType = line.slice(6).trim(); } else if (line.startsWith('data:')) { const chunk = line.slice(5).trim(); eventData = eventData ? `${eventData}\n${chunk}` : chunk; } else if (line.startsWith('id:')) { lastEventId = line.slice(3).trim(); } } } } catch (err) { if (signal.aborted) return; console.error('[sse] stream error', err); recordEvent({ type: 'error', data: err?.message || '', lastEventId: '' }); sseAutomat.setState({ status: 'error' }); } finally { if (!signal.aborted) { scheduleReconnect(); } } } function startEventSource(url, companyId) { const es = new EventSource(url.href, { withCredentials: true }); eventSource = es; addTrackedListener(es, 'open', () => { sseAutomat.setState({ status: 'connected', companyId }); }); addTrackedListener(es, 'user', (event) => { dispatchSseEvent(event); }); addTrackedListener(es, 'message', (event) => { dispatchSseEvent(event); }); es.onerror = (e) => { console.error('[sse] event stream error', e); recordEvent(e || { type: 'error', data: '', lastEventId: '' }); sseAutomat.setState({ status: 'error' }); }; attachHandlers(es); } // #region connect // MARK: opens GET /api/events stream when hasAuth is true function connect() { clearReconnectTimer(); 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 }); if (typeof fetch === 'function' && typeof ReadableStream !== 'undefined') { void startFetchStream(url, companyId); } else { startEventSource(url, companyId); } } // #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 && (streamAbortController || 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: [], lastEvent: null }); }, /** * Registers a handler for a named backend event or pattern. * Supports exact event names (e.g. `category`, `articles`), `article-*` wildcard, or `*`. * * @param {string} event Event name or pattern. * @param {Function} 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 && event !== '*' && event !== 'article-*') { entry.detach = addTrackedListener(eventSource, event, handler); } } return () => { entry.detach?.(); set.delete(entry); if (set.size === 0) handlers.delete(event); }; }, /** * Subscribes to any article mutation event (`article-`). * * @param {(sku: string, event: object) => void} handler Called with changed SKU. * @returns {() => void} Unsubscribe callback. */ onArticle(handler) { return actions.on('article-*', (sku, event) => handler(sku, event)); }, /** * Subscribes to `article-` signals for one specific 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 rawKey = String(sku).trim(); const lowerKey = rawKey.toLowerCase(); let set = articleWatchers.get(lowerKey); if (!set) { set = new Set(); articleWatchers.set(lowerKey, set); attachArticleListener(rawKey); } set.add(handler); return () => { const current = articleWatchers.get(lowerKey); if (!current) return; current.delete(handler); if (current.size === 0) articleWatchers.delete(lowerKey); }; }, /** * Programmatically dispatches an event through sseAutomat. * * @param {string} type Event type (e.g. 'article-ART-000001', 'articles'). * @param {any} [data] Event data payload. * @param {string} [lastEventId] Optional event ID. * @returns {void} */ dispatch(type, data = {}, lastEventId = '') { dispatchSseEvent({ type, data: typeof data === 'string' ? data : JSON.stringify(data), lastEventId, }); }, }; /** * 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(() => {}); } if (articleHistoryAutomat.has(key)) { void articleHistoryAutomat.reload(key).catch(() => {}); } // Reload assigned suppliers and sync with companyAutomat void articleSupplierAutomat.reload(key).catch(() => {}); } authAutomat.subscribe(handleAuthSync); handleAuthSync(authAutomat.state); export default sseAutomat;