Compare commits

...

93 Commits

Author SHA1 Message Date
a299958fcb chore: high severity vulnerability 2021-09-17 23:14:42 +08:00
f07e18f71d fix: nest build can not output dist folder. 2021-09-17 22:58:35 +08:00
eb7adc0ef2 build: 更新部分依赖。 2021-09-16 23:58:04 +08:00
122dca689d build: 更新部分依赖。 2021-09-16 23:55:08 +08:00
fcca1508eb fix: 依赖问题导致无法通过类型检查 2021-09-16 23:17:08 +08:00
Ivan Li
34cfc71a18 Merge pull request 'feat-built-in-pm2' (#9) from feat-built-in-pm2 into master
Reviewed-on: #9
2021-09-12 19:54:49 +08:00
de2e9fa8c4 chore: 使用自建的包管理仓库 2021-09-12 10:55:39 +08:00
0f1466bf91 feat: built in pm2 2021-09-08 23:04:11 +08:00
Ivan Li
ed71d83581 chore: clean code. 2021-07-24 16:57:01 +08:00
Ivan Li
574e7ecae7 Merge pull request 'single-deploy-folder' (#8) from single-deploy-folder into master
Reviewed-on: #8
2021-07-22 20:59:55 +08:00
Ivan Li
1b469e34f9 feat: 使用单独目录部署。 2021-07-22 20:48:30 +08:00
Ivan Li
c86772a5dd bak 2021-07-20 22:30:00 +08:00
Ivan Li
aec7f65434 Merge pull request 'feat: jwt auth.' (#7) from feat-jwt-auth into master
Reviewed-on: #7
2021-07-20 20:45:16 +08:00
Ivan Li
ab4ef36bf8 feat: 支持 Subscription 鉴权。 2021-07-20 20:37:46 +08:00
Ivan Li
0a03bcd36e feat: 使用 @nestjs-lib/auth 鉴权。 2021-07-18 22:40:33 +08:00
Ivan Li
ec351d12f2 fix(pipelines): 任务脚本不得为空 2021-07-10 18:50:01 +08:00
Ivan Li
c3c73fbe65 fix(account middleware): 使用了未定义的变量。 2021-07-10 17:37:21 +08:00
Ivan
02059ee54f feat: 使用中间件 从 jwt 中获取用户信息 2021-07-09 18:20:36 +08:00
Ivan
5ed17cc04b test: 完善测试用例 2021-07-09 10:17:16 +08:00
Ivan
256878890b fix(projects): required id when create project 2021-07-06 15:57:03 +08:00
Ivan Li
9908bd229e build: add pm2 ecosystem config 2021-06-30 20:25:00 +08:00
Ivan Li
07f19101a5 Merge pull request 'feat_the_progress_of_tasks' (#5) from feat_the_progress_of_tasks into master
Reviewed-on: #5
2021-06-27 19:57:12 +08:00
Ivan Li
7d84017f9e feat:v0.1.0 2021-06-27 19:44:01 +08:00
Ivan Li
a231a02c28 feat: 修改项目后自动删除项目工作区相关目录 2021-06-27 19:35:49 +08:00
Ivan Li
7e17de0f15 feat(repos): 获取 commit logs 时,允许返回错误信息给调用方。 2021-06-27 10:33:32 +08:00
Ivan Li
9d735c582c fix(pipelines): 更新时重复判断时误判问题。 2021-06-27 10:32:39 +08:00
Ivan Li
b626eed859 fix(pipeline-tasks): lost unit logs. 2021-06-20 20:30:14 +08:00
Ivan Li
5b5a657651 feat(pipeline-tasks): 强制停止任务接口。 2021-06-20 15:41:16 +08:00
Ivan Li
a510f411a7 feat(pipeline-tasks): 在数据库中保存任务执行者所在实例。 2021-06-20 10:52:12 +08:00
Ivan Li
246623b5db feat: 任务状态改变时,自动更新数据库。 2021-06-20 10:39:53 +08:00
Ivan Li
37f8ae19be Merge branch 'feat_the_progress_of_tasks' of ssh://git.ivanli.cc:7018/fennec/fennec-be into feat_the_progress_of_tasks 2021-06-20 00:06:17 +08:00
Ivan Li
133439bb49 feat: 任务完成后更新数据库中的数据。 2021-06-20 00:06:09 +08:00
Ivan Li
646f68d298 feat: 任务完成后更新数据库中的数据。 2021-06-20 00:05:24 +08:00
Ivan
0c3310d3a5 feat(pipeline-task):通过任务完成消息更新数据库中的任务信息。 2021-06-08 14:52:10 +08:00
Ivan Li
ead32a1204 feat(pipeline-task): flush service. 2021-06-06 22:21:03 +08:00
Ivan Li
20612d4301 feat(pipeline-task): add pipelineTaskEvent api. 2021-06-06 08:40:25 +08:00
Ivan Li
7091f9df6a feat(pipeline-task): logger. 2021-06-05 19:11:39 +08:00
Ivan Li
3ee41ece67 refactor(pipeline, repo): rabbitmq 2021-05-30 22:42:20 +08:00
Ivan Li
b3a2b11db9 refactor(pipeline-tasks-runner): rabbitmq 2021-05-23 17:51:54 +08:00
Ivan Li
4041a6fd2a feat(pipelines): add commits query api. 2021-05-09 16:54:59 +08:00
Ivan Li
5a8b699e2f feat(pipelines): normalization name. 2021-05-07 20:49:35 +08:00
Ivan Li
86c8bce9ea feat(projects): normalization name. 2021-05-04 21:05:20 +08:00
Ivan Li
246c0bd8f8 fix(pipeline-tasks): 修复日志有概率丢失最后几条的问题
日志记录结束前,任务信息已入库存档。
2021-04-06 22:54:51 +08:00
Ivan Li
24a2f80e46 test: pass cases 2021-04-05 20:20:43 +08:00
Ivan Li
0e0781c4c4 feat(pipeline-tasks): 任务更新推送时机修改。 2021-04-05 16:34:12 +08:00
Ivan Li
a82f663354 feat: clean up. 2021-04-04 15:51:03 +08:00
Ivan Li
752db8a0c5 Merge pull request 'feat-pub-sub' (#3) from feat-pub-sub into master
Reviewed-on: #3
2021-04-04 13:28:58 +08:00
Ivan
46fb41f856 feat: 完善消息发布。 2021-04-04 12:20:08 +08:00
Ivan
b4307f05d6 feat: 使用 PubSub 进行消息广播。 2021-04-04 00:36:58 +08:00
Ivan
bb3efd3714 feat: pubsub base redis. 2021-04-03 19:19:02 +08:00
Ivan
092cf9c418 Merge branch 'master' of ssh://git.ivanli.cc:7018/Fennec/fennec-be 2021-03-28 20:27:28 +08:00
Ivan
039f4b6d15 test: pass case. 2021-03-28 20:27:20 +08:00
Ivan Li
22be1ffb33 Merge branch 'master' of ssh://git.ivanli.cc:7018/Fennec/fennec-be 2021-03-28 19:38:18 +08:00
Ivan Li
ef47f8049e fix(queue): 修复消息队列未使用配置的reids. 2021-03-28 19:38:09 +08:00
Ivan
032aa89b05 feat(pipelines): list logs 时提供关联的 tasks 。 2021-03-28 19:37:16 +08:00
Ivan Li
da6bc9a068 提供 gieea webhooks (#2)
chore: debug log 仅输出app的log

fix(commons): fix sanitize not return value.

feat(webhooks): add gitea webhooks api.

Co-authored-by: Ivan Li <ivanli@live.cn>
Co-authored-by: Ivan <ivanli@live.cn>
Reviewed-on: #2
Co-Authored-By: Ivan Li <ivan@noreply.%(DOMAIN)s>
Co-Committed-By: Ivan Li <ivan@noreply.%(DOMAIN)s>
2021-03-28 10:24:12 +08:00
Ivan
429de1eaed test(pipeline-tasks): pass test cases. 2021-03-27 15:52:21 +08:00
Ivan
08e5c7e7d3 refactor(pipeline-tasks): remove unused queue. 2021-03-27 15:34:08 +08:00
Ivan
713f5b2426 feat(pipeline-tasks): debug log. 2021-03-25 18:09:45 +08:00
Ivan
607a4f57de feat(pipeline-tasks): 任务失败后,也调用执行下一个任务。 2021-03-25 11:39:15 +08:00
Ivan
211a90590f test: pass the test case. 2021-03-25 10:51:16 +08:00
Ivan Li
07fc98bc86 Merge pull request 'feat-pipelines' (#1) from feat-pipelines into master
Reviewed-on: #1
2021-03-24 20:50:40 +08:00
Ivan Li
9bdd991cfb feat(pipeline-tasks): 任务更新推送。 2021-03-24 20:35:24 +08:00
Ivan Li
9078835c28 test(pipeline-tasks): 测试用例不通过的问题 2021-03-21 20:31:25 +08:00
Ivan Li
42c5e4d608 fix(pipeline-tasks): 修复未保存任务状态到数据库的问题。 2021-03-21 20:28:48 +08:00
Ivan Li
cdc28cb102 feat(pipeline-tasks): 添加 部署任务查询接口和任务日志推送。 2021-03-20 14:30:26 +08:00
Ivan Li
7923ae6d41 fix: 字段类型错误的问题 2021-03-15 13:49:37 +08:00
Ivan Li
4e7c825170 feat(pipeline-task): 流水线任务日志订阅接口。 2021-03-15 13:30:52 +08:00
Ivan Li
aa92c518f0 feat: 新增 创建CI CD任务接口。 2021-03-12 23:00:12 +08:00
Ivan Li
cba4c0464c feat(pipelines): list commit logs 使用直接订阅。
替代了先查询再订阅,确保数据不丢失。
2021-03-10 21:41:47 +08:00
Ivan Li
d02cea2115 fix(repos): 完善 commit log 订阅接口。 2021-03-09 22:50:26 +08:00
Ivan
22d3dc299c feat(repos): 添加订阅commit log 接口。 2021-03-08 11:26:12 +08:00
Ivan
ba0ba46a35 test(pipeline-tasks): 更新测试用例。 2021-03-08 10:48:46 +08:00
Ivan Li
f00f75673b feat(pipelines): 添加查询 commit logs 接口。 2021-03-07 22:53:16 +08:00
Ivan Li
bba7963949 feat(commons): 为实体统一添加软删除相关字段。 2021-03-07 22:12:15 +08:00
Ivan Li
bf4590bd4c feat(exception): 优化错误响应格式。 2021-03-07 16:20:41 +08:00
Ivan Li
e908d2981d feat(pipelines): 修改接口名。
findPipelines -> listPipelines.
2021-03-06 23:30:12 +08:00
Ivan Li
0dadc09ec5 fix: Graphql 类型报错 2021-03-06 12:24:20 +08:00
Ivan Li
38d3cb0db8 feat(redis): 添加 redis 及其配置 2021-03-06 12:23:55 +08:00
Ivan
8901c49bb3 back. 2021-03-05 18:13:10 +08:00
Ivan
7913184174 feat(pipeline-tasks): 完善 doTask 2021-03-05 17:12:06 +08:00
Ivan
f39c801fc2 feat(repos): 检出相关逻辑更改为传入任务信息而不是项目信息。 2021-03-04 17:02:07 +08:00
Ivan
33b09594f5 test(pipeline-tasks): task consumer unit test. 2021-03-04 11:43:23 +08:00
Ivan
64ec1433a6 feat(pipeline-tasks): 完善流程和测试用例。 2021-03-03 17:24:22 +08:00
Ivan
31a200206f feat(pipelines): 添加消息队列 2021-03-02 16:28:37 +08:00
Ivan
22d9bf47d3 feat(pipeline-tasks): 流水线人物。 2021-03-01 18:14:13 +08:00
Ivan
e3e698b8cb feat(pipelines): 添加流水线模块。 2021-03-01 15:22:45 +08:00
Ivan
ea4ca724e3 fix(test): repos resolver 测试用例不通过的问题。 2021-03-01 11:47:04 +08:00
Ivan Li
32102ffefd feat(repos): 添加checkout接口。 2021-02-28 21:51:52 +08:00
Ivan Li
11cf2a6c12 fix(repos): 修复提交日志接口,分支筛选无效的问题。 2021-02-28 02:23:20 +08:00
Ivan Li
2d5763ac02 fix(repos): 查询分支列表接口名称错误的问题。 2021-02-28 02:07:56 +08:00
Ivan Li
3b7c50438f fix(repos): ssh url 校验规则修改,不强制要求仅有用户名时要求追加:分隔。 2021-02-27 22:55:00 +08:00
Ivan
1d8b99fe8e feat(repos): 检出路径添加分支名称或Commit Number。 2021-02-25 18:15:12 +08:00
96 changed files with 27390 additions and 6446 deletions

3
.gitignore vendored
View File

@ -33,4 +33,5 @@ lerna-debug.log*
!.vscode/launch.json
!.vscode/extensions.json
/config.yml
/config.yml
tsconfig.build.tsbuildinfo

3
.vscode/extensions.json vendored Normal file
View File

@ -0,0 +1,3 @@
{
"recommendations": []
}

0
.vscode/launch.json vendored Normal file
View File

17
.vscode/settings.json vendored
View File

@ -1,5 +1,20 @@
{
"cSpell.words": [
"Repos"
"Mutex",
"Repos",
"amqp",
"boardcat",
"errout",
"fanout",
"gitea",
"golevelup",
"lpush",
"lrange",
"metatype",
"pmessage",
"psubscribe",
"rabbitmq",
"rpop",
"rpush"
]
}

View File

@ -9,5 +9,16 @@ db:
database: fennec
username: fennec
password:
redis:
host: 192.168.31.194
port: 6379
password:
prefix: fennec
rabbitmq:
uri: 'amqp://fennec:fennec@192.168.31.194:5672'
etcd:
hosts:
- 'http://192.168.31.194:2379'
workspaces:
root: '/Users/ivanli/Projects/fennec/workspaces'

33
docs/ci-cd.md Normal file
View File

@ -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"
]
}
}
}
```

14
ecosystem.config.js Normal file
View File

@ -0,0 +1,14 @@
module.exports = {
apps: [
{
name: 'fennec-be',
script: 'npm',
args: 'run start:prod',
watch: false,
ignore_watch: ['node_modules'],
log_date_format: 'MM-DD HH:mm:ss.SSS Z',
env: {},
max_restarts: 5,
},
],
};

30081
package-lock.json generated

File diff suppressed because it is too large Load Diff

View File

@ -1,8 +1,8 @@
{
"name": "fennec-be",
"version": "0.0.1",
"description": "",
"author": "",
"version": "0.1.1",
"description": "a ci/cd project.",
"author": "Ivan Li\b<ivanli2048@gmail.com>",
"private": true,
"license": "UNLICENSED",
"scripts": {
@ -10,8 +10,8 @@
"build": "nest build",
"format": "prettier --write \"src/**/*.ts\" \"test/**/*.ts\"",
"start": "nest start",
"start:dev": "nest start --watch",
"start:debug": "DEBUG=simple-git,simple-git:* nest start --debug --watch",
"start:dev": "DEBUG=fennec:* nest start --watch",
"start:debug": "DEBUG=fennec:* nest start --debug --watch",
"start:prod": "node dist/main",
"lint": "eslint \"{src,apps,libs,test}/**/*.ts\" --fix",
"test": "jest",
@ -21,26 +21,34 @@
"test:e2e": "jest --config ./test/jest-e2e.json"
},
"dependencies": {
"@nestjs/bull": "^0.3.1",
"@golevelup/nestjs-rabbitmq": "^1.16.1",
"@nestjs-lib/auth": "^0.2.1",
"@nestjs/common": "^7.5.1",
"@nestjs/config": "^0.6.2",
"@nestjs/core": "^7.5.1",
"@nestjs/graphql": "^7.9.8",
"@nestjs/platform-express": "^7.5.1",
"@nestjs/typeorm": "^7.1.5",
"@neuralegion/class-sanitizer": "^0.3.2",
"@types/bull": "^3.15.0",
"@types/amqplib": "^0.8.0",
"@types/ramda": "^0.27.38",
"apollo-server-express": "^2.19.2",
"bcrypt": "^5.0.0",
"bull": "^3.20.1",
"body-parser": "^1.19.0",
"class-transformer": "^0.3.2",
"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",
"nestjs-etcd": "^0.2.0",
"nestjs-pino": "^1.4.0",
"nestjs-redis": "^1.2.8",
"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",
@ -49,13 +57,17 @@
"typeorm": "^0.2.30"
},
"devDependencies": {
"@nestjs/cli": "^7.5.1",
"@nestjs/cli": "^7.6.0",
"@nestjs/schematics": "^7.1.3",
"@nestjs/testing": "^7.5.1",
"@types/body-parser": "^1.19.0",
"@types/debug": "^4.1.5",
"@types/express": "^4.17.8",
"@types/ioredis": "^4.22.2",
"@types/jest": "^26.0.15",
"@types/js-yaml": "^4.0.0",
"@types/node": "^14.14.6",
"@types/pino-pretty": "^4.7.0",
"@types/supertest": "^2.0.10",
"@typescript-eslint/eslint-plugin": "^4.6.1",
"@typescript-eslint/parser": "^4.6.1",
@ -85,6 +97,9 @@
"collectCoverageFrom": [
"**/*.(t|j)s"
],
"moduleNameMapper": {
"^jose/(.*)$": "<rootDir>/../node_modules/jose/dist/node/cjs/$1"
},
"coverageDirectory": "../coverage",
"testEnvironment": "node"
}

View File

@ -1,4 +1,5 @@
import { Module } from '@nestjs/common';
import { CommonsModule } from './commons/commons.module';
import { MiddlewareConsumer, Module, NestModule } from '@nestjs/common';
import { ConfigModule, ConfigService } from '@nestjs/config';
import { GraphQLModule } from '@nestjs/graphql';
import { TypeOrmModule } from '@nestjs/typeorm';
@ -7,13 +8,41 @@ 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 { WebhooksModule } from './webhooks/webhooks.module';
import { RawBodyMiddleware } from './commons/middleware/raw-body.middleware';
import { GiteaWebhooksController } from './webhooks/gitea-webhooks.controller';
import { ParseBodyMiddleware } from './commons/middleware/parse-body.middleware';
import { LoggerModule } from 'nestjs-pino';
import { EtcdModule } from 'nestjs-etcd';
import pinoPretty from 'pino-pretty';
import { fromPairs, map, pipe, toPairs } from 'ramda';
@Module({
imports: [
ConfigModule.forRoot({
load: [configuration],
}),
LoggerModule.forRootAsync({
imports: [ConfigModule],
useFactory: (configService: ConfigService) => {
const isDev = configService.get<'dev' | 'prod'>('env') === 'dev';
return {
pinoHttp: {
prettyPrint: isDev
? {
levelFirst: true,
}
: false,
prettifier: pinoPretty,
},
};
},
inject: [ConfigService],
}),
TypeOrmModule.forRootAsync({
imports: [ConfigModule],
useFactory: (configService: ConfigService) => ({
@ -34,13 +63,59 @@ import configuration from './commons/config/configuration';
debug: configService.get<string>('env') !== 'prod',
playground: true,
autoSchemaFile: true,
installSubscriptionHandlers: true,
context: ({ req, connection, ...args }) => {
return connection ? { req: connection.context } : { req };
},
subscriptions: {
onConnect: (connectionParams: Record<string, string>) => {
const connectionParamsWithLowerKeys = pipe(
toPairs,
map(([key, value]) => [key.toLowerCase(), value]),
fromPairs,
)(connectionParams);
return {
headers: connectionParamsWithLowerKeys,
};
},
},
}),
inject: [ConfigService],
}),
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],
}),
EtcdModule.forRootAsync({
imports: [ConfigModule],
useFactory: (configService: ConfigService) => ({
hosts: configService.get<string>('db.etcd.hosts', 'localhost:2379'),
}),
inject: [ConfigService],
}),
WebhooksModule,
CommonsModule,
],
controllers: [AppController],
providers: [AppService, AppResolver],
})
export class AppModule {}
export class AppModule implements NestModule {
public configure(consumer: MiddlewareConsumer): void {
consumer
.apply(RawBodyMiddleware)
.forRoutes(GiteaWebhooksController)
.apply(ParseBodyMiddleware)
.forRoutes('*');
}
}

View File

@ -1,8 +1,11 @@
import { Module } from '@nestjs/common';
import { PasswordConverter } from './services/password-converter';
import { RedisMutexModule } from './redis-mutex/redis-mutex.module';
import { AuthModule } from '@nestjs-lib/auth';
@Module({
imports: [RedisMutexModule, AuthModule],
providers: [PasswordConverter],
exports: [PasswordConverter],
exports: [PasswordConverter, RedisMutexModule, AuthModule],
})
export class CommonsModule {}

View File

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

View File

@ -1,11 +1,11 @@
import { pick } from 'ramda';
export class ApplicationException extends Error {
code: number;
error: Error;
constructor(
message:
| string
| { error?: Error; message?: string | object; code?: number },
message: string | { error?: Error; message?: string | any; code?: number },
) {
if (message instanceof Object) {
super();
@ -18,4 +18,8 @@ export class ApplicationException extends Error {
super((message as unknown) as any);
}
}
toJSON() {
return pick(['code', 'message'], this);
}
}

View File

@ -1,55 +1,58 @@
import {
ArgumentsHost,
Catch,
ExceptionFilter,
Catch,
ArgumentsHost,
HttpException,
HttpStatus,
} from '@nestjs/common';
import { EntityNotFoundError } from 'typeorm/error/EntityNotFoundError';
import { ApolloError } from 'apollo-server-errors';
import { PinoLogger, InjectPinoLogger } from 'nestjs-pino';
@Catch()
export class AllExceptionsFilter implements ExceptionFilter {
catch(exception: any, host: ArgumentsHost) {
const ctx = host.switchToHttp();
const response = ctx.getResponse();
const request = ctx.getRequest();
@Catch(HttpException)
export class HttpExceptionFilter implements ExceptionFilter {
constructor(
@InjectPinoLogger(HttpExceptionFilter.name)
private readonly logger: PinoLogger,
) {}
catch(exception: HttpException, host: ArgumentsHost) {
switch (host.getType<'http' | 'graphql' | string>()) {
case 'graphql': {
const errorName = exception.message;
const extensions: Record<string, any> = {};
const err = exception.getResponse();
if (typeof err === 'string') {
extensions.message = err;
} else {
Object.assign(extensions, (err as any).extension);
extensions.message = (err as any).message;
}
extensions.error = errorName;
this.logger.error(extensions);
return new ApolloError(
extensions.message,
exception.getStatus().toString(),
extensions,
);
}
case 'http': {
const ctx = host.switchToHttp();
const response = ctx.getResponse();
const request = ctx.getRequest();
const status =
exception instanceof HttpException
? exception.getStatus()
: HttpStatus.INTERNAL_SERVER_ERROR;
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 {
response.status(status).json({
message: ex,
timestamp: Date.now(),
statusCode: status,
message: exception.message,
timestamp: new Date().toISOString(),
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,
});
default:
throw exception;
}
}
}

View File

@ -0,0 +1,7 @@
import { ParseBodyMiddleware } from './parse-body.middleware';
describe('ParseBodyMiddleware', () => {
it('should be defined', () => {
expect(new ParseBodyMiddleware()).toBeDefined();
});
});

View File

@ -0,0 +1,13 @@
import { Injectable, NestMiddleware } from '@nestjs/common';
import { json, urlencoded, text } from 'body-parser';
import { Request, Response, NextFunction } from 'express';
@Injectable()
export class ParseBodyMiddleware implements NestMiddleware {
use(req: Request, res: Response, next: NextFunction) {
json()(req, res, () =>
urlencoded()(req, res, () => text()(req, res, next)),
);
// next();
}
}

View File

@ -0,0 +1,7 @@
import { RawBodyMiddleware } from './raw-body.middleware';
describe('RawBodyMiddleware', () => {
it('should be defined', () => {
expect(new RawBodyMiddleware()).toBeDefined();
});
});

View File

@ -0,0 +1,10 @@
import { Injectable, NestMiddleware } from '@nestjs/common';
import { raw } from 'body-parser';
import { Request, Response, NextFunction } from 'express';
@Injectable()
export class RawBodyMiddleware implements NestMiddleware {
use(req: Request, res: Response, next: NextFunction) {
raw({ type: '*/*' })(req, res, next);
}
}

View File

@ -0,0 +1,25 @@
import { ArgumentMetadata, Injectable, PipeTransform } from '@nestjs/common';
import { plainToClass } from 'class-transformer';
@Injectable()
export class SanitizePipe implements PipeTransform {
transform(value: any, metadata: ArgumentMetadata) {
if (
!(value instanceof Object) ||
value instanceof Buffer ||
value instanceof Array
) {
return value;
}
const constructorFunction = metadata.metatype;
if (!constructorFunction || value instanceof constructorFunction) {
return value;
}
try {
return plainToClass(constructorFunction, value);
} catch (err) {
console.error(err);
throw err;
}
}
}

View File

@ -0,0 +1,10 @@
import { Module } from '@nestjs/common';
import { RedisMutexService } from './redis-mutex.service';
import { RedisModule } from 'nestjs-redis';
@Module({
imports: [RedisModule],
providers: [RedisMutexService],
exports: [RedisMutexService],
})
export class RedisMutexModule {}

View File

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

View File

@ -0,0 +1,71 @@
import { Injectable } from '@nestjs/common';
import { RedisService } from 'nestjs-redis';
import * as uuid from 'uuid';
import { ApplicationException } from '../exceptions/application.exception';
export interface RedisMutexOption {
/**
* seconds
*/
expires?: number;
/**
* seconds
*/
timeout?: number | null;
/**
* milliseconds
*/
retryDelay?: number;
}
@Injectable()
export class RedisMutexService {
constructor(private readonly redisClient: RedisService) {}
public async lock(
key: string,
{ expires = 100, timeout = 10, retryDelay = 100 }: RedisMutexOption = {
expires: 100,
timeout: 10,
retryDelay: 100,
},
) {
const redisKey = `${'mutex-lock'}:${key}`;
const redis = this.redisClient.getClient();
const value = uuid.v4();
const timeoutAt = timeout ? Date.now() + timeout * 1000 : null;
while (
!(await redis
.set(redisKey, value, 'EX', expires, 'NX')
.then(() => true)
.catch(() => false))
) {
if (timeoutAt && timeoutAt > Date.now()) {
throw new ApplicationException('lock timeout');
}
await new Promise((resolve) => setTimeout(resolve, retryDelay));
}
const renewTimer = setInterval(() => {
redis.expire(redisKey, expires);
}, (expires * 1000) / 2);
return async () => {
clearInterval(renewTimer);
await redis.eval(
`
if redis.call("get", KEYS[1]) == ARGV[1]
then
return redis.call("del", KEYS[1])
else
return 0
end
`,
1,
redisKey,
value,
);
};
}
}

View File

@ -53,8 +53,22 @@ export class BaseDbService<Entity extends AppBaseEntity> extends TypeormHelper {
async isDuplicateEntityForUpdate<Dto extends Entity>(
id: string,
dto: Partial<Dto>,
extendsFields?: Array<keyof Dto & string>,
): Promise<false | never>;
async isDuplicateEntityForUpdate<Dto extends Entity>(
old: Entity,
dto: Partial<Dto>,
extendsFields?: Array<keyof Dto & string>,
): Promise<false | never>;
async isDuplicateEntityForUpdate<Dto extends Entity>(
id: string | Entity,
dto: Partial<Dto>,
extendsFields: Array<keyof Dto & string> = [],
): Promise<false | never> {
if (typeof id !== 'string') {
dto = Object.assign({}, id, dto);
id = id.id;
}
const qb = this.repository.createQueryBuilder('entity');
const compareFields = this.getCompareFields(dto, [
...this.uniqueFields,

View File

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

View File

@ -1,12 +1,22 @@
import { PinoLogger } from 'nestjs-pino';
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 app = await NestFactory.create(AppModule, { bodyParser: false });
const configService = app.get(ConfigService);
app.useGlobalPipes(new ValidationPipe());
app.useGlobalPipes(new SanitizePipe());
app.useGlobalPipes(
new ValidationPipe({
transform: true,
}),
);
const httpExceptionFilterLogger = await app.resolve(PinoLogger);
app.useGlobalFilters(new HttpExceptionFilter(httpExceptionFilterLogger));
await app.listen(configService.get<number>('http.port'));
}
bootstrap();

View File

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

View File

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

View File

@ -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: '流水线单元',
});

View File

@ -0,0 +1,15 @@
import { registerEnumType } from '@nestjs/graphql';
export enum TaskStatuses {
success = 'success',
failed = 'failed',
working = 'working',
pending = 'pending',
}
registerEnumType(TaskStatuses, {
name: 'TaskStatuses',
description: '任务状态',
});
export const terminalTaskStatuses = [TaskStatuses.success, TaskStatuses.failed];

View File

@ -0,0 +1,25 @@
import { Field, ObjectType } from '@nestjs/graphql';
import { PipelineUnits } from '../enums/pipeline-units.enum';
import { TaskStatuses } from '../enums/task-statuses.enum';
import { Type } from 'class-transformer';
@ObjectType()
export class PipelineTaskEvent {
@Field()
taskId: string;
@Field()
pipelineId: string;
@Field()
projectId: string;
@Field(() => PipelineUnits, { nullable: true })
unit: PipelineUnits | null;
@Field()
@Type(() => Date)
emittedAt: Date;
@Field()
message: string;
@Field()
messageType: 'stdout' | 'stderr' | 'stdin';
@Field(() => TaskStatuses)
status: TaskStatuses;
}

View File

@ -0,0 +1,34 @@
import { PipelineTask } from './../pipeline-task.entity';
import { PipelineUnits } from '../enums/pipeline-units.enum';
import { Field, HideField, ObjectType } from '@nestjs/graphql';
import { Type } from 'class-transformer';
@ObjectType()
export class PipelineTaskLogMessage {
@HideField()
task: PipelineTask;
@Field(() => PipelineUnits, { nullable: true })
unit?: PipelineUnits;
@Field()
@Type(() => Date)
time: Date;
@Field()
message: string;
@Field()
isError: boolean;
static create(
task: PipelineTask,
unit: PipelineUnits,
message: string,
isError: boolean,
) {
return Object.assign(new PipelineTaskLogMessage(), {
task,
message,
time: new Date(),
unit,
isError,
});
}
}

View File

@ -0,0 +1,17 @@
import { TaskStatuses } from '../enums/task-statuses.enum';
import { PipelineUnits } from '../enums/pipeline-units.enum';
import { Field, ObjectType } from '@nestjs/graphql';
import { Type } from 'class-transformer';
@ObjectType()
export class PipelineTaskLogs {
@Field(() => PipelineUnits)
unit: PipelineUnits;
@Field(() => TaskStatuses)
status: TaskStatuses;
@Type(() => Date)
startedAt?: Date;
@Type(() => Date)
endedAt?: Date;
logs = '';
}

View File

@ -0,0 +1,16 @@
import { Field, InputType, Int, ObjectType } from '@nestjs/graphql';
import { Type } from 'class-transformer';
import { IsInstance, isInstance, ValidateNested } from 'class-validator';
import { WorkUnit } from './work-unit.model';
@InputType('WorkUnitMetadataInput')
@ObjectType()
export class WorkUnitMetadata {
@Field(() => Int)
version = 1;
@Type(() => WorkUnit)
@IsInstance(WorkUnit, { each: true })
@ValidateNested({ each: true })
units: WorkUnit[];
}

View File

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

View File

@ -0,0 +1,88 @@
import { AmqpConnection } from '@golevelup/nestjs-rabbitmq';
import { Test, TestingModule } from '@nestjs/testing';
import { RedisService } from 'nestjs-redis';
import { PipelineTaskFlushService } from './pipeline-task-flush.service';
import { PipelineTaskEvent } from './models/pipeline-task-event';
import { TaskStatuses } from './enums/task-statuses.enum';
import {
EXCHANGE_PIPELINE_TASK_TOPIC,
ROUTE_PIPELINE_TASK_DONE,
} from './pipeline-tasks.constants';
describe('PipelineTaskFlushService', () => {
let service: PipelineTaskFlushService;
let redisService: RedisService;
let amqpConnection: AmqpConnection;
beforeEach(async () => {
const redisClient = {
rpush: jest.fn(() => Promise.resolve()),
lrange: jest.fn(() => Promise.resolve()),
expire: jest.fn(() => Promise.resolve()),
};
const module: TestingModule = await Test.createTestingModule({
providers: [
PipelineTaskFlushService,
{
provide: RedisService,
useValue: {
getClient() {
return redisClient;
},
},
},
{
provide: AmqpConnection,
useValue: {
request: jest.fn(() => Promise.resolve()),
},
},
],
}).compile();
service = module.get<PipelineTaskFlushService>(PipelineTaskFlushService);
redisService = module.get<RedisService>(RedisService);
amqpConnection = module.get<AmqpConnection>(AmqpConnection);
});
it('should be defined', () => {
expect(service).toBeDefined();
});
describe('write', () => {
const amqpMsg = {
properties: { headers: { sender: 'test' } },
} as any;
it('normal', async () => {
const testEvent = new PipelineTaskEvent();
testEvent.taskId = 'test';
testEvent.status = TaskStatuses.working;
const rpush = jest.spyOn(redisService.getClient(), 'rpush');
const request = jest.spyOn(amqpConnection, 'request');
await service.write(testEvent, amqpMsg);
expect(rpush).toBeCalledTimes(1);
expect(rpush.mock.calls[0][0]).toEqual('p-task:log:test');
expect(rpush.mock.calls[0][1]).toEqual(JSON.stringify(testEvent));
expect(request).toBeCalledTimes(1);
});
it('event for which task done', async () => {
const testEvent = new PipelineTaskEvent();
testEvent.taskId = 'test';
testEvent.status = TaskStatuses.success;
const rpush = jest.spyOn(redisService.getClient(), 'rpush');
const request = jest.spyOn(amqpConnection, 'request');
await service.write(testEvent, amqpMsg);
expect(rpush).toBeCalledTimes(1);
expect(request).toBeCalledTimes(1);
expect(request.mock.calls[0][0]).toMatchObject({
exchange: EXCHANGE_PIPELINE_TASK_TOPIC,
routingKey: ROUTE_PIPELINE_TASK_DONE,
payload: {
taskId: 'test',
status: TaskStatuses.success,
runOn: 'test',
},
});
});
});
});

View File

@ -0,0 +1,66 @@
import { AmqpConnection, RabbitSubscribe } from '@golevelup/nestjs-rabbitmq';
import { Injectable } from '@nestjs/common';
import { ConsumeMessage } from 'amqplib';
import { deserialize } from 'class-transformer';
import { RedisService } from 'nestjs-redis';
import { isNil } from 'ramda';
import { getSelfInstanceQueueKey } from '../commons/utils/rabbit-mq';
import { PipelineTaskEvent } from './models/pipeline-task-event';
import {
EXCHANGE_PIPELINE_TASK_TOPIC,
ROUTE_PIPELINE_TASK_DONE,
} from './pipeline-tasks.constants';
import {
EXCHANGE_PIPELINE_TASK_FANOUT,
ROUTE_PIPELINE_TASK_LOG,
QUEUE_WRITE_PIPELINE_TASK_LOG,
} from './pipeline-tasks.constants';
@Injectable()
export class PipelineTaskFlushService {
constructor(
private readonly redisService: RedisService,
private readonly amqpConnection: AmqpConnection,
) {}
@RabbitSubscribe({
exchange: EXCHANGE_PIPELINE_TASK_FANOUT,
routingKey: ROUTE_PIPELINE_TASK_LOG,
queue: getSelfInstanceQueueKey(QUEUE_WRITE_PIPELINE_TASK_LOG),
queueOptions: {
autoDelete: true,
durable: true,
},
})
async write(message: PipelineTaskEvent, amqpMsg: ConsumeMessage) {
const client = this.redisService.getClient();
await client.rpush(this.getKey(message.taskId), JSON.stringify(message));
await client.expire(this.getKey(message.taskId), 600); // ten minutes
if (isNil(message.unit)) {
try {
await this.amqpConnection.request({
exchange: EXCHANGE_PIPELINE_TASK_TOPIC,
routingKey: ROUTE_PIPELINE_TASK_DONE,
payload: {
taskId: message.taskId,
status: message.status,
runOn: amqpMsg.properties.headers.sender,
},
});
} catch (error) {
console.log(error);
}
}
}
async read(taskId: string) {
const raw = await this.redisService
.getClient()
.lrange(this.getKey(taskId), 0, -1);
return raw.map((it) => deserialize(PipelineTaskEvent, it));
}
private getKey(taskId: string) {
return `p-task:log:${taskId}`;
}
}

View File

@ -0,0 +1,42 @@
import { AppBaseEntity } from './../commons/entities/app-base-entity';
import { ObjectType } from '@nestjs/graphql';
import { Column, Entity, ManyToOne, ValueTransformer } 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';
import { plainToClass } from 'class-transformer';
const logsTransformer: ValueTransformer = {
from: (value) => plainToClass(PipelineTaskLogs, value),
to: (value) => value,
};
@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: '[]', transformer: logsTransformer })
logs: PipelineTaskLogs[];
@Column({ type: 'enum', enum: TaskStatuses, default: TaskStatuses.pending })
status: TaskStatuses;
@Column({ nullable: true })
startedAt?: Date;
@Column({ nullable: true })
endedAt?: Date;
@Column({ nullable: true })
runOn: string;
}

View File

@ -0,0 +1,77 @@
import { Test, TestingModule } from '@nestjs/testing';
import { PipelineTaskLogger } from './pipeline-task.logger';
import { PipelineTaskEvent } from './models/pipeline-task-event';
import { take, timeout } from 'rxjs/operators';
describe('PipelineTaskRunner', () => {
let logger: PipelineTaskLogger;
let module: TestingModule;
beforeEach(async () => {
module = await Test.createTestingModule({
providers: [PipelineTaskLogger],
}).compile();
logger = module.get(PipelineTaskLogger);
});
it('should be defined', () => {
expect(logger).toBeDefined();
});
describe('getMessage$', () => {
it('normal', async () => {
const event = new PipelineTaskEvent();
event.taskId = 'test';
const emittedAt = new Date();
event.emittedAt = emittedAt.toISOString() as any;
const message$ = logger.getMessage$('test');
let receiveEvent;
message$.pipe(take(1)).subscribe((value) => (receiveEvent = value));
await logger.handleEvent(event);
expect(receiveEvent).toMatchObject({
...event,
emittedAt,
});
});
it('no match', async () => {
const event = new PipelineTaskEvent();
event.taskId = 'test';
const message$ = logger.getMessage$('other');
setTimeout(() => {
logger.handleEvent(event);
});
await expect(
message$.pipe(take(1), timeout(100)).toPromise(),
).rejects.toThrow(/timeout/i);
});
it('multiple subscribers', async () => {
const event = new PipelineTaskEvent();
event.taskId = 'test';
const message$ = logger.getMessage$('test');
const message2$ = logger.getMessage$('test');
setTimeout(() => {
logger.handleEvent(event);
logger.handleEvent(event);
});
await Promise.all([
expect(
message$.pipe(take(1), timeout(100)).toPromise(),
).resolves.toEqual(event),
expect(
message2$.pipe(take(1), timeout(100)).toPromise(),
).resolves.toEqual(event),
]);
});
});
describe('onModuleDestroy', () => {
it('complete observable when destroying module', async () => {
logger.onModuleDestroy();
await expect(
(logger as any).message$.toPromise(),
).resolves.toBeUndefined();
});
});
});

View File

@ -0,0 +1,38 @@
import { RabbitSubscribe } from '@golevelup/nestjs-rabbitmq';
import { Injectable, OnModuleDestroy } from '@nestjs/common';
import { plainToClass } from 'class-transformer';
import { Observable, Subject } from 'rxjs';
import { filter } from 'rxjs/operators';
import { PipelineTaskEvent } from './models/pipeline-task-event';
import {
EXCHANGE_PIPELINE_TASK_FANOUT,
QUEUE_HANDLE_PIPELINE_TASK_LOG_EVENT,
ROUTE_PIPELINE_TASK_LOG,
} from './pipeline-tasks.constants';
@Injectable()
export class PipelineTaskLogger implements OnModuleDestroy {
private readonly messageSubject = new Subject<PipelineTaskEvent>();
private readonly message$: Observable<PipelineTaskEvent> =
this.messageSubject.pipe();
@RabbitSubscribe({
exchange: EXCHANGE_PIPELINE_TASK_FANOUT,
routingKey: ROUTE_PIPELINE_TASK_LOG,
queue: QUEUE_HANDLE_PIPELINE_TASK_LOG_EVENT,
queueOptions: {
autoDelete: true,
},
})
async handleEvent(message: PipelineTaskEvent) {
this.messageSubject.next(plainToClass(PipelineTaskEvent, message));
}
getMessage$(taskId: string) {
return this.message$.pipe(filter((event) => event.taskId === taskId));
}
onModuleDestroy() {
this.messageSubject.complete();
}
}

View File

@ -0,0 +1,351 @@
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';
import { PipelineTask } from './pipeline-task.entity';
import { Pipeline } from '../pipelines/pipeline.entity';
import { Project } from '../projects/project.entity';
import { TaskStatuses } from './enums/task-statuses.enum';
import { getLoggerToken, PinoLogger } from 'nestjs-pino';
import { PipelineTaskRunner } from './pipeline-task.runner';
import { WorkUnitMetadata } from './models/work-unit-metadata.model';
import { AmqpConnection } from '@golevelup/nestjs-rabbitmq';
describe('PipelineTaskRunner', () => {
let runner: PipelineTaskRunner;
let deployByPM2Service: DeployByPm2Service;
beforeEach(async () => {
const module: TestingModule = await Test.createTestingModule({
providers: [
{
provide: ReposService,
useValue: {
getWorkspaceRootByTask: () => 'workspace-root',
checkout: async () => undefined,
},
},
{
provide: getLoggerToken(PipelineTaskRunner.name),
useValue: new PinoLogger({}),
},
{
provide: 'spawn',
useValue: () => undefined,
},
PipelineTaskRunner,
{
provide: AmqpConnection,
useValue: {},
},
{
provide: DeployByPm2Service,
useValue: {
deploy: () => Promise.resolve(),
},
},
],
}).compile();
module.get(ReposService);
runner = module.get(PipelineTaskRunner);
deployByPM2Service = module.get(DeployByPm2Service);
});
it('should be defined', () => {
expect(runner).toBeDefined();
});
it('onNewTask', async () => {
const task = new PipelineTask();
let tmpTask;
const doTask = jest
.spyOn(runner, 'doTask')
.mockImplementation(async (task) => {
tmpTask = task;
});
await runner.onNewTask(task);
expect(tmpTask).toEqual(task);
expect(doTask).toBeCalledTimes(1);
});
describe('test biz', () => {
let emitEvent: jest.SpyInstance;
beforeEach(() => {
emitEvent = jest
.spyOn(runner, 'emitEvent')
.mockImplementation(() => Promise.resolve());
});
describe('doTask', () => {
let checkout: jest.SpyInstance;
let doTaskUnit: jest.SpyInstance;
beforeEach(() => {
checkout = jest
.spyOn(runner, 'checkout')
.mockImplementation(() => Promise.resolve('/null'));
doTaskUnit = jest
.spyOn(runner, 'doTaskUnit')
.mockImplementation(() => Promise.resolve());
});
it('only checkout', async () => {
const task = new PipelineTask();
(task.id = 'taskId'), (task.pipeline = new Pipeline());
task.units = [PipelineUnits.checkout];
task.pipeline.id = 'pipelineId';
task.pipeline.project = new Project();
task.pipeline.project.id = 'projectId';
task.pipeline.workUnitMetadata = new WorkUnitMetadata();
task.pipeline.workUnitMetadata.version = 1;
task.pipeline.workUnitMetadata.units = [
{
type: PipelineUnits.checkout,
scripts: [],
},
];
await runner.doTask(task);
expect(checkout).toBeCalledTimes(1);
expect(doTaskUnit).toBeCalledTimes(0);
expect(emitEvent).toBeCalledTimes(2);
expect(emitEvent.mock.calls[0][0]).toMatchObject(task);
expect(emitEvent.mock.calls[0][1]).toBeNull();
expect(emitEvent.mock.calls[0][2]).toEqual(TaskStatuses.working);
expect(emitEvent.mock.calls[1][0]).toMatchObject(task);
expect(emitEvent.mock.calls[1][1]).toBeNull();
expect(emitEvent.mock.calls[1][2]).toEqual(TaskStatuses.success);
});
it('many units', async () => {
const task = new PipelineTask();
(task.id = 'taskId'), (task.pipeline = new Pipeline());
task.units = [
PipelineUnits.checkout,
PipelineUnits.test,
PipelineUnits.deploy,
];
task.pipeline.id = 'pipelineId';
task.pipeline.project = new Project();
task.pipeline.project.id = 'projectId';
task.pipeline.workUnitMetadata = new WorkUnitMetadata();
task.pipeline.workUnitMetadata.version = 1;
task.pipeline.workUnitMetadata.units = [
{
type: PipelineUnits.checkout,
scripts: [],
},
{
type: PipelineUnits.installDependencies,
scripts: ['pwd'],
},
{
type: PipelineUnits.test,
scripts: ['pwd'],
},
{
type: PipelineUnits.deploy,
scripts: ['pwd', 'uname'],
},
];
await runner.doTask(task);
expect(checkout).toBeCalledTimes(1);
expect(doTaskUnit).toBeCalledTimes(1);
expect(emitEvent).toBeCalledTimes(2);
});
it('unit work failed', async () => {
const task = new PipelineTask();
(task.id = 'taskId'), (task.pipeline = new Pipeline());
task.units = [PipelineUnits.checkout, PipelineUnits.test];
task.pipeline.id = 'pipelineId';
task.pipeline.project = new Project();
task.pipeline.project.id = 'projectId';
task.pipeline.workUnitMetadata = new WorkUnitMetadata();
task.pipeline.workUnitMetadata.version = 1;
task.pipeline.workUnitMetadata.units = [
{
type: PipelineUnits.checkout,
scripts: [],
},
{
type: PipelineUnits.test,
scripts: ['pwd'],
},
];
doTaskUnit = jest
.spyOn(runner, 'doTaskUnit')
.mockImplementation(() => Promise.reject(new Error('test error')));
await runner.doTask(task);
expect(checkout).toBeCalledTimes(1);
expect(doTaskUnit).toBeCalledTimes(1);
expect(emitEvent).toBeCalledTimes(2);
expect(emitEvent.mock.calls[1][0]).toMatchObject(task);
expect(emitEvent.mock.calls[1][1]).toBeNull();
expect(emitEvent.mock.calls[1][2]).toEqual(TaskStatuses.failed);
});
});
describe('doTaskUnit', () => {
it('success', async () => {
const runScript = jest
.spyOn(runner, 'runScript')
.mockImplementation(() => Promise.resolve());
const task = new PipelineTask();
const unit = PipelineUnits.test;
const workspacePath = '/null';
await runner.doTaskUnit(unit, ['pwd'], task, workspacePath);
expect(emitEvent.mock.calls[0][0]).toEqual(task);
expect(emitEvent.mock.calls[0][1]).toEqual(unit);
expect(emitEvent.mock.calls[0][2]).toEqual(TaskStatuses.working);
expect(emitEvent.mock.calls[1][0]).toEqual(task);
expect(emitEvent.mock.calls[1][1]).toEqual(unit);
expect(emitEvent.mock.calls[1][2]).toEqual(TaskStatuses.success);
expect(runScript.mock.calls[0][0]).toEqual('pwd');
expect(runScript.mock.calls[0][1]).toEqual(workspacePath);
expect(runScript.mock.calls[0][2]).toEqual(task);
expect(runScript.mock.calls[0][3]).toEqual(unit);
});
it('failed', async () => {
const runScript = jest
.spyOn(runner, 'runScript')
.mockImplementation(() => Promise.reject(new Error('test error')));
const task = new PipelineTask();
const unit = PipelineUnits.test;
const workspacePath = '/null';
await expect(
runner.doTaskUnit(unit, ['pwd'], task, workspacePath),
).rejects.toThrow('test error');
expect(emitEvent.mock.calls[1]?.[0]).toEqual(task);
expect(emitEvent.mock.calls[1]?.[1]).toEqual(unit);
expect(emitEvent.mock.calls[1]?.[2]).toEqual(TaskStatuses.failed);
expect(runScript).toBeCalledTimes(1);
});
});
describe('runScript', () => {
it('normal', async () => {
const spawn = jest.fn<any, any>(() => ({
stdout: {
on: () => undefined,
},
stderr: {
on: () => undefined,
},
addListener: (_: any, fn: (code: number) => void) => {
fn(0);
},
}));
(runner as any).spawn = spawn;
const task = new PipelineTask();
task.id = 'taskId';
const unit = PipelineUnits.deploy;
await runner.runScript('script name', 'workspaceRoot', task, unit);
expect(spawn).toHaveBeenCalledTimes(1);
expect(spawn.mock.calls[0][0]).toEqual('script name');
expect(spawn.mock.calls[0][1]).toMatchObject({
shell: true,
cwd: 'workspaceRoot',
});
});
it('failed', async () => {
const spawn = jest.fn(() => ({
stdout: {
on: () => undefined,
},
stderr: {
on: () => undefined,
},
addListener: (_: any, fn: (code: number) => void) => {
fn(1);
},
}));
(runner as any).spawn = spawn;
const task = new PipelineTask();
task.id = 'taskId';
const unit = PipelineUnits.deploy;
expect(
runner.runScript('script name', 'workspaceRoot', task, unit),
).rejects.toThrowError();
});
it('wait emit message done', async () => {
let finishedFn: () => void;
const on = jest.fn((_: any, fn: (buff: Buffer) => void) => {
setTimeout(() => {
fn(Buffer.from('message 1'));
setTimeout(() => {
fn(Buffer.from('message 2'));
setTimeout(() => {
fn(Buffer.from('message 3'));
finishedFn();
}, 1000);
}, 10);
}, 10);
});
const spawn = jest.fn(() => ({
stdout: {
on,
},
stderr: {
on,
},
addListener: (_: any, fn: (code: number) => void) => {
finishedFn = () => fn(0);
},
}));
let emitSuccessCount = 0;
jest.spyOn(runner, 'emitEvent').mockImplementation(() => {
return new Promise((resolve) => {
setTimeout(() => {
emitSuccessCount++;
resolve();
}, 1000);
});
});
(runner as any).spawn = spawn;
const task = new PipelineTask();
task.id = 'taskId';
const unit = PipelineUnits.deploy;
await runner.runScript('script name', 'workspaceRoot', task, unit);
expect(emitSuccessCount).toEqual(1 + 6);
});
});
});
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);
});
});
});

View File

@ -0,0 +1,339 @@
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';
import { ApplicationException } from '../commons/exceptions/application.exception';
import { PipelineUnits } from './enums/pipeline-units.enum';
import { TaskStatuses } from './enums/task-statuses.enum';
import { InjectPinoLogger, PinoLogger } from 'nestjs-pino';
import {
AmqpConnection,
RabbitRPC,
RabbitSubscribe,
} from '@golevelup/nestjs-rabbitmq';
import { PipelineTaskEvent } from './models/pipeline-task-event';
import { last } from 'ramda';
import { Inject } from '@nestjs/common';
import {
EXCHANGE_PIPELINE_TASK_TOPIC,
QUEUE_PIPELINE_TASK_KILL,
ROUTE_PIPELINE_TASK_KILL,
} from './pipeline-tasks.constants';
import {
EXCHANGE_PIPELINE_TASK_FANOUT,
ROUTE_PIPELINE_TASK_LOG,
} from './pipeline-tasks.constants';
import {
getInstanceName,
getSelfInstanceQueueKey,
getSelfInstanceRouteKey,
} from '../commons/utils/rabbit-mq';
import { rm } from 'fs/promises';
import { rename } from 'fs/promises';
import { join } from 'path';
type Spawn = typeof spawn;
export class PipelineTaskRunner {
readonly processes = new Map<string, ChildProcessWithoutNullStreams>();
readonly stopTaskIds = new Set<string>();
constructor(
private readonly reposService: ReposService,
@InjectPinoLogger(PipelineTaskRunner.name)
private readonly logger: PinoLogger,
@Inject('spawn')
private readonly spawn: Spawn,
private readonly amqpConnection: AmqpConnection,
private readonly deployByPM2Service: DeployByPm2Service,
) {}
@RabbitSubscribe({
exchange: 'new-pipeline-task',
routingKey: 'mac',
queue: 'mac.new-pipeline-task',
})
async onNewTask(task: PipelineTask) {
this.logger.info({ task }, 'on new task [%s].', task.id);
try {
await this.doTask(task);
} catch (err) {
this.logger.error({ task, err }, err.message);
}
}
@RabbitRPC({
exchange: EXCHANGE_PIPELINE_TASK_TOPIC,
routingKey: getSelfInstanceRouteKey(ROUTE_PIPELINE_TASK_KILL),
queue: getSelfInstanceQueueKey(QUEUE_PIPELINE_TASK_KILL),
queueOptions: {
autoDelete: true,
durable: true,
},
})
async onStopTask(task: PipelineTask) {
this.logger.info({ task }, 'on stop task [%s].', task.id);
this.stopTaskIds.add(task.id);
const process = this.processes.get(task.id);
if (process) {
this.logger.info({ task }, 'send signal SIGINT to child process.');
process.kill('SIGINT');
setTimeout(() => {
setTimeout(() => {
this.stopTaskIds.delete(task.id);
}, 10_000);
if (process === this.processes.get(task.id)) {
this.logger.info({ task }, 'send signal SIGKILL to child process.');
process.kill('SIGKILL');
return;
}
if (this.processes.has(task.id)) {
this.logger.error(
{ task },
'this pipeline task not stop yet. there is a new process running, maybe is a bug about error capture',
);
}
}, 10_000);
} else {
this.logger.info({ task }, 'child process is not running.');
}
return true;
}
async doTask(task: PipelineTask) {
if (task.pipeline.workUnitMetadata.version !== 1) {
throw new ApplicationException(
'work unit metadata version is not match.',
);
}
await this.emitEvent(
task,
null,
TaskStatuses.working,
`[start task]`,
'stdout',
);
this.logger.info('running task [%s].', task.id);
try {
let workspaceRoot = await this.checkout(task);
const units = task.units
.filter((unit) => unit !== PipelineUnits.checkout)
.map(
(type) =>
task.pipeline.workUnitMetadata.units.find(
(unit) => unit.type === type,
) ?? { type: type, scripts: [] },
);
this.logger.info({ units }, 'begin run units.');
for (const unit of units) {
if (unit.type === PipelineUnits.deploy) {
const oldRoot = workspaceRoot;
workspaceRoot = this.reposService.getDeployRoot(task);
if (oldRoot !== workspaceRoot) {
await rm(workspaceRoot, { force: true, recursive: true });
await rename(oldRoot, workspaceRoot);
}
await this.emitEvent(
task,
unit.type,
TaskStatuses.success,
`[deploy] change deploy folder content success`,
'stdout',
);
}
await this.doTaskUnit(unit.type, unit.scripts, task, workspaceRoot);
}
await this.emitEvent(
task,
null,
TaskStatuses.success,
`[finished task] success`,
'stdout',
);
this.logger.info({ task }, 'task [%s] completed.', task.id);
} catch (err) {
await this.emitEvent(
task,
null,
TaskStatuses.failed,
`[finished unit] ${err.message}`,
'stderr',
);
this.logger.error({ task, error: err }, 'task [%s] failed.', task.id);
} finally {
}
}
async doTaskUnit(
unit: PipelineUnits,
scripts: string[],
task: PipelineTask,
workspaceRoot: string,
) {
await this.emitEvent(
task,
unit,
TaskStatuses.working,
`[begin unit] ${unit}`,
'stdin',
);
this.logger.info({ task }, 'curr unit is %s', unit);
try {
for (const script of scripts) {
this.logger.debug('begin runScript %s', script);
if (this.stopTaskIds.has(task.id)) {
throw new ApplicationException('Task is be KILLED');
}
await this.runScript(script, workspaceRoot, task, unit);
this.logger.debug('end runScript %s', script);
}
await this.emitEvent(
task,
unit,
TaskStatuses.success,
`[finished unit] ${unit}`,
'stdout',
);
} catch (err) {
await this.emitEvent(
task,
unit,
TaskStatuses.failed,
`[finished unit] ${err.message}`,
'stderr',
);
throw err;
}
}
async checkout(task: PipelineTask) {
await this.emitEvent(
task,
PipelineUnits.checkout,
TaskStatuses.working,
'[begin unit] checkout',
'stdin',
);
try {
const path = await this.reposService.checkout4Task(task);
await this.emitEvent(
task,
PipelineUnits.checkout,
TaskStatuses.success,
'checkout success.',
'stdout',
);
return path;
} catch (err) {
await this.emitEvent(
task,
PipelineUnits.checkout,
TaskStatuses.failed,
'checkout failed.',
'stderr',
);
throw err;
}
}
async emitEvent(
task: PipelineTask,
unit: PipelineUnits | null,
status: TaskStatuses,
message: string,
messageType: 'stderr' | 'stdout' | 'stdin',
) {
const event: PipelineTaskEvent = {
taskId: task.id,
pipelineId: task.pipeline.id,
projectId: task.pipeline.project.id,
unit,
emittedAt: new Date(),
message: last(message) === '\n' ? message : message + '\n',
messageType,
status,
};
this.amqpConnection
.publish(EXCHANGE_PIPELINE_TASK_FANOUT, ROUTE_PIPELINE_TASK_LOG, event, {
headers: {
sender: getInstanceName(),
},
})
.catch((error) => {
this.logger.error(
{ error, event },
'send event message to queue failed. %s',
error.message,
);
});
}
async runScript(
script: string,
workspaceRoot: string,
task: PipelineTask,
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,
cwd: workspaceRoot,
});
this.processes.set(task.id, sub);
let loggingCount = 0; // semaphore
sub.stderr.on('data', (data: Buffer) => {
const str = data.toString();
loggingCount++;
this.emitEvent(task, unit, TaskStatuses.working, str, 'stdout').finally(
() => loggingCount--,
);
});
sub.stdout.on('data', (data: Buffer) => {
const str = data.toString();
loggingCount++;
this.emitEvent(task, unit, TaskStatuses.working, str, 'stderr').finally(
() => loggingCount--,
);
});
sub.addListener('close', async (code) => {
this.processes.delete(task.id);
await new Promise<void>(async (resolve) => {
for (let i = 0; i < 10 && loggingCount > 0; i++) {
await new Promise((resolve) => setTimeout(resolve, 500));
this.logger.debug('waiting logging... (%dx500ms)', i);
}
resolve();
});
if (code === 0) {
return resolve();
}
if (this.stopTaskIds.has(task.id)) {
throw reject(new ApplicationException('Task is be KILLED'));
}
return reject(new ApplicationException('exec script failed'));
});
});
}
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;
}
}
}

View File

@ -0,0 +1,11 @@
export const EXCHANGE_PIPELINE_TASK_TOPIC = 'pipeline-task.topic';
export const EXCHANGE_PIPELINE_TASK_FANOUT = 'pipeline-task.fanout';
export const ROUTE_PIPELINE_TASK_LOG = 'pipeline-task-log';
export const QUEUE_HANDLE_PIPELINE_TASK_LOG_EVENT = 'pipeline-task-log';
export const QUEUE_WRITE_PIPELINE_TASK_LOG = 'write-pipeline-task-log';
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');

View File

@ -0,0 +1,76 @@
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 { RabbitMQModule } from '@golevelup/nestjs-rabbitmq';
import { ConfigModule, ConfigService } from '@nestjs/config';
import { PipelineTaskRunner } from './pipeline-task.runner';
import { spawn } from 'child_process';
import {
EXCHANGE_PIPELINE_TASK_FANOUT,
EXCHANGE_PIPELINE_TASK_TOPIC,
} from './pipeline-tasks.constants';
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: [
CommonsModule,
TypeOrmModule.forFeature([PipelineTask, Pipeline]),
RedisModule,
ReposModule,
RabbitMQModule.forRootAsync(RabbitMQModule, {
imports: [ConfigModule],
useFactory: (configService: ConfigService) => ({
uri: configService.get<string>('db.rabbitmq.uri'),
exchanges: [
{
name: 'new-pipeline-task',
type: 'fanout',
options: {
durable: true,
autoDelete: true,
},
},
{
name: EXCHANGE_PIPELINE_TASK_FANOUT,
type: 'fanout',
options: {
durable: false,
autoDelete: true,
},
},
{
name: EXCHANGE_PIPELINE_TASK_TOPIC,
type: 'topic',
options: {
durable: false,
autoDelete: true,
},
},
],
}),
inject: [ConfigService],
}),
],
providers: [
PipelineTasksService,
PipelineTasksResolver,
PipelineTaskRunner,
PipelineTaskLogger,
{
provide: 'spawn',
useValue: spawn,
},
PipelineTaskFlushService,
DeployByPm2Service,
],
exports: [PipelineTasksService],
})
export class PipelineTasksModule {}

View File

@ -0,0 +1,35 @@
import { JwtService } from '@nestjs-lib/auth';
import { Test, TestingModule } from '@nestjs/testing';
import { PipelineTaskLogger } from './pipeline-task.logger';
import { PipelineTasksResolver } from './pipeline-tasks.resolver';
import { PipelineTasksService } from './pipeline-tasks.service';
describe('PipelineTasksResolver', () => {
let resolver: PipelineTasksResolver;
beforeEach(async () => {
const module: TestingModule = await Test.createTestingModule({
providers: [
PipelineTasksResolver,
{
provide: PipelineTasksService,
useValue: {},
},
{
provide: PipelineTaskLogger,
useValue: {},
},
{
provide: JwtService,
useValue: {},
},
],
}).compile();
resolver = module.get<PipelineTasksResolver>(PipelineTasksResolver);
});
it('should be defined', () => {
expect(resolver).toBeDefined();
});
});

View File

@ -0,0 +1,63 @@
import { Resolver, Args, Mutation, Subscription, Query } from '@nestjs/graphql';
import { PipelineTask } from './pipeline-task.entity';
import { PipelineTasksService } from './pipeline-tasks.service';
import { CreatePipelineTaskInput } from './dtos/create-pipeline-task.input';
import { PipelineTaskLogArgs } from './dtos/pipeline-task-log.args';
import { plainToClass } from 'class-transformer';
import { PipelineTaskLogger } from './pipeline-task.logger';
import { observableToAsyncIterable } from '@graphql-tools/utils';
import { PipelineTaskEvent } from './models/pipeline-task-event';
import { Roles, AccountRole } from '@nestjs-lib/auth';
@Roles(AccountRole.admin, AccountRole.super)
@Resolver()
export class PipelineTasksResolver {
constructor(
private readonly service: PipelineTasksService,
private readonly taskLogger: PipelineTaskLogger,
) {}
@Mutation(() => PipelineTask)
async createPipelineTask(@Args('task') taskDto: CreatePipelineTaskInput) {
return await this.service.addTask(taskDto);
}
@Subscription(() => PipelineTaskEvent, {
resolve: (value) => {
const data = plainToClass(PipelineTaskEvent, value);
return data;
},
})
async pipelineTaskEvent(@Args() args: PipelineTaskLogArgs) {
const task = await this.service.findTaskById(args.taskId);
return observableToAsyncIterable<PipelineTaskEvent>(
this.taskLogger.getMessage$(task.id),
);
}
@Subscription(() => PipelineTask, {
resolve: (value) => {
return value;
},
})
async pipelineTaskChanged(@Args('id') id: string) {
// return await this.service.watchTaskUpdated(id);
}
@Query(() => [PipelineTask])
async listPipelineTaskByPipelineId(@Args('pipelineId') pipelineId: string) {
return await this.service.listTasksByPipelineId(pipelineId);
}
@Query(() => PipelineTask)
async pipelineTask(@Args('id') id: string) {
return await this.service.findTaskById(id);
}
@Mutation(() => Boolean)
async stopPipelineTask(@Args('id') id: string) {
const task = await this.service.findTaskById(id);
await this.service.stopTask(task);
return true;
}
}

View File

@ -0,0 +1,99 @@
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 } from '../pipelines/pipeline.entity';
import { Repository } from 'typeorm';
import { AmqpConnection } from '@golevelup/nestjs-rabbitmq';
import { PipelineTaskFlushService } from './pipeline-task-flush.service';
import { getLoggerToken, PinoLogger } from 'nestjs-pino';
describe('PipelineTasksService', () => {
let service: PipelineTasksService;
let module: TestingModule;
let taskRepository: Repository<PipelineTask>;
let pipelineRepository: Repository<Pipeline>;
beforeEach(async () => {
module = await Test.createTestingModule({
providers: [
PipelineTasksService,
{
provide: getRepositoryToken(PipelineTask),
useValue: new Repository(),
},
{
provide: getRepositoryToken(Pipeline),
useValue: new Repository(),
},
{
provide: AmqpConnection,
useValue: {},
},
{
provide: PipelineTaskFlushService,
useValue: {},
},
{
provide: getLoggerToken(PipelineTasksService.name),
useValue: new PinoLogger({}),
},
],
}).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);
jest.spyOn(taskRepository, 'findOne').mockImplementation(async () => null);
});
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);
// const findOne = jest.spyOn(taskRepository, 'findOne');
// await service.addTask({ pipelineId: 'test', commit: 'test', units: [] }),
// expect(save.mock.calls[0][0]).toMatchObject({
// pipelineId: 'test',
// commit: 'test',
// units: [],
// });
// expect(findOne).toBeCalled();
// });
// it('add task', async () => {
// const lpush = jest.spyOn(redisClient, 'lpush');
// 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(),
// });
// });
// });
});

View File

@ -0,0 +1,158 @@
import { BadRequestException, 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 { Pipeline } from '../pipelines/pipeline.entity';
import debug from 'debug';
import { AmqpConnection, RabbitRPC } from '@golevelup/nestjs-rabbitmq';
import {
EXCHANGE_PIPELINE_TASK_TOPIC,
QUEUE_PIPELINE_TASK_DONE,
ROUTE_PIPELINE_TASK_DONE,
} from './pipeline-tasks.constants';
import { PipelineTaskFlushService } from './pipeline-task-flush.service';
import { find, isNil, propEq } from 'ramda';
import { PipelineTaskLogs } from './models/pipeline-task-logs.model';
import { TaskStatuses, terminalTaskStatuses } from './enums/task-statuses.enum';
import { InjectPinoLogger, PinoLogger } from 'nestjs-pino';
import { getAppInstanceRouteKey } from '../commons/utils/rabbit-mq';
import { ROUTE_PIPELINE_TASK_KILL } from './pipeline-tasks.constants';
const log = debug('fennec:pipeline-tasks:service');
@Injectable()
export class PipelineTasksService {
constructor(
@InjectRepository(PipelineTask)
private readonly repository: Repository<PipelineTask>,
@InjectRepository(Pipeline)
private readonly pipelineRepository: Repository<Pipeline>,
private readonly amqpConnection: AmqpConnection,
private readonly eventFlushService: PipelineTaskFlushService,
@InjectPinoLogger(PipelineTasksService.name)
private readonly logger: PinoLogger,
) {}
async addTask(dto: CreatePipelineTaskInput) {
const pipeline = await this.pipelineRepository.findOneOrFail({
where: { id: dto.pipelineId },
relations: ['project'],
});
// const hasUnfinishedTask = await this.repository
// .findOne({
// pipelineId: dto.pipelineId,
// commit: dto.commit,
// status: In([TaskStatuses.pending, TaskStatuses.working]),
// })
// .then((val) => !isNil(val));
// if (hasUnfinishedTask) {
// throw new ConflictException(
// 'There are the same tasks among the unfinished tasks!',
// );
// }
const task = await this.repository.save(this.repository.create(dto));
task.pipeline = pipeline;
this.amqpConnection.publish('new-pipeline-task', 'mac', task);
return task;
}
async findTaskById(id: string) {
return await this.repository.findOneOrFail({ id });
}
async listTasksByPipelineId(pipelineId: string) {
return await this.repository.find({ pipelineId });
}
async listTasksByCommitHash(hash: string) {
return await this.repository.find({
where: { commit: hash },
order: { createdAt: 'DESC' },
});
}
getRedisTokens(pipeline: Pipeline): [string, string] {
return [`pipeline-${pipeline.id}:lck`, `pipeline-${pipeline.id}:tasks`];
}
@RabbitRPC({
exchange: EXCHANGE_PIPELINE_TASK_TOPIC,
routingKey: ROUTE_PIPELINE_TASK_DONE,
queue: QUEUE_PIPELINE_TASK_DONE,
queueOptions: {
autoDelete: true,
durable: true,
},
})
async updateByEvent({ taskId, runOn }: { taskId: string; runOn: string }) {
try {
const [events, task] = await Promise.all([
this.eventFlushService.read(taskId),
this.findTaskById(taskId),
]);
this.logger.info('[updateByEvent] start. taskId: %s', taskId);
for (const event of events) {
if (isNil(event.unit)) {
if (
event.status !== TaskStatuses.pending &&
task.status === TaskStatuses.pending
) {
task.startedAt = event.emittedAt;
} else if (terminalTaskStatuses.includes(event.status)) {
task.endedAt = event.emittedAt;
}
task.status = event.status;
} else {
let l: PipelineTaskLogs = find<PipelineTaskLogs>(
propEq('unit', event.unit),
task.logs,
);
if (isNil(l)) {
l = {
unit: event.unit,
startedAt: event.emittedAt,
endedAt: null,
logs: event.message,
status: event.status,
};
task.logs.push(l);
} else {
l.logs += event.message;
}
if (terminalTaskStatuses.includes(event.status)) {
l.endedAt = event.emittedAt;
}
l.status = event.status;
}
}
task.runOn = runOn;
await this.repository.update({ id: taskId }, task);
this.logger.info('[updateByEvent] success. taskId: %s', taskId);
return task;
} catch (error) {
this.logger.error(
{ error },
'[updateByEvent] failed. taskId: %s',
taskId,
);
}
}
async stopTask(task: PipelineTask) {
if (isNil(task.runOn)) {
throw new BadRequestException(
"the task have not running instance on database. field 'runOn' is nil",
);
}
await this.amqpConnection.request({
exchange: EXCHANGE_PIPELINE_TASK_TOPIC,
routingKey: getAppInstanceRouteKey(ROUTE_PIPELINE_TASK_KILL, task.runOn),
payload: task,
});
}
}

View File

@ -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);
});
});

View File

@ -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;
}
}

View File

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

View File

@ -0,0 +1,44 @@
import { Roles, AccountRole } from '@nestjs-lib/auth';
import { Query } from '@nestjs/graphql';
import {
Args,
Parent,
ResolveField,
Resolver,
Subscription,
} from '@nestjs/graphql';
import { PipelineTasksService } from '../pipeline-tasks/pipeline-tasks.service';
import { Commit, LogFields } from '../repos/dtos/log-list.model';
import { PipelinesService } from './pipelines.service';
@Roles(AccountRole.admin, AccountRole.super)
@Resolver(() => Commit)
export class CommitLogsResolver {
constructor(
private readonly service: PipelinesService,
private readonly taskServices: PipelineTasksService,
) {}
@Subscription(() => String, { resolve: (val) => val, nullable: true })
async syncCommits(
@Args('pipelineId', { type: () => String })
pipelineId: string,
@Args('appInstance', { type: () => String, nullable: true })
appInstance?: string,
) {
const pipeline = await this.service.findOneWithProject(pipelineId);
const syncCommitsPromise = this.service.syncCommits(pipeline, appInstance);
return (async function* () {
yield await syncCommitsPromise;
})();
}
@ResolveField()
async tasks(@Parent() commit: LogFields) {
return await this.taskServices.listTasksByCommitHash(commit.hash);
}
@Query(() => [Commit], { nullable: true })
async commits(@Args('pipelineId', { type: () => String }) id: string) {
const pipeline = await this.service.findOneWithProject(id);
return await this.service.listCommits(pipeline);
}
}

View File

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

View File

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

View File

@ -0,0 +1,9 @@
import { InputType, OmitType } from '@nestjs/graphql';
import { CreatePipelineInput } from './create-pipeline.input';
@InputType()
export class UpdatePipelineInput extends OmitType(CreatePipelineInput, [
'projectId',
]) {
id: string;
}

View File

@ -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;
}

View File

@ -0,0 +1,29 @@
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 { CommitLogsResolver } from './commit-logs.resolver';
import { PipelineTasksModule } from '../pipeline-tasks/pipeline-tasks.module';
import { ReposModule } from '../repos/repos.module';
import { RabbitMQModule } from '@golevelup/nestjs-rabbitmq';
import { ConfigModule, ConfigService } from '@nestjs/config';
import { CommonsModule } from '../commons/commons.module';
@Module({
imports: [
CommonsModule,
TypeOrmModule.forFeature([Pipeline]),
PipelineTasksModule,
RabbitMQModule.forRootAsync(RabbitMQModule, {
imports: [ConfigModule],
useFactory: (configService: ConfigService) => ({
uri: configService.get<string>('db.rabbitmq.uri'),
exchanges: [],
}),
inject: [ConfigService],
}),
],
providers: [PipelinesResolver, PipelinesService, CommitLogsResolver],
})
export class PipelinesModule {}

View File

@ -0,0 +1,30 @@
import { JwtService } from '@nestjs-lib/auth';
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: {},
},
{
provide: JwtService,
useValue: {},
},
],
}).compile();
resolver = module.get<PipelinesResolver>(PipelinesResolver);
});
it('should be defined', () => {
expect(resolver).toBeDefined();
});
});

View File

@ -0,0 +1,45 @@
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 { Roles, AccountRole } from '@nestjs-lib/auth';
@Roles(AccountRole.admin, AccountRole.super)
@Resolver()
export class PipelinesResolver {
constructor(private readonly service: PipelinesService) {}
@Query(() => [Pipeline])
async pipelines(@Args() dto: ListPipelineArgs) {
return await this.service.list(dto);
}
@Query(() => Pipeline)
async pipeline(@Args('id', { type: () => String }) id: string) {
return await this.service.findOne(id);
}
@Mutation(() => Pipeline)
async createPipeline(
@Args('pipeline', { type: () => CreatePipelineInput })
dto: CreatePipelineInput,
) {
return await this.service.create(dto);
}
@Mutation(() => Pipeline)
async updatePipeline(
@Args('pipeline', { type: () => UpdatePipelineInput })
dto: UpdatePipelineInput,
) {
const tmp = await this.service.update(dto);
console.log(tmp);
return tmp;
}
@Mutation(() => Number)
async deletePipeline(@Args('id', { type: () => String }) id: string) {
return await this.service.remove(id);
}
}

View File

@ -0,0 +1,49 @@
import { Test, TestingModule } from '@nestjs/testing';
import { PipelinesService } from './pipelines.service';
import { Pipeline } from './pipeline.entity';
import { getRepositoryToken } from '@nestjs/typeorm';
import { Repository } from 'typeorm';
import { Project } from '../projects/project.entity';
import { AmqpConnection } from '@golevelup/nestjs-rabbitmq';
describe('PipelinesService', () => {
let service: PipelinesService;
let repository: Repository<Pipeline>;
let pipeline: Pipeline;
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: AmqpConnection,
useValue: {},
},
],
}).compile();
service = module.get<PipelinesService>(PipelinesService);
repository = module.get(getRepositoryToken(Pipeline));
});
it('should be defined', () => {
expect(service).toBeDefined();
});
});

View File

@ -0,0 +1,78 @@
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 {
EXCHANGE_REPO,
ROUTE_FETCH,
ROUTE_LIST_COMMITS,
} from '../repos/repos.constants';
import { AmqpConnection } from '@golevelup/nestjs-rabbitmq';
import { Commit } from '../repos/dtos/log-list.model';
import { getAppInstanceRouteKey } from '../commons/utils/rabbit-mq';
import { ApplicationException } from '../commons/exceptions/application.exception';
import { plainToClass } from 'class-transformer';
@Injectable()
export class PipelinesService extends BaseDbService<Pipeline> {
readonly uniqueFields: Array<Array<keyof Pipeline>> = [['projectId', 'name']];
constructor(
@InjectRepository(Pipeline)
readonly repository: Repository<Pipeline>,
private readonly amqpConnection: AmqpConnection,
) {
super(repository);
}
async list(dto: ListPipelineArgs) {
return this.repository.find(dto);
}
async findOneWithProject(id: string) {
return await this.repository.findOne({
where: { id },
relations: ['project'],
});
}
async create(dto: CreatePipelineInput) {
await this.isDuplicateEntity(dto);
return await this.repository.save(this.repository.create(dto));
}
async update(dto: UpdatePipelineInput) {
const old = await this.findOne(dto.id);
await this.isDuplicateEntityForUpdate(old, dto);
return await this.repository.save(this.repository.merge(old, dto));
}
async remove(id: string) {
return (await this.repository.softDelete({ id })).affected;
}
async syncCommits(pipeline: Pipeline, appInstance?: string) {
return await this.amqpConnection.request<string | null>({
exchange: EXCHANGE_REPO,
routingKey: getAppInstanceRouteKey(ROUTE_FETCH, appInstance),
payload: pipeline,
timeout: 120_000,
});
}
async listCommits(pipeline: Pipeline) {
return await this.amqpConnection
.request<[Error, Commit[]]>({
exchange: EXCHANGE_REPO,
routingKey: ROUTE_LIST_COMMITS,
payload: pipeline,
timeout: 30_000,
})
.then(([error, list]) => {
if (error) {
throw new ApplicationException(error);
}
return plainToClass(Commit, list);
});
}
}

View File

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

View File

@ -1,5 +1,9 @@
import { InputType } from '@nestjs/graphql';
import { IsUUID } from 'class-validator';
import { CreateProjectInput } from './create-project.input';
@InputType()
export class UpdateProjectInput extends CreateProjectInput {}
export class UpdateProjectInput extends CreateProjectInput {
@IsUUID()
id: string;
}

View File

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

View File

@ -0,0 +1,3 @@
export const EXCHANGE_PROJECT_TOPIC = 'project.topic';
export const EXCHANGE_PROJECT_FANOUT = 'project.fanout';
export const ROUTE_PROJECT_CHANGE = 'project-change';

View File

@ -3,9 +3,33 @@ import { ProjectsService } from './projects.service';
import { ProjectsResolver } from './projects.resolver';
import { TypeOrmModule } from '@nestjs/typeorm';
import { Project } from './project.entity';
import { RabbitMQModule } from '@golevelup/nestjs-rabbitmq';
import { ConfigModule, ConfigService } from '@nestjs/config';
import { EXCHANGE_PROJECT_FANOUT } from './projects.constants';
import { CommonsModule } from '../commons/commons.module';
@Module({
imports: [TypeOrmModule.forFeature([Project])],
imports: [
CommonsModule,
TypeOrmModule.forFeature([Project]),
RabbitMQModule.forRootAsync(RabbitMQModule, {
imports: [ConfigModule],
useFactory: (configService: ConfigService) => ({
uri: configService.get<string>('db.rabbitmq.uri'),
exchanges: [
{
name: EXCHANGE_PROJECT_FANOUT,
type: 'fanout',
options: {
durable: false,
autoDelete: true,
},
},
],
}),
inject: [ConfigService],
}),
],
providers: [ProjectsService, ProjectsResolver],
exports: [ProjectsService],
})

View File

@ -1,3 +1,4 @@
import { JwtService } from '@nestjs-lib/auth';
import { Test, TestingModule } from '@nestjs/testing';
import { ProjectsResolver } from './projects.resolver';
import { ProjectsService } from './projects.service';
@ -13,6 +14,10 @@ describe('ProjectsResolver', () => {
provide: ProjectsService,
useValue: {},
},
{
provide: JwtService,
useValue: {},
},
],
}).compile();

View File

@ -1,43 +1,44 @@
import { AccountRole, Roles } from '@nestjs-lib/auth';
import { Args, Mutation, Query, Resolver } from '@nestjs/graphql';
import { CreateProjectInput } from './dtos/create-project.input';
import { UpdateProjectInput } from './dtos/update-project.input';
import { Project } from './project.entity';
import { ProjectsService } from './projects.service';
@Roles(AccountRole.admin, AccountRole.super)
@Resolver(() => Project)
export class ProjectsResolver {
constructor(private readonly service: ProjectsService) {}
@Query(() => [Project])
async findProjects() {
async projects() {
return await this.service.list();
}
@Query(() => Project)
async findProject(@Args('id', { type: () => String }) id: string) {
async project(@Args('id', { type: () => String }) id: string) {
return await this.service.findOne(id);
}
@Mutation(() => Project)
async createProject(
@Args('project', { type: () => CreateProjectInput })
dto: UpdateProjectInput,
dto: CreateProjectInput,
) {
return await this.service.create(dto);
}
@Mutation(() => Project)
async modifyProject(
@Args('id', { type: () => String }) id: string,
async updateProject(
@Args('project', { type: () => UpdateProjectInput })
dto: UpdateProjectInput,
) {
const tmp = await this.service.update(id, dto);
const tmp = await this.service.update(dto);
console.log(tmp);
return tmp;
}
@Mutation(() => Number)
async deleteProject(@Args('id', { type: () => String }) id: string) {
async removeProject(@Args('id', { type: () => String }) id: string) {
return await this.service.remove(id);
}
}

View File

@ -2,6 +2,7 @@ import { Test, TestingModule } from '@nestjs/testing';
import { ProjectsService } from './projects.service';
import { getRepositoryToken } from '@nestjs/typeorm';
import { Project } from './project.entity';
import { AmqpConnection } from '@golevelup/nestjs-rabbitmq';
describe('ProjectsService', () => {
let service: ProjectsService;
@ -14,6 +15,10 @@ describe('ProjectsService', () => {
provide: getRepositoryToken(Project),
useValue: {},
},
{
provide: AmqpConnection,
useValue: {},
},
],
}).compile();

View File

@ -4,6 +4,12 @@ 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';
import { AmqpConnection } from '@golevelup/nestjs-rabbitmq';
import {
EXCHANGE_PROJECT_FANOUT,
ROUTE_PROJECT_CHANGE,
} from './projects.constants';
@Injectable()
export class ProjectsService extends BaseDbService<Project> {
@ -11,6 +17,7 @@ export class ProjectsService extends BaseDbService<Project> {
constructor(
@InjectRepository(Project)
readonly repository: Repository<Project>,
private readonly amqpConnection: AmqpConnection,
) {
super(repository);
}
@ -24,10 +31,15 @@ export class ProjectsService extends BaseDbService<Project> {
return await this.repository.save(this.repository.create(dto));
}
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));
async update(dto: UpdateProjectInput) {
await this.isDuplicateEntityForUpdate(dto.id, dto);
const old = await this.findOne(dto.id);
const project = await this.repository.save(this.repository.merge(old, dto));
this.amqpConnection.publish(EXCHANGE_PROJECT_FANOUT, ROUTE_PROJECT_CHANGE, [
project,
old,
]);
return project;
}
async remove(id: string) {

View File

@ -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;
}

View File

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

View File

@ -1,6 +1,20 @@
import { ObjectType, Field } from '@nestjs/graphql';
import { Type } from 'class-transformer';
import { LogResult, DefaultLogFields } from 'simple-git';
import { PipelineTask } from '../../pipeline-tasks/pipeline-task.entity';
@ObjectType()
export class Commit {
hash: string;
@Type(() => Date)
date: Date;
message: string;
refs: string;
body: string;
author_name: string;
author_email: string;
tasks: PipelineTask[];
}
@ObjectType()
export class LogFields {
hash: string;
@ -10,6 +24,7 @@ export class LogFields {
body: string;
author_name: string;
author_email: string;
tasks: PipelineTask[];
}
@ObjectType()

View File

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

View File

@ -0,0 +1,6 @@
export const EXCHANGE_REPO = 'fennec.repo';
export const ROUTE_FETCH = 'fetch';
export const ROUTE_LIST_COMMITS = 'list-commits';
export const QUEUE_LIST_COMMITS = 'list-commits';
export const QUEUE_FETCH = 'repo-fetch';
export const QUEUE_REFRESH_REPO = 'refresh-repo';

View File

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

View File

@ -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();

View File

@ -1,26 +1,4 @@
import { Args, Query, Resolver } from '@nestjs/graphql';
import { ListLogsArgs } from './dtos/list-logs.args';
import { ReposService } from './repos.service';
import { LogList } from './dtos/log-list.model';
import { ListBranchesArgs } from './dtos/list-branches.args';
import { BranchList } from './dtos/branch-list.model';
import { Resolver } from '@nestjs/graphql';
@Resolver()
export class ReposResolver {
constructor(private readonly service: ReposService) {}
@Query(() => LogList)
async listLogs(@Args('listLogsArgs') dto: ListLogsArgs) {
return await this.service.listLogs(dto);
}
@Query(() => BranchList)
async ListBranchesArgs(
@Args('listBranchesArgs') dto: ListBranchesArgs,
): Promise<BranchList> {
return await this.service.listBranches(dto).then((data) => {
return {
...data,
branches: Object.values(data.branches),
};
});
}
}
export class ReposResolver {}

View File

@ -1,12 +1,18 @@
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';
import { getLoggerToken, PinoLogger } from 'nestjs-pino';
import { Nack } from '@golevelup/nestjs-rabbitmq';
import { getInstanceName } from '../commons/utils/rabbit-mq';
import { RedisMutexService } from '../commons/redis-mutex/redis-mutex.service';
const getTest1Project = () =>
({
@ -43,6 +49,18 @@ describe('ReposService', () => {
provide: getRepositoryToken(Project),
useFactory: repositoryMockFactory,
},
{
provide: getLoggerToken(ReposService.name),
useValue: new PinoLogger({}),
},
{
provide: RedisMutexService,
useValue: {
lock: jest.fn(() =>
Promise.resolve(() => Promise.resolve(undefined)),
),
},
},
],
}).compile();
@ -61,7 +79,10 @@ describe('ReposService', () => {
});
describe('listLogs', () => {
it('should be return logs', async () => {
const result = await service.listLogs({ projectId: '1' });
const result = await service.listLogs({
project: getTest1Project(),
branch: 'master',
});
expect(result).toBeDefined();
}, 20_000);
});
@ -69,63 +90,122 @@ describe('ReposService', () => {
it('should be return branches', async () => {
const result = await service.listBranches({ projectId: '1' });
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);
}, 20_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$/,
);
});
});
describe('fetch', () => {
it('success', async () => {
const project = new Project();
const pipeline = new Pipeline();
pipeline.branch = 'test';
const fetch = jest.fn((_: any) => Promise.resolve());
pipeline.project = project;
const getGit = jest.spyOn(service, 'getGit').mockImplementation(() =>
Promise.resolve({
fetch,
} as any),
);
await expect(service.fetch(pipeline)).resolves.toEqual(getInstanceName());
expect(getGit).toBeCalledTimes(1);
expect(getGit.mock.calls[0]?.[0]).toEqual(project);
expect(fetch).toBeCalledTimes(1);
expect(fetch.mock.calls[0]?.[0]).toMatchObject([
'origin',
'test',
'--depth=100',
]);
});
it('failed a', async () => {
const project = new Project();
const pipeline = new Pipeline();
pipeline.branch = 'test';
const fetch = jest.fn((_: any) => Promise.resolve());
pipeline.project = project;
const getGit = jest
.spyOn(service, 'getGit')
.mockImplementation(() => Promise.reject('error'));
await expect(service.fetch(pipeline)).resolves.toMatchObject(new Nack());
expect(getGit).toBeCalledTimes(1);
expect(getGit.mock.calls[0]?.[0]).toEqual(project);
expect(fetch).toBeCalledTimes(0);
});
it('failed b', async () => {
const project = new Project();
const pipeline = new Pipeline();
pipeline.branch = 'test';
const fetch = jest.fn((_: any) => Promise.reject('error'));
pipeline.project = project;
const getGit = jest.spyOn(service, 'getGit').mockImplementation(() =>
Promise.resolve({
fetch,
} as any),
);
await expect(service.fetch(pipeline)).resolves.toMatchObject(new Nack());
expect(getGit).toBeCalledTimes(1);
expect(fetch).toBeCalledTimes(1);
});
});
});

View File

@ -1,3 +1,5 @@
import { ListLogsOption } from './models/list-logs.options';
import { PipelineTask } from './../pipeline-tasks/pipeline-task.entity';
import { Injectable, NotFoundException } from '@nestjs/common';
import { InjectRepository } from '@nestjs/typeorm';
import { F_OK } from 'constants';
@ -7,57 +9,89 @@ import { gitP } from 'simple-git';
import { Repository } from 'typeorm';
import { Project } from '../projects/project.entity';
import { ListBranchesArgs } from './dtos/list-branches.args';
import { ListLogsArgs } from './dtos/list-logs.args';
import { ConfigService } from '@nestjs/config';
import { Commit } from './dtos/log-list.model';
import { Nack, RabbitRPC, RabbitSubscribe } from '@golevelup/nestjs-rabbitmq';
import { Pipeline } from '../pipelines/pipeline.entity';
import { InjectPinoLogger, PinoLogger } from 'nestjs-pino';
import {
EXCHANGE_REPO,
QUEUE_FETCH,
QUEUE_LIST_COMMITS,
QUEUE_REFRESH_REPO,
ROUTE_FETCH,
ROUTE_LIST_COMMITS,
} from './repos.constants';
import { getSelfInstanceQueueKey } from '../commons/utils/rabbit-mq';
import {
getInstanceName,
getSelfInstanceRouteKey,
} from '../commons/utils/rabbit-mq';
import { ApplicationException } from '../commons/exceptions/application.exception';
import {
EXCHANGE_PROJECT_FANOUT,
ROUTE_PROJECT_CHANGE,
} from '../projects/projects.constants';
import { RedisMutexService } from '../commons/redis-mutex/redis-mutex.service';
import { rm } from 'fs/promises';
const DEFAULT_REMOTE_NAME = 'origin';
const INFO_PATH = '@info';
@Injectable()
export class ReposService {
constructor(
@InjectRepository(Project)
private readonly projectRepository: Repository<Project>,
private readonly configService: ConfigService,
@InjectPinoLogger(ReposService.name)
private readonly logger: PinoLogger,
private readonly redisMutexService: RedisMutexService,
) {}
getWorkspaceRoot(project: Project): string {
return join(
this.configService.get<string>('workspaces.root'),
project.name,
encodeURIComponent(project.name),
INFO_PATH,
);
}
async lockWorkspace(workspaceRoot: string) {
// TODO: 获取锁,失败抛错。
getDeployRoot(task: PipelineTask) {
return join(
this.configService.get<string>('workspaces.root'),
encodeURIComponent(task.pipeline.project.name),
encodeURIComponent(`deploy-${task.pipeline.name}`),
);
}
async getGit(project: Project) {
const workspaceRoot = this.getWorkspaceRoot(project);
await this.lockWorkspace(workspaceRoot);
async getGit(
project: Project,
workspaceRoot?: string,
{ fetch = true } = {},
) {
if (!workspaceRoot) {
workspaceRoot = this.getWorkspaceRoot(project);
}
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().catch(() => false))) {
await git.init();
await git.addRemote(DEFAULT_REMOTE_NAME, project.sshUrl);
}
if (fetch) {
await git.fetch();
}
return git;
}
async listLogs(dto: ListLogsArgs) {
const project = await this.projectRepository.findOneOrFail({
id: dto.projectId,
});
async listLogs({ project, branch }: ListLogsOption) {
const git = await this.getGit(project);
await git.fetch();
return await git.log({
'--branches': dto.branch ?? '',
'--remotes': DEFAULT_REMOTE_NAME,
});
return await git.log(
branch ? ['--branches', `remotes/origin/${branch}`, '--'] : ['--all'],
);
}
async listBranches(dto: ListBranchesArgs) {
@ -68,26 +102,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 +112,129 @@ 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}`),
);
}
async checkout4Task(task: PipelineTask): Promise<string> {
const path = this.getWorkspaceRootByTask(task);
await this.checkout(task, path);
return path;
}
@RabbitRPC({
exchange: EXCHANGE_REPO,
routingKey: [
ROUTE_LIST_COMMITS,
getSelfInstanceRouteKey(ROUTE_LIST_COMMITS),
],
queue: getSelfInstanceQueueKey(QUEUE_LIST_COMMITS),
queueOptions: {
autoDelete: true,
},
})
async listCommits(pipeline: Pipeline): Promise<[Error, Commit[]?]> {
const git = await this.getGit(pipeline.project, undefined, {
fetch: false,
});
try {
const data = await git.log([
'-100',
'--branches',
`remotes/origin/${pipeline.branch}`,
'--',
]);
return [
null,
data.all.map(
(it) =>
({
...it,
date: new Date(it.date),
} as Commit),
),
];
} catch (error) {
this.logger.error(
{ error, pipeline },
'[listCommits] %s',
error?.message,
);
return [new ApplicationException(error)];
}
}
@RabbitRPC({
exchange: EXCHANGE_REPO,
routingKey: [ROUTE_FETCH, getSelfInstanceRouteKey(ROUTE_FETCH)],
queue: getSelfInstanceQueueKey(QUEUE_FETCH),
queueOptions: {
autoDelete: true,
},
})
async fetch(pipeline: Pipeline): Promise<string | null | Nack> {
const unlock = await this.redisMutexService.lock(
`repo-project-${pipeline.projectId}`,
);
try {
const git = await this.getGit(pipeline.project, undefined, {
fetch: false,
});
await git.fetch(['origin', pipeline.branch, '--depth=100']);
return getInstanceName();
} catch (error) {
this.logger.error({ error, pipeline }, '[fetch] %s', error?.message);
return new Nack();
} finally {
await unlock();
}
}
@RabbitSubscribe({
exchange: EXCHANGE_PROJECT_FANOUT,
routingKey: ROUTE_PROJECT_CHANGE,
queue: QUEUE_REFRESH_REPO,
queueOptions: {
autoDelete: true,
durable: true,
},
})
async refreshRepo([project]: [Project]) {
this.logger.info({ project }, '[refreshRepo] start');
const unlock = await this.redisMutexService.lock(
`repo-project-${project.id}`,
{
timeout: null,
},
);
try {
const path = join(
this.configService.get<string>('workspaces.root'),
encodeURIComponent(project.name),
);
await rm(path, { recursive: true });
this.logger.info({ project }, '[refreshRepo] success');
} catch (error) {
this.logger.error(
{ project, error },
'[refreshRepo] failed. $s',
error.message,
);
} finally {
await unlock();
}
}
}

