Idempotency

pnpm dlx blockend-cli add idempotency

/**
 * Master barrel export for the idempotency library.
 *
 * Re-exports the complete public API so consumers can import everything from a
 * single entry point, exactly as they could from the previous monolithic
 * `IdempotencyHandler` module:
 *
 *   import { IdempotencyHandler, DEFAULT_CACHE_TTL } from "idempotency/core";
 *
 * The module is decomposed into focused sub-modules (errors, interfaces, types,
 * utils, services, jobs) — see `core/idempotency-handler.ts` for the main class.
 */

// Errors
export { IDEMPOTENCY_ERROR_CODES } from "./errors/codes";
export type { IdempotencyErrorCode } from "./errors/codes";
export { IdempotencyError } from "./errors/idempotency-errors";

// Interfaces
export type { IdempotencyCache } from "./interfaces/cache";
export type {
  DeleteExpiredParams,
  FindAllFilters,
  IdempotencyStore,
  RecoverStuckRecordsOptions,
  RecoveryResult
} from "./interfaces/store";
export { noopLogger, noopMetrics } from "./interfaces/observability";
export type { Logger, Metrics, MetricName } from "./interfaces/observability";

// Types
export type {
  CachedResponse,
  CreateIdempotencyRecord,
  CreateProcessingResult,
  CreateRecordResult,
  IdempotencyRecord,
  IdempotencyStatus
} from "./types/index";

// Services
export { CacheManager, DEFAULT_CACHE_TTL } from "./services/cache-manager";

// Utils
export {
  DEFAULT_MAX_KEY_LENGTH,
  DEFAULT_MIN_KEY_LENGTH,
  UUID_REGEX,
  validateKey
} from "./utils/key-validator";
export type { KeyValidationOptions } from "./utils/key-validator";
export { hashRequest, serialize } from "./utils/serializer";

// Jobs
export { cleanupExpiredRecords } from "./jobs/record-cleaner";
export type { CleanupExpiredOptions, CleanupResult } from "./jobs/record-cleaner";
export { recoverStuckRecords } from "./jobs/record-recovery";

// Core handler
export { IdempotencyHandler } from "./core/idempotency-handler";
export type { IdempotencyHandlerOptions } from "./core/idempotency-handler";

import { IDEMPOTENCY_ERROR_CODES } from "../errors/codes";
import { IdempotencyError } from "../errors/idempotency-errors";
import type { IdempotencyCache } from "../interfaces/cache";
import type { Logger, Metrics } from "../interfaces/observability";
import { noopLogger, noopMetrics } from "../interfaces/observability";
import type {
  IdempotencyStore,
  RecoverStuckRecordsOptions,
  RecoveryResult
} from "../interfaces/store";
import type {
  CreateIdempotencyRecord,
  CreateProcessingResult,
  IdempotencyRecord
} from "../types/index";
import { validateKey } from "../utils/key-validator";
import type { KeyValidationOptions } from "../utils/key-validator";
import { hashRequest, serialize } from "../utils/serializer";
import { constantTimeEquals } from "../utils/constant-time-equals";
import { CacheManager } from "../services/cache-manager";
import { cleanupExpiredRecords } from "../jobs/record-cleaner";
import type { CleanupExpiredOptions, CleanupResult } from "../jobs/record-cleaner";
import { recoverStuckRecords } from "../jobs/record-recovery";

export type IdempotencyHandlerOptions = {
  logger?: Logger;
  metrics?: Metrics;
};

const DEFAULT_RECORD_TTL = 24 * 60 * 60 * 1000;
const MAX_ALLOWED_RETRIES = 3;

/**
 * IdempotencyHandler wraps business logic so that retries of the same request
 * are safe: a request keyed by (operation, userId, key) executes exactly once,
 * and subsequent attempts receive the stored result instead.
 *
 * The class deliberately focuses on orchestration only. All heavy lifting is
 * delegated to focused modules:
 *   - key validation   -> utils/key-validator
 *   - hashing          -> utils/serializer
 *   - L1 cache         -> services/cache-manager
 *   - background jobs  -> jobs/record-cleaner, jobs/record-recovery
 *
 * Public API (execute, validateKey, serialize, hashRequest,
 * cleanupExpiredRecords, recoverStuckRecords, retry) is fully
 * backward-compatible with the previous monolithic implementation.
 */
export class IdempotencyHandler {
  private readonly logger: Logger;
  private readonly metrics: Metrics;
  private readonly cacheManager: CacheManager;

  constructor(
    private readonly store: IdempotencyStore,
    cache?: IdempotencyCache,
    options: IdempotencyHandlerOptions = {}
  ) {
    this.logger = options.logger ?? noopLogger;
    this.metrics = options.metrics ?? noopMetrics;
    this.cacheManager = new CacheManager(cache, this.logger, this.metrics);
  }

  /**
   * Sanitizes and validates the incoming idempotency key.
   * Delegates to `utils/key-validator`, which trims BEFORE length checks so
   * whitespace alone can't bypass length limits or bloat storage.
   *
   * @returns the cleaned key; callers MUST use this returned value, not the raw input.
   */
  public validateKey(key: unknown, options?: KeyValidationOptions): string {
    return validateKey(key, options);
  }

  /**
   * Deterministically stringifies a payload structure (recursively sorted keys,
   * order-preserving arrays, non-ambiguous primitive encodings). Rejects
   * non-deterministic values (undefined, NaN, Infinity, BigInt, functions,
   * symbols). See `utils/serializer` for details.
   */
  public static serialize(value: unknown): string {
    return serialize(value);
  }

  /**
   * Produces a stable SHA-256 fingerprint of a request payload so identical
   * payloads (regardless of key order) hash identically.
   */
  public hashRequest(body: unknown): string {
    return hashRequest(body);
  }

  /**
   * Intended to be run periodically by a background worker or cron job.
   *
   * SAFETY INVARIANT: only deletes records with status 'SUCCESS' or 'FAILED'.
   * Records marked 'PROCESSING' are intentionally excluded here, even if
   * expired — purging an active lock would allow a duplicate incoming request
   * to grab a new lock and run concurrently. Stuck locks are handled by
   * `recoverStuckRecords()` instead.
   */
  public async cleanupExpiredRecords(options?: CleanupExpiredOptions): Promise<CleanupResult> {
    return cleanupExpiredRecords(this.store, this.logger, this.metrics, options);
  }

  /**
   * Recovers operations stuck in 'PROCESSING' (e.g., node crashed mid-execution)
   * by marking them 'FAILED' so clients can safely retry or inspect the outcome.
   */
  public async recoverStuckRecords(options?: RecoverStuckRecordsOptions): Promise<RecoveryResult> {
    return recoverStuckRecords(this.store, this.logger, this.metrics, options);
  }

  /**
   * Manually re-opens a failed record to 'PROCESSING' to allow retry attempts.
   */
  public async retry(
    userId: string,
    operation: string,
    key: string
  ): Promise<IdempotencyRecord | null> {
    const record = await this.store.find(key, operation, userId);

    if (!record) {
      return null;
    }

    // Already completed successfully; no retry needed
    if (record.status === "SUCCESS") {
      return record;
    }

    // Another thread or node is already processing this job
    if (record.status === "PROCESSING") {
      this.metrics.increment("idempotency_duplicates_total", 1, {
        operation,
        status: "processing"
      });
      throw new IdempotencyError(
        IDEMPOTENCY_ERROR_CODES.REQUEST_IN_PROGRESS,
        "The request is already being processed."
      );
    }

    // Re-acquire lock for failed operation
    if (record.status === "FAILED") {
      await this.markRecordProcessing(userId, operation, key);

      return {
        ...record,
        status: "PROCESSING",
        updatedAt: new Date()
      };
    }

    throw new IdempotencyError(
      IDEMPOTENCY_ERROR_CODES.STORE_UNAVAILABLE,
      "Unable to retry the idempotency record."
    );
  }

