Merge pull request #10 from jsonjoy-com/local-server-crud

Local repo, edit session and demo server improvements

Author: streamich

.gitignore

@@ -127,3 +127,4 @@ typedocs/
 out.bin
 
 /gh-pages/
+/db/

package.json

@@ -43,7 +43,7 @@
     "prettier:check": "prettier --ignore-path .gitignore --list-different 'src/**/*.{ts,tsx,js,jsx}'",
     "lint": "yarn tslint",
     "tslint": "tslint 'src/**/*.{js,jsx,ts,tsx}' -t verbose --project .",
-    "clean": "rimraf lib typedocs coverage gh-pages yarn-error.log",
+    "clean": "rimraf lib typedocs coverage gh-pages yarn-error.log db",
     "build": "tsc --project tsconfig.build.json --module commonjs --target es2020 --outDir lib",
     "jest": "node -r ts-node/register ./node_modules/.bin/jest",
     "test": "jest --maxWorkers 7",
@@ -72,18 +72,23 @@
   },
   "dependencies": {
     "@jsonjoy.com/jit-router": "^1.0.1",
-    "@jsonjoy.com/json-pack": "^1.0.2",
-    "@jsonjoy.com/util": "^1.0.0",
-    "json-joy": "^15.8.0",
-    "memfs": "^5.0.0-next.1",
-    "sonic-forest": "^1.0.0",
+    "@jsonjoy.com/json-pack": "^1.1.0",
+    "@jsonjoy.com/util": "^1.3.0",
+    "abstract-level": "^2.0.0",
+    "browser-level": "^1.0.1",
+    "fs-zoo": "^1.1.0",
+    "json-joy": "^16.24.0",
+    "memfs": "^4.11.0",
+    "memory-level": "^1.0.0",
+    "sonic-forest": "^1.0.3",
     "thingies": "^2.1.0"
   },
   "devDependencies": {
     "@types/benchmark": "^2.1.5",
     "@types/jest": "^29.5.12",
     "@types/ws": "^8.5.10",
     "benchmark": "^2.1.4",
+    "classic-level": "^1.4.1",
     "jest": "^29.7.0",
     "prettier": "^3.2.5",
     "rimraf": "^5.0.5",
@@ -95,7 +100,7 @@
     "tslint-config-common": "^1.6.2",
     "typedoc": "^0.25.13",
     "typescript": "^5.4.5",
-    "uWebSockets.js": "uNetworking/uWebSockets.js#v20.23.0",
+    "uWebSockets.js": "uNetworking/uWebSockets.js#v20.44.0",
     "websocket": "^1.0.34",
     "ws": "^8.16.0"
   },

src/__demos__/json-crdt-server/__tests__/block.spec.ts

@@ -1,5 +1,15 @@
-import {setup} from './setup';
+import {setupMemory, setupLevelMemory, setupLevelClassic} from './setup';
 import {runBlockTests} from '../../../__tests__/json-crdt-server/block';
 import type {ApiTestSetup} from '../../../common/rpc/__tests__/runApiTests';
 
-runBlockTests(setup as ApiTestSetup);
+describe('MemoryStore', () => {
+  runBlockTests(setupMemory as ApiTestSetup);
+});
+
+describe('LevelStore(memory-level)', () => {
+  runBlockTests(setupLevelMemory as ApiTestSetup);
+});
+
+describe('LevelStore(classic-level)', () => {
+  runBlockTests(setupLevelClassic as ApiTestSetup);
+});

src/__demos__/json-crdt-server/__tests__/setup.ts

@@ -1,9 +1,14 @@
+import {MemoryLevel} from 'memory-level';
 import {buildE2eClient} from '../../../common/testing/buildE2eClient';
 import {createCaller} from '../routes';
+import {LevelStore} from '../services/blocks/store/level/LevelStore';
 import {Services} from '../services/Services';
+import {ClassicLevel} from 'classic-level';
+import {MemoryStore} from '../services/blocks/store/MemoryStore';
+import type {Store} from '../services/blocks/store/types';
 
