feat:v0.1.0

.vscode/settings.json

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

package-lock.json

@@ -16,6 +16,7 @@
         "@nestjs/graphql": "^7.9.8",
         "@nestjs/platform-express": "^7.5.1",
         "@nestjs/typeorm": "^7.1.5",
+        "@types/amqplib": "^0.8.0",
         "@types/bull": "^3.15.0",
         "@types/ramda": "^0.27.38",
         "apollo-server-express": "^2.19.2",
@@ -2758,6 +2759,15 @@
         "@types/node": "*"
       }
     },
+    "node_modules/@types/amqplib": {
+      "version": "0.8.0",
+      "resolved": "https://registry.npmjs.org/@types/amqplib/-/amqplib-0.8.0.tgz",
+      "integrity": "sha512-RDojJ8WACs43HIfWSQGnAVwgNzjMGx4YMNeW7jptgAFgkG1EpNQqts+cND5HYWdYgTM58b+RHe675b0i4A9WpQ==",
+      "dependencies": {
+        "@types/bluebird": "*",
+        "@types/node": "*"
+      }
+    },
     "node_modules/@types/babel__core": {
       "version": "7.1.14",
       "resolved": "https://registry.npmjs.org/@types/babel__core/-/babel__core-7.1.14.tgz",
@@ -2799,6 +2809,11 @@
         "@babel/types": "^7.3.0"
       }
     },
+    "node_modules/@types/bluebird": {
+      "version": "3.5.35",
+      "resolved": "https://registry.npmjs.org/@types/bluebird/-/bluebird-3.5.35.tgz",
+      "integrity": "sha512-2WeeXK7BuQo7yPI4WGOBum90SzF/f8rqlvpaXx4rjeTmNssGRDHWf7fgDUH90xMB3sUOu716fUK5d+OVx0+ncQ=="
+    },
     "node_modules/@types/body-parser": {
       "version": "1.19.0",
       "resolved": "https://registry.npmjs.org/@types/body-parser/-/body-parser-1.19.0.tgz",
@@ -17553,6 +17568,15 @@
         "@types/node": "*"
       }
     },
+    "@types/amqplib": {
+      "version": "0.8.0",
+      "resolved": "https://registry.npmjs.org/@types/amqplib/-/amqplib-0.8.0.tgz",
+      "integrity": "sha512-RDojJ8WACs43HIfWSQGnAVwgNzjMGx4YMNeW7jptgAFgkG1EpNQqts+cND5HYWdYgTM58b+RHe675b0i4A9WpQ==",
+      "requires": {
+        "@types/bluebird": "*",
+        "@types/node": "*"
+      }
+    },
     "@types/babel__core": {
       "version": "7.1.14",
       "resolved": "https://registry.npmjs.org/@types/babel__core/-/babel__core-7.1.14.tgz",
@@ -17594,6 +17618,11 @@
         "@babel/types": "^7.3.0"
       }
     },
