Add trailConvo method

src/state/messages/events/agent.ts

@@ -44,6 +44,7 @@ export class MessagesEventBus {
     this.resume = this.resume.bind(this)
     this.setPollInterval = this.setPollInterval.bind(this)
     this.trail = this.trail.bind(this)
+    this.trailConvo = this.trailConvo.bind(this)
   }
 
   private commit() {
@@ -79,6 +80,7 @@ export class MessagesEventBus {
           error: undefined,
           setPollInterval: this.setPollInterval,
           trail: this.trail,
+          trailConvo: this.trailConvo,
         }
       }
       case MessagesEventBusStatus.Ready: {
@@ -88,6 +90,7 @@ export class MessagesEventBus {
           error: undefined,
           setPollInterval: this.setPollInterval,
           trail: this.trail,
+          trailConvo: this.trailConvo,
         }
       }
       case MessagesEventBusStatus.Suspended: {
@@ -97,6 +100,7 @@ export class MessagesEventBus {
           error: undefined,
           setPollInterval: this.setPollInterval,
           trail: this.trail,
+          trailConvo: this.trailConvo,
         }
       }
       case MessagesEventBusStatus.Error: {
@@ -111,6 +115,7 @@ export class MessagesEventBus {
           },
           setPollInterval: this.setPollInterval,
           trail: this.trail,
+          trailConvo: this.trailConvo,
         }
       }
       default: {
@@ -120,6 +125,7 @@ export class MessagesEventBus {
           error: undefined,
           setPollInterval: this.setPollInterval,
           trail: this.trail,
+          trailConvo: this.trailConvo,
         }
       }
     }
@@ -312,6 +318,29 @@ export class MessagesEventBus {
     }
   }
 
+  trailConvo(
+    convoId: string,
+    handler: (events: ChatBskyConvoGetLog.OutputSchema['logs']) => void,
+  ) {
+    const handle = (events: ChatBskyConvoGetLog.OutputSchema['logs']) => {
+      const convoEvents = events.filter(ev => {
+        if (typeof ev.convoId === 'string' && ev.convoId === convoId) {
+          return ev.convoId === convoId
+        }
+        return false
+      })
+
+      if (convoEvents.length > 0) {
+        handler(convoEvents)
+      }
+    }
+
+    this.emitter.on('events', handle)
+    return () => {
+      this.emitter.off('events', handle)
+    }
+  }
+
   private async initializeLatestRev() {
     logger.debug(
       `${LOGGER_CONTEXT}: initialize latest rev`,

src/state/messages/events/types.ts

@@ -56,58 +56,56 @@ export type MessagesEventBusDispatch =
       payload: MessagesEventBusError
     }
 
+export type TrailHandler = (
+  events: ChatBskyConvoGetLog.OutputSchema['logs'],
+) => void
+
 export type MessagesEventBusState =
   | {
       status: MessagesEventBusStatus.Uninitialized
       rev: undefined
       error: undefined
       setPollInterval: (interval: number) => void
-      trail: (
+      trail: (handler: TrailHandler) => () => void
-        handler: (events: ChatBskyConvoGetLog.OutputSchema['logs']) => void,
+      trailConvo: (convoId: string, handler: TrailHandler) => () => void
-      ) => () => void
     }
   | {
       status: MessagesEventBusStatus.Initializing
       rev: undefined
       error: undefined
       setPollInterval: (interval: number) => void
-      trail: (
+      trail: (handler: TrailHandler) => () => void
-        handler: (events: ChatBskyConvoGetLog.OutputSchema['logs']) => void,
+      trailConvo: (convoId: string, handler: TrailHandler) => () => void
-      ) => () => void
     }
   | {
       status: MessagesEventBusStatus.Ready
       rev: string
       error: undefined
       setPollInterval: (interval: number) => void
-      trail: (
+      trail: (handler: TrailHandler) => () => void
-        handler: (events: ChatBskyConvoGetLog.OutputSchema['logs']) => void,
+      trailConvo: (convoId: string, handler: TrailHandler) => () => void
-      ) => () => void
     }
   | {
       status: MessagesEventBusStatus.Backgrounded
       rev: string | undefined
       error: undefined
       setPollInterval: (interval: number) => void
-      trail: (
+      trail: (handler: TrailHandler) => () => void
-        handler: (events: ChatBskyConvoGetLog.OutputSchema['logs']) => void,
+      trailConvo: (convoId: string, handler: TrailHandler) => () => void
-      ) => () => void
     }
   | {
       status: MessagesEventBusStatus.Suspended
       rev: string | undefined
       error: undefined
       setPollInterval: (interval: number) => void
-      trail: (
+      trail: (handler: TrailHandler) => () => void
-        handler: (events: ChatBskyConvoGetLog.OutputSchema['logs']) => void,
+      trailConvo: (convoId: string, handler: TrailHandler) => () => void
-      ) => () => void
     }
   | {
       status: MessagesEventBusStatus.Error
       rev: string | undefined
       error: MessagesEventBusError
       setPollInterval: (interval: number) => void
-      trail: (
+      trail: (handler: TrailHandler) => () => void
-        handler: (events: ChatBskyConvoGetLog.OutputSchema['logs']) => void,
+      trailConvo: (convoId: string, handler: TrailHandler) => () => void
-      ) => () => void
     }