fix(wallet): fund-leak on lead rejection, broken expiry refund, dead admin_adjust

Three independent Tracecoin correctness bugs found auditing the
services-marketplace lead flow:

1. reject_request passed lead.user_role_profile_id -- the wrong id --
   to try_release_reserved_tracecoins, which looks the wallet up by
   user_id. It never found a matching wallet, silently returned
   Ok(false), and the handler returned 409 to the customer -- but
   lead_requests.status had already committed to REJECTED on the
   previous statement. Every rejected lead permanently stranded the
   professional's reserved Tracecoins, with no cron sweeping REJECTED
   leads to recover them. Now uses professional_user_id, matching
   approve_request's (correct) debit call.

2. apps/cron/src/tasks/leads.rs's stale-PENDING-lead refund wrote
   `UPDATE tracecoin_wallets SET current_balance = current_balance +
   $1` -- that column doesn't exist (it's `balance`), so this errored
   on every run that hit a reserved-coins expiry, aborting the whole
   function via `?` before the transaction committed (rolling back the
   EXPIRED status flip with it). Leads with reserved coins never
   actually expired or refunded, silently, every 15 minutes. Fixed the
   column name and made it also decrement `reserved` (previously it
   only ever credited `balance`, never releasing the hold itself).

3. crates/wallet (credit/reserve/release/confirm/admin_adjust) and its
   hold/escrow submodule were built against a tracecoin_ledger schema
   (`type`, `reason`, `balance_after`, `actor_user_id`, `metadata`
   columns, a `lock_tracecoin_wallet()` SQL function) that was never
   actually migrated -- the migration that would have added it was
   disabled (`.up.sql.skip`), and even that skipped file assumed
   column names that don't match this repo's real, separately-evolved
   ledger schema (`transaction_type`/`reference_type`). Every call into
   crates/wallet errored before doing anything. The one live caller is
   the admin manual wallet-adjustment endpoint (apps/payments/src/
   admin.rs), so every admin credit/debit adjustment 500'd. Migration
   adds the function plus the genuinely-new audit columns
   (balance_after/actor_user_id/metadata) and an idempotency index,
   while the code now reads/writes through the existing transaction_
   type/reference_type columns instead of introducing a second,
   competing pair.

Also fixes the underlying check-then-act race on lead_requests.status:
approve_request/reject_request read status, branched in Rust, then
called an unconditional UPDATE with no re-check -- concurrent
approve/reject on the same lead could both pass the in-memory check.
LeadRequestRepository::update_status_from adds a `WHERE status =
$from` guard, used both for the initial PENDING transition and to
compensate (revert to PENDING) if the wallet debit/release fails after
the status flip already committed.

hold::place (the escrow "reserve" half) has zero callers anywhere in
the codebase -- flagging separately, not fixed here, since wiring it
up or removing it is a product decision, not a bug fix.

apps/cron/src/tasks/leads.rs

