Merge pull request #57 from xtr-dev/dev

Fix TypeScript build error in jobScheduler.ts

dev/app/api/test-email/route.ts

@@ -1,6 +1,6 @@
 import { getPayload } from 'payload'
 import config from '@payload-config'
-import { sendEmail, processEmailById } from '@xtr-dev/payload-mailing'
+import { sendEmail } from '@xtr-dev/payload-mailing'
 
 export async function POST(request: Request) {
   try {
@@ -55,35 +55,21 @@ export async function POST(request: Request) {
       emailOptions.data.scheduledAt = scheduledAt ? new Date(scheduledAt) : new Date(Date.now() + 60000)
     }
 
+    // Set processImmediately for "send now" type
+    const processImmediately = (type === 'send' && !scheduledAt)
+    emailOptions.processImmediately = processImmediately
+
     const result = await sendEmail(payload, emailOptions)
 
-    // If it's "send now" (not scheduled), process the email immediately
-    if (type === 'send' && !scheduledAt) {
-      try {
-        await processEmailById(payload, String(result.id))
     return Response.json({
       success: true,
       emailId: result.id,
-          message: 'Email sent successfully',
-          status: 'sent'
-        })
-      } catch (processError) {
-        // If immediate processing fails, return that it's queued
-        console.warn('Failed to process email immediately, left in queue:', processError)
-        return Response.json({
-          success: true,
-          emailId: result.id,
-          message: 'Email queued successfully (immediate processing failed)',
-          status: 'queued'
-        })
-      }
-    }
-
-    return Response.json({
-      success: true,
-      emailId: result.id,
-      message: scheduledAt ? 'Email scheduled successfully' : 'Email queued successfully',
-      status: scheduledAt ? 'scheduled' : 'queued'
+      message: processImmediately ? 'Email sent successfully' :
+               scheduledAt ? 'Email scheduled successfully' :
+               'Email queued successfully',
+      status: processImmediately ? 'sent' :
+              scheduledAt ? 'scheduled' :
+              'queued'
     })
   } catch (error) {
     console.error('Test email error:', error)

package.json

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

src/collections/Emails.ts

@@ -1,10 +1,12 @@
 import type { CollectionConfig } from 'payload'
+import { findExistingJobs, ensureEmailJob, updateEmailJobRelationship } from '../utils/jobScheduler.js'
+import { createContextLogger } from '../utils/logger.js'
 
 const Emails: CollectionConfig = {
   slug: 'emails',
   admin: {
     useAsTitle: 'subject',
-    defaultColumns: ['subject', 'to', 'status', 'scheduledAt', 'sentAt'],
+    defaultColumns: ['subject', 'to', 'status', 'jobs', 'scheduledAt', 'sentAt'],
     group: 'Mailing',
     description: 'Email delivery and status tracking',
   },
@@ -164,22 +166,76 @@ const Emails: CollectionConfig = {
         description: 'Email priority (1=highest, 10=lowest)',
       },
     },
+    {
+      name: 'jobs',
+      type: 'relationship',
+      relationTo: 'payload-jobs',
+      hasMany: true,
+      admin: {
+        description: 'Processing jobs associated with this email',
+        allowCreate: false,
+        readOnly: true,
+      },
+      filterOptions: ({ id }) => {
+        return {
+          'input.emailId': {
+            equals: id,
+          },
+        }
+      },
+    },
   ],
+  hooks: {
+    // Simple approach: Only use afterChange hook for job management
+    // This avoids complex interaction between hooks and ensures document ID is always available
+    afterChange: [
+      async ({ doc, previousDoc, req, operation }) => {
+        // Skip if:
+        // 1. Email is not pending status
+        // 2. Jobs are not configured
+        // 3. Email already has jobs (unless status just changed to pending)
+
+        const shouldSkip =
+          doc.status !== 'pending' ||
+          !req.payload.jobs ||
+          (doc.jobs?.length > 0 && previousDoc?.status === 'pending')
+
+        if (shouldSkip) {
+          return
+        }
+
+        try {
+          // Ensure a job exists for this email
+          // This function handles:
+          // - Checking for existing jobs (duplicate prevention)
+          // - Creating new job if needed
+          // - Returning all job IDs
+          const result = await ensureEmailJob(req.payload, doc.id, {
+            scheduledAt: doc.scheduledAt,
+          })
+
+          // Update the email's job relationship if we have jobs
+          // This handles both new jobs and existing jobs that weren't in the relationship
+          if (result.jobIds.length > 0) {
+            await updateEmailJobRelationship(req.payload, doc.id, result.jobIds, 'emails')
+          }
+        } catch (error) {
+          // Log error but don't throw - we don't want to fail the email operation
+          const logger = createContextLogger(req.payload, 'EMAILS_HOOK')
+          logger.error(`Failed to ensure job for email ${doc.id}:`, error)
+        }
+      }
+    ]
+  },
   timestamps: true,
-  // indexes: [
-  //   {
-  //     fields: {
-  //       status: 1,
-  //       scheduledAt: 1,
-  //     },
-  //   },
-  //   {
-  //     fields: {
-  //       priority: -1,
-  //       createdAt: 1,
-  //     },
-  //   },
-  // ],
+  indexes: [
+    {
+      fields: ['status', 'scheduledAt'],
+    },
+    {
+      fields: ['priority', 'createdAt'],
+    },
+  ],
 }
 
 export default Emails

