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

refactor: strengthen client boundaries and quality gates

Browse Source
This commit is contained in:
hectorzhao
2026-08-28 14:26:58 +08:00
parent 3af145abe5
commit ad27acad7e
51 changed files with 7703 additions and 697 deletions
Download Patch File Download Diff File Expand all files Collapse all files
api/src/open-api/client-open-api.controller.ts
+54 -10
View File
@@ -1,22 +1,66 @@
import { Body, Controller, Get, Param, Post, Put } from '@nestjs/common';
import { Body, Controller, Get, Param, ParseEnumPipe, Post, Put, UsePipes } from '@nestjs/common';
import { ApiTags } from '@nestjs/swagger';
import { RequireRecentAuthentication } from '../auth/require-recent-authentication.decorator';
import { CurrentTenantId } from '../auth/current-tenant-id.decorator';
import { CurrentSessionUserId } from '../auth/current-session-user.decorator';
import { OpenApiService } from './open-api.service';
import { strictValidationPipe } from '../common/strict-validation.pipe';
import { ClientHttpCredentialDto, ClientWebhookDto, ClientWebhookEventType } from './client-open-api.dto';
@ApiTags('client-http-open-api-management')
@Controller('client/applications/:applicationId/http-api')
export class ClientOpenApiController {
constructor(private readonly service: OpenApiService) {}
@Get() getConfig(@Param('applicationId') applicationId: string, @CurrentTenantId() tenantId: string) { return this.service.getConfig(applicationId, tenantId); }
@Get('credentials') listCredentials(@Param('applicationId') applicationId: string, @CurrentTenantId() tenantId: string) { return this.service.listCredentials(applicationId, tenantId); }
@Post('credentials') @RequireRecentAuthentication() createCredential(@Param('applicationId') applicationId: string, @Body() body: { name?: string; expiresAt?: string; createdById?: string }, @CurrentTenantId() tenantId: string, @CurrentSessionUserId() createdById?: string) { return this.service.createCredential(applicationId, { ...body, createdById }, tenantId, true); }
@Post('credentials/:credentialId/revoke') @RequireRecentAuthentication() revokeCredential(@Param('applicationId') applicationId: string, @Param('credentialId') credentialId: string, @CurrentTenantId() tenantId: string) { return this.service.revokeCredential(applicationId, credentialId, tenantId); }
@Get('webhooks') getWebhooks(@Param('applicationId') applicationId: string, @CurrentTenantId() tenantId: string) { return this.service.getWebhookEndpoints(applicationId, tenantId); }
@Put('webhooks/:eventType') @RequireRecentAuthentication() upsertWebhook(@Param('applicationId') applicationId: string, @Param('eventType') eventType: string, @Body() body: { url: string; rotateSecret?: boolean; status?: string }, @CurrentTenantId() tenantId: string) { return this.service.upsertWebhookEndpoint(applicationId, eventType, body, tenantId); }
@Get('requests') listRequests(@Param('applicationId') applicationId: string, @CurrentTenantId() tenantId: string) { return this.service.listRequestLogs(applicationId, tenantId); }
@Get('webhook-deliveries') listDeliveries(@Param('applicationId') applicationId: string, @CurrentTenantId() tenantId: string) { return this.service.listWebhookDeliveries(applicationId, tenantId); }
@Post('webhook-deliveries/:deliveryId/retry') @RequireRecentAuthentication() retryDelivery(@Param('applicationId') applicationId: string, @Param('deliveryId') deliveryId: string, @CurrentTenantId() tenantId: string) { return this.service.retryWebhookDelivery(applicationId, deliveryId, tenantId); }
@Get() getConfig(@Param('applicationId') applicationId: string, @CurrentTenantId() tenantId: string) {
return this.service.getConfig(applicationId, tenantId);
}
@Get('credentials') listCredentials(
@Param('applicationId') applicationId: string,
@CurrentTenantId() tenantId: string,
) {
return this.service.listCredentials(applicationId, tenantId);
}
@Post('credentials') @RequireRecentAuthentication() @UsePipes(strictValidationPipe) createCredential(
@Param('applicationId') applicationId: string,
@Body() body: ClientHttpCredentialDto,
@CurrentTenantId() tenantId: string,
@CurrentSessionUserId() createdById?: string,
) {
return this.service.createCredential(applicationId, { ...body, createdById }, tenantId, true);
}
@Post('credentials/:credentialId/revoke') @RequireRecentAuthentication() revokeCredential(
@Param('applicationId') applicationId: string,
@Param('credentialId') credentialId: string,
@CurrentTenantId() tenantId: string,
) {
return this.service.revokeCredential(applicationId, credentialId, tenantId);
}
@Get('webhooks') getWebhooks(@Param('applicationId') applicationId: string, @CurrentTenantId() tenantId: string) {
return this.service.getWebhookEndpoints(applicationId, tenantId);
}
@Put('webhooks/:eventType') @RequireRecentAuthentication() @UsePipes(strictValidationPipe) upsertWebhook(
@Param('applicationId') applicationId: string,
@Param('eventType', new ParseEnumPipe(ClientWebhookEventType)) eventType: ClientWebhookEventType,
@Body() body: ClientWebhookDto,
@CurrentTenantId() tenantId: string,
) {
return this.service.upsertWebhookEndpoint(applicationId, eventType, body, tenantId);
}
@Get('requests') listRequests(@Param('applicationId') applicationId: string, @CurrentTenantId() tenantId: string) {
return this.service.listRequestLogs(applicationId, tenantId);
}
@Get('webhook-deliveries') listDeliveries(
@Param('applicationId') applicationId: string,
@CurrentTenantId() tenantId: string,
) {
return this.service.listWebhookDeliveries(applicationId, tenantId);
}
@Post('webhook-deliveries/:deliveryId/retry') @RequireRecentAuthentication() retryDelivery(
@Param('applicationId') applicationId: string,
@Param('deliveryId') deliveryId: string,
@CurrentTenantId() tenantId: string,
) {
return this.service.retryWebhookDelivery(applicationId, deliveryId, tenantId);
}
}
api/src/open-api/client-open-api.dto.spec.ts
+28
View File
@@ -0,0 +1,28 @@
import { BadRequestException } from '@nestjs/common';
import { strictValidationPipe } from '../common/strict-validation.pipe';
import { ClientHttpCredentialDto, ClientWebhookDto } from './client-open-api.dto';
function validate<T>(metatype: new () => T, value: unknown) {
return strictValidationPipe.transform(value, { type: 'body', metatype, data: undefined });
}
describe('client HTTP API DTOs', () => {
it('rejects client-supplied operator identity and malformed expiry dates', async () => {
await expect(validate(ClientHttpCredentialDto, { name: '凭据', createdById: 'spoofed' })).rejects.toBeInstanceOf(
BadRequestException,
);
await expect(validate(ClientHttpCredentialDto, { expiresAt: 'tomorrow' })).rejects.toBeInstanceOf(
BadRequestException,
);
});
it('accepts a blank webhook URL for deletion and rejects unsafe fields', async () => {
await expect(validate(ClientWebhookDto, { url: ' ' })).resolves.toEqual(expect.objectContaining({ url: '' }));
await expect(validate(ClientWebhookDto, { url: 'javascript:alert(1)' })).rejects.toBeInstanceOf(
BadRequestException,
);
await expect(
validate(ClientWebhookDto, { url: 'https://example.com/hook', status: 'approved' }),
).rejects.toBeInstanceOf(BadRequestException);
});
});
api/src/open-api/client-open-api.dto.ts
+35
View File
@@ -0,0 +1,35 @@
import { Transform } from 'class-transformer';
import { IsBoolean, IsDateString, IsIn, IsOptional, IsString, IsUrl, MaxLength, ValidateIf } from 'class-validator';
export enum ClientWebhookEventType {
Receipt = 'receipt',
Uplink = 'uplink',
}
export class ClientHttpCredentialDto {
@IsOptional()
@IsString()
@MaxLength(100)
name?: string;
@IsOptional()
@IsDateString({ strict: true })
expiresAt?: string;
}
export class ClientWebhookDto {
@Transform(({ value }) => (typeof value === 'string' ? value.trim() : value))
@IsString()
@MaxLength(2048)
@ValidateIf(({ url }) => url !== '')
@IsUrl({ require_protocol: true, require_tld: false, protocols: ['http', 'https'] })
url!: string;
@IsOptional()
@IsBoolean()
rotateSecret?: boolean;
@IsOptional()
@IsIn(['active', 'inactive'])
status?: 'active' | 'inactive';
}
api/src/open-api/open-api.service.spec.ts
+175 -43
View File
@@ -3,7 +3,9 @@ import { decryptSecret, encryptSecret } from './open-api.crypto';
import { OpenApiService } from './open-api.service';
describe('OpenApiService', () => {
beforeAll(() => { process.env.HTTP_API_MASTER_KEY = 'test-master-key-with-at-least-32-characters'; });
beforeAll(() => {
process.env.HTTP_API_MASTER_KEY = 'test-master-key-with-at-least-32-characters';
});
it('encrypts secrets with authenticated encryption', () => {
const encrypted = encryptSecret('customer-secret');
@@ -15,11 +17,15 @@ describe('OpenApiService', () => {
const previous = process.env.HTTP_API_PUBLIC_ORIGIN;
process.env.HTTP_API_PUBLIC_ORIGIN = 'https://api.lisglo.com/';
const prisma = {
smsApplication: { findFirst: jest.fn().mockResolvedValue({ id: 'app-1', name: '应用A', httpConfig: null, httpIpAllowlist: [] }) },
smsApplication: {
findFirst: jest.fn().mockResolvedValue({ id: 'app-1', name: '应用A', httpConfig: null, httpIpAllowlist: [] }),
},
};
try {
const service = new OpenApiService(prisma as never, {} as never);
await expect(service.getConfig('app-1')).resolves.toEqual(expect.objectContaining({ publicOrigin: 'https://api.lisglo.com' }));
await expect(service.getConfig('app-1')).resolves.toEqual(
expect.objectContaining({ publicOrigin: 'https://api.lisglo.com' }),
);
} finally {
if (previous === undefined) delete process.env.HTTP_API_PUBLIC_ORIGIN;
else process.env.HTTP_API_PUBLIC_ORIGIN = previous;
@@ -32,13 +38,17 @@ describe('OpenApiService', () => {
process.env.HTTP_API_PUBLIC_ORIGIN = 'http://100.93.204.60:12026/';
delete process.env.HTTP_API_ALLOW_INSECURE_ORIGIN;
const prisma = {
smsApplication: { findFirst: jest.fn().mockResolvedValue({ id: 'app-1', name: '应用A', httpConfig: null, httpIpAllowlist: [] }) },
smsApplication: {
findFirst: jest.fn().mockResolvedValue({ id: 'app-1', name: '应用A', httpConfig: null, httpIpAllowlist: [] }),
},
};
try {
const service = new OpenApiService(prisma as never, {} as never);
await expect(service.getConfig('app-1')).rejects.toThrow('HTTP_API_ALLOW_INSECURE_ORIGIN');
process.env.HTTP_API_ALLOW_INSECURE_ORIGIN = 'true';
await expect(service.getConfig('app-1')).resolves.toEqual(expect.objectContaining({ publicOrigin: 'http://100.93.204.60:12026' }));
await expect(service.getConfig('app-1')).resolves.toEqual(
expect.objectContaining({ publicOrigin: 'http://100.93.204.60:12026' }),
);
} finally {
if (previousOrigin === undefined) delete process.env.HTTP_API_PUBLIC_ORIGIN;
else process.env.HTTP_API_PUBLIC_ORIGIN = previousOrigin;
@@ -49,25 +59,62 @@ describe('OpenApiService', () => {
it('replays a completed request for the same idempotency key and body', async () => {
const prisma = {
openApiRequest: { findUnique: jest.fn().mockResolvedValue({ bodyHash: 'same', status: 'completed', responseBody: { code: 'ACCEPTED', messageId: 'MSG-1' } }) },
openApiRequest: {
findUnique: jest
.fn()
.mockResolvedValue({
bodyHash: 'same',
status: 'completed',
responseBody: { code: 'ACCEPTED', messageId: 'MSG-1' },
}),
},
};
const sendChain = { createHttpBatchTask: jest.fn() };
const service = new OpenApiService(prisma as never, sendChain as never);
const result = await service.sendMessage(auth() as never, { mobile: '18821203795', content: '【测试】短信' }, { idempotencyKey: 'idem-0001', bodyHash: 'same' });
const result = await service.sendMessage(
auth() as never,
{ mobile: '18821203795', content: '【测试】短信' },
{ idempotencyKey: 'idem-0001', bodyHash: 'same' },
);
expect(result).toEqual({ code: 'ACCEPTED', messageId: 'MSG-1' });
expect(sendChain.createHttpBatchTask).not.toHaveBeenCalled();
});
it('rejects reuse of an idempotency key with a different body', async () => {
const prisma = { openApiRequest: { findUnique: jest.fn().mockResolvedValue({ bodyHash: 'old', status: 'completed' }) } };
const prisma = {
openApiRequest: { findUnique: jest.fn().mockResolvedValue({ bodyHash: 'old', status: 'completed' }) },
};
const service = new OpenApiService(prisma as never, { createHttpBatchTask: jest.fn() } as never);
await expect(service.sendMessage(auth() as never, { mobile: '18821203795', content: '【测试】短信' }, { idempotencyKey: 'idem-0001', bodyHash: 'new' })).rejects.toBeInstanceOf(ConflictException);
await expect(
service.sendMessage(
auth() as never,
{ mobile: '18821203795', content: '【测试】短信' },
{ idempotencyKey: 'idem-0001', bodyHash: 'new' },
),
).rejects.toBeInstanceOf(ConflictException);
});
it('replays the same persisted business rejection', async () => {
const prisma = { openApiRequest: { findUnique: jest.fn().mockResolvedValue({ bodyHash: 'same', status: 'failed', httpStatus: 422, responseBody: { code: 'SEND_REJECTED', message: '模板不匹配' } }) } };
const prisma = {
openApiRequest: {
findUnique: jest
.fn()
.mockResolvedValue({
bodyHash: 'same',
status: 'failed',
httpStatus: 422,
responseBody: { code: 'SEND_REJECTED', message: '模板不匹配' },
}),
},
};
const service = new OpenApiService(prisma as never, { createHttpBatchTask: jest.fn() } as never);
await expect(service.sendMessage(auth() as never, { mobile: '18821203795', content: '【测试】短信' }, { idempotencyKey: 'idem-0001', bodyHash: 'same' })).rejects.toMatchObject({ status: 422 });
await expect(
service.sendMessage(
auth() as never,
{ mobile: '18821203795', content: '【测试】短信' },
{ idempotencyKey: 'idem-0001', bodyHash: 'same' },
),
).rejects.toMatchObject({ status: 422 });
});
it('uses the real send chain and persists the accepted response', async () => {
@@ -79,47 +126,107 @@ describe('OpenApiService', () => {
},
smsMessageRecord: { findFirst: jest.fn().mockResolvedValue(null) },
};
const sendChain = { createHttpBatchTask: jest.fn().mockResolvedValue({ status: 'ready', messages: [{ id: 'row-1', messageId: 'MSG-1', status: 'queued' }] }) };
const sendChain = {
createHttpBatchTask: jest
.fn()
.mockResolvedValue({ status: 'ready', messages: [{ id: 'row-1', messageId: 'MSG-1', status: 'queued' }] }),
};
const service = new OpenApiService(prisma as never, sendChain as never);
const result = await service.sendMessage(auth() as never, { mobile: '18821203795', content: '【测试】短信', clientMessageId: 'client-1' }, { idempotencyKey: 'idem-0001', bodyHash: 'hash' });
expect(sendChain.createHttpBatchTask).toHaveBeenCalledWith(expect.objectContaining({ phones: ['18821203795'], clientMessageId: 'client-1' }));
expect(result).toEqual(expect.objectContaining({ code: 'ACCEPTED', messageId: 'MSG-1', clientMessageId: 'client-1' }));
expect(prisma.openApiRequest.update).toHaveBeenCalledWith(expect.objectContaining({ data: expect.objectContaining({ status: 'completed', httpStatus: 202, messageRecordId: 'row-1' }) }));
const result = await service.sendMessage(
auth() as never,
{ mobile: '18821203795', content: '【测试】短信', clientMessageId: 'client-1' },
{ idempotencyKey: 'idem-0001', bodyHash: 'hash' },
);
expect(sendChain.createHttpBatchTask).toHaveBeenCalledWith(
expect.objectContaining({ phones: ['18821203795'], clientMessageId: 'client-1' }),
);
expect(result).toEqual(
expect.objectContaining({ code: 'ACCEPTED', messageId: 'MSG-1', clientMessageId: 'client-1' }),
);
expect(prisma.openApiRequest.update).toHaveBeenCalledWith(
expect.objectContaining({
data: expect.objectContaining({ status: 'completed', httpStatus: 202, messageRecordId: 'row-1' }),
}),
);
});
it('persists a 422 result when the real send chain rejects the business request', async () => {
const prisma = {
openApiRequest: { findUnique: jest.fn().mockResolvedValue(null), create: jest.fn().mockResolvedValue({ id: 'request-row-1' }), update: jest.fn().mockResolvedValue({}) },
openApiRequest: {
findUnique: jest.fn().mockResolvedValue(null),
create: jest.fn().mockResolvedValue({ id: 'request-row-1' }),
update: jest.fn().mockResolvedValue({}),
},
smsMessageRecord: { findFirst: jest.fn().mockResolvedValue(null) },
};
const service = new OpenApiService(prisma as never, { createHttpBatchTask: jest.fn().mockRejectedValue(new BadRequestException('短信未匹配模板')) } as never);
await expect(service.sendMessage(auth() as never, { mobile: '18821203795', content: '未匹配模板' }, { idempotencyKey: 'idem-0001', bodyHash: 'hash' })).rejects.toMatchObject({ status: 422 });
expect(prisma.openApiRequest.update).toHaveBeenCalledWith(expect.objectContaining({ data: expect.objectContaining({ status: 'failed', httpStatus: 422, businessCode: 'SEND_REJECTED' }) }));
const service = new OpenApiService(
prisma as never,
{ createHttpBatchTask: jest.fn().mockRejectedValue(new BadRequestException('短信未匹配模板')) } as never,
);
await expect(
service.sendMessage(
auth() as never,
{ mobile: '18821203795', content: '未匹配模板' },
{ idempotencyKey: 'idem-0001', bodyHash: 'hash' },
),
).rejects.toMatchObject({ status: 422 });
expect(prisma.openApiRequest.update).toHaveBeenCalledWith(
expect.objectContaining({
data: expect.objectContaining({ status: 'failed', httpStatus: 422, businessCode: 'SEND_REJECTED' }),
}),
);
});
it('creates an HTTP webhook event when HTTP and the event capability are enabled', async () => {
const prisma = {
smsApplication: { findUnique: jest.fn().mockResolvedValue({ httpConfig: { enabled: true, receiptWebhookEnabled: true, receiptDeliveryMode: 'http' } }) },
smsApplication: {
findUnique: jest
.fn()
.mockResolvedValue({
httpConfig: { enabled: true, receiptWebhookEnabled: true, receiptDeliveryMode: 'http' },
}),
},
httpWebhookEndpoint: { findUnique: jest.fn().mockResolvedValue({ id: 'endpoint-1', status: 'active' }) },
httpWebhookEvent: { upsert: jest.fn().mockResolvedValue({ id: 'event-row-1' }) },
httpWebhookDelivery: { upsert: jest.fn().mockResolvedValue({ id: 'delivery-1', status: 'pending' }) },
};
const service = new OpenApiService(prisma as never, {} as never);
const input = { tenantId: 'tenant-1', applicationId: 'app-1', eventType: 'receipt' as const, messageRecordId: 'record-1', messageId: 'MSG-1', payload: { receiptStatus: 'delivered' } };
const input = {
tenantId: 'tenant-1',
applicationId: 'app-1',
eventType: 'receipt' as const,
messageRecordId: 'record-1',
messageId: 'MSG-1',
payload: { receiptStatus: 'delivered' },
};
await service.queueWebhookEvent(input);
await service.queueWebhookEvent(input);
expect(prisma.httpWebhookEvent.upsert).toHaveBeenCalledWith(expect.objectContaining({
where: { eventId: 'evt_receipt_record-1' },
}));
expect(prisma.httpWebhookEvent.upsert).toHaveBeenCalledWith(
expect.objectContaining({
where: { eventId: 'evt_receipt_record-1' },
}),
);
expect(prisma.httpWebhookEvent.upsert).toHaveBeenCalledTimes(2);
expect(prisma.httpWebhookDelivery.upsert).toHaveBeenCalledWith(expect.objectContaining({
create: { eventId: 'event-row-1', endpointId: 'endpoint-1' },
}));
expect(prisma.httpWebhookDelivery.upsert).toHaveBeenCalledWith(
expect.objectContaining({
create: { eventId: 'event-row-1', endpointId: 'endpoint-1' },
}),
);
});
it('defaults a newly enabled HTTP interface to all six capabilities and automatic dual delivery', async () => {
const prisma = {
smsApplication: { findFirst: jest.fn().mockResolvedValue({ id: 'app-1', name: '应用A', interfaceEnabled: true, httpConfig: null, httpIpAllowlist: [] }) },
smsApplication: {
findFirst: jest
.fn()
.mockResolvedValue({
id: 'app-1',
name: '应用A',
interfaceEnabled: true,
httpConfig: null,
httpIpAllowlist: [],
}),
},
smsApplicationHttpConfig: { upsert: jest.fn().mockImplementation(({ create }) => Promise.resolve(create)) },
smsApplicationHttpIpAllowlist: { deleteMany: jest.fn().mockResolvedValue({ count: 0 }), createMany: jest.fn() },
$transaction: jest.fn((operations) => Promise.all(operations)),
@@ -128,19 +235,21 @@ describe('OpenApiService', () => {
await service.updateConfig('app-1', { enabled: true });
expect(prisma.smsApplicationHttpConfig.upsert).toHaveBeenCalledWith(expect.objectContaining({
create: expect.objectContaining({
enabled: true,
sendEnabled: true,
messageQueryEnabled: true,
receiptWebhookEnabled: true,
uplinkWebhookEnabled: true,
uplinkQueryEnabled: true,
credentialSelfServiceEnabled: true,
receiptDeliveryMode: 'both',
uplinkDeliveryMode: 'both',
expect(prisma.smsApplicationHttpConfig.upsert).toHaveBeenCalledWith(
expect.objectContaining({
create: expect.objectContaining({
enabled: true,
sendEnabled: true,
messageQueryEnabled: true,
receiptWebhookEnabled: true,
uplinkWebhookEnabled: true,
uplinkQueryEnabled: true,
credentialSelfServiceEnabled: true,
receiptDeliveryMode: 'both',
uplinkDeliveryMode: 'both',
}),
}),
}));
);
});
it('removes a webhook endpoint when an operator saves a blank address', async () => {
@@ -160,12 +269,35 @@ describe('OpenApiService', () => {
};
const service = new OpenApiService(prisma as never, {} as never);
await expect(service.upsertWebhookEndpoint('app-1', 'receipt', { url: ' ' }))
.resolves.toEqual(expect.objectContaining({ eventType: 'receipt', url: '', status: 'inactive', deleted: true }));
await expect(service.upsertWebhookEndpoint('app-1', 'receipt', { url: ' ' })).resolves.toEqual(
expect.objectContaining({ eventType: 'receipt', url: '', status: 'inactive', deleted: true }),
);
expect(prisma.httpWebhookEndpoint.deleteMany).toHaveBeenCalledWith({
where: { applicationId: 'app-1', eventType: 'receipt' },
});
});
it('rejects an already expired credential before writing a secret', async () => {
const prisma = {
smsApplication: {
findFirst: jest
.fn()
.mockResolvedValue({
id: 'app-1',
name: '应用A',
interfaceEnabled: true,
httpConfig: { enabled: true, credentialSelfServiceEnabled: true, maxCredentialCount: 3 },
httpIpAllowlist: [],
}),
},
httpApiCredential: { count: jest.fn().mockResolvedValue(0), create: jest.fn() },
};
const service = new OpenApiService(prisma as never, {} as never);
await expect(
service.createCredential('app-1', { expiresAt: '2020-01-01T00:00:00.000Z' }, 'tenant-1', true),
).rejects.toThrow('凭据过期时间必须晚于当前时间');
expect(prisma.httpApiCredential.create).not.toHaveBeenCalled();
});
});
function auth() {
api/src/open-api/open-api.service.ts
+538 -130
View File
@@ -1,4 +1,17 @@
import { BadRequestException, ConflictException, ForbiddenException, forwardRef, HttpException, Inject, Injectable, NotFoundException, OnModuleDestroy, OnModuleInit, Optional, UnprocessableEntityException } from '@nestjs/common';
import {
BadRequestException,
ConflictException,
ForbiddenException,
forwardRef,
HttpException,
Inject,
Injectable,
NotFoundException,
OnModuleDestroy,
OnModuleInit,
Optional,
UnprocessableEntityException,
} from '@nestjs/common';
import { Prisma } from '@prisma/client';
import { Queue, Worker } from 'bullmq';
import { createHash, createHmac, randomBytes, randomUUID } from 'node:crypto';
@@ -59,7 +72,10 @@ export class OpenApiService implements OnModuleInit, OnModuleDestroy {
// 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 });
this.worker = new Worker(WEBHOOK_QUEUE, (job) => this.deliverWebhook(job.data.deliveryId), {
connection,
concurrency: 10,
});
}
async onModuleDestroy() {
@@ -89,7 +105,13 @@ export class OpenApiService implements OnModuleInit, OnModuleDestroy {
update: data,
}),
this.prisma.smsApplicationHttpIpAllowlist.deleteMany({ where: { applicationId } }),
...(ipAllowlist.length > 0 ? [this.prisma.smsApplicationHttpIpAllowlist.createMany({ data: ipAllowlist.map((ipCidr) => ({ applicationId, ipCidr })) })] : []),
...(ipAllowlist.length > 0
? [
this.prisma.smsApplicationHttpIpAllowlist.createMany({
data: ipAllowlist.map((ipCidr) => ({ applicationId, ipCidr })),
}),
]
: []),
]);
return { applicationId, publicOrigin: httpApiPublicOrigin(), config, ipAllowlist };
}
@@ -98,37 +120,69 @@ export class OpenApiService implements OnModuleInit, OnModuleDestroy {
await this.requireApplication(applicationId, tenantId);
return this.prisma.httpApiCredential.findMany({
where: { applicationId },
select: { id: true, name: true, accessKey: true, secretLast4: true, status: true, expiresAt: true, lastUsedAt: true, lastUsedIp: true, createdAt: true, revokedAt: true },
select: {
id: true,
name: true,
accessKey: true,
secretLast4: true,
status: true,
expiresAt: true,
lastUsedAt: true,
lastUsedIp: true,
createdAt: true,
revokedAt: true,
},
orderBy: { createdAt: 'desc' },
});
}
async createCredential(applicationId: string, data: { name?: string; expiresAt?: string; createdById?: string }, tenantId?: string, selfService = false) {
async createCredential(
applicationId: string,
data: { name?: string; expiresAt?: string; createdById?: string },
tenantId?: string,
selfService = false,
) {
const application = await this.requireApplication(applicationId, tenantId);
const config = application.httpConfig;
if (!config?.enabled) throw new BadRequestException('请先开通该应用的HTTP接口');
if (selfService && !config.credentialSelfServiceEnabled) throw new ForbiddenException('该应用未开通客户端凭据自助管理');
if (selfService && !config.credentialSelfServiceEnabled)
throw new ForbiddenException('该应用未开通客户端凭据自助管理');
const activeCount = await this.prisma.httpApiCredential.count({ where: { applicationId, status: 'active' } });
if (activeCount >= config.maxCredentialCount) throw new BadRequestException(`有效凭据最多允许 ${config.maxCredentialCount} 个`);
if (activeCount >= config.maxCredentialCount)
throw new BadRequestException(`有效凭据最多允许 ${config.maxCredentialCount} 个`);
const expiresAt = data.expiresAt ? new Date(data.expiresAt) : undefined;
if (expiresAt && expiresAt.getTime() <= Date.now()) throw new BadRequestException('凭据过期时间必须晚于当前时间');
const secret = randomBytes(32).toString('base64url');
const credential = await this.prisma.httpApiCredential.create({
data: {
applicationId,
name: String(data.name ?? '默认凭据').trim().slice(0, 100) || '默认凭据',
name:
String(data.name ?? '默认凭据')
.trim()
.slice(0, 100) || '默认凭据',
accessKey: `ak_${randomBytes(18).toString('base64url')}`,
secretEncrypted: encryptSecret(secret),
secretLast4: secret.slice(-4),
expiresAt: data.expiresAt ? new Date(data.expiresAt) : undefined,
expiresAt,
createdById: data.createdById,
},
select: { id: true, name: true, accessKey: true, secretLast4: true, status: true, expiresAt: true, createdAt: true },
select: {
id: true,
name: true,
accessKey: true,
secretLast4: true,
status: true,
expiresAt: true,
createdAt: true,
},
});
return { ...credential, secret, secretShownOnce: true };
}
async revokeCredential(applicationId: string, credentialId: string, tenantId?: string) {
const application = await this.requireApplication(applicationId, tenantId);
if (tenantId && !application.httpConfig?.credentialSelfServiceEnabled) throw new ForbiddenException('该应用未开通客户端凭据自助管理');
if (tenantId && !application.httpConfig?.credentialSelfServiceEnabled)
throw new ForbiddenException('该应用未开通客户端凭据自助管理');
const result = await this.prisma.httpApiCredential.updateMany({
where: { id: credentialId, applicationId, status: 'active' },
data: { status: 'revoked', revokedAt: new Date() },
@@ -141,14 +195,29 @@ export class OpenApiService implements OnModuleInit, OnModuleDestroy {
await this.requireApplication(applicationId, tenantId);
return this.prisma.httpWebhookEndpoint.findMany({
where: { applicationId },
select: { id: true, eventType: true, url: true, secretLast4: true, status: true, lastTestAt: true, lastTestStatus: true, updatedAt: true },
select: {
id: true,
eventType: true,
url: true,
secretLast4: true,
status: true,
lastTestAt: true,
lastTestStatus: true,
updatedAt: true,
},
orderBy: { eventType: 'asc' },
});
}
async upsertWebhookEndpoint(applicationId: string, eventType: string, data: { url: string; rotateSecret?: boolean; status?: string }, tenantId?: string) {
async upsertWebhookEndpoint(
applicationId: string,
eventType: string,
data: { url: string; rotateSecret?: boolean; status?: string },
tenantId?: string,
) {
const application = await this.requireApplication(applicationId, tenantId);
if (!['receipt', 'uplink'].includes(eventType)) throw new BadRequestException('eventType only supports receipt or uplink');
if (!['receipt', 'uplink'].includes(eventType))
throw new BadRequestException('eventType only supports receipt or uplink');
if (!String(data.url ?? '').trim()) {
await this.prisma.httpWebhookEndpoint.deleteMany({ where: { applicationId, eventType } });
return {
@@ -162,49 +231,103 @@ export class OpenApiService implements OnModuleInit, OnModuleDestroy {
};
}
const url = await validateWebhookUrl(data.url, application.httpConfig?.requireHttps ?? true);
const existing = await this.prisma.httpWebhookEndpoint.findUnique({ where: { applicationId_eventType: { applicationId, eventType } } });
const existing = await this.prisma.httpWebhookEndpoint.findUnique({
where: { applicationId_eventType: { applicationId, eventType } },
});
const secret = !existing || data.rotateSecret ? randomBytes(32).toString('base64url') : undefined;
const endpoint = await this.prisma.httpWebhookEndpoint.upsert({
where: { applicationId_eventType: { applicationId, eventType } },
create: { applicationId, eventType, url, status: data.status ?? 'active', secretEncrypted: encryptSecret(secret!), secretLast4: secret!.slice(-4) },
update: { url, status: data.status ?? existing?.status ?? 'active', ...(secret ? { secretEncrypted: encryptSecret(secret), secretLast4: secret.slice(-4) } : {}) },
create: {
applicationId,
eventType,
url,
status: data.status ?? 'active',
secretEncrypted: encryptSecret(secret!),
secretLast4: secret!.slice(-4),
},
update: {
url,
status: data.status ?? existing?.status ?? 'active',
...(secret ? { secretEncrypted: encryptSecret(secret), secretLast4: secret.slice(-4) } : {}),
},
select: { id: true, eventType: true, url: true, secretLast4: true, status: true, updatedAt: true },
});
return { ...endpoint, ...(secret ? { secret, secretShownOnce: true } : {}) };
}
async sendMessage(auth: OpenApiAuthContext, input: { mobile?: string; content?: string; clientMessageId?: string }, meta: { idempotencyKey?: string; bodyHash: string; userAgent?: string }) {
if (!auth.config.sendEnabled) throw new ForbiddenException({ code: 'SEND_NOT_ENABLED', message: '该应用未开通HTTP短信发送' });
async sendMessage(
auth: OpenApiAuthContext,
input: { mobile?: string; content?: string; clientMessageId?: string },
meta: { idempotencyKey?: string; bodyHash: string; userAgent?: string },
) {
if (!auth.config.sendEnabled)
throw new ForbiddenException({ code: 'SEND_NOT_ENABLED', message: '该应用未开通HTTP短信发送' });
const mobile = String(input.mobile ?? '').trim();
const content = String(input.content ?? '');
if (!/^1[3-9]\d{9}$/.test(mobile)) throw new BadRequestException({ code: 'MOBILE_INVALID', message: '手机号格式非法' });
if (!/^1[3-9]\d{9}$/.test(mobile))
throw new BadRequestException({ code: 'MOBILE_INVALID', message: '手机号格式非法' });
if (!content.trim()) throw new BadRequestException({ code: 'CONTENT_REQUIRED', message: '短信内容不能为空' });
const idempotencyKey = String(meta.idempotencyKey ?? '').trim();
if (!/^[A-Za-z0-9._:-]{8,128}$/.test(idempotencyKey)) throw new BadRequestException({ code: 'IDEMPOTENCY_KEY_INVALID', message: 'Idempotency-Key 必填且长度为8至128位' });
const existing = await this.prisma.openApiRequest.findUnique({ where: { applicationId_idempotencyKey: { applicationId: auth.application.id, idempotencyKey } } });
if (!/^[A-Za-z0-9._:-]{8,128}$/.test(idempotencyKey))
throw new BadRequestException({
code: 'IDEMPOTENCY_KEY_INVALID',
message: 'Idempotency-Key 必填且长度为8至128位',
});
const existing = await this.prisma.openApiRequest.findUnique({
where: { applicationId_idempotencyKey: { applicationId: auth.application.id, idempotencyKey } },
});
if (existing) {
if (existing.bodyHash !== meta.bodyHash) throw new ConflictException({ code: 'IDEMPOTENCY_CONFLICT', message: '同一Idempotency-Key对应的请求内容不一致' });
if (existing.bodyHash !== meta.bodyHash)
throw new ConflictException({
code: 'IDEMPOTENCY_CONFLICT',
message: '同一Idempotency-Key对应的请求内容不一致',
});
if (existing.status === 'completed' && existing.responseBody) return existing.responseBody;
if (existing.status === 'failed' && existing.responseBody && existing.httpStatus) throw new HttpException(existing.responseBody as Record<string, unknown>, existing.httpStatus);
if (existing.status === 'failed' && existing.responseBody && existing.httpStatus)
throw new HttpException(existing.responseBody as Record<string, unknown>, existing.httpStatus);
throw new ConflictException({ code: 'REQUEST_PROCESSING', message: '同一请求正在处理中，请稍后查询' });
}
if (input.clientMessageId) {
const duplicateClientMessage = await this.prisma.smsMessageRecord.findFirst({ where: { applicationId: auth.application.id, clientMessageId: input.clientMessageId }, select: { messageId: true } });
if (duplicateClientMessage) throw new ConflictException({ code: 'CLIENT_MESSAGE_ID_CONFLICT', message: `clientMessageId已关联短信 ${duplicateClientMessage.messageId}` });
const duplicateClientMessage = await this.prisma.smsMessageRecord.findFirst({
where: { applicationId: auth.application.id, clientMessageId: input.clientMessageId },
select: { messageId: true },
});
if (duplicateClientMessage)
throw new ConflictException({
code: 'CLIENT_MESSAGE_ID_CONFLICT',
message: `clientMessageId已关联短信 ${duplicateClientMessage.messageId}`,
});
}
const requestId = `req_${randomUUID()}`;
const startedAt = Date.now();
let request;
try {
request = await this.prisma.openApiRequest.create({
data: { tenantId: auth.application.tenantId, applicationId: auth.application.id, credentialId: auth.credentialId, requestId, idempotencyKey, bodyHash: meta.bodyHash, clientMessageId: input.clientMessageId, sourceIp: auth.sourceIp, userAgent: meta.userAgent },
data: {
tenantId: auth.application.tenantId,
applicationId: auth.application.id,
credentialId: auth.credentialId,
requestId,
idempotencyKey,
bodyHash: meta.bodyHash,
clientMessageId: input.clientMessageId,
sourceIp: auth.sourceIp,
userAgent: meta.userAgent,
},
});
} catch (error) {
if ((error as { code?: string }).code === 'P2002') {
const raced = await this.prisma.openApiRequest.findUnique({ where: { applicationId_idempotencyKey: { applicationId: auth.application.id, idempotencyKey } } });
if (raced?.bodyHash !== meta.bodyHash) throw new ConflictException({ code: 'IDEMPOTENCY_CONFLICT', message: '同一Idempotency-Key对应的请求内容不一致' });
const raced = await this.prisma.openApiRequest.findUnique({
where: { applicationId_idempotencyKey: { applicationId: auth.application.id, idempotencyKey } },
});
if (raced?.bodyHash !== meta.bodyHash)
throw new ConflictException({
code: 'IDEMPOTENCY_CONFLICT',
message: '同一Idempotency-Key对应的请求内容不一致',
});
if (raced?.status === 'completed' && raced.responseBody) return raced.responseBody;
if (raced?.status === 'failed' && raced.responseBody && raced.httpStatus) throw new HttpException(raced.responseBody as Record<string, unknown>, raced.httpStatus);
if (raced?.status === 'failed' && raced.responseBody && raced.httpStatus)
throw new HttpException(raced.responseBody as Record<string, unknown>, raced.httpStatus);
throw new ConflictException({ code: 'REQUEST_PROCESSING', message: '同一请求正在处理中，请稍后查询' });
}
throw error;
@@ -221,10 +344,31 @@ export class OpenApiService implements OnModuleInit, OnModuleDestroy {
});
const message = task.messages?.[0];
if (task.status === 'rejected' || message?.status === 'rejected') {
throw new UnprocessableEntityException({ code: 'SEND_REJECTED', message: message?.errorMessage ?? task.rejectReason ?? '短信未通过业务校验' });
throw new UnprocessableEntityException({
code: 'SEND_REJECTED',
message: message?.errorMessage ?? task.rejectReason ?? '短信未通过业务校验',
});
}
const response = { code: 'ACCEPTED', requestId, messageId: message?.messageId, clientMessageId: input.clientMessageId ?? null, status: message?.status ?? task.status, acceptedAt: new Date().toISOString() };
await this.prisma.openApiRequest.update({ where: { id: request.id }, data: { status: 'completed', httpStatus: 202, businessCode: 'ACCEPTED', responseBody: response, messageRecordId: message?.id, durationMs: Date.now() - startedAt, completedAt: new Date() } });
const response = {
code: 'ACCEPTED',
requestId,
messageId: message?.messageId,
clientMessageId: input.clientMessageId ?? null,
status: message?.status ?? task.status,
acceptedAt: new Date().toISOString(),
};
await this.prisma.openApiRequest.update({
where: { id: request.id },
data: {
status: 'completed',
httpStatus: 202,
businessCode: 'ACCEPTED',
responseBody: response,
messageRecordId: message?.id,
durationMs: Date.now() - startedAt,
completedAt: new Date(),
},
});
this.protocolLogs?.record({
protocol: 'http',
direction: 'client_to_platform',
@@ -245,11 +389,24 @@ export class OpenApiService implements OnModuleInit, OnModuleDestroy {
let outwardError = error;
if (error instanceof HttpException && error.getStatus() === 400) {
const response = error.getResponse();
const message = typeof response === 'object' && response && 'message' in response ? (response as { message: unknown }).message : error.message;
const message =
typeof response === 'object' && response && 'message' in response
? (response as { message: unknown }).message
: error.message;
outwardError = new UnprocessableEntityException({ code: 'SEND_REJECTED', message });
}
const failure = normalizeOpenApiFailure(outwardError);
await this.prisma.openApiRequest.update({ where: { id: request.id }, data: { status: 'failed', httpStatus: failure.httpStatus, businessCode: failure.code, responseBody: failure.responseBody, durationMs: Date.now() - startedAt, completedAt: new Date() } });
await this.prisma.openApiRequest.update({
where: { id: request.id },
data: {
status: 'failed',
httpStatus: failure.httpStatus,
businessCode: failure.code,
responseBody: failure.responseBody,
durationMs: Date.now() - startedAt,
completedAt: new Date(),
},
});
this.protocolLogs?.record({
protocol: 'http',
direction: 'client_to_platform',
@@ -268,21 +425,41 @@ export class OpenApiService implements OnModuleInit, OnModuleDestroy {
}
async getMessage(auth: OpenApiAuthContext, messageId: string) {
if (!auth.config.messageQueryEnabled) throw new ForbiddenException({ code: 'MESSAGE_QUERY_NOT_ENABLED', message: '该应用未开通短信状态查询' });
if (!auth.config.messageQueryEnabled)
throw new ForbiddenException({ code: 'MESSAGE_QUERY_NOT_ENABLED', message: '该应用未开通短信状态查询' });
const message = await this.prisma.smsMessageRecord.findFirst({
where: { applicationId: auth.application.id, OR: [{ messageId }, { clientMessageId: messageId }] },
select: { messageId: true, clientMessageId: true, phoneNumber: true, status: true, submitStatus: true, receiptStatus: true, errorCode: true, errorMessage: true, queuedAt: true, submittedAt: true, deliveredAt: true, updatedAt: true },
select: {
messageId: true,
clientMessageId: true,
phoneNumber: true,
status: true,
submitStatus: true,
receiptStatus: true,
errorCode: true,
errorMessage: true,
queuedAt: true,
submittedAt: true,
deliveredAt: true,
updatedAt: true,
},
});
if (!message) throw new NotFoundException({ code: 'MESSAGE_NOT_FOUND', message: '短信记录不存在' });
return message;
}
async listUplinks(auth: OpenApiAuthContext, query: Record<string, string | undefined>) {
if (!auth.config.uplinkQueryEnabled) throw new ForbiddenException({ code: 'UPLINK_QUERY_NOT_ENABLED', message: '该应用未开通上行查询' });
if (!auth.config.uplinkQueryEnabled)
throw new ForbiddenException({ code: 'UPLINK_QUERY_NOT_ENABLED', message: '该应用未开通上行查询' });
const endTime = query.endTime ? new Date(query.endTime) : new Date();
const startTime = query.startTime ? new Date(query.startTime) : new Date(endTime.getTime() - 24 * 3600_000);
if (!Number.isFinite(startTime.getTime()) || !Number.isFinite(endTime.getTime()) || startTime > endTime) throw new BadRequestException({ code: 'TIME_RANGE_INVALID', message: '查询时间范围非法' });
if (endTime.getTime() - startTime.getTime() > auth.config.maxQueryRangeDays * 86400_000) throw new BadRequestException({ code: 'TIME_RANGE_TOO_LARGE', message: `单次查询不能超过${auth.config.maxQueryRangeDays}天` });
if (!Number.isFinite(startTime.getTime()) || !Number.isFinite(endTime.getTime()) || startTime > endTime)
throw new BadRequestException({ code: 'TIME_RANGE_INVALID', message: '查询时间范围非法' });
if (endTime.getTime() - startTime.getTime() > auth.config.maxQueryRangeDays * 86400_000)
throw new BadRequestException({
code: 'TIME_RANGE_TOO_LARGE',
message: `单次查询不能超过${auth.config.maxQueryRangeDays}天`,
});
const limit = Math.min(Math.max(Number(query.limit) || 50, 1), auth.config.maxPageSize);
const cursor = decodeCursor(query.cursor);
const rows = await this.prisma.smsUplinkMessage.findMany({
@@ -293,9 +470,22 @@ export class OpenApiService implements OnModuleInit, OnModuleDestroy {
phoneNumber: query.mobile,
destId: query.accessNumber,
content: query.keyword ? { contains: query.keyword } : undefined,
...(cursor ? { OR: [{ receivedAt: { lt: cursor.receivedAt } }, { receivedAt: cursor.receivedAt, id: { lt: cursor.id } }] } : {}),
...(cursor
? {
OR: [{ receivedAt: { lt: cursor.receivedAt } }, { receivedAt: cursor.receivedAt, id: { lt: cursor.id } }],
}
: {}),
},
select: {
id: true,
messageId: true,
phoneNumber: true,
destId: true,
content: true,
matchStatus: true,
matchReason: true,
receivedAt: true,
},
select: { id: true, messageId: true, phoneNumber: true, destId: true, content: true, matchStatus: true, matchReason: true, receivedAt: true },
orderBy: [{ receivedAt: 'desc' }, { id: 'desc' }],
take: limit + 1,
});
@@ -306,29 +496,55 @@ export class OpenApiService implements OnModuleInit, OnModuleDestroy {
}
async getUplink(auth: OpenApiAuthContext, uplinkId: string) {
if (!auth.config.uplinkQueryEnabled) throw new ForbiddenException({ code: 'UPLINK_QUERY_NOT_ENABLED', message: '该应用未开通上行查询' });
const row = await this.prisma.smsUplinkMessage.findFirst({ where: { id: uplinkId, applicationId: auth.application.id, matchStatus: 'matched' } });
if (!auth.config.uplinkQueryEnabled)
throw new ForbiddenException({ code: 'UPLINK_QUERY_NOT_ENABLED', message: '该应用未开通上行查询' });
const row = await this.prisma.smsUplinkMessage.findFirst({
where: { id: uplinkId, applicationId: auth.application.id, matchStatus: 'matched' },
});
if (!row) throw new NotFoundException({ code: 'UPLINK_NOT_FOUND', message: '上行记录不存在' });
return row;
}
async queueWebhookEvent(data: { tenantId: string; applicationId?: string | null; messageRecordId?: string | null; messageId?: string | null; uplinkMessageId?: string | null; eventType: 'receipt' | 'uplink'; payload: Record<string, unknown> }) {
async queueWebhookEvent(data: {
tenantId: string;
applicationId?: string | null;
messageRecordId?: string | null;
messageId?: string | null;
uplinkMessageId?: string | null;
eventType: 'receipt' | 'uplink';
payload: Record<string, unknown>;
}) {
if (!data.applicationId) return null;
const application = await this.prisma.smsApplication.findUnique({ where: { id: data.applicationId }, include: { httpConfig: true } });
const application = await this.prisma.smsApplication.findUnique({
where: { id: data.applicationId },
include: { httpConfig: true },
});
const config = application?.httpConfig;
const enabled = data.eventType === 'receipt' ? config?.receiptWebhookEnabled : config?.uplinkWebhookEnabled;
if (!config?.enabled || !enabled) return null;
const endpoint = await this.prisma.httpWebhookEndpoint.findUnique({ where: { applicationId_eventType: { applicationId: data.applicationId, eventType: data.eventType } } });
const endpoint = await this.prisma.httpWebhookEndpoint.findUnique({
where: { applicationId_eventType: { applicationId: data.applicationId, eventType: data.eventType } },
});
if (!endpoint || endpoint.status !== 'active') return null;
const eventId = data.eventType === 'receipt' && data.messageRecordId
? `evt_receipt_${data.messageRecordId}`
: data.eventType === 'uplink' && data.uplinkMessageId
? `evt_uplink_${data.uplinkMessageId}`
: `evt_${randomUUID()}`;
const eventId =
data.eventType === 'receipt' && data.messageRecordId
? `evt_receipt_${data.messageRecordId}`
: data.eventType === 'uplink' && data.uplinkMessageId
? `evt_uplink_${data.uplinkMessageId}`
: `evt_${randomUUID()}`;
const event = await this.prisma.httpWebhookEvent.upsert({
where: { eventId },
update: {},
create: { eventId, tenantId: data.tenantId, applicationId: data.applicationId, eventType: data.eventType, messageRecordId: data.messageRecordId, messageId: data.messageId, uplinkMessageId: data.uplinkMessageId, payload: data.payload as Prisma.InputJsonValue },
create: {
eventId,
tenantId: data.tenantId,
applicationId: data.applicationId,
eventType: data.eventType,
messageRecordId: data.messageRecordId,
messageId: data.messageId,
uplinkMessageId: data.uplinkMessageId,
payload: data.payload as Prisma.InputJsonValue,
},
});
const delivery = await this.prisma.httpWebhookDelivery.upsert({
where: { eventId_endpointId: { eventId: event.id, endpointId: endpoint.id } },
@@ -336,57 +552,135 @@ export class OpenApiService implements OnModuleInit, OnModuleDestroy {
create: { eventId: event.id, endpointId: endpoint.id },
});
if (delivery.status === 'delivered') return delivery;
await this.queue?.add('deliver', { deliveryId: delivery.id }, { jobId: delivery.id, removeOnComplete: 1000, removeOnFail: 1000 });
await this.queue?.add(
'deliver',
{ deliveryId: delivery.id },
{ jobId: delivery.id, removeOnComplete: 1000, removeOnFail: 1000 },
);
return delivery;
}
async listRequestLogs(applicationId: string, tenantId?: string) {
await this.requireApplication(applicationId, tenantId);
return this.prisma.openApiRequest.findMany({ where: { applicationId }, select: { id: true, requestId: true, clientMessageId: true, sourceIp: true, httpStatus: true, businessCode: true, status: true, durationMs: true, createdAt: true, completedAt: true }, orderBy: { createdAt: 'desc' }, take: 100 });
return this.prisma.openApiRequest.findMany({
where: { applicationId },
select: {
id: true,
requestId: true,
clientMessageId: true,
sourceIp: true,
httpStatus: true,
businessCode: true,
status: true,
durationMs: true,
createdAt: true,
completedAt: true,
},
orderBy: { createdAt: 'desc' },
take: 100,
});
}
async listWebhookDeliveries(applicationId: string, tenantId?: string) {
await this.requireApplication(applicationId, tenantId);
return this.prisma.httpWebhookDelivery.findMany({ where: { event: { applicationId } }, include: { event: true, endpoint: { select: { eventType: true, url: true } }, attempts: { orderBy: { attemptNo: 'desc' }, take: 5 } }, orderBy: { createdAt: 'desc' }, take: 100 });
return this.prisma.httpWebhookDelivery.findMany({
where: { event: { applicationId } },
include: {
event: true,
endpoint: { select: { eventType: true, url: true } },
attempts: { orderBy: { attemptNo: 'desc' }, take: 5 },
},
orderBy: { createdAt: 'desc' },
take: 100,
});
}
async retryWebhookDelivery(applicationId: string, deliveryId: string, tenantId?: string) {
const application = await this.requireApplication(applicationId, tenantId);
if (tenantId && !application.httpConfig?.allowClientManualRetry) throw new ForbiddenException('该应用未开通客户端手动重投');
const delivery = await this.prisma.httpWebhookDelivery.findFirst({ where: { id: deliveryId, event: { applicationId } } });
if (tenantId && !application.httpConfig?.allowClientManualRetry)
throw new ForbiddenException('该应用未开通客户端手动重投');
const delivery = await this.prisma.httpWebhookDelivery.findFirst({
where: { id: deliveryId, event: { applicationId } },
});
if (!delivery) throw new NotFoundException('Webhook投递记录不存在');
await this.prisma.httpWebhookDelivery.update({ where: { id: delivery.id }, data: { status: 'pending', nextRetryAt: null, lastError: null } });
await this.queue?.add('deliver', { deliveryId }, { jobId: `${deliveryId}:manual:${Date.now()}`, removeOnComplete: 1000, removeOnFail: 1000 });
await this.prisma.httpWebhookDelivery.update({
where: { id: delivery.id },
data: { status: 'pending', nextRetryAt: null, lastError: null },
});
await this.queue?.add(
'deliver',
{ deliveryId },
{ jobId: `${deliveryId}:manual:${Date.now()}`, removeOnComplete: 1000, removeOnFail: 1000 },
);
return { id: deliveryId, status: 'pending' };
}
private async deliverWebhook(deliveryId: string) {
const delivery = await this.prisma.httpWebhookDelivery.findUnique({ where: { id: deliveryId }, include: { event: true, endpoint: true } });
const delivery = await this.prisma.httpWebhookDelivery.findUnique({
where: { id: deliveryId },
include: { event: true, endpoint: true },
});
if (!delivery || delivery.status === 'delivered') return;
const config = await this.prisma.smsApplicationHttpConfig.findUnique({ where: { applicationId: delivery.event.applicationId } });
const config = await this.prisma.smsApplicationHttpConfig.findUnique({
where: { applicationId: delivery.event.applicationId },
});
if (!config) return;
const attemptNo = delivery.attemptCount + 1;
const timestamp = String(Math.floor(Date.now() / 1000));
const body = JSON.stringify({ eventId: delivery.event.eventId, eventType: delivery.event.eventType, occurredAt: delivery.event.createdAt.toISOString(), data: delivery.event.payload });
const signature = createHmac('sha256', decryptSecret(delivery.endpoint.secretEncrypted)).update(`${timestamp}\n${body}`).digest('hex');
const body = JSON.stringify({
eventId: delivery.event.eventId,
eventType: delivery.event.eventType,
occurredAt: delivery.event.createdAt.toISOString(),
data: delivery.event.payload,
});
const signature = createHmac('sha256', decryptSecret(delivery.endpoint.secretEncrypted))
.update(`${timestamp}\n${body}`)
.digest('hex');
const startedAt = Date.now();
let responseStatus: number | undefined;
let responseSummary: string | undefined;
let errorMessage: string | undefined;
try {
const response = await postWebhook(delivery.endpoint.url, body, {
'content-type': 'application/json',
'x-event-id': delivery.event.eventId,
'x-event-type': delivery.event.eventType,
'x-timestamp': timestamp,
'x-signature': `sha256=${signature}`,
}, config.webhookTimeoutSeconds * 1000, config.requireHttps);
const response = await postWebhook(
delivery.endpoint.url,
body,
{
'content-type': 'application/json',
'x-event-id': delivery.event.eventId,
'x-event-type': delivery.event.eventType,
'x-timestamp': timestamp,
'x-signature': `sha256=${signature}`,
},
config.webhookTimeoutSeconds * 1000,
config.requireHttps,
);
responseStatus = response.status;
responseSummary = response.body;
} catch (error) { errorMessage = error instanceof Error ? error.message : 'Webhook request failed'; }
} catch (error) {
errorMessage = error instanceof Error ? error.message : 'Webhook request failed';
}
const success = responseStatus !== undefined && responseStatus >= 200 && responseStatus < 300;
const retryable = errorMessage !== undefined || responseStatus === 408 || responseStatus === 429 || (responseStatus !== undefined && responseStatus >= 500);
await this.prisma.httpWebhookAttempt.create({ data: { deliveryId, attemptNo, responseStatus, responseSummary, errorMessage, durationMs: Date.now() - startedAt, requestHeaders: { 'x-event-id': delivery.event.eventId, 'x-event-type': delivery.event.eventType, 'x-timestamp': timestamp, 'x-signature': 'sha256=***' } } });
const retryable =
errorMessage !== undefined ||
responseStatus === 408 ||
responseStatus === 429 ||
(responseStatus !== undefined && responseStatus >= 500);
await this.prisma.httpWebhookAttempt.create({
data: {
deliveryId,
attemptNo,
responseStatus,
responseSummary,
errorMessage,
durationMs: Date.now() - startedAt,
requestHeaders: {
'x-event-id': delivery.event.eventId,
'x-event-type': delivery.event.eventType,
'x-timestamp': timestamp,
'x-signature': 'sha256=***',
},
},
});
this.protocolLogs?.record({
protocol: 'http',
direction: 'platform_to_client',
@@ -403,22 +697,62 @@ export class OpenApiService implements OnModuleInit, OnModuleDestroy {
detail: { deliveryId, attemptNo, error: errorMessage },
});
if (success) {
await this.prisma.httpWebhookDelivery.update({ where: { id: deliveryId }, data: { status: 'delivered', attemptCount: attemptNo, lastHttpStatus: responseStatus, lastError: null, deliveredAt: new Date(), nextRetryAt: null } });
await this.prisma.httpWebhookDelivery.update({
where: { id: deliveryId },
data: {
status: 'delivered',
attemptCount: attemptNo,
lastHttpStatus: responseStatus,
lastError: null,
deliveredAt: new Date(),
nextRetryAt: null,
},
});
return;
}
const maxAttempts = Math.min(config.webhookMaxAttempts, RETRY_DELAYS_SECONDS.length);
if (config.webhookRetryEnabled && retryable && attemptNo < maxAttempts) {
const delaySeconds = RETRY_DELAYS_SECONDS[attemptNo] ?? RETRY_DELAYS_SECONDS.at(-1)!;
const nextRetryAt = new Date(Date.now() + delaySeconds * 1000);
await this.prisma.httpWebhookDelivery.update({ where: { id: deliveryId }, data: { status: 'retrying', attemptCount: attemptNo, lastHttpStatus: responseStatus, lastError: errorMessage ?? `HTTP ${responseStatus}`, nextRetryAt } });
await this.queue?.add('deliver', { deliveryId }, { jobId: `${deliveryId}:${attemptNo + 1}`, delay: delaySeconds * 1000, removeOnComplete: 1000, removeOnFail: 1000 });
await this.prisma.httpWebhookDelivery.update({
where: { id: deliveryId },
data: {
status: 'retrying',
attemptCount: attemptNo,
lastHttpStatus: responseStatus,
lastError: errorMessage ?? `HTTP ${responseStatus}`,
nextRetryAt,
},
});
await this.queue?.add(
'deliver',
{ deliveryId },
{
jobId: `${deliveryId}:${attemptNo + 1}`,
delay: delaySeconds * 1000,
removeOnComplete: 1000,
removeOnFail: 1000,
},
);
return;
}
await this.prisma.httpWebhookDelivery.update({ where: { id: deliveryId }, data: { status: 'failed', attemptCount: attemptNo, lastHttpStatus: responseStatus, lastError: errorMessage ?? `HTTP ${responseStatus}`, nextRetryAt: null } });
await this.prisma.httpWebhookDelivery.update({
where: { id: deliveryId },
data: {
status: 'failed',
attemptCount: attemptNo,
lastHttpStatus: responseStatus,
lastError: errorMessage ?? `HTTP ${responseStatus}`,
nextRetryAt: null,
},
});
}
private async requireApplication(applicationId: string, tenantId?: string) {
const application = await this.prisma.smsApplication.findFirst({ where: { id: applicationId, tenantId }, include: { httpConfig: true, httpIpAllowlist: true } });
const application = await this.prisma.smsApplication.findFirst({
where: { id: applicationId, tenantId },
include: { httpConfig: true, httpIpAllowlist: true },
});
if (!application) throw new NotFoundException('企业应用不存在');
return application;
}
@@ -428,12 +762,22 @@ function httpApiPublicOrigin() {
const configured = process.env.HTTP_API_PUBLIC_ORIGIN?.trim().replace(/\/+$/, '');
if (!configured) return undefined;
const url = new URL(configured);
const insecureHttpExplicitlyAllowed = process.env.HTTP_API_ALLOW_INSECURE_ORIGIN === 'true' && url.protocol === 'http:';
if ((url.protocol !== 'https:' && !insecureHttpExplicitlyAllowed) || url.username || url.password || url.pathname !== '/' || url.search || url.hash) {
const insecureHttpExplicitlyAllowed =
process.env.HTTP_API_ALLOW_INSECURE_ORIGIN === 'true' && url.protocol === 'http:';
if (
(url.protocol !== 'https:' && !insecureHttpExplicitlyAllowed) ||
url.username ||
url.password ||
url.pathname !== '/' ||
url.search ||
url.hash
) {
// This value is copied into customer integration parameters, so fail closed instead of
// publishing an insecure or path-dependent endpoint unless an isolated test environment
// has explicitly opted into plain HTTP.
throw new Error('HTTP_API_PUBLIC_ORIGIN必须是无路径、无凭据的HTTPS源地址；隔离测试环境如需HTTP须显式启用HTTP_API_ALLOW_INSECURE_ORIGIN');
throw new Error(
'HTTP_API_PUBLIC_ORIGIN必须是无路径、无凭据的HTTPS源地址；隔离测试环境如需HTTP须显式启用HTTP_API_ALLOW_INSECURE_ORIGIN',
);
}
return url.origin;
}
@@ -441,15 +785,22 @@ function httpApiPublicOrigin() {
function normalizeOpenApiFailure(error: unknown) {
if (error instanceof HttpException) {
const value = error.getResponse();
const object = typeof value === 'object' && value ? value as Record<string, unknown> : {};
const object = typeof value === 'object' && value ? (value as Record<string, unknown>) : {};
const rawMessage = object.message ?? error.message;
return {
httpStatus: error.getStatus(),
code: String(object.code ?? 'SEND_REJECTED'),
responseBody: { code: String(object.code ?? 'SEND_REJECTED'), message: Array.isArray(rawMessage) ? rawMessage.join('；') : String(rawMessage) } as Prisma.InputJsonValue,
responseBody: {
code: String(object.code ?? 'SEND_REJECTED'),
message: Array.isArray(rawMessage) ? rawMessage.join('；') : String(rawMessage),
} as Prisma.InputJsonValue,
};
}
return { httpStatus: 500, code: 'INTERNAL_ERROR', responseBody: { code: 'INTERNAL_ERROR', message: 'Internal server error' } as Prisma.InputJsonValue };
return {
httpStatus: 500,
code: 'INTERNAL_ERROR',
responseBody: { code: 'INTERNAL_ERROR', message: 'Internal server error' } as Prisma.InputJsonValue,
};
}
function normalizeConfig(
@@ -458,15 +809,17 @@ function normalizeConfig(
cmppEnabled: boolean,
) {
const enabling = input.enabled === true && existing?.enabled !== true;
const effective = enabling ? {
sendEnabled: true,
messageQueryEnabled: true,
receiptWebhookEnabled: true,
uplinkWebhookEnabled: true,
uplinkQueryEnabled: true,
credentialSelfServiceEnabled: true,
...input,
} : input;
const effective = enabling
? {
sendEnabled: true,
messageQueryEnabled: true,
receiptWebhookEnabled: true,
uplinkWebhookEnabled: true,
uplinkQueryEnabled: true,
credentialSelfServiceEnabled: true,
...input,
}
: input;
const httpEnabled = effective.enabled ?? existing?.enabled ?? false;
const deliveryMode = automaticDeliveryMode(cmppEnabled, httpEnabled);
return {
@@ -496,22 +849,31 @@ function normalizeConfig(
function bounded(value: number | undefined, min: number, max: number, label: string) {
if (value === undefined) return undefined;
if (!Number.isInteger(value) || value < min || value > max) throw new BadRequestException(`${label}必须在${min}至${max}之间`);
if (!Number.isInteger(value) || value < min || value > max)
throw new BadRequestException(`${label}必须在${min}至${max}之间`);
return value;
}
function normalizeIpAllowlist(values?: string[]) {
return [...new Set((values ?? []).map((item) => item.trim()).filter(Boolean).map((item) => {
const [ip, prefix] = item.split('/');
const version = isIP(ip);
if (!version) throw new BadRequestException(`IP白名单格式非法：${item}`);
if (prefix !== undefined) {
const bits = Number(prefix);
const max = version === 4 ? 32 : 128;
if (!Number.isInteger(bits) || bits < 0 || bits > max) throw new BadRequestException(`CIDR格式非法：${item}`);
}
return item;
}))];
return [
...new Set(
(values ?? [])
.map((item) => item.trim())
.filter(Boolean)
.map((item) => {
const [ip, prefix] = item.split('/');
const version = isIP(ip);
if (!version) throw new BadRequestException(`IP白名单格式非法：${item}`);
if (prefix !== undefined) {
const bits = Number(prefix);
const max = version === 4 ? 32 : 128;
if (!Number.isInteger(bits) || bits < 0 || bits > max)
throw new BadRequestException(`CIDR格式非法：${item}`);
}
return item;
}),
),
];
}
async function validateWebhookUrl(value: string, requireHttps: boolean) {
@@ -520,37 +882,54 @@ async function validateWebhookUrl(value: string, requireHttps: boolean) {
async function resolveWebhookTarget(value: string, requireHttps: boolean) {
let url: URL;
try { url = new URL(String(value ?? '').trim()); } catch { throw new BadRequestException('Webhook URL格式非法'); }
try {
url = new URL(String(value ?? '').trim());
} catch {
throw new BadRequestException('Webhook URL格式非法');
}
if (!['http:', 'https:'].includes(url.protocol)) throw new BadRequestException('Webhook仅支持HTTP/HTTPS');
if (requireHttps && url.protocol !== 'https:') throw new BadRequestException('当前应用要求Webhook使用HTTPS');
if (url.username || url.password) throw new BadRequestException('Webhook URL不能包含用户名或密码');
const addresses = isIP(url.hostname) ? [{ address: url.hostname }] : await lookup(url.hostname, { all: true });
if (addresses.some(({ address }) => isPrivateAddress(address))) throw new BadRequestException('Webhook URL不能指向内网、环回或链路本地地址');
if (addresses.some(({ address }) => isPrivateAddress(address)))
throw new BadRequestException('Webhook URL不能指向内网、环回或链路本地地址');
const selected = addresses[0];
if (!selected) throw new BadRequestException('Webhook域名未解析到可用地址');
return { url, address: selected.address, family: isIP(selected.address) };
}
async function postWebhook(urlText: string, body: string, headers: Record<string, string>, timeoutMs: number, requireHttps: boolean) {
async function postWebhook(
urlText: string,
body: string,
headers: Record<string, string>,
timeoutMs: number,
requireHttps: boolean,
) {
const target = await resolveWebhookTarget(urlText, requireHttps);
return new Promise<{ status: number; body: string }>((resolve, reject) => {
const requestFn = target.url.protocol === 'https:' ? httpsRequest : httpRequest;
const request = requestFn(target.url, {
method: 'POST',
headers: { ...headers, 'content-length': String(Buffer.byteLength(body)) },
lookup: (_hostname, _options, callback) => callback(null, target.address, target.family),
}, (response) => {
const chunks: Buffer[] = [];
let size = 0;
response.on('data', (chunk: Buffer) => {
if (size < 1000) {
const buffer = Buffer.from(chunk);
chunks.push(buffer.subarray(0, 1000 - size));
size += buffer.length;
}
});
response.on('end', () => resolve({ status: response.statusCode ?? 0, body: Buffer.concat(chunks).toString('utf8') }));
});
const request = requestFn(
target.url,
{
method: 'POST',
headers: { ...headers, 'content-length': String(Buffer.byteLength(body)) },
lookup: (_hostname, _options, callback) => callback(null, target.address, target.family),
},
(response) => {
const chunks: Buffer[] = [];
let size = 0;
response.on('data', (chunk: Buffer) => {
if (size < 1000) {
const buffer = Buffer.from(chunk);
chunks.push(buffer.subarray(0, 1000 - size));
size += buffer.length;
}
});
response.on('end', () =>
resolve({ status: response.statusCode ?? 0, body: Buffer.concat(chunks).toString('utf8') }),
);
},
);
request.setTimeout(timeoutMs, () => request.destroy(new Error('Webhook request timed out')));
request.on('error', reject);
request.end(body);
@@ -559,13 +938,33 @@ async function postWebhook(urlText: string, body: string, headers: Record<string
function isPrivateAddress(address: string) {
const normalized = address.replace(/^::ffff:/, '');
if (normalized === '::1' || normalized === '::' || normalized.startsWith('fc') || normalized.startsWith('fd') || normalized.startsWith('fe8') || normalized.startsWith('fe9') || normalized.startsWith('fea') || normalized.startsWith('feb')) return true;
if (
normalized === '::1' ||
normalized === '::' ||
normalized.startsWith('fc') ||
normalized.startsWith('fd') ||
normalized.startsWith('fe8') ||
normalized.startsWith('fe9') ||
normalized.startsWith('fea') ||
normalized.startsWith('feb')
)
return true;
if (isIP(normalized) !== 4) return false;
const [a, b] = normalized.split('.').map(Number);
return a === 10 || a === 127 || a === 0 || (a === 169 && b === 254) || (a === 172 && b >= 16 && b <= 31) || (a === 192 && b === 168) || (a === 100 && b >= 64 && b <= 127);
return (
a === 10 ||
a === 127 ||
a === 0 ||
(a === 169 && b === 254) ||
(a === 172 && b >= 16 && b <= 31) ||
(a === 192 && b === 168) ||
(a === 100 && b >= 64 && b <= 127)
);
}
function encodeCursor(receivedAt: Date, id: string) { return Buffer.from(JSON.stringify([receivedAt.toISOString(), id])).toString('base64url'); }
function encodeCursor(receivedAt: Date, id: string) {
return Buffer.from(JSON.stringify([receivedAt.toISOString(), id])).toString('base64url');
}
function decodeCursor(value?: string) {
if (!value) return null;
try {
@@ -573,10 +972,19 @@ function decodeCursor(value?: string) {
const receivedAt = new Date(date);
if (!id || !Number.isFinite(receivedAt.getTime())) throw new Error();
return { receivedAt, id };
} catch { throw new BadRequestException({ code: 'CURSOR_INVALID', message: 'cursor格式非法' }); }
} catch {
throw new BadRequestException({ code: 'CURSOR_INVALID', message: 'cursor格式非法' });
}
}
function bullmqConnection() {
const url = new URL(process.env.REDIS_URL ?? 'redis://127.0.0.1:6379');
return { host: url.hostname, port: Number(url.port || 6379), username: url.username || undefined, password: url.password || undefined, db: Number(url.pathname.slice(1) || 0), maxRetriesPerRequest: null as null };
return {
host: url.hostname,
port: Number(url.port || 6379),
username: url.username || undefined,
password: url.password || undefined,
db: Number(url.pathname.slice(1) || 0),
maxRetriesPerRequest: null as null,
};
}