import {
  Inject,
  CallHandler,
  ExecutionContext,
  Injectable,
  NestInterceptor,
  SetMetadata,
} from "@nestjs/common";
import { createHash } from "node:crypto";
import { Observable, from, of } from "rxjs";
import { switchMap, tap } from "rxjs/operators";
import { RedisService } from "../redis.service.js";
import { logger } from "../logger.js";

/**
 * Idempotent-write middleware (TASK-0095 R1) — the server half of the offline
 * outbox. The client tags every REST mutation with an `Idempotency-Key: <key>`
 * header (a client-generated `key_` + 32 hex); a mutation whose idempotent
 * replay would be wrong (a `POST /records` create — the second send must not
 * double-create) is exactly what the key prevents.
 *
 * RULE 29 — recorded deviation: the task spec names this a *middleware* at
 * `common/middleware/idempotency.ts`. A true Express/Nest **middleware runs
 * BEFORE the global guard** (auth), so it has no `req.auth` (the `userId` the
 * `idem:{userId}:{key}` keyspace needs) and could not capture the response
 * body. The correct Nest primitive is a **global interceptor** (it runs after
 * the guard, so `req.auth` is present, and it wraps the handler result, so it
 * can store the response). The spec's FILE PATH is respected
 * (`common/middleware/idempotency.ts`); the primitive is an interceptor.
 *
 * The standard idempotent-write pattern (key checked BEFORE execution):
 *   - a SEEN key (a Redis `idem:{userId}:{key}` set by a prior execution,
 *     TTL 24 h) → the STORED response is replayed (the handler is NOT
 *     re-run — a double-replay creates no duplicate);
 *   - a FRESH key → the handler runs; a SUCCESSFUL (2xx) result is stored
 *     under the key (a 4xx/5xx was rejected, not executed — nothing to
 *     replay; the outbox moves the item to its failed list).
 *
 * Scope (the 6-case matrix, the spec's named unit surface):
 *   1. key HIT  → the stored body is replayed, the handler is skipped;
 *   2. key MISS → the handler runs, a 2xx body is stored;
 *   3. expiry   → a key past its 24 h TTL reads as MISS (Redis enforces the
 *                 TTL; the client re-executes — a day-old mutation is a new
 *                 attempt, matching the outbox's retry=fresh-key rule);
 *   4. storage  → only a 2xx body is stored (a 4xx/5xx is not);
 *   5. GET excl. → a GET never carries a key (the client only tags mutations);
 *                 a GET with a key is a no-op here (defensive — the keyspace
 *                 is for writes);
 *   6. no-header → a request without `Idempotency-Key` passes through
 *                 untouched (every Phase-3/5 endpoint keeps working — the
 *                 middleware is additive).
 *
 * Fail-open: a Redis error degrades to "execute, don't store" (the mutation
 * still succeeds; the worst case is a duplicate, never a lost write — the
 * same degradation policy as the rate limiter). Logged, not thrown.
 *
 * Concurrency note (recorded): the 6-case matrix does not include two LIVE
 * clients racing the SAME key (a true single-flight would need a SETNX lock).
 * The key is per-tab unique (client-generated), so a same-key race only
 * happens if two clients deliberately share a key — a client bug, not a
 * protocol case. A crash-mid-drain REPLAY (the outbox's actual concern) is
 * sequential and fully covered.
 */

/** The 24 h keyspace TTL (the task's stored-response bound). */
export const IDEM_TTL_SECONDS = 24 * 60 * 60;

/**
 * The single-flight lock TTL (docs §9: "concurrent duplicate → the
 * in-flight result is awaited, Redis lock, 10 s"). A MISS caller takes
 * `idem:lock:{redisKey}` (SET NX) before executing; a concurrent duplicate
 * of the SAME operation (same credential + route + body digest) that misses
 * the memo but cannot take the lock WAITS for the memo the first caller
 * stores, instead of executing a second time. The lock is never released
 * explicitly — the memo (checked BEFORE the lock) supersedes it, and the 10 s
 * TTL is the backstop.
 */
export const IDEM_LOCK_TTL_SECONDS = 10;

/** Poll cadence for a waiter on an in-flight duplicate (≤ ~1.5 s). */
const IDEM_WAIT_POLL_MS = 75;
const IDEM_WAIT_POLLS = 20;

/** The header the client tags mutations with (the frozen shared contract). */
export const IDEMPOTENCY_KEY_HEADER = "idempotency-key";

