BC-8403 Code improvements

.env.test

@@ -10,6 +10,7 @@ S3_SECRET_KEY=miniouser
 S3_SSL=false
 
 FEATURE_TLDRAW_ENABLED=true
-TLDRAW_WEBSOCKET_URL=ws://localhost:3345
+TLDRAW_WEBSOCKET_URL=ws://localhost:3399
+TLDRAW_WEBSOCKET_PORT=3399
 
 X_API_ALLOWED_KEYS=randomString

src/apps/tldraw-server.app.ts

@@ -29,13 +29,13 @@ async function bootstrap(): Promise<void> {
 	await metricsApp.listen(metricsPort, async () => {
 		const logger = await metricsApp.resolve(Logger);
 		logger.setContext('METRICS');
-		logger.log(`Metrics server is running on port ${metricsPort}`);
+		logger.info(`Metrics server is running on port ${metricsPort}`);
 	});
 
 	await nestApp.listen(httpPort, async () => {
 		const logger = await nestApp.resolve(Logger);
 		logger.setContext('TLDRAW');
-		logger.log(`Server is running on port ${httpPort}`);
+		logger.info(`Server is running on port ${httpPort}`);
 	});
 }
 bootstrap();

src/apps/tldraw-worker.app.ts

@@ -1,9 +1,19 @@
 import { NestFactory } from '@nestjs/core';
 import { WorkerModule } from '../modules/worker/worker.module.js';
+import { WorkerService } from '../modules/worker/worker.service.js';
 
 async function bootstrap(): Promise<void> {
 	const nestApp = await NestFactory.createApplicationContext(WorkerModule);
 
 	await nestApp.init();
+	const workerService = await nestApp.resolve(WorkerService);
+
+	try {
+		workerService.start();
+	} catch (error) {
+		console.error(error);
+		workerService.stop();
+		process.exit(1);
+	}
 }
 bootstrap();

src/infra/authorization/authorization.service.ts

@@ -94,7 +94,6 @@ export class AuthorizationService {
 
 	private createErrorResponsePayload(code: number, reason: string): ResponsePayload {
 		const response = ResponsePayloadBuilder.buildWithError(code, reason);
-		this.logger.log(`Error: ${code} - ${reason}`);
 
 		return response;
 	}

src/infra/logger/logger.spec.ts

@@ -6,7 +6,7 @@ import { RequestLoggingBody } from './interfaces/logger.interface.js';
 import { Logger } from './logger.js';
 
 describe('Logger', () => {
-	let service: Logger;
+	let logger: Logger;
 	let processStdoutWriteSpy: jest.SpyInstance<
 		boolean,
 		[str: string | Uint8Array, encoding?: BufferEncoding | undefined, cb?: ((err?: Error) => void) | undefined],
@@ -30,7 +30,7 @@ describe('Logger', () => {
 			],
 		}).compile();
 
-		service = await module.resolve(Logger);
+		logger = await module.resolve(Logger);
 		winstonLogger = module.get(WINSTON_MODULE_PROVIDER);
 	});
 
@@ -44,26 +44,26 @@ describe('Logger', () => {
 		processStderrWriteSpy.mockRestore();
 	});
 
-	describe('WHEN log logging', () => {
+	describe('WHEN info logging', () => {
 		it('should call winstonLogger.info', () => {
 			const error = new Error('custom error');
-			service.log(error.message, error.stack);
+			logger.info(error.message, error.stack);
 			expect(winstonLogger.info).toHaveBeenCalled();
 		});
 	});
 
 	describe('WHEN warn logging', () => {
 		it('should call winstonLogger.warning', () => {
 			const error = new Error('custom error');
-			service.warn(error.message, error.stack);
+			logger.warning(error.message, error.stack);
 			expect(winstonLogger.warning).toHaveBeenCalled();
 		});
 	});
 
 	describe('WHEN debug logging', () => {
 		it('should call winstonLogger.debug', () => {
 			const error = new Error('custom error');
-			service.debug(error.message, error.stack);
+			logger.debug(error.message, error.stack);
 			expect(winstonLogger.debug).toHaveBeenCalled();
 		});
 	});
@@ -81,7 +81,7 @@ describe('Logger', () => {
 				},
 				error,
 			};
-			service.http(message, error.stack);
+			logger.http(message, error.stack);
 			expect(winstonLogger.notice).toHaveBeenCalled();
 		});
 	});

