chore: housework.

Reviewed-on: #9

.gitignore

@@ -34,3 +34,5 @@ lerna-debug.log*
 !.vscode/extensions.json
 
 /config.yml
+tsconfig.build.tsbuildinfo
+.eslintcache

.husky/.gitignore

@@ -0,0 +1 @@
+_

.husky/pre-commit

@@ -0,0 +1,5 @@
+#!/bin/sh
+. "$(dirname "$0")/_/husky.sh"
+
+npx lint-staged
+npm test

.vscode/extensions.json

@@ -0,0 +1,3 @@
+{
+  "recommendations": []
+}

.vscode/settings.json

@@ -1,5 +1,6 @@
 {
   "cSpell.words": [
+    "Mutex",
     "Repos",
     "amqp",
     "boardcat",

config.yml.example

@@ -16,5 +16,9 @@ db:
     prefix: fennec
   rabbitmq:
     uri: 'amqp://fennec:fennec@192.168.31.194:5672'
+  etcd:
+    hosts:
+      - 'http://192.168.31.194:2379'
+
 workspaces:
   root: '/Users/ivanli/Projects/fennec/workspaces'

ecosystem.config.js

@@ -0,0 +1,14 @@
+module.exports = {
+  apps: [
+    {
+      name: 'fennec-be',
+      script: 'npm',
+      args: 'run start:prod',
+      watch: false,
+      ignore_watch: ['node_modules'],
+      log_date_format: 'MM-DD HH:mm:ss.SSS Z',
+      env: {},
+      max_restarts: 5,
+    },
+  ],
+};

package.json

@@ -1,8 +1,8 @@
 {
   "name": "fennec-be",
-  "version": "0.0.1",
-  "description": "",
-  "author": "",
+  "version": "0.1.1",
+  "description": "a ci/cd project.",
+  "author": "Ivan Li\b<ivanli2048@gmail.com>",
   "private": true,
   "license": "UNLICENSED",
   "scripts": {
@@ -18,35 +18,39 @@
     "test:watch": "jest --watch",
     "test:cov": "jest --coverage",
     "test:debug": "node --inspect-brk -r tsconfig-paths/register -r ts-node/register node_modules/.bin/jest --runInBand",
-    "test:e2e": "jest --config ./test/jest-e2e.json"
+    "test:e2e": "jest --config ./test/jest-e2e.json",
+    "prepare": "husky install"
   },
   "dependencies": {
     "@golevelup/nestjs-rabbitmq": "^1.16.1",
-    "@nestjs/bull": "^0.3.1",
+    "@nestjs-lib/auth": "^0.2.1",
     "@nestjs/common": "^7.5.1",
     "@nestjs/config": "^0.6.2",
     "@nestjs/core": "^7.5.1",
     "@nestjs/graphql": "^7.9.8",
     "@nestjs/platform-express": "^7.5.1",
     "@nestjs/typeorm": "^7.1.5",
-    "@types/bull": "^3.15.0",
+    "@types/amqplib": "^0.8.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",
+    "configuration": "file:../configuration",
     "debug": "^4.3.1",
     "graphql": "^15.5.0",
-    "graphql-tools": "^7.0.2",
+    "graphql-tools": "^8.1.0",
     "ioredis": "^4.25.0",
+    "jose": "^3.14.0",
     "js-yaml": "^4.0.0",
+    "nestjs-etcd": "^0.2.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",
+    "pm2": "^5.1.0",
     "ramda": "^0.27.1",
     "reflect-metadata": "^0.1.13",
     "rimraf": "^3.0.2",
@@ -55,7 +59,7 @@
     "typeorm": "^0.2.30"
   },
   "devDependencies": {
-    "@nestjs/cli": "^7.5.7",
+    "@nestjs/cli": "^7.6.0",
     "@nestjs/schematics": "^7.1.3",
     "@nestjs/testing": "^7.5.1",
     "@types/body-parser": "^1.19.0",
@@ -72,7 +76,9 @@
     "eslint": "^7.12.1",
     "eslint-config-prettier": "7.2.0",
     "eslint-plugin-prettier": "^3.1.4",
+    "husky": "^7.0.2",
     "jest": "^26.6.3",
+    "lint-staged": "^11.1.2",
     "prettier": "^2.1.2",
     "supertest": "^6.0.0",
     "ts-jest": "^26.4.3",
@@ -95,7 +101,13 @@
     "collectCoverageFrom": [
       "**/*.(t|j)s"
     ],
+    "moduleNameMapper": {
+      "^jose/(.*)$": "<rootDir>/../node_modules/jose/dist/node/cjs/$1"
+    },
     "coverageDirectory": "../coverage",
     "testEnvironment": "node"
+  },
+  "lint-staged": {
+    "{src,apps,libs,test}/**/*.ts": "eslint --cache --fix"
   }
 }

src/app.module.ts

@@ -1,3 +1,4 @@
+import { CommonsModule } from './commons/commons.module';
 import { MiddlewareConsumer, Module, NestModule } from '@nestjs/common';
 import { ConfigModule, ConfigService } from '@nestjs/config';
 import { GraphQLModule } from '@nestjs/graphql';
@@ -12,13 +13,14 @@ 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 { RawBodyMiddleware } from './commons/middleware/raw-body.middleware';
 import { GiteaWebhooksController } from './webhooks/gitea-webhooks.controller';
-import { ParseBodyMiddleware } from './commons/middlewares/parse-body.middleware';
-import { BullModule } from '@nestjs/bull';
+import { ParseBodyMiddleware } from './commons/middleware/parse-body.middleware';
 import { LoggerModule } from 'nestjs-pino';
-
+import { EtcdModule } from 'nestjs-etcd';
 import pinoPretty from 'pino-pretty';
+import { fromPairs, map, pipe, toPairs } from 'ramda';
+import { ConfigurationsModule } from './configurations/configurations.module';
 
 @Module({
   imports: [
@@ -63,16 +65,21 @@ import pinoPretty from 'pino-pretty';
         playground: true,
         autoSchemaFile: true,
         installSubscriptionHandlers: true,
-      }),
-      inject: [ConfigService],
-    }),
-    BullModule.forRootAsync({
-      imports: [ConfigModule],
-      useFactory: (configService: ConfigService) => ({
-        redis: {
-          host: configService.get<string>('db.redis.host', 'localhost'),
-          port: configService.get<number>('db.redis.port', undefined),
-          password: configService.get<string>('db.redis.password', undefined),
+        context: ({ req, connection }) => {
+          return connection ? { req: connection.context } : { req };
+        },
+        subscriptions: {
+          onConnect: (connectionParams: Record<string, string>) => {
+            const connectionParamsWithLowerKeys = pipe(
+              toPairs,
+              map(([key, value]) => [key.toLowerCase(), value]),
+              fromPairs,
+            )(connectionParams);
+
+            return {
+              headers: connectionParamsWithLowerKeys,
+            };
+          },
         },
       }),
       inject: [ConfigService],
@@ -91,7 +98,16 @@ import pinoPretty from 'pino-pretty';
       }),
       inject: [ConfigService],
     }),
+    EtcdModule.forRootAsync({
+      imports: [ConfigModule],
+      useFactory: (configService: ConfigService) => ({
+        hosts: configService.get<string>('db.etcd.hosts', 'localhost:2379'),
+      }),
+      inject: [ConfigService],
+    }),
     WebhooksModule,
+    CommonsModule,
+    ConfigurationsModule,
   ],
   controllers: [AppController],
   providers: [AppService, AppResolver],

src/commons/commons.module.ts

@@ -1,8 +1,12 @@
-import { Module } from '@nestjs/common';
+import { Global, Module } from '@nestjs/common';
 import { PasswordConverter } from './services/password-converter';
+import { RedisMutexModule } from './redis-mutex/redis-mutex.module';
+import { AuthModule } from '@nestjs-lib/auth';
 
+@Global()
 @Module({
+  imports: [RedisMutexModule, AuthModule],
   providers: [PasswordConverter],
-  exports: [PasswordConverter],
+  exports: [PasswordConverter, RedisMutexModule, AuthModule],
 })
 export class CommonsModule {}

src/commons/exceptions/application.exception.ts

