Implement idempotent escrow operations

scripts/replay-escrow.mjs

@@ -2,6 +2,7 @@ import { Networks } from "@stellar/stellar-sdk";
 import { getDb } from "../src/lib/mongodb.js";
 import { createJsonRpcEventSource } from "../src/lib/indexer/stellarIndexer.js";
 import { applyEscrowEvent } from "../src/lib/indexer/escrowIndexer.js";
+import { reconcileEscrowOperations } from "../src/lib/escrow/escrowOperations.js";
 
 const rpcUrl = process.env.NEXT_PUBLIC_STELLAR_RPC_URL;
 const networkPassphrase =
@@ -18,37 +19,76 @@ if (!contractId) {
   throw new Error("TRUSTLESS_WORK_CONTRACT_ID_TESTNET or NEXT_PUBLIC_TRUSTLESS_WORK_CONTRACT_ID is required to run the escrow indexer");
 }
 
-const startLedger = parseInt(process.argv[2], 10);
-const limit = parseInt(process.argv[3], 10) || 100;
+const mode = process.argv[2] === "reconcile" ? "reconcile" : "replay";
+const startLedger = parseInt(mode === "reconcile" ? process.argv[3] : process.argv[2], 10);
+const limit = parseInt(mode === "reconcile" ? process.argv[4] : process.argv[3], 10) || 100;
+const actor = process.env.ESCROW_REPLAY_ACTOR || process.env.USER || "admin-cli";
 
 if (!startLedger || isNaN(startLedger)) {
   console.error("Usage: node scripts/replay-escrow.mjs <startLedger> [limit]");
+  console.error("   or: node scripts/replay-escrow.mjs reconcile <startLedger> [limit]");
   process.exit(1);
 }
 
 const db = await getDb();
 const eventSource = createJsonRpcEventSource({ rpcUrl, contractId, networkPassphrase });
 
