import { createServer } from 'node:http'; import { createHash } from 'node:crypto'; import { pathToFileURL } from 'node:url'; import { WebSocketServer } from 'ws'; function selectLatestSourceJobId(product) { const versions = (product?.versions ?? []) .filter((version) => Array.isArray(version?.wordcloud_archives)) .sort((left, right) => { const leftTime = Date.parse(left?.created_at ?? ''); const rightTime = Date.parse(right?.created_at ?? ''); return (Number.isFinite(rightTime) ? rightTime : 0) - (Number.isFinite(leftTime) ? leftTime : 0); }); for (const version of versions) { const archive = version.wordcloud_archives.find((item) => ( typeof item?.source_job_id === 'string' && item.source_job_id.length > 0 )); if (archive) { return archive.source_job_id; } } return null; } function isFocusEvent(value) { return ( typeof value === 'object' && value !== null && value.type === 'FOCUS_NAME' && typeof value.cloudId === 'string' && typeof value.name === 'string' && typeof value.messageId === 'string' && typeof value.sentAt === 'number' && (value.targetIndex === undefined || (typeof value.targetIndex === 'number' && Number.isInteger(value.targetIndex) && value.targetIndex >= 0)) ); } export function createRealtimeServer( port = Number(process.env.REALTIME_PORT || 8787), options = {}, ) { const upstreamUrl = options.upstreamUrl ?? process.env.WORDCLOUD_API_UPSTREAM ?? 'http://192.168.31.213:8000'; const adminPassword = options.adminPassword ?? process.env.ORDERS_ADMIN_PASSWORD ?? 'zhihui2024'; const tokenSecret = options.tokenSecret ?? process.env.ORDERS_TOKEN_SECRET ?? 'wordcloud-orders-demo'; const fetchImpl = options.fetchImpl ?? fetch; const upstream = upstreamUrl.replace(/\/+$/, ''); const server = createServer((request, response) => { if (request.url === '/healthz') { response.writeHead(200, { 'content-type': 'application/json' }); response.end(JSON.stringify({ status: 'ok' })); return; } const sourceJobMatch = request.url.match(/^\/api\/products\/([^/]+)\/source-job$/); if (sourceJobMatch && request.method === 'GET') { const productId = decodeURIComponent(sourceJobMatch[1]); const sendError = (status, message) => { response.writeHead(status, { 'content-type': 'application/json' }); response.end(JSON.stringify({ detail: message })); }; fetchImpl(`${upstream}/api/login`, { method: 'POST', headers: { 'content-type': 'application/json' }, body: JSON.stringify({ password: adminPassword }), }) .then(async (loginResponse) => { if (!loginResponse.ok) { return sendError(502, '词云产品服务认证失败'); } const { token } = await loginResponse.json(); return fetchImpl(`${upstream}/api/products/${encodeURIComponent(productId)}`, { headers: { Authorization: `Bearer ${token}` }, }); }) .then(async (productResponse) => { if (!productResponse.ok) { return sendError( productResponse.status === 404 ? 404 : 502, productResponse.status === 404 ? 'product not found' : '词云产品服务不可用', ); } const sourceJobId = selectLatestSourceJobId(await productResponse.json()); if (!sourceJobId) { return sendError(404, 'product has no wordcloud archive'); } response.writeHead(200, { 'content-type': 'application/json' }); response.end(JSON.stringify({ product_id: productId, source_job_id: sourceJobId })); }) .catch(() => sendError(502, '词云产品服务不可用')); return; } response.writeHead(404, { 'content-type': 'text/plain' }); response.end('Not found'); }); const realtime = new WebSocketServer({ server, maxPayload: 16 * 1024, }); const heartbeat = setInterval(() => { for (const socket of realtime.clients) { if (socket.isAlive === false) { socket.terminate(); continue; } socket.isAlive = false; socket.ping(); } }, 30_000); heartbeat.unref(); realtime.on('connection', (socket) => { socket.isAlive = true; socket.on('pong', () => { socket.isAlive = true; }); socket.on('message', (data) => { let value; try { value = JSON.parse(data.toString()); } catch { return; } if (!isFocusEvent(value)) { return; } const message = JSON.stringify(value); for (const client of realtime.clients) { if (client !== socket && client.readyState === client.OPEN) { client.send(message); } } }); }); realtime.on('close', () => { clearInterval(heartbeat); }); return server; } if (import.meta.url === pathToFileURL(process.argv[1] || '').href) { const port = Number(process.env.REALTIME_PORT || 8787); createRealtimeServer(port).listen(port, () => { console.log(`Realtime WebSocket server listening on port ${port}`); }); }