Notification Service

notifications/
  tasks.ts     # one task per channel + a periodic digest
  service.ts   # the producer API the rest of the app calls
  worker.ts    # the worker process

import { Queue } from "@byteveda/flexiq";

export const queue = new Queue({ dbPath: "notifications.db" });

queue.task("sendEmail", async (to: string, subject: string, body: string) => {
  await emailProvider.send(to, subject, body);
}, { maxRetries: 5, retryBackoff: { baseMs: 2_000, maxMs: 300_000 } });

queue.task("sendSms", async (to: string, text: string) => {
  await smsProvider.send(to, text);
}, { maxRetries: 3, rateLimit: "100/m" });

queue.task("sendDigest", async () => {
  const due = await pendingDigests();
  queue.enqueueMany("sendEmail", due.map((d) => ({
    args: [d.email, "Your daily digest", d.html],
  })));
});

import { queue } from "./tasks";

// 1. Delayed scheduling — remind 24h from now.
export function scheduleReminder(email: string) {
  return queue.enqueue("sendEmail", [email, "Reminder", "..."], {
    delayMs: 24 * 60 * 60 * 1000,
  });
}

// 2. Idempotency — a duplicate enqueue with the same key is a no-op while the
//    first is still pending or running.
export function sendWelcomeOnce(userId: string, email: string) {
  return queue.enqueue("sendEmail", [email, "Welcome", "..."], {
    uniqueKey: `welcome:${userId}`,
  });
}

// 3. Priority — security alerts preempt routine mail.
export function sendSecurityAlert(phone: string, text: string) {
  return queue.enqueue("sendSms", [phone, text], { priority: 100 });
}

// 4. Cancellation — pull a not-yet-started send.
export function cancelScheduled(jobId: string) {
  return queue.cancelJob(jobId); // false if it already started
}

// 5. Inspection — what's still pending?
export async function pendingSends() {
  return queue.listJobs({ status: "pending", task: "sendEmail", limit: 50 });
}

import { queue } from "./tasks";

queue.registerPeriodic("daily-digest", "sendDigest", "0 8 * * *", {
  timezone: "America/New_York",
});

const worker = queue.runWorker({ queues: ["default"] });
process.on("SIGINT", () => {
  worker.stop();
  process.exit(0);
});