platform/src/server.tsx

import './servers/bootstrap';
import type { ServerWebSocket } from 'bun';
import { serve } from 'bun';
import { honoServer, PROTECTED_API_PREFIXES, UNPROTECTED_API_PREFIXES } from './servers/hono';
import { assertCapabilityTotality } from './servers/capabilities/totality';
import { verify } from './servers/jwt';
import { isWsProviderAllowed } from './servers/capabilities/authorize';
import { isTokenBlacklisted } from 'officerdb';
import { terminalWebsocket } from './servers/api/terminal/websocket';
import { chatWebsocket } from './servers/api/chat/websocket';
import { taskRunnerWebsocket } from './servers/api/tasks/task-executor';
import { pipelineWebsocket } from './servers/api/tasks/pipeline-executor';
import { cliampWebsocket, cliampAudioWebsocket } from './servers/api/cliamp/relay';
import { desktopWebsocket } from './servers/api/desktop/websocket';
import { vaultWebsocket, upgradeVaultWs } from './servers/api/vault/websocket';
import officerWeb from './apps/officer-web/index.gen.html';
import { startBrowserRelay } from './servers/api/browser/relay';
import { registerSidecar, unregisterSidecar, handleSidecarMessage } from './servers/sidecar-registry';
import './servers/api/chat/opencode/sidecar-server'; // subscribe to the opencode sidecar's port report
import type { SidecarRegistration } from './servers/sidecar/registration-protocol';
import { toShellUsername } from './servers/data-path';

const { PORT = '5000' } = process.env;

// Build static file routes from public/
const publicRoutes: Record<string, (req: Request) => Response> = {};
for await (const file of new Bun.Glob('**').scan({ cwd: './public' })) {
  const bunFile = Bun.file(`./public/${file}`);
  publicRoutes[`/${file}`] = () => new Response(bunFile);
}

type WSData = {
  userId: number;
  email: string;
  username: string;
  provider:
    | 'terminal'
    | 'chat'
    | 'task-runner'
    | 'pipeline'
    | 'cliamp'
    | 'cliamp-audio'
    | 'desktop'
    | 'vault'
    | 'sidecar';
  sessionId?: string;
  cwd?: string;
  command?: string;
  cols?: number;
  rows?: number;
  search?: string; // raw query string, for providers that relay it to a sidecar
};

// Sidecar registration WebSocket handler
const sidecarConnections = new Map<ServerWebSocket<WSData>, string>(); // ws → sidecar ID

const sidecarWebsocket = {
  open(_ws: ServerWebSocket<WSData>) {
    // Wait for registration message
  },
  message(ws: ServerWebSocket<WSData>, raw: string | Buffer) {
    try {
      const data = typeof raw === 'string' ? raw : raw.toString();
      const msg = JSON.parse(data);

      if (msg.type === 'register') {
        const id = registerSidecar(ws, msg as SidecarRegistration);
        sidecarConnections.set(ws, id);
        ws.send(JSON.stringify({ type: 'registered', id }));
        return;
      }

      // Handle queue commands from sidecars (e.g., email sidecar enqueuing jobs)
      if (typeof msg.type === 'string' && msg.type.startsWith('queue:') && msg.id) {
        handleSidecarQueueCommand(ws, msg);
        return;
      }

      const id = sidecarConnections.get(ws);
      if (id) {
        handleSidecarMessage(id, msg);
      }
    } catch {
      // skip malformed messages
    }
  },
  close(ws: ServerWebSocket<WSData>) {
    const id = sidecarConnections.get(ws);
    if (id) {
      unregisterSidecar(id);
      sidecarConnections.delete(ws);
    }
  },
  drain() {},
};