/**
 * RULE 29 — recorded deviation (TASK-0126): a route that OWNS its idempotency
 * and streams (SSE) must opt OUT of this body-replay interceptor. The
 * interceptor's replay contract is "re-run the handler with the STORED BODY" —
 * for a stream the handler writes the socket itself, so a key HIT that returns
 * the stored body WITHOUT running the handler leaves the `res` never written
 * nor ended (the client hangs). SSE endpoints (the AI generate endpoint)
 * implement idempotency in-band instead: a key HIT re-emits the terminal frame
 * (the created resource is not re-created — the same "no double-create"
 * guarantee, SSE-safe). `@SkipIdempotency()` marks such a route; the header is
 * still honored (read by the handler), just not by the body interceptor.
 */
export const SKIP_IDEMPOTENCY_KEY = "skip_idempotency";
export const SkipIdempotency = (): MethodDecorator & ClassDecorator & PropertyDecorator =>
  SetMetadata(SKIP_IDEMPOTENCY_KEY, true);

/** A stored body that was EMPTY (204 / nil body) — replayed as `undefined`. */
const EMPTY_BODY = "__IDEM_EMPTY__";

/** The JSON the keyspace stores under a key. */
interface IdemRecord {
  /** The body the API wrote (EMPTY_BODY marker for a nil body). */
  readonly body: string;
}

/** The mutating HTTP methods the key applies to (GET is excluded). */
const MUTATING = new Set(["POST", "PATCH", "PUT", "DELETE"]);

@Injectable()
export class IdempotencyInterceptor implements NestInterceptor {
  constructor(@Inject(RedisService) private readonly redis: RedisService) {}

  intercept(context: ExecutionContext, next: CallHandler): Observable<unknown> {
    const req = context.switchToHttp().getRequest<{
      method: string;
      path: string;
      body?: unknown;
      auth?:
        { kind?: string; userId?: string; apiKeyId?: string; workspaceId?: string } | undefined;
      headers: Record<string, string | string[] | undefined>;
    }>();
    const res = context
      .switchToHttp()
      .getResponse<{ statusCode: number; setHeader: (k: string, v: string) => void }>();

    // GATE (cases 5 + 6 + auth): only an authenticated actor's MUTATING
    // request carrying an `Idempotency-Key` header is honored. TASK-0127:
    // API keys are first-class idempotent callers (CI retries are exactly
    // the replay case the header exists for) — the keyspace is scoped by
    // CREDENTIAL, so no cross-key / cross-workspace replay is possible
    // (spec §Security: "the idempotency memo is scoped by credential").
    // Everything else (GET / no header / unauthenticated) passes through
    // untouched — the middleware is additive, existing endpoints unchanged.
    const key = IdempotencyInterceptor.headerValue(req.headers?.[IDEMPOTENCY_KEY_HEADER]);
    const handler = context.getHandler?.();
    const cls = context.getClass?.();
    const skip =
      (handler != null && Reflect.getMetadata(SKIP_IDEMPOTENCY_KEY, handler) === true) ||
      (cls != null && Reflect.getMetadata(SKIP_IDEMPOTENCY_KEY, cls) === true);
    const auth = req.auth;
    const credential =
      auth?.kind === "user" && auth.userId
        ? { scope: `user:${auth.userId}` }
        : auth?.kind === "api" && auth.apiKeyId && auth.workspaceId
          ? { scope: `api:${auth.workspaceId}:${auth.apiKeyId}` }
          : null;
    if (skip || !MUTATING.has(req.method) || !key || !credential) {
      return next.handle();
    }

    // TASK-0127: for API keys the memo identity is credential + ROUTE +
    // BODY digest (a different body with the same key is a different
    // operation — the spec's "body is in the memo hash" case). The user
    // keyspace keeps the TASK-0095 identity (`idem:{userId}:{key}`) for
    // compatibility with already-stored outbox records.
    const digest = credential.scope.startsWith("api:")
      ? createHash("sha256")
          .update(`${req.method} ${req.path}`)
          .update(JSON.stringify(req.body ?? null))
          .digest("hex")
          .slice(0, 32)
      : "";
    const redisKey = credential.scope.startsWith("api:")
      ? `idem:${credential.scope}:${digest}:${key}`
      : `idem:${auth!.userId}:${key}`;

    // FAIL-OPEN HIT check (cases 1 + 3): a stored body → replay it (the
    // handler is NOT re-run); a miss or a Redis error → the single-flight
    // MISS path below (a Redis error degrades to "execute, don't store").
    return from(this.lookup(redisKey)).pipe(
      switchMap((stored) => {
        if (stored !== null) {
          // TASK-0127: the replayed response carries the marker header
          // (docs/api/authentication.md §9: "Idempotency-Replayed: true
          // when a cached result is returned within 24 h").
          res.setHeader("Idempotency-Replayed", "true");
          return of(IdempotencyInterceptor.decodeBody(stored.body));
        }
        // MISS: acquire the single-flight lock (docs §9: "concurrent
        // duplicate → the in-flight result is awaited, Redis lock, 10 s"),
        // then either execute (acquired) or wait for the in-flight caller's
        // memo (a concurrent duplicate of the same operation).
        return from(this.acquireLock(redisKey)).pipe(
          switchMap((acquired) => {
            if (acquired) {
              return this.executeAndStore(next, res, redisKey);
            }
            return from(this.waitForMemo(redisKey)).pipe(
              switchMap((memo) => {
                if (memo !== null) {
                  res.setHeader("Idempotency-Replayed", "true");
                  return of(IdempotencyInterceptor.decodeBody(memo.body));
                }
                // The in-flight caller did not store a memo (a non-2xx
                // result, a crash, or a Redis hiccup) within the wait
                // window — fail open: execute, then store (the same
                // degradation as a lookup miss).
                logger.warn(`idempotency wait timed out (${redisKey}) — executing (fail-open)`);
                return this.executeAndStore(next, res, redisKey);
              }),
            );
          }),
        );
      }),
    );
  }

