feat(pipeline-task): flush service.

日志记录结束前，任务信息已入库存档。

.vscode/settings.json

@@ -1,5 +1,19 @@
 {
   "cSpell.words": [
-    "Repos"
+    "Repos",
+    "amqp",
+    "boardcat",
+    "errout",
+    "fanout",
+    "gitea",
+    "golevelup",
+    "lpush",
+    "lrange",
+    "metatype",
+    "pmessage",
+    "psubscribe",
+    "rabbitmq",
+    "rpop",
+    "rpush"
   ]
 }

config.yml.example

@@ -1,6 +1,7 @@
 env: dev
 http:
   port: 7122
+
 db:
   postgres:
     host: 192.168.31.194
@@ -8,9 +9,12 @@ db:
     database: fennec
     username: fennec
     password:
-redis:
+  redis:
-  mq:
+    host: 192.168.31.194
-    host: localhost
     port: 6379
+    password:
+    prefix: fennec
+  rabbitmq:
+    uri: 'amqp://fennec:fennec@192.168.31.194:5672'
 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

@@ -10,8 +10,8 @@
     "build": "nest build",
     "format": "prettier --write \"src/**/*.ts\" \"test/**/*.ts\"",
     "start": "nest start",
-    "start:dev": "nest start --watch",
+    "start:dev": "DEBUG=fennec:* nest start --watch",
-    "start:debug": "DEBUG=simple-git,simple-git:* nest start --debug --watch",
+    "start:debug": "DEBUG=fennec:* nest start --debug --watch",
     "start:prod": "node dist/main",
     "lint": "eslint \"{src,apps,libs,test}/**/*.ts\" --fix",
     "test": "jest",
