feat: filter tags

.changeset/yummy-breads-roll.md

@@ -0,0 +1,5 @@
+---
+"@toapi/server": minor
+---
+
+add option to filter tags based on the revalidation stream request

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(

packages/toapi-server/src/create-request-handler.ts

@@ -60,6 +60,7 @@ export function createRequestHandler(
       return streamRevalidatedTags({
         cache: api.cache,
         config: api.revalidationStreamConfig,
+        req,
       });
     }
 

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,
   );
 }
 

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,15 +17,15 @@ 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");
   });
 
   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();
+  });
 });

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<typeof setInterval> | null = null;
   let timeout: ReturnType<typeof setTimeout> | 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`),
           );