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

perf: merge CMPP inbox validation and persistence

Browse Source
This commit is contained in:
hectorzhao
2026-08-20 18:55:09 +08:00
parent 633e7a7055
commit 99fb346566
6 changed files with 303 additions and 32 deletions
Download Patch File Download Diff File Expand all files Collapse all files
api/src/send-chain/send-chain.service.spec.ts
+82 -22
View File
@@ -2119,6 +2119,24 @@ describe('SendChainService', () => {
try {
const { service, prisma, billing, riskReview, phoneFrequency } = createService();
service.enqueueBatchTask = jest.fn();
prisma.$queryRaw.mockImplementationOnce((query) => {
const payloadHash = query.values.find((value: unknown) => typeof value === 'string' && /^[a-f0-9]{64}$/.test(value));
return Promise.resolve([{
validationError: null,
payloadHash,
response: {
accepted: true,
tenantId: 'tenant-1',
applicationId: 'app-1',
taskId: '',
messageId: 'MSG-fast',
messageRecordId: '',
status: 'accepted_pending',
phoneCount: 2,
messages: [],
},
}]);
});
const result = await service.submitInboundMessage({
requestId: 'cmpp-inbound:test-fast-path',
@@ -2134,13 +2152,12 @@ describe('SendChainService', () => {
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.smsApplication.findFirst).not.toHaveBeenCalled();
expect(prisma.$queryRaw).toHaveBeenCalledTimes(1);
const sql = prisma.$queryRaw.mock.calls[0][0].strings.join(' ');
expect(sql).toContain('INSERT INTO "CmppInboundSubmissionInbox"');
expect(sql).toContain('JOIN "Tenant"');
expect(sql).toContain('ON CONFLICT ("requestKey") DO NOTHING');
expect(prisma.smsBatchTask.create).not.toHaveBeenCalled();
expect(prisma.smsMessageRecord.create).not.toHaveBeenCalled();
expect(riskReview.evaluateTask).not.toHaveBeenCalled();
@@ -2158,7 +2175,6 @@ describe('SendChainService', () => {
process.env.CMPP_INBOUND_FAST_PATH_ENABLED = 'true';
try {
const { service, prisma } = createService();
let originalHash = '';
const storedResponse = {
accepted: true,
tenantId: 'tenant-1',
@@ -2170,19 +2186,10 @@ describe('SendChainService', () => {
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,
}));
prisma.$queryRaw.mockImplementation((query) => {
const payloadHash = query.values.find((value: unknown) => typeof value === 'string' && /^[a-f0-9]{64}$/.test(value));
return Promise.resolve([{ validationError: null, payloadHash, response: storedResponse }]);
});
const request = {
requestId: 'cmpp-inbound:test-retry',
account: '100001',
@@ -2194,7 +2201,8 @@ describe('SendChainService', () => {
await service.submitInboundMessage(request);
await expect(service.submitInboundMessage(request)).resolves.toEqual(storedResponse);
expect(prisma.cmppInboundSubmissionInbox.create).toHaveBeenCalledTimes(2);
expect(prisma.$queryRaw).toHaveBeenCalledTimes(2);
expect(prisma.cmppInboundSubmissionInbox.findUnique).not.toHaveBeenCalled();
expect(prisma.smsMessageRecord.create).not.toHaveBeenCalled();
} finally {
if (previous == null) delete process.env.CMPP_INBOUND_FAST_PATH_ENABLED;
@@ -2202,6 +2210,58 @@ describe('SendChainService', () => {
}
});
it('rejects a disabled application before the merged Inbox statement can insert', async () => {
const previous = process.env.CMPP_INBOUND_FAST_PATH_ENABLED;
process.env.CMPP_INBOUND_FAST_PATH_ENABLED = 'true';
try {
const { service, prisma } = createService();
prisma.$queryRaw.mockResolvedValueOnce([{
validationError: 'CMPP account is disabled for new submissions',
payloadHash: null,
response: null,
}]);
await expect(service.submitInboundMessage({
requestId: 'cmpp-inbound:disabled',
account: '100001',
phoneNumber: '13800000001',
content: 'hello',
})).rejects.toThrow('CMPP account is disabled for new submissions');
expect(prisma.cmppInboundSubmissionInbox.findUnique).not.toHaveBeenCalled();
} finally {
if (previous == null) delete process.env.CMPP_INBOUND_FAST_PATH_ENABLED;
else process.env.CMPP_INBOUND_FAST_PATH_ENABLED = previous;
}
});
it('uses a read-only recovery only for a concurrent Inbox insert outside the CTE snapshot', async () => {
const previous = process.env.CMPP_INBOUND_FAST_PATH_ENABLED;
process.env.CMPP_INBOUND_FAST_PATH_ENABLED = 'true';
try {
const { service, prisma } = createService();
let payloadHash = '';
prisma.$queryRaw.mockImplementationOnce((query) => {
payloadHash = query.values.find((value: unknown) => typeof value === 'string' && /^[a-f0-9]{64}$/.test(value));
return Promise.resolve([{ validationError: null, payloadHash: null, response: null }]);
});
prisma.cmppInboundSubmissionInbox.findUnique.mockImplementationOnce(() => Promise.resolve({
payloadHash,
response: { accepted: true, messageId: 'MSG-concurrent', status: 'accepted_pending' },
}));
await expect(service.submitInboundMessage({
requestId: 'cmpp-inbound:concurrent',
account: '100001',
phoneNumber: '13800000001',
content: 'hello',
})).resolves.toEqual(expect.objectContaining({ messageId: 'MSG-concurrent' }));
expect(prisma.cmppInboundSubmissionInbox.findUnique).toHaveBeenCalledTimes(1);
} finally {
if (previous == null) delete process.env.CMPP_INBOUND_FAST_PATH_ENABLED;
else process.env.CMPP_INBOUND_FAST_PATH_ENABLED = previous;
}
});
it('persists an idempotent daily quota reservation with a Prisma Date value', async () => {
const { service, prisma } = createService();
prisma.$queryRaw.mockResolvedValueOnce([{ tenantId: 'tenant-1', dailyLimit: 100000, usedCount: 1 }]);
api/src/send-chain/send-inbound-entry.service.ts
+193 -10
View File
@@ -33,6 +33,26 @@ type ClaimedInboundWorkflow = {
payload: Prisma.JsonValue;
};
type PersistedInboundWorkflowRow = {
validationError: string | null;
payloadHash: string | null;
response: Prisma.JsonValue | null;
};
type InboundApplication = NonNullable<Awaited<ReturnType<SendSubmissionService['findInboundApplication']>>>;
type InboundWorkflowResponse = {
accepted: boolean;
tenantId: string;
applicationId: string;
taskId: string;
messageId: string;
messageRecordId: string;
status: string;
phoneCount: number;
messages: Array<{ phoneNumber: string; messageId: string; messageRecordId: string; taskId: string; status: string }>;
};
/**
* R9 inboundEntry implementation. Cross-method calls return through the stable SendChainService seam.
*/
@@ -183,6 +203,13 @@ async submitInboundMessage(data: GatewayInboundSubmitDto) {
throw new BadRequestException('CMPP submit phone number is invalid');
}
if (this.inboundFastPathEnabled() && !data.longMessage) {
return this.measureInboundStage(
'inbox_persist',
() => this.persistValidatedInboundWorkflow(data, phoneNumbers),
);
}
const application = await this.measureInboundStage(
'application_lookup',
() => this.facade.findInboundApplication(data.account),
@@ -276,12 +303,6 @@ 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),
@@ -292,6 +313,162 @@ async submitInboundMessage(data: GatewayInboundSubmitDto) {
return process.env.CMPP_INBOUND_FAST_PATH_ENABLED === 'true';
}
private async persistValidatedInboundWorkflow(
data: GatewayInboundSubmitDto,
phoneNumbers: string[],
) {
const requestKey = data.requestId?.trim();
if (!requestKey || requestKey.length > 160) {
throw new BadRequestException('CMPP inbound requestId is required for fast-path idempotency');
}
const submitGroupMessageId = `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 payloadHash = createHash('sha256').update(JSON.stringify({
data: payload.data,
phoneNumbers,
requestedGroupMessageId: null,
})).digest('hex');
const responseMessages = phoneNumbers.map((phoneNumber, index) => ({
phoneNumber,
messageId: messageIds[index],
messageRecordId: '',
taskId: '',
status: 'accepted_pending',
}));
const remoteIp = data.remoteIp?.replace(/^::ffff:/, '').trim() || null;
const submittedSrcId = data.srcId?.trim() ?? '';
// One indexed statement must both validate the current application state and create the
// durable Inbox row. Keeping those operations in one database snapshot prevents a disable
// racing between a separate SELECT and INSERT, while the unique request key remains the
// authoritative idempotency boundary.
const rows = await this.prisma.$queryRaw<PersistedInboundWorkflowRow[]>(Prisma.sql`
WITH application AS (
SELECT app.id,
app."tenantId",
app.status,
app."interfaceEnabled",
app."queuePriority",
app."cmppApplicationExtension",
app."cmppAccessNumberFillEnabled",
app."cmppAccessNumberFillPrefix",
app."cmppClientSrcId",
tenant.status AS "tenantStatus"
FROM "SmsApplication" AS app
JOIN "Tenant" AS tenant ON tenant.id = app."tenantId"
WHERE app."cmppAccount" = ${data.account}
LIMIT 1
), validation AS (
SELECT application.*,
CASE
WHEN application.status <> 'active'
OR application."tenantStatus" <> 'active'
OR NOT application."interfaceEnabled"
THEN 'CMPP account is disabled for new submissions'
WHEN ${remoteIp}::text IS NOT NULL
AND EXISTS (
SELECT 1 FROM "SmsApplicationIpAllowlist" allowlist
WHERE allowlist."applicationId" = application.id
)
AND NOT EXISTS (
SELECT 1
FROM "SmsApplicationIpAllowlist" allowlist
WHERE allowlist."applicationId" = application.id
AND (
(position('/' IN trim(allowlist."ipCidr")) = 0
AND regexp_replace(trim(allowlist."ipCidr"), '^::ffff:', '') = ${remoteIp})
OR (
trim(allowlist."ipCidr") ~ '^[0-9]{1,3}(\\.[0-9]{1,3}){3}/([0-9]|[12][0-9]|3[0-2])$'
AND ${remoteIp} ~ '^[0-9]{1,3}(\\.[0-9]{1,3}){3}$'
AND ${remoteIp}::inet <<= trim(allowlist."ipCidr")::cidr
)
)
)
THEN 'CMPP source IP is not in application allowlist'
WHEN coalesce(trim(application."cmppApplicationExtension"), '') <> ''
AND ${submittedSrcId} <> coalesce(
nullif(trim(application."cmppClientSrcId"), ''),
CASE WHEN application."cmppAccessNumberFillEnabled"
THEN coalesce(trim(application."cmppAccessNumberFillPrefix"), '')
ELSE ''
END || trim(application."cmppApplicationExtension")
)
THEN 'CMPP Src_Id must equal the access number assigned to this application: ' || coalesce(
nullif(trim(application."cmppClientSrcId"), ''),
CASE WHEN application."cmppAccessNumberFillEnabled"
THEN coalesce(trim(application."cmppAccessNumberFillPrefix"), '')
ELSE ''
END || trim(application."cmppApplicationExtension")
)
ELSE NULL
END AS "validationError"
FROM application
), inserted AS (
INSERT INTO "CmppInboundSubmissionInbox" (
id, "requestKey", "payloadHash", "tenantId", "applicationId", "queuePriority",
payload, response, status, attempts, "nextAttemptAt", "createdAt", "updatedAt"
)
SELECT ${randomUUID()}, ${requestKey}, ${payloadHash}, validation."tenantId", validation.id,
CASE WHEN validation."queuePriority" = 'priority' THEN 'priority' ELSE 'normal' END,
${JSON.stringify(payload)}::jsonb,
jsonb_build_object(
'accepted', true,
'tenantId', validation."tenantId",
'applicationId', validation.id,
'taskId', '',
'messageId', ${submitGroupMessageId},
'messageRecordId', '',
'status', 'accepted_pending',
'phoneCount', ${phoneNumbers.length},
'messages', ${JSON.stringify(responseMessages)}::jsonb
),
'pending', 0,
(NOW() AT TIME ZONE 'UTC'),
(NOW() AT TIME ZONE 'UTC'),
(NOW() AT TIME ZONE 'UTC')
FROM validation
WHERE validation."validationError" IS NULL
ON CONFLICT ("requestKey") DO NOTHING
RETURNING "payloadHash", response
)
SELECT validation."validationError",
coalesce(inserted."payloadHash", existing."payloadHash") AS "payloadHash",
coalesce(inserted.response, existing.response) AS response
FROM validation
LEFT JOIN inserted ON true
LEFT JOIN "CmppInboundSubmissionInbox" existing
ON existing."requestKey" = ${requestKey}
LIMIT 1
`);
const row = rows[0];
if (!row) {
throw new BadRequestException('CMPP account is invalid');
}
if (row.validationError) {
throw new BadRequestException(row.validationError);
}
if (row.response == null) {
// A concurrent insert that wins after this statement's MVCC snapshot is not visible to
// the CTE. The unique key has already prevented duplication, so only that rare retry path
// needs a second read; normal submissions still use exactly one database round trip.
const concurrent = await this.prisma.cmppInboundSubmissionInbox.findUnique({ where: { requestKey } });
if (!concurrent || concurrent.payloadHash !== payloadHash) {
throw new BadRequestException('CMPP inbound requestId conflicts with another payload');
}
return concurrent.response as InboundWorkflowResponse;
}
if (row.payloadHash !== payloadHash) {
throw new BadRequestException('CMPP inbound requestId conflicts with another payload');
}
return row.response as InboundWorkflowResponse;
}
private async persistInboundWorkflow(
data: GatewayInboundSubmitDto,
phoneNumbers: string[],
@@ -1051,10 +1228,17 @@ startInboundWorkflowWorker() {
try {
await this.refreshInboundWorkflowMetrics();
const claimed = await this.claimInboundWorkflows(available);
const applications = claimed.length === 0
? []
: await this.prisma.smsApplication.findMany({
where: { id: { in: [...new Set(claimed.map((item) => item.applicationId))] } },
include: { tenant: true, ipAllowlist: true },
});
const applicationById = new Map(applications.map((application) => [application.id, application]));
for (const item of claimed) {
this.inboundWorkflowInFlight += 1;
this.metrics?.setInboundWorkflowSlots(concurrency, this.inboundWorkflowInFlight);
const task = this.processClaimedInboundWorkflow(item)
const task = this.processClaimedInboundWorkflow(item, applicationById.get(item.applicationId))
.catch((error) => this.logger.error(`CMPP inbound workflow ${item.id} failed to settle: ${String(error)}`))
.finally(() => {
this.inboundWorkflowInFlight = Math.max(0, this.inboundWorkflowInFlight - 1);
@@ -1110,11 +1294,10 @@ startInboundWorkflowWorker() {
`);
}
private async processClaimedInboundWorkflow(item: ClaimedInboundWorkflow) {
private async processClaimedInboundWorkflow(item: ClaimedInboundWorkflow, application?: InboundApplication) {
try {
const payload = parseInboundWorkflowPayload(item.payload);
const application = await this.facade.findInboundApplication(payload.data.account);
if (!application || application.id !== item.applicationId) {
if (!application || application.cmppAccount !== payload.data.account) {
throw new Error('CMPP inbound application no longer matches persisted workflow');
}
const result = await this.facade.submitCompleteInboundMessage(