  /**
   * Main entrypoint for wrapping business logic with idempotency.
   *
   * Flow:
   * 1. Validate key -> Hash body payload.
   * 2. Check L1 cache (Redis/Memory). Hit -> return response immediately.
   * 3. Acquire DB lock (create PROCESSING record).
   * 4. Run `execute()` callback.
   * 5. Success -> Mark DB record SUCCESS & sync L1 cache.
   * 6. Error -> Check if error is transient or permanent:
   *    - Permanent error: Mark FAILED immediately (blocks identical retries).
   *    - Transient error: Increment retry count. Delete record if retries
   *      remain so the client can retry.
   */
  public async execute(
    userId: string,
    operation: string,
    key: string,
    body: unknown,
    execute: () => Promise<unknown>,
    options?: { isPermanentError?: (err: unknown) => boolean }
  ): Promise<unknown> {
    const startTime = performance.now();

    // The entire flow — including key validation and body hashing — runs
    // inside the try so the latency histogram is always emitted, even when an
    // attacker-controlled key/body is rejected up front.
    try {
      const validKey = this.validateKey(key);
      const requestHash = this.hashRequest(body);

      // 1. Fast path: check short-lived cache hit
      const cachedResponse = await this.cacheManager.getCachedResponse(
        userId,
        operation,
        validKey,
        requestHash
      );

      if (cachedResponse?.status === "HIT") {
        return cachedResponse.data.response;
      }

      const expiresAt = new Date(Date.now() + DEFAULT_RECORD_TTL);

      // 2. Lock phase: Atomic insert into DB
      const result = await this.createProcessingRecord({
        key: validKey,
        request_hash: requestHash,
        response: null,
        status: "PROCESSING",
        expires_at: expiresAt,
        user_id: userId,
        operation
      });

      // 3. Return cached DB result if request was completed previously
      if (result.kind === "EXISTING_SUCCESS") {
        return result.response;
      }

      // 4. Lock acquired successfully -> run actual business logic
      try {
        const response = await execute();

        await this.markRecordSuccess(userId, operation, validKey, response);

        // Async write to L1 cache; non-blocking failure
        await this.cacheManager.setCachedResponse(
          userId,
          operation,
          validKey,
          requestHash,
          response
        );

        this.logger.info(
          { key: validKey, userId, operation, status: "executed_success" },
          "Idempotent operation executed and recorded successfully"
        );

        return response;
      } catch (error) {
        // FAIL CLOSED: if the caller's error classifier itself throws, treat
        // the failure as permanent (mark the record FAILED) and rethrow the
        // ORIGINAL error — never let a classifier bug swallow the real cause
        // or leave the lock dangling.
        let permanent = true;

        try {
          permanent = options?.isPermanentError ? options.isPermanentError(error) : true;
        } catch {
          permanent = true;
        }

        if (permanent) {
          // Permanent business logic failure (e.g., validation fail) -> save as FAILED
          await this.markRecordFailed(userId, operation, validKey);
        } else {
          // Transient failure (e.g., DB connection drop) -> increment counter
          const retryCount = await this.incrementRetryCount(userId, operation, validKey);

          if (retryCount >= MAX_ALLOWED_RETRIES) {
            await this.markRecordFailed(userId, operation, validKey);
          } else {
            // Delete lock row so caller can retry immediately on next request
            await this.deleteRecord(userId, operation, validKey);
          }
        }

        this.logger.error(
          {
            key: validKey,
            userId,
            operation,
            permanent,
            error: error instanceof Error ? error.message : error
          },
          "Idempotent operation execution failed"
        );

        throw error;
      }
    } finally {
      const durationInSeconds = (performance.now() - startTime) / 1000;
      this.metrics.histogram("idempotency_latency_seconds", durationInSeconds, {
        operation
      });
    }
  }

  // ---------------------------------------------------------------------------
  // Private store-coordination helpers (used by execute()/retry() orchestration)
  // ---------------------------------------------------------------------------

  /**
   * Attempts an atomic insert of a PROCESSING lock record.
   *
   * - On `CREATED`: the lock is ours; execute the business logic.
   * - On `DUPLICATE`: the key already exists; inspect the existing row and
   *   either return its stored SUCCESS response, or reject with the
   *   appropriate idempotency error (in-progress / failed / hash mismatch).
   */
  private async createProcessingRecord({
    key,
    request_hash,
    response,
    expires_at,
    status,
    user_id,
    operation
  }: CreateIdempotencyRecord & {
    operation: string;
  }): Promise<CreateProcessingResult> {
    const record: IdempotencyRecord = {
      key,
      requestHash: request_hash,
      response,
      expiresAt: expires_at,
      status,
      userId: user_id,
      createdAt: new Date(),
      updatedAt: new Date(),
      operation
    };

    try {
      // Primary DB call: attempts an atomic insert (e.g. INSERT INTO ... ON CONFLICT DO NOTHING)
      const result = await this.store.create(record);

      if (result.status === "CREATED") {
        return { kind: "CREATED", record };
      }

      // Insert failed due to duplicate key; fetch existing row state
      const existingData = await this.findRecordByKey(user_id, operation, key);

      return this.handleExistingRecord(existingData, request_hash);
    } catch (error) {
      if (error instanceof IdempotencyError) {
        throw error;
      }

      this.metrics.increment("idempotency_store_errors_total", 1, {
        operation
      });
      this.logger.error(
        {
          key,
          userId: user_id,
          operation,
          error: error instanceof Error ? error.message : error
        },
        "Idempotency store error during record creation"
      );

      throw new IdempotencyError(
        IDEMPOTENCY_ERROR_CODES.STORE_UNAVAILABLE,
        "The idempotency store is currently unavailable."
      );
    }
  }

  /**
   * Inspects a pre-existing record (the insert was a DUPLICATE) and decides the
   * outcome: return the stored SUCCESS response, or throw the correct error.
   */
  private handleExistingRecord(
    existingData: IdempotencyRecord,
    requestHash: string
  ): CreateProcessingResult {
    this.metrics.increment("idempotency_duplicates_total", 1, {
      operation: existingData.operation,
      status: existingData.status.toLowerCase()
    });

    // Key reuse with payload mismatch -> reject immediately. Compared in
    // constant time so hash probing can't leak prefix information.
    if (!constantTimeEquals(existingData.requestHash, requestHash)) {
      this.logger.warn(
        {
          key: existingData.key,
          userId: existingData.userId,
          operation: existingData.operation
        },
        "Idempotency key reused with different request body"
      );
      throw new IdempotencyError(
        IDEMPOTENCY_ERROR_CODES.KEY_REUSED_WITH_DIFFERENT_REQUEST,
        "Idempotency key was already used with a different request."
      );
    }

    if (existingData.status === "SUCCESS") {
      this.logger.info(
        {
          key: existingData.key,
          userId: existingData.userId,
          operation: existingData.operation
        },
        "Returning existing success response from store"
      );
      return {
        kind: "EXISTING_SUCCESS",
        // Return the stored response payload (not the full record) so callers
        // receive the same shape as a fresh execution.
        response: existingData.response
      };
    }

    if (existingData.status === "PROCESSING") {
      this.logger.warn(
        {
          key: existingData.key,
          userId: existingData.userId,
          operation: existingData.operation
        },
        "Concurrent request blocked; key is currently processing"
      );
      throw new IdempotencyError(
        IDEMPOTENCY_ERROR_CODES.REQUEST_IN_PROGRESS,
        "The request is already being processed."
      );
    }

    if (existingData.status === "FAILED") {
      throw new IdempotencyError(
        IDEMPOTENCY_ERROR_CODES.REQUEST_FAILED,
        "The previous request with this idempotency key failed."
      );
    }

    throw new IdempotencyError(
      IDEMPOTENCY_ERROR_CODES.STORE_UNAVAILABLE,
      "Unknown idempotency record status."
    );
  }

  private async findRecordByKey(userId: string, operation: string, key: string) {
    const record = await this.store.find(key, operation, userId);

    if (!record) {
      throw new IdempotencyError(
        IDEMPOTENCY_ERROR_CODES.RECORD_NOT_FOUND,
        "No idempotency record found."
      );
    }

    return record;
  }

