feat: 🎸 use one path for updates and creates

Author: streamich

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

@@ -141,7 +141,7 @@ export const runBlockTests = (_setup: ApiTestSetup, params: {staticOnly?: true}
         stop();
       });
 
-      test('throws BLOCK_NOT_FOUND when "create" flag missing', async () => {
+      test('throws NOT_FOUND when "create" flag missing', async () => {
         const {call, stop} = await setup();
         const id = getId();
         const model = Model.withLogicalClock();

src/json-crdt-repo/local/server-crud/ServerCrudLocalHistorySync.ts

@@ -50,12 +50,27 @@ export class ServerCrudLocalHistorySync {
           const core = this.core;
           if (!core.connected$.getValue()) return false;
           const meta = await this.getMeta(collection, id);
-          const isNewBlock = meta.time < 1;
-          if (isNewBlock) {
-            await this.pushNewBlock(collection, id);
-          } else {
-            await this.pushExistingBlock(collection, id, meta.time);
+          const blob = await core.read(collection, id);
+          const {history} = core.decoder.decode(blob, {format: 'seq.cbor', history: true});
+          const patches: RemoteBlockPatch[] = [];
+          let time = 0;
+          // TODO: perf: use a binary search to find the first patch to sync.
+          history!.patches.forEach(({v: patch}) => {
+            const id = patch.getId();
+            if (!id) return;
+            // TODO: also filter in SESSION.GLOBAL patches.
+            if (id.sid === core.sid && id.time > meta.time) {
+              patches.push({blob: patch.toBinary()});
+              time = id.time;
+            }
+          });
+          if (!patches.length) {
+            await this.putMeta(collection, id, {time, ts: Date.now()});
+            return true;
           }
+          const remoteId = [...collection, id].join('/');
+          await this.core.remote.update(remoteId, patches);
+          await this.putMeta(collection, id, {time, ts: Date.now()});
           await this.markTidy(collection, id);
           return true;
         });
@@ -66,51 +81,6 @@ export class ServerCrudLocalHistorySync {
     });
   }
 
-  private async pushNewBlock(collection: string[], id: string): Promise<void> {
-    const core = this.core;
-    const blob = await core.read(collection, id);
-    const {history} = core.decoder.decode(blob, {format: 'seq.cbor', history: true});
-    const patches: RemoteBlockPatch[] = [];
-    let time = 0;
-    history!.patches.forEach(({v: patch}) => {
-      const id = patch.getId();
-      if (!id) return;
-      if (id.sid === core.sid) {
-        patches.push({blob: patch.toBinary()});
-        time = id.time;
-      }
-    });
-    if (!patches.length) return;
-    const remoteId = [...collection, id].join('/');
-    await this.core.remote.create(remoteId, patches);
-    await this.putMeta(collection, id, {time, ts: Date.now()});
-  }
-
-  /**
-   * @todo Unify this with `pushNewBlock`.
-   */
-  private async pushExistingBlock(collection: string[], id: string, syncedTime: number): Promise<void> {
-    const core = this.core;
-    const blob = await core.read(collection, id);
-    const {history} = core.decoder.decode(blob, {format: 'seq.cbor', history: true});
-    const patches: RemoteBlockPatch[] = [];
-    let time = 0;
-    // TODO: perf: use a binary search to find the first patch to sync.
-    history!.patches.forEach(({v: patch}) => {
-      const id = patch.getId();
-      if (!id) return;
-      // TODO: also filter in SESSION.GLOBAL patches.
-      if (id.sid === core.sid && id.time > syncedTime) {
-        patches.push({blob: patch.toBinary()});
-        time = id.time;
-      }
-    });
-    if (!patches.length) return;
-    const remoteId = [...collection, id].join('/');
-    await this.core.remote.update(remoteId, patches);
-    await this.putMeta(collection, id, {time, ts: Date.now()});
-  }
-
   public async lock<T>(
     {
       collection,

src/json-crdt-repo/local/server-crud/__tests__/ServerCrudLocalHistory.spec.ts

@@ -157,6 +157,10 @@ describe('.create()', () => {
         await tick(15);
         throw new Error('Remote call failed');
       };
+      remote.remote.update = async (...args) => {
+        await tick(15);
+        throw new Error('Remote call failed');
+      };
       return setup({remote});
     };
 
@@ -218,6 +222,10 @@ describe('.create()', () => {
         await tick(500);
         throw new Error('something went wrong');
       };
+      remote.remote.update = async (...args) => {
+        await tick(500);
+        throw new Error('something went wrong');
+      };
       return setup({
         remote,
         local: {
@@ -286,6 +294,10 @@ describe('.create()', () => {
         await tick(200);
         return create(...args);
       };
+      remote.remote.update = async (...args) => {
+        await tick(200);
+        return create(...args);
+      };
       return setup({
         remote,
         local: {

src/json-crdt-repo/remote/DemoServerRemoteHistory.ts

@@ -78,6 +78,7 @@ export class DemoServerRemoteHistory
     patches: Pick<DemoServerPatch, 'blob'>[],
   ): Promise<{patches: Omit<DemoServerPatch, 'blob'>[]}> {
     const res = await this.client.call('block.upd', {
+      create: true,
       id,
       patches: patches.map((patch) => ({
         blob: patch.blob,