Merge pull request #15 from xtr-dev/dev

Add automatic job scheduling and rescheduling

package.json

@@ -1,6 +1,6 @@
 {
   "name": "@xtr-dev/payload-mailing",
-  "version": "0.0.4",
+  "version": "0.0.8",
   "description": "Template-based email system with scheduling and job processing for PayloadCMS",
   "type": "module",
   "main": "dist/index.js",

src/plugin.ts

@@ -4,9 +4,61 @@ import { MailingService } from './services/MailingService.js'
 import { createEmailTemplatesCollection } from './collections/EmailTemplates.js'
 import Emails from './collections/Emails.js'
 
+// Helper function to schedule the email processing job
+async function scheduleEmailProcessingJob(payload: any, queueName: string, delayMs: number = 60000): Promise<boolean> {
+  if (!queueName || typeof queueName !== 'string') {
+    throw new Error('Invalid queueName: must be a non-empty string')
+  }
+
+  const jobSlug = 'process-email-queue'
+
+  // Check if there's already a scheduled job for this task
+  const existingJobs = await payload.find({
+    collection: 'payload-jobs',
+    where: {
+      and: [
+        {
+          taskSlug: {
+            equals: jobSlug,
+          },
+        },
+        {
+          hasCompleted: {
+            equals: false,
+          },
+        },
+      ],
+    },
+    limit: 1,
+  })
+
+  // If no existing job, schedule a new one
+  if (existingJobs.docs.length === 0) {
+    await payload.create({
+      collection: 'payload-jobs',
+      data: {
+        taskSlug: jobSlug,
+        input: {},
+        queue: queueName,
+        waitUntil: new Date(Date.now() + delayMs),
+      },
+    })
+    console.log(`🔄 Scheduled email processing job in queue: ${queueName}`)
+    return true
+  } else {
+    console.log(`✅ Email processing job already scheduled in queue: ${queueName}`)
+    return false
+  }
+}
+
 export const mailingPlugin = (pluginConfig: MailingPluginConfig) => (config: Config): Config => {
   const queueName = pluginConfig.queue || 'default'
 
+  // Validate queueName
+  if (!queueName || typeof queueName !== 'string') {
+    throw new Error('Invalid queue configuration: queue must be a non-empty string')
+  }
+
   // Handle templates collection configuration
   const templatesConfig = pluginConfig.collections?.templates
   const templatesSlug = typeof templatesConfig === 'string' ? templatesConfig : 'email-templates'
@@ -80,8 +132,11 @@ export const mailingPlugin = (pluginConfig: MailingPluginConfig) => (config: Con
         {
           slug: 'process-email-queue',
           handler: async ({ job, req }: { job: any; req: any }) => {
+            const payload = (req as any).payload
+            let jobResult = null
+
             try {
-              const mailingService = new MailingService((req as any).payload, pluginConfig)
+              const mailingService = new MailingService(payload, pluginConfig)
 
               console.log('🔄 Processing email queue (pending + failed emails)...')
 
@@ -91,19 +146,41 @@ export const mailingPlugin = (pluginConfig: MailingPluginConfig) => (config: Con
               // Then retry failed emails
               await mailingService.retryFailedEmails()
 
-              return {
+              jobResult = {
                 output: {
                   success: true,
                   message: 'Email queue processed successfully (pending and failed emails)'
                 }
               }
+
+              console.log('✅ Email queue processing completed successfully')
+
             } catch (error) {
               console.error('❌ Error processing email queue:', error)
               const errorMessage = error instanceof Error ? error.message : 'Unknown error'
 
-              // Properly fail the job by throwing the error
-              throw new Error(`Email queue processing failed: ${errorMessage}`)
+              jobResult = new Error(`Email queue processing failed: ${errorMessage}`)
             }
+
+            // Always reschedule the next job (success or failure) using duplicate prevention
+            let rescheduled = false
+            try {
+              rescheduled = await scheduleEmailProcessingJob(payload, queueName, 300000) // Reschedule in 5 minutes
+              if (rescheduled) {
+                console.log(`🔄 Rescheduled next email processing job in ${queueName} queue`)
+              }
+            } catch (rescheduleError) {
+              console.error('❌ Failed to reschedule email processing job:', rescheduleError)
+              // If rescheduling fails, we should warn but not fail the current job
+              // since the email processing itself may have succeeded
+              console.warn('⚠️  Email processing completed but next job could not be scheduled')
+            }
+
+            // Return the original result or throw the error
+            if (jobResult instanceof Error) {
+              throw jobResult
+            }
+            return jobResult
           },
           interfaceName: 'ProcessEmailQueueJob',
         },
@@ -130,6 +207,13 @@ export const mailingPlugin = (pluginConfig: MailingPluginConfig) => (config: Con
 
       console.log('PayloadCMS Mailing Plugin initialized successfully')
 
+      // Schedule the email processing job if not already scheduled
+      try {
+        await scheduleEmailProcessingJob(payload, queueName)
+      } catch (error) {
+        console.error('Failed to schedule email processing job:', error)
+      }
+
       // Call onReady callback if provided
       if (pluginConfig.onReady) {
         await pluginConfig.onReady(payload)

src/services/MailingService.ts

@@ -15,7 +15,7 @@ import { serializeRichTextToHTML, serializeRichTextToText } from '../utils/richT
 export class MailingService implements IMailingService {
   private payload: Payload
   private config: MailingPluginConfig
-  private transporter!: Transporter
+  private transporter!: Transporter | any
   private templatesCollection: string
   private emailsCollection: string
 
@@ -40,11 +40,28 @@ export class MailingService implements IMailingService {
       } else {
         this.transporter = nodemailer.createTransport(this.config.transport as MailingTransportConfig)
       }
+    } else if (this.payload.email && 'sendMail' in this.payload.email) {
+      // Use Payload's configured mailer (cast to any to handle different adapter types)
+      this.transporter = this.payload.email as any
     } else {
-      throw new Error('Email transport configuration is required')
+      throw new Error('Email transport configuration is required either in plugin config or Payload config')
     }
   }
 
+  private getDefaultFrom(): string {
+    const fromEmail = this.config.defaultFrom
+    const fromName = this.config.defaultFromName
+
+    // Check if fromName exists, is not empty after trimming, and fromEmail exists
+    if (fromName && fromName.trim() && fromEmail) {
+      // Escape quotes in the display name to prevent malformed headers
+      const escapedName = fromName.replace(/"/g, '\\"')
+      return `"${escapedName}" <${fromEmail}>`
+    }
+
+    return fromEmail || ''
+  }
+
   private registerHandlebarsHelpers(): void {
     Handlebars.registerHelper('formatDate', (date: Date, format?: string) => {
       if (!date) return ''
@@ -125,7 +142,7 @@ export class MailingService implements IMailingService {
       to: Array.isArray(options.to) ? options.to : [options.to],
       cc: options.cc ? (Array.isArray(options.cc) ? options.cc : [options.cc]) : undefined,
       bcc: options.bcc ? (Array.isArray(options.bcc) ? options.bcc : [options.bcc]) : undefined,
-      from: options.from || this.config.defaultFrom,
+      from: options.from || this.getDefaultFrom(),
       replyTo: options.replyTo,
       subject: subject || options.subject,
       html,
@@ -242,7 +259,7 @@ export class MailingService implements IMailingService {
       }) as QueuedEmail
 
       let emailObject: EmailObject = {
-        from: email.from || this.config.defaultFrom,
+        from: email.from || this.getDefaultFrom(),
         to: email.to,
         cc: email.cc || undefined,
         bcc: email.bcc || undefined,
@@ -259,7 +276,7 @@ export class MailingService implements IMailingService {
       }
 
       const mailOptions = {
-        from: emailObject.from || this.config.defaultFrom,
+        from: emailObject.from,
         to: emailObject.to,
         cc: emailObject.cc || undefined,
         bcc: emailObject.bcc || undefined,

src/types/index.ts

@@ -22,6 +22,7 @@ export interface MailingPluginConfig {
     emails?: string | Partial<CollectionConfig>
   }
   defaultFrom?: string
+  defaultFromName?: string
   transport?: Transporter | MailingTransportConfig
   queue?: string
   retryAttempts?: number