feat(server): write analytics to the storage event sink

.changeset/server-analytics-sink.md

@@ -0,0 +1,5 @@
+---
+"@chatcops/server": minor
+---
+
+With `analytics: true` and a `storage` that has an event sink, the server writes analytics events there too, buffered with `BufferedEventSink` and flushed before each reply's stream ends, so serverless runtimes don't drop them. `chatcopsAnalyticsHandler` then counts from the store, so the stats cover every instance, for `?from=` and `?to=` in milliseconds (default: the last 30 days). The handler's new `analyticsStats(range)` does the same; `getAnalytics()` still describes this process only.

packages/server/src/adapters/express.ts

@@ -107,11 +107,15 @@ function constantTimeEqual(a: string, b: string): boolean {
   return mismatch === 0;
 }
 
+/**
+ * Analytics stats as JSON. With a storage that has an event sink, they cover
+ * every instance, for `?from=` and `?to=` (milliseconds) or the last 30 days.
+ */
 export function chatcopsAnalyticsHandler(config: ChatCopsServerConfig) {
-  const { getAnalytics } = getOrCreateChatHandler(config);
+  const { analyticsStats } = getOrCreateChatHandler(config);
   let warned = false;
 
-  return (req: Request, res: Response) => {
+  return async (req: Request, res: Response) => {
     if (config.analyticsToken) {
       const auth = req.headers?.authorization;
       const token = typeof auth === 'string' && auth.startsWith('Bearer ') ? auth.slice(7) : undefined;
@@ -128,6 +132,24 @@ export function chatcopsAnalyticsHandler(config: ChatCopsServerConfig) {
       );
     }
 
-    res.json(getAnalytics());
+    const from = timeParam(req.query?.from);
+    const to = timeParam(req.query?.to);
+    if (from === null || to === null) {
+      res.status(400).json({ error: 'invalid_range' });
+      return;
+    }
+    try {
+      res.json(await analyticsStats({ from, to }));
+    } catch (err) {
+      console.error('[chatcops] Reading analytics failed:', err instanceof Error ? err.message : String(err));
+      res.status(500).json({ error: 'internal_error' });
+    }
   };
 }
+
+/** A `from` or `to` query value: unset, a time in milliseconds, or `null` when it isn't one. */
+function timeParam(value: unknown): number | undefined | null {
+  if (value === undefined) return undefined;
+  const time = typeof value === 'string' && value.trim() !== '' ? Number(value) : Number.NaN;
+  return Number.isFinite(time) ? time : null;
+}

packages/server/src/handler.ts

@@ -3,6 +3,9 @@ import {
   isAIProvider,
   ConversationManager,
   AnalyticsCollector,
+  BufferedEventSink,
+  statsFromSink,
+  type AnalyticsStats,
   createBudgetGuard,
   withDefaultBudgets,
   createRateLimiter,
@@ -74,7 +77,19 @@ export function createChatHandler(config: ChatCopsServerConfig) {
     limit: rateLimit.maxRequests,
     windowMs: rateLimit.windowMs,
   });
-  const analytics = config.analytics ? new AnalyticsCollector() : null;
+  // Events always count in this process, and go to the storage's event sink
+  // too when it has one, so stats can cover every instance.
+  const collector = config.analytics ? new AnalyticsCollector() : null;
+  const eventSink = config.analytics ? storage?.events : undefined;
+  const bufferedEvents = eventSink ? new BufferedEventSink(eventSink) : undefined;
+  const analytics = collector
+    ? {
+        track(type: string, data?: Record<string, unknown>) {
+          collector.track(type, data);
+          bufferedEvents?.track(type, data);
+        },
+      }
+    : null;
   const webhooks = config.webhooks?.length ? new WebhookDispatcher(config.webhooks) : null;
   const toolDefs = config.tools?.map(toolToDefinition) ?? [];
   // A broken tool definition is a code mistake, so fail here, at startup,
@@ -159,9 +174,25 @@ export function createChatHandler(config: ChatCopsServerConfig) {
     } finally {
       await release();
     }
-    // Keeping the stream open until lead webhooks finish keeps serverless
-    // runtimes from cutting them off once the response ends.
+    // Keeping the stream open until lead webhooks and analytics writes
+    // finish keeps serverless runtimes from cutting them off once the
+    // response ends. A failed analytics write is logged, never thrown.
     await awaitPendingWebhooks(pendingWebhooks);
+    await bufferedEvents?.flush();
+  }
+
+  /**
+   * Analytics stats for `range` (default: the last 30 days), counted in the
+   * storage's event sink when it can be queried, so they cover every
+   * instance. Otherwise they come from this process only, as
+   * `getAnalytics()` does. `null` when `analytics` is off.
+   */
+  async function analyticsStats(range?: { from?: number; to?: number }): Promise<AnalyticsStats | null> {
+    if (!collector) return null;
+    if (!bufferedEvents || !eventSink?.query) return collector.getStats();
+    await bufferedEvents.flush();
+    const to = range?.to ?? Date.now() + 1;
+    return statsFromSink(eventSink, { from: range?.from ?? to - 30 * DAY_MS, to });
   }
 
   /**
@@ -391,7 +422,9 @@ export function createChatHandler(config: ChatCopsServerConfig) {
       ? (body: unknown, origin: string | null, clientIp?: string) => sessions.mint(body, origin, clientIp)
       : undefined,
     protocolStats: () => ({ ...requests }),
-    getAnalytics: () => analytics?.getStats() ?? null,
+    /** Stats from this process only. `analyticsStats()` reads the storage's event sink. */
+    getAnalytics: () => collector?.getStats() ?? null,
+    analyticsStats,
     cleanup,
   };
 }

