racing-data/src/server.ts

import dgram from "node:dgram";
import { createReadStream, existsSync } from "node:fs";
import { createServer, IncomingMessage, ServerResponse } from "node:http";
import { extname, join, normalize } from "node:path";
import { getLeaderboard, updateLeaderboard } from "./leaderboard.js";
import { parsePacket, UDP_PORT, type ParsedPacket } from "./parser.js";

const HTTP_PORT = Number(process.env.HTTP_PORT ?? 3000);
const publicDir = join(process.cwd(), "public");
const docsDir = join(process.cwd(), "docs");
const sseClients = new Set<ServerResponse>();
const latestByType = new Map<string, ParsedPacket>();
const recentPackets: ParsedPacket[] = [];

const contentTypes: Record<string, string> = {
  ".html": "text/html; charset=utf-8",
  ".css": "text/css; charset=utf-8",
  ".js": "text/javascript; charset=utf-8",
  ".json": "application/json; charset=utf-8"
};

function sendEvent(response: ServerResponse, event: string, payload: unknown) {
  response.write(`event: ${event}\n`);
  response.write(`data: ${JSON.stringify(payload)}\n\n`);
}

function broadcast(packet: ParsedPacket) {
  const state = {
    latest: Object.fromEntries(latestByType),
    recent: recentPackets
  };
  const leaderboard = getLeaderboard(latestByType, recentPackets);

  for (const client of sseClients) {
    sendEvent(client, "packet", { packet, state, leaderboard });
    sendEvent(client, "leaderboard", leaderboard);
  }
}

function processPacket(packet: ParsedPacket) {
  const mergedPacket = mergeLatestPacket(packet);
  latestByType.set(String(mergedPacket.base.packetType), mergedPacket);
  recentPackets.unshift(packet);
  recentPackets.splice(50);
  updateLeaderboard(latestByType);
  broadcast(mergedPacket);
}

function readRequestBody(request: IncomingMessage): Promise<string> {
  return new Promise((resolve, reject) => {
    const chunks: Buffer[] = [];
    request.on("data", (chunk) => chunks.push(Buffer.from(chunk)));
    request.on("end", () => resolve(Buffer.concat(chunks).toString("utf8")));
    request.on("error", reject);
  });
}

function isParsedPacket(value: unknown): value is ParsedPacket {
  if (!isRecord(value) || !isRecord(value.base) || !isRecord(value.data)) {
    return false;
  }

  return typeof value.base.packetType === "number";
}

function isRecord(value: unknown): value is Record<string, unknown> {
  return Boolean(value) && typeof value === "object" && !Array.isArray(value);
}

function records(value: unknown): Record<string, unknown>[] {
  return Array.isArray(value) ? value.filter(isRecord) : [];
}

function mergeArrayByKey(
  previous: unknown,
  next: unknown,
  key: string,
  fallbackKey = "slot"
): Record<string, unknown>[] {
  const merged = new Map<string, Record<string, unknown>>();

  for (const item of [...records(previous), ...records(next)]) {
    const keyValue = item[key] ?? item[fallbackKey];
    if (keyValue === undefined || keyValue === null) continue;
    merged.set(String(keyValue), item);
  }

  return Array.from(merged.values());
}

function mergeLatestPacket(packet: ParsedPacket): ParsedPacket {
  const existing = latestByType.get(String(packet.base.packetType));
  if (!existing) {
    return packet;
  }

  if (packet.base.packetType === 2) {
    return {
      ...packet,
      data: {
        ...existing.data,
        ...packet.data,
        participants: mergeArrayByKey(
          existing.data.participants,
          packet.data.participants,
          "slot"
        )
      }
    };
  }

  if (packet.base.packetType === 8) {
    return {
      ...packet,
      data: {
        ...existing.data,
        ...packet.data,
        vehicles: mergeArrayByKey(existing.data.vehicles, packet.data.vehicles, "index"),
        classes: mergeArrayByKey(existing.data.classes, packet.data.classes, "classIndex")
      }
    };
  }

  return packet;
}

function serveStatic(pathname: string, response: ServerResponse) {
  const requestedPath = pathname === "/" ? "/index.html" : pathname;
  const filePath = normalize(join(publicDir, requestedPath));

  if (!filePath.startsWith(publicDir) || !existsSync(filePath)) {
    response.writeHead(404, { "content-type": "text/plain; charset=utf-8" });
    response.end("Not found");
    return;
  }

  response.writeHead(200, {
    "content-type": contentTypes[extname(filePath)] ?? "application/octet-stream"
  });
  createReadStream(filePath).pipe(response);
}

