120 lines
3.2 KiB
TypeScript
120 lines
3.2 KiB
TypeScript
import type { PrismaClient } from '@prisma/client';
|
|
import { logger } from '@/infrastructure/monitoring/logger';
|
|
import {
|
|
createGatewayRailAdapter,
|
|
isGatewayRailAdapterId,
|
|
} from '@/core/gateway/adapters/gateway-adapter-registry';
|
|
|
|
export interface GatewayOutboxWorkerOptions {
|
|
batchSize?: number;
|
|
/** After this many send attempts, row stays FAILED (DLQ). */
|
|
maxAttempts?: number;
|
|
}
|
|
|
|
/**
|
|
* Processes `gateway_outbox` rows (PENDING → adapter.send → SENT or PENDING retry / FAILED).
|
|
* Uses compare-and-swap on `(id, status, sendAttempts)` to limit double-sends under multiple workers.
|
|
*/
|
|
export class GatewayOutboxWorker {
|
|
private readonly batchSize: number;
|
|
private readonly maxAttempts: number;
|
|
|
|
constructor(
|
|
private readonly prisma: PrismaClient,
|
|
opts: GatewayOutboxWorkerOptions = {},
|
|
) {
|
|
this.batchSize = opts.batchSize ?? parseInt(process.env.GATEWAY_OUTBOX_BATCH_SIZE || '25', 10);
|
|
this.maxAttempts =
|
|
opts.maxAttempts ?? parseInt(process.env.GATEWAY_OUTBOX_MAX_ATTEMPTS || '10', 10);
|
|
}
|
|
|
|
async runOnce(): Promise<number> {
|
|
const candidates = await this.prisma.gateway_outbox.findMany({
|
|
where: {
|
|
status: 'PENDING',
|
|
sendAttempts: { lt: this.maxAttempts },
|
|
},
|
|
orderBy: { createdAt: 'asc' },
|
|
take: this.batchSize,
|
|
});
|
|
|
|
let processed = 0;
|
|
for (const row of candidates) {
|
|
const claimed = await this.prisma.gateway_outbox.updateMany({
|
|
where: {
|
|
id: row.id,
|
|
status: 'PENDING',
|
|
sendAttempts: row.sendAttempts,
|
|
},
|
|
data: {
|
|
sendAttempts: row.sendAttempts + 1,
|
|
lastAttemptAt: new Date(),
|
|
},
|
|
});
|
|
if (claimed.count !== 1) {
|
|
continue;
|
|
}
|
|
|
|
const attemptsAfterClaim = row.sendAttempts + 1;
|
|
try {
|
|
await this.dispatchSend(row.id, row.txnId, row.adapterId, row.payloadHash, attemptsAfterClaim);
|
|
processed += 1;
|
|
} catch (err) {
|
|
logger.error('GatewayOutboxWorker: dispatch error', {
|
|
id: row.id,
|
|
error: err instanceof Error ? err.message : String(err),
|
|
});
|
|
await this.prisma.gateway_outbox.update({
|
|
where: { id: row.id },
|
|
data: {
|
|
status: attemptsAfterClaim >= this.maxAttempts ? 'FAILED' : 'PENDING',
|
|
},
|
|
});
|
|
processed += 1;
|
|
}
|
|
}
|
|
return processed;
|
|
}
|
|
|
|
private async dispatchSend(
|
|
id: string,
|
|
txnId: string,
|
|
adapterId: string,
|
|
payloadHash: string,
|
|
attemptsAfterClaim: number,
|
|
): Promise<void> {
|
|
if (!isGatewayRailAdapterId(adapterId)) {
|
|
logger.warn('GatewayOutboxWorker: unknown adapter', { id, adapterId });
|
|
await this.prisma.gateway_outbox.update({
|
|
where: { id },
|
|
data: { status: attemptsAfterClaim >= this.maxAttempts ? 'FAILED' : 'PENDING' },
|
|
});
|
|
return;
|
|
}
|
|
|
|
const adapter = createGatewayRailAdapter(adapterId)!;
|
|
await adapter.initialize({}, undefined);
|
|
const result = await adapter.send({
|
|
txnId,
|
|
payloadHash,
|
|
envelope: { source: 'gateway_outbox', scaffold: true },
|
|
});
|
|
|
|
if (result.status === 'SENT') {
|
|
await this.prisma.gateway_outbox.update({
|
|
where: { id },
|
|
data: { status: 'SENT', lastAttemptAt: new Date() },
|
|
});
|
|
return;
|
|
}
|
|
|
|
await this.prisma.gateway_outbox.update({
|
|
where: { id },
|
|
data: {
|
|
status: attemptsAfterClaim >= this.maxAttempts ? 'FAILED' : 'PENDING',
|
|
lastAttemptAt: new Date(),
|
|
},
|
|
});
|
|
}
|
|
}
|