fix optimistic like uri race

Fixes APP-T47N

src/state/queries/post.test.tsx

@@ -0,0 +1,317 @@
+import {deleteLike, deleteRepost, like, repost} from '@bsky/sdk'
+import {useMutation, useQueryClient} from '@tanstack/react-query'
+import {act, renderHook} from '@testing-library/react-native'
+
+import {usePdsClient, useSession} from '#/state/session'
+import {usePostLikeMutationQueue, usePostRepostMutationQueue} from './post'
+
+jest.mock('@tanstack/react-query', () => ({
+  useMutation: jest.fn(),
+  useQuery: jest.fn(),
+  useQueryClient: jest.fn(),
+}))
+
+jest.mock('#/analytics', () => ({
+  useAnalytics: jest.fn(() => ({metric: jest.fn()})),
+}))
+
+jest.mock('#/state/cache/post-shadow', () => ({
+  updatePostShadow: jest.fn(),
+}))
+
+jest.mock('#/state/session', () => ({
+  useAppviewClient: jest.fn(),
+  usePdsClient: jest.fn(),
+  useSession: jest.fn(() => ({currentAccount: undefined})),
+}))
+
+jest.mock('#/state/userActionHistory', () => ({
+  like: jest.fn(),
+  unlike: jest.fn(),
+}))
+
+jest.mock('./profile', () => ({
+  findProfileQueryData: jest.fn(),
+}))
+
+function deferred<T>() {
+  let resolve!: (value: T) => void
+  const promise = new Promise<T>(res => {
+    resolve = res
+  })
+  return {promise, resolve}
+}
+
+function post(viewer: {like?: string; repost?: string}) {
+  return {
+    uri: 'at://did:plc:author/app.bsky.feed.post/post',
+    cid: 'post-cid',
+    author: {did: 'did:plc:author'},
+    viewer,
+  } as unknown as Parameters<typeof usePostLikeMutationQueue>[0]
+}
+
+const queryClient = {}
+const pdsClient = {call: jest.fn()}
+
+beforeEach(() => {
+  jest.clearAllMocks()
+  jest.mocked(useQueryClient).mockReturnValue(queryClient as never)
+  jest.mocked(usePdsClient).mockReturnValue(pdsClient as never)
+  jest.mocked(useSession).mockReturnValue({
+    currentAccount: {did: 'did:plc:default'},
+  } as never)
+  jest.mocked(useMutation).mockImplementation(
+    options =>
+      ({
+        mutateAsync: options.mutationFn,
+      }) as ReturnType<typeof useMutation>,
+  )
+})
+
+it('does not deduplicate likes across accounts', async () => {
+  const firstRequest = deferred<{uri: string}>()
+  const secondRequest = deferred<{uri: string}>()
+  const firstLikeUri = 'at://did:plc:first/app.bsky.feed.like/like'
+  const secondLikeUri = 'at://did:plc:second/app.bsky.feed.like/like'
+  let likeCalls = 0
+  pdsClient.call.mockImplementation(method => {
+    if (method === like) {
+      return [firstRequest.promise, secondRequest.promise][likeCalls++]
+    }
+    if (method === deleteLike) return Promise.resolve()
+    throw new Error('Unexpected PDS call')
+  })
+
+  jest.mocked(useSession).mockReturnValue({
+    currentAccount: {did: 'did:plc:first'},
+  } as never)
+  const first = renderHook(() =>
+    usePostLikeMutationQueue(post({}), undefined, undefined, 'ImmersiveVideo'),
+  )
+  let firstPromise!: Promise<string | undefined>
+  act(() => {
+    firstPromise = first.result.current[0]()
+  })
+
+  jest.mocked(useSession).mockReturnValue({
+    currentAccount: {did: 'did:plc:second'},
+  } as never)
+  const second = renderHook(() =>
+    usePostLikeMutationQueue(post({}), undefined, undefined, 'ImmersiveVideo'),
+  )
+  let secondPromise!: Promise<string | undefined>
+  act(() => {
+    secondPromise = second.result.current[0]()
+  })
+
+  expect(pdsClient.call).toHaveBeenCalledTimes(2)
+  await act(async () => {
+    firstRequest.resolve({uri: firstLikeUri})
+    secondRequest.resolve({uri: secondLikeUri})
+    await Promise.all([firstPromise, secondPromise])
+  })
+  expect(likeCalls).toBe(2)
+})
+
+it('does not remove a newer like URI when an older unlike completes', async () => {
+  const firstLikeRequest = deferred<{uri: string}>()
+  const secondLikeRequest = deferred<{uri: string}>()
+  const firstUnlikeRequest = deferred<void>()
+  const firstLikeUri = 'at://did:plc:default/app.bsky.feed.like/first'
+  const secondLikeUri = 'at://did:plc:default/app.bsky.feed.like/second'
+  let likeCalls = 0
+  let unlikeCalls = 0
+  pdsClient.call.mockImplementation(method => {
+    if (method === like) {
+      return [firstLikeRequest.promise, secondLikeRequest.promise][likeCalls++]
+    }
+    if (method === deleteLike) {
+      unlikeCalls++
+      return unlikeCalls === 1 ? firstUnlikeRequest.promise : Promise.resolve()
+    }
+    throw new Error('Unexpected PDS call')
+  })
+
+  const owner = renderHook(
+    ({viewerLike}: {viewerLike?: string}) =>
+      usePostLikeMutationQueue(
+        post({like: viewerLike}),
+        undefined,
+        undefined,
+        'ImmersiveVideo',
+      ),
+    {initialProps: {viewerLike: undefined as string | undefined}},
+  )
+  let firstLikePromise!: Promise<string | undefined>
+  act(() => {
+    firstLikePromise = owner.result.current[0]()
+  })
+  await act(async () => {
+    firstLikeRequest.resolve({uri: firstLikeUri})
+    await firstLikePromise
+  })
+  owner.rerender({viewerLike: firstLikeUri})
+
+  let firstUnlikePromise!: Promise<string | undefined>
+  act(() => {
+    firstUnlikePromise = owner.result.current[1]()
+  })
+
+  const peer = renderHook(() =>
+    usePostLikeMutationQueue(post({}), undefined, undefined, 'ImmersiveVideo'),
+  )
+  let secondLikePromise!: Promise<string | undefined>
+  act(() => {
+    secondLikePromise = peer.result.current[0]()
+  })
+  const stale = renderHook(() =>
+    usePostLikeMutationQueue(
+      post({like: 'pending'}),
+      undefined,
+      undefined,
+      'ImmersiveVideo',
+    ),
+  )
+  let secondUnlikePromise!: Promise<string | undefined>
+  act(() => {
+    secondUnlikePromise = stale.result.current[1]()
+  })
+
+  await act(async () => {
+    firstUnlikeRequest.resolve()
+    await firstUnlikePromise
+  })
+  await act(async () => {
+    secondLikeRequest.resolve({uri: secondLikeUri})
+    await Promise.all([secondLikePromise, secondUnlikePromise])
+  })
+
+  expect(pdsClient.call).toHaveBeenLastCalledWith(deleteLike, secondLikeUri)
+})
+
+it('deduplicates likes and shares the URI with another queue instance', async () => {
+  const likeRequest = deferred<{uri: string}>()
+  const likeUri = 'at://did:plc:me/app.bsky.feed.like/like'
+  pdsClient.call.mockImplementation(method => {
+    if (method === like) return likeRequest.promise
+    if (method === deleteLike) return Promise.resolve()
+    throw new Error('Unexpected PDS call')
+  })
+
+  const owner = renderHook(
+    ({viewerLike}: {viewerLike?: string}) =>
+      usePostLikeMutationQueue(
+        post({like: viewerLike}),
+        undefined,
+        undefined,
+        'ImmersiveVideo',
+      ),
+    {initialProps: {viewerLike: undefined as string | undefined}},
+  )
+  const peer = renderHook(() =>
+    usePostLikeMutationQueue(post({}), undefined, undefined, 'ImmersiveVideo'),
+  )
+
+  let likePromise!: Promise<string | undefined>
+  let peerLikePromise!: Promise<string | undefined>
+  act(() => {
+    likePromise = owner.result.current[0]()
+    peerLikePromise = peer.result.current[0]()
+  })
+  expect(pdsClient.call).toHaveBeenCalledTimes(1)
+  owner.rerender({viewerLike: 'pending'})
+  const other = renderHook(() =>
+    usePostLikeMutationQueue(
+      post({like: 'pending'}),
+      undefined,
+      undefined,
+      'ImmersiveVideo',
+    ),
+  )
+  await act(async () => {
+    likeRequest.resolve({uri: likeUri})
+    await Promise.all([likePromise, peerLikePromise])
+  })
+  const latePeer = renderHook(() =>
+    usePostLikeMutationQueue(post({}), undefined, undefined, 'ImmersiveVideo'),
+  )
+  await act(async () => {
+    await latePeer.result.current[0]()
+  })
+  expect(pdsClient.call).toHaveBeenCalledTimes(1)
+
+  await act(async () => {
+    await other.result.current[1]()
+  })
+
+  expect(pdsClient.call).toHaveBeenLastCalledWith(deleteLike, likeUri)
+})
+
+it('deduplicates reposts and shares the URI with another queue instance', async () => {
+  const repostRequest = deferred<{uri: string}>()
+  const repostUri = 'at://did:plc:me/app.bsky.feed.repost/repost'
+  pdsClient.call.mockImplementation(method => {
+    if (method === repost) return repostRequest.promise
+    if (method === deleteRepost) return Promise.resolve()
+    throw new Error('Unexpected PDS call')
+  })
+
+  const owner = renderHook(
+    ({viewerRepost}: {viewerRepost?: string}) =>
+      usePostRepostMutationQueue(
+        post({repost: viewerRepost}),
+        undefined,
+        undefined,
+        'ImmersiveVideo',
+      ),
+    {initialProps: {viewerRepost: undefined as string | undefined}},
+  )
+  const peer = renderHook(() =>
+    usePostRepostMutationQueue(
+      post({}),
+      undefined,
+      undefined,
+      'ImmersiveVideo',
+    ),
+  )
+
+  let repostPromise!: Promise<string | undefined>
+  let peerRepostPromise!: Promise<string | undefined>
+  act(() => {
+    repostPromise = owner.result.current[0]()
+    peerRepostPromise = peer.result.current[0]()
+  })
+  expect(pdsClient.call).toHaveBeenCalledTimes(1)
+  owner.rerender({viewerRepost: 'pending'})
+  const other = renderHook(() =>
+    usePostRepostMutationQueue(
+      post({repost: 'pending'}),
+      undefined,
+      undefined,
+      'ImmersiveVideo',
+    ),
+  )
+  await act(async () => {
+    repostRequest.resolve({uri: repostUri})
+    await Promise.all([repostPromise, peerRepostPromise])
+  })
+  const latePeer = renderHook(() =>
+    usePostRepostMutationQueue(
+      post({}),
+      undefined,
+      undefined,
+      'ImmersiveVideo',
+    ),
+  )
+  await act(async () => {
+    await latePeer.result.current[0]()
+  })
+  expect(pdsClient.call).toHaveBeenCalledTimes(1)
+
+  await act(async () => {
+    await other.result.current[1]()
+  })
+
+  expect(pdsClient.call).toHaveBeenLastCalledWith(deleteRepost, repostUri)
+})

