refactor: rename defineQueues to defineJobs & remove proxy

Author: LouisHaftmann

README.md

@@ -13,7 +13,7 @@ npm add @falcondev-oss/queue
 #### 1. Define your job schemas and handlers
 
 ```ts
-const jobs = {
+const jobs = defineJobs({
   video: {
     process: defineJob({
       schema: z.object({
@@ -39,32 +39,26 @@ const jobs = {
       // send email to `payload.to` with `payload.subject` and `payload.body`
     },
   }),
-}
-```
-
-#### 2. Define queues
-
-```ts
-const queues = defineQueues(jobs)
+})
 ```
 
-#### 3. Start workers
+#### 2. Start workers
 
 ```ts
 await startWorkers(jobs)
 ```
 
-#### 4. Queue jobs
+#### 3. Queue jobs
 
 Single jobs:
 
 ```ts
-queues.video.process.queue({
+await jobs.video.process.queue({
   // payload type inferred from schema
   outputFormat: 'mp4',
   path: '/path/to/video.mov',
 })
-queues.sendEmail.queue({
+await jobs.sendEmail.queue({
   to: 'user@example.com',
   subject: 'Hello',
   body: 'This is a test email',
@@ -74,7 +68,7 @@ queues.sendEmail.queue({
 Bulk jobs:
 
 ```ts
