feat(queue): priority lists and * wildcard for queue:work

CHANGELOG.md

@@ -8,6 +8,7 @@ The format follows semantic versioning principles.
 
 ### Added
 
+- `ark queue:work --queue` accepts a comma-separated priority list (`--queue=webhook,ocr,default`) and the `*` wildcard for every queue on the connection (`--queue=*`, or `--queue=webhook,*` to put some first). `Worker.runNextJob()` and `daemon({ queue })` take a string or a list. `QueueContract.popAny()` backs the wildcard: the `database` driver takes the oldest due job across all queues, and the `redis` driver discovers queues with `SCAN` and works them round-robin. A bare `queue:work` still works only the default queue.
 - `transports.pusher.public_host` / `public_port` / `public_use_tls` (`PUSHER_PUBLIC_*`) set the Pusher-protocol server address handed to browsers when it differs from the one the app uses, such as soketi inside Docker or behind a TLS proxy. Each defaults to its server-side counterpart.
 - `RealtimeConfig.importModule` loads the transport SDKs (`pusher-js`, `firebase/*`) for every transport and for device tokens, so pages without a bundler can load them from a CDN. Defaults to a dynamic `import()`. The `createPusherTransport`, `createFirebaseTransport`, `createFirebaseDatabaseTransport` and `requestDeviceToken` helpers accept it as an extra argument.
 - `RealtimeClient.deviceToken({ serviceWorker })` takes a service worker registration or script URL (registered with the Firebase web config as `firebaseConfig`), returns `null` where web push is unsupported, and falls back to the `firebase` config when streaming over another transport. `requestDeviceToken()` issues a token without a client.

create-arkstack/src/catalog.ts

@@ -4,26 +4,26 @@
 
 /** The default workspace catalog (`catalog:`). */
 export const catalog: Record<string, string> = {
-    "@h3ravel/musket": "^2.2.13",
-    "@h3ravel/shared": "^2.2.7",
-    "@h3ravel/support": "^2.2.7",
-    "@h3ravel/collect.js": "^5.4.0",
-    "@faker-js/faker": "^10.4.0",
-    "@arkormx/plugin-clear-router": "^0.1.57",
-    "@resora/plugin-clear-router": "^1.0.72",
-    "@kanun-hq/plugin-file": "^0.1.12",
-    "@kanun-hq/plugin-phone": "^0.1.8",
-    "arkormx": "^2.12.8",
-    "clear-router": "^2.9.3",
-    "dotenv": "^17.4.2",
-    "express": "^5.2.1",
-    "h3": "^2.0.1-rc.22",
-    "kysely": "^0.29.2",
-    "kanun": "^1.2.0",
-    "parasito": "^0.2.6",
-    "pg": "^8.22.0",
-    "resora": "^1.3.34",
-    "tsdown": "^0.22.14"
+  '@h3ravel/musket': '^2.2.13',
+  '@h3ravel/shared': '^2.2.7',
+  '@h3ravel/support': '^2.2.7',
+  '@h3ravel/collect.js': '^5.4.0',
+  '@faker-js/faker': '^10.4.0',
+  '@arkormx/plugin-clear-router': '^0.1.57',
+  '@resora/plugin-clear-router': '^1.0.72',
+  '@kanun-hq/plugin-file': '^0.1.12',
+  '@kanun-hq/plugin-phone': '^0.1.8',
+  arkormx: '^2.12.8',
+  'clear-router': '^2.9.3',
+  dotenv: '^17.4.2',
+  express: '^5.2.1',
+  h3: '^2.0.1-rc.22',
+  kysely: '^0.29.2',
+  kanun: '^1.2.0',
+  parasito: '^0.2.6',
+  pg: '^8.22.0',
+  resora: '^1.3.34',
+  tsdown: '^0.22.14',
 }
 
 /** Named workspace catalogs (`catalog:<name>`). */

docs/guide/jobs.md

@@ -87,6 +87,8 @@ With the default `sync` connection the job runs immediately. Configure a `databa
 ark queue:work
 ```
 
+A bare `ark queue:work` processes only the connection's default queue. If jobs go to other queues (`onQueue(...)` or a `queue` property), name them, e.g. `ark queue:work --queue=emails,default`, or use `ark queue:work --queue=*` to work them all.
+
 ## Retries & failure handling
 
 Control attempts and backoff with instance properties, and react to permanent failure with a `failed` hook:

docs/guide/queue.md

@@ -88,21 +88,42 @@ With the `sync` connection, pushed work runs immediately and inline.
 A worker pulls jobs from a connection and runs them. Run one from the CLI:
 
 ```sh
-ark queue:work                      # daemon on the default connection
+ark queue:work                      # daemon on the default connection's default queue
 ark queue:work redis --queue=emails
+ark queue:work --queue=high,default # priority list: high first, then default
+ark queue:work --queue=*            # every queue on the connection
 ark queue:work --once               # process a single job and exit
 ark queue:clear redis --queue=emails
 ```
 