@@ -21,6 +21,7 @@
     "test:e2e": "jest --config ./test/jest-e2e.json"
   },
   "dependencies": {
+    "@golevelup/nestjs-rabbitmq": "^1.16.1",
     "@nestjs/bull": "^0.3.1",
     "@nestjs/common": "^7.5.1",
     "@nestjs/config": "^0.6.2",
@@ -28,19 +29,24 @@
     "@nestjs/graphql": "^7.9.8",
     "@nestjs/platform-express": "^7.5.1",
     "@nestjs/typeorm": "^7.1.5",
-    "@neuralegion/class-sanitizer": "^0.3.2",
     "@types/bull": "^3.15.0",
     "@types/ramda": "^0.27.38",
     "apollo-server-express": "^2.19.2",
     "bcrypt": "^5.0.0",
+    "body-parser": "^1.19.0",
     "bull": "^3.20.1",
     "class-transformer": "^0.3.2",
     "class-validator": "^0.13.1",
+    "debug": "^4.3.1",
     "graphql": "^15.5.0",
     "graphql-tools": "^7.0.2",
+    "ioredis": "^4.25.0",
     "js-yaml": "^4.0.0",
+    "nestjs-pino": "^1.4.0",
     "nestjs-redis": "^1.2.8",
+    "observable-to-async-generator": "^1.0.1-rc",
     "pg": "^8.5.1",
+    "pino-pretty": "^4.7.1",
     "ramda": "^0.27.1",
     "reflect-metadata": "^0.1.13",
     "rimraf": "^3.0.2",
@@ -49,13 +55,17 @@
     "typeorm": "^0.2.30"
   },
   "devDependencies": {
-    "@nestjs/cli": "^7.5.1",
+    "@nestjs/cli": "^7.5.7",
     "@nestjs/schematics": "^7.1.3",
     "@nestjs/testing": "^7.5.1",
+    "@types/body-parser": "^1.19.0",
+    "@types/debug": "^4.1.5",
     "@types/express": "^4.17.8",
+    "@types/ioredis": "^4.22.2",
     "@types/jest": "^26.0.15",
     "@types/js-yaml": "^4.0.0",
     "@types/node": "^14.14.6",
+    "@types/pino-pretty": "^4.7.0",
     "@types/supertest": "^2.0.10",
     "@typescript-eslint/eslint-plugin": "^4.6.1",
     "@typescript-eslint/parser": "^4.6.1",

src/app.module.ts

@@ -1,4 +1,4 @@
-import { Module } from '@nestjs/common';
+import { MiddlewareConsumer, Module, NestModule } from '@nestjs/common';
 import { ConfigModule, ConfigService } from '@nestjs/config';
 import { GraphQLModule } from '@nestjs/graphql';
 import { TypeOrmModule } from '@nestjs/typeorm';
@@ -7,14 +7,41 @@ 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';
+import { WebhooksModule } from './webhooks/webhooks.module';
+import { RawBodyMiddleware } from './commons/middlewares/raw-body.middleware';
+import { GiteaWebhooksController } from './webhooks/gitea-webhooks.controller';
+import { ParseBodyMiddleware } from './commons/middlewares/parse-body.middleware';
 import { BullModule } from '@nestjs/bull';
+import { LoggerModule } from 'nestjs-pino';
+
+import pinoPretty from 'pino-pretty';
 
 @Module({
   imports: [
     ConfigModule.forRoot({
       load: [configuration],
     }),
+    LoggerModule.forRootAsync({
+      imports: [ConfigModule],
+      useFactory: (configService: ConfigService) => {
+        const isDev = configService.get<'dev' | 'prod'>('env') === 'dev';
+        return {
+          pinoHttp: {
+            prettyPrint: isDev
+              ? {
+                  levelFirst: true,
+                }
+              : false,
+            prettifier: pinoPretty,
+          },
+        };
+      },
+      inject: [ConfigService],
+    }),
     TypeOrmModule.forRootAsync({
       imports: [ConfigModule],
       useFactory: (configService: ConfigService) => ({
@@ -35,23 +62,46 @@ import { BullModule } from '@nestjs/bull';
         debug: configService.get<string>('env') !== 'prod',
         playground: true,
         autoSchemaFile: true,
+        installSubscriptionHandlers: true,
       }),
       inject: [ConfigService],
     }),
     BullModule.forRootAsync({
       imports: [ConfigModule],
-      useFactory: async (configService: ConfigService) => ({
+      useFactory: (configService: ConfigService) => ({
         redis: {
-          host: configService.get<string>('redis.mq.host', 'localhost'),
+          host: configService.get<string>('db.redis.host', 'localhost'),
-          port: configService.get<number>('redis.mq.port', 6379),
+          port: configService.get<number>('db.redis.port', undefined),
+          password: configService.get<string>('db.redis.password', undefined),
         },
       }),
       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],
+    }),
+    WebhooksModule,
   ],
   controllers: [AppController],
   providers: [AppService, AppResolver],
 })
-export class AppModule {}
+export class AppModule implements NestModule {
+  public configure(consumer: MiddlewareConsumer): void {
+    consumer
+      .apply(RawBodyMiddleware)
+      .forRoutes(GiteaWebhooksController)
+      .apply(ParseBodyMiddleware)
+      .forRoutes('*');
+  }
+}

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,15 +1,33 @@
 import {
-  ArgumentsHost,
-  Catch,
   ExceptionFilter,
+  Catch,
+  ArgumentsHost,
   HttpException,
   HttpStatus,
 } from '@nestjs/common';
-import { EntityNotFoundError } from 'typeorm/error/EntityNotFoundError';
+import { ApolloError } from 'apollo-server-errors';
 
-@Catch()
+@Catch(HttpException)
-export class AllExceptionsFilter implements ExceptionFilter {
+export class HttpExceptionFilter implements ExceptionFilter {
-  catch(exception: any, host: ArgumentsHost) {
+  catch(exception: HttpException, host: ArgumentsHost) {
+    switch (host.getType<'http' | 'graphql' | string>()) {
+      case 'graphql': {
+        const message = exception.message;
+        const extensions: Record<string, any> = {};
+        const err = exception.getResponse();
+        if (typeof err === 'string') {
+          extensions.message = err;
+        } else {
+          Object.assign(extensions, (err as any).extension);
+          extensions.message = (err as any).message;
+        }
+        return new ApolloError(
+          message,
+          exception.getStatus().toString(),
+          extensions,
+        );
+      }
+      case 'http': {
         const ctx = host.switchToHttp();
         const response = ctx.getResponse();
         const request = ctx.getRequest();
@@ -19,37 +37,15 @@ export class AllExceptionsFilter implements ExceptionFilter {
             ? exception.getStatus()
             : HttpStatus.INTERNAL_SERVER_ERROR;
 
-    if (exception instanceof HttpException) {
-      const ex = exception.getResponse();
-      if (ex instanceof Object) {
         response.status(status).json({
-          ...ex,
+          statusCode: status,
-          timestamp: Date.now(),
+          message: exception.message,
-          path: request.url,
+          timestamp: new Date().toISOString(),
-        });
-      } else {
-        response.status(status).json({
-          message: ex,
-          timestamp: Date.now(),
           path: request.url,
         });
       }
-    } else if (exception instanceof EntityNotFoundError) {
+      default:
-      response.status(HttpStatus.NOT_FOUND).json({
+        throw exception;
-        message: '资源未找到!',
-        timestamp: Date.now(),
-        path: request.url,
-      });
-    } 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,
-      });
     }
   }
 }

src/commons/middlewares/parse-body.middleware.spec.ts

@@ -0,0 +1,7 @@
+import { ParseBodyMiddleware } from './parse-body.middleware';
+
+describe('ParseBodyMiddleware', () => {
+  it('should be defined', () => {
+    expect(new ParseBodyMiddleware()).toBeDefined();
+  });
+});

src/commons/middlewares/parse-body.middleware.ts

@@ -0,0 +1,13 @@
+import { Injectable, NestMiddleware } from '@nestjs/common';
+import { json, urlencoded, text } from 'body-parser';
+import { Request, Response, NextFunction } from 'express';
+
+@Injectable()
+export class ParseBodyMiddleware implements NestMiddleware {
+  use(req: Request, res: Response, next: NextFunction) {
+    json()(req, res, () =>
+      urlencoded()(req, res, () => text()(req, res, next)),
+    );
+    // next();
+  }
+}

src/commons/middlewares/raw-body.middleware.spec.ts

@@ -0,0 +1,7 @@
+import { RawBodyMiddleware } from './raw-body.middleware';
+
+describe('RawBodyMiddleware', () => {
+  it('should be defined', () => {
+    expect(new RawBodyMiddleware()).toBeDefined();
+  });
+});

src/commons/middlewares/raw-body.middleware.ts

@@ -0,0 +1,10 @@
+import { Injectable, NestMiddleware } from '@nestjs/common';
+import { raw } from 'body-parser';
+import { Request, Response, NextFunction } from 'express';
+
+@Injectable()
+export class RawBodyMiddleware implements NestMiddleware {
+  use(req: Request, res: Response, next: NextFunction) {
+    raw({ type: '*/*' })(req, res, next);
+  }
+}

src/commons/pipes/sanitize.pipe.ts

@@ -0,0 +1,25 @@
+import { ArgumentMetadata, Injectable, PipeTransform } from '@nestjs/common';
+import { plainToClass } from 'class-transformer';
+
+@Injectable()
+export class SanitizePipe implements PipeTransform {
+  transform(value: any, metadata: ArgumentMetadata) {
+    if (
+      !(value instanceof Object) ||
+      value instanceof Buffer ||
+      value instanceof Array
+    ) {
+      return value;
+    }
+    const constructorFunction = metadata.metatype;
+    if (!constructorFunction || value instanceof constructorFunction) {
+      return value;
+    }
+    try {
+      return plainToClass(constructorFunction, value);
+    } catch (err) {
+      console.error(err);
+      throw err;
+    }
+  }
+}

src/commons/utils/rabbit-mq.ts

@@ -0,0 +1,21 @@
+import { hostname } from 'os';
+
+export function getInstanceName() {
+  return hostname();
+}
+
+export function getSelfInstanceRouteKey(key: string) {
+  return getAppInstanceRouteKey(key, getInstanceName());
+}
+
+export function getAppInstanceRouteKey(key: string, appInstance?: string) {
+  return appInstance ? `${key}.${appInstance}` : key;
+}
+
+export function getSelfInstanceQueueKey(key: string) {
+  return getAppInstanceQueueKey(key, getInstanceName());
+}
+
+export function getAppInstanceQueueKey(key: string, appInstance?: string) {
+  return appInstance ? `${key}.${appInstance}` : key;
+}

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 app = await NestFactory.create(AppModule, { bodyParser: false });
   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,15 @@
+import { registerEnumType } from '@nestjs/graphql';
+
+export enum TaskStatuses {
+  success = 'success',
+  failed = 'failed',
+  working = 'working',
+  pending = 'pending',
+}
+
+registerEnumType(TaskStatuses, {
+  name: 'TaskStatuses',
+  description: '任务状态',
+});
+
+export const terminalTaskStatuses = [TaskStatuses.success, TaskStatuses.failed];

src/pipeline-tasks/models/pipeline-task-event.ts

@@ -0,0 +1,25 @@
+import { Field, ObjectType } from '@nestjs/graphql';
+import { PipelineUnits } from '../enums/pipeline-units.enum';
+import { TaskStatuses } from '../enums/task-statuses.enum';
+import { Type } from 'class-transformer';
+
+@ObjectType()
+export class PipelineTaskEvent {
+  @Field()
+  taskId: string;
+  @Field()
+  pipelineId: string;
+  @Field()
+  projectId: string;
+  @Field(() => PipelineUnits, { nullable: true })
+  unit: PipelineUnits | null;
+  @Field()
+  @Type(() => Date)
+  emittedAt: Date;
+  @Field()
+  message: string;
+  @Field()
+  messageType: 'stdout' | 'stderr' | 'stdin';
+  @Field(() => TaskStatuses)
+  status: TaskStatuses;
+}

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

@@ -0,0 +1,34 @@
+import { PipelineTask } from './../pipeline-task.entity';
+import { PipelineUnits } from '../enums/pipeline-units.enum';
+import { Field, HideField, ObjectType } from '@nestjs/graphql';
+import { Type } from 'class-transformer';
+
+@ObjectType()
+export class PipelineTaskLogMessage {
+  @HideField()
+  task: PipelineTask;
+  @Field(() => PipelineUnits, { nullable: true })
+  unit?: PipelineUnits;
+  @Field()
+  @Type(() => Date)
+  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,17 @@
+import { TaskStatuses } from '../enums/task-statuses.enum';
+import { PipelineUnits } from '../enums/pipeline-units.enum';
+import { Field, ObjectType } from '@nestjs/graphql';
+import { Type } from 'class-transformer';
+
+@ObjectType()
+export class PipelineTaskLogs {
+  @Field(() => PipelineUnits)
+  unit: PipelineUnits;
+  @Field(() => TaskStatuses)
+  status: TaskStatuses;
+  @Type(() => Date)
+  startedAt?: Date;
+  @Type(() => 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-flush.service.spec.ts

@@ -0,0 +1,80 @@
+import { AmqpConnection } from '@golevelup/nestjs-rabbitmq';
+import { Test, TestingModule } from '@nestjs/testing';
+import { RedisService } from 'nestjs-redis';
+import { PipelineTaskFlushService } from './pipeline-task-flush.service';
+import { PipelineTaskEvent } from './models/pipeline-task-event';
+import { TaskStatuses } from './enums/task-statuses.enum';
+import {
+  EXCHANGE_PIPELINE_TASK_TOPIC,
+  ROUTE_PIPELINE_TASK_DONE,
+} from './pipeline-tasks.constants';
+
+describe('PipelineTaskFlushService', () => {
+  let service: PipelineTaskFlushService;
+  let redisService: RedisService;
+  let amqpConnection: AmqpConnection;
+
+  beforeEach(async () => {
+    const redisClient = {
+      rpush: jest.fn(() => Promise.resolve()),
+      lrange: jest.fn(() => Promise.resolve()),
+    };
+    const module: TestingModule = await Test.createTestingModule({
+      providers: [
+        PipelineTaskFlushService,
+        {
+          provide: RedisService,
+          useValue: {
+            getClient() {
+              return redisClient;
+            },
+          },
+        },
+        {
+          provide: AmqpConnection,
+          useValue: {
+            request: jest.fn(() => Promise.resolve()),
+          },
+        },
+      ],
+    }).compile();
+
+    service = module.get<PipelineTaskFlushService>(PipelineTaskFlushService);
+    redisService = module.get<RedisService>(RedisService);
+    amqpConnection = module.get<AmqpConnection>(AmqpConnection);
+  });
+
+  it('should be defined', () => {
+    expect(service).toBeDefined();
+  });
+
+  describe('write', () => {
+    it('normal', async () => {
+      const testEvent = new PipelineTaskEvent();
+      testEvent.taskId = 'test';
+      testEvent.status = TaskStatuses.working;
+      const rpush = jest.spyOn(redisService.getClient(), 'rpush');
+      const request = jest.spyOn(amqpConnection, 'request');
+      await service.write(testEvent);
+      expect(rpush).toBeCalledTimes(1);
+      expect(rpush.mock.calls[0][0]).toEqual('p-task:log:test');
+      expect(rpush.mock.calls[0][1]).toEqual(JSON.stringify(testEvent));
+      expect(request).toBeCalledTimes(0);
+    });
+    it('event for which task done', async () => {
+      const testEvent = new PipelineTaskEvent();
+      testEvent.taskId = 'test';
+      testEvent.status = TaskStatuses.success;
+      const rpush = jest.spyOn(redisService.getClient(), 'rpush');
+      const request = jest.spyOn(amqpConnection, 'request');
+      await service.write(testEvent);
+      expect(rpush).toBeCalledTimes(1);
+      expect(request).toBeCalledTimes(1);
+      expect(request.mock.calls[0][0]).toMatchObject({
+        exchange: EXCHANGE_PIPELINE_TASK_TOPIC,
+        routingKey: ROUTE_PIPELINE_TASK_DONE,
+        payload: { taskId: 'test', status: TaskStatuses.success },
+      });
+    });
+  });
+});

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

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

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

@@ -0,0 +1,39 @@
+import { AppBaseEntity } from './../commons/entities/app-base-entity';
+import { ObjectType } from '@nestjs/graphql';
+import { Column, Entity, ManyToOne, ValueTransformer } 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';
+import { plainToClass } from 'class-transformer';
+
+const logsTransformer: ValueTransformer = {
+  from: (value) => plainToClass(PipelineTaskLogs, value),
+  to: (value) => value,
+};
+@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: '[]', transformer: logsTransformer })
+  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-task.logger.spec.ts

@@ -0,0 +1,74 @@
+import { Test, TestingModule } from '@nestjs/testing';
+import { PipelineTaskLogger } from './pipeline-task.logger';
+import { PipelineTaskEvent } from './models/pipeline-task-event';
+import { take, timeout } from 'rxjs/operators';
+
+describe('PipelineTaskRunner', () => {
+  let logger: PipelineTaskLogger;
+  let module: TestingModule;
+
+  beforeEach(async () => {
+    module = await Test.createTestingModule({
+      providers: [PipelineTaskLogger],
+    }).compile();
+
+    logger = module.get(PipelineTaskLogger);
+  });
+
+  it('should be defined', () => {
+    expect(logger).toBeDefined();
+  });
+
+  describe('getMessage$', () => {
+    it('normal', async () => {
+      const event = new PipelineTaskEvent();
+      event.taskId = 'test';
+      const emittedAt = new Date();
+      event.emittedAt = emittedAt.toISOString() as any;
+      const message$ = logger.getMessage$('test');
+
+      let receiveEvent;
+      message$.pipe(take(1)).subscribe((value) => (receiveEvent = value));
+      await logger.handleEvent(event);
+      expect(receiveEvent).toMatchObject({
+        ...event,
+        emittedAt,
+      });
+    });
+    it('no match', async () => {
+      const event = new PipelineTaskEvent();
+      event.taskId = 'test';
+      const message$ = logger.getMessage$('other');
+      setTimeout(() => {
+        logger.handleEvent(event);
+      });
+      expect(message$.pipe(take(1), timeout(100)).toPromise()).rejects.toMatch(
+        'timeout',
+      );
+    });
+    it('multiple subscribers', async () => {
+      const event = new PipelineTaskEvent();
+      event.taskId = 'test';
+      const message$ = logger.getMessage$('test');
+      const message2$ = logger.getMessage$('test');
+      setTimeout(() => {
+        logger.handleEvent(event);
+      });
+      expect(message$.pipe(take(1), timeout(100)).toPromise()).resolves.toEqual(
+        event,
+      );
+      expect(
+        message2$.pipe(take(1), timeout(100)).toPromise(),
+      ).resolves.toEqual(event);
+    });
+  });
+
+  describe('onModuleDestroy', () => {
+    it('complete observable when destroying module', async () => {
+      logger.onModuleDestroy();
+      await expect(
+        (logger as any).message$.toPromise(),
+      ).resolves.toBeUndefined();
+    });
+  });
+});

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

@@ -0,0 +1,37 @@
+import { RabbitSubscribe } from '@golevelup/nestjs-rabbitmq';
+import { Injectable, OnModuleDestroy } from '@nestjs/common';
+import { plainToClass } from 'class-transformer';
+import { Observable, Subject } from 'rxjs';
+import { filter } from 'rxjs/operators';
+import { PipelineTaskEvent } from './models/pipeline-task-event';
+import {
+  EXCHANGE_PIPELINE_TASK_FANOUT,
+  QUEUE_HANDLE_PIPELINE_TASK_LOG_EVENT,
+  ROUTE_PIPELINE_TASK_LOG,
+} from './pipeline-tasks.constants';
+
+@Injectable()
+export class PipelineTaskLogger implements OnModuleDestroy {
+  private readonly messageSubject = new Subject<PipelineTaskEvent>();
+  private readonly message$: Observable<PipelineTaskEvent> = this.messageSubject.pipe();
+
+  @RabbitSubscribe({
+    exchange: EXCHANGE_PIPELINE_TASK_FANOUT,
+    routingKey: ROUTE_PIPELINE_TASK_LOG,
+    queue: QUEUE_HANDLE_PIPELINE_TASK_LOG_EVENT,
+    queueOptions: {
+      autoDelete: true,
+    },
+  })
+  async handleEvent(message: PipelineTaskEvent) {
+    this.messageSubject.next(plainToClass(PipelineTaskEvent, message));
+  }
+
+  getMessage$(taskId: string) {
+    return this.message$.pipe(filter((event) => event.taskId === taskId));
+  }
+
+  onModuleDestroy() {
+    this.messageSubject.complete();
+  }
+}

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

@@ -0,0 +1,326 @@
+import { Test, TestingModule } from '@nestjs/testing';
+import { ReposService } from '../repos/repos.service';
+import { PipelineUnits } from './enums/pipeline-units.enum';
+import { PipelineTask } from './pipeline-task.entity';
+import { Pipeline } from '../pipelines/pipeline.entity';
+import { Project } from '../projects/project.entity';
+import { TaskStatuses } from './enums/task-statuses.enum';
+import { getLoggerToken, PinoLogger } from 'nestjs-pino';
+import { PipelineTaskRunner } from './pipeline-task.runner';
+import { WorkUnitMetadata } from './models/work-unit-metadata.model';
+import { AmqpConnection } from '@golevelup/nestjs-rabbitmq';
+describe('PipelineTaskRunner', () => {
+  let runner: PipelineTaskRunner;
+  let reposService: ReposService;
+
+  beforeEach(async () => {
+    const module: TestingModule = await Test.createTestingModule({
+      providers: [
+        {
+          provide: ReposService,
+          useValue: {
+            getWorkspaceRootByTask: () => 'workspace-root',
+            checkout: async () => undefined,
+          },
+        },
+        {
+          provide: getLoggerToken(PipelineTaskRunner.name),
+          useValue: new PinoLogger({}),
+        },
+        {
+          provide: 'spawn',
+          useValue: () => undefined,
+        },
+        PipelineTaskRunner,
+        {
+          provide: AmqpConnection,
+          useValue: {},
+        },
+      ],
+    }).compile();
+
+    reposService = module.get(ReposService);
+    runner = module.get(PipelineTaskRunner);
+  });
+
+  it('should be defined', () => {
+    expect(runner).toBeDefined();
+  });
+
+  it('onNewTask', async () => {
+    const task = new PipelineTask();
+    let tmpTask;
+    const doTask = jest
+      .spyOn(runner, 'doTask')
+      .mockImplementation(async (task) => {
+        tmpTask = task;
+      });
+    await runner.onNewTask(task);
+    expect(tmpTask).toEqual(task);
+    expect(doTask).toBeCalledTimes(1);
+  });
+
+  describe('test biz', () => {
+    let emitEvent: jest.SpyInstance;
+    beforeEach(() => {
+      emitEvent = jest
+        .spyOn(runner, 'emitEvent')
+        .mockImplementation((..._) => Promise.resolve());
+    });
+
+    describe('doTask', () => {
+      let checkout: jest.SpyInstance;
+      let doTaskUnit: jest.SpyInstance;
+
+      beforeEach(() => {
+        checkout = jest
+          .spyOn(runner, 'checkout')
+          .mockImplementation((..._) => Promise.resolve('/null'));
+        doTaskUnit = jest
+          .spyOn(runner, 'doTaskUnit')
+          .mockImplementation((..._) => Promise.resolve());
+      });
+
+      it('only checkout', async () => {
+        const task = new PipelineTask();
+        (task.id = 'taskId'), (task.pipeline = new Pipeline());
+        task.units = [PipelineUnits.checkout];
+        task.pipeline.id = 'pipelineId';
+        task.pipeline.project = new Project();
+        task.pipeline.project.id = 'projectId';
+        task.pipeline.workUnitMetadata = new WorkUnitMetadata();
+        task.pipeline.workUnitMetadata.version = 1;
+        task.pipeline.workUnitMetadata.units = [
+          {
+            type: PipelineUnits.checkout,
+            scripts: [],
+          },
+        ];
+
+        await runner.doTask(task);
+
+        expect(checkout).toBeCalledTimes(1);
+        expect(doTaskUnit).toBeCalledTimes(0);
+        expect(emitEvent).toBeCalledTimes(2);
+        expect(emitEvent.mock.calls[0][0]).toMatchObject(task);
+        expect(emitEvent.mock.calls[0][1]).toBeNull();
+        expect(emitEvent.mock.calls[0][2]).toEqual(TaskStatuses.working);
+        expect(emitEvent.mock.calls[1][0]).toMatchObject(task);
+        expect(emitEvent.mock.calls[1][1]).toBeNull();
+        expect(emitEvent.mock.calls[1][2]).toEqual(TaskStatuses.success);
+      });
+
+      it('many units', async () => {
+        const task = new PipelineTask();
+        (task.id = 'taskId'), (task.pipeline = new Pipeline());
+        task.units = [
+          PipelineUnits.checkout,
+          PipelineUnits.test,
+          PipelineUnits.deploy,
+        ];
+        task.pipeline.id = 'pipelineId';
+        task.pipeline.project = new Project();
+        task.pipeline.project.id = 'projectId';
+        task.pipeline.workUnitMetadata = new WorkUnitMetadata();
+        task.pipeline.workUnitMetadata.version = 1;
+        task.pipeline.workUnitMetadata.units = [
+          {
+            type: PipelineUnits.checkout,
+            scripts: [],
+          },
+          {
+            type: PipelineUnits.installDependencies,
+            scripts: ['pwd'],
+          },
+          {
+            type: PipelineUnits.test,
+            scripts: ['pwd'],
+          },
+          {
+            type: PipelineUnits.deploy,
+            scripts: ['pwd', 'uname'],
+          },
+        ];
+
+        await runner.doTask(task);
+
+        expect(checkout).toBeCalledTimes(1);
+        expect(doTaskUnit).toBeCalledTimes(2);
+        expect(emitEvent).toBeCalledTimes(2);
+      });
+
+      it('unit work failed', async () => {
+        const task = new PipelineTask();
+        (task.id = 'taskId'), (task.pipeline = new Pipeline());
+        task.units = [PipelineUnits.checkout, PipelineUnits.test];
+        task.pipeline.id = 'pipelineId';
+        task.pipeline.project = new Project();
+        task.pipeline.project.id = 'projectId';
+        task.pipeline.workUnitMetadata = new WorkUnitMetadata();
+        task.pipeline.workUnitMetadata.version = 1;
+        task.pipeline.workUnitMetadata.units = [
+          {
+            type: PipelineUnits.checkout,
+            scripts: [],
+          },
+          {
+            type: PipelineUnits.test,
+            scripts: ['pwd'],
+          },
+        ];
+
+        doTaskUnit = jest
+          .spyOn(runner, 'doTaskUnit')
+          .mockImplementation((..._) =>
+            Promise.reject(new Error('test error')),
+          );
+        await runner.doTask(task);
+
+        expect(checkout).toBeCalledTimes(1);
+        expect(doTaskUnit).toBeCalledTimes(1);
+        expect(emitEvent).toBeCalledTimes(2);
+        expect(emitEvent.mock.calls[1][0]).toMatchObject(task);
+        expect(emitEvent.mock.calls[1][1]).toBeNull();
+        expect(emitEvent.mock.calls[1][2]).toEqual(TaskStatuses.failed);
+      });
+    });
+
+    describe('doTaskUnit', () => {
+      it('success', async () => {
+        const runScript = jest
+          .spyOn(runner, 'runScript')
+          .mockImplementation((..._) => Promise.resolve());
+        const task = new PipelineTask();
+
+        const unit = PipelineUnits.test;
+        const workspacePath = '/null';
+        await runner.doTaskUnit(unit, ['pwd'], task, workspacePath);
+
+        expect(emitEvent.mock.calls[0][0]).toEqual(task);
+        expect(emitEvent.mock.calls[0][1]).toEqual(unit);
+        expect(emitEvent.mock.calls[0][2]).toEqual(TaskStatuses.working);
+        expect(emitEvent.mock.calls[1][0]).toEqual(task);
+        expect(emitEvent.mock.calls[1][1]).toEqual(unit);
+        expect(emitEvent.mock.calls[1][2]).toEqual(TaskStatuses.success);
+        expect(runScript.mock.calls[0][0]).toEqual('pwd');
+        expect(runScript.mock.calls[0][1]).toEqual(workspacePath);
+        expect(runScript.mock.calls[0][2]).toEqual(task);
+        expect(runScript.mock.calls[0][3]).toEqual(unit);
+      });
+      it('failed', async () => {
+        const runScript = jest
+          .spyOn(runner, 'runScript')
+          .mockImplementation((..._) =>
+            Promise.reject(new Error('test error')),
+          );
+        const task = new PipelineTask();
+
+        const unit = PipelineUnits.test;
+        const workspacePath = '/null';
+        await expect(
+          runner.doTaskUnit(unit, ['pwd'], task, workspacePath),
+        ).rejects.toThrow('test error');
+
+        expect(emitEvent.mock.calls[1]?.[0]).toEqual(task);
+        expect(emitEvent.mock.calls[1]?.[1]).toEqual(unit);
+        expect(emitEvent.mock.calls[1]?.[2]).toEqual(TaskStatuses.failed);
+        expect(runScript).toBeCalledTimes(1);
+      });
+    });
+
+    describe('runScript', () => {
+      it('normal', async () => {
+        const spawn = jest.fn((..._: any[]) => ({
+          stdout: {
+            on: () => undefined,
+          },
+          stderr: {
+            on: () => undefined,
+          },
+          addListener: (_: any, fn: (code: number) => void) => {
+            fn(0);
+          },
+        }));
+        (runner as any).spawn = spawn;
+
+        const task = new PipelineTask();
+        task.id = 'taskId';
+        const unit = PipelineUnits.deploy;
+
+        await runner.runScript('script name', 'workspaceRoot', task, unit);
+        expect(spawn).toHaveBeenCalledTimes(1);
+        expect(spawn.mock.calls[0][0]).toEqual('script name');
+        expect(spawn.mock.calls[0][1]).toMatchObject({
+          shell: true,
+          cwd: 'workspaceRoot',
+        });
+      });
+      it('failed', async () => {
+        const spawn = jest.fn((..._: any[]) => ({
+          stdout: {
+            on: () => undefined,
+          },
+          stderr: {
+            on: () => undefined,
+          },
+          addListener: (_: any, fn: (code: number) => void) => {
+            fn(1);
+          },
+        }));
+        (runner as any).spawn = spawn;
+
+        const task = new PipelineTask();
+        task.id = 'taskId';
+        const unit = PipelineUnits.deploy;
+
+        expect(
+          runner.runScript('script name', 'workspaceRoot', task, unit),
+        ).rejects.toThrowError();
+      });
+      it('wait emit message done', async () => {
+        let finishedFn: () => void;
+        const on = jest.fn((_: any, fn: (buff: Buffer) => void) => {
+          setTimeout(() => {
+            fn(Buffer.from('message 1'));
+            setTimeout(() => {
+              fn(Buffer.from('message 2'));
+              setTimeout(() => {
+                fn(Buffer.from('message 3'));
+                finishedFn();
+              }, 1000);
+            }, 10);
+          }, 10);
+        });
+        const spawn = jest.fn((..._: any[]) => ({
+          stdout: {
+            on,
+          },
+          stderr: {
+            on,
+          },
+          addListener: (_: any, fn: (code: number) => void) => {
+            finishedFn = () => fn(0);
+          },
+        }));
+
+        let emitSuccessCount = 0;
+        jest.spyOn(runner, 'emitEvent').mockImplementation((..._: any[]) => {
+          return new Promise((resolve) => {
+            setTimeout(() => {
+              emitSuccessCount++;
+              resolve();
+            }, 1000);
+          });
+        });
+        (runner as any).spawn = spawn;
+
+        const task = new PipelineTask();
+        task.id = 'taskId';
+        const unit = PipelineUnits.deploy;
+
+        await runner.runScript('script name', 'workspaceRoot', task, unit);
+        expect(emitSuccessCount).toEqual(1 + 6);
+      });
+    });
+  });
+});

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

@@ -0,0 +1,267 @@
+import { ReposService } from '../repos/repos.service';
+import { spawn, ChildProcessWithoutNullStreams } from 'child_process';
+import { PipelineTask } from './pipeline-task.entity';
+import { ApplicationException } from '../commons/exceptions/application.exception';
+import { PipelineUnits } from './enums/pipeline-units.enum';
+import { TaskStatuses } from './enums/task-statuses.enum';
+import { InjectPinoLogger, PinoLogger } from 'nestjs-pino';
+import { AmqpConnection, RabbitSubscribe } from '@golevelup/nestjs-rabbitmq';
+import { PipelineTaskEvent } from './models/pipeline-task-event';
+import { last } from 'ramda';
+import { Inject } from '@nestjs/common';
+import {
+  EXCHANGE_PIPELINE_TASK_FANOUT,
+  ROUTE_PIPELINE_TASK_LOG,
+} from './pipeline-tasks.constants';
+
+type Spawn = typeof spawn;
+
+export class PipelineTaskRunner {
+  readonly processes = new Map<string, ChildProcessWithoutNullStreams>();
+
+  constructor(
+    private readonly reposService: ReposService,
+    @InjectPinoLogger(PipelineTaskRunner.name)
+    private readonly logger: PinoLogger,
+    @Inject('spawn')
+    private readonly spawn: Spawn,
+    private readonly amqpConnection: AmqpConnection,
+  ) {}
+  @RabbitSubscribe({
+    exchange: 'new-pipeline-task',
+    routingKey: 'mac',
+    queue: 'mac.new-pipeline-task',
+  })
+  async onNewTask(task: PipelineTask) {
+    this.logger.info({ task }, 'on new task [%s].', task.id);
+    try {
+      await this.doTask(task);
+    } catch (err) {
+      this.logger.error({ task, err }, err.message);
+    }
+  }
+  @RabbitSubscribe({
+    exchange: 'stop-pipeline-task',
+    routingKey: 'mac',
+    queue: 'mac.stop-pipeline-task',
+  })
+  async onStopTask(task: PipelineTask) {
+    this.logger.info({ task }, 'on stop task [%s].', task.id);
+    const process = this.processes.get(task.id);
+    if (process) {
+      this.logger.info({ task }, 'send signal SIGINT to child process.');
+      process.kill('SIGINT');
+
+      setTimeout(() => {
+        if (process === this.processes.get(task.id)) {
+          this.logger.info({ task }, 'send signal SIGKILL to child process.');
+          process.kill('SIGKILL');
+          return;
+        }
+        if (this.processes.has(task.id)) {
+          this.logger.error(
+            { task },
+            'this pipeline task not stop yet. there is a new process running, maybe is a bug about error capture',
+          );
+        }
+      }, 10_000);
+    } else {
+      this.logger.info({ task }, 'child process is not running.');
+    }
+  }
+
+  async doTask(task: PipelineTask) {
+    if (task.pipeline.workUnitMetadata.version !== 1) {
+      throw new ApplicationException(
+        'work unit metadata version is not match.',
+      );
+    }
+    await this.emitEvent(
+      task,
+      null,
+      TaskStatuses.working,
+      `[start task]`,
+      'stdout',
+    );
+
+    this.logger.info('running task [%s].', task.id);
+    try {
+      const workspaceRoot = await this.checkout(task);
+      const units = task.units
+        .filter((unit) => unit !== PipelineUnits.checkout)
+        .map(
+          (type) =>
+            task.pipeline.workUnitMetadata.units.find(
+              (unit) => unit.type === type,
+            ) ?? { type: type, scripts: [] },
+        );
+      this.logger.info({ units }, 'begin run units.');
+      for (const unit of units) {
+        await this.doTaskUnit(unit.type, unit.scripts, task, workspaceRoot);
+      }
+      await this.emitEvent(
+        task,
+        null,
+        TaskStatuses.success,
+        `[finished task] success`,
+        'stdout',
+      );
+      this.logger.info({ task }, 'task [%s] completed.', task.id);
+    } catch (err) {
+      await this.emitEvent(
+        task,
+        null,
+        TaskStatuses.failed,
+        `[finished unit] ${err.message}`,
+        'stderr',
+      );
+      this.logger.error({ task, error: err }, 'task [%s] failed.', task.id);
+    } finally {
+    }
+  }
+
+  async doTaskUnit(
+    unit: PipelineUnits,
+    scripts: string[],
+    task: PipelineTask,
+    workspaceRoot: string,
+  ) {
+    await this.emitEvent(
+      task,
+      unit,
+      TaskStatuses.working,
+      `[begin unit] ${unit}`,
+      'stdin',
+    );
+    this.logger.info({ task }, 'curr unit is %s', unit);
+    try {
+      for (const script of scripts) {
+        this.logger.debug('begin runScript %s', script);
+        await this.runScript(script, workspaceRoot, task, unit);
+        this.logger.debug('end runScript %s', script);
+      }
+
+      await this.emitEvent(
+        task,
+        unit,
+        TaskStatuses.success,
+        `[finished unit] ${unit}`,
+        'stdout',
+      );
+    } catch (err) {
+      await this.emitEvent(
+        task,
+        unit,
+        TaskStatuses.failed,
+        `[finished unit] ${err.message}`,
+        'stderr',
+      );
+      throw err;
+    }
+  }
+
+  async checkout(task: PipelineTask) {
+    await this.emitEvent(
+      task,
+      PipelineUnits.checkout,
+      TaskStatuses.working,
+      '[begin unit] checkout',
+      'stdin',
+    );
+    try {
+      const path = await this.reposService.checkout4Task(task);
+      await this.emitEvent(
+        task,
+        PipelineUnits.checkout,
+        TaskStatuses.success,
+        'checkout success.',
+        'stdout',
+      );
+      return path;
+    } catch (err) {
+      await this.emitEvent(
+        task,
+        PipelineUnits.checkout,
+        TaskStatuses.failed,
+        'checkout failed.',
+        'stderr',
+      );
+    }
+  }
+
+  async emitEvent(
+    task: PipelineTask,
+    unit: PipelineUnits | null,
+    status: TaskStatuses,
+    message: string,
+    messageType: 'stderr' | 'stdout' | 'stdin',
+  ) {
+    const event: PipelineTaskEvent = {
+      taskId: task.id,
+      pipelineId: task.pipeline.id,
+      projectId: task.pipeline.project.id,
+      unit,
+      emittedAt: new Date(),
+      message: last(message) === '\n' ? message : message + '\n',
+      messageType,
+      status,
+    };
+    this.amqpConnection
+      .publish(EXCHANGE_PIPELINE_TASK_FANOUT, ROUTE_PIPELINE_TASK_LOG, event)
+      .catch((error) => {
+        this.logger.error(
+          { error, event },
+          'send event message to queue failed. %s',
+          error.message,
+        );
+      });
+  }
+
+  async runScript(
+    script: string,
+    workspaceRoot: string,
+    task: PipelineTask,
+    unit: PipelineUnits,
+  ): Promise<void> {
+    await this.emitEvent(task, unit, TaskStatuses.working, script, 'stdin');
+    return new Promise((resolve, reject) => {
+      const sub = this.spawn(script, {
+        shell: true,
+        cwd: workspaceRoot,
+      });
+      this.processes.set(task.id, sub);
+      let loggingCount = 0; // semaphore
+
+      sub.stderr.on('data', (data: Buffer) => {
+        const str = data.toString();
+        loggingCount++;
+
+        this.emitEvent(task, unit, TaskStatuses.working, str, 'stdout').finally(
+          () => loggingCount--,
+        );
+      });
+      sub.stdout.on('data', (data: Buffer) => {
+        const str = data.toString();
+        loggingCount++;
+
+        this.emitEvent(task, unit, TaskStatuses.working, str, 'stderr').finally(
+          () => loggingCount--,
+        );
+      });
+      sub.addListener('close', async (code) => {
+        this.processes.delete(task.id);
+        await new Promise<void>(async (resolve) => {
+          for (let i = 0; i < 10 && loggingCount > 0; i++) {
+            await new Promise((resolve) => setTimeout(resolve, 500));
+            this.logger.debug('waiting logging... (%dx500ms)', i);
+          }
+          resolve();
+        });
+        if (code === 0) {
+          return resolve();
+        }
+        return reject(new ApplicationException('exec script failed'));
+      });
+    });
+  }
+}

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

@@ -0,0 +1,6 @@
+export const EXCHANGE_PIPELINE_TASK_TOPIC = 'pipeline-task.topic';
+export const EXCHANGE_PIPELINE_TASK_FANOUT = 'pipeline-task.fanout';
+export const ROUTE_PIPELINE_TASK_LOG = 'pipeline-task-log';
+export const QUEUE_HANDLE_PIPELINE_TASK_LOG_EVENT = 'pipeline-task-log';
+export const QUEUE_WRITE_PIPELINE_TASK_LOG = 'write-pipeline-task-log';
+export const ROUTE_PIPELINE_TASK_DONE = 'pipeline-task-done';

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

@@ -0,0 +1,89 @@
+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 { RabbitMQModule } from '@golevelup/nestjs-rabbitmq';
+import { ConfigModule, ConfigService } from '@nestjs/config';
+import { PipelineTaskRunner } from './pipeline-task.runner';
+import { spawn } from 'child_process';
+import {
+  EXCHANGE_PIPELINE_TASK_FANOUT,
+  EXCHANGE_PIPELINE_TASK_TOPIC,
+} from './pipeline-tasks.constants';
+import { PipelineTaskLogger } from './pipeline-task.logger';
+import { PipelineTaskFlushService } from './pipeline-task-flush.service';
+
+@Module({
+  imports: [
+    TypeOrmModule.forFeature([PipelineTask, Pipeline]),
+    RedisModule,
+    ReposModule,
+
+    RabbitMQModule.forRootAsync(RabbitMQModule, {
+      imports: [ConfigModule],
+      useFactory: (configService: ConfigService) => ({
+        uri: configService.get<string>('db.rabbitmq.uri'),
+        exchanges: [
+          {
+            name: 'new-pipeline-task',
+            type: 'fanout',
+            options: {
+              durable: true,
+              autoDelete: true,
+            },
+          },
+          {
+            name: 'stop-pipeline-task',
+            type: 'fanout',
+            options: {
+              durable: true,
+              autoDelete: true,
+            },
+          },
+          {
+            name: 'update-pipeline-task',
+            type: 'fanout',
+            options: {
+              durable: false,
+              autoDelete: true,
+            },
+          },
+          {
+            name: EXCHANGE_PIPELINE_TASK_FANOUT,
+            type: 'fanout',
+            options: {
+              durable: false,
+              autoDelete: true,
+            },
+          },
+          {
+            name: EXCHANGE_PIPELINE_TASK_TOPIC,
+            type: 'topic',
+            options: {
+              durable: false,
+              autoDelete: true,
+            },
+          },
+        ],
+      }),
+      inject: [ConfigService],
+    }),
+  ],
+  providers: [
+    PipelineTasksService,
+    PipelineTasksResolver,
+    PipelineTaskRunner,
+    PipelineTaskLogger,
+    {
+      provide: 'spawn',
+      useValue: spawn,
+    },
+    PipelineTaskFlushService,
+  ],
+  exports: [PipelineTasksService],
+})
+export class PipelineTasksModule {}

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

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

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

@@ -0,0 +1,59 @@
+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 { PipelineTaskLogArgs } from './dtos/pipeline-task-log.args';
+import { plainToClass } from 'class-transformer';
+import { PipelineTaskLogger } from './pipeline-task.logger';
+import { observableToAsyncIterable } from '@graphql-tools/utils';
+import { PipelineTaskEvent } from './models/pipeline-task-event';
+
+@Resolver()
+export class PipelineTasksResolver {
+  constructor(
+    private readonly service: PipelineTasksService,
+    private readonly taskLogger: PipelineTaskLogger,
+  ) {}
+
+  @Mutation(() => PipelineTask)
+  async createPipelineTask(@Args('task') taskDto: CreatePipelineTaskInput) {
+    return await this.service.addTask(taskDto);
+  }
+
+  @Subscription(() => PipelineTaskEvent, {
+    resolve: (value) => {
+      const data = plainToClass(PipelineTaskEvent, value);
+      return data;
+    },
+  })
+  async pipelineTaskEvent(@Args() args: PipelineTaskLogArgs) {
+    const task = await this.service.findTaskById(args.taskId);
+    return observableToAsyncIterable<PipelineTaskEvent>(
+      this.taskLogger.getMessage$(task.id),
+    );
+  }
+
+  @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 pipelineTask(@Args('id') id: string) {
+    return await this.service.findTaskById(id);
+  }
+
+  @Mutation(() => Boolean)
+  async stopPipelineTask(@Args('id') id: string) {
+    const task = await this.service.findTaskById(id);
+  }
+}

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

@@ -0,0 +1,102 @@
+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 } from '../pipelines/pipeline.entity';
+import { Repository } from 'typeorm';
+import { Queue } from 'bull';
+import { AmqpConnection } from '@golevelup/nestjs-rabbitmq';
+
+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;
+
+  beforeEach(async () => {
+    module = await Test.createTestingModule({
+      providers: [
+        PipelineTasksService,
+        {
+          provide: getRepositoryToken(PipelineTask),
+          useValue: new Repository(),
+        },
+        {
+          provide: getRepositoryToken(Pipeline),
+          useValue: new Repository(),
+        },
+        {
+          provide: AmqpConnection,
+          useValue: {},
+        },
+      ],
+    }).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);
+    jest.spyOn(taskRepository, 'findOne').mockImplementation(async () => null);
+  });
+
+  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);
+  //     const findOne = jest.spyOn(taskRepository, 'findOne');
+  //     await service.addTask({ pipelineId: 'test', commit: 'test', units: [] }),
+  //       expect(save.mock.calls[0][0]).toMatchObject({
+  //         pipelineId: 'test',
+  //         commit: 'test',
+  //         units: [],
+  //       });
+  //     expect(findOne).toBeCalled();
+  //   });
+  //   it('add task', async () => {
+  //     const lpush = jest.spyOn(redisClient, 'lpush');
+  //     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(),
+  //     });
+  //   });
+  // });
+});

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

