feat(pipeline-task): 流水线任务日志订阅接口。

.vscode/settings.json

@@ -2,6 +2,8 @@
   "cSpell.words": [
     "Repos",
     "lpush",
-    "rpop"
+    "lrange",
+    "rpop",
+    "rpush"
   ]
 }

package.json

@@ -40,6 +40,7 @@
     "graphql-tools": "^7.0.2",
     "js-yaml": "^4.0.0",
     "nestjs-redis": "^1.2.8",
+    "observable-to-async-generator": "^1.0.1-rc",
     "pg": "^8.5.1",
     "ramda": "^0.27.1",
     "reflect-metadata": "^0.1.13",

src/pipeline-tasks/dtos/create-pipeline-task.input.ts

@@ -7,6 +7,5 @@ export class CreatePipelineTaskInput {
 
   commit: string;
 
-  @Field(() => PipelineUnits)
   units: PipelineUnits[];
 }

src/pipeline-tasks/dtos/pipeline-task-log.args.ts

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

src/pipeline-tasks/enums/task-statuses.enum.ts

@@ -1,6 +1,13 @@
+import { registerEnumType } from '@nestjs/graphql';
+
 export enum TaskStatuses {
   success = 'success',
   failed = 'failed',
   working = 'working',
   pending = 'pending',
 }
+
+registerEnumType(TaskStatuses, {
+  name: 'TaskStatuses',
+  description: '任务状态',
+});

src/pipeline-tasks/models/pipeline-task-log-message.module.ts

@@ -1,7 +1,13 @@
 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)
+  unit: PipelineUnits;
   time: Date;
   message: string;
 

src/pipeline-tasks/models/pipeline-task-logs.model.ts

@@ -1,7 +1,12 @@
 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(() => PipelineUnits)
   status: TaskStatuses;
   startedAt?: Date;
   endedAt?: Date;

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

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

src/pipeline-tasks/pipeline-task-logs.service.ts

@@ -0,0 +1,50 @@
+import { Injectable } from '@nestjs/common';
+import { PubSub } from 'graphql-subscriptions';
+import { RedisService } from 'nestjs-redis';
+import { omit } from 'ramda';
+import { PipelineTaskLogMessage } from './models/pipeline-task-log-message.module';
+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 readLogs(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;
+      }),
+    );
+  }
+
+  watchLogs(task: PipelineTask) {
+    return this.pubSub.asyncIterator(this.getKeys(task));
+  }
+}

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

@@ -18,13 +18,13 @@ import { ApplicationException } from '../commons/exceptions/application.exceptio
 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,
-    @InjectQueue(PIPELINE_TASK_LOG_QUEUE)
+    private readonly logsService: PipelineTaskLogsService,
-    private readonly logQueue: Queue<PipelineTaskLogMessage>,
   ) {}
   @Process()
   async doTask({ data: task, update }: Job<PipelineTask>) {
@@ -68,13 +68,14 @@ export class PipelineTaskConsumer {
         } finally {
           unitLog.endedAt = new Date();
           task.logs.push(unitLog);
+          // await update(task);
         }
       }
     } catch (err) {
       console.log(err);
     } finally {
       task = await this.service.updateTask(task);
-      update(task);
+      await update(task);
     }
   }
 
@@ -94,12 +95,12 @@ export class PipelineTaskConsumer {
         const str = data.toString();
         errorMessages.push(str);
         logs.push(str);
-        this.logQueue.add(PipelineTaskLogMessage.create(task, str));
+        this.logsService.recordLog(PipelineTaskLogMessage.create(task, str));
       });
       sub.stdout.on('data', (data: Buffer) => {
         const str = data.toString();
         logs.push(str);
-        this.logQueue.add(PipelineTaskLogMessage.create(task, str));
+        this.logsService.recordLog(PipelineTaskLogMessage.create(task, str));
       });
       sub.addListener('close', (code) => {
         if (code === 0) {

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

@@ -1,5 +1,5 @@
 import { AppBaseEntity } from './../commons/entities/app-base-entity';
-import { ObjectType } from '@nestjs/graphql';
+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';
@@ -26,9 +26,9 @@ export class PipelineTask extends AppBaseEntity {
   @Column({ type: 'enum', enum: TaskStatuses, default: TaskStatuses.pending })
   status: TaskStatuses;
 
-  @Column()
+  @Column({ nullable: true })
-  startedAt: Date;
+  startedAt?: Date;
 
-  @Column()
+  @Column({ nullable: true })
-  endedAt: Date;
+  endedAt?: Date;
 }

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

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

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

@@ -7,10 +7,14 @@ 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: [
@@ -24,6 +28,15 @@ import {
     RedisModule,
     ReposModule,
   ],
-  providers: [PipelineTasksService, PipelineTasksResolver],
+  providers: [
+    PipelineTasksService,
+    PipelineTasksResolver,
+    PipelineTaskConsumer,
+    PipelineTaskLogsService,
+    {
+      provide: Symbol(PIPELINE_TASK_LOG_PUBSUB),
+      useValue: new PubSub(),
+    },
+  ],
 })
 export class PipelineTasksModule {}

src/pipeline-tasks/pipeline-tasks.resolver.ts

@@ -1,4 +1,27 @@
-import { Resolver } from '@nestjs/graphql';
+import { Resolver, Args, Mutation, Subscription } 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 {}
+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)
+  async pipelineTaskLog(@Args() args: PipelineTaskLogArgs) {
+    const task = await this.service.findTaskById(args.taskId);
+    const asyncIterator = this.logsService.watchLogs(task);
+    return asyncIterator;
+  }
+}

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

@@ -33,6 +33,11 @@ export class PipelineTasksService {
     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 doNextTask(pipeline: Pipeline) {