feat: 🎸 add "blocks.listen" method

Author: streamich

src/server/__tests__/blocks.spec.ts

@@ -1,7 +1,9 @@
+import {of} from 'rxjs';
 import {Model} from '../../json-crdt';
 import {Value} from '../../reactive-rpc/common/messages/Value';
 import {RpcError, RpcErrorCodes} from '../../reactive-rpc/common/rpc/caller';
 import {setup} from './setup';
+import {tick, until} from '../../__tests__/util';
 
 describe('blocks.*', () => {
   describe('blocks.create', () => {
@@ -11,7 +13,7 @@ describe('blocks.*', () => {
       const {block} = (await caller.call('blocks.get', {id: 'my-block'}, {})).data;
       expect(block).toMatchObject({
         id: 'my-block',
-        seq: 0,
+        seq: -1,
         blob: expect.any(Uint8Array),
         created: expect.any(Number),
         updated: expect.any(Number),
@@ -282,4 +284,74 @@ describe('blocks.*', () => {
       expect(patches[2].blob).toStrictEqual(patch3.toBinary());
     });
   });
+
+  describe('blocks.listen', () => {
+    test('can listen for block changes', async () => {
+      const {call, caller} = setup();
+      await call('blocks.create', {id: 'my-block', patches: []});
+      await tick(11);
+      const emits: any[] = [];
+      caller.call$('blocks.listen', of({id: 'my-block'}), {}).subscribe((data) => emits.push(data));
+      const model = Model.withLogicalClock(); 
+      model.api.root({
+        text: 'Hell',
+      });
+      const patch1 = model.api.flush();
+      await tick(12);
+      expect(emits.length).toBe(0);
+      await call('blocks.edit', {id: 'my-block', patches: [{seq: 0, created: Date.now(), blob: patch1.toBinary()}]});
+      await until(() => emits.length === 1);
+      expect(emits.length).toBe(1);
+      expect(emits[0].data.patches.length).toBe(1);
+      expect(emits[0].data.patches[0].seq).toBe(0);
+      model.api.root({
+        text: 'Hello',
+      });
+      const patch2 = model.api.flush();
+      await tick(12);
+      expect(emits.length).toBe(1);
+      await call('blocks.edit', {id: 'my-block', patches: [{seq: 1, created: Date.now(), blob: patch2.toBinary()}]});
+      await until(() => emits.length === 2);
+      expect(emits.length).toBe(2);
+      expect(emits[1].data.patches.length).toBe(1);
+      expect(emits[1].data.patches[0].seq).toBe(1);
+    });
+
+    test('can subscribe before block is created', async () => {
+      const {call, caller} = setup();
+      const emits: any[] = [];
+      caller.call$('blocks.listen', of({id: 'my-block'}), {}).subscribe((data) => emits.push(data));
+      const model = Model.withLogicalClock(); 
+      model.api.root({
+        text: 'Hell',
+      });
+      const patch1 = model.api.flush();
+      await tick(12);
+      expect(emits.length).toBe(0);
+      await call('blocks.create', {id: 'my-block', patches: [
+        {
+          seq: 0,
+          created: Date.now(),
+          blob: patch1.toBinary(),
+        },
+      ]});
+      await until(() => emits.length === 1);
+      expect(emits.length).toBe(1);
+      expect(emits[0].data.patches.length).toBe(1);
+      expect(emits[0].data.patches[0].seq).toBe(0);
+      expect(emits[0].data.patches[0].blob).toStrictEqual(patch1.toBinary());
+    });
+
+    test('can receive deletion events', async () => {
+      const {call, caller} = setup();
+      const emits: any[] = [];
+      caller.call$('blocks.listen', of({id: 'my-block'}), {}).subscribe((data) => emits.push(data));
+      await call('blocks.create', {id: 'my-block', patches: []});
+      await until(() => emits.length === 1);
+      expect(emits[0].data.block.seq).toBe(-1);
+      await call('blocks.remove', {id: 'my-block'});
+      await until(() => emits.length === 2);
+      expect(emits[1].data.deleted).toBe(true);
+    });
+  });
 });

src/server/routes/blocks/index.ts

@@ -2,6 +2,7 @@ import {create} from './methods/create';
 import {get} from './methods/get';
 import {remove} from './methods/remove';
 import {edit} from './methods/edit';
+import {listen} from './methods/listen';
 import {Block, BlockId, BlockPatch, BlockSeq} from './schema';
 import type {RoutesBase, TypeRouter} from '../../../json-type/system/TypeRouter';
 import type {RouteDeps} from '../types';
@@ -20,5 +21,6 @@ export const blocks =
       ( get(d)
       ( remove(d)
       ( edit(d)
-      ( r ))))));
+      ( listen(d)
+      ( r )))))));
   };