@@ -0,0 +1,63 @@
+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 { Pipeline } from '../pipelines/pipeline.entity';
+import debug from 'debug';
+import { AmqpConnection } from '@golevelup/nestjs-rabbitmq';
+
+const log = debug('fennec:pipeline-tasks:service');
+
+@Injectable()
+export class PipelineTasksService {
+  constructor(
+    @InjectRepository(PipelineTask)
+    private readonly repository: Repository<PipelineTask>,
+    @InjectRepository(Pipeline)
+    private readonly pipelineRepository: Repository<Pipeline>,
+    private readonly amqpConnection: AmqpConnection,
+  ) {}
+  async addTask(dto: CreatePipelineTaskInput) {
+    const pipeline = await this.pipelineRepository.findOneOrFail({
+      where: { id: dto.pipelineId },
+      relations: ['project'],
+    });
+    // const hasUnfinishedTask = await this.repository
+    //   .findOne({
+    //     pipelineId: dto.pipelineId,
+    //     commit: dto.commit,
+    //     status: In([TaskStatuses.pending, TaskStatuses.working]),
+    //   })
+    //   .then((val) => !isNil(val));
+    // if (hasUnfinishedTask) {
+    //   throw new ConflictException(
+    //     'There are the same tasks among the unfinished tasks!',
+    //   );
+    // }
+    const task = await this.repository.save(this.repository.create(dto));
+    task.pipeline = pipeline;
+
+    this.amqpConnection.publish('new-pipeline-task', 'mac', task);
+    return task;
+  }
+
+  async findTaskById(id: string) {
+    return await this.repository.findOneOrFail({ id });
+  }
+
+  async listTasksByPipelineId(pipelineId: string) {
+    return await this.repository.find({ pipelineId });
+  }
+
+  async listTasksByCommitHash(hash: string) {
+    return await this.repository.find({
+      where: { commit: hash },
+      order: { createdAt: 'DESC' },
+    });
+  }
+
+  getRedisTokens(pipeline: Pipeline): [string, string] {
+    return [`pipeline-${pipeline.id}:lck`, `pipeline-${pipeline.id}:tasks`];
+  }
+}