src/state/queries/post.ts

@@ -21,8 +21,73 @@ import {useIsThreadMuted, useSetThreadMute} from '../cache/thread-mutes'
 import {findProfileQueryData} from './profile'
 
 const RQKEY_ROOT = 'post'
+const PENDING_URI = 'pending'
+type CreatedUri = {
+  promise: Promise<AtUriString>
+  isPending: boolean
+  isDeleting: boolean
+  uri?: AtUriString
+}
+const createdLikeUris = new Map<string, CreatedUri>()
+const createdRepostUris = new Map<string, CreatedUri>()
 export const RQKEY = (postUri: string) => [RQKEY_ROOT, postUri]
 
+function createdUriKey(accountDid: string | undefined, postUri: string) {
+  return `${accountDid ?? ''}\0${postUri}`
+}
+
+function getOrCreateUri(
+  records: Map<string, CreatedUri>,
+  postUri: string,
+  create: () => Promise<AtUriString>,
+) {
+  const existing = records.get(postUri)
+  if (existing && !existing.isDeleting) {
+    return {recordUri: existing.promise, created: false}
+  }
+
+  const recordUri = create()
+  const entry: CreatedUri = {
+    promise: recordUri,
+    isPending: true,
+    isDeleting: false,
+  }
+  records.set(postUri, entry)
+  void recordUri.then(
+    uri => {
+      if (records.get(postUri) === entry) {
+        entry.isPending = false
+        entry.uri = uri
+      }
+    },
+    () => {
+      if (records.get(postUri) === entry) records.delete(postUri)
+    },
+  )
+  return {recordUri, created: true}
+}
+
+function getCreatedUri(records: Map<string, CreatedUri>, postUri: string) {
+  const entry = records.get(postUri)
+  if (!entry) {
+    const err = new Error('The record URI is not available yet')
+    err.name = 'AbortError'
+    throw err
+  }
+  return entry.promise
+}
+
+function removeCreatedUri(
+  records: Map<string, CreatedUri>,
+  postUri: string,
+  entry: CreatedUri | undefined,
+  uri: AtUriString,
+) {
+  if (entry && records.get(postUri) === entry && entry.uri === uri) {
+    records.delete(postUri)
+  }
+}
+
 export function usePostQuery(uri: string | undefined) {
   const client = useAppviewClient()
   return useQuery<app.bsky.feed.defs.PostView>({
@@ -122,10 +187,13 @@ export function usePostLikeMutationQueue(
   feedDescriptor: string | undefined,
   logContext: Metrics['post:like']['logContext'],
 ) {
+  const {currentAccount} = useSession()
   const queryClient = useQueryClient()
   const postUri = post.uri
+  const uriKey = createdUriKey(currentAccount?.did, postUri)
   const postCid = post.cid
-  const initialLikeUri = post.viewer?.like
+  const initialLikeUri = post.viewer?.like as
+    AtUriString | typeof PENDING_URI | undefined
   const likeMutation = usePostLikeMutation(feedDescriptor, logContext, post)
   const unlikeMutation = usePostUnlikeMutation(feedDescriptor, logContext, post)
 
@@ -133,19 +201,43 @@ export function usePostLikeMutationQueue(
     initialState: initialLikeUri,
     runMutation: async (prevLikeUri, shouldLike) => {
       if (shouldLike) {
-        const {uri: likeUri} = await likeMutation.mutateAsync({
+        const {recordUri, created} = getOrCreateUri(
-          uri: postUri,
+          createdLikeUris,
-          cid: postCid,
+          uriKey,
-          via: viaRepost,
+          () =>
-        })
+            likeMutation
-        userActionHistory.like([postUri])
+              .mutateAsync({
+                uri: postUri,
+                cid: postCid,
+                via: viaRepost,
+              })
+              .then(({uri}) => uri),
+        )
+        const likeUri = await recordUri
+        if (created) {
+          userActionHistory.like([postUri])
+        }
         return likeUri
       } else {
-        if (prevLikeUri) {
+        const createdUri = createdLikeUris.get(uriKey)
-          await unlikeMutation.mutateAsync({
+        const likeUri =
-            postUri: postUri,
+          prevLikeUri === PENDING_URI
-            likeUri: prevLikeUri,
+            ? await getCreatedUri(createdLikeUris, uriKey)
-          })
+            : prevLikeUri
+        if (likeUri) {
+          if (createdUri?.uri === likeUri) createdUri.isDeleting = true
+          try {
+            await unlikeMutation.mutateAsync({
+              postUri: postUri,
+              likeUri,
+            })
+          } catch (err) {
+            if (createdUri && createdLikeUris.get(uriKey) === createdUri) {
+              createdUri.isDeleting = false
+            }
+            throw err
+          }
+          removeCreatedUri(createdLikeUris, uriKey, createdUri, likeUri)
           userActionHistory.unlike([postUri])
         }
         return undefined
@@ -162,7 +254,7 @@ export function usePostLikeMutationQueue(
   const queueLike = useCallback(() => {
     // optimistically update
     updatePostShadow(queryClient, postUri, {
-      likeUri: 'pending',
+      likeUri: PENDING_URI,
     })
     return queueToggle(true)
   }, [queryClient, postUri, queueToggle])
@@ -252,10 +344,13 @@ export function usePostRepostMutationQueue(
   feedDescriptor: string | undefined,
   logContext: Metrics['post:repost']['logContext'],
 ) {
+  const {currentAccount} = useSession()
   const queryClient = useQueryClient()
   const postUri = post.uri
+  const uriKey = createdUriKey(currentAccount?.did, postUri)
   const postCid = post.cid
-  const initialRepostUri = post.viewer?.repost
+  const initialRepostUri = post.viewer?.repost as
+    AtUriString | typeof PENDING_URI | undefined
   const repostMutation = usePostRepostMutation(feedDescriptor, logContext, post)
   const unrepostMutation = usePostUnrepostMutation(
     feedDescriptor,
@@ -267,18 +362,37 @@ export function usePostRepostMutationQueue(
     initialState: initialRepostUri,
     runMutation: async (prevRepostUri, shouldRepost) => {
       if (shouldRepost) {
-        const {uri: repostUri} = await repostMutation.mutateAsync({
+        const {recordUri} = getOrCreateUri(createdRepostUris, uriKey, () =>
-          uri: postUri,
+          repostMutation
-          cid: postCid,
+            .mutateAsync({
-          via: viaRepost,
+              uri: postUri,
-        })
+              cid: postCid,
+              via: viaRepost,
+            })
+            .then(({uri}) => uri),
+        )
+        const repostUri = await recordUri
         return repostUri
       } else {
-        if (prevRepostUri) {
+        const createdUri = createdRepostUris.get(uriKey)
-          await unrepostMutation.mutateAsync({
+        const repostUri =
-            postUri: postUri,
+          prevRepostUri === PENDING_URI
-            repostUri: prevRepostUri,
+            ? await getCreatedUri(createdRepostUris, uriKey)
-          })
+            : prevRepostUri
+        if (repostUri) {
+          if (createdUri?.uri === repostUri) createdUri.isDeleting = true
+          try {
+            await unrepostMutation.mutateAsync({
+              postUri: postUri,
+              repostUri,
+            })
+          } catch (err) {
+            if (createdUri && createdRepostUris.get(uriKey) === createdUri) {
+              createdUri.isDeleting = false
+            }
+            throw err
+          }
+          removeCreatedUri(createdRepostUris, uriKey, createdUri, repostUri)
         }
         return undefined
       }
@@ -294,7 +408,7 @@ export function usePostRepostMutationQueue(
   const queueRepost = useCallback(() => {
     // optimistically update
     updatePostShadow(queryClient, postUri, {
-      repostUri: 'pending',
+      repostUri: PENDING_URI,
     })
     return queueToggle(true)
   }, [queryClient, postUri, queueToggle])