+`--queue` takes one queue, a comma-separated list, or `*`:
+
+- **No `--queue`**: only the connection's default queue (`queue` in its config). Jobs
+  sent elsewhere with `onQueue(...)` or a job's `queue` property are **not** processed.
+- **A list** (`--queue=webhook,ocr,default`) is a strict priority order: on every
+  loop the worker takes the first job it finds walking the list, so later queues
+  run only while earlier ones are empty.
+- **`*`** works every queue on the connection. The `database` driver takes the
+  oldest due job across all queues; the `redis` driver discovers queues with
+  `SCAN` and works them round-robin. Combine it with names to prioritize some:
+  `--queue=webhook,*`.
+
+Every queue that receives jobs needs a worker that lists it (or `*`), or its jobs
+wait forever.
+
 Or drive a worker programmatically:
 
 ```ts
 import { Queue } from '@arkstack/queue'
 
 const worker = Queue.worker('database')
 await worker.daemon({ queue: 'default', sleep: 3 })
+await worker.daemon({ queue: ['webhook', '*'], sleep: 3 }) // priority list / wildcard
 ```
 
+A custom driver supports `*` by overriding `popAny()` on `QueueContract`; the base
+implementation works only the default queue.
+
 ### Retries & failures
 
 A job is **deleted** on success. On failure it is **released** for another attempt until it exhausts its `tries`, at which point it is **marked failed** — invoking the job's `failed` hook — and removed. Released jobs become available again after the job's `backoff` (seconds).

packages/console/src/prepare/BuildInterfaces.ts

