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

@@ -0,0 +1,6 @@
+import { WorkUnit } from './work-unit.model';
+
+export class WorkUnitMetadata {
+  version = 1;
+  units: WorkUnit[];
+}

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

@@ -0,0 +1,6 @@
+import { PipelineUnits as PipelineUnitTypes } from '../enums/pipeline-units.enum';
+
+export class WorkUnit {
+  type: PipelineUnitTypes;
+  scripts: string[];
+}

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

@@ -1,5 +1,6 @@
 import { Test, TestingModule } from '@nestjs/testing';
 import { Job } from 'bull';
+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';
@@ -25,6 +26,12 @@ describe('PipelineTaskConsumer', () => {
             doNextTask: () => undefined,
           },
         },
+        {
+          provide: ReposService,
+          useValue: {
+            getWorkspaceRootByTask: () => 'workspace-root',
+          },
+        },
         PipelineTaskConsumer,
       ],
     }).compile();
@@ -45,4 +52,8 @@ describe('PipelineTaskConsumer', () => {
       expect(doNextTask).toHaveBeenCalledTimes(1);
     });
   });
+
+  describe('doTask', () => {
+    it('should do all task', () => {});
+  });
 });

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

@@ -1,13 +1,57 @@
+import { ReposService } from './../repos/repos.service';
 import { OnQueueCompleted, Process, Processor } from '@nestjs/bull';
 import { Job } from 'bull';
+import { spawn } from 'child_process';
 import { PipelineTask } from './pipeline-task.entity';
 import { PIPELINE_TASK_QUEUE } from './pipeline-tasks.constants';
 import { PipelineTasksService } from './pipeline-tasks.service';
+import { ApplicationException } from '../commons/exceptions/application.exception';
+import { PipelineUnits } from './enums/pipeline-units.enum';
 @Processor(PIPELINE_TASK_QUEUE)
 export class PipelineTaskConsumer {
-  constructor(private readonly service: PipelineTasksService) {}
+  constructor(
+    private readonly service: PipelineTasksService,
+    private readonly reposService: ReposService,
+  ) {}
   @Process()
-  async doTask() {}
+  async doTask({ data: task }: Job<PipelineTask>) {
+    const workspaceRoot = this.reposService.getWorkspaceRootByTask(task);
+
+    const units = task.units.map(
+      (type) =>
+        task.pipeline.workUnitMetadata.units.find(
+          (unit) => unit.type === type,
+        ) ?? { type: type, scripts: [] },
+    );
+
+    for (const unit of units) {
+      // 检出代码时，不执行其他脚本。
+      if (unit.type === PipelineUnits.checkout) {
+        await this.reposService.checkout(task, workspaceRoot);
+        continue;
+      }
+      for (const script of unit.scripts) {
+        await this.runScript(task, script, workspaceRoot);
+      }
+    }
+  }
+
+  async runScript(task: PipelineTask, script: string, workspaceRoot: string) {
+    return new Promise((resolve, reject) => {
+      const errorMessages: string[] = [];
+      const sub = spawn(script, {
+        shell: true,
+        cwd: workspaceRoot,
+      });
+      sub.stderr.on('data', (data) => errorMessages.push(data));
+      sub.addListener('close', (code) => {
+        if (code === 0) {
+          return resolve(code);
+        }
+        return reject(new ApplicationException(errorMessages.join('\n')));
+      });
+    });
+  }
 
   @OnQueueCompleted()
   onCompleted(job: Job<PipelineTask>) {

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

@@ -1,3 +1,4 @@
+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';
@@ -7,7 +8,7 @@ import { PipelineUnits } from './enums/pipeline-units.enum';
 
 @ObjectType()
 @Entity()
-export class PipelineTask {
+export class PipelineTask extends AppBaseEntity {
   @ManyToOne(() => Pipeline)
   pipeline: Pipeline;
   @Column()

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

@@ -21,7 +21,10 @@ describe('PipelineTasksService', () => {
       id: 'test',
       name: '测试流水线',
       branch: 'master',
-      workUnitMetadata: [],
+      workUnitMetadata: {},
+      project: {
+        id: 'test-project',
+      },
     } as Pipeline);
   let redisClient;
   let taskQueue: Queue;

src/pipelines/pipeline.entity.ts

@@ -2,6 +2,7 @@ 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()
@@ -18,5 +19,5 @@ export class Pipeline extends AppBaseEntity {
   name: string;
 
   @Column({ type: 'jsonb' })
-  workUnitMetadata: any;
+  workUnitMetadata: WorkUnitMetadata;
 }

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 = () =>
   ({
@@ -59,73 +61,79 @@ describe('ReposService', () => {
   it('getWorkspaceRoot', () => {
     expect(service.getWorkspaceRoot(getTest1Project())).toBeDefined();
   });
-  describe.skip('listLogs', () => {
+  describe('listLogs', () => {
     it('should be return logs', async () => {
       const result = await service.listLogs({ projectId: '1' });
       expect(result).toBeDefined();
     }, 20_000);
   });
-  describe.skip('listBranch', () => {
+  describe('listBranch', () => {
     it('should be return branches', async () => {
       const result = await service.listBranches({ projectId: '1' });
       expect(result).toBeDefined();
     }, 10_000);
   });
-  describe.skip('checkoutBranch', () => {
-    it('should be checkout', async () => {
-      await service.checkoutBranch(getTest1Project(), 'master');
-      const filePath = join(
-        service.getWorkspaceRoot(getTest1Project(), 'master'),
-        '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'),
-        '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(), 'master'),
-        'README.md',
-      );
-      const text = await readFile(filePath, { encoding: 'utf-8' });
-      expect(text).toMatch(/Commit 1/gi);
-    }, 30_000);
-  });
 
-  describe.skip('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(), '498c782685'),
-        '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(), '7f7123fe5b'),
-        '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(
@@ -19,33 +22,28 @@ export class ReposService {
     private readonly configService: ConfigService,
   ) {}
 
-  getWorkspaceRoot(project: Project, subDir = ''): string {
+  getWorkspaceRoot(project: Project): string {
     return join(
       this.configService.get<string>('workspaces.root'),
-      project.name,
-      encodeURIComponent(subDir),
+      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, subDir = 'default') {
-    const workspaceRoot = this.getWorkspaceRoot(project, subDir);
-    await this.lockWorkspace(workspaceRoot);
-
-    const firstInit = await access(workspaceRoot, F_OK)
-      .then(() => false)
-      .catch(async () => {
-        await mkdir(workspaceRoot, { recursive: true });
-        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;
   }
 
@@ -54,7 +52,6 @@ export class ReposService {
       id: dto.projectId,
     });
     const git = await this.getGit(project);
-    await git.fetch();
     return await git.log(
       dto.branch ? ['--branches', dto.branch, '--'] : ['--all'],
     );
@@ -68,26 +65,8 @@ export class ReposService {
     return git.branch();
   }
 
-  async checkoutBranch(project: Project, branch: string) {
-    const git = await this.getGit(project, branch);
-    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, commitNumber);
+  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}`),
+    );
   }
 }