13 Commits

Author SHA1 Message Date
24a2f80e46 test: pass cases 2021-04-05 20:20:43 +08:00
0e0781c4c4 feat(pipeline-tasks): 任务更新推送时机修改。 2021-04-05 16:34:12 +08:00
a82f663354 feat: clean up. 2021-04-04 15:51:03 +08:00
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
46fb41f856 feat: 完善消息发布。 2021-04-04 12:20:08 +08:00
b4307f05d6 feat: 使用 PubSub 进行消息广播。 2021-04-04 00:36:58 +08:00
bb3efd3714 feat: pubsub base redis. 2021-04-03 19:19:02 +08:00
092cf9c418 Merge branch 'master' of ssh://git.ivanli.cc:7018/Fennec/fennec-be 2021-03-28 20:27:28 +08:00
039f4b6d15 test: pass case. 2021-03-28 20:27:20 +08:00
22be1ffb33 Merge branch 'master' of ssh://git.ivanli.cc:7018/Fennec/fennec-be 2021-03-28 19:38:18 +08:00
ef47f8049e fix(queue): 修复消息队列未使用配置的reids. 2021-03-28 19:38:09 +08:00
032aa89b05 feat(pipelines): list logs 时提供关联的 tasks 。 2021-03-28 19:37:16 +08:00
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
51 changed files with 4018 additions and 20012 deletions

View File

@ -1,8 +1,13 @@
{
"cSpell.words": [
"Repos",
"boardcat",
"gitea",
"lpush",
"lrange",
"metatype",
"pmessage",
"psubscribe",
"rpop",
"rpush"
]

22984
package-lock.json generated

File diff suppressed because it is too large Load Diff

View File

@ -10,8 +10,8 @@
"build": "nest build",
"format": "prettier --write \"src/**/*.ts\" \"test/**/*.ts\"",
"start": "nest start",
"start:dev": "DEBUG=fennec:*,simple-git:* 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",
@ -28,17 +28,18 @@
"@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/ramda": "^0.27.38",
"apollo-server-express": "^2.19.2",
"bcrypt": "^5.0.0",
"body-parser": "^1.19.0",
"bull": "^3.20.1",
"class-transformer": "^0.3.2",
"class-validator": "^0.13.1",
"debug": "^4.3.1",
"graphql": "^15.5.0",
"graphql-tools": "^7.0.2",
"ioredis": "^4.25.0",
"js-yaml": "^4.0.0",
"nestjs-redis": "^1.2.8",
"observable-to-async-generator": "^1.0.1-rc",
@ -51,11 +52,13 @@
"typeorm": "^0.2.30"
},
"devDependencies": {
"@nestjs/cli": "^7.5.1",
"@nestjs/cli": "^7.5.7",
"@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",

View File

@ -1,4 +1,4 @@
import { Module } from '@nestjs/common';
import { MiddlewareConsumer, Module, NestModule } from '@nestjs/common';
import { ConfigModule, ConfigService } from '@nestjs/config';
import { GraphQLModule } from '@nestjs/graphql';
import { TypeOrmModule } from '@nestjs/typeorm';
@ -11,6 +11,12 @@ 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/middlewares/raw-body.middleware';
import { GiteaWebhooksController } from './webhooks/gitea-webhooks.controller';
import { ParseBodyMiddleware } from './commons/middlewares/parse-body.middleware';
import { BullModule } from '@nestjs/bull';
import { PubSubModule } from './commons/pub-sub/pub-sub.module';
@Module({
imports: [
@ -41,6 +47,28 @@ import { RedisModule } from 'nestjs-redis';
}),
inject: [ConfigService],
}),
BullModule.forRootAsync({
imports: [ConfigModule],
useFactory: (configService: ConfigService) => ({
redis: {
host: configService.get<string>('db.redis.host', 'localhost'),
port: configService.get<number>('db.redis.port', undefined),
password: configService.get<string>('db.redis.password', undefined),
},
}),
inject: [ConfigService],
}),
PubSubModule.forRootAsync({
imports: [ConfigModule],
useFactory: (configService: ConfigService) => ({
redis: {
host: configService.get<string>('db.redis.host', 'localhost'),
port: configService.get<number>('db.redis.port', undefined),
password: configService.get<string>('db.redis.password', undefined),
},
}),
inject: [ConfigService],
}),
ProjectsModule,
ReposModule,
PipelinesModule,
@ -55,8 +83,17 @@ import { RedisModule } from 'nestjs-redis';
}),
inject: [ConfigService],
}),
WebhooksModule,
],
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,10 @@
import { Module } from '@nestjs/common';
import { PasswordConverter } from './services/password-converter';
import { PubSubModule } from './pub-sub/pub-sub.module';
@Module({
providers: [PasswordConverter],
exports: [PasswordConverter],
imports: [PubSubModule],
})
export class CommonsModule {}

