refactor(pipeline, repo): rabbitmq

src/app.module.ts

@@ -16,9 +16,7 @@ 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';
 import { LoggerModule } from 'nestjs-pino';
-import { RabbitMQModule } from '@golevelup/nestjs-rabbitmq';
 
 import pinoPretty from 'pino-pretty';
 
@@ -79,17 +77,6 @@ import pinoPretty from 'pino-pretty';
       }),
       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/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/commons/utils/rabbit-mq.ts

@@ -0,0 +1,21 @@
+import { hostname } from 'os';
+
+export function getInstanceName() {
+  return hostname();
+}
+
+export function getSelfInstanceRouteKey(key: string) {
+  return getAppInstanceRouteKey(key, getInstanceName());
+}
+
+export function getAppInstanceRouteKey(key: string, appInstance?: string) {
+  return appInstance ? `${key}.${appInstance}` : key;
+}
+
+export function getSelfInstanceQueueKey(key: string) {
+  return getAppInstanceQueueKey(key, getInstanceName());
+}
+
+export function getAppInstanceQueueKey(key: string, appInstance?: string) {
+  return appInstance ? `${key}.${appInstance}` : key;
+}

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

@@ -1,30 +0,0 @@
-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;
-
-  beforeEach(async () => {
-    const module: TestingModule = await Test.createTestingModule({
-      providers: [
-        PipelineTaskLogsService,
-        {
-          provide: RedisService,
-          useValue: {},
-        },
-        {
-          provide: getPubSubToken(),
-          useValue: {},
-        },
-      ],
-    }).compile();
-
-    service = module.get<PipelineTaskLogsService>(PipelineTaskLogsService);
-  });
-
-  it('should be defined', () => {
-    expect(service).toBeDefined();
-  });
-});

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

@@ -1,81 +0,0 @@
-import { Injectable } from '@nestjs/common';
-import { observableToAsyncIterable } from 'graphql-tools';
-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';
-import { PipelineTaskLogs } from './models/pipeline-task-logs.model';
-import { PipelineTask } from './pipeline-task.entity';
-
-const LOG_TIMEOUT_SECONDS = 10_000;
-
-@Injectable()
-export class PipelineTaskLogsService {
-  constructor(
-    private readonly redisService: RedisService,
-    @InjectPubSub()
-    private readonly pubSub: PubSub,
-  ) {}
-
-  get redis() {
-    return this.redisService.getClient();
-  }
-
-  getKeys(task: PipelineTask) {
-    return `ptl:${task.id}`;
-  }
-
-  async recordLog(log: PipelineTaskLogMessage) {
-    const logDto = omit(['task'], log);
-    await Promise.all([
-      this.pubSub.publish(this.getKeys(log.task), logDto),
-      this.redis
-        .expire(this.getKeys(log.task), LOG_TIMEOUT_SECONDS)
-        .then(() =>
-          this.redis.rpush(this.getKeys(log.task), JSON.stringify(logDto)),
-        ),
-    ]);
-  }
-
-  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;
-        log.task = task;
-        log.time = new Date(log.time);
-        return log;
-      }),
-    );
-  }
-
-  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 observableToAsyncIterable(this.pubSub.message$(this.getKeys(task)));
-  }
-}

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

@@ -1,243 +0,0 @@
-import { Test, TestingModule } from '@nestjs/testing';
-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 { 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';
-import { getLoggerToken, PinoLogger } from 'nestjs-pino';
-
-describe('PipelineTaskConsumer', () => {
-  let consumer: PipelineTaskConsumer;
-  let tasksService: PipelineTasksService;
-  let logsService: PipelineTaskLogsService;
-  const getJob = () =>
-    ({
-      data: {
-        pipelineId: 'test',
-        units: [PipelineUnits.checkout, PipelineUnits.test],
-      },
-    } as Job<PipelineTask>);
-
-  beforeEach(async () => {
-    const module: TestingModule = await Test.createTestingModule({
-      providers: [
-        {
-          provide: PipelineTasksService,
-          useValue: {
-            doNextTask: () => undefined,
-            updateTask: async (value) => value,
-          },
-        },
-        {
-          provide: ReposService,
-          useValue: {
-            getWorkspaceRootByTask: () => 'workspace-root',
-            checkout: async () => undefined,
-          },
-        },
-        {
-          provide: PipelineTaskLogsService,
-          useValue: {
-            recordLog: async () => undefined,
-            readLogsAsPipelineTaskLogs: async () => [],
-          },
-        },
-        {
-          provide: getLoggerToken(PipelineTaskConsumer.name),
-          useValue: new PinoLogger({}),
-        },
-        PipelineTaskConsumer,
-      ],
-    }).compile();
-
-    tasksService = module.get(PipelineTasksService);
-    logsService = module.get(PipelineTaskLogsService);
-    consumer = module.get(PipelineTaskConsumer);
-  });
-
-  it('should be defined', () => {
-    expect(consumer).toBeDefined();
-  });
-
-  describe('onCompleted', () => {
-    it('should call doNextTask()', () => {
-      const job = getJob();
-      const doNextTask = jest.spyOn(tasksService, 'doNextTask');
-      consumer.onCompleted(job);
-      expect(doNextTask).toHaveBeenCalledTimes(1);
-    });
-  });
-
-  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 () => {
-      await consumer.runScript(
-        'node one-second-work.js',
-        join(__dirname, '../../test/data'),
-      );
-      expect(logText).toMatch(
-        /node one-second-work\.js.+10.+20.+30.+40.+50.+60.+70.+80.+90/s,
-      );
-      expect(recordLog).toHaveBeenCalled();
-    });
-    it('should failed and log right message', async () => {
-      await expect(
-        consumer.runScript(
-          'node bad-work.js',
-          join(__dirname, '../../test/data'),
-        ),
-      ).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);
-    });
-    it('should log with task', async () => {
-      const task = new PipelineTask();
-      task.id = 'test';
-
-      const recordLog = jest.spyOn(logsService, 'recordLog');
-      await expect(
-        consumer.runScript(
-          'node bad-work.js',
-          join(__dirname, '../../test/data'),
-          task,
-        ),
-      ).rejects.toThrowError(/exec script failed/);
-
-      expect(errorText).toMatch(/Error Message 2/);
-      expect(
-        ((recordLog.mock.calls[2][0] as unknown) as PipelineTaskLogMessage)
-          .task,
-      ).toMatchObject(task);
-    });
-  });
-
-  describe('doTask', () => {
-    let task: PipelineTask;
-
-    beforeEach(() => {
-      task = new PipelineTask();
-      task.id = 'test-id';
-      task.logs = [];
-      task.pipeline = new Pipeline();
-      task.pipeline.workUnitMetadata = {
-        version: 1,
-        units: [
-          {
-            type: PipelineUnits.checkout,
-            scripts: [],
-          },
-          {
-            type: PipelineUnits.installDependencies,
-            scripts: ["echo ' Hello, Fennec!'"],
-          },
-        ],
-      };
-      task.units = task.pipeline.workUnitMetadata.units.map(
-        (unit) => unit.type,
-      );
-      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(4);
-      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(4);
-      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,
-        update: jest.fn().mockImplementation(() => undefined),
-      } as unknown) as Job;
-
-      const runScript = jest
-        .spyOn(consumer, 'runScript')
-        .mockImplementation(async () => undefined);
-      const updateTask = jest.spyOn(tasksService, 'updateTask');
-
-      await consumer.doTask(job);
-
-      expect(runScript).toHaveBeenCalledTimes(1);
-      expect(updateTask).toHaveBeenCalledTimes(4);
-      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);
-    });
-    it('should log error message', async () => {
-      const job: Job = ({
-        data: task,
-        update: jest.fn().mockImplementation(() => undefined),
-      } as unknown) as Job;
-
-      jest.spyOn(consumer, 'runScript').mockImplementation(async () => {
-        throw new Error('bad message');
-      });
-      const updateTask = jest.spyOn(tasksService, 'updateTask');
-
-      await consumer.doTask(job);
-
-      expect(updateTask).toHaveBeenCalledTimes(4);
-      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);
-    });
-  });
-});

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

