// src/services/email-worker.boot.ts
import { defineService, loadService } from '@hile/core'
import redisService from '@hile/ioredis'
import { RedisIdempotency, stableHash } from '@hile/redis-idempotency'
import { RedisStreamQueue, defineQueue } from '@hile/redis-stream-queue'
type EmailPayload = {
tenantId: string
userId: string
template: 'welcome'
}
const emailQueue = defineQueue<EmailPayload>('email')
export default defineService('email.worker', async (shutdown) => {
const redis = await loadService(redisService)
const queue = new RedisStreamQueue(redis, { prefix: 'app:' })
const idempotency = new RedisIdempotency(redis)
const worker = queue.worker(emailQueue, async (job) => {
await idempotency.run(
`idem:email:${job.data.tenantId}:${job.jobId ?? job.id}`,
() => sendEmail(job.data),
{
lockTtl: 60_000,
resultTtl: 86_400_000,
fingerprint: stableHash(job.data),
},
)
}, {
group: 'email-workers',
consumer: process.env.HOSTNAME ?? `${process.pid}`,
concurrency: 8,
claimIdle: 60_000,
})
worker.start()
shutdown(() => worker.stop())
return { queue, worker }
})