konnosDashboard
index.ts615 lines · main
1import { resolve } from 'node:path';
2
3import { constantTimeEqual, resolveBuildIdentity } from '@briven/shared';
4import { createLogger } from '@briven/shared/observability';
5import { z } from 'zod';
6
7import { env } from './env.js';
8import { incCounter, registerGauge, renderPrometheus } from './metrics.js';
9import { PollManager } from './poll-manager.js';
10import { SubscriptionRegistry } from './subscription-registry.js';
11
12const BOOT_TIME = new Date().toISOString();
13// apps/realtime runs from /app/apps/realtime; the repo root is two
14// levels up.
15const { buildSha: BUILD_SHA, buildAt: BUILD_AT } = resolveBuildIdentity(
16 resolve(process.cwd(), '../../.git'),
17);
18
19const log = createLogger({
20 service: 'realtime',
21 env: env.BRIVEN_ENV,
22 level: env.BRIVEN_LOG_LEVEL,
23});
24
25/**
26 * Reactive WebSocket service.
27 *
28 * Wire protocol (JSON frames over a single WS connection):
29 * client → server:
30 * {type:'subscribe', subscriptionId, projectId, functionName, args}
31 * {type:'unsubscribe', subscriptionId}
32 * server → client:
33 * {type:'hello', protocol:1}
34 * {type:'data', subscriptionId, ok, value | code/message, durationMs}
35 * {type:'unsubscribed', subscriptionId}
36 * {type:'error', code}
37 *
38 * Subscription lifecycle:
39 * 1. Client subscribes → realtime calls apps/api invoke endpoint
40 * 2. Response includes `touchedTables`; realtime watches those tables
41 * for changes (Phase 2: Dolt commit-diff polling)
42 * 3. Change detected → realtime re-invokes every subscription that
43 * touched that table, sends a fresh `data` frame
44 * 4. Unsubscribe / disconnect → drop the subscription, decrement channel
45 * refcounts, stop polling the project when its last channel is removed
46 *
47 * @README-BRIVEN Phase 2: Postgres LISTEN/NOTIFY replaced with DoltGres
48 * commit-diff polling. The PollManager queries `DOLT_HASHOF('HEAD')`
49 * for each active project at the configured interval; when the hash
50 * changes it fires every channel belonging to that project.
51 */
52
53const subscribeSchema = z.object({
54 type: z.literal('subscribe'),
55 subscriptionId: z.string().min(1),
56 projectId: z.string().min(1),
57 functionName: z.string().min(1),
58 args: z.unknown(),
59});
60
61const unsubscribeSchema = z.object({
62 type: z.literal('unsubscribe'),
63 subscriptionId: z.string().min(1),
64});
65
66const clientMessage = z.discriminatedUnion('type', [subscribeSchema, unsubscribeSchema]);
67
68interface Subscription {
69 subscriptionId: string;
70 projectId: string;
71 functionName: string;
72 args: unknown;
73 channels: Set<string>;
74 send: (frame: Record<string, unknown>) => void;
75 // Wall-clock ms at subscription open. Closed-out subs flush their
76 // duration into closedSubSecondsByProject (cumulative billable
77 // seconds-of-connection per project). Active subs contribute to the
78 // pull-based gauge below — at scrape time we add (now - startedAt)
79 // for every still-live sub so the metric never under-counts.
80 startedAt: number;
81}
82
83const subscriptions = new Map<string, Subscription>(); // subscriptionId → sub
84const registry = new SubscriptionRegistry(); // channel ↔ subId, ref-counted
85const sockets = new WeakMap<object, Set<string>>(); // ws → set of subscriptionIds it owns
86// projectId → live sub count. Maintained inline with the subscriptions
87// map so the per-project cap can be enforced in O(1) without scanning.
88const subsByProject = new Map<string, number>();
89
90// Cumulative connection-seconds, per project, from subs that have
91// already closed. Phase 3 usage-metering pillar #3 — Polar metering
92// push reads this via /metrics. Kept in memory (resets on process
93// restart); a durable rollup is the natural follow-up.
94const closedSubSecondsByProject = new Map<string, number>();
95
96/**
97 * Tier-aware per-project subscription cap. Resolved from the api's internal
98 * /v1/internal/projects/:id/limits endpoint on first subscribe per project,
99 * cached for 5 min so tier changes propagate within one window. Returns
100 * null when the project doesn't exist OR the api is unreachable — caller
101 * falls back to the env ceiling so a metadata outage doesn't lock out
102 * legitimate subscriptions.
103 */
104const TIER_CAP_CACHE_TTL_MS = 5 * 60_000;
105interface CapCacheEntry {
106 cap: number | null;
107 expiresAt: number;
108}
109const capByProject = new Map<string, CapCacheEntry>();
110
111async function resolveProjectCap(projectId: string): Promise<number | null> {
112 const now = Date.now();
113 const hit = capByProject.get(projectId);
114 if (hit && hit.expiresAt > now) return hit.cap;
115 if (!env.BRIVEN_RUNTIME_SHARED_SECRET) return null;
116 try {
117 const url = `${env.BRIVEN_API_INTERNAL_URL}/v1/internal/projects/${encodeURIComponent(
118 projectId,
119 )}/limits`;
120 const res = await fetch(url, {
121 headers: { authorization: `Bearer ${env.BRIVEN_RUNTIME_SHARED_SECRET}` },
122 });
123 if (!res.ok) {
124 capByProject.set(projectId, { cap: null, expiresAt: now + TIER_CAP_CACHE_TTL_MS });
125 return null;
126 }
127 const body = (await res.json()) as {
128 limits: { concurrentSubscriptions?: number } | null;
129 };
130 const cap = body.limits?.concurrentSubscriptions ?? null;
131 capByProject.set(projectId, { cap, expiresAt: now + TIER_CAP_CACHE_TTL_MS });
132 return cap;
133 } catch (err) {
134 log.warn('realtime_resolve_cap_failed', {
135 projectId,
136 message: err instanceof Error ? err.message : String(err),
137 });
138 return null;
139 }
140}
141
142function dbNameFor(projectId: string): string {
143 return `proj_${projectId.replace(/[^a-zA-Z0-9]/g, '_').toLowerCase()}`;
144}
145
146function channelFor(projectId: string, table: string): string {
147 return `briven_${dbNameFor(projectId)}_${table}`;
148}
149
150// ── PollManager — Dolt commit-diff polling (Phase 2) ──────────────
151// Replaces Postgres LISTEN/NOTIFY. Each project is polled for HEAD
152// changes; when a change is detected every channel for that project
153// fires via the existing `fireChannel` path.
154
155const pollManager = new PollManager(registry, fireChannel, env.BRIVEN_REALTIME_POLL_MS);
156
157// Per-project refcount — when the first channel is attached for a
158// project we call pollManager.addProject; when the last is detached
159// we call pollManager.removeProject. Keeps the poll set tight.
160const projectRefCount = new Map<string, number>();
161
162function projectIdFromChannel(channel: string): string | null {
163 // channels are `briven_<dbName>_<table>` where dbName = `proj_<sanitizedId>`.
164 // Strip the `briven_` prefix and the trailing `_<table>` suffix to recover
165 // the database name, then strip `proj_` to recover the raw project id.
166 const withoutPrefix = channel.slice('briven_'.length);
167 const lastUnderscore = withoutPrefix.lastIndexOf('_');
168 if (lastUnderscore === -1) return null;
169 const dbName = withoutPrefix.slice(0, lastUnderscore);
170 // dbName is `proj_<sanitized>`. The project id is everything after `proj_`.
171 // Since project ids are sanitised (alphanumeric + underscores only), this is
172 // a reversible mapping (the sanitised id is what's stored in the db name).
173 return dbName.slice('proj_'.length);
174}
175
176async function startListen(channel: string): Promise<void> {
177 const pid = projectIdFromChannel(channel);
178 if (!pid) return;
179 const prev = projectRefCount.get(pid) ?? 0;
180 projectRefCount.set(pid, prev + 1);
181 if (prev === 0) {
182 // First channel for this project — start polling it.
183 pollManager.addProject(pid);
184 log.info('realtime_project_watch_started', { projectId: pid, channel });
185 }
186}
187
188async function stopListen(channel: string): Promise<void> {
189 const pid = projectIdFromChannel(channel);
190 if (!pid) return;
191 const prev = projectRefCount.get(pid) ?? 0;
192 if (prev <= 1) {
193 projectRefCount.delete(pid);
194 pollManager.removeProject(pid);
195 log.info('realtime_project_watch_stopped', { projectId: pid, channel });
196 } else {
197 projectRefCount.set(pid, prev - 1);
198 }
199}
200
201async function fireChannel(channel: string): Promise<void> {
202 // SubscriptionRegistry.subsForChannel returns a snapshot — safe to iterate
203 // while attach/detach inside invokeOnce mutates the channel's live set.
204 const snapshot = registry.subsForChannel(channel);
205 if (snapshot.length === 0) return;
206 incCounter('briven_realtime_notifies_total');
207 for (const subId of snapshot) {
208 const sub = subscriptions.get(subId);
209 if (!sub) continue;
210 const result = await invokeOnce(sub);
211 incCounter('briven_realtime_reinvoke_total', {
212 outcome: (result as { ok?: boolean }).ok ? 'ok' : 'err',
213 });
214 sub.send({ type: 'data', subscriptionId: sub.subscriptionId, ...result });
215 }
216}
217
218async function invokeOnce(sub: Subscription): Promise<Record<string, unknown>> {
219 // why: the public /v1/projects/:id/functions/:name route requires a
220 // session or api-key with developer role. Realtime is a system caller,
221 // not a user — it authenticates as the runtime via the shared secret
222 // and uses the internal invoke endpoint, which threads `auth: null`.
223 const url = `${env.BRIVEN_API_INTERNAL_URL}/v1/internal/projects/${sub.projectId}/functions/${sub.functionName}`;
224 const headers: Record<string, string> = { 'content-type': 'application/json' };
225 if (env.BRIVEN_RUNTIME_SHARED_SECRET) {
226 headers['authorization'] = `Bearer ${env.BRIVEN_RUNTIME_SHARED_SECRET}`;
227 }
228 try {
229 const res = await fetch(url, {
230 method: 'POST',
231 headers,
232 body: JSON.stringify(sub.args ?? {}),
233 });
234 if (!res.ok) {
235 const text = await res.text().catch(() => '');
236 return { ok: false, code: 'invoke_failed', message: text || `http ${res.status}` };
237 }
238 const body = (await res.json()) as {
239 ok: boolean;
240 value?: unknown;
241 code?: string;
242 message?: string;
243 durationMs?: number;
244 touchedTables?: string[];
245 };
246 // Update channel subscriptions to match what the executor actually
247 // touched. Adding new ones is idempotent; removing dropped ones keeps
248 // the LISTEN set tight.
249 const next = new Set((body.touchedTables ?? []).map((t) => channelFor(sub.projectId, t)));
250 for (const ch of sub.channels) {
251 if (!next.has(ch)) await detachSubFromChannel(sub.subscriptionId, ch);
252 }
253 for (const ch of next) {
254 if (!sub.channels.has(ch)) await attachSubToChannel(sub.subscriptionId, ch);
255 }
256 sub.channels = next;
257 return body as unknown as Record<string, unknown>;
258 } catch (err) {
259 return {
260 ok: false,
261 code: 'invoke_error',
262 message: err instanceof Error ? err.message : 'unknown',
263 };
264 }
265}
266
267async function attachSubToChannel(subId: string, channel: string): Promise<void> {
268 if (registry.attach(subId, channel)) await startListen(channel);
269}
270
271async function detachSubFromChannel(subId: string, channel: string): Promise<void> {
272 if (registry.detach(subId, channel)) await stopListen(channel);
273}
274
275async function dropSubscription(subId: string): Promise<void> {
276 const sub = subscriptions.get(subId);
277 if (!sub) return;
278 for (const ch of sub.channels) await detachSubFromChannel(subId, ch);
279 // Flush this sub's connection seconds into the per-project tally so
280 // the metric survives the sub teardown. (Math.max guards against
281 // a clock skew producing a negative — postgres NTP is solid here
282 // but the realtime container's clock isn't authoritative.)
283 const seconds = Math.max(0, (Date.now() - sub.startedAt) / 1000);
284 closedSubSecondsByProject.set(
285 sub.projectId,
286 (closedSubSecondsByProject.get(sub.projectId) ?? 0) + seconds,
287 );
288 // Decrement the per-project counter. Map entry deleted at 0 so the
289 // map doesn't grow unbounded with dormant projects.
290 const next = (subsByProject.get(sub.projectId) ?? 0) - 1;
291 if (next <= 0) subsByProject.delete(sub.projectId);
292 else subsByProject.set(sub.projectId, next);
293 subscriptions.delete(subId);
294}
295
296/**
297 * Authorise an incoming WS upgrade. Refuses ALL requests when the shared
298 * secret is unset — prior behaviour returned `true` (fail-open), which
299 * combined with deployments that forgot to set the env var produced an
300 * unauthenticated WebSocket service. apps/api/src/routes/internal.ts
301 * already does the right thing (503 when secret unset); this matches.
302 */
303/**
304 * Extract the bearer token from a WS upgrade request. Accepts the Authorization
305 * header (server-to-server) OR a `?token=` / `?bearer=` query param — browsers
306 * CANNOT set headers on a WebSocket, so the SDK passes the token in the URL.
307 */
308function extractToken(req: Request): string | null {
309 const auth = req.headers.get('authorization');
310 const headerToken = auth?.startsWith('Bearer ') ? auth.slice('Bearer '.length).trim() : null;
311 if (headerToken) return headerToken;
312 try {
313 const p = new URL(req.url).searchParams;
314 return p.get('token') ?? p.get('bearer');
315 } catch {
316 return null;
317 }
318}
319
320/**
321 * Authorise an incoming WS upgrade. Accepts either the global shared secret
322 * (trusted server-to-server callers) or a structurally-valid project key
323 * (`brk_…`). A project key's validity FOR A SPECIFIC PROJECT is verified at
324 * SUBSCRIBE time (see verifyProjectKey) — the projectId isn't known until the
325 * subscribe frame arrives. Refuses ALL requests when the shared secret is unset
326 * (fail-closed), matching apps/api/src/routes/internal.ts.
327 */
328function authorise(req: Request): boolean {
329 if (!env.BRIVEN_RUNTIME_SHARED_SECRET) return false;
330 const token = extractToken(req);
331 if (!token) return false;
332 if (constantTimeEqual(token, env.BRIVEN_RUNTIME_SHARED_SECRET)) return true;
333 return token.startsWith('brk_');
334}
335
336/**
337 * Verify a project key (`brk_…`) is valid FOR `projectId` by asking apps/api
338 * (which owns the api-key table) over the shared-secret internal channel.
339 * Cached briefly so a reconnect storm doesn't hammer the control plane.
340 */
341const keyVerifyCache = new Map<string, { ok: boolean; exp: number }>();
342async function verifyProjectKey(projectId: string, token: string): Promise<boolean> {
343 if (!token.startsWith('brk_')) return false;
344 const cacheKey = `${projectId}:${token}`;
345 const now = Date.now();
346 const hit = keyVerifyCache.get(cacheKey);
347 if (hit && hit.exp > now) return hit.ok;
348 let ok = false;
349 try {
350 const res = await fetch(
351 `${env.BRIVEN_API_INTERNAL_URL}/v1/internal/projects/${projectId}/verify-key`,
352 {
353 method: 'POST',
354 headers: {
355 'content-type': 'application/json',
356 ...(env.BRIVEN_RUNTIME_SHARED_SECRET
357 ? { authorization: `Bearer ${env.BRIVEN_RUNTIME_SHARED_SECRET}` }
358 : {}),
359 },
360 body: JSON.stringify({ token }),
361 },
362 );
363 if (res.ok) ok = ((await res.json()) as { valid?: boolean }).valid === true;
364 } catch {
365 ok = false;
366 }
367 // Short positive TTL (keys can be revoked); shorter negative TTL.
368 keyVerifyCache.set(cacheKey, { ok, exp: now + (ok ? 60_000 : 5_000) });
369 return ok;
370}
371
372if (!env.BRIVEN_RUNTIME_SHARED_SECRET) {
373 log.warn(
374 'realtime_boot_warning: BRIVEN_RUNTIME_SHARED_SECRET is unset — every WS upgrade will be rejected with 401 until configured',
375 );
376}
377
378// Pull-based gauges — snapshot at scrape time. Keeps the subscribe /
379// unsubscribe hot paths free of gauge bookkeeping.
380registerGauge('briven_realtime_subscriptions_active', () => [
381 { labels: {}, value: subscriptions.size },
382]);
383registerGauge('briven_realtime_channels_active', () => [
384 { labels: {}, value: registry.channelCount },
385]);
386
387// Per-project cumulative connection-seconds. The closed-sub tally is the
388// authoritative base; we add (now - startedAt) for every still-live sub
389// so a scrape during a long-running subscription doesn't show 0. Polar
390// metering reads this via /metrics on the aggregation cron in apps/api.
391registerGauge('briven_realtime_connection_seconds_total', () => {
392 const out: { labels: { project: string }; value: number }[] = [];
393 const now = Date.now();
394 const liveByProject = new Map<string, number>();
395 for (const sub of subscriptions.values()) {
396 liveByProject.set(
397 sub.projectId,
398 (liveByProject.get(sub.projectId) ?? 0) + (now - sub.startedAt) / 1000,
399 );
400 }
401 const projects = new Set<string>([
402 ...closedSubSecondsByProject.keys(),
403 ...liveByProject.keys(),
404 ]);
405 for (const projectId of projects) {
406 const value =
407 (closedSubSecondsByProject.get(projectId) ?? 0) + (liveByProject.get(projectId) ?? 0);
408 out.push({ labels: { project: projectId }, value });
409 }
410 return out;
411});
412
413// Record the data-plane DSN so the first subscription doesn't block on
414// setup. When BRIVEN_DATA_PLANE_URL is unset the init is a no-op;
415// PollManager opens per-project clients lazily and handles the absent
416// DSN gracefully (polling stays disabled).
417if (env.BRIVEN_DATA_PLANE_URL) {
418 pollManager.init(env.BRIVEN_DATA_PLANE_URL).catch((err) => {
419 log.error('realtime_poll_manager_init_failed', {
420 message: err instanceof Error ? err.message : String(err),
421 });
422 });
423}
424
425log.info('realtime_boot', {
426 port: env.BRIVEN_REALTIME_PORT,
427 apiUrl: env.BRIVEN_API_INTERNAL_URL,
428 auth: env.BRIVEN_RUNTIME_SHARED_SECRET ? 'shared_secret' : 'rejecting_all',
429 poll: env.BRIVEN_DATA_PLANE_URL ? 'enabled' : 'disabled',
430 pollIntervalMs: env.BRIVEN_REALTIME_POLL_MS,
431 phase: 2, // @README-BRIVEN Phase 2: commit-diff polling live
432});
433
434interface SocketHandle {
435 send: (data: string) => void;
436}
437
438export default {
439 port: env.BRIVEN_REALTIME_PORT,
440 fetch(req: Request, server: { upgrade: (req: Request, options?: { data?: unknown }) => boolean }) {
441 const url = new URL(req.url);
442 if (url.pathname === '/health') return Response.json({ status: 'ok', service: 'realtime' });
443 if (url.pathname === '/ready') {
444 return Response.json({
445 status: env.BRIVEN_DATA_PLANE_URL ? 'ready' : 'degraded',
446 poll: env.BRIVEN_DATA_PLANE_URL ? 'enabled' : 'disabled',
447 pollIntervalMs: env.BRIVEN_REALTIME_POLL_MS,
448 activeProjects: pollManager.projectCount,
449 });
450 }
451 if (url.pathname === '/info') {
452 return Response.json({
453 service: 'realtime',
454 env: env.BRIVEN_ENV,
455 buildSha: BUILD_SHA,
456 buildAt: BUILD_AT,
457 bootedAt: BOOT_TIME,
458 uptimeSec: Math.floor(process.uptime()),
459 });
460 }
461 if (url.pathname === '/metrics') {
462 return new Response(renderPrometheus(), {
463 status: 200,
464 headers: { 'content-type': 'text/plain; version=0.0.4; charset=utf-8' },
465 });
466 }
467 if (url.pathname === '/v1/realtime/stats') {
468 // Operator-facing live snapshot. Shared-secret gated — same gate
469 // as /v1/subscribe so an unauth scrape can't enumerate project
470 // ids. Numbers are O(subscriptions.size), fine for the year-one
471 // 10k target; if it ever needs to scale further the natural
472 // aggregation is a periodic write to a stats endpoint on the api.
473 if (!authorise(req)) return Response.json({ code: 'unauthorized' }, { status: 401 });
474 const byProject: { projectId: string; subscriptions: number }[] = [];
475 for (const [projectId, count] of subsByProject) {
476 byProject.push({ projectId, subscriptions: count });
477 }
478 byProject.sort((a, b) => b.subscriptions - a.subscriptions);
479 const byChannel = registry.channelCounts();
480 byChannel.sort((a, b) => b.subscriptions - a.subscriptions);
481 return Response.json({
482 totalSubscriptions: subscriptions.size,
483 totalChannels: registry.channelCount,
484 limits: {
485 perWs: env.BRIVEN_REALTIME_MAX_SUBS_PER_WS,
486 perProject: env.BRIVEN_REALTIME_MAX_SUBS_PER_PROJECT,
487 },
488 // byProject returns every active project — at year-one scale
489 // (~25 projects) the response stays under a kilobyte; we rely on
490 // sort-descending so dashboards can render top-N client-side.
491 // byChannel keeps its top-50 clamp since channel cardinality can
492 // explode with many tables × many subscribers.
493 byProject,
494 byChannel: byChannel.slice(0, 50),
495 });
496 }
497 if (url.pathname === '/v1/subscribe') {
498 if (!authorise(req)) return Response.json({ code: 'unauthorized' }, { status: 401 });
499 // Carry the upgrade token onto the socket so the subscribe handler can do
500 // the per-project authorization check once the projectId is known.
501 if (server.upgrade(req, { data: { token: extractToken(req) } })) return undefined;
502 return new Response('upgrade required', { status: 426 });
503 }
504 return Response.json({ code: 'not_found' }, { status: 404 });
505 },
506 websocket: {
507 open(ws: SocketHandle) {
508 sockets.set(ws as unknown as object, new Set<string>());
509 ws.send(JSON.stringify({ type: 'hello', protocol: 1 }));
510 },
511 async message(ws: SocketHandle, raw: string | Buffer) {
512 const text = typeof raw === 'string' ? raw : raw.toString('utf8');
513 let parsed: z.infer<typeof clientMessage>;
514 try {
515 parsed = clientMessage.parse(JSON.parse(text));
516 } catch {
517 ws.send(JSON.stringify({ type: 'error', code: 'malformed_message' }));
518 return;
519 }
520
521 const owned = sockets.get(ws as unknown as object);
522 if (!owned) return;
523
524 if (parsed.type === 'unsubscribe') {
525 owned.delete(parsed.subscriptionId);
526 await dropSubscription(parsed.subscriptionId);
527 ws.send(JSON.stringify({ type: 'unsubscribed', subscriptionId: parsed.subscriptionId }));
528 return;
529 }
530
531 // Per-project authorization. A connection authed with the global shared
532 // secret is a trusted server-to-server caller and may subscribe to any
533 // project. A connection authed with a project key (brk_…) may ONLY
534 // subscribe to the project that key belongs to — verified against
535 // apps/api. Without this, one browser token could read another tenant's
536 // data (the subscribe frame's projectId is otherwise unchecked).
537 const connToken = (ws as unknown as { data?: { token?: string } }).data?.token ?? null;
538 const trusted =
539 !!connToken &&
540 !!env.BRIVEN_RUNTIME_SHARED_SECRET &&
541 constantTimeEqual(connToken, env.BRIVEN_RUNTIME_SHARED_SECRET);
542 if (!trusted) {
543 const allowed = connToken ? await verifyProjectKey(parsed.projectId, connToken) : false;
544 if (!allowed) {
545 incCounter('briven_realtime_subscribe_rejected_total', { reason: 'forbidden_project' });
546 ws.send(
547 JSON.stringify({
548 type: 'error',
549 code: 'forbidden',
550 subscriptionId: parsed.subscriptionId,
551 }),
552 );
553 return;
554 }
555 }
556
557 if (owned.size >= env.BRIVEN_REALTIME_MAX_SUBS_PER_WS) {
558 // Defensive cap so a single client (bug or malicious) can't open
559 // unbounded subscriptions and degrade the service for everyone.
560 // The client sees an error frame referencing the offending sub
561 // id so it can correlate; the server-side counter has already
562 // ignored this sub, so unsubscribe isn't required.
563 incCounter('briven_realtime_subscribe_rejected_total', { reason: 'ws_limit' });
564 ws.send(
565 JSON.stringify({
566 type: 'error',
567 code: 'subscription_limit_ws',
568 subscriptionId: parsed.subscriptionId,
569 limit: env.BRIVEN_REALTIME_MAX_SUBS_PER_WS,
570 }),
571 );
572 return;
573 }
574 const projectCount = subsByProject.get(parsed.projectId) ?? 0;
575 // Tier-aware cap when the api answers; env ceiling as the floor so
576 // a metadata outage doesn't lock out legitimate Team customers.
577 const tierCap = await resolveProjectCap(parsed.projectId);
578 const projectLimit = tierCap ?? env.BRIVEN_REALTIME_MAX_SUBS_PER_PROJECT;
579 if (projectCount >= projectLimit) {
580 incCounter('briven_realtime_subscribe_rejected_total', { reason: 'project_limit' });
581 ws.send(
582 JSON.stringify({
583 type: 'error',
584 code: 'subscription_limit_project',
585 subscriptionId: parsed.subscriptionId,
586 limit: projectLimit,
587 source: tierCap !== null ? 'tier' : 'env',
588 }),
589 );
590 return;
591 }
592
593 const sub: Subscription = {
594 subscriptionId: parsed.subscriptionId,
595 projectId: parsed.projectId,
596 functionName: parsed.functionName,
597 args: parsed.args,
598 channels: new Set<string>(),
599 send: (frame) => ws.send(JSON.stringify(frame)),
600 startedAt: Date.now(),
601 };
602 subscriptions.set(sub.subscriptionId, sub);
603 owned.add(sub.subscriptionId);
604 subsByProject.set(sub.projectId, projectCount + 1);
605 const result = await invokeOnce(sub);
606 ws.send(JSON.stringify({ type: 'data', subscriptionId: sub.subscriptionId, ...result }));
607 },
608 async close(ws: object) {
609 const owned = sockets.get(ws);
610 if (!owned) return;
611 for (const subId of owned) await dropSubscription(subId);
612 sockets.delete(ws);
613 },
614 },
615};