278 lines
11 KiB
TypeScript
278 lines
11 KiB
TypeScript
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 DEFAULT_CLEANUP_INTERVAL_MS = 60 * 60 * 1000;
|
|
const CLEANUP_RETENTION_DAYS = 7;
|
|
const SUPPORTED_APPOINTMENT_NOTIFICATION_TYPES = new Set(["creation", "reminder", "cancellation"]);
|
|
|
|
function resolveCleanupIntervalMs(): number {
|
|
const raw = process.env.NOTIFICATION_JOB_CLEANUP_INTERVAL_MS;
|
|
if (!raw) return DEFAULT_CLEANUP_INTERVAL_MS;
|
|
|
|
const parsed = Number(raw);
|
|
return Number.isFinite(parsed) && parsed >= 0 ? parsed : DEFAULT_CLEANUP_INTERVAL_MS;
|
|
}
|
|
|
|
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);
|
|
}
|
|
|
|
export function buildStartOfDay(date: Date): Date {
|
|
return new Date(date.getFullYear(), date.getMonth(), date.getDate());
|
|
}
|
|
|
|
export function buildCleanupCutoff(now: Date): Date {
|
|
const cutoff = buildStartOfDay(now);
|
|
cutoff.setDate(cutoff.getDate() - CLEANUP_RETENTION_DAYS);
|
|
return cutoff;
|
|
}
|
|
|
|
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;
|
|
private cleanupTimeout?: NodeJS.Timeout;
|
|
private cleanupIntervalMs: number;
|
|
|
|
constructor() {
|
|
this.jobModel = new NotificationJobModel();
|
|
this.throttle = new WhatsAppThrottleClass();
|
|
this.cleanupIntervalMs = resolveCleanupIntervalMs();
|
|
}
|
|
|
|
public async startPolling(): Promise<void> {
|
|
if (this.polling) return;
|
|
this.polling = true;
|
|
logIntent("JobProcessor: starting poll loop");
|
|
await this.cleanupOldJobs();
|
|
this.scheduleCleanup();
|
|
this.poll();
|
|
}
|
|
|
|
public stopPolling(): void {
|
|
this.polling = false;
|
|
if (this.cleanupTimeout) {
|
|
clearTimeout(this.cleanupTimeout);
|
|
this.cleanupTimeout = undefined;
|
|
}
|
|
logIntent("JobProcessor: stopping poll loop");
|
|
}
|
|
|
|
private scheduleCleanup(): void {
|
|
if (!this.polling || this.cleanupIntervalMs <= 0) return;
|
|
|
|
this.cleanupTimeout = setTimeout(async () => {
|
|
await this.cleanupOldJobs();
|
|
this.scheduleCleanup();
|
|
}, this.cleanupIntervalMs);
|
|
}
|
|
|
|
private async cleanupOldJobs(now = new Date()): Promise<void> {
|
|
const cutoff = buildCleanupCutoff(now);
|
|
|
|
try {
|
|
const deletedCount = await this.jobModel.deleteBeforeScheduledAt(cutoff);
|
|
logDone(`JobProcessor: cleaned ${deletedCount} old notification jobs before ${cutoff.toISOString()}`);
|
|
} catch (error) {
|
|
logError(`JobProcessor: cleanup error before ${cutoff.toISOString()}`, error);
|
|
}
|
|
}
|
|
|
|
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);
|
|
}
|
|
}
|
|
}
|