-export const setup = async () => {
-  const services = new Services();
+export const setup = async (store: Store = new MemoryStore(), close?: () => Promise<void>) => {
+  const services = new Services({store});
   const {caller} = createCaller(services);
   const {client} = buildE2eClient(caller, {
     writerDefaultBufferKb: [1, 32],
@@ -14,8 +19,31 @@ export const setup = async () => {
   });
   const call = client.call.bind(client);
   const call$ = client.call$.bind(client);
-  const stop = () => {};
+  const stop = async (): Promise<void> => {
+    await close?.();
+  };
   return {call, call$, stop};
 };
 
+export const setupMemory = async () => {
+  const store = new MemoryStore();
+  return setup(store);
+};
+
+export const setupLevelMemory = async () => {
+  const kv = new MemoryLevel<string, Uint8Array>({
+    keyEncoding: 'utf8',
+    valueEncoding: 'view',
+  });
+  const store = new LevelStore(<any>kv);
+  return setup(store);
+};
+
+export const setupLevelClassic = async () => {
+  const kv = new ClassicLevel<string, Uint8Array>('./db', {valueEncoding: 'view'});
+  await kv.open();
+  const store = new LevelStore(<any>kv);
+  return setup(store, async () => kv.close());
+};
+
 export type JsonCrdtTestSetup = typeof setup;

src/__demos__/json-crdt-server/main-uws.ts

@@ -15,3 +15,6 @@ const app = new RpcApp<MyCtx>({
   port: +(process.env.PORT || 9999),
 });
 app.startWithDefaults();
+
+// tslint:disable-next-line:no-console
+console.log(app + '');

src/__demos__/json-crdt-server/routes/block/index.ts

@@ -1,22 +1,28 @@
 import {new_} from './methods/new';
 import {get} from './methods/get';
+import {view} from './methods/view';
 import {upd} from './methods/upd';
 import {del} from './methods/del';
 import {scan} from './methods/scan';
 import {listen} from './methods/listen';
 import {
-  Block,
   BlockId,
   BlockPatch,
   BlockPatchPartial,
   BlockPatchPartialReturn,
   BlockCur,
-  BlockNew,
   BlockSnapshot,
   NewBlockSnapshotResponse,
   BlockEvent,
+  BlockBatch,
+  BlockBatchPartial,
+  BlockBatchPartialReturn,
+  BlockBatchSeq,
+  BlockSnapshotReturn,
+  Block,
 } from './schema';
 import type {RouteDeps, Router, RouterBase} from '../types';
+import {pull} from './methods/pull';
 
 export const block =
   (d: RouteDeps) =>
@@ -25,25 +31,32 @@ export const block =
 
     system.alias('BlockId', BlockId);
     system.alias('BlockCur', BlockCur);
-    system.alias('BlockNew', BlockNew);
+    system.alias('BlockBatchSeq', BlockBatchSeq);
+
     system.alias('Block', Block);
 
+    system.alias('BlockSnapshotReturn', BlockSnapshotReturn);
     system.alias('BlockSnapshot', BlockSnapshot);
     system.alias('NewBlockSnapshotResponse', NewBlockSnapshotResponse);
 
     system.alias('BlockPatch', BlockPatch);
     system.alias('BlockPatchPartial', BlockPatchPartial);
     system.alias('BlockPatchPartialReturn', BlockPatchPartialReturn);
+    system.alias('BlockBatch', BlockBatch);
+    system.alias('BlockBatchPartial', BlockBatchPartial);
+    system.alias('BlockBatchPartialReturn', BlockBatchPartialReturn);
 
     system.alias('BlockEvent', BlockEvent);
 
     // prettier-ignore
     return (
     ( new_(d)
     ( get(d)
+    ( view(d)
     ( upd(d)
     ( del(d)
     ( listen(d)
     ( scan(d)
-    ( r ))))))));
+    ( pull(d)
+    ( r ))))))))));
   };

src/__demos__/json-crdt-server/routes/block/methods/get.ts

@@ -1,5 +1,5 @@
 import {ResolveType} from 'json-joy/lib/json-type';
-import {BlockRef, BlockIdRef} from '../schema';
+import {BlockIdRef, BlockRef} from '../schema';
 import type {RouteDeps, Router, RouterBase} from '../../types';
 
 export const get =
@@ -12,27 +12,19 @@ export const get =
       }),
     );
 
-    const Response = t.Object(t.prop('block', BlockRef));
+    // prettier-ignore
+    const Response = t.Object(
+      t.prop('block', BlockRef),
+    );
 
     const Func = t.Function(Request, Response).options({
       title: 'Read Block',
       intro: 'Retrieves a block by ID.',
     });
 
     return r.prop('block.get', Func, async ({id}) => {
-      const {snapshot} = await services.blocks.get(id);
-      const response: ResolveType<typeof Response> = {
-        block: {
-          id: snapshot.id,
-          ts: snapshot.created,
-          snapshot: {
-            blob: snapshot.blob,
-            cur: snapshot.seq,
-            ts: snapshot.created,
-          },
-          tip: [],
-        },
-      };
+      const {block} = await services.blocks.get(id);
+      const response: ResolveType<typeof Response> = {block};
       return response;
     });
   };

src/__demos__/json-crdt-server/routes/block/methods/listen.ts

@@ -1,4 +1,4 @@
-import {map, switchMap, tap} from 'rxjs';
+import {map, switchMap} from 'rxjs';
 import {BlockEventRef, BlockIdRef} from '../schema';
 import type {RouteDeps, Router, RouterBase} from '../../types';
 

src/__demos__/json-crdt-server/routes/block/methods/new.ts