// Handle queue commands from sidecars (e.g., email sidecar enqueuing jobs)
async function handleSidecarQueueCommand(ws: ServerWebSocket<WSData>, msg: Record<string, unknown>) {
  const id = msg.id as string;
  try {
    switch (msg.type) {
      case 'queue:enqueue': {
        const job = await queueEnqueue(msg.params as import('./servers/queue/types').EnqueueParams);
        ws.send(JSON.stringify({ type: 'queue:enqueued', id, job }));
        break;
      }
      case 'queue:cancel': {
        const job = await queueCancel(msg.jobId as string);
        ws.send(JSON.stringify({ type: 'queue:cancelled', id, job }));
        break;
      }
      case 'queue:list': {
        const jobs = await queueList();
        ws.send(JSON.stringify({ type: 'queue:list', id, jobs }));
        break;
      }
      case 'queue:get': {
        const job = await queueGet(msg.jobId as string);
        ws.send(JSON.stringify({ type: 'queue:get', id, job }));
        break;
      }
      default:
        ws.send(JSON.stringify({ type: 'queue:error', id, error: `Unknown queue command: ${msg.type}` }));
    }
  } catch (err) {
    ws.send(JSON.stringify({ type: 'queue:error', id, error: err instanceof Error ? err.message : String(err) }));
  }
}

const handlers: Record<string, any> = {
  terminal: terminalWebsocket,
  chat: chatWebsocket,
  'task-runner': taskRunnerWebsocket,
  pipeline: pipelineWebsocket,
  cliamp: cliampWebsocket,
  'cliamp-audio': cliampAudioWebsocket,
  desktop: desktopWebsocket,
  vault: vaultWebsocket,
  sidecar: sidecarWebsocket,
};

// Both doors into the platform, checked against the capability registry before either opens.
//
// This throws rather than warns, and it throws HERE — before serve() — so a surface nobody has gated
// cannot be reached even once. The HTTP half comes from hono.ts's own mount table and the socket half
// from the map directly above, so neither list can be a stale copy of the thing it describes.
assertCapabilityTotality({
  apiPrefixes: [...PROTECTED_API_PREFIXES, ...UNPROTECTED_API_PREFIXES],
  wsProviders: Object.keys(handlers),
});

async function upgradeWs(
  req: Request,
  server: any,
  provider: 'terminal' | 'chat' | 'task-runner' | 'pipeline' | 'cliamp' | 'cliamp-audio' | 'desktop',
) {
  const token = new URL(req.url).searchParams.get('token');
  if (!token) return new Response('Unauthorized', { status: 401 });

  try {
    const user = await verify(token);
    if (!user) return new Response('Unauthorized', { status: 401 });

    if (user.jti) {
      if (await isTokenBlacklisted(user.jti)) return new Response('Unauthorized', { status: 401 });
    }

    // The capability backstop, applied to sockets. Everything above this line AUTHENTICATES — it proves
    // who is calling and never asks what they may reach. That is why a Member with a valid token could
    // open a terminal here in the same minute it was 403'd on GET /api/tasks.
    //
    // This resolves against the same registry as the HTTP door rather than a parallel list, which is the
    // whole point: the two doors cannot disagree about what a role holds, because there is only one
    // declaration to read. `terminal`, `chat`, `task-runner`, `pipeline` and `desktop` are refused here
    // by being `execution` capabilities, not by being absent from an array someone has to maintain.
    if (!(await isWsProviderAllowed(user.id, provider))) {
      return new Response('Forbidden', { status: 403 });
    }

    const url = new URL(req.url);
    const sessionId = url.searchParams.get('sessionId') ?? undefined;
    const cwd = url.searchParams.get('cwd') ?? undefined;
    const command = url.searchParams.get('command') ?? undefined;
    const cols = url.searchParams.get('cols') ? Number(url.searchParams.get('cols')) : undefined;
    const rows = url.searchParams.get('rows') ? Number(url.searchParams.get('rows')) : undefined;
    const ok = server.upgrade(req, {
      data: {
        userId: user.id,
        email: user.email,
        username: toShellUsername(user.username ?? '', user.email),
        provider,
        sessionId,
        cwd,
        command,
        cols,
        rows,
        search: url.search,
      },
    });
    if (!ok) return new Response('Upgrade failed', { status: 500 });
  } catch {
    return new Response('Unauthorized', { status: 401 });
  }
}

