feat: Implement Phase 2 cross-cutting services with email and notification enhancements

- Refactor email verification and password reset utilities to use new email service.
- Introduce email delivery queue and notification delivery model for better tracking.
- Enhance file validation and storage services for improved security and ownership management.
- Add cron job for cleaning inactive notifications with retention policy.
- Update document worker to handle document generation and storage more efficiently.
- Implement logging improvements in activity and log workers.
- Create comprehensive documentation for new API endpoints and services.
- Add unit tests for file validation and notification policies to ensure robustness.
This commit is contained in:
Sathira Sri Sathara
2026-09-03 14:22:34 +05:30
parent 9d3d431416
commit b6b345f245
54 changed files with 593 additions and 1248 deletions
+17
View File
@@ -0,0 +1,17 @@
const { Worker, UnrecoverableError } = require("bullmq");
const connection = require("../config/redisClient");
const db = require("../models");
const { sendMail } = require("../utils/mail.util");
const { templates } = require("../services/email/email.service");
const permanent = (error) => ["EENVELOPE", "EMESSAGE"].includes(error.code) || /^5\d\d$/.test(String(error.responseCode || ""));
module.exports = () => {
const worker = new Worker("email-delivery", async (job) => {
const { deliveryId, templateKey, recipient, variables } = job.data;
const delivery = await db.NotificationDelivery.findByPk(deliveryId);
await delivery?.update({ status: "PROCESSING", attemptCount: job.attemptsMade + 1, lastAttemptAt: new Date() });
try { const result = await sendMail({ to: recipient, subject: templates[templateKey].subject, templateName: templateKey, templateVars: variables }); await delivery?.update({ status: "SENT", sentAt: new Date(), failureCode: null }); return { messageId: result.messageId }; }
catch (error) { if (permanent(error)) throw new UnrecoverableError(error.code || "PERMANENT_REJECTION"); throw error; }
}, { connection, concurrency: Number(process.env.EMAIL_QUEUE_CONCURRENCY || 5) });
worker.on("failed", async (job, error) => { if (job && (job.attemptsMade >= (job.opts.attempts || 1) || error instanceof UnrecoverableError)) await db.NotificationDelivery.update({ status: "FAILED", failedAt: new Date(), failureCode: String(error.code || error.name || "DELIVERY_FAILED").slice(0, 100) }, { where: { id: job.data.deliveryId } }).catch(() => undefined); console.error("Email delivery failed", { jobId: job?.id, code: error.code || error.name }); });
return worker;
};