src/pipelines/commit-logs.resolver.spec.ts

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

src/pipelines/commit-logs.resolver.ts

@@ -0,0 +1,42 @@
+import { Query } from '@nestjs/graphql';
+import {
+  Args,
+  Parent,
+  ResolveField,
+  Resolver,
+  Subscription,
+} from '@nestjs/graphql';
+import { PipelineTasksService } from '../pipeline-tasks/pipeline-tasks.service';
+import { Commit, LogFields } from '../repos/dtos/log-list.model';
+import { PipelinesService } from './pipelines.service';
+
+@Resolver(() => Commit)
+export class CommitLogsResolver {
+  constructor(
+    private readonly service: PipelinesService,
+    private readonly taskServices: PipelineTasksService,
+  ) {}
+  @Subscription(() => String, { resolve: (val) => val, nullable: true })
+  async syncCommits(
+    @Args('pipelineId', { type: () => String })
+    pipelineId: string,
+    @Args('appInstance', { type: () => String, nullable: true })
+    appInstance?: string,
+  ) {
+    const pipeline = await this.service.findOneWithProject(pipelineId);
+    const syncCommitsPromise = this.service.syncCommits(pipeline, appInstance);
+    return (async function* () {
+      yield await syncCommitsPromise;
+    })();
+  }
+  @ResolveField()
+  async tasks(@Parent() commit: LogFields) {
+    return await this.taskServices.listTasksByCommitHash(commit.hash);
+  }
+
+  @Query(() => [Commit], { nullable: true })
+  async commits(@Args('pipelineId', { type: () => String }) id: string) {
+    const pipeline = await this.service.findOneWithProject(id);
+    return await this.service.listCommits(pipeline);
+  }
+}

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,7 @@
+import { InputType } from '@nestjs/graphql';
+import { CreatePipelineInput } from './create-pipeline.input';
+
+@InputType()
+export class UpdatePipelineInput extends CreatePipelineInput {
+  id: string;
+}

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,27 @@
+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 { CommitLogsResolver } from './commit-logs.resolver';
+import { PipelineTasksModule } from '../pipeline-tasks/pipeline-tasks.module';
+import { ReposModule } from '../repos/repos.module';
+import { RabbitMQModule } from '@golevelup/nestjs-rabbitmq';
+import { ConfigModule, ConfigService } from '@nestjs/config';
+
+@Module({
+  imports: [
+    TypeOrmModule.forFeature([Pipeline]),
+    PipelineTasksModule,
+    RabbitMQModule.forRootAsync(RabbitMQModule, {
+      imports: [ConfigModule],
+      useFactory: (configService: ConfigService) => ({
+        uri: configService.get<string>('db.rabbitmq.uri'),
+        exchanges: [],
+      }),
+      inject: [ConfigService],
+    }),
+  ],
+  providers: [PipelinesResolver, PipelinesService, CommitLogsResolver],
+})
+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,43 @@
+import { Args, Mutation, Query, Resolver } 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';
+
+@Resolver()
+export class PipelinesResolver {
+  constructor(private readonly service: PipelinesService) {}
+  @Query(() => [Pipeline])
+  async pipelines(@Args() dto: ListPipelineArgs) {
+    return await this.service.list(dto);
+  }
+
+  @Query(() => Pipeline)
+  async pipeline(@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 updatePipeline(
+    @Args('Pipeline', { type: () => UpdatePipelineInput })
+    dto: UpdatePipelineInput,
+  ) {
+    const tmp = await this.service.update(dto);
+    console.log(tmp);
+    return tmp;
+  }
+
+  @Mutation(() => Number)
+  async deletePipeline(@Args('id', { type: () => String }) id: string) {
+    return await this.service.remove(id);
+  }
+}

src/pipelines/pipelines.service.spec.ts

@@ -0,0 +1,49 @@
+import { Test, TestingModule } from '@nestjs/testing';
+import { PipelinesService } from './pipelines.service';
+import { Pipeline } from './pipeline.entity';
+import { getRepositoryToken } from '@nestjs/typeorm';
+import { Repository } from 'typeorm';
+import { Project } from '../projects/project.entity';
+import { AmqpConnection } from '@golevelup/nestjs-rabbitmq';
+
+describe('PipelinesService', () => {
+  let service: PipelinesService;
+  let repository: Repository<Pipeline>;
+  let pipeline: Pipeline;
+
+  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: AmqpConnection,
+          useValue: {},
+        },
+      ],
+    }).compile();
+
+    service = module.get<PipelinesService>(PipelinesService);
+    repository = module.get(getRepositoryToken(Pipeline));
+  });
+
+  it('should be defined', () => {
+    expect(service).toBeDefined();
+  });
+});