View File

@ -0,0 +1,8 @@
import { IsString } from 'class-validator';
export class GiteaHookPayloadDto {
@IsString()
ref: string;
@IsString()
after: string;
}

View File

@ -0,0 +1,3 @@
export enum SourceService {
gitea = 'gitea',
}

View File

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

View File

@ -0,0 +1,37 @@
import { Body, Controller, Headers, Param, Post } from '@nestjs/common';
import { validateOrReject } from 'class-validator';
import { pick } from 'ramda';
import { GiteaHookPayloadDto } from './dtos/gitea-hook-payload.dto';
import { SourceService } from './enums/source-service.enum';
import { WebhookLog } from './webhook-log.entity';
import { WebhooksService } from './webhooks.service';
@Controller('gitea-webhooks')
export class GiteaWebhooksController {
constructor(private readonly service: WebhooksService) {}
@Post(':pipelineId')
async onCall(
@Headers('X-Gitea-Delivery') delivery: string,
@Headers('X-Gitea-Event') event: string,
@Headers('X-Gitea-Signature') signature: string,
@Body() body: Buffer,
@Param('pipelineId') pipelineId: string,
) {
const payload = Object.assign(
new GiteaHookPayloadDto(),
JSON.parse(body.toString('utf-8')),
);
await validateOrReject(payload);
await this.service.verifySignature(body, signature, 'boardcat');
return await this.service
.onCall(pipelineId, {
payload,
sourceDelivery: delivery,
sourceEvent: event,
sourceService: SourceService.gitea,
})
.then((data) =>
pick<keyof WebhookLog>(['id', 'createdAt', 'localEvent'])(data),
);
}
}

