feat: 🎸 allow local ops to be applied using .applyPatch()

Author: streamich

src/json-crdt-repo/JsonCrdtRepo.ts

@@ -18,6 +18,7 @@ export interface JsonCrdtRepoOpts {
 export class JsonCrdtRepo {
   public readonly sessions: EditSessionFactory;
   public readonly opts: JsonCrdtRepoOpts;
+  public readonly remote: DemoServerRemoteHistory;
 
   constructor(opts: Partial<JsonCrdtRepoOpts>) {
     this.opts = {
@@ -26,7 +27,7 @@ export class JsonCrdtRepo {
       ...opts,
     };
     const client = createBinaryWsRpcClient(this.opts.wsUrl) as DemoServerClient;
-    const remote = new DemoServerRemoteHistory(client);
+    this.remote = new DemoServerRemoteHistory(client);
     const kv: BinStrLevel = new BrowserLevel(this.opts.name, {
       keyEncoding: 'utf8',
       valueEncoding: 'view',
@@ -39,7 +40,7 @@ export class JsonCrdtRepo {
       kv,
       locks,
       sid,
-      rpc: remote,
+      rpc: this.remote,
       pubsub,
       connected$,
     });

src/json-crdt-repo/session/EditSession.ts

@@ -34,11 +34,21 @@ export class EditSession {
     protected readonly session: number = Math.floor(Math.random() * 0x7fffffff),
   ) {
     this.log = new Log(() => this.start.clone());
-    const flushUnsubscribe = this.log.end.api.onFlush.listen((a) => {
+    const api = this.log.end.api;
+    const flushUnsubscribe = api.onFlush.listen((a) => {
       this.syncLog();
     });
+    const patchUnsubscribe = api.onPatch.listen((patch) => {
+      const id = patch.getId();
+      if (!id) return;
+      const clock = this.model.clock;
+      if (id.sid === clock.sid && clock.time >= id.time) {
+        this.syncLog();
+      }
+    });
     this._stop$.pipe(first()).subscribe(() => {
       flushUnsubscribe();
+      patchUnsubscribe();
     });
     this.repo.change$(this.id).pipe(takeUntil(this._stop$)).subscribe(this.onEvent);
   }
@@ -114,8 +124,10 @@ export class EditSession {
 
   public syncLog(): void {
     if (!this.log.patches.size()) return;
-    this.sync().then((error) => {
-      this.onsyncerror?.(error);
+    this._syncRace(() => {
+      this.sync().then((error) => {
+        this.onsyncerror?.(error);
+      });
     });
   }
 

src/json-crdt-repo/session/__tests__/EditSession.del.spec.ts

@@ -1,8 +1,8 @@
 import {s} from 'json-joy/lib/json-crdt';
 import {setup} from './setup';
+import {EditSessionFactory} from '../EditSessionFactory';
 import {until} from 'thingies/lib/until';
 import {of} from 'thingies';
-import {EditSessionFactory} from '../EditSessionFactory';
 import {BehaviorSubject} from 'rxjs';
 import {Testbed} from '../../__tests__/testbed';