  private async markRecordSuccess(
    userId: string,
    operation: string,
    key: string,
    response: unknown
  ): Promise<void> {
    try {
      await this.store.markSuccess(key, response, userId, operation);
    } catch (error) {
      this.metrics.increment("idempotency_store_errors_total", 1, {
        operation
      });
      if (error instanceof IdempotencyError) {
        throw error;
      }

      throw new IdempotencyError(
        IDEMPOTENCY_ERROR_CODES.STORE_UNAVAILABLE,
        "Failed to update the record to success."
      );
    }
  }

  private async markRecordProcessing(
    userId: string,
    operation: string,
    key: string
  ): Promise<void> {
    try {
      await this.store.markProcessing(key, operation, userId);
    } catch (error) {
      this.metrics.increment("idempotency_store_errors_total", 1, {
        operation
      });
      if (error instanceof IdempotencyError) {
        throw error;
      }

      throw new IdempotencyError(
        IDEMPOTENCY_ERROR_CODES.STORE_UNAVAILABLE,
        "Failed to update the record to processing."
      );
    }
  }

  private async markRecordFailed(userId: string, operation: string, key: string): Promise<void> {
    try {
      await this.store.markFailed(key, operation, userId);
    } catch (error) {
      this.metrics.increment("idempotency_store_errors_total", 1, {
        operation
      });
      if (error instanceof IdempotencyError) {
        throw error;
      }

      throw new IdempotencyError(
        IDEMPOTENCY_ERROR_CODES.STORE_UNAVAILABLE,
        "Failed to update the record to failed."
      );
    }
  }

  private async incrementRetryCount(
    userId: string,
    operation: string,
    key: string
  ): Promise<number> {
    try {
      return await this.store.incrementRetryCount(key, operation, userId);
    } catch (error) {
      this.metrics.increment("idempotency_store_errors_total", 1, {
        operation
      });
      if (error instanceof IdempotencyError) {
        throw error;
      }

      throw new IdempotencyError(
        IDEMPOTENCY_ERROR_CODES.STORE_UNAVAILABLE,
        "Failed to increment retry count."
      );
    }
  }

  private async deleteRecord(userId: string, operation: string, key: string): Promise<void> {
    try {
      await this.store.delete(key, operation, userId);
    } catch (error) {
      this.metrics.increment("idempotency_store_errors_total", 1, {
        operation
      });
      this.logger.error(
        {
          key,
          userId,
          operation,
          error: error instanceof Error ? error.message : error
        },
        "Failed to delete idempotency record"
      );
    }
  }
}

export const IDEMPOTENCY_ERROR_CODES = {
  KEY_INVALID_TYPE: "KEY_INVALID_TYPE",
  KEY_REQUIRED: "KEY_REQUIRED",
  KEY_TOO_SHORT: "KEY_TOO_SHORT",
  KEY_TOO_LONG: "KEY_TOO_LONG",
  KEY_INVALID_FORMAT: "KEY_INVALID_FORMAT",

  KEY_REUSED_WITH_DIFFERENT_REQUEST: "KEY_REUSED_WITH_DIFFERENT_REQUEST",
  UNSUPPORTED_REQUEST_VALUE: "UNSUPPORTED_REQUEST_VALUE",
  REQUEST_IN_PROGRESS: "REQUEST_IN_PROGRESS",
  REQUEST_FAILED: "REQUEST_FAILED",
  RECORD_NOT_FOUND: "RECORD_NOT_FOUND",

  STORE_UNAVAILABLE: "STORE_UNAVAILABLE",
  CACHE_UNAVAILABLE: "CACHE_UNAVAILABLE",

  INVALID_CACHED_RESPONSE: "INVALID_CACHED_RESPONSE"
} as const;

export type IdempotencyErrorCode =
  (typeof IDEMPOTENCY_ERROR_CODES)[keyof typeof IDEMPOTENCY_ERROR_CODES];

import type { IdempotencyErrorCode } from "./codes";

export class IdempotencyError extends Error {
  constructor(
    public readonly code: IdempotencyErrorCode,
    message: string
  ) {
    super(message);

    this.name = "IdempotencyError";
  }
}

import type { CreateRecordResult, IdempotencyRecord } from "../types/index";

export interface FindAllFilters {
  status?: "PROCESSING" | "SUCCESS" | "FAILED";
  updatedBefore?: Date;
  limit?: number;
}

export interface RecoverStuckRecordsOptions {
  timeoutInMs?: number;
  limit?: number;
}

export interface RecoveryResult {
  processed: number;
  succeeded: number;
  failed: number;
}

export interface DeleteExpiredParams {
  /** Cutoff timestamp; purge rows where expires_at < expiredBefore */
  expiredBefore: Date;
  /**
   * Only delete non-active records (SUCCESS/FAILED).
   * Excludes PROCESSING records to prevent deleting active locks during long jobs.
   */
  statuses: Array<"SUCCESS" | "FAILED">;
  /** Optional batch limit to prevent database lock escalation during cleanup */
  limit?: number;
}

export interface IdempotencyStore {
  create(record: IdempotencyRecord): Promise<CreateRecordResult>;

  find(key: string, operation: string, userId: string): Promise<IdempotencyRecord | null>;

  /** Query records with optional filtering for background jobs or adapters */
  findAll(filters?: FindAllFilters): Promise<IdempotencyRecord[]>;

  markSuccess(key: string, response: unknown, userId: string, operation: string): Promise<void>;

  markProcessing(key: string, operation: string, userId: string): Promise<void>;

  markFailed(key: string, operation: string, userId: string): Promise<void>;

  /**
   * Increments the retry counter for a key and returns the updated count.
   *
   * CONTRACT: the counter MUST survive record deletion. The handler deletes a
   * record after a transient failure (so the client can retry) and then
   * creates a fresh one on the next attempt — if the counter lived on the row
   * and reset on re-creation, `MAX_ALLOWED_RETRIES` would never be reached and
   * the key would never be poisoned. Implementations should keep the count in
   * a location that outlives the row (e.g. a dedicated counter table/column
   * upserted by key, never reset on re-create).
   */
  incrementRetryCount(key: string, operation: string, userId: string): Promise<number>;

  /** Deletes/releases an idempotency record (e.g., after a transient error) */
  delete(key: string, operation: string, userId: string): Promise<void>;

  /**
   * Deletes expired records matching specific status conditions.
   * Returns the count of deleted rows.
   */
  deleteExpired(params: DeleteExpiredParams): Promise<number>;
}

import type { CachedResponse } from "../types/index";

type CacheResult =
  | {
      status: "HIT";
      data: CachedResponse;
    }
  | {
      status: "MISS";
    }
  | {
      status: "UNAVAILABLE";
    };

export interface IdempotencyCache {
  get(key: string): Promise<CacheResult>;

  set(key: string, value: CachedResponse, ttl: number): Promise<void>;

  delete(key: string): Promise<void>;
}

/**
 * Observability contracts for the idempotency module.
 *
 * Kept intentionally lightweight so consumers don't need a heavy logging or
 * metrics library to adopt the handler. Both interfaces are duck-typed: any
 * logger (pino, winston, console, ...) that satisfies the shape can be passed
 * via `IdempotencyHandlerOptions`.
 */

/**
 * Canonical metric names emitted by the idempotency module.
 * Every `metrics.increment` / `metrics.histogram` call must use one of these
 * names so dashboards stay stable across versions.
 */
export type MetricName =
  | "idempotency_cache_hits_total"
  | "idempotency_cache_misses_total"
  | "idempotency_duplicates_total"
  | "idempotency_store_errors_total"
  | "idempotency_recovered_records_total"
  | "idempotency_cleaned_records_total"
  | "idempotency_latency_seconds";

/**
 * Structured logger contract. The first argument is always a context object
 * (which may be empty), and the second is the human-readable message.
 */
export interface Logger {
  debug(ctx: Record<string, unknown>, msg?: string): void;
  info(ctx: Record<string, unknown>, msg?: string): void;
  warn(ctx: Record<string, unknown>, msg?: string): void;
  error(ctx: Record<string, unknown>, msg?: string): void;
}

