From e9003dec9ef751e48b87eb0dff6833b4d7199fac Mon Sep 17 00:00:00 2001 From: fundon Date: Mon, 28 Apr 2025 02:28:48 +0000 Subject: [PATCH] feat(editor): add blobState$ to BlobEngine (#11756) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Closes: [BS-3137](https://linear.app/affine-design/issue/BS-3137/在-bs-添加-blobstate-接口) ## 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. --- blocksuite/framework/sync/src/blob/engine.ts | 4 ++++ blocksuite/framework/sync/src/blob/source.ts | 11 +++++++++++ packages/common/nbstore/src/frontend/blob.ts | 4 ++++ packages/common/nbstore/src/sync/blob/index.ts | 2 +- packages/common/nbstore/src/sync/blob/peer.ts | 14 ++++++++------ packages/common/nbstore/src/worker/client.ts | 4 ++-- packages/common/nbstore/src/worker/consumer.ts | 3 ++- packages/common/nbstore/src/worker/ops.ts | 2 +- .../src/modules/workspace/entities/workspace.ts | 2 ++ 9 files changed, 35 insertions(+), 11 deletions(-) diff --git a/blocksuite/framework/sync/src/blob/engine.ts b/blocksuite/framework/sync/src/blob/engine.ts index a95732939..89546eeba 100644 --- a/blocksuite/framework/sync/src/blob/engine.ts +++ b/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; diff --git a/blocksuite/framework/sync/src/blob/source.ts b/blocksuite/framework/sync/src/blob/source.ts index ebb675db5..7b9d8382f 100644 --- a/blocksuite/framework/sync/src/blob/source.ts +++ b/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; delete: (key: string) => Promise; list: () => Promise; + // This state is only available when uploading to the cloud or downloading from the cloud. + blobState$?: (key: string) => Observable | null; } diff --git a/packages/common/nbstore/src/frontend/blob.ts b/packages/common/nbstore/src/frontend/blob.ts index 011234c4c..620ce377b 100644 --- a/packages/common/nbstore/src/frontend/blob.ts +++ b/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); diff --git a/packages/common/nbstore/src/sync/blob/index.ts b/packages/common/nbstore/src/sync/blob/index.ts index b96edbba4..82b93f579 100644 --- a/packages/common/nbstore/src/sync/blob/index.ts +++ b/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 => ({ diff --git a/packages/common/nbstore/src/sync/blob/peer.ts b/packages/common/nbstore/src/sync/blob/peer.ts index 3dcdf49ac..79858e5af 100644 --- a/packages/common/nbstore/src/sync/blob/peer.ts +++ b/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 => { - if (updatedBlobId === blobId || updatedBlobId === true) { - next(); - } - }); + const dispose = this.statusUpdatedSubject$ + .pipe( + filter( + updatedBlobId => updatedBlobId === blobId || updatedBlobId === true + ) + ) + .subscribe(() => next()); return () => { dispose.unsubscribe(); }; diff --git a/packages/common/nbstore/src/worker/client.ts b/packages/common/nbstore/src/worker/client.ts index b179842a8..d9b104a2b 100644 --- a/packages/common/nbstore/src/worker/client.ts +++ b/packages/common/nbstore/src/worker/client.ts @@ -268,8 +268,8 @@ class WorkerBlobSync implements BlobSync { downloadBlob(blobId: string): Promise { return this.client.call('blobSync.downloadBlob', blobId); } - uploadBlob(blob: BlobRecord): Promise { - return this.client.call('blobSync.uploadBlob', blob); + uploadBlob(blob: BlobRecord, force?: boolean): Promise { + return this.client.call('blobSync.uploadBlob', { blob, force }); } fullDownload(peerId?: string, signal?: AbortSignal): Promise { return new Promise((resolve, reject) => { diff --git a/packages/common/nbstore/src/worker/consumer.ts b/packages/common/nbstore/src/worker/consumer.ts index 0fea4d91f..f9105ec2d 100644 --- a/packages/common/nbstore/src/worker/consumer.ts +++ b/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(); diff --git a/packages/common/nbstore/src/worker/ops.ts b/packages/common/nbstore/src/worker/ops.ts index 8e8730a36..a2e1a9ea1 100644 --- a/packages/common/nbstore/src/worker/ops.ts +++ b/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]; }; diff --git a/packages/frontend/core/src/modules/workspace/entities/workspace.ts b/packages/frontend/core/src/modules/workspace/entities/workspace.ts index a99c19632..27e2644e1 100644 --- a/packages/frontend/core/src/modules/workspace/entities/workspace.ts +++ b/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, },