@@ -61,8 +61,20 @@ pub async fn expire_stale_lead_requests(
         }
 
         if rec.tracecoins_reserved > 0 {
+            // Was `current_balance = current_balance + $1` -- that column
+            // doesn't exist on tracecoin_wallets (it's `balance`), so this
+            // UPDATE errored on every run that hit a reserved-coins expiry,
+            // aborting the whole function via `?` before the transaction
+            // committed (the EXPIRED status flip above rolled back with
+            // it). Net effect: leads with reserved coins never actually
+            // expired, and their coins were never refunded -- silently,
+            // forever, every 15 minutes. Also restores `reserved`, not just
+            // `balance` -- the reservation incremented both `reserved` (the
+            // held amount) and left `balance` unchanged, so releasing it
+            // must symmetrically decrement `reserved`, or the coins would
+            // count as both refunded to balance AND still held in reserved.
             sqlx::query(
-                "UPDATE tracecoin_wallets SET current_balance = current_balance + $1, updated_at = NOW() WHERE user_id = $2"
+                "UPDATE tracecoin_wallets SET balance = balance + $1, reserved = reserved - $1, updated_at = NOW() WHERE user_id = $2"
             )
             .bind(rec.tracecoins_reserved)
             .bind(rec.user_id)

apps/customers/src/handlers.rs

@@ -385,35 +385,42 @@ async fn approve_request(
         _ => return (StatusCode::NOT_FOUND, "Lead request not found").into_response(),
     };
 
-    if lead.status != "PENDING" {
-        return (StatusCode::BAD_REQUEST, "Lead already resolved").into_response();
-    }
-
-    match LeadRequestRepository::update_status(&state.pool, lead.id, "ACCEPTED").await {
-        Ok(updated) => {
-            match TracecoinWalletRepository::try_debit_reserved_tracecoins(
-                &state.pool,
-                lead.professional_user_id.unwrap(),
-                lead.tracecoins_reserved,
-                lead.id,
-            ).await {
-                Ok(true) => {}
-                Ok(false) => return (StatusCode::CONFLICT, "Reserved Tracecoins unavailable").into_response(),
-                Err(e) => {
-                    tracing::error!("approve_request debit error: {}", e);
-                    return (StatusCode::INTERNAL_SERVER_ERROR, "Internal server error").into_response();
-                }
-            }
-
-            (StatusCode::OK, Json(serde_json::json!({
-                "lead_request": updated,
-            }))).into_response()
-        },
+    // update_status_from's WHERE status = 'PENDING' guard makes this
+    // atomic against a concurrent approve/reject on the same lead --
+    // previously this was a plain SELECT-then-branch-then-UPDATE with no
+    // re-check, so two racing requests could both pass the `lead.status !=
+    // "PENDING"` check above before either UPDATE committed. See its
+    // doc-comment for the full reasoning.
+    let updated = match LeadRequestRepository::update_status_from(&state.pool, lead.id, "PENDING", "ACCEPTED").await {
+        Ok(Some(updated)) => updated,
+        Ok(None) => return (StatusCode::BAD_REQUEST, "Lead already resolved").into_response(),
         Err(e) => {
             tracing::error!("approve_request update_status error: {}", e);
-            (StatusCode::INTERNAL_SERVER_ERROR, "Internal server error").into_response()
+            return (StatusCode::INTERNAL_SERVER_ERROR, "Internal server error").into_response();
+        }
+    };
+
+    match TracecoinWalletRepository::try_debit_reserved_tracecoins(
+        &state.pool,
+        lead.professional_user_id.unwrap(),
+        lead.tracecoins_reserved,
+        lead.id,
+    ).await {
+        Ok(true) => {}
+        Ok(false) | Err(_) => {
+            // The debit failed after the status flip already committed --
+            // revert PENDING so the lead isn't left stuck ACCEPTED with its
+            // Tracecoins never actually debited from reserved.
+            if let Err(e) = LeadRequestRepository::update_status_from(&state.pool, lead.id, "ACCEPTED", "PENDING").await {
+                tracing::error!("approve_request failed to revert status after debit failure for lead {}: {}", lead.id, e);
+            }
+            return (StatusCode::CONFLICT, "Reserved Tracecoins unavailable").into_response();
         }
     }
+
+    (StatusCode::OK, Json(serde_json::json!({
+        "lead_request": updated,
+    }))).into_response()
 }
 
 async fn reject_request(
@@ -447,32 +454,40 @@ async fn reject_request(
         _ => return (StatusCode::NOT_FOUND, "Lead request not found").into_response(),
     };
 
-    if lead.status != "PENDING" {
-        return (StatusCode::BAD_REQUEST, "Lead already resolved").into_response();
-    }
-
-    match LeadRequestRepository::update_status(&state.pool, lead.id, "REJECTED").await {
-        Ok(updated) => {
-            match TracecoinWalletRepository::try_release_reserved_tracecoins(
-                &state.pool,
-                lead.user_role_profile_id.unwrap(),
-                lead.tracecoins_reserved,
-                lead.id,
-                "LEAD_REJECTED",
-            ).await {
-                Ok(true) => {}
-                Ok(false) => return (StatusCode::CONFLICT, "Reserved Tracecoins unavailable").into_response(),
-                Err(e) => {
-                    tracing::error!("reject_request release error: {}", e);
-                    return (StatusCode::INTERNAL_SERVER_ERROR, "Internal server error").into_response();
-                }
-            }
-
-            (StatusCode::OK, Json(updated)).into_response()
-        },
+    let updated = match LeadRequestRepository::update_status_from(&state.pool, lead.id, "PENDING", "REJECTED").await {
+        Ok(Some(updated)) => updated,
+        Ok(None) => return (StatusCode::BAD_REQUEST, "Lead already resolved").into_response(),
         Err(e) => {
             tracing::error!("reject_request update_status error: {}", e);
-            (StatusCode::INTERNAL_SERVER_ERROR, "Internal server error").into_response()
+            return (StatusCode::INTERNAL_SERVER_ERROR, "Internal server error").into_response();
+        }
+    };
+
+    match TracecoinWalletRepository::try_release_reserved_tracecoins(
+        &state.pool,
+        // Was lead.user_role_profile_id -- the wrong id. The wallet is
+        // keyed by user_id, and try_release_reserved_tracecoins looks it up
+        // by user_id, so passing the role-profile id meant this never found
+        // a matching wallet: it silently returned Ok(false), the caller
+        // returned 409 to the customer, but the status flip to REJECTED
+        // above had already committed -- permanently stranding the
+        // professional's reserved Tracecoins with nothing that ever swept
+        // REJECTED leads to release them. approve_request's debit call
+        // (above) already used the correct professional_user_id; this now
+        // matches it.
+        lead.professional_user_id.unwrap(),
+        lead.tracecoins_reserved,
+        lead.id,
+        "LEAD_REJECTED",
+    ).await {
+        Ok(true) => {}
+        Ok(false) | Err(_) => {
+            if let Err(e) = LeadRequestRepository::update_status_from(&state.pool, lead.id, "REJECTED", "PENDING").await {
+                tracing::error!("reject_request failed to revert status after release failure for lead {}: {}", lead.id, e);
+            }
+            return (StatusCode::CONFLICT, "Reserved Tracecoins unavailable").into_response();
         }
     }
+
+    (StatusCode::OK, Json(updated)).into_response()
 }

crates/db/migrations/20260818000002_tracecoin_ledger_wallet_crate_columns.down.sql

@@ -0,0 +1,11 @@
+BEGIN;
+
+DROP FUNCTION IF EXISTS lock_tracecoin_wallet(UUID);
+DROP INDEX IF EXISTS uq_tracecoin_ledger_wallet_reference_type;
+
+ALTER TABLE tracecoin_ledger
+    DROP COLUMN IF EXISTS balance_after,
+    DROP COLUMN IF EXISTS actor_user_id,
+    DROP COLUMN IF EXISTS metadata;
+
+COMMIT;

crates/db/migrations/20260818000002_tracecoin_ledger_wallet_crate_columns.up.sql

@@ -0,0 +1,65 @@
+-- crates/wallet (lib.rs + hold.rs) is a second, more sophisticated
+-- Tracecoin implementation (idempotent-by-reference credit/reserve/
+-- release/confirm/admin_adjust, plus a hold/escrow subsystem) that was
+-- built against a `tracecoin_ledger` schema with `type`/`reason`/
+-- `balance_after`/`actor_user_id`/`metadata` columns, and a
+-- `lock_tracecoin_wallet(uuid)` helper function. Neither the columns nor
+-- the function were ever actually created: the migration that would have
+-- added them (20260627000000_tracecoin_security_hardening.up.sql.skip) was
+-- disabled, and itself assumed columns (`type`, `reason`) that don't match
+-- this repo's actual, separately-evolved ledger schema (`transaction_type`,
+-- `reference_type` -- see 20260318233000_tracecoin_ledger_immutable.up.sql).
+--
+-- Net effect: every call into crates/wallet errors at the DB layer before
+-- it can do anything (`function lock_tracecoin_wallet(uuid) does not
+-- exist`). The one live, reachable caller is the admin manual wallet
+-- adjustment endpoint (apps/payments/src/admin.rs -> wallet::admin_adjust),
+-- so every admin credit/debit adjustment 500s today. This migration adds
+-- what crates/wallet actually needs, reusing the existing
+-- transaction_type/reference_type columns (via the handler-side fix that
+-- accompanies this migration) instead of introducing a second, competing
+-- set of `type`/`reason` columns -- balance_after/actor_user_id/metadata
+-- are genuinely new, additive audit fields with no existing equivalent.
+BEGIN;
+
+ALTER TABLE tracecoin_ledger
+    ADD COLUMN IF NOT EXISTS balance_after INTEGER,
+    ADD COLUMN IF NOT EXISTS actor_user_id UUID REFERENCES users(id),
+    ADD COLUMN IF NOT EXISTS metadata JSONB;
+
+-- Idempotency: at most one ledger entry per (wallet, reference,
+-- transaction_type). Lets a single payment/reservation/admin-action credit
+-- or debit exactly once on retry, while still allowing separate RESERVE /
+-- RELEASE / DEBIT rows for the same reference_id (e.g. one lead_request).
+CREATE UNIQUE INDEX IF NOT EXISTS uq_tracecoin_ledger_wallet_reference_type
+    ON tracecoin_ledger (wallet_id, reference_id, transaction_type)
+    WHERE reference_id IS NOT NULL;
+
+-- The single point where concurrent access to a wallet is serialized:
+-- get-or-create the wallet row, lock it FOR UPDATE for the rest of the
+-- caller's transaction, return its current state. Mirrors the inline
+-- SELECT...FOR UPDATE pattern crates/db/src/models/tracecoin_wallet.rs
+-- already uses correctly elsewhere in this codebase, just as a reusable
+-- function for crates/wallet's callers.
+CREATE OR REPLACE FUNCTION lock_tracecoin_wallet(p_user_id UUID)
+RETURNS TABLE(wallet_id UUID, balance INTEGER, reserved INTEGER)
+LANGUAGE plpgsql
+AS $$
+DECLARE
+    v_id UUID;
+BEGIN
+    INSERT INTO tracecoin_wallets (user_id, balance, reserved)
+    VALUES (p_user_id, 0, 0)
+    ON CONFLICT (user_id) DO NOTHING;
+
+    SELECT id INTO v_id FROM tracecoin_wallets WHERE user_id = p_user_id;
+
+    RETURN QUERY
+        SELECT tw.id, tw.balance, tw.reserved
+        FROM tracecoin_wallets tw
+        WHERE tw.id = v_id
+        FOR UPDATE;
+END;
+$$;
+
+COMMIT;

crates/db/src/models/lead_request.rs

@@ -123,4 +123,43 @@ impl LeadRequestRepository {
 
         Ok(req)
     }
+
+    /// Transitions status only if the row is currently `from_status`,
+    /// atomically -- the `WHERE status = $3` makes this safe to call from
+    /// two concurrent requests racing the same lead_request (e.g. approve
+    /// racing reject, or a double-click on approve): only one call's
+    /// `UPDATE` matches a row and gets `Some(..)` back, the other gets
+    /// `None` because by the time its `UPDATE` runs the row's status has
+    /// already moved. Callers previously read status in a separate
+    /// `SELECT`, branched in Rust, then called the unconditional
+    /// `update_status` above with no re-check -- the same check-then-act
+    /// shape as the job_applications duplicate-application bug.
+    ///
+    /// Also used to compensate: if a wallet debit/release fails after this
+    /// transitions PENDING -> ACCEPTED/REJECTED, callers revert with
+    /// `update_status_from(pool, id, "ACCEPTED", "PENDING")` so a failed
+    /// debit doesn't leave the lead stuck resolved with no Tracecoins
+    /// actually moved.
+    pub async fn update_status_from(
+        pool: &PgPool,
+        id: Uuid,
+        from_status: &str,
+        to_status: &str,
+    ) -> Result<Option<LeadRequest>, sqlx::Error> {
+        let req = sqlx::query_as::<_, LeadRequest>(
+            r#"
+            UPDATE lead_requests
+            SET status = $1, resolved_at = NOW(), updated_at = NOW()
+            WHERE id = $2 AND status = $3
+            RETURNING *
+            "#,
+        )
+        .bind(to_status)
+        .bind(id)
+        .bind(from_status)
+        .fetch_optional(pool)
+        .await?;
+
+        Ok(req)
+    }
 }

