chore: debug log 仅输出app的log

Reviewed-on: #1

.vscode/settings.json

@@ -1,8 +1,11 @@
 {
   "cSpell.words": [
     "Repos",
+    "boardcat",
+    "gitea",
     "lpush",
     "lrange",
+    "metatype",
     "rpop",
     "rpush"
   ]

package.json

@@ -10,8 +10,8 @@
     "build": "nest build",
     "format": "prettier --write \"src/**/*.ts\" \"test/**/*.ts\"",
     "start": "nest start",
-    "start:dev": "nest start --watch",
-    "start:debug": "DEBUG=simple-git,simple-git:* nest start --debug --watch",
+    "start:dev": "DEBUG=fennec:* nest start --watch",
+    "start:debug": "DEBUG=fennec:* nest start --debug --watch",
     "start:prod": "node dist/main",
     "lint": "eslint \"{src,apps,libs,test}/**/*.ts\" --fix",
     "test": "jest",
@@ -33,9 +33,11 @@
     "@types/ramda": "^0.27.38",
     "apollo-server-express": "^2.19.2",
     "bcrypt": "^5.0.0",
+    "body-parser": "^1.19.0",
     "bull": "^3.20.1",
     "class-transformer": "^0.3.2",
     "class-validator": "^0.13.1",
+    "debug": "^4.3.1",
     "graphql": "^15.5.0",
     "graphql-tools": "^7.0.2",
     "js-yaml": "^4.0.0",
@@ -50,9 +52,11 @@
     "typeorm": "^0.2.30"
   },
   "devDependencies": {
-    "@nestjs/cli": "^7.5.1",
+    "@nestjs/cli": "^7.5.7",
     "@nestjs/schematics": "^7.1.3",
     "@nestjs/testing": "^7.5.1",
+    "@types/body-parser": "^1.19.0",
+    "@types/debug": "^4.1.5",
     "@types/express": "^4.17.8",
     "@types/jest": "^26.0.15",
     "@types/js-yaml": "^4.0.0",

src/app.module.ts

@@ -1,4 +1,4 @@
-import { Module } from '@nestjs/common';
+import { MiddlewareConsumer, Module, NestModule } from '@nestjs/common';
 import { ConfigModule, ConfigService } from '@nestjs/config';
 import { GraphQLModule } from '@nestjs/graphql';
 import { TypeOrmModule } from '@nestjs/typeorm';
@@ -11,6 +11,10 @@ import { PipelinesModule } from './pipelines/pipelines.module';
 import { PipelineTasksModule } from './pipeline-tasks/pipeline-tasks.module';
 import configuration from './commons/config/configuration';
 import { RedisModule } from 'nestjs-redis';
+import { WebhooksModule } from './webhooks/webhooks.module';
+import { RawBodyMiddleware } from './commons/middlewares/raw-body.middleware';
+import { GiteaWebhooksController } from './webhooks/gitea-webhooks.controller';
+import { ParseBodyMiddleware } from './commons/middlewares/parse-body.middleware';
 
 @Module({
   imports: [
@@ -51,12 +55,21 @@ import { RedisModule } from 'nestjs-redis';
         host: configService.get<string>('db.redis.host', 'localhost'),
         port: configService.get<number>('db.redis.port', 6379),
         password: configService.get<string>('db.redis.password', ''),
-        keyPrefix: configService.get<string>('db.redis.prefix', 'fennec'),
+        keyPrefix: configService.get<string>('db.redis.prefix', 'fennec') + ':',
       }),
       inject: [ConfigService],
     }),
+    WebhooksModule,
   ],
   controllers: [AppController],
   providers: [AppService, AppResolver],
 })
-export class AppModule {}
+export class AppModule implements NestModule {
+  public configure(consumer: MiddlewareConsumer): void {
+    consumer
+      .apply(RawBodyMiddleware)
+      .forRoutes(GiteaWebhooksController)
+      .apply(ParseBodyMiddleware)
+      .forRoutes('*');
+  }
+}

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

