[ws] Allow server to manipulate store data after persistence starts

jamesgpearce · jamesgpearce · commit db121d51866e · 2024-07-25T15:54:17.000-05:00
diff --git a/src/@types/synchronizers/synchronizer-ws-server/docs.js b/src/@types/synchronizers/synchronizer-ws-server/docs.js
@@ -493,10 +493,20 @@
  * by the WsServer. As a result, the server MergeableStore will be kept in sync
  * with the clients on that path, and in turn with whatever persistence layer
  * you have configured. See the example below.
+ *
+ * It is not safe to add or manipulate data in the MergeableStore during the
+ * `createPersisterForPath` function, since changes will probably be overwritten
+ * when the Persister starts. If you wish to modify data - or upgrade a schema,
+ * for example - you can have that function instead return an array containing
+ * the Persister _and_ a callback that takes the MergeableStore. That callback
+ * will get called after the Persister has started, and is an appropriate place
+ * to manipulate data in a way that will be transmitted to clients. Again, see
+ * the example below.
  * @param webSocketServer A WebSocketServer object from your server environment.
  * @param createPersisterForPath An optional function that will create a
- * Persister (with a MergeableStore) to synchronize with the clients on a given
- * path.
+ * Persister to synchronize with the clients on a given path (or a two-item
+ * array of Persister and callback that lets you handle data after persistence
+ * has started).
  * @returns A reference to the new WsServer object.
  * @example
  * This example creates a WsServer that synchronizes two clients on a shared