src/server/routes/blocks/methods/listen.ts

@@ -0,0 +1,47 @@
+import {switchMap} from 'rxjs';
+import type {RoutesBase, TypeRouter} from '../../../../json-type/system/TypeRouter';
+import type {RouteDeps} from '../../types';
+import type {BlockId, BlockPatch, Block} from '../schema';
+
+export const listen =
+  ({services}: RouteDeps) =>
+  <R extends RoutesBase>(router: TypeRouter<R>) => {
+    const t = router.t;
+    const PatchType = t.Ref<typeof BlockPatch>('BlockPatch');
+
+    const Request = t.Object(
+      t.prop('id', t.Ref<typeof BlockId>('BlockId')).options({
+        title: 'Block ID',
+        description: 'The ID of the block to subscribe to.',
+      }),
+    );
+
+    const Response = t.Object(
+      t.propOpt('deleted', t.Boolean()).options({
+        title: 'Deleted',
+        description: 'Emitted only when the block is deleted.',
+      }),
+      t.propOpt('block', t.Ref<typeof Block>('Block')).options({
+        title: 'Block',
+        description: 'The whole block object, emitted only when the block is created.',
+      }),
+      t.propOpt('patches', t.Array(PatchType)).options({
+        title: 'Latest patches',
+        description: 'Patches that have been applied to the block.',
+      }),
+    );
+
+    const Func = t
+      .Function$(Request, Response)
+      .options({
+        title: 'Listen for block changes',
+        description: 'Subscribe to a block to receive updates when it changes.',
+      })
+      .implement((req$) => {
+        return req$.pipe(
+          switchMap(({id}) => services.pubsub.listen$(`__block:${id}`)),
+        ) as any;
+      });
+
+    return router.fn$('blocks.listen', Func);
+  };

src/server/services/blocks/BlocksServices.ts

@@ -1,7 +1,7 @@
 import {MemoryStore} from './MemoryStore';
 import {StorePatch} from './types';
-import type {Services} from '../Services';
 import {RpcError, RpcErrorCodes} from '../../../reactive-rpc/common/rpc/caller';
+import type {Services} from '../Services';
 
 export class BlocksServices {
   protected readonly store = new MemoryStore();
@@ -11,6 +11,13 @@ export class BlocksServices {
   public async create(id: string, patches: StorePatch[]) {
     const {store} = this;
     const {block} = await store.create(id, patches);
+    const data = {
+      block,
+      patches,
+    };
+    this.services.pubsub.publish(`__block:${id}`, data).catch(error => {
+      console.error('Error publishing block patches', error);
+    });
     return {block};
   }
 
@@ -24,6 +31,9 @@ export class BlocksServices {
 
   public async remove(id: string) {
     await this.store.remove(id);
+    this.services.pubsub.publish(`__block:${id}`, {deleted: true}).catch(error => {
+      console.error('Error publishing block deletion', error);
+    });
   }
 
   public async edit(id: string, patches: any[]) {
@@ -32,6 +42,9 @@ export class BlocksServices {
     const seq = patches[0].seq;
     const {store} = this;
     const {block} = await store.edit(id, patches);
+    this.services.pubsub.publish(`__block:${id}`, {patches}).catch(error => {
+      console.error('Error publishing block patches', error);
+    });
     const expectedBlockSeq = seq + patches.length - 1;
     const hadConcurrentEdits = block.seq !== expectedBlockSeq;
     let patchesBack: StorePatch[] = [];

src/server/services/blocks/MemoryStore.ts

@@ -18,16 +18,15 @@ export class MemoryStore implements types.Store {
     if (!Array.isArray(patches)) throw new Error('NO_PATCHES');
     if (this.blocks.has(id)) throw new Error('BLOCK_EXISTS');
     const model = Model.withLogicalClock();
-    let seq = 0;
+    let seq = -1;
     const now = Date.now();
     if (patches.length) {
       for (const patch of patches) {
+        seq++;
         if (seq !== patch.seq) throw new Error('PATCHES_OUT_OF_ORDER');
         model.applyPatch(Patch.fromBinary(patch.blob));
         if (patch.created > now) patch.created = now;
-        seq++;
       }
-      seq--;
     }
     const block: types.StoreBlock = {
       id,