/**
 * Metrics contract. `increment` bumps counters (optionally labelled) and
 * `histogram` records latency observations in seconds.
 */
export interface Metrics {
  increment(metric: MetricName, value?: number, labels?: Record<string, string>): void;
  histogram(metric: MetricName, valueInSeconds: number, labels?: Record<string, string>): void;
}

/**
 * Fallback stubs so the handler runs cleanly even when logger/metrics are not
 * injected. They intentionally swallow all output — observability is optional.
 */
export const noopLogger: Logger = {
  debug: () => {},
  info: () => {},
  warn: () => {},
  error: () => {}
};

export const noopMetrics: Metrics = {
  increment: () => {},
  histogram: () => {}
};

/**
 * Adapter-style logging contracts (pino-like call signatures) used by
 * `utils/create-logger.ts` to bridge arbitrary logger implementations.
 *
 * The canonical observability contracts — `Logger`, `Metrics`, `MetricName`
 * and the noop implementations — live in `./observability` and are re-exported
 * here for backward compatibility with existing imports.
 */

export interface LogFn {
  (msg: string, ...args: unknown[]): void;
  (obj: object, msg?: string, ...args: unknown[]): void;
}

export interface CoreLogger {
  debug: LogFn;
  info: LogFn;
  warn: LogFn;
  error: LogFn;
}

export type { Logger, Metrics, MetricName } from "./observability";

export type CachedResponse = {
  status: "SUCCESS" | "FAILED";
  request_hash: string;
  response: unknown;
};

export type IdempotencyStatus = "PROCESSING" | "SUCCESS" | "FAILED";

export type IdempotencyRecord = {
  key: string;
  userId: string;
  operation: string;
  requestHash: string;
  status: IdempotencyStatus;
  response?: unknown;
  expiresAt: Date;
  createdAt: Date;
  updatedAt: Date;
};

/**
 * Input shape for creating a new idempotency record.
 *
 * Deliberately structural (no ORM imports): the core stays framework- and
 * package-agnostic. Concrete adapters (e.g. a Prisma store) map these fields to
 * their own schema.
 */
export type CreateIdempotencyRecord = {
  key: string;
  request_hash: string;
  user_id: string;
  status: IdempotencyStatus;
  response: unknown;
  operation: string;
  expires_at: Date;
};

export type CreateRecordResult = { status: "CREATED" } | { status: "DUPLICATE" };

export type CreateProcessingResult =
  | { kind: "CREATED"; record: IdempotencyRecord }
  | { kind: "EXISTING_SUCCESS"; response: unknown }
  | { kind: "EXISTING_PROCESSING" } // throw
  | { kind: "EXISTING_FAILED" }; // throw

import { IDEMPOTENCY_ERROR_CODES } from "../errors/codes";
import { IdempotencyError } from "../errors/idempotency-errors";
import type { IdempotencyCache } from "../interfaces/cache";
import type { Logger, Metrics } from "../interfaces/observability";
import type { CachedResponse } from "../types/index";
import { constantTimeEquals } from "../utils/constant-time-equals";

export const DEFAULT_CACHE_TTL = 24 * 60 * 60 * 1000; // 24 hours in ms

type CacheHit = { status: "HIT"; data: CachedResponse };

/**
 * L1 cache orchestration for the idempotency module.
 *
 * Responsibilities:
 *  - Build and own the cache key layout, which scopes entries per operation
 *    AND per tenant (user) so the same key used for different operations or
 *    users never collides.
 *  - Distinguish HIT / MISS / UNAVAILABLE outcomes and emit hit/miss metrics.
 *  - Enforce the hash-mismatch security check: a cache entry whose stored
 *    `request_hash` differs from the current request's hash means the key was
 *    reused with a different payload — this is an error, never a silent hit.
 *  - Keep the cache non-blocking: any underlying cache failure (e.g. Redis
 *    down) is swallowed so callers fall back to the database, which remains
 *    the source of truth.
 */
export class CacheManager {
  constructor(
    private readonly cache: IdempotencyCache | undefined,
    private readonly logger: Logger,
    private readonly metrics: Metrics
  ) {}

  /**
   * Builds a collision-free cache key.
   *
   * SECURITY INVARIANT: a naive `idem:<operation>:<userId>:<key>` join is
   * ambiguous because every component may legitimately contain the `:`
   * delimiter (key charset allows `:`, operations default to `method:path`,
   * and `getUserId` may read client-supplied headers). Two DIFFERENT scopes
   * could therefore map to the SAME cache key, letting one tenant read another
   * tenant's cached response. Length-prefixing each component makes the
   * encoding injective: the key can be decoded unambiguously, so distinct
   * scopes always produce distinct cache keys.
   */
  private buildCacheKey(userId: string, operation: string, key: string): string {
    const encode = (segment: string): string => `${segment.length}:${segment}`;

    return `idem:v1:${encode(userId)}:${encode(operation)}:${encode(key)}`;
  }

  /**
   * Reads a cached idempotent response.
   *
   * Returns the cache payload on a HIT (after verifying the request hash
   * matches), or `null` on MISS / UNAVAILABLE / no cache configured.
   *
   * SECURITY INVARIANT: on a HIT with a mismatched request hash we re-throw an
   * `IdempotencyError` — a cached success must never be returned for a
   * different payload. Only non-idempotency (infrastructure) errors are
   * swallowed so the caller can fall back to the store.
   */
  public async getCachedResponse(
    userId: string,
    operation: string,
    key: string,
    requestHash: string
  ): Promise<CacheHit | null> {
    if (!this.cache) {
      return null;
    }

    const cacheKey = this.buildCacheKey(userId, operation, key);

    try {
      const result = await this.cache.get(cacheKey);

      if (result.status === "MISS" || result.status === "UNAVAILABLE") {
        this.metrics.increment("idempotency_cache_misses_total", 1, {
          operation
        });
        this.logger.debug(
          { key, userId, operation, status: result.status },
          "Idempotency cache miss"
        );
        return null;
      }

      // SECURITY INVARIANT: never serve a cache entry whose shape is not a
      // well-formed SUCCESS payload. A corrupted, stale-format, or foreign
      // entry (wrong status, missing request_hash) must not be replayed as a
      // successful response — fall back to the store (source of truth) and
      // best-effort invalidate the bad entry.
      const data = result.data;
      if (!data || data.status !== "SUCCESS" || typeof data.request_hash !== "string") {
        this.metrics.increment("idempotency_cache_misses_total", 1, {
          operation
        });
        this.logger.warn(
          { key, userId, operation, status: data?.status },
          "Invalid cached idempotency response; falling back to store"
        );

        // Non-blocking invalidation: never let a bad cache entry break the
        // request — the DB write that produced it already succeeded.
        try {
          await this.cache.delete(cacheKey);
        } catch {
          // swallowed; the entry will expire via TTL
        }

        return null;
      }

      // Security check: Same idempotency key, but payload hash changed mid-flight
      if (!constantTimeEquals(data.request_hash, requestHash)) {
        this.logger.warn(
          { key, userId, operation },
          "Idempotency key reused with different request payload in cache"
        );
        throw new IdempotencyError(
          IDEMPOTENCY_ERROR_CODES.KEY_REUSED_WITH_DIFFERENT_REQUEST,
          "Idempotency key was already used with a different request."
        );
      }

      this.metrics.increment("idempotency_cache_hits_total", 1, { operation });
      this.logger.info({ key, userId, operation, status: "cache_hit" }, "Idempotency cache hit");
      return result;
    } catch (error) {
      // Re-throw validation/mismatch errors; swallow underlying redis/cache
      // failures so the request falls back to the database.
      if (error instanceof IdempotencyError) {
        throw error;
      }

      this.logger.error(
        {
          key,
          userId,
          operation,
          error: error instanceof Error ? error.message : error
        },
        "Error fetching idempotency record from cache"
      );
      return null;
    }
  }

