feat: add beforeSync hook

Author: nperez0111

docs/server/hooks.md

@@ -49,7 +49,6 @@ By way of illustration, if a user isn’t allowed to connect: Just throw an erro
 | `beforeBroadcastStateless` | Before broadcast a stateless message      | [Read more](/server/hooks#before-broadcast-stateless) |
 | `afterUnloadDocument`      | When a document is closed                 | [Read more](/server/hooks#after-unload-document)      |
 
-
 ## Usage
 
 ```js
@@ -759,7 +758,6 @@ const server = new Server({
 server.listen();
 ```
 
-
 ### onStateless
 
 The `onStateless` hooks are called after the server has received a stateless message. It should return a Promise.
@@ -796,6 +794,37 @@ const server = new Server({
 server.listen()
 ```
 
+### beforeSync
+
+The `beforeSync` hooks are called before a sync message is handled. This is useful if you want to inspect the sync message that will be applied to the document.
+
+**Hook payload**
+
+```js
+const data = {
+  documentName: string,
+  document: Document,
+  // The y-protocols/sync message type
+  type: number,
+  // The payload of the y-protocols/sync message
+  payload: Uint8Array,
+}
+```
+
+**Example**
+
+```js
+import { Server } from '@hocuspocus/server'
+
+const server = new Server({
+  async beforeSync({ payload, document, documentName, type }) {
+    console.log(`Server will handle a sync message: "${payload}"!`)
+  },
+})
+
+server.listen()
+```
+
 ### beforeBroadcastStateless
 
 The `beforeBroadcastStateless` hooks are called before the server broadcast a stateless message.

packages/server/src/ClientConnection.ts

@@ -16,6 +16,7 @@ import { OutgoingMessage } from './OutgoingMessage.ts'
 import type {
   ConnectionConfiguration,
   beforeHandleMessagePayload,
+  beforeSyncPayload,
   onDisconnectPayload,
 } from './types.ts'
 import {
@@ -199,6 +200,20 @@ export class ClientConnection {
       return this.hooks('beforeHandleMessage', beforeHandleMessagePayload)
     })
 
+    instance.beforeSync((connection, payload) => {
+      const beforeSyncPayload: beforeSyncPayload = {
+        clientsCount: document.getConnectionsCount(),
+        context: hookPayload.context,
+        document,
+        documentName: document.name,
+        connection,
+        type: payload.type,
+        payload: payload.payload,
+      }
+
+      return this.hooks('beforeSync', beforeSyncPayload)
+    })
+
     return instance
   }
 

packages/server/src/Connection.ts

@@ -8,7 +8,7 @@ import type Document from './Document.ts'
 import { IncomingMessage } from './IncomingMessage.ts'
 import { MessageReceiver } from './MessageReceiver.ts'
 import { OutgoingMessage } from './OutgoingMessage.ts'
-import type { onStatelessPayload } from './types.ts'
+import type { beforeSyncPayload, onStatelessPayload } from './types.ts'
 
 export class Connection {
 
@@ -23,6 +23,7 @@ export class Connection {
   callbacks = {
     onClose: [(document: Document, event?: CloseEvent) => {}],
     beforeHandleMessage: (connection: Connection, update: Uint8Array) => Promise.resolve(),
+    beforeSync: (connection: Connection, payload: Pick<beforeSyncPayload, 'type' | 'payload'>) => Promise.resolve(),
     statelessCallback: (payload: onStatelessPayload) => Promise.resolve(),
   }
 
@@ -81,6 +82,15 @@ export class Connection {
     return this
   }
 
+  /**
+   * Set a callback that will be triggered before a sync message is handled
+   */
+  beforeSync(callback: (connection: Connection, payload: Pick<beforeSyncPayload, 'type' | 'payload'>) => Promise<any>): Connection {
+    this.callbacks.beforeSync = callback
+
+    return this
+  }
+
   /**
    * Send the given message
    */

packages/server/src/Hocuspocus.ts

@@ -49,6 +49,7 @@ export class Hocuspocus {
     onConnect: () => new Promise(r => r(null)),
     connected: () => new Promise(r => r(null)),
     beforeHandleMessage: () => new Promise(r => r(null)),
+    beforeSync: () => new Promise(r => r(null)),
     beforeBroadcastStateless: () => new Promise(r => r(null)),
     onStateless: () => new Promise(r => r(null)),
     onChange: () => new Promise(r => r(null)),
@@ -111,6 +112,7 @@ export class Hocuspocus {
       afterLoadDocument: this.configuration.afterLoadDocument,
       beforeHandleMessage: this.configuration.beforeHandleMessage,
       beforeBroadcastStateless: this.configuration.beforeBroadcastStateless,
+      beforeSync: this.configuration.beforeSync,
       onStateless: this.configuration.onStateless,
       onChange: this.configuration.onChange,
       onStoreDocument: this.configuration.onStoreDocument,

packages/server/src/IncomingMessage.ts

@@ -49,6 +49,13 @@ export class IncomingMessage {
     return readVarUint8Array(this.decoder)
   }
 
+  peekVarUint8Array() {
+    const { pos } = this.decoder
+    const result = readVarUint8Array(this.decoder)
+    this.decoder.pos = pos
+    return result
+  }
+
   readVarUint() {
     return readVarUint(this.decoder)
   }

packages/server/src/MessageReceiver.ts

@@ -104,6 +104,13 @@ export class MessageReceiver {
   readSyncMessage(message: IncomingMessage, document: Document, connection?: Connection, reply?: (message: Uint8Array) => void, requestFirstSync = true) {
     const type = message.readVarUint()
 
+    if (connection) {
+      connection.callbacks.beforeSync(connection, {
+        type,
+        payload: message.peekVarUint8Array(),
+      })
+    }
+
     switch (type) {
       case messageYjsSyncStep1: {
         readSyncStep1(message.decoder, message.encoder, document)

packages/server/src/types.ts

@@ -44,6 +44,7 @@ export interface Extension {
   onLoadDocument?(data: onLoadDocumentPayload): Promise<any>;
   afterLoadDocument?(data: afterLoadDocumentPayload): Promise<any>;
   beforeHandleMessage?(data: beforeHandleMessagePayload): Promise<any>;
+  beforeSync?(data: beforeSyncPayload): Promise<any>;
   beforeBroadcastStateless?(data: beforeBroadcastStatelessPayload): Promise<any>;
   onStateless?(payload: onStatelessPayload): Promise<any>;
   onChange?(data: onChangePayload): Promise<any>;
@@ -69,6 +70,7 @@ export type HookName =
   'afterLoadDocument' |
   'beforeHandleMessage' |
   'beforeBroadcastStateless' |
+  'beforeSync' |
   'onStateless' |
   'onChange' |
   'onStoreDocument' |
@@ -92,6 +94,7 @@ export type HookPayloadByName = {
   afterLoadDocument: afterLoadDocumentPayload,
   beforeHandleMessage: beforeHandleMessagePayload,
   beforeBroadcastStateless: beforeBroadcastStatelessPayload,
+  beforeSync: beforeSyncPayload,
   onStateless: onStatelessPayload,
   onChange: onChangePayload,
   onStoreDocument: onStoreDocumentPayload,
@@ -250,6 +253,28 @@ export interface beforeHandleMessagePayload {
   connection: Connection
 }
 
+export interface beforeSyncPayload {
+  clientsCount: number,
+  context: any,
+  document: Document,
+  documentName: string,
+  connection: Connection,
+  /**
+   * The y-protocols/sync message type
+   * @example
+   * 0: SyncStep1
+   * 1: SyncStep2
+   * 2: YjsUpdate
+   *
+   * @see https://github.com/yjs/y-protocols/blob/master/sync.js#L13-L40
+   */
+  type: number,
+  /**
+   * The payload of the y-sync message.
+   */
+  payload: Uint8Array,
+}
+
 export interface beforeBroadcastStatelessPayload {
   document: Document,
   documentName: string,

tests/server/beforeSync.ts

@@ -0,0 +1,164 @@
+import test from 'ava'
+import { newHocuspocus, newHocuspocusProvider } from '../utils/index.ts'
+import { retryableAssertion } from '../utils/retryableAssertion.ts'
+
+test('beforeSync gets called in proper order', async t => {
+  await new Promise(async resolve => {
+    const mockContext = {
+      user: 123,
+    }
+
+    const expectedValuesByCallNumber = [
+      undefined, // beforeSync 1
+      'bar', // beforeSync 2
+      'bar', // beforeSync 3
+    ]
+    let callNumber = 0
+
+    const server = await newHocuspocus({
+      async onConnect() {
+        return mockContext
+      },
+      async beforeSync({ document, context, payload }) {
+        t.deepEqual(context, mockContext)
+
+        const value = document.getArray('foo').get(0)
+
+        t.is(value, expectedValuesByCallNumber[callNumber])
+        callNumber += 1
+
+        if (callNumber === expectedValuesByCallNumber.length - 1) {
+          resolve('done')
+        }
+      },
+      async onChange({ context, document }) {
+        t.deepEqual(context, mockContext)
+
+        const value = document.getArray('foo').get(0)
+
+        t.is(value, expectedValuesByCallNumber[2])
+      },
+    })
+
+    const provider = newHocuspocusProvider(server, {
+      onSynced() {
+        provider.document.getArray('foo').insert(0, ['bar'])
+      },
+    })
+  })
+})
+
+test('beforeSync callback is called for every sync', async t => {
+  let onConnectCount = 0
+  let updateCount = 0
+  let syncstep1Count = 0
+  let syncstep2Count = 0
+
+  await new Promise(async resolve => {
+    const server = await newHocuspocus({
+      async onConnect() {
+        onConnectCount += 1
+      },
+      async beforeSync({ type }) {
+        if (type === 0){
+          syncstep1Count += 1
+        } else if (type === 1) {
+          syncstep2Count += 1
+        } else if (type === 2) {
+          updateCount += 1
+        }
+      },
+    })
+
+    await Promise.all([
+      new Promise(done => {
+        newHocuspocusProvider(server, {
+          onClose() {
+            t.fail()
+          },
+          onSynced() {
+            done('done')
+          },
+        })
+      }),
+      new Promise(done => {
+        newHocuspocusProvider(server, {
+          onClose() {
+            t.fail()
+          },
+          onSynced() {
+            done('done')
+          },
+        })
+      }),
+    ])
+
+    resolve('done')
+  })
+
+  await retryableAssertion(t, tt => {
+    tt.is(onConnectCount, 2)
+    tt.is(syncstep1Count, 2)
+    tt.is(syncstep2Count, 2)
+    tt.is(updateCount, 0)
+  })
+})
+
+
+test('beforeSync callback is called on every update', async t => {
+  let onConnectCount = 0
+  let updateCount = 0
+  let syncstep1Count = 0
+  let syncstep2Count = 0
+
+
+  await new Promise(async resolve => {
+    const server = await newHocuspocus({
+      async onConnect() {
+        onConnectCount += 1
+      },
+      async beforeSync({ type }) {
+        if (type === 0){
+          syncstep1Count += 1
+        } else if (type === 1) {
+          syncstep2Count += 1
+        } else if (type === 2) {
+          updateCount += 1
+        }
+      },
+    })
+
+    await Promise.all([
+      new Promise(done => {
+        newHocuspocusProvider(server, {
+          onClose() {
+            t.fail()
+          },
+          onSynced() {
+            done('done')
+          },
+        })
+      }),
+      new Promise(done => {
+        const provider = newHocuspocusProvider(server, {
+          onClose() {
+            t.fail()
+          },
+          onSynced() {
+            provider.document.getArray('foo').insert(0, ['bar'])
+            done('done')
+          },
+        })
+      }),
+    ])
+
+    resolve('done')
+  })
+
+  await retryableAssertion(t, tt => {
+    tt.is(onConnectCount, 2)
+    tt.is(syncstep1Count, 2)
+    tt.is(syncstep2Count, 2)
+    tt.is(updateCount, 1)
+  })
+})