src/index.ts

@@ -11,9 +11,9 @@ export { MailingService } from './services/MailingService.js'
 export { default as EmailTemplates, createEmailTemplatesCollection } from './collections/EmailTemplates.js'
 export { default as Emails } from './collections/Emails.js'
 
-// Jobs (includes the send email task)
-export { mailingJobs, sendEmailJob } from './jobs/index.js'
-export type { SendEmailTaskInput } from './jobs/sendEmailTask.js'
+// Jobs (includes the individual email processing job)
+export { mailingJobs } from './jobs/index.js'
+export type { ProcessEmailJobInput } from './jobs/processEmailJob.js'
 
 // Main email sending function
 export { sendEmail, type SendEmailOptions } from './sendEmail.js'
@@ -26,7 +26,12 @@ export {
   processEmails,
   retryFailedEmails,
   parseAndValidateEmails,
+  sanitizeDisplayName,
+  sanitizeFromName,
 } from './utils/helpers.js'
 
 // Email processing utilities
-export { processEmailById, processAllEmails } from './utils/emailProcessor.js'
+export { processEmailById, processJobById, processAllEmails } from './utils/emailProcessor.js'
+
+// Job scheduling utilities
+export { findExistingJobs, ensureEmailJob, updateEmailJobRelationship } from './utils/jobScheduler.js'

src/jobs/index.ts

@@ -1,14 +1,16 @@
 import { processEmailsJob } from './processEmailsTask.js'
-import { sendEmailJob } from './sendEmailTask.js'
+import { processEmailJob } from './processEmailJob.js'
 
 /**
  * All mailing-related jobs that get registered with Payload
+ *
+ * Note: The sendEmailJob has been removed as each email now gets its own individual processEmailJob
  */
 export const mailingJobs = [
-  processEmailsJob,
-  sendEmailJob,
+  processEmailsJob, // Kept for backward compatibility and batch processing if needed
+  processEmailJob,  // New individual email processing job
 ]
 
 // Re-export everything from individual job files
 export * from './processEmailsTask.js'
-export * from './sendEmailTask.js'
+export * from './processEmailJob.js'

src/jobs/processEmailJob.ts

@@ -0,0 +1,72 @@
+import type { PayloadRequest } from 'payload'
+import { processEmailById } from '../utils/emailProcessor.js'
+
+/**
+ * Data passed to the individual email processing job
+ */
+export interface ProcessEmailJobInput {
+  /**
+   * The ID of the email to process
+   */
+  emailId: string | number
+}
+
+/**
+ * Job definition for processing a single email
+ * This replaces the batch processing approach with individual email jobs
+ */
+export const processEmailJob = {
+  slug: 'process-email',
+  label: 'Process Individual Email',
+  inputSchema: [
+    {
+      name: 'emailId',
+      type: 'text' as const,
+      required: true,
+      label: 'Email ID',
+      admin: {
+        description: 'The ID of the email to process and send'
+      }
+    }
+  ],
+  outputSchema: [
+    {
+      name: 'success',
+      type: 'checkbox' as const
+    },
+    {
+      name: 'emailId',
+      type: 'text' as const
+    },
+    {
+      name: 'status',
+      type: 'text' as const
+    }
+  ],
+  handler: async ({ input, req }: { input: ProcessEmailJobInput; req: PayloadRequest }) => {
+    const payload = (req as any).payload
+    const { emailId } = input
+
+    if (!emailId) {
+      throw new Error('Email ID is required for processing')
+    }
+
+    try {
+      // Process the individual email
+      await processEmailById(payload, String(emailId))
+
+      return {
+        output: {
+          success: true,
+          emailId: String(emailId),
+          status: 'sent',
+          message: `Email ${emailId} processed successfully`
+        }
+      }
+    } catch (error) {
+      throw new Error(`Failed to process email ${emailId}: ${String(error)}`)
+    }
+  }
+}
+
+export default processEmailJob

src/jobs/processEmailsTask.ts

@@ -1,5 +1,6 @@
 import type { PayloadRequest, Payload } from 'payload'
 import { processAllEmails } from '../utils/emailProcessor.js'
