xtr.dev/payload-mailing
1
0
Fork 0
mirror of https://github.com/xtr-dev/payload-mailing.git synced 2025-12-10 08:13:23 +00:00
Code Issues Packages Projects Releases Wiki Activity

Flatten email processing structure to individual jobs per email

Browse Source 
BREAKING CHANGE: Replaced batch email processing with individual jobs per email

Changes:
- Remove sendEmailTask.ts - no longer needed as each email gets its own job
- Add processEmailJob.ts - handles individual email processing
- Update sendEmail() to automatically create individual job per email
- Add processImmediately option to sendEmail() for instant processing
- Add processJobById() utility to run specific jobs immediately
- Update job registration to use new individual job structure
- Update dev API routes to use new processImmediately pattern
- Fix all TypeScript compilation errors

Benefits:
- Better job queue visibility (one job per email)
- More granular control over individual email processing
- Easier job monitoring and failure tracking
- Maintains backward compatibility via processImmediately option
- Simpler job queue management

Migration:
- Replace sendEmailJob usage with sendEmail({ processImmediately: true })
- Individual emails now appear as separate jobs in queue
- Batch processing still available via processEmailsTask if needed

🤖 Generated with [Claude Code](https://claude.ai/code)

Co-Authored-By: Claude <noreply@anthropic.com>
This commit is contained in:
Bas van den Aakster
2025-09-14 20:19:07 +02:00
parent 08f017abed
commit caa3686f1a
7 changed files with 164 additions and 290 deletions
Download Patch File Download Diff File Expand all files Collapse all files

6
src/index.ts
View File

@@ -11,9 +11,9 @@ export { MailingService } from './services/MailingService.js'
export { default as EmailTemplates, createEmailTemplatesCollection } from './collections/EmailTemplates.js'
export { default as Emails } from './collections/Emails.js'
// Jobs (includes the send email task)
export { mailingJobs, sendEmailJob } from './jobs/index.js'
export type { SendEmailTaskInput } from './jobs/sendEmailTask.js'
// Jobs (includes the individual email processing job)
export { mailingJobs } from './jobs/index.js'
export type { ProcessEmailJobInput } from './jobs/processEmailJob.js'
// Main email sending function
export { sendEmail, type SendEmailOptions } from './sendEmail.js'

10
src/jobs/index.ts
View File

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

78
src/jobs/processEmailJob.ts Normal file
View File

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

256
src/jobs/sendEmailTask.ts
View File

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

43
src/sendEmail.ts
View File

@@ -1,6 +1,7 @@
import { Payload } from 'payload'
import { getMailing, renderTemplate, parseAndValidateEmails } from './utils/helpers.js'
import { BaseEmailDocument } from './types/index.js'
import { processJobById } from './utils/emailProcessor.js'
// Options for sending emails
export interface SendEmailOptions<T extends BaseEmailDocument = BaseEmailDocument> {
@@ -13,6 +14,8 @@ export interface SendEmailOptions<T extends BaseEmailDocument = BaseEmailDocumen
data?: Partial<T>
// Common options
collectionSlug?: string // defaults to 'emails'
processImmediately?: boolean // if true, creates job and processes it immediately
queue?: string // queue name for the job, defaults to mailing config queue
}
/**
@@ -39,8 +42,8 @@ export const sendEmail = async <TEmail extends BaseEmailDocument = BaseEmailDocu
payload: Payload,
options: SendEmailOptions<TEmail>
): Promise<TEmail> => {
const mailing = getMailing(payload)
const collectionSlug = options.collectionSlug || mailing.collections.emails || 'emails'
const mailingConfig = getMailing(payload)
const collectionSlug = options.collectionSlug || mailingConfig.collections.emails || 'emails'
let emailData: Partial<TEmail> = { ...options.data } as Partial<TEmail>
@@ -139,6 +142,42 @@ 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
const queueName = options.queue || mailingConfig.queue || 'default'
let jobId: string | undefined
if (payload.jobs) {
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) {
console.warn(`Failed to create job for email ${email.id}:`, error)
// Don't fail the entire sendEmail operation if job creation fails
}
} else {
console.warn('PayloadCMS jobs not configured - emails will not be processed automatically')
}
// If processImmediately is true and we have a job, process it now
if (options.processImmediately && jobId) {
try {
await processJobById(payload, jobId)
} catch (error) {
console.warn(`Failed to process email ${email.id} immediately:`, error)
// Don't fail the entire sendEmail operation if immediate processing fails
}
}
return email as TEmail
}

25
src/utils/emailProcessor.ts
View File

@@ -29,6 +29,31 @@ 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)
await payload.jobs.run({
where: {
id: {
equals: jobId
}
}
})
} catch (error) {
throw new Error(`Failed to process job ${jobId}: ${error instanceof Error ? error.message : String(error)}`)
}
}
/**
* Processes all pending and failed emails using the mailing service
* @param payload Payload instance