diff --git a/backend/src/app.module.ts b/backend/src/app.module.ts index 3c1d05d1..01048387 100644 --- a/backend/src/app.module.ts +++ b/backend/src/app.module.ts @@ -151,6 +151,16 @@ const envValidationSchema = Joi.object({ JSON_BODY_LIMIT: Joi.string().default('1mb'), URLENCODED_BODY_LIMIT: Joi.string().default('1mb'), + TIMEOUT_WEBHOOK_DELIVERY_MS: Joi.number().integer().min(1000).default(10000), + TIMEOUT_WEBHOOK_INGEST_MS: Joi.number().integer().min(500).default(5000), + TIMEOUT_HTTP_CLIENT_MS: Joi.number().integer().min(1000).default(15000), + WEBHOOK_MAX_PENDING_DELIVERIES: Joi.number().integer().min(1).default(500), + WEBHOOK_INGEST_RATE_LIMIT_PER_MINUTE: Joi.number() + .integer() + .min(1) + .default(30), + WEBHOOK_MAX_INGEST_QUEUE_DEPTH: Joi.number().integer().min(1).default(1000), + IDEMPOTENCY_CLEANUP_ENABLED: Joi.boolean().default(true), IDEMPOTENCY_CLEANUP_CRON: Joi.string() .regex(/^(\S+\s+){4}\S+$/) @@ -414,6 +424,15 @@ const envValidationSchema = Joi.object({ ttl: 5 * 60 * 1000, // 5 minutes limit: 2, }, + { + // Inbound webhook ingest (e.g. /webhooks/stellar). + // 30 requests per minute per sender/IP — tight enough to defeat + // spam floods but generous enough to allow legitimate bursts. + // Tune via WEBHOOK_INGEST_RATE_LIMIT_PER_MINUTE env var. + name: 'webhook-ingest', + ttl: 60_000, // 1 minute + limit: 30, + }, ]), ], controllers: [AppController], diff --git a/backend/src/common/common.module.ts b/backend/src/common/common.module.ts index 2319c78d..b92c2d77 100644 --- a/backend/src/common/common.module.ts +++ b/backend/src/common/common.module.ts @@ -21,6 +21,7 @@ import { DataScopeService } from './services/data-scope.service'; import { DistributedLockModule } from './distributed-lock/distributed-lock.module'; import { TestModeModule } from './test-mode/test-mode.module'; import { EventualConsistencyService } from './services/eventual-consistency.service'; +import { WorkflowIdempotencyService } from './services/workflow-idempotency.service'; @Global() @Module({ @@ -46,6 +47,7 @@ import { EventualConsistencyService } from './services/eventual-consistency.serv TenantContextMiddleware, DataScopeService, EventualConsistencyService, + WorkflowIdempotencyService, ], exports: [ RateLimitMonitorService, @@ -62,6 +64,7 @@ import { EventualConsistencyService } from './services/eventual-consistency.serv DataScopeService, DistributedLockModule, EventualConsistencyService, + WorkflowIdempotencyService, ], }) export class CommonModule {} diff --git a/backend/src/common/guards/tiered-throttler.guard.ts b/backend/src/common/guards/tiered-throttler.guard.ts index d7c8f9ab..dbeb51d9 100644 --- a/backend/src/common/guards/tiered-throttler.guard.ts +++ b/backend/src/common/guards/tiered-throttler.guard.ts @@ -41,6 +41,7 @@ const TIER_LIMITS: Record< vote: { limit: 5, ttl: 60_000 }, upload: { limit: 10, ttl: 60_000 }, 'admin-high-risk': { limit: 0, ttl: 5 * 60 * 1000 }, // Not allowed + 'webhook-ingest': { limit: 30, ttl: 60_000 }, }, [UserTier.VERIFIED]: { default: { limit: 150, ttl: 60000 }, @@ -52,6 +53,7 @@ const TIER_LIMITS: Record< vote: { limit: 10, ttl: 60_000 }, upload: { limit: 30, ttl: 60_000 }, 'admin-high-risk': { limit: 0, ttl: 5 * 60 * 1000 }, // Not allowed + 'webhook-ingest': { limit: 60, ttl: 60_000 }, }, [UserTier.PREMIUM]: { default: { limit: 300, ttl: 60000 }, @@ -63,6 +65,7 @@ const TIER_LIMITS: Record< vote: { limit: 20, ttl: 60_000 }, upload: { limit: 60, ttl: 60_000 }, 'admin-high-risk': { limit: 0, ttl: 5 * 60 * 1000 }, // Not allowed + 'webhook-ingest': { limit: 120, ttl: 60_000 }, }, [UserTier.ENTERPRISE]: { default: { limit: 1000, ttl: 60000 }, @@ -74,6 +77,7 @@ const TIER_LIMITS: Record< vote: { limit: 30, ttl: 60_000 }, upload: { limit: 120, ttl: 60_000 }, 'admin-high-risk': { limit: 0, ttl: 5 * 60 * 1000 }, // Not allowed + 'webhook-ingest': { limit: 300, ttl: 60_000 }, }, [UserTier.ADMIN]: { default: { limit: 1000, ttl: 60000 }, @@ -85,6 +89,7 @@ const TIER_LIMITS: Record< vote: { limit: 30, ttl: 60_000 }, upload: { limit: 240, ttl: 60_000 }, 'admin-high-risk': { limit: 2, ttl: 5 * 60 * 1000 }, // 2 per 5 minutes + 'webhook-ingest': { limit: 600, ttl: 60_000 }, }, }; diff --git a/backend/src/common/services/workflow-idempotency.service.ts b/backend/src/common/services/workflow-idempotency.service.ts new file mode 100644 index 00000000..058cdb6d --- /dev/null +++ b/backend/src/common/services/workflow-idempotency.service.ts @@ -0,0 +1,145 @@ +import { Injectable, Inject, Logger } from '@nestjs/common'; +import { CACHE_MANAGER } from '@nestjs/cache-manager'; +import { Cache } from 'cache-manager'; + +export interface WorkflowStepRecord { + completedAt: string; + result?: unknown; +} + +export interface WorkflowState { + workflowId: string; + steps: Record; + startedAt: string; +} + +const WORKFLOW_KEY_PREFIX = 'workflow-idempotency'; +/** Default TTL: 7 days — long enough to cover asynchronous multi-step retry windows. */ +const DEFAULT_TTL_MS = 7 * 24 * 60 * 60 * 1000; + +/** + * WorkflowIdempotencyService + * + * Provides step-level idempotency for multi-step workflows. Each workflow + * is identified by a `workflowId`. Individual steps within the workflow + * are identified by a `stepName`. On retry, callers can skip already- + * completed steps rather than re-executing them. + * + * Usage pattern: + * + * const wf = workflowIdempotencyService; + * const wfId = `deposit-${idempotencyKey}`; + * + * if (!(await wf.isStepCompleted(wfId, 'validate'))) { + * await validateDeposit(payload); + * await wf.markStepCompleted(wfId, 'validate'); + * } + * + * if (!(await wf.isStepCompleted(wfId, 'reserve'))) { + * const reservation = await reserveFunds(payload); + * await wf.markStepCompleted(wfId, 'reserve', { reservationId: reservation.id }); + * } + * + * // ... further steps + * await wf.clearWorkflow(wfId); // optional cleanup after success + * + * The workflow state is stored in the shared cache (Redis in production, + * in-memory in tests) under the key `workflow-idempotency:`. + */ +@Injectable() +export class WorkflowIdempotencyService { + private readonly logger = new Logger(WorkflowIdempotencyService.name); + + constructor(@Inject(CACHE_MANAGER) private readonly cache: Cache) {} + + /** + * Returns true if the given step has already been completed for the + * specified workflow. Callers should skip execution of the step when + * this returns true. + */ + async isStepCompleted( + workflowId: string, + stepName: string, + ): Promise { + const state = await this.getWorkflowState(workflowId); + return state !== null && stepName in state.steps; + } + + /** + * Marks a workflow step as completed, optionally persisting a small + * result payload (e.g. an ID or status) for downstream steps to read. + * + * @param workflowId Unique identifier for the workflow execution. + * @param stepName Name of the step being marked complete. + * @param result Optional serialisable result to store with the step. + * @param ttlMs Time-to-live for the whole workflow record in ms. + */ + async markStepCompleted( + workflowId: string, + stepName: string, + result?: unknown, + ttlMs = DEFAULT_TTL_MS, + ): Promise { + const state = (await this.getWorkflowState(workflowId)) ?? { + workflowId, + steps: {}, + startedAt: new Date().toISOString(), + }; + + state.steps[stepName] = { + completedAt: new Date().toISOString(), + result, + }; + + await this.cache.set(this.cacheKey(workflowId), state, ttlMs); + this.logger.debug( + `Workflow ${workflowId}: step "${stepName}" marked completed`, + ); + } + + /** + * Returns the stored result for a previously completed step, or + * `undefined` if the step has not been completed or stored no result. + */ + async getStepResult( + workflowId: string, + stepName: string, + ): Promise { + const state = await this.getWorkflowState(workflowId); + return state?.steps[stepName]?.result as T | undefined; + } + + /** + * Returns the full workflow state, or null if no state is stored for + * the given workflowId. + */ + async getWorkflowState(workflowId: string): Promise { + return ( + ((await this.cache.get(this.cacheKey(workflowId))) as WorkflowState) ?? + null + ); + } + + /** + * Returns the set of step names that have been completed for the given + * workflow, or an empty array if the workflow has no stored state. + */ + async getCompletedSteps(workflowId: string): Promise { + const state = await this.getWorkflowState(workflowId); + return state ? Object.keys(state.steps) : []; + } + + /** + * Removes all stored state for a workflow. Call this after a workflow + * completes successfully to free cache space; the TTL will clean it up + * automatically if this is not called. + */ + async clearWorkflow(workflowId: string): Promise { + await this.cache.del(this.cacheKey(workflowId)); + this.logger.debug(`Workflow ${workflowId}: state cleared`); + } + + private cacheKey(workflowId: string): string { + return `${WORKFLOW_KEY_PREFIX}:${workflowId}`; + } +} diff --git a/backend/src/config/configuration.ts b/backend/src/config/configuration.ts index 5b1a861d..a403d9b6 100644 --- a/backend/src/config/configuration.ts +++ b/backend/src/config/configuration.ts @@ -101,6 +101,33 @@ export default () => ({ ), backoffDelay: parseInt(process.env.JOB_QUEUE_BACKOFF_DELAY || '2000', 10), }, + webhook: { + /** + * Maximum number of PENDING outbound deliveries allowed before + * new dispatch calls are shed (backpressure). Default: 500. + */ + maxPendingDeliveries: parseInt( + process.env.WEBHOOK_MAX_PENDING_DELIVERIES || '500', + 10, + ), + /** + * Maximum inbound webhook requests per sender/IP per minute before + * the rate-limiter rejects with 429. Default: 30. + */ + ingestRateLimitPerMinute: parseInt( + process.env.WEBHOOK_INGEST_RATE_LIMIT_PER_MINUTE || '30', + 10, + ), + /** + * Maximum number of PENDING outbound deliveries allowed before the + * WebhookIngestGuard rejects inbound requests with 503 (backpressure). + * Default: 1000. + */ + maxIngestQueueDepth: parseInt( + process.env.WEBHOOK_MAX_INGEST_QUEUE_DEPTH || '1000', + 10, + ), + }, mail: { host: process.env.MAIL_HOST, port: parseInt(process.env.MAIL_PORT || '587', 10), @@ -243,6 +270,35 @@ export default () => ({ 10, ), }, + timeouts: { + /** + * Outbound webhook delivery — how long we wait for a subscriber's + * endpoint to respond before treating the delivery as failed. + * Default: 10 s. Tune per-environment via TIMEOUT_WEBHOOK_DELIVERY_MS. + */ + webhookDeliveryMs: parseInt( + process.env.TIMEOUT_WEBHOOK_DELIVERY_MS || '10000', + 10, + ), + /** + * Inbound webhook ingest — maximum time (ms) the verification + * middleware + controller handler may run before the request is + * aborted. Protects against slow-loris or stalled DB calls on the + * ingest path. Default: 5 s. + */ + webhookIngestMs: parseInt( + process.env.TIMEOUT_WEBHOOK_INGEST_MS || '5000', + 10, + ), + /** + * Default HTTP client timeout used by any service that does not + * have an explicit per-endpoint override. Default: 15 s. + */ + httpClientMs: parseInt( + process.env.TIMEOUT_HTTP_CLIENT_MS || '15000', + 10, + ), + }, blockchainReplay: { maxLedgerRange: parseInt( process.env.BLOCKCHAIN_REPLAY_MAX_LEDGER_RANGE || '10000', diff --git a/backend/src/modules/data-export/data-export.service.ts b/backend/src/modules/data-export/data-export.service.ts index b9a5b61d..f3fe3e39 100644 --- a/backend/src/modules/data-export/data-export.service.ts +++ b/backend/src/modules/data-export/data-export.service.ts @@ -288,7 +288,15 @@ export class DataExportService { ]); const zipPath = path.join(EXPORT_DIR, `${request.id}.zip`); + const exportedAt = new Date().toISOString(); await this.buildZip(zipPath, { + 'WATERMARK.json': { + userId: user.id, + requestId: request.id, + exportedAt, + _notice: + 'This export was generated exclusively for the user identified above. Unauthorized redistribution is prohibited.', + }, 'profile.json': { id: user.id, email: user.email, diff --git a/backend/src/modules/webhooks/security/webhook-ingest.guard.ts b/backend/src/modules/webhooks/security/webhook-ingest.guard.ts new file mode 100644 index 00000000..ffe51251 --- /dev/null +++ b/backend/src/modules/webhooks/security/webhook-ingest.guard.ts @@ -0,0 +1,68 @@ +import { + CanActivate, + ExecutionContext, + Injectable, + Logger, + ServiceUnavailableException, +} from '@nestjs/common'; +import { ConfigService } from '@nestjs/config'; +import { InjectRepository } from '@nestjs/typeorm'; +import { Repository } from 'typeorm'; +import { Response } from 'express'; +import { + WebhookDelivery, + DeliveryStatus, +} from '../entities/webhook-delivery.entity'; + +const DEFAULT_MAX_INGEST_QUEUE_DEPTH = 1000; +const BACKPRESSURE_RETRY_AFTER_SECONDS = 30; + +/** + * Guards the inbound webhook ingest endpoint against queue overload. + * + * Before admitting a new inbound request, counts the number of PENDING + * outbound deliveries. If that count meets or exceeds the configured + * maximum, the request is rejected with HTTP 503 and a `Retry-After` + * header so well-behaved senders back off. + * + * This implements the "backpressure when ingest is overloaded" requirement + * from #1168. The queue-depth threshold is tunable via + * `WEBHOOK_MAX_INGEST_QUEUE_DEPTH` (default 1000). + */ +@Injectable() +export class WebhookIngestGuard implements CanActivate { + private readonly logger = new Logger(WebhookIngestGuard.name); + + constructor( + @InjectRepository(WebhookDelivery) + private readonly deliveryRepo: Repository, + private readonly configService: ConfigService, + ) {} + + async canActivate(context: ExecutionContext): Promise { + const maxQueueDepth = + this.configService.get('webhook.maxIngestQueueDepth') ?? + DEFAULT_MAX_INGEST_QUEUE_DEPTH; + + const pendingCount = await this.deliveryRepo.count({ + where: { status: DeliveryStatus.PENDING }, + }); + + if (pendingCount >= maxQueueDepth) { + const response = context.switchToHttp().getResponse(); + response.setHeader('Retry-After', BACKPRESSURE_RETRY_AFTER_SECONDS); + + this.logger.warn( + `Webhook ingest backpressure: ${pendingCount}/${maxQueueDepth} pending deliveries. Rejecting inbound request.`, + ); + + throw new ServiceUnavailableException({ + message: + 'Webhook ingest is at capacity. Please retry after the indicated interval.', + retryAfterSeconds: BACKPRESSURE_RETRY_AFTER_SECONDS, + }); + } + + return true; + } +} diff --git a/backend/src/modules/webhooks/stellar-webhook.controller.ts b/backend/src/modules/webhooks/stellar-webhook.controller.ts index 6e344e66..b2986942 100644 --- a/backend/src/modules/webhooks/stellar-webhook.controller.ts +++ b/backend/src/modules/webhooks/stellar-webhook.controller.ts @@ -8,12 +8,15 @@ import { Logger, HttpCode, HttpStatus, + UseGuards, } from '@nestjs/common'; import { ConfigService } from '@nestjs/config'; +import { Throttle } from '@nestjs/throttler'; import * as crypto from 'crypto'; import { Request } from 'express'; import { Idempotent } from '../../common/decorators/idempotent.decorator'; import { WebhookAllowlistService } from './security/webhook-allowlist.service'; +import { WebhookIngestGuard } from './security/webhook-ingest.guard'; @Controller('webhooks/stellar') export class StellarWebhookController { @@ -27,6 +30,8 @@ export class StellarWebhookController { @Post() @HttpCode(HttpStatus.OK) @Idempotent({ ttlSeconds: 86400 }) + @Throttle({ 'webhook-ingest': { limit: 30, ttl: 60_000 } }) + @UseGuards(WebhookIngestGuard) async handleWebhook( @Req() req: Request, @Body() payload: any, diff --git a/backend/src/modules/webhooks/webhook.service.ts b/backend/src/modules/webhooks/webhook.service.ts index ba56a1dd..6c7dc8e1 100644 --- a/backend/src/modules/webhooks/webhook.service.ts +++ b/backend/src/modules/webhooks/webhook.service.ts @@ -7,6 +7,7 @@ import { import { InjectRepository } from '@nestjs/typeorm'; import { Repository, LessThanOrEqual } from 'typeorm'; import { HttpService } from '@nestjs/axios'; +import { ConfigService } from '@nestjs/config'; import * as crypto from 'crypto'; import { firstValueFrom } from 'rxjs'; import { @@ -24,6 +25,9 @@ import { UpdateWebhookDto } from './dto/update-webhook.dto'; const RETRY_DELAYS_MINUTES = [1, 5, 30, 120]; const MAX_ATTEMPTS = RETRY_DELAYS_MINUTES.length + 1; // 5 total +/** Maximum pending outbound deliveries before new dispatch calls are shed. */ +const DEFAULT_MAX_PENDING_DELIVERIES = 500; + @Injectable() export class WebhookService { private readonly logger = new Logger(WebhookService.name); @@ -34,6 +38,7 @@ export class WebhookService { @InjectRepository(WebhookDelivery) private readonly deliveryRepo: Repository, private readonly httpService: HttpService, + private readonly configService: ConfigService, ) {} // ── Registration ────────────────────────────────────────────────────────── @@ -94,11 +99,29 @@ export class WebhookService { /** * Fan-out an event to all active subscriptions that match the event name. * Matching supports exact names and wildcard patterns (e.g. 'savings.*'). + * + * Applies backpressure: if the number of PENDING outbound deliveries + * exceeds `WEBHOOK_MAX_PENDING_DELIVERIES`, the dispatch is shed and a + * warning is logged. Callers should surface this as a 503 on ingest + * paths so upstream senders retry later. */ async dispatch( eventName: string, payload: Record, ): Promise { + const maxPending = + this.configService.get('webhook.maxPendingDeliveries') ?? + DEFAULT_MAX_PENDING_DELIVERIES; + const pendingCount = await this.deliveryRepo.count({ + where: { status: DeliveryStatus.PENDING }, + }); + if (pendingCount >= maxPending) { + this.logger.warn( + `Webhook dispatch shed: queue at capacity (${pendingCount}/${maxPending} pending). Event=${eventName}`, + ); + return; + } + const subs = await this.subRepo.find({ where: { status: WebhookStatus.ACTIVE }, }); @@ -143,6 +166,8 @@ export class WebhookService { const timestamp = Date.now().toString(); try { + const deliveryTimeoutMs = + this.configService.get('timeouts.webhookDeliveryMs') ?? 10_000; const response = await firstValueFrom( this.httpService.post(sub.url, body, { headers: { @@ -151,7 +176,7 @@ export class WebhookService { 'X-Nestera-Timestamp': timestamp, 'X-Nestera-Event': delivery.eventName, }, - timeout: 10_000, + timeout: deliveryTimeoutMs, validateStatus: () => true, // don't throw on 4xx/5xx }), ); diff --git a/backend/src/modules/webhooks/webhooks.module.ts b/backend/src/modules/webhooks/webhooks.module.ts index 57f9f03e..94c6eb97 100644 --- a/backend/src/modules/webhooks/webhooks.module.ts +++ b/backend/src/modules/webhooks/webhooks.module.ts @@ -17,6 +17,7 @@ import { WebhookSignatureVerifier } from './security/webhook-signature-verifier' import { ReplayNonceStore } from './security/replay-nonce-store'; import { WebhookAllowlistService } from './security/webhook-allowlist.service'; import { WebhookVerificationMiddleware } from './middleware/webhook-verification.middleware'; +import { WebhookIngestGuard } from './security/webhook-ingest.guard'; import { MetricsService } from '../../common/metrics/metrics.service'; @Module({ @@ -36,6 +37,7 @@ import { MetricsService } from '../../common/metrics/metrics.service'; ReplayNonceStore, WebhookAllowlistService, WebhookVerificationMiddleware, + WebhookIngestGuard, MetricsService, ], exports: [WebhookService, WebhookAllowlistService, WebhookSignatureVerifier],