diff --git a/src/domain/accounts/index.types.d.ts b/src/domain/accounts/index.types.d.ts index e55dc11d3..8cadb3b5b 100644 --- a/src/domain/accounts/index.types.d.ts +++ b/src/domain/accounts/index.types.d.ts @@ -185,6 +185,22 @@ interface IAccountsRepository { findByNpub(npub: Npub): Promise update(account: Account): Promise + transitionBridgeKycStatus( + id: AccountId, + nextStatus: + | "open" + | "not_started" + | "incomplete" + | "awaiting_questionnaire" + | "awaiting_ubo" + | "under_review" + | "paused" + | "approved" + | "rejected" + | "offboarded", + ): Promise< + { changed: boolean; previousStatus?: Account["bridgeKycStatus"] } | RepositoryError + > updateBridgeFields( id: AccountId, fields: { diff --git a/src/services/bridge/webhook-server/routes/external-account.ts b/src/services/bridge/webhook-server/routes/external-account.ts index 2163a3cba..fa76eae8c 100644 --- a/src/services/bridge/webhook-server/routes/external-account.ts +++ b/src/services/bridge/webhook-server/routes/external-account.ts @@ -32,6 +32,18 @@ export const externalAccountHandler = async (req: Request, res: Response) => { } try { + // Idempotency FIRST: the lock used to be taken after the external-account + // upsert, so a retried delivery re-persisted before being detected. + const lockKey = `bridge-external-account:${event_id}` + const lockResult = await LockService().lockIdempotencyKey(lockKey as IdempotencyKey) + if (lockResult instanceof Error) { + baseLogger.info( + { customer_id, event_id, id }, + "Duplicate Bridge external account webhook", + ) + return res.status(200).json({ status: "already_processed" }) + } + const bridgeCustomerId = toBridgeCustomerId(customer_id) const account = await AccountsRepository().findByBridgeCustomerId(bridgeCustomerId) if (account instanceof Error) { @@ -65,16 +77,6 @@ export const externalAccountHandler = async (req: Request, res: Response) => { "Bridge external account persisted", ) - const lockKey = `bridge-external-account:${event_id}` - const lockResult = await LockService().lockIdempotencyKey(lockKey as IdempotencyKey) - if (lockResult instanceof Error) { - baseLogger.info( - { customer_id, event_id, id }, - "Duplicate Bridge external account webhook", - ) - return res.status(200).json({ status: "already_processed" }) - } - return res.status(200).json({ status: "success" }) } catch (error) { baseLogger.error( diff --git a/src/services/bridge/webhook-server/routes/kyc.ts b/src/services/bridge/webhook-server/routes/kyc.ts index d789d6cb3..c0d153d27 100644 --- a/src/services/bridge/webhook-server/routes/kyc.ts +++ b/src/services/bridge/webhook-server/routes/kyc.ts @@ -70,6 +70,17 @@ export const kycHandler = async (req: Request, res: Response) => { } try { + // Idempotency FIRST: Bridge retries reuse the same event_id, and the lock + // used to be taken only AFTER all side effects ran (pure bookkeeping — + // it prevented nothing). Locked means a delivery of this exact event is + // already being (or was) processed. + const lockKey = `bridge-kyc:${event_id}` + const lockResult = await LockService().lockIdempotencyKey(lockKey as IdempotencyKey) + if (lockResult instanceof Error) { + baseLogger.info({ customerId, event_id }, "Duplicate Bridge KYC webhook") + return res.status(200).json({ status: "already_processed" }) + } + const bridgeCustomerId = toBridgeCustomerId(customerId) const account = await AccountsRepository().findByBridgeCustomerId(bridgeCustomerId) if (account instanceof Error) { @@ -88,71 +99,64 @@ export const kycHandler = async (req: Request, res: Response) => { "paused", ]) - // Map Bridge customer status fields to our internal kyc status - // Bridge customer.status values: not_started, active (approved), rejected, offboarded + // Map Bridge customer status fields to our internal kyc status. + // Bridge customer.status values: not_started, active (approved), rejected, + // offboarded, plus the pending set above. + let nextStatus: NonNullable | undefined if (status === "not_started") { - const nextStatus = "not_started" as const - const result = await AccountsRepository().updateBridgeFields(account.id, { - bridgeKycStatus: nextStatus, - }) - - if (result instanceof Error) { - baseLogger.error( - { accountId: account.id, error: result }, - "Failed to update KYC status", - ) - return res.status(500).json({ error: "Failed to update status" }) - } - - baseLogger.info({ accountId: account.id, customerId }, "Bridge KYC not started") + nextStatus = "not_started" } else if (PENDING_BRIDGE_STATUSES.has(status)) { - const nextStatus = status as NonNullable - const result = await AccountsRepository().updateBridgeFields(account.id, { - bridgeKycStatus: nextStatus, - }) - - if (result instanceof Error) { - baseLogger.error( - { accountId: account.id, error: result }, - "Failed to update KYC status", - ) - return res.status(500).json({ error: "Failed to update status" }) - } - - await notifyKycStatusChange({ - account, - previousStatus: account.bridgeKycStatus, - nextStatus, - rejectionReasons, - }) + nextStatus = status as NonNullable + } else if (status === "active" || status === "approved") { + nextStatus = "approved" + } else if (status === "rejected") { + nextStatus = "rejected" + } else if (status === "offboarded") { + nextStatus = "offboarded" + } + if (!nextStatus) { baseLogger.info( - { accountId: account.id, customerId, status }, - "Bridge KYC moved to pending", + { accountId: account.id, customerId, status, event_type }, + "Unhandled Bridge customer status — no action taken", ) - } else if (status === "active" || status === "approved") { - const nextStatus = "approved" as const - const result = await AccountsRepository().updateBridgeFields(account.id, { - bridgeKycStatus: nextStatus, - }) + return res.status(200).json({ status: "success" }) + } - if (result instanceof Error) { - baseLogger.error( - { accountId: account.id, error: result }, - "Failed to update KYC status", - ) - return res.status(500).json({ error: "Failed to update status" }) - } + // Atomic compare-and-swap: only the delivery that actually changes the + // status wins. Losers (distinct events carrying the same status, e.g. + // customer.created racing customer.updated — the double-push bug) skip + // notifications and side effects entirely. + const transition = await AccountsRepository().transitionBridgeKycStatus( + account.id, + nextStatus, + ) + if (transition instanceof Error) { + baseLogger.error( + { accountId: account.id, error: transition }, + "Failed to update KYC status", + ) + return res.status(500).json({ error: "Failed to update status" }) + } + if (!transition.changed) { + baseLogger.info( + { accountId: account.id, customerId, status, event_type }, + "Bridge KYC status already current — skipping notification/side effects", + ) + return res.status(200).json({ status: "already_current" }) + } + if (nextStatus !== "not_started") { await notifyKycStatusChange({ account, - previousStatus: account.bridgeKycStatus, + previousStatus: transition.previousStatus, nextStatus, rejectionReasons, }) + } + if (nextStatus === "approved") { baseLogger.info({ accountId: account.id, customerId }, "Bridge KYC approved") - const vaResult = await BridgeService.createVirtualAccount(account.id) if (vaResult instanceof Error) { baseLogger.error( @@ -165,71 +169,20 @@ export const kycHandler = async (req: Request, res: Response) => { "Virtual account auto-created after KYC approval", ) } - } else if (status === "rejected") { - const nextStatus = "rejected" as const - const result = await AccountsRepository().updateBridgeFields(account.id, { - bridgeKycStatus: nextStatus, - }) - - if (result instanceof Error) { - baseLogger.error( - { accountId: account.id, error: result }, - "Failed to update KYC status", - ) - return res.status(500).json({ error: "Failed to update status" }) - } - - await notifyKycStatusChange({ - account, - previousStatus: account.bridgeKycStatus, - nextStatus, - rejectionReasons, - }) - + } else if (nextStatus === "rejected") { baseLogger.warn( - { - accountId: account.id, - customerId, - rejectionReasons, - }, + { accountId: account.id, customerId, rejectionReasons }, "Bridge KYC rejected", ) - } else if (status === "offboarded") { - const nextStatus = "offboarded" as const - const result = await AccountsRepository().updateBridgeFields(account.id, { - bridgeKycStatus: nextStatus, - }) - - if (result instanceof Error) { - baseLogger.error( - { accountId: account.id, error: result }, - "Failed to update KYC status", - ) - return res.status(500).json({ error: "Failed to update status" }) - } - - await notifyKycStatusChange({ - account, - previousStatus: account.bridgeKycStatus, - nextStatus, - rejectionReasons, - }) - + } else if (nextStatus === "offboarded") { baseLogger.warn({ accountId: account.id, customerId }, "Bridge KYC offboarded") } else { baseLogger.info( - { accountId: account.id, customerId, status, event_type }, - "Unhandled Bridge customer status — no action taken", + { accountId: account.id, customerId, status: nextStatus }, + "Bridge KYC status updated", ) } - const lockKey = `bridge-kyc:${event_id}` - const lockResult = await LockService().lockIdempotencyKey(lockKey as IdempotencyKey) - if (lockResult instanceof Error) { - baseLogger.info({ customerId, event_id }, "Duplicate Bridge KYC webhook") - return res.status(200).json({ status: "already_processed" }) - } - return res.status(200).json({ status: "success" }) } catch (error) { baseLogger.error({ error, customerId }, "Error processing Bridge KYC webhook") diff --git a/src/services/mongoose/accounts.ts b/src/services/mongoose/accounts.ts index 0e6d2fb64..447fef1b5 100644 --- a/src/services/mongoose/accounts.ts +++ b/src/services/mongoose/accounts.ts @@ -174,6 +174,48 @@ export const AccountsRepository = (): IAccountsRepository => { } } + // Atomic KYC status transition (bridge webhook dedupe): updates only when + // the status actually differs, and reports whether THIS call made the + // change. Concurrent webhook deliveries for the same target status (Bridge + // retries, or customer.created + customer.updated racing) collapse to one + // winner — losers see changed:false and must skip notifications and + // side effects. Returns the pre-update status of the winning transition. + const transitionBridgeKycStatus = async ( + id: AccountId, + nextStatus: + | "open" + | "not_started" + | "incomplete" + | "awaiting_questionnaire" + | "awaiting_ubo" + | "under_review" + | "paused" + | "approved" + | "rejected" + | "offboarded", + ): Promise< + { changed: boolean; previousStatus?: Account["bridgeKycStatus"] } | RepositoryError + > => { + try { + const result = await Account.findOneAndUpdate( + { _id: toObjectId(id), bridgeKycStatus: { $ne: nextStatus } }, + { $set: { bridgeKycStatus: nextStatus } }, + { new: false }, + ) + if (!result) { + const exists = await Account.exists({ _id: toObjectId(id) }) + if (!exists) return new RepositoryError("Account not found") + return { changed: false } + } + return { + changed: true, + previousStatus: result.bridgeKycStatus as Account["bridgeKycStatus"], + } + } catch (error) { + return parseRepositoryError(error) + } + } + const updateBridgeFields = async ( id: AccountId, fields: { @@ -238,6 +280,7 @@ export const AccountsRepository = (): IAccountsRepository => { findByUsername, findByNpub, update, + transitionBridgeKycStatus, updateBridgeFields, findByBridgeEthereumAddress, findByBridgeCustomerId, diff --git a/test/flash/unit/services/bridge/webhook-server/kyc.spec.ts b/test/flash/unit/services/bridge/webhook-server/kyc.spec.ts index 162333228..5a66609d7 100644 --- a/test/flash/unit/services/bridge/webhook-server/kyc.spec.ts +++ b/test/flash/unit/services/bridge/webhook-server/kyc.spec.ts @@ -43,6 +43,7 @@ import { Request, Response } from "express" import { kycHandler } from "@services/bridge/webhook-server/routes/kyc" import { AccountsRepository } from "@services/mongoose/accounts" import { LockService } from "@services/lock" +import BridgeService from "@services/bridge" import { sendBridgeKycNotificationBestEffort } from "@app/bridge/send-kyc-notification" const makeRes = () => { @@ -66,7 +67,9 @@ describe("kycHandler", () => { jest.clearAllMocks() ;(AccountsRepository as jest.Mock).mockReturnValue({ findByBridgeCustomerId: jest.fn().mockResolvedValue(mockAccount), - updateBridgeFields: jest.fn().mockResolvedValue(mockAccount), + transitionBridgeKycStatus: jest + .fn() + .mockResolvedValue({ changed: true, previousStatus: "incomplete" }), }) ;(LockService as jest.Mock).mockReturnValue({ lockIdempotencyKey: jest.fn().mockResolvedValue(true), @@ -122,7 +125,9 @@ describe("kycHandler", () => { ...mockAccount, bridgeKycStatus: "open", }), - updateBridgeFields: jest.fn().mockResolvedValue(mockAccount), + transitionBridgeKycStatus: jest + .fn() + .mockResolvedValue({ changed: true, previousStatus: "open" }), }) const req = makeReq({ @@ -151,7 +156,9 @@ describe("kycHandler", () => { ...mockAccount, bridgeKycStatus: undefined, }), - updateBridgeFields: jest.fn().mockResolvedValue(mockAccount), + transitionBridgeKycStatus: jest + .fn() + .mockResolvedValue({ changed: true, previousStatus: undefined }), }) const req = makeReq({ @@ -175,7 +182,7 @@ describe("kycHandler", () => { ...mockAccount, bridgeKycStatus: "approved", }), - updateBridgeFields: jest.fn().mockResolvedValue(mockAccount), + transitionBridgeKycStatus: jest.fn().mockResolvedValue({ changed: false }), }) const req = makeReq({ @@ -214,4 +221,47 @@ describe("kycHandler", () => { rejectionReasons: [{ reason: "Document expired" }], }) }) + + it("skips side effects when a racing delivery already applied the status", async () => { + // customer.created and customer.updated race: the loser's CAS reports + // changed:false and must not notify or create a virtual account. + ;(AccountsRepository as jest.Mock).mockReturnValue({ + findByBridgeCustomerId: jest.fn().mockResolvedValue(mockAccount), + transitionBridgeKycStatus: jest.fn().mockResolvedValue({ changed: false }), + }) + + const req = makeReq({ + event_id: "evt-race-loser", + event_type: "customer.created", + event_object: { id: customerId, status: "active" }, + }) + const res = makeRes() + + await kycHandler(req, res) + + expect(sendBridgeKycNotificationBestEffort).not.toHaveBeenCalled() + expect(BridgeService.createVirtualAccount).not.toHaveBeenCalled() + expect(res.json).toHaveBeenCalledWith({ status: "already_current" }) + }) + + it("rejects duplicate event ids before touching the account (lock is first)", async () => { + const findByBridgeCustomerId = jest.fn() + ;(AccountsRepository as jest.Mock).mockReturnValue({ findByBridgeCustomerId }) + ;(LockService as jest.Mock).mockReturnValue({ + lockIdempotencyKey: jest.fn().mockResolvedValue(new Error("already locked")), + }) + + const req = makeReq({ + event_id: "evt-retry-dup", + event_type: "customer.updated.status_transitioned", + event_object: { id: customerId, status: "active" }, + }) + const res = makeRes() + + await kycHandler(req, res) + + expect(res.json).toHaveBeenCalledWith({ status: "already_processed" }) + expect(findByBridgeCustomerId).not.toHaveBeenCalled() + expect(sendBridgeKycNotificationBestEffort).not.toHaveBeenCalled() + }) })