add tasks and jobs

Author: tacxou

src/_common/abstracts/schemas/abstract.schema.ts

@@ -3,8 +3,8 @@ import { Types, Document } from 'mongoose';
 import { MetadataPart, MetadataPartSchema } from '~/_common/abstracts/schemas/parts/metadata.part.schema';
 
 export abstract class AbstractSchema extends Document {
-  public readonly _id: Types.ObjectId | any // eslint-disable-line
+  public readonly _id?: Types.ObjectId | any // eslint-disable-line
 
   @Prop({ type: MetadataPartSchema })
-  public metadata: MetadataPart;
+  public metadata?: MetadataPart;
 }

src/core/backends/_interfaces/execute-job-options.interface.ts

@@ -1,9 +1,11 @@
 import { JobsOptions } from 'bullmq';
+import { Types } from 'mongoose';
 
 export interface ExecuteJobOptions {
   job?: JobsOptions;
   async?: boolean;
   syncTimeout?: number;
   timeoutDiscard?: boolean;
   comment?: string;
+  task?: Types.ObjectId;
 }

src/core/backends/backends.controller.ts

@@ -19,6 +19,7 @@ import { Public } from '~/_common/decorators/public.decorator';
 import { ExecuteJobDto } from './_dto/execute-job.dto';
 import { BackendsService } from './backends.service';
 import { SyncIdentitiesDto } from './_dto/sync-identities.dto';
