Feature/escrow event listener

apps/backend/notifications/README.md

@@ -9,3 +9,13 @@ pnpm --filter @delego/notifications dev
 ```
 
 Health check: `GET http://localhost:3015/health`
+
+## Environment Variables
+
+- `ESCROW_CONTRACT_ID`: The Soroban contract ID for the escrow contract, to listen for events.
+- `SOROBAN_RPC_URL`: The Stellar RPC URL to poll for on-chain events.
+- `REDIS_URL`: Redis connection URL, used for worker idempotency and deduplication.
+- `DATABASE_URL`: PostgreSQL connection URL, used for wallet lookup adapter.
+- `SENDGRID_API_KEY`: SendGrid API key for emails.
+- `FROM_EMAIL`: Sender email address.
+- `LOG_LEVEL`: Log level (e.g. info, debug).

apps/backend/notifications/src/escrowListener.test.ts

@@ -0,0 +1,109 @@
+import { describe, it, expect, vi, beforeEach, afterEach } from "vitest";
+import { startEscrowEventListener, stopEscrowEventListener } from "./escrowListener.js";
+import { rpc as SorobanRpc, xdr, nativeToScVal } from "@stellar/stellar-sdk";
+import { Redis } from "ioredis";
+import * as idempotency from "./idempotency.js";
+import { setWalletLookupAdapter, resetWalletLookupAdapter } from "./walletLookup.js";
+import * as email from "../email/index.js";
+import * as push from "../push/index.js";
+
+// Mock dependencies
+vi.mock("ioredis", () => {
+  const Redis = vi.fn();
+  Redis.prototype.get = vi.fn().mockResolvedValue(null);
+  Redis.prototype.set = vi.fn().mockResolvedValue("OK");
+  Redis.prototype.smembers = vi.fn().mockResolvedValue([]);
+  return { Redis };
+});
+
+vi.mock("./idempotency.js", () => ({
+  checkAndMarkDispatched: vi.fn().mockResolvedValue(true),
+}));
+
+vi.mock("../email/index.js", () => ({
+  sendEmail: vi.fn().mockResolvedValue(undefined),
+}));
+
+vi.mock("../push/index.js", () => ({
+  sendPushNotification: vi.fn().mockResolvedValue(undefined),
+}));
+
+describe("escrowListener", () => {
+  beforeEach(() => {
+    vi.useFakeTimers();
+    vi.clearAllMocks();
+
+    // Mock the Stellar SDK RPC Server
+    const mockGetLatestLedger = vi.fn().mockResolvedValue({ sequence: 100 });
+    const mockGetEvents = vi.fn().mockResolvedValue({ events: [] });
+
+    SorobanRpc.Server.prototype.getLatestLedger = mockGetLatestLedger;
+    SorobanRpc.Server.prototype.getEvents = mockGetEvents;
+
+    setWalletLookupAdapter({
+      lookupByWalletAddress: vi.fn().mockImplementation(async (address: string) => {
+        if (address === "merchant-addr") {
+          return {
+            walletAddress: "merchant-addr",
+            userId: "merchant-123",
+            email: "merchant@example.com",
+            pushEnabled: false,
+          };
+        }
+        return null;
+      }),
+    });
+  });
+
+  afterEach(() => {
+    stopEscrowEventListener();
+    resetWalletLookupAdapter();
+    vi.useRealTimers();
+  });
+
+  it("decodes on-chain EscrowCreatedEvent and dispatches notification", async () => {
+    const mockGetEvents = vi.fn().mockResolvedValue({
+      events: [
+        {
+          type: "contract",
+          inSuccessfulContractCall: true,
+          contractId: "C123",
+          ledger: 100,
+          txHash: "abcd",
+          id: "event-1",
+          topic: [
+            nativeToScVal("escrow"),
+            nativeToScVal("created"),
+          ],
+          value: nativeToScVal({
+            buyer: "buyer-addr",
+            seller: "merchant-addr",
+            amount: 1000n, // u128 map
+            order_id: "order-99",
+          }),
+        },
+      ],
+    });
+    SorobanRpc.Server.prototype.getEvents = mockGetEvents;
+
+    startEscrowEventListener("http://localhost:8000", "C123");
+
+    // Fast forward to allow async operations to run
+    await vi.runOnlyPendingTimersAsync();
+
+    expect(mockGetEvents).toHaveBeenCalled();
+    expect(idempotency.checkAndMarkDispatched).toHaveBeenCalledWith(
+      expect.anything(),
+      expect.objectContaining({
+        userId: "merchant-123",
+        eventType: "escrow_created",
+      })
+    );
+    expect(email.sendEmail).toHaveBeenCalledWith(
+      expect.objectContaining({
+        to: "merchant@example.com",
+        templateName: "escrow-funded",
+      })
+    );
+  });
+});

