Move events mgmt into Convo class

src/state/messages/convo/agent.ts

@@ -9,6 +9,10 @@ import {nanoid} from 'nanoid/non-secure'
 
 import {logger} from '#/logger'
 import {isNative} from '#/platform/detection'
+import {
+  ACTIVE_POLL_INTERVAL,
+  BACKGROUND_POLL_INTERVAL,
+} from '#/state/messages/convo/const'
 import {
   ConvoDispatch,
   ConvoDispatchEvent,
@@ -19,6 +23,7 @@ import {
   ConvoState,
   ConvoStatus,
 } from '#/state/messages/convo/types'
+import {MessagesEventBus} from '#/state/messages/events/agent'
 import {MessagesEventBusError} from '#/state/messages/events/types'
 
 // TODO temporary
@@ -39,6 +44,7 @@ export class Convo {
   private id: string
 
   private agent: BskyAgent
+  private events: MessagesEventBus
   private __tempFromUserDid: string
 
   private status: ConvoStatus = ConvoStatus.Uninitialized
@@ -49,9 +55,9 @@ export class Convo {
         retry: () => void
       }
     | undefined
-  private historyCursor: string | undefined | null = undefined
+  private oldestRev: string | undefined | null = undefined
   private isFetchingHistory = false
-  private eventsCursor: string | undefined = undefined
+  private latestRev: string | undefined = undefined
 
   private pastMessages: Map<
     string,
@@ -81,6 +87,7 @@ export class Convo {
     this.id = nanoid(3)
     this.convoId = params.convoId
     this.agent = params.agent
+    this.events = params.events
     this.__tempFromUserDid = params.__tempFromUserDid
 
     this.subscribe = this.subscribe.bind(this)
@@ -99,6 +106,12 @@ export class Convo {
     } else {
       DEBUG_ACTIVE_CHAT = this.convoId
     }
+
+    this.events.trailConvo(this.convoId, events => {
+      this.ingestFirehose(events)
+    })
+    this.events.onConnect(this.onFirehoseConnect)
+    this.events.onError(this.onFirehoseError)
   }
 
   private commit() {
@@ -346,8 +359,8 @@ export class Convo {
 
     this.status = ConvoStatus.Uninitialized
     this.error = undefined
-    this.historyCursor = undefined
+    this.oldestRev = undefined
-    this.eventsCursor = undefined
+    this.latestRev = undefined
 
     this.pastMessages = new Map()
     this.newMessages = new Map()
@@ -401,21 +414,36 @@ export class Convo {
 
   init() {
     this.dispatch({event: ConvoDispatchEvent.Init})
+    this.requestPollInterval(ACTIVE_POLL_INTERVAL)
   }
 
   resume() {
     this.dispatch({event: ConvoDispatchEvent.Resume})
+    this.requestPollInterval(ACTIVE_POLL_INTERVAL)
   }
 
   background() {
     this.dispatch({event: ConvoDispatchEvent.Background})
+    this.requestPollInterval(BACKGROUND_POLL_INTERVAL)
   }
 
   suspend() {
     this.dispatch({event: ConvoDispatchEvent.Suspend})
+    this.withdrawRequestedPollInterval()
     DEBUG_ACTIVE_CHAT = undefined
   }
 
+  private requestedPollInterval: (() => void) | undefined
+  private requestPollInterval(interval: number) {
+    this.withdrawRequestedPollInterval()
+    this.requestedPollInterval = this.events.requestPollInterval(interval)
+  }
+  private withdrawRequestedPollInterval() {
+    if (this.requestedPollInterval) {
+      this.requestedPollInterval()
+    }
+  }
+
   private pendingFetchConvo:
     | Promise<{
         convo: ChatBskyConvoDefs.ConvoView
@@ -489,9 +517,9 @@ export class Convo {
     logger.debug('Convo: fetch message history', {}, logger.DebugContext.convo)
 
     /*
-     * If historyCursor is null, we've fetched all history.
+     * If oldestRev is null, we've fetched all history.
      */
-    if (this.historyCursor === null) return
+    if (this.oldestRev === null) return
 
     /*
      * Don't fetch again if a fetch is already in progress
@@ -519,7 +547,7 @@ export class Convo {
 
       const response = await this.agent.api.chat.bsky.convo.getMessages(
         {
-          cursor: this.historyCursor,
+          cursor: this.oldestRev,
           convoId: this.convoId,
           limit: isNative ? 25 : 50,
         },
@@ -531,21 +559,22 @@ export class Convo {
       )
       const {cursor, messages} = response.data
 
-      this.historyCursor = cursor ?? null
+      this.oldestRev = cursor ?? null
 
       for (const message of messages) {
         if (
           ChatBskyConvoDefs.isMessageView(message) ||
           ChatBskyConvoDefs.isDeletedMessageView(message)
         ) {
-          this.pastMessages.set(message.id, message)
+          /*
-
+           * If this message is already in new messages, it was added by the
-          // set to latest rev
+           * firehose ingestion, and we can safely overwrite it. This trusts
-          if (
+           * the server on ordering, and keeps it in sync.
-            message.rev > (this.eventsCursor = this.eventsCursor || message.rev)
+           */
-          ) {
+          if (this.newMessages.has(message.id)) {
-            this.eventsCursor = message.rev
+            this.newMessages.delete(message.id)
           }
+          this.pastMessages.set(message.id, message)
         }
       }
     } catch (e: any) {
@@ -594,14 +623,25 @@ export class Convo {
        * know what it is.
        */
       if (typeof ev.rev === 'string') {
+        const isUninitialized = !this.latestRev
+        const isNewEvent = this.latestRev && ev.rev > this.latestRev
+
+        /*
+         * We received an event prior to fetching any history, so we can safely
+         * use this as the initial history cursor
+         */
+        if (this.oldestRev === undefined && isUninitialized) {
+          this.oldestRev = ev.rev
+        }
+
         /*
          * We only care about new events
          */
-        if (ev.rev > (this.eventsCursor = this.eventsCursor || ev.rev)) {
+        if (isNewEvent || isUninitialized) {
           /*
            * Update rev regardless of if it's a ev type we care about or not
            */
-          this.eventsCursor = ev.rev
+          this.latestRev = ev.rev
 
           /*
            * This is VERY important. We don't want to insert any messages from
@@ -613,8 +653,14 @@ export class Convo {
             ChatBskyConvoDefs.isLogCreateMessage(ev) &&
             ChatBskyConvoDefs.isMessageView(ev.message)
           ) {
+            /**
+             * If this message is already in new messages, it was added by our
+             * sending logic, and is based on client-ordering. When we receive
+             * the "commited" event from the log, we should replace this
+             * reference and re-insert in order to respect the order we receied
+             * from the log.
+             */
             if (this.newMessages.has(ev.message.id)) {
-              // Trust the ev as the source of truth on ordering
               this.newMessages.delete(ev.message.id)
             }
             this.newMessages.set(ev.message.id, ev.message)
@@ -626,6 +672,7 @@ export class Convo {
             /*
              * Update if we have this in state. If we don't, don't worry about it.
              */
+            // TODO check for other storage spots
             if (this.pastMessages.has(ev.message.id)) {
               /*
                * For now, we remove deleted messages from the thread, if we receive one.

src/state/messages/convo/const.ts

@@ -1 +1,2 @@
 export const ACTIVE_POLL_INTERVAL = 1e3
+export const BACKGROUND_POLL_INTERVAL = 5e3

src/state/messages/convo/index.tsx

@@ -4,7 +4,6 @@ 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'
@@ -29,6 +28,7 @@ export function ConvoProvider({
   const isScreenFocused = useIsFocused()
   const {serviceUrl} = useDmServiceUrlStorage()
   const {getAgent} = useAgent()
+  const events = useMessagesEventBus()
   const [convo] = useState(
     () =>
       new Convo({
@@ -36,33 +36,15 @@ export function ConvoProvider({
         agent: new BskyAgent({
           service: serviceUrl,
         }),
+        events,
         __tempFromUserDid: getAgent().session?.did!,
       }),
   )
   const service = useSyncExternalStore(convo.subscribe, convo.getSnapshot)
   const {mutate: markAsRead} = useMarkAsReadMutation()
-  const events = useMessagesEventBus()
-
-  React.useEffect(() => {
-    const trail = events.trailConvo(convoId, events => {
-      convo.ingestFirehose(events)
-    })
-    const onConnect = events.onConnect(convo.onFirehoseConnect)
-    const onError = events.onError(convo.onFirehoseError)
-    return () => {
-      trail()
-      onConnect()
-      onError()
-    }
-  }, [convoId, convo, events])
 
   useFocusEffect(
     React.useCallback(() => {
-      if (!requestedPollInterval.current) {
-        requestedPollInterval.current =
-          events.requestPollInterval(ACTIVE_POLL_INTERVAL)
-      }
-
       convo.resume()
       markAsRead({convoId})
 
@@ -74,7 +56,7 @@ export function ConvoProvider({
         convo.background()
         markAsRead({convoId})
       }
-    }, [convo, convoId, markAsRead, events]),
+    }, [convo, convoId, markAsRead]),
   )
 
   React.useEffect(() => {
@@ -82,16 +64,8 @@ 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})
@@ -103,7 +77,7 @@ export function ConvoProvider({
     return () => {
       sub.remove()
     }
-  }, [convoId, convo, isScreenFocused, markAsRead, events])
+  }, [convoId, convo, isScreenFocused, markAsRead])
 
   return <ChatContext.Provider value={service}>{children}</ChatContext.Provider>
 }

src/state/messages/convo/types.ts

@@ -5,9 +5,12 @@ import {
   ChatBskyConvoSendMessage,
 } from '@atproto-labs/api'
 
+import {MessagesEventBus} from '#/state/messages/events/agent'
+
 export type ConvoParams = {
   convoId: string
   agent: BskyAgent
+  events: MessagesEventBus
   __tempFromUserDid: string
 }