packages/server/tests/adapters/express.test.ts

@@ -355,14 +355,14 @@ describe('express analytics sharing', () => {
     }
 
     const { fake: statsFake, res: statsRes } = createResponse();
-    chatcopsAnalyticsHandler(config)(createRequest('GET'), statsRes);
+    await chatcopsAnalyticsHandler(config)(createRequest('GET'), statsRes);
 
     expect(statsFake.body).toMatchObject({ totalConversations: 2 });
     // One handler for the config, shared by the middleware and the stats route.
     expect(createChatHandler).toHaveBeenCalledTimes(1);
   });
 
-  it('warns once across several requests when analyticsToken is not configured', () => {
+  it('warns once across several requests when analyticsToken is not configured', async () => {
     const warnSpy = vi.spyOn(console, 'warn').mockImplementation(() => {});
     const config: ChatCopsServerConfig = {
       provider: { type: 'claude', apiKey: 'test-key' },
@@ -372,11 +372,11 @@ describe('express analytics sharing', () => {
     const handler = chatcopsAnalyticsHandler(config);
 
     const { fake: firstFake, res: firstRes } = createResponse();
-    handler(createRequest('GET'), firstRes);
+    await handler(createRequest('GET'), firstRes);
     expect(firstFake.statusCode).toBe(200);
 
     const { fake: secondFake, res: secondRes } = createResponse();
-    handler(createRequest('GET'), secondRes);
+    await handler(createRequest('GET'), secondRes);
     expect(secondFake.statusCode).toBe(200);
 
     expect(warnSpy).toHaveBeenCalledTimes(1);
@@ -385,7 +385,7 @@ describe('express analytics sharing', () => {
     );
   });
 
-  it('requires a bearer token when analyticsToken is configured', () => {
+  it('requires a bearer token when analyticsToken is configured', async () => {
     const warnSpy = vi.spyOn(console, 'warn').mockImplementation(() => {});
     const config: ChatCopsServerConfig = {
       provider: { type: 'claude', apiKey: 'test-key' },
@@ -397,28 +397,28 @@ describe('express analytics sharing', () => {
 
     // Missing token
     const { fake: missingFake, res: missingRes } = createResponse();
-    handler(createRequest('GET'), missingRes);
+    await handler(createRequest('GET'), missingRes);
     expect(missingFake.statusCode).toBe(401);
     expect(missingFake.headers['WWW-Authenticate']).toBe('Bearer');
     expect(missingFake.body).toEqual({ error: 'unauthorized' });
 
     // Wrong token of different length
     const { fake: diffLengthFake, res: diffLengthRes } = createResponse();
-    handler(createRequest('GET', undefined, { authorization: 'Bearer short' }), diffLengthRes);
+    await handler(createRequest('GET', undefined, { authorization: 'Bearer short' }), diffLengthRes);
     expect(diffLengthFake.statusCode).toBe(401);
     expect(diffLengthFake.headers['WWW-Authenticate']).toBe('Bearer');
     expect(diffLengthFake.body).toEqual({ error: 'unauthorized' });
 
     // Wrong token of same length
     const { fake: wrongFake, res: wrongRes } = createResponse();
-    handler(createRequest('GET', undefined, { authorization: 'Bearer secret-tokeX' }), wrongRes);
+    await handler(createRequest('GET', undefined, { authorization: 'Bearer secret-tokeX' }), wrongRes);
     expect(wrongFake.statusCode).toBe(401);
     expect(wrongFake.headers['WWW-Authenticate']).toBe('Bearer');
     expect(wrongFake.body).toEqual({ error: 'unauthorized' });
 
     // Right token
     const { fake: rightFake, res: rightRes } = createResponse();
-    handler(createRequest('GET', undefined, { authorization: 'Bearer secret-token' }), rightRes);
+    await handler(createRequest('GET', undefined, { authorization: 'Bearer secret-token' }), rightRes);
     expect(rightFake.statusCode).toBe(200);
     expect(rightFake.body).toMatchObject({ totalConversations: 0 });
 

packages/server/tests/analytics-sink.test.ts

@@ -0,0 +1,151 @@
+import { afterEach, describe, expect, it, vi } from 'vitest';
+import type { Request as ExpressRequest, Response as ExpressResponse } from 'express';
+import { memoryStorage, type AnalyticsEvent, type ChatCopsStorage, type EventSink } from '@chatcops/core';
+import { scriptedProvider, textStep } from '@chatcops/core/testing/provider';
+import { createChatHandler } from '../src/handler.js';
+import { chatcopsAnalyticsHandler } from '../src/adapters/express.js';
+import type { ChatCopsServerConfig } from '../src/config.js';
+
+function config(storage: ChatCopsStorage, overrides: Partial<ChatCopsServerConfig> = {}): ChatCopsServerConfig {
+  return {
+    provider: scriptedProvider(Array.from({ length: 5 }, () => textStep('Hello'))),
+    systemPrompt: 'Test',
+    budgets: false,
+    analytics: true,
+    storage,
+    ...overrides,
+  };
+}
+
+/** Sends one message and reads the reply to the end. */
+async function chat(handler: ReturnType<typeof createChatHandler>, conversationId: string) {
+  for await (const _ of handler.handleChat({ conversationId, message: 'Hi' })) {
+    // drain
+  }
+}
+
+const DAY = 86_400_000;
+
+afterEach(() => {
+  vi.restoreAllMocks();
+});
+
+describe('analytics on the storage event sink', () => {
+  it('writes the events by the time the reply ends, and counts them across instances', async () => {
+    const storage = memoryStorage();
+    // Two server instances sharing one store, as on serverless.
+    const first = createChatHandler(config(storage));
+    const second = createChatHandler(config(storage));
+
+    await chat(first, 'c1');
+    await chat(second, 'c2');
+
+    const now = Date.now();
+    const written = await storage.events?.query?.({ from: now - DAY, to: now + 1 });
+    expect(written).toEqual(
+      expect.arrayContaining([
+        expect.objectContaining({ type: 'conversation:started', count: 2 }),
+        expect.objectContaining({ type: 'message:sent', count: 2 }),
+        expect.objectContaining({ type: 'message:received', count: 2 }),
+      ])
+    );
+
+    expect(await first.analyticsStats()).toMatchObject({ totalConversations: 2, totalMessages: 4 });
+    // getAnalytics() still describes this process only.
+    expect(first.getAnalytics()).toMatchObject({ totalConversations: 1, totalMessages: 2 });
+  });
+
+  it('counts only the range asked for', async () => {
+    const storage = memoryStorage();
+    const handler = createChatHandler(config(storage));
+    await chat(handler, 'c1');
+
+    const now = Date.now();
+    expect(await handler.analyticsStats({ from: now + DAY, to: now + 2 * DAY })).toMatchObject({ totalConversations: 0 });
+    expect(await handler.analyticsStats({ from: now - DAY })).toMatchObject({ totalConversations: 1 });
+  });
+
+  it('falls back to this process for a sink that can only be written to', async () => {
+    const write = vi.fn(async (_events: AnalyticsEvent[]) => undefined);
+    const events: EventSink = { write };
+    const handler = createChatHandler(config({ ...memoryStorage(), events }));
+    await chat(handler, 'c1');
+
+    expect(write).toHaveBeenCalled();
+    expect(write.mock.calls.flatMap(([batch]) => batch.map((event) => event.type))).toEqual(['conversation:started', 'message:sent', 'message:received']);
+    expect(await handler.analyticsStats()).toMatchObject({ totalConversations: 1 });
+  });
+
+  it('still answers when the sink fails, and logs it', async () => {
+    const error = vi.spyOn(console, 'error').mockImplementation(() => undefined);
+    const events: EventSink = { write: async () => Promise.reject(new Error('db down')) };
+    const handler = createChatHandler(config({ ...memoryStorage(), events }));
+
+    const chunks: string[] = [];
+    for await (const chunk of handler.handleChat({ conversationId: 'c1', message: 'Hi' })) chunks.push(chunk);
+
+    expect(chunks.join('')).toContain('Hello');
+    expect(error).toHaveBeenCalledWith('[chatcops] analytics write failed: db down');
+  });
+
+  it('writes nothing with analytics off', async () => {
+    const storage = memoryStorage();
+    const write = vi.spyOn(storage.events as EventSink, 'write');
+    const handler = createChatHandler(config(storage, { analytics: false }));
+    await chat(handler, 'c1');
+
+    expect(write).not.toHaveBeenCalled();
+    expect(await handler.analyticsStats()).toBeNull();
+  });
+});
+
+describe('the Express analytics route', () => {
+  function response() {
+    const res = {
+      statusCode: 200,
+      body: undefined as unknown,
+      setHeader: vi.fn(),
+      status(code: number) {
+        this.statusCode = code;
+        return this;
+      },
+      json(value: unknown) {
+        this.body = value;
+      },
+    };
+    return res;
+  }
+  const get = (query: Record<string, string> = {}) =>
+    ({ method: 'GET', headers: { authorization: 'Bearer token' }, query }) as unknown as ExpressRequest;
+
+  it('reads the sink for the range in the query', async () => {
+    const storage = memoryStorage();
+    const serverConfig = config(storage, { analyticsToken: 'token' });
+    await chat(createChatHandler(serverConfig), 'other-instance');
+
+    const res = response();
+    await chatcopsAnalyticsHandler(serverConfig)(get({ from: String(Date.now() - DAY) }), res as unknown as ExpressResponse);
+    expect(res.statusCode).toBe(200);
+    expect(res.body).toMatchObject({ totalConversations: 1 });
+
+    const empty = response();
+    await chatcopsAnalyticsHandler(serverConfig)(get({ from: String(Date.now() + DAY) }), empty as unknown as ExpressResponse);
+    expect(empty.body).toMatchObject({ totalConversations: 0 });
+  });
+
+  it('refuses a range that is not a time in milliseconds', async () => {
+    const res = response();
+    await chatcopsAnalyticsHandler(config(memoryStorage(), { analyticsToken: 'token' }))(get({ from: 'yesterday' }), res as unknown as ExpressResponse);
+    expect(res.statusCode).toBe(400);
+    expect(res.body).toEqual({ error: 'invalid_range' });
+  });
+
+  it('answers 500 when the sink cannot be read', async () => {
+    vi.spyOn(console, 'error').mockImplementation(() => undefined);
+    const events: EventSink = { write: async () => undefined, query: async () => Promise.reject(new Error('db down')) };
+    const res = response();
+    await chatcopsAnalyticsHandler(config({ ...memoryStorage(), events }, { analyticsToken: 'token' }))(get(), res as unknown as ExpressResponse);
+    expect(res.statusCode).toBe(500);
+    expect(res.body).toEqual({ error: 'internal_error' });
+  });
+});

website/src/content/docs/core/analytics.mdx

@@ -17,7 +17,7 @@ chatcopsMiddleware({
 
 ## Reading Stats with Express
 
-`chatcopsAnalyticsHandler` returns the stats for the traffic handled by `chatcopsMiddleware`. Pass the **same config object** to both so they share one handler and one `AnalyticsCollector`:
+`chatcopsAnalyticsHandler` returns the stats for the traffic handled by `chatcopsMiddleware`. Pass the **same config object** to both so they share one handler:
 
 ```typescript
 import express from 'express';
@@ -34,7 +34,7 @@ const config = {
 const app = express();
 app.use(express.json());
 
-app.all('/chat', chatcopsMiddleware(config));
+app.use('/chat', chatcopsMiddleware(config));
 app.get('/analytics', chatcopsAnalyticsHandler(config));
 
 app.listen(3001);
@@ -48,6 +48,14 @@ curl -H "Authorization: Bearer <token>" http://localhost:3001/analytics
 
 When unset, the endpoint is public and logs a warning the first time it is hit.
 
+With a `storage` that has an event sink (`memoryStorage()`, Postgres, Redis, SQLite and Cloudflare stores all do), the server writes every event there too, and the endpoint counts from the store. Then the stats cover every server instance, not just the one answering. Pick the range with `from` and `to` in milliseconds. The default is the last 30 days:
+
+```bash
+curl -H "Authorization: Bearer <token>" "http://localhost:3001/analytics?from=1792368000000"
+```
+
+Without a store that can be queried, the stats come from the memory of the process that answers, which on serverless is one instance at a time. A range that isn't a number gets `400 {"error":"invalid_range"}`.
+
 `GET /analytics` responds with the aggregated stats (or `null` if `analytics` is not enabled):
 
 ```json
@@ -119,7 +127,7 @@ interface AnalyticsEvent {
 
 ## Storage
 
-The built-in collector stores events in memory (max 1000 events). To write events to a storage-backed `EventSink`, use `BufferedEventSink`. It batches up to 50 events by default and flushes every 2 seconds. Call `flush()` when you need to wait for queued writes, such as before a process exits:
+The built-in collector stores events in memory (max 1000 events). With `analytics: true` and a `storage`, the server already writes events to the store's sink, buffered and flushed as each reply ends. To do the same in your own code, use `BufferedEventSink`. It batches up to 50 events by default and flushes every 2 seconds. Call `flush()` when you need to wait for queued writes, such as before a process exits:
 
 ```typescript
 import { BufferedEventSink, memoryStorage, statsFromSink } from '@chatcops/core';