Merge pull request #69 from xtr-dev/dev

Bump version to 0.4.21

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)
     }
 
-    const result = await sendEmail(payload, emailOptions)
+    // Set processImmediately for "send now" type
+    const processImmediately = (type === 'send' && !scheduledAt)
+    emailOptions.processImmediately = processImmediately
 
-    // If it's "send now" (not scheduled), process the email immediately
+    const result = await sendEmail(payload, emailOptions)
-    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',
+      message: processImmediately ? 'Email sent successfully' :
-      status: scheduledAt ? 'scheduled' : 'queued'
+               scheduledAt ? 'Email scheduled successfully' :
+               'Email queued successfully',
+      status: processImmediately ? 'sent' :
+              scheduledAt ? 'scheduled' :
+              'queued'
     })
   } catch (error) {
     console.error('Test email error:', error)

dev/payload.config.ts

@@ -123,123 +123,6 @@ export default buildConfig({
         retryDelay: 60000, // 1 minute for dev
         queue: 'default',
 
-        // Example: Collection overrides for customization
-        // Uncomment and modify as needed for your use case
-        /*
-        collections: {
-          templates: {
-            // Custom access controls - restrict who can manage templates
-            access: {
-              read: ({ req: { user } }) => {
-                if (!user) return false
-                return user.role === 'admin' || user.permissions?.includes('mailing:read')
-              },
-              create: ({ req: { user } }) => {
-                if (!user) return false
-                return user.role === 'admin' || user.permissions?.includes('mailing:create')
-              },
-              update: ({ req: { user } }) => {
-                if (!user) return false
-                return user.role === 'admin' || user.permissions?.includes('mailing:update')
-              },
-              delete: ({ req: { user } }) => {
-                if (!user) return false
-                return user.role === 'admin'
-              },
-            },
-            // Custom admin UI settings
-            admin: {
-              group: 'Marketing',
-              description: 'Email templates with enhanced security and categorization'
-            },
-            // Add custom fields to templates
-            fields: [
-              // Default plugin fields are automatically included
-              {
-                name: 'category',
-                type: 'select',
-                options: [
-                  { label: 'Marketing', value: 'marketing' },
-                  { label: 'Transactional', value: 'transactional' },
-                  { label: 'System Notifications', value: 'system' }
-                ],
-                defaultValue: 'transactional',
-                admin: {
-                  position: 'sidebar',
-                  description: 'Template category for organization'
-                }
-              },
-              {
-                name: 'tags',
-                type: 'text',
-                hasMany: true,
-                admin: {
-                  position: 'sidebar',
-                  description: 'Tags for easy template filtering'
-                }
-              },
-              {
-                name: 'isActive',
-                type: 'checkbox',
-                defaultValue: true,
-                admin: {
-                  position: 'sidebar',
-                  description: 'Only active templates can be used'
-                }
-              }
-            ],
-            // Custom validation hooks
-            hooks: {
-              beforeChange: [
-                ({ data, req }) => {
-                  // Example: Only admins can create system templates
-                  if (data.category === 'system' && req.user?.role !== 'admin') {
-                    throw new Error('Only administrators can create system notification templates')
-                  }
-
-                  // Example: Auto-generate slug if not provided
-                  if (!data.slug && data.name) {
-                    data.slug = data.name.toLowerCase()
-                      .replace(/[^a-z0-9]/g, '-')
-                      .replace(/-+/g, '-')
-                      .replace(/^-|-$/g, '')
-                  }
-
-                  return data
-                }
-              ]
-            }
-          },
-          emails: {
-            // Restrict access to emails collection
-            access: {
-              read: ({ req: { user } }) => {
-                if (!user) return false
-                return user.role === 'admin' || user.permissions?.includes('mailing:read')
-              },
-              create: ({ req: { user } }) => {
-                if (!user) return false
-                return user.role === 'admin' || user.permissions?.includes('mailing:create')
-              },
-              update: ({ req: { user } }) => {
-                if (!user) return false
-                return user.role === 'admin' || user.permissions?.includes('mailing:update')
-              },
-              delete: ({ req: { user } }) => {
-                if (!user) return false
-                return user.role === 'admin'
-              },
-            },
-            // Custom admin configuration for emails
-            admin: {
-              group: 'Marketing',
-              description: 'Email delivery tracking and management',
-              defaultColumns: ['subject', 'to', 'status', 'priority', 'scheduledAt'],
-            }
-          }
-        },
-        */
-
         // Optional: Custom rich text editor configuration
         // Comment out to use default lexical editor
         richTextEditor: lexicalEditor({
@@ -256,12 +139,6 @@ export default buildConfig({
             // etc.
           ],
         }),
-
-
-        // Called after mailing plugin is fully initialized
-        onReady: async (payload) => {
-          await seedUser(payload)
-        },
       }),
     ],
     secret: process.env.PAYLOAD_SECRET || 'test-secret_key',

package.json

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

src/collections/Emails.ts

@@ -1,13 +1,36 @@
 import type { CollectionConfig } from 'payload'
+import { findExistingJobs, ensureEmailJob, updateEmailJobRelationship } from '../utils/jobScheduler.js'
+import { createContextLogger } from '../utils/logger.js'
+import { resolveID } from '../utils/helpers.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',
   },
