[🐴] Error recovery (#4036)

* Handle block state when sending messages

* Handle different pending failures

* Use existing profile data to handle blocks

* Better cleanup, leave room for more

* Attempt recover upon next send

* Reset pending failure

* Capture unexpected error

* Gracefully handle network errors and recovery

* Re-align error components and types

* Include history fetching in recoverable states

1 files changed, 128 insertions, 52 deletions

diff --git a/src/state/messages/convo/agent.ts b/src/state/messages/convo/agent.ts
index 94bb8dda4..8673c70ad 100644
--- a/src/state/messages/convo/agent.ts
+++ b/src/state/messages/convo/agent.ts
@@ -5,6 +5,8 @@ import {
   ChatBskyConvoGetLog,
   ChatBskyConvoSendMessage,
 } from '@atproto/api'
+import {XRPCError} from '@atproto/xrpc'
+import EventEmitter from 'eventemitter3'
 import {nanoid} from 'nanoid/non-secure'
 
 import {networkRetry} from '#/lib/async/retry'
@@ -14,11 +16,14 @@ import {
   ACTIVE_POLL_INTERVAL,
   BACKGROUND_POLL_INTERVAL,
   INACTIVE_TIMEOUT,
+  NETWORK_FAILURE_STATUSES,
 } from '#/state/messages/convo/const'
 import {
   ConvoDispatch,
   ConvoDispatchEvent,
+  ConvoError,
   ConvoErrorCode,
+  ConvoEvent,
   ConvoItem,
   ConvoItemError,
   ConvoParams,
@@ -51,13 +56,7 @@ export class Convo {
   private senderUserDid: string
 
   private status: ConvoStatus = ConvoStatus.Uninitialized
-  private error:
-    | {
-        code: ConvoErrorCode
-        exception?: Error
-        retry: () => void
-      }
-    | undefined
+  private error: ConvoError | undefined
   private oldestRev: string | undefined | null = undefined
   private isFetchingHistory = false
   private latestRev: string | undefined = undefined
@@ -75,13 +74,13 @@ export class Convo {
     {id: string; message: ChatBskyConvoSendMessage.InputSchema['message']}
   > = new Map()
   private deletedMessages: Set<string> = new Set()
-  private footerItems: Map<string, ConvoItem> = new Map()
-  private headerItems: Map<string, ConvoItem> = new Map()
 
   private isProcessingPendingMessages = false
 
   private lastActiveTimestamp: number | undefined
 
+  private emitter = new EventEmitter<{event: [ConvoEvent]}>()
+
   convoId: string
   convo: ChatBskyConvoDefs.ConvoView | undefined
   sender: AppBskyActorDefs.ProfileViewBasic | undefined
@@ -174,7 +173,7 @@ export class Convo {
           status: ConvoStatus.Error,
           items: [],
           convo: undefined,
-          error: this.error,
+          error: this.error!,
           sender: undefined,
           recipients: undefined,
           isFetchingHistory: false,
@@ -282,6 +281,7 @@ export class Convo {
               if (this.convo) {
                 this.status = ConvoStatus.Ready
                 this.refreshConvo()
+                this.maybeRecoverFromNetworkError()
               } else {
                 this.status = ConvoStatus.Initializing
                 this.setup()
@@ -379,12 +379,30 @@ export class Convo {
     this.newMessages = new Map()
     this.pendingMessages = new Map()
     this.deletedMessages = new Set()
-    this.footerItems = new Map()
-    this.headerItems = new Map()
+
+    this.pendingMessageFailure = null
+    this.fetchMessageHistoryError = undefined
+    this.firehoseError = undefined
 
     this.dispatch({event: ConvoDispatchEvent.Init})
   }
 
+  maybeRecoverFromNetworkError() {
+    if (this.firehoseError) {
+      this.firehoseError.retry()
+      this.firehoseError = undefined
+      this.commit()
+    } else {
+      this.batchRetryPendingMessages()
+    }
+
+    if (this.fetchMessageHistoryError) {
+      this.fetchMessageHistoryError.retry()
+      this.fetchMessageHistoryError = undefined
+      this.commit()
+    }
+  }
+
   private async setup() {
     try {
       const {convo, sender, recipients} = await this.fetchConvo()
@@ -520,6 +538,11 @@ export class Convo {
     }
   }
 
+  private fetchMessageHistoryError:
+    | {
+        retry: () => void
+      }
+    | undefined
   async fetchMessageHistory() {
     logger.debug('Convo: fetch message history', {}, logger.DebugContext.convo)
 
@@ -537,7 +560,7 @@ export class Convo {
      * If we've rendered a retry state for history fetching, exit. Upon retry,
      * this will be removed and we'll try again.
      */
-    if (this.headerItems.has(ConvoItemError.HistoryFailed)) return
+    if (this.fetchMessageHistoryError) return
 
     try {
       this.isFetchingHistory = true
@@ -586,15 +609,11 @@ export class Convo {
     } catch (e: any) {
       logger.error('Convo: failed to fetch message history')
 
-      this.headerItems.set(ConvoItemError.HistoryFailed, {
-        type: 'error-recoverable',
-        key: ConvoItemError.HistoryFailed,
-        code: ConvoItemError.HistoryFailed,
+      this.fetchMessageHistoryError = {
         retry: () => {
-          this.headerItems.delete(ConvoItemError.HistoryFailed)
           this.fetchMessageHistory()
         },
-      })
+      }
     } finally {
       this.isFetchingHistory = false
       this.commit()
@@ -628,22 +647,16 @@ export class Convo {
     )
   }
 
+  private firehoseError: MessagesEventBusError | undefined
+
   onFirehoseConnect() {
-    this.footerItems.delete(ConvoItemError.FirehoseFailed)
+    this.firehoseError = undefined
+    this.batchRetryPendingMessages()
     this.commit()
   }
 
   onFirehoseError(error?: MessagesEventBusError) {
-    this.footerItems.set(ConvoItemError.FirehoseFailed, {
-      type: 'error-recoverable',
-      key: ConvoItemError.FirehoseFailed,
-      code: ConvoItemError.FirehoseFailed,
-      retry: () => {
-        this.footerItems.delete(ConvoItemError.FirehoseFailed)
-        this.commit()
-        error?.retry()
-      },
-    })
+    this.firehoseError = error
     this.commit()
   }
 
@@ -724,7 +737,7 @@ export class Convo {
     }
   }
 
-  private pendingFailed = false
+  private pendingMessageFailure: 'recoverable' | 'unrecoverable' | null = null
 
   async sendMessage(message: ChatBskyConvoSendMessage.InputSchema['message']) {
     // Ignore empty messages for now since they have no other purpose atm
@@ -734,13 +747,14 @@ export class Convo {
 
     const tempId = nanoid()
 
+    this.pendingMessageFailure = null
     this.pendingMessages.set(tempId, {
       id: tempId,
       message,
     })
     this.commit()
 
-    if (!this.isProcessingPendingMessages && !this.pendingFailed) {
+    if (!this.isProcessingPendingMessages && !this.pendingMessageFailure) {
       this.processPendingMessages()
     }
   }
@@ -765,7 +779,6 @@ export class Convo {
     try {
       this.isProcessingPendingMessages = true
 
-      // throw new Error('UNCOMMENT TO TEST RETRY')
       const {id, message} = pendingMessage
 
       const response = await networkRetry(2, () => {
@@ -794,23 +807,65 @@ export class Convo {
       this.commit()
     } catch (e: any) {
       logger.error(e, {context: `Convo: failed to send message`})
-      this.pendingFailed = true
-      this.commit()
+      this.handleSendMessageFailure(e)
     } finally {
       this.isProcessingPendingMessages = false
     }
   }
 
+  private handleSendMessageFailure(e: any) {
+    if (e instanceof XRPCError) {
+      if (NETWORK_FAILURE_STATUSES.includes(e.status)) {
+        this.pendingMessageFailure = 'recoverable'
+      } else {
+        switch (e.message) {
+          case 'block between recipient and sender':
+            this.pendingMessageFailure = 'unrecoverable'
+            this.emitter.emit('event', {
+              type: 'invalidate-block-state',
+              accountDids: [
+                this.sender!.did,
+                ...this.recipients!.map(r => r.did),
+              ],
+            })
+            break
+          default:
+            logger.warn(
+              `Convo handleSendMessageFailure could not handle error`,
+              {
+                status: e.status,
+                message: e.message,
+              },
+            )
+            break
+        }
+      }
+    } else {
+      logger.error(e, {
+        context: `Convo handleSendMessageFailure received unknown error`,
+      })
+    }
+
+    this.commit()
+  }
+
   async batchRetryPendingMessages() {
+    if (this.pendingMessageFailure === null) return
+
+    const messageArray = Array.from(this.pendingMessages.values())
+    if (messageArray.length === 0) return
+
+    this.pendingMessageFailure = null
+    this.commit()
+
     logger.debug(
-      `Convo: retrying ${this.pendingMessages.size} pending messages`,
+      `Convo: batch retrying ${this.pendingMessages.size} pending messages`,
       {},
       logger.DebugContext.convo,
     )
 
     try {
       // throw new Error('UNCOMMENT TO TEST RETRY')
-      const messageArray = Array.from(this.pendingMessages.values())
       const {data} = await networkRetry(2, () => {
         return this.agent.api.chat.bsky.convo.sendMessageBatch(
           {
@@ -848,8 +903,7 @@ export class Convo {
       )
     } catch (e: any) {
       logger.error(e, {context: `Convo: failed to batch retry messages`})
-      this.pendingFailed = true
-      this.commit()
+      this.handleSendMessageFailure(e)
     }
   }
 
@@ -877,6 +931,14 @@ export class Convo {
     }
   }
 
+  on(handler: (event: ConvoEvent) => void) {
+    this.emitter.on('event', handler)
+
+    return () => {
+      this.emitter.off('event', handler)
+    }
+  }
+
   /*
    * Items in reverse order, since FlatList inverts
    */
@@ -901,9 +963,16 @@ export class Convo {
       }
     })
 
-    this.headerItems.forEach(item => {
-      items.unshift(item)
-    })
+    if (this.fetchMessageHistoryError) {
+      items.unshift({
+        type: 'error',
+        code: ConvoItemError.HistoryFailed,
+        key: ConvoItemError.HistoryFailed,
+        retry: () => {
+          this.maybeRecoverFromNetworkError()
+        },
+      })
+    }
 
     this.newMessages.forEach(m => {
       if (ChatBskyConvoDefs.isMessageView(m)) {
@@ -940,19 +1009,26 @@ export class Convo {
           sender: this.sender!,
         },
         nextMessage: null,
-        retry: this.pendingFailed
-          ? () => {
-              this.pendingFailed = false
-              this.commit()
-              this.batchRetryPendingMessages()
-            }
-          : undefined,
+        failed: this.pendingMessageFailure !== null,
+        retry:
+          this.pendingMessageFailure === 'recoverable'
+            ? () => {
+                this.maybeRecoverFromNetworkError()
+              }
+            : undefined,
       })
     })
 
-    this.footerItems.forEach(item => {
-      items.push(item)
-    })
+    if (this.firehoseError) {
+      items.push({
+        type: 'error',
+        code: ConvoItemError.FirehoseFailed,
+        key: ConvoItemError.FirehoseFailed,
+        retry: () => {
+          this.firehoseError?.retry()
+        },
+      })
+    }
 
     return items
       .filter(item => {