feat(dashboard): add performance and operations slices
This commit is contained in:
1 parent
20fa83f245
commit
5917b63755
13 files changed
+1059
No files matched your search
@@ -0,0 +1,143 @@
|
||||
import type { QueryKey } from '@tanstack/react-query';
|
||||
import {
|
||||
DashboardEventSchema,
|
||||
type DashboardEvent,
|
||||
} from '../../../../../packages/api-contracts/src/index.js';
|
||||
|
||||
export const DASHBOARD_EVENT_TYPES = [
|
||||
'bot.status.changed',
|
||||
'message.sent',
|
||||
'message.failed',
|
||||
'conversation.updated',
|
||||
'reply.received',
|
||||
'sync.started',
|
||||
'sync.progress',
|
||||
'sync.completed',
|
||||
'sync.failed',
|
||||
'rate-limit.changed',
|
||||
'reddit.session.changed',
|
||||
'log.created',
|
||||
] as const;
|
||||
|
||||
export interface DashboardEventSource {
|
||||
addEventListener(type: string, listener: (event: unknown) => void): void;
|
||||
close(): void;
|
||||
}
|
||||
|
||||
export interface RealtimeConnectionSnapshot {
|
||||
status: 'idle' | 'connecting' | 'connected' | 'reconnecting' | 'closed';
|
||||
isStale: boolean;
|
||||
lastEventId?: string;
|
||||
}
|
||||
|
||||
export interface DashboardRealtimeClient {
|
||||
start(): void;
|
||||
stop(): void;
|
||||
getSnapshot(): RealtimeConnectionSnapshot;
|
||||
}
|
||||
|
||||
export interface DashboardRealtimeClientOptions {
|
||||
url?: string;
|
||||
sourceFactory(url: string): DashboardEventSource;
|
||||
onEvent(event: DashboardEvent): void;
|
||||
onStateChange?(snapshot: RealtimeConnectionSnapshot): void;
|
||||
onInvalidEvent?(value: unknown): void;
|
||||
}
|
||||
|
||||
function eventData(value: unknown): string | undefined {
|
||||
if (typeof value !== 'object' || value === null || !('data' in value)) return undefined;
|
||||
return typeof value.data === 'string' ? value.data : undefined;
|
||||
}
|
||||
|
||||
export function createDashboardRealtimeClient(
|
||||
options: DashboardRealtimeClientOptions,
|
||||
): DashboardRealtimeClient {
|
||||
let source: DashboardEventSource | undefined;
|
||||
let snapshot: RealtimeConnectionSnapshot = { status: 'idle', isStale: true };
|
||||
|
||||
function update(next: RealtimeConnectionSnapshot): void {
|
||||
snapshot = next;
|
||||
options.onStateChange?.(snapshot);
|
||||
}
|
||||
|
||||
function receive(value: unknown): void {
|
||||
const data = eventData(value);
|
||||
if (data === undefined) {
|
||||
options.onInvalidEvent?.(value);
|
||||
return;
|
||||
}
|
||||
let decoded: unknown;
|
||||
try {
|
||||
decoded = JSON.parse(data);
|
||||
} catch {
|
||||
options.onInvalidEvent?.(data);
|
||||
return;
|
||||
}
|
||||
const result = DashboardEventSchema.safeParse(decoded);
|
||||
if (!result.success) {
|
||||
options.onInvalidEvent?.(decoded);
|
||||
return;
|
||||
}
|
||||
snapshot = { ...snapshot, lastEventId: result.data.id };
|
||||
options.onEvent(result.data);
|
||||
}
|
||||
|
||||
return {
|
||||
start() {
|
||||
if (source !== undefined) return;
|
||||
update({ ...snapshot, status: 'connecting', isStale: true });
|
||||
source = options.sourceFactory(options.url ?? '/api/events/stream');
|
||||
source.addEventListener('open', () => {
|
||||
update({ ...snapshot, status: 'connected', isStale: false });
|
||||
});
|
||||
source.addEventListener('error', () => {
|
||||
update({ ...snapshot, status: 'reconnecting', isStale: true });
|
||||
});
|
||||
for (const type of DASHBOARD_EVENT_TYPES) source.addEventListener(type, receive);
|
||||
},
|
||||
stop() {
|
||||
source?.close();
|
||||
source = undefined;
|
||||
update({ ...snapshot, status: 'closed', isStale: true });
|
||||
},
|
||||
getSnapshot() {
|
||||
return snapshot;
|
||||
},
|
||||
};
|
||||
}
|
||||
|
||||
export function queryKeysForDashboardEvent(event: DashboardEvent): readonly QueryKey[] {
|
||||
switch (event.type) {
|
||||
case 'bot.status.changed':
|
||||
return [['dashboard', 'stats']];
|
||||
case 'rate-limit.changed':
|
||||
return [['dashboard', 'stats'], ['dashboard', 'timers']];
|
||||
case 'message.sent':
|
||||
case 'message.failed':
|
||||
return [['conversations'], ['performance']];
|
||||
case 'conversation.updated':
|
||||
case 'reply.received':
|
||||
return [['conversations'], ['operations', 'activity']];
|
||||
case 'sync.started':
|
||||
case 'sync.progress':
|
||||
case 'sync.completed':
|
||||
case 'sync.failed':
|
||||
return [['operations', 'sync-status'], ['conversations'], ['media']];
|
||||
case 'reddit.session.changed':
|
||||
return [['reddit', 'session']];
|
||||
case 'log.created':
|
||||
return [];
|
||||
}
|
||||
}
|
||||
|
||||
export function createBrowserEventSource(url: string): DashboardEventSource {
|
||||
const source = new EventSource(url);
|
||||
return {
|
||||
addEventListener(type, listener) {
|
||||
source.addEventListener(type, (event) => listener(event));
|
||||
},
|
||||
close() {
|
||||
source.close();
|
||||
},
|
||||
};
|
||||
}
|
||||
Reference in new issue
Block a user