From 7d56f71c919bbdebb4439d49dc2b5c41fc00cd59 Mon Sep 17 00:00:00 2001 From: Jon Baldie <8376953+jonbaldie@users.noreply.github.com> Date: Tue, 27 Oct 2026 10:02:13 +0000 Subject: [PATCH] feat(logging): extract access log middleware with injectable sinks --- main.ts | 19 ++---- mutation/mutasaurus_ci.ts | 1 + src/handler.ts | 32 +--------- src/middleware.ts | 34 ++++++++++ tests/handler_test.ts | 6 +- tests/logging_test.ts | 130 ++++++++++++++++++++++++++++++++++++++ 6 files changed, 180 insertions(+), 42 deletions(-) create mode 100644 tests/logging_test.ts diff --git a/main.ts b/main.ts index f85fc31..f57b583 100644 --- a/main.ts +++ b/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,23 +53,23 @@ 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 { - 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); } @@ -82,7 +77,7 @@ async function shutdown(signal: string): Promise { PERSIST_ENGINE.close(); - writeLog("Goodbye!"); + DEFAULT_CONSOLE_SINK.write("Goodbye!"); Deno.exit(0); } diff --git a/mutation/mutasaurus_ci.ts b/mutation/mutasaurus_ci.ts index 0acdb60..14a1ba9 100644 --- a/mutation/mutasaurus_ci.ts +++ b/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", diff --git a/src/handler.ts b/src/handler.ts index 7f2d3e8..7b95148 100644 --- a/src/handler.ts +++ b/src/handler.ts @@ -1,6 +1,6 @@ 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"; @@ -8,8 +8,6 @@ type JsonPayload = Payload.Payload; type RouteHandler = Parameters[1]; type RouteMatch = Parameters[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,14 +144,11 @@ function registerRoutes(router: Router, mgr: QueueManager): 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, apiToken: string, rateLimitRequests?: number, + logger?: LogSink, ) { const rateLimiter = new RateLimiter(rateLimitRequests ?? 100); const router = new Router(); @@ -161,26 +156,5 @@ export function createHandler( const handlerWithAuth = withAuth(apiToken)(router.handle); const handlerWithRateLimit = withRateLimit(rateLimiter)(handlerWithAuth); - return async function handler( - request: Request, - info?: Deno.ServeHandlerInfo, - ): Promise { - 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); } diff --git a/src/middleware.ts b/src/middleware.ts index f921150..1eed553 100644 --- a/src/middleware.ts +++ b/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 => { + 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(); diff --git a/tests/handler_test.ts b/tests/handler_test.ts index d0d88b5..44932a8 100644 --- a/tests/handler_test.ts +++ b/tests/handler_test.ts @@ -1,11 +1,15 @@ import { assertEquals, assertNotEquals, assertThrows, assertRejects } from "jsr:@std/assert@1.0"; 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, token: string, rateLimit?: number) { + return createHandlerWithLogging(mgr, token, rateLimit, { write() {} }); +} + // Shared helpers const API_TOKEN = "test-token"; const authHeaders = { "Authorization": `Bearer ${API_TOKEN}` }; diff --git a/tests/logging_test.ts b/tests/logging_test.ts new file mode 100644 index 0000000..4284aa2 --- /dev/null +++ b/tests/logging_test.ts @@ -0,0 +1,130 @@ +import { assertEquals, assertMatch, assertRejects, assertStrictEquals } from "jsr:@std/assert@1.0"; +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(new MemoryStore()); + 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), ""); +});