fix: debounce revalidation stream

.changeset/two-worms-take.md

@@ -0,0 +1,5 @@
+---
+"@toapi/server": patch
+---
+
+debounce and dedupe tags stream

.github/workflows/ci.yml

@@ -53,5 +53,5 @@ jobs:
       - run: pnpm install
       - run: pnpm run lint
       - run: pnpm run build
-      - run: pnpm exec playwright install --with-deps
+      - run: pnpm exec playwright install --with-deps --no-progress
       - run: pnpm run test

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

@@ -59,6 +59,7 @@ export function createRequestHandler(
     if (url.pathname === `${basePath}${INVALIDATIONS_ROUTE}`) {
       return streamRevalidatedTags({
         cache: api.cache,
+        config: api.revalidationStreamConfig,
       });
     }
 

packages/toapi-server/src/define-api.ts

@@ -3,6 +3,7 @@ import type { Logger } from "@toapi/common";
 import type { Path as BasePath, StrictParams } from "@toapi/common";
 import type { Route } from "@toapi/common";
 import { type Cache, PubSub } from "./cache.js";
+import type { RevalidationStreamConfig } from "./revalidation-stream.js";
 
 export interface OasInfo {
   title: string;
@@ -13,10 +14,16 @@ interface Options {
   cache?: Cache;
   oas?: OasInfo;
   logger?: Logger;
+  revalidationStream?: RevalidationStreamConfig;
 }
 
 export function defineApi(options: Options = {}) {
-  return new ApiDefinition({}, options?.cache ?? new PubSub(), options?.oas, options?.logger);
+  return new ApiDefinition(
+    {},
+    options?.cache ?? new PubSub(),
+    options?.oas,
+    options?.logger,
+  );
 }
 
 export class ApiDefinition<Routes extends Record<BasePath, unknown>> {
@@ -25,10 +32,11 @@ export class ApiDefinition<Routes extends Record<BasePath, unknown>> {
     public cache: Cache,
     public oas?: OasInfo,
     public logger?: Logger,
+    public revalidationStreamConfig?: RevalidationStreamConfig,
   ) {}
 
   async invalidate(tags: string[]) {
-    await this.cache.delete(tags)
+    await this.cache.delete(tags);
   }
 
   route<

packages/toapi-server/src/revalidation-stream.ts

@@ -1,22 +1,33 @@
-import {
-  SESSION_COOKIE_NAME,
-  TAGS_CONTENT_TYPE,
-} from "@toapi/common";
+import { SESSION_COOKIE_NAME, TAGS_CONTENT_TYPE } from "@toapi/common";
 import type { Cache } from "./cache.js";
 
-const KEEPALIVE_INTERVAL = 10 * 1000;
+const DEFAULT_KEEPALIVE_INTERVAL = 10 * 1000;
+const DEFAULT_THROTTLE_TIMEOUT = 500;
+
+export interface RevalidationStreamConfig {
+  throttleTimeout?: number;
+  keepaliveInterval?: number;
+}
 
 interface Options {
   cache: Cache;
+  config?: RevalidationStreamConfig;
 }
 
-export function streamRevalidatedTags({ cache }: Options) {
+export function streamRevalidatedTags({ cache, config = {} }: Options) {
+  const {
+    throttleTimeout = DEFAULT_THROTTLE_TIMEOUT,
+    keepaliveInterval = DEFAULT_KEEPALIVE_INTERVAL,
+  } = config;
   const id = crypto.randomUUID();
   let interval: ReturnType<typeof setInterval> | null = null;
+  let timeout: ReturnType<typeof setTimeout> | null = null;
   let unsubscribe = () => {};
   const stream = new ReadableStream({
     async start(controller) {
+      let queue = new Set<string>();
       const textEncoder = new TextEncoder();
+
       // subscribe to tag invalidations
       unsubscribe = cache.subscribe((tags, meta) => {
         // ignore our own invalidations
@@ -27,19 +38,36 @@ export function streamRevalidatedTags({ cache }: Options) {
           meta.clientId === id
         )
           return;
+
+        for (const tag of tags) queue.add(tag);
+
         // send tags to client
-        controller.enqueue(textEncoder.encode(`${tags.join(" ")}\n`));
+        if (!timeout) {
+          controller.enqueue(
+            textEncoder.encode(`${Array.from(queue).join(" ")}\n`),
+          );
+          queue = new Set();
+
+          timeout = setTimeout(() => {
+            controller.enqueue(
+              textEncoder.encode(`${Array.from(queue).join(" ")}\n`),
+            );
+            queue = new Set();
+            timeout = null;
+          }, throttleTimeout);
+        }
       });
 
       // keepalive. The first one is sent right away so the response headers
       // are flushed immediately instead of only once the interval first
       // fires — consumers skip empty lines, so this is a no-op for them.
       const keepalive = () => controller.enqueue(textEncoder.encode("\n"));
       keepalive();
-      interval = setInterval(keepalive, KEEPALIVE_INTERVAL);
+      interval = setInterval(keepalive, keepaliveInterval);
     },
     cancel() {
       if (interval) clearInterval(interval);
+      if (timeout) clearTimeout(timeout);
       unsubscribe();
     },
   });