  /**
   * Writes a successful response into the cache.
   *
   * NON-BLOCKING INVARIANT: the database write has already succeeded by the
   * time this runs, so a cache failure must never surface to the end user.
   * Errors are logged as warnings and swallowed.
   */
  public async setCachedResponse(
    userId: string,
    operation: string,
    key: string,
    requestHash: string,
    response: unknown
  ): Promise<void> {
    if (!this.cache) {
      return;
    }

    const cacheKey = this.buildCacheKey(userId, operation, key);

    try {
      await this.cache.set(
        cacheKey,
        {
          request_hash: requestHash,
          response,
          status: "SUCCESS"
        },
        DEFAULT_CACHE_TTL
      );
    } catch (error) {
      // Non-blocking: standard DB write succeeded, so cache failures shouldn't throw to end-user
      this.logger.warn(
        {
          key,
          userId,
          operation,
          error: error instanceof Error ? error.message : error
        },
        "Failed to update idempotency cache (database remains source of truth)"
      );
    }
  }
}

import type { Logger, Metrics } from "../interfaces/observability";
import type { IdempotencyStore } from "../interfaces/store";

export type CleanupExpiredOptions = {
  limit?: number;
};

export type CleanupResult = {
  deleted: number;
};

/**
 * Deletes expired idempotency records. Intended to be run periodically by a
 * background worker or cron job.
 *
 * SAFETY INVARIANT: only records with status 'SUCCESS' or 'FAILED' are deleted.
 * Records marked 'PROCESSING' are intentionally excluded here, even if expired:
 * if a long-running job passed its expiry timestamp, purging its lock would let
 * a duplicate incoming request acquire a fresh lock and run concurrently.
 * Active or stuck 'PROCESSING' locks are instead handled by
 * `recoverStuckRecords()` in `jobs/record-recovery`.
 */
export async function cleanupExpiredRecords(
  store: IdempotencyStore,
  logger: Logger,
  metrics: Metrics,
  { limit = 500 }: CleanupExpiredOptions = {}
): Promise<CleanupResult> {
  const now = new Date();

  try {
    const deleted = await store.deleteExpired({
      expiredBefore: now,
      statuses: ["SUCCESS", "FAILED"],
      limit
    });

    metrics.increment("idempotency_cleaned_records_total", deleted);
    logger.info(
      { deleted, expiredBefore: now, limit },
      "Expired idempotency records cleanup completed"
    );

    return { deleted };
  } catch (error) {
    metrics.increment("idempotency_store_errors_total", 1, {
      operation: "cleanup"
    });
    logger.error(
      { error: error instanceof Error ? error.message : error },
      "Failed to clean up expired idempotency records"
    );
    throw error;
  }
}

import { IDEMPOTENCY_ERROR_CODES } from "../errors/codes";
import { IdempotencyError } from "../errors/idempotency-errors";
import type { Logger, Metrics } from "../interfaces/observability";
import type {
  IdempotencyStore,
  RecoverStuckRecordsOptions,
  RecoveryResult
} from "../interfaces/store";

/**
 * Recovers operations that got stuck in 'PROCESSING' state (e.g., a node
 * crashed mid-execution and never released its lock). Such records are marked
 * 'FAILED' so clients can safely retry or inspect the outcome.
 *
 * RESILIENCE INVARIANT: a single failing `markFailed` call (e.g. a DB hiccup)
 * must not abort recovery of the remaining records — each record is processed
 * independently and the outcome is counted in `failed`.
 *
 * LEASE-RACE WARNING: recovery assumes a PROCESSING record whose `updatedAt` is
 * older than `timeoutInMs` is dead. If a legitimate operation runs LONGER than
 * the timeout (no heartbeat), recovery can mark it FAILED while it is still
 * executing — allowing a retry to re-run it (double execution) and/or losing
 * its result. Mitigations for long-running jobs: (1) keep a heartbeat that
 * touches `updatedAt`; (2) make the store's `markFailed` conditional
 * (`UPDATE ... WHERE status = 'PROCESSING'`) so a just-completed SUCCESS
 * record can never be flipped to FAILED.
 */
export async function recoverStuckRecords(
  store: IdempotencyStore,
  logger: Logger,
  metrics: Metrics,
  {
    timeoutInMs = 5 * 60 * 1000, // 5 minutes threshold by default
    limit = 100
  }: RecoverStuckRecordsOptions = {}
): Promise<RecoveryResult> {
  const cutoffTime = new Date(Date.now() - timeoutInMs);

  const stuckRecords = await store.findAll({
    status: "PROCESSING",
    updatedBefore: cutoffTime,
    limit
  });

  const result: RecoveryResult = {
    processed: stuckRecords.length,
    succeeded: 0,
    failed: 0
  };

  for (const record of stuckRecords) {
    try {
      await markRecordFailed(store, metrics, record.userId, record.operation, record.key);
      result.succeeded++;
    } catch {
      result.failed++;
    }
  }

  metrics.increment("idempotency_recovered_records_total", result.succeeded, {
    status: "success"
  });
  if (result.failed > 0) {
    metrics.increment("idempotency_recovered_records_total", result.failed, {
      status: "failed"
    });
  }

  logger.info(
    {
      processed: result.processed,
      succeeded: result.succeeded,
      failed: result.failed
    },
    "Stuck idempotency records recovery completed"
  );

  return result;
}

/**
 * Marks a single record FAILED, translating infrastructure failures into a
 * typed `IdempotencyError` while preserving the original error for
 * `IdempotencyError` cases.
 */
async function markRecordFailed(
  store: IdempotencyStore,
  metrics: Metrics,
  userId: string,
  operation: string,
  key: string
): Promise<void> {
  try {
    await store.markFailed(key, operation, userId);
  } catch (error) {
    metrics.increment("idempotency_store_errors_total", 1, { operation });
    if (error instanceof IdempotencyError) {
      throw error;
    }

    throw new IdempotencyError(
      IDEMPOTENCY_ERROR_CODES.STORE_UNAVAILABLE,
      "Failed to update the record to failed."
    );
  }
}

import { IDEMPOTENCY_ERROR_CODES } from "../errors/codes";
import { IdempotencyError } from "../errors/idempotency-errors";

export type KeyValidationOptions = {
  minLength?: number;
  maxLength?: number;
  requireUuid?: boolean;
  pattern?: {
    value: RegExp;
    message: string;
  };
};

export const DEFAULT_MIN_KEY_LENGTH = 1;
export const DEFAULT_MAX_KEY_LENGTH = 128;

// Strict RFC 4122 pattern for UUID v1-v5 (case-insensitive). v1-v5 is
// distinguished by the version nibble `[1-5]` and the variant nibble `[89abAB]`.
export const UUID_REGEX =
  /^[0-9a-fA-F]{8}-[0-9a-fA-F]{4}-[1-5][0-9a-fA-F]{3}-[89abAB][0-9a-fA-F]{3}-[0-9a-fA-F]{12}$/;

// Default allowed charset: letters, digits, dots, underscores, hyphens, colons.
const DEFAULT_PATTERN = {
  value: /^[A-Za-z0-9._:-]+$/,
  message:
    "Only letters, numbers, dots (.), underscores (_), hyphens (-), and colons (:) are allowed."
};

/**
 * Sanitizes and validates an idempotency key, returning the cleaned value.
 *
 * INVARIANT (ordering matters): the key is trimmed FIRST, before any length or
 * format checks, so whitespace-only input can't bypass length limits, and so
 * callers never store raw input that contains trailing/leading whitespace.
 *
 * Callers MUST use the returned (trimmed) key, never the raw input.
 */
