feat: clean up.

Reviewed-on: #3

.vscode/settings.json

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

package.json

@@ -10,8 +10,8 @@
     "build": "nest build",
     "format": "prettier --write \"src/**/*.ts\" \"test/**/*.ts\"",
     "start": "nest start",
-    "start:dev": "DEBUG=fennec:*,simple-git:* nest start --watch",
+    "start:dev": "DEBUG=fennec:* nest start --watch",
-    "start:debug": "DEBUG=simple-git,simple-git:* nest start --debug --watch",
+    "start:debug": "DEBUG=fennec:* nest start --debug --watch",
     "start:prod": "node dist/main",
     "lint": "eslint \"{src,apps,libs,test}/**/*.ts\" --fix",
     "test": "jest",
@@ -28,17 +28,18 @@
     "@nestjs/graphql": "^7.9.8",
     "@nestjs/platform-express": "^7.5.1",
     "@nestjs/typeorm": "^7.1.5",
-    "@neuralegion/class-sanitizer": "^0.3.2",
     "@types/bull": "^3.15.0",
     "@types/ramda": "^0.27.38",
     "apollo-server-express": "^2.19.2",
     "bcrypt": "^5.0.0",
+    "body-parser": "^1.19.0",
     "bull": "^3.20.1",
     "class-transformer": "^0.3.2",
     "class-validator": "^0.13.1",
     "debug": "^4.3.1",
     "graphql": "^15.5.0",
     "graphql-tools": "^7.0.2",
+    "ioredis": "^4.25.0",
     "js-yaml": "^4.0.0",
     "nestjs-redis": "^1.2.8",
     "observable-to-async-generator": "^1.0.1-rc",
@@ -51,11 +52,13 @@
     "typeorm": "^0.2.30"
   },
   "devDependencies": {
-    "@nestjs/cli": "^7.5.1",
+    "@nestjs/cli": "^7.5.7",
     "@nestjs/schematics": "^7.1.3",
     "@nestjs/testing": "^7.5.1",
+    "@types/body-parser": "^1.19.0",
     "@types/debug": "^4.1.5",
     "@types/express": "^4.17.8",
+    "@types/ioredis": "^4.22.2",
     "@types/jest": "^26.0.15",
     "@types/js-yaml": "^4.0.0",
     "@types/node": "^14.14.6",

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,12 @@ import { PipelinesModule } from './pipelines/pipelines.module';
 import { PipelineTasksModule } from './pipeline-tasks/pipeline-tasks.module';
 import configuration from './commons/config/configuration';
 import { RedisModule } from 'nestjs-redis';
+import { WebhooksModule } from './webhooks/webhooks.module';
+import { RawBodyMiddleware } from './commons/middlewares/raw-body.middleware';
+import { GiteaWebhooksController } from './webhooks/gitea-webhooks.controller';
+import { ParseBodyMiddleware } from './commons/middlewares/parse-body.middleware';
+import { BullModule } from '@nestjs/bull';
+import { PubSubModule } from './commons/pub-sub/pub-sub.module';
 
 @Module({
   imports: [
@@ -41,6 +47,28 @@ import { RedisModule } from 'nestjs-redis';
       }),
       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),
+        },
+      }),
+      inject: [ConfigService],
+    }),
+    PubSubModule.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),
+        },
+      }),
+      inject: [ConfigService],
+    }),
     ProjectsModule,
     ReposModule,
     PipelinesModule,
@@ -55,8 +83,17 @@ import { RedisModule } from 'nestjs-redis';
       }),
       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/commons.module.ts

@@ -1,8 +1,10 @@
 import { Module } from '@nestjs/common';
 import { PasswordConverter } from './services/password-converter';
