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

import { ConflictException, Injectable } from '@nestjs/common';
import { InjectRepository } from '@nestjs/typeorm';
import { PipelineTask } from './pipeline-task.entity';
import { In, Repository } from 'typeorm';
import { CreatePipelineTaskInput } from './dtos/create-pipeline-task.input';
import { RedisService } from 'nestjs-redis';
import { Pipeline } from '../pipelines/pipeline.entity';
import { InjectQueue } from '@nestjs/bull';
import { PIPELINE_TASK_QUEUE } from './pipeline-tasks.constants';
import { Queue } from 'bull';
import { LockFailedException } from '../commons/exceptions/lock-failed.exception';
import { PubSub } from 'apollo-server-express';
import { TaskStatuses } from './enums/task-statuses.enum';
import { isNil } from 'ramda';

@Injectable()
export class PipelineTasksService {
  pubSub = new PubSub();
  constructor(
    @InjectRepository(PipelineTask)
    private readonly repository: Repository<PipelineTask>,
    @InjectRepository(Pipeline)
    private readonly pipelineRepository: Repository<Pipeline>,
    @InjectQueue(PIPELINE_TASK_QUEUE)
    private readonly queue: Queue<PipelineTask>,
    private readonly redis: RedisService,
  ) {}
  async addTask(dto: CreatePipelineTaskInput) {
    const pipeline = await this.pipelineRepository.findOneOrFail({
      where: { id: dto.pipelineId },
      relations: ['project'],
    });
    // const hasUnfinishedTask = await this.repository
    //   .findOne({
    //     pipelineId: pipeline.id,
    //     status: In([TaskStatuses.pending, TaskStatuses.working]),
    //   })
    //   .then((val) => !isNil(val));
    // if (hasUnfinishedTask) {
    //   throw new ConflictException(
    //     'there are still unfinished task in the current pipeline.',
    //   );
    // }
    const task = await this.repository.save(this.repository.create(dto));
    task.pipeline = pipeline;

    const tasksKey = this.getRedisTokens(pipeline)[1];
    const redis = this.redis.getClient();
    await redis.lpush(tasksKey, JSON.stringify(task));
    await this.doNextTask(pipeline);
    return task;
  }

  async findTaskById(id: string) {
    return await this.repository.findOneOrFail({ id });
  }

  async listTasksByPipelineId(pipelineId: string) {
    return await this.repository.find({ pipelineId });
  }

  async doNextTask(pipeline: Pipeline) {
    const [lckKey, tasksKey] = this.getRedisTokens(pipeline);
    const redis = this.redis.getClient();

    const unLck = await new Promise<() => Promise<void>>(
      async (resolve, reject) => {
        const lckValue = Date.now().toString();
        for (let i = 0; i < 5; i++) {
          if (
            await redis
              .set(lckKey, 0, 'EX', lckValue, 'NX')
              .then(() => true)
              .catch(() => false)
          ) {
            resolve(async () => {
              if ((await redis.get(lckKey)) === lckValue) {
                await redis.del(lckKey);
              }
            });
            return;
          }
          await new Promise((resolve) => setTimeout(resolve, 2000));
        }
        reject(new LockFailedException(lckKey));
      },
    );

    const task = JSON.parse(
      (await redis.rpop(tasksKey).finally(() => unLck())) ?? 'null',
    );
    if (task) {
      await this.queue.add(task);
    }
  }

  async updateTask(task: PipelineTask) {
    this.pubSub.publish(task.id, task);
    return await this.repository.save(task);
  }

  async watchTaskUpdated(id: string) {
    return this.pubSub.asyncIterator(id);
  }

  getRedisTokens(pipeline: Pipeline): [string, string] {
    return [`pipeline-${pipeline.id}:lck`, `pipeline-${pipeline.id}:tasks`];
  }
}