konnosDashboard
logs.ts140 lines · main
1import type Redis from 'ioredis';
2import { Hono } from 'hono';
3import { streamSSE } from 'hono/streaming';
4
5import { getRedis } from '../lib/redis.js';
6import { log } from '../lib/logger.js';
7import { requireProjectAuth } from '../middleware/project-auth.js';
8import type { ProjectAppEnv as AppEnv } from '../types/app-env.js';
9
10const MAX_CONCURRENT_PER_PROJECT = 10;
11const HEARTBEAT_MS = 15_000;
12
13export const logsRouter = new Hono<AppEnv>();
14
15logsRouter.use('/v1/projects/:id/logs/*', requireProjectAuth());
16
17/**
18 * GET /v1/projects/:id/logs/stream?since=<id>&n=<N>
19 *
20 * Server-Sent Events stream of function invocation envelopes. Opens a
21 * dedicated Redis connection per caller so XREAD BLOCK doesn't starve
22 * the shared command pool. Concurrency is capped per project via a
23 * counter held on the *shared* Redis to enforce the limit across
24 * api replicas, not just this process.
25 */
26logsRouter.get('/v1/projects/:id/logs/stream', async (c) => {
27 const projectId = c.req.param('id');
28 const since = c.req.query('since');
29 const replayN = Number(c.req.query('n') ?? 0);
30
31 const sharedRedis = getRedis();
32 if (!sharedRedis) {
33 return c.json({ code: 'not_configured', message: 'redis is not configured' }, 503);
34 }
35
36 const concurrentKey = `logs:subscribers:${projectId}`;
37 const current = await sharedRedis.incr(concurrentKey);
38 if (current === 1) await sharedRedis.pexpire(concurrentKey, 60_000);
39 if (current > MAX_CONCURRENT_PER_PROJECT) {
40 await sharedRedis.decr(concurrentKey);
41 return c.json(
42 { code: 'too_many_subscribers', message: 'concurrent log streams exceeded' },
43 429,
44 );
45 }
46
47 const streamKey = `logs:${projectId}`;
48 // Dedicated duplicate connection so blocking XREAD doesn't block any
49 // other command on the shared ioredis pool.
50 const blockingRedis = sharedRedis.duplicate();
51
52 return streamSSE(c, async (stream) => {
53 let closed = false;
54 const release = async (): Promise<void> => {
55 if (closed) return;
56 closed = true;
57 try {
58 await sharedRedis.decr(concurrentKey);
59 } catch {
60 // non-fatal
61 }
62 blockingRedis.disconnect();
63 };
64
65 stream.onAbort(() => {
66 void release();
67 });
68
69 try {
70 // Optional replay-before-follow window.
71 let cursor = since ?? '0-0';
72 if (!since && replayN > 0) {
73 const tail = await readLastN(sharedRedis, streamKey, replayN);
74 for (const msg of tail) {
75 await stream.writeSSE({ data: JSON.stringify(msg.fields), id: msg.id });
76 cursor = msg.id;
77 }
78 } else if (since) {
79 // Range replay from `since` inclusive, up to the live head.
80 const ranged = await sharedRedis.xrange(streamKey, since, '+');
81 for (const [id, fields] of ranged) {
82 const parsed = parseFields(fields);
83 await stream.writeSSE({ data: JSON.stringify(parsed), id });
84 cursor = id;
85 }
86 }
87
88 // Live follow loop.
89 while (!closed) {
90 const reply = (await blockingRedis.xread(
91 'BLOCK',
92 HEARTBEAT_MS,
93 'STREAMS',
94 streamKey,
95 cursor,
96 )) as Array<[string, Array<[string, string[]]>]> | null;
97
98 if (!reply) {
99 // Heartbeat so proxies don't sever the connection.
100 await stream.writeSSE({ event: 'ping', data: '' });
101 continue;
102 }
103 for (const [, entries] of reply) {
104 for (const [id, fields] of entries) {
105 const parsed = parseFields(fields);
106 await stream.writeSSE({ data: JSON.stringify(parsed), id });
107 cursor = id;
108 }
109 }
110 }
111 } catch (err) {
112 log.warn('logs_stream_error', {
113 projectId,
114 message: err instanceof Error ? err.message : String(err),
115 });
116 } finally {
117 await release();
118 }
119 });
120});
121
122interface ParsedMessage {
123 id: string;
124 fields: Record<string, string>;
125}
126
127async function readLastN(redis: Redis, key: string, n: number): Promise<ParsedMessage[]> {
128 const rev = await redis.xrevrange(key, '+', '-', 'COUNT', n);
129 return rev.map(([id, fields]) => ({ id, fields: parseFields(fields) })).reverse();
130}
131
132function parseFields(flat: string[]): Record<string, string> {
133 const out: Record<string, string> = {};
134 for (let i = 0; i < flat.length; i += 2) {
135 const k = flat[i];
136 const v = flat[i + 1];
137 if (typeof k === 'string' && typeof v === 'string') out[k] = v;
138 }
139 return out;
140}