+import { createContextLogger } from '../utils/logger.js'
 
 /**
  * Data passed to the process emails task
@@ -67,7 +68,8 @@ export const scheduleEmailsJob = async (
   delay?: number
 ) => {
   if (!payload.jobs) {
-    console.warn('PayloadCMS jobs not configured - emails will not be processed automatically')
+    const logger = createContextLogger(payload, 'SCHEDULER')
+    logger.warn('PayloadCMS jobs not configured - emails will not be processed automatically')
     return
   }
 
@@ -79,6 +81,7 @@ export const scheduleEmailsJob = async (
       waitUntil: delay ? new Date(Date.now() + delay) : undefined,
     } as any)
   } catch (error) {
-    console.error('Failed to schedule email processing job:', error)
+    const logger = createContextLogger(payload, 'SCHEDULER')
+    logger.error('Failed to schedule email processing job:', error)
   }
 }

src/jobs/sendEmailTask.ts

@@ -1,256 +0,0 @@
-import { sendEmail } from '../sendEmail.js'
-import { BaseEmailDocument } from '../types/index.js'
-import { processEmailById } from '../utils/emailProcessor.js'
-
-export interface SendEmailTaskInput {
-  // Template mode fields
-  templateSlug?: string
-  variables?: Record<string, any>
-
-  // Direct email mode fields
-  subject?: string
-  html?: string
-  text?: string
-
-  // Common fields
-  to: string | string[]
-  cc?: string | string[]
-  bcc?: string | string[]
-  from?: string
-  fromName?: string
-  replyTo?: string
-  scheduledAt?: string | Date // ISO date string or Date object
-  priority?: number
-  processImmediately?: boolean // If true, process the email immediately instead of waiting for the queue
-
-  // Allow any additional fields that users might have in their email collection
-  [key: string]: any
-}
-
-/**
- * Transforms task input into sendEmail options by separating template and data fields
- */
-function transformTaskInputToSendEmailOptions(taskInput: SendEmailTaskInput) {
-  const sendEmailOptions: any = {
-    data: {}
-  }
-
-  // If using template mode, set template options
-  if (taskInput.templateSlug) {
-    sendEmailOptions.template = {
-      slug: taskInput.templateSlug,
-      variables: taskInput.variables || {}
-    }
-  }
-
-  // Standard email fields that should be copied to data
-  const standardFields = ['to', 'cc', 'bcc', 'from', 'fromName', 'replyTo', 'subject', 'html', 'text', 'scheduledAt', 'priority']
-
-  // Fields that should not be copied to data
-  const excludedFields = ['templateSlug', 'variables', 'processImmediately']
-
-  // Copy standard fields to data
-  standardFields.forEach(field => {
-    if (taskInput[field] !== undefined) {
-      sendEmailOptions.data[field] = taskInput[field]
-    }
-  })
-
-  // Copy any additional custom fields that aren't excluded or standard fields
-  Object.keys(taskInput).forEach(key => {
-    if (!excludedFields.includes(key) && !standardFields.includes(key)) {
-      sendEmailOptions.data[key] = taskInput[key]
-    }
-  })
-
-  return sendEmailOptions
-}
-
-/**
- * Job definition for sending emails
- * Can be used through Payload's job queue system to send emails programmatically
- */
-export const sendEmailJob = {
-  slug: 'send-email',
-  label: 'Send Email',
-  inputSchema: [
-    {
-      name: 'processImmediately',
-      type: 'checkbox' as const,
-      label: 'Process Immediately',
-      defaultValue: false,
-      admin: {
-        description: 'Process and send the email immediately instead of waiting for the queue processor'
-      }
-    },
-    {
-      name: 'templateSlug',
-      type: 'text' as const,
-      label: 'Template Slug',
-      admin: {
-        description: 'Use a template (leave empty for direct email)',
-        condition: (data: any) => !data.subject && !data.html
-      }
-    },
-    {
-      name: 'variables',
-      type: 'json' as const,
-      label: 'Template Variables',
-      admin: {
-        description: 'JSON object with variables for template rendering',
-        condition: (data: any) => Boolean(data.templateSlug)
-      }
-    },
-    {
-      name: 'subject',
-      type: 'text' as const,
-      label: 'Subject',
-      admin: {
-        description: 'Email subject (required if not using template)',
-        condition: (data: any) => !data.templateSlug
-      }
-    },
-    {
-      name: 'html',
-      type: 'textarea' as const,
-      label: 'HTML Content',
-      admin: {
-        description: 'HTML email content (required if not using template)',
-        condition: (data: any) => !data.templateSlug
-      }
-    },
-    {
-      name: 'text',
-      type: 'textarea' as const,
-      label: 'Text Content',
-      admin: {
-        description: 'Plain text email content (optional)',
-        condition: (data: any) => !data.templateSlug
-      }
-    },
-    {
-      name: 'to',
-      type: 'text' as const,
-      required: true,
-      label: 'To (Email Recipients)',
-      admin: {
-        description: 'Comma-separated list of email addresses'
-      }
-    },
-    {
-      name: 'cc',
-      type: 'text' as const,
-      label: 'CC (Carbon Copy)',
-      admin: {
-        description: 'Optional comma-separated list of CC email addresses'
-      }
-    },
-    {
-      name: 'bcc',
-      type: 'text' as const,
-      label: 'BCC (Blind Carbon Copy)',
-      admin: {
-        description: 'Optional comma-separated list of BCC email addresses'
-      }
-    },
-    {
-      name: 'from',
-      type: 'text' as const,
-      label: 'From Email',
-      admin: {
-        description: 'Optional sender email address (uses default if not provided)'
-      }
-    },
-    {
-      name: 'fromName',
-      type: 'text' as const,
-      label: 'From Name',
-      admin: {
-        description: 'Optional sender display name (e.g., "John Doe")'
-      }
-    },
-    {
-      name: 'replyTo',
-      type: 'text' as const,
-      label: 'Reply To',
-      admin: {
-        description: 'Optional reply-to email address'
-      }
-    },
-    {
-      name: 'scheduledAt',
-      type: 'date' as const,
-      label: 'Schedule For',
-      admin: {
-        description: 'Optional date/time to schedule email for future delivery',
-        condition: (data: any) => !data.processImmediately
-      }
-    },
-    {
-      name: 'priority',
-      type: 'number' as const,
-      label: 'Priority',
-      min: 1,
-      max: 10,
-      defaultValue: 5,
-      admin: {
-        description: 'Email priority (1 = highest, 10 = lowest)'
-      }
-    }
-  ],
-  outputSchema: [
-    {
-      name: 'id',
-      type: 'text' as const
-    }
-  ],
-  handler: async ({ input, payload }: any) => {
-    // Cast input to our expected type
-    const taskInput = input as SendEmailTaskInput
-    const shouldProcessImmediately = taskInput.processImmediately || false
-
-    try {
-      // Transform task input into sendEmail options using helper function
-      const sendEmailOptions = transformTaskInputToSendEmailOptions(taskInput)
-
-      // Use the sendEmail helper to create the email
-      const email = await sendEmail<BaseEmailDocument>(payload, sendEmailOptions)
-
-      // If processImmediately is true, process the email now
-      if (shouldProcessImmediately) {
-        console.log(`⚡ Processing email ${email.id} immediately...`)
-        await processEmailById(payload, String(email.id))
-        console.log(`✅ Email ${email.id} processed and sent immediately`)
-
-        return {
-          output: {
-            success: true,
-            id: email.id,
-            status: 'sent',
-            processedImmediately: true
-          }
-        }
-      }
-
-      return {
-        output: {
-          success: true,
-          id: email.id,
-          status: 'queued',
-          processedImmediately: false
-        }
-      }
-
-    } catch (error) {
-      // Re-throw Error instances to preserve stack trace and error context
-      if (error instanceof Error) {
-        throw error
-      } else {
-        // Only wrap non-Error values
-        throw new Error(`Failed to process email: ${String(error)}`)
-      }
-    }
-  }
-}
-
-export default sendEmailJob