View File

@ -3,25 +3,49 @@ import {
Catch,
ArgumentsHost,
HttpException,
HttpStatus,
} from '@nestjs/common';
import { ApolloError } from 'apollo-server-errors';
@Catch(HttpException)
export class HttpExceptionFilter implements ExceptionFilter {
catch(exception: HttpException, host: ArgumentsHost) {
const message = 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;
switch (host.getType<'http' | 'graphql' | string>()) {
case 'graphql': {
const message = 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;
}
return new ApolloError(
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;
response.status(status).json({
statusCode: status,
message: exception.message,
timestamp: new Date().toISOString(),
path: request.url,
});
}
default:
throw exception;
}
return new ApolloError(
message,
exception.getStatus().toString(),
extensions,
);
}
}

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

@ -1,15 +1,25 @@
import { ArgumentMetadata, Injectable, PipeTransform } from '@nestjs/common';
import { sanitize } from '@neuralegion/class-sanitizer/dist';
import { plainToClass } from 'class-transformer';
@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);
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;
}
return value;
}
}

View File

@ -0,0 +1,5 @@
import { Inject } from '@nestjs/common';
import { getPubSubToken } from '../utils/token';
export const InjectPubSub = (name?: string): ParameterDecorator =>
Inject(getPubSubToken(name));

View File

@ -0,0 +1,8 @@
import { ModuleMetadata } from '@nestjs/common';
import { PubSubOptions } from './pub-sub-options.interface';
export interface PubSubAsyncConfig extends Pick<ModuleMetadata, 'imports'> {
name?: string;
useFactory: (...args: any[]) => Promise<PubSubOptions> | PubSubOptions;
inject?: any[];
}

View File

@ -0,0 +1,6 @@
import { RedisOptions } from 'ioredis';
export interface PubSubOptions {
name?: string;
redis: RedisOptions;
}

View File

@ -0,0 +1,15 @@
export type PubSubRawMessage<T> =
| PubSubRawNextMessage<T>
| PubSubRawErrorMessage<T>
| PubSubRawCompleteMessage<T>;
export interface PubSubRawNextMessage<T> {
type: 'next';
message: T;
}
export interface PubSubRawErrorMessage<T> {
type: 'error';
error: string;
}
export interface PubSubRawCompleteMessage<T> {
type: 'complete';
}

View File

@ -0,0 +1 @@
export const DEFAULT_PUB_SUB_NAME = 'default';

View File

@ -0,0 +1,48 @@
import { DynamicModule, Module } from '@nestjs/common';
import { PubSubService } from './pub-sub.service';
import {
createAsyncPubSubProviders,
createPubSubProvider,
} from './pub-sub.providers';
import { PubSubOptions } from './interfaces/pub-sub-options.interface';
import { PubSubAsyncConfig } from './interfaces/pub-sub-async-config.interface';
import { getPubSubConfigToken } from './utils/token';
@Module({
providers: [PubSubService],
})
export class PubSubModule {
public static forRoot(options: PubSubOptions): DynamicModule {
const providers = [
{
provide: getPubSubConfigToken(options.name),
useValue: options,
},
createPubSubProvider(options.name),
];
return {
global: true,
module: PubSubModule,
providers,
exports: providers,
};
}
public static forRootAsync(...configs: PubSubAsyncConfig[]): DynamicModule {
const providers = createAsyncPubSubProviders(configs);
return {
global: true,
module: PubSubModule,
imports: configs
.map((config) => config.imports)
.flat()
.filter((o, i, a) => a.indexOf(o) === i),
providers,
exports: providers,
};
}
public static forFeature(): DynamicModule {
return {
module: PubSubModule,
};
}
}