@@ -1,161 +0,0 @@
-import { PipelineTaskLogs } from './models/pipeline-task-logs.model';
-import { ReposService } from './../repos/repos.service';
-import {
-  OnQueueCompleted,
-  OnQueueFailed,
-  Process,
-  Processor,
-} from '@nestjs/bull';
-import { Job } from 'bull';
-import { spawn } from 'child_process';
-import { PipelineTask } from './pipeline-task.entity';
-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 { InjectPinoLogger, PinoLogger } from 'nestjs-pino';
-
-@Processor(PIPELINE_TASK_QUEUE)
-export class PipelineTaskConsumer {
-  constructor(
-    private readonly service: PipelineTasksService,
-    private readonly reposService: ReposService,
-    private readonly logsService: PipelineTaskLogsService,
-    @InjectPinoLogger(PipelineTaskConsumer.name)
-    private readonly logger: PinoLogger,
-  ) {}
-  @Process()
-  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);
-    this.logger.info({ task }, 'running task [%s].', task.id);
-    await job.update(task);
-
-    const workspaceRoot = this.reposService.getWorkspaceRootByTask(task);
-
-    const units = task.units.map(
-      (type) =>
-        task.pipeline.workUnitMetadata.units.find(
-          (unit) => unit.type === type,
-        ) ?? { type: type, scripts: [] },
-    );
-
-    this.logger.info({ units }, 'begin run units.');
-    try {
-      for (const unit of units) {
-        const unitLog = new PipelineTaskLogs();
-        unitLog.unit = unit.type;
-        unitLog.startedAt = new Date();
-        this.logger.info('curr unit is %s', unit.type);
-        try {
-          // 检出代码前执行 git checkout
-          if (unit.type === PipelineUnits.checkout) {
-            this.logger.debug('begin checkout');
-            await this.reposService.checkout(task, workspaceRoot);
-            unitLog.status = TaskStatuses.success;
-            this.logger.debug('end checkout');
-          }
-          for (const script of unit.scripts) {
-            unitLog.logs += `[RUN SCRIPT] ${script}`;
-            this.logger.debug('begin runScript %s', script);
-            await this.runScript(script, workspaceRoot, task, unit.type);
-            this.logger.debug('end runScript %s', script);
-          }
-          unitLog.status = TaskStatuses.success;
-        } catch (err) {
-          unitLog.status = TaskStatuses.failed;
-          unitLog.logs += err.message;
-          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);
-          task = await this.service.updateTask(task);
-          await job.update(task);
-        }
-      }
-
-      task.status = TaskStatuses.success;
-      this.logger.info({ task }, 'task [%s] completed.', task.id);
-    } catch (err) {
-      task.status = TaskStatuses.failed;
-      this.logger.error({ task, error: err }, 'task [%s] failed.', task.id);
-    } finally {
-      task.endedAt = new Date();
-      task = await this.service.updateTask(task);
-      await job.update(task);
-    }
-  }
-
-  async runScript(
-    script: string,
-    workspaceRoot: string,
-    task?: PipelineTask,
-    unit?: PipelineUnits,
-  ): Promise<void> {
-    await this.logsService.recordLog(
-      PipelineTaskLogMessage.create(task, unit, script + '\n', false),
-    );
-    return new Promise((resolve, reject) => {
-      const sub = spawn(script, {
-        shell: true,
-        cwd: workspaceRoot,
-      });
-      let loggingCount = 0; // semaphore
-
-      sub.stderr.on('data', (data: Buffer) => {
-        const str = data.toString();
-        loggingCount++;
-        this.logsService
-          .recordLog(PipelineTaskLogMessage.create(task, unit, str, true))
-          .finally(() => loggingCount--);
-      });
-      sub.stdout.on('data', (data: Buffer) => {
-        const str = data.toString();
-        loggingCount++;
-        this.logsService
-          .recordLog(PipelineTaskLogMessage.create(task, unit, str, false))
-          .finally(() => loggingCount--);
-      });
-      sub.addListener('close', async (code) => {
-        await new Promise<void>(async (resolve) => {
-          for (let i = 0; i < 10 && loggingCount > 0; i++) {
-            await new Promise((resolve) => setTimeout(resolve, 500));
-            this.logger.debug('waiting logging... (%dx500ms)', i);
-          }
-          resolve();
-        });
-        if (code === 0) {
-          return resolve();
-        }
-        return reject(new ApplicationException('exec script failed'));
-      });
-    });
-  }
-
-  @OnQueueCompleted()
-  onCompleted(job: Job<PipelineTask>) {
-    this.service.doNextTask(job.data.pipeline);
-  }
-
-  @OnQueueFailed()
-  onFailed(job: Job<PipelineTask>) {
-    this.service.doNextTask(job.data.pipeline);
-  }
-}

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

