From 27feda0445df33cb1ae23a148013ee4b79ea2d12 Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?Bj=C3=B8rge=20N=C3=A6ss?= <bjoerge@gmail.com>
Date: Wed, 8 Jan 2025 12:16:47 +0100
Subject: [PATCH] fix(sanity): fix race condition introduced by #8120 (#8211)

---
 .../document/document-pair/checkoutPair.ts    |  4 ----
 .../document/document-pair/memoizedPair.ts    | 19 ++++++++++++-------
 2 files changed, 12 insertions(+), 11 deletions(-)

diff --git a/packages/sanity/src/core/store/_legacy/document/document-pair/checkoutPair.ts b/packages/sanity/src/core/store/_legacy/document/document-pair/checkoutPair.ts
index 34a42be66b2..19e9975cfe8 100644
--- a/packages/sanity/src/core/store/_legacy/document/document-pair/checkoutPair.ts
+++ b/packages/sanity/src/core/store/_legacy/document/document-pair/checkoutPair.ts
@@ -65,7 +65,6 @@ export interface Pair {
   transactionsPendingEvents$: Observable<PendingMutationsEvent>
   published: DocumentVersion
   draft: DocumentVersion
-  _keepalive: Observable<never>
 }
 
 function setVersion<T>(version: 'draft' | 'published') {
@@ -252,8 +251,5 @@ export function checkoutPair(
       consistency$: published.consistency$,
       remoteSnapshot$: published.remoteSnapshot$.pipe(map(setVersion('published'))),
     },
-    // Use this to keep the mutation pipeline active.
-    // It won't ever emit any events, but it will prevent the eventsource connection from completing for as long as it is subscribed to
-    _keepalive: merge(listenerEvents$, commits$).pipe(mergeMap(() => EMPTY)),
   }
 }
diff --git a/packages/sanity/src/core/store/_legacy/document/document-pair/memoizedPair.ts b/packages/sanity/src/core/store/_legacy/document/document-pair/memoizedPair.ts
index f2d6dd1637e..dfeb7cf8d25 100644
--- a/packages/sanity/src/core/store/_legacy/document/document-pair/memoizedPair.ts
+++ b/packages/sanity/src/core/store/_legacy/document/document-pair/memoizedPair.ts
@@ -1,5 +1,6 @@
 import {type SanityClient} from '@sanity/client'
-import {merge, type Observable, of, ReplaySubject, share, timer} from 'rxjs'
+import {EMPTY, merge, Observable, of, ReplaySubject, share, timer} from 'rxjs'
+import {mergeMap} from 'rxjs/operators'
 
 import {type PairListenerOptions} from '../getPairListener'
 import {type IdPair} from '../types'
@@ -24,12 +25,16 @@ export const memoizedPair: (
     serverActionsEnabled: Observable<boolean>,
     pairListenerOptions?: PairListenerOptions,
   ): Observable<Pair> => {
-    const pair = checkoutPair(client, idPair, serverActionsEnabled, pairListenerOptions)
-    return merge(
-      of(pair),
-      // makes sure the pair listener is kept alive for as long as there are subscribers
-      pair._keepalive,
-    ).pipe(
+    return new Observable<Pair>((subscriber) => {
+      const pair = checkoutPair(client, idPair, serverActionsEnabled, pairListenerOptions)
+      return merge(
+        of(pair),
+        // merge in draft events and published events to makes sure they receive
+        // the events they need for as long as the pair is subscribed to
+        pair.draft.events.pipe(mergeMap(() => EMPTY)),
+        pair.published.events.pipe(mergeMap(() => EMPTY)),
+      ).subscribe(subscriber)
+    }).pipe(
       share({
         connector: () => new ReplaySubject(1),
         resetOnComplete: true,