/* eslint-disable max-lines */ /* eslint-disable @typescript-eslint/no-unused-vars */ /// import { Buffer } from 'buffer'; import { saveAs } from 'file-saver'; import { fromEvent } from 'file-selector'; import LRUCache from 'lru-cache'; import { nanoid } from 'nanoid'; import { debounce } from 'ts-debounce'; import { Awareness } from 'y-protocols/awareness.js'; import { applyUpdate, Array as YArray, Doc, encodeStateAsUpdate, Map as YMap, snapshot, transact, } from 'yjs'; import { BlockItem, BlockTypes } from '../types'; import { getLogger, sha3, sleep } from '../utils'; import { AsyncDatabaseAdapter, BlockListener, ChangedStateKeys, Connectivity, HistoryManager, } from './types'; import { YjsRemoteBinaries } from './binary'; import { YjsBlockInstance } from './block'; import { GateKeeper } from './gatekeeper'; import { YjsHistoryManager } from './history'; import { EmitEvents, Suspend } from './listener'; import { DO_NOT_USE_THIS_OR_YOU_WILL_BE_FIRED_SYMBOL_INTO_INNER as INTO_INNER, YjsContentOperation, } from './operation'; import { YjsProvider } from './provider'; declare const JWT_DEV: boolean; // @ts-ignore const logger = getLogger('BlockDB:yjs'); type ConnectivityListener = ( workspace: string, connectivity: Connectivity ) => void; type YjsProviders = { awareness: Awareness; binaries: Doc; doc: Doc; gatekeeper: GateKeeper; connListener: { listeners?: ConnectivityListener }; userId: string; remoteToken: string | undefined; // remote storage token providers: unknown[]; }; const _yjsDatabaseInstance = new Map(); const _asyncInitLoading = new Set(); const _waitLoading = async (workspace: string) => { while (_asyncInitLoading.has(workspace)) { await sleep(); } }; async function _initYjsDatabase( workspace: string, options: { userId: string; token?: string | undefined; provider?: Record | undefined; } ): Promise { if (_asyncInitLoading.has(workspace)) { await _waitLoading(workspace); } const instance = _yjsDatabaseInstance.get(workspace); // TODO: temporarily handle this if ( instance && (instance.userId === options.userId || options.userId === 'default') ) { return instance; } // if (instance) return instance; _asyncInitLoading.add(workspace); const { userId, token } = options; const doc = new Doc({ autoLoad: true, shouldLoad: true }); // const idb = await new IndexedDBProvider(workspace, doc).whenSynced; const binaries = new Doc({ autoLoad: true, shouldLoad: true }); const awareness = new Awareness(doc); const gateKeeperData = doc.getMap>('gatekeeper'); const gatekeeper = new GateKeeper( userId, gateKeeperData.get('creators') || gateKeeperData.set('creators', new YMap()), gateKeeperData.get('common') || gateKeeperData.set('common', new YMap()) ); const connListener: { listeners?: ConnectivityListener } = {}; let providers: unknown[] = []; if (options.provider) { const emitState = (c: Connectivity) => connListener.listeners?.(workspace, c); providers = await Promise.all( Object.entries(options.provider).flatMap(([name, p]) => [ p({ awareness, doc, token, workspace, emitState }).then(p => { console.log(p); return { [name]: p, }; }), p({ awareness, doc: binaries, token, workspace: `${workspace}_binaries`, emitState, }).then(p => ({ [`${name}_binaries`]: p })), ]) ); } const newInstance = { awareness, binaries, doc, gatekeeper, providers, connListener, userId, remoteToken: token, }; _yjsDatabaseInstance.set(workspace, newInstance); _asyncInitLoading.delete(workspace); return newInstance; } export type { YjsBlockInstance } from './block'; export type { YjsContentOperation } from './operation'; export { getYjsProviders } from './provider'; export type { YjsProviderOptions } from './provider'; export type YjsInitOptions = { userId?: string; token?: string; provider?: Record; }; export class YjsAdapter implements AsyncDatabaseAdapter { private readonly _provider: YjsProviders; private readonly _doc: Doc; // doc instance private readonly _awareness: Awareness; // lightweight state synchronization private readonly _gatekeeper: GateKeeper; // Simple access control private readonly _history!: YjsHistoryManager; // Block Collection // key is a randomly generated global id private readonly _blocks!: YMap>; private readonly _blockUpdated!: YMap; // Maximum cache Block 1024, ttl 10 minutes private readonly _blockCaches!: LRUCache; private readonly _binaries!: YjsRemoteBinaries; private readonly _listener: Map>; private readonly _reload: () => void; static async init( workspace: string, options: YjsInitOptions ): Promise { const { userId = 'default', token, provider } = options; const providers = await _initYjsDatabase(workspace, { userId, token, provider, }); return new YjsAdapter(providers); } private constructor(providers: YjsProviders) { this._provider = providers; this._doc = providers.doc; this._awareness = providers.awareness; this._gatekeeper = providers.gatekeeper; this._reload = () => { const blocks = this._doc.getMap>('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.binaries.getMap(), providers.remoteToken ); // @ts-ignore this._history = new YjsHistoryManager(this._blocks); }; this._reload(); this._listener = new Map(); providers.connListener.listeners = ( workspace: string, connectivity: Connectivity ) => { this._listener.get('connectivity')?.( new Map([[workspace, connectivity]]) ); }; const debounced_editing_notifier = debounce( () => { const listener: BlockListener> | undefined = this._listener.get('editing'); if (listener) { const mapping = this._awareness.getStates(); const editing_mapping: Record = {}; for (const { userId, editing, updated, } of mapping.values()) { // Only return the status with refresh time within 10 seconds if ( userId && editing && updated && typeof updated === 'number' && updated + 1000 * 10 > Date.now() ) { if (!editing_mapping[editing]) { editing_mapping[editing] = []; } editing_mapping[editing]?.push(userId); } } listener( new Map( Object.entries(editing_mapping).map(([k, v]) => [ k, new Set(v), ]) ) ); } }, 200, { maxWait: 1000 } ); this._awareness.setLocalStateField('userId', providers.userId); this._awareness.on('update', debounced_editing_notifier); this._blocks.observeDeep(events => { const now = Date.now(); const keys = events.flatMap(e => { if ((e.path?.length | 0) > 0) { return [ [e.path[0], 'update'] as [string, ChangedStateKeys], ]; } else { return Array.from(e.changes.keys.entries()).map( ([k, { action }]) => [k, action] as [string, ChangedStateKeys] ); } }); EmitEvents(keys, this._listener.get('updated')); transact(this._doc, () => { for (const [key, action] of keys) { if (action === 'delete') { this._blockUpdated.delete(key); } else { this._blockUpdated.set(key, now); } } }); }); } reload() { this._reload(); } getUserId(): string { return this._provider.userId; } inspector() { const resolve_block = (blocks: Record, id: string) => { const block = blocks[id]; if (block) { return { ...block, children: block.children.map((id: string) => resolve_block(blocks, id) ), }; } }; return { save: () => { const binary = encodeStateAsUpdate(this._doc); saveAs( new Blob([binary]), `affine_workspace_${new Date().toDateString()}.affine` ); }, load: async () => { try { const handles = await window.showOpenFilePicker({ types: [ { description: 'AFFiNE Package', accept: { // eslint-disable-next-line @typescript-eslint/naming-convention 'application/affine': ['.affine'], }, }, ], }); const [file] = (await fromEvent(handles)) as File[]; const binary = await file?.arrayBuffer(); console.log(this._provider.providers); let { indexeddb } = ( this._provider.providers as any[] ).find(p => p.indexeddb); await indexeddb?.idb?.clearData(); const doc = new Doc({ autoLoad: true, shouldLoad: true }); let updated = 0; let isUpdated = false; doc.on('update', () => { isUpdated = true; updated += 1; }); setInterval(() => { if (updated > 0) { updated -= 1; } }, 500); const update_check = new Promise(resolve => { const check = async () => { while (!isUpdated || updated > 0) { await sleep(); } resolve(); }; check(); }); await new indexeddb.ctor(indexeddb.idb.name, doc) .whenSynced; if (binary) { applyUpdate(doc, new Uint8Array(binary)); await update_check; } return true; } catch (err) { console.log(err); return false; } }, parse: () => this._doc.toJSON(), // eslint-disable-next-line @typescript-eslint/naming-convention parse_page: (page_id: string) => { const blocks = this._blocks.toJSON(); return resolve_block(blocks, page_id); }, // eslint-disable-next-line @typescript-eslint/naming-convention parse_pages: (resolve = false) => { const blocks = this._blocks.toJSON(); return Object.fromEntries( Object.entries(blocks) .filter(([, block]) => block.flavor === 'page') .map(([key, block]) => { if (resolve) { return resolve_block(blocks, key); } else { return [key, block]; } }) ); }, clear: () => { this._blocks.clear(); this._blockUpdated.clear(); this._gatekeeper.clear(); this._doc.getMap('blocks').clear(); this._doc.getMap('gatekeeper').clear(); }, // eslint-disable-next-line @typescript-eslint/naming-convention clear_old: () => { this._doc.getMap('block_updated').clear(); this._doc.getMap('blocks').clear(); this._doc.getMap('common').clear(); this._doc.getMap('creators').clear(); }, snapshot: () => { return snapshot(this._doc); }, }; } async createBlock( options: Pick, 'type' | 'flavor'> & { uuid: string | undefined; binary: ArrayBufferLike | undefined; } ): Promise { const uuid = options.uuid || `affine${nanoid(16)}`; if (options.type === BlockTypes.binary) { if (options.binary && options.binary instanceof ArrayBuffer) { const array = new YArray(); array.insert(0, [options.binary]); const block = { type: options.type, flavor: options.flavor, children: [] as string[], created: Date.now(), content: new YjsContentOperation(array), hash: sha3(Buffer.from(options.binary)), }; await this.set_block(uuid, block); return (await this.getBlock(uuid))!; } else { throw new Error(`Invalid binary type: ${options.binary}`); } } else { const block = { type: options.type, flavor: options.flavor, children: [] as string[], created: Date.now(), content: new YjsContentOperation(new YMap()), }; await this.set_block(uuid, block); return (await this.getBlock(uuid))!; } } private get_updated(id: string) { return this._blockUpdated.get(id); } private get_creator(id: string) { return this._gatekeeper.getCreator(id); } private get_block_sync(id: string): YjsBlockInstance | undefined { const cached = this._blockCaches.get(id); if (cached) { // Synchronous read cannot read binary if (cached.type === BlockTypes.block) { return cached; } return undefined; } const block = this._blocks.get(id); // Synchronous read cannot read binary if (block && block.get('type') === BlockTypes.block) { return new YjsBlockInstance({ id, block, setBlock: this.set_block.bind(this), getUpdated: this.get_updated.bind(this), getCreator: this.get_creator.bind(this), getBlockInstance: this.get_block_sync.bind(this), }); } return undefined; } async getBlock(id: string): Promise { const block_instance = this.get_block_sync(id); if (block_instance) { return block_instance; } const block = this._blocks.get(id); if (block && block.get('type') === BlockTypes.binary) { const binary = await this._binaries.get( block.get('hash') as string ); if (binary) { return new YjsBlockInstance({ id, block, binary, setBlock: this.set_block.bind(this), getUpdated: this.get_updated.bind(this), getCreator: this.get_creator.bind(this), getBlockInstance: this.get_block_sync.bind(this), }); } } return undefined; } async getBlockByFlavor( flavor: BlockItem['flavor'] ): Promise { const keys: string[] = []; this._blocks.forEach((doc, key) => { if (doc.get('flavor') === flavor) { keys.push(key); } }); return keys; } async getBlockByType( type: BlockItem['type'] ): Promise { const keys: string[] = []; this._blocks.forEach((doc, key) => { if (doc.get('type') === type) { keys.push(key); } }); return keys; } private async set_block( key: string, item: BlockItem & { hash?: string } ): Promise { return new Promise((resolve, reject) => { const block = this._blocks.get(key) || new YMap(); transact(this._doc, () => { // Insert only if the block doesn't exist yet // Other modification operations are done in the block instance let uploaded: Promise | undefined; if (!block.size) { const content = item.content[INTO_INNER](); if (!content) { return reject(); } const children = new YArray(); children.push(item.children); block.set('type', item.type); block.set('flavor', item.flavor); block.set('children', children); block.set('created', item.created); if (item.type === BlockTypes.block) { block.set('content', content); } else if (item.type === BlockTypes.binary && item.hash) { if (content instanceof YArray) { block.set('hash', item.hash); if (!this._binaries.has(item.hash)) { uploaded = this._binaries.set( item.hash, content ); } } else { throw new Error( 'binary content must be an buffer yarray' ); } } else { throw new Error('invalid block type: ' + item.type); } this._blocks.set(key, block); } if (item.flavor === 'page') { this._awareness.setLocalStateField('editing', key); this._awareness.setLocalStateField('updated', Date.now()); } // References do not add delete restrictions if (item.flavor === 'reference') { this._gatekeeper.setCommon(key); } else { this._gatekeeper.setCreator(key); } if (uploaded) { // TODO: there should be a mechanism to retry the upload uploaded.catch(err => { // undo set on failure console.error('Failed to upload object: ', err); this.deleteBlocks([key]); reject(err); }); } resolve(); }); }); } async checkBlocks(keys: string[]): Promise { return ( keys.filter(key => !!this._blocks.get(key)).length === keys.length ); } async deleteBlocks(keys: string[]): Promise { const [success, fail] = this._gatekeeper.checkDeleteLists(keys); transact(this._doc, () => { for (const key of success) { this._blocks.delete(key); } }); return fail; } on( key: 'editing' | 'updated' | 'connectivity', listener: BlockListener ): void { this._listener.set(key, listener); } suspend(suspend: boolean) { Suspend(suspend); } public history(): HistoryManager { return this._history; } }