[🐴] update convo list from message bus (#4189)

* update convo list from message bus

* don't increase unread count if you're the sender

* add refetch interval back

* Fix deleted message state copy

* only enable if `hasSession`

* Fix logged out handling

* increase refetch interval to 60s

* request 10s interval when message screen active

* use useAppState hook for convo resume/background

* Combine forces

* fix useFocusEffect logic

---------

Co-authored-by: Eric Bailey <git@esb.lol>

src/lib/hooks/useAppState.ts

@@ -0,0 +1,15 @@
+import {useEffect, useState} from 'react'
+import {AppState} from 'react-native'
+
+export function useAppState() {
+  const [state, setState] = useState(AppState.currentState)
+
+  useEffect(() => {
+    const sub = AppState.addEventListener('change', nextAppState => {
+      setState(nextAppState)
+    })
+    return () => sub.remove()
+  }, [])
+
+  return state
+}

src/screens/Messages/List/ChatListItem.tsx

@@ -105,7 +105,9 @@ function ChatListItemReady({
     lastMessageSentAt = convo.lastMessage.sentAt
   }
   if (ChatBskyConvoDefs.isDeletedMessageView(convo.lastMessage)) {
-    lastMessage = _(msg`Conversation deleted`)
+    lastMessage = isDeletedAccount
+      ? _(msg`Conversation deleted`)
+      : _(msg`Message deleted`)
   }
 
   const [showActions, setShowActions] = useState(false)

src/screens/Messages/List/index.tsx

@@ -1,16 +1,20 @@
-import React, {useCallback, useMemo, useState} from 'react'
+import React, {useCallback, useEffect, useMemo, useState} from 'react'
 import {View} from 'react-native'
 import {ChatBskyConvoDefs} from '@atproto/api'
 import {msg, Trans} from '@lingui/macro'
 import {useLingui} from '@lingui/react'
+import {useFocusEffect} from '@react-navigation/native'
 import {NativeStackScreenProps} from '@react-navigation/native-stack'
 
+import {useAppState} from '#/lib/hooks/useAppState'
 import {useInitialNumToRender} from '#/lib/hooks/useInitialNumToRender'
 import {MessagesTabNavigatorParams} from '#/lib/routes/types'
 import {cleanError} from '#/lib/strings/errors'
 import {logger} from '#/logger'
 import {isNative} from '#/platform/detection'
-import {useListConvos} from '#/state/queries/messages/list-converations'
+import {MESSAGE_SCREEN_POLL_INTERVAL} from '#/state/messages/convo/const'
+import {useMessagesEventBus} from '#/state/messages/events'
+import {useListConvosQuery} from '#/state/queries/messages/list-converations'
 import {List} from '#/view/com/util/List'
 import {ViewHeader} from '#/view/com/util/ViewHeader'
 import {CenteredView} from '#/view/com/util/Views'
@@ -52,7 +56,7 @@ export function MessagesScreen({navigation, route}: Props) {
   // this tab. We should immediately push to the conversation after pressing the notification.
   // After we push, reset with `setParams` so that this effect will fire next time we press a notification, even if
   // the conversation is the same as before
-  React.useEffect(() => {
+  useEffect(() => {
     if (pushToConversation) {
       navigation.navigate('MessagesConversation', {
         conversation: pushToConversation,
@@ -61,6 +65,22 @@ export function MessagesScreen({navigation, route}: Props) {
     }
   }, [navigation, pushToConversation])
 
+  // Request the poll interval to be 10s (or whatever the MESSAGE_SCREEN_POLL_INTERVAL is set to in the future)
+  // but only when the screen is active
+  const messagesBus = useMessagesEventBus()
+  const state = useAppState()
+  const isActive = state === 'active'
+  useFocusEffect(
+    useCallback(() => {
+      if (isActive) {
+        const unsub = messagesBus.requestPollInterval(
+          MESSAGE_SCREEN_POLL_INTERVAL,
+        )
+        return () => unsub()
+      }
+    }, [messagesBus, isActive]),
+  )
+
   const renderButton = useCallback(() => {
     return (
       <Link
@@ -88,7 +108,7 @@ export function MessagesScreen({navigation, route}: Props) {
     isError,
     error,
     refetch,
-  } = useListConvos({refetchInterval: 15_000})
+  } = useListConvosQuery()
 
   useRefreshOnFocus(refetch)
 

src/state/messages/convo/const.ts

@@ -1,4 +1,5 @@
 export const ACTIVE_POLL_INTERVAL = 3e3
+export const MESSAGE_SCREEN_POLL_INTERVAL = 10e3
 export const BACKGROUND_POLL_INTERVAL = 60e3
 export const INACTIVE_TIMEOUT = 60e3 * 5
 

src/state/messages/convo/index.tsx

@@ -1,8 +1,8 @@
 import React, {useContext, useState, useSyncExternalStore} from 'react'
-import {AppState} from 'react-native'
+import {useFocusEffect} from '@react-navigation/native'
-import {useFocusEffect, useIsFocused} from '@react-navigation/native'
 import {useQueryClient} from '@tanstack/react-query'
 
+import {useAppState} from '#/lib/hooks/useAppState'
 import {Convo} from '#/state/messages/convo/agent'
 import {
   ConvoParams,
@@ -58,7 +58,6 @@ export function ConvoProvider({
   convoId,
 }: Pick<ConvoParams, 'convoId'> & {children: React.ReactNode}) {
   const queryClient = useQueryClient()
-  const isScreenFocused = useIsFocused()
   const {getAgent} = useAgent()
   const events = useMessagesEventBus()
   const [convo] = useState(
@@ -72,16 +71,20 @@ export function ConvoProvider({
   const service = useSyncExternalStore(convo.subscribe, convo.getSnapshot)
   const {mutate: markAsRead} = useMarkAsReadMutation()
 
+  const appState = useAppState()
+  const isActive = appState === 'active'
   useFocusEffect(
     React.useCallback(() => {
-      convo.resume()
+      if (isActive) {
-      markAsRead({convoId})
+        convo.resume()
-
-      return () => {
-        convo.background()
         markAsRead({convoId})
+
+        return () => {
+          convo.background()
+          markAsRead({convoId})
+        }
       }
-    }, [convo, convoId, markAsRead]),
+    }, [isActive, convo, convoId, markAsRead]),
   )
 
   React.useEffect(() => {
@@ -101,25 +104,5 @@ export function ConvoProvider({
     })
   }, [convo, queryClient])
 
-  React.useEffect(() => {
-    const handleAppStateChange = (nextAppState: string) => {
-      if (isScreenFocused) {
-        if (nextAppState === 'active') {
-          convo.resume()
-        } else {
-          convo.background()
-        }
-
-        markAsRead({convoId})
-      }
-    }
-
-    const sub = AppState.addEventListener('change', handleAppStateChange)
-
-    return () => {
-      sub.remove()
-    }
-  }, [convoId, convo, isScreenFocused, markAsRead])
-
   return <ChatContext.Provider value={service}>{children}</ChatContext.Provider>
 }

src/state/messages/index.tsx

@@ -2,13 +2,16 @@ import React from 'react'
 
 import {CurrentConvoIdProvider} from '#/state/messages/current-convo-id'
 import {MessagesEventBusProvider} from '#/state/messages/events'
+import {ListConvosProvider} from '#/state/queries/messages/list-converations'
 import {MessageDraftsProvider} from './message-drafts'
 
 export function MessagesProvider({children}: {children: React.ReactNode}) {
   return (
     <CurrentConvoIdProvider>
       <MessageDraftsProvider>
-        <MessagesEventBusProvider>{children}</MessagesEventBusProvider>
+        <MessagesEventBusProvider>
+          <ListConvosProvider>{children}</ListConvosProvider>
+        </MessagesEventBusProvider>
       </MessageDraftsProvider>
     </CurrentConvoIdProvider>
   )

src/state/queries/messages/list-converations.ts

@@ -1,196 +0,0 @@
-import {useCallback, useMemo} from 'react'
-import {
-  ChatBskyConvoDefs,
-  ChatBskyConvoListConvos,
-  moderateProfile,
-} from '@atproto/api'
-import {
-  InfiniteData,
-  QueryClient,
-  useInfiniteQuery,
-  useQueryClient,
-} from '@tanstack/react-query'
-
-import {useCurrentConvoId} from '#/state/messages/current-convo-id'
-import {useModerationOpts} from '#/state/preferences/moderation-opts'
-import {DM_SERVICE_HEADERS} from '#/state/queries/messages/const'
-import {useAgent, useSession} from '#/state/session'
-
-export const RQKEY = ['convo-list']
-type RQPageParam = string | undefined
-
-export function useListConvos({refetchInterval}: {refetchInterval: number}) {
-  const {getAgent} = useAgent()
-
-  return useInfiniteQuery({
-    queryKey: RQKEY,
-    queryFn: async ({pageParam}) => {
-      const {data} = await getAgent().api.chat.bsky.convo.listConvos(
-        {cursor: pageParam},
-        {headers: DM_SERVICE_HEADERS},
-      )
-
-      return data
-    },
-    initialPageParam: undefined as RQPageParam,
-    getNextPageParam: lastPage => lastPage.cursor,
-    refetchInterval,
-  })
-}
-
-export function useUnreadMessageCount() {
-  const {currentConvoId} = useCurrentConvoId()
-  const {currentAccount} = useSession()
-  const convos = useListConvos({
-    refetchInterval: 30_000,
-  })
-  const moderationOpts = useModerationOpts()
-
-  const count = useMemo(() => {
-    return (
-      convos.data?.pages
-        .flatMap(page => page.convos)
-        .filter(convo => convo.id !== currentConvoId)
-        .reduce((acc, convo) => {
-          const otherMember = convo.members.find(
-            member => member.did !== currentAccount?.did,
-          )
-
-          if (!otherMember || !moderationOpts) return acc
-
-          const moderation = moderateProfile(otherMember, moderationOpts)
-          const shouldIgnore =
-            convo.muted ||
-            moderation.blocked ||
-            otherMember.did === 'missing.invalid'
-          const unreadCount = !shouldIgnore && convo.unreadCount > 0 ? 1 : 0
-
-          return acc + unreadCount
-        }, 0) ?? 0
-    )
-  }, [convos.data, currentAccount?.did, currentConvoId, moderationOpts])
-
-  return useMemo(() => {
-    return {
-      count,
-      numUnread: count > 0 ? (count > 30 ? '30+' : String(count)) : undefined,
-    }
-  }, [count])
-}
-
-type ConvoListQueryData = {
-  pageParams: Array<string | undefined>
-  pages: Array<ChatBskyConvoListConvos.OutputSchema>
-}
-
-export function useOnDeleteMessage() {
-  const queryClient = useQueryClient()
-
-  return useCallback(
-    (chatId: string, messageId: string) => {
-      queryClient.setQueryData(RQKEY, (old: ConvoListQueryData) => {
-        return optimisticUpdate(chatId, old, convo =>
-          messageId === convo.lastMessage?.id
-            ? {
-                ...convo,
-                lastMessage: {
-                  $type: 'chat.bsky.convo.defs#deletedMessageView',
-                  id: messageId,
-                  rev: '',
-                },
-              }
-            : convo,
-        )
-      })
-    },
-    [queryClient],
-  )
-}
-
-export function useOnNewMessage() {
-  const queryClient = useQueryClient()
-
-  return useCallback(
-    (chatId: string, message: ChatBskyConvoDefs.MessageView) => {
-      queryClient.setQueryData(RQKEY, (old: ConvoListQueryData) => {
-        return optimisticUpdate(chatId, old, convo => ({
-          ...convo,
-          lastMessage: message,
-          unreadCount: convo.unreadCount + 1,
-        }))
-      })
-      queryClient.invalidateQueries({queryKey: RQKEY})
-    },
-    [queryClient],
-  )
-}
-
-export function useOnCreateConvo() {
-  const queryClient = useQueryClient()
-
-  return useCallback(() => {
-    queryClient.invalidateQueries({queryKey: RQKEY})
-  }, [queryClient])
-}
-
-export function useOnMarkAsRead() {
-  const queryClient = useQueryClient()
-
-  return useCallback(
-    (chatId: string) => {
-      queryClient.setQueryData(RQKEY, (old: ConvoListQueryData) => {
-        return optimisticUpdate(chatId, old, convo => ({
-          ...convo,
-          unreadCount: 0,
-        }))
-      })
-    },
-    [queryClient],
-  )
-}
-
-function optimisticUpdate(
-  chatId: string,
-  old: ConvoListQueryData,
-  updateFn: (convo: ChatBskyConvoDefs.ConvoView) => ChatBskyConvoDefs.ConvoView,
-) {
-  if (!old) {
-    return old
-  }
-
-  return {
-    ...old,
-    pages: old.pages.map(page => ({
-      ...page,
-      convos: page.convos.map(convo =>
-        chatId === convo.id ? updateFn(convo) : convo,
-      ),
-    })),
-  }
-}
-
-export function* findAllProfilesInQueryData(
-  queryClient: QueryClient,
-  did: string,
-) {
-  const queryDatas = queryClient.getQueriesData<
-    InfiniteData<ChatBskyConvoListConvos.OutputSchema>
-  >({
-    queryKey: RQKEY,
-  })
-  for (const [_queryKey, queryData] of queryDatas) {
-    if (!queryData?.pages) {
-      continue
-    }
-
-    for (const page of queryData.pages) {
-      for (const convo of page.convos) {
-        for (const member of convo.members) {
-          if (member.did === did) {
-            yield member
-          }
-        }
-      }
-    }
-  }
-}

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

@@ -0,0 +1,317 @@
+import React, {
+  createContext,
+  useCallback,
+  useContext,
+  useEffect,
+  useMemo,
+} from 'react'
+import {
+  ChatBskyConvoDefs,
+  ChatBskyConvoListConvos,
+  moderateProfile,
+} from '@atproto/api'
+import {
+  InfiniteData,
+  QueryClient,
+  useInfiniteQuery,
+  useQueryClient,
+} from '@tanstack/react-query'
+
+import {useCurrentConvoId} from '#/state/messages/current-convo-id'
+import {useMessagesEventBus} from '#/state/messages/events'
+import {useModerationOpts} from '#/state/preferences/moderation-opts'
+import {DM_SERVICE_HEADERS} from '#/state/queries/messages/const'
+import {useAgent, useSession} from '#/state/session'
+
+export const RQKEY = ['convo-list']
+type RQPageParam = string | undefined
+
+export function useListConvosQuery() {
+  const {getAgent} = useAgent()
+
+  return useInfiniteQuery({
+    queryKey: RQKEY,
+    queryFn: async ({pageParam}) => {
+      const {data} = await getAgent().api.chat.bsky.convo.listConvos(
+        {cursor: pageParam},
+        {headers: DM_SERVICE_HEADERS},
+      )
+
+      return data
+    },
+    initialPageParam: undefined as RQPageParam,
+    getNextPageParam: lastPage => lastPage.cursor,
+    // refetch every 60 seconds since we can't get *all* info from the logs
+    // i.e. reading chats on another device won't update the unread count
+    refetchInterval: 60_000,
+  })
+}
+
+const ListConvosContext = createContext<ChatBskyConvoDefs.ConvoView[] | null>(
+  null,
+)
+
+export function useListConvos() {
+  const ctx = useContext(ListConvosContext)
+  if (!ctx) {
+    throw new Error('useListConvos must be used within a ListConvosProvider')
+  }
+  return ctx
+}
+
+export function ListConvosProvider({children}: {children: React.ReactNode}) {
+  const {hasSession} = useSession()
+
+  if (!hasSession) {
+    return (
+      <ListConvosContext.Provider value={[]}>
+        {children}
+      </ListConvosContext.Provider>
+    )
+  }
+
+  return <ListConvosProviderInner>{children}</ListConvosProviderInner>
+}
+
+export function ListConvosProviderInner({
+  children,
+}: {
+  children: React.ReactNode
+}) {
+  const {refetch, data} = useListConvosQuery()
+  const messagesBus = useMessagesEventBus()
+  const queryClient = useQueryClient()
+  const {currentConvoId} = useCurrentConvoId()
+  const {currentAccount} = useSession()
+
+  useEffect(() => {
+    const unsub = messagesBus.on(
+      events => {
+        if (events.type !== 'logs') return
+
+        events.logs.forEach(log => {
+          if (ChatBskyConvoDefs.isLogBeginConvo(log)) {
+            refetch()
+          } else if (ChatBskyConvoDefs.isLogLeaveConvo(log)) {
+            queryClient.setQueryData(RQKEY, (old: ConvoListQueryData) =>
+              optimisticDelete(log.convoId, old),
+            )
+          } else if (ChatBskyConvoDefs.isLogDeleteMessage(log)) {
+            queryClient.setQueryData(RQKEY, (old: ConvoListQueryData) =>
+              optimisticUpdate(log.convoId, old, convo =>
+                log.message.id === convo.lastMessage?.id
+                  ? {
+                      ...convo,
+                      rev: log.rev,
+                      lastMessage: log.message,
+                    }
+                  : convo,
+              ),
+            )
+          } else if (ChatBskyConvoDefs.isLogCreateMessage(log)) {
+            queryClient.setQueryData(RQKEY, (old: ConvoListQueryData) => {
+              if (!old) return old
+
+              function updateConvo(convo: ChatBskyConvoDefs.ConvoView) {
+                if (!ChatBskyConvoDefs.isLogCreateMessage(log)) return convo
+
+                let unreadCount = convo.unreadCount
+                if (convo.id !== currentConvoId) {
+                  if (
+                    ChatBskyConvoDefs.isMessageView(log.message) ||
+                    ChatBskyConvoDefs.isDeletedMessageView(log.message)
+                  ) {
+                    if (log.message.sender.did !== currentAccount?.did) {
+                      unreadCount++
+                    }
+                  }
+                } else {
+                  unreadCount = 0
+                }
+
+                return {
+                  ...convo,
+                  rev: log.rev,
+                  lastMessage: log.message,
+                  unreadCount,
+                }
+              }
+
+              function filterConvoFromPage(
+                convo: ChatBskyConvoDefs.ConvoView[],
+              ) {
+                return convo.filter(c => c.id !== log.convoId)
+              }
+
+              const existingConvo = getConvoFromQueryData(log.convoId, old)
+
+              if (existingConvo) {
+                return {
+                  ...old,
+                  pages: old.pages.map((page, i) => {
+                    if (i === 0) {
+                      return {
+                        ...page,
+                        convos: [
+                          updateConvo(existingConvo),
+                          ...filterConvoFromPage(page.convos),
+                        ],
+                      }
+                    }
+                    return {
+                      ...page,
+                      convos: filterConvoFromPage(page.convos),
+                    }
+                  }),
+                }
+              } else {
+                refetch()
+              }
+            })
+          }
+        })
+      },
+      {
+        // get events for all chats
+        convoId: undefined,
+      },
+    )
+
+    return () => unsub()
+  }, [messagesBus, currentConvoId, refetch, queryClient, currentAccount?.did])
+
+  const ctx = useMemo(() => {
+    return data?.pages.flatMap(page => page.convos) ?? []
+  }, [data])
+
+  return (
+    <ListConvosContext.Provider value={ctx}>
+      {children}
+    </ListConvosContext.Provider>
+  )
+}
+
+export function useUnreadMessageCount() {
+  const {currentConvoId} = useCurrentConvoId()
+  const {currentAccount} = useSession()
+  const convos = useListConvos()
+  const moderationOpts = useModerationOpts()
+
+  const count = useMemo(() => {
+    return (
+      convos
+        .filter(convo => convo.id !== currentConvoId)
+        .reduce((acc, convo) => {
+          const otherMember = convo.members.find(
+            member => member.did !== currentAccount?.did,
+          )
+
+          if (!otherMember || !moderationOpts) return acc
+
+          const moderation = moderateProfile(otherMember, moderationOpts)
+          const shouldIgnore =
+            convo.muted ||
+            moderation.blocked ||
+            otherMember.did === 'missing.invalid'
+          const unreadCount = !shouldIgnore && convo.unreadCount > 0 ? 1 : 0
+
+          return acc + unreadCount
+        }, 0) ?? 0
+    )
+  }, [convos, currentAccount?.did, currentConvoId, moderationOpts])
+
+  return useMemo(() => {
+    return {
+      count,
+      numUnread: count > 0 ? (count > 30 ? '30+' : String(count)) : undefined,
+    }
+  }, [count])
+}
+
+type ConvoListQueryData = {
+  pageParams: Array<string | undefined>
+  pages: Array<ChatBskyConvoListConvos.OutputSchema>
+}
+
+export function useOnMarkAsRead() {
+  const queryClient = useQueryClient()
+
+  return useCallback(
+    (chatId: string) => {
+      queryClient.setQueryData(RQKEY, (old: ConvoListQueryData) => {
+        return optimisticUpdate(chatId, old, convo => ({
+          ...convo,
+          unreadCount: 0,
+        }))
+      })
+    },
+    [queryClient],
+  )
+}
+
+function optimisticUpdate(
+  chatId: string,
+  old: ConvoListQueryData,
+  updateFn: (convo: ChatBskyConvoDefs.ConvoView) => ChatBskyConvoDefs.ConvoView,
+) {
+  if (!old) return old
+
+  return {
+    ...old,
+    pages: old.pages.map(page => ({
+      ...page,
+      convos: page.convos.map(convo =>
+        chatId === convo.id ? updateFn(convo) : convo,
+      ),
+    })),
+  }
+}
+
+function optimisticDelete(chatId: string, old: ConvoListQueryData) {
+  if (!old) return old
+
+  return {
+    ...old,
+    pages: old.pages.map(page => ({
+      ...page,
+      convos: page.convos.filter(convo => chatId !== convo.id),
+    })),
+  }
+}
+
+function getConvoFromQueryData(chatId: string, old: ConvoListQueryData) {
+  for (const page of old.pages) {
+    for (const convo of page.convos) {
+      if (convo.id === chatId) {
+        return convo
+      }
+    }
+  }
+  return null
+}
+
+export function* findAllProfilesInQueryData(
+  queryClient: QueryClient,
+  did: string,
+) {
+  const queryDatas = queryClient.getQueriesData<
+    InfiniteData<ChatBskyConvoListConvos.OutputSchema>
+  >({
+    queryKey: RQKEY,
+  })
+  for (const [_queryKey, queryData] of queryDatas) {
+    if (!queryData?.pages) {
+      continue
+    }
+
+    for (const page of queryData.pages) {
+      for (const convo of page.convos) {
+        for (const member of convo.members) {
+          if (member.did === did) {
+            yield member
+          }
+        }
+      }
+    }
+  }
+}