feat: tranche-based partial refund countdown and automated alert system

apps/api/src/db/migrations/013_add_tranche_refund_alerts.sql

@@ -0,0 +1,15 @@
+CREATE TYPE alert_notification_status AS ENUM ('PENDING', 'WARNING_SENT', 'REFUND_EXECUTED', 'CANCELLED');
+
+CREATE TABLE tranche_refund_schedules (
+    id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
+    trade_id VARCHAR(64) NOT NULL UNIQUE REFERENCES cash_requests(id) ON DELETE CASCADE,
+    total_tranches INT NOT NULL,
+    unreleased_tranches INT NOT NULL,
+    unreleased_amount BIGINT NOT NULL,
+    timeout_ledger_sequence INT NOT NULL,
+    status alert_notification_status NOT NULL DEFAULT 'PENDING',
+    created_at TIMESTAMP WITH TIME ZONE DEFAULT CURRENT_TIMESTAMP,
+    updated_at TIMESTAMP WITH TIME ZONE DEFAULT CURRENT_TIMESTAMP
+);
+
+CREATE INDEX idx_tranche_timeout ON tranche_refund_schedules(timeout_ledger_sequence, status);

apps/api/src/index.ts

@@ -8,6 +8,7 @@ import { StellarIndexerWorker, PgAdvisoryLock } from "./lib/workers/stellarIndex
 import { startChatCleanupWorker } from "./lib/workers/chatCleanupWorker.js";
 import { startBatchAuctionWorker } from "./lib/workers/batchAuctionWorker.js";
 import { startSessionRotationWorker } from "./lib/workers/sessionRotationWorker.js";
+import { startTrancheRefundWorker } from "./lib/workers/trancheRefundWorker.js";
 import { createSessionKeyRegistryStore } from "./lib/session-registry-store.js";
 import { createClient } from "redis";
 import { CircuitBreakerStore } from "./lib/circuit-breaker-store.js";

apps/api/src/lib/workers/trancheRefundWorker.ts