+  defaultPopulate: {
+    templateSlug: true,
+    to: true,
+    cc: true,
+    bcc: true,
+    from: true,
+    replyTo: true,
+    jobs: true,
+    status: true,
+    attempts: true,
+    lastAttemptAt: true,
+    error: true,
+    priority: true,
+    scheduledAt: true,
+    sentAt: true,
+    variables: true,
+    html: true,
+    text: true,
+    createdAt: true,
+  },
   fields: [
     {
       name: 'template',
@@ -17,6 +40,14 @@ const Emails: CollectionConfig = {
         description: 'Email template used (optional if custom content provided)',
       },
     },
+    {
+      name: 'templateSlug',
+      type: 'text',
+      admin: {
+        description: 'Slug of the email template (auto-populated from template relationship)',
+        readOnly: true,
+      },
+    },
     {
       name: 'to',
       type: 'text',
@@ -164,22 +195,98 @@ 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 }) => {
+        const emailId = resolveID(id)
+        return {
+          'input.emailId': {
+            equals: emailId ? String(emailId) : '',
+          },
+        }
+      },
+    },
   ],
+  hooks: {
+    beforeChange: [
+      async ({ data, req }) => {
+        // Auto-populate templateSlug from template relationship
+        if (data.template) {
+          try {
+            const template = await req.payload.findByID({
+              collection: 'email-templates',
+              id: typeof data.template === 'string' ? data.template : data.template.id,
+            })
+            data.templateSlug = template.slug
+          } catch (error) {
+            // If template lookup fails, clear the slug
+            data.templateSlug = undefined
+          }
+        } else {
+          // Clear templateSlug if template is removed
+          data.templateSlug = undefined
+        }
+        return data
+      }
+    ],
+    // 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: [
+  indexes: [
-  //   {
+    {
-  //     fields: {
+      fields: ['status', 'scheduledAt'],
-  //       status: 1,
+    },
-  //       scheduledAt: 1,
+    {
-  //     },
+      fields: ['priority', 'createdAt'],
-  //   },
+    },
-  //   {
+  ],
-  //     fields: {
-  //       priority: -1,
-  //       createdAt: 1,
-  //     },
-  //   },
-  // ],
 }
 
 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)
+// Jobs (includes the individual email processing job)
-export { mailingJobs, sendEmailJob } from './jobs/index.js'
+export { mailingJobs } from './jobs/index.js'
-export type { SendEmailTaskInput } from './jobs/sendEmailTask.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,11 @@
-import { processEmailsJob } from './processEmailsTask.js'
+import { processEmailJob } from './processEmailJob.js'
-import { sendEmailJob } from './sendEmailTask.js'
 
 /**
  * All mailing-related jobs that get registered with Payload
  */
 export const mailingJobs = [
-  processEmailsJob,
+  processEmailJob,
-  sendEmailJob,
 ]
 
 // Re-export everything from individual job files
-export * from './processEmailsTask.js'
+export * from './processEmailJob.js'
-export * from './sendEmailTask.js'

src/jobs/processEmailJob.ts

