Compare commits
10 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
| f716c498e5 | |||
| 29a4147b5e | |||
| 15389156d3 | |||
| 168904728b | |||
| 3ca07e0f4c | |||
| 6098fe1e14 | |||
| a2f1e32f21 | |||
| f27e968c15 | |||
| 971b7e099d | |||
| 6d0ccb15ea |
@@ -6,6 +6,7 @@ on:
|
||||
push:
|
||||
branches:
|
||||
- 'main'
|
||||
- 'feature/collab-long-polling'
|
||||
tags:
|
||||
- 'v*'
|
||||
pull_request:
|
||||
|
||||
+5
-1
@@ -8,17 +8,21 @@ and this project adheres to
|
||||
|
||||
## [Unreleased]
|
||||
|
||||
## Added
|
||||
|
||||
- ✨Collaboration long polling fallback #517
|
||||
|
||||
## Changed
|
||||
|
||||
- 🛂(frontend) Restore version visibility #629
|
||||
- 📝(doc) minor README.md formatting and wording enhancements
|
||||
- ♻️Stop setting a default title on doc creation #634
|
||||
- 📝(readme) remove front-end local run instructions local.md #651
|
||||
|
||||
## Fixed
|
||||
|
||||
- ♻️(frontend) improve table pdf rendering
|
||||
|
||||
|
||||
## [2.2.0] - 2025-02-10
|
||||
|
||||
## Added
|
||||
|
||||
@@ -40,7 +40,7 @@ Docs is a collaborative text editor designed to address common challenges in kno
|
||||
* 📚 Built-in wiki functionality to turn your team's collaborative work into organized knowledge `ETA 02/2025`
|
||||
|
||||
### Self-host
|
||||
* 🚀 Easy to install, scalable and secure alternative to Notion and Outline.
|
||||
* 🚀 Easy to install, scalable and secure alternative to Notion, Outline or Confluence
|
||||
|
||||
## Getting started 🔧
|
||||
|
||||
@@ -100,6 +100,26 @@ password: impress
|
||||
$ make run
|
||||
```
|
||||
|
||||
⚠️ For the frontend developer, it is often better to run the frontend in development mode locally.
|
||||
|
||||
To do so, install the frontend dependencies with the following command:
|
||||
|
||||
```shellscript
|
||||
$ make frontend-development-install
|
||||
```
|
||||
|
||||
And run the frontend locally in development mode with the following command:
|
||||
|
||||
```shellscript
|
||||
$ make run-frontend-development
|
||||
```
|
||||
|
||||
To start all the services, except the frontend container, you can use the following command:
|
||||
|
||||
```shellscript
|
||||
$ make run-backend
|
||||
```
|
||||
|
||||
**Adding content**
|
||||
You can create a basic demo site by running:
|
||||
|
||||
|
||||
@@ -0,0 +1 @@
|
||||
proxy_cache_path /tmp/auth_cache levels=1:2 keys_zone=auth_cache:10m inactive=60s max_size=100m;
|
||||
@@ -1,4 +1,3 @@
|
||||
|
||||
server {
|
||||
listen 8083;
|
||||
server_name localhost;
|
||||
@@ -6,6 +5,14 @@ server {
|
||||
|
||||
# Proxy auth for collaboration server
|
||||
location /collaboration/ws/ {
|
||||
if ($request_method = OPTIONS) {
|
||||
add_header 'Access-Control-Allow-Origin' 'http://localhost:3000';
|
||||
add_header 'Access-Control-Allow-Credentials' 'true';
|
||||
add_header 'Access-Control-Allow-Methods' 'GET, POST, OPTIONS';
|
||||
add_header 'Access-Control-Allow-Headers' 'Authorization, Content-Type';
|
||||
return 204;
|
||||
}
|
||||
|
||||
# Collaboration Auth request configuration
|
||||
auth_request /collaboration-auth;
|
||||
auth_request_set $authHeader $upstream_http_authorization;
|
||||
@@ -34,6 +41,10 @@ server {
|
||||
}
|
||||
|
||||
location /collaboration-auth {
|
||||
proxy_cache auth_cache;
|
||||
proxy_cache_key "$http_authorization-$arg_room";
|
||||
proxy_cache_valid 200 30s;
|
||||
|
||||
proxy_pass http://app-dev:8000/api/v1.0/documents/collaboration-auth/;
|
||||
proxy_set_header Host $host;
|
||||
proxy_set_header X-Real-IP $remote_addr;
|
||||
@@ -43,10 +54,11 @@ server {
|
||||
# Prevent the body from being passed
|
||||
proxy_pass_request_body off;
|
||||
proxy_set_header Content-Length "";
|
||||
proxy_set_header Accept "application/json";
|
||||
proxy_set_header X-Original-Method $request_method;
|
||||
}
|
||||
|
||||
location /collaboration/api/ {
|
||||
location /collaboration/api/ {
|
||||
# Collaboration server
|
||||
proxy_pass http://y-provider:4444;
|
||||
proxy_set_header Host $host;
|
||||
@@ -76,7 +88,7 @@ server {
|
||||
proxy_set_header X-Real-IP $remote_addr;
|
||||
proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
|
||||
proxy_set_header X-Original-URL $request_uri;
|
||||
|
||||
|
||||
# Prevent the body from being passed
|
||||
proxy_pass_request_body off;
|
||||
proxy_set_header Content-Length "";
|
||||
|
||||
@@ -0,0 +1,84 @@
|
||||
# Architecture Overview
|
||||
|
||||
This architecture showcases different ways for clients to interact with a **Hocus Pocus Server** (a [Y.js](https://github.com/yjs/yjs) provider) through either WebSockets, HTTP fallbacks, or Server-Sent Events (SSE) when WebSockets are not available.
|
||||
|
||||
**Main Components**:
|
||||
|
||||
- **Client**: The front-end application or user agent.
|
||||
- **Nginx**: A reverse proxy handling incoming requests, forwarding them to the appropriate services, and managing SSL/TLS termination if needed.
|
||||
- **Auth Sub Request (Django)**: Handles authentication/authorization, ensuring requests have valid credentials or permissions.
|
||||
- **Hocus Pocus Server**: The core collaborative editing server (powered by [Y.js](https://github.com/yjs/yjs) libraries) that manages document state and synchronization.
|
||||
- **Express**: Fallback server to handle push or pull requests when WebSocket connections fail.
|
||||
- **SSE**: A mechanism (Server-Sent Events) for real-time updates when WebSockets are unavailable.
|
||||
|
||||
## Mermaid Diagram
|
||||
|
||||
```mermaid
|
||||
flowchart TD
|
||||
title1[WebSocket Success]-->Client1(Client)<--->|WebSocket Success|WS1(Websocket) --> Nginx1(Ngnix) <--> Auth1("Auth Sub Request (Django)") --->|With the good right|YServer1("Hocus Pocus Server")
|
||||
YServer1 --> WS1
|
||||
YServer1 <--> clients(Dispatch to clients)
|
||||
|
||||
title2[WebSocket Fails - Push data]-->Client2(Client)---|WebSocket fails|HTTP2(HTTP) --> Nginx2(Ngnix) <--> Auth2("Auth Sub Request (Django)")--->|With the good right|Express2(Express) --> YServer2("Hocus Pocus Server") --> clients(Dispatch to clients)
|
||||
|
||||
title3[WebSocket Fails - Pull data]-->Client3(Client)<--->|WebSocket fails|SSE(SSE) --> Nginx3(Ngnix) <--> Auth3("Auth Sub Request (Django)") --->|With the good right|Express3(Express) --> YServer3("Listen Hocus Pocus Server")
|
||||
YServer3("Listen Hocus Pocus Server") --> SSE
|
||||
YServer3("Listen Hocus Pocus Server") <--> clients(Data from clients)
|
||||
```
|
||||
|
||||
---
|
||||
|
||||
## Detailed Flows
|
||||
|
||||
### 1. WebSocket Success
|
||||
1. **Client** attempts a WebSocket connection.
|
||||
2. **Nginx** proxies the WebSocket connection through the **Auth Sub Request (Django)** for authentication.
|
||||
3. Once authenticated, traffic is routed to the **Hocus Pocus Server**.
|
||||
4. The server can broadcast data to all clients connected through WebSockets.
|
||||
- Note: The path `YServer1 --> WS1` indicates the two-way real-time communication between the server and client(s).
|
||||
|
||||
### 2. WebSocket Fails — Push Data (HTTP)
|
||||
If WebSocket connections fail, clients can **push** data via HTTP:
|
||||
1. **Client** detects WebSocket failure and falls back to sending data over **HTTP**.
|
||||
2. **Nginx** handles HTTP requests and authenticates them via the **Auth Sub Request (Django)**.
|
||||
3. After successful authentication, the requests go to an **Express** server.
|
||||
4. The **Express** server relays changes to the **Hocus Pocus Server**.
|
||||
5. The **Hocus Pocus Server** dispatches updated content to connected clients.
|
||||
|
||||
### 3. WebSocket Fails — Pull Data (SSE)
|
||||
For continuously receiving data when WebSockets fail, the client can **pull** data using SSE:
|
||||
1. **Client** sets up an **SSE** connection.
|
||||
2. **Nginx** proxies the SSE stream request through the **Auth Sub Request (Django)** for authentication.
|
||||
3. Once authenticated, the **Express** server listens to the **Hocus Pocus Server** for changes.
|
||||
4. The server then sends updates back to the **Client** through SSE in near real-time.
|
||||
|
||||
---
|
||||
|
||||
## Component Responsibilities
|
||||
|
||||
| **Component** | **Responsibility** |
|
||||
|-----------------------------|-----------------------------------------------------------------------------------------|
|
||||
| **Client** | Initiates connections (WebSocket/HTTP/SSE), displays and interacts with data |
|
||||
| **Nginx** | Acts as a reverse proxy, routes traffic, handles SSL, and passes auth sub requests |
|
||||
| **Auth Sub Request (Django)** | Validates requests, ensuring correct permissions and tokens |
|
||||
| **WebSocket** | Real-time two-way communication channel |
|
||||
| **HTTP** | Fallback method for sending updates when WebSockets are not available |
|
||||
| **Express** | Fallback server for handling requests (push/pull of data) |
|
||||
| **SSE** | Mechanism for real-time one-way updates from server to client |
|
||||
| **Hocus Pocus Server** | Core Y.js server for collaboration, managing document states and synchronization |
|
||||
|
||||
---
|
||||
|
||||
## Why This Setup?
|
||||
|
||||
- **Reliability:** Ensures that when a user’s browser or network environment does not support WebSockets, there are fallback mechanisms (HTTP for push updates and SSE for server-initiated updates).
|
||||
- **Scalability:** Nginx can efficiently proxy requests and scale horizontally, while the authentication step is centralized in Django.
|
||||
- **Security:** The Auth Sub Request in Django enforces proper permissions before data is relayed to the collaboration server.
|
||||
- **Real-time Collaboration:** The Hocus Pocus Server provides low-latency updates, essential for collaborative editing, supported by [Y.js](https://github.com/yjs/yjs).
|
||||
|
||||
---
|
||||
|
||||
### Contributing
|
||||
If you have any suggestions or improvements, feel free to open an issue or submit a pull request.
|
||||
|
||||
**Thank you for exploring this architecture!** If you have any questions or need more detailed explanations, please let us know.
|
||||
@@ -1,92 +0,0 @@
|
||||
# Run Docs locally
|
||||
|
||||
> ⚠️ Running Docs locally using the methods described below is for testing purposes only. It is based on building Docs using Minio as the S3 storage solution: if you want to use Minio for production deployment of Docs, you will need to comply with Minio's AGPL-3.0 licence.
|
||||
|
||||
**Prerequisite**
|
||||
|
||||
Make sure you have a recent version of Docker and [Docker Compose](https://docs.docker.com/compose/install) installed on your laptop:
|
||||
|
||||
```shellscript
|
||||
$ docker -v
|
||||
|
||||
Docker version 20.10.2, build 2291f61
|
||||
|
||||
$ docker compose version
|
||||
|
||||
Docker Compose version v2.32.4
|
||||
```
|
||||
|
||||
> ⚠️ You may need to run the following commands with sudo but this can be avoided by adding your user to the `docker` group.
|
||||
|
||||
**Project bootstrap**
|
||||
|
||||
The easiest way to start working on the project is to use GNU Make:
|
||||
|
||||
```shellscript
|
||||
$ make bootstrap FLUSH_ARGS='--no-input'
|
||||
```
|
||||
|
||||
This command builds the `app` container, installs dependencies, performs database migrations and compile translations. It's a good idea to use this command each time you are pulling code from the project repository to avoid dependency-related or migration-related issues.
|
||||
|
||||
Your Docker services should now be up and running 🎉
|
||||
|
||||
You can access to the project by going to <http://localhost:3000>.
|
||||
|
||||
You will be prompted to log in, the default credentials are:
|
||||
|
||||
```
|
||||
username: impress
|
||||
password: impress
|
||||
```
|
||||
|
||||
📝 Note that if you need to run them afterwards, you can use the eponym Make rule:
|
||||
|
||||
```shellscript
|
||||
$ make run
|
||||
```
|
||||
|
||||
**Adding content**
|
||||
You can create a basic demo site by running:
|
||||
|
||||
```shellscript
|
||||
$ make demo
|
||||
```
|
||||
|
||||
Finally, you can check all available Make rules using:
|
||||
|
||||
```shellscript
|
||||
$ make help
|
||||
```
|
||||
|
||||
**Django admin**
|
||||
|
||||
You can access the Django admin site at
|
||||
|
||||
<http://localhost:8071/admin>.
|
||||
|
||||
You first need to create a superuser account:
|
||||
|
||||
```shellscript
|
||||
$ make superuser
|
||||
```
|
||||
|
||||
## Front-end dev instructions
|
||||
⚠️ For the frontend developer, it is often better to run the frontend in development mode locally.
|
||||
|
||||
To do so, install the frontend dependencies with the following command:
|
||||
|
||||
```shellscript
|
||||
$ make frontend-development-install
|
||||
```
|
||||
|
||||
And run the frontend locally in development mode with the following command:
|
||||
|
||||
```shellscript
|
||||
$ make run-frontend-development
|
||||
```
|
||||
|
||||
To start all the services, except the frontend container, you can use the following command:
|
||||
|
||||
```shellscript
|
||||
$ make run-backend
|
||||
```
|
||||
@@ -0,0 +1,97 @@
|
||||
import { expect, test } from '@playwright/test';
|
||||
|
||||
import { createDoc, verifyDocName } from './common';
|
||||
|
||||
test.beforeEach(async ({ page }) => {
|
||||
await page.goto('/');
|
||||
});
|
||||
|
||||
test.describe('Doc Collaboration', () => {
|
||||
/**
|
||||
* We check:
|
||||
* - connection to the collaborative server
|
||||
* - signal of the backend to the collaborative server (connection should close)
|
||||
* - reconnection to the collaborative server
|
||||
*/
|
||||
test('checks the connection with collaborative server', async ({
|
||||
page,
|
||||
browserName,
|
||||
}) => {
|
||||
let webSocketPromise = page.waitForEvent('websocket', (webSocket) => {
|
||||
return webSocket
|
||||
.url()
|
||||
.includes('ws://localhost:8083/collaboration/ws/?room=');
|
||||
});
|
||||
|
||||
const [title] = await createDoc(page, 'doc-editor', browserName, 1);
|
||||
await verifyDocName(page, title);
|
||||
|
||||
let webSocket = await webSocketPromise;
|
||||
expect(webSocket.url()).toContain(
|
||||
'ws://localhost:8083/collaboration/ws/?room=',
|
||||
);
|
||||
|
||||
// Is connected
|
||||
let framesentPromise = webSocket.waitForEvent('framesent');
|
||||
|
||||
await page.locator('.ProseMirror.bn-editor').click();
|
||||
await page.locator('.ProseMirror.bn-editor').fill('Hello World');
|
||||
|
||||
let framesent = await framesentPromise;
|
||||
expect(framesent.payload).not.toBeNull();
|
||||
|
||||
await page.getByRole('button', { name: 'Share' }).click();
|
||||
|
||||
const selectVisibility = page.getByLabel('Visibility', { exact: true });
|
||||
|
||||
// When the visibility is changed, the ws should closed the connection (backend signal)
|
||||
const wsClosePromise = webSocket.waitForEvent('close');
|
||||
|
||||
await selectVisibility.click();
|
||||
await page
|
||||
.getByRole('button', {
|
||||
name: 'Connected',
|
||||
})
|
||||
.click();
|
||||
|
||||
// Assert that the doc reconnects to the ws
|
||||
const wsClose = await wsClosePromise;
|
||||
expect(wsClose.isClosed()).toBeTruthy();
|
||||
|
||||
// Checkt the ws is connected again
|
||||
webSocketPromise = page.waitForEvent('websocket', (webSocket) => {
|
||||
return webSocket
|
||||
.url()
|
||||
.includes('ws://localhost:8083/collaboration/ws/?room=');
|
||||
});
|
||||
|
||||
webSocket = await webSocketPromise;
|
||||
framesentPromise = webSocket.waitForEvent('framesent');
|
||||
framesent = await framesentPromise;
|
||||
expect(framesent.payload).not.toBeNull();
|
||||
});
|
||||
|
||||
test('checks the connection switch to polling after websocket failure', async ({
|
||||
page,
|
||||
browserName,
|
||||
}) => {
|
||||
const responsePromise = page.waitForResponse(
|
||||
(response) =>
|
||||
response.url().includes('/poll/') && response.status() === 200,
|
||||
);
|
||||
|
||||
await page.routeWebSocket(
|
||||
'ws://localhost:8083/collaboration/ws/**',
|
||||
async (ws) => {
|
||||
await ws.close();
|
||||
},
|
||||
);
|
||||
|
||||
await page.reload();
|
||||
|
||||
await createDoc(page, 'doc-polling', browserName, 1);
|
||||
|
||||
const response = await responsePromise;
|
||||
expect(response.ok()).toBeTruthy();
|
||||
});
|
||||
});
|
||||
@@ -88,70 +88,6 @@ test.describe('Doc Editor', () => {
|
||||
).toBeVisible();
|
||||
});
|
||||
|
||||
/**
|
||||
* We check:
|
||||
* - connection to the collaborative server
|
||||
* - signal of the backend to the collaborative server (connection should close)
|
||||
* - reconnection to the collaborative server
|
||||
*/
|
||||
test('checks the connection with collaborative server', async ({
|
||||
page,
|
||||
browserName,
|
||||
}) => {
|
||||
let webSocketPromise = page.waitForEvent('websocket', (webSocket) => {
|
||||
return webSocket
|
||||
.url()
|
||||
.includes('ws://localhost:8083/collaboration/ws/?room=');
|
||||
});
|
||||
|
||||
const randomDoc = await createDoc(page, 'doc-editor', browserName, 1);
|
||||
await verifyDocName(page, randomDoc[0]);
|
||||
|
||||
let webSocket = await webSocketPromise;
|
||||
expect(webSocket.url()).toContain(
|
||||
'ws://localhost:8083/collaboration/ws/?room=',
|
||||
);
|
||||
|
||||
// Is connected
|
||||
let framesentPromise = webSocket.waitForEvent('framesent');
|
||||
|
||||
await page.locator('.ProseMirror.bn-editor').click();
|
||||
await page.locator('.ProseMirror.bn-editor').fill('Hello World');
|
||||
|
||||
let framesent = await framesentPromise;
|
||||
expect(framesent.payload).not.toBeNull();
|
||||
|
||||
await page.getByRole('button', { name: 'Share' }).click();
|
||||
|
||||
const selectVisibility = page.getByLabel('Visibility', { exact: true });
|
||||
|
||||
// When the visibility is changed, the ws should closed the connection (backend signal)
|
||||
const wsClosePromise = webSocket.waitForEvent('close');
|
||||
|
||||
await selectVisibility.click();
|
||||
await page
|
||||
.getByRole('button', {
|
||||
name: 'Connected',
|
||||
})
|
||||
.click();
|
||||
|
||||
// Assert that the doc reconnects to the ws
|
||||
const wsClose = await wsClosePromise;
|
||||
expect(wsClose.isClosed()).toBeTruthy();
|
||||
|
||||
// Checkt the ws is connected again
|
||||
webSocketPromise = page.waitForEvent('websocket', (webSocket) => {
|
||||
return webSocket
|
||||
.url()
|
||||
.includes('ws://localhost:8083/collaboration/ws/?room=');
|
||||
});
|
||||
|
||||
webSocket = await webSocketPromise;
|
||||
framesentPromise = webSocket.waitForEvent('framesent');
|
||||
framesent = await framesentPromise;
|
||||
expect(framesent.payload).not.toBeNull();
|
||||
});
|
||||
|
||||
test('markdown button converts from markdown to the editor syntax json', async ({
|
||||
page,
|
||||
browserName,
|
||||
|
||||
@@ -62,6 +62,7 @@
|
||||
"@types/node": "*",
|
||||
"@types/react": "18.3.12",
|
||||
"@types/react-dom": "*",
|
||||
"@types/ws": "8.5.13",
|
||||
"cross-env": "7.0.3",
|
||||
"dotenv": "16.4.7",
|
||||
"eslint-config-impress": "*",
|
||||
|
||||
@@ -2,12 +2,10 @@ import { useRouter } from 'next/router';
|
||||
import { useCallback, useEffect, useRef, useState } from 'react';
|
||||
import * as Y from 'yjs';
|
||||
|
||||
import { useUpdateDoc } from '@/features/docs/doc-management/';
|
||||
import { toBase64, useUpdateDoc } from '@/features/docs/doc-management/';
|
||||
import { KEY_LIST_DOC_VERSIONS } from '@/features/docs/doc-versioning';
|
||||
import { isFirefox } from '@/utils/userAgent';
|
||||
|
||||
import { toBase64 } from '../utils';
|
||||
|
||||
const useSaveDoc = (docId: string, doc: Y.Doc, canSave: boolean) => {
|
||||
const { mutate: updateDoc } = useUpdateDoc({
|
||||
listInvalideQueries: [KEY_LIST_DOC_VERSIONS],
|
||||
|
||||
@@ -22,6 +22,3 @@ function hslToHex(h: number, s: number, l: number) {
|
||||
};
|
||||
return `#${f(0)}${f(8)}${f(4)}`;
|
||||
}
|
||||
|
||||
export const toBase64 = (str: Uint8Array) =>
|
||||
Buffer.from(str).toString('base64');
|
||||
|
||||
+67
@@ -0,0 +1,67 @@
|
||||
import { APIError, errorCauses } from '@/api';
|
||||
|
||||
interface PollOutgoingMessageParams {
|
||||
pollUrl: string;
|
||||
message64: string;
|
||||
}
|
||||
interface PollOutgoingMessageResponse {
|
||||
updated?: boolean;
|
||||
}
|
||||
|
||||
export const pollOutgoingMessageRequest = async ({
|
||||
pollUrl,
|
||||
message64,
|
||||
}: PollOutgoingMessageParams): Promise<PollOutgoingMessageResponse> => {
|
||||
const response = await fetch(pollUrl, {
|
||||
method: 'POST',
|
||||
credentials: 'include',
|
||||
headers: {
|
||||
'Content-Type': 'application/json',
|
||||
},
|
||||
body: JSON.stringify({
|
||||
message64,
|
||||
}),
|
||||
});
|
||||
|
||||
if (!response.ok) {
|
||||
throw new APIError(
|
||||
`Post poll message request failed`,
|
||||
await errorCauses(response),
|
||||
);
|
||||
}
|
||||
|
||||
return response.json() as Promise<PollOutgoingMessageResponse>;
|
||||
};
|
||||
|
||||
interface PollSyncParams {
|
||||
pollUrl: string;
|
||||
localDoc64: string;
|
||||
}
|
||||
interface PollSyncResponse {
|
||||
syncDoc64?: string;
|
||||
}
|
||||
|
||||
export const postPollSyncRequest = async ({
|
||||
pollUrl,
|
||||
localDoc64,
|
||||
}: PollSyncParams): Promise<PollSyncResponse> => {
|
||||
const response = await fetch(pollUrl, {
|
||||
method: 'POST',
|
||||
credentials: 'include',
|
||||
headers: {
|
||||
'Content-Type': 'application/json',
|
||||
},
|
||||
body: JSON.stringify({
|
||||
localDoc64,
|
||||
}),
|
||||
});
|
||||
|
||||
if (!response.ok) {
|
||||
throw new APIError(
|
||||
`Sync request failed: ${response.status} ${response.statusText}`,
|
||||
await errorCauses(response),
|
||||
);
|
||||
}
|
||||
|
||||
return response.json() as Promise<PollSyncResponse>;
|
||||
};
|
||||
+16
-3
@@ -6,17 +6,29 @@ import { useBroadcastStore } from '@/stores';
|
||||
import { useProviderStore } from '../stores/useProviderStore';
|
||||
import { Base64 } from '../types';
|
||||
|
||||
export const useCollaboration = (room?: string, initialContent?: Base64) => {
|
||||
export const useCollaboration = (
|
||||
room?: string,
|
||||
initialContent?: Base64,
|
||||
canEdit?: boolean,
|
||||
) => {
|
||||
const collaborationUrl = useCollaborationUrl(room);
|
||||
const { setBroadcastProvider } = useBroadcastStore();
|
||||
const { provider, createProvider, destroyProvider } = useProviderStore();
|
||||
|
||||
/**
|
||||
* Initialize the provider
|
||||
*/
|
||||
useEffect(() => {
|
||||
if (!room || !collaborationUrl || provider) {
|
||||
if (!room || !collaborationUrl || provider || canEdit === undefined) {
|
||||
return;
|
||||
}
|
||||
|
||||
const newProvider = createProvider(collaborationUrl, room, initialContent);
|
||||
const newProvider = createProvider(
|
||||
collaborationUrl,
|
||||
room,
|
||||
canEdit,
|
||||
initialContent,
|
||||
);
|
||||
setBroadcastProvider(newProvider);
|
||||
}, [
|
||||
provider,
|
||||
@@ -25,6 +37,7 @@ export const useCollaboration = (room?: string, initialContent?: Base64) => {
|
||||
initialContent,
|
||||
createProvider,
|
||||
setBroadcastProvider,
|
||||
canEdit,
|
||||
]);
|
||||
|
||||
/**
|
||||
|
||||
+333
@@ -0,0 +1,333 @@
|
||||
import crypto from 'crypto';
|
||||
|
||||
import {
|
||||
CompleteHocuspocusProviderConfiguration,
|
||||
CompleteHocuspocusProviderWebsocketConfiguration,
|
||||
HocuspocusProvider,
|
||||
HocuspocusProviderConfiguration,
|
||||
WebSocketStatus,
|
||||
onOutgoingMessageParameters,
|
||||
onStatusParameters,
|
||||
} from '@hocuspocus/provider';
|
||||
import type { MessageEvent } from 'ws';
|
||||
import * as Y from 'yjs';
|
||||
|
||||
import { isAPIError } from '@/api';
|
||||
|
||||
import {
|
||||
pollOutgoingMessageRequest,
|
||||
postPollSyncRequest,
|
||||
} from '../api/collaborationRequests';
|
||||
import { toBase64 } from '../utils';
|
||||
|
||||
type HocuspocusProviderConfigurationUrl = Required<
|
||||
Pick<CompleteHocuspocusProviderConfiguration, 'name'>
|
||||
> &
|
||||
Partial<CompleteHocuspocusProviderConfiguration> &
|
||||
Required<Pick<CompleteHocuspocusProviderWebsocketConfiguration, 'url'>>;
|
||||
|
||||
export const isHocuspocusProviderConfigurationUrl = (
|
||||
data: HocuspocusProviderConfiguration,
|
||||
): data is HocuspocusProviderConfigurationUrl => {
|
||||
return 'url' in data;
|
||||
};
|
||||
|
||||
type CollaborationProviderConfiguration = HocuspocusProviderConfiguration & {
|
||||
canEdit: boolean;
|
||||
};
|
||||
|
||||
export class CollaborationProvider extends HocuspocusProvider {
|
||||
/**
|
||||
* If the user can edit the document
|
||||
*/
|
||||
public canEdit = false;
|
||||
|
||||
/**
|
||||
* If the long polling is started
|
||||
* it is used to avoid starting it multiple times
|
||||
* when the websocket is failed.
|
||||
*/
|
||||
public isLongPollingStarted = false;
|
||||
|
||||
/**
|
||||
* If the document is syncing with the server
|
||||
* it is used to avoid starting it multiple times.
|
||||
*/
|
||||
public isSyncing = false;
|
||||
/**
|
||||
* The document can pass out of sync
|
||||
* then sync again with a next updates so
|
||||
* we add a counter to avoid syncing the document
|
||||
* to quickly.
|
||||
*/
|
||||
public seemsUnsyncCount = 0;
|
||||
public seemsUnsyncMaxCount = 5;
|
||||
|
||||
/**
|
||||
* In Safari or Firefox the websocket takes time before passing in
|
||||
* mode failed, it can takes up to 1 minutes. To avoid this latence
|
||||
* we set isWebsocketFailed to true, it is connects it will switch
|
||||
* to false.
|
||||
*/
|
||||
public isWebsocketFailed = true;
|
||||
/**
|
||||
* There is a ping-pong mechanism with awareness, receipt awareness is send again,
|
||||
* it creates useless requests.
|
||||
* We use this variable to avoid treating the same awareness message twice.
|
||||
*/
|
||||
private treatedAwarenessMessage: string | null = null;
|
||||
/**
|
||||
* Server-Sent Events
|
||||
*/
|
||||
protected sse: EventSource | null = null;
|
||||
/**
|
||||
* Polling timeout
|
||||
* It is used to avoid starting the polling to quickly
|
||||
* to let the class init properly.
|
||||
*/
|
||||
protected pollTimeout: NodeJS.Timeout | null = null;
|
||||
/**
|
||||
* Easy way to get the url of the server
|
||||
*/
|
||||
protected url = '';
|
||||
|
||||
public constructor(configuration: CollaborationProviderConfiguration) {
|
||||
let url = '';
|
||||
if (isHocuspocusProviderConfigurationUrl(configuration)) {
|
||||
url = configuration.url;
|
||||
let withWS = true;
|
||||
if (
|
||||
new URLSearchParams(window.location.search).get('withoutWS') === 'true'
|
||||
) {
|
||||
withWS = false;
|
||||
}
|
||||
|
||||
configuration.url = !withWS ? 'ws://localhost:6666' : configuration.url;
|
||||
}
|
||||
|
||||
super(configuration);
|
||||
|
||||
this.url = url;
|
||||
this.canEdit = configuration.canEdit;
|
||||
|
||||
if (configuration.canEdit) {
|
||||
this.on('outgoingMessage', this.onPollOutgoingMessage.bind(this));
|
||||
}
|
||||
}
|
||||
|
||||
public setPollDefaultValues(): void {
|
||||
this.isLongPollingStarted = false;
|
||||
this.isWebsocketFailed = false;
|
||||
this.seemsUnsyncCount = 0;
|
||||
this.sse?.close();
|
||||
this.sse = null;
|
||||
if (this.pollTimeout) {
|
||||
clearTimeout(this.pollTimeout);
|
||||
}
|
||||
}
|
||||
|
||||
public destroy(): void {
|
||||
super.destroy();
|
||||
this.setPollDefaultValues();
|
||||
}
|
||||
|
||||
public onStatus({ status }: onStatusParameters) {
|
||||
if (status === WebSocketStatus.Connecting) {
|
||||
this.isWebsocketFailed = true;
|
||||
if (this.pollTimeout) {
|
||||
clearTimeout(this.pollTimeout);
|
||||
}
|
||||
this.pollTimeout = setTimeout(() => {
|
||||
this.initPolling();
|
||||
}, 5000);
|
||||
} else if (status === WebSocketStatus.Connected) {
|
||||
this.setPollDefaultValues();
|
||||
}
|
||||
|
||||
super.onStatus({ status });
|
||||
}
|
||||
|
||||
public initPolling() {
|
||||
if (this.isLongPollingStarted || !this.isWebsocketFailed) {
|
||||
return;
|
||||
}
|
||||
|
||||
this.isLongPollingStarted = true;
|
||||
void this.pollSync(true);
|
||||
this.initCollaborationSSE();
|
||||
}
|
||||
|
||||
protected toPollUrl(endpoint: string): string {
|
||||
let pollUrl = this.url.replace('ws:', 'http:');
|
||||
if (pollUrl.includes('wss:')) {
|
||||
pollUrl = pollUrl.replace('wss:', 'https:');
|
||||
}
|
||||
|
||||
pollUrl = pollUrl.replace('/ws/', '/ws/poll/' + endpoint + '/');
|
||||
|
||||
// To have our requests not cached
|
||||
return `${pollUrl}&${Date.now()}`;
|
||||
}
|
||||
|
||||
protected isDuplicateAwareness(message64: string): boolean {
|
||||
if (this.treatedAwarenessMessage === message64) {
|
||||
return true;
|
||||
}
|
||||
|
||||
this.treatedAwarenessMessage = message64;
|
||||
|
||||
return false;
|
||||
}
|
||||
|
||||
/**
|
||||
* Outgoing message event
|
||||
*
|
||||
* Sent to the server the message to
|
||||
* be sent to the other users
|
||||
*/
|
||||
public async onPollOutgoingMessage({ message }: onOutgoingMessageParameters) {
|
||||
if (!this.isWebsocketFailed || !this.canEdit) {
|
||||
return;
|
||||
}
|
||||
|
||||
const message64 = Buffer.from(message.toUint8Array()).toString('base64');
|
||||
if (this.isDuplicateAwareness(message64)) {
|
||||
return;
|
||||
}
|
||||
|
||||
try {
|
||||
const { updated } = await pollOutgoingMessageRequest({
|
||||
pollUrl: this.toPollUrl('message'),
|
||||
message64,
|
||||
});
|
||||
|
||||
if (!updated) {
|
||||
await this.pollSync();
|
||||
}
|
||||
} catch (error: unknown) {
|
||||
if (isAPIError(error)) {
|
||||
// The user is not allowed to send messages
|
||||
if (error.status === 403) {
|
||||
this.off('outgoingMessage', this.onPollOutgoingMessage.bind(this));
|
||||
this.canEdit = false;
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* EventSource is a API for opening an HTTP
|
||||
* connection for receiving push notifications
|
||||
* from a server in real-time.
|
||||
* We use it to sync the document with the server
|
||||
*/
|
||||
protected initCollaborationSSE() {
|
||||
if (!this.isWebsocketFailed) {
|
||||
return;
|
||||
}
|
||||
|
||||
this.sse = new EventSource(this.toPollUrl('message'), {
|
||||
withCredentials: true,
|
||||
});
|
||||
|
||||
this.sse.onmessage = (event) => {
|
||||
const { updatedDoc64, stateFingerprint, awareness64 } = JSON.parse(
|
||||
// eslint-disable-next-line @typescript-eslint/no-unsafe-argument
|
||||
event.data,
|
||||
) as {
|
||||
updatedDoc64?: string;
|
||||
stateFingerprint?: string;
|
||||
awareness64?: string;
|
||||
};
|
||||
|
||||
if (awareness64) {
|
||||
if (this.isDuplicateAwareness(awareness64)) {
|
||||
return;
|
||||
}
|
||||
|
||||
this.treatedAwarenessMessage = awareness64;
|
||||
const awareness = Buffer.from(awareness64, 'base64');
|
||||
|
||||
this.onMessage({
|
||||
data: awareness,
|
||||
} as MessageEvent);
|
||||
}
|
||||
|
||||
if (updatedDoc64) {
|
||||
this.document.transact(() => {
|
||||
Y.applyUpdate(this.document, Buffer.from(updatedDoc64, 'base64'));
|
||||
}, this);
|
||||
}
|
||||
|
||||
const localStateFingerprint = this.getStateFingerprint(this.document);
|
||||
if (localStateFingerprint !== stateFingerprint) {
|
||||
void this.pollSync();
|
||||
} else {
|
||||
this.seemsUnsyncCount = 0;
|
||||
}
|
||||
};
|
||||
|
||||
this.sse.onopen = () => {};
|
||||
|
||||
this.sse.onerror = (err) => {
|
||||
console.error('SSE error:', err);
|
||||
this.sse?.close();
|
||||
|
||||
setTimeout(() => {
|
||||
this.initCollaborationSSE();
|
||||
}, 5000);
|
||||
};
|
||||
}
|
||||
|
||||
/**
|
||||
* Sync the document with the server.
|
||||
*
|
||||
* In some rare cases, the document may be out of sync.
|
||||
* We use a fingerprint to compare documents,
|
||||
* it happens that the local fingerprint is different from the server one
|
||||
* when awareness plus the document are updated quickly.
|
||||
* The system is resilient to this kind of problems, so `seemsUnsyncCount` should
|
||||
* go back to 0 after a few seconds. If not, we will force a sync.
|
||||
*/
|
||||
public async pollSync(forseSync = false) {
|
||||
if (!this.isWebsocketFailed || this.isSyncing) {
|
||||
return;
|
||||
}
|
||||
|
||||
this.seemsUnsyncCount++;
|
||||
|
||||
if (this.seemsUnsyncCount < this.seemsUnsyncMaxCount && !forseSync) {
|
||||
return;
|
||||
}
|
||||
|
||||
this.isSyncing = true;
|
||||
|
||||
try {
|
||||
const { syncDoc64 } = await postPollSyncRequest({
|
||||
pollUrl: this.toPollUrl('sync'),
|
||||
localDoc64: toBase64(Y.encodeStateAsUpdate(this.document)),
|
||||
});
|
||||
|
||||
if (syncDoc64) {
|
||||
const uint8Array = Buffer.from(syncDoc64, 'base64');
|
||||
Y.applyUpdate(this.document, uint8Array);
|
||||
this.seemsUnsyncCount = 0;
|
||||
}
|
||||
} catch (error) {
|
||||
console.error('Polling sync failed:', error);
|
||||
} finally {
|
||||
this.isSyncing = false;
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Create a hash SHA-256 of the state vector of the document.
|
||||
* Usefull to compare the state of the document.
|
||||
* @param doc
|
||||
* @returns
|
||||
*/
|
||||
public getStateFingerprint(doc: Y.Doc): string {
|
||||
const stateVector = Y.encodeStateVector(doc);
|
||||
return crypto.createHash('sha256').update(stateVector).digest('base64');
|
||||
}
|
||||
}
|
||||
+179
@@ -0,0 +1,179 @@
|
||||
import { WebSocketStatus } from '@hocuspocus/provider';
|
||||
import fetchMock from 'fetch-mock';
|
||||
import * as Y from 'yjs';
|
||||
|
||||
if (typeof EventSource === 'undefined') {
|
||||
const mockEventSource = jest.fn();
|
||||
class MockEventSource {
|
||||
constructor(...args: any[]) {
|
||||
return mockEventSource(...args);
|
||||
}
|
||||
}
|
||||
(global as any).EventSource = MockEventSource;
|
||||
}
|
||||
|
||||
import { CollaborationProvider } from '../CollaborationProvider';
|
||||
|
||||
const mockApplyUpdate = jest.fn();
|
||||
jest.mock('yjs', () => ({
|
||||
...jest.requireActual('yjs'),
|
||||
applyUpdate: (...args: any) => mockApplyUpdate(...args),
|
||||
}));
|
||||
|
||||
describe('CollaborationProvider', () => {
|
||||
let config: any;
|
||||
let provider: CollaborationProvider;
|
||||
let fakeWebsocketProvider: any;
|
||||
|
||||
beforeEach(() => {
|
||||
fakeWebsocketProvider = {
|
||||
on: jest.fn(),
|
||||
open: jest.fn(),
|
||||
attach: jest.fn(),
|
||||
};
|
||||
config = {
|
||||
name: 'test',
|
||||
url: 'ws://localhost/ws/',
|
||||
canEdit: true,
|
||||
websocketProvider: fakeWebsocketProvider,
|
||||
};
|
||||
provider = new CollaborationProvider(config);
|
||||
});
|
||||
|
||||
afterEach(() => {
|
||||
jest.clearAllMocks();
|
||||
fetchMock.restore();
|
||||
});
|
||||
|
||||
test('constructor initializes properties and attaches event handlers', () => {
|
||||
expect(provider.canEdit).toBe(true);
|
||||
expect((provider as any).url).toBe('ws://localhost/ws/');
|
||||
expect(fakeWebsocketProvider.on).toHaveBeenCalled();
|
||||
});
|
||||
|
||||
test('getStateFingerprint returns a consistent hash', () => {
|
||||
const fingerprint1 = provider.getStateFingerprint(provider.document);
|
||||
const fingerprint2 = provider.getStateFingerprint(provider.document);
|
||||
expect(typeof fingerprint1).toBe('string');
|
||||
expect(fingerprint1).toBe(fingerprint2);
|
||||
});
|
||||
|
||||
test('onPollOutgoingMessage does nothing when websocket is not failed', async () => {
|
||||
fetchMock.post(/http:\/\/localhost\/ws\/poll\/message\/.*/, {
|
||||
body: JSON.stringify({ updated: false }),
|
||||
});
|
||||
|
||||
provider.isWebsocketFailed = false;
|
||||
const dummyMessage = {
|
||||
toUint8Array: () => new Uint8Array([1, 2, 3]),
|
||||
} as any;
|
||||
await provider.onPollOutgoingMessage({ message: dummyMessage });
|
||||
expect(fetchMock.called()).toBe(false);
|
||||
});
|
||||
|
||||
test('onPollOutgoingMessage calls pollOutgoingMessageRequest and pollSync when updated is false', async () => {
|
||||
provider.isWebsocketFailed = true;
|
||||
const dummyMessage = {
|
||||
toUint8Array: () => new Uint8Array([4, 5, 6]),
|
||||
} as any;
|
||||
fetchMock.post(/http:\/\/localhost\/ws\/poll\/message\/.*/, {
|
||||
body: JSON.stringify({ updated: false }),
|
||||
});
|
||||
const pollSyncSpy = jest.spyOn(provider, 'pollSync').mockResolvedValue();
|
||||
await provider.onPollOutgoingMessage({ message: dummyMessage });
|
||||
expect(fetchMock.lastUrl()).toContain('http://localhost/ws/poll/message/');
|
||||
expect(pollSyncSpy).toHaveBeenCalled();
|
||||
});
|
||||
|
||||
test('onPollOutgoingMessage disables editing (canEdit becomes false) if API returns a 403 error', async () => {
|
||||
provider.isWebsocketFailed = true;
|
||||
const dummyMessage = {
|
||||
toUint8Array: () => new Uint8Array([7, 8, 9]),
|
||||
} as any;
|
||||
fetchMock.post(/http:\/\/localhost\/ws\/poll\/message\/.*/, {
|
||||
status: 403,
|
||||
body: JSON.stringify({}),
|
||||
});
|
||||
|
||||
// Stub the off method (inherited from event emitter) to observe its call.
|
||||
provider.off = jest.fn();
|
||||
await provider.onPollOutgoingMessage({ message: dummyMessage });
|
||||
expect(fetchMock.lastUrl()).toContain('http://localhost/ws/poll/message/');
|
||||
expect(provider.off).toHaveBeenCalled();
|
||||
expect(provider.canEdit).toBe(false);
|
||||
});
|
||||
|
||||
test('pollSync does nothing if websocket is not failed', async () => {
|
||||
fetchMock.post(/http:\/\/localhost\/ws\/poll\/sync\/.*/, {
|
||||
body: JSON.stringify({ syncDoc64: '123456' }),
|
||||
});
|
||||
|
||||
provider.isWebsocketFailed = false;
|
||||
await provider.pollSync();
|
||||
expect(fetchMock.called()).toBe(false);
|
||||
});
|
||||
|
||||
test('pollSync calls postPollSyncRequest when unsync count threshold is reached', async () => {
|
||||
const update = Y.encodeStateAsUpdate(provider.document);
|
||||
const syncDoc64 = Buffer.from(update).toString('base64');
|
||||
|
||||
fetchMock.post(/http:\/\/localhost\/ws\/poll\/sync\/.*/, {
|
||||
body: JSON.stringify({ syncDoc64 }),
|
||||
});
|
||||
|
||||
provider.isWebsocketFailed = true;
|
||||
provider.seemsUnsyncCount = provider.seemsUnsyncMaxCount - 1;
|
||||
|
||||
await provider.pollSync();
|
||||
const uint8Array = Buffer.from(syncDoc64, 'base64');
|
||||
expect(mockApplyUpdate).toHaveBeenCalledWith(provider.document, uint8Array);
|
||||
});
|
||||
|
||||
describe('onStatus', () => {
|
||||
beforeEach(() => {
|
||||
jest.useFakeTimers();
|
||||
});
|
||||
|
||||
afterEach(() => {
|
||||
jest.useRealTimers();
|
||||
});
|
||||
|
||||
test('sets websocket failed and schedules polling on Connecting', () => {
|
||||
const initPollingSpy = jest
|
||||
.spyOn(provider, 'initPolling')
|
||||
.mockImplementation(() => {});
|
||||
const superOnStatusSpy = jest.spyOn(
|
||||
Object.getPrototypeOf(provider),
|
||||
'onStatus',
|
||||
);
|
||||
|
||||
provider.onStatus({ status: WebSocketStatus.Connecting });
|
||||
|
||||
expect(provider.isWebsocketFailed).toBe(true);
|
||||
// Fast-forward timer to trigger the scheduled initPolling call.
|
||||
jest.runAllTimers();
|
||||
expect(initPollingSpy).toHaveBeenCalled();
|
||||
|
||||
expect(superOnStatusSpy).toHaveBeenCalledWith({
|
||||
status: WebSocketStatus.Connecting,
|
||||
});
|
||||
});
|
||||
|
||||
test('calls setPollDefaultValues on Connected', () => {
|
||||
const setPollDefaultValuesSpy = jest
|
||||
.spyOn(provider, 'setPollDefaultValues')
|
||||
.mockImplementation(() => {});
|
||||
const superOnStatusSpy = jest.spyOn(
|
||||
Object.getPrototypeOf(provider),
|
||||
'onStatus',
|
||||
);
|
||||
|
||||
provider.onStatus({ status: WebSocketStatus.Connected });
|
||||
|
||||
expect(setPollDefaultValuesSpy).toHaveBeenCalled();
|
||||
expect(superOnStatusSpy).toHaveBeenCalledWith({
|
||||
status: WebSocketStatus.Connected,
|
||||
});
|
||||
});
|
||||
});
|
||||
});
|
||||
+6
-2
@@ -4,10 +4,13 @@ import { create } from 'zustand';
|
||||
|
||||
import { Base64 } from '@/features/docs/doc-management';
|
||||
|
||||
import { CollaborationProvider } from '../libs/CollaborationProvider';
|
||||
|
||||
export interface UseCollaborationStore {
|
||||
createProvider: (
|
||||
providerUrl: string,
|
||||
storeId: string,
|
||||
canEdit: boolean,
|
||||
initialDoc?: Base64,
|
||||
) => HocuspocusProvider;
|
||||
destroyProvider: () => void;
|
||||
@@ -20,7 +23,7 @@ const defaultValues = {
|
||||
|
||||
export const useProviderStore = create<UseCollaborationStore>((set, get) => ({
|
||||
...defaultValues,
|
||||
createProvider: (wsUrl, storeId, initialDoc) => {
|
||||
createProvider: (wsUrl, storeId, canEdit, initialDoc) => {
|
||||
const doc = new Y.Doc({
|
||||
guid: storeId,
|
||||
});
|
||||
@@ -29,10 +32,11 @@ export const useProviderStore = create<UseCollaborationStore>((set, get) => ({
|
||||
Y.applyUpdate(doc, Buffer.from(initialDoc, 'base64'));
|
||||
}
|
||||
|
||||
const provider = new HocuspocusProvider({
|
||||
const provider = new CollaborationProvider({
|
||||
url: wsUrl,
|
||||
name: storeId,
|
||||
document: doc,
|
||||
canEdit,
|
||||
});
|
||||
|
||||
set({
|
||||
|
||||
@@ -12,6 +12,9 @@ export const currentDocRole = (abilities: Doc['abilities']): Role => {
|
||||
: Role.READER;
|
||||
};
|
||||
|
||||
export const toBase64 = (str: Uint8Array) =>
|
||||
Buffer.from(str).toString('base64');
|
||||
|
||||
export const base64ToYDoc = (base64: string) => {
|
||||
const uint8Array = Buffer.from(base64, 'base64');
|
||||
const ydoc = new Y.Doc();
|
||||
|
||||
@@ -63,7 +63,7 @@ const DocPage = ({ id }: DocProps) => {
|
||||
const { addTask } = useBroadcastStore();
|
||||
const queryClient = useQueryClient();
|
||||
const { replace } = useRouter();
|
||||
useCollaboration(doc?.id, doc?.content);
|
||||
useCollaboration(doc?.id, doc?.content, doc?.abilities.partial_update);
|
||||
|
||||
useEffect(() => {
|
||||
if (doc?.title) {
|
||||
|
||||
@@ -0,0 +1,185 @@
|
||||
/* eslint-disable @typescript-eslint/no-require-imports */
|
||||
import { EventEmitter } from 'events';
|
||||
|
||||
import { MessageType } from '@hocuspocus/server';
|
||||
import { Response } from 'express';
|
||||
import * as Y from 'yjs';
|
||||
|
||||
import { PollSync, PollSyncRequest } from '../src/libs/PollSync';
|
||||
|
||||
const { base64ToYDoc } = require('@/utils');
|
||||
const { logger } = require('@/utils');
|
||||
|
||||
const mockEncodeStateAsUpdate = jest.fn();
|
||||
jest.mock('yjs', () => ({
|
||||
...jest.requireActual('yjs'),
|
||||
encodeStateAsUpdate: () => mockEncodeStateAsUpdate(),
|
||||
}));
|
||||
|
||||
jest.mock('@/utils', () => ({
|
||||
base64ToYDoc: jest.fn((_b64: string) => new Y.Doc()),
|
||||
toBase64: jest.fn((data: Uint8Array) => Buffer.from(data).toString('base64')),
|
||||
logger: jest.fn(),
|
||||
}));
|
||||
|
||||
jest.mock('@hocuspocus/server', () => {
|
||||
const originalModule = jest.requireActual('@hocuspocus/server');
|
||||
return {
|
||||
__esModule: true,
|
||||
...originalModule,
|
||||
IncomingMessage: jest.fn().mockImplementation((buf: Buffer) => ({
|
||||
buffer: buf,
|
||||
readVarString: jest.fn(() => 'testRoom'),
|
||||
readVarUint: jest.fn(() => MessageType.Sync),
|
||||
writeVarUint: jest.fn(),
|
||||
decoder: {},
|
||||
encoder: {},
|
||||
})),
|
||||
};
|
||||
});
|
||||
|
||||
describe('PollSync', () => {
|
||||
let req: PollSyncRequest<any>;
|
||||
let res: Response;
|
||||
let dummyDoc: any;
|
||||
let pollSync: PollSync<any>;
|
||||
let mockHocuspocusServer: any;
|
||||
|
||||
beforeEach(() => {
|
||||
req = Object.assign(new EventEmitter(), {
|
||||
on: jest.fn((event, cb) => {
|
||||
// For 'close' event, store the callback for manual trigger in tests.
|
||||
if (event === 'close') {
|
||||
req.destroy = cb;
|
||||
}
|
||||
}),
|
||||
}) as unknown as PollSyncRequest<any>;
|
||||
|
||||
res = {
|
||||
write: jest.fn(),
|
||||
end: jest.fn(),
|
||||
} as unknown as Response;
|
||||
|
||||
// Create a dummy document with required methods/properties
|
||||
dummyDoc = {
|
||||
name: 'testRoom',
|
||||
merge: jest.fn((_other: Y.Doc) => {
|
||||
// Simulate merging by returning a new Y.Doc instance
|
||||
return new Y.Doc();
|
||||
}),
|
||||
getConnections: jest.fn(() => [{ handleMessage: jest.fn() }]),
|
||||
awareness: {
|
||||
on: jest.fn(),
|
||||
off: jest.fn(),
|
||||
},
|
||||
addDirectConnection: jest.fn(),
|
||||
removeDirectConnection: jest.fn(),
|
||||
on: jest.fn(),
|
||||
off: jest.fn(),
|
||||
};
|
||||
|
||||
pollSync = new PollSync(req, 'testRoom', true);
|
||||
// Pre-set the document for non-init tests
|
||||
(pollSync as any)._hpDocument = dummyDoc;
|
||||
|
||||
// Create a dummy Hocuspocus server
|
||||
mockHocuspocusServer = {
|
||||
loadingDocuments: { get: jest.fn() },
|
||||
documents: new Map<string, any>(),
|
||||
createDocument: jest.fn(() => ({
|
||||
name: 'newDoc',
|
||||
merge: (doc: Y.Doc) => doc,
|
||||
})),
|
||||
};
|
||||
});
|
||||
|
||||
afterEach(() => {
|
||||
jest.clearAllMocks();
|
||||
});
|
||||
|
||||
describe('initHocuspocusDocument', () => {
|
||||
it('should return document from loadingDocuments when available', async () => {
|
||||
mockHocuspocusServer.loadingDocuments.get.mockResolvedValue(dummyDoc);
|
||||
pollSync = new PollSync(req, 'testRoom', true);
|
||||
const doc = await pollSync.initHocuspocusDocument(mockHocuspocusServer);
|
||||
expect(doc).toBe(dummyDoc);
|
||||
expect(mockHocuspocusServer.loadingDocuments.get).toHaveBeenCalledWith(
|
||||
'testRoom',
|
||||
);
|
||||
});
|
||||
|
||||
it('should return document from documents when available', async () => {
|
||||
mockHocuspocusServer.loadingDocuments.get.mockResolvedValue(undefined);
|
||||
mockHocuspocusServer.documents.set('testRoom', dummyDoc);
|
||||
pollSync = new PollSync(req, 'testRoom', false);
|
||||
const doc = await pollSync.initHocuspocusDocument(mockHocuspocusServer);
|
||||
expect(doc).toBe(dummyDoc);
|
||||
});
|
||||
|
||||
it('should create a new document when none exists and canEdit is true', async () => {
|
||||
mockHocuspocusServer.loadingDocuments.get.mockResolvedValue(undefined);
|
||||
pollSync = new PollSync(req, 'testRoom', true);
|
||||
const newDoc = { name: 'newDoc', merge: (doc: any) => doc };
|
||||
mockHocuspocusServer.createDocument.mockResolvedValue(newDoc);
|
||||
const doc = await pollSync.initHocuspocusDocument(mockHocuspocusServer);
|
||||
expect(doc).toBe(newDoc);
|
||||
expect(mockHocuspocusServer.createDocument).toHaveBeenCalled();
|
||||
});
|
||||
});
|
||||
|
||||
describe('sync', () => {
|
||||
it('should encode state without merging when canEdit is false', () => {
|
||||
// When user cannot edit, merge should not be called.
|
||||
pollSync = new PollSync(req, 'testRoom', false);
|
||||
(pollSync as any)._hpDocument = dummyDoc;
|
||||
mockEncodeStateAsUpdate.mockReturnValue(Uint8Array.from([1, 2, 3]));
|
||||
|
||||
const result = pollSync.sync('dummyLocalDoc64');
|
||||
expect(dummyDoc.merge).not.toHaveBeenCalled();
|
||||
expect(result).toBe(
|
||||
Buffer.from(Uint8Array.from([1, 2, 3])).toString('base64'),
|
||||
);
|
||||
});
|
||||
|
||||
it('should merge local doc when canEdit is true', () => {
|
||||
pollSync = new PollSync(req, 'testRoom', true);
|
||||
(pollSync as any)._hpDocument = dummyDoc;
|
||||
const localDoc = new Y.Doc();
|
||||
// Mock base64ToYDoc to return our localDoc
|
||||
base64ToYDoc.mockReturnValue(localDoc);
|
||||
mockEncodeStateAsUpdate.mockReturnValue(Uint8Array.from([4, 5, 6]));
|
||||
pollSync.sync('localDoc64');
|
||||
expect(dummyDoc.merge).toHaveBeenCalledWith(localDoc);
|
||||
});
|
||||
});
|
||||
|
||||
describe('sendClientsMessages', () => {
|
||||
it('should log an error if room names do not match', () => {
|
||||
// Set doc name different from what IncomingMessage returns ('testRoom')
|
||||
dummyDoc.name = 'differentRoom';
|
||||
(pollSync as any)._hpDocument = dummyDoc;
|
||||
const fakeMessage = Buffer.from('fakeMessage').toString('base64');
|
||||
pollSync.sendClientsMessages(fakeMessage);
|
||||
expect(logger).toHaveBeenCalled();
|
||||
});
|
||||
});
|
||||
|
||||
describe('pullClientsMessages', () => {
|
||||
it('should register event listeners and cleanup on request close', () => {
|
||||
(pollSync as any)._hpDocument = dummyDoc;
|
||||
pollSync.pullClientsMessages(res);
|
||||
expect(dummyDoc.addDirectConnection).toHaveBeenCalled();
|
||||
expect(dummyDoc.on).toHaveBeenCalled();
|
||||
expect(dummyDoc.awareness.on).toHaveBeenCalled();
|
||||
|
||||
// Simulate a 'close' event on the request
|
||||
if (req.destroy) {
|
||||
req.destroy();
|
||||
}
|
||||
expect(dummyDoc.removeDirectConnection).toHaveBeenCalled();
|
||||
// Verify that document listeners are removed (off was called)
|
||||
expect(dummyDoc.off).toHaveBeenCalled();
|
||||
expect(dummyDoc.awareness.off).toHaveBeenCalled();
|
||||
});
|
||||
});
|
||||
});
|
||||
@@ -0,0 +1,193 @@
|
||||
import { Response } from 'express';
|
||||
|
||||
import {
|
||||
collaborationPollPostMessageHandler,
|
||||
collaborationPollSSEMessageHandler,
|
||||
collaborationPollSyncDocHandler,
|
||||
} from '../src/handlers/collaborationPollHandler';
|
||||
|
||||
const mockInitHocuspocusDocument = jest.fn();
|
||||
const mockSendClientsMessages = jest.fn();
|
||||
const mockSync = jest.fn();
|
||||
const mockPullClientsMessages = jest.fn();
|
||||
|
||||
jest.mock('@/libs/PollSync', () => {
|
||||
return {
|
||||
PollSync: jest.fn().mockImplementation(() => ({
|
||||
initHocuspocusDocument: mockInitHocuspocusDocument,
|
||||
sendClientsMessages: mockSendClientsMessages,
|
||||
sync: mockSync,
|
||||
pullClientsMessages: mockPullClientsMessages,
|
||||
})),
|
||||
};
|
||||
});
|
||||
|
||||
jest.mock('@/servers/hocusPocusServer', () => ({
|
||||
hocusPocusServer: {},
|
||||
}));
|
||||
|
||||
// Helper function to create a mock response
|
||||
function createResponse<T>() {
|
||||
const res: Partial<Response<T>> = {};
|
||||
res.status = jest.fn().mockReturnValue(res);
|
||||
res.json = jest.fn().mockReturnValue(res);
|
||||
res.setHeader = jest.fn();
|
||||
res.write = jest.fn();
|
||||
res.headersSent = false;
|
||||
return res as Response<T>;
|
||||
}
|
||||
|
||||
describe('collaborationPollPostMessageHandler', () => {
|
||||
beforeEach(() => {
|
||||
jest.clearAllMocks();
|
||||
mockInitHocuspocusDocument.mockResolvedValue({ doc: 'exists' });
|
||||
});
|
||||
|
||||
it('should return 403 if user is not allowed to edit', async () => {
|
||||
const req = {
|
||||
query: { room: 'test-room' },
|
||||
headers: { 'x-can-edit': 'False' },
|
||||
body: { message64: 'testMessage' },
|
||||
} as any;
|
||||
const res = createResponse();
|
||||
await collaborationPollPostMessageHandler(req, res);
|
||||
expect(res.status).toHaveBeenCalledWith(403);
|
||||
expect(res.json).toHaveBeenCalledWith({ error: 'Forbidden' });
|
||||
});
|
||||
|
||||
it('should return 400 if room is not provided', async () => {
|
||||
const req = {
|
||||
query: {},
|
||||
headers: { 'x-can-edit': 'True' },
|
||||
body: { message64: 'testMessage' },
|
||||
} as any;
|
||||
const res = createResponse();
|
||||
await collaborationPollPostMessageHandler(req, res);
|
||||
expect(res.status).toHaveBeenCalledWith(400);
|
||||
expect(res.json).toHaveBeenCalledWith({ error: 'Room name not provided' });
|
||||
});
|
||||
|
||||
it('should return 404 if document is not found', async () => {
|
||||
mockInitHocuspocusDocument.mockResolvedValue(null);
|
||||
const req = {
|
||||
query: { room: 'test-room' },
|
||||
headers: { 'x-can-edit': 'True' },
|
||||
body: { message64: 'testMessage' },
|
||||
} as any;
|
||||
const res = createResponse();
|
||||
res.headersSent = false;
|
||||
await collaborationPollPostMessageHandler(req, res);
|
||||
expect(res.status).toHaveBeenCalledWith(404);
|
||||
expect(res.json).toHaveBeenCalledWith({ error: 'Document not found' });
|
||||
});
|
||||
|
||||
it('should process message and return updated true when successful', async () => {
|
||||
// Reset headerSent to false to simulate a proper response
|
||||
const req = {
|
||||
query: { room: 'test-room' },
|
||||
headers: { 'x-can-edit': 'True' },
|
||||
body: { message64: 'testMessage' },
|
||||
} as any;
|
||||
const res = createResponse();
|
||||
res.headersSent = false;
|
||||
await collaborationPollPostMessageHandler(req, res);
|
||||
expect(mockSendClientsMessages).toHaveBeenCalledWith('testMessage');
|
||||
expect(res.status).toHaveBeenCalledWith(200);
|
||||
expect(res.json).toHaveBeenCalledWith({ updated: true });
|
||||
});
|
||||
});
|
||||
|
||||
describe('collaborationPollSyncDocHandler', () => {
|
||||
beforeEach(() => {
|
||||
jest.clearAllMocks();
|
||||
mockInitHocuspocusDocument.mockResolvedValue({ doc: 'exists' });
|
||||
mockSync.mockReturnValue('syncDocEncoded');
|
||||
});
|
||||
|
||||
it('should return 400 if room is not provided', async () => {
|
||||
const req = {
|
||||
query: {},
|
||||
headers: { 'x-can-edit': 'True' },
|
||||
body: { localDoc64: 'localDocEncoded' },
|
||||
} as any;
|
||||
const res = createResponse();
|
||||
await collaborationPollSyncDocHandler(req, res);
|
||||
expect(res.status).toHaveBeenCalledWith(400);
|
||||
expect(res.json).toHaveBeenCalledWith({ error: 'Room name not provided' });
|
||||
});
|
||||
|
||||
it('should return 404 if document is not found', async () => {
|
||||
mockInitHocuspocusDocument.mockResolvedValue(null);
|
||||
const req = {
|
||||
query: { room: 'test-room' },
|
||||
headers: { 'x-can-edit': 'True' },
|
||||
body: { localDoc64: 'localDocEncoded' },
|
||||
} as any;
|
||||
const res = createResponse();
|
||||
await collaborationPollSyncDocHandler(req, res);
|
||||
expect(res.status).toHaveBeenCalledWith(404);
|
||||
expect(res.json).toHaveBeenCalledWith({ error: 'Document not found' });
|
||||
});
|
||||
|
||||
it('should sync document and return syncDoc64 when successful', async () => {
|
||||
const req = {
|
||||
query: { room: 'test-room' },
|
||||
headers: { 'x-can-edit': 'True' },
|
||||
body: { localDoc64: 'localDocEncoded' },
|
||||
} as any;
|
||||
const res = createResponse();
|
||||
res.headersSent = false;
|
||||
await collaborationPollSyncDocHandler(req, res);
|
||||
expect(mockSync).toHaveBeenCalledWith('localDocEncoded');
|
||||
expect(res.status).toHaveBeenCalledWith(200);
|
||||
expect(res.json).toHaveBeenCalledWith({ syncDoc64: 'syncDocEncoded' });
|
||||
});
|
||||
});
|
||||
|
||||
describe('collaborationPollSSEMessageHandler', () => {
|
||||
beforeEach(() => {
|
||||
jest.clearAllMocks();
|
||||
mockInitHocuspocusDocument.mockResolvedValue({ doc: 'exists' });
|
||||
});
|
||||
|
||||
it('should return 400 if room is not provided', async () => {
|
||||
const req = {
|
||||
query: {},
|
||||
headers: { 'x-can-edit': 'True' },
|
||||
} as any;
|
||||
const res = createResponse();
|
||||
await collaborationPollSSEMessageHandler(req, res);
|
||||
expect(res.status).toHaveBeenCalledWith(400);
|
||||
expect(res.json).toHaveBeenCalledWith({ error: 'Room name not provided' });
|
||||
});
|
||||
|
||||
it('should return 404 if document is not found', async () => {
|
||||
mockInitHocuspocusDocument.mockResolvedValue(null);
|
||||
const req = {
|
||||
query: { room: 'test-room' },
|
||||
headers: { 'x-can-edit': 'True' },
|
||||
} as any;
|
||||
const res = createResponse();
|
||||
await collaborationPollSSEMessageHandler(req, res);
|
||||
expect(res.status).toHaveBeenCalledWith(404);
|
||||
expect(res.json).toHaveBeenCalledWith({ error: 'Document not found' });
|
||||
});
|
||||
|
||||
it('should set SSE headers and send connected message when successful', async () => {
|
||||
const req = {
|
||||
query: { room: 'test-room' },
|
||||
headers: { 'x-can-edit': 'True' },
|
||||
} as any;
|
||||
const res = createResponse();
|
||||
res.headersSent = false;
|
||||
await collaborationPollSSEMessageHandler(req, res);
|
||||
expect(res.setHeader).toHaveBeenCalledWith(
|
||||
'Content-Type',
|
||||
'text/event-stream',
|
||||
);
|
||||
expect(res.setHeader).toHaveBeenCalledWith('Cache-Control', 'no-cache');
|
||||
expect(res.setHeader).toHaveBeenCalledWith('Connection', 'keep-alive');
|
||||
expect(res.write).toHaveBeenCalledWith(': connected\n\n');
|
||||
expect(mockPullClientsMessages).toHaveBeenCalledWith(res);
|
||||
});
|
||||
});
|
||||
@@ -7,6 +7,7 @@ var config = {
|
||||
moduleNameMapper: {
|
||||
'^@/(.*)$': '<rootDir>/../src/$1',
|
||||
'^@blocknote/server-util$': '<rootDir>/../__mocks__/mock.js',
|
||||
'^y-protocols/awareness.js$': '<rootDir>/../__mocks__/mock.js',
|
||||
},
|
||||
};
|
||||
export default config;
|
||||
|
||||
@@ -23,6 +23,7 @@
|
||||
"cors": "2.8.5",
|
||||
"express": "4.21.2",
|
||||
"express-ws": "5.0.2",
|
||||
"uuid": "11.0.5",
|
||||
"y-protocols": "1.0.6",
|
||||
"yjs": "13.6.23"
|
||||
},
|
||||
|
||||
@@ -4,7 +4,8 @@ export const COLLABORATION_SERVER_ORIGIN =
|
||||
process.env.COLLABORATION_SERVER_ORIGIN || 'http://localhost:3000';
|
||||
export const COLLABORATION_SERVER_SECRET =
|
||||
process.env.COLLABORATION_SERVER_SECRET || 'secret-api-key';
|
||||
export const Y_PROVIDER_API_KEY =
|
||||
process.env.Y_PROVIDER_API_KEY || 'yprovider-api-key';
|
||||
export const PORT = Number(process.env.PORT || 4444);
|
||||
export const SENTRY_DSN = process.env.SENTRY_DSN || '';
|
||||
export const SENTRY_ENV = process.env.SENTRY_ENV || 'Development';
|
||||
export const Y_PROVIDER_API_KEY =
|
||||
process.env.Y_PROVIDER_API_KEY || 'yprovider-api-key';
|
||||
|
||||
@@ -0,0 +1,132 @@
|
||||
import { Response } from 'express';
|
||||
|
||||
import { PollSync, PollSyncRequest } from '@/libs/PollSync';
|
||||
import { hocusPocusServer } from '@/servers/hocusPocusServer';
|
||||
|
||||
interface CollaborationPollPostMessagePayload {
|
||||
message64: string;
|
||||
}
|
||||
interface CollaborationPollPostMessageResponse {
|
||||
updated?: boolean;
|
||||
error?: string;
|
||||
}
|
||||
|
||||
export const collaborationPollPostMessageHandler = async (
|
||||
req: PollSyncRequest<CollaborationPollPostMessagePayload>,
|
||||
res: Response<CollaborationPollPostMessageResponse>,
|
||||
) => {
|
||||
const room = req.query.room;
|
||||
const canEdit = req.headers['x-can-edit'] === 'True';
|
||||
|
||||
// Only editors can send messages
|
||||
if (!canEdit) {
|
||||
res.status(403).json({ error: 'Forbidden' });
|
||||
return;
|
||||
}
|
||||
|
||||
if (!room) {
|
||||
res.status(400).json({ error: 'Room name not provided' });
|
||||
return;
|
||||
}
|
||||
|
||||
const pollSynch = new PollSync<CollaborationPollPostMessagePayload>(
|
||||
req,
|
||||
room,
|
||||
canEdit,
|
||||
);
|
||||
const hpDoc = await pollSynch.initHocuspocusDocument(hocusPocusServer);
|
||||
|
||||
if (!res.headersSent && !hpDoc) {
|
||||
res.status(404).json({ error: 'Document not found' });
|
||||
return;
|
||||
}
|
||||
|
||||
pollSynch.sendClientsMessages(req.body.message64);
|
||||
|
||||
if (!res.headersSent) {
|
||||
res.status(200).json({ updated: true });
|
||||
}
|
||||
};
|
||||
|
||||
/**
|
||||
* Polling way of handling collaboration
|
||||
* @param req
|
||||
* @param res
|
||||
*/
|
||||
interface CollaborationPollSyncDocResponse {
|
||||
syncDoc64?: string;
|
||||
error?: string;
|
||||
}
|
||||
interface CollaborationPollSyncDocBody {
|
||||
localDoc64: string;
|
||||
}
|
||||
|
||||
export const collaborationPollSyncDocHandler = async (
|
||||
req: PollSyncRequest<CollaborationPollSyncDocBody>,
|
||||
res: Response<CollaborationPollSyncDocResponse>,
|
||||
) => {
|
||||
const room = req.query.room;
|
||||
const canEdit = req.headers['x-can-edit'] === 'True';
|
||||
|
||||
if (!room) {
|
||||
res.status(400).json({ error: 'Room name not provided' });
|
||||
return;
|
||||
}
|
||||
|
||||
const pollSynch = new PollSync<CollaborationPollSyncDocBody>(
|
||||
req,
|
||||
room,
|
||||
canEdit,
|
||||
);
|
||||
const hpDoc = await pollSynch.initHocuspocusDocument(hocusPocusServer);
|
||||
|
||||
if (!hpDoc) {
|
||||
res.status(404).json({ error: 'Document not found' });
|
||||
return;
|
||||
}
|
||||
|
||||
const syncDoc64 = pollSynch.sync(req.body.localDoc64);
|
||||
|
||||
if (!res.headersSent) {
|
||||
res.status(200).json({ syncDoc64 });
|
||||
}
|
||||
};
|
||||
|
||||
/**
|
||||
* SSE message handling
|
||||
* @param req
|
||||
* @param res
|
||||
*/
|
||||
interface CollaborationPollSSEMessageResponse {
|
||||
updatedDoc64?: string;
|
||||
stateFingerprint?: string;
|
||||
awareness64?: string;
|
||||
error?: string;
|
||||
}
|
||||
export const collaborationPollSSEMessageHandler = async (
|
||||
req: PollSyncRequest<void>,
|
||||
res: Response<CollaborationPollSSEMessageResponse>,
|
||||
) => {
|
||||
const room = req.query.room;
|
||||
const canEdit = req.headers['x-can-edit'] === 'True';
|
||||
|
||||
if (!room) {
|
||||
res.status(400).json({ error: 'Room name not provided' });
|
||||
return;
|
||||
}
|
||||
|
||||
const pollSynch = new PollSync<void>(req, room, canEdit);
|
||||
const hpDoc = await pollSynch.initHocuspocusDocument(hocusPocusServer);
|
||||
|
||||
if (!hpDoc) {
|
||||
res.status(404).json({ error: 'Document not found' });
|
||||
return;
|
||||
}
|
||||
|
||||
res.setHeader('Content-Type', 'text/event-stream');
|
||||
res.setHeader('Cache-Control', 'no-cache');
|
||||
res.setHeader('Connection', 'keep-alive');
|
||||
res.write(': connected\n\n');
|
||||
|
||||
pollSynch.pullClientsMessages(res);
|
||||
};
|
||||
@@ -1,3 +1,4 @@
|
||||
export * from './collaborationResetConnectionsHandler';
|
||||
export * from './collaborationWSHandler';
|
||||
export * from './collaborationPollHandler';
|
||||
export * from './convertMarkdownHandler';
|
||||
export * from './collaborationWSHandler';
|
||||
|
||||
@@ -1,6 +1,6 @@
|
||||
export const promiseDone = () => {
|
||||
let done: (value: void | PromiseLike<void>) => void = () => {};
|
||||
const promise = new Promise<void>((resolve) => {
|
||||
export const promiseDone = <T = void>() => {
|
||||
let done: (value: T | PromiseLike<T>) => void = () => {};
|
||||
const promise = new Promise<T>((resolve) => {
|
||||
done = resolve;
|
||||
});
|
||||
|
||||
|
||||
@@ -0,0 +1,227 @@
|
||||
/* eslint-disable @typescript-eslint/no-unused-vars */
|
||||
import crypto from 'crypto';
|
||||
|
||||
import {
|
||||
AwarenessUpdate,
|
||||
Document,
|
||||
Hocuspocus,
|
||||
IncomingMessage,
|
||||
MessageType,
|
||||
OutgoingMessage,
|
||||
} from '@hocuspocus/server';
|
||||
import { Request, Response } from 'express';
|
||||
import { v4 as uuid } from 'uuid';
|
||||
import { applyAwarenessUpdate } from 'y-protocols/awareness.js';
|
||||
import { readSyncMessage } from 'y-protocols/sync';
|
||||
import * as Y from 'yjs';
|
||||
|
||||
import { base64ToYDoc, logger, toBase64 } from '@/utils';
|
||||
|
||||
export type PollSyncRequestQuery = {
|
||||
room?: string;
|
||||
};
|
||||
|
||||
export type PollSyncRequest<T> = Request<
|
||||
object,
|
||||
object,
|
||||
T,
|
||||
PollSyncRequestQuery
|
||||
>;
|
||||
|
||||
export class PollSync<T> {
|
||||
public readonly canEdit: boolean;
|
||||
public readonly req: PollSyncRequest<T>;
|
||||
public readonly room: string;
|
||||
private _hpDocument?: Document;
|
||||
|
||||
constructor(req: PollSyncRequest<T>, room: string, canEdit: boolean) {
|
||||
this.room = room;
|
||||
this.canEdit = canEdit;
|
||||
this.req = req;
|
||||
}
|
||||
|
||||
public get hpDocument() {
|
||||
return this._hpDocument;
|
||||
}
|
||||
|
||||
public async initHocuspocusDocument(hocusPocusServer: Hocuspocus) {
|
||||
const { req, room, canEdit } = this;
|
||||
this._hpDocument = await hocusPocusServer.loadingDocuments.get(room);
|
||||
|
||||
if (this._hpDocument) {
|
||||
return this._hpDocument;
|
||||
}
|
||||
|
||||
this._hpDocument = hocusPocusServer.documents.get(room);
|
||||
|
||||
if (this._hpDocument || (!this._hpDocument && !canEdit)) {
|
||||
return this._hpDocument;
|
||||
}
|
||||
|
||||
/**
|
||||
* createDocument is used to create a new document if it does not exist.
|
||||
* If the document exists, it will return the existing document.
|
||||
*/
|
||||
this._hpDocument = await hocusPocusServer.createDocument(
|
||||
room,
|
||||
req,
|
||||
uuid(),
|
||||
{
|
||||
readOnly: false,
|
||||
requiresAuthentication: false,
|
||||
isAuthenticated: false,
|
||||
},
|
||||
);
|
||||
|
||||
return this._hpDocument;
|
||||
}
|
||||
|
||||
/**
|
||||
* Sync the document with the latest changes
|
||||
*
|
||||
* @param localDoc64
|
||||
* @returns
|
||||
*/
|
||||
public sync(localDoc64: string): string | undefined {
|
||||
const hpDoc = this.getHpDocument();
|
||||
let syncYDoc = hpDoc;
|
||||
|
||||
// Merge the coming document with the latest changes (only if the user can edit)
|
||||
if (this.canEdit) {
|
||||
const localDoc = base64ToYDoc(localDoc64);
|
||||
syncYDoc = hpDoc.merge(localDoc);
|
||||
}
|
||||
|
||||
return toBase64(Y.encodeStateAsUpdate(syncYDoc));
|
||||
}
|
||||
|
||||
/**
|
||||
* Create a hash SHA-256 of the state vector of the document.
|
||||
* Usefull to compare the state of the document.
|
||||
* @param doc
|
||||
* @returns
|
||||
*/
|
||||
protected getStateFingerprint(doc: Y.Doc): string {
|
||||
const stateVector = Y.encodeStateVector(doc);
|
||||
return crypto.createHash('sha256').update(stateVector).digest('base64'); // or 'hex'
|
||||
}
|
||||
|
||||
/**
|
||||
* Send messages to other clients
|
||||
*/
|
||||
public sendClientsMessages(message64: string) {
|
||||
const hpDoc = this.getHpDocument();
|
||||
const messageBuffer = Buffer.from(message64, 'base64');
|
||||
const message = new IncomingMessage(messageBuffer);
|
||||
const room = message.readVarString();
|
||||
|
||||
if (hpDoc.name !== room) {
|
||||
logger('Send messages problem, room different', room, hpDoc.name);
|
||||
return;
|
||||
}
|
||||
|
||||
// We write the sync to the current doc - it will propagate to others by itself
|
||||
const type = message.readVarUint() as MessageType;
|
||||
if (type === MessageType.Sync) {
|
||||
message.writeVarUint(MessageType.Sync);
|
||||
readSyncMessage(message.decoder, message.encoder, hpDoc, null);
|
||||
} else if (type === MessageType.Awareness) {
|
||||
const awarenessUpdate = message.readVarUint8Array();
|
||||
applyAwarenessUpdate(
|
||||
hpDoc.awareness,
|
||||
awarenessUpdate,
|
||||
hpDoc.awareness.clientID,
|
||||
);
|
||||
} else {
|
||||
hpDoc.getConnections().forEach((connection) => {
|
||||
connection.handleMessage(messageBuffer);
|
||||
});
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Pull messages from other clients
|
||||
*
|
||||
* We listen 2 kind of messages:
|
||||
* - Document updates (change in the document)
|
||||
* - Awareness messages (cursor, selection, etc.)
|
||||
*
|
||||
* @param res
|
||||
*/
|
||||
public pullClientsMessages(res: Response) {
|
||||
const hpDoc = this.getHpDocument();
|
||||
hpDoc.addDirectConnection();
|
||||
|
||||
const updateMessagesFn = (
|
||||
update: Uint8Array,
|
||||
_origin: string,
|
||||
updatedDoc: Y.Doc,
|
||||
_transaction: Y.Transaction,
|
||||
) => {
|
||||
res.write(
|
||||
`data: ${JSON.stringify({
|
||||
time: new Date(),
|
||||
updatedDoc64: toBase64(update),
|
||||
stateFingerprint: this.getStateFingerprint(updatedDoc),
|
||||
})}\n\n`,
|
||||
);
|
||||
};
|
||||
|
||||
const destroyFn = (updatedDoc: Y.Doc) => {
|
||||
res.write(
|
||||
`data: ${JSON.stringify({
|
||||
time: new Date(),
|
||||
updatedDoc64: undefined,
|
||||
stateFingerprint: this.getStateFingerprint(updatedDoc),
|
||||
})}\n\n`,
|
||||
);
|
||||
|
||||
hpDoc.off('destroy', destroyFn);
|
||||
hpDoc.off('update', updateMessagesFn);
|
||||
|
||||
// Close the connection
|
||||
res.end();
|
||||
};
|
||||
|
||||
const updateAwarenessFn = ({
|
||||
added,
|
||||
updated,
|
||||
removed,
|
||||
}: AwarenessUpdate) => {
|
||||
const changedClients = added.concat(updated, removed);
|
||||
const awarenessMessage = new OutgoingMessage(
|
||||
this.room,
|
||||
).createAwarenessUpdateMessage(hpDoc.awareness, changedClients);
|
||||
|
||||
res.write(
|
||||
`data: ${JSON.stringify({
|
||||
time: new Date(),
|
||||
awareness64: toBase64(awarenessMessage.toUint8Array()),
|
||||
stateFingerprint: this.getStateFingerprint(hpDoc),
|
||||
})}\n\n`,
|
||||
);
|
||||
};
|
||||
|
||||
hpDoc.awareness.off('update', updateAwarenessFn);
|
||||
hpDoc.awareness.on('update', updateAwarenessFn);
|
||||
hpDoc.off('update', updateMessagesFn);
|
||||
hpDoc.off('destroy', destroyFn);
|
||||
hpDoc.on('update', updateMessagesFn);
|
||||
hpDoc.on('destroy', destroyFn);
|
||||
|
||||
this.req.on('close', () => {
|
||||
hpDoc.off('update', updateMessagesFn);
|
||||
hpDoc.off('destroy', destroyFn);
|
||||
hpDoc.awareness.off('update', updateAwarenessFn);
|
||||
hpDoc.removeDirectConnection();
|
||||
});
|
||||
}
|
||||
|
||||
protected getHpDocument() {
|
||||
if (!this.hpDocument) {
|
||||
throw new Error('HocusPocus document not initialized');
|
||||
}
|
||||
|
||||
return this.hpDocument;
|
||||
}
|
||||
}
|
||||
@@ -1,5 +1,7 @@
|
||||
export const routes = {
|
||||
COLLABORATION_WS: '/collaboration/ws/',
|
||||
COLLABORATION_POLL_MESSAGE: '/collaboration/ws/poll/message/',
|
||||
COLLABORATION_POLL_SYNC: '/collaboration/ws/poll/sync/',
|
||||
COLLABORATION_RESET_CONNECTIONS: '/collaboration/api/reset-connections/',
|
||||
CONVERT_MARKDOWN: '/api/convert-markdown/',
|
||||
};
|
||||
|
||||
@@ -6,6 +6,9 @@ import expressWebsockets from 'express-ws';
|
||||
|
||||
import { PORT } from '../env';
|
||||
import {
|
||||
collaborationPollPostMessageHandler,
|
||||
collaborationPollSSEMessageHandler,
|
||||
collaborationPollSyncDocHandler,
|
||||
collaborationResetConnectionsHandler,
|
||||
collaborationWSHandler,
|
||||
convertMarkdownHandler,
|
||||
@@ -27,9 +30,24 @@ export const initServer = () => {
|
||||
app.use(corsMiddleware);
|
||||
|
||||
/**
|
||||
* Route to handle WebSocket connections
|
||||
* Routes to handle collaboration connections
|
||||
*/
|
||||
app.ws(routes.COLLABORATION_WS, wsSecurity, collaborationWSHandler);
|
||||
app.get(
|
||||
routes.COLLABORATION_POLL_MESSAGE,
|
||||
httpSecurity,
|
||||
collaborationPollSSEMessageHandler,
|
||||
);
|
||||
app.post(
|
||||
routes.COLLABORATION_POLL_MESSAGE,
|
||||
httpSecurity,
|
||||
collaborationPollPostMessageHandler,
|
||||
);
|
||||
app.post(
|
||||
routes.COLLABORATION_POLL_SYNC,
|
||||
httpSecurity,
|
||||
collaborationPollSyncDocHandler,
|
||||
);
|
||||
|
||||
/**
|
||||
* Route to reset connections in a room:
|
||||
|
||||
@@ -1,11 +1,12 @@
|
||||
import * as Sentry from '@sentry/node';
|
||||
import { nodeProfilingIntegration } from '@sentry/profiling-node';
|
||||
|
||||
import { SENTRY_DSN } from '../env';
|
||||
import { SENTRY_DSN, SENTRY_ENV } from '../env';
|
||||
|
||||
Sentry.init({
|
||||
dsn: SENTRY_DSN,
|
||||
integrations: [nodeProfilingIntegration()],
|
||||
tracesSampleRate: 0.1,
|
||||
profilesSampleRate: 1.0,
|
||||
environment: SENTRY_ENV,
|
||||
});
|
||||
|
||||
@@ -1,3 +1,5 @@
|
||||
import * as Y from 'yjs';
|
||||
|
||||
import { COLLABORATION_LOGGING } from './env';
|
||||
|
||||
// eslint-disable-next-line @typescript-eslint/no-explicit-any
|
||||
@@ -8,6 +10,13 @@ export function logger(...args: any[]) {
|
||||
}
|
||||
}
|
||||
|
||||
export const toBase64 = function (str: Uint8Array) {
|
||||
return Buffer.from(str).toString('base64');
|
||||
export const toBase64 = function (uInt8Array: Uint8Array) {
|
||||
return Buffer.from(uInt8Array).toString('base64');
|
||||
};
|
||||
|
||||
export const base64ToYDoc = (base64: string) => {
|
||||
const uint8Array = Buffer.from(base64, 'base64');
|
||||
const ydoc = new Y.Doc();
|
||||
Y.applyUpdate(ydoc, uint8Array);
|
||||
return ydoc;
|
||||
};
|
||||
|
||||
@@ -5281,6 +5281,13 @@
|
||||
dependencies:
|
||||
"@types/node" "*"
|
||||
|
||||
"@types/ws@8.5.13":
|
||||
version "8.5.13"
|
||||
resolved "https://registry.yarnpkg.com/@types/ws/-/ws-8.5.13.tgz#6414c280875e2691d0d1e080b05addbf5cb91e20"
|
||||
integrity sha512-osM/gWBTPKgHV8XkTunnegTRIsvF6owmf5w+JtAfOw472dptdm0dlGv4xCt6GwQRcC2XVOvvRE/0bAoQcL2QkA==
|
||||
dependencies:
|
||||
"@types/node" "*"
|
||||
|
||||
"@types/yargs-parser@*":
|
||||
version "21.0.3"
|
||||
resolved "https://registry.yarnpkg.com/@types/yargs-parser/-/yargs-parser-21.0.3.tgz#815e30b786d2e8f0dcd85fd5bcf5e1a04d008f15"
|
||||
@@ -13934,7 +13941,7 @@ utils-merge@1.0.1:
|
||||
resolved "https://registry.yarnpkg.com/utils-merge/-/utils-merge-1.0.1.tgz#9f95710f50a267947b2ccc124741c1028427e713"
|
||||
integrity sha512-pMZTvIkT1d+TFGvDOqodOclx0QWkkgi6Tdoa8gC8ffGAAqz9pzPTZWAybbsHHoED/ztMtkv/VoYTYyShUn81hA==
|
||||
|
||||
uuid@^11.0.3:
|
||||
uuid@11.0.5, uuid@^11.0.3:
|
||||
version "11.0.5"
|
||||
resolved "https://registry.yarnpkg.com/uuid/-/uuid-11.0.5.tgz#07b46bdfa6310c92c3fb3953a8720f170427fc62"
|
||||
integrity sha512-508e6IcKLrhxKdBbcA2b4KQZlLVp2+J5UwQ6F7Drckkc5N9ZJwFa4TgWtsww9UG8fGHbm6gbV19TdM5pQ4GaIA==
|
||||
|
||||
@@ -1,5 +1,5 @@
|
||||
apiVersion: v2
|
||||
type: application
|
||||
name: docs
|
||||
version: 2.2.0-beta.1
|
||||
version: 2.2.0-beta.2
|
||||
appVersion: latest
|
||||
|
||||
@@ -82,7 +82,11 @@ ingressCollaborationWS:
|
||||
## @param ingressCollaborationWS.annotations.nginx.ingress.kubernetes.io/proxy-send-timeout
|
||||
## @param ingressCollaborationWS.annotations.nginx.ingress.kubernetes.io/upstream-hash-by
|
||||
annotations:
|
||||
nginx.ingress.kubernetes.io/auth-cache-key: "$http_authorization-$arg_room"
|
||||
nginx.ingress.kubernetes.io/auth-cache-duration: 200 30s
|
||||
nginx.ingress.kubernetes.io/auth-response-headers: "Authorization, X-Can-Edit, X-User-Id"
|
||||
nginx.ingress.kubernetes.io/auth-snippet: |
|
||||
proxy_set_header Accept "application/json";
|
||||
nginx.ingress.kubernetes.io/auth-url: https://impress.example.com/api/v1.0/documents/collaboration-auth/
|
||||
nginx.ingress.kubernetes.io/enable-websocket: "true"
|
||||
nginx.ingress.kubernetes.io/proxy-read-timeout: "86400"
|
||||
|
||||
Reference in New Issue
Block a user