Feature/distributed transaction coordinator

src/app.module.ts

@@ -28,6 +28,7 @@ import { ComplianceModule } from './modules/compliance/compliance.module';
 import { RateLimitingModule } from './modules/rate-limiting/rate-limiting.module';
 import { PortfolioModule } from './modules/portfolio/portfolio.module';
 import { WebhookModule } from './modules/webhooks/webhook.module';
+import { TransactionCoordinatorModule } from './modules/transaction-coordinator/transaction-coordinator.module';
 
 @Module({
   imports: [
@@ -59,6 +60,7 @@ import { WebhookModule } from './modules/webhooks/webhook.module';
     RateLimitingModule,
     PortfolioModule,
     WebhookModule,
+    TransactionCoordinatorModule,
   ],
   controllers: [AppController],
   providers: [AppService],

src/modules/transaction-coordinator/dto/batch-response.dto.ts

@@ -0,0 +1,33 @@
+import { ApiProperty } from '@nestjs/swagger';
+import { TransactionBatch } from '../entities/transaction-batch.entity';
+import { BatchLeg } from '../entities/batch-leg.entity';
+
+/**
+ * Detailed response for a transaction batch including its legs and audit trail.
+ */
+export class BatchDetailResponse {
+  @ApiProperty({ description: 'The transaction batch' })
+  batch: TransactionBatch;
+
+  @ApiProperty({
+    description: 'The swap legs in this batch',
+    type: [BatchLeg],
+  })
+  legs: BatchLeg[];
+
+  @ApiProperty({
+    description: 'Dependency graph of the legs',
+  })
+  dependencyGraph: Record<string, string[]>;
+
+  @ApiProperty({
+    description: 'Execution plan (groups of legs that can run in parallel)',
+    type: [[String]],
+  })
+  executionPlan: string[][];
+
+  @ApiProperty({
+    description: 'Estimated total duration in milliseconds',
+  })
+  estimatedDurationMs: number;
+}

src/modules/transaction-coordinator/dto/create-batch.dto.ts

@@ -0,0 +1,160 @@
+import {
+  IsString,
+  IsArray,
+  ValidateNested,
+  IsOptional,
+  IsNumber,
+  IsBoolean,
+  Min,
+  IsEnum,
+  ValidateIf,
+  IsObject,
+} from 'class-validator';
+import { Type } from 'class-transformer';
+import { ApiProperty, ApiPropertyOptional } from '@nestjs/swagger';
+
+export enum ConditionType {
+  PRICE_GT = 'price_gt',
+  PRICE_LT = 'price_lt',
+  PRICE_GTE = 'price_gte',
+  PRICE_LTE = 'price_lte',
+  AMOUNT_GT = 'amount_gt',
+  AMOUNT_LT = 'amount_lt',
+}
+
+export class ConditionalLegDto {
+  @ApiPropertyOptional({ description: 'Type of condition to evaluate' })
+  @IsOptional()
+  @IsEnum(ConditionType)
+  conditionType?: ConditionType;
+
+  @ApiPropertyOptional({
+    description:
+      'Condition expression (e.g., "1.05" for price_gt, meaning execute only if price > 1.05)',
+  })
+  @IsOptional()
+  @IsString()
+  conditionExpression?: string;
+}
+
+export class SwapLegDto {
+  @ApiProperty({ description: 'Soroban contract ID to invoke' })
+  @IsString()
+  contractId: string;
+
+  @ApiProperty({ description: 'Contract method to invoke' })
+  @IsString()
+  method: string;
+
+  @ApiPropertyOptional({
+    description: 'Arguments for the contract invocation',
+  })
+  @IsOptional()
+  @IsObject()
+  args?: Record<string, any>;
+
+  @ApiProperty({ description: 'Source asset code for the swap' })
+  @IsString()
+  sourceAssetCode: string;
+
+  @ApiPropertyOptional({ description: 'Source asset issuer (null for native)' })
+  @IsOptional()
+  @IsString()
+  sourceAssetIssuer?: string;
+
+  @ApiProperty({ description: 'Destination asset code for the swap' })
+  @IsString()
+  destAssetCode: string;
+
+  @ApiPropertyOptional({
+    description: 'Destination asset issuer (null for native)',
+  })
+  @IsOptional()
+  @IsString()
+  destAssetIssuer?: string;
+
+  @ApiProperty({ description: 'Amount to swap' })
+  @IsNumber()
+  @Min(0.0000001)
+  amount: number;
+
+  @ApiProperty({
+    description: 'Minimum acceptable output amount (slippage protection)',
+  })
+  @IsNumber()
+  @Min(0)
+  minAmountOut: number;
+
+  @ApiPropertyOptional({
+    description: 'Maximum acceptable output amount (price bounds)',
+  })
+  @IsOptional()
+  @IsNumber()
+  @Min(0)
+  maxAmountOut?: number;
+
+  @ApiPropertyOptional({
+    description:
+      'IDs of legs that must complete before this one (by order index)',
+    type: [Number],
+  })
+  @IsOptional()
+  @IsArray()
+  @IsNumber({}, { each: true })
+  dependencies?: number[];
+
+  @ApiPropertyOptional({ description: 'Whether this leg has a conditional' })
+  @IsOptional()
+  @IsBoolean()
+  isConditional?: boolean;
+
+  @ApiPropertyOptional({
+    description: 'Conditional execution details',
+    type: ConditionalLegDto,
+  })
+  @IsOptional()
+  @ValidateNested()
+  @Type(() => ConditionalLegDto)
+  @ValidateIf((obj) => obj.isConditional === true)
+  conditional?: ConditionalLegDto;
+}
+
+export class CreateBatchDto {
+  @ApiProperty({ description: 'Human-readable name for this batch' })
+  @IsString()
+  name: string;
+
+  @ApiProperty({
+    description: 'Swap legs to execute atomically',
+    type: [SwapLegDto],
+    minItems: 1,
+  })
+  @IsArray()
+  @ValidateNested({ each: true })
+  @Type(() => SwapLegDto)
+  legs: SwapLegDto[];
+
+  @ApiPropertyOptional({
+    description: 'Maximum time allowed for the entire batch (ms)',
+    default: 30000,
+  })
+  @IsOptional()
+  @IsNumber()
+  @Min(1000)
+  timeoutMs?: number;
+
+  @ApiPropertyOptional({
+    description: 'Maximum number of retry attempts for transient failures',
+    default: 3,
+  })
+  @IsOptional()
+  @IsNumber()
+  @Min(0)
+  @IsNumber()
+  maxRetries?: number;
+
+  @ApiPropertyOptional({ description: 'Optional metadata for the batch' })
+  @IsOptional()
+  @IsObject()
+  metadata?: Record<string, any>;
+}

src/modules/transaction-coordinator/dto/query-batch.dto.ts

@@ -0,0 +1,21 @@
+import { ApiPropertyOptional } from '@nestjs/swagger';
+import { IsOptional, IsEnum, IsUUID, IsString } from 'class-validator';
+import { PaginationQueryDto } from '@app/common';
+import { BatchStatus } from '../entities/transaction-batch.entity';
+
+export class QueryBatchDto extends PaginationQueryDto {
+  @ApiPropertyOptional({ description: 'Filter by batch status' })
+  @IsOptional()
+  @IsEnum(BatchStatus)
+  status?: BatchStatus;
+
+  @ApiPropertyOptional({ description: 'Filter by user ID' })
+  @IsOptional()
+  @IsUUID()
+  userId?: string;
+
+  @ApiPropertyOptional({ description: 'Search by batch name' })
+  @IsOptional()
+  @IsString()
+  name?: string;
+}

src/modules/transaction-coordinator/entities/batch-audit-log.entity.ts

@@ -0,0 +1,69 @@
+import { Column, Entity, Index } from 'typeorm';
+import { BaseEntity } from '@app/common';
+
+export enum BatchAuditAction {
+  BATCH_CREATED = 'batch_created',
+  BATCH_PREPARE_STARTED = 'batch_prepare_started',
+  BATCH_PREPARED = 'batch_prepared',
+  BATCH_COMMIT_STARTED = 'batch_commit_started',
+  BATCH_COMMITTED = 'batch_committed',
+  BATCH_ROLLBACK_STARTED = 'batch_rollback_started',
+  BATCH_ROLLED_BACK = 'batch_rolled_back',
+  BATCH_FAILED = 'batch_failed',
+  BATCH_EXPIRED = 'batch_expired',
+  LEG_PREPARED = 'leg_prepared',
+  LEG_EXECUTED = 'leg_executed',
+  LEG_ROLLED_BACK = 'leg_rolled_back',
+  LEG_FAILED = 'leg_failed',
+  LEG_RETRIED = 'leg_retried',
+  CONDITION_EVALUATED = 'condition_evaluated',
+  CONSISTENCY_CHECK_PASSED = 'consistency_check_passed',
+  CONSISTENCY_CHECK_FAILED = 'consistency_check_failed',
+  PARTIAL_FILL_RECOVERY = 'partial_fill_recovery',
+}
+
+/**
+ * Complete audit trail for the transaction coordinator.
+ *
+ * Every state transition in the batch lifecycle is recorded here for
+ * dispute resolution and debugging.
+ */
+@Entity('batch_audit_logs')
+@Index('IDX_audit_batch_id', ['batchId'])
+@Index('IDX_audit_action', ['action'])
+@Index('IDX_audit_created_at', ['createdAt'])
+export class BatchAuditLog extends BaseEntity {
+  @Index()
+  @Column({ type: 'uuid' })
+  batchId: string;
+
+  /** Leg ID if this audit entry is leg-specific (null for batch-level) */
+  @Column({ type: 'uuid', nullable: true })
+  legId?: string | null;
+
+  @Column({
+    type: 'enum',
+    enum: BatchAuditAction,
+  })
+  action: BatchAuditAction;
+
+  /** The actor who triggered this action (userId or 'system') */
+  @Column({ type: 'varchar' })
+  actorId: string;
+
+  /** State before the action */
+  @Column({ type: 'jsonb', nullable: true })
+  previousState?: Record<string, any> | null;
+
+  /** State after the action */
+  @Column({ type: 'jsonb', nullable: true })
+  newState?: Record<string, any> | null;
+
+  /** Additional context (errors, gas estimates, timing, etc.) */
+  @Column({ type: 'jsonb', nullable: true })
+  metadata?: Record<string, any> | null;
+
+  /** Duration of the operation in milliseconds */
+  @Column({ type: 'int', nullable: true })
+  durationMs?: number | null;
+}