@@ -1,11 +1,11 @@
+import { pick } from 'ramda';
+
 export class ApplicationException extends Error {
   code: number;
   error: Error;
 
   constructor(
-    message:
-      | string
-      | { error?: Error; message?: string | object; code?: number },
+    message: string | { error?: Error; message?: string | any; code?: number },
   ) {
     if (message instanceof Object) {
       super();
@@ -13,9 +13,13 @@ export class ApplicationException extends Error {
       this.error = message.error;
       this.message = message.message as any;
     } else if (typeof message === 'string') {
-      super((message as unknown) as any);
+      super(message as unknown as any);
     } else {
-      super((message as unknown) as any);
+      super(message as unknown as any);
     }
   }
+
+  toJSON() {
+    return pick(['code', 'message'], this);
+  }
 }

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

@@ -6,23 +6,34 @@ import {
   HttpStatus,
 } from '@nestjs/common';
 import { ApolloError } from 'apollo-server-errors';
+import { PinoLogger, InjectPinoLogger } from 'nestjs-pino';
 
 @Catch(HttpException)
 export class HttpExceptionFilter implements ExceptionFilter {
+  constructor(
+    @InjectPinoLogger(HttpExceptionFilter.name)
+    private readonly logger: PinoLogger,
+  ) {}
   catch(exception: HttpException, host: ArgumentsHost) {
     switch (host.getType<'http' | 'graphql' | string>()) {
       case 'graphql': {
-        const message = exception.message;
+        const errorName = exception.message;
         const extensions: Record<string, any> = {};
-        const err = exception.getResponse();
+        const err = exception.getResponse() as any;
         if (typeof err === 'string') {
           extensions.message = err;
         } else {
-          Object.assign(extensions, (err as any).extension);
-          extensions.message = (err as any).message;
+          Object.assign(extensions, err.extension);
+          if (typeof err.message === 'string') {
+            extensions.message = err.message;
+          } else {
+            extensions.message = err.error;
           }
+        }
+        extensions.error = errorName;
+        this.logger.error(extensions);
         return new ApolloError(
-          message,
+          extensions.message,
           exception.getStatus().toString(),
           extensions,
         );

src/commons/redis-mutex/redis-mutex.module.ts

@@ -0,0 +1,10 @@
+import { Module } from '@nestjs/common';
+import { RedisMutexService } from './redis-mutex.service';
+import { RedisModule } from 'nestjs-redis';
+
+@Module({
+  imports: [RedisModule],
+  providers: [RedisMutexService],
+  exports: [RedisMutexService],
+})
+export class RedisMutexModule {}

src/commons/redis-mutex/redis-mutex.service.spec.ts

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

src/commons/redis-mutex/redis-mutex.service.ts

@@ -0,0 +1,71 @@
+import { Injectable } from '@nestjs/common';
+import { RedisService } from 'nestjs-redis';
+import * as uuid from 'uuid';
+import { ApplicationException } from '../exceptions/application.exception';
+
+export interface RedisMutexOption {
+  /**
+   * seconds
+   */
+  expires?: number;
+  /**
+   * seconds
+   */
+  timeout?: number | null;
+  /**
+   * milliseconds
+   */
+  retryDelay?: number;
+}
+
+@Injectable()
+export class RedisMutexService {
+  constructor(private readonly redisClient: RedisService) {}
+
+  public async lock(
+    key: string,
+    { expires = 100, timeout = 10, retryDelay = 100 }: RedisMutexOption = {
+      expires: 100,
+      timeout: 10,
+      retryDelay: 100,
+    },
+  ) {
+    const redisKey = `${'mutex-lock'}:${key}`;
+    const redis = this.redisClient.getClient();
+    const value = uuid.v4();
+    const timeoutAt = timeout ? Date.now() + timeout * 1000 : null;
+
+    while (
+      !(await redis
+        .set(redisKey, value, 'EX', expires, 'NX')
+        .then(() => true)
+        .catch(() => false))
+    ) {
+      if (timeoutAt && timeoutAt > Date.now()) {
+        throw new ApplicationException('lock timeout');
+      }
+      await new Promise((resolve) => setTimeout(resolve, retryDelay));
+    }
+
+    const renewTimer = setInterval(() => {
+      redis.expire(redisKey, expires);
+    }, (expires * 1000) / 2);
+
+    return async () => {
+      clearInterval(renewTimer);
+      await redis.eval(
+        `
+      if redis.call("get", KEYS[1]) == ARGV[1]
+      then
+          return redis.call("del", KEYS[1])
+      else
+          return 0
+      end
+    `,
+        1,
+        redisKey,
+        value,
+      );
+    };
+  }
+}

src/commons/services/base-db.service.ts

@@ -53,8 +53,22 @@ export class BaseDbService<Entity extends AppBaseEntity> extends TypeormHelper {
   async isDuplicateEntityForUpdate<Dto extends Entity>(
     id: string,
     dto: Partial<Dto>,
+    extendsFields?: Array<keyof Dto & string>,
+  ): Promise<false | never>;
+  async isDuplicateEntityForUpdate<Dto extends Entity>(
+    old: Entity,
+    dto: Partial<Dto>,
+    extendsFields?: Array<keyof Dto & string>,
+  ): Promise<false | never>;
+  async isDuplicateEntityForUpdate<Dto extends Entity>(
+    id: string | Entity,
+    dto: Partial<Dto>,
     extendsFields: Array<keyof Dto & string> = [],
   ): Promise<false | never> {
+    if (typeof id !== 'string') {
+      dto = Object.assign({}, id, dto);
+      id = id.id;
+    }
     const qb = this.repository.createQueryBuilder('entity');
     const compareFields = this.getCompareFields(dto, [
       ...this.uniqueFields,
@@ -112,6 +126,7 @@ export class BaseDbService<Entity extends AppBaseEntity> extends TypeormHelper {
     }
   }
 
+  // eslint-disable-next-line @typescript-eslint/no-unused-vars
   async canYouRemoveWithIds(ids: string[]): Promise<void> {
     return;
   }

src/configurations/configurations.module.ts

@@ -0,0 +1,11 @@
+import { Configuration } from './entities/configuration.entity';
+import { TypeOrmModule } from '@nestjs/typeorm';
+import { Module } from '@nestjs/common';
+import { ConfigurationsService } from './configurations.service';
+import { ConfigurationsResolver } from './configurations.resolver';
+
+@Module({
+  imports: [TypeOrmModule.forFeature([Configuration])],
+  providers: [ConfigurationsResolver, ConfigurationsService],
+})
+export class ConfigurationsModule {}

src/configurations/configurations.resolver.spec.ts

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

src/configurations/configurations.resolver.ts

@@ -0,0 +1,39 @@
+import { UnprocessableEntityException } from '@nestjs/common';
+import { GetConfigurationArgs } from './dto/get-configuration.args';
+import { SetConfigurationInput } from './dto/set-configuration.input';
+import { Resolver, Mutation, Args, Query } from '@nestjs/graphql';
+import { ConfigurationsService } from './configurations.service';
+import { Configuration } from './entities/configuration.entity';
+import { any, pipe, values } from 'ramda';
+import { AccountRole, Roles } from '@nestjs-lib/auth';
+
+@Roles(AccountRole.admin, AccountRole.super)
+@Resolver(() => Configuration)
+export class ConfigurationsResolver {
+  constructor(private readonly configurationsService: ConfigurationsService) {}
+
+  @Mutation(() => Configuration)
+  setConfiguration(
+    @Args('setConfigurationInput', { type: () => SetConfigurationInput })
+    setConfigurationInput: SetConfigurationInput,
+  ) {
+    return this.configurationsService.setConfiguration(setConfigurationInput);
+  }
+
+  @Query(() => Configuration, { nullable: true })
+  getConfiguration(
+    @Args()
+    getConfigurationArgs: GetConfigurationArgs,
+  ) {
+    if (
+      pipe(
+        values,
+        any((value) => !value),
+      )(getConfigurationArgs)
+    ) {
+      throw new UnprocessableEntityException('Must pass a parameter');
+    }
+
+    return this.configurationsService.findOneByConditions(getConfigurationArgs);
+  }
+}

src/configurations/configurations.service.spec.ts

@@ -0,0 +1,49 @@
+import { getRepositoryToken } from '@nestjs/typeorm';
+import { Test, TestingModule } from '@nestjs/testing';
+import { ConfigurationsService } from './configurations.service';
+import { Configuration } from './entities/configuration.entity';
+import { IsNull } from 'typeorm';
+import { Etcd3 } from 'etcd3';
+
+describe('ConfigurationsService', () => {
+  let service: ConfigurationsService;
+
+  beforeEach(async () => {
+    const module: TestingModule = await Test.createTestingModule({
+      providers: [
+        ConfigurationsService,
+        {
+          provide: getRepositoryToken(Configuration),
+          useValue: {},
+        },
+        {
+          provide: Etcd3,
+          useValue: {},
+        },
+      ],
+    }).compile();
+
+    service = module.get<ConfigurationsService>(ConfigurationsService);
+  });
+
+  it('should be defined', () => {
+    expect(service).toBeDefined();
+  });
+
+  describe('findOneByConditions', () => {
+    it('should select by projectId only', async () => {
+      const entity = new Configuration();
+      const findOne = jest.fn<any, [any]>(() => Promise.resolve(entity));
+      service['repository'].findOne = findOne;
+
+      await expect(
+        service.findOneByConditions({ projectId: 'uuid' }),
+      ).resolves.toEqual(entity);
+
+      expect(findOne.mock.calls[0][0]).toMatchObject({
+        projectId: 'uuid',
+        pipelineId: IsNull(),
+      });
+    });
+  });
+});

src/configurations/configurations.service.ts

@@ -0,0 +1,55 @@
+import { GetConfigurationArgs } from './dto/get-configuration.args';
+import { BaseDbService } from './../commons/services/base-db.service';
+import { Injectable } from '@nestjs/common';
+import { Configuration } from './entities/configuration.entity';
+import { InjectRepository } from '@nestjs/typeorm';
+import { FindConditions, IsNull, Repository } from 'typeorm';
+import { SetConfigurationInput } from './dto/set-configuration.input';
+import { pick } from 'ramda';
+import { Etcd3 } from 'etcd3';
+
+@Injectable()
+export class ConfigurationsService extends BaseDbService<Configuration> {
+  constructor(
+    @InjectRepository(Configuration)
+    configurationRepository: Repository<Configuration>,
+    private readonly etcd: Etcd3,
+  ) {
+    super(configurationRepository);
+  }
+
+  async setConfiguration(dto: SetConfigurationInput) {
+    let entity = await this.repository.findOne(
+      pick(['pipelineId', 'projectId'], dto),
+    );
+    if (!entity) {
+      entity = this.repository.create(dto);
+    }
+    entity = await this.repository.save(entity);
+    await this.syncToEtcd(entity);
+    return entity;
+  }
+
+  async findOneByConditions(dto: FindConditions<GetConfigurationArgs>) {
+    if (dto.projectId && !dto.pipelineId) {
+      dto.pipelineId = IsNull();
+    }
+    return await this.repository.findOne(dto);
+  }
+
+  async syncToEtcd({ pipelineId, id }: { pipelineId?: string; id?: string }) {
+    const config = await this.repository.findOneOrFail({
+      where: { pipelineId, id },
+      relations: ['pipeline', 'project'],
+    });
+
+    await this.etcd
+      .put(`share/config/${config.id}`)
+      .value(config.content)
+      .exec();
+    await this.etcd
+      .put(`share/config/${config.pipeline.environment}/${config.project.name}`)
+      .value(config.content)
+      .exec();
+  }
+}

src/configurations/dto/get-configuration.args.ts

@@ -0,0 +1,17 @@
+import { ArgsType } from '@nestjs/graphql';
+import { IsUUID, IsOptional } from 'class-validator';
+
+@ArgsType()
+export class GetConfigurationArgs {
+  @IsUUID()
+  @IsOptional()
+  pipelineId?: string;
+
+  @IsUUID()
+  @IsOptional()
+  projectId?: string;
+
+  @IsUUID()
+  @IsOptional()
+  id?: string;
+}

src/configurations/dto/set-configuration.input.ts

@@ -0,0 +1,26 @@
+import { ConfigurationLanguage } from './../enums/configuration-language.enum';
+import { IsEnum, IsString, IsUUID, Length, IsOptional } from 'class-validator';
+import { InputType } from '@nestjs/graphql';
+
+@InputType()
+export class SetConfigurationInput {
+  @IsOptional()
+  @IsUUID()
+  id?: string;
+
+  @IsUUID()
+  pipelineId: string;
+
+  @IsUUID()
+  projectId: string;
+
+  @IsString()
+  content: string;
+
+  @IsEnum(ConfigurationLanguage)
+  language: ConfigurationLanguage;
+
+  @Length(0, 100)
+  @IsOptional()
+  name = 'Default Configuration';
+}

src/configurations/entities/configuration.entity.ts

@@ -0,0 +1,35 @@
+import { Project } from './../../projects/project.entity';
+import { ConfigurationLanguage } from './../enums/configuration-language.enum';
+import { Pipeline } from './../../pipelines/pipeline.entity';
+import { AppBaseEntity } from './../../commons/entities/app-base-entity';
+import { ObjectType } from '@nestjs/graphql';
+import { Column, Entity, ManyToOne } from 'typeorm';
+
+@Entity()
+@ObjectType()
+export class Configuration extends AppBaseEntity {
+  @ManyToOne(() => Pipeline)
+  pipeline: Pipeline;
+
+  @Column({ unique: true, nullable: true })
+  pipelineId: string;
+
+  @ManyToOne(() => Project)
+  project: Project;
+
+  @Column()
+  projectId: string;
+
+  @Column({ comment: 'language defined in type field.' })
+  content: string;
+
+  @Column({ comment: '配置名称' })
+  name: string;
+
+  @Column({
+    type: 'enum',
+    enum: ConfigurationLanguage,
+    comment: 'configuration content language',
+  })
+  language: ConfigurationLanguage;
+}

src/configurations/enums/configuration-language.enum.ts

@@ -0,0 +1,10 @@
+import { registerEnumType } from '@nestjs/graphql';
+
+export enum ConfigurationLanguage {
+  JavaScript = 'JavaScript',
+  YAML = 'YAML',
+}
+
+registerEnumType(ConfigurationLanguage, {
+  name: 'ConfigurationLanguage',
+});

src/main.ts

@@ -1,9 +1,11 @@
+import { PinoLogger } from 'nestjs-pino';
 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';
+import { ServiceRegister } from 'configuration';
 
 async function bootstrap() {
   const app = await NestFactory.create(AppModule, { bodyParser: false });
@@ -14,7 +16,12 @@ async function bootstrap() {
       transform: true,
     }),
   );
-  app.useGlobalFilters(new HttpExceptionFilter());
-  await app.listen(configService.get<number>('http.port'));
+  const httpExceptionFilterLogger = await app.resolve(PinoLogger);
+  app.useGlobalFilters(new HttpExceptionFilter(httpExceptionFilterLogger));
+  const server = await app.listen(configService.get<number>('http.port', 0));
+  const port = server.address().port;
+  const register = new ServiceRegister({ etcd: { hosts: 'http://rpi:2379' } });
+  register.register('fennec/api', `http://localhost:${port}`);
+  register.register('api.fennec', `http://localhost:${port}`);
 }
 bootstrap();

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

@@ -1,4 +1,4 @@
-import { Field, InputType } from '@nestjs/graphql';
+import { InputType } from '@nestjs/graphql';
 import { PipelineUnits } from '../enums/pipeline-units.enum';
 
 @InputType()

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

@@ -1,9 +1,16 @@
-import { InputType, ObjectType } from '@nestjs/graphql';
+import { Field, InputType, Int, ObjectType } from '@nestjs/graphql';
+import { Type } from 'class-transformer';
+import { IsInstance, ValidateNested } from 'class-validator';
 import { WorkUnit } from './work-unit.model';
 
 @InputType('WorkUnitMetadataInput')
 @ObjectType()
 export class WorkUnitMetadata {
+  @Field(() => Int)
   version = 1;
+
+  @Type(() => WorkUnit)
+  @IsInstance(WorkUnit, { each: true })
+  @ValidateNested({ each: true })
   units: WorkUnit[];
 }

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

@@ -1,4 +1,5 @@
 import { Field, InputType, ObjectType } from '@nestjs/graphql';
+import { IsNotEmpty } from 'class-validator';
 import {
   PipelineUnits,
   PipelineUnits as PipelineUnitTypes,
@@ -9,5 +10,7 @@ import {
 export class WorkUnit {
   @Field(() => PipelineUnits)
   type: PipelineUnitTypes;
+
+  @IsNotEmpty({ each: true })
   scripts: string[];
 }

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

@@ -18,6 +18,7 @@ describe('PipelineTaskFlushService', () => {
     const redisClient = {
       rpush: jest.fn(() => Promise.resolve()),
       lrange: jest.fn(() => Promise.resolve()),
+      expire: jest.fn(() => Promise.resolve()),
     };
     const module: TestingModule = await Test.createTestingModule({
       providers: [
@@ -49,17 +50,20 @@ describe('PipelineTaskFlushService', () => {
   });
 
   describe('write', () => {
+    const amqpMsg = {
+      properties: { headers: { sender: 'test' } },
+    } as any;
     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);
+      await service.write(testEvent, amqpMsg);
       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);
+      expect(request).toBeCalledTimes(1);
     });
     it('event for which task done', async () => {
       const testEvent = new PipelineTaskEvent();
@@ -67,13 +71,17 @@ describe('PipelineTaskFlushService', () => {
       testEvent.status = TaskStatuses.success;
       const rpush = jest.spyOn(redisService.getClient(), 'rpush');
       const request = jest.spyOn(amqpConnection, 'request');
-      await service.write(testEvent);
+      await service.write(testEvent, amqpMsg);
       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 },
+        payload: {
+          taskId: 'test',
+          status: TaskStatuses.success,
+          runOn: 'test',
+        },
       });
     });
   });

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

@@ -1,10 +1,10 @@
 import { AmqpConnection, RabbitSubscribe } from '@golevelup/nestjs-rabbitmq';
 import { Injectable } from '@nestjs/common';
+import { ConsumeMessage } from 'amqplib';
 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,
@@ -29,18 +29,27 @@ export class PipelineTaskFlushService {
     queue: getSelfInstanceQueueKey(QUEUE_WRITE_PIPELINE_TASK_LOG),
     queueOptions: {
       autoDelete: true,
+      durable: 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({
+  async write(message: PipelineTaskEvent, amqpMsg: ConsumeMessage) {
+    const client = this.redisService.getClient();
+    await client.rpush(this.getKey(message.taskId), JSON.stringify(message));
+    await client.expire(this.getKey(message.taskId), 600); // ten minutes
+    if (isNil(message.unit)) {
+      try {
+        await this.amqpConnection.request({
           exchange: EXCHANGE_PIPELINE_TASK_TOPIC,
           routingKey: ROUTE_PIPELINE_TASK_DONE,
-        payload: { taskId: message.taskId, status: message.status },
+          payload: {
+            taskId: message.taskId,
+            status: message.status,
+            runOn: amqpMsg.properties.headers.sender,
+          },
         });
+      } catch (error) {
+        console.log(error);
+      }
     }
   }
 

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

@@ -36,4 +36,7 @@ export class PipelineTask extends AppBaseEntity {
 
   @Column({ nullable: true })
   endedAt?: Date;
+
+  @Column({ nullable: true })
+  runOn: string;
 }

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

@@ -42,9 +42,9 @@ describe('PipelineTaskRunner', () => {
       setTimeout(() => {
         logger.handleEvent(event);
       });
-      expect(message$.pipe(take(1), timeout(100)).toPromise()).rejects.toMatch(
-        'timeout',
-      );
+      await expect(
+        message$.pipe(take(1), timeout(100)).toPromise(),
+      ).rejects.toThrow(/timeout/i);
     });
     it('multiple subscribers', async () => {
       const event = new PipelineTaskEvent();
@@ -53,13 +53,16 @@ describe('PipelineTaskRunner', () => {
       const message2$ = logger.getMessage$('test');
       setTimeout(() => {
         logger.handleEvent(event);
+        logger.handleEvent(event);
       });
-      expect(message$.pipe(take(1), timeout(100)).toPromise()).resolves.toEqual(
-        event,
-      );
+      await Promise.all([
+        expect(
+          message$.pipe(take(1), timeout(100)).toPromise(),
+        ).resolves.toEqual(event),
         expect(
           message2$.pipe(take(1), timeout(100)).toPromise(),
-      ).resolves.toEqual(event);
+        ).resolves.toEqual(event),
+      ]);
     });
   });
 

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

@@ -13,7 +13,8 @@ import {
 @Injectable()
 export class PipelineTaskLogger implements OnModuleDestroy {
   private readonly messageSubject = new Subject<PipelineTaskEvent>();
-  private readonly message$: Observable<PipelineTaskEvent> = this.messageSubject.pipe();
+  private readonly message$: Observable<PipelineTaskEvent> =
+    this.messageSubject.pipe();
 
   @RabbitSubscribe({
     exchange: EXCHANGE_PIPELINE_TASK_FANOUT,

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

@@ -1,3 +1,4 @@
+import { DeployByPm2Service } from './runners/deploy-by-pm2/deploy-by-pm2.service';
 import { Test, TestingModule } from '@nestjs/testing';
 import { ReposService } from '../repos/repos.service';
 import { PipelineUnits } from './enums/pipeline-units.enum';
@@ -11,7 +12,7 @@ import { WorkUnitMetadata } from './models/work-unit-metadata.model';
 import { AmqpConnection } from '@golevelup/nestjs-rabbitmq';
 describe('PipelineTaskRunner', () => {
   let runner: PipelineTaskRunner;
-  let reposService: ReposService;
+  let deployByPM2Service: DeployByPm2Service;
 
   beforeEach(async () => {
     const module: TestingModule = await Test.createTestingModule({
@@ -36,11 +37,18 @@ describe('PipelineTaskRunner', () => {
           provide: AmqpConnection,
           useValue: {},
         },
+        {
+          provide: DeployByPm2Service,
+          useValue: {
+            deploy: () => Promise.resolve(),
+          },
+        },
       ],
     }).compile();
 
-    reposService = module.get(ReposService);
+    module.get(ReposService);
     runner = module.get(PipelineTaskRunner);
+    deployByPM2Service = module.get(DeployByPm2Service);
   });
 
   it('should be defined', () => {
@@ -65,7 +73,7 @@ describe('PipelineTaskRunner', () => {
     beforeEach(() => {
       emitEvent = jest
         .spyOn(runner, 'emitEvent')
-        .mockImplementation((..._) => Promise.resolve());
+        .mockImplementation(() => Promise.resolve());
     });
 
     describe('doTask', () => {
@@ -75,10 +83,10 @@ describe('PipelineTaskRunner', () => {
       beforeEach(() => {
         checkout = jest
           .spyOn(runner, 'checkout')
-          .mockImplementation((..._) => Promise.resolve('/null'));
+          .mockImplementation(() => Promise.resolve('/null'));
         doTaskUnit = jest
           .spyOn(runner, 'doTaskUnit')
-          .mockImplementation((..._) => Promise.resolve());
+          .mockImplementation(() => Promise.resolve());
       });
 
       it('only checkout', async () => {
@@ -145,7 +153,7 @@ describe('PipelineTaskRunner', () => {
         await runner.doTask(task);
 
         expect(checkout).toBeCalledTimes(1);
-        expect(doTaskUnit).toBeCalledTimes(2);
+        expect(doTaskUnit).toBeCalledTimes(1);
         expect(emitEvent).toBeCalledTimes(2);
       });
 
@@ -171,9 +179,7 @@ describe('PipelineTaskRunner', () => {
 
         doTaskUnit = jest
           .spyOn(runner, 'doTaskUnit')
-          .mockImplementation((..._) =>
-            Promise.reject(new Error('test error')),
-          );
+          .mockImplementation(() => Promise.reject(new Error('test error')));
         await runner.doTask(task);
 
         expect(checkout).toBeCalledTimes(1);
@@ -189,7 +195,7 @@ describe('PipelineTaskRunner', () => {
       it('success', async () => {
         const runScript = jest
           .spyOn(runner, 'runScript')
-          .mockImplementation((..._) => Promise.resolve());
+          .mockImplementation(() => Promise.resolve());
         const task = new PipelineTask();
 
         const unit = PipelineUnits.test;
@@ -210,9 +216,7 @@ describe('PipelineTaskRunner', () => {
       it('failed', async () => {
         const runScript = jest
           .spyOn(runner, 'runScript')
-          .mockImplementation((..._) =>
-            Promise.reject(new Error('test error')),
-          );
+          .mockImplementation(() => Promise.reject(new Error('test error')));
         const task = new PipelineTask();
 
         const unit = PipelineUnits.test;
@@ -230,7 +234,7 @@ describe('PipelineTaskRunner', () => {
 
     describe('runScript', () => {
       it('normal', async () => {
-        const spawn = jest.fn((..._: any[]) => ({
+        const spawn = jest.fn<any, any>(() => ({
           stdout: {
             on: () => undefined,
           },
@@ -256,7 +260,7 @@ describe('PipelineTaskRunner', () => {
         });
       });
       it('failed', async () => {
-        const spawn = jest.fn((..._: any[]) => ({
+        const spawn = jest.fn(() => ({
           stdout: {
             on: () => undefined,
           },
@@ -291,7 +295,7 @@ describe('PipelineTaskRunner', () => {
             }, 10);
           }, 10);
         });
-        const spawn = jest.fn((..._: any[]) => ({
+        const spawn = jest.fn(() => ({
           stdout: {
             on,
           },
@@ -304,7 +308,7 @@ describe('PipelineTaskRunner', () => {
         }));
 
         let emitSuccessCount = 0;
-        jest.spyOn(runner, 'emitEvent').mockImplementation((..._: any[]) => {
+        jest.spyOn(runner, 'emitEvent').mockImplementation(() => {
           return new Promise((resolve) => {
             setTimeout(() => {
               emitSuccessCount++;
@@ -323,4 +327,25 @@ describe('PipelineTaskRunner', () => {
       });
     });
   });
+
+  describe('tryRunDeployScript', () => {
+    it('should be call deploy with right args', async () => {
+      const deploy = jest.spyOn(deployByPM2Service, 'deploy');
+      await expect(
+        runner['tryRunDeployScript'](
+          '/test/dir',
+          '@@DEPLOY ecosystem.config.js',
+        ),
+      ).resolves.toBe(true);
+      expect(deploy.mock.calls[0][0]).toEqual('/test/dir/ecosystem.config.js');
+    });
+    it('should return false', async () => {
+      await expect(
+        runner['tryRunDeployScript'](
+          '/test/dir',
+          'pm2 start ecosystem.config.js',
+        ),
+      ).resolves.toBe(false);
+    });
+  });
 });

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

@@ -1,3 +1,4 @@
+import { DeployByPm2Service } from './runners/deploy-by-pm2/deploy-by-pm2.service';
 import { ReposService } from '../repos/repos.service';
 import { spawn, ChildProcessWithoutNullStreams } from 'child_process';
 import { PipelineTask } from './pipeline-task.entity';
@@ -5,19 +6,37 @@ import { ApplicationException } from '../commons/exceptions/application.exceptio
 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 {
+  AmqpConnection,
+  RabbitRPC,
+  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_TOPIC,
+  QUEUE_PIPELINE_TASK_KILL,
+  ROUTE_PIPELINE_TASK_KILL,
+} from './pipeline-tasks.constants';
 import {
   EXCHANGE_PIPELINE_TASK_FANOUT,
   ROUTE_PIPELINE_TASK_LOG,
 } from './pipeline-tasks.constants';
+import {
+  getInstanceName,
+  getSelfInstanceQueueKey,
+  getSelfInstanceRouteKey,
+} from '../commons/utils/rabbit-mq';
+import { rm } from 'fs/promises';
+import { rename } from 'fs/promises';
+import { join } from 'path';
 
 type Spawn = typeof spawn;
 
 export class PipelineTaskRunner {
   readonly processes = new Map<string, ChildProcessWithoutNullStreams>();
+  readonly stopTaskIds = new Set<string>();
 
   constructor(
     private readonly reposService: ReposService,
@@ -26,6 +45,7 @@ export class PipelineTaskRunner {
     @Inject('spawn')
     private readonly spawn: Spawn,
     private readonly amqpConnection: AmqpConnection,
+    private readonly deployByPM2Service: DeployByPm2Service,
   ) {}
   @RabbitSubscribe({
     exchange: 'new-pipeline-task',
@@ -40,19 +60,27 @@ export class PipelineTaskRunner {
       this.logger.error({ task, err }, err.message);
     }
   }
-  @RabbitSubscribe({
-    exchange: 'stop-pipeline-task',
-    routingKey: 'mac',
-    queue: 'mac.stop-pipeline-task',
+  @RabbitRPC({
+    exchange: EXCHANGE_PIPELINE_TASK_TOPIC,
+    routingKey: getSelfInstanceRouteKey(ROUTE_PIPELINE_TASK_KILL),
+    queue: getSelfInstanceQueueKey(QUEUE_PIPELINE_TASK_KILL),
+    queueOptions: {
+      autoDelete: true,
+      durable: true,
+    },
   })
   async onStopTask(task: PipelineTask) {
     this.logger.info({ task }, 'on stop task [%s].', task.id);
+    this.stopTaskIds.add(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(() => {
+        setTimeout(() => {
+          this.stopTaskIds.delete(task.id);
+        }, 10_000);
         if (process === this.processes.get(task.id)) {
           this.logger.info({ task }, 'send signal SIGKILL to child process.');
           process.kill('SIGKILL');
@@ -68,6 +96,7 @@ export class PipelineTaskRunner {
     } else {
       this.logger.info({ task }, 'child process is not running.');
     }
+    return true;
   }
 
   async doTask(task: PipelineTask) {
@@ -86,7 +115,7 @@ export class PipelineTaskRunner {
 
     this.logger.info('running task [%s].', task.id);
     try {
-      const workspaceRoot = await this.checkout(task);
+      let workspaceRoot = await this.checkout(task);
       const units = task.units
         .filter((unit) => unit !== PipelineUnits.checkout)
         .map(
@@ -97,6 +126,22 @@ export class PipelineTaskRunner {
         );
       this.logger.info({ units }, 'begin run units.');
       for (const unit of units) {
+        if (unit.type === PipelineUnits.deploy) {
+          const oldRoot = workspaceRoot;
+          workspaceRoot = this.reposService.getDeployRoot(task);
+          if (oldRoot !== workspaceRoot) {
+            await rm(workspaceRoot, { force: true, recursive: true });
+            await rename(oldRoot, workspaceRoot);
+          }
+
+          await this.emitEvent(
+            task,
+            unit.type,
+            TaskStatuses.success,
+            `[deploy] change deploy folder content success`,
+            'stdout',
+          );
+        }
         await this.doTaskUnit(unit.type, unit.scripts, task, workspaceRoot);
       }
       await this.emitEvent(
@@ -137,6 +182,9 @@ export class PipelineTaskRunner {
     try {
       for (const script of scripts) {
         this.logger.debug('begin runScript %s', script);
+        if (this.stopTaskIds.has(task.id)) {
+          throw new ApplicationException('Task is be KILLED');
+        }
         await this.runScript(script, workspaceRoot, task, unit);
         this.logger.debug('end runScript %s', script);
       }
@@ -186,6 +234,7 @@ export class PipelineTaskRunner {
         'checkout failed.',
         'stderr',
       );
+      throw err;
     }
   }
 
@@ -207,7 +256,11 @@ export class PipelineTaskRunner {
       status,
     };
     this.amqpConnection
-      .publish(EXCHANGE_PIPELINE_TASK_FANOUT, ROUTE_PIPELINE_TASK_LOG, event)
+      .publish(EXCHANGE_PIPELINE_TASK_FANOUT, ROUTE_PIPELINE_TASK_LOG, event, {
+        headers: {
+          sender: getInstanceName(),
+        },
+      })
       .catch((error) => {
         this.logger.error(
           { error, event },
@@ -224,6 +277,9 @@ export class PipelineTaskRunner {
     unit: PipelineUnits,
   ): Promise<void> {
     await this.emitEvent(task, unit, TaskStatuses.working, script, 'stdin');
+    if (await this.tryRunDeployScript(workspaceRoot, script)) {
+      return;
+    }
     return new Promise((resolve, reject) => {
       const sub = this.spawn(script, {
         shell: true,
@@ -260,8 +316,24 @@ export class PipelineTaskRunner {
         if (code === 0) {
           return resolve();
         }
+        if (this.stopTaskIds.has(task.id)) {
+          throw reject(new ApplicationException('Task is be KILLED'));
+        }
         return reject(new ApplicationException('exec script failed'));
       });
     });
   }
+
+  private async tryRunDeployScript(workspaceRoot: string, script: string) {
+    const match = /^@@DEPLOY\s+(\S*)/.exec(script);
+    if (match) {
+      await this.deployByPM2Service.deploy(
+        join(workspaceRoot, match[1]),
+        workspaceRoot,
+      );
+      return true;
+    } else {
+      return false;
+    }
+  }
 }

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

@@ -4,3 +4,8 @@ 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';
+export const QUEUE_PIPELINE_TASK_DONE = 'pipeline-task-done';
+export const ROUTE_PIPELINE_TASK_KILL = 'pipeline-task-kill';
+export const QUEUE_PIPELINE_TASK_KILL = 'pipeline-task-kill';
+
+export const PM2_TOKEN = Symbol('pm2-token');

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

@@ -16,13 +16,13 @@ import {
 } from './pipeline-tasks.constants';
 import { PipelineTaskLogger } from './pipeline-task.logger';
 import { PipelineTaskFlushService } from './pipeline-task-flush.service';
+import { DeployByPm2Service } from './runners/deploy-by-pm2/deploy-by-pm2.service';
 
 @Module({
   imports: [
     TypeOrmModule.forFeature([PipelineTask, Pipeline]),
     RedisModule,
     ReposModule,
-
     RabbitMQModule.forRootAsync(RabbitMQModule, {
       imports: [ConfigModule],
       useFactory: (configService: ConfigService) => ({
@@ -36,22 +36,6 @@ import { PipelineTaskFlushService } from './pipeline-task-flush.service';
               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',
@@ -83,6 +67,7 @@ import { PipelineTaskFlushService } from './pipeline-task-flush.service';
       useValue: spawn,
     },
     PipelineTaskFlushService,
+    DeployByPm2Service,
   ],
   exports: [PipelineTasksService],
 })

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

@@ -1,3 +1,4 @@
+import { JwtService } from '@nestjs-lib/auth';
 import { Test, TestingModule } from '@nestjs/testing';
 import { PipelineTaskLogger } from './pipeline-task.logger';
 import { PipelineTasksResolver } from './pipeline-tasks.resolver';
@@ -18,6 +19,10 @@ describe('PipelineTasksResolver', () => {
           provide: PipelineTaskLogger,
           useValue: {},
         },
+        {
+          provide: JwtService,
+          useValue: {},
+        },
       ],
     }).compile();
 

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

@@ -7,7 +7,9 @@ import { plainToClass } from 'class-transformer';
 import { PipelineTaskLogger } from './pipeline-task.logger';
 import { observableToAsyncIterable } from '@graphql-tools/utils';
 import { PipelineTaskEvent } from './models/pipeline-task-event';
+import { Roles, AccountRole } from '@nestjs-lib/auth';
 
+@Roles(AccountRole.admin, AccountRole.super)
 @Resolver()
 export class PipelineTasksResolver {
   constructor(
@@ -33,15 +35,6 @@ export class PipelineTasksResolver {
     );
   }
 
-  @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);
@@ -55,5 +48,7 @@ export class PipelineTasksResolver {
   @Mutation(() => Boolean)
   async stopPipelineTask(@Args('id') id: string) {
     const task = await this.service.findTaskById(id);
+    await this.service.stopTask(task);
+    return true;
   }
 }

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

@@ -4,26 +4,14 @@ 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';
+import { PipelineTaskFlushService } from './pipeline-task-flush.service';
+import { getLoggerToken, PinoLogger } from 'nestjs-pino';
 
 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({
@@ -41,12 +29,19 @@ describe('PipelineTasksService', () => {
           provide: AmqpConnection,
           useValue: {},
         },
+        {
+          provide: PipelineTaskFlushService,
+          useValue: {},
+        },
+        {
+          provide: getLoggerToken(PipelineTasksService.name),
+          useValue: new PinoLogger({}),
+        },
       ],
     }).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);

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

@@ -1,13 +1,22 @@
-import { Injectable } from '@nestjs/common';
+import { BadRequestException, 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');
+import { AmqpConnection, RabbitRPC } from '@golevelup/nestjs-rabbitmq';
+import {
+  EXCHANGE_PIPELINE_TASK_TOPIC,
+  QUEUE_PIPELINE_TASK_DONE,
+  ROUTE_PIPELINE_TASK_DONE,
+} from './pipeline-tasks.constants';
+import { PipelineTaskFlushService } from './pipeline-task-flush.service';
+import { find, isNil, propEq } from 'ramda';
+import { PipelineTaskLogs } from './models/pipeline-task-logs.model';
+import { TaskStatuses, terminalTaskStatuses } from './enums/task-statuses.enum';
+import { InjectPinoLogger, PinoLogger } from 'nestjs-pino';
+import { getAppInstanceRouteKey } from '../commons/utils/rabbit-mq';
+import { ROUTE_PIPELINE_TASK_KILL } from './pipeline-tasks.constants';
 
 @Injectable()
 export class PipelineTasksService {
@@ -17,6 +26,9 @@ export class PipelineTasksService {
     @InjectRepository(Pipeline)
     private readonly pipelineRepository: Repository<Pipeline>,
     private readonly amqpConnection: AmqpConnection,
+    private readonly eventFlushService: PipelineTaskFlushService,
+    @InjectPinoLogger(PipelineTasksService.name)
+    private readonly logger: PinoLogger,
   ) {}
   async addTask(dto: CreatePipelineTaskInput) {
     const pipeline = await this.pipelineRepository.findOneOrFail({
@@ -60,4 +72,84 @@ export class PipelineTasksService {
   getRedisTokens(pipeline: Pipeline): [string, string] {
     return [`pipeline-${pipeline.id}:lck`, `pipeline-${pipeline.id}:tasks`];
   }
+
+  @RabbitRPC({
+    exchange: EXCHANGE_PIPELINE_TASK_TOPIC,
+    routingKey: ROUTE_PIPELINE_TASK_DONE,
+    queue: QUEUE_PIPELINE_TASK_DONE,
+    queueOptions: {
+      autoDelete: true,
+      durable: true,
+    },
+  })
+  async updateByEvent({ taskId, runOn }: { taskId: string; runOn: string }) {
+    try {
+      const [events, task] = await Promise.all([
+        this.eventFlushService.read(taskId),
+        this.findTaskById(taskId),
+      ]);
+      this.logger.info('[updateByEvent] start. taskId: %s', taskId);
+
+      for (const event of events) {
+        if (isNil(event.unit)) {
+          if (
+            event.status !== TaskStatuses.pending &&
+            task.status === TaskStatuses.pending
+          ) {
+            task.startedAt = event.emittedAt;
+          } else if (terminalTaskStatuses.includes(event.status)) {
+            task.endedAt = event.emittedAt;
+          }
+          task.status = event.status;
+        } else {
+          let l: PipelineTaskLogs = find<PipelineTaskLogs>(
+            propEq('unit', event.unit),
+            task.logs,
+          );
+
+          if (isNil(l)) {
+            l = {
+              unit: event.unit,
+              startedAt: event.emittedAt,
+              endedAt: null,
+              logs: event.message,
+              status: event.status,
+            };
+
+            task.logs.push(l);
+          } else {
+            l.logs += event.message;
+          }
+
+          if (terminalTaskStatuses.includes(event.status)) {
+            l.endedAt = event.emittedAt;
+          }
+          l.status = event.status;
+        }
+      }
+      task.runOn = runOn;
+      await this.repository.update({ id: taskId }, task);
+      this.logger.info('[updateByEvent] success. taskId: %s', taskId);
+      return task;
+    } catch (error) {
+      this.logger.error(
+        { error },
+        '[updateByEvent] failed. taskId: %s',
+        taskId,
+      );
+    }
+  }
+
+  async stopTask(task: PipelineTask) {
+    if (isNil(task.runOn)) {
+      throw new BadRequestException(
+        "the task have not running instance on database. field 'runOn' is nil",
+      );
+    }
+    await this.amqpConnection.request({
+      exchange: EXCHANGE_PIPELINE_TASK_TOPIC,
+      routingKey: getAppInstanceRouteKey(ROUTE_PIPELINE_TASK_KILL, task.runOn),
+      payload: task,
+    });
+  }
 }

src/pipeline-tasks/runners/deploy-by-pm2/deploy-by-pm2.service.spec.ts

@@ -0,0 +1,117 @@
+import { Test, TestingModule } from '@nestjs/testing';
+import { getLoggerToken, PinoLogger } from 'nestjs-pino';
+import { join } from 'path';
+import { DeployByPm2Service } from './deploy-by-pm2.service';
+
+describe('DeployByPm2Service', () => {
+  let service: DeployByPm2Service;
+
+  beforeEach(async () => {
+    const module: TestingModule = await Test.createTestingModule({
+      providers: [
+        DeployByPm2Service,
+        {
+          provide: getLoggerToken(DeployByPm2Service.name),
+          useValue: new PinoLogger({
+            pinoHttp: {
+              level: 'silent',
+            },
+          }),
+        },
+      ],
+    }).compile();
+
+    service = module.get<DeployByPm2Service>(DeployByPm2Service);
+  });
+
+  it('should be defined', () => {
+    expect(service).toBeDefined();
+  });
+
+  describe('getAppsSn', () => {
+    it('should return right value', () => {
+      expect(
+        service['getAppsSn']([
+          { name: 'app' },
+          { name: 'app#4' },
+          { name: 'app#1' },
+        ]),
+      ).toEqual(4 + 1);
+    });
+    it('should return 1 when no match', () => {
+      expect(
+        service['getAppsSn']([
+          { name: 'bar' },
+          { name: 'foo#4' },
+          { name: 'foo#1' },
+        ]),
+      ).toEqual(4 + 1);
+    });
+  });
+
+  describe('filterOldApps', () => {
+    it('should return right value', () => {
+      expect(
+        service['filterOldApps'](
+          [{ name: 'app' }],
+          [
+            { name: 'app' },
+            { name: 'app#4' },
+            { name: 'foo#2' },
+            { name: 'bar' },
+          ],
+        ),
+      ).toEqual([{ name: 'app' }, { name: 'app#4' }]);
+    });
+    it('should return [] when no match', () => {
+      expect(
+        service['filterOldApps'](
+          [{ name: 'app' }],
+          [{ name: 'foo#2' }, { name: 'bar' }],
+        ),
+      ).toEqual([]);
+    });
+  });
+
+  describe('replaceAppName', () => {
+    it('should be replaced with right value', () => {
+      const getAppsSn = jest
+        .spyOn(service, 'getAppsSn' as any)
+        .mockImplementation(() => 1);
+      const options = [{ name: 'app' }, { name: 'foo' }];
+      service['replaceAppName'](options, []);
+      expect(options).toEqual([{ name: 'app#1' }, { name: 'foo#1' }]);
+      expect(getAppsSn).toBeCalledTimes(1);
+      expect(getAppsSn.mock.calls[0][0]).toEqual([]);
+    });
+  });
+
+  describe('deploy', () => {
+    it('should be success', async () => {
+      const filterOldApps = jest
+        .spyOn(service, 'filterOldApps' as any)
+        .mockImplementation(() => [{ name: 'app#1' }, { name: 'app#2' }]);
+      const replaceAppName = jest
+        .spyOn(service, 'replaceAppName' as any)
+        .mockImplementation((options) => (options[0].name = 'app#2'));
+      const stopApps = jest
+        .spyOn(service, 'stopApps' as any)
+        .mockImplementation(() => Promise.resolve());
+      await expect(
+        service['deploy'](
+          join(
+            __dirname,
+            '../../../../test/__mocks__/deploy-service/ecosystem.config.js',
+          ),
+          join(__dirname, '../../../../test/__mocks__/deploy-service'),
+        ),
+      ).resolves.toBeFalsy();
+      expect(filterOldApps).toBeCalledTimes(1);
+      expect(replaceAppName).toBeCalledTimes(1);
+      expect(stopApps).toBeCalledTimes(1);
+      stopApps.mockReset();
+
+      await service['stopApps']([{ name: 'app#2' }]);
+    }, 10_000);
+  });
+});

src/pipeline-tasks/runners/deploy-by-pm2/deploy-by-pm2.service.ts

@@ -0,0 +1,98 @@
+import { InjectPinoLogger, PinoLogger } from 'nestjs-pino';
+import { Injectable } from '@nestjs/common';
+import { promisify } from 'util';
+import * as pm2 from 'pm2';
+import { Proc, ProcessDescription, StartOptions } from 'pm2';
+import { clone, last } from 'ramda';
+
+@Injectable()
+export class DeployByPm2Service {
+  constructor(
+    @InjectPinoLogger(DeployByPm2Service.name)
+    private readonly logger: PinoLogger,
+  ) {}
+  async deploy(filePath: string, workspace: string) {
+    const baseConfig: { apps: StartOptions[] } = await import(filePath);
+    const appOptionsList: StartOptions[] = clone(baseConfig.apps);
+
+    await promisify<void>(pm2.connect.bind(pm2))();
+    const allApps = await promisify(pm2.list.bind(pm2))();
+    try {
+      if (!Array.isArray(baseConfig.apps)) {
+        this.logger.error(
+          'the "apps" in the PM2 ecosystem configuration is not array',
+        );
+        throw new Error('apps is not array');
+      }
+
+      const oldApps = this.filterOldApps(appOptionsList, allApps);
+
+      this.replaceAppName(appOptionsList, oldApps);
+      for (const appOptions of appOptionsList) {
+        const proc = await promisify<StartOptions, Proc>(pm2.start.bind(pm2))({
+          ...appOptions,
+          cwd: workspace,
+        });
+        this.logger.info({ proc }, `start ${appOptions.name}`);
+      }
+      await this.stopApps(oldApps);
+    } catch (err) {
+      await this.stopApps(appOptionsList);
+      throw err;
+    } finally {
+      pm2.disconnect();
+    }
+  }
+
+  private async stopApps(apps: ProcessDescription[] | StartOptions[]) {
+    await Promise.all(
+      apps.map(async (app: ProcessDescription | StartOptions) => {
+        let procAtStop: ProcessDescription;
+        let procAtDelete: ProcessDescription;
+        try {
+          const idOrName = 'pm_id' in app ? app.pm_id : app.name;
+          procAtStop = await promisify(pm2.stop.bind(pm2))(idOrName);
+          procAtDelete = await promisify(pm2.delete.bind(pm2))(idOrName);
+          this.logger.info('stop & delete %s success', app.name);
+        } catch (error) {
+          this.logger.error(
+            { error, procAtStop, procAtDelete },
+            'stop & delete %s error',
+            app.name,
+          );
+        }
+      }),
+    );
+  }
+
+  private replaceAppName(
+    optionsList: StartOptions[],
+    oldApps: ProcessDescription[],
+  ) {
+    const appSn = this.getAppsSn(oldApps);
+
+    optionsList.forEach((options) => {
+      if (!options.name) {
+        this.logger.error('please give a name for application');
+        throw new Error('app name is not given');
+      }
+      options.name = `${options.name}#${appSn}`;
+    });
+  }
+
+  private filterOldApps(
+    optionsList: StartOptions[],
+    apps: ProcessDescription[],
+  ) {
+    return apps.filter((app) =>
+      optionsList.some((options) => app.name.split('#')[0] === options.name),
+    );
+  }
+
+  private getAppsSn(oldApps: ProcessDescription[]) {
+    const appsSn: number[] = oldApps.map(
+      (app) => +(app.name.split('#')?.[1] ?? 0),
+    );
+    return (last(appsSn.sort()) ?? 0) + 1;
+  }
+}

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

@@ -1,3 +1,4 @@
+import { JwtService } from '@nestjs-lib/auth';
 import { Test, TestingModule } from '@nestjs/testing';
 import { PipelineTasksService } from '../pipeline-tasks/pipeline-tasks.service';
 import { CommitLogsResolver } from './commit-logs.resolver';
@@ -18,6 +19,10 @@ describe('CommitLogsResolver', () => {
           provide: PipelineTasksService,
           useValue: {},
         },
+        {
+          provide: JwtService,
+          useValue: {},
+        },
       ],
     }).compile();
 

src/pipelines/commit-logs.resolver.ts

@@ -1,3 +1,4 @@
+import { Roles, AccountRole } from '@nestjs-lib/auth';
 import { Query } from '@nestjs/graphql';
 import {
   Args,
@@ -10,6 +11,7 @@ import { PipelineTasksService } from '../pipeline-tasks/pipeline-tasks.service';
 import { Commit, LogFields } from '../repos/dtos/log-list.model';
 import { PipelinesService } from './pipelines.service';
 
+@Roles(AccountRole.admin, AccountRole.super)
 @Resolver(() => Commit)
 export class CommitLogsResolver {
   constructor(

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

@@ -1,11 +1,13 @@
+import { Type } from 'class-transformer';
 import { InputType } from '@nestjs/graphql';
 import { WorkUnitMetadata } from '../../pipeline-tasks/models/work-unit-metadata.model';
 import {
-  IsObject,
+  IsInstance,
   IsOptional,
   IsString,
   IsUUID,
   MaxLength,
+  ValidateNested,
 } from 'class-validator';
 
 @InputType({ isAbstract: true })
@@ -21,7 +23,13 @@ export class CreatePipelineInput {
   @MaxLength(32)
   name: string;
 
+  @Type(() => WorkUnitMetadata)
   @IsOptional()
-  @IsObject()
+  @ValidateNested()
+  @IsInstance(WorkUnitMetadata)
   workUnitMetadata: WorkUnitMetadata;
+
+  @IsString()
+  @MaxLength(100)
+  environment: string;
 }

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

@@ -1,7 +1,9 @@
-import { InputType } from '@nestjs/graphql';
+import { InputType, OmitType } from '@nestjs/graphql';
 import { CreatePipelineInput } from './create-pipeline.input';
 
 @InputType()
-export class UpdatePipelineInput extends CreatePipelineInput {
+export class UpdatePipelineInput extends OmitType(CreatePipelineInput, [
+  'projectId',
+]) {
   id: string;
 }

src/pipelines/pipeline.entity.ts

@@ -12,7 +12,7 @@ export class Pipeline extends AppBaseEntity {
   @Column()
   projectId: string;
 
-  @Column({ comment: 'eg: remotes/origin/master' })
+  @Column({ comment: 'E.g., remotes/origin/master' })
   branch: string;
 
   @Column()
@@ -20,4 +20,7 @@ export class Pipeline extends AppBaseEntity {
 
   @Column({ type: 'jsonb' })
   workUnitMetadata: WorkUnitMetadata;
+
+  @Column()
+  environment: string;
 }

src/pipelines/pipelines.module.ts

@@ -5,7 +5,6 @@ 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';
 

src/pipelines/pipelines.resolver.spec.ts

@@ -1,3 +1,4 @@
+import { JwtService } from '@nestjs-lib/auth';
 import { Test, TestingModule } from '@nestjs/testing';
 import { PipelinesResolver } from './pipelines.resolver';
 import { PipelinesService } from './pipelines.service';
@@ -13,6 +14,10 @@ describe('PipelinesResolver', () => {
           provide: PipelinesService,
           useValue: {},
         },
+        {
+          provide: JwtService,
+          useValue: {},
+        },
       ],
     }).compile();
 

src/pipelines/pipelines.resolver.ts

@@ -4,7 +4,9 @@ import { UpdatePipelineInput } from './dtos/update-pipeline.input';
 import { Pipeline } from './pipeline.entity';
 import { PipelinesService } from './pipelines.service';
 import { ListPipelineArgs } from './dtos/list-pipelines.args';
+import { Roles, AccountRole } from '@nestjs-lib/auth';
 
+@Roles(AccountRole.admin, AccountRole.super)
 @Resolver()
 export class PipelinesResolver {
   constructor(private readonly service: PipelinesService) {}
@@ -21,14 +23,14 @@ export class PipelinesResolver {
   @Mutation(() => Pipeline)
   async createPipeline(
     @Args('pipeline', { type: () => CreatePipelineInput })
-    dto: UpdatePipelineInput,
+    dto: CreatePipelineInput,
   ) {
     return await this.service.create(dto);
   }
 
   @Mutation(() => Pipeline)
   async updatePipeline(
-    @Args('Pipeline', { type: () => UpdatePipelineInput })
+    @Args('pipeline', { type: () => UpdatePipelineInput })
     dto: UpdatePipelineInput,
   ) {
     const tmp = await this.service.update(dto);

src/pipelines/pipelines.service.spec.ts

@@ -2,13 +2,11 @@ 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 () => {
@@ -40,7 +38,6 @@ describe('PipelinesService', () => {
     }).compile();
 
     service = module.get<PipelinesService>(PipelinesService);
-    repository = module.get(getRepositoryToken(Pipeline));
   });
 
   it('should be defined', () => {

src/pipelines/pipelines.service.ts

@@ -14,10 +14,14 @@ import {
 import { AmqpConnection } from '@golevelup/nestjs-rabbitmq';
 import { Commit } from '../repos/dtos/log-list.model';
 import { getAppInstanceRouteKey } from '../commons/utils/rabbit-mq';
+import { ApplicationException } from '../commons/exceptions/application.exception';
+import { plainToClass } from 'class-transformer';
 
 @Injectable()
 export class PipelinesService extends BaseDbService<Pipeline> {
-  readonly uniqueFields: Array<Array<keyof Pipeline>> = [['projectId', 'name']];
+  readonly uniqueFields: Array<Array<keyof Pipeline>> = [
+    ['projectId', 'name', 'environment'],
+  ];
   constructor(
     @InjectRepository(Pipeline)
     readonly repository: Repository<Pipeline>,
@@ -42,8 +46,8 @@ export class PipelinesService extends BaseDbService<Pipeline> {
   }
 
   async update(dto: UpdatePipelineInput) {
-    await this.isDuplicateEntityForUpdate(dto.id, dto);
     const old = await this.findOne(dto.id);
+    await this.isDuplicateEntityForUpdate(old, dto);
     return await this.repository.save(this.repository.merge(old, dto));
   }
 
@@ -55,22 +59,22 @@ export class PipelinesService extends BaseDbService<Pipeline> {
       exchange: EXCHANGE_REPO,
       routingKey: getAppInstanceRouteKey(ROUTE_FETCH, appInstance),
       payload: pipeline,
-      timeout: 30_000,
+      timeout: 120_000,
     });
   }
   async listCommits(pipeline: Pipeline) {
     return await this.amqpConnection
-      .request<Commit[]>({
+      .request<[Error, Commit[]]>({
         exchange: EXCHANGE_REPO,
         routingKey: ROUTE_LIST_COMMITS,
         payload: pipeline,
-        timeout: 10_000,
+        timeout: 30_000,
       })
-      .then((list) =>
-        list.map((item) => {
-          item.date = new Date(item.date);
-          return item;
-        }),
-      );
+      .then(([error, list]) => {
+        if (error) {
+          throw new ApplicationException(error);
+        }
+        return plainToClass(Commit, list);
+      });
   }
 }

src/projects/project.entity.ts

@@ -1,5 +1,5 @@
 import { ObjectType } from '@nestjs/graphql';
-import { Entity, Column, DeleteDateColumn } from 'typeorm';
+import { Entity, Column } from 'typeorm';
 import { AppBaseEntity } from '../commons/entities/app-base-entity';
 
 @ObjectType()

src/projects/projects.constants.ts

@@ -0,0 +1,3 @@
+export const EXCHANGE_PROJECT_TOPIC = 'project.topic';
+export const EXCHANGE_PROJECT_FANOUT = 'project.fanout';
+export const ROUTE_PROJECT_CHANGE = 'project-change';

src/projects/projects.module.ts

@@ -3,9 +3,31 @@ import { ProjectsService } from './projects.service';
 import { ProjectsResolver } from './projects.resolver';
 import { TypeOrmModule } from '@nestjs/typeorm';
 import { Project } from './project.entity';
+import { RabbitMQModule } from '@golevelup/nestjs-rabbitmq';
+import { ConfigModule, ConfigService } from '@nestjs/config';
+import { EXCHANGE_PROJECT_FANOUT } from './projects.constants';
 
 @Module({
-  imports: [TypeOrmModule.forFeature([Project])],
+  imports: [
+    TypeOrmModule.forFeature([Project]),
+    RabbitMQModule.forRootAsync(RabbitMQModule, {
+      imports: [ConfigModule],
+      useFactory: (configService: ConfigService) => ({
+        uri: configService.get<string>('db.rabbitmq.uri'),
+        exchanges: [
+          {
+            name: EXCHANGE_PROJECT_FANOUT,
+            type: 'fanout',
+            options: {
+              durable: false,
+              autoDelete: true,
+            },
+          },
+        ],
+      }),
+      inject: [ConfigService],
+    }),
+  ],
   providers: [ProjectsService, ProjectsResolver],
   exports: [ProjectsService],
 })

src/projects/projects.resolver.spec.ts

@@ -1,3 +1,4 @@
+import { JwtService } from '@nestjs-lib/auth';
 import { Test, TestingModule } from '@nestjs/testing';
 import { ProjectsResolver } from './projects.resolver';
 import { ProjectsService } from './projects.service';
@@ -13,6 +14,10 @@ describe('ProjectsResolver', () => {
           provide: ProjectsService,
           useValue: {},
         },
+        {
+          provide: JwtService,
+          useValue: {},
+        },
       ],
     }).compile();
 

src/projects/projects.resolver.ts

@@ -1,9 +1,11 @@
+import { AccountRole, Roles } from '@nestjs-lib/auth';
 import { Args, Mutation, Query, Resolver } from '@nestjs/graphql';
 import { CreateProjectInput } from './dtos/create-project.input';
 import { UpdateProjectInput } from './dtos/update-project.input';
 import { Project } from './project.entity';
 import { ProjectsService } from './projects.service';
 
+@Roles(AccountRole.admin, AccountRole.super)
 @Resolver(() => Project)
 export class ProjectsResolver {
   constructor(private readonly service: ProjectsService) {}
@@ -20,7 +22,7 @@ export class ProjectsResolver {
   @Mutation(() => Project)
   async createProject(
     @Args('project', { type: () => CreateProjectInput })
-    dto: UpdateProjectInput,
+    dto: CreateProjectInput,
   ) {
     return await this.service.create(dto);
   }

src/projects/projects.service.spec.ts

@@ -2,6 +2,7 @@ import { Test, TestingModule } from '@nestjs/testing';
 import { ProjectsService } from './projects.service';
 import { getRepositoryToken } from '@nestjs/typeorm';
 import { Project } from './project.entity';
+import { AmqpConnection } from '@golevelup/nestjs-rabbitmq';
 
 describe('ProjectsService', () => {
   let service: ProjectsService;
@@ -14,6 +15,10 @@ describe('ProjectsService', () => {
           provide: getRepositoryToken(Project),
           useValue: {},
         },
+        {
+          provide: AmqpConnection,
+          useValue: {},
+        },
       ],
     }).compile();
 

src/projects/projects.service.ts

@@ -5,6 +5,11 @@ import { Repository } from 'typeorm';
 import { CreateProjectInput } from './dtos/create-project.input';
 import { Project } from './project.entity';
 import { UpdateProjectInput } from './dtos/update-project.input';
+import { AmqpConnection } from '@golevelup/nestjs-rabbitmq';
+import {
+  EXCHANGE_PROJECT_FANOUT,
+  ROUTE_PROJECT_CHANGE,
+} from './projects.constants';
 
 @Injectable()
 export class ProjectsService extends BaseDbService<Project> {
@@ -12,6 +17,7 @@ export class ProjectsService extends BaseDbService<Project> {
   constructor(
     @InjectRepository(Project)
     readonly repository: Repository<Project>,
+    private readonly amqpConnection: AmqpConnection,
   ) {
     super(repository);
   }
@@ -28,7 +34,12 @@ export class ProjectsService extends BaseDbService<Project> {
   async update(dto: UpdateProjectInput) {
     await this.isDuplicateEntityForUpdate(dto.id, dto);
     const old = await this.findOne(dto.id);
-    return await this.repository.save(this.repository.merge(old, dto));
+    const project = await this.repository.save(this.repository.merge(old, dto));
+    this.amqpConnection.publish(EXCHANGE_PROJECT_FANOUT, ROUTE_PROJECT_CHANGE, [
+      project,
+      old,
+    ]);
+    return project;
   }
 
   async remove(id: string) {

src/repos/dtos/branch-list.model.ts

@@ -1,10 +1,5 @@
 import { ObjectType, Field } from '@nestjs/graphql';
-import {
-  LogResult,
-  DefaultLogFields,
-  BranchSummary,
-  BranchSummaryBranch,
-} from 'simple-git';
+import { BranchSummaryBranch } from 'simple-git';
 
 @ObjectType()
 export class Branch implements BranchSummaryBranch {

src/repos/dtos/log-list.model.ts

@@ -1,10 +1,12 @@
 import { ObjectType, Field } from '@nestjs/graphql';
+import { Type } from 'class-transformer';
 import { LogResult, DefaultLogFields } from 'simple-git';
 import { PipelineTask } from '../../pipeline-tasks/pipeline-task.entity';
 
 @ObjectType()
 export class Commit {
   hash: string;
+  @Type(() => Date)
   date: Date;
   message: string;
   refs: string;

src/repos/repos.constants.ts

@@ -3,3 +3,4 @@ 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';
+export const QUEUE_REFRESH_REPO = 'refresh-repo';

src/repos/repos.service.spec.ts

@@ -12,6 +12,7 @@ import { readFile } from 'fs/promises';
 import { getLoggerToken, PinoLogger } from 'nestjs-pino';
 import { Nack } from '@golevelup/nestjs-rabbitmq';
 import { getInstanceName } from '../commons/utils/rabbit-mq';
+import { RedisMutexService } from '../commons/redis-mutex/redis-mutex.service';
 
 const getTest1Project = () =>
   ({
@@ -52,6 +53,14 @@ describe('ReposService', () => {
           provide: getLoggerToken(ReposService.name),
           useValue: new PinoLogger({}),
         },
+        {
+          provide: RedisMutexService,
+          useValue: {
+            lock: jest.fn(() =>
+              Promise.resolve(() => Promise.resolve(undefined)),
+            ),
+          },
+        },
       ],
     }).compile();
 
@@ -152,7 +161,7 @@ describe('ReposService', () => {
       const project = new Project();
       const pipeline = new Pipeline();
       pipeline.branch = 'test';
-      const fetch = jest.fn((_: any) => Promise.resolve());
+      const fetch = jest.fn<any, [any]>(() => Promise.resolve());
       pipeline.project = project;
       const getGit = jest.spyOn(service, 'getGit').mockImplementation(() =>
         Promise.resolve({
@@ -173,7 +182,7 @@ describe('ReposService', () => {
       const project = new Project();
       const pipeline = new Pipeline();
       pipeline.branch = 'test';
-      const fetch = jest.fn((_: any) => Promise.resolve());
+      const fetch = jest.fn<any, [any]>(() => Promise.resolve());
       pipeline.project = project;
       const getGit = jest
         .spyOn(service, 'getGit')
@@ -187,7 +196,7 @@ describe('ReposService', () => {
       const project = new Project();
       const pipeline = new Pipeline();
       pipeline.branch = 'test';
-      const fetch = jest.fn((_: any) => Promise.reject('error'));
+      const fetch = jest.fn<any, [any]>(() => Promise.reject('error'));
       pipeline.project = project;
       const getGit = jest.spyOn(service, 'getGit').mockImplementation(() =>
         Promise.resolve({

src/repos/repos.service.ts

@@ -11,13 +11,14 @@ import { Project } from '../projects/project.entity';
 import { ListBranchesArgs } from './dtos/list-branches.args';
 import { ConfigService } from '@nestjs/config';
 import { Commit } from './dtos/log-list.model';
-import { Nack, RabbitRPC } from '@golevelup/nestjs-rabbitmq';
+import { Nack, RabbitRPC, RabbitSubscribe } from '@golevelup/nestjs-rabbitmq';
 import { Pipeline } from '../pipelines/pipeline.entity';
-import { InjectPinoLogger, Logger } from 'nestjs-pino';
+import { InjectPinoLogger, PinoLogger } from 'nestjs-pino';
 import {
   EXCHANGE_REPO,
   QUEUE_FETCH,
   QUEUE_LIST_COMMITS,
+  QUEUE_REFRESH_REPO,
   ROUTE_FETCH,
   ROUTE_LIST_COMMITS,
 } from './repos.constants';
@@ -26,6 +27,13 @@ import {
   getInstanceName,
   getSelfInstanceRouteKey,
 } from '../commons/utils/rabbit-mq';
+import { ApplicationException } from '../commons/exceptions/application.exception';
+import {
+  EXCHANGE_PROJECT_FANOUT,
+  ROUTE_PROJECT_CHANGE,
+} from '../projects/projects.constants';
+import { RedisMutexService } from '../commons/redis-mutex/redis-mutex.service';
+import { rm } from 'fs/promises';
 
 const DEFAULT_REMOTE_NAME = 'origin';
 const INFO_PATH = '@info';
@@ -36,7 +44,8 @@ export class ReposService {
     private readonly projectRepository: Repository<Project>,
     private readonly configService: ConfigService,
     @InjectPinoLogger(ReposService.name)
-    private readonly logger: Logger,
+    private readonly logger: PinoLogger,
+    private readonly redisMutexService: RedisMutexService,
   ) {}
 
   getWorkspaceRoot(project: Project): string {
@@ -47,6 +56,14 @@ export class ReposService {
     );
   }
 
+  getDeployRoot(task: PipelineTask) {
+    return join(
+      this.configService.get<string>('workspaces.root'),
+      encodeURIComponent(task.pipeline.project.name),
+      encodeURIComponent(`deploy-${task.pipeline.name}`),
+    );
+  }
+
   async getGit(
     project: Project,
     workspaceRoot?: string,
@@ -129,7 +146,7 @@ export class ReposService {
       autoDelete: true,
     },
   })
-  async listCommits(pipeline: Pipeline): Promise<Commit[] | Nack> {
+  async listCommits(pipeline: Pipeline): Promise<[Error, Commit[]?]> {
     const git = await this.getGit(pipeline.project, undefined, {
       fetch: false,
     });
@@ -140,20 +157,23 @@ export class ReposService {
         `remotes/origin/${pipeline.branch}`,
         '--',
       ]);
-      return data.all.map(
+      return [
+        null,
+        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();
+      return [new ApplicationException(error)];
     }
   }
 
@@ -166,6 +186,9 @@ export class ReposService {
     },
   })
   async fetch(pipeline: Pipeline): Promise<string | null | Nack> {
+    const unlock = await this.redisMutexService.lock(
+      `repo-project-${pipeline.projectId}`,
+    );
     try {
       const git = await this.getGit(pipeline.project, undefined, {
         fetch: false,
@@ -175,6 +198,43 @@ export class ReposService {
     } catch (error) {
       this.logger.error({ error, pipeline }, '[fetch] %s', error?.message);
       return new Nack();
+    } finally {
+      await unlock();
+    }
+  }
+
+  @RabbitSubscribe({
+    exchange: EXCHANGE_PROJECT_FANOUT,
+    routingKey: ROUTE_PROJECT_CHANGE,
+    queue: QUEUE_REFRESH_REPO,
+    queueOptions: {
+      autoDelete: true,
+      durable: true,
+    },
+  })
+  async refreshRepo([project]: [Project]) {
+    this.logger.info({ project }, '[refreshRepo] start');
+    const unlock = await this.redisMutexService.lock(
+      `repo-project-${project.id}`,
+      {
+        timeout: null,
+      },
+    );
+    try {
+      const path = join(
+        this.configService.get<string>('workspaces.root'),
+        encodeURIComponent(project.name),
+      );
+      await rm(path, { recursive: true });
+      this.logger.info({ project }, '[refreshRepo] success');
+    } catch (error) {
+      this.logger.error(
+        { project, error },
+        '[refreshRepo] failed. $s',
+        error.message,
+      );
+    } finally {
+      await unlock();
     }
   }
 }

src/webhooks/webhooks.module.ts

@@ -1,4 +1,4 @@
-import { MiddlewareConsumer, Module } from '@nestjs/common';
+import { Module } from '@nestjs/common';
 import { TypeOrmModule } from '@nestjs/typeorm';
 import { PipelineTasksModule } from '../pipeline-tasks/pipeline-tasks.module';
 import { GiteaWebhooksController } from './gitea-webhooks.controller';
@@ -10,5 +10,4 @@ import { WebhooksService } from './webhooks.service';
   controllers: [GiteaWebhooksController],
   providers: [WebhooksService],
 })
-export class WebhooksModule {
-}
+export class WebhooksModule {}

test/__mocks__/deploy-service/ecosystem.config.js

@@ -0,0 +1,15 @@
+module.exports = {
+  apps: [
+    {
+      name: 'app',
+      script: __dirname + '/index.js',
+      watch: false,
+      ignore_watch: ['node_modules'],
+      log_date_format: 'MM-DD HH:mm:ss.SSS Z',
+      env: {},
+      max_restarts: 5,
+      kill_timeout: 10_000,
+      wait_ready: true,
+    },
+  ],
+};

test/__mocks__/deploy-service/index.js

@@ -0,0 +1,23 @@
+import { createServer } from 'http';
+
+var app = createServer(function (req, res) {
+  res.writeHead(200);
+  setTimeout(() => {
+    res.end('hey');
+  }, 2000);
+});
+
+var listener = app.listen(0, function () {
+  console.log('Listening on port ' + listener.address().port);
+  setTimeout(() => {
+    // Here we send the ready signal to PM2
+    process.send('ready');
+  }, 5000);
+});
+
+process.on('SIGINT', function () {
+  listener.close();
+  setTimeout(() => {
+    process.exit(0);
+  }, 2000);
+});

test/__mocks__/deploy-service/package.json

@@ -0,0 +1,10 @@
+{
+  "name": "deploy-service",
+  "version": "1.0.0",
+  "description": "For Test",
+  "main": "index.js",
+  "type": "module",
+  "scripts": {},
+  "author": "Ivan Li",
+  "license": "ISC"
+}

test/__mocks__/deploy-service/test.js

@@ -0,0 +1,16 @@
+import { createServer } from 'http';
+
+var app = createServer(function (req, res) {
+  res.writeHead(200);
+  setTimeout(() => {
+    res.end('pm2');
+  }, 2000);
+});
+
+var listener = app.listen(33333, function () {
+  console.log('Listening on port ' + listener.address().port);
+  setTimeout(() => {
+    // Here we send the ready signal to PM2
+    process.send('ready');
+  }, 5000);
+});

tsconfig.json

@@ -7,9 +7,9 @@
     "experimentalDecorators": true,
     "allowSyntheticDefaultImports": true,
     "target": "es2017",
+    "lib": ["ES2021"],
     "sourceMap": true,
     "outDir": "./dist",
     "baseUrl": "./",
-    "incremental": true
-  }
+  },
 }