feat(pipeline-task): flush service.

src/pipeline-tasks/pipeline-task-flush.service.ts

@@ -0,0 +1,57 @@
+import { AmqpConnection, RabbitSubscribe } from '@golevelup/nestjs-rabbitmq';
+import { Injectable } from '@nestjs/common';
+import { deserialize } from 'class-transformer';
+import { RedisService } from 'nestjs-redis';
+import { isNil } from 'ramda';
+import { getSelfInstanceQueueKey } from '../commons/utils/rabbit-mq';
+import { terminalTaskStatuses } from './enums/task-statuses.enum';
+import { PipelineTaskEvent } from './models/pipeline-task-event';
+import {
+  EXCHANGE_PIPELINE_TASK_TOPIC,
+  ROUTE_PIPELINE_TASK_DONE,
+} from './pipeline-tasks.constants';
+import {
+  EXCHANGE_PIPELINE_TASK_FANOUT,
+  ROUTE_PIPELINE_TASK_LOG,
+  QUEUE_WRITE_PIPELINE_TASK_LOG,
+} from './pipeline-tasks.constants';
+
+@Injectable()
+export class PipelineTaskFlushService {
+  constructor(
+    private readonly redisService: RedisService,
+    private readonly amqpConnection: AmqpConnection,
+  ) {}
+
+  @RabbitSubscribe({
+    exchange: EXCHANGE_PIPELINE_TASK_FANOUT,
+    routingKey: ROUTE_PIPELINE_TASK_LOG,
+    queue: getSelfInstanceQueueKey(QUEUE_WRITE_PIPELINE_TASK_LOG),
+    queueOptions: {
+      autoDelete: true,
+    },
+  })
+  async write(message: PipelineTaskEvent) {
+    await this.redisService
+      .getClient()
+      .rpush(this.getKey(message.taskId), JSON.stringify(message));
+    if (isNil(message.unit) && terminalTaskStatuses.includes(message.status)) {
+      this.amqpConnection.request({
+        exchange: EXCHANGE_PIPELINE_TASK_TOPIC,
+        routingKey: ROUTE_PIPELINE_TASK_DONE,
+        payload: { taskId: message.taskId, status: message.status },
+      });
+    }
+  }
+
+  async read(taskId: string) {
+    const raw = await this.redisService
+      .getClient()
+      .lrange(this.getKey(taskId), 0, -1);
+    return raw.map((it) => deserialize(PipelineTaskEvent, it));
+  }
+
+  private getKey(taskId: string) {
+    return `p-task:log:${taskId}`;
+  }
+}