perf: batch gateway submits and isolate callbacks
This commit is contained in:
hectorzhao
@@ -22,18 +22,25 @@ export class PhoneRoutingLookupService {
|
||||
}
|
||||
|
||||
async identifyProvince(phoneNumber: string) {
|
||||
const prefixes = phonePrefixes(phoneNumber);
|
||||
if (prefixes.length === 0) return null;
|
||||
return (await this.identifyProvinces([phoneNumber])).get(phoneNumber) ?? null;
|
||||
}
|
||||
|
||||
async identifyProvinces(phoneNumbers: string[]) {
|
||||
const uniquePhones = [...new Set(phoneNumbers)];
|
||||
const prefixesByPhone = new Map(uniquePhones.map((phone) => [phone, phonePrefixes(phone)]));
|
||||
const prefixes = [...new Set([...prefixesByPhone.values()].flat())];
|
||||
if (prefixes.length === 0) return new Map(uniquePhones.map((phone) => [phone, null]));
|
||||
const segments = await this.prisma.phoneSegment.findMany({
|
||||
where: { prefix: { in: prefixes } },
|
||||
select: { prefix: true, province: true },
|
||||
});
|
||||
const provinceByPrefix = new Map(segments.map((segment) => [segment.prefix, segment.province]));
|
||||
for (const prefix of prefixes) {
|
||||
const province = provinceByPrefix.get(prefix);
|
||||
if (province) return province;
|
||||
}
|
||||
return null;
|
||||
return new Map(uniquePhones.map((phone) => {
|
||||
const province = (prefixesByPhone.get(phone) ?? [])
|
||||
.map((prefix) => provinceByPrefix.get(prefix))
|
||||
.find((value): value is string => Boolean(value)) ?? null;
|
||||
return [phone, province];
|
||||
}));
|
||||
}
|
||||
|
||||
invalidateCarrierRules() {
|
||||
|
||||
@@ -0,0 +1,31 @@
|
||||
import { Module } from '@nestjs/common';
|
||||
import { ConfigModule } from '@nestjs/config';
|
||||
import { BillingService } from './billing/billing.service';
|
||||
import { PhoneRoutingLookupService } from './dictionaries/phone-routing-lookup.service';
|
||||
import { MetricsModule } from './metrics/metrics.module';
|
||||
import { OpenApiService } from './open-api/open-api.service';
|
||||
import { PrismaModule } from './prisma/prisma.module';
|
||||
import { ProtocolLogsModule } from './protocol-logs/protocol-logs.module';
|
||||
import { PhoneFrequencyService } from './risk-review/phone-frequency.service';
|
||||
import { RiskReviewService } from './risk-review/risk-review.service';
|
||||
import { GatewayCallbackController } from './send-chain/gateway-callback.controller';
|
||||
import { SendChainService } from './send-chain/send-chain.service';
|
||||
|
||||
@Module({
|
||||
imports: [
|
||||
ConfigModule.forRoot({ isGlobal: true, envFilePath: ['.env.local', '.env'] }),
|
||||
PrismaModule,
|
||||
MetricsModule,
|
||||
ProtocolLogsModule,
|
||||
],
|
||||
controllers: [GatewayCallbackController],
|
||||
providers: [
|
||||
BillingService,
|
||||
RiskReviewService,
|
||||
PhoneFrequencyService,
|
||||
PhoneRoutingLookupService,
|
||||
SendChainService,
|
||||
OpenApiService,
|
||||
],
|
||||
})
|
||||
export class GatewayCallbackModule {}
|
||||
@@ -0,0 +1,44 @@
|
||||
import 'reflect-metadata';
|
||||
import { NestFactory } from '@nestjs/core';
|
||||
import { createServer } from 'node:http';
|
||||
import type { NestExpressApplication } from '@nestjs/platform-express';
|
||||
import { GatewayCallbackModule } from './gateway-callback.module';
|
||||
import { configureHttpBodyParsers } from './http-body-limits';
|
||||
import { MetricsService } from './metrics/metrics.service';
|
||||
|
||||
Object.defineProperty(BigInt.prototype, 'toJSON', {
|
||||
configurable: true,
|
||||
value(this: bigint) {
|
||||
const result = Number(this);
|
||||
if (!Number.isSafeInteger(result)) throw new RangeError('金额超过 JavaScript 安全整数范围');
|
||||
return result;
|
||||
},
|
||||
});
|
||||
|
||||
async function bootstrap() {
|
||||
if (process.env.CMPP_PROCESS_ROLE !== 'callback') {
|
||||
throw new Error('gateway-callback requires CMPP_PROCESS_ROLE=callback');
|
||||
}
|
||||
const app = await NestFactory.create<NestExpressApplication>(GatewayCallbackModule, { rawBody: true, bodyParser: false });
|
||||
app.setGlobalPrefix('api');
|
||||
configureHttpBodyParsers(app);
|
||||
app.enableShutdownHooks();
|
||||
const host = process.env.API_CALLBACK_HOST?.trim() || '127.0.0.1';
|
||||
const port = Number(process.env.API_CALLBACK_PORT ?? 3001);
|
||||
await app.listen(port, host);
|
||||
|
||||
const metrics = app.get(MetricsService);
|
||||
const metricsHost = process.env.API_CALLBACK_METRICS_HOST?.trim() || '127.0.0.1';
|
||||
const metricsPort = Number(process.env.API_CALLBACK_METRICS_PORT ?? 9468);
|
||||
const metricsServer = createServer((request, response) => {
|
||||
if (request.method !== 'GET' || request.url !== '/metrics') return void response.writeHead(404).end();
|
||||
response.writeHead(200, { 'Content-Type': 'text/plain; version=0.0.4; charset=utf-8', 'Cache-Control': 'no-store' });
|
||||
response.end(metrics.render());
|
||||
});
|
||||
await new Promise<void>((resolve, reject) => {
|
||||
metricsServer.once('error', reject);
|
||||
metricsServer.listen(metricsPort, metricsHost, resolve);
|
||||
});
|
||||
}
|
||||
|
||||
void bootstrap();
|
||||
@@ -7,13 +7,29 @@ describe('MetricsService', () => {
|
||||
service.finishRequest(startedAt, 'GET', '/api/admin/tenants/:id', 200);
|
||||
const inboundStartedAt = service.beginCmppInboundStage();
|
||||
service.finishCmppInboundStage(inboundStartedAt, 'application_lookup', 'success');
|
||||
const sendStartedAt = service.beginSendWorkerStage();
|
||||
service.finishSendWorkerStage(sendStartedAt, 'route_lookup', 'success');
|
||||
service.setSendWorkerSlots(20, 3);
|
||||
service.setSendWorkerQueueJobs('waiting', 12);
|
||||
service.recordSendWorkerResult('completed');
|
||||
service.setSendWorkerDatabasePool('max', 8);
|
||||
service.setSendWorkerDatabasePool('waiting', 2);
|
||||
const output = service.render();
|
||||
|
||||
expect(output).toContain('cmpp_api_process_resident_memory_bytes');
|
||||
expect(output).toContain('cmpp_api_http_requests_total{method="GET",route="/api/admin/tenants/:id",status="200"} 1');
|
||||
expect(output).toContain('cmpp_api_http_request_duration_seconds_bucket');
|
||||
expect(output).toContain('cmpp_api_cmpp_inbound_stage_duration_seconds_count{stage="application_lookup",result="success"} 1');
|
||||
expect(output).toContain('cmpp_worker_send_stage_duration_seconds_count{stage="route_lookup",result="success"} 1');
|
||||
expect(output).toContain('cmpp_worker_send_slots{state="configured"} 20');
|
||||
expect(output).toContain('cmpp_worker_send_slots{state="in_flight"} 3');
|
||||
expect(output).toContain('cmpp_worker_send_queue_jobs{state="waiting"} 12');
|
||||
expect(output).toContain('cmpp_worker_send_jobs_total{result="completed"} 1');
|
||||
expect(output).toContain('cmpp_worker_database_pool_connections{state="max"} 8');
|
||||
expect(output).toContain('cmpp_worker_database_pool_connections{state="waiting"} 2');
|
||||
expect(output).not.toContain('phone_number');
|
||||
expect(output).not.toContain('tenant_id');
|
||||
expect(output).not.toContain('channel_id');
|
||||
service.onModuleDestroy();
|
||||
});
|
||||
});
|
||||
|
||||
@@ -3,6 +3,7 @@ import { monitorEventLoopDelay } from 'node:perf_hooks';
|
||||
|
||||
const HTTP_DURATION_BUCKETS = [0.05, 0.1, 0.25, 0.5, 1, 3, 10] as const;
|
||||
const CMPP_INBOUND_DURATION_BUCKETS = [0.005, 0.01, 0.025, 0.05, 0.1, 0.25, 0.5, 1, 3, 10] as const;
|
||||
const SEND_WORKER_DURATION_BUCKETS = [0.005, 0.01, 0.025, 0.05, 0.1, 0.25, 0.5, 1, 3, 10] as const;
|
||||
|
||||
export type CmppInboundStage =
|
||||
| 'application_lookup'
|
||||
@@ -25,6 +26,22 @@ export type CmppInboundStage =
|
||||
|
||||
export type CmppInboundStageResult = 'success' | 'error';
|
||||
|
||||
export type SendWorkerStage =
|
||||
| 'message_load'
|
||||
| 'phone_routing'
|
||||
| 'route_lookup'
|
||||
| 'signature_candidates'
|
||||
| 'signature_final_check'
|
||||
| 'rate_limit'
|
||||
| 'submit_transaction'
|
||||
| 'gateway_bullmq_publish'
|
||||
| 'gateway_stream_publish'
|
||||
| 'task_progress'
|
||||
| 'total';
|
||||
|
||||
export type SendWorkerStageResult = 'success' | 'error' | 'skipped';
|
||||
export type SendWorkerQueueState = 'waiting' | 'active' | 'completed' | 'failed' | 'delayed' | 'prioritized';
|
||||
|
||||
type HttpMetric = {
|
||||
count: number;
|
||||
durationSum: number;
|
||||
@@ -48,6 +65,12 @@ export class MetricsService implements OnModuleDestroy {
|
||||
private readonly eventLoopDelay = monitorEventLoopDelay({ resolution: 20 });
|
||||
private readonly http = new Map<string, HttpMetric>();
|
||||
private readonly cmppInbound = new Map<string, HttpMetric>();
|
||||
private readonly sendWorkerStages = new Map<string, HttpMetric>();
|
||||
private readonly sendWorkerQueueJobs = new Map<SendWorkerQueueState, number>();
|
||||
private readonly sendWorkerResults = new Map<string, number>();
|
||||
private sendWorkerConfiguredSlots = 0;
|
||||
private sendWorkerInFlightSlots = 0;
|
||||
private readonly sendWorkerDatabasePool = new Map<'max' | 'total' | 'idle' | 'waiting', number>();
|
||||
private inFlight = 0;
|
||||
private inboundWorkflowPending = 0;
|
||||
private inboundWorkflowProcessing = 0;
|
||||
@@ -101,6 +124,43 @@ export class MetricsService implements OnModuleDestroy {
|
||||
this.cmppInbound.set(key, metric);
|
||||
}
|
||||
|
||||
beginSendWorkerStage() {
|
||||
return process.hrtime.bigint();
|
||||
}
|
||||
|
||||
finishSendWorkerStage(startedAt: bigint, stage: SendWorkerStage, result: SendWorkerStageResult) {
|
||||
const key = `${stage}\u0000${result}`;
|
||||
const metric = this.sendWorkerStages.get(key) ?? {
|
||||
count: 0,
|
||||
durationSum: 0,
|
||||
buckets: SEND_WORKER_DURATION_BUCKETS.map(() => 0),
|
||||
};
|
||||
const durationSeconds = Number(process.hrtime.bigint() - startedAt) / 1_000_000_000;
|
||||
metric.count += 1;
|
||||
metric.durationSum += durationSeconds;
|
||||
SEND_WORKER_DURATION_BUCKETS.forEach((bucket, index) => {
|
||||
if (durationSeconds <= bucket) metric.buckets[index] += 1;
|
||||
});
|
||||
this.sendWorkerStages.set(key, metric);
|
||||
}
|
||||
|
||||
setSendWorkerSlots(configured: number, inFlight: number) {
|
||||
this.sendWorkerConfiguredSlots = Math.max(0, configured);
|
||||
this.sendWorkerInFlightSlots = Math.max(0, inFlight);
|
||||
}
|
||||
|
||||
setSendWorkerQueueJobs(state: SendWorkerQueueState, count: number) {
|
||||
this.sendWorkerQueueJobs.set(state, Math.max(0, count));
|
||||
}
|
||||
|
||||
recordSendWorkerResult(result: 'completed' | 'failed' | 'skipped') {
|
||||
this.sendWorkerResults.set(result, (this.sendWorkerResults.get(result) ?? 0) + 1);
|
||||
}
|
||||
|
||||
setSendWorkerDatabasePool(state: 'max' | 'total' | 'idle' | 'waiting', count: number) {
|
||||
this.sendWorkerDatabasePool.set(state, Math.max(0, count));
|
||||
}
|
||||
|
||||
setInboundWorkflowState(pending: number, processing: number, oldestPendingAgeSeconds: number) {
|
||||
this.inboundWorkflowPending = Math.max(0, pending);
|
||||
this.inboundWorkflowProcessing = Math.max(0, processing);
|
||||
@@ -144,6 +204,18 @@ export class MetricsService implements OnModuleDestroy {
|
||||
'# TYPE cmpp_api_http_request_duration_seconds histogram',
|
||||
'# HELP cmpp_api_cmpp_inbound_stage_duration_seconds CMPP inbound processing duration by bounded stage and result.',
|
||||
'# TYPE cmpp_api_cmpp_inbound_stage_duration_seconds histogram',
|
||||
'# HELP cmpp_worker_send_stage_duration_seconds Send worker processing duration by bounded stage and result.',
|
||||
'# TYPE cmpp_worker_send_stage_duration_seconds histogram',
|
||||
'# HELP cmpp_worker_send_queue_jobs BullMQ send jobs by queue state.',
|
||||
'# TYPE cmpp_worker_send_queue_jobs gauge',
|
||||
'# HELP cmpp_worker_send_slots Send worker concurrency slots by state.',
|
||||
'# TYPE cmpp_worker_send_slots gauge',
|
||||
metricLine('cmpp_worker_send_slots', this.sendWorkerConfiguredSlots, { state: 'configured' }),
|
||||
metricLine('cmpp_worker_send_slots', this.sendWorkerInFlightSlots, { state: 'in_flight' }),
|
||||
'# HELP cmpp_worker_send_jobs_total Send worker processing outcomes.',
|
||||
'# TYPE cmpp_worker_send_jobs_total counter',
|
||||
'# HELP cmpp_worker_database_pool_connections Worker PostgreSQL client pool slots by state.',
|
||||
'# TYPE cmpp_worker_database_pool_connections gauge',
|
||||
'# HELP cmpp_worker_inbound_workflow_items Current durable CMPP inbound workflow items by state.',
|
||||
'# TYPE cmpp_worker_inbound_workflow_items gauge',
|
||||
metricLine('cmpp_worker_inbound_workflow_items', this.inboundWorkflowPending, { state: 'pending' }),
|
||||
@@ -179,6 +251,25 @@ export class MetricsService implements OnModuleDestroy {
|
||||
lines.push(metricLine('cmpp_api_cmpp_inbound_stage_duration_seconds_sum', metric.durationSum, labels));
|
||||
lines.push(metricLine('cmpp_api_cmpp_inbound_stage_duration_seconds_count', metric.count, labels));
|
||||
}
|
||||
for (const [key, metric] of this.sendWorkerStages) {
|
||||
const [stage, result] = key.split('\u0000');
|
||||
const labels = { stage, result };
|
||||
SEND_WORKER_DURATION_BUCKETS.forEach((bucket, index) => {
|
||||
lines.push(metricLine('cmpp_worker_send_stage_duration_seconds_bucket', metric.buckets[index], { ...labels, le: String(bucket) }));
|
||||
});
|
||||
lines.push(metricLine('cmpp_worker_send_stage_duration_seconds_bucket', metric.count, { ...labels, le: '+Inf' }));
|
||||
lines.push(metricLine('cmpp_worker_send_stage_duration_seconds_sum', metric.durationSum, labels));
|
||||
lines.push(metricLine('cmpp_worker_send_stage_duration_seconds_count', metric.count, labels));
|
||||
}
|
||||
for (const [state, count] of this.sendWorkerQueueJobs) {
|
||||
lines.push(metricLine('cmpp_worker_send_queue_jobs', count, { state }));
|
||||
}
|
||||
for (const [result, count] of this.sendWorkerResults) {
|
||||
lines.push(metricLine('cmpp_worker_send_jobs_total', count, { result }));
|
||||
}
|
||||
for (const [state, count] of this.sendWorkerDatabasePool) {
|
||||
lines.push(metricLine('cmpp_worker_database_pool_connections', count, { state }));
|
||||
}
|
||||
for (const [result, count] of this.inboundWorkflowResults) {
|
||||
lines.push(metricLine('cmpp_worker_inbound_workflow_results_total', count, { result }));
|
||||
}
|
||||
|
||||
@@ -55,6 +55,10 @@ export class OpenApiService implements OnModuleInit, OnModuleDestroy {
|
||||
onModuleInit() {
|
||||
const connection = bullmqConnection();
|
||||
this.queue = new Queue(WEBHOOK_QUEUE, { connection });
|
||||
// The isolated Gateway callback process only enqueues customer callbacks.
|
||||
// Delivery remains owned by the main API process so callback DB/HTTP capacity
|
||||
// cannot be consumed by slow customer webhook endpoints.
|
||||
if (process.env.CMPP_PROCESS_ROLE === 'callback') return;
|
||||
this.worker = new Worker(WEBHOOK_QUEUE, (job) => this.deliverWebhook(job.data.deliveryId), { connection, concurrency: 10 });
|
||||
}
|
||||
|
||||
|
||||
@@ -7,7 +7,25 @@ describe('PrismaService', () => {
|
||||
expect(Object.getOwnPropertyDescriptor(prisma, 'operationLog')).toEqual(
|
||||
expect.objectContaining({ configurable: true }),
|
||||
);
|
||||
expect(prisma.getPoolState()).toEqual({ max: 32, total: 0, idle: 0, waiting: 0 });
|
||||
|
||||
await prisma.$disconnect();
|
||||
});
|
||||
|
||||
it('reserves a bounded database pool for the isolated Gateway callback process', async () => {
|
||||
const previousRole = process.env.CMPP_PROCESS_ROLE;
|
||||
const previousMax = process.env.API_CALLBACK_DB_POOL_MAX;
|
||||
process.env.CMPP_PROCESS_ROLE = 'callback';
|
||||
process.env.API_CALLBACK_DB_POOL_MAX = '12';
|
||||
try {
|
||||
const prisma = new PrismaService();
|
||||
expect(prisma.getPoolState()).toEqual({ max: 12, total: 0, idle: 0, waiting: 0 });
|
||||
await prisma.$disconnect();
|
||||
} finally {
|
||||
if (previousRole === undefined) delete process.env.CMPP_PROCESS_ROLE;
|
||||
else process.env.CMPP_PROCESS_ROLE = previousRole;
|
||||
if (previousMax === undefined) delete process.env.API_CALLBACK_DB_POOL_MAX;
|
||||
else process.env.API_CALLBACK_DB_POOL_MAX = previousMax;
|
||||
}
|
||||
});
|
||||
});
|
||||
|
||||
@@ -1,30 +1,46 @@
|
||||
import { Injectable, OnModuleDestroy } from '@nestjs/common';
|
||||
import { PrismaPg } from '@prisma/adapter-pg';
|
||||
import { PrismaClient } from '@prisma/client';
|
||||
import { Pool } from 'pg';
|
||||
import { requestContext } from '../common/request-context';
|
||||
|
||||
@Injectable()
|
||||
export class PrismaService extends PrismaClient implements OnModuleDestroy {
|
||||
private readonly databasePool: Pool;
|
||||
private readonly databasePoolMax: number;
|
||||
|
||||
constructor() {
|
||||
const workerRole = process.env.CMPP_PROCESS_ROLE === 'worker';
|
||||
const databaseUrl = workerRole
|
||||
? process.env.API_WORKER_DATABASE_URL || process.env.DATABASE_URL
|
||||
: process.env.DATABASE_URL;
|
||||
const configuredPoolMax = Number(workerRole
|
||||
? process.env.API_WORKER_DB_POOL_MAX ?? 8
|
||||
: process.env.API_DB_POOL_MAX ?? 32);
|
||||
const processRole = process.env.CMPP_PROCESS_ROLE?.trim() || 'api';
|
||||
const workerRole = processRole === 'worker';
|
||||
const outboxRole = processRole === 'outbox';
|
||||
const callbackRole = processRole === 'callback';
|
||||
const databaseUrl = outboxRole
|
||||
? process.env.API_OUTBOX_DATABASE_URL || process.env.DATABASE_URL
|
||||
: callbackRole
|
||||
? process.env.API_CALLBACK_DATABASE_URL || process.env.DATABASE_URL
|
||||
: workerRole
|
||||
? process.env.API_WORKER_DATABASE_URL || process.env.DATABASE_URL
|
||||
: process.env.DATABASE_URL;
|
||||
const configuredPoolMax = Number(outboxRole
|
||||
? process.env.API_OUTBOX_DB_POOL_MAX ?? 6
|
||||
: callbackRole
|
||||
? process.env.API_CALLBACK_DB_POOL_MAX ?? 16
|
||||
: workerRole
|
||||
? process.env.API_WORKER_DB_POOL_MAX ?? 8
|
||||
: process.env.API_DB_POOL_MAX ?? 32);
|
||||
const poolMax = Number.isInteger(configuredPoolMax) && configuredPoolMax > 0
|
||||
? configuredPoolMax
|
||||
: workerRole ? 8 : 32;
|
||||
super({
|
||||
adapter: new PrismaPg({
|
||||
connectionString: databaseUrl
|
||||
?? 'postgresql://cmpp:cmpp_password@localhost:5432/cmpp_platform?schema=public',
|
||||
// API capacity must be reserved independently from the heavier Worker
|
||||
// transactions; explicit bounds also protect PostgreSQL max_connections.
|
||||
max: poolMax,
|
||||
}),
|
||||
: outboxRole ? 6 : callbackRole ? 16 : workerRole ? 8 : 32;
|
||||
const databasePool = new Pool({
|
||||
connectionString: databaseUrl
|
||||
?? 'postgresql://cmpp:cmpp_password@localhost:5432/cmpp_platform?schema=public',
|
||||
// API capacity must be reserved independently from the heavier Worker
|
||||
// transactions; explicit bounds also protect PostgreSQL max_connections.
|
||||
max: poolMax,
|
||||
});
|
||||
super({ adapter: new PrismaPg(databasePool, { disposeExternalPool: true }) });
|
||||
this.databasePool = databasePool;
|
||||
this.databasePoolMax = poolMax;
|
||||
const operationLog = this.operationLog;
|
||||
Object.defineProperty(this, 'operationLog', {
|
||||
value: new Proxy(operationLog, {
|
||||
@@ -46,6 +62,15 @@ export class PrismaService extends PrismaClient implements OnModuleDestroy {
|
||||
});
|
||||
}
|
||||
|
||||
getPoolState() {
|
||||
return {
|
||||
max: this.databasePoolMax,
|
||||
total: this.databasePool.totalCount,
|
||||
idle: this.databasePool.idleCount,
|
||||
waiting: this.databasePool.waitingCount,
|
||||
};
|
||||
}
|
||||
|
||||
async onModuleDestroy() {
|
||||
await this.$disconnect();
|
||||
}
|
||||
|
||||
@@ -0,0 +1,43 @@
|
||||
import { BadRequestException } from '@nestjs/common';
|
||||
import { GatewayCallbackController } from './gateway-callback.controller';
|
||||
|
||||
describe('GatewayCallbackController', () => {
|
||||
const sendChain = {
|
||||
handleSubmitResult: jest.fn(), handleSubmitSegmentResult: jest.fn(),
|
||||
intakeReceipt: jest.fn(), handleReceipt: jest.fn(), handleUplink: jest.fn(),
|
||||
recordGatewaySubmitDeadLetter: jest.fn(),
|
||||
};
|
||||
const protocolLogs = { record: jest.fn() };
|
||||
const prisma = { $queryRaw: jest.fn(), getPoolState: jest.fn().mockReturnValue({ max: 16, total: 2, idle: 1, waiting: 0 }) };
|
||||
const controller = new GatewayCallbackController(sendChain as never, protocolLogs as never, prisma as never);
|
||||
|
||||
beforeEach(() => jest.clearAllMocks());
|
||||
|
||||
it('keeps Submit result persistence on the callback process without duplicating protocol logs', async () => {
|
||||
sendChain.handleSubmitResult.mockResolvedValue({ accepted: true });
|
||||
await expect(controller.submitResult({
|
||||
messageId: 'MSG-1', channelId: 'channel-1', gatewayMessageId: '1', sequenceId: 1,
|
||||
submitStatus: 'accepted',
|
||||
})).resolves.toEqual({ accepted: true });
|
||||
expect(sendChain.handleSubmitResult).toHaveBeenCalledTimes(1);
|
||||
expect(protocolLogs.record).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it('durably intakes receipts before recording the callback protocol event', async () => {
|
||||
sendChain.intakeReceipt.mockResolvedValue({ accepted: true, inboxId: 'inbox-1' });
|
||||
await controller.receiptIntake({
|
||||
messageId: 'MSG-1', channelId: 'channel-1', gatewayMessageId: '1',
|
||||
phoneNumber: '13800000001', receiptStatus: 'delivered', rawStatus: 'DELIVRD',
|
||||
});
|
||||
expect(sendChain.intakeReceipt).toHaveBeenCalledTimes(1);
|
||||
expect(protocolLogs.record).toHaveBeenCalledWith(expect.objectContaining({
|
||||
eventType: 'deliver_receipt', messageId: 'MSG-1', status: 'success',
|
||||
}));
|
||||
});
|
||||
|
||||
it('rejects inbound/client packet logs from the isolated supplier callback surface', () => {
|
||||
expect(() => controller.protocolLog({
|
||||
protocol: 'cmpp', direction: 'client_to_platform', eventType: 'submit', status: 'success',
|
||||
})).toThrow(BadRequestException);
|
||||
});
|
||||
});
|
||||
@@ -0,0 +1,102 @@
|
||||
import { BadRequestException, Body, Controller, Get, Post } from '@nestjs/common';
|
||||
import { PrismaService } from '../prisma/prisma.service';
|
||||
import { ProtocolLogsService, type ProtocolLogInput } from '../protocol-logs/protocol-logs.service';
|
||||
import type {
|
||||
GatewayReceiptEventDto,
|
||||
GatewaySubmitDeadLetterDto,
|
||||
GatewaySubmitResultDto,
|
||||
GatewaySubmitSegmentResultDto,
|
||||
GatewayUplinkEventDto,
|
||||
} from './send-chain.contracts';
|
||||
import { SendChainService } from './send-chain.service';
|
||||
|
||||
@Controller()
|
||||
export class GatewayCallbackController {
|
||||
constructor(
|
||||
private readonly sendChain: SendChainService,
|
||||
private readonly protocolLogs: ProtocolLogsService,
|
||||
private readonly prisma: PrismaService,
|
||||
) {}
|
||||
|
||||
@Get('health')
|
||||
async health() {
|
||||
await this.prisma.$queryRaw`SELECT 1`;
|
||||
return { status: 'ok', role: 'gateway-callback', databasePool: this.prisma.getPoolState() };
|
||||
}
|
||||
|
||||
@Post('gateway/events/submit-result')
|
||||
submitResult(@Body() body: GatewaySubmitResultDto) {
|
||||
return this.sendChain.handleSubmitResult(body);
|
||||
}
|
||||
|
||||
@Post('gateway/events/submit-segment-result')
|
||||
submitSegmentResult(@Body() body: GatewaySubmitSegmentResultDto) {
|
||||
return this.sendChain.handleSubmitSegmentResult(body);
|
||||
}
|
||||
|
||||
@Post('gateway/events/receipt/intake')
|
||||
receiptIntake(@Body() body: GatewayReceiptEventDto) {
|
||||
return this.track('deliver_receipt', body, () => this.sendChain.intakeReceipt(body));
|
||||
}
|
||||
|
||||
@Post('gateway/events/receipt')
|
||||
receipt(@Body() body: GatewayReceiptEventDto) {
|
||||
return this.track('deliver_receipt', body, () => this.sendChain.handleReceipt(body));
|
||||
}
|
||||
|
||||
@Post('gateway/events/uplink')
|
||||
uplink(@Body() body: GatewayUplinkEventDto) {
|
||||
return this.track('deliver_uplink', body, () => this.sendChain.handleUplink(body));
|
||||
}
|
||||
|
||||
@Post('gateway/events/protocol-log')
|
||||
protocolLog(@Body() body: ProtocolLogInput) {
|
||||
const allowedPacket = (
|
||||
body.direction === 'platform_to_channel' && ['submit', 'deliver_resp'].includes(body.eventType)
|
||||
) || (
|
||||
body.direction === 'channel_to_platform' && body.eventType === 'submit_resp'
|
||||
);
|
||||
if (body.protocol !== 'cmpp' || !allowedPacket || !['success', 'failed'].includes(body.status)) {
|
||||
throw new BadRequestException('Unsupported Gateway callback protocol log event');
|
||||
}
|
||||
this.protocolLogs.record(body);
|
||||
return { accepted: true };
|
||||
}
|
||||
|
||||
@Post('gateway/events/dead-letter')
|
||||
deadLetter(@Body() body: GatewaySubmitDeadLetterDto) {
|
||||
return this.sendChain.recordGatewaySubmitDeadLetter(body);
|
||||
}
|
||||
|
||||
private async track<T>(eventType: string, body: object, action: () => Promise<T> | T) {
|
||||
const startedAt = Date.now();
|
||||
const value = body as Record<string, unknown>;
|
||||
const common: Omit<ProtocolLogInput, 'status'> = {
|
||||
protocol: 'cmpp', direction: 'channel_to_platform', eventType,
|
||||
tenantId: value.tenantId as string, applicationId: value.applicationId as string,
|
||||
channelId: value.channelId as string,
|
||||
messageId: (value.messageId ?? value.platformMessageId) as string,
|
||||
gatewayMessageId: (value.gatewayMessageId ?? value.msgId ?? value.upstreamMessageId) as string,
|
||||
phone: (value.phoneNumber ?? value.srcTerminalId) as string,
|
||||
resultCode: (value.rawStatus ?? value.status ?? value.stat) as string,
|
||||
};
|
||||
try {
|
||||
const result = await action();
|
||||
const resolved = result && typeof result === 'object' ? result as Record<string, unknown> : {};
|
||||
this.protocolLogs.record({
|
||||
...common,
|
||||
tenantId: (resolved.tenantId ?? common.tenantId) as string,
|
||||
applicationId: (resolved.applicationId ?? common.applicationId) as string,
|
||||
messageId: (resolved.messageId ?? common.messageId) as string,
|
||||
status: 'success', durationMs: Date.now() - startedAt,
|
||||
});
|
||||
return result;
|
||||
} catch (error) {
|
||||
this.protocolLogs.record({
|
||||
...common, status: 'failed', durationMs: Date.now() - startedAt,
|
||||
detail: { error: error instanceof Error ? error.message : String(error) },
|
||||
});
|
||||
throw error;
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -158,10 +158,12 @@ function createPrismaMock() {
|
||||
findUnique: jest.fn().mockResolvedValue(channel),
|
||||
},
|
||||
cmppSubmitSession: {
|
||||
findUnique: jest.fn().mockResolvedValue({ id: 'session-1' }),
|
||||
upsert: jest.fn().mockResolvedValue({ id: 'session-1' }),
|
||||
},
|
||||
smsSubmitRecord: {
|
||||
create: jest.fn().mockResolvedValue({ id: 'submit-1' }),
|
||||
createMany: jest.fn().mockResolvedValue({ count: 2 }),
|
||||
updateMany: jest.fn().mockResolvedValue({ count: 1 }),
|
||||
findFirst: jest.fn().mockResolvedValue({ id: 'submit-1', submitId: 'SUB-1', submitStatus: 'accepted', createdAt: new Date('2026-07-01T10:00:00.000Z') }),
|
||||
findUnique: jest.fn().mockImplementation(({ where }) => Promise.resolve(
|
||||
@@ -329,6 +331,11 @@ function createPrismaMock() {
|
||||
lastError: 'downstream client is not connected',
|
||||
}),
|
||||
},
|
||||
gatewaySubmitOutbox: {
|
||||
create: jest.fn().mockResolvedValue({ id: 'outbox-1', submitId: 'SUB-1', status: 'pending' }),
|
||||
createMany: jest.fn().mockResolvedValue({ count: 2 }),
|
||||
updateMany: jest.fn().mockResolvedValue({ count: 1 }),
|
||||
},
|
||||
cmppInboundSubmissionInbox: {
|
||||
create: jest.fn().mockResolvedValue({ id: 'inbox-1' }),
|
||||
findUnique: jest.fn().mockResolvedValue(null),
|
||||
@@ -926,6 +933,7 @@ describe('SendChainService', () => {
|
||||
|
||||
it('coalesces concurrent batch progress refreshes and keeps a trailing refresh', async () => {
|
||||
const { service, prisma } = createService();
|
||||
prisma.$executeRaw.mockResolvedValue(0);
|
||||
let resolveFirst: ((value: Array<{ status: string; _count: { _all: number } }>) => void) | undefined;
|
||||
prisma.smsMessageRecord.groupBy
|
||||
.mockImplementationOnce(() => new Promise((resolve) => { resolveFirst = resolve; }))
|
||||
@@ -946,6 +954,37 @@ describe('SendChainService', () => {
|
||||
});
|
||||
});
|
||||
|
||||
it('updates a known single-message CMPP task without grouping all message states', async () => {
|
||||
const { service, prisma } = createService();
|
||||
|
||||
await service['submission'].refreshTaskProgress('task-cmpp-1', 'submit_queued');
|
||||
|
||||
expect(prisma.smsBatchTask.updateMany).toHaveBeenCalledWith({
|
||||
where: { id: 'task-cmpp-1', sourceType: 'cmpp', phoneTotal: 1 },
|
||||
data: {
|
||||
progressTotal: 1,
|
||||
submittedTotal: 1,
|
||||
successTotal: 0,
|
||||
failedTotal: 0,
|
||||
unknownTotal: 0,
|
||||
timeoutTotal: 0,
|
||||
status: 'sending',
|
||||
},
|
||||
});
|
||||
expect(prisma.smsMessageRecord.groupBy).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it('refreshes callback progress for a single-message CMPP task in one direct SQL update', async () => {
|
||||
const { service, prisma } = createService();
|
||||
prisma.$executeRaw.mockResolvedValue(1);
|
||||
|
||||
await service['submission'].refreshTaskProgress('task-cmpp-callback');
|
||||
|
||||
expect(prisma.$executeRaw).toHaveBeenCalledTimes(1);
|
||||
expect(prisma.smsMessageRecord.groupBy).not.toHaveBeenCalled();
|
||||
expect(prisma.smsBatchTask.update).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it('does not expose CMPP internal tasks through client task detail or messages', async () => {
|
||||
const { service, prisma } = createService();
|
||||
prisma.smsBatchTask.findFirst.mockResolvedValue(null);
|
||||
@@ -2038,9 +2077,7 @@ describe('SendChainService', () => {
|
||||
|
||||
it('routes queued messages to gateway submit commands', async () => {
|
||||
const { service, prisma } = createService();
|
||||
const gatewayAdd = jest.fn().mockResolvedValue(undefined);
|
||||
service['waitForChannelRateLimit'] = jest.fn().mockResolvedValue(undefined);
|
||||
service['getGatewayQueue'] = jest.fn().mockReturnValue({ add: gatewayAdd });
|
||||
|
||||
await expect(service.processSendJob({ messageRecordId: 'record-1' })).resolves.toEqual(
|
||||
expect.objectContaining({ submitted: true, messageRecordId: 'record-1', channelId: 'channel-1' }),
|
||||
@@ -2059,8 +2096,7 @@ describe('SendChainService', () => {
|
||||
where: { id: 'record-1' },
|
||||
data: expect.objectContaining({ channelId: 'channel-1', carrier: 'mobile', province: '山东', status: 'submit_queued' }),
|
||||
});
|
||||
expect(gatewayAdd).toHaveBeenCalledWith(
|
||||
'submit-command',
|
||||
expect(service['publishGatewaySubmitCommand']).toHaveBeenCalledWith(
|
||||
expect.objectContaining({
|
||||
schemaVersion: 'v1',
|
||||
messageType: 'SubmitCommand',
|
||||
@@ -2074,9 +2110,117 @@ describe('SendChainService', () => {
|
||||
}),
|
||||
);
|
||||
expect(service['publishGatewaySubmitCommand']).toHaveBeenCalledWith(expect.objectContaining({ messageId: 'MSG-1' }));
|
||||
expect(prisma.cmppSubmitSession.findUnique).toHaveBeenCalledWith({
|
||||
where: { sessionNo: 'OPEN-channel-1' },
|
||||
select: { id: true },
|
||||
});
|
||||
expect(prisma.cmppSubmitSession.upsert).not.toHaveBeenCalled();
|
||||
expect(service['postGatewayControl']).not.toHaveBeenCalledWith('/upstream/submit', expect.anything());
|
||||
});
|
||||
|
||||
it('shadow-writes the durable submit Outbox without replacing the direct stream path', async () => {
|
||||
const previousShadow = process.env.SEND_SUBMIT_OUTBOX_SHADOW_ENABLED;
|
||||
const previousPublish = process.env.SEND_SUBMIT_OUTBOX_PUBLISH_ENABLED;
|
||||
process.env.SEND_SUBMIT_OUTBOX_SHADOW_ENABLED = 'true';
|
||||
delete process.env.SEND_SUBMIT_OUTBOX_PUBLISH_ENABLED;
|
||||
try {
|
||||
const { service, prisma } = createService();
|
||||
service['waitForChannelRateLimit'] = jest.fn().mockResolvedValue(undefined);
|
||||
|
||||
await service.processSendJob({ messageRecordId: 'record-1' });
|
||||
|
||||
expect(prisma.gatewaySubmitOutbox.create).toHaveBeenCalledWith({
|
||||
data: expect.objectContaining({
|
||||
messageRecordId: 'record-1',
|
||||
channelId: 'channel-1',
|
||||
payload: expect.objectContaining({ messageType: 'SubmitCommand', messageId: 'MSG-1' }),
|
||||
}),
|
||||
});
|
||||
expect(service['publishGatewaySubmitCommand']).toHaveBeenCalledTimes(1);
|
||||
} finally {
|
||||
if (previousShadow === undefined) delete process.env.SEND_SUBMIT_OUTBOX_SHADOW_ENABLED;
|
||||
else process.env.SEND_SUBMIT_OUTBOX_SHADOW_ENABLED = previousShadow;
|
||||
if (previousPublish === undefined) delete process.env.SEND_SUBMIT_OUTBOX_PUBLISH_ENABLED;
|
||||
else process.env.SEND_SUBMIT_OUTBOX_PUBLISH_ENABLED = previousPublish;
|
||||
}
|
||||
});
|
||||
|
||||
it('uses only the durable Outbox when formal publishing is enabled', async () => {
|
||||
const previousShadow = process.env.SEND_SUBMIT_OUTBOX_SHADOW_ENABLED;
|
||||
const previousPublish = process.env.SEND_SUBMIT_OUTBOX_PUBLISH_ENABLED;
|
||||
process.env.SEND_SUBMIT_OUTBOX_SHADOW_ENABLED = 'true';
|
||||
process.env.SEND_SUBMIT_OUTBOX_PUBLISH_ENABLED = 'true';
|
||||
try {
|
||||
const { service, prisma } = createService();
|
||||
service['waitForChannelRateLimit'] = jest.fn().mockResolvedValue(undefined);
|
||||
|
||||
await service.processSendJob({ messageRecordId: 'record-1' });
|
||||
|
||||
expect(prisma.gatewaySubmitOutbox.create).toHaveBeenCalledTimes(1);
|
||||
expect(service['publishGatewaySubmitCommand']).not.toHaveBeenCalled();
|
||||
} finally {
|
||||
if (previousShadow === undefined) delete process.env.SEND_SUBMIT_OUTBOX_SHADOW_ENABLED;
|
||||
else process.env.SEND_SUBMIT_OUTBOX_SHADOW_ENABLED = previousShadow;
|
||||
if (previousPublish === undefined) delete process.env.SEND_SUBMIT_OUTBOX_PUBLISH_ENABLED;
|
||||
else process.env.SEND_SUBMIT_OUTBOX_PUBLISH_ENABLED = previousPublish;
|
||||
}
|
||||
});
|
||||
|
||||
it('plans routes once and bulk-creates Submit records and Outbox rows for a Worker batch', async () => {
|
||||
const previousPublish = process.env.SEND_SUBMIT_OUTBOX_PUBLISH_ENABLED;
|
||||
process.env.SEND_SUBMIT_OUTBOX_PUBLISH_ENABLED = 'true';
|
||||
try {
|
||||
const { service, prisma } = createService();
|
||||
const base = await prisma.smsMessageRecord.findUnique({ where: { id: 'record-1' } });
|
||||
const messages = [
|
||||
{ ...base, id: 'record-1', messageId: 'MSG-1', phoneNumber: '13800000001', amountCents: 3n, carrier: null, province: null, batchTask: { sourceType: 'cmpp', phoneTotal: 1 } },
|
||||
{ ...base, id: 'record-2', batchTaskId: 'task-2', messageId: 'MSG-2', phoneNumber: '13800000002', amountCents: 3n, carrier: null, province: null, batchTask: { sourceType: 'cmpp', phoneTotal: 1 } },
|
||||
];
|
||||
const channel = {
|
||||
id: 'channel-1', code: 'CMPP-A', account: 'cmpp-account', srcId: '10690000',
|
||||
rateLimitPerSecond: 100, unitPrice: 3n, status: 'active', carrier: 'mobile', sendRegion: '全国',
|
||||
gatewayHost: '127.0.0.1', gatewayPort: 17890, passwordCipher: 'secret', cmppVersion: '3.0',
|
||||
config: { serviceId: 'SMS' },
|
||||
connectionStates: [{ status: 'connected', currentConnections: 1, desiredConnections: 1 }],
|
||||
reportTasks: [{ signatureId: 'sig-1', carrier: 'mobile', approvalScope: 'carrier_specific' }],
|
||||
};
|
||||
prisma.smsMessageRecord.findMany.mockResolvedValue(messages);
|
||||
prisma.channelRouteRule.findMany.mockResolvedValue([{
|
||||
tenantId: 'tenant-1', applicationId: 'app-1', carrier: 'mobile', groupId: 'group-1',
|
||||
group: {
|
||||
name: '默认通道组', carrier: 'mobile', status: 'active',
|
||||
items: [{ id: 'item-1', groupId: 'group-1', channelId: 'channel-1', carrier: 'mobile', province: null, priority: 1, weight: 1, isBackup: false, channel }],
|
||||
},
|
||||
}]);
|
||||
service['waitForChannelRateLimit'] = jest.fn().mockResolvedValue(undefined);
|
||||
|
||||
const gatewaySubmit = (service as any).submission.gatewaySubmit;
|
||||
const result = await gatewaySubmit.processSendJobBatch([
|
||||
{ messageRecordId: 'record-1' }, { messageRecordId: 'record-2' },
|
||||
]);
|
||||
|
||||
expect(prisma.channelRouteRule.findMany).toHaveBeenCalledTimes(1);
|
||||
expect(prisma.phoneSegment.findMany).toHaveBeenCalledTimes(1);
|
||||
expect(prisma.smsSubmitRecord.createMany).toHaveBeenCalledWith({
|
||||
data: expect.arrayContaining([
|
||||
expect.objectContaining({ messageRecordId: 'record-1', channelId: 'channel-1' }),
|
||||
expect.objectContaining({ messageRecordId: 'record-2', channelId: 'channel-1' }),
|
||||
]),
|
||||
});
|
||||
expect(prisma.gatewaySubmitOutbox.createMany).toHaveBeenCalledWith({
|
||||
data: expect.arrayContaining([
|
||||
expect.objectContaining({ messageRecordId: 'record-1', payload: expect.objectContaining({ messageId: 'MSG-1' }) }),
|
||||
expect.objectContaining({ messageRecordId: 'record-2', payload: expect.objectContaining({ messageId: 'MSG-2' }) }),
|
||||
]),
|
||||
});
|
||||
expect(result.get('record-1')).toEqual(expect.objectContaining({ submitted: true }));
|
||||
expect(result.get('record-2')).toEqual(expect.objectContaining({ submitted: true }));
|
||||
} finally {
|
||||
if (previousPublish === undefined) delete process.env.SEND_SUBMIT_OUTBOX_PUBLISH_ENABLED;
|
||||
else process.env.SEND_SUBMIT_OUTBOX_PUBLISH_ENABLED = previousPublish;
|
||||
}
|
||||
});
|
||||
|
||||
it('appends the real application extension to the upstream channel base number', async () => {
|
||||
const { service, prisma } = createService();
|
||||
const queuedMessage = await prisma.smsMessageRecord.findUnique({ where: { id: 'record-1' } });
|
||||
@@ -2085,14 +2229,11 @@ describe('SendChainService', () => {
|
||||
applicationExtension: '0001',
|
||||
clientSrcId: '000001',
|
||||
});
|
||||
const gatewayAdd = jest.fn().mockResolvedValue(undefined);
|
||||
service['waitForChannelRateLimit'] = jest.fn().mockResolvedValue(undefined);
|
||||
service['getGatewayQueue'] = jest.fn().mockReturnValue({ add: gatewayAdd });
|
||||
|
||||
await service.processSendJob({ messageRecordId: 'record-1' });
|
||||
|
||||
expect(gatewayAdd).toHaveBeenCalledWith(
|
||||
'submit-command',
|
||||
expect(service['publishGatewaySubmitCommand']).toHaveBeenCalledWith(
|
||||
expect.objectContaining({ cmpp: expect.objectContaining({ srcId: '106900000001' }) }),
|
||||
);
|
||||
});
|
||||
@@ -2105,17 +2246,14 @@ describe('SendChainService', () => {
|
||||
prisma.channelRouteRule.findFirst.mockResolvedValue({
|
||||
...baseRoute,
|
||||
group: { ...baseRoute.group, items: [
|
||||
{ ...baseRoute.group.items[0], channelId: primary.id, priority: 1, channel: primary },
|
||||
{ ...baseRoute.group.items[0], id: 'item-2', channelId: backup.id, priority: 2, channel: backup },
|
||||
] },
|
||||
});
|
||||
prisma.channelSignatureReportTask.findMany.mockResolvedValue([{ channelId: backup.id }]);
|
||||
const gatewayAdd = jest.fn().mockResolvedValue(undefined);
|
||||
service['waitForChannelRateLimit'] = jest.fn().mockResolvedValue(undefined);
|
||||
service['getGatewayQueue'] = jest.fn().mockReturnValue({ add: gatewayAdd });
|
||||
|
||||
await expect(service.processSendJob({ messageRecordId: 'record-1' })).resolves.toEqual(expect.objectContaining({ submitted: true, channelId: backup.id }));
|
||||
expect(prisma.channelSignatureReportTask.findMany).toHaveBeenCalledWith(expect.objectContaining({ where: expect.objectContaining({ reportType: 'signature' }) }));
|
||||
expect(prisma.channelSignatureReportTask.findMany).not.toHaveBeenCalled();
|
||||
expect(JSON.stringify(prisma.channelRouteRule.findFirst.mock.calls.at(-1)?.[0])).toContain('sig-1');
|
||||
expect(prisma.smsMessageRecord.update).toHaveBeenCalledWith(expect.objectContaining({ data: expect.objectContaining({ channelId: backup.id }) }));
|
||||
});
|
||||
|
||||
@@ -2127,6 +2265,7 @@ describe('SendChainService', () => {
|
||||
id: 'record-1',
|
||||
tenantId: 'tenant-1',
|
||||
applicationId: 'app-1',
|
||||
signatureId: 'sig-1',
|
||||
phoneNumber: '13800000001',
|
||||
})).rejects.toThrow('企业应用未配置对应运营商通道组');
|
||||
|
||||
@@ -2554,9 +2693,7 @@ describe('SendChainService', () => {
|
||||
queuedAt: new Date(),
|
||||
}, '回执失败补发')).resolves.toEqual(expect.objectContaining({ channelId: backup.id }));
|
||||
|
||||
expect(prisma.channelSignatureReportTask.findMany).toHaveBeenCalledWith(expect.objectContaining({
|
||||
where: expect.objectContaining({ signatureId: 'sig-direct' }),
|
||||
}));
|
||||
expect(JSON.stringify(prisma.channelRouteRule.findFirst.mock.calls.at(-1)?.[0])).toContain('sig-direct');
|
||||
expect(submitMessageToGateway).toHaveBeenCalledWith(
|
||||
expect.objectContaining({ signatureId: 'sig-direct' }),
|
||||
expect.objectContaining({ channel: expect.objectContaining({ id: backup.id }) }),
|
||||
@@ -2617,7 +2754,7 @@ describe('SendChainService', () => {
|
||||
|
||||
expect(results.filter((result) => result.submitted)).toHaveLength(1);
|
||||
expect(results.filter((result) => result.duplicateRetry)).toHaveLength(2);
|
||||
expect(queueAdd).toHaveBeenCalledTimes(1);
|
||||
expect(queueAdd).not.toHaveBeenCalled();
|
||||
expect(service['publishGatewaySubmitCommand']).toHaveBeenCalledTimes(1);
|
||||
expect(prisma.smsMessageRecord.update).toHaveBeenCalledTimes(1);
|
||||
});
|
||||
@@ -3378,10 +3515,9 @@ describe('SendChainService', () => {
|
||||
|
||||
it('blocks submit when signature is not approved on the selected channel', async () => {
|
||||
const { service, prisma } = createService();
|
||||
const gatewayAdd = jest.fn().mockResolvedValue(undefined);
|
||||
service['waitForChannelRateLimit'] = jest.fn().mockResolvedValue(undefined);
|
||||
service['getGatewayQueue'] = jest.fn().mockReturnValue({ add: gatewayAdd });
|
||||
prisma.channelSignatureReportTask.findMany.mockResolvedValue([]);
|
||||
const route = await prisma.channelRouteRule.findFirst();
|
||||
prisma.channelRouteRule.findFirst.mockResolvedValue({ ...route, group: { ...route.group, items: [] } });
|
||||
prisma.smsMessageRecord.findUnique.mockResolvedValue({
|
||||
id: 'record-1',
|
||||
tenantId: 'tenant-1',
|
||||
@@ -3396,14 +3532,14 @@ describe('SendChainService', () => {
|
||||
amountCents: 3,
|
||||
status: 'queued',
|
||||
queuePriority: 'normal',
|
||||
batchTask: { sourceType: 'cmpp' },
|
||||
batchTask: { sourceType: 'cmpp', phoneTotal: 1 },
|
||||
template: { signature: { id: 'sig-1', name: '签名' } },
|
||||
});
|
||||
|
||||
await expect(service.processSendJob({ messageRecordId: 'record-1' })).resolves.toEqual(
|
||||
expect.objectContaining({ submitted: false, status: 'failed', reason: '无已报备通过且在线的可用通道' }),
|
||||
);
|
||||
expect(gatewayAdd).not.toHaveBeenCalled();
|
||||
expect(service['publishGatewaySubmitCommand']).not.toHaveBeenCalled();
|
||||
expect(prisma.smsReceiptRecord.create).toHaveBeenCalledWith({
|
||||
data: expect.objectContaining({ receiptStatus: 'undelivered', errorCode: 'ROUTE' }),
|
||||
});
|
||||
@@ -3411,9 +3547,7 @@ describe('SendChainService', () => {
|
||||
|
||||
it('only selects channel group items allocated to the matched carrier', async () => {
|
||||
const { service, prisma } = createService();
|
||||
const gatewayAdd = jest.fn().mockResolvedValue(undefined);
|
||||
service['waitForChannelRateLimit'] = jest.fn().mockResolvedValue(undefined);
|
||||
service['getGatewayQueue'] = jest.fn().mockReturnValue({ add: gatewayAdd });
|
||||
prisma.channelRouteRule.findFirst.mockResolvedValue({
|
||||
id: 'route-1',
|
||||
tenantId: 'tenant-1',
|
||||
@@ -3475,8 +3609,7 @@ describe('SendChainService', () => {
|
||||
await expect(service.processSendJob({ messageRecordId: 'record-1' })).resolves.toEqual(
|
||||
expect.objectContaining({ channelId: 'channel-all' }),
|
||||
);
|
||||
expect(gatewayAdd).toHaveBeenCalledWith(
|
||||
'submit-command',
|
||||
expect(service['publishGatewaySubmitCommand']).toHaveBeenCalledWith(
|
||||
expect.objectContaining({ channelId: 'channel-all', route: expect.objectContaining({ channelCode: 'CMPP-ALL', carrier: 'mobile' }) }),
|
||||
);
|
||||
});
|
||||
@@ -4429,7 +4562,7 @@ describe('SendChainService', () => {
|
||||
where: { id: 'record-1', status: 'timeout', timeoutReceiptQueuedAt: null },
|
||||
data: { timeoutReceiptQueuedAt: expect.any(Date) },
|
||||
});
|
||||
expect(prisma.smsBatchTask.update).toHaveBeenCalled();
|
||||
expect(prisma.$executeRaw).toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it('queues an explicit HTTP failure webhook when a receipt times out', async () => {
|
||||
|
||||
@@ -79,7 +79,11 @@ export class SendChainService implements OnModuleInit, OnModuleDestroy {
|
||||
|
||||
onModuleInit() {
|
||||
const processRole = process.env.CMPP_PROCESS_ROLE?.trim() || 'all';
|
||||
if (processRole === 'api') return;
|
||||
if (processRole === 'api' || processRole === 'callback') return;
|
||||
if (processRole === 'outbox') {
|
||||
this.submission.startSubmitOutboxPublisher();
|
||||
return;
|
||||
}
|
||||
if (process.env.API_ENABLE_SEND_WORKER === 'true') {
|
||||
this.startWorker();
|
||||
}
|
||||
@@ -701,8 +705,8 @@ export class SendChainService implements OnModuleInit, OnModuleDestroy {
|
||||
return this.submission.selectChannelForMessage(message, options);
|
||||
}
|
||||
|
||||
private async findApplicationRoute(tenantId: string, applicationId: string | undefined, carrier: string) {
|
||||
return this.submission.findApplicationRoute(tenantId, applicationId, carrier);
|
||||
private async findApplicationRoute(tenantId: string, applicationId: string | undefined, carrier: string, signatureId?: string) {
|
||||
return this.submission.findApplicationRoute(tenantId, applicationId, carrier, signatureId);
|
||||
}
|
||||
|
||||
private async identifyCarrier(phoneNumber: string) {
|
||||
@@ -836,8 +840,8 @@ export class SendChainService implements OnModuleInit, OnModuleDestroy {
|
||||
return this.submission.waitForChannelRateLimit(channelId, tps);
|
||||
}
|
||||
|
||||
private async refreshTaskProgress(batchTaskId: string) {
|
||||
return this.submission.refreshTaskProgress(batchTaskId);
|
||||
private async refreshTaskProgress(batchTaskId: string, knownSingleMessageStatus?: string) {
|
||||
return this.submission.refreshTaskProgress(batchTaskId, knownSingleMessageStatus);
|
||||
}
|
||||
|
||||
private smsMessageSegmentAuditDelegate() {
|
||||
|
||||
@@ -8,6 +8,7 @@ import { BillingService } from '../billing/billing.service';
|
||||
import { isIpAllowed } from '../common/ip-allowlist';
|
||||
import { moneyToNumber } from '../common/money';
|
||||
import { PhoneRoutingLookupService } from '../dictionaries/phone-routing-lookup.service';
|
||||
import { MetricsService, SendWorkerQueueState, SendWorkerStage } from '../metrics/metrics.service';
|
||||
import { PrismaService } from '../prisma/prisma.service';
|
||||
import { RiskReviewService } from '../risk-review/risk-review.service';
|
||||
import { PhoneFrequencyService } from '../risk-review/phone-frequency.service';
|
||||
@@ -15,6 +16,12 @@ import type { CreateBatchTaskDto, CreateHttpBatchTaskDto, GatewayInboundAuthDto,
|
||||
import { SEND_QUEUE, GATEWAY_SUBMIT_QUEUE, GATEWAY_SUBMIT_STREAM, DEFAULT_SCHEDULED_DISPATCH_STALE_MS, DEFAULT_INBOUND_LONG_MESSAGE_PROCESSING_STALE_SECONDS, GATEWAY_SUBMIT_REQUEUE_IDEMPOTENCY_TTL_SECONDS, BULLMQ_PRIORITY, drainageRejectionReason, statusFromRisk, parseSchedule, parseImportRows, splitImportLine, cellByHeader, normalizeCarrier, normalizeQueuePriority, getPositiveConfigInteger, getNonNegativeConfigInteger, isCarrierCompatible, matchTemplateContent, isNationalChannel, validateInboundApplicationSrcId, composeUpstreamSrcId, positiveInteger, parseOptionalSequenceId, shanghaiDateKey, bullmqConnection, matchesApplicationSecret, octetString, selectChannelCandidate } from './send-chain.helpers';
|
||||
import type { SendSubmissionCallbacks, SendSubmissionService } from './send-submission.service';
|
||||
|
||||
type PendingSendBatchItem = {
|
||||
job: SendJob;
|
||||
resolve: (value: unknown) => void;
|
||||
reject: (reason: unknown) => void;
|
||||
};
|
||||
|
||||
/**
|
||||
* R9 gatewaySubmit implementation. Cross-method calls return through the stable SendChainService seam.
|
||||
*/
|
||||
@@ -24,8 +31,18 @@ export class SendGatewaySubmitService {
|
||||
private sendQueue?: Queue<SendJob, unknown, 'send-message'>;
|
||||
private gatewayQueue?: Queue;
|
||||
private worker?: Worker<SendJob>;
|
||||
private sendQueueMetricsTimer?: ReturnType<typeof setInterval>;
|
||||
private submitOutboxTimer?: ReturnType<typeof setInterval>;
|
||||
private submitOutboxRunning = false;
|
||||
private readonly submitOutboxLeaseOwner = `send-worker-${process.pid}-${randomUUID()}`;
|
||||
private sendWorkerInFlight = 0;
|
||||
private sendWorkerConfiguredSlots = 0;
|
||||
private pendingSendBatch: PendingSendBatchItem[] = [];
|
||||
private sendBatchTimer?: ReturnType<typeof setTimeout>;
|
||||
private sendBatchFlushing = false;
|
||||
private readonly taskProgressRefreshes = new Map<string, Promise<void>>();
|
||||
private readonly dirtyTaskProgressRefreshes = new Set<string>();
|
||||
private readonly openSubmitSessionIds = new Map<string, Promise<string>>();
|
||||
|
||||
constructor(
|
||||
private readonly prisma: PrismaService,
|
||||
@@ -35,9 +52,14 @@ export class SendGatewaySubmitService {
|
||||
private readonly phoneRouting: PhoneRoutingLookupService,
|
||||
private readonly facade: SendSubmissionService,
|
||||
private readonly callbacks: SendSubmissionCallbacks,
|
||||
private readonly metrics?: MetricsService,
|
||||
) {}
|
||||
|
||||
async onModuleDestroy() {
|
||||
if (this.sendQueueMetricsTimer) clearInterval(this.sendQueueMetricsTimer);
|
||||
if (this.submitOutboxTimer) clearInterval(this.submitOutboxTimer);
|
||||
if (this.sendBatchTimer) clearTimeout(this.sendBatchTimer);
|
||||
await this.flushSendBatch();
|
||||
await this.worker?.close();
|
||||
await this.sendQueue?.close();
|
||||
await this.gatewayQueue?.close();
|
||||
@@ -112,46 +134,390 @@ startWorker() {
|
||||
return { status: 'already_started' };
|
||||
}
|
||||
const connection = bullmqConnection();
|
||||
const configuredConcurrency = Number(process.env.API_SEND_WORKER_CONCURRENCY ?? 20);
|
||||
this.sendWorkerConfiguredSlots = Number.isInteger(configuredConcurrency) && configuredConcurrency > 0 ? configuredConcurrency : 20;
|
||||
this.metrics?.setSendWorkerSlots(this.sendWorkerConfiguredSlots, this.sendWorkerInFlight);
|
||||
this.worker = new Worker<SendJob>(
|
||||
SEND_QUEUE,
|
||||
async (job) => this.facade.processSendJob(job.data),
|
||||
{ connection, concurrency: Number(process.env.API_SEND_WORKER_CONCURRENCY ?? 20) },
|
||||
async (job) => {
|
||||
this.sendWorkerInFlight += 1;
|
||||
this.metrics?.setSendWorkerSlots(this.sendWorkerConfiguredSlots, this.sendWorkerInFlight);
|
||||
try {
|
||||
return process.env.API_SEND_WORKER_BATCH_ENABLED === 'false'
|
||||
? await this.facade.processSendJob(job.data)
|
||||
: await this.enqueueSendBatch(job.data);
|
||||
} finally {
|
||||
this.sendWorkerInFlight = Math.max(0, this.sendWorkerInFlight - 1);
|
||||
this.metrics?.setSendWorkerSlots(this.sendWorkerConfiguredSlots, this.sendWorkerInFlight);
|
||||
}
|
||||
},
|
||||
{ connection, concurrency: this.sendWorkerConfiguredSlots },
|
||||
);
|
||||
void this.refreshSendQueueMetrics();
|
||||
this.sendQueueMetricsTimer = setInterval(() => void this.refreshSendQueueMetrics(), 5_000);
|
||||
this.sendQueueMetricsTimer.unref?.();
|
||||
if (this.submitOutboxEnabled() && process.env.SEND_SUBMIT_OUTBOX_SEPARATE_PROCESS_ENABLED !== 'true') {
|
||||
void this.publishSubmitOutboxBatch();
|
||||
this.submitOutboxTimer = setInterval(
|
||||
() => void this.publishSubmitOutboxBatch(),
|
||||
getPositiveConfigInteger(process.env, 'SEND_SUBMIT_OUTBOX_POLL_INTERVAL_MS', 25),
|
||||
);
|
||||
this.submitOutboxTimer.unref?.();
|
||||
}
|
||||
return { status: 'started' };
|
||||
}
|
||||
|
||||
async processSendJob(job: SendJob) {
|
||||
const message = await this.prisma.smsMessageRecord.findUnique({
|
||||
where: { id: job.messageRecordId },
|
||||
include: { batchTask: true, template: { include: { signature: true } }, signature: true },
|
||||
});
|
||||
if (!message || message.status !== 'queued') {
|
||||
return { skipped: true };
|
||||
}
|
||||
if (!message.tenantId || !message.batchTaskId) {
|
||||
return { skipped: true, reason: 'standalone channel test message' };
|
||||
}
|
||||
const businessMessage = message as typeof message & { tenantId: string; batchTaskId: string };
|
||||
try {
|
||||
const routed = await this.facade.selectChannelForMessage(businessMessage);
|
||||
return await this.facade.submitMessageToGateway(businessMessage, routed, 0);
|
||||
} catch (error) {
|
||||
const reason = error instanceof Error ? error.message : '无可用通道组或通道';
|
||||
await this.prisma.smsMessageRecord.update({
|
||||
where: { id: message.id },
|
||||
data: { status: 'failed', errorMessage: reason },
|
||||
});
|
||||
await this.releaseMessageReservation(businessMessage, reason);
|
||||
if (message.batchTask?.sourceType === 'cmpp') {
|
||||
await this.recordCmppFailureReceipt(businessMessage, 'ROUTE', reason);
|
||||
} else {
|
||||
await this.facade.refreshTaskProgress(businessMessage.batchTaskId);
|
||||
startSubmitOutboxPublisher() {
|
||||
if (this.submitOutboxTimer) return { status: 'already_started' };
|
||||
if (!this.submitOutboxEnabled()) return { status: 'disabled' };
|
||||
void this.publishSubmitOutboxBatch();
|
||||
this.submitOutboxTimer = setInterval(
|
||||
() => void this.publishSubmitOutboxBatch(),
|
||||
getPositiveConfigInteger(process.env, 'SEND_SUBMIT_OUTBOX_POLL_INTERVAL_MS', 25),
|
||||
);
|
||||
this.submitOutboxTimer.unref?.();
|
||||
return { status: 'started' };
|
||||
}
|
||||
|
||||
private enqueueSendBatch(job: SendJob) {
|
||||
return new Promise((resolve, reject) => {
|
||||
this.pendingSendBatch.push({ job, resolve, reject });
|
||||
const batchSize = Math.min(128, getPositiveConfigInteger(process.env, 'API_SEND_WORKER_BATCH_SIZE', 32));
|
||||
if (this.pendingSendBatch.length >= batchSize) {
|
||||
if (this.sendBatchTimer) clearTimeout(this.sendBatchTimer);
|
||||
this.sendBatchTimer = undefined;
|
||||
queueMicrotask(() => void this.flushSendBatch());
|
||||
return;
|
||||
}
|
||||
return { submitted: false, messageRecordId: message.id, status: 'failed', reason };
|
||||
if (!this.sendBatchTimer) {
|
||||
this.sendBatchTimer = setTimeout(
|
||||
() => {
|
||||
this.sendBatchTimer = undefined;
|
||||
void this.flushSendBatch();
|
||||
},
|
||||
Math.min(25, getPositiveConfigInteger(process.env, 'API_SEND_WORKER_BATCH_WAIT_MS', 3)),
|
||||
);
|
||||
this.sendBatchTimer.unref?.();
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
private async flushSendBatch() {
|
||||
if (this.sendBatchFlushing) return;
|
||||
this.sendBatchFlushing = true;
|
||||
try {
|
||||
const batchSize = Math.min(128, getPositiveConfigInteger(process.env, 'API_SEND_WORKER_BATCH_SIZE', 32));
|
||||
while (this.pendingSendBatch.length > 0) {
|
||||
const batch = this.pendingSendBatch.splice(0, batchSize);
|
||||
try {
|
||||
const results = await this.processSendJobBatch(batch.map((item) => item.job));
|
||||
for (const item of batch) item.resolve(results.get(item.job.messageRecordId) ?? { skipped: true });
|
||||
} catch (error) {
|
||||
for (const item of batch) item.reject(error);
|
||||
}
|
||||
}
|
||||
} finally {
|
||||
this.sendBatchFlushing = false;
|
||||
if (this.pendingSendBatch.length > 0) queueMicrotask(() => void this.flushSendBatch());
|
||||
}
|
||||
}
|
||||
|
||||
async submitMessageToGateway(
|
||||
private async processSendJobBatch(jobs: SendJob[]) {
|
||||
if (jobs.length === 1) {
|
||||
return new Map([[jobs[0].messageRecordId, await this.processSendJob(jobs[0])]]);
|
||||
}
|
||||
const ids = [...new Set(jobs.map((job) => job.messageRecordId))];
|
||||
const messages = await this.measureSendStage('message_load', () => this.prisma.smsMessageRecord.findMany({
|
||||
where: { id: { in: ids } },
|
||||
include: { batchTask: true, template: { include: { signature: true } }, signature: true },
|
||||
}));
|
||||
const messageById = new Map(messages.map((message) => [message.id, message]));
|
||||
const results = new Map<string, unknown>();
|
||||
const businessMessages = messages.filter((message) => {
|
||||
if (message.status !== 'queued' || !message.tenantId || !message.batchTaskId) {
|
||||
results.set(message.id, { skipped: true });
|
||||
return false;
|
||||
}
|
||||
return true;
|
||||
}) as Array<typeof messages[number] & { tenantId: string; batchTaskId: string }>;
|
||||
for (const id of ids) if (!messageById.has(id)) results.set(id, { skipped: true });
|
||||
if (businessMessages.length === 0) return results;
|
||||
|
||||
const { planned, failed } = await this.planRoutesBatch(businessMessages);
|
||||
if (failed.length > 0) await this.failRouteBatch(failed, results);
|
||||
if (planned.length === 0) return results;
|
||||
|
||||
await Promise.all(planned.map(({ routed }) => (
|
||||
this.facade.waitForChannelRateLimit(routed.channel.id, routed.channel.rateLimitPerSecond)
|
||||
)));
|
||||
const sessionByChannel = new Map<string, string>();
|
||||
await Promise.all([...new Set(planned.map(({ routed }) => routed.channel.id))].map(async (channelId) => {
|
||||
sessionByChannel.set(channelId, await this.getOpenSubmitSessionId(channelId));
|
||||
}));
|
||||
const prepared = planned.map(({ message, routed }) => {
|
||||
const submitId = `SUB-${randomUUID()}`;
|
||||
const upstreamSrcId = composeUpstreamSrcId(routed.channel.srcId, message.applicationExtension);
|
||||
return {
|
||||
message,
|
||||
routed,
|
||||
submitId,
|
||||
command: this.buildGatewaySubmitCommand(message, routed, 0, submitId, upstreamSrcId),
|
||||
sessionId: sessionByChannel.get(routed.channel.id),
|
||||
};
|
||||
});
|
||||
const writeOutbox = this.submitOutboxEnabled();
|
||||
await this.measureSendStage('submit_transaction', () => this.prisma.$transaction(async (tx) => {
|
||||
await tx.smsSubmitRecord.createMany({
|
||||
data: prepared.map(({ message, routed, submitId, sessionId }) => ({
|
||||
id: randomUUID(),
|
||||
tenantId: message.tenantId,
|
||||
batchTaskId: message.batchTaskId,
|
||||
messageRecordId: message.id,
|
||||
channelId: routed.channel.id,
|
||||
channelGroupId: routed.groupId,
|
||||
channelGroupName: routed.groupName,
|
||||
sessionId,
|
||||
submitId,
|
||||
submitStatus: 'queued',
|
||||
costUnitPrice: routed.channel.unitPrice ?? 0,
|
||||
costAmountCents: moneyToNumber(routed.channel.unitPrice) * Math.max(1, message.billingUnits ?? 1),
|
||||
})),
|
||||
});
|
||||
const updates = Prisma.join(prepared.map(({ message, routed, submitId }) => Prisma.sql`(
|
||||
${message.id}::text, ${routed.channel.id}::text, ${routed.carrier}::text,
|
||||
${routed.province ?? null}::text, ${submitId}::text
|
||||
)`));
|
||||
await tx.$executeRaw(Prisma.sql`
|
||||
UPDATE "SmsMessageRecord" AS message
|
||||
SET "channelId" = updates."channelId",
|
||||
carrier = updates.carrier,
|
||||
province = updates.province,
|
||||
"submitId" = updates."submitId",
|
||||
status = 'submit_queued',
|
||||
"submitStatus" = 'queued',
|
||||
"receiptStatus" = NULL,
|
||||
"errorCode" = NULL,
|
||||
"errorMessage" = NULL,
|
||||
"updatedAt" = CURRENT_TIMESTAMP
|
||||
FROM (VALUES ${updates}) AS updates(id, "channelId", carrier, province, "submitId")
|
||||
WHERE message.id = updates.id AND message.status = 'queued'
|
||||
`);
|
||||
if (writeOutbox) {
|
||||
await tx.gatewaySubmitOutbox.createMany({
|
||||
data: prepared.map(({ message, routed, submitId, command }) => ({
|
||||
id: randomUUID(), submitId, messageRecordId: message.id,
|
||||
channelId: routed.channel.id, payload: command as Prisma.InputJsonValue,
|
||||
})),
|
||||
});
|
||||
}
|
||||
}));
|
||||
if (!this.submitOutboxPublishEnabled()) {
|
||||
await Promise.all(prepared.map(({ command }) => this.facade.publishGatewaySubmitCommand(command)));
|
||||
}
|
||||
await this.refreshTaskProgressBatch(prepared.map(({ message }) => message));
|
||||
for (const { message, routed, submitId } of prepared) {
|
||||
results.set(message.id, {
|
||||
submitted: true, messageRecordId: message.id, channelId: routed.channel.id, attempt: 0, submitId,
|
||||
});
|
||||
this.metrics?.recordSendWorkerResult('completed');
|
||||
}
|
||||
return results;
|
||||
}
|
||||
|
||||
private async planRoutesBatch<T extends {
|
||||
id: string; tenantId: string; batchTaskId: string; applicationId?: string | null;
|
||||
templateId?: string | null; signatureId?: string | null; phoneNumber: string;
|
||||
carrier?: string | null; province?: string | null;
|
||||
template?: { signature?: { id?: string | null } | null } | null;
|
||||
signature?: { id?: string | null } | null;
|
||||
}>(messages: T[]) {
|
||||
const unresolvedPhones = messages.filter((message) => !message.carrier).map((message) => message.phoneNumber);
|
||||
const provinces = await this.measureSendStage('phone_routing', () => this.phoneRouting.identifyProvinces(unresolvedPhones));
|
||||
const routeInputs = await Promise.all(messages.map(async (message) => ({
|
||||
message,
|
||||
carrier: message.carrier ? normalizeCarrier(message.carrier) : normalizeCarrier(await this.phoneRouting.identifyCarrier(message.phoneNumber)),
|
||||
province: message.carrier ? message.province ?? null : provinces.get(message.phoneNumber) ?? null,
|
||||
signatureId: message.signatureId ?? message.template?.signature?.id ?? message.signature?.id ?? null,
|
||||
})));
|
||||
const valid = routeInputs.filter((input) => input.message.applicationId && input.signatureId);
|
||||
const signatures = [...new Set(valid.map((input) => input.signatureId as string))];
|
||||
const routes = valid.length === 0 ? [] : await this.measureSendStage('route_lookup', () => this.prisma.channelRouteRule.findMany({
|
||||
where: {
|
||||
status: 'active', channelId: null, province: null,
|
||||
OR: valid.map((input) => ({
|
||||
tenantId: input.message.tenantId,
|
||||
applicationId: input.message.applicationId,
|
||||
carrier: input.carrier,
|
||||
})),
|
||||
},
|
||||
include: {
|
||||
group: {
|
||||
include: {
|
||||
items: {
|
||||
include: {
|
||||
channel: {
|
||||
include: {
|
||||
connectionStates: { where: { status: 'connected', currentConnections: { gt: 0 } } },
|
||||
reportTasks: { where: { signatureId: { in: signatures }, reportType: 'signature', status: 'approved' } },
|
||||
},
|
||||
},
|
||||
},
|
||||
orderBy: { priority: 'asc' },
|
||||
},
|
||||
},
|
||||
},
|
||||
},
|
||||
orderBy: { priority: 'asc' },
|
||||
}));
|
||||
const routeByKey = new Map<string, typeof routes[number]>();
|
||||
for (const route of routes) {
|
||||
const key = `${route.tenantId}:${route.applicationId}:${normalizeCarrier(route.carrier)}`;
|
||||
if (!routeByKey.has(key)) routeByKey.set(key, route);
|
||||
}
|
||||
const planned: Array<{ message: T; routed: RoutedChannel }> = [];
|
||||
const failed: Array<{ message: T; reason: string }> = [];
|
||||
for (const input of routeInputs) {
|
||||
if (!input.message.applicationId) {
|
||||
failed.push({ message: input.message, reason: '短信应用未配置，无法选择通道组' });
|
||||
continue;
|
||||
}
|
||||
if (!input.signatureId) {
|
||||
failed.push({ message: input.message, reason: '短信签名未配置，无法选择已报备通道' });
|
||||
continue;
|
||||
}
|
||||
const route = routeByKey.get(`${input.message.tenantId}:${input.message.applicationId}:${input.carrier}`);
|
||||
if (!route) {
|
||||
failed.push({ message: input.message, reason: '企业应用未配置对应运营商通道组' });
|
||||
continue;
|
||||
}
|
||||
if (route.group.status !== 'active' || normalizeCarrier(route.group.carrier) !== input.carrier) {
|
||||
failed.push({ message: input.message, reason: '企业应用绑定的通道组已停用或运营商不一致' });
|
||||
continue;
|
||||
}
|
||||
const approvedItems = route.group.items.filter((item) => item.channel.status === 'active'
|
||||
&& item.channel.connectionStates.length > 0
|
||||
&& item.channel.reportTasks.some((task) => task.signatureId === input.signatureId
|
||||
&& (task.carrier === input.carrier
|
||||
|| (process.env.SIGNATURE_REPORT_STRICT_CARRIER !== 'true' && task.approvalScope === 'legacy_channel'))));
|
||||
const selected = selectChannelCandidate(approvedItems, {
|
||||
carrier: input.carrier,
|
||||
province: input.province,
|
||||
excludedChannelIds: new Set(),
|
||||
approvedChannelIds: new Set(approvedItems.map((item) => item.channelId)),
|
||||
routingKey: input.message.id,
|
||||
});
|
||||
if (!selected) {
|
||||
failed.push({ message: input.message, reason: '无已报备通过且在线的可用通道' });
|
||||
continue;
|
||||
}
|
||||
planned.push({
|
||||
message: input.message,
|
||||
routed: {
|
||||
channel: { ...selected.channel, unitPrice: moneyToNumber(selected.channel.unitPrice) },
|
||||
carrier: input.carrier, province: input.province,
|
||||
groupId: route.groupId, groupName: route.group.name,
|
||||
routeScope: isNationalChannel(selected) ? 'national' : 'province',
|
||||
},
|
||||
});
|
||||
}
|
||||
return { planned, failed };
|
||||
}
|
||||
|
||||
private async failRouteBatch<T extends {
|
||||
id: string; tenantId: string; batchTaskId: string; applicationId?: string | null;
|
||||
messageId: string; phoneNumber: string; amountCents: bigint; billingUnits: number;
|
||||
cmppSubmitSequenceId?: string | null; cmppSubmitGroupMessageId?: string | null;
|
||||
batchTask?: { sourceType?: string | null; phoneTotal?: number | null } | null;
|
||||
}>(failed: Array<{ message: T; reason: string }>, results: Map<string, unknown>) {
|
||||
const values = Prisma.join(failed.map(({ message, reason }) => Prisma.sql`(${message.id}::text, ${reason.slice(0, 1000)}::text)`));
|
||||
await this.prisma.$executeRaw(Prisma.sql`
|
||||
UPDATE "SmsMessageRecord" AS message
|
||||
SET status = 'failed', "errorMessage" = failures.reason, "updatedAt" = CURRENT_TIMESTAMP
|
||||
FROM (VALUES ${values}) AS failures(id, reason)
|
||||
WHERE message.id = failures.id AND message.status = 'queued'
|
||||
`);
|
||||
await Promise.all(failed.map(async ({ message, reason }) => {
|
||||
await this.releaseMessageReservation(message, reason);
|
||||
if (message.batchTask?.sourceType === 'cmpp') await this.recordCmppFailureReceipt(message, 'ROUTE', reason);
|
||||
else await this.facade.refreshTaskProgress(message.batchTaskId);
|
||||
results.set(message.id, { submitted: false, messageRecordId: message.id, status: 'failed', reason });
|
||||
this.metrics?.recordSendWorkerResult('failed');
|
||||
}));
|
||||
}
|
||||
|
||||
private async refreshTaskProgressBatch(messages: Array<{
|
||||
batchTaskId: string; batchTask?: { sourceType?: string | null; phoneTotal?: number | null } | null;
|
||||
}>) {
|
||||
const singleCmppIds = [...new Set(messages
|
||||
.filter((message) => message.batchTask?.sourceType === 'cmpp' && message.batchTask.phoneTotal === 1)
|
||||
.map((message) => message.batchTaskId))];
|
||||
if (singleCmppIds.length > 0) {
|
||||
await this.prisma.smsBatchTask.updateMany({
|
||||
where: { id: { in: singleCmppIds }, sourceType: 'cmpp', phoneTotal: 1 },
|
||||
data: singleMessageTaskProgress('submit_queued'),
|
||||
});
|
||||
}
|
||||
const otherTaskIds = [...new Set(messages
|
||||
.filter((message) => !singleCmppIds.includes(message.batchTaskId))
|
||||
.map((message) => message.batchTaskId))];
|
||||
await Promise.all(otherTaskIds.map((taskId) => this.facade.refreshTaskProgress(taskId)));
|
||||
}
|
||||
|
||||
async processSendJob(job: SendJob) {
|
||||
const totalStartedAt = this.metrics?.beginSendWorkerStage();
|
||||
let totalFinished = false;
|
||||
const finish = (result: 'completed' | 'failed' | 'skipped') => {
|
||||
if (totalFinished) return;
|
||||
totalFinished = true;
|
||||
if (totalStartedAt != null) {
|
||||
this.metrics?.finishSendWorkerStage(totalStartedAt, 'total', result === 'completed' ? 'success' : result === 'skipped' ? 'skipped' : 'error');
|
||||
}
|
||||
this.metrics?.recordSendWorkerResult(result);
|
||||
};
|
||||
try {
|
||||
const message = await this.measureSendStage('message_load', () => this.prisma.smsMessageRecord.findUnique({
|
||||
where: { id: job.messageRecordId },
|
||||
include: { batchTask: true, template: { include: { signature: true } }, signature: true },
|
||||
}));
|
||||
if (!message || message.status !== 'queued') {
|
||||
finish('skipped');
|
||||
return { skipped: true };
|
||||
}
|
||||
if (!message.tenantId || !message.batchTaskId) {
|
||||
finish('skipped');
|
||||
return { skipped: true, reason: 'standalone channel test message' };
|
||||
}
|
||||
const businessMessage = message as typeof message & { tenantId: string; batchTaskId: string };
|
||||
try {
|
||||
const routed = await this.facade.selectChannelForMessage(businessMessage);
|
||||
const result = await this.facade.submitMessageToGateway(businessMessage, routed, 0);
|
||||
finish(result.submitted ? 'completed' : 'skipped');
|
||||
return result;
|
||||
} catch (error) {
|
||||
const reason = error instanceof Error ? error.message : '无可用通道组或通道';
|
||||
await this.prisma.smsMessageRecord.update({
|
||||
where: { id: message.id },
|
||||
data: { status: 'failed', errorMessage: reason },
|
||||
});
|
||||
await this.releaseMessageReservation(businessMessage, reason);
|
||||
if (message.batchTask?.sourceType === 'cmpp') {
|
||||
await this.recordCmppFailureReceipt(businessMessage, 'ROUTE', reason);
|
||||
} else {
|
||||
await this.facade.refreshTaskProgress(
|
||||
businessMessage.batchTaskId,
|
||||
message.batchTask?.sourceType === 'cmpp' && message.batchTask.phoneTotal === 1 ? 'failed' : undefined,
|
||||
);
|
||||
}
|
||||
finish('failed');
|
||||
return { submitted: false, messageRecordId: message.id, status: 'failed', reason };
|
||||
}
|
||||
} catch (error) {
|
||||
finish('failed');
|
||||
throw error;
|
||||
}
|
||||
}
|
||||
|
||||
async submitMessageToGateway(
|
||||
message: {
|
||||
id: string;
|
||||
tenantId: string;
|
||||
@@ -169,6 +535,7 @@ async submitMessageToGateway(
|
||||
applicationExtension?: string | null;
|
||||
template?: { signature?: { id?: string | null; name?: string | null } | null } | null;
|
||||
signature?: { id?: string | null; name?: string | null } | null;
|
||||
batchTask?: { sourceType?: string | null; phoneTotal?: number | null } | null;
|
||||
},
|
||||
routed: RoutedChannel,
|
||||
attempt: number,
|
||||
@@ -176,16 +543,13 @@ async submitMessageToGateway(
|
||||
) {
|
||||
const channel = routed.channel;
|
||||
const upstreamSrcId = composeUpstreamSrcId(channel.srcId, message.applicationExtension);
|
||||
await this.facade.ensureSignatureReportedForChannel(message, channel.id, routed.carrier);
|
||||
await this.facade.waitForChannelRateLimit(channel.id, channel.rateLimitPerSecond);
|
||||
await this.measureSendStage('rate_limit', () => this.facade.waitForChannelRateLimit(channel.id, channel.rateLimitPerSecond));
|
||||
const submitId = `SUB-${randomUUID()}`;
|
||||
const sessionId = await this.getOpenSubmitSessionId(channel.id);
|
||||
const command = this.buildGatewaySubmitCommand(message, routed, attempt, submitId, upstreamSrcId);
|
||||
const writeOutbox = this.submitOutboxEnabled();
|
||||
try {
|
||||
await this.prisma.$transaction(async (tx) => {
|
||||
const session = await tx.cmppSubmitSession.upsert({
|
||||
where: { sessionNo: `OPEN-${channel.id}` },
|
||||
update: { submitTotal: { increment: 1 } },
|
||||
create: { channelId: channel.id, sessionNo: `OPEN-${channel.id}`, submitTotal: 1 },
|
||||
});
|
||||
await this.measureSendStage('submit_transaction', () => this.prisma.$transaction(async (tx) => {
|
||||
await tx.smsSubmitRecord.create({
|
||||
data: {
|
||||
tenantId: message.tenantId,
|
||||
@@ -194,7 +558,7 @@ async submitMessageToGateway(
|
||||
channelId: channel.id,
|
||||
channelGroupId: routed.groupId,
|
||||
channelGroupName: routed.groupName,
|
||||
sessionId: session.id,
|
||||
sessionId,
|
||||
retryOfSubmitRecordId,
|
||||
submitId,
|
||||
submitStatus: 'queued',
|
||||
@@ -216,7 +580,17 @@ async submitMessageToGateway(
|
||||
errorMessage: attempt > 0 ? `第 ${attempt + 1} 次提交，路由至${routed.routeScope === 'national' ? '全国' : '省网'}通道` : undefined,
|
||||
},
|
||||
});
|
||||
});
|
||||
if (writeOutbox) {
|
||||
await tx.gatewaySubmitOutbox.create({
|
||||
data: {
|
||||
submitId,
|
||||
messageRecordId: message.id,
|
||||
channelId: channel.id,
|
||||
payload: command as Prisma.InputJsonValue,
|
||||
},
|
||||
});
|
||||
}
|
||||
}));
|
||||
if (retryOfSubmitRecordId) {
|
||||
this.logger.log(`sms_retry_claim_acquired ${JSON.stringify({
|
||||
messageId: message.messageId,
|
||||
@@ -255,7 +629,31 @@ async submitMessageToGateway(
|
||||
}
|
||||
throw error;
|
||||
}
|
||||
const command = {
|
||||
if (!this.submitOutboxPublishEnabled()) {
|
||||
await this.measureSendStage('gateway_stream_publish', () => this.facade.publishGatewaySubmitCommand(command));
|
||||
}
|
||||
await this.measureSendStage('task_progress', () => this.facade.refreshTaskProgress(
|
||||
message.batchTaskId,
|
||||
message.batchTask?.sourceType === 'cmpp' && message.batchTask.phoneTotal === 1 ? 'submit_queued' : undefined,
|
||||
));
|
||||
return { submitted: true, messageRecordId: message.id, channelId: channel.id, attempt };
|
||||
}
|
||||
|
||||
private buildGatewaySubmitCommand(
|
||||
message: {
|
||||
id: string; tenantId: string; batchTaskId: string; applicationId?: string | null;
|
||||
templateId?: string | null; messageId: string; phoneNumber: string; content: string;
|
||||
billingUnits: number; queuePriority?: string | null; applicationExtension?: string | null;
|
||||
template?: { signature?: { name?: string | null } | null } | null;
|
||||
signature?: { name?: string | null } | null;
|
||||
},
|
||||
routed: RoutedChannel,
|
||||
attempt: number,
|
||||
submitId: string,
|
||||
upstreamSrcId: string,
|
||||
) {
|
||||
const channel = routed.channel;
|
||||
return {
|
||||
schemaVersion: 'v1',
|
||||
messageType: 'SubmitCommand',
|
||||
traceId: randomUUID(),
|
||||
@@ -304,10 +702,124 @@ async submitMessageToGateway(
|
||||
},
|
||||
retry: { attempt, maxAttempts: 1 },
|
||||
};
|
||||
await this.facade.getGatewayQueue().add('submit-command', command);
|
||||
await this.facade.publishGatewaySubmitCommand(command);
|
||||
await this.facade.refreshTaskProgress(message.batchTaskId);
|
||||
return { submitted: true, messageRecordId: message.id, channelId: channel.id, attempt };
|
||||
}
|
||||
|
||||
private submitOutboxEnabled() {
|
||||
return process.env.SEND_SUBMIT_OUTBOX_SHADOW_ENABLED === 'true'
|
||||
|| process.env.SEND_SUBMIT_OUTBOX_PUBLISH_ENABLED === 'true';
|
||||
}
|
||||
|
||||
private submitOutboxPublishEnabled() {
|
||||
return process.env.SEND_SUBMIT_OUTBOX_PUBLISH_ENABLED === 'true';
|
||||
}
|
||||
|
||||
private async publishSubmitOutboxBatch() {
|
||||
if (this.submitOutboxRunning || !this.submitOutboxEnabled()) return;
|
||||
this.submitOutboxRunning = true;
|
||||
try {
|
||||
const batchSize = Math.min(500, getPositiveConfigInteger(process.env, 'SEND_SUBMIT_OUTBOX_BATCH_SIZE', 64));
|
||||
const leaseSeconds = Math.min(300, getPositiveConfigInteger(process.env, 'SEND_SUBMIT_OUTBOX_LEASE_SECONDS', 30));
|
||||
const rows = await this.prisma.$queryRaw<Array<{ id: string; submitId: string; payload: Prisma.JsonValue }>>(Prisma.sql`
|
||||
WITH candidates AS (
|
||||
SELECT id
|
||||
FROM "GatewaySubmitOutbox"
|
||||
WHERE (
|
||||
(status = 'pending' AND "nextAttemptAt" <= CURRENT_TIMESTAMP)
|
||||
OR (status = 'publishing' AND "leaseExpiresAt" < CURRENT_TIMESTAMP)
|
||||
)
|
||||
ORDER BY "createdAt", id
|
||||
LIMIT ${batchSize}
|
||||
FOR UPDATE SKIP LOCKED
|
||||
)
|
||||
UPDATE "GatewaySubmitOutbox" AS outbox
|
||||
SET status = 'publishing',
|
||||
"leaseOwner" = ${this.submitOutboxLeaseOwner},
|
||||
"leaseExpiresAt" = CURRENT_TIMESTAMP + (${leaseSeconds} * INTERVAL '1 second'),
|
||||
"attemptCount" = outbox."attemptCount" + 1,
|
||||
"updatedAt" = CURRENT_TIMESTAMP
|
||||
FROM candidates
|
||||
WHERE outbox.id = candidates.id
|
||||
RETURNING outbox.id, outbox."submitId", outbox.payload
|
||||
`);
|
||||
if (rows.length === 0) return;
|
||||
const results = this.submitOutboxPublishEnabled()
|
||||
? await this.publishGatewaySubmitCommandBatch(rows)
|
||||
: rows.map((row) => ({ row, streamEntryId: `shadow:${row.submitId}` }));
|
||||
const succeeded = results.filter((result): result is { row: typeof rows[number]; streamEntryId: string } => 'streamEntryId' in result);
|
||||
if (succeeded.length > 0) {
|
||||
const values = Prisma.join(succeeded.map(({ row, streamEntryId }) => Prisma.sql`(${row.id}, ${streamEntryId})`));
|
||||
await this.prisma.$executeRaw(Prisma.sql`
|
||||
UPDATE "GatewaySubmitOutbox" AS outbox
|
||||
SET status = 'published',
|
||||
"streamEntryId" = published."streamEntryId",
|
||||
"publishedAt" = CURRENT_TIMESTAMP,
|
||||
"leaseOwner" = NULL,
|
||||
"leaseExpiresAt" = NULL,
|
||||
"lastError" = NULL,
|
||||
"updatedAt" = CURRENT_TIMESTAMP
|
||||
FROM (VALUES ${values}) AS published(id, "streamEntryId")
|
||||
WHERE outbox.id = published.id
|
||||
AND outbox.status = 'publishing'
|
||||
AND outbox."leaseOwner" = ${this.submitOutboxLeaseOwner}
|
||||
`);
|
||||
}
|
||||
for (const result of results) {
|
||||
if ('streamEntryId' in result) continue;
|
||||
const { row, error } = result;
|
||||
try {
|
||||
const message = error instanceof Error ? error.message : String(error);
|
||||
await this.prisma.$executeRaw(Prisma.sql`
|
||||
UPDATE "GatewaySubmitOutbox"
|
||||
SET status = CASE WHEN "attemptCount" >= 10 THEN 'dead' ELSE 'pending' END,
|
||||
"nextAttemptAt" = CURRENT_TIMESTAMP + (LEAST(60, POWER(2, LEAST("attemptCount", 6))) * INTERVAL '1 second'),
|
||||
"leaseOwner" = NULL,
|
||||
"leaseExpiresAt" = NULL,
|
||||
"lastError" = ${message.slice(0, 1000)},
|
||||
"updatedAt" = CURRENT_TIMESTAMP
|
||||
WHERE id = ${row.id} AND "leaseOwner" = ${this.submitOutboxLeaseOwner}
|
||||
`);
|
||||
} catch (recordError) {
|
||||
this.logger.error(`gateway_submit_outbox_failure_record_failed ${recordError instanceof Error ? recordError.message : String(recordError)}`);
|
||||
}
|
||||
}
|
||||
} catch (error) {
|
||||
this.logger.error(`gateway_submit_outbox_publish_failed ${error instanceof Error ? error.message : String(error)}`);
|
||||
} finally {
|
||||
this.submitOutboxRunning = false;
|
||||
}
|
||||
}
|
||||
|
||||
private async publishGatewaySubmitCommandBatch(
|
||||
rows: Array<{ id: string; submitId: string; payload: Prisma.JsonValue }>,
|
||||
): Promise<Array<
|
||||
{ row: typeof rows[number]; streamEntryId: string }
|
||||
| { row: typeof rows[number]; error: unknown }
|
||||
>> {
|
||||
const redis = this.facade.getRedis();
|
||||
const stream = process.env.GATEWAY_SUBMIT_STREAM ?? GATEWAY_SUBMIT_STREAM;
|
||||
const script = `local existing = redis.call('GET', KEYS[2])
|
||||
if existing then return existing end
|
||||
local streamId = redis.call('XADD', KEYS[1], '*', 'messageType', 'SubmitCommand', 'data', ARGV[1])
|
||||
redis.call('SET', KEYS[2], streamId, 'EX', ARGV[2])
|
||||
return streamId`;
|
||||
const pipeline = redis.pipeline();
|
||||
for (const row of rows) {
|
||||
pipeline.eval(
|
||||
script,
|
||||
2,
|
||||
stream,
|
||||
`gateway:submit:outbox:${row.submitId}`,
|
||||
JSON.stringify(row.payload),
|
||||
String(GATEWAY_SUBMIT_REQUEUE_IDEMPOTENCY_TTL_SECONDS),
|
||||
);
|
||||
}
|
||||
const replies = await pipeline.exec();
|
||||
if (!replies || replies.length !== rows.length) {
|
||||
return rows.map((row) => ({ row, error: new Error('Redis Outbox pipeline result count mismatch') }));
|
||||
}
|
||||
return replies.map(([error, value], index) => error
|
||||
? { row: rows[index], error }
|
||||
: { row: rows[index], streamEntryId: typeof value === 'string' ? value : String(value ?? '') });
|
||||
}
|
||||
|
||||
async selectChannelForMessage(
|
||||
@@ -318,33 +830,31 @@ async selectChannelForMessage(
|
||||
throw new BadRequestException('短信应用未配置，无法选择通道组');
|
||||
}
|
||||
const hasPersistedRouting = Boolean(message.carrier);
|
||||
const [carrier, province] = hasPersistedRouting
|
||||
? [normalizeCarrier(message.carrier), message.province ?? null]
|
||||
: await Promise.all([
|
||||
this.facade.identifyCarrier(message.phoneNumber),
|
||||
this.facade.identifyProvince(message.phoneNumber),
|
||||
]);
|
||||
if (!hasPersistedRouting) {
|
||||
await this.prisma.smsMessageRecord.update({
|
||||
where: { id: message.id },
|
||||
data: { carrier, province },
|
||||
});
|
||||
}
|
||||
const route = await this.facade.findApplicationRoute(message.tenantId, message.applicationId, carrier);
|
||||
const excluded = new Set(options.excludeChannelIds ?? []);
|
||||
const signatureId = await this.facade.resolveMessageSignatureId(message);
|
||||
if (!signatureId) throw new BadRequestException('短信签名未配置，无法选择已报备通道');
|
||||
const approvedTasks = await this.prisma.channelSignatureReportTask.findMany({
|
||||
where: {
|
||||
signatureId,
|
||||
reportType: 'signature',
|
||||
status: 'approved',
|
||||
channelId: { in: route.group.items.map((item) => item.channelId) },
|
||||
OR: signatureReportApprovalScopes(carrier),
|
||||
},
|
||||
select: { channelId: true },
|
||||
const [carrier, province] = await this.measureSendStage('phone_routing', async () => {
|
||||
const resolved = hasPersistedRouting
|
||||
? [normalizeCarrier(message.carrier), message.province ?? null] as const
|
||||
: await Promise.all([
|
||||
this.facade.identifyCarrier(message.phoneNumber),
|
||||
this.facade.identifyProvince(message.phoneNumber),
|
||||
]);
|
||||
if (!hasPersistedRouting) {
|
||||
await this.prisma.smsMessageRecord.update({
|
||||
where: { id: message.id },
|
||||
data: { carrier: resolved[0], province: resolved[1] },
|
||||
});
|
||||
}
|
||||
return resolved;
|
||||
});
|
||||
const approvedChannelIds = new Set(approvedTasks.map((task) => task.channelId));
|
||||
const signatureId = await this.measureSendStage('signature_candidates', () => this.facade.resolveMessageSignatureId(message));
|
||||
if (!signatureId) throw new BadRequestException('短信签名未配置，无法选择已报备通道');
|
||||
const route = await this.measureSendStage('route_lookup', () => this.facade.findApplicationRoute(
|
||||
message.tenantId,
|
||||
message.applicationId ?? undefined,
|
||||
carrier,
|
||||
signatureId,
|
||||
));
|
||||
const excluded = new Set(options.excludeChannelIds ?? []);
|
||||
const approvedChannelIds = new Set(route.group.items.map((item) => item.channelId));
|
||||
const selected = selectChannelCandidate(route.group.items, {
|
||||
carrier,
|
||||
province,
|
||||
@@ -366,7 +876,55 @@ async selectChannelForMessage(
|
||||
};
|
||||
}
|
||||
|
||||
async findApplicationRoute(tenantId: string, applicationId: string | undefined, carrier: string) {
|
||||
private async measureSendStage<T>(stage: SendWorkerStage, operation: () => Promise<T>): Promise<T> {
|
||||
const startedAt = this.metrics?.beginSendWorkerStage();
|
||||
try {
|
||||
const result = await operation();
|
||||
if (startedAt != null) this.metrics?.finishSendWorkerStage(startedAt, stage, 'success');
|
||||
return result;
|
||||
} catch (error) {
|
||||
if (startedAt != null) this.metrics?.finishSendWorkerStage(startedAt, stage, 'error');
|
||||
throw error;
|
||||
}
|
||||
}
|
||||
|
||||
private async refreshSendQueueMetrics() {
|
||||
if (!this.metrics) return;
|
||||
try {
|
||||
const counts = await this.facade.getSendQueue().getJobCounts('wait', 'active', 'completed', 'failed', 'delayed', 'prioritized');
|
||||
const mappings: Array<[SendWorkerQueueState, number]> = [
|
||||
['waiting', counts.wait ?? 0],
|
||||
['active', counts.active ?? 0],
|
||||
['completed', counts.completed ?? 0],
|
||||
['failed', counts.failed ?? 0],
|
||||
['delayed', counts.delayed ?? 0],
|
||||
['prioritized', counts.prioritized ?? 0],
|
||||
];
|
||||
for (const [state, count] of mappings) this.metrics.setSendWorkerQueueJobs(state, count);
|
||||
const pool = this.prisma.getPoolState();
|
||||
for (const state of ['max', 'total', 'idle', 'waiting'] as const) {
|
||||
this.metrics.setSendWorkerDatabasePool(state, pool[state]);
|
||||
}
|
||||
} catch (error) {
|
||||
this.logger.warn(`send_queue_metrics_refresh_failed ${error instanceof Error ? error.message : String(error)}`);
|
||||
}
|
||||
}
|
||||
|
||||
async findApplicationRoute(tenantId: string, applicationId: string | undefined, carrier: string, signatureId?: string) {
|
||||
const approvedChannelWhere = signatureId
|
||||
? {
|
||||
status: 'active',
|
||||
connectionStates: { some: { status: 'connected', currentConnections: { gt: 0 } } },
|
||||
reportTasks: {
|
||||
some: {
|
||||
signatureId,
|
||||
reportType: 'signature',
|
||||
status: 'approved',
|
||||
OR: signatureReportApprovalScopes(carrier),
|
||||
},
|
||||
},
|
||||
}
|
||||
: undefined;
|
||||
const route = await this.prisma.channelRouteRule.findFirst({
|
||||
where: {
|
||||
status: 'active',
|
||||
@@ -376,7 +934,25 @@ async findApplicationRoute(tenantId: string, applicationId: string | undefined,
|
||||
channelId: null,
|
||||
province: null,
|
||||
},
|
||||
include: { group: { include: { items: { include: { channel: { include: { connectionStates: true } } }, orderBy: { priority: 'asc' } } } } },
|
||||
include: {
|
||||
group: {
|
||||
include: {
|
||||
items: {
|
||||
where: approvedChannelWhere ? { channel: approvedChannelWhere } : undefined,
|
||||
include: {
|
||||
channel: {
|
||||
include: {
|
||||
connectionStates: approvedChannelWhere
|
||||
? { where: { status: 'connected', currentConnections: { gt: 0 } } }
|
||||
: true,
|
||||
},
|
||||
},
|
||||
},
|
||||
orderBy: { priority: 'asc' },
|
||||
},
|
||||
},
|
||||
},
|
||||
},
|
||||
orderBy: { priority: 'asc' },
|
||||
});
|
||||
if (!route) {
|
||||
@@ -450,7 +1026,40 @@ async waitForChannelRateLimit(channelId: string, tps: number) {
|
||||
}
|
||||
}
|
||||
|
||||
async refreshTaskProgress(batchTaskId: string) {
|
||||
async refreshTaskProgress(batchTaskId: string, knownSingleMessageStatus?: string) {
|
||||
if (knownSingleMessageStatus) {
|
||||
const direct = await this.prisma.smsBatchTask.updateMany({
|
||||
where: { id: batchTaskId, sourceType: 'cmpp', phoneTotal: 1 },
|
||||
data: singleMessageTaskProgress(knownSingleMessageStatus),
|
||||
});
|
||||
if (direct.count === 1) return;
|
||||
} else {
|
||||
// Gateway结果、回执和超时回调已经先提交了消息状态。对CMPP单号码内部任务，
|
||||
// 在同一条UPDATE中读取该唯一消息的当前状态并写入精确计数，避免每次回调都
|
||||
// 对一个只有一行的任务执行GROUP BY；多号码任务继续使用下方聚合路径。
|
||||
const direct = await this.prisma.$executeRaw(Prisma.sql`
|
||||
UPDATE "SmsBatchTask" AS task
|
||||
SET
|
||||
"progressTotal" = 1,
|
||||
"submittedTotal" = CASE WHEN message.status IN ('submit_queued', 'submitted', 'delivered', 'failed', 'unknown', 'timeout') THEN 1 ELSE 0 END,
|
||||
"successTotal" = CASE WHEN message.status = 'delivered' THEN 1 ELSE 0 END,
|
||||
"failedTotal" = CASE WHEN message.status IN ('submit_failed', 'failed') THEN 1 ELSE 0 END,
|
||||
"unknownTotal" = CASE WHEN message.status = 'unknown' THEN 1 ELSE 0 END,
|
||||
"timeoutTotal" = CASE WHEN message.status = 'timeout' THEN 1 ELSE 0 END,
|
||||
status = CASE
|
||||
WHEN message.status IN ('delivered', 'submit_failed', 'failed', 'timeout') THEN 'finished'
|
||||
WHEN message.status IN ('submit_queued', 'submitted', 'unknown') THEN 'sending'
|
||||
ELSE 'queued'
|
||||
END,
|
||||
"updatedAt" = CURRENT_TIMESTAMP
|
||||
FROM "SmsMessageRecord" AS message
|
||||
WHERE task.id = ${batchTaskId}
|
||||
AND task."sourceType" = 'cmpp'
|
||||
AND task."phoneTotal" = 1
|
||||
AND message."batchTaskId" = task.id
|
||||
`);
|
||||
if (direct === 1) return;
|
||||
}
|
||||
const running = this.taskProgressRefreshes.get(batchTaskId);
|
||||
if (running) {
|
||||
// A state transition committed after the running aggregate may not be visible
|
||||
@@ -519,7 +1128,7 @@ getRedis() {
|
||||
return this.redis;
|
||||
}
|
||||
|
||||
async publishGatewaySubmitCommand(command: unknown, idempotencyKey?: string) {
|
||||
async publishGatewaySubmitCommand(command: unknown, idempotencyKey?: string) {
|
||||
const redis = this.facade.getRedis();
|
||||
const stream = process.env.GATEWAY_SUBMIT_STREAM ?? GATEWAY_SUBMIT_STREAM;
|
||||
const payload = JSON.stringify(command);
|
||||
@@ -540,6 +1149,44 @@ return streamId`,
|
||||
);
|
||||
return typeof result === 'string' ? result : String(result ?? '');
|
||||
}
|
||||
|
||||
private getOpenSubmitSessionId(channelId: string) {
|
||||
const cached = this.openSubmitSessionIds.get(channelId);
|
||||
if (cached) return cached;
|
||||
const sessionNo = `OPEN-${channelId}`;
|
||||
const pending = this.prisma.cmppSubmitSession.findUnique({
|
||||
where: { sessionNo },
|
||||
select: { id: true },
|
||||
}).then((existing) => existing ?? this.prisma.cmppSubmitSession.upsert({
|
||||
where: { sessionNo },
|
||||
update: {},
|
||||
create: { channelId, sessionNo, submitTotal: 0 },
|
||||
select: { id: true },
|
||||
})).then((session) => session.id).catch((error) => {
|
||||
this.openSubmitSessionIds.delete(channelId);
|
||||
throw error;
|
||||
});
|
||||
this.openSubmitSessionIds.set(channelId, pending);
|
||||
return pending;
|
||||
}
|
||||
}
|
||||
|
||||
function singleMessageTaskProgress(status: string) {
|
||||
const submittedTotal = ['submit_queued', 'submitted', 'delivered', 'failed', 'unknown', 'timeout'].includes(status) ? 1 : 0;
|
||||
const successTotal = status === 'delivered' ? 1 : 0;
|
||||
const failedTotal = ['submit_failed', 'failed'].includes(status) ? 1 : 0;
|
||||
const unknownTotal = status === 'unknown' ? 1 : 0;
|
||||
const timeoutTotal = status === 'timeout' ? 1 : 0;
|
||||
const doneTotal = successTotal + failedTotal + timeoutTotal;
|
||||
return {
|
||||
progressTotal: 1,
|
||||
submittedTotal,
|
||||
successTotal,
|
||||
failedTotal,
|
||||
unknownTotal,
|
||||
timeoutTotal,
|
||||
status: doneTotal >= 1 ? 'finished' : submittedTotal > 0 ? 'sending' : 'queued',
|
||||
};
|
||||
}
|
||||
|
||||
function signatureReportApprovalScopes(carrier: string) {
|
||||
|
||||
@@ -1542,53 +1542,6 @@ startInboundWorkflowWorker() {
|
||||
return { candidate, workflowDigest, taskId, messageRecordId, content, drainageDetection, billing };
|
||||
}));
|
||||
await this.measureInboundStage('message_persist', () => this.prisma.$transaction(async (tx) => {
|
||||
// Lock accounts in a stable order and reserve the whole tenant subtotal once.
|
||||
// Per-message idempotency rows remain separate so retries, releases and charges
|
||||
// keep the original accounting contract without one account update per Inbox row.
|
||||
const tenantIds = [...new Set(prepared.map(({ candidate }) => candidate.application.tenantId))].sort();
|
||||
for (const tenantId of tenantIds) {
|
||||
await tx.$executeRaw`SELECT pg_advisory_xact_lock(hashtextextended(${'tenant-account:' + tenantId}, 0))`;
|
||||
await tx.tenantAccount.upsert({
|
||||
where: { tenantId }, update: {},
|
||||
create: { tenantId, balanceCents: 0, creditCents: 0, status: 'active' },
|
||||
});
|
||||
const account = await tx.tenantAccount.findUniqueOrThrow({ where: { tenantId } });
|
||||
const paid = prepared.filter(({ candidate, billing }) => candidate.application.tenantId === tenantId && billing.amountCents > 0);
|
||||
const totalAmount = paid.reduce((sum, entry) => sum + entry.billing.amountCents, 0);
|
||||
if (totalAmount === 0) continue;
|
||||
const available = moneyToNumber(account.balanceCents) + moneyToNumber(account.creditCents);
|
||||
if (account.status !== 'active' || available < totalAmount) {
|
||||
throw new BadRequestException('企业账户余额不足');
|
||||
}
|
||||
const existing = await tx.accountTransaction.findMany({
|
||||
where: { idempotencyKey: { in: paid.map(({ candidate }) => `${candidate.item.requestKey}:freeze`) } },
|
||||
select: { idempotencyKey: true },
|
||||
});
|
||||
if (existing.length > 0) {
|
||||
throw new ConflictException('批量计费幂等流水已存在，转入逐条恢复');
|
||||
}
|
||||
const balanceBefore = moneyToNumber(account.balanceCents);
|
||||
let reserved = 0;
|
||||
await tx.accountTransaction.createMany({
|
||||
data: paid.map(({ candidate, taskId, billing }) => {
|
||||
reserved += billing.amountCents;
|
||||
return {
|
||||
tenantId,
|
||||
transactionType: 'frozen',
|
||||
idempotencyKey: `${candidate.item.requestKey}:freeze`,
|
||||
amountCents: -billing.amountCents,
|
||||
balanceAfter: balanceBefore - reserved,
|
||||
relatedType: 'sms_batch_task',
|
||||
relatedId: taskId,
|
||||
remark: 'CMPP 入站短信批量冻结',
|
||||
};
|
||||
}),
|
||||
});
|
||||
await tx.tenantAccount.update({
|
||||
where: { tenantId },
|
||||
data: { balanceCents: { decrement: totalAmount } },
|
||||
});
|
||||
}
|
||||
await tx.smsBatchTask.createMany({
|
||||
data: prepared.map(({ candidate, workflowDigest, taskId, content }) => ({
|
||||
id: taskId,
|
||||
@@ -1637,6 +1590,56 @@ startInboundWorkflowWorker() {
|
||||
status: 'queued',
|
||||
})),
|
||||
});
|
||||
|
||||
// Acquire the tenant account lock only after the independent workflow rows
|
||||
// have been staged. PostgreSQL holds transaction-scoped locks until commit;
|
||||
// keeping this section last preserves the all-or-nothing boundary while
|
||||
// avoiding serialization across task/API/message persistence for one tenant.
|
||||
// Per-message idempotency rows remain separate for retry/release/charge repair.
|
||||
const tenantIds = [...new Set(prepared.map(({ candidate }) => candidate.application.tenantId))].sort();
|
||||
for (const tenantId of tenantIds) {
|
||||
const paid = prepared.filter(({ candidate, billing }) => candidate.application.tenantId === tenantId && billing.amountCents > 0);
|
||||
const totalAmount = paid.reduce((sum, entry) => sum + entry.billing.amountCents, 0);
|
||||
if (totalAmount === 0) continue;
|
||||
await tx.$executeRaw`SELECT pg_advisory_xact_lock(hashtextextended(${'tenant-account:' + tenantId}, 0))`;
|
||||
await tx.tenantAccount.upsert({
|
||||
where: { tenantId }, update: {},
|
||||
create: { tenantId, balanceCents: 0, creditCents: 0, status: 'active' },
|
||||
});
|
||||
const account = await tx.tenantAccount.findUniqueOrThrow({ where: { tenantId } });
|
||||
const available = moneyToNumber(account.balanceCents) + moneyToNumber(account.creditCents);
|
||||
if (account.status !== 'active' || available < totalAmount) {
|
||||
throw new BadRequestException('企业账户余额不足');
|
||||
}
|
||||
const existing = await tx.accountTransaction.findMany({
|
||||
where: { idempotencyKey: { in: paid.map(({ candidate }) => `${candidate.item.requestKey}:freeze`) } },
|
||||
select: { idempotencyKey: true },
|
||||
});
|
||||
if (existing.length > 0) {
|
||||
throw new ConflictException('批量计费幂等流水已存在，转入逐条恢复');
|
||||
}
|
||||
const balanceBefore = moneyToNumber(account.balanceCents);
|
||||
let reserved = 0;
|
||||
await tx.accountTransaction.createMany({
|
||||
data: paid.map(({ candidate, taskId, billing }) => {
|
||||
reserved += billing.amountCents;
|
||||
return {
|
||||
tenantId,
|
||||
transactionType: 'frozen',
|
||||
idempotencyKey: `${candidate.item.requestKey}:freeze`,
|
||||
amountCents: -billing.amountCents,
|
||||
balanceAfter: balanceBefore - reserved,
|
||||
relatedType: 'sms_batch_task',
|
||||
relatedId: taskId,
|
||||
remark: 'CMPP 入站短信批量冻结',
|
||||
};
|
||||
}),
|
||||
});
|
||||
await tx.tenantAccount.update({
|
||||
where: { tenantId },
|
||||
data: { balanceCents: { decrement: totalAmount } },
|
||||
});
|
||||
}
|
||||
}));
|
||||
await this.measureInboundStage('queue_publish', () => this.facade.getSendQueue().addBulk(prepared.map(({ candidate, messageRecordId }) => ({
|
||||
name: 'send-message' as const,
|
||||
|
||||
@@ -63,7 +63,7 @@ export class SendSubmissionService {
|
||||
this.inboundEntry = new SendInboundEntryService(prisma, billing, riskReview, phoneFrequency, phoneRouting, facade, callbacks, metrics);
|
||||
this.reviewContinuation = new SendReviewContinuationService(prisma, billing, riskReview, phoneFrequency, phoneRouting, facade, callbacks);
|
||||
this.scheduledDispatch = new SendScheduledDispatchService(prisma, billing, riskReview, phoneFrequency, phoneRouting, facade, callbacks);
|
||||
this.gatewaySubmit = new SendGatewaySubmitService(prisma, billing, riskReview, phoneFrequency, phoneRouting, facade, callbacks);
|
||||
this.gatewaySubmit = new SendGatewaySubmitService(prisma, billing, riskReview, phoneFrequency, phoneRouting, facade, callbacks, metrics);
|
||||
}
|
||||
|
||||
onModuleDestroy() {
|
||||
@@ -229,6 +229,10 @@ startWorker() {
|
||||
return this.gatewaySubmit.startWorker();
|
||||
}
|
||||
|
||||
startSubmitOutboxPublisher() {
|
||||
return this.gatewaySubmit.startSubmitOutboxPublisher();
|
||||
}
|
||||
|
||||
startInboundWorkflowWorker() {
|
||||
return this.inboundEntry.startInboundWorkflowWorker();
|
||||
}
|
||||
@@ -259,6 +263,7 @@ async submitMessageToGateway(
|
||||
applicationExtension?: string | null;
|
||||
template?: { signature?: { id?: string | null; name?: string | null } | null } | null;
|
||||
signature?: { id?: string | null; name?: string | null } | null;
|
||||
batchTask?: { sourceType?: string | null; phoneTotal?: number | null } | null;
|
||||
},
|
||||
routed: RoutedChannel,
|
||||
attempt: number,
|
||||
@@ -274,8 +279,8 @@ async selectChannelForMessage(
|
||||
return this.gatewaySubmit.selectChannelForMessage(message, options);
|
||||
}
|
||||
|
||||
async findApplicationRoute(tenantId: string, applicationId: string | undefined, carrier: string) {
|
||||
return this.gatewaySubmit.findApplicationRoute(tenantId, applicationId, carrier);
|
||||
async findApplicationRoute(tenantId: string, applicationId: string | undefined, carrier: string, signatureId?: string) {
|
||||
return this.gatewaySubmit.findApplicationRoute(tenantId, applicationId, carrier, signatureId);
|
||||
}
|
||||
|
||||
async identifyCarrier(phoneNumber: string) {
|
||||
@@ -307,8 +312,8 @@ async waitForChannelRateLimit(channelId: string, tps: number) {
|
||||
return this.gatewaySubmit.waitForChannelRateLimit(channelId, tps);
|
||||
}
|
||||
|
||||
async refreshTaskProgress(batchTaskId: string) {
|
||||
return this.gatewaySubmit.refreshTaskProgress(batchTaskId);
|
||||
async refreshTaskProgress(batchTaskId: string, knownSingleMessageStatus?: string) {
|
||||
return this.gatewaySubmit.refreshTaskProgress(batchTaskId, knownSingleMessageStatus);
|
||||
}
|
||||
|
||||
getSendQueue(): Queue<SendJob, unknown, 'send-message'> {
|
||||
|
||||
@@ -0,0 +1,30 @@
|
||||
import 'reflect-metadata';
|
||||
import { NestFactory } from '@nestjs/core';
|
||||
import { createServer } from 'node:http';
|
||||
import { MetricsService } from './metrics/metrics.service';
|
||||
import { SendWorkerModule } from './send-worker.module';
|
||||
|
||||
async function bootstrap() {
|
||||
if (process.env.CMPP_PROCESS_ROLE !== 'outbox') {
|
||||
throw new Error('submit-outbox-worker requires CMPP_PROCESS_ROLE=outbox');
|
||||
}
|
||||
const app = await NestFactory.createApplicationContext(SendWorkerModule);
|
||||
app.enableShutdownHooks();
|
||||
const metrics = app.get(MetricsService);
|
||||
const host = process.env.API_OUTBOX_METRICS_HOST?.trim() || '127.0.0.1';
|
||||
const port = Number(process.env.API_OUTBOX_METRICS_PORT ?? 9467);
|
||||
const server = createServer((request, response) => {
|
||||
if (request.method !== 'GET' || request.url !== '/metrics') {
|
||||
response.writeHead(404).end();
|
||||
return;
|
||||
}
|
||||
response.writeHead(200, { 'Content-Type': 'text/plain; version=0.0.4; charset=utf-8', 'Cache-Control': 'no-store' });
|
||||
response.end(metrics.render());
|
||||
});
|
||||
await new Promise<void>((resolve, reject) => {
|
||||
server.once('error', reject);
|
||||
server.listen(port, host, resolve);
|
||||
});
|
||||
}
|
||||
|
||||
void bootstrap();
|
||||
Reference in New Issue
Block a user