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

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

.vscode/settings.json

@@ -1,5 +1,9 @@
 {
   "cSpell.words": [
-    "Repos"
+    "Repos",
+    "lpush",
+    "lrange",
+    "rpop",
+    "rpush"
   ]
 }

config.yml.example

@@ -9,5 +9,10 @@ db:
     database: fennec
     username: fennec
     password:
+  redis:
+    host: 192.168.31.194
+    port: 6379
+    password:
+    prefix: fennec
 workspaces:
   root: '/Users/ivanli/Projects/fennec/workspaces'

docs/ci-cd.md

@@ -0,0 +1,33 @@
+# CI/CD 流程
+0. 准备
+  - project information
+  - commit hash
+1. checkout
+2. install dependencies
+3. run test script
+5. run deploy script
+6. clear workspace
+
+## 流水线任务单元描述
+```json
+{
+  "version": 1,
+  "unit": {
+    "install-dependencies": {
+      "script": "npm ci"
+    },
+    "test": {
+      "script": "npm test"
+    },
+    "build": {
+      "script": "npm build"
+    },
+    "deploy": {
+      "script": [
+        "npm build"
+      ]
+    }
+  }
+}
+
+```

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

@@ -7,7 +7,10 @@ import { AppResolver } from './app.resolver';
 import { AppService } from './app.service';
 import { ProjectsModule } from './projects/projects.module';
 import { ReposModule } from './repos/repos.module';
+import { PipelinesModule } from './pipelines/pipelines.module';
+import { PipelineTasksModule } from './pipeline-tasks/pipeline-tasks.module';
 import configuration from './commons/config/configuration';
+import { RedisModule } from 'nestjs-redis';
 
 @Module({
   imports: [
@@ -34,11 +37,24 @@ import configuration from './commons/config/configuration';
         debug: configService.get<string>('env') !== 'prod',
         playground: true,
         autoSchemaFile: true,
+        installSubscriptionHandlers: true,
       }),
       inject: [ConfigService],
     }),
     ProjectsModule,
     ReposModule,
+    PipelinesModule,
+    PipelineTasksModule,
+    RedisModule.forRootAsync({
+      imports: [ConfigModule],
+      useFactory: (configService: ConfigService) => ({
+        host: configService.get<string>('db.redis.host', 'localhost'),
+        port: configService.get<number>('db.redis.port', 6379),
+        password: configService.get<string>('db.redis.password', ''),
+        keyPrefix: configService.get<string>('db.redis.prefix', 'fennec'),
+      }),
+      inject: [ConfigService],
+    }),
   ],
   controllers: [AppController],
   providers: [AppService, AppResolver],

src/commons/entities/app-base-entity.ts

@@ -1,6 +1,7 @@
 import { Field, ID, ObjectType } from '@nestjs/graphql';
 import {
   CreateDateColumn,
+  DeleteDateColumn,
   PrimaryGeneratedColumn,
   UpdateDateColumn,
 } from 'typeorm';
@@ -16,4 +17,7 @@ export class AppBaseEntity {
 
   @UpdateDateColumn({ select: false })
   updatedAt: Date;
+
+  @DeleteDateColumn({ select: false })
+  deletedAt?: Date;
 }

src/commons/filters/all.exception-filter.ts

@@ -1,55 +1,27 @@
 import {
-  ArgumentsHost,
-  Catch,
   ExceptionFilter,
+  Catch,
+  ArgumentsHost,
   HttpException,
-  HttpStatus,
 } from '@nestjs/common';
-import { EntityNotFoundError } from 'typeorm/error/EntityNotFoundError';
+import { ApolloError } from 'apollo-server-errors';
 