@@ -0,0 +1,71 @@
+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
+ */
+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,84 +0,0 @@
-import type { PayloadRequest, Payload } from 'payload'
-import { processAllEmails } from '../utils/emailProcessor.js'
-
-/**
- * Data passed to the process emails task
- */
-export interface ProcessEmailsTaskData {
-  // Currently no data needed - always processes both pending and failed emails
-}
-
-/**
- * Handler function for processing emails
- * Used internally by the task definition
- */
-export const processEmailsTaskHandler = async (
-  job: { data: ProcessEmailsTaskData },
-  context: { req: PayloadRequest }
-) => {
-  const { req } = context
-  const payload = (req as any).payload
-
-  // Use the shared email processing logic
-  await processAllEmails(payload)
-}
-
-/**
- * Task definition for processing emails
- * This is what gets registered with Payload's job system
- */
-export const processEmailsTask = {
-  slug: 'process-emails',
-  handler: async ({ job, req }: { job: any; req: any }) => {
-    // Get mailing context from payload
-    const payload = (req as any).payload
-    const mailingContext = payload.mailing
-
-    if (!mailingContext) {
-      throw new Error('Mailing plugin not properly initialized')
-    }
-
-    // Use the task handler
-    await processEmailsTaskHandler(
-      job as { data: ProcessEmailsTaskData },
-      { req }
-    )
-
-    return {
-      output: {
-        success: true,
-        message: 'Email queue processing completed successfully'
-      }
-    }
-  },
-  interfaceName: 'ProcessEmailsTask',
-}
-
-// For backward compatibility, export as processEmailsJob
-export const processEmailsJob = processEmailsTask
-
-/**
- * Helper function to schedule an email processing job
- * Used by the plugin during initialization and can be used by developers
- */
-export const scheduleEmailsJob = async (
-  payload: Payload,
-  queueName: string,
-  delay?: number
-) => {
-  if (!payload.jobs) {
-    console.warn('PayloadCMS jobs not configured - emails will not be processed automatically')
-    return
-  }
-
-  try {
-    await payload.jobs.queue({
-      queue: queueName,
-      task: 'process-emails',
-      input: {},
-      waitUntil: delay ? new Date(Date.now() + delay) : undefined,
-    } as any)
-  } catch (error) {
-    console.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,9 @@
 import { Payload } from 'payload'
-import { getMailing, renderTemplate, parseAndValidateEmails } from './utils/helpers.js'
+import { getMailing, renderTemplateWithId, parseAndValidateEmails, sanitizeFromName } from './utils/helpers.js'
 import { BaseEmailDocument } from './types/index.js'
+import { processJobById } from './utils/emailProcessor.js'
+import { createContextLogger } from './utils/logger.js'
+import { pollForJobId } from './utils/jobPolling.js'
 
 // Options for sending emails
 export interface SendEmailOptions<T extends BaseEmailDocument = BaseEmailDocument> {
@@ -13,6 +16,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,22 +44,22 @@ export const sendEmail = async <TEmail extends BaseEmailDocument = BaseEmailDocu
   payload: Payload,
   options: SendEmailOptions<TEmail>
 ): Promise<TEmail> => {
-  const mailing = getMailing(payload)
+  const mailingConfig = getMailing(payload)
-  const collectionSlug = options.collectionSlug || mailing.collections.emails || 'emails'
+  const collectionSlug = options.collectionSlug || mailingConfig.collections.emails || 'emails'
 
   let emailData: Partial<TEmail> = { ...options.data } as Partial<TEmail>
 
-  // If using a template, render it first
   if (options.template) {
-    const { html, text, subject } = await renderTemplate(
+    // Look up and render the template in a single operation to avoid duplicate lookups
+    const { html, text, subject, templateId } = await renderTemplateWithId(
       payload,
       options.template.slug,
       options.template.variables || {}
     )
 
-    // Template values take precedence over data values
     emailData = {
       ...emailData,
+      template: templateId,
       subject,
       html,
       text,
@@ -66,20 +71,16 @@ export const sendEmail = async <TEmail extends BaseEmailDocument = BaseEmailDocu
     throw new Error('Field "to" is required for sending emails')
   }
 
-  // Validate required fields based on whether template was used
   if (options.template) {
-    // When using template, subject and html should have been set by renderTemplate
     if (!emailData.subject || !emailData.html) {
       throw new Error(`Template rendering failed: template "${options.template.slug}" did not provide required subject and html content`)
     }
   } else {
-    // When not using template, user must provide subject and html directly
     if (!emailData.subject || !emailData.html) {
       throw new Error('Fields "subject" and "html" are required when sending direct emails without a template')
     }
   }
 
-  // Process email addresses using shared validation (handle null values)
   if (emailData.to) {
     emailData.to = parseAndValidateEmails(emailData.to as string | string[])
   }
@@ -91,27 +92,15 @@ export const sendEmail = async <TEmail extends BaseEmailDocument = BaseEmailDocu
   }
   if (emailData.replyTo) {
     const validated = parseAndValidateEmails(emailData.replyTo as string | string[])
-    // replyTo should be a single email, so take the first one if array
     emailData.replyTo = validated && validated.length > 0 ? validated[0] : undefined
   }
   if (emailData.from) {
     const validated = parseAndValidateEmails(emailData.from as string | string[])
-    // from should be a single email, so take the first one if array
     emailData.from = validated && validated.length > 0 ? validated[0] : undefined
   }
 
-  // Sanitize fromName to prevent header injection
+  emailData.fromName = sanitizeFromName(emailData.fromName as string)
-  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
-  }
 
-  // Normalize Date objects to ISO strings for consistent database storage
   if (emailData.scheduledAt instanceof Date) {
     emailData.scheduledAt = emailData.scheduledAt.toISOString()
   }
@@ -128,17 +117,40 @@ export const sendEmail = async <TEmail extends BaseEmailDocument = BaseEmailDocu
     emailData.updatedAt = emailData.updatedAt.toISOString()
   }
 
-  // Create the email in the collection with proper typing
   const email = await payload.create({
     collection: collectionSlug,
     data: emailData
   })
 
-  // Validate that the created email has the expected structure
   if (!email || typeof email !== 'object' || !email.id) {
     throw new Error('Failed to create email: invalid response from database')
   }
 
+  if (options.processImmediately) {
+    const logger = createContextLogger(payload, 'IMMEDIATE')
+
+    if (!payload.jobs) {
+      throw new Error('PayloadCMS jobs not configured - cannot process email immediately')
+    }
+
+    // Poll for the job ID using configurable polling mechanism
+    const { jobId } = await pollForJobId({
+      payload,
+      collectionSlug,
+      emailId: email.id,
+      config: mailingConfig.jobPolling,
+      logger,
+    })
+
+    try {
+      await processJobById(payload, jobId)
+      logger.debug(`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

@@ -1,17 +1,17 @@
-import { Payload } from 'payload'
+import {CollectionSlug, EmailAdapter, Payload, SendEmailOptions} from 'payload'
 import { Liquid } from 'liquidjs'
 import {
   MailingPluginConfig,
   TemplateVariables,
   MailingService as IMailingService,
-  BaseEmail, BaseEmailTemplate, BaseEmailDocument, BaseEmailTemplateDocument
+  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
   private config: MailingPluginConfig
-  private emailAdapter: any
   private templatesCollection: string
   private emailsCollection: string
   private liquid: Liquid | null | false = null
@@ -30,31 +30,23 @@ export class MailingService implements IMailingService {
     if (!this.payload.email) {
       throw new Error('Payload email configuration is required. Please configure email in your Payload config.')
     }
-    this.emailAdapter = this.payload.email
   }
 
   private ensureInitialized(): void {
     if (!this.payload || !this.payload.db) {
       throw new Error('MailingService payload not properly initialized')
     }
-    if (!this.emailAdapter) {
+    if (!this.payload.email) {
       throw new Error('Email adapter not configured. Please ensure Payload has email configured.')
     }
   }
 
   /**
    * 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
+    return sanitizeDisplayName(name, true) // escapeQuotes = true for email headers
-      .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, '\\"')
   }
 
   /**
@@ -133,6 +125,17 @@ export class MailingService implements IMailingService {
       throw new Error(`Email template not found: ${templateSlug}`)
     }
 
+    return this.renderTemplateDocument(template, variables)
+  }
+
+  /**
+   * Render a template document (for when you already have the template loaded)
+   * This avoids duplicate template lookups
+   * @internal
+   */
+  async renderTemplateDocument(template: BaseEmailTemplateDocument, variables: TemplateVariables): Promise<{ html: string; text: string; subject: string }> {
+    this.ensureInitialized()
+
     const emailContent = await this.renderEmailTemplate(template, variables)
     const subject = await this.renderTemplateString(template.subject || '', variables)
 
@@ -148,7 +151,7 @@ export class MailingService implements IMailingService {
     const currentTime = new Date().toISOString()
 
     const { docs: pendingEmails } = await this.payload.find({
-      collection: this.emailsCollection as any,
+      collection: this.emailsCollection as CollectionSlug,
       where: {
         and: [
           {
@@ -188,7 +191,7 @@ export class MailingService implements IMailingService {
     const retryTime = new Date(Date.now() - retryDelay).toISOString()
 
     const { docs: failedEmails } = await this.payload.find({
-      collection: this.emailsCollection as any,
+      collection: this.emailsCollection as CollectionSlug,
       where: {
         and: [
           {
@@ -228,7 +231,7 @@ export class MailingService implements IMailingService {
   async processEmailItem(emailId: string): Promise<void> {
     try {
       await this.payload.update({
-        collection: this.emailsCollection as any,
+        collection: this.emailsCollection as CollectionSlug,
         id: emailId,
         data: {
           status: 'processing',
@@ -237,8 +240,9 @@ export class MailingService implements IMailingService {
       })
 
       const email = await this.payload.findByID({
-        collection: this.emailsCollection as any,
+        collection: this.emailsCollection as CollectionSlug,
         id: emailId,
+        depth: 1,
       }) as BaseEmailDocument
 
       // Combine from and fromName for nodemailer using proper sanitization
@@ -249,7 +253,7 @@ export class MailingService implements IMailingService {
         fromField = this.getDefaultFrom()
       }
 
-      let mailOptions: any = {
+      let mailOptions: SendEmailOptions = {
         from: fromField,
         to: email.to,
         cc: email.cc || undefined,
@@ -260,6 +264,19 @@ export class MailingService implements IMailingService {
         text: email.text || undefined,
       }
 
+      if (!mailOptions.from) {
+        throw new Error('Email from field is required')
+      }
+      if (!mailOptions.to || (Array.isArray(mailOptions.to) && mailOptions.to.length === 0)) {
+        throw new Error('Email to field is required')
+      }
+      if (!mailOptions.subject) {
+        throw new Error('Email subject is required')
+      }
+      if (!mailOptions.html && !mailOptions.text) {
+        throw new Error('Email content is required')
+      }
+
       // Call beforeSend hook if configured
       if (this.config.beforeSend) {
         try {
@@ -285,10 +302,10 @@ export class MailingService implements IMailingService {
       }
 
       // Send email using Payload's email adapter
-      await this.emailAdapter.sendEmail(mailOptions)
+      await this.payload.email.sendEmail(mailOptions)
 
       await this.payload.update({
-        collection: this.emailsCollection as any,
+        collection: this.emailsCollection as CollectionSlug,
         id: emailId,
         data: {
           status: 'sent',
@@ -302,7 +319,7 @@ export class MailingService implements IMailingService {
       const maxAttempts = this.config.retryAttempts || 3
 
       await this.payload.update({
-        collection: this.emailsCollection as any,
+        collection: this.emailsCollection as CollectionSlug,
         id: emailId,
         data: {
           status: attempts >= maxAttempts ? 'failed' : 'pending',
@@ -319,14 +336,14 @@ export class MailingService implements IMailingService {
 
   private async incrementAttempts(emailId: string): Promise<number> {
     const email = await this.payload.findByID({
-      collection: this.emailsCollection as any,
+      collection: this.emailsCollection as CollectionSlug,
       id: emailId,
     })
 
     const newAttempts = ((email as any).attempts || 0) + 1
 
     await this.payload.update({
-      collection: this.emailsCollection as any,
+      collection: this.emailsCollection as CollectionSlug,
       id: emailId,
       data: {
         attempts: newAttempts,
@@ -372,7 +389,7 @@ export class MailingService implements IMailingService {
     if (engine === 'liquidjs') {
       try {
         await this.ensureLiquidJSInitialized()
-        if (this.liquid && typeof this.liquid !== 'boolean') {
+        if (this.liquid) {
           return await this.liquid.parseAndRender(template, variables)
         }
       } catch (error) {

src/types/index.ts

@@ -1,6 +1,12 @@
-import { Payload } from 'payload'
+import {Payload, SendEmailOptions} from 'payload'
 import type { CollectionConfig, RichTextField } from 'payload'
 
+// Payload ID type (string or number)
+export type PayloadID = string | number
+
+// Payload relation type - can be populated (object with id) or unpopulated (just the ID)
+export type PayloadRelation<T extends { id: PayloadID }> = T | PayloadID
+
 // JSON value type that matches Payload's JSON field type
 export type JSONValue = string | number | boolean | { [k: string]: unknown } | unknown[] | null | undefined
 
@@ -8,6 +14,7 @@ export type JSONValue = string | number | boolean | { [k: string]: unknown } | u
 export interface BaseEmailDocument {
   id: string | number
   template?: any
+  templateSlug?: string | null
   to: string[]
   cc?: string[] | null
   bcc?: string[] | null
@@ -39,28 +46,18 @@ export interface BaseEmailTemplateDocument {
   updatedAt?: string | Date | null
 }
 
-export type BaseEmail<TEmail extends BaseEmailDocument = BaseEmailDocument, TEmailTemplate extends BaseEmailTemplateDocument = BaseEmailTemplateDocument> = Omit<TEmail, 'id' | 'template'> & {template: Omit<TEmailTemplate, 'id'> | TEmailTemplate['id'] | undefined | null}
-
-export type BaseEmailTemplate<TEmailTemplate extends BaseEmailTemplateDocument = BaseEmailTemplateDocument> = Omit<TEmailTemplate, 'id'>
-
 export type TemplateRendererHook = (template: string, variables: Record<string, any>) => string | Promise<string>
 
 export type TemplateEngine = 'liquidjs' | 'mustache' | 'simple'
 
-export interface BeforeSendMailOptions {
+export type BeforeSendHook = (options: SendEmailOptions, email: BaseEmailDocument) => SendEmailOptions | Promise<SendEmailOptions>
-  from: string
-  to: string[]
-  cc?: string[]
-  bcc?: string[]
-  replyTo?: string
-  subject: string
-  html: string
-  text?: string
-  attachments?: any[]
-  [key: string]: any
-}
 
-export type BeforeSendHook = (options: BeforeSendMailOptions, email: BaseEmailDocument) => BeforeSendMailOptions | Promise<BeforeSendMailOptions>
+export interface JobPollingConfig {
+  maxAttempts?: number        // Maximum number of polling attempts (default: 5)
+  initialDelay?: number       // Initial delay in milliseconds (default: 25)
+  maxTotalTime?: number       // Maximum total polling time in milliseconds (default: 3000)
+  maxBackoffDelay?: number    // Maximum delay between attempts in milliseconds (default: 400)
+}
 
 export interface MailingPluginConfig {
   collections?: {
@@ -76,8 +73,8 @@ export interface MailingPluginConfig {
   templateEngine?: TemplateEngine
   richTextEditor?: RichTextField['editor']
   beforeSend?: BeforeSendHook
-  onReady?: (payload: any) => Promise<void>
   initOrder?: 'before' | 'after'
+  jobPolling?: JobPollingConfig
 }
 
 export interface QueuedEmail {

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,33 @@ 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> {
+  if (!payload.jobs) {
+    throw new Error('PayloadCMS jobs not configured - cannot process job immediately')
+  }
+
+  try {
+    // Run a specific job by its ID (using where clause to find the job)
+    const result = await payload.jobs.run({
+      where: {
+        id: {
+          equals: jobId
+        }
+      }
+    })
+  } catch (error) {
+    const logger = createContextLogger(payload, 'PROCESSOR')
+    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

@@ -1,5 +1,5 @@
 import { Payload } from 'payload'
-import { TemplateVariables } from '../types/index.js'
+import { TemplateVariables, PayloadID, PayloadRelation } from '../types/index.js'
 
 /**
  * Parse and validate email addresses
@@ -36,6 +36,87 @@ 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
+}
+
+/**
+ * Type guard to check if a Payload relation is populated (object) or unpopulated (ID)
+ */
+export const isPopulated = <T extends { id: PayloadID }>(
+  value: PayloadRelation<T> | null | undefined
+): value is T => {
+  return value !== null && value !== undefined && typeof value === 'object' && 'id' in value
+}
+
+/**
+ * Resolves a Payload relation to just the ID
+ * Handles both populated (object with id) and unpopulated (string/number) values
+ */
+export const resolveID = <T extends { id: PayloadID }>(
+  value: PayloadRelation<T> | null | undefined
+): PayloadID | undefined => {
+  if (value === null || value === undefined) return undefined
+
+  if (typeof value === 'string' || typeof value === 'number') {
+    return value
+  }
+
+  if (typeof value === 'object' && 'id' in value) {
+    return value.id
+  }
+
+  return undefined
+}
+
+/**
+ * Resolves an array of Payload relations to an array of IDs
+ * Handles mixed arrays of populated and unpopulated values
+ */
+export const resolveIDs = <T extends { id: PayloadID }>(
+  values: (PayloadRelation<T> | null | undefined)[] | null | undefined
+): PayloadID[] => {
+  if (!values || !Array.isArray(values)) return []
+
+  return values
+    .map(value => resolveID(value))
+    .filter((id): id is PayloadID => id !== undefined)
+}
+
 export const getMailing = (payload: Payload) => {
   const mailing = (payload as any).mailing
   if (!mailing) {
@@ -49,6 +130,53 @@ export const renderTemplate = async (payload: Payload, templateSlug: string, var
   return mailing.service.renderTemplate(templateSlug, variables)
 }
 
+/**
+ * Render a template and return both rendered content and template ID
+ * This is used by sendEmail to avoid duplicate template lookups
+ * @internal
+ */
+export const renderTemplateWithId = async (
+  payload: Payload,
+  templateSlug: string,
+  variables: TemplateVariables
+): Promise<{ html: string; text: string; subject: string; templateId: PayloadID }> => {
+  const mailing = getMailing(payload)
+  const templatesCollection = mailing.config.collections?.templates || 'email-templates'
+
+  // Runtime validation: Ensure the collection exists in Payload
+  if (!payload.collections[templatesCollection]) {
+    throw new Error(
+      `Templates collection '${templatesCollection}' not found. ` +
+      `Available collections: ${Object.keys(payload.collections).join(', ')}`
+    )
+  }
+
+  // Look up the template document once
+  const { docs: templateDocs } = await payload.find({
+    collection: templatesCollection as any,
+    where: {
+      slug: {
+        equals: templateSlug,
+      },
+    },
+    limit: 1,
+  })
+
+  if (!templateDocs || templateDocs.length === 0) {
+    throw new Error(`Template not found: ${templateSlug}`)
+  }
+
+  const templateDoc = templateDocs[0]
+
+  // Render using the document directly to avoid duplicate lookup
+  const rendered = await mailing.service.renderTemplateDocument(templateDoc, variables)
+
+  return {
+    ...rendered,
+    templateId: templateDoc.id,
+  }
+}
+
 export const processEmails = async (payload: Payload): Promise<void> => {
   const mailing = getMailing(payload)
   return mailing.service.processEmails()

src/utils/jobPolling.ts

@@ -0,0 +1,115 @@
+import { Payload } from 'payload'
+import { JobPollingConfig } from '../types/index.js'
+
+export interface PollForJobIdOptions {
+  payload: Payload
+  collectionSlug: string
+  emailId: string | number
+  config?: JobPollingConfig
+  logger?: {
+    debug: (message: string, ...args: any[]) => void
+    info: (message: string, ...args: any[]) => void
+    warn: (message: string, ...args: any[]) => void
+    error: (message: string, ...args: any[]) => void
+  }
+}
+
+export interface PollForJobIdResult {
+  jobId: string
+  attempts: number
+  elapsedTime: number
+}
+
+// Default job polling configuration values
+const DEFAULT_JOB_POLLING_CONFIG: Required<JobPollingConfig> = {
+  maxAttempts: 5,
+  initialDelay: 25,
+  maxTotalTime: 3000,
+  maxBackoffDelay: 400,
+}
+
+/**
+ * Polls for a job ID associated with an email document using exponential backoff.
+ * This utility handles the complexity of waiting for auto-scheduled jobs to be created.
+ *
+ * The polling mechanism uses exponential backoff with configurable parameters:
+ * - Starts with an initial delay and doubles on each retry
+ * - Caps individual delays at maxBackoffDelay
+ * - Enforces a maximum total polling time
+ *
+ * @param options - Polling options including payload, collection, email ID, and config
+ * @returns Promise resolving to job ID and timing information
+ * @throws Error if job is not found within the configured limits
+ */
+export const pollForJobId = async (options: PollForJobIdOptions): Promise<PollForJobIdResult> => {
+  const { payload, collectionSlug, emailId, logger } = options
+
+  // Merge user config with defaults
+  const config: Required<JobPollingConfig> = {
+    ...DEFAULT_JOB_POLLING_CONFIG,
+    ...options.config,
+  }
+
+  const { maxAttempts, initialDelay, maxTotalTime, maxBackoffDelay } = config
+  const startTime = Date.now()
+  let jobId: string | undefined
+
+  for (let attempt = 0; attempt < maxAttempts; attempt++) {
+    const elapsedTime = Date.now() - startTime
+
+    // Check if we've exceeded the maximum total polling time
+    if (elapsedTime > maxTotalTime) {
+      throw new Error(
+        `Job polling timed out after ${maxTotalTime}ms for email ${emailId}. ` +
+        `The auto-scheduling may have failed or is taking longer than expected.`
+      )
+    }
+
+    // Calculate exponential backoff delay, capped at maxBackoffDelay
+    const delay = Math.min(initialDelay * Math.pow(2, attempt), maxBackoffDelay)
+
+    // Wait before checking (skip on first attempt)
+    if (attempt > 0) {
+      await new Promise(resolve => setTimeout(resolve, delay))
+    }
+
+    // Fetch the email document to check for associated jobs
+    const emailWithJobs = await payload.findByID({
+      collection: collectionSlug,
+      id: emailId,
+    })
+
+    // Check if jobs array exists and has entries
+    if (emailWithJobs.jobs && emailWithJobs.jobs.length > 0) {
+      const firstJob = Array.isArray(emailWithJobs.jobs) ? emailWithJobs.jobs[0] : emailWithJobs.jobs
+      jobId = typeof firstJob === 'string' ? firstJob : String(firstJob.id || firstJob)
+
+      return {
+        jobId,
+        attempts: attempt + 1,
+        elapsedTime: Date.now() - startTime,
+      }
+    }
+
+    // Log progress for attempts after the second try
+    if (attempt >= 2 && logger) {
+      logger.debug(`Waiting for job creation for email ${emailId}, attempt ${attempt + 1}/${maxAttempts}`)
+    }
+  }
+
+  // If we reach here, job was not found
+  const elapsedTime = Date.now() - startTime
+  const timeoutMsg = elapsedTime >= maxTotalTime
+  const errorType = timeoutMsg ? 'POLLING_TIMEOUT' : 'JOB_NOT_FOUND'
+  const baseMessage = timeoutMsg
+    ? `Job polling timed out after ${maxTotalTime}ms for email ${emailId}`
+    : `No processing job found for email ${emailId} after ${maxAttempts} attempts (${elapsedTime}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('${emailId}').`
+  )
+}

src/utils/jobScheduler.ts

@@ -0,0 +1,152 @@
+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')
+
+  // 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 {
+    // 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}`)
+
+    return {
+      jobIds: [job.id],
+      created: true
+    }
+  } catch (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)
+
+
+    if (existingJobs.totalDocs > 0) {
+      // Found existing jobs - return them (race condition handled successfully)
+      logger.debug(`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.warn(`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(`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,
+    })
+
+    // Extract IDs from job objects or use the value directly if it's already an ID
+    // Jobs can be populated (objects with id field) or just IDs (strings/numbers)
+    const currentJobs = (currentEmail.jobs || []).map((job: any) =>
+      typeof job === 'object' && job !== null && job.id ? String(job.id) : 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),
+  }
+}