feat: finish viewer service

apps/twitch-viewer-service/src/app/app.module.ts

@@ -3,13 +3,15 @@ import { ConfigModule } from '@nestjs/config';
 import { VersionModule } from './version/version.module';
 import { HealthModule } from './health/health.module';
 import { MetricsModule } from './metrics/metrics.module';
+import { ViewerModule } from './viewer/viewer.module';
 
 @Module({
 	imports: [
 		ConfigModule.forRoot({ isGlobal: true }),
 		HealthModule,
 		MetricsModule,
 		VersionModule,
+		ViewerModule,
 	],
 })
 export class AppModule {}

apps/twitch-viewer-service/src/app/viewer/viewer.controller.ts

@@ -0,0 +1,144 @@
+import { Controller } from '@nestjs/common';
+import { ViewerService } from './viewer.service';
+import { MessagePattern, Payload, RpcException } from '@nestjs/microservices';
+import { TwitchViewerTopics } from '@toxictoast/azkaban-broker-rabbitmq';
+import { Nullable } from '@toxictoast/azkaban-base-types';
+import { ViewerDAO } from '@azkaban/twitch-infrastructure';
+
+@Controller('viewer')
+export class ViewerController {
+	constructor(private readonly service: ViewerService) {}
+
+	@MessagePattern(TwitchViewerTopics.LIST)
+	async getViewerList(
+		@Payload('limit') limit: number,
+		@Payload('offset') offset: number,
+	) {
+		try {
+			return await this.service.getList(limit, offset);
+		} catch (error) {
+			throw new RpcException(error);
+		}
+	}
+
+	@MessagePattern(TwitchViewerTopics.ID)
+	async getViewerById(@Payload('id') id: string) {
+		try {
+			return await this.service.getById(id);
+		} catch (error) {
+			throw new RpcException(error);
+		}
+	}
+
+	@MessagePattern(TwitchViewerTopics.JOIN)
+	async updateViewerJoin(@Payload('display_name') display_name: string) {
+		try {
+			const viewer = await this.getViewerByDisplayName(display_name);
+			if (!viewer) {
+				return await this.createViewerByDisplayName(display_name);
+			}
+			const newJoins = viewer.joins + 1;
+			return await this.service.updateViewer(viewer.id, newJoins);
+		} catch (error) {
+			throw new RpcException(error);
+		}
+	}
+
+	@MessagePattern(TwitchViewerTopics.PART)
+	async updateViewerPart(@Payload('display_name') display_name: string) {
+		try {
+			const viewer = await this.getViewerByDisplayName(display_name);
+			if (!viewer) {
+				return await this.createViewerByDisplayName(display_name);
+			}
+			const newParts = viewer.parts + 1;
+			return await this.service.updateViewer(
+				viewer.id,
+				undefined,
+				newParts,
+			);
+		} catch (error) {
+			throw new RpcException(error);
+		}
+	}
+
+	@MessagePattern(TwitchViewerTopics.TIMEOUT)
+	async updateViewerTimeout(@Payload('display_name') display_name: string) {
+		try {
+			const viewer = await this.getViewerByDisplayName(display_name);
+			if (!viewer) {
+				return await this.createViewerByDisplayName(display_name);
+			}
+			const newTimeouts = viewer.timeouts + 1;
+			return await this.service.updateViewer(
+				viewer.id,
+				undefined,
+				undefined,
+				undefined,
+				newTimeouts,
+			);
+		} catch (error) {
+			throw new RpcException(error);
+		}
+	}
+
+	@MessagePattern(TwitchViewerTopics.BAN)
+	async updateViewerBan(@Payload('display_name') display_name: string) {
+		try {
+			const viewer = await this.getViewerByDisplayName(display_name);
+			if (!viewer) {
+				return await this.createViewerByDisplayName(display_name);
+			}
+			const newBans = viewer.bans + 1;
+			return await this.service.updateViewer(
+				viewer.id,
+				undefined,
+				undefined,
+				undefined,
+				undefined,
+				newBans,
+			);
+		} catch (error) {
+			throw new RpcException(error);
+		}
+	}
+
+	@MessagePattern(TwitchViewerTopics.DELETE)
+	async deleteViewer(@Payload('id') id: string) {
+		try {
+			return await this.service.deleteViewer(id);
+		} catch (error) {
+			throw new RpcException(error);
+		}
+	}
+
+	@MessagePattern(TwitchViewerTopics.RESTORE)
+	async restoreViewer(@Payload('id') id: string) {
+		try {
+			return await this.service.restoreViewer(id);
+		} catch (error) {
+			throw new RpcException(error);
+		}
+	}
+
+	private async getViewerByDisplayName(
+		display_name: string,
+	): Promise<Nullable<ViewerDAO>> {
+		try {
+			const viewer = await this.service.getByDisplayName(display_name);
+			return viewer ?? null;
+		} catch (error) {
+			throw new RpcException(error);
+		}
+	}
+
+	private async createViewerByDisplayName(
+		display_name: string,
+	): Promise<ViewerDAO> {
+		try {
+			return await this.service.createViewer(display_name);
+		} catch (error) {
+			throw new RpcException(error);
+		}
+	}
+}

