feat: bridge reconciliation + USDT provisioning additions

dev/apollo-federation/supergraph.graphql

@@ -1646,7 +1646,6 @@ type Query
   @join__type(graph: PUBLIC)
 {
   accountDefaultWallet(username: Username!, walletCurrency: WalletCurrency): PublicWallet!
-  latestAccountUpgradeRequest: AccountUpgradeRequestPayload!
   bridgeExternalAccounts: [BridgeExternalAccount]
   bridgeKycStatus: String
   bridgeVirtualAccount: BridgeVirtualAccount
@@ -1657,6 +1656,7 @@ type Query
   currencyList: [Currency!]!
   globals: Globals
   isFlashNpub(input: IsFlashNpubInput!): IsFlashNpubPayload
+  latestAccountUpgradeRequest: AccountUpgradeRequestPayload!
   lnInvoicePaymentStatus(input: LnInvoicePaymentStatusInput!): LnInvoicePaymentStatusPayload!
   me: User
   mobileVersions: [MobileVersions]

dev/bruno/Flash GraphQL API/environments/local.bru

@@ -8,6 +8,6 @@ vars {
   token:
   walletId:
   walletIdUsd: c593736e-5a58-42e4-93fa-dc895856c1f1
-  userEmail: maurienteso@gmail.com
-  userFullName: maurienteso
+  userEmail: mauriente@gmail.com
+  userFullName: maurientes
 }

src/app/authentication/login.ts

@@ -309,7 +309,10 @@ export const loginDeviceUpgradeWithPhone = async ({
   if (deviceWallets instanceof Error) return deviceWallets
   let deviceAccountHasBalance = false
   for (const wallet of deviceWallets) {
-    const balance = await getBalanceForWallet({ walletId: wallet.id })
+    const balance = await getBalanceForWallet({
+      walletId: wallet.id,
+      currency: wallet.currency,
+    })
     if (balance instanceof Error) return balance
     if (!balance.isZero()) {
       deviceAccountHasBalance = true

src/domain/pubsub/index.ts

@@ -5,6 +5,7 @@ export const PubSubDefaultTriggers = {
   UserPriceUpdate: "USER_PRICE_UPDATE",
   AccountUpdate: "ACCOUNT_UPDATE",
   LnPaymentStatus: "LN_PAYMENT_STATUS",
+  BridgeReconciliationUpdate: "BRIDGE_RECONCILIATION_UPDATE",
 } as const
 
 export const customPubSubTrigger = ({

src/graphql/admin/queries.ts

@@ -15,6 +15,7 @@ import AccountDetailsByAccountId from "./root/query/account-details-by-account-i
 import MerchantsPendingApprovalQuery from "./root/query/merchants-pending-approval-listing"
 import IdDocumentReadUrlQuery from "./root/query/id-document-read-url"
 import NotificationTopicsQuery from "./root/query/notification-topics"
+import BridgeReconciliationOrphansQuery from "./root/query/bridge-reconciliation-orphans"
 
 export const queryFields = {
   unauthed: {},
@@ -34,6 +35,7 @@ export const queryFields = {
     merchantsPendingApproval: MerchantsPendingApprovalQuery,
     idDocumentReadUrl: IdDocumentReadUrlQuery,
     notificationTopics: NotificationTopicsQuery,
+    bridgeReconciliationOrphans: BridgeReconciliationOrphansQuery,
   },
 }
 

src/graphql/admin/root/query/bridge-reconciliation-orphans.ts

@@ -0,0 +1,40 @@
+import { GT } from "@graphql/index"
+import BridgeReconciliationOrphanObject from "@graphql/admin/types/object/bridge-reconciliation-orphan"
+import { findOrphans } from "@services/mongoose/bridge-reconciliation-orphan"
+
+const BridgeReconciliationOrphansQuery = GT.Field({
+  type: GT.NonNullList(BridgeReconciliationOrphanObject),
+  args: {
+    status: { type: GT.String, defaultValue: null },
+    orphanType: { type: GT.String, defaultValue: null },
+    limit: { type: GT.Int, defaultValue: 50 },
+  },
+  resolve: async (
+    _: unknown,
+    {
+      status,
+      orphanType,
+      limit,
+    }: { status?: string; orphanType?: string; limit?: number },
+  ) => {
+    const result = await findOrphans({
+      status: status as "unmatched" | "resolved" | undefined,
+      orphanType: orphanType as
+        | "bridge_without_ibex"
+        | "ibex_without_bridge"
+        | undefined,
+      limit: limit ?? 50,
+    })
+
+    if (result instanceof Error) throw result
+
+    return result.map((o) => ({
+      ...o,
+      detectedAt: o.detectedAt.toISOString(),
+      resolvedAt: o.resolvedAt?.toISOString() ?? null,
+      triageContext: JSON.stringify(o.triageContext),
+    }))
+  },
+})
+
+export default BridgeReconciliationOrphansQuery

src/graphql/admin/schema.graphql

@@ -110,6 +110,21 @@ type BTCWallet implements Wallet {
   walletCurrency: WalletCurrency!
 }
 
+type BridgeReconciliationOrphan {
+  amount: String
+  currency: String
+  customerId: String
+  detectedAt: String!
+  id: ID!
+  orphanKey: String!
+  orphanType: String!
+  resolvedAt: String
+  status: String!
+  transferId: String
+  triageContext: String!
+  txHash: String
+}
+
 input BusinessDeleteMapInfoInput {
   username: Username!
 }
@@ -319,6 +334,7 @@ type Query {
   accountDetailsByUserPhone(phone: Phone!): AuditedAccount!
   accountDetailsByUsername(username: Username!): AuditedAccount!
   allLevels: [AccountLevel!]!
+  bridgeReconciliationOrphans(limit: Int = 50, orphanType: String = null, status: String = null): [BridgeReconciliationOrphan!]!
   idDocumentReadUrl(
     """Storage key of the ID document file"""
     fileKey: String!
@@ -551,6 +567,50 @@ type UsdWallet implements Wallet {
   walletCurrency: WalletCurrency!
 }
 
+"""
+A wallet belonging to an account which contains a USDT balance and a list of transactions.
+"""
+type UsdtWallet implements Wallet {
+  accountId: ID!
+  balance: FractionalCentAmount!
+  id: ID!
+  isExternal: Boolean!
+  lnurlp: Lnurl
+
+  """An unconfirmed incoming onchain balance."""
+  pendingIncomingBalance: SignedAmount!
+  transactions(
+    """Returns the items in the list that come after the specified cursor."""
+    after: String
+
+    """Returns the items in the list that come before the specified cursor."""
+    before: String
+
+    """Returns the first n items from the list."""
+    first: Int
+
+    """Returns the last n items from the list."""
+    last: Int
+  ): TransactionConnection
+  transactionsByAddress(
+    """Returns the items that include this address."""
+    address: OnChainAddress!
+
+    """Returns the items in the list that come after the specified cursor."""
+    after: String
+
+    """Returns the items in the list that come before the specified cursor."""
+    before: String
+
+    """Returns the first n items from the list."""
+    first: Int
+
+    """Returns the last n items from the list."""
+    last: Int
+  ): TransactionConnection
+  walletCurrency: WalletCurrency!
+}
+
 input UserUpdatePhoneInput {
   accountUuid: ID!
   phone: Phone!

src/graphql/admin/types/index.ts

@@ -1,5 +1,6 @@
 import BtcWallet from "@graphql/shared/types/object/btc-wallet"
 import GraphQLApplicationError from "@graphql/shared/types/object/graphql-application-error"
 import UsdWallet from "@graphql/shared/types/object/usd-wallet"
+import UsdtWallet from "@graphql/shared/types/object/usdt-wallet"
 
-export const ALL_INTERFACE_TYPES = [GraphQLApplicationError, BtcWallet, UsdWallet]
+export const ALL_INTERFACE_TYPES = [GraphQLApplicationError, BtcWallet, UsdWallet, UsdtWallet]

src/graphql/admin/types/object/bridge-reconciliation-orphan.ts

@@ -0,0 +1,21 @@
+import { GT } from "@graphql/index"
+
+const BridgeReconciliationOrphanObject = GT.Object({
+  name: "BridgeReconciliationOrphan",
+  fields: () => ({
+    id: { type: GT.NonNullID },
+    orphanKey: { type: GT.NonNull(GT.String) },
+    orphanType: { type: GT.NonNull(GT.String) },
+    status: { type: GT.NonNull(GT.String) },
+    txHash: { type: GT.String },
+    transferId: { type: GT.String },
+    customerId: { type: GT.String },
+    amount: { type: GT.String },
+    currency: { type: GT.String },
+    detectedAt: { type: GT.NonNull(GT.String) },
+    resolvedAt: { type: GT.String },
+    triageContext: { type: GT.NonNull(GT.String) },
+  }),
+})
+
+export default BridgeReconciliationOrphanObject

src/graphql/public/root/mutation/onchain-payment-send-all.ts

@@ -10,7 +10,7 @@ import WalletId from "@graphql/shared/types/scalar/wallet-id"
 
 import { Wallets } from "@app"
 import { getBalanceForWallet } from "@app/wallets"
-import { USDAmount } from "@domain/shared"
+import { USDAmount, WalletCurrency } from "@domain/shared"
 
 const OnChainPaymentSendAllInput = GT.Input({
   name: "OnChainPaymentSendAllInput",
@@ -63,7 +63,10 @@ const OnChainPaymentSendAllMutation = GT.Field<
       return { errors: [{ message: speed.message }] }
     }
 
-    const amount = await getBalanceForWallet({ walletId })
+    const amount = await getBalanceForWallet({
+      walletId,
+      currency: WalletCurrency.Usd,
+    })
     if (amount instanceof Error) return amount
     if (!(amount instanceof USDAmount)) {
       return { errors: [{ message: "Onchain payments require a USD wallet" }] }

src/graphql/public/schema.graphql

@@ -1290,7 +1290,6 @@ type PublicWallet {
 
 type Query {
   accountDefaultWallet(username: Username!, walletCurrency: WalletCurrency): PublicWallet!
-  latestAccountUpgradeRequest: AccountUpgradeRequestPayload!
   bridgeExternalAccounts: [BridgeExternalAccount]
   bridgeKycStatus: String
   bridgeVirtualAccount: BridgeVirtualAccount
@@ -1301,6 +1300,7 @@ type Query {
   currencyList: [Currency!]!
   globals: Globals
   isFlashNpub(input: IsFlashNpubInput!): IsFlashNpubPayload
+  latestAccountUpgradeRequest: AccountUpgradeRequestPayload!
   lnInvoicePaymentStatus(input: LnInvoicePaymentStatusInput!): LnInvoicePaymentStatusPayload!
   me: User
   mobileVersions: [MobileVersions]

src/graphql/shared/types/object/btc-wallet.ts

@@ -52,7 +52,10 @@ const BtcWallet = GT.Object<Wallet>({
       description: "A balance stored in BTC.",
       resolve: async (source) => {
         if (source.type === WalletType.External) return null
-        const balanceSats = await Wallets.getBalanceForWallet({ walletId: source.id })
+        const balanceSats = await Wallets.getBalanceForWallet({
+          walletId: source.id,
+          currency: source.currency,
+        })
         if (balanceSats instanceof Error) {
           throw mapError(balanceSats)
         }

src/scripts/reconcile-bridge-ibex-deposits.ts

@@ -9,14 +9,14 @@ import { reconcileBridgeAndIbexDeposits } from "@services/bridge/reconciliation"
 const args = yargs(hideBin(process.argv))
   .option("window-hours", {
     type: "number",
-    default: 24,
-    describe: "Reconciliation window in hours",
+    default: 0.25,
+    describe: "Reconciliation window in hours (default: 15 minutes)",
   })
   .option("configPath", { type: "string", demandOption: true })
   .parseSync()
 
 const main = async () => {
-  const windowMs = Math.max(1, Math.floor(args["window-hours"])) * 60 * 60 * 1000
+  const windowMs = args["window-hours"] * 60 * 60 * 1000
   const result = await reconcileBridgeAndIbexDeposits({ windowMs })
   if (result instanceof Error) throw result
   baseLogger.info(result, "Bridge↔IBEX reconciliation finished")

src/servers/cron.ts

@@ -65,10 +65,14 @@ const swapOutJob = async () => {
   if (swapResult instanceof Error) throw swapResult
 }
 
+// Window covers 15 min of events — real-time webhook reconciliation handles everything
+// else immediately. This batch pass is only a safety net for missed/delayed webhooks.
+const RECONCILE_WINDOW_MS = 15 * 60 * 1000
+
 const reconcileBridgeDepositsJob = async () => {
   if (!BridgeConfig.enabled) return
 
-  const result = await reconcileBridgeAndIbexDeposits()
+  const result = await reconcileBridgeAndIbexDeposits({ windowMs: RECONCILE_WINDOW_MS })
   if (result instanceof Error) throw result
 }