diff --git a/prisma/schema/migrations/20260830000000_add_soroban_wal/migration.sql b/prisma/schema/migrations/20260830000000_add_soroban_wal/migration.sql new file mode 100644 index 0000000..6ed0089 --- /dev/null +++ b/prisma/schema/migrations/20260830000000_add_soroban_wal/migration.sql @@ -0,0 +1,29 @@ +CREATE TYPE "SorobanWALOperation" AS ENUM ('buy', 'sell'); +CREATE TYPE "SorobanWALState" AS ENUM ('PENDING', 'SUBMITTED', 'CONFIRMED', 'FAILED', 'ROLLED_BACK'); + +CREATE TABLE "soroban_wal_entries" ( + "id" TEXT NOT NULL, + "idempotencyKey" TEXT NOT NULL, + "operation" "SorobanWALOperation" NOT NULL, + "wallet" TEXT NOT NULL, + "creatorWallet" TEXT NOT NULL, + "amount" DECIMAL(65,30) NOT NULL, + "expectedSupplyBefore" DECIMAL(65,30) NOT NULL, + "xdrPayload" TEXT NOT NULL, + "state" "SorobanWALState" NOT NULL DEFAULT 'PENDING', + "txHash" TEXT, + "submittedAt" TIMESTAMP(3), + "confirmedAt" TIMESTAMP(3), + "error" TEXT, + "createdAt" TIMESTAMP(3) NOT NULL DEFAULT CURRENT_TIMESTAMP, + "updatedAt" TIMESTAMP(3) NOT NULL, + + CONSTRAINT "soroban_wal_entries_pkey" PRIMARY KEY ("id") +); + +CREATE UNIQUE INDEX "soroban_wal_entries_idempotencyKey_key" + ON "soroban_wal_entries"("idempotencyKey"); +CREATE UNIQUE INDEX "soroban_wal_entries_txHash_key" + ON "soroban_wal_entries"("txHash"); +CREATE INDEX "soroban_wal_entries_state_createdAt_idx" + ON "soroban_wal_entries"("state", "createdAt"); diff --git a/prisma/schema/soroban-wal.prisma b/prisma/schema/soroban-wal.prisma new file mode 100644 index 0000000..f961f37 --- /dev/null +++ b/prisma/schema/soroban-wal.prisma @@ -0,0 +1,33 @@ +enum SorobanWALOperation { + BUY @map("buy") + SELL @map("sell") +} + +enum SorobanWALState { + PENDING + SUBMITTED + CONFIRMED + FAILED + ROLLED_BACK +} + +model SorobanWALEntry { + id String @id @default(cuid()) + idempotencyKey String @unique + operation SorobanWALOperation + wallet String + creatorWallet String + amount Decimal + expectedSupplyBefore Decimal + xdrPayload String @db.Text + state SorobanWALState @default(PENDING) + txHash String? @unique + submittedAt DateTime? + confirmedAt DateTime? + error String? @db.Text + createdAt DateTime @default(now()) + updatedAt DateTime @updatedAt + + @@index([state, createdAt]) + @@map("soroban_wal_entries") +} diff --git a/src/config.schema.ts b/src/config.schema.ts index 18abee4..afa5ab2 100644 --- a/src/config.schema.ts +++ b/src/config.schema.ts @@ -1,302 +1,302 @@ -import { z } from 'zod'; - -/** - * Centralized Zod schema for all environment variables. - * - * Configuration Source Precedence (highest to lowest): - * 1. Environment Variables (.env file or system environment) - * 2. Schema Defaults (defined below with .default()) - * 3. Validation Failure (startup fails if required field missing) - * - * Extracted into its own module so tests can validate the schema - * in isolation (via `.safeParse()`) without triggering the eager - * `envSchema.parse(process.env)` side-effect in `config.ts`. - * - * See docs/configuration.md for complete documentation. - * See docs/CONFIG_SOURCE_PRECEDENCE.md for visual reference. - */ -/** - * Helper to correctly coerce boolean strings from environment variables. - * Zod's default z.coerce.boolean() returns true for any non-empty string, - * including "false" and "0", which is usually not what we want for .env files. - */ -const booleanCoerce = z.preprocess(val => { - if (typeof val === 'string') { - const lower = val.toLowerCase(); - if (lower === 'true' || lower === '1') return true; - if (lower === 'false' || lower === '0') return false; - } - return val; -}, z.coerce.boolean()); - -const optionalNonEmptyString = z.preprocess(val => { - if (typeof val === 'string' && val.trim().length === 0) { - return undefined; - } - - return val; -}, z.string().min(1).optional()); - -export const envSchema = z - .object({ - PORT: z.coerce.number().default(3000), - MODE: z - .enum(['development', 'production', 'test']) - .default('development'), - DATABASE_URL: z - .string() - .min(1, 'DATABASE_URL is required in the environment variables'), - NODE_ID: z.string().default('node-local'), - - GMAIL_USER: z.string(), - GMAIL_APP_PASSWORD: z.string(), - // Google OAuth - GOOGLE_CLIENT_ID: z - .string() - .min(1, 'GOOGLE_CLIENT_ID is required for Google OAuth'), - GOOGLE_CLIENT_SECRET: z - .string() - .min(1, 'GOOGLE_CLIENT_SECRET is required for Google OAuth'), - - // URLs - BACKEND_URL: z.string().url(), - FRONTEND_URL: z - .string() - .url('FRONTEND_URL must be a valid URL') - .min(1, 'FRONTEND_URL is required'), - - // Cloudinary - CLOUDINARY_CLOUD_NAME: z - .string() - .min(1, 'CLOUDINARY_CLOUD_NAME is required for image uploads'), - CLOUDINARY_API_KEY: z - .string() - .min(1, 'CLOUDINARY_API_KEY is required for image uploads'), - CLOUDINARY_API_SECRET: z - .string() - .min(1, 'CLOUDINARY_API_SECRET is required for image uploads'), - - PAYSTACK_SECRET_KEY: z - .string() - .min(1, 'PAYSTACK_SECRET_KEY is required for payment processing'), - PAYSTACK_PUBLIC_KEY: optionalNonEmptyString, - ENABLE_RESPONSE_TIMING: booleanCoerce.default(true), - API_VERSION: z.string().default('1.0.0'), - ENABLE_API_VERSION_HEADER: booleanCoerce.default(true), - ENABLE_SCHEMA_VERSION_HEADER: booleanCoerce.default(true), - ENABLE_REQUEST_LOGGING: booleanCoerce.default(true), - DB_QUERY_TIMEOUT_MS: z.coerce.number().default(5000), - DB_POOL_WAIT_WARN_MS: z.coerce.number().int().positive().default(2000), - DB_POOL_WAIT_ERROR_MS: z.coerce.number().int().positive().default(5000), - WEBHOOK_MAX_PER_CREATOR: z.coerce.number().int().positive().default(5), - WEBHOOK_RETRY_MAX_ATTEMPTS: z.coerce.number().int().positive().default(3), - - APP_SECRET: z - .string() - .min(32, 'APP_SECRET should be at least 32 characters') - .default('accesslayer_default_development_secret_key_32_bytes_long'), - - // JWT auth - JWT_SECRET: z - .string() - .min(32, 'JWT_SECRET should be at least 32 characters') - .default('accesslayer_default_development_jwt_secret_key_32_bytes'), - JWT_ISSUER: z.string().default('accesslayer-server'), - JWT_EXPIRES_IN: z.string().default('15m'), - JWT_ACCESS_TOKEN_TTL_SECONDS: z.coerce - .number() - .int() - .positive() - .default(900), - - // Redis cache - REDIS_URL: z.string().default('redis://localhost:6379'), - ENABLE_REDIS_CACHE: booleanCoerce.default(true), - - // Key trade lockup - LOCKUP_DURATION_SECONDS: z.coerce.number().int().nonnegative().default(0), - - // Leaderboard volume - LEADERBOARD_VOLUME_WINDOW_DAYS: z.coerce - .number() - .int() - .positive() - .default(7), - LEADERBOARD_VOLUME_CACHE_TTL_SECONDS: z.coerce - .number() - .int() - .positive() - .default(300), - - INDEXER_JITTER_FACTOR: z.coerce.number().min(0).max(1).default(0.1), - BACKGROUND_JOB_LOCK_TTL_MS: z.coerce - .number() - .int() - .positive() - .default(300000), - SLOW_QUERY_THRESHOLD_MS: z.coerce.number().int().positive().default(500), - CREATOR_LIST_SLOW_QUERY_THRESHOLD_MS: z.coerce - .number() - .int() - .positive() - .default(500), - INDEXER_CURSOR_STALE_AGE_WARNING_MS: z.coerce - .number() - .int() - .positive() - .default(300000), - INDEXER_HEARTBEAT_STALE_THRESHOLD_MS: z.coerce - .number() - .positive() - .default(300000), - - // Indexer feature flags - ENABLE_INDEXER_DEDUPE: booleanCoerce.default(true), - ENABLE_INDEXER_DLQ: booleanCoerce.default(true), - ENABLE_INDEXER_CURSOR_STALENESS_WARNING: booleanCoerce.default(true), - - // Stellar auth — optional server keypair secret used for SEP-10 challenge - // signing. When absent the server falls back to an ephemeral random keypair. - STELLAR_AUTH_SECRET: optionalNonEmptyString, - - // Stellar network - STELLAR_NETWORK: z - .enum(['testnet', 'mainnet'], { - message: - 'STELLAR_NETWORK must be "testnet" or "mainnet". Set STELLAR_NETWORK in your .env file.', - }) - .default('testnet'), - STELLAR_HORIZON_URL: z - .string() - .url( - 'STELLAR_HORIZON_URL must be a valid URL (e.g. https://horizon-testnet.stellar.org)' - ) - .default('https://horizon-testnet.stellar.org'), - STELLAR_SOROBAN_RPC_URL: z - .string() - .url( - 'STELLAR_SOROBAN_RPC_URL must be a valid URL (e.g. https://soroban-testnet.stellar.org)' - ) - .default('https://soroban-testnet.stellar.org'), - - // Ownership snapshot cleanup job - OWNERSHIP_SNAPSHOT_TABLE_NAME: z - .string() - .min(1) - .default('creator_ownership_snapshots'), - OWNERSHIP_SNAPSHOT_CLEANUP_DRY_RUN: z.coerce.boolean().default(true), - OWNERSHIP_SNAPSHOT_RETENTION_DAYS: z.coerce - .number() - .int() - .positive() - .default(30), - OWNERSHIP_SNAPSHOT_CLEANUP_ENABLED: z.coerce.boolean().default(false), - OWNERSHIP_SNAPSHOT_CLEANUP_INTERVAL_MINUTES: z.coerce - .number() - .int() - .positive() - .default(60), - - // Price movement detection job (feeds price_moved notifications) - DETECT_PRICE_MOVEMENTS_ENABLED: booleanCoerce.default(true), - DETECT_PRICE_MOVEMENTS_INTERVAL_MINUTES: z.coerce - .number() - .int() - .positive() - .default(5), - - // Governance proposal sync job - GOVERNANCE_SYNC_ENABLED: booleanCoerce.default(false), - GOVERNANCE_SYNC_INTERVAL_MINUTES: z.coerce - .number() - .int() - .positive() - .default(5), - - // Request body size limits (see docs/body-size-limits.md). - // Accepts any size string understood by the `bytes` package used - // internally by body-parser (e.g. '100kb', '1mb', '10mb'). - BODY_SIZE_LIMIT_DEFAULT: z.string().min(1).default('10mb'), - BODY_SIZE_LIMIT_AUTH: optionalNonEmptyString, - BODY_SIZE_LIMIT_ADMIN: optionalNonEmptyString, - BODY_SIZE_LIMIT_CREATORS: optionalNonEmptyString, - - // Distributed tracing - // Shared secret trusted internal callers present in the - // `x-internal-service-token` header to have their incoming - // `X-Trace-Id` header honored instead of a freshly generated one. - // Left unset by default, so no caller is trusted unless configured. - TRACE_ID_TRUSTED_TOKEN: optionalNonEmptyString, - INTERNAL_SERVICE_KEY: optionalNonEmptyString, - - // Query cost governor (#755): rolling per-wallet (or per-IP, when - // unauthenticated) database query budget. See - // src/middlewares/query-cost-governor.middleware.ts. - QUERY_COST_BUDGET: z.coerce.number().int().positive().default(200), - QUERY_COST_WINDOW_MS: z.coerce - .number() - .int() - .positive() - .default(60_000), - // JSON object overriding/extending the default route->cost map in - // src/constants/query-cost.constants.ts, e.g. - // '{"GET /search": 8, "GET /custom-route": 2}'. Merged over the - // defaults, not a full replacement, so operators only need to - // specify what differs. - QUERY_COST_MAP_JSON: optionalNonEmptyString, - // Comma-separated wallet addresses that bypass the governor entirely. - QUERY_COST_ADMIN_WALLETS: optionalNonEmptyString, - HORIZON_WEBHOOK_SECRET: optionalNonEmptyString, - WEBHOOK_RETRY_BASE_DELAY_MS: z.coerce - .number() - .int() - .positive() - .default(1000), - SSE_HEARTBEAT_INTERVAL_MS: z.coerce - .number() - .int() - .positive() - .default(15000), - SSE_QUEUE_CAPACITY: z.coerce.number().int().positive().default(1000), - SSE_QUEUE_FULL_TIMEOUT_MS: z.coerce - .number() - .int() - .positive() - .default(5000), - SSE_REPLAY_MAX_EVENTS: z.coerce.number().int().positive().default(100), - - // SSE subscription management (src/modules/subscriptions) — a wallet's - // subscription set, persisted in Redis, distinct from the per-connection - // heartbeat/queue/replay tuning above. - SSE_MAX_CONNECTIONS_PER_WALLET: z.coerce - .number() - .int() - .positive() - .default(10), - SSE_SUBSCRIPTION_TTL_MS: z.coerce - .number() - .int() - .positive() - .default(300000), - SSE_MAX_SUBSCRIPTIONS_PER_WALLET: z.coerce - .number() - .int() - .positive() - .default(10), - SSE_THROTTLE_DURATION_MS: z.coerce - .number() - .int() - .positive() - .default(1000), - - }) - .superRefine((data, ctx) => { - if (data.MODE === 'production' && data.STELLAR_NETWORK === 'testnet') { - ctx.addIssue({ - code: z.ZodIssueCode.custom, - path: ['STELLAR_NETWORK'], - message: - 'STELLAR_NETWORK should be "mainnet" when MODE is "production"', - }); - } - }); +import { z } from 'zod'; + +/** + * Centralized Zod schema for all environment variables. + * + * Configuration Source Precedence (highest to lowest): + * 1. Environment Variables (.env file or system environment) + * 2. Schema Defaults (defined below with .default()) + * 3. Validation Failure (startup fails if required field missing) + * + * Extracted into its own module so tests can validate the schema + * in isolation (via `.safeParse()`) without triggering the eager + * `envSchema.parse(process.env)` side-effect in `config.ts`. + * + * See docs/configuration.md for complete documentation. + * See docs/CONFIG_SOURCE_PRECEDENCE.md for visual reference. + */ +/** + * Helper to correctly coerce boolean strings from environment variables. + * Zod's default z.coerce.boolean() returns true for any non-empty string, + * including "false" and "0", which is usually not what we want for .env files. + */ +const booleanCoerce = z.preprocess(val => { + if (typeof val === 'string') { + const lower = val.toLowerCase(); + if (lower === 'true' || lower === '1') return true; + if (lower === 'false' || lower === '0') return false; + } + return val; +}, z.coerce.boolean()); + +const optionalNonEmptyString = z.preprocess(val => { + if (typeof val === 'string' && val.trim().length === 0) { + return undefined; + } + + return val; +}, z.string().min(1).optional()); + +export const envSchema = z + .object({ + PORT: z.coerce.number().default(3000), + MODE: z + .enum(['development', 'production', 'test']) + .default('development'), + DATABASE_URL: z + .string() + .min(1, 'DATABASE_URL is required in the environment variables'), + NODE_ID: z.string().default('node-local'), + + GMAIL_USER: z.string(), + GMAIL_APP_PASSWORD: z.string(), + // Google OAuth + GOOGLE_CLIENT_ID: z + .string() + .min(1, 'GOOGLE_CLIENT_ID is required for Google OAuth'), + GOOGLE_CLIENT_SECRET: z + .string() + .min(1, 'GOOGLE_CLIENT_SECRET is required for Google OAuth'), + + // URLs + BACKEND_URL: z.string().url(), + FRONTEND_URL: z + .string() + .url('FRONTEND_URL must be a valid URL') + .min(1, 'FRONTEND_URL is required'), + + // Cloudinary + CLOUDINARY_CLOUD_NAME: z + .string() + .min(1, 'CLOUDINARY_CLOUD_NAME is required for image uploads'), + CLOUDINARY_API_KEY: z + .string() + .min(1, 'CLOUDINARY_API_KEY is required for image uploads'), + CLOUDINARY_API_SECRET: z + .string() + .min(1, 'CLOUDINARY_API_SECRET is required for image uploads'), + + PAYSTACK_SECRET_KEY: z + .string() + .min(1, 'PAYSTACK_SECRET_KEY is required for payment processing'), + PAYSTACK_PUBLIC_KEY: optionalNonEmptyString, + ENABLE_RESPONSE_TIMING: booleanCoerce.default(true), + API_VERSION: z.string().default('1.0.0'), + ENABLE_API_VERSION_HEADER: booleanCoerce.default(true), + ENABLE_SCHEMA_VERSION_HEADER: booleanCoerce.default(true), + ENABLE_REQUEST_LOGGING: booleanCoerce.default(true), + DB_QUERY_TIMEOUT_MS: z.coerce.number().default(5000), + DB_POOL_WAIT_WARN_MS: z.coerce.number().int().positive().default(2000), + DB_POOL_WAIT_ERROR_MS: z.coerce.number().int().positive().default(5000), + WEBHOOK_MAX_PER_CREATOR: z.coerce.number().int().positive().default(5), + WEBHOOK_RETRY_MAX_ATTEMPTS: z.coerce.number().int().positive().default(3), + + APP_SECRET: z + .string() + .min(32, 'APP_SECRET should be at least 32 characters') + .default('accesslayer_default_development_secret_key_32_bytes_long'), + + // JWT auth + JWT_SECRET: z + .string() + .min(32, 'JWT_SECRET should be at least 32 characters') + .default('accesslayer_default_development_jwt_secret_key_32_bytes'), + JWT_ISSUER: z.string().default('accesslayer-server'), + JWT_EXPIRES_IN: z.string().default('15m'), + JWT_ACCESS_TOKEN_TTL_SECONDS: z.coerce + .number() + .int() + .positive() + .default(900), + + // Redis cache + REDIS_URL: z.string().default('redis://localhost:6379'), + ENABLE_REDIS_CACHE: booleanCoerce.default(true), + + // Key trade lockup + LOCKUP_DURATION_SECONDS: z.coerce.number().int().nonnegative().default(0), + + // Leaderboard volume + LEADERBOARD_VOLUME_WINDOW_DAYS: z.coerce + .number() + .int() + .positive() + .default(7), + LEADERBOARD_VOLUME_CACHE_TTL_SECONDS: z.coerce + .number() + .int() + .positive() + .default(300), + + INDEXER_JITTER_FACTOR: z.coerce.number().min(0).max(1).default(0.1), + BACKGROUND_JOB_LOCK_TTL_MS: z.coerce + .number() + .int() + .positive() + .default(300000), + SLOW_QUERY_THRESHOLD_MS: z.coerce.number().int().positive().default(500), + CREATOR_LIST_SLOW_QUERY_THRESHOLD_MS: z.coerce + .number() + .int() + .positive() + .default(500), + INDEXER_CURSOR_STALE_AGE_WARNING_MS: z.coerce + .number() + .int() + .positive() + .default(300000), + INDEXER_HEARTBEAT_STALE_THRESHOLD_MS: z.coerce + .number() + .positive() + .default(300000), + + // Indexer feature flags + ENABLE_INDEXER_DEDUPE: booleanCoerce.default(true), + ENABLE_INDEXER_DLQ: booleanCoerce.default(true), + ENABLE_INDEXER_CURSOR_STALENESS_WARNING: booleanCoerce.default(true), + + // Stellar auth — optional server keypair secret used for SEP-10 challenge + // signing. When absent the server falls back to an ephemeral random keypair. + STELLAR_AUTH_SECRET: optionalNonEmptyString, + + // Stellar network + STELLAR_NETWORK: z + .enum(['testnet', 'mainnet'], { + message: + 'STELLAR_NETWORK must be "testnet" or "mainnet". Set STELLAR_NETWORK in your .env file.', + }) + .default('testnet'), + STELLAR_HORIZON_URL: z + .string() + .url( + 'STELLAR_HORIZON_URL must be a valid URL (e.g. https://horizon-testnet.stellar.org)' + ) + .default('https://horizon-testnet.stellar.org'), + STELLAR_SOROBAN_RPC_URL: z + .string() + .url( + 'STELLAR_SOROBAN_RPC_URL must be a valid URL (e.g. https://soroban-testnet.stellar.org)' + ) + .default('https://soroban-testnet.stellar.org'), + + // Ownership snapshot cleanup job + OWNERSHIP_SNAPSHOT_TABLE_NAME: z + .string() + .min(1) + .default('creator_ownership_snapshots'), + OWNERSHIP_SNAPSHOT_CLEANUP_DRY_RUN: z.coerce.boolean().default(true), + OWNERSHIP_SNAPSHOT_RETENTION_DAYS: z.coerce + .number() + .int() + .positive() + .default(30), + OWNERSHIP_SNAPSHOT_CLEANUP_ENABLED: z.coerce.boolean().default(false), + OWNERSHIP_SNAPSHOT_CLEANUP_INTERVAL_MINUTES: z.coerce + .number() + .int() + .positive() + .default(60), + + // Price movement detection job (feeds price_moved notifications) + DETECT_PRICE_MOVEMENTS_ENABLED: booleanCoerce.default(true), + DETECT_PRICE_MOVEMENTS_INTERVAL_MINUTES: z.coerce + .number() + .int() + .positive() + .default(5), + + // Governance proposal sync job + GOVERNANCE_SYNC_ENABLED: booleanCoerce.default(false), + GOVERNANCE_SYNC_INTERVAL_MINUTES: z.coerce + .number() + .int() + .positive() + .default(5), + + // Request body size limits (see docs/body-size-limits.md). + // Accepts any size string understood by the `bytes` package used + // internally by body-parser (e.g. '100kb', '1mb', '10mb'). + BODY_SIZE_LIMIT_DEFAULT: z.string().min(1).default('10mb'), + BODY_SIZE_LIMIT_AUTH: optionalNonEmptyString, + BODY_SIZE_LIMIT_ADMIN: optionalNonEmptyString, + BODY_SIZE_LIMIT_CREATORS: optionalNonEmptyString, + + // Distributed tracing + // Shared secret trusted internal callers present in the + // `x-internal-service-token` header to have their incoming + // `X-Trace-Id` header honored instead of a freshly generated one. + // Left unset by default, so no caller is trusted unless configured. + TRACE_ID_TRUSTED_TOKEN: optionalNonEmptyString, + INTERNAL_SERVICE_KEY: optionalNonEmptyString, + + // Query cost governor (#755): rolling per-wallet (or per-IP, when + // unauthenticated) database query budget. See + // src/middlewares/query-cost-governor.middleware.ts. + QUERY_COST_BUDGET: z.coerce.number().int().positive().default(200), + QUERY_COST_WINDOW_MS: z.coerce + .number() + .int() + .positive() + .default(60_000), + // JSON object overriding/extending the default route->cost map in + // src/constants/query-cost.constants.ts, e.g. + // '{"GET /search": 8, "GET /custom-route": 2}'. Merged over the + // defaults, not a full replacement, so operators only need to + // specify what differs. + QUERY_COST_MAP_JSON: optionalNonEmptyString, + // Comma-separated wallet addresses that bypass the governor entirely. + QUERY_COST_ADMIN_WALLETS: optionalNonEmptyString, + HORIZON_WEBHOOK_SECRET: optionalNonEmptyString, + WEBHOOK_RETRY_BASE_DELAY_MS: z.coerce + .number() + .int() + .positive() + .default(1000), + SSE_HEARTBEAT_INTERVAL_MS: z.coerce + .number() + .int() + .positive() + .default(15000), + SSE_QUEUE_CAPACITY: z.coerce.number().int().positive().default(1000), + SSE_QUEUE_FULL_TIMEOUT_MS: z.coerce + .number() + .int() + .positive() + .default(5000), + SSE_REPLAY_MAX_EVENTS: z.coerce.number().int().positive().default(100), + + // SSE subscription management (src/modules/subscriptions) — a wallet's + // subscription set, persisted in Redis, distinct from the per-connection + // heartbeat/queue/replay tuning above. + SSE_MAX_CONNECTIONS_PER_WALLET: z.coerce + .number() + .int() + .positive() + .default(10), + SSE_SUBSCRIPTION_TTL_MS: z.coerce + .number() + .int() + .positive() + .default(300000), + SSE_MAX_SUBSCRIPTIONS_PER_WALLET: z.coerce + .number() + .int() + .positive() + .default(10), + SSE_THROTTLE_DURATION_MS: z.coerce + .number() + .int() + .positive() + .default(1000), + + }) + .superRefine((data, ctx) => { + if (data.MODE === 'production' && data.STELLAR_NETWORK === 'testnet') { + ctx.addIssue({ + code: z.ZodIssueCode.custom, + path: ['STELLAR_NETWORK'], + message: + 'STELLAR_NETWORK should be "mainnet" when MODE is "production"', + }); + } + }); diff --git a/src/constants/error.constants.ts b/src/constants/error.constants.ts index bfee94f..9034092 100644 --- a/src/constants/error.constants.ts +++ b/src/constants/error.constants.ts @@ -1,21 +1,21 @@ -// src/constants/error.constants.ts -/** - * Shared API error codes. - */ -export const ErrorCode = { - VALIDATION_ERROR: 'VALIDATION_ERROR', - UNPROCESSABLE_ENTITY: 'UNPROCESSABLE_ENTITY', - NOT_FOUND: 'NOT_FOUND', - UNAUTHORIZED: 'UNAUTHORIZED', - FORBIDDEN: 'FORBIDDEN', - CONFLICT: 'CONFLICT', - BAD_REQUEST: 'BAD_REQUEST', - INTERNAL_ERROR: 'INTERNAL_ERROR', - RATE_LIMIT: 'RATE_LIMIT', - PRISMA_ERROR: 'DATABASE_ERROR', - JWT_ERROR: 'TOKEN_ERROR', - INSUFFICIENT_BALANCE: 'insufficient_balance', - NOT_A_CREATOR: 'not_a_creator', -} as const; - -export type ErrorCodeType = (typeof ErrorCode)[keyof typeof ErrorCode]; +// src/constants/error.constants.ts +/** + * Shared API error codes. + */ +export const ErrorCode = { + VALIDATION_ERROR: 'VALIDATION_ERROR', + UNPROCESSABLE_ENTITY: 'UNPROCESSABLE_ENTITY', + NOT_FOUND: 'NOT_FOUND', + UNAUTHORIZED: 'UNAUTHORIZED', + FORBIDDEN: 'FORBIDDEN', + CONFLICT: 'CONFLICT', + BAD_REQUEST: 'BAD_REQUEST', + INTERNAL_ERROR: 'INTERNAL_ERROR', + RATE_LIMIT: 'RATE_LIMIT', + PRISMA_ERROR: 'DATABASE_ERROR', + JWT_ERROR: 'TOKEN_ERROR', + INSUFFICIENT_BALANCE: 'insufficient_balance', + NOT_A_CREATOR: 'not_a_creator', +} as const; + +export type ErrorCodeType = (typeof ErrorCode)[keyof typeof ErrorCode]; diff --git a/src/jobs/soroban-wal-recovery.job.ts b/src/jobs/soroban-wal-recovery.job.ts new file mode 100644 index 0000000..febacd9 --- /dev/null +++ b/src/jobs/soroban-wal-recovery.job.ts @@ -0,0 +1,61 @@ +import { logger } from '../utils/logger.utils'; +import { HorizonSorobanGateway } from '../modules/soroban-wal/horizon-soroban-gateway'; +import { PrismaSorobanWALStore } from '../modules/soroban-wal/prisma-soroban-wal.store'; +import { SorobanWALService } from '../modules/soroban-wal/soroban-wal.service'; +import { applyDefaultSorobanSideEffects } from '../modules/soroban-wal/soroban-wal-side-effects'; + +const RECOVERY_INTERVAL_MS = 60_000; +const STALE_AFTER_MS = 30_000; + +const recoveryService = new SorobanWALService( + new PrismaSorobanWALStore(), + new HorizonSorobanGateway() +); + +let recoveryTimer: NodeJS.Timeout | null = null; +let recoveryRunning = false; + +export async function recoverStaleSorobanTransactions( + now = new Date() +): Promise { + const olderThan = new Date(now.getTime() - STALE_AFTER_MS); + const recovered = await recoveryService.recover( + olderThan, + applyDefaultSorobanSideEffects + ); + return recovered.length; +} + +export function startSorobanWALRecoveryJob(): void { + if (recoveryTimer) return; + + const run = async () => { + if (recoveryRunning) return; + recoveryRunning = true; + try { + const recoveredEntries = await recoverStaleSorobanTransactions(); + if (recoveredEntries > 0) { + logger.info( + { recoveredEntries }, + 'Soroban WAL recovery pass completed' + ); + } + } catch (error) { + logger.error({ error }, 'Soroban WAL recovery pass failed'); + } finally { + recoveryRunning = false; + } + }; + + void run(); + recoveryTimer = setInterval(() => void run(), RECOVERY_INTERVAL_MS); + recoveryTimer.unref?.(); + logger.info('Soroban WAL recovery job started'); +} + +export function stopSorobanWALRecoveryJob(): void { + if (!recoveryTimer) return; + clearInterval(recoveryTimer); + recoveryTimer = null; + logger.info('Soroban WAL recovery job stopped'); +} diff --git a/src/modules/creator/creator.routes.ts b/src/modules/creator/creator.routes.ts index 70a2a93..fa1aba0 100644 --- a/src/modules/creator/creator.routes.ts +++ b/src/modules/creator/creator.routes.ts @@ -1,106 +1,106 @@ -import { Router } from 'express'; -import { httpListCreators, httpGetCreatorStats } from '../creators/creators.controllers'; -import { cacheControl } from '../../middlewares/cache-control.middleware'; -import { CREATOR_PUBLIC_ROUTE_CACHE_PRESETS } from '../../constants/creator-public-cache.constants'; -import { CREATOR_PUBLIC_ROUTE_NAMES } from '../../constants/creator-public-routes.constants'; -import { createCreatorReadMetricsMiddleware } from '../../utils/creator-read-metrics.utils'; -import { normalizeTrailingSlash } from '../../middlewares/trailing-slash-normalizer.middleware'; -import { requireKeyCreator, AuthenticatedRequest } from '../../middlewares/jwt-auth.middleware'; -import { sendError, sendSuccess } from '../../utils/api-response.utils'; -import { ErrorCode } from '../../constants/error.constants'; -import { prisma } from '../../utils/prisma.utils'; - -const creatorsRouter = Router(); - -// Normalize trailing slashes for all creator routes so that, e.g., -// GET /api/v1/creators/ reaches the same handler as GET /api/v1/creators. -// Scoped to this router to avoid side-effects on other route groups. -creatorsRouter.use(normalizeTrailingSlash); - -/** - * GET /api/v1/creators - * - * List all creators with pagination and filtering. - * Public endpoint with 5-minute cache. - */ -creatorsRouter.get( - '/', - createCreatorReadMetricsMiddleware('list'), - cacheControl(CREATOR_PUBLIC_ROUTE_CACHE_PRESETS[CREATOR_PUBLIC_ROUTE_NAMES.LIST]), - httpListCreators -); -// 405 handler for / -creatorsRouter.all('/', (_req, res) => { - res.set('Allow', 'GET').sendStatus(405); -}); - -/** - * GET /api/v1/creators/:id/stats - * - * Get public stats for a specific creator. - * Public endpoint with 5-minute cache. - */ -creatorsRouter.get( - '/:id/stats', - createCreatorReadMetricsMiddleware('detail'), - cacheControl(CREATOR_PUBLIC_ROUTE_CACHE_PRESETS[CREATOR_PUBLIC_ROUTE_NAMES.GET_STATS]), - httpGetCreatorStats -); -// 405 handler for /:id/stats -creatorsRouter.all('/:id/stats', (_req, res) => { - res.set('Allow', 'GET').sendStatus(405); -}); - -/** - * POST /api/v1/creator/:keyId/holder-cap - * Update holder cap endpoint for creators to change max keys a single wallet can hold (#841). - */ -creatorsRouter.post( - '/:keyId/holder-cap', - requireKeyCreator('keyId'), - async (req: AuthenticatedRequest, res, next) => { - const { capBps } = req.body || {}; - if ( - capBps === undefined || - capBps === null || - typeof capBps !== 'number' || - capBps < 100 || - capBps > 2500 - ) { - sendError( - res, - 422, - ErrorCode.UNPROCESSABLE_ENTITY, - 'capBps must be between 100 and 2500' - ); - return; - } - - const keyId = Array.isArray(req.params.keyId) - ? req.params.keyId[0] - : req.params.keyId; - try { - const creatorProfile = await prisma.creatorProfile.findFirst({ - where: { OR: [{ id: keyId }, { handle: keyId }] }, - }); - if (!creatorProfile) { - sendError(res, 404, ErrorCode.NOT_FOUND, 'Key not found'); - return; - } - - const updated = await prisma.creatorProfile.update({ - where: { id: creatorProfile.id }, - data: { holderCapBps: capBps }, - }); - - sendSuccess(res, { - holderCapBps: updated.holderCapBps, - percentage: `${updated.holderCapBps / 100}%`, - }); - } catch (error) { - next(error); - } - } -); - +import { Router } from 'express'; +import { httpListCreators, httpGetCreatorStats } from '../creators/creators.controllers'; +import { cacheControl } from '../../middlewares/cache-control.middleware'; +import { CREATOR_PUBLIC_ROUTE_CACHE_PRESETS } from '../../constants/creator-public-cache.constants'; +import { CREATOR_PUBLIC_ROUTE_NAMES } from '../../constants/creator-public-routes.constants'; +import { createCreatorReadMetricsMiddleware } from '../../utils/creator-read-metrics.utils'; +import { normalizeTrailingSlash } from '../../middlewares/trailing-slash-normalizer.middleware'; +import { requireKeyCreator, AuthenticatedRequest } from '../../middlewares/jwt-auth.middleware'; +import { sendError, sendSuccess } from '../../utils/api-response.utils'; +import { ErrorCode } from '../../constants/error.constants'; +import { prisma } from '../../utils/prisma.utils'; + +const creatorsRouter = Router(); + +// Normalize trailing slashes for all creator routes so that, e.g., +// GET /api/v1/creators/ reaches the same handler as GET /api/v1/creators. +// Scoped to this router to avoid side-effects on other route groups. +creatorsRouter.use(normalizeTrailingSlash); + +/** + * GET /api/v1/creators + * + * List all creators with pagination and filtering. + * Public endpoint with 5-minute cache. + */ +creatorsRouter.get( + '/', + createCreatorReadMetricsMiddleware('list'), + cacheControl(CREATOR_PUBLIC_ROUTE_CACHE_PRESETS[CREATOR_PUBLIC_ROUTE_NAMES.LIST]), + httpListCreators +); +// 405 handler for / +creatorsRouter.all('/', (_req, res) => { + res.set('Allow', 'GET').sendStatus(405); +}); + +/** + * GET /api/v1/creators/:id/stats + * + * Get public stats for a specific creator. + * Public endpoint with 5-minute cache. + */ +creatorsRouter.get( + '/:id/stats', + createCreatorReadMetricsMiddleware('detail'), + cacheControl(CREATOR_PUBLIC_ROUTE_CACHE_PRESETS[CREATOR_PUBLIC_ROUTE_NAMES.GET_STATS]), + httpGetCreatorStats +); +// 405 handler for /:id/stats +creatorsRouter.all('/:id/stats', (_req, res) => { + res.set('Allow', 'GET').sendStatus(405); +}); + +/** + * POST /api/v1/creator/:keyId/holder-cap + * Update holder cap endpoint for creators to change max keys a single wallet can hold (#841). + */ +creatorsRouter.post( + '/:keyId/holder-cap', + requireKeyCreator('keyId'), + async (req: AuthenticatedRequest, res, next) => { + const { capBps } = req.body || {}; + if ( + capBps === undefined || + capBps === null || + typeof capBps !== 'number' || + capBps < 100 || + capBps > 2500 + ) { + sendError( + res, + 422, + ErrorCode.UNPROCESSABLE_ENTITY, + 'capBps must be between 100 and 2500' + ); + return; + } + + const keyId = Array.isArray(req.params.keyId) + ? req.params.keyId[0] + : req.params.keyId; + try { + const creatorProfile = await prisma.creatorProfile.findFirst({ + where: { OR: [{ id: keyId }, { handle: keyId }] }, + }); + if (!creatorProfile) { + sendError(res, 404, ErrorCode.NOT_FOUND, 'Key not found'); + return; + } + + const updated = await prisma.creatorProfile.update({ + where: { id: creatorProfile.id }, + data: { holderCapBps: capBps }, + }); + + sendSuccess(res, { + holderCapBps: updated.holderCapBps, + percentage: `${updated.holderCapBps / 100}%`, + }); + } catch (error) { + next(error); + } + } +); + export default creatorsRouter; \ No newline at end of file diff --git a/src/modules/soroban-wal/horizon-soroban-gateway.ts b/src/modules/soroban-wal/horizon-soroban-gateway.ts new file mode 100644 index 0000000..18a25f1 --- /dev/null +++ b/src/modules/soroban-wal/horizon-soroban-gateway.ts @@ -0,0 +1,39 @@ +import { horizonGet } from '../../clients/horizon.client'; +import { + SorobanTransactionGateway, + SorobanTransactionStatus, +} from './soroban-wal.types'; + +export type SubmitSorobanXdr = ( + xdrPayload: string +) => Promise<{ txHash: string }>; + +export class HorizonSorobanGateway implements SorobanTransactionGateway { + constructor(private readonly submitXdr?: SubmitSorobanXdr) {} + + async submit(xdrPayload: string): Promise<{ txHash: string }> { + if (!this.submitXdr) { + throw new Error('Soroban submission adapter is not configured'); + } + return this.submitXdr(xdrPayload); + } + + async getStatus(txHash: string): Promise { + const response = await horizonGet( + `/transactions/${encodeURIComponent(txHash)}` + ); + + if (response.status === 404) return { status: 'PENDING' }; + if (!response.ok) { + return { + status: 'FAILED', + error: `Horizon returned HTTP ${response.status} for ${txHash}`, + }; + } + + const transaction = (await response.json()) as { successful?: boolean }; + return transaction.successful === false + ? { status: 'FAILED', error: `Transaction ${txHash} failed on-chain` } + : { status: 'CONFIRMED' }; + } +} diff --git a/src/modules/soroban-wal/prisma-soroban-wal.store.ts b/src/modules/soroban-wal/prisma-soroban-wal.store.ts new file mode 100644 index 0000000..ecae996 --- /dev/null +++ b/src/modules/soroban-wal/prisma-soroban-wal.store.ts @@ -0,0 +1,218 @@ +import { prisma } from '../../utils/prisma.utils'; +import { + ApplySorobanSideEffects, + CreateSorobanWALEntry, + RollbackSorobanOptimisticState, + SorobanWALEntry, + SorobanWALStore, + ValidateSorobanOperation, +} from './soroban-wal.types'; + +type RawEntry = Omit & { + amount: { toString(): string } | string | number; + expectedSupplyBefore: { toString(): string } | string | number; +}; + +interface WalDelegate { + findUnique(args: unknown): Promise; + findMany(args: unknown): Promise; + create(args: unknown): Promise; + update(args: unknown): Promise; +} + +export interface SorobanWALDatabaseTransaction { + sorobanWALEntry: WalDelegate; + $queryRawUnsafe(query: string, ...values: unknown[]): Promise; + [model: string]: unknown; +} + +interface WalClient extends SorobanWALDatabaseTransaction { + $transaction( + operation: (transaction: SorobanWALDatabaseTransaction) => Promise, + options?: { isolationLevel: 'Serializable' } + ): Promise; +} + +const client = prisma as unknown as WalClient; + +export class PrismaSorobanWALStore implements SorobanWALStore { + async createPending( + input: CreateSorobanWALEntry, + validate: ValidateSorobanOperation + ): Promise<{ entry: SorobanWALEntry; created: boolean }> { + try { + return await client.$transaction( + async transaction => { + const existing = await transaction.sorobanWALEntry.findUnique({ + where: { idempotencyKey: input.idempotencyKey }, + }); + if (existing) { + return { entry: normalizeEntry(existing), created: false }; + } + + await validate(transaction); + const entry = await transaction.sorobanWALEntry.create({ + data: { ...input, state: 'PENDING' }, + }); + return { entry: normalizeEntry(entry), created: true }; + }, + { isolationLevel: 'Serializable' } + ); + } catch (error) { + if (!isUniqueConstraintError(error)) throw error; + + const existing = await client.sorobanWALEntry.findUnique({ + where: { idempotencyKey: input.idempotencyKey }, + }); + if (!existing) throw error; + return { entry: normalizeEntry(existing), created: false }; + } + } + + async markSubmitted(id: string, txHash: string): Promise { + return client.$transaction( + async transaction => { + const entry = await lockEntry(transaction, id); + if (entry.state !== 'PENDING') { + throw new Error( + `Cannot submit WAL entry ${id} from state ${entry.state}` + ); + } + return normalizeEntry( + await transaction.sorobanWALEntry.update({ + where: { id }, + data: { + state: 'SUBMITTED', + txHash, + submittedAt: new Date(), + error: null, + }, + }) + ); + }, + { isolationLevel: 'Serializable' } + ); + } + + async markFailed( + id: string, + error: string, + rollback: RollbackSorobanOptimisticState + ): Promise { + return this.finishWithRollback(id, 'FAILED', error, rollback); + } + + async markRolledBack( + id: string, + rollback: RollbackSorobanOptimisticState + ): Promise { + return this.finishWithRollback(id, 'ROLLED_BACK', null, rollback); + } + + async confirmAtomically( + id: string, + txHash: string, + applySideEffects: ApplySorobanSideEffects + ): Promise { + return client.$transaction( + async transaction => { + const raw = await lockEntry(transaction, id); + const entry = normalizeEntry(raw); + + if (entry.state === 'CONFIRMED') return entry; + if (entry.state !== 'SUBMITTED' || entry.txHash !== txHash) { + throw new Error( + `Cannot confirm WAL entry ${id} from state ${entry.state}` + ); + } + + await applySideEffects(transaction, entry); + + return normalizeEntry( + await transaction.sorobanWALEntry.update({ + where: { id }, + data: { + state: 'CONFIRMED', + confirmedAt: new Date(), + error: null, + }, + }) + ); + }, + { isolationLevel: 'Serializable' } + ); + } + + async listRecoverable(olderThan: Date): Promise { + const entries = await client.sorobanWALEntry.findMany({ + where: { + state: { in: ['PENDING', 'SUBMITTED'] }, + createdAt: { lt: olderThan }, + }, + orderBy: { createdAt: 'asc' }, + }); + return entries.map(normalizeEntry); + } + + private async finishWithRollback( + id: string, + state: 'FAILED' | 'ROLLED_BACK', + error: string | null, + rollback: RollbackSorobanOptimisticState + ): Promise { + return client.$transaction( + async transaction => { + const raw = await lockEntry(transaction, id); + const entry = normalizeEntry(raw); + if ( + entry.state === 'CONFIRMED' || + entry.state === 'FAILED' || + entry.state === 'ROLLED_BACK' + ) { + return entry; + } + + await rollback(transaction, entry); + return normalizeEntry( + await transaction.sorobanWALEntry.update({ + where: { id }, + data: { state, error }, + }) + ); + }, + { isolationLevel: 'Serializable' } + ); + } +} + +async function lockEntry( + transaction: SorobanWALDatabaseTransaction, + id: string +): Promise { + await transaction.$queryRawUnsafe( + 'SELECT "id" FROM "soroban_wal_entries" WHERE "id" = $1 FOR UPDATE', + id + ); + const entry = await transaction.sorobanWALEntry.findUnique({ + where: { id }, + }); + if (!entry) throw new Error(`Soroban WAL entry ${id} was not found`); + return entry; +} + +function normalizeEntry(entry: RawEntry): SorobanWALEntry { + return { + ...entry, + amount: entry.amount.toString(), + expectedSupplyBefore: entry.expectedSupplyBefore.toString(), + }; +} + +function isUniqueConstraintError(error: unknown): boolean { + return ( + typeof error === 'object' && + error !== null && + 'code' in error && + (error as { code?: string }).code === 'P2002' + ); +} diff --git a/src/modules/soroban-wal/soroban-wal-side-effects.ts b/src/modules/soroban-wal/soroban-wal-side-effects.ts new file mode 100644 index 0000000..a55b805 --- /dev/null +++ b/src/modules/soroban-wal/soroban-wal-side-effects.ts @@ -0,0 +1,94 @@ +import { SorobanWALEntry } from './soroban-wal.types'; +import { SorobanWALDatabaseTransaction } from './prisma-soroban-wal.store'; + +interface SideEffectTransaction extends SorobanWALDatabaseTransaction { + creatorProfile: any; + keyOwnership: any; + creatorPriceSnapshot: any; + activity: any; + activityLog: any; +} + +export async function applyDefaultSorobanSideEffects( + databaseTransaction: unknown, + entry: SorobanWALEntry +): Promise { + const transaction = databaseTransaction as SideEffectTransaction; + const creator = await transaction.creatorProfile.findFirst({ + where: { + user: { + is: { + stellarWallet: { is: { address: entry.creatorWallet } }, + }, + }, + }, + select: { id: true, circulatingSupply: true }, + }); + + if (!creator) { + throw new Error( + `Creator profile for wallet ${entry.creatorWallet} was not found` + ); + } + + if (creator.circulatingSupply.toString() !== entry.expectedSupplyBefore) { + throw new Error( + `Supply changed before confirmation for WAL entry ${entry.id}` + ); + } + + const amount = entry.amount; + const delta = entry.operation === 'BUY' ? amount : `-${amount}`; + const confirmedAt = new Date(); + + await transaction.keyOwnership.upsert({ + where: { + ownerAddress_creatorId: { + ownerAddress: entry.wallet, + creatorId: creator.id, + }, + }, + create: { + ownerAddress: entry.wallet, + creatorId: creator.id, + balance: delta, + lastBuyAt: entry.operation === 'BUY' ? confirmedAt : null, + }, + update: { + balance: { increment: delta }, + ...(entry.operation === 'BUY' ? { lastBuyAt: confirmedAt } : {}), + }, + }); + + await transaction.creatorProfile.update({ + where: { id: creator.id }, + data: { circulatingSupply: { increment: delta } }, + }); + + await transaction.creatorPriceSnapshot.upsert({ + where: { creatorId: creator.id }, + create: { creatorId: creator.id, lastTradeAt: confirmedAt }, + update: { lastTradeAt: confirmedAt }, + }); + + await transaction.activity.create({ + data: { + type: entry.operation === 'BUY' ? 'KEY_BOUGHT' : 'KEY_SOLD', + actor: entry.wallet, + creatorId: creator.id, + payload: { amount, txHash: entry.txHash }, + createdAt: confirmedAt, + }, + }); + + await transaction.activityLog.create({ + data: { + type: entry.operation.toLowerCase(), + actor: entry.wallet, + keyId: creator.id, + amount, + txHash: entry.txHash, + timestamp: confirmedAt, + }, + }); +} diff --git a/src/modules/soroban-wal/soroban-wal.service.test.ts b/src/modules/soroban-wal/soroban-wal.service.test.ts new file mode 100644 index 0000000..dbd34d3 --- /dev/null +++ b/src/modules/soroban-wal/soroban-wal.service.test.ts @@ -0,0 +1,246 @@ +import { SorobanWALService } from './soroban-wal.service'; +import { + ApplySorobanSideEffects, + CreateSorobanWALEntry, + RollbackSorobanOptimisticState, + SorobanTransactionGateway, + SorobanTransactionStatus, + SorobanWALEntry, + SorobanWALStore, + ValidateSorobanOperation, +} from './soroban-wal.types'; + +const input: CreateSorobanWALEntry = { + idempotencyKey: 'trade-1', + operation: 'BUY', + wallet: 'GBUYER', + creatorWallet: 'GCREATOR', + amount: '2', + expectedSupplyBefore: '10', + xdrPayload: 'AAAA', +}; + +class MemoryWALStore implements SorobanWALStore { + readonly entries = new Map(); + private sequence = 0; + + async createPending( + createInput: CreateSorobanWALEntry, + validate: ValidateSorobanOperation + ): Promise<{ entry: SorobanWALEntry; created: boolean }> { + const existing = [...this.entries.values()].find( + entry => entry.idempotencyKey === createInput.idempotencyKey + ); + if (existing) return { entry: existing, created: false }; + + await validate({}); + const now = new Date(); + const entry: SorobanWALEntry = { + ...createInput, + id: `wal-${++this.sequence}`, + state: 'PENDING', + txHash: null, + submittedAt: null, + confirmedAt: null, + error: null, + createdAt: now, + updatedAt: now, + }; + this.entries.set(entry.id, entry); + return { entry, created: true }; + } + + async markSubmitted(id: string, txHash: string): Promise { + return this.update(id, { + state: 'SUBMITTED', + txHash, + submittedAt: new Date(), + }); + } + + async markFailed( + id: string, + error: string, + rollback: RollbackSorobanOptimisticState + ): Promise { + const entry = this.get(id); + if (isTerminal(entry)) return entry; + await rollback({}, entry); + return this.update(id, { state: 'FAILED', error }); + } + + async markRolledBack( + id: string, + rollback: RollbackSorobanOptimisticState + ): Promise { + const entry = this.get(id); + if (isTerminal(entry)) return entry; + await rollback({}, entry); + return this.update(id, { state: 'ROLLED_BACK' }); + } + + async confirmAtomically( + id: string, + txHash: string, + applySideEffects: ApplySorobanSideEffects + ): Promise { + const entry = this.get(id); + if (entry.state === 'CONFIRMED') return entry; + if (entry.state !== 'SUBMITTED' || entry.txHash !== txHash) { + throw new Error('invalid confirmation transition'); + } + await applySideEffects({}, entry); + return this.update(id, { + state: 'CONFIRMED', + confirmedAt: new Date(), + }); + } + + async listRecoverable(olderThan: Date): Promise { + return [...this.entries.values()].filter( + entry => + (entry.state === 'PENDING' || entry.state === 'SUBMITTED') && + entry.createdAt < olderThan + ); + } + + private get(id: string): SorobanWALEntry { + const entry = this.entries.get(id); + if (!entry) throw new Error('missing WAL entry'); + return entry; + } + + private update( + id: string, + change: Partial + ): SorobanWALEntry { + const entry = { ...this.get(id), ...change, updatedAt: new Date() }; + this.entries.set(id, entry); + return entry; + } +} + +function isTerminal(entry: SorobanWALEntry): boolean { + return ['CONFIRMED', 'FAILED', 'ROLLED_BACK'].includes(entry.state); +} + +function createGateway( + statuses: SorobanTransactionStatus[] = [{ status: 'CONFIRMED' }] +): SorobanTransactionGateway & { + submit: jest.Mock; + getStatus: jest.Mock; +} { + return { + submit: jest.fn().mockResolvedValue({ txHash: 'tx-1' }), + getStatus: jest.fn( + async (): Promise => + statuses.shift() ?? { status: 'PENDING' } + ), + }; +} + +function createService( + store: SorobanWALStore, + gateway: SorobanTransactionGateway +): SorobanWALService { + return new SorobanWALService(store, gateway, { + maxConfirmationAttempts: 3, + baseRetryDelayMs: 1, + sleep: async () => {}, + }); +} + +describe('SorobanWALService', () => { + it('writes PENDING after validation and before chain submission', async () => { + const order: string[] = []; + const store = new MemoryWALStore(); + const gateway = createGateway(); + gateway.submit.mockImplementation(async () => { + order.push('submit'); + expect([...store.entries.values()][0].state).toBe('PENDING'); + return { txHash: 'tx-1' }; + }); + + await createService(store, gateway).execute({ + ...input, + validate: async () => void order.push('validate'), + applySideEffects: async () => {}, + }); + + expect(order).toEqual(['validate', 'submit']); + }); + + it('rolls back stale PENDING entries without applying side effects', async () => { + const store = new MemoryWALStore(); + await store.createPending(input, async () => {}); + const apply = jest.fn(); + const rollback = jest.fn(); + + const recovered = await createService(store, createGateway()).recover( + new Date(Date.now() + 31_000), + apply, + rollback + ); + + expect(recovered[0].state).toBe('ROLLED_BACK'); + expect(rollback).toHaveBeenCalledTimes(1); + expect(apply).not.toHaveBeenCalled(); + }); + + it('recovers SUBMITTED entries and applies side effects exactly once', async () => { + const store = new MemoryWALStore(); + const { entry } = await store.createPending(input, async () => {}); + await store.markSubmitted(entry.id, 'tx-1'); + const apply = jest.fn(); + const service = createService(store, createGateway()); + const cutoff = new Date(Date.now() + 31_000); + + await service.recover(cutoff, apply); + await service.recover(cutoff, apply); + + expect(apply).toHaveBeenCalledTimes(1); + expect(store.entries.get(entry.id)?.state).toBe('CONFIRMED'); + }); + + it('deduplicates retries by idempotency key', async () => { + const store = new MemoryWALStore(); + const gateway = createGateway(); + const service = createService(store, gateway); + const apply = jest.fn(); + + await service.execute({ ...input, applySideEffects: apply }); + await service.execute({ ...input, applySideEffects: apply }); + + expect(store.entries.size).toBe(1); + expect(gateway.submit).toHaveBeenCalledTimes(1); + expect(apply).toHaveBeenCalledTimes(1); + }); + + it('produces the same state after recovery as a crash-free execution', async () => { + const crashFreeStore = new MemoryWALStore(); + let crashFreeBalance = 0; + await createService(crashFreeStore, createGateway()).execute({ + ...input, + applySideEffects: async () => { + crashFreeBalance += Number(input.amount); + }, + }); + + const recoveredStore = new MemoryWALStore(); + const { entry } = await recoveredStore.createPending( + input, + async () => {} + ); + await recoveredStore.markSubmitted(entry.id, 'tx-1'); + let recoveredBalance = 0; + await createService(recoveredStore, createGateway()).recover( + new Date(Date.now() + 31_000), + async () => { + recoveredBalance += Number(input.amount); + } + ); + + expect(recoveredBalance).toBe(crashFreeBalance); + expect(recoveredStore.entries.get(entry.id)?.state).toBe('CONFIRMED'); + }); +}); diff --git a/src/modules/soroban-wal/soroban-wal.service.ts b/src/modules/soroban-wal/soroban-wal.service.ts new file mode 100644 index 0000000..59b43df --- /dev/null +++ b/src/modules/soroban-wal/soroban-wal.service.ts @@ -0,0 +1,159 @@ +import { + ApplySorobanSideEffects, + CreateSorobanWALEntry, + RollbackSorobanOptimisticState, + SorobanTransactionGateway, + SorobanWALEntry, + SorobanWALStore, + ValidateSorobanOperation, +} from './soroban-wal.types'; + +const noValidation: ValidateSorobanOperation = async () => {}; +const noRollback: RollbackSorobanOptimisticState = async () => {}; + +export interface SorobanWALServiceOptions { + maxConfirmationAttempts?: number; + baseRetryDelayMs?: number; + sleep?: (milliseconds: number) => Promise; +} + +export interface ExecuteSorobanOperation extends CreateSorobanWALEntry { + validate?: ValidateSorobanOperation; + applySideEffects: ApplySorobanSideEffects; + rollbackOptimisticState?: RollbackSorobanOptimisticState; +} + +export class SorobanWALService { + private readonly maxConfirmationAttempts: number; + private readonly baseRetryDelayMs: number; + private readonly sleep: (milliseconds: number) => Promise; + + constructor( + private readonly store: SorobanWALStore, + private readonly gateway: SorobanTransactionGateway, + options: SorobanWALServiceOptions = {} + ) { + this.maxConfirmationAttempts = options.maxConfirmationAttempts ?? 10; + this.baseRetryDelayMs = options.baseRetryDelayMs ?? 1000; + this.sleep = + options.sleep ?? + (milliseconds => + new Promise(resolve => setTimeout(resolve, milliseconds))); + } + + async execute(input: ExecuteSorobanOperation): Promise { + const { entry, created } = await this.store.createPending( + input, + input.validate ?? noValidation + ); + + if (!created) { + if (entry.state === 'SUBMITTED' && entry.txHash) { + return this.settleSubmitted( + entry, + input.applySideEffects, + input.rollbackOptimisticState ?? noRollback + ); + } + return entry; + } + + let submitted: SorobanWALEntry; + try { + const { txHash } = await this.gateway.submit(entry.xdrPayload); + submitted = await this.store.markSubmitted(entry.id, txHash); + } catch (error) { + return this.store.markFailed( + entry.id, + getErrorMessage(error), + input.rollbackOptimisticState ?? noRollback + ); + } + + return this.settleSubmitted( + submitted, + input.applySideEffects, + input.rollbackOptimisticState ?? noRollback + ); + } + + async recover( + olderThan: Date, + applySideEffects: ApplySorobanSideEffects, + rollbackOptimisticState: RollbackSorobanOptimisticState = noRollback + ): Promise { + const entries = await this.store.listRecoverable(olderThan); + const recovered: SorobanWALEntry[] = []; + + for (const entry of entries) { + if (entry.state === 'PENDING') { + recovered.push( + await this.store.markRolledBack( + entry.id, + rollbackOptimisticState + ) + ); + continue; + } + + if (entry.state === 'SUBMITTED' && entry.txHash) { + recovered.push( + await this.settleSubmitted( + entry, + applySideEffects, + rollbackOptimisticState + ) + ); + } + } + + return recovered; + } + + private async settleSubmitted( + entry: SorobanWALEntry, + applySideEffects: ApplySorobanSideEffects, + rollbackOptimisticState: RollbackSorobanOptimisticState + ): Promise { + const txHash = entry.txHash; + if (!txHash) { + return this.store.markFailed( + entry.id, + 'SUBMITTED WAL entry is missing its transaction hash', + rollbackOptimisticState + ); + } + + for (let attempt = 0; attempt < this.maxConfirmationAttempts; attempt++) { + const result = await this.gateway.getStatus(txHash); + if (result.status === 'CONFIRMED') { + return this.store.confirmAtomically( + entry.id, + txHash, + applySideEffects + ); + } + if (result.status === 'FAILED') { + return this.store.markFailed( + entry.id, + result.error, + rollbackOptimisticState + ); + } + + if (attempt + 1 < this.maxConfirmationAttempts) { + await this.sleep(this.baseRetryDelayMs * 2 ** attempt); + } + } + + return this.store.markFailed( + entry.id, + `Transaction ${txHash} was still pending after ${this.maxConfirmationAttempts} checks`, + rollbackOptimisticState + ); + } +} + +function getErrorMessage(error: unknown): string { + return error instanceof Error ? error.message : String(error); +} diff --git a/src/modules/soroban-wal/soroban-wal.types.ts b/src/modules/soroban-wal/soroban-wal.types.ts new file mode 100644 index 0000000..ee69136 --- /dev/null +++ b/src/modules/soroban-wal/soroban-wal.types.ts @@ -0,0 +1,78 @@ +export type SorobanWALOperation = 'BUY' | 'SELL'; +export type SorobanWALState = + | 'PENDING' + | 'SUBMITTED' + | 'CONFIRMED' + | 'FAILED' + | 'ROLLED_BACK'; + +export interface SorobanWALEntry { + id: string; + idempotencyKey: string; + operation: SorobanWALOperation; + wallet: string; + creatorWallet: string; + amount: string; + expectedSupplyBefore: string; + xdrPayload: string; + state: SorobanWALState; + txHash: string | null; + submittedAt: Date | null; + confirmedAt: Date | null; + error: string | null; + createdAt: Date; + updatedAt: Date; +} + +export interface CreateSorobanWALEntry { + idempotencyKey: string; + operation: SorobanWALOperation; + wallet: string; + creatorWallet: string; + amount: string; + expectedSupplyBefore: string; + xdrPayload: string; +} + +export type DatabaseTransaction = unknown; +export type ValidateSorobanOperation = ( + transaction: DatabaseTransaction +) => Promise; +export type ApplySorobanSideEffects = ( + transaction: DatabaseTransaction, + entry: SorobanWALEntry +) => Promise; +export type RollbackSorobanOptimisticState = ApplySorobanSideEffects; + +export interface SorobanWALStore { + createPending( + input: CreateSorobanWALEntry, + validate: ValidateSorobanOperation + ): Promise<{ entry: SorobanWALEntry; created: boolean }>; + markSubmitted(id: string, txHash: string): Promise; + markFailed( + id: string, + error: string, + rollback: RollbackSorobanOptimisticState + ): Promise; + markRolledBack( + id: string, + rollback: RollbackSorobanOptimisticState + ): Promise; + confirmAtomically( + id: string, + txHash: string, + applySideEffects: ApplySorobanSideEffects + ): Promise; + listRecoverable(olderThan: Date): Promise; +} + +export type SorobanTransactionStatus = + | { status: 'PENDING' } + | { status: 'CONFIRMED' } + | { status: 'FAILED'; error: string }; + +export interface SorobanTransactionGateway { + submit(xdrPayload: string): Promise<{ txHash: string }>; + getStatus(txHash: string): Promise; +} diff --git a/src/server.ts b/src/server.ts index f0a68d8..4b53b0c 100644 --- a/src/server.ts +++ b/src/server.ts @@ -21,8 +21,15 @@ import { stopGovernanceSyncJob, } from './jobs/governance-sync.job'; import { connectRedis, disconnectRedis } from './utils/redis.utils'; -import { broadcastServerClosing, closeAllConnections } from './utils/sse-fanout.utils'; +import { + broadcastServerClosing, + closeAllConnections, +} from './utils/sse-fanout.utils'; import { buildStartupConfigSummary } from './utils/config-summary.utils'; +import { + startSorobanWALRecoveryJob, + stopSorobanWALRecoveryJob, +} from './jobs/soroban-wal-recovery.job'; async function startServer() { try { @@ -72,6 +79,7 @@ async function startServer() { startDetectPriceMovementsJob(); startGovernanceSyncJob(); + startSorobanWALRecoveryJob(); const server = app.listen(envConfig.PORT, () => { logger.info(`Server running on port ${envConfig.PORT}`); @@ -107,6 +115,7 @@ function createGracefulShutdownHandler(server: ReturnType) { stopOwnershipSnapshotCleanupJob(); stopDetectPriceMovementsJob(); stopGovernanceSyncJob(); + stopSorobanWALRecoveryJob(); await prisma.$disconnect(); logger.info('Database connection closed');