function serveFile(filePath: string, response: ServerResponse, contentType?: string) {
  if (!existsSync(filePath)) {
    response.writeHead(404, { "content-type": "text/plain; charset=utf-8" });
    response.end("Not found");
    return;
  }

  response.writeHead(200, {
    "content-type": contentType ?? contentTypes[extname(filePath)] ?? "application/octet-stream"
  });
  createReadStream(filePath).pipe(response);
}

const httpServer = createServer((request, response) => {
  const url = new URL(request.url ?? "/", `http://${request.headers.host}`);

  if (request.method === "OPTIONS") {
    response.writeHead(204, {
      "access-control-allow-origin": "*",
      "access-control-allow-methods": "GET,POST,OPTIONS",
      "access-control-allow-headers": "content-type"
    });
    response.end();
    return;
  }

  if (url.pathname === "/events") {
    response.writeHead(200, {
      "content-type": "text/event-stream",
      "cache-control": "no-cache",
      connection: "keep-alive",
      "access-control-allow-origin": "*"
    });
    response.write("\n");
    sseClients.add(response);
    sendEvent(response, "state", {
      latest: Object.fromEntries(latestByType),
      recent: recentPackets,
      leaderboard: getLeaderboard(latestByType, recentPackets)
    });
    request.on("close", () => sseClients.delete(response));
    return;
  }

  if (url.pathname === "/api/state") {
    response.writeHead(200, { "content-type": "application/json; charset=utf-8" });
    response.end(JSON.stringify({
      latest: Object.fromEntries(latestByType),
      recent: recentPackets
    }));
    return;
  }

  if (url.pathname === "/api/leaderboard") {
    response.writeHead(200, { "content-type": "application/json; charset=utf-8" });
    response.end(JSON.stringify(getLeaderboard(latestByType, recentPackets)));
    return;
  }

  if (url.pathname === "/api/ingest/shared-memory" && request.method === "POST") {
    void readRequestBody(request)
      .then((body) => {
        const payload = JSON.parse(body) as unknown;
        const incomingPackets = isRecord(payload) && Array.isArray(payload.packets)
          ? payload.packets.filter(isParsedPacket)
          : [];

        if (!incomingPackets.length) {
          response.writeHead(400, {
            "content-type": "application/json; charset=utf-8",
            "access-control-allow-origin": "*"
          });
          response.end(JSON.stringify({ ok: false, error: "Expected a packets array." }));
          return;
        }

        for (const packet of incomingPackets) {
          processPacket({
            ...packet,
            receivedAt: packet.receivedAt ?? new Date().toISOString(),
            source: packet.source || "shared-memory"
          });
        }

        response.writeHead(200, {
          "content-type": "application/json; charset=utf-8",
          "access-control-allow-origin": "*"
        });
        response.end(JSON.stringify({ ok: true, packets: incomingPackets.length }));
      })
      .catch((error) => {
        response.writeHead(400, {
          "content-type": "application/json; charset=utf-8",
          "access-control-allow-origin": "*"
        });
        response.end(JSON.stringify({
          ok: false,
          error: error instanceof Error ? error.message : String(error)
        }));
      });
    return;
  }

  if (url.pathname === "/api/openapi.json") {
    serveFile(join(publicDir, "openapi.json"), response, "application/json; charset=utf-8");
    return;
  }

  if (url.pathname === "/schemas/udp-packets.schema.json") {
    serveFile(
      join(docsDir, "udp-packets.schema.json"),
      response,
      "application/schema+json; charset=utf-8"
    );
    return;
  }

  serveStatic(url.pathname, response);
});

const udpServer = dgram.createSocket("udp4");

udpServer.on("message", (message, remote) => {
  try {
    const packet = parsePacket(message, `${remote.address}:${remote.port}`);
    processPacket(packet);
  } catch (error) {
    console.error("Failed to parse UDP packet:", error);
  }
});

udpServer.on("listening", () => {
  const address = udpServer.address();
  console.log(`UDP listener ready on ${address.address}:${address.port}`);
});

udpServer.bind(UDP_PORT, "0.0.0.0");

httpServer.listen(HTTP_PORT, () => {
  console.log(`Frontend ready at http://localhost:${HTTP_PORT}`);
});