fix(queue): make queue:work actually process jobs, and say what it did

docs/guide/jobs.md

@@ -122,4 +122,12 @@ serialize() {
 }
 ```
 
-For dedicated worker **processes**, make sure your job modules are imported — or call `JobRegistry.register(MyJob)` — so the names are known before jobs are processed.
+A dedicated worker **process** constructs none of your jobs, so `ark queue:work` imports every job module in `src/app/jobs` (or its build output) before it starts working, registering each class it finds under its class name. Jobs kept elsewhere can be registered by loading that directory instead:
+
+```ts
+import { loadJobs } from '@arkstack/jobs';
+
+await loadJobs('domain/jobs'); // relative to src/, or the build output
+```
+
+`JobRegistry.register(MyJob)` still registers a single class explicitly.

packages/jobs/src/JobRegistry.ts

@@ -2,17 +2,30 @@ import type { JobPayload } from '@arkstack/queue'
 import type { Job } from './Job'
 import type { JobConstructor } from './types'
 
+// Back the registry with a global symbol so a job module evaluated more than
+// once — the app's own import and a worker's jiti-loaded copy are different
+// module instances — registers into the one map the worker resolves against.
+const REGISTRY = Symbol.for('arkstack.jobs.registry')
+
+const registry = (): Map<string, JobConstructor> => {
+    const store = globalThis as unknown as Record<symbol, Map<string, JobConstructor>>
+
+    return (store[REGISTRY] ??= new Map())
+}
+
 /**
  * A registry mapping job names to their classes so a worker can reconstruct job
  * instances from a stored payload.
  *
  * Job classes register themselves when instantiated, which covers same-process
- * dispatch + work. For dedicated worker processes, ensure the job modules are
- * imported (or call {@link JobRegistry.register} explicitly) so the names are
- * known before jobs are processed.
+ * dispatch + work. A dedicated worker process constructs none of them, so
+ * `queue:work` calls {@link loadJobs} to import the application's job modules
+ * before it starts working.
  */
 export class JobRegistry {
-    private static classes = new Map<string, JobConstructor>()
+    private static get classes(): Map<string, JobConstructor> {
+        return registry()
+    }
 
     /**
      * Register a job class under an explicit name (defaults to the class name).

packages/jobs/src/index.ts

@@ -7,5 +7,6 @@ export * from './Job'
 export * from './JobRegistry'
 export * from './PendingDispatch'
 export * from './dispatch'
+export * from './loader'
 export * from './bridge'
 export * from './types'

packages/jobs/src/loader.ts

@@ -0,0 +1,80 @@
+import { importFile, nodeEnv, outputDir } from '@arkstack/common'
+import path, { join } from 'node:path'
+
+import { Arkstack } from '@arkstack/contract'
+import { Dirent } from 'node:fs'
+import { JobConstructor } from './types'
+import { JobRegistry } from './JobRegistry'
+import { readdirSync } from 'node:fs'
+
+/**
+ * Whether a module export is a runnable job class.
+ *
+ * The test is `handle` on the prototype rather than `instanceof Job`: a job
+ * module loaded through jiti carries its own copy of the base class, so an
+ * identity check would reject every job it finds. An abstract base declares no
+ * `handle` of its own and is skipped.
+ */
+const isJobClass = (value: unknown): value is JobConstructor => typeof value === 'function'
+    && typeof (value as { prototype?: { handle?: unknown } }).prototype?.handle === 'function'
+
+/** The job modules in the first candidate directory that has any. */
+const jobFiles = (directories: string[]): [string, Dirent<string>[]] => {
+    for (const directory of directories) {
+        try {
+            const files = readdirSync(directory, { withFileTypes: true }).filter(
+                (file) => file.isFile() && ['.ts', '.js', '.mjs'].includes(path.extname(file.name)),
+            )
+
+            if (files.length) return [directory, files]
+        } catch {
+            // Directory missing — try the next candidate.
+        }
+    }
+
+    return ['', []]
+}
+
+/**
+ * Import the application's job classes so a worker can reconstruct them.
+ *
+ * A dedicated `queue:work` process never constructs the app's jobs itself, and
+ * a job class only reaches the {@link JobRegistry} when its module is loaded —
+ * without this a worker cannot resolve a single payload it pops, and every job
+ * is released back onto the queue until it exhausts its attempts.
+ *
+ * Every concrete `Job` subclass a module exports is registered under its class
+ * name, which is the name payloads are serialized with.
+ *
+ * @param subPath  Directory (under `src/`, or the build output) to load from.
+ * @returns        The names registered, in discovery order.
+ */
+export const loadJobs = async (subPath: string = join('app', 'jobs')): Promise<string[]> => {
+    const root = Arkstack.rootDir()
+
+    // Production prefers the build output (a deploy ships only `dist`); dev
+    // prefers source so edits land without a rebuild. Either falls back.
+    const [directory, files] = jobFiles(nodeEnv() === 'prod'
+        ? [join(outputDir(), subPath), join(root, 'src', subPath)]
+        : [join(root, 'src', subPath), join(outputDir(), subPath)])
+
+    const registered: string[] = []
+
+    for (const file of files) {
+        // One unloadable job module must not cost the worker every other job.
+        try {
+            const module = await importFile<Record<string, unknown>>(join(directory, file.name))
+
+            for (const value of Object.values(module)) {
+                if (!isJobClass(value)) continue
+
+                JobRegistry.register(value)
+                registered.push(value.name)
+            }
+        } catch {
+            continue
+        }
+    }
+
+    return registered
+}

packages/jobs/tests/jobs.test.ts

@@ -1,4 +1,4 @@
-import { Job, JobRegistry, dispatch } from '../src'
+import { Job, JobRegistry, dispatch, loadJobs } from '../src'
 import { type JobPayload, QueueContract, type Queueable, Worker, serializeJob } from '@arkstack/queue'
 import { afterAll, afterEach, beforeAll, describe, expect, it } from 'vitest'
 import { dirname, resolve } from 'node:path'
@@ -153,6 +153,36 @@ describe('Jobs', () => {
         })
     })
 
+    describe('loadJobs', () => {
+        it('registers the job classes a worker process would otherwise not know', async () => {
+            // A dedicated worker constructs none of the app's jobs, so nothing
+            // reaches the registry until their modules are loaded.
+            JobRegistry.clear()
+
+            expect(JobRegistry.has('LoadedJob')).toBe(false)
+
+            const registered = await loadJobs()
+
+            expect(registered).toContain('LoadedJob')
+            expect(registered).toContain('AlsoLoadedJob')
+            expect(JobRegistry.has('LoadedJob')).toBe(true)
+            expect(JobRegistry.has('AlsoLoadedJob')).toBe(true)
+        })
+
+        it('skips abstract bases, non-job exports and modules that fail to load', async () => {
+            const registered = await loadJobs()
+
+            expect(registered).not.toContain('BaseFixtureJob')
+            expect(registered).not.toContain('notAJob')
+            // BrokenJob.ts throws on import; the rest still load.
+            expect(registered).toContain('LoadedJob')
+        })
+
+        it('returns nothing when the directory does not exist', async () => {
+            expect(await loadJobs('app/nowhere')).toEqual([])
+        })
+    })
+
     describe('End to end with a worker', () => {
         it('serializes, stores, resolves and runs the job', async () => {
             const queue = new MemoryQueue()

packages/jobs/tests/src/app/jobs/BrokenJob.ts

@@ -0,0 +1 @@
+throw new Error('this module cannot be loaded')

packages/jobs/tests/src/app/jobs/FixtureJobs.ts

@@ -0,0 +1,15 @@
+import { Job } from '../../../../src'
+
+/** An abstract base: not instantiable, so it must not be registered. */
+export abstract class BaseFixtureJob extends Job { }
+
+export class LoadedJob extends BaseFixtureJob {
+    async handle () { }
+}
+
+export class AlsoLoadedJob extends Job {
+    async handle () { }
+}
+
+/** A plain export that isn't a job at all. */
+export const notAJob = () => 'nope'

packages/queue/src/Contracts/QueueContract.ts

@@ -23,6 +23,14 @@ export abstract class QueueContract {
         return this
     }
 
+    /**
+     * The queue this connection works when none is named. Drivers with a
+     * configurable queue override it.
+     */
+    getDefaultQueue (): string {
+        return 'default'
+    }
+
     /**
      * Push a job onto the queue. Returns the job id.
      */

packages/queue/src/Worker.ts

@@ -13,6 +13,23 @@ export interface WorkerOptions {
     stopWhenEmpty?: boolean
 }
 
+/**
+ * Observers for what a worker does with each job. A worker executes jobs the
+ * caller can't see — a dedicated worker process reports what happened through
+ * these, so a job that keeps failing isn't indistinguishable from an idle queue.
+ */
+export interface WorkerHandlers {
+    /** A job has been reserved and is about to run. */
+    onProcessing?: (job: Job) => void | Promise<void>
+    /** A job completed and has been removed from the queue. */
+    onProcessed?: (job: Job) => void | Promise<void>
+    /**
+     * A job threw. `released` is `true` when it goes back on the queue for
+     * another attempt, `false` when it has exhausted its tries and failed.
+     */
+    onFailed?: (job: Job, error: unknown, released: boolean) => void | Promise<void>
+}
+
 const wait = (ms: number): Promise<void> => new Promise((resolve) => setTimeout(resolve, ms))
 
 /**
@@ -24,6 +41,7 @@ const wait = (ms: number): Promise<void> => new Promise((resolve) => setTimeout(
  */
 export class Worker {
     private shouldStop = false
+    private readonly handlers: WorkerHandlers = {}
 
     constructor(private readonly connection: QueueContract) { }
 
@@ -32,6 +50,18 @@ export class Worker {
         this.shouldStop = true
     }
 
+    /**
+     * Observe what the worker does with each job. Handlers merge, so they can be
+     * registered in more than one call.
+     *
+     * @param handlers  The callbacks to add.
+     */
+    on (handlers: WorkerHandlers): this {
+        Object.assign(this.handlers, handlers)
+
+        return this
+    }
+
     /**
      * Pop and process the next job. Returns `true` if a job was processed,
      * `false` when the queue was empty.
@@ -53,10 +83,13 @@ export class Worker {
      */
     async process (job: Job): Promise<void> {
         try {
+            await this.handlers.onProcessing?.(job)
+
             const instance = await resolveJob(job.payload)
 
             await instance.handle()
             await job.delete()
+            await this.handlers.onProcessed?.(job)
         } catch (error) {
             await this.handleFailure(job, error)
         }
@@ -107,10 +140,12 @@ export class Worker {
             }
 
             await job.delete()
+            await this.handlers.onFailed?.(job, error, false)
 
             return
         }
 
         await job.release(job.backoff())
+        await this.handlers.onFailed?.(job, error, true)
     }
 }

packages/queue/src/commands/QueueWorkCommand.ts

@@ -1,7 +1,13 @@
 import { Command } from '@h3ravel/musket'
 import { Queue } from '../QueueManager'
+import { Worker } from '../Worker'
 import { bootArkorm } from '@arkstack/database'
 
+/** The message of a thrown value, whatever it is. */
+const reason = (error: unknown): string => error instanceof Error
+    ? error.message
+    : String(error)
+
 /**
  * Process jobs from a queue connection.
  */
@@ -17,29 +23,72 @@ export class QueueWorkCommand extends Command {
     protected description = 'Start processing jobs on the queue as a daemon.'
 
     async handle() {
-        const connection = this.argument('connection') as string | undefined
-        const worker = Queue.worker(connection)
-        const queue = this.option('queue') as string | undefined
+        const connection = Queue.connection(this.argument('connection') as string | undefined)
+        const worker = new Worker(connection).on({
+            onProcessed: (job) => {
+                this.success(`Processed: ${job.name()}`)
+            },
+            onFailed: (job, error, released) => {
+                this.error(released
+                    ? `Failed (attempt ${job.attempts()}, retrying): ${job.name()} — ${reason(error)}`
+                    : `Failed permanently: ${job.name()} — ${reason(error)}`)
+            },
+        })
+
+        const queue = (this.option('queue') as string | undefined) ?? connection.getDefaultQueue()
+
+        try {
+            bootArkorm()
+        } catch {/** */ }
+
+        // The worker runs in its own process, so nothing has constructed the
+        // application's jobs and the registry a payload resolves against is
+        // empty. Load them before working, or every job pops and fails.
+        await this.registerJobs()
 
         if (this.option('once')) {
             const handled = await worker.runNextJob(queue)
 
-            this.info(handled ? 'Processed one job.' : 'No jobs available.')
+            if (!handled) this.info('No jobs available.')
 
             return
         }
 
-        this.info(`Processing jobs from [${connection ?? 'default'}] connection.`)
-
-        try {
-            bootArkorm()
-        } catch {/** */ }
+        this.info(`Processing jobs from the [${connection.getConnectionName()}] connection on the [${queue}] queue.`)
 
         await worker.daemon({
             queue,
             sleep: Number(this.option('sleep') ?? 3),
-            maxJobs: Number(this.option('max-jobs') ?? 0),
-            stopWhenEmpty: Boolean(this.option('stop-when-empty')),
+            maxJobs: Number(this.flag('maxJobs', 'max-jobs') ?? 0),
+            stopWhenEmpty: Boolean(this.flag('stopWhenEmpty', 'stop-when-empty')),
         })
     }
+
+    /**
+     * Read a multi-word flag. Musket hands over the parsed options camelCased,
+     * so `--stop-when-empty` arrives as `stopWhenEmpty`; the kebab-case name is
+     * accepted too so the flag can't go quietly missing again.
+     *
+     * @param camel  The camelCase key.
+     * @param kebab  The flag as written in the signature.
+     */
+    private flag(camel: string, kebab: string): unknown {
+        return this.option(camel) ?? this.option(kebab)
+    }
+
+    /**
+     * Register the application's job classes with `@arkstack/jobs` when it is
+     * installed. Without it a payload cannot be turned back into a job.
+     */
+    private async registerJobs(): Promise<void> {
+        try {
+            const specifier = '@arkstack/jobs'
+            const { loadJobs } = await import(specifier)
+            const names = (await loadJobs()) as string[]
+
+            if (names.length) this.line(`Loaded ${names.length} job class(es).`)
+        } catch {
+            // `@arkstack/jobs` isn't installed; the app registers its own jobs.
+        }
+    }
 }