konnosDashboard
poll-manager.ts210 lines · main
1import { createLogger } from '@briven/shared/observability';
2import pg from 'pg';
3
4import { env } from './env.js';
5import type { SubscriptionRegistry } from './subscription-registry.js';
6
7const log = createLogger({
8 service: 'realtime',
9 env: env.BRIVEN_ENV,
10 level: env.BRIVEN_LOG_LEVEL,
11});
12
13/**
14 * PollManager — DoltGres commit-diff polling engine.
15 *
16 * Replaces Postgres LISTEN/NOTIFY with periodic polling of DoltGres's
17 * commit log. Each active project is polled at a configurable interval
18 * (default 500 ms, configurable via BRIVEN_REALTIME_POLL_MS).
19 *
20 * DoltGres speaks the Postgres wire protocol, so we use the `pg` driver
21 * (node-postgres) — the same driver the converged data plane uses.
22 * postgres.js is deliberately NOT used: its extended-protocol pipelining
23 * desyncs against DoltGres. Postgres cannot switch database mid-connection
24 * (there is no `USE`), so we maintain one `pg` pool per project, each bound
25 * to that project's database (`proj_<id>`) via the `database` option.
26 *
27 * Poll cycle per project:
28 * 1. Resolve the project's `pg` pool (bound to proj_<id>)
29 * 2. Query `SELECT DOLT_HASHOF('HEAD') AS h` to get the current commit
30 * hash — DOLT_HASHOF is DoltGres's real commit-hash function
31 * 3. If the hash changed since last poll, fire every channel for that
32 * project via the provided `onChange` callback
33 *
34 * @README-BRIVEN Phase 2: This replaces the Phase 1 stubs in
35 * `apps/realtime/src/index.ts:startListen` / `stopListen`.
36 * When no projects are active, the interval timer stops to avoid
37 * burning CPU on an empty poll set.
38 */
39export class PollManager {
40 /** Base data-plane URL (a postgres:// DSN). Set by init(). */
41 private baseUrl: string | null = null;
42 /** Per-project `pg` pools, each bound to proj_<id>. */
43 private readonly clients = new Map<string, pg.Pool>();
44 private timer: ReturnType<typeof setInterval> | null = null;
45 private readonly lastHashes = new Map<string, string>();
46 private readonly activeProjects = new Set<string>();
47 private readonly intervalMs: number;
48
49 constructor(
50 /** Registry used to discover which channels belong to a project. */
51 private readonly registry: SubscriptionRegistry,
52 /**
53 * Called for every channel whose project's HEAD changed since the
54 * last poll. The caller is responsible for re-invoking subscriptions
55 * and shipping data frames.
56 */
57 private readonly onChange: (channel: string) => Promise<void>,
58 intervalMs = 500,
59 ) {
60 this.intervalMs = Math.max(100, Math.min(intervalMs, 5000));
61 }
62
63 /**
64 * Record the data-plane DSN. Call once at boot. Per-project clients are
65 * opened lazily (in addProject / on first poll) since each binds to a
66 * different database via the postgres.js `database` option.
67 */
68 async init(dataPlaneUrl: string): Promise<void> {
69 this.baseUrl = dataPlaneUrl;
70 }
71
72 /** Start watching a project for changes. Idempotent. */
73 addProject(projectId: string): void {
74 if (this.activeProjects.has(projectId)) return;
75 this.activeProjects.add(projectId);
76 // Open the project's client eagerly so the first poll doesn't pay
77 // connection setup inline. No-op if init() hasn't run yet.
78 this.clientFor(projectId);
79 this.tryStartPolling();
80 }
81
82 /** Stop watching a project. Called when its last channel is removed. */
83 removeProject(projectId: string): void {
84 this.activeProjects.delete(projectId);
85 this.lastHashes.delete(projectId);
86 const client = this.clients.get(projectId);
87 if (client) {
88 this.clients.delete(projectId);
89 // Fire-and-forget close; a failed teardown must not block removal.
90 client.end().catch(() => undefined);
91 }
92 if (this.activeProjects.size === 0) this.stopPolling();
93 }
94
95 /** True when at least one project is being watched. */
96 get active(): boolean {
97 return this.activeProjects.size > 0;
98 }
99
100 /** Number of projects currently watched. */
101 get projectCount(): number {
102 return this.activeProjects.size;
103 }
104
105 // ── internals ──────────────────────────────────────────────────────
106
107 /**
108 * Resolve (lazily creating) the `pg` pool bound to a project's DoltGres
109 * database. Returns null until init() has supplied the DSN.
110 */
111 private clientFor(projectId: string): pg.Pool | null {
112 if (!this.baseUrl) return null;
113 let client = this.clients.get(projectId);
114 if (!client) {
115 const base = new URL(this.baseUrl);
116 client = new pg.Pool({
117 host: base.hostname,
118 port: Number(base.port || 5432),
119 user: decodeURIComponent(base.username),
120 password: decodeURIComponent(base.password),
121 database: dbNameFor(projectId),
122 max: 2,
123 idleTimeoutMillis: 30000,
124 connectionTimeoutMillis: 5000,
125 });
126 this.clients.set(projectId, client);
127 }
128 return client;
129 }
130
131 private tryStartPolling(): void {
132 if (this.timer || this.activeProjects.size === 0) return;
133 this.timer = setInterval(() => {
134 this.poll().catch(() => {
135 /* errors logged inside poll() */
136 });
137 }, this.intervalMs);
138 // Run an immediate first poll so a fresh subscription gets its
139 // initial hash seeded without waiting for the interval.
140 this.poll().catch(() => undefined);
141 }
142
143 private stopPolling(): void {
144 if (this.timer) {
145 clearInterval(this.timer);
146 this.timer = null;
147 }
148 }
149
150 private async poll(): Promise<void> {
151 if (!this.baseUrl || this.activeProjects.size === 0) return;
152
153 const projects = [...this.activeProjects];
154 for (const projectId of projects) {
155 try {
156 const hash = await this.fetchHeadHash(projectId);
157 if (hash === null) continue;
158 const last = this.lastHashes.get(projectId);
159 if (last !== undefined && hash === last) continue;
160 // First poll: store hash but don't fire (no baseline).
161 // Subsequent polls: hash changed → fire channels.
162 this.lastHashes.set(projectId, hash);
163 if (last !== undefined) {
164 await this.fireProjectChannels(projectId);
165 }
166 } catch (err) {
167 // Per-project failure shouldn't block other projects. Surface it
168 // at warn level so a broken DSN / missing database / DoltGres
169 // outage is visible instead of silently looking like "no live
170 // updates". No secrets or URLs are logged — only the project id
171 // and the error message.
172 log.warn('realtime_poll_failed', {
173 projectId,
174 message: err instanceof Error ? err.message : String(err),
175 });
176 }
177 }
178 }
179
180 private async fetchHeadHash(projectId: string): Promise<string | null> {
181 const pool = this.clientFor(projectId);
182 if (!pool) return null;
183 const { rows } = await pool.query<{ h: string | null }>("SELECT DOLT_HASHOF('HEAD') AS h");
184 return rows[0]?.h ?? null;
185 }
186
187 private async fireProjectChannels(projectId: string): Promise<void> {
188 const channels = this.registry.channelsForProject(projectId);
189 for (const channel of channels) {
190 await this.onChange(channel);
191 }
192 }
193
194 /** Shut down the poll timer and all per-project clients. Graceful-exit hook. */
195 async close(): Promise<void> {
196 this.stopPolling();
197 const clients = [...this.clients.values()];
198 this.clients.clear();
199 await Promise.all(clients.map((c) => c.end().catch(() => undefined)));
200 }
201}
202
203/**
204 * Derive the DoltGres database name for a project id.
205 * Must stay in sync with `apps/api/src/db/data-plane.ts:dbNameFor`.
206 */
207function dbNameFor(projectId: string): string {
208 const safe = projectId.replace(/[^a-zA-Z0-9]/g, '_').toLowerCase();
209 return `proj_${safe}`;
210}