.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/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/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/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 logQueue: Queue<PipelineTaskLogMessage>,
+    private readonly logsService: PipelineTaskLogsService,
   ) {}
   @Process()
   async doTask({ data: task, update }: Job<PipelineTask>) {
@@ -68,14 +68,14 @@ export class PipelineTaskConsumer {
         } finally {
           unitLog.endedAt = new Date();
           task.logs.push(unitLog);
-          update(task);
+          // await update(task);
         }
       }
     } catch (err) {
       console.log(err);
     } finally {
       task = await this.service.updateTask(task);
-      update(task);
+      await update(task);
     }
   }
 
@@ -95,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-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

@@ -11,7 +11,10 @@ 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: [
@@ -29,6 +32,11 @@ import {
     PipelineTasksService,
     PipelineTasksResolver,
     PipelineTaskConsumer,
+    PipelineTaskLogsService,
+    {
+      provide: Symbol(PIPELINE_TASK_LOG_PUBSUB),
+      useValue: new PubSub(),
+    },
   ],
 })
 export class PipelineTasksModule {}

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

@@ -4,10 +4,14 @@ 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) {}
+  constructor(
+    private readonly service: PipelineTasksService,
+    private readonly logsService: PipelineTaskLogsService,
+  ) {}
 
   @Mutation(() => PipelineTask)
   async createPipelineTask(@Args('task') taskDto: CreatePipelineTaskInput) {
@@ -15,4 +19,9 @@ export class PipelineTasksResolver {
   }
 
   @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

@@ -36,6 +36,10 @@ export class PipelineTasksService {
     return task;
   }
 
+  async findTaskById(id: string) {
+    return await this.repository.findOneOrFail({ id });
+  }
+
   async doNextTask(pipeline: Pipeline) {
     const [lckKey, tasksKey] = this.getRedisTokens(pipeline);
     const redis = this.redis.getClient();