platform/src/servers/sidecar-registry.ts

import { resolve } from 'node:path';
import type { ServerWebSocket } from 'bun';
import type { Subprocess } from 'bun';
import type {
  SidecarCommand,
  SidecarEvent,
  ClaudeState,
  ClaudeSpawnParams,
  ClaudeSpawnStreamingParams,
  ClaudeCodeResult,
  OpenCodeRunParams,
  PtyCommand,
  PtyEvent,
  VncStartParams,
  VncSessionInfo,
} from './sidecar/protocol';
import type { SidecarRegistration } from './sidecar/registration-protocol';
import type { ChatEvent } from './api/chat/types';

// ── Types ──

type RegisteredSidecar = {
  id: string;
  name: string;
  capabilities: string[];
  ws: ServerWebSocket<any>;
};

type PendingRequest = {
  resolve: (value: any) => void;
  reject: (error: Error) => void;
  timer: Timer;
};

type EventHandler = (event: SidecarEvent | PtyEvent) => void;

// ── State ──

const sidecars = new Map<string, RegisteredSidecar>();
const pending = new Map<string, PendingRequest>();
const eventHandlers = new Map<string, Set<EventHandler>>();
let cachedState: ClaudeState | null = null;
let idCounter = 0;

function nextId(): string {
  return `sr_${Date.now()}_${++idCounter}`;
}

// ── Registration ──

export function registerSidecar(ws: ServerWebSocket<any>, registration: SidecarRegistration): string {
  // If a sidecar with the same name is already registered, unregister it first
  for (const [id, sc] of sidecars) {
    if (sc.name === registration.name) {
      console.log(`[registry] replacing existing sidecar "${registration.name}" (id=${id})`);
      unregisterSidecar(id);
      break;
    }
  }

  const id = `sc_${registration.name}_${Date.now()}`;
  sidecars.set(id, {
    id,
    name: registration.name,
    capabilities: registration.capabilities,
    ws,
  });
  console.log(
    `[registry] registered sidecar "${registration.name}" (id=${id}, capabilities=[${registration.capabilities.join(', ')}])`,
  );
  return id;
}

export function unregisterSidecar(id: string): void {
  const sc = sidecars.get(id);
  if (!sc) return;
  sidecars.delete(id);
  console.log(`[registry] unregistered sidecar "${sc.name}" (id=${id})`);

  // Reject all pending requests for this sidecar
  for (const [reqId, req] of pending) {
    clearTimeout(req.timer);
    req.reject(new Error(`Sidecar "${sc.name}" disconnected`));
    pending.delete(reqId);
  }

  // Clear cached state if the claude sidecar disconnects
  if (sc.capabilities.includes('proxy')) {
    cachedState = null;
  }
}

export function handleSidecarMessage(id: string, msg: SidecarEvent | PtyEvent): void {
  // Check if this is a response to a pending request
  if ('id' in msg && msg.id && pending.has(msg.id)) {
    const req = pending.get(msg.id)!;
    pending.delete(msg.id);
    clearTimeout(req.timer);
    req.resolve(msg);
    return;
  }

  // Otherwise dispatch as event
  dispatchEvent(msg);
}

// ── Lookup ──

function findSidecarByCapability(cap: string): RegisteredSidecar | undefined {
  for (const sc of sidecars.values()) {
    if (sc.capabilities.includes(cap)) return sc;
  }
  return undefined;
}

function findSidecarByName(name: string): RegisteredSidecar | undefined {
  for (const sc of sidecars.values()) {
    if (sc.name === name) return sc;
  }
  return undefined;
}

// ── Event dispatch ──

function dispatchEvent(msg: SidecarEvent | PtyEvent) {
  const handlers = eventHandlers.get(msg.type);
  if (handlers) {
    for (const handler of handlers) {
      try {
        handler(msg);
      } catch {
        /* ignore */
      }
    }
  }
}

export function on(eventType: string, handler: EventHandler): () => void {
  if (!eventHandlers.has(eventType)) {
    eventHandlers.set(eventType, new Set());
  }
  eventHandlers.get(eventType)!.add(handler);
  return () => {
    eventHandlers.get(eventType)?.delete(handler);
  };
}

// ── Command sending ──

const DEFAULT_TIMEOUT_MS = 30_000;
const LONG_TIMEOUT_MS = 6 * 60 * 1000;

function sendCommand(cap: string, cmd: SidecarCommand | PtyCommand, timeoutMs = DEFAULT_TIMEOUT_MS): Promise<any> {
  return new Promise((resolve, reject) => {
    const sc = findSidecarByCapability(cap);
    if (!sc) {
      reject(new Error(`No sidecar with capability "${cap}" is connected`));
      return;
    }

    const timer = setTimeout(() => {
      pending.delete((cmd as any).id);
      reject(new Error(`Sidecar command ${cmd.type} timed out`));
    }, timeoutMs);

    pending.set((cmd as any).id, { resolve, reject, timer });
    sc.ws.send(JSON.stringify(cmd));
  });
}

