feat(repos): 添加消息队列分发。BAK

.vscode/settings.json

@@ -1,7 +1,5 @@
 {
   "cSpell.words": [
-    "Repos",
-    "lpush",
-    "rpop"
+    "Repos"
   ]
 }

config.yml.example

@@ -1,7 +1,6 @@
 env: dev
 http:
   port: 7122
-
 db:
   postgres:
     host: 192.168.31.194
@@ -9,10 +8,9 @@ db:
     database: fennec
     username: fennec
     password:
-  redis:
-    host: 192.168.31.194
+redis:
+  mq:
+    host: localhost
     port: 6379
-    password:
-    prefix: fennec
 workspaces:
   root: '/Users/ivanli/Projects/fennec/workspaces'

docs/ci-cd.md

@@ -1,33 +0,0 @@
-# CI/CD 流程
-0. 准备
-  - project information
-  - commit hash
-1. checkout
-2. install dependencies
-3. run test script
-5. run deploy script
-6. clear workspace
-
-## 流水线任务单元描述
-```json
-{
-  "version": 1,
-  "unit": {
-    "install-dependencies": {
-      "script": "npm ci"
-    },
-    "test": {
-      "script": "npm test"
-    },
-    "build": {
-      "script": "npm build"
-    },
-    "deploy": {
-      "script": [
-        "npm build"
-      ]
-    }
-  }
-}
-
-```

package-lock.json

@@ -9087,6 +9087,14 @@
         }
       }
     },
+    "mkdirp": {
+      "version": "0.5.5",
+      "resolved": "https://registry.npmjs.org/mkdirp/-/mkdirp-0.5.5.tgz",
+      "integrity": "sha512-NKmAlESf6jMGym1++R0Ra7wvhV+wFW63FaSOFPwRahvea0gMUcGUhVeAg/0BC0wiv9ih5NYPB1Wn1UEI1/L+xQ==",
+      "requires": {
+        "minimist": "^1.2.5"
+      }
+    },
     "moment": {
       "version": "2.29.1",
       "resolved": "https://registry.npmjs.org/moment/-/moment-2.29.1.tgz",
@@ -9118,16 +9126,6 @@
         "on-finished": "^2.3.0",
         "type-is": "^1.6.4",
         "xtend": "^4.0.0"
-      },
-      "dependencies": {
-        "mkdirp": {
-          "version": "0.5.5",
-          "resolved": "https://registry.npmjs.org/mkdirp/-/mkdirp-0.5.5.tgz",
-          "integrity": "sha512-NKmAlESf6jMGym1++R0Ra7wvhV+wFW63FaSOFPwRahvea0gMUcGUhVeAg/0BC0wiv9ih5NYPB1Wn1UEI1/L+xQ==",
-          "requires": {
-            "minimist": "^1.2.5"
-          }
-        }
       }
     },
     "multimatch": {
@@ -9491,14 +9489,6 @@
             "minipass": "^2.9.0"
           }
         },
-        "mkdirp": {
-          "version": "0.5.5",
-          "resolved": "https://registry.npmjs.org/mkdirp/-/mkdirp-0.5.5.tgz",
-          "integrity": "sha512-NKmAlESf6jMGym1++R0Ra7wvhV+wFW63FaSOFPwRahvea0gMUcGUhVeAg/0BC0wiv9ih5NYPB1Wn1UEI1/L+xQ==",
-          "requires": {
-            "minimist": "^1.2.5"
-          }
-        },
         "rimraf": {
           "version": "2.7.1",
           "resolved": "https://registry.npmjs.org/rimraf/-/rimraf-2.7.1.tgz",

src/app.module.ts

@@ -7,10 +7,8 @@ import { AppResolver } from './app.resolver';
 import { AppService } from './app.service';
 import { ProjectsModule } from './projects/projects.module';
 import { ReposModule } from './repos/repos.module';
-import { PipelinesModule } from './pipelines/pipelines.module';
-import { PipelineTasksModule } from './pipeline-tasks/pipeline-tasks.module';
 import configuration from './commons/config/configuration';
-import { RedisModule } from 'nestjs-redis';
+import { BullModule } from '@nestjs/bull';
 
 @Module({
   imports: [
@@ -40,20 +38,18 @@ import { RedisModule } from 'nestjs-redis';
       }),
       inject: [ConfigService],
     }),
