Address the release review findings for the Monerium EUR onramp

apps/api/src/api/services/monerium/monerium.service.test.ts

@@ -1,4 +1,4 @@
-import { afterAll, afterEach, beforeAll, beforeEach, describe, expect, it, mock } from "bun:test";
+import { afterAll, afterEach, beforeAll, beforeEach, describe, expect, it, mock, spyOn } from "bun:test";
 // Load this shared consumer before the module mocks below; Bun does not unregister mock.module
 // replacements, and the API suite may import transfer eligibility after this file.
 import "../recipients/transfer-eligibility.service";
@@ -8,6 +8,7 @@ import "./wallet";
 // module mocks are process-wide and would poison later test files (e.g. integration tests
 // that need the real sequelize instance and models).
 import * as databaseNamespace from "../../../config/database";
+import logger from "../../../config/logger";
 import * as kycCaseNamespace from "../../../models/kycCase.model";
 import * as providerCustomerNamespace from "../../../models/providerCustomer.model";
 import * as customerEntityNamespace from "../customer-entity.service";
@@ -368,6 +369,46 @@ describe("Monerium OAuth", () => {
     expect(tokenCalls).toBe(2);
   });
 
+  it("keeps the session and logs a warning when the refresh grant fails upstream", async () => {
+    let tokenCalls = 0;
+    globalThis.fetch = mock(async (input: string | URL | Request) => {
+      const url = String(input);
+      if (url.endsWith("/auth/token")) {
+        tokenCalls += 1;
+        return tokenCalls === 1
+          ? jsonResponse({ access_token: "old-access", expires_in: 1, refresh_token: "old-refresh" })
+          : new Response("upstream unavailable", { status: 503 });
+      }
+      if (url.endsWith("/auth/context")) {
+        return jsonResponse({
+          email: "owner@example.com",
+          profiles: [{ id: "profile-a", kind: "personal" }],
+          userId: "monerium-user-a"
+        });
+      }
+      return jsonResponse({ id: "profile-a", kind: "personal", state: "pending" });
+    }) as unknown as typeof fetch;
+    const warn = spyOn(logger, "warn").mockImplementation(() => logger);
+
+    try {
+      const { authorizationUrl } = await service.startMoneriumOAuth("owner", "owner@example.com", "individual");
+      const state = new URL(authorizationUrl).searchParams.get("state") as string;
+      await service.completeMoneriumOAuth("owner", "authorization-code", state);
+
+      await expect(service.getMoneriumStatus("owner", "individual")).rejects.toMatchObject({ status: 502 });
+      // The session survives an outage: the next call retries the refresh instead of demanding reauthentication.
+      await expect(service.getMoneriumStatus("owner", "individual")).rejects.toMatchObject({ status: 502 });
+      expect(tokenCalls).toBe(3);
+      expect(warn).toHaveBeenCalledTimes(2);
+      const logged = JSON.stringify(warn.mock.calls);
+      expect(logged).toContain("token refresh");
+      expect(logged).toContain("503");
+      expect(logged).not.toContain("old-refresh");
+    } finally {
+      warn.mockRestore();
+    }
+  });
+
   it("asks for reauthentication when Monerium revokes an otherwise unexpired access token", async () => {
     let profileReads = 0;
     globalThis.fetch = mock(async (input: string | URL | Request) => {

apps/api/src/api/services/monerium/monerium.service.ts

@@ -2,6 +2,7 @@ import crypto from "crypto";
 import httpStatus from "http-status";
 import NodeCache from "node-cache";
 import sequelize from "../../../config/database";
+import logger from "../../../config/logger";
 import { config } from "../../../config/vars";
 import KycCase from "../../../models/kycCase.model";
 import ProviderCustomer, {
@@ -234,6 +235,17 @@ async function getValidCredentials(customerEntityId: string, customerType: Provi
           type: MONERIUM_REAUTHENTICATION_REQUIRED
         });
       }
+      // An upstream failure keeps the session for the next attempt, but must not stay invisible: a
+      // sustained outage on a truly expired token would otherwise read as an endless 502.
+      const reason =
+        error instanceof MoneriumUpstreamError
+          ? `upstream status ${error.upstreamStatus ?? "unknown"}`
+          : error instanceof Error
+            ? error.message
+            : String(error);
+      logger.warn(
+        `MoneriumOAuth: token refresh for ${customerType} entity ${customerEntityId} failed (${reason}); keeping the session for retry`
+      );
       throw error;
     }
   );

apps/api/src/api/services/monerium/wallet.test.ts

@@ -165,6 +165,33 @@ describe("linkMoneriumWallet", () => {
     expect(error.message).toContain("Contract wallets are not supported");
     expect(monerium.linkAddress).not.toHaveBeenCalled();
   });
+
+  it("holds the profile lock while linking the address and requesting the IBAN", async () => {
+    const monerium = client();
+    const order: string[] = [];
+    monerium.linkAddress.mockImplementation(async () => {
+      order.push("link");
+      return { httpStatus: 201 as const };
+    });
+    monerium.requestIban.mockImplementation(async () => {
+      order.push("iban");
+      return { httpStatus: 202 as const };
+    });
+    const runWithProfileLock = async <T>(profileId: string, work: (transaction: Transaction) => Promise<T>): Promise<T> => {
+      order.push(`lock:${profileId}`);
+      const result = await work(undefined as unknown as Transaction);
+      order.push("unlock");
+      return result;
+    };
+
+    await linkMoneriumWallet(
+      "user-1",
+      { address: OWNER.address, chain: "polygon", signature: await ownerSignature() },
+      deps(monerium, { runWithProfileLock })
+    );
+
+    expect(order).toEqual([`lock:${PROFILE_ID}`, "link", "iban", "unlock"]);
+  });
 });
 
 describe("moveMoneriumIban", () => {

apps/api/src/api/services/monerium/wallet.ts

@@ -146,33 +146,43 @@ export async function linkMoneriumWallet(
     });
   }
 
