fix: deduplicate notification events

src/services/__tests__/notificationRouter.test.ts

@@ -1,4 +1,5 @@
 import { NotificationRouter, NotificationSeverity } from "../notificationRouter";
+import { notificationDeduplicator } from "../notificationDeduplicator";
 import { UserModel } from "../../models/users";
 import { Transaction } from "../../models/transaction";
 
@@ -60,6 +61,21 @@ const mockPushService = (global as any).mockPushService;
 const mockWhatsappService = (global as any).mockWhatsappService;
 const mockPagerDutyService = (global as any).mockPagerDutyService;
 
+function makeTransaction(overrides: Partial<Transaction> = {}): Transaction {
+  return {
+    id: "tx-1",
+    referenceNumber: "REF-1",
+    type: "deposit",
+    amount: "100",
+    phoneNumber: "+15551234567",
+    provider: "mtn",
+    status: "completed",
+    userId: "user-1",
+    createdAt: new Date(),
+    updatedAt: new Date(),
+    ...overrides,
+  };
+}
 
 describe("NotificationRouter", () => {
   let notificationRouter: NotificationRouter;
@@ -75,10 +91,12 @@ describe("NotificationRouter", () => {
     } as any;
 
     notificationRouter = new NotificationRouter(mockUserModel);
+    notificationDeduplicator.reset();
   });
 
   describe("routeNotification", () => {
     it("should route low severity notifications to push channel only", async () => {
+      const logSpy = jest.spyOn(console, "log").mockImplementation(() => {});
       const context = {
         severity: "low" as NotificationSeverity,
         category: "test",
@@ -88,10 +106,85 @@ describe("NotificationRouter", () => {
 
       await notificationRouter.routeNotification(context);
 
-      // Verify push service was called
-      expect(mockPushService.sendToUser).toHaveBeenCalled();
+      // Low-severity system notifications select push as the only channel.
+      expect(logSpy).toHaveBeenCalledWith(
+        expect.stringContaining(
+          "Routing low notification to channels: push",
+        ),
+      );
       expect(mockEmailService.sendEmail).not.toHaveBeenCalled();
       expect(mockSmsService.notifyTransactionEvent).not.toHaveBeenCalled();
+      logSpy.mockRestore();
     });
   });
-});
+
+  describe("deduplication", () => {
+    beforeEach(() => {
+      mockUserModel.findById.mockResolvedValue({
+        email: "user@example.com",
+        preferredLanguage: "en",
+        displayName: "Test User",
+      } as any);
+    });
+
+    it("sends a transaction notification only once for the same event", async () => {
+      const transaction = makeTransaction();
+
+      await notificationRouter.routeTransactionNotification(transaction, "completed");
+      await notificationRouter.routeTransactionNotification(transaction, "completed");
+
+      expect(mockEmailService.sendTransactionReceipt).toHaveBeenCalledTimes(1);
+      expect(mockSmsService.notifyTransactionEvent).toHaveBeenCalledTimes(1);
+      expect(mockPushService.sendTransactionComplete).toHaveBeenCalledTimes(1);
+    });
+
+    it("sends distinct transaction events even within the dedup window", async () => {
+      const transaction = makeTransaction();
+
+      await notificationRouter.routeTransactionNotification(transaction, "completed");
+      await notificationRouter.routeTransactionNotification(
+        makeTransaction({ id: "tx-2", referenceNumber: "REF-2" }),
+        "completed",
+      );
+
+      expect(mockEmailService.sendTransactionReceipt).toHaveBeenCalledTimes(2);
+      expect(mockSmsService.notifyTransactionEvent).toHaveBeenCalledTimes(2);
+      expect(mockPushService.sendTransactionComplete).toHaveBeenCalledTimes(2);
+    });
+
+    it("treats completed and failed as distinct events for the same transaction", async () => {
+      const transaction = makeTransaction();
+
+      await notificationRouter.routeTransactionNotification(transaction, "completed");
+      await notificationRouter.routeTransactionNotification(
+        transaction,
+        "failed",
+        "provider rejected",
+      );
+
+      expect(mockEmailService.sendTransactionReceipt).toHaveBeenCalledTimes(1);
+      expect(mockEmailService.sendTransactionFailure).toHaveBeenCalledTimes(1);
+      expect(mockPushService.sendTransactionComplete).toHaveBeenCalledTimes(1);
+      expect(mockPushService.sendTransactionFailed).toHaveBeenCalledTimes(1);
+    });
+
+    it("deduplicates system notifications that share the same entity", async () => {
+      const logSpy = jest.spyOn(console, "log").mockImplementation(() => {});
+      const context = {
+        severity: "low" as NotificationSeverity,
+        category: "subscription",
+        title: "Subscription Created",
+        message: "Subscription sub-1 created",
+        data: { subscriptionId: "sub-1" },
+      };
+
+      await notificationRouter.routeNotification(context);
+      await notificationRouter.routeNotification(context);
+
+      expect(logSpy).toHaveBeenCalledWith(
+        expect.stringContaining("Skipping duplicate notification"),
+      );
+      logSpy.mockRestore();
+    });
+  });
+});

