fix: remove ongoing download jobs if exist

Author: oae

src/server/queue/download.ts

@@ -1,15 +1,22 @@
 import { Prisma } from '@prisma/client';
 import { Job, Queue, Worker } from 'bullmq';
 import { sanitizer } from '../../utils';
+import { logger } from '../../utils/logging';
 import { prisma } from '../db/client';
-import { downloadChapter, getChapterFromLocal } from '../utils/mangal';
+import { downloadChapter, getChapterFromLocal, getMangaPath, removeManga } from '../utils/mangal';
 import { notificationQueue } from './notify';
 
 const mangaWithLibraryAndMetadata = Prisma.validator<Prisma.MangaArgs>()({
   include: { library: true, metadata: true },
 });
 
-export type MangaWithLibraryAndMetadata = Prisma.MangaGetPayload<typeof mangaWithLibraryAndMetadata>;
+type MangaWithLibraryAndMetadata = Prisma.MangaGetPayload<typeof mangaWithLibraryAndMetadata>;
+
+const mangaWithChaptersAndLibrary = Prisma.validator<Prisma.MangaArgs>()({
+  include: { chapters: true, library: true },
+});
+
+type MangaWithChaptersAndLibrary = Prisma.MangaGetPayload<typeof mangaWithChaptersAndLibrary>;
 export interface IDownloadWorkerData {
   manga: MangaWithLibraryAndMetadata;
   chapterIndex: number;
@@ -19,8 +26,19 @@ export const downloadWorker = new Worker(
   'downloadQueue',
   async (job: Job) => {
     const { chapterIndex, manga }: IDownloadWorkerData = job.data;
+    let filePath;
     try {
-      const filePath = await downloadChapter(manga.title, manga.source, chapterIndex, manga.library.path);
+      const mangaInDb = await prisma.manga.findUnique({ where: { id: manga.id } });
+      if (!mangaInDb) {
+        job.log(`Manga ${manga.title} with id ${manga.id} is removed from db.`);
+        downloadWorker.on('completed', async ({ id }) => {
+          if (id === job.id) {
+            await job.remove();
+          }
+        });
+        return;
+      }
+      filePath = await downloadChapter(manga.title, manga.source, chapterIndex, manga.library.path);
       const chapter = await getChapterFromLocal(filePath);
 
       await prisma.chapter.deleteMany({
@@ -46,6 +64,10 @@ export const downloadWorker = new Worker(
       await job.updateProgress(100);
     } catch (err) {
       await job.log(`${err}`);
+      if (err instanceof Prisma.PrismaClientKnownRequestError && err.code === 'P2003' && filePath) {
+        await removeManga(getMangaPath(manga.library.path, manga.title));
+        await job.log(`Removed ${manga.title} folder because it was not found in db`);
+      }
       throw err;
     }
   },
@@ -71,3 +93,19 @@ export const downloadQueue = new Queue('downloadQueue', {
     },
   },
 });
+
+export const removeDownloadJobs = async (manga: MangaWithChaptersAndLibrary) => {
+  await Promise.all(
+    manga.chapters.map(async (chapter) => {
+      const jobId = `${sanitizer(manga.title)}_${chapter.index}_download`;
+      try {
+        const job = await downloadQueue.getJob(jobId);
+        if (job) {
+          await job.remove();
+        }
+      } catch (err) {
+        logger.error(`job could not be cancelled. err: ${err}`);
+      }
+    }),
+  );
+};

src/server/trpc/router/manga.ts

@@ -3,7 +3,7 @@ import path from 'path';
 import { z } from 'zod';
 import { isCronValid, sanitizer } from '../../../utils';
 import { checkChaptersQueue, removeJob, schedule } from '../../queue/checkChapters';
-import { downloadQueue } from '../../queue/download';
+import { downloadQueue, downloadWorker, removeDownloadJobs } from '../../queue/download';
 import {
   bindTitleToAnilistId,
   getAvailableSources,
@@ -89,9 +89,11 @@ export const mangaRouter = t.router({
     )
     .mutation(async ({ input, ctx }) => {
       const { id, shouldRemoveFiles } = input;
+      await downloadWorker.pause(true);
       const removed = await ctx.prisma.manga.delete({
         include: {
           library: true,
+          chapters: true,
         },
         where: {
           id,
@@ -102,11 +104,13 @@ export const mangaRouter = t.router({
           id: removed.metadataId,
         },
       });
+      await removeJob(removed.title);
+      await removeDownloadJobs(removed);
       if (shouldRemoveFiles === true) {
         const mangaPath = path.resolve(removed.library.path, sanitizer(removed.title));
         await removeManga(mangaPath);
       }
-      await removeJob(removed.title);
+      downloadWorker.resume();
     }),
   add: t.procedure
     .input(