feat(pipeline-tasks): 任务更新推送。

替代了先查询再订阅，确保数据不丢失。

.vscode/settings.json

@@ -2,6 +2,8 @@
   "cSpell.words": [
     "Repos",
     "lpush",
-    "rpop"
+    "lrange",
+    "rpop",
+    "rpush"
   ]
 }

package.json

@@ -40,6 +40,7 @@
     "graphql-tools": "^7.0.2",
     "js-yaml": "^4.0.0",
     "nestjs-redis": "^1.2.8",
+    "observable-to-async-generator": "^1.0.1-rc",
     "pg": "^8.5.1",
     "ramda": "^0.27.1",
     "reflect-metadata": "^0.1.13",

src/app.module.ts

@@ -37,6 +37,7 @@ import { RedisModule } from 'nestjs-redis';
         debug: configService.get<string>('env') !== 'prod',
         playground: true,
         autoSchemaFile: true,
+        installSubscriptionHandlers: true,
       }),
       inject: [ConfigService],
     }),

src/pipeline-tasks/dtos/create-pipeline-task.input.ts

@@ -7,6 +7,5 @@ export class CreatePipelineTaskInput {
 
   commit: string;
 
-  @Field(() => PipelineUnits)
   units: PipelineUnits[];
 }

src/pipeline-tasks/dtos/pipeline-task-log.args.ts

@@ -0,0 +1,8 @@
+import { ArgsType } from '@nestjs/graphql';
+import { IsUUID } from 'class-validator';
+
+@ArgsType()
+export class PipelineTaskLogArgs {
+  @IsUUID()
+  taskId: string;
+}

src/pipeline-tasks/enums/task-statuses.enum.ts

@@ -1,6 +1,13 @@
+import { registerEnumType } from '@nestjs/graphql';
+
 export enum TaskStatuses {
   success = 'success',
   failed = 'failed',
   working = 'working',
   pending = 'pending',
 }
+
+registerEnumType(TaskStatuses, {
+  name: 'TaskStatuses',
+  description: '任务状态',
+});

src/pipeline-tasks/models/pipeline-task-log-message.module.ts

@@ -1,15 +1,32 @@
 import { PipelineTask } from './../pipeline-task.entity';
+import { PipelineUnits } from '../enums/pipeline-units.enum';
+import { Field, HideField, ObjectType } from '@nestjs/graphql';
 
+@ObjectType()
 export class PipelineTaskLogMessage {
+  @HideField()
   task: PipelineTask;
+  @Field(() => PipelineUnits, { nullable: true })
+  unit?: PipelineUnits;
+  @Field()
   time: Date;
+  @Field()
   message: string;
+  @Field()
+  isError: boolean;
 
-  static create(task: PipelineTask, message: string) {
+  static create(
+    task: PipelineTask,
+    unit: PipelineUnits,
+    message: string,
+    isError: boolean,
+  ) {
     return Object.assign(new PipelineTaskLogMessage(), {
       task,
       message,
       time: new Date(),
+      unit,
+      isError,
     });
   }
 }

src/pipeline-tasks/models/pipeline-task-logs.model.ts

@@ -1,7 +1,12 @@
 import { TaskStatuses } from '../enums/task-statuses.enum';
 import { PipelineUnits } from '../enums/pipeline-units.enum';
+import { Field, ObjectType } from '@nestjs/graphql';
+
+@ObjectType()
 export class PipelineTaskLogs {
+  @Field(() => PipelineUnits)
   unit: PipelineUnits;
+  @Field(() => TaskStatuses)
   status: TaskStatuses;
   startedAt?: Date;
   endedAt?: Date;

src/pipeline-tasks/pipeline-task-logs.service.spec.ts

@@ -0,0 +1,25 @@
+import { Test, TestingModule } from '@nestjs/testing';
+import { PipelineTaskLogsService } from './pipeline-task-logs.service';
+import { RedisService } from 'nestjs-redis';
+
+describe('PipelineTaskLogsService', () => {
+  let service: PipelineTaskLogsService;
+
+  beforeEach(async () => {
+    const module: TestingModule = await Test.createTestingModule({
+      providers: [
+        PipelineTaskLogsService,
+        {
+          provide: RedisService,
+          useValue: {},
+        },
+      ],
+    }).compile();
+
+    service = module.get<PipelineTaskLogsService>(PipelineTaskLogsService);
+  });
+
+  it('should be defined', () => {
+    expect(service).toBeDefined();
+  });
+});

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

@@ -0,0 +1,78 @@
+import { Injectable } from '@nestjs/common';
+import { log } from 'console';
+import { PubSub } from 'graphql-subscriptions';
+import { RedisService } from 'nestjs-redis';
+import { find, omit, propEq } from 'ramda';
+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) {}
+
+  pubSub = new 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 this.pubSub.asyncIterator(this.getKeys(task));
+  }
+}