+import { PubSubModule } from './pub-sub/pub-sub.module';
 
 @Module({
   providers: [PasswordConverter],
   exports: [PasswordConverter],
+  imports: [PubSubModule],
 })
 export class CommonsModule {}

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;
+    switch (host.getType<'http' | 'graphql' | string>()) {
-    const extensions: Record<string, any> = {};
+      case 'graphql': {
-    const err = exception.getResponse();
+        const message = exception.message;
-    if (typeof err === 'string') {
+        const extensions: Record<string, any> = {};
-      extensions.message = err;
+        const err = exception.getResponse();
-    } else {
+        if (typeof err === 'string') {
-      Object.assign(extensions, (err as any).extension);
+          extensions.message = err;
-      extensions.message = (err as any).message;
+        } 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

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

src/commons/pub-sub/decorators/inject-pub-sub.decorator.ts

@@ -0,0 +1,5 @@
+import { Inject } from '@nestjs/common';
+import { getPubSubToken } from '../utils/token';
+
+export const InjectPubSub = (name?: string): ParameterDecorator =>
+  Inject(getPubSubToken(name));

src/commons/pub-sub/interfaces/pub-sub-async-config.interface.ts

@@ -0,0 +1,8 @@
+import { ModuleMetadata } from '@nestjs/common';
+import { PubSubOptions } from './pub-sub-options.interface';
+
+export interface PubSubAsyncConfig extends Pick<ModuleMetadata, 'imports'> {
+  name?: string;
+  useFactory: (...args: any[]) => Promise<PubSubOptions> | PubSubOptions;
+  inject?: any[];
+}

src/commons/pub-sub/interfaces/pub-sub-options.interface.ts

@@ -0,0 +1,6 @@
+import { RedisOptions } from 'ioredis';
+
+export interface PubSubOptions {
+  name?: string;
+  redis: RedisOptions;
+}

src/commons/pub-sub/interfaces/pub-sub-raw-message.interface.ts

@@ -0,0 +1,15 @@
+export type PubSubRawMessage<T> =
+  | PubSubRawNextMessage<T>
+  | PubSubRawErrorMessage<T>
+  | PubSubRawCompleteMessage<T>;
+export interface PubSubRawNextMessage<T> {
+  type: 'next';
+  message: T;
+}
+export interface PubSubRawErrorMessage<T> {
+  type: 'error';
+  error: string;
+}
+export interface PubSubRawCompleteMessage<T> {
+  type: 'complete';
+}

src/commons/pub-sub/pub-sub.constants.ts

@@ -0,0 +1 @@
+export const DEFAULT_PUB_SUB_NAME = 'default';

src/commons/pub-sub/pub-sub.module.ts

@@ -0,0 +1,48 @@
+import { DynamicModule, Module } from '@nestjs/common';
+import { PubSubService } from './pub-sub.service';
+import {
+  createAsyncPubSubProviders,
+  createPubSubProvider,
+} from './pub-sub.providers';
+import { PubSubOptions } from './interfaces/pub-sub-options.interface';
+import { PubSubAsyncConfig } from './interfaces/pub-sub-async-config.interface';
+import { getPubSubConfigToken } from './utils/token';
+
+@Module({
+  providers: [PubSubService],
+})
+export class PubSubModule {
+  public static forRoot(options: PubSubOptions): DynamicModule {
+    const providers = [
+      {
+        provide: getPubSubConfigToken(options.name),
+        useValue: options,
+      },
+      createPubSubProvider(options.name),
+    ];
+    return {
+      global: true,
+      module: PubSubModule,
+      providers,
+      exports: providers,
+    };
+  }
+  public static forRootAsync(...configs: PubSubAsyncConfig[]): DynamicModule {
+    const providers = createAsyncPubSubProviders(configs);
+    return {
+      global: true,
+      module: PubSubModule,
+      imports: configs
+        .map((config) => config.imports)
+        .flat()
+        .filter((o, i, a) => a.indexOf(o) === i),
+      providers,
+      exports: providers,
+    };
+  }
+  public static forFeature(): DynamicModule {
+    return {
+      module: PubSubModule,
+    };
+  }
+}

src/commons/pub-sub/pub-sub.providers.ts

@@ -0,0 +1,31 @@
+import { Provider } from '@nestjs/common';
+import { PubSubAsyncConfig } from './interfaces/pub-sub-async-config.interface';
+import { PubSubOptions } from './interfaces/pub-sub-options.interface';
+import { PubSub } from './pub-sub';
+import { getPubSubConfigToken, getPubSubToken } from './utils/token';
+
+export function createPubSubProvider(name: string): Provider {
+  return {
+    provide: getPubSubToken(name),
+    useFactory: (option: PubSubOptions) => new PubSub(option),
+    inject: [getPubSubConfigToken(name)],
+  };
+}
+
+export function createOptionsProvider(config: PubSubAsyncConfig): Provider {
+  return {
+    provide: getPubSubConfigToken(config.name),
+    useFactory: config.useFactory,
+    inject: config.inject || [],
+  };
+}
+export function createAsyncPubSubProviders(
+  configs: PubSubAsyncConfig[],
+): Provider[] {
+  return configs
+    .map((config) => [
+      createOptionsProvider(config),
+      createPubSubProvider(config.name),
+    ])
+    .flat();
+}

src/commons/pub-sub/pub-sub.service.spec.ts

@@ -0,0 +1,18 @@
+import { Test, TestingModule } from '@nestjs/testing';
+import { PubSubService } from './pub-sub.service';
+
+describe('PubsubService', () => {
+  let service: PubSubService;
+
+  beforeEach(async () => {
+    const module: TestingModule = await Test.createTestingModule({
+      providers: [PubSubService],
+    }).compile();
+
+    service = module.get<PubSubService>(PubSubService);
+  });
+
+  it('should be defined', () => {
+    expect(service).toBeDefined();
+  });
+});

src/commons/pub-sub/pub-sub.service.ts

@@ -0,0 +1,9 @@
+import { Injectable } from '@nestjs/common';
+import { PubSubOptions } from './interfaces/pub-sub-options.interface';
+
+@Injectable()
+export class PubSubService {
+  private options = new Map<string, PubSubOptions>();
+  private pubClient;
+  private;
+}

src/commons/pub-sub/pub-sub.spec.ts

@@ -0,0 +1,87 @@
+import debug from 'debug';
+import { tap } from 'rxjs/operators';
+import { PubSub } from './pub-sub';
+
+debug.enable('app:pubsub:*');
+
+describe('PubSub', () => {
+  let instance: PubSub;
+
+  beforeEach(async () => {
+    instance = new PubSub({
+      name: 'default',
+      redis: {
+        host: 'localhost',
+      },
+    });
+  });
+
+  it('should be defined', () => {
+    expect(instance).toBeDefined();
+  });
+
+  it('should can send and receive message', async () => {
+    const arr = new Array(10)
+      .fill(undefined)
+      .map(() => Math.random().toString(36).slice(2, 8));
+    const results: string[] = [];
+    instance
+      .message$<string>('test')
+      .pipe(
+        tap((val) => {
+          console.log(val);
+        }),
+      )
+      .subscribe((val) => results.push(val));
+    await new Promise((r) => setTimeout(r, 1000));
+    await Promise.all([...arr.map((str) => instance.publish('test', str))]);
+    await new Promise((r) => setTimeout(r, 1000));
+    expect(results).toMatchObject(arr);
+  });
+
+  it('should complete', async () => {
+    const arr = new Array(10)
+      .fill(undefined)
+      .map(() => Math.random().toString(36).slice(2, 8));
+    const results: string[] = [];
+    instance
+      .message$<string>('test')
+      .pipe(
+        tap((val) => {
+          console.log(val);
+        }),
+      )
+      .subscribe((val) => results.push(val));
+    await new Promise((r) => setTimeout(r, 1000));
+    await Promise.all([...arr.map((str) => instance.publish('test', str))]);
+    await instance.finish('test');
+    await Promise.all([...arr.map((str) => instance.publish('test', str))]);
+    await new Promise((r) => setTimeout(r, 1000));
+    expect(results).toMatchObject(arr);
+  });
+  it('should error', async () => {
+    const arr = new Array(10)
+      .fill(undefined)
+      .map(() => Math.random().toString(36).slice(2, 8));
+    const results: string[] = [];
+    let error: string;
+    instance
+      .message$<string>('test')
+      .pipe(
+        tap((val) => {
+          console.log(val);
+        }),
+      )
+      .subscribe({
+        next: (val) => results.push(val),
+        error: (err) => (error = err.message),
+      });
+    await new Promise((r) => setTimeout(r, 1000));
+    await Promise.all([...arr.map((str) => instance.publish('test', str))]);
+    await instance.throwError('test', 'TEST ERROR MESSAGE');
+    await Promise.all([...arr.map((str) => instance.publish('test', str))]);
+    await new Promise((r) => setTimeout(r, 1000));
+    expect(results).toMatchObject(arr);
+    expect(error).toEqual('TEST ERROR MESSAGE');
+  });
+});

src/commons/pub-sub/pub-sub.ts

@@ -0,0 +1,115 @@
+import debug from 'debug';
+import { EventEmitter } from 'events';
+import IORedis, { Redis } from 'ioredis';
+import { from, fromEvent, Observable } from 'rxjs';
+import { filter, map, share, switchMap, takeWhile, tap } from 'rxjs/operators';
+import { ApplicationException } from '../exceptions/application.exception';
+import { PubSubOptions } from './interfaces/pub-sub-options.interface';
+import {
+  PubSubRawMessage,
+  PubSubRawNextMessage,
+} from './interfaces/pub-sub-raw-message.interface';
+
+const log = debug('fennec:pubsub:instance');
+export class PubSub extends EventEmitter {
+  pubRedis: Redis;
+  pSubRedis: Redis;
+  channels: string[] = [];
+  event$: Observable<[string, string, any]>;
+
+  constructor(private readonly options: PubSubOptions) {
+    super();
+    this.pubRedis = new IORedis(this.options.redis);
+    this.pSubRedis = new IORedis(this.options.redis);
+
+    this.pSubRedis.on('pmessage', (...args) =>
+      log.extend('raw')('%s %s %o', ...args),
+    );
+
+    this.event$ = fromEvent<[string, string, string]>(
+      this.pSubRedis,
+      'pmessage',
+    ).pipe(
+      map((ev) => {
+        try {
+          ev[2] = JSON.parse(ev[2]);
+        } catch (err) {
+          log('WARN: is not json');
+          return null;
+        }
+        log('on message: %s %s %o', ...ev);
+        return ev;
+      }),
+      filter((v) => !!v),
+      share(),
+    );
+  }
+
+  async disconnect() {
+    log('disconnecting to redis...');
+    this.pubRedis.disconnect();
+    this.pSubRedis.disconnect();
+    log('disconnected');
+  }
+
+  async registerChannel(channel: string) {
+    if (this.channels.includes(channel)) {
+      return;
+    }
+    this.channels.push(channel);
+
+    return await this.pSubRedis.psubscribe(channel);
+  }
+
+  private async redisPublish<T>(channel: string, message: PubSubRawMessage<T>) {
+    log.extend('publish')('channel: %s, message: %O', channel, message);
+    return await this.pubRedis.publish(channel, JSON.stringify(message));
+  }
+
+  async publish(channel: string, message: any): Promise<number> {
+    return await this.redisPublish(channel, {
+      type: 'next',
+      message,
+    });
+  }
+
+  async finish(channel: string): Promise<number> {
+    return await this.redisPublish(channel, {
+      type: 'complete',
+    });
+  }
+
+  async throwError(channel: string, error: string): Promise<number> {
+    return await this.redisPublish(channel, {
+      type: 'error',
+      error,
+    });
+  }
+
+  message$ = <T>(channel: string): Observable<T> => {
+    return from(this.registerChannel(channel))
+      .pipe(switchMap(() => this.event$))
+      .pipe(
+        filter(([pattern]) => pattern === channel),
+        tap(([pattern, channel, message]) => {
+          log.extend('subscribe')(
+            'channel: %s, match: %s, message: %O',
+            channel,
+            pattern,
+            message,
+          );
+        }),
+        takeWhile(([pattern, channel, message]) => {
+          if (pattern === channel) {
+            if (message.type === 'error') {
+              throw new ApplicationException(message.error);
+            }
+            return message.type !== 'complete';
+          }
+          return true;
+        }),
+        map((ev) => ev[2] as PubSubRawMessage<T>),
+        map((message: PubSubRawNextMessage<T>) => message.message),
+      );
+  };
+}

src/commons/pub-sub/utils/token.ts

@@ -0,0 +1,8 @@
+import { DEFAULT_PUB_SUB_NAME } from '../pub-sub.constants';
+
+export function getPubSubToken(name) {
+  return `app:pub-usb:${name || DEFAULT_PUB_SUB_NAME}`;
+}
+export function getPubSubConfigToken(name) {
+  return `app:pub-usb:config:${name || DEFAULT_PUB_SUB_NAME}`;
+}

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

@@ -1,6 +1,7 @@
 import { PipelineTask } from './../pipeline-task.entity';
 import { PipelineUnits } from '../enums/pipeline-units.enum';
 import { Field, HideField, ObjectType } from '@nestjs/graphql';
+import { Type } from 'class-transformer';
 
 @ObjectType()
 export class PipelineTaskLogMessage {
@@ -9,6 +10,7 @@ export class PipelineTaskLogMessage {
   @Field(() => PipelineUnits, { nullable: true })
   unit?: PipelineUnits;
   @Field()
+  @Type(() => Date)
   time: Date;
   @Field()
   message: string;

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

@@ -1,8 +1,9 @@
 import { Injectable } from '@nestjs/common';
-import { log } from 'console';
+import { observableToAsyncIterable } from 'graphql-tools';
-import { PubSub } from 'graphql-subscriptions';
 import { RedisService } from 'nestjs-redis';
 import { find, omit, propEq } from 'ramda';
+import { InjectPubSub } from '../commons/pub-sub/decorators/inject-pub-sub.decorator';
+import { PubSub } from '../commons/pub-sub/pub-sub';
 import { PipelineUnits } from './enums/pipeline-units.enum';
 import { TaskStatuses } from './enums/task-statuses.enum';
 import { PipelineTaskLogMessage } from './models/pipeline-task-log-message.module';
@@ -13,9 +14,11 @@ const LOG_TIMEOUT_SECONDS = 10_000;
 
 @Injectable()
 export class PipelineTaskLogsService {
-  constructor(private readonly redisService: RedisService) {}
+  constructor(
-
+    private readonly redisService: RedisService,
-  pubSub = new PubSub();
+    @InjectPubSub()
+    private readonly pubSub: PubSub,
+  ) {}
 
   get redis() {
     return this.redisService.getClient();
@@ -73,6 +76,6 @@ export class PipelineTaskLogsService {
   }
 
   watchLogs(task: PipelineTask) {
-    return this.pubSub.asyncIterator(this.getKeys(task));
+    return observableToAsyncIterable(this.pubSub.message$(this.getKeys(task)));
   }
 }

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

@@ -1,3 +1,2 @@
 export const PIPELINE_TASK_QUEUE = 'PIPELINE_TASK_QUEUE';
 export const PIPELINE_TASK_LOG_QUEUE = 'PIPELINE_TASK_LOG_QUEUE';
-export const PIPELINE_TASK_LOG_PUBSUB = 'PIPELINE_TASK_LOG_PUBSUB';

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

@@ -8,13 +8,9 @@ import { ReposModule } from '../repos/repos.module';
 import { RedisModule } from 'nestjs-redis';
 import { BullModule } from '@nestjs/bull';
 import { PipelineTaskConsumer } from './pipeline-task.consumer';
-import {
+import { PIPELINE_TASK_QUEUE } from './pipeline-tasks.constants';
-  PIPELINE_TASK_QUEUE,
-  PIPELINE_TASK_LOG_QUEUE,
-  PIPELINE_TASK_LOG_PUBSUB,
-} from './pipeline-tasks.constants';
 import { PipelineTaskLogsService } from './pipeline-task-logs.service';
-import { PubSub } from 'apollo-server-express';
+import { PubSubModule } from '../commons/pub-sub/pub-sub.module';
 
 @Module({
   imports: [
@@ -22,6 +18,7 @@ import { PubSub } from 'apollo-server-express';
     BullModule.registerQueue({
       name: PIPELINE_TASK_QUEUE,
     }),
+    PubSubModule.forFeature(),
     RedisModule,
     ReposModule,
   ],
@@ -30,10 +27,7 @@ import { PubSub } from 'apollo-server-express';
     PipelineTasksResolver,
     PipelineTaskConsumer,
     PipelineTaskLogsService,
-    {
-      provide: Symbol(PIPELINE_TASK_LOG_PUBSUB),
-      useValue: new PubSub(),
-    },
   ],
+  exports: [PipelineTasksService],
 })
 export class PipelineTasksModule {}

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

@@ -5,6 +5,7 @@ import { CreatePipelineTaskInput } from './dtos/create-pipeline-task.input';
 import { PipelineTaskLogMessage } from './models/pipeline-task-log-message.module';
 import { PipelineTaskLogArgs } from './dtos/pipeline-task-log.args';
 import { PipelineTaskLogsService } from './pipeline-task-logs.service';
+import { plainToClass } from 'class-transformer';
 
 @Resolver()
 export class PipelineTasksResolver {
@@ -20,7 +21,8 @@ export class PipelineTasksResolver {
 
   @Subscription(() => PipelineTaskLogMessage, {
     resolve: (value) => {
-      return value;
+      const data = plainToClass(PipelineTaskLogMessage, value);
+      return data;
     },
   })
   async pipelineTaskLog(@Args() args: PipelineTaskLogArgs) {

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

@@ -9,16 +9,17 @@ 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';
+import { InjectPubSub } from '../commons/pub-sub/decorators/inject-pub-sub.decorator';
+import { PubSub } from '../commons/pub-sub/pub-sub';
+import { observableToAsyncIterable } from '@graphql-tools/utils';
 
 const log = debug('fennec:pipeline-tasks:service');
 
 @Injectable()
 export class PipelineTasksService {
-  pubSub = new PubSub();
   constructor(
     @InjectRepository(PipelineTask)
     private readonly repository: Repository<PipelineTask>,
@@ -27,6 +28,8 @@ export class PipelineTasksService {
     @InjectQueue(PIPELINE_TASK_QUEUE)
     private readonly queue: Queue<PipelineTask>,
     private readonly redis: RedisService,
+    @InjectPubSub()
+    private readonly pubSub: PubSub,
   ) {}
   async addTask(dto: CreatePipelineTaskInput) {
     const pipeline = await this.pipelineRepository.findOneOrFail({
@@ -69,6 +72,10 @@ export class PipelineTasksService {
     return await this.repository.find({ pipelineId });
   }
 
+  async listTasksByCommitHash(hash: string) {
+    return await this.repository.find({ commit: hash });
+  }
+
   async doNextTask(pipeline: Pipeline) {
     const [lckKey, tasksKey] = this.getRedisTokens(pipeline);
     const redis = this.redis.getClient();
@@ -114,12 +121,13 @@ export class PipelineTasksService {
   }
 
   async updateTask(task: PipelineTask) {
-    this.pubSub.publish(task.id, task);
+    this.pubSub.publish(`task:${task.id}`, task);
     return await this.repository.save(task);
   }
 
   async watchTaskUpdated(id: string) {
-    return this.pubSub.asyncIterator(id);
+    log('watchTaskUpdated %s', id);
+    return observableToAsyncIterable(this.pubSub.message$(`task:${id}`));
   }
 
   getRedisTokens(pipeline: Pipeline): [string, string] {

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

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

src/pipelines/commit-logs.resolver.ts

@@ -0,0 +1,33 @@
+import {
+  Args,
+  Parent,
+  ResolveField,
+  Resolver,
+  Subscription,
+} from '@nestjs/graphql';
+import { PipelineTasksService } from '../pipeline-tasks/pipeline-tasks.service';
+import { LogFields, LogList } from '../repos/dtos/log-list.model';
+import { PipelinesService } from './pipelines.service';
+
+@Resolver(() => LogFields)
+export class CommitLogsResolver {
+  constructor(
+    private readonly service: PipelinesService,
+    private readonly taskServices: PipelineTasksService,
+  ) {}
+  @Subscription(() => LogList, {
+    resolve: (value) => {
+      return value;
+    },
+  })
+  async listLogsForPipeline(@Args('id', { type: () => String }) id: string) {
+    const job = await this.service.listLogsForPipeline(id);
+    return (async function* () {
+      yield await job.finished();
+    })();
+  }
+  @ResolveField()
+  async tasks(@Parent() commit: LogFields) {
+    return await this.taskServices.listTasksByCommitHash(commit.hash);
+  }
+}

src/pipelines/pipelines.module.ts

@@ -5,6 +5,8 @@ import { TypeOrmModule } from '@nestjs/typeorm';
 import { Pipeline } from './pipeline.entity';
 import { BullModule } from '@nestjs/bull';
 import { LIST_LOGS_TASK } from '../repos/repos.constants';
+import { CommitLogsResolver } from './commit-logs.resolver';
+import { PipelineTasksModule } from '../pipeline-tasks/pipeline-tasks.module';
 
 @Module({
   imports: [
@@ -12,7 +14,8 @@ import { LIST_LOGS_TASK } from '../repos/repos.constants';
     BullModule.registerQueue({
       name: LIST_LOGS_TASK,
     }),
+    PipelineTasksModule,
   ],
-  providers: [PipelinesResolver, PipelinesService],
+  providers: [PipelinesResolver, PipelinesService, CommitLogsResolver],
 })
 export class PipelinesModule {}

src/pipelines/pipelines.resolver.ts

@@ -1,10 +1,9 @@
-import { Args, Mutation, Query, Resolver, Subscription } from '@nestjs/graphql';
+import { Args, Mutation, Query, Resolver } from '@nestjs/graphql';
 import { CreatePipelineInput } from './dtos/create-pipeline.input';
 import { UpdatePipelineInput } from './dtos/update-pipeline.input';
 import { Pipeline } from './pipeline.entity';
 import { PipelinesService } from './pipelines.service';
 import { ListPipelineArgs } from './dtos/list-pipelines.args';
-import { LogList } from '../repos/dtos/log-list.model';
 
 @Resolver()
 export class PipelinesResolver {
@@ -42,16 +41,4 @@ export class PipelinesResolver {
   async deletePipeline(@Args('id', { type: () => String }) id: string) {
     return await this.service.remove(id);
   }
-
-  @Subscription(() => LogList, {
-    resolve: (value) => {
-      return value;
-    },
-  })
-  async listLogsForPipeline(@Args('id', { type: () => String }) id: string) {
-    const job = await this.service.listLogsForPipeline(id);
-    return (async function* () {
-      yield await job.finished();
-    })();
-  }
 }

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

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

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

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

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

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

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

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

src/webhooks/gitea-webhooks.controller.ts

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

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

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

src/webhooks/webhook-log.entity.ts

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

src/webhooks/webhooks.module.ts

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

src/webhooks/webhooks.service.spec.ts

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

src/webhooks/webhooks.service.ts

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

test/data/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"
+  }
+}