Integrate event bus

src/state/messages/convo/agent.ts

@@ -19,9 +19,7 @@ import {
   ConvoState,
   ConvoStatus,
 } from '#/state/messages/convo/types'
-
+import {MessagesEventBusError} from '#/state/messages/events/types'
-const ACTIVE_POLL_INTERVAL = 1e3
-const BACKGROUND_POLL_INTERVAL = 10e3
 
 // TODO temporary
 let DEBUG_ACTIVE_CHAT: string | undefined
@@ -44,7 +42,6 @@ export class Convo {
   private __tempFromUserDid: string
 
   private status: ConvoStatus = ConvoStatus.Uninitialized
-  private pollInterval = ACTIVE_POLL_INTERVAL
   private error:
     | {
         code: ConvoErrorCode
@@ -73,7 +70,6 @@ export class Convo {
   private headerItems: Map<string, ConvoItem> = new Map()
 
   private isProcessingPendingMessages = false
-  private nextPoll: NodeJS.Timeout | undefined
 
   convoId: string
   convo: ChatBskyConvoDefs.ConvoView | undefined
@@ -92,6 +88,9 @@ export class Convo {
     this.sendMessage = this.sendMessage.bind(this)
     this.deleteMessage = this.deleteMessage.bind(this)
     this.fetchMessageHistory = this.fetchMessageHistory.bind(this)
+    this.ingestFirehose = this.ingestFirehose.bind(this)
+    this.onFirehoseConnect = this.onFirehoseConnect.bind(this)
+    this.onFirehoseError = this.onFirehoseError.bind(this)
 
     if (DEBUG_ACTIVE_CHAT) {
       logger.error(`Convo: another chat was already active`, {
@@ -207,18 +206,12 @@ export class Convo {
         switch (action.event) {
           case ConvoDispatchEvent.Ready: {
             this.status = ConvoStatus.Ready
-            this.pollInterval = ACTIVE_POLL_INTERVAL
+            this.fetchMessageHistory()
-            this.fetchMessageHistory().then(() => {
-              this.restartPoll()
-            })
             break
           }
           case ConvoDispatchEvent.Background: {
             this.status = ConvoStatus.Backgrounded
-            this.pollInterval = BACKGROUND_POLL_INTERVAL
+            this.fetchMessageHistory()
-            this.fetchMessageHistory().then(() => {
-              this.restartPoll()
-            })
             break
           }
           case ConvoDispatchEvent.Suspend: {
@@ -237,24 +230,19 @@ export class Convo {
         switch (action.event) {
           case ConvoDispatchEvent.Resume: {
             this.refreshConvo()
-            this.restartPoll()
             break
           }
           case ConvoDispatchEvent.Background: {
             this.status = ConvoStatus.Backgrounded
-            this.pollInterval = BACKGROUND_POLL_INTERVAL
-            this.restartPoll()
             break
           }
           case ConvoDispatchEvent.Suspend: {
             this.status = ConvoStatus.Suspended
-            this.cancelNextPoll()
             break
           }
           case ConvoDispatchEvent.Error: {
             this.status = ConvoStatus.Error
             this.error = action.payload
-            this.cancelNextPoll()
             break
           }
         }
@@ -262,23 +250,24 @@ export class Convo {
       }
       case ConvoStatus.Backgrounded: {
         switch (action.event) {
+          // TODO truncate history if needed
           case ConvoDispatchEvent.Resume: {
-            this.status = ConvoStatus.Ready
+            if (this.convo) {
-            this.pollInterval = ACTIVE_POLL_INTERVAL
+              this.status = ConvoStatus.Ready
-            this.refreshConvo()
+              this.refreshConvo()
-            // TODO truncate history if needed
+            } else {
-            this.restartPoll()
+              this.status = ConvoStatus.Initializing
+              this.setup()
+            }
             break
           }
           case ConvoDispatchEvent.Suspend: {
             this.status = ConvoStatus.Suspended
-            this.cancelNextPoll()
             break
           }
           case ConvoDispatchEvent.Error: {
             this.status = ConvoStatus.Error
             this.error = action.payload
-            this.cancelNextPoll()
             break
           }
         }
@@ -286,19 +275,20 @@ export class Convo {
       }
       case ConvoStatus.Suspended: {
         switch (action.event) {
+          // TODO truncate history if needed
           case ConvoDispatchEvent.Init: {
-            this.status = ConvoStatus.Ready
+            if (this.convo) {
-            this.pollInterval = ACTIVE_POLL_INTERVAL
+              this.status = ConvoStatus.Ready
-            this.refreshConvo()
+              this.refreshConvo()
-            // TODO truncate history if needed
+            } else {
-            this.restartPoll()
+              this.status = ConvoStatus.Initializing
+              this.setup()
+            }
             break
           }
           case ConvoDispatchEvent.Resume: {
             this.status = ConvoStatus.Ready
-            this.pollInterval = ACTIVE_POLL_INTERVAL
             this.refreshConvo()
-            this.restartPoll()
             break
           }
           case ConvoDispatchEvent.Error: {
@@ -576,84 +566,25 @@ export class Convo {
     }
   }
 
-  private restartPoll() {
+  onFirehoseConnect() {
-    this.cancelNextPoll()
+    this.footerItems.delete(ConvoItemError.PollFailed)
-    this.pollLatestEvents()
+    this.commit()
   }
 
-  private cancelNextPoll() {
+  onFirehoseError(error?: MessagesEventBusError) {
-    if (this.nextPoll) clearTimeout(this.nextPoll)
+    this.footerItems.set(ConvoItemError.PollFailed, {
-  }
+      type: 'error-recoverable',
-
+      key: ConvoItemError.PollFailed,
-  private pollLatestEvents() {
+      code: ConvoItemError.PollFailed,
-    /*
+      retry: () => {
-     * Uncomment to view poll events
+        this.footerItems.delete(ConvoItemError.PollFailed)
-     */
+        this.commit()
-    logger.debug('Convo: poll events', {id: this.id}, logger.DebugContext.convo)
+        error?.retry()
-
+      },
-    try {
-      this.fetchLatestEvents().then(({events}) => {
-        this.applyLatestEvents(events)
-      })
-      this.nextPoll = setTimeout(() => {
-        this.pollLatestEvents()
-      }, this.pollInterval)
-    } catch (e: any) {
-      logger.error('Convo: poll events failed')
-
-      this.cancelNextPoll()
-
-      this.footerItems.set(ConvoItemError.PollFailed, {
-        type: 'error-recoverable',
-        key: ConvoItemError.PollFailed,
-        code: ConvoItemError.PollFailed,
-        retry: () => {
-          this.footerItems.delete(ConvoItemError.PollFailed)
-          this.commit()
-          this.pollLatestEvents()
-        },
-      })
-
-      this.commit()
-    }
-  }
-
-  private pendingFetchLatestEvents:
-    | Promise<{
-        events: ChatBskyConvoGetLog.OutputSchema['logs']
-      }>
-    | undefined
-  async fetchLatestEvents() {
-    if (this.pendingFetchLatestEvents) return this.pendingFetchLatestEvents
-
-    this.pendingFetchLatestEvents = new Promise<{
-      events: ChatBskyConvoGetLog.OutputSchema['logs']
-    }>(async (resolve, reject) => {
-      try {
-        // throw new Error('UNCOMMENT TO TEST POLL FAILURE')
-        const response = await this.agent.api.chat.bsky.convo.getLog(
-          {
-            cursor: this.eventsCursor,
-          },
-          {
-            headers: {
-              Authorization: this.__tempFromUserDid,
-            },
-          },
-        )
-        const {logs} = response.data
-        resolve({events: logs})
-      } catch (e) {
-        reject(e)
-      } finally {
-        this.pendingFetchLatestEvents = undefined
-      }
     })
-
-    return this.pendingFetchLatestEvents
   }
 
-  private applyLatestEvents(events: ChatBskyConvoGetLog.OutputSchema['logs']) {
+  ingestFirehose(events: ChatBskyConvoGetLog.OutputSchema['logs']) {
     let needsCommit = false
 
     for (const ev of events) {

src/state/messages/convo/const.ts

@@ -0,0 +1 @@
+export const ACTIVE_POLL_INTERVAL = 1e3

src/state/messages/convo/index.tsx

@@ -4,7 +4,9 @@ import {BskyAgent} from '@atproto-labs/api'
 import {useFocusEffect, useIsFocused} from '@react-navigation/native'
 
 import {Convo} from '#/state/messages/convo/agent'
+import {ACTIVE_POLL_INTERVAL} from '#/state/messages/convo/const'
 import {ConvoParams, ConvoState} from '#/state/messages/convo/types'
+import {useMessagesEventBus} from '#/state/messages/events'
 import {useMarkAsReadMutation} from '#/state/queries/messages/conversation'
 import {useAgent} from '#/state/session'
 import {useDmServiceUrlStorage} from '#/screens/Messages/Temp/useDmServiceUrlStorage'
@@ -23,6 +25,7 @@ export function ConvoProvider({
   children,
   convoId,
 }: Pick<ConvoParams, 'convoId'> & {children: React.ReactNode}) {
+  const requestedPollInterval = React.useRef<(() => void) | void>()
   const isScreenFocused = useIsFocused()
   const {serviceUrl} = useDmServiceUrlStorage()
   const {getAgent} = useAgent()
@@ -38,17 +41,36 @@ export function ConvoProvider({
   )
   const service = useSyncExternalStore(convo.subscribe, convo.getSnapshot)
   const {mutate: markAsRead} = useMarkAsReadMutation()
+  const events = useMessagesEventBus()
+
+  React.useEffect(() => {
+    const remove = events.trailConvo(convoId, events => {
+      convo.ingestFirehose(events)
+    })
+    return () => {
+      remove()
+    }
+  }, [convoId, convo, events])
 
   useFocusEffect(
     React.useCallback(() => {
+      if (!requestedPollInterval.current) {
+        requestedPollInterval.current =
+          events.requestPollInterval(ACTIVE_POLL_INTERVAL)
+      }
+
       convo.resume()
       markAsRead({convoId})
 
       return () => {
+        if (requestedPollInterval.current) {
+          requestedPollInterval.current = requestedPollInterval.current()
+        }
+
         convo.background()
         markAsRead({convoId})
       }
-    }, [convo, convoId, markAsRead]),
+    }, [convo, convoId, markAsRead, events]),
   )
 
   React.useEffect(() => {
@@ -56,8 +78,16 @@ export function ConvoProvider({
       if (isScreenFocused) {
         if (nextAppState === 'active') {
           convo.resume()
+
+          if (!requestedPollInterval.current) {
+            requestedPollInterval.current =
+              events.requestPollInterval(ACTIVE_POLL_INTERVAL)
+          }
         } else {
           convo.background()
+          if (requestedPollInterval.current) {
+            requestedPollInterval.current = requestedPollInterval.current()
+          }
         }
 
         markAsRead({convoId})
@@ -69,7 +99,7 @@ export function ConvoProvider({
     return () => {
       sub.remove()
     }
-  }, [convoId, convo, isScreenFocused, markAsRead])
+  }, [convoId, convo, isScreenFocused, markAsRead, events])
 
   return <ChatContext.Provider value={service}>{children}</ChatContext.Provider>
 }