|
| 1 | +/** |
| 2 | + * Idempotency middleware. |
| 3 | + * |
| 4 | + * Prevents duplicate payment submission to Stellar on network retries: a |
| 5 | + * client sends an `Idempotency-Key` header; the first request is processed and |
| 6 | + * its response cached; identical retries within the TTL return the original |
| 7 | + * response without re-submitting to the chain. Pairs with the contract's |
| 8 | + * nonce-uniqueness guard for defence in depth. |
| 9 | + * |
| 10 | + * Storage is pluggable: a Redis-backed store for production (atomic SET NX), |
| 11 | + * and an in-memory TTL store for dev/tests when Redis is unavailable. |
| 12 | + */ |
| 13 | + |
| 14 | +import { Request, Response, NextFunction } from 'express'; |
| 15 | + |
| 16 | +export interface CachedResponse { |
| 17 | + status: number; |
| 18 | + body: unknown; |
| 19 | + /** Headers worth replaying (content-type only) */ |
| 20 | + headers?: Record<string, string>; |
| 21 | +} |
| 22 | + |
| 23 | +export interface IdempotencyStore { |
| 24 | + /** Returns 'acquired' if we won the lock, 'processing' if a request is in |
| 25 | + * flight, 'exists' if a final result is already cached. */ |
| 26 | + tryAcquire(key: string, ttlSeconds: number): Promise<'acquired' | 'processing' | 'exists'>; |
| 27 | + getResult(key: string): Promise<CachedResponse | undefined>; |
| 28 | + setResult(key: string, value: CachedResponse, ttlSeconds: number): Promise<void>; |
| 29 | + /** Release the in-flight lock without caching a result (e.g. on 5xx). */ |
| 30 | + releaseLock(key: string): Promise<void>; |
| 31 | +} |
| 32 | + |
| 33 | +const OK = 'OK'; |
| 34 | + |
| 35 | +/** Redis-backed store using atomic SET NX EX for the lock. */ |
| 36 | +export class RedisIdempotencyStore implements IdempotencyStore { |
| 37 | + constructor(private client: { set: (k: string, v: string, mode: string, ttl: string, seconds: number) => Promise<string | null>; get: (k: string) => Promise<string | null>; setex: (k: string, ttl: number, v: string) => Promise<string | null>; del: (k: string) => Promise<number>; }) {} |
| 38 | + |
| 39 | + async tryAcquire(key: string, ttlSeconds: number): Promise<'acquired' | 'processing' | 'exists'> { |
| 40 | + const resultKey = this.resultKey(key); |
| 41 | + // If a final result already exists, replay it. |
| 42 | + const existing = await this.client.get(resultKey); |
| 43 | + if (existing) return 'exists'; |
| 44 | + // Try to acquire the processing lock. |
| 45 | + const acquired = await this.client.set(key, 'processing', 'NX', 'EX', ttlSeconds); |
| 46 | + return acquired === OK ? 'acquired' : 'processing'; |
| 47 | + } |
| 48 | + |
| 49 | + async getResult(key: string): Promise<CachedResponse | undefined> { |
| 50 | + const raw = await this.client.get(this.resultKey(key)); |
| 51 | + if (!raw) return undefined; |
| 52 | + try { |
| 53 | + return JSON.parse(raw) as CachedResponse; |
| 54 | + } catch { |
| 55 | + return undefined; |
| 56 | + } |
| 57 | + } |
| 58 | + |
| 59 | + async setResult(key: string, value: CachedResponse, ttlSeconds: number): Promise<void> { |
| 60 | + await this.client.setex(this.resultKey(key), ttlSeconds, JSON.stringify(value)); |
| 61 | + await this.client.del(key); // release the processing lock |
| 62 | + } |
| 63 | + |
| 64 | + async releaseLock(key: string): Promise<void> { |
| 65 | + await this.client.del(key); |
| 66 | + } |
| 67 | + |
| 68 | + private resultKey = (key: string) => `${key}:result`; |
| 69 | +} |
| 70 | + |
| 71 | +interface MemEntry { value: 'processing' | CachedResponse; expiresAt: number; } |
| 72 | + |
| 73 | +/** In-memory TTL store (dev/tests/fallback). */ |
| 74 | +export class MemoryIdempotencyStore implements IdempotencyStore { |
| 75 | + private store = new Map<string, MemEntry>(); |
| 76 | + private now: () => number; |
| 77 | + constructor(now: () => number = Date.now) { this.now = now; } |
| 78 | + |
| 79 | + private prune(key: string): void { |
| 80 | + const e = this.store.get(key); |
| 81 | + if (e && e.expiresAt < this.now()) this.store.delete(key); |
| 82 | + } |
| 83 | + |
| 84 | + async tryAcquire(key: string, ttlSeconds: number): Promise<'acquired' | 'processing' | 'exists'> { |
| 85 | + this.prune(key); |
| 86 | + const e = this.store.get(key); |
| 87 | + if (e && e.value !== 'processing') return 'exists'; |
| 88 | + if (e && e.value === 'processing') return 'processing'; |
| 89 | + this.store.set(key, { value: 'processing', expiresAt: this.now() + ttlSeconds * 1000 }); |
| 90 | + return 'acquired'; |
| 91 | + } |
| 92 | + |
| 93 | + async getResult(key: string): Promise<CachedResponse | undefined> { |
| 94 | + this.prune(key); |
| 95 | + const e = this.store.get(key); |
| 96 | + return e && e.value !== 'processing' ? (e.value as CachedResponse) : undefined; |
| 97 | + } |
| 98 | + |
| 99 | + async setResult(key: string, value: CachedResponse, ttlSeconds: number): Promise<void> { |
| 100 | + this.store.set(key, { value, expiresAt: this.now() + ttlSeconds * 1000 }); |
| 101 | + } |
| 102 | + |
| 103 | + async releaseLock(key: string): Promise<void> { |
| 104 | + this.store.delete(key); |
| 105 | + } |
| 106 | +} |
| 107 | + |
| 108 | +export interface IdempotencyOptions { |
| 109 | + headerName?: string; |
| 110 | + ttlSeconds?: number; |
| 111 | + store?: IdempotencyStore; |
| 112 | +} |
| 113 | + |
| 114 | +/** Build a scoped cache key from method + route + client key. */ |
| 115 | +function buildKey(req: Request, clientKey: string): string { |
| 116 | + const route = (req.route?.path || req.path || '').toString().slice(0, 64); |
| 117 | + return `idem:${req.method}:${route}:${clientKey}`; |
| 118 | +} |
| 119 | + |
| 120 | +/** |
| 121 | + * Express middleware. If no `Idempotency-Key` header is present the request |
| 122 | + * passes through unchanged (the contract nonce still guards on-chain replay). |
| 123 | + */ |
| 124 | +export function idempotency(opts: IdempotencyOptions = {}) { |
| 125 | + const headerName = (opts.headerName || 'idempotency-key').toLowerCase(); |
| 126 | + const ttlSeconds = opts.ttlSeconds ?? 24 * 60 * 60; // 24h >= Stellar finality windows |
| 127 | + const store = opts.store || new MemoryIdempotencyStore(); |
| 128 | + |
| 129 | + return async (req: Request, res: Response, next: NextFunction) => { |
| 130 | + const clientKey = (req.headers[headerName] as string | undefined)?.trim(); |
| 131 | + if (!clientKey) return next(); // optional; contract nonce still protects |
| 132 | + |
| 133 | + const key = buildKey(req, clientKey); |
| 134 | + const state = await store.tryAcquire(key, ttlSeconds); |
| 135 | + |
| 136 | + if (state === 'exists') { |
| 137 | + const cached = await store.getResult(key); |
| 138 | + if (cached) { |
| 139 | + res.status(cached.status); |
| 140 | + if (cached.headers) for (const [k, v] of Object.entries(cached.headers)) res.set(k, v); |
| 141 | + res.set('X-Idempotent-Replay', 'true'); |
| 142 | + return res.json(cached.body); |
| 143 | + } |
| 144 | + } |
| 145 | + if (state === 'processing') { |
| 146 | + return res.status(409).json({ error: 'IDEMPOTENCY_IN_FLIGHT', message: 'A request with this Idempotency-Key is already being processed' }); |
| 147 | + } |
| 148 | + |
| 149 | + // state === 'acquired': capture the response and cache it. |
| 150 | + const originalJson = res.json.bind(res); |
| 151 | + res.json = ((body: unknown) => { |
| 152 | + const captured: CachedResponse = { status: res.statusCode, body }; |
| 153 | + // Only cache successful + client-error responses; 5xx should be retryable, |
| 154 | + // so release the in-flight lock to let the client retry with the same key. |
| 155 | + if (res.statusCode < 500) { |
| 156 | + store.setResult(key, captured, ttlSeconds).catch(() => { /* non-fatal */ }); |
| 157 | + } else { |
| 158 | + store.releaseLock(key).catch(() => { /* non-fatal */ }); |
| 159 | + } |
| 160 | + return originalJson(body); |
| 161 | + }) as Response['json']; |
| 162 | + |
| 163 | + next(); |
| 164 | + }; |
| 165 | +} |
0 commit comments