feat(editor): add blobState$ to BlobEngine (#11756)

Closes: [BS-3137](https://linear.app/affine-design/issue/BS-3137/在-bs-添加-blobstate-接口)

<!-- This is an auto-generated comment: release notes by coderabbit.ai -->

## Summary by CodeRabbit

- **New Features**
  - Introduced real-time status tracking for file upload and download operations, allowing users to monitor progress and errors more effectively.
- **Improvements**
  - Enhanced icon display for attachments, providing a refreshed visual experience.
  - Improved update frequency and accuracy for file synchronization status indicators.
- **Bug Fixes**
  - Refined internal logic for filtering status updates, ensuring more precise feedback during file operations.

<!-- end of auto-generated comment: release notes by coderabbit.ai -->

blocksuite/framework/sync/src/blob/engine.ts

@@ -104,6 +104,10 @@ export class BlobEngine {
     return key;
   }
 
+  blobState$(key: string) {
+    return this.main.blobState$?.(key) ?? null;
+  }
+
   start() {
     if (this._abort) {
       return;

blocksuite/framework/sync/src/blob/source.ts

@@ -1,3 +1,12 @@
+import type { Observable } from 'rxjs';
+
+export interface BlobState {
+  uploading: boolean;
+  downloading: boolean;
+  errorMessage?: string | null;
+  overSize: boolean;
+}
+
 export interface BlobSource {
   name: string;
   readonly: boolean;
@@ -5,4 +14,6 @@ export interface BlobSource {
   set: (key: string, value: Blob) => Promise<string>;
   delete: (key: string) => Promise<void>;
   list: () => Promise<string[]>;
+  // This state is only available when uploading to the cloud or downloading from the cloud.
+  blobState$?: (key: string) => Observable<BlobState> | null;
 }

packages/common/nbstore/src/frontend/blob.ts

@@ -15,6 +15,10 @@ export class BlobFrontend {
     return this.sync.state$;
   }
 
+  blobState$(blobId: string) {
+    return this.sync.blobState$(blobId);
+  }
+
   async get(blobId: string) {
     await this.waitForConnected();
     await using lock = await this.lock.lock('blob', blobId);

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

@@ -97,7 +97,7 @@ export class BlobSyncImpl implements BlobSync {
     return combineLatest(
       this.peers.map(peer => peer.blobPeerState$(blobId))
     ).pipe(
-      throttleTime(1000),
+      throttleTime(1000, undefined, { leading: true, trailing: true }),
       map(
         peers =>
           ({

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

@@ -1,5 +1,5 @@
 import { difference } from 'lodash-es';
-import { Observable, ReplaySubject, share, Subject } from 'rxjs';
+import { filter, Observable, ReplaySubject, share, Subject } from 'rxjs';
 
 import type { BlobRecord, BlobStorage } from '../../storage';
 import { OverCapacityError, OverSizeError } from '../../storage';
@@ -412,11 +412,13 @@ class BlobSyncPeerStatus {
         });
       };
       next();
-      const dispose = this.statusUpdatedSubject$.subscribe(updatedBlobId => {
+      const dispose = this.statusUpdatedSubject$
-        if (updatedBlobId === blobId || updatedBlobId === true) {
+        .pipe(
-          next();
+          filter(
-        }
+            updatedBlobId => updatedBlobId === blobId || updatedBlobId === true
-      });
+          )
+        )
+        .subscribe(() => next());
       return () => {
         dispose.unsubscribe();
       };

packages/common/nbstore/src/worker/client.ts

@@ -268,8 +268,8 @@ class WorkerBlobSync implements BlobSync {
   downloadBlob(blobId: string): Promise<boolean> {
     return this.client.call('blobSync.downloadBlob', blobId);
   }
-  uploadBlob(blob: BlobRecord): Promise<true> {
+  uploadBlob(blob: BlobRecord, force?: boolean): Promise<true> {
-    return this.client.call('blobSync.uploadBlob', blob);
+    return this.client.call('blobSync.uploadBlob', { blob, force });
   }
   fullDownload(peerId?: string, signal?: AbortSignal): Promise<void> {
     return new Promise((resolve, reject) => {

packages/common/nbstore/src/worker/consumer.ts

@@ -216,7 +216,8 @@ class StoreConsumer {
       'blobSync.state': () => this.blobSync.state$,
       'blobSync.blobState': blobId => this.blobSync.blobState$(blobId),
       'blobSync.downloadBlob': key => this.blobSync.downloadBlob(key),
-      'blobSync.uploadBlob': blob => this.blobSync.uploadBlob(blob),
+      'blobSync.uploadBlob': ({ blob, force }) =>
+        this.blobSync.uploadBlob(blob, force),
       'blobSync.fullDownload': peerId =>
         new Observable(subscriber => {
           const abortController = new AbortController();

packages/common/nbstore/src/worker/ops.ts

@@ -110,7 +110,7 @@ interface GroupedWorkerOps {
     state: [void, BlobSyncState];
     blobState: [string, BlobSyncBlobState];
     downloadBlob: [string, boolean];
-    uploadBlob: [BlobRecord, true];
+    uploadBlob: [{ blob: BlobRecord; force?: boolean }, true];
     fullDownload: [string | null, void];
   };
 

packages/frontend/core/src/modules/workspace/entities/workspace.ts

@@ -46,6 +46,8 @@ export class Workspace extends Entity {
             });
             return id;
           },
+          /* eslint-disable rxjs/finnish */
+          blobState$: key => this.engine.blob.blobState$(key),
           name: 'blob',
           readonly: false,
         },