feat(logging): extract access log middleware with injectable sinks

main.ts

@@ -3,12 +3,7 @@ import QueueManager from "./src/manager.ts";
 import { createHandler } from "./src/handler.ts";
 import { ENV_VAR_NAMES, parseConfig } from "./src/config.ts";
 import type { Payload } from "./src/payload.ts";
-
-const LOG_ENCODER = new TextEncoder();
-
-function writeLog(message: string): void {
-    Deno.stdout.writeSync(LOG_ENCODER.encode(`${message}\n`));
-}
+import { DEFAULT_CONSOLE_SINK } from "./src/middleware.ts";
 
 // Read each config var by name rather than Deno.env.toObject(), which
 // enumerates the entire process environment and therefore requires
@@ -47,7 +42,7 @@ if (PERSIST_ENGINE instanceof Persistency.FileStore) {
         Deno.exit(1);
     }
 
-    writeLog("Loading in data from persist.dat...\n");
+    DEFAULT_CONSOLE_SINK.write("Loading in data from persist.dat...\n");
 
     MANAGER.load();
 }
@@ -58,31 +53,31 @@ const HANDLER = createHandler(MANAGER, CONFIG.apiToken, CONFIG.rateLimitRequests
 const SERVER = Deno.serve({
     hostname: CONFIG.host,
     port: CONFIG.port,
-    onListen: ({ hostname, port }) => writeLog(`Listening on ${hostname}:${port}`),
+    onListen: ({ hostname, port }) => DEFAULT_CONSOLE_SINK.write(`Listening on ${hostname}:${port}`),
 }, HANDLER);
 
 async function shutdown(signal: string): Promise<void> {
-    writeLog(`Received ${signal}, shutting down gracefully...`);
+    DEFAULT_CONSOLE_SINK.write(`Received ${signal}, shutting down gracefully...`);
 
     // Stop accepting new connections
     await SERVER.shutdown();
 
     // If we're using file persistency, save all current state to persistant storage
     if (PERSIST_ENGINE instanceof Persistency.FileStore) {
-        writeLog("Flushing data to persist.dat...\n");
+        DEFAULT_CONSOLE_SINK.write("Flushing data to persist.dat...\n");
         try {
             MANAGER.save();
         } catch (error) {
             // The previous persist.dat is left in place; it still holds every logged item.
-            writeLog(`Failed to flush data to persist.dat: ${error instanceof Error ? error.message : error}`);
+            DEFAULT_CONSOLE_SINK.write(`Failed to flush data to persist.dat: ${error instanceof Error ? error.message : error}`);
             PERSIST_ENGINE.close();
             Deno.exit(1);
         }
     }
 
     PERSIST_ENGINE.close();
 
-    writeLog("Goodbye!");
+    DEFAULT_CONSOLE_SINK.write("Goodbye!");
 
     Deno.exit(0);
 }

mutation/mutasaurus_ci.ts

@@ -41,6 +41,7 @@ const mutasaurus = new Mutasaurus({
     "./tests/config_test.ts",
     "./tests/e2e_test.ts",
     "./tests/handler_test.ts",
+    "./tests/logging_test.ts",
     "./tests/manager_test.ts",
     "./tests/payload_test.ts",
     "./tests/persist_test.ts",

src/handler.ts

@@ -1,15 +1,13 @@
 import QueueManager, { QueueNameTooLongError } from "./manager.ts";
 import { RateLimiter } from "./rate_limiter.ts";
-import { withAuth, withRateLimit } from "./middleware.ts";
+import { type LogSink, withAuth, withLogging, withRateLimit } from "./middleware.ts";
 import { Router } from "./router.ts";
 import * as Payload from "./payload.ts";
 
 type JsonPayload = Payload.Payload;
 type RouteHandler = Parameters<Router["get"]>[1];
 type RouteMatch = Parameters<RouteHandler>[1];
 
-const LOG_ENCODER = Reflect.construct(TextEncoder, []);
-
 function extractQueueName(match: RouteMatch): { name: string } | { error: Response } {
     const raw = match.pathname.groups.queue;
     if (raw === undefined) {
@@ -146,41 +144,17 @@ function registerRoutes(router: Router, mgr: QueueManager<JsonPayload>): void {
     router.get("/length/:queue", lengthHandler(mgr));
 }
 
-function writeLog(destination: { writeSync(data: Uint8Array): number }, message: string): void {
-    destination.writeSync(LOG_ENCODER.encode(`${message}\n`));
-}
-
 export function createHandler(
     mgr: QueueManager<JsonPayload>,
     apiToken: string,
     rateLimitRequests?: number,
+    logger?: LogSink,
 ) {
     const rateLimiter = new RateLimiter(rateLimitRequests ?? 100);
     const router = new Router();
     registerRoutes(router, mgr);
     const handlerWithAuth = withAuth(apiToken)(router.handle);
     const handlerWithRateLimit = withRateLimit(rateLimiter)(handlerWithAuth);
 
-    return async function handler(
-        request: Request,
-        info?: Deno.ServeHandlerInfo,
-    ): Promise<Response> {
-        const start = performance.now();
-        try {
-            const response = await handlerWithRateLimit(request, info);
-            const duration = performance.now() - start;
-            writeLog(
-                Deno.stdout,
-                `${request.method} ${request.url} ${response.status} ${duration.toFixed(2)}ms`,
-            );
-            return response;
-        } catch (error) {
-            const duration = performance.now() - start;
-            writeLog(
-                Deno.stderr,
-                `${request.method} ${request.url} 500 ${duration.toFixed(2)}ms`,
-            );
-            throw error;
-        }
-    };
+    return withLogging(logger)(handlerWithRateLimit);
 }

src/middleware.ts

@@ -4,6 +4,40 @@ export type HttpHandler = (request: Request, info?: Deno.ServeHandlerInfo) => Pr
 export type Middleware = (next: HttpHandler) => HttpHandler;
 
 const HEALTH_PATTERN = new URLPattern({ pathname: "/health{/}?" });
+const LOG_ENCODER = new TextEncoder();
+
+export interface LogSink {
+    write(message: string, isError?: boolean): void;
+}
+
+export const DEFAULT_CONSOLE_SINK: LogSink = {
+    write(message: string, isError = false): void {
+        const destination = isError ? Deno.stderr : Deno.stdout;
+        destination.writeSync(LOG_ENCODER.encode(`${message}\n`));
+    },
+};
+
+export function withLogging(sink: LogSink = DEFAULT_CONSOLE_SINK) {
+    return (next: HttpHandler) => async (request: Request, info?: Deno.ServeHandlerInfo): Promise<Response> => {
+        const start = performance.now();
+        try {
+            const response = await next(request, info);
+            const duration = performance.now() - start;
+            sink.write(
+                `${request.method} ${request.url} ${response.status} ${duration.toFixed(2)}ms`,
+                false,
+            );
+            return response;
+        } catch (error) {
+            const duration = performance.now() - start;
+            sink.write(
+                `${request.method} ${request.url} 500 ${duration.toFixed(2)}ms`,
+                true,
+            );
+            throw error;
+        }
+    };
+}
 
 export function withAuth(apiToken: string): Middleware {
     const trimmedToken = apiToken?.trim();

tests/handler_test.ts

@@ -1,11 +1,15 @@
 import { assertEquals, assertNotEquals, assertThrows, assertRejects } from "jsr:@std/[email protected]";
 import QueueManager, { QueueNameTooLongError } from "../src/manager.ts";
-import { createHandler } from "../src/handler.ts";
+import { createHandler as createHandlerWithLogging } from "../src/handler.ts";
 import * as Persistency from "../src/persist.ts";
 import { RateLimiter } from "../src/rate_limiter.ts";
 import { parseConfig, ConfigError } from "../src/config.ts";
 import type { Payload } from "../src/payload.ts";
 
+function createHandler(mgr: QueueManager<Payload>, token: string, rateLimit?: number) {
+    return createHandlerWithLogging(mgr, token, rateLimit, { write() {} });
+}
+
 // Shared helpers
 const API_TOKEN = "test-token";
 const authHeaders = { "Authorization": `Bearer ${API_TOKEN}` };

tests/logging_test.ts

@@ -0,0 +1,130 @@
+import { assertEquals, assertMatch, assertRejects, assertStrictEquals } from "jsr:@std/[email protected]";
+import { createHandler } from "../src/handler.ts";
+import QueueManager from "../src/manager.ts";
+import { MemoryStore } from "../src/persist.ts";
+import type { Payload } from "../src/payload.ts";
+import { withLogging } from "../src/middleware.ts";
+
+Deno.test("logging: a custom sink records health, auth, and rate-limit responses", async () => {
+    const records: { message: string; isError: boolean | undefined }[] = [];
+    const sink = {
+        write(message: string, isError?: boolean) {
+            records.push({ message, isError });
+        },
+    };
+    const manager = new QueueManager<Payload>(new MemoryStore<Payload>());
+    const handler = createHandler(manager, "token", 1, sink);
+    const requests = [
+        new Request("http://localhost/health"),
+        new Request("http://localhost/queues"),
+        new Request("http://localhost/queues", {
+            headers: { Authorization: "Bearer token" },
+        }),
+        new Request("http://localhost/health/"),
+    ];
+    const statuses = [];
+    for (const request of requests) {
+        const response = await handler(request);
+        statuses.push(response.status);
+        await response.body?.cancel();
+    }
+    assertEquals(statuses, [200, 401, 429, 200]);
+    assertEquals(records.map(({ message, isError }) => {
+        assertMatch(message, / \d+\.\d{2}ms$/);
+        return { message: message.replace(/ \d+\.\d{2}ms$/, ""), isError };
+    }), [
+        { message: "GET http://localhost/health 200", isError: false },
+        { message: "GET http://localhost/queues 401", isError: false },
+        { message: "GET http://localhost/queues 429", isError: false },
+        { message: "GET http://localhost/health/ 200", isError: false },
+    ]);
+});
+
+Deno.test("logging: middleware forwards request and server info and preserves the response", async () => {
+    const messages: string[] = [];
+    const request = new Request("http://localhost/enqueue/jobs", { method: "POST" });
+    const info: Deno.ServeHandlerInfo = {
+        remoteAddr: { transport: "tcp", hostname: "192.0.2.1", port: 1234 },
+        completed: Promise.resolve(),
+    };
+    const response = new Response("created", { status: 201, headers: { "X-Queue": "jobs" } });
+    const handler = withLogging({
+        write(message, isError) {
+            assertEquals(isError, false);
+            messages.push(message);
+        },
+    })((receivedRequest, receivedInfo) => {
+        assertStrictEquals(receivedRequest, request);
+        assertStrictEquals(receivedInfo, info);
+        return response;
+    });
+    assertStrictEquals(await handler(request, info), response);
+    assertEquals(await response.text(), "created");
+    assertEquals(response.headers.get("X-Queue"), "jobs");
+    assertEquals(messages.length, 1);
+    assertMatch(messages[0], /^POST http:\/\/localhost\/enqueue\/jobs 201 \d+\.\d{2}ms$/);
+});
+
+Deno.test("logging: synchronous throws and async rejections log 500 and retain the error", async () => {
+    const failure = new Error("queue unavailable");
+    for (const next of [() => { throw failure; }, async () => { throw failure; }]) {
+        const records: { message: string; isError: boolean | undefined }[] = [];
+        const handler = withLogging({
+            write(message, isError) {
+                records.push({ message, isError });
+            },
+        })(next);
+        const caught = await assertRejects(
+            async () => await handler(new Request("http://localhost/queues")),
+            Error,
+            "queue unavailable",
+        );
+        assertStrictEquals(caught, failure);
+        assertEquals(records.length, 1);
+        assertEquals(records[0].isError, true);
+        assertMatch(records[0].message, /^GET http:\/\/localhost\/queues 500 \d+\.\d{2}ms$/);
+    }
+});
+
+Deno.test("logging: default sink writes UTF-8 lines to stdout and errors to stderr", async () => {
+    const moduleUrl = new URL("../src/middleware.ts", import.meta.url).href;
+    const result = await new Deno.Command(Deno.execPath(), {
+        args: ["eval", `
+            import { DEFAULT_CONSOLE_SINK, withLogging } from ${JSON.stringify(moduleUrl)};
+            DEFAULT_CONSOLE_SINK.write("Queue café");
+            DEFAULT_CONSOLE_SINK.write("Queue failed", true);
+            const request = new Request("http://localhost/queues");
+            await withLogging()(() => new Response("failed", { status: 500 }))(request);
+            try {
+                await withLogging()(() => { throw new Error("unavailable"); })(request);
+            } catch (error) {
+                if (error.message !== "unavailable") throw error;
+            }
+        `],
+    }).output();
+    assertEquals(result.code, 0);
+    const decoder = new TextDecoder();
+    assertMatch(decoder.decode(result.stdout), /^Queue café\nGET http:\/\/localhost\/queues 500 \d+\.\d{2}ms\n$/);
+    assertMatch(decoder.decode(result.stderr), /^Queue failed\nGET http:\/\/localhost\/queues 500 \d+\.\d{2}ms\n$/);
+});
+
+Deno.test("logging: a silent sink leaves process streams empty and HTTP responses intact", async () => {
+    const handlerUrl = new URL("../src/handler.ts", import.meta.url).href;
+    const managerUrl = new URL("../src/manager.ts", import.meta.url).href;
+    const persistUrl = new URL("../src/persist.ts", import.meta.url).href;
+    const result = await new Deno.Command(Deno.execPath(), {
+        args: ["eval", `
+            import { createHandler } from ${JSON.stringify(handlerUrl)};
+            import QueueManager from ${JSON.stringify(managerUrl)};
+            import { MemoryStore } from ${JSON.stringify(persistUrl)};
+            const handler = createHandler(new QueueManager(new MemoryStore()), "token", 100, { write() {} });
+            const response = await handler(new Request("http://localhost/health"));
+            if (response.status !== 200 || (await response.json()).status !== "ok") {
+                throw new Error("Health response changed");
+            }
+        `],
+    }).output();
+    assertEquals(result.code, 0);
+    assertEquals(new TextDecoder().decode(result.stdout), "");
+    assertEquals(new TextDecoder().decode(result.stderr), "");
+});