BC-5736 - Redis for tldraw

.github/workflows/dependency-review.yml

@@ -13,7 +13,6 @@ jobs:
       - name: 'Dependency Review'
         uses: actions/dependency-review-action@v3
         with:
-          allow-licenses: AGPL-3.0-only, LGPL-3.0, MIT, Apache-2.0, BSD-2-Clause, BSD-3-Clause, ISC, X11, 0BSD, GPL-3.0 AND BSD-3-Clause-Clear, Unlicense
-          allow-dependencies-licenses: 'pkg:npm/parse-mongo-url'
+          allow-licenses: AGPL-3.0-only, LGPL-3.0, MIT, Apache-2.0, BSD-2-Clause, BSD-3-Clause, ISC, X11, 0BSD, GPL-3.0, Unlicense
           # temporarily ignore dependency error for upgrade mongodb 4.9 to 4.11, remove when mikroORM is upgraded to 5.9
           allow-ghsas: 'GHSA-vxvm-qww3-2fh7'

apps/server/src/modules/tldraw/config.ts

@@ -10,6 +10,7 @@ export interface TldrawConfig {
 	FEATURE_TLDRAW_ENABLED: boolean;
 	TLDRAW_PING_TIMEOUT: number;
 	TLDRAW_GC_ENABLED: number;
+	REDIS_URI: string;
 }
 
 const tldrawConnectionString: string = Configuration.get('TLDRAW_DB_URL') as string;
@@ -24,6 +25,7 @@ const tldrawConfig = {
 	CONNECTION_STRING: tldrawConnectionString,
 	TLDRAW_PING_TIMEOUT: Configuration.get('TLDRAW__PING_TIMEOUT') as number,
 	TLDRAW_GC_ENABLED: Configuration.get('TLDRAW__GC_ENABLED') as boolean,
+	REDIS_URI: Configuration.has('REDIS_URI') ? (Configuration.get('REDIS_URI') as string) : 'redis://127.0.0.1:6379',
 };
 
 export const SOCKET_PORT = Configuration.get('TLDRAW__SOCKET_PORT') as number;

apps/server/src/modules/tldraw/controller/api-test/tldraw.ws.api.spec.ts

