.live() observer form: push each rowset into a callback#93
Merged
Conversation
ryanrasti
force-pushed
the
ryan_4_live_observer
branch
from
July 23, 2026 03:11
64dcdb3 to
3a70ff9
Compare
There was a problem hiding this comment.
Pull request overview
Adds a push/observer consumption mode for QueryBuilder.live() to support RPC environments (Cap’n Web) where AsyncIterable cannot be serialized, returning a LiveSubscription handle for teardown.
Changes:
- Added
LiveObserver/LiveSubscriptionandsubscribeLive()to adapt anAsyncIterableinto an observer-driven stream with backpressure and teardown. - Extended
QueryBuilder.live()with overloads for observer-first and(conn, observer)forms while preserving the existing iterator form. - Added Vitest coverage for observer mode on SQLite and exported the new types from the public index.
Reviewed changes
Copilot reviewed 4 out of 4 changed files in this pull request and generated 2 comments.
| File | Description |
|---|---|
| src/live/sqlite/db-live.test.ts | Adds a new test validating observer-mode pushes and unsubscribe behavior. |
| src/live/observer.ts | Introduces observer subscription adapter + subscription handle with teardown and RPC callback retention. |
| src/index.ts | Exposes LiveSubscription and LiveObserver publicly. |
| src/builder/query.ts | Implements .live() overloads and argument validation for observer mode, wiring to subscribeLive(). |
💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.
Comment on lines
+70
to
+75
| const stop = (): void => { | ||
| if (!stopped) { | ||
| stopped = true; | ||
| void iter.return?.(); | ||
| } | ||
| }; |
Comment on lines
+31
to
+38
| @expose() | ||
| unsubscribe(): void { | ||
| this.#stop(); | ||
| } | ||
|
|
||
| [Symbol.dispose](): void { | ||
| this.#stop(); | ||
| } |
ryanrasti
force-pushed
the
ryan_4_live_observer
branch
2 times, most recently
from
July 23, 2026 17:43
6fb273c to
62603da
Compare
Collapses the three live() overloads to one signature — live(conn?) returns a LiveQuery, which is an AsyncIterable (iterate `for await` locally) and exposes @expose'd .observe(observer) → LiveSubscription (the push form, which crosses capnweb; the observer's callbacks cross by reference). The body is just asObservable(conn.live(this)). This also removes the Connection|LiveObserver union off live(), which combined with LiveObserver's now-`| undefined` optional fields lets the observer schema be a plain z.object (colocated in observer.ts) instead of a hand-rolled z.custom predicate — the union and exactOptional mismatch were what broke the @expose decorator's inference before. The callback field type is `(...args: unknown[]) => unknown` so the schema stays assignable to every LiveObserver field (incl. the zero-arg onReturn) under the generic method param. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
ryanrasti
force-pushed
the
ryan_4_live_observer
branch
from
July 23, 2026 17:46
62603da to
0f61310
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
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.
QueryBuilder.live gains an observer overload — .live({ onNext, onThrow?, onReturn? }) — returning a LiveSubscription (@expose'd unsubscribe + Symbol.dispose). Rationale: Cap'n Web can't serialize AsyncIterators, so the callback pattern is needed.
The iterator form is unchanged and stays local-only; the observer form is the wire-native shape: each committed change pushes the rowset into onNext, awaited for backpressure — a disconnected peer rejects the push and the subscription tears down. RPC-delivered callbacks are dup()'d for the subscription's lifetime. Observer-first (.live({onNext})) resolves the default connection (needs PR3).
Covered in src/live/sqlite/db-live.test.ts.