ENG-695: make initiateCashout single-use per offerId

src/app/offers/CashoutManager.ts

@@ -179,6 +179,28 @@ const CashoutManager = {
       return validOffer
     }
 
+    // Single-use gate (ENG-695): exactly one execution per offer gets past it.
+    // It sits after validation, so an offer that fails a transient check (an
+    // ERPNext bank-account lookup, say) is not burned and can be retried. It
+    // sits before execute(), the only step that drafts the ERPNext cashout and
+    // pays IBEX. A Redis error here refuses the payout (fail closed). Once
+    // consumed, an offer is never executed again, even if execute() fails: the
+    // user needs a new offer, and a partial failure is reconciled by ops.
+    const consumed = await Storage.consume(id)
+    if (consumed instanceof Error) {
+      notifyOpsEvent({
+        flow: "cashout",
+        phase: "failed",
+        status: "failed",
+        accountId: providedWallet.accountId,
+        amount: displayAmount,
+        step: "consumeOffer",
+        error: consumed.constructor.name,
+        meta: { offerId: id },
+      })
+      return consumed
+    }
+
     const executedOffer = await validOffer.execute()
     if (executedOffer instanceof Error) return executedOffer
     else {

src/app/offers/storage/Redis.ts

@@ -1,7 +1,14 @@
 import { randomUUID } from "crypto"
 
-import { CacheServiceError, CacheUndefinedError, OfferNotFound } from "@domain/cache"
-import { RedisCacheService } from "@services/cache"
+import {
+  CacheServiceError,
+  CacheUndefinedError,
+  OfferAlreadyExecuted,
+  OfferNotFound,
+} from "@domain/cache"
+import { UuidRegex } from "@domain/shared"
+import { RedisCacheService, consumeCacheKey } from "@services/cache"
+import { baseLogger } from "@services/logger"
 
 import ValidOffer from "../ValidOffer"
 
@@ -10,30 +17,100 @@ import { parseRepositoryError } from "../../../services/mongoose/utils"
 import { OffersSerde } from "./OffersSerde"
 import PersistedOffer from "./PersistedOffer"
 
+const OFFER_TTL_SECS = 3600 as Seconds
+
+const offerKey = (id: OfferId): string => `offers:${id}`
+// Left behind when an offer is consumed, so a later attempt to execute it is
+// told the offer was already used rather than that it never existed. It is for
+// error specificity only: the DEL in `consume` alone decides who executes. It
+// sits outside the offers: prefix, so no offer id can make offerKey name it.
+const executedKey = (id: OfferId): string => `offer-executed:${id}`
+
+// `add` mints every offer id with randomUUID(). The id `get` and `consume`
+// receive is client input (initiateCashout's offerId), so anything that is not
+// a UUID is refused before it becomes part of a Redis key.
+const isOfferId = (id: OfferId): boolean => UuidRegex.test(id)
+
+// The offer record is gone: it expired, never existed, or an execution consumed
+// it. Only the last leaves a tombstone.
+const missingOfferError = async (
+  id: OfferId,
+): Promise<OfferAlreadyExecuted | OfferNotFound> => {
+  const tombstone = await RedisCacheService().get<string>({ key: executedKey(id) })
+  if (tombstone instanceof CacheUndefinedError) return new OfferNotFound()
+  if (tombstone instanceof CacheServiceError) {
+    baseLogger.warn(
+      { offerId: id, err: tombstone },
+      "Could not read the executed-offer tombstone; reporting the offer as not found",
+    )
+    return new OfferNotFound()
+  }
+  return new OfferAlreadyExecuted()
+}
+
 const Redis = {
   add: async (o: ValidOffer): Promise<PersistedOffer | CacheServiceError> => {
     const id = randomUUID() as OfferId // could use hash of offer details with getOrSet
     const result = await RedisCacheService().set({
-      key: `offers:${id}`,
+      key: offerKey(id),
       value: OffersSerde.serialize(o.details),
-      ttlSecs: 3600 as Seconds,
+      ttlSecs: OFFER_TTL_SECS,
     })
     if (result instanceof CacheServiceError) return result
     return new PersistedOffer(id, o.details)
   },
 
   get: async (id: OfferId): Promise<PersistedOffer | CacheServiceError> => {
+    if (!isOfferId(id)) return new OfferNotFound()
     try {
       const result: string | CacheServiceError = await RedisCacheService().get({
-        key: `offers:${id}`,
+        key: offerKey(id),
       })
-      if (result instanceof CacheUndefinedError) return new OfferNotFound()
+      if (result instanceof CacheUndefinedError) return await missingOfferError(id)
       if (result instanceof CacheServiceError) return result
       else return new PersistedOffer(id, OffersSerde.deserialize(result))
     } catch (err) {
       return parseRepositoryError(err)
     }
   },
+
+  /**
+   * Single-use gate for executing an offer: takes the record out of Redis with
+   * one DEL, so exactly one caller per offer gets `true`. A concurrent
+   * duplicate gets OfferAlreadyExecuted. A Redis error is returned as is, and
+   * the caller must not execute an offer it could not consume (fail closed).
+   * An id that is not a UUID names no offer: OfferNotFound, and no Redis call.
+   */
+  consume: async (
+    id: OfferId,
+  ): Promise<true | OfferAlreadyExecuted | CacheServiceError> => {
+    if (!isOfferId(id)) return new OfferNotFound()
+
+    // The tombstone goes first, so no retry can land between a won DEL and the
+    // tombstone write. It is best effort. If it fails, a later retry sees
+    // OfferNotFound instead of OfferAlreadyExecuted, and nothing is paid twice
+    // either way. A tombstone left by a caller whose DEL then fails or loses is
+    // harmless, because `get` reads it only once the offer record is gone.
+    const tombstone = await RedisCacheService().set({
+      key: executedKey(id),
+      value: "1",
+      ttlSecs: OFFER_TTL_SECS,
+    })
+    if (tombstone instanceof CacheServiceError) {
+      baseLogger.warn(
+        { offerId: id, err: tombstone },
+        "Could not write the executed-offer tombstone; consuming the offer anyway",
+      )
+    }
+
+    const consumed = await consumeCacheKey({ key: offerKey(id) })
+    if (consumed instanceof CacheServiceError) return consumed
+    // The caller read this offer moments ago, so a key that is already gone was
+    // removed by a concurrent execution of the same offer (or, at the very edge
+    // of its TTL, expired). Either way, this caller must not execute it.
+    if (!consumed) return new OfferAlreadyExecuted()
+    return true
+  },
 }
 
 export default Redis

src/domain/cache/errors.ts

@@ -10,3 +10,4 @@ export class UnknownCacheServiceError extends CacheServiceError {
 }
 
 export class OfferNotFound extends CacheServiceError {}
+export class OfferAlreadyExecuted extends CacheServiceError {}

src/graphql/error-map.ts

@@ -636,6 +636,16 @@ export const mapError = (error: ApplicationError): CustomApolloError => {
         message: "Offer not available. Try again.",
         logger: baseLogger,
       })
