refactor: share proxy capture event recording

src/proxy-capture/proxy-server.ts

@@ -7,6 +7,7 @@ import { URL } from "node:url";
 import { ensureDebugProxyCa } from "./ca.js";
 import type { DebugProxySettings } from "./env.js";
 import { getDebugProxyCaptureStore } from "./store.sqlite.js";
+import type { CaptureEventRecord } from "./types.js";
 
 const TRUTHY_ENV = new Set(["1", "true", "yes", "on"]);
 const DEBUG_PROXY_DIRECT_CONNECT_OVERRIDE =
@@ -39,6 +40,26 @@ type DebugProxyServerHandle = {
   stop: () => Promise<void>;
 };
 
+type ProxyCaptureEventInput = Omit<
+  CaptureEventRecord,
+  "sessionId" | "ts" | "sourceScope" | "sourceProcess"
+>;
+
+function createProxyCaptureRecorder(params: {
+  store: ReturnType<typeof getDebugProxyCaptureStore>;
+  settings: DebugProxySettings;
+}) {
+  return (event: ProxyCaptureEventInput): void => {
+    params.store.recordEvent({
+      sessionId: params.settings.sessionId,
+      ts: Date.now(),
+      sourceScope: "openclaw",
+      sourceProcess: params.settings.sourceProcess,
+      ...event,
+    });
+  };
+}
+
 export function parseConnectTarget(rawTarget: string | undefined): {
   hostname: string;
   port: number;
@@ -97,6 +118,7 @@ export async function startDebugProxyServer(params: {
 }): Promise<DebugProxyServerHandle> {
   await ensureDebugProxyCa(params.settings.certDir);
   const store = getDebugProxyCaptureStore(params.settings.dbPath, params.settings.blobDir);
+  const recordProxyEvent = createProxyCaptureRecorder({ store, settings: params.settings });
   const host = params.host?.trim() || "127.0.0.1";
 
   const server = createServer(async (req: IncomingMessage, res: ServerResponse) => {
@@ -106,11 +128,7 @@ export async function startDebugProxyServer(params: {
       target = normalizeTargetUrl(req);
     } catch (error) {
       const message = "Invalid proxy target URL";
-      store.recordEvent({
-        sessionId: params.settings.sessionId,
-        ts: Date.now(),
-        sourceScope: "openclaw",
-        sourceProcess: params.settings.sourceProcess,
+      recordProxyEvent({
         protocol: "http",
         direction: "local",
         kind: "error",
@@ -129,22 +147,26 @@ export async function startDebugProxyServer(params: {
       res.end(responseBody);
       return;
     }
+    const targetProtocol = target.protocol === "https:" ? "https" : "http";
+    const targetPath = `${target.pathname}${target.search}`;
+    const recordTargetEvent = (
+      event: Omit<ProxyCaptureEventInput, "protocol" | "flowId" | "method" | "host" | "path">,
+    ) =>
+      recordProxyEvent({
+        protocol: targetProtocol,
+        flowId,
+        method: req.method,
+        host: target.host,
+        path: targetPath,
+        ...event,
+      });
     try {
       assertDebugProxyDirectUpstreamAllowed();
     } catch (error) {
       const message = error instanceof Error ? error.message : String(error);
-      store.recordEvent({
-        sessionId: params.settings.sessionId,
-        ts: Date.now(),
-        sourceScope: "openclaw",
-        sourceProcess: params.settings.sourceProcess,
-        protocol: target.protocol === "https:" ? "https" : "http",
+      recordTargetEvent({
         direction: "local",
         kind: "error",
-        flowId,
-        method: req.method,
-        host: target.host,
-        path: `${target.pathname}${target.search}`,
         errorText: message,
       });
       const responseBody = `${message}\n`;
@@ -157,18 +179,9 @@ export async function startDebugProxyServer(params: {
       return;
     }
     const body = await readBody(req);
-    store.recordEvent({
-      sessionId: params.settings.sessionId,
-      ts: Date.now(),
-      sourceScope: "openclaw",
-      sourceProcess: params.settings.sourceProcess,
-      protocol: target.protocol === "https:" ? "https" : "http",
+    recordTargetEvent({
       direction: "outbound",
       kind: "request",
-      flowId,
-      method: req.method,
-      host: target.host,
-      path: `${target.pathname}${target.search}`,
       headersJson: JSON.stringify(req.headers),
       dataText: body.subarray(0, 8192).toString("utf8"),
     });
@@ -187,18 +200,9 @@ export async function startDebugProxyServer(params: {
         });
         upstreamRes.on("end", () => {
           const responseBody = Buffer.concat(chunks);
-          store.recordEvent({
-            sessionId: params.settings.sessionId,
-            ts: Date.now(),
-            sourceScope: "openclaw",
-            sourceProcess: params.settings.sourceProcess,
-            protocol: target.protocol === "https:" ? "https" : "http",
+          recordTargetEvent({
             direction: "inbound",
             kind: "response",
-            flowId,
-            method: req.method,
-            host: target.host,
-            path: `${target.pathname}${target.search}`,
             status: upstreamRes.statusCode ?? undefined,
             headersJson: JSON.stringify(upstreamRes.headers),
             dataText: responseBody.subarray(0, 8192).toString("utf8"),
@@ -209,18 +213,9 @@ export async function startDebugProxyServer(params: {
       },
     );
     upstream.on("error", (error) => {
-      store.recordEvent({
-        sessionId: params.settings.sessionId,
-        ts: Date.now(),
-        sourceScope: "openclaw",
-        sourceProcess: params.settings.sourceProcess,
-        protocol: target.protocol === "https:" ? "https" : "http",
+      recordTargetEvent({
         direction: "local",
         kind: "error",
-        flowId,
-        method: req.method,
-        host: target.host,
-        path: `${target.pathname}${target.search}`,
         errorText: error.message,
       });
       res.statusCode = 502;
@@ -241,11 +236,7 @@ export async function startDebugProxyServer(params: {
       hostname = parsed.hostname;
       port = parsed.port;
     } catch (error) {
-      store.recordEvent({
-        sessionId: params.settings.sessionId,
-        ts: Date.now(),
-        sourceScope: "openclaw",
-        sourceProcess: params.settings.sourceProcess,
+      recordProxyEvent({
         protocol: "connect",
         direction: "local",
         kind: "error",
@@ -257,11 +248,7 @@ export async function startDebugProxyServer(params: {
       clientSocket.end("HTTP/1.1 400 Bad Request\r\n\r\n");
       return;
     }
-    store.recordEvent({
-      sessionId: params.settings.sessionId,
-      ts: Date.now(),
-      sourceScope: "openclaw",
-      sourceProcess: params.settings.sourceProcess,
+    recordProxyEvent({
       protocol: "connect",
       direction: "local",
       kind: "connect",
@@ -274,11 +261,7 @@ export async function startDebugProxyServer(params: {
       assertDebugProxyDirectUpstreamAllowed();
     } catch (error) {
       const message = error instanceof Error ? error.message : String(error);
-      store.recordEvent({
-        sessionId: params.settings.sessionId,
-        ts: Date.now(),
-        sourceScope: "openclaw",
-        sourceProcess: params.settings.sourceProcess,
+      recordProxyEvent({
         protocol: "connect",
         direction: "local",
         kind: "error",
@@ -302,11 +285,7 @@ export async function startDebugProxyServer(params: {
       upstreamSocket.pipe(clientSocket);
     });
     clientSocket.on("error", (error) => {
-      store.recordEvent({
-        sessionId: params.settings.sessionId,
-        ts: Date.now(),
-        sourceScope: "openclaw",
-        sourceProcess: params.settings.sourceProcess,
+      recordProxyEvent({
         protocol: "connect",
         direction: "local",
         kind: "error",
@@ -318,11 +297,7 @@ export async function startDebugProxyServer(params: {
       upstreamSocket.destroy();
     });
     upstreamSocket.on("error", (error) => {
-      store.recordEvent({
-        sessionId: params.settings.sessionId,
-        ts: Date.now(),
-        sourceScope: "openclaw",
-        sourceProcess: params.settings.sourceProcess,
+      recordProxyEvent({
         protocol: "connect",
         direction: "local",
         kind: "error",