View File

@ -0,0 +1,31 @@
import { Provider } from '@nestjs/common';
import { PubSubAsyncConfig } from './interfaces/pub-sub-async-config.interface';
import { PubSubOptions } from './interfaces/pub-sub-options.interface';
import { PubSub } from './pub-sub';
import { getPubSubConfigToken, getPubSubToken } from './utils/token';
export function createPubSubProvider(name: string): Provider {
return {
provide: getPubSubToken(name),
useFactory: (option: PubSubOptions) => new PubSub(option),
inject: [getPubSubConfigToken(name)],
};
}
export function createOptionsProvider(config: PubSubAsyncConfig): Provider {
return {
provide: getPubSubConfigToken(config.name),
useFactory: config.useFactory,
inject: config.inject || [],
};
}
export function createAsyncPubSubProviders(
configs: PubSubAsyncConfig[],
): Provider[] {
return configs
.map((config) => [
createOptionsProvider(config),
createPubSubProvider(config.name),
])
.flat();
}

View File

@ -0,0 +1,18 @@
import { Test, TestingModule } from '@nestjs/testing';
import { PubSubService } from './pub-sub.service';
describe('PubsubService', () => {
let service: PubSubService;
beforeEach(async () => {
const module: TestingModule = await Test.createTestingModule({
providers: [PubSubService],
}).compile();
service = module.get<PubSubService>(PubSubService);
});
it('should be defined', () => {
expect(service).toBeDefined();
});
});

View File

@ -0,0 +1,9 @@
import { Injectable } from '@nestjs/common';
import { PubSubOptions } from './interfaces/pub-sub-options.interface';
@Injectable()
export class PubSubService {
private options = new Map<string, PubSubOptions>();
private pubClient;
private;
}

View File

@ -0,0 +1,87 @@
import debug from 'debug';
import { tap } from 'rxjs/operators';
import { PubSub } from './pub-sub';
debug.enable('app:pubsub:*');
describe('PubSub', () => {
let instance: PubSub;
beforeEach(async () => {
instance = new PubSub({
name: 'default',
redis: {
host: 'localhost',
},
});
});
it('should be defined', () => {
expect(instance).toBeDefined();
});
it('should can send and receive message', async () => {
const arr = new Array(10)
.fill(undefined)
.map(() => Math.random().toString(36).slice(2, 8));
const results: string[] = [];
instance
.message$<string>('test')
.pipe(
tap((val) => {
console.log(val);
}),
)
.subscribe((val) => results.push(val));
await new Promise((r) => setTimeout(r, 1000));
await Promise.all([...arr.map((str) => instance.publish('test', str))]);
await new Promise((r) => setTimeout(r, 1000));
expect(results).toMatchObject(arr);
});
it('should complete', async () => {
const arr = new Array(10)
.fill(undefined)
.map(() => Math.random().toString(36).slice(2, 8));
const results: string[] = [];
instance
.message$<string>('test')
.pipe(
tap((val) => {
console.log(val);
}),
)
.subscribe((val) => results.push(val));
await new Promise((r) => setTimeout(r, 1000));
await Promise.all([...arr.map((str) => instance.publish('test', str))]);
await instance.finish('test');
await Promise.all([...arr.map((str) => instance.publish('test', str))]);
await new Promise((r) => setTimeout(r, 1000));
expect(results).toMatchObject(arr);
});
it('should error', async () => {
const arr = new Array(10)
.fill(undefined)
.map(() => Math.random().toString(36).slice(2, 8));
const results: string[] = [];
let error: string;
instance
.message$<string>('test')
.pipe(
tap((val) => {
console.log(val);
}),
)
.subscribe({
next: (val) => results.push(val),
error: (err) => (error = err.message),
});
await new Promise((r) => setTimeout(r, 1000));
await Promise.all([...arr.map((str) => instance.publish('test', str))]);
await instance.throwError('test', 'TEST ERROR MESSAGE');
await Promise.all([...arr.map((str) => instance.publish('test', str))]);
await new Promise((r) => setTimeout(r, 1000));
expect(results).toMatchObject(arr);
expect(error).toEqual('TEST ERROR MESSAGE');
});
});