function sendFire(cap: string, cmd: SidecarCommand | PtyCommand): void {
  const sc = findSidecarByCapability(cap);
  if (sc) {
    sc.ws.send(JSON.stringify(cmd));
  }
}

function sendCommandToSidecar(
  sc: RegisteredSidecar,
  cmd: SidecarCommand | PtyCommand,
  timeoutMs = DEFAULT_TIMEOUT_MS,
): Promise<any> {
  return new Promise((resolve, reject) => {
    const timer = setTimeout(() => {
      pending.delete((cmd as any).id);
      reject(new Error(`Sidecar command ${cmd.type} timed out`));
    }, timeoutMs);

    pending.set((cmd as any).id, { resolve, reject, timer });
    sc.ws.send(JSON.stringify(cmd));
  });
}

function sendFireToSidecar(sc: RegisteredSidecar, cmd: SidecarCommand | PtyCommand): void {
  sc.ws.send(JSON.stringify(cmd));
}

// ── On-demand Claude sidecar spawning ──

const USER_INSTANCE_SCRIPT = resolve(import.meta.dir, 'sidecar/claude/user-instance.ts');
const SIDECAR_SPAWN_TIMEOUT_MS = 15_000;

const claudeProcs = new Map<string, Subprocess>();
const claudeSpawnWaiters = new Map<string, Promise<RegisteredSidecar>>();

async function ensureClaudeSidecar(email: string): Promise<RegisteredSidecar> {
  const name = `claude:${email}`;

  // Already registered?
  const existing = findSidecarByName(name);
  if (existing) return existing;

  // Already spawning?
  const waiter = claudeSpawnWaiters.get(email);
  if (waiter) return waiter;

  // Spawn and wait for registration
  const promise = spawnAndWaitForRegistration(email, name);
  claudeSpawnWaiters.set(email, promise);
  try {
    return await promise;
  } finally {
    claudeSpawnWaiters.delete(email);
  }
}

async function spawnAndWaitForRegistration(email: string, name: string): Promise<RegisteredSidecar> {
  const proxySecret = await getProxySecret();
  const proxyPort = process.env.ANTHROPIC_PROXY_PORT ?? '5051';

  const env: Record<string, string> = {
    ...(process.env as Record<string, string>),
    CLAUDE_USER_EMAIL: email,
    ANTHROPIC_BASE_URL: `http://127.0.0.1:${proxyPort}`,
    ANTHROPIC_API_KEY: proxySecret,
  };

  const proc = Bun.spawn(['bun', 'run', USER_INSTANCE_SCRIPT], {
    env,
    stdout: 'inherit',
    stderr: 'inherit',
  });

  claudeProcs.set(email, proc);

  // Clean up on exit
  proc.exited.then(() => {
    claudeProcs.delete(email);
  });

  // Wait for the sidecar to register
  return new Promise<RegisteredSidecar>((resolve, reject) => {
    const timeout = setTimeout(() => {
      unsub();
      reject(new Error(`Claude sidecar for ${email} failed to register within ${SIDECAR_SPAWN_TIMEOUT_MS}ms`));
    }, SIDECAR_SPAWN_TIMEOUT_MS);

    // Poll for registration (the sidecar connects via WebSocket and registerSidecar is called)
    const check = () => {
      const sc = findSidecarByName(name);
      if (sc) {
        clearTimeout(timeout);
        clearInterval(interval);
        resolve(sc);
      }
    };
    const interval = setInterval(check, 50);

    const unsub = () => {
      clearTimeout(timeout);
      clearInterval(interval);
    };
  });
}

// ── Public API ──

export function isConnected(): boolean {
  return findSidecarByCapability('proxy') !== undefined;
}

export function getCachedState(): ClaudeState | null {
  return cachedState;
}

export async function syncState(): Promise<ClaudeState> {
  const res = await sendCommand('proxy', { type: 'state:sync', id: nextId() });
  if (res.type === 'state:sync') {
    cachedState = res.state;
    return res.state;
  }
  throw new Error('Unexpected response');
}

export async function getProxySecret(): Promise<string> {
  if (cachedState?.proxySecret) return cachedState.proxySecret;
  const res = await sendCommand('proxy', { type: 'proxy:secret', id: nextId() });
  if (res.type === 'proxy:secret') return res.secret;
  throw new Error('Failed to get proxy secret');
}

export function getProxySecretSync(): string {
  return cachedState?.proxySecret ?? '';
}

// ── Claude Code (per-user routing) ──

