munisp · devin-ai-integration · Jul 6, 2026 · Jul 6, 2026
diff --git a/server/_core/transferPipeline.ts b/server/_core/transferPipeline.ts
@@ -29,6 +29,17 @@ import { broadcastUserEvent } from "../sse.service";
 import { createAuditLog } from "../db";
 import { checkFraud, checkVelocity } from "../fraud.service";
 
+/**
+ * Deterministic TigerBeetle transfer id derived from the transfer UUID.
+ * Both the pipeline (which creates the pending hold) and the saga compensation
+ * (which must void that exact hold) derive the same id, so compensation can
+ * always target the real pending transfer instead of a fabricated random id.
+ */
+function pendingTransferIdFor(transferId: string): bigint {
+  const hex = transferId.replace(/-/g, "").slice(0, 32).padEnd(32, "0");
+  return BigInt(`0x${hex}`);
+}
+
 // ─── Types ───────────────────────────────────────────────────────────────────
 
 export interface TransferPipelineInput {
@@ -49,6 +60,12 @@ export interface TransferPipelineInput {
   skipKycTier?: boolean;
   /** Skip velocity check */
   skipVelocity?: boolean;
+  /**
+   * Skip the TigerBeetle ledger step. Set by callers (e.g. property escrow) that
+   * already create their own ledger-backed hold on a dedicated account/ledger, so
+   * the pipeline does not double-book a second pending transfer for the same funds.
+   */
+  skipLedger?: boolean;
   /** Additional metadata for audit log */
   metadata?: Record<string, unknown>;
 }
@@ -188,8 +205,9 @@ export async function executeTransferPipeline(input: TransferPipelineInput): Pro
   // 4. TigerBeetle double-entry ledger (FAIL-CLOSED in production)
   // Uses two-phase transfer: create pending hold, then post after settlement.
   // Validates balance before creating the transfer.
-  {
-    const transferBigId = BigInt(Date.now()) * BigInt(1000) + BigInt(Math.floor(Math.random() * 1000));
+  // Skipped when the caller already recorded a ledger-backed hold (e.g. escrow).
+  if (!input.skipLedger) {
+    const transferBigId = pendingTransferIdFor(input.transferId);
     const debitAccountId = BigInt(input.userId);
     const creditAccountId = BigInt(input.userId + 1_000_000);
     const amountCents = BigInt(Math.round(input.amount * 100));
@@ -358,7 +376,7 @@ export async function compensateFailedTransfer(input: {
   try {
     // Void the pending TigerBeetle transfer (two-phase: void releases the hold)
     const voidId = BigInt(Date.now()) * BigInt(1000) + BigInt(Math.floor(Math.random() * 1000));
-    const pendingId = BigInt(Date.now()) * BigInt(999) + BigInt(Math.floor(Math.random() * 999));
+    const pendingId = pendingTransferIdFor(input.transferId);
     await tigerBeetle.voidPendingTransfer({
       id: voidId,
       pendingId,

diff --git a/server/routers/propertyEscrow.ts b/server/routers/propertyEscrow.ts
@@ -410,16 +410,26 @@ const escrowPlanRouter = router({
       }).where(and(eq(wallets.id, wallet.id), sql`CAST(${wallets.balance} AS DECIMAL(18,2)) >= ${depositUsd}`)).returning();
       if (!debitedDeposit) throw new TRPCError({ code: "BAD_REQUEST", message: "Insufficient balance (concurrent update)" });
 
-      // Lock deposit in TigerBeetle (FAIL-CLOSED — escrow MUST be ledger-backed)
-      await tigerBeetle.createPendingTransfer({
-        id: BigInt(Date.now()),
-        debitAccountId: BigInt(ctx.user.id),
-        creditAccountId: plan.tigerBeetleEscrowAccount ?? BigInt(0),
-        amount: BigInt(Math.round(depositUsd * 100)),
-        ledger: ESCROW_LEDGER,
-        code: ESCROW_CODE,
-        timeoutSeconds: 86400, // 24h timeout for escrow holds
-      });
+      // Lock deposit in TigerBeetle (FAIL-CLOSED — escrow MUST be ledger-backed).
+      // If the ledger hold fails, refund the wallet debit so the buyer is not
+      // left short without an escrow record, then propagate the failure.
+      try {
+        await tigerBeetle.createPendingTransfer({
+          id: BigInt(Date.now()),
+          debitAccountId: BigInt(ctx.user.id),
+          creditAccountId: plan.tigerBeetleEscrowAccount ?? BigInt(0),
+          amount: BigInt(Math.round(depositUsd * 100)),
+          ledger: ESCROW_LEDGER,
+          code: ESCROW_CODE,
+          timeoutSeconds: 86400, // 24h timeout for escrow holds
+        });
+      } catch (err) {
+        await db.update(wallets).set({
+          balance: sql`CAST(CAST(${wallets.balance} AS DECIMAL(18,2)) + ${depositUsd} AS VARCHAR)`,
+          updatedAt: new Date(),
+        }).where(eq(wallets.id, wallet.id));
+        throw err;
+      }
 
       // Record transaction
       await db.insert(transactions).values({
@@ -461,6 +471,8 @@ const escrowPlanRouter = router({
         featureLabel: "property_escrow_deposit",
         transferId: `ESCROW-DEP-${plan.planId}`,
         description: `Escrow deposit: $${depositUsd.toFixed(2)} for plan ${plan.planId}`,
+        // Escrow already created its own ledger-backed hold above — do not double-book.
+        skipLedger: true,
         metadata: { planId: plan.planId, depositPct: Number(plan.depositPct) },
       });
 
@@ -500,16 +512,25 @@ const escrowPlanRouter = router({
       }).where(and(eq(wallets.id, wallet.id), sql`CAST(${wallets.balance} AS DECIMAL(18,2)) >= ${amount}`)).returning();
       if (!debitedInstallment) throw new TRPCError({ code: "BAD_REQUEST", message: "Insufficient balance (concurrent update)" });
 
-      // Lock in TigerBeetle (FAIL-CLOSED — installment MUST be ledger-backed)
-      await tigerBeetle.createPendingTransfer({
-        id: BigInt(Date.now()),
-        debitAccountId: BigInt(ctx.user.id),
-        creditAccountId: plan.tigerBeetleEscrowAccount ?? BigInt(0),
-        amount: BigInt(Math.round(amount * 100)),
-        ledger: ESCROW_LEDGER,
-        code: ESCROW_CODE,
-        timeoutSeconds: 86400,
-      });
+      // Lock in TigerBeetle (FAIL-CLOSED — installment MUST be ledger-backed).
+      // Refund the wallet debit if the ledger hold fails, then propagate.
+      try {
+        await tigerBeetle.createPendingTransfer({
+          id: BigInt(Date.now()),
+          debitAccountId: BigInt(ctx.user.id),
+          creditAccountId: plan.tigerBeetleEscrowAccount ?? BigInt(0),
+          amount: BigInt(Math.round(amount * 100)),
+          ledger: ESCROW_LEDGER,
+          code: ESCROW_CODE,
+          timeoutSeconds: 86400,
+        });
+      } catch (err) {
+        await db.update(wallets).set({
+          balance: sql`CAST(CAST(${wallets.balance} AS DECIMAL(18,2)) + ${amount} AS VARCHAR)`,
+          updatedAt: new Date(),
+        }).where(eq(wallets.id, wallet.id));
+        throw err;
+      }
 
       // Record transaction
       const [tx] = await db.insert(transactions).values({

diff --git a/services/ledger-service/go.mod b/services/ledger-service/go.mod
@@ -3,6 +3,6 @@ module github.com/remitflow/ledger-service
 go 1.22
 
 require (
-	github.com/gin-gonic/gin v1.10.0
 	github.com/google/uuid v1.6.0
+	github.com/lib/pq v1.12.3
 )
diff --git a/services/ledger-service/go.sum b/services/ledger-service/go.sum
@@ -0,0 +1,4 @@
+github.com/google/uuid v1.6.0 h1:NIvaJDMOsjHA8n1jAhLSgzrAzy1Hgr+hNrb57e+94F0=
+github.com/google/uuid v1.6.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo=
+github.com/lib/pq v1.12.3 h1:tTWxr2YLKwIvK90ZXEw8GP7UFHtcbTtty8zsI+YjrfQ=
+github.com/lib/pq v1.12.3/go.mod h1:/p+8NSbOcwzAEI7wiMXFlgydTwcgTr3OSKMsD2BitpA=
diff --git a/services/ledger-service/main.go b/services/ledger-service/main.go
@@ -53,13 +53,13 @@ const (
 
 // Transfer codes
 const (
-	TransferCodeStandard   = 1
-	TransferCodeReversal   = 2
-	TransferCodeFee        = 3
-	TransferCodeEscrowLock = 4
+	TransferCodeStandard      = 1
+	TransferCodeReversal      = 2
+	TransferCodeFee           = 3
+	TransferCodeEscrowLock    = 4
 	TransferCodeEscrowRelease = 5
-	TransferCodeFX         = 6
-	TransferCodePayroll    = 7
+	TransferCodeFX            = 6
+	TransferCodePayroll       = 7
 )
 
 // Transfer flags (TigerBeetle two-phase protocol)
@@ -153,6 +153,10 @@ func minorToAmount(minor int64) float64 {
 func initDB() *sql.DB {
 	dbURL := os.Getenv("DATABASE_URL")
 	if dbURL == "" {
+		// FAIL-CLOSED: never fall back to hardcoded credentials in production.
+		if os.Getenv("NODE_ENV") == "production" || os.Getenv("GO_ENV") == "production" {
+			log.Fatalf("[ledger-service] DATABASE_URL is required in production")
+		}
 		dbURL = "postgresql://remitflow:remitflow123@localhost:5432/remitflow?sslmode=disable"
 	}
 
@@ -250,16 +254,16 @@ func (s *LedgerService) handleHealth(w http.ResponseWriter, r *http.Request) {
 
 	resp := map[string]interface{}{
 		"status":                status,
-		"service":              "go-ledger-service",
-		"version":             "v2.0.0-production",
+		"service":               "go-ledger-service",
+		"version":               "v2.0.0-production",
 		"tigerbeetle_connected": true,
-		"postgres_connected":   pgOK,
-		"kafka_connected":      true,
-		"fail_closed":         s.isProduction,
-		"two_phase_enabled":   true,
-		"dapr_enabled":        true,
-		"temporal_enabled":    true,
-		"timestamp":           time.Now().UTC().Format(time.RFC3339),
+		"postgres_connected":    pgOK,
+		"kafka_connected":       true,
+		"fail_closed":           s.isProduction,
+		"two_phase_enabled":     true,
+		"dapr_enabled":          true,
+		"temporal_enabled":      true,
+		"timestamp":             time.Now().UTC().Format(time.RFC3339),
 	}
 	w.Header().Set("Content-Type", "application/json")
 	json.NewEncoder(w).Encode(resp)
@@ -383,6 +387,18 @@ func (s *LedgerService) handleCreateTransfer(w http.ResponseWriter, r *http.Requ
 		return
 	}
 
+	// Reject non-positive amounts. A negative amount would pass the balance
+	// pre-check (available < negative is false) and reverse the fund flow.
+	if req.Amount <= 0 {
+		w.WriteHeader(http.StatusBadRequest)
+		json.NewEncoder(w).Encode(map[string]interface{}{
+			"error":   "INVALID_AMOUNT",
+			"amount":  req.Amount,
+			"message": "Transfer amount must be greater than zero",
+		})
+		return
+	}
+
 	// Idempotency check
 	if req.IdempotencyKey != "" {
 		var existingID string
@@ -453,13 +469,24 @@ func (s *LedgerService) handleCreateTransfer(w http.ResponseWriter, r *http.Requ
 	if req.IdempotencyKey != "" {
 		idempKey = &req.IdempotencyKey
 	}
-	_, err = s.db.Exec(
+
+	// Insert the transfer and update balances atomically: a crash between the
+	// insert and the balance updates would otherwise leave the ledger inconsistent.
+	tx, err := s.db.Begin()
+	if err != nil {
+		log.Printf("[ledger-service] FAIL-CLOSED: begin tx failed: %v", err)
+		w.WriteHeader(http.StatusInternalServerError)
+		json.NewEncoder(w).Encode(map[string]string{"error": "TRANSFER_FAILED", "message": "Failed to persist transfer — operation blocked"})
+		return
+	}
+	defer tx.Rollback()
+
+	if _, err = tx.Exec(
 		`INSERT INTO ledger_transfers (id, debit_account_id, credit_account_id, amount, ledger, code, flags, timeout, status, idempotency_key)
 		 VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10)`,
 		transferID, req.DebitAccountID, req.CreditAccountID,
 		fmt.Sprintf("%d", amountMinor), 1, code, flags, req.TimeoutSeconds, status, idempKey,
-	)
-	if err != nil {
+	); err != nil {
 		log.Printf("[ledger-service] FAIL-CLOSED: Transfer persistence failed: %v", err)
 		w.WriteHeader(http.StatusInternalServerError)
 		json.NewEncoder(w).Encode(map[string]string{
@@ -469,22 +496,38 @@ func (s *LedgerService) handleCreateTransfer(w http.ResponseWriter, r *http.Requ
 		return
 	}
 
-	// Update balances
+	// Update balances (within the same transaction)
+	var balErr error
 	if req.Pending {
-		s.db.Exec("UPDATE ledger_accounts SET debits_pending = debits_pending + $1, updated_at = NOW() WHERE id = $2",
-			fmt.Sprintf("%d", amountMinor), req.DebitAccountID)
-		s.db.Exec("UPDATE ledger_accounts SET credits_pending = credits_pending + $1, updated_at = NOW() WHERE id = $2",
-			fmt.Sprintf("%d", amountMinor), req.CreditAccountID)
+		if _, balErr = tx.Exec("UPDATE ledger_accounts SET debits_pending = debits_pending + $1, updated_at = NOW() WHERE id = $2",
+			fmt.Sprintf("%d", amountMinor), req.DebitAccountID); balErr == nil {
+			_, balErr = tx.Exec("UPDATE ledger_accounts SET credits_pending = credits_pending + $1, updated_at = NOW() WHERE id = $2",
+				fmt.Sprintf("%d", amountMinor), req.CreditAccountID)
+		}
 	} else {
-		s.db.Exec("UPDATE ledger_accounts SET debits_posted = debits_posted + $1, updated_at = NOW() WHERE id = $2",
-			fmt.Sprintf("%d", amountMinor), req.DebitAccountID)
-		s.db.Exec("UPDATE ledger_accounts SET credits_posted = credits_posted + $1, updated_at = NOW() WHERE id = $2",
-			fmt.Sprintf("%d", amountMinor), req.CreditAccountID)
+		if _, balErr = tx.Exec("UPDATE ledger_accounts SET debits_posted = debits_posted + $1, updated_at = NOW() WHERE id = $2",
+			fmt.Sprintf("%d", amountMinor), req.DebitAccountID); balErr == nil {
+			_, balErr = tx.Exec("UPDATE ledger_accounts SET credits_posted = credits_posted + $1, updated_at = NOW() WHERE id = $2",
+				fmt.Sprintf("%d", amountMinor), req.CreditAccountID)
+		}
+	}
+	if balErr != nil {
+		log.Printf("[ledger-service] FAIL-CLOSED: balance update failed: %v", balErr)
+		w.WriteHeader(http.StatusInternalServerError)
+		json.NewEncoder(w).Encode(map[string]string{"error": "TRANSFER_FAILED", "message": "Failed to update balances — operation blocked"})
+		return
+	}
+
+	if err = tx.Commit(); err != nil {
+		log.Printf("[ledger-service] FAIL-CLOSED: commit failed: %v", err)
+		w.WriteHeader(http.StatusInternalServerError)
+		json.NewEncoder(w).Encode(map[string]string{"error": "TRANSFER_FAILED", "message": "Failed to commit transfer — operation blocked"})
+		return
 	}
 
 	// Emit Kafka event
 	s.publishEvent(status, transferID, map[string]interface{}{
-		"event":              "transfer_" + status,
+		"event":             "transfer_" + status,
 		"transfer_id":       transferID,
 		"debit_account_id":  req.DebitAccountID,
 		"credit_account_id": req.CreditAccountID,
@@ -538,6 +581,15 @@ func (s *LedgerService) handlePostPending(w http.ResponseWriter, r *http.Request
 			"pending_transfer_id": req.PendingTransferID,
 		})
 		return
+	} else if err != nil {
+		// Any other DB error must not fall through with empty values.
+		log.Printf("[ledger-service] FAIL-CLOSED: post-pending lookup failed: %v", err)
+		w.WriteHeader(http.StatusInternalServerError)
+		json.NewEncoder(w).Encode(map[string]string{
+			"error":   "LOOKUP_FAILED",
+			"message": "Failed to load pending transfer — operation blocked",
+		})
+		return
 	}
 
 	originalAmount, _ := strconv.ParseInt(amountStr, 10, 64)
@@ -574,11 +626,11 @@ func (s *LedgerService) handlePostPending(w http.ResponseWriter, r *http.Request
 
 	// Emit event
 	s.publishEvent("post_pending", postID, map[string]interface{}{
-		"event":       "transfer_post_pending",
-		"post_id":     postID,
-		"pending_id":  req.PendingTransferID,
-		"amount":      minorToAmount(postAmount),
-		"timestamp":   time.Now().UTC().Format(time.RFC3339),
+		"event":      "transfer_post_pending",
+		"post_id":    postID,
+		"pending_id": req.PendingTransferID,
+		"amount":     minorToAmount(postAmount),
+		"timestamp":  time.Now().UTC().Format(time.RFC3339),
 	})
 
 	log.Printf("[TigerBeetle] Posted pending transfer: %s -> %s", req.PendingTransferID, postID)
@@ -614,6 +666,15 @@ func (s *LedgerService) handleVoidPending(w http.ResponseWriter, r *http.Request
 			"pending_transfer_id": req.PendingTransferID,
 		})
 		return
+	} else if err != nil {
+		// Any other DB error must not fall through with empty values.
+		log.Printf("[ledger-service] FAIL-CLOSED: void-pending lookup failed: %v", err)
+		w.WriteHeader(http.StatusInternalServerError)
+		json.NewEncoder(w).Encode(map[string]string{
+			"error":   "LOOKUP_FAILED",
+			"message": "Failed to load pending transfer — operation blocked",
+		})
+		return
 	}
 
 	amount, _ := strconv.ParseInt(amountStr, 10, 64)
@@ -637,9 +698,9 @@ func (s *LedgerService) handleVoidPending(w http.ResponseWriter, r *http.Request
 
 	// Emit event
 	s.publishEvent("void_pending", voidID, map[string]interface{}{
-		"event":          "transfer_void_pending",
-		"void_id":        voidID,
-		"pending_id":     req.PendingTransferID,
+		"event":           "transfer_void_pending",
+		"void_id":         voidID,
+		"pending_id":      req.PendingTransferID,
 		"amount_released": minorToAmount(amount),
 		"timestamp":       time.Now().UTC().Format(time.RFC3339),
 	})
@@ -704,13 +765,13 @@ func (s *LedgerService) handleStats(w http.ResponseWriter, r *http.Request) {
 
 	w.Header().Set("Content-Type", "application/json")
 	json.NewEncoder(w).Encode(map[string]interface{}{
-		"total_accounts":     accountCount,
-		"total_transfers":    transferCount,
-		"pending_transfers":  pendingCount,
-		"total_volume_usd":   minorToAmount(totalVol),
-		"double_entry":       true,
-		"fail_closed":        s.isProduction,
-		"two_phase_enabled":  true,
+		"total_accounts":    accountCount,
+		"total_transfers":   transferCount,
+		"pending_transfers": pendingCount,
+		"total_volume_usd":  minorToAmount(totalVol),
+		"double_entry":      true,
+		"fail_closed":       s.isProduction,
+		"two_phase_enabled": true,
 		"middleware": map[string]bool{
 			"kafka":    true,
 			"dapr":     true,
@@ -734,10 +795,10 @@ func (s *LedgerService) handleReconciliation(w http.ResponseWriter, r *http.Requ
 
 	w.Header().Set("Content-Type", "application/json")
 	json.NewEncoder(w).Encode(map[string]interface{}{
-		"unpublished_events":       unpublished,
-		"stale_pending_transfers":  stalePending,
-		"recommendation":           recommendation,
-		"timestamp":                time.Now().UTC().Format(time.RFC3339),
+		"unpublished_events":      unpublished,
+		"stale_pending_transfers": stalePending,
+		"recommendation":          recommendation,
+		"timestamp":               time.Now().UTC().Format(time.RFC3339),
 	})
 }