[Chat] Sync convo agent with shadow cache (#10827)

Co-authored-by: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Co-authored-by: DS Boyce <260543580+ds-boyce@users.noreply.github.com>
Co-authored-by: Samuel Newman <mozzius@protonmail.com>

src/state/cache/profile-shadow.ts

@@ -53,6 +53,21 @@ const shadows: WeakMap<
 > = new WeakMap()
 const emitter = new EventEmitter()
 
+type ShadowUpdateEventPayload = {did: string; shadow: Partial<ProfileShadow>}
+
+/**
+ * Subscribe to all profile shadow updates, regardless of did. Useful for
+ * non-React consumers like the Convo agent. Returns an unlisten function.
+ */
+export function listenProfileShadowUpdate(
+  listener: (payload: ShadowUpdateEventPayload) => void,
+): () => void {
+  emitter.addListener('shadow-update', listener)
+  return () => {
+    emitter.removeListener('shadow-update', listener)
+  }
+}
+
 export function useProfileShadow<
   TProfileView extends bsky.profile.AnyProfileView,
 >(profile: TProfileView): Shadow<TProfileView> {
@@ -206,10 +221,47 @@ export function updateProfileShadow(
   }
   batchedUpdates(() => {
     emitter.emit(did, value)
+    emitter.emit('shadow-update', {
+      did,
+      shadow: value,
+    } satisfies ShadowUpdateEventPayload)
   })
 }
 
-function mergeShadow<TProfileView extends bsky.profile.AnyProfileView>(
+/**
+ * Returns true if merging `shadow` into `profile` would change nothing, i.e.
+ * `mergeShadow` would be a no-op. Object-valued fields are compared by
+ * reference, so this can return false negatives - callers may do redundant
+ * merges, but never skip a real change.
+ */
+export function isProfileShadowApplied<
+  TProfileView extends bsky.profile.AnyProfileView,
+>(profile: TProfileView, shadow: Partial<ProfileShadow>): boolean {
+  if ('followingUri' in shadow) {
+    if (profile.viewer?.following !== shadow.followingUri) return false
+  }
+  if ('muted' in shadow) {
+    if (profile.viewer?.muted !== shadow.muted) return false
+  }
+  if ('blockingUri' in shadow) {
+    if (profile.viewer?.blocking !== shadow.blockingUri) return false
+  }
+  if ('activitySubscription' in shadow) {
+    if (profile.viewer?.activitySubscription !== shadow.activitySubscription) {
+      return false
+    }
+  }
+  if ('verification' in shadow) {
+    if (profile.verification !== shadow.verification) return false
+  }
+  if ('status' in shadow) {
+    const current = 'status' in profile ? profile.status : undefined
+    if (current !== shadow.status) return false
+  }
+  return true
+}
+
+export function mergeShadow<TProfileView extends bsky.profile.AnyProfileView>(
   profile: TProfileView,
   shadow: Partial<ProfileShadow>,
 ): Shadow<TProfileView> {

src/state/messages/convo/agent.ts

@@ -20,6 +20,12 @@ import {
   isNetworkError,
 } from '#/lib/strings/errors'
 import {Logger} from '#/logger'
+import {
+  isProfileShadowApplied,
+  listenProfileShadowUpdate,
+  mergeShadow,
+  type ProfileShadow,
+} from '#/state/cache/profile-shadow'
 import {
   ACTIVE_POLL_INTERVAL,
   BACKGROUND_POLL_INTERVAL,
@@ -118,6 +124,15 @@ export class Convo {
   private deletedMessages: Set<string> = new Set()
   private relatedProfiles: Map<string, ChatBskyActorDefs.ProfileViewBasic> =
     new Map()
+  /**
+   * Accumulated profile shadow state, keyed by did. The profiles this agent
+   * holds come from direct service fetches, so they are invisible to the
+   * shadow cache's react-query scan. We keep our own overlay and re-apply it
+   * whenever server data overwrites `relatedProfiles` or `convo.members`,
+   * otherwise refreshes would revert optimistic state (e.g. a block) until
+   * the server catches up.
+   */
+  private profileShadows: Map<string, Partial<ProfileShadow>> = new Map()
 
   private isProcessingPendingMessages = false
 
@@ -168,16 +183,29 @@ export class Convo {
   private subscribers: (() => void)[] = []
 
   subscribe(subscriber: () => void) {
-    if (this.subscribers.length === 0) this.init()
+    if (this.subscribers.length === 0) {
+      this.cleanupProfileShadowListener = listenProfileShadowUpdate(
+        ({did, shadow}) => {
+          this.mergeProfileShadow(did, shadow)
+        },
+      )
+      this.init()
+    }
 
     this.subscribers.push(subscriber)
 
     return () => {
       this.subscribers = this.subscribers.filter(s => s !== subscriber)
-      if (this.subscribers.length === 0) this.suspend()
+      if (this.subscribers.length === 0) {
+        this.cleanupProfileShadowListener?.()
+        this.cleanupProfileShadowListener = undefined
+        this.suspend()
+      }
     }
   }
 
+  private cleanupProfileShadowListener: (() => void) | undefined
+
   getSnapshot(): ConvoState {
     if (!this.snapshot) this.snapshot = this.generateSnapshot()
     // logger.debug('snapshotted', {})
@@ -496,6 +524,9 @@ export class Convo {
     this.pendingMessages = new Map()
     this.deletedMessages = new Set()
     this.relatedProfiles = new Map()
+    // Shadow updates fired while suspended are missed, so the overlay may be
+    // stale - drop it and trust the from-scratch refetch.
+    this.profileShadows = new Map()
 
     this.pendingMessageFailure = null
     this.fetchMessageHistoryError = undefined
@@ -527,6 +558,7 @@ export class Convo {
         this.relatedProfiles.set(member.did, member)
       }
     }
+    this.applyProfileShadows()
   }
 
   private updateConvo(convo: Partial<ChatBskyConvoDefs.ConvoView>) {
@@ -537,6 +569,7 @@ export class Convo {
       for (const member of this.convo.members) {
         this.relatedProfiles.set(member.did, member)
       }
+      this.applyProfileShadows()
     }
   }
 
@@ -707,6 +740,7 @@ export class Convo {
         this.relatedProfiles.set(member.did, member)
       }
     } while (cursor)
+    this.applyProfileShadows()
   }
 
   private fetchMessageHistoryError: {retry: () => void} | undefined
@@ -753,6 +787,7 @@ export class Convo {
         for (const profile of relatedProfiles) {
           this.relatedProfiles.set(profile.did, profile)
         }
+        this.applyProfileShadows()
       }
 
       /*
@@ -875,6 +910,7 @@ export class Convo {
             for (const profile of ev.relatedProfiles) {
               this.relatedProfiles.set(profile.did, profile)
             }
+            this.applyProfileShadows()
           }
 
           if (
@@ -1485,4 +1521,86 @@ export class Convo {
       throw error
     }
   }
+
+  mergeProfileShadow(did: string, shadow: Partial<ProfileShadow>) {
+    // Accumulate even if the did isn't held yet - the profile may arrive
+    // later via message history or the member list, and must get the shadow.
+    this.profileShadows.set(did, {
+      ...this.profileShadows.get(did),
+      ...shadow,
+    })
+    if (this.applyProfileShadow(did, shadow)) {
+      this.commit()
+    }
+  }
+
+  /**
+   * Re-applies all accumulated shadows. Must be called after any server data
+   * lands in `relatedProfiles` or `this.convo`, since raw server profiles
+   * would otherwise clobber optimistic state.
+   */
+  private applyProfileShadows() {
+    for (const [did, shadow] of this.profileShadows) {
+      this.applyProfileShadow(did, shadow)
+    }
+  }
+
+  private applyProfileShadow(
+    did: string,
+    shadow: Partial<ProfileShadow>,
+  ): boolean {
+    let changed = false
+
+    const related = this.relatedProfiles.get(did)
+    if (related && !isProfileShadowApplied(related, shadow)) {
+      this.relatedProfiles.set(did, mergeShadow(related, shadow))
+      changed = true
+    }
+
+    if (this.convo) {
+      const next = applyShadowToConvo(this.convo, did, shadow)
+      if (next) {
+        this.convo = next
+        changed = true
+      }
+    }
+
+    return changed
+  }
+}
+
+/**
+ * Returns a new convo with the shadow merged into the matching member (and
+ * `primaryMember`, if it's the same profile), or null if nothing changed.
+ */
+function applyShadowToConvo(
+  convo: ConvoWithDetails,
+  did: string,
+  shadow: Partial<ProfileShadow>,
+): ConvoWithDetails | null {
+  const i = convo.members.findIndex(m => m.did === did)
+  if (i === -1) return null
+  if (isProfileShadowApplied(convo.members[i], shadow)) return null
+
+  // The branches are identical, but narrowing the union is what lets the
+  // member arrays keep their per-kind types.
+  if (convo.kind === 'group') {
+    const members = convo.members.slice()
+    members[i] = mergeShadow(members[i], shadow)
+    return {
+      ...convo,
+      members,
+      primaryMember:
+        convo.primaryMember?.did === did ? members[i] : convo.primaryMember,
+    }
+  } else {
+    const members = convo.members.slice()
+    members[i] = mergeShadow(members[i], shadow)
+    return {
+      ...convo,
+      members,
+      primaryMember:
+        convo.primaryMember.did === did ? members[i] : convo.primaryMember,
+    }
+  }
 }

src/state/queries/messages/get-status.ts

@@ -12,6 +12,8 @@ export function useChatActorStatusQuery() {
   const agent = useAgent()
 
   return useQuery({
+    gcTime: STALE.INFINITY,
+    staleTime: STALE.SECONDS.FIFTEEN,
     queryKey: chatActorStatusQueryKey(),
     queryFn: async () => {
       const {data} = await agent.chat.bsky.actor.getStatus(
@@ -21,7 +23,5 @@ export function useChatActorStatusQuery() {
 
       return data
     },
-    staleTime: STALE.INFINITY,
-    gcTime: STALE.INFINITY,
   })
 }

src/state/queries/profile.ts

@@ -528,6 +528,10 @@ export function useProfileBlockMutationQueue(
       updateProfileShadow(queryClient, did, {
         blockingUri: finalBlockingUri,
       })
+      // The shadow only reaches components that read profiles through shadow
+      // hooks. The convo list is also read raw (e.g. the unread badge's
+      // calculateCount, getMessageInfo), and blocks emit no chat log event,
+      // so without a refetch that data stays stale indefinitely.
       void queryClient.invalidateQueries({queryKey: [RQKEY_LIST_CONVOS]})
     },
   })