Skip to content
Open
5 changes: 4 additions & 1 deletion apps/api/src/api/controllers/alfredpay.controller.ts
Original file line number Diff line number Diff line change
Expand Up @@ -776,8 +776,11 @@ export class AlfredpayController {
const alfredpayService = AlfredpayApiService.getInstance();
const details = await alfredpayService.getKybBusinessDetails(alfredPayCustomer.alfredPayId);

// submissionId is what lets the caller pick the related persons belonging to the submission it is
// actually filing — a customer that retried can carry several businesses here.
const minimized = details.map(business => ({
relatedPersons: (business.relatedPersons ?? []).map(person => ({ idRelatedPerson: person.idRelatedPerson }))
relatedPersons: (business.relatedPersons ?? []).map(person => ({ idRelatedPerson: person.idRelatedPerson })),
submissionId: business.submissionId
}));

res.json(minimized);
Expand Down
53 changes: 53 additions & 0 deletions apps/api/src/api/middlewares/validators.ts
Original file line number Diff line number Diff line change
Expand Up @@ -17,6 +17,7 @@ import {
PriceProvider,
QuoteError,
RampDirection,
SubmitKybInformationRequest,
SubmitKycInformationRequest,
TokenConfig,
VALID_CRYPTO_CURRENCIES,
Expand Down Expand Up @@ -537,6 +538,58 @@ const countryValidators: Record<string, (body: SubmitKycInformationRequest) => s
}
};

/**
* Alfredpay refuses to finalize a KYB submission missing any of these (`110002 "Invalid field(s)"`),
* and it only says so at `sendKybSubmission` — after the documents have been uploaded. Rejecting the
* incomplete payload here keeps that failure at the point of entry, and keeps the contract enforced
* for callers that are not our own KYB wizard. Mirrors GET …/penny/kybRequirements?country=,
* including its two `requiredIf` branches.
*/
const validateKybQuestionnaire = (body: SubmitKybInformationRequest): string | null => {
const requiredText: Array<keyof SubmitKybInformationRequest> = [
"walletAddresses",
"sourceOfFunds",
"businessActivities",
"accountPurpose"
];
for (const field of requiredText) {
const value = body[field];
if (typeof value !== "string" || !value.trim()) return `${field} is required`;
}

const requiredBooleans: Array<keyof SubmitKybInformationRequest> = [
"transmitsCustomerFunds",
"operatesInSanctionedCountries",
"isRegulatedBusiness"
];
for (const field of requiredBooleans) {
if (typeof body[field] !== "boolean") return `${field} is required`;
}

for (const field of ["expectedMonthlyVolumeUsd", "expectedMonthlyTransactions"] as const) {
const value = body[field];
if (typeof value !== "number" || !Number.isFinite(value) || value < 0) return `${field} must be a non-negative number`;
}
if (!Number.isInteger(body.expectedMonthlyTransactions)) return "expectedMonthlyTransactions must be a whole number";

if (body.transmitsCustomerFunds && typeof body.conductsComplianceScreening !== "boolean") {
return "conductsComplianceScreening is required when transmitting customer funds";
}
if (body.conductsComplianceScreening && !body.complianceScreeningDescription?.trim()) {
return "complianceScreeningDescription is required when conducting compliance screening";
}
return null;
};

export const validateKybSubmission: RequestHandler = (req, res, next) => {
const error = validateKybQuestionnaire(req.body as SubmitKybInformationRequest);
if (error) {
next(new APIError({ errors: [{ message: error }], message: error, status: httpStatus.BAD_REQUEST }));
return;
}
next();
};

