From 9bbc3befdd6ae2c6fd4a2c39f6d85e3f0cfd275d Mon Sep 17 00:00:00 2001 From: Samuel Newman Date: Tue, 4 Aug 2026 03:17:17 +0300 Subject: [PATCH] migrate the stateful chat agents to the chat client Convo and MessagesEventBus were the last two useAgent consumers holding a long-lived agent rather than reading one per render. Both now take a chat client, which carries the atproto-proxy header itself, so the per-call DM_SERVICE_HEADERS drop out under the double-set rule. Both providers gained an effect that pushes the current client into the instance. The instances are created once with useState and outlive the client they were built with: an account switch, a cross-tab token sync or an expiry rescue replaces the bundle and disposes the previous clients, so without the sync a convo would keep sending through a dead session. `handleSendMessageFailure` narrows on XrpcResponseError rather than XRPCError. The lex client throws the former, so the status branch (and the recoverable / unrecoverable split that drives the retry banner) had silently stopped firing. Co-Authored-By: Claude Fable 5 --- .../Messages/components/MessagesList.tsx | 14 +- src/state/messages/convo/agent.ts | 123 +++++++++--------- src/state/messages/convo/index.tsx | 16 ++- src/state/messages/convo/types.ts | 9 +- src/state/messages/events/agent.ts | 37 +++--- src/state/messages/events/index.tsx | 16 ++- src/state/messages/events/types.ts | 9 +- 7 files changed, 127 insertions(+), 97 deletions(-) diff --git a/src/screens/Messages/components/MessagesList.tsx b/src/screens/Messages/components/MessagesList.tsx index e9d336a67a..eab025665c 100644 --- a/src/screens/Messages/components/MessagesList.tsx +++ b/src/screens/Messages/components/MessagesList.tsx @@ -70,7 +70,7 @@ import {Loader} from '#/components/Loader' import {Text} from '#/components/Typography' import {useAnalytics} from '#/analytics' import {IS_ANDROID, IS_NATIVE, IS_WEB} from '#/env' -import {app} from '#/lexicons' +import {app, type chat, type com} from '#/lexicons' import * as bsky from '#/types/bsky' import {ChatStatusInfo} from './ChatStatusInfo' import {groupSystemMessages, type RenderItem} from './groupSystemMessages' @@ -537,10 +537,7 @@ export function MessagesList({ */ rt.detectFacetsWithoutResolution() - let embed: - | $Typed - | $Typed - | undefined + let embed: chat.bsky.convo.defs.MessageInput['embed'] let embedView: | $Typed | $Typed @@ -576,10 +573,15 @@ export function MessagesList({ if (post) { embed = { $type: 'app.bsky.embed.record', + /* + * `getPost` still returns an `@atproto/api` view, whose `uri` and + * `cid` are plain strings rather than the branded syntax types + * the lexicon input declares. + */ record: { uri: post.uri, cid: post.cid, - }, + } as com.atproto.repo.strongRef.Main, } embedView = { diff --git a/src/state/messages/convo/agent.ts b/src/state/messages/convo/agent.ts index b1dc348e79..521e7f77ea 100644 --- a/src/state/messages/convo/agent.ts +++ b/src/state/messages/convo/agent.ts @@ -1,20 +1,17 @@ import { type $Typed, type AppBskyEmbedRecord, - type AtpAgent, type ChatBskyActorDefs, ChatBskyConvoDefs, type ChatBskyConvoGetLog, - type ChatBskyConvoSendMessage, type ChatBskyEmbedJoinLink, type ChatBskyGroupDefs, } from '@atproto/api' -import {XRPCError} from '@atproto/api' +import {type Client, XrpcResponseError} from '@atproto/lex' import {EventEmitter} from 'eventemitter3' import {nanoid} from 'nanoid/non-secure' import {networkRetry} from '#/lib/async/retry' -import {DM_SERVICE_HEADERS} from '#/lib/constants' import { isErrorMaybeAppPasswordPermissions, isNetworkError, @@ -52,6 +49,7 @@ import { parseConvoView, } from '#/components/dms/util' import {IS_NATIVE} from '#/env' +import {chat} from '#/lexicons' const logger = Logger.create(Logger.Context.ConversationAgent) @@ -105,7 +103,7 @@ function toDeletedMessageView( export class Convo { private id: string - private agent: AtpAgent + private chatClient: Client private events: MessagesEventBus private senderUserDid: string @@ -131,7 +129,7 @@ export class Convo { string, { id: string - message: ChatBskyConvoSendMessage.InputSchema['message'] + message: chat.bsky.convo.defs.MessageInput optimisticEmbedView?: | $Typed | $Typed @@ -166,9 +164,9 @@ export class Convo { constructor(params: ConvoParams) { this.id = nanoid(3) this.convoId = params.convoId - this.agent = params.agent + this.chatClient = params.chatClient this.events = params.events - this.senderUserDid = params.agent.assertDid + this.senderUserDid = params.chatClient.assertDid if (params.placeholderData) { this.setupPlaceholderData(params.placeholderData) @@ -197,6 +195,15 @@ export class Convo { this.subscribers.forEach(subscriber => subscriber()) } + /** + * Point the convo at a new chat client, for when the session bundle is + * replaced underneath it. Conversation state is unaffected: the same account + * is being served over a fresh session. + */ + updateClient(chatClient: Client) { + this.chatClient = chatClient + } + private subscribers: (() => void)[] = [] subscribe(subscriber: () => void) { @@ -720,13 +727,12 @@ export class Convo { this.pendingFetchConvo = (async () => { try { const response = await networkRetry(2, () => { - return this.agent.chat.bsky.convo.getConvo( - {convoId: this.convoId}, - {headers: DM_SERVICE_HEADERS}, - ) + return this.chatClient.call(chat.bsky.convo.getConvo, { + convoId: this.convoId, + }) }) - const convo = response.data.convo + const convo = response.convo return { convo, @@ -763,18 +769,15 @@ export class Convo { let cursor: string | undefined do { const result = await networkRetry(2, () => { - return this.agent.chat.bsky.convo.getConvoMembers( - { - convoId: this.convoId, - limit: 50, - cursor, - }, - {headers: DM_SERVICE_HEADERS}, - ) + return this.chatClient.call(chat.bsky.convo.getConvoMembers, { + convoId: this.convoId, + limit: 50, + cursor, + }) }) - cursor = result.data.cursor + cursor = result.cursor - for (const member of result.data.members) { + for (const member of result.members) { this.relatedProfiles.set(member.did, member) } } while (cursor) @@ -808,16 +811,13 @@ export class Convo { const nextCursor = this.oldestRev // for TS const response = await networkRetry(2, () => { - return this.agent.chat.bsky.convo.getMessages( - { - cursor: nextCursor, - convoId: this.convoId, - limit: IS_NATIVE ? 30 : 60, - }, - {headers: DM_SERVICE_HEADERS}, - ) + return this.chatClient.call(chat.bsky.convo.getMessages, { + cursor: nextCursor, + convoId: this.convoId, + limit: IS_NATIVE ? 30 : 60, + }) }) - const {cursor, messages, relatedProfiles} = response.data + const {cursor, messages, relatedProfiles} = response // Trust the cursor for pagination. We can't infer "no more pages" from a // short page: the server pages by raw rows but strips deleted messages @@ -1031,7 +1031,7 @@ export class Convo { private pendingMessageFailure: 'recoverable' | 'unrecoverable' | null = null sendMessage( - message: ChatBskyConvoSendMessage.InputSchema['message'], + message: chat.bsky.convo.defs.MessageInput, optimisticEmbedView?: | $Typed | $Typed, @@ -1165,14 +1165,10 @@ export class Convo { const {id, message} = pendingMessage - const response = await this.agent.chat.bsky.convo.sendMessage( - { - convoId: this.convoId, - message, - }, - {encoding: 'application/json', headers: DM_SERVICE_HEADERS}, - ) - const res = response.data + const res = await this.chatClient.call(chat.bsky.convo.sendMessage, { + convoId: this.convoId, + message, + }) // remove from queue this.pendingMessages.delete(id) @@ -1197,8 +1193,15 @@ export class Convo { } } - private handleSendMessageFailure(e: Error | XRPCError) { - if (e instanceof XRPCError) { + /* + * The lex client throws `XrpcResponseError`, not `@atproto/api`'s + * `XRPCError`, so the status/message branch narrows on the lex class. Only + * a genuine server response carries a status: transport and internal lex + * failures are `XrpcInternalError`s and fall through to the generic arm, + * where `isNetworkError` keeps them out of the logs. + */ + private handleSendMessageFailure(e: Error | XrpcResponseError) { + if (e instanceof XrpcResponseError) { if (NETWORK_FAILURE_STATUSES.includes(e.status)) { this.pendingMessageFailure = 'recoverable' } else { @@ -1261,16 +1264,15 @@ export class Convo { ) try { - const {data} = await this.agent.chat.bsky.convo.sendMessageBatch( + const {items} = await this.chatClient.call( + chat.bsky.convo.sendMessageBatch, { items: messageArray.map(({message}) => ({ convoId: this.convoId, message, })), }, - {encoding: 'application/json', headers: DM_SERVICE_HEADERS}, ) - const {items} = data /* * Insert into `newMessages` as soon as we have a real ID. That way, when @@ -1304,13 +1306,10 @@ export class Convo { try { await networkRetry(2, () => { - return this.agent.chat.bsky.convo.deleteMessageForSelf( - { - convoId: this.convoId, - messageId, - }, - {encoding: 'application/json', headers: DM_SERVICE_HEADERS}, - ) + return this.chatClient.call(chat.bsky.convo.deleteMessageForSelf, { + convoId: this.convoId, + messageId, + }) }) } catch (err) { const e = err as Error @@ -1529,10 +1528,11 @@ export class Convo { try { logger.debug(`Adding reaction ${emoji} to message ${messageId}`) - const {data} = await this.agent.chat.bsky.convo.addReaction( - {messageId, value: emoji, convoId: this.convoId}, - {encoding: 'application/json', headers: DM_SERVICE_HEADERS}, - ) + const data = await this.chatClient.call(chat.bsky.convo.addReaction, { + messageId, + value: emoji, + convoId: this.convoId, + }) if (ChatBskyConvoDefs.isMessageView(data.message)) { if (this.pastMessages.has(messageId)) { this.pastMessages.set(messageId, data.message) @@ -1594,10 +1594,11 @@ export class Convo { try { logger.debug(`Removing reaction ${emoji} from message ${messageId}`) - await this.agent.chat.bsky.convo.removeReaction( - {messageId, value: emoji, convoId: this.convoId}, - {encoding: 'application/json', headers: DM_SERVICE_HEADERS}, - ) + await this.chatClient.call(chat.bsky.convo.removeReaction, { + messageId, + value: emoji, + convoId: this.convoId, + }) } catch (error) { if (restore) restore() throw error diff --git a/src/state/messages/convo/index.tsx b/src/state/messages/convo/index.tsx index 69ad1e237c..b37b060701 100644 --- a/src/state/messages/convo/index.tsx +++ b/src/state/messages/convo/index.tsx @@ -28,7 +28,7 @@ import { } from '#/state/queries/messages/conversation' import {RQKEY_ROOT as ListConvosQueryKeyRoot} from '#/state/queries/messages/list-conversations' import {RQKEY as createProfileQueryKey} from '#/state/queries/profile' -import {useAgent} from '#/state/session' +import {useChatClient} from '#/state/session' import {type GroupConvoMember} from '#/components/dms/util' export * from '#/state/messages/convo/util' @@ -80,7 +80,7 @@ export function ConvoProvider({ convoId, }: Pick & {children: React.ReactNode}) { const queryClient = useQueryClient() - const agent = useAgent() + const chatClient = useChatClient() const events = useMessagesEventBus() const [convo] = useState(() => { const placeholder = queryClient.getQueryData( @@ -88,7 +88,7 @@ export function ConvoProvider({ ) return new Convo({ convoId, - agent, + chatClient, events, placeholderData: placeholder ? {convo: placeholder} : undefined, }) @@ -96,6 +96,16 @@ export function ConvoProvider({ const service = useSyncExternalStore(convo.subscribe, convo.getSnapshot) const {mutate: markAsRead} = useMarkAsReadMutation() + /* + * The convo outlives the client it was constructed with: replacing the + * session bundle builds fresh clients over the new session and disposes the + * old ones, so a convo still holding the previous client would send through a + * dead session. + */ + useEffect(() => { + convo.updateClient(chatClient) + }, [convo, chatClient]) + const appState = useAppState() const isActive = appState === 'active' useFocusEffect( diff --git a/src/state/messages/convo/types.ts b/src/state/messages/convo/types.ts index e330d6eea3..5bc5777e82 100644 --- a/src/state/messages/convo/types.ts +++ b/src/state/messages/convo/types.ts @@ -1,19 +1,20 @@ import { type $Typed, type AppBskyEmbedRecord, - type AtpAgent, type ChatBskyActorDefs, type ChatBskyConvoDefs, - type ChatBskyConvoSendMessage, type ChatBskyEmbedJoinLink, } from '@atproto/api' +import {type Client} from '@atproto/lex' import {type MessagesEventBus} from '#/state/messages/events/agent' import {type ConvoWithDetails} from '#/components/dms/util' +import {type chat} from '#/lexicons' export type ConvoParams = { convoId: string - agent: AtpAgent + /** The chat client, which proxies `chat.bsky.*` to the chat service. */ + chatClient: Client events: MessagesEventBus placeholderData?: { convo: ChatBskyConvoDefs.ConvoView @@ -108,7 +109,7 @@ export type ConvoItem = type DeleteMessage = (messageId: string) => Promise type SendMessage = ( - message: ChatBskyConvoSendMessage.InputSchema['message'], + message: chat.bsky.convo.defs.MessageInput, optimisticEmbedView: | $Typed | $Typed diff --git a/src/state/messages/events/agent.ts b/src/state/messages/events/agent.ts index 56e6bed173..9bc7158187 100644 --- a/src/state/messages/events/agent.ts +++ b/src/state/messages/events/agent.ts @@ -1,9 +1,8 @@ -import {type AtpAgent, type ChatBskyConvoGetLog} from '@atproto/api' +import {type Client} from '@atproto/lex' import {EventEmitter} from 'eventemitter3' import {nanoid} from 'nanoid/non-secure' import {networkRetry} from '#/lib/async/retry' -import {DM_SERVICE_HEADERS} from '#/lib/constants' import { isErrorMaybeAppPasswordPermissions, isNetworkError, @@ -21,13 +20,14 @@ import { type MessagesEventBusParams, MessagesEventBusStatus, } from '#/state/messages/events/types' +import {chat} from '#/lexicons' const logger = Logger.create(Logger.Context.DMsAgent) export class MessagesEventBus { private id: string - private agent: AtpAgent + private chatClient: Client private emitter = new EventEmitter<{event: [MessagesEventBusEvent]}>() private status: MessagesEventBusStatus = MessagesEventBusStatus.Initializing @@ -37,11 +37,20 @@ export class MessagesEventBus { constructor(params: MessagesEventBusParams) { this.id = nanoid(3) - this.agent = params.agent + this.chatClient = params.chatClient this.init() } + /** + * Point the bus at a new chat client, for when the session bundle is + * replaced underneath it. The poll cursor and subscribers are unaffected: the + * same account is being served over a fresh session. + */ + updateClient(chatClient: Client) { + this.chatClient = chatClient + } + requestPollInterval(interval: number) { const id = nanoid() this.requestedPollIntervals.set(id, interval) @@ -260,14 +269,11 @@ export class MessagesEventBus { try { const response = await networkRetry(2, () => { - return this.agent.chat.bsky.convo.getLog( - {}, - {headers: DM_SERVICE_HEADERS}, - ) + return this.chatClient.call(chat.bsky.convo.getLog, {}) }) // throw new Error('UNCOMMENT TO TEST INIT FAILURE') - const {cursor} = response.data + const {cursor} = response // should always be defined if (cursor) { @@ -355,21 +361,18 @@ export class MessagesEventBus { // ) let needsEmit = false - let batch: ChatBskyConvoGetLog.OutputSchema['logs'] = [] + let batch: chat.bsky.convo.getLog.$OutputBody['logs'] = [] try { const response = await networkRetry(2, () => { - return this.agent.chat.bsky.convo.getLog( - { - cursor: this.latestRev, - }, - {headers: DM_SERVICE_HEADERS}, - ) + return this.chatClient.call(chat.bsky.convo.getLog, { + cursor: this.latestRev, + }) }) // throw new Error('UNCOMMENT TO TEST POLL FAILURE') - const {logs: events} = response.data + const {logs: events} = response for (const ev of events) { /* diff --git a/src/state/messages/events/index.tsx b/src/state/messages/events/index.tsx index 5d1e2ec505..d6e0abe600 100644 --- a/src/state/messages/events/index.tsx +++ b/src/state/messages/events/index.tsx @@ -2,7 +2,7 @@ import {createContext, useContext, useEffect, useState} from 'react' import {AppState} from 'react-native' import {MessagesEventBus} from '#/state/messages/events/agent' -import {useAgent, useSession} from '#/state/session' +import {useChatClient, useSession} from '#/state/session' const MessagesEventBusContext = createContext(null) MessagesEventBusContext.displayName = 'MessagesEventBusContext' @@ -42,14 +42,24 @@ export function MessagesEventBusProviderInner({ }: { children: React.ReactNode }) { - const agent = useAgent() + const chatClient = useChatClient() const [bus] = useState( () => new MessagesEventBus({ - agent, + chatClient, }), ) + /* + * The bus outlives the client it was constructed with: replacing the session + * bundle (account switch, cross-tab token sync, expiry rescue) builds fresh + * clients over the new session and disposes the old ones, so a bus still + * holding the previous client would poll through a dead session. + */ + useEffect(() => { + bus.updateClient(chatClient) + }, [bus, chatClient]) + useEffect(() => { bus.resume() diff --git a/src/state/messages/events/types.ts b/src/state/messages/events/types.ts index 67d6bdd452..2a743a07a2 100644 --- a/src/state/messages/events/types.ts +++ b/src/state/messages/events/types.ts @@ -1,7 +1,10 @@ -import {type AtpAgent, type ChatBskyConvoGetLog} from '@atproto/api' +import {type Client} from '@atproto/lex' + +import {type chat} from '#/lexicons' export type MessagesEventBusParams = { - agent: AtpAgent + /** The chat client, which proxies `chat.bsky.*` to the chat service. */ + chatClient: Client } export enum MessagesEventBusStatus { @@ -64,5 +67,5 @@ export type MessagesEventBusEvent = } | { type: 'logs' - logs: ChatBskyConvoGetLog.OutputSchema['logs'] + logs: chat.bsky.convo.getLog.$OutputBody['logs'] }