src/infra/logger/logger.ts

@@ -16,11 +16,11 @@ export class Logger {
 
 	public constructor(@Inject(WINSTON_MODULE_PROVIDER) private readonly logger: winston.Logger) {}
 
-	public log(message: unknown, context?: string): void {
+	public info(message: unknown, context?: string): void {
 		this.logger.info(this.createMessage(message, context));
 	}
 
-	public warn(message: unknown, context?: string): void {
+	public warning(message: unknown, context?: string): void {
 		this.logger.warning(this.createMessage(message, context));
 	}
 

src/infra/metrics/index.ts

@@ -1,2 +1,2 @@
-export * from './metrics.module.js';
-export * from './metrics.service.js';
+export { MetricsModule } from './metrics.module.js';
+export { MetricsService } from './metrics.service.js';

src/infra/redis/index.ts

@@ -1,2 +1,2 @@
+export * from './redis.factory.js';
 export * from './redis.module.js';
-export * from './redis.service.js';

src/infra/redis/interfaces/redis-adapter.ts

@@ -11,11 +11,11 @@ export interface RedisAdapter {
 	exists(stream: string): Promise<number>;
 	createGroup(): Promise<void>;
 	quit(): Promise<void>;
-	readStreams(streams: StreamNameClockPair[]): Promise<StreamMessagesReply>;
-	readMessagesFromStream(streamName: string): Promise<StreamMessagesReply>;
+	readStreams(streams: StreamNameClockPair[]): Promise<StreamMessagesReply[]>;
+	readMessagesFromStream(streamName: string): Promise<StreamMessagesReply[]>;
 	reclaimTasks(consumerName: string, redisTaskDebounce: number, tryClaimCount: number): Promise<XAutoClaimResponse>;
 	getDeletedDocEntries(): Promise<StreamMessageReply[]>;
-	deleteDeleteDocEntry(id: string): Promise<number>;
+	deleteDeletedDocEntry(id: string): Promise<number>;
 	tryClearTask(task: Task): Promise<number>;
 	tryDeduplicateTask(task: Task, lastId: number, redisMinMessageLifetime: number): Promise<void>;
 }

src/infra/redis/interfaces/stream-message-reply.ts

@@ -1,23 +1,17 @@
 import { RedisKey } from 'ioredis';
 
 interface Message {
-	key: RedisKey;
-	m?: RedisKey;
+	m?: Buffer;
 	docName?: string;
-	compact?: string;
+	compact?: Buffer;
 }
 
 export interface StreamMessageReply {
 	id: RedisKey;
-	message: Record<keyof Message, RedisKey>;
+	message: Message;
 }
 
-export interface StreamMessagesSingleReply {
+export interface StreamMessagesReply {
 	name: string;
 	messages: StreamMessageReply[] | null;
 }
-
-export type StreamMessagesReply = {
-	name: string;
-	messages: StreamMessageReply[] | null;
-}[];

src/infra/redis/ioredis.adapter.spec.ts

@@ -194,7 +194,7 @@ describe(IoRedisAdapter.name, () => {
 
 			await redisAdapter.createGroup();
 
-			expect(logger.log).toHaveBeenCalledWith(error);
+			expect(logger.info).toHaveBeenCalledWith(error);
 		});
 	});
 
@@ -412,15 +412,15 @@ describe(IoRedisAdapter.name, () => {
 		it('should call redis xdel with correct values', async () => {
 			const { id, xdelSpy, expectedProps, redisAdapter } = await setup();
 
-			await redisAdapter.deleteDeleteDocEntry(id);
+			await redisAdapter.deleteDeletedDocEntry(id);
 
 			expect(xdelSpy).toHaveBeenCalledWith(...expectedProps);
 		});
 
 		it('should return correct value', async () => {
 			const { id, redisAdapter } = await setup();
 
-			const result = await redisAdapter.deleteDeleteDocEntry(id);
+			const result = await redisAdapter.deleteDeletedDocEntry(id);
 
 			expect(result).toBe(1);
 		});

src/infra/redis/ioredis.adapter.ts

@@ -76,7 +76,7 @@ export class IoRedisAdapter implements RedisAdapter {
 		try {
 			await this.redis.xgroup('CREATE', this.redisWorkerStreamName, this.redisWorkerGroupName, '0', 'MKSTREAM');
 		} catch (e) {
-			this.logger.log(e);
+			this.logger.info(e);
 			// It is okay when the group already exists, so we can ignore this error.
 			if (e.message !== 'BUSYGROUP Consumer Group name already exists') {
 				throw e;
@@ -88,7 +88,7 @@ export class IoRedisAdapter implements RedisAdapter {
 		await this.redis.quit();
 	}
 
-	public async readStreams(streams: StreamNameClockPair[]): Promise<StreamMessagesReply> {
+	public async readStreams(streams: StreamNameClockPair[]): Promise<StreamMessagesReply[]> {
 		const reads = await this.redis.xreadBuffer(
 			'COUNT',
 			1000,
@@ -104,7 +104,7 @@ export class IoRedisAdapter implements RedisAdapter {
 		return streamReplyRes;
 	}
 
-	public async readMessagesFromStream(streamName: string): Promise<StreamMessagesReply> {
+	public async readMessagesFromStream(streamName: string): Promise<StreamMessagesReply[]> {
 		const reads = await this.redis.xreadBuffer('STREAMS', streamName, '0');
 
 		const streamReplyRes = mapToStreamMessagesReply(reads);
@@ -140,7 +140,7 @@ export class IoRedisAdapter implements RedisAdapter {
 		return transformedDeletedTasks;
 	}
 
-	public deleteDeleteDocEntry(id: string): Promise<number> {
+	public deleteDeletedDocEntry(id: string): Promise<number> {
 		const result = this.redis.xdel(this.redisDeleteStreamName, id);
 
 		return result;

src/infra/redis/mapper.ts

@@ -42,7 +42,7 @@ export function mapToStreamMessagesReplies(messages: XItems | unknown): StreamMe
 	return result;
 }
 
-export function mapToStreamMessagesReply(streamReply: XReadBufferReply | unknown): StreamMessagesReply {
+export function mapToStreamMessagesReply(streamReply: XReadBufferReply | unknown): StreamMessagesReply[] {
 	if (streamReply === null) {
 		return [];
 	}

src/infra/redis/redis.service.spec.ts → src/infra/redis/redis.factory.spec.ts

@@ -4,7 +4,7 @@ import * as util from 'util';
 import { Logger } from '../logger/index.js';
 import { IoRedisAdapter } from './ioredis.adapter.js';
 import { RedisConfig } from './redis.config.js';
-import { RedisService } from './redis.service.js';
+import { RedisFactory } from './redis.factory.js';
 
 jest.mock('ioredis', () => {
 	return {
@@ -14,7 +14,7 @@ jest.mock('ioredis', () => {
 
 jest.mock<IoRedisAdapter>('./ioredis.adapter.js');
 
-describe('Redis Service', () => {
+describe(RedisFactory.name, () => {
 	beforeEach(() => {
 		jest.resetAllMocks();
 	});
@@ -50,7 +50,7 @@ describe('Redis Service', () => {
 				const constructorSpy = jest.spyOn(Redis.prototype, 'constructor');
 
 				const logger = createMock<Logger>();
-				const service = new RedisService(config, logger);
+				const factory = new RedisFactory(config, logger);
 
 				const expectedProps = {
 					sentinels: [
@@ -62,29 +62,29 @@ describe('Redis Service', () => {
 					name: 'sentinelName',
 				};
 
-				return { resolveSrv, sentinelServiceName, service, constructorSpy, expectedProps };
+				return { resolveSrv, sentinelServiceName, factory, constructorSpy, expectedProps };
 			};
 
 			it('calls resolveSrv', async () => {
-				const { resolveSrv, sentinelServiceName, service } = setup();
+				const { resolveSrv, sentinelServiceName, factory } = setup();
 
-				await service.createRedisInstance();
+				await factory.createRedisInstance();
 
 				expect(resolveSrv).toHaveBeenLastCalledWith(sentinelServiceName);
 			});
 
 			it('create new Redis instance with correctly props', async () => {
-				const { service, constructorSpy, expectedProps } = setup();
+				const { factory, constructorSpy, expectedProps } = setup();
 
-				await service.createRedisInstance();
+				await factory.createRedisInstance();
 
 				expect(constructorSpy).toHaveBeenCalledWith(expectedProps);
 			});
 
 			it('creates a new Redis instance', async () => {
-				const { service } = setup();
+				const { factory } = setup();
 
-				const redisInstance = await service.createRedisInstance();
+				const redisInstance = await factory.createRedisInstance();
 
 				expect(redisInstance).toBeInstanceOf(IoRedisAdapter);
 			});
@@ -105,33 +105,33 @@ describe('Redis Service', () => {
 				const constructorSpy = jest.spyOn(Redis.prototype, 'constructor');
 
 				const logger = createMock<Logger>();
-				const service = new RedisService(config, logger);
+				const factory = new RedisFactory(config, logger);
 
 				const expectedProps = redisUrl;
 
-				return { resolveSrv, service, redisMock, constructorSpy, expectedProps };
+				return { resolveSrv, factory, redisMock, constructorSpy, expectedProps };
 			};
 
 			it('calls resolveSrv', async () => {
-				const { resolveSrv, service } = setup();
+				const { resolveSrv, factory } = setup();
 
-				await service.createRedisInstance();
+				await factory.createRedisInstance();
 
 				expect(resolveSrv).not.toHaveBeenCalled();
 			});
 
 			it('create new Redis instance with correctly props', async () => {
-				const { service, constructorSpy, expectedProps } = setup();
+				const { factory, constructorSpy, expectedProps } = setup();
 
-				await service.createRedisInstance();
+				await factory.createRedisInstance();
 
 				expect(constructorSpy).toHaveBeenCalledWith(expectedProps);
 			});
 
 			it('creates a new Redis instance', async () => {
-				const { service } = setup();
+				const { factory } = setup();
 
-				const redisInstance = await service.createRedisInstance();
+				const redisInstance = await factory.createRedisInstance();
 
 				expect(redisInstance).toBeInstanceOf(IoRedisAdapter);
 			});

src/infra/redis/redis.service.ts → src/infra/redis/redis.factory.ts

@@ -1,4 +1,3 @@
-import { Injectable } from '@nestjs/common';
 import * as dns from 'dns';
 import { Redis } from 'ioredis';
 import * as util from 'util';
@@ -7,14 +6,11 @@ import { RedisAdapter } from './interfaces/index.js';
 import { IoRedisAdapter } from './ioredis.adapter.js';
 import { RedisConfig } from './redis.config.js';
 
-@Injectable()
-export class RedisService {
+export class RedisFactory {
 	public constructor(
 		private readonly config: RedisConfig,
 		private readonly logger: Logger,
-	) {
-		this.logger.setContext(RedisService.name);
-	}
+	) {}
 
 	public async createRedisInstance(): Promise<RedisAdapter> {
 		let redisInstance: Redis;
@@ -39,7 +35,7 @@ export class RedisService {
 		const sentinelName = this.config.REDIS_SENTINEL_NAME;
 		const sentinelPassword = this.config.REDIS_SENTINEL_PASSWORD;
 		const sentinels = await this.discoverSentinelHosts();
-		this.logger.log(`Discovered sentinels: ${JSON.stringify(sentinels)}`);
+		this.logger.info(`Discovered sentinels: ${JSON.stringify(sentinels)}`);
 
 		const redisInstance = new Redis({
 			sentinels,
@@ -63,7 +59,7 @@ export class RedisService {
 
 			return hosts;
 		} catch (err) {
-			this.logger.log('Error during service discovery:', err);
+			this.logger.info('Error during service discovery:', err);
 			throw err;
 		}
 	}