feat: detect DB vs on-chain balance drift in reconciliation closes(#229)

app/api/activity/route.ts

@@ -1,9 +1,9 @@
 import { NextResponse } from "next/server";
-import { db, encodeCursor, decodeCursor } from "@/app/lib/db";
-import { getClientIdentity, checkRateLimit, rateLimitResponse } from "@/app/lib/rate-limit";
-import { recordThrottle, recordRequest } from "@/app/lib/rate-limit-metrics";
+import { decodeCursor, encodeCursor, getStore } from "@/app/lib/db";
+import { checkRateLimit, getClientIdentity, rateLimitResponse } from "@/app/lib/rate-limit";
 import { getLimitForRoute } from "@/app/lib/rate-limit-config";
-import { getCorrelationContext, logger } from "@/app/lib/logger";
+import { recordRequest, recordThrottle } from "@/app/lib/rate-limit-metrics";
+import { getCorrelationContext, logger, withCorrelationContext } from "@/app/lib/logger";
 
 function createErrorResponse(code: string, message: string, status: number) {
   const context = getCorrelationContext();
@@ -46,17 +46,37 @@ export async function GET(request: Request) {
     if (type) {
       events = events.filter((event) => event.type === type);
     }
-  }
 
-  const paginatedEvents = events.slice(0, limit);
-  const hasNext = events.length > limit;
-  const nextCursor = hasNext && paginatedEvents.length > 0 ? encodeCursor(paginatedEvents[paginatedEvents.length - 1].id) : null;
+    if (cursor) {
+      let cursorId: string;
+      try {
+        cursorId = decodeCursor(cursor);
+      } catch {
+        return createErrorResponse("INVALID_CURSOR", "Malformed cursor", 422);
+      }
+
+      const cursorIndex = events.findIndex((event) => event.id === cursorId);
+      if (cursorIndex >= 0) {
+        events = events.slice(cursorIndex + 1);
+      }
+    }
+
+    const paginatedEvents = events.slice(0, limit);
+    const hasNext = events.length > limit;
+    const nextCursor =
+      hasNext && paginatedEvents.length > 0
+        ? encodeCursor(paginatedEvents[paginatedEvents.length - 1].id)
+        : null;
 
-  logger.info('Activity list completed', { count: paginatedEvents.length, total: db.activity.size });
+    logger.info("Activity list completed", {
+      count: paginatedEvents.length,
+      total: streamRepository.activity.size,
+    });
 
-  return NextResponse.json({
-    data: paginatedEvents,
-    meta: { hasNext, nextCursor, total: db.activity.size },
-    links: { self: `/api/v1/activity?limit=${limit}` },
+    return NextResponse.json({
+      data: paginatedEvents,
+      meta: { hasNext, nextCursor, total: streamRepository.activity.size },
+      links: { self: `/api/v1/activity?limit=${limit}` },
+    });
   });
 }

app/api/audit/route.test.ts

@@ -6,7 +6,7 @@ import { JWT_SECRET } from "@/app/lib/auth";
 import { auditLogStore, resetAuditLogStore } from "@/app/lib/audit-log";
 
 function signAccessToken(role: string, actorId: string) {
-  return jwt.sign({ sub: `${actorId}-wallet`, role, actorId, iss: "streampay" }, JWT_SECRET, {
+  return jwt.sign({ sub: `${actorId}-wallet`, role, actorId, iss: "streampay", aud: "streampay-api" }, JWT_SECRET, {
     expiresIn: "15m",
   });
 }

app/api/debug/kms-sign/route.test.ts

@@ -1,111 +1,168 @@
-/**
- * Tests for POST /api/debug/kms-sign
- *
- * Covers:
- *  - 200 success path with signature
- *  - 400 for missing/empty payload
- *  - 403 in production
- *  - 500 on unexpected error
- *  - canonical error envelope shape on all error paths
- */
-
+/** @jest-environment node */
 import { POST } from "./route";
+import { requireInternalServiceAuth } from "@/app/lib/internal-service-auth";
+import { NextResponse } from "next/server";
 
-jest.mock("next/server", () => ({
-  NextResponse: {
-    json: <T>(body: T, init?: { status?: number }) => ({
-      status: init?.status ?? 200,
-      body,
-      json: async () => body,
-    }),
-  },
-}));
+// Mock the dependencies
+const mockSigner = {
+  getProviderName: () => "local-mock",
+  sign: jest.fn().mockResolvedValue(Buffer.from("mock-signature")),
+  getPublicKey: jest.fn().mockResolvedValue("mock-public-key"),
+};
 
-jest.mock("next/headers", () => ({
-  headers: () => ({ get: () => null }),
+jest.mock("../../../lib/kms/factory", () => ({
+  getSigner: () => mockSigner,
 }));
 
-function makeRequest(body: unknown) {
-  return {
-    json: async () => {
-      if (body === "THROW") throw new Error("parse error");
-      return body;
-    },
-    headers: { get: () => null },
-  } as unknown as import("next/server").NextRequest;
-}
+jest.mock("../../../lib/internal-service-auth", () => ({
+  requireInternalServiceAuth: jest.fn(),
+}));
 
-const originalEnv = process.env.NODE_ENV;
+describe("KMS Debug Sign Route", () => {
+  let originalNodeEnv: string | undefined;
 
-afterEach(() => {
-  Object.defineProperty(process.env, "NODE_ENV", { value: originalEnv, writable: true });
-});
+  beforeAll(() => {
+    originalNodeEnv = process.env.NODE_ENV;
+  });
 
-describe("POST /api/debug/kms-sign", () => {
-  it("returns 200 with a signature for a valid payload", async () => {
-    const res = await POST(makeRequest({ payload: "aGVsbG8=" }));
-    expect(res.status).toBe(200);
-    const body = (res as unknown as { body: { signature: string } }).body;
-    expect(typeof body.signature).toBe("string");
-    expect(body.signature.length).toBeGreaterThan(0);
+  afterAll(() => {
+    (process.env as any).NODE_ENV = originalNodeEnv;
   });
 
-  it("returns 400 when payload is missing", async () => {
-    const res = await POST(makeRequest({}));
-    expect(res.status).toBe(400);
-    const body = (res as unknown as { body: { error: { code: string } } }).body;
-    expect(body.error.code).toBe("KMS_SIGN_INVALID_INPUT");
+  beforeEach(() => {
+    jest.clearAllMocks();
+    (process.env as any).NODE_ENV = "development";
   });
 
-  it("returns 400 when payload is an empty string", async () => {
-    const res = await POST(makeRequest({ payload: "   " }));
-    expect(res.status).toBe(400);
-    const body = (res as unknown as { body: { error: { code: string } } }).body;
-    expect(body.error.code).toBe("KMS_SIGN_INVALID_INPUT");
+  it("returns 404 NOT_FOUND error envelope in production", async () => {
+    (process.env as any).NODE_ENV = "production";
+    const request = new Request("http://localhost/api/debug/kms-sign", {
+      method: "POST",
+      body: JSON.stringify({ payload: "hello" }),
+    });
+
+    const response = await POST(request);
+    expect(response.status).toBe(404);
+
+    const body = await response.json();
+    expect(body.code).toBe("NOT_FOUND");
+    expect(body.status).toBe(404);
   });
 
-  it("returns 400 when payload is not a string", async () => {
-    const res = await POST(makeRequest({ payload: 42 }));
-    expect(res.status).toBe(400);
-    const body = (res as unknown as { body: { error: { code: string } } }).body;
-    expect(body.error.code).toBe("KMS_SIGN_INVALID_INPUT");
+  it("returns 404 NOT_FOUND error envelope when internal auth fails", async () => {
+    // requireInternalServiceAuth returns a NextResponse (like a 404 / 401 response) if auth fails
+    const mockAuthFailureResponse = NextResponse.json({ error: "Auth failed" }, { status: 404 });
+    (requireInternalServiceAuth as jest.Mock).mockResolvedValue(mockAuthFailureResponse);
+
+    const request = new Request("http://localhost/api/debug/kms-sign", {
+      method: "POST",
+      body: JSON.stringify({ payload: "hello" }),
+    });
+
+    const response = await POST(request);
+    expect(response.status).toBe(404);
+
+    const body = await response.json();
+    expect(body.code).toBe("NOT_FOUND");
+    expect(body.status).toBe(404);
   });
 
-  it("returns 400 when body is null", async () => {
-    const res = await POST(makeRequest(null));
-    expect(res.status).toBe(400);
-    const body = (res as unknown as { body: { error: { code: string } } }).body;
-    expect(body.error.code).toBe("KMS_SIGN_INVALID_INPUT");
+  it("signs request successfully when auth is valid", async () => {
+    // requireInternalServiceAuth returns the identity details on success
+    (requireInternalServiceAuth as jest.Mock).mockResolvedValue({
+      serviceName: "debug-client",
+      keyId: "current",
+      timestamp: new Date().toISOString(),
+    });
+
+    const request = new Request("http://localhost/api/debug/kms-sign", {
+      method: "POST",
+      body: JSON.stringify({ payload: "hello world" }),
+    });
+
+    const response = await POST(request);
+    expect(response.status).toBe(200);
+
+    const body = await response.json();
+    expect(body.provider).toBe("local-mock");
+    expect(body.publicKey).toBe("mock-public-key");
+    expect(body.signature).toBe(Buffer.from("mock-signature").toString("hex"));
   });
 
-  it("returns 403 in production", async () => {
-    Object.defineProperty(process.env, "NODE_ENV", { value: "production", writable: true });
-    const res = await POST(makeRequest({ payload: "aGVsbG8=" }));
-    expect(res.status).toBe(403);
-    const body = (res as unknown as { body: { error: { code: string } } }).body;
-    expect(body.error.code).toBe("FORBIDDEN");
+  it("returns 422 INVALID_REQUEST error envelope when content-length header is too large", async () => {
+    (requireInternalServiceAuth as jest.Mock).mockResolvedValue({
+      serviceName: "debug-client",
+    });
+
+    const request = new Request("http://localhost/api/debug/kms-sign", {
+      method: "POST",
+      headers: {
+        "content-length": String(16 * 1024 + 2000),
+      },
+      body: JSON.stringify({ payload: "hello" }),
+    });
+
+    const response = await POST(request);
+    expect(response.status).toBe(422);
+
+    const body = await response.json();
+    expect(body.code).toBe("INVALID_REQUEST");
+    expect(body.status).toBe(422);
   });
 
-  it("returns 500 canonical error when json() throws", async () => {
-    const res = await POST(makeRequest("THROW"));
-    expect(res.status).toBe(500);
-    const body = (res as unknown as { body: { error: { code: string } } }).body;
-    expect(body.error.code).toBe("KMS_SIGN_FAILED");
+  it("returns 422 INVALID_FIELD_VALUE error envelope when payload size exceeds 16KB", async () => {
+    (requireInternalServiceAuth as jest.Mock).mockResolvedValue({
+      serviceName: "debug-client",
+    });
+
+    // Create a payload larger than 16KB (16 * 1024 bytes)
+    const largePayload = "a".repeat(16 * 1024 + 1);
+    const request = new Request("http://localhost/api/debug/kms-sign", {
+      method: "POST",
+      body: JSON.stringify({ payload: largePayload }),
+    });
+
+    const response = await POST(request);
+    expect(response.status).toBe(422);
+
+    const body = await response.json();
+    expect(body.code).toBe("INVALID_FIELD_VALUE");
+    expect(body.status).toBe(422);
   });
 
-  it("error envelope always has code, message, request_id", async () => {
-    const res = await POST(makeRequest({}));
-    const body = (res as unknown as { body: { error: Record<string, unknown> } }).body;
-    expect(body.error).toHaveProperty("code");
-    expect(body.error).toHaveProperty("message");
-    expect(body.error).toHaveProperty("request_id");
+  it("returns 400 MISSING_REQUIRED_FIELD error envelope when payload is empty", async () => {
+    (requireInternalServiceAuth as jest.Mock).mockResolvedValue({
+      serviceName: "debug-client",
+    });
+
+    const request = new Request("http://localhost/api/debug/kms-sign", {
+      method: "POST",
+      body: JSON.stringify({ payload: "" }),
+    });
+
+    const response = await POST(request);
+    expect(response.status).toBe(400);
+
+    const body = await response.json();
+    expect(body.code).toBe("MISSING_REQUIRED_FIELD");
+    expect(body.status).toBe(400);
   });
 
-  it("does not expose { success, error } shape (old divergent shape)", async () => {
-    const res = await POST(makeRequest({ payload: "aGVsbG8=" }));
-    const body = (res as unknown as { body: Record<string, unknown> }).body;
-    // Must NOT have top-level 'success' or top-level 'error' string
-    expect(body).not.toHaveProperty("success");
-    expect(typeof body.error).not.toBe("string");
+  it("returns 422 INVALID_REQUEST error envelope when payload is not a string", async () => {
+    (requireInternalServiceAuth as jest.Mock).mockResolvedValue({
+      serviceName: "debug-client",
+    });
+
+    const request = new Request("http://localhost/api/debug/kms-sign", {
+      method: "POST",
+      body: JSON.stringify({ payload: 12345 }),
+    });
+
+    const response = await POST(request);
+    expect(response.status).toBe(422);
+
+    const body = await response.json();
+    expect(body.code).toBe("INVALID_REQUEST");
+    expect(body.status).toBe(422);
   });
 });

app/api/debug/kms-sign/route.ts

@@ -10,9 +10,16 @@ import { errorResponse, ErrorCode } from "@/app/lib/errors";
  * Body: { "payload": "<base64-encoded string>" }
  * Response: { "signature": "<base64-encoded signature>" }
  */
-export async function POST(req: NextRequest) {
-  if (process.env.NODE_ENV === "production") {
-    return errorResponse(ErrorCode.FORBIDDEN, "This endpoint is not available in production.", 403);
+export async function POST(request: Request) {
+  // Hard-disable in production
+  if (process.env.NODE_ENV === 'production') {
+    return NextResponse.json(createError('NOT_FOUND'), { status: 404 });
+  }
+
+  // Internal-service auth (concealFailure hides auth failures as 404)
+  const authResult = await requireInternalServiceAuth(request, { concealFailure: true });
+  if (authResult instanceof NextResponse) {
+    return NextResponse.json(createError('NOT_FOUND'), { status: 404 });
   }
 
   try {

app/api/exports/exports.test.ts

@@ -6,7 +6,7 @@ import { GET as getExport } from "./[id]/route";
 const JWT_SECRET = "streampay-dev-secret-do-not-use-in-prod";
 
 function makeToken(walletAddress: string, role = "user"): string {
-  return jwt.sign({ sub: walletAddress, role }, JWT_SECRET, { expiresIn: "1h" });
+  return jwt.sign({ sub: walletAddress, role, iss: "streampay", aud: "streampay-api" }, JWT_SECRET, { expiresIn: "1h" });
 }
 
 function authRequest(url: string, token?: string): Request {
@@ -214,7 +214,7 @@ describe("Exports API — authentication and scoping", () => {
         { params: Promise.resolve({ id: data.id }) }
       );
       expect(downloadRes.status).toBe(200);
-      expect(db.exportAudit.some((r) => r.type === "export.downloaded" && r.exportId === data.id)).toBe(true);
+      expect(db.exportAudit.some((r: any) => r.type === "export.downloaded" && r.exportId === data.id)).toBe(true);
     });
 
     it("cross-tenant actor cannot use a valid signed URL for another tenant's job", async () => {
@@ -255,7 +255,6 @@ describe("Exports API — authentication and scoping", () => {
         nextAction: "pause",
         createdAt: "2026-04-01T00:00:00Z",
         updatedAt: "2026-04-01T00:00:00Z",
-        // @ts-expect-error ownerId not on Stream type but used for scoping
         ownerId: "GOWNER1",
       });
       db.streams.set("s-other", {
@@ -267,7 +266,6 @@ describe("Exports API — authentication and scoping", () => {
         nextAction: "pause",
         createdAt: "2026-04-01T00:00:00Z",
         updatedAt: "2026-04-01T00:00:00Z",
-        // @ts-expect-error ownerId not on Stream type but used for scoping
         ownerId: "GOTHER2",
       });
 

app/api/identity/me/route.test.ts

@@ -16,7 +16,7 @@ function requestWithAuthorization(authorization?: string) {
 }
 
 function signToken(payload: Record<string, unknown>, secret = TEST_SECRET) {
-  return jwt.sign(payload, secret, { algorithm: "HS256", expiresIn: "15m" });
+  return jwt.sign({ iss: "streampay", aud: "streampay-api", ...payload }, secret, { algorithm: "HS256", expiresIn: "15m" });
 }
 
 function unsignedToken(payload: Record<string, unknown>) {