+  const signature = input.signature;
   const identity = await dependencies.resolveIdentity(userId, undefined, input.customerType);
-  const before = await readDestinations(identity, chain);
-  if (!before.addresses.some(entry => sameAddress(entry.address, address))) {
-    await identity.client.linkAddress({
-      address,
-      chain,
-      message: MONERIUM_ADDRESS_OWNERSHIP_MESSAGE,
-      profile: identity.profileId,
-      signature: input.signature
-    });
-    logger.info(`MoneriumWallet: linked ${address} on ${chain} through the ${identity.source} app`);
-  }
-
-  if (before.ibans.some(entry => entry.chain === chain && sameAddress(entry.address, address))) {
-    return { address, chain, iban: "provisioned" };
-  }
-  if (before.ibans.length > 0) return { address, chain, iban: "elsewhere" };
+  // Serialise with IBAN moves and issue registration on the same profile: both read the same
+  // address and IBAN lists this call mutates.
+  return (dependencies.runWithProfileLock ?? runWithProfileLock)(
+    identity.profileId,
+    async (): Promise<MoneriumWalletLinkResult> => {
+      const before = await readDestinations(identity, chain);
+      if (!before.addresses.some(entry => sameAddress(entry.address, address))) {
+        await identity.client.linkAddress({
+          address,
+          chain,
+          message: MONERIUM_ADDRESS_OWNERSHIP_MESSAGE,
+          profile: identity.profileId,
+          signature
+        });
+        logger.info(`MoneriumWallet: linked ${address} on ${chain} through the ${identity.source} app`);
+      }
 
-  try {
-    await identity.client.requestIban({ address, chain });
-  } catch (error) {
-    // Monerium keeps one IBAN per profile and answers 400 when one is already requested.
-    if (!(error instanceof MoneriumApiError && error.status === 400)) throw error;
-    logger.warn(`MoneriumWallet: POST /ibans answered 400 for ${address} on ${chain}; assuming an IBAN is already requested`);
-  }
-  logger.info(`MoneriumWallet: requested an IBAN for ${address} on ${chain}`);
-  return { address, chain, iban: "pending" };
+      if (before.ibans.some(entry => entry.chain === chain && sameAddress(entry.address, address))) {
+        return { address, chain, iban: "provisioned" };
+      }
+      if (before.ibans.length > 0) return { address, chain, iban: "elsewhere" };
+
+      try {
+        await identity.client.requestIban({ address, chain });
+      } catch (error) {
+        // Monerium keeps one IBAN per profile and answers 400 when one is already requested.
+        if (!(error instanceof MoneriumApiError && error.status === 400)) throw error;
+        logger.warn(
+          `MoneriumWallet: POST /ibans answered 400 for ${address} on ${chain}; assuming an IBAN is already requested`
+        );
+      }
+      logger.info(`MoneriumWallet: requested an IBAN for ${address} on ${chain}`);
+      return { address, chain, iban: "pending" };
+    }
+  );
 }
 
 /** Moves the profile's single IBAN to an already-linked address. Only ever called on the owner's explicit request. */

apps/api/src/api/services/phases/blocks/__tests__/monerium-issue.executor.test.ts

@@ -3,6 +3,7 @@ import { BalanceCheckError, BalanceCheckErrorType, Networks } from "@vortexfi/sh
 import Big from "big.js";
 import { MoneriumOnrampMintExecutor } from "../phases/monerium-issue/execution";
 import { MoneriumIssue } from "../phases/monerium-issue";
+import { MONERIUM_ISSUE_NETWORKS } from "../phases/monerium-issue/simulation";
 
 const OWNER = "0x1212121212121212121212121212121212121212";
 