const server = serve({
  port: Number(PORT),
  idleTimeout: 60,
  maxRequestBodySize: 1024 * 1024 * 1024 * 50, // 50 GB
  routes: {
    ...publicRoutes,
    '/novnc/*': (req) => {
      const file = Bun.file(`public${new URL(req.url).pathname}`);
      return new Response(file);
    },
    '/vendor/*': (req) => {
      const file = Bun.file(`public${new URL(req.url).pathname}`);
      return new Response(file);
    },
    // Vaultwarden notifications hub: upgrade the WebSocket here (proxied to upstream by vaultWebsocket);
    // everything else on this path (SignalR long-poll negotiate/poll) falls through to the HTTP proxy.
    '/api/vault/notifications/*': (req, server) => {
      if (req.headers.get('upgrade') === 'websocket') return upgradeVaultWs(req, server);
      return honoServer.fetch(req, server);
    },
    '/api/sidecar/register': (req: Request, server: any) => {
      const ok = server.upgrade(req, {
        data: { provider: 'sidecar', userId: 0, email: '', username: '' },
      });
      if (!ok) return new Response('Upgrade failed', { status: 500 });
    },
    '/api/tasks/run/ws': (req, server) => upgradeWs(req, server, 'task-runner'),
    '/api/tasks/pipeline/ws': (req, server) => upgradeWs(req, server, 'pipeline'),
    '/api/terminal/ws': (req, server) => upgradeWs(req, server, 'terminal'),
    '/api/chat/ws': (req, server) => upgradeWs(req, server, 'chat'),
    '/api/cliamp/ws': (req, server) => upgradeWs(req, server, 'cliamp'),
    '/api/cliamp/audio/ws': (req, server) => upgradeWs(req, server, 'cliamp-audio'),
    '/api/desktop/ws': (req, server) => upgradeWs(req, server, 'desktop'),
    // CalDAV/CardDAV. These live OUTSIDE /api because DAV clients are given a bare domain and probe
    // fixed, spec-defined paths — `/.well-known/caldav` unauthenticated, before they hold any
    // credential at all. They need naming explicitly here or the `/*` SPA fallback below swallows them
    // and the phone gets an HTML page where it expected a redirect.
    '/.well-known/caldav': honoServer.fetch,
    '/.well-known/carddav': honoServer.fetch,
    '/dav': honoServer.fetch,
    '/dav/*': honoServer.fetch,
    '/': officerWeb,
    '/*': officerWeb,
    '/api': honoServer.fetch,
    '/api/*': honoServer.fetch,
  },

  websocket: {
    open(ws) {
      const { provider } = (ws as unknown as ServerWebSocket<WSData>).data;
      handlers[provider]!.open(ws as any);
    },
    message(ws, raw) {
      const { provider } = (ws as unknown as ServerWebSocket<WSData>).data;
      handlers[provider]!.message(ws as any, raw);
    },
    close(ws) {
      const { provider } = (ws as unknown as ServerWebSocket<WSData>).data;
      handlers[provider]!.close(ws as any);
    },
    drain() {},
  },

  development: process.env.NODE_ENV !== 'production' && {
    hmr: true,
  },
});

console.log(`🚀 Server running at ${server.url}`);

const BROWSER_RELAY_PORT = Number(process.env.BROWSER_RELAY_PORT ?? '18792');
try {
  await startBrowserRelay(BROWSER_RELAY_PORT);
  console.log(`[browser-relay] listening on port ${BROWSER_RELAY_PORT}`);
} catch (err) {
  console.error('[browser-relay] failed to start:', err instanceof Error ? err.message : err);
}

// Initialize queue engine in API server process
import {
  initQueue,
  enqueueJob as queueEnqueue,
  cancelJob as queueCancel,
  listAllJobs as queueList,
  readJob as queueGet,
} from './servers/queue/init';
initQueue().catch((err) => console.error('[queue] failed to initialize:', err));

// Mark any orphaned pipeline jobs from previous server run
import { cleanupOnStartup } from './servers/api/tasks/pipeline-job-manager';
cleanupOnStartup().catch((err) => console.error('[pipeline-jobs] startup cleanup failed:', err));

// PulseAudio and the `virtual_out` sink used to be set up here, at every boot of a process that has no
// audio responsibilities. They belong to the music sidecar, which owns both cliamp halves now
// (sidecar/music/pulse-audio.ts).

// Pi check/install is handled by bootstrap.ts (imported above)