This commit is contained in:
seb
2026-10-28 00:08:45 +01:00
parent fab291f68e
commit 96d0ae6742
88 changed files with 3871 additions and 5334 deletions

View File

@@ -1,44 +1,26 @@
/**
* @file sseAutomat.js
* @description Central Server-Sent Events hub managing the single persistent
* `GET /api/events` connection and fanning mutation signals into domain automats.
* `GET /api/events` connection and fanning mutation signals into active automats.
*
* Upstream Triggers:
* - `authAutomat`: Subscribes to auth state changes — reconnects stream when
* `selectedCompanyId` changes/becomes available; disconnects on `'anon'` (logout).
* - `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 specific automats:
* - `artikelAutomat`: Re-fetches article via `reloadArticle(sku)` on `article-<sku>` signals.
* - `articleDetailAutomat` (`articleHistoryAutomat`, `articleSupplierAutomat`):
* Reloads SKU-keyed views on `article-<sku>` 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.
* 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-<sku>`.
* - `orderAutomat`, `quoteAutomat`, `shipmentAutomat`, `deliveryAutomat`, `retoureAutomat`:
* Reloads when those automats are active.
*/
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 deliveryAutomat from './deliveryAutomat';
import orderAutomat from './orderAutomat';
import quoteAutomat from './quoteAutomat';
import retoureAutomat from './retoureAutomat';
import shipmentAutomat from './shipmentAutomat';
import carrierAutomat from './carrierAutomat';
/**
* @typedef {Object} SseState
@@ -49,7 +31,6 @@ import carrierAutomat from './carrierAutomat';
*/
const EVENT_HISTORY_LIMIT = 200;
const SSE_CLIENT_ID_KEY = 'sse.cid';
function createClientId() {
@@ -63,8 +44,6 @@ function createClientId() {
/**
* 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.
*/
@@ -81,7 +60,6 @@ export function getSseClientId() {
}
}
// Created once per tab at automat creation time and stored to sessionStorage.
const sseClientId = typeof window !== 'undefined' ? getSseClientId() : null;
/** @type {SseState} */
@@ -172,10 +150,6 @@ function attachHandlers(es) {
/**
* Binds the single `article-<sku>` 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}
*/
@@ -189,119 +163,179 @@ function attachArticleListener(sku) {
});
}
// #region connect
// MARK: opens GET /api/events EventSource when hasAuth is true
function connect() {
const companyId = authAutomat.state.user?.selectedCompanyId;
if (companyId == null) {
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: String(companyId) });
console.log('[sse] connecting');
sseAutomat.setState({ status: 'connecting', 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();
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', () => {
invalidateAllCategoryViews();
void articleIndexAutomat.actions.reload().catch((err) => {
console.error('[sse] article index 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 failed', err);
});
}
});
addTrackedListener(es, 'warehouse', () => {
void warehouseAutomat.actions.reload().catch((err) => {
console.error('[sse] warehouse refresh failed', err);
});
invalidateAllCategoryViews();
void articleIndexAutomat.actions.reload().catch((err) => {
console.error('[sse] article index refresh on warehouse event failed', err);
});
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', () => {
void carrierAutomat.actions.reload().catch((err) => {
console.error('[sse] carrier refresh failed', err);
});
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', () => {
void attributeAutomat.actions.load().catch((err) => {
console.error('[sse] attributename refresh failed', err);
});
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', () => {
void propertyAutomat.actions.load().catch((err) => {
console.error('[sse] propertyname refresh failed', err);
});
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', () => {
void supplierAutomat.actions.reload().catch((err) => {
console.error('[sse] supplier refresh failed', err);
});
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', () => {
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);
});
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', () => {
void quoteAutomat.actions.reload().catch((err) => {
console.error('[sse] quote refresh failed', err);
});
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', () => {
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);
});
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', () => {
void deliveryAutomat.actions.reload().catch((err) => {
console.error('[sse] delivery refresh failed', err);
});
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', () => {
void retoureAutomat.actions.reload().catch((err) => {
console.error('[sse] retoure refresh failed', err);
});
void shipmentAutomat.actions.reload().catch((err) => {
console.error('[sse] shipment refresh failed', err);
});
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', () => {});
// EventSource reconnects on its own; only surface the state.
es.onerror = (e) => {
console.error('[sse] event stream error', e);
recordEvent(e || { type: 'error', data: '', lastEventId: '' });
@@ -311,11 +345,37 @@ function connect() {
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. Required after a company switch,
* because the stream is scoped to the selected company at connect time.
* Closes and reopens the event stream.
*
* @returns {void}
*/
@@ -324,6 +384,15 @@ const actions = {
connect();
},
/**
* Synchronizes the SSE connection with `authAutomat` state.
*
* @param {import('./authAutomat').AuthState} authState
*/
syncWithAuth(authState) {
handleAuthSync(authState ?? authAutomat.state);
},
clearEvents() {
sseAutomat.setState({ events: [] });
},
@@ -331,9 +400,6 @@ const actions = {
/**
* 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.
@@ -360,9 +426,6 @@ const actions = {
/**
* Subscribes to `article-<sku>` 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.
@@ -389,9 +452,6 @@ const actions = {
/**
* 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<SseState>}
*/
sseAutomat = new Automat(initialState, actions, { name: 'sse' });
@@ -399,48 +459,26 @@ 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(() => {});
const art = Automat.get('artikel');
if (art?.actions?.reloadArticle) {
void art.actions.reloadArticle(key).catch(() => {});
}
if (articleSupplierAutomat.has(key)) {
void articleSupplierAutomat.reload(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(() => {});
}
}
// 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();
});
authAutomat.subscribe(handleAuthSync);
handleAuthSync(authAutomat.state);
export default sseAutomat;