@@ -40,7 +41,7 @@ describe("MoneriumOnrampMintExecutor", () => {
 
     expect(await executor.executePhase(rampState)).toBe(rampState);
     expect(waitForBalance).toHaveBeenCalledWith(
-      "0x18ec0A6E18E5bc3784fDd3a3634b31245ab704F6",
+      MONERIUM_ISSUE_NETWORKS[Networks.Polygon].eureAddress,
       OWNER,
       "598",
       5000,

apps/api/src/api/services/phases/blocks/__tests__/uniswap-v3-fixed-swap.test.ts

@@ -28,13 +28,31 @@ import {
 
 const financialOperationReal = { ...financialOperationNamespace };
 const operationAttempts: string[] = [];
+/** Journal rows by attempt class; a seeded `submitted` row must be reconciled, never performed again. */
+const operationRows = new Map<string, "submitted" | "confirmed">();
 
 mock.module("../core/financial-operation", () => ({
   ...financialOperationReal,
   requireFinancialFlowIdentity: () => ({ id: "test-flow", version: 1 }),
-  runFinancialOperation: async ({ attemptClass, perform }: { attemptClass: string; perform(key: string): Promise<unknown> }) => {
+  runFinancialOperation: async ({
+    attemptClass,
+    perform,
+    reconcile
+  }: {
+    attemptClass: string;
+    perform(key: string): Promise<unknown>;
+    reconcile?: (operation: unknown) => Promise<unknown>;
+  }) => {
     operationAttempts.push(attemptClass);
-    return perform(`test-${attemptClass}`);
+    if (operationRows.get(attemptClass) === "submitted") {
+      const reconciled = reconcile ? await reconcile({}) : null;
+      if (reconciled === null) throw new Error(`operation ${attemptClass} has submitted outcome`);
+      operationRows.set(attemptClass, "confirmed");
+      return reconciled;
+    }
+    const result = await perform(`test-${attemptClass}`);
+    operationRows.set(attemptClass, "confirmed");
+    return result;
   }
 }));
 
@@ -431,6 +449,7 @@ describe("fixed Polygon Uniswap V3 execution failure branches", () => {
 
   async function runSwap(overrides: Partial<NonNullable<Deps>>, now = Date.now()) {
     operationAttempts.length = 0;
+    operationRows.clear();
     const { approval, simulated, state, swap } = await prepared(now);
     const signedApproval = await sign(approval);
     const signedSwap = await sign(swap);
@@ -468,6 +487,52 @@ describe("fixed Polygon Uniswap V3 execution failure branches", () => {
     expect(operationAttempts).toEqual([]);
   });
 
+  /** Crash after the swap was mined: allowance consumed, EURe gone, but the deterministic hash has a receipt. */
+  function settledSwapDependencies() {
+    const sendRawTransaction = mock(async (): Promise<`0x${string}`> => {
+      throw new Error("must not resend a settled swap");
+    });
+    const simulateTransaction = mock(async () => {
+      throw new Error("must not simulate a settled swap");
+    });
+    const getReceipt = mock(async (): Promise<{ status: "success" }> => ({ status: "success" }));
+    return {
+      getAllowance: async () => 0n,
+      getBalance: async (token: `0x${string}`) => (token === POLYGON_EURE ? 0n : 116_000_000n),
+      getReceipt,
+      sendRawTransaction,
+      simulateTransaction
+    };
+  }
+
+  it("finishes a resumed swap whose transaction already settled instead of pausing", async () => {
+    const dependencies = settledSwapDependencies();
+    const error = await runSwap(dependencies);
+    expect(error).toBeNull();
+    // One lookup decides the resume and also settles the journal: no second read that could fail
+    // transiently, no resend of the mined transaction.
+    expect(dependencies.getReceipt).toHaveBeenCalledTimes(1);
+    expect(dependencies.sendRawTransaction).not.toHaveBeenCalled();
+    expect(dependencies.simulateTransaction).not.toHaveBeenCalled();
+    expect(operationAttempts).toEqual(["uniswap-presigned-broadcast"]);
+    expect(operationRows.get("uniswap-presigned-broadcast")).toBe("confirmed");
+  });
+
+  it("confirms a journal row left submitted by the crash from the receipt it already read", async () => {
+    const dependencies = settledSwapDependencies();
+    // runSwap clears the rows, so seed the submitted row from the receipt read, which precedes the journal call.
+    const getReceipt = mock(async (): Promise<{ status: "success" }> => {
+      operationRows.set("uniswap-presigned-broadcast", "submitted");
+      return { status: "success" };
+    });
+    const error = await runSwap({ ...dependencies, getReceipt });
+    expect(error).toBeNull();
+    expect(getReceipt).toHaveBeenCalledTimes(1);
+    expect(dependencies.sendRawTransaction).not.toHaveBeenCalled();
+    expect(operationAttempts).toEqual(["uniswap-presigned-broadcast"]);
+    expect(operationRows.get("uniswap-presigned-broadcast")).toBe("confirmed");
+  });
+
   it("retries later when the live quote moved below the soft minimum", async () => {
     const error = await runSwap({ quote: async () => 113_679_999n });
     expect(error).toBeInstanceOf(RecoverablePhaseError);