Largepayloads

.env.example

@@ -2,6 +2,11 @@
 # PostgreSQL connection URL (used by Prisma) — REQUIRED
 DATABASE_URL="postgresql://parashield:secret@localhost:5432/parashield_db?schema=public"
 
+# Read replica URL (#439) — optional. When set, all read-only queries are
+# routed to this instance via PrismaService.reader; writes and transactions
+# always go to the primary DATABASE_URL. Use the same format as DATABASE_URL.
+# DATABASE_REPLICA_URL="postgresql://parashield:secret@replica-host:5432/parashield_db?schema=public"
+
 # Prisma connection pool tuning (#381) — optional.
 # The backend applies these automatically to the datasource URL at runtime;
 # values already present as DATABASE_URL query params (e.g. ?connection_limit=10)
@@ -110,5 +115,5 @@ CORS_ORIGIN="http://localhost:3000"
 
 # Optional tuning (defaults shown). See README "CORS configuration".
 # CORS_METHODS=GET,POST,PUT,DELETE,OPTIONS
-# CORS_ALLOWED_HEADERS=Content-Type,Authorization,x-wallet-address,x-wallet-signature,x-wallet-message,x-api-key,x-admin-api-key
+# CORS_ALLOWED_HEADERS=Content-Type,Authorization,x-wallet-address,x-wallet-signature,x-wallet-message,x-api-key,x-admin-api-key,Idempotency-Key
 # CORS_CREDENTIALS=false

src/claims/claims.controller.ts