+    "@types/bluebird": {
+      "version": "3.5.35",
+      "resolved": "https://registry.npmjs.org/@types/bluebird/-/bluebird-3.5.35.tgz",
+      "integrity": "sha512-2WeeXK7BuQo7yPI4WGOBum90SzF/f8rqlvpaXx4rjeTmNssGRDHWf7fgDUH90xMB3sUOu716fUK5d+OVx0+ncQ=="
+    },
     "@types/body-parser": {
       "version": "1.19.0",
       "resolved": "https://registry.npmjs.org/@types/body-parser/-/body-parser-1.19.0.tgz",

package.json

@@ -1,6 +1,6 @@
 {
   "name": "fennec-be",
-  "version": "0.0.1",
+  "version": "0.1.0",
   "description": "",
   "author": "",
   "private": true,
@@ -29,6 +29,7 @@
     "@nestjs/graphql": "^7.9.8",
     "@nestjs/platform-express": "^7.5.1",
     "@nestjs/typeorm": "^7.1.5",
+    "@types/amqplib": "^0.8.0",
     "@types/bull": "^3.15.0",
     "@types/ramda": "^0.27.38",
     "apollo-server-express": "^2.19.2",

src/commons/commons.module.ts

@@ -1,8 +1,10 @@
 import { Module } from '@nestjs/common';
 import { PasswordConverter } from './services/password-converter';
+import { RedisMutexModule } from './redis-mutex/redis-mutex.module';
 
 @Module({
   providers: [PasswordConverter],
-  exports: [PasswordConverter],
+  exports: [PasswordConverter, RedisMutexModule],
+  imports: [RedisMutexModule],
 })
 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:
+    message: string | { error?: Error; message?: string | any; code?: number },
-      | string
-      | { error?: Error; message?: string | object; code?: number },
   ) {
     if (message instanceof Object) {
       super();
@@ -18,4 +18,8 @@ export class ApplicationException extends Error {
       super((message as unknown) as any);
     }
   }
+
+  toJSON() {
+    return pick(['code', 'message'], this);
+  }
 }

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,18 @@
+import { Test, TestingModule } from '@nestjs/testing';
+import { RedisMutexService } from './redis-mutex.service';
+
+describe('RedisMutexService', () => {
+  let service: RedisMutexService;
+
+  beforeEach(async () => {
+    const module: TestingModule = await Test.createTestingModule({
+      providers: [RedisMutexService],
+    }).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,

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

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

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) {
+  async write(message: PipelineTaskEvent, amqpMsg: ConsumeMessage) {
-    await this.redisService
+    const client = this.redisService.getClient();
-      .getClient()
+    await client.rpush(this.getKey(message.taskId), JSON.stringify(message));
-      .rpush(this.getKey(message.taskId), JSON.stringify(message));
+    await client.expire(this.getKey(message.taskId), 600); // ten minutes
-    if (isNil(message.unit) && terminalTaskStatuses.includes(message.status)) {
+    if (isNil(message.unit)) {
-      this.amqpConnection.request({
+      try {
-        exchange: EXCHANGE_PIPELINE_TASK_TOPIC,
+        await this.amqpConnection.request({
-        routingKey: ROUTE_PIPELINE_TASK_DONE,
+          exchange: EXCHANGE_PIPELINE_TASK_TOPIC,
-        payload: { taskId: message.taskId, status: message.status },
+          routingKey: ROUTE_PIPELINE_TASK_DONE,
-      });
+          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.runner.ts

@@ -5,19 +5,34 @@ 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';
 
 type Spawn = typeof spawn;
 
 export class PipelineTaskRunner {
   readonly processes = new Map<string, ChildProcessWithoutNullStreams>();
+  readonly stopTaskIds = new Set<string>();
 
   constructor(
     private readonly reposService: ReposService,
@@ -40,19 +55,27 @@ export class PipelineTaskRunner {
       this.logger.error({ task, err }, err.message);
     }
   }
-  @RabbitSubscribe({
+  @RabbitRPC({
-    exchange: 'stop-pipeline-task',
+    exchange: EXCHANGE_PIPELINE_TASK_TOPIC,
-    routingKey: 'mac',
+    routingKey: getSelfInstanceRouteKey(ROUTE_PIPELINE_TASK_KILL),
-    queue: 'mac.stop-pipeline-task',
+    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 +91,7 @@ export class PipelineTaskRunner {
     } else {
       this.logger.info({ task }, 'child process is not running.');
     }
+    return true;
   }
 
   async doTask(task: PipelineTask) {
@@ -137,6 +161,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);
       }
@@ -207,7 +234,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 },
@@ -260,6 +291,9 @@ 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'));
       });
     });

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

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

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

@@ -22,7 +22,6 @@ import { PipelineTaskFlushService } from './pipeline-task-flush.service';
     TypeOrmModule.forFeature([PipelineTask, Pipeline]),
     RedisModule,
     ReposModule,
-
     RabbitMQModule.forRootAsync(RabbitMQModule, {
       imports: [ConfigModule],
       useFactory: (configService: ConfigService) => ({
@@ -36,22 +35,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',

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

@@ -55,5 +55,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,15 @@ 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,6 +30,14 @@ describe('PipelineTasksService', () => {
           provide: AmqpConnection,
           useValue: {},
         },
+        {
+          provide: PipelineTaskFlushService,
+          useValue: {},
+        },
+        {
+          provide: getLoggerToken(PipelineTasksService.name),
+          useValue: new PinoLogger({}),
+        },
       ],
     }).compile();
 

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

@@ -1,11 +1,23 @@
-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';
+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';
 
 const log = debug('fennec:pipeline-tasks:service');
 
@@ -17,6 +29,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 +75,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/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/pipelines.resolver.ts

@@ -21,14 +21,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.ts

@@ -14,6 +14,8 @@ 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> {
@@ -42,8 +44,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 +57,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) =>
+      .then(([error, list]) => {
-        list.map((item) => {
+        if (error) {
-          item.date = new Date(item.date);
+          throw new ApplicationException(error);
-          return item;
+        }
-        }),
+        return plainToClass(Commit, list);
-      );
+      });
   }
 }

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.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/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.module.ts

@@ -7,12 +7,14 @@ import { ConfigModule, ConfigService } from '@nestjs/config';
 import { ProjectsModule } from '../projects/projects.module';
 import { EXCHANGE_REPO } from './repos.constants';
 import { RabbitMQModule } from '@golevelup/nestjs-rabbitmq';
+import { CommonsModule } from '../commons/commons.module';
 
 @Module({
   imports: [
     TypeOrmModule.forFeature([Project]),
     ConfigModule,
     ProjectsModule,
+    CommonsModule,
     RabbitMQModule.forRootAsync(RabbitMQModule, {
       imports: [ConfigModule],
       useFactory: (configService: ConfigService) => ({

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 {
@@ -129,7 +138,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 +149,23 @@ export class ReposService {
         `remotes/origin/${pipeline.branch}`,
         '--',
       ]);
-      return data.all.map(
+      return [
-        (it) =>
+        null,
-          ({
+        data.all.map(
-            ...it,
+          (it) =>
-            date: new Date(it.date),
+            ({
-          } as Commit),
+              ...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 +178,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 +190,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();
     }
   }
 }