-    ProjectsModule,
-    ReposModule,
-    PipelinesModule,
-    PipelineTasksModule,
-    RedisModule.forRootAsync({
+    BullModule.forRootAsync({
       imports: [ConfigModule],
-      useFactory: (configService: ConfigService) => ({
-        host: configService.get<string>('db.redis.host', 'localhost'),
-        port: configService.get<number>('db.redis.port', 6379),
-        password: configService.get<string>('db.redis.password', ''),
-        keyPrefix: configService.get<string>('db.redis.prefix', 'fennec'),
+      useFactory: async (configService: ConfigService) => ({
+        redis: {
+          host: configService.get<string>('redis.mq.host', 'localhost'),
+          port: configService.get<number>('redis.mq.port', 6379),
+        },
       }),
       inject: [ConfigService],
     }),
+    ProjectsModule,
+    ReposModule,
   ],
   controllers: [AppController],
   providers: [AppService, AppResolver],

src/commons/entities/app-base-entity.ts

@@ -1,7 +1,6 @@
 import { Field, ID, ObjectType } from '@nestjs/graphql';
 import {
   CreateDateColumn,
-  DeleteDateColumn,
   PrimaryGeneratedColumn,
   UpdateDateColumn,
 } from 'typeorm';
@@ -17,7 +16,4 @@ export class AppBaseEntity {
 
   @UpdateDateColumn({ select: false })
   updatedAt: Date;
-
-  @DeleteDateColumn({ select: false })
-  deletedAt?: Date;
 }

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

@@ -1,27 +1,55 @@
 import {
-  ExceptionFilter,
-  Catch,
   ArgumentsHost,
+  Catch,
+  ExceptionFilter,
   HttpException,
+  HttpStatus,
 } from '@nestjs/common';
-import { ApolloError } from 'apollo-server-errors';
+import { EntityNotFoundError } from 'typeorm/error/EntityNotFoundError';
 
-@Catch(HttpException)
-export class HttpExceptionFilter implements ExceptionFilter {
-  catch(exception: HttpException, host: ArgumentsHost) {
-    const message = exception.message;
-    const extensions: Record<string, any> = {};
-    const err = exception.getResponse();
-    if (typeof err === 'string') {
-      extensions.message = err;
+@Catch()
+export class AllExceptionsFilter implements ExceptionFilter {
+  catch(exception: any, host: ArgumentsHost) {
+    const ctx = host.switchToHttp();
+    const response = ctx.getResponse();
+    const request = ctx.getRequest();
+
+    const status =
+      exception instanceof HttpException
+        ? exception.getStatus()
+        : HttpStatus.INTERNAL_SERVER_ERROR;
+
+    if (exception instanceof HttpException) {
+      const ex = exception.getResponse();
+      if (ex instanceof Object) {
+        response.status(status).json({
+          ...ex,
+          timestamp: Date.now(),
+          path: request.url,
+        });
       } else {
-      Object.assign(extensions, (err as any).extension);
-      extensions.message = (err as any).message;
+        response.status(status).json({
+          message: ex,
+          timestamp: Date.now(),
+          path: request.url,
+        });
+      }
+    } else if (exception instanceof EntityNotFoundError) {
+      response.status(HttpStatus.NOT_FOUND).json({
+        message: '资源未找到!',
+        timestamp: Date.now(),
+        path: request.url,
+      });
+    } else {
+      console.error('服务器内部错误');
+      console.error(exception);
+      response.status(HttpStatus.INTERNAL_SERVER_ERROR).json({
+        code: status,
+        timestamp: new Date().toISOString(),
+        message: '服务器内部错误',
+        error: exception,
+        path: request.url,
+      });
     }
-    return new ApolloError(
-      message,
-      exception.getStatus().toString(),
-      extensions,
-    );
   }
 }

src/commons/pipes/sanitize.pipe.ts

@@ -1,15 +0,0 @@
-import { ArgumentMetadata, Injectable, PipeTransform } from '@nestjs/common';
-import { sanitize } from '@neuralegion/class-sanitizer/dist';
-
-@Injectable()
-export class SanitizePipe implements PipeTransform {
-  transform(value: any, metadata: ArgumentMetadata) {
-    // console.log(value, typeof value);
-    if (value instanceof Object) {
-      value = Object.assign(new metadata.metatype(), value);
-      sanitize(value);
-      // console.log(value);
-    }
-    return value;
-  }
-}

src/main.ts

@@ -2,19 +2,11 @@ import { ValidationPipe } from '@nestjs/common';
 import { ConfigService } from '@nestjs/config';
 import { NestFactory } from '@nestjs/core';
 import { AppModule } from './app.module';
-import { HttpExceptionFilter } from './commons/filters/all.exception-filter';
-import { SanitizePipe } from './commons/pipes/sanitize.pipe';
 
 async function bootstrap() {
   const app = await NestFactory.create(AppModule);
   const configService = app.get(ConfigService);
-  app.useGlobalPipes(new SanitizePipe());
-  app.useGlobalPipes(
-    new ValidationPipe({
-      transform: true,
-    }),
-  );
-  app.useGlobalFilters(new HttpExceptionFilter());
+  app.useGlobalPipes(new ValidationPipe());
   await app.listen(configService.get<number>('http.port'));
 }
 bootstrap();

src/pipeline-tasks/dtos/create-pipeline-task.input.ts

@@ -1,12 +0,0 @@
-import { Field, InputType } from '@nestjs/graphql';
-import { PipelineUnits } from '../enums/pipeline-units.enum';
-
-@InputType()
-export class CreatePipelineTaskInput {
-  pipelineId: string;
-
-  commit: string;
-
-  @Field(() => PipelineUnits)
-  units: PipelineUnits[];
-}

src/pipeline-tasks/enums/pipeline-units.enum.ts

@@ -1,14 +0,0 @@
-import { registerEnumType } from '@nestjs/graphql';
-
-export enum PipelineUnits {
-  checkout = 'checkout',
-  installDependencies = 'installDependencies',
-  test = 'test',
-  deploy = 'deploy',
-  cleanUp = 'cleanUp',
-}
-
-registerEnumType(PipelineUnits, {
-  name: 'PipelineUnits',
-  description: '流水线单元',
-});

src/pipeline-tasks/enums/task-statuses.enum.ts

@@ -1,6 +0,0 @@
-export enum TaskStatuses {
-  success = 'success',
-  failed = 'failed',
-  working = 'working',
-  pending = 'pending',
-}

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

@@ -1,15 +0,0 @@
-import { PipelineTask } from './../pipeline-task.entity';
-
-export class PipelineTaskLogMessage {
-  task: PipelineTask;
-  time: Date;
-  message: string;
-
-  static create(task: PipelineTask, message: string) {
-    return Object.assign(new PipelineTaskLogMessage(), {
-      task,
-      message,
-      time: new Date(),
-    });
-  }
-}

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

@@ -1,9 +0,0 @@
-import { TaskStatuses } from '../enums/task-statuses.enum';
-import { PipelineUnits } from '../enums/pipeline-units.enum';
-export class PipelineTaskLogs {
-  unit: PipelineUnits;
-  status: TaskStatuses;
-  startedAt?: Date;
-  endedAt?: Date;
-  logs = '';
-}

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

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

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

@@ -1,13 +0,0 @@
-import { Field, InputType, ObjectType } from '@nestjs/graphql';
-import {
-  PipelineUnits,
-  PipelineUnits as PipelineUnitTypes,
-} from '../enums/pipeline-units.enum';
-
-@ObjectType()
-@InputType('WorkUnitInput')
-export class WorkUnit {
-  @Field(() => PipelineUnits)
-  type: PipelineUnitTypes;
-  scripts: string[];
-}

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

@@ -1,192 +0,0 @@
-import { PIPELINE_TASK_LOG_QUEUE } from './pipeline-tasks.constants';
-import { Test, TestingModule } from '@nestjs/testing';
-import { Job, Queue } from 'bull';
-import { join } from 'path';
-import { ReposService } from '../repos/repos.service';
-import { PipelineUnits } from './enums/pipeline-units.enum';
-import { PipelineTaskConsumer } from './pipeline-task.consumer';
-import { PipelineTask } from './pipeline-task.entity';
-import { PipelineTasksService } from './pipeline-tasks.service';
-import { getQueueToken } from '@nestjs/bull';
-import { PipelineTaskLogMessage } from './models/pipeline-task-log-message.module';
-import { Pipeline } from '../pipelines/pipeline.entity';
-import { Project } from '../projects/project.entity';
-import { TaskStatuses } from './enums/task-statuses.enum';
-
-describe('PipelineTaskConsumer', () => {
-  let consumer: PipelineTaskConsumer;
-  let tasksService: PipelineTasksService;
-  let logQueue: Queue<PipelineTaskLogMessage>;
-  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,
-          },
-        },
-        PipelineTaskConsumer,
-        {
-          provide: getQueueToken(PIPELINE_TASK_LOG_QUEUE),
-          useValue: {
-            add: () => undefined,
-          },
-        },
-      ],
-    }).compile();
-
-    tasksService = module.get(PipelineTasksService);
-    consumer = module.get(PipelineTaskConsumer);
-    logQueue = module.get(getQueueToken(PIPELINE_TASK_LOG_QUEUE));
-  });
-
-  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', () => {
-    it('should success and log right message', async () => {
-      const add = jest.spyOn(logQueue, 'add');
-      await expect(
-        consumer
-          .runScript(
-            'node one-second-work.js',
-            join(__dirname, '../../test/data'),
-          )
-          .then((arr) => arr.join('')),
-      ).resolves.toMatch(/10.+20.+30.+40.+50.+60.+70.+80.+90/s);
-      // expect(add).toHaveBeenCalledTimes(10);
-      expect(
-        ((add.mock.calls[8][0] as unknown) as PipelineTaskLogMessage).message,
-      ).toMatch(/^90/);
-    });
-    it('should failed and log right message', async () => {
-      const add = jest.spyOn(logQueue, 'add');
-      await expect(
-        consumer.runScript(
-          'node bad-work.js',
-          join(__dirname, '../../test/data'),
-        ),
-      ).rejects.toThrowError(/Error Message/);
-      // expect(add).toHaveBeenCalledTimes(8);
-      const logs = add.mock.calls
-        .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 add = jest.spyOn(logQueue, 'add');
-      await expect(
-        consumer.runScript(
-          'node bad-work.js',
-          join(__dirname, '../../test/data'),
-          task,
-        ),
-      ).rejects.toThrowError(/Error Message 2/);
-      expect(
-        ((add.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('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 () => []);
-      const updateTask = jest.spyOn(tasksService, 'updateTask');
-
-      await consumer.doTask(job);
-
-      expect(runScript).toHaveBeenCalledTimes(1);
-      expect(updateTask).toHaveBeenCalledTimes(1);
-      const taskDto: PipelineTask = updateTask.mock.calls[0][0];
-      expect(taskDto.logs).toHaveLength(2);
-      expect(taskDto.logs[0].status).toEqual(TaskStatuses.success);
-      expect(taskDto.logs[0].unit).toEqual(PipelineUnits.checkout);
-      expect(taskDto.logs[1].logs).toMatch(/Hello, Fennec!/);
-    });
-    it('should log error message', async () => {
-      const job: Job = ({
-        data: task,
-        update: jest.fn().mockImplementation(() => undefined),
-      } as unknown) as Job;
-
-      const runScript = jest
-        .spyOn(consumer, 'runScript')
-        .mockImplementation(async () => {
-          throw new Error('bad message');
-        });
-      const updateTask = jest.spyOn(tasksService, 'updateTask');
-
-      await consumer.doTask(job);
-
-      expect(updateTask).toHaveBeenCalledTimes(1);
-      const taskDto: PipelineTask = updateTask.mock.calls[0][0];
-      expect(taskDto.logs).toHaveLength(2);
-      expect(taskDto.logs[0].status).toEqual(TaskStatuses.success);
-      expect(taskDto.logs[1].status).toEqual(TaskStatuses.failed);
-      expect(taskDto.logs[1].logs).toMatch(/bad message/);
-    });
-  });
-});

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

@@ -1,118 +0,0 @@
-import { PipelineTaskLogs } from './models/pipeline-task-logs.model';
-import { ReposService } from './../repos/repos.service';
-import {
-  InjectQueue,
-  OnQueueCompleted,
-  Process,
-  Processor,
-} from '@nestjs/bull';
-import { Job, Queue } from 'bull';
-import { spawn } from 'child_process';
-import { PipelineTask } from './pipeline-task.entity';
-import {
-  PIPELINE_TASK_LOG_QUEUE,
-  PIPELINE_TASK_QUEUE,
-} from './pipeline-tasks.constants';
-import { 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';
-@Processor(PIPELINE_TASK_QUEUE)
-export class PipelineTaskConsumer {
-  constructor(
-    private readonly service: PipelineTasksService,
-    private readonly reposService: ReposService,
-    @InjectQueue(PIPELINE_TASK_LOG_QUEUE)
-    private readonly logQueue: Queue<PipelineTaskLogMessage>,
-  ) {}
-  @Process()
-  async doTask({ data: task, update }: Job<PipelineTask>) {
-    if (task.pipeline.workUnitMetadata.version !== 1) {
-      throw new ApplicationException(
-        'work unit metadata version is not match.',
-      );
-    }
-
-    const workspaceRoot = this.reposService.getWorkspaceRootByTask(task);
-
-    const units = task.units.map(
-      (type) =>
-        task.pipeline.workUnitMetadata.units.find(
-          (unit) => unit.type === type,
-        ) ?? { type: type, scripts: [] },
-    );
-
-    try {
-      for (const unit of units) {
-        const unitLog = new PipelineTaskLogs();
-        unitLog.unit = unit.type;
-        unitLog.startedAt = new Date();
-        try {
-          // 检出代码时，不执行其他脚本。
-          if (unit.type === PipelineUnits.checkout) {
-            await this.reposService.checkout(task, workspaceRoot);
-            unitLog.status = TaskStatuses.success;
-            continue;
-          }
-          for (const script of unit.scripts) {
-            unitLog.logs += `[RUN SCRIPT] ${script}`;
-            const messages = await this.runScript(script, workspaceRoot, task);
-            unitLog.logs += messages.join('');
-          }
-          unitLog.status = TaskStatuses.success;
-        } catch (err) {
-          unitLog.status = TaskStatuses.failed;
-          unitLog.logs += err.message;
-          throw err;
-        } finally {
-          unitLog.endedAt = new Date();
-          task.logs.push(unitLog);
-        }
-      }
-    } catch (err) {
-      console.log(err);
-    } finally {
-      task = await this.service.updateTask(task);
-      update(task);
-    }
-  }
-
-  async runScript(
-    script: string,
-    workspaceRoot: string,
-    task?: PipelineTask,
-  ): Promise<string[]> {
-    return new Promise((resolve, reject) => {
-      const errorMessages: string[] = [];
-      const logs: string[] = [];
-      const sub = spawn(script, {
-        shell: true,
-        cwd: workspaceRoot,
-      });
-      sub.stderr.on('data', (data: Buffer) => {
-        const str = data.toString();
-        errorMessages.push(str);
-        logs.push(str);
-        this.logQueue.add(PipelineTaskLogMessage.create(task, str));
-      });
-      sub.stdout.on('data', (data: Buffer) => {
-        const str = data.toString();
-        logs.push(str);
-        this.logQueue.add(PipelineTaskLogMessage.create(task, str));
-      });
-      sub.addListener('close', (code) => {
-        if (code === 0) {
-          sub.stdout;
-          return resolve(logs);
-        }
-        return reject(new ApplicationException(errorMessages.join('')));
-      });
-    });
-  }
-
-  @OnQueueCompleted()
-  onCompleted(job: Job<PipelineTask>) {
-    this.service.doNextTask(job.data.pipeline);
-  }
-}

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

@@ -1,34 +0,0 @@
-import { AppBaseEntity } from './../commons/entities/app-base-entity';
-import { 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';
-
-@ObjectType()
-@Entity()
-export class PipelineTask extends AppBaseEntity {
-  @ManyToOne(() => Pipeline)
-  pipeline: Pipeline;
-  @Column()
-  pipelineId: string;
-
-  @Column()
-  commit: string;
-
-  @Column({ type: 'enum', enum: PipelineUnits, array: true })
-  units: PipelineUnits[];
-
-  @Column({ type: 'jsonb', default: '[]' })
-  logs: PipelineTaskLogs[];
-
-  @Column({ type: 'enum', enum: TaskStatuses, default: TaskStatuses.pending })
-  status: TaskStatuses;
-
-  @Column()
-  startedAt: Date;
-
-  @Column()
-  endedAt: Date;
-}

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

@@ -1,29 +0,0 @@
-import { Module } from '@nestjs/common';
-import { PipelineTasksService } from './pipeline-tasks.service';
-import { PipelineTasksResolver } from './pipeline-tasks.resolver';
-import { TypeOrmModule } from '@nestjs/typeorm';
-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 {
-  PIPELINE_TASK_QUEUE,
-  PIPELINE_TASK_LOG_QUEUE,
-} from './pipeline-tasks.constants';
-
-@Module({
-  imports: [
-    TypeOrmModule.forFeature([PipelineTask, Pipeline]),
-    BullModule.registerQueue(
-      {
-        name: PIPELINE_TASK_QUEUE,
-      },
-      { name: PIPELINE_TASK_LOG_QUEUE },
-    ),
-    RedisModule,
-    ReposModule,
-  ],
-  providers: [PipelineTasksService, PipelineTasksResolver],
-})
-export class PipelineTasksModule {}

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

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

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

@@ -1,4 +0,0 @@
-import { Resolver } from '@nestjs/graphql';
-
-@Resolver()
-export class PipelineTasksResolver {}

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

@@ -1,190 +0,0 @@
-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';
-
-describe('PipelineTasksService', () => {
-  let service: PipelineTasksService;
-  let module: TestingModule;
-  let taskRepository: Repository<PipelineTask>;
-  let pipelineRepository: Repository<Pipeline>;
-  const getBasePipeline = () =>
-    ({
-      id: 'test',
-      name: '测试流水线',
-      branch: 'master',
-      workUnitMetadata: {},
-      project: {
-        id: 'test-project',
-      },
-    } as Pipeline);
-  let redisClient;
-  let taskQueue: Queue;
-  const getTask = () =>
-    ({
-      pipelineId: 'test',
-      commit: 'test',
-      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,
-        {
-          provide: getRepositoryToken(PipelineTask),
-          useValue: new Repository(),
-        },
-        {
-          provide: getRepositoryToken(Pipeline),
-          useValue: new Repository(),
-        },
-        {
-          provide: getQueueToken(PIPELINE_TASK_QUEUE),
-          useValue: taskQueue,
-        },
-        {
-          provide: RedisService,
-          useValue: {
-            getClient: jest.fn(() => redisClient),
-          },
-        },
-      ],
-    }).compile();
-
-    service = module.get<PipelineTasksService>(PipelineTasksService);
-    taskRepository = module.get(getRepositoryToken(PipelineTask));
-    pipelineRepository = module.get(getRepositoryToken(Pipeline));
-    jest
-      .spyOn(taskRepository, 'save')
-      .mockImplementation(async (data: any) => data);
-    jest
-      .spyOn(taskRepository, 'create')
-      .mockImplementation((data: any) => data);
-  });
-
-  it('should be defined', () => {
-    expect(service).toBeDefined();
-  });
-
-  describe('addTask', () => {
-    beforeEach(() => {
-      jest
-        .spyOn(pipelineRepository, 'findOneOrFail')
-        .mockImplementation(async () => getBasePipeline());
-    });
-    it('pipeline not found', async () => {
-      jest.spyOn(taskRepository, 'findOneOrFail').mockImplementation(() => {
-        throw new EntityNotFoundError(Pipeline, {});
-      });
-      await expect(
-        service.addTask({ pipelineId: 'test', commit: 'test', units: [] }),
-      ).rejects;
-    });
-    it('create task on db', async () => {
-      const save = jest
-        .spyOn(taskRepository, 'save')
-        .mockImplementation(async (data: any) => data);
-      jest
-        .spyOn(service, 'doNextTask')
-        .mockImplementation(async () => undefined);
-      await service.addTask({ pipelineId: 'test', commit: 'test', units: [] }),
-        expect(save.mock.calls[0][0]).toMatchObject({
-          pipelineId: 'test',
-          commit: 'test',
-          units: [],
-        });
-    });
-    it('add task', async () => {
-      const lpush = jest.spyOn(redisClient, 'lpush');
-      const doNextTask = jest.spyOn(service, 'doNextTask');
-      jest
-        .spyOn(service, 'doNextTask')
-        .mockImplementation(async () => undefined);
-      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;
-
-      const incr = jest
-        .spyOn(redisClient, 'incr')
-        .mockImplementation(() => remainTimes--);
-      const rpop = jest.spyOn(redisClient, 'rpop');
-      const decr = jest.spyOn(redisClient, 'decr');
-      const add = jest.spyOn(taskQueue, 'add');
-
-      await service.doNextTask(getBasePipeline());
-
-      expect(rpop).toHaveBeenCalledTimes(1);
-      expect(incr).toHaveBeenCalledTimes(3);
-      expect(decr).toHaveBeenCalledTimes(3);
-      expect(add).toHaveBeenCalledWith(getTask());
-    });
-    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,80 +0,0 @@
-import { Injectable } from '@nestjs/common';
-import { InjectRepository } from '@nestjs/typeorm';
-import { PipelineTask } from './pipeline-task.entity';
-import { Repository } from 'typeorm';
-import { CreatePipelineTaskInput } from './dtos/create-pipeline-task.input';
-import { 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';
-
-@Injectable()
-export class PipelineTasksService {
-  constructor(
-    @InjectRepository(PipelineTask)
-    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,
-  ) {}
-  async addTask(dto: CreatePipelineTaskInput) {
-    const pipeline = await this.pipelineRepository.findOneOrFail({
-      where: { id: dto.pipelineId },
-      relations: ['project'],
-    });
-    const task = await this.repository.save(this.repository.create(dto));
-    task.pipeline = pipeline;
-
-    const tasksKey = this.getRedisTokens(pipeline)[1];
-    const redis = this.redis.getClient();
-    await redis.lpush(tasksKey, JSON.stringify(task));
-    await this.doNextTask(pipeline);
-  }
-
-  async doNextTask(pipeline: Pipeline) {
-    const [lckKey, tasksKey] = this.getRedisTokens(pipeline);
-    const redis = this.redis.getClient();
-
-    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) {
-      await this.queue.add(task);
-    }
-  }
-
-  async updateTask(task: PipelineTask) {
-    return await this.repository.save(task);
-  }
-
-  getRedisTokens(pipeline: Pipeline): [string, string] {
-    return [`pipeline-${pipeline.id}:lck`, `pipeline-${pipeline.id}:tasks`];
-  }
-}

src/pipelines/dtos/create-pipeline.input.ts

@@ -1,27 +0,0 @@
-import { InputType } from '@nestjs/graphql';
-import { WorkUnitMetadata } from '../../pipeline-tasks/models/work-unit-metadata.model';
-import {
-  IsObject,
-  IsOptional,
-  IsString,
-  IsUUID,
-  MaxLength,
-} from 'class-validator';
-
-@InputType({ isAbstract: true })
-export class CreatePipelineInput {
-  @IsUUID()
-  projectId: string;
-
-  @IsString()
-  @MaxLength(100)
-  branch: string;
-
-  @IsString()
-  @MaxLength(32)
-  name: string;
-
-  @IsOptional()
-  @IsObject()
-  workUnitMetadata: WorkUnitMetadata;
-}

src/pipelines/dtos/list-pipelines.args.ts

@@ -1,8 +0,0 @@
-import { ArgsType } from '@nestjs/graphql';
-import { IsUUID } from 'class-validator';
-
-@ArgsType()
-export class ListPipelineArgs {
-  @IsUUID()
-  projectId?: string;
-}

src/pipelines/dtos/update-pipeline.input.ts

@@ -1,5 +0,0 @@
-import { InputType } from '@nestjs/graphql';
-import { CreatePipelineInput } from './create-pipeline.input';
-
-@InputType()
-export class UpdatePipelineInput extends CreatePipelineInput {}

src/pipelines/pipeline.entity.ts

@@ -1,23 +0,0 @@
-import { Column, Entity, ManyToOne } from 'typeorm';
-import { AppBaseEntity } from '../commons/entities/app-base-entity';
-import { Project } from '../projects/project.entity';
-import { ObjectType } from '@nestjs/graphql';
-import { WorkUnitMetadata } from '../pipeline-tasks/models/work-unit-metadata.model';
-
-@ObjectType()
-@Entity()
-export class Pipeline extends AppBaseEntity {
-  @ManyToOne(() => Project)
-  project: Project;
-  @Column()
-  projectId: string;
-
-  @Column({ comment: 'eg: remotes/origin/master' })
-  branch: string;
-
-  @Column()
-  name: string;
-
-  @Column({ type: 'jsonb' })
-  workUnitMetadata: WorkUnitMetadata;
-}

src/pipelines/pipelines.module.ts

@@ -1,18 +0,0 @@
-import { Module } from '@nestjs/common';
-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';
-
-@Module({
-  imports: [
-    TypeOrmModule.forFeature([Pipeline]),
-    BullModule.registerQueue({
-      name: LIST_LOGS_TASK,
-    }),
-  ],
-  providers: [PipelinesResolver, PipelinesService],
-})
-export class PipelinesModule {}

src/pipelines/pipelines.resolver.spec.ts

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

src/pipelines/pipelines.resolver.ts

@@ -1,50 +0,0 @@
-import { Args, Mutation, Query, Resolver } from '@nestjs/graphql';
-import { CreatePipelineInput } from './dtos/create-pipeline.input';
-import { UpdatePipelineInput } from './dtos/update-pipeline.input';
-import { Pipeline } from './pipeline.entity';
-import { PipelinesService } from './pipelines.service';
-import { ListPipelineArgs } from './dtos/list-pipelines.args';
-import { LogList } from '../repos/dtos/log-list.model';
-
-@Resolver()
-export class PipelinesResolver {
-  constructor(private readonly service: PipelinesService) {}
-  @Query(() => [Pipeline])
-  async listPipelines(@Args() dto: ListPipelineArgs) {
-    return await this.service.list(dto);
-  }
-
-  @Query(() => Pipeline)
-  async findPipeline(@Args('id', { type: () => String }) id: string) {
-    return await this.service.findOne(id);
-  }
-
-  @Mutation(() => Pipeline)
-  async createPipeline(
-    @Args('pipeline', { type: () => CreatePipelineInput })
-    dto: UpdatePipelineInput,
-  ) {
-    return await this.service.create(dto);
-  }
-
-  @Mutation(() => Pipeline)
-  async modifyPipeline(
-    @Args('id', { type: () => String }) id: string,
-    @Args('Pipeline', { type: () => UpdatePipelineInput })
-    dto: UpdatePipelineInput,
-  ) {
-    const tmp = await this.service.update(id, dto);
-    console.log(tmp);
-    return tmp;
-  }
-
-  @Mutation(() => Number)
-  async deletePipeline(@Args('id', { type: () => String }) id: string) {
-    return await this.service.remove(id);
-  }
-
-  @Query(() => String)
-  async listLogsForPipeline(@Args('id', { type: () => String }) id: string) {
-    return await this.service.listLogsForPipeline(id);
-  }
-}

src/pipelines/pipelines.service.spec.ts

@@ -1,69 +0,0 @@
-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 { 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(), {
-      id: 'test-pipeline',
-      name: 'pipeline',
-      branch: 'master',
-      projectId: 'test-project',
-      project: Object.assign(new Project(), {
-        id: 'test-project',
-        name: 'project',
-      } as Project),
-    } as Pipeline);
-
-    const module: TestingModule = await Test.createTestingModule({
-      providers: [
-        PipelinesService,
-        {
-          provide: getRepositoryToken(Pipeline),
-          useValue: {
-            findOneOrFail: jest.fn().mockImplementation(() => pipeline),
-          },
-        },
-        {
-          provide: getQueueToken(LIST_LOGS_TASK),
-          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(1);
-      expect(add).toBeCalledWith({
-        project: pipeline.project,
-        branch: pipeline.branch,
-      });
-    });
-  });
-});

src/pipelines/pipelines.service.ts

@@ -1,57 +0,0 @@
-import { Injectable } from '@nestjs/common';
-import { InjectRepository } from '@nestjs/typeorm';
-import { Pipeline } from './pipeline.entity';
-import { Repository } from 'typeorm';
-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 { LIST_LOGS_TASK } from '../repos/repos.constants';
-import { Queue } from 'bull';
-import { ListLogsOption } from '../repos/models/list-logs.options';
-
-@Injectable()
-export class PipelinesService extends BaseDbService<Pipeline> {
-  readonly uniqueFields: Array<Array<keyof Pipeline>> = [
-    ['branch', 'projectId'],
-  ];
-  constructor(
-    @InjectRepository(Pipeline)
-    readonly repository: Repository<Pipeline>,
-    @InjectQueue(LIST_LOGS_TASK)
-    private readonly listLogsQueue: Queue<ListLogsOption>,
-  ) {
-    super(repository);
-  }
-  async list(dto: ListPipelineArgs) {
-    return this.repository.find(dto);
-  }
-
-  async create(dto: CreatePipelineInput) {
-    await this.isDuplicateEntity(dto);
-    return await this.repository.save(this.repository.create(dto));
-  }
-
-  async update(id: string, dto: UpdatePipelineInput) {
-    await this.isDuplicateEntityForUpdate(id, dto);
-    const old = await this.findOne(id);
-    return await this.repository.save(this.repository.merge(old, dto));
-  }
-
-  async remove(id: string) {
-    return (await this.repository.softDelete({ id })).affected;
-  }
-
-  async listLogsForPipeline(id: string) {
-    const pipeline = await this.repository.findOneOrFail({
-      where: { id },
-      relations: ['project'],
-    });
-    const job = await this.listLogsQueue.add({
-      project: pipeline.project,
-      branch: pipeline.branch,
-    });
-    return job.id;
-  }
-}

src/projects/dtos/create-project.input.ts

@@ -3,7 +3,6 @@ import {
   IsOptional,
   IsString,
   IsUrl,
-  Matches,
   MaxLength,
   MinLength,
 } from 'class-validator';
@@ -20,9 +19,7 @@ export class CreateProjectInput {
   @MinLength(2)
   comment: string;
 
-  @Matches(
-    /^(?:ssh:\/\/)?(?:[\w\d-_]+@)(?:[\w\d-_]+\.)*\w{2,10}(?::\d{1,5})?(?:\/[\w\d-_.]+)*/,
-  )
+  @IsUrl({ protocols: ['ssh'] })
   @MaxLength(256)
   sshUrl: string;
 

src/projects/project.entity.ts

@@ -22,4 +22,7 @@ export class Project extends AppBaseEntity {
 
   @Column({ nullable: true })
   webHookSecret?: string;
+
+  @DeleteDateColumn()
+  deletedAt?: Date;
 }

src/projects/projects.service.ts

@@ -4,7 +4,6 @@ import { BaseDbService } from '../commons/services/base-db.service';
 import { Repository } from 'typeorm';
 import { CreateProjectInput } from './dtos/create-project.input';
 import { Project } from './project.entity';
-import { UpdateProjectInput } from './dtos/update-project.input';
 
 @Injectable()
 export class ProjectsService extends BaseDbService<Project> {
@@ -25,7 +24,7 @@ export class ProjectsService extends BaseDbService<Project> {
     return await this.repository.save(this.repository.create(dto));
   }
 
-  async update(id: string, dto: UpdateProjectInput) {
+  async update(id: string, dto: CreateProjectInput) {
     await this.isDuplicateEntityForUpdate(id, dto);
     const old = await this.findOne(id);
     return await this.repository.save(this.repository.merge(old, dto));

src/repos/dtos/checkout.input.ts

@@ -1,16 +0,0 @@
-import { InputType } from '@nestjs/graphql';
-import { IsOptional, IsString, IsUUID } from 'class-validator';
-
-@InputType()
-export class CheckoutInput {
-  @IsUUID()
-  pipelineId: string;
-
-  @IsString()
-  @IsOptional()
-  branch?: string;
-
-  @IsString()
-  @IsOptional()
-  commitNumber?: string;
-}

src/repos/dtos/list-logs.args.ts

@@ -1,4 +1,4 @@
-import { InputType } from '@nestjs/graphql';
+import { InputType, ObjectType } from '@nestjs/graphql';
 import { IsOptional, IsString, IsUUID } from 'class-validator';
 
 @InputType()

src/repos/models/list-logs.options.ts

@@ -1,5 +0,0 @@
-import { Project } from '../../projects/project.entity';
-export interface ListLogsOption {
-  project: Project;
-  branch: string;
-}

src/repos/repos.constants.ts

@@ -1 +1 @@
-export const LIST_LOGS_TASK = 'LIST_LOGS_TASK';
+export const WORKSPACE_ACTION = 'workspace-action';

src/repos/repos.module.ts

@@ -4,11 +4,17 @@ import { Project } from '../projects/project.entity';
 import { ReposResolver } from './repos.resolver';
 import { ReposService } from './repos.service';
 import { ConfigModule } from '@nestjs/config';
-import { ProjectsModule } from '../projects/projects.module';
+import { BullModule } from '@nestjs/bull';
+import { WORKSPACE_ACTION } from './repos.constants';
 
 @Module({
-  imports: [TypeOrmModule.forFeature([Project]), ConfigModule, ProjectsModule],
+  imports: [
+    TypeOrmModule.forFeature([Project]),
+    ConfigModule,
+    BullModule.registerQueue({
+      name: WORKSPACE_ACTION,
+    }),
+  ],
   providers: [ReposResolver, ReposService],
-  exports: [ReposService],
 })
 export class ReposModule {}

src/repos/repos.resolver.spec.ts

@@ -1,7 +1,6 @@
 import { Test, TestingModule } from '@nestjs/testing';
 import { ReposResolver } from './repos.resolver';
 import { ReposService } from './repos.service';
-import { ProjectsService } from '../projects/projects.service';
 
 describe('ReposResolver', () => {
   let resolver: ReposResolver;
@@ -14,10 +13,6 @@ describe('ReposResolver', () => {
           provide: ReposService,
           useValue: {},
         },
-        {
-          provide: ProjectsService,
-          useValue: {},
-        },
       ],
     }).compile();
 

src/repos/repos.resolver.ts

@@ -13,7 +13,7 @@ export class ReposResolver {
     return await this.service.listLogs(dto);
   }
   @Query(() => BranchList)
-  async listBranches(
+  async ListBranchesArgs(
     @Args('listBranchesArgs') dto: ListBranchesArgs,
   ): Promise<BranchList> {
     return await this.service.listBranches(dto).then((data) => {

src/repos/repos.service.spec.ts

@@ -1,14 +1,12 @@
-import { Pipeline } from './../pipelines/pipeline.entity';
 import { Test, TestingModule } from '@nestjs/testing';
 import { getRepositoryToken } from '@nestjs/typeorm';
 import { Project } from '../projects/project.entity';
 import { ReposService } from './repos.service';
 import { ConfigModule } from '@nestjs/config';
-import { rm } from 'fs/promises';
-import configuration from '../commons/config/configuration';
-import { PipelineTask } from '../pipeline-tasks/pipeline-task.entity';
+import { readFile, rm } from 'fs/promises';
 import { join } from 'path';
-import { readFile } from 'fs/promises';
+import configuration from '../commons/config/configuration';
+import { NotFoundException } from '@nestjs/common';
 
 const getTest1Project = () =>
   ({
@@ -73,67 +71,61 @@ describe('ReposService', () => {
       expect(result).toBeDefined();
     }, 10_000);
   });
-
-  describe.skip('checkout', () => {
-    let task: PipelineTask;
-    let workspaceRoot: string;
-    beforeEach(() => {
-      const project = new Project();
-      const pipeline = new Pipeline();
-      task = new PipelineTask();
-      pipeline.project = project;
-      task.pipeline = pipeline;
-      project.id = 'pid';
-      project.name = 'pname';
-      pipeline.id = 'lid';
-      pipeline.name = 'pipeline';
-      task.id = 'tid';
-      task.commit = '123123hash';
-      workspaceRoot = service.getWorkspaceRootByTask(task);
+  describe('checkoutBranch', () => {
+    it('should be checkout', async () => {
+      await service.checkoutBranch(getTest1Project(), 'master');
+      const filePath = join(
+        service.getWorkspaceRoot(getTest1Project()),
+        'README.md',
+      );
+      const text = await readFile(filePath, { encoding: 'utf-8' });
+      expect(text).toMatch(/Commit 1/gi);
+    }, 30_000);
+    it('multiplexing workspace', async () => {
+      await service.checkoutBranch(getTest1Project(), 'master');
+      await service.checkoutBranch(getTest1Project(), 'branch-a');
+      await service.checkoutBranch(getTest1Project(), 'branch-b');
+      const filePath = join(
+        service.getWorkspaceRoot(getTest1Project()),
+        'branch-b.md',
+      );
+      const text = await readFile(filePath, { encoding: 'utf-8' });
+      expect(text).toMatch(/Commit branch b/gi);
+    }, 30_000);
+    it('nonexistent branch', async () => {
+      return expect(
+        service.checkoutBranch(getTest1Project(), 'nonexistent'),
+      ).rejects.toBeInstanceOf(NotFoundException);
+    }, 30_000);
+    it('checkout the specified version', async () => {
+      await service.checkoutBranch(getTest1Project(), 'master');
+      const filePath = join(
+        service.getWorkspaceRoot(getTest1Project()),
+        'README.md',
+      );
+      const text = await readFile(filePath, { encoding: 'utf-8' });
+      expect(text).toMatch(/Commit 1/gi);
+    }, 30_000);
   });
 
+  describe('checkoutCommit', () => {
     it('should be checkout', async () => {
-      task.commit = '498c782685';
-      await service.checkout(task, workspaceRoot);
-      const filePath = join(workspaceRoot, 'README.md');
+      await service.checkoutCommit(getTest1Project(), '498c782685');
+      const filePath = join(
+        service.getWorkspaceRoot(getTest1Project()),
+        'README.md',
+      );
       const text = await readFile(filePath, { encoding: 'utf-8' });
       expect(text).toMatch(/Commit 1/gi);
     }, 20_000);
     it('should be checkout right commit', async () => {
-      task.commit = '7f7123fe5b';
-      await service.checkout(task, workspaceRoot);
-      const filePath = join(workspaceRoot, 'README.md');
+      await service.checkoutCommit(getTest1Project(), '7f7123fe5b');
+      const filePath = join(
+        service.getWorkspaceRoot(getTest1Project()),
+        'README.md',
+      );
       const text = await readFile(filePath, { encoding: 'utf-8' });
       expect(text).toMatch(/(?!Commit 1)/gi);
     }, 20_000);
-    it('should be checkout right commit (复用)', async () => {
-      task.commit = '498c782685';
-      await service.checkout(task, workspaceRoot);
-      task.commit = '7f7123fe5b';
-      await service.checkout(task, workspaceRoot);
-      const filePath = join(workspaceRoot, 'README.md');
-      const text = await readFile(filePath, { encoding: 'utf-8' });
-      expect(text).toMatch(/(?!Commit 1)/gi);
-    }, 30_000);
-  });
-
-  describe('getWorkspaceRootByTask', () => {
-    it('should be return right path', () => {
-      const project = new Project();
-      const pipeline = new Pipeline();
-      const task = new PipelineTask();
-      pipeline.project = project;
-      task.pipeline = pipeline;
-      project.id = 'pid';
-      project.name = 'pname';
-      pipeline.id = 'lid';
-      pipeline.name = 'pipeline/\\-名称';
-      task.id = 'tid';
-      task.commit = '123123hash';
-
-      expect(service.getWorkspaceRootByTask(task)).toMatch(
-        /\/pname\/pipeline%2F%5C-%E5%90%8D%E7%A7%B0-123123hash$/,
-      );
-    });
   });
 });

src/repos/repos.service.ts

@@ -1,5 +1,3 @@
-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';
 import { F_OK } from 'constants';
@@ -13,7 +11,6 @@ import { ListLogsArgs } from './dtos/list-logs.args';
 import { ConfigService } from '@nestjs/config';
 
 const DEFAULT_REMOTE_NAME = 'origin';
-const INFO_PATH = '@info';
 @Injectable()
 export class ReposService {
   constructor(
@@ -25,25 +22,29 @@ export class ReposService {
   getWorkspaceRoot(project: Project): string {
     return join(
       this.configService.get<string>('workspaces.root'),
-      encodeURIComponent(project.name),
-      INFO_PATH,
+      project.name,
     );
   }
 
-  async getGit(project: Project, workspaceRoot?: string) {
-    if (!workspaceRoot) {
-      workspaceRoot = this.getWorkspaceRoot(project);
+  async lockWorkspace(workspaceRoot: string) {
+    // TODO:  获取锁，失败抛错。
   }
 
-    await access(workspaceRoot, F_OK).catch(async () => {
-      await mkdir(workspaceRoot, { recursive: true });
+  async getGit(project: Project) {
+    const workspaceRoot = this.getWorkspaceRoot(project);
+    await this.lockWorkspace(workspaceRoot);
+
+    const firstInit = await access(workspaceRoot, F_OK)
+      .then(() => false)
+      .catch(async () => {
+        await mkdir(workspaceRoot);
+        return true;
       });
     const git = gitP(workspaceRoot);
-    if (!(await git.checkIsRepo())) {
+    if (firstInit) {
       await git.init();
       await git.addRemote(DEFAULT_REMOTE_NAME, project.sshUrl);
     }
-    await git.fetch();
     return git;
   }
 
@@ -52,9 +53,11 @@ export class ReposService {
       id: dto.projectId,
     });
     const git = await this.getGit(project);
-    return await git.log(
-      dto.branch ? ['--branches', dto.branch, '--'] : ['--all'],
-    );
+    await git.fetch();
+    return await git.log({
+      '--branches': dto.branch ?? '',
+      '--remotes': DEFAULT_REMOTE_NAME,
+    });
   }
 
   async listBranches(dto: ListBranchesArgs) {
@@ -65,8 +68,26 @@ export class ReposService {
     return git.branch();
   }
 
-  async checkout(task: PipelineTask, workspaceRoot: string) {
-    const git = await this.getGit(task.pipeline.project, workspaceRoot);
+  async checkoutBranch(project: Project, branch: string) {
+    const git = await this.getGit(project);
+    try {
+      await git.fetch(DEFAULT_REMOTE_NAME, branch);
+    } catch (err) {
+      if (err.message.includes("couldn't find remote ref nonexistent")) {
+        throw new NotFoundException(err.message);
+      }
+      throw err;
+    }
+    await git.checkout([
+      '-B',
+      branch,
+      '--track',
+      `${DEFAULT_REMOTE_NAME}/${branch}`,
+    ]);
+  }
+
+  async checkoutCommit(project: Project, commitNumber: string) {
+    const git = await this.getGit(project);
     try {
       await git.fetch(DEFAULT_REMOTE_NAME);
     } catch (err) {
@@ -75,20 +96,6 @@ export class ReposService {
       }
       throw err;
     }
-    await git.checkout([task.commit]);
-  }
-
-  /**
-   * get workspace root absolute path
-   *
-   * ! example: `/var/tmp/fennec-workspaces-root/project/pipeline_name-commit_hash`
-   * @param task {PipelineTask} task (with pipeline and project info)
-   */
-  getWorkspaceRootByTask(task: PipelineTask) {
-    return join(
-      this.configService.get<string>('workspaces.root'),
-      encodeURIComponent(task.pipeline.project.name),
-      encodeURIComponent(`${task.pipeline.name}-${task.commit}`),
-    );
+    await git.checkout([commitNumber]);
   }
 }

src/repos/workspace-action.consumer.ts

@@ -0,0 +1,8 @@
+import { Process, Processor } from '@nestjs/bull';
+import { WORKSPACE_ACTION } from './repos.constants';
+
+@Processor(WORKSPACE_ACTION)
+export class WorkspaceActionConsumer {
+  @Process()
+  async dispatch() {}
+}

src/repos/workspace-action.model.ts

@@ -0,0 +1,5 @@
+import { WorkspaceActions } from './workspace-actions.enum';
+
+export class WorkspaceAction {
+  action: WorkspaceActions;
+}

src/repos/workspace-actions.enum.ts

@@ -0,0 +1,6 @@
+export enum WorkspaceActions {
+  checkoutBranch = 'checkoutBranch',
+  checkoutCommit = 'checkoutCommit',
+  listLogs = 'listLogs',
+  listBranches = 'listBranches',
+}

test/data/bad-work.js

@@ -1,8 +0,0 @@
-for (let i = 1; i <= 5; i++) {
-  console.log(i * 10);
-}
-console.error('Error Message');
-console.error('Error Message 2');
-console.log('Bye-bye');
-
-process.exit(1);

test/data/one-second-work.js

@@ -1,7 +0,0 @@
-let timer;
-let count = 0;
-setTimeout(() => clearInterval(timer), 1_000);
-
-timer = setInterval(() => {
-  console.log(++count * 10);
-}, 95);