Clean up polling, add backgrounding

src/state/messages/events/agent.ts

@@ -15,7 +15,8 @@ import {
 
 const LOGGER_CONTEXT = 'MessagesEventBus'
 
-const ACTIVE_POLL_INTERVAL = 3e3
+const ACTIVE_POLL_INTERVAL = 5e3
+const BACKGROUND_POLL_INTERVAL = 20e3
 
 export class MessagesEventBus {
   private id: string
@@ -29,8 +30,6 @@ export class MessagesEventBus {
   private error: MessagesEventBusError | undefined
   private latestRev: string | undefined = undefined
 
-  private nextPoll: NodeJS.Timeout | undefined
-
   snapshot: MessagesEventBusState | undefined
 
   constructor(params: MessagesEventBusParams) {
@@ -144,8 +143,16 @@ export class MessagesEventBus {
         switch (action.event) {
           case MessagesEventBusDispatchEvent.Ready: {
             this.status = MessagesEventBusStatus.Ready
-            this.pollInterval = ACTIVE_POLL_INTERVAL
+            this.setPollInterval(ACTIVE_POLL_INTERVAL)
-            this.restartPoll()
+            break
+          }
+          case MessagesEventBusDispatchEvent.Background: {
+            this.status = MessagesEventBusStatus.Backgrounded
+            this.setPollInterval(BACKGROUND_POLL_INTERVAL)
+            break
+          }
+          case MessagesEventBusDispatchEvent.Suspend: {
+            this.status = MessagesEventBusStatus.Suspended
             break
           }
           case MessagesEventBusDispatchEvent.Error: {
@@ -158,15 +165,41 @@ export class MessagesEventBus {
       }
       case MessagesEventBusStatus.Ready: {
         switch (action.event) {
+          case MessagesEventBusDispatchEvent.Background: {
+            this.status = MessagesEventBusStatus.Backgrounded
+            this.setPollInterval(BACKGROUND_POLL_INTERVAL)
+            break
+          }
           case MessagesEventBusDispatchEvent.Suspend: {
             this.status = MessagesEventBusStatus.Suspended
-            this.cancelNextPoll()
+            this.stopPoll()
             break
           }
           case MessagesEventBusDispatchEvent.Error: {
             this.status = MessagesEventBusStatus.Error
             this.error = action.payload
-            this.cancelNextPoll()
+            this.stopPoll()
+            break
+          }
+        }
+        break
+      }
+      case MessagesEventBusStatus.Backgrounded: {
+        switch (action.event) {
+          case MessagesEventBusDispatchEvent.Resume: {
+            this.status = MessagesEventBusStatus.Ready
+            this.setPollInterval(ACTIVE_POLL_INTERVAL)
+            break
+          }
+          case MessagesEventBusDispatchEvent.Suspend: {
+            this.status = MessagesEventBusStatus.Suspended
+            this.stopPoll()
+            break
+          }
+          case MessagesEventBusDispatchEvent.Error: {
+            this.status = MessagesEventBusStatus.Error
+            this.error = action.payload
+            this.stopPoll()
             break
           }
         }
@@ -176,14 +209,18 @@ export class MessagesEventBus {
         switch (action.event) {
           case MessagesEventBusDispatchEvent.Resume: {
             this.status = MessagesEventBusStatus.Ready
-            this.pollInterval = ACTIVE_POLL_INTERVAL
+            this.setPollInterval(ACTIVE_POLL_INTERVAL)
-            this.restartPoll()
+            break
+          }
+          case MessagesEventBusDispatchEvent.Background: {
+            this.status = MessagesEventBusStatus.Backgrounded
+            this.setPollInterval(BACKGROUND_POLL_INTERVAL)
             break
           }
           case MessagesEventBusDispatchEvent.Error: {
             this.status = MessagesEventBusStatus.Error
             this.error = action.payload
-            this.cancelNextPoll()
+            this.stopPoll()
             break
           }
         }
@@ -191,6 +228,7 @@ export class MessagesEventBus {
       }
       case MessagesEventBusStatus.Error: {
         switch (action.event) {
+          case MessagesEventBusDispatchEvent.Resume:
           case MessagesEventBusDispatchEvent.Init: {
             this.status = MessagesEventBusStatus.Initializing
             this.error = undefined
@@ -223,12 +261,11 @@ export class MessagesEventBus {
 
     try {
       await this.initializeLatestRev()
-
-      // await new Promise(y => setTimeout(y, 2000))
-      // throw new Error('UNCOMMENT TO TEST INIT FAILURE')
       this.dispatch({event: MessagesEventBusDispatchEvent.Ready})
     } catch (e: any) {
-      logger.error(`${LOGGER_CONTEXT}: setup failed`)
+      logger.error(e, {
+        context: `${LOGGER_CONTEXT}: setup failed`,
+      })
 
       this.dispatch({
         event: MessagesEventBusDispatchEvent.Error,
@@ -248,6 +285,11 @@ export class MessagesEventBus {
     this.dispatch({event: MessagesEventBusDispatchEvent.Init})
   }
 
+  background() {
+    logger.debug(`${LOGGER_CONTEXT}: background`, {}, logger.DebugContext.convo)
+    this.dispatch({event: MessagesEventBusDispatchEvent.Background})
+  }
+
   suspend() {
     logger.debug(`${LOGGER_CONTEXT}: suspend`, {}, logger.DebugContext.convo)
     this.dispatch({event: MessagesEventBusDispatchEvent.Suspend})
@@ -260,7 +302,7 @@ export class MessagesEventBus {
 
   setPollInterval(interval: number) {
     this.pollInterval = interval
-    this.restartPoll()
+    this.resetPoll()
   }
 
   trail(handler: (events: ChatBskyConvoGetLog.OutputSchema['logs']) => void) {
@@ -277,8 +319,6 @@ export class MessagesEventBus {
       logger.DebugContext.convo,
     )
 
-    // throw new Error('UNCOMMENT TO TEST INIT FAILURE')
-
     const response = await this.agent.api.chat.bsky.convo.listConvos(
       {
         limit: 1,
@@ -293,131 +333,105 @@ export class MessagesEventBus {
     const {convos} = response.data
 
     for (const convo of convos) {
-      // set to latest rev
       if (convo.rev > (this.latestRev = this.latestRev || convo.rev)) {
         this.latestRev = convo.rev
       }
     }
   }
 
-  private restartPoll() {
+  /*
-    logger.debug(
+   * Polling
-      `${LOGGER_CONTEXT}: restart poll`,
+   */
-      {},
+
-      logger.DebugContext.convo,
+  private isPolling = false
-    )
+  private pollIntervalRef: NodeJS.Timeout | undefined
-    this.cancelNextPoll()
+
-    this.pollLatestEvents()
+  private resetPoll() {
+    this.stopPoll()
+    this.startPoll()
   }
 
-  private cancelNextPoll() {
+  private startPoll() {
-    logger.debug(
+    if (!this.isPolling) this.poll()
-      `${LOGGER_CONTEXT}: cancel next poll`,
+
-      {},
+    this.pollIntervalRef = setInterval(() => {
-      logger.DebugContext.convo,
+      if (this.isPolling) return
-    )
+      this.poll()
-    if (this.nextPoll) clearTimeout(this.nextPoll)
+    }, this.pollInterval)
   }
 
-  private pollLatestEvents() {
+  private stopPoll() {
-    /*
+    if (this.pollIntervalRef) clearInterval(this.pollIntervalRef)
-     * Uncomment to view poll events
+  }
-     */
+
+  private async poll() {
+    if (this.isPolling) return
+
+    this.isPolling = true
+
     logger.debug(`${LOGGER_CONTEXT}: poll`, {}, logger.DebugContext.convo)
 
-    this.nextPoll = setTimeout(() => {
+    try {
-      this.pollLatestEvents()
+      const response = await this.agent.api.chat.bsky.convo.getLog(
-    }, this.pollInterval)
+        {
-
+          cursor: this.latestRev,
-    this.fetchLatestEvents()
+        },
-      .then(({events}) => {
+        {
-        this.processLatestEvents(events)
+          headers: {
-      })
+            Authorization: this.__tempFromUserDid,
-      .catch(e => {
-        logger.error(`${LOGGER_CONTEXT}: poll events failed`)
-
-        this.dispatch({
-          event: MessagesEventBusDispatchEvent.Error,
-          payload: {
-            exception: e,
-            code: MessagesEventBusErrorCode.PollFailed,
-            retry: () => {
-              this.init()
-            },
           },
-        })
+        },
-      })
+      )
-  }
 
-  private pendingFetchLatestEvents:
+      const {logs: events} = response.data
-    | Promise<{
-        events: ChatBskyConvoGetLog.OutputSchema['logs']
-      }>
-    | undefined
-  async fetchLatestEvents() {
-    if (this.pendingFetchLatestEvents) return this.pendingFetchLatestEvents
 
-    this.pendingFetchLatestEvents = new Promise<{
+      let needsEmit = false
-      events: ChatBskyConvoGetLog.OutputSchema['logs']
+      let batch: 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.latestRev,
-          },
-          {
-            headers: {
-              Authorization: this.__tempFromUserDid,
-            },
-          },
-        )
-        const {logs} = response.data
-        resolve({events: logs})
-      } catch (e) {
-        reject(e)
-      } finally {
-        this.pendingFetchLatestEvents = undefined
-      }
-    })
 
-    return this.pendingFetchLatestEvents
+      for (const ev of events) {
-  }
-
-  private processLatestEvents(
-    events: ChatBskyConvoGetLog.OutputSchema['logs'],
-  ) {
-    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
-       * know what it is.
-       */
-      if (typeof ev.rev === 'string') {
         /*
-         * We only care about new events
+         * If there's a rev, we should handle it. If there's not a rev, we don't
+         * know what it is.
          */
-        if (ev.rev > (this.latestRev = this.latestRev || ev.rev)) {
+        if (typeof ev.rev === 'string') {
           /*
-           * Update rev regardless of if it's a ev type we care about or not
+           * We only care about new events
            */
-          this.latestRev = ev.rev
+          if (ev.rev > (this.latestRev = this.latestRev || ev.rev)) {
-          needsEmit = true
+            /*
-          batch.push(ev)
+             * Update rev regardless of if it's a ev type we care about or not
+             */
+            this.latestRev = ev.rev
+            needsEmit = true
+            batch.push(ev)
+          }
         }
       }
-    }
 
-    if (needsEmit) {
+      if (needsEmit) {
-      try {
+        try {
-        this.emitter.emit('events', batch)
+          this.emitter.emit('events', batch)
-      } catch (e: any) {
+        } catch (e: any) {
-        logger.error(e, {
+          logger.error(e, {
-          context: `${LOGGER_CONTEXT}: process latest events`,
+            context: `${LOGGER_CONTEXT}: process latest events`,
-        })
+          })
+        }
       }
+    } catch (e: any) {
+      logger.error(e, {context: `${LOGGER_CONTEXT}: poll events failed`})
+
+      this.dispatch({
+        event: MessagesEventBusDispatchEvent.Error,
+        payload: {
+          exception: e,
+          code: MessagesEventBusErrorCode.PollFailed,
+          retry: () => {
+            this.init()
+          },
+        },
+      })
+    } finally {
+      this.isPolling = false
     }
   }
 }

src/state/messages/events/index.tsx

@@ -48,7 +48,7 @@ export function MessagesEventBusProvider({
       if (nextAppState === 'active') {
         bus.resume()
       } else {
-        bus.suspend()
+        bus.background()
       }
     }
 

src/state/messages/events/types.ts

@@ -10,6 +10,7 @@ export enum MessagesEventBusStatus {
   Initializing = 'initializing',
   Ready = 'ready',
   Error = 'error',
+  Backgrounded = 'backgrounded',
   Suspended = 'suspended',
 }
 
@@ -17,6 +18,7 @@ export enum MessagesEventBusDispatchEvent {
   Init = 'init',
   Ready = 'ready',
   Error = 'error',
+  Background = 'background',
   Suspend = 'suspend',
   Resume = 'resume',
 }
@@ -40,6 +42,9 @@ export type MessagesEventBusDispatch =
   | {
       event: MessagesEventBusDispatchEvent.Ready
     }
+  | {
+      event: MessagesEventBusDispatchEvent.Background
+    }
   | {
       event: MessagesEventBusDispatchEvent.Suspend
     }
@@ -79,6 +84,15 @@ export type MessagesEventBusState =
         handler: (events: ChatBskyConvoGetLog.OutputSchema['logs']) => void,
       ) => () => void
     }
+  | {
+      status: MessagesEventBusStatus.Backgrounded
+      rev: string | undefined
+      error: undefined
+      setPollInterval: (interval: number) => void
+      trail: (
+        handler: (events: ChatBskyConvoGetLog.OutputSchema['logs']) => void,
+      ) => () => void
+    }
   | {
       status: MessagesEventBusStatus.Suspended
       rev: string | undefined