konnosDashboard
poll-manager.integration.test.ts121 lines · main
1/**
2 * Per-table change scoping — Phase 4.1 (the whole point of the phase).
3 *
4 * Proves the REAL PollManager, against a live DoltGres, fires ONLY the
5 * channels whose table actually changed — a write to table_a must NOT wake
6 * a subscriber watching table_b. Before Phase 4 every commit re-invoked
7 * every subscription in the project; this test locks in the cure so a
8 * regression fails CI.
9 *
10 * How to run (needs a DoltGres container; e.g. the throwaway probe on :5456):
11 * BRIVEN_DATA_PLANE_URL=postgres://postgres:password@127.0.0.1:5456/postgres \
12 * bun test src/poll-manager.integration.test.ts
13 *
14 * Skips entirely when BRIVEN_DATA_PLANE_URL is unset (keeps the no-DB unit run green).
15 */
16import { afterAll, beforeAll, describe, expect, test } from 'bun:test';
17import pg from 'pg';
18
19import { PollManager } from './poll-manager.js';
20import { SubscriptionRegistry } from './subscription-registry.js';
21
22const DSN = process.env.BRIVEN_DATA_PLANE_URL;
23const HAS_DB = Boolean(DSN);
24
25// Throwaway project id, unique per run. Sanitised form must match dbNameFor.
26const PROJECT_ID = `p4s${Date.now().toString(36)}`;
27function dbNameFor(id: string): string {
28 return `proj_${id.replace(/[^a-zA-Z0-9]/g, '_').toLowerCase()}`;
29}
30function channelFor(id: string, table: string): string {
31 return `briven_${dbNameFor(id)}_${table}`;
32}
33const DB_NAME = dbNameFor(PROJECT_ID);
34const CH_A = channelFor(PROJECT_ID, 'table_a');
35const CH_B = channelFor(PROJECT_ID, 'table_b');
36
37const AUTHOR = "--author=probe <p@p.io>";
38const sleep = (ms: number) => new Promise((r) => setTimeout(r, ms));
39
40function basePool(): pg.Pool {
41 const u = new URL(DSN as string);
42 return new pg.Pool({
43 host: u.hostname,
44 port: Number(u.port || 5432),
45 user: decodeURIComponent(u.username),
46 password: decodeURIComponent(u.password),
47 database: u.pathname.replace(/^\//, '') || 'postgres',
48 max: 1,
49 });
50}
51function projPool(): pg.Pool {
52 const u = new URL(DSN as string);
53 return new pg.Pool({
54 host: u.hostname,
55 port: Number(u.port || 5432),
56 user: decodeURIComponent(u.username),
57 password: decodeURIComponent(u.password),
58 database: DB_NAME,
59 max: 1,
60 });
61}
62
63describe.skipIf(!HAS_DB)('PollManager per-table scoping', () => {
64 let manager: PollManager | null = null;
65
66 beforeAll(async () => {
67 // Provision a throwaway DoltGres database with two tables + a baseline commit.
68 const base = basePool();
69 await base.query(`DROP DATABASE IF EXISTS ${DB_NAME}`).catch(() => {});
70 await base.query(`CREATE DATABASE ${DB_NAME}`);
71 await base.end();
72
73 const proj = projPool();
74 await proj.query('CREATE TABLE IF NOT EXISTS table_a (id int primary key, v text)');
75 await proj.query('CREATE TABLE IF NOT EXISTS table_b (id int primary key, v text)');
76 await proj.query(`SELECT DOLT_COMMIT('-A', '-m', 'baseline', '${AUTHOR}')`);
77 await proj.end();
78 });
79
80 afterAll(async () => {
81 if (manager) await manager.close().catch(() => {});
82 const base = basePool();
83 await base.query(`DROP DATABASE IF EXISTS ${DB_NAME}`).catch(() => {});
84 await base.end();
85 });
86
87 test('a commit touching only table_a fires CH_A and NOT CH_B', async () => {
88 const fired: string[] = [];
89 const registry = new SubscriptionRegistry();
90 // Two subscribers: one on table_a, one on table_b.
91 registry.attach('sub-a', CH_A);
92 registry.attach('sub-b', CH_B);
93
94 manager = new PollManager(
95 registry,
96 async (channel) => {
97 fired.push(channel);
98 },
99 150, // fast poll for the test (clamped floor is 100ms)
100 );
101 await manager.init(DSN as string);
102 manager.addProject(PROJECT_ID);
103
104 // Let the first poll seed the baseline HEAD hash (no fire — no baseline yet).
105 await sleep(600);
106 expect(fired).toEqual([]); // nothing fired before any change
107
108 // Commit a change to table_a ONLY.
109 const proj = projPool();
110 await proj.query("INSERT INTO table_a (id, v) VALUES (1, 'x')");
111 await proj.query(`SELECT DOLT_COMMIT('-A', '-m', 'touch a', '${AUTHOR}')`);
112 await proj.end();
113
114 // Let the next poll detect + scope + fire.
115 await sleep(800);
116
117 // CH_A must have fired; CH_B must NOT — that is per-table scoping.
118 expect(fired).toContain(CH_A);
119 expect(fired).not.toContain(CH_B);
120 }, 15_000);
121});