feat: built in pm2

package.json

@@ -38,7 +38,7 @@
     "class-validator": "^0.13.1",
     "debug": "^4.3.1",
     "graphql": "^15.5.0",
-    "graphql-tools": "^7.0.2",
+    "graphql-tools": "^8.1.0",
     "ioredis": "^4.25.0",
     "jose": "^3.14.0",
     "js-yaml": "^4.0.0",
@@ -48,6 +48,7 @@
     "observable-to-async-generator": "^1.0.1-rc",
     "pg": "^8.5.1",
     "pino-pretty": "^4.7.1",
+    "pm2": "^5.1.0",
     "ramda": "^0.27.1",
     "reflect-metadata": "^0.1.13",
     "rimraf": "^3.0.2",
@@ -56,7 +57,7 @@
     "typeorm": "^0.2.30"
   },
   "devDependencies": {
-    "@nestjs/cli": "^7.5.7",
+    "@nestjs/cli": "^7.6.0",
     "@nestjs/schematics": "^7.1.3",
     "@nestjs/testing": "^7.5.1",
     "@types/body-parser": "^1.19.0",

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

@@ -42,9 +42,9 @@ describe('PipelineTaskRunner', () => {
       setTimeout(() => {
         logger.handleEvent(event);
       });
-      expect(message$.pipe(take(1), timeout(100)).toPromise()).rejects.toMatch(
+      await expect(
-        'timeout',
+        message$.pipe(take(1), timeout(100)).toPromise(),
-      );
+      ).rejects.toThrow(/timeout/i);
     });
     it('multiple subscribers', async () => {
       const event = new PipelineTaskEvent();
@@ -53,13 +53,16 @@ describe('PipelineTaskRunner', () => {
       const message2$ = logger.getMessage$('test');
       setTimeout(() => {
         logger.handleEvent(event);
+        logger.handleEvent(event);
       });
-      expect(message$.pipe(take(1), timeout(100)).toPromise()).resolves.toEqual(
+      await Promise.all([
-        event,
+        expect(
-      );
+          message$.pipe(take(1), timeout(100)).toPromise(),
-      expect(
+        ).resolves.toEqual(event),
-        message2$.pipe(take(1), timeout(100)).toPromise(),
+        expect(
-      ).resolves.toEqual(event);
+          message2$.pipe(take(1), timeout(100)).toPromise(),
+        ).resolves.toEqual(event),
+      ]);
     });
   });
 

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

@@ -13,7 +13,8 @@ import {
 @Injectable()
 export class PipelineTaskLogger implements OnModuleDestroy {
   private readonly messageSubject = new Subject<PipelineTaskEvent>();
-  private readonly message$: Observable<PipelineTaskEvent> = this.messageSubject.pipe();
+  private readonly message$: Observable<PipelineTaskEvent> =
+    this.messageSubject.pipe();
 
   @RabbitSubscribe({
     exchange: EXCHANGE_PIPELINE_TASK_FANOUT,

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

@@ -1,3 +1,4 @@
+import { DeployByPm2Service } from './runners/deploy-by-pm2/deploy-by-pm2.service';
 import { Test, TestingModule } from '@nestjs/testing';
 import { ReposService } from '../repos/repos.service';
 import { PipelineUnits } from './enums/pipeline-units.enum';
@@ -11,7 +12,7 @@ import { WorkUnitMetadata } from './models/work-unit-metadata.model';
 import { AmqpConnection } from '@golevelup/nestjs-rabbitmq';
 describe('PipelineTaskRunner', () => {
   let runner: PipelineTaskRunner;
-  let reposService: ReposService;
+  let deployByPM2Service: DeployByPm2Service;
 
   beforeEach(async () => {
     const module: TestingModule = await Test.createTestingModule({
@@ -36,11 +37,18 @@ describe('PipelineTaskRunner', () => {
           provide: AmqpConnection,
           useValue: {},
         },
+        {
+          provide: DeployByPm2Service,
+          useValue: {
+            deploy: () => Promise.resolve(),
+          },
+        },
       ],
     }).compile();
 
-    reposService = module.get(ReposService);
+    module.get(ReposService);
     runner = module.get(PipelineTaskRunner);
+    deployByPM2Service = module.get(DeployByPm2Service);
   });
 
   it('should be defined', () => {
@@ -65,7 +73,7 @@ describe('PipelineTaskRunner', () => {
     beforeEach(() => {
       emitEvent = jest
         .spyOn(runner, 'emitEvent')
-        .mockImplementation((..._) => Promise.resolve());
+        .mockImplementation(() => Promise.resolve());
     });
 
     describe('doTask', () => {
@@ -75,10 +83,10 @@ describe('PipelineTaskRunner', () => {
       beforeEach(() => {
         checkout = jest
           .spyOn(runner, 'checkout')
-          .mockImplementation((..._) => Promise.resolve('/null'));
+          .mockImplementation(() => Promise.resolve('/null'));
         doTaskUnit = jest
           .spyOn(runner, 'doTaskUnit')
-          .mockImplementation((..._) => Promise.resolve());
+          .mockImplementation(() => Promise.resolve());
       });
 
       it('only checkout', async () => {
@@ -171,9 +179,7 @@ describe('PipelineTaskRunner', () => {
 
         doTaskUnit = jest
           .spyOn(runner, 'doTaskUnit')
-          .mockImplementation((..._) =>
+          .mockImplementation(() => Promise.reject(new Error('test error')));
-            Promise.reject(new Error('test error')),
-          );
         await runner.doTask(task);
 
         expect(checkout).toBeCalledTimes(1);
@@ -189,7 +195,7 @@ describe('PipelineTaskRunner', () => {
       it('success', async () => {
         const runScript = jest
           .spyOn(runner, 'runScript')
-          .mockImplementation((..._) => Promise.resolve());
+          .mockImplementation(() => Promise.resolve());
         const task = new PipelineTask();
 
         const unit = PipelineUnits.test;
@@ -210,9 +216,7 @@ describe('PipelineTaskRunner', () => {
       it('failed', async () => {
         const runScript = jest
           .spyOn(runner, 'runScript')
-          .mockImplementation((..._) =>
+          .mockImplementation(() => Promise.reject(new Error('test error')));
-            Promise.reject(new Error('test error')),
-          );
         const task = new PipelineTask();
 
         const unit = PipelineUnits.test;
@@ -230,7 +234,7 @@ describe('PipelineTaskRunner', () => {
 
     describe('runScript', () => {
       it('normal', async () => {
-        const spawn = jest.fn((..._: any[]) => ({
+        const spawn = jest.fn<any, any>(() => ({
           stdout: {
             on: () => undefined,
           },
@@ -256,7 +260,7 @@ describe('PipelineTaskRunner', () => {
         });
       });
       it('failed', async () => {
-        const spawn = jest.fn((..._: any[]) => ({
+        const spawn = jest.fn(() => ({
           stdout: {
             on: () => undefined,
           },
@@ -291,7 +295,7 @@ describe('PipelineTaskRunner', () => {
             }, 10);
           }, 10);
         });
-        const spawn = jest.fn((..._: any[]) => ({
+        const spawn = jest.fn(() => ({
           stdout: {
             on,
           },
@@ -304,7 +308,7 @@ describe('PipelineTaskRunner', () => {
         }));
 
         let emitSuccessCount = 0;
-        jest.spyOn(runner, 'emitEvent').mockImplementation((..._: any[]) => {
+        jest.spyOn(runner, 'emitEvent').mockImplementation(() => {
           return new Promise((resolve) => {
             setTimeout(() => {
               emitSuccessCount++;
@@ -323,4 +327,25 @@ describe('PipelineTaskRunner', () => {
       });
     });
   });
+
+  describe('tryRunDeployScript', () => {
+    it('should be call deploy with right args', async () => {
+      const deploy = jest.spyOn(deployByPM2Service, 'deploy');
+      await expect(
+        runner['tryRunDeployScript'](
+          '/test/dir',
+          '@@DEPLOY ecosystem.config.js',
+        ),
+      ).resolves.toBe(true);
+      expect(deploy.mock.calls[0][0]).toEqual('/test/dir/ecosystem.config.js');
+    });
+    it('should return false', async () => {
+      await expect(
+        runner['tryRunDeployScript'](
+          '/test/dir',
+          'pm2 start ecosystem.config.js',
+        ),
+      ).resolves.toBe(false);
+    });
+  });
 });

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

@@ -1,3 +1,4 @@
+import { DeployByPm2Service } from './runners/deploy-by-pm2/deploy-by-pm2.service';
 import { ReposService } from '../repos/repos.service';
 import { spawn, ChildProcessWithoutNullStreams } from 'child_process';
 import { PipelineTask } from './pipeline-task.entity';
@@ -29,6 +30,7 @@ import {
 } from '../commons/utils/rabbit-mq';
 import { rm } from 'fs/promises';
 import { rename } from 'fs/promises';
+import { join } from 'path';
 
 type Spawn = typeof spawn;
 
@@ -43,6 +45,7 @@ export class PipelineTaskRunner {
     @Inject('spawn')
     private readonly spawn: Spawn,
     private readonly amqpConnection: AmqpConnection,
+    private readonly deployByPM2Service: DeployByPm2Service,
   ) {}
   @RabbitSubscribe({
     exchange: 'new-pipeline-task',
@@ -274,6 +277,9 @@ export class PipelineTaskRunner {
     unit: PipelineUnits,
   ): Promise<void> {
     await this.emitEvent(task, unit, TaskStatuses.working, script, 'stdin');
+    if (await this.tryRunDeployScript(workspaceRoot, script)) {
+      return;
+    }
     return new Promise((resolve, reject) => {
       const sub = this.spawn(script, {
         shell: true,
@@ -317,4 +323,17 @@ export class PipelineTaskRunner {
       });
     });
   }
+
+  private async tryRunDeployScript(workspaceRoot: string, script: string) {
+    const match = /^@@DEPLOY\s+(\S*)/.exec(script);
+    if (match) {
+      await this.deployByPM2Service.deploy(
+        join(workspaceRoot, match[1]),
+        workspaceRoot,
+      );
+      return true;
+    } else {
+      return false;
+    }
+  }
 }

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

@@ -7,3 +7,5 @@ export const ROUTE_PIPELINE_TASK_DONE = 'pipeline-task-done';
 export const QUEUE_PIPELINE_TASK_DONE = 'pipeline-task-done';
 export const ROUTE_PIPELINE_TASK_KILL = 'pipeline-task-kill';
 export const QUEUE_PIPELINE_TASK_KILL = 'pipeline-task-kill';
+
+export const PM2_TOKEN = Symbol('pm2-token');

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

@@ -17,6 +17,7 @@ import {
 import { PipelineTaskLogger } from './pipeline-task.logger';
 import { PipelineTaskFlushService } from './pipeline-task-flush.service';
 import { CommonsModule } from '../commons/commons.module';
+import { DeployByPm2Service } from './runners/deploy-by-pm2/deploy-by-pm2.service';
 
 @Module({
   imports: [
@@ -68,6 +69,7 @@ import { CommonsModule } from '../commons/commons.module';
       useValue: spawn,
     },
     PipelineTaskFlushService,
+    DeployByPm2Service,
   ],
   exports: [PipelineTasksService],
 })

src/pipeline-tasks/runners/deploy-by-pm2/deploy-by-pm2.service.spec.ts

@@ -0,0 +1,117 @@
+import { Test, TestingModule } from '@nestjs/testing';
+import { getLoggerToken, PinoLogger } from 'nestjs-pino';
+import { join } from 'path';
+import { DeployByPm2Service } from './deploy-by-pm2.service';
+
+describe('DeployByPm2Service', () => {
+  let service: DeployByPm2Service;
+
+  beforeEach(async () => {
+    const module: TestingModule = await Test.createTestingModule({
+      providers: [
+        DeployByPm2Service,
+        {
+          provide: getLoggerToken(DeployByPm2Service.name),
+          useValue: new PinoLogger({
+            pinoHttp: {
+              level: 'silent',
+            },
+          }),
+        },
+      ],
+    }).compile();
+
+    service = module.get<DeployByPm2Service>(DeployByPm2Service);
+  });
+
+  it('should be defined', () => {
+    expect(service).toBeDefined();
+  });
+
+  describe('getAppsSn', () => {
+    it('should return right value', () => {
+      expect(
+        service['getAppsSn']([
+          { name: 'app' },
+          { name: 'app#4' },
+          { name: 'app#1' },
+        ]),
+      ).toEqual(4 + 1);
+    });
+    it('should return 1 when no match', () => {
+      expect(
+        service['getAppsSn']([
+          { name: 'bar' },
+          { name: 'foo#4' },
+          { name: 'foo#1' },
+        ]),
+      ).toEqual(4 + 1);
+    });
+  });
+
+  describe('filterOldApps', () => {
+    it('should return right value', () => {
+      expect(
+        service['filterOldApps'](
+          [{ name: 'app' }],
+          [
+            { name: 'app' },
+            { name: 'app#4' },
+            { name: 'foo#2' },
+            { name: 'bar' },
+          ],
+        ),
+      ).toEqual([{ name: 'app' }, { name: 'app#4' }]);
+    });
+    it('should return [] when no match', () => {
+      expect(
+        service['filterOldApps'](
+          [{ name: 'app' }],
+          [{ name: 'foo#2' }, { name: 'bar' }],
+        ),
+      ).toEqual([]);
+    });
+  });
+
+  describe('replaceAppName', () => {
+    it('should be replaced with right value', () => {
+      const getAppsSn = jest
+        .spyOn(service, 'getAppsSn' as any)
+        .mockImplementation(() => 1);
+      const options = [{ name: 'app' }, { name: 'foo' }];
+      service['replaceAppName'](options, []);
+      expect(options).toEqual([{ name: 'app#1' }, { name: 'foo#1' }]);
+      expect(getAppsSn).toBeCalledTimes(1);
+      expect(getAppsSn.mock.calls[0][0]).toEqual([]);
+    });
+  });
+
+  describe('deploy', () => {
+    it('should be success', async () => {
+      const filterOldApps = jest
+        .spyOn(service, 'filterOldApps' as any)
+        .mockImplementation(() => [{ name: 'app#1' }, { name: 'app#2' }]);
+      const replaceAppName = jest
+        .spyOn(service, 'replaceAppName' as any)
+        .mockImplementation((options) => (options[0].name = 'app#2'));
+      const stopApps = jest
+        .spyOn(service, 'stopApps' as any)
+        .mockImplementation(() => Promise.resolve());
+      await expect(
+        service['deploy'](
+          join(
+            __dirname,
+            '../../../../test/__mocks__/deploy-service/ecosystem.config.js',
+          ),
+          join(__dirname, '../../../../test/__mocks__/deploy-service'),
+        ),
+      ).resolves.toBeFalsy();
+      expect(filterOldApps).toBeCalledTimes(1);
+      expect(replaceAppName).toBeCalledTimes(1);
+      expect(stopApps).toBeCalledTimes(1);
+      stopApps.mockReset();
+
+      await service['stopApps']([{ name: 'app#2' }]);
+    }, 10_000);
+  });
+});

src/pipeline-tasks/runners/deploy-by-pm2/deploy-by-pm2.service.ts

@@ -0,0 +1,98 @@
+import { InjectPinoLogger, PinoLogger } from 'nestjs-pino';
+import { Injectable } from '@nestjs/common';
+import { promisify } from 'util';
+import * as pm2 from 'pm2';
+import { Proc, ProcessDescription, StartOptions } from 'pm2';
+import { clone, last } from 'ramda';
+
+@Injectable()
+export class DeployByPm2Service {
+  constructor(
+    @InjectPinoLogger(DeployByPm2Service.name)
+    private readonly logger: PinoLogger,
+  ) {}
+  async deploy(filePath: string, workspace: string) {
+    const baseConfig: { apps: StartOptions[] } = await import(filePath);
+    const appOptionsList: StartOptions[] = clone(baseConfig.apps);
+
+    await promisify<void>(pm2.connect.bind(pm2))();
+    const allApps = await promisify(pm2.list.bind(pm2))();
+    try {
+      if (!Array.isArray(baseConfig.apps)) {
+        this.logger.error(
+          'the "apps" in the PM2 ecosystem configuration is not array',
+        );
+        throw new Error('apps is not array');
+      }
+
+      const oldApps = this.filterOldApps(appOptionsList, allApps);
+
+      this.replaceAppName(appOptionsList, oldApps);
+      for (const appOptions of appOptionsList) {
+        const proc = await promisify<StartOptions, Proc>(pm2.start.bind(pm2))({
+          ...appOptions,
+          cwd: workspace,
+        });
+        this.logger.info({ proc }, `start ${appOptions.name}`);
+      }
+      await this.stopApps(oldApps);
+    } catch (err) {
+      await this.stopApps(appOptionsList);
+      throw err;
+    } finally {
+      pm2.disconnect();
+    }
+  }
+
+  private async stopApps(apps: ProcessDescription[] | StartOptions[]) {
+    await Promise.all(
+      apps.map(async (app: ProcessDescription | StartOptions) => {
+        let procAtStop: ProcessDescription;
+        let procAtDelete: ProcessDescription;
+        try {
+          const idOrName = 'pm_id' in app ? app.pm_id : app.name;
+          procAtStop = await promisify(pm2.stop.bind(pm2))(idOrName);
+          procAtDelete = await promisify(pm2.delete.bind(pm2))(idOrName);
+          this.logger.info('stop & delete %s success', app.name);
+        } catch (error) {
+          this.logger.error(
+            { error, procAtStop, procAtDelete },
+            'stop & delete %s error',
+            app.name,
+          );
+        }
+      }),
+    );
+  }
+
+  private replaceAppName(
+    optionsList: StartOptions[],
+    oldApps: ProcessDescription[],
+  ) {
+    const appSn = this.getAppsSn(oldApps);
+
+    optionsList.forEach((options) => {
+      if (!options.name) {
+        this.logger.error('please give a name for application');
+        throw new Error('app name is not given');
+      }
+      options.name = `${options.name}#${appSn}`;
+    });
+  }
+
+  private filterOldApps(
+    optionsList: StartOptions[],
+    apps: ProcessDescription[],
+  ) {
+    return apps.filter((app) =>
+      optionsList.some((options) => app.name.split('#')[0] === options.name),
+    );
+  }
+
+  private getAppsSn(oldApps: ProcessDescription[]) {
+    const appsSn: number[] = oldApps.map(
+      (app) => +(app.name.split('#')?.[1] ?? 0),
+    );
+    return (last(appsSn.sort()) ?? 0) + 1;
+  }
+}

test/__mocks__/deploy-service/ecosystem.config.js

@@ -0,0 +1,15 @@
+module.exports = {
+  apps: [
+    {
+      name: 'app',
+      script: __dirname + '/index.js',
+      watch: false,
+      ignore_watch: ['node_modules'],
+      log_date_format: 'MM-DD HH:mm:ss.SSS Z',
+      env: {},
+      max_restarts: 5,
+      kill_timeout: 10_000,
+      wait_ready: true,
+    },
+  ],
+};

test/__mocks__/deploy-service/index.js

@@ -0,0 +1,23 @@
+import { createServer } from 'http';
+
+var app = createServer(function (req, res) {
+  res.writeHead(200);
+  setTimeout(() => {
+    res.end('hey');
+  }, 2000);
+});
+
+var listener = app.listen(0, function () {
+  console.log('Listening on port ' + listener.address().port);
+  setTimeout(() => {
+    // Here we send the ready signal to PM2
+    process.send('ready');
+  }, 5000);
+});
+
+process.on('SIGINT', function () {
+  listener.close();
+  setTimeout(() => {
+    process.exit(0);
+  }, 2000);
+});

test/__mocks__/deploy-service/package.json

@@ -0,0 +1,10 @@
+{
+  "name": "deploy-service",
+  "version": "1.0.0",
+  "description": "For Test",
+  "main": "index.js",
+  "type": "module",
+  "scripts": {},
+  "author": "Ivan Li",
+  "license": "ISC"
+}

test/__mocks__/deploy-service/test.js

@@ -0,0 +1,16 @@
+import { createServer } from 'http';
+
+var app = createServer(function (req, res) {
+  res.writeHead(200);
+  setTimeout(() => {
+    res.end('pm2');
+  }, 2000);
+});
+
+var listener = app.listen(33333, function () {
+  console.log('Listening on port ' + listener.address().port);
+  setTimeout(() => {
+    // Here we send the ready signal to PM2
+    process.send('ready');
+  }, 5000);
+});

tsconfig.json

@@ -10,7 +10,8 @@
     "lib": ["ES2021"],
     "sourceMap": true,
     "outDir": "./dist",
+    "rootDir": "./src",
     "baseUrl": "./",
-    "incremental": true
+    "incremental": true,
-  }
+  },
 }