u

src/automats/sseAutomat.js

@@ -28,6 +28,7 @@ import authAutomat from './authAutomat';
  * @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;
@@ -68,6 +69,7 @@ const initialState = {
   companyId: null,
   clientId: sseClientId,
   events: [],
+  lastEvent: null,
 };
 
 /** Event name prefix for per-article signals. */
@@ -76,18 +78,20 @@ const ARTICLE_EVENT_PREFIX = 'article-';
 let sseAutomat;
 /** @type {EventSource | null} */
 let eventSource = null;
+/** @type {AbortController | null} */
+let streamAbortController = null;
+let reconnectTimer = null;
 
 /**
- * Handlers for statically named events (`category`, `articles`, …).
- * @type {Map<string, Set<{handler: (event: MessageEvent) => void, detach: (() => void)|null}>>}
+ * Handlers for event subscriptions (`category`, `articles`, `article-*`, …).
+ * @type {Map<string, Set<{handler: Function, detach: (() => void)|null}>>}
  */
 const handlers = new Map();
 let nextEventId = 0;
 const seenEvents = new WeakSet();
 
 /**
- * Handlers per watched SKU, so one `article-<sku>` listener is attached while
- * at least one view is open on that SKU.
+ * Handlers per watched SKU.
  * @type {Map<string, Set<(sku: string) => void>>}
  */
 const articleWatchers = new Map();
@@ -99,25 +103,41 @@ function recordEvent(event) {
   const entry = {
     id: ++nextEventId,
     type: event.type,
-    data: typeof event.data === 'string' ? event.data : '',
+    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 addTrackedListener(es, type, handler) {
-  const listener = (event) => {
-    recordEvent(event);
-    handler(event);
-  };
-  es.addEventListener(type, listener);
-  return () => es.removeEventListener(type, listener);
+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 {
@@ -125,21 +145,136 @@ function disconnect() {
     } catch {}
     eventSource = null;
   }
-  if (sseAutomat.state.status !== 'disconnected') {
+  if (sseAutomat && sseAutomat.state.status !== 'disconnected') {
     sseAutomat.setState({ status: 'disconnected' });
   }
 }
 
 /**
- * Attaches every registered handler to a freshly opened stream.
+ * Dispatches an incoming event through registered handlers and article watchers.
  *
- * @param {EventSource} es Open event stream.
- * @returns {void}
+ * @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 === '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);
+
+    // Watchers registered via watchArticle(sku, handler)
+    const watchers = 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, entry.handler);
+      entry.detach = addTrackedListener(es, event, (ev) => dispatchSseEvent(ev));
     }
   }
   for (const [sku] of articleWatchers) {
@@ -147,25 +282,127 @@ function attachHandlers(es) {
   }
 }
 
-/**
- * Binds the single `article-<sku>` 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);
+    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 EventSource when hasAuth is true
+// 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) {
@@ -180,196 +417,11 @@ function connect() {
   console.log('[sse] connecting');
   sseAutomat.setState({ status: 'connecting', companyId });
 
-  const es = new EventSource(url.href, { withCredentials: true });
-
-  addTrackedListener(es, 'user', (event) => {
-    try {
-      const payload = JSON.parse(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);
-    }
-  });
-
-  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);
+  if (typeof fetch === 'function' && typeof ReadableStream !== 'undefined') {
+    void startFetchStream(url, companyId);
+  } else {
+    startEventSource(url, companyId);
+  }
 }
 // #endregion
 
@@ -390,7 +442,7 @@ function handleAuthSync(auth) {
     return;
   }
 
-  if (!relogin && eventSource && companyId === sseAutomat.state.companyId) {
+  if (!relogin && (streamAbortController || eventSource) && companyId === sseAutomat.state.companyId) {
     return;
   }
 
@@ -420,14 +472,15 @@ const actions = {
   },
 
   clearEvents() {
-    sseAutomat.setState({ events: [] });
+    sseAutomat.setState({ events: [], lastEvent: null });
   },
 
   /**
-   * Registers a handler for a statically named backend event.
+   * 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 (e.g. `category`, `articles`).
-   * @param {(event: MessageEvent) => void} handler Signal handler.
+   * @param {string} event Event name or pattern.
+   * @param {Function} handler Signal handler.
    * @returns {() => void} Unsubscribe callback.
    */
   on(event, handler) {
@@ -440,7 +493,9 @@ const actions = {
     if (!entry) {
       entry = { handler, detach: null };
       set.add(entry);
-      if (eventSource) entry.detach = addTrackedListener(eventSource, event, handler);
+      if (eventSource && event !== '*' && event !== 'article-*') {
+        entry.detach = addTrackedListener(eventSource, event, handler);
+      }
     }
     return () => {
       entry.detach?.();
@@ -450,7 +505,17 @@ const actions = {
   },
 
   /**
-   * Subscribes to `article-<sku>` signals for one SKU.
+   * Subscribes to any article mutation event (`article-<sku>`).
+   *
+   * @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-<sku>` signals for one specific SKU.
    *
    * @param {string} sku Article SKU to watch.
    * @param {(sku: string) => void} handler Called with the changed SKU.
@@ -473,6 +538,22 @@ const actions = {
       if (current.size === 0) articleWatchers.delete(key);
     };
   },
+
+  /**
+   * 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,
+    });
+  },
 };
 
 /**