View File

@ -0,0 +1,9 @@
import { SourceService } from '../enums/source-service.enum';
import { WebhookLog } from '../webhook-log.entity';
export class CreateWebhookLogModel<T> implements Partial<WebhookLog> {
sourceDelivery: string;
sourceEvent: string;
sourceService: SourceService;
payload: T;
}

View File

@ -0,0 +1,19 @@
import { Column, Entity } from 'typeorm';
import { AppBaseEntity } from './../commons/entities/app-base-entity';
import { SourceService } from './enums/source-service.enum';
@Entity()
export class WebhookLog extends AppBaseEntity {
@Column()
sourceDelivery: string;
@Column({ type: 'enum', enum: SourceService })
sourceService: SourceService;
@Column()
sourceEvent: string;
@Column({ type: 'jsonb' })
payload: any;
@Column()
localEvent: string;
@Column({ type: 'jsonb' })
localPayload: any;
}

View File

@ -0,0 +1,14 @@
import { MiddlewareConsumer, Module } from '@nestjs/common';
import { TypeOrmModule } from '@nestjs/typeorm';
import { PipelineTasksModule } from '../pipeline-tasks/pipeline-tasks.module';
import { GiteaWebhooksController } from './gitea-webhooks.controller';
import { WebhookLog } from './webhook-log.entity';
import { WebhooksService } from './webhooks.service';
@Module({
imports: [TypeOrmModule.forFeature([WebhookLog]), PipelineTasksModule],
controllers: [GiteaWebhooksController],
providers: [WebhooksService],
})
export class WebhooksModule {
}

