server/index.ts

import path from 'node:path';
import {fileURLToPath} from 'node:url';
import cors from 'cors';
import express from 'express';
import {
  CLUSTER_MODELS,
  DEFAULT_CLUSTER_MODEL_ID,
  isValidClusterModelId,
} from '../shared/clusterModels.js';
import {ApiTaskHandler} from './ApiTaskHandler.js';
import {validateHostRegistered} from './hostValidation.js';
import {initSse, writeSse, writeSseComment} from './sse.js';
import {store} from './store.js';
import type {RegisterHostBody} from './types.js';

const PORT = Number(process.env.PORT ?? 8787);
const TASK_TIMEOUT_MS = Number(process.env.TASK_TIMEOUT_MS ?? 120_000);
const SERVE_STATIC = process.env.SERVE_STATIC === '1';
const staticDir = path.resolve(
  path.dirname(fileURLToPath(import.meta.url)),
  '../docs',
);

const app = express();

app.use((_req, res, next) => {
  res.setHeader('Cross-Origin-Opener-Policy', 'same-origin');
  res.setHeader('Cross-Origin-Embedder-Policy', 'require-corp');
  next();
});

app.use(cors());
app.use(express.json({limit: '32mb'}));

app.get('/health', (_req, res) => {
  res.json({status: 'ok', api_version: 2});
});

app.get('/v1/hosts', (_req, res) => {
  res.json({hosts: store.listHosts()});
});

app.get('/v1/models', (_req, res) => {
  res.json({
    models: CLUSTER_MODELS.map((model) => ({
      id: model.id,
      label: model.label,
      description: model.description,
      implemented: model.implemented,
    })),
  });
});

app.post('/api/hosts/register', (req, res) => {
  const body = req.body as RegisterHostBody;
  if (!body?.id?.trim()) {
    res.status(400).json({error: 'id is required'});
    return;
  }

  const modelId = body.model?.trim() || DEFAULT_CLUSTER_MODEL_ID;
  if (!isValidClusterModelId(modelId)) {
    res.status(400).json({
      error: `Unknown model '${modelId}'. Supported: ${CLUSTER_MODELS.map((m) => m.id).join(', ')}`,
    });
    return;
  }

  const host = store.registerHost(body.id.trim(), modelId);
  res.json({
    id: host.id,
    model: host.model,
    online: store.isHostOnline(host.id),
  });
});

app.get('/api/hosts/stream', (req, res) => {
  const hostId = String(req.query.host_id ?? '').trim();
  if (!hostId) {
    res.status(400).json({error: 'host_id query parameter is required'});
    return;
  }

  const hostError = validateHostRegistered(store, hostId);
  if (hostError) {
    res.status(hostError.status).json({error: hostError.message});
    return;
  }

  initSse(res);

  const heartbeat = setInterval(() => {
    store.heartbeat(hostId);
    writeSseComment(res, 'ping');
  }, 15_000);

  store.attachHostStream(hostId, {
    write: (event, data) => writeSse(res, event, data),
    close: () => {
      clearInterval(heartbeat);
      if (!res.writableEnded) {
        res.end();
      }
    },
  });

  writeSse(res, 'ready', {host_id: hostId});

  req.on('close', () => {
    clearInterval(heartbeat);
    store.detachHostStream(hostId);
  });
});

new ApiTaskHandler(store, TASK_TIMEOUT_MS).mount(app);

if (SERVE_STATIC) {
  app.use(express.static(staticDir));
  app.get('/monitor.html', (_req, res) => {
    res.sendFile(path.join(staticDir, 'monitor.html'));
  });
  app.get('/', (_req, res) => {
    res.sendFile(path.join(staticDir, 'index.html'));
  });
}

app.listen(PORT, '0.0.0.0', () => {
  console.log(`Detection broker listening on http://0.0.0.0:${PORT}`);
  if (SERVE_STATIC) {
    console.log(`  Static UI: / and /monitor.html`);
  }
  console.log(`  POST /v1/detect`);
  console.log(`  POST /v1/describe`);
  console.log(`  GET  /v1/hosts`);
  console.log(`  GET  /v1/models`);
  console.log(`  GET  /api/hosts/stream?host_id=...`);
});