src/plugin.ts

@@ -3,7 +3,7 @@ import { MailingPluginConfig, MailingContext } from './types/index.js'
 import { MailingService } from './services/MailingService.js'
 import { createEmailTemplatesCollection } from './collections/EmailTemplates.js'
 import Emails from './collections/Emails.js'
-import { mailingJobs, scheduleEmailsJob } from './jobs/index.js'
+import { mailingJobs } from './jobs/index.js'
 
 
 export const mailingPlugin = (pluginConfig: MailingPluginConfig) => (config: Config): Config => {
@@ -106,18 +106,6 @@ export const mailingPlugin = (pluginConfig: MailingPluginConfig) => (config: Con
         },
       } as MailingContext
 
-      // Schedule the initial email processing job
-      try {
-        await scheduleEmailsJob(payload, queueName, 60000) // Schedule in 1 minute
-      } catch (error) {
-        console.error('Failed to schedule email processing job:', error)
-      }
-
-      // Call onReady callback if provided
-      if (pluginConfig.onReady) {
-        await pluginConfig.onReady(payload)
-      }
-
       if (pluginConfig.initOrder !== 'after' && config.onInit) {
         await config.onInit(payload)
       }

src/sendEmail.ts

@@ -1,6 +1,8 @@
 import { Payload } from 'payload'
-import { getMailing, renderTemplate, parseAndValidateEmails } from './utils/helpers.js'
+import { getMailing, renderTemplate, parseAndValidateEmails, sanitizeFromName } from './utils/helpers.js'
 import { BaseEmailDocument } from './types/index.js'