View File

@ -0,0 +1,115 @@
import debug from 'debug';
import { EventEmitter } from 'events';
import IORedis, { Redis } from 'ioredis';
import { from, fromEvent, Observable } from 'rxjs';
import { filter, map, share, switchMap, takeWhile, tap } from 'rxjs/operators';
import { ApplicationException } from '../exceptions/application.exception';
import { PubSubOptions } from './interfaces/pub-sub-options.interface';
import {
PubSubRawMessage,
PubSubRawNextMessage,
} from './interfaces/pub-sub-raw-message.interface';
const log = debug('fennec:pubsub:instance');
export class PubSub extends EventEmitter {
pubRedis: Redis;
pSubRedis: Redis;
channels: string[] = [];
event$: Observable<[string, string, any]>;
constructor(private readonly options: PubSubOptions) {
super();
this.pubRedis = new IORedis(this.options.redis);
this.pSubRedis = new IORedis(this.options.redis);
this.pSubRedis.on('pmessage', (...args) =>
log.extend('raw')('%s %s %o', ...args),
);
this.event$ = fromEvent<[string, string, string]>(
this.pSubRedis,
'pmessage',
).pipe(
map((ev) => {
try {
ev[2] = JSON.parse(ev[2]);
} catch (err) {
log('WARN: is not json');
return null;
}
log('on message: %s %s %o', ...ev);
return ev;
}),
filter((v) => !!v),
share(),
);
}
async disconnect() {
log('disconnecting to redis...');
this.pubRedis.disconnect();
this.pSubRedis.disconnect();
log('disconnected');
}
async registerChannel(channel: string) {
if (this.channels.includes(channel)) {
return;
}
this.channels.push(channel);
return await this.pSubRedis.psubscribe(channel);
}
private async redisPublish<T>(channel: string, message: PubSubRawMessage<T>) {
log.extend('publish')('channel: %s, message: %O', channel, message);
return await this.pubRedis.publish(channel, JSON.stringify(message));
}
async publish(channel: string, message: any): Promise<number> {
return await this.redisPublish(channel, {
type: 'next',
message,
});
}
async finish(channel: string): Promise<number> {
return await this.redisPublish(channel, {
type: 'complete',
});
}
async throwError(channel: string, error: string): Promise<number> {
return await this.redisPublish(channel, {
type: 'error',
error,
});
}
message$ = <T>(channel: string): Observable<T> => {
return from(this.registerChannel(channel))
.pipe(switchMap(() => this.event$))
.pipe(
filter(([pattern]) => pattern === channel),
tap(([pattern, channel, message]) => {
log.extend('subscribe')(
'channel: %s, match: %s, message: %O',
channel,
pattern,
message,
);
}),
takeWhile(([pattern, channel, message]) => {
if (pattern === channel) {
if (message.type === 'error') {
throw new ApplicationException(message.error);
}
return message.type !== 'complete';
}
return true;
}),
map((ev) => ev[2] as PubSubRawMessage<T>),
map((message: PubSubRawNextMessage<T>) => message.message),
);
};
}

View File

@ -0,0 +1,8 @@
import { DEFAULT_PUB_SUB_NAME } from '../pub-sub.constants';
export function getPubSubToken(name = DEFAULT_PUB_SUB_NAME) {
return `app:pub-usb:${name || DEFAULT_PUB_SUB_NAME}`;
}
export function getPubSubConfigToken(name = DEFAULT_PUB_SUB_NAME) {
return `app:pub-usb:config:${name || DEFAULT_PUB_SUB_NAME}`;
}