@@ -0,0 +1,165 @@
+import { createClient } from "redis";
+import { pgPool } from "../../app.js";
+import { getLatestLedgerSequence, refundEscrow } from "../stellar.js";
+import { getCashRequest } from "../store.js";
+import { sendRefundCountdownAlert } from "../webhook.js";
+
+const QUEUE_NAME = "velo:tranche-refund-queue";
+const GROUP_NAME = "tranche-refund-group";
+const DLQ_NAME = "velo:tranche-refund-dlq";
+const POLL_INTERVAL_MS = 5000;
+const MAX_ATTEMPTS = 5;
+
+export async function startTrancheRefundWorker() {
+  const redisUrl = process.env.REDIS_URL || "redis://localhost:6379";
+  const redis = createClient({ url: redisUrl });
+
+  await redis.connect();
+
+  try {
+    await redis.xGroupCreate(QUEUE_NAME, GROUP_NAME, "0", { MKSTREAM: true });
+  } catch (err: any) {
+    if (!err.message.includes("BUSYGROUP")) {
+      console.error("Error creating redis group", err);
+    }
+  }
+
+  let stopped = false;
+  let ticking = false;
+
+  async function checkExpiringLedgers() {
+    if (!pgPool) return;
+    try {
+      const currentLedger = await getLatestLedgerSequence();
+
+      const client = await pgPool.connect();
+      try {
+        await client.query("BEGIN");
+
+        // Find pending schedules that need warnings
+        const warningRes = await client.query(
+          `SELECT trade_id, unreleased_tranches, unreleased_amount, timeout_ledger_sequence
+           FROM tranche_refund_schedules
+           WHERE status = 'PENDING' AND timeout_ledger_sequence - $1 <= 100
+           FOR UPDATE SKIP LOCKED`,
+          [currentLedger]
+        );
+
+        for (const row of warningRes.rows) {
+          const trade = getCashRequest(row.trade_id);
+          if (trade) {
+             const estimatedSeconds = Math.max(0, row.timeout_ledger_sequence - currentLedger) * 6;
+             await sendRefundCountdownAlert({
+               tradeId: row.trade_id,
+               amountStroops: row.unreleased_amount.toString(),
+               buyer: trade.buyer,
+               seller: trade.seller,
+               timeoutLedger: row.timeout_ledger_sequence,
+               latestLedger: currentLedger,
+               ledgersUntilRefund: Math.max(0, row.timeout_ledger_sequence - currentLedger),
+               estimatedSecondsUntilRefund: estimatedSeconds
+             });
+          }
+          await client.query(`UPDATE tranche_refund_schedules SET status = 'WARNING_SENT' WHERE trade_id = $1`, [row.trade_id]);
+        }
+
+        // Find schedules that reached timeout but haven't been queued yet
+        // For simplicity, we assume they get moved to REFUND_EXECUTED by the manual trigger or the worker itself.
+        // The automated fallback execution worker triggers refundEscrow() for remaining unreleased tranches when thresholds are reached.
+        const executeRes = await client.query(
+          `SELECT trade_id
+           FROM tranche_refund_schedules
+           WHERE (status = 'PENDING' OR status = 'WARNING_SENT') AND $1 > timeout_ledger_sequence
+           FOR UPDATE SKIP LOCKED`,
+          [currentLedger]
+        );
+
+        for (const row of executeRes.rows) {
+           await redis.xAdd(QUEUE_NAME, "*", { tradeId: row.trade_id });
+        }
+
+        await client.query("COMMIT");
+      } catch (err) {
+        await client.query("ROLLBACK");
+        console.error("Error checking expiring ledgers", err);
+      } finally {
+        client.release();
+      }
+    } catch (err) {
+      console.error("Error fetching latest ledger", err);
+    }
+  }
+
+  async function processQueue() {
+    try {
+      const response = await redis.xReadGroup(
+        GROUP_NAME,
+        `consumer-${process.pid}`,
+        [{ key: QUEUE_NAME, id: ">" }],
+        { COUNT: 10 }
+      );
+
+      if (!response) return;
+
+      for (const stream of response as any[]) {
+        for (const entry of stream.messages) {
+          const { tradeId } = entry.message;
+          let success = false;
+
+          for (let attempt = 1; attempt <= MAX_ATTEMPTS; attempt++) {
+             try {
+                // Perform the on-chain refund. In a real app we need the trade details.
+                const trade = getCashRequest(tradeId);
+                if (trade) {
+                   await refundEscrow({ contractId: trade.contractId, tradeId });
+                }
+
+                // Update DB via the trigger route logic (or directly here if it was automated).
+                // If it was already set to REFUND_EXECUTED by the route, we just execute on-chain.
+                if (pgPool) {
+                   await pgPool.query(`UPDATE tranche_refund_schedules SET status = 'REFUND_EXECUTED' WHERE trade_id = $1`, [tradeId]);
+                }
+
+                success = true;
+                break;
+             } catch (err) {
+                console.error(`Refund failed for trade ${tradeId}, attempt ${attempt}`, err);
+                await new Promise(r => setTimeout(r, 1000 * Math.pow(2, attempt)));
+             }
+          }
+
+          if (success) {
+            await redis.xAck(QUEUE_NAME, GROUP_NAME, entry.id);
+          } else {
+            await redis.xAdd(DLQ_NAME, "*", { tradeId, reason: "Max attempts reached" });
+            await redis.xAck(QUEUE_NAME, GROUP_NAME, entry.id);
+          }
+        }
+      }
+    } catch (err) {
+      console.error("Error processing refund queue", err);
+    }
+  }
+
+  async function tick() {
+    if (stopped || ticking) return;
+    ticking = true;
+    try {
+      await checkExpiringLedgers();
+      await processQueue();
+    } finally {
+      ticking = false;
+    }
+  }
+
+  const timer = setInterval(() => {
+    void tick();
+  }, POLL_INTERVAL_MS);
+  timer.unref();
+
+  return async () => {
+    stopped = true;
+    clearInterval(timer);
+    await redis.quit();
+  };
+}

