chore: add settle pending onchain payments script (#4668)

* chore: add settle pending onchain payments script

* fix: how to run comment

* fix: lint issues

* fix: handle proto generator default values issue

* fix: spell lint

core/api/src/debug/settle-pending-onchain-payments.ts

@@ -0,0 +1,115 @@
+/**
+ * how to run:
+ *
+ * pnpm tsx src/debug/settle-pending-onchain-payments.ts
+ *
+ */
+
+import { Wallets } from "@/app"
+
+import { OnChainService } from "@/services/bria"
+import { LedgerService } from "@/services/ledger"
+import * as LedgerFacade from "@/services/ledger/facade"
+import { lndsConnect } from "@/services/lnd/auth"
+import { isUp } from "@/services/lnd/health"
+import { setupMongoConnection } from "@/services/mongodb"
+
+const onChainService = OnChainService()
+
+const processPayment = async (payment: LedgerTransaction<WalletCurrency>) => {
+  const payout = await onChainService.findPayoutByLedgerJournalId(payment.journalId)
+  if (payout instanceof Error) {
+    return new Error(`Failed to get payout: ${payout.name} - ${payout.message}`)
+  }
+  if (!payout.batchId || !payout.txId || payout.vout === undefined) {
+    return new Error("Missing required payout details")
+  }
+
+  const setTxIdResult = await LedgerFacade.setOnChainTxIdByPayoutId({
+    payoutId: payout.id,
+    txId: payout.txId,
+    vout: payout.vout,
+  })
+  if (setTxIdResult instanceof Error) {
+    return new Error(
+      `Failed to set transaction ID: ${setTxIdResult.name} - ${setTxIdResult.message}`,
+    )
+  }
+
+  const settledPayout = await Wallets.settlePayout(payout.id)
+  if (settledPayout instanceof Error) {
+    return new Error(
+      `Failed to settle payout: ${settledPayout.name} - ${settledPayout.message}`,
+    )
+  }
+
+  return true
+}
+
+const settlePendingOnchainPayments = async () => {
+  const pendingPayments = LedgerService().listPendingOnchainPayments()
+  if (pendingPayments instanceof Error) return pendingPayments
+
+  let totalPayments = 0
+  let successCount = 0
+  let errorCount = 0
+  const errors = []
+
+  for await (const payment of pendingPayments) {
+    totalPayments++
+    console.log(`Processing payment ${totalPayments}`)
+
+    const result = await processPayment(payment)
+    if (result instanceof Error) {
+      errorCount++
+      errors.push({
+        journalId: payment.journalId,
+        payoutId: payment.payoutId,
+        error: result.message,
+      })
+      console.error(`Failed to process payment: ${result.message}`)
+      continue
+    }
+
+    successCount++
+    console.log(
+      `Successfully processed payout ${payment.payoutId} for journal ${payment.journalId}`,
+    )
+  }
+
+  return {
+    successCount,
+    errorCount,
+    total: totalPayments,
+    errors,
+  }
+}
+
+const main = async () => {
+  const result = await settlePendingOnchainPayments()
+  if (result instanceof Error) {
+    console.error("Error:", result)
+    return
+  }
+  console.log("Settlement process completed")
+  console.log(`Total Processed: ${result.total}`)
+  console.log(`Successful: ${result.successCount}`)
+  console.log(`Failed: ${result.errorCount}`)
+
+  if (result.errors.length > 0) {
+    console.log("\nErrors:")
+    result.errors.forEach(({ journalId, payoutId, error }) => {
+      console.log(
+        `- Journal ${journalId}${payoutId ? ` (Payout ${payoutId})` : ""}: ${error}`,
+      )
+    })
+  }
+}
+
+setupMongoConnection()
+  .then(async (mongoose) => {
+    await Promise.all(lndsConnect.map((lndParams) => isUp(lndParams)))
+    await main()
+    if (mongoose) await mongoose.connection.close()
+  })
+  .catch((err) => console.log(err))

core/api/src/domain/bitcoin/onchain/index.types.d.ts

@@ -10,6 +10,7 @@ type OnChainAddressRequestId = string & { readonly brand: unique symbol }
 type BlockId = string & { readonly brand: unique symbol }
 type OnChainTxHash = string & { readonly brand: unique symbol }
 type PayoutId = string & { readonly brand: unique symbol }
+type BatchId = string & { readonly brand: unique symbol }
 type OnChainTxVout = number & { readonly brand: unique symbol }
 type ScanDepth = number & { readonly brand: unique symbol }
 type TxOut = {
@@ -92,6 +93,9 @@ type OnChainPayout = {
   id: PayoutId
   journalId: LedgerJournalId
   batchInclusionEstimatedAt: number | undefined
+  batchId: BatchId | undefined
+  txId: OnChainTxHash | undefined
+  vout: OnChainTxVout | undefined
 }
 
 type OnChainEventHandler = (event: OnChainEvent) => true | ApplicationError

core/api/src/domain/ledger/index.types.d.ts

@@ -330,6 +330,10 @@ interface ILedgerService {
 
   listWalletIdsWithPendingPayments: () => AsyncGenerator<WalletId> | LedgerServiceError
 
+  listPendingOnchainPayments: () =>
+    | AsyncGenerator<LedgerTransaction<WalletCurrency>>
+    | LedgerServiceError
+
   addColdStorageTxReceive<T extends DisplayCurrency>(
     args: AddColdStorageTxReceiveArgs<T>,
   ): Promise<LedgerJournal | LedgerServiceError>

core/api/src/services/bria/grpc-client.ts

@@ -7,6 +7,8 @@ import {
   EstimatePayoutFeeResponse,
   GetAddressRequest,
   GetAddressResponse,
+  GetBatchRequest,
+  GetBatchResponse,
   GetPayoutRequest,
   GetPayoutResponse,
   GetWalletBalanceSummaryRequest,
@@ -58,3 +60,7 @@ export const estimatePayoutFee = promisify<
 >(bitcoinBridgeClient.estimatePayoutFee.bind(bitcoinBridgeClient))
 
 export const subscribeAll = bitcoinBridgeClient.subscribeAll.bind(bitcoinBridgeClient)
+
+export const getBatch = promisify<GetBatchRequest, Metadata, GetBatchResponse>(
+  bitcoinBridgeClient.getBatch.bind(bitcoinBridgeClient),
+)

core/api/src/services/bria/index.ts

@@ -233,16 +233,17 @@ export const OnChainService = (): IOnChainService => {
 
       const response = await getPayout(request, metadata)
       const foundPayout = response.getPayout()
-
       if (foundPayout === undefined) return new PayoutNotFoundError()
-      let batchInclusionEstimatedAt = undefined
-      if (foundPayout.hasBatchInclusionEstimatedAt()) {
-        batchInclusionEstimatedAt = foundPayout.getBatchInclusionEstimatedAt()
-      }
+
+      //fix issue with proto gen default values
+      const txId = (foundPayout.getTxId() as OnChainTxHash) || undefined
       return {
         id: foundPayout.getId() as PayoutId,
         journalId: foundPayout.getExternalId() as LedgerJournalId,
-        batchInclusionEstimatedAt,
+        batchInclusionEstimatedAt: foundPayout.getBatchInclusionEstimatedAt(),
+        batchId: foundPayout.getBatchId() as BatchId,
+        txId,
+        vout: txId ? (foundPayout.getVout() as OnChainTxVout) : undefined,
       }
     } catch (err) {
       if (
@@ -281,6 +282,9 @@ export const OnChainService = (): IOnChainService => {
         id: response.getId() as PayoutId,
         journalId,
         batchInclusionEstimatedAt: response.getBatchInclusionEstimatedAt(),
+        batchId: undefined,
+        txId: undefined,
+        vout: undefined,
       }
     } catch (err) {
       const errMsg = parseErrorMessageFromUnknown(err)

core/api/src/services/ledger/index.ts

@@ -493,6 +493,35 @@ export const LedgerService = (): ILedgerService => {
     }
   }
 
+  const listPendingOnchainPayments = async function* ():
+    | AsyncGenerator<LedgerTransaction<WalletCurrency>>
+    | LedgerServiceError {
+    try {
+      const bankOwnerWalletId = await caching.getBankOwnerWalletId()
+      const dealerUsdWalletId = await caching.getDealerUsdWalletId()
+      const dealerBtcWalletId = await caching.getDealerBtcWalletId()
+
+      const excludedAccounts = [
+        toLiabilitiesWalletId(bankOwnerWalletId),
+        toLiabilitiesWalletId(dealerUsdWalletId),
+        toLiabilitiesWalletId(dealerBtcWalletId),
+      ]
+
+      const transactions = Transaction.find({
+        type: LedgerTransactionType.OnchainPayment,
+        pending: true,
+        account_path: liabilitiesMainAccount,
+        accounts: { $nin: excludedAccounts },
+      }).cursor({ batchSize: 100 })
+
+      for await (const tx of transactions) {
+        yield translateToLedgerTx(tx)
+      }
+    } catch (error) {
+      return new UnknownLedgerError(error)
+    }
+  }
+
   return wrapAsyncFunctionsToRunInSpan({
     namespace: "services.ledger",
     fns: {
@@ -516,6 +545,7 @@ export const LedgerService = (): ILedgerService => {
       isLnTxRecorded,
       getWalletIdByPaymentHash,
       listWalletIdsWithPendingPayments,
+      listPendingOnchainPayments,
       ...admin,
       ...send,
     },

typos.toml

@@ -1,31 +1,29 @@
 files.extend-exclude = [
-  "dev",
-  "core/api/dev",
-  "core/api/src/domain/users/languages.ts",
-  "core/api/docker-compose.yml",
-  "quickstart/*.yml",
-  "quickstart/galoy",
-  "quickstart/dev",
-  "prelude",
-  "core/notifications/locales/*.yml",
-  "apps/admin-panel/components/notification/languages.ts",
-  "apps/pay/components/success-animation.json",
+    "dev",
+    "core/api/dev",
+    "core/api/src/domain/users/languages.ts",
+    "core/api/docker-compose.yml",
+    "quickstart/*.yml",
+    "quickstart/galoy",
+    "quickstart/dev",
+    "prelude",
+    "core/notifications/locales/*.yml",
+    "apps/admin-panel/components/notification/languages.ts",
+    "apps/pay/components/success-animation.json",
 ]
 
-default.extend-ignore-identifiers-re = [
-  "^bc1\\w+",
-  "^lnbc\\w+",
-  "[Ww][Ss]",
-]
+default.extend-ignore-identifiers-re = ["^bc1\\w+", "^lnbc\\w+", "[Ww][Ss]"]
 
-default.extend-ignore-re = ["\"eyJ[a-zA-Z0-9_\\-\\.]+\"",
-  "moneyImportantGovernement",
-  "GovernementCanPrintMoney",
-  "moneySocialAggrement",
+default.extend-ignore-re = [
+    "\"eyJ[a-zA-Z0-9_\\-\\.]+\"",
+    "moneyImportantGovernement",
+    "GovernementCanPrintMoney",
+    "moneySocialAggrement",
 ]
 
 [default.extend-words]
 # returned from twilio API referenced in src/debug/text-sms.ts
 Ons = "Ons"
 Lsat = "Lsat"
 quizs = "quizs"
+nin = "nin"