🔧 Refactoring the import queue process

services/importqueue.service.ts

@@ -34,12 +34,7 @@ SOFTWARE.
 "use strict";
 
 import { refineQuery } from "filename-parser";
-import {
+import { Context, Service, ServiceBroker, ServiceSchema } from "moleculer";
-	Context,
-	Service,
-	ServiceBroker,
-	ServiceSchema
-} from "moleculer";
 import BullMQMixin, { SandboxedJob } from "moleculer-bull";
 import { DbMixin } from "../mixins/db.mixin";
 import Comic from "../models/comic.model";
@@ -56,18 +51,14 @@ export default class QueueService extends Service {
 		schema: ServiceSchema<{}> = { name: "importqueue" }
 	) {
 		super(broker);
-		console.log(this.io);
 		this.parseServiceSchema({
 			name: "importqueue",
-			mixins: [
+			mixins: [BullMQMixin(REDIS_URI), DbMixin("comics", Comic)],
-				BullMQMixin(REDIS_URI),
-				DbMixin("comics", Comic),
-			],
 			settings: {},
 			hooks: {},
 			queues: {
 				"process.import": {
-					concurrency: 30,
+					concurrency: 1,
 					async process(job: SandboxedJob) {
 						console.info("New job received!", job.data);
 						console.info(`Processing queue...`);
@@ -130,11 +121,12 @@ export default class QueueService extends Service {
 								},
 							}
 						);
-						return Promise.resolve({
+						console.log("JAMA", dbImportResult);
+						return {
 							dbImportResult,
 							id: job.id,
 							worker: process.pid,
-						});
+						};
 					},
 				},
 			},
@@ -195,12 +187,11 @@ export default class QueueService extends Service {
 				await this.getQueue("process.import").on(
 					"completed",
 					async (job, res) => {
-						await this.broker.call('socket.broadcast', {
+						await this.broker.call("socket.broadcast", {
-							namespace: '/', //optional
+							namespace: "/", //optional
 							event: "action",
 							args: [{ type: "LS_COVER_EXTRACTED", result: res }], //optional
-
+						});
-						})
 						console.info(`Job with the id '${job.id}' completed.`);
 					}
 				);

services/library.service.ts

@@ -101,7 +101,7 @@ export default class ImportService extends Service {
 						klaw(path.resolve(COMICS_DIRECTORY))
 							// 1.1 Filter on .cb* extensions
 							.pipe(
-								through2.obj(function (item, enc, next) {
+								through2.obj(function(item, enc, next) {
 									let fileExtension = path.extname(item.path);
 									if (
 										[".cbz", ".cbr", ".cb7"].includes(
@@ -123,18 +123,20 @@ export default class ImportService extends Service {
 								let comicExists = await Comic.exists({
 									"rawFileDetails.name": `${path.basename(
 										item.path,
-										path.extname(item.path),
+										path.extname(item.path)
-										
 									)}`,
 								});
 								if (!comicExists) {
 									// 2. Send the extraction job to the queue
-									await broker.call("importqueue.processImport", {
+									await broker.call(
+										"importqueue.processImport",
+										{
 											fileObject: {
 												filePath: item.path,
 												fileSize: item.stats.size,
 											},
-									});
+										}
+									);
 								} else {
 									console.log(
 										"Comic already exists in the library."
@@ -169,45 +171,34 @@ export default class ImportService extends Service {
 							};
 						}>
 					) {
+						try {
 							let volumeDetails;
 							const comicMetadata = ctx.params;
-						console.log(comicMetadata);
+							console.log(JSON.stringify(comicMetadata, null, 4));
 							// When an issue is added from the search CV feature
 							if (
 								comicMetadata.sourcedMetadata.comicvine &&
 								!isNil(
-								comicMetadata.sourcedMetadata.comicvine.volume
+									comicMetadata.sourcedMetadata.comicvine
+										.volume
 								)
 							) {
 								volumeDetails = await this.broker.call(
 									"comicvine.getVolumes",
 									{
 										volumeURI:
-										comicMetadata.sourcedMetadata.comicvine
+											comicMetadata.sourcedMetadata
-											.volume.api_detail_url,
+												.comicvine.volume
+												.api_detail_url,
 									}
 								);
 								comicMetadata.sourcedMetadata.comicvine.volumeInformation =
 									volumeDetails.results;
 							}
-						return await Comic.create(
+							return await Comic.create(ctx.params);
-							ctx.params,
+						} catch (error) {
-							(error, data) => {
+							console.log(error);
-								if (data) {
-									return data;
-								} else if (error) {
-									console.log("data", data);
-									console.log("error", error);
-									throw new Errors.MoleculerError(
-										"Failed to import comic book",
-										400,
-										"IMS_FAILED_COMIC_BOOK_IMPORT",
-										error
-									);
 						}
-							}
-						);
-						
 					},
 				},
 				applyComicVineMetadata: {
@@ -552,16 +543,29 @@ export default class ImportService extends Service {
 							.drop()
 							.then(async (data) => {
 								console.info(data);
-								const coversFolderDeleteResult = fsExtra.emptyDirSync(
+								const coversFolderDeleteResult =
-									path.resolve(`${USERDATA_DIRECTORY}/covers`)
+									fsExtra.emptyDirSync(
+										path.resolve(
+											`${USERDATA_DIRECTORY}/covers`
+										)
 									);
-								const  expandedFolderDeleteResult = fsExtra.emptyDirSync(
+								const expandedFolderDeleteResult =
+									fsExtra.emptyDirSync(
 										path.resolve(
 											`${USERDATA_DIRECTORY}/expanded`
 										)
 									);
-								const eSIndicesDeleteResult = await ctx.broker.call("search.deleteElasticSearchIndices", {});
+								const eSIndicesDeleteResult =
-								return { data, coversFolderDeleteResult, expandedFolderDeleteResult, eSIndicesDeleteResult };
+									await ctx.broker.call(
+										"search.deleteElasticSearchIndices",
+										{}
+									);
+								return {
+									data,
+									coversFolderDeleteResult,
+									expandedFolderDeleteResult,
+									eSIndicesDeleteResult,
+								};
 							})
 							.catch((error) => error);
 					},