  /** Run the handler once; store a 2xx body under the memo key. */
  private executeAndStore(
    next: CallHandler,
    res: { statusCode: number },
    redisKey: string,
  ): Observable<unknown> {
    return next.handle().pipe(
      tap({
        next: (data) => {
          const status = res.statusCode;
          if (status >= 200 && status < 300) {
            void this.store(redisKey, data);
          }
        },
      }),
    );
  }

  /** Take the single-flight lock (SET NX EX 10). Fail-open (Redis down). */
  private async acquireLock(redisKey: string): Promise<boolean> {
    try {
      const ok = await this.redis
        .getRedis()
        .set(`idem:lock:${redisKey}`, "1", "EX", IDEM_LOCK_TTL_SECONDS, "NX");
      return ok === "OK";
    } catch {
      return true; // Redis down → execute (availability over strictness)
    }
  }

  /** Poll the memo for an in-flight duplicate's result (≤ ~1.5 s). */
  private async waitForMemo(redisKey: string): Promise<IdemRecord | null> {
    for (let i = 0; i < IDEM_WAIT_POLLS; i++) {
      await new Promise((r) => setTimeout(r, IDEM_WAIT_POLL_MS));
      const memo = await this.lookup(redisKey);
      if (memo !== null) return memo;
    }
    return null;
  }

  /** Read + parse a stored record (null = miss / error → execute). */
  private async lookup(redisKey: string): Promise<IdemRecord | null> {
    try {
      const raw = await this.redis.getRedis().get(redisKey);
      if (raw === null) return null;
      return JSON.parse(raw) as IdemRecord;
    } catch (err) {
      logger.warn(`idempotency lookup failed (${redisKey}): ${String(err)}`);
      return null; // fail-open → execute
    }
  }

  /** Store a 2xx body under the key (TTL 24 h). Fail-open on Redis error. */
  private async store(redisKey: string, data: unknown): Promise<void> {
    const body = data === undefined || data === null ? EMPTY_BODY : JSON.stringify(data);
    try {
      await this.redis
        .getRedis()
        .set(redisKey, JSON.stringify({ body } satisfies IdemRecord), "EX", IDEM_TTL_SECONDS);
    } catch (err) {
      logger.warn(`idempotency store failed (${redisKey}): ${String(err)}`);
    }
  }

  /** Reconstruct the handler's result from the stored body. */
  private static decodeBody(body: string): unknown {
    if (body === EMPTY_BODY) return undefined; // 204 / nil → an empty response
    try {
      return JSON.parse(body);
    } catch {
      return undefined;
    }
  }

  /** The header value (a single value, or the first of a repeated one). */
  private static headerValue(v: string | string[] | undefined): string | undefined {
    if (Array.isArray(v)) return v[0];
    if (typeof v === "string" && v.length > 0) return v;
    return undefined;
  }
}