+import { processJobById } from './utils/emailProcessor.js'
+import { createContextLogger } from './utils/logger.js'
 
 // Options for sending emails
 export interface SendEmailOptions<T extends BaseEmailDocument = BaseEmailDocument> {
@@ -13,6 +15,8 @@ export interface SendEmailOptions<T extends BaseEmailDocument = BaseEmailDocumen
   data?: Partial<T>
   // Common options
   collectionSlug?: string // defaults to 'emails'
+  processImmediately?: boolean // if true, creates job and processes it immediately
+  queue?: string // queue name for the job, defaults to mailing config queue
 }
 
 /**
@@ -39,8 +43,8 @@ export const sendEmail = async <TEmail extends BaseEmailDocument = BaseEmailDocu
   payload: Payload,
   options: SendEmailOptions<TEmail>
 ): Promise<TEmail> => {
-  const mailing = getMailing(payload)
-  const collectionSlug = options.collectionSlug || mailing.collections.emails || 'emails'
+  const mailingConfig = getMailing(payload)
+  const collectionSlug = options.collectionSlug || mailingConfig.collections.emails || 'emails'
 
   let emailData: Partial<TEmail> = { ...options.data } as Partial<TEmail>
 
@@ -101,15 +105,7 @@ export const sendEmail = async <TEmail extends BaseEmailDocument = BaseEmailDocu
   }
 
   // Sanitize fromName to prevent header injection
-  if (emailData.fromName) {
-    emailData.fromName = emailData.fromName
-      .trim()
-      // Remove/replace newlines and carriage returns to prevent header injection
-      .replace(/[\r\n]/g, ' ')
-      // Remove control characters (except space and printable characters)
-      .replace(/[\x00-\x1F\x7F-\x9F]/g, '')
-      // Note: We don't escape quotes here as that's handled in MailingService
-  }
+  emailData.fromName = sanitizeFromName(emailData.fromName as string)
 
   // Normalize Date objects to ISO strings for consistent database storage
   if (emailData.scheduledAt instanceof Date) {
@@ -129,6 +125,7 @@ export const sendEmail = async <TEmail extends BaseEmailDocument = BaseEmailDocu
   }
 
   // Create the email in the collection with proper typing
+  // The hooks will automatically create and populate the job relationship
   const email = await payload.create({
     collection: collectionSlug,
     data: emailData
@@ -139,6 +136,93 @@ export const sendEmail = async <TEmail extends BaseEmailDocument = BaseEmailDocu
     throw new Error('Failed to create email: invalid response from database')
   }
 
+  // If processImmediately is true, get the job from the relationship and process it now
+  if (options.processImmediately) {
+    const logger = createContextLogger(payload, 'IMMEDIATE')
+    logger.debug(`Starting immediate processing for email ${email.id}`)
+
+    if (!payload.jobs) {
+      throw new Error('PayloadCMS jobs not configured - cannot process email immediately')
+    }
+
+    // Poll for the job with optimized backoff and timeout protection
+    // This handles the async nature of hooks and ensures we wait for job creation
+    const maxAttempts = 5 // Reduced from 10 to minimize delay
+    const initialDelay = 25 // Reduced from 50ms for faster response
+    const maxTotalTime = 3000 // 3 second total timeout
+    const startTime = Date.now()
+    let jobId: string | undefined
+
+    logger.debug(`Polling for job creation (max ${maxAttempts} attempts, ${maxTotalTime}ms timeout)`)
+
+    for (let attempt = 0; attempt < maxAttempts; attempt++) {
+      // Check total timeout before continuing
+      if (Date.now() - startTime > maxTotalTime) {
+        throw new Error(
+          `Job polling timed out after ${maxTotalTime}ms for email ${email.id}. ` +
+          `The auto-scheduling may have failed or is taking longer than expected.`
+        )
+      }
+
+      // Calculate delay with exponential backoff (25ms, 50ms, 100ms, 200ms, 400ms)
+      // Cap at 400ms per attempt for better responsiveness
+      const delay = Math.min(initialDelay * Math.pow(2, attempt), 400)
+
+      if (attempt > 0) {
+        await new Promise(resolve => setTimeout(resolve, delay))
+      }
+
+      // Refetch the email to check for jobs
+      const emailWithJobs = await payload.findByID({
+        collection: collectionSlug,
+        id: email.id,
+      })
+
+      logger.debug(`Attempt ${attempt + 1}/${maxAttempts}: Found ${emailWithJobs.jobs?.length || 0} jobs for email ${email.id}`)
+
+      if (emailWithJobs.jobs && emailWithJobs.jobs.length > 0) {
+        // Job found! Get the first job ID (should only be one for a new email)
+        jobId = Array.isArray(emailWithJobs.jobs)
+          ? String(emailWithJobs.jobs[0])
+          : String(emailWithJobs.jobs)
+        logger.info(`Found job ID: ${jobId}`)
+        break
+      }
+
+      // Log on later attempts to help with debugging (reduced threshold)
+      if (attempt >= 1) {
+        logger.debug(`Waiting for job creation for email ${email.id}, attempt ${attempt + 1}/${maxAttempts}`)
+      }
+    }
+
+    if (!jobId) {
+      // Distinguish between different failure scenarios for better error handling
+      const timeoutMsg = Date.now() - startTime >= maxTotalTime
+      const errorType = timeoutMsg ? 'POLLING_TIMEOUT' : 'JOB_NOT_FOUND'
+
+      const baseMessage = timeoutMsg
+        ? `Job polling timed out after ${maxTotalTime}ms for email ${email.id}`
+        : `No processing job found for email ${email.id} after ${maxAttempts} attempts (${Date.now() - startTime}ms)`
+
+      throw new Error(
+        `${errorType}: ${baseMessage}. ` +
+        `This indicates the email was created but job auto-scheduling failed. ` +
+        `The email exists in the database but immediate processing cannot proceed. ` +
+        `You may need to: 1) Check job queue configuration, 2) Verify database hooks are working, ` +
+        `3) Process the email later using processEmailById('${email.id}').`
+      )
+    }
+
+    logger.info(`Starting job execution for job ${jobId}`)
+    try {
+      await processJobById(payload, jobId)
+      logger.info(`Successfully processed email ${email.id} immediately`)
+    } catch (error) {
+      logger.error(`Failed to process email ${email.id} immediately:`, error)
+      throw new Error(`Failed to process email ${email.id} immediately: ${String(error)}`)
+    }
+  }
+
   return email as TEmail
 }
 

src/services/MailingService.ts

@@ -7,6 +7,7 @@ import {
   BaseEmail, BaseEmailTemplate, BaseEmailDocument, BaseEmailTemplateDocument
 } from '../types/index.js'
 import { serializeRichTextToHTML, serializeRichTextToText } from '../utils/richTextSerializer.js'
+import { sanitizeDisplayName } from '../utils/helpers.js'
 
 export class MailingService implements IMailingService {
   public payload: Payload
@@ -44,17 +45,10 @@ export class MailingService implements IMailingService {
 
   /**
    * Sanitizes a display name for use in email headers to prevent header injection
-   * and ensure proper formatting
+   * Uses the centralized sanitization utility with quote escaping for headers
    */
   private sanitizeDisplayName(name: string): string {
-    return name
-      .trim()
-      // Remove/replace newlines and carriage returns to prevent header injection
-      .replace(/[\r\n]/g, ' ')
-      // Remove control characters (except space and printable characters)
-      .replace(/[\x00-\x1F\x7F-\x9F]/g, '')
-      // Escape quotes to prevent malformed headers
-      .replace(/"/g, '\\"')
+    return sanitizeDisplayName(name, true) // escapeQuotes = true for email headers
   }
 
   /**

src/types/index.ts

@@ -76,7 +76,6 @@ export interface MailingPluginConfig {
   templateEngine?: TemplateEngine
   richTextEditor?: RichTextField['editor']
   beforeSend?: BeforeSendHook
-  onReady?: (payload: any) => Promise<void>
   initOrder?: 'before' | 'after'
 }
 

src/utils/emailProcessor.ts

@@ -1,4 +1,5 @@
 import type { Payload } from 'payload'
+import { createContextLogger } from './logger.js'
 
 /**
  * Processes a single email by ID using the mailing service
@@ -29,6 +30,39 @@ export async function processEmailById(payload: Payload, emailId: string): Promi
   await mailingContext.service.processEmailItem(emailId)
 }
 
+/**
+ * Processes a job immediately by finding and executing it
+ * @param payload Payload instance
+ * @param jobId The ID of the job to run immediately
+ * @returns Promise that resolves when job is processed
+ */
+export async function processJobById(payload: Payload, jobId: string): Promise<void> {
+  const logger = createContextLogger(payload, 'PROCESSOR')
+  logger.debug(`Starting processJobById for job ${jobId}`)
+
+  if (!payload.jobs) {
+    throw new Error('PayloadCMS jobs not configured - cannot process job immediately')
+  }
+
+  try {
+    logger.debug(`Running job ${jobId} with payload.jobs.run()`)
+
+    // Run a specific job by its ID (using where clause to find the job)
+    const result = await payload.jobs.run({
+      where: {
+        id: {
+          equals: jobId
+        }
+      }
+    })
+
+    logger.info(`Job ${jobId} execution completed`, { result })
+  } catch (error) {
+    logger.error(`Job ${jobId} execution failed:`, error)
+    throw new Error(`Failed to process job ${jobId}: ${String(error)}`)
+  }
+}
+
 /**
  * Processes all pending and failed emails using the mailing service
  * @param payload Payload instance

src/utils/helpers.ts

@@ -36,6 +36,44 @@ export const parseAndValidateEmails = (emails: string | string[] | null | undefi
   return emailList
 }
 
+/**
+ * Sanitize display names to prevent email header injection
+ * Removes newlines, carriage returns, and control characters
+ * @param displayName - The display name to sanitize
+ * @param escapeQuotes - Whether to escape quotes (for email headers)
+ * @returns Sanitized display name
+ */
+export const sanitizeDisplayName = (displayName: string, escapeQuotes = false): string => {
+  if (!displayName) return displayName
+
+  let sanitized = displayName
+    .trim()
+    // Remove/replace newlines and carriage returns to prevent header injection
+    .replace(/[\r\n]/g, ' ')
+    // Remove control characters (except space and printable characters)
+    .replace(/[\x00-\x1F\x7F-\x9F]/g, '')
+
+  // Escape quotes if needed (for email headers)
+  if (escapeQuotes) {
+    sanitized = sanitized.replace(/"/g, '\\"')
+  }
+
+  return sanitized
+}
+
+/**
+ * Sanitize and validate fromName for emails
+ * Wrapper around sanitizeDisplayName for consistent fromName handling
+ * @param fromName - The fromName to sanitize
+ * @returns Sanitized fromName or undefined if empty after sanitization
+ */
+export const sanitizeFromName = (fromName: string | null | undefined): string | undefined => {
+  if (!fromName) return undefined
+
+  const sanitized = sanitizeDisplayName(fromName, false)
+  return sanitized.length > 0 ? sanitized : undefined
+}
+
 export const getMailing = (payload: Payload) => {
   const mailing = (payload as any).mailing
   if (!mailing) {

src/utils/jobScheduler.ts

@@ -0,0 +1,160 @@
+import type { Payload } from 'payload'
+import { createContextLogger } from './logger.js'
+
+/**
+ * Finds existing processing jobs for an email
+ */
+export async function findExistingJobs(
+  payload: Payload,
+  emailId: string | number
+): Promise<{ docs: any[], totalDocs: number }> {
+  return await payload.find({
+    collection: 'payload-jobs',
+    where: {
+      'input.emailId': {
+        equals: String(emailId),
+      },
+      task: {
+        equals: 'process-email',
+      },
+    },
+    limit: 10,
+  })
+}
+
+/**
+ * Ensures a processing job exists for an email
+ * Creates one if it doesn't exist, or returns existing job IDs
+ *
+ * This function is idempotent and safe for concurrent calls:
+ * - Uses atomic check-and-create pattern with retry logic
+ * - Multiple concurrent calls will only create one job
+ * - Database-level uniqueness prevents duplicate jobs
+ * - Race conditions are handled with exponential backoff retry
+ */
+export async function ensureEmailJob(
+  payload: Payload,
+  emailId: string | number,
+  options?: {
+    scheduledAt?: string | Date
+    queueName?: string
+  }
+): Promise<{ jobIds: (string | number)[], created: boolean }> {
+  if (!payload.jobs) {
+    throw new Error('PayloadCMS jobs not configured - cannot create email job')
+  }
+
+  const normalizedEmailId = String(emailId)
+  const mailingContext = (payload as any).mailing
+  const queueName = options?.queueName || mailingContext?.config?.queue || 'default'
+
+  const logger = createContextLogger(payload, 'JOB_SCHEDULER')
+  logger.debug(`Ensuring job for email ${normalizedEmailId}`)
+  logger.debug(`Queue: ${queueName}, scheduledAt: ${options?.scheduledAt || 'immediate'}`)
+
+  // First, optimistically try to create the job
+  // If it fails due to uniqueness constraint, then check for existing jobs
+  // This approach minimizes the race condition window
+
+  try {
+    logger.debug(`Attempting to create new job for email ${normalizedEmailId}`)
+    // Attempt to create job - rely on database constraints for duplicate prevention
+    const job = await payload.jobs.queue({
+      queue: queueName,
+      task: 'process-email',
+      input: {
+        emailId: normalizedEmailId
+      },
+      waitUntil: options?.scheduledAt ? new Date(options.scheduledAt) : undefined
+    })
+
+    logger.info(`Auto-scheduled processing job ${job.id} for email ${normalizedEmailId}`)
+    logger.debug(`Job details`, {
+      jobId: job.id,
+      emailId: normalizedEmailId,
+      scheduledAt: options?.scheduledAt || 'immediate',
+      task: 'process-email',
+      queue: queueName
+    })
+
+    return {
+      jobIds: [job.id],
+      created: true
+    }
+  } catch (createError) {
+    logger.warn(`Job creation failed for email ${normalizedEmailId}: ${String(createError)}`)
+
+    // Job creation failed - likely due to duplicate constraint or system issue
+
+    // Check if duplicate jobs exist (handles race condition where another process created job)
+    const existingJobs = await findExistingJobs(payload, normalizedEmailId)
+
+    logger.debug(`Found ${existingJobs.totalDocs} existing jobs after creation failure`)
+
+    if (existingJobs.totalDocs > 0) {
+      // Found existing jobs - return them (race condition handled successfully)
+      logger.info(`Using existing jobs for email ${normalizedEmailId}: ${existingJobs.docs.map(j => j.id).join(', ')}`)
+      return {
+        jobIds: existingJobs.docs.map(job => job.id),
+        created: false
+      }
+    }
+
+    // No existing jobs found - this is a genuine error
+    // Enhanced error context for better debugging
+    const errorMessage = String(createError)
+    const isLikelyUniqueConstraint = errorMessage.toLowerCase().includes('duplicate') ||
+                                     errorMessage.toLowerCase().includes('unique') ||
+                                     errorMessage.toLowerCase().includes('constraint')
+
+    if (isLikelyUniqueConstraint) {
+      // This should not happen if our check above worked, but provide a clear error
+      logger.error(`Unique constraint violation but no existing jobs found for email ${normalizedEmailId}`)
+      throw new Error(
+        `Database uniqueness constraint violation for email ${normalizedEmailId}, but no existing jobs found. ` +
+        `This indicates a potential data consistency issue. Original error: ${errorMessage}`
+      )
+    }
+
+    // Non-constraint related error
+    logger.error(`Non-constraint job creation error for email ${normalizedEmailId}: ${errorMessage}`)
+    throw new Error(`Failed to create job for email ${normalizedEmailId}: ${errorMessage}`)
+  }
+}
+
+/**
+ * Updates an email document to include job IDs in the relationship field
+ */
+export async function updateEmailJobRelationship(
+  payload: Payload,
+  emailId: string | number,
+  jobIds: (string | number)[],
+  collectionSlug: string = 'emails'
+): Promise<void> {
+  try {
+    const normalizedEmailId = String(emailId)
+    const normalizedJobIds = jobIds.map(id => String(id))
+
+    // Get current jobs to avoid overwriting
+    const currentEmail = await payload.findByID({
+      collection: collectionSlug,
+      id: normalizedEmailId,
+    })
+
+    const currentJobs = (currentEmail.jobs || []).map((job: any) => String(job))
+    const allJobs = [...new Set([...currentJobs, ...normalizedJobIds])] // Deduplicate with normalized strings
+
+    await payload.update({
+      collection: collectionSlug,
+      id: normalizedEmailId,
+      data: {
+        jobs: allJobs
+      }
+    })
+  } catch (error) {
+    const normalizedEmailId = String(emailId)
+    const logger = createContextLogger(payload, 'JOB_SCHEDULER')
+    logger.error(`Failed to update email ${normalizedEmailId} with job relationship:`, error)
+    throw error
+  }
+}

src/utils/logger.ts

@@ -0,0 +1,48 @@
+import type { Payload } from 'payload'
+
+let pluginLogger: any = null
+
+/**
+ * Get or create the plugin logger instance
+ * Uses PAYLOAD_MAILING_LOG_LEVEL environment variable to configure log level
+ * Defaults to 'info' if not set
+ */
+export function getPluginLogger(payload: Payload) {
+  if (!pluginLogger && payload.logger) {
+    const logLevel = process.env.PAYLOAD_MAILING_LOG_LEVEL || 'info'
+
+    pluginLogger = payload.logger.child({
+      level: logLevel,
+      plugin: '@xtr-dev/payload-mailing'
+    })
+
+    // Log the configured log level on first initialization
+    pluginLogger.info(`Logger initialized with level: ${logLevel}`)
+  }
+
+  // Fallback to console if logger not available (shouldn't happen in normal operation)
+  if (!pluginLogger) {
+    return {
+      debug: (...args: any[]) => console.log('[MAILING DEBUG]', ...args),
+      info: (...args: any[]) => console.log('[MAILING INFO]', ...args),
+      warn: (...args: any[]) => console.warn('[MAILING WARN]', ...args),
+      error: (...args: any[]) => console.error('[MAILING ERROR]', ...args),
+    }
+  }
+
+  return pluginLogger
+}
+
+/**
+ * Create a context-specific logger for a particular operation
+ */
+export function createContextLogger(payload: Payload, context: string) {
+  const logger = getPluginLogger(payload)
+
+  return {
+    debug: (message: string, ...args: any[]) => logger.debug(`[${context}] ${message}`, ...args),
+    info: (message: string, ...args: any[]) => logger.info(`[${context}] ${message}`, ...args),
+    warn: (message: string, ...args: any[]) => logger.warn(`[${context}] ${message}`, ...args),
+    error: (message: string, ...args: any[]) => logger.error(`[${context}] ${message}`, ...args),
+  }
+}