hectorzhao/lislgosms
1
0
Fork 0
Code Issues Pull Requests Actions 1 Packages Projects Releases Wiki Activity

perf(cmpp): add durable inbound fast path

Browse Source
This commit is contained in:
hectorzhao
2026-08-20 17:34:45 +08:00
parent 26ef67fb6a
commit 0b63bcd74e
29 changed files with 1136 additions and 87 deletions
Download Patch File Download Diff File Expand all files Collapse all files
api/prisma/migrations/20260820170000_add_cmpp_inbound_submission_inbox/migration.sql
+91
View File
@@ -0,0 +1,91 @@
CREATE TABLE "CmppInboundSubmissionInbox" (
"id" TEXT NOT NULL,
"requestKey" TEXT NOT NULL,
"payloadHash" TEXT NOT NULL,
"tenantId" TEXT NOT NULL,
"applicationId" TEXT NOT NULL,
"queuePriority" TEXT NOT NULL DEFAULT 'normal',
"payload" JSONB NOT NULL,
"response" JSONB NOT NULL,
"status" TEXT NOT NULL DEFAULT 'pending',
"attempts" INTEGER NOT NULL DEFAULT 0,
"nextAttemptAt" TIMESTAMP(3) NOT NULL DEFAULT CURRENT_TIMESTAMP,
"lockedAt" TIMESTAMP(3),
"lockedBy" TEXT,
"lastError" TEXT,
"result" JSONB,
"completedAt" TIMESTAMP(3),
"createdAt" TIMESTAMP(3) NOT NULL DEFAULT CURRENT_TIMESTAMP,
"updatedAt" TIMESTAMP(3) NOT NULL,
CONSTRAINT "CmppInboundSubmissionInbox_pkey" PRIMARY KEY ("id")
);
CREATE UNIQUE INDEX "CmppInboundSubmissionInbox_requestKey_key"
ON "CmppInboundSubmissionInbox"("requestKey");
CREATE INDEX "CmppInboundSubmissionInbox_status_nextAttemptAt_createdAt_idx"
ON "CmppInboundSubmissionInbox"("status", "nextAttemptAt", "createdAt");
CREATE INDEX "CmppInboundSubmissionInbox_status_lockedAt_idx"
ON "CmppInboundSubmissionInbox"("status", "lockedAt");
CREATE INDEX "CmppInboundSubmissionInbox_applicationId_createdAt_idx"
ON "CmppInboundSubmissionInbox"("applicationId", "createdAt");
CREATE INDEX "CmppInboundSubmissionInbox_queuePriority_status_createdAt_idx"
ON "CmppInboundSubmissionInbox"("queuePriority", "status", "createdAt");
ALTER TABLE "CmppInboundSubmissionInbox"
ADD CONSTRAINT "CmppInboundSubmissionInbox_tenantId_fkey"
FOREIGN KEY ("tenantId") REFERENCES "Tenant"("id") ON DELETE RESTRICT ON UPDATE CASCADE;
ALTER TABLE "CmppInboundSubmissionInbox"
ADD CONSTRAINT "CmppInboundSubmissionInbox_applicationId_fkey"
FOREIGN KEY ("applicationId") REFERENCES "SmsApplication"("id") ON DELETE RESTRICT ON UPDATE CASCADE;
CREATE TABLE "SmsApplicationDailyReservation" (
"id" TEXT NOT NULL,
"reservationKey" TEXT NOT NULL,
"tenantId" TEXT NOT NULL,
"applicationId" TEXT NOT NULL,
"usageDate" DATE NOT NULL,
"requestedCount" INTEGER NOT NULL,
"dailyLimit" INTEGER NOT NULL,
"usedCount" INTEGER,
"reserved" BOOLEAN NOT NULL,
"createdAt" TIMESTAMP(3) NOT NULL DEFAULT CURRENT_TIMESTAMP,
CONSTRAINT "SmsApplicationDailyReservation_pkey" PRIMARY KEY ("id")
);
CREATE UNIQUE INDEX "SmsApplicationDailyReservation_reservationKey_key"
ON "SmsApplicationDailyReservation"("reservationKey");
CREATE INDEX "SmsApplicationDailyReservation_applicationId_usageDate_idx"
ON "SmsApplicationDailyReservation"("applicationId", "usageDate");
CREATE INDEX "SmsApplicationDailyReservation_tenantId_createdAt_idx"
ON "SmsApplicationDailyReservation"("tenantId", "createdAt");
ALTER TABLE "SmsApplicationDailyReservation" ADD CONSTRAINT "SmsApplicationDailyReservation_tenantId_fkey"
FOREIGN KEY ("tenantId") REFERENCES "Tenant"("id") ON DELETE RESTRICT ON UPDATE CASCADE;
ALTER TABLE "SmsApplicationDailyReservation" ADD CONSTRAINT "SmsApplicationDailyReservation_applicationId_fkey"
FOREIGN KEY ("applicationId") REFERENCES "SmsApplication"("id") ON DELETE RESTRICT ON UPDATE CASCADE;
CREATE TABLE "PhoneFrequencyReservation" (
"id" TEXT NOT NULL,
"reservationKey" TEXT NOT NULL,
"tenantId" TEXT NOT NULL,
"applicationId" TEXT NOT NULL,
"result" JSONB NOT NULL,
"createdAt" TIMESTAMP(3) NOT NULL DEFAULT CURRENT_TIMESTAMP,
CONSTRAINT "PhoneFrequencyReservation_pkey" PRIMARY KEY ("id")
);
CREATE UNIQUE INDEX "PhoneFrequencyReservation_reservationKey_key"
ON "PhoneFrequencyReservation"("reservationKey");
CREATE INDEX "PhoneFrequencyReservation_applicationId_createdAt_idx"
ON "PhoneFrequencyReservation"("applicationId", "createdAt");
CREATE INDEX "PhoneFrequencyReservation_tenantId_createdAt_idx"
ON "PhoneFrequencyReservation"("tenantId", "createdAt");
ALTER TABLE "PhoneFrequencyReservation" ADD CONSTRAINT "PhoneFrequencyReservation_tenantId_fkey"
FOREIGN KEY ("tenantId") REFERENCES "Tenant"("id") ON DELETE RESTRICT ON UPDATE CASCADE;
ALTER TABLE "PhoneFrequencyReservation" ADD CONSTRAINT "PhoneFrequencyReservation_applicationId_fkey"
FOREIGN KEY ("applicationId") REFERENCES "SmsApplication"("id") ON DELETE RESTRICT ON UPDATE CASCADE;
api/prisma/schema.prisma
+69
View File
@@ -52,6 +52,9 @@ model Tenant {
openApiRequests OpenApiRequest[]
httpWebhookEvents HttpWebhookEvent[]
cmppInboundLongMessages CmppInboundLongMessage[]
cmppInboundSubmissionInboxes CmppInboundSubmissionInbox[]
smsApplicationDailyReservations SmsApplicationDailyReservation[]
phoneFrequencyReservations PhoneFrequencyReservation[]
}
model EnterpriseCertification {
@@ -467,10 +470,13 @@ model SmsApplication {
httpWebhookEndpoints HttpWebhookEndpoint[]
httpWebhookEvents HttpWebhookEvent[]
dailyUsages SmsApplicationDailyUsage[]
dailyReservations SmsApplicationDailyReservation[]
inboundLongMessages CmppInboundLongMessage[]
inboundSubmissionInboxes CmppInboundSubmissionInbox[]
riskRules RiskRule[]
phoneFrequencyStates PhoneFrequencyState[]
phoneFrequencyHits PhoneFrequencyHit[]
phoneFrequencyReservations PhoneFrequencyReservation[]
@@index([tenantId, status])
@@index([status, createdAt])
@@ -533,6 +539,25 @@ model SmsApplicationDailyUsage {
@@index([usageDate])
}
model SmsApplicationDailyReservation {
id String @id @default(cuid())
reservationKey String @unique
tenantId String
applicationId String
usageDate DateTime @db.Date
requestedCount Int
dailyLimit Int
usedCount Int?
reserved Boolean
createdAt DateTime @default(now())
tenant Tenant @relation(fields: [tenantId], references: [id])
application SmsApplication @relation(fields: [applicationId], references: [id], onDelete: Restrict)
@@index([applicationId, usageDate])
@@index([tenantId, createdAt])
}
model SmsApplicationHttpIpAllowlist {
id String @id @default(cuid())
applicationId String
@@ -1578,6 +1603,21 @@ model PhoneFrequencyHit {
@@index([releasedById])
}
model PhoneFrequencyReservation {
id String @id @default(cuid())
reservationKey String @unique
tenantId String
applicationId String
result Json
createdAt DateTime @default(now())
tenant Tenant @relation(fields: [tenantId], references: [id])
application SmsApplication @relation(fields: [applicationId], references: [id], onDelete: Restrict)
@@index([applicationId, createdAt])
@@index([tenantId, createdAt])
}
model PhoneFrequencyWhitelist {
id String @id @default(cuid())
phoneNumber String @unique
@@ -1952,6 +1992,35 @@ model CmppInboundLongMessageSegment {
@@index([sequenceId])
}
model CmppInboundSubmissionInbox {
id String @id @default(cuid())
requestKey String @unique
payloadHash String
tenantId String
applicationId String
queuePriority String @default("normal")
payload Json
response Json
status String @default("pending")
attempts Int @default(0)
nextAttemptAt DateTime @default(now())
lockedAt DateTime?
lockedBy String?
lastError String?
result Json?
completedAt DateTime?
createdAt DateTime @default(now())
updatedAt DateTime @updatedAt
tenant Tenant @relation(fields: [tenantId], references: [id])
application SmsApplication @relation(fields: [applicationId], references: [id])
@@index([status, nextAttemptAt, createdAt])
@@index([status, lockedAt])
@@index([applicationId, createdAt])
@@index([queuePriority, status, createdAt])
}
model SmsReceiptRecord {
id String @id @default(cuid())
tenantId String?
api/src/metrics/metrics.service.ts
+38
View File
@@ -6,6 +6,7 @@ const CMPP_INBOUND_DURATION_BUCKETS = [0.005, 0.01, 0.025, 0.05, 0.1, 0.25, 0.5,
export type CmppInboundStage =
| 'application_lookup'
| 'inbox_persist'
| 'long_message_fragment'
| 'submission_precheck'
| 'template_match'
@@ -45,6 +46,12 @@ export class MetricsService implements OnModuleDestroy {
private readonly http = new Map<string, HttpMetric>();
private readonly cmppInbound = new Map<string, HttpMetric>();
private inFlight = 0;
private inboundWorkflowPending = 0;
private inboundWorkflowProcessing = 0;
private inboundWorkflowOldestPendingAgeSeconds = 0;
private inboundWorkflowConfiguredSlots = 0;
private inboundWorkflowInFlightSlots = 0;
private readonly inboundWorkflowResults = new Map<string, number>();
constructor() {
this.eventLoopDelay.enable();
@@ -91,6 +98,21 @@ export class MetricsService implements OnModuleDestroy {
this.cmppInbound.set(key, metric);
}
setInboundWorkflowState(pending: number, processing: number, oldestPendingAgeSeconds: number) {
this.inboundWorkflowPending = Math.max(0, pending);
this.inboundWorkflowProcessing = Math.max(0, processing);
this.inboundWorkflowOldestPendingAgeSeconds = Math.max(0, oldestPendingAgeSeconds);
}
setInboundWorkflowSlots(configured: number, inFlight: number) {
this.inboundWorkflowConfiguredSlots = Math.max(0, configured);
this.inboundWorkflowInFlightSlots = Math.max(0, inFlight);
}
recordInboundWorkflowResult(result: 'completed' | 'retry') {
this.inboundWorkflowResults.set(result, (this.inboundWorkflowResults.get(result) ?? 0) + 1);
}
render() {
const memory = process.memoryUsage();
const uptime = Number(process.hrtime.bigint() - this.startedAt) / 1_000_000_000;
@@ -119,6 +141,19 @@ 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_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' }),
metricLine('cmpp_worker_inbound_workflow_items', this.inboundWorkflowProcessing, { state: 'processing' }),
'# HELP cmpp_worker_inbound_workflow_slots Durable workflow worker slots by state.',
'# TYPE cmpp_worker_inbound_workflow_slots gauge',
metricLine('cmpp_worker_inbound_workflow_slots', this.inboundWorkflowConfiguredSlots, { state: 'configured' }),
metricLine('cmpp_worker_inbound_workflow_slots', this.inboundWorkflowInFlightSlots, { state: 'in_flight' }),
'# HELP cmpp_worker_inbound_workflow_oldest_pending_age_seconds Age of the oldest pending durable workflow.',
'# TYPE cmpp_worker_inbound_workflow_oldest_pending_age_seconds gauge',
metricLine('cmpp_worker_inbound_workflow_oldest_pending_age_seconds', this.inboundWorkflowOldestPendingAgeSeconds),
'# HELP cmpp_worker_inbound_workflow_results_total Durable workflow processing outcomes.',
'# TYPE cmpp_worker_inbound_workflow_results_total counter',
];
for (const [key, metric] of this.http) {
const [method, route, status] = key.split('\u0000');
@@ -141,6 +176,9 @@ 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 [result, count] of this.inboundWorkflowResults) {
lines.push(metricLine('cmpp_worker_inbound_workflow_results_total', count, { result }));
}
this.eventLoopDelay.reset();
return `${lines.join('\n')}\n`;
}
api/src/prisma/prisma.service.ts
+4 -1
View File
@@ -6,9 +6,12 @@ import { requestContext } from '../common/request-context';
@Injectable()
export class PrismaService extends PrismaClient implements OnModuleDestroy {
constructor() {
const databaseUrl = process.env.CMPP_PROCESS_ROLE === 'worker'
? process.env.API_WORKER_DATABASE_URL || process.env.DATABASE_URL
: process.env.DATABASE_URL;
super({
adapter: new PrismaPg(
process.env.DATABASE_URL ??
databaseUrl ??
'postgresql://cmpp:cmpp_password@localhost:5432/cmpp_platform?schema=public',
),
});
api/src/risk-review/phone-frequency.service.ts
+46 -2
View File
@@ -80,6 +80,7 @@ export class PhoneFrequencyService {
phones: string[],
sourceType?: string,
requestedAt = new Date(),
reservationKey?: string,
) {
if (!applicationId) return new Map<string, PhoneFrequencyRejection>();
const normalizedPhones = [...new Set(phones.map((phone) => phone.trim()).filter(Boolean))].sort();
@@ -87,14 +88,35 @@ export class PhoneFrequencyService {
await this.riskReview.ensureDefaultRules();
const rules = await this.effectiveRules(applicationId);
if (rules.length === 0) return new Map<string, PhoneFrequencyRejection>();
const normalizedReservationKey = reservationKey?.trim();
return this.prisma.$transaction(async (tx) => {
if (normalizedReservationKey) {
// Frequency counters and the reservation result commit together. Retrying a reclaimed
// Inbox item therefore returns the original decision without incrementing either window.
await tx.$executeRaw`SELECT pg_advisory_xact_lock(hashtextextended(${'phone-frequency:' + normalizedReservationKey}, 0))`;
const existingReservation = await tx.phoneFrequencyReservation.findUnique({
where: { reservationKey: normalizedReservationKey },
});
if (existingReservation) {
if (existingReservation.tenantId !== tenantId || existingReservation.applicationId !== applicationId) {
throw new BadRequestException('号码频控幂等键已用于另一笔预留');
}
return frequencyRejectionsFromJson(existingReservation.result);
}
}
const rejected = new Map<string, PhoneFrequencyRejection>();
// 平台级白名单只截断号码频控链路；调用 reserve 之前已执行的格式、黑名单等校验不受影响。
const whitelistedPhones = await this.findActiveWhitelistedPhones(tx, normalizedPhones);
const controlledPhones = normalizedPhones.filter((phone) => !whitelistedPhones.has(phone));
if (controlledPhones.length === 0) return rejected;
if (controlledPhones.length === 0 || rules.length === 0) {
if (normalizedReservationKey) {
await tx.phoneFrequencyReservation.create({
data: { reservationKey: normalizedReservationKey, tenantId, applicationId, result: [] },
});
}
return rejected;
}
for (const rule of rules) {
const window = fixedShanghaiWindow(requestedAt, readPeriodSeconds(rule));
// 分块限制 SQL 参数数量，但两条规则的全部分块仍在同一事务中提交或回滚。
@@ -144,6 +166,16 @@ export class PhoneFrequencyService {
}
}
}
if (normalizedReservationKey) {
await tx.phoneFrequencyReservation.create({
data: {
reservationKey: normalizedReservationKey,
tenantId,
applicationId,
result: [...rejected.entries()].map(([phoneNumber, rejection]) => ({ phoneNumber, ...rejection })),
},
});
}
return rejected;
}, { isolationLevel: Prisma.TransactionIsolationLevel.ReadCommitted });
}
@@ -566,6 +598,18 @@ export class PhoneFrequencyService {
}
}
function frequencyRejectionsFromJson(value: Prisma.JsonValue) {
const result = new Map<string, PhoneFrequencyRejection>();
if (!Array.isArray(value)) return result;
for (const item of value) {
if (!item || typeof item !== 'object' || Array.isArray(item)) continue;
const phoneNumber = typeof item.phoneNumber === 'string' ? item.phoneNumber : '';
const reason = typeof item.reason === 'string' ? item.reason : '';
if (phoneNumber && reason) result.set(phoneNumber, { code: 'PHONE_FREQUENCY_LIMIT', reason });
}
return result;
}
const whitelistUserInclude = {
createdBy: { select: { id: true, username: true, displayName: true } },
updatedBy: { select: { id: true, username: true, displayName: true } },
api/src/send-chain/send-batch-entry.service.ts
+47 -7
View File
@@ -1,4 +1,4 @@
import { BadRequestException, HttpException, HttpStatus, Logger, NotFoundException } from '@nestjs/common';
import { BadRequestException, ConflictException, HttpException, HttpStatus, Logger, NotFoundException } from '@nestjs/common';
import { Prisma } from '@prisma/client';
import { Queue, Worker } from 'bullmq';
import IORedis from 'ioredis';
@@ -525,15 +525,16 @@ async reserveDailySendQuota(applicationId: string, requestedCount: number) {
return result;
}
async tryReserveDailySendQuota(applicationId: string, requestedCount: number) {
async tryReserveDailySendQuota(applicationId: string, requestedCount: number, reservationKey?: string) {
if (!Number.isInteger(requestedCount) || requestedCount <= 0) {
throw new BadRequestException('发送号码数量必须为正整数');
}
const usageDate = shanghaiDateKey();
const reservationId = randomUUID();
const rows = await this.prisma.$queryRaw<Array<{ dailyLimit: number; usedCount: number | null }>>(Prisma.sql`
const reserve = (client: Pick<Prisma.TransactionClient, '$queryRaw'>) => {
const reservationId = randomUUID();
return client.$queryRaw<Array<{ tenantId: string; dailyLimit: number; usedCount: number | null }>>(Prisma.sql`
WITH application_limit AS (
SELECT id, COALESCE("dailyLimit", 100000)::integer AS "dailyLimit"
SELECT id, "tenantId", COALESCE("dailyLimit", 100000)::integer AS "dailyLimit"
FROM "SmsApplication"
WHERE id = ${applicationId}
), reservation AS (
@@ -550,10 +551,49 @@ async tryReserveDailySendQuota(applicationId: string, requestedCount: number) {
<= (SELECT "dailyLimit" FROM application_limit)
RETURNING "usedCount"
)
SELECT application_limit."dailyLimit", reservation."usedCount"
SELECT application_limit."tenantId", application_limit."dailyLimit", reservation."usedCount"
FROM application_limit
LEFT JOIN reservation ON TRUE
`);
`);
};
const normalizedReservationKey = reservationKey?.trim();
const rows = normalizedReservationKey
? await this.prisma.$transaction(async (tx) => {
// The quota increment and its idempotency record share one short transaction. A worker
// crash can therefore neither lose a successful reservation nor increment it twice.
await tx.$executeRaw`SELECT pg_advisory_xact_lock(hashtextextended(${'daily-quota:' + normalizedReservationKey}, 0))`;
const existing = await tx.smsApplicationDailyReservation.findUnique({
where: { reservationKey: normalizedReservationKey },
});
if (existing) {
if (existing.applicationId !== applicationId || existing.requestedCount !== requestedCount) {
throw new ConflictException('日发送配额幂等键已用于另一笔预留');
}
return [{
tenantId: existing.tenantId,
dailyLimit: existing.dailyLimit,
usedCount: existing.usedCount,
}];
}
const reservedRows = await reserve(tx);
if (reservedRows.length > 0) {
const row = reservedRows[0];
await tx.smsApplicationDailyReservation.create({
data: {
reservationKey: normalizedReservationKey,
tenantId: row.tenantId,
applicationId,
usageDate,
requestedCount,
dailyLimit: Number(row.dailyLimit),
usedCount: row.usedCount == null ? null : Number(row.usedCount),
reserved: row.usedCount != null,
},
});
}
return reservedRows;
})
: await reserve(this.prisma);
if (rows.length === 0) {
throw new NotFoundException('短信应用不存在');
}
api/src/send-chain/send-chain.contracts.ts
+1
View File
@@ -30,6 +30,7 @@ export interface GatewayInboundAuthDto {
}
export interface GatewayInboundSubmitDto {
requestId?: string;
account: string;
phoneNumber?: string;
phoneNumbers?: string[];
api/src/send-chain/send-chain.service.spec.ts
+96
View File
@@ -329,6 +329,13 @@ function createPrismaMock() {
lastError: 'downstream client is not connected',
}),
},
cmppInboundSubmissionInbox: {
create: jest.fn().mockResolvedValue({ id: 'inbox-1' }),
findUnique: jest.fn().mockResolvedValue(null),
updateMany: jest.fn().mockResolvedValue({ count: 1 }),
count: jest.fn().mockResolvedValue(0),
findFirst: jest.fn().mockResolvedValue(null),
},
smsBillingRecord: {
findFirst: jest.fn().mockResolvedValue(null),
create: jest.fn().mockResolvedValue({ id: 'bill-1' }),
@@ -2098,6 +2105,95 @@ describe('SendChainService', () => {
});
});
it('returns from the durable CMPP Inbox fast path before risk, billing, or queue publication', async () => {
const previous = process.env.CMPP_INBOUND_FAST_PATH_ENABLED;
process.env.CMPP_INBOUND_FAST_PATH_ENABLED = 'true';
try {
const { service, prisma, billing, riskReview, phoneFrequency } = createService();
service.enqueueBatchTask = jest.fn();
const result = await service.submitInboundMessage({
requestId: 'cmpp-inbound:test-fast-path',
account: '100001',
phoneNumbers: ['13800000001', '13900000002'],
content: 'hello',
sequenceId: 777,
remoteIp: '127.0.0.1',
});
expect(result).toEqual(expect.objectContaining({
accepted: true,
status: 'accepted_pending',
phoneCount: 2,
}));
expect(prisma.cmppInboundSubmissionInbox.create).toHaveBeenCalledWith({
data: expect.objectContaining({
requestKey: 'cmpp-inbound:test-fast-path',
tenantId: 'tenant-1',
applicationId: 'app-1',
}),
});
expect(prisma.smsBatchTask.create).not.toHaveBeenCalled();
expect(prisma.smsMessageRecord.create).not.toHaveBeenCalled();
expect(riskReview.evaluateTask).not.toHaveBeenCalled();
expect(phoneFrequency.reserve).not.toHaveBeenCalled();
expect(billing.freeze).not.toHaveBeenCalled();
expect(service.enqueueBatchTask).not.toHaveBeenCalled();
} finally {
if (previous == null) delete process.env.CMPP_INBOUND_FAST_PATH_ENABLED;
else process.env.CMPP_INBOUND_FAST_PATH_ENABLED = previous;
}
});
it('returns the stored SubmitResp for an idempotent CMPP Inbox retry', async () => {
const previous = process.env.CMPP_INBOUND_FAST_PATH_ENABLED;
process.env.CMPP_INBOUND_FAST_PATH_ENABLED = 'true';
try {
const { service, prisma } = createService();
let originalHash = '';
const storedResponse = {
accepted: true,
tenantId: 'tenant-1',
applicationId: 'app-1',
taskId: '',
messageId: 'MSG-stable',
messageRecordId: '',
status: 'accepted_pending',
phoneCount: 1,
messages: [{ phoneNumber: '13800000001', messageId: 'MSG-stable', messageRecordId: '', taskId: '', status: 'accepted_pending' }],
};
prisma.cmppInboundSubmissionInbox.create
.mockImplementationOnce(({ data }) => {
originalHash = data.payloadHash;
return Promise.resolve({ id: 'inbox-1' });
})
.mockRejectedValueOnce(new Prisma.PrismaClientKnownRequestError('duplicate request key', {
code: 'P2002',
clientVersion: '7.9.0',
}));
prisma.cmppInboundSubmissionInbox.findUnique.mockImplementation(() => Promise.resolve({
payloadHash: originalHash,
response: storedResponse,
}));
const request = {
requestId: 'cmpp-inbound:test-retry',
account: '100001',
phoneNumber: '13800000001',
content: 'hello',
sequenceId: 778,
remoteIp: '127.0.0.1',
};
await service.submitInboundMessage(request);
await expect(service.submitInboundMessage(request)).resolves.toEqual(storedResponse);
expect(prisma.cmppInboundSubmissionInbox.create).toHaveBeenCalledTimes(2);
expect(prisma.smsMessageRecord.create).not.toHaveBeenCalled();
} finally {
if (previous == null) delete process.env.CMPP_INBOUND_FAST_PATH_ENABLED;
else process.env.CMPP_INBOUND_FAST_PATH_ENABLED = previous;
}
});
it('enqueues a freshly persisted inbound message without querying the task and message again', async () => {
const { service, prisma } = createService();
const add = jest.fn().mockResolvedValue(undefined);
api/src/send-chain/send-chain.service.ts
+15 -6
View File
@@ -78,9 +78,14 @@ export class SendChainService implements OnModuleInit, OnModuleDestroy {
}
onModuleInit() {
const processRole = process.env.CMPP_PROCESS_ROLE?.trim() || 'all';
if (processRole === 'api') return;
if (process.env.API_ENABLE_SEND_WORKER === 'true') {
this.startWorker();
}
if (process.env.CMPP_INBOUND_WORKFLOW_WORKER_ENABLED === 'true') {
this.submission.startInboundWorkflowWorker();
}
if (process.env.SMS_RECEIPT_TIMEOUT_SCAN_ENABLED !== 'false') {
this.receiptTimeoutInitialTimer = setTimeout(() => void this.runReceiptTimeoutScan(), RECEIPT_TIMEOUT_INITIAL_DELAY_MS);
this.receiptTimeoutInitialTimer.unref?.();
@@ -159,6 +164,7 @@ export class SendChainService implements OnModuleInit, OnModuleDestroy {
await this.sendQueue?.close();
await this.gatewayQueue?.close();
this.redis?.disconnect();
await this.submission.onModuleDestroy();
}
async createBatchTask(data: CreateBatchTaskDto) {
@@ -579,8 +585,10 @@ export class SendChainService implements OnModuleInit, OnModuleDestroy {
phoneNumbers: string[],
application: Awaited<ReturnType<SendChainService['findInboundApplication']>>,
requestedGroupMessageId?: string,
requestedMessageIds?: string[],
workflowKey?: string,
) {
return this.submission.submitCompleteInboundMessage(data, phoneNumbers, application, requestedGroupMessageId);
return this.submission.submitCompleteInboundMessage(data, phoneNumbers, application, requestedGroupMessageId, requestedMessageIds, workflowKey);
}
private async collectInboundLongMessageFragment(
@@ -602,8 +610,9 @@ export class SendChainService implements OnModuleInit, OnModuleDestroy {
application: NonNullable<Awaited<ReturnType<SendChainService['findInboundApplication']>>>,
synchronousRejection?: { code: string; reason: string },
receiptRejection?: { code: string; reason: string },
workflowItemKey?: string,
) {
return this.submission.submitInboundSingleMessage(data, messageId, submitGroupMessageId, application, synchronousRejection, receiptRejection);
return this.submission.submitInboundSingleMessage(data, messageId, submitGroupMessageId, application, synchronousRejection, receiptRejection, workflowItemKey);
}
/**
@@ -618,8 +627,8 @@ export class SendChainService implements OnModuleInit, OnModuleDestroy {
variables?: Record<string, unknown>;
phoneNumber: string;
sourceType: 'cmpp';
}) {
return this.submission.evaluateRiskWithPhoneFrequency(input);
}, reservationKey?: string) {
return this.submission.evaluateRiskWithPhoneFrequency(input, reservationKey);
}
async markUnknownTimeout(data: TimeoutUnknownDto) {
@@ -774,8 +783,8 @@ export class SendChainService implements OnModuleInit, OnModuleDestroy {
return this.submission.reserveDailySendQuota(applicationId, requestedCount);
}
private async tryReserveDailySendQuota(applicationId: string, requestedCount: number) {
return this.submission.tryReserveDailySendQuota(applicationId, requestedCount);
private async tryReserveDailySendQuota(applicationId: string, requestedCount: number, reservationKey?: string) {
return this.submission.tryReserveDailySendQuota(applicationId, requestedCount, reservationKey);
}
private async chargeAcceptedMessage(message: {
api/src/send-chain/send-inbound-entry.service.ts
+419 -54
View File
@@ -3,6 +3,7 @@ import { Prisma } from '@prisma/client';
import { Queue, Worker } from 'bullmq';
import IORedis from 'ioredis';
import { createHash, randomUUID } from 'node:crypto';
import { hostname } from 'node:os';
import { setTimeout as sleep } from 'node:timers/promises';
import { BillingService } from '../billing/billing.service';
import { isIpAllowed } from '../common/ip-allowlist';
@@ -17,11 +18,33 @@ import { SEND_QUEUE, GATEWAY_SUBMIT_QUEUE, GATEWAY_SUBMIT_STREAM, DEFAULT_SCHEDU
import type { SendSubmissionCallbacks, SendSubmissionService } from './send-submission.service';
import { detectDrainageContent } from './drainage-content-detection';
type InboundWorkflowPayload = {
data: GatewayInboundSubmitDto;
phoneNumbers: string[];
submitGroupMessageId: string;
messageIds: string[];
};
type ClaimedInboundWorkflow = {
id: string;
requestKey: string;
applicationId: string;
attempts: number;
payload: Prisma.JsonValue;
};
/**
* R9 inboundEntry implementation. Cross-method calls return through the stable SendChainService seam.
*/
export class SendInboundEntryService {
private readonly logger = new Logger('SendChainService');
private readonly inboundWorkflowWorkerId = `${hostname()}:${process.pid}:${randomUUID()}`;
private readonly inboundWorkflowTasks = new Set<Promise<void>>();
private inboundWorkflowTimer?: ReturnType<typeof setTimeout>;
private inboundWorkflowInFlight = 0;
private inboundWorkflowPumping = false;
private inboundWorkflowStopping = false;
private inboundWorkflowMetricsUpdatedAt = 0;
constructor(
private readonly prisma: PrismaService,
@@ -201,6 +224,26 @@ async submitInboundMessage(data: GatewayInboundSubmitDto) {
};
}
try {
if (this.inboundFastPathEnabled()) {
const response = await this.measureInboundStage('inbox_persist', () => (
this.persistInboundWorkflow({
...data,
content: collection.content,
sequenceId: collection.sequenceId,
registeredDelivery: collection.registeredDelivery ? 1 : 0,
longMessage: undefined,
}, phoneNumbers, application, collection.messageId)
));
await this.prisma.cmppInboundLongMessage.update({
where: { id: collection.groupId },
data: {
status: 'completed',
response: JSON.parse(JSON.stringify(response)) as Prisma.InputJsonValue,
completedAt: new Date(),
},
});
return response;
}
const response = await this.measureInboundStage('complete_submit', async () => (
await this.facade.recoverCompletedInboundLongMessageResponse(
collection.messageId,
@@ -233,12 +276,90 @@ async submitInboundMessage(data: GatewayInboundSubmitDto) {
throw error;
}
}
if (this.inboundFastPathEnabled()) {
return this.measureInboundStage(
'inbox_persist',
() => this.persistInboundWorkflow(data, phoneNumbers, application),
);
}
return this.measureInboundStage(
'complete_submit',
() => this.facade.submitCompleteInboundMessage(data, phoneNumbers, application),
);
}
private inboundFastPathEnabled() {
return process.env.CMPP_INBOUND_FAST_PATH_ENABLED === 'true';
}
private async persistInboundWorkflow(
data: GatewayInboundSubmitDto,
phoneNumbers: string[],
application: NonNullable<Awaited<ReturnType<SendSubmissionService['findInboundApplication']>>>,
requestedGroupMessageId?: string,
) {
const requestKey = data.requestId?.trim();
if (!requestKey || requestKey.length > 160) {
throw new BadRequestException('CMPP inbound requestId is required for fast-path idempotency');
}
if (data.remoteIp && !isIpAllowed(data.remoteIp, application.ipAllowlist.map((item) => item.ipCidr))) {
throw new BadRequestException('CMPP source IP is not in application allowlist');
}
validateInboundApplicationSrcId(data.srcId, application);
const submitGroupMessageId = requestedGroupMessageId ?? `MSG-${randomUUID()}`;
const messageIds = phoneNumbers.map((_, index) => index === 0 ? submitGroupMessageId : `MSG-${randomUUID()}`);
const payload: InboundWorkflowPayload = {
data: JSON.parse(JSON.stringify(data)) as GatewayInboundSubmitDto,
phoneNumbers,
submitGroupMessageId,
messageIds,
};
const payloadJson = JSON.parse(JSON.stringify(payload)) as Prisma.InputJsonValue;
const payloadHash = createHash('sha256').update(JSON.stringify({
data: payload.data,
phoneNumbers,
requestedGroupMessageId: requestedGroupMessageId ?? null,
})).digest('hex');
const response = {
accepted: true,
tenantId: application.tenantId,
applicationId: application.id,
taskId: '',
messageId: submitGroupMessageId,
messageRecordId: '',
status: 'accepted_pending',
phoneCount: phoneNumbers.length,
messages: phoneNumbers.map((phoneNumber, index) => ({
phoneNumber,
messageId: messageIds[index],
messageRecordId: '',
taskId: '',
status: 'accepted_pending',
})),
};
try {
await this.prisma.cmppInboundSubmissionInbox.create({
data: {
requestKey,
payloadHash,
tenantId: application.tenantId,
applicationId: application.id,
queuePriority: normalizeQueuePriority(application.queuePriority),
payload: payloadJson,
response: JSON.parse(JSON.stringify(response)) as Prisma.InputJsonValue,
},
});
return response;
} catch (error) {
if (!(error instanceof Prisma.PrismaClientKnownRequestError) || error.code !== 'P2002') throw error;
const existing = await this.prisma.cmppInboundSubmissionInbox.findUnique({ where: { requestKey } });
if (!existing || existing.payloadHash !== payloadHash) {
throw new BadRequestException('CMPP inbound requestId conflicts with another payload');
}
return existing.response as typeof response;
}
}
async recoverCompletedInboundLongMessageResponse(messageId: string, phoneNumbers: string[]) {
const existing = await this.prisma.smsMessageRecord.findMany({
where: {
@@ -289,6 +410,8 @@ async submitCompleteInboundMessage(
phoneNumbers: string[],
application: Awaited<ReturnType<SendSubmissionService['findInboundApplication']>>,
requestedGroupMessageId?: string,
requestedMessageIds?: string[],
workflowKey?: string,
) {
if (!application) {
throw new BadRequestException('CMPP account is invalid');
@@ -309,6 +432,7 @@ async submitCompleteInboundMessage(
phoneNumber: true,
status: true,
errorCode: true,
batchTask: { select: { status: true } },
},
})
: [];
@@ -318,7 +442,11 @@ async submitCompleteInboundMessage(
!persistedByPhone.has(phoneNumber) && !phoneRejections.has(phoneNumber)
)).length;
const dailyQuota = missingPhoneCount > 0
? await this.facade.tryReserveDailySendQuota(application.id, missingPhoneCount)
? await this.facade.tryReserveDailySendQuota(
application.id,
missingPhoneCount,
workflowKey ? `${workflowKey}:daily-quota` : undefined,
)
: { reserved: true, dailyLimit: application.dailyLimit ?? 100000 };
return { persistedByPhone, phoneRejections, dailyQuota, missingPhoneCount };
});
@@ -336,13 +464,19 @@ async submitCompleteInboundMessage(
persisted: persistedByPhone.get(phoneNumber),
receiptRejection: phoneRejections.get(phoneNumber),
messageId: persistedByPhone.get(phoneNumber)?.messageId
?? requestedMessageIds?.[index]
?? (index === 0 ? submitGroupMessageId : `MSG-${randomUUID()}`),
workflowItemKey: workflowKey ? `${workflowKey}:message:${index}` : undefined,
}));
const results: GatewayInboundSingleSubmitResult[] = [];
const concurrency = 10;
for (let offset = 0; offset < submissions.length; offset += concurrency) {
const batch = submissions.slice(offset, offset + concurrency);
results.push(...await Promise.all(batch.map((submission) => submission.persisted
&& !(workflowKey && (
submission.persisted.status === 'validating'
|| (submission.persisted.status === 'queued' && submission.persisted.batchTask?.status !== 'queued')
))
? Promise.resolve({
accepted: submission.persisted.errorCode !== 'DAILY_LIMIT',
tenantId: submission.persisted.tenantId ?? application.tenantId,
@@ -356,7 +490,7 @@ async submitCompleteInboundMessage(
...data,
phoneNumber: submission.phoneNumber,
phoneNumbers: undefined,
}, submission.messageId, submitGroupMessageId, application, submission.receiptRejection ? undefined : dailyLimitRejection, submission.receiptRejection))));
}, submission.messageId, submitGroupMessageId, application, submission.receiptRejection ? undefined : dailyLimitRejection, submission.receiptRejection, submission.workflowItemKey))));
}
const first = results[0];
return {
@@ -553,6 +687,7 @@ async submitInboundSingleMessage(
application: NonNullable<Awaited<ReturnType<SendSubmissionService['findInboundApplication']>>>,
synchronousRejection?: { code: string; reason: string },
receiptRejection?: { code: string; reason: string },
workflowItemKey?: string,
) {
// 入口已按账号取得并校验同一个应用快照；复用它可避免每个目标号码再次查询应用、企业和IP白名单。
if (data.remoteIp && !isIpAllowed(data.remoteIp, application.ipAllowlist.map((item) => item.ipCidr))) {
@@ -573,60 +708,101 @@ async submitInboundSingleMessage(
phoneCount: 1,
unitPrice,
});
const task = await this.measureInboundStage('task_persist', () => this.prisma.smsBatchTask.create({
data: {
tenantId: application.tenantId,
applicationId: application.id,
templateId: template?.id,
taskNo: `BT-${Date.now()}-${randomUUID().slice(0, 8)}`,
sourceType: 'cmpp',
content: data.content,
phoneTotal: 1,
status: synchronousRejection ? 'rejected' : 'validating',
auditStatus: synchronousRejection ? 'rejected' : undefined,
rejectReason: synchronousRejection?.reason,
progressTotal: 1,
},
}));
await this.measureInboundStage('api_request_persist', () => this.prisma.smsApiRequest.create({
data: {
tenantId: application.tenantId,
batchTaskId: task.id,
requestId: `REQ-${Date.now()}-${randomUUID().slice(0, 8)}`,
sourceIp: data.remoteIp,
userAgent: 'cmpp-gateway',
payloadSummary: { phoneTotal: 1, contentLength: [...data.content].length, account: data.account },
status: synchronousRejection ? 'rejected' : 'accepted',
},
}));
const drainageDetection = await this.measureInboundStage(
'content_detection',
() => detectDrainageContent(this.prisma, data.content),
);
const message = await this.measureInboundStage('message_persist', () => this.prisma.smsMessageRecord.create({
data: {
tenantId: application.tenantId,
batchTaskId: task.id,
applicationId: application.id,
templateId: template?.id,
messageId,
phoneNumber: data.phoneNumber,
content: data.content,
...drainageDetection,
billingUnits: billing.billingUnitsPerMessage,
unitPrice: receiptRejection ? 0 : billing.unitPrice,
amountCents: receiptRejection ? 0 : billing.amountCents,
const workflowDigest = workflowItemKey
? createHash('sha256').update(workflowItemKey).digest('hex').slice(0, 32)
: undefined;
let recoveredExisting = false;
let persisted;
try {
persisted = await this.measureInboundStage('message_persist', () => this.prisma.$transaction(async (tx) => {
const task = await tx.smsBatchTask.create({
data: {
tenantId: application.tenantId,
applicationId: application.id,
templateId: template?.id,
taskNo: workflowDigest ? `BT-IN-${workflowDigest}` : `BT-${Date.now()}-${randomUUID().slice(0, 8)}`,
sourceType: 'cmpp',
content: data.content,
phoneTotal: 1,
status: synchronousRejection ? 'rejected' : 'validating',
auditStatus: synchronousRejection ? 'rejected' : undefined,
rejectReason: synchronousRejection?.reason,
progressTotal: 1,
},
});
await tx.smsApiRequest.create({
data: {
tenantId: application.tenantId,
batchTaskId: task.id,
requestId: workflowDigest ? `REQ-IN-${workflowDigest}` : `REQ-${Date.now()}-${randomUUID().slice(0, 8)}`,
sourceIp: data.remoteIp,
userAgent: 'cmpp-gateway',
payloadSummary: { phoneTotal: 1, contentLength: [...data.content].length, account: data.account },
status: synchronousRejection ? 'rejected' : 'accepted',
},
});
const message = await tx.smsMessageRecord.create({
data: {
tenantId: application.tenantId,
batchTaskId: task.id,
applicationId: application.id,
templateId: template?.id,
messageId,
phoneNumber: data.phoneNumber,
content: data.content,
...drainageDetection,
billingUnits: billing.billingUnitsPerMessage,
unitPrice: receiptRejection ? 0 : billing.unitPrice,
amountCents: receiptRejection ? 0 : billing.amountCents,
queuePriority,
cmppSubmitSequenceId: data.sequenceId == null ? null : String(data.sequenceId),
cmppSubmitGroupMessageId: submitGroupMessageId,
cmppRegisteredDelivery: data.registeredDelivery !== 0,
clientSrcId,
applicationExtension: application.cmppApplicationExtension,
status: synchronousRejection ? 'rejected' : 'validating',
errorCode: synchronousRejection?.code,
errorMessage: synchronousRejection?.reason,
},
});
return { task, message };
}));
} catch (error) {
if (!workflowItemKey || !(error instanceof Prisma.PrismaClientKnownRequestError) || error.code !== 'P2002') throw error;
const existing = await this.prisma.smsMessageRecord.findUnique({
where: { messageId },
include: { batchTask: true },
});
if (!existing?.batchTask || existing.cmppSubmitGroupMessageId !== submitGroupMessageId
|| existing.phoneNumber !== data.phoneNumber || existing.applicationId !== application.id) {
throw error;
}
recoveredExisting = true;
persisted = { task: existing.batchTask, message: existing };
}
const { task, message } = persisted;
if (recoveredExisting && message.status === 'queued' && task.status !== 'queued') {
await this.facade.enqueueBatchTask(task.id, {
messageRecordId: message.id,
queuePriority,
cmppSubmitSequenceId: data.sequenceId == null ? null : String(data.sequenceId),
cmppSubmitGroupMessageId: submitGroupMessageId,
cmppRegisteredDelivery: data.registeredDelivery !== 0,
clientSrcId,
applicationExtension: application.cmppApplicationExtension,
status: synchronousRejection ? 'rejected' : 'validating',
errorCode: synchronousRejection?.code,
errorMessage: synchronousRejection?.reason,
},
}));
});
}
if (recoveredExisting && message.status !== 'validating') {
return {
accepted: message.status !== 'rejected' && message.status !== 'failed',
tenantId: application.tenantId,
applicationId: application.id,
taskId: task.id,
messageId: message.messageId,
messageRecordId: message.id,
status: message.status,
};
}
if (synchronousRejection) {
return {
@@ -658,7 +834,7 @@ async submitInboundSingleMessage(
variables: options.templateId ? templateVariables : undefined,
phoneNumber: data.phoneNumber,
sourceType: 'cmpp',
});
}, workflowItemKey ? `${workflowItemKey}:frequency` : undefined);
return { drainageInfoId: drainage?.id, risk: evaluatedRisk };
});
if (risk.status === 'rejected') {
@@ -693,6 +869,7 @@ async submitInboundSingleMessage(
relatedType: 'sms_batch_task',
relatedId: task.id,
remark: 'CMPP 入站短信冻结',
idempotencyKey: workflowItemKey ? `${workflowItemKey}:freeze` : undefined,
});
}
return check;
@@ -736,7 +913,7 @@ async submitInboundSingleMessage(
content: data.content,
phoneNumber: data.phoneNumber,
sourceType: 'cmpp',
});
}, workflowItemKey ? `${workflowItemKey}:frequency` : undefined);
if (risk.status === 'rejected') {
await reject('RISK', risk.reason || '短信被风控拒绝');
} else {
@@ -754,6 +931,7 @@ async submitInboundSingleMessage(
relatedType: 'sms_batch_task',
relatedId: task.id,
remark: 'CMPP 模板不匹配待审核短信冻结',
idempotencyKey: workflowItemKey ? `${workflowItemKey}:freeze` : undefined,
});
}
const reviewTask = risk.status === 'pending_review' && risk.task
@@ -813,7 +991,7 @@ async evaluateRiskWithPhoneFrequency(input: {
variables?: Record<string, unknown>;
phoneNumber: string;
sourceType: 'cmpp';
}) {
}, reservationKey?: string) {
const risk = await this.riskReview.evaluateTask({
tenantId: input.tenantId,
applicationId: input.applicationId,
@@ -829,6 +1007,8 @@ async evaluateRiskWithPhoneFrequency(input: {
input.applicationId,
[input.phoneNumber],
input.sourceType,
new Date(),
reservationKey,
);
const rejection = frequencyRejections.get(input.phoneNumber);
return rejection
@@ -836,6 +1016,170 @@ async evaluateRiskWithPhoneFrequency(input: {
: risk;
}
startInboundWorkflowWorker() {
if (this.inboundWorkflowTimer || this.inboundWorkflowPumping || this.inboundWorkflowTasks.size > 0) {
return { status: 'already_started' };
}
this.inboundWorkflowStopping = false;
this.scheduleInboundWorkflowPump(0);
return { status: 'started' };
}
async stopInboundWorkflowWorker() {
this.inboundWorkflowStopping = true;
if (this.inboundWorkflowTimer) clearTimeout(this.inboundWorkflowTimer);
this.inboundWorkflowTimer = undefined;
await Promise.allSettled([...this.inboundWorkflowTasks]);
}
private scheduleInboundWorkflowPump(delayMs: number) {
if (this.inboundWorkflowStopping || this.inboundWorkflowTimer) return;
this.inboundWorkflowTimer = setTimeout(() => {
this.inboundWorkflowTimer = undefined;
void this.pumpInboundWorkflow();
}, delayMs);
this.inboundWorkflowTimer.unref?.();
}
private async pumpInboundWorkflow() {
if (this.inboundWorkflowStopping || this.inboundWorkflowPumping) return;
const concurrency = positiveInteger(process.env.API_INBOUND_WORKFLOW_CONCURRENCY, 32);
this.metrics?.setInboundWorkflowSlots(concurrency, this.inboundWorkflowInFlight);
const available = Math.max(0, concurrency - this.inboundWorkflowInFlight);
if (available === 0) return;
this.inboundWorkflowPumping = true;
try {
await this.refreshInboundWorkflowMetrics();
const claimed = await this.claimInboundWorkflows(available);
for (const item of claimed) {
this.inboundWorkflowInFlight += 1;
this.metrics?.setInboundWorkflowSlots(concurrency, this.inboundWorkflowInFlight);
const task = this.processClaimedInboundWorkflow(item)
.catch((error) => this.logger.error(`CMPP inbound workflow ${item.id} failed to settle: ${String(error)}`))
.finally(() => {
this.inboundWorkflowInFlight = Math.max(0, this.inboundWorkflowInFlight - 1);
this.metrics?.setInboundWorkflowSlots(concurrency, this.inboundWorkflowInFlight);
this.inboundWorkflowTasks.delete(task);
this.scheduleInboundWorkflowPump(0);
});
this.inboundWorkflowTasks.add(task);
}
if (claimed.length === 0) {
this.scheduleInboundWorkflowPump(positiveInteger(process.env.API_INBOUND_WORKFLOW_POLL_INTERVAL_MS, 100));
}
} catch (error) {
this.logger.error(`Failed to claim CMPP inbound workflow: ${String(error)}`);
this.scheduleInboundWorkflowPump(1000);
} finally {
this.inboundWorkflowPumping = false;
if (this.inboundWorkflowInFlight < concurrency && !this.inboundWorkflowTimer) {
this.scheduleInboundWorkflowPump(0);
}
}
}
private claimInboundWorkflows(limit: number) {
const staleSeconds = positiveInteger(process.env.API_INBOUND_WORKFLOW_STALE_SECONDS, 300);
return this.prisma.$queryRaw<ClaimedInboundWorkflow[]>(Prisma.sql`
WITH candidates AS (
SELECT id
FROM "CmppInboundSubmissionInbox"
WHERE (
status = 'pending'
AND "nextAttemptAt" <= NOW()
) OR (
status = 'processing'
AND "lockedAt" <= NOW() - make_interval(secs => ${staleSeconds})
)
-- Priority applications enter the same durable Inbox, but are claimed first while
-- preserving FIFO within each class. This keeps the V5 priority contract effective
-- before BullMQ without adding another non-durable queue.
ORDER BY CASE WHEN "queuePriority" = 'priority' THEN 0 ELSE 1 END, "createdAt" ASC
LIMIT ${limit}
FOR UPDATE SKIP LOCKED
)
UPDATE "CmppInboundSubmissionInbox" AS inbox
SET status = 'processing',
attempts = inbox.attempts + 1,
"lockedAt" = NOW(),
"lockedBy" = ${this.inboundWorkflowWorkerId},
"updatedAt" = NOW()
FROM candidates
WHERE inbox.id = candidates.id
RETURNING inbox.id, inbox."requestKey", inbox."applicationId", inbox.attempts, inbox.payload
`);
}
private async processClaimedInboundWorkflow(item: ClaimedInboundWorkflow) {
try {
const payload = parseInboundWorkflowPayload(item.payload);
const application = await this.facade.findInboundApplication(payload.data.account);
if (!application || application.id !== item.applicationId) {
throw new Error('CMPP inbound application no longer matches persisted workflow');
}
const result = await this.facade.submitCompleteInboundMessage(
payload.data,
payload.phoneNumbers,
application,
payload.submitGroupMessageId,
payload.messageIds,
item.requestKey,
);
const settled = await this.prisma.cmppInboundSubmissionInbox.updateMany({
where: { id: item.id, status: 'processing', lockedBy: this.inboundWorkflowWorkerId },
data: {
status: 'completed',
result: JSON.parse(JSON.stringify(result)) as Prisma.InputJsonValue,
completedAt: new Date(),
lockedAt: null,
lockedBy: null,
lastError: null,
},
});
if (settled.count !== 1) throw new Error('CMPP inbound workflow lease was lost before completion');
this.metrics?.recordInboundWorkflowResult('completed');
} catch (error) {
const reason = (error instanceof Error ? error.message : String(error)).slice(0, 2000);
const delayMs = Math.min(60_000, 250 * 2 ** Math.min(8, Math.max(0, item.attempts - 1)));
const released = await this.prisma.cmppInboundSubmissionInbox.updateMany({
where: { id: item.id, status: 'processing', lockedBy: this.inboundWorkflowWorkerId },
data: {
status: 'pending',
nextAttemptAt: new Date(Date.now() + delayMs),
lockedAt: null,
lockedBy: null,
lastError: reason,
},
});
if (released.count === 1) {
this.metrics?.recordInboundWorkflowResult('retry');
this.logger.warn(`CMPP inbound workflow ${item.id} will retry after attempt ${item.attempts}: ${reason}`);
return;
}
throw error;
}
}
private async refreshInboundWorkflowMetrics() {
const now = Date.now();
if (now - this.inboundWorkflowMetricsUpdatedAt < 5_000) return;
this.inboundWorkflowMetricsUpdatedAt = now;
const [pending, processing, oldest] = await Promise.all([
this.prisma.cmppInboundSubmissionInbox.count({ where: { status: 'pending' } }),
this.prisma.cmppInboundSubmissionInbox.count({ where: { status: 'processing' } }),
this.prisma.cmppInboundSubmissionInbox.findFirst({
where: { status: 'pending' },
select: { createdAt: true },
orderBy: { createdAt: 'asc' },
}),
]);
this.metrics?.setInboundWorkflowState(
pending,
processing,
oldest ? Math.max(0, (now - oldest.createdAt.getTime()) / 1000) : 0,
);
}
findInboundApplication(account: string) {
return this.prisma.smsApplication.findFirst({
where: { cmppAccount: account },
@@ -917,3 +1261,24 @@ async attachMessageToReviewTask(reviewTaskId: string, messageRecordId: string, s
return this.prisma.smsSendTask.findUnique({ where: { id: reviewTaskId } });
}
}
function parseInboundWorkflowPayload(value: Prisma.JsonValue): InboundWorkflowPayload {
if (!value || typeof value !== 'object' || Array.isArray(value)) throw new Error('CMPP inbound workflow payload is invalid');
const data = value.data;
const phoneNumbers = value.phoneNumbers;
const submitGroupMessageId = value.submitGroupMessageId;
const messageIds = value.messageIds;
if (!data || typeof data !== 'object' || Array.isArray(data)
|| !Array.isArray(phoneNumbers) || phoneNumbers.some((item) => typeof item !== 'string')
|| typeof submitGroupMessageId !== 'string'
|| !Array.isArray(messageIds) || messageIds.some((item) => typeof item !== 'string')
|| phoneNumbers.length === 0 || phoneNumbers.length !== messageIds.length) {
throw new Error('CMPP inbound workflow payload fields are invalid');
}
return {
data: data as unknown as GatewayInboundSubmitDto,
phoneNumbers: phoneNumbers as string[],
submitGroupMessageId,
messageIds: messageIds as string[],
};
}
api/src/send-chain/send-submission.service.ts
+21 -7
View File
@@ -67,7 +67,10 @@ export class SendSubmissionService {
}
onModuleDestroy() {
return this.gatewaySubmit.onModuleDestroy();
return Promise.all([
this.gatewaySubmit.onModuleDestroy(),
this.inboundEntry.stopInboundWorkflowWorker(),
]);
}
async createBatchTask(data: CreateBatchTaskDto) {
@@ -123,8 +126,8 @@ async reserveDailySendQuota(applicationId: string, requestedCount: number) {
return this.batchEntry.reserveDailySendQuota(applicationId, requestedCount);
}
async tryReserveDailySendQuota(applicationId: string, requestedCount: number) {
return this.batchEntry.tryReserveDailySendQuota(applicationId, requestedCount);
async tryReserveDailySendQuota(applicationId: string, requestedCount: number, reservationKey?: string) {
return this.batchEntry.tryReserveDailySendQuota(applicationId, requestedCount, reservationKey);
}
async authenticateInboundApplication(data: GatewayInboundAuthDto) {
@@ -144,8 +147,10 @@ async submitCompleteInboundMessage(
phoneNumbers: string[],
application: Awaited<ReturnType<SendSubmissionService['findInboundApplication']>>,
requestedGroupMessageId?: string,
requestedMessageIds?: string[],
workflowKey?: string,
) {
return this.inboundEntry.submitCompleteInboundMessage(data, phoneNumbers, application, requestedGroupMessageId);
return this.inboundEntry.submitCompleteInboundMessage(data, phoneNumbers, application, requestedGroupMessageId, requestedMessageIds, workflowKey);
}
async collectInboundLongMessageFragment(
@@ -167,8 +172,9 @@ async submitInboundSingleMessage(
application: NonNullable<Awaited<ReturnType<SendSubmissionService['findInboundApplication']>>>,
synchronousRejection?: { code: string; reason: string },
receiptRejection?: { code: string; reason: string },
workflowItemKey?: string,
) {
return this.inboundEntry.submitInboundSingleMessage(data, messageId, submitGroupMessageId, application, synchronousRejection, receiptRejection);
return this.inboundEntry.submitInboundSingleMessage(data, messageId, submitGroupMessageId, application, synchronousRejection, receiptRejection, workflowItemKey);
}
async evaluateRiskWithPhoneFrequency(input: {
@@ -179,8 +185,8 @@ async evaluateRiskWithPhoneFrequency(input: {
variables?: Record<string, unknown>;
phoneNumber: string;
sourceType: 'cmpp';
}) {
return this.inboundEntry.evaluateRiskWithPhoneFrequency(input);
}, reservationKey?: string) {
return this.inboundEntry.evaluateRiskWithPhoneFrequency(input, reservationKey);
}
findInboundApplication(account: string) {
@@ -223,6 +229,14 @@ startWorker() {
return this.gatewaySubmit.startWorker();
}
startInboundWorkflowWorker() {
return this.inboundEntry.startInboundWorkflowWorker();
}
stopInboundWorkflowWorker() {
return this.inboundEntry.stopInboundWorkflowWorker();
}
async processSendJob(job: SendJob) {
return this.gatewaySubmit.processSendJob(job);
}
api/src/send-worker.module.ts
+20
View File
@@ -0,0 +1,20 @@
import { Module } from '@nestjs/common';
import { ConfigModule } from '@nestjs/config';
import { BillingModule } from './billing/billing.module';
import { PhoneRoutingLookupService } from './dictionaries/phone-routing-lookup.service';
import { MetricsModule } from './metrics/metrics.module';
import { PrismaModule } from './prisma/prisma.module';
import { PhoneFrequencyService } from './risk-review/phone-frequency.service';
import { RiskReviewService } from './risk-review/risk-review.service';
import { SendChainService } from './send-chain/send-chain.service';
@Module({
imports: [
ConfigModule.forRoot({ isGlobal: true, envFilePath: ['.env.local', '.env'] }),
PrismaModule,
BillingModule,
MetricsModule,
],
providers: [RiskReviewService, PhoneFrequencyService, PhoneRoutingLookupService, SendChainService],
})
export class SendWorkerModule {}
api/src/send-worker.ts
+45
View File
@@ -0,0 +1,45 @@
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';
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?.trim() || 'worker') !== 'worker') {
throw new Error('send-worker requires CMPP_PROCESS_ROLE=worker');
}
const app = await NestFactory.createApplicationContext(SendWorkerModule);
app.enableShutdownHooks();
const metrics = app.get(MetricsService);
const host = process.env.API_WORKER_METRICS_HOST?.trim() || '127.0.0.1';
const port = Number(process.env.API_WORKER_METRICS_PORT ?? 9465);
const metricsServer = 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) => {
metricsServer.once('error', reject);
metricsServer.listen(port, host, resolve);
});
const close = async () => {
await new Promise<void>((resolve) => metricsServer.close(() => resolve()));
await app.close();
};
process.once('SIGTERM', () => void close());
process.once('SIGINT', () => void close());
}
void bootstrap();