apps/api/src/routes/__tests__/tranche-refund.test.ts

@@ -0,0 +1,39 @@
+import { describe, it, expect, vi } from "vitest";
+import { app } from "../../app.js";
+import { pgPool } from "../../app.js";
+import { getCashRequest, saveCashRequest } from "../../lib/store.js";
+import { getLatestLedgerSequence } from "../../lib/stellar.js";
+
+vi.mock("../../lib/stellar.js", async (importOriginal) => {
+  const mod = await importOriginal<any>();
+  return {
+    ...mod,
+    getLatestLedgerSequence: vi.fn().mockResolvedValue(100),
+  };
+});
+
+describe("Tranche Refund API", () => {
+  it("should return 400 if timeout not reached", async () => {
+    if (!pgPool) return;
+
+    // Setup DB
+    const client = await pgPool.connect();
+    await client.query("BEGIN");
+    await client.query(
+      `INSERT INTO tranche_refund_schedules (trade_id, total_tranches, unreleased_tranches, unreleased_amount, timeout_ledger_sequence)
+       VALUES ('test_trade_1', 3, 2, 200, 150)`
+    );
+    await client.query("COMMIT");
+    client.release();
+
+    const response = await app.inject({
+      method: "POST",
+      url: "/api/v1/tranche-refund/trigger",
+      payload: { tradeId: "test_trade_1".padEnd(64, "0") }
+    });
+
+    expect(response.statusCode).toBe(400);
+    const body = JSON.parse(response.payload);
+    expect(body.error.code).toBe("TIMEOUT_NOT_REACHED");
+  });
+});

apps/api/src/routes/tranche-refund.ts

@@ -0,0 +1,109 @@
+import type { FastifyPluginAsync } from "fastify";
+import { z } from "zod";
+import { pgPool } from "../app.js";
+import { getLatestLedgerSequence, submitRefundTx } from "../lib/stellar.js";
+import { getCashRequest, updateStatus } from "../lib/store.js";
+import { CONTRACTS } from "@velo/shared";
+import { createClient } from "redis";
+
+export const TriggerTrancheRefundSchema = z.object({
+  tradeId: z.string().length(64, "Trade ID must be a 64-character hex string"),
+});
+
+export const trancheRefundRoutes: FastifyPluginAsync = async (fastify) => {
+  fastify.post("/tranche-refund/trigger", async (req, reply) => {
+    try {
+      const parsed = TriggerTrancheRefundSchema.safeParse(req.body);
+      if (!parsed.success) {
+        return reply.status(400).send({ error: "VALIDATION_ERROR", details: parsed.error.issues });
+      }
+
+      const { tradeId } = parsed.data;
+
+      if (!pgPool) {
+         return reply.status(500).send({ error: "DATABASE_NOT_CONFIGURED" });
+      }
+
+      const client = await pgPool.connect();
+      try {
+        await client.query("BEGIN");
+
+        const result = await client.query(
+          `SELECT trade_id, unreleased_tranches, unreleased_amount, status, timeout_ledger_sequence 
+           FROM tranche_refund_schedules 
+           WHERE trade_id = $1 
+           FOR UPDATE`,
+          [tradeId]
+        );
+
+        if (result.rowCount === 0) {
+          await client.query("ROLLBACK");
+          return reply.status(404).send({ error: "NOT_FOUND", message: "Trade schedule not found." });
+        }
+
+        const schedule = result.rows[0];
+
+        if (schedule.status === 'REFUND_EXECUTED' || schedule.status === 'CANCELLED') {
+          await client.query("ROLLBACK");
+          return reply.status(409).send({
+            error: {
+              code: "TRANCHE_ALREADY_SETTLED",
+              message: "Tranche trade has already been fully released or refunded.",
+              requestId: req.id
+            }
+          });
+        }
+
+        const currentLedger = await getLatestLedgerSequence();
+        if (currentLedger <= schedule.timeout_ledger_sequence) {
+          await client.query("ROLLBACK");
+          return reply.status(400).send({
+            error: {
+              code: "TIMEOUT_NOT_REACHED",
+              message: "Current Stellar ledger height has not reached the expiration threshold.",
+              requestId: req.id
+            }
+          });
+        }
+
+        // Update DB
+        await client.query(
+          `UPDATE tranche_refund_schedules SET status = 'REFUND_EXECUTED', updated_at = CURRENT_TIMESTAMP WHERE trade_id = $1`,
+          [tradeId]
+        );
+
+        // Update in-memory store
+        const trade = getCashRequest(tradeId);
+        if (trade) {
+          updateStatus(tradeId, "refunded");
+        }
+
+        await client.query("COMMIT");
+
+        // Async Relayer Offload
+        const redisUrl = process.env.REDIS_URL || "redis://localhost:6379";
+        const redis = createClient({ url: redisUrl });
+        await redis.connect();
+
+        await redis.xAdd('velo:tranche-refund-queue', '*', { tradeId });
+        await redis.quit();
+
+        return reply.status(200).send({
+          tradeId,
+          refundedAmount: schedule.unreleased_amount,
+          refundedTranches: schedule.unreleased_tranches,
+          status: 'REFUND_EXECUTED'
+        });
+
+      } catch (err) {
+        await client.query("ROLLBACK");
+        throw err;
+      } finally {
+        client.release();
+      }
+    } catch (err) {
+      req.log.error(err, "Tranche refund trigger failed");
+      return reply.status(500).send({ error: "INTERNAL_SERVER_ERROR" });
+    }
+  });
+};