@@ -25,9 +25,7 @@ export class BuildInterfaces {
 
   static async tsconfig() {
     const pkg = JSON.parse(await readFile(path.join(process.cwd(), 'package.json'), 'utf-8'))
-    const deps = Object
-      .keys(pkg.dependencies ?? {})
-      .concat(Object.keys(pkg.devDependencies ?? {}))
+    const deps = Object.keys(pkg.dependencies ?? {}).concat(Object.keys(pkg.devDependencies ?? {}))
 
     if (deps.includes('@arkstack/inertia')) {
       TSConfig['include'] = [...TSConfig['include'], '../resources']
@@ -36,7 +34,7 @@ export class BuildInterfaces {
     if (deps.includes('react')) {
       TSConfig['compilerOptions'] = {
         ...TSConfig['compilerOptions'],
-        jsx: 'react-jsx'
+        jsx: 'react-jsx',
       } as never
     }
 

packages/queue/README.md

@@ -49,21 +49,42 @@ await Queue.clear('emails')
 ## Workers
 
 ```bash
-ark queue:work                      # daemon on the default connection
+ark queue:work                      # daemon on the default connection's default queue
 ark queue:work redis --queue=emails
+ark queue:work --queue=high,default # priority list: high first, then default
+ark queue:work --queue=*            # every queue on the connection
 ark queue:work --once               # process a single job and exit
 ark queue:clear redis --queue=emails
 ```
 
+`--queue` takes one queue, a comma-separated list, or `*`:
+
+- **No `--queue`**: only the connection's default queue (`queue` in its config). Jobs
+  sent elsewhere with `onQueue(...)` or a job's `queue` property are **not** processed.
+- **A list** (`--queue=webhook,ocr,default`) is a strict priority order: on every
+  loop the worker takes the first job it finds walking the list, so later queues
+  run only while earlier ones are empty.
+- **`*`** works every queue on the connection. The `database` driver takes the
+  oldest due job across all queues; the `redis` driver discovers queues with
+  `SCAN` and works them round-robin. Combine it with names to prioritize some:
+  `--queue=webhook,*`.
+
+Every queue that receives jobs needs a worker that lists it (or `*`), or its jobs
+wait forever.
+
 Programmatically:
 
 ```ts
 import { Queue } from '@arkstack/queue'
 
 const worker = Queue.worker('database')
 await worker.daemon({ queue: 'default', sleep: 3 })
+await worker.daemon({ queue: ['webhook', '*'], sleep: 3 }) // priority list / wildcard
 ```
 
+A custom driver supports `*` by overriding `popAny()` on `QueueContract`; the base
+implementation works only the default queue.
+
 A job is **deleted** on success, **released** for another attempt on failure, and **marked failed** (invoking its `failed` hook) once it exhausts `tries`.
 
 ## Job (de)serialization

packages/queue/src/Contracts/QueueContract.ts

@@ -51,6 +51,16 @@ export abstract class QueueContract {
    */
   abstract pop(queue?: string): Promise<Job | null>
 
+  /**
+   * Reserve the next job from any queue on this connection, for the worker's `*`
+   * wildcard. Drivers that can list their queues override this; the default
+   * works only the default queue, so a driver without that ability behaves
+   * exactly as a worker given no queue.
+   */
+  async popAny(): Promise<Job | null> {
+    return this.pop()
+  }
+
   /**
    * The number of jobs waiting on the given queue.
    */

packages/queue/src/Worker.ts

@@ -2,9 +2,16 @@ import { Job } from './Job'
 import { QueueContract } from './Contracts/QueueContract'
 import { resolveJob } from './serialization'
 
+/** Matches every queue on the connection in a worker's queue list. */
+export const ANY_QUEUE = '*'
+
 export interface WorkerOptions {
-  /** The queue name to pull from. */
-  queue?: string
+  /**
+   * The queue(s) to pull from. A list is a strict priority order: each loop
+   * takes the first job found walking the list. {@link ANY_QUEUE} (`*`) matches
+   * any queue, so `['webhook', '*']` means "webhook first, then anything".
+   */
+  queue?: string | string[]
   /** Seconds to wait when the queue is empty before polling again. */
   sleep?: number
   /** Stop after processing this many jobs (0 = unlimited). */
@@ -64,10 +71,13 @@ export class Worker {
 
   /**
    * Pop and process the next job. Returns `true` if a job was processed,
-   * `false` when the queue was empty.
+   * `false` when every queue asked for was empty.
+   *
+   * @param queue  A queue, or a priority-ordered list of queues (see
+   *               {@link WorkerOptions.queue}). Omitted: the connection's default.
    */
-  async runNextJob(queue?: string): Promise<boolean> {
-    const job = await this.connection.pop(queue)
+  async runNextJob(queue?: string | string[]): Promise<boolean> {
+    const job = await this.nextJob(queue)
 
     if (!job) {
       return false
@@ -78,6 +88,25 @@ export class Worker {
     return true
   }
 
+  /**
+   * Reserve the next job, walking the queues in priority order.
+   *
+   * @param queue
+   */
+  private async nextJob(queue?: string | string[]): Promise<Job | null> {
+    const queues = Array.isArray(queue) ? queue : [queue]
+
+    for (const name of queues) {
+      const job = name === ANY_QUEUE ? await this.connection.popAny() : await this.connection.pop(name)
+
+      if (job) {
+        return job
+      }
+    }
+
+    return null
+  }
+
   /**
    * Execute a single reserved job, handling success and failure.
    */

packages/queue/src/commands/QueueWorkCommand.ts

@@ -1,6 +1,6 @@
 import { Command } from '@h3ravel/musket'
 import { Queue } from '../QueueManager'
-import { Worker } from '../Worker'
+import { ANY_QUEUE, Worker } from '../Worker'
 import { bootArkorm } from '@arkstack/database'
 
 /** The message of a thrown value, whatever it is. */
@@ -12,7 +12,7 @@ const reason = (error: unknown): string => (error instanceof Error ? error.messa
 export class QueueWorkCommand extends Command {
   protected signature = `queue:work
         {connection? : The queue connection to work. Defaults to the configured default.}
-        {--queue= : The queue to process.}
+        {--queue= : The queue(s) to process: a name, a comma-separated priority list, or * for every queue.}
         {--sleep=3 : Seconds to sleep when no job is available.}
         {--max-jobs=0 : Stop after processing this many jobs (0 for unlimited).}
         {--once : Process a single job and exit.}
@@ -35,7 +35,7 @@ export class QueueWorkCommand extends Command {
       },
     })
 
-    const queue = (this.option('queue') as string | undefined) ?? connection.getDefaultQueue()
+    const queue = this.queues(connection.getDefaultQueue())
 
     try {
       bootArkorm()
@@ -56,7 +56,7 @@ export class QueueWorkCommand extends Command {
       return
     }
 
-    this.info(`Processing jobs from the [${connection.getConnectionName()}] connection on the [${queue}] queue.`)
+    this.info(`Processing jobs from the [${connection.getConnectionName()}] connection on ${this.describe(queue)}.`)
 
     await worker.daemon({
       queue,
@@ -66,6 +66,34 @@ export class QueueWorkCommand extends Command {
     })
   }
 
+  /**
+   * The queues to work, from `--queue`: `ocr` → `['ocr']`, `webhook,ocr` → a
+   * priority list, `*` → every queue. Falls back to the connection's default.
+   *
+   * @param fallback  The connection's default queue.
+   */
+  private queues(fallback: string): string[] {
+    const names = String(this.option('queue') ?? '')
+      .split(',')
+      .map((name) => name.trim())
+      .filter(Boolean)
+
+    return names.length ? [...new Set(names)] : [fallback]
+  }
+
+  /**
+   * How the worker's queues read in its startup line.
+   *
+   * @param queues
+   */
+  private describe(queues: string[]): string {
+    if (queues.length === 1) {
+      return queues[0] === ANY_QUEUE ? 'every queue' : `the [${queues[0]}] queue`
+    }
+
+    return `the [${queues.join(', ')}] queues, in that order`
+  }
+
   /**
    * Read a multi-word flag. Musket hands over the parsed options camelCased,
    * so `--stop-when-empty` arrives as `stopWhenEmpty`; the kebab-case name is