|
| 1 | +import { runtimeStore } from "../runtime/store.js"; |
| 2 | +import { readVsockResponse } from "../runtime/protocol.js"; |
| 3 | +import { getVmSocket } from "../runtime/transport.js"; |
| 4 | +import { getSession, createSession, touchSession } from "./session.js"; |
| 5 | +import { gatewayLogger } from "../utils/logger.js"; |
| 6 | +import crypto from "crypto"; |
| 7 | +import type { Vm } from "../types/types.js"; |
| 8 | + |
| 9 | +import { createVm } from "../runtime/vm-manager.js"; |
| 10 | +import { Deque } from "../runtime/deque.js"; |
| 11 | + |
| 12 | +export async function ensureSession(sessionId: string): Promise<Vm> { |
| 13 | + let session = getSession(sessionId); |
| 14 | + if (session?.state === "active") { |
| 15 | + const fn = runtimeStore.functions.get(sessionId); |
| 16 | + if (fn && fn.vms.length > 0) return fn.vms[0]!; |
| 17 | + } |
| 18 | + |
| 19 | + session = session || createSession(sessionId); |
| 20 | + |
| 21 | + let fn = runtimeStore.functions.get(sessionId); |
| 22 | + if (!fn) { |
| 23 | + fn = { |
| 24 | + functionId: sessionId, |
| 25 | + queue: new Deque(), |
| 26 | + vms: [], |
| 27 | + readyVms: new Set(), |
| 28 | + weight: 1, |
| 29 | + inflightCount: 0, |
| 30 | + deficit: 0, |
| 31 | + pendingCreations: 0, |
| 32 | + }; |
| 33 | + runtimeStore.functions.set(sessionId, fn); |
| 34 | + } |
| 35 | + |
| 36 | + if (fn.vms.length === 0) { |
| 37 | + await createVm(sessionId, fn, "__exec__"); |
| 38 | + } |
| 39 | + |
| 40 | + session.state = "active"; |
| 41 | + return fn.vms[0]!; |
| 42 | +} |
| 43 | + |
| 44 | +export async function sendSessionMessage( |
| 45 | + sessionId: string, |
| 46 | + message: Record<string, any>, |
| 47 | + onStream?: (chunk: any) => void, |
| 48 | + timeout: number = 60000, |
| 49 | +): Promise<any> { |
| 50 | + const vm = await ensureSession(sessionId); |
| 51 | + touchSession(sessionId); |
| 52 | + |
| 53 | + const id = message.id || crypto.randomUUID(); |
| 54 | + const fullMessage = { ...message, id }; |
| 55 | + |
| 56 | + const socket = await getVmSocket(vm); |
| 57 | + socket.write(JSON.stringify(fullMessage) + "\n"); |
| 58 | + |
| 59 | + gatewayLogger.debug( |
| 60 | + { sessionId, messageType: message.type, messageId: id }, |
| 61 | + "message sent to VM" |
| 62 | + ); |
| 63 | + |
| 64 | + const result = await readVsockResponse(socket, timeout, onStream); |
| 65 | + return { ...result, messageId: id }; |
| 66 | +} |
0 commit comments