feat: add schema adapters for Prisma, MongoDB, PostgreSQL, and EventEmitter - #3
Merged
Merged
Conversation
|
Note Reviews pausedIt looks like this branch is under active development. To avoid overwhelming you with review comments due to an influx of new commits, CodeRabbit has automatically paused this review. You can configure this behavior by changing the Use the following commands to manage reviews:
Use the checkboxes below for quick actions:
📝 WalkthroughWalkthroughAdds a pluggable SSE adapter system (types + Prisma, MongoDB, PostgreSQL, EventEmitter adapters), integrates adapter lifecycle into channels via channel.watch()/close(), and extends defineSchema() with a resources field that auto-expands .created/.updated/.deleted while explicit events take precedence. Changes
Sequence Diagram(s)sequenceDiagram
rect rgba(220,240,255,0.5)
participant Dev
end
participant Channel
participant Adapter as SSEAdapter
participant DB as Database
participant Client as SSE_Client
Dev->>Channel: watch(adapter)
Channel->>Adapter: start(emit)
Adapter->>DB: subscribe / LISTEN / watch
DB-->>Adapter: change / notification
Adapter->>Adapter: map source -> schema event
Adapter->>Channel: emit(eventType, payload)
Channel->>Client: broadcast SSE (event + JSON)
Dev->>Channel: close()
Channel->>Adapter: stop()
Adapter->>DB: unsubscribe / close
sequenceDiagram
participant Dev
participant Schema as defineSchema()
participant Expansion
participant Result
Dev->>Schema: defineSchema({ resources: { orders: {...} } })
Schema->>Expansion: expandResource('orders')
Expansion->>Expansion: generate orders.created / orders.updated / orders.deleted
Schema->>Schema: merge explicit events (explicit wins)
Schema->>Result: freeze & return SchemaResult
Estimated code review effort🎯 4 (Complex) | ⏱️ ~60 minutes Possibly related PRs
Poem
🚥 Pre-merge checks | ✅ 2 | ❌ 1❌ Failed checks (1 warning)
✅ Passed checks (2 passed)
✏️ Tip: You can configure your own custom pre-merge checks in the settings. ✨ Finishing Touches🧪 Generate unit tests (beta)
Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out. Comment |
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 6
🧹 Nitpick comments (4)
src/server/adapters/emitter.ts (1)
45-50: Consider resettingstartedflag instop()for reusability.The
startedflag is not reset instop(), which means after callingstop()and thenstart()again, the adapter won't re-register handlers. If one-shot usage is intentional, this is fine. Otherwise, consider resetting the flag to allow adapter restart.♻️ Proposed fix to allow restart
stop(): void { for (const [emitterEvent, handler] of handlers) { emitter.off(emitterEvent, handler) } handlers.clear() + started = false },🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@src/server/adapters/emitter.ts` around lines 45 - 50, The stop() method currently removes handlers and clears the handlers map but does not reset the started flag, preventing subsequent calls to start() from re-registering handlers; update stop() to set started = false (in the same scope where stop() and start() live) after calling emitter.off and handlers.clear so the adapter becomes restartable, and ensure this change keeps stop() idempotent with respect to handlers and emitter.off.src/server/adapters/pg.ts (1)
99-106:stop()issues UNLISTEN even whenstart()was never called.When
stop()is called without a priorstart(), it will still issueUNLISTENqueries for all mapped channels. While PostgreSQL toleratesUNLISTENon non-listened channels, this is wasteful and could mask bugs. Consider guarding with a check.♻️ Proposed fix to guard UNLISTEN
async stop(): Promise<void> { removeListener() + if (!started) return started = false // Issue UNLISTEN for each channel with proper identifier quoting await Promise.all( channels.map((ch) => client.query(`UNLISTEN ${quoteIdentifier(ch)}`)), ) },🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@src/server/adapters/pg.ts` around lines 99 - 106, stop() currently issues UNLISTEN for every channel even if start() wasn't called; change stop() to check the started flag before running removeListener() and issuing UNLISTENs. Specifically, in stop() guard the logic with if (!started) return (or skip UNLISTEN), so removeListener(), started = false, and the Promise.all over channels.map((ch) => client.query(`UNLISTEN ${quoteIdentifier(ch)}`)) only run when started is true; use the existing started variable, channels array, removeListener(), client.query(), and quoteIdentifier() identifiers to locate and update the function.src/__tests__/adapter-mongodb.test.ts (1)
140-141: Small timeout values may cause test flakiness in slow CI environments.The 10-20ms timeouts used throughout the tests (e.g.,
setTimeout(r, 10)) rely on async iteration completing within that window. Consider using slightly larger values or a more deterministic approach to avoid intermittent failures.🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@src/__tests__/adapter-mongodb.test.ts` around lines 140 - 141, The test uses a short sleep (await new Promise((r) => setTimeout(r, 10))) to allow async iteration to start which can be flaky in slow CI; change these sleeps in src/__tests__/adapter-mongodb.test.ts to a more robust approach—either increase the timeout to a safer value (e.g., 50–100ms) or replace the ad-hoc sleep with a deterministic helper (e.g., flushPromises using setImmediate/process.nextTick or awaiting the actual async iterator consumer) wherever the test currently uses the Promise with setTimeout to ensure async iteration in the test reliably completes.src/schema.ts (1)
17-27: Unnecessary type cast on line 19.The
opvariable is already typed asResourceOpfrom iteratingRESOURCE_OPS, so theas ResourceOpcast is redundant.♻️ Remove redundant cast
for (const op of RESOURCE_OPS) { const eventName = `${resourceName}.${op}` - const opDef = resourceDef[op as ResourceOp] + const opDef = resourceDef[op] expanded[eventName] = {🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@src/schema.ts` around lines 17 - 27, The loop is using a redundant type cast on op; change the lookup from resourceDef[op as ResourceOp] to resourceDef[op] (while keeping the rest of the block identical) so RESOURCE_OPS, resourceName, resourceDef, ResourceOp, expanded, eventName and opDef are used without the unnecessary cast; ensure TypeScript still compiles and types flow correctly after removing the cast.
🤖 Prompt for all review comments with AI agents
Verify each finding against the current code and only fix it if needed.
Inline comments:
In `@prd/0003-schema-adapters.md`:
- Around line 40-45: This PRD lists implementation priority but lacks the
required tracked decomposition step; update the document to run and reference
the `/ateam plan <prd-file>` output before the "Implementation Priority"
section, and insert the generated tracked work items (tickets/tasks) that map to
the listed components (e.g., SSEAdapter, channel.watch(adapter), enhanced schema
resources, Prisma adapter, MongoDB/Postgres/EventEmitter adapters) so that each
implementation item is associated with a tracked ticket ID and acceptance
criteria; ensure the PRD explicitly states "Do not implement until `/ateam plan
prd/0003-schema-adapters.md` tasks are created" and include the created task IDs
and short descriptions for each adapter and core infra item.
In `@README.md`:
- Around line 236-243: Update the README examples for channel.watch to show
async usage: clarify that for async adapters you must await their startup (e.g.,
await adapter.start()) before calling channel.watch or await channel.watch(...)
if watch is implemented to return a Promise; keep the existing sync example
as-is but add a second async example and a note that cleanup() will still call
adapter.stop() and that channel.close() will await/stop all watched adapters.
Reference channel.watch, adapter.stop, cleanup(), and channel.close in the text
so readers know where to apply the awaiting behavior.
In `@src/__tests__/adapter-exports.test.ts`:
- Around line 38-45: The test "should re-export SSEAdapter type (verified via
type-level import)" only does a runtime import and doesn't confirm the type-only
export; update the test to include a compile-time type assertion that references
the SSEAdapter type from the barrel (e.g., add a type-only import or an `//
`@ts-expect-error``-free type alias like `type _T =
import('../server/adapters/index').SSEAdapter` or similar) so TypeScript will
fail the test suite if the barrel no longer re-exports SSEAdapter; keep the
existing runtime `barrel` import and expect but add this type-level check in the
same test to ensure type resolution is validated.
In `@src/server/adapters/mongodb.ts`:
- Around line 104-111: The start method currently launches
runStream(emit).catch(()=>{}) but never resets the adapter state if the
background stream exits, leaving started=true and preventing future restarts;
change the start implementation to attach cleanup logic to the runStream promise
(use then/finally or an async wrapper) so that whenever runStream resolves or
rejects it sets started=false and ensures stopped=true as appropriate; locate
the start(emit: ...) method and the runStream call and add a finally/cleanup
block that resets the started/stopped flags so subsequent start() calls can
recover the adapter.
In `@src/server/adapters/prisma.ts`:
- Around line 52-58: start() currently sets started = true before prisma.$use()
and stop() doesn't reset started, which can permanently disable the adapter;
change start() to set started = true only after successful prisma.$use()
registration and ensure that on any exception during registration you clear
started/active and emitFn so the adapter can be retried, and update stop() to
reset started = false (and also clear active and emitFn) so subsequent start()
calls can reinitialize; refer to the start(), stop(), started, active, emitFn
symbols and prisma.$use() call when making these changes.
In `@src/server/index.ts`:
- Around line 386-397: The issue is that stoppedAdapters is a global Set so once
an adapter is stopped it prevents later watch() calls from running stop again
for that new lifecycle; to fix, make the cleanup idempotency local to the watch
call instead of relying on the global stoppedAdapters: inside watch() introduce
a local stopped boolean (e.g. let localStopped = false) and have cleanup
check/set that (if localStopped return; localStopped = true;
watchedAdapters.delete(adapter); await adapter.stop()); do not add the adapter
to the global stoppedAdapters from this watch (or if you must keep the global
Set for other logic, ensure you also remove the adapter from stoppedAdapters at
the start of watch() so subsequent watches aren’t short-circuited). This change
affects the watch() function, its cleanup closure, stoppedAdapters and
watchedAdapters usage, and the adapter.stop() call.
---
Nitpick comments:
In `@src/__tests__/adapter-mongodb.test.ts`:
- Around line 140-141: The test uses a short sleep (await new Promise((r) =>
setTimeout(r, 10))) to allow async iteration to start which can be flaky in slow
CI; change these sleeps in src/__tests__/adapter-mongodb.test.ts to a more
robust approach—either increase the timeout to a safer value (e.g., 50–100ms) or
replace the ad-hoc sleep with a deterministic helper (e.g., flushPromises using
setImmediate/process.nextTick or awaiting the actual async iterator consumer)
wherever the test currently uses the Promise with setTimeout to ensure async
iteration in the test reliably completes.
In `@src/schema.ts`:
- Around line 17-27: The loop is using a redundant type cast on op; change the
lookup from resourceDef[op as ResourceOp] to resourceDef[op] (while keeping the
rest of the block identical) so RESOURCE_OPS, resourceName, resourceDef,
ResourceOp, expanded, eventName and opDef are used without the unnecessary cast;
ensure TypeScript still compiles and types flow correctly after removing the
cast.
In `@src/server/adapters/emitter.ts`:
- Around line 45-50: The stop() method currently removes handlers and clears the
handlers map but does not reset the started flag, preventing subsequent calls to
start() from re-registering handlers; update stop() to set started = false (in
the same scope where stop() and start() live) after calling emitter.off and
handlers.clear so the adapter becomes restartable, and ensure this change keeps
stop() idempotent with respect to handlers and emitter.off.
In `@src/server/adapters/pg.ts`:
- Around line 99-106: stop() currently issues UNLISTEN for every channel even if
start() wasn't called; change stop() to check the started flag before running
removeListener() and issuing UNLISTENs. Specifically, in stop() guard the logic
with if (!started) return (or skip UNLISTEN), so removeListener(), started =
false, and the Promise.all over channels.map((ch) => client.query(`UNLISTEN
${quoteIdentifier(ch)}`)) only run when started is true; use the existing
started variable, channels array, removeListener(), client.query(), and
quoteIdentifier() identifiers to locate and update the function.
ℹ️ Review info
Configuration used: defaults
Review profile: CHILL
Plan: Pro
📒 Files selected for processing (23)
CHANGELOG.mdREADME.mddocs/API.mdpackage.jsonprd/0003-schema-adapters.mdsrc/__tests__/adapter-emitter.test.tssrc/__tests__/adapter-exports.test.tssrc/__tests__/adapter-mongodb.test.tssrc/__tests__/adapter-pg.test.tssrc/__tests__/adapter-prisma.test.tssrc/__tests__/adapter-types.test.tssrc/__tests__/channel-watch.test.tssrc/__tests__/schema-resources.test.tssrc/schema.tssrc/server/adapters/emitter.tssrc/server/adapters/index.tssrc/server/adapters/mongodb.tssrc/server/adapters/pg.tssrc/server/adapters/prisma.tssrc/server/adapters/types.tssrc/server/index.tssrc/types.tstsconfig.emit.json
Comment on lines
+104
to
+111
| start(emit: (eventType: string, payload: unknown) => void): void { | ||
| if (started) return | ||
|
|
||
| stopped = false | ||
| started = true | ||
| // Kick off async iteration without blocking start() | ||
| runStream(emit).catch(() => {}) | ||
| }, |
There was a problem hiding this comment.
Reset adapter state when background stream exits.
If the stream loop stops (error/end/max reconnect path), started stays true, so later start() calls are ignored and the adapter can’t recover.
🔧 Proposed fix
start(emit: (eventType: string, payload: unknown) => void): void {
if (started) return
stopped = false
started = true
// Kick off async iteration without blocking start()
- runStream(emit).catch(() => {})
+ runStream(emit)
+ .catch(() => {})
+ .finally(() => {
+ if (!stopped) started = false
+ })
},📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
Suggested change
| start(emit: (eventType: string, payload: unknown) => void): void { | |
| if (started) return | |
| stopped = false | |
| started = true | |
| // Kick off async iteration without blocking start() | |
| runStream(emit).catch(() => {}) | |
| }, | |
| start(emit: (eventType: string, payload: unknown) => void): void { | |
| if (started) return | |
| stopped = false | |
| started = true | |
| // Kick off async iteration without blocking start() | |
| runStream(emit) | |
| .catch(() => {}) | |
| .finally(() => { | |
| if (!stopped) started = false | |
| }) | |
| }, |
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.
In `@src/server/adapters/mongodb.ts` around lines 104 - 111, The start method
currently launches runStream(emit).catch(()=>{}) but never resets the adapter
state if the background stream exits, leaving started=true and preventing future
restarts; change the start implementation to attach cleanup logic to the
runStream promise (use then/finally or an async wrapper) so that whenever
runStream resolves or rejects it sets started=false and ensures stopped=true as
appropriate; locate the start(emit: ...) method and the runStream call and add a
finally/cleanup block that resets the started/stopped flags so subsequent
start() calls can recover the adapter.
Owner
Author
There was a problem hiding this comment.
This was already addressed in our second commit (7b7eb28) — start() is now async and awaits runStream(emit) directly. The .catch(() => {}) fire-and-forget pattern is gone. Additionally, stop() now resets started = false (line 115), so the adapter can recover after stream exit. No further changes needed.
There was a problem hiding this comment.
@queso Thank you for the clarification! I can only see the diff provided in this review context, which shows the fire-and-forget pattern. If commit 7b7eb28 already makes start() async and properly awaits runStream(emit), and stop() now resets started = false, then the concern is indeed resolved. That's the right approach for proper adapter lifecycle management. 👍
🐰
There was a problem hiding this comment.
Actionable comments posted: 1
♻️ Duplicate comments (2)
src/server/index.ts (1)
388-399:⚠️ Potential issue | 🟠 MajorClear
stoppedAdaptersentry at the start ofwatch()to allow adapter reuse.To support repeated watch/cleanup cycles with the same adapter instance, remove the adapter from
stoppedAdapterswhen a new watch begins.🐛 Proposed fix
watch( adapter: SSEAdapter, ): (() => Promise<void>) | Promise<() => Promise<void>> { if (closed) throw new Error('Channel is closed') + // Reset stopped state for this adapter instance to allow reuse + stoppedAdapters.delete(adapter) // Bug 3 fix: idempotent cleanup — only stop once regardless of who calls it const cleanup = async (): Promise<void> => { + watchedAdapters.delete(adapter) if (stoppedAdapters.has(adapter)) return stoppedAdapters.add(adapter) - watchedAdapters.delete(adapter) await adapter.stop() }🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@src/server/index.ts` around lines 388 - 399, In watch(adapter: SSEAdapter) ensure the adapter can be reused by removing it from stoppedAdapters at the start of the method; specifically, in the watch function (before the closed check or immediately after) call stoppedAdapters.delete(adapter) so a previously-stopped adapter can be watched again, while keeping the existing idempotent cleanup logic in cleanup() that checks stoppedAdapters.has(adapter), deletes from watchedAdapters, adds to stoppedAdapters, and calls adapter.stop().src/server/adapters/mongodb.ts (1)
101-111:⚠️ Potential issue | 🟠 Major
start()blocks indefinitely until the stream ends, preventing callers from proceeding.The
await runStream(emit)on Line 110 blocksstart()until the change stream terminates. This deviates from theSSEAdapterinterface expectation wherestart()should return promptly after initializing the adapter. Callers likechannel.watch()will hang until the stream closes.Additionally, when
runStreamexits (error/max reconnects),startedremainstrue, preventing subsequentstart()calls from re-initializing the adapter (as noted in a prior review).🐛 Proposed fix: Run stream in background and reset state on exit
async start( emit: (eventType: string, payload: unknown) => void, ): Promise<void> { if (started) return stopped = false started = true reconnectAttempts = 0 - await runStream(emit) + // Run stream in background; reset state if it exits unexpectedly + runStream(emit) + .catch(() => {}) + .finally(() => { + if (!stopped) started = false + }) },🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@src/server/adapters/mongodb.ts` around lines 101 - 111, start() currently awaits runStream(emit) which blocks callers and leaves started true after runStream exits; change start to launch runStream in the background (do not await) so start() returns immediately and keep the existing guard (if (started) return) to prevent concurrent starts. Inside runStream (or in the promise's finally/catch handler created when you call it from start), reset state variables (set started = false, stopped = true, and optionally reconnectAttempts = 0) and log/propagate any final error so the adapter can be restarted later; reference the start function, runStream function call, and the started/stopped/reconnectAttempts variables when making these changes.
🧹 Nitpick comments (5)
src/types.ts (1)
203-208:SchemaDefinitionintersection type may allow unintended shapes.The intersection
Record<string, SchemaEventDefinition | Record<string, ResourceDefinition> | undefined>allows any key to be either aSchemaEventDefinitionor aRecord<string, ResourceDefinition>. This means a top-level key likeorderscould accidentally accept a resource-like shape{ created: { key: '...' } }without being under theresourcesfield, which would not be expanded at runtime.Consider narrowing the union to only
SchemaEventDefinition | undefinedfor non-resourceskeys:♻️ Proposed type refinement
export type SchemaDefinition = { resources?: Record<string, ResourceDefinition> -} & Record< - string, - SchemaEventDefinition | Record<string, ResourceDefinition> | undefined -> +} & { + [key: string]: SchemaEventDefinition | undefined +}Note: This may require adjusting how
resourcesinteracts with the index signature. An alternative is using a discriminated approach or Omit to excluderesourcesfrom the index signature.🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@src/types.ts` around lines 203 - 208, The current SchemaDefinition intersection lets any top-level key (e.g., orders) be either a SchemaEventDefinition or a Record<string,ResourceDefinition>, which allows resource-like shapes outside the dedicated resources field; update SchemaDefinition so the index signature only permits SchemaEventDefinition | undefined for non-resources keys (or alternatively use an Omit-based/discriminated approach that excludes the "resources" key from the index signature) and keep resources?: Record<string,ResourceDefinition> as the sole place for ResourceDefinition maps to ensure resources are only accepted under the resources field; adjust types referencing SchemaDefinition (search for SchemaDefinition, resources, SchemaEventDefinition, ResourceDefinition) accordingly.src/__tests__/schema-resources.test.ts (1)
322-326: EmptyafterEachblock has a misleading comment.The comment says "Restore console.warn after each test" but the body is empty. Each test already calls
mockRestore()explicitly, so this is harmless but could confuse readers. Consider removing the empty block or adding the actual restore logic.♻️ Proposed fix: Remove empty afterEach or centralize mock restoration
describe('collision warning — explicit event overrides resource-generated event', () => { - afterEach(() => { - // Restore console.warn after each test in this describe block - }) + // Each test restores its own mock via mockRestore()🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@src/__tests__/schema-resources.test.ts` around lines 322 - 326, The empty afterEach block inside the describe('collision warning — explicit event overrides resource-generated event', ...) is misleading because it only contains a comment but no restore logic; either remove this empty afterEach entirely or implement the intended restore (e.g., call console.warn.mockRestore() or a central cleanup helper) so mocks are consistently restored — update the afterEach in that describe block (or add a centralized afterEach helper used by the tests) to perform the mock restoration rather than leaving an empty block.src/server/adapters/mongodb.ts (1)
71-76: RecursiverunStream()call can exceed call stack on rapid invalidate events.If the MongoDB cluster experiences repeated rapid invalidate events, the recursive
await runStream(emit)could grow the call stack unboundedly. Consider using a loop pattern instead of recursion.♻️ Proposed iterative approach
Convert the recursive call into a while-loop in the outer
runStreamor use tail-call style by returning and having the caller re-invoke. For example, track ashouldReconnectflag and loop at the top level rather than recursing.🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@src/server/adapters/mongodb.ts` around lines 71 - 76, The current recursive await runStream(emit) inside the invalidate/reconnect branch can blow the call stack on rapid invalidations; change runStream to use an iterative reconnect loop instead: introduce a local loop flag (e.g., shouldReconnect) or wrap the main stream logic in a while(!stopped && reconnectAttempts < MAX_RECONNECT_ATTEMPTS) loop, increment reconnectAttempts and perform the reopen/resume actions inside the loop rather than calling runStream recursively, and ensure you reset/clear shouldReconnect and respect the stopped and MAX_RECONNECT_ATTEMPTS checks so the function returns normally when done. Reference symbols: runStream, reconnectAttempts, MAX_RECONNECT_ATTEMPTS, stopped, emit.src/server/adapters/prisma.ts (1)
51-57: Consider supporting restart afterstop()by reactivating emissions.Since Prisma middleware cannot be unregistered, the current design permanently disables the adapter after
stop(). If restart capability is needed,start()could checkstartedand simply reactivateactiveandemitFnwithout re-calling$use().♻️ Proposed fix to support restart
start(emit: (eventType: string, payload: unknown) => void): void { + emitFn = emit + if (started) { + // Middleware already registered; just reactivate emissions. + active = true + return + } - if (started) return - - emitFn = emit - active = true - started = true try { prisma.$use(async (params, next) => { // ... middleware logic }) + started = true + active = true } catch (err) { - active = false - started = false emitFn = undefined + active = false throw err } },🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@src/server/adapters/prisma.ts` around lines 51 - 57, The adapter currently sets started=true when first started and never re-enables emissions after stop() because start() returns early; modify start(emit) so that if (started) it does not re-register Prisma middleware but instead reassigns emitFn = emit and sets active = true (reactivating emissions), while the first-time path still calls prisma.$use(...) and sets started = true; ensure stop() continues to set active = false but leaves started true so restart works without re-calling $use. Include references to the start, stop, started, active, emitFn symbols and the prisma.$use registration in your change.src/server/adapters/pg.ts (1)
101-108:stop()issues UNLISTEN even if adapter was never started.When
stop()is called without a priorstart(), it still issues UNLISTEN queries for all channels. While harmless, this is wasteful and could mask usage errors. Consider guarding UNLISTEN behind thestartedcheck.♻️ Proposed fix
async stop(): Promise<void> { removeListener() + if (!started) return started = false // Issue UNLISTEN for each channel with proper identifier quoting await Promise.all( channels.map((ch) => client.query(`UNLISTEN ${quoteIdentifier(ch)}`)), ) },🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@src/server/adapters/pg.ts` around lines 101 - 108, The stop() method currently executes UNLISTEN for all channels even when the adapter was never started; modify stop() to first check the started flag and only run removeListener() and the Promise.all of client.query(`UNLISTEN ${quoteIdentifier(ch)}`) when started is true, leaving stop() as a no-op (or just ensure started is false) otherwise; reference the stop() function, the started variable, removeListener(), channels array, client.query(...) and quoteIdentifier(...) when applying this guard so UNLISTENs are only issued after a successful start().
🤖 Prompt for all review comments with AI agents
Verify each finding against the current code and only fix it if needed.
Inline comments:
In `@src/server/adapters/pg.ts`:
- Around line 67-74: The adapter sets started = true before awaiting the LISTEN
queries, which can leave started true if any client.query fails; move the
assignment so started is set only after the await Promise.all(...) completes
successfully (or wrap the await in try/catch and reset started = false before
rethrowing), updating the start() flow that uses the started flag and
referencing started, start(), channels.map(...), client.query(...) and
quoteIdentifier(...) so the adapter isn't left in a started state when LISTEN
fails.
---
Duplicate comments:
In `@src/server/adapters/mongodb.ts`:
- Around line 101-111: start() currently awaits runStream(emit) which blocks
callers and leaves started true after runStream exits; change start to launch
runStream in the background (do not await) so start() returns immediately and
keep the existing guard (if (started) return) to prevent concurrent starts.
Inside runStream (or in the promise's finally/catch handler created when you
call it from start), reset state variables (set started = false, stopped = true,
and optionally reconnectAttempts = 0) and log/propagate any final error so the
adapter can be restarted later; reference the start function, runStream function
call, and the started/stopped/reconnectAttempts variables when making these
changes.
In `@src/server/index.ts`:
- Around line 388-399: In watch(adapter: SSEAdapter) ensure the adapter can be
reused by removing it from stoppedAdapters at the start of the method;
specifically, in the watch function (before the closed check or immediately
after) call stoppedAdapters.delete(adapter) so a previously-stopped adapter can
be watched again, while keeping the existing idempotent cleanup logic in
cleanup() that checks stoppedAdapters.has(adapter), deletes from
watchedAdapters, adds to stoppedAdapters, and calls adapter.stop().
---
Nitpick comments:
In `@src/__tests__/schema-resources.test.ts`:
- Around line 322-326: The empty afterEach block inside the describe('collision
warning — explicit event overrides resource-generated event', ...) is misleading
because it only contains a comment but no restore logic; either remove this
empty afterEach entirely or implement the intended restore (e.g., call
console.warn.mockRestore() or a central cleanup helper) so mocks are
consistently restored — update the afterEach in that describe block (or add a
centralized afterEach helper used by the tests) to perform the mock restoration
rather than leaving an empty block.
In `@src/server/adapters/mongodb.ts`:
- Around line 71-76: The current recursive await runStream(emit) inside the
invalidate/reconnect branch can blow the call stack on rapid invalidations;
change runStream to use an iterative reconnect loop instead: introduce a local
loop flag (e.g., shouldReconnect) or wrap the main stream logic in a
while(!stopped && reconnectAttempts < MAX_RECONNECT_ATTEMPTS) loop, increment
reconnectAttempts and perform the reopen/resume actions inside the loop rather
than calling runStream recursively, and ensure you reset/clear shouldReconnect
and respect the stopped and MAX_RECONNECT_ATTEMPTS checks so the function
returns normally when done. Reference symbols: runStream, reconnectAttempts,
MAX_RECONNECT_ATTEMPTS, stopped, emit.
In `@src/server/adapters/pg.ts`:
- Around line 101-108: The stop() method currently executes UNLISTEN for all
channels even when the adapter was never started; modify stop() to first check
the started flag and only run removeListener() and the Promise.all of
client.query(`UNLISTEN ${quoteIdentifier(ch)}`) when started is true, leaving
stop() as a no-op (or just ensure started is false) otherwise; reference the
stop() function, the started variable, removeListener(), channels array,
client.query(...) and quoteIdentifier(...) when applying this guard so UNLISTENs
are only issued after a successful start().
In `@src/server/adapters/prisma.ts`:
- Around line 51-57: The adapter currently sets started=true when first started
and never re-enables emissions after stop() because start() returns early;
modify start(emit) so that if (started) it does not re-register Prisma
middleware but instead reassigns emitFn = emit and sets active = true
(reactivating emissions), while the first-time path still calls prisma.$use(...)
and sets started = true; ensure stop() continues to set active = false but
leaves started true so restart works without re-calling $use. Include references
to the start, stop, started, active, emitFn symbols and the prisma.$use
registration in your change.
In `@src/types.ts`:
- Around line 203-208: The current SchemaDefinition intersection lets any
top-level key (e.g., orders) be either a SchemaEventDefinition or a
Record<string,ResourceDefinition>, which allows resource-like shapes outside the
dedicated resources field; update SchemaDefinition so the index signature only
permits SchemaEventDefinition | undefined for non-resources keys (or
alternatively use an Omit-based/discriminated approach that excludes the
"resources" key from the index signature) and keep resources?:
Record<string,ResourceDefinition> as the sole place for ResourceDefinition maps
to ensure resources are only accepted under the resources field; adjust types
referencing SchemaDefinition (search for SchemaDefinition, resources,
SchemaEventDefinition, ResourceDefinition) accordingly.
ℹ️ Review info
Configuration used: defaults
Review profile: CHILL
Plan: Pro
📒 Files selected for processing (9)
src/__tests__/adapter-pg.test.tssrc/__tests__/adapter-prisma.test.tssrc/__tests__/schema-resources.test.tssrc/schema.tssrc/server/adapters/mongodb.tssrc/server/adapters/pg.tssrc/server/adapters/prisma.tssrc/server/index.tssrc/types.ts
✅ Files skipped from review due to trivial changes (1)
- src/tests/adapter-prisma.test.ts
🚧 Files skipped from review as they are similar to previous changes (1)
- src/schema.ts
There was a problem hiding this comment.
Actionable comments posted: 2
🧹 Nitpick comments (1)
src/server/index.ts (1)
182-183: PreferWeakSetfor stopped adapter tracking.
stoppedAdaptersonly needs membership checks; usingWeakSetavoids retaining stopped adapter instances in long-lived channels.♻️ Suggested refactor
- const stoppedAdapters = new Set<SSEAdapter>() + const stoppedAdapters = new WeakSet<SSEAdapter>()🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@src/server/index.ts` around lines 182 - 183, stoppedAdapters is currently a Set<SSEAdapter> but only used for membership checks, so replace its declaration with a WeakSet<SSEAdapter> (symbol: stoppedAdapters) to avoid retaining adapter instances; update any code that iterates over stoppedAdapters (WeakSet is not iterable) to instead track necessary iteration targets elsewhere or perform membership checks directly against stoppedAdapters in places like the SSEAdapter lifecycle handlers, and keep watchedAdapters as a normal Set<SSEAdapter> if you still need iteration.
🤖 Prompt for all review comments with AI agents
Verify each finding against the current code and only fix it if needed.
Inline comments:
In `@src/server/adapters/emitter.ts`:
- Around line 28-50: The adapter marks started = true before all listeners are
registered in start(), and stop() can abort if emitter.off throws, causing
leaked listeners and a stuck started flag; change start() to only set started =
true after successfully registering all listeners (iterate mapping, add handlers
to a local list and call emitter.on for each, and if any emitter.on throws,
unregister any already-registered handlers and leave started false), and make
stop() exception-safe by wrapping each emitter.off call in try/catch (still
clear handlers and set started = false regardless), updating the handlers Map
and using the same unique symbols: start, stop, started, handlers, mapping,
emitter.on, emitter.off, and emit.
In `@src/server/adapters/prisma.ts`:
- Around line 16-18: The Prisma middleware usage via prisma.$use is
deprecated/removed; update the adapter by replacing any prisma.$use
registrations (and the PrismaClient interface stub) with Prisma Client
Extensions using prisma.$extends({ query: { $allModels: { create: ..., update:
..., delete: ... } } }) and move your middleware logic into those
create/update/delete handlers, or alternatively document and enforce a strict
Prisma version (e.g., prisma@^6.x) in package.json and README so prisma.$use
remains valid; locate and replace all uses of prisma.$use and the local
PrismaClient interface (and the middleware functions referenced there) with the
$extends pattern or add the version constraint consistently.
---
Nitpick comments:
In `@src/server/index.ts`:
- Around line 182-183: stoppedAdapters is currently a Set<SSEAdapter> but only
used for membership checks, so replace its declaration with a
WeakSet<SSEAdapter> (symbol: stoppedAdapters) to avoid retaining adapter
instances; update any code that iterates over stoppedAdapters (WeakSet is not
iterable) to instead track necessary iteration targets elsewhere or perform
membership checks directly against stoppedAdapters in places like the SSEAdapter
lifecycle handlers, and keep watchedAdapters as a normal Set<SSEAdapter> if you
still need iteration.
ℹ️ Review info
Configuration used: defaults
Review profile: CHILL
Plan: Pro
📒 Files selected for processing (6)
README.mdsrc/__tests__/adapter-exports.test.tssrc/server/adapters/emitter.tssrc/server/adapters/pg.tssrc/server/adapters/prisma.tssrc/server/index.ts
✅ Files skipped from review due to trivial changes (1)
- README.md
🚧 Files skipped from review as they are similar to previous changes (1)
- src/server/adapters/pg.ts
Comment on lines
+28
to
+50
| start(emit: (eventType: string, payload: unknown) => void): void { | ||
| if (started) return | ||
| started = true | ||
|
|
||
| for (const [emitterEvent, schemaEvent] of Object.entries(mapping)) { | ||
| const handler: EventListener = (...args: unknown[]) => { | ||
| try { | ||
| emit(schemaEvent, args[0]) | ||
| } catch { | ||
| // emit() errors must not propagate through the emitter's event dispatch | ||
| } | ||
| } | ||
| handlers.set(emitterEvent, handler) | ||
| emitter.on(emitterEvent, handler) | ||
| } | ||
| }, | ||
|
|
||
| stop(): void { | ||
| for (const [emitterEvent, handler] of handlers) { | ||
| emitter.off(emitterEvent, handler) | ||
| } | ||
| handlers.clear() | ||
| started = false |
There was a problem hiding this comment.
Make adapter lifecycle exception-safe to avoid partial subscriptions.
start() marks started before registration is fully complete, and stop() can abort early if off() throws. That can leave leaked listeners and a stuck adapter state.
🔧 Suggested fix
return {
start(emit: (eventType: string, payload: unknown) => void): void {
if (started) return
- started = true
-
- for (const [emitterEvent, schemaEvent] of Object.entries(mapping)) {
- const handler: EventListener = (...args: unknown[]) => {
- try {
- emit(schemaEvent, args[0])
- } catch {
- // emit() errors must not propagate through the emitter's event dispatch
- }
- }
- handlers.set(emitterEvent, handler)
- emitter.on(emitterEvent, handler)
- }
+ const registered: Array<[string, EventListener]> = []
+ try {
+ for (const [emitterEvent, schemaEvent] of Object.entries(mapping)) {
+ const handler: EventListener = (...args: unknown[]) => {
+ try {
+ emit(schemaEvent, args[0])
+ } catch {
+ // emit() errors must not propagate through the emitter's event dispatch
+ }
+ }
+ emitter.on(emitterEvent, handler)
+ handlers.set(emitterEvent, handler)
+ registered.push([emitterEvent, handler])
+ }
+ started = true
+ } catch (err) {
+ for (const [event, handler] of registered) {
+ try {
+ emitter.off(event, handler)
+ } catch {
+ // best-effort rollback
+ }
+ }
+ handlers.clear()
+ started = false
+ throw err
+ }
},
stop(): void {
for (const [emitterEvent, handler] of handlers) {
- emitter.off(emitterEvent, handler)
+ try {
+ emitter.off(emitterEvent, handler)
+ } catch {
+ // best-effort cleanup
+ }
}
handlers.clear()
started = false
},
}📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
Suggested change
| start(emit: (eventType: string, payload: unknown) => void): void { | |
| if (started) return | |
| started = true | |
| for (const [emitterEvent, schemaEvent] of Object.entries(mapping)) { | |
| const handler: EventListener = (...args: unknown[]) => { | |
| try { | |
| emit(schemaEvent, args[0]) | |
| } catch { | |
| // emit() errors must not propagate through the emitter's event dispatch | |
| } | |
| } | |
| handlers.set(emitterEvent, handler) | |
| emitter.on(emitterEvent, handler) | |
| } | |
| }, | |
| stop(): void { | |
| for (const [emitterEvent, handler] of handlers) { | |
| emitter.off(emitterEvent, handler) | |
| } | |
| handlers.clear() | |
| started = false | |
| start(emit: (eventType: string, payload: unknown) => void): void { | |
| if (started) return | |
| const registered: Array<[string, EventListener]> = [] | |
| try { | |
| for (const [emitterEvent, schemaEvent] of Object.entries(mapping)) { | |
| const handler: EventListener = (...args: unknown[]) => { | |
| try { | |
| emit(schemaEvent, args[0]) | |
| } catch { | |
| // emit() errors must not propagate through the emitter's event dispatch | |
| } | |
| } | |
| emitter.on(emitterEvent, handler) | |
| handlers.set(emitterEvent, handler) | |
| registered.push([emitterEvent, handler]) | |
| } | |
| started = true | |
| } catch (err) { | |
| for (const [event, handler] of registered) { | |
| try { | |
| emitter.off(event, handler) | |
| } catch { | |
| // best-effort rollback | |
| } | |
| } | |
| handlers.clear() | |
| started = false | |
| throw err | |
| } | |
| }, | |
| stop(): void { | |
| for (const [emitterEvent, handler] of handlers) { | |
| try { | |
| emitter.off(emitterEvent, handler) | |
| } catch { | |
| // best-effort cleanup | |
| } | |
| } | |
| handlers.clear() | |
| started = false | |
| }, |
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.
In `@src/server/adapters/emitter.ts` around lines 28 - 50, The adapter marks
started = true before all listeners are registered in start(), and stop() can
abort if emitter.off throws, causing leaked listeners and a stuck started flag;
change start() to only set started = true after successfully registering all
listeners (iterate mapping, add handlers to a local list and call emitter.on for
each, and if any emitter.on throws, unregister any already-registered handlers
and leave started false), and make stop() exception-safe by wrapping each
emitter.off call in try/catch (still clear handlers and set started = false
regardless), updating the handlers Map and using the same unique symbols: start,
stop, started, handlers, mapping, emitter.on, emitter.off, and emit.
Owner
Author
There was a problem hiding this comment.
Reasonable hardening. The emitter adapter works with any on/off interface, so we should be defensive. Fixing now — will set started = true after registration loop and wrap off() calls in try/catch.
There was a problem hiding this comment.
@queso, sounds good! The defensive approach makes sense given the adapter works with any on/off interface. Looking forward to seeing the hardened implementation with exception-safe registration and cleanup. 👍
🧠 Learnings used
Learnt from: queso
Repo: queso/reactiveSWR PR: 3
File: src/server/adapters/prisma.ts:52-58
Timestamp: 2026-02-27T16:33:37.175Z
Learning: In `src/server/adapters/prisma.ts`, the Prisma adapter intentionally does not reset `started = false` in `stop()` because Prisma's `$use()` method registers middleware permanently with no removal API. The `active` flag controls emission after `stop()`, and users requiring a fresh adapter should create a new instance rather than reusing a stopped one.
Comment on lines
+16
to
+18
| interface PrismaClient { | ||
| $use(middleware: PrismaMiddlewareFn): void | ||
| } |
There was a problem hiding this comment.
🧩 Analysis chain
🌐 Web query:
Prisma Client middleware $use deprecation removal timeline 2025 2026
💡 Result:
- Deprecation: Prisma Client middleware via
prisma.$use()was deprecated in Prisma ORM v4.16.0 (released June 20, 2023). [1][2] - Removal: The client middleware API (
prisma.$use) is removed in Prisma ORM v7 (migration guide explicitly lists “Client middleware has been removed”). [3] - When v7 shipped: Prisma announced Prisma ORM 7.0.0 on November 19, 2025, so the practical “removal moment” is the v7 upgrade (late 2025)—meaning in 2026 you should assume
$useis already gone if you’re on v7+. [4]
Action: migrate middleware logic to Prisma Client Extensions ($extends) as the supported replacement. [3]
Sources:
[1] https://www.gitclear.com/open_repos/prisma/prisma/release/4.16.0
[2] 47ng/prisma-field-encryption#63
[3] https://www.prisma.io/docs/v6/orm/more/upgrades/to-v7
[4] https://www.prisma.io/blog/announcing-prisma-orm-7-0-0
🌐 Web query:
Prisma $use middleware API alternative interceptor create update delete
💡 Result:
Prisma’s recommended “middleware / interceptor” alternative for intercepting create / update / delete is Prisma Client Extensions via $extends({ query: { … } }) (the query extension component hooks the query lifecycle and can mutate args and/or the result). [1][2]
(prisma.$use() has been treated as deprecated/discouraged by the community since Prisma 4.16.0 when Client Extensions became GA.) [2]
Intercept create / update / delete (all models)
import { PrismaClient } from "@prisma/client";
const prisma = new PrismaClient().$extends({
query: {
$allModels: {
async create({ model, args, query }) {
// e.g. add audit fields
// args.data = { ...args.data, createdAt: new Date() };
return query(args);
},
async update({ model, args, query }) {
// args.data = { ...args.data, updatedAt: new Date() };
return query(args);
},
async delete({ model, args, query }) {
// e.g. block deletes, log, or convert to soft-delete
return query(args);
},
},
},
});Prisma also supports $allOperations (all operations), targeting a single model, or a single operation on a single model. [1]
Soft-delete style: override delete → update (per model)
const prisma = new PrismaClient().$extends({
query: {
user: {
async delete({ args, query }) {
return query.update({
where: args.where,
data: { deletedAt: new Date() },
});
},
},
},
});This pattern (changing a model operation’s behavior via the query extension) is discussed by Prisma maintainers/users. [3]
Sources:
[1] Prisma Docs – query extensions (hook lifecycle, $allModels, $allOperations, per-operation overrides) (prisma.io)
[2] Community report of prisma.$use() deprecation starting Prisma 4.16.0 (github.com)
[3] Prisma discussion showing overriding delete via Client Extensions query (github.com)
Citations:
- 1: https://www.prisma.io/docs/v6/orm/prisma-client/client-extensions/query
- 2: prisma.$use() deprecated 47ng/prisma-field-encryption#63
- 3: Possible to change the operation inside Client Extension? prisma/orm#21530
Migrate from prisma.$use() to $extends() or document strict version constraint.
Prisma Client middleware via $use() was deprecated in v4.16.0 and removed entirely in Prisma v7.0.0 (November 2025). This adapter is incompatible with Prisma v7+.
Choose one path:
- Migrate to Prisma Client Extensions (
$extends({ query: { ... } })) — the officially supported replacement for intercepting operations. - Or document and enforce a version constraint (
prisma@^6.x) in package.json andREADME.md.
The suggested runtime guard will catch the failure but masks a deeper incompatibility. If supporting multiple Prisma versions, migrate the middleware registration to use $extends({ query: { $allModels: { create, update, delete } } }) with equivalent logic.
Also applies to: 59-61
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.
In `@src/server/adapters/prisma.ts` around lines 16 - 18, The Prisma middleware
usage via prisma.$use is deprecated/removed; update the adapter by replacing any
prisma.$use registrations (and the PrismaClient interface stub) with Prisma
Client Extensions using prisma.$extends({ query: { $allModels: { create: ...,
update: ..., delete: ... } } }) and move your middleware logic into those
create/update/delete handlers, or alternatively document and enforce a strict
Prisma version (e.g., prisma@^6.x) in package.json and README so prisma.$use
remains valid; locate and replace all uses of prisma.$use and the local
PrismaClient interface (and the middleware functions referenced there) with the
$extends pattern or add the version constraint consistently.
Owner
Author
There was a problem hiding this comment.
Valid point — $use() is removed in Prisma v7. This is a real compatibility concern but out of scope for this PR. Opening an issue to track the migration to $extends(). For now, the adapter works with Prisma v4-v6 which covers the vast majority of production deployments. Will add a note to the README.
There was a problem hiding this comment.
@queso, that's a pragmatic approach — deferring the v7 migration while supporting the established v4-v6 base makes sense for this PR's scope. Adding a README note about the supported Prisma versions will help users understand the current compatibility boundaries.
Would you like me to open the GitHub issue to track the $extends() migration? I can include:
- The migration path from
$use()to$extends({ query: { ... } }) - A backlink to this PR and comment thread
- The supported version constraint context
Just let me know and I'll create it for you.
🐰✨
/> />
(='.'=)
(")_(")
[thinking...]
✏️ Learnings added
Learnt from: queso
Repo: queso/reactiveSWR PR: 3
File: src/server/adapters/prisma.ts:16-18
Timestamp: 2026-02-27T16:47:24.881Z
Learning: In src/server/adapters/prisma.ts, the Prisma adapter currently targets Prisma v4-v6 using prisma.$use() middleware. Migration to Prisma v7's $extends() API is tracked separately and is not in scope for the initial adapter implementation.
Note: Learnings are effective only in the context of similar code segments. To apply general coding guidelines, please configure review instructions. You can manage existing learnings in the app.
🧠 Learnings used
Learnt from: queso
Repo: queso/reactiveSWR PR: 3
File: src/server/adapters/prisma.ts:52-58
Timestamp: 2026-02-27T16:33:37.175Z
Learning: In src/server/adapters/prisma.ts, do not reset the started flag to false in stop() because Prisma's $use() registers middleware permanently with no removal API. The active flag should govern whether emission occurs after stop(). If a fresh adapter is required, instantiate a new Prisma adapter instead of reusing a stopped one. This guidance is specific to the Prisma adapter implementation and should be kept as a file-specific rule.
There was a problem hiding this comment.
♻️ Duplicate comments (1)
src/server/adapters/emitter.ts (1)
27-43:⚠️ Potential issue | 🟡 MinorPartial registration failure leaves leaked handlers.
If
emitter.on()throws mid-registration (e.g., on the second event), handlers already registered are left attached. Sincestartedremainsfalse, a retry ofstart()would double-register those events. The previous review suggested rollback logic for this scenario.🛡️ Proposed fix to rollback on partial registration failure
start(emit: (eventType: string, payload: unknown) => void): void { if (started) return + const registered: Array<[string, EventListener]> = [] - for (const [emitterEvent, schemaEvent] of Object.entries(mapping)) { - const handler: EventListener = (...args: unknown[]) => { - try { - emit(schemaEvent, args[0]) - } catch { - // emit() errors must not propagate through the emitter's event dispatch + try { + for (const [emitterEvent, schemaEvent] of Object.entries(mapping)) { + const handler: EventListener = (...args: unknown[]) => { + try { + emit(schemaEvent, args[0]) + } catch { + // emit() errors must not propagate through the emitter's event dispatch + } } + emitter.on(emitterEvent, handler) + handlers.set(emitterEvent, handler) + registered.push([emitterEvent, handler]) } - handlers.set(emitterEvent, handler) - emitter.on(emitterEvent, handler) + started = true + } catch (err) { + for (const [event, handler] of registered) { + try { + emitter.off(event, handler) + } catch { + // best-effort rollback + } + } + handlers.clear() + throw err } - started = true },🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@src/server/adapters/emitter.ts` around lines 27 - 43, The start function can leave previously registered handlers attached if emitter.on(emitterEvent, handler) throws partway through mapping; implement rollback so partial registration is undone: when iterating mapping in start (use mapping, handlers, emitter, handler, started), wrap each emitter.on call in try/catch and on any exception remove all handlers that were already added (call emitter.off/removeListener for each key in handlers and clear handlers), ensure started remains false on failure and only set started = true after successful completion of the whole loop; also avoid leaving stale entries in handlers by only calling handlers.set after emitter.on succeeds.
🤖 Prompt for all review comments with AI agents
Verify each finding against the current code and only fix it if needed.
Duplicate comments:
In `@src/server/adapters/emitter.ts`:
- Around line 27-43: The start function can leave previously registered handlers
attached if emitter.on(emitterEvent, handler) throws partway through mapping;
implement rollback so partial registration is undone: when iterating mapping in
start (use mapping, handlers, emitter, handler, started), wrap each emitter.on
call in try/catch and on any exception remove all handlers that were already
added (call emitter.off/removeListener for each key in handlers and clear
handlers), ensure started remains false on failure and only set started = true
after successful completion of the whole loop; also avoid leaving stale entries
in handlers by only calling handlers.set after emitter.on succeeds.
…Emitter support
Adds a pluggable SSE adapter system and resource-based schema expansion:
- SSEAdapter interface and AdapterMapping type for building database adapters
- channel.watch(adapter) with idempotent cleanup and re-watch support
- createPrismaAdapter: intercepts $use() middleware (Prisma v4-v6)
- createMongoAdapter: Change Streams with resume tokens and bounded reconnect
- createPgAdapter: LISTEN/NOTIFY with always-quoted identifiers
- createEmitterAdapter: bridges any on/off-compatible event source
- defineSchema({ resources }) auto-expands CRUD event triplets
- SchemaResult type includes resource-expanded event keys
- Tree-shakeable subpath exports for each adapter
- Double-start guards, try/catch around emit(), exception-safe cleanup
- 723 tests passing, full TypeScript and Biome compliance
Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
queso
force-pushed
the
feat/schema-adapters
branch
from
February 28, 2026 19:18
9337197 to
d0f22f8
Compare
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
1 participant
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Summary
AdapterMappinggeneric type for building database adapters with start/stop lifecycledefineSchema()— auto-expand CRUD event triplets (orders→orders.created,orders.updated,orders.deleted)channel.watch(adapter)method bridging adapters to SSE broadcast channels with idempotent cleanup$use()middleware), MongoDB (Change Streams with resume tokens), PostgreSQL (LISTEN/NOTIFY), EventEmitter (generic on/off)package.jsonsubpath configuration for tree-shakeable individual adapter importsTest plan
🤖 Generated with Claude Code
Co-Authored-By: Claude Opus 4.6 noreply@anthropic.com
Summary by CodeRabbit
New Features
Bug Fixes
Changed
Documentation
Tests
Chores