feat(nbstore): add doc sync state (#9131)

packages/common/nbstore/src/sync/doc/index.ts

@@ -1,16 +1,55 @@
+import type { Observable } from 'rxjs';
+import { combineLatest, map } from 'rxjs';
+
 import type { DocStorage, SyncStorage } from '../../storage';
 import { DocSyncPeer } from './peer';
 
+export interface DocSyncState {
+  total: number;
+  syncing: number;
+  retrying: boolean;
+  errorMessage: string | null;
+}
+
+export interface DocSyncDocState {
+  syncing: boolean;
+  retrying: boolean;
+  errorMessage: string | null;
+}
+
 export class DocSync {
-  private readonly peers: DocSyncPeer[];
+  private readonly peers: DocSyncPeer[] = this.remotes.map(
+    remote => new DocSyncPeer(this.local, this.sync, remote)
+  );
   private abort: AbortController | null = null;
 
+  readonly state$: Observable<DocSyncState> = combineLatest(
+    this.peers.map(peer => peer.peerState$)
+  ).pipe(
+    map(allPeers => ({
+      total: allPeers.reduce((acc, peer) => acc + peer.total, 0),
+      syncing: allPeers.reduce((acc, peer) => acc + peer.syncing, 0),
+      retrying: allPeers.some(peer => peer.retrying),
+      errorMessage:
+        allPeers.find(peer => peer.errorMessage)?.errorMessage ?? null,
+    }))
+  );
+
   constructor(
     readonly local: DocStorage,
     readonly sync: SyncStorage,
     readonly remotes: DocStorage[]
-  ) {
-    this.peers = remotes.map(remote => new DocSyncPeer(local, sync, remote));
+  ) {}
+
+  docState$(docId: string): Observable<DocSyncDocState> {
+    return combineLatest(this.peers.map(peer => peer.docState$(docId))).pipe(
+      map(allPeers => ({
+        errorMessage:
+          allPeers.find(peer => peer.errorMessage)?.errorMessage ?? null,
+        retrying: allPeers.some(peer => peer.retrying),
+        syncing: allPeers.some(peer => peer.syncing),
+      }))
+    );
   }
 
   start() {

packages/common/nbstore/src/sync/doc/peer.ts

@@ -1,6 +1,6 @@
 import { remove } from 'lodash-es';
 import { nanoid } from 'nanoid';
-import { Subject } from 'rxjs';
+import { Observable, Subject } from 'rxjs';
 import { diffUpdate, encodeStateVectorFromUpdate, mergeUpdates } from 'yjs';
 
 import type { DocStorage, SyncStorage } from '../../storage';
@@ -46,6 +46,19 @@ interface Status {
   errorMessage: string | null;
 }
 
+interface PeerState {
+  total: number;
+  syncing: number;
+  retrying: boolean;
+  errorMessage: string | null;
+}
+
+interface PeerDocState {
+  syncing: boolean;
+  retrying: boolean;
+  errorMessage: string | null;
+}
+
 interface DocSyncPeerOptions {
   mergeUpdates?: (updates: Uint8Array[]) => Promise<Uint8Array> | Uint8Array;
 }
@@ -101,6 +114,50 @@ export class DocSyncPeer {
   };
   private readonly statusUpdatedSubject$ = new Subject<string | true>();
 
+  peerState$ = new Observable<PeerState>(subscribe => {
+    const next = () => {
+      if (!this.status.syncing) {
+        // if syncing = false, jobMap is empty
+        subscribe.next({
+          total: this.status.docs.size,
+          syncing: this.status.docs.size,
+          retrying: this.status.retrying,
+          errorMessage: this.status.errorMessage,
+        });
+      } else {
+        const syncing = this.status.jobMap.size;
+        subscribe.next({
+          total: this.status.docs.size,
+          syncing: syncing,
+          retrying: this.status.retrying,
+          errorMessage: this.status.errorMessage,
+        });
+      }
+    };
+    next();
+    return this.statusUpdatedSubject$.subscribe(() => {
+      next();
+    });
+  });
+
+  docState$(docId: string) {
+    return new Observable<PeerDocState>(subscribe => {
+      const next = () => {
+        subscribe.next({
+          syncing:
+            !this.status.connectedDocs.has(docId) ||
+            this.status.jobMap.has(docId),
+          retrying: this.status.retrying,
+          errorMessage: this.status.errorMessage,
+        });
+      };
+      next();
+      return this.statusUpdatedSubject$.subscribe(updatedId => {
+        if (updatedId === true || updatedId === docId) next();
+      });
+    });
+  }
+
   private readonly jobs = createJobErrorCatcher({
     connect: async (docId: string, signal?: AbortSignal) => {
       const pushedClock =