@@ -3,25 +3,49 @@ import {
   Catch,
   ArgumentsHost,
   HttpException,
+  HttpStatus,
 } from '@nestjs/common';
 import { ApolloError } from 'apollo-server-errors';
 
 @Catch(HttpException)
 export class HttpExceptionFilter implements ExceptionFilter {
   catch(exception: HttpException, host: ArgumentsHost) {
-    const message = exception.message;
-    const extensions: Record<string, any> = {};
-    const err = exception.getResponse();
-    if (typeof err === 'string') {
-      extensions.message = err;
-    } else {
-      Object.assign(extensions, (err as any).extension);
-      extensions.message = (err as any).message;
+    switch (host.getType<'http' | 'graphql' | string>()) {
+      case 'graphql': {
+        const message = exception.message;
+        const extensions: Record<string, any> = {};
+        const err = exception.getResponse();
+        if (typeof err === 'string') {
+          extensions.message = err;
+        } else {
+          Object.assign(extensions, (err as any).extension);
+          extensions.message = (err as any).message;
+        }
+        return new ApolloError(
+          message,
+          exception.getStatus().toString(),
+          extensions,
+        );
+      }
+      case 'http': {
+        const ctx = host.switchToHttp();
+        const response = ctx.getResponse();
+        const request = ctx.getRequest();
+
+        const status =
+          exception instanceof HttpException
+            ? exception.getStatus()
+            : HttpStatus.INTERNAL_SERVER_ERROR;
+
+        response.status(status).json({
+          statusCode: status,
+          message: exception.message,
+          timestamp: new Date().toISOString(),
+          path: request.url,
+        });
+      }
+      default:
+        throw exception;
     }
-    return new ApolloError(
-      message,
-      exception.getStatus().toString(),
-      extensions,
-    );
   }
 }

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

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

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

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

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

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

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

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

src/commons/pipes/sanitize.pipe.ts

@@ -4,12 +4,24 @@ import { sanitize } from '@neuralegion/class-sanitizer/dist';
 @Injectable()
 export class SanitizePipe implements PipeTransform {
   transform(value: any, metadata: ArgumentMetadata) {
-    // console.log(value, typeof value);
-    if (value instanceof Object) {
-      value = Object.assign(new metadata.metatype(), value);
-      sanitize(value);
-      // console.log(value);
+    if (
+      !(value instanceof Object) ||
+      value instanceof Buffer ||
+      value instanceof Array
+    ) {
+      return value;
+    }
+    const constructorFunction = metadata.metatype;
+    if (!constructorFunction) {
+      return value;
+    }
+    value = Object.assign(new constructorFunction(), value);
+    try {
+      sanitize(value);
+      return value;
+    } catch (err) {
+      console.error(err);
+      throw err;
     }
-    return value;
   }
 }

src/main.ts

@@ -6,7 +6,7 @@ import { HttpExceptionFilter } from './commons/filters/all.exception-filter';
 import { SanitizePipe } from './commons/pipes/sanitize.pipe';
 
 async function bootstrap() {
-  const app = await NestFactory.create(AppModule);
+  const app = await NestFactory.create(AppModule, { bodyParser: false });
   const configService = app.get(ConfigService);
   app.useGlobalPipes(new SanitizePipe());
   app.useGlobalPipes(

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

@@ -12,13 +12,21 @@ export class PipelineTaskLogMessage {
   time: Date;
   @Field()
   message: string;
+  @Field()
+  isError: boolean;
 
-  static create(task: PipelineTask, unit: PipelineUnits, message: string) {
+  static create(
+    task: PipelineTask,
+    unit: PipelineUnits,
+    message: string,
+    isError: boolean,
+  ) {
     return Object.assign(new PipelineTaskLogMessage(), {
       task,
       message,
       time: new Date(),
       unit,
+      isError,
     });
   }
 }

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

@@ -1,8 +1,12 @@
 import { Injectable } from '@nestjs/common';
+import { log } from 'console';
 import { PubSub } from 'graphql-subscriptions';
 import { RedisService } from 'nestjs-redis';
-import { omit } from 'ramda';
+import { find, omit, propEq } from 'ramda';
+import { PipelineUnits } from './enums/pipeline-units.enum';
+import { TaskStatuses } from './enums/task-statuses.enum';
 import { PipelineTaskLogMessage } from './models/pipeline-task-log-message.module';
+import { PipelineTaskLogs } from './models/pipeline-task-logs.model';
 import { PipelineTask } from './pipeline-task.entity';
 
 const LOG_TIMEOUT_SECONDS = 10_000;
@@ -33,7 +37,7 @@ export class PipelineTaskLogsService {
     ]);
   }
 
-  async readLogs(task: PipelineTask): Promise<PipelineTaskLogMessage[]> {
+  async readLog(task: PipelineTask): Promise<PipelineTaskLogMessage[]> {
     return await this.redis.lrange(this.getKeys(task), 0, -1).then((items) =>
       items.map((item) => {
         const log = JSON.parse(item) as PipelineTaskLogMessage;
@@ -44,6 +48,30 @@ export class PipelineTaskLogsService {
     );
   }
 
+  async readLogsAsPipelineTaskLogs(
+    task: PipelineTask,
+  ): Promise<PipelineTaskLogs[]> {
+    const logs = await this.readLog(task);
+    const taskLogs: PipelineTaskLogs[] = [];
+    for (const log of logs) {
+      const taskLog = find<PipelineTaskLogs>(
+        propEq('unit', log.unit),
+        taskLogs,
+      );
+      if (!taskLog) {
+        taskLogs.push({
+          unit: (log.unit as unknown) as PipelineUnits,
+          status: TaskStatuses.working,
+          startedAt: log.time,
+          logs: log.message,
+        });
+      } else {
+        taskLog.logs += log.message;
+      }
+    }
+    return taskLogs;
+  }
+
   watchLogs(task: PipelineTask) {
     return this.pubSub.asyncIterator(this.getKeys(task));
   }

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

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

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

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

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

@@ -10,7 +10,6 @@ import { BullModule } from '@nestjs/bull';
 import { PipelineTaskConsumer } from './pipeline-task.consumer';
 import {
   PIPELINE_TASK_QUEUE,
-  PIPELINE_TASK_LOG_QUEUE,
   PIPELINE_TASK_LOG_PUBSUB,
 } from './pipeline-tasks.constants';
 import { PipelineTaskLogsService } from './pipeline-task-logs.service';
@@ -19,12 +18,9 @@ import { PubSub } from 'apollo-server-express';
 @Module({
   imports: [
     TypeOrmModule.forFeature([PipelineTask, Pipeline]),
-    BullModule.registerQueue(
-      {
-        name: PIPELINE_TASK_QUEUE,
-      },
-      { name: PIPELINE_TASK_LOG_QUEUE },
-    ),
+    BullModule.registerQueue({
+      name: PIPELINE_TASK_QUEUE,
+    }),
     RedisModule,
     ReposModule,
   ],
@@ -38,5 +34,6 @@ import { PubSub } from 'apollo-server-express';
       useValue: new PubSub(),
     },
   ],
+  exports: [PipelineTasksService],
 })
 export class PipelineTasksModule {}

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

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

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

@@ -29,6 +29,15 @@ export class PipelineTasksResolver {
     return asyncIterator;
   }
 
+  @Subscription(() => PipelineTask, {
+    resolve: (value) => {
+      return value;
+    },
+  })
+  async pipelineTaskChanged(@Args('id') id: string) {
+    return await this.service.watchTaskUpdated(id);
+  }
+
   @Query(() => [PipelineTask])
   async listPipelineTaskByPipelineId(@Args('pipelineId') pipelineId: string) {
     return await this.service.listTasksByPipelineId(pipelineId);

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

@@ -32,6 +32,7 @@ describe('PipelineTasksService', () => {
     ({
       pipelineId: 'test',
       commit: 'test',
+      pipeline: { branch: 'master' },
       units: [],
     } as PipelineTask);
 
@@ -79,6 +80,7 @@ describe('PipelineTasksService', () => {
     jest
       .spyOn(taskRepository, 'create')
       .mockImplementation((data: any) => data);
+    jest.spyOn(taskRepository, 'findOne').mockImplementation(async () => null);
   });
 
   it('should be defined', () => {
@@ -103,6 +105,7 @@ describe('PipelineTasksService', () => {
       const save = jest
         .spyOn(taskRepository, 'save')
         .mockImplementation(async (data: any) => data);
+      const findOne = jest.spyOn(taskRepository, 'findOne');
       jest
         .spyOn(service, 'doNextTask')
         .mockImplementation(async () => undefined);
@@ -112,6 +115,7 @@ describe('PipelineTasksService', () => {
           commit: 'test',
           units: [],
         });
+      expect(findOne).toBeCalled();
     });
     it('add task', async () => {
       const lpush = jest.spyOn(redisClient, 'lpush');

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

@@ -1,7 +1,7 @@
-import { Injectable } from '@nestjs/common';
+import { ConflictException, Injectable } from '@nestjs/common';
 import { InjectRepository } from '@nestjs/typeorm';
 import { PipelineTask } from './pipeline-task.entity';
-import { Repository } from 'typeorm';
+import { In, Repository } from 'typeorm';
 import { CreatePipelineTaskInput } from './dtos/create-pipeline-task.input';
 import { RedisService } from 'nestjs-redis';
 import { Pipeline } from '../pipelines/pipeline.entity';
@@ -9,9 +9,16 @@ import { InjectQueue } from '@nestjs/bull';
 import { PIPELINE_TASK_QUEUE } from './pipeline-tasks.constants';
 import { Queue } from 'bull';
 import { LockFailedException } from '../commons/exceptions/lock-failed.exception';
+import { PubSub } from 'apollo-server-express';
+import { TaskStatuses } from './enums/task-statuses.enum';
+import { isNil } from 'ramda';
+import debug from 'debug';
+
+const log = debug('fennec:pipeline-tasks:service');
 
 @Injectable()
 export class PipelineTasksService {
+  pubSub = new PubSub();
   constructor(
     @InjectRepository(PipelineTask)
     private readonly repository: Repository<PipelineTask>,
@@ -26,12 +33,30 @@ export class PipelineTasksService {
       where: { id: dto.pipelineId },
       relations: ['project'],
     });
+    const hasUnfinishedTask = await this.repository
+      .findOne({
+        pipelineId: dto.pipelineId,
+        commit: dto.commit,
+        status: In([TaskStatuses.pending, TaskStatuses.working]),
+      })
+      .then((val) => !isNil(val));
+    if (hasUnfinishedTask) {
+      throw new ConflictException(
+        'There are the same tasks among the unfinished tasks!',
+      );
+    }
     const task = await this.repository.save(this.repository.create(dto));
     task.pipeline = pipeline;
 
     const tasksKey = this.getRedisTokens(pipeline)[1];
     const redis = this.redis.getClient();
     await redis.lpush(tasksKey, JSON.stringify(task));
+    log(
+      'add task %s:%s-%s',
+      task.id,
+      task.pipeline.branch,
+      task.commit.slice(0, 6),
+    );
     await this.doNextTask(pipeline);
     return task;
   }
@@ -48,6 +73,7 @@ export class PipelineTasksService {
     const [lckKey, tasksKey] = this.getRedisTokens(pipeline);
     const redis = this.redis.getClient();
 
+    log('doNextTask()');
     const unLck = await new Promise<() => Promise<void>>(
       async (resolve, reject) => {
         const lckValue = Date.now().toString();
@@ -75,14 +101,27 @@ export class PipelineTasksService {
       (await redis.rpop(tasksKey).finally(() => unLck())) ?? 'null',
     );
     if (task) {
+      log(
+        'add task (%s:%s-%s) to queue',
+        task.id,
+        task.pipeline.branch,
+        task.commit.slice(0, 6),
+      );
       await this.queue.add(task);
+    } else {
+      log('task is empty');
     }
   }
 
   async updateTask(task: PipelineTask) {
+    this.pubSub.publish(task.id, task);
     return await this.repository.save(task);
   }
 
+  async watchTaskUpdated(id: string) {
+    return this.pubSub.asyncIterator(id);
+  }
+
   getRedisTokens(pipeline: Pipeline): [string, string] {
     return [`pipeline-${pipeline.id}:lck`, `pipeline-${pipeline.id}:tasks`];
   }

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

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

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

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

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

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

src/webhooks/gitea-webhooks.controller.ts

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

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

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

src/webhooks/webhook-log.entity.ts

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

src/webhooks/webhooks.module.ts

@@ -0,0 +1,20 @@
+import { MiddlewareConsumer, Module } from '@nestjs/common';
+import { TypeOrmModule } from '@nestjs/typeorm';
+import { PipelineTasksModule } from '../pipeline-tasks/pipeline-tasks.module';
+import { GiteaWebhooksController } from './gitea-webhooks.controller';
+import { WebhookLog } from './webhook-log.entity';
+import { WebhooksService } from './webhooks.service';
+import { raw } from 'body-parser';
+
+@Module({
+  imports: [TypeOrmModule.forFeature([WebhookLog]), PipelineTasksModule],
+  controllers: [GiteaWebhooksController],
+  providers: [WebhooksService],
+})
+export class WebhooksModule {
+  // configure(consumer: MiddlewareConsumer) {
+  //   consumer
+  //     .apply(raw({ type: 'application/json' }))
+  //     .forRoutes(GiteaWebhooksController);
+  // }
+}

src/webhooks/webhooks.service.spec.ts

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

src/webhooks/webhooks.service.ts

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

test/data/gitea-hook-payload.json.bin

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