chore: wip

Author: glennmichael123

app/Jobs/ExampleJob.ts

@@ -7,8 +7,8 @@ export default new Job({
   tries: 3, // optional, defaults to 3 retries, in case of failures
   backoff: 3, // optional, defaults to 3-second delays between retries
   rate: Every.Minute, // optional, '* * * * *' in cron syntax
-  handle: () => {
-    console.log('hello world')
+  handle: (payload: any) => {
+    console.log(payload)
   },
   // action: 'SendWelcomeEmail', // instead of handle, you may target an action or `action: () => {`
 })

storage/framework/core/actions/src/queue/work.ts

@@ -6,8 +6,6 @@ const options = parseArgs()
 
 const queue = options.queue
 
-console.log(queue)
-
 const result = await processJobs(queue)
 
 if (result?.isErr()) {

storage/framework/core/queue/src/job.ts

@@ -7,7 +7,7 @@ import { storeJob } from './utils'
 const queueDriver = 'database'
 
 interface JobConfig {
-  handle?: () => Promise<void>
+  handle?: (payload?: any) => Promise<void>
   action?: string | (() => Promise<void>)
 }
 
@@ -66,7 +66,7 @@ export async function runJob(name: string, options: JobOptions = {}): Promise<vo
     }
     // If handle is defined, execute it
     else if (job.handle) {
-      await job.handle()
+      await job.handle(options.payload)
     }
     // If no handle or action, try to execute the module directly
     else if (typeof jobModule.default === 'function') {

storage/framework/core/queue/src/process.ts

@@ -1,6 +1,6 @@
 import { ok, type Ok } from '@stacksjs/error-handling'
 import { log } from '@stacksjs/logging'
-import { Job } from '../../../orm/src/models/Job'
+import { Job, type JobModel } from '../../../orm/src/models/Job'
 import { runJob } from './job'
 
 interface QueuePayload {
@@ -12,51 +12,62 @@ interface QueuePayload {
 }
 
 export async function processJobs(queue: string | undefined): Promise<Ok<string, never>> {
-  setInterval(async () => {
-    await executeJobs(queue)
-  }, 1000)
+  async function process() {
+    try {
+      await executeJobs(queue)
+    } catch (error) {
+      log.error('Error processing jobs:', error)
+    }
+
+    setTimeout(process, 1000)
+  }
+
+  process()
 
-  return ok('All jobs processed successfully!')
+  return ok('Job processing has started successfully!')
 }
 
 async function executeJobs(queue: string | undefined): Promise<void> {
-  const jobs = await Job.when(queue !== undefined, (query: any) => {
-    return query.where('queue', queue)
-  }).get()
+  const jobs = await Job.when(queue !== undefined, (query: any) => query.where('queue', queue)).get()
 
   for (const job of jobs) {
-    if (job.payload) {
-      if (job.available_at && job.available_at > timestampNow())
-        return
-
-      const payload: QueuePayload = JSON.parse(job.payload)
-      const currentAttempts = job.attempts || 0
-
-      log.info(`Running ${payload.displayName}`)
-
-      await job.update({ attempts: currentAttempts + 1 })
-
-      try {
-        await runJob(payload.name, {
-          queue: job.queue,
-          payload: {},
-          context: '',
-          maxTries: payload.maxTries,
-          timeout: 60,
-        })
-
-        await job.delete()
-      }
-      catch (error) {
-        log.info(`${payload.displayName} failed`)
-        log.error(error)
-      }
-
-      log.info(`Successfully ran ${payload.displayName}`)
+
+    if (!job.payload) continue
+
+    if (job.available_at && job.available_at > timestampNow()) continue
+
+    const payload: QueuePayload = JSON.parse(job.payload)
+    const currentAttempts = job.attempts || 0
+
+    log.info(`Running job: ${payload.displayName}`)
+
+    await updateJobAttempts(job, currentAttempts)
+
+    try {
+      await runJob(payload.name, {
+        queue: job.queue,
+        payload: {},
+        context: '',
+        maxTries: payload.maxTries,
+        timeout: 60,
+      })
+
+      await job.delete()
+      log.info(`Successfully ran job: ${payload.displayName}`)
+    } catch (error) {
+      log.error(`Job failed: ${payload.displayName}`, error)
     }
   }
 }
 
+async function updateJobAttempts(job: any, currentAttempts: number): Promise<void> {
+  try {
+    await job.update({ attempts: currentAttempts + 1 })
+  } catch (error) {
+    log.error('Failed to update job attempts:', error)
+  }
+}
+
 function timestampNow(): number {
   const now = Date.now()
   return Math.floor(now / 1000)