View File

@ -0,0 +1,57 @@
import { UnauthorizedException } from '@nestjs/common';
import { Test, TestingModule } from '@nestjs/testing';
import { getRepositoryToken } from '@nestjs/typeorm';
import { readFile } from 'fs/promises';
import { join } from 'path';
import { Repository } from 'typeorm';
import { PipelineTasksService } from '../pipeline-tasks/pipeline-tasks.service';
import { WebhookLog } from './webhook-log.entity';
import { WebhooksService } from './webhooks.service';
describe('WebhooksService', () => {
let service: WebhooksService;
beforeEach(async () => {
const module: TestingModule = await Test.createTestingModule({
providers: [
WebhooksService,
{
provide: PipelineTasksService,
useValue: {},
},
{
provide: getRepositoryToken(WebhookLog),
useValue: new Repository(),
},
],
}).compile();
service = module.get<WebhooksService>(WebhooksService);
});
it('should be defined', () => {
expect(service).toBeDefined();
});
describe('verifySignature', () => {
const signature =
'b175e07189a6106f386b62253b18b5879c4b1f3af2f11fe13a294602671e361a';
const secret = 'boardcat';
let payload: Buffer;
beforeAll(async () => {
payload = await readFile(
join(__dirname, '../../test/data/gitea-hook-payload.json.bin'),
);
});
it('must be valid', async () => {
await expect(
service.verifySignature(payload, signature, secret),
).resolves.toEqual(undefined);
});
it('must be invalid', async () => {
await expect(
service.verifySignature(payload, 'test', secret),
).rejects.toThrowError(UnauthorizedException);
});
});
});

