fennec-be/src/pipeline-tasks/pipeline-task-logs.service.ts

import { Injectable } from '@nestjs/common';
import { observableToAsyncIterable } from 'graphql-tools';
import { RedisService } from 'nestjs-redis';
import { find, omit, propEq } from 'ramda';
import { InjectPubSub } from '../commons/pub-sub/decorators/inject-pub-sub.decorator';
import { PubSub } from '../commons/pub-sub/pub-sub';
import { PipelineUnits } from './enums/pipeline-units.enum';
import { TaskStatuses } from './enums/task-statuses.enum';
import { PipelineTaskLogMessage } from './models/pipeline-task-log-message.module';
import { PipelineTaskLogs } from './models/pipeline-task-logs.model';
import { PipelineTask } from './pipeline-task.entity';

const LOG_TIMEOUT_SECONDS = 10_000;

@Injectable()
export class PipelineTaskLogsService {
  constructor(
    private readonly redisService: RedisService,
    @InjectPubSub()
    private readonly pubSub: PubSub,
  ) {}

  get redis() {
    return this.redisService.getClient();
  }

  getKeys(task: PipelineTask) {
    return `ptl:${task.id}`;
  }

  async recordLog(log: PipelineTaskLogMessage) {
    const logDto = omit(['task'], log);
    await Promise.all([
      this.pubSub.publish(this.getKeys(log.task), logDto),
      this.redis
        .expire(this.getKeys(log.task), LOG_TIMEOUT_SECONDS)
        .then(() =>
          this.redis.rpush(this.getKeys(log.task), JSON.stringify(logDto)),
        ),
    ]);
  }

  async readLog(task: PipelineTask): Promise<PipelineTaskLogMessage[]> {
    return await this.redis.lrange(this.getKeys(task), 0, -1).then((items) =>
      items.map((item) => {
        const log = JSON.parse(item) as PipelineTaskLogMessage;
        log.task = task;
        log.time = new Date(log.time);
        return log;
      }),
    );
  }

  async readLogsAsPipelineTaskLogs(
    task: PipelineTask,
  ): Promise<PipelineTaskLogs[]> {
    const logs = await this.readLog(task);
    const taskLogs: PipelineTaskLogs[] = [];
    for (const log of logs) {
      const taskLog = find<PipelineTaskLogs>(
        propEq('unit', log.unit),
        taskLogs,
      );
      if (!taskLog) {
        taskLogs.push({
          unit: (log.unit as unknown) as PipelineUnits,
          status: TaskStatuses.working,
          startedAt: log.time,
          logs: log.message,
        });
      } else {
        taskLog.logs += log.message;
      }
    }
    return taskLogs;
  }

  watchLogs(task: PipelineTask) {
    return observableToAsyncIterable(this.pubSub.message$(this.getKeys(task)));
  }
}