View File

@ -6,7 +6,7 @@ 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 SanitizePipe());
app.useGlobalPipes(

View File

@ -1,6 +1,7 @@
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 {
@ -9,6 +10,7 @@ export class PipelineTaskLogMessage {
@Field(() => PipelineUnits, { nullable: true })
unit?: PipelineUnits;
@Field()
@Type(() => Date)
time: Date;
@Field()
message: string;

View File

@ -1,6 +1,7 @@
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 {
@ -8,7 +9,9 @@ export class PipelineTaskLogs {
unit: PipelineUnits;
@Field(() => TaskStatuses)
status: TaskStatuses;
@Type(() => Date)
startedAt?: Date;
@Type(() => Date)
endedAt?: Date;
logs = '';
}

View File

@ -1,6 +1,7 @@
import { Test, TestingModule } from '@nestjs/testing';
import { PipelineTaskLogsService } from './pipeline-task-logs.service';
import { RedisService } from 'nestjs-redis';
import { getPubSubToken } from '../commons/pub-sub/utils/token';
describe('PipelineTaskLogsService', () => {
let service: PipelineTaskLogsService;
@ -13,6 +14,10 @@ describe('PipelineTaskLogsService', () => {
provide: RedisService,
useValue: {},
},
{
provide: getPubSubToken(),
useValue: {},
},
],
}).compile();

View File

@ -1,8 +1,9 @@
import { Injectable } from '@nestjs/common';
import { log } from 'console';
import { PubSub } from 'graphql-subscriptions';
import { observableToAsyncIterable } from 'graphql-tools';
import { RedisService } from 'nestjs-redis';
import { find, omit, propEq } from 'ramda';
import { InjectPubSub } from '../commons/pub-sub/decorators/inject-pub-sub.decorator';
import { PubSub } from '../commons/pub-sub/pub-sub';
import { PipelineUnits } from './enums/pipeline-units.enum';
import { TaskStatuses } from './enums/task-statuses.enum';
import { PipelineTaskLogMessage } from './models/pipeline-task-log-message.module';
@ -13,9 +14,11 @@ const LOG_TIMEOUT_SECONDS = 10_000;
@Injectable()
export class PipelineTaskLogsService {
constructor(private readonly redisService: RedisService) {}
pubSub = new PubSub();
constructor(
private readonly redisService: RedisService,
@InjectPubSub()
private readonly pubSub: PubSub,
) {}
get redis() {
return this.redisService.getClient();
@ -73,6 +76,6 @@ export class PipelineTaskLogsService {
}
watchLogs(task: PipelineTask) {
return this.pubSub.asyncIterator(this.getKeys(task));
return observableToAsyncIterable(this.pubSub.message$(this.getKeys(task)));
}
}

View File

@ -174,7 +174,7 @@ describe('PipelineTaskConsumer', () => {
await consumer.doTask(job);
expect(updateTask).toHaveBeenCalledTimes(2);
expect(updateTask).toHaveBeenCalledTimes(4);
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);
@ -192,7 +192,7 @@ describe('PipelineTaskConsumer', () => {
await consumer.doTask(job);
expect(updateTask).toHaveBeenCalledTimes(2);
expect(updateTask).toHaveBeenCalledTimes(4);
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);
@ -211,7 +211,7 @@ describe('PipelineTaskConsumer', () => {
await consumer.doTask(job);
expect(runScript).toHaveBeenCalledTimes(1);
expect(updateTask).toHaveBeenCalledTimes(2);
expect(updateTask).toHaveBeenCalledTimes(4);
const taskDto: PipelineTask = updateTask.mock.calls[0][0];
expect(taskDto.logs).toHaveLength(2);
expect(taskDto.logs[0].status).toEqual(TaskStatuses.success);
@ -232,7 +232,7 @@ describe('PipelineTaskConsumer', () => {
await consumer.doTask(job);
expect(updateTask).toHaveBeenCalledTimes(2);
expect(updateTask).toHaveBeenCalledTimes(4);
const taskDto: PipelineTask = updateTask.mock.calls[0][0];
expect(taskDto.logs).toHaveLength(2);
expect(taskDto.logs[0].status).toEqual(TaskStatuses.success);

View File

@ -86,6 +86,7 @@ export class PipelineTaskConsumer {
taskLogs.find((tl) => tl.unit === unit.type)?.logs ?? '',
);
task.logs.push(unitLog);
task = await this.service.updateTask(task);
await job.update(task);
}
}

View File

@ -1,11 +1,16 @@
import { AppBaseEntity } from './../commons/entities/app-base-entity';
import { Field, ObjectType } from '@nestjs/graphql';
import { Column, Entity, ManyToOne } from 'typeorm';
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 {
@ -20,7 +25,7 @@ export class PipelineTask extends AppBaseEntity {
@Column({ type: 'enum', enum: PipelineUnits, array: true })
units: PipelineUnits[];
@Column({ type: 'jsonb', default: '[]' })
@Column({ type: 'jsonb', default: '[]', transformer: logsTransformer })
logs: PipelineTaskLogs[];
@Column({ type: 'enum', enum: TaskStatuses, default: TaskStatuses.pending })

View File

@ -1,3 +1,2 @@
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';

View File

@ -8,13 +8,9 @@ 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 { PIPELINE_TASK_QUEUE } from './pipeline-tasks.constants';
import { PipelineTaskLogsService } from './pipeline-task-logs.service';
import { PubSub } from 'apollo-server-express';
import { PubSubModule } from '../commons/pub-sub/pub-sub.module';
@Module({
imports: [
@ -22,6 +18,7 @@ import { PubSub } from 'apollo-server-express';
BullModule.registerQueue({
name: PIPELINE_TASK_QUEUE,
}),
PubSubModule.forFeature(),
RedisModule,
ReposModule,
],
@ -30,10 +27,7 @@ import { PubSub } from 'apollo-server-express';
PipelineTasksResolver,
PipelineTaskConsumer,
PipelineTaskLogsService,
{
provide: Symbol(PIPELINE_TASK_LOG_PUBSUB),
useValue: new PubSub(),
},
],
exports: [PipelineTasksService],
})
export class PipelineTasksModule {}

View File

@ -5,6 +5,7 @@ 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';
import { plainToClass } from 'class-transformer';
@Resolver()
export class PipelineTasksResolver {
@ -20,7 +21,8 @@ export class PipelineTasksResolver {
@Subscription(() => PipelineTaskLogMessage, {
resolve: (value) => {
return value;
const data = plainToClass(PipelineTaskLogMessage, value);
return data;
},
})
async pipelineTaskLog(@Args() args: PipelineTaskLogArgs) {

View File

@ -10,6 +10,7 @@ import { EntityNotFoundError } from 'typeorm/error/EntityNotFoundError';
import { Repository } from 'typeorm';
import { Queue } from 'bull';
import { LockFailedException } from '../commons/exceptions/lock-failed.exception';
import { getPubSubToken } from '../commons/pub-sub/utils/token';
describe('PipelineTasksService', () => {
let service: PipelineTasksService;
@ -68,6 +69,10 @@ describe('PipelineTasksService', () => {
getClient: jest.fn(() => redisClient),
},
},
{
provide: getPubSubToken(),
useValue: {},
},
],
}).compile();

View File

@ -9,16 +9,17 @@ 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';
import { TaskStatuses } from './enums/task-statuses.enum';
import { isNil } from 'ramda';
import debug from 'debug';
import { InjectPubSub } from '../commons/pub-sub/decorators/inject-pub-sub.decorator';
import { PubSub } from '../commons/pub-sub/pub-sub';
import { observableToAsyncIterable } from '@graphql-tools/utils';
const log = debug('fennec:pipeline-tasks:service');
@Injectable()
export class PipelineTasksService {
pubSub = new PubSub();
constructor(
@InjectRepository(PipelineTask)
private readonly repository: Repository<PipelineTask>,
@ -27,6 +28,8 @@ export class PipelineTasksService {
@InjectQueue(PIPELINE_TASK_QUEUE)
private readonly queue: Queue<PipelineTask>,
private readonly redis: RedisService,
@InjectPubSub()
private readonly pubSub: PubSub,
) {}
async addTask(dto: CreatePipelineTaskInput) {
const pipeline = await this.pipelineRepository.findOneOrFail({
@ -69,6 +72,10 @@ export class PipelineTasksService {
return await this.repository.find({ pipelineId });
}
async listTasksByCommitHash(hash: string) {
return await this.repository.find({ commit: hash });
}
async doNextTask(pipeline: Pipeline) {
const [lckKey, tasksKey] = this.getRedisTokens(pipeline);
const redis = this.redis.getClient();
@ -114,12 +121,14 @@ export class PipelineTasksService {
}
async updateTask(task: PipelineTask) {
this.pubSub.publish(task.id, task);
this.pubSub.publish(`pipeline-task:${task.id}`, task);
return await this.repository.save(task);
}
async watchTaskUpdated(id: string) {
return this.pubSub.asyncIterator(id);
return observableToAsyncIterable(
this.pubSub.message$(`pipeline-task:${id}`),
);
}
getRedisTokens(pipeline: Pipeline): [string, string] {

View File

@ -0,0 +1,30 @@
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: {},
},
],
}).compile();
resolver = module.get<CommitLogsResolver>(CommitLogsResolver);
});
it('should be defined', () => {
expect(resolver).toBeDefined();
});
});

View File

@ -0,0 +1,33 @@
import {
Args,
Parent,
ResolveField,
Resolver,
Subscription,
} from '@nestjs/graphql';
import { PipelineTasksService } from '../pipeline-tasks/pipeline-tasks.service';
import { LogFields, LogList } from '../repos/dtos/log-list.model';
import { PipelinesService } from './pipelines.service';
@Resolver(() => LogFields)
export class CommitLogsResolver {
constructor(
private readonly service: PipelinesService,
private readonly taskServices: PipelineTasksService,
) {}
@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();
})();
}
@ResolveField()
async tasks(@Parent() commit: LogFields) {
return await this.taskServices.listTasksByCommitHash(commit.hash);
}
}

View File

@ -5,6 +5,8 @@ import { TypeOrmModule } from '@nestjs/typeorm';
import { Pipeline } from './pipeline.entity';
import { BullModule } from '@nestjs/bull';
import { LIST_LOGS_TASK } from '../repos/repos.constants';
import { CommitLogsResolver } from './commit-logs.resolver';
import { PipelineTasksModule } from '../pipeline-tasks/pipeline-tasks.module';
@Module({
imports: [
@ -12,7 +14,8 @@ import { LIST_LOGS_TASK } from '../repos/repos.constants';
BullModule.registerQueue({
name: LIST_LOGS_TASK,
}),
PipelineTasksModule,
],
providers: [PipelinesResolver, PipelinesService],
providers: [PipelinesResolver, PipelinesService, CommitLogsResolver],
})
export class PipelinesModule {}

View File

@ -1,10 +1,9 @@
import { Args, Mutation, Query, Resolver, Subscription } from '@nestjs/graphql';
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 { LogList } from '../repos/dtos/log-list.model';
@Resolver()
export class PipelinesResolver {
@ -42,16 +41,4 @@ export class PipelinesResolver {
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();
})();
}
}

View File

@ -1,5 +1,6 @@
import { ObjectType, Field } from '@nestjs/graphql';
import { LogResult, DefaultLogFields } from 'simple-git';
import { PipelineTask } from '../../pipeline-tasks/pipeline-task.entity';
@ObjectType()
export class LogFields {
@ -10,6 +11,7 @@ export class LogFields {
body: string;
author_name: string;
author_email: string;
tasks: PipelineTask[];
}
@ObjectType()

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