wordcloud-board/server/realtime-server.mjs

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}`);
  });
}