feat: background job overhaul with bullmq

Author: jakeaturner

admin/.env.example

@@ -4,7 +4,6 @@ LOG_LEVEL=info
 APP_KEY=some_random_key
 NODE_ENV=development
 SESSION_DRIVER=cookie
-DRIVE_DISK=fs
 DB_HOST=localhost
 DB_PORT=3306
 DB_USER=root

admin/adonisrc.ts

@@ -52,8 +52,8 @@ export default defineConfig({
     () => import('@adonisjs/cors/cors_provider'),
     () => import('@adonisjs/lucid/database_provider'),
     () => import('@adonisjs/inertia/inertia_provider'),
-    () => import('@adonisjs/drive/drive_provider'),
-    () => import('@adonisjs/transmit/transmit_provider')
+    () => import('@adonisjs/transmit/transmit_provider'),
+    () => import('#providers/map_static_provider')
   ],
 
   /*

admin/app/controllers/downloads_controller.ts

@@ -0,0 +1,18 @@
+import type { HttpContext } from '@adonisjs/core/http'
+import { DownloadService } from '#services/download_service'
+import { downloadJobsByFiletypeSchema } from '#validators/download'
+import { inject } from '@adonisjs/core'
+
+@inject()
+export default class DownloadsController {
+  constructor(private downloadService: DownloadService) {}
+
+  async index() {
+    return this.downloadService.listDownloadJobs()
+  }
+
+  async filetype({ request }: HttpContext) {
+    const payload = await request.validateUsing(downloadJobsByFiletypeSchema)
+    return this.downloadService.listDownloadJobs(payload.params.filetype)
+  }
+}

admin/app/controllers/maps_controller.ts

@@ -1,6 +1,6 @@
 import { MapService } from '#services/map_service'
 import {
-  filenameValidator,
+  filenameParamValidator,
   remoteDownloadValidator,
   remoteDownloadValidatorOptional,
 } from '#validators/common'
@@ -53,14 +53,14 @@ export default class MapsController {
   }
 
   async delete({ request, response }: HttpContext) {
-    const payload = await request.validateUsing(filenameValidator)
+    const payload = await request.validateUsing(filenameParamValidator)
 
     try {
-      await this.mapService.delete(payload.filename)
+      await this.mapService.delete(payload.params.filename)
     } catch (error) {
       if (error.message === 'not_found') {
         return response.status(404).send({
-          message: `Map file with key ${payload.filename} not found`,
+          message: `Map file with key ${payload.params.filename} not found`,
         })
       }
       throw error // Re-throw any other errors and let the global error handler catch

admin/app/controllers/zim_controller.ts

@@ -1,7 +1,7 @@
 import { ZimService } from '#services/zim_service'
 import {
   downloadCollectionValidator,
-  filenameValidator,
+  filenameParamValidator,
   remoteDownloadValidator,
 } from '#validators/common'
 import { listRemoteZimValidator } from '#validators/zim'
@@ -24,11 +24,12 @@ export default class ZimController {
 
   async downloadRemote({ request }: HttpContext) {
     const payload = await request.validateUsing(remoteDownloadValidator)
-    const filename = await this.zimService.downloadRemote(payload.url)
+    const { filename, jobId } = await this.zimService.downloadRemote(payload.url)
 
     return {
       message: 'Download started successfully',
       filename,
+      jobId,
       url: payload.url,
     }
   }
@@ -44,10 +45,6 @@ export default class ZimController {
     }
   }
 
-  async listActiveDownloads({}: HttpContext) {
-    return this.zimService.listActiveDownloads()
-  }
-
   async listCuratedCollections({}: HttpContext) {
     return this.zimService.listCuratedCollections()
   }
@@ -58,14 +55,14 @@ export default class ZimController {
   }
 
   async delete({ request, response }: HttpContext) {
-    const payload = await request.validateUsing(filenameValidator)
+    const payload = await request.validateUsing(filenameParamValidator)
 
     try {
-      await this.zimService.delete(payload.filename)
+      await this.zimService.delete(payload.params.filename)
     } catch (error) {
       if (error.message === 'not_found') {
         return response.status(404).send({
-          message: `ZIM file with key ${payload.filename} not found`,
+          message: `ZIM file with key ${payload.params.filename} not found`,
         })
       }
       throw error // Re-throw any other errors and let the global error handler catch

admin/app/jobs/run_download_job.ts

@@ -0,0 +1,107 @@
+import { Job } from 'bullmq'
+import { RunDownloadJobParams } from '../../types/downloads.js'
+import { QueueService } from '#services/queue_service'
+import { doResumableDownload } from '../utils/downloads.js'
+import { createHash } from 'crypto'
+import { DockerService } from '#services/docker_service'
+import { ZimService } from '#services/zim_service'
+
+export class RunDownloadJob {
+  static get queue() {
+    return 'downloads'
+  }
+
+  static get key() {
+    return 'run-download'
+  }
+
+  static getJobId(url: string): string {
+    return createHash('sha256').update(url).digest('hex').slice(0, 16)
+  }
+
+  async handle(job: Job) {
+    const { url, filepath, timeout, allowedMimeTypes, forceNew, filetype } =
+      job.data as RunDownloadJobParams
+
+    //   console.log("Simulating delay for job for URL:", url)
+    // await new Promise((resolve) => setTimeout(resolve, 30000)) // Simulate initial delay
+    // console.log("Starting download for URL:", url)
+
+    // // simulate progress updates for demonstration
+    // for (let progress = 0; progress <= 100; progress += 10) {
+    //   await new Promise((resolve) => setTimeout(resolve, 20000)) // Simulate time taken for each progress step
+    //   job.updateProgress(progress)
+    //   console.log(`Job progress for URL ${url}: ${progress}%`)
+    // }
+
+    await doResumableDownload({
+      url,
+      filepath,
+      timeout,
+      allowedMimeTypes,
+      forceNew,
+      onProgress(progress) {
+        const progressPercent = (progress.downloadedBytes / (progress.totalBytes || 1)) * 100
+        job.updateProgress(Math.floor(progressPercent))
+      },
+      async onComplete(url) {
+        if (filetype === 'zim') {
+          try {
+            const dockerService = new DockerService()
+            const zimService = new ZimService(dockerService)
+            await zimService.downloadRemoteSuccessCallback([url], true)
+          } catch (error) {
+            console.error(
+              `[RunDownloadJob] Error in ZIM download success callback for URL ${url}:`,
+              error
+            )
+          }
+        }
+        job.updateProgress(100)
+      },
+    })
+
+    return {
+      url,
+      filepath,
+    }
+  }
+
+  static async getByUrl(url: string): Promise<Job | undefined> {
+    const queueService = new QueueService()
+    const queue = queueService.getQueue(this.queue)
+    const jobId = this.getJobId(url)
+    return await queue.getJob(jobId)
+  }
+
+  static async dispatch(params: RunDownloadJobParams) {
+    const queueService = new QueueService()
+    const queue = queueService.getQueue(this.queue)
+    const jobId = this.getJobId(params.url)
+
+    try {
+      const job = await queue.add(this.key, params, {
+        jobId,
+        attempts: 3,
+        backoff: { type: 'exponential', delay: 2000 },
+        removeOnComplete: true,
+      })
+
+      return {
+        job,
+        created: true,
+        message: `Dispatched download job for URL ${params.url}`,
+      }
+    } catch (error) {
+      if (error.message.includes('job already exists')) {
+        const existing = await queue.getJob(jobId)
+        return {
+          job: existing,
+          created: false,
+          message: `Job already exists for URL ${params.url}`,
+        }
+      }
+      throw error
+    }
+  }
+}

admin/app/middleware/maps_static_middleware.ts

@@ -0,0 +1,20 @@
+import type { HttpContext } from '@adonisjs/core/http'
+import type { NextFn } from '@adonisjs/core/types/http'
+import StaticMiddleware from '@adonisjs/static/static_middleware'
+import { AssetsConfig } from '@adonisjs/static/types'
+
+/**
+ * See #providers/map_static_provider.ts for explanation
+ * of why this middleware exists.
+ */
+export default class MapsStaticMiddleware {
+  constructor(
+    private path: string,
+    private config: AssetsConfig
+  ) {}
+
+  async handle(ctx: HttpContext, next: NextFn) {
+    const staticMiddleware = new StaticMiddleware(this.path, this.config)
+    return staticMiddleware.handle(ctx, next)
+  }
+}

admin/app/services/docker_service.ts

@@ -4,7 +4,7 @@ import logger from '@adonisjs/core/services/logger'
 import { inject } from '@adonisjs/core'
 import { ServiceStatus } from '../../types/services.js'
 import transmit from '@adonisjs/transmit/services/main'
-import { doSimpleDownload } from '../utils/downloads.js'
+import { doResumableDownloadWithRetry } from '../utils/downloads.js'
 import path from 'path'
 
 @inject()
@@ -347,10 +347,15 @@ export class DockerService {
     )
 
     try {
-      await doSimpleDownload({
+      await doResumableDownloadWithRetry({
         url: WIKIPEDIA_ZIM_URL,
         filepath,
         timeout: 60000,
+        allowedMimeTypes: [
+          'application/x-zim',
+          'application/x-openzim',
+          'application/octet-stream',
+        ],
       })
 
       this._broadcast(