fix: 🐛 filter events correctly

Author: streamich

package.json

@@ -77,7 +77,7 @@
     "abstract-level": "^2.0.0",
     "browser-level": "^1.0.1",
     "fs-zoo": "^1.1.0",
-    "json-joy": "^16.23.2",
+    "json-joy": "^16.24.0",
     "memfs": "^4.11.0",
     "memory-level": "^1.0.0",
     "sonic-forest": "^1.0.3",

src/json-crdt-repo/local/level/__tests__/LevelLocalRepo.events.spec.ts

@@ -0,0 +1,93 @@
+import {Model, s} from 'json-joy/lib/json-crdt';
+import {setup} from './setup';
+import {BehaviorSubject} from 'rxjs';
+import {until} from 'thingies';
+import {LocalRepoRebaseEvent} from '../../types';
+
+describe('events', () => {
+  test('can emit to another tab', async () => {
+    const kit = await setup({local: {connected$: new BehaviorSubject(false)}});
+    const local2 = await kit.createLocal();
+    const bus1 = kit.pubsub;
+    const bus2 = local2.pubsub;
+    const msgs: any[] = [];
+    bus1.bus$.subscribe((msg) => {
+      msgs.push(msg);
+    });
+    bus2.bus$.subscribe((msg) => {
+      msgs.push(msg);
+    });
+    (bus1 as any).pub({foo: 'bar', bin: new Uint8Array([1, 2, 3])});
+    await until(() => msgs.length === 2);
+    expect(msgs).toEqual([{foo: 'bar', bin: new Uint8Array([1, 2, 3])}, {foo: 'bar', bin: new Uint8Array([1, 2, 3])}]);
+    await bus1.end();
+    await bus2.end();
+  });
+  
+  test('emits "rebase" event (in same tab)', async () => {
+    const kit = await setup({local: {connected$: new BehaviorSubject(false)}});
+    const model = Model.create(s.obj({foo: s.con(1)}), kit.sid);
+    const patch = model.api.flush();
+    const res1 = await kit.local.sync({
+      id: kit.blockId,
+      patches: [patch],
+    });
+    const events: LocalRepoRebaseEvent[] = [];
+    const sub = kit.local.change$(kit.blockId).subscribe((event) => {
+      if ((event as LocalRepoRebaseEvent).rebase) events.push(event as LocalRepoRebaseEvent);
+    });
+    const model2 = model.clone();
+    model.api.obj([]).set({foo: 2});
+    const patch2 = model.api.flush();
+    const res2 = await kit.local.sync({
+      id: kit.blockId,
+      patches: [patch2],
+      cursor: res1.cursor,
+    });
+    const get1 = await kit.local.get({id: kit.blockId});
+    expect(get1.model.view()).toEqual({foo: 2});
+    await until(() => events.length === 1);
+    const event = events[0];
+    expect(model2.view()).toEqual({foo: 1});
+    for (const patch of event.rebase) {
+      model2.applyPatch(patch);
+    }
+    expect(model2.view()).toEqual({foo: 2});
+    sub.unsubscribe();
+    await kit.stop();
+  });
+
+  test('emits "rebase" event (across tabs)', async () => {
+    const kit = await setup({local: {connected$: new BehaviorSubject(false)}});
+    const local2 = await kit.createLocal();
+    const model = Model.create(s.obj({foo: s.con(1)}), kit.sid);
+    const patch = model.api.flush();
+    const res1 = await kit.local.sync({
+      id: kit.blockId,
+      patches: [patch],
+    });
+    const events: LocalRepoRebaseEvent[] = [];
+    const sub = local2.local.change$(kit.blockId).subscribe((event) => {
+      if ((event as LocalRepoRebaseEvent).rebase) events.push(event as LocalRepoRebaseEvent);
+    });
+    const model2 = model.clone();
+    model.api.obj([]).set({foo: 2});
+    const patch2 = model.api.flush();
+    const res2 = await kit.local.sync({
+      id: kit.blockId,
+      patches: [patch2],
+      cursor: res1.cursor,
+    });
+    const get1 = await kit.local.get({id: kit.blockId});
+    expect(get1.model.view()).toEqual({foo: 2});
+    await until(() => events.length === 1);
+    const event = events[0];
+    expect(model2.view()).toEqual({foo: 1});
+    for (const patch of event.rebase) {
+      model2.applyPatch(patch);
+    }
+    expect(model2.view()).toEqual({foo: 2});
+    sub.unsubscribe();
+    await kit.stop();
+  });
+});

src/json-crdt-repo/local/level/__tests__/setup.ts

@@ -25,7 +25,8 @@ export const setup = async (
   }) as unknown as BinStrLevel;
   const blockId = [...col, id];
   const createLocal = (sid: number = 12345678) => {
-    const pubsub = createPubsub('test-' + blockId.join('/')) as LevelLocalRepoPubSub;
+    const busName = 'test-' + id;
+    const pubsub = createPubsub(busName) as LevelLocalRepoPubSub;
     const local = new LevelLocalRepo({
       kv,
       locks,

src/json-crdt-repo/pubsub/__tests__/index.spec.ts

@@ -6,12 +6,15 @@ test('can emit to another instance', async () => {
   const bus1 = pubsub(id);
   const bus2 = pubsub(id);
   const msgs: any[] = [];
+  bus1.bus$.subscribe((msg) => {
+    msgs.push(msg);
+  });
   bus2.bus$.subscribe((msg) => {
     msgs.push(msg);
   });
   bus1.pub('hello');
-  await until(() => msgs.length === 1);
-  expect(msgs).toEqual(['hello']);
+  await until(() => msgs.length === 2);
+  expect(msgs).toEqual(['hello', 'hello']);
   await bus1.end();
   await bus2.end();
 });

yarn.lock

@@ -2020,10 +2020,10 @@ jsesc@^2.5.1:
   resolved "https://registry.yarnpkg.com/jsesc/-/jsesc-2.5.2.tgz#80564d2e483dacf6e8ef209650a67df3f0c283a4"
   integrity sha512-OYu7XEzjkCQ3C5Ps3QIZsQfNpqoJyZZA99wd9aWd05NCtC5pWOkShK2mkL6HXQR6/Cy2lbNdPlZBpuQHXE63gA==
 
-json-joy@^16.23.2:
-  version "16.23.2"
-  resolved "https://registry.yarnpkg.com/json-joy/-/json-joy-16.23.2.tgz#e7132edd50aa7338f2d7d335e5c08e85f306d57a"
-  integrity sha512-DgCJYEpU91IijGXWXBTtZJICBDBiSIV0c1Lvu5CYm4WjcT42Kz4SXxRC6aS8FJKNj+JR2VsjPoi0V0YA9hhdEA==
+json-joy@^16.24.0:
+  version "16.24.0"
+  resolved "https://registry.yarnpkg.com/json-joy/-/json-joy-16.24.0.tgz#ae68c10524d9cde99a5363baf46fe8e3f6aca4c4"
+  integrity sha512-Y9yvAOCLl8ar/tHgstxWi7snXj3Vl/6wWtSb2YH1QNRxRHYJSC2dmJm+xolXYkxoD41lQA898n7TLJgiQceToQ==
   dependencies:
     arg "^5.0.2"
     hyperdyperid "^1.2.0"