fix(realtime): filter subscription events before they become events

[pyramation]

Summary

A sparse-set subscriber — onItemChanged(ids: [watched]) — was told about every other row's changes. Not the row data, but the fact and the timing, which is the part that leaks.

The cause is a sentinel that couldn't be honoured. Filtering ran inside listen()'s mapping callback, and that callback has no way to say don't emit: returning null still produced an event, so the field resolvers had to invent something to put in a String!:

// before — the filter, and the resolver that undoes it
if (!hasMatch) return null;                                  // "dropped"
event: (p as ParsedPayload | null)?.event ?? 'UNKNOWN',      // ...delivered
overflow: (p as ParsedPayload | null)?.overflow ?? false,

The throttle's drop leaked identically. null meant both "filtered" and "no data", and ?? turned both into data.

The fix is to filter upstream of the stream, where "don't emit" is expressible. createGatedSubscriber() wraps the pgSubscriber per subscription and owns parse + throttle + sparse-set gate, so a payload that shouldn't be delivered is never yielded:

const $subscriber = lambda([$pgSubscriber, $ids], ([sub, ids]) =>
  createGatedSubscriber(sub, { ids, threshold: overflowThreshold }));

return listen($subscriber, $topic, $payload => object({ parsed: $payload }));

Everything after it is total, which is the actual point:

• parsed is non-null by construction, so every ?? fallback is deleted rather than corrected;
• a payload emit_change and this plugin disagree about (no colon, empty, unknown op) throws MalformedNotifyPayloadError — a divergence between trigger and plugin is a deployment fault, and 'UNKNOWN' was hiding it behind data the client acts on;
• subscribedIds leaves the payload object entirely: the gate narrows rowIds to the subscribed set, so nothing downstream re-intersects.

Second bug, found on the way. The throttle was one instance per table constructed at schema build time and shared by every subscriber, so one noisy client throttled everyone on that table. It's per-subscription now — which is what its own comment always claimed (Per-subscriber, per-table event rate tracker).

Why the tests didn't catch it

They asserted the bug was the contract. The websocket suite filtered UNKNOWN out of the received events and then required exactly one of them:

- const relevant = events.filter(e => e.onItemChanged.event !== 'UNKNOWN');
- expect(relevant.length).toBe(1);
- const filtered = events.filter(e => e.onItemChanged.event === 'UNKNOWN');
- expect(filtered.length).toBe(1);          // the leak, pinned as expected
+ expect(events).toHaveLength(1);
+ expect(events[0].onItemChanged.event).toBe('INSERT');

And the unit suite's sparse-set tests re-implemented the intersection inline (parsed.rowIds.some(...) in the test body) or asserted that .get('subscribedIds') had been called — never that the stream withheld anything. They now drive a real async iterable through the gate and assert on what it yields, including the burst→INVALIDATE→silence sequence and per-subscription throttle isolation.

graphile-realtime-test comes out of EXCLUSIONS in scripts/check-test-coverage.cjs and into the pg-graphile-extras batch; 94/94 unit and 16/16 integration pass.

Closes the first item of constructive-planning#1426.

Link to Devin session: https://app.devin.ai/sessions/087553534c774929918ec4d378845881
Requested by: @pyramation

[devin-ai-integration]

🤖 Devin AI Engineer

I'll be helping with this pull request! Here's what you should know:

✅ I will automatically:

• Address comments on this PR. Add '(aside)' to your comment to have me ignore it.
• Look at CI failures and help fix them

Note: I can only respond to comments from users who have write access to this repository.

⚙️ Control Options:

• Disable automatic comment, CI, and merge conflict monitoring