fix(metrics): compute true stream backlog via consumer group lag and set retention to 100k

- Cap Redis stream retention at 100,000 messages (down from 10,000,000) in
  src/config/index.ts and .env.example to prevent unbounded memory growth.
- In src/services/api/handlers.ts, update getSystemMetrics to resolve stream
  queue depths using consumer group lag and in-flight pending entries (via
  XINFO GROUPS) instead of raw XLEN. This prevents acknowledged and delivered
  notifications from persisting as ghost backlog in the dashboard.
- Retain XLEN measurement for DEAD_LETTER queue where messages persist until
  replayed or deleted.
- Adjust dashboard capacity scaling to 100,000 items in AnalyticsPage.tsx and
  update backpressure warning threshold to 80,000 in DashboardHeader.tsx.
- Ensure 0-depth queues render empty (0% width) in the backlog gauge meter.
- Update queue test expectations in tests/queue.test.ts and add unit tests
  verifying consumer group lag resolution in tests/api-admin.test.ts.

.env.example

@@ -35,7 +35,7 @@ SEGMENT_MAX_USERS=10000
 # In-flight messages a worker will process concurrently.
 WORKER_CONCURRENCY=10
 # Redis stream trim length. The cap on unconsumed backlog.
-QUEUE_MAX_LEN=10000000
+QUEUE_MAX_LEN=100000
 # Postgres pool size, per process. Multiply by your replica count.
 DB_MAX_CONNECTIONS=2
 

dashboard/src/components/DashboardHeader.tsx