export function validateKey(
  key: unknown,
  {
    minLength = DEFAULT_MIN_KEY_LENGTH,
    maxLength = DEFAULT_MAX_KEY_LENGTH,
    requireUuid = false,
    pattern = DEFAULT_PATTERN
  }: KeyValidationOptions = {}
): string {
  if (typeof key !== "string") {
    throw new IdempotencyError(
      IDEMPOTENCY_ERROR_CODES.KEY_INVALID_TYPE,
      "Invalid idempotency key. The value must be a string."
    );
  }

  // Always sanitize first; raw input might contain trailing spaces or control whitespace
  const trimmedKey = key.trim();

  if (trimmedKey.length === 0) {
    throw new IdempotencyError(
      IDEMPOTENCY_ERROR_CODES.KEY_REQUIRED,
      "An idempotency key is required."
    );
  }

  if (trimmedKey.length < minLength) {
    throw new IdempotencyError(
      IDEMPOTENCY_ERROR_CODES.KEY_TOO_SHORT,
      `The idempotency key must be at least ${minLength} characters long.`
    );
  }

  if (trimmedKey.length > maxLength) {
    throw new IdempotencyError(
      IDEMPOTENCY_ERROR_CODES.KEY_TOO_LONG,
      `The idempotency key must not exceed ${maxLength} characters.`
    );
  }

  // UUID mode overrides the standard charset regex when explicitly enabled.
  if (requireUuid) {
    if (!UUID_REGEX.test(trimmedKey)) {
      throw new IdempotencyError(
        IDEMPOTENCY_ERROR_CODES.KEY_INVALID_FORMAT,
        "Invalid idempotency key. Key must be a valid UUID v1-v5."
      );
    }
  } else if (!pattern.value.test(trimmedKey)) {
    throw new IdempotencyError(
      IDEMPOTENCY_ERROR_CODES.KEY_INVALID_FORMAT,
      `Invalid idempotency key. ${pattern.message}`
    );
  }

  // Return the cleaned key; callers MUST use this returned value, not the raw input
  return trimmedKey;
}

import crypto from "crypto";
import { IDEMPOTENCY_ERROR_CODES } from "../errors/codes";
import { IdempotencyError } from "../errors/idempotency-errors";

/**
 * Maximum object/array nesting depth accepted by `serialize`.
 *
 * SECURITY: without a depth bound, an attacker-controlled body that nests
 * objects thousands of levels deep (or contains a cyclic reference) would
 * recurse until the JS stack overflows, surfacing as a raw `RangeError` (500)
 * instead of a clean, typed client error. Bounding the depth turns that DoS
 * vector into a deterministic `UNSUPPORTED_REQUEST_VALUE` rejection.
 */
const MAX_SERIALIZE_DEPTH = 100;

/**
 * Deterministically stringifies a payload structure.
 *
 * Why not plain `JSON.stringify`? `JSON.stringify({ a: 1, b: 2 })` differs from
 * `JSON.stringify({ b: 2, a: 1 })`. By sorting object keys recursively, two
 * logically-identical payloads always produce byte-identical strings, which is
 * the foundation for stable request hashing.
 *
 * REPRESENTATION SAFETY: primitives are encoded with `JSON.stringify` so that
 * `null` ("null") is distinguishable from the string "null" ('"null"'), and
 * numbers from numeric strings. Arrays keep their order (order is significant).
 *
 * REJECTED VALUES: `undefined`, `NaN`, `±Infinity`, `BigInt`, functions,
 * symbols, cycles, and nesting deeper than `MAX_SERIALIZE_DEPTH` are rejected
 * because their stringification is non-deterministic, runtime-dependent, or a
 * stack-overflow risk — hashing them would produce unstable fingerprints.
 */
export function serialize(value: unknown): string {
  return serializeInternal(value, 0, new WeakSet<object>());
}

function serializeInternal(value: unknown, depth: number, seen: WeakSet<object>): string {
  if (value === null) {
    return "null";
  }

  if (typeof value === "number") {
    // NaN or Infinity produce inconsistent output across JS runtimes
    if (!Number.isFinite(value)) {
      throw unsupported();
    }

    return JSON.stringify(value);
  }

  if (typeof value === "string" || typeof value === "boolean") {
    return JSON.stringify(value);
  }

  if (Array.isArray(value)) {
    if (depth > MAX_SERIALIZE_DEPTH) {
      throw unsupported();
    }

    // Cycle detection: a container that is already on the current recursion
    // path is a cyclic reference, which would otherwise recurse forever.
    // Containers are removed from `seen` when their subtree finishes so shared
    // (diamond) references — which are legal — are not mistaken for cycles.
    if (seen.has(value)) {
      throw unsupported();
    }

    seen.add(value);

    try {
      const items = value.map((item) => serializeInternal(item, depth + 1, seen));

      return `[${items.join(",")}]`;
    } finally {
      seen.delete(value);
    }
  }

  if (typeof value === "object") {
    if (depth > MAX_SERIALIZE_DEPTH) {
      throw unsupported();
    }

    if (seen.has(value)) {
      throw unsupported();
    }

    seen.add(value);

    try {
      const object = value as Record<string, unknown>;

      // Sorting object keys is what guarantees payload hashing consistency
      const entries = Object.keys(object)
        .sort()
        .map((key) => {
          const serializedKey = JSON.stringify(key);
          const serializedValue = serializeInternal(object[key], depth + 1, seen);

          return `${serializedKey}:${serializedValue}`;
        });

      return `{${entries.join(",")}}`;
    } finally {
      seen.delete(value);
    }
  }

  throw unsupported();
}

function unsupported(): IdempotencyError {
  return new IdempotencyError(
    IDEMPOTENCY_ERROR_CODES.UNSUPPORTED_REQUEST_VALUE,
    "Unsupported request value for deterministic serialization."
  );
}

/**
 * Produces a stable SHA-256 fingerprint of a request payload.
 *
 * The digest is a 64-character lowercase hex string and is the value stored as
 * `request_hash` on records and cache entries. Two identical payloads always
 * yield the same hash regardless of object key ordering; two different payloads
 * yield different hashes (barring SHA-256 collisions).
 */
export function hashRequest(body: unknown): string {
  const serialized = serialize(body);

  return crypto.createHash("sha256").update(serialized).digest("hex");
}

import crypto from "crypto";

/**
 * Constant-time string comparison for secret-ish values such as request hashes.
 *
 * WHY: comparing two strings with `!==` short-circuits on the first differing
 * byte, so an attacker probing stored values could measure timing differences
 * and learn prefix information. `crypto.timingSafeEqual` runs in time
 * proportional to the input length regardless of where the difference is.
 *
 * NOTE: this is defense-in-depth — a 256-bit SHA-256 digest cannot realistically
 * be recovered via timing on its own — but the check is cheap and removes the
 * leak class entirely. Lengths are compared first because `timingSafeEqual`
 * throws when the buffers differ in length (and request hashes are always
 * 64-char lowercase hex, so a length mismatch means "different" anyway).
 */
export function constantTimeEquals(a: string, b: string): boolean {
  // Defensive: a malformed/legacy record may carry a non-string hash (e.g.
  // `undefined`). Treat it as a mismatch — never crash the request with a
  // TypeError, and never let a missing hash pass the comparison.
  if (typeof a !== "string" || typeof b !== "string") {
    return false;
  }

  const bufferA = Buffer.from(a);
  const bufferB = Buffer.from(b);

  if (bufferA.length !== bufferB.length) {
    return false;
  }

  return crypto.timingSafeEqual(bufferA, bufferB);
}

// utils/create-logger.ts
import type { CoreLogger, LogFn } from "../interfaces/logger";

export type MinimalLogHandler = (
  level: "debug" | "info" | "warn" | "error",
  msg: string,
  meta?: object
) => void;

/**
 * Utility to convert any custom log function into a CoreLogger in seconds.
 */
export function createLoggerAdapter(handler: MinimalLogHandler): CoreLogger {
  const wrap = (level: "debug" | "info" | "warn" | "error"): LogFn => {
    return (first: unknown, second?: unknown) => {
      if (typeof first === "string") {
        handler(level, first);
      } else if (typeof first === "object" && first !== null) {
        handler(level, typeof second === "string" ? second : "", first);
      }
    };
  };

  return {
    debug: wrap("debug"),
    info: wrap("info"),
    warn: wrap("warn"),
    error: wrap("error")
  };
}

