Merge branch 'feat/filesystem_and_search' of github.com:toeverything/AFFiNE into feat/filesystem_and_search

This commit is contained in:
DiamondThree
2022-12-22 09:49:04 +08:00
10 changed files with 582 additions and 18 deletions
+3 -1
View File
@@ -28,6 +28,7 @@
"css-spring": "^4.1.0",
"dayjs": "^1.11.7",
"i18next": "^21.9.1",
"lib0": "^0.2.58",
"lit": "^2.3.1",
"next": "13.0.1",
"prettier": "^2.7.1",
@@ -35,7 +36,8 @@
"quill-cursors": "^4.0.0",
"react": "18.2.0",
"react-dom": "18.2.0",
"react-i18next": "^11.18.4"
"react-i18next": "^11.18.4",
"y-protocols": "^1.0.5"
},
"devDependencies": {
"@types/node": "18.7.18",
+1 -1
View File
@@ -58,7 +58,7 @@ export const usePageHelper = (): EditorHandlers => {
});
return trash;
},
search: (query: string) => {
search: (query: QueryContent) => {
return currentWorkspace!.search(query);
},
changePageMode: async (pageId, mode) => {
@@ -7,7 +7,8 @@ import type {
} from '@blocksuite/store';
import type { EditorContainer } from '@blocksuite/editor';
export type LoadWorkspaceHandler = (
workspaceId: string
workspaceId: string,
websocket?: boolean
) => Promise<StoreWorkspace | null> | null;
export type CreateEditorHandler = (page: StorePage) => EditorContainer | null;
@@ -13,7 +13,8 @@ import type {
LoadWorkspaceHandler,
CreateEditorHandler,
} from './context';
import { downloadWorkspace } from '@pathfinder/data-services';
import { downloadWorkspace, getToken } from '@pathfinder/data-services';
import { WebsocketProvider } from './y-websocket';
const getEditorParams = (workspaceId: string) => {
const providers = [];
@@ -44,12 +45,46 @@ const DynamicBlocksuite = ({
setCreateEditorHandler,
}: Props) => {
useEffect(() => {
const openWorkspace: LoadWorkspaceHandler = (workspaceId: string) =>
const openWorkspace: LoadWorkspaceHandler = (
workspaceId: string,
websocket = false
) =>
new Promise(async resolve => {
const workspace = new Workspace({
...getEditorParams(workspaceId as string),
room: workspaceId,
providers: [IndexedDBDocProvider],
}).register(BlockSchema);
const refreshToken = getToken()?.refreshToken;
if (
websocket &&
refreshToken &&
location.search.includes('sync=websocket')
) {
// FIXME: if add websocket provider, the first page will be blank
const ws = new WebsocketProvider(
`ws${window.location.protocol === 'https:' ? 's' : ''}://${
window.location.host
}/api/sync/`,
workspaceId,
workspace.doc,
{
params: {
token: refreshToken,
},
awareness: workspace.meta.awareness.awareness,
}
);
ws.shouldConnect = false;
// FIXME: there needs some method to destroy websocket.
// Or we need a manager to manage websocket.
// @ts-expect-error
workspace.__ws__ = ws;
}
const indexDBProvider = workspace.providers.find(
p => p instanceof IndexedDBDocProvider
);
@@ -50,7 +50,18 @@ export const AppStateProvider = ({ children }: { children?: ReactNode }) => {
const loadWorkspace =
useRef<(workspaceId: string) => Promise<Workspace | null> | null>();
loadWorkspace.current = async (workspaceId: string) => {
const workspace = (await loadWorkspaceHandler?.(workspaceId)) || null;
if (state.currentWorkspaceId === workspaceId) {
return state.currentWorkspace;
}
const workspace = (await loadWorkspaceHandler?.(workspaceId, true)) || null;
// @ts-expect-error
window.workspace = workspace;
// FIXME: there needs some method to destroy websocket.
// Or we need a manager to manage websocket.
// @ts-expect-error
state.currentWorkspace?.__ws__?.destroy();
setState(state => ({
...state,
currentWorkspace: workspace,
@@ -0,0 +1,507 @@
/**
* @module provider/websocket
*/
/* eslint-env browser */
// import * as Y from 'yjs'; // eslint-disable-line
import * as bc from 'lib0/broadcastchannel';
import * as time from 'lib0/time';
import * as encoding from 'lib0/encoding';
import * as decoding from 'lib0/decoding';
import * as syncProtocol from 'y-protocols/sync';
import * as authProtocol from 'y-protocols/auth';
import * as awarenessProtocol from 'y-protocols/awareness';
import { Observable } from 'lib0/observable';
import * as math from 'lib0/math';
import * as url from 'lib0/url';
export const messageSync = 0;
export const messageQueryAwareness = 3;
export const messageAwareness = 1;
export const messageAuth = 2;
/**
* encoder, decoder, provider, emitSynced, messageType
* @type {Array<function(encoding.Encoder, decoding.Decoder, WebsocketProvider, boolean, number):void>}
*/
const messageHandlers = [];
messageHandlers[messageSync] = (
encoder,
decoder,
provider,
emitSynced,
_messageType
) => {
encoding.writeVarUint(encoder, messageSync);
const syncMessageType = syncProtocol.readSyncMessage(
decoder,
encoder,
provider.doc,
provider
);
if (
emitSynced &&
syncMessageType === syncProtocol.messageYjsSyncStep2 &&
!provider.synced
) {
provider.synced = true;
}
};
messageHandlers[messageQueryAwareness] = (
encoder,
_decoder,
provider,
_emitSynced,
_messageType
) => {
encoding.writeVarUint(encoder, messageAwareness);
encoding.writeVarUint8Array(
encoder,
awarenessProtocol.encodeAwarenessUpdate(
provider.awareness,
Array.from(provider.awareness.getStates().keys())
)
);
};
messageHandlers[messageAwareness] = (
_encoder,
decoder,
provider,
_emitSynced,
_messageType
) => {
awarenessProtocol.applyAwarenessUpdate(
provider.awareness,
decoding.readVarUint8Array(decoder),
provider
);
};
messageHandlers[messageAuth] = (
_encoder,
decoder,
provider,
_emitSynced,
_messageType
) => {
authProtocol.readAuthMessage(decoder, provider.doc, (_ydoc, reason) =>
permissionDeniedHandler(provider, reason)
);
};
// @todo - this should depend on awareness.outdatedTime
const messageReconnectTimeout = 30000;
/**
* @param {WebsocketProvider} provider
* @param {string} reason
*/
const permissionDeniedHandler = (provider, reason) =>
console.warn(`Permission denied to access ${provider.url}.\n${reason}`);
/**
* @param {WebsocketProvider} provider
* @param {Uint8Array} buf
* @param {boolean} emitSynced
* @return {encoding.Encoder}
*/
const readMessage = (provider, buf, emitSynced) => {
const decoder = decoding.createDecoder(buf);
const encoder = encoding.createEncoder();
const messageType = decoding.readVarUint(decoder);
const messageHandler = provider.messageHandlers[messageType];
if (/** @type {any} */ (messageHandler)) {
messageHandler(encoder, decoder, provider, emitSynced, messageType);
} else {
console.error('Unable to compute message');
}
return encoder;
};
/**
* @param {WebsocketProvider} provider
*/
const setupWS = provider => {
if (provider.shouldConnect && provider.ws === null) {
const websocket = new provider._WS(provider.url);
websocket.binaryType = 'arraybuffer';
provider.ws = websocket;
provider.wsconnecting = true;
provider.wsconnected = false;
provider.synced = false;
websocket.onmessage = event => {
provider.wsLastMessageReceived = time.getUnixTime();
const encoder = readMessage(provider, new Uint8Array(event.data), true);
if (encoding.length(encoder) > 1) {
websocket.send(encoding.toUint8Array(encoder));
}
};
websocket.onerror = event => {
provider.emit('connection-error', [event, provider]);
};
websocket.onclose = event => {
provider.emit('connection-close', [event, provider]);
provider.ws = null;
provider.wsconnecting = false;
if (provider.wsconnected) {
provider.wsconnected = false;
provider.synced = false;
// update awareness (all users except local left)
awarenessProtocol.removeAwarenessStates(
provider.awareness,
Array.from(provider.awareness.getStates().keys()).filter(
client => client !== provider.doc.clientID
),
provider
);
provider.emit('status', [
{
status: 'disconnected',
},
]);
} else {
provider.wsUnsuccessfulReconnects++;
}
// Start with no reconnect timeout and increase timeout by
// using exponential backoff starting with 100ms
setTimeout(
setupWS,
math.min(
math.pow(2, provider.wsUnsuccessfulReconnects) * 100,
provider.maxBackoffTime
),
provider
);
};
websocket.onopen = () => {
provider.wsLastMessageReceived = time.getUnixTime();
provider.wsconnecting = false;
provider.wsconnected = true;
provider.wsUnsuccessfulReconnects = 0;
provider.emit('status', [
{
status: 'connected',
},
]);
// always send sync step 1 when connected
const encoder = encoding.createEncoder();
encoding.writeVarUint(encoder, messageSync);
syncProtocol.writeSyncStep1(encoder, provider.doc);
websocket.send(encoding.toUint8Array(encoder));
// broadcast local awareness state
if (provider.awareness.getLocalState() !== null) {
const encoderAwarenessState = encoding.createEncoder();
encoding.writeVarUint(encoderAwarenessState, messageAwareness);
encoding.writeVarUint8Array(
encoderAwarenessState,
awarenessProtocol.encodeAwarenessUpdate(provider.awareness, [
provider.doc.clientID,
])
);
websocket.send(encoding.toUint8Array(encoderAwarenessState));
}
};
provider.emit('status', [
{
status: 'connecting',
},
]);
}
};
/**
* @param {WebsocketProvider} provider
* @param {ArrayBuffer} buf
*/
const broadcastMessage = (provider, buf) => {
if (provider.wsconnected) {
/** @type {WebSocket} */ (provider.ws).send(buf);
}
if (provider.bcconnected) {
bc.publish(provider.bcChannel, buf, provider);
}
};
/**
* Websocket Provider for Yjs. Creates a websocket connection to sync the shared document.
* The document name is attached to the provided url. I.e. the following example
* creates a websocket connection to http://localhost:1234/my-document-name
*
* @example
* import * as Y from 'yjs'
* import { WebsocketProvider } from 'y-websocket'
* const doc = new Y.Doc()
* const provider = new WebsocketProvider('http://localhost:1234', 'my-document-name', doc)
*
* @extends {Observable<string>}
*/
export class WebsocketProvider extends Observable {
/**
* @param {string} serverUrl
* @param {string} roomname
* @param {Y.Doc} doc
* @param {object} [opts]
* @param {boolean} [opts.connect]
* @param {awarenessProtocol.Awareness} [opts.awareness]
* @param {Object<string,string>} [opts.params]
* @param {typeof WebSocket} [opts.WebSocketPolyfill] Optionall provide a WebSocket polyfill
* @param {number} [opts.resyncInterval] Request server state every `resyncInterval` milliseconds
* @param {number} [opts.maxBackoffTime] Maximum amount of time to wait before trying to reconnect (we try to reconnect using exponential backoff)
* @param {boolean} [opts.disableBc] Disable cross-tab BroadcastChannel communication
*/
constructor(
serverUrl,
roomname,
doc,
{
connect = true,
awareness = new awarenessProtocol.Awareness(doc),
params = {},
WebSocketPolyfill = WebSocket,
resyncInterval = -1,
maxBackoffTime = 2500,
disableBc = false,
} = {}
) {
super();
// ensure that url is always ends with /
while (serverUrl[serverUrl.length - 1] === '/') {
serverUrl = serverUrl.slice(0, serverUrl.length - 1);
}
const encodedParams = url.encodeQueryParams(params);
this.maxBackoffTime = maxBackoffTime;
this.bcChannel = serverUrl + '/' + roomname;
this.url =
serverUrl +
'/' +
roomname +
(encodedParams.length === 0 ? '' : '?' + encodedParams);
this.roomname = roomname;
this.doc = doc;
this._WS = WebSocketPolyfill;
this.awareness = awareness;
this.wsconnected = false;
this.wsconnecting = false;
this.bcconnected = false;
this.disableBc = disableBc;
this.wsUnsuccessfulReconnects = 0;
this.messageHandlers = messageHandlers.slice();
/**
* @type {boolean}
*/
this._synced = false;
/**
* @type {WebSocket?}
*/
this.ws = null;
this.wsLastMessageReceived = 0;
/**
* Whether to connect to other peers or not
* @type {boolean}
*/
this.shouldConnect = connect;
/**
* @type {number}
*/
this._resyncInterval = 0;
if (resyncInterval > 0) {
this._resyncInterval = /** @type {any} */ (
setInterval(() => {
if (this.ws && this.ws.readyState === WebSocket.OPEN) {
// resend sync step 1
const encoder = encoding.createEncoder();
encoding.writeVarUint(encoder, messageSync);
syncProtocol.writeSyncStep1(encoder, doc);
this.ws.send(encoding.toUint8Array(encoder));
}
}, resyncInterval)
);
}
/**
* @param {ArrayBuffer} data
* @param {any} origin
*/
this._bcSubscriber = (data, origin) => {
if (origin !== this) {
const encoder = readMessage(this, new Uint8Array(data), false);
if (encoding.length(encoder) > 1) {
bc.publish(this.bcChannel, encoding.toUint8Array(encoder), this);
}
}
};
/**
* Listens to Yjs updates and sends them to remote peers (ws and broadcastchannel)
* @param {Uint8Array} update
* @param {any} origin
*/
this._updateHandler = (update, origin) => {
if (origin !== this) {
const encoder = encoding.createEncoder();
encoding.writeVarUint(encoder, messageSync);
syncProtocol.writeUpdate(encoder, update);
broadcastMessage(this, encoding.toUint8Array(encoder));
}
};
this.doc.on('update', this._updateHandler);
/**
* @param {any} changed
* @param {any} _origin
*/
this._awarenessUpdateHandler = ({ added, updated, removed }, _origin) => {
const changedClients = added.concat(updated).concat(removed);
const encoder = encoding.createEncoder();
encoding.writeVarUint(encoder, messageAwareness);
encoding.writeVarUint8Array(
encoder,
awarenessProtocol.encodeAwarenessUpdate(awareness, changedClients)
);
broadcastMessage(this, encoding.toUint8Array(encoder));
};
this._unloadHandler = () => {
awarenessProtocol.removeAwarenessStates(
this.awareness,
[doc.clientID],
'window unload'
);
};
if (typeof window !== 'undefined') {
window.addEventListener('unload', this._unloadHandler);
} else if (typeof process !== 'undefined') {
process.on('exit', this._unloadHandler);
}
awareness.on('update', this._awarenessUpdateHandler);
this._checkInterval = /** @type {any} */ (
setInterval(() => {
if (
this.wsconnected &&
messageReconnectTimeout <
time.getUnixTime() - this.wsLastMessageReceived
) {
// no message received in a long time - not even your own awareness
// updates (which are updated every 15 seconds)
/** @type {WebSocket} */ (this.ws).close();
}
}, messageReconnectTimeout / 10)
);
if (connect) {
this.connect();
}
}
/**
* @type {boolean}
*/
get synced() {
return this._synced;
}
set synced(state) {
if (this._synced !== state) {
this._synced = state;
this.emit('synced', [state]);
this.emit('sync', [state]);
}
}
destroy() {
if (this._resyncInterval !== 0) {
clearInterval(this._resyncInterval);
}
clearInterval(this._checkInterval);
this.disconnect();
if (typeof window !== 'undefined') {
window.removeEventListener('unload', this._unloadHandler);
} else if (typeof process !== 'undefined') {
process.off('exit', this._unloadHandler);
}
this.awareness.off('update', this._awarenessUpdateHandler);
this.doc.off('update', this._updateHandler);
super.destroy();
}
connectBc() {
if (this.disableBc) {
return;
}
if (!this.bcconnected) {
bc.subscribe(this.bcChannel, this._bcSubscriber);
this.bcconnected = true;
}
// send sync step1 to bc
// write sync step 1
const encoderSync = encoding.createEncoder();
encoding.writeVarUint(encoderSync, messageSync);
syncProtocol.writeSyncStep1(encoderSync, this.doc);
bc.publish(this.bcChannel, encoding.toUint8Array(encoderSync), this);
// broadcast local state
const encoderState = encoding.createEncoder();
encoding.writeVarUint(encoderState, messageSync);
syncProtocol.writeSyncStep2(encoderState, this.doc);
bc.publish(this.bcChannel, encoding.toUint8Array(encoderState), this);
// write queryAwareness
const encoderAwarenessQuery = encoding.createEncoder();
encoding.writeVarUint(encoderAwarenessQuery, messageQueryAwareness);
bc.publish(
this.bcChannel,
encoding.toUint8Array(encoderAwarenessQuery),
this
);
// broadcast local awareness state
const encoderAwarenessState = encoding.createEncoder();
encoding.writeVarUint(encoderAwarenessState, messageAwareness);
encoding.writeVarUint8Array(
encoderAwarenessState,
awarenessProtocol.encodeAwarenessUpdate(this.awareness, [
this.doc.clientID,
])
);
bc.publish(
this.bcChannel,
encoding.toUint8Array(encoderAwarenessState),
this
);
}
disconnectBc() {
// broadcast message with local awareness state set to null (indicating disconnect)
const encoder = encoding.createEncoder();
encoding.writeVarUint(encoder, messageAwareness);
encoding.writeVarUint8Array(
encoder,
awarenessProtocol.encodeAwarenessUpdate(
this.awareness,
[this.doc.clientID],
new Map()
)
);
broadcastMessage(this, encoding.toUint8Array(encoder));
if (this.bcconnected) {
bc.unsubscribe(this.bcChannel, this._bcSubscriber);
this.bcconnected = false;
}
}
disconnect() {
this.shouldConnect = false;
this.disconnectBc();
if (this.ws !== null) {
this.ws.close();
}
}
connect() {
this.shouldConnect = true;
if (!this.wsconnected && this.ws === null) {
setupWS(this);
this.connectBc();
}
}
}
+1 -1
View File
@@ -1,4 +1,4 @@
export { request } from './axios';
export { ServiceError } from './ServiceError';
export { setToken, authorizationEvent } from './request';
export { setToken, getToken, authorizationEvent } from './request';
export type { AccessTokenMessage } from './request';
@@ -28,7 +28,10 @@ export function setToken(token: Token | null): void {
authorizationEvent.triggerChange(parseAccessToken(token.accessToken));
}
function getToken(): { accessToken: string; refreshToken: string } | null {
export function getToken(): {
accessToken: string;
refreshToken: string;
} | null {
try {
return JSON.parse(window.localStorage.getItem(TOKEN_KEY) || '');
} catch (error) {
@@ -1,6 +1,7 @@
export {
setAuthorization,
setToken,
getToken,
authorizationEvent,
} from './authorization';
export type { AccessTokenMessage } from './authorization';
+13 -9
View File
@@ -56,6 +56,7 @@ importers:
eslint-config-prettier: ^8.5.0
eslint-plugin-prettier: ^4.2.1
i18next: ^21.9.1
lib0: ^0.2.58
lit: ^2.3.1
next: 13.0.1
prettier: ^2.7.1
@@ -65,6 +66,7 @@ importers:
react-dom: 18.2.0
react-i18next: ^11.18.4
typescript: 4.8.3
y-protocols: ^1.0.5
dependencies:
'@blocksuite/blocks': 0.3.0-20221221064254-e3ebf2f
'@blocksuite/editor': 0.3.0-20221221064254-e3ebf2f
@@ -85,6 +87,7 @@ importers:
css-spring: 4.1.0
dayjs: 1.11.7
i18next: 21.10.0
lib0: 0.2.58
lit: 2.4.0
next: 13.0.1_biqbaboplfbrettd7655fr4n2y
prettier: 2.7.1
@@ -93,6 +96,7 @@ importers:
react: 18.2.0
react-dom: 18.2.0_react@18.2.0
react-i18next: 11.18.6_vfm63zmruocgezzfl2v26zlzpy
y-protocols: 1.0.5
devDependencies:
'@types/node': 18.7.18
'@types/react': 18.0.20
@@ -549,7 +553,7 @@ packages:
flexsearch: 0.7.21
idb-keyval: 6.2.0
ky: 0.32.2
lib0: 0.2.52
lib0: 0.2.58
y-indexeddb: 9.0.9_yjs@13.5.41
y-protocols: 1.0.5
y-webrtc: 10.2.3
@@ -3295,7 +3299,7 @@ packages:
eslint-import-resolver-webpack:
optional: true
dependencies:
'@typescript-eslint/parser': 5.38.0_tg6quxtr5dyl3tikvj7rwonxxi
'@typescript-eslint/parser': 5.38.0_76twfck5d7crjqrmw4yltga7zm
debug: 3.2.7
eslint: 8.22.0
eslint-import-resolver-node: 0.3.6
@@ -3314,7 +3318,7 @@ packages:
'@typescript-eslint/parser':
optional: true
dependencies:
'@typescript-eslint/parser': 5.38.0_tg6quxtr5dyl3tikvj7rwonxxi
'@typescript-eslint/parser': 5.38.0_76twfck5d7crjqrmw4yltga7zm
array-includes: 3.1.5
array.prototype.flat: 1.3.0
debug: 2.6.9
@@ -4698,8 +4702,8 @@ packages:
type-check: 0.4.0
dev: true
/lib0/0.2.52:
resolution: {integrity: sha512-CjxlM7UgICfN6b2OPALBXchIBiNk6jE+1g7JP8ha+dh1xKRDSYpH0WQl1+rMqCju49xUnwPG34v4CR5/rPOZhg==}
/lib0/0.2.58:
resolution: {integrity: sha512-6ovqPaYfOKU7GkkVxz/wjMR0zsqmNsISLvH+h9Lx5YNtWDZey69aYsTGXaSVpUPpJ+ZFtIvcZHsTGL3MbwOM8A==}
engines: {node: '>=14'}
dependencies:
isomorphic.js: 0.2.5
@@ -6205,14 +6209,14 @@ packages:
peerDependencies:
yjs: ^13.0.0
dependencies:
lib0: 0.2.52
lib0: 0.2.58
yjs: 13.5.41
dev: false
/y-protocols/1.0.5:
resolution: {integrity: sha512-Wil92b7cGk712lRHDqS4T90IczF6RkcvCwAD0A2OPg+adKmOe+nOiT/N2hvpQIWS3zfjmtL4CPaH5sIW1Hkm/A==}
dependencies:
lib0: 0.2.52
lib0: 0.2.58
dev: false
/y-webrtc/10.2.3:
@@ -6220,7 +6224,7 @@ packages:
engines: {node: '>=12'}
hasBin: true
dependencies:
lib0: 0.2.52
lib0: 0.2.58
simple-peer: 9.11.1
y-protocols: 1.0.5
optionalDependencies:
@@ -6283,7 +6287,7 @@ packages:
/yjs/13.5.41:
resolution: {integrity: sha512-4eSTrrs8OeI0heXKKioRY4ag7V5Bk85Z4MeniUyown3o3y0G7G4JpAZWrZWfTp7pzw2b53GkAQWKqHsHi9j9JA==}
dependencies:
lib0: 0.2.52
lib0: 0.2.58
dev: false
/yocto-queue/0.1.0: