chore: debug log 仅输出app的log

.vscode/settings.json

@@ -6,8 +6,6 @@
     "lpush",
     "lrange",
     "metatype",
-    "pmessage",
-    "psubscribe",
     "rpop",
     "rpush"
   ]

package.json

@@ -28,6 +28,7 @@
     "@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",
@@ -39,7 +40,6 @@
     "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",
@@ -58,7 +58,6 @@
     "@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

@@ -15,8 +15,6 @@ 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: [
@@ -47,28 +45,6 @@ import { PubSubModule } from './commons/pub-sub/pub-sub.module';
       }),
       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,

src/commons/commons.module.ts

@@ -1,10 +1,8 @@
 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/pipes/sanitize.pipe.ts

@@ -1,5 +1,5 @@
 import { ArgumentMetadata, Injectable, PipeTransform } from '@nestjs/common';
-import { plainToClass } from 'class-transformer';
+import { sanitize } from '@neuralegion/class-sanitizer/dist';
 
 @Injectable()
 export class SanitizePipe implements PipeTransform {
@@ -12,11 +12,13 @@ export class SanitizePipe implements PipeTransform {
       return value;
     }
     const constructorFunction = metadata.metatype;
-    if (!constructorFunction || value instanceof constructorFunction) {
+    if (!constructorFunction) {
       return value;
     }
+    value = Object.assign(new constructorFunction(), value);
     try {
-      return plainToClass(constructorFunction, value);
+      sanitize(value);
+      return value;
     } catch (err) {
       console.error(err);
       throw err;

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

@@ -1,5 +0,0 @@
-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

@@ -1,8 +0,0 @@
-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

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

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

@@ -1,15 +0,0 @@
-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

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

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

@@ -1,48 +0,0 @@
-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

@@ -1,31 +0,0 @@
-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

@@ -1,18 +0,0 @@
-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

@@ -1,9 +0,0 @@
-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

@@ -1,87 +0,0 @@
-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

@@ -1,115 +0,0 @@
-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

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

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

@@ -1,7 +1,6 @@
 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 {
@@ -10,7 +9,6 @@ export class PipelineTaskLogMessage {
   @Field(() => PipelineUnits, { nullable: true })
   unit?: PipelineUnits;
   @Field()
-  @Type(() => Date)
   time: Date;
   @Field()
   message: string;

src/pipeline-tasks/models/pipeline-task-logs.model.ts

@@ -1,7 +1,6 @@
 import { TaskStatuses } from '../enums/task-statuses.enum';
 import { PipelineUnits } from '../enums/pipeline-units.enum';
 import { Field, ObjectType } from '@nestjs/graphql';
-import { Type } from 'class-transformer';
 
 @ObjectType()
 export class PipelineTaskLogs {
@@ -9,9 +8,7 @@ export class PipelineTaskLogs {
   unit: PipelineUnits;
   @Field(() => TaskStatuses)
   status: TaskStatuses;
-  @Type(() => Date)
   startedAt?: Date;
-  @Type(() => Date)
   endedAt?: Date;
   logs = '';
 }

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

@@ -1,7 +1,6 @@
 import { Test, TestingModule } from '@nestjs/testing';
 import { PipelineTaskLogsService } from './pipeline-task-logs.service';
 import { RedisService } from 'nestjs-redis';
-import { getPubSubToken } from '../commons/pub-sub/utils/token';
 
 describe('PipelineTaskLogsService', () => {
   let service: PipelineTaskLogsService;
@@ -14,10 +13,6 @@ describe('PipelineTaskLogsService', () => {
           provide: RedisService,
           useValue: {},
         },
-        {
-          provide: getPubSubToken(),
-          useValue: {},
-        },
       ],
     }).compile();
 

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

@@ -1,9 +1,8 @@
 import { Injectable } from '@nestjs/common';
-import { observableToAsyncIterable } from 'graphql-tools';
+import { log } from 'console';
+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';
@@ -14,11 +13,9 @@ const LOG_TIMEOUT_SECONDS = 10_000;
 
 @Injectable()
 export class PipelineTaskLogsService {
-  constructor(
-    private readonly redisService: RedisService,
-    @InjectPubSub()
-    private readonly pubSub: PubSub,
-  ) {}
+  constructor(private readonly redisService: RedisService) {}
+
+  pubSub = new PubSub();
 
   get redis() {
     return this.redisService.getClient();
@@ -76,6 +73,6 @@ export class PipelineTaskLogsService {
   }
 
   watchLogs(task: PipelineTask) {
-    return observableToAsyncIterable(this.pubSub.message$(this.getKeys(task)));
+    return this.pubSub.asyncIterator(this.getKeys(task));
   }
 }

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

@@ -174,7 +174,7 @@ describe('PipelineTaskConsumer', () => {
 
       await consumer.doTask(job);
 
-      expect(updateTask).toHaveBeenCalledTimes(4);
+      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);
@@ -192,7 +192,7 @@ describe('PipelineTaskConsumer', () => {
 
       await consumer.doTask(job);
 
-      expect(updateTask).toHaveBeenCalledTimes(4);
+      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);
@@ -211,7 +211,7 @@ describe('PipelineTaskConsumer', () => {
       await consumer.doTask(job);
 
       expect(runScript).toHaveBeenCalledTimes(1);
-      expect(updateTask).toHaveBeenCalledTimes(4);
+      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);
@@ -232,7 +232,7 @@ describe('PipelineTaskConsumer', () => {
 
       await consumer.doTask(job);
 
-      expect(updateTask).toHaveBeenCalledTimes(4);
+      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);

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

@@ -86,7 +86,6 @@ export class PipelineTaskConsumer {
                 taskLogs.find((tl) => tl.unit === unit.type)?.logs ?? '',
             );
           task.logs.push(unitLog);
-          task = await this.service.updateTask(task);
           await job.update(task);
         }
       }

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

@@ -1,16 +1,11 @@
 import { AppBaseEntity } from './../commons/entities/app-base-entity';
-import { ObjectType } from '@nestjs/graphql';
-import { Column, Entity, ManyToOne, ValueTransformer } from 'typeorm';
+import { Field, ObjectType } from '@nestjs/graphql';
+import { Column, Entity, ManyToOne } from 'typeorm';
 import { Pipeline } from '../pipelines/pipeline.entity';
 import { PipelineTaskLogs } from './models/pipeline-task-logs.model';
 import { TaskStatuses } from './enums/task-statuses.enum';
 import { PipelineUnits } from './enums/pipeline-units.enum';
-import { plainToClass } from 'class-transformer';
 
-const logsTransformer: ValueTransformer = {
-  from: (value) => plainToClass(PipelineTaskLogs, value),
-  to: (value) => value,
-};
 @ObjectType()
 @Entity()
 export class PipelineTask extends AppBaseEntity {
@@ -25,7 +20,7 @@ export class PipelineTask extends AppBaseEntity {
   @Column({ type: 'enum', enum: PipelineUnits, array: true })
   units: PipelineUnits[];
 
-  @Column({ type: 'jsonb', default: '[]', transformer: logsTransformer })
+  @Column({ type: 'jsonb', default: '[]' })
   logs: PipelineTaskLogs[];
 
   @Column({ type: 'enum', enum: TaskStatuses, default: TaskStatuses.pending })

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

@@ -1,2 +1,3 @@
 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,9 +8,12 @@ import { ReposModule } from '../repos/repos.module';
 import { RedisModule } from 'nestjs-redis';
 import { BullModule } from '@nestjs/bull';
 import { PipelineTaskConsumer } from './pipeline-task.consumer';
-import { PIPELINE_TASK_QUEUE } from './pipeline-tasks.constants';
+import {
+  PIPELINE_TASK_QUEUE,
+  PIPELINE_TASK_LOG_PUBSUB,
+} from './pipeline-tasks.constants';
 import { PipelineTaskLogsService } from './pipeline-task-logs.service';
-import { PubSubModule } from '../commons/pub-sub/pub-sub.module';
+import { PubSub } from 'apollo-server-express';
 
 @Module({
   imports: [
@@ -18,7 +21,6 @@ import { PubSubModule } from '../commons/pub-sub/pub-sub.module';
     BullModule.registerQueue({
       name: PIPELINE_TASK_QUEUE,
     }),
-    PubSubModule.forFeature(),
     RedisModule,
     ReposModule,
   ],
@@ -27,6 +29,10 @@ import { PubSubModule } from '../commons/pub-sub/pub-sub.module';
     PipelineTasksResolver,
     PipelineTaskConsumer,
     PipelineTaskLogsService,
+    {
+      provide: Symbol(PIPELINE_TASK_LOG_PUBSUB),
+      useValue: new PubSub(),
+    },
   ],
   exports: [PipelineTasksService],
 })

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

@@ -5,7 +5,6 @@ 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 {
@@ -21,8 +20,7 @@ export class PipelineTasksResolver {
 
   @Subscription(() => PipelineTaskLogMessage, {
     resolve: (value) => {
-      const data = plainToClass(PipelineTaskLogMessage, value);
-      return data;
+      return value;
     },
   })
   async pipelineTaskLog(@Args() args: PipelineTaskLogArgs) {

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

@@ -10,7 +10,6 @@ import { EntityNotFoundError } from 'typeorm/error/EntityNotFoundError';
 import { Repository } from 'typeorm';
 import { Queue } from 'bull';
 import { LockFailedException } from '../commons/exceptions/lock-failed.exception';
-import { getPubSubToken } from '../commons/pub-sub/utils/token';
 
 describe('PipelineTasksService', () => {
   let service: PipelineTasksService;
@@ -69,10 +68,6 @@ describe('PipelineTasksService', () => {
             getClient: jest.fn(() => redisClient),
           },
         },
-        {
-          provide: getPubSubToken(),
-          useValue: {},
-        },
       ],
     }).compile();
 

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

@@ -9,17 +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';
-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>,
@@ -28,8 +27,6 @@ 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({
@@ -72,10 +69,6 @@ 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();
@@ -121,14 +114,12 @@ export class PipelineTasksService {
   }
 
   async updateTask(task: PipelineTask) {
-    this.pubSub.publish(`pipeline-task:${task.id}`, task);
+    this.pubSub.publish(task.id, task);
     return await this.repository.save(task);
   }
 
   async watchTaskUpdated(id: string) {
-    return observableToAsyncIterable(
-      this.pubSub.message$(`pipeline-task:${id}`),
-    );
+    return this.pubSub.asyncIterator(id);
   }
 
   getRedisTokens(pipeline: Pipeline): [string, string] {

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

@@ -1,30 +0,0 @@
-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

@@ -1,33 +0,0 @@
-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,8 +5,6 @@ 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: [
@@ -14,8 +12,7 @@ import { PipelineTasksModule } from '../pipeline-tasks/pipeline-tasks.module';
     BullModule.registerQueue({
       name: LIST_LOGS_TASK,
     }),
-    PipelineTasksModule,
   ],
-  providers: [PipelinesResolver, PipelinesService, CommitLogsResolver],
+  providers: [PipelinesResolver, PipelinesService],
 })
 export class PipelinesModule {}

src/pipelines/pipelines.resolver.ts

@@ -1,9 +1,10 @@
-import { Args, Mutation, Query, Resolver } from '@nestjs/graphql';
+import { Args, Mutation, Query, Resolver, Subscription } 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 {
@@ -41,4 +42,16 @@ 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,6 +1,5 @@
 import { ObjectType, Field } from '@nestjs/graphql';
 import { LogResult, DefaultLogFields } from 'simple-git';
-import { PipelineTask } from '../../pipeline-tasks/pipeline-task.entity';
 
 @ObjectType()
 export class LogFields {
@@ -11,7 +10,6 @@ export class LogFields {
   body: string;
   author_name: string;
   author_email: string;
-  tasks: PipelineTask[];
 }
 
 @ObjectType()

src/webhooks/webhooks.module.ts

@@ -4,6 +4,7 @@ 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],
@@ -11,4 +12,9 @@ import { WebhooksService } from './webhooks.service';
   providers: [WebhooksService],
 })
 export class WebhooksModule {
+  // configure(consumer: MiddlewareConsumer) {
+  //   consumer
+  //     .apply(raw({ type: 'application/json' }))
+  //     .forRoutes(GiteaWebhooksController);
+  // }
 }