-queues.sendEmail.queueBulk([
+await jobs.sendEmail.queueBulk([
   {
     to: 'user@example.com',
     subject: 'Hello',

package.json

@@ -44,7 +44,8 @@
     "@standard-schema/spec": "^1.0.0",
     "@types/node": "^24.5.2",
     "bullmq": "^5.58.7",
-    "ioredis": "^5.7.0"
+    "ioredis": "^5.7.0",
+    "remeda": "^2.32.0"
   },
   "devDependencies": {
     "@falcondev-oss/configs": "^5.0.2",

pnpm-lock.yaml

@@ -8,16 +8,17 @@ import type {
   DefineFlowOptions,
   DefineJobOptions,
   Flow,
+  FlowAccessor,
   FlowStep,
   Job,
+  JobAccessor,
   JobDefinitionsObject,
-  QueueFlowProxyAccessor,
-  QueueJobProxy,
-  QueueJobProxyAccessor,
+  Jobs,
   WorkerOptions,
 } from './types'
 import { Queue as BullQueue, Worker as BullWorker, FlowProducer } from 'bullmq'
 import IORedis from 'ioredis'
+import { mapValues } from 'remeda'
 import { FlowBuilder, flowSymbol, jobSymbol, StandardSchemaV1Error } from './types'
 
 export function defineJob<Schema extends StandardSchemaV1, Output>(
@@ -121,7 +122,7 @@ export function defineFlow<Schema extends StandardSchemaV1, Output>(
   }
 }
 
-export function defineQueues<J extends JobDefinitionsObject>(jobs: J, opts?: BullQueueOptions) {
+export function defineJobs<J extends JobDefinitionsObject>(jobs: J, opts?: BullQueueOptions) {
   const connection =
     opts?.connection ??
     new IORedis({
@@ -162,49 +163,40 @@ export function defineQueues<J extends JobDefinitionsObject>(jobs: J, opts?: Bul
     return flowProducer
   }
 
-  function createProxy(obj: JobDefinitionsObject, path: string[]) {
-    return new Proxy(obj, {
-      get(target, p, receiver) {
-        if (typeof p !== 'string') return
+  function traverse(obj: JobDefinitionsObject, path: string[]): Jobs<any> {
+    return mapValues(obj, (jobOrJobs, p) => {
+      if (typeof jobOrJobs !== 'object') throw new Error('Job definition must be an object')
 
-        const jobOrJobs = Reflect.get(target, p, receiver)
-        if (typeof jobOrJobs !== 'object') return
+      const fullPath = [...path, p]
 
-        const fullPath = [...path, p]
-
-        if (jobSymbol in jobOrJobs) {
-          const jobName = fullPath.join('-')
+      if (jobSymbol in jobOrJobs) {
+        const jobName = fullPath.join('-')
 
-          return {
-            queue: async (payload: unknown) => {
-              return jobOrJobs[jobSymbol].addToQueue(await getQueue(jobName), payload)
-            },
-            queueBulk: async (payloads: unknown[]) => {
-              return jobOrJobs[jobSymbol].addToQueueBulk(await getQueue(jobName), payloads)
-            },
-          } satisfies QueueJobProxyAccessor<any, any>
-        } else if (flowSymbol in jobOrJobs) {
-          const flowName = fullPath.join('-')
+        return {
+          queue: async (payload: unknown) => {
+            return jobOrJobs[jobSymbol].addToQueue(await getQueue(jobName), payload)
+          },
+          queueBulk: async (payloads: unknown[]) => {
+            return jobOrJobs[jobSymbol].addToQueueBulk(await getQueue(jobName), payloads)
+          },
+        } satisfies JobAccessor<any, any>
+      } else if (flowSymbol in jobOrJobs) {
+        const flowName = fullPath.join('-')
 
-          return {
-            queue: async (payload: unknown) => {
-              return jobOrJobs[flowSymbol].addToQueue(flowName, await getFlowProducer(), payload)
-            },
-            queueBulk: async (payloads: unknown[]) => {
-              return jobOrJobs[flowSymbol].addToQueueBulk(
-                flowName,
-                await getFlowProducer(),
-                payloads,
-              )
-            },
-          } satisfies QueueFlowProxyAccessor<any>
-        }
+        return {
+          queue: async (payload: unknown) => {
+            return jobOrJobs[flowSymbol].addToQueue(flowName, await getFlowProducer(), payload)
+          },
+          queueBulk: async (payloads: unknown[]) => {
+            return jobOrJobs[flowSymbol].addToQueueBulk(flowName, await getFlowProducer(), payloads)
+          },
+        } satisfies FlowAccessor<any>
+      }
 
-        return createProxy(jobOrJobs, fullPath)
-      },
+      return traverse(jobOrJobs, fullPath)
     })
   }
-  return createProxy(jobs, []) as unknown as QueueJobProxy<J>
+  return traverse(jobs, []) as Jobs<J>
 }
 
 export async function startWorkers<J extends JobDefinitionsObject>(

src/index.ts

@@ -99,25 +99,25 @@ export type JobDefinitionsObject = {
   [key: string]: Job<any, any> | Flow<any, any> | JobDefinitionsObject
 }
 
-export type QueueJobProxyAccessor<S extends StandardSchemaV1, Output> = {
+export type JobAccessor<S extends StandardSchemaV1, Output> = {
   queue: (
     payload: StandardSchemaV1.InferInput<S>,
   ) => Promise<BullJob<StandardSchemaV1.InferOutput<S>, Output, string>>
   queueBulk: (
     payloads: StandardSchemaV1.InferInput<S>[],
   ) => Promise<BullJob<StandardSchemaV1.InferOutput<S>, Output, string>[]>
 }
-export type QueueFlowProxyAccessor<S extends StandardSchemaV1> = {
+export type FlowAccessor<S extends StandardSchemaV1> = {
   queue: (payload: StandardSchemaV1.InferInput<S>) => Promise<JobNode>
   queueBulk: (payloads: StandardSchemaV1.InferInput<S>[]) => Promise<JobNode[]>
 }
-export type QueueJobProxy<J extends JobDefinitionsObject> = {
+export type Jobs<J extends JobDefinitionsObject> = {
   [K in keyof J]: J[K] extends Job<infer S, infer O>
-    ? QueueJobProxyAccessor<S, O>
+    ? JobAccessor<S, O>
     : J[K] extends Flow<infer S, any>
-      ? QueueFlowProxyAccessor<S>
+      ? FlowAccessor<S>
       : J[K] extends JobDefinitionsObject
-        ? QueueJobProxy<J[K]>
+        ? Jobs<J[K]>
         : never
 }