328 lines
11 KiB
JavaScript
328 lines
11 KiB
JavaScript
// ==========================================
|
||
// Chat Switchboard – EventBus
|
||
// ==========================================
|
||
// Labeled event bus with WebSocket bridge.
|
||
// Components subscribe by label (supports * wildcard).
|
||
// Events route: local-only, FE→BE, BE→FE, or both.
|
||
//
|
||
// Usage:
|
||
// Events.on('chat.message.*', (payload, meta) => { ... });
|
||
// Events.emit('chat.typing.abc123', { user: 'jeff' });
|
||
// Events.connect('/ws');
|
||
// ==========================================
|
||
|
||
const Events = {
|
||
|
||
_handlers: new Map(), // label → Set<{fn, once}>
|
||
_ws: null,
|
||
_wsUrl: null,
|
||
_wsReconnectMs: 1000,
|
||
_wsMaxReconnectMs: 30000,
|
||
_wsReconnectTimer: null,
|
||
_wsConnected: false,
|
||
_wsQueue: [], // buffered while disconnected
|
||
|
||
// Labels that should NOT cross the wire
|
||
_localOnly: new Set([
|
||
'ui.modal.open', 'ui.modal.close',
|
||
'ui.sidebar.toggle', 'ui.sidebar.collapse',
|
||
'ui.toast',
|
||
]),
|
||
|
||
// Labels the FE should never receive from BE
|
||
// (enforced server-side, this is defense-in-depth)
|
||
_serverOnly: new Set([
|
||
'plugin.hook.pre_completion',
|
||
'plugin.hook.post_completion',
|
||
'internal.',
|
||
'db.',
|
||
]),
|
||
|
||
// ── Subscribe ────────────────────────────
|
||
|
||
/**
|
||
* Subscribe to events matching a label pattern.
|
||
* Supports exact match and trailing wildcard:
|
||
* 'chat.message.abc123' — exact
|
||
* 'chat.message.*' — any chat.message.{id}
|
||
* 'chat.*' — any chat.{anything}
|
||
*
|
||
* @param {string} label - Event label or pattern
|
||
* @param {Function} fn - Handler(payload, meta)
|
||
* @returns {Function} unsubscribe function
|
||
*/
|
||
on(label, fn) {
|
||
if (!this._handlers.has(label)) this._handlers.set(label, new Set());
|
||
const entry = { fn, once: false };
|
||
this._handlers.get(label).add(entry);
|
||
return () => this._handlers.get(label)?.delete(entry);
|
||
},
|
||
|
||
/**
|
||
* Subscribe for a single event, then auto-unsubscribe.
|
||
*/
|
||
once(label, fn) {
|
||
if (!this._handlers.has(label)) this._handlers.set(label, new Set());
|
||
const entry = { fn, once: true };
|
||
this._handlers.get(label).add(entry);
|
||
return () => this._handlers.get(label)?.delete(entry);
|
||
},
|
||
|
||
/**
|
||
* Remove all handlers for a label, or a specific handler.
|
||
*/
|
||
off(label, fn) {
|
||
if (!fn) {
|
||
this._handlers.delete(label);
|
||
} else {
|
||
const set = this._handlers.get(label);
|
||
if (set) {
|
||
for (const entry of set) {
|
||
if (entry.fn === fn) { set.delete(entry); break; }
|
||
}
|
||
}
|
||
}
|
||
},
|
||
|
||
// ── Emit ─────────────────────────────────
|
||
|
||
/**
|
||
* Emit an event. Dispatches locally and sends via WebSocket
|
||
* unless the label is in _localOnly.
|
||
*
|
||
* @param {string} label - Event label
|
||
* @param {*} payload - Event data
|
||
* @param {object} opts - { localOnly: bool, room: string }
|
||
*/
|
||
emit(label, payload, opts = {}) {
|
||
const meta = {
|
||
event: label,
|
||
room: opts.room || null,
|
||
ts: Date.now(),
|
||
local: true,
|
||
};
|
||
|
||
// Local dispatch
|
||
this._dispatch(label, payload, meta);
|
||
|
||
// Send over WebSocket unless local-only
|
||
if (!opts.localOnly && !this._isLocalOnly(label)) {
|
||
this._send({ event: label, room: meta.room, payload, ts: meta.ts });
|
||
}
|
||
},
|
||
|
||
// ── Internal dispatch ────────────────────
|
||
|
||
_dispatch(label, payload, meta) {
|
||
// Check every registered pattern against this label
|
||
for (const [pattern, entries] of this._handlers) {
|
||
if (this._match(label, pattern)) {
|
||
for (const entry of entries) {
|
||
try {
|
||
entry.fn(payload, meta);
|
||
} catch (e) {
|
||
console.error(`[EventBus] Handler error for ${pattern}:`, e);
|
||
}
|
||
if (entry.once) entries.delete(entry);
|
||
}
|
||
}
|
||
}
|
||
},
|
||
|
||
/**
|
||
* Match a concrete label against a subscription pattern.
|
||
* 'chat.message.abc' matches:
|
||
* 'chat.message.abc' (exact)
|
||
* 'chat.message.*' (wildcard last segment)
|
||
* 'chat.*' (wildcard all after chat.)
|
||
* '*' (match everything)
|
||
*/
|
||
_match(label, pattern) {
|
||
if (pattern === '*') return true;
|
||
if (pattern === label) return true;
|
||
if (!pattern.endsWith('*')) return false;
|
||
const prefix = pattern.slice(0, -1); // 'chat.message.' from 'chat.message.*'
|
||
return label.startsWith(prefix);
|
||
},
|
||
|
||
_isLocalOnly(label) {
|
||
if (this._localOnly.has(label)) return true;
|
||
for (const prefix of this._localOnly) {
|
||
if (prefix.endsWith('.') && label.startsWith(prefix)) return true;
|
||
}
|
||
return false;
|
||
},
|
||
|
||
// ── WebSocket Bridge ─────────────────────
|
||
|
||
/**
|
||
* Connect to WebSocket endpoint. Auto-reconnects on drop.
|
||
* @param {string} url - WebSocket URL (e.g. '/ws' or 'wss://host/ws')
|
||
*/
|
||
connect(url) {
|
||
this._wsUrl = url;
|
||
this._wsReconnectMs = 1000;
|
||
this._wsRetries = 0;
|
||
this._wsMaxRetries = 5;
|
||
this._doConnect();
|
||
},
|
||
|
||
disconnect() {
|
||
if (this._wsReconnectTimer) clearTimeout(this._wsReconnectTimer);
|
||
this._wsReconnectTimer = null;
|
||
this._wsRetries = 0;
|
||
if (this._ws) {
|
||
this._ws.onclose = null; // prevent reconnect
|
||
this._ws.close();
|
||
this._ws = null;
|
||
}
|
||
this._wsConnected = false;
|
||
this._dispatch('ws.disconnected', {}, { event: 'ws.disconnected', ts: Date.now(), local: true });
|
||
},
|
||
|
||
_doConnect() {
|
||
if (!this._wsUrl) return;
|
||
|
||
// Build full URL
|
||
let url = this._wsUrl;
|
||
if (url.startsWith('/')) {
|
||
const proto = location.protocol === 'https:' ? 'wss:' : 'ws:';
|
||
url = `${proto}//${location.host}${url}`;
|
||
}
|
||
|
||
// Add auth token as query param (WebSocket doesn't support headers)
|
||
const tokens = JSON.parse(localStorage.getItem(typeof _storageKey !== 'undefined' ? _storageKey : 'sb_auth') || '{}');
|
||
if (tokens.access) {
|
||
url += (url.includes('?') ? '&' : '?') + `token=${encodeURIComponent(tokens.access)}`;
|
||
} else {
|
||
console.warn('[EventBus] No auth token — skipping WebSocket');
|
||
return;
|
||
}
|
||
|
||
try {
|
||
this._ws = new WebSocket(url);
|
||
} catch (e) {
|
||
console.warn('[EventBus] WebSocket creation failed:', e);
|
||
this._scheduleReconnect();
|
||
return;
|
||
}
|
||
|
||
this._ws.onopen = () => {
|
||
console.log('[EventBus] WebSocket connected');
|
||
this._wsConnected = true;
|
||
this._wsReconnectMs = 1000;
|
||
this._wsRetries = 0;
|
||
|
||
// Flush queued events
|
||
while (this._wsQueue.length > 0) {
|
||
const msg = this._wsQueue.shift();
|
||
this._ws.send(JSON.stringify(msg));
|
||
}
|
||
|
||
this._dispatch('ws.connected', {}, { event: 'ws.connected', ts: Date.now(), local: true });
|
||
};
|
||
|
||
this._ws.onmessage = (evt) => {
|
||
try {
|
||
const msg = JSON.parse(evt.data);
|
||
|
||
// Server pong
|
||
if (msg.event === 'pong') return;
|
||
|
||
// Drop events that shouldn't reach the frontend
|
||
for (const prefix of this._serverOnly) {
|
||
if (msg.event.startsWith(prefix)) return;
|
||
}
|
||
|
||
const meta = {
|
||
event: msg.event,
|
||
room: msg.room || null,
|
||
ts: msg.ts || Date.now(),
|
||
local: false,
|
||
};
|
||
|
||
this._dispatch(msg.event, msg.payload, meta);
|
||
} catch (e) {
|
||
console.warn('[EventBus] Bad message:', evt.data);
|
||
}
|
||
};
|
||
|
||
this._ws.onclose = (evt) => {
|
||
this._wsConnected = false;
|
||
this._ws = null;
|
||
if (evt.code !== 1000) {
|
||
// Abnormal close — log with context
|
||
console.log(`[EventBus] WebSocket closed (code=${evt.code}, reason=${evt.reason || 'none'})`);
|
||
}
|
||
this._dispatch('ws.disconnected', { code: evt.code }, { event: 'ws.disconnected', ts: Date.now(), local: true });
|
||
this._scheduleReconnect();
|
||
};
|
||
|
||
this._ws.onerror = () => {
|
||
// onclose will fire after this — don't double-log
|
||
};
|
||
|
||
// Heartbeat
|
||
this._startHeartbeat();
|
||
},
|
||
|
||
_scheduleReconnect() {
|
||
if (!this._wsUrl) return;
|
||
if (this._wsReconnectTimer) return;
|
||
|
||
this._wsRetries++;
|
||
if (this._wsRetries > this._wsMaxRetries) {
|
||
console.warn(`[EventBus] WebSocket failed after ${this._wsMaxRetries} attempts — giving up. Real-time features unavailable.`);
|
||
this._dispatch('ws.failed', { retries: this._wsRetries }, { event: 'ws.failed', ts: Date.now(), local: true });
|
||
return;
|
||
}
|
||
|
||
console.log(`[EventBus] Reconnecting in ${this._wsReconnectMs}ms (attempt ${this._wsRetries}/${this._wsMaxRetries})`);
|
||
this._wsReconnectTimer = setTimeout(() => {
|
||
this._wsReconnectTimer = null;
|
||
this._doConnect();
|
||
}, this._wsReconnectMs);
|
||
|
||
// Exponential backoff with cap
|
||
this._wsReconnectMs = Math.min(this._wsReconnectMs * 2, this._wsMaxReconnectMs);
|
||
},
|
||
|
||
_send(msg) {
|
||
if (this._wsConnected && this._ws?.readyState === WebSocket.OPEN) {
|
||
this._ws.send(JSON.stringify(msg));
|
||
} else {
|
||
// Buffer up to 100 events while disconnected
|
||
if (this._wsQueue.length < 100) this._wsQueue.push(msg);
|
||
}
|
||
},
|
||
|
||
_heartbeatInterval: null,
|
||
_startHeartbeat() {
|
||
if (this._heartbeatInterval) clearInterval(this._heartbeatInterval);
|
||
this._heartbeatInterval = setInterval(() => {
|
||
if (this._wsConnected) {
|
||
this._send({ event: 'ping', payload: {}, ts: Date.now() });
|
||
}
|
||
}, 30000);
|
||
},
|
||
|
||
// ── Utility ──────────────────────────────
|
||
|
||
/** Check if WebSocket is connected */
|
||
get connected() { return this._wsConnected; },
|
||
|
||
/** Remove all handlers (for cleanup / logout) */
|
||
clear() {
|
||
this._handlers.clear();
|
||
},
|
||
|
||
/** Debug: list all subscriptions */
|
||
debug() {
|
||
const subs = {};
|
||
for (const [label, entries] of this._handlers) {
|
||
subs[label] = entries.size;
|
||
}
|
||
return subs;
|
||
}
|
||
};
|