Fix chat firehose event handling and error recovery (#10884)

src/state/messages/events/agent.ts

@@ -211,17 +211,9 @@ export class MessagesEventBus {
       }
       case MessagesEventBusStatus.Error: {
         switch (action.event) {
-          case MessagesEventBusDispatchEvent.UpdatePoll: {
+          case MessagesEventBusDispatchEvent.UpdatePoll:
-            // basically reset
-            this.status = MessagesEventBusStatus.Initializing
-            this.latestRev = undefined
-            this.init()
-            break
-          }
           case MessagesEventBusDispatchEvent.Resume: {
-            this.status = MessagesEventBusStatus.Ready
+            this.recoverFromError()
-            this.resetPoll()
-            this.emitter.emit('event', {type: 'connect'})
             break
           }
         }
@@ -238,6 +230,31 @@ export class MessagesEventBus {
     })
   }
 
+  private recoverFromError() {
+    logger.debug(`recoverFromError`, {hasRev: !!this.latestRev})
+
+    if (this.latestRev === undefined) {
+      /*
+       * init() never succeeded, so we have no cursor to resume from. Re-run
+       * init() to seed latestRev. Its success path dispatches Ready, which from
+       * Initializing transitions us to Ready + resetPoll + emit connect.
+       */
+      this.status = MessagesEventBusStatus.Initializing
+      this.init()
+    } else {
+      /*
+       * A poll failed mid-session but we still have a valid cursor. Resume
+       * polling from it directly. We must NOT route through init() here: its
+       * seeding logic takes the max of the existing rev and the server's
+       * current cursor, which would skip any events that arrived while we were
+       * offline.
+       */
+      this.status = MessagesEventBusStatus.Ready
+      this.resetPoll()
+      this.emitter.emit('event', {type: 'connect'})
+    }
+  }
+
   private async init() {
     logger.debug(`init`, {})
 
@@ -337,6 +354,9 @@ export class MessagesEventBus {
     //   },
     // )
 
+    let needsEmit = false
+    let batch: ChatBskyConvoGetLog.OutputSchema['logs'] = []
+
     try {
       const response = await networkRetry(2, () => {
         return this.agent.chat.bsky.convo.getLog(
@@ -351,9 +371,6 @@ export class MessagesEventBus {
 
       const {logs: events} = response.data
 
-      let needsEmit = false
-      let batch: ChatBskyConvoGetLog.OutputSchema['logs'] = []
-
       for (const ev of events) {
         /*
          * If there's a rev, we should handle it. If there's not a rev, we don't
@@ -373,10 +390,6 @@ export class MessagesEventBus {
           }
         }
       }
-
-      if (needsEmit) {
-        this.emitter.emit('event', {type: 'logs', logs: batch})
-      }
     } catch (e: any) {
       if (!isNetworkError(e) && !isErrorMaybeAppPasswordPermissions(e)) {
         logger.error(`poll events failed`, {
@@ -397,5 +410,22 @@ export class MessagesEventBus {
     } finally {
       this.isPolling = false
     }
+
+    /*
+     * Emit outside the try/catch above so a throwing subscriber is not
+     * misreported as a poll failure (which would show a network-error banner
+     * and drop the batch, since the revs have already been consumed). poll()
+     * runs from setInterval, so we must not let the exception escape - log it
+     * and move on without dispatching Error.
+     */
+    if (needsEmit) {
+      try {
+        this.emitter.emit('event', {type: 'logs', logs: batch})
+      } catch (e) {
+        logger.error(`subscriber error handling chat events`, {
+          safeMessage: e instanceof Error ? e.message : String(e),
+        })
+      }
+    }
   }
 }

src/state/queries/messages/list-conversations.tsx

@@ -282,8 +282,11 @@ export function ListConvosProviderInner({
           mutateMembers(convoId, list =>
             list.some(m => m.did === did) ? list : list.concat(newMember),
           )
-          mutateConvoView(convoId, convo =>
+          mutateConvoView(
-            addMemberToConvoView(convo, newMember, rev, alreadyKnownMember),
+            convoId,
+            withRevGuard(rev, convo =>
+              addMemberToConvoView(convo, newMember, rev, alreadyKnownMember),
+            ),
           )
         }
 
@@ -301,8 +304,11 @@ export function ListConvosProviderInner({
               >(listConvoMembersQueryKey(convoId))
               ?.some(m => m.did === did) === false
           mutateMembers(convoId, list => list.filter(m => m.did !== did))
-          mutateConvoView(convoId, convo =>
+          mutateConvoView(
-            removeMemberFromConvoView(convo, did, rev, alreadyRemovedMember),
+            convoId,
+            withRevGuard(rev, convo =>
+              removeMemberFromConvoView(convo, did, rev, alreadyRemovedMember),
+            ),
           )
         }
 
@@ -317,24 +323,27 @@ export function ListConvosProviderInner({
             // link preview so its viewer state reflects the lost membership.
             void invalidateJoinLinkPreviewsForConvo(queryClient, log.convoId)
           } else if (ChatBskyConvoDefs.isLogDeleteMessage(log)) {
-            updateConvoInAllLists(log.convoId, convo => {
+            updateConvoInAllLists(
-              if (
+              log.convoId,
-                (ChatBskyConvoDefs.isDeletedMessageView(log.message) ||
+              withRevGuard(log.rev, convo => {
-                  ChatBskyConvoDefs.isMessageView(log.message)) &&
+                if (
-                (ChatBskyConvoDefs.isDeletedMessageView(convo.lastMessage) ||
+                  (ChatBskyConvoDefs.isDeletedMessageView(log.message) ||
-                  ChatBskyConvoDefs.isMessageView(convo.lastMessage))
+                    ChatBskyConvoDefs.isMessageView(log.message)) &&
-              ) {
+                  (ChatBskyConvoDefs.isDeletedMessageView(convo.lastMessage) ||
-                return log.message.id === convo.lastMessage.id
+                    ChatBskyConvoDefs.isMessageView(convo.lastMessage))
-                  ? {
+                ) {
-                      ...convo,
+                  return log.message.id === convo.lastMessage.id
-                      rev: log.rev,
+                    ? {
-                      lastMessage: log.message,
+                        ...convo,
-                    }
+                        rev: log.rev,
-                  : convo
+                        lastMessage: log.message,
-              } else {
+                      }
-                return convo
+                    : convo
-              }
+                } else {
-            })
+                  return convo
+                }
+              }),
+            )
           } else if (ChatBskyConvoDefs.isLogCreateMessage(log)) {
             // Store in a new var to avoid TS errors due to closures.
             const logRef: ChatBskyConvoDefs.LogCreateMessage = log
@@ -356,9 +365,20 @@ export function ListConvosProviderInner({
             }
 
             if (!foundConvo) {
-              // Convo not found, trigger refetch
+              // Convo not found, trigger refetch. Use continue (not return) so
+              // the remaining logs in this batch still apply - the bus advances
+              // its cursor past this batch, so a dropped log is never
+              // redelivered.
               debouncedRefetch()
-              return
+              continue
+            }
+
+            // Rev guard. updatedConvo is built once from foundConvo and applied
+            // across caches, so guarding here (rather than per-cache) is both
+            // simplest and correct - skip if the log isn't newer than the
+            // cached convo.
+            if (logRef.rev <= foundConvo.rev) {
+              continue
             }
 
             // add relatedProfiles to members list, but making sure to dedupe
@@ -449,17 +469,23 @@ export function ListConvosProviderInner({
               )
             }
           } else if (ChatBskyConvoDefs.isLogReadMessage(log)) {
-            updateConvoInAllLists(log.convoId, convo => ({
+            updateConvoInAllLists(
-              ...convo,
+              log.convoId,
-              unreadCount: 0,
+              withRevGuard(log.rev, convo => ({
-              rev: log.rev,
+                ...convo,
-            }))
+                unreadCount: 0,
+                rev: log.rev,
+              })),
+            )
           } else if (ChatBskyConvoDefs.isLogReadConvo(log)) {
-            updateConvoInAllLists(log.convoId, convo => ({
+            updateConvoInAllLists(
-              ...convo,
+              log.convoId,
-              unreadCount: 0,
+              withRevGuard(log.rev, convo => ({
-              rev: log.rev,
+                ...convo,
-            }))
+                unreadCount: 0,
+                rev: log.rev,
+              })),
+            )
           } else if (ChatBskyConvoDefs.isLogAcceptConvo(log)) {
             const requestQueries =
               queryClient.getQueriesData<ConvoListQueryData>({
@@ -472,13 +498,37 @@ export function ListConvosProviderInner({
               if (foundConvo) break
             }
             if (!foundConvo) {
+              // Use continue (not return) so the remaining logs in this batch
+              // still apply - the bus advances its cursor past this batch, so a
+              // dropped log is never redelivered.
               debouncedRefetch()
-              return
+              continue
+            }
+            if (log.rev <= foundConvo.rev) {
+              continue
             }
             const acceptedConvo: ChatBskyConvoDefs.ConvoView = {
               ...foundConvo,
               status: 'accepted',
+              rev: log.rev,
             }
+            // Flip status to 'accepted' in every cache that already holds this
+            // convo - including 'all'-status caches like the provider's
+            // always-mounted unread query, which the request->accepted move
+            // below otherwise never touches. Without this the stale 'all' copy
+            // keeps status: 'request', and the next isLogCreateMessage can seed
+            // foundConvo from it and resurrect the convo in the requests inbox.
+            // Runs before the delete-from-request below: it updates in place
+            // (never inserts), so the 'request' caches get the accepted copy and
+            // are then cleared by the delete, leaving no stale request entries.
+            updateConvoInAllLists(
+              log.convoId,
+              withRevGuard(log.rev, convo => ({
+                ...convo,
+                status: 'accepted',
+                rev: log.rev,
+              })),
+            )
             queryClient.setQueriesData(
               {queryKey: RQKEY_PARTIAL('request')},
               (old?: ConvoListQueryData) => optimisticDelete(log.convoId, old),
@@ -519,60 +569,75 @@ export function ListConvosProviderInner({
               },
             )
           } else if (ChatBskyConvoDefs.isLogMuteConvo(log)) {
-            mutateConvoView(log.convoId, convo => ({
+            mutateConvoView(
-              ...convo,
+              log.convoId,
-              muted: true,
+              withRevGuard(log.rev, convo => ({
-              rev: log.rev,
+                ...convo,
-            }))
+                muted: true,
+                rev: log.rev,
+              })),
+            )
           } else if (ChatBskyConvoDefs.isLogUnmuteConvo(log)) {
-            mutateConvoView(log.convoId, convo => ({
+            mutateConvoView(
-              ...convo,
+              log.convoId,
-              muted: false,
+              withRevGuard(log.rev, convo => ({
-              rev: log.rev,
+                ...convo,
-            }))
+                muted: false,
+                rev: log.rev,
+              })),
+            )
           } else if (ChatBskyConvoDefs.isLogLockConvo(log)) {
-            mutateConvoView(log.convoId, convo => {
+            mutateConvoView(
-              if (ChatBskyConvoDefs.isGroupConvo(convo.kind)) {
+              log.convoId,
-                return {
+              withRevGuard(log.rev, convo => {
-                  ...convo,
+                if (ChatBskyConvoDefs.isGroupConvo(convo.kind)) {
-                  kind: {...convo.kind, lockStatus: 'locked'},
+                  return {
-                  rev: log.rev,
+                    ...convo,
+                    kind: {...convo.kind, lockStatus: 'locked'},
+                    rev: log.rev,
+                  }
                 }
-              }
+                return {...convo, rev: log.rev}
-              return {...convo, rev: log.rev}
+              }),
-            })
+            )
             // The log event doesn't say whether the lock is forced by a
             // moderation override, so refetch to pick up the flag.
             void queryClient.invalidateQueries({
               queryKey: CONVO_KEY(log.convoId),
             })
           } else if (ChatBskyConvoDefs.isLogUnlockConvo(log)) {
-            mutateConvoView(log.convoId, convo => {
+            mutateConvoView(
-              if (ChatBskyConvoDefs.isGroupConvo(convo.kind)) {
+              log.convoId,
-                return {
+              withRevGuard(log.rev, convo => {
-                  ...convo,
+                if (ChatBskyConvoDefs.isGroupConvo(convo.kind)) {
-                  kind: {
+                  return {
-                    ...convo.kind,
+                    ...convo,
-                    lockStatus: 'unlocked',
+                    kind: {
-                    // An unlocked convo cannot be moderation-locked.
+                      ...convo.kind,
-                    lockStatusModerationOverride: false,
+                      lockStatus: 'unlocked',
-                  },
+                      // An unlocked convo cannot be moderation-locked.
-                  rev: log.rev,
+                      lockStatusModerationOverride: false,
+                    },
+                    rev: log.rev,
+                  }
                 }
-              }
+                return {...convo, rev: log.rev}
-              return {...convo, rev: log.rev}
+              }),
-            })
+            )
           } else if (ChatBskyConvoDefs.isLogLockConvoPermanently(log)) {
-            mutateConvoView(log.convoId, convo => {
+            mutateConvoView(
-              if (ChatBskyConvoDefs.isGroupConvo(convo.kind)) {
+              log.convoId,
-                return {
+              withRevGuard(log.rev, convo => {
-                  ...convo,
+                if (ChatBskyConvoDefs.isGroupConvo(convo.kind)) {
-                  kind: {...convo.kind, lockStatus: 'locked-permanently'},
+                  return {
-                  rev: log.rev,
+                    ...convo,
+                    kind: {...convo.kind, lockStatus: 'locked-permanently'},
+                    rev: log.rev,
+                  }
                 }
-              }
+                return {...convo, rev: log.rev}
-              return {...convo, rev: log.rev}
+              }),
-            })
+            )
           } else if (
             ChatBskyConvoDefs.isLogCreateJoinLink(log) ||
             ChatBskyConvoDefs.isLogEditJoinLink(log) ||
@@ -592,30 +657,39 @@ export function ListConvosProviderInner({
             // Route through mutateConvoView (not updateConvoInAllLists) so the
             // single-convo cache updates too, keeping the in-convo requests
             // banner in sync.
-            mutateConvoView(log.convoId, convo =>
+            mutateConvoView(
-              applyJoinRequestCountDelta(convo, log.rev, -1),
+              log.convoId,
+              withRevGuard(log.rev, convo =>
+                applyJoinRequestCountDelta(convo, log.rev, -1),
+              ),
             )
           } else if (ChatBskyConvoDefs.isLogIncomingJoinRequest(log)) {
             // Route through mutateConvoView (not updateConvoInAllLists) so the
             // single-convo cache updates too, letting the in-convo requests
             // banner appear live.
-            mutateConvoView(log.convoId, convo =>
+            mutateConvoView(
-              applyJoinRequestCountDelta(convo, log.rev, 1),
+              log.convoId,
+              withRevGuard(log.rev, convo =>
+                applyJoinRequestCountDelta(convo, log.rev, 1),
+              ),
             )
           } else if (ChatBskyConvoDefs.isLogReadJoinRequests(log)) {
             // The owner marked join requests as read (possibly on another
             // device). Zero the unread count but keep the total, mirroring the
             // useMarkJoinRequestsRead mutation.
-            mutateConvoView(log.convoId, convo => {
+            mutateConvoView(
-              if (!ChatBskyConvoDefs.isGroupConvo(convo.kind)) {
+              log.convoId,
-                return {...convo, rev: log.rev}
+              withRevGuard(log.rev, convo => {
-              }
+                if (!ChatBskyConvoDefs.isGroupConvo(convo.kind)) {
-              return {
+                  return {...convo, rev: log.rev}
-                ...convo,
+                }
-                kind: {...convo.kind, unreadJoinRequestCount: 0},
+                return {
-                rev: log.rev,
+                  ...convo,
-              }
+                  kind: {...convo.kind, unreadJoinRequestCount: 0},
-            })
+                  rev: log.rev,
+                }
+              }),
+            )
           } else if (ChatBskyConvoDefs.isLogOutgoingJoinRequest(log)) {
             // Viewer isn't in the chat yet, but the inbox surfaces outgoing
             // requests, so refetch to pick up the new entry.
@@ -623,8 +697,11 @@ export function ListConvosProviderInner({
           } else if (ChatBskyConvoDefs.isLogWithdrawIncomingJoinRequest(log)) {
             // A requester rescinded their request to a group the viewer owns.
             // Mirror of isLogIncomingJoinRequest: decrement the counts.
-            mutateConvoView(log.convoId, convo =>
+            mutateConvoView(
-              applyJoinRequestCountDelta(convo, log.rev, -1),
+              log.convoId,
+              withRevGuard(log.rev, convo =>
+                applyJoinRequestCountDelta(convo, log.rev, -1),
+              ),
             )
           } else if (ChatBskyConvoDefs.isLogWithdrawOutgoingJoinRequest(log)) {
             // The viewer rescinded their own outgoing join request (possibly on
@@ -634,25 +711,28 @@ export function ListConvosProviderInner({
               old => optimisticDeleteJoinRequest(log.convoId, old),
             )
           } else if (ChatBskyConvoDefs.isLogAddReaction(log)) {
-            updateConvoInAllLists(log.convoId, convo => {
+            updateConvoInAllLists(
-              // add relatedProfiles to members list, but making sure to dedupe
+              log.convoId,
-              const relatedProfilesSansMembers = (
+              withRevGuard(log.rev, convo => {
-                log.relatedProfiles ?? []
+                // add relatedProfiles to members list, but making sure to dedupe
-              ).filter(
+                const relatedProfilesSansMembers = (
-                profile =>
+                  log.relatedProfiles ?? []
-                  !convo.members.some(member => member.did === profile.did),
+                ).filter(
-              )
+                  profile =>
-              return {
+                    !convo.members.some(member => member.did === profile.did),
-                ...convo,
+                )
-                members: [...convo.members, ...relatedProfilesSansMembers],
+                return {
-                lastReaction: {
+                  ...convo,
-                  $type: 'chat.bsky.convo.defs#messageAndReactionView',
+                  members: [...convo.members, ...relatedProfilesSansMembers],
-                  reaction: log.reaction,
+                  lastReaction: {
-                  message: log.message,
+                    $type: 'chat.bsky.convo.defs#messageAndReactionView',
-                },
+                    reaction: log.reaction,
-                rev: log.rev,
+                    message: log.message,
-              }
+                  },
-            })
+                  rev: log.rev,
+                }
+              }),
+            )
           } else if (ChatBskyConvoDefs.isLogAddMember(log)) {
             const data = log.message.data
             if (
@@ -725,31 +805,35 @@ export function ListConvosProviderInner({
             queryClient.setQueriesData(
               {queryKey: [RQKEY_ROOT]},
               (old?: ConvoListQueryData) =>
-                optimisticUpdate(log.convoId, old, convo => {
+                optimisticUpdate(
-                  if (
+                  log.convoId,
-                    // if the convo is the same
+                  old,
-                    log.convoId === convo.id &&
+                  withRevGuard(log.rev, convo => {
-                    ChatBskyConvoDefs.isMessageAndReactionView(
+                    if (
-                      convo.lastReaction,
+                      // if the convo is the same
-                    ) &&
+                      log.convoId === convo.id &&
-                    ChatBskyConvoDefs.isMessageView(log.message) &&
+                      ChatBskyConvoDefs.isMessageAndReactionView(
-                    // ...and the message is the same
+                        convo.lastReaction,
-                    convo.lastReaction.message.id === log.message.id &&
+                      ) &&
-                    // ...and the reaction is the same
+                      ChatBskyConvoDefs.isMessageView(log.message) &&
-                    convo.lastReaction.reaction.sender.did ===
+                      // ...and the message is the same
-                      log.reaction.sender.did &&
+                      convo.lastReaction.message.id === log.message.id &&
-                    convo.lastReaction.reaction.value === log.reaction.value
+                      // ...and the reaction is the same
-                  ) {
+                      convo.lastReaction.reaction.sender.did ===
-                    return {
+                        log.reaction.sender.did &&
-                      ...convo,
+                      convo.lastReaction.reaction.value === log.reaction.value
-                      // ...remove the reaction. hopefully they didn't react twice in a row!
+                    ) {
-                      lastReaction: undefined,
+                      return {
-                      rev: log.rev,
+                        ...convo,
+                        // ...remove the reaction. hopefully they didn't react twice in a row!
+                        lastReaction: undefined,
+                        rev: log.rev,
+                      }
+                    } else {
+                      return convo
                     }
-                  } else {
+                  }),
-                    return convo
+                ),
-                  }
-                }),
             )
           }
         }
@@ -905,6 +989,25 @@ export function useOnMarkAsRead() {
   )
 }
 
+/**
+ * Wraps a log-driven convo update so it's skipped when the log is not newer
+ * than the cached convo. Lists are fed by two unsynchronized channels (full
+ * listConvos refetches and the log stream), so a stale refetch snapshot can be
+ * written after a log already applied, or a refetch can already include a
+ * message whose log then arrives and double-counts. Rev comparison as plain
+ * string comparison is safe here - revs are fixed-width. ConvoView.rev is a
+ * required field per the lexicon, so convo.rev always exists.
+ *
+ * Only used in the log-event paths (which have a log.rev), never in the generic
+ * optimisticUpdate helper, which mutations without a rev also call.
+ */
+function withRevGuard(
+  rev: string,
+  fn: (convo: ChatBskyConvoDefs.ConvoView) => ChatBskyConvoDefs.ConvoView,
+): (convo: ChatBskyConvoDefs.ConvoView) => ChatBskyConvoDefs.ConvoView {
+  return convo => (rev <= convo.rev ? convo : fn(convo))
+}
+
 function optimisticUpdate(
   chatId: string,
   old?: ConvoListQueryData,