feat: Subscription refactor (#333)

Inspiration for this from the tRPC example repo: https://github.com/trpc/examples-next-sse-chat/blob/ee39f922e1f0fab6ea8598fe14916ca2c1cd444b/src/server/routers/channel.ts#L13

Author: charlesvien

ARCHITECTURE.md

@@ -349,18 +349,18 @@ export class MyService extends TypedEventEmitter<MyServiceEvents> {
 
 ### 3. Create Subscriptions in Router
 
-Use a helper to reduce boilerplate. For global events (broadcast to all subscribers):
+Use `toIterable()` on the service to convert events to an async iterable. For global events (broadcast to all subscribers):
 
 ```typescript
 // src/main/trpc/routers/my-router.ts
-import { on } from "node:events";
 import { MyServiceEvent, type MyServiceEvents } from "../../services/my-service/schemas";
 
 function subscribe<K extends keyof MyServiceEvents>(event: K) {
   return publicProcedure.subscription(async function* (opts) {
     const service = getService();
-    for await (const [payload] of on(service, event, { signal: opts.signal })) {
-      yield payload as MyServiceEvents[K];
+    const iterable = service.toIterable(event, { signal: opts.signal });
+    for await (const data of iterable) {
+      yield data;
     }
   });
 }
@@ -382,15 +382,15 @@ export interface ShellEvents {
 }
 
 // Router filters events to the specific session
-function subscribeToSession<K extends keyof ShellEvents>(event: K) {
+function subscribeFiltered<K extends keyof ShellEvents>(event: K) {
   return publicProcedure
     .input(sessionIdInput)
     .subscription(async function* (opts) {
       const service = getService();
       const targetSessionId = opts.input.sessionId;
+      const iterable = service.toIterable(event, { signal: opts.signal });
 
-      for await (const [payload] of on(service, event, { signal: opts.signal })) {
-        const data = payload as ShellEvents[K];
+      for await (const data of iterable) {
         if (data.sessionId === targetSessionId) {
           yield data;
         }
@@ -399,8 +399,8 @@ function subscribeToSession<K extends keyof ShellEvents>(event: K) {
 }
 
 export const shellRouter = router({
-  onData: subscribeToSession(ShellEvent.Data),
-  onExit: subscribeToSession(ShellEvent.Exit),
+  onData: subscribeFiltered(ShellEvent.Data),
+  onExit: subscribeFiltered(ShellEvent.Exit),
 });
 ```
 

apps/array/package.json

@@ -110,9 +110,9 @@
     "@tiptap/react": "^3.11.0",
     "@tiptap/starter-kit": "^3.11.0",
     "@tiptap/suggestion": "^3.11.0",
-    "@trpc/client": "^11.7.2",
-    "@trpc/react-query": "^11.7.2",
-    "@trpc/server": "^11.7.2",
+    "@trpc/client": "^11.8.0",
+    "@trpc/react-query": "^11.8.0",
+    "@trpc/server": "^11.8.0",
     "@xterm/addon-fit": "^0.10.0",
     "@xterm/addon-serialize": "^0.13.0",
     "@xterm/addon-web-links": "^0.11.0",

apps/array/src/main/lib/typed-event-emitter.ts

@@ -1,4 +1,4 @@
-import { EventEmitter } from "node:events";
+import { EventEmitter, on } from "node:events";
 
 export class TypedEventEmitter<TEvents> extends EventEmitter {
   emit<K extends keyof TEvents & string>(
@@ -21,4 +21,13 @@ export class TypedEventEmitter<TEvents> extends EventEmitter {
   ): this {
     return super.off(event, listener);
   }
+
+  async *toIterable<K extends keyof TEvents & string>(
+    event: K,
+    opts?: { signal?: AbortSignal },
+  ): AsyncIterable<TEvents[K]> {
+    for await (const [payload] of on(this, event, opts)) {
+      yield payload as TEvents[K];
+    }
+  }
 }

apps/array/src/main/services/types.ts

@@ -2,5 +2,3 @@
  * Auto-import all IPC service types for declaration merging
  * This file is auto-generated by vite-plugin-auto-services.ts
  */
-
-

apps/array/src/main/services/ui/schemas.ts

@@ -6,9 +6,10 @@ export const UIServiceEvent = {
   ClearStorage: "clear-storage",
 } as const;
 
+// UI events are simple signals - payload is just a marker that the event fired
 export interface UIServiceEvents {
-  [UIServiceEvent.OpenSettings]: undefined;
-  [UIServiceEvent.NewTask]: undefined;
-  [UIServiceEvent.ResetLayout]: undefined;
-  [UIServiceEvent.ClearStorage]: undefined;
+  [UIServiceEvent.OpenSettings]: true;
+  [UIServiceEvent.NewTask]: true;
+  [UIServiceEvent.ResetLayout]: true;
+  [UIServiceEvent.ClearStorage]: true;
 }

apps/array/src/main/services/ui/service.ts

@@ -9,18 +9,18 @@ import { UIServiceEvent, type UIServiceEvents } from "./schemas.js";
 @injectable()
 export class UIService extends TypedEventEmitter<UIServiceEvents> {
   openSettings(): void {
-    this.emit(UIServiceEvent.OpenSettings, undefined);
+    this.emit(UIServiceEvent.OpenSettings, true);
   }
 
   newTask(): void {
-    this.emit(UIServiceEvent.NewTask, undefined);
+    this.emit(UIServiceEvent.NewTask, true);
   }
 
   resetLayout(): void {
-    this.emit(UIServiceEvent.ResetLayout, undefined);
+    this.emit(UIServiceEvent.ResetLayout, true);
   }
 
   clearStorage(): void {
-    this.emit(UIServiceEvent.ClearStorage, undefined);
+    this.emit(UIServiceEvent.ClearStorage, true);
   }
 }

apps/array/src/main/services/updates/schemas.ts

@@ -31,7 +31,7 @@ export type UpdatesStatusPayload = {
 };
 
 export interface UpdatesEvents {
-  [UpdatesEvent.Ready]: undefined;
+  [UpdatesEvent.Ready]: true;
   [UpdatesEvent.Status]: UpdatesStatusPayload;
-  [UpdatesEvent.CheckFromMenu]: undefined;
+  [UpdatesEvent.CheckFromMenu]: true;
 }

apps/array/src/main/services/updates/service.ts

@@ -55,7 +55,7 @@ export class UpdatesService extends TypedEventEmitter<UpdatesEvents> {
   }
 
   triggerMenuCheck(): void {
-    this.emit(UpdatesEvent.CheckFromMenu, undefined);
+    this.emit(UpdatesEvent.CheckFromMenu, true);
   }
 
   checkForUpdates(): CheckForUpdatesOutput {
@@ -120,7 +120,7 @@ export class UpdatesService extends TypedEventEmitter<UpdatesEvents> {
 
   private flushPendingNotification(): void {
     if (this.updateReady && this.pendingNotification) {
-      this.emit(UpdatesEvent.Ready, undefined);
+      this.emit(UpdatesEvent.Ready, true);
       this.pendingNotification = false;
     }
   }

apps/array/src/main/trpc/routers/agent.ts

@@ -1,10 +1,8 @@
-import { on } from "node:events";
 import type { ContentBlock } from "@agentclientprotocol/sdk";
 import { container } from "../../di/container.js";
 import { MAIN_TOKENS } from "../../di/tokens.js";
 import {
   AgentServiceEvent,
-  type AgentSessionEventPayload,
   cancelPromptInput,
   cancelSessionInput,
   promptInput,
@@ -64,14 +62,11 @@ export const agentRouter = router({
     .subscription(async function* (opts) {
       const service = getService();
       const targetSessionId = opts.input.sessionId;
-      const options = opts.signal ? { signal: opts.signal } : undefined;
+      const iterable = service.toIterable(AgentServiceEvent.SessionEvent, {
+        signal: opts.signal,
+      });
 
-      for await (const [payload] of on(
-        service,
-        AgentServiceEvent.SessionEvent,
-        options,
-      )) {
-        const event = payload as AgentSessionEventPayload;
+      for await (const event of iterable) {
         if (event.sessionId === targetSessionId) {
           yield event.payload;
         }

apps/array/src/main/trpc/routers/deep-link.ts

@@ -1,9 +1,7 @@
-import { on } from "node:events";
 import { container } from "../../di/container.js";
 import { MAIN_TOKENS } from "../../di/tokens.js";
 import {
   TaskLinkEvent,
-  type TaskLinkEvents,
   type TaskLinkService,
 } from "../../services/task-link/service.js";
 import { publicProcedure, router } from "../trpc.js";
@@ -18,13 +16,11 @@ export const deepLinkRouter = router({
    */
   onOpenTask: publicProcedure.subscription(async function* (opts) {
     const service = getService();
-    const options = opts.signal ? { signal: opts.signal } : undefined;
-    for await (const [payload] of on(
-      service,
-      TaskLinkEvent.OpenTask,
-      options,
-    )) {
-      yield payload as TaskLinkEvents[typeof TaskLinkEvent.OpenTask];
+    const iterable = service.toIterable(TaskLinkEvent.OpenTask, {
+      signal: opts.signal,
+    });
+    for await (const data of iterable) {
+      yield data;
     }
   }),