refactor: local storage

libs/datasource/db-service/src/services/base.ts

@@ -154,6 +154,14 @@ export abstract class ServiceBaseClass {
         await this.database.unregisterTagExporter(workspace, name);
     }
 
+    async setupDataExporter(
+        workspace: string,
+        initialData: Uint8Array,
+        cb: (data: Uint8Array) => Promise<void>
+    ) {
+        await this.database.setupDataExporter(workspace, initialData, cb);
+    }
+
     protected async _observe(
         workspace: string,
         blockId: string,

libs/datasource/db-service/src/services/database/index.ts

@@ -192,4 +192,13 @@ export class Database {
             }
         }
     }
+
+    async setupDataExporter(
+        workspace: string,
+        initialData: Uint8Array,
+        callback: (binary: Uint8Array) => Promise<void>
+    ) {
+        const db = await this.getDatabase(workspace);
+        await db.setupDataExporter(initialData, callback);
+    }
 }

libs/datasource/jwt-rpc/src/sqlite.ts

@@ -5,7 +5,7 @@ import { Observable } from 'lib0/observable.js';
 const PREFERRED_TRIM_SIZE = 500;
 
 const _stmts = {
-    create: 'CREATE TABLE updates (key INTEGER PRIMARY KEY AUTOINCREMENT, value BLOB);',
+    create: 'CREATE TABLE IF NOT EXISTS updates (key INTEGER PRIMARY KEY AUTOINCREMENT, value BLOB);',
     selectAll: 'SELECT * FROM updates where key >= $idx',
     selectCount: 'SELECT count(*) FROM updates',
     insert: 'INSERT INTO updates VALUES (null, $data);',
@@ -59,7 +59,7 @@ export class SQLiteProvider extends Observable<string> {
     private _size: number;
     private _destroyed: boolean;
     private _db: Promise<Database>;
-    private _saver?: (binary: Uint8Array) => void;
+    private _saver?: (binary: Uint8Array) => Promise<void> | undefined;
     private _destroy: () => void;
 
     constructor(name: string, doc: Y.Doc, origin?: Uint8Array) {
@@ -82,8 +82,9 @@ export class SQLiteProvider extends Observable<string> {
         this.whenSynced = this._db.then(async db => {
             this.db = db;
             const currState = Y.encodeStateAsUpdate(doc);
-            await this._fetchUpdates();
+            await this._fetchUpdates(true);
             db.exec(_stmts.insert, { $data: currState });
+            this._storeState();
             if (this._destroyed) return this;
             this.emit('synced', [this]);
             this.synced = true;
@@ -91,21 +92,38 @@ export class SQLiteProvider extends Observable<string> {
         });
 
         // Timeout in ms until data is merged and persisted in sqlite.
-        const storeTimeout = 1000;
+        const storeTimeout = 500;
         let storeTimeoutId: NodeJS.Timer | undefined = undefined;
+        let lastSize = 0;
+
+        const debouncedStoreState = (force = false) => {
+            // debounce store call
+            if (storeTimeoutId) clearTimeout(storeTimeoutId);
+
+            if (force) {
+                if (lastSize !== this._size) {
+                    this._storeState();
+                    storeTimeoutId = undefined;
+                    lastSize = this._size;
+                }
+            } else {
+                storeTimeoutId = setTimeout(() => {
+                    this._storeState();
+                    storeTimeoutId = undefined;
+                }, storeTimeout);
+            }
+        };
+        const storeStateInterval = setInterval(
+            () => debouncedStoreState(true),
+            1000
+        );
 
         const storeUpdate = (update: Uint8Array, origin: any) => {
             if (this._saver && this.db && origin !== this) {
                 this.db.exec(_stmts.insert, { $data: update });
 
                 if (++this._size >= PREFERRED_TRIM_SIZE) {
-                    // debounce store call
-                    if (storeTimeoutId) clearTimeout(storeTimeoutId);
-
-                    storeTimeoutId = setTimeout(() => {
-                        this._storeState();
-                        storeTimeoutId = undefined;
-                    }, storeTimeout);
+                    debouncedStoreState();
                 }
             }
         };
@@ -116,34 +134,53 @@ export class SQLiteProvider extends Observable<string> {
 
         this._destroy = () => {
             if (storeTimeoutId) clearTimeout(storeTimeoutId);
+            if (storeStateInterval) clearInterval(storeStateInterval);
 
             this.doc.off('update', storeUpdate);
             this.doc.off('destroy', this.destroy);
         };
     }
 
-    registerExporter(saver: (binary: Uint8Array) => void) {
+    registerExporter(saver: (binary: Uint8Array) => Promise<void> | undefined) {
         this._saver = saver;
     }
 
-    private async _storeState() {
+    private async _storeState(force?: boolean) {
         await this._fetchUpdates();
 
-        if (this.db && this._size >= PREFERRED_TRIM_SIZE) {
-            this.db.exec(_stmts.insert, {
-                $data: Y.encodeStateAsUpdate(this.doc),
-            });
+        if (this.db) {
+            if (force || this._size >= PREFERRED_TRIM_SIZE) {
+                this.db.exec(_stmts.insert, {
+                    $data: Y.encodeStateAsUpdate(this.doc),
+                });
 
-            clearUpdates(this.db, this._ref);
+                clearUpdates(this.db, this._ref);
 
-            this._size = countUpdates(this.db);
+                this._size = countUpdates(this.db);
+            }
 
-            this._saver?.(this.db?.export());
+            await this._saver?.(this.db?.export());
         }
     }
 
-    private async _fetchUpdates() {
+    private _waitUpdate(sync = false) {
+        if (sync) {
+            return new Promise<void>((resolve, reject) => {
+                const final = (_: any, origin: any) => {
+                    if (origin === this) {
+                        this.doc.off('update', final);
+                        resolve();
+                    }
+                };
+                this.doc.on('update', final);
+            });
+        }
+        return undefined;
+    }
+
+    private async _fetchUpdates(sync = false) {
         if (this.db) {
+            const wait = this._waitUpdate(sync);
             const updates = getAllUpdates(this.db, this._ref);
 
             Y.transact(
@@ -160,6 +197,7 @@ export class SQLiteProvider extends Observable<string> {
             const lastKey = Math.max(...updates.map(([idx]) => idx));
             this._ref = lastKey + 1;
             this._size = countUpdates(this.db);
+            await wait;
         }
     }
 

libs/datasource/jwt/src/adapter/index.ts

@@ -136,6 +136,7 @@ interface BlockInstance<C extends ContentOperation> {
 
 interface AsyncDatabaseAdapter<C extends ContentOperation> {
     inspector(): Record<string, any>;
+    reload(): void;
     createBlock(
         options: Pick<BlockItem<C>, 'type' | 'flavor'> & {
             binary?: ArrayBuffer;
@@ -156,6 +157,33 @@ interface AsyncDatabaseAdapter<C extends ContentOperation> {
     getUserId(): string;
 }
 
+export type DataExporter = (binary: Uint8Array) => Promise<void>;
+
+export const getDataExporter = () => {
+    let exporter: DataExporter | undefined = undefined;
+    let importer: (() => Uint8Array | undefined) | undefined = undefined;
+
+    const importData = () => importer?.();
+    const exportData = (binary: Uint8Array) => exporter?.(binary);
+    const hasExporter = () => !!exporter;
+
+    const installExporter = (
+        initialData: Uint8Array | undefined,
+        cb: DataExporter
+    ) => {
+        return new Promise<void>(resolve => {
+            importer = () => initialData;
+            exporter = async (data: Uint8Array) => {
+                exporter = cb;
+                await cb(data);
+                resolve();
+            };
+        });
+    };
+
+    return { importData, exportData, hasExporter, installExporter };
+};
+
 export type {
     AsyncDatabaseAdapter,
     BlockPosition,

libs/datasource/jwt/src/adapter/yjs/index.ts

@@ -153,19 +153,21 @@ export class YjsAdapter implements AsyncDatabaseAdapter<YjsContentOperation> {
     private readonly _doc: Doc; // doc instance
     private readonly _awareness: Awareness; // lightweight state synchronization
     private readonly _gatekeeper: GateKeeper; // Simple access control
-    private readonly _history: YjsHistoryManager;
+    private readonly _history!: YjsHistoryManager;
 
     // Block Collection
     // key is a randomly generated global id
-    private readonly _blocks: YMap<YMap<unknown>>;
-    private readonly _blockUpdated: YMap<number>;
+    private readonly _blocks!: YMap<YMap<unknown>>;
+    private readonly _blockUpdated!: YMap<number>;
     // Maximum cache Block 1024, ttl 10 minutes
-    private readonly _blockCaches: LRUCache<string, YjsBlockInstance>;
+    private readonly _blockCaches!: LRUCache<string, YjsBlockInstance>;
 
-    private readonly _binaries: YjsRemoteBinaries;
+    private readonly _binaries!: YjsRemoteBinaries;
 
     private readonly _listener: Map<string, BlockListener<any>>;
 
+    private readonly _reload: () => void;
+
     static async init(
         workspace: string,
         options: YjsInitOptions
@@ -184,18 +186,28 @@ export class YjsAdapter implements AsyncDatabaseAdapter<YjsContentOperation> {
         this._doc = providers.idb.doc;
         this._awareness = providers.awareness;
         this._gatekeeper = providers.gatekeeper;
-
-        const blocks = this._doc.getMap<YMap<any>>('blocks');
-        this._blocks =
-            blocks.get('content') || blocks.set('content', new YMap());
-        this._blockUpdated =
-            blocks.get('updated') || blocks.set('updated', new YMap());
-        this._blockCaches = new LRUCache({ max: 1024, ttl: 1000 * 60 * 10 });
-        this._binaries = new YjsRemoteBinaries(
-            providers.binariesIdb.doc.getMap(),
-            providers.remoteToken
-        );
-        this._history = new YjsHistoryManager(this._blocks);
+        this._reload = () => {
+            const blocks = this._doc.getMap<YMap<any>>('blocks');
+            // @ts-ignore
+            this._blocks =
+                blocks.get('content') || blocks.set('content', new YMap());
+            // @ts-ignore
+            this._blockUpdated =
+                blocks.get('updated') || blocks.set('updated', new YMap());
+            // @ts-ignore
+            this._blockCaches = new LRUCache({
+                max: 1024,
+                ttl: 1000 * 60 * 10,
+            });
+            // @ts-ignore
+            this._binaries = new YjsRemoteBinaries(
+                providers.binariesIdb.doc.getMap(),
+                providers.remoteToken
+            );
+            // @ts-ignore
+            this._history = new YjsHistoryManager(this._blocks);
+        };
+        this._reload();
 
         this._listener = new Map();
 
@@ -281,6 +293,10 @@ export class YjsAdapter implements AsyncDatabaseAdapter<YjsContentOperation> {
         });
     }
 
+    reload() {
+        this._reload();
+    }
+
     getUserId(): string {
         return this._provider.userId;
     }

libs/datasource/jwt/src/adapter/yjs/provider.ts

@@ -22,8 +22,9 @@ export type YjsProvider = (instances: YjsDefaultInstances) => Promise<void>;
 export type YjsProviderOptions = {
     backend: typeof BucketBackend[keyof typeof BucketBackend];
     params?: Record<string, string>;
-    importData?: Uint8Array;
-    exportData?: (binary: Uint8Array) => void;
+    importData?: () => Promise<Uint8Array> | Uint8Array | undefined;
+    exportData?: (binary: Uint8Array) => Promise<void> | undefined;
+    hasExporter?: () => boolean;
 };
 
 export const getYjsProviders = (
@@ -31,13 +32,20 @@ export const getYjsProviders = (
 ): Record<string, YjsProvider> => {
     return {
         sqlite: async (instances: YjsDefaultInstances) => {
-            const fs = new SQLiteProvider(
-                instances.workspace,
-                instances.doc,
-                options.importData
-            );
-            if (options.exportData) fs.registerExporter(options.exportData);
-            await fs.whenSynced;
+            const fsHandle = setInterval(async () => {
+                if (options.hasExporter?.()) {
+                    clearInterval(fsHandle);
+                    const fs = new SQLiteProvider(
+                        instances.workspace,
+                        instances.doc,
+                        await options.importData?.()
+                    );
+                    if (options.exportData) {
+                        fs.registerExporter(options.exportData);
+                    }
+                    await fs.whenSynced;
+                }
+            }, 500);
         },
         ws: async (instances: YjsDefaultInstances) => {
             if (instances.token) {

libs/datasource/jwt/src/index.ts

@@ -14,6 +14,8 @@ import {
     HistoryManager,
     ContentTypes,
     Connectivity,
+    DataExporter,
+    getDataExporter,
 } from './adapter';
 import {
     getYjsProviders,
@@ -66,6 +68,10 @@ type BlockClientOptions = {
     content?: BlockExporters<string>;
     metadata?: BlockExporters<Array<[string, number | string | string[]]>>;
     tagger?: BlockExporters<string[]>;
+    installExporter: (
+        initialData: Uint8Array,
+        exporter: DataExporter
+    ) => Promise<void>;
 };
 
 export class BlockClient<
@@ -95,10 +101,15 @@ export class BlockClient<
 
     private readonly _root: { node?: BaseBlock<B, C> };
 
+    private readonly _installExporter: (
+        initialData: Uint8Array,
+        exporter: DataExporter
+    ) => Promise<void>;
+
     private constructor(
         adapter: A,
         workspace: string,
-        options?: BlockClientOptions
+        options: BlockClientOptions
     ) {
         this._adapter = adapter;
         this._workspace = workspace;
@@ -142,6 +153,7 @@ export class BlockClient<
             });
 
         this._root = {};
+        this._installExporter = options.installExporter;
     }
 
     public addBlockListener(tag: string, listener: BlockListener) {
@@ -590,21 +602,34 @@ export class BlockClient<
         return this._adapter.history();
     }
 
+    public async setupDataExporter(initialData: Uint8Array, cb: DataExporter) {
+        await this._installExporter(initialData, cb);
+        this._adapter.reload();
+    }
+
     public static async init(
         workspace: string,
         options: Partial<
             YjsInitOptions & YjsProviderOptions & BlockClientOptions
         > = {}
     ): Promise<BlockClientInstance> {
+        const { importData, exportData, hasExporter, installExporter } =
+            getDataExporter();
+
         const instance = await YjsAdapter.init(workspace, {
             provider: getYjsProviders({
                 backend: BucketBackend.YjsWebSocketAffine,
-                exportData: console.log.bind(console),
+                importData,
+                exportData,
+                hasExporter,
                 ...options,
             }),
             ...options,
         });
-        return new BlockClient(instance, workspace, options);
+        return new BlockClient(instance, workspace, {
+            ...options,
+            installExporter,
+        });
     }
 }
 

libs/datasource/state/src/user.ts

@@ -55,28 +55,39 @@ const _useUserAndSpace = () => {
 
     const currentSpaceId: string | undefined = useMemo(() => user?.id, [user]);
 
-    return {
-        user,
-        currentSpaceId,
-        loading,
-    };
+    return { user, currentSpaceId, loading };
 };
 
+const BRAND_ID = 'AFFiNE';
+
+const _localTrigger = atom<boolean>(false);
 const _useUserAndSpacesForFreeLogin = () => {
+    const [user, setUser] = useAtom(_userAtom);
     const [loading, setLoading] = useAtom(_loadingAtom);
+    const [localTrigger] = useAtom(_localTrigger);
 
     useEffect(() => setLoading(false), []);
-    const BRAND_ID = 'AFFiNE';
-    return {
-        user: {
-            photo: '',
-            id: BRAND_ID,
-            nickname: BRAND_ID,
-            email: '',
-        } as UserInfo,
-        currentSpaceId: BRAND_ID,
-        loading,
-    };
+
+    useEffect(() => {
+        if (localTrigger) {
+            setUser({
+                photo: '',
+                id: BRAND_ID,
+                username: BRAND_ID,
+                nickname: BRAND_ID,
+                email: '',
+            });
+        }
+    }, [localTrigger, setLoading, setUser]);
+
+    const currentSpaceId: string | undefined = useMemo(() => user?.id, [user]);
+
+    return { user, currentSpaceId, loading };
+};
+
+export const useLocalTrigger = () => {
+    const [, setTrigger] = useAtom(_localTrigger);
+    return () => setTrigger(true);
 };
 
 export const useUserAndSpaces = process.env['NX_LOCAL']