chore: wip

chore: wip

Author: chrisbbreuer

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

@@ -26,61 +26,33 @@ export class QueueStack {
     const jobsDir = path.jobsPath()
 
     try {
-      const files = await fs.readdir(jobsDir)
-
-      for (const file of files) {
-        if (file.endsWith('.ts')) {
-          const filePath = path.jobsPath(file)
-
-          // Await the loading of the job module
-          const jobModule = await this.loadJobModule(filePath)
-
-          // Now you can safely access jobModule.default.rate
-          const rate = jobModule.default?.rate
-
-          // if no rate or job is disabled, no need to schedule, skip
-          if (!rate || jobModule.default?.enabled === false)
-            continue
-
-          // Convert the rate to a Schedule object
-          const schedule = Schedule.cron(this.cronScheduleFromRate(rate))
-
-          const id = `QueueRule${pascalCase(file.replace('.ts', ''))}`
-
-          // Perform operations with the jobModule.default as needed
-          const rule = new Rule(this.scope, id, {
-            // schedule to run every second
-            ruleName: `${this.props.appName}-${this.props.appEnv}-queue-rule-${slug(file.replace('.ts', ''))}`,
-            schedule,
-          })
-
-          rule.addTarget(new EcsTask({
-            cluster: this.props.cluster,
-            taskDefinition: this.props.taskDefinition,
-            containerOverrides: [
-              {
-                containerName: `${this.props.appName}-${this.props.appEnv}-api`,
-                environment: [
-                  {
-                    name: 'QUEUE_WORKER',
-                    value: 'true',
-                  },
-                  {
-                    name: 'JOB',
-                    value: file,
-                  },
-                ],
-              },
-            ],
-
-            retryAttempts: jobModule.default.tries || 3,
-
-            subnetSelection: {
-              subnetType: ec2.SubnetType.PUBLIC, // SubnetType.PRIVATE_WITH_EGRESS
-            },
-          }))
-        }
+      const jobFiles = await fs.readdir(jobsDir)
+      const actionFiles = await fs.readdir(actionsDir)
+      const jobs = []
+
+      // then, need to loop through all app/Jobs/*.ts and create a rule for each, potentially overwriting the Schedule.ts jobs
+      for (const file of jobFiles) {
+        if (!file.endsWith('.ts'))
+          continue
+
+        const filePath = path.jobsPath(file)
+
+        // Await the loading of the job module
+        jobs.push(await this.loadModule(filePath))
+      }
+
+      for (const file of actionFiles) {
+        if (!file.endsWith('.ts'))
+          continue
+
+        const filePath = path.actionsPath(file)
+
+        // Await the loading of the job module
+        jobs.push(await this.loadModule(filePath))
       }
+
+      for (const job of jobs)
+        await this.createQueueRule(job)
     }
     catch (err) {
       console.error('Error reading the jobs directory:', err)
@@ -100,9 +72,73 @@ export class QueueStack {
     }
   }
 
-  async loadJobModule(filePath: string) {
+  async loadModule(filePath: string) {
     const jobModule = await import(filePath)
 
     return jobModule
   }
+
+  createQueueRule(job: any) {
+    // Now you can safely access job.default.rate
+    const rate = job.default?.rate
+
+    // if no rate or job is disabled, no need to schedule, skip
+    if (!rate || job.default?.enabled === false)
+      return
+
+    // Convert the rate to a Schedule object
+    const schedule = Schedule.cron(this.cronScheduleFromRate(rate))
+
+    const id = `QueueRule${pascalCase(file.replace('.ts', ''))}`
+
+    // Perform operations with the jobModule.default as needed
+    const rule = new Rule(this.scope, id, {
+      // schedule to run every second
+      ruleName: `${this.props.appName}-${this.props.appEnv}-queue-rule-${slug(file.replace('.ts', ''))}`,
+      schedule,
+    })
+
+    rule.addTarget(new EcsTask({
+      cluster: this.props.cluster,
+      taskDefinition: this.props.taskDefinition,
+      containerOverrides: [
+        {
+          containerName: `${this.props.appName}-${this.props.appEnv}-api`,
+          environment: [
+            {
+              name: 'QUEUE_WORKER',
+              value: 'true',
+            },
+            {
+              name: 'JOB',
+              value: file,
+            },
+            {
+              name: 'JOB_BACKOFF_FACTOR',
+              value: jobModule.default?.backoffFactor,
+            },
+            {
+              name: 'JOB_RETRIES',
+              value: jobModule.default?.tries,
+            },
+            {
+              name: 'JOB_INITIAL_DELAY',
+              value: jobModule.default?.initialDelay,
+            },
+            {
+              name: 'JOB_JITTER',
+              value: jobModule.default?.jitter,
+            },
+          ],
+        },
+      ],
+
+      retryAttempts: 1, // we utilize a custom retry mechanism in the job itself
+      // retryAttempts: jobModule.default.tries || 3,
+
+      subnetSelection: {
+        subnetType: ec2.SubnetType.PUBLIC, // SubnetType.PRIVATE_WITH_EGRESS
+      },
+    }))
+  }
 }

storage/framework/core/path/src/index.ts

@@ -366,10 +366,6 @@ export function scriptsPath(path?: string) {
   return frameworkPath(`scripts/${path || ''}`)
 }
 
-export function schedulePath() {
-  return appPath(`Schedule.ts`)
-}
-
 export function schedulerPath(path?: string) {
   return corePath(`scheduler/${path || ''}`)
 }
@@ -527,7 +523,6 @@ export const path = {
   signalsPath,
   slugPath,
   scriptsPath,
-  schedulePath,
   securityPath,
   serverPath,
   serverlessPath,