feat: BullMQ email:send Queue — enqueueEmailSend mit maxAttempts 3, exponentielles Backoff
Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
70
src/queues/email-send.queue.test.ts
Normal file
70
src/queues/email-send.queue.test.ts
Normal file
@@ -0,0 +1,70 @@
|
|||||||
|
import { describe, it, expect, vi, beforeEach } from 'vitest'
|
||||||
|
|
||||||
|
const mockAdd = vi.hoisted(() => vi.fn().mockResolvedValue({ id: 'job-1' }))
|
||||||
|
|
||||||
|
vi.mock('bullmq', () => ({
|
||||||
|
Queue: vi.fn().mockImplementation(() => ({
|
||||||
|
add: mockAdd,
|
||||||
|
})),
|
||||||
|
}))
|
||||||
|
|
||||||
|
import { enqueueEmailSend, type EmailSendJobData, resetQueueForTesting } from './email-send.queue'
|
||||||
|
import { Queue } from 'bullmq'
|
||||||
|
|
||||||
|
describe('enqueueEmailSend', () => {
|
||||||
|
beforeEach(() => {
|
||||||
|
vi.clearAllMocks()
|
||||||
|
resetQueueForTesting()
|
||||||
|
})
|
||||||
|
|
||||||
|
it('enqueued Job mit korrekten Daten', async () => {
|
||||||
|
const data: EmailSendJobData = {
|
||||||
|
tenantId: 'tenant1',
|
||||||
|
campaignId: 'campaign-uuid',
|
||||||
|
recipientEmail: 'empfaenger@example.com',
|
||||||
|
recipientHash: 'abc123hash',
|
||||||
|
}
|
||||||
|
const result = await enqueueEmailSend(data)
|
||||||
|
expect(result.ok).toBe(true)
|
||||||
|
expect(mockAdd).toHaveBeenCalledWith('send', data, expect.objectContaining({ attempts: 3 }))
|
||||||
|
})
|
||||||
|
|
||||||
|
it('gibt Job-ID zurück', async () => {
|
||||||
|
const data: EmailSendJobData = {
|
||||||
|
tenantId: 'tenant1',
|
||||||
|
campaignId: 'c1',
|
||||||
|
recipientEmail: 'x@example.com',
|
||||||
|
recipientHash: 'hash1',
|
||||||
|
}
|
||||||
|
const result = await enqueueEmailSend(data)
|
||||||
|
if (result.ok) expect(result.data).toBe('job-1')
|
||||||
|
})
|
||||||
|
|
||||||
|
it('gibt err zurück wenn Queue.add wirft', async () => {
|
||||||
|
mockAdd.mockRejectedValueOnce(new Error('Redis down'))
|
||||||
|
const result = await enqueueEmailSend({
|
||||||
|
tenantId: 't1', campaignId: 'c1', recipientEmail: 'x@x.com', recipientHash: 'h1',
|
||||||
|
})
|
||||||
|
expect(result.ok).toBe(false)
|
||||||
|
if (!result.ok) expect(result.error.message).toBe('Redis down')
|
||||||
|
})
|
||||||
|
|
||||||
|
it('nutzt exponentielles Backoff', async () => {
|
||||||
|
await enqueueEmailSend({
|
||||||
|
tenantId: 't1', campaignId: 'c1', recipientEmail: 'x@x.com', recipientHash: 'h1',
|
||||||
|
})
|
||||||
|
expect(mockAdd).toHaveBeenCalledWith(
|
||||||
|
'send',
|
||||||
|
expect.anything(),
|
||||||
|
expect.objectContaining({
|
||||||
|
backoff: expect.objectContaining({ type: 'exponential' }),
|
||||||
|
})
|
||||||
|
)
|
||||||
|
})
|
||||||
|
|
||||||
|
it('Queue wird mit Queue-Name "email:send" initialisiert', async () => {
|
||||||
|
await enqueueEmailSend({ tenantId: 't1', campaignId: 'c1', recipientEmail: 'x@x.com', recipientHash: 'h1' })
|
||||||
|
const { Queue: MockQueue } = await import('bullmq')
|
||||||
|
expect(MockQueue).toHaveBeenCalledWith('email:send', expect.anything())
|
||||||
|
})
|
||||||
|
})
|
||||||
42
src/queues/email-send.queue.ts
Normal file
42
src/queues/email-send.queue.ts
Normal file
@@ -0,0 +1,42 @@
|
|||||||
|
import { Queue } from 'bullmq'
|
||||||
|
import { ok, err, type Result } from '../lib/result'
|
||||||
|
|
||||||
|
export interface EmailSendJobData {
|
||||||
|
tenantId: string
|
||||||
|
campaignId: string
|
||||||
|
recipientEmail: string
|
||||||
|
recipientHash: string
|
||||||
|
}
|
||||||
|
|
||||||
|
const connection = {
|
||||||
|
host: process.env.REDIS_HOST ?? 'localhost',
|
||||||
|
port: Number(process.env.REDIS_PORT ?? 6379),
|
||||||
|
}
|
||||||
|
|
||||||
|
let _queue: Queue<EmailSendJobData> | null = null
|
||||||
|
|
||||||
|
function getQueue(): Queue<EmailSendJobData> {
|
||||||
|
if (!_queue) {
|
||||||
|
_queue = new Queue<EmailSendJobData>('email:send', { connection })
|
||||||
|
}
|
||||||
|
return _queue
|
||||||
|
}
|
||||||
|
|
||||||
|
/** Nur für Tests — setzt den Queue-Singleton zurück */
|
||||||
|
export function resetQueueForTesting(): void {
|
||||||
|
_queue = null
|
||||||
|
}
|
||||||
|
|
||||||
|
export async function enqueueEmailSend(data: EmailSendJobData): Promise<Result<string>> {
|
||||||
|
try {
|
||||||
|
const job = await getQueue().add('send', data, {
|
||||||
|
attempts: 3,
|
||||||
|
backoff: { type: 'exponential', delay: 2000 },
|
||||||
|
removeOnComplete: 100,
|
||||||
|
removeOnFail: { count: 500 },
|
||||||
|
})
|
||||||
|
return ok(job.id ?? 'unknown')
|
||||||
|
} catch (e) {
|
||||||
|
return err(e instanceof Error ? e : new Error(String(e)))
|
||||||
|
}
|
||||||
|
}
|
||||||
Reference in New Issue
Block a user