View File

@ -0,0 +1,65 @@
import {
BadRequestException,
Injectable,
UnauthorizedException,
} from '@nestjs/common';
import { InjectRepository } from '@nestjs/typeorm';
import { createHmac } from 'crypto';
import { Repository } from 'typeorm';
import { PipelineUnits } from '../pipeline-tasks/enums/pipeline-units.enum';
import { PipelineTasksService } from '../pipeline-tasks/pipeline-tasks.service';
import { GiteaHookPayloadDto } from './dtos/gitea-hook-payload.dto';
import { CreateWebhookLogModel } from './models/create-webhook-log.model';
import { WebhookLog } from './webhook-log.entity';
@Injectable()
export class WebhooksService {
constructor(
@InjectRepository(WebhookLog)
private readonly repository: Repository<WebhookLog>,
private readonly taskService: PipelineTasksService,
) {}
async onCall(
pipelineId: string,
model: CreateWebhookLogModel<GiteaHookPayloadDto>,
) {
if (model.sourceEvent.toLowerCase() === 'push') {
const taskDto = {
pipelineId,
commit: model.payload.after,
units: Object.values(PipelineUnits),
};
await this.taskService.addTask(taskDto);
return await this.repository.save(
this.repository.create({
...model,
localEvent: 'create-pipeline-task',
localPayload: taskDto,
}),
);
} else {
throw new BadRequestException('无法处理的请求');
}
}
async verifySignature(payload: any, signature: string, secret: string) {
const local = await new Promise<string>((resolve, reject) => {
const hmac = createHmac('sha256', secret);
hmac.on('readable', () => {
const data = hmac.read();
if (data) {
resolve(data.toString('hex'));
}
});
hmac.on('error', (err) => {
reject(err);
});
hmac.write(payload);
hmac.end();
});
if (local !== signature) {
throw new UnauthorizedException();
}
}
}

