refactor(nextcloud-talk): extract webhook pipeline and shared test harness

extensions/nextcloud-talk/src/monitor.auth-order.test.ts

@@ -1,50 +1,5 @@
-import { type AddressInfo } from "node:net";
-import { afterEach, describe, expect, it, vi } from "vitest";
-import { createNextcloudTalkWebhookServer } from "./monitor.js";
-
-type WebhookHarness = {
-  webhookUrl: string;
-  stop: () => Promise<void>;
-};
-
-const cleanupFns: Array<() => Promise<void>> = [];
-
-afterEach(async () => {
-  while (cleanupFns.length > 0) {
-    const cleanup = cleanupFns.pop();
-    if (cleanup) {
-      await cleanup();
-    }
-  }
-});
-
-async function startWebhookServer(params: {
-  path: string;
-  maxBodyBytes: number;
-  readBody?: (req: import("node:http").IncomingMessage, maxBodyBytes: number) => Promise<string>;
-}): Promise<WebhookHarness> {
-  const { server, start } = createNextcloudTalkWebhookServer({
-    port: 0,
-    host: "127.0.0.1",
-    path: params.path,
-    secret: "nextcloud-secret",
-    maxBodyBytes: params.maxBodyBytes,
-    readBody: params.readBody,
-    onMessage: vi.fn(),
-  });
-  await start();
-  const address = server.address() as AddressInfo | null;
-  if (!address) {
-    throw new Error("missing server address");
-  }
-  return {
-    webhookUrl: `http://127.0.0.1:${address.port}${params.path}`,
-    stop: () =>
-      new Promise<void>((resolve) => {
-        server.close(() => resolve());
-      }),
-  };
-}
+import { describe, expect, it, vi } from "vitest";
+import { startWebhookServer } from "./monitor.test-harness.js";
 
 describe("createNextcloudTalkWebhookServer auth order", () => {
   it("rejects missing signature headers before reading request body", async () => {
@@ -55,8 +10,8 @@ describe("createNextcloudTalkWebhookServer auth order", () => {
       path: "/nextcloud-auth-order",
       maxBodyBytes: 128,
       readBody,
+      onMessage: vi.fn(),
     });
-    cleanupFns.push(harness.stop);
 
     const response = await fetch(harness.webhookUrl, {
       method: "POST",

extensions/nextcloud-talk/src/monitor.backend.test.ts

@@ -1,51 +1,7 @@
-import { type AddressInfo } from "node:net";
-import { afterEach, describe, expect, it, vi } from "vitest";
-import { createNextcloudTalkWebhookServer } from "./monitor.js";
+import { describe, expect, it, vi } from "vitest";
+import { startWebhookServer } from "./monitor.test-harness.js";
 import { generateNextcloudTalkSignature } from "./signature.js";
 
-type WebhookHarness = {
-  webhookUrl: string;
-  stop: () => Promise<void>;
-};
-
-const cleanupFns: Array<() => Promise<void>> = [];
-
-afterEach(async () => {
-  while (cleanupFns.length > 0) {
-    const cleanup = cleanupFns.pop();
-    if (cleanup) {
-      await cleanup();
-    }
-  }
-});
-
-async function startWebhookServer(params: {
-  path: string;
-  isBackendAllowed: (backend: string) => boolean;
-  onMessage: () => void | Promise<void>;
-}): Promise<WebhookHarness> {
-  const { server, start } = createNextcloudTalkWebhookServer({
-    port: 0,
-    host: "127.0.0.1",
-    path: params.path,
-    secret: "nextcloud-secret",
-    isBackendAllowed: params.isBackendAllowed,
-    onMessage: params.onMessage,
-  });
-  await start();
-  const address = server.address() as AddressInfo | null;
-  if (!address) {
-    throw new Error("missing server address");
-  }
-  return {
-    webhookUrl: `http://127.0.0.1:${address.port}${params.path}`,
-    stop: () =>
-      new Promise<void>((resolve) => {
-        server.close(() => resolve());
-      }),
-  };
-}
-
 describe("createNextcloudTalkWebhookServer backend allowlist", () => {
   it("rejects requests from unexpected backend origins", async () => {
     const onMessage = vi.fn(async () => {});
@@ -54,7 +10,6 @@ describe("createNextcloudTalkWebhookServer backend allowlist", () => {
       isBackendAllowed: (backend) => backend === "https://nextcloud.expected",
       onMessage,
     });
-    cleanupFns.push(harness.stop);
 
     const payload = {
       type: "Create",

extensions/nextcloud-talk/src/monitor.replay.test.ts

@@ -1,54 +1,7 @@
-import { type AddressInfo } from "node:net";
-import { afterEach, describe, expect, it, vi } from "vitest";
-import { createNextcloudTalkWebhookServer } from "./monitor.js";
+import { describe, expect, it, vi } from "vitest";
+import { startWebhookServer } from "./monitor.test-harness.js";
 import { generateNextcloudTalkSignature } from "./signature.js";
-
-type WebhookHarness = {
-  webhookUrl: string;
-  stop: () => Promise<void>;
-};
-
-const cleanupFns: Array<() => Promise<void>> = [];
-
-afterEach(async () => {
-  while (cleanupFns.length > 0) {
-    const cleanup = cleanupFns.pop();
-    if (cleanup) {
-      await cleanup();
-    }
-  }
-});
-
-async function startWebhookServer(params: {
-  path: string;
-  shouldProcessMessage?: (
-    message: Parameters<
-      NonNullable<Parameters<typeof createNextcloudTalkWebhookServer>[0]["onMessage"]>
-    >[0],
-  ) => boolean | Promise<boolean>;
-  onMessage: (message: { messageId: string }) => void | Promise<void>;
-}): Promise<WebhookHarness> {
-  const { server, start } = createNextcloudTalkWebhookServer({
-    port: 0,
-    host: "127.0.0.1",
-    path: params.path,
-    secret: "nextcloud-secret",
-    shouldProcessMessage: params.shouldProcessMessage,
-    onMessage: params.onMessage,
-  });
-  await start();
-  const address = server.address() as AddressInfo | null;
-  if (!address) {
-    throw new Error("missing server address");
-  }
-  return {
-    webhookUrl: `http://127.0.0.1:${address.port}${params.path}`,
-    stop: () =>
-      new Promise<void>((resolve) => {
-        server.close(() => resolve());
-      }),
-  };
-}
+import type { NextcloudTalkInboundMessage } from "./types.js";
 
 function createSignedRequest(body: string): { random: string; signature: string } {
   return generateNextcloudTalkSignature({
@@ -61,7 +14,7 @@ describe("createNextcloudTalkWebhookServer replay handling", () => {
   it("acknowledges replayed requests and skips onMessage side effects", async () => {
     const seen = new Set<string>();
     const onMessage = vi.fn(async () => {});
-    const shouldProcessMessage = vi.fn(async (message: { messageId: string }) => {
+    const shouldProcessMessage = vi.fn(async (message: NextcloudTalkInboundMessage) => {
       if (seen.has(message.messageId)) {
         return false;
       }
@@ -73,7 +26,6 @@ describe("createNextcloudTalkWebhookServer replay handling", () => {
       shouldProcessMessage,
       onMessage,
     });
-    cleanupFns.push(harness.stop);
 
     const payload = {
       type: "Create",

extensions/nextcloud-talk/src/monitor.test-harness.ts

@@ -0,0 +1,59 @@
+import { type AddressInfo } from "node:net";
+import { afterEach } from "vitest";
+import { createNextcloudTalkWebhookServer } from "./monitor.js";
+import type { NextcloudTalkWebhookServerOptions } from "./types.js";
+
+export type WebhookHarness = {
+  webhookUrl: string;
+  stop: () => Promise<void>;
+};
+
+const cleanupFns: Array<() => Promise<void>> = [];
+
+afterEach(async () => {
+  while (cleanupFns.length > 0) {
+    const cleanup = cleanupFns.pop();
+    if (cleanup) {
+      await cleanup();
+    }
+  }
+});
+
+export type StartWebhookServerParams = Omit<
+  NextcloudTalkWebhookServerOptions,
+  "port" | "host" | "path" | "secret"
+> & {
+  path: string;
+  secret?: string;
+  host?: string;
+  port?: number;
+};
+
+export async function startWebhookServer(
+  params: StartWebhookServerParams,
+): Promise<WebhookHarness> {
+  const host = params.host ?? "127.0.0.1";
+  const port = params.port ?? 0;
+  const secret = params.secret ?? "nextcloud-secret";
+  const { server, start } = createNextcloudTalkWebhookServer({
+    ...params,
+    port,
+    host,
+    secret,
+  });
+  await start();
+  const address = server.address() as AddressInfo | null;
+  if (!address) {
+    throw new Error("missing server address");
+  }
+
+  const harness: WebhookHarness = {
+    webhookUrl: `http://${host}:${address.port}${params.path}`,
+    stop: () =>
+      new Promise<void>((resolve) => {
+        server.close(() => resolve());
+      }),
+  };
+  cleanupFns.push(harness.stop);
+  return harness;
+}

extensions/nextcloud-talk/src/monitor.ts

@@ -15,6 +15,7 @@ import { extractNextcloudTalkHeaders, verifyNextcloudTalkSignature } from "./sig
 import type {
   CoreConfig,
   NextcloudTalkInboundMessage,
+  NextcloudTalkWebhookHeaders,
   NextcloudTalkWebhookPayload,
   NextcloudTalkWebhookServerOptions,
 } from "./types.js";
@@ -25,6 +26,14 @@ const DEFAULT_WEBHOOK_PATH = "/nextcloud-talk-webhook";
 const DEFAULT_WEBHOOK_MAX_BODY_BYTES = 1024 * 1024;
 const DEFAULT_WEBHOOK_BODY_TIMEOUT_MS = 30_000;
 const HEALTH_PATH = "/healthz";
+const WEBHOOK_ERRORS = {
+  missingSignatureHeaders: "Missing signature headers",
+  invalidBackend: "Invalid backend",
+  invalidSignature: "Invalid signature",
+  invalidPayloadFormat: "Invalid payload format",
+  payloadTooLarge: "Payload too large",
+  internalServerError: "Internal server error",
+} as const;
 
 function formatError(err: unknown): string {
   if (err instanceof Error) {
@@ -61,6 +70,83 @@ function parseWebhookPayload(body: string): NextcloudTalkWebhookPayload | null {
   }
 }
 
+function writeJsonResponse(
+  res: ServerResponse,
+  status: number,
+  body?: Record<string, unknown>,
+): void {
+  if (body) {
+    res.writeHead(status, { "Content-Type": "application/json" });
+    res.end(JSON.stringify(body));
+    return;
+  }
+  res.writeHead(status);
+  res.end();
+}
+
+function writeWebhookError(res: ServerResponse, status: number, error: string): void {
+  if (res.headersSent) {
+    return;
+  }
+  writeJsonResponse(res, status, { error });
+}
+
+function validateWebhookHeaders(params: {
+  req: IncomingMessage;
+  res: ServerResponse;
+  isBackendAllowed?: (backend: string) => boolean;
+}): NextcloudTalkWebhookHeaders | null {
+  const headers = extractNextcloudTalkHeaders(
+    params.req.headers as Record<string, string | string[] | undefined>,
+  );
+  if (!headers) {
+    writeWebhookError(params.res, 400, WEBHOOK_ERRORS.missingSignatureHeaders);
+    return null;
+  }
+  if (params.isBackendAllowed && !params.isBackendAllowed(headers.backend)) {
+    writeWebhookError(params.res, 401, WEBHOOK_ERRORS.invalidBackend);
+    return null;
+  }
+  return headers;
+}
+
+function verifyWebhookSignature(params: {
+  headers: NextcloudTalkWebhookHeaders;
+  body: string;
+  secret: string;
+  res: ServerResponse;
+}): boolean {
+  const isValid = verifyNextcloudTalkSignature({
+    signature: params.headers.signature,
+    random: params.headers.random,
+    body: params.body,
+    secret: params.secret,
+  });
+  if (!isValid) {
+    writeWebhookError(params.res, 401, WEBHOOK_ERRORS.invalidSignature);
+    return false;
+  }
+  return true;
+}
+
+function decodeWebhookCreateMessage(params: {
+  body: string;
+  res: ServerResponse;
+}):
+  | { kind: "message"; message: NextcloudTalkInboundMessage }
+  | { kind: "ignore" }
+  | { kind: "invalid" } {
+  const payload = parseWebhookPayload(params.body);
+  if (!payload) {
+    writeWebhookError(params.res, 400, WEBHOOK_ERRORS.invalidPayloadFormat);
+    return { kind: "invalid" };
+  }
+  if (payload.type !== "Create") {
+    return { kind: "ignore" };
+  }
+  return { kind: "message", message: payloadToInboundMessage(payload) };
+}
+
 function payloadToInboundMessage(
   payload: NextcloudTalkWebhookPayload,
 ): NextcloudTalkInboundMessage {
@@ -120,60 +206,49 @@ export function createNextcloudTalkWebhookServer(opts: NextcloudTalkWebhookServe
     }
 
     try {
-      const headers = extractNextcloudTalkHeaders(
-        req.headers as Record<string, string | string[] | undefined>,
-      );
+      const headers = validateWebhookHeaders({
+        req,
+        res,
+        isBackendAllowed,
+      });
       if (!headers) {
-        res.writeHead(400, { "Content-Type": "application/json" });
-        res.end(JSON.stringify({ error: "Missing signature headers" }));
-        return;
-      }
-      if (isBackendAllowed && !isBackendAllowed(headers.backend)) {
-        res.writeHead(401, { "Content-Type": "application/json" });
-        res.end(JSON.stringify({ error: "Invalid backend" }));
         return;
       }
 
       const body = await readBody(req, maxBodyBytes);
 
-      const isValid = verifyNextcloudTalkSignature({
-        signature: headers.signature,
-        random: headers.random,
+      const hasValidSignature = verifyWebhookSignature({
+        headers,
         body,
         secret,
+        res,
       });
-
-      if (!isValid) {
-        res.writeHead(401, { "Content-Type": "application/json" });
-        res.end(JSON.stringify({ error: "Invalid signature" }));
+      if (!hasValidSignature) {
         return;
       }
 
-      const payload = parseWebhookPayload(body);
-      if (!payload) {
-        res.writeHead(400, { "Content-Type": "application/json" });
-        res.end(JSON.stringify({ error: "Invalid payload format" }));
+      const decoded = decodeWebhookCreateMessage({
+        body,
+        res,
+      });
+      if (decoded.kind === "invalid") {
+        return;
+      }
+      if (decoded.kind === "ignore") {
+        writeJsonResponse(res, 200);
         return;
       }
 
-      if (payload.type !== "Create") {
-        res.writeHead(200);
-        res.end();
-        return;
-      }
-
-      const message = payloadToInboundMessage(payload);
+      const message = decoded.message;
       if (shouldProcessMessage) {
         const shouldProcess = await shouldProcessMessage(message);
         if (!shouldProcess) {
-          res.writeHead(200);
-          res.end();
+          writeJsonResponse(res, 200);
           return;
         }
       }
 
-      res.writeHead(200);
-      res.end();
+      writeJsonResponse(res, 200);
 
       try {
         await onMessage(message);
@@ -182,25 +257,16 @@ export function createNextcloudTalkWebhookServer(opts: NextcloudTalkWebhookServe
       }
     } catch (err) {
       if (isRequestBodyLimitError(err, "PAYLOAD_TOO_LARGE")) {
-        if (!res.headersSent) {
-          res.writeHead(413, { "Content-Type": "application/json" });
-          res.end(JSON.stringify({ error: "Payload too large" }));
-        }
+        writeWebhookError(res, 413, WEBHOOK_ERRORS.payloadTooLarge);
         return;
       }
       if (isRequestBodyLimitError(err, "REQUEST_BODY_TIMEOUT")) {
-        if (!res.headersSent) {
-          res.writeHead(408, { "Content-Type": "application/json" });
-          res.end(JSON.stringify({ error: requestBodyErrorToText("REQUEST_BODY_TIMEOUT") }));
-        }
+        writeWebhookError(res, 408, requestBodyErrorToText("REQUEST_BODY_TIMEOUT"));
         return;
       }
       const error = err instanceof Error ? err : new Error(formatError(err));
       onError?.(error);
-      if (!res.headersSent) {
-        res.writeHead(500, { "Content-Type": "application/json" });
-        res.end(JSON.stringify({ error: "Internal server error" }));
-      }
+      writeWebhookError(res, 500, WEBHOOK_ERRORS.internalServerError);
     }
   });