+    case "OfferAlreadyExecuted":
+      // Mostly a retry after a lost response, so the first execution may have
+      // paid, and the client has no cashout id to check (ENG-716). The copy
+      // sends the user to their transactions before a new cashout, and stays
+      // true when that first execution failed before any money moved.
+      return bridgeGqlError({
+        code: "OFFER_ALREADY_EXECUTED",
+        message:
+          "This cashout offer was already submitted and can't be reused. Check your transactions before starting a new cashout.",
+      })
 
     case "BridgeInvalidAmountError":
       message =

src/graphql/public/root/mutation/offers/initiate-cash-out.ts

@@ -1,8 +1,10 @@
 import CashoutManager from "@app/offers/CashoutManager"
 import { Cashout } from "@config"
+import { OfferAlreadyExecuted } from "@domain/cache"
 import { NotImplementedError } from "@domain/errors"
 import { ErrorLevel } from "@domain/shared"
 import { InternalServerError } from "@graphql/error"
+import { mapToGqlErrorList } from "@graphql/error-map"
 import { GT } from "@graphql/index"
 import IError from "@graphql/shared/types/abstract/error"
 import WalletId from "@graphql/shared/types/scalar/wallet-id"
@@ -54,6 +56,11 @@ const InitiateCashoutMutation = GT.Field({
     }
 
     const offer = await CashoutManager.executeCashout(offerId, walletId, domainAccount.id)
+    // Expected, not a fault: this offer was already executed, by an earlier
+    // call or a concurrent duplicate. Return the typed code so the client can
+    // ask for a new offer, and keep it out of the Critical exception stream.
+    if (offer instanceof OfferAlreadyExecuted)
+      return { errors: mapToGqlErrorList(offer), id: null }
     if (offer instanceof Error) {
       recordExceptionInCurrentSpan({
         error: offer,

test/flash/integration/offers/execute-offer.spec.ts

@@ -1,8 +1,12 @@
 import CashoutManager from "@app/offers/CashoutManager"
 
-import { JMDAmount, USDAmount } from "@domain/shared"
+import { Cashout } from "@config"
+import { OfferAlreadyExecuted, OfferNotFound } from "@domain/cache"
+import { JMDAmount, USDAmount, ValidationError } from "@domain/shared"
 import ErpNext, { CashoutId } from "@services/frappe/ErpNext"
+import { BankAccountQueryError } from "@services/frappe/errors"
 import Ibex from "@services/ibex/client"
+import { IbexError, UnexpectedIbexResponse } from "@services/ibex/errors"
 
 import { alice } from "../jest.setup"
 
@@ -109,4 +113,123 @@ describe("Offers", () => {
     // make assertions against ledger
     expect(status).toBeDefined()
   })
+
+  describe("single use per offerId", () => {
+    // dev/config/base-config.yaml sets cashout.skipPayment, which skips the
+    // IBEX leg of ValidOffer.execute entirely. Put it back on the path so the
+    // payInvoice call count below is evidence of what a duplicate would pay.
+    let skipPayment: jest.ReplaceProperty<boolean>
+    beforeEach(() => {
+      skipPayment = jest.replaceProperty(Cashout, "SkipPayment", false)
+    })
+    afterEach(() => {
+      skipPayment.restore()
+    })
+
+    const createOffer = async () => {
+      const offer = await CashoutManager.createOffer(
+        alice.usdWalletD.id,
+        send,
+        "",
+        alice.account.id,
+      )
+      if (offer instanceof Error) throw offer
+      return offer
+    }
+
+    const executeOffer = (id: OfferId) =>
+      CashoutManager.executeCashout(id, alice.usdWalletD.id, alice.account.id)
+
+    it("rejects a second execution of the same offer before ERPNext or IBEX", async () => {
+      const { id } = await createOffer()
+
+      const first = await executeOffer(id)
+      const second = await executeOffer(id)
+
+      // the money-moving side effects ran for the first call only
+      expect(mockedErpNext.draftCashout).toHaveBeenCalledTimes(1)
+      expect(mockedIbex.payInvoice).toHaveBeenCalledTimes(1)
+      expect(first).not.toBeInstanceOf(Error)
+      expect(second).toBeInstanceOf(OfferAlreadyExecuted)
+    })
+
+    it("lets exactly one of two concurrent executions through", async () => {
+      const { id } = await createOffer()
+
+      // Hold both executions inside offer validation, after Storage.get and
+      // before the consume gate, until both have arrived. Both callers have
+      // then read the offer, so only the atomic consume can stop the second.
+      // If a regression stops one execution before validation, the timer lets
+      // the other through, so the arrived count fails the test rather than the
+      // jest timeout.
+      let arrived = 0
+      let releaseBoth: () => void = () => undefined
+      const bothArrived = new Promise<void>((resolve) => {
+        releaseBoth = resolve
+      })
+      const safetyRelease = setTimeout(() => releaseBoth(), 5_000)
+      mockedErpNext.getBankAccountsByCustomer.mockImplementation(async () => {
+        arrived += 1
+        if (arrived === 2) releaseBoth()
+        await bothArrived
+        return [bankAccount]
+      })
+
+      const results = await Promise.all([executeOffer(id), executeOffer(id)]).finally(
+        () => clearTimeout(safetyRelease),
+      )
+
+      expect(arrived).toBe(2)
+      expect(mockedErpNext.draftCashout).toHaveBeenCalledTimes(1)
+      expect(mockedIbex.payInvoice).toHaveBeenCalledTimes(1)
+      expect(results.filter((r) => !(r instanceof Error))).toHaveLength(1)
+      expect(results.filter((r) => r instanceof OfferAlreadyExecuted)).toHaveLength(1)
+    })
+
+    it("keeps an offer whose execution fails validation, so a retry can run it", async () => {
+      const { id } = await createOffer()
+      mockedErpNext.getBankAccountsByCustomer.mockResolvedValueOnce(
+        new BankAccountQueryError("ERPNext unavailable"),
+      )
+
+      const failed = await executeOffer(id)
+      const retried = await executeOffer(id)
+
+      expect(failed).toBeInstanceOf(ValidationError)
+      expect(retried).not.toBeInstanceOf(Error)
+      expect(mockedErpNext.draftCashout).toHaveBeenCalledTimes(1)
+      expect(mockedIbex.payInvoice).toHaveBeenCalledTimes(1)
+    })
+
+    // The offer stays consumed when execute() fails. After an IBEX timeout the
+    // payment may still have gone through, so handing the offer back for a
+    // retry could pay the same cashout twice.
+    it("never executes an offer again after its execution fails at IBEX", async () => {
+      const { id } = await createOffer()
+      mockedIbex.payInvoice.mockResolvedValueOnce(new UnexpectedIbexResponse("timeout"))
+
+      const failed = await executeOffer(id)
+      const retried = await executeOffer(id)
+
+      expect(failed).toBeInstanceOf(IbexError)
+      expect(retried).toBeInstanceOf(OfferAlreadyExecuted)
+      expect(mockedErpNext.draftCashout).toHaveBeenCalledTimes(1)
+      expect(mockedIbex.payInvoice).toHaveBeenCalledTimes(1)
+    })
+
+    // offerId is client input. A consumed offer's tombstone must not come back
+    // as an offer when the client names it, and the call must return a typed
+    // error rather than throw.
+    it("answers an offerId that names a tombstone with OfferNotFound", async () => {
+      const { id } = await createOffer()
+      const executed = await executeOffer(id)
+
+      const forged = await executeOffer(`executed:${id}` as OfferId)
+
+      expect(executed).not.toBeInstanceOf(Error)
+      expect(forged).toBeInstanceOf(OfferNotFound)
+      expect(mockedErpNext.draftCashout).toHaveBeenCalledTimes(1)
+      expect(mockedIbex.payInvoice).toHaveBeenCalledTimes(1)
+    })
+  })
 })