/**
 * Adapter-agnostic HTTP helpers shared by every framework adapter.
 *
 * The framework adapters (Express / Fastify / Hono) translate `IdempotencyError`
 * into HTTP responses. The status mapping and the error envelope are identical
 * across frameworks, so they live here once instead of being copied into each
 * adapter.
 */
import type { IdempotencyErrorCode } from "../errors/codes";
import { IdempotencyError } from "../errors/idempotency-errors";

// Default error code -> HTTP status mapping.
export const DEFAULT_STATUS_MAP: Record<IdempotencyErrorCode, number> = {
  // Client input problems -> 400
  KEY_INVALID_TYPE: 400,
  KEY_REQUIRED: 400,
  KEY_TOO_SHORT: 400,
  KEY_TOO_LONG: 400,
  KEY_INVALID_FORMAT: 400,
  UNSUPPORTED_REQUEST_VALUE: 400,
  // Idempotency conflicts -> 409 Conflict
  KEY_REUSED_WITH_DIFFERENT_REQUEST: 409,
  REQUEST_IN_PROGRESS: 409,
  REQUEST_FAILED: 409,
  // Lookups
  RECORD_NOT_FOUND: 404,
  // Infrastructure
  STORE_UNAVAILABLE: 503,
  CACHE_UNAVAILABLE: 503,
  // Should not happen; treat as a server error
  INVALID_CACHED_RESPONSE: 500
};

/**
 * Maps an `IdempotencyError` to the HTTP status it should be served with.
 */
export function idempotencyErrorStatus(
  error: IdempotencyError,
  statusMap?: Partial<Record<IdempotencyErrorCode, number>>
): number {
  return statusMap?.[error.code] ?? DEFAULT_STATUS_MAP[error.code] ?? 500;
}

/**
 * Builds the standard error envelope: `{ error: { code, message } }`.
 */
export function buildErrorEnvelope(error: IdempotencyError): {
  error: { code: string; message: string };
} {
  return { error: { code: error.code, message: error.message } };
}

/**
 * Thin Express adapter for `IdempotencyHandler`.
 *
 * The core (`core/idempotency-handler.ts` and everything it imports) is deliberately
 * framework- and package-agnostic — it only depends on Node built-ins. This
 * file is the OPTIONAL bridge that translates between Express requests and the
 * handler's plain-typed API. It imports `express` only as *types*, so it has
 * no runtime dependency on Express either.
 *
 * Two usage styles are provided:
 *
 * 1. `idempotent(handler, route, options)` — wrap a route whose business logic
 *    RETURNS the response payload. Cleanest and fully typed:
 *
 *        app.post("/payments", idempotent(handler, async (req) => {
 *          const charge = await createCharge(req.body);
 *          return { id: charge.id, status: charge.status };
 *        }));
 *
 * 2. `idempotencyMiddleware(handler, options)` — drop-in middleware for routes
 *    that already call `res.json(...)` themselves. The middleware intercepts
 *    the JSON body the downstream handler sends and stores it as the
 *    idempotent response:
 *
 *        app.post("/payments", idempotencyMiddleware(handler), async (req, res) => {
 *          const charge = await createCharge(req.body);
 *          res.json(charge);
 *        });
 *
 * Both styles forward business-logic errors to Express's error middleware and
 * map `IdempotencyError`s to HTTP status codes (see `idempotencyErrorStatus`).
 */
import type { NextFunction, Request, RequestHandler, Response } from "express";
import { IDEMPOTENCY_ERROR_CODES } from "../errors/codes";
import type { IdempotencyErrorCode } from "../errors/codes";
import { IdempotencyError } from "../errors/idempotency-errors";
import type { IdempotencyHandler } from "../core/idempotency-handler";
import { hashRequest } from "../utils/serializer";
import { buildErrorEnvelope, idempotencyErrorStatus } from "./shared";

// Re-exported for API parity — the mapping is shared by every framework adapter.
export { idempotencyErrorStatus } from "./shared";

/** Where to look for the idempotency key in a request. */
export type IdempotencyKeySource = {
  /** Header name (e.g. "Idempotency-Key"). Default: "Idempotency-Key". */
  header?: string;
  /** Optional JSON body field (e.g. "idempotencyKey"). Disabled by default. */
  bodyField?: string;
  /** Optional query parameter (e.g. "idempotency_key"). Disabled by default. */
  queryParam?: string;
};

export type IdempotencyContext = {
  key: string;
  userId: string;
  operation: string;
  requestHash: string;
  outcome: "processing" | "completed" | "failed" | "skipped";
};

export type ExpressIdempotencyOptions = {
  /** How to extract the key. Default: `Idempotency-Key` header only. */
  key?: IdempotencyKeySource;
  /**
   * When true, a request without a key is rejected with 400 instead of
   * silently proceeding without idempotency protection. Default: false.
   */
  requireKey?: boolean;
  /**
   * Resolve the tenant/user id. Default: `req.user?.id` or "anonymous".
   *
   * SECURITY: this value scopes the idempotency key. If it resolves to a
   * constant (e.g. the default "anonymous" when `req.user` is unset), keys
   * become GLOBAL across all users — any client that learns a key + payload
   * could replay another client's stored response. Always provide a real,
   * authenticated per-user id (or `getOperation` that embeds the tenant).
   */
  getUserId?: (req: Request) => string | undefined;
  /** Resolve the operation name. Default: `${req.method}:${req.path}`. */
  getOperation?: (req: Request) => string;
  /** Resolve the payload to hash. Default: `req.body ?? {}`. */
  getBody?: (req: Request) => unknown;
  /**
   * Classify business-logic failures as permanent (mark FAILED) vs transient
   * (increment retry count / release the lock). Forwarded to
   * `IdempotencyHandler.execute`.
   */
  isPermanentError?: (err: unknown) => boolean;
  /** Attach metadata to `res.locals.idempotency`. Default: true. */
  attachContext?: boolean;
  /** Override the default `IdempotencyError` -> HTTP status mapping. */
  statusMap?: Partial<Record<IdempotencyErrorCode, number>>;
};

/**
 * Extracts the idempotency key from a request.
 *
 * Priority: header -> JSON body field -> query parameter. The first non-empty
 * string wins; `undefined` means "no key provided".
 */
export function getIdempotencyKey(
  req: Request,
  options: ExpressIdempotencyOptions = {}
): string | undefined {
  const { header = "Idempotency-Key", bodyField, queryParam } = options.key ?? {};

  const fromHeader = header ? req.get(header) : undefined;
  if (typeof fromHeader === "string" && fromHeader.length > 0) {
    return fromHeader;
  }

  if (bodyField) {
    const body = req.body as Record<string, unknown> | undefined;
    const fromBody = body?.[bodyField];
    if (typeof fromBody === "string" && fromBody.length > 0) {
      return fromBody;
    }
  }

  if (queryParam) {
    const fromQuery = req.query[queryParam];
    if (typeof fromQuery === "string" && fromQuery.length > 0) {
      return fromQuery;
    }
  }

  return undefined;
}

/**
 * Responds with the standard error envelope: `{ error: { code, message } }`.
 */
export function sendIdempotencyError(
  res: Response,
  error: IdempotencyError,
  statusMap?: ExpressIdempotencyOptions["statusMap"]
): void {
  res.status(idempotencyErrorStatus(error, statusMap)).json(buildErrorEnvelope(error));
}

function defaultUserId(req: Request): string {
  const user = (req as { user?: { id?: unknown } }).user;

  return typeof user?.id === "string" ? user.id : "anonymous";
}

function defaultOperation(req: Request): string {
  return `${req.method}:${req.path}`;
}

/**
 * Wraps a route handler that RETURNS its response payload.
 *
 * The returned Express handler:
 *  - extracts the key (missing key + `requireKey` -> 400, otherwise pass-through),
 *  - resolves userId/operation/body through the option hooks,
 *  - runs the business logic inside `IdempotencyHandler.execute` (which
 *    guarantees exactly-once semantics and stores the returned value),
 *  - sends the result with `res.json(result)` unless the route already
 *    responded.
 */