-@Catch()
-export class AllExceptionsFilter implements ExceptionFilter {
-  catch(exception: any, host: ArgumentsHost) {
-    const ctx = host.switchToHttp();
-    const response = ctx.getResponse();
-    const request = ctx.getRequest();
-
-    const status =
-      exception instanceof HttpException
-        ? exception.getStatus()
-        : HttpStatus.INTERNAL_SERVER_ERROR;
-
-    if (exception instanceof HttpException) {
-      const ex = exception.getResponse();
-      if (ex instanceof Object) {
-        response.status(status).json({
-          ...ex,
-          timestamp: Date.now(),
-          path: request.url,
-        });
-      } else {
-        response.status(status).json({
-          message: ex,
-          timestamp: Date.now(),
-          path: request.url,
-        });
-      }
-    } else if (exception instanceof EntityNotFoundError) {
-      response.status(HttpStatus.NOT_FOUND).json({
-        message: '资源未找到!',
-        timestamp: Date.now(),
-        path: request.url,
-      });
+@Catch(HttpException)
+export class HttpExceptionFilter implements ExceptionFilter {
+  catch(exception: HttpException, host: ArgumentsHost) {
+    const message = exception.message;
+    const extensions: Record<string, any> = {};
+    const err = exception.getResponse();
+    if (typeof err === 'string') {
+      extensions.message = err;
     } else {
-      console.error('服务器内部错误');
-      console.error(exception);
-      response.status(HttpStatus.INTERNAL_SERVER_ERROR).json({
-        code: status,
-        timestamp: new Date().toISOString(),
-        message: '服务器内部错误',
-        error: exception,
-        path: request.url,
-      });
+      Object.assign(extensions, (err as any).extension);
+      extensions.message = (err as any).message;
     }
+    return new ApolloError(
+      message,
+      exception.getStatus().toString(),
+      extensions,
+    );
   }
 }

src/commons/pipes/sanitize.pipe.ts

@@ -0,0 +1,15 @@
+import { ArgumentMetadata, Injectable, PipeTransform } from '@nestjs/common';
+import { sanitize } from '@neuralegion/class-sanitizer/dist';
+
+@Injectable()
+export class SanitizePipe implements PipeTransform {
+  transform(value: any, metadata: ArgumentMetadata) {
+    // console.log(value, typeof value);
+    if (value instanceof Object) {
+      value = Object.assign(new metadata.metatype(), value);
+      sanitize(value);
+      // console.log(value);
+    }
+    return value;
+  }
+}

src/main.ts

@@ -2,11 +2,19 @@ import { ValidationPipe } from '@nestjs/common';
 import { ConfigService } from '@nestjs/config';
 import { NestFactory } from '@nestjs/core';
 import { AppModule } from './app.module';
+import { HttpExceptionFilter } from './commons/filters/all.exception-filter';
+import { SanitizePipe } from './commons/pipes/sanitize.pipe';
 
 async function bootstrap() {
   const app = await NestFactory.create(AppModule);
   const configService = app.get(ConfigService);
-  app.useGlobalPipes(new ValidationPipe());
+  app.useGlobalPipes(new SanitizePipe());
+  app.useGlobalPipes(
+    new ValidationPipe({
+      transform: true,
+    }),
+  );
+  app.useGlobalFilters(new HttpExceptionFilter());
   await app.listen(configService.get<number>('http.port'));
 }
 bootstrap();

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

@@ -0,0 +1,11 @@
+import { Field, InputType } from '@nestjs/graphql';
+import { PipelineUnits } from '../enums/pipeline-units.enum';
+
+@InputType()
+export class CreatePipelineTaskInput {
+  pipelineId: string;
+
+  commit: string;
+
+  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/pipeline-units.enum.ts

@@ -0,0 +1,14 @@
+import { registerEnumType } from '@nestjs/graphql';
+
+export enum PipelineUnits {
+  checkout = 'checkout',
+  installDependencies = 'installDependencies',
+  test = 'test',
+  deploy = 'deploy',
+  cleanUp = 'cleanUp',
+}
+
+registerEnumType(PipelineUnits, {
+  name: 'PipelineUnits',
+  description: '流水线单元',
+});

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

@@ -0,0 +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

@@ -0,0 +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,
+    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

@@ -0,0 +1,14 @@
+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;
+  logs = '';
+}

src/pipeline-tasks/models/work-unit-metadata.model.ts

@@ -0,0 +1,9 @@
+import { InputType, ObjectType } from '@nestjs/graphql';
+import { WorkUnit } from './work-unit.model';
+
+@InputType('WorkUnitMetadataInput')
+@ObjectType()
+export class WorkUnitMetadata {
+  version = 1;
+  units: WorkUnit[];
+}

src/pipeline-tasks/models/work-unit.model.ts

@@ -0,0 +1,13 @@
+import { Field, InputType, ObjectType } from '@nestjs/graphql';
+import {
+  PipelineUnits,
+  PipelineUnits as PipelineUnitTypes,
+} from '../enums/pipeline-units.enum';
+
+@ObjectType()
+@InputType('WorkUnitInput')
+export class WorkUnit {
+  @Field(() => PipelineUnits)
+  type: PipelineUnitTypes;
+  scripts: string[];
+}

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

@@ -0,0 +1,242 @@
+import { Test, TestingModule } from '@nestjs/testing';
+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 { 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 logsService: PipelineTaskLogsService;
+  const getJob = () =>
+    ({
+      data: {
+        pipelineId: 'test',
+        units: [PipelineUnits.checkout, PipelineUnits.test],
+      },
+    } as Job<PipelineTask>);
+
+  beforeEach(async () => {
+    const module: TestingModule = await Test.createTestingModule({
+      providers: [
+        {
+          provide: PipelineTasksService,
+          useValue: {
+            doNextTask: () => undefined,
+            updateTask: async (value) => value,
+          },
+        },
+        {
+          provide: ReposService,
+          useValue: {
+            getWorkspaceRootByTask: () => 'workspace-root',
+            checkout: async () => undefined,
+          },
+        },
+        {
+          provide: PipelineTaskLogsService,
+          useValue: {
+            recordLog: async () => undefined,
+            readLogsAsPipelineTaskLogs: async () => [],
+          },
+        },
+        PipelineTaskConsumer,
+      ],
+    }).compile();
+
+    tasksService = module.get(PipelineTasksService);
+    logsService = module.get(PipelineTaskLogsService);
+    consumer = module.get(PipelineTaskConsumer);
+  });
+
+  it('should be defined', () => {
+    expect(consumer).toBeDefined();
+  });
+
+  describe('onCompleted', () => {
+    it('should call doNextTask()', () => {
+      const job = getJob();
+      const doNextTask = jest.spyOn(tasksService, 'doNextTask');
+      consumer.onCompleted(job);
+      expect(doNextTask).toHaveBeenCalledTimes(1);
+    });
+  });
+
+  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 () => {
+      await consumer.runScript(
+        'node one-second-work.js',
+        join(__dirname, '../../test/data'),
+      );
+      expect(logText).toMatch(/10.+20.+30.+40.+50.+60.+70.+80.+90/s);
+      expect(recordLog).toHaveBeenCalledTimes(10);
+      expect(
+        ((recordLog.mock.calls[8][0] as unknown) as PipelineTaskLogMessage)
+          .message,
+      ).toMatch(/^90/);
+    });
+    it('should failed and log right message', async () => {
+      await expect(
+        consumer.runScript(
+          'node bad-work.js',
+          join(__dirname, '../../test/data'),
+        ),
+      ).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);
+    });
+    it('should log with task', async () => {
+      const task = new PipelineTask();
+      task.id = 'test';
+
+      const recordLog = jest.spyOn(logsService, 'recordLog');
+      await expect(
+        consumer.runScript(
+          'node bad-work.js',
+          join(__dirname, '../../test/data'),
+          task,
+        ),
+      ).rejects.toThrowError(/exec script failed/);
+
+      expect(errorText).toMatch(/Error Message 2/);
+      expect(
+        ((recordLog.mock.calls[2][0] as unknown) as PipelineTaskLogMessage)
+          .task,
+      ).toMatchObject(task);
+    });
+  });
+
+  describe('doTask', () => {
+    let task: PipelineTask;
+
+    beforeEach(() => {
+      task = new PipelineTask();
+      task.id = 'test-id';
+      task.logs = [];
+      task.pipeline = new Pipeline();
+      task.pipeline.workUnitMetadata = {
+        version: 1,
+        units: [
+          {
+            type: PipelineUnits.checkout,
+            scripts: [],
+          },
+          {
+            type: PipelineUnits.installDependencies,
+            scripts: ["echo ' Hello, Fennec!'"],
+          },
+        ],
+      };
+      task.units = task.pipeline.workUnitMetadata.units.map(
+        (unit) => unit.type,
+      );
+      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,
+        update: jest.fn().mockImplementation(() => undefined),
+      } as unknown) as Job;
+
+      const runScript = jest
+        .spyOn(consumer, 'runScript')
+        .mockImplementation(async () => undefined);
+      const updateTask = jest.spyOn(tasksService, 'updateTask');
+
+      await consumer.doTask(job);
+
+      expect(runScript).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);
+    });
+    it('should log error message', async () => {
+      const job: Job = ({
+        data: task,
+        update: jest.fn().mockImplementation(() => undefined),
+      } as unknown) as Job;
+
+      const runScript = jest
+        .spyOn(consumer, 'runScript')
+        .mockImplementation(async () => {
+          throw new Error('bad message');
+        });
+      const updateTask = jest.spyOn(tasksService, 'updateTask');
+
+      await consumer.doTask(job);
+
+      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);
+    });
+  });
+});

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