apps/backend/notifications/src/escrowListener.ts

@@ -0,0 +1,252 @@
+import { rpc as SorobanRpc, scValToNative } from "@stellar/stellar-sdk";
+import { createLogger } from "@delego/utils";
+import { Redis } from "ioredis";
+import { getWalletLookupAdapter } from "./walletLookup.js";
+import { checkAndMarkDispatched } from "./idempotency.js";
+import { sendEmail } from "../email/index.js";
+import { sendPushNotification, type PushPayload, type PushSubscription } from "../push/index.js";
+
+const log = createLogger(
+  "notifications:escrow-listener",
+  process.env.LOG_LEVEL ?? "info"
+);
+
+const redis = new Redis(process.env.REDIS_URL ?? "redis://localhost:6379");
+
+export interface EscrowContractEvent {
+  contractId: string;
+  eventType: "escrow_created" | "escrow_released" | "escrow_refunded" | "escrow_disputed";
+  orderId: string;
+  buyer: string;
+  merchant: string;
+  amountStroops: string;
+  ledger: number;
+  txHash: string;
+}
+
+let isRunning = false;
+let timeoutId: NodeJS.Timeout | null = null;
+
+export function startEscrowEventListener(rpcUrl: string, contractId: string): void {
+  if (isRunning) {
+    log.warn("Escrow event listener is already running");
+    return;
+  }
+  isRunning = true;
+
+  const server = new SorobanRpc.Server(rpcUrl);
+  log.info("Starting escrow event listener", { rpcUrl, contractId });
+
+  const poll = async () => {
+    if (!isRunning) return;
+
+    try {
+      const latestLedgerResponse = await server.getLatestLedger();
+      const latestLedger = latestLedgerResponse.sequence;
+
+      const lastProcessedKey = `escrow_listener:last_ledger:${contractId}`;
+      const lastProcessedStr = await redis.get(lastProcessedKey);
+
+      let startLedger = lastProcessedStr ? parseInt(lastProcessedStr, 10) + 1 : latestLedger - 100;
+
+      if (startLedger > latestLedger) {
+        startLedger = latestLedger;
+      }
+
+      if (startLedger <= latestLedger) {
+        const eventsResponse = await server.getEvents({
+          startLedger,
+          filters: [
+            {
+              type: "contract",
+              contractIds: [contractId],
+              topics: [["*"]], // Match any topics for this contract
+            },
+          ],
+          limit: 1000,
+        });
+
+        for (const event of eventsResponse.events) {
+          if (event.type !== "contract" || !event.inSuccessfulContractCall) {
+            continue;
+          }
+
+          try {
+            const topics = event.topic.map(t => scValToNative(t));
+            if (topics[0] !== "escrow" || !topics[1]) continue;
+
+            const eventTypeRaw = topics[1];
+            let eventType: EscrowContractEvent["eventType"];
+
+            switch (eventTypeRaw) {
+              case "created":
+                eventType = "escrow_created";
+                break;
+              case "resolved":
+                eventType = "escrow_released"; 
+                break;
+              case "refunded":
+                eventType = "escrow_refunded";
+                break;
+              case "disputed":
+                eventType = "escrow_disputed";
+                break;
+              default:
+                continue;
+            }
+
+            const value = scValToNative(event.value);
+
+            if (eventTypeRaw === "resolved" && value && value.release_to_seller === false) {
+              eventType = "escrow_refunded";
+            }
+            if (eventTypeRaw === "resolved" && value && value.release_to_seller === true) {
+              eventType = "escrow_released";
+            }
+
+            const escrowEvent: EscrowContractEvent = {
+              contractId: event.contractId,
+              eventType,
+              orderId: value.order_id || value.orderId || "", 
+              buyer: value.buyer || "",
+              merchant: value.seller || value.merchant || "", 
+              amountStroops: value.amount ? value.amount.toString() : "0",
+              ledger: event.ledger,
+              txHash: event.txHash,
+            };
+
+            await dispatchEscrowNotification(escrowEvent, `${event.txHash}-${event.id}`);
+          } catch (e) {
+            log.warn("Failed to parse or process event", { error: e, event });
+          }
+        }
+
+        await redis.set(lastProcessedKey, latestLedger.toString());
+      }
+    } catch (error) {
+      log.error("Error polling escrow events", { error });
+    }
+
+    if (isRunning) {
+      timeoutId = setTimeout(poll, 5000);
+    }
+  };
+
+  poll();
+}
+
+export function stopEscrowEventListener(): void {
+  isRunning = false;
+  if (timeoutId) {
+    clearTimeout(timeoutId);
+    timeoutId = null;
+  }
+}
+
+async function dispatchEscrowNotification(event: EscrowContractEvent, eventId: string) {
+  const adapter = getWalletLookupAdapter();
+
+  // Decide who needs the notification
+  let targetAddress: string | null = null;
+  let subject = "";
+  let templateName = "";
+  let body = "";
+
+  if (event.eventType === "escrow_created") {
+    // Escrow created by buyer, notify merchant
+    targetAddress = event.merchant;
+    subject = "New Escrow Funded";
+    templateName = "escrow-funded";
+    body = `A new escrow for order ${event.orderId} has been funded.`;
+  } else if (event.eventType === "escrow_released") {
+    // Funds released to merchant, notify merchant
+    targetAddress = event.merchant;
+    subject = "Escrow Released";
+    templateName = "escrow-released";
+    body = `Funds for order ${event.orderId} have been released to you.`;
+  } else if (event.eventType === "escrow_refunded") {
+    // Funds refunded to buyer, notify buyer
+    targetAddress = event.buyer;
+    subject = "Escrow Refunded";
+    templateName = "escrow-refunded";
+    body = `Funds for order ${event.orderId} have been refunded to you.`;
+  } else if (event.eventType === "escrow_disputed") {
+    // Disputed, notify both? For simplicity, notify merchant
+    targetAddress = event.merchant;
+    subject = "Escrow Disputed";
+    templateName = "escrow-disputed";
+    body = `A dispute has been opened for order ${event.orderId}.`;
+  }
+
+  if (!targetAddress) return;
+
+  const target = await adapter.lookupByWalletAddress(targetAddress);
+  if (!target) {
+    log.info("No notification target found for wallet", { targetAddress, eventId });
+    return;
+  }
+
+  const tasks: Promise<void>[] = [];
+
+  // Email
+  if (target.email) {
+    const shouldSend = await checkAndMarkDispatched(redis, {
+      userId: target.userId,
+      channel: "email",
+      eventType: event.eventType,
+      eventId,
+    });
+
+    if (shouldSend) {
+      tasks.push(
+        sendEmail({
+          to: target.email,
+          subject,
+          templateName,
+          templateData: {
+            orderId: event.orderId,
+            amount: event.amountStroops,
+          },
+        }).catch((err) =>
+          log.error("Failed to send escrow email", { error: err, userId: target.userId })
+        )
+      );
+    }
+  }
+
+  // Push
+  if (target.pushEnabled) {
+    const shouldSend = await checkAndMarkDispatched(redis, {
+      userId: target.userId,
+      channel: "push",
+      eventType: event.eventType,
+      eventId,
+    });
+
+    if (shouldSend) {
+      const payload: PushPayload = {
+        title: subject,
+        body,
+        data: {
+          type: event.eventType,
+          orderId: event.orderId,
+          amount: event.amountStroops,
+        },
+      };
+
+      tasks.push(
+        (async () => {
+          const subscriptions = await redis.smembers(`push:subscriptions:${target.userId}`);
+          for (const subStr of subscriptions) {
+            const sub = JSON.parse(subStr) as PushSubscription;
+            await sendPushNotification(sub, payload).catch((err) =>
+              log.error("Failed to send escrow push", { error: err, userId: target.userId })
+            );
+          }
+        })()
+      );
+    }
+  }
+
+  await Promise.all(tasks);
+}

apps/backend/notifications/src/index.ts

@@ -9,6 +9,7 @@ import {
   dispatchTransactionApproval,
 } from "./dispatcher.js";
 import { getVapidPublicKey } from "../push/index.js";
+import { startEscrowEventListener } from "./escrowListener.js";
 import type { IncomingMessage, ServerResponse } from "node:http";
 
 const SERVICE_NAME = "notifications";
@@ -125,3 +126,11 @@ const server = startHttpServer({
 });
 
 initWebSocketServer(server);
+
+const escrowContractId = process.env.ESCROW_CONTRACT_ID;
+const rpcUrl = process.env.SOROBAN_RPC_URL;
+if (escrowContractId && rpcUrl) {
+  startEscrowEventListener(rpcUrl, escrowContractId);
+} else {
+  log.warn("Escrow listener not started: ESCROW_CONTRACT_ID or SOROBAN_RPC_URL missing");
+}