crates/wallet/src/hold.rs

@@ -283,7 +283,7 @@ pub async fn settle(
     let ledger_id: Uuid = sqlx::query_scalar(
         r#"
         INSERT INTO tracecoin_ledger (
-            wallet_id, type, amount, balance_after, reason,
+            wallet_id, transaction_type, amount, balance_after, reference_type,
             reference_id, actor_user_id, metadata
         )
         VALUES ($1, 'DEBIT', $2, $3, 'HOLD_SETTLED', $4, NULL, $5)

crates/wallet/src/lib.rs

@@ -288,16 +288,16 @@ async fn write_ledger_entry(
             r#"
             SELECT
                 id, wallet_id,
-                type AS entry_type,
+                transaction_type AS entry_type,
                 amount,
                 balance_after,
-                reason,
+                reference_type AS reason,
                 reference_id,
                 actor_user_id,
                 metadata,
                 created_at
             FROM tracecoin_ledger
-            WHERE wallet_id = $1 AND reference_id = $2 AND type = $3
+            WHERE wallet_id = $1 AND reference_id = $2 AND transaction_type = $3
             "#,
         )
         .bind(wallet_id)
@@ -314,16 +314,16 @@ async fn write_ledger_entry(
     let row = sqlx::query(
         r#"
         INSERT INTO tracecoin_ledger (
-            wallet_id, type, amount, balance_after, reason,
+            wallet_id, transaction_type, amount, balance_after, reference_type,
             reference_id, actor_user_id, metadata
         )
         VALUES ($1, $2, $3, $4, $5, $6, $7, $8)
         RETURNING
             id, wallet_id,
-            type AS entry_type,
+            transaction_type AS entry_type,
             amount,
             balance_after,
-            reason,
+            reference_type AS reason,
             reference_id,
             actor_user_id,
             metadata,
@@ -382,16 +382,16 @@ pub async fn credit(
             r#"
             SELECT
                 id, wallet_id,
-                type AS entry_type,
+                transaction_type AS entry_type,
                 amount,
                 balance_after,
-                reason,
+                reference_type AS reason,
                 reference_id,
                 actor_user_id,
                 metadata,
                 created_at
             FROM tracecoin_ledger
-            WHERE wallet_id = $1 AND reference_id = $2 AND type = 'CREDIT'
+            WHERE wallet_id = $1 AND reference_id = $2 AND transaction_type = 'CREDIT'
             "#,
         )
         .bind(wallet_id)
@@ -466,16 +466,16 @@ pub async fn reserve(
         r#"
         SELECT
             id, wallet_id,
-            type AS entry_type,
+            transaction_type AS entry_type,
             amount,
             balance_after,
-            reason,
+            reference_type AS reason,
             reference_id,
             actor_user_id,
             metadata,
             created_at
         FROM tracecoin_ledger
-        WHERE wallet_id = $1 AND reference_id = $2 AND type = 'RESERVE'
+        WHERE wallet_id = $1 AND reference_id = $2 AND transaction_type = 'RESERVE'
         "#,
     )
     .bind(wallet_id)
@@ -551,16 +551,16 @@ pub async fn release(
         r#"
         SELECT
             id, wallet_id,
-            type AS entry_type,
+            transaction_type AS entry_type,
             amount,
             balance_after,
-            reason,
+            reference_type AS reason,
             reference_id,
             actor_user_id,
             metadata,
             created_at
         FROM tracecoin_ledger
-        WHERE wallet_id = $1 AND reference_id = $2 AND type = 'RELEASE'
+        WHERE wallet_id = $1 AND reference_id = $2 AND transaction_type = 'RELEASE'
         "#,
     )
     .bind(wallet_id)
@@ -632,16 +632,16 @@ pub async fn confirm(
         r#"
         SELECT
             id, wallet_id,
-            type AS entry_type,
+            transaction_type AS entry_type,
             amount,
             balance_after,
-            reason,
+            reference_type AS reason,
             reference_id,
             actor_user_id,
             metadata,
             created_at
         FROM tracecoin_ledger
-        WHERE wallet_id = $1 AND reference_id = $2 AND type = 'DEBIT'
+        WHERE wallet_id = $1 AND reference_id = $2 AND transaction_type = 'DEBIT'
         "#,
     )
     .bind(wallet_id)
@@ -771,10 +771,10 @@ pub async fn list_ledger(
         r#"
         SELECT
             tl.id, tl.wallet_id,
-            tl.type AS entry_type,
+            tl.transaction_type AS entry_type,
             tl.amount,
             tl.balance_after,
-            tl.reason,
+            tl.reference_type AS reason,
             tl.reference_id,
             tl.actor_user_id,
             tl.metadata,