diff --git a/docs/stellar-reconciliation.md b/docs/stellar-reconciliation.md new file mode 100644 index 0000000..d4732b1 --- /dev/null +++ b/docs/stellar-reconciliation.md @@ -0,0 +1,31 @@ +# Stellar reconciliation + +The reconciliation module records confirmed Stellar payments, matches them to internal invoices, and preserves every decision in an audit trail. + +## Testnet polling + +Set `STELLAR_RECONCILIATION_ACCOUNT` to the Stellar destination account monitored by the service. The service polls `STELLAR_HORIZON_URL` every 60 seconds and defaults to `https://horizon-testnet.stellar.org`. Horizon paging tokens provide idempotent continuation. Duplicate transaction hashes are ignored and recorded as retry-safe audit entries. + +For webhook or queue consumers, send a confirmed payment to `POST /reconcile/stellar/transactions` with its transaction hash, destination account, amount, asset, and memo. The endpoint accepts the same payload shape as the Horizon adapter. + +## Invoice lifecycle + +Register an invoice with `POST /reconcile/stellar/invoice`. + +```json +{ + "invoiceId": "INV-2026-0001", + "expectedAmount": "125.5000000", + "destinationAccount": "GABC...DEST", + "paymentReference": "order-0001", + "assetCode": "XLM" +} +``` + +Matching requires destination account and asset equality. When an invoice has a payment reference, the incoming memo must match it. The invoice moves from `open` to `partial` or `paid`. Transactions without a matching invoice remain `unmatched`. + +## Lookup and operations + +`GET /reconcile/stellar/tx/:txid` returns the transaction and its decisions. `GET /reconcile/stellar/invoice/:invoiceId` returns the invoice and its decisions. `POST /reconcile/stellar/invoice/:invoiceId/reconcile` retries matching for accounting staff. Administrators can inspect unmatched transactions with `GET /reconcile/stellar/admin/unmatched` and audit records with `GET /reconcile/stellar/admin/audit`. + +The admin endpoints require an authenticated administrator with verified two-factor authentication. All decision records include the invoice, transaction, decision, reason, attempt, and relevant matching metadata. diff --git a/src/app.module.ts b/src/app.module.ts index cbd5d08..22e2d65 100644 --- a/src/app.module.ts +++ b/src/app.module.ts @@ -54,6 +54,7 @@ import { LoggerModule } from "./logging/logger.module"; // Modules – cache import { CacheModule } from "./common/cache/cache.module"; import { BillingModule } from "./billing/billing.module"; +import { ReconciliationModule } from "./reconciliation/reconciliation.module"; // Auth entities import { User } from "./core/user/entities/user.entity"; @@ -108,6 +109,10 @@ import { WebhookDeadLetter } from "./infrastructure/webhooks/entities/webhook-de import { UploadedFile } from "./infrastructure/file-upload/entities/uploaded-file.entity"; import { FileThumbnail } from "./infrastructure/file-upload/entities/file-thumbnail.entity"; import { FileScanResult } from "./infrastructure/file-upload/entities/file-scan-result.entity"; +// Reconciliation entities +import { ReconciliationAudit } from "./reconciliation/entities/reconciliation-audit.entity"; +import { ReconciliationInvoice } from "./reconciliation/entities/reconciliation-invoice.entity"; +import { StellarTransaction } from "./reconciliation/entities/stellar-transaction.entity"; // Modules – webhooks import { WebhookModule } from "./infrastructure/webhooks/webhook.module"; // Modules – file upload @@ -141,7 +146,7 @@ import { GraphqlGatewayModule } from "./graphql/graphql.module"; throw new Error( `Environment validation failed: ${errors .map((e) => Object.values(e.constraints || {}).join(", ")) - .join(", ")}`, + .join(", ")}` ); } return validatedConfig; @@ -202,9 +207,12 @@ import { GraphqlGatewayModule } from "./graphql/graphql.module"; WebhookEvent, WebhookDelivery, WebhookDeadLetter, - UploadedFile, - FileThumbnail, - FileScanResult, + UploadedFile, + FileThumbnail, + FileScanResult, + ReconciliationAudit, + ReconciliationInvoice, + StellarTransaction, ], synchronize: true, logging: true, @@ -253,6 +261,7 @@ import { GraphqlGatewayModule } from "./graphql/graphql.module"; }), }), BillingModule, + ReconciliationModule, ], controllers: [AppController], @@ -284,7 +293,7 @@ import { GraphqlGatewayModule } from "./graphql/graphql.module"; export class AppModule implements NestModule, OnModuleInit { constructor( @Inject(SubmissionVerifierService) - private readonly verifier: SubmissionVerifierService, + private readonly verifier: SubmissionVerifierService ) {} configure(consumer: MiddlewareConsumer) { @@ -293,7 +302,7 @@ export class AppModule implements NestModule, OnModuleInit { consumer .apply( (req, res, next) => loggingMiddleware.use(req, res, next), - ProfilingMiddleware, + ProfilingMiddleware ) .forRoutes("*"); } diff --git a/src/reconciliation/dto/reconciliation.dto.ts b/src/reconciliation/dto/reconciliation.dto.ts new file mode 100644 index 0000000..cd9c51b --- /dev/null +++ b/src/reconciliation/dto/reconciliation.dto.ts @@ -0,0 +1,100 @@ +import { ApiProperty, ApiPropertyOptional } from "@nestjs/swagger"; +import { + IsDateString, + IsNumberString, + IsObject, + IsOptional, + IsString, + Length, + MaxLength, +} from "class-validator"; + +export class CreateReconciliationInvoiceDto { + @ApiProperty({ example: "INV-2026-0001" }) + @IsString() + @MaxLength(255) + invoiceId: string; + + @ApiProperty({ example: "125.5000000" }) + @IsNumberString() + expectedAmount: string; + + @ApiProperty({ example: "GABC...DEST" }) + @IsString() + @Length(56, 56) + destinationAccount: string; + + @ApiPropertyOptional({ example: "order-0001" }) + @IsOptional() + @IsString() + @MaxLength(128) + paymentReference?: string; + + @ApiPropertyOptional({ example: "XLM" }) + @IsOptional() + @IsString() + @MaxLength(12) + assetCode?: string; + + @ApiPropertyOptional({ type: Object }) + @IsOptional() + @IsObject() + metadata?: Record; +} + +export class IngestStellarTransactionDto { + @ApiProperty({ example: "a transaction hash" }) + @IsString() + @MaxLength(128) + transactionId: string; + + @ApiPropertyOptional({ example: "123456" }) + @IsOptional() + @IsString() + ledger?: string; + + @ApiPropertyOptional({ example: "GABC...SOURCE" }) + @IsOptional() + @IsString() + @Length(56, 56) + sourceAccount?: string; + + @ApiProperty({ example: "GABC...DEST" }) + @IsString() + @Length(56, 56) + destinationAccount: string; + + @ApiProperty({ example: "125.5000000" }) + @IsNumberString() + amount: string; + + @ApiPropertyOptional({ example: "XLM" }) + @IsOptional() + @IsString() + @MaxLength(12) + assetCode?: string; + + @ApiPropertyOptional({ example: "order-0001" }) + @IsOptional() + @IsString() + @MaxLength(128) + memo?: string; + + @ApiPropertyOptional({ example: "2026-08-21T12:00:00.000Z" }) + @IsOptional() + @IsDateString() + observedAt?: string; + + @ApiPropertyOptional({ type: Object }) + @IsOptional() + @IsObject() + rawPayload?: Record; +} + +export class ManualReconciliationDto { + @ApiPropertyOptional({ example: "manual-review-123" }) + @IsOptional() + @IsString() + @MaxLength(255) + note?: string; +} diff --git a/src/reconciliation/entities/reconciliation-audit.entity.ts b/src/reconciliation/entities/reconciliation-audit.entity.ts new file mode 100644 index 0000000..3d5a000 --- /dev/null +++ b/src/reconciliation/entities/reconciliation-audit.entity.ts @@ -0,0 +1,33 @@ +import { Column, Entity, Index } from "typeorm"; +import { BaseEntity } from "../../common/database/entities/base.entity"; + +export enum ReconciliationDecision { + MATCHED = "matched", + PARTIAL = "partial", + UNMATCHED = "unmatched", + FAILED = "failed", + RETRY = "retry", +} + +@Entity("reconciliation_audits") +@Index(["invoiceId", "createdAt"]) +@Index(["transactionId", "createdAt"]) +export class ReconciliationAudit extends BaseEntity { + @Column({ type: "varchar", length: 255, nullable: true }) + invoiceId: string; + + @Column({ type: "varchar", length: 128, nullable: true }) + transactionId: string; + + @Column({ type: "varchar", length: 16 }) + decision: ReconciliationDecision; + + @Column({ type: "text", nullable: true }) + reason: string; + + @Column({ type: "integer", default: 0 }) + attempt: number; + + @Column({ type: "jsonb", nullable: true }) + metadata: Record; +} diff --git a/src/reconciliation/entities/reconciliation-invoice.entity.ts b/src/reconciliation/entities/reconciliation-invoice.entity.ts new file mode 100644 index 0000000..6a3b2b6 --- /dev/null +++ b/src/reconciliation/entities/reconciliation-invoice.entity.ts @@ -0,0 +1,42 @@ +import { Column, Entity, Index } from "typeorm"; +import { BaseEntity } from "../../common/database/entities/base.entity"; + +export enum ReconciliationInvoiceStatus { + OPEN = "open", + PARTIAL = "partial", + PAID = "paid", + FAILED = "failed", +} + +@Entity("reconciliation_invoices") +@Index(["invoiceId"], { unique: true }) +@Index(["status"]) +export class ReconciliationInvoice extends BaseEntity { + @Column({ type: "varchar", length: 255 }) + invoiceId: string; + + @Column({ type: "numeric", precision: 30, scale: 7 }) + expectedAmount: string; + + @Column({ type: "numeric", precision: 30, scale: 7, default: "0" }) + paidAmount: string; + + @Column({ type: "varchar", length: 12, default: "XLM" }) + assetCode: string; + + @Column({ type: "varchar", length: 64 }) + destinationAccount: string; + + @Column({ type: "varchar", length: 128, nullable: true }) + paymentReference: string; + + @Column({ + type: "varchar", + length: 16, + default: ReconciliationInvoiceStatus.OPEN, + }) + status: ReconciliationInvoiceStatus; + + @Column({ type: "jsonb", nullable: true }) + metadata: Record; +} diff --git a/src/reconciliation/entities/stellar-transaction.entity.ts b/src/reconciliation/entities/stellar-transaction.entity.ts new file mode 100644 index 0000000..8d91818 --- /dev/null +++ b/src/reconciliation/entities/stellar-transaction.entity.ts @@ -0,0 +1,55 @@ +import { Column, Entity, Index } from "typeorm"; +import { BaseEntity } from "../../common/database/entities/base.entity"; + +export enum StellarTransactionStatus { + UNMATCHED = "unmatched", + PARTIAL = "partial", + MATCHED = "matched", + FAILED = "failed", +} + +@Entity("stellar_reconciliation_transactions") +@Index(["transactionId"], { unique: true }) +@Index(["status"]) +@Index(["paymentReference"]) +export class StellarTransaction extends BaseEntity { + @Column({ type: "varchar", length: 128 }) + transactionId: string; + + @Column({ type: "bigint", nullable: true }) + ledger: string; + + @Column({ type: "varchar", length: 64, nullable: true }) + sourceAccount: string; + + @Column({ type: "varchar", length: 64 }) + destinationAccount: string; + + @Column({ type: "numeric", precision: 30, scale: 7 }) + amount: string; + + @Column({ type: "varchar", length: 12, default: "XLM" }) + assetCode: string; + + @Column({ type: "varchar", length: 128, nullable: true }) + memo: string; + + @Column({ type: "varchar", length: 128, nullable: true }) + paymentReference: string; + + @Column({ + type: "varchar", + length: 16, + default: StellarTransactionStatus.UNMATCHED, + }) + status: StellarTransactionStatus; + + @Column({ type: "text", nullable: true }) + failureReason: string; + + @Column({ type: "timestamptz", nullable: true }) + observedAt: Date; + + @Column({ type: "jsonb", nullable: true }) + rawPayload: Record; +} diff --git a/src/reconciliation/horizon-polling.service.ts b/src/reconciliation/horizon-polling.service.ts new file mode 100644 index 0000000..aba5b9d --- /dev/null +++ b/src/reconciliation/horizon-polling.service.ts @@ -0,0 +1,13 @@ +import { Injectable } from "@nestjs/common"; +import { Interval } from "@nestjs/schedule"; +import { ReconciliationService } from "./reconciliation.service"; + +@Injectable() +export class HorizonPollingService { + constructor(private readonly reconciliationService: ReconciliationService) {} + + @Interval(60_000) + poll() { + return this.reconciliationService.pollHorizon(); + } +} diff --git a/src/reconciliation/reconciliation.controller.ts b/src/reconciliation/reconciliation.controller.ts new file mode 100644 index 0000000..15a6f91 --- /dev/null +++ b/src/reconciliation/reconciliation.controller.ts @@ -0,0 +1,102 @@ +import { + Body, + Controller, + Get, + Param, + ParseIntPipe, + Post, + Query, + UseGuards, +} from "@nestjs/common"; +import { + ApiBearerAuth, + ApiOperation, + ApiParam, + ApiQuery, + ApiTags, +} from "@nestjs/swagger"; +import { JwtAuthGuard } from "../core/auth/guards/jwt-auth.guard"; +import { AdminTwoFactorGuard } from "../core/auth/guards/admin-two-factor.guard"; +import { RolesGuard } from "../common/guard/roles.guard"; +import { Roles } from "../common/guard/roles.decorator"; +import { Role } from "../common/guard/roles.enum"; +import { + CreateReconciliationInvoiceDto, + IngestStellarTransactionDto, + ManualReconciliationDto, +} from "./dto/reconciliation.dto"; +import { ReconciliationService } from "./reconciliation.service"; + +@ApiTags("Stellar Reconciliation") +@Controller("reconcile/stellar") +export class ReconciliationController { + constructor(private readonly reconciliationService: ReconciliationService) {} + + @Post("invoice") + @ApiOperation({ summary: "Register an invoice for Stellar reconciliation" }) + createInvoice(@Body() dto: CreateReconciliationInvoiceDto) { + return this.reconciliationService.createInvoice(dto); + } + + @Post("transactions") + @ApiOperation({ + summary: "Ingest a confirmed Stellar payment", + description: + "Idempotently records a payment and reconciles it against an invoice by destination, asset, and memo/reference.", + }) + ingestTransaction(@Body() dto: IngestStellarTransactionDto) { + return this.reconciliationService.ingestTransaction(dto); + } + + @Get("tx/:txid") + @ApiParam({ name: "txid", description: "Stellar transaction hash" }) + @ApiOperation({ summary: "Look up a Stellar transaction and its decisions" }) + getTransaction(@Param("txid") transactionId: string) { + return this.reconciliationService.getTransaction(transactionId); + } + + @Get("invoice/:invoiceId") + @ApiParam({ name: "invoiceId", description: "Internal invoice identifier" }) + @ApiOperation({ + summary: "Look up an invoice and its reconciliation audit trail", + }) + getInvoice(@Param("invoiceId") invoiceId: string) { + return this.reconciliationService.getInvoice(invoiceId); + } + + @Post("invoice/:invoiceId/reconcile") + @ApiParam({ name: "invoiceId", description: "Internal invoice identifier" }) + @ApiOperation({ summary: "Retry reconciliation for an invoice" }) + manualReconcile( + @Param("invoiceId") invoiceId: string, + @Body() dto: ManualReconciliationDto + ) { + return this.reconciliationService.manualReconcile(invoiceId, dto); + } + + @Get("admin/unmatched") + @UseGuards(JwtAuthGuard, RolesGuard, AdminTwoFactorGuard) + @Roles(Role.ADMIN) + @ApiBearerAuth() + @ApiOperation({ summary: "List unmatched Stellar transactions" }) + @ApiQuery({ name: "limit", required: false, type: Number, example: 50 }) + listUnmatched( + @Query("limit", new ParseIntPipe({ optional: true })) limit?: number + ) { + return this.reconciliationService.listUnmatched(limit); + } + + @Get("admin/audit") + @UseGuards(JwtAuthGuard, RolesGuard, AdminTwoFactorGuard) + @Roles(Role.ADMIN) + @ApiBearerAuth() + @ApiOperation({ summary: "List reconciliation decisions" }) + @ApiQuery({ name: "invoiceId", required: false }) + @ApiQuery({ name: "transactionId", required: false }) + listAudits( + @Query("invoiceId") invoiceId?: string, + @Query("transactionId") transactionId?: string + ) { + return this.reconciliationService.listAudits(invoiceId, transactionId); + } +} diff --git a/src/reconciliation/reconciliation.module.ts b/src/reconciliation/reconciliation.module.ts new file mode 100644 index 0000000..5b834ec --- /dev/null +++ b/src/reconciliation/reconciliation.module.ts @@ -0,0 +1,22 @@ +import { Module } from "@nestjs/common"; +import { TypeOrmModule } from "@nestjs/typeorm"; +import { ReconciliationController } from "./reconciliation.controller"; +import { ReconciliationService } from "./reconciliation.service"; +import { HorizonPollingService } from "./horizon-polling.service"; +import { ReconciliationAudit } from "./entities/reconciliation-audit.entity"; +import { ReconciliationInvoice } from "./entities/reconciliation-invoice.entity"; +import { StellarTransaction } from "./entities/stellar-transaction.entity"; + +@Module({ + imports: [ + TypeOrmModule.forFeature([ + ReconciliationAudit, + ReconciliationInvoice, + StellarTransaction, + ]), + ], + controllers: [ReconciliationController], + providers: [ReconciliationService, HorizonPollingService], + exports: [ReconciliationService], +}) +export class ReconciliationModule {} diff --git a/src/reconciliation/reconciliation.service.spec.ts b/src/reconciliation/reconciliation.service.spec.ts new file mode 100644 index 0000000..fdb5b4d --- /dev/null +++ b/src/reconciliation/reconciliation.service.spec.ts @@ -0,0 +1,174 @@ +import { ConfigService } from "@nestjs/config"; +import { + ReconciliationAudit, + ReconciliationDecision, +} from "./entities/reconciliation-audit.entity"; +import { + ReconciliationInvoice, + ReconciliationInvoiceStatus, +} from "./entities/reconciliation-invoice.entity"; +import { + StellarTransaction, + StellarTransactionStatus, +} from "./entities/stellar-transaction.entity"; +import { ReconciliationService } from "./reconciliation.service"; + +function repository() { + return { + findOne: jest.fn(), + find: jest.fn(), + create: jest.fn((value: Partial) => value as T), + save: jest.fn(async (value: T) => value), + }; +} + +describe("ReconciliationService", () => { + const invoiceRepo = repository(); + const transactionRepo = repository(); + const auditRepo = repository(); + let service: ReconciliationService; + + beforeEach(() => { + jest.clearAllMocks(); + service = new ReconciliationService( + invoiceRepo as any, + transactionRepo as any, + auditRepo as any, + new ConfigService() + ); + }); + + it("creates an open invoice with normalized decimal amounts", async () => { + invoiceRepo.findOne.mockResolvedValue(undefined); + transactionRepo.find.mockResolvedValue([]); + const invoice = await service.createInvoice({ + invoiceId: "INV-1", + expectedAmount: "10.5", + destinationAccount: "G".repeat(56), + paymentReference: "order-1", + }); + + expect(invoice.expectedAmount).toBe("10.5000000"); + expect(invoice.paidAmount).toBe("0.0000000"); + expect(invoice.status).toBe(ReconciliationInvoiceStatus.OPEN); + expect(invoiceRepo.save).toHaveBeenCalled(); + }); + + it("marks an invoice paid when an exact matching transaction arrives", async () => { + const invoice: ReconciliationInvoice = { + invoiceId: "INV-1", + expectedAmount: "10.0000000", + paidAmount: "0.0000000", + assetCode: "XLM", + destinationAccount: "G".repeat(56), + paymentReference: "order-1", + status: ReconciliationInvoiceStatus.OPEN, + } as ReconciliationInvoice; + invoiceRepo.findOne.mockResolvedValue(invoice); + transactionRepo.findOne.mockResolvedValue(undefined); + auditRepo.create.mockImplementation( + (value) => value as ReconciliationAudit + ); + + const transaction = await service.ingestTransaction({ + transactionId: "tx-1", + destinationAccount: invoice.destinationAccount, + amount: "10", + memo: "order-1", + }); + + expect(transaction.status).toBe(StellarTransactionStatus.MATCHED); + expect(invoice.status).toBe(ReconciliationInvoiceStatus.PAID); + expect(invoice.paidAmount).toBe("10.0000000"); + expect(auditRepo.create).toHaveBeenCalledWith( + expect.objectContaining({ decision: ReconciliationDecision.MATCHED }) + ); + }); + + it("marks an invoice partial until the expected amount is reached", async () => { + const invoice: ReconciliationInvoice = { + invoiceId: "INV-2", + expectedAmount: "10.0000000", + paidAmount: "4.0000000", + assetCode: "XLM", + destinationAccount: "G".repeat(56), + paymentReference: "order-2", + status: ReconciliationInvoiceStatus.OPEN, + } as ReconciliationInvoice; + invoiceRepo.findOne.mockResolvedValue(invoice); + transactionRepo.findOne.mockResolvedValue(undefined); + auditRepo.create.mockImplementation( + (value) => value as ReconciliationAudit + ); + + const transaction = await service.ingestTransaction({ + transactionId: "tx-2", + destinationAccount: invoice.destinationAccount, + amount: "3.5", + memo: "order-2", + }); + + expect(transaction.status).toBe(StellarTransactionStatus.PARTIAL); + expect(invoice.status).toBe(ReconciliationInvoiceStatus.PARTIAL); + expect(invoice.paidAmount).toBe("7.5000000"); + expect(auditRepo.create).toHaveBeenCalledWith( + expect.objectContaining({ decision: ReconciliationDecision.PARTIAL }) + ); + }); + + it("keeps transactions unmatched when no invoice reference matches", async () => { + invoiceRepo.findOne.mockResolvedValue(undefined); + transactionRepo.findOne.mockResolvedValue(undefined); + auditRepo.create.mockImplementation( + (value) => value as ReconciliationAudit + ); + + const transaction = await service.ingestTransaction({ + transactionId: "tx-unmatched", + destinationAccount: "G".repeat(56), + amount: "1", + memo: "unknown-order", + }); + + expect(transaction.status).toBe(StellarTransactionStatus.UNMATCHED); + expect(auditRepo.create).toHaveBeenCalledWith( + expect.objectContaining({ decision: ReconciliationDecision.UNMATCHED }) + ); + }); + + it("ignores duplicate transaction events and records an idempotent retry audit", async () => { + const existing = { + transactionId: "tx-duplicate", + status: StellarTransactionStatus.MATCHED, + } as StellarTransaction; + transactionRepo.findOne.mockResolvedValue(existing); + auditRepo.create.mockImplementation( + (value) => value as ReconciliationAudit + ); + + const result = await service.ingestTransaction({ + transactionId: "tx-duplicate", + destinationAccount: "G".repeat(56), + amount: "1", + }); + + expect(result).toBe(existing); + expect(transactionRepo.save).not.toHaveBeenCalled(); + expect(auditRepo.create).toHaveBeenCalledWith( + expect.objectContaining({ + decision: ReconciliationDecision.RETRY, + metadata: { idempotent: true }, + }) + ); + }); + + it("bounds the unmatched dashboard result to 200 records", async () => { + transactionRepo.find.mockResolvedValue([]); + + await service.listUnmatched(9999); + + expect(transactionRepo.find).toHaveBeenCalledWith( + expect.objectContaining({ take: 200 }) + ); + }); +}); diff --git a/src/reconciliation/reconciliation.service.ts b/src/reconciliation/reconciliation.service.ts new file mode 100644 index 0000000..1e6f14c --- /dev/null +++ b/src/reconciliation/reconciliation.service.ts @@ -0,0 +1,375 @@ +import { Injectable, Logger, NotFoundException } from "@nestjs/common"; +import { ConfigService } from "@nestjs/config"; +import { InjectRepository } from "@nestjs/typeorm"; +import { In, Repository } from "typeorm"; +import { + CreateReconciliationInvoiceDto, + IngestStellarTransactionDto, + ManualReconciliationDto, +} from "./dto/reconciliation.dto"; +import { + ReconciliationInvoice, + ReconciliationInvoiceStatus, +} from "./entities/reconciliation-invoice.entity"; +import { + ReconciliationAudit, + ReconciliationDecision, +} from "./entities/reconciliation-audit.entity"; +import { + StellarTransaction, + StellarTransactionStatus, +} from "./entities/stellar-transaction.entity"; + +const SCALE = 7n; +const SCALE_FACTOR = 10n ** SCALE; + +@Injectable() +export class ReconciliationService { + private readonly logger = new Logger(ReconciliationService.name); + private horizonCursor: string | undefined; + + constructor( + @InjectRepository(ReconciliationInvoice) + private readonly invoiceRepo: Repository, + @InjectRepository(StellarTransaction) + private readonly transactionRepo: Repository, + @InjectRepository(ReconciliationAudit) + private readonly auditRepo: Repository, + private readonly configService: ConfigService + ) {} + + async createInvoice( + dto: CreateReconciliationInvoiceDto + ): Promise { + const existing = await this.invoiceRepo.findOne({ + where: { invoiceId: dto.invoiceId }, + }); + if (existing) return existing; + + const invoice = this.invoiceRepo.create({ + invoiceId: dto.invoiceId, + expectedAmount: this.normalizeAmount(dto.expectedAmount), + paidAmount: "0.0000000", + assetCode: (dto.assetCode ?? "XLM").toUpperCase(), + destinationAccount: dto.destinationAccount, + paymentReference: dto.paymentReference, + metadata: dto.metadata, + status: ReconciliationInvoiceStatus.OPEN, + }); + const saved = await this.invoiceRepo.save(invoice); + const candidates = await this.transactionRepo.find({ + where: { + destinationAccount: saved.destinationAccount, + assetCode: saved.assetCode, + status: StellarTransactionStatus.UNMATCHED, + }, + order: { observedAt: "ASC" }, + }); + for (const transaction of candidates) { + if (this.matchesInvoice(transaction, saved)) { + await this.reconcileTransaction( + transaction, + "Invoice registered after payment" + ); + await this.transactionRepo.save(transaction); + } + } + return saved; + } + + async ingestTransaction( + dto: IngestStellarTransactionDto + ): Promise { + const existing = await this.transactionRepo.findOne({ + where: { transactionId: dto.transactionId }, + }); + if (existing) { + await this.writeAudit({ + transactionId: existing.transactionId, + decision: ReconciliationDecision.RETRY, + reason: "Duplicate transaction event ignored", + attempt: 0, + metadata: { idempotent: true }, + }); + return existing; + } + + const transaction = this.transactionRepo.create({ + transactionId: dto.transactionId, + ledger: dto.ledger, + sourceAccount: dto.sourceAccount, + destinationAccount: dto.destinationAccount, + amount: this.normalizeAmount(dto.amount), + assetCode: (dto.assetCode ?? "XLM").toUpperCase(), + memo: dto.memo, + paymentReference: dto.memo, + observedAt: dto.observedAt ? new Date(dto.observedAt) : new Date(), + rawPayload: dto.rawPayload, + status: StellarTransactionStatus.UNMATCHED, + }); + const saved = await this.transactionRepo.save(transaction); + try { + await this.reconcileTransaction(saved); + } catch (error) { + saved.status = StellarTransactionStatus.FAILED; + saved.failureReason = + error instanceof Error ? error.message : String(error); + await this.writeAudit({ + transactionId: saved.transactionId, + decision: ReconciliationDecision.FAILED, + reason: saved.failureReason, + attempt: 1, + metadata: { retryable: true }, + }); + } + return this.transactionRepo.save(saved); + } + + async getTransaction(transactionId: string) { + const transaction = await this.transactionRepo.findOne({ + where: { transactionId }, + }); + if (!transaction) { + throw new NotFoundException( + `Stellar transaction ${transactionId} not found` + ); + } + + const audits = await this.auditRepo.find({ + where: { transactionId }, + order: { createdAt: "DESC" }, + }); + return { transaction, audits }; + } + + async getInvoice(invoiceId: string) { + const invoice = await this.invoiceRepo.findOne({ where: { invoiceId } }); + if (!invoice) { + throw new NotFoundException(`Invoice ${invoiceId} not found`); + } + + const audits = await this.auditRepo.find({ + where: { invoiceId }, + order: { createdAt: "DESC" }, + }); + return { invoice, audits }; + } + + async listUnmatched(limit = 50): Promise { + return this.transactionRepo.find({ + where: { status: StellarTransactionStatus.UNMATCHED }, + order: { createdAt: "DESC" }, + take: Math.min(Math.max(limit, 1), 200), + }); + } + + async listAudits(invoiceId?: string, transactionId?: string) { + const where: Record = {}; + if (invoiceId) where.invoiceId = invoiceId; + if (transactionId) where.transactionId = transactionId; + return this.auditRepo.find({ + where, + order: { createdAt: "DESC" }, + take: 200, + }); + } + + async manualReconcile(invoiceId: string, dto: ManualReconciliationDto) { + const invoice = await this.invoiceRepo.findOne({ where: { invoiceId } }); + if (!invoice) throw new NotFoundException(`Invoice ${invoiceId} not found`); + + const candidates = await this.transactionRepo.find({ + where: { + destinationAccount: invoice.destinationAccount, + assetCode: invoice.assetCode, + status: In([ + StellarTransactionStatus.UNMATCHED, + StellarTransactionStatus.PARTIAL, + ]), + }, + order: { observedAt: "ASC" }, + }); + + for (const transaction of candidates) { + if (this.matchesInvoice(transaction, invoice)) { + await this.reconcileTransaction( + transaction, + dto.note ?? "Manual retry" + ); + await this.transactionRepo.save(transaction); + } + } + + return this.getInvoice(invoiceId); + } + + async pollHorizon(): Promise<{ skipped?: boolean; ingested: number }> { + const account = this.configService.get( + "STELLAR_RECONCILIATION_ACCOUNT" + ); + if (!account) return { skipped: true, ingested: 0 }; + + const baseUrl = this.configService.get( + "STELLAR_HORIZON_URL", + "https://horizon-testnet.stellar.org" + ); + const params = new URLSearchParams({ order: "asc", limit: "200" }); + if (this.horizonCursor) params.set("cursor", this.horizonCursor); + + try { + const response = await fetch( + `${baseUrl.replace(/\/$/, "")}/accounts/${account}/payments?${params}` + ); + if (!response.ok) { + throw new Error(`Horizon returned HTTP ${response.status}`); + } + + const payload = (await response.json()) as { + _embedded?: { records?: Array> }; + }; + const records = payload._embedded?.records ?? []; + let ingested = 0; + for (const record of records) { + if ( + record.type !== "payment" || + typeof record.transaction_hash !== "string" + ) { + if (typeof record.paging_token === "string") { + this.horizonCursor = record.paging_token; + } + continue; + } + + await this.ingestTransaction({ + transactionId: record.transaction_hash, + ledger: + typeof record.ledger === "number" || + typeof record.ledger === "string" + ? String(record.ledger) + : undefined, + sourceAccount: + typeof record.from === "string" ? record.from : undefined, + destinationAccount: String(record.to ?? ""), + amount: String(record.amount ?? "0"), + assetCode: + record.asset_type === "native" + ? "XLM" + : String(record.asset_code ?? "XLM"), + memo: typeof record.memo === "string" ? record.memo : undefined, + rawPayload: record, + }); + ingested += 1; + if (typeof record.paging_token === "string") { + this.horizonCursor = record.paging_token; + } + } + return { ingested }; + } catch (error) { + this.logger.warn(`Horizon polling failed: ${error.message}`); + await this.writeAudit({ + decision: ReconciliationDecision.RETRY, + reason: error.message, + attempt: 1, + metadata: { source: "horizon", retryable: true }, + }); + return { ingested: 0 }; + } + } + + private async reconcileTransaction( + transaction: StellarTransaction, + reason?: string + ): Promise { + const invoice = await this.invoiceRepo.findOne({ + where: { destinationAccount: transaction.destinationAccount }, + order: { createdAt: "ASC" }, + }); + + if (!invoice || !this.matchesInvoice(transaction, invoice)) { + transaction.status = StellarTransactionStatus.UNMATCHED; + await this.writeAudit({ + invoiceId: invoice?.invoiceId, + transactionId: transaction.transactionId, + decision: ReconciliationDecision.UNMATCHED, + reason: + reason ?? "No invoice matched destination, asset, and reference", + attempt: 0, + metadata: { destinationAccount: transaction.destinationAccount }, + }); + return; + } + + const paidUnits = + this.toUnits(invoice.paidAmount) + this.toUnits(transaction.amount); + const expectedUnits = this.toUnits(invoice.expectedAmount); + const isPaid = paidUnits >= expectedUnits; + invoice.paidAmount = this.fromUnits(paidUnits); + invoice.status = isPaid + ? ReconciliationInvoiceStatus.PAID + : ReconciliationInvoiceStatus.PARTIAL; + transaction.status = isPaid + ? StellarTransactionStatus.MATCHED + : StellarTransactionStatus.PARTIAL; + await this.invoiceRepo.save(invoice); + await this.writeAudit({ + invoiceId: invoice.invoiceId, + transactionId: transaction.transactionId, + decision: isPaid + ? ReconciliationDecision.MATCHED + : ReconciliationDecision.PARTIAL, + reason: + reason ?? (isPaid ? "Invoice fully paid" : "Invoice partially paid"), + attempt: 0, + metadata: { + expectedAmount: invoice.expectedAmount, + paidAmount: invoice.paidAmount, + assetCode: invoice.assetCode, + }, + }); + } + + private matchesInvoice( + transaction: StellarTransaction, + invoice: ReconciliationInvoice + ): boolean { + if (transaction.destinationAccount !== invoice.destinationAccount) + return false; + if (transaction.assetCode !== invoice.assetCode) return false; + if ( + invoice.paymentReference && + invoice.paymentReference !== transaction.paymentReference + ) { + return false; + } + return true; + } + + private async writeAudit(values: Partial) { + const audit = this.auditRepo.create(values); + return this.auditRepo.save(audit); + } + + private normalizeAmount(value: string): string { + return this.fromUnits(this.toUnits(value)); + } + + private toUnits(value: string): bigint { + const normalized = String(value).trim(); + if (!/^\d+(\.\d+)?$/.test(normalized)) { + throw new Error(`Invalid Stellar amount: ${value}`); + } + const [whole, fraction = ""] = normalized.split("."); + return ( + BigInt(whole) * SCALE_FACTOR + + BigInt(fraction.padEnd(Number(SCALE), "0").slice(0, Number(SCALE))) + ); + } + + private fromUnits(units: bigint): string { + const whole = units / SCALE_FACTOR; + const fraction = (units % SCALE_FACTOR) + .toString() + .padStart(Number(SCALE), "0"); + return `${whole}.${fraction}`; + } +}