export function idempotent(
  handler: IdempotencyHandler,
  route: (req: Request, res: Response) => unknown | Promise<unknown>,
  options: ExpressIdempotencyOptions = {}
): RequestHandler {
  return async (req: Request, res: Response, next: NextFunction) => {
    const key = getIdempotencyKey(req, options);

    if (key === undefined) {
      if (options.requireKey) {
        if (res.headersSent) {
          return;
        }
        sendIdempotencyError(
          res,
          new IdempotencyError(
            IDEMPOTENCY_ERROR_CODES.KEY_REQUIRED,
            "Missing idempotency key. Provide the configured header, body field, or query parameter."
          ),
          options.statusMap
        );
        return;
      }

      // No key and none required: run the business logic WITHOUT idempotency
      // protection. In this wrapper style the route IS the downstream handler,
      // so we invoke it directly (calling next() would skip it entirely).
      attachContext(req, res, options, { outcome: "skipped" });
      const unguarded = await route(req, res);
      if (!res.headersSent) {
        res.json(unguarded);
      }
      return;
    }

    const userId = options.getUserId?.(req) ?? defaultUserId(req);
    const operation = options.getOperation?.(req) ?? defaultOperation(req);
    const body = options.getBody?.(req) ?? req.body ?? {};

    // Attach the context BEFORE executing so downstream middleware (including
    // the route itself and error middleware) can read it mid-flight. The
    // outcome is updated once the execution settles.
    attachContext(req, res, options, {
      key,
      userId,
      operation,
      outcome: "processing"
    });

    try {
      const result = await handler.execute(
        userId,
        operation,
        key,
        body,
        // `async` wraps the route's (possibly plain) return value in a promise,
        // satisfying the handler's `() => Promise<unknown>` callback contract.
        async () => route(req, res),
        options.isPermanentError ? { isPermanentError: options.isPermanentError } : undefined
      );

      attachContext(req, res, options, {
        key,
        userId,
        operation,
        outcome: "completed"
      });

      if (!res.headersSent) {
        res.json(result);
      }
    } catch (error) {
      attachContext(req, res, options, {
        key,
        userId,
        operation,
        outcome: "failed"
      });

      // SECURITY/ROBUSTNESS: if the route already sent a response before the
      // idempotency layer failed, never attempt a second send — Express would
      // throw "Cannot set headers after they are sent" and crash the request.
      if (res.headersSent) {
        return;
      }

      if (error instanceof IdempotencyError) {
        sendIdempotencyError(res, error, options.statusMap);
        return;
      }

      next(error);
    }
  };
}

/**
 * Drop-in middleware for routes that send their own response.
 *
 * The downstream handler's `res.json`/`res.send` payload is captured and stored
 * as the idempotent response. If the downstream chain errors (a 4xx/5xx
 * response, e.g. from your error middleware), the idempotency record is marked
 * FAILED instead of SUCCESS, and the error is NOT re-sent — your error
 * middleware already handled it.
 */
export function idempotencyMiddleware(
  handler: IdempotencyHandler,
  options: ExpressIdempotencyOptions = {}
): RequestHandler {
  return async (req: Request, res: Response, next: NextFunction) => {
    const key = getIdempotencyKey(req, options);

    if (key === undefined) {
      if (options.requireKey) {
        sendIdempotencyError(
          res,
          new IdempotencyError(
            IDEMPOTENCY_ERROR_CODES.KEY_REQUIRED,
            "Missing idempotency key. Provide the configured header, body field, or query parameter."
          ),
          options.statusMap
        );
        return;
      }

      attachContext(req, res, options, { outcome: "skipped" });
      next();
      return;
    }

    const userId = options.getUserId?.(req) ?? defaultUserId(req);
    const operation = options.getOperation?.(req) ?? defaultOperation(req);
    const body = options.getBody?.(req) ?? req.body ?? {};

    // Attach the context BEFORE executing so downstream middleware can read it
    // mid-flight. The outcome is updated once the execution settles.
    attachContext(req, res, options, {
      key,
      userId,
      operation,
      outcome: "processing"
    });

    try {
      const result = await handler.execute(
        userId,
        operation,
        key,
        body,
        () => runDownstream(req, res, next),
        options.isPermanentError ? { isPermanentError: options.isPermanentError } : undefined
      );

      attachContext(req, res, options, {
        key,
        userId,
        operation,
        outcome: "completed"
      });

      // The downstream handler already sent its response; only send if it
      // produced no payload.
      if (!res.headersSent) {
        res.json(result);
      }
    } catch (error) {
      attachContext(req, res, options, {
        key,
        userId,
        operation,
        outcome: "failed"
      });

      // The downstream error middleware already responded — don't double-send.
      if (res.headersSent) {
        return;
      }

      if (error instanceof IdempotencyError) {
        sendIdempotencyError(res, error, options.statusMap);
        return;
      }

      next(error);
    }
  };
}

function attachContext(
  req: Request,
  res: Response,
  options: ExpressIdempotencyOptions,
  partial: { outcome: IdempotencyContext["outcome"] } & Partial<
    Pick<IdempotencyContext, "key" | "userId" | "operation">
  >
): void {
  if (options.attachContext === false) {
    return;
  }

  const key = partial.key ?? getIdempotencyKey(req, options) ?? "";
  const userId = partial.userId ?? options.getUserId?.(req) ?? defaultUserId(req);
  const operation = partial.operation ?? options.getOperation?.(req) ?? defaultOperation(req);

  res.locals.idempotency = {
    key,
    userId,
    operation,
    requestHash: hashOf(req, options),
    outcome: partial.outcome
  };
}

function hashOf(req: Request, options: ExpressIdempotencyOptions): string {
  // The hash only feeds observability metadata on `res.locals`; never let it
  // break the request if the body can't be serialized.
  try {
    return hashRequest(options.getBody?.(req) ?? req.body ?? {});
  } catch {
    return "";
  }
}

/**
 * Runs the downstream Express stack, resolving with the JSON payload the route
 * sends via `res.json`/`res.send`.
 */
function runDownstream(req: Request, res: Response, next: NextFunction): Promise<unknown> {
  return new Promise((resolve, reject) => {
    let captured: unknown;
    let settled = false;

    // Keep the RAW references for restoration so any spies/mocks installed on
    // the response stay inspectable; use the bound versions for invoking (so
    // `this` remains the response object, as Express expects).
    const rawJson = res.json;
    const rawSend = res.send;
    const originalJson = res.json.bind(res);
    const originalSend = res.send.bind(res);

    const settle = (value: unknown): void => {
      if (!settled) {
        settled = true;
        res.json = rawJson;
        res.send = rawSend;
        resolve(value);
      }
    };
    const fail = (err: unknown): void => {
      if (!settled) {
        settled = true;
        res.json = rawJson;
        res.send = rawSend;
        reject(err);
      }
    };

    // Any 4xx/5xx response — including one sent by the app's error middleware
    // via res.json/res.send — means the downstream execution FAILED. Check the
    // status at capture time so an error body is never recorded as a success.
    const respond = (value: unknown): void => {
      if (res.statusCode >= 400) {
        fail(new Error(`Downstream handler responded with status ${res.statusCode}`));
      } else {
        settle(value);
      }
    };

    res.json = (value: unknown) => {
      captured = value;
      respond(value);
      return originalJson(value);
    };
    res.send = (value: unknown) => {
      captured = value;
      respond(value);
      return originalSend(value);
    };

    res.once("finish", () => {
      // If the downstream chain errored, the app's error middleware typically
      // responds with 4xx/5xx. Treat that as a failed execution so the
      // idempotency record is marked FAILED rather than SUCCESS.
      if (res.statusCode >= 400) {
        fail(new Error(`Downstream handler responded with status ${res.statusCode}`));
      } else {
        settle(captured);
      }
    });

    next();
  });
}

declare global {
  // eslint-disable-next-line @typescript-eslint/no-namespace
  namespace Express {
    interface Locals {
      idempotency?: IdempotencyContext;
    }
  }
}