diff --git a/api/src/send-chain/send-chain.service.spec.ts b/api/src/send-chain/send-chain.service.spec.ts index 7ff9f6c..02a2bbe 100644 --- a/api/src/send-chain/send-chain.service.spec.ts +++ b/api/src/send-chain/send-chain.service.spec.ts @@ -106,6 +106,7 @@ function createPrismaMock() { findMany: jest.fn(), }, smsMessageRecord: { + create: jest.fn().mockImplementation(({ data }) => Promise.resolve({ ...message, ...data })), createMany: jest.fn().mockResolvedValue({ count: 2 }), findMany: jest.fn().mockResolvedValue([{ id: 'record-1', batchTaskId: 'task-1' }]), findUnique: jest.fn().mockResolvedValue(message), @@ -149,6 +150,7 @@ function createPrismaMock() { }, smsReceiptRecord: { create: jest.fn().mockResolvedValue({ id: 'receipt-1' }), + findFirst: jest.fn().mockResolvedValue(null), findMany: jest.fn(), }, smsUplinkMessage: { @@ -474,7 +476,7 @@ describe('SendChainService', () => { })).rejects.toThrow('CMPP interface is disabled for this application'); }); - it('rejects Gateway submit when application interface is disabled', async () => { + it('records and acknowledges Gateway submit with a failure receipt when the application interface was disabled after bind', async () => { const { service, prisma } = createService(); prisma.smsApplication.findFirst.mockResolvedValue({ id: 'app-1', @@ -493,9 +495,35 @@ describe('SendChainService', () => { phoneNumber: '13800000001', content: 'hello', remoteIp: '127.0.0.1', - })).rejects.toThrow('CMPP interface is disabled for this application'); + })).resolves.toEqual(expect.objectContaining({ accepted: true, messageRecordId: 'record-1' })); - expect(prisma.smsBatchTask.create).not.toHaveBeenCalled(); + expect(prisma.smsMessageRecord.create).toHaveBeenCalledWith({ data: expect.objectContaining({ status: 'validating' }) }); + expect(prisma.smsReceiptRecord.create).toHaveBeenCalledWith({ + data: expect.objectContaining({ messageRecordId: 'record-1', receiptStatus: 'undelivered', errorCode: 'INTERFACE' }), + }); + expect(prisma.cmppDownstreamDelivery.create).toHaveBeenCalledWith({ + data: expect.objectContaining({ messageRecordId: 'record-1', deliveryType: 'receipt', status: 'pending' }), + }); + }); + + it('records an unreported CMPP message and returns success before delivering the template failure receipt', async () => { + const { service, prisma } = createService(); + prisma.smsTemplate.findFirst.mockResolvedValue(null); + + await expect(service.submitInboundMessage({ + account: '100001', + phoneNumber: '13800000001', + content: 'unreported content', + remoteIp: '127.0.0.1', + })).resolves.toEqual(expect.objectContaining({ accepted: true, messageRecordId: 'record-1' })); + + expect(prisma.smsReceiptRecord.create).toHaveBeenCalledWith({ + data: expect.objectContaining({ receiptStatus: 'undelivered', rawStatus: 'REJECTD', errorCode: 'TEMPLATE' }), + }); + expect(service['postGatewayControl']).toHaveBeenCalledWith( + '/downstream/receipt', + expect.objectContaining({ receiptStatus: 'undelivered', rawStatus: 'REJECTD', errorCode: 'TEMPLATE' }), + ); }); it('previews imported phone files with duplicate, invalid, blacklist, and variable errors', async () => { @@ -896,11 +924,31 @@ describe('SendChainService', () => { service['waitForChannelRateLimit'] = jest.fn().mockResolvedValue(undefined); service['getGatewayQueue'] = jest.fn().mockReturnValue({ add: gatewayAdd }); prisma.channelSignatureReportTask.findFirst.mockResolvedValue(null); + prisma.smsMessageRecord.findUnique.mockResolvedValue({ + id: 'record-1', + tenantId: 'tenant-1', + batchTaskId: 'task-1', + applicationId: 'app-1', + templateId: 'tpl-1', + messageId: 'MSG-1', + phoneNumber: '13800000001', + content: 'hello', + billingUnits: 1, + unitPrice: 3, + amountCents: 3, + status: 'queued', + queuePriority: 'normal', + batchTask: { sourceType: 'cmpp' }, + 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(prisma.smsReceiptRecord.create).toHaveBeenCalledWith({ + data: expect.objectContaining({ receiptStatus: 'undelivered', errorCode: 'ROUTE' }), + }); }); it('only selects channel group items allocated to the matched carrier', async () => { diff --git a/api/src/send-chain/send-chain.service.ts b/api/src/send-chain/send-chain.service.ts index 5f580d5..eb4062b 100644 --- a/api/src/send-chain/send-chain.service.ts +++ b/api/src/send-chain/send-chain.service.ts @@ -626,13 +626,20 @@ export class SendChainService implements OnModuleInit, OnModuleDestroy { data: { status: 'failed', errorMessage: reason }, }); await this.releaseMessageReservation(businessMessage, reason); - await this.refreshTaskProgress(businessMessage.batchTaskId); + if (message.batchTask?.sourceType === 'cmpp') { + await this.recordCmppFailureReceipt(businessMessage, 'ROUTE', reason); + } else { + await this.refreshTaskProgress(businessMessage.batchTaskId); + } return { submitted: false, messageRecordId: message.id, status: 'failed', reason }; } } async handleSubmitResult(data: GatewaySubmitResultDto) { const message = await this.requireMessageByGatewayEvent(data.messageId, data.gatewayMessageId); + const batchTask = message.batchTaskId + ? await this.prisma.smsBatchTask.findUnique({ where: { id: message.batchTaskId }, select: { sourceType: true } }) + : null; const submittedAt = data.submittedAt ? new Date(data.submittedAt) : new Date(); await this.prisma.smsSubmitRecord.updateMany({ where: { OR: [{ submitId: data.submitId }, { messageRecordId: message.id }] }, @@ -675,6 +682,13 @@ export class SendChainService implements OnModuleInit, OnModuleDestroy { timeoutAt: data.submitStatus === 'timeout' ? submittedAt : undefined, }, }); + if (data.submitStatus !== 'accepted' && batchTask?.sourceType === 'cmpp' && message.tenantId && message.applicationId) { + await this.recordCmppFailureReceipt( + message as typeof message & { tenantId: string; applicationId: string; batchTaskId: string }, + data.errorCode || 'SUBMIT', + data.errorMessage || (data.submitStatus === 'timeout' ? '上游提交超时' : '上游拒绝短信'), + ); + } await this.prisma.gatewaySubmitDeadLetter.updateMany({ where: { status: { in: ['pending', 'requeued'] }, @@ -1422,11 +1436,8 @@ export class SendChainService implements OnModuleInit, OnModuleDestroy { async submitInboundMessage(data: GatewayInboundSubmitDto) { const application = await this.findInboundApplication(data.account); - if (!application || application.status !== 'active' || application.tenant.status !== 'active') { - throw new BadRequestException('CMPP account is invalid or disabled'); - } - if (!application.interfaceEnabled) { - throw new BadRequestException('CMPP interface is disabled for this application'); + if (!application) { + throw new BadRequestException('CMPP account is invalid'); } if (data.remoteIp && !isApplicationIpAllowed(data.remoteIp, application.ipAllowlist.map((item) => item.ipCidr))) { throw new BadRequestException('CMPP source IP is not in application allowlist'); @@ -1434,26 +1445,130 @@ export class SendChainService implements OnModuleInit, OnModuleDestroy { if (!/^1[3-9]\d{9}$/.test(data.phoneNumber)) { throw new BadRequestException('CMPP submit phone number is invalid'); } - const template = await this.resolveInboundTemplate(application.id, data.content); - const task = await this.createBatchTask({ + const template = await this.resolveInboundTemplateCandidate(application.id, data.content); + const unitPrice = application.customerUnitPrice ?? 0; + const queuePriority = normalizeQueuePriority(application.queuePriority); + const billing = this.billing.estimateSmsCost({ tenantId: application.tenantId, applicationId: application.id, - templateId: template.id, content: data.content, - phones: [data.phoneNumber], - sourceType: 'cmpp', - sourceIp: data.remoteIp, - userAgent: 'cmpp-gateway', + phoneCount: 1, + unitPrice, }); - const message = task?.messages?.[0]; + const task = await 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: 'validating', + progressTotal: 1, + }, + }); + await 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: 'accepted', + }, + }); + const message = await this.prisma.smsMessageRecord.create({ + data: { + tenantId: application.tenantId, + batchTaskId: task.id, + applicationId: application.id, + templateId: template?.id, + messageId: `MSG-${randomUUID()}`, + phoneNumber: data.phoneNumber, + content: data.content, + billingUnits: billing.billingUnitsPerMessage, + unitPrice: billing.unitPrice, + amountCents: billing.amountCents, + queuePriority, + status: 'validating', + }, + }); + + const reject = async (code: string, reason: string) => { + await this.prisma.smsBatchTask.update({ + where: { id: task.id }, + data: { status: 'rejected', auditStatus: 'rejected', rejectReason: reason }, + }); + await this.recordCmppFailureReceipt(message, code, reason); + }; + if (application.status !== 'active' || application.tenant.status !== 'active') { + await reject('ACCOUNT', '企业或短信应用已停用'); + } else if (!application.interfaceEnabled) { + await reject('INTERFACE', '短信应用 CMPP 接口已停用'); + } else if (application.tenant.certificationStatus !== 'approved') { + await reject('CERT', '企业认证未通过'); + } else if (!template) { + await reject('TEMPLATE', '短信内容未匹配到已报备模板'); + } else if (template.auditStatus !== 'approved') { + await reject('TEMPLATE', '短信模板尚未审核通过'); + } else if (!template.signature || template.signature.auditStatus !== 'approved') { + await reject('SIGNATURE', '短信签名尚未审核通过'); + } else if (template.signature.reportStatus !== 'approved') { + await reject('REPORT', '短信签名尚未报备通过'); + } else { + const risk = await this.riskReview.evaluateTask({ + tenantId: application.tenantId, + applicationId: application.id, + templateId: template.id, + content: data.content, + phones: [data.phoneNumber], + }); + if (risk.status === 'rejected') { + await reject('RISK', risk.reason || '短信被风控拒绝'); + } else if (risk.status === 'pending_review') { + await this.prisma.smsMessageRecord.update({ where: { id: message.id }, data: { status: 'pending_review' } }); + await this.prisma.smsBatchTask.update({ + where: { id: task.id }, + data: { status: 'pending_review', riskTaskId: risk.task?.id, auditStatus: 'pending', reviewReason: risk.reason }, + }); + } else { + const accountCheck = await this.billing.checkAccount({ + tenantId: application.tenantId, + amountCents: billing.amountCents, + smsUnits: billing.totalBillingUnits, + }); + if (!accountCheck.canSend) { + await reject('BALANCE', '企业账户余额、套餐余量或授信额度不足'); + } else { + if (billing.amountCents + billing.totalBillingUnits > 0) { + await this.billing.freeze({ + tenantId: application.tenantId, + amountCents: billing.amountCents, + smsUnits: billing.totalBillingUnits, + relatedType: 'sms_batch_task', + relatedId: task.id, + remark: 'CMPP 入站短信冻结', + }); + } + await this.prisma.smsMessageRecord.update({ where: { id: message.id }, data: { status: 'queued' } }); + await this.prisma.smsBatchTask.update({ + where: { id: task.id }, + data: { status: 'ready', riskTaskId: risk.task?.id, auditStatus: 'approved' }, + }); + await this.enqueueBatchTask(task.id); + } + } + } return { accepted: true, tenantId: application.tenantId, applicationId: application.id, - taskId: task?.id, - messageId: message?.messageId, - messageRecordId: message?.id, - status: message?.status ?? task?.status, + taskId: task.id, + messageId: message.messageId, + messageRecordId: message.id, + status: 'accepted', }; } @@ -1762,24 +1877,71 @@ export class SendChainService implements OnModuleInit, OnModuleDestroy { }); } - private async resolveInboundTemplate(applicationId: string, content: string) { - const template = await this.prisma.smsTemplate.findFirst({ + private resolveInboundTemplateCandidate(applicationId: string, content: string) { + return this.prisma.smsTemplate.findFirst({ where: { applicationId, content, - auditStatus: 'approved', - signature: { - auditStatus: 'approved', - reportStatus: 'approved', - }, }, include: { signature: true }, orderBy: { updatedAt: 'desc' }, }); - if (!template) { - throw new BadRequestException('CMPP submit content does not match an approved template and signature'); - } - return template; + } + + private async recordCmppFailureReceipt( + message: { + id: string; + tenantId?: string | null; + batchTaskId?: string | null; + applicationId?: string | null; + messageId: string; + phoneNumber: string; + }, + errorCode: string, + reason: string, + ) { + if (!message.tenantId || !message.applicationId) return null; + const existing = await this.prisma.smsReceiptRecord.findFirst({ + where: { messageRecordId: message.id, gatewayMessageId: `PLATFORM:${message.messageId}` }, + }); + if (existing) return existing; + const deliveredAt = new Date(); + await this.prisma.smsMessageRecord.update({ + where: { id: message.id }, + data: { status: 'failed', receiptStatus: 'undelivered', errorCode, errorMessage: reason, deliveredAt }, + }); + const receipt = await this.prisma.smsReceiptRecord.create({ + data: { + tenantId: message.tenantId, + batchTaskId: message.batchTaskId, + messageRecordId: message.id, + messageId: message.messageId, + gatewayMessageId: `PLATFORM:${message.messageId}`, + receiptStatus: 'undelivered', + rawStatus: 'REJECTD', + errorCode, + deliveredAt, + }, + }); + await this.queueAndTryDownstreamDelivery({ + tenantId: message.tenantId, + applicationId: message.applicationId, + messageRecordId: message.id, + messageId: message.messageId, + deliveryType: 'receipt', + payload: { + messageId: message.messageId, + gatewayMessageId: `PLATFORM:${message.messageId}`, + phoneNumber: message.phoneNumber, + receiptStatus: 'undelivered', + rawStatus: 'REJECTD', + errorCode, + errorMessage: reason, + deliveredAt: deliveredAt.toISOString(), + }, + }); + if (message.batchTaskId) await this.refreshTaskProgress(message.batchTaskId); + return receipt; } private async validateSendResources(tenantId: string, applicationId?: string, templateId?: string) { diff --git a/docs/first-version-development-requirements.md b/docs/first-version-development-requirements.md index c80680d..cd0b2ab 100644 --- a/docs/first-version-development-requirements.md +++ b/docs/first-version-development-requirements.md @@ -210,14 +210,14 @@ #### 4.8.2 下游客户 CMPP 接入能力 1. Gateway 必须监听生产 CMPP 端口 `17890`，作为平台侧 CMPP Server 接收企业客户系统连接；该端口不是 HTTP 健康检查或控制接口。 -2. Gateway 必须按企业应用生成的 CMPP 接入参数校验客户 connect/login，包括客户侧企业代码、账号、密码、CMPP 版本、源 IP 白名单、短信接口开关、应用状态、企业状态、连接数上限；客户侧企业代码必须来自 `SmsApplication.cmppEnterpriseCode`，且 `interfaceEnabled=false` 时必须拒绝鉴权和后续 submit。Gateway 必须根据 CONNECT `Version` 为每条 TCP 连接独立协商 CMPP2.0/2.1/3.0 解包与响应类型，不得用固定 CMPP3.0 结构解析 CMPP2.0 Submit。 +2. Gateway 必须按企业应用生成的 CMPP 接入参数校验客户 connect/login，包括客户侧企业代码、账号、密码、CMPP 版本、源 IP 白名单、短信接口开关、应用状态、企业状态、连接数上限；客户侧企业代码必须来自 `SmsApplication.cmppEnterpriseCode`，且 `interfaceEnabled=false` 时必须拒绝新的鉴权。已完成 bind 的连接若随后被停用，其后续参数合法 Submit 必须按业务失败记录并通过 Deliver Receipt 回执。Gateway 必须根据 CONNECT `Version` 为每条 TCP 连接独立协商 CMPP2.0/2.1/3.0 解包与响应类型，不得用固定 CMPP3.0 结构解析 CMPP2.0 Submit。 3. 客户端应用的 IP 白名单必须对 CMPP 下游连接生效；未命中白名单、应用停用、企业停用、密码错误、超过连接数上限时必须拒绝连接并记录系统日志。 4. Gateway 必须维护应用级下游连接状态，回写 applicationId、tenantId、connectionId、currentConnections、desiredConnections、lastHeartbeatAt、lastError，运营端企业应用列表和连接详情必须来自这些真实状态。 5. Gateway 必须实现下游 ActiveTest、Terminate、异常断开处理；断开后连接数和状态必须及时回写。 6. Gateway 必须处理客户提交的 CMPP Submit，将手机号、内容、源地址、企业应用、客户消息序号等转换为平台发送请求。应用身份必须来自当前已鉴权 TCP 连接绑定的 `cmppAccount`；Submit `MsgSrc` 是独立的企业代码，必须与该应用 `cmppEnterpriseCode` 匹配，不得把 `MsgSrc` 当作登录账号查找应用。 -7. 下游 CMPP Submit 进入平台后，不创建客户端批量任务，但必须按手机号维度创建 `sms_message_record`，source 标记为 `cmpp`，并保留客户侧 sequence/msgId 映射。 +7. 下游 CMPP Submit 进入平台后，不创建客户可见的批量发送任务，但必须按手机号维度创建 `sms_message_record`，source 标记为 `cmpp`，并保留客户侧 sequence/msgId 映射。系统可使用内部批次承载计费、风控和队列，不得把该内部批次误展示为客户端手工批量任务。 8. 下游 CMPP Submit 必须复用 NestJS 发送前校验：企业/应用状态、IP 白名单、签名/模板报备、模板匹配策略、风控、黑名单、余额/授信、运营商识别、通道组路由。 -9. 对客户 Submit 的响应必须符合 CMPP 协议：参数错误、鉴权失败、余额不足、模板或签名未通过、无可用通道、风控拒绝等应映射为明确失败状态；已接收进入平台发送链路时返回成功并生成可追踪平台 messageId。 +9. 对客户 Submit 的响应必须符合 CMPP 协议：鉴权失败、账号或源 IP 不合法、PDU/手机号等参数不合法时直接返回非零 SubmitResp，且不创建短信记录。账号已识别且参数合法后必须先创建可追踪平台 messageId 和短信记录，再返回成功 SubmitResp；模板/签名/报备、风控、余额、应用后续停用、无可用通道以及上游最终提交失败等业务失败必须保存真实失败记录，并通过客户侧 Deliver Receipt 返回失败，不得因业务校验失败丢失客户 Submit 审计。 10. Gateway 必须支持平台最终回执向下游客户连接投递 Deliver Receipt；若客户连接已断开，应按策略缓存、重试或记录投递失败，不能丢失平台最终状态。 11. Gateway 必须支持下游客户上行接入场景：收到运营商上行后，按接入号、手机号、应用、时间窗口匹配并向客户连接推送 Deliver，上行同时入库。 12. 下游客户连接与上游通道连接必须隔离管理：客户侧账号密码不能用于连接上游通道，上游通道账号密码也不能作为客户接入凭据。 diff --git a/docs/system-functional-test-cases.md b/docs/system-functional-test-cases.md index 8cc4c4c..5951754 100644 --- a/docs/system-functional-test-cases.md +++ b/docs/system-functional-test-cases.md @@ -1009,7 +1009,8 @@ - 密码错误、应用停用、企业停用、短信接口关闭、IP 不在白名单时 connect/login 被拒绝。 - submit 被接受后返回 CMPP SubmitResp 成功，并在真实数据库创建 `sourceType=cmpp` 的发送记录，进入真实发送链路。 - Submit 应用身份使用 bind 已鉴权账号；`MsgSrc` 使用应用级企业代码并独立校验。企业代码与登录账号不同时仍能正确定位应用，企业代码不匹配时返回失败。 - - submit 内容不匹配审核模板、余额不足、短信接口关闭、无可用通道时返回明确失败，不得伪造成功。 + - 鉴权失败、IP 白名单不符、手机号等协议参数不合法时返回非零 SubmitResp，且不创建短信记录。 + - 已鉴权且参数合法的 Submit 必须先返回成功 SubmitResp 和平台 Msg_Id；内容不匹配审核模板、签名/报备未通过、余额不足、应用在 bind 后停用、无可用通道、上游 Submit 最终失败时，均须真实创建短信记录、`SmsReceiptRecord` 和 `CmppDownstreamDelivery`，并向客户下发 `undelivered/REJECTD` Deliver Receipt，不得仅以 SubmitResp 失败替代回执。 - Gateway 对每次 submit 记录 `submit_received` 和 `submit_accepted`/`submit_rejected`；日志可按账号、IP、sequenceId、号码和 messageId 定位，拒绝时包含 NestJS 真实业务原因和 CMPP result，但不包含明文短信正文。 - CMPP 包在进入 handler 前因长度、命令字、读包或 Unpack 失败时，Gateway 记录 `read/unpack packet failed`、远端地址、协议模式、错误类型和原始错误，不得静默断开。 diff --git a/docs/testing-progress.md b/docs/testing-progress.md index 237b49a..9a5f817 100644 --- a/docs/testing-progress.md +++ b/docs/testing-progress.md @@ -1558,3 +1558,11 @@ git diff --check - 已执行：`channels.service.spec.ts`（24 项通过）、`dictionaries.service.spec.ts`（4 项通过）、`operations.service.spec.ts`（12 项通过）、API build、前端 build 与 `git diff --check` 均通过；前端仍仅有既有 chunk size warning。 - 已部署生产验证：`cmpp-api`、`cmpp-gateway`、Nginx、MinIO 均为 active，API/Gateway health 均通过。真实认证 API 返回四类待审明细并与总数一致；手机号段第 2 页返回 25 条、总数 516217、`page=2/pageSize=25`，证明页面分页不再依赖 cursor 猜测总页数。 - 浏览器自动化在登录页连接阶段超时，未使用 CAPTCHA 绕过或修改生产数据；登录后页面视觉验收需在下一轮以人工登录或可用浏览器会话补充截图。其余项目以源码、真实 API 和构建结果验收，不能将该未完成的视觉截图记录成已完成。 + +## 2026-07-11 CMPP 业务失败回执闭环 + +- 修复下游 CMPP 入站的审计缺口：客户已完成 bind、账号可识别且手机号参数合法后，NestJS 会先创建真实 `SmsBatchTask`、`SmsApiRequest` 和 `SmsMessageRecord`，再执行模板、签名/报备、风控和余额校验；不再因模板未报备等业务失败而直接丢弃客户 Submit。 +- 协议、鉴权、源 IP 和手机号参数错误仍由 Gateway/NestJS 返回非零 SubmitResp，且不创建短信记录。其余业务失败返回成功 SubmitResp 与平台 Msg_Id，并创建真实 `SmsReceiptRecord(rawStatus=REJECTD)` 和 `CmppDownstreamDelivery`，客户通过 Deliver Receipt 获得 `undelivered` 结果。 +- 同一回执策略覆盖最终通道签名报备失败、无可用路由，以及上游 Submit rejected/timeout 在补发耗尽后的终态失败；失败记录、错误码和错误原因均可在运营端真实短信记录链路查询。 +- Gateway 在客户 Submit 成功并建立 messageId-连接映射后立即冲刷该账号 pending 下游投递，避免 API 先创建失败回执时只能等待周期补投。 +- 已执行 `npm --prefix api test -- --runInBand send-chain.service.spec.ts`（34 项通过）、`npm --prefix api run build`、`go test ./...`（Gateway 全量通过）。待本轮全量 API/前端构建及生产验证完成后补充最终部署结果。 diff --git a/gateway/internal/inbound/server.go b/gateway/internal/inbound/server.go index 4d11086..aae9b66 100644 --- a/gateway/internal/inbound/server.go +++ b/gateway/internal/inbound/server.go @@ -254,6 +254,11 @@ func (s Server) handleSubmit(response *cmpp.Response, packet *cmpp.Packet, logge presence: s.PresenceStore, instanceID: s.gatewayInstanceID(), }) + go func() { + if _, err := s.flushPending(account, logger); err != nil { + logger.Printf("cmpp inbound event=post_submit_pending_flush_failed account=%s message_id=%s error=%q", account, result.MessageID, err.Error()) + } + }() logger.Printf( "cmpp inbound event=submit_accepted protocol=%s packet_type=%s account=%s remote=%s seq=%d phone=%s result=0 message_id=%s gateway_message_id=%d duration_ms=%d content_chars=%d content_hash=%s", clientProtocol, req.protocol, account, remote, req.sequenceID, phone, result.MessageID, gatewayMsgID, time.Since(startedAt).Milliseconds(), len([]rune(content)), contentHash,