feat(json-crdt): 🎸 start LocalHistory implementation

Author: streamich

package.json

@@ -83,7 +83,7 @@
     "arg": "^5.0.2",
     "hyperdyperid": "^1.2.0",
     "multibase": "^4.0.6",
-    "thingies": "^1.18.0"
+    "thingies": "^1.20.0"
   },
   "devDependencies": {
     "@automerge/automerge": "2.1.7",
@@ -136,6 +136,7 @@
     "lodash": "^4.17.21",
     "loro-crdt": "^0.4.1",
     "markdown-it": "^13.0.1",
+    "memfs": "^4.8.1",
     "messagepack": "^1.1.12",
     "msgpack-lite": "^0.1.26",
     "msgpack5": "^6.0.2",

src/json-crdt/history/LocalHistoryCrud.ts

@@ -0,0 +1,71 @@
+import {File, FileOptions} from "../file/File";
+import {CborEncoder} from "../../json-pack/cbor/CborEncoder";
+import type {CrudApi} from "memfs/lib/crud/types";
+import type {Locks} from "thingies/es2020/Locks";
+import type {Patch} from "../../json-crdt-patch";
+import type {PatchLog} from "./PatchLog";
+import type {LocalHistory} from "./types";
+
+export const genId = (octets: number = 8): string => {
+  const uint8 = crypto.getRandomValues(new Uint8Array(octets));
+  let hex = '';
+  for (let i = 0; i < octets; i++) hex += uint8[i].toString(16).padStart(2, '0');
+  return hex;
+};
+
+const STATE_FILE_NAME = 'state.seq.cbor';
+
+export class LocalHistoryCrud implements LocalHistory {
+  protected fileOpts: FileOptions = {
+    cborEncoder: new CborEncoder(),
+  };
+
+  constructor(
+    protected readonly crud: CrudApi,
+    protected readonly locks: Locks,
+  ) {}
+
+  public async create(collection: string[], log: PatchLog): Promise<{id: string}> {
+    // TODO: Remove `log.end`, just `log` should be enough.
+    const file = new File(log.end, log, this.fileOpts);
+    const blob = file.toBinary({
+      format: "seq.cbor",
+      model: 'binary',
+    });
+    const id = genId();
+    await this.lock(collection, id, async () => {
+      await this.crud.put([...collection, id], STATE_FILE_NAME, blob, {throwIf: 'exists'});
+    });
+    return {id};
+  }
+
+  public async read(collection: string[], id: string): Promise<{log: PatchLog, cursor: string}> {
+    const blob = await this.crud.get([...collection, id], STATE_FILE_NAME);
+    const {log} = File.fromSeqCbor(blob);
+    return {
+      log,
+      cursor: '',
+    };
+  }
+
+  public readHistory(collection: string[], id: string, cursor: string): Promise<{log: PatchLog, cursor: string}> {
+    throw new Error('Method not implemented.');
+  }
+
+  public update(collection: string[], id: string, patches: Patch[]): Promise<void> {
+    throw new Error('Method not implemented.');
+  }
+
+  public async delete(collection: string[], id: string): Promise<void> {
+    await this.lock(collection, id, async () => {
+      await this.crud.drop(collection, true);
+    });
+  }
+
+  protected async lock(collection: string[], id: string, fn: () => Promise<void>): Promise<void> {
+    const key = collection.join('/') + '/' + id;
+    await this.locks.lock(key, 250, 500)(async () => {
+      await fn();
+    });
+  }
+}

src/json-crdt/history/__tests__/LocalHistoryCrud.spec.ts

@@ -0,0 +1,62 @@
+import {memfs} from 'memfs';
+import {NodeCrud} from 'memfs/lib/node-to-crud';
+import {Locks} from 'thingies/es2020/Locks';
+import {LocalHistoryCrud} from '../LocalHistoryCrud';
+import {PatchLog} from '../PatchLog';
+import {Model} from '../../model';
+
+const setup = async () => {
+  const {fs, vol} = memfs();
+  const crud = new NodeCrud({fs: fs.promises, dir: '/'});
+  const locks = new Locks();
+  const local = new LocalHistoryCrud(crud, locks);
+  return {
+    fs,
+    vol,
+    crud,
+    locks,
+    local,
+  };
+};
+
+test('can create a new document', async () => {
+  const {local} = await setup();
+  const model = Model.withLogicalClock();
+  model.api.root({
+    foo: 'spam',
+  });
+  const log = PatchLog.fromNewModel(model);
+  const {id} = await local.create(['test'], log);
+  expect(typeof id).toBe('string');
+  expect(id.length > 6).toBe(true);
+  const {log: log2} = await local.read(['test'], id);
+  expect(log2.end.view()).toStrictEqual({foo: 'spam'});
+});
+
+test('throws on non-existing document', async () => {
+  const {local} = await setup();
+  try {
+    await local.read(['test'], 'asdfasdf');
+    throw new Error('FAIL');
+  } catch (err) {
+    expect((err as Error).message).toBe('Collection /test/asdfasdf does not exist');
+  }
+});
+
+test('can delete a document', async () => {
+  const {local} = await setup();
+  const model = Model.withLogicalClock();
+  model.api.root({
+    foo: 'spam',
+  });
+  const log = PatchLog.fromNewModel(model);
+  const {id} = await local.create(['test'], log);
+  await local.read(['test'], id);
+  await local.delete(['test'], id);
+  try {
+    await local.read(['test'], id);
+    throw new Error('FAIL');
+  } catch (err) {
+    expect((err as Error).message).toBe(`Collection /test/${id} does not exist`);
+  }
+});

src/json-crdt/history/types.ts

@@ -1,6 +1,6 @@
-import {Patch} from '../../json-crdt-patch';
-import {PatchLog} from '../history/PatchLog';
-import {Model} from '../model';
+import type {Patch} from '../../json-crdt-patch';
+import type {PatchLog} from '../history/PatchLog';
+import type {Model} from '../model';
 
 /**
  * A history of patches that have been applied to a model, stored on the
@@ -37,9 +37,11 @@ export interface RemoteHistory<Cursor> {
 }
 
 export interface LocalHistory {
-  load(id: string): Promise<EditingSessionHistory>;
-  // loadHistory(id: string): Promise<PatchLog>;
-  apply(id: string, patches: Patch[]): Promise<void>;
+  create(collection: string[], log: PatchLog): Promise<{id: string}>;
+  read(collection: string[], id: string): Promise<{log: PatchLog, cursor: string}>;
+  readHistory(collection: string[], id: string, cursor: string): Promise<{log: PatchLog, cursor: string}>;
+  update(collection: string[], id: string, patches: Patch[]): Promise<void>;
+  delete(collection: string[], id: string): Promise<void>;
 }
 
 export interface EditingSessionHistory {

yarn.lock

@@ -3997,6 +3997,13 @@ memfs@^3.4.1, memfs@^3.4.3:
   dependencies:
     fs-monkey "^1.0.4"
 
+memfs@^4.8.1:
+  version "4.8.1"
+  resolved "https://registry.yarnpkg.com/memfs/-/memfs-4.8.1.tgz#1e02c15c4397212a9a1b037fa4324c6f7dd45b47"
+  integrity sha512-7q/AdPzf2WpwPlPL4v1kE2KsJsHl7EF4+hAeVzlyanr2+YnR21NVn9mDqo+7DEaKDRsQy8nvxPlKH4WqMtiO0w==
+  dependencies:
+    tslib "^2.0.0"
+
 merge-descriptors@1.0.1:
   version "1.0.1"
   resolved "https://registry.yarnpkg.com/merge-descriptors/-/merge-descriptors-1.0.1.tgz#b00aaa556dd8b44568150ec9d1b953f3f90cbb61"
@@ -5348,10 +5355,10 @@ test-exclude@^6.0.0:
     glob "^7.1.4"
     minimatch "^3.0.4"
 
-thingies@^1.18.0:
-  version "1.18.0"
-  resolved "https://registry.yarnpkg.com/thingies/-/thingies-1.18.0.tgz#827141872d96f3c3c2c0b432ab0dfdb581b4b4ac"
-  integrity sha512-WiB26BQP0MF47Bbvbq0P19KpyfrvdTK07L8xnltobpZ/aJPmu52CBGhYjLsnFgjyawmusJ0gVkTplnnoz2hBkQ==
+thingies@^1.20.0:
+  version "1.20.0"
+  resolved "https://registry.yarnpkg.com/thingies/-/thingies-1.20.0.tgz#27bf93397c39c3ff36601197e8cf78f43b7b2319"
+  integrity sha512-WvXY4CjHp/Uim2Ri0daqu6jkNTHJTk1H8NvuMQiOL0mgtdkqoSH5fkENy2M6XnvsLOp5iwyPcbmokoBjVb4lnQ==
 
 thunky@^1.0.2:
   version "1.1.0"
@@ -5453,7 +5460,7 @@ tslib@^1.13.0, tslib@^1.8.1:
   resolved "https://registry.yarnpkg.com/tslib/-/tslib-1.14.1.tgz#cf2d38bdc34a134bcaf1091c41f6619e2f672d00"
   integrity sha512-Xni35NKzjgMrwevysHTCArtLDpPvye8zV/0E4EyYn43P7/7qvQwPh9BGkHewbMulVntbigmcT7rdX3BNo9wRJg==
 
-tslib@^2.0.1, tslib@^2.0.3, tslib@^2.1.0, tslib@^2.6.2:
+tslib@^2.0.0, tslib@^2.0.1, tslib@^2.0.3, tslib@^2.1.0, tslib@^2.6.2:
   version "2.6.2"
   resolved "https://registry.yarnpkg.com/tslib/-/tslib-2.6.2.tgz#703ac29425e7b37cd6fd456e92404d46d1f3e4ae"
   integrity sha512-AEYxH93jGFPn/a2iVAwW87VuUIkR1FVUKB77NwMF7nBTDkDrrT/Hpt/IrCJ0QXhW27jTBDcf5ZY7w6RiqTMw2Q==