contracts/atomic-swap/src/lib.rs

@@ -410,7 +410,7 @@ impl Htlc for AtomicSwapContract {
             .publish((Symbol::new(&env, "released"), id), secret);
     }
 
-    fn refund(env: Env, id: BytesN<32>) {
+    fn refund(env: Env, id: BytesN<32>) -> i128 {
         let key = DataKey::Trade(id.clone());
         let mut state: TradeState = env
             .storage()
@@ -420,7 +420,7 @@ impl Htlc for AtomicSwapContract {
 
         // No-op if already released or refunded (trait invariant).
         if state.status != TradeStatus::Locked {
-            return;
+            return 0;
         }
         if env.ledger().sequence() < state.timeout_ledger {
             panic_with_error(&env, Error::TimeoutNotReached);
@@ -439,6 +439,8 @@ impl Htlc for AtomicSwapContract {
 
         env.events()
             .publish((Symbol::new(&env, "refunded"), id), state.amount);
+
+        state.amount
     }
 }
 

contracts/escrow/src/lib.rs

@@ -2159,7 +2159,7 @@ impl Htlc for EscrowContract {
         }
     }
 
-    fn refund(env: Env, id: BytesN<32>) {
+    fn refund(env: Env, id: BytesN<32>) -> i128 {
         // Issue #266: intentionally does NOT call check_not_paused — same
         // reasoning as `release`: already-locked funds must never be trapped
         // by the circuit breaker.
@@ -2206,6 +2206,8 @@ impl Htlc for EscrowContract {
 
         env.events()
             .publish((symbol_short(&env, "refunded"), id), refund_amount);
+
+        refund_amount
     }
 }
 

contracts/htlc-core/src/lib.rs

@@ -74,8 +74,8 @@ pub trait Htlc {
 
     /// Permissionless refund back to the buyer once timeout_ledger has
     /// passed. Anyone can call this — it does not require the buyer's
-    /// signature, only that the timeout has elapsed.
-    fn refund(env: Env, id: BytesN<32>);
+    /// signature, only that the timeout has elapsed. Returns the amount refunded.
+    fn refund(env: Env, id: BytesN<32>) -> i128;
 }
 
 /// Denominator for basis-point arithmetic: 10_000 bps == 100%.