diff --git a/src/lib/settlement-observability.ts b/src/lib/settlement-observability.ts new file mode 100644 index 0000000..7475f38 --- /dev/null +++ b/src/lib/settlement-observability.ts @@ -0,0 +1,83 @@ +import type { AppLogger } from "../observability/logger"; +import { truncateWalletAddress } from "./kyc"; +import { redactString } from "../observability/redaction-formatter"; +import { stroopsToXlm } from "./stellar-format"; +import { logSettlementCompletion } from "./settlement-completion-log"; + +export interface SettlementStartInput { + invoiceId: string; + actorWallet: string; + startedAt: string; +} + +export interface SettlementFailureInput { + invoiceId: string; + error: unknown; + durationMs?: number; + distributionTxHash?: string | null; + category?: string; + retryable?: boolean; +} + +export interface SettlementSuccessInput { + invoiceId: string; + totalProceedsStroops: bigint; + investorCount: number; + durationMs?: number; + distributionTxHash?: string | null; +} + +export function logSettlementStart(logger: AppLogger, input: SettlementStartInput): void { + logger.info("Starting settlement flow.", { + event: "settlement_started", + invoice_id: input.invoiceId, + actor_wallet: truncateWalletAddress(input.actorWallet), + started_at: input.startedAt, + }); +} + +export function logSettlementFailure(logger: AppLogger, input: SettlementFailureInput): void { + const safeMessage = redactString( + input.error instanceof Error ? input.error.message : String(input.error ?? "") + ); + + logger.warn("Settlement flow failed.", { + event: "settlement_failed", + invoice_id: input.invoiceId, + category: + input.category ?? + (input.error instanceof Error && (input.error as any).code + ? (input.error as any).code + : "unknown"), + retryable: Boolean(input.retryable), + duration_ms: input.durationMs ?? null, + distribution_tx_hash: input.distributionTxHash ?? null, + error_reason: safeMessage, + }); +} + +export function logSettlementSuccess(logger: AppLogger, input: SettlementSuccessInput): void { + // Emit the established completion log used by tests + logSettlementCompletion(logger, { + invoiceId: input.invoiceId, + totalProceedsStroops: input.totalProceedsStroops, + investorCount: input.investorCount, + }); + + // Additional lightweight success event with duration and correlation + logger.info("Settlement flow completed.", { + event: "settlement_completed", + invoice_id: input.invoiceId, + total_proceeds: stroopsToXlm(input.totalProceedsStroops), + investor_count: input.investorCount, + duration_ms: input.durationMs ?? null, + distribution_tx_hash: input.distributionTxHash ?? null, + settled_at: new Date().toISOString(), + }); +} + +export default { + logSettlementStart, + logSettlementFailure, + logSettlementSuccess, +}; diff --git a/src/services/settlement.service.ts b/src/services/settlement.service.ts index 49dfa24..ce8c153 100644 --- a/src/services/settlement.service.ts +++ b/src/services/settlement.service.ts @@ -11,6 +11,11 @@ import { decimalStringToScaledBigInt, scaledBigIntToDecimalString } from "../lib import { logInvoiceTransition } from "../lib/invoice-lifecycle-log"; import { logSettlementCompletion } from "../lib/settlement-completion-log"; import { logger } from "../observability/logger"; +import { + logSettlementStart, + logSettlementFailure, + logSettlementSuccess, +} from "../lib/settlement-observability"; import type { PaymentDistributorContractService } from "./stellar/payment-distributor-contract.service"; // settlement.service.ts stores/computes amounts as decimal strings scaled by @@ -25,7 +30,10 @@ export interface SettleInvoiceInput { actorWallet: string; } -export interface PaymentDistributorSettlementConfig { feeRecipient: string; feeBps: number; } +export interface PaymentDistributorSettlementConfig { + feeRecipient: string; + feeBps: number; +} export interface InvestorSettlement { investmentId: string; @@ -43,7 +51,11 @@ export interface SettleInvoiceResult { } export class SettlementService { - constructor(private readonly dataSource: DataSource, private readonly paymentDistributor?: PaymentDistributorContractService, private readonly distributorConfig?: PaymentDistributorSettlementConfig) {} + constructor( + private readonly dataSource: DataSource, + private readonly paymentDistributor?: PaymentDistributorContractService, + private readonly distributorConfig?: PaymentDistributorSettlementConfig + ) {} /** * Settles a funded invoice by distributing proceeds to each investor @@ -54,145 +66,203 @@ export class SettlementService { const proceeds = new Decimal(proceedsInput); if (proceeds.isNegative() || proceeds.isZero()) { - throw new ServiceError( - "INVALID_PROCEEDS", - "Settlement proceeds must be greater than zero", - ); + throw new ServiceError("INVALID_PROCEEDS", "Settlement proceeds must be greater than zero"); } - return await this.dataSource.transaction(async (transactionalEntityManager: EntityManager) => { - // 1. Lock the invoice row for update (if supported by the driver). - // SQLite does not support row-level locking, so we fall back to a plain read. - let invoice: Invoice | null; - try { - invoice = await transactionalEntityManager - .createQueryBuilder(Invoice, "invoice") - .setLock("pessimistic_write") - .where("invoice.id = :id", { id: invoiceId }) - .getOne(); - } catch { - invoice = await transactionalEntityManager - .createQueryBuilder(Invoice, "invoice") - .where("invoice.id = :id", { id: invoiceId }) - .getOne(); - } + const startedAt = Date.now(); + const startedAtIso = new Date(startedAt).toISOString(); + logSettlementStart(logger, { invoiceId, actorWallet, startedAt: startedAtIso }); - if (!invoice) { - throw new ServiceError("INVOICE_NOT_FOUND", "Invoice not found", 404); - } + try { + return await this.dataSource.transaction( + async (transactionalEntityManager: EntityManager) => { + // 1. Lock the invoice row for update (if supported by the driver). + // SQLite does not support row-level locking, so we fall back to a plain read. + let invoice: Invoice | null; + try { + invoice = await transactionalEntityManager + .createQueryBuilder(Invoice, "invoice") + .setLock("pessimistic_write") + .where("invoice.id = :id", { id: invoiceId }) + .getOne(); + } catch { + invoice = await transactionalEntityManager + .createQueryBuilder(Invoice, "invoice") + .where("invoice.id = :id", { id: invoiceId }) + .getOne(); + } - // 2. Validate invoice status - if (invoice.status !== InvoiceStatus.FUNDED) { - throw new ServiceError( - "INVALID_INVOICE_STATUS", - `INVALID_INVOICE_STATUS: Cannot settle an invoice with status ${invoice.status}`, - ); - } + if (!invoice) { + throw new ServiceError("INVOICE_NOT_FOUND", "Invoice not found", 404); + } - // 3. Find confirmed investments backing this invoice - const investments = await transactionalEntityManager.find(Investment, { - where: { invoiceId: invoice.id, status: InvestmentStatus.CONFIRMED }, - relations: { investor: true }, - }); + // 2. Validate invoice status + if (invoice.status !== InvoiceStatus.FUNDED) { + throw new ServiceError( + "INVALID_INVOICE_STATUS", + `INVALID_INVOICE_STATUS: Cannot settle an invoice with status ${invoice.status}` + ); + } - if (investments.length === 0) { - throw new ServiceError( - "NO_CONFIRMED_INVESTMENTS", - "Invoice has no confirmed investments to settle", - ); - } + // 3. Find confirmed investments backing this invoice + const investments = await transactionalEntityManager.find(Investment, { + where: { invoiceId: invoice.id, status: InvestmentStatus.CONFIRMED }, + relations: { investor: true }, + }); - // 4. Distribute proceeds pro-rata to each investor's share of the total funded amount - const totalFunded = investments.reduce( - (sum, investment) => sum.plus(new Decimal(investment.investmentAmount)), - new Decimal(0), - ); - const totalFundedScaled = decimalStringToScaledBigInt(totalFunded.toFixed(4)); - const proceedsScaled = decimalStringToScaledBigInt(proceeds.toFixed(4)); - const feeScaled = this.distributorConfig - ? (proceedsScaled * BigInt(this.distributorConfig.feeBps)) / 10_000n - : 0n; - const distributableScaled = proceedsScaled - feeScaled; - const settlements: InvestorSettlement[] = []; - let distributionTransactionHash: string | undefined; - - if (this.paymentDistributor) { - if (!this.distributorConfig) { - throw new ServiceError("DISTRIBUTOR_CONFIGURATION_MISSING", "Payment distributor fee configuration is required"); + if (investments.length === 0) { + throw new ServiceError( + "NO_CONFIRMED_INVESTMENTS", + "Invoice has no confirmed investments to settle" + ); + } + + // 4. Distribute proceeds pro-rata to each investor's share of the total funded amount + const totalFunded = investments.reduce( + (sum, investment) => sum.plus(new Decimal(investment.investmentAmount)), + new Decimal(0) + ); + const totalFundedScaled = decimalStringToScaledBigInt(totalFunded.toFixed(4)); + const proceedsScaled = decimalStringToScaledBigInt(proceeds.toFixed(4)); + const feeScaled = this.distributorConfig + ? (proceedsScaled * BigInt(this.distributorConfig.feeBps)) / 10_000n + : 0n; + const distributableScaled = proceedsScaled - feeScaled; + const settlements: InvestorSettlement[] = []; + let distributionTransactionHash: string | undefined; + + if (this.paymentDistributor) { + if (!this.distributorConfig) { + throw new ServiceError( + "DISTRIBUTOR_CONFIGURATION_MISSING", + "Payment distributor fee configuration is required" + ); + } + const distribution = await this.paymentDistributor.distributePayouts({ + invoiceId: invoice.id, + totalAmountStroops: proceedsScaled * DECIMAL_SCALE_TO_STROOP_FACTOR, + feeRecipient: this.distributorConfig.feeRecipient, + feeBps: this.distributorConfig.feeBps, + recipients: investments.map((investment) => ({ + address: investment.investor?.stellarAddress ?? investment.investorId, + amountStroops: + computeInvestorReturn( + decimalStringToScaledBigInt(investment.investmentAmount), + totalFundedScaled, + distributableScaled + ) * DECIMAL_SCALE_TO_STROOP_FACTOR, + })), + }); + distributionTransactionHash = distribution.transactionHash; + await transactionalEntityManager.save( + Transaction, + transactionalEntityManager.create(Transaction, { + userId: invoice.sellerId, + invoiceId: invoice.id, + investmentId: null, + type: TransactionType.PAYMENT, + amount: proceeds.toFixed(4), + stellarTxHash: distribution.transactionHash, + stellarOperationIndex: 0, + status: TransactionStatus.COMPLETED, + }) + ); + } + + for (const investment of investments) { + const investmentAmountScaled = decimalStringToScaledBigInt(investment.investmentAmount); + const actualReturnScaled = computeInvestorReturn( + investmentAmountScaled, + totalFundedScaled, + distributableScaled + ); + + investment.actualReturn = scaledBigIntToDecimalString(actualReturnScaled); + investment.status = InvestmentStatus.SETTLED; + await transactionalEntityManager.save(Investment, investment); + + settlements.push({ + investmentId: investment.id, + investorId: investment.investorId, + investmentAmount: investment.investmentAmount, + actualReturn: investment.actualReturn, + }); + } + + // 5. Transition invoice to SETTLED + const previousStatus = invoice.status; + invoice.status = InvoiceStatus.SETTLED; + await transactionalEntityManager.save(Invoice, invoice); + + logInvoiceTransition(logger, { + invoiceId: invoice.id, + fromState: previousStatus, + toState: InvoiceStatus.SETTLED, + actorWallet, + reason: "admin_settled", + }); + + // Emit completion logs (keeps existing completion log for tests) + const durationMs = Date.now() - startedAt; + logSettlementSuccess(logger, { + invoiceId: invoice.id, + totalProceedsStroops: proceedsScaled * DECIMAL_SCALE_TO_STROOP_FACTOR, + investorCount: settlements.length, + durationMs, + distributionTxHash: distributionTransactionHash ?? null, + }); + + return { + invoiceId: invoice.id, + status: InvoiceStatus.SETTLED as const, + proceeds: proceeds.toFixed(4), + settlements, + distributionTransactionHash, + }; } - const distribution = await this.paymentDistributor.distributePayouts({ - invoiceId: invoice.id, - totalAmountStroops: proceedsScaled * DECIMAL_SCALE_TO_STROOP_FACTOR, - feeRecipient: this.distributorConfig.feeRecipient, - feeBps: this.distributorConfig.feeBps, - recipients: investments.map((investment) => ({ - address: investment.investor?.stellarAddress ?? investment.investorId, - amountStroops: computeInvestorReturn(decimalStringToScaledBigInt(investment.investmentAmount), totalFundedScaled, distributableScaled) * DECIMAL_SCALE_TO_STROOP_FACTOR, - })), - }); - distributionTransactionHash = distribution.transactionHash; - await transactionalEntityManager.save(Transaction, transactionalEntityManager.create(Transaction, { - userId: invoice.sellerId, - invoiceId: invoice.id, - investmentId: null, - type: TransactionType.PAYMENT, - amount: proceeds.toFixed(4), - stellarTxHash: distribution.transactionHash, - stellarOperationIndex: 0, - status: TransactionStatus.COMPLETED, - })); - } + ); + } catch (err) { + const durationMs = Date.now() - startedAt; - for (const investment of investments) { - const investmentAmountScaled = decimalStringToScaledBigInt(investment.investmentAmount); - const actualReturnScaled = computeInvestorReturn( - investmentAmountScaled, - totalFundedScaled, - distributableScaled, - ); - - investment.actualReturn = scaledBigIntToDecimalString(actualReturnScaled); - investment.status = InvestmentStatus.SETTLED; - await transactionalEntityManager.save(Investment, investment); - - settlements.push({ - investmentId: investment.id, - investorId: investment.investorId, - investmentAmount: investment.investmentAmount, - actualReturn: investment.actualReturn, - }); + // Determine retryability: Soroban-mapped ServiceErrors use codes starting with 'soroban_' + let retryable = false; + let category = "unknown"; + if (err instanceof ServiceError) { + category = err.code; + if (category.startsWith("soroban_")) { + // Treat some soroban codes as retryable + const retryableCodes = new Set([ + "soroban_timeout", + "soroban_rate_limited", + "soroban_capacity_exceeded", + "soroban_try_again_later", + ]); + retryable = retryableCodes.has(category); + } else { + retryable = false; + } + } else { + retryable = true; // unknown errors likely transient } - // 5. Transition invoice to SETTLED - const previousStatus = invoice.status; - invoice.status = InvoiceStatus.SETTLED; - await transactionalEntityManager.save(Invoice, invoice); - - logInvoiceTransition(logger, { - invoiceId: invoice.id, - fromState: previousStatus, - toState: InvoiceStatus.SETTLED, - actorWallet, - reason: "admin_settled", + logSettlementFailure(logger, { + invoiceId, + error: err, + durationMs, + distributionTxHash: undefined, + category, + retryable, }); - logSettlementCompletion(logger, { - invoiceId: invoice.id, - totalProceedsStroops: proceedsScaled * DECIMAL_SCALE_TO_STROOP_FACTOR, - investorCount: settlements.length, - }); - - return { - invoiceId: invoice.id, - status: InvoiceStatus.SETTLED as const, - proceeds: proceeds.toFixed(4), - settlements, - distributionTransactionHash, - }; - }); + throw err; + } } } -export function createSettlementService(dataSource: DataSource, paymentDistributor?: PaymentDistributorContractService, distributorConfig?: PaymentDistributorSettlementConfig): SettlementService { +export function createSettlementService( + dataSource: DataSource, + paymentDistributor?: PaymentDistributorContractService, + distributorConfig?: PaymentDistributorSettlementConfig +): SettlementService { return new SettlementService(dataSource, paymentDistributor, distributorConfig); } diff --git a/src/services/stellar/payment-distributor-contract.service.ts b/src/services/stellar/payment-distributor-contract.service.ts index 0da9875..eba4699 100644 --- a/src/services/stellar/payment-distributor-contract.service.ts +++ b/src/services/stellar/payment-distributor-contract.service.ts @@ -1,4 +1,13 @@ -import { Contract, Address, nativeToScVal, xdr, SorobanRpc, Keypair, TransactionBuilder, BASE_FEE } from "stellar-sdk"; +import { + Contract, + Address, + nativeToScVal, + xdr, + SorobanRpc, + Keypair, + TransactionBuilder, + BASE_FEE, +} from "stellar-sdk"; import type { AppLogger } from "../../observability/logger"; import { logger as globalLogger } from "../../observability/logger"; @@ -31,7 +40,10 @@ export interface DistributePayoutsInput { feeBps: number; } -export interface DistributePayoutsResult { transactionHash: string; ledger: number | null; } +export interface DistributePayoutsResult { + transactionHash: string; + ledger: number | null; +} const MAX_FEE_BPS = 10_000; @@ -59,7 +71,7 @@ export class PaymentDistributorContractService { constructor( dependenciesOrContractId: string | PaymentDistributorContractServiceDependencies, - logger?: AppLogger, + logger?: AppLogger ) { if (typeof dependenciesOrContractId === "string") { if (!dependenciesOrContractId) { @@ -108,7 +120,7 @@ export class PaymentDistributorContractService { invoiceId: string, recipients: PayoutRecipient[], platformFeeAccount: string, - feeBps: number, + feeBps: number ): xdr.Operation { if (recipients.length === 0) { throw new Error("At least one payout recipient is required."); @@ -118,7 +130,7 @@ export class PaymentDistributorContractService { } const recipientAddressesScVal = xdr.ScVal.scvVec( - recipients.map((recipient) => new Address(recipient.address).toScVal()), + recipients.map((recipient) => new Address(recipient.address).toScVal()) ); const recipientAmountsScVal = xdr.ScVal.scvVec( @@ -128,7 +140,7 @@ export class PaymentDistributorContractService { ? recipient.amountStroops : BigInt(recipient.amountStroops); return nativeToScVal(amountBigInt, { type: "i128" }); - }), + }) ); return this.contract.call( @@ -137,7 +149,7 @@ export class PaymentDistributorContractService { recipientAddressesScVal, recipientAmountsScVal, new Address(platformFeeAccount).toScVal(), - nativeToScVal(feeBps, { type: "u32" }), + nativeToScVal(feeBps, { type: "u32" }) ); } @@ -148,7 +160,10 @@ export class PaymentDistributorContractService { if (this.verifyDistributorWiring && !(await this.verifyDistributorWiring())) { throw new Error("Payment distributor is not initialized on the invoice escrow contract."); } - const recipientTotal = input.recipients.reduce((sum, recipient) => sum + BigInt(recipient.amountStroops), 0n); + const recipientTotal = input.recipients.reduce( + (sum, recipient) => sum + BigInt(recipient.amountStroops), + 0n + ); const fee = (input.totalAmountStroops * BigInt(input.feeBps)) / 10_000n; if (recipientTotal + fee > input.totalAmountStroops) { throw new Error("Payout recipients and protocol fee exceed the settlement total."); @@ -159,21 +174,60 @@ export class PaymentDistributorContractService { const transaction = new TransactionBuilder(account, { fee: BASE_FEE, networkPassphrase: this.networkPassphrase, - }).addOperation(this.buildDistributePayoutsTx(input.invoiceId, input.recipients, input.feeRecipient, input.feeBps)).setTimeout(30).build(); + }) + .addOperation( + this.buildDistributePayoutsTx( + input.invoiceId, + input.recipients, + input.feeRecipient, + input.feeBps + ) + ) + .setTimeout(30) + .build(); const prepared = await this.rpcServer.prepareTransaction(transaction); prepared.sign(signer); - const submitted = await this.rpcServer.sendTransaction(prepared); - if (submitted.status === "ERROR") throw new Error("Payment distribution transaction was rejected by Soroban RPC."); - for (let attempt = 0; attempt < this.confirmationAttempts; attempt++) { - const result = await this.rpcServer.getTransaction(submitted.hash); - if (result.status === "SUCCESS") { - this.logger.info("Payment distribution confirmed on-chain.", { invoice_id: input.invoiceId, transaction_hash: submitted.hash }); - return { transactionHash: submitted.hash, ledger: "ledger" in result ? Number(result.ledger) : null }; + // Submit and map Soroban RPC errors to ServiceError with retryability + try { + const submitted = await this.rpcServer.sendTransaction(prepared); + if (submitted.status === "ERROR") { + // Map to a ServiceError (non-retryable transaction rejection) + throw require("../stellar/soroban-error-mapper").mapSorobanError(submitted, { + contractId: this.contractId, + }).error; + } + + for (let attempt = 0; attempt < this.confirmationAttempts; attempt++) { + const result = await this.rpcServer.getTransaction(submitted.hash); + if (result.status === "SUCCESS") { + this.logger.info("Payment distribution confirmed on-chain.", { + invoice_id: input.invoiceId, + transaction_hash: submitted.hash, + }); + return { + transactionHash: submitted.hash, + ledger: "ledger" in result ? Number(result.ledger) : null, + }; + } + if (result.status === "FAILED") { + throw require("../stellar/soroban-error-mapper").mapSorobanError( + { status: "FAILED" }, + { contractId: this.contractId } + ).error; + } + await new Promise((resolve) => setTimeout(resolve, this.confirmationPollMs)); } - if (result.status === "FAILED") throw new Error("Payment distribution transaction reverted on-chain."); - await new Promise((resolve) => setTimeout(resolve, this.confirmationPollMs)); + throw new Error("Timed out waiting for payment distribution confirmation."); + } catch (err) { + // If it's already a ServiceError, rethrow; otherwise map and throw a sanitized ServiceError + if (err instanceof Error && (err as any).name === "ServiceError") throw err; + const mapper = require("../stellar/soroban-error-mapper"); + const mapped = mapper.mapSorobanError(err, { + contractId: this.contractId, + invoiceId: input.invoiceId, + }); + throw mapped.error; } - throw new Error("Timed out waiting for payment distribution confirmation."); } } diff --git a/src/services/stellar/soroban-error-mapper.ts b/src/services/stellar/soroban-error-mapper.ts new file mode 100644 index 0000000..3a83cdc --- /dev/null +++ b/src/services/stellar/soroban-error-mapper.ts @@ -0,0 +1,126 @@ +import { ServiceError } from "../../utils/service-error"; + +export interface MappedSorobanError { + error: ServiceError; + retryable: boolean; + cause?: string; +} + +/** + * Map common Soroban RPC / contract errors into stable ServiceError categories. + * + * This intentionally returns a sanitized ServiceError and a `retryable` flag. + * Avoid embedding raw provider payloads or secrets in the returned error. + */ +export function mapSorobanError( + input: unknown, + context: { contractId?: string; invoiceId?: string } = {} +): MappedSorobanError { + // Normalize message + const message = + input instanceof Error + ? input.message + : typeof input === "string" + ? input + : JSON.stringify(input ?? {}); + + const lower = (message ?? "").toLowerCase(); + + // Timeouts / network blips + if (/timeout|timed out|etimedout/.test(lower)) { + return { + error: new ServiceError("soroban_timeout", "Soroban RPC timed out.", 503, { + contractId: context.contractId, + invoiceId: context.invoiceId, + }), + retryable: true, + cause: "timeout", + }; + } + + // Rate limiting + if (/429|rate limit|too many requests/.test(lower)) { + return { + error: new ServiceError("soroban_rate_limited", "Soroban RPC rate limited.", 503, { + contractId: context.contractId, + }), + retryable: true, + cause: "rate_limited", + }; + } + + // Simulation failures (contract-level reverts, validation failures) + if (/simulate|simulation failed|simulation error|revert|reverted|contract failed/.test(lower)) { + return { + error: new ServiceError( + "soroban_simulation_failed", + "Soroban transaction simulation failed.", + 422, + { contractId: context.contractId } + ), + retryable: false, + cause: "simulation_failed", + }; + } + + // Authorization failures + if (/authorization|auth failed|unauthorized|not authorized|forbidden/.test(lower)) { + return { + error: new ServiceError("soroban_unauthorized", "Soroban authorization failed.", 403, { + contractId: context.contractId, + }), + retryable: false, + cause: "unauthorized", + }; + } + + // Capacity / resource exhaustion + if ( + /capacity|out of memory|insufficient resources|gas|resource limit|limit exceeded/.test(lower) + ) { + return { + error: new ServiceError("soroban_capacity_exceeded", "Soroban node capacity exceeded.", 503, { + contractId: context.contractId, + }), + retryable: true, + cause: "capacity_exceeded", + }; + } + + // Transaction-level rejection reported by sendTransaction result + if (typeof input === "object" && input !== null && "status" in (input as any)) { + const st = String((input as any).status).toUpperCase(); + if (st === "ERROR" || st === "FAILED") { + return { + error: new ServiceError("soroban_tx_rejected", "Soroban transaction was rejected.", 422, { + contractId: context.contractId, + }), + retryable: false, + cause: "tx_rejected", + }; + } + if (st === "TRY_AGAIN_LATER") { + return { + error: new ServiceError( + "soroban_try_again_later", + "Soroban RPC asked to try again later.", + 503, + { contractId: context.contractId } + ), + retryable: true, + cause: "try_again_later", + }; + } + } + + // Fallback: generic RPC error, mark retryable (network/backpressure) + return { + error: new ServiceError("soroban_rpc_error", "Soroban RPC error.", 502, { + contractId: context.contractId, + }), + retryable: true, + cause: "rpc_error", + }; +} + +export default mapSorobanError; diff --git a/src/services/stellar/verify-payment.service.ts b/src/services/stellar/verify-payment.service.ts index fb37dc7..c903ba2 100644 --- a/src/services/stellar/verify-payment.service.ts +++ b/src/services/stellar/verify-payment.service.ts @@ -4,6 +4,7 @@ import { Investment } from "../../models/Investment.model"; import { Transaction } from "../../models/Transaction.model"; import { InvestmentStatus, TransactionStatus, TransactionType } from "../../types/enums"; import { ServiceError } from "../../utils/service-error"; +import type { AppLogger } from "../../observability/logger"; import { normalizeHorizonPayment, normalizeHorizonTransaction } from "../../utils/horizon-response"; type FetchLike = typeof fetch; @@ -65,7 +66,7 @@ interface PaymentVerificationUnitOfWork { interface PaymentTransactionRunner { runInTransaction( - callback: (unitOfWork: PaymentVerificationUnitOfWork) => Promise, + callback: (unitOfWork: PaymentVerificationUnitOfWork) => Promise ): Promise; } @@ -75,6 +76,7 @@ interface VerifyPaymentServiceDependencies { config: PaymentVerificationConfig; fetchImplementation?: FetchLike; sleep?: SleepFn; + logger?: AppLogger; } export class VerifyPaymentService { @@ -83,18 +85,29 @@ export class VerifyPaymentService { private readonly config: PaymentVerificationConfig; private readonly fetchImplementation: FetchLike; private readonly sleep: SleepFn; + private readonly logger: AppLogger; + + private static NOOP_LOGGER: AppLogger = { + debug: () => undefined, + info: () => undefined, + warn: () => undefined, + error: () => undefined, + child: () => VerifyPaymentService.NOOP_LOGGER, + }; constructor(dependencies: VerifyPaymentServiceDependencies) { this.investmentReader = dependencies.investmentReader; this.transactionRunner = dependencies.transactionRunner; this.config = dependencies.config; this.fetchImplementation = dependencies.fetchImplementation ?? fetch; - this.sleep = dependencies.sleep ?? ((ms: number) => new Promise((resolve) => setTimeout(resolve, ms))); + this.sleep = + dependencies.sleep ?? ((ms: number) => new Promise((resolve) => setTimeout(resolve, ms))); + this.logger = (dependencies.logger ?? VerifyPaymentService.NOOP_LOGGER).child({ + component: "stellar-verify-payment", + }); } - async verifyPayment( - input: PaymentVerificationInput, - ): Promise { + async verifyPayment(input: PaymentVerificationInput): Promise { const investment = await this.investmentReader.findById(input.investmentId); if (!investment) { @@ -104,7 +117,8 @@ export class VerifyPaymentService { if (investment.status === InvestmentStatus.CONFIRMED) { if ( investment.transactionHash === input.stellarTxHash && - investment.stellarOperationIndex === (input.operationIndex ?? investment.stellarOperationIndex) + investment.stellarOperationIndex === + (input.operationIndex ?? investment.stellarOperationIndex) ) { return { outcome: "already_verified", @@ -119,15 +133,32 @@ export class VerifyPaymentService { throw new ServiceError( "reconciliation_conflict", "Investment is already confirmed with a different Stellar payment.", - 409, + 409 ); } - const matchedPayment = await this.fetchAndValidatePayment( - input.stellarTxHash, - investment.investmentAmount, - input.operationIndex, - ); + let matchedPayment; + try { + matchedPayment = await this.fetchAndValidatePayment( + input.stellarTxHash, + investment.investmentAmount, + input.operationIndex + ); + } catch (err) { + // Emit a structured log for validation rejections + this.logger.warn("Investment funding validation rejected.", { + event: "investment_funding_rejected", + investment_id: investment.id, + invoice_id: investment.invoiceId, + wallet_id: investment.investorId, + stellar_tx_hash: input.stellarTxHash, + operation_index: input.operationIndex ?? null, + error_code: err instanceof ServiceError ? err.code : undefined, + error_reason: err instanceof Error ? err.message : String(err), + }); + + throw err; + } return this.transactionRunner.runInTransaction(async (unitOfWork) => { const lockedInvestment = await unitOfWork.findInvestmentByIdForUpdate(input.investmentId); @@ -137,14 +168,14 @@ export class VerifyPaymentService { } const linkedTransactions = await unitOfWork.findTransactionsByInvestmentIdForUpdate( - lockedInvestment.id, + lockedInvestment.id ); if (linkedTransactions.length > 1) { throw new ServiceError( "reconciliation_conflict", "Multiple transaction rows are linked to the same investment.", - 409, + 409 ); } @@ -168,7 +199,7 @@ export class VerifyPaymentService { throw new ServiceError( "reconciliation_conflict", "Investment was confirmed by another transaction while verification was in progress.", - 409, + 409 ); } @@ -182,7 +213,7 @@ export class VerifyPaymentService { throw new ServiceError( "reconciliation_conflict", "Transaction row is already linked to a different Stellar hash.", - 409, + 409 ); } @@ -216,6 +247,18 @@ export class VerifyPaymentService { const savedTransaction = await unitOfWork.saveTransaction(transaction); await unitOfWork.saveInvestment(lockedInvestment); + // Structured log for accepted funding + this.logger.info("Investment funding accepted.", { + event: "investment_funding_accepted", + investment_id: lockedInvestment.id, + invoice_id: lockedInvestment.invoiceId, + wallet_id: lockedInvestment.investorId, + transaction_id: savedTransaction.id, + stellar_tx_hash: input.stellarTxHash, + operation_index: matchedPayment.operationIndex, + amount: lockedInvestment.investmentAmount, + }); + return { outcome: "verified" as const, investmentId: lockedInvestment.id, @@ -230,22 +273,18 @@ export class VerifyPaymentService { private async fetchAndValidatePayment( stellarTxHash: string, expectedAmount: string, - operationIndex?: number, + operationIndex?: number ): Promise { - const transaction = normalizeHorizonTransaction(await this.fetchJson( - `/transactions/${stellarTxHash}`, - )); + const transaction = normalizeHorizonTransaction( + await this.fetchJson(`/transactions/${stellarTxHash}`) + ); if (!transaction.successful) { - throw new ServiceError( - "invalid_payment", - "The Stellar transaction was not successful.", - 422, - ); + throw new ServiceError("invalid_payment", "The Stellar transaction was not successful.", 422); } const operations = await this.fetchJson( - `/transactions/${stellarTxHash}/operations?limit=200&order=asc`, + `/transactions/${stellarTxHash}/operations?limit=200&order=asc` ); const paymentOperations = (operations._embedded?.records ?? []) @@ -265,11 +304,7 @@ export class VerifyPaymentService { operation.assetIssuer === this.config.usdcAssetIssuer && operation.destination === this.config.escrowPublicKey && operation.amount !== null && - amountsWithinDelta( - operation.amount, - expectedAmount, - this.config.allowedAmountDelta, - ) + amountsWithinDelta(operation.amount, expectedAmount, this.config.allowedAmountDelta) ); }); @@ -277,7 +312,7 @@ export class VerifyPaymentService { throw new ServiceError( "invalid_payment", "No Stellar payment operation matched the expected asset, amount, and destination.", - 422, + 422 ); } @@ -285,7 +320,7 @@ export class VerifyPaymentService { throw new ServiceError( "invalid_payment", "Multiple payment operations matched. Supply operationIndex to disambiguate.", - 422, + 422 ); } @@ -316,7 +351,7 @@ export class VerifyPaymentService { throw new ServiceError( "transaction_not_found", "The Stellar transaction could not be found in Horizon.", - 404, + 404 ); } @@ -328,7 +363,7 @@ export class VerifyPaymentService { throw new ServiceError( "horizon_request_failed", "Horizon rejected the verification request.", - 502, + 502 ); } @@ -342,7 +377,7 @@ export class VerifyPaymentService { throw new ServiceError( "horizon_unavailable", "Horizon is temporarily unavailable. Please retry later.", - 503, + 503 ); } @@ -353,7 +388,7 @@ export class VerifyPaymentService { throw new ServiceError( "horizon_unavailable", "Horizon is temporarily unavailable. Please retry later.", - 503, + 503 ); } } @@ -372,7 +407,7 @@ class TypeOrmTransactionRunner implements PaymentTransactionRunner { constructor(private readonly dataSource: DataSource) {} runInTransaction( - callback: (unitOfWork: PaymentVerificationUnitOfWork) => Promise, + callback: (unitOfWork: PaymentVerificationUnitOfWork) => Promise ): Promise { return this.dataSource.transaction(async (manager) => callback({ @@ -390,14 +425,14 @@ class TypeOrmTransactionRunner implements PaymentTransactionRunner { manager.getRepository(Transaction).save(transaction), createTransaction: (input: Partial) => manager.getRepository(Transaction).create(input), - }), + }) ); } } export function createVerifyPaymentService( dataSource: DataSource, - config: PaymentVerificationConfig, + config: PaymentVerificationConfig ): VerifyPaymentService { return new VerifyPaymentService({ investmentReader: new TypeOrmInvestmentReader(dataSource.getRepository(Investment)), @@ -418,9 +453,8 @@ function amountsWithinDelta(actual: string, expected: string, delta: string): bo const expectedValue = toScaledBigInt(expected, scale); const deltaValue = toScaledBigInt(delta, scale); - const difference = actualValue >= expectedValue - ? actualValue - expectedValue - : expectedValue - actualValue; + const difference = + actualValue >= expectedValue ? actualValue - expectedValue : expectedValue - actualValue; return difference <= deltaValue; } diff --git a/src/workers/reconcile-pending-stellar-state.worker.ts b/src/workers/reconcile-pending-stellar-state.worker.ts index 611c38d..b97920f 100644 --- a/src/workers/reconcile-pending-stellar-state.worker.ts +++ b/src/workers/reconcile-pending-stellar-state.worker.ts @@ -88,8 +88,7 @@ export class ReconcilePendingStellarStateWorker { }); this.now = dependencies.now ?? (() => new Date()); this.yieldControl = - dependencies.yieldControl ?? - (() => new Promise((resolve) => setImmediate(resolve))); + dependencies.yieldControl ?? (() => new Promise((resolve) => setImmediate(resolve))); this.setIntervalFn = dependencies.setIntervalFn ?? setInterval; this.clearIntervalFn = dependencies.clearIntervalFn ?? clearInterval; } @@ -132,10 +131,7 @@ export class ReconcilePendingStellarStateWorker { const deadline = startedAt.getTime() + this.config.maxRuntimeMs; try { - const candidates = await this.repository.findPendingCandidates( - cutoff, - this.config.batchSize, - ); + const candidates = await this.repository.findPendingCandidates(cutoff, this.config.batchSize); const cycleId = randomUUID(); this.logger.info("Started Stellar reconciliation tick.", { @@ -175,12 +171,13 @@ export class ReconcilePendingStellarStateWorker { result.processed += 1; result.failed += 1; this.logger.warn("Failed to reconcile pending Stellar state.", { - investmentId: candidate.investmentId, - stellarTxHash: candidate.stellarTxHash, - operationIndex: candidate.operationIndex, + event: "investment_funding_failed", + investment_id: candidate.investmentId, + stellar_tx_hash: candidate.stellarTxHash, + operation_index: candidate.operationIndex, source: candidate.source, - errorCode: error instanceof ServiceError ? error.code : undefined, - error: error instanceof Error ? error.message : "Unknown error", + error_code: error instanceof ServiceError ? error.code : undefined, + error_reason: error instanceof Error ? error.message : "Unknown error", }); } @@ -240,12 +237,10 @@ export class ReconcilePendingStellarStateWorker { } } -class TypeOrmReconciliationCandidateRepository - implements ReconciliationCandidateRepository -{ +class TypeOrmReconciliationCandidateRepository implements ReconciliationCandidateRepository { constructor( private readonly investmentRepository: Repository, - private readonly transactionRepository: Repository, + private readonly transactionRepository: Repository ) {} async findPendingCandidates(olderThan: Date, limit: number): Promise { @@ -318,12 +313,12 @@ export function createReconcilePendingStellarStateWorker( dataSource: DataSource, paymentVerifier: VerifyPaymentService, config: AppConfig["reconciliation"], - logger: AppLogger, + logger: AppLogger ): ReconcilePendingStellarStateWorker { return new ReconcilePendingStellarStateWorker({ repository: new TypeOrmReconciliationCandidateRepository( dataSource.getRepository(Investment), - dataSource.getRepository(Transaction), + dataSource.getRepository(Transaction) ), paymentVerifier, config,