fix(mobile): share pull request refresh subscriptions

packages/client-runtime/src/state/pullRequests.test.ts

@@ -1176,6 +1176,59 @@ it.effect("keeps hover previews fresh after edits and turns", () =>
   ),
 );
 
+it.effect("shares one refresh stream between independently created pull request readers", () =>
+  Effect.scoped(
+    Effect.gen(function* () {
+      const revision = yield* SubscriptionRef.make(0);
+      let subscriptions = 0;
+      let title = "Original title";
+      const client = {
+        [WS_METHODS.pullRequestsSubscribeRefreshes]: () =>
+          Stream.unwrap(
+            Effect.sync(() => {
+              subscriptions++;
+              return SubscriptionRef.changes(revision);
+            }),
+          ),
+        [WS_METHODS.pullRequestsSummary]: () => Effect.sync(() => ({ title, state: "open" })),
+        [WS_METHODS.pullRequestsDetail]: () => Effect.sync(() => ({ title, state: "open" })),
+      } as unknown as WsRpcProtocolClient;
+      const { runtime, atoms, registry } = yield* makeTestRuntime(client);
+      const target = {
+        environmentId: TARGET.environmentId,
+        input: { projectId: ProjectId.make("project-1"), repository: "acme/web", number: 1 },
+      };
+      const summary = createLinkedPullRequestSummaryAtomFamily(runtime)(target);
+      const detail = atoms.detail(target);
+      for (const atom of [summary, detail]) {
+        const unmount = registry.mount(atom);
+        yield* Effect.addFinalizer(() => Effect.sync(unmount));
+        expect((yield* AtomRegistry.getResult(registry, atom)).title).toBe(title);
+      }
+      yield* AtomRegistry.getResult(
+        registry,
+        atoms.refreshes({ environmentId: target.environmentId, input: {} }),
+      );
+      expect(subscriptions).toBe(1);
+
+      const changed = [summary, detail].map((atom) => {
+        const ready = Latch.makeUnsafe();
+        const stop = registry.subscribe(atom, (result) => {
+          if (AsyncResult.isSuccess(result) && result.value.title === "Updated title") {
+            ready.openUnsafe();
+          }
+        });
+        return { ready, stop };
+      });
+      yield* Effect.addFinalizer(() => Effect.sync(() => changed.forEach(({ stop }) => stop())));
+      title = "Updated title";
+      yield* SubscriptionRef.update(revision, (value) => value + 1);
+      for (const { ready } of changed) yield* ready.await;
+      expect(subscriptions).toBe(1);
+    }),
+  ),
+);
+
 it.effect("shares close, reopen, and merge with an untouched client's mounted PR readers", () =>
   Effect.scoped(
     Effect.gen(function* () {

packages/client-runtime/src/state/pullRequests.ts

@@ -106,7 +106,7 @@ function updateCached<A, E>(
   registry.update(atom, AsyncResult.map(update));
 }
 
-function createPullRequestRefreshAtomFamily<R, E>(
+function makePullRequestRefreshAtomFamily<R, E>(
   runtime: Atom.AtomRuntime<EnvironmentRegistry | R, E>,
 ) {
   return createEnvironmentRpcSubscriptionAtomFamily(runtime, {
@@ -115,6 +115,21 @@ function createPullRequestRefreshAtomFamily<R, E>(
   });
 }
 
+const refreshFamiliesByRuntime = new WeakMap<
+  object,
+  ReturnType<typeof makePullRequestRefreshAtomFamily>
+>();
+
+function createPullRequestRefreshAtomFamily<R, E>(
+  runtime: Atom.AtomRuntime<EnvironmentRegistry | R, E>,
+): ReturnType<typeof makePullRequestRefreshAtomFamily<R, E>> {
+  const existing = refreshFamiliesByRuntime.get(runtime);
+  if (existing) return existing as ReturnType<typeof makePullRequestRefreshAtomFamily<R, E>>;
+  const refreshes = makePullRequestRefreshAtomFamily(runtime);
+  refreshFamiliesByRuntime.set(runtime, refreshes);
+  return refreshes;
+}
+
 export function createLinkedPullRequestSummaryAtomFamily<R, E>(
   runtime: Atom.AtomRuntime<EnvironmentRegistry | R, E>,
   refreshes = createPullRequestRefreshAtomFamily(runtime),