From 18194e21663f957c521b4a19b4118dbfa36a9fa1 Mon Sep 17 00:00:00 2001 From: Samuel Newman Date: Fri, 12 Jun 2026 10:52:06 +0300 Subject: [PATCH] extract chat log event handler for testability Pure refactor: move the log-event handling out of the ListConvosProviderInner effect closure into an exported handleConvoLogEvents function, so tests can drive it directly with a QueryClient and a batch of logs. Co-Authored-By: Claude Fable 5 --- .../queries/messages/list-conversations.tsx | 1272 +++++++++-------- 1 file changed, 642 insertions(+), 630 deletions(-) diff --git a/src/state/queries/messages/list-conversations.tsx b/src/state/queries/messages/list-conversations.tsx index 3ba857e155..bddfe08080 100644 --- a/src/state/queries/messages/list-conversations.tsx +++ b/src/state/queries/messages/list-conversations.tsx @@ -19,6 +19,7 @@ import throttle from 'lodash.throttle' import {DM_SERVICE_HEADERS} from '#/lib/constants' import {useCurrentConvoId} from '#/state/messages/current-convo-id' import {useMessagesEventBus} from '#/state/messages/events' +import {type MessagesEventBusEvent} from '#/state/messages/events/types' import {useModerationOpts} from '#/state/preferences/moderation-opts' import {invalidateJoinLinkPreviewsForConvo} from '#/state/queries/join-links' import {useAgent, useSession} from '#/state/session' @@ -207,636 +208,13 @@ export function ListConvosProviderInner({ const unsub = messagesBus.on( events => { if (events.type !== 'logs') return - - function mutateMembers( - convoId: string, - fn: ( - members: ChatBskyActorDefs.ProfileViewBasic[], - ) => ChatBskyActorDefs.ProfileViewBasic[], - ) { - queryClient.setQueryData( - listConvoMembersQueryKey(convoId), - old => { - if (!old) return // query doesn't exist yet, skip - return fn(old) - }, - ) - } - - function updateConvoInAllLists( - convoId: string, - fn: ( - convo: ChatBskyConvoDefs.ConvoView, - ) => ChatBskyConvoDefs.ConvoView, - ) { - queryClient.setQueriesData( - {queryKey: [RQKEY_ROOT]}, - old => optimisticUpdate(convoId, old, fn), - ) - queryClient.setQueriesData( - {queryKey: [REQUESTS_RQKEY_ROOT]}, - old => optimisticUpdateRequest(convoId, old, fn), - ) - } - - function mutateConvoView( - convoId: string, - fn: ( - convo: ChatBskyConvoDefs.ConvoView, - ) => ChatBskyConvoDefs.ConvoView, - ) { - queryClient.setQueryData( - CONVO_KEY(convoId), - old => (old ? fn(old) : old), - ) - updateConvoInAllLists(convoId, fn) - } - - function deleteConvoFromAllLists(convoId: string) { - queryClient.setQueriesData( - {queryKey: [RQKEY_ROOT]}, - old => optimisticDelete(convoId, old), - ) - queryClient.setQueriesData( - {queryKey: [REQUESTS_RQKEY_ROOT]}, - old => optimisticDeleteRequest(convoId, old), - ) - } - - function handleMemberAdded( - convoId: string, - did: string, - relatedProfiles: ChatBskyActorDefs.ProfileViewBasic[], - rev: string, - ) { - const newMember = relatedProfiles.find(r => r.did === did) - if (!newMember) return - // If the optimistic add already added them, skip the - // memberCount bump to avoid double-counting. - const alreadyKnownMember = - queryClient - .getQueryData< - ChatBskyActorDefs.ProfileViewBasic[] - >(listConvoMembersQueryKey(convoId)) - ?.some(m => m.did === did) ?? false - mutateMembers(convoId, list => - list.some(m => m.did === did) ? list : list.concat(newMember), - ) - mutateConvoView( - convoId, - withRevGuard(rev, convo => - addMemberToConvoView(convo, newMember, rev, alreadyKnownMember), - ), - ) - } - - function handleMemberRemoved( - convoId: string, - did: string, - rev: string, - ) { - // If the optimistic remove already dropped them from the full - // list, skip the memberCount decrement to avoid double-counting. - const alreadyRemovedMember = - queryClient - .getQueryData< - ChatBskyActorDefs.ProfileViewBasic[] - >(listConvoMembersQueryKey(convoId)) - ?.some(m => m.did === did) === false - mutateMembers(convoId, list => list.filter(m => m.did !== did)) - mutateConvoView( - convoId, - withRevGuard(rev, convo => - removeMemberFromConvoView(convo, did, rev, alreadyRemovedMember), - ), - ) - } - - for (const log of events.logs) { - if (ChatBskyConvoDefs.isLogBeginConvo(log)) { - debouncedRefetch() - } else if (ChatBskyConvoDefs.isLogLeaveConvo(log)) { - deleteConvoFromAllLists(log.convoId) - // The viewer is no longer in this convo (they left on another - // device, or were removed - removed members receive a - // logLeaveConvo, not a logRemoveMember). Refetch any cached join - // link preview so its viewer state reflects the lost membership. - void invalidateJoinLinkPreviewsForConvo(queryClient, log.convoId) - } else if (ChatBskyConvoDefs.isLogDeleteMessage(log)) { - updateConvoInAllLists( - log.convoId, - withRevGuard(log.rev, convo => { - if ( - (ChatBskyConvoDefs.isDeletedMessageView(log.message) || - ChatBskyConvoDefs.isMessageView(log.message)) && - (ChatBskyConvoDefs.isDeletedMessageView(convo.lastMessage) || - ChatBskyConvoDefs.isMessageView(convo.lastMessage)) - ) { - return log.message.id === convo.lastMessage.id - ? { - ...convo, - rev: log.rev, - lastMessage: log.message, - } - : 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 - - // Get all matching queries - const queries = queryClient.getQueriesData({ - queryKey: [RQKEY_ROOT], - }) - - // Check if convo exists in any query - let foundConvo: ChatBskyConvoDefs.ConvoView | null = null - for (const [_key, query] of queries) { - if (!query) continue - const convo = getConvoFromQueryData(logRef.convoId, query) - if (convo) { - foundConvo = convo - break - } - } - - if (!foundConvo) { - // 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() - 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 - const relatedProfilesSansMembers = ( - logRef.relatedProfiles ?? [] - ).filter( - profile => - !foundConvo.members.some(member => member.did === profile.did), - ) - - // Update the convo - const updatedConvo = { - ...foundConvo, - members: [...foundConvo.members, ...relatedProfilesSansMembers], - rev: logRef.rev, - lastMessage: logRef.message, - unreadCount: - foundConvo.id !== currentConvoId - ? (ChatBskyConvoDefs.isMessageView(logRef.message) || - ChatBskyConvoDefs.isDeletedMessageView(logRef.message)) && - logRef.message.sender.did !== currentAccount?.did - ? foundConvo.unreadCount + 1 - : foundConvo.unreadCount - : 0, - } - - function filterConvoFromPage(convo: ChatBskyConvoDefs.ConvoView[]) { - return convo.filter(c => c.id !== logRef.convoId) - } - - // Update all matching queries - function updateFn(old?: ConvoListQueryData) { - if (!old) return old - return { - ...old, - pages: old.pages.map((page, i) => { - if (i === 0) { - return { - ...page, - convos: [ - updatedConvo, - ...filterConvoFromPage(page.convos), - ], - } - } - return { - ...page, - convos: filterConvoFromPage(page.convos), - } - }), - } - } - // always update the unread ones, where the convo qualifies - queryClient.setQueriesData( - { - queryKey: RQKEY_PARTIAL('all', 'unread'), - predicate: convoListQueryPredicate(updatedConvo), - }, - (old?: ConvoListQueryData) => - old - ? updateFn(old) - : ({ - pageParams: [undefined], - pages: [{convos: [updatedConvo], cursor: undefined}], - } satisfies ConvoListQueryData), - ) - // update the other ones based on status of the incoming message - if (updatedConvo.status === 'accepted') { - queryClient.setQueriesData( - { - queryKey: RQKEY_PARTIAL('accepted'), - predicate: convoListQueryPredicate(updatedConvo), - }, - updateFn, - ) - } else if (updatedConvo.status === 'request') { - queryClient.setQueriesData( - { - queryKey: RQKEY_PARTIAL('request'), - predicate: convoListQueryPredicate(updatedConvo), - }, - updateFn, - ) - // also move-to-top in the new requests cache - queryClient.setQueriesData( - {queryKey: [REQUESTS_RQKEY_ROOT]}, - old => moveConvoToTopInRequests(updatedConvo, old), - ) - } - } else if (ChatBskyConvoDefs.isLogReadMessage(log)) { - updateConvoInAllLists( - log.convoId, - withRevGuard(log.rev, convo => ({ - ...convo, - unreadCount: 0, - rev: log.rev, - })), - ) - } else if (ChatBskyConvoDefs.isLogReadConvo(log)) { - updateConvoInAllLists( - log.convoId, - withRevGuard(log.rev, convo => ({ - ...convo, - unreadCount: 0, - rev: log.rev, - })), - ) - } else if (ChatBskyConvoDefs.isLogAcceptConvo(log)) { - const requestQueries = - queryClient.getQueriesData({ - queryKey: RQKEY_PARTIAL('request'), - }) - let foundConvo: ChatBskyConvoDefs.ConvoView | null = null - for (const [_key, data] of requestQueries) { - if (!data) continue - foundConvo = getConvoFromQueryData(log.convoId, data) - 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() - 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), - ) - // also remove from the new requests cache - queryClient.setQueriesData( - {queryKey: [REQUESTS_RQKEY_ROOT]}, - old => optimisticDeleteRequest(log.convoId, old), - ) - queryClient.setQueriesData( - { - queryKey: RQKEY_PARTIAL('accepted'), - predicate: convoListQueryPredicate(acceptedConvo), - }, - (old?: ConvoListQueryData) => { - if (!old) { - debouncedRefetch() - return old - } - return { - ...old, - pages: old.pages.map((page, i) => { - if (i === 0) { - return { - ...page, - convos: [ - acceptedConvo, - ...page.convos.filter(c => c.id !== log.convoId), - ], - } - } - return { - ...page, - convos: page.convos.filter(c => c.id !== log.convoId), - } - }), - } - }, - ) - } else if (ChatBskyConvoDefs.isLogMuteConvo(log)) { - mutateConvoView( - log.convoId, - withRevGuard(log.rev, convo => ({ - ...convo, - muted: true, - rev: log.rev, - })), - ) - } else if (ChatBskyConvoDefs.isLogUnmuteConvo(log)) { - mutateConvoView( - log.convoId, - withRevGuard(log.rev, convo => ({ - ...convo, - muted: false, - rev: log.rev, - })), - ) - } else if (ChatBskyConvoDefs.isLogLockConvo(log)) { - mutateConvoView( - log.convoId, - withRevGuard(log.rev, convo => { - if (ChatBskyConvoDefs.isGroupConvo(convo.kind)) { - return { - ...convo, - kind: {...convo.kind, lockStatus: 'locked'}, - 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, - withRevGuard(log.rev, convo => { - if (ChatBskyConvoDefs.isGroupConvo(convo.kind)) { - return { - ...convo, - kind: { - ...convo.kind, - lockStatus: 'unlocked', - // An unlocked convo cannot be moderation-locked. - lockStatusModerationOverride: false, - }, - rev: log.rev, - } - } - return {...convo, rev: log.rev} - }), - ) - } else if (ChatBskyConvoDefs.isLogLockConvoPermanently(log)) { - mutateConvoView( - log.convoId, - withRevGuard(log.rev, convo => { - if (ChatBskyConvoDefs.isGroupConvo(convo.kind)) { - return { - ...convo, - kind: {...convo.kind, lockStatus: 'locked-permanently'}, - rev: log.rev, - } - } - return {...convo, rev: log.rev} - }), - ) - } else if ( - ChatBskyConvoDefs.isLogCreateJoinLink(log) || - ChatBskyConvoDefs.isLogEditJoinLink(log) || - ChatBskyConvoDefs.isLogEnableJoinLink(log) || - ChatBskyConvoDefs.isLogDisableJoinLink(log) - ) { - // Join link data not included in the log event, trigger refetch to get it - debouncedRefetch() - } else if (ChatBskyConvoDefs.isLogEditGroup(log)) { - // Updated group details (name etc.) aren't included in the log - // event, so refetch to pick them up. - debouncedRefetch() - } else if ( - ChatBskyConvoDefs.isLogApproveJoinRequest(log) || - ChatBskyConvoDefs.isLogRejectJoinRequest(log) - ) { - // Route through mutateConvoView (not updateConvoInAllLists) so the - // single-convo cache updates too, keeping the in-convo requests - // banner in sync. - mutateConvoView( - 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, - 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, - withRevGuard(log.rev, convo => { - if (!ChatBskyConvoDefs.isGroupConvo(convo.kind)) { - return {...convo, rev: log.rev} - } - return { - ...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. - debouncedRefetch() - } 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, - withRevGuard(log.rev, convo => - applyJoinRequestCountDelta(convo, log.rev, -1), - ), - ) - } else if (ChatBskyConvoDefs.isLogWithdrawOutgoingJoinRequest(log)) { - // The viewer rescinded their own outgoing join request (possibly on - // another device). Remove it from the requests inbox cache. - queryClient.setQueriesData( - {queryKey: [REQUESTS_RQKEY_ROOT]}, - old => optimisticDeleteJoinRequest(log.convoId, old), - ) - } else if (ChatBskyConvoDefs.isLogAddReaction(log)) { - updateConvoInAllLists( - log.convoId, - withRevGuard(log.rev, convo => { - // add relatedProfiles to members list, but making sure to dedupe - const relatedProfilesSansMembers = ( - log.relatedProfiles ?? [] - ).filter( - profile => - !convo.members.some(member => member.did === profile.did), - ) - return { - ...convo, - members: [...convo.members, ...relatedProfilesSansMembers], - lastReaction: { - $type: 'chat.bsky.convo.defs#messageAndReactionView', - reaction: log.reaction, - message: log.message, - }, - rev: log.rev, - } - }), - ) - } else if (ChatBskyConvoDefs.isLogAddMember(log)) { - const data = log.message.data - if ( - bsky.dangerousIsType( - data, - ChatBskyConvoDefs.isSystemMessageDataAddMember, - ) - ) { - handleMemberAdded( - log.convoId, - data.member.did, - log.relatedProfiles, - log.rev, - ) - } - // Refetch so the server can refresh the curated members list. - void queryClient.invalidateQueries({ - queryKey: CONVO_KEY(log.convoId), - }) - debouncedRefetch() - } else if (ChatBskyConvoDefs.isLogRemoveMember(log)) { - const data = log.message.data - if ( - bsky.dangerousIsType( - data, - ChatBskyConvoDefs.isSystemMessageDataRemoveMember, - ) - ) { - handleMemberRemoved(log.convoId, data.member.did, log.rev) - } - // Refetch so the server can refill the curated members list. - void queryClient.invalidateQueries({ - queryKey: CONVO_KEY(log.convoId), - }) - debouncedRefetch() - } else if (ChatBskyConvoDefs.isLogMemberJoin(log)) { - const data = log.message.data - if ( - bsky.dangerousIsType( - data, - ChatBskyConvoDefs.isSystemMessageDataMemberJoin, - ) - ) { - handleMemberAdded( - log.convoId, - data.member.did, - log.relatedProfiles, - log.rev, - ) - } - void queryClient.invalidateQueries({ - queryKey: CONVO_KEY(log.convoId), - }) - debouncedRefetch() - } else if (ChatBskyConvoDefs.isLogMemberLeave(log)) { - const data = log.message.data - if ( - bsky.dangerousIsType( - data, - ChatBskyConvoDefs.isSystemMessageDataMemberLeave, - ) - ) { - handleMemberRemoved(log.convoId, data.member.did, log.rev) - } - void queryClient.invalidateQueries({ - queryKey: CONVO_KEY(log.convoId), - }) - debouncedRefetch() - } else if (ChatBskyConvoDefs.isLogRemoveReaction(log)) { - queryClient.setQueriesData( - {queryKey: [RQKEY_ROOT]}, - (old?: ConvoListQueryData) => - optimisticUpdate( - log.convoId, - old, - withRevGuard(log.rev, convo => { - if ( - // if the convo is the same - log.convoId === convo.id && - ChatBskyConvoDefs.isMessageAndReactionView( - convo.lastReaction, - ) && - ChatBskyConvoDefs.isMessageView(log.message) && - // ...and the message is the same - convo.lastReaction.message.id === log.message.id && - // ...and the reaction is the same - convo.lastReaction.reaction.sender.did === - log.reaction.sender.did && - convo.lastReaction.reaction.value === log.reaction.value - ) { - return { - ...convo, - // ...remove the reaction. hopefully they didn't react twice in a row! - lastReaction: undefined, - rev: log.rev, - } - } else { - return convo - } - }), - ), - ) - } - } + handleConvoLogEvents({ + queryClient, + logs: events.logs, + currentConvoId, + currentAccountDid: currentAccount?.did, + onRefetchNeeded: debouncedRefetch, + }) }, { // get events for all chats @@ -868,6 +246,640 @@ export function ListConvosProviderInner({ ) } +/** + * Applies a batch of chat log events to the React Query caches (convo lists, + * single-convo, members, requests). Exported as a pure function - separate + * from the event-bus subscription in ListConvosProviderInner - so it can be + * unit-tested directly without mounting the provider. + */ +export function handleConvoLogEvents({ + queryClient, + logs, + currentConvoId, + currentAccountDid, + onRefetchNeeded, +}: { + queryClient: QueryClient + logs: Extract['logs'] + currentConvoId: string | undefined + currentAccountDid: string | undefined + onRefetchNeeded: () => void +}): void { + function mutateMembers( + convoId: string, + fn: ( + members: ChatBskyActorDefs.ProfileViewBasic[], + ) => ChatBskyActorDefs.ProfileViewBasic[], + ) { + queryClient.setQueryData( + listConvoMembersQueryKey(convoId), + old => { + if (!old) return // query doesn't exist yet, skip + return fn(old) + }, + ) + } + + function updateConvoInAllLists( + convoId: string, + fn: (convo: ChatBskyConvoDefs.ConvoView) => ChatBskyConvoDefs.ConvoView, + ) { + queryClient.setQueriesData( + {queryKey: [RQKEY_ROOT]}, + old => optimisticUpdate(convoId, old, fn), + ) + queryClient.setQueriesData( + {queryKey: [REQUESTS_RQKEY_ROOT]}, + old => optimisticUpdateRequest(convoId, old, fn), + ) + } + + function mutateConvoView( + convoId: string, + fn: (convo: ChatBskyConvoDefs.ConvoView) => ChatBskyConvoDefs.ConvoView, + ) { + queryClient.setQueryData( + CONVO_KEY(convoId), + old => (old ? fn(old) : old), + ) + updateConvoInAllLists(convoId, fn) + } + + function deleteConvoFromAllLists(convoId: string) { + queryClient.setQueriesData( + {queryKey: [RQKEY_ROOT]}, + old => optimisticDelete(convoId, old), + ) + queryClient.setQueriesData( + {queryKey: [REQUESTS_RQKEY_ROOT]}, + old => optimisticDeleteRequest(convoId, old), + ) + } + + function handleMemberAdded( + convoId: string, + did: string, + relatedProfiles: ChatBskyActorDefs.ProfileViewBasic[], + rev: string, + ) { + const newMember = relatedProfiles.find(r => r.did === did) + if (!newMember) return + // If the optimistic add already added them, skip the + // memberCount bump to avoid double-counting. + const alreadyKnownMember = + queryClient + .getQueryData< + ChatBskyActorDefs.ProfileViewBasic[] + >(listConvoMembersQueryKey(convoId)) + ?.some(m => m.did === did) ?? false + mutateMembers(convoId, list => + list.some(m => m.did === did) ? list : list.concat(newMember), + ) + mutateConvoView( + convoId, + withRevGuard(rev, convo => + addMemberToConvoView(convo, newMember, rev, alreadyKnownMember), + ), + ) + } + + function handleMemberRemoved(convoId: string, did: string, rev: string) { + // If the optimistic remove already dropped them from the full + // list, skip the memberCount decrement to avoid double-counting. + const alreadyRemovedMember = + queryClient + .getQueryData< + ChatBskyActorDefs.ProfileViewBasic[] + >(listConvoMembersQueryKey(convoId)) + ?.some(m => m.did === did) === false + mutateMembers(convoId, list => list.filter(m => m.did !== did)) + mutateConvoView( + convoId, + withRevGuard(rev, convo => + removeMemberFromConvoView(convo, did, rev, alreadyRemovedMember), + ), + ) + } + + for (const log of logs) { + if (ChatBskyConvoDefs.isLogBeginConvo(log)) { + onRefetchNeeded() + } else if (ChatBskyConvoDefs.isLogLeaveConvo(log)) { + deleteConvoFromAllLists(log.convoId) + // The viewer is no longer in this convo (they left on another + // device, or were removed - removed members receive a + // logLeaveConvo, not a logRemoveMember). Refetch any cached join + // link preview so its viewer state reflects the lost membership. + void invalidateJoinLinkPreviewsForConvo(queryClient, log.convoId) + } else if (ChatBskyConvoDefs.isLogDeleteMessage(log)) { + updateConvoInAllLists( + log.convoId, + withRevGuard(log.rev, convo => { + if ( + (ChatBskyConvoDefs.isDeletedMessageView(log.message) || + ChatBskyConvoDefs.isMessageView(log.message)) && + (ChatBskyConvoDefs.isDeletedMessageView(convo.lastMessage) || + ChatBskyConvoDefs.isMessageView(convo.lastMessage)) + ) { + return log.message.id === convo.lastMessage.id + ? { + ...convo, + rev: log.rev, + lastMessage: log.message, + } + : 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 + + // Get all matching queries + const queries = queryClient.getQueriesData({ + queryKey: [RQKEY_ROOT], + }) + + // Check if convo exists in any query + let foundConvo: ChatBskyConvoDefs.ConvoView | null = null + for (const [_key, query] of queries) { + if (!query) continue + const convo = getConvoFromQueryData(logRef.convoId, query) + if (convo) { + foundConvo = convo + break + } + } + + if (!foundConvo) { + // 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. + onRefetchNeeded() + 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 + const relatedProfilesSansMembers = (logRef.relatedProfiles ?? []).filter( + profile => + !foundConvo.members.some(member => member.did === profile.did), + ) + + // Update the convo + const updatedConvo = { + ...foundConvo, + members: [...foundConvo.members, ...relatedProfilesSansMembers], + rev: logRef.rev, + lastMessage: logRef.message, + unreadCount: + foundConvo.id !== currentConvoId + ? (ChatBskyConvoDefs.isMessageView(logRef.message) || + ChatBskyConvoDefs.isDeletedMessageView(logRef.message)) && + logRef.message.sender.did !== currentAccountDid + ? foundConvo.unreadCount + 1 + : foundConvo.unreadCount + : 0, + } + + function filterConvoFromPage(convo: ChatBskyConvoDefs.ConvoView[]) { + return convo.filter(c => c.id !== logRef.convoId) + } + + // Update all matching queries + function updateFn(old?: ConvoListQueryData) { + if (!old) return old + return { + ...old, + pages: old.pages.map((page, i) => { + if (i === 0) { + return { + ...page, + convos: [updatedConvo, ...filterConvoFromPage(page.convos)], + } + } + return { + ...page, + convos: filterConvoFromPage(page.convos), + } + }), + } + } + // always update the unread ones, where the convo qualifies + queryClient.setQueriesData( + { + queryKey: RQKEY_PARTIAL('all', 'unread'), + predicate: convoListQueryPredicate(updatedConvo), + }, + (old?: ConvoListQueryData) => + old + ? updateFn(old) + : ({ + pageParams: [undefined], + pages: [{convos: [updatedConvo], cursor: undefined}], + } satisfies ConvoListQueryData), + ) + // update the other ones based on status of the incoming message + if (updatedConvo.status === 'accepted') { + queryClient.setQueriesData( + { + queryKey: RQKEY_PARTIAL('accepted'), + predicate: convoListQueryPredicate(updatedConvo), + }, + updateFn, + ) + } else if (updatedConvo.status === 'request') { + queryClient.setQueriesData( + { + queryKey: RQKEY_PARTIAL('request'), + predicate: convoListQueryPredicate(updatedConvo), + }, + updateFn, + ) + // also move-to-top in the new requests cache + queryClient.setQueriesData( + {queryKey: [REQUESTS_RQKEY_ROOT]}, + old => moveConvoToTopInRequests(updatedConvo, old), + ) + } + } else if (ChatBskyConvoDefs.isLogReadMessage(log)) { + updateConvoInAllLists( + log.convoId, + withRevGuard(log.rev, convo => ({ + ...convo, + unreadCount: 0, + rev: log.rev, + })), + ) + } else if (ChatBskyConvoDefs.isLogReadConvo(log)) { + updateConvoInAllLists( + log.convoId, + withRevGuard(log.rev, convo => ({ + ...convo, + unreadCount: 0, + rev: log.rev, + })), + ) + } else if (ChatBskyConvoDefs.isLogAcceptConvo(log)) { + const requestQueries = queryClient.getQueriesData({ + queryKey: RQKEY_PARTIAL('request'), + }) + let foundConvo: ChatBskyConvoDefs.ConvoView | null = null + for (const [_key, data] of requestQueries) { + if (!data) continue + foundConvo = getConvoFromQueryData(log.convoId, data) + 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. + onRefetchNeeded() + 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), + ) + // also remove from the new requests cache + queryClient.setQueriesData( + {queryKey: [REQUESTS_RQKEY_ROOT]}, + old => optimisticDeleteRequest(log.convoId, old), + ) + queryClient.setQueriesData( + { + queryKey: RQKEY_PARTIAL('accepted'), + predicate: convoListQueryPredicate(acceptedConvo), + }, + (old?: ConvoListQueryData) => { + if (!old) { + onRefetchNeeded() + return old + } + return { + ...old, + pages: old.pages.map((page, i) => { + if (i === 0) { + return { + ...page, + convos: [ + acceptedConvo, + ...page.convos.filter(c => c.id !== log.convoId), + ], + } + } + return { + ...page, + convos: page.convos.filter(c => c.id !== log.convoId), + } + }), + } + }, + ) + } else if (ChatBskyConvoDefs.isLogMuteConvo(log)) { + mutateConvoView( + log.convoId, + withRevGuard(log.rev, convo => ({ + ...convo, + muted: true, + rev: log.rev, + })), + ) + } else if (ChatBskyConvoDefs.isLogUnmuteConvo(log)) { + mutateConvoView( + log.convoId, + withRevGuard(log.rev, convo => ({ + ...convo, + muted: false, + rev: log.rev, + })), + ) + } else if (ChatBskyConvoDefs.isLogLockConvo(log)) { + mutateConvoView( + log.convoId, + withRevGuard(log.rev, convo => { + if (ChatBskyConvoDefs.isGroupConvo(convo.kind)) { + return { + ...convo, + kind: {...convo.kind, lockStatus: 'locked'}, + 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, + withRevGuard(log.rev, convo => { + if (ChatBskyConvoDefs.isGroupConvo(convo.kind)) { + return { + ...convo, + kind: { + ...convo.kind, + lockStatus: 'unlocked', + // An unlocked convo cannot be moderation-locked. + lockStatusModerationOverride: false, + }, + rev: log.rev, + } + } + return {...convo, rev: log.rev} + }), + ) + } else if (ChatBskyConvoDefs.isLogLockConvoPermanently(log)) { + mutateConvoView( + log.convoId, + withRevGuard(log.rev, convo => { + if (ChatBskyConvoDefs.isGroupConvo(convo.kind)) { + return { + ...convo, + kind: {...convo.kind, lockStatus: 'locked-permanently'}, + rev: log.rev, + } + } + return {...convo, rev: log.rev} + }), + ) + } else if ( + ChatBskyConvoDefs.isLogCreateJoinLink(log) || + ChatBskyConvoDefs.isLogEditJoinLink(log) || + ChatBskyConvoDefs.isLogEnableJoinLink(log) || + ChatBskyConvoDefs.isLogDisableJoinLink(log) + ) { + // Join link data not included in the log event, trigger refetch to get it + onRefetchNeeded() + } else if (ChatBskyConvoDefs.isLogEditGroup(log)) { + // Updated group details (name etc.) aren't included in the log + // event, so refetch to pick them up. + onRefetchNeeded() + } else if ( + ChatBskyConvoDefs.isLogApproveJoinRequest(log) || + ChatBskyConvoDefs.isLogRejectJoinRequest(log) + ) { + // Route through mutateConvoView (not updateConvoInAllLists) so the + // single-convo cache updates too, keeping the in-convo requests + // banner in sync. + mutateConvoView( + 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, + 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, + withRevGuard(log.rev, convo => { + if (!ChatBskyConvoDefs.isGroupConvo(convo.kind)) { + return {...convo, rev: log.rev} + } + return { + ...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. + onRefetchNeeded() + } 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, + withRevGuard(log.rev, convo => + applyJoinRequestCountDelta(convo, log.rev, -1), + ), + ) + } else if (ChatBskyConvoDefs.isLogWithdrawOutgoingJoinRequest(log)) { + // The viewer rescinded their own outgoing join request (possibly on + // another device). Remove it from the requests inbox cache. + queryClient.setQueriesData( + {queryKey: [REQUESTS_RQKEY_ROOT]}, + old => optimisticDeleteJoinRequest(log.convoId, old), + ) + } else if (ChatBskyConvoDefs.isLogAddReaction(log)) { + updateConvoInAllLists( + log.convoId, + withRevGuard(log.rev, convo => { + // add relatedProfiles to members list, but making sure to dedupe + const relatedProfilesSansMembers = (log.relatedProfiles ?? []).filter( + profile => + !convo.members.some(member => member.did === profile.did), + ) + return { + ...convo, + members: [...convo.members, ...relatedProfilesSansMembers], + lastReaction: { + $type: 'chat.bsky.convo.defs#messageAndReactionView', + reaction: log.reaction, + message: log.message, + }, + rev: log.rev, + } + }), + ) + } else if (ChatBskyConvoDefs.isLogAddMember(log)) { + const data = log.message.data + if ( + bsky.dangerousIsType( + data, + ChatBskyConvoDefs.isSystemMessageDataAddMember, + ) + ) { + handleMemberAdded( + log.convoId, + data.member.did, + log.relatedProfiles, + log.rev, + ) + } + // Refetch so the server can refresh the curated members list. + void queryClient.invalidateQueries({ + queryKey: CONVO_KEY(log.convoId), + }) + onRefetchNeeded() + } else if (ChatBskyConvoDefs.isLogRemoveMember(log)) { + const data = log.message.data + if ( + bsky.dangerousIsType( + data, + ChatBskyConvoDefs.isSystemMessageDataRemoveMember, + ) + ) { + handleMemberRemoved(log.convoId, data.member.did, log.rev) + } + // Refetch so the server can refill the curated members list. + void queryClient.invalidateQueries({ + queryKey: CONVO_KEY(log.convoId), + }) + onRefetchNeeded() + } else if (ChatBskyConvoDefs.isLogMemberJoin(log)) { + const data = log.message.data + if ( + bsky.dangerousIsType( + data, + ChatBskyConvoDefs.isSystemMessageDataMemberJoin, + ) + ) { + handleMemberAdded( + log.convoId, + data.member.did, + log.relatedProfiles, + log.rev, + ) + } + void queryClient.invalidateQueries({ + queryKey: CONVO_KEY(log.convoId), + }) + onRefetchNeeded() + } else if (ChatBskyConvoDefs.isLogMemberLeave(log)) { + const data = log.message.data + if ( + bsky.dangerousIsType( + data, + ChatBskyConvoDefs.isSystemMessageDataMemberLeave, + ) + ) { + handleMemberRemoved(log.convoId, data.member.did, log.rev) + } + void queryClient.invalidateQueries({ + queryKey: CONVO_KEY(log.convoId), + }) + onRefetchNeeded() + } else if (ChatBskyConvoDefs.isLogRemoveReaction(log)) { + queryClient.setQueriesData( + {queryKey: [RQKEY_ROOT]}, + (old?: ConvoListQueryData) => + optimisticUpdate( + log.convoId, + old, + withRevGuard(log.rev, convo => { + if ( + // if the convo is the same + log.convoId === convo.id && + ChatBskyConvoDefs.isMessageAndReactionView( + convo.lastReaction, + ) && + ChatBskyConvoDefs.isMessageView(log.message) && + // ...and the message is the same + convo.lastReaction.message.id === log.message.id && + // ...and the reaction is the same + convo.lastReaction.reaction.sender.did === + log.reaction.sender.did && + convo.lastReaction.reaction.value === log.reaction.value + ) { + return { + ...convo, + // ...remove the reaction. hopefully they didn't react twice in a row! + lastReaction: undefined, + rev: log.rev, + } + } else { + return convo + } + }), + ), + ) + } + } +} + export function useUnreadMessageCount() { const {currentConvoId} = useCurrentConvoId() const {currentAccount} = useSession()