threetwo-core-service/services/jobqueue.service.ts

import { Context, Service, ServiceBroker, ServiceSchema } from "moleculer";
const { MoleculerError } = require("moleculer").Errors;
import JobResult from "../models/jobresult.model";
import { refineQuery } from "filename-parser";
import BullMqMixin, { BullMQAdapter, Queue } from "moleculer-bullmq";
import { extractFromArchive } from "../utils/uncompression.utils";
import { isNil, isUndefined } from "lodash";

console.log(process.env.REDIS_URI);
export default class JobQueueService extends Service {
	public constructor(public broker: ServiceBroker) {
		super(broker);

		this.parseServiceSchema({
			name: "jobqueue",
			hooks: {},
			mixins: [BullMqMixin],
			settings: {
				bullmq: {
					client: process.env.REDIS_URI,
				},
			},
			actions: {
				getJobStatuses: {
					rest: "GET /getJobStatuses",
					handler: async (ctx: Context<{}>) => {
						const foo = await this.getJobStatuses("enqueue.async");
						console.log(foo);
						return foo;
					},
				},
				enqueue: {
					queue: true,
					rest: "/GET enqueue",
					handler: async (ctx: Context<{}>) => {
						// Enqueue the job
						const job = await this.localQueue(
							ctx,
							"enqueue.async",
							ctx.params,
							{ priority: 10 }
						);
						console.log(`Job ${job.id} enqueued`);

						return job.id;
					},
				},
				// Comic Book Import Job Queue
				"enqueue.async": {
					handler: async (
						ctx: Context<{
							socketSessionId: String;
						}>
					) => {
						try {
							console.log(
								`Recieved Job ID ${ctx.locals.job.id}, processing...`
							);

							// 1. De-structure the job params
							const { fileObject } = ctx.locals.job.data.params;

							// 2. Extract metadata from the archive
							const result = await extractFromArchive(
								fileObject.filePath
							);
							const {
								name,
								filePath,
								fileSize,
								extension,
								mimeType,
								cover,
								containedIn,
								comicInfoJSON,
							} = result;

							// 3a. Infer any issue-related metadata from the filename
							const { inferredIssueDetails } = refineQuery(
								result.name
							);
							console.log(
								"Issue metadata inferred: ",
								JSON.stringify(inferredIssueDetails, null, 2)
							);

							// 3b. Orchestrate the payload
							const payload = {
								importStatus: {
									isImported: true,
									tagged: false,
									matchedResult: {
										score: "0",
									},
								},
								rawFileDetails: {
									name,
									filePath,
									fileSize,
									extension,
									mimeType,
									containedIn,
									cover,
								},
								inferredMetadata: {
									issue: inferredIssueDetails,
								},
								sourcedMetadata: {
									// except for ComicInfo.xml, everything else should be copied over from the
									// parent comic
									comicInfo: comicInfoJSON,
								},
								// since we already have at least 1 copy
								// mark it as not wanted by default
								"acquisition.source.wanted": false,

								// clear out the downloads array
								// "acquisition.directconnect.downloads": [],

								// mark the metadata source
								"acquisition.source.name":
									ctx.locals.job.data.params.sourcedFrom,
							};

							// 3c. Add the bundleId, if present to the payload
							let bundleId = null;
							if (!isNil(ctx.locals.job.data.params.bundleId)) {
								bundleId = ctx.locals.job.data.params.bundleId;
							}

							// 3d. Add the sourcedMetadata, if present
							if (
								!isNil(
									ctx.locals.job.data.params.sourcedMetadata
								) &&
								!isUndefined(
									ctx.locals.job.data.params.sourcedMetadata
										.comicvine
								)
							) {
								Object.assign(
									payload.sourcedMetadata,
									ctx.locals.job.data.params.sourcedMetadata
								);
							}

							// 4. write to mongo
							const importResult = await this.broker.call(
								"library.rawImportToDB",
								{
									importType:
										ctx.locals.job.data.params.importType,
									bundleId,
									payload,
								}
							);
							return {
								data: {
									importResult,
								},
								id: ctx.locals.job.id,
								socketSessionId: ctx.params.socketSessionId,
							};
						} catch (error) {
							console.error(
								`An error occurred processing Job ID ${ctx.locals.job.id}`
							);
							throw new MoleculerError(
								error,
								500,
								"IMPORT_JOB_ERROR",
								{ data: ctx.params.socketSessionId }
							);
						}
					},
				},
			},

			events: {
				// use the `${QUEUE_NAME}.QUEUE_EVENT` scheme
				async "enqueue.async.active"(ctx: Context<{ id: Number }>) {
					console.log(`Job ID ${ctx.params.id} is set to active.`);
				},

				async "enqueue.async.completed"(ctx: Context<{ id: Number }>) {
					const jobState = await this.job(ctx.params.id);

					await JobResult.create({
						id: ctx.params.id,
						status: "completed",
						failedReason: {},
					});
					console.log(`Job ID ${ctx.params.id} completed.`);
				},

				async "enqueue.async.failed"(ctx) {
					const jobState = await this.job(ctx.params.id);
					await JobResult.create({
						id: ctx.params.id,
						status: "failed",
						failedReason: jobState.failedReason,
					});
				},
			},
		});
	}
}