apps/twitch-viewer-service/src/app/viewer/viewer.module.ts

@@ -0,0 +1,18 @@
+import { Module } from '@nestjs/common';
+import {
+	viewerProvider,
+	datasourceProvider,
+	ViewerEntity,
+} from '@azkaban/twitch-infrastructure';
+import { ViewerService } from './viewer.service';
+import { ViewerController } from './viewer.controller';
+
+@Module({
+	controllers: [ViewerController],
+	providers: [
+		...datasourceProvider([ViewerEntity]),
+		...viewerProvider,
+		ViewerService,
+	],
+})
+export class ViewerModule {}

apps/twitch-viewer-service/src/app/viewer/viewer.service.ts

@@ -0,0 +1,92 @@
+import { Inject, Injectable } from '@nestjs/common';
+import {
+	ViewerDAO,
+	ViewerEntity,
+	ViewerRepository,
+	ViewerService as BaseService,
+} from '@azkaban/twitch-infrastructure';
+import { Repository } from 'typeorm';
+import { Nullable, Optional } from '@toxictoast/azkaban-base-types';
+
+@Injectable()
+export class ViewerService {
+	private readonly infrastructureRepository: ViewerRepository;
+	private readonly infrastructureService: BaseService;
+
+	constructor(
+		@Inject('VIEWER_REPOSITORY')
+		private readonly viewerRepository: Repository<ViewerEntity>,
+	) {
+		this.infrastructureRepository = new ViewerRepository(
+			this.viewerRepository,
+		);
+		this.infrastructureService = new BaseService(
+			this.infrastructureRepository,
+		);
+	}
+
+	async getList(limit: number, offset: number): Promise<Array<ViewerDAO>> {
+		return await this.infrastructureService.getViewerList(limit, offset);
+	}
+
+	async getById(id: string): Promise<ViewerDAO> {
+		return await this.infrastructureService.getViewerById(id);
+	}
+
+	async getByDisplayName(display_name: string): Promise<ViewerDAO> {
+		return await this.infrastructureService.getViewerByDisplayName(
+			display_name,
+		);
+	}
+
+	async createViewer(display_name: string): Promise<ViewerDAO> {
+		return await this.infrastructureService.createViewer({
+			display_name,
+		});
+	}
+
+	async updateViewer(
+		id: string,
+		joins?: Optional<number>,
+		parts?: Optional<number>,
+		messages?: Optional<number>,
+		timeouts?: Optional<number>,
+		bans?: Optional<number>,
+		minutes_watched?: Optional<number>,
+		lastseen_at?: Optional<Nullable<Date>>,
+	): Promise<ViewerDAO> {
+		if (joins !== undefined) {
+			await this.infrastructureService.updateJoins(id, joins);
+		}
+		if (parts !== undefined) {
+			await this.infrastructureService.updateParts(id, parts);
+		}
+		if (messages !== undefined) {
+			await this.infrastructureService.updateMessages(id, messages);
+		}
+		if (timeouts !== undefined) {
+			await this.infrastructureService.updateTimeouts(id, timeouts);
+		}
+		if (bans !== undefined) {
+			await this.infrastructureService.updateBans(id, bans);
+		}
+		if (minutes_watched !== undefined) {
+			await this.infrastructureService.updateMinutesWatched(
+				id,
+				minutes_watched,
+			);
+		}
+		if (lastseen_at !== undefined) {
+			await this.infrastructureService.updateLastSeenAt(id, lastseen_at);
+		}
+		return await this.infrastructureService.getViewerById(id);
+	}
+
+	async deleteViewer(id: string): Promise<ViewerDAO> {
+		return await this.infrastructureService.deleteViewer(id);
+	}
+
+	async restoreViewer(id: string): Promise<ViewerDAO> {
+		return await this.infrastructureService.restoreViewer(id);
+	}
+}

libs/twitch-infrastructure/src/lib/typeorm/services/viewer.service.ts

@@ -136,4 +136,24 @@ export class ViewerService {
 			throw new BadRequestException(errorMessage);
 		}
 	}
+
+	async deleteViewer(id: string): Promise<ViewerDAO> {
+		const result = await this.domainService.deleteViewer(id);
+		if (result.isSuccess) {
+			return result.value;
+		} else {
+			const errorMessage = result.errorValue;
+			throw new NotFoundException(errorMessage);
+		}
+	}
+
+	async restoreViewer(id: string): Promise<ViewerDAO> {
+		const result = await this.domainService.restoreViewer(id);
+		if (result.isSuccess) {
+			return result.value;
+		} else {
+			const errorMessage = result.errorValue;
+			throw new NotFoundException(errorMessage);
+		}
+	}
 }