agent-ui/server/services/sync-server.ts

/**
 * Sync Server
 * Unified WebSocket server for real-time synchronization:
 * - Git change notifications
 * - Torch (multi-browser MCP control)
 */

import { PORT_GIT, WORKING_DIR } from '../config'
import { setupGitWatcher, handleGitClient, cleanupGitWatcher } from './handlers/git-handler'
import { setupComponentsWatcher, cleanupComponentsWatcher } from './handlers/components-handler'
import { setupTranscriptDebugWatcher, cleanupTranscriptDebugWatcher } from './handlers/transcript-debug-handler'
import { setTranscriptDebugBroadcast } from '../routes/transcript-debug'
import { setHooksApprovalBroadcast } from '../routes/hooks-approval'
import { handleTorchMessage, handleTorchConnect, handleTorchDisconnect, getTorchStatus, cleanupTorchHandler } from './handlers/torch-handler'

// Connected clients
const clients = new Set<any>()

export function broadcast(message: string, filter?: (ws: any) => boolean) {
  for (const ws of clients) {
    if (filter && !filter(ws)) continue
    try {
      ws.send(message)
    } catch {
      clients.delete(ws)
    }
  }
}

export function getClients() {
  return clients
}

export function startSyncServer() {
  const server = Bun.serve({
    port: PORT_GIT,
    async fetch(req, server) {
      const url = new URL(req.url)

      // CORS headers
      const corsHeaders = {
        'Access-Control-Allow-Origin': '*',
        'Access-Control-Allow-Methods': 'GET, POST, OPTIONS',
        'Access-Control-Allow-Headers': 'Content-Type'
      }

      if (req.method === 'OPTIONS') {
        return new Response(null, { headers: corsHeaders })
      }

      // Broadcast endpoint (used by API server cross-process)
      if (url.pathname === '/broadcast' && req.method === 'POST') {
        try {
          const message = await req.text()
          console.log(`[Sync] /broadcast received (${message.length} bytes) → ${clients.size} clients`)
          broadcast(message)
          return Response.json({ ok: true, clients: clients.size }, { headers: corsHeaders })
        } catch (e: any) {
          return Response.json({ error: e.message }, { status: 400, headers: corsHeaders })
        }
      }

      // Health check
      if (url.pathname === '/health') {
        const torchStatus = getTorchStatus()
        return Response.json({
          status: 'ok',
          clients: clients.size,
          torch: torchStatus
        }, { headers: corsHeaders })
      }

      // WebSocket upgrade
      const upgrade = req.headers.get('upgrade')
      if (upgrade?.toLowerCase() === 'websocket') {
        const success = server.upgrade(req)
        if (success) return undefined
        return new Response('WebSocket upgrade failed', { status: 400 })
      }

      return new Response('Sync WebSocket Server (Git + Torch)', { status: 200 })
    },
    websocket: {
      open(ws) {
        clients.add(ws)
        console.log(`[Sync] Client connected (${clients.size} total)`)

        // Initialize client for both services
        handleGitClient(ws)
        handleTorchConnect(ws, broadcast)
      },
      message(ws, message) {
        try {
          const data = JSON.parse(message.toString())

          // Route to appropriate handler based on message type
          if (data.type?.startsWith('torch-') || ['register', 'request', 'release', 'transfer', 'update-name', 'set-auto-request'].includes(data.type)) {
            handleTorchMessage(ws, data, broadcast)
          }
          // Git doesn't expect messages from client
        } catch (e) {
          console.error('[Sync] Invalid message:', e)
        }
      },
      close(ws) {
        clients.delete(ws)
        console.log(`[Sync] Client disconnected (${clients.size} total)`)
        handleTorchDisconnect(ws, broadcast)
      }
    }
  })

  console.log(`[Sync] WebSocket server on port ${PORT_GIT}`)

  // Start file watchers
  setupGitWatcher(WORKING_DIR, broadcast)
  setupComponentsWatcher(WORKING_DIR, broadcast)
  setupTranscriptDebugWatcher(WORKING_DIR, broadcast)

  // Give the route handler access to broadcast for process-complete notifications
  setTranscriptDebugBroadcast(broadcast)
  setHooksApprovalBroadcast(broadcast)

  return server
}

export function stopSyncServer() {
  cleanupGitWatcher()
  cleanupComponentsWatcher()
  cleanupTranscriptDebugWatcher()
  cleanupTorchHandler()
  clients.clear()
}