@@ -610,6 +620,46 @@
  * rmSync('petShop.json');
  * ```
  * @example
+ * This example creates a WsServer that persists a MergeableStore to file that
+ * is synchronized with two clients on a shared path, but also which updates its
+ * data once synchronization has started.
+ *
+ * ```js
+ * import {WebSocketServer} from 'ws';
+ * import {createFilePersister} from 'tinybase/persisters/persister-file';
+ * import {createMergeableStore} from 'tinybase';
+ * import {createWsServer} from 'tinybase/synchronizers/synchronizer-ws-server';
+ * import {createWsSynchronizer} from 'tinybase/synchronizers/synchronizer-ws-client';
+ * import {rmSync} from 'fs';
+ *
+ * // Server
+ * const server = createWsServer(
+ *   new WebSocketServer({port: 8047}),
+ *   (pathId) => [
+ *     createFilePersister(createMergeableStore(), pathId + '.json'),
+ *     (store) => store.setValue('pathId', pathId),
+ *   ],
+ * );
+ *
+ * const clientStore = createMergeableStore();
+ * clientStore.setCell('pets', 'fido', 'species', 'dog');
+ * const synchronizer = await createWsSynchronizer(
+ *   clientStore,
+ *   new WebSocket('ws://localhost:8047/petShop'),
+ * );
+ * await synchronizer.startSync();
+ * // ...
+ *
+ * console.log(clientStore.getContent());
+ * // -> [{pets: {fido: {species: 'dog'}}}, {"pathId": "petShop"}]
+ *
+ * synchronizer.destroy();
+ * server.destroy();
+ *
+ * // Remove file for the purposes of this demo.
+ * rmSync('petShop.json');
+ * ```
+ * @example
  * This example creates a WsServer with a custom listener that displays
  * information about the address of the client that connects to it.
  *
diff --git a/src/@types/synchronizers/synchronizer-ws-server/index.d.ts b/src/@types/synchronizers/synchronizer-ws-server/index.d.ts
@@ -3,6 +3,7 @@
 import type {Id, IdOrNull, Ids} from '../../common/index.d.ts';
 import type {Persister, Persists} from '../../persisters/index.d.ts';
 import type {IdAddedOrRemoved} from '../../store/index.d.ts';
+import type {MergeableStore} from '../../mergeable-store/index.d.ts';
 import type {WebSocketServer} from 'ws';
 
 /// PathIdsListener
@@ -55,5 +56,10 @@ export function createWsServer<
   >,
 >(
   webSocketServer: WebSocketServer,
-  createPersisterForPath?: (pathId: Id) => PathPersister | undefined,
+  createPersisterForPath?: (
+    pathId: Id,
+  ) =>
+    | PathPersister
+    | [PathPersister, (store: MergeableStore) => void]
+    | undefined,
 ): WsServer;
diff --git a/src/@types/synchronizers/synchronizer-ws-server/with-schemas/index.d.ts b/src/@types/synchronizers/synchronizer-ws-server/with-schemas/index.d.ts
@@ -9,6 +9,7 @@ import type {
   Persister,
   Persists,
 } from '../../../persisters/with-schemas/index.d.ts';
+import type {MergeableStore} from '../../../mergeable-store/with-schemas/index.d.ts';
 import type {WebSocketServer} from 'ws';
 
 /// PathIdsListener
@@ -63,5 +64,10 @@ export function createWsServer<
   >,
 >(
   webSocketServer: WebSocketServer,
-  createPersisterForPath?: (pathId: Id) => PathPersister | undefined,
+  createPersisterForPath?: (
+    pathId: Id,
+  ) =>
+    | PathPersister
+    | [PathPersister, (store: MergeableStore<Schemas>) => void]
+    | undefined,
 ): WsServer;
diff --git a/src/synchronizers/synchronizer-ws-server/index.ts b/src/synchronizers/synchronizer-ws-server/index.ts
@@ -37,11 +37,11 @@ import {
   ifPayloadValid,
   receivePayload,
 } from '../common.ts';
+import {ifNotUndefined, isArray} from '../../common/other.ts';
 import {IdSet2} from '../../common/set.ts';
 import type {MergeableStore} from '../../@types/mergeable-store/index.d.ts';
 import {createCustomSynchronizer} from '../index.ts';
 import {getListenerFunctions} from '../../common/listeners.ts';
-import {ifNotUndefined} from '../../common/other.ts';
 import {objFreeze} from '../../common/obj.ts';
 
 enum Sc {
@@ -50,6 +50,7 @@ enum Sc {
   Synchronizer = 2,
   Send = 3,
   Buffer = 4,
+  Then = 5,
 }
 enum ScState {
   Ready,
@@ -66,14 +67,20 @@ export const createWsServer = (<
   >,
 >(
   webSocketServer: WebSocketServer,
-  createPersisterForPath?: (pathId: Id) => PathPersister | undefined,
+  createPersisterForPath?: (
+    pathId: Id,
+  ) =>
+    | [PathPersister, (store: MergeableStore) => void]
+    | PathPersister
+    | undefined,
 ) => {
   type ServerClient = [
     state: ScState,
     persister: PathPersister,
     synchronizer: Synchronizer,
     send: (payload: string) => void,
     buffer: [],
+    then: (store: MergeableStore) => void,
   ];
 
   const pathIdListeners: IdSet2 = mapNew();
@@ -90,16 +97,18 @@ export const createWsServer = (<
     pathId: Id,
     clients: IdMap<WebSocket>,
   ) =>
-    ifNotUndefined(createPersisterForPath?.(pathId), (persister) => {
+    ifNotUndefined(createPersisterForPath?.(pathId), (persisterMaybeThen) => {
       serverClient[Sc.State] = 1;
-      serverClient[Sc.Persister] = persister;
+      serverClient[Sc.Persister] = isArray(persisterMaybeThen)
+        ? persisterMaybeThen[0]
+        : persisterMaybeThen;
       const messageHandler = getMessageHandler(
         SERVER_CLIENT_ID,
         clients,
         serverClient,
       );
       serverClient[Sc.Synchronizer] = createCustomSynchronizer(
-        persister.getStore() as MergeableStore,
+        serverClient[Sc.Persister].getStore() as MergeableStore,
         (toClientId, requestId, message, body) =>
           messageHandler(createPayload(toClientId, requestId, message, body)),
         (receive: Receive) =>
@@ -109,6 +118,9 @@ export const createWsServer = (<
         1,
       );
       serverClient[Sc.Buffer] = [];
+      serverClient[Sc.Then] = isArray(persisterMaybeThen)
+        ? persisterMaybeThen[1]
+        : (_) => 0;
     });
 
   const startServerClient = async (serverClient: ServerClient) => {
@@ -118,6 +130,9 @@ export const createWsServer = (<
       serverClient[Sc.Persister].startAutoSave,
       serverClient[Sc.Synchronizer].startSync,
     );
+    serverClient[Sc.Then](
+      serverClient[Sc.Persister].getStore() as MergeableStore,
+    );
     serverClient[Sc.State] = ScState.Ready;
   };
 
diff --git a/test/unit/mergeables/synchronizer-ws-server.test.ts b/test/unit/mergeables/synchronizer-ws-server.test.ts
@@ -2,6 +2,7 @@
 
 import type {Id, MergeableStore} from 'tinybase';
 import {WebSocket, WebSocketServer} from 'ws';
+import {readFileSync, writeFileSync} from 'fs';
 import type {WsServer} from 'tinybase/synchronizers/synchronizer-ws-server';
 import type {WsSynchronizer} from 'tinybase/synchronizers/synchronizer-ws-client';
 import {createFilePersister} from 'tinybase/persisters/persister-file';
@@ -10,7 +11,6 @@ import {createWsServer} from 'tinybase/synchronizers/synchronizer-ws-server';
 import {createWsSynchronizer} from 'tinybase/synchronizers/synchronizer-ws-client';
 import {join} from 'path';
 import {pause} from '../common/other.ts';
-import {readFileSync} from 'fs';
 import {resetHlc} from '../common/mergeable.ts';
 import tmp from 'tmp';
 
@@ -259,6 +259,106 @@ describe('Persistence', () => {
     wsServer.destroy();
   });
 
+  describe('single client to existing path', () => {
+    let serverStore: MergeableStore;
+
+    beforeEach(() => {
+      serverStore = createMergeableStore('ss');
+      writeFileSync(
+        join(tmpDir, 'p1.json'),
+        JSON.stringify([
+          [
+            {
+              t1: [
+                {
+                  r1: [
+                    {c1: [1, 'Nn1JUF-----7JQY8', 1003668370]},
+                    '',
+                    550994372,
+                  ],
+                },
+                '',
+                1072852846,
+              ],
+            },
+            '',
+            1771939739,
+          ],
+          [{}, '', 0],
+        ]),
+        'utf-8',
+      );
+    });
+
+    test('alters data prematurely', async () => {
+      const wsServer = createWsServer(
+        new WebSocketServer({port: 8049}),
+        (pathId) => {
+          serverStore.setValue('p', pathId);
+          return createPersister(serverStore, pathId);
+        },
+      );
+
+      const clientStore = createMergeableStore('s1');
+      const synchronizer = await createWsSynchronizer(
+        clientStore,
+        new WebSocket('ws://localhost:8049/p1'),
+      );
+      await synchronizer.startSync();
+
+      await pause();
+      expect(serverStore.getContent()).toEqual([{t1: {r1: {c1: 1}}}, {}]);
+
+      synchronizer.destroy();
+      wsServer.destroy();
+    });
+
+    test('alters data after path first persisted', async () => {
+      const wsServer = createWsServer(
+        new WebSocketServer({port: 8049}),
+        (pathId) => {
+          serverStore.setValue('p', pathId);
+          return [
+            createPersister(serverStore, pathId),
+            (store) => store.setValue('p', pathId),
+          ];
+        },
+      );
+
+      const clientStore = createMergeableStore('s1');
+      const synchronizer = await createWsSynchronizer(
+        clientStore,
+        new WebSocket('ws://localhost:8049/p1'),
+      );
+      await synchronizer.startSync();
+
+      await pause();
+      expect(serverStore.getContent()).toEqual([
+        {t1: {r1: {c1: 1}}},
+        {p: 'p1'},
+      ]);
+      expect(
+        JSON.parse(readFileSync(join(tmpDir, 'p1.json'), 'utf-8')),
+      ).toEqual([
+        [
+          {
+            t1: [
+              {r1: [{c1: [1, 'Nn1JUF-----7JQY8', 1003668370]}, '', 550994372]},
+              '',
+              1072852846,
+            ],
+          },
+          '',
+          1771939739,
+        ],
+        [{p: ['p1', 'Nn1JUF----05JWdY', 328213929]}, '', 1622699135],
+      ]);
+
+      synchronizer.destroy();
+      wsServer.destroy();
+    });
+  });
+
   test('multiple clients, one path', async () => {
     const serverStore = createMergeableStore('ss');
     const wsServer = createWsServer(
