platform/src/servers/sidecar/notify/index.ts

import type { SidecarCommand, SidecarEvent } from '../protocol';
import { createSidecarConnector } from '../connect';
import { dispatch, configuredChannels } from './dispatch';
import { handleDeviceRoute } from './devices';
import { resolveNotifyUser } from './resolve-user';
import { closeApnsSessions } from './apns';
import type { Notification, NotifyType } from './types';

// The officer-notify sidecar. The one place anything leaves this machine to tell the owner something.
//
// It exists as a sidecar rather than as platform code because the producers are spread out — the queue,
// the email sidecar, the agent sidecar — and a platform-owned notifier would force every sidecar to call
// BACK into the platform. That is the inversion just removed from email. Here, anything POSTs over
// loopback and the direction of every arrow stays the same.
//
// ─────────────────────────────────────────────────────────────────────────────────────────────────
// HTTP CONTRACT — the platform strips its /api/notify mount prefix before forwarding.
//
//   GET  /_health                    liveness, plus which channels are configured
//   POST /_officer/notify            send one. Body: { type, userId, id?, count?, ok?, appSlug? }
//   POST /_officer/devices           register/refresh a device (idempotent; call every launch)
//   DELETE /_officer/devices/:token  deregister, on sign-out
//   GET  /_officer/devices           the caller's registered devices
//
// The notify body is deliberately narrow: a category and an id, never content. See ./text.ts for why
// the visible string is composed here rather than sent by the producer.
// ─────────────────────────────────────────────────────────────────────────────────────────────────

const API_URL = process.env.API_URL ?? `ws://127.0.0.1:${process.env.PORT ?? '5000'}`;

const VALID_TYPES: NotifyType[] = ['job', 'mail', 'agent', 'download', 'test'];

/** Grab an ephemeral free port by briefly binding one and releasing it. */
function getFreePort(): number {
  const probe = Bun.serve({ port: 0, hostname: '127.0.0.1', fetch: () => new Response('') });
  const p = probe.port;
  probe.stop(true);
  if (p == null) throw new Error('failed to acquire a free port');
  return p;
}

const port = getFreePort();

const server = Bun.serve({
  port,
  hostname: '127.0.0.1',
  async fetch(req) {
    const url = new URL(req.url);

    if (url.pathname === '/_health') {
      return Response.json({ ok: true, channels: configuredChannels() });
    }

    if (url.pathname === '/_officer/notify' && req.method === 'POST') {
      let body: Partial<Notification>;
      try {
        body = (await req.json()) as Partial<Notification>;
      } catch {
        return Response.json({ error: 'invalid json' }, { status: 400 });
      }

      if (!body.type || !VALID_TYPES.includes(body.type)) {
        return Response.json({ error: `type must be one of ${VALID_TYPES.join(', ')}` }, { status: 400 });
      }
      // The header wins over the body wherever it is present — see ./resolve-user.ts for why that
      // ordering is the whole access control on this route.
      const userId = resolveNotifyUser({ header: req.headers.get('X-Officer-User'), bodyUserId: body.userId });
      if (userId === null) {
        return Response.json({ error: 'userId is required (body or X-Officer-User)' }, { status: 400 });
      }

      const results = await dispatch({ ...body, userId } as Notification);
      return Response.json({ ok: true, results });
    }

    if (url.pathname.startsWith('/_officer/devices')) {
      try {
        const res = await handleDeviceRoute(req, url);
        return res ?? new Response('not found', { status: 404 });
      } catch (err) {
        console.error(`[notify] ${req.method} ${url.pathname} failed`, err);
        return Response.json({ error: 'internal error' }, { status: 500 });
      }
    }

    return new Response('not found', { status: 404 });
  },
});

console.log(
  `[notify] listening on http://127.0.0.1:${server.port} — channels: ${configuredChannels().join(', ') || 'none configured'}`,
);

// ── Connect to the API server ──

const connection = createSidecarConnector({
  apiUrl: `${API_URL}/api/sidecar/register`,
  name: 'notify',
  capabilities: ['notify'],
  onCommand(cmd, reply) {
    const c = cmd as SidecarCommand;
    if (c.type === 'ping') return reply({ type: 'pong', id: c.id } as SidecarEvent);
    reply({ type: 'error', id: (c as { id?: string }).id, error: `Unknown command type: ${c.type}` } as SidecarEvent);
  },
  onConnected() {
    // Re-announced on every reconnect: the platform forgets the port when the socket drops, and this
    // listener outlives an officer restart.
    connection.send({ type: 'notify:server', port: server.port } as SidecarEvent);
  },
});

function shutdown(signal: string) {
  console.log(`[notify] ${signal} received, shutting down...`);
  closeApnsSessions();
  connection.destroy();
  process.exit(0);
}
process.on('SIGINT', () => shutdown('SIGINT'));
process.on('SIGTERM', () => shutdown('SIGTERM'));