@@ -0,0 +1,124 @@
+import { PipelineTaskLogs } from './models/pipeline-task-logs.model';
+import { ReposService } from './../repos/repos.service';
+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_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,
+    private readonly logsService: PipelineTaskLogsService,
+  ) {}
+  @Process()
+  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(
+      (type) =>
+        task.pipeline.workUnitMetadata.units.find(
+          (unit) => unit.type === type,
+        ) ?? { type: type, scripts: [] },
+    );
+
+    try {
+      for (const unit of units) {
+        const unitLog = new PipelineTaskLogs();
+        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;
+          }
+          for (const script of unit.scripts) {
+            unitLog.logs += `[RUN SCRIPT] ${script}`;
+            await this.runScript(script, workspaceRoot, task, unit.type);
+          }
+          unitLog.status = TaskStatuses.success;
+        } catch (err) {
+          unitLog.status = TaskStatuses.failed;
+          unitLog.logs += err.message;
+          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);
+      await job.update(task);
+    }
+  }
+
+  async runScript(
+    script: string,
+    workspaceRoot: string,
+    task?: PipelineTask,
+    unit?: PipelineUnits,
+  ): Promise<void> {
+    return new Promise((resolve, reject) => {
+      const sub = spawn(script, {
+        shell: true,
+        cwd: workspaceRoot,
+      });
+      sub.stderr.on('data', (data: Buffer) => {
+        const str = data.toString();
+        this.logsService.recordLog(
+          PipelineTaskLogMessage.create(task, unit, str, true),
+        );
+      });
+      sub.stdout.on('data', (data: Buffer) => {
+        const str = data.toString();
+        this.logsService.recordLog(
+          PipelineTaskLogMessage.create(task, unit, str, false),
+        );
+      });
+      sub.addListener('close', (code) => {
+        if (code === 0) {
+          return resolve();
+        }
+        return reject(new ApplicationException('exec script failed'));
+      });
+    });
+  }
+
+  @OnQueueCompleted()
+  onCompleted(job: Job<PipelineTask>) {
+    this.service.doNextTask(job.data.pipeline);
+  }
+}

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

@@ -0,0 +1,34 @@
+import { AppBaseEntity } from './../commons/entities/app-base-entity';
+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';
+import { TaskStatuses } from './enums/task-statuses.enum';
+import { PipelineUnits } from './enums/pipeline-units.enum';
+
+@ObjectType()
+@Entity()
+export class PipelineTask extends AppBaseEntity {
+  @ManyToOne(() => Pipeline)
+  pipeline: Pipeline;
+  @Column()
+  pipelineId: string;
+
+  @Column()
+  commit: string;
+
+  @Column({ type: 'enum', enum: PipelineUnits, array: true })
+  units: PipelineUnits[];
+
+  @Column({ type: 'jsonb', default: '[]' })
+  logs: PipelineTaskLogs[];
+
+  @Column({ type: 'enum', enum: TaskStatuses, default: TaskStatuses.pending })
+  status: TaskStatuses;
+
+  @Column({ nullable: true })
+  startedAt?: Date;
+
+  @Column({ nullable: true })
+  endedAt?: Date;
+}

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

@@ -0,0 +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

@@ -0,0 +1,42 @@
+import { Module } from '@nestjs/common';
+import { PipelineTasksService } from './pipeline-tasks.service';
+import { PipelineTasksResolver } from './pipeline-tasks.resolver';
+import { TypeOrmModule } from '@nestjs/typeorm';
+import { PipelineTask } from './pipeline-task.entity';
+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: [
+    TypeOrmModule.forFeature([PipelineTask, Pipeline]),
+    BullModule.registerQueue(
+      {
+        name: PIPELINE_TASK_QUEUE,
+      },
+      { name: PIPELINE_TASK_LOG_QUEUE },
+    ),
+    RedisModule,
+    ReposModule,
+  ],
+  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

@@ -0,0 +1,25 @@
+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,
+        {
+          provide: PipelineTaskLogsService,
+          useValue: {},
+        },
+      ],
+    }).compile();
+
+    resolver = module.get<PipelineTasksResolver>(PipelineTasksResolver);
+  });
+
+  it('should be defined', () => {
+    expect(resolver).toBeDefined();
+  });
+});

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

