spb/agent-os

1/**2 * `/api/v1/events?cursor=<lastEventId>&missionId=<optional>`: the one WebSocket of the console.3 * Authentication (cookie + completed second factor) and the Origin check happen in the global4 * guard during the HTTP upgrade; query validation and mission ownership are checked here in5 * `preValidation`, so a refused connection is a plain HTTP error and never becomes a socket.6 * Server → browser only: one JSON `AosEvent` per text frame (see events-hub.ts).7 */8import type { FastifyInstance, FastifyRequest } from 'fastify';9import type WebSocket from 'ws';10import { z } from 'zod';11import { Id } from '@aos/shared';12import { notice } from '../events-hub.js';13import { authOf, notFound, parse, type Ctx } from '../http.js';14import { findSessionById } from '../security/sessions.js';1516const EventsQuery = z.object({17  cursor: z.string().regex(/^\d{1,15}$/).transform(Number).optional(),18  missionId: Id.optional(),19});2021/** Registers the socket and re-validates its auth session periodically (revoked/expired ⇒ 4401). */22export function watchAuth(ctx: Ctx, socket: WebSocket, req: FastifyRequest, beforeClose?: () => void): () => void {23  const a = authOf(req);24  const unregister = ctx.live.add(socket, a.sessionId, 'events', beforeClose);25  const timer = setInterval(() => {26    void findSessionById(ctx.db, a.sessionId).then(27      (row) => {28        if (!row || !row.second_factor) ctx.live.kickAuthSessions([a.sessionId]);29      },30      () => undefined, // database hiccup: keep the socket, retry on the next tick31    );32  }, ctx.config.wsAuthRecheckMs);33  timer.unref();34  const stop = (): void => {35    clearInterval(timer);36    unregister();37  };38  socket.once('close', stop);39  return stop;40}4142export default async function eventsWsPlugin(app: FastifyInstance, ctx: Ctx): Promise<void> {43  app.get('/events', {44    websocket: true,45    config: { doc: { summary: 'AosEvent stream (text JSON): replay after `cursor`, then live; transient model.delta has id -1', query: EventsQuery, ws: true } },46    preValidation: async (req) => {47      const q = parse(EventsQuery, req.query);48      if (q.missionId && !(await ctx.db.one('SELECT 1 AS x FROM missions WHERE id = $1 AND owner_id = $2', [q.missionId, authOf(req).userId]))) throw notFound('mission');49    },50  }, (socket, req) => {51    const a = authOf(req);52    const q = parse(EventsQuery, req.query);53    socket.on('message', () => undefined); // server → browser only54    socket.on('error', () => undefined);55    const detach = ctx.events.attach(socket, a.userId, { cursor: q.cursor ?? null, missionId: q.missionId ?? null });56    watchAuth(ctx, socket, req, () => {57      detach();58      ctx.events.sendTo(socket, notice('auth.expired', 'session ended'));59    });60    socket.once('close', detach);61  });62}63