View File

@ -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,
},
],
};

View File

@ -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);
});

View File

@ -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"
}

View File

@ -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);
});

8
test/data/bad-work.js Normal file
View File

@ -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);

View File

@ -0,0 +1,115 @@
{
"secret": "boardcat",
"ref": "refs/heads/master",
"before": "429de1eaedf1da83f1e0e3ac3d8b20e771b7051c",
"after": "429de1eaedf1da83f1e0e3ac3d8b20e771b7051c",
"compare_url": "",
"commits": [
{
"id": "429de1eaedf1da83f1e0e3ac3d8b20e771b7051c",
"message": "test(pipeline-tasks): pass test cases.\n",
"url": "https://git.ivanli.cc/Fennec/fennec-be/commit/429de1eaedf1da83f1e0e3ac3d8b20e771b7051c",
"author": {
"name": "Ivan",
"email": "ivanli@live.cn",
"username": ""
},
"committer": {
"name": "Ivan",
"email": "ivanli@live.cn",
"username": ""
},
"verification": null,
"timestamp": "0001-01-01T00:00:00Z",
"added": null,
"removed": null,
"modified": null
}
],
"head_commit": null,
"repository": {
"id": 3,
"owner": {
"id": 3,
"login": "Fennec",
"full_name": "",
"email": "",
"avatar_url": "https://git.ivanli.cc/user/avatar/Fennec/-1",
"language": "",
"is_admin": false,
"last_login": "1970-01-01T08:00:00+08:00",
"created": "2021-01-30T16:46:11+08:00",
"username": "Fennec"
},
"name": "fennec-be",
"full_name": "Fennec/fennec-be",
"description": "Fennec CI/CD Back-End",
"empty": false,
"private": false,
"fork": false,
"template": false,
"parent": null,
"mirror": false,
"size": 1897,
"html_url": "https://git.ivanli.cc/Fennec/fennec-be",
"ssh_url": "ssh://gitea@git.ivanli.cc:7018/Fennec/fennec-be.git",
"clone_url": "https://git.ivanli.cc/Fennec/fennec-be.git",
"original_url": "",
"website": "",
"stars_count": 1,
"forks_count": 0,
"watchers_count": 1,
"open_issues_count": 0,
"open_pr_counter": 0,
"release_counter": 0,
"default_branch": "master",
"archived": false,
"created_at": "2021-01-31T09:58:38+08:00",
"updated_at": "2021-03-27T15:57:00+08:00",
"permissions": {
"admin": false,
"push": false,
"pull": false
},
"has_issues": true,
"internal_tracker": {
"enable_time_tracker": true,
"allow_only_contributors_to_track_time": true,
"enable_issue_dependencies": true
},
"has_wiki": true,
"has_pull_requests": true,
"has_projects": true,
"ignore_whitespace_conflicts": false,
"allow_merge_commits": true,
"allow_rebase": true,
"allow_rebase_explicit": true,
"allow_squash_merge": true,
"avatar_url": "",
"internal": false
},
"pusher": {
"id": 1,
"login": "Ivan",
"full_name": "Ivan Li",
"email": "ivan@noreply.%(DOMAIN)s",
"avatar_url": "https://git.ivanli.cc/user/avatar/Ivan/-1",
"language": "zh-CN",
"is_admin": true,
"last_login": "2021-03-26T22:28:05+08:00",
"created": "2021-01-23T18:15:30+08:00",
"username": "Ivan"
},
"sender": {
"id": 1,
"login": "Ivan",
"full_name": "Ivan Li",
"email": "ivan@noreply.%(DOMAIN)s",
"avatar_url": "https://git.ivanli.cc/user/avatar/Ivan/-1",
"language": "zh-CN",
"is_admin": true,
"last_login": "2021-03-26T22:28:05+08:00",
"created": "2021-01-23T18:15:30+08:00",
"username": "Ivan"
}
}

View File

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

View File

@ -7,9 +7,10 @@
"experimentalDecorators": true,
"allowSyntheticDefaultImports": true,
"target": "es2017",
"lib": ["ES2021"],
"sourceMap": true,
"outDir": "./dist",
"rootDir": "./src",
"baseUrl": "./",
"incremental": true
}
},
}