dashfront/src/automats/sseAutomat.js

/**
 * @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-<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.
 */

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 { handleDraftRequest, 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<string, Set<{handler: (event: MessageEvent) => void, 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.
 * @type {Map<string, Set<(sku: string) => 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-<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}
 */
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);
    });
  });

  addTrackedListener(es, 'quote', () => {
    void quoteAutomat.actions.reload().catch((err) => {
      console.error('[sse] quote refresh failed', err);
    });
  });

  addTrackedListener(es, 'shipment', () => {
    void orderAutomat.actions.reload().catch((err) => {
      console.error('[sse] shipment refresh failed', err);
    });
  });

  addTrackedListener(es, 'draft:request', (event) => {
    try {
      const payload = typeof event.data === 'string' ? JSON.parse(event.data) : event.data;
      void handleDraftRequest(payload);
    } catch (err) {
      console.error('[sse] draft:request handling 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-<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.
   */
  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<SseState>}
 */
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;