@@ -1,4 +1,5 @@
-import { Body, Controller, ForbiddenException, Get, Param, Post, Query, Req, UseGuards, UnauthorizedException, NotFoundException, Throttle } from '@nestjs/common';
+import { Body, Controller, ForbiddenException, Get, Param, Post, Query, Req, UseGuards, UseInterceptors, UnauthorizedException, NotFoundException, Throttle } from '@nestjs/common';
+import { StreamingInterceptor } from '../common/interceptors/streaming.interceptor';
 import {
   ApiTags,
   ApiOperation,
@@ -55,11 +56,18 @@ export class ClaimsController {
   /** GET /api/v1/claims?wallet=... — get claim history for the authenticated wallet */
   @Get()
   @UseGuards(JwtAuthGuard)
+  @UseInterceptors(StreamingInterceptor)
   @ApiBearerAuth()
   @ApiOperation({ summary: 'Get claim history for a wallet address (query param)' })
   @ApiQuery({ name: 'wallet', required: true, description: 'Stellar wallet address' })
   @ApiQuery({ name: 'page', required: false, description: 'Page number' })
   @ApiQuery({ name: 'limit', required: false, description: 'Items per page' })
+  @ApiQuery({
+    name: 'stream',
+    required: false,
+    description: "Set to 'true' to receive the data array as NDJSON (one item per line). Alternatively send Accept: application/x-ndjson. Pagination metadata available in X-Total-Count, X-Page, X-Limit headers.",
+    example: 'true',
+  })
   @ApiResponse({
     status: 200,
     description: 'Returns paginated claim history — { success, data, total, page, limit }',
@@ -138,11 +146,18 @@ export class ClaimsController {
   /** GET /api/v1/claims/history/:wallet — get all claims for a wallet address */
   @Get('history/:wallet')
   @UseGuards(JwtAuthGuard)
+  @UseInterceptors(StreamingInterceptor)
   @ApiBearerAuth()
   @ApiOperation({ summary: 'Get all claims for a wallet address' })
   @ApiParam({ name: 'wallet', description: 'Stellar wallet address' })
   @ApiQuery({ name: 'page', required: false, description: 'Page number' })
   @ApiQuery({ name: 'limit', required: false, description: 'Items per page' })
+  @ApiQuery({
+    name: 'stream',
+    required: false,
+    description: "Set to 'true' to receive the data array as NDJSON (one item per line). Alternatively send Accept: application/x-ndjson. Pagination metadata available in X-Total-Count, X-Page, X-Limit headers.",
+    example: 'true',
+  })
   @ApiResponse({
     status: 200,
     description: 'Returns paginated claim history — { success, data, total, page, limit }',

src/common/interceptors/streaming.interceptor.ts

@@ -0,0 +1,123 @@
+import {
+  Injectable,
+  NestInterceptor,
+  ExecutionContext,
+  CallHandler,
+  StreamableFile,
+} from '@nestjs/common';
+import { Observable } from 'rxjs';
+import { switchMap } from 'rxjs/operators';
+import { PassThrough } from 'stream';
+import { Request, Response } from 'express';
+
+/**
+ * StreamingInterceptor — streams large list responses as NDJSON to reduce
+ * peak memory usage (#440).
+ *
+ * Instead of buffering the full result array and serialising it as a single
+ * JSON payload, the interceptor writes each item as an individual JSON line
+ * (newline-delimited JSON / NDJSON) through a PassThrough Node stream. The
+ * Express response is flushed chunk-by-chunk with Transfer-Encoding: chunked,
+ * so the server never holds the entire serialised payload in memory at once.
+ *
+ * Opt-in trigger (either condition activates streaming):
+ *   • Request header:  Accept: application/x-ndjson
+ *   • Query parameter: ?stream=true
+ *
+ * When neither condition is present the interceptor is a no-op and the
+ * standard JSON response path is used unchanged.
+ *
+ * The NDJSON format uses one JSON line per item:
+ *   {"id":"...","status":"ACTIVE",...}\n
+ *   {"id":"...","status":"ACTIVE",...}\n
+ *   ...
+ *
+ * Clients that need the full metadata envelope (success, total, page, limit)
+ * can either use the standard (non-streaming) response or read the custom
+ * response headers:
+ *   X-Total-Count  — total number of items in the list
+ *   X-Page         — page number returned (when applicable)
+ *   X-Limit        — page size (when applicable)
+ *
+ * Usage:
+ *   @UseInterceptors(StreamingInterceptor)
+ *   @Get('some-list')
+ *   async list(...) { ... }
+ */
+@Injectable()
+export class StreamingInterceptor implements NestInterceptor {
+  intercept(context: ExecutionContext, next: CallHandler): Observable<unknown> {
+    const http     = context.switchToHttp();
+    const request  = http.getRequest<Request>();
+    const response = http.getResponse<Response>();
+
+    const wantsStream =
+      request.headers['accept'] === 'application/x-ndjson' ||
+      request.query['stream'] === 'true';
+
+    if (!wantsStream) {
+      return next.handle();
+    }
+
+    return next.handle().pipe(
+      switchMap((payload: unknown) => {
+        // Only stream responses that carry a data array.
+        // Non-list responses (single objects, errors) pass through unchanged.
+        if (
+          payload === null ||
+          typeof payload !== 'object' ||
+          !Array.isArray((payload as Record<string, unknown>)['data'])
+        ) {
+          return new Observable(subscriber => {
+            subscriber.next(payload);
+            subscriber.complete();
+          });
+        }
+
+        const envelope  = payload as Record<string, unknown>;
+        const items     = envelope['data'] as unknown[];
+        const total     = envelope['total'];
+        const page      = envelope['page'];
+        const limit     = envelope['limit'];
+
+        // Expose pagination metadata via response headers so NDJSON clients
+        // don't have to parse a wrapper object just to get count info.
+        response.setHeader('Content-Type', 'application/x-ndjson; charset=utf-8');
+        if (total !== undefined) {
+          response.setHeader('X-Total-Count', String(total));
+        }
+        if (page !== undefined) {
+          response.setHeader('X-Page', String(page));
+        }
+        if (limit !== undefined) {
+          response.setHeader('X-Limit', String(limit));
+        }
+
+        // Build a PassThrough stream and write each item as a JSON line.
+        const passThrough = new PassThrough();
+
+        // Schedule writes asynchronously so the stream is returned to NestJS
+        // before we start pushing data (avoids blocking the event loop).
+        setImmediate(() => {
+          try {
+            for (const item of items) {
+              passThrough.write(JSON.stringify(item) + '\n');
+            }
+            passThrough.end();
+          } catch (err) {
+            passThrough.destroy(err instanceof Error ? err : new Error(String(err)));
+          }
+        });
+
+        // Return a StreamableFile so NestJS hands off the stream to Express
+        // and uses Transfer-Encoding: chunked automatically.
+        return new Observable(subscriber => {
+          subscriber.next(new StreamableFile(passThrough, {
+            type: 'application/x-ndjson',
+          }));
+          subscriber.complete();
+        });
+      }),
+    );
+  }
+}

src/common/middleware/idempotency.middleware.ts

@@ -0,0 +1,102 @@
+import { Injectable, NestMiddleware, Logger } from '@nestjs/common';
+import { Request, Response, NextFunction } from 'express';
+import Redis from 'ioredis';
+
+// #438 — Idempotency keys prevent duplicate processing of the same request.
+// A client that retries a POST (e.g. after a network timeout) sends the same
+// Idempotency-Key header; we return the cached response instead of re-running
+// the handler. Only applies to mutating methods (POST, PUT, PATCH).
+const IDEMPOTENCY_TTL_SECONDS = 86_400; // 24 h — matches typical API conventions
+const IDEMPOTENCY_KEY_MAX_LENGTH = 255;
+
+/**
+ * IdempotencyMiddleware (#438)
+ *
+ * Intercepts POST/PUT/PATCH requests that include an `Idempotency-Key` header.
+ *
+ * First call  — processes normally, caches the response body + status in Redis
+ *               for 24 hours under the key `idempotency:{method}:{path}:{key}`.
+ * Repeat call — returns the cached response immediately with a
+ *               `X-Idempotent-Replayed: true` header so callers can detect it.
+ *
+ * Requests without the header pass through unchanged, so the middleware is
+ * completely opt-in and does not affect existing clients.
+ *
+ * Register in main.ts after body parsers:
+ *   const idempotency = new IdempotencyMiddleware(redisClient);
+ *   app.use((req, res, next) => idempotency.use(req, res, next));
+ */
+@Injectable()
+export class IdempotencyMiddleware implements NestMiddleware {
+  private readonly logger = new Logger(IdempotencyMiddleware.name);
+
+  constructor(private readonly redis: Redis) {}
+
+  use(req: Request, res: Response, next: NextFunction): void {
+    const MUTATING_METHODS = ['POST', 'PUT', 'PATCH'];
+    if (!MUTATING_METHODS.includes(req.method)) {
+      return next();
+    }
+
+    const rawKey = req.headers['idempotency-key'] as string | undefined;
+    if (!rawKey) {
+      return next();
+    }
+
+    // Basic validation — reject keys that are too long or contain newlines
+    // (could be used to construct arbitrary Redis keys).
+    if (rawKey.length > IDEMPOTENCY_KEY_MAX_LENGTH || /[\r\n]/.test(rawKey)) {
+      res.status(400).json({
+        success: false,
+        errorCode: 'BAD_REQUEST',
+        error: 'Invalid Idempotency-Key header value',
+        statusCode: 400,
+      });
+      return;
+    }
+
+    const storeKey = `idempotency:${req.method}:${req.path}:${rawKey}`;
+
+    this.redis.get(storeKey).then((cached) => {
+      if (cached) {
+        try {
+          const { status, body } = JSON.parse(cached) as { status: number; body: unknown };
+          this.logger.debug(`Replaying idempotent response for key=${rawKey}`);
+          res.setHeader('X-Idempotent-Replayed', 'true');
+          res.status(status).json(body);
+          return;
+        } catch {
+          // Corrupted cache entry — fall through to normal processing.
+          this.logger.warn(`Failed to parse cached idempotency entry for key=${rawKey}`);
+        }
+      }
+
+      // Intercept the response so we can cache it before it is sent.
+      const originalJson = res.json.bind(res);
+      res.json = (body: unknown) => {
+        // Only cache successful responses (2xx) to avoid caching transient errors.
+        if (res.statusCode >= 200 && res.statusCode < 300) {
+          const entry = JSON.stringify({ status: res.statusCode, body });
+          this.redis
+            .set(storeKey, entry, 'EX', IDEMPOTENCY_TTL_SECONDS)
+            .catch((err: unknown) =>
+              this.logger.warn(
+                `Failed to store idempotency key=${rawKey}: ${err instanceof Error ? err.message : String(err)}`,
+              ),
+            );
+        }
+        return originalJson(body);
+      };
+
+      next();
+    }).catch((err: unknown) => {
+      // Redis unavailable — fail open (process the request normally) so a
+      // Redis outage doesn't take down the API. Log as a warning so ops can
+      // detect the degradation.
+      this.logger.warn(
+        `Idempotency Redis lookup failed for key=${rawKey}: ${err instanceof Error ? err.message : String(err)}. Processing request without idempotency check.`,
+      );
+      next();
+    });
+  }
+}

src/health/dto/health-response.dto.ts

@@ -43,6 +43,15 @@ export class StellarCheckDto {
   @ApiProperty({ description: 'Keeper account native XLM balance (7-decimal fixed point)', required: false })
   keeperBalanceXlm?: string;
 
+  @ApiProperty({ description: 'Stellar RPC (Soroban) connectivity status', enum: ['ok', 'error'], required: false })
+  rpcStatus?: 'ok' | 'error';
+
+  @ApiProperty({ description: 'Stellar RPC round-trip latency in milliseconds', required: false })
+  rpcLatencyMs?: number;
+
+  @ApiProperty({ description: 'Latest ledger sequence number returned by Stellar RPC', required: false })
+  rpcLedger?: number;
+
   @ApiProperty({ description: 'Error message when status is "error"', required: false })
   error?: string;
 }

src/health/health.controller.ts

@@ -61,6 +61,10 @@ export class HealthController {
     let stellarStatus: 'ok' | 'error' = 'ok';
     let stellarError: string | undefined;
     let keeperBalanceXlm: string | undefined;
+    // #441 — Direct Soroban RPC connectivity check fields
+    let stellarRpcStatus: 'ok' | 'error' = 'ok';
+    let stellarRpcLatencyMs: number | undefined;
+    let stellarRpcLedger: number | undefined;
     let queueStatus: 'ok' | 'error' = 'ok';
     let queueError: string | undefined;
 
@@ -96,6 +100,24 @@ export class HealthController {
       // Non-fatal: pg_stat_activity may be restricted on managed databases.
     }
 
+    // #441 — Direct Stellar RPC (Soroban) connectivity check.
+    // getLatestLedger is the lightest available probe: it requires no
+    // authentication, touches no account state, and always succeeds when
+    // the RPC node is reachable. We record latency so ops can distinguish
+    // a slow node from a fully unreachable one. A failure here is fatal
+    // (stellarStatus → 'error') because contract invocations — claims,
+    // policy submissions, oracle writes — all depend on the Soroban RPC.
+    try {
+      const rpcProbe = await this.stellar.checkRpcConnectivity(HEALTH_CHECK_RPC_TIMEOUT_MS);
+      stellarRpcLatencyMs = rpcProbe.latencyMs;
+      stellarRpcLedger    = rpcProbe.ledger;
+    } catch (err) {
+      stellarRpcStatus = 'error';
+      stellarStatus    = 'error';
+      stellarError     = `Stellar RPC unreachable: ${err instanceof Error ? err.message : String(err)}`;
+      this.logger.error(`Health check: ${stellarError}`);
+    }
+
     try {
       keeperBalanceXlm = await this.stellar.getAccountBalance(
         this.stellar.keeperKeypair.publicKey(),
@@ -117,7 +139,7 @@ export class HealthController {
       }
     } catch (err) {
       stellarStatus = 'error';
-      this.logger.error(`Health check Stellar RPC failed: ${err instanceof Error ? err.message : String(err)}`);
+      this.logger.error(`Health check Stellar keeper/Horizon failed: ${err instanceof Error ? err.message : String(err)}`);
     }
 
     // #403 — Redis/message queue connectivity check.
@@ -229,6 +251,9 @@ export class HealthController {
       },
       stellar: {
         status: stellarStatus,
+        rpcStatus: stellarRpcStatus,
+        ...(stellarRpcLatencyMs !== undefined ? { rpcLatencyMs: stellarRpcLatencyMs } : {}),
+        ...(stellarRpcLedger !== undefined ? { rpcLedger: stellarRpcLedger } : {}),
         ...(keeperBalanceXlm !== undefined ? { keeperBalanceXlm } : {}),
         ...(stellarError ? { error: stellarError } : {}),
       },

src/main.ts

@@ -10,12 +10,14 @@ import { JsonLogger } from './common/logging/json-logger.service';
 import { InputSanitizationMiddleware } from './common/middleware/input-sanitization.middleware';
 import { RequestTimeoutMiddleware } from './common/middleware/request-timeout.middleware';
 import { UsdcPrecisionValidationMiddleware } from './common/middleware/usdc-precision-validation.middleware';
+import { IdempotencyMiddleware } from './common/middleware/idempotency.middleware';
 import { loadVaultSecrets } from './common/secrets/vault-secrets.loader';
 import { applyRateLimitHeaders } from './common/swagger/rate-limit-headers';
 import { initializeOpenTelemetry } from './common/telemetry/opentelemetry';
 import helmet from 'helmet';
 import { ConfigService } from '@nestjs/config';
 import { json, urlencoded } from 'express';
+import Redis from 'ioredis';
 
 const REQUEST_BODY_LIMIT = '1mb';
 const SERVER_TIMEOUT_MS = 30_000;
@@ -31,6 +33,7 @@ const DEFAULT_CORS_ALLOWED_HEADERS = [
   'x-wallet-message',
   'x-api-key',
   'x-admin-api-key',
+  'Idempotency-Key',
 ];
 
 function parseCsvEnv(value: string | undefined): string[] | undefined {
@@ -79,6 +82,13 @@ async function bootstrap() {
   const usdcValidator = new UsdcPrecisionValidationMiddleware();
   app.use((req, res, next) => usdcValidator.use(req, res, next));
 
+  // #438 — idempotency key deduplication for mutating requests (POST/PUT/PATCH).
+  // Clients include an `Idempotency-Key` header; replayed requests with the
+  // same key get the cached response instead of re-executing the handler.
+  const redisClient = app.get<Redis>('REDIS_CLIENT');
+  const idempotency = new IdempotencyMiddleware(redisClient);
+  app.use((req, res, next) => idempotency.use(req, res, next));
+
   // Global exception filter
   app.useGlobalFilters(new GlobalExceptionFilter());