@@ -8,7 +8,6 @@ import { TaskStatuses } from './enums/task-statuses.enum';
 import { getLoggerToken, PinoLogger } from 'nestjs-pino';
 import { PipelineTaskRunner } from './pipeline-task.runner';
 import { WorkUnitMetadata } from './models/work-unit-metadata.model';
-import { Code } from 'typeorm';
 describe('PipelineTaskRunner', () => {
   let runner: PipelineTaskRunner;
   let reposService: ReposService;

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

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

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

@@ -6,11 +6,6 @@ import { PipelineTask } from './pipeline-task.entity';
 import { Pipeline } from '../pipelines/pipeline.entity';
 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 { PipelineTaskLogsService } from './pipeline-task-logs.service';
-import { PubSubModule } from '../commons/pub-sub/pub-sub.module';
 import { RabbitMQModule } from '@golevelup/nestjs-rabbitmq';
 import { ConfigModule, ConfigService } from '@nestjs/config';
 import { PipelineTaskRunner } from './pipeline-task.runner';
@@ -19,10 +14,6 @@ import { spawn } from 'child_process';
 @Module({
   imports: [
     TypeOrmModule.forFeature([PipelineTask, Pipeline]),
-    BullModule.registerQueue({
-      name: PIPELINE_TASK_QUEUE,
-    }),
-    PubSubModule.forFeature(),
     RedisModule,
     ReposModule,
 
@@ -60,8 +51,6 @@ import { spawn } from 'child_process';
   providers: [
     PipelineTasksService,
     PipelineTasksResolver,
-    PipelineTaskConsumer,
-    PipelineTaskLogsService,
     PipelineTaskRunner,
     {
       provide: 'spawn',

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

@@ -1,6 +1,5 @@
 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', () => {
@@ -14,10 +13,6 @@ describe('PipelineTasksResolver', () => {
           provide: PipelineTasksService,
           useValue: {},
         },
-        {
-          provide: PipelineTaskLogsService,
-          useValue: {},
-        },
       ],
     }).compile();
 

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

@@ -4,15 +4,11 @@ import { PipelineTasksService } from './pipeline-tasks.service';
 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 {
-  constructor(
+  constructor(private readonly service: PipelineTasksService) {}
-    private readonly service: PipelineTasksService,
-    private readonly logsService: PipelineTaskLogsService,
-  ) {}
 
   @Mutation(() => PipelineTask)
   async createPipelineTask(@Args('task') taskDto: CreatePipelineTaskInput) {
@@ -26,9 +22,9 @@ export class PipelineTasksResolver {
     },
   })
   async pipelineTaskLog(@Args() args: PipelineTaskLogArgs) {
-    const task = await this.service.findTaskById(args.taskId);
+    // const task = await this.service.findTaskById(args.taskId);
-    const asyncIterator = this.logsService.watchLogs(task);
+    // const asyncIterator = this.logsService.watchLogs(task);
-    return asyncIterator;
+    // return asyncIterator;
   }
 
   @Subscription(() => PipelineTask, {
@@ -37,7 +33,7 @@ export class PipelineTasksResolver {
     },
   })
   async pipelineTaskChanged(@Args('id') id: string) {
-    return await this.service.watchTaskUpdated(id);
+    // return await this.service.watchTaskUpdated(id);
   }
 
   @Query(() => [PipelineTask])

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

@@ -2,15 +2,10 @@ import { Test, TestingModule } from '@nestjs/testing';
 import { PipelineTasksService } from './pipeline-tasks.service';
 import { getRepositoryToken } from '@nestjs/typeorm';
 import { PipelineTask } from './pipeline-task.entity';
-import { PIPELINE_TASK_QUEUE } from './pipeline-tasks.constants';
-import { getQueueToken } from '@nestjs/bull';
-import { RedisService } from 'nestjs-redis';
 import { Pipeline } from '../pipelines/pipeline.entity';
-import { EntityNotFoundError } from 'typeorm/error/EntityNotFoundError';
 import { Repository } from 'typeorm';
 import { Queue } from 'bull';
-import { LockFailedException } from '../commons/exceptions/lock-failed.exception';
+import { AmqpConnection } from '@golevelup/nestjs-rabbitmq';
-import { getPubSubToken } from '../commons/pub-sub/utils/token';
 
 describe('PipelineTasksService', () => {
   let service: PipelineTasksService;
@@ -29,25 +24,8 @@ describe('PipelineTasksService', () => {
     } as Pipeline);
   let redisClient;
   let taskQueue: Queue;
-  const getTask = () =>
-    ({
-      pipelineId: 'test',
-      commit: 'test',
-      pipeline: { branch: 'master' },
-      units: [],
-    } as PipelineTask);
 
   beforeEach(async () => {
-    redisClient = (() => ({
-      set: jest.fn().mockImplementation(async () => 'OK'),
-      del: jest.fn().mockImplementation(async () => 'test'),
-      get: jest.fn().mockImplementation(async () => 'test'),
-      lpush: jest.fn().mockImplementation(async () => 1),
-      rpop: jest.fn().mockImplementation(async () => JSON.stringify(getTask())),
-    }))() as any;
-    taskQueue = (() => ({
-      add: jest.fn().mockImplementation(async () => null),
-    }))() as any;
     module = await Test.createTestingModule({
       providers: [
         PipelineTasksService,
@@ -60,17 +38,7 @@ describe('PipelineTasksService', () => {
           useValue: new Repository(),
         },
         {
-          provide: getQueueToken(PIPELINE_TASK_QUEUE),
+          provide: AmqpConnection,
-          useValue: taskQueue,
-        },
-        {
-          provide: RedisService,
-          useValue: {
-            getClient: jest.fn(() => redisClient),
-          },
-        },
-        {
-          provide: getPubSubToken(),
           useValue: {},
         },
       ],
@@ -92,119 +60,43 @@ describe('PipelineTasksService', () => {
     expect(service).toBeDefined();
   });
 
-  describe('addTask', () => {
+  // describe('addTask', () => {
-    beforeEach(() => {
+  //   beforeEach(() => {
-      jest
+  //     jest
-        .spyOn(pipelineRepository, 'findOneOrFail')
+  //       .spyOn(pipelineRepository, 'findOneOrFail')
-        .mockImplementation(async () => getBasePipeline());
+  //       .mockImplementation(async () => getBasePipeline());
-    });
+  //   });
-    it('pipeline not found', async () => {
+  //   it('pipeline not found', async () => {
-      jest.spyOn(taskRepository, 'findOneOrFail').mockImplementation(() => {
+  //     jest.spyOn(taskRepository, 'findOneOrFail').mockImplementation(() => {
-        throw new EntityNotFoundError(Pipeline, {});
+  //       throw new EntityNotFoundError(Pipeline, {});
-      });
+  //     });
-      await expect(
+  //     await expect(
-        service.addTask({ pipelineId: 'test', commit: 'test', units: [] }),
+  //       service.addTask({ pipelineId: 'test', commit: 'test', units: [] }),
-      ).rejects;
+  //     ).rejects;
-    });
+  //   });
-    it('create task on db', async () => {
+  //   it('create task on db', async () => {
-      const save = jest
+  //     const save = jest
-        .spyOn(taskRepository, 'save')
+  //       .spyOn(taskRepository, 'save')
-        .mockImplementation(async (data: any) => data);
+  //       .mockImplementation(async (data: any) => data);
-      const findOne = jest.spyOn(taskRepository, 'findOne');
+  //     const findOne = jest.spyOn(taskRepository, 'findOne');
-      jest
+  //     await service.addTask({ pipelineId: 'test', commit: 'test', units: [] }),
-        .spyOn(service, 'doNextTask')
+  //       expect(save.mock.calls[0][0]).toMatchObject({
-        .mockImplementation(async () => undefined);
+  //         pipelineId: 'test',
-      await service.addTask({ pipelineId: 'test', commit: 'test', units: [] }),
+  //         commit: 'test',
-        expect(save.mock.calls[0][0]).toMatchObject({
+  //         units: [],
-          pipelineId: 'test',
+  //       });
-          commit: 'test',
+  //     expect(findOne).toBeCalled();
-          units: [],
+  //   });
-        });
+  //   it('add task', async () => {
-      expect(findOne).toBeCalled();
+  //     const lpush = jest.spyOn(redisClient, 'lpush');
-    });
+  //     await service.addTask({ pipelineId: 'test', commit: 'test', units: [] });
-    it('add task', async () => {
+  //     expect(typeof lpush.mock.calls[0][1] === 'string').toBeTruthy();
-      const lpush = jest.spyOn(redisClient, 'lpush');
+  //     expect(JSON.parse(lpush.mock.calls[0][1] as string)).toMatchObject({
-      const doNextTask = jest.spyOn(service, 'doNextTask');
+  //       pipelineId: 'test',
-      jest
+  //       commit: 'test',
-        .spyOn(service, 'doNextTask')
+  //       units: [],
-        .mockImplementation(async () => undefined);
+  //       pipeline: getBasePipeline(),
-      await service.addTask({ pipelineId: 'test', commit: 'test', units: [] });
+  //     });
-      expect(typeof lpush.mock.calls[0][1] === 'string').toBeTruthy();
+  //   });
-      expect(JSON.parse(lpush.mock.calls[0][1] as string)).toMatchObject({
+  // });
-        pipelineId: 'test',
-        commit: 'test',
-        units: [],
-        pipeline: getBasePipeline(),
-      });
-      expect(doNextTask).toHaveBeenCalledWith(getBasePipeline());
-    });
-  });
-
-  describe('doNextTask', () => {
-    it('add task to queue', async () => {
-      let lckValue: string;
-      const set = jest
-        .spyOn(redisClient, 'set')
-        .mockImplementation(async (...args) => (lckValue = args[3] as string));
-      const get = jest
-        .spyOn(redisClient, 'get')
-        .mockImplementation(async () => lckValue);
-      const del = jest.spyOn(redisClient, 'del');
-      const rpop = jest.spyOn(redisClient, 'rpop');
-      const add = jest.spyOn(taskQueue, 'add');
-
-      await service.doNextTask(getBasePipeline());
-
-      expect(add).toHaveBeenCalledWith(getTask());
-      expect(set).toHaveBeenCalledTimes(1);
-      expect(rpop).toHaveBeenCalledTimes(1);
-      expect(get).toHaveBeenCalledTimes(1);
-      expect(del).toHaveBeenCalledTimes(1);
-    });
-    it('pipeline is busy', async () => {
-      let remainTimes = 3;
-
-      let lckValue: string;
-      const set = jest
-        .spyOn(redisClient, 'set')
-        .mockImplementation(async (...args) => {
-          if (remainTimes-- > 0) {
-            throw new Error();
-          } else {
-            lckValue = args[3] as string;
-          }
-        });
-      const get = jest
-        .spyOn(redisClient, 'get')
-        .mockImplementation(async () => lckValue);
-      const del = jest.spyOn(redisClient, 'del');
-      const rpop = jest.spyOn(redisClient, 'rpop');
-      const add = jest.spyOn(taskQueue, 'add');
-
-      await service.doNextTask(getBasePipeline());
-
-      expect(rpop).toHaveBeenCalledTimes(1);
-      expect(set).toHaveBeenCalledTimes(4);
-      expect(get).toHaveBeenCalledTimes(1);
-      expect(del).toHaveBeenCalledTimes(1);
-      expect(add).toHaveBeenCalledWith(getTask());
-    }, 10_000);
-    it('pipeline always busy and timeout', async () => {
-      const set = jest
-        .spyOn(redisClient, 'set')
-        .mockImplementation(async () => {
-          throw new Error();
-        });
-      const get = jest.spyOn(redisClient, 'get');
-      const del = jest.spyOn(redisClient, 'del');
-
-      await expect(
-        service.doNextTask(getBasePipeline()),
-      ).rejects.toBeInstanceOf(LockFailedException);
-
-      expect(set).toHaveBeenCalledTimes(5);
-      expect(get).toHaveBeenCalledTimes(0);
-      expect(del).toHaveBeenCalledTimes(0);
-    }, 15_000);
-  });
 });

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

@@ -1,20 +1,10 @@
-import { ConflictException, Injectable } from '@nestjs/common';
+import { Injectable } from '@nestjs/common';
 import { InjectRepository } from '@nestjs/typeorm';
 import { PipelineTask } from './pipeline-task.entity';
-import { In, Repository } from 'typeorm';
+import { Repository } from 'typeorm';
 import { CreatePipelineTaskInput } from './dtos/create-pipeline-task.input';
-import { RedisService } from 'nestjs-redis';
 import { Pipeline } from '../pipelines/pipeline.entity';
-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 { 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';
 import { AmqpConnection } from '@golevelup/nestjs-rabbitmq';
 
 const log = debug('fennec:pipeline-tasks:service');
@@ -26,11 +16,6 @@ export class PipelineTasksService {
     private readonly repository: Repository<PipelineTask>,
     @InjectRepository(Pipeline)
     private readonly pipelineRepository: Repository<Pipeline>,
-    @InjectQueue(PIPELINE_TASK_QUEUE)
-    private readonly queue: Queue<PipelineTask>,
-    private readonly redis: RedisService,
-    @InjectPubSub()
-    private readonly pubSub: PubSub,
     private readonly amqpConnection: AmqpConnection,
   ) {}
   async addTask(dto: CreatePipelineTaskInput) {
@@ -69,61 +54,6 @@ export class PipelineTasksService {
     return await this.repository.find({ commit: hash });
   }
 
-  async doNextTask(pipeline: Pipeline) {
-    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();
-        for (let i = 0; i < 5; i++) {
-          if (
-            await redis
-              .set(lckKey, 0, 'EX', lckValue, 'NX')
-              .then(() => true)
-              .catch(() => false)
-          ) {
-            resolve(async () => {
-              if ((await redis.get(lckKey)) === lckValue) {
-                await redis.del(lckKey);
-              }
-            });
-            return;
-          }
-          await new Promise((resolve) => setTimeout(resolve, 2000));
-        }
-        reject(new LockFailedException(lckKey));
-      },
-    );
-
-    const task = JSON.parse(
-      (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(`pipeline-task:${task.id}`, task);
-    return await this.repository.save(task);
-  }
-
-  async watchTaskUpdated(id: string) {
-    return observableToAsyncIterable(
-      this.pubSub.message$(`pipeline-task:${id}`),
-    );
-  }
-
   getRedisTokens(pipeline: Pipeline): [string, string] {
     return [`pipeline-${pipeline.id}:lck`, `pipeline-${pipeline.id}:tasks`];
   }

src/pipelines/commit-logs.resolver.ts

@@ -7,8 +7,7 @@ import {
   Subscription,
 } from '@nestjs/graphql';
 import { PipelineTasksService } from '../pipeline-tasks/pipeline-tasks.service';
-import { Commit, LogFields, LogList } from '../repos/dtos/log-list.model';
+import { Commit, LogFields } from '../repos/dtos/log-list.model';
-import { ReposService } from '../repos/repos.service';
 import { PipelinesService } from './pipelines.service';
 
 @Resolver(() => Commit)
@@ -16,17 +15,18 @@ export class CommitLogsResolver {
   constructor(
     private readonly service: PipelinesService,
     private readonly taskServices: PipelineTasksService,
-    private readonly reposService: ReposService,
   ) {}
-  @Subscription(() => LogList, {
+  @Subscription(() => String, { resolve: (val) => val, nullable: true })
-    resolve: (value) => {
+  async syncCommits(
-      return value;
+    @Args('pipelineId', { type: () => String })
-    },
+    pipelineId: string,
-  })
+    @Args('appInstance', { type: () => String, nullable: true })
-  async listLogsForPipeline(@Args('id', { type: () => String }) id: string) {
+    appInstance?: string,
-    const job = await this.service.listLogsForPipeline(id);
+  ) {
+    const pipeline = await this.service.findOneWithProject(pipelineId);
+    const syncCommitsPromise = this.service.syncCommits(pipeline, appInstance);
     return (async function* () {
-      yield await job.finished();
+      yield await syncCommitsPromise;
     })();
   }
   @ResolveField()
@@ -34,13 +34,9 @@ export class CommitLogsResolver {
     return await this.taskServices.listTasksByCommitHash(commit.hash);
   }
 
-  @Query(() => [Commit])
+  @Query(() => [Commit], { nullable: true })
   async commits(@Args('pipelineId', { type: () => String }) id: string) {
     const pipeline = await this.service.findOneWithProject(id);
-    const data = await this.reposService.listCommits(
+    return await this.service.listCommits(pipeline);
-      pipeline.project,
-      pipeline.branch,
-    );
-    return data;
   }
 }

src/pipelines/pipelines.module.ts

@@ -3,20 +3,24 @@ import { PipelinesResolver } from './pipelines.resolver';
 import { PipelinesService } from './pipelines.service';
 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';
 import { ReposModule } from '../repos/repos.module';
+import { RabbitMQModule } from '@golevelup/nestjs-rabbitmq';
+import { ConfigModule, ConfigService } from '@nestjs/config';
 
 @Module({
   imports: [
     TypeOrmModule.forFeature([Pipeline]),
-    BullModule.registerQueue({
-      name: LIST_LOGS_TASK,
-    }),
     PipelineTasksModule,
-    ReposModule,
+    RabbitMQModule.forRootAsync(RabbitMQModule, {
+      imports: [ConfigModule],
+      useFactory: (configService: ConfigService) => ({
+        uri: configService.get<string>('db.rabbitmq.uri'),
+        exchanges: [],
+      }),
+      inject: [ConfigService],
+    }),
   ],
   providers: [PipelinesResolver, PipelinesService, CommitLogsResolver],
 })

src/pipelines/pipelines.service.spec.ts

@@ -2,18 +2,14 @@ import { Test, TestingModule } from '@nestjs/testing';
 import { PipelinesService } from './pipelines.service';
 import { Pipeline } from './pipeline.entity';
 import { getRepositoryToken } from '@nestjs/typeorm';
-import { getQueueToken } from '@nestjs/bull';
-import { LIST_LOGS_TASK } from '../repos/repos.constants';
 import { Repository } from 'typeorm';
 import { Project } from '../projects/project.entity';
-import { Job, Queue } from 'bull';
+import { AmqpConnection } from '@golevelup/nestjs-rabbitmq';
-import { ListLogsOption } from '../repos/models/list-logs.options';
 
 describe('PipelinesService', () => {
   let service: PipelinesService;
   let repository: Repository<Pipeline>;
   let pipeline: Pipeline;
-  let queue: Queue<ListLogsOption>;
 
   beforeEach(async () => {
     pipeline = Object.assign(new Pipeline(), {
@@ -37,33 +33,17 @@ describe('PipelinesService', () => {
           },
         },
         {
-          provide: getQueueToken(LIST_LOGS_TASK),
+          provide: AmqpConnection,
-          useValue: {
+          useValue: {},
-            add: jest.fn().mockImplementation(() => ({ id: 1 } as Job)),
-          },
         },
       ],
     }).compile();
 
     service = module.get<PipelinesService>(PipelinesService);
     repository = module.get(getRepositoryToken(Pipeline));
-    queue = module.get(getQueueToken(LIST_LOGS_TASK));
   });
 
   it('should be defined', () => {
     expect(service).toBeDefined();
   });
-
-  describe('listLogsForPipeline', () => {
-    it('should send task to queue.', async () => {
-      const add = jest.spyOn(queue, 'add');
-      await expect(
-        service.listLogsForPipeline('test-pipeline'),
-      ).resolves.toEqual({ id: 1 });
-      expect(add).toBeCalledWith({
-        project: pipeline.project,
-        branch: pipeline.branch,
-      });
-    });
-  });
 });

src/pipelines/pipelines.service.ts

@@ -6,10 +6,14 @@ import { BaseDbService } from '../commons/services/base-db.service';
 import { CreatePipelineInput } from './dtos/create-pipeline.input';
 import { UpdatePipelineInput } from './dtos/update-pipeline.input';
 import { ListPipelineArgs } from './dtos/list-pipelines.args';
-import { InjectQueue } from '@nestjs/bull';
+import {
-import { LIST_LOGS_TASK } from '../repos/repos.constants';
+  EXCHANGE_REPO,
-import { Queue } from 'bull';
+  ROUTE_FETCH,
-import { ListLogsOption } from '../repos/models/list-logs.options';
+  ROUTE_LIST_COMMITS,
+} from '../repos/repos.constants';
+import { AmqpConnection } from '@golevelup/nestjs-rabbitmq';
+import { Commit } from '../repos/dtos/log-list.model';
+import { getAppInstanceRouteKey } from '../commons/utils/rabbit-mq';
 
 @Injectable()
 export class PipelinesService extends BaseDbService<Pipeline> {
@@ -17,8 +21,7 @@ export class PipelinesService extends BaseDbService<Pipeline> {
   constructor(
     @InjectRepository(Pipeline)
     readonly repository: Repository<Pipeline>,
-    @InjectQueue(LIST_LOGS_TASK)
+    private readonly amqpConnection: AmqpConnection,
-    private readonly listLogsQueue: Queue<ListLogsOption>,
   ) {
     super(repository);
   }
@@ -47,16 +50,27 @@ export class PipelinesService extends BaseDbService<Pipeline> {
   async remove(id: string) {
     return (await this.repository.softDelete({ id })).affected;
   }
-
+  async syncCommits(pipeline: Pipeline, appInstance?: string) {
-  async listLogsForPipeline(id: string) {
+    return await this.amqpConnection.request<string | null>({
-    const pipeline = await this.repository.findOneOrFail({
+      exchange: EXCHANGE_REPO,
-      where: { id },
+      routingKey: getAppInstanceRouteKey(ROUTE_FETCH, appInstance),
-      relations: ['project'],
+      payload: pipeline,
+      timeout: 30_000,
     });
-    const job = await this.listLogsQueue.add({
+  }
-      project: pipeline.project,
+  async listCommits(pipeline: Pipeline) {
-      branch: pipeline.branch,
+    return await this.amqpConnection
-    });
+      .request<Commit[]>({
-    return job;
+        exchange: EXCHANGE_REPO,
+        routingKey: ROUTE_LIST_COMMITS,
+        payload: pipeline,
+        timeout: 10_000,
+      })
+      .then((list) =>
+        list.map((item) => {
+          item.date = new Date(item.date);
+          return item;
+        }),
+      );
   }
 }

src/repos/list-logs.consumer.ts

@@ -1,14 +0,0 @@
-import { ReposService } from './repos.service';
-import { Processor, Process } from '@nestjs/bull';
-import { Job } from 'bull';
-import { ListLogsOption } from './models/list-logs.options';
-import { LIST_LOGS_TASK } from './repos.constants';
-@Processor(LIST_LOGS_TASK)
-export class ListLogsConsumer {
-  constructor(private readonly service: ReposService) {}
-  @Process()
-  async listLogs(job: Job<ListLogsOption>) {
-    const logs = await this.service.listLogs(job.data);
-    return logs;
-  }
-}

src/repos/repos.constants.ts

@@ -1,3 +1,5 @@
-export const LIST_LOGS_TASK = 'LIST_LOGS_TASK';
+export const EXCHANGE_REPO = 'fennec.repo';
-export const LIST_LOGS_PUB_SUB = 'LIST_LOGS_PUB_SUB';
+export const ROUTE_FETCH = 'fetch';
-export const LIST_LOGS_DONE = 'LIST_LOGS_DONE';
+export const ROUTE_LIST_COMMITS = 'list-commits';
+export const QUEUE_LIST_COMMITS = 'list-commits';
+export const QUEUE_FETCH = 'repo-fetch';

src/repos/repos.module.ts

@@ -3,22 +3,35 @@ import { TypeOrmModule } from '@nestjs/typeorm';
 import { Project } from '../projects/project.entity';
 import { ReposResolver } from './repos.resolver';
 import { ReposService } from './repos.service';
-import { ConfigModule } from '@nestjs/config';
+import { ConfigModule, ConfigService } from '@nestjs/config';
 import { ProjectsModule } from '../projects/projects.module';
-import { BullModule } from '@nestjs/bull';
+import { EXCHANGE_REPO } from './repos.constants';
-import { LIST_LOGS_TASK } from './repos.constants';
+import { RabbitMQModule } from '@golevelup/nestjs-rabbitmq';
-import { ListLogsConsumer } from './list-logs.consumer';
 
 @Module({
   imports: [
     TypeOrmModule.forFeature([Project]),
     ConfigModule,
     ProjectsModule,
-    BullModule.registerQueue({
+    RabbitMQModule.forRootAsync(RabbitMQModule, {
-      name: LIST_LOGS_TASK,
+      imports: [ConfigModule],
+      useFactory: (configService: ConfigService) => ({
+        uri: configService.get<string>('db.rabbitmq.uri'),
+        exchanges: [
+          {
+            name: EXCHANGE_REPO,
+            type: 'topic',
+            options: {
+              durable: true,
+              autoDelete: true,
+            },
+          },
+        ],
+      }),
+      inject: [ConfigService],
     }),
   ],
-  providers: [ReposResolver, ReposService, ListLogsConsumer],
+  providers: [ReposResolver, ReposService],
   exports: [ReposService],
 })
 export class ReposModule {}

src/repos/repos.service.spec.ts

@@ -9,6 +9,9 @@ import configuration from '../commons/config/configuration';
 import { PipelineTask } from '../pipeline-tasks/pipeline-task.entity';
 import { join } from 'path';
 import { readFile } from 'fs/promises';
+import { getLoggerToken, PinoLogger } from 'nestjs-pino';
+import { Nack } from '@golevelup/nestjs-rabbitmq';
+import { getInstanceName } from '../commons/utils/rabbit-mq';
 
 const getTest1Project = () =>
   ({
@@ -45,6 +48,10 @@ describe('ReposService', () => {
           provide: getRepositoryToken(Project),
           useFactory: repositoryMockFactory,
         },
+        {
+          provide: getLoggerToken(ReposService.name),
+          useValue: new PinoLogger({}),
+        },
       ],
     }).compile();
 
@@ -139,4 +146,57 @@ describe('ReposService', () => {
       );
     });
   });
+
+  describe('fetch', () => {
+    it('success', async () => {
+      const project = new Project();
+      const pipeline = new Pipeline();
+      pipeline.branch = 'test';
+      const fetch = jest.fn((_: any) => Promise.resolve());
+      pipeline.project = project;
+      const getGit = jest.spyOn(service, 'getGit').mockImplementation(() =>
+        Promise.resolve({
+          fetch,
+        } as any),
+      );
+      await expect(service.fetch(pipeline)).resolves.toEqual(getInstanceName());
+      expect(getGit).toBeCalledTimes(1);
+      expect(getGit.mock.calls[0]?.[0]).toEqual(project);
+      expect(fetch).toBeCalledTimes(1);
+      expect(fetch.mock.calls[0]?.[0]).toMatchObject([
+        'origin',
+        'test',
+        '--depth=100',
+      ]);
+    });
+    it('failed a', async () => {
+      const project = new Project();
+      const pipeline = new Pipeline();
+      pipeline.branch = 'test';
+      const fetch = jest.fn((_: any) => Promise.resolve());
+      pipeline.project = project;
+      const getGit = jest
+        .spyOn(service, 'getGit')
+        .mockImplementation(() => Promise.reject('error'));
+      await expect(service.fetch(pipeline)).resolves.toMatchObject(new Nack());
+      expect(getGit).toBeCalledTimes(1);
+      expect(getGit.mock.calls[0]?.[0]).toEqual(project);
+      expect(fetch).toBeCalledTimes(0);
+    });
+    it('failed b', async () => {
+      const project = new Project();
+      const pipeline = new Pipeline();
+      pipeline.branch = 'test';
+      const fetch = jest.fn((_: any) => Promise.reject('error'));
+      pipeline.project = project;
+      const getGit = jest.spyOn(service, 'getGit').mockImplementation(() =>
+        Promise.resolve({
+          fetch,
+        } as any),
+      );
+      await expect(service.fetch(pipeline)).resolves.toMatchObject(new Nack());
+      expect(getGit).toBeCalledTimes(1);
+      expect(fetch).toBeCalledTimes(1);
+    });
+  });
 });

src/repos/repos.service.ts

@@ -1,5 +1,4 @@
 import { ListLogsOption } from './models/list-logs.options';
-import { Pipeline } from './../pipelines/pipeline.entity';
 import { PipelineTask } from './../pipeline-tasks/pipeline-task.entity';
 import { Injectable, NotFoundException } from '@nestjs/common';
 import { InjectRepository } from '@nestjs/typeorm';
@@ -12,25 +11,32 @@ import { Project } from '../projects/project.entity';
 import { ListBranchesArgs } from './dtos/list-branches.args';
 import { ConfigService } from '@nestjs/config';
 import { Commit } from './dtos/log-list.model';
+import { Nack, RabbitRPC } from '@golevelup/nestjs-rabbitmq';
+import { Pipeline } from '../pipelines/pipeline.entity';
+import { InjectPinoLogger, Logger } from 'nestjs-pino';
+import {
+  EXCHANGE_REPO,
+  QUEUE_FETCH,
+  QUEUE_LIST_COMMITS,
+  ROUTE_FETCH,
+  ROUTE_LIST_COMMITS,
+} from './repos.constants';
+import { getSelfInstanceQueueKey } from '../commons/utils/rabbit-mq';
+import {
+  getInstanceName,
+  getSelfInstanceRouteKey,
+} from '../commons/utils/rabbit-mq';
 
 const DEFAULT_REMOTE_NAME = 'origin';
 const INFO_PATH = '@info';
 @Injectable()
 export class ReposService {
-  async listCommits(project: Project, branch?: string) {
-    const git = await this.getGit(project, undefined, { fetch: false });
-    const data = await git.log(
-      branch ? ['--branches', `remotes/origin/${branch}`, '--'] : ['--all'],
-    );
-    return data.all.map((it) => ({
-      ...it,
-      date: new Date(it.date),
-    }));
-  }
   constructor(
     @InjectRepository(Project)
     private readonly projectRepository: Repository<Project>,
     private readonly configService: ConfigService,
+    @InjectPinoLogger(ReposService.name)
+    private readonly logger: Logger,
   ) {}
 
   getWorkspaceRoot(project: Project): string {
@@ -111,4 +117,64 @@ export class ReposService {
     await this.checkout(task, path);
     return path;
   }
+
+  @RabbitRPC({
+    exchange: EXCHANGE_REPO,
+    routingKey: [
+      ROUTE_LIST_COMMITS,
+      getSelfInstanceRouteKey(ROUTE_LIST_COMMITS),
+    ],
+    queue: getSelfInstanceQueueKey(QUEUE_LIST_COMMITS),
+    queueOptions: {
+      autoDelete: true,
+    },
+  })
+  async listCommits(pipeline: Pipeline): Promise<Commit[] | Nack> {
+    const git = await this.getGit(pipeline.project, undefined, {
+      fetch: false,
+    });
+    try {
+      const data = await git.log([
+        '-100',
+        '--branches',
+        `remotes/origin/${pipeline.branch}`,
+        '--',
+      ]);
+      return data.all.map(
+        (it) =>
+          ({
+            ...it,
+            date: new Date(it.date),
+          } as Commit),
+      );
+    } catch (error) {
+      this.logger.error(
+        { error, pipeline },
+        '[listCommits] %s',
+        error?.message,
+      );
+      return new Nack();
+    }
+  }
+
+  @RabbitRPC({
+    exchange: EXCHANGE_REPO,
+    routingKey: [ROUTE_FETCH, getSelfInstanceRouteKey(ROUTE_FETCH)],
+    queue: getSelfInstanceQueueKey(QUEUE_FETCH),
+    queueOptions: {
+      autoDelete: true,
+    },
+  })
+  async fetch(pipeline: Pipeline): Promise<string | null | Nack> {
+    try {
+      const git = await this.getGit(pipeline.project, undefined, {
+        fetch: false,
+      });
+      await git.fetch(['origin', pipeline.branch, '--depth=100']);
+      return getInstanceName();
+    } catch (error) {
+      this.logger.error({ error, pipeline }, '[fetch] %s', error?.message);
+      return new Nack();
+    }
+  }
 }