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,