Files
hapi/hub/src/web/server.ts
T

521 lines
22 KiB
TypeScript

import { Hono } from 'hono'
import { cors } from 'hono/cors'
import { logger } from 'hono/logger'
import { join } from 'node:path'
import { existsSync } from 'node:fs'
import { serveStatic } from 'hono/bun'
import { getConfiguration } from '../configuration'
import { PROTOCOL_VERSION } from '@hapi/protocol'
import { buildGeminiLiveSetupMessage, QWEN_REALTIME_MODEL } from '@hapi/protocol/voice'
import { createQwenProxyWebSocketHandler } from './qwenProxyHandler'
import { decodeVoiceSystemPromptParam } from '../voiceSystemPromptParam'
import type { SyncEngine } from '../sync/syncEngine'
import type { PushService } from '../push/pushService'
import { createAuthMiddleware, type WebAppEnv } from './middleware/auth'
import { createAuthRoutes } from './routes/auth'
import { createBindRoutes } from './routes/bind'
import { createEventsRoutes } from './routes/events'
import { createSessionsRoutes } from './routes/sessions'
import { createMessagesRoutes } from './routes/messages'
import { createPermissionsRoutes } from './routes/permissions'
import { createMachinesRoutes } from './routes/machines'
import { createStorageRoutes } from './routes/storage'
import { createUsageRoutes } from './routes/usage'
import { createGitRoutes } from './routes/git'
import { createCliRoutes } from './routes/cli'
import { createCodexDesktopRoutes } from './routes/codexDesktop'
import { createPushRoutes } from './routes/push'
import { createNotificationPreferencesRoutes } from './routes/notificationPreferences'
import { createDevicesRoutes } from './routes/devices'
import { createVoiceRoutes } from './routes/voice'
import { createClaudeModelsRoutes } from './routes/claudeModels'
import { createNotificationCopyRoutes } from './routes/notificationCopy'
import type { SSEManager } from '../sse/sseManager'
import type { VisibilityTracker } from '../visibility/visibilityTracker'
import type { Server as BunServer, ServerWebSocket } from 'bun'
import type { Server as SocketEngine } from '@socket.io/bun-engine'
import { jwtVerify } from 'jose'
import type { WebSocketData } from '@socket.io/bun-engine'
import { loadEmbeddedAssetMap, type EmbeddedWebAsset } from './embeddedAssets'
import { isBunCompiled } from '../utils/bunCompiled'
import type { Store } from '../store'
// Normalise upstream close codes before forwarding to the browser client.
// Codes 1005/1006/1015 are reserved and cannot be sent in a close frame;
// abnormal upstream drops commonly produce 1006, which would throw on clientWs.close().
function toClientCloseCode(code: number): number {
return code >= 1000 && code <= 4999 && code !== 1005 && code !== 1006 && code !== 1015
? code
: 1011
}
function decodeWsText(message: string | ArrayBuffer | Uint8Array): string {
if (typeof message === 'string') return message
const bytes = message instanceof Uint8Array ? message : new Uint8Array(message)
return new TextDecoder().decode(bytes)
}
function isGeminiSetupFrame(message: string | ArrayBuffer | Uint8Array): boolean {
try {
const parsed = JSON.parse(decodeWsText(message)) as unknown
return parsed !== null && typeof parsed === 'object' && 'setup' in (parsed as object)
} catch {
return false
}
}
function isGeminiSetupCompleteFrame(message: string | ArrayBuffer | Uint8Array): boolean {
try {
const parsed = JSON.parse(decodeWsText(message)) as unknown
return parsed !== null && typeof parsed === 'object' && 'setupComplete' in (parsed as object)
} catch {
return false
}
}
const MAX_GEMINI_PENDING_BYTES = 1024 * 1024 // 1 MiB — rejects setup-window floods
function frameByteSize(msg: string | ArrayBuffer | Uint8Array): number {
return typeof msg === 'string' ? msg.length : (msg as ArrayBuffer | Uint8Array).byteLength
}
// Gemini Live WebSocket proxy — relays browser WS to Google, bypassing region restrictions
function createGeminiProxyWebSocketHandler() {
const GEMINI_WS_BASE = 'wss://generativelanguage.googleapis.com/ws/google.ai.generativelanguage.v1beta.GenerativeService.BidiGenerateContent'
const upstreamMap = new WeakMap<ServerWebSocket<unknown>, WebSocket>()
// pendingMap holds queued client frames until Google acknowledges setup via setupComplete.
// Flushed on setupComplete; until then message() queues rather than forwards.
const pendingMap = new WeakMap<ServerWebSocket<unknown>, Array<string | ArrayBuffer | Uint8Array>>()
const pendingBytesMap = new WeakMap<ServerWebSocket<unknown>, number>()
return {
open(clientWs: ServerWebSocket<unknown>) {
const data = clientWs.data as {
_geminiProxy: boolean
apiKey: string
language?: string
voiceName?: string
systemInstruction?: string
affectiveDialog?: boolean
}
const upstreamUrl = `${process.env.GEMINI_LIVE_WS_URL || GEMINI_WS_BASE}?key=${encodeURIComponent(data.apiKey)}`
const pending: Array<string | ArrayBuffer | Uint8Array> = []
pendingMap.set(clientWs, pending)
pendingBytesMap.set(clientWs, 0)
const upstream = new WebSocket(upstreamUrl)
upstreamMap.set(clientWs, upstream)
upstream.onopen = () => {
// Hub-owned setup only — never forward client setup (prevents generic Gemini proxy abuse).
// Do NOT flush pending here: wait for Google's setupComplete before forwarding client frames.
upstream.send(JSON.stringify(buildGeminiLiveSetupMessage(
data.language,
data.voiceName,
data.systemInstruction,
{ affectiveDialog: data.affectiveDialog }
)))
}
upstream.onmessage = (event) => {
try {
if (clientWs.readyState === 1) {
clientWs.send(typeof event.data === 'string' ? event.data : new Uint8Array(event.data as ArrayBuffer))
}
} catch { /* client gone */ }
// Flush queued client frames only after Google acknowledges setup.
const pending = pendingMap.get(clientWs)
if (pending && isGeminiSetupCompleteFrame(event.data as string | ArrayBuffer)) {
pendingMap.delete(clientWs)
pendingBytesMap.delete(clientWs)
for (const queued of pending) {
try { upstream.send(queued) } catch { /* upstream gone */ }
}
}
}
upstream.onerror = () => {
pendingMap.delete(clientWs)
pendingBytesMap.delete(clientWs)
try { clientWs.close(1011, 'Upstream error') } catch { /* */ }
}
upstream.onclose = (event) => {
pendingMap.delete(clientWs)
pendingBytesMap.delete(clientWs)
try { clientWs.close(toClientCloseCode(event.code), event.reason || 'Upstream closed') } catch { /* client gone */ }
upstreamMap.delete(clientWs)
}
},
message(clientWs: ServerWebSocket<unknown>, message: string | ArrayBuffer | Uint8Array) {
if (isGeminiSetupFrame(message)) {
try { clientWs.close(1008, 'Client-provided Gemini setup is not allowed') } catch { /* */ }
return
}
const upstream = upstreamMap.get(clientWs)
const pending = pendingMap.get(clientWs)
if (pending) {
// Still awaiting setupComplete — queue, but cap to prevent setup-window floods.
const total = (pendingBytesMap.get(clientWs) ?? 0) + frameByteSize(message)
if (total > MAX_GEMINI_PENDING_BYTES) {
try { clientWs.close(1009, 'Setup-window frame budget exceeded') } catch { /* */ }
return
}
pendingBytesMap.set(clientWs, total)
pending.push(message)
} else if (upstream?.readyState === WebSocket.OPEN) {
upstream.send(message)
}
},
close(clientWs: ServerWebSocket<unknown>, code: number, reason: string) {
const upstream = upstreamMap.get(clientWs)
pendingMap.delete(clientWs)
pendingBytesMap.delete(clientWs)
if (upstream) {
try { upstream.close(toClientCloseCode(code), (reason || 'Client closed').slice(0, 123)) } catch { /* */ }
upstreamMap.delete(clientWs)
}
}
}
}
// Qwen Realtime WebSocket proxy — bridges browser (no custom headers) to DashScope
// (requires Authorization header). Implementation extracted to `./qwenProxyHandler` so
// the ack-gating behaviour is unit-testable; `createQwenProxyWebSocketHandler` is imported above.
function findWebappDistDir(): { distDir: string; indexHtmlPath: string } {
const candidates = [
join(process.cwd(), '..', 'web', 'dist'),
join(import.meta.dir, '..', '..', '..', 'web', 'dist'),
join(process.cwd(), 'web', 'dist')
]
for (const distDir of candidates) {
const indexHtmlPath = join(distDir, 'index.html')
if (existsSync(indexHtmlPath)) {
return { distDir, indexHtmlPath }
}
}
const distDir = candidates[0]
return { distDir, indexHtmlPath: join(distDir, 'index.html') }
}
function serveEmbeddedAsset(asset: EmbeddedWebAsset): Response {
const headers: Record<string, string> = {
'Content-Type': asset.mimeType
}
if (asset.path === '/sw.js') {
headers['Cache-Control'] = 'no-store, no-cache, must-revalidate'
headers['CDN-Cache-Control'] = 'no-store'
headers['Cloudflare-CDN-Cache-Control'] = 'no-store'
}
return new Response(Bun.file(asset.sourcePath), {
headers
})
}
function createWebApp(options: {
getSyncEngine: () => SyncEngine | null
getSseManager: () => SSEManager | null
getVisibilityTracker: () => VisibilityTracker | null
jwtSecret: Uint8Array
store: Store
vapidPublicKey: string
pushService: PushService
corsOrigins?: string[]
embeddedAssetMap: Map<string, EmbeddedWebAsset> | null
relayMode?: boolean
officialWebUrl?: string
dataDir: string
}): Hono<WebAppEnv> {
const app = new Hono<WebAppEnv>()
app.use('*', logger())
// Health check endpoint (no auth required)
app.get('/health', (c) => c.json({ status: 'ok', protocolVersion: PROTOCOL_VERSION }))
const configuration = getConfiguration()
const corsOrigins = options.corsOrigins ?? configuration.corsOrigins
const corsOriginOption = corsOrigins.includes('*') ? '*' : corsOrigins
const corsMiddleware = cors({
origin: corsOriginOption,
allowMethods: ['GET', 'POST', 'PUT', 'PATCH', 'DELETE', 'OPTIONS'],
allowHeaders: ['authorization', 'content-type']
})
app.use('/api/*', corsMiddleware)
app.use('/cli/*', corsMiddleware)
app.route('/cli', createCliRoutes(options.getSyncEngine))
app.route('/api', createAuthRoutes(options.jwtSecret, options.store))
app.route('/api', createBindRoutes(options.jwtSecret, options.store))
app.use('/api/*', createAuthMiddleware(options.jwtSecret))
app.route('/api', createEventsRoutes(options.getSseManager, options.getSyncEngine, options.getVisibilityTracker))
app.route('/api', createSessionsRoutes(options.getSyncEngine))
app.route('/api', createMessagesRoutes(options.getSyncEngine))
app.route('/api', createPermissionsRoutes(options.getSyncEngine))
app.route('/api', createMachinesRoutes(options.getSyncEngine))
app.route('/api', createStorageRoutes(configuration.dbPath))
app.route('/api', createUsageRoutes(options.store))
app.route('/api', createGitRoutes(options.getSyncEngine))
// 中文注释:这里提供两类 Codex 辅助能力:扫描本地 transcript 以导入到 Hapi,以及按需重启 Codex Desktop 客户端。
app.route('/api', createCodexDesktopRoutes({
store: options.store,
getSyncEngine: options.getSyncEngine
}))
app.route('/api', createPushRoutes(options.store, options.vapidPublicKey, options.pushService))
app.route('/api', createNotificationPreferencesRoutes(options.store))
app.route('/api', createClaudeModelsRoutes(options.dataDir))
app.route('/api', createNotificationCopyRoutes(options.dataDir))
app.route('/api', createDevicesRoutes(options.store))
app.route('/api', createVoiceRoutes())
// Skip static serving in relay mode, show helpful message on root
if (options.relayMode) {
const officialUrl = options.officialWebUrl || 'https://app.hapi.run'
app.get('/', (c) => {
return c.html(`<!DOCTYPE html>
<html>
<head><meta charset="utf-8"><title>HAPI Hub</title></head>
<body style="font-family: system-ui; padding: 2rem; max-width: 600px;">
<h1>HAPI Hub</h1>
<p>This hub is running in relay mode. Please use the official web app:</p>
<p><a href="${officialUrl}">${officialUrl}</a></p>
<details>
<summary>Why am I seeing this?</summary>
<p style="margin-top: 0.5rem; color: #666;">
When relay mode is enabled, all traffic flows through our relay infrastructure with end-to-end encryption.
To reduce bandwidth and improve performance, the frontend is served separately
from GitHub Pages instead of through the relay tunnel.
</p>
</details>
</body>
</html>`)
})
return app
}
if (options.embeddedAssetMap) {
const embeddedAssetMap = options.embeddedAssetMap
const indexHtmlAsset = embeddedAssetMap.get('/index.html')
if (!indexHtmlAsset) {
app.get('*', (c) => {
return c.text(
'Embedded Mini App is missing index.html. Rebuild the executable after running bun run build:web.',
503
)
})
return app
}
app.use('*', async (c, next) => {
if (c.req.path.startsWith('/api')) {
return await next()
}
if (c.req.method !== 'GET' && c.req.method !== 'HEAD') {
return await next()
}
const asset = embeddedAssetMap.get(c.req.path)
if (asset) {
return serveEmbeddedAsset(asset)
}
return await next()
})
app.get('*', async (c, next) => {
if (c.req.path.startsWith('/api')) {
await next()
return
}
return serveEmbeddedAsset(indexHtmlAsset)
})
return app
}
const { distDir, indexHtmlPath } = findWebappDistDir()
if (!existsSync(indexHtmlPath)) {
app.get('/', (c) => {
return c.text(
'Mini App is not built.\n\nRun:\n cd web\n bun install\n bun run build\n',
503
)
})
return app
}
app.use('/assets/*', serveStatic({ root: distDir }))
app.use('*', async (c, next) => {
if (c.req.path.startsWith('/api')) {
await next()
return
}
return await serveStatic({ root: distDir })(c, next)
})
app.get('*', async (c, next) => {
if (c.req.path.startsWith('/api')) {
await next()
return
}
return await serveStatic({ root: distDir, path: 'index.html' })(c, next)
})
return app
}
export async function startWebServer(options: {
getSyncEngine: () => SyncEngine | null
getSseManager: () => SSEManager | null
getVisibilityTracker: () => VisibilityTracker | null
jwtSecret: Uint8Array
store: Store
vapidPublicKey: string
pushService: PushService
socketEngine: SocketEngine
corsOrigins?: string[]
relayMode?: boolean
officialWebUrl?: string
dataDir: string
}): Promise<BunServer<WebSocketData>> {
const isCompiled = isBunCompiled()
const embeddedAssetMap = isCompiled ? await loadEmbeddedAssetMap() : null
const app = createWebApp({
getSyncEngine: options.getSyncEngine,
getSseManager: options.getSseManager,
getVisibilityTracker: options.getVisibilityTracker,
jwtSecret: options.jwtSecret,
store: options.store,
vapidPublicKey: options.vapidPublicKey,
pushService: options.pushService,
corsOrigins: options.corsOrigins,
embeddedAssetMap,
dataDir: options.dataDir,
relayMode: options.relayMode,
officialWebUrl: options.officialWebUrl
})
const configuration = getConfiguration()
const socketHandler = options.socketEngine.handler()
// Wrap socket.io websocket handler to also support Gemini/Qwen proxy connections
const originalWsHandler = socketHandler.websocket
const geminiProxyHandler = createGeminiProxyWebSocketHandler()
const qwenProxyHandler = createQwenProxyWebSocketHandler()
// eslint-disable-next-line @typescript-eslint/no-explicit-any
const server = (Bun.serve as any)({
hostname: configuration.listenHost,
port: configuration.listenPort,
idleTimeout: Math.max(30, socketHandler.idleTimeout),
maxRequestBodySize: Math.max(socketHandler.maxRequestBodySize, 68 * 1024 * 1024),
websocket: {
...originalWsHandler,
open(ws: unknown) {
const wsAny = ws as ServerWebSocket<{ _qwenProxy?: boolean; _geminiProxy?: boolean }>
if (wsAny.data?._geminiProxy) {
geminiProxyHandler.open(wsAny)
} else if (wsAny.data?._qwenProxy) {
qwenProxyHandler.open(wsAny)
} else {
originalWsHandler.open?.(ws as never)
}
},
message(ws: unknown, message: unknown) {
const wsAny = ws as ServerWebSocket<{ _qwenProxy?: boolean; _geminiProxy?: boolean }>
if (wsAny.data?._geminiProxy) {
geminiProxyHandler.message(wsAny, message as string)
} else if (wsAny.data?._qwenProxy) {
qwenProxyHandler.message(wsAny, message as string)
} else {
originalWsHandler.message?.(ws as never, message as never)
}
},
close(ws: unknown, code: number, reason: string) {
const wsAny = ws as ServerWebSocket<{ _qwenProxy?: boolean; _geminiProxy?: boolean }>
if (wsAny.data?._geminiProxy) {
geminiProxyHandler.close(wsAny, code, reason)
} else if (wsAny.data?._qwenProxy) {
qwenProxyHandler.close(wsAny, code, reason)
} else {
originalWsHandler.close?.(ws as never, code as never, reason as never)
}
}
},
fetch: async (req: Request, server: { upgrade: (req: Request, opts?: unknown) => boolean }) => {
const url = new URL(req.url)
if (url.pathname.startsWith('/socket.io/')) {
return socketHandler.fetch(req, server as never)
}
// Voice WebSocket proxies — require JWT auth via query param
// (browser WebSocket API cannot set custom headers)
if (url.pathname === '/api/voice/gemini-ws' || url.pathname === '/api/voice/qwen-ws') {
const token = url.searchParams.get('token')
if (!token) {
return new Response('Missing authorization token', { status: 401 })
}
try {
await jwtVerify(token, options.jwtSecret, { algorithms: ['HS256'] })
} catch {
return new Response('Invalid token', { status: 401 })
}
}
// Gemini Live WebSocket proxy
if (url.pathname === '/api/voice/gemini-ws') {
const apiKey = process.env.GEMINI_API_KEY || process.env.GOOGLE_API_KEY
if (!apiKey) {
return new Response('Gemini API key not configured', { status: 400 })
}
const language = url.searchParams.get('language') ?? undefined
const voiceParam = url.searchParams.get('voice')?.trim() || undefined
const systemInstruction = decodeVoiceSystemPromptParam(url.searchParams.get('systemPrompt'))
const affectiveDialog = url.searchParams.get('affectiveDialog') === '1'
const upgraded = (server as unknown as { upgrade: (req: Request, opts: unknown) => boolean }).upgrade(req, {
data: { _geminiProxy: true, apiKey, language, voiceName: voiceParam, systemInstruction, affectiveDialog }
})
if (!upgraded) {
return new Response('WebSocket upgrade failed', { status: 500 })
}
return undefined as unknown as Response
}
// Qwen Realtime WebSocket proxy
if (url.pathname === '/api/voice/qwen-ws') {
const apiKey = process.env.DASHSCOPE_API_KEY || process.env.QWEN_API_KEY
const model = QWEN_REALTIME_MODEL
const language = url.searchParams.get('language') ?? undefined
const voiceParam = url.searchParams.get('voice')?.trim() || undefined
const systemInstruction = decodeVoiceSystemPromptParam(url.searchParams.get('systemPrompt'))
if (!apiKey) {
return new Response('DashScope API key not configured', { status: 400 })
}
const upgraded = (server as unknown as { upgrade: (req: Request, opts: unknown) => boolean }).upgrade(req, {
data: { _qwenProxy: true, apiKey, model, language, voiceName: voiceParam, systemInstruction }
})
if (!upgraded) {
return new Response('WebSocket upgrade failed', { status: 500 })
}
return undefined as unknown as Response
}
return app.fetch(req)
}
})
console.log(`[Web] hub listening on ${configuration.listenHost}:${configuration.listenPort}`)
console.log(`[Web] public URL: ${configuration.publicUrl}`)
return server
}