feat(pipelines): 修改接口名。

findPipelines -> listPipelines.

.vscode/settings.json

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

config.yml.example

@@ -9,5 +9,10 @@ db:
     database: fennec
     username: fennec
     password:
+  redis:
+    host: 192.168.31.194
+    port: 6379
+    password:
+    prefix: fennec
 workspaces:
   root: '/Users/ivanli/Projects/fennec/workspaces'

docs/ci-cd.md

@@ -0,0 +1,33 @@
+# 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,14 +9087,6 @@
         }
       }
     },
-    "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",
@@ -9126,6 +9118,16 @@
         "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": {
@@ -9489,6 +9491,14 @@
             "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,7 +7,10 @@ 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';
 
 @Module({
   imports: [
@@ -39,6 +42,18 @@ import configuration from './commons/config/configuration';
     }),
     ProjectsModule,
     ReposModule,
+    PipelinesModule,
+    PipelineTasksModule,
+    RedisModule.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'),
+      }),
+      inject: [ConfigService],
+    }),
   ],
   controllers: [AppController],
   providers: [AppService, AppResolver],

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

@@ -0,0 +1,27 @@
+import { InputType } from '@nestjs/graphql';
+import { WorkUnitMetadata } from '../../pipeline-tasks/models/work-unit-metadata.model';
+import {
+  IsInstance,
+  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()
+  @IsInstance(WorkUnitMetadata)
+  workUnitMetadata: WorkUnitMetadata;
+}

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

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

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

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

src/pipelines/pipeline.entity.ts

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

@@ -0,0 +1,11 @@
+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';
+
+@Module({
+  imports: [TypeOrmModule.forFeature([Pipeline])],
+  providers: [PipelinesResolver, PipelinesService],
+})
+export class PipelinesModule {}

src/pipelines/pipelines.resolver.spec.ts

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

@@ -0,0 +1,44 @@
+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';
+
+@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);
+  }
+}

src/pipelines/pipelines.service.spec.ts

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

src/pipelines/pipelines.service.ts

@@ -0,0 +1,37 @@
+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';
+
+@Injectable()
+export class PipelinesService extends BaseDbService<Pipeline> {
+  readonly uniqueFields: Array<keyof Pipeline> = ['branch', 'projectId'];
+  constructor(
+    @InjectRepository(Pipeline)
+    readonly repository: Repository<Pipeline>,
+  ) {
+    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;
+  }
+}

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

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

src/projects/projects.service.ts

@@ -4,6 +4,7 @@ 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> {
@@ -24,7 +25,7 @@ export class ProjectsService extends BaseDbService<Project> {
     return await this.repository.save(this.repository.create(dto));
   }
 
-  async update(id: string, dto: CreateProjectInput) {
+  async update(id: string, dto: UpdateProjectInput) {
     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

@@ -0,0 +1,16 @@
+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, ObjectType } from '@nestjs/graphql';
+import { InputType } from '@nestjs/graphql';
 import { IsOptional, IsString, IsUUID } from 'class-validator';
 
 @InputType()

src/repos/repos.module.ts

@@ -4,9 +4,11 @@ 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';
 
 @Module({
-  imports: [TypeOrmModule.forFeature([Project]), ConfigModule],
+  imports: [TypeOrmModule.forFeature([Project]), ConfigModule, ProjectsModule],
   providers: [ReposResolver, ReposService],
+  exports: [ReposService],
 })
 export class ReposModule {}

src/repos/repos.resolver.spec.ts

@@ -1,6 +1,7 @@
 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;
@@ -13,6 +14,10 @@ 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 ListBranchesArgs(
+  async listBranches(
     @Args('listBranchesArgs') dto: ListBranchesArgs,
   ): Promise<BranchList> {
     return await this.service.listBranches(dto).then((data) => {

src/repos/repos.service.spec.ts

@@ -1,12 +1,14 @@
+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 { readFile, rm } from 'fs/promises';
-import { join } from 'path';
+import { rm } from 'fs/promises';
 import configuration from '../commons/config/configuration';
-import { NotFoundException } from '@nestjs/common';
+import { PipelineTask } from '../pipeline-tasks/pipeline-task.entity';
+import { join } from 'path';
+import { readFile } from 'fs/promises';
 
 const getTest1Project = () =>
   ({
@@ -71,61 +73,67 @@ describe('ReposService', () => {
       expect(result).toBeDefined();
     }, 10_000);
   });
-  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', () => {
+  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);
+    });
+
     it('should be checkout', async () => {
-      await service.checkoutCommit(getTest1Project(), '498c782685');
-      const filePath = join(
-        service.getWorkspaceRoot(getTest1Project()),
-        'README.md',
-      );
+      task.commit = '498c782685';
+      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);
     }, 20_000);
     it('should be checkout right commit', async () => {
-      await service.checkoutCommit(getTest1Project(), '7f7123fe5b');
-      const filePath = join(
-        service.getWorkspaceRoot(getTest1Project()),
-        'README.md',
-      );
+      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);
     }, 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,3 +1,5 @@
+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';
@@ -11,6 +13,7 @@ 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(
@@ -22,29 +25,25 @@ export class ReposService {
   getWorkspaceRoot(project: Project): string {
     return join(
       this.configService.get<string>('workspaces.root'),
-      project.name,
+      encodeURIComponent(project.name),
+      INFO_PATH,
     );
   }
 
-  async lockWorkspace(workspaceRoot: string) {
-    // TODO:  获取锁，失败抛错。
-  }
+  async getGit(project: Project, workspaceRoot?: string) {
+    if (!workspaceRoot) {
+      workspaceRoot = this.getWorkspaceRoot(project);
+    }
 
-  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;
-      });
+    await access(workspaceRoot, F_OK).catch(async () => {
+      await mkdir(workspaceRoot, { recursive: true });
+    });
     const git = gitP(workspaceRoot);
-    if (firstInit) {
+    if (!(await git.checkIsRepo())) {
       await git.init();
       await git.addRemote(DEFAULT_REMOTE_NAME, project.sshUrl);
     }
+    await git.fetch();
     return git;
   }
 
@@ -53,11 +52,9 @@ export class ReposService {
       id: dto.projectId,
     });
     const git = await this.getGit(project);
-    await git.fetch();
-    return await git.log({
-      '--branches': dto.branch ?? '',
-      '--remotes': DEFAULT_REMOTE_NAME,
-    });
+    return await git.log(
+      dto.branch ? ['--branches', dto.branch, '--'] : ['--all'],
+    );
   }
 
   async listBranches(dto: ListBranchesArgs) {
@@ -68,26 +65,8 @@ export class ReposService {
     return git.branch();
   }
 
-  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);
+  async checkout(task: PipelineTask, workspaceRoot: string) {
+    const git = await this.getGit(task.pipeline.project, workspaceRoot);
     try {
       await git.fetch(DEFAULT_REMOTE_NAME);
     } catch (err) {
@@ -96,6 +75,20 @@ export class ReposService {
       }
       throw err;
     }
-    await git.checkout([commitNumber]);
+    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}`),
+    );
   }
 }

test/data/bad-work.js

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

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