1import { 2 Inject, 3 CallHandler, 4 ExecutionContext, 5 Injectable, 6 NestInterceptor, 7 SetMetadata, 8} from "@nestjs/common"; 9import { createHash } from "node:crypto"; 10import { Observable, from, of } from "rxjs"; 11import { switchMap, tap } from "rxjs/operators"; 12import { RedisService } from "../redis.service.js"; 13import { logger } from "../logger.js"; 14 15/** 16 * Idempotent-write middleware (TASK-0095 R1) — the server half of the offline 17 * outbox. The client tags every REST mutation with an `Idempotency-Key: <key>` 18 * header (a client-generated `key_` + 32 hex); a mutation whose idempotent 19 * replay would be wrong (a `POST /records` create — the second send must not 20 * double-create) is exactly what the key prevents. 21 * 22 * RULE 29 — recorded deviation: the task spec names this a *middleware* at 23 * `common/middleware/idempotency.ts`. A true Express/Nest **middleware runs 24 * BEFORE the global guard** (auth), so it has no `req.auth` (the `userId` the 25 * `idem:{userId}:{key}` keyspace needs) and could not capture the response 26 * body. The correct Nest primitive is a **global interceptor** (it runs after 27 * the guard, so `req.auth` is present, and it wraps the handler result, so it 28 * can store the response). The spec's FILE PATH is respected 29 * (`common/middleware/idempotency.ts`); the primitive is an interceptor. 30 * 31 * The standard idempotent-write pattern (key checked BEFORE execution): 32 * - a SEEN key (a Redis `idem:{userId}:{key}` set by a prior execution, 33 * TTL 24 h) → the STORED response is replayed (the handler is NOT 34 * re-run — a double-replay creates no duplicate); 35 * - a FRESH key → the handler runs; a SUCCESSFUL (2xx) result is stored 36 * under the key (a 4xx/5xx was rejected, not executed — nothing to 37 * replay; the outbox moves the item to its failed list). 38 * 39 * Scope (the 6-case matrix, the spec's named unit surface): 40 * 1. key HIT → the stored body is replayed, the handler is skipped; 41 * 2. key MISS → the handler runs, a 2xx body is stored; 42 * 3. expiry → a key past its 24 h TTL reads as MISS (Redis enforces the 43 * TTL; the client re-executes — a day-old mutation is a new 44 * attempt, matching the outbox's retry=fresh-key rule); 45 * 4. storage → only a 2xx body is stored (a 4xx/5xx is not); 46 * 5. GET excl. → a GET never carries a key (the client only tags mutations); 47 * a GET with a key is a no-op here (defensive — the keyspace 48 * is for writes); 49 * 6. no-header → a request without `Idempotency-Key` passes through 50 * untouched (every Phase-3/5 endpoint keeps working — the 51 * middleware is additive). 52 * 53 * Fail-open: a Redis error degrades to "execute, don't store" (the mutation 54 * still succeeds; the worst case is a duplicate, never a lost write — the 55 * same degradation policy as the rate limiter). Logged, not thrown. 56 * 57 * Concurrency note (recorded): the 6-case matrix does not include two LIVE 58 * clients racing the SAME key (a true single-flight would need a SETNX lock). 59 * The key is per-tab unique (client-generated), so a same-key race only 60 * happens if two clients deliberately share a key — a client bug, not a 61 * protocol case. A crash-mid-drain REPLAY (the outbox's actual concern) is 62 * sequential and fully covered. 63 */ 64 65/** The 24 h keyspace TTL (the task's stored-response bound). */ 66export const IDEM_TTL_SECONDS = 24 * 60 * 60; 67 68/** 69 * The single-flight lock TTL (docs §9: "concurrent duplicate → the 70 * in-flight result is awaited, Redis lock, 10 s"). A MISS caller takes 71 * `idem:lock:{redisKey}` (SET NX) before executing; a concurrent duplicate 72 * of the SAME operation (same credential + route + body digest) that misses 73 * the memo but cannot take the lock WAITS for the memo the first caller 74 * stores, instead of executing a second time. The lock is never released 75 * explicitly — the memo (checked BEFORE the lock) supersedes it, and the 10 s 76 * TTL is the backstop. 77 */ 78export const IDEM_LOCK_TTL_SECONDS = 10; 79 80/** Poll cadence for a waiter on an in-flight duplicate (≤ ~1.5 s). */ 81const IDEM_WAIT_POLL_MS = 75; 82const IDEM_WAIT_POLLS = 20; 83 84/** The header the client tags mutations with (the frozen shared contract). */ 85export const IDEMPOTENCY_KEY_HEADER = "idempotency-key"; 86 87/** 88 * RULE 29 — recorded deviation (TASK-0126): a route that OWNS its idempotency 89 * and streams (SSE) must opt OUT of this body-replay interceptor. The 90 * interceptor's replay contract is "re-run the handler with the STORED BODY" — 91 * for a stream the handler writes the socket itself, so a key HIT that returns 92 * the stored body WITHOUT running the handler leaves the `res` never written 93 * nor ended (the client hangs). SSE endpoints (the AI generate endpoint) 94 * implement idempotency in-band instead: a key HIT re-emits the terminal frame 95 * (the created resource is not re-created — the same "no double-create" 96 * guarantee, SSE-safe). `@SkipIdempotency()` marks such a route; the header is 97 * still honored (read by the handler), just not by the body interceptor. 98 */ 99export const SKIP_IDEMPOTENCY_KEY = "skip_idempotency"; 100export const SkipIdempotency = (): MethodDecorator & ClassDecorator & PropertyDecorator => 101 SetMetadata(SKIP_IDEMPOTENCY_KEY, true); 102 103/** A stored body that was EMPTY (204 / nil body) — replayed as `undefined`. */ 104const EMPTY_BODY = "__IDEM_EMPTY__"; 105 106/** The JSON the keyspace stores under a key. */ 107interface IdemRecord { 108 /** The body the API wrote (EMPTY_BODY marker for a nil body). */ 109 readonly body: string; 110} 111 112/** The mutating HTTP methods the key applies to (GET is excluded). */ 113const MUTATING = new Set(["POST", "PATCH", "PUT", "DELETE"]); 114 115@Injectable() 116export class IdempotencyInterceptor implements NestInterceptor { 117 constructor(@Inject(RedisService) private readonly redis: RedisService) {} 118 119 intercept(context: ExecutionContext, next: CallHandler): Observable<unknown> { 120 const req = context.switchToHttp().getRequest<{ 121 method: string; 122 path: string; 123 body?: unknown; 124 auth?: 125 { kind?: string; userId?: string; apiKeyId?: string; workspaceId?: string } | undefined; 126 headers: Record<string, string | string[] | undefined>; 127 }>(); 128 const res = context 129 .switchToHttp() 130 .getResponse<{ statusCode: number; setHeader: (k: string, v: string) => void }>(); 131 132 // GATE (cases 5 + 6 + auth): only an authenticated actor's MUTATING 133 // request carrying an `Idempotency-Key` header is honored. TASK-0127: 134 // API keys are first-class idempotent callers (CI retries are exactly 135 // the replay case the header exists for) — the keyspace is scoped by 136 // CREDENTIAL, so no cross-key / cross-workspace replay is possible 137 // (spec §Security: "the idempotency memo is scoped by credential"). 138 // Everything else (GET / no header / unauthenticated) passes through 139 // untouched — the middleware is additive, existing endpoints unchanged. 140 const key = IdempotencyInterceptor.headerValue(req.headers?.[IDEMPOTENCY_KEY_HEADER]); 141 const handler = context.getHandler?.(); 142 const cls = context.getClass?.(); 143 const skip = 144 (handler != null && Reflect.getMetadata(SKIP_IDEMPOTENCY_KEY, handler) === true) || 145 (cls != null && Reflect.getMetadata(SKIP_IDEMPOTENCY_KEY, cls) === true); 146 const auth = req.auth; 147 const credential = 148 auth?.kind === "user" && auth.userId 149 ? { scope: `user:${auth.userId}` } 150 : auth?.kind === "api" && auth.apiKeyId && auth.workspaceId 151 ? { scope: `api:${auth.workspaceId}:${auth.apiKeyId}` } 152 : null; 153 if (skip || !MUTATING.has(req.method) || !key || !credential) { 154 return next.handle(); 155 } 156 157 // TASK-0127: for API keys the memo identity is credential + ROUTE + 158 // BODY digest (a different body with the same key is a different 159 // operation — the spec's "body is in the memo hash" case). The user 160 // keyspace keeps the TASK-0095 identity (`idem:{userId}:{key}`) for 161 // compatibility with already-stored outbox records. 162 const digest = credential.scope.startsWith("api:") 163 ? createHash("sha256") 164 .update(`${req.method} ${req.path}`) 165 .update(JSON.stringify(req.body ?? null)) 166 .digest("hex") 167 .slice(0, 32) 168 : ""; 169 const redisKey = credential.scope.startsWith("api:") 170 ? `idem:${credential.scope}:${digest}:${key}` 171 : `idem:${auth!.userId}:${key}`; 172 173 // FAIL-OPEN HIT check (cases 1 + 3): a stored body → replay it (the 174 // handler is NOT re-run); a miss or a Redis error → the single-flight 175 // MISS path below (a Redis error degrades to "execute, don't store"). 176 return from(this.lookup(redisKey)).pipe( 177 switchMap((stored) => { 178 if (stored !== null) { 179 // TASK-0127: the replayed response carries the marker header 180 // (docs/api/authentication.md §9: "Idempotency-Replayed: true 181 // when a cached result is returned within 24 h"). 182 res.setHeader("Idempotency-Replayed", "true"); 183 return of(IdempotencyInterceptor.decodeBody(stored.body)); 184 } 185 // MISS: acquire the single-flight lock (docs §9: "concurrent 186 // duplicate → the in-flight result is awaited, Redis lock, 10 s"), 187 // then either execute (acquired) or wait for the in-flight caller's 188 // memo (a concurrent duplicate of the same operation). 189 return from(this.acquireLock(redisKey)).pipe( 190 switchMap((acquired) => { 191 if (acquired) { 192 return this.executeAndStore(next, res, redisKey); 193 } 194 return from(this.waitForMemo(redisKey)).pipe( 195 switchMap((memo) => { 196 if (memo !== null) { 197 res.setHeader("Idempotency-Replayed", "true"); 198 return of(IdempotencyInterceptor.decodeBody(memo.body)); 199 } 200 // The in-flight caller did not store a memo (a non-2xx 201 // result, a crash, or a Redis hiccup) within the wait 202 // window — fail open: execute, then store (the same 203 // degradation as a lookup miss). 204 logger.warn(`idempotency wait timed out (${redisKey}) — executing (fail-open)`); 205 return this.executeAndStore(next, res, redisKey); 206 }), 207 ); 208 }), 209 ); 210 }), 211 ); 212 } 213 214 /** Run the handler once; store a 2xx body under the memo key. */ 215 private executeAndStore( 216 next: CallHandler, 217 res: { statusCode: number }, 218 redisKey: string, 219 ): Observable<unknown> { 220 return next.handle().pipe( 221 tap({ 222 next: (data) => { 223 const status = res.statusCode; 224 if (status >= 200 && status < 300) { 225 void this.store(redisKey, data); 226 } 227 }, 228 }), 229 ); 230 } 231 232 /** Take the single-flight lock (SET NX EX 10). Fail-open (Redis down). */ 233 private async acquireLock(redisKey: string): Promise<boolean> { 234 try { 235 const ok = await this.redis 236 .getRedis() 237 .set(`idem:lock:${redisKey}`, "1", "EX", IDEM_LOCK_TTL_SECONDS, "NX"); 238 return ok === "OK"; 239 } catch { 240 return true; // Redis down → execute (availability over strictness) 241 } 242 } 243 244 /** Poll the memo for an in-flight duplicate's result (≤ ~1.5 s). */ 245 private async waitForMemo(redisKey: string): Promise<IdemRecord | null> { 246 for (let i = 0; i < IDEM_WAIT_POLLS; i++) { 247 await new Promise((r) => setTimeout(r, IDEM_WAIT_POLL_MS)); 248 const memo = await this.lookup(redisKey); 249 if (memo !== null) return memo; 250 } 251 return null; 252 } 253 254 /** Read + parse a stored record (null = miss / error → execute). */ 255 private async lookup(redisKey: string): Promise<IdemRecord | null> { 256 try { 257 const raw = await this.redis.getRedis().get(redisKey); 258 if (raw === null) return null; 259 return JSON.parse(raw) as IdemRecord; 260 } catch (err) { 261 logger.warn(`idempotency lookup failed (${redisKey}): ${String(err)}`); 262 return null; // fail-open → execute 263 } 264 } 265 266 /** Store a 2xx body under the key (TTL 24 h). Fail-open on Redis error. */ 267 private async store(redisKey: string, data: unknown): Promise<void> { 268 const body = data === undefined || data === null ? EMPTY_BODY : JSON.stringify(data); 269 try { 270 await this.redis 271 .getRedis() 272 .set(redisKey, JSON.stringify({ body } satisfies IdemRecord), "EX", IDEM_TTL_SECONDS); 273 } catch (err) { 274 logger.warn(`idempotency store failed (${redisKey}): ${String(err)}`); 275 } 276 } 277 278 /** Reconstruct the handler's result from the stored body. */ 279 private static decodeBody(body: string): unknown { 280 if (body === EMPTY_BODY) return undefined; // 204 / nil → an empty response 281 try { 282 return JSON.parse(body); 283 } catch { 284 return undefined; 285 } 286 } 287 288 /** The header value (a single value, or the first of a repeated one). */ 289 private static headerValue(v: string | string[] | undefined): string | undefined { 290 if (Array.isArray(v)) return v[0]; 291 if (typeof v === "string" && v.length > 0) return v; 292 return undefined; 293 } 294} 295