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.10",
+  "version": "0.4.13",
   "description": "Template-based email system with scheduling and job processing for PayloadCMS",
   "type": "module",
   "main": "dist/index.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,87 +0,0 @@
-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
- */
-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) {
-    const logger = createContextLogger(payload, 'SCHEDULER')
-    logger.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) {
-    const logger = createContextLogger(payload, 'SCHEDULER')
-    logger.error('Failed to schedule email processing job:', error)
-  }
-}

src/sendEmail.ts

@@ -139,7 +139,6 @@ export const sendEmail = async <TEmail extends BaseEmailDocument = BaseEmailDocu
   // 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')
@@ -153,7 +152,6 @@ export const sendEmail = async <TEmail extends BaseEmailDocument = BaseEmailDocu
     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
@@ -178,20 +176,19 @@ export const sendEmail = async <TEmail extends BaseEmailDocument = BaseEmailDocu
         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)
+        const firstJob = Array.isArray(emailWithJobs.jobs) ? emailWithJobs.jobs[0] : emailWithJobs.jobs
-          ? String(emailWithJobs.jobs[0])
+        jobId = typeof firstJob === 'string' ? firstJob : String(firstJob.id || firstJob)
-          : 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 (attempt >= 2) {
+          logger.debug(`Waiting for job creation for email ${email.id}, attempt ${attempt + 1}/${maxAttempts}`)
+        }
       }
     }
 
@@ -213,10 +210,9 @@ export const sendEmail = async <TEmail extends BaseEmailDocument = BaseEmailDocu
       )
     }
 
-    logger.info(`Starting job execution for job ${jobId}`)
     try {
       await processJobById(payload, jobId)
-      logger.info(`Successfully processed email ${email.id} immediately`)
+      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)}`)

src/services/MailingService.ts

@@ -366,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

@@ -37,16 +37,11 @@ export async function processEmailById(payload: Payload, emailId: string): Promi
  * @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: {
@@ -55,9 +50,8 @@ export async function processJobById(payload: Payload, jobId: string): Promise<v
         }
       }
     })
-
-    logger.info(`Job ${jobId} execution completed`, { result })
   } 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/jobScheduler.ts

@@ -49,15 +49,12 @@ export async function ensureEmailJob(
   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,
@@ -69,31 +66,22 @@ export async function ensureEmailJob(
     })
 
     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(', ')}`)
+      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
@@ -109,7 +97,7 @@ export async function ensureEmailJob(
 
     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}`)
+      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}`
@@ -117,7 +105,7 @@ export async function ensureEmailJob(
     }
 
     // Non-constraint related error
-    logger.error(`Non-constraint job creation error for email ${normalizedEmailId}: ${errorMessage}`)
+    logger.error(`Job creation error for email ${normalizedEmailId}: ${errorMessage}`)
     throw new Error(`Failed to create job for email ${normalizedEmailId}: ${errorMessage}`)
   }
 }