diff --git a/server/_core/transferPipeline.ts b/server/_core/transferPipeline.ts index f488f342..ffa2c964 100644 --- 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; } @@ -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 index f63b3d34..e86209c1 100644 --- 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 index 2696d555..594bc89e 100644 --- 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 new file mode 100644 index 00000000..9e3584b4 --- /dev/null +++ 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 index d0a2e0d4..92d1a3e5 100644 --- 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), }) } diff --git a/services/python-tigerbeetle-reconciliation/main.py b/services/python-tigerbeetle-reconciliation/main.py index 20214b8c..e8bb31c7 100644 --- a/services/python-tigerbeetle-reconciliation/main.py +++ b/services/python-tigerbeetle-reconciliation/main.py @@ -152,6 +152,16 @@ class Discrepancy: # ─── TigerBeetle Service Client ────────────────────────────────────────────── +def _to_minor(value) -> int: + """Convert a float major-unit balance to integer minor units (10^6). + + Uses round() rather than int() truncation: the ledger services return + balances as floats, and e.g. 0.07 * 1_000_000 == 69999.99999999999, so a + plain int() cast would truncate to 69999 and manufacture a phantom 1-unit + drift (or mask a real one) during reconciliation. + """ + return int(round(float(value or 0) * 1_000_000)) + def fetch_tb_account_balance(account_id: str) -> Optional[AccountBalance]: """Fetch account balance from Rust TigerBeetle service.""" try: @@ -165,10 +175,10 @@ def fetch_tb_account_balance(account_id: str) -> Optional[AccountBalance]: return None return AccountBalance( account_id=account_id, - debits_pending=int(float(data.get("debits_pending", 0)) * 1_000_000), - debits_posted=int(float(data.get("debits_posted", 0)) * 1_000_000), - credits_pending=int(float(data.get("credits_pending", 0)) * 1_000_000), - credits_posted=int(float(data.get("credits_posted", 0)) * 1_000_000), + debits_pending=_to_minor(data.get("debits_pending", 0)), + debits_posted=_to_minor(data.get("debits_posted", 0)), + credits_pending=_to_minor(data.get("credits_pending", 0)), + credits_posted=_to_minor(data.get("credits_posted", 0)), ) except Exception as e: logger.warning(f"Failed to fetch TB balance for {account_id}: {e}") @@ -187,10 +197,10 @@ def fetch_go_account_balance(account_id: str) -> Optional[AccountBalance]: return None return AccountBalance( account_id=account_id, - debits_pending=int(float(data.get("debits_pending", 0)) * 1_000_000), - debits_posted=int(float(data.get("debits_posted", 0)) * 1_000_000), - credits_pending=int(float(data.get("credits_pending", 0)) * 1_000_000), - credits_posted=int(float(data.get("credits_posted", 0)) * 1_000_000), + debits_pending=_to_minor(data.get("debits_pending", 0)), + debits_posted=_to_minor(data.get("debits_posted", 0)), + credits_pending=_to_minor(data.get("credits_pending", 0)), + credits_posted=_to_minor(data.get("credits_posted", 0)), ) except Exception as e: logger.warning(f"Failed to fetch Go ledger balance for {account_id}: {e}") diff --git a/services/rust-tigerbeetle-service/src/main.rs b/services/rust-tigerbeetle-service/src/main.rs index 33837290..08d72408 100644 --- a/services/rust-tigerbeetle-service/src/main.rs +++ b/services/rust-tigerbeetle-service/src/main.rs @@ -332,6 +332,7 @@ async fn db_record_transfer(pool: &PgPool, id: &str, debit_id: &str, credit_id: Ok(()) } +#[allow(dead_code)] async fn db_update_balances(pool: &PgPool, debit_id: &str, credit_id: &str, amount: u128, is_pending: bool) -> Result<(), sqlx::Error> { if is_pending { sqlx::query("UPDATE tb_accounts SET debits_pending = debits_pending + $1, updated_at = NOW() WHERE id = $2") @@ -347,6 +348,37 @@ async fn db_update_balances(pool: &PgPool, debit_id: &str, credit_id: &str, amou Ok(()) } +/// Records a transfer and updates both account balances atomically in a single +/// DB transaction, so a failure between the insert and the balance updates can +/// never leave the ledger in an inconsistent state. +async fn db_record_transfer_atomic(pool: &PgPool, id: &str, debit_id: &str, credit_id: &str, amount: u128, ledger: u32, code: u16, flags: u16, status: &str, idempotency_key: Option<&str>, is_pending: bool) -> Result<(), sqlx::Error> { + let mut tx = pool.begin().await?; + sqlx::query( + "INSERT INTO tb_transfers (id, debit_account_id, credit_account_id, amount, pending_id, ledger, code, flags, status, idempotency_key) + VALUES ($1, $2, $3, $4, NULL, $5, $6, $7, $8, $9) + ON CONFLICT (id) DO UPDATE SET status = $8" + ) + .bind(id).bind(debit_id).bind(credit_id) + .bind(amount.to_string()) + .bind(ledger as i32).bind(code as i16).bind(flags as i16) + .bind(status).bind(idempotency_key) + .execute(&mut *tx).await?; + + if is_pending { + sqlx::query("UPDATE tb_accounts SET debits_pending = debits_pending + $1, updated_at = NOW() WHERE id = $2") + .bind(amount.to_string()).bind(debit_id).execute(&mut *tx).await?; + sqlx::query("UPDATE tb_accounts SET credits_pending = credits_pending + $1, updated_at = NOW() WHERE id = $2") + .bind(amount.to_string()).bind(credit_id).execute(&mut *tx).await?; + } else { + sqlx::query("UPDATE tb_accounts SET debits_posted = debits_posted + $1, updated_at = NOW() WHERE id = $2") + .bind(amount.to_string()).bind(debit_id).execute(&mut *tx).await?; + sqlx::query("UPDATE tb_accounts SET credits_posted = credits_posted + $1, updated_at = NOW() WHERE id = $2") + .bind(amount.to_string()).bind(credit_id).execute(&mut *tx).await?; + } + tx.commit().await?; + Ok(()) +} + async fn db_emit_event(pool: &PgPool, event_type: &str, transfer_id: &str, payload: &serde_json::Value) -> Result<(), sqlx::Error> { sqlx::query( "INSERT INTO tb_events (event_type, transfer_id, payload) VALUES ($1, $2, $3)" @@ -478,6 +510,16 @@ async fn initiate_transfer( State(state): State>, Json(req): Json, ) -> (StatusCode, Json) { + // Reject non-positive / non-finite amounts. A negative f64 cast to u128 wraps + // to a huge value and a negative amount would also bypass the balance pre-check. + if !req.amount.is_finite() || req.amount <= 0.0 { + return (StatusCode::BAD_REQUEST, Json(serde_json::json!({ + "error": "INVALID_AMOUNT", + "amount": req.amount, + "message": "Transfer amount must be greater than zero" + }))); + } + let debit_id = req.debit_account_id.clone(); let credit_id = req.credit_account_id.clone(); let amount_minor = amount_to_minor(req.amount); @@ -531,11 +573,11 @@ async fn initiate_transfer( let flags = if is_pending { transfer_flags::PENDING } else { transfer_flags::NONE }; let status = if is_pending { "pending" } else { "posted" }; - // Record transfer in PostgreSQL - if let Err(e) = db_record_transfer( + // Record transfer and update balances atomically in PostgreSQL (FAIL-CLOSED) + if let Err(e) = db_record_transfer_atomic( &state.db_pool, &transfer_id_str, &debit_id, &credit_id, - amount_minor, None, 1, transfer_code, flags, status, - req.idempotency_key.as_deref(), + amount_minor, 1, transfer_code, flags, status, + req.idempotency_key.as_deref(), is_pending, ).await { error!(err = %e, "[TigerBeetle] FAIL-CLOSED: Transfer persistence failed"); return (StatusCode::INTERNAL_SERVER_ERROR, Json(serde_json::json!({ @@ -544,11 +586,6 @@ async fn initiate_transfer( }))); } - // Update account balances - if let Err(e) = db_update_balances(&state.db_pool, &debit_id, &credit_id, amount_minor, is_pending).await { - error!(err = %e, "[TigerBeetle] Balance update failed"); - } - // Emit Kafka event let event = serde_json::json!({ "event": if is_pending { "transfer_pending" } else { "transfer_posted" },