@@ -22,6 +22,7 @@ describe('WebSocketController (WsAdapter)', () => {
 		const testingModule = await Test.createTestingModule({
 			imports: [TldrawWsTestModule],
 		}).compile();
+
 		gateway = testingModule.get<TldrawWs>(TldrawWs);
 		app = testingModule.createNestApplication();
 		app.useWebSocketAdapter(new WsAdapter(app));

apps/server/src/modules/tldraw/controller/tldraw.controller.spec.ts

@@ -1,7 +1,7 @@
 import { createMock } from '@golevelup/ts-jest';
 import { Test, TestingModule } from '@nestjs/testing';
 import { TldrawController } from './tldraw.controller';
-import { TldrawService } from '../service/tldraw.service';
+import { TldrawService } from '../service';
 import { TldrawDeleteParams } from './tldraw.params';
 
 describe('TldrawController', () => {

apps/server/src/modules/tldraw/controller/tldraw.controller.ts

@@ -1,7 +1,7 @@
 import { ApiOperation, ApiResponse, ApiTags } from '@nestjs/swagger';
 import { Controller, Delete, ForbiddenException, HttpCode, NotFoundException, Param } from '@nestjs/common';
 import { ApiValidationError } from '@shared/common';
-import { TldrawService } from '../service/tldraw.service';
+import { TldrawService } from '../service';
 import { TldrawDeleteParams } from './tldraw.params';
 
 @ApiTags('Tldraw Document')

apps/server/src/modules/tldraw/controller/tldraw.ws.ts

@@ -28,17 +28,8 @@ export class TldrawWs implements OnGatewayInit, OnGatewayConnection {
 		}
 	}
 
-	public afterInit(): void {
-		this.tldrawWsService.setPersistence({
-			bindState: async (docName, ydoc) => {
-				await this.tldrawWsService.updateDocument(docName, ydoc);
-			},
-			writeState: async (docName) => {
-				// This is called when all connections to the document are closed.
-				// eslint-disable-next-line @typescript-eslint/no-unsafe-call,@typescript-eslint/no-unsafe-member-access
-				await this.tldrawWsService.flushDocument(docName);
-			},
-		});
+	public async afterInit(): Promise<void> {
+		await this.tldrawWsService.createDbIndex();
 	}
 
 	private getDocNameFromRequest(request: Request): string {

apps/server/src/modules/tldraw/domain/index.ts

@@ -0,0 +1 @@
+export * from './ws-shared-doc.do';

apps/server/src/modules/tldraw/domain/ws-shared-doc.do.spec.ts

@@ -2,17 +2,29 @@ import { Test } from '@nestjs/testing';
 import { INestApplication } from '@nestjs/common';
 import WebSocket from 'ws';
 import { WsAdapter } from '@nestjs/platform-ws';
-import { CoreModule } from '@src/core';
-import { ConfigModule } from '@nestjs/config';
-import { createConfigModuleOptions } from '@src/config';
-import { createMock } from '@golevelup/ts-jest';
 import * as AwarenessProtocol from 'y-protocols/awareness';
-import { config } from '../config';
-import { TldrawBoardRepo } from '../repo/tldraw-board.repo';
+import * as Yjs from 'yjs';
+import * as Ioredis from 'ioredis';
 import { TldrawWsService } from '../service';
 import { WsSharedDocDo } from './ws-shared-doc.do';
-import { TldrawWs } from '../controller';
 import { TestConnection } from '../testing/test-connection';
+import { TldrawWsTestModule } from '../tldraw-ws-test.module';
+
+jest.mock('y-protocols/awareness', () => {
+	const moduleMock: unknown = {
+		__esModule: true,
+		...jest.requireActual('y-protocols/awareness'),
+	};
+	return moduleMock;
+});
+
+jest.mock('yjs', () => {
+	const moduleMock: unknown = {
+		__esModule: true,
+		...jest.requireActual('yjs'),
+	};
+	return moduleMock;
+});
 
 describe('WsSharedDocDo', () => {
 	let app: INestApplication;
@@ -24,18 +36,14 @@ describe('WsSharedDocDo', () => {
 
 	jest.useFakeTimers();
 
+	const delay = (ms: number) =>
+		new Promise((resolve) => {
+			setTimeout(resolve, ms);
+		});
+
 	beforeAll(async () => {
-		const imports = [CoreModule, ConfigModule.forRoot(createConfigModuleOptions(config))];
 		const testingModule = await Test.createTestingModule({
-			imports,
-			providers: [
-				TldrawWs,
-				TldrawBoardRepo,
-				{
-					provide: TldrawWsService,
-					useValue: createMock<TldrawWsService>(),
-				},
-			],
+			imports: [TldrawWsTestModule],
 		}).compile();
 
 		service = testingModule.get<TldrawWsService>(TldrawWsService);
@@ -49,13 +57,112 @@ describe('WsSharedDocDo', () => {
 		await app.close();
 	});
 
+	describe('creating new instance', () => {
+		const setup = () => {
+			const doc = new WsSharedDocDo('TEST', service);
+			const redisOnSpy = jest.spyOn(Ioredis.Redis.prototype, 'on');
+			const loggerSpy = jest.spyOn(service, 'logYDocError');
+
+			return {
+				doc,
+				redisOnSpy,
+				loggerSpy,
+			};
+		};
+
+		describe('when subscribing to redis channel', () => {
+			it('should register new listener', async () => {
+				jest.spyOn(Ioredis.Redis.prototype, 'subscribe').mockResolvedValueOnce(() => 1);
+
+				const { doc, redisOnSpy } = setup();
+
+				await delay(20);
+				expect(doc).toBeDefined();
+				expect(redisOnSpy).toHaveBeenCalled();
+			});
+
+			it('should call logYDocError function in service on error', async () => {
+				jest.spyOn(Ioredis.Redis.prototype, 'subscribe').mockRejectedValueOnce(() => new Error('error'));
+
+				const { doc, loggerSpy } = setup();
+
+				await delay(20);
+				expect(doc).toBeDefined();
+				expect(loggerSpy).toHaveBeenCalled();
+			});
+		});
+	});
+
+	describe('destroy', () => {
+		const setup = () => {
+			const doc = new WsSharedDocDo('TEST', service);
+			doc.awarenessChannel = 'TEST-AWARENESS';
+			const loggerSpy = jest.spyOn(service, 'logYDocError');
+
+			return {
+				doc,
+				loggerSpy,
+			};
+		};
+
+		describe('when unsubscribing from redis channels', () => {
+			it('should call logYDocError function in service on error', async () => {
+				jest.spyOn(Ioredis.Redis.prototype, 'unsubscribe').mockRejectedValueOnce(() => new Error('error'));
+
+				const { doc, loggerSpy } = setup();
+				doc.destroy();
+
+				await delay(20);
+				expect(doc).toBeDefined();
+				expect(loggerSpy).toHaveBeenCalled();
+			});
+		});
+	});
+
+	describe('redis message handler', () => {
+		const setup = () => {
+			const applyUpdateSpy = jest.spyOn(Yjs, 'applyUpdate').mockReturnValueOnce();
+			const applyAwarenessUpdateSpy = jest.spyOn(AwarenessProtocol, 'applyAwarenessUpdate').mockReturnValueOnce();
+
+			const doc = new WsSharedDocDo('TEST', service);
+			doc.awarenessChannel = 'TEST-AWARENESS';
+
+			return {
+				doc,
+				applyUpdateSpy,
+				applyAwarenessUpdateSpy,
+			};
+		};
+
+		describe('when channel name is the same as docName', () => {
+			it('should call applyUpdate', () => {
+				const { doc, applyUpdateSpy } = setup();
+
+				doc.redisMessageHandler(Buffer.from('TEST'), Buffer.from('message'));
+
+				expect(applyUpdateSpy).toHaveBeenCalled();
+			});
+		});
+
+		describe('when channel name is the same as docAwarenessChannel name', () => {
+			it('should call applyAwarenessUpdate', () => {
+				const { doc, applyAwarenessUpdateSpy } = setup();
+
+				doc.redisMessageHandler(Buffer.from('TEST-AWARENESS'), Buffer.from('message'));
+
+				expect(applyAwarenessUpdateSpy).toHaveBeenCalled();
+			});
+		});
+	});
+
 	describe('ydoc client awareness change handler', () => {
 		const setup = async () => {
 			ws = await TestConnection.setupWs(wsUrl);
 
 			class MockAwareness {
 				on = jest.fn();
 			}
+
 			const doc = new WsSharedDocDo('TEST', service);
 			doc.awareness = new MockAwareness() as unknown as AwarenessProtocol.Awareness;
 			const awarenessMetaMock = new Map<number, { clock: number; lastUpdated: number }>();

apps/server/src/modules/tldraw/domain/ws-shared-doc.do.ts

@@ -1,9 +1,9 @@
-import { Doc } from 'yjs';
+import { applyUpdate, Doc } from 'yjs';
 import WebSocket from 'ws';
-import { Awareness, encodeAwarenessUpdate } from 'y-protocols/awareness';
+import { applyAwarenessUpdate, Awareness, encodeAwarenessUpdate } from 'y-protocols/awareness';
 import { encoding } from 'lib0';
 import { TldrawWsService } from '@modules/tldraw/service';
-import { WSMessageType } from '../types/connection-enum';
+import { WSMessageType } from '../types';
 
 export class WsSharedDocDo extends Doc {
 	public name: string;
@@ -12,6 +12,8 @@ export class WsSharedDocDo extends Doc {
 
 	public awareness: Awareness;
 
+	public awarenessChannel: string;
+
 	/**
 	 * @param {string} name
 	 * @param {TldrawWsService} tldrawService
@@ -23,11 +25,20 @@ export class WsSharedDocDo extends Doc {
 		this.conns = new Map();
 		this.awareness = new Awareness(this);
 		this.awareness.setLocalState(null);
+		this.awarenessChannel = `${name}-awareness`;
 
 		this.awareness.on('update', this.awarenessChangeHandler);
+
 		this.on('update', (update: Uint8Array, origin, doc: WsSharedDocDo) => {
 			this.tldrawService.updateHandler(update, origin, doc);
 		});
+
+		this.tldrawService.sub
+			.subscribe(this.name, this.awarenessChannel)
+			.then(() => this.tldrawService.sub.on('messageBuffer', this.redisMessageHandler))
+			.catch((err) =>
+				this.tldrawService.logYDocError(this.name, 'Error while subscribing to Redis channels', err as Error)
+			);
 	}
 
 	/**
@@ -43,6 +54,31 @@ export class WsSharedDocDo extends Doc {
 		this.sendAwarenessMessage(buff);
 	};
 
+	/**
+	 * @param {{ Buffer }} channel redis channel
+	 * @param {{ Buffer }} update update message
+	 */
+	public redisMessageHandler = (channel: Buffer, update: Buffer): void => {
+		const channelId = channel.toString();
+
+		if (channelId === this.name) {
+			applyUpdate(this, update, this.tldrawService.sub);
+		}
+
+		if (channelId === this.awarenessChannel) {
+			applyAwarenessUpdate(this.awareness, update, this.tldrawService.sub);
+		}
+	};
+
+	public destroy() {
+		super.destroy();
+		this.tldrawService.sub
+			.unsubscribe(this.name, this.awarenessChannel)
+			.catch((err) =>
+				this.tldrawService.logYDocError(this.name, 'Error while unsubscribing from Redis channels', err as Error)
+			);
+	}
+
 	/**
 	 * @param {{ added: Array<number>, updated: Array<number>, removed: Array<number> }} changes
 	 * @param {WebSocket | null} wsConnection Origin is the connection that made the change

apps/server/src/modules/tldraw/entities/tldraw-drawing.entity.spec.ts

@@ -12,16 +12,24 @@ describe('tldraw entity', () => {
 				const tldraw = new TldrawDrawing({
 					docName: 'test',
 					version: 'v1_tst',
-					value: 'bindatamock',
+					value: Buffer.from('bindatamock'),
 					_id: 'test-id',
 					clock: 0,
 					action: 'update',
+					part: 1,
 				});
 				expect(tldraw).toBeInstanceOf(TldrawDrawing);
 			});
 
 			it('should throw with empty docName', () => {
-				const call = () => new TldrawDrawing({ docName: '', version: 'v1_tst', value: 'bindatamock', _id: 'test-id' });
+				const call = () =>
+					new TldrawDrawing({
+						docName: '',
+						version: 'v1_tst',
+						value: Buffer.from('bindatamock'),
+						_id: 'test-id',
+						part: 1,
+					});
 				expect(call).toThrow();
 			});
 		});

apps/server/src/modules/tldraw/entities/tldraw-drawing.entity.ts

@@ -15,6 +15,9 @@ export class TldrawDrawing {
 		if (props.action) {
 			this.action = props.action;
 		}
+		if (typeof props.part === 'number') {
+			this.part = props.part;
+		}
 	}
 
 	@PrimaryKey()
@@ -27,13 +30,16 @@ export class TldrawDrawing {
 	version: string;
 
 	@Property({ nullable: false })
-	value: string;
+	value: Buffer;
 
 	@Property({ nullable: true })
 	clock?: number;
 
 	@Property({ nullable: true })
 	action?: string;
+
+	@Property({ nullable: true })
+	part?: number;
 }
 
 export interface TldrawDrawingProps {
@@ -42,5 +48,6 @@ export interface TldrawDrawingProps {
 	version: string;
 	clock?: number;
 	action?: string;
-	value: string;
+	value: Buffer;
+	part?: number;
 }

apps/server/src/modules/tldraw/factory/tldraw.factory.ts

@@ -7,8 +7,11 @@ export const tldrawEntityFactory = BaseFactory.define<TldrawDrawing, TldrawDrawi
 		return {
 			_id: 'test-id',
 			docName: 'test-name',
-			value: 'test-value',
-			version: `test-version-${sequence}`,
+			value: Buffer.from('test'),
+			version: `v1`,
+			action: 'update',
+			clock: sequence,
+			part: sequence,
 		};
 	}
 );