Custom job data formatter

src/@types/app.ts

@@ -1,7 +1,9 @@
+import * as Redis from 'ioredis'
+
 import { Job, JobOptions } from 'bull'
 import { Job as JobMq, JobsOptions } from 'bullmq'
+
 import React from 'react'
-import * as Redis from 'ioredis'
 import { Status } from '../ui/components/constants'
 
 export type JobCleanStatus =
@@ -11,12 +13,17 @@ export type JobCleanStatus =
   | 'delayed'
   | 'failed'
 
+export type AdapterOptions = {
+  dataFormatter(job: Job | JobMq): AppJob
+}
+
 export type JobStatus = Status
 
 export type JobCounts = Record<JobStatus, number>
 
 export interface QueueAdapter {
   readonly client: Promise<Redis.Redis>
+  readonly options?: AdapterOptions
   getName(): string
 
   getJob(id: string): Promise<Job | JobMq | undefined | null>
@@ -65,6 +72,8 @@ export interface AppJob {
   returnValue: string | Record<string | number, any> | null
 }
 
+export type DataFormatter = (data: JobMq['data']) => JobMq['data']
+
 export interface AppQueue {
   name: string
   counts: Record<Status, number>

src/queueAdapters/bull.ts

@@ -1,17 +1,18 @@
-import { Job, Queue } from 'bull'
 import {
+  AdapterOptions,
   JobCleanStatus,
   JobCounts,
   JobStatus,
   QueueAdapter,
 } from '../@types/app'
+import { Job, Queue } from 'bull'
 
 export class BullAdapter implements QueueAdapter {
   public get client() {
     return Promise.resolve(this.queue.client)
   }
 
-  constructor(public queue: Queue) {}
+  constructor(public queue: Queue, public options?: AdapterOptions) {}
 
   public getName(): string {
     return this.queue.name

src/queueAdapters/bullMQ.ts

@@ -1,11 +1,13 @@
-import { Job, Queue } from 'bullmq'
 import * as Redis from 'ioredis'
+
 import {
+  AdapterOptions,
   JobCleanStatus,
   JobCounts,
   JobStatus,
   QueueAdapter,
 } from '../@types/app'
+import { Job, Queue } from 'bullmq'
 
 export class BullMQAdapter implements QueueAdapter {
   private readonly LIMIT = 1000
@@ -14,7 +16,7 @@ export class BullMQAdapter implements QueueAdapter {
     return (this.queue.client as unknown) as Promise<Redis.Redis>
   }
 
-  constructor(private queue: Queue) {}
+  constructor(private queue: Queue, public options?: AdapterOptions) {}
 
   public getName(): string {
     return this.queue.toKey('~')

src/routes/queues.ts

@@ -1,12 +1,13 @@
-import { Job } from 'bull'
-import { Job as JobMq } from 'bullmq'
-import { Request, RequestHandler, Response } from 'express-serve-static-core'
-import { parse as parseRedisInfo } from 'redis-info'
-
 import * as api from '../@types/api'
 import * as app from '../@types/app'
+
+import { Request, RequestHandler, Response } from 'express-serve-static-core'
+
+import { Job } from 'bull'
+import { Job as JobMq } from 'bullmq'
 import { JobStatus } from '../@types/app'
 import { Status } from '../ui/components/constants'
+import { parse as parseRedisInfo } from 'redis-info'
 
 type MetricName = keyof app.ValidMetrics
 
@@ -39,23 +40,25 @@ const getStats = async ({
   return validMetrics
 }
 
-const formatJob = (job: Job | JobMq): app.AppJob => {
-  const jobProps = job.toJSON()
+const formatJob = (dataFormatter: app.DataFormatter = (d) => d) => {
+  return (job: Job | JobMq): app.AppJob => {
+    const jobProps = job.toJSON()
 
-  return {
-    id: jobProps.id,
-    timestamp: jobProps.timestamp,
-    processedOn: jobProps.processedOn,
-    finishedOn: jobProps.finishedOn,
-    progress: jobProps.progress,
-    attempts: jobProps.attemptsMade,
-    delay: job.opts.delay,
-    failedReason: jobProps.failedReason,
-    stacktrace: jobProps.stacktrace,
-    opts: jobProps.opts,
-    data: jobProps.data,
-    name: jobProps.name,
-    returnValue: jobProps.returnvalue,
+    return {
+      id: jobProps.id,
+      timestamp: jobProps.timestamp,
+      processedOn: jobProps.processedOn,
+      finishedOn: jobProps.finishedOn,
+      progress: jobProps.progress,
+      attempts: jobProps.attemptsMade,
+      delay: job.opts.delay,
+      failedReason: jobProps.failedReason,
+      stacktrace: jobProps.stacktrace,
+      opts: jobProps.opts,
+      data: dataFormatter(jobProps.data),
+      name: jobProps.name,
+      returnValue: jobProps.returnvalue,
+    }
   }
 }
 
@@ -88,11 +91,11 @@ const getDataForQueues = async (
       const status =
         query[name] === 'latest' ? statuses : (query[name] as JobStatus[])
       const jobs = await queue.getJobs(status, 0, 10)
-
+      const formatter = formatJob(queue.options?.dataFormatter)
       return {
         name,
         counts: counts as Record<Status, number>,
-        jobs: jobs.map(formatJob),
+        jobs: jobs.map(formatter),
       }
     }),
   )