src/services/notificationDeduplicator.ts

@@ -0,0 +1,117 @@
+import { createHash } from "crypto";
+import { redisClient } from "../config/redis";
+
+/**
+ * Notification deduplication
+ *
+ * The same logical notification event (e.g. a transaction reaching
+ * "completed") can be triggered through multiple paths — the queue worker
+ * routes directly, while the notification worker reacts to Redis pub/sub
+ * messages that `updateStatus()` publishes to both the broadcast and
+ * per-transaction channels, and more than one process may be running.
+ *
+ * This helper claims a per-event key atomically so only the first caller
+ * proceeds; everyone else within the dedup window is a duplicate.
+ *
+ * Failure policy: fail-open. If the dedup store is unavailable we fall back
+ * to an in-process store, and if that is unavailable we let the notification
+ * through — a duplicate is preferable to a missed notification.
+ */
+
+const DEFAULT_TTL_SECONDS = parseInt(
+  process.env.NOTIFICATION_DEDUP_TTL_SECONDS || "300",
+  10,
+);
+
+const DEDUP_KEY_PREFIX = "notif:dedup:";
+
+/**
+ * Builds a stable fingerprint for dedup keys, hashing the parts when no
+ * entity identifier is available.
+ */
+export function hashDedupParts(
+  parts: Array<string | undefined | null>,
+): string {
+  return createHash("sha256")
+    .update(parts.filter(Boolean).join("|"))
+    .digest("hex")
+    .slice(0, 24);
+}
+
+/** In-process SET NX equivalent — used when Redis is not connected. */
+class InMemoryDeduplicator {
+  private readonly store = new Map<string, number>(); // key -> expiry (ms)
+  private readonly ttlMs: number;
+
+  constructor(ttlMs: number) {
+    this.ttlMs = ttlMs;
+  }
+
+  tryAcquire(key: string): boolean {
+    const now = Date.now();
+
+    // Opportunistic cleanup to avoid unbounded growth.
+    if (this.store.size >= 10_000) {
+      for (const [storedKey, expiresAt] of this.store) {
+        if (expiresAt <= now) this.store.delete(storedKey);
+      }
+    }
+
+    const expiresAt = this.store.get(key);
+    if (expiresAt !== undefined && expiresAt > now) {
+      return false;
+    }
+
+    this.store.set(key, now + this.ttlMs);
+    return true;
+  }
+
+  reset(): void {
+    this.store.clear();
+  }
+}
+
+export class NotificationDeduplicator {
+  private readonly ttlSeconds: number;
+  private readonly memory: InMemoryDeduplicator;
+
+  constructor(ttlSeconds: number = DEFAULT_TTL_SECONDS) {
+    this.ttlSeconds = ttlSeconds;
+    this.memory = new InMemoryDeduplicator(ttlSeconds * 1000);
+  }
+
+  /**
+   * Atomically claims a notification event key.
+   *
+   * @returns `true` if this caller is the first to claim the event (send the
+   *          notification), `false` if the event was already claimed within
+   *          the dedup window (skip — it is a duplicate).
+   */
+  async claim(key: string): Promise<boolean> {
+    const redisKey = `${DEDUP_KEY_PREFIX}${key}`;
+
+    if (redisClient.isOpen) {
+      try {
+        const result = await redisClient.set(redisKey, "1", {
+          NX: true,
+          PX: this.ttlSeconds * 1000,
+        });
+        return result !== null;
+      } catch (err) {
+        console.warn(
+          "NotificationDeduplicator: Redis claim failed, using in-process dedup:",
+          err,
+        );
+      }
+    }
+
+    return this.memory.tryAcquire(key);
+  }
+
+  /** Clears the in-process store — primarily for tests. */
+  reset(): void {
+    this.memory.reset();
+  }
+}
+
+export const notificationDeduplicator = new NotificationDeduplicator();

