208 lines
8.7 KiB
TypeScript
208 lines
8.7 KiB
TypeScript
import React, { useEffect, useRef, useCallback, useState } from 'react';
|
|
import type {
|
|
WSDashboardInit,
|
|
WSAgentOffline,
|
|
WSStatsUpdate,
|
|
WSCommandResult,
|
|
WSAgentLog,
|
|
} from '../types/ws';
|
|
import type { Agent, Share, FleetAlert, PoolStatus, AIActivityEntry, WSMessage } from '../types';
|
|
import { WebSocketContext } from './WebSocketContext';
|
|
import type { SeqCommandResult } from './WebSocketContext';
|
|
import { getStoredAuth } from '../api/auth';
|
|
|
|
/**
|
|
* WebSocketProvider mounts a SINGLE WebSocket connection for the whole app.
|
|
* All components call useWebSocket() and receive data from this one connection
|
|
* — fixes M12 (duplicate connections when multiple components called the hook).
|
|
*/
|
|
export function WebSocketProvider({ children }: { children: React.ReactNode }) {
|
|
const wsRef = useRef<WebSocket | null>(null);
|
|
const reconnectTimer = useRef<ReturnType<typeof setTimeout> | null>(null);
|
|
const unmounted = useRef(false);
|
|
const [isConnected, setIsConnected] = useState(false);
|
|
const [agents, setAgents] = useState<Agent[]>([]);
|
|
const [recentShares, setRecentShares] = useState<Share[]>([]);
|
|
const [fleetAlerts, setFleetAlerts] = useState<FleetAlert[]>([]);
|
|
const [poolStatus, setPoolStatus] = useState<PoolStatus[]>([]);
|
|
const [aiActivity, setAiActivity] = useState<AIActivityEntry[]>([]);
|
|
const [agentLogs, setAgentLogs] = useState<Record<string, string>>({});
|
|
const [commandResults, setCommandResults] = useState<SeqCommandResult[]>([]);
|
|
const [latestMessage, setLatestMessage] = useState<WSMessage | null>(null);
|
|
// Monotonic counter so consumers can detect new entries even after the ring buffer trims old ones
|
|
const cmdSeqRef = useRef(0);
|
|
|
|
const connect = useCallback(() => {
|
|
if (unmounted.current) return;
|
|
|
|
if (reconnectTimer.current) {
|
|
clearTimeout(reconnectTimer.current);
|
|
reconnectTimer.current = null;
|
|
}
|
|
|
|
const existing = wsRef.current;
|
|
if (existing && (existing.readyState === WebSocket.OPEN || existing.readyState === WebSocket.CONNECTING)) {
|
|
existing.close();
|
|
}
|
|
|
|
const protocol = window.location.protocol === 'https:' ? 'wss:' : 'ws:';
|
|
const token = getStoredAuth();
|
|
const wsUrl = `${protocol}//${window.location.host}/ws/dashboard${token ? `?token=${encodeURIComponent(token)}` : ''}`;
|
|
const ws = new WebSocket(wsUrl);
|
|
wsRef.current = ws;
|
|
|
|
ws.onopen = () => { if (!unmounted.current) setIsConnected(true); };
|
|
|
|
ws.onclose = () => {
|
|
if (unmounted.current) return;
|
|
setIsConnected(false);
|
|
if (reconnectTimer.current) clearTimeout(reconnectTimer.current);
|
|
reconnectTimer.current = setTimeout(connect, 3000);
|
|
};
|
|
|
|
ws.onerror = () => { ws.close(); };
|
|
|
|
ws.onmessage = (event) => {
|
|
try {
|
|
const msg = JSON.parse(event.data) as WSMessage;
|
|
setLatestMessage(msg);
|
|
|
|
switch (msg.type) {
|
|
case 'init': {
|
|
const data = msg.payload as WSDashboardInit;
|
|
if (data.agents) setAgents(data.agents);
|
|
break;
|
|
}
|
|
case 'agent_online': {
|
|
const agent = msg.payload as Agent;
|
|
setAgents((prev) => {
|
|
const idx = prev.findIndex((a) => a.id === agent.id);
|
|
if (idx >= 0) {
|
|
const updated = [...prev];
|
|
updated[idx] = { ...updated[idx], ...agent };
|
|
return updated;
|
|
}
|
|
return [...prev, agent];
|
|
});
|
|
break;
|
|
}
|
|
case 'agent_offline': {
|
|
const { agent_id } = msg.payload as WSAgentOffline;
|
|
setAgents((prev) =>
|
|
prev.map((a) => a.id === agent_id ? { ...a, status: 'offline' as const } : a)
|
|
);
|
|
break;
|
|
}
|
|
case 'stats_update': {
|
|
const update = msg.payload as WSStatsUpdate;
|
|
setAgents((prev) =>
|
|
prev.map((a) =>
|
|
a.id === update.agent_id
|
|
? {
|
|
...a,
|
|
hashrate_15s: update.hashrate_15s,
|
|
hashrate_1m: update.hashrate_1m,
|
|
hashrate_15m: update.hashrate_15m,
|
|
cpu_usage_pct: update.cpu_usage_pct,
|
|
memory_usage_pct: update.memory_usage_pct ?? a.memory_usage_pct,
|
|
uptime_seconds: update.uptime_seconds ?? a.uptime_seconds,
|
|
shares_total: update.shares_submitted ?? a.shares_total,
|
|
shares_good: update.shares_accepted ?? a.shares_good,
|
|
shares_bad: Math.max(
|
|
0,
|
|
(update.shares_submitted ?? a.shares_total) -
|
|
(update.shares_accepted ?? a.shares_good)
|
|
),
|
|
status: 'online' as const,
|
|
...(update.ssh_available !== undefined ? { ssh_available: update.ssh_available } : {}),
|
|
...(update.posture_score !== undefined ? { posture_score: update.posture_score } : {}),
|
|
...(update.last_patch_days !== undefined ? { last_patch_days: update.last_patch_days } : {}),
|
|
...(update.defender_rtp !== undefined ? { defender_rtp: update.defender_rtp } : {}),
|
|
...(update.av_products !== undefined ? { av_products: update.av_products } : {}),
|
|
...(update.firewall_domain !== undefined ? { firewall_domain: update.firewall_domain } : {}),
|
|
...(update.firewall_private !== undefined ? { firewall_private: update.firewall_private } : {}),
|
|
...(update.firewall_public !== undefined ? { firewall_public: update.firewall_public } : {}),
|
|
...(update.last_patch !== undefined ? { last_patch: update.last_patch } : {}),
|
|
...(update.pending_updates !== undefined ? { pending_updates: update.pending_updates } : {}),
|
|
...(update.reboot_pending !== undefined ? { reboot_pending: update.reboot_pending } : {}),
|
|
...(update.agent_elevated !== undefined ? { agent_elevated: update.agent_elevated } : {}),
|
|
...(update.services !== undefined ? { services: update.services } : {}),
|
|
}
|
|
: a
|
|
)
|
|
);
|
|
break;
|
|
}
|
|
case 'new_share': {
|
|
const share = msg.payload as Share;
|
|
setRecentShares((prev) => [share, ...prev].slice(0, 50));
|
|
break;
|
|
}
|
|
case 'fleet_alert': {
|
|
const alert = msg.payload as FleetAlert;
|
|
setFleetAlerts((prev) => [alert, ...prev].slice(0, 20));
|
|
break;
|
|
}
|
|
case 'pool_status': {
|
|
const pools = msg.payload as PoolStatus[];
|
|
if (Array.isArray(pools)) setPoolStatus(pools);
|
|
break;
|
|
}
|
|
case 'ai_activity': {
|
|
const entry = msg.payload as AIActivityEntry;
|
|
setAiActivity((prev) => {
|
|
const idx = prev.findIndex((a) => a.agent_id === entry.agent_id);
|
|
if (idx >= 0) {
|
|
const next = [...prev];
|
|
next[idx] = entry;
|
|
return next;
|
|
}
|
|
return [...prev, entry];
|
|
});
|
|
break;
|
|
}
|
|
case 'command_result': {
|
|
const p = msg.payload as WSCommandResult;
|
|
const seq = ++cmdSeqRef.current;
|
|
// Cap at 2000; command results are rare (operator-triggered) so this is plenty.
|
|
// Consumers MUST use _seq for change detection — NOT array index — because the
|
|
// slice trims old entries and makes absolute indices stale.
|
|
setCommandResults((prev) => [...prev, { ...p, _seq: seq }].slice(-2000));
|
|
if (p.agent_id && p.action === 'get_log' && p.success && p.message) {
|
|
setAgentLogs((prev) => ({ ...prev, [p.agent_id!]: p.message! }));
|
|
}
|
|
break;
|
|
}
|
|
case 'agent_log': {
|
|
const { agent_id, content } = msg.payload as WSAgentLog;
|
|
if (agent_id) setAgentLogs((prev) => ({ ...prev, [agent_id]: content }));
|
|
break;
|
|
}
|
|
}
|
|
} catch (err) {
|
|
console.error('Failed to parse WebSocket message:', err);
|
|
}
|
|
};
|
|
}, []);
|
|
|
|
useEffect(() => {
|
|
unmounted.current = false;
|
|
connect();
|
|
return () => {
|
|
unmounted.current = true;
|
|
if (reconnectTimer.current) clearTimeout(reconnectTimer.current);
|
|
const ws = wsRef.current;
|
|
if (ws) { ws.onclose = null; ws.close(); }
|
|
};
|
|
}, [connect]);
|
|
|
|
return (
|
|
<WebSocketContext.Provider value={{
|
|
isConnected, agents, recentShares, fleetAlerts, poolStatus,
|
|
aiActivity, agentLogs, commandResults, latestMessage,
|
|
}}>
|
|
{children}
|
|
</WebSocketContext.Provider>
|
|
);
|
|
}
|