@@ -1,10 +1,4 @@
-import {
-  BlockIdRef,
-  BlockPatchPartialRef,
-  BlockPatchPartialReturnRef,
-  BlockNewRef,
-  NewBlockSnapshotResponseRef,
-} from '../schema';
+import {BlockIdRef, BlockBatchPartialRef, BlockSnapshotReturnRef} from '../schema';
 import type {RouteDeps, Router, RouterBase} from '../../types';
 
 export const new_ =
@@ -16,26 +10,20 @@ export const new_ =
         description:
           'The ID of the new block. Must be a unique ID, if the block already exists it will return an error.',
       }),
-      t.prop('patches', t.Array(BlockPatchPartialRef)).options({
-        title: 'Patches',
-        description: 'The patches to apply to the document.',
+      t.propOpt('batch', BlockBatchPartialRef).options({
+        title: 'Batch',
+        description: 'A collection of patches to apply to the new block.',
       }),
     );
 
-    const Response = t
-      .Object(
-        t.prop('block', BlockNewRef),
-        t.prop('snapshot', NewBlockSnapshotResponseRef),
-        t.prop('patches', t.Array(BlockPatchPartialReturnRef)).options({
-          title: 'Patches',
-          description: 'The list of patches to apply to the newly created block.',
-        }),
-      )
-      .options({
-        title: 'New block creation response',
-        description:
-          'The response object for the new block creation, contains server generated metadata without blobs supplied by the client.',
-      });
+    // prettier-ignore
+    const Response = t.Object(
+      t.prop('snapshot', BlockSnapshotReturnRef),
+    ).options({
+      title: 'New block creation response',
+      description:
+        'The response object for the new block creation, contains server generated metadata without blobs supplied by the client.',
+    });
 
     const Func = t.Function(Request, Response).options({
       title: 'Create Block',
@@ -44,20 +32,15 @@ export const new_ =
         'Creates a new block out of supplied patches. A block starts empty with an `undefined` state, and patches are applied to it.',
     });
 
-    return r.prop('block.new', Func, async ({id, patches}) => {
-      const res = await services.blocks.create(id, patches);
+    return r.prop('block.new', Func, async ({id, batch}) => {
+      const {block} = await services.blocks.create(id, batch);
+      const snapshot = block.snapshot;
       return {
-        block: {
-          id: res.snapshot.id,
-          ts: res.snapshot.created,
-        },
         snapshot: {
-          cur: res.snapshot.seq,
-          ts: res.snapshot.created,
+          id,
+          seq: snapshot.seq,
+          ts: snapshot.ts,
         },
-        patches: res.patches.map((patch) => ({
-          ts: patch.created,
-        })),
       };
     });
   };

src/__demos__/json-crdt-server/routes/block/methods/pull.ts

@@ -0,0 +1,51 @@
+import {BlockIdRef, BlockCurRef, BlockBatchRef, BlockSnapshotRef} from '../schema';
+import type {RouteDeps, Router, RouterBase} from '../../types';
+
+export const pull =
+  ({t, services}: RouteDeps) =>
+  <R extends RouterBase>(r: Router<R>) => {
+    // prettier-ignore
+    const Request = t.Object(
+      t.prop('id', BlockIdRef).options({
+        title: 'Block ID',
+        description: 'The ID of the block.',
+      }),
+      t.prop('seq', BlockCurRef).options({
+        title: 'Last Known Sequence Number',
+        description: 'The sequence number that the client is caught up to. If '
+          + 'the client is not caught up to the latest state of the block, the '
+          + 'server will return a list of batches that the client needs to apply '
+          + 'to get to the latest state. If the client is too far behind, the '
+          + 'server will return a snapshot of the block.'
+          + '\n\n'
+          + 'The initial value should be `-1`.',
+      }),
+      t.propOpt('create', t.bool).options({
+        title: 'Create Block',
+        description: 'Whether to create a new block if it does not exist.',
+      }),
+    );
+
+    // prettier-ignore
+    const Response = t.Object(
+      t.prop('batches', t.Array(BlockBatchRef)).options({
+        title: 'Batches',
+        description: 'List of batches that the client need to apply to the local state. ' +
+          'Or, if `snapshot` is provided, the list of batches that the client need to apply to the snapshot to get to the latest state.',
+      }),
+      t.propOpt('snapshot', BlockSnapshotRef).options({
+        title: 'Snapshot',
+        description: 'The state of the block right before the first batch in the result.',
+      }),
+    );
+
+    const Func = t.Function(Request, Response).options({
+      title: 'Pull Block',
+      intro: 'Catch up to the latest state of a block.',
+      description: 'Returns a list of most recent change batches or a snapshot of a block.',
+    });
+
+    return r.prop('block.pull', Func, async ({id, seq, create}) => {
+      return await services.blocks.pull(id, seq, !!create);
+    });
+  };