Merge pull request #61 from xtr-dev/dev

Remove deprecated `processEmailsTask` and associated helpers

package.json

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

src/collections/Emails.ts

@@ -1,4 +1,6 @@
 import type { CollectionConfig } from 'payload'
+import { findExistingJobs, ensureEmailJob, updateEmailJobRelationship } from '../utils/jobScheduler.js'
+import { createContextLogger } from '../utils/logger.js'
 
 const Emails: CollectionConfig = {
   slug: 'emails',
@@ -183,21 +185,57 @@ const Emails: CollectionConfig = {
       },
     },
   ],
+  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: [
+  indexes: [
-  //   {
+    {
-  //     fields: {
+      fields: ['status', 'scheduledAt'],
-  //       status: 1,
+    },
-  //       scheduledAt: 1,
+    {
-  //     },
+      fields: ['priority', 'createdAt'],
-  //   },
+    },
-  //   {
+  ],
-  //     fields: {
-  //       priority: -1,
-  //       createdAt: 1,
-  //     },
-  //   },
-  // ],
 }
 
 export default Emails

src/index.ts

@@ -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,16 +1,11 @@
-import { processEmailsJob } from './processEmailsTask.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, // Kept for backward compatibility and batch processing if needed
+  processEmailJob,
-  processEmailJob,  // New individual email processing job
 ]
 
 // Re-export everything from individual job files
-export * from './processEmailsTask.js'
 export * from './processEmailJob.js'

src/jobs/processEmailJob.ts

@@ -13,7 +13,6 @@ export interface ProcessEmailJobInput {
 
 /**
  * Job definition for processing a single email
- * This replaces the batch processing approach with individual email jobs
  */
 export const processEmailJob = {
   slug: 'process-email',
@@ -69,4 +68,4 @@ export const processEmailJob = {
   }
 }
 
-export default processEmailJob
+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/sendEmail.ts

@@ -1,7 +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> {
@@ -104,15 +105,7 @@ export const sendEmail = async <TEmail extends BaseEmailDocument = BaseEmailDocu
   }
 
   // Sanitize fromName to prevent header injection
-  if (emailData.fromName) {
+  emailData.fromName = sanitizeFromName(emailData.fromName as string)
-    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) {
@@ -132,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
@@ -142,54 +136,85 @@ export const sendEmail = async <TEmail extends BaseEmailDocument = BaseEmailDocu
     throw new Error('Failed to create email: invalid response from database')
   }
 
-  // Create an individual job for this email
+  // If processImmediately is true, get the job from the relationship and process it now
-  const queueName = options.queue || mailingConfig.queue || 'default'
-
-  if (!payload.jobs) {
-    if (options.processImmediately) {
-      throw new Error('PayloadCMS jobs not configured - cannot process email immediately')
-    } else {
-      console.warn('PayloadCMS jobs not configured - emails will not be processed automatically')
-      return email as TEmail
-    }
-  }
-
-  let jobId: string
-  try {
-    const job = await payload.jobs.queue({
-      queue: queueName,
-      task: 'process-email',
-      input: {
-        emailId: String(email.id)
-      },
-      // If scheduled, set the waitUntil date
-      waitUntil: emailData.scheduledAt ? new Date(emailData.scheduledAt) : undefined
-    })
-
-    jobId = String(job.id)
-  } catch (error) {
-    // Clean up the orphaned email since job creation failed
-    try {
-      await payload.delete({
-        collection: collectionSlug,
-        id: email.id
-      })
-    } catch (deleteError) {
-      console.error(`Failed to clean up orphaned email ${email.id} after job creation failure:`, deleteError)
-    }
-
-    // Throw the original job creation error
-    const errorMsg = `Failed to create processing job for email ${email.id}: ${String(error)}`
-    throw new Error(errorMsg)
-  }
-
-  // If processImmediately is true, process the job now
   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 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
+
+
+    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,
+      })
+
+
+      if (emailWithJobs.jobs && emailWithJobs.jobs.length > 0) {
+        // Job found! Get the first job ID (should only be one for a new email)
+        const firstJob = Array.isArray(emailWithJobs.jobs) ? emailWithJobs.jobs[0] : emailWithJobs.jobs
+        jobId = typeof firstJob === 'string' ? firstJob : String(firstJob.id || firstJob)
+        break
+      }
+
+      // Log on later attempts to help with debugging (reduced threshold)
+      if (attempt >= 1) {
+        if (attempt >= 2) {
+          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}').`
+      )
+    }
+
     try {
       await processJobById(payload, jobId)
+      logger.debug(`Successfully processed email ${email.id} immediately`)
     } catch (error) {
-      // For immediate processing failures, we could consider cleanup, but the job exists and could be retried later
+      logger.error(`Failed to process email ${email.id} immediately:`, error)
-      // So we'll leave the email and job in place for potential retry
       throw new Error(`Failed to process email ${email.id} immediately: ${String(error)}`)
     }
   }

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
+    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, '\\"')
   }
 
   /**
@@ -372,7 +366,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/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
@@ -42,7 +43,7 @@ export async function processJobById(payload: Payload, jobId: string): Promise<v
 
   try {
     // Run a specific job by its ID (using where clause to find the job)
-    await payload.jobs.run({
+    const result = await payload.jobs.run({
       where: {
         id: {
           equals: jobId
@@ -50,6 +51,8 @@ export async function processJobById(payload: Payload, jobId: string): Promise<v
       }
     })
   } catch (error) {
+    const logger = createContextLogger(payload, 'PROCESSOR')
+    logger.error(`Job ${jobId} execution failed:`, error)
     throw new Error(`Failed to process job ${jobId}: ${String(error)}`)
   }
 }

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,148 @@
+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,
+    })
+
+    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),
+  }
+}