@@ -0,0 +1,50 @@
+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 {
+  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

@@ -0,0 +1,201 @@
+import { Test, TestingModule } from '@nestjs/testing';
+import { PipelineTasksService } from './pipeline-tasks.service';
+import { getRepositoryToken } from '@nestjs/typeorm';
+import { PipelineTask } from './pipeline-task.entity';
+import { PIPELINE_TASK_QUEUE } from './pipeline-tasks.constants';
+import { getQueueToken } from '@nestjs/bull';
+import { RedisService } from 'nestjs-redis';
+import { Pipeline } from '../pipelines/pipeline.entity';
+import { EntityNotFoundError } from 'typeorm/error/EntityNotFoundError';
+import { Repository } from 'typeorm';
+import { Queue } from 'bull';
+import { LockFailedException } from '../commons/exceptions/lock-failed.exception';
+
+describe('PipelineTasksService', () => {
+  let service: PipelineTasksService;
+  let module: TestingModule;
+  let taskRepository: Repository<PipelineTask>;
+  let pipelineRepository: Repository<Pipeline>;
+  const getBasePipeline = () =>
+    ({
+      id: 'test',
+      name: '测试流水线',
+      branch: 'master',
+      workUnitMetadata: {},
+      project: {
+        id: 'test-project',
+      },
+    } as Pipeline);
+  let redisClient;
+  let taskQueue: Queue;
+  const getTask = () =>
+    ({
+      pipelineId: 'test',
+      commit: 'test',
+      units: [],
+    } as PipelineTask);
+
+  beforeEach(async () => {
+    redisClient = (() => ({
+      set: jest.fn().mockImplementation(async () => 'OK'),
+      del: jest.fn().mockImplementation(async () => 'test'),
+      get: jest.fn().mockImplementation(async () => 'test'),
+      lpush: jest.fn().mockImplementation(async () => 1),
+      rpop: jest.fn().mockImplementation(async () => JSON.stringify(getTask())),
+    }))() as any;
+    taskQueue = (() => ({
+      add: jest.fn().mockImplementation(async () => null),
+    }))() as any;
+    module = await Test.createTestingModule({
+      providers: [
+        PipelineTasksService,
+        {
+          provide: getRepositoryToken(PipelineTask),
+          useValue: new Repository(),
+        },
+        {
+          provide: getRepositoryToken(Pipeline),
+          useValue: new Repository(),
+        },
+        {
+          provide: getQueueToken(PIPELINE_TASK_QUEUE),
+          useValue: taskQueue,
+        },
+        {
+          provide: RedisService,
+          useValue: {
+            getClient: jest.fn(() => redisClient),
+          },
+        },
+      ],
+    }).compile();
+
+    service = module.get<PipelineTasksService>(PipelineTasksService);
+    taskRepository = module.get(getRepositoryToken(PipelineTask));
+    pipelineRepository = module.get(getRepositoryToken(Pipeline));
+    jest
+      .spyOn(taskRepository, 'save')
+      .mockImplementation(async (data: any) => data);
+    jest
+      .spyOn(taskRepository, 'create')
+      .mockImplementation((data: any) => data);
+  });
+
+  it('should be defined', () => {
+    expect(service).toBeDefined();
+  });
+
+  describe('addTask', () => {
+    beforeEach(() => {
+      jest
+        .spyOn(pipelineRepository, 'findOneOrFail')
+        .mockImplementation(async () => getBasePipeline());
+    });
+    it('pipeline not found', async () => {
+      jest.spyOn(taskRepository, 'findOneOrFail').mockImplementation(() => {
+        throw new EntityNotFoundError(Pipeline, {});
+      });
+      await expect(
+        service.addTask({ pipelineId: 'test', commit: 'test', units: [] }),
+      ).rejects;
+    });
+    it('create task on db', async () => {
+      const save = jest
+        .spyOn(taskRepository, 'save')
+        .mockImplementation(async (data: any) => data);
+      jest
+        .spyOn(service, 'doNextTask')
+        .mockImplementation(async () => undefined);
+      await service.addTask({ pipelineId: 'test', commit: 'test', units: [] }),
+        expect(save.mock.calls[0][0]).toMatchObject({
+          pipelineId: 'test',
+          commit: 'test',
+          units: [],
+        });
+    });
+    it('add task', async () => {
+      const lpush = jest.spyOn(redisClient, 'lpush');
+      const doNextTask = jest.spyOn(service, 'doNextTask');
+      jest
+        .spyOn(service, 'doNextTask')
+        .mockImplementation(async () => undefined);
+      await service.addTask({ pipelineId: 'test', commit: 'test', units: [] });
+      expect(typeof lpush.mock.calls[0][1] === 'string').toBeTruthy();
+      expect(JSON.parse(lpush.mock.calls[0][1] as string)).toMatchObject({
+        pipelineId: 'test',
+        commit: 'test',
+        units: [],
+        pipeline: getBasePipeline(),
+      });
+      expect(doNextTask).toHaveBeenCalledWith(getBasePipeline());
+    });
+  });
+
+  describe('doNextTask', () => {
+    it('add task to queue', async () => {
+      let lckValue: string;
+      const set = jest
+        .spyOn(redisClient, 'set')
+        .mockImplementation(async (...args) => (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 add = jest.spyOn(taskQueue, 'add');
+
+      await service.doNextTask(getBasePipeline());
+
+      expect(add).toHaveBeenCalledWith(getTask());
+      expect(set).toHaveBeenCalledTimes(1);
+      expect(rpop).toHaveBeenCalledTimes(1);
+      expect(get).toHaveBeenCalledTimes(1);
+      expect(del).toHaveBeenCalledTimes(1);
+    });
+    it('pipeline is busy', async () => {
+      let remainTimes = 3;
+
+      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 add = jest.spyOn(taskQueue, 'add');
+
+      await service.doNextTask(getBasePipeline());
+
+      expect(rpop).toHaveBeenCalledTimes(1);
+      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')
+        .mockImplementation(async () => {
+          throw new Error();
+        });
+      const get = jest.spyOn(redisClient, 'get');
+      const del = jest.spyOn(redisClient, 'del');
+
+      await expect(
+        service.doNextTask(getBasePipeline()),
+      ).rejects.toBeInstanceOf(LockFailedException);
+
+      expect(set).toHaveBeenCalledTimes(5);
+      expect(get).toHaveBeenCalledTimes(0);
+      expect(del).toHaveBeenCalledTimes(0);
+    }, 15_000);
+  });
+});

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

@@ -0,0 +1,96 @@
+import { Injectable } from '@nestjs/common';
+import { InjectRepository } from '@nestjs/typeorm';
+import { PipelineTask } from './pipeline-task.entity';
+import { 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';
+
+@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 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`];
+  }
+}

src/pipelines/dtos/create-pipeline.input.ts

@@ -0,0 +1,27 @@
+import { InputType } from '@nestjs/graphql';
+import { WorkUnitMetadata } from '../../pipeline-tasks/models/work-unit-metadata.model';
+import {
+  IsObject,
+  IsOptional,
+  IsString,
+  IsUUID,
+  MaxLength,
+} from 'class-validator';
+
+@InputType({ isAbstract: true })
+export class CreatePipelineInput {
+  @IsUUID()
+  projectId: string;
+
+  @IsString()
+  @MaxLength(100)
+  branch: string;
+
+  @IsString()
+  @MaxLength(32)
+  name: string;
+
+  @IsOptional()
+  @IsObject()
+  workUnitMetadata: WorkUnitMetadata;
+}

src/pipelines/dtos/list-pipelines.args.ts

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

src/pipelines/dtos/update-pipeline.input.ts

@@ -0,0 +1,5 @@
+import { InputType } from '@nestjs/graphql';
+import { CreatePipelineInput } from './create-pipeline.input';
+
+@InputType()
+export class UpdatePipelineInput extends CreatePipelineInput {}

src/pipelines/pipeline.entity.ts

@@ -0,0 +1,23 @@
+import { Column, Entity, ManyToOne } from 'typeorm';
+import { AppBaseEntity } from '../commons/entities/app-base-entity';
+import { Project } from '../projects/project.entity';
+import { ObjectType } from '@nestjs/graphql';
+import { WorkUnitMetadata } from '../pipeline-tasks/models/work-unit-metadata.model';
+
+@ObjectType()
+@Entity()
+export class Pipeline extends AppBaseEntity {
+  @ManyToOne(() => Project)
+  project: Project;
+  @Column()
+  projectId: string;
+
+  @Column({ comment: 'eg: remotes/origin/master' })
+  branch: string;
+
+  @Column()
+  name: string;
+
+  @Column({ type: 'jsonb' })
+  workUnitMetadata: WorkUnitMetadata;
+}

src/pipelines/pipelines.module.ts

@@ -0,0 +1,18 @@
+import { Module } from '@nestjs/common';
+import { PipelinesResolver } from './pipelines.resolver';
+import { PipelinesService } from './pipelines.service';
+import { TypeOrmModule } from '@nestjs/typeorm';
+import { Pipeline } from './pipeline.entity';
+import { BullModule } from '@nestjs/bull';
+import { LIST_LOGS_TASK } from '../repos/repos.constants';
+
+@Module({
+  imports: [
+    TypeOrmModule.forFeature([Pipeline]),
+    BullModule.registerQueue({
+      name: LIST_LOGS_TASK,
+    }),
+  ],
+  providers: [PipelinesResolver, PipelinesService],
+})
+export class PipelinesModule {}

src/pipelines/pipelines.resolver.spec.ts

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

src/pipelines/pipelines.resolver.ts

@@ -0,0 +1,57 @@
+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';
+import { PipelinesService } from './pipelines.service';
+import { ListPipelineArgs } from './dtos/list-pipelines.args';
+import { LogList } from '../repos/dtos/log-list.model';
+
+@Resolver()
+export class PipelinesResolver {
+  constructor(private readonly service: PipelinesService) {}
+  @Query(() => [Pipeline])
+  async listPipelines(@Args() dto: ListPipelineArgs) {
+    return await this.service.list(dto);
+  }
+
+  @Query(() => Pipeline)
+  async findPipeline(@Args('id', { type: () => String }) id: string) {
+    return await this.service.findOne(id);
+  }
+
+  @Mutation(() => Pipeline)
+  async createPipeline(
+    @Args('pipeline', { type: () => CreatePipelineInput })
+    dto: UpdatePipelineInput,
+  ) {
+    return await this.service.create(dto);
+  }
+
+  @Mutation(() => Pipeline)
+  async modifyPipeline(
+    @Args('id', { type: () => String }) id: string,
+    @Args('Pipeline', { type: () => UpdatePipelineInput })
+    dto: UpdatePipelineInput,
+  ) {
+    const tmp = await this.service.update(id, dto);
+    console.log(tmp);
+    return tmp;
+  }
+
+  @Mutation(() => Number)
+  async deletePipeline(@Args('id', { type: () => String }) id: string) {
+    return await this.service.remove(id);
+  }
+
+  @Subscription(() => LogList, {
+    resolve: (value) => {
+      return value;
+    },
+  })
+  async listLogsForPipeline(@Args('id', { type: () => String }) id: string) {
+    const job = await this.service.listLogsForPipeline(id);
+    return (async function* () {
+      yield await job.finished();
+    })();
+  }
+}

src/pipelines/pipelines.service.spec.ts

@@ -0,0 +1,69 @@
+import { Test, TestingModule } from '@nestjs/testing';
+import { PipelinesService } from './pipelines.service';
+import { Pipeline } from './pipeline.entity';
+import { getRepositoryToken } from '@nestjs/typeorm';
+import { getQueueToken } from '@nestjs/bull';
+import { LIST_LOGS_TASK } from '../repos/repos.constants';
+import { Repository } from 'typeorm';
+import { Project } from '../projects/project.entity';
+import { Job, Queue } from 'bull';
+import { ListLogsOption } from '../repos/models/list-logs.options';
+
+describe('PipelinesService', () => {
+  let service: PipelinesService;
+  let repository: Repository<Pipeline>;
+  let pipeline: Pipeline;
+  let queue: Queue<ListLogsOption>;
+
+  beforeEach(async () => {
+    pipeline = Object.assign(new Pipeline(), {
+      id: 'test-pipeline',
+      name: 'pipeline',
+      branch: 'master',
+      projectId: 'test-project',
+      project: Object.assign(new Project(), {
+        id: 'test-project',
+        name: 'project',
+      } as Project),
+    } as Pipeline);
+
+    const module: TestingModule = await Test.createTestingModule({
+      providers: [
+        PipelinesService,
+        {
+          provide: getRepositoryToken(Pipeline),
+          useValue: {
+            findOneOrFail: jest.fn().mockImplementation(() => pipeline),
+          },
+        },
+        {
+          provide: getQueueToken(LIST_LOGS_TASK),
+          useValue: {
+            add: jest.fn().mockImplementation(() => ({ id: 1 } as Job)),
+          },
+        },
+      ],
+    }).compile();
+
+    service = module.get<PipelinesService>(PipelinesService);
+    repository = module.get(getRepositoryToken(Pipeline));
+    queue = module.get(getQueueToken(LIST_LOGS_TASK));
+  });
+
+  it('should be defined', () => {
+    expect(service).toBeDefined();
+  });
+
+  describe('listLogsForPipeline', () => {
+    it('should send task to queue.', async () => {
+      const add = jest.spyOn(queue, 'add');
+      await expect(
+        service.listLogsForPipeline('test-pipeline'),
+      ).resolves.toEqual({ id: 1 });
+      expect(add).toBeCalledWith({
+        project: pipeline.project,
+        branch: pipeline.branch,
+      });
+    });
+  });
+});

src/pipelines/pipelines.service.ts

@@ -0,0 +1,55 @@
+import { Injectable } from '@nestjs/common';
+import { InjectRepository } from '@nestjs/typeorm';
+import { Pipeline } from './pipeline.entity';
+import { Repository } from 'typeorm';
+import { BaseDbService } from '../commons/services/base-db.service';
+import { CreatePipelineInput } from './dtos/create-pipeline.input';
+import { UpdatePipelineInput } from './dtos/update-pipeline.input';
+import { ListPipelineArgs } from './dtos/list-pipelines.args';
+import { InjectQueue } from '@nestjs/bull';
+import { LIST_LOGS_TASK } from '../repos/repos.constants';
+import { Queue } from 'bull';
+import { ListLogsOption } from '../repos/models/list-logs.options';
+
+@Injectable()
+export class PipelinesService extends BaseDbService<Pipeline> {
+  readonly uniqueFields: Array<Array<keyof Pipeline>> = [['projectId', 'name']];
+  constructor(
+    @InjectRepository(Pipeline)
+    readonly repository: Repository<Pipeline>,
+    @InjectQueue(LIST_LOGS_TASK)
+    private readonly listLogsQueue: Queue<ListLogsOption>,
+  ) {
+    super(repository);
+  }
+  async list(dto: ListPipelineArgs) {
+    return this.repository.find(dto);
+  }
+
+  async create(dto: CreatePipelineInput) {
+    await this.isDuplicateEntity(dto);
+    return await this.repository.save(this.repository.create(dto));
+  }
+
+  async update(id: string, dto: UpdatePipelineInput) {
+    await this.isDuplicateEntityForUpdate(id, dto);
+    const old = await this.findOne(id);
+    return await this.repository.save(this.repository.merge(old, dto));
+  }
+
+  async remove(id: string) {
+    return (await this.repository.softDelete({ id })).affected;
+  }
+
+  async listLogsForPipeline(id: string) {
+    const pipeline = await this.repository.findOneOrFail({
+      where: { id },
+      relations: ['project'],
+    });
+    const job = await this.listLogsQueue.add({
+      project: pipeline.project,
+      branch: pipeline.branch,
+    });
+    return job;
+  }
+}

src/projects/dtos/create-project.input.ts

@@ -3,6 +3,7 @@ import {
   IsOptional,
   IsString,
   IsUrl,
+  Matches,
   MaxLength,
   MinLength,
 } from 'class-validator';
@@ -19,7 +20,9 @@ export class CreateProjectInput {
   @MinLength(2)
   comment: string;
 
-  @IsUrl({ protocols: ['ssh'] })
+  @Matches(
+    /^(?:ssh:\/\/)?(?:[\w\d-_]+@)(?:[\w\d-_]+\.)*\w{2,10}(?::\d{1,5})?(?:\/[\w\d-_.]+)*/,
+  )
   @MaxLength(256)
   sshUrl: string;
 

src/projects/project.entity.ts

@@ -22,7 +22,4 @@ export class Project extends AppBaseEntity {
 
   @Column({ nullable: true })
   webHookSecret?: string;
-
-  @DeleteDateColumn()
-  deletedAt?: Date;
 }

src/projects/projects.service.ts

@@ -4,6 +4,7 @@ import { BaseDbService } from '../commons/services/base-db.service';
 import { Repository } from 'typeorm';
 import { CreateProjectInput } from './dtos/create-project.input';
 import { Project } from './project.entity';
+import { UpdateProjectInput } from './dtos/update-project.input';
 
 @Injectable()
 export class ProjectsService extends BaseDbService<Project> {
@@ -24,7 +25,7 @@ export class ProjectsService extends BaseDbService<Project> {
     return await this.repository.save(this.repository.create(dto));
   }
 
-  async update(id: string, dto: CreateProjectInput) {
+  async update(id: string, dto: UpdateProjectInput) {
     await this.isDuplicateEntityForUpdate(id, dto);
     const old = await this.findOne(id);
     return await this.repository.save(this.repository.merge(old, dto));

src/repos/dtos/checkout.input.ts

@@ -0,0 +1,16 @@
+import { InputType } from '@nestjs/graphql';
+import { IsOptional, IsString, IsUUID } from 'class-validator';
+
+@InputType()
+export class CheckoutInput {
+  @IsUUID()
+  pipelineId: string;
+
+  @IsString()
+  @IsOptional()
+  branch?: string;
+
+  @IsString()
+  @IsOptional()
+  commitNumber?: string;
+}

src/repos/dtos/list-logs.args.ts

@@ -1,4 +1,4 @@
-import { InputType, ObjectType } from '@nestjs/graphql';
+import { InputType } from '@nestjs/graphql';
 import { IsOptional, IsString, IsUUID } from 'class-validator';
 
 @InputType()

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/models/list-logs.options.ts

@@ -0,0 +1,5 @@
+import { Project } from '../../projects/project.entity';
+export interface ListLogsOption {
+  project: Project;
+  branch: string;
+}

src/repos/repos.constants.ts

@@ -0,0 +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

@@ -4,9 +4,22 @@ import { Project } from '../projects/project.entity';
 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],
-  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.spec.ts

@@ -1,6 +1,7 @@
 import { Test, TestingModule } from '@nestjs/testing';
 import { ReposResolver } from './repos.resolver';
 import { ReposService } from './repos.service';
+import { ProjectsService } from '../projects/projects.service';
 
 describe('ReposResolver', () => {
   let resolver: ReposResolver;
@@ -13,6 +14,10 @@ describe('ReposResolver', () => {
           provide: ReposService,
           useValue: {},
         },
+        {
+          provide: ProjectsService,
+          useValue: {},
+        },
       ],
     }).compile();
 

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 ListBranchesArgs(
-    @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

@@ -1,12 +1,14 @@
+import { Pipeline } from './../pipelines/pipeline.entity';
 import { Test, TestingModule } from '@nestjs/testing';
 import { getRepositoryToken } from '@nestjs/typeorm';
 import { Project } from '../projects/project.entity';
 import { ReposService } from './repos.service';
 import { ConfigModule } from '@nestjs/config';
-import { readFile, rm } from 'fs/promises';
-import { join } from 'path';
+import { rm } from 'fs/promises';
 import configuration from '../commons/config/configuration';
-import { NotFoundException } from '@nestjs/common';
+import { PipelineTask } from '../pipeline-tasks/pipeline-task.entity';
+import { join } from 'path';
+import { readFile } from 'fs/promises';
 
 const getTest1Project = () =>
   ({
@@ -61,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);
   });
@@ -69,63 +74,69 @@ describe('ReposService', () => {
     it('should be return branches', async () => {
       const result = await service.listBranches({ projectId: '1' });
       expect(result).toBeDefined();
-    }, 10_000);
-  });
-  describe('checkoutBranch', () => {
-    it('should be checkout', async () => {
-      await service.checkoutBranch(getTest1Project(), 'master');
-      const filePath = join(
-        service.getWorkspaceRoot(getTest1Project()),
-        'README.md',
-      );
-      const text = await readFile(filePath, { encoding: 'utf-8' });
-      expect(text).toMatch(/Commit 1/gi);
-    }, 30_000);
-    it('multiplexing workspace', async () => {
-      await service.checkoutBranch(getTest1Project(), 'master');
-      await service.checkoutBranch(getTest1Project(), 'branch-a');
-      await service.checkoutBranch(getTest1Project(), 'branch-b');
-      const filePath = join(
-        service.getWorkspaceRoot(getTest1Project()),
-        'branch-b.md',
-      );
-      const text = await readFile(filePath, { encoding: 'utf-8' });
-      expect(text).toMatch(/Commit branch b/gi);
-    }, 30_000);
-    it('nonexistent branch', async () => {
-      return expect(
-        service.checkoutBranch(getTest1Project(), 'nonexistent'),
-      ).rejects.toBeInstanceOf(NotFoundException);
-    }, 30_000);
-    it('checkout the specified version', async () => {
-      await service.checkoutBranch(getTest1Project(), 'master');
-      const filePath = join(
-        service.getWorkspaceRoot(getTest1Project()),
-        'README.md',
-      );
-      const text = await readFile(filePath, { encoding: 'utf-8' });
-      expect(text).toMatch(/Commit 1/gi);
-    }, 30_000);
+    }, 20_000);
   });
 
-  describe('checkoutCommit', () => {
+  describe.skip('checkout', () => {
+    let task: PipelineTask;
+    let workspaceRoot: string;
+    beforeEach(() => {
+      const project = new Project();
+      const pipeline = new Pipeline();
+      task = new PipelineTask();
+      pipeline.project = project;
+      task.pipeline = pipeline;
+      project.id = 'pid';
+      project.name = 'pname';
+      pipeline.id = 'lid';
+      pipeline.name = 'pipeline';
+      task.id = 'tid';
+      task.commit = '123123hash';
+      workspaceRoot = service.getWorkspaceRootByTask(task);
+    });
+
     it('should be checkout', async () => {
-      await service.checkoutCommit(getTest1Project(), '498c782685');
-      const filePath = join(
-        service.getWorkspaceRoot(getTest1Project()),
-        'README.md',
-      );
+      task.commit = '498c782685';
+      await service.checkout(task, workspaceRoot);
+      const filePath = join(workspaceRoot, 'README.md');
       const text = await readFile(filePath, { encoding: 'utf-8' });
       expect(text).toMatch(/Commit 1/gi);
     }, 20_000);
     it('should be checkout right commit', async () => {
-      await service.checkoutCommit(getTest1Project(), '7f7123fe5b');
-      const filePath = join(
-        service.getWorkspaceRoot(getTest1Project()),
-        'README.md',
-      );
+      task.commit = '7f7123fe5b';
+      await service.checkout(task, workspaceRoot);
+      const filePath = join(workspaceRoot, 'README.md');
       const text = await readFile(filePath, { encoding: 'utf-8' });
       expect(text).toMatch(/(?!Commit 1)/gi);
     }, 20_000);
+    it('should be checkout right commit (复用)', async () => {
+      task.commit = '498c782685';
+      await service.checkout(task, workspaceRoot);
+      task.commit = '7f7123fe5b';
+      await service.checkout(task, workspaceRoot);
+      const filePath = join(workspaceRoot, 'README.md');
+      const text = await readFile(filePath, { encoding: 'utf-8' });
+      expect(text).toMatch(/(?!Commit 1)/gi);
+    }, 30_000);
+  });
+
+  describe('getWorkspaceRootByTask', () => {
+    it('should be return right path', () => {
+      const project = new Project();
+      const pipeline = new Pipeline();
+      const task = new PipelineTask();
+      pipeline.project = project;
+      task.pipeline = pipeline;
+      project.id = 'pid';
+      project.name = 'pname';
+      pipeline.id = 'lid';
+      pipeline.name = 'pipeline/\\-名称';
+      task.id = 'tid';
+      task.commit = '123123hash';
+
+      expect(service.getWorkspaceRootByTask(task)).toMatch(
+        /\/pname\/pipeline%2F%5C-%E5%90%8D%E7%A7%B0-123123hash$/,
+      );
+    });
   });
 });

src/repos/repos.service.ts

@@ -1,3 +1,6 @@
+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';
 import { InjectRepository } from '@nestjs/typeorm';
 import { F_OK } from 'constants';
@@ -11,6 +14,7 @@ import { ListLogsArgs } from './dtos/list-logs.args';
 import { ConfigService } from '@nestjs/config';
 
 const DEFAULT_REMOTE_NAME = 'origin';
+const INFO_PATH = '@info';
 @Injectable()
 export class ReposService {
   constructor(
@@ -22,42 +26,33 @@ export class ReposService {
   getWorkspaceRoot(project: Project): string {
     return join(
       this.configService.get<string>('workspaces.root'),
-      project.name,
+      encodeURIComponent(project.name),
+      INFO_PATH,
     );
   }
 
-  async lockWorkspace(workspaceRoot: string) {
-    // TODO:  获取锁，失败抛错。
-  }
+  async getGit(project: Project, workspaceRoot?: string) {
+    if (!workspaceRoot) {
+      workspaceRoot = this.getWorkspaceRoot(project);
+    }
 
-  async getGit(project: Project) {
-    const workspaceRoot = this.getWorkspaceRoot(project);
-    await this.lockWorkspace(workspaceRoot);
-
-    const firstInit = await access(workspaceRoot, F_OK)
-      .then(() => false)
-      .catch(async () => {
-        await mkdir(workspaceRoot);
-        return true;
-      });
+    await access(workspaceRoot, F_OK).catch(async () => {
+      await mkdir(workspaceRoot, { recursive: true });
+    });
     const git = gitP(workspaceRoot);
-    if (firstInit) {
+    if (!(await git.checkIsRepo().catch(() => false))) {
       await git.init();
       await git.addRemote(DEFAULT_REMOTE_NAME, project.sshUrl);
     }
+    await git.fetch();
     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);
-    await git.fetch();
-    return await git.log({
-      '--branches': dto.branch ?? '',
-      '--remotes': DEFAULT_REMOTE_NAME,
-    });
+    return await git.log(
+      branch ? ['--branches', `remotes/origin/${branch}`, '--'] : ['--all'],
+    );
   }
 
   async listBranches(dto: ListBranchesArgs) {
@@ -68,26 +63,8 @@ export class ReposService {
     return git.branch();
   }
 
-  async checkoutBranch(project: Project, branch: string) {
-    const git = await this.getGit(project);
-    try {
-      await git.fetch(DEFAULT_REMOTE_NAME, branch);
-    } catch (err) {
-      if (err.message.includes("couldn't find remote ref nonexistent")) {
-        throw new NotFoundException(err.message);
-      }
-      throw err;
-    }
-    await git.checkout([
-      '-B',
-      branch,
-      '--track',
-      `${DEFAULT_REMOTE_NAME}/${branch}`,
-    ]);
-  }
-
-  async checkoutCommit(project: Project, commitNumber: string) {
-    const git = await this.getGit(project);
+  async checkout(task: PipelineTask, workspaceRoot: string) {
+    const git = await this.getGit(task.pipeline.project, workspaceRoot);
     try {
       await git.fetch(DEFAULT_REMOTE_NAME);
     } catch (err) {
@@ -96,6 +73,20 @@ export class ReposService {
       }
       throw err;
     }
-    await git.checkout([commitNumber]);
+    await git.checkout([task.commit]);
+  }
+
+  /**
+   * get workspace root absolute path
+   *
+   * ! example: `/var/tmp/fennec-workspaces-root/project/pipeline_name-commit_hash`
+   * @param task {PipelineTask} task (with pipeline and project info)
+   */
+  getWorkspaceRootByTask(task: PipelineTask) {
+    return join(
+      this.configService.get<string>('workspaces.root'),
+      encodeURIComponent(task.pipeline.project.name),
+      encodeURIComponent(`${task.pipeline.name}-${task.commit}`),
+    );
   }
 }

test/data/bad-work.js

@@ -0,0 +1,8 @@
+for (let i = 1; i <= 5; i++) {
+  console.log(i * 10);
+}
+console.error('Error Message');
+console.error('Error Message 2');
+console.log('Bye-bye');
+
+process.exit(1);

test/data/one-second-work.js

@@ -0,0 +1,7 @@
+let timer;
+let count = 0;
+setTimeout(() => clearInterval(timer), 1_000);
+
+timer = setInterval(() => {
+  console.log(++count * 10);
+}, 95);