feat(webhooks): harden hooks and idempotency

Author: BjornMelin

src/app/api/hooks/files/__tests__/route.test.ts

@@ -1,12 +1,16 @@
 /** @vitest-environment node */
 
 import { beforeEach, describe, expect, it, vi } from "vitest";
+import type { ReserveKeyOptions } from "@/lib/idempotency/redis";
 import type { WebhookPayload } from "@/lib/webhooks/payload";
 import { createMockNextRequest, getMockCookiesForTest } from "@/test/helpers/route";
 
 type ParseAndVerify = (req: Request) => Promise<ParseResult>;
 type BuildEventKey = (payload: WebhookPayload) => string;
-type TryReserveKey = (key: string, ttlSeconds?: number) => Promise<boolean>;
+type TryReserveKey = (
+  key: string,
+  ttlSecondsOrOptions?: number | ReserveKeyOptions
+) => Promise<boolean>;
 
 type ParseResult = { ok: boolean; payload?: WebhookPayload };
 type FilesRouteModule = typeof import("../route");
@@ -50,7 +54,8 @@ vi.mock("@/lib/idempotency/redis", () => ({
       this.name = "IdempotencyServiceUnavailableError";
     }
   },
-  tryReserveKey: (key: string, ttl?: number) => tryReserveKeyMock(key, ttl),
+  tryReserveKey: (key: string, ttlSecondsOrOptions?: number | ReserveKeyOptions) =>
+    tryReserveKeyMock(key, ttlSecondsOrOptions),
 }));
 
 vi.mock("@/lib/supabase/admin", () => ({
@@ -162,7 +167,10 @@ describe("POST /api/hooks/files", () => {
     expect(res.status).toBe(200);
     expect(json.duplicate).toBe(true);
     expect(json.ok).toBe(true);
-    expect(tryReserveKeyMock).toHaveBeenCalledWith("file-event-key-1", 300);
+    expect(tryReserveKeyMock).toHaveBeenCalledWith("file-event-key-1", {
+      degradedMode: "fail_closed",
+      ttlSeconds: 300,
+    });
   });
 
   it("processes INSERT with uploading status successfully", async () => {

src/app/api/hooks/trips/__tests__/route.test.ts

@@ -10,7 +10,11 @@ const { afterAllHook: upstashAfterAllHook, beforeEachHook: upstashBeforeEachHook
 
 type ParseAndVerify = (req: Request) => Promise<ParseResult>;
 type BuildEventKey = (payload: WebhookPayload) => string;
-type TryReserveKey = (key: string, ttlSeconds?: number) => Promise<boolean>;
+type TryReserveKeyOptions = { degradedMode?: string; ttlSeconds?: number };
+type TryReserveKey = (
+  key: string,
+  ttlSecondsOrOptions?: number | TryReserveKeyOptions
+) => Promise<boolean>;
 type SendNotifications = (
   payload: WebhookPayload,
   eventKey: string
@@ -63,7 +67,8 @@ vi.mock("@/lib/idempotency/redis", () => ({
       this.name = "IdempotencyServiceUnavailableError";
     }
   },
-  tryReserveKey: (key: string, ttl?: number) => tryReserveKeyMock(key, ttl),
+  tryReserveKey: (key: string, ttlSecondsOrOptions?: number | TryReserveKeyOptions) =>
+    tryReserveKeyMock(key, ttlSecondsOrOptions),
 }));
 
 vi.mock("@/lib/notifications/collaborators", () => ({
@@ -238,7 +243,10 @@ describe("POST /api/hooks/trips", () => {
     const json = await res.json();
     expect(res.status).toBe(200);
     expect(json.duplicate).toBe(true);
-    expect(tryReserveKeyMock).toHaveBeenCalledWith("event-key-1", 300);
+    expect(tryReserveKeyMock).toHaveBeenCalledWith("event-key-1", {
+      degradedMode: "fail_closed",
+      ttlSeconds: 300,
+    });
   });
 
   it("enqueues to QStash when configured", async () => {

src/app/api/jobs/notify-collaborators/__tests__/route.test.ts

@@ -10,7 +10,7 @@ const {
   beforeEachHook: upstashBeforeEachHook,
   mocks: upstashMocks,
 } = setupUpstashTestEnvironment();
-type TryReserveKey = (key: string, ttlSeconds?: number) => Promise<boolean>;
+type TryReserveKey = (key: string, ttlSecondsOrOptions?: unknown) => Promise<boolean>;
 type SendNotifications = (
   payload: NotifyJob["payload"],
   eventKey: string
@@ -50,7 +50,8 @@ vi.mock("@/lib/env/server", () => ({
 }));
 
 vi.mock("@/lib/idempotency/redis", () => ({
-  tryReserveKey: (key: string, ttl?: number) => tryReserveKeyMock(key, ttl),
+  tryReserveKey: (key: string, ttlSecondsOrOptions?: unknown) =>
+    tryReserveKeyMock(key, ttlSecondsOrOptions),
 }));
 
 vi.mock("@/lib/notifications/collaborators", () => ({

src/app/api/jobs/notify-collaborators/route.ts

@@ -7,11 +7,11 @@ import "server-only";
 import { notifyJobSchema } from "@schemas/webhooks";
 import { NextResponse } from "next/server";
 import { errorResponse, validateSchema } from "@/lib/api/route-helpers";
-import { getClientIpFromHeaders } from "@/lib/http/ip";
 import { tryReserveKey } from "@/lib/idempotency/redis";
 import { sendCollaboratorNotifications } from "@/lib/notifications/collaborators";
 import { pushToDLQ } from "@/lib/qstash/dlq";
 import { getQstashReceiver, verifyQstashRequest } from "@/lib/qstash/receiver";
+import { getTrustedRateLimitIdentifierFromHeaders } from "@/lib/ratelimit/identifier";
 import { withTelemetrySpan } from "@/lib/telemetry/span";
 
 /** Max retries configured for QStash (per ADR-0048) */
@@ -70,12 +70,13 @@ export async function POST(req: Request) {
         const verified = await verifyQstashRequest(req, receiver);
         if (!verified.ok) {
           try {
-            const ip = getClientIpFromHeaders(req.headers);
+            const ipHash = getTrustedRateLimitIdentifierFromHeaders(req.headers);
+            const pathname = new URL(req.url).pathname;
             span.addEvent("unauthorized_attempt", {
               hasSignature: verified.reason !== "missing_signature",
-              ip: ip === "unknown" ? undefined : ip,
+              ipHash: ipHash === "unknown" ? undefined : ipHash,
+              path: pathname,
               reason: verified.reason,
-              url: req.url,
             });
           } catch (spanError) {
             span.recordException(spanError as Error);
@@ -97,7 +98,10 @@ export async function POST(req: Request) {
         span.setAttribute("op", payload.type);
 
         // De-duplicate at worker level as well to avoid double-send on retries
-        const unique = await tryReserveKey(`notify:${eventKey}`, 300);
+        const unique = await tryReserveKey(`notify:${eventKey}`, {
+          degradedMode: "fail_closed",
+          ttlSeconds: 300,
+        });
         if (!unique) {
           span.setAttribute("event.duplicate", true);
           return NextResponse.json({ duplicate: true, ok: true });

src/lib/idempotency/__tests__/redis.test.ts

@@ -3,6 +3,7 @@
 import { beforeEach, describe, expect, it, vi } from "vitest";
 
 const warnRedisUnavailableMock = vi.fn();
+const emitOperationalAlertOncePerWindowMock = vi.fn();
 
 // Mock redis client and factories
 const existsMock = vi.fn(async (_key: string) => 0);
@@ -28,6 +29,11 @@ vi.mock("@/lib/telemetry/redis", () => ({
   warnRedisUnavailable: (...args: unknown[]) => warnRedisUnavailableMock(...args),
 }));
 
+vi.mock("@/lib/telemetry/degraded-mode", () => ({
+  emitOperationalAlertOncePerWindow: (...args: unknown[]) =>
+    emitOperationalAlertOncePerWindowMock(...args),
+}));
+
 describe("idempotency redis helpers", () => {
   beforeEach(() => {
     existsMock.mockReset();
@@ -36,6 +42,7 @@ describe("idempotency redis helpers", () => {
     redisClient = { del: delMock, exists: existsMock, set: setMock };
     getRedisMock.mockReset();
     warnRedisUnavailableMock.mockReset();
+    emitOperationalAlertOncePerWindowMock.mockReset();
   });
 
   describe("hasKey", () => {

src/lib/idempotency/redis.ts

@@ -12,6 +12,7 @@ import "server-only";
 import { z } from "zod";
 
 import { getRedis } from "@/lib/redis";
+import { emitOperationalAlertOncePerWindow } from "@/lib/telemetry/degraded-mode";
 import { warnRedisUnavailable } from "@/lib/telemetry/redis";
 
 const REDIS_FEATURE = "idempotency.keys";
@@ -61,6 +62,28 @@ export interface ReserveKeyOptions {
    * Can also be set globally via IDEMPOTENCY_FAIL_OPEN env var (read at module load).
    */
   failOpen?: boolean;
+
+  /**
+   * Alias for fail-open behavior with explicit naming.
+   * When set, overrides failOpen.
+   */
+  degradedMode?: "fail_closed" | "fail_open";
+}
+
+function resolveDegradedMode(options: {
+  degradedMode?: "fail_closed" | "fail_open";
+  failOpen?: boolean;
+}): "fail_closed" | "fail_open" {
+  if (options.degradedMode) return options.degradedMode;
+  if (options.failOpen === true) return "fail_open";
+  if (options.failOpen === false) return "fail_closed";
+  return DEFAULT_FAIL_OPEN ? "fail_open" : "fail_closed";
+}
+
+function getIdempotencyNamespace(key: string): string {
+  const idx = key.indexOf(":");
+  const namespace = idx === -1 ? key : key.slice(0, idx);
+  return namespace.trim().slice(0, 64) || "unknown";
 }
 
 /**
@@ -91,7 +114,8 @@ export async function tryReserveKey(
       : ttlSecondsOrOptions;
 
   const ttlSeconds = options.ttlSeconds ?? 300;
-  const failOpen = options.failOpen ?? DEFAULT_FAIL_OPEN;
+  const degradedMode = resolveDegradedMode(options);
+  const failOpen = degradedMode === "fail_open";
 
   const redis = getRedis();
   if (!redis) {
@@ -101,13 +125,40 @@ export async function tryReserveKey(
       throw new IdempotencyServiceUnavailableError();
     }
 
+    emitOperationalAlertOncePerWindow({
+      attributes: {
+        degradedMode: "fail_open",
+        namespace: getIdempotencyNamespace(key),
+        reason: "redis_unavailable",
+      },
+      event: "idempotency.degraded",
+      windowMs: 60_000,
+    });
+
     // Fail open: allow processing (may cause duplicates during Redis outage)
     return true;
   }
 
   const namespaced = `idemp:${key}`;
-  const result = await redis.set(namespaced, "1", { ex: ttlSeconds, nx: true });
-  return result === "OK";
+  try {
+    const result = await redis.set(namespaced, "1", { ex: ttlSeconds, nx: true });
+    return result === "OK";
+  } catch (_error) {
+    warnRedisUnavailable(REDIS_FEATURE);
+    if (!failOpen) {
+      throw new IdempotencyServiceUnavailableError();
+    }
+    emitOperationalAlertOncePerWindow({
+      attributes: {
+        degradedMode: "fail_open",
+        namespace: getIdempotencyNamespace(key),
+        reason: "redis_error",
+      },
+      event: "idempotency.degraded",
+      windowMs: 60_000,
+    });
+    return true;
+  }
 }
 
 /**
@@ -121,7 +172,8 @@ export async function hasKey(
   key: string,
   options?: { failOpen?: boolean }
 ): Promise<boolean> {
-  const failOpen = options?.failOpen ?? DEFAULT_FAIL_OPEN;
+  const degradedMode = resolveDegradedMode({ failOpen: options?.failOpen });
+  const failOpen = degradedMode === "fail_open";
 
   const redis = getRedis();
   if (!redis) {
@@ -132,8 +184,34 @@ export async function hasKey(
   }
 
   const namespaced = `idemp:${key}`;
-  const result = await redis.exists(namespaced);
-  return result > 0;
+  try {
+    const result = await redis.exists(namespaced);
+    return result > 0;
+  } catch (_error) {
+    warnRedisUnavailable(REDIS_FEATURE);
+    if (!failOpen) {
+      emitOperationalAlertOncePerWindow({
+        attributes: {
+          degradedMode: "fail_closed",
+          namespace: getIdempotencyNamespace(key),
+          reason: "redis_error",
+        },
+        event: "idempotency.degraded",
+        windowMs: 60_000,
+      });
+      return true;
+    }
+    emitOperationalAlertOncePerWindow({
+      attributes: {
+        degradedMode: "fail_open",
+        namespace: getIdempotencyNamespace(key),
+        reason: "redis_error",
+      },
+      event: "idempotency.degraded",
+      windowMs: 60_000,
+    });
+    return false;
+  }
 }
 
 /**
@@ -146,7 +224,8 @@ export async function releaseKey(
   key: string,
   options?: { failOpen?: boolean }
 ): Promise<boolean> {
-  const failOpen = options?.failOpen ?? DEFAULT_FAIL_OPEN;
+  const degradedMode = resolveDegradedMode({ failOpen: options?.failOpen });
+  const failOpen = degradedMode === "fail_open";
 
   const redis = getRedis();
   if (!redis) {
@@ -158,6 +237,23 @@ export async function releaseKey(
   }
 
   const namespaced = `idemp:${key}`;
-  const result = await redis.del(namespaced);
-  return result > 0;
+  try {
+    const result = await redis.del(namespaced);
+    return result > 0;
+  } catch (_error) {
+    warnRedisUnavailable(REDIS_FEATURE);
+    if (!failOpen) {
+      throw new IdempotencyServiceUnavailableError();
+    }
+    emitOperationalAlertOncePerWindow({
+      attributes: {
+        degradedMode: "fail_open",
+        namespace: getIdempotencyNamespace(key),
+        reason: "redis_error",
+      },
+      event: "idempotency.degraded",
+      windowMs: 60_000,
+    });
+    return false;
+  }
 }