feat: implement comprehensive notification engine with policies, preferences, and automated dispatching services
This commit is contained in:
@@ -0,0 +1,228 @@
|
||||
import { logDone, logError, logIntent } from "../../config/logger.js";
|
||||
import { NotificationJobModel, INotificationJob } from "./NotificationJob.Adapter.Mongoose.js";
|
||||
import { WhatsAppThrottleClass } from "./WhatsAppThrottle.js";
|
||||
import { dispatchWhatsApp, dispatchEmail, dispatchSystem } from "./ChannelDispatchers.js";
|
||||
import { resolveNotificationContent } from "./NotificationContentResolver.js";
|
||||
|
||||
const POLL_INTERVAL_MS = 5000;
|
||||
const BASE_RETRY_DELAY_MS = 5000;
|
||||
const MAX_JITTER_MS = 3000;
|
||||
const THROTTLE_RETRY_DELAY_MS = 8000;
|
||||
const SUPPORTED_APPOINTMENT_NOTIFICATION_TYPES = new Set(["creation", "reminder", "cancellation"]);
|
||||
|
||||
function resolveAppointmentNotificationType(type: unknown): string {
|
||||
return typeof type === "string" && SUPPORTED_APPOINTMENT_NOTIFICATION_TYPES.has(type) ? type : "reminder";
|
||||
}
|
||||
|
||||
export function buildPendingJobQuery(now: Date) {
|
||||
return {
|
||||
status: "pending",
|
||||
scheduledAt: { $lte: now },
|
||||
$or: [{ nextRetryAt: { $exists: false } }, { nextRetryAt: null }, { nextRetryAt: { $lte: now } }],
|
||||
};
|
||||
}
|
||||
|
||||
export function calculateNextRetryAt(attempt: number, now: Date): Date {
|
||||
const delay = BASE_RETRY_DELAY_MS * Math.pow(2, attempt - 1);
|
||||
const jitter = Math.floor(Math.random() * MAX_JITTER_MS);
|
||||
return new Date(now.getTime() + delay + jitter);
|
||||
}
|
||||
|
||||
function getJobId(job: { id?: unknown; _id?: unknown; appointmentId?: unknown; channel?: unknown; get?: (path: string) => unknown }): string {
|
||||
const value = job.id || job._id || job.get?.("_id");
|
||||
if (value) return String(value);
|
||||
|
||||
const appointmentId = job.appointmentId || job.get?.("appointmentId");
|
||||
const channel = job.channel || job.get?.("channel");
|
||||
return appointmentId || channel ? `unknown appointment=${String(appointmentId || "n/a")} channel=${String(channel || "n/a")}` : "unknown";
|
||||
}
|
||||
|
||||
export class JobProcessor {
|
||||
private jobModel: NotificationJobModel;
|
||||
private throttle: WhatsAppThrottleClass;
|
||||
private polling = false;
|
||||
|
||||
constructor() {
|
||||
this.jobModel = new NotificationJobModel();
|
||||
this.throttle = new WhatsAppThrottleClass();
|
||||
}
|
||||
|
||||
public async startPolling(): Promise<void> {
|
||||
if (this.polling) return;
|
||||
this.polling = true;
|
||||
logIntent("JobProcessor: starting poll loop");
|
||||
this.poll();
|
||||
}
|
||||
|
||||
public stopPolling(): void {
|
||||
this.polling = false;
|
||||
logIntent("JobProcessor: stopping poll loop");
|
||||
}
|
||||
|
||||
private async poll(): Promise<void> {
|
||||
if (!this.polling) return;
|
||||
|
||||
try {
|
||||
const now = new Date();
|
||||
const pendingJobs = await this.jobModel.notificationJobList
|
||||
.find(buildPendingJobQuery(now))
|
||||
.sort({ scheduledAt: 1 })
|
||||
.exec();
|
||||
|
||||
if (pendingJobs.length > 0) {
|
||||
logIntent(`JobProcessor: found ${pendingJobs.length} pending jobs`);
|
||||
await this.processJobs(pendingJobs);
|
||||
}
|
||||
} catch (error) {
|
||||
logError("JobProcessor: poll error", error);
|
||||
}
|
||||
|
||||
setTimeout(() => this.poll(), POLL_INTERVAL_MS);
|
||||
}
|
||||
|
||||
private async processJobs(jobs: any[]): Promise<void> {
|
||||
// Group by companyId
|
||||
const grouped = new Map<string, any[]>();
|
||||
for (const job of jobs) {
|
||||
const companyId = String(job.companyId);
|
||||
if (!grouped.has(companyId)) {
|
||||
grouped.set(companyId, []);
|
||||
}
|
||||
grouped.get(companyId)!.push(job);
|
||||
}
|
||||
|
||||
for (const [companyId, companyJobs] of grouped) {
|
||||
for (const job of companyJobs) {
|
||||
await this.processJob(job, companyId);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
private async processJob(job: any, companyId: string): Promise<void> {
|
||||
const now = new Date();
|
||||
const jobId = getJobId(job);
|
||||
|
||||
try {
|
||||
// Check throttle for WhatsApp
|
||||
if (job.channel === "whatsapp") {
|
||||
if (!this.throttle.canSend(companyId)) {
|
||||
const nextRetryAt = new Date(now.getTime() + THROTTLE_RETRY_DELAY_MS);
|
||||
await this.jobModel.notificationJobList.updateOne(
|
||||
{ _id: job._id },
|
||||
{ $set: { status: "pending", nextRetryAt } }
|
||||
).exec();
|
||||
logIntent(`JobProcessor: job ${jobId} throttled WhatsApp for company ${companyId}; retry at ${nextRetryAt.toISOString()}`);
|
||||
return;
|
||||
}
|
||||
}
|
||||
|
||||
const nextAttempt = job.attempts + 1;
|
||||
await this.jobModel.notificationJobList.updateOne(
|
||||
{ _id: job._id },
|
||||
{ $set: { status: "processing", attempts: nextAttempt } }
|
||||
).exec();
|
||||
|
||||
const dispatchJob = {
|
||||
...job,
|
||||
_id: job._id || job.get?.("_id"),
|
||||
appointmentId: job.appointmentId || job.get?.("appointmentId"),
|
||||
channel: job.channel || job.get?.("channel"),
|
||||
attempts: nextAttempt,
|
||||
};
|
||||
const content = await resolveNotificationContent(job);
|
||||
const appointmentNotificationType = resolveAppointmentNotificationType(job.type);
|
||||
|
||||
let result;
|
||||
switch (job.channel) {
|
||||
case "whatsapp":
|
||||
result = await dispatchWhatsApp({
|
||||
companyId,
|
||||
appointmentId: String(job.appointmentId),
|
||||
type: appointmentNotificationType,
|
||||
systemToken: `${process.env.API_KEY}`,
|
||||
phoneNumber: content.phoneNumber || "",
|
||||
message: content.message,
|
||||
companyOwnerId: content.companyOwnerId || "",
|
||||
});
|
||||
break;
|
||||
case "email":
|
||||
result = await dispatchEmail({
|
||||
appointmentId: String(job.appointmentId),
|
||||
systemToken: `${process.env.API_KEY}`,
|
||||
type: appointmentNotificationType,
|
||||
email: content.email || "",
|
||||
subject: content.subject,
|
||||
message: content.message,
|
||||
});
|
||||
break;
|
||||
case "system":
|
||||
result = await dispatchSystem({
|
||||
systemToken: `${process.env.API_KEY}`,
|
||||
userId: content.userId || "",
|
||||
subject: content.subject,
|
||||
message: content.message,
|
||||
companyId: String(job.companyId),
|
||||
type: job.type,
|
||||
code: String(job.appointmentId),
|
||||
});
|
||||
break;
|
||||
default:
|
||||
result = { success: false, error: `Unknown channel: ${job.channel}` };
|
||||
}
|
||||
|
||||
if (result.success) {
|
||||
if (job.channel === "whatsapp") {
|
||||
this.throttle.recordSent(companyId);
|
||||
}
|
||||
|
||||
await this.jobModel.notificationJobList.updateOne(
|
||||
{ _id: job._id },
|
||||
{
|
||||
$set: {
|
||||
status: "sent",
|
||||
sentAt: new Date(),
|
||||
providerResponse: result.providerResponse,
|
||||
},
|
||||
}
|
||||
).exec();
|
||||
logDone(`JobProcessor: job ${jobId} sent via ${job.channel}`);
|
||||
} else {
|
||||
await this.handleFailure(dispatchJob, result.error || "Unknown error");
|
||||
}
|
||||
} catch (error) {
|
||||
const msg = error instanceof Error ? error.message : String(error);
|
||||
await this.handleFailure({ ...job, attempts: job.attempts + 1 }, msg);
|
||||
}
|
||||
}
|
||||
|
||||
private async handleFailure(job: any, errorMessage: string): Promise<void> {
|
||||
const now = new Date();
|
||||
const jobId = getJobId(job);
|
||||
|
||||
if (job.attempts >= job.maxAttempts) {
|
||||
await this.jobModel.notificationJobList.updateOne(
|
||||
{ _id: job._id },
|
||||
{
|
||||
$set: {
|
||||
status: "failed",
|
||||
lastError: errorMessage,
|
||||
},
|
||||
}
|
||||
).exec();
|
||||
logError(`JobProcessor: job ${jobId} failed permanently after ${job.attempts} attempts`, errorMessage);
|
||||
} else {
|
||||
const nextRetryAt = calculateNextRetryAt(job.attempts, now);
|
||||
await this.jobModel.notificationJobList.updateOne(
|
||||
{ _id: job._id },
|
||||
{
|
||||
$set: {
|
||||
status: "pending",
|
||||
lastError: errorMessage,
|
||||
nextRetryAt,
|
||||
},
|
||||
}
|
||||
).exec();
|
||||
logError(`JobProcessor: job ${jobId} failed attempt ${job.attempts}/${job.maxAttempts}; retry at ${nextRetryAt.toISOString()}`, errorMessage);
|
||||
}
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user