+import { Types } from 'mongoose';
 
 function fireMessage(observer: Subscriber<MessageEvent>, channel: string, message: any, loggername: string) {
   try {
@@ -72,11 +73,16 @@ export class BackendsController {
   ): Promise<Response> {
     const async = /true|on|yes|1/i.test(asyncQuery);
     const timeoutDiscard = /true|on|yes|1/i.test(timeoutDiscardQuery);
-    const [job, response] = await this.backendsService.executeJob(body.action, body.id, body.payload, {
-      async,
-      syncTimeout,
-      timeoutDiscard,
-    });
+    const [job, response] = await this.backendsService.executeJob(
+      body.action,
+      new Types.ObjectId(body.id),
+      body.payload,
+      {
+        async,
+        syncTimeout,
+        timeoutDiscard,
+      },
+    );
     return res.status(HttpStatus.ACCEPTED).json({ async, job, response });
   }
 

src/core/backends/backends.service.ts

@@ -11,6 +11,7 @@ import { JobsService } from '../jobs/jobs.service';
 import { TasksService } from '../tasks/tasks.service';
 import { ActionType } from './_enum/action-type.enum';
 import { ExecuteJobOptions } from './_interfaces/execute-job-options.interface';
+import { Tasks } from '../tasks/_schemas/tasks.schema';
 
 const DEFAULT_SYNC_TIMEOUT = 30_000;
 
@@ -28,7 +29,6 @@ export class BackendsService extends AbstractQueueProcessor {
   public async onModuleInit() {
     await super.onModuleInit();
 
-    //TODO: resync failed + completed
     const jobsCompleted = await this._queue.getCompleted();
     for (const job of jobsCompleted) {
       const isSyncedJob = await this.jobsService.model.findOneAndUpdate<Jobs>(
@@ -43,6 +43,11 @@ export class BackendsService extends AbstractQueueProcessor {
         { new: true },
       );
       if (isSyncedJob) {
+        await this.identitiesService.model.findByIdAndUpdate(isSyncedJob.concernedTo, {
+          $set: {
+            state: IdentityState.SYNCED,
+          },
+        });
         this.logger.warn(`Job already completed, syncing... [${job.id}::COMPLETED]`);
       }
     }
@@ -64,7 +69,7 @@ export class BackendsService extends AbstractQueueProcessor {
 
     this.queueEvents.on('failed', async (payload) => {
       this.logger.debug(`Job failed ! [${payload.jobId}]`);
-      await this.jobsService.model.findOneAndUpdate<Jobs>(
+      const failedJob = await this.jobsService.model.findOneAndUpdate<Jobs>(
         { jobId: payload.jobId, state: { $ne: JobState.COMPLETED } },
         {
           $set: {
@@ -75,10 +80,15 @@ export class BackendsService extends AbstractQueueProcessor {
         },
         { new: true },
       );
+      await this.identitiesService.model.findByIdAndUpdate(failedJob.concernedTo, {
+        $set: {
+          state: IdentityState.ON_ERROR,
+        },
+      });
     });
 
     this.queueEvents.on('completed', async (payload) => {
-      await this.jobsService.model.findOneAndUpdate<Jobs>(
+      const completedJob = await this.jobsService.model.findOneAndUpdate<Jobs>(
         { jobId: payload.jobId, state: { $ne: JobState.COMPLETED } },
         {
           $set: {
@@ -87,8 +97,13 @@ export class BackendsService extends AbstractQueueProcessor {
             result: payload.returnvalue,
           },
         },
-        { upsert: true },
+        { upsert: true, new: true },
       );
+      await this.identitiesService.model.findByIdAndUpdate(completedJob.concernedTo, {
+        $set: {
+          state: IdentityState.SYNCED,
+        },
+      });
       this.logger.log(`Job completed... Syncing [${payload.jobId}]`);
     });
   }
@@ -114,9 +129,21 @@ export class BackendsService extends AbstractQueueProcessor {
       });
     }
 
+    const task: Tasks = await this.tasksService.create<Tasks>({
+      jobs: identities.map((identity) => identity.identity._id),
+    });
+
     const result = {};
     for (const identity of identities) {
-      const [executedJob] = await this.executeJob(identity.action, identity.identity._id, { identity }, options);
+      const [executedJob] = await this.executeJob(
+        identity.action,
+        identity.identity._id,
+        { identity },
+        {
+          ...options,
+          task: task._id,
+        },
+      );
       result[identity.identity._id] = executedJob;
     }
     return result;
@@ -148,6 +175,7 @@ export class BackendsService extends AbstractQueueProcessor {
       params: payload,
       concernedTo: concernedTo,
       comment: options?.comment,
+      task: options?.task,
       state: JobState.CREATED,
       ...optionals,
     });
@@ -163,6 +191,11 @@ export class BackendsService extends AbstractQueueProcessor {
             result: response,
           },
         });
+        await this.identitiesService.model.findByIdAndUpdate(concernedTo, {
+          $set: {
+            state: IdentityState.SYNCED,
+          },
+        });
         return [jobStoreUpdated as unknown as Jobs, response];
       } catch (err) {
         error = err;
@@ -174,6 +207,11 @@ export class BackendsService extends AbstractQueueProcessor {
           result: error,
         },
       });
+      await this.identitiesService.model.findByIdAndUpdate(concernedTo, {
+        $set: {
+          state: IdentityState.ON_ERROR,
+        },
+      });
       if (options?.timeoutDiscard !== false) {
         job.discard();
         throw new BadRequestException({

src/core/jobs/_schemas/jobs.schema.ts

@@ -25,6 +25,9 @@ export class Jobs extends AbstractSchema {
   })
   public concernedTo?: Types.ObjectId;
 
+  @Prop({ type: Types.ObjectId })
+  public task?: Types.ObjectId;
+
   @Prop({ type: String })
   public comment?: string;
 

src/core/tasks/tasks.service.ts

@@ -1,12 +1,12 @@
 import { Injectable } from '@nestjs/common';
 import { InjectModel } from '@nestjs/mongoose';
 import { Model } from 'mongoose';
-import { AbstractService } from '~/_common/abstracts/abstract.service';
+import { AbstractServiceSchema } from '~/_common/abstracts/abstract.service.schema';
 import { Tasks } from './_schemas/tasks.schema';
 
 @Injectable()
-export class TasksService extends AbstractService {
-  constructor(@InjectModel(Tasks.name) protected _model: Model<Tasks>) {
+export class TasksService extends AbstractServiceSchema {
+  public constructor(@InjectModel(Tasks.name) protected _model: Model<Tasks>) {
     super();
   }
 }

src/management/passwd/passwd.service.ts

@@ -30,7 +30,7 @@ export class PasswdService extends AbstractService {
     const iv = crypto.randomBytes(12).toString('base64');
     const key = crypto.randomBytes(16).toString('hex');
     const cipher = crypto.createCipheriv('aes-256-gcm', key, iv);
-    const dataStruct = { uid: askToken.uid, mail: askToken.mail };
+    const dataStruct = { uid: askToken.id, mail: askToken.mail };
     let ciphertext = cipher.update(JSON.stringify(dataStruct), 'utf8', 'base64');
     ciphertext += cipher.final('base64');
     const tag = cipher.getAuthTag();