export async function spawnClaude(params: ClaudeSpawnParams): Promise<ClaudeCodeResult> {
  const sc = await ensureClaudeSidecar(params.email);
  const res = await sendCommandToSidecar(sc, { type: 'claude:spawn', id: nextId(), params }, LONG_TIMEOUT_MS);
  if (res.type === 'claude:result') return res.result;
  if (res.type === 'claude:error') throw new Error(res.error);
  throw new Error('Unexpected response');
}

export async function spawnClaudeStreaming(params: ClaudeSpawnStreamingParams): Promise<void> {
  const sc = await ensureClaudeSidecar(params.email);
  const res = await sendCommandToSidecar(sc, { type: 'claude:spawn-streaming', id: nextId(), params });
  if (res.type === 'claude:spawned') return;
  if (res.type === 'claude:error') throw new Error(res.error);
  throw new Error('Unexpected response');
}

export function killClaude(sessionKey: string, email: string): void {
  const sc = findSidecarByName(`claude:${email}`);
  if (sc) sendFireToSidecar(sc, { type: 'claude:kill', id: nextId(), sessionKey });
}

export function clearClaudeSession(sessionKey: string, email?: string): void {
  if (email) {
    const sc = findSidecarByName(`claude:${email}`);
    if (sc) sendFireToSidecar(sc, { type: 'claude:clear-session', id: nextId(), sessionKey });
  } else {
    // Broadcast to all claude sidecars (used when email is not available)
    for (const sc of sidecars.values()) {
      if (sc.capabilities.includes('claude')) {
        sendFireToSidecar(sc, { type: 'claude:clear-session', id: nextId(), sessionKey });
      }
    }
  }
}

export function onClaudeEvent(handler: (sessionKey: string, event: ChatEvent) => void): () => void {
  return on('claude:event', (msg) => {
    if (msg.type === 'claude:event') {
      handler(
        (msg as SidecarEvent & { type: 'claude:event' }).sessionKey,
        (msg as SidecarEvent & { type: 'claude:event' }).event,
      );
    }
  });
}

// ── OpenCode Code (single sidecar, capability 'opencode') ──

export async function spawnOpenCodeStreaming(params: OpenCodeRunParams): Promise<void> {
  const res = await sendCommand('opencode', { type: 'opencode:run-streaming', id: nextId(), params });
  if (res.type === 'opencode:spawned') return;
  if (res.type === 'opencode:error') throw new Error(res.error);
  throw new Error('Unexpected response');
}

export function killOpenCode(sessionKey: string): void {
  sendFire('opencode', { type: 'opencode:kill', id: nextId(), sessionKey });
}

export function onOpenCodeEvent(handler: (sessionKey: string, event: ChatEvent) => void): () => void {
  return on('opencode:event', (msg) => {
    if (msg.type === 'opencode:event') {
      handler(
        (msg as SidecarEvent & { type: 'opencode:event' }).sessionKey,
        (msg as SidecarEvent & { type: 'opencode:event' }).event,
      );
    }
  });
}

export function onOpenCodeSession(handler: (sessionKey: string, sessionId: string) => void): () => void {
  return on('opencode:session', (msg) => {
    if (msg.type === 'opencode:session') {
      handler(
        (msg as SidecarEvent & { type: 'opencode:session' }).sessionKey,
        (msg as SidecarEvent & { type: 'opencode:session' }).sessionId,
      );
    }
  });
}

// ── Terminal (PTY sidecar) ──

export function sendPtyCommand(cmd: PtyCommand): void {
  sendFire('terminal', cmd);
}

export async function sendPtyCommandAsync(cmd: PtyCommand, timeoutMs = DEFAULT_TIMEOUT_MS): Promise<PtyEvent> {
  return sendCommand('terminal', cmd, timeoutMs);
}

export function isTerminalConnected(): boolean {
  return findSidecarByCapability('terminal') !== undefined;
}

// ── VNC ──

export async function startVnc(params: VncStartParams): Promise<{ port: number; display: number }> {
  const res = await sendCommand('vnc', { type: 'vnc:start', id: nextId(), params });
  if (res.type === 'vnc:started') return { port: res.port, display: res.display };
  if (res.type === 'vnc:error') throw new Error(res.error);
  throw new Error('Unexpected response');
}

export function stopVnc(email: string): void {
  sendFire('vnc', { type: 'vnc:stop', id: nextId(), email });
}

export async function getVncStatus(email: string): Promise<VncSessionInfo | null> {
  const res = await sendCommand('vnc', { type: 'vnc:status', id: nextId(), email });
  if (res.type === 'vnc:status') return res.session;
  throw new Error('Unexpected response');
}

export function isVncConnected(): boolean {
  return findSidecarByCapability('vnc') !== undefined;
}