feat(imessage): typing event + unified subscribeEvent routing - #41
feat(imessage): typing event + unified subscribeEvent routing#41Ryan Zhu (underthestars-zhy) wants to merge 2 commits into
subscribeEvent routing#41Conversation
Replaces the split `subscribeMessages` / custom-event-streams approach with a single `subscribeEvent(name)` path that handles all events, including `messages`, uniformly. All events now yield `[Space, payload]` tuples; `space` is stripped from the payload half. Broadcasters are keyed by `platform:eventName` instead of platform alone. Also adds an iMessage `typing` event backed by a new resilient `subscribeTyping` stream with exponential-backoff reconnection.
Custom event fan-in (`CustomEventStreams`, `mergedEventStreams`, and the lazy proxy) are removed. Non-`messages` events are now only accessible through platform-specific accessors. The messages stream is simplified to a direct merge without the generic `createMergedEventStream` abstraction.
📝 WalkthroughWalkthroughThe PR unifies the platform event subscription system by replacing separate handling paths for Changes
Sequence Diagram(s)sequenceDiagram
participant Consumer
participant PlatformInstance
participant EventCache
participant Runtime
participant EventBroadcaster
Consumer->>PlatformInstance: Access im.typing<br/>(or other event)
PlatformInstance->>EventCache: Check cache for event
alt Event cached
EventCache-->>PlatformInstance: Return cached async iterable
else Event not cached
PlatformInstance->>Runtime: subscribeEvent("typing")
Runtime->>EventBroadcaster: Subscribe to typed event stream
EventBroadcaster-->>Runtime: Return ManagedStream
Runtime-->>PlatformInstance: Return ManagedStream
PlatformInstance->>EventCache: Cache iterable for "typing"
EventCache-->>PlatformInstance: Return and cache async iterable
end
PlatformInstance-->>Consumer: Return async iterable
Consumer->>Consumer: Iterate over [Space, payload] tuples
sequenceDiagram
participant subscribeTyping
participant AdvancedIMessage
participant ChatClient
participant clientTypingStream
subscribeTyping->>subscribeTyping: Iterate over clients
loop For each AdvancedIMessage client
subscribeTyping->>clientTypingStream: Subscribe client
clientTypingStream->>ChatClient: chatTypingIndicators.stream()
ChatClient-->>clientTypingStream: Typing event stream
clientTypingStream->>clientTypingStream: Filter chat.typingIndicator events
alt On error
clientTypingStream->>clientTypingStream: Log error, close subscription
clientTypingStream->>clientTypingStream: Wait (jittered exponential backoff)
clientTypingStream->>ChatClient: Resubscribe
else On success
clientTypingStream-->>subscribeTyping: Emit IMessageTypingRecord
end
end
subscribeTyping-->>subscribeTyping: Merge all streams into ManagedStream
Estimated code review effort🎯 4 (Complex) | ⏱️ ~75 minutes Possibly related PRs
Suggested labels
Poem
🚥 Pre-merge checks | ✅ 4 | ❌ 1❌ Failed checks (1 warning)
✅ Passed checks (4 passed)
✏️ Tip: You can configure your own custom pre-merge checks in the settings. ✨ Finishing Touches📝 Generate docstrings
🧪 Generate unit tests (beta)
Review rate limit: 4/5 reviews remaining, refill in 12 minutes. Comment |
There was a problem hiding this comment.
Pull request overview
This PR refactors Spectrum’s event subscription model to route all platform events (including messages) through a single subscribeEvent(name) entrypoint backed by a shared broadcaster-per-(platform,event) cache, and adds an iMessage typing event stream.
Changes:
- Replace split message/custom-event subscription paths with unified
PlatformRuntime.subscribeEvent(name)routing and a singleeventBroadcastersmap. - Remove Spectrum-level custom event proxy/fan-in (
app.<event>), shifting non-messagesconsumption to platform-specific accessors (e.g.imessage(app).typing). - Add iMessage remote
typingevent stream with reconnect/backoff behavior; local mode yields an empty stream.
Reviewed changes
Copilot reviewed 5 out of 5 changed files in this pull request and generated 2 comments.
Show a summary per file
| File | Description |
|---|---|
| packages/spectrum-ts/src/spectrum.ts | Introduces unified per-(platform,event) broadcasters and projects provider event records into [Space, payload] tuples. |
| packages/spectrum-ts/src/platform/types.ts | Adds ProviderEventRecord/EventPayload types, tightens event typing, and replaces subscribeMessages() with subscribeEvent(name). |
| packages/spectrum-ts/src/platform/define.ts | Generates per-platform accessors for all events via runtime.subscribeEvent(name) and caches per-event access. |
| packages/spectrum-ts/src/providers/imessage/remote/typing-events.ts | New resilient merged typing-indicator stream for remote iMessage clients. |
| packages/spectrum-ts/src/providers/imessage/index.ts | Wires the typing event into the iMessage platform definition (remote only). |
💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.
| // Lazily resolve every event (including `messages`) through the runtime's | ||
| // single `subscribeEvent` entry point. Cached per-name so a second | ||
| // `for await (...of im.<event>)` reuses the broadcaster's existing | ||
| // subscription instead of opening a new upstream stream. | ||
| const eventCache = new Map<string, AsyncIterable<[Space, unknown]>>(); |
| if (!subscription) { | ||
| return; | ||
| } | ||
| cached = subscription; |
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 2
🤖 Prompt for all review comments with AI agents
Verify each finding against the current code and only fix it if needed.
Inline comments:
In `@packages/spectrum-ts/src/platform/define.ts`:
- Around line 146-160: The getter for dynamic events is incorrectly caching the
subscription in eventCache so subsequent reads reuse the same AsyncIterable;
remove the per-access caching: delete the eventCache Map and its get/set calls
and change the property getter on base (the Object.defineProperty for each
eventName in def.events) to always call runtime.subscribeEvent(eventName) and
return that value (still handle a falsy/undefined return from
runtime.subscribeEvent by returning undefined). This ensures each access to
base[eventName] (e.g., imessage(app).typing) receives a fresh broadcaster
consumer.
In `@packages/spectrum-ts/src/providers/imessage/remote/typing-events.ts`:
- Around line 61-62: client.chats.subscribe() can throw before entering
runTypingSubscription, bypassing the retry/backoff and logTypingStreamError;
wrap the subscribe call in a try/catch inside the existing retry loop (the same
loop that calls runTypingSubscription) so any error from
client.chats.subscribe() is caught, forwarded to logTypingStreamError, triggers
the backoff/retry posture, and does not escape to tear the stream down. Apply
the same try/catch pattern to both occurrences where active =
client.chats.subscribe() appears, ensuring you only call runTypingSubscription
when subscribe succeeded and that caught errors loop back into the retry logic.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: Organization UI
Review profile: CHILL
Plan: Pro
Run ID: 1e581d9f-c2c6-49e4-8815-bc5238280868
📒 Files selected for processing (5)
packages/spectrum-ts/src/platform/define.tspackages/spectrum-ts/src/platform/types.tspackages/spectrum-ts/src/providers/imessage/index.tspackages/spectrum-ts/src/providers/imessage/remote/typing-events.tspackages/spectrum-ts/src/spectrum.ts
📜 Review details
⏰ Context from checks skipped due to timeout of 90000ms. You can increase the timeout in your CodeRabbit configuration to a maximum of 15 minutes (900000ms). (1)
- GitHub Check: Agent
🧰 Additional context used
📓 Path-based instructions (2)
**/*.{ts,tsx}
📄 CodeRabbit inference engine (AGENTS.md)
**/*.{ts,tsx}: Use explicit types for function parameters and return values when they enhance clarity in TypeScript
Preferunknownoveranywhen the type is genuinely unknown in TypeScript
Use const assertions (as const) for immutable values and literal types in TypeScript
Leverage TypeScript's type narrowing instead of type assertions
Files:
packages/spectrum-ts/src/providers/imessage/index.tspackages/spectrum-ts/src/providers/imessage/remote/typing-events.tspackages/spectrum-ts/src/platform/define.tspackages/spectrum-ts/src/platform/types.tspackages/spectrum-ts/src/spectrum.ts
**/*.{js,ts,jsx,tsx}
📄 CodeRabbit inference engine (AGENTS.md)
**/*.{js,ts,jsx,tsx}: Use meaningful variable names instead of magic numbers - extract constants with descriptive names
Use arrow functions for callbacks and short functions in JavaScript/TypeScript
Preferfor...ofloops over.forEach()and indexedforloops in JavaScript/TypeScript
Use optional chaining (?.) and nullish coalescing (??) for safer property access in JavaScript/TypeScript
Prefer template literals over string concatenation in JavaScript/TypeScript
Use destructuring for object and array assignments in JavaScript/TypeScript
Useconstby default,letonly when reassignment is needed, nevervarin JavaScript/TypeScript
Alwaysawaitpromises in async functions - don't forget to use the return value in JavaScript/TypeScript
Useasync/awaitsyntax instead of promise chains for better readability in JavaScript/TypeScript
Handle errors appropriately in async code with try-catch blocks in JavaScript/TypeScript
Don't use async functions as Promise executors in JavaScript/TypeScript
Removeconsole.log,debugger, andalertstatements from production code in JavaScript/TypeScript
ThrowErrorobjects with descriptive messages, not strings or other values in JavaScript/TypeScript
Usetry-catchblocks meaningfully - don't catch errors just to rethrow them in JavaScript/TypeScript
Prefer early returns over nested conditionals for error cases in JavaScript/TypeScript
Keep functions focused and under reasonable cognitive complexity limits
Extract complex conditions into well-named boolean variables in JavaScript/TypeScript
Use early returns to reduce nesting in JavaScript/TypeScript
Prefer simple conditionals over nested ternary operators in JavaScript/TypeScript
Group related code together and separate concerns in JavaScript/TypeScript
Don't useeval()or assign directly todocument.cookiein JavaScript/TypeScript
Validate and sanitize user input in JavaScript/TypeScript
Avoid spread syntax in accumulators within loops in JavaScript/Ty...
Files:
packages/spectrum-ts/src/providers/imessage/index.tspackages/spectrum-ts/src/providers/imessage/remote/typing-events.tspackages/spectrum-ts/src/platform/define.tspackages/spectrum-ts/src/platform/types.tspackages/spectrum-ts/src/spectrum.ts
| const eventCache = new Map<string, AsyncIterable<[Space, unknown]>>(); | ||
| for (const eventName of Object.keys(def.events)) { | ||
| if (eventName === "messages") { | ||
| continue; | ||
| } | ||
| const producer = def.events[eventName] as | ||
| | ((ctx: { client: unknown; config: unknown }) => AsyncIterable<unknown>) | ||
| | undefined; | ||
| if (producer) { | ||
| eventProperties[eventName] = producer({ | ||
| client: runtime.client, | ||
| config: runtime.config, | ||
| }); | ||
| } | ||
| Object.defineProperty(base, eventName, { | ||
| enumerable: true, | ||
| get() { | ||
| let cached = eventCache.get(eventName); | ||
| if (!cached) { | ||
| const subscription = runtime.subscribeEvent(eventName); | ||
| if (!subscription) { | ||
| return; | ||
| } | ||
| cached = subscription; | ||
| eventCache.set(eventName, cached); | ||
| } | ||
| return cached; |
There was a problem hiding this comment.
Don't cache the per-access event subscription.
runtime.subscribeEvent(eventName) already gives you a fresh broadcaster consumer. Caching that object here means the second imessage(app).typing read reuses the same downstream stream instead of getting its own subscription, so one closed/exhausted loop can starve later consumers.
Suggested fix
- const eventCache = new Map<string, AsyncIterable<[Space, unknown]>>();
for (const eventName of Object.keys(def.events)) {
Object.defineProperty(base, eventName, {
enumerable: true,
get() {
- let cached = eventCache.get(eventName);
- if (!cached) {
- const subscription = runtime.subscribeEvent(eventName);
- if (!subscription) {
- return;
- }
- cached = subscription;
- eventCache.set(eventName, cached);
- }
- return cached;
+ return runtime.subscribeEvent(eventName);
},
});
}📝 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.
| const eventCache = new Map<string, AsyncIterable<[Space, unknown]>>(); | |
| for (const eventName of Object.keys(def.events)) { | |
| if (eventName === "messages") { | |
| continue; | |
| } | |
| const producer = def.events[eventName] as | |
| | ((ctx: { client: unknown; config: unknown }) => AsyncIterable<unknown>) | |
| | undefined; | |
| if (producer) { | |
| eventProperties[eventName] = producer({ | |
| client: runtime.client, | |
| config: runtime.config, | |
| }); | |
| } | |
| Object.defineProperty(base, eventName, { | |
| enumerable: true, | |
| get() { | |
| let cached = eventCache.get(eventName); | |
| if (!cached) { | |
| const subscription = runtime.subscribeEvent(eventName); | |
| if (!subscription) { | |
| return; | |
| } | |
| cached = subscription; | |
| eventCache.set(eventName, cached); | |
| } | |
| return cached; | |
| for (const eventName of Object.keys(def.events)) { | |
| Object.defineProperty(base, eventName, { | |
| enumerable: true, | |
| get() { | |
| return runtime.subscribeEvent(eventName); | |
| }, | |
| }); | |
| } |
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.
In `@packages/spectrum-ts/src/platform/define.ts` around lines 146 - 160, The
getter for dynamic events is incorrectly caching the subscription in eventCache
so subsequent reads reuse the same AsyncIterable; remove the per-access caching:
delete the eventCache Map and its get/set calls and change the property getter
on base (the Object.defineProperty for each eventName in def.events) to always
call runtime.subscribeEvent(eventName) and return that value (still handle a
falsy/undefined return from runtime.subscribeEvent by returning undefined). This
ensures each access to base[eventName] (e.g., imessage(app).typing) receives a
fresh broadcaster consumer.
| stream<IMessageTypingRecord>((emit, end) => { | ||
| let active = client.chats.subscribe(); |
There was a problem hiding this comment.
Handle client.chats.subscribe() failures inside the retry loop.
Line 62 and Line 106 can throw before runTypingSubscription(...) is entered, so those failures bypass logTypingStreamError, skip backoff, and tear the stream down permanently. That breaks the resilience this helper is supposed to provide.
Suggested fix
- let active = client.chats.subscribe();
+ let active:
+ | ReturnType<AdvancedIMessage["chats"]["subscribe"]>
+ | undefined;
@@
const pump = (async () => {
while (!closed) {
try {
+ active = client.chats.subscribe();
await runTypingSubscription(active, emit, () => {
retryDelayMs = RECONNECT_INITIAL_DELAY_MS;
});
} catch (e) {
if (!closed) {
logTypingStreamError(e);
}
} finally {
- await active.close();
+ await active?.close();
}
if (!closed) {
await sleep(retryDelayMs);
retryDelayMs = Math.min(retryDelayMs * 2, RECONNECT_MAX_DELAY_MS);
- active = client.chats.subscribe();
}
}
end();
})();As per coding guidelines, Handle errors appropriately in async code with try-catch blocks in JavaScript/TypeScript.
Also applies to: 103-106
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.
In `@packages/spectrum-ts/src/providers/imessage/remote/typing-events.ts` around
lines 61 - 62, client.chats.subscribe() can throw before entering
runTypingSubscription, bypassing the retry/backoff and logTypingStreamError;
wrap the subscribe call in a try/catch inside the existing retry loop (the same
loop that calls runTypingSubscription) so any error from
client.chats.subscribe() is caught, forwarded to logTypingStreamError, triggers
the backoff/retry posture, and does not escape to tear the stream down. Apply
the same try/catch pattern to both occurrences where active =
client.chats.subscribe() appears, ensuring you only call runTypingSubscription
when subscribe succeeded and that caught errors loop back into the retry logic.
Summary
subscribeMessages/ custom-event-streams paths into a singlesubscribeEvent(name)entry point onPlatformRuntime. All events — includingmessages— flow through one broadcaster-per-(platform, event)map, share one upstream subscription across consumers, and yield uniform[Space, payload]tuples (the producer'sspacefield is stripped from the payload half).CustomEventStreams, the lazyProxyonSpectrumInstance,createCustomEventStream,mergedEventStreams). Non-messagesevents are now consumed only through the platform-specific accessor (e.g.imessage(app).typing); theapp.<event>shape is gone.typingevent backed by a new resilientsubscribeTypingstream — wrapsclient.chats.subscribe(), filters forchat.typingIndicator, and reconnects on failure with exponential backoff + jitter (sharedRECONNECT_INITIAL_DELAY_MS/RECONNECT_MAX_DELAY_MSfromresumable-stream). Local clients yield an empty stream.EventProducerso every event payload must extendProviderEventRecord({ space: { id } }), and updatesPlatformInstanceto surface[PlatformSpace<Def>, EventPayload<...>]for non-messagesevents.Usage
Note: a previous
app.typingshape (the spectrum-level proxy) is removed in this PR — consumers must go throughimessage(app).typing.Changes
packages/spectrum-ts/src/spectrum.tsmessageBroadcasters+customEventStreamswith oneeventBroadcastersmap keyed byplatform:eventName. NewcreateProviderEventStream/getOrCreateEventBroadcasthandle every event uniformly;messagesstill runs throughwrapProviderMessageandflattenGroupswhile other events yield the raw record minusspace. ThecustomEventProxyand outerProxywrapper around the returned instance are removed.packages/spectrum-ts/src/platform/types.tsProviderEventRecordandEventPayload<P>; tightensEventProducerandPlatformDef.eventsindex signature to require records that carryspace. ReplacessubscribeMessagesonPlatformRuntimewithsubscribeEvent(name). UpdatesPlatformInstancemapped type to yield[PlatformSpace<Def>, EventPayload<...>]per event. Drops the unusedExtractCustomEventNames/ToCustomEventVariant/AllCustomEventNames/UnifiedCustomEvent/CustomEventStreamsmachinery.packages/spectrum-ts/src/platform/define.tsPlatformInstanceaccessors are now generated for every event (includingmessages) by lazily callingruntime.subscribeEvent(name)and caching the resulting iterable per-name, so a secondfor awaitreuses the broadcaster's subscription. The old eagereventProperties/ dedicatedmessagesObject.definePropertyare removed.packages/spectrum-ts/src/providers/imessage/remote/typing-events.tssubscribeTyping(clients)returns a mergedManagedStream<IMessageTypingRecord>. Per-client stream filterschat.typingIndicator, emits{ space: { id: chatGuid }, isTyping, displayName, timestamp }, and reconnects with backoff (resets the delay after any successful event). Errors are logged, never propagated.packages/spectrum-ts/src/providers/imessage/index.tstypingevent into the iMessage platform definition (remote only — local yields an empty async iterable).Behavior notes
isFromMeflag on the typing indicator. Consumers that need to ignore self-typing should branch ondisplayName.subscribeEvent(name)returnsundefinedwhen the platform doesn't define that event, so the merged messages stream and platform-level accessors silently skip unsupported events instead of throwing.Test plan
bun x ultracite checkpassesbun run typecheckpassesimessage(app).typing, confirmisTypingtoggles when a peer starts/stops typingchats.subscribe()connection mid-iteration, confirm the typing stream reconnects with backoff and resumes emittingfor await (const [space, msg] of app.messages)still works across iMessage + WhatsApp providersfor await (...of imessage(app).messages)reuses the broadcaster (no duplicate upstream subscription)app.stop()shuts down the typing stream cleanlyNeed help on this PR? Tag
@codesmithwith what you need.Summary by CodeRabbit
New Features
Improvements