src/services/notificationRouter.ts

@@ -3,6 +3,10 @@ import { SmsService, smsService } from "./sms";
 import { PushNotificationService, pushNotificationService } from "./push";
 import { WhatsappService, whatsappService } from "./whatsapp";
 import { PagerDutyService, pagerDutyService } from "./pagerDutyService";
+import {
+  hashDedupParts,
+  notificationDeduplicator,
+} from "./notificationDeduplicator";
 import { UserModel } from "../models/users";
 import { Transaction } from "../models/transaction";
 
@@ -20,6 +24,13 @@ export interface NotificationContext {
   message: string;
   data?: Record<string, any>;
   locale?: string;
+  /**
+   * Stable identity of the notification event used for deduplication.
+   * When omitted, one is derived from the context (category, entity id,
+   * or a hash of the content). Only the first send within the dedup window
+   * is delivered.
+   */
+  dedupKey?: string;
 }
 
 export interface DisputeNotificationContext {
@@ -329,6 +340,38 @@ export class NotificationRouter {
     console.log(`PagerDuty alert: ${context.title} - ${context.message}`);
   }
 
+  /**
+   * Derive a stable dedup key from the context when none was supplied.
+   * Prefers a real entity id (subscription, dispute, transaction, …) so
+   * distinct events for the same entity are not conflated; falls back to
+   * hashing the content so exact repeats are still suppressed.
+   */
+  private buildDedupKey(context: NotificationContext): string | null {
+    if (context.category === "transaction") {
+      return context.transactionId
+        ? `tx:${context.transactionId}:${context.severity}`
+        : null;
+    }
+
+    const entity =
+      context.data?.subscriptionId ??
+      context.data?.disputeId ??
+      context.data?.transactionId ??
+      context.data?.provider ??
+      context.data?.userId;
+
+    if (entity) {
+      return `sys:${context.category}:${context.severity}:${String(entity)}`;
+    }
+
+    return `sys:${context.category}:${context.severity}:${hashDedupParts([
+      context.category,
+      context.severity,
+      context.title,
+      context.message,
+    ])}`;
+  }
+
   /**
    * Route and send notification based on severity
    */
@@ -339,6 +382,18 @@ export class NotificationRouter {
       return;
     }
 
+    // Deduplicate: the same logical event can arrive through multiple paths
+    // (queue worker, Redis pub/sub broadcast + per-transaction channels,
+    // multiple worker instances). Only the first claim within the window is
+    // delivered.
+    const dedupKey = context.dedupKey ?? this.buildDedupKey(context);
+    if (dedupKey && !(await notificationDeduplicator.claim(dedupKey))) {
+      console.log(
+        `Skipping duplicate notification: ${dedupKey} (${context.category}/${context.severity})`,
+      );
+      return;
+    }
+
     const enabledChannels = await this.getEnabledChannels(context, rule);
 
     if (enabledChannels.length === 0) {
@@ -384,6 +439,7 @@ export class NotificationRouter {
       title,
       message,
       locale: "en", // Could be retrieved from user preferences
+      dedupKey: `tx:${transaction.id}:${status}`,
     });
   }
 
@@ -396,13 +452,15 @@ export class NotificationRouter {
     title: string,
     message: string,
     data?: Record<string, any>,
+    dedupKey?: string,
   ): Promise<void> {
     await this.routeNotification({
       severity,
       category,
       title,
       message,
       data,
+      dedupKey,
     });
   }
 
@@ -423,6 +481,7 @@ export class NotificationRouter {
         status: context.status,
         ...context.metadata,
       },
+      `dispute:${context.disputeId}:${context.event}`,
     );
   }
 }