src/pipeline-tasks/pipeline-task.consumer.spec.ts

@@ -1,22 +1,22 @@
-import { PIPELINE_TASK_LOG_QUEUE } from './pipeline-tasks.constants';
 import { Test, TestingModule } from '@nestjs/testing';
-import { Job, Queue } from 'bull';
+import { Job } from 'bull';
 import { join } from 'path';
 import { ReposService } from '../repos/repos.service';
 import { PipelineUnits } from './enums/pipeline-units.enum';
 import { PipelineTaskConsumer } from './pipeline-task.consumer';
 import { PipelineTask } from './pipeline-task.entity';
 import { PipelineTasksService } from './pipeline-tasks.service';
-import { getQueueToken } from '@nestjs/bull';
 import { PipelineTaskLogMessage } from './models/pipeline-task-log-message.module';
 import { Pipeline } from '../pipelines/pipeline.entity';
 import { Project } from '../projects/project.entity';
 import { TaskStatuses } from './enums/task-statuses.enum';
+import { PipelineTaskLogsService } from './pipeline-task-logs.service';
+import { ApplicationException } from '../commons/exceptions/application.exception';
 
 describe('PipelineTaskConsumer', () => {
   let consumer: PipelineTaskConsumer;
   let tasksService: PipelineTasksService;
-  let logQueue: Queue<PipelineTaskLogMessage>;
+  let logsService: PipelineTaskLogsService;
   const getJob = () =>
     ({
       data: {
@@ -42,19 +42,20 @@ describe('PipelineTaskConsumer', () => {
             checkout: async () => undefined,
           },
         },
-        PipelineTaskConsumer,
         {
-          provide: getQueueToken(PIPELINE_TASK_LOG_QUEUE),
+          provide: PipelineTaskLogsService,
           useValue: {
-            add: () => undefined,
+            recordLog: async () => undefined,
+            readLogsAsPipelineTaskLogs: async () => [],
           },
         },
+        PipelineTaskConsumer,
       ],
     }).compile();
 
     tasksService = module.get(PipelineTasksService);
+    logsService = module.get(PipelineTaskLogsService);
     consumer = module.get(PipelineTaskConsumer);
-    logQueue = module.get(getQueueToken(PIPELINE_TASK_LOG_QUEUE));
   });
 
   it('should be defined', () => {
@@ -71,31 +72,42 @@ describe('PipelineTaskConsumer', () => {
   });
 
   describe('runScript', () => {
+    let logText: string;
+    let errorText: string;
+    let recordLog: jest.SpyInstance;
+    beforeEach(() => {
+      logText = '';
+      errorText = '';
+      recordLog = jest
+        .spyOn(logsService, 'recordLog')
+        .mockImplementation(async (log: PipelineTaskLogMessage) => {
+          logText += log.message;
+          if (log.isError) {
+            errorText += log.message;
+          }
+        });
+    });
     it('should success and log right message', async () => {
-      const add = jest.spyOn(logQueue, 'add');
-      await expect(
-        consumer
-          .runScript(
+      await consumer.runScript(
         'node one-second-work.js',
         join(__dirname, '../../test/data'),
-          )
-          .then((arr) => arr.join('')),
-      ).resolves.toMatch(/10.+20.+30.+40.+50.+60.+70.+80.+90/s);
-      // expect(add).toHaveBeenCalledTimes(10);
+      );
+      expect(logText).toMatch(/10.+20.+30.+40.+50.+60.+70.+80.+90/s);
+      expect(recordLog).toHaveBeenCalledTimes(10);
       expect(
-        ((add.mock.calls[8][0] as unknown) as PipelineTaskLogMessage).message,
+        ((recordLog.mock.calls[8][0] as unknown) as PipelineTaskLogMessage)
+          .message,
       ).toMatch(/^90/);
     });
     it('should failed and log right message', async () => {
-      const add = jest.spyOn(logQueue, 'add');
       await expect(
         consumer.runScript(
           'node bad-work.js',
           join(__dirname, '../../test/data'),
         ),
-      ).rejects.toThrowError(/Error Message/);
-      // expect(add).toHaveBeenCalledTimes(8);
-      const logs = add.mock.calls
+      ).rejects.toThrowError(/exec script failed/);
+      expect(errorText).toMatch(/Error Message/);
+      const logs = recordLog.mock.calls
         .map((call) => ((call[0] as unknown) as PipelineTaskLogMessage).message)
         .join('');
       expect(logs).toMatch(/10.+20.+30.+40.+50/s);
@@ -104,16 +116,19 @@ describe('PipelineTaskConsumer', () => {
       const task = new PipelineTask();
       task.id = 'test';
 
-      const add = jest.spyOn(logQueue, 'add');
+      const recordLog = jest.spyOn(logsService, 'recordLog');
       await expect(
         consumer.runScript(
           'node bad-work.js',
           join(__dirname, '../../test/data'),
           task,
         ),
-      ).rejects.toThrowError(/Error Message 2/);
+      ).rejects.toThrowError(/exec script failed/);
+
+      expect(errorText).toMatch(/Error Message 2/);
       expect(
-        ((add.mock.calls[2][0] as unknown) as PipelineTaskLogMessage).task,
+        ((recordLog.mock.calls[2][0] as unknown) as PipelineTaskLogMessage)
+          .task,
       ).toMatchObject(task);
     });
   });
@@ -145,6 +160,43 @@ describe('PipelineTaskConsumer', () => {
       task.pipeline.project = new Project();
       task.pipeline.project.name = 'test-project';
     });
+
+    it('success and update task on db', async () => {
+      const job: Job = ({
+        data: task,
+        update: jest.fn().mockImplementation(() => undefined),
+      } as unknown) as Job;
+
+      jest
+        .spyOn(consumer, 'runScript')
+        .mockImplementation(async () => undefined);
+      const updateTask = jest.spyOn(tasksService, 'updateTask');
+
+      await consumer.doTask(job);
+
+      expect(updateTask).toHaveBeenCalledTimes(2);
+      expect(updateTask.mock.calls[0][0].startedAt).toBeDefined();
+      expect(updateTask.mock.calls[1][0].endedAt).toBeDefined();
+      expect(updateTask.mock.calls[1][0].status).toEqual(TaskStatuses.success);
+    });
+    it('failed and update task on db', async () => {
+      const job: Job = ({
+        data: task,
+        update: jest.fn().mockImplementation(() => undefined),
+      } as unknown) as Job;
+
+      jest.spyOn(consumer, 'runScript').mockImplementation(async () => {
+        throw new ApplicationException('exec script failed');
+      });
+      const updateTask = jest.spyOn(tasksService, 'updateTask');
+
+      await consumer.doTask(job);
+
+      expect(updateTask).toHaveBeenCalledTimes(2);
+      expect(updateTask.mock.calls[0][0].startedAt).toBeDefined();
+      expect(updateTask.mock.calls[1][0].endedAt).toBeDefined();
+      expect(updateTask.mock.calls[1][0].status).toEqual(TaskStatuses.failed);
+    });
     it('should do all task', async () => {
       const job: Job = ({
         data: task,
@@ -153,18 +205,17 @@ describe('PipelineTaskConsumer', () => {
 
       const runScript = jest
         .spyOn(consumer, 'runScript')
-        .mockImplementation(async () => []);
+        .mockImplementation(async () => undefined);
       const updateTask = jest.spyOn(tasksService, 'updateTask');
 
       await consumer.doTask(job);
 
       expect(runScript).toHaveBeenCalledTimes(1);
-      expect(updateTask).toHaveBeenCalledTimes(1);
+      expect(updateTask).toHaveBeenCalledTimes(2);
       const taskDto: PipelineTask = updateTask.mock.calls[0][0];
       expect(taskDto.logs).toHaveLength(2);
       expect(taskDto.logs[0].status).toEqual(TaskStatuses.success);
       expect(taskDto.logs[0].unit).toEqual(PipelineUnits.checkout);
-      expect(taskDto.logs[1].logs).toMatch(/Hello, Fennec!/);
     });
     it('should log error message', async () => {
       const job: Job = ({
@@ -181,12 +232,11 @@ describe('PipelineTaskConsumer', () => {
 
       await consumer.doTask(job);
 
-      expect(updateTask).toHaveBeenCalledTimes(1);
+      expect(updateTask).toHaveBeenCalledTimes(2);
       const taskDto: PipelineTask = updateTask.mock.calls[0][0];
       expect(taskDto.logs).toHaveLength(2);
       expect(taskDto.logs[0].status).toEqual(TaskStatuses.success);
       expect(taskDto.logs[1].status).toEqual(TaskStatuses.failed);
-      expect(taskDto.logs[1].logs).toMatch(/bad message/);
     });
   });
 });

src/pipeline-tasks/pipeline-task.consumer.ts

@@ -1,39 +1,37 @@
 import { PipelineTaskLogs } from './models/pipeline-task-logs.model';
 import { ReposService } from './../repos/repos.service';
-import {
-  InjectQueue,
-  OnQueueCompleted,
-  Process,
-  Processor,
-} from '@nestjs/bull';
-import { Job, Queue } from 'bull';
+import { OnQueueCompleted, Process, Processor } from '@nestjs/bull';
+import { Job } from 'bull';
 import { spawn } from 'child_process';
 import { PipelineTask } from './pipeline-task.entity';
-import {
-  PIPELINE_TASK_LOG_QUEUE,
-  PIPELINE_TASK_QUEUE,
-} from './pipeline-tasks.constants';
+import { PIPELINE_TASK_QUEUE } from './pipeline-tasks.constants';
 import { PipelineTasksService } from './pipeline-tasks.service';
 import { ApplicationException } from '../commons/exceptions/application.exception';
 import { PipelineUnits } from './enums/pipeline-units.enum';
 import { PipelineTaskLogMessage } from './models/pipeline-task-log-message.module';
 import { TaskStatuses } from './enums/task-statuses.enum';
+import { PipelineTaskLogsService } from './pipeline-task-logs.service';
 @Processor(PIPELINE_TASK_QUEUE)
 export class PipelineTaskConsumer {
   constructor(
     private readonly service: PipelineTasksService,
     private readonly reposService: ReposService,
-    @InjectQueue(PIPELINE_TASK_LOG_QUEUE)
-    private readonly logQueue: Queue<PipelineTaskLogMessage>,
+    private readonly logsService: PipelineTaskLogsService,
   ) {}
   @Process()
-  async doTask({ data: task, update }: Job<PipelineTask>) {
+  async doTask(job: Job<PipelineTask>) {
+    let task = job.data;
     if (task.pipeline.workUnitMetadata.version !== 1) {
       throw new ApplicationException(
         'work unit metadata version is not match.',
       );
     }
 
+    task.startedAt = new Date();
+    task.status = TaskStatuses.working;
+    task = await this.service.updateTask(task);
+    await job.update(task);
+
     const workspaceRoot = this.reposService.getWorkspaceRootByTask(task);
 
     const units = task.units.map(
@@ -49,16 +47,14 @@ export class PipelineTaskConsumer {
         unitLog.unit = unit.type;
         unitLog.startedAt = new Date();
         try {
-          // 检出代码时，不执行其他脚本。
+          // 检出代码前执行 git checkout
           if (unit.type === PipelineUnits.checkout) {
             await this.reposService.checkout(task, workspaceRoot);
             unitLog.status = TaskStatuses.success;
-            continue;
           }
           for (const script of unit.scripts) {
             unitLog.logs += `[RUN SCRIPT] ${script}`;
-            const messages = await this.runScript(script, workspaceRoot, task);
-            unitLog.logs += messages.join('');
+            await this.runScript(script, workspaceRoot, task, unit.type);
           }
           unitLog.status = TaskStatuses.success;
         } catch (err) {
@@ -67,14 +63,25 @@ export class PipelineTaskConsumer {
           throw err;
         } finally {
           unitLog.endedAt = new Date();
+          unitLog.logs = await this.logsService
+            .readLogsAsPipelineTaskLogs(task)
+            .then(
+              (taskLogs) =>
+                taskLogs.find((tl) => tl.unit === unit.type)?.logs ?? '',
+            );
           task.logs.push(unitLog);
+          await job.update(task);
         }
       }
+
+      task.status = TaskStatuses.success;
     } catch (err) {
+      task.status = TaskStatuses.failed;
       console.log(err);
     } finally {
+      task.endedAt = new Date();
       task = await this.service.updateTask(task);
-      update(task);
+      await job.update(task);
     }
   }
 
@@ -82,31 +89,30 @@ export class PipelineTaskConsumer {
     script: string,
     workspaceRoot: string,
     task?: PipelineTask,
-  ): Promise<string[]> {
+    unit?: PipelineUnits,
+  ): Promise<void> {
     return new Promise((resolve, reject) => {
-      const errorMessages: string[] = [];
-      const logs: string[] = [];
       const sub = spawn(script, {
         shell: true,
         cwd: workspaceRoot,
       });
       sub.stderr.on('data', (data: Buffer) => {
         const str = data.toString();
-        errorMessages.push(str);
-        logs.push(str);
-        this.logQueue.add(PipelineTaskLogMessage.create(task, str));
+        this.logsService.recordLog(
+          PipelineTaskLogMessage.create(task, unit, str, true),
+        );
       });
       sub.stdout.on('data', (data: Buffer) => {
         const str = data.toString();
-        logs.push(str);
-        this.logQueue.add(PipelineTaskLogMessage.create(task, str));
+        this.logsService.recordLog(
+          PipelineTaskLogMessage.create(task, unit, str, false),
+        );
       });
       sub.addListener('close', (code) => {
         if (code === 0) {
-          sub.stdout;
-          return resolve(logs);
+          return resolve();
         }
-        return reject(new ApplicationException(errorMessages.join('')));
+        return reject(new ApplicationException('exec script failed'));
       });
     });
   }

src/pipeline-tasks/pipeline-task.entity.ts

@@ -1,5 +1,5 @@
 import { AppBaseEntity } from './../commons/entities/app-base-entity';
-import { ObjectType } from '@nestjs/graphql';
+import { Field, ObjectType } from '@nestjs/graphql';
 import { Column, Entity, ManyToOne } from 'typeorm';
 import { Pipeline } from '../pipelines/pipeline.entity';
 import { PipelineTaskLogs } from './models/pipeline-task-logs.model';
@@ -26,9 +26,9 @@ export class PipelineTask extends AppBaseEntity {
   @Column({ type: 'enum', enum: TaskStatuses, default: TaskStatuses.pending })
   status: TaskStatuses;
 
-  @Column()
-  startedAt: Date;
+  @Column({ nullable: true })
+  startedAt?: Date;
 
-  @Column()
-  endedAt: Date;
+  @Column({ nullable: true })
+  endedAt?: Date;
 }

src/pipeline-tasks/pipeline-tasks.constants.ts

@@ -1,2 +1,3 @@
 export const PIPELINE_TASK_QUEUE = 'PIPELINE_TASK_QUEUE';
 export const PIPELINE_TASK_LOG_QUEUE = 'PIPELINE_TASK_LOG_QUEUE';
+export const PIPELINE_TASK_LOG_PUBSUB = 'PIPELINE_TASK_LOG_PUBSUB';

src/pipeline-tasks/pipeline-tasks.module.ts

@@ -7,10 +7,14 @@ import { Pipeline } from '../pipelines/pipeline.entity';
 import { ReposModule } from '../repos/repos.module';
 import { RedisModule } from 'nestjs-redis';
 import { BullModule } from '@nestjs/bull';
+import { PipelineTaskConsumer } from './pipeline-task.consumer';
 import {
   PIPELINE_TASK_QUEUE,
   PIPELINE_TASK_LOG_QUEUE,
+  PIPELINE_TASK_LOG_PUBSUB,
 } from './pipeline-tasks.constants';
+import { PipelineTaskLogsService } from './pipeline-task-logs.service';
+import { PubSub } from 'apollo-server-express';
 
 @Module({
   imports: [
@@ -24,6 +28,15 @@ import {
     RedisModule,
     ReposModule,
   ],
-  providers: [PipelineTasksService, PipelineTasksResolver],
+  providers: [
+    PipelineTasksService,
+    PipelineTasksResolver,
+    PipelineTaskConsumer,
+    PipelineTaskLogsService,
+    {
+      provide: Symbol(PIPELINE_TASK_LOG_PUBSUB),
+      useValue: new PubSub(),
+    },
+  ],
 })
 export class PipelineTasksModule {}

src/pipeline-tasks/pipeline-tasks.resolver.spec.ts

@@ -1,12 +1,19 @@
 import { Test, TestingModule } from '@nestjs/testing';
 import { PipelineTasksResolver } from './pipeline-tasks.resolver';
+import { PipelineTaskLogsService } from './pipeline-task-logs.service';
 
 describe('PipelineTasksResolver', () => {
   let resolver: PipelineTasksResolver;
 
   beforeEach(async () => {
     const module: TestingModule = await Test.createTestingModule({
-      providers: [PipelineTasksResolver],
+      providers: [
+        PipelineTasksResolver,
+        {
+          provide: PipelineTaskLogsService,
+          useValue: {},
+        },
+      ],
     }).compile();
 
     resolver = module.get<PipelineTasksResolver>(PipelineTasksResolver);

src/pipeline-tasks/pipeline-tasks.resolver.ts

@@ -1,4 +1,50 @@
-import { Resolver } from '@nestjs/graphql';
+import { Resolver, Args, Mutation, Subscription, Query } from '@nestjs/graphql';
+import { PipelineTask } from './pipeline-task.entity';
+import { PipelineTasksService } from './pipeline-tasks.service';
+import { CreatePipelineTaskInput } from './dtos/create-pipeline-task.input';
+import { PipelineTaskLogMessage } from './models/pipeline-task-log-message.module';
+import { PipelineTaskLogArgs } from './dtos/pipeline-task-log.args';
+import { PipelineTaskLogsService } from './pipeline-task-logs.service';
 
 @Resolver()
-export class PipelineTasksResolver {}
+export class PipelineTasksResolver {
+  constructor(
+    private readonly service: PipelineTasksService,
+    private readonly logsService: PipelineTaskLogsService,
+  ) {}
+
+  @Mutation(() => PipelineTask)
+  async createPipelineTask(@Args('task') taskDto: CreatePipelineTaskInput) {
+    return await this.service.addTask(taskDto);
+  }
+
+  @Subscription(() => PipelineTaskLogMessage, {
+    resolve: (value) => {
+      return value;
+    },
+  })
+  async pipelineTaskLog(@Args() args: PipelineTaskLogArgs) {
+    const task = await this.service.findTaskById(args.taskId);
+    const asyncIterator = this.logsService.watchLogs(task);
+    return asyncIterator;
+  }
+
+  @Subscription(() => PipelineTask, {
+    resolve: (value) => {
+      return value;
+    },
+  })
+  async pipelineTaskChanged(@Args('id') id: string) {
+    return await this.service.watchTaskUpdated(id);
+  }
+
+  @Query(() => [PipelineTask])
+  async listPipelineTaskByPipelineId(@Args('pipelineId') pipelineId: string) {
+    return await this.service.listTasksByPipelineId(pipelineId);
+  }
+
+  @Query(() => PipelineTask)
+  async findPipelineTask(@Args('id') id: string) {
+    return await this.service.findTaskById(id);
+  }
+}

src/pipeline-tasks/pipeline-tasks.service.spec.ts

@@ -155,20 +155,31 @@ describe('PipelineTasksService', () => {
     it('pipeline is busy', async () => {
       let remainTimes = 3;
 
-      const incr = jest
-        .spyOn(redisClient, 'incr')
-        .mockImplementation(() => remainTimes--);
+      let lckValue: string;
+      const set = jest
+        .spyOn(redisClient, 'set')
+        .mockImplementation(async (...args) => {
+          if (remainTimes-- > 0) {
+            throw new Error();
+          } else {
+            lckValue = args[3] as string;
+          }
+        });
+      const get = jest
+        .spyOn(redisClient, 'get')
+        .mockImplementation(async () => lckValue);
+      const del = jest.spyOn(redisClient, 'del');
       const rpop = jest.spyOn(redisClient, 'rpop');
-      const decr = jest.spyOn(redisClient, 'decr');
       const add = jest.spyOn(taskQueue, 'add');
 
       await service.doNextTask(getBasePipeline());
 
       expect(rpop).toHaveBeenCalledTimes(1);
-      expect(incr).toHaveBeenCalledTimes(3);
-      expect(decr).toHaveBeenCalledTimes(3);
+      expect(set).toHaveBeenCalledTimes(4);
+      expect(get).toHaveBeenCalledTimes(1);
+      expect(del).toHaveBeenCalledTimes(1);
       expect(add).toHaveBeenCalledWith(getTask());
-    });
+    }, 10_000);
     it('pipeline always busy and timeout', async () => {
       const set = jest
         .spyOn(redisClient, 'set')

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

@@ -9,9 +9,11 @@ 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';
 
 @Injectable()
 export class PipelineTasksService {
+  pubSub = new PubSub();
   constructor(
     @InjectRepository(PipelineTask)
     private readonly repository: Repository<PipelineTask>,
@@ -33,6 +35,15 @@ export class PipelineTasksService {
     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) {
@@ -71,9 +82,14 @@ export class PipelineTasksService {
   }
 
   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`];
   }

src/pipelines/pipelines.resolver.ts

@@ -1,4 +1,4 @@
-import { Args, Mutation, Query, Resolver } from '@nestjs/graphql';
+import { Args, Mutation, Query, Resolver, Subscription } from '@nestjs/graphql';
 import { CreatePipelineInput } from './dtos/create-pipeline.input';
 import { UpdatePipelineInput } from './dtos/update-pipeline.input';
 import { Pipeline } from './pipeline.entity';
@@ -43,8 +43,15 @@ export class PipelinesResolver {
     return await this.service.remove(id);
   }
 
-  @Query(() => String)
+  @Subscription(() => LogList, {
+    resolve: (value) => {
+      return value;
+    },
+  })
   async listLogsForPipeline(@Args('id', { type: () => String }) id: string) {
-    return await this.service.listLogsForPipeline(id);
+    const job = await this.service.listLogsForPipeline(id);
+    return (async function* () {
+      yield await job.finished();
+    })();
   }
 }

src/pipelines/pipelines.service.spec.ts

@@ -59,7 +59,7 @@ describe('PipelinesService', () => {
       const add = jest.spyOn(queue, 'add');
       await expect(
         service.listLogsForPipeline('test-pipeline'),
-      ).resolves.toEqual(1);
+      ).resolves.toEqual({ id: 1 });
       expect(add).toBeCalledWith({
         project: pipeline.project,
         branch: pipeline.branch,

src/pipelines/pipelines.service.ts

@@ -13,9 +13,7 @@ import { ListLogsOption } from '../repos/models/list-logs.options';
 
 @Injectable()
 export class PipelinesService extends BaseDbService<Pipeline> {
-  readonly uniqueFields: Array<Array<keyof Pipeline>> = [
-    ['branch', 'projectId'],
-  ];
+  readonly uniqueFields: Array<Array<keyof Pipeline>> = [['projectId', 'name']];
   constructor(
     @InjectRepository(Pipeline)
     readonly repository: Repository<Pipeline>,
@@ -52,6 +50,6 @@ export class PipelinesService extends BaseDbService<Pipeline> {
       project: pipeline.project,
       branch: pipeline.branch,
     });
-    return job.id;
+    return job;
   }
 }

src/repos/list-logs.consumer.ts

@@ -0,0 +1,14 @@
+import { ReposService } from './repos.service';
+import { Processor, Process } from '@nestjs/bull';
+import { Job } from 'bull';
+import { ListLogsOption } from './models/list-logs.options';
+import { LIST_LOGS_TASK } from './repos.constants';
+@Processor(LIST_LOGS_TASK)
+export class ListLogsConsumer {
+  constructor(private readonly service: ReposService) {}
+  @Process()
+  async listLogs(job: Job<ListLogsOption>) {
+    const logs = await this.service.listLogs(job.data);
+    return logs;
+  }
+}

src/repos/repos.constants.ts

@@ -1 +1,3 @@
 export const LIST_LOGS_TASK = 'LIST_LOGS_TASK';
+export const LIST_LOGS_PUB_SUB = 'LIST_LOGS_PUB_SUB';
+export const LIST_LOGS_DONE = 'LIST_LOGS_DONE';

src/repos/repos.module.ts

@@ -5,10 +5,21 @@ import { ReposResolver } from './repos.resolver';
 import { ReposService } from './repos.service';
 import { ConfigModule } from '@nestjs/config';
 import { ProjectsModule } from '../projects/projects.module';
+import { BullModule } from '@nestjs/bull';
+import { LIST_LOGS_TASK, LIST_LOGS_PUB_SUB } from './repos.constants';
+import { PubSub } from 'graphql-subscriptions';
+import { ListLogsConsumer } from './list-logs.consumer';
 
 @Module({
-  imports: [TypeOrmModule.forFeature([Project]), ConfigModule, ProjectsModule],
-  providers: [ReposResolver, ReposService],
+  imports: [
+    TypeOrmModule.forFeature([Project]),
+    ConfigModule,
+    ProjectsModule,
+    BullModule.registerQueue({
+      name: LIST_LOGS_TASK,
+    }),
+  ],
+  providers: [ReposResolver, ReposService, ListLogsConsumer],
   exports: [ReposService],
 })
 export class ReposModule {}

src/repos/repos.resolver.ts

@@ -1,26 +1,4 @@
-import { Args, Query, Resolver } from '@nestjs/graphql';
-import { ListLogsArgs } from './dtos/list-logs.args';
-import { ReposService } from './repos.service';
-import { LogList } from './dtos/log-list.model';
-import { ListBranchesArgs } from './dtos/list-branches.args';
-import { BranchList } from './dtos/branch-list.model';
+import { Resolver } from '@nestjs/graphql';
 
 @Resolver()
-export class ReposResolver {
-  constructor(private readonly service: ReposService) {}
-  @Query(() => LogList)
-  async listLogs(@Args('listLogsArgs') dto: ListLogsArgs) {
-    return await this.service.listLogs(dto);
-  }
-  @Query(() => BranchList)
-  async listBranches(
-    @Args('listBranchesArgs') dto: ListBranchesArgs,
-  ): Promise<BranchList> {
-    return await this.service.listBranches(dto).then((data) => {
-      return {
-        ...data,
-        branches: Object.values(data.branches),
-      };
-    });
-  }
-}
+export class ReposResolver {}

src/repos/repos.service.spec.ts

@@ -63,7 +63,10 @@ describe('ReposService', () => {
   });
   describe('listLogs', () => {
     it('should be return logs', async () => {
-      const result = await service.listLogs({ projectId: '1' });
+      const result = await service.listLogs({
+        project: getTest1Project(),
+        branch: 'master',
+      });
       expect(result).toBeDefined();
     }, 20_000);
   });
@@ -71,7 +74,7 @@ describe('ReposService', () => {
     it('should be return branches', async () => {
       const result = await service.listBranches({ projectId: '1' });
       expect(result).toBeDefined();
-    }, 10_000);
+    }, 20_000);
   });
 
   describe.skip('checkout', () => {

src/repos/repos.service.ts

@@ -1,3 +1,4 @@
+import { ListLogsOption } from './models/list-logs.options';
 import { Pipeline } from './../pipelines/pipeline.entity';
 import { PipelineTask } from './../pipeline-tasks/pipeline-task.entity';
 import { Injectable, NotFoundException } from '@nestjs/common';
@@ -39,7 +40,7 @@ export class ReposService {
       await mkdir(workspaceRoot, { recursive: true });
     });
     const git = gitP(workspaceRoot);
-    if (!(await git.checkIsRepo())) {
+    if (!(await git.checkIsRepo().catch(() => false))) {
       await git.init();
       await git.addRemote(DEFAULT_REMOTE_NAME, project.sshUrl);
     }
@@ -47,13 +48,10 @@ export class ReposService {
     return git;
   }
 
-  async listLogs(dto: ListLogsArgs) {
-    const project = await this.projectRepository.findOneOrFail({
-      id: dto.projectId,
-    });
+  async listLogs({ project, branch }: ListLogsOption) {
     const git = await this.getGit(project);
     return await git.log(
-      dto.branch ? ['--branches', dto.branch, '--'] : ['--all'],
+      branch ? ['--branches', `remotes/origin/${branch}`, '--'] : ['--all'],
     );
   }