src/pipelines/pipelines.service.ts

@@ -0,0 +1,76 @@
+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 {
+  EXCHANGE_REPO,
+  ROUTE_FETCH,
+  ROUTE_LIST_COMMITS,
+} from '../repos/repos.constants';
+import { AmqpConnection } from '@golevelup/nestjs-rabbitmq';
+import { Commit } from '../repos/dtos/log-list.model';
+import { getAppInstanceRouteKey } from '../commons/utils/rabbit-mq';
+
+@Injectable()
+export class PipelinesService extends BaseDbService<Pipeline> {
+  readonly uniqueFields: Array<Array<keyof Pipeline>> = [['projectId', 'name']];
+  constructor(
+    @InjectRepository(Pipeline)
+    readonly repository: Repository<Pipeline>,
+    private readonly amqpConnection: AmqpConnection,
+  ) {
+    super(repository);
+  }
+  async list(dto: ListPipelineArgs) {
+    return this.repository.find(dto);
+  }
+
+  async findOneWithProject(id: string) {
+    return await this.repository.findOne({
+      where: { id },
+      relations: ['project'],
+    });
+  }
+
+  async create(dto: CreatePipelineInput) {
+    await this.isDuplicateEntity(dto);
+    return await this.repository.save(this.repository.create(dto));
+  }
+
+  async update(dto: UpdatePipelineInput) {
+    await this.isDuplicateEntityForUpdate(dto.id, dto);
+    const old = await this.findOne(dto.id);
+    return await this.repository.save(this.repository.merge(old, dto));
+  }
+
+  async remove(id: string) {
+    return (await this.repository.softDelete({ id })).affected;
+  }
+  async syncCommits(pipeline: Pipeline, appInstance?: string) {
+    return await this.amqpConnection.request<string | null>({
+      exchange: EXCHANGE_REPO,
+      routingKey: getAppInstanceRouteKey(ROUTE_FETCH, appInstance),
+      payload: pipeline,
+      timeout: 30_000,
+    });
+  }
+  async listCommits(pipeline: Pipeline) {
+    return await this.amqpConnection
+      .request<Commit[]>({
+        exchange: EXCHANGE_REPO,
+        routingKey: ROUTE_LIST_COMMITS,
+        payload: pipeline,
+        timeout: 10_000,
+      })
+      .then((list) =>
+        list.map((item) => {
+          item.date = new Date(item.date);
+          return item;
+        }),
+      );
+  }
+}

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

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

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