@@ -25,7 +25,7 @@ export default function DashboardHeader({ isConnected }: { isConnected: boolean
           const data = await res.json();
           const streams = data.streams || {};
           const maxDepth = Math.max(...(Object.values(streams) as number[]), 0);
-          setHighBackpressure(maxDepth > 8000);
+          setHighBackpressure(maxDepth > 80000);
         }
       } catch {
         // Ignore background header check errors

dashboard/src/pages/AnalyticsPage.tsx

@@ -169,7 +169,7 @@ export default function AnalyticsPage() {
             <div className="grid grid-cols-1 md:grid-cols-2 lg:grid-cols-3 gap-6">
               {streamNames.map((st) => {
                 const depth = data?.streams?.[st.key] ?? 0;
-                const maxCapacity = 10000;
+                const maxCapacity = 100000;
                 const pct = Math.min(100, Math.round((depth / maxCapacity) * 100));
 
                 return (
@@ -186,12 +186,12 @@ export default function AnalyticsPage() {
                     <div className="w-full h-2.5 rounded-full bg-muted overflow-hidden relative">
                       <div
                         className="h-full bg-primary transition-all duration-300 rounded-full"
-                        style={{ width: `${Math.max(4, pct)}%` }}
+                        style={{ width: `${depth === 0 ? 0 : Math.max(4, pct)}%` }}
                       />
                     </div>
                     <div className="flex justify-between text-[10px] text-muted-foreground">
                       <span>0</span>
-                      <span>Cap: 10,000</span>
+                      <span>Cap: 100,000</span>
                     </div>
                   </div>
                 );

src/config/index.ts

@@ -47,7 +47,7 @@ export const baseConfigSchema = z.object({
   ADMIN_PASSWORD: z.string().trim().min(6).optional(),
   ADMIN_USERNAME: z.string().trim().optional(),
   WORKER_CONCURRENCY: z.coerce.number().int().min(1).default(10),
-  QUEUE_MAX_LEN: z.coerce.number().int().min(1).default(10000000),
+  QUEUE_MAX_LEN: z.coerce.number().int().min(1).default(100000),
   DB_MAX_CONNECTIONS: z.coerce.number().int().min(1).default(2),
   LOG_FLUSH_INTERVAL_MS: z.coerce.number().int().min(50).default(500),
   LOG_BUFFER_MAX_SIZE: z.coerce.number().int().min(100).default(5000),

src/services/api/handlers.ts

@@ -37,6 +37,7 @@ import {
   CreateWorkflowSchema,
   TriggerWorkflowSchema,
   STREAMS,
+  CONSUMER_GROUPS,
   PUBSUB_CHANNELS,
 } from "@/contracts/index.js";
 import { renderWithTemplate, type TemplateCache } from "@/templates/index.js";
@@ -1186,31 +1187,76 @@ export function createHandlers(deps: Deps) {
     });
   }
 
+  // Helper to resolve true unconsumed backlog (consumer group lag + pending entries)
+  async function getStreamBacklog(
+    redis: any,
+    streamKey: string,
+    targetGroup?: string,
+  ): Promise<number> {
+    if (!targetGroup) {
+      return await redis.xlen(streamKey);
+    }
+
+    if (typeof redis.xinfo === "function") {
+      try {
+        const groups = await redis.xinfo("GROUPS", streamKey);
+        if (Array.isArray(groups) && groups.length > 0) {
+          for (const item of groups) {
+            const groupData: Record<string, any> = {};
+            if (Array.isArray(item)) {
+              for (let i = 0; i < item.length; i += 2) {
+                groupData[String(item[i])] = item[i + 1];
+              }
+            } else if (item && typeof item === "object") {
+              Object.assign(groupData, item);
+            }
+
+            if (groupData.name === targetGroup) {
+              const pending = typeof groupData.pending === "number" ? groupData.pending : 0;
+              if (typeof groupData.lag === "number") {
+                return groupData.lag + pending;
+              }
+              if (groupData["last-delivered-id"] === "0-0") {
+                const totalLen = await redis.xlen(streamKey);
+                return totalLen + pending;
+              }
+              return pending;
+            }
+          }
+        }
+      } catch {
+        // Group or stream may not exist yet, fallback to xlen
+      }
+    }
+
+    return await redis.xlen(streamKey);
+  }
+
   // ── GET /v1/system/metrics — getSystemMetrics ────────────────────────────
   async function getSystemMetrics(
     _req: IncomingMessage,
     res: ServerResponse,
     ctx: RouteContext,
   ): Promise<void> {
-    const streamMap: Record<string, string> = {
-      INBOUND_CRITICAL: STREAMS.INBOUND_CRITICAL,
-      INBOUND_NORMAL: STREAMS.INBOUND_NORMAL,
-      INBOUND_LOW: STREAMS.INBOUND_LOW,
-      ENRICHED_NORMAL: STREAMS.ENRICHED_NORMAL,
-      OUTBOUND_CRITICAL: STREAMS.OUTBOUND_CRITICAL,
-      OUTBOUND_NORMAL: STREAMS.OUTBOUND_NORMAL,
-      OUTBOUND_LOW: STREAMS.OUTBOUND_LOW,
-      WORKFLOW_INBOUND: STREAMS.WORKFLOW_INBOUND,
-      EVENTS_INBOUND: STREAMS.EVENTS_INBOUND,
-      DEAD_LETTER: STREAMS.DEAD_LETTER,
+    const streamMap: Record<string, { key: string; group?: string }> = {
+      INBOUND_CRITICAL: { key: STREAMS.INBOUND_CRITICAL, group: CONSUMER_GROUPS.ENRICHER },
+      INBOUND_NORMAL: { key: STREAMS.INBOUND_NORMAL, group: CONSUMER_GROUPS.ENRICHER },
+      INBOUND_LOW: { key: STREAMS.INBOUND_LOW, group: CONSUMER_GROUPS.ENRICHER },
+      ENRICHED_NORMAL: { key: STREAMS.ENRICHED_NORMAL, group: CONSUMER_GROUPS.ENGINE },
+      OUTBOUND_CRITICAL: { key: STREAMS.OUTBOUND_CRITICAL, group: CONSUMER_GROUPS.DELIVERY },
+      OUTBOUND_NORMAL: { key: STREAMS.OUTBOUND_NORMAL, group: CONSUMER_GROUPS.DELIVERY },
+      OUTBOUND_LOW: { key: STREAMS.OUTBOUND_LOW, group: CONSUMER_GROUPS.DELIVERY },
+      WORKFLOW_INBOUND: { key: STREAMS.WORKFLOW_INBOUND, group: CONSUMER_GROUPS.WORKFLOW },
+      EVENTS_INBOUND: { key: STREAMS.EVENTS_INBOUND, group: CONSUMER_GROUPS.EVENTS },
+      DEAD_LETTER: { key: STREAMS.DEAD_LETTER },
     };
 
     const streamDepths: Record<string, number> = {};
     if (ctx.isAdmin !== false) {
-      for (const [key, realRedisKey] of Object.entries(streamMap)) {
+      for (const [key, { key: realRedisKey, group }] of Object.entries(streamMap)) {
         try {
-          const len = await deps.redis.native.xlen(realRedisKey);
-          streamDepths[key] = len;
+          const depth = await getStreamBacklog(deps.redis.native, realRedisKey, group);
+          streamDepths[key] = depth;
         } catch {
           streamDepths[key] = 0;
         }

tests/api-admin.test.ts

@@ -227,6 +227,27 @@ describe("API operational handlers", () => {
       expect(body.deliveryStats.delivered).toBe(10);
       expect(body.deliveryStats.failed).toBe(0);
     });
+
+    it("uses consumer group lag and pending entries when xinfo is available", async () => {
+      selectRows = [[{ count: 0 }], [{ count: 0 }]];
+      deps.redis.native.xinfo = vi.fn().mockImplementation(async (_cmd: string, stream: string) => {
+        if (stream.includes("outbound:critical")) {
+          return [["name", "notifkit:group:delivery", "pending", 0, "lag", 0]];
+        }
+        if (stream.includes("outbound:normal")) {
+          return [["name", "notifkit:group:delivery", "pending", 2, "lag", 5]];
+        }
+        return [];
+      });
+
+      const res = createMockRes();
+      await handlers.getSystemMetrics(createMockReq(), res, ctx());
+
+      const body = parse(res);
+      expect(body.streams.OUTBOUND_CRITICAL).toBe(0); // 0 lag + 0 pending
+      expect(body.streams.OUTBOUND_LOW).toBe(3); // fallback to xlen when xinfo returns empty
+      expect(body.streams.OUTBOUND_NORMAL).toBe(7); // 5 lag + 2 pending
+    });
   });
 
   describe("DLQ", () => {

tests/queue.test.ts

@@ -727,7 +727,7 @@ describe("StreamProducer", () => {
       "my-stream",
       "MAXLEN",
       "~",
-      "10000000",
+      "100000",
       "*",
       "data",
       expect.any(String),