🥭 Aggregation for jobResult

services/jobqueue.service.ts

@@ -1,7 +1,7 @@
-import { Context, Service, ServiceBroker, ServiceSchema } from "moleculer";
+import { Context, Service, ServiceBroker } from "moleculer";
 import JobResult from "../models/jobresult.model";
 import { refineQuery } from "filename-parser";
-import BullMqMixin, { BullMQAdapter, Queue } from "moleculer-bullmq";
+import BullMqMixin from "moleculer-bullmq";
 import { extractFromArchive } from "../utils/uncompression.utils";
 import { isNil, isUndefined } from "lodash";
 import { pubClient } from "../config/redis.config";
@@ -12,7 +12,6 @@ console.log(process.env.REDIS_URI);
 export default class JobQueueService extends Service {
 	public constructor(public broker: ServiceBroker) {
 		super(broker);
-
 		this.parseServiceSchema({
 			name: "jobqueue",
 			hooks: {},
@@ -168,6 +167,23 @@ export default class JobQueueService extends Service {
 						}
 					},
 				},
+				getJobResultStatistics: {
+					rest: "GET /getJobResultStatistics",
+					handler: async (ctx: Context<{}>) => {
+						const result = await JobResult.aggregate([
+							{
+								$group: {
+									_id: "$_id",
+									jobResults: { $addToSet: "status" }
+								},
+								
+							},
+							{ $sort: { timestamp: -1 } },
+							{ $skip: 0 },
+							{ $limit: 5 },
+						])
+					}
+				},
 			},
 
 			events: {
@@ -188,14 +204,13 @@ export default class JobQueueService extends Service {
 				async "enqueue.async.completed"(ctx: Context<{ id: Number }>) {
 					// 1. Fetch the job result using the job Id
 					const job = await this.job(ctx.params.id);
-					console.log(job);
 					// 2. Increment the completed job counter
 					await pubClient.incr("completedJobCount");
 					// 3. Fetch the completed job count for the final payload to be sent to the client
 					const completedJobCount = await pubClient.get("completedJobCount");
 					// 4. Emit the LS_COVER_EXTRACTED event with the necessary details
 					await this.broker.call("socket.broadcast", {
-						namespace: "/", //optional
+						namespace: "/",
 						event: "action",
 						args: [
 							{
@@ -226,6 +241,7 @@ export default class JobQueueService extends Service {
 						id: ctx.params.id,
 						status: "failed",
 						failedReason: job.failedReason,
+						timestamp: job.timestamp,
 					});
 
 					// 4. Emit the LS_COVER_EXTRACTION_FAILED event with the necessary details