diff --git a/.github/workflows/run-tests.yaml b/.github/workflows/run-tests.yaml index b6d993594..ea4db4a11 100644 --- a/.github/workflows/run-tests.yaml +++ b/.github/workflows/run-tests.yaml @@ -247,7 +247,7 @@ jobs: - batch: pg-graphql packages: 'graphile/graphile-search graphile/graphile-ltree graphile/graphile-bulk-mutations graphile/graphile-function-bindings graphile/graphile-history graphile/graphile-meta graphile/graphile-schema graphql/orm-test graphql/test graphql/playwright-test' - batch: pg-graphile-extras - packages: 'graphile/graphile-i18n graphile/graphile-pg-aggregates graphile/graphile-query' + packages: 'graphile/graphile-i18n graphile/graphile-pg-aggregates graphile/graphile-query graphile/graphile-realtime-test' env: PGHOST: localhost diff --git a/graphile/graphile-realtime-subscriptions/__tests__/plugin.test.ts b/graphile/graphile-realtime-subscriptions/__tests__/plugin.test.ts index 723560853..7f6669bbf 100644 --- a/graphile/graphile-realtime-subscriptions/__tests__/plugin.test.ts +++ b/graphile/graphile-realtime-subscriptions/__tests__/plugin.test.ts @@ -53,10 +53,13 @@ jest.mock('graphile-utils', () => ({ gql: jest.fn((strings: TemplateStringsArray) => strings.join('')), })); +import type { EventGateOptions } from '../src/event-gate'; import { + createGatedSubscriber, createRealtimeSubscriptionsPlugin, DEFAULT_OVERFLOW_THRESHOLD, EventThrottle, + MalformedNotifyPayloadError, parseNotifyPayload, RealtimeSubscriptionsPlugin, } from '../src/plugin'; @@ -142,31 +145,20 @@ describe('parseNotifyPayload', () => { }); }); - it('handles payload with no colon as bare event', () => { - const result = parseNotifyPayload('INSERT'); - expect(result).toEqual({ - event: 'INSERT', - rowIds: [], - overflow: false, - }); + // A payload this cannot read means emit_change and this plugin have + // diverged. Inventing an event name for it hides a deployment fault behind + // data the client acts on, so every unreadable shape is a hard failure. + it.each([ + ['a payload with no colon', 'INSERT'], + ['an empty payload', ''], + ['an operation with no row ids', 'INSERT:'], + ['an unknown operation', 'TRUNCATE:abc-123'], + ])('throws on %s', (_label, raw) => { + expect(() => parseNotifyPayload(raw)).toThrow(MalformedNotifyPayloadError); }); - it('handles empty string as UNKNOWN', () => { - const result = parseNotifyPayload(''); - expect(result).toEqual({ - event: 'UNKNOWN', - rowIds: [], - overflow: false, - }); - }); - - it('handles operation with empty ID list', () => { - const result = parseNotifyPayload('INSERT:'); - expect(result).toEqual({ - event: 'INSERT', - rowIds: [], - overflow: false, - }); + it('names the offending payload in the error', () => { + expect(() => parseNotifyPayload('TRUNCATE:abc')).toThrow(/"TRUNCATE:abc"/); }); }); @@ -500,39 +492,13 @@ describe('createRealtimeSubscriptionsPlugin', () => { }); const result = capturedFactory!(build); - const mockParent = { get: jest.fn((key: string) => { - if (key === 'parsed') return { event: 'INSERT', rowIds: ['row-uuid'], overflow: false }; - if (key === 'subscribedIds') return null; - return null; - }) }; + const mockParent = { get: jest.fn(() => ({ event: 'INSERT', rowIds: ['row-uuid'], overflow: false })) }; result.plans['TasksSubscriptionPayload'].tasks(mockParent); + // The gate already narrowed rowIds to the subscription's ids, so the + // resolver reads nothing but 'parsed'. expect(mockParent.get).toHaveBeenCalledWith('parsed'); - expect(mockParent.get).toHaveBeenCalledWith('subscribedIds'); - expect(mockResource.get).toHaveBeenCalled(); - }); - - it('payload row resolver uses first matching ID when ids provided', () => { - createRealtimeSubscriptionsPlugin(); - - const codec = createMockCodec('tasks', { realtime: true }); - const mockResource = { - ...createMockResource('tasks', codec), - get: jest.fn(), - }; - const build = createMockBuild({ - tasks: mockResource, - }); - - const result = capturedFactory!(build); - const mockParent = { get: jest.fn((key: string) => { - if (key === 'parsed') return { event: 'INSERT', rowIds: ['id-a', 'id-b', 'id-c'], overflow: false }; - if (key === 'subscribedIds') return ['id-b', 'id-d']; - return null; - }) }; - - result.plans['TasksSubscriptionPayload'].tasks(mockParent); - expect(mockParent.get).toHaveBeenCalledWith('subscribedIds'); + expect(mockParent.get).not.toHaveBeenCalledWith('subscribedIds'); expect(mockResource.get).toHaveBeenCalled(); }); }); @@ -563,120 +529,124 @@ describe('createRealtimeSubscriptionsPlugin', () => { }); }); + // These drive a real async iterable through the gate and assert on what the + // stream *yields*. The previous versions re-implemented the intersection + // inline and asserted on `.get()` call sites, which is why a filtered event + // reaching the client as `event: 'UNKNOWN'` went unnoticed. describe('sparse set filtering (ids argument)', () => { - it('subscribePlan passes ids through object step', () => { - createRealtimeSubscriptionsPlugin(); - - const codec = createMockCodec('tasks', { realtime: true }); - const build = createMockBuild({ - tasks: createMockResource('tasks', codec), - }); - - const result = capturedFactory!(build); - const mockArgs = { getRaw: jest.fn((key: string) => { - if (key === 'ids') return ['id-a', 'id-b']; - return null; - }) }; - - result.plans['Subscription']['onTasksChanged'].subscribePlan(null, mockArgs); + function subscriberEmitting(...payloads: string[]) { + return { + // eslint-disable-next-line @typescript-eslint/require-await + async *subscribe() { + yield* payloads; + }, + }; + } - expect(mockArgs.getRaw).toHaveBeenCalledWith('ids'); + async function collect(subscriber: { subscribe(topic: string): any }) { + const out = []; + for await (const event of await subscriber.subscribe('realtime:app_public.tasks')) { + out.push(event); + } + return out; + } - // The listen callback is captured but not invoked by the mock. - // Invoke it manually to verify ids are threaded through. - expect(mockListen).toHaveBeenCalled(); - const listenCallback = mockListen.mock.calls[mockListen.mock.calls.length - 1][2]; - listenCallback('INSERT:id-a'); + it('yields nothing at all for an unsubscribed row', async () => { + const gated = createGatedSubscriber(subscriberEmitting('UPDATE:id-x,id-y'), { + ids: ['id-a', 'id-b'], + threshold: 50, + }); - expect(mockObject).toHaveBeenCalled(); - const objectArg = mockObject.mock.calls[mockObject.mock.calls.length - 1][0]; - expect(objectArg).toHaveProperty('subscribedIds'); + await expect(collect(gated)).resolves.toEqual([]); }); - it('drops events with no row ID intersection in sparse set mode', () => { - const parsed = parseNotifyPayload('INSERT:id-x,id-y'); - const subscribedIds = ['id-a', 'id-b']; + it('narrows rowIds to the subscribed set so no consumer re-intersects', async () => { + const gated = createGatedSubscriber(subscriberEmitting('UPDATE:id-x,id-b,id-a'), { + ids: ['id-a', 'id-b'], + threshold: 50, + }); - const hasMatch = parsed.rowIds.some((rid: string) => subscribedIds.includes(rid)); - expect(hasMatch).toBe(false); + await expect(collect(gated)).resolves.toEqual([ + { event: 'UPDATE', rowIds: ['id-b', 'id-a'], overflow: false }, + ]); }); - it('delivers events with row ID intersection in sparse set mode', () => { - const parsed = parseNotifyPayload('UPDATE:id-a,id-x'); - const subscribedIds = ['id-a', 'id-b']; - - const hasMatch = parsed.rowIds.some((rid: string) => subscribedIds.includes(rid)); - expect(hasMatch).toBe(true); - }); + it('passes every event through in full collection mode', async () => { + const gated = createGatedSubscriber(subscriberEmitting('INSERT:id-x', 'DELETE:id-y'), { + ids: null, + threshold: 50, + }); - it('delivers INVALIDATE events regardless of sparse set', () => { - const parsed = parseNotifyPayload('INVALIDATE'); - expect(parsed.overflow).toBe(true); - expect(parsed.rowIds).toEqual([]); + await expect(collect(gated)).resolves.toEqual([ + { event: 'INSERT', rowIds: ['id-x'], overflow: false }, + { event: 'DELETE', rowIds: ['id-y'], overflow: false }, + ]); }); - it('rowId resolver returns first matching ID from sparse set', () => { - createRealtimeSubscriptionsPlugin(); - - const codec = createMockCodec('tasks', { realtime: true }); - const build = createMockBuild({ - tasks: { ...createMockResource('tasks', codec), get: jest.fn() }, + it('delivers INVALIDATE regardless of the sparse set', async () => { + const gated = createGatedSubscriber(subscriberEmitting('INVALIDATE'), { + ids: ['id-a'], + threshold: 50, }); - const result = capturedFactory!(build); - const payload = result.plans['TasksSubscriptionPayload']; + await expect(collect(gated)).resolves.toEqual([ + { event: 'INVALIDATE', rowIds: [], overflow: true }, + ]); + }); - const mockParent = { get: jest.fn((key: string) => { - if (key === 'parsed') return { event: 'UPDATE', rowIds: ['id-x', 'id-b', 'id-a'], overflow: false }; - if (key === 'subscribedIds') return ['id-a', 'id-b']; - return null; - }) }; + it('collapses a burst to one INVALIDATE and then yields nothing', async () => { + const gated = createGatedSubscriber( + subscriberEmitting('INSERT:a', 'INSERT:b', 'INSERT:c', 'INSERT:d'), + { ids: null, threshold: 2 } + ); - payload.rowId(mockParent); - expect(mockParent.get).toHaveBeenCalledWith('parsed'); - expect(mockParent.get).toHaveBeenCalledWith('subscribedIds'); + await expect(collect(gated)).resolves.toEqual([ + { event: 'INSERT', rowIds: ['a'], overflow: false }, + { event: 'INSERT', rowIds: ['b'], overflow: false }, + { event: 'INVALIDATE', rowIds: [], overflow: true }, + ]); }); - it('rowId resolver returns null when no sparse set match', () => { - createRealtimeSubscriptionsPlugin(); - - const codec = createMockCodec('tasks', { realtime: true }); - const build = createMockBuild({ - tasks: { ...createMockResource('tasks', codec), get: jest.fn() }, - }); + it('throttles each subscription independently', async () => { + const opts: EventGateOptions = { ids: null, threshold: 1 }; + const noisy = createGatedSubscriber(subscriberEmitting('INSERT:a', 'INSERT:b'), opts); + await collect(noisy); - const result = capturedFactory!(build); - const payload = result.plans['TasksSubscriptionPayload']; + const quiet = createGatedSubscriber(subscriberEmitting('INSERT:c'), opts); + await expect(collect(quiet)).resolves.toEqual([ + { event: 'INSERT', rowIds: ['c'], overflow: false }, + ]); + }); - const mockParent = { get: jest.fn((key: string) => { - if (key === 'parsed') return { event: 'INSERT', rowIds: ['id-x'], overflow: false }; - if (key === 'subscribedIds') return ['id-a', 'id-b']; - return null; - }) }; + it('surfaces a malformed payload instead of emitting an event for it', async () => { + const gated = createGatedSubscriber(subscriberEmitting('TRUNCATE:a'), { + ids: null, + threshold: 50, + }); - payload.rowId(mockParent); - expect(mockParent.get).toHaveBeenCalledWith('subscribedIds'); + await expect(collect(gated)).rejects.toThrow(MalformedNotifyPayloadError); }); - it('rowId resolver falls back to first rowId when no sparse set provided', () => { + it('subscribePlan reads the ids argument and gates the subscriber', () => { createRealtimeSubscriptionsPlugin(); const codec = createMockCodec('tasks', { realtime: true }); const build = createMockBuild({ - tasks: { ...createMockResource('tasks', codec), get: jest.fn() }, + tasks: createMockResource('tasks', codec), }); const result = capturedFactory!(build); - const payload = result.plans['TasksSubscriptionPayload']; + const mockArgs = { getRaw: jest.fn((key: string) => (key === 'ids' ? ['id-a'] : null)) }; - const mockParent = { get: jest.fn((key: string) => { - if (key === 'parsed') return { event: 'INSERT', rowIds: ['id-first', 'id-second'], overflow: false }; - if (key === 'subscribedIds') return null; - return null; - }) }; + result.plans['Subscription']['onTasksChanged'].subscribePlan(null, mockArgs); - payload.rowId(mockParent); - expect(mockParent.get).toHaveBeenCalledWith('subscribedIds'); + expect(mockArgs.getRaw).toHaveBeenCalledWith('ids'); + + // 'parsed' is the whole payload now — nothing downstream needs the ids. + const listenCallback = mockListen.mock.calls[mockListen.mock.calls.length - 1][2]; + listenCallback({ event: 'INSERT', rowIds: ['id-a'], overflow: false }); + const objectArg = mockObject.mock.calls[mockObject.mock.calls.length - 1][0]; + expect(Object.keys(objectArg)).toEqual(['parsed']); }); }); diff --git a/graphile/graphile-realtime-subscriptions/src/event-gate.ts b/graphile/graphile-realtime-subscriptions/src/event-gate.ts new file mode 100644 index 000000000..b1747173a --- /dev/null +++ b/graphile/graphile-realtime-subscriptions/src/event-gate.ts @@ -0,0 +1,177 @@ +/** + * The gate between PostgreSQL's NOTIFY stream and a GraphQL subscription. + * + * Everything that can decide "this event should not reach the client" lives + * here, upstream of grafast, because grafast's `listen` maps each event to a + * step and offers no way to suppress one. Deciding downstream forced the + * plugin to represent "dropped" as a null payload, which the field resolvers + * then had to coalesce (`p?.event ?? 'UNKNOWN'`) — so a filtered event still + * reached the client, carrying an invented event name. Filtering here means + * the stream only ever yields events that should exist, and every resolver + * downstream is total. + * + * The gate also owns parsing and the overflow throttle, so a payload the + * trigger and the plugin disagree about fails loudly instead of arriving as + * an `UNKNOWN`-shaped record. + */ + +import type { GrafastSubscriber } from 'grafast'; + +/** Operations `emit_change` is allowed to report. */ +const KNOWN_OPERATIONS = new Set(['INSERT', 'UPDATE', 'DELETE']); + +export interface ParsedPayload { + event: string; + /** + * Row ids being reported. Already narrowed to the subscription's `ids` when + * it supplied any, so no consumer needs to re-intersect. Empty for + * INVALIDATE. + */ + rowIds: string[]; + overflow: boolean; +} + +export class MalformedNotifyPayloadError extends Error { + constructor(raw: string, reason: string) { + super( + `Malformed realtime NOTIFY payload ${JSON.stringify(raw)}: ${reason}. ` + + 'Expected "INVALIDATE" or ":[,...]" ' + + 'as emitted by the emit_change trigger.' + ); + this.name = 'MalformedNotifyPayloadError'; + } +} + +/** + * Parse the NOTIFY payload from `emit_change`. + * Format: `"TG_OP:id1,id2,..."` or `"INVALIDATE"`. + * + * Throws on anything else: a payload this cannot read means the trigger and + * this plugin have diverged, and inventing an event name for it hides a + * deployment fault behind data the client will act on. + */ +export function parseNotifyPayload(raw: string): ParsedPayload { + if (raw === 'INVALIDATE') { + return { event: 'INVALIDATE', rowIds: [], overflow: true }; + } + + const colonIdx = raw.indexOf(':'); + if (colonIdx === -1) { + throw new MalformedNotifyPayloadError(raw, 'no ":" separating the operation from the row ids'); + } + + const event = raw.substring(0, colonIdx); + if (!KNOWN_OPERATIONS.has(event)) { + throw new MalformedNotifyPayloadError(raw, `unknown operation ${JSON.stringify(event)}`); + } + + const idsPart = raw.substring(colonIdx + 1); + const rowIds = idsPart.length > 0 ? idsPart.split(',') : []; + if (rowIds.length === 0) { + throw new MalformedNotifyPayloadError(raw, `${event} carries no row ids`); + } + + return { event, rowIds, overflow: false }; +} + +/** + * Per-subscription event rate tracker. Counts events in a sliding 1-second + * window. + */ +export class EventThrottle { + private windowStart = 0; + private eventCount = 0; + private overflowSent = false; + + constructor(private readonly threshold: number) {} + + /** + * Record an event and return whether it should be delivered. + * Returns 'deliver' for normal events, 'overflow' when the threshold + * is first exceeded, or 'drop' for subsequent events in the same window. + */ + check(): 'deliver' | 'overflow' | 'drop' { + const now = Date.now(); + + if (now - this.windowStart >= 1000) { + this.windowStart = now; + this.eventCount = 0; + this.overflowSent = false; + } + + this.eventCount++; + + if (this.eventCount <= this.threshold) { + return 'deliver'; + } + + if (!this.overflowSent) { + this.overflowSent = true; + return 'overflow'; + } + + return 'drop'; + } +} + +export interface EventGateOptions { + /** Sparse-set subscription: deliver only events touching these row ids. */ + ids?: readonly string[] | null; + /** Events per second before the gate collapses the stream to one INVALIDATE. */ + threshold: number; +} + +/** + * Apply the gate to one raw NOTIFY payload. + * + * Returns the payload to deliver, or `null` meaning *do not emit* — a decision + * only this layer is allowed to make, because only this layer can act on it. + */ +function gate(raw: string, ids: readonly string[] | null, throttle: EventThrottle): ParsedPayload | null { + const parsed = parseNotifyPayload(raw); + + // An INVALIDATE is the overflow signal itself; throttling it would drop the + // one event telling the client to refetch. + if (parsed.overflow) return parsed; + + const action = throttle.check(); + if (action === 'drop') return null; + if (action === 'overflow') { + return { event: 'INVALIDATE', rowIds: [], overflow: true }; + } + + if (!ids || ids.length === 0) return parsed; + + const matched = parsed.rowIds.filter(rowId => ids.includes(rowId)); + if (matched.length === 0) return null; + + return { ...parsed, rowIds: matched }; +} + +/** + * Wrap a `GrafastSubscriber` so its stream yields parsed, gated payloads. + * + * Built per subscription (the throttle and `ids` are per-subscriber state), so + * one noisy client can no longer throttle every other subscriber to the same + * table — which a single build-time throttle instance did. + */ +export function createGatedSubscriber( + inner: GrafastSubscriber>, + { ids, threshold }: EventGateOptions +): GrafastSubscriber> { + const subscribedIds = ids && ids.length > 0 ? ids : null; + const throttle = new EventThrottle(threshold); + + return { + async *subscribe(topic: string) { + const source = await inner.subscribe(topic); + for await (const raw of source) { + const payload = gate(String(raw), subscribedIds, throttle); + if (payload !== null) yield payload; + } + } + // Deliberately no release(): `inner` is the shared pgSubscriber from the + // request context and is not ours to tear down. Ending iteration returns + // the underlying iterator, which is the whole of our cleanup. + }; +} diff --git a/graphile/graphile-realtime-subscriptions/src/plugin.ts b/graphile/graphile-realtime-subscriptions/src/plugin.ts index 85a38958d..a013d778b 100644 --- a/graphile/graphile-realtime-subscriptions/src/plugin.ts +++ b/graphile/graphile-realtime-subscriptions/src/plugin.ts @@ -17,8 +17,10 @@ * 1. A row is inserted/updated/deleted * 2. The emit_change trigger fires pg_notify with TG_OP:row_ids or INVALIDATE * 3. PostGraphile's pgSubscriber receives the NOTIFY - * 4. The plugin parses the payload and fetches the specific changed row(s) - * 5. The client receives { event, row, rowId, overflow } + * 4. The gate (see event-gate.ts) parses, throttles and filters it — a + * payload this subscription should not see never becomes an event + * 5. The plugin fetches the changed row(s) + * 6. The client receives { event, row, rowId, overflow } * * Cursor tracking (at-least-once delivery): * The CursorTracker class provides a complementary polling-based delivery @@ -31,7 +33,7 @@ * * Overflow protection: * - Database-side: statements affecting > 50 rows send INVALIDATE - * - Plugin-side: per-subscriber throttle (default 50 events/second/table) + * - Plugin-side: per-subscription throttle (default 50 events/second/table) * drops individual events and sends a single INVALIDATE when exceeded * * Security / RLS enforcement: @@ -51,6 +53,8 @@ import { constant, context as grafastContext, lambda,listen, object } from 'graf import type { GraphileConfig } from 'graphile-config'; import { extendSchema } from 'graphile-utils'; +import type { ParsedPayload } from './event-gate'; +import { createGatedSubscriber } from './event-gate'; import type { RealtimeSubscriptionsPluginOptions } from './types'; const log = new Logger('graphile-realtime-subscriptions'); @@ -69,73 +73,6 @@ interface RealtimeTableInfo { pgTable: string; } -interface ParsedPayload { - event: string; - rowIds: string[]; - overflow: boolean; -} - -/** - * Parse the NOTIFY payload from emit_change. - * Format: "TG_OP:id1,id2,..." or "INVALIDATE" - */ -function parseNotifyPayload(raw: string): ParsedPayload { - if (raw === 'INVALIDATE') { - return { event: 'INVALIDATE', rowIds: [], overflow: true }; - } - - const colonIdx = raw.indexOf(':'); - if (colonIdx === -1) { - return { event: raw || 'UNKNOWN', rowIds: [], overflow: false }; - } - - const event = raw.substring(0, colonIdx); - const idsPart = raw.substring(colonIdx + 1); - const rowIds = idsPart.length > 0 ? idsPart.split(',') : []; - - return { event, rowIds, overflow: false }; -} - -/** - * Per-subscriber, per-table event rate tracker. - * Counts events in a sliding 1-second window. - */ -class EventThrottle { - private windowStart = 0; - private eventCount = 0; - private overflowSent = false; - - constructor(private readonly threshold: number) {} - - /** - * Record an event and return whether it should be delivered. - * Returns 'deliver' for normal events, 'overflow' when the threshold - * is first exceeded, or 'drop' for subsequent events in the same window. - */ - check(): 'deliver' | 'overflow' | 'drop' { - const now = Date.now(); - - if (now - this.windowStart >= 1000) { - this.windowStart = now; - this.eventCount = 0; - this.overflowSent = false; - } - - this.eventCount++; - - if (this.eventCount <= this.threshold) { - return 'deliver'; - } - - if (!this.overflowSent) { - this.overflowSent = true; - return 'overflow'; - } - - return 'drop'; - } -} - function discoverRealtimeTables(build: any): RealtimeTableInfo[] { const { pgRegistry } = build.input; const resources = pgRegistry.pgResources; @@ -201,6 +138,30 @@ function buildTypeDefs(tables: RealtimeTableInfo[]): string { return `extend type Subscription {\n${subscriptionFields}\n}\n\n${payloadTypes}`; } +/** + * The gate never emits a null payload, so one here means the plan graph was + * rewired wrongly — fail rather than invent an event for the client. + */ +function requirePayload(payload: unknown): ParsedPayload { + if (payload === null || payload === undefined) { + throw new Error( + 'Realtime subscription payload is missing: the gated subscriber only ever ' + + 'yields parsed payloads, so this event bypassed createGatedSubscriber.', + ); + } + return payload as ParsedPayload; +} + +/** + * The row this event reports. `rowIds` is already narrowed to the + * subscription's `ids`, so the first entry is the one to surface; INVALIDATE + * carries none, which is a genuine absence rather than a suppressed error. + */ +function reportedRowId(payload: ParsedPayload): string | null { + if (payload.overflow) return null; + return payload.rowIds[0] ?? null; +} + function buildPlans( tables: RealtimeTableInfo[], overflowThreshold: number, @@ -209,47 +170,21 @@ function buildPlans( const allPlans: Record = {}; for (const { resource, fieldName, payloadTypeName, rowFieldName, notifyChannel } of tables) { - const throttle = new EventThrottle(overflowThreshold); - subscriptionPlans[fieldName] = { subscribePlan(_$root: any, args: any) { const $pgSubscriber = (grafastContext() as any).get('pgSubscriber'); const $topic = constant(notifyChannel); const $ids = args.getRaw('ids'); - return listen($pgSubscriber, $topic, ($payload: any) => { - const $parsed = lambda([$payload, $ids], (pair: unknown) => { - const [raw, subscribedIds] = pair as readonly [unknown, string[] | null | undefined]; - const parsed = parseNotifyPayload(String(raw)); - - const action = parsed.overflow ? 'deliver' : throttle.check(); - - if (action === 'drop') { - return null; - } - - if (action === 'overflow') { - return { - event: 'INVALIDATE', - rowIds: [], - overflow: true, - }; - } - - // Sparse set filtering: only deliver events for subscribed row IDs - if (subscribedIds && subscribedIds.length > 0) { - const hasMatch = parsed.rowIds.some((rid: string) => subscribedIds.includes(rid)); - if (!hasMatch) return null; - } - - return parsed; - }); - - return object({ - parsed: $parsed, - subscribedIds: $ids, - }); + // Parsing, throttling and sparse-set filtering all happen in the gate, + // built once per subscription, so the stream below yields only events + // that should reach this client — every step after it is total. + const $subscriber = lambda([$pgSubscriber, $ids], (pair: unknown) => { + const [pgSubscriber, ids] = pair as readonly [any, string[] | null | undefined]; + return createGatedSubscriber(pgSubscriber, { ids, threshold: overflowThreshold }); }); + + return listen($subscriber, $topic, ($payload: any) => object({ parsed: $payload })); }, plan($event: any) { return $event; @@ -258,47 +193,17 @@ function buildPlans( allPlans[payloadTypeName] = { event($parent: any) { - const $parsed = $parent.get('parsed'); - return lambda($parsed, (p: unknown) => (p as ParsedPayload | null)?.event ?? 'UNKNOWN'); + return lambda($parent.get('parsed'), (p: unknown) => requirePayload(p).event); }, rowId($parent: any) { - const $parsed = $parent.get('parsed'); - const $subscribedIds = $parent.get('subscribedIds'); - return lambda([$parsed, $subscribedIds], (pair: unknown) => { - const [p, subscribedIds] = pair as readonly [ParsedPayload | null, string[] | null | undefined]; - if (!p || p.overflow || p.rowIds.length === 0) return null; - - // When ids are provided, return the first matching row ID - if (subscribedIds && subscribedIds.length > 0) { - return p.rowIds.find((rid: string) => subscribedIds.includes(rid)) ?? null; - } - - return p.rowIds[0]; - }); + return lambda($parent.get('parsed'), (p: unknown) => reportedRowId(requirePayload(p))); }, overflow($parent: any) { - const $parsed = $parent.get('parsed'); - return lambda($parsed, (p: unknown) => (p as ParsedPayload | null)?.overflow ?? false); + return lambda($parent.get('parsed'), (p: unknown) => requirePayload(p).overflow); }, [rowFieldName]($parent: any) { - const $parsed = $parent.get('parsed'); - const $subscribedIds = $parent.get('subscribedIds'); - - const $rowId = lambda( - [$parsed, $subscribedIds], - (tuple: unknown) => { - const [p, subscribedIds] = tuple as readonly [ - ParsedPayload | null, - string[] | null | undefined, - ]; - if (!p || p.overflow || p.rowIds.length === 0) return null; - // When ids are provided, return first matching row ID - if (subscribedIds && subscribedIds.length > 0) { - return p.rowIds.find((rid: string) => subscribedIds.includes(rid)) ?? null; - } - // Full collection mode: return first row ID - return p.rowIds[0]; - }, + const $rowId = lambda($parent.get('parsed'), (p: unknown) => + reportedRowId(requirePayload(p)), ); return resource.get({ id: $rowId }); @@ -340,8 +245,13 @@ export { createRealtimeSubscriptionsPlugin as RealtimeSubscriptionsPlugin }; // Re-export CursorTracker and RealtimeManager for convenience export { CursorTracker } from './cursor-tracker'; +export type { ParsedPayload } from './event-gate'; +export { + createGatedSubscriber, + EventThrottle, + MalformedNotifyPayloadError, + parseNotifyPayload, +} from './event-gate'; export { RealtimeManager } from './realtime-manager'; export type { ChangeLogEntry, CursorTrackerOptions, Queryable, RealtimeManagerOptions } from './types'; - -// Exported for testing -export { DEFAULT_OVERFLOW_THRESHOLD,EventThrottle, parseNotifyPayload }; +export { DEFAULT_OVERFLOW_THRESHOLD }; diff --git a/graphile/graphile-realtime-test/__tests__/realtime-websocket.integration.test.ts b/graphile/graphile-realtime-test/__tests__/realtime-websocket.integration.test.ts index 56e75b0b3..8652424a1 100644 --- a/graphile/graphile-realtime-test/__tests__/realtime-websocket.integration.test.ts +++ b/graphile/graphile-realtime-test/__tests__/realtime-websocket.integration.test.ts @@ -177,14 +177,13 @@ describe('realtime WebSocket E2E (real graphql-ws over ws)', () => { unsubscribe(); - const relevant = events.filter(e => e.onItemChanged.event !== 'UNKNOWN'); - expect(relevant.length).toBe(1); - expect(relevant[0].onItemChanged.event).toBe('INSERT'); - expect(relevant[0].onItemChanged.rowId).toBe(watchedId); - - const filtered = events.filter(e => e.onItemChanged.event === 'UNKNOWN'); - expect(filtered.length).toBe(1); - expect(filtered[0].onItemChanged.rowId).toBeNull(); + // The change to the unwatched row must produce no event at all. This + // previously asserted the opposite — one event with `event: 'UNKNOWN'` + // and a null rowId — which told a subscriber that *something* it isn't + // watching changed, and when. + expect(events).toHaveLength(1); + expect(events[0].onItemChanged.event).toBe('INSERT'); + expect(events[0].onItemChanged.rowId).toBe(watchedId); }, 15000); // ─── Multiple concurrent WebSocket subscribers ──────────────────────── diff --git a/scripts/check-test-coverage.cjs b/scripts/check-test-coverage.cjs index 811b24ceb..783cbc4af 100644 --- a/scripts/check-test-coverage.cjs +++ b/scripts/check-test-coverage.cjs @@ -26,9 +26,6 @@ const WORKFLOW = path.join(ROOT, '.github/workflows/run-tests.yaml'); * so this list cannot rot quietly. */ const EXCLUSIONS = { - 'graphile/graphile-realtime-test': - 'Sparse-set filtering delivers a bogus UNKNOWN event instead of dropping ' + - 'the filtered one — a real plugin bug, not test drift. constructive-planning#1426.', 'graphql/react': 'Requires an external GraphQL endpoint via $TESTING_URL and throws at ' + 'import time without one — a manual suite, not a CI one.'