feat(bridge): promote topup deposit audit row to Completed on crypto receive

src/services/frappe/BridgeTransferRequestWriter.ts

@@ -51,6 +51,29 @@ const upsert = async (
   return ErpNext.upsertBridgeTransferRequest(request)
 }
 
+// The IBEX crypto receive settle row (`ibex:<txHash>`), when it already
+// exists at Settled — i.e. the crypto side of this topup landed before this
+// deposit event. Carries the credited account/wallet so the deposit row gets
+// the same attribution the promotion path stamps. A lookup failure degrades
+// to undefined so the deposit audit write never fails on the enrichment; the
+// row just stays at Fiat Received until the crypto-receive handler (or a
+// Bridge retry) promotes it.
+const findSettledIbexReceive = async (
+  txHash: string,
+): Promise<{ accountId?: string; walletId?: string } | undefined> => {
+  if (!ErpNext?.findBridgeTransferRequest) return undefined
+  const doc = await ErpNext.findBridgeTransferRequest(`ibex:${txHash}`)
+  if (doc instanceof Error) {
+    baseLogger.warn(
+      { txHash, error: doc },
+      "Failed to check IBEX settle row for Bridge deposit; keeping Fiat Received",
+    )
+    return undefined
+  }
+  if (!doc || doc.status !== BridgeTransferRequestStatus.Settled) return undefined
+  return { accountId: doc.account_id, walletId: doc.wallet_id }
+}
+
 export const writeBridgeDepositRequest = async ({
   eventId,
   eventObject,
@@ -87,11 +110,20 @@ export const writeBridgeDepositRequest = async ({
     return true
   }
 
+  const destinationTxHash = receipt?.destination_tx_hash
+  const settledReceive = destinationTxHash
+    ? await findSettledIbexReceive(destinationTxHash)
+    : undefined
+
   return upsert(
     new BridgeTransferRequest({
       requestId: stableRequestId,
       transactionType: BridgeTransferRequestTransactionType.Topup,
-      status: BridgeTransferRequestStatus.FiatReceived,
+      status: settledReceive
+        ? BridgeTransferRequestStatus.Completed
+        : BridgeTransferRequestStatus.FiatReceived,
+      accountId: settledReceive?.accountId,
+      walletId: settledReceive?.walletId,
       amount: String(eventObject.amount),
       currency: String(currency),
       developerFee:
@@ -106,12 +138,34 @@ export const writeBridgeDepositRequest = async ({
       ibexTxHash: receipt?.destination_tx_hash,
       sourceEventId: eventId,
       sourceEventType: `deposit.${state}`,
-      sourceSystemsSeen: ["bridge_deposit"],
+      sourceSystemsSeen: settledReceive
+        ? ["bridge_deposit", "ibex_crypto_receive"]
+        : ["bridge_deposit"],
       rawPayload,
     }),
   )
 }
 
+// Called by the IBEX crypto-receive handler after it writes the settle row:
+// promotes the matching deposit-side Topup row (joined on ibex_tx_hash) to
+// Completed and stamps the credited account/wallet on it.
+export const promoteBridgeDepositForCryptoReceive = async ({
+  txHash,
+  accountId,
+  walletId,
+}: {
+  txHash: string
+  accountId: AccountId
+  walletId: WalletId
+}): Promise<
+  "completed" | "already_completed" | "not_found" | BridgeTransferRequestUpsertError
+> => {
+  if (!ErpNext?.completeBridgeTopupByTxHash) {
+    return new BridgeTransferRequestUpsertError("ERPNext client is not configured")
+  }
+  return ErpNext.completeBridgeTopupByTxHash({ txHash, accountId, walletId })
+}
+
 export const writeIbexCryptoReceiveRequest = async ({
   txHash,
   address,

src/services/frappe/ErpNext.ts

@@ -26,7 +26,12 @@ import {
   BankAccountUpdateRequest,
   ErpNextBankAccountUpdateRequestData,
 } from "./models/BankAccountUpdateRequest"
-import { BridgeTransferRequest } from "./models/BridgeTransferRequest"
+import {
+  BridgeTransferRequest,
+  BridgeTransferRequestStatus,
+  BridgeTransferRequestTransactionType,
+  toFrappeDatetime,
+} from "./models/BridgeTransferRequest"
 import { Filter } from "./SearchFilters"
 
 export type AccountUpgradeRequestFilters = { username?: Filter; status?: Filter }
@@ -48,6 +53,40 @@ export const toJson = (filters: AccountUpgradeRequestFilters): string => {
 
 export type CashoutId = string & { readonly brand: unique symbol }
 
+// Topup rows are written by two independent webhook streams (Bridge deposit
+// events and the IBEX crypto receive). Statuses must only ever move forward —
+// a Bridge retry of an early deposit event must not stomp a row that has
+// already been promoted past it.
+const BRIDGE_TRANSFER_STATUS_RANK: Record<string, number> = {
+  [BridgeTransferRequestStatus.Pending]: 0,
+  [BridgeTransferRequestStatus.FiatReceived]: 1,
+  [BridgeTransferRequestStatus.Settled]: 2,
+  [BridgeTransferRequestStatus.Completed]: 3,
+  [BridgeTransferRequestStatus.Failed]: 4,
+}
+
+const mergeSourceSystemsSeen = (
+  existing?: string,
+  incoming?: string,
+): string | undefined => {
+  const merged = [
+    ...new Set(
+      [...(existing?.split(",") ?? []), ...(incoming?.split(",") ?? [])]
+        .map((system) => system.trim())
+        .filter(Boolean),
+    ),
+  ]
+  return merged.length ? merged.join(",") : undefined
+}
+
+export type BridgeTransferRequestDoc = {
+  name: string
+  status?: string
+  source_systems_seen?: string
+  account_id?: string
+  wallet_id?: string
+}
+
 export class ErpNext {
   url: string
   headers: Record<string, string>
@@ -464,13 +503,13 @@ export class ErpNext {
     const requestId = payload.request_id
 
     try {
-      const existingName = await this.findBridgeTransferRequestName(requestId)
-      if (existingName instanceof BridgeTransferRequestUpsertError) return existingName
+      const existing = await this.findBridgeTransferRequest(requestId)
+      if (existing instanceof BridgeTransferRequestUpsertError) return existing
 
-      if (existingName) {
+      if (existing) {
         await axios.put(
-          `${this.url}/api/resource/${encodeURIComponent(BridgeTransferRequest.doctype)}/${encodeURIComponent(existingName)}`,
-          payload,
+          `${this.url}/api/resource/${encodeURIComponent(BridgeTransferRequest.doctype)}/${encodeURIComponent(existing.name)}`,
+          this.applyUpdateGuards(payload, existing),
           { headers: this.headers },
         )
         return true
@@ -486,13 +525,13 @@ export class ErpNext {
       } catch (err) {
         if (!this.isDuplicateRequestError(err)) throw err
 
-        const racedName = await this.findBridgeTransferRequestName(requestId)
-        if (racedName instanceof BridgeTransferRequestUpsertError) return racedName
-        if (!racedName) throw err
+        const raced = await this.findBridgeTransferRequest(requestId)
+        if (raced instanceof BridgeTransferRequestUpsertError) return raced
+        if (!raced) throw err
 
         await axios.put(
-          `${this.url}/api/resource/${encodeURIComponent(BridgeTransferRequest.doctype)}/${encodeURIComponent(racedName)}`,
-          payload,
+          `${this.url}/api/resource/${encodeURIComponent(BridgeTransferRequest.doctype)}/${encodeURIComponent(raced.name)}`,
+          this.applyUpdateGuards(payload, raced),
           { headers: this.headers },
         )
         return true
@@ -511,14 +550,116 @@ export class ErpNext {
     }
   }
 
-  private async findBridgeTransferRequestName(
+  // Promote the deposit-side Topup audit row (keyed by Bridge deposit id, not
+  // the `ibex:<txHash>` settle row) to Completed once the IBEX crypto receive
+  // has been observed for its destination tx hash. "not_found" is normal when
+  // the crypto receive lands before Bridge's payment_processed webhook has
+  // stamped the deposit row with the tx hash — the deposit writer covers that
+  // ordering by checking for the settle row itself.
+  async completeBridgeTopupByTxHash({
+    txHash,
+    accountId,
+    walletId,
+  }: {
+    txHash: string
+    accountId?: string
+    walletId?: string
+  }): Promise<
+    "completed" | "already_completed" | "not_found" | BridgeTransferRequestUpsertError
+  > {
+    try {
+      const filters = JSON.stringify([
+        [BridgeTransferRequest.doctype, "ibex_tx_hash", "=", txHash],
+        [
+          BridgeTransferRequest.doctype,
+          "transaction_type",
+          "=",
+          BridgeTransferRequestTransactionType.Topup,
+        ],
+        [BridgeTransferRequest.doctype, "request_id", "not like", "ibex:%"],
+      ])
+      const fields = JSON.stringify(["name", "status", "source_systems_seen"])
+      const resp = await axios.get(
+        `${this.url}/api/resource/${encodeURIComponent(BridgeTransferRequest.doctype)}`,
+        {
+          params: { filters, fields, limit_page_length: 1 },
+          headers: this.headers,
+        },
+      )
+
+      const doc: BridgeTransferRequestDoc | undefined = resp.data?.data?.[0]
+      if (!doc?.name) return "not_found"
+      if (doc.status === BridgeTransferRequestStatus.Completed) {
+        return "already_completed"
+      }
+
+      await axios.put(
+        `${this.url}/api/resource/${encodeURIComponent(BridgeTransferRequest.doctype)}/${encodeURIComponent(doc.name)}`,
+        {
+          status: BridgeTransferRequestStatus.Completed,
+          account_id: accountId,
+          wallet_id: walletId,
+          source_systems_seen: mergeSourceSystemsSeen(
+            doc.source_systems_seen,
+            "ibex_crypto_receive",
+          ),
+          last_seen_at: toFrappeDatetime(),
+        },
+        { headers: this.headers },
+      )
+      return "completed"
+    } catch (err) {
+      const responseData = isAxiosError(err) ? err.response?.data : undefined
+      baseLogger.error(
+        { err, responseData, txHash },
+        "Error promoting Bridge Transfer Request to Completed in ERPNext",
+      )
+      recordExceptionInCurrentSpan({
+        error: err,
+        attributes: { "erpnext.exception": responseData?.exception },
+      })
+      return new BridgeTransferRequestUpsertError(err)
+    }
+  }
+
+  private applyUpdateGuards(
+    payload: ReturnType<BridgeTransferRequest["toErpnext"]>,
+    existing: BridgeTransferRequestDoc,
+  ): ReturnType<BridgeTransferRequest["toErpnext"]> {
+    const guarded = {
+      ...payload,
+      source_systems_seen: mergeSourceSystemsSeen(
+        existing.source_systems_seen,
+        payload.source_systems_seen,
+      ),
+    }
+
+    if (
+      payload.transaction_type === BridgeTransferRequestTransactionType.Topup &&
+      existing.status &&
+      (BRIDGE_TRANSFER_STATUS_RANK[existing.status] ?? -1) >
+        (BRIDGE_TRANSFER_STATUS_RANK[payload.status] ?? -1)
+    ) {
+      guarded.status = existing.status as BridgeTransferRequestStatus
+    }
+
+    return guarded
+  }
+
+  async findBridgeTransferRequest(
     requestId: string,
-  ): Promise<string | undefined | BridgeTransferRequestUpsertError> {
+  ): Promise<BridgeTransferRequestDoc | undefined | BridgeTransferRequestUpsertError> {
     try {
       const filters = JSON.stringify([
         [BridgeTransferRequest.doctype, "request_id", "=", requestId],
       ])
-      const fields = JSON.stringify(["name"])
+      const fields = JSON.stringify([
+        "name",
+        "status",
+        "source_systems_seen",
+        "account_id",
+        "wallet_id",
+      ])
       const resp = await axios.get(
         `${this.url}/api/resource/${encodeURIComponent(BridgeTransferRequest.doctype)}`,
         {
@@ -527,7 +668,8 @@ export class ErpNext {
         },
       )
 
-      return resp.data?.data?.[0]?.name
+      const doc = resp.data?.data?.[0]
+      return doc?.name ? doc : undefined
     } catch (err) {
       const responseData = isAxiosError(err) ? err.response?.data : undefined
       baseLogger.error(

src/services/frappe/models/BridgeTransferRequest.ts

@@ -39,7 +39,7 @@ export type BridgeTransferRequestInput = {
   failureReason?: string
 }
 
-const toFrappeDatetime = (value?: string): string => {
+export const toFrappeDatetime = (value?: string): string => {
   const date = value ? new Date(value) : new Date()
 
   if (Number.isNaN(date.getTime())) return value ?? ""

src/services/ibex/webhook-server/routes/crypto-receive.ts

@@ -12,7 +12,10 @@ import {
   alertIbexCryptoReceiveFailure,
   alertIbexReconciliationFailed,
 } from "@services/alerts/ibex-bridge-movement"
-import { writeIbexCryptoReceiveRequest } from "@services/frappe/BridgeTransferRequestWriter"
+import {
+  promoteBridgeDepositForCryptoReceive,
+  writeIbexCryptoReceiveRequest,
+} from "@services/frappe/BridgeTransferRequestWriter"
 import { ibexWebhookPaths } from "@services/ibex/webhook-config"
 
 import { authenticate, logRequest, validateIbexIp } from "../middleware"
@@ -192,6 +195,34 @@ const cryptoReceiveHandler = async (req: Request, res: Response) => {
           return { status: "error", code: "erpnext_audit_failed" } as CryptoReceiveResult
         }
 
+        // Best-effort: the funds are already credited and the settle row is
+        // written, so a promotion failure must not fail the webhook. The
+        // deposit writer's reverse check (and Bridge webhook retries) will
+        // converge the row on a later event.
+        const promoteResult = await promoteBridgeDepositForCryptoReceive({
+          txHash: String(tx_hash),
+          accountId: account.id,
+          walletId: usdtWallet.id,
+        })
+        if (promoteResult instanceof Error) {
+          baseLogger.error(
+            { error: promoteResult, tx_hash, accountId: account.id },
+            "Failed to promote Bridge deposit audit row to Completed",
+          )
+          alertIbexCryptoReceiveFailure({
+            txHash: String(tx_hash),
+            code: "erpnext_promote_failed",
+            title: "Bridge deposit audit row promotion failed",
+            detail: promoteResult.message,
+            context: { accountId: account.id, walletId: usdtWallet.id, address },
+          })
+        } else {
+          baseLogger.info(
+            { tx_hash, promoteResult },
+            "Bridge deposit audit row promotion result",
+          )
+        }
+
         await sendBridgeDepositNotificationBestEffort({
           accountId: account.id,
           amount: String(usdtAmount.asNumber()),