.vscode/settings.json

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

package.json

@@ -40,7 +40,6 @@
     "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,5 +7,6 @@ export class CreatePipelineTaskInput {
 
   commit: string;
 
+  @Field(() => PipelineUnits)
   units: PipelineUnits[];
 }

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

@@ -1,8 +0,0 @@
-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,13 +1,6 @@
-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,13 +1,7 @@
 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,12 +1,7 @@
 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

@@ -1,25 +0,0 @@
-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

@@ -1,50 +0,0 @@
-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,
-    private readonly logsService: PipelineTaskLogsService,
+    @InjectQueue(PIPELINE_TASK_LOG_QUEUE)
+    private readonly logQueue: Queue<PipelineTaskLogMessage>,
   ) {}
   @Process()
   async doTask({ data: task, update }: Job<PipelineTask>) {
@@ -68,14 +68,13 @@ 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);
-      await update(task);
+      update(task);
     }
   }
 
@@ -95,12 +94,12 @@ export class PipelineTaskConsumer {
         const str = data.toString();
         errorMessages.push(str);
         logs.push(str);
-        this.logsService.recordLog(PipelineTaskLogMessage.create(task, str));
+        this.logQueue.add(PipelineTaskLogMessage.create(task, str));
       });
       sub.stdout.on('data', (data: Buffer) => {
         const str = data.toString();
         logs.push(str);
-        this.logsService.recordLog(PipelineTaskLogMessage.create(task, str));
+        this.logQueue.add(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 { Field, ObjectType } from '@nestjs/graphql';
+import { 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({ nullable: true })
-  startedAt?: Date;
+  @Column()
+  startedAt: Date;
 
-  @Column({ nullable: true })
-  endedAt?: Date;
+  @Column()
+  endedAt: Date;
 }

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

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

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

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

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

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

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

@@ -33,11 +33,6 @@ 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) {