-console.log(`Replaying escrow events from ledger ${startLedger} with limit ${limit}`);
+function matchesOperation(event, operation) {
+  if (operation.transactionHash && (event.transactionHash === operation.transactionHash || event.txHash === operation.transactionHash)) {
+    return true;
+  }
+  if (operation.payload?.escrowId && event.escrowId === operation.payload.escrowId) {
+    return true;
+  }
+  return false;
+}
 
 try {
-  const batch = await eventSource.getEvents({ startLedger, limit });
-  const events = batch.events || [];
-
-  console.log(`Found ${events.length} events to replay.`);
-
-  let applied = 0;
-  let skipped = 0;
-
-  for (const event of events) {
-    const result = await applyEscrowEvent(db, { ...event, source: "escrow" });
-    if (result.skipped) skipped += 1;
-    else applied += 1;
+  if (mode === "reconcile") {
+    console.log(`Reconciling escrow operations from ledger ${startLedger} with event limit ${limit}`);
+    const batch = await eventSource.getEvents({ startLedger, limit });
+    const events = batch.events || [];
+    const result = await reconcileEscrowOperations(db, {
+      actor,
+      limit,
+      queryChainState: async (operation) => {
+        const event = events.find((candidate) => matchesOperation(candidate, operation));
+        if (!event) return { found: false, reason: "no matching escrow event found" };
+        return {
+          found: true,
+          confirmed: true,
+          transactionHash: event.transactionHash || event.txHash || operation.transactionHash || null,
+          ledgerSequence: event.ledger ?? event.ledgerSequence ?? operation.ledgerSequence ?? null,
+          event,
+        };
+      },
+      project: async (operation) => {
+        const event = operation.onChainState?.event;
+        if (!event) throw new Error("missing on-chain event for escrow projection");
+        await applyEscrowEvent(db, { ...event, source: "escrow" });
+      },
+    });
+    console.log(JSON.stringify({ mode, actor, ...result }, null, 2));
+  } else {
+    console.log(`Replaying escrow events from ledger ${startLedger} with limit ${limit}`);
+    const batch = await eventSource.getEvents({ startLedger, limit });
+    const events = batch.events || [];
+
+    console.log(`Found ${events.length} events to replay.`);
+
+    let applied = 0;
+    let skipped = 0;
+
+    for (const event of events) {
+      const result = await applyEscrowEvent(db, { ...event, source: "escrow" });
+      if (result.skipped) skipped += 1;
+      else applied += 1;
+    }
+
+    console.log(`Replay complete. Applied: ${applied}, Skipped: ${skipped}`);
   }
-
-  console.log(`Replay complete. Applied: ${applied}, Skipped: ${skipped}`);
 } catch (err) {
-  console.error("Replay failed:", err);
+  console.error(`${mode === "reconcile" ? "Reconcile" : "Replay"} failed:`, err);
 }
 
 process.exit(0);

src/app/api/escrow-operations/[idempotencyKey]/route.js

@@ -0,0 +1,70 @@
+import { NextResponse } from "next/server";
+
+import { getUserFromCookie } from "@/lib/api/auth";
+import { withApiHardening } from "@/lib/api/hardening";
+import { COLLECTIONS } from "@/lib/backend/schemaContracts";
+import { getDb } from "@/lib/mongodb";
+
+export const dynamic = "force-dynamic";
+export const runtime = "nodejs";
+
+function publicOperation(operation) {
+  return {
+    idempotencyKey: operation.idempotencyKey,
+    operationType: operation.operationType,
+    state: operation.state,
+    stage: operation.stage || null,
+    transactionHash: operation.transactionHash || null,
+    ledgerSequence: operation.ledgerSequence ?? null,
+    retryCount: operation.retryCount || 0,
+    reconciliationFailureCount: operation.reconciliationFailureCount || 0,
+    terminal: operation.terminal === true,
+    nextAttemptAt: operation.nextAttemptAt || null,
+    createdAt: operation.createdAt,
+    updatedAt: operation.updatedAt,
+  };
+}
+
+export async function GET(request, { params }) {
+  return withApiHardening(
+    request,
+    { route: "escrow-operation-status", rateLimit: { limit: 60, windowMs: 60_000 } },
+    async () => {
+      const user = await getUserFromCookie(request);
+      if (!user) {
+        return NextResponse.json({ error: "Unauthorized" }, { status: 401 });
+      }
+
+      const { idempotencyKey } = await params;
+      if (!idempotencyKey) {
+        return NextResponse.json({ error: "Missing idempotency key" }, { status: 400 });
+      }
+
+      const db = await getDb();
+      const operation = await db
+        .collection(COLLECTIONS.escrowOperations)
+        .findOne({ idempotencyKey: decodeURIComponent(idempotencyKey) });
+
+      if (!operation) {
+        return NextResponse.json({ error: "Not found" }, { status: 404 });
+      }
+
+      const userWallets = [
+        user.walletAddress,
+        user.walletAddressLower,
+        user.payoutWalletAddress,
+        user.payoutWalletAddressLower,
+        user.address,
+        user.id,
+      ]
+        .filter(Boolean)
+        .map((value) => String(value).toLowerCase());
+      const actor = operation.actor ? String(operation.actor).toLowerCase() : null;
+      if (actor && !userWallets.includes(actor) && user.role !== "admin") {
+        return NextResponse.json({ error: "Forbidden" }, { status: 403 });
+      }
+
+      return NextResponse.json(publicOperation(operation));
+    },
+  );
+}

src/hooks/useEscrow.js

@@ -27,6 +27,31 @@ async function fetchEscrowMilestones(escrowId) {
   return res.json();
 }
 
+async function fetchEscrowOperation(idempotencyKey) {
+  if (!idempotencyKey) return null;
+  const res = await fetch(`/api/escrow-operations/${idempotencyKey}`);
+  if (!res.ok) {
+    if (res.status === 404) return null;
+    throw new Error("Failed to fetch escrow operation");
+  }
+  return res.json();
+}
+
+export function getEscrowOperationStatus(operation) {
+  if (!operation) return { state: "unknown", recoverable: false, pending: false, failed: false };
+  const failed = operation.state === "failed";
+  const pending = ["pending", "submitted", "reconciling"].includes(operation.state);
+  return {
+    state: operation.state,
+    stage: operation.stage || null,
+    pending,
+    failed,
+    recoverable: failed && operation.terminal !== true,
+    retryCount: operation.retryCount || 0,
+    reconciliationFailureCount: operation.reconciliationFailureCount || 0,
+  };
+}
+
 export function useEscrow(escrowId, { enabled = true, refetchInterval = false } = {}) {
   return useQuery({
     queryKey: ["escrow", escrowId],
@@ -36,6 +61,19 @@ export function useEscrow(escrowId, { enabled = true, refetchInterval = false }
   });
 }
 
+export function useEscrowOperation(idempotencyKey, { enabled = true, refetchInterval = false } = {}) {
+  return useQuery({
+    queryKey: ["escrow-operation", idempotencyKey],
+    queryFn: () => fetchEscrowOperation(idempotencyKey),
+    enabled: enabled && !!idempotencyKey,
+    refetchInterval,
+    select: (operation) => ({
+      operation,
+      status: getEscrowOperationStatus(operation),
+    }),
+  });
+}
+
 export function useUserEscrows(walletAddress, { enabled = true } = {}) {
   return useQuery({
     queryKey: ["escrows", "user", walletAddress],

src/lib/backend/schemaContracts.js

@@ -33,6 +33,13 @@ export const COLLECTIONS = Object.freeze({
   files: "files",
   fileCleanupOutbox: "file_cleanup_outbox",
   uploadQuarantine: "upload_quarantine",
+
+  // Escrow / Trustless Work.
+  escrows: "escrows",
+  milestones: "milestones",
+  payouts: "payouts",
+  escrowOperations: "escrow_operations",
+  escrowOperationAudit: "escrow_operation_audit",
 });
 
 // File lifecycle states (#98). A file object moves:
@@ -62,10 +69,6 @@ export const FILE_PURPOSES = Object.freeze({
   MILESTONE_EVIDENCE: { purpose: "milestone_evidence", visibility: FILE_VISIBILITY.PRIVATE, maxBytes: 25 * 1024 * 1024 },
   PAYOUT_DOCUMENT: { purpose: "payout_document", visibility: FILE_VISIBILITY.PRIVATE, maxBytes: 10 * 1024 * 1024 },
   FEEDBACK_ATTACHMENT: { purpose: "feedback_attachment", visibility: FILE_VISIBILITY.PRIVATE, maxBytes: 10 * 1024 * 1024 },
-  // Escrow / Trustless Work.
-  escrows: "escrows",
-  milestones: "milestones",
-  payouts: "payouts",
 });
 
 export const REQUIRED_INDEXES = Object.freeze({
@@ -574,6 +577,10 @@ export const REQUIRED_INDEXES = Object.freeze({
     {
       name: "quarantine_status_lease",
       keys: { status: 1, leaseUntil: 1 },
+      options: {},
+    },
+  ],
+
   escrows: [
     {
       name: "escrows_escrow_id_unique",
@@ -625,6 +632,39 @@ export const REQUIRED_INDEXES = Object.freeze({
       options: {},
     },
   ],
+
+  escrow_operations: [
+    {
+      name: "escrow_operations_idempotency_key_unique",
+      keys: { idempotencyKey: 1 },
+      options: { unique: true },
+    },
+    {
+      name: "escrow_operations_state_next_attempt",
+      keys: { state: 1, nextAttemptAt: 1 },
+      options: {},
+    },
+    {
+      name: "escrow_operations_transaction_hash",
+      keys: { transactionHash: 1 },
+      options: {
+        partialFilterExpression: { transactionHash: { $type: "string" } },
+      },
+    },
+  ],
+
+  escrow_operation_audit: [
+    {
+      name: "escrow_operation_audit_operation_created_at",
+      keys: { operationId: 1, createdAt: -1 },
+      options: {},
+    },
+    {
+      name: "escrow_operation_audit_actor_created_at",
+      keys: { actor: 1, createdAt: -1 },
+      options: {},
+    },
+  ],
 });
 
 // ── Material field contracts ───────────────────────────────────────────────
@@ -1073,6 +1113,51 @@ export const COLLECTION_VALIDATORS = Object.freeze({
       },
     },
   },
+
+  escrow_operations: {
+    $jsonSchema: {
+      bsonType: "object",
+      required: [
+        "idempotencyKey",
+        "operationType",
+        "payloadHash",
+        "actor",
+        "state",
+        "createdAt",
+        "updatedAt",
+      ],
+      properties: {
+        idempotencyKey: { bsonType: "string", minLength: 1 },
+        operationType: { bsonType: "string", minLength: 1 },
+        payloadHash: { bsonType: "string", minLength: 64 },
+        actor: { bsonType: ["string", "null"] },
+        state: { enum: ["pending", "submitted", "confirmed", "failed", "reconciling"] },
+        transactionHash: { bsonType: ["string", "null"] },
+        ledgerSequence: { bsonType: ["int", "long", "null"] },
+        retryCount: { bsonType: ["int", "long"] },
+        reconciliationFailureCount: { bsonType: ["int", "long"] },
+        terminal: { bsonType: "bool" },
+        nextAttemptAt: { bsonType: ["date", "null"] },
+        createdAt: { bsonType: "date" },
+        updatedAt: { bsonType: "date" },
+      },
+    },
+  },
+
+  escrow_operation_audit: {
+    $jsonSchema: {
+      bsonType: "object",
+      required: ["operationId", "action", "actor", "createdAt"],
+      properties: {
+        operationId: { bsonType: "string", minLength: 1 },
+        action: { bsonType: "string", minLength: 1 },
+        actor: { bsonType: ["string", "null"] },
+        before: { bsonType: ["object", "null"] },
+        after: { bsonType: ["object", "null"] },
+        createdAt: { bsonType: "date" },
+      },
+    },
+  },
 });
 
 export const EDITABLE_MATERIAL_FIELDS = Object.freeze([