diff --git a/.changeset/yummy-breads-roll.md b/.changeset/yummy-breads-roll.md new file mode 100644 index 00000000..59aa81d9 --- /dev/null +++ b/.changeset/yummy-breads-roll.md @@ -0,0 +1,5 @@ +--- +"@toapi/server": minor +--- + +add option to filter tags based on the revalidation stream request diff --git a/packages/toapi-server/src/create-request-handler.test.ts b/packages/toapi-server/src/create-request-handler.test.ts index f72fcdd2..983e6145 100644 --- a/packages/toapi-server/src/create-request-handler.test.ts +++ b/packages/toapi-server/src/create-request-handler.test.ts @@ -7,8 +7,8 @@ import { import { defineApi } from "./define-api.js"; import { defineHandler } from "./define-handler.js"; import z from "zod"; -import { TResponse } from "@toapi/common"; -import type { Cache } from "./cache.js"; +import { INVALIDATIONS_ROUTE, TResponse } from "@toapi/common"; +import { type Cache, PubSub } from "./cache.js"; describe("compilePathRegex", () => { test("match a simple route", () => { @@ -67,6 +67,30 @@ describe("compilePathRegex", () => { }); describe("createRequestHandler", () => { + test("applies the configured tag filter to the invalidation stream", async () => { + const cache = new PubSub(); + const request = new Request(`http://localhost:3000${INVALIDATIONS_ROUTE}`, { + headers: { "X-Allowed-Tag": "visible" }, + }); + const filter = vi.fn((req: Request) => (tag: string) => + tag === req.headers.get("X-Allowed-Tag"), + ); + const handler = createRequestHandler( + defineApi({ cache, revalidationStream: { filter } }), + ); + const response = await handler(request); + const reader = response.body!.getReader(); + + await reader.read(); // initial keepalive + await cache.delete(["hidden", "visible"]); + + expect(new TextDecoder().decode((await reader.read()).value)).toBe( + "visible\n", + ); + expect(filter).toHaveBeenCalledWith(request); + await reader.cancel(); + }); + test("returns 500 for arbitrary errors in handler", async () => { const errorHook = vi.fn(); const sut = createRequestHandler( diff --git a/packages/toapi-server/src/create-request-handler.ts b/packages/toapi-server/src/create-request-handler.ts index 3f7c8cc8..37681773 100644 --- a/packages/toapi-server/src/create-request-handler.ts +++ b/packages/toapi-server/src/create-request-handler.ts @@ -60,6 +60,7 @@ export function createRequestHandler( return streamRevalidatedTags({ cache: api.cache, config: api.revalidationStreamConfig, + req, }); } diff --git a/packages/toapi-server/src/define-api.ts b/packages/toapi-server/src/define-api.ts index 1995d5f1..f9ebd328 100644 --- a/packages/toapi-server/src/define-api.ts +++ b/packages/toapi-server/src/define-api.ts @@ -23,6 +23,7 @@ export function defineApi(options: Options = {}) { options?.cache ?? new PubSub(), options?.oas, options?.logger, + options?.revalidationStream, ); } diff --git a/packages/toapi-server/src/revalidation-stream.test.ts b/packages/toapi-server/src/revalidation-stream.test.ts index 1f57a301..f0695caf 100644 --- a/packages/toapi-server/src/revalidation-stream.test.ts +++ b/packages/toapi-server/src/revalidation-stream.test.ts @@ -1,12 +1,14 @@ -import { describe, expect, test } from "vitest"; +import { describe, expect, test, vi } from "vitest"; import { PubSub } from "./cache.js"; import { streamRevalidatedTags } from "./revalidation-stream.js"; import { SESSION_COOKIE_NAME } from "@toapi/common"; describe("revalidation stream", () => { + const req = new Request("http://localhost:3000/invalidations"); + test("should set session cookie", async () => { const cache = new PubSub(); - const response = streamRevalidatedTags({ cache }); + const response = streamRevalidatedTags({ cache, req }); expect( response.headers.get("Set-Cookie")?.startsWith(`${SESSION_COOKIE_NAME}=`) @@ -15,7 +17,7 @@ describe("revalidation stream", () => { test("should flush an initial keepalive so headers are sent immediately", async () => { const cache = new PubSub(); - const response = streamRevalidatedTags({ cache }); + const response = streamRevalidatedTags({ cache, req }); const result = await response.body?.getReader().read(); expect(new TextDecoder().decode(result?.value)).toBe("\n"); @@ -23,7 +25,7 @@ describe("revalidation stream", () => { test("should send revalidated tags", async () => { const cache = new PubSub(); - const response = streamRevalidatedTags({ cache }); + const response = streamRevalidatedTags({ cache, req }); const reader = response.body!.getReader(); // consume the initial keepalive @@ -34,4 +36,29 @@ describe("revalidation stream", () => { const result = await reader.read(); expect(new TextDecoder().decode(result?.value)).toBe("tag1\n"); }); + + test("should filter revalidated tags using the stream request", async () => { + const cache = new PubSub(); + const request = new Request("http://localhost:3000/invalidations", { + headers: { "X-Allowed-Tag": "visible" }, + }); + const filter = vi.fn((req: Request) => (tag: string) => + tag === req.headers.get("X-Allowed-Tag"), + ); + const response = streamRevalidatedTags({ + cache, + req: request, + config: { filter }, + }); + const reader = response.body!.getReader(); + + await reader.read(); // initial keepalive + await cache.delete(["hidden", "visible"]); + + expect(new TextDecoder().decode((await reader.read()).value)).toBe( + "visible\n", + ); + expect(filter).toHaveBeenCalledWith(request); + await reader.cancel(); + }); }); diff --git a/packages/toapi-server/src/revalidation-stream.ts b/packages/toapi-server/src/revalidation-stream.ts index 5491e812..99d280e3 100644 --- a/packages/toapi-server/src/revalidation-stream.ts +++ b/packages/toapi-server/src/revalidation-stream.ts @@ -7,18 +7,22 @@ const DEFAULT_THROTTLE_TIMEOUT = 500; export interface RevalidationStreamConfig { throttleTimeout?: number; keepaliveInterval?: number; + filter?: (req: Request) => (tag: string) => boolean; } interface Options { cache: Cache; config?: RevalidationStreamConfig; + req: Request; } -export function streamRevalidatedTags({ cache, config = {} }: Options) { +export function streamRevalidatedTags({ cache, config = {}, req }: Options) { const { throttleTimeout = DEFAULT_THROTTLE_TIMEOUT, keepaliveInterval = DEFAULT_KEEPALIVE_INTERVAL, + filter = () => () => true, } = config; + const filterTag = filter(req); const id = crypto.randomUUID(); let interval: ReturnType | null = null; let timeout: ReturnType | null = null; @@ -39,10 +43,14 @@ export function streamRevalidatedTags({ cache, config = {} }: Options) { ) return; - for (const tag of tags) queue.add(tag); + for (const tag of tags) { + if (filterTag(tag)) { + queue.add(tag); + } + } // send tags to client - if (!timeout) { + if (queue.size > 0 && !timeout) { controller.enqueue( textEncoder.encode(`${Array.from(queue).join(" ")}\n`), );