@@ -1,5 +1,9 @@
 import { InputType } from '@nestjs/graphql';
+import { IsUUID } from 'class-validator';
 import { CreateProjectInput } from './create-project.input';
 
 @InputType()
-export class UpdateProjectInput extends CreateProjectInput {}
+export class UpdateProjectInput extends CreateProjectInput {
+  @IsUUID()
+  id: 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.resolver.ts

@@ -8,12 +8,12 @@ import { ProjectsService } from './projects.service';
 export class ProjectsResolver {
   constructor(private readonly service: ProjectsService) {}
   @Query(() => [Project])
-  async findProjects() {
+  async projects() {
     return await this.service.list();
   }
 
   @Query(() => Project)
-  async findProject(@Args('id', { type: () => String }) id: string) {
+  async project(@Args('id', { type: () => String }) id: string) {
     return await this.service.findOne(id);
   }
 
@@ -26,18 +26,17 @@ export class ProjectsResolver {
   }
 
   @Mutation(() => Project)
-  async modifyProject(
+  async updateProject(
-    @Args('id', { type: () => String }) id: string,
     @Args('project', { type: () => UpdateProjectInput })
     dto: UpdateProjectInput,
   ) {
-    const tmp = await this.service.update(id, dto);
+    const tmp = await this.service.update(dto);
     console.log(tmp);
     return tmp;
   }
 
   @Mutation(() => Number)
-  async deleteProject(@Args('id', { type: () => String }) id: string) {
+  async removeProject(@Args('id', { type: () => String }) id: string) {
     return await this.service.remove(id);
   }
 }

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,9 +25,9 @@ export class ProjectsService extends BaseDbService<Project> {
     return await this.repository.save(this.repository.create(dto));
   }
 
-  async update(id: string, dto: CreateProjectInput) {
+  async update(dto: UpdateProjectInput) {
-    await this.isDuplicateEntityForUpdate(id, dto);
+    await this.isDuplicateEntityForUpdate(dto.id, dto);
-    const old = await this.findOne(id);
+    const old = await this.findOne(dto.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/dtos/log-list.model.ts

@@ -1,6 +1,18 @@
 import { ObjectType, Field } from '@nestjs/graphql';
 import { LogResult, DefaultLogFields } from 'simple-git';
+import { PipelineTask } from '../../pipeline-tasks/pipeline-task.entity';
 
+@ObjectType()
+export class Commit {
+  hash: string;
+  date: Date;
+  message: string;
+  refs: string;
+  body: string;
+  author_name: string;
+  author_email: string;
+  tasks: PipelineTask[];
+}
 @ObjectType()
 export class LogFields {
   hash: string;
@@ -10,6 +22,7 @@ export class LogFields {
   body: string;
   author_name: string;
   author_email: string;
+  tasks: PipelineTask[];
 }
 
 @ObjectType()

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

@@ -1 +1,5 @@
-export const WORKSPACE_ACTION = 'workspace-action';
+export const EXCHANGE_REPO = 'fennec.repo';
+export const ROUTE_FETCH = 'fetch';
+export const ROUTE_LIST_COMMITS = 'list-commits';
+export const QUEUE_LIST_COMMITS = 'list-commits';
+export const QUEUE_FETCH = 'repo-fetch';

src/repos/repos.module.ts

@@ -3,18 +3,35 @@ import { TypeOrmModule } from '@nestjs/typeorm';
 import { Project } from '../projects/project.entity';
 import { ReposResolver } from './repos.resolver';
 import { ReposService } from './repos.service';
-import { ConfigModule } from '@nestjs/config';
+import { ConfigModule, ConfigService } from '@nestjs/config';
-import { BullModule } from '@nestjs/bull';
+import { ProjectsModule } from '../projects/projects.module';
-import { WORKSPACE_ACTION } from './repos.constants';
+import { EXCHANGE_REPO } from './repos.constants';
+import { RabbitMQModule } from '@golevelup/nestjs-rabbitmq';
 
 @Module({
   imports: [
     TypeOrmModule.forFeature([Project]),
     ConfigModule,
-    BullModule.registerQueue({
+    ProjectsModule,
-      name: WORKSPACE_ACTION,
+    RabbitMQModule.forRootAsync(RabbitMQModule, {
+      imports: [ConfigModule],
+      useFactory: (configService: ConfigService) => ({
+        uri: configService.get<string>('db.rabbitmq.uri'),
+        exchanges: [
+          {
+            name: EXCHANGE_REPO,
+            type: 'topic',
+            options: {
+              durable: true,
+              autoDelete: true,
+            },
+          },
+        ],
+      }),
+      inject: [ConfigService],
     }),
   ],
   providers: [ReposResolver, ReposService],
+  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 { 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';
 
 @Resolver()
-export class ReposResolver {
+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),
-      };
-    });
-  }
-}

src/repos/repos.service.spec.ts

@@ -1,12 +1,17 @@
+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 { rm } from 'fs/promises';
-import { join } from 'path';
 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';
+import { getLoggerToken, PinoLogger } from 'nestjs-pino';
+import { Nack } from '@golevelup/nestjs-rabbitmq';
+import { getInstanceName } from '../commons/utils/rabbit-mq';
 
 const getTest1Project = () =>
   ({
@@ -43,6 +48,10 @@ describe('ReposService', () => {
           provide: getRepositoryToken(Project),
           useFactory: repositoryMockFactory,
         },
+        {
+          provide: getLoggerToken(ReposService.name),
+          useValue: new PinoLogger({}),
+        },
       ],
     }).compile();
 
@@ -61,7 +70,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 +81,122 @@ describe('ReposService', () => {
     it('should be return branches', async () => {
       const result = await service.listBranches({ projectId: '1' });
       expect(result).toBeDefined();
-    }, 10_000);
+    }, 20_000);
-  });
+  });
-  describe('checkoutBranch', () => {
+
-    it('should be checkout', async () => {
+  describe.skip('checkout', () => {
-      await service.checkoutBranch(getTest1Project(), 'master');
+    let task: PipelineTask;
-      const filePath = join(
+    let workspaceRoot: string;
-        service.getWorkspaceRoot(getTest1Project()),
+    beforeEach(() => {
-        'README.md',
+      const project = new Project();
-      );
+      const pipeline = new Pipeline();
-      const text = await readFile(filePath, { encoding: 'utf-8' });
+      task = new PipelineTask();
-      expect(text).toMatch(/Commit 1/gi);
+      pipeline.project = project;
-    }, 30_000);
+      task.pipeline = pipeline;
-    it('multiplexing workspace', async () => {
+      project.id = 'pid';
-      await service.checkoutBranch(getTest1Project(), 'master');
+      project.name = 'pname';
-      await service.checkoutBranch(getTest1Project(), 'branch-a');
+      pipeline.id = 'lid';
-      await service.checkoutBranch(getTest1Project(), 'branch-b');
+      pipeline.name = 'pipeline';
-      const filePath = join(
+      task.id = 'tid';
-        service.getWorkspaceRoot(getTest1Project()),
+      task.commit = '123123hash';
-        'branch-b.md',
+      workspaceRoot = service.getWorkspaceRootByTask(task);
-      );
-      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);
     });
 
-  describe('checkoutCommit', () => {
     it('should be checkout', async () => {
-      await service.checkoutCommit(getTest1Project(), '498c782685');
+      task.commit = '498c782685';
-      const filePath = join(
+      await service.checkout(task, workspaceRoot);
-        service.getWorkspaceRoot(getTest1Project()),
+      const filePath = join(workspaceRoot, 'README.md');
-        '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');
+      task.commit = '7f7123fe5b';
-      const filePath = join(
+      await service.checkout(task, workspaceRoot);
-        service.getWorkspaceRoot(getTest1Project()),
+      const filePath = join(workspaceRoot, 'README.md');
-        '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$/,
+      );
+    });
+  });
+
+  describe('fetch', () => {
+    it('success', async () => {
+      const project = new Project();
+      const pipeline = new Pipeline();
+      pipeline.branch = 'test';
+      const fetch = jest.fn((_: any) => Promise.resolve());
+      pipeline.project = project;
+      const getGit = jest.spyOn(service, 'getGit').mockImplementation(() =>
+        Promise.resolve({
+          fetch,
+        } as any),
+      );
+      await expect(service.fetch(pipeline)).resolves.toEqual(getInstanceName());
+      expect(getGit).toBeCalledTimes(1);
+      expect(getGit.mock.calls[0]?.[0]).toEqual(project);
+      expect(fetch).toBeCalledTimes(1);
+      expect(fetch.mock.calls[0]?.[0]).toMatchObject([
+        'origin',
+        'test',
+        '--depth=100',
+      ]);
+    });
+    it('failed a', async () => {
+      const project = new Project();
+      const pipeline = new Pipeline();
+      pipeline.branch = 'test';
+      const fetch = jest.fn((_: any) => Promise.resolve());
+      pipeline.project = project;
+      const getGit = jest
+        .spyOn(service, 'getGit')
+        .mockImplementation(() => Promise.reject('error'));
+      await expect(service.fetch(pipeline)).resolves.toMatchObject(new Nack());
+      expect(getGit).toBeCalledTimes(1);
+      expect(getGit.mock.calls[0]?.[0]).toEqual(project);
+      expect(fetch).toBeCalledTimes(0);
+    });
+    it('failed b', async () => {
+      const project = new Project();
+      const pipeline = new Pipeline();
+      pipeline.branch = 'test';
+      const fetch = jest.fn((_: any) => Promise.reject('error'));
+      pipeline.project = project;
+      const getGit = jest.spyOn(service, 'getGit').mockImplementation(() =>
+        Promise.resolve({
+          fetch,
+        } as any),
+      );
+      await expect(service.fetch(pipeline)).resolves.toMatchObject(new Nack());
+      expect(getGit).toBeCalledTimes(1);
+      expect(fetch).toBeCalledTimes(1);
+    });
   });
 });

src/repos/repos.service.ts

@@ -1,3 +1,5 @@
+import { ListLogsOption } from './models/list-logs.options';
+import { PipelineTask } from './../pipeline-tasks/pipeline-task.entity';
 import { Injectable, NotFoundException } from '@nestjs/common';
 import { InjectRepository } from '@nestjs/typeorm';
 import { F_OK } from 'constants';
@@ -7,57 +9,72 @@ import { gitP } from 'simple-git';
 import { Repository } from 'typeorm';
 import { Project } from '../projects/project.entity';
 import { ListBranchesArgs } from './dtos/list-branches.args';
-import { ListLogsArgs } from './dtos/list-logs.args';
 import { ConfigService } from '@nestjs/config';
+import { Commit } from './dtos/log-list.model';
+import { Nack, RabbitRPC } from '@golevelup/nestjs-rabbitmq';
+import { Pipeline } from '../pipelines/pipeline.entity';
+import { InjectPinoLogger, Logger } from 'nestjs-pino';
+import {
+  EXCHANGE_REPO,
+  QUEUE_FETCH,
+  QUEUE_LIST_COMMITS,
+  ROUTE_FETCH,
+  ROUTE_LIST_COMMITS,
+} from './repos.constants';
+import { getSelfInstanceQueueKey } from '../commons/utils/rabbit-mq';
+import {
+  getInstanceName,
+  getSelfInstanceRouteKey,
+} from '../commons/utils/rabbit-mq';
 
 const DEFAULT_REMOTE_NAME = 'origin';
+const INFO_PATH = '@info';
 @Injectable()
 export class ReposService {
   constructor(
     @InjectRepository(Project)
     private readonly projectRepository: Repository<Project>,
     private readonly configService: ConfigService,
+    @InjectPinoLogger(ReposService.name)
+    private readonly logger: Logger,
   ) {}
 
   getWorkspaceRoot(project: Project): string {
     return join(
       this.configService.get<string>('workspaces.root'),
-      project.name,
+      encodeURIComponent(project.name),
+      INFO_PATH,
     );
   }
 
-  async lockWorkspace(workspaceRoot: string) {
+  async getGit(
-    // TODO:  获取锁，失败抛错。
+    project: Project,
+    workspaceRoot?: string,
+    { fetch = true } = {},
+  ) {
+    if (!workspaceRoot) {
+      workspaceRoot = this.getWorkspaceRoot(project);
     }
 
-  async getGit(project: Project) {
+    await access(workspaceRoot, F_OK).catch(async () => {
-    const workspaceRoot = this.getWorkspaceRoot(project);
+      await mkdir(workspaceRoot, { recursive: true });
-    await this.lockWorkspace(workspaceRoot);
-
-    const firstInit = await access(workspaceRoot, F_OK)
-      .then(() => false)
-      .catch(async () => {
-        await mkdir(workspaceRoot);
-        return true;
     });
     const git = gitP(workspaceRoot);
-    if (firstInit) {
+    if (!(await git.checkIsRepo().catch(() => false))) {
       await git.init();
       await git.addRemote(DEFAULT_REMOTE_NAME, project.sshUrl);
     }
+    if (fetch) {
+      await git.fetch();
+    }
     return git;
   }
 
-  async listLogs(dto: ListLogsArgs) {
+  async listLogs({ project, branch }: ListLogsOption) {
-    const project = await this.projectRepository.findOneOrFail({
-      id: dto.projectId,
-    });
     const git = await this.getGit(project);
-    await git.fetch();
+    return await git.log(
-    return await git.log({
+      branch ? ['--branches', `remotes/origin/${branch}`, '--'] : ['--all'],
-      '--branches': dto.branch ?? '',
+    );
-      '--remotes': DEFAULT_REMOTE_NAME,
-    });
   }
 
   async listBranches(dto: ListBranchesArgs) {
@@ -68,26 +85,8 @@ export class ReposService {
     return git.branch();
   }
 
-  async checkoutBranch(project: Project, branch: string) {
+  async checkout(task: PipelineTask, workspaceRoot: string) {
-    const git = await this.getGit(project);
+    const git = await this.getGit(task.pipeline.project, workspaceRoot);
-    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);
     try {
       await git.fetch(DEFAULT_REMOTE_NAME);
     } catch (err) {
@@ -96,6 +95,86 @@ 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}`),
+    );
+  }
+
+  async checkout4Task(task: PipelineTask): Promise<string> {
+    const path = this.getWorkspaceRootByTask(task);
+    await this.checkout(task, path);
+    return path;
+  }
+
+  @RabbitRPC({
+    exchange: EXCHANGE_REPO,
+    routingKey: [
+      ROUTE_LIST_COMMITS,
+      getSelfInstanceRouteKey(ROUTE_LIST_COMMITS),
+    ],
+    queue: getSelfInstanceQueueKey(QUEUE_LIST_COMMITS),
+    queueOptions: {
+      autoDelete: true,
+    },
+  })
+  async listCommits(pipeline: Pipeline): Promise<Commit[] | Nack> {
+    const git = await this.getGit(pipeline.project, undefined, {
+      fetch: false,
+    });
+    try {
+      const data = await git.log([
+        '-100',
+        '--branches',
+        `remotes/origin/${pipeline.branch}`,
+        '--',
+      ]);
+      return data.all.map(
+        (it) =>
+          ({
+            ...it,
+            date: new Date(it.date),
+          } as Commit),
+      );
+    } catch (error) {
+      this.logger.error(
+        { error, pipeline },
+        '[listCommits] %s',
+        error?.message,
+      );
+      return new Nack();
+    }
+  }
+
+  @RabbitRPC({
+    exchange: EXCHANGE_REPO,
+    routingKey: [ROUTE_FETCH, getSelfInstanceRouteKey(ROUTE_FETCH)],
+    queue: getSelfInstanceQueueKey(QUEUE_FETCH),
+    queueOptions: {
+      autoDelete: true,
+    },
+  })
+  async fetch(pipeline: Pipeline): Promise<string | null | Nack> {
+    try {
+      const git = await this.getGit(pipeline.project, undefined, {
+        fetch: false,
+      });
+      await git.fetch(['origin', pipeline.branch, '--depth=100']);
+      return getInstanceName();
+    } catch (error) {
+      this.logger.error({ error, pipeline }, '[fetch] %s', error?.message);
+      return new Nack();
+    }
   }
 }

src/repos/workspace-action.consumer.ts

@@ -1,8 +0,0 @@
-import { Process, Processor } from '@nestjs/bull';
-import { WORKSPACE_ACTION } from './repos.constants';
-
-@Processor(WORKSPACE_ACTION)
-export class WorkspaceActionConsumer {
-  @Process()
-  async dispatch() {}
-}

src/repos/workspace-action.model.ts

@@ -1,5 +0,0 @@
-import { WorkspaceActions } from './workspace-actions.enum';
-
-export class WorkspaceAction {
-  action: WorkspaceActions;
-}

src/repos/workspace-actions.enum.ts

@@ -1,6 +0,0 @@
-export enum WorkspaceActions {
-  checkoutBranch = 'checkoutBranch',
-  checkoutCommit = 'checkoutCommit',
-  listLogs = 'listLogs',
-  listBranches = 'listBranches',
-}

src/webhooks/dtos/gitea-hook-payload.dto.ts

@@ -0,0 +1,8 @@
+import { IsString } from 'class-validator';
+
+export class GiteaHookPayloadDto {
+  @IsString()
+  ref: string;
+  @IsString()
+  after: string;
+}

src/webhooks/enums/source-service.enum.ts

@@ -0,0 +1,3 @@
+export enum SourceService {
+  gitea = 'gitea',
+}

src/webhooks/gitea-webhooks.controller.spec.ts

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

src/webhooks/gitea-webhooks.controller.ts

@@ -0,0 +1,37 @@
+import { Body, Controller, Headers, Param, Post } from '@nestjs/common';
+import { validateOrReject } from 'class-validator';
+import { pick } from 'ramda';
+import { GiteaHookPayloadDto } from './dtos/gitea-hook-payload.dto';
+import { SourceService } from './enums/source-service.enum';
+import { WebhookLog } from './webhook-log.entity';
+import { WebhooksService } from './webhooks.service';
+
+@Controller('gitea-webhooks')
+export class GiteaWebhooksController {
+  constructor(private readonly service: WebhooksService) {}
+  @Post(':pipelineId')
+  async onCall(
+    @Headers('X-Gitea-Delivery') delivery: string,
+    @Headers('X-Gitea-Event') event: string,
+    @Headers('X-Gitea-Signature') signature: string,
+    @Body() body: Buffer,
+    @Param('pipelineId') pipelineId: string,
+  ) {
+    const payload = Object.assign(
+      new GiteaHookPayloadDto(),
+      JSON.parse(body.toString('utf-8')),
+    );
+    await validateOrReject(payload);
+    await this.service.verifySignature(body, signature, 'boardcat');
+    return await this.service
+      .onCall(pipelineId, {
+        payload,
+        sourceDelivery: delivery,
+        sourceEvent: event,
+        sourceService: SourceService.gitea,
+      })
+      .then((data) =>
+        pick<keyof WebhookLog>(['id', 'createdAt', 'localEvent'])(data),
+      );
+  }
+}

src/webhooks/models/create-webhook-log.model.ts

@@ -0,0 +1,9 @@
+import { SourceService } from '../enums/source-service.enum';
+import { WebhookLog } from '../webhook-log.entity';
+
+export class CreateWebhookLogModel<T> implements Partial<WebhookLog> {
+  sourceDelivery: string;
+  sourceEvent: string;
+  sourceService: SourceService;
+  payload: T;
+}

src/webhooks/webhook-log.entity.ts

@@ -0,0 +1,19 @@
+import { Column, Entity } from 'typeorm';
+import { AppBaseEntity } from './../commons/entities/app-base-entity';
+import { SourceService } from './enums/source-service.enum';
+
+@Entity()
+export class WebhookLog extends AppBaseEntity {
+  @Column()
+  sourceDelivery: string;
+  @Column({ type: 'enum', enum: SourceService })
+  sourceService: SourceService;
+  @Column()
+  sourceEvent: string;
+  @Column({ type: 'jsonb' })
+  payload: any;
+  @Column()
+  localEvent: string;
+  @Column({ type: 'jsonb' })
+  localPayload: any;
+}

src/webhooks/webhooks.module.ts

@@ -0,0 +1,14 @@
+import { MiddlewareConsumer, Module } from '@nestjs/common';
+import { TypeOrmModule } from '@nestjs/typeorm';
+import { PipelineTasksModule } from '../pipeline-tasks/pipeline-tasks.module';
+import { GiteaWebhooksController } from './gitea-webhooks.controller';
+import { WebhookLog } from './webhook-log.entity';
+import { WebhooksService } from './webhooks.service';
+
+@Module({
+  imports: [TypeOrmModule.forFeature([WebhookLog]), PipelineTasksModule],
+  controllers: [GiteaWebhooksController],
+  providers: [WebhooksService],
+})
+export class WebhooksModule {
+}

src/webhooks/webhooks.service.spec.ts

@@ -0,0 +1,57 @@
+import { UnauthorizedException } from '@nestjs/common';
+import { Test, TestingModule } from '@nestjs/testing';
+import { getRepositoryToken } from '@nestjs/typeorm';
+import { readFile } from 'fs/promises';
+import { join } from 'path';
+import { Repository } from 'typeorm';
+import { PipelineTasksService } from '../pipeline-tasks/pipeline-tasks.service';
+import { WebhookLog } from './webhook-log.entity';
+import { WebhooksService } from './webhooks.service';
+
+describe('WebhooksService', () => {
+  let service: WebhooksService;
+
+  beforeEach(async () => {
+    const module: TestingModule = await Test.createTestingModule({
+      providers: [
+        WebhooksService,
+        {
+          provide: PipelineTasksService,
+          useValue: {},
+        },
+        {
+          provide: getRepositoryToken(WebhookLog),
+          useValue: new Repository(),
+        },
+      ],
+    }).compile();
+
+    service = module.get<WebhooksService>(WebhooksService);
+  });
+
+  it('should be defined', () => {
+    expect(service).toBeDefined();
+  });
+
+  describe('verifySignature', () => {
+    const signature =
+      'b175e07189a6106f386b62253b18b5879c4b1f3af2f11fe13a294602671e361a';
+    const secret = 'boardcat';
+    let payload: Buffer;
+    beforeAll(async () => {
+      payload = await readFile(
+        join(__dirname, '../../test/data/gitea-hook-payload.json.bin'),
+      );
+    });
+    it('must be valid', async () => {
+      await expect(
+        service.verifySignature(payload, signature, secret),
+      ).resolves.toEqual(undefined);
+    });
+    it('must be invalid', async () => {
+      await expect(
+        service.verifySignature(payload, 'test', secret),
+      ).rejects.toThrowError(UnauthorizedException);
+    });
+  });
+});

src/webhooks/webhooks.service.ts

@@ -0,0 +1,65 @@
+import {
+  BadRequestException,
+  Injectable,
+  UnauthorizedException,
+} from '@nestjs/common';
+import { InjectRepository } from '@nestjs/typeorm';
+import { createHmac } from 'crypto';
+import { Repository } from 'typeorm';
+import { PipelineUnits } from '../pipeline-tasks/enums/pipeline-units.enum';
+import { PipelineTasksService } from '../pipeline-tasks/pipeline-tasks.service';
+import { GiteaHookPayloadDto } from './dtos/gitea-hook-payload.dto';
+import { CreateWebhookLogModel } from './models/create-webhook-log.model';
+import { WebhookLog } from './webhook-log.entity';
+
+@Injectable()
+export class WebhooksService {
+  constructor(
+    @InjectRepository(WebhookLog)
+    private readonly repository: Repository<WebhookLog>,
+    private readonly taskService: PipelineTasksService,
+  ) {}
+
+  async onCall(
+    pipelineId: string,
+    model: CreateWebhookLogModel<GiteaHookPayloadDto>,
+  ) {
+    if (model.sourceEvent.toLowerCase() === 'push') {
+      const taskDto = {
+        pipelineId,
+        commit: model.payload.after,
+        units: Object.values(PipelineUnits),
+      };
+      await this.taskService.addTask(taskDto);
+      return await this.repository.save(
+        this.repository.create({
+          ...model,
+          localEvent: 'create-pipeline-task',
+          localPayload: taskDto,
+        }),
+      );
+    } else {
+      throw new BadRequestException('无法处理的请求');
+    }
+  }
+
+  async verifySignature(payload: any, signature: string, secret: string) {
+    const local = await new Promise<string>((resolve, reject) => {
+      const hmac = createHmac('sha256', secret);
+      hmac.on('readable', () => {
+        const data = hmac.read();
+        if (data) {
+          resolve(data.toString('hex'));
+        }
+      });
+      hmac.on('error', (err) => {
+        reject(err);
+      });
+      hmac.write(payload);
+      hmac.end();
+    });
+    if (local !== signature) {
+      throw new UnauthorizedException();
+    }
+  }
+}

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/gitea-hook-payload.json.bin

@@ -0,0 +1,115 @@
+{
+  "secret": "boardcat",
+  "ref": "refs/heads/master",
+  "before": "429de1eaedf1da83f1e0e3ac3d8b20e771b7051c",
+  "after": "429de1eaedf1da83f1e0e3ac3d8b20e771b7051c",
+  "compare_url": "",
+  "commits": [
+    {
+      "id": "429de1eaedf1da83f1e0e3ac3d8b20e771b7051c",
+      "message": "test(pipeline-tasks): pass test cases.\n",
+      "url": "https://git.ivanli.cc/Fennec/fennec-be/commit/429de1eaedf1da83f1e0e3ac3d8b20e771b7051c",
+      "author": {
+        "name": "Ivan",
+        "email": "ivanli@live.cn",
+        "username": ""
+      },
+      "committer": {
+        "name": "Ivan",
+        "email": "ivanli@live.cn",
+        "username": ""
+      },
+      "verification": null,
+      "timestamp": "0001-01-01T00:00:00Z",
+      "added": null,
+      "removed": null,
+      "modified": null
+    }
+  ],
+  "head_commit": null,
+  "repository": {
+    "id": 3,
+    "owner": {
+      "id": 3,
+      "login": "Fennec",
+      "full_name": "",
+      "email": "",
+      "avatar_url": "https://git.ivanli.cc/user/avatar/Fennec/-1",
+      "language": "",
+      "is_admin": false,
+      "last_login": "1970-01-01T08:00:00+08:00",
+      "created": "2021-01-30T16:46:11+08:00",
+      "username": "Fennec"
+    },
+    "name": "fennec-be",
+    "full_name": "Fennec/fennec-be",
+    "description": "Fennec CI/CD Back-End",
+    "empty": false,
+    "private": false,
+    "fork": false,
+    "template": false,
+    "parent": null,
+    "mirror": false,
+    "size": 1897,
+    "html_url": "https://git.ivanli.cc/Fennec/fennec-be",
+    "ssh_url": "ssh://gitea@git.ivanli.cc:7018/Fennec/fennec-be.git",
+    "clone_url": "https://git.ivanli.cc/Fennec/fennec-be.git",
+    "original_url": "",
+    "website": "",
+    "stars_count": 1,
+    "forks_count": 0,
+    "watchers_count": 1,
+    "open_issues_count": 0,
+    "open_pr_counter": 0,
+    "release_counter": 0,
+    "default_branch": "master",
+    "archived": false,
+    "created_at": "2021-01-31T09:58:38+08:00",
+    "updated_at": "2021-03-27T15:57:00+08:00",
+    "permissions": {
+      "admin": false,
+      "push": false,
+      "pull": false
+    },
+    "has_issues": true,
+    "internal_tracker": {
+      "enable_time_tracker": true,
+      "allow_only_contributors_to_track_time": true,
+      "enable_issue_dependencies": true
+    },
+    "has_wiki": true,
+    "has_pull_requests": true,
+    "has_projects": true,
+    "ignore_whitespace_conflicts": false,
+    "allow_merge_commits": true,
+    "allow_rebase": true,
+    "allow_rebase_explicit": true,
+    "allow_squash_merge": true,
+    "avatar_url": "",
+    "internal": false
+  },
+  "pusher": {
+    "id": 1,
+    "login": "Ivan",
+    "full_name": "Ivan Li",
+    "email": "ivan@noreply.%(DOMAIN)s",
+    "avatar_url": "https://git.ivanli.cc/user/avatar/Ivan/-1",
+    "language": "zh-CN",
+    "is_admin": true,
+    "last_login": "2021-03-26T22:28:05+08:00",
+    "created": "2021-01-23T18:15:30+08:00",
+    "username": "Ivan"
+  },
+  "sender": {
+    "id": 1,
+    "login": "Ivan",
+    "full_name": "Ivan Li",
+    "email": "ivan@noreply.%(DOMAIN)s",
+    "avatar_url": "https://git.ivanli.cc/user/avatar/Ivan/-1",
+    "language": "zh-CN",
+    "is_admin": true,
+    "last_login": "2021-03-26T22:28:05+08:00",
+    "created": "2021-01-23T18:15:30+08:00",
+    "username": "Ivan"
+  }
+}

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);