export const validateKycSubmission: RequestHandler = (req, res, next) => {
const body = req.body as SubmitKycInformationRequest;
const validator = countryValidators[body.country];
Expand Down
10 changes: 8 additions & 2 deletions apps/api/src/api/routes/v1/alfredpay.route.ts
Original file line number Diff line number Diff line change
Expand Up @@ -4,7 +4,7 @@ import { AlfredpayController } from "../../controllers/alfredpay.controller";
import { validateResultCountry } from "../../middlewares/alfredpay.middleware";
import { requirePartnerOrUserAuth } from "../../middlewares/dualAuth";
import { requireAuth } from "../../middlewares/supabaseAuth";
import { validateKycSubmission } from "../../middlewares/validators";
import { validateKybSubmission, validateKycSubmission } from "../../middlewares/validators";

const router = Router();
const upload = multer({ limits: { fileSize: 5 * 1024 * 1024 }, storage: multer.memoryStorage() });
Expand All @@ -31,7 +31,13 @@ router.post("/submitKycFile", requireAuth, upload.single("file"), validateResult
router.post("/sendKycSubmission", requireAuth, validateResultCountry, AlfredpayController.sendKycSubmission);

// Business API-based KYB
router.post("/submitKybInformation", requireAuth, validateResultCountry, AlfredpayController.submitKybInformation);
router.post(
"/submitKybInformation",
requireAuth,
validateResultCountry,
validateKybSubmission,
AlfredpayController.submitKybInformation
);
router.post("/submitKybFile", requireAuth, upload.single("file"), validateResultCountry, AlfredpayController.submitKybFile);
router.get("/findKybCustomerAndBusiness", requireAuth, validateResultCountry, AlfredpayController.findKybCustomerAndBusiness);
router.post(
Expand Down
Original file line number Diff line number Diff line change
@@ -0,0 +1,22 @@
import { describe, expect, it } from "bun:test";
import QuoteTicket from "../../../../models/quoteTicket.model";
import RampState from "../../../../models/rampState.model";
import { RecoverablePhaseError } from "../../../errors/phase-error";
import { SquidRouterPayPhaseHandler } from "./squid-router-pay-phase-handler";

type BridgeStatusChecker = {
checkBridgeStatus(state: RampState, swapHash: string, quote: QuoteTicket, timeoutMs?: number): Promise<void>;
};

describe("SquidRouterPayPhaseHandler bridge polling timeout", () => {
it("rejects with a recoverable error when its deadline expires", async () => {
const handler = Object.create(SquidRouterPayPhaseHandler.prototype) as BridgeStatusChecker;
const state = { state: {} } as RampState;
const quote = {} as QuoteTicket;

const result = handler.checkBridgeStatus(state, "0xswap", quote, 0);

await expect(result).rejects.toBeInstanceOf(RecoverablePhaseError);
await expect(result).rejects.toThrow("Bridge status check timed out after 0ms");
});
});
Original file line number Diff line number Diff line change
Expand Up @@ -29,17 +29,12 @@ import RampState from "../../../../models/rampState.model";
import { SubsidyToken } from "../../../../models/subsidy.model";
import { BasePhaseHandler } from "../base-phase-handler";
import { getEvmFundingAccount } from "../evm-funding";
import { getSquidRouterPayTimeoutMs } from "../phase-processor-config";

const AXELAR_POLLING_INTERVAL_MS = 10000; // 10 seconds
const SQUIDROUTER_INITIAL_DELAY_MS = 60000; // 60 seconds
const AXL_GAS_SERVICE_EVM = "0x2d5d7d31F671F86C782533cc367F14109a082712";
const BALANCE_POLLING_TIME_MS = 10000;
// NOTE: This timeout is intentionally longer (15 minutes) than the 3–5 minute balance
// checks in other handlers. For SquidRouter/Axelar bridge flows we wait for cross-chain
// settlement and gas payment on the destination chain, which can legitimately take longer
// under network congestion or bridge delays. Reducing this timeout risks premature failure
// of otherwise successful bridge operations.
const EVM_BALANCE_CHECK_TIMEOUT_MS = 15 * 60 * 1000; // 15 minutes
const DEFAULT_SQUIDROUTER_GAS_ESTIMATE = "1600000"; // Estimate used to calculate part of the gas fee for SquidRouter transactions.
/**
* Handler for the squidRouter pay phase. Checks the status of the Axelar bridge and pays on native GLMR fee.
Expand Down Expand Up @@ -128,6 +123,7 @@ export class SquidRouterPayPhaseHandler extends BasePhaseHandler {
}

const toChain = quote.to as EvmNetworks;
const pollingTimeoutMs = getSquidRouterPayTimeoutMs();

let balanceCheckPromise: Promise<Big>;

Expand All @@ -141,7 +137,7 @@ export class SquidRouterPayPhaseHandler extends BasePhaseHandler {
chain: toChain,
intervalMs: BALANCE_POLLING_TIME_MS,
ownerAddress: ephemeralAddress,
timeoutMs: EVM_BALANCE_CHECK_TIMEOUT_MS,
timeoutMs: pollingTimeoutMs,
tokenDetails: outTokenDetails
});
} else {
Expand Down Expand Up @@ -179,7 +175,7 @@ export class SquidRouterPayPhaseHandler extends BasePhaseHandler {

if (balanceError instanceof BalanceCheckError) {
if (balanceError.type === BalanceCheckErrorType.Timeout) {
errorMessage += ` Balance check timed out after ${EVM_BALANCE_CHECK_TIMEOUT_MS}ms.`;
errorMessage += ` Balance check timed out after ${pollingTimeoutMs}ms.`;
} else if (balanceError.type === BalanceCheckErrorType.ReadFailure) {
errorMessage += ` Balance check read failure (unexpected infrastructure issue): ${balanceError.message}.`;
}
Expand All @@ -189,7 +185,7 @@ export class SquidRouterPayPhaseHandler extends BasePhaseHandler {
errorMessage += ` Bridge check error: ${bridgeError instanceof Error ? bridgeError.message : String(bridgeError)}.`;
}

throw new Error(errorMessage);
throw this.createRecoverableError(errorMessage);
}
throw error;
}
Expand All @@ -199,13 +195,23 @@ export class SquidRouterPayPhaseHandler extends BasePhaseHandler {
* Gets the status of the Axelar bridge
* @param txHash The swap (bridgeCall) transaction hash
*/
private async checkBridgeStatus(state: RampState, swapHash: string, quote: QuoteTicket): Promise<void> {
private async checkBridgeStatus(
state: RampState,
swapHash: string,
quote: QuoteTicket,
timeoutMs = getSquidRouterPayTimeoutMs()
): Promise<void> {
let isExecuted = false;
let payTxHash: string | undefined = state.state.squidRouterPayTxHash;
const timeoutAt = Date.now() + timeoutMs;

await new Promise(resolve => setTimeout(resolve, SQUIDROUTER_INITIAL_DELAY_MS));
await new Promise(resolve => setTimeout(resolve, Math.min(SQUIDROUTER_INITIAL_DELAY_MS, timeoutMs)));

while (!isExecuted) {
if (Date.now() >= timeoutAt) {
throw this.createRecoverableError(`SquidRouterPayPhaseHandler: Bridge status check timed out after ${timeoutMs}ms`);
}

try {
const squidRouterStatus = await this.getSquidrouterStatus(swapHash, state, quote);

Expand Down
21 changes: 21 additions & 0 deletions apps/api/src/api/services/phases/phase-processor-config.test.ts
Original file line number Diff line number Diff line change
@@ -0,0 +1,21 @@
import { afterEach, describe, expect, it } from "bun:test";
import { getPhaseProcessorMaxExecutionTimeMs, getSquidRouterPayTimeoutMs } from "./phase-processor-config";

const originalProcessorTimeout = process.env.PHASE_PROCESSOR_MAX_EXECUTION_TIME_MS;

describe("phase processor timeout configuration", () => {
afterEach(() => {
if (originalProcessorTimeout === undefined) {
delete process.env.PHASE_PROCESSOR_MAX_EXECUTION_TIME_MS;
} else {
process.env.PHASE_PROCESSOR_MAX_EXECUTION_TIME_MS = originalProcessorTimeout;
}
});

it("limits SquidRouterPay polling to 80% of the processor timeout", () => {
process.env.PHASE_PROCESSOR_MAX_EXECUTION_TIME_MS = "1000";

expect(getPhaseProcessorMaxExecutionTimeMs()).toBe(1000);
expect(getSquidRouterPayTimeoutMs()).toBe(800);
});
});
16 changes: 16 additions & 0 deletions apps/api/src/api/services/phases/phase-processor-config.ts
Original file line number Diff line number Diff line change
@@ -0,0 +1,16 @@
function positiveIntFromEnv(value: string | undefined, fallback: number): number {
const parsed = value ? parseInt(value, 10) : Number.NaN;
return Number.isFinite(parsed) && parsed > 0 ? parsed : fallback;
}

export function getPhaseProcessorMaxExecutionTimeMs(): number {
return positiveIntFromEnv(process.env.PHASE_PROCESSOR_MAX_EXECUTION_TIME_MS, 600000);
}

export function getSquidRouterPayTimeoutMs(): number {
return Math.floor(getPhaseProcessorMaxExecutionTimeMs() * 0.8);
}

export function getPhaseProcessorRetryDelayMs(): number {
return positiveIntFromEnv(process.env.PHASE_PROCESSOR_RETRY_DELAY_MS, 30000);
}
Original file line number Diff line number Diff line change
Expand Up @@ -29,10 +29,13 @@ const TEST_PHASE = "nablaSwap";
describe("PhaseProcessor execution cancellation", () => {
let polls = 0;
const receivedSignals: (AbortSignal | undefined)[] = [];
const observedLockTimes: number[] = [];

const hangingHandler: PhaseHandler = {
execute: async (_state: RampState, signal?: AbortSignal) => {
execute: async (state: RampState, signal?: AbortSignal) => {
receivedSignals.push(signal);
const persistedState = await RampState.findByPk(state.id);
observedLockTimes.push(new Date(persistedState?.processingLock.lockedAt as unknown as string).getTime());
// Poll forever, like a phase waiting for a payment that never arrives.
await waitUntilTrue(
async () => {
Expand Down Expand Up @@ -86,6 +89,8 @@ describe("PhaseProcessor execution cancellation", () => {
expect(receivedSignals.length).toBeGreaterThanOrEqual(2);
expect(receivedSignals.every(signal => signal instanceof AbortSignal)).toBe(true);
expect(receivedSignals.every(signal => signal?.aborted)).toBe(true);
expect(observedLockTimes.length).toBe(receivedSignals.length);
expect(observedLockTimes.at(-1)).toBeGreaterThan(observedLockTimes[0]);

// Regression: without cancellation the abandoned loops keep polling forever.
const pollsAfterGivingUp = polls;
Expand Down
31 changes: 22 additions & 9 deletions apps/api/src/api/services/phases/phase-processor.ts
Original file line number Diff line number Diff line change
Expand Up @@ -5,15 +5,9 @@ import { config } from "../../../config/vars";
import RampState from "../../../models/rampState.model";
import { APIError } from "../../errors/api-error";
import { PhaseError, RecoverablePhaseError } from "../../errors/phase-error";
import { getPhaseProcessorMaxExecutionTimeMs, getPhaseProcessorRetryDelayMs } from "./phase-processor-config";
import phaseRegistry from "./phase-registry";

// A malformed env override must fall back to the default: parseInt would yield NaN
// and setTimeout(..., NaN) fires immediately, which would time out every phase instantly.
function positiveIntFromEnv(value: string | undefined, fallback: number): number {
const parsed = value ? parseInt(value, 10) : Number.NaN;
return Number.isFinite(parsed) && parsed > 0 ? parsed : fallback;
}

/**
* Process phases for a ramping process
*/
Expand All @@ -22,9 +16,9 @@ export class PhaseProcessor {
private retriesMap = new Map<string, number>();
private readonly MAX_RETRIES = 8;
// Overridable so tests don't wait 10 minutes for the execution timeout.
private readonly MAX_EXECUTION_TIME_MS = positiveIntFromEnv(process.env.PHASE_PROCESSOR_MAX_EXECUTION_TIME_MS, 600000); // 10 minutes
private readonly MAX_EXECUTION_TIME_MS = getPhaseProcessorMaxExecutionTimeMs(); // 10 minutes
// Overridable so tests don't wait 30s between recoverable-error retries.
private readonly DEFAULT_RETRY_DELAY_MS = positiveIntFromEnv(process.env.PHASE_PROCESSOR_RETRY_DELAY_MS, 30000);
private readonly DEFAULT_RETRY_DELAY_MS = getPhaseProcessorRetryDelayMs();
private lockedRamps = new Set<string>();

/**
Expand Down Expand Up @@ -137,6 +131,21 @@ export class PhaseProcessor {
}
}

/**
* Keep the lock fresh while a ramp advances through phases and retries.
*/
private async refreshLock(state: RampState): Promise<void> {
await RampState.update(
{
processingLock: {
locked: true,
lockedAt: new Date()
}
},
{ where: { id: state.id } }
);
}

/**
* Check if the lock has expired. We do this to avoid stale locks that can happen when the service crashes or restarts.
* @param state The current ramp state
Expand Down Expand Up @@ -172,6 +181,10 @@ export class PhaseProcessor {
*/
private async processPhase(state: RampState): Promise<void> {
try {
// The lock is acquired once for the whole processing chain. Refresh it before
// every phase attempt so long phases and retries are not mistaken for crashed work.
await this.refreshLock(state);

const { currentPhase } = state;
logger.info(`Processing phase ${currentPhase} for ramp ${state.id}`);

Expand Down
Loading
Loading