feat-pipelines #1
6
.vscode/settings.json
vendored
6
.vscode/settings.json
vendored
@ -1,5 +1,9 @@
|
|||||||
{
|
{
|
||||||
"cSpell.words": [
|
"cSpell.words": [
|
||||||
"Repos"
|
"Repos",
|
||||||
|
"lpush",
|
||||||
|
"lrange",
|
||||||
|
"rpop",
|
||||||
|
"rpush"
|
||||||
]
|
]
|
||||||
}
|
}
|
@ -9,5 +9,10 @@ db:
|
|||||||
database: fennec
|
database: fennec
|
||||||
username: fennec
|
username: fennec
|
||||||
password:
|
password:
|
||||||
|
redis:
|
||||||
|
host: 192.168.31.194
|
||||||
|
port: 6379
|
||||||
|
password:
|
||||||
|
prefix: fennec
|
||||||
workspaces:
|
workspaces:
|
||||||
root: '/Users/ivanli/Projects/fennec/workspaces'
|
root: '/Users/ivanli/Projects/fennec/workspaces'
|
33
docs/ci-cd.md
Normal file
33
docs/ci-cd.md
Normal 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"
|
||||||
|
]
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
```
|
15696
package-lock.json
generated
15696
package-lock.json
generated
File diff suppressed because it is too large
Load Diff
@ -40,6 +40,7 @@
|
|||||||
"graphql-tools": "^7.0.2",
|
"graphql-tools": "^7.0.2",
|
||||||
"js-yaml": "^4.0.0",
|
"js-yaml": "^4.0.0",
|
||||||
"nestjs-redis": "^1.2.8",
|
"nestjs-redis": "^1.2.8",
|
||||||
|
"observable-to-async-generator": "^1.0.1-rc",
|
||||||
"pg": "^8.5.1",
|
"pg": "^8.5.1",
|
||||||
"ramda": "^0.27.1",
|
"ramda": "^0.27.1",
|
||||||
"reflect-metadata": "^0.1.13",
|
"reflect-metadata": "^0.1.13",
|
||||||
|
@ -7,7 +7,10 @@ import { AppResolver } from './app.resolver';
|
|||||||
import { AppService } from './app.service';
|
import { AppService } from './app.service';
|
||||||
import { ProjectsModule } from './projects/projects.module';
|
import { ProjectsModule } from './projects/projects.module';
|
||||||
import { ReposModule } from './repos/repos.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 configuration from './commons/config/configuration';
|
||||||
|
import { RedisModule } from 'nestjs-redis';
|
||||||
|
|
||||||
@Module({
|
@Module({
|
||||||
imports: [
|
imports: [
|
||||||
@ -34,11 +37,24 @@ import configuration from './commons/config/configuration';
|
|||||||
debug: configService.get<string>('env') !== 'prod',
|
debug: configService.get<string>('env') !== 'prod',
|
||||||
playground: true,
|
playground: true,
|
||||||
autoSchemaFile: true,
|
autoSchemaFile: true,
|
||||||
|
installSubscriptionHandlers: true,
|
||||||
}),
|
}),
|
||||||
inject: [ConfigService],
|
inject: [ConfigService],
|
||||||
}),
|
}),
|
||||||
ProjectsModule,
|
ProjectsModule,
|
||||||
ReposModule,
|
ReposModule,
|
||||||
|
PipelinesModule,
|
||||||
|
PipelineTasksModule,
|
||||||
|
RedisModule.forRootAsync({
|
||||||
|
imports: [ConfigModule],
|
||||||
|
useFactory: (configService: ConfigService) => ({
|
||||||
|
host: configService.get<string>('db.redis.host', 'localhost'),
|
||||||
|
port: configService.get<number>('db.redis.port', 6379),
|
||||||
|
password: configService.get<string>('db.redis.password', ''),
|
||||||
|
keyPrefix: configService.get<string>('db.redis.prefix', 'fennec'),
|
||||||
|
}),
|
||||||
|
inject: [ConfigService],
|
||||||
|
}),
|
||||||
],
|
],
|
||||||
controllers: [AppController],
|
controllers: [AppController],
|
||||||
providers: [AppService, AppResolver],
|
providers: [AppService, AppResolver],
|
||||||
|
@ -1,6 +1,7 @@
|
|||||||
import { Field, ID, ObjectType } from '@nestjs/graphql';
|
import { Field, ID, ObjectType } from '@nestjs/graphql';
|
||||||
import {
|
import {
|
||||||
CreateDateColumn,
|
CreateDateColumn,
|
||||||
|
DeleteDateColumn,
|
||||||
PrimaryGeneratedColumn,
|
PrimaryGeneratedColumn,
|
||||||
UpdateDateColumn,
|
UpdateDateColumn,
|
||||||
} from 'typeorm';
|
} from 'typeorm';
|
||||||
@ -16,4 +17,7 @@ export class AppBaseEntity {
|
|||||||
|
|
||||||
@UpdateDateColumn({ select: false })
|
@UpdateDateColumn({ select: false })
|
||||||
updatedAt: Date;
|
updatedAt: Date;
|
||||||
|
|
||||||
|
@DeleteDateColumn({ select: false })
|
||||||
|
deletedAt?: Date;
|
||||||
}
|
}
|
||||||
|
@ -1,55 +1,27 @@
|
|||||||
import {
|
import {
|
||||||
ArgumentsHost,
|
|
||||||
Catch,
|
|
||||||
ExceptionFilter,
|
ExceptionFilter,
|
||||||
|
Catch,
|
||||||
|
ArgumentsHost,
|
||||||
HttpException,
|
HttpException,
|
||||||
HttpStatus,
|
|
||||||
} from '@nestjs/common';
|
} from '@nestjs/common';
|
||||||
import { EntityNotFoundError } from 'typeorm/error/EntityNotFoundError';
|
import { ApolloError } from 'apollo-server-errors';
|
||||||
|
|
||||||
@Catch()
|
@Catch(HttpException)
|
||||||
export class AllExceptionsFilter implements ExceptionFilter {
|
export class HttpExceptionFilter implements ExceptionFilter {
|
||||||
catch(exception: any, host: ArgumentsHost) {
|
catch(exception: HttpException, host: ArgumentsHost) {
|
||||||
const ctx = host.switchToHttp();
|
const message = exception.message;
|
||||||
const response = ctx.getResponse();
|
const extensions: Record<string, any> = {};
|
||||||
const request = ctx.getRequest();
|
const err = exception.getResponse();
|
||||||
|
if (typeof err === 'string') {
|
||||||
const status =
|
extensions.message = err;
|
||||||
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 {
|
} else {
|
||||||
response.status(status).json({
|
Object.assign(extensions, (err as any).extension);
|
||||||
message: ex,
|
extensions.message = (err as any).message;
|
||||||
timestamp: Date.now(),
|
|
||||||
path: request.url,
|
|
||||||
});
|
|
||||||
}
|
|
||||||
} else if (exception instanceof EntityNotFoundError) {
|
|
||||||
response.status(HttpStatus.NOT_FOUND).json({
|
|
||||||
message: '资源未找到!',
|
|
||||||
timestamp: Date.now(),
|
|
||||||
path: request.url,
|
|
||||||
});
|
|
||||||
} else {
|
|
||||||
console.error('服务器内部错误');
|
|
||||||
console.error(exception);
|
|
||||||
response.status(HttpStatus.INTERNAL_SERVER_ERROR).json({
|
|
||||||
code: status,
|
|
||||||
timestamp: new Date().toISOString(),
|
|
||||||
message: '服务器内部错误',
|
|
||||||
error: exception,
|
|
||||||
path: request.url,
|
|
||||||
});
|
|
||||||
}
|
}
|
||||||
|
return new ApolloError(
|
||||||
|
message,
|
||||||
|
exception.getStatus().toString(),
|
||||||
|
extensions,
|
||||||
|
);
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
15
src/commons/pipes/sanitize.pipe.ts
Normal file
15
src/commons/pipes/sanitize.pipe.ts
Normal file
@ -0,0 +1,15 @@
|
|||||||
|
import { ArgumentMetadata, Injectable, PipeTransform } from '@nestjs/common';
|
||||||
|
import { sanitize } from '@neuralegion/class-sanitizer/dist';
|
||||||
|
|
||||||
|
@Injectable()
|
||||||
|
export class SanitizePipe implements PipeTransform {
|
||||||
|
transform(value: any, metadata: ArgumentMetadata) {
|
||||||
|
// console.log(value, typeof value);
|
||||||
|
if (value instanceof Object) {
|
||||||
|
value = Object.assign(new metadata.metatype(), value);
|
||||||
|
sanitize(value);
|
||||||
|
// console.log(value);
|
||||||
|
}
|
||||||
|
return value;
|
||||||
|
}
|
||||||
|
}
|
10
src/main.ts
10
src/main.ts
@ -2,11 +2,19 @@ import { ValidationPipe } from '@nestjs/common';
|
|||||||
import { ConfigService } from '@nestjs/config';
|
import { ConfigService } from '@nestjs/config';
|
||||||
import { NestFactory } from '@nestjs/core';
|
import { NestFactory } from '@nestjs/core';
|
||||||
import { AppModule } from './app.module';
|
import { AppModule } from './app.module';
|
||||||
|
import { HttpExceptionFilter } from './commons/filters/all.exception-filter';
|
||||||
|
import { SanitizePipe } from './commons/pipes/sanitize.pipe';
|
||||||
|
|
||||||
async function bootstrap() {
|
async function bootstrap() {
|
||||||
const app = await NestFactory.create(AppModule);
|
const app = await NestFactory.create(AppModule);
|
||||||
const configService = app.get(ConfigService);
|
const configService = app.get(ConfigService);
|
||||||
app.useGlobalPipes(new ValidationPipe());
|
app.useGlobalPipes(new SanitizePipe());
|
||||||
|
app.useGlobalPipes(
|
||||||
|
new ValidationPipe({
|
||||||
|
transform: true,
|
||||||
|
}),
|
||||||
|
);
|
||||||
|
app.useGlobalFilters(new HttpExceptionFilter());
|
||||||
await app.listen(configService.get<number>('http.port'));
|
await app.listen(configService.get<number>('http.port'));
|
||||||
}
|
}
|
||||||
bootstrap();
|
bootstrap();
|
||||||
|
11
src/pipeline-tasks/dtos/create-pipeline-task.input.ts
Normal file
11
src/pipeline-tasks/dtos/create-pipeline-task.input.ts
Normal 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[];
|
||||||
|
}
|
8
src/pipeline-tasks/dtos/pipeline-task-log.args.ts
Normal file
8
src/pipeline-tasks/dtos/pipeline-task-log.args.ts
Normal file
@ -0,0 +1,8 @@
|
|||||||
|
import { ArgsType } from '@nestjs/graphql';
|
||||||
|
import { IsUUID } from 'class-validator';
|
||||||
|
|
||||||
|
@ArgsType()
|
||||||
|
export class PipelineTaskLogArgs {
|
||||||
|
@IsUUID()
|
||||||
|
taskId: string;
|
||||||
|
}
|
14
src/pipeline-tasks/enums/pipeline-units.enum.ts
Normal file
14
src/pipeline-tasks/enums/pipeline-units.enum.ts
Normal 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: '流水线单元',
|
||||||
|
});
|
13
src/pipeline-tasks/enums/task-statuses.enum.ts
Normal file
13
src/pipeline-tasks/enums/task-statuses.enum.ts
Normal file
@ -0,0 +1,13 @@
|
|||||||
|
import { registerEnumType } from '@nestjs/graphql';
|
||||||
|
|
||||||
|
export enum TaskStatuses {
|
||||||
|
success = 'success',
|
||||||
|
failed = 'failed',
|
||||||
|
working = 'working',
|
||||||
|
pending = 'pending',
|
||||||
|
}
|
||||||
|
|
||||||
|
registerEnumType(TaskStatuses, {
|
||||||
|
name: 'TaskStatuses',
|
||||||
|
description: '任务状态',
|
||||||
|
});
|
@ -0,0 +1,32 @@
|
|||||||
|
import { PipelineTask } from './../pipeline-task.entity';
|
||||||
|
import { PipelineUnits } from '../enums/pipeline-units.enum';
|
||||||
|
import { Field, HideField, ObjectType } from '@nestjs/graphql';
|
||||||
|
|
||||||
|
@ObjectType()
|
||||||
|
export class PipelineTaskLogMessage {
|
||||||
|
@HideField()
|
||||||
|
task: PipelineTask;
|
||||||
|
@Field(() => PipelineUnits, { nullable: true })
|
||||||
|
unit?: PipelineUnits;
|
||||||
|
@Field()
|
||||||
|
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,
|
||||||
|
});
|
||||||
|
}
|
||||||
|
}
|
14
src/pipeline-tasks/models/pipeline-task-logs.model.ts
Normal file
14
src/pipeline-tasks/models/pipeline-task-logs.model.ts
Normal file
@ -0,0 +1,14 @@
|
|||||||
|
import { TaskStatuses } from '../enums/task-statuses.enum';
|
||||||
|
import { PipelineUnits } from '../enums/pipeline-units.enum';
|
||||||
|
import { Field, ObjectType } from '@nestjs/graphql';
|
||||||
|
|
||||||
|
@ObjectType()
|
||||||
|
export class PipelineTaskLogs {
|
||||||
|
@Field(() => PipelineUnits)
|
||||||
|
unit: PipelineUnits;
|
||||||
|
@Field(() => TaskStatuses)
|
||||||
|
status: TaskStatuses;
|
||||||
|
startedAt?: Date;
|
||||||
|
endedAt?: Date;
|
||||||
|
logs = '';
|
||||||
|
}
|
9
src/pipeline-tasks/models/work-unit-metadata.model.ts
Normal file
9
src/pipeline-tasks/models/work-unit-metadata.model.ts
Normal file
@ -0,0 +1,9 @@
|
|||||||
|
import { InputType, ObjectType } from '@nestjs/graphql';
|
||||||
|
import { WorkUnit } from './work-unit.model';
|
||||||
|
|
||||||
|
@InputType('WorkUnitMetadataInput')
|
||||||
|
@ObjectType()
|
||||||
|
export class WorkUnitMetadata {
|
||||||
|
version = 1;
|
||||||
|
units: WorkUnit[];
|
||||||
|
}
|
13
src/pipeline-tasks/models/work-unit.model.ts
Normal file
13
src/pipeline-tasks/models/work-unit.model.ts
Normal file
@ -0,0 +1,13 @@
|
|||||||
|
import { Field, InputType, ObjectType } from '@nestjs/graphql';
|
||||||
|
import {
|
||||||
|
PipelineUnits,
|
||||||
|
PipelineUnits as PipelineUnitTypes,
|
||||||
|
} from '../enums/pipeline-units.enum';
|
||||||
|
|
||||||
|
@ObjectType()
|
||||||
|
@InputType('WorkUnitInput')
|
||||||
|
export class WorkUnit {
|
||||||
|
@Field(() => PipelineUnits)
|
||||||
|
type: PipelineUnitTypes;
|
||||||
|
scripts: string[];
|
||||||
|
}
|
25
src/pipeline-tasks/pipeline-task-logs.service.spec.ts
Normal file
25
src/pipeline-tasks/pipeline-task-logs.service.spec.ts
Normal file
@ -0,0 +1,25 @@
|
|||||||
|
import { Test, TestingModule } from '@nestjs/testing';
|
||||||
|
import { PipelineTaskLogsService } from './pipeline-task-logs.service';
|
||||||
|
import { RedisService } from 'nestjs-redis';
|
||||||
|
|
||||||
|
describe('PipelineTaskLogsService', () => {
|
||||||
|
let service: PipelineTaskLogsService;
|
||||||
|
|
||||||
|
beforeEach(async () => {
|
||||||
|
const module: TestingModule = await Test.createTestingModule({
|
||||||
|
providers: [
|
||||||
|
PipelineTaskLogsService,
|
||||||
|
{
|
||||||
|
provide: RedisService,
|
||||||
|
useValue: {},
|
||||||
|
},
|
||||||
|
],
|
||||||
|
}).compile();
|
||||||
|
|
||||||
|
service = module.get<PipelineTaskLogsService>(PipelineTaskLogsService);
|
||||||
|
});
|
||||||
|
|
||||||
|
it('should be defined', () => {
|
||||||
|
expect(service).toBeDefined();
|
||||||
|
});
|
||||||
|
});
|
78
src/pipeline-tasks/pipeline-task-logs.service.ts
Normal file
78
src/pipeline-tasks/pipeline-task-logs.service.ts
Normal file
@ -0,0 +1,78 @@
|
|||||||
|
import { Injectable } from '@nestjs/common';
|
||||||
|
import { log } from 'console';
|
||||||
|
import { PubSub } from 'graphql-subscriptions';
|
||||||
|
import { RedisService } from 'nestjs-redis';
|
||||||
|
import { find, omit, propEq } from 'ramda';
|
||||||
|
import { PipelineUnits } from './enums/pipeline-units.enum';
|
||||||
|
import { TaskStatuses } from './enums/task-statuses.enum';
|
||||||
|
import { PipelineTaskLogMessage } from './models/pipeline-task-log-message.module';
|
||||||
|
import { PipelineTaskLogs } from './models/pipeline-task-logs.model';
|
||||||
|
import { PipelineTask } from './pipeline-task.entity';
|
||||||
|
|
||||||
|
const LOG_TIMEOUT_SECONDS = 10_000;
|
||||||
|
|
||||||
|
@Injectable()
|
||||||
|
export class PipelineTaskLogsService {
|
||||||
|
constructor(private readonly redisService: RedisService) {}
|
||||||
|
|
||||||
|
pubSub = new PubSub();
|
||||||
|
|
||||||
|
get redis() {
|
||||||
|
return this.redisService.getClient();
|
||||||
|
}
|
||||||
|
|
||||||
|
getKeys(task: PipelineTask) {
|
||||||
|
return `ptl:${task.id}`;
|
||||||
|
}
|
||||||
|
|
||||||
|
async recordLog(log: PipelineTaskLogMessage) {
|
||||||
|
const logDto = omit(['task'], log);
|
||||||
|
await Promise.all([
|
||||||
|
this.pubSub.publish(this.getKeys(log.task), logDto),
|
||||||
|
this.redis
|
||||||
|
.expire(this.getKeys(log.task), LOG_TIMEOUT_SECONDS)
|
||||||
|
.then(() =>
|
||||||
|
this.redis.rpush(this.getKeys(log.task), JSON.stringify(logDto)),
|
||||||
|
),
|
||||||
|
]);
|
||||||
|
}
|
||||||
|
|
||||||
|
async readLog(task: PipelineTask): Promise<PipelineTaskLogMessage[]> {
|
||||||
|
return await this.redis.lrange(this.getKeys(task), 0, -1).then((items) =>
|
||||||
|
items.map((item) => {
|
||||||
|
const log = JSON.parse(item) as PipelineTaskLogMessage;
|
||||||
|
log.task = task;
|
||||||
|
log.time = new Date(log.time);
|
||||||
|
return log;
|
||||||
|
}),
|
||||||
|
);
|
||||||
|
}
|
||||||
|
|
||||||
|
async readLogsAsPipelineTaskLogs(
|
||||||
|
task: PipelineTask,
|
||||||
|
): Promise<PipelineTaskLogs[]> {
|
||||||
|
const logs = await this.readLog(task);
|
||||||
|
const taskLogs: PipelineTaskLogs[] = [];
|
||||||
|
for (const log of logs) {
|
||||||
|
const taskLog = find<PipelineTaskLogs>(
|
||||||
|
propEq('unit', log.unit),
|
||||||
|
taskLogs,
|
||||||
|
);
|
||||||
|
if (!taskLog) {
|
||||||
|
taskLogs.push({
|
||||||
|
unit: (log.unit as unknown) as PipelineUnits,
|
||||||
|
status: TaskStatuses.working,
|
||||||
|
startedAt: log.time,
|
||||||
|
logs: log.message,
|
||||||
|
});
|
||||||
|
} else {
|
||||||
|
taskLog.logs += log.message;
|
||||||
|
}
|
||||||
|
}
|
||||||
|
return taskLogs;
|
||||||
|
}
|
||||||
|
|
||||||
|
watchLogs(task: PipelineTask) {
|
||||||
|
return this.pubSub.asyncIterator(this.getKeys(task));
|
||||||
|
}
|
||||||
|
}
|
242
src/pipeline-tasks/pipeline-task.consumer.spec.ts
Normal file
242
src/pipeline-tasks/pipeline-task.consumer.spec.ts
Normal file
@ -0,0 +1,242 @@
|
|||||||
|
import { Test, TestingModule } from '@nestjs/testing';
|
||||||
|
import { Job } from 'bull';
|
||||||
|
import { join } from 'path';
|
||||||
|
import { ReposService } from '../repos/repos.service';
|
||||||
|
import { PipelineUnits } from './enums/pipeline-units.enum';
|
||||||
|
import { PipelineTaskConsumer } from './pipeline-task.consumer';
|
||||||
|
import { PipelineTask } from './pipeline-task.entity';
|
||||||
|
import { PipelineTasksService } from './pipeline-tasks.service';
|
||||||
|
import { PipelineTaskLogMessage } from './models/pipeline-task-log-message.module';
|
||||||
|
import { Pipeline } from '../pipelines/pipeline.entity';
|
||||||
|
import { Project } from '../projects/project.entity';
|
||||||
|
import { TaskStatuses } from './enums/task-statuses.enum';
|
||||||
|
import { PipelineTaskLogsService } from './pipeline-task-logs.service';
|
||||||
|
import { ApplicationException } from '../commons/exceptions/application.exception';
|
||||||
|
|
||||||
|
describe('PipelineTaskConsumer', () => {
|
||||||
|
let consumer: PipelineTaskConsumer;
|
||||||
|
let tasksService: PipelineTasksService;
|
||||||
|
let logsService: PipelineTaskLogsService;
|
||||||
|
const getJob = () =>
|
||||||
|
({
|
||||||
|
data: {
|
||||||
|
pipelineId: 'test',
|
||||||
|
units: [PipelineUnits.checkout, PipelineUnits.test],
|
||||||
|
},
|
||||||
|
} as Job<PipelineTask>);
|
||||||
|
|
||||||
|
beforeEach(async () => {
|
||||||
|
const module: TestingModule = await Test.createTestingModule({
|
||||||
|
providers: [
|
||||||
|
{
|
||||||
|
provide: PipelineTasksService,
|
||||||
|
useValue: {
|
||||||
|
doNextTask: () => undefined,
|
||||||
|
updateTask: async (value) => value,
|
||||||
|
},
|
||||||
|
},
|
||||||
|
{
|
||||||
|
provide: ReposService,
|
||||||
|
useValue: {
|
||||||
|
getWorkspaceRootByTask: () => 'workspace-root',
|
||||||
|
checkout: async () => undefined,
|
||||||
|
},
|
||||||
|
},
|
||||||
|
{
|
||||||
|
provide: PipelineTaskLogsService,
|
||||||
|
useValue: {
|
||||||
|
recordLog: async () => undefined,
|
||||||
|
readLogsAsPipelineTaskLogs: async () => [],
|
||||||
|
},
|
||||||
|
},
|
||||||
|
PipelineTaskConsumer,
|
||||||
|
],
|
||||||
|
}).compile();
|
||||||
|
|
||||||
|
tasksService = module.get(PipelineTasksService);
|
||||||
|
logsService = module.get(PipelineTaskLogsService);
|
||||||
|
consumer = module.get(PipelineTaskConsumer);
|
||||||
|
});
|
||||||
|
|
||||||
|
it('should be defined', () => {
|
||||||
|
expect(consumer).toBeDefined();
|
||||||
|
});
|
||||||
|
|
||||||
|
describe('onCompleted', () => {
|
||||||
|
it('should call doNextTask()', () => {
|
||||||
|
const job = getJob();
|
||||||
|
const doNextTask = jest.spyOn(tasksService, 'doNextTask');
|
||||||
|
consumer.onCompleted(job);
|
||||||
|
expect(doNextTask).toHaveBeenCalledTimes(1);
|
||||||
|
});
|
||||||
|
});
|
||||||
|
|
||||||
|
describe('runScript', () => {
|
||||||
|
let logText: string;
|
||||||
|
let errorText: string;
|
||||||
|
let recordLog: jest.SpyInstance;
|
||||||
|
beforeEach(() => {
|
||||||
|
logText = '';
|
||||||
|
errorText = '';
|
||||||
|
recordLog = jest
|
||||||
|
.spyOn(logsService, 'recordLog')
|
||||||
|
.mockImplementation(async (log: PipelineTaskLogMessage) => {
|
||||||
|
logText += log.message;
|
||||||
|
if (log.isError) {
|
||||||
|
errorText += log.message;
|
||||||
|
}
|
||||||
|
});
|
||||||
|
});
|
||||||
|
it('should success and log right message', async () => {
|
||||||
|
await consumer.runScript(
|
||||||
|
'node one-second-work.js',
|
||||||
|
join(__dirname, '../../test/data'),
|
||||||
|
);
|
||||||
|
expect(logText).toMatch(/10.+20.+30.+40.+50.+60.+70.+80.+90/s);
|
||||||
|
expect(recordLog).toHaveBeenCalledTimes(10);
|
||||||
|
expect(
|
||||||
|
((recordLog.mock.calls[8][0] as unknown) as PipelineTaskLogMessage)
|
||||||
|
.message,
|
||||||
|
).toMatch(/^90/);
|
||||||
|
});
|
||||||
|
it('should failed and log right message', async () => {
|
||||||
|
await expect(
|
||||||
|
consumer.runScript(
|
||||||
|
'node bad-work.js',
|
||||||
|
join(__dirname, '../../test/data'),
|
||||||
|
),
|
||||||
|
).rejects.toThrowError(/exec script failed/);
|
||||||
|
expect(errorText).toMatch(/Error Message/);
|
||||||
|
const logs = recordLog.mock.calls
|
||||||
|
.map((call) => ((call[0] as unknown) as PipelineTaskLogMessage).message)
|
||||||
|
.join('');
|
||||||
|
expect(logs).toMatch(/10.+20.+30.+40.+50/s);
|
||||||
|
});
|
||||||
|
it('should log with task', async () => {
|
||||||
|
const task = new PipelineTask();
|
||||||
|
task.id = 'test';
|
||||||
|
|
||||||
|
const recordLog = jest.spyOn(logsService, 'recordLog');
|
||||||
|
await expect(
|
||||||
|
consumer.runScript(
|
||||||
|
'node bad-work.js',
|
||||||
|
join(__dirname, '../../test/data'),
|
||||||
|
task,
|
||||||
|
),
|
||||||
|
).rejects.toThrowError(/exec script failed/);
|
||||||
|
|
||||||
|
expect(errorText).toMatch(/Error Message 2/);
|
||||||
|
expect(
|
||||||
|
((recordLog.mock.calls[2][0] as unknown) as PipelineTaskLogMessage)
|
||||||
|
.task,
|
||||||
|
).toMatchObject(task);
|
||||||
|
});
|
||||||
|
});
|
||||||
|
|
||||||
|
describe('doTask', () => {
|
||||||
|
let task: PipelineTask;
|
||||||
|
|
||||||
|
beforeEach(() => {
|
||||||
|
task = new PipelineTask();
|
||||||
|
task.id = 'test-id';
|
||||||
|
task.logs = [];
|
||||||
|
task.pipeline = new Pipeline();
|
||||||
|
task.pipeline.workUnitMetadata = {
|
||||||
|
version: 1,
|
||||||
|
units: [
|
||||||
|
{
|
||||||
|
type: PipelineUnits.checkout,
|
||||||
|
scripts: [],
|
||||||
|
},
|
||||||
|
{
|
||||||
|
type: PipelineUnits.installDependencies,
|
||||||
|
scripts: ["echo ' Hello, Fennec!'"],
|
||||||
|
},
|
||||||
|
],
|
||||||
|
};
|
||||||
|
task.units = task.pipeline.workUnitMetadata.units.map(
|
||||||
|
(unit) => unit.type,
|
||||||
|
);
|
||||||
|
task.pipeline.project = new Project();
|
||||||
|
task.pipeline.project.name = 'test-project';
|
||||||
|
});
|
||||||
|
|
||||||
|
it('success and update task on db', async () => {
|
||||||
|
const job: Job = ({
|
||||||
|
data: task,
|
||||||
|
update: jest.fn().mockImplementation(() => undefined),
|
||||||
|
} as unknown) as Job;
|
||||||
|
|
||||||
|
jest
|
||||||
|
.spyOn(consumer, 'runScript')
|
||||||
|
.mockImplementation(async () => undefined);
|
||||||
|
const updateTask = jest.spyOn(tasksService, 'updateTask');
|
||||||
|
|
||||||
|
await consumer.doTask(job);
|
||||||
|
|
||||||
|
expect(updateTask).toHaveBeenCalledTimes(2);
|
||||||
|
expect(updateTask.mock.calls[0][0].startedAt).toBeDefined();
|
||||||
|
expect(updateTask.mock.calls[1][0].endedAt).toBeDefined();
|
||||||
|
expect(updateTask.mock.calls[1][0].status).toEqual(TaskStatuses.success);
|
||||||
|
});
|
||||||
|
it('failed and update task on db', async () => {
|
||||||
|
const job: Job = ({
|
||||||
|
data: task,
|
||||||
|
update: jest.fn().mockImplementation(() => undefined),
|
||||||
|
} as unknown) as Job;
|
||||||
|
|
||||||
|
jest.spyOn(consumer, 'runScript').mockImplementation(async () => {
|
||||||
|
throw new ApplicationException('exec script failed');
|
||||||
|
});
|
||||||
|
const updateTask = jest.spyOn(tasksService, 'updateTask');
|
||||||
|
|
||||||
|
await consumer.doTask(job);
|
||||||
|
|
||||||
|
expect(updateTask).toHaveBeenCalledTimes(2);
|
||||||
|
expect(updateTask.mock.calls[0][0].startedAt).toBeDefined();
|
||||||
|
expect(updateTask.mock.calls[1][0].endedAt).toBeDefined();
|
||||||
|
expect(updateTask.mock.calls[1][0].status).toEqual(TaskStatuses.failed);
|
||||||
|
});
|
||||||
|
it('should do all task', async () => {
|
||||||
|
const job: Job = ({
|
||||||
|
data: task,
|
||||||
|
update: jest.fn().mockImplementation(() => undefined),
|
||||||
|
} as unknown) as Job;
|
||||||
|
|
||||||
|
const runScript = jest
|
||||||
|
.spyOn(consumer, 'runScript')
|
||||||
|
.mockImplementation(async () => undefined);
|
||||||
|
const updateTask = jest.spyOn(tasksService, 'updateTask');
|
||||||
|
|
||||||
|
await consumer.doTask(job);
|
||||||
|
|
||||||
|
expect(runScript).toHaveBeenCalledTimes(1);
|
||||||
|
expect(updateTask).toHaveBeenCalledTimes(2);
|
||||||
|
const taskDto: PipelineTask = updateTask.mock.calls[0][0];
|
||||||
|
expect(taskDto.logs).toHaveLength(2);
|
||||||
|
expect(taskDto.logs[0].status).toEqual(TaskStatuses.success);
|
||||||
|
expect(taskDto.logs[0].unit).toEqual(PipelineUnits.checkout);
|
||||||
|
});
|
||||||
|
it('should log error message', async () => {
|
||||||
|
const job: Job = ({
|
||||||
|
data: task,
|
||||||
|
update: jest.fn().mockImplementation(() => undefined),
|
||||||
|
} as unknown) as Job;
|
||||||
|
|
||||||
|
const runScript = jest
|
||||||
|
.spyOn(consumer, 'runScript')
|
||||||
|
.mockImplementation(async () => {
|
||||||
|
throw new Error('bad message');
|
||||||
|
});
|
||||||
|
const updateTask = jest.spyOn(tasksService, 'updateTask');
|
||||||
|
|
||||||
|
await consumer.doTask(job);
|
||||||
|
|
||||||
|
expect(updateTask).toHaveBeenCalledTimes(2);
|
||||||
|
const taskDto: PipelineTask = updateTask.mock.calls[0][0];
|
||||||
|
expect(taskDto.logs).toHaveLength(2);
|
||||||
|
expect(taskDto.logs[0].status).toEqual(TaskStatuses.success);
|
||||||
|
expect(taskDto.logs[1].status).toEqual(TaskStatuses.failed);
|
||||||
|
});
|
||||||
|
});
|
||||||
|
});
|
124
src/pipeline-tasks/pipeline-task.consumer.ts
Normal file
124
src/pipeline-tasks/pipeline-task.consumer.ts
Normal file
@ -0,0 +1,124 @@
|
|||||||
|
import { PipelineTaskLogs } from './models/pipeline-task-logs.model';
|
||||||
|
import { ReposService } from './../repos/repos.service';
|
||||||
|
import { OnQueueCompleted, Process, Processor } from '@nestjs/bull';
|
||||||
|
import { Job } from 'bull';
|
||||||
|
import { spawn } from 'child_process';
|
||||||
|
import { PipelineTask } from './pipeline-task.entity';
|
||||||
|
import { PIPELINE_TASK_QUEUE } from './pipeline-tasks.constants';
|
||||||
|
import { PipelineTasksService } from './pipeline-tasks.service';
|
||||||
|
import { ApplicationException } from '../commons/exceptions/application.exception';
|
||||||
|
import { PipelineUnits } from './enums/pipeline-units.enum';
|
||||||
|
import { PipelineTaskLogMessage } from './models/pipeline-task-log-message.module';
|
||||||
|
import { TaskStatuses } from './enums/task-statuses.enum';
|
||||||
|
import { PipelineTaskLogsService } from './pipeline-task-logs.service';
|
||||||
|
@Processor(PIPELINE_TASK_QUEUE)
|
||||||
|
export class PipelineTaskConsumer {
|
||||||
|
constructor(
|
||||||
|
private readonly service: PipelineTasksService,
|
||||||
|
private readonly reposService: ReposService,
|
||||||
|
private readonly logsService: PipelineTaskLogsService,
|
||||||
|
) {}
|
||||||
|
@Process()
|
||||||
|
async doTask(job: Job<PipelineTask>) {
|
||||||
|
let task = job.data;
|
||||||
|
if (task.pipeline.workUnitMetadata.version !== 1) {
|
||||||
|
throw new ApplicationException(
|
||||||
|
'work unit metadata version is not match.',
|
||||||
|
);
|
||||||
|
}
|
||||||
|
|
||||||
|
task.startedAt = new Date();
|
||||||
|
task.status = TaskStatuses.working;
|
||||||
|
task = await this.service.updateTask(task);
|
||||||
|
await job.update(task);
|
||||||
|
|
||||||
|
const workspaceRoot = this.reposService.getWorkspaceRootByTask(task);
|
||||||
|
|
||||||
|
const units = task.units.map(
|
||||||
|
(type) =>
|
||||||
|
task.pipeline.workUnitMetadata.units.find(
|
||||||
|
(unit) => unit.type === type,
|
||||||
|
) ?? { type: type, scripts: [] },
|
||||||
|
);
|
||||||
|
|
||||||
|
try {
|
||||||
|
for (const unit of units) {
|
||||||
|
const unitLog = new PipelineTaskLogs();
|
||||||
|
unitLog.unit = unit.type;
|
||||||
|
unitLog.startedAt = new Date();
|
||||||
|
try {
|
||||||
|
// 检出代码前执行 git checkout
|
||||||
|
if (unit.type === PipelineUnits.checkout) {
|
||||||
|
await this.reposService.checkout(task, workspaceRoot);
|
||||||
|
unitLog.status = TaskStatuses.success;
|
||||||
|
}
|
||||||
|
for (const script of unit.scripts) {
|
||||||
|
unitLog.logs += `[RUN SCRIPT] ${script}`;
|
||||||
|
await this.runScript(script, workspaceRoot, task, unit.type);
|
||||||
|
}
|
||||||
|
unitLog.status = TaskStatuses.success;
|
||||||
|
} catch (err) {
|
||||||
|
unitLog.status = TaskStatuses.failed;
|
||||||
|
unitLog.logs += err.message;
|
||||||
|
throw err;
|
||||||
|
} finally {
|
||||||
|
unitLog.endedAt = new Date();
|
||||||
|
unitLog.logs = await this.logsService
|
||||||
|
.readLogsAsPipelineTaskLogs(task)
|
||||||
|
.then(
|
||||||
|
(taskLogs) =>
|
||||||
|
taskLogs.find((tl) => tl.unit === unit.type)?.logs ?? '',
|
||||||
|
);
|
||||||
|
task.logs.push(unitLog);
|
||||||
|
await job.update(task);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
task.status = TaskStatuses.success;
|
||||||
|
} catch (err) {
|
||||||
|
task.status = TaskStatuses.failed;
|
||||||
|
console.log(err);
|
||||||
|
} finally {
|
||||||
|
task.endedAt = new Date();
|
||||||
|
task = await this.service.updateTask(task);
|
||||||
|
await job.update(task);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
async runScript(
|
||||||
|
script: string,
|
||||||
|
workspaceRoot: string,
|
||||||
|
task?: PipelineTask,
|
||||||
|
unit?: PipelineUnits,
|
||||||
|
): Promise<void> {
|
||||||
|
return new Promise((resolve, reject) => {
|
||||||
|
const sub = spawn(script, {
|
||||||
|
shell: true,
|
||||||
|
cwd: workspaceRoot,
|
||||||
|
});
|
||||||
|
sub.stderr.on('data', (data: Buffer) => {
|
||||||
|
const str = data.toString();
|
||||||
|
this.logsService.recordLog(
|
||||||
|
PipelineTaskLogMessage.create(task, unit, str, true),
|
||||||
|
);
|
||||||
|
});
|
||||||
|
sub.stdout.on('data', (data: Buffer) => {
|
||||||
|
const str = data.toString();
|
||||||
|
this.logsService.recordLog(
|
||||||
|
PipelineTaskLogMessage.create(task, unit, str, false),
|
||||||
|
);
|
||||||
|
});
|
||||||
|
sub.addListener('close', (code) => {
|
||||||
|
if (code === 0) {
|
||||||
|
return resolve();
|
||||||
|
}
|
||||||
|
return reject(new ApplicationException('exec script failed'));
|
||||||
|
});
|
||||||
|
});
|
||||||
|
}
|
||||||
|
|
||||||
|
@OnQueueCompleted()
|
||||||
|
onCompleted(job: Job<PipelineTask>) {
|
||||||
|
this.service.doNextTask(job.data.pipeline);
|
||||||
|
}
|
||||||
|
}
|
34
src/pipeline-tasks/pipeline-task.entity.ts
Normal file
34
src/pipeline-tasks/pipeline-task.entity.ts
Normal file
@ -0,0 +1,34 @@
|
|||||||
|
import { AppBaseEntity } from './../commons/entities/app-base-entity';
|
||||||
|
import { Field, ObjectType } from '@nestjs/graphql';
|
||||||
|
import { Column, Entity, ManyToOne } from 'typeorm';
|
||||||
|
import { Pipeline } from '../pipelines/pipeline.entity';
|
||||||
|
import { PipelineTaskLogs } from './models/pipeline-task-logs.model';
|
||||||
|
import { TaskStatuses } from './enums/task-statuses.enum';
|
||||||
|
import { PipelineUnits } from './enums/pipeline-units.enum';
|
||||||
|
|
||||||
|
@ObjectType()
|
||||||
|
@Entity()
|
||||||
|
export class PipelineTask extends AppBaseEntity {
|
||||||
|
@ManyToOne(() => Pipeline)
|
||||||
|
pipeline: Pipeline;
|
||||||
|
@Column()
|
||||||
|
pipelineId: string;
|
||||||
|
|
||||||
|
@Column()
|
||||||
|
commit: string;
|
||||||
|
|
||||||
|
@Column({ type: 'enum', enum: PipelineUnits, array: true })
|
||||||
|
units: PipelineUnits[];
|
||||||
|
|
||||||
|
@Column({ type: 'jsonb', default: '[]' })
|
||||||
|
logs: PipelineTaskLogs[];
|
||||||
|
|
||||||
|
@Column({ type: 'enum', enum: TaskStatuses, default: TaskStatuses.pending })
|
||||||
|
status: TaskStatuses;
|
||||||
|
|
||||||
|
@Column({ nullable: true })
|
||||||
|
startedAt?: Date;
|
||||||
|
|
||||||
|
@Column({ nullable: true })
|
||||||
|
endedAt?: Date;
|
||||||
|
}
|
3
src/pipeline-tasks/pipeline-tasks.constants.ts
Normal file
3
src/pipeline-tasks/pipeline-tasks.constants.ts
Normal file
@ -0,0 +1,3 @@
|
|||||||
|
export const PIPELINE_TASK_QUEUE = 'PIPELINE_TASK_QUEUE';
|
||||||
|
export const PIPELINE_TASK_LOG_QUEUE = 'PIPELINE_TASK_LOG_QUEUE';
|
||||||
|
export const PIPELINE_TASK_LOG_PUBSUB = 'PIPELINE_TASK_LOG_PUBSUB';
|
42
src/pipeline-tasks/pipeline-tasks.module.ts
Normal file
42
src/pipeline-tasks/pipeline-tasks.module.ts
Normal file
@ -0,0 +1,42 @@
|
|||||||
|
import { Module } from '@nestjs/common';
|
||||||
|
import { PipelineTasksService } from './pipeline-tasks.service';
|
||||||
|
import { PipelineTasksResolver } from './pipeline-tasks.resolver';
|
||||||
|
import { TypeOrmModule } from '@nestjs/typeorm';
|
||||||
|
import { PipelineTask } from './pipeline-task.entity';
|
||||||
|
import { Pipeline } from '../pipelines/pipeline.entity';
|
||||||
|
import { ReposModule } from '../repos/repos.module';
|
||||||
|
import { RedisModule } from 'nestjs-redis';
|
||||||
|
import { BullModule } from '@nestjs/bull';
|
||||||
|
import { PipelineTaskConsumer } from './pipeline-task.consumer';
|
||||||
|
import {
|
||||||
|
PIPELINE_TASK_QUEUE,
|
||||||
|
PIPELINE_TASK_LOG_QUEUE,
|
||||||
|
PIPELINE_TASK_LOG_PUBSUB,
|
||||||
|
} from './pipeline-tasks.constants';
|
||||||
|
import { PipelineTaskLogsService } from './pipeline-task-logs.service';
|
||||||
|
import { PubSub } from 'apollo-server-express';
|
||||||
|
|
||||||
|
@Module({
|
||||||
|
imports: [
|
||||||
|
TypeOrmModule.forFeature([PipelineTask, Pipeline]),
|
||||||
|
BullModule.registerQueue(
|
||||||
|
{
|
||||||
|
name: PIPELINE_TASK_QUEUE,
|
||||||
|
},
|
||||||
|
{ name: PIPELINE_TASK_LOG_QUEUE },
|
||||||
|
),
|
||||||
|
RedisModule,
|
||||||
|
ReposModule,
|
||||||
|
],
|
||||||
|
providers: [
|
||||||
|
PipelineTasksService,
|
||||||
|
PipelineTasksResolver,
|
||||||
|
PipelineTaskConsumer,
|
||||||
|
PipelineTaskLogsService,
|
||||||
|
{
|
||||||
|
provide: Symbol(PIPELINE_TASK_LOG_PUBSUB),
|
||||||
|
useValue: new PubSub(),
|
||||||
|
},
|
||||||
|
],
|
||||||
|
})
|
||||||
|
export class PipelineTasksModule {}
|
25
src/pipeline-tasks/pipeline-tasks.resolver.spec.ts
Normal file
25
src/pipeline-tasks/pipeline-tasks.resolver.spec.ts
Normal file
@ -0,0 +1,25 @@
|
|||||||
|
import { Test, TestingModule } from '@nestjs/testing';
|
||||||
|
import { PipelineTasksResolver } from './pipeline-tasks.resolver';
|
||||||
|
import { PipelineTaskLogsService } from './pipeline-task-logs.service';
|
||||||
|
|
||||||
|
describe('PipelineTasksResolver', () => {
|
||||||
|
let resolver: PipelineTasksResolver;
|
||||||
|
|
||||||
|
beforeEach(async () => {
|
||||||
|
const module: TestingModule = await Test.createTestingModule({
|
||||||
|
providers: [
|
||||||
|
PipelineTasksResolver,
|
||||||
|
{
|
||||||
|
provide: PipelineTaskLogsService,
|
||||||
|
useValue: {},
|
||||||
|
},
|
||||||
|
],
|
||||||
|
}).compile();
|
||||||
|
|
||||||
|
resolver = module.get<PipelineTasksResolver>(PipelineTasksResolver);
|
||||||
|
});
|
||||||
|
|
||||||
|
it('should be defined', () => {
|
||||||
|
expect(resolver).toBeDefined();
|
||||||
|
});
|
||||||
|
});
|
50
src/pipeline-tasks/pipeline-tasks.resolver.ts
Normal file
50
src/pipeline-tasks/pipeline-tasks.resolver.ts
Normal file
@ -0,0 +1,50 @@
|
|||||||
|
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 { PipelineTaskLogMessage } from './models/pipeline-task-log-message.module';
|
||||||
|
import { PipelineTaskLogArgs } from './dtos/pipeline-task-log.args';
|
||||||
|
import { PipelineTaskLogsService } from './pipeline-task-logs.service';
|
||||||
|
|
||||||
|
@Resolver()
|
||||||
|
export class PipelineTasksResolver {
|
||||||
|
constructor(
|
||||||
|
private readonly service: PipelineTasksService,
|
||||||
|
private readonly logsService: PipelineTaskLogsService,
|
||||||
|
) {}
|
||||||
|
|
||||||
|
@Mutation(() => PipelineTask)
|
||||||
|
async createPipelineTask(@Args('task') taskDto: CreatePipelineTaskInput) {
|
||||||
|
return await this.service.addTask(taskDto);
|
||||||
|
}
|
||||||
|
|
||||||
|
@Subscription(() => PipelineTaskLogMessage, {
|
||||||
|
resolve: (value) => {
|
||||||
|
return value;
|
||||||
|
},
|
||||||
|
})
|
||||||
|
async pipelineTaskLog(@Args() args: PipelineTaskLogArgs) {
|
||||||
|
const task = await this.service.findTaskById(args.taskId);
|
||||||
|
const asyncIterator = this.logsService.watchLogs(task);
|
||||||
|
return asyncIterator;
|
||||||
|
}
|
||||||
|
|
||||||
|
@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 findPipelineTask(@Args('id') id: string) {
|
||||||
|
return await this.service.findTaskById(id);
|
||||||
|
}
|
||||||
|
}
|
201
src/pipeline-tasks/pipeline-tasks.service.spec.ts
Normal file
201
src/pipeline-tasks/pipeline-tasks.service.spec.ts
Normal file
@ -0,0 +1,201 @@
|
|||||||
|
import { Test, TestingModule } from '@nestjs/testing';
|
||||||
|
import { PipelineTasksService } from './pipeline-tasks.service';
|
||||||
|
import { getRepositoryToken } from '@nestjs/typeorm';
|
||||||
|
import { PipelineTask } from './pipeline-task.entity';
|
||||||
|
import { PIPELINE_TASK_QUEUE } from './pipeline-tasks.constants';
|
||||||
|
import { getQueueToken } from '@nestjs/bull';
|
||||||
|
import { RedisService } from 'nestjs-redis';
|
||||||
|
import { Pipeline } from '../pipelines/pipeline.entity';
|
||||||
|
import { EntityNotFoundError } from 'typeorm/error/EntityNotFoundError';
|
||||||
|
import { Repository } from 'typeorm';
|
||||||
|
import { Queue } from 'bull';
|
||||||
|
import { LockFailedException } from '../commons/exceptions/lock-failed.exception';
|
||||||
|
|
||||||
|
describe('PipelineTasksService', () => {
|
||||||
|
let service: PipelineTasksService;
|
||||||
|
let module: TestingModule;
|
||||||
|
let taskRepository: Repository<PipelineTask>;
|
||||||
|
let pipelineRepository: Repository<Pipeline>;
|
||||||
|
const getBasePipeline = () =>
|
||||||
|
({
|
||||||
|
id: 'test',
|
||||||
|
name: '测试流水线',
|
||||||
|
branch: 'master',
|
||||||
|
workUnitMetadata: {},
|
||||||
|
project: {
|
||||||
|
id: 'test-project',
|
||||||
|
},
|
||||||
|
} as Pipeline);
|
||||||
|
let redisClient;
|
||||||
|
let taskQueue: Queue;
|
||||||
|
const getTask = () =>
|
||||||
|
({
|
||||||
|
pipelineId: 'test',
|
||||||
|
commit: 'test',
|
||||||
|
units: [],
|
||||||
|
} as PipelineTask);
|
||||||
|
|
||||||
|
beforeEach(async () => {
|
||||||
|
redisClient = (() => ({
|
||||||
|
set: jest.fn().mockImplementation(async () => 'OK'),
|
||||||
|
del: jest.fn().mockImplementation(async () => 'test'),
|
||||||
|
get: jest.fn().mockImplementation(async () => 'test'),
|
||||||
|
lpush: jest.fn().mockImplementation(async () => 1),
|
||||||
|
rpop: jest.fn().mockImplementation(async () => JSON.stringify(getTask())),
|
||||||
|
}))() as any;
|
||||||
|
taskQueue = (() => ({
|
||||||
|
add: jest.fn().mockImplementation(async () => null),
|
||||||
|
}))() as any;
|
||||||
|
module = await Test.createTestingModule({
|
||||||
|
providers: [
|
||||||
|
PipelineTasksService,
|
||||||
|
{
|
||||||
|
provide: getRepositoryToken(PipelineTask),
|
||||||
|
useValue: new Repository(),
|
||||||
|
},
|
||||||
|
{
|
||||||
|
provide: getRepositoryToken(Pipeline),
|
||||||
|
useValue: new Repository(),
|
||||||
|
},
|
||||||
|
{
|
||||||
|
provide: getQueueToken(PIPELINE_TASK_QUEUE),
|
||||||
|
useValue: taskQueue,
|
||||||
|
},
|
||||||
|
{
|
||||||
|
provide: RedisService,
|
||||||
|
useValue: {
|
||||||
|
getClient: jest.fn(() => redisClient),
|
||||||
|
},
|
||||||
|
},
|
||||||
|
],
|
||||||
|
}).compile();
|
||||||
|
|
||||||
|
service = module.get<PipelineTasksService>(PipelineTasksService);
|
||||||
|
taskRepository = module.get(getRepositoryToken(PipelineTask));
|
||||||
|
pipelineRepository = module.get(getRepositoryToken(Pipeline));
|
||||||
|
jest
|
||||||
|
.spyOn(taskRepository, 'save')
|
||||||
|
.mockImplementation(async (data: any) => data);
|
||||||
|
jest
|
||||||
|
.spyOn(taskRepository, 'create')
|
||||||
|
.mockImplementation((data: any) => data);
|
||||||
|
});
|
||||||
|
|
||||||
|
it('should be defined', () => {
|
||||||
|
expect(service).toBeDefined();
|
||||||
|
});
|
||||||
|
|
||||||
|
describe('addTask', () => {
|
||||||
|
beforeEach(() => {
|
||||||
|
jest
|
||||||
|
.spyOn(pipelineRepository, 'findOneOrFail')
|
||||||
|
.mockImplementation(async () => getBasePipeline());
|
||||||
|
});
|
||||||
|
it('pipeline not found', async () => {
|
||||||
|
jest.spyOn(taskRepository, 'findOneOrFail').mockImplementation(() => {
|
||||||
|
throw new EntityNotFoundError(Pipeline, {});
|
||||||
|
});
|
||||||
|
await expect(
|
||||||
|
service.addTask({ pipelineId: 'test', commit: 'test', units: [] }),
|
||||||
|
).rejects;
|
||||||
|
});
|
||||||
|
it('create task on db', async () => {
|
||||||
|
const save = jest
|
||||||
|
.spyOn(taskRepository, 'save')
|
||||||
|
.mockImplementation(async (data: any) => data);
|
||||||
|
jest
|
||||||
|
.spyOn(service, 'doNextTask')
|
||||||
|
.mockImplementation(async () => undefined);
|
||||||
|
await service.addTask({ pipelineId: 'test', commit: 'test', units: [] }),
|
||||||
|
expect(save.mock.calls[0][0]).toMatchObject({
|
||||||
|
pipelineId: 'test',
|
||||||
|
commit: 'test',
|
||||||
|
units: [],
|
||||||
|
});
|
||||||
|
});
|
||||||
|
it('add task', async () => {
|
||||||
|
const lpush = jest.spyOn(redisClient, 'lpush');
|
||||||
|
const doNextTask = jest.spyOn(service, 'doNextTask');
|
||||||
|
jest
|
||||||
|
.spyOn(service, 'doNextTask')
|
||||||
|
.mockImplementation(async () => undefined);
|
||||||
|
await service.addTask({ pipelineId: 'test', commit: 'test', units: [] });
|
||||||
|
expect(typeof lpush.mock.calls[0][1] === 'string').toBeTruthy();
|
||||||
|
expect(JSON.parse(lpush.mock.calls[0][1] as string)).toMatchObject({
|
||||||
|
pipelineId: 'test',
|
||||||
|
commit: 'test',
|
||||||
|
units: [],
|
||||||
|
pipeline: getBasePipeline(),
|
||||||
|
});
|
||||||
|
expect(doNextTask).toHaveBeenCalledWith(getBasePipeline());
|
||||||
|
});
|
||||||
|
});
|
||||||
|
|
||||||
|
describe('doNextTask', () => {
|
||||||
|
it('add task to queue', async () => {
|
||||||
|
let lckValue: string;
|
||||||
|
const set = jest
|
||||||
|
.spyOn(redisClient, 'set')
|
||||||
|
.mockImplementation(async (...args) => (lckValue = args[3] as string));
|
||||||
|
const get = jest
|
||||||
|
.spyOn(redisClient, 'get')
|
||||||
|
.mockImplementation(async () => lckValue);
|
||||||
|
const del = jest.spyOn(redisClient, 'del');
|
||||||
|
const rpop = jest.spyOn(redisClient, 'rpop');
|
||||||
|
const add = jest.spyOn(taskQueue, 'add');
|
||||||
|
|
||||||
|
await service.doNextTask(getBasePipeline());
|
||||||
|
|
||||||
|
expect(add).toHaveBeenCalledWith(getTask());
|
||||||
|
expect(set).toHaveBeenCalledTimes(1);
|
||||||
|
expect(rpop).toHaveBeenCalledTimes(1);
|
||||||
|
expect(get).toHaveBeenCalledTimes(1);
|
||||||
|
expect(del).toHaveBeenCalledTimes(1);
|
||||||
|
});
|
||||||
|
it('pipeline is busy', async () => {
|
||||||
|
let remainTimes = 3;
|
||||||
|
|
||||||
|
let lckValue: string;
|
||||||
|
const set = jest
|
||||||
|
.spyOn(redisClient, 'set')
|
||||||
|
.mockImplementation(async (...args) => {
|
||||||
|
if (remainTimes-- > 0) {
|
||||||
|
throw new Error();
|
||||||
|
} else {
|
||||||
|
lckValue = args[3] as string;
|
||||||
|
}
|
||||||
|
});
|
||||||
|
const get = jest
|
||||||
|
.spyOn(redisClient, 'get')
|
||||||
|
.mockImplementation(async () => lckValue);
|
||||||
|
const del = jest.spyOn(redisClient, 'del');
|
||||||
|
const rpop = jest.spyOn(redisClient, 'rpop');
|
||||||
|
const add = jest.spyOn(taskQueue, 'add');
|
||||||
|
|
||||||
|
await service.doNextTask(getBasePipeline());
|
||||||
|
|
||||||
|
expect(rpop).toHaveBeenCalledTimes(1);
|
||||||
|
expect(set).toHaveBeenCalledTimes(4);
|
||||||
|
expect(get).toHaveBeenCalledTimes(1);
|
||||||
|
expect(del).toHaveBeenCalledTimes(1);
|
||||||
|
expect(add).toHaveBeenCalledWith(getTask());
|
||||||
|
}, 10_000);
|
||||||
|
it('pipeline always busy and timeout', async () => {
|
||||||
|
const set = jest
|
||||||
|
.spyOn(redisClient, 'set')
|
||||||
|
.mockImplementation(async () => {
|
||||||
|
throw new Error();
|
||||||
|
});
|
||||||
|
const get = jest.spyOn(redisClient, 'get');
|
||||||
|
const del = jest.spyOn(redisClient, 'del');
|
||||||
|
|
||||||
|
await expect(
|
||||||
|
service.doNextTask(getBasePipeline()),
|
||||||
|
).rejects.toBeInstanceOf(LockFailedException);
|
||||||
|
|
||||||
|
expect(set).toHaveBeenCalledTimes(5);
|
||||||
|
expect(get).toHaveBeenCalledTimes(0);
|
||||||
|
expect(del).toHaveBeenCalledTimes(0);
|
||||||
|
}, 15_000);
|
||||||
|
});
|
||||||
|
});
|
96
src/pipeline-tasks/pipeline-tasks.service.ts
Normal file
96
src/pipeline-tasks/pipeline-tasks.service.ts
Normal file
@ -0,0 +1,96 @@
|
|||||||
|
import { Injectable } from '@nestjs/common';
|
||||||
|
import { InjectRepository } from '@nestjs/typeorm';
|
||||||
|
import { PipelineTask } from './pipeline-task.entity';
|
||||||
|
import { Repository } from 'typeorm';
|
||||||
|
import { CreatePipelineTaskInput } from './dtos/create-pipeline-task.input';
|
||||||
|
import { RedisService } from 'nestjs-redis';
|
||||||
|
import { Pipeline } from '../pipelines/pipeline.entity';
|
||||||
|
import { InjectQueue } from '@nestjs/bull';
|
||||||
|
import { PIPELINE_TASK_QUEUE } from './pipeline-tasks.constants';
|
||||||
|
import { Queue } from 'bull';
|
||||||
|
import { LockFailedException } from '../commons/exceptions/lock-failed.exception';
|
||||||
|
import { PubSub } from 'apollo-server-express';
|
||||||
|
|
||||||
|
@Injectable()
|
||||||
|
export class PipelineTasksService {
|
||||||
|
pubSub = new PubSub();
|
||||||
|
constructor(
|
||||||
|
@InjectRepository(PipelineTask)
|
||||||
|
private readonly repository: Repository<PipelineTask>,
|
||||||
|
@InjectRepository(Pipeline)
|
||||||
|
private readonly pipelineRepository: Repository<Pipeline>,
|
||||||
|
@InjectQueue(PIPELINE_TASK_QUEUE)
|
||||||
|
private readonly queue: Queue<PipelineTask>,
|
||||||
|
private readonly redis: RedisService,
|
||||||
|
) {}
|
||||||
|
async addTask(dto: CreatePipelineTaskInput) {
|
||||||
|
const pipeline = await this.pipelineRepository.findOneOrFail({
|
||||||
|
where: { id: dto.pipelineId },
|
||||||
|
relations: ['project'],
|
||||||
|
});
|
||||||
|
const task = await this.repository.save(this.repository.create(dto));
|
||||||
|
task.pipeline = pipeline;
|
||||||
|
|
||||||
|
const tasksKey = this.getRedisTokens(pipeline)[1];
|
||||||
|
const redis = this.redis.getClient();
|
||||||
|
await redis.lpush(tasksKey, JSON.stringify(task));
|
||||||
|
await this.doNextTask(pipeline);
|
||||||
|
return task;
|
||||||
|
}
|
||||||
|
|
||||||
|
async findTaskById(id: string) {
|
||||||
|
return await this.repository.findOneOrFail({ id });
|
||||||
|
}
|
||||||
|
|
||||||
|
async listTasksByPipelineId(pipelineId: string) {
|
||||||
|
return await this.repository.find({ pipelineId });
|
||||||
|
}
|
||||||
|
|
||||||
|
async doNextTask(pipeline: Pipeline) {
|
||||||
|
const [lckKey, tasksKey] = this.getRedisTokens(pipeline);
|
||||||
|
const redis = this.redis.getClient();
|
||||||
|
|
||||||
|
const unLck = await new Promise<() => Promise<void>>(
|
||||||
|
async (resolve, reject) => {
|
||||||
|
const lckValue = Date.now().toString();
|
||||||
|
for (let i = 0; i < 5; i++) {
|
||||||
|
if (
|
||||||
|
await redis
|
||||||
|
.set(lckKey, 0, 'EX', lckValue, 'NX')
|
||||||
|
.then(() => true)
|
||||||
|
.catch(() => false)
|
||||||
|
) {
|
||||||
|
resolve(async () => {
|
||||||
|
if ((await redis.get(lckKey)) === lckValue) {
|
||||||
|
await redis.del(lckKey);
|
||||||
|
}
|
||||||
|
});
|
||||||
|
return;
|
||||||
|
}
|
||||||
|
await new Promise((resolve) => setTimeout(resolve, 2000));
|
||||||
|
}
|
||||||
|
reject(new LockFailedException(lckKey));
|
||||||
|
},
|
||||||
|
);
|
||||||
|
|
||||||
|
const task = JSON.parse(
|
||||||
|
(await redis.rpop(tasksKey).finally(() => unLck())) ?? 'null',
|
||||||
|
);
|
||||||
|
if (task) {
|
||||||
|
await this.queue.add(task);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
async updateTask(task: PipelineTask) {
|
||||||
|
this.pubSub.publish(task.id, task);
|
||||||
|
return await this.repository.save(task);
|
||||||
|
}
|
||||||
|
|
||||||
|
async watchTaskUpdated(id: string) {
|
||||||
|
return this.pubSub.asyncIterator(id);
|
||||||
|
}
|
||||||
|
|
||||||
|
getRedisTokens(pipeline: Pipeline): [string, string] {
|
||||||
|
return [`pipeline-${pipeline.id}:lck`, `pipeline-${pipeline.id}:tasks`];
|
||||||
|
}
|
||||||
|
}
|
27
src/pipelines/dtos/create-pipeline.input.ts
Normal file
27
src/pipelines/dtos/create-pipeline.input.ts
Normal file
@ -0,0 +1,27 @@
|
|||||||
|
import { InputType } from '@nestjs/graphql';
|
||||||
|
import { WorkUnitMetadata } from '../../pipeline-tasks/models/work-unit-metadata.model';
|
||||||
|
import {
|
||||||
|
IsObject,
|
||||||
|
IsOptional,
|
||||||
|
IsString,
|
||||||
|
IsUUID,
|
||||||
|
MaxLength,
|
||||||
|
} from 'class-validator';
|
||||||
|
|
||||||
|
@InputType({ isAbstract: true })
|
||||||
|
export class CreatePipelineInput {
|
||||||
|
@IsUUID()
|
||||||
|
projectId: string;
|
||||||
|
|
||||||
|
@IsString()
|
||||||
|
@MaxLength(100)
|
||||||
|
branch: string;
|
||||||
|
|
||||||
|
@IsString()
|
||||||
|
@MaxLength(32)
|
||||||
|
name: string;
|
||||||
|
|
||||||
|
@IsOptional()
|
||||||
|
@IsObject()
|
||||||
|
workUnitMetadata: WorkUnitMetadata;
|
||||||
|
}
|
8
src/pipelines/dtos/list-pipelines.args.ts
Normal file
8
src/pipelines/dtos/list-pipelines.args.ts
Normal file
@ -0,0 +1,8 @@
|
|||||||
|
import { ArgsType } from '@nestjs/graphql';
|
||||||
|
import { IsUUID } from 'class-validator';
|
||||||
|
|
||||||
|
@ArgsType()
|
||||||
|
export class ListPipelineArgs {
|
||||||
|
@IsUUID()
|
||||||
|
projectId?: string;
|
||||||
|
}
|
5
src/pipelines/dtos/update-pipeline.input.ts
Normal file
5
src/pipelines/dtos/update-pipeline.input.ts
Normal file
@ -0,0 +1,5 @@
|
|||||||
|
import { InputType } from '@nestjs/graphql';
|
||||||
|
import { CreatePipelineInput } from './create-pipeline.input';
|
||||||
|
|
||||||
|
@InputType()
|
||||||
|
export class UpdatePipelineInput extends CreatePipelineInput {}
|
23
src/pipelines/pipeline.entity.ts
Normal file
23
src/pipelines/pipeline.entity.ts
Normal 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;
|
||||||
|
}
|
18
src/pipelines/pipelines.module.ts
Normal file
18
src/pipelines/pipelines.module.ts
Normal file
@ -0,0 +1,18 @@
|
|||||||
|
import { Module } from '@nestjs/common';
|
||||||
|
import { PipelinesResolver } from './pipelines.resolver';
|
||||||
|
import { PipelinesService } from './pipelines.service';
|
||||||
|
import { TypeOrmModule } from '@nestjs/typeorm';
|
||||||
|
import { Pipeline } from './pipeline.entity';
|
||||||
|
import { BullModule } from '@nestjs/bull';
|
||||||
|
import { LIST_LOGS_TASK } from '../repos/repos.constants';
|
||||||
|
|
||||||
|
@Module({
|
||||||
|
imports: [
|
||||||
|
TypeOrmModule.forFeature([Pipeline]),
|
||||||
|
BullModule.registerQueue({
|
||||||
|
name: LIST_LOGS_TASK,
|
||||||
|
}),
|
||||||
|
],
|
||||||
|
providers: [PipelinesResolver, PipelinesService],
|
||||||
|
})
|
||||||
|
export class PipelinesModule {}
|
25
src/pipelines/pipelines.resolver.spec.ts
Normal file
25
src/pipelines/pipelines.resolver.spec.ts
Normal file
@ -0,0 +1,25 @@
|
|||||||
|
import { Test, TestingModule } from '@nestjs/testing';
|
||||||
|
import { PipelinesResolver } from './pipelines.resolver';
|
||||||
|
import { PipelinesService } from './pipelines.service';
|
||||||
|
|
||||||
|
describe('PipelinesResolver', () => {
|
||||||
|
let resolver: PipelinesResolver;
|
||||||
|
|
||||||
|
beforeEach(async () => {
|
||||||
|
const module: TestingModule = await Test.createTestingModule({
|
||||||
|
providers: [
|
||||||
|
PipelinesResolver,
|
||||||
|
{
|
||||||
|
provide: PipelinesService,
|
||||||
|
useValue: {},
|
||||||
|
},
|
||||||
|
],
|
||||||
|
}).compile();
|
||||||
|
|
||||||
|
resolver = module.get<PipelinesResolver>(PipelinesResolver);
|
||||||
|
});
|
||||||
|
|
||||||
|
it('should be defined', () => {
|
||||||
|
expect(resolver).toBeDefined();
|
||||||
|
});
|
||||||
|
});
|
57
src/pipelines/pipelines.resolver.ts
Normal file
57
src/pipelines/pipelines.resolver.ts
Normal file
@ -0,0 +1,57 @@
|
|||||||
|
import { Args, Mutation, Query, Resolver, Subscription } from '@nestjs/graphql';
|
||||||
|
import { CreatePipelineInput } from './dtos/create-pipeline.input';
|
||||||
|
import { UpdatePipelineInput } from './dtos/update-pipeline.input';
|
||||||
|
import { Pipeline } from './pipeline.entity';
|
||||||
|
import { PipelinesService } from './pipelines.service';
|
||||||
|
import { ListPipelineArgs } from './dtos/list-pipelines.args';
|
||||||
|
import { LogList } from '../repos/dtos/log-list.model';
|
||||||
|
|
||||||
|
@Resolver()
|
||||||
|
export class PipelinesResolver {
|
||||||
|
constructor(private readonly service: PipelinesService) {}
|
||||||
|
@Query(() => [Pipeline])
|
||||||
|
async listPipelines(@Args() dto: ListPipelineArgs) {
|
||||||
|
return await this.service.list(dto);
|
||||||
|
}
|
||||||
|
|
||||||
|
@Query(() => Pipeline)
|
||||||
|
async findPipeline(@Args('id', { type: () => String }) id: string) {
|
||||||
|
return await this.service.findOne(id);
|
||||||
|
}
|
||||||
|
|
||||||
|
@Mutation(() => Pipeline)
|
||||||
|
async createPipeline(
|
||||||
|
@Args('pipeline', { type: () => CreatePipelineInput })
|
||||||
|
dto: UpdatePipelineInput,
|
||||||
|
) {
|
||||||
|
return await this.service.create(dto);
|
||||||
|
}
|
||||||
|
|
||||||
|
@Mutation(() => Pipeline)
|
||||||
|
async modifyPipeline(
|
||||||
|
@Args('id', { type: () => String }) id: string,
|
||||||
|
@Args('Pipeline', { type: () => UpdatePipelineInput })
|
||||||
|
dto: UpdatePipelineInput,
|
||||||
|
) {
|
||||||
|
const tmp = await this.service.update(id, dto);
|
||||||
|
console.log(tmp);
|
||||||
|
return tmp;
|
||||||
|
}
|
||||||
|
|
||||||
|
@Mutation(() => Number)
|
||||||
|
async deletePipeline(@Args('id', { type: () => String }) id: string) {
|
||||||
|
return await this.service.remove(id);
|
||||||
|
}
|
||||||
|
|
||||||
|
@Subscription(() => LogList, {
|
||||||
|
resolve: (value) => {
|
||||||
|
return value;
|
||||||
|
},
|
||||||
|
})
|
||||||
|
async listLogsForPipeline(@Args('id', { type: () => String }) id: string) {
|
||||||
|
const job = await this.service.listLogsForPipeline(id);
|
||||||
|
return (async function* () {
|
||||||
|
yield await job.finished();
|
||||||
|
})();
|
||||||
|
}
|
||||||
|
}
|
69
src/pipelines/pipelines.service.spec.ts
Normal file
69
src/pipelines/pipelines.service.spec.ts
Normal file
@ -0,0 +1,69 @@
|
|||||||
|
import { Test, TestingModule } from '@nestjs/testing';
|
||||||
|
import { PipelinesService } from './pipelines.service';
|
||||||
|
import { Pipeline } from './pipeline.entity';
|
||||||
|
import { getRepositoryToken } from '@nestjs/typeorm';
|
||||||
|
import { getQueueToken } from '@nestjs/bull';
|
||||||
|
import { LIST_LOGS_TASK } from '../repos/repos.constants';
|
||||||
|
import { Repository } from 'typeorm';
|
||||||
|
import { Project } from '../projects/project.entity';
|
||||||
|
import { Job, Queue } from 'bull';
|
||||||
|
import { ListLogsOption } from '../repos/models/list-logs.options';
|
||||||
|
|
||||||
|
describe('PipelinesService', () => {
|
||||||
|
let service: PipelinesService;
|
||||||
|
let repository: Repository<Pipeline>;
|
||||||
|
let pipeline: Pipeline;
|
||||||
|
let queue: Queue<ListLogsOption>;
|
||||||
|
|
||||||
|
beforeEach(async () => {
|
||||||
|
pipeline = Object.assign(new Pipeline(), {
|
||||||
|
id: 'test-pipeline',
|
||||||
|
name: 'pipeline',
|
||||||
|
branch: 'master',
|
||||||
|
projectId: 'test-project',
|
||||||
|
project: Object.assign(new Project(), {
|
||||||
|
id: 'test-project',
|
||||||
|
name: 'project',
|
||||||
|
} as Project),
|
||||||
|
} as Pipeline);
|
||||||
|
|
||||||
|
const module: TestingModule = await Test.createTestingModule({
|
||||||
|
providers: [
|
||||||
|
PipelinesService,
|
||||||
|
{
|
||||||
|
provide: getRepositoryToken(Pipeline),
|
||||||
|
useValue: {
|
||||||
|
findOneOrFail: jest.fn().mockImplementation(() => pipeline),
|
||||||
|
},
|
||||||
|
},
|
||||||
|
{
|
||||||
|
provide: getQueueToken(LIST_LOGS_TASK),
|
||||||
|
useValue: {
|
||||||
|
add: jest.fn().mockImplementation(() => ({ id: 1 } as Job)),
|
||||||
|
},
|
||||||
|
},
|
||||||
|
],
|
||||||
|
}).compile();
|
||||||
|
|
||||||
|
service = module.get<PipelinesService>(PipelinesService);
|
||||||
|
repository = module.get(getRepositoryToken(Pipeline));
|
||||||
|
queue = module.get(getQueueToken(LIST_LOGS_TASK));
|
||||||
|
});
|
||||||
|
|
||||||
|
it('should be defined', () => {
|
||||||
|
expect(service).toBeDefined();
|
||||||
|
});
|
||||||
|
|
||||||
|
describe('listLogsForPipeline', () => {
|
||||||
|
it('should send task to queue.', async () => {
|
||||||
|
const add = jest.spyOn(queue, 'add');
|
||||||
|
await expect(
|
||||||
|
service.listLogsForPipeline('test-pipeline'),
|
||||||
|
).resolves.toEqual({ id: 1 });
|
||||||
|
expect(add).toBeCalledWith({
|
||||||
|
project: pipeline.project,
|
||||||
|
branch: pipeline.branch,
|
||||||
|
});
|
||||||
|
});
|
||||||
|
});
|
||||||
|
});
|
55
src/pipelines/pipelines.service.ts
Normal file
55
src/pipelines/pipelines.service.ts
Normal file
@ -0,0 +1,55 @@
|
|||||||
|
import { Injectable } from '@nestjs/common';
|
||||||
|
import { InjectRepository } from '@nestjs/typeorm';
|
||||||
|
import { Pipeline } from './pipeline.entity';
|
||||||
|
import { Repository } from 'typeorm';
|
||||||
|
import { BaseDbService } from '../commons/services/base-db.service';
|
||||||
|
import { CreatePipelineInput } from './dtos/create-pipeline.input';
|
||||||
|
import { UpdatePipelineInput } from './dtos/update-pipeline.input';
|
||||||
|
import { ListPipelineArgs } from './dtos/list-pipelines.args';
|
||||||
|
import { InjectQueue } from '@nestjs/bull';
|
||||||
|
import { LIST_LOGS_TASK } from '../repos/repos.constants';
|
||||||
|
import { Queue } from 'bull';
|
||||||
|
import { ListLogsOption } from '../repos/models/list-logs.options';
|
||||||
|
|
||||||
|
@Injectable()
|
||||||
|
export class PipelinesService extends BaseDbService<Pipeline> {
|
||||||
|
readonly uniqueFields: Array<Array<keyof Pipeline>> = [['projectId', 'name']];
|
||||||
|
constructor(
|
||||||
|
@InjectRepository(Pipeline)
|
||||||
|
readonly repository: Repository<Pipeline>,
|
||||||
|
@InjectQueue(LIST_LOGS_TASK)
|
||||||
|
private readonly listLogsQueue: Queue<ListLogsOption>,
|
||||||
|
) {
|
||||||
|
super(repository);
|
||||||
|
}
|
||||||
|
async list(dto: ListPipelineArgs) {
|
||||||
|
return this.repository.find(dto);
|
||||||
|
}
|
||||||
|
|
||||||
|
async create(dto: CreatePipelineInput) {
|
||||||
|
await this.isDuplicateEntity(dto);
|
||||||
|
return await this.repository.save(this.repository.create(dto));
|
||||||
|
}
|
||||||
|
|
||||||
|
async update(id: string, dto: UpdatePipelineInput) {
|
||||||
|
await this.isDuplicateEntityForUpdate(id, dto);
|
||||||
|
const old = await this.findOne(id);
|
||||||
|
return await this.repository.save(this.repository.merge(old, dto));
|
||||||
|
}
|
||||||
|
|
||||||
|
async remove(id: string) {
|
||||||
|
return (await this.repository.softDelete({ id })).affected;
|
||||||
|
}
|
||||||
|
|
||||||
|
async listLogsForPipeline(id: string) {
|
||||||
|
const pipeline = await this.repository.findOneOrFail({
|
||||||
|
where: { id },
|
||||||
|
relations: ['project'],
|
||||||
|
});
|
||||||
|
const job = await this.listLogsQueue.add({
|
||||||
|
project: pipeline.project,
|
||||||
|
branch: pipeline.branch,
|
||||||
|
});
|
||||||
|
return job;
|
||||||
|
}
|
||||||
|
}
|
@ -22,7 +22,4 @@ export class Project extends AppBaseEntity {
|
|||||||
|
|
||||||
@Column({ nullable: true })
|
@Column({ nullable: true })
|
||||||
webHookSecret?: string;
|
webHookSecret?: string;
|
||||||
|
|
||||||
@DeleteDateColumn()
|
|
||||||
deletedAt?: Date;
|
|
||||||
}
|
}
|
||||||
|
@ -4,6 +4,7 @@ import { BaseDbService } from '../commons/services/base-db.service';
|
|||||||
import { Repository } from 'typeorm';
|
import { Repository } from 'typeorm';
|
||||||
import { CreateProjectInput } from './dtos/create-project.input';
|
import { CreateProjectInput } from './dtos/create-project.input';
|
||||||
import { Project } from './project.entity';
|
import { Project } from './project.entity';
|
||||||
|
import { UpdateProjectInput } from './dtos/update-project.input';
|
||||||
|
|
||||||
@Injectable()
|
@Injectable()
|
||||||
export class ProjectsService extends BaseDbService<Project> {
|
export class ProjectsService extends BaseDbService<Project> {
|
||||||
@ -24,7 +25,7 @@ export class ProjectsService extends BaseDbService<Project> {
|
|||||||
return await this.repository.save(this.repository.create(dto));
|
return await this.repository.save(this.repository.create(dto));
|
||||||
}
|
}
|
||||||
|
|
||||||
async update(id: string, dto: CreateProjectInput) {
|
async update(id: string, dto: UpdateProjectInput) {
|
||||||
await this.isDuplicateEntityForUpdate(id, dto);
|
await this.isDuplicateEntityForUpdate(id, dto);
|
||||||
const old = await this.findOne(id);
|
const old = await this.findOne(id);
|
||||||
return await this.repository.save(this.repository.merge(old, dto));
|
return await this.repository.save(this.repository.merge(old, dto));
|
||||||
|
@ -4,7 +4,7 @@ import { IsOptional, IsString, IsUUID } from 'class-validator';
|
|||||||
@InputType()
|
@InputType()
|
||||||
export class CheckoutInput {
|
export class CheckoutInput {
|
||||||
@IsUUID()
|
@IsUUID()
|
||||||
projectId: string;
|
pipelineId: string;
|
||||||
|
|
||||||
@IsString()
|
@IsString()
|
||||||
@IsOptional()
|
@IsOptional()
|
||||||
|
14
src/repos/list-logs.consumer.ts
Normal file
14
src/repos/list-logs.consumer.ts
Normal file
@ -0,0 +1,14 @@
|
|||||||
|
import { ReposService } from './repos.service';
|
||||||
|
import { Processor, Process } from '@nestjs/bull';
|
||||||
|
import { Job } from 'bull';
|
||||||
|
import { ListLogsOption } from './models/list-logs.options';
|
||||||
|
import { LIST_LOGS_TASK } from './repos.constants';
|
||||||
|
@Processor(LIST_LOGS_TASK)
|
||||||
|
export class ListLogsConsumer {
|
||||||
|
constructor(private readonly service: ReposService) {}
|
||||||
|
@Process()
|
||||||
|
async listLogs(job: Job<ListLogsOption>) {
|
||||||
|
const logs = await this.service.listLogs(job.data);
|
||||||
|
return logs;
|
||||||
|
}
|
||||||
|
}
|
5
src/repos/models/list-logs.options.ts
Normal file
5
src/repos/models/list-logs.options.ts
Normal file
@ -0,0 +1,5 @@
|
|||||||
|
import { Project } from '../../projects/project.entity';
|
||||||
|
export interface ListLogsOption {
|
||||||
|
project: Project;
|
||||||
|
branch: string;
|
||||||
|
}
|
3
src/repos/repos.constants.ts
Normal file
3
src/repos/repos.constants.ts
Normal file
@ -0,0 +1,3 @@
|
|||||||
|
export const LIST_LOGS_TASK = 'LIST_LOGS_TASK';
|
||||||
|
export const LIST_LOGS_PUB_SUB = 'LIST_LOGS_PUB_SUB';
|
||||||
|
export const LIST_LOGS_DONE = 'LIST_LOGS_DONE';
|
@ -5,9 +5,21 @@ import { ReposResolver } from './repos.resolver';
|
|||||||
import { ReposService } from './repos.service';
|
import { ReposService } from './repos.service';
|
||||||
import { ConfigModule } from '@nestjs/config';
|
import { ConfigModule } from '@nestjs/config';
|
||||||
import { ProjectsModule } from '../projects/projects.module';
|
import { ProjectsModule } from '../projects/projects.module';
|
||||||
|
import { BullModule } from '@nestjs/bull';
|
||||||
|
import { LIST_LOGS_TASK, LIST_LOGS_PUB_SUB } from './repos.constants';
|
||||||
|
import { PubSub } from 'graphql-subscriptions';
|
||||||
|
import { ListLogsConsumer } from './list-logs.consumer';
|
||||||
|
|
||||||
@Module({
|
@Module({
|
||||||
imports: [TypeOrmModule.forFeature([Project]), ConfigModule, ProjectsModule],
|
imports: [
|
||||||
providers: [ReposResolver, ReposService],
|
TypeOrmModule.forFeature([Project]),
|
||||||
|
ConfigModule,
|
||||||
|
ProjectsModule,
|
||||||
|
BullModule.registerQueue({
|
||||||
|
name: LIST_LOGS_TASK,
|
||||||
|
}),
|
||||||
|
],
|
||||||
|
providers: [ReposResolver, ReposService, ListLogsConsumer],
|
||||||
|
exports: [ReposService],
|
||||||
})
|
})
|
||||||
export class ReposModule {}
|
export class ReposModule {}
|
||||||
|
@ -1,37 +1,4 @@
|
|||||||
import { Args, Mutation, Query, Resolver } from '@nestjs/graphql';
|
import { 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 { CheckoutInput } from './dtos/checkout.input';
|
|
||||||
import { ProjectsService } from '../projects/projects.service';
|
|
||||||
|
|
||||||
@Resolver()
|
@Resolver()
|
||||||
export class ReposResolver {
|
export class ReposResolver {}
|
||||||
constructor(
|
|
||||||
private readonly service: ReposService,
|
|
||||||
private readonly projectService: ProjectsService,
|
|
||||||
) {}
|
|
||||||
@Query(() => LogList)
|
|
||||||
async listLogs(@Args('listLogsArgs') dto: ListLogsArgs) {
|
|
||||||
return await this.service.listLogs(dto);
|
|
||||||
}
|
|
||||||
@Query(() => BranchList)
|
|
||||||
async listBranches(
|
|
||||||
@Args('listBranchesArgs') dto: ListBranchesArgs,
|
|
||||||
): Promise<BranchList> {
|
|
||||||
return await this.service.listBranches(dto).then((data) => {
|
|
||||||
return {
|
|
||||||
...data,
|
|
||||||
branches: Object.values(data.branches),
|
|
||||||
};
|
|
||||||
});
|
|
||||||
}
|
|
||||||
@Mutation(() => Boolean)
|
|
||||||
async checkout(@Args('checkoutInput') dto: CheckoutInput): Promise<true> {
|
|
||||||
const project = await this.projectService.findOne(dto.projectId);
|
|
||||||
await this.service.checkoutCommit(project, dto.commitNumber);
|
|
||||||
return true;
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
@ -1,12 +1,14 @@
|
|||||||
|
import { Pipeline } from './../pipelines/pipeline.entity';
|
||||||
import { Test, TestingModule } from '@nestjs/testing';
|
import { Test, TestingModule } from '@nestjs/testing';
|
||||||
import { getRepositoryToken } from '@nestjs/typeorm';
|
import { getRepositoryToken } from '@nestjs/typeorm';
|
||||||
import { Project } from '../projects/project.entity';
|
import { Project } from '../projects/project.entity';
|
||||||
import { ReposService } from './repos.service';
|
import { ReposService } from './repos.service';
|
||||||
import { ConfigModule } from '@nestjs/config';
|
import { ConfigModule } from '@nestjs/config';
|
||||||
import { readFile, rm } from 'fs/promises';
|
import { rm } from 'fs/promises';
|
||||||
import { join } from 'path';
|
|
||||||
import configuration from '../commons/config/configuration';
|
import configuration from '../commons/config/configuration';
|
||||||
import { NotFoundException } from '@nestjs/common';
|
import { PipelineTask } from '../pipeline-tasks/pipeline-task.entity';
|
||||||
|
import { join } from 'path';
|
||||||
|
import { readFile } from 'fs/promises';
|
||||||
|
|
||||||
const getTest1Project = () =>
|
const getTest1Project = () =>
|
||||||
({
|
({
|
||||||
@ -61,7 +63,10 @@ describe('ReposService', () => {
|
|||||||
});
|
});
|
||||||
describe('listLogs', () => {
|
describe('listLogs', () => {
|
||||||
it('should be return logs', async () => {
|
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();
|
expect(result).toBeDefined();
|
||||||
}, 20_000);
|
}, 20_000);
|
||||||
});
|
});
|
||||||
@ -69,63 +74,69 @@ describe('ReposService', () => {
|
|||||||
it('should be return branches', async () => {
|
it('should be return branches', async () => {
|
||||||
const result = await service.listBranches({ projectId: '1' });
|
const result = await service.listBranches({ projectId: '1' });
|
||||||
expect(result).toBeDefined();
|
expect(result).toBeDefined();
|
||||||
}, 10_000);
|
}, 20_000);
|
||||||
});
|
});
|
||||||
describe('checkoutBranch', () => {
|
|
||||||
it('should be checkout', async () => {
|
describe.skip('checkout', () => {
|
||||||
await service.checkoutBranch(getTest1Project(), 'master');
|
let task: PipelineTask;
|
||||||
const filePath = join(
|
let workspaceRoot: string;
|
||||||
service.getWorkspaceRoot(getTest1Project(), 'master'),
|
beforeEach(() => {
|
||||||
'README.md',
|
const project = new Project();
|
||||||
);
|
const pipeline = new Pipeline();
|
||||||
const text = await readFile(filePath, { encoding: 'utf-8' });
|
task = new PipelineTask();
|
||||||
expect(text).toMatch(/Commit 1/gi);
|
pipeline.project = project;
|
||||||
}, 30_000);
|
task.pipeline = pipeline;
|
||||||
it('multiplexing workspace', async () => {
|
project.id = 'pid';
|
||||||
await service.checkoutBranch(getTest1Project(), 'master');
|
project.name = 'pname';
|
||||||
await service.checkoutBranch(getTest1Project(), 'branch-a');
|
pipeline.id = 'lid';
|
||||||
await service.checkoutBranch(getTest1Project(), 'branch-b');
|
pipeline.name = 'pipeline';
|
||||||
const filePath = join(
|
task.id = 'tid';
|
||||||
service.getWorkspaceRoot(getTest1Project(), 'branch-b'),
|
task.commit = '123123hash';
|
||||||
'branch-b.md',
|
workspaceRoot = service.getWorkspaceRootByTask(task);
|
||||||
);
|
|
||||||
const text = await readFile(filePath, { encoding: 'utf-8' });
|
|
||||||
expect(text).toMatch(/Commit branch b/gi);
|
|
||||||
}, 30_000);
|
|
||||||
it('nonexistent branch', async () => {
|
|
||||||
return expect(
|
|
||||||
service.checkoutBranch(getTest1Project(), 'nonexistent'),
|
|
||||||
).rejects.toBeInstanceOf(NotFoundException);
|
|
||||||
}, 30_000);
|
|
||||||
it('checkout the specified version', async () => {
|
|
||||||
await service.checkoutBranch(getTest1Project(), 'master');
|
|
||||||
const filePath = join(
|
|
||||||
service.getWorkspaceRoot(getTest1Project(), 'master'),
|
|
||||||
'README.md',
|
|
||||||
);
|
|
||||||
const text = await readFile(filePath, { encoding: 'utf-8' });
|
|
||||||
expect(text).toMatch(/Commit 1/gi);
|
|
||||||
}, 30_000);
|
|
||||||
});
|
});
|
||||||
|
|
||||||
describe('checkoutCommit', () => {
|
|
||||||
it('should be checkout', async () => {
|
it('should be checkout', async () => {
|
||||||
await service.checkoutCommit(getTest1Project(), '498c782685');
|
task.commit = '498c782685';
|
||||||
const filePath = join(
|
await service.checkout(task, workspaceRoot);
|
||||||
service.getWorkspaceRoot(getTest1Project(), '498c782685'),
|
const filePath = join(workspaceRoot, 'README.md');
|
||||||
'README.md',
|
|
||||||
);
|
|
||||||
const text = await readFile(filePath, { encoding: 'utf-8' });
|
const text = await readFile(filePath, { encoding: 'utf-8' });
|
||||||
expect(text).toMatch(/Commit 1/gi);
|
expect(text).toMatch(/Commit 1/gi);
|
||||||
}, 20_000);
|
}, 20_000);
|
||||||
it('should be checkout right commit', async () => {
|
it('should be checkout right commit', async () => {
|
||||||
await service.checkoutCommit(getTest1Project(), '7f7123fe5b');
|
task.commit = '7f7123fe5b';
|
||||||
const filePath = join(
|
await service.checkout(task, workspaceRoot);
|
||||||
service.getWorkspaceRoot(getTest1Project(), '7f7123fe5b'),
|
const filePath = join(workspaceRoot, 'README.md');
|
||||||
'README.md',
|
|
||||||
);
|
|
||||||
const text = await readFile(filePath, { encoding: 'utf-8' });
|
const text = await readFile(filePath, { encoding: 'utf-8' });
|
||||||
expect(text).toMatch(/(?!Commit 1)/gi);
|
expect(text).toMatch(/(?!Commit 1)/gi);
|
||||||
}, 20_000);
|
}, 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$/,
|
||||||
|
);
|
||||||
|
});
|
||||||
});
|
});
|
||||||
});
|
});
|
||||||
|
@ -1,3 +1,6 @@
|
|||||||
|
import { ListLogsOption } from './models/list-logs.options';
|
||||||
|
import { Pipeline } from './../pipelines/pipeline.entity';
|
||||||
|
import { PipelineTask } from './../pipeline-tasks/pipeline-task.entity';
|
||||||
import { Injectable, NotFoundException } from '@nestjs/common';
|
import { Injectable, NotFoundException } from '@nestjs/common';
|
||||||
import { InjectRepository } from '@nestjs/typeorm';
|
import { InjectRepository } from '@nestjs/typeorm';
|
||||||
import { F_OK } from 'constants';
|
import { F_OK } from 'constants';
|
||||||
@ -11,6 +14,7 @@ import { ListLogsArgs } from './dtos/list-logs.args';
|
|||||||
import { ConfigService } from '@nestjs/config';
|
import { ConfigService } from '@nestjs/config';
|
||||||
|
|
||||||
const DEFAULT_REMOTE_NAME = 'origin';
|
const DEFAULT_REMOTE_NAME = 'origin';
|
||||||
|
const INFO_PATH = '@info';
|
||||||
@Injectable()
|
@Injectable()
|
||||||
export class ReposService {
|
export class ReposService {
|
||||||
constructor(
|
constructor(
|
||||||
@ -19,44 +23,35 @@ export class ReposService {
|
|||||||
private readonly configService: ConfigService,
|
private readonly configService: ConfigService,
|
||||||
) {}
|
) {}
|
||||||
|
|
||||||
getWorkspaceRoot(project: Project, subDir = ''): string {
|
getWorkspaceRoot(project: Project): string {
|
||||||
return join(
|
return join(
|
||||||
this.configService.get<string>('workspaces.root'),
|
this.configService.get<string>('workspaces.root'),
|
||||||
project.name,
|
encodeURIComponent(project.name),
|
||||||
encodeURIComponent(subDir),
|
INFO_PATH,
|
||||||
);
|
);
|
||||||
}
|
}
|
||||||
|
|
||||||
async lockWorkspace(workspaceRoot: string) {
|
async getGit(project: Project, workspaceRoot?: string) {
|
||||||
// TODO: 获取锁,失败抛错。
|
if (!workspaceRoot) {
|
||||||
|
workspaceRoot = this.getWorkspaceRoot(project);
|
||||||
}
|
}
|
||||||
|
|
||||||
async getGit(project: Project, subDir = 'default') {
|
await access(workspaceRoot, F_OK).catch(async () => {
|
||||||
const workspaceRoot = this.getWorkspaceRoot(project, subDir);
|
|
||||||
await this.lockWorkspace(workspaceRoot);
|
|
||||||
|
|
||||||
const firstInit = await access(workspaceRoot, F_OK)
|
|
||||||
.then(() => false)
|
|
||||||
.catch(async () => {
|
|
||||||
await mkdir(workspaceRoot, { recursive: true });
|
await mkdir(workspaceRoot, { recursive: true });
|
||||||
return true;
|
|
||||||
});
|
});
|
||||||
const git = gitP(workspaceRoot);
|
const git = gitP(workspaceRoot);
|
||||||
if (firstInit) {
|
if (!(await git.checkIsRepo().catch(() => false))) {
|
||||||
await git.init();
|
await git.init();
|
||||||
await git.addRemote(DEFAULT_REMOTE_NAME, project.sshUrl);
|
await git.addRemote(DEFAULT_REMOTE_NAME, project.sshUrl);
|
||||||
}
|
}
|
||||||
|
await git.fetch();
|
||||||
return git;
|
return git;
|
||||||
}
|
}
|
||||||
|
|
||||||
async listLogs(dto: ListLogsArgs) {
|
async listLogs({ project, branch }: ListLogsOption) {
|
||||||
const project = await this.projectRepository.findOneOrFail({
|
|
||||||
id: dto.projectId,
|
|
||||||
});
|
|
||||||
const git = await this.getGit(project);
|
const git = await this.getGit(project);
|
||||||
await git.fetch();
|
|
||||||
return await git.log(
|
return await git.log(
|
||||||
dto.branch ? ['--branches', dto.branch, '--'] : ['--all'],
|
branch ? ['--branches', `remotes/origin/${branch}`, '--'] : ['--all'],
|
||||||
);
|
);
|
||||||
}
|
}
|
||||||
|
|
||||||
@ -68,26 +63,8 @@ export class ReposService {
|
|||||||
return git.branch();
|
return git.branch();
|
||||||
}
|
}
|
||||||
|
|
||||||
async checkoutBranch(project: Project, branch: string) {
|
async checkout(task: PipelineTask, workspaceRoot: string) {
|
||||||
const git = await this.getGit(project, branch);
|
const git = await this.getGit(task.pipeline.project, workspaceRoot);
|
||||||
try {
|
|
||||||
await git.fetch(DEFAULT_REMOTE_NAME, branch);
|
|
||||||
} catch (err) {
|
|
||||||
if (err.message.includes("couldn't find remote ref nonexistent")) {
|
|
||||||
throw new NotFoundException(err.message);
|
|
||||||
}
|
|
||||||
throw err;
|
|
||||||
}
|
|
||||||
await git.checkout([
|
|
||||||
'-B',
|
|
||||||
branch,
|
|
||||||
'--track',
|
|
||||||
`${DEFAULT_REMOTE_NAME}/${branch}`,
|
|
||||||
]);
|
|
||||||
}
|
|
||||||
|
|
||||||
async checkoutCommit(project: Project, commitNumber: string) {
|
|
||||||
const git = await this.getGit(project, commitNumber);
|
|
||||||
try {
|
try {
|
||||||
await git.fetch(DEFAULT_REMOTE_NAME);
|
await git.fetch(DEFAULT_REMOTE_NAME);
|
||||||
} catch (err) {
|
} catch (err) {
|
||||||
@ -96,6 +73,20 @@ export class ReposService {
|
|||||||
}
|
}
|
||||||
throw err;
|
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}`),
|
||||||
|
);
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
8
test/data/bad-work.js
Normal file
8
test/data/bad-work.js
Normal 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);
|
7
test/data/one-second-work.js
Normal file
7
test/data/one-second-work.js
Normal file
@ -0,0 +1,7 @@
|
|||||||
|
let timer;
|
||||||
|
let count = 0;
|
||||||
|
setTimeout(() => clearInterval(timer), 1_000);
|
||||||
|
|
||||||
|
timer = setInterval(() => {
|
||||||
|
console.log(++count * 10);
|
||||||
|
}, 95);
|
Loading…
Reference in New Issue
Block a user