diff --git a/.env.example b/.env.example new file mode 100644 index 0000000..cbbfe86 --- /dev/null +++ b/.env.example @@ -0,0 +1,29 @@ +# Web +VITE_API_BASE_URL=http://localhost:3000/api + +# API +API_PORT=3000 +DATABASE_URL=postgresql://cmpp:cmpp_password@localhost:5432/cmpp_platform?schema=public +REDIS_HOST=127.0.0.1 +REDIS_PORT=6379 +REDIS_URL=redis://127.0.0.1:6379 +ADMIN_SESSION_IDLE_TIMEOUT_MS=3600000 +CLIENT_SESSION_IDLE_TIMEOUT_MS=7200000 +SESSION_LOCK_RECOVERY_MS=14400000 +SESSION_ABSOLUTE_TIMEOUT_MS=43200000 +SESSION_RECENT_AUTH_MS=1800000 +# Local HTTP development only. Production must use HTTPS and true. +SESSION_COOKIE_SECURE=false +MINIO_ENDPOINT=localhost:9000 +MINIO_ACCESS_KEY=cmpp_minio +MINIO_SECRET_KEY=cmpp_minio_password +MINIO_BUCKET=cmpp-platform + +# Gateway +GATEWAY_HEALTH_ADDR=:8090 +GATEWAY_REDIS_ADDR=127.0.0.1:6379 +GATEWAY_CMPP_VERSION=3.0 +GATEWAY_CMPP_ADDR=127.0.0.1:7890 +GATEWAY_CMPP_USER=900001 +GATEWAY_CMPP_PASSWORD=888888 +CMPP_DOWNSTREAM_ACK_TIMEOUT_SECONDS=30 diff --git a/.gitignore b/.gitignore index 444ebd8..b6391b3 100644 --- a/.gitignore +++ b/.gitignore @@ -2,6 +2,7 @@ node_modules/ dist/ .env .env.* +!.env.example *.local .local-tools/ .local-data/ @@ -11,6 +12,7 @@ yarn-debug.log* yarn-error.log* pnpm-debug.log* logs/ +gateway/gateway.exe dump.rdb .DS_Store diff --git a/api/prisma/migrations/20260714130000_add_downstream_delivery_ack_tracking/migration.sql b/api/prisma/migrations/20260714130000_add_downstream_delivery_ack_tracking/migration.sql new file mode 100644 index 0000000..92ebccb --- /dev/null +++ b/api/prisma/migrations/20260714130000_add_downstream_delivery_ack_tracking/migration.sql @@ -0,0 +1,32 @@ +ALTER TABLE "SmsApplication" + ADD COLUMN "downstreamReceiptRetryEnabled" BOOLEAN NOT NULL DEFAULT true, + ADD COLUMN "downstreamUplinkRetryEnabled" BOOLEAN NOT NULL DEFAULT true; + +ALTER TABLE "CmppDownstreamDelivery" + ADD COLUMN "retryEnabled" BOOLEAN NOT NULL DEFAULT true, + ADD COLUMN "sentAt" TIMESTAMP(3), + ADD COLUMN "acknowledgedAt" TIMESTAMP(3), + ADD COLUMN "ackDeadlineAt" TIMESTAMP(3), + ADD COLUMN "ackResult" INTEGER, + ADD COLUMN "ackSequenceId" TEXT, + ADD COLUMN "ackMessageId" TEXT, + ADD COLUMN "connectionId" TEXT; + +UPDATE "CmppDownstreamDelivery" +SET + "status" = 'unconfirmed', + "sentAt" = "deliveredAt", + "deliveredAt" = NULL, + "lastError" = '历史记录仅确认 Gateway 已写出，未留存 CMPP_DELIVER_RESP' +WHERE "status" = 'delivered'; + +UPDATE "CmppDownstreamDelivery" +SET "retryEnabled" = CASE + WHEN "deliveryType" = 'uplink' THEN app."downstreamUplinkRetryEnabled" + ELSE app."downstreamReceiptRetryEnabled" +END +FROM "SmsApplication" app +WHERE app.id = "CmppDownstreamDelivery"."applicationId"; + +CREATE INDEX "CmppDownstreamDelivery_status_ackDeadlineAt_idx" + ON "CmppDownstreamDelivery"("status", "ackDeadlineAt"); diff --git a/api/prisma/schema.prisma b/api/prisma/schema.prisma index 41469b6..67698be 100644 --- a/api/prisma/schema.prisma +++ b/api/prisma/schema.prisma @@ -360,6 +360,8 @@ model SmsApplication { queuePriority String @default("normal") maxPhonesPerTask Int @default(1000000) templateMismatchMode String @default("reject") + downstreamReceiptRetryEnabled Boolean @default(true) + downstreamUplinkRetryEnabled Boolean @default(true) status String @default("active") createdAt DateTime @default(now()) updatedAt DateTime @updatedAt @@ -1188,8 +1190,16 @@ model CmppDownstreamDelivery { deliveryType String status String @default("pending") payload Json + retryEnabled Boolean @default(true) retryCount Int @default(0) nextRetryAt DateTime? + sentAt DateTime? + acknowledgedAt DateTime? + ackDeadlineAt DateTime? + ackResult Int? + ackSequenceId String? + ackMessageId String? + connectionId String? deliveredAt DateTime? lastError String? createdAt DateTime @default(now()) @@ -1204,6 +1214,7 @@ model CmppDownstreamDelivery { @@index([messageId]) @@index([messageRecordId]) @@index([deliveryType, status]) + @@index([status, ackDeadlineAt]) } model GatewaySubmitDeadLetter { diff --git a/api/src/auth/auth.controller.ts b/api/src/auth/auth.controller.ts index 363e9fc..6a6ecbb 100644 --- a/api/src/auth/auth.controller.ts +++ b/api/src/auth/auth.controller.ts @@ -1,13 +1,22 @@ -import { Body, Controller, Get, Post, UnauthorizedException } from '@nestjs/common'; +import { Body, Controller, Get, Post, Req, Res, UnauthorizedException } from '@nestjs/common'; import { ApiTags } from '@nestjs/swagger'; import { CurrentSessionUserId } from './current-session-user.decorator'; import { AuthService, LoginDto } from './auth.service'; +import { SessionService } from './session.service'; +import type { SessionRequest } from './session-validation.middleware'; import { UsersService } from '../users/users.service'; +import { PrismaService } from '../prisma/prisma.service'; +import { Prisma } from '@prisma/client'; + +type CookieResponse = { + cookie(name: string, value: string, options: Record): void; + clearCookie(name: string, options: Record): void; +}; @ApiTags('auth') @Controller() export class AuthController { - constructor(private readonly auth: AuthService, private readonly users: UsersService) {} + constructor(private readonly auth: AuthService, private readonly users: UsersService, private readonly sessions: SessionService, private readonly prisma: PrismaService) {} @Get('admin/auth/captcha') adminCaptcha() { @@ -15,8 +24,8 @@ export class AuthController { } @Post('admin/auth/login') - adminLogin(@Body() body: LoginDto) { - return this.auth.login(body, 'admin'); + async adminLogin(@Body() body: LoginDto, @Req() request: SessionRequest, @Res({ passthrough: true }) response: CookieResponse) { + return this.finishLogin(await this.auth.login(body, 'admin'), request, response); } @Get('client/auth/captcha') @@ -25,15 +34,101 @@ export class AuthController { } @Post('client/auth/login') - clientLogin(@Body() body: LoginDto) { - return this.auth.login(body, 'client'); + async clientLogin(@Body() body: LoginDto, @Req() request: SessionRequest, @Res({ passthrough: true }) response: CookieResponse) { + return this.finishLogin(await this.auth.login(body, 'client'), request, response); + } + + @Get('auth/session') + currentSession(@Req() request: SessionRequest) { + this.assertSession(request); + return this.sessions.publicSession(request.authSession!); + } + + @Post('auth/session/touch') + async touch(@Req() request: SessionRequest) { + this.assertSession(request); + const result = await this.sessions.touch(request.sessionToken!); + if (result.status !== 'active') throw new UnauthorizedException({ code: 'SESSION_LOCKED', message: '会话已锁定' }); + return this.sessions.publicSession(result.record); + } + + @Post('auth/session/lock') + async lock(@Req() request: SessionRequest) { + this.assertSession(request); + const record = await this.sessions.lock(request.sessionToken!); + if (record) await this.writeLog(request, 'auth.session_locked', { portal: record.portal, reason: 'client_idle_timer' }); + return { locked: Boolean(record) }; + } + + @Post('auth/session/unlock') + async unlock(@Req() request: SessionRequest, @Body('password') password: string, @Res({ passthrough: true }) response: CookieResponse) { + this.assertSession(request); + const result = await this.auth.unlock(request.sessionToken!, request.sessionUserId!, password); + if (result.status !== 'active' || !('token' in result)) throw new UnauthorizedException({ code: 'SESSION_LOCK_TIMEOUT', message: '锁定时间过长，请重新登录' }); + this.setCookie(response, result.token); + await this.writeLog(request, 'auth.session_unlocked', { portal: result.record.portal }); + return this.sessions.publicSession(result.record); + } + + @Post('auth/reauthenticate') + async reauthenticate(@Req() request: SessionRequest, @Body('password') password: string) { + this.assertSession(request); + const result = await this.auth.reauthenticate(request.sessionToken!, request.sessionUserId!, password); + if (result.status !== 'active') throw new UnauthorizedException({ code: 'SESSION_LOCKED', message: '会话已锁定' }); + await this.writeLog(request, 'auth.session_reauthenticated', { portal: result.record.portal }); + return this.sessions.publicSession(result.record); + } + + @Post('auth/logout') + async logout(@Req() request: SessionRequest, @Res({ passthrough: true }) response: CookieResponse) { + if (request.sessionToken) await this.sessions.remove(request.sessionToken); + if (request.sessionUserId) await this.writeLog(request, 'auth.session_logged_out', { portal: request.authSession?.portal }); + this.clearCookie(response); + return { success: true }; } @Post('auth/password') changeOwnPassword(@CurrentSessionUserId() userId: string | undefined, @Body() body: { currentPassword?: string; password?: string }) { - if (!userId) { - throw new UnauthorizedException('登录会话无效，请重新登录'); - } + if (!userId) throw new UnauthorizedException('登录会话无效，请重新登录'); return this.users.changeOwnPassword(userId, body.currentPassword ?? '', body.password ?? ''); } + + private async finishLogin(result: Awaited>, request: SessionRequest, response: CookieResponse) { + this.setCookie(response, result.sessionToken); + await this.prisma.operationLog.create({ + data: { userId: result.user.id, tenantId: result.user.tenantId, action: 'auth.session_created', resource: 'auth_session', userAgent: request.header('user-agent'), detail: { portal: result.portal } }, + }); + const { sessionToken: _, ...publicResult } = result; + return publicResult; + } + + private writeLog(request: SessionRequest, action: string, detail: Record) { + return this.prisma.operationLog.create({ + data: { userId: request.sessionUserId, action, resource: 'auth_session', userAgent: request.header('user-agent'), detail: JSON.parse(JSON.stringify(detail)) as Prisma.InputJsonValue }, + }); + } + + private assertSession(request: SessionRequest) { + if (!request.sessionToken || !request.sessionUserId || !request.authSession) { + throw new UnauthorizedException({ code: 'SESSION_INVALID', message: '登录会话无效，请重新登录' }); + } + } + + private setCookie(response: CookieResponse, token: string) { + response.cookie(this.sessions.cookieName, token, { + httpOnly: true, + secure: this.sessions.cookieSecure, + sameSite: 'lax', + path: '/', + }); + } + + private clearCookie(response: CookieResponse) { + response.clearCookie(this.sessions.cookieName, { + httpOnly: true, + secure: this.sessions.cookieSecure, + sameSite: 'lax', + path: '/', + }); + } } diff --git a/api/src/auth/auth.module.ts b/api/src/auth/auth.module.ts index ecbcc3a..d427ebe 100644 --- a/api/src/auth/auth.module.ts +++ b/api/src/auth/auth.module.ts @@ -1,11 +1,20 @@ import { Module } from '@nestjs/common'; +import { APP_GUARD } from '@nestjs/core'; import { UsersModule } from '../users/users.module'; import { AuthController } from './auth.controller'; import { AuthService } from './auth.service'; +import { RecentAuthenticationGuard } from './recent-authentication.guard'; +import { SessionService } from './session.service'; @Module({ imports: [UsersModule], controllers: [AuthController], - providers: [AuthService], + providers: [ + AuthService, + SessionService, + RecentAuthenticationGuard, + { provide: APP_GUARD, useExisting: RecentAuthenticationGuard }, + ], + exports: [SessionService], }) export class AuthModule {} diff --git a/api/src/auth/auth.service.spec.ts b/api/src/auth/auth.service.spec.ts index 4d88c52..3527a47 100644 --- a/api/src/auth/auth.service.spec.ts +++ b/api/src/auth/auth.service.spec.ts @@ -25,6 +25,17 @@ function createUsersMock(roleCode: string, overrides: Record = }; } +function createSessionsMock() { + const record = { + userId: 'user-1', portal: 'admin', sessionVersion: 0, createdAt: 1, lastActivityAt: 1, + lastAuthenticatedAt: 1, absoluteExpiresAt: Date.now() + 1000, + }; + return { + create: jest.fn().mockResolvedValue({ token: 'opaque-session-token', record }), + publicSession: jest.fn().mockReturnValue({ idleTimeoutSeconds: 3600, absoluteExpiresAt: new Date(record.absoluteExpiresAt).toISOString() }), + }; +} + async function loginWithCaptcha(service: AuthService, portal: 'admin' | 'client', password = 'secret1') { const captcha = service.createCaptcha(); const answer = captcha.challenge.split('=')[0].split('+').map((part) => Number(part.trim())).reduce((sum, value) => sum + value, 0); @@ -39,21 +50,23 @@ async function loginWithCaptcha(service: AuthService, portal: 'admin' | 'client' describe('AuthService', () => { it('allows platform admins to login admin portal', async () => { const users = createUsersMock('platform_admin'); - const service = new AuthService(users as never); - await expect(loginWithCaptcha(service, 'admin')).resolves.toEqual(expect.objectContaining({ portal: 'admin', accessToken: 'dev-token:user-1:0' })); + const sessions = createSessionsMock(); + const service = new AuthService(users as never, sessions as never); + await expect(loginWithCaptcha(service, 'admin')).resolves.toEqual(expect.objectContaining({ portal: 'admin', sessionToken: 'opaque-session-token' })); expect(users.recordLoginSuccess).toHaveBeenCalledWith('user-1'); + expect(sessions.create).toHaveBeenCalledWith('user-1', 'admin', 0); }); it('rejects enterprise admins on admin portal', async () => { const users = createUsersMock('enterprise_admin'); - const service = new AuthService(users as never); + const service = new AuthService(users as never, createSessionsMock() as never); await expect(loginWithCaptcha(service, 'admin')).rejects.toBeInstanceOf(UnauthorizedException); expect(users.recordLoginFailure).toHaveBeenCalledWith('user-1'); }); it('locks user after five failed password attempts', async () => { const users = createUsersMock('platform_admin'); - const service = new AuthService(users as never); + const service = new AuthService(users as never, createSessionsMock() as never); for (let index = 0; index < 5; index += 1) { await expect(loginWithCaptcha(service, 'admin', 'bad-password')).rejects.toBeInstanceOf(UnauthorizedException); } diff --git a/api/src/auth/auth.service.ts b/api/src/auth/auth.service.ts index 27ce5fb..8126a37 100644 --- a/api/src/auth/auth.service.ts +++ b/api/src/auth/auth.service.ts @@ -1,6 +1,7 @@ import { randomUUID } from 'node:crypto'; import { BadRequestException, Injectable, UnauthorizedException } from '@nestjs/common'; import { hashPassword, UsersService } from '../users/users.service'; +import { SessionService } from './session.service'; export interface LoginDto { login: string; @@ -21,7 +22,7 @@ const anonymousFailures = new Map) { + return { + getHandler: jest.fn(), + getClass: jest.fn(), + switchToHttp: () => ({ getRequest: () => ({ authSession: record }) }), + }; +} + +describe('RecentAuthenticationGuard', () => { + it('requires a fresh password verification on decorated operations', () => { + const reflector = { getAllAndOverride: jest.fn().mockReturnValue(true) }; + const sessions = { isRecentlyAuthenticated: jest.fn().mockReturnValue(false) }; + const guard = new RecentAuthenticationGuard(reflector as never, sessions as never); + expect(() => guard.canActivate(context({}) as never)).toThrow(ForbiddenException); + }); + + it('allows a decorated operation during the recent authentication window', () => { + const reflector = { getAllAndOverride: jest.fn().mockReturnValue(true) }; + const sessions = { isRecentlyAuthenticated: jest.fn().mockReturnValue(true) }; + const guard = new RecentAuthenticationGuard(reflector as never, sessions as never); + expect(guard.canActivate(context({}) as never)).toBe(true); + }); +}); diff --git a/api/src/auth/recent-authentication.guard.ts b/api/src/auth/recent-authentication.guard.ts new file mode 100644 index 0000000..f343052 --- /dev/null +++ b/api/src/auth/recent-authentication.guard.ts @@ -0,0 +1,21 @@ +import { CanActivate, ExecutionContext, ForbiddenException, Injectable } from '@nestjs/common'; +import { Reflector } from '@nestjs/core'; +import { REQUIRE_RECENT_AUTHENTICATION } from './require-recent-authentication.decorator'; +import { SessionService } from './session.service'; +import type { SessionRequest } from './session-validation.middleware'; + +@Injectable() +export class RecentAuthenticationGuard implements CanActivate { + constructor(private readonly reflector: Reflector, private readonly sessions: SessionService) {} + + canActivate(context: ExecutionContext) { + const required = this.reflector.getAllAndOverride(REQUIRE_RECENT_AUTHENTICATION, [ + context.getHandler(), + context.getClass(), + ]); + if (!required) return true; + const request = context.switchToHttp().getRequest(); + if (request.authSession && this.sessions.isRecentlyAuthenticated(request.authSession)) return true; + throw new ForbiddenException({ code: 'RECENT_AUTHENTICATION_REQUIRED', message: '该操作需要重新验证当前密码' }); + } +} diff --git a/api/src/auth/require-recent-authentication.decorator.ts b/api/src/auth/require-recent-authentication.decorator.ts new file mode 100644 index 0000000..043ef45 --- /dev/null +++ b/api/src/auth/require-recent-authentication.decorator.ts @@ -0,0 +1,4 @@ +import { SetMetadata } from '@nestjs/common'; + +export const REQUIRE_RECENT_AUTHENTICATION = 'requireRecentAuthentication'; +export const RequireRecentAuthentication = () => SetMetadata(REQUIRE_RECENT_AUTHENTICATION, true); diff --git a/api/src/auth/session-validation.middleware.spec.ts b/api/src/auth/session-validation.middleware.spec.ts index 6c3110c..d89cf00 100644 --- a/api/src/auth/session-validation.middleware.spec.ts +++ b/api/src/auth/session-validation.middleware.spec.ts @@ -1,27 +1,51 @@ import { UnauthorizedException } from '@nestjs/common'; import { SessionValidationMiddleware, type SessionRequest } from './session-validation.middleware'; -function request(authorization?: string): SessionRequest { - return { header: jest.fn().mockReturnValue(authorization) }; +const record = { + userId: 'user-1', portal: 'admin' as const, sessionVersion: 3, createdAt: 1, + lastActivityAt: 1, lastAuthenticatedAt: 1, absoluteExpiresAt: Date.now() + 1000, +}; + +function request(path = '/api/admin/users', cookie = 'cmpp_session=opaque-token'): SessionRequest { + return { + originalUrl: path, + header: jest.fn((name: string) => name === 'cookie' ? cookie : undefined), + }; } describe('SessionValidationMiddleware', () => { - it('accepts the current user session version and exposes the session user id', async () => { + it('accepts an active Redis session and exposes its user and record', async () => { const prisma = { user: { findUnique: jest.fn().mockResolvedValue({ id: 'user-1', status: 'active', deletedAt: null, sessionVersion: 3 }) } }; - const middleware = new SessionValidationMiddleware(prisma as never); - const currentRequest = request('Bearer dev-token:user-1:3'); + const sessions = { validate: jest.fn().mockResolvedValue({ status: 'active', record }), remove: jest.fn() }; + const middleware = new SessionValidationMiddleware(prisma as never, sessions as never); + const currentRequest = request(); const next = jest.fn(); - await middleware.use(currentRequest, {} as never, next); + await middleware.use(currentRequest, {}, next); expect(next).toHaveBeenCalledTimes(1); expect(currentRequest.sessionUserId).toBe('user-1'); + expect(currentRequest.sessionToken).toBe('opaque-token'); + expect(currentRequest.authSession).toEqual(record); }); - it('rejects an old session token after the user session version changes', async () => { + it('rejects a session after the user session version changes', async () => { const prisma = { user: { findUnique: jest.fn().mockResolvedValue({ id: 'user-1', status: 'active', deletedAt: null, sessionVersion: 4 }) } }; - const middleware = new SessionValidationMiddleware(prisma as never); + const sessions = { validate: jest.fn().mockResolvedValue({ status: 'active', record }), remove: jest.fn() }; + const middleware = new SessionValidationMiddleware(prisma as never, sessions as never); - await expect(middleware.use(request('Bearer dev-token:user-1:3'), {} as never, jest.fn())).rejects.toBeInstanceOf(UnauthorizedException); + await expect(middleware.use(request(), {}, jest.fn())).rejects.toBeInstanceOf(UnauthorizedException); + expect(sessions.remove).toHaveBeenCalledWith('opaque-token'); + }); + + it('only lets a locked session reach unlock and logout endpoints', async () => { + const prisma = { user: { findUnique: jest.fn().mockResolvedValue({ id: 'user-1', status: 'active', deletedAt: null, sessionVersion: 3 }) } }; + const sessions = { validate: jest.fn().mockResolvedValue({ status: 'locked', record }), remove: jest.fn() }; + const middleware = new SessionValidationMiddleware(prisma as never, sessions as never); + + await expect(middleware.use(request(), {}, jest.fn())).rejects.toBeInstanceOf(UnauthorizedException); + const next = jest.fn(); + await middleware.use(request('/api/auth/session/unlock'), {}, next); + expect(next).toHaveBeenCalled(); }); }); diff --git a/api/src/auth/session-validation.middleware.ts b/api/src/auth/session-validation.middleware.ts index 79e281f..f238e3f 100644 --- a/api/src/auth/session-validation.middleware.ts +++ b/api/src/auth/session-validation.middleware.ts @@ -1,33 +1,76 @@ import { Injectable, NestMiddleware, UnauthorizedException } from '@nestjs/common'; import { PrismaService } from '../prisma/prisma.service'; +import { AuthSessionRecord, DEVELOPMENT_SESSION_COOKIE_NAME, SESSION_COOKIE_NAME, SessionService } from './session.service'; export type SessionRequest = { header(name: string): string | undefined; + originalUrl?: string; + url?: string; sessionUserId?: string; + sessionToken?: string; + authSession?: AuthSessionRecord; }; @Injectable() export class SessionValidationMiddleware implements NestMiddleware { - constructor(private readonly prisma: PrismaService) {} + constructor(private readonly prisma: PrismaService, private readonly sessions: SessionService) {} async use(request: SessionRequest, _: unknown, next: () => void) { - const authorization = request.header('authorization'); - if (!authorization) { + const path = request.originalUrl ?? request.url ?? ''; + if (/\/(admin|client)\/auth\/(captcha|login)(?:\?|$)/.test(path)) { next(); return; } - const match = /^Bearer dev-token:([^:]+):(\d+)$/.exec(authorization.trim()); - if (!match) { - throw new UnauthorizedException('登录会话无效，请重新登录'); + + const token = this.readCookie(request.header('cookie')); + if (!token) { + if ((path.includes('/admin/') && !path.includes('/admin/gateway/')) || path.includes('/client/') || path.includes('/auth/')) { + throw this.unauthorized('SESSION_INVALID', '请先登录'); + } + next(); + return; } + const result = await this.sessions.validate(token, request.header('x-session-activity') === 'user'); + if (result.status === 'expired') { + throw this.unauthorized(result.code, result.code === 'SESSION_ABSOLUTE_TIMEOUT' ? '登录已满 12 小时，请重新登录' : '登录会话已失效，请重新登录'); + } + const user = await this.prisma.user.findUnique({ - where: { id: match[1] }, + where: { id: result.record.userId }, select: { id: true, status: true, deletedAt: true, sessionVersion: true }, }); - if (!user || user.status !== 'active' || user.deletedAt || user.sessionVersion !== Number(match[2])) { - throw new UnauthorizedException('登录会话已失效，请重新登录'); + if (!user || user.status !== 'active' || user.deletedAt || user.sessionVersion !== result.record.sessionVersion) { + await this.sessions.remove(token); + throw this.unauthorized('SESSION_REVOKED', '登录会话已被撤销，请重新登录'); } + if ((path.includes('/admin/') && result.record.portal !== 'admin') || (path.includes('/client/') && result.record.portal !== 'client')) { + throw this.unauthorized('SESSION_PORTAL_MISMATCH', '登录入口与当前会话不匹配'); + } + request.sessionUserId = user.id; + request.sessionToken = token; + request.authSession = result.record; + if (result.status === 'locked' && !path.includes('/auth/session/unlock') && !path.includes('/auth/logout')) { + if (result.newlyLocked) { + await this.prisma.operationLog.create({ + data: { userId: user.id, action: 'auth.session_locked', resource: 'auth_session', detail: { portal: result.record.portal, reason: 'idle_timeout' } }, + }); + } + throw this.unauthorized('SESSION_LOCKED', '由于长时间未操作，会话已安全锁定'); + } next(); } + + private readCookie(cookieHeader?: string) { + if (!cookieHeader) return undefined; + for (const part of cookieHeader.split(';')) { + const [name, ...value] = part.trim().split('='); + if (name === SESSION_COOKIE_NAME || name === DEVELOPMENT_SESSION_COOKIE_NAME) return decodeURIComponent(value.join('=')); + } + return undefined; + } + + private unauthorized(code: string, message: string) { + return new UnauthorizedException({ code, message }); + } } diff --git a/api/src/auth/session.service.spec.ts b/api/src/auth/session.service.spec.ts new file mode 100644 index 0000000..ea0982f --- /dev/null +++ b/api/src/auth/session.service.spec.ts @@ -0,0 +1,73 @@ +const values = new Map(); +const redis = { + get: jest.fn((key: string) => Promise.resolve(values.get(key) ?? null)), + set: jest.fn((key: string, value: string) => { values.set(key, value); return Promise.resolve('OK'); }), + del: jest.fn((key: string) => { values.delete(key); return Promise.resolve(1); }), + disconnect: jest.fn(), +}; + +jest.mock('ioredis', () => jest.fn(() => redis)); + +import { SessionService } from './session.service'; + +describe('SessionService', () => { + let now = 1_700_000_000_000; + + beforeEach(() => { + values.clear(); + jest.clearAllMocks(); + now = 1_700_000_000_000; + jest.spyOn(Date, 'now').mockImplementation(() => now); + process.env.ADMIN_SESSION_IDLE_TIMEOUT_MS = '3600000'; + process.env.CLIENT_SESSION_IDLE_TIMEOUT_MS = '7200000'; + process.env.SESSION_ABSOLUTE_TIMEOUT_MS = '43200000'; + process.env.SESSION_LOCK_RECOVERY_MS = '14400000'; + }); + + afterEach(() => { + jest.restoreAllMocks(); + delete process.env.ADMIN_SESSION_IDLE_TIMEOUT_MS; + delete process.env.CLIENT_SESSION_IDLE_TIMEOUT_MS; + delete process.env.SESSION_ABSOLUTE_TIMEOUT_MS; + delete process.env.SESSION_LOCK_RECOVERY_MS; + }); + + it('locks an admin session after 60 minutes without extending the 12 hour absolute deadline', async () => { + const service = new SessionService(); + const created = await service.create('user-1', 'admin', 2); + expect(created.record.absoluteExpiresAt).toBe(now + 12 * 60 * 60 * 1000); + + now += 60 * 60 * 1000 + 1; + const result = await service.validate(created.token, false); + expect(result.status).toBe('locked'); + if (result.status === 'locked') expect(result.record.lockedAt).toBe(now); + }); + + it('keeps a client session active before its 120 minute inactivity limit', async () => { + const service = new SessionService(); + const created = await service.create('user-1', 'client', 2); + now += 90 * 60 * 1000; + await expect(service.validate(created.token, false)).resolves.toEqual(expect.objectContaining({ status: 'active' })); + }); + + it('requires a full login after a session stays locked for four hours', async () => { + const service = new SessionService(); + const created = await service.create('user-1', 'admin', 2); + await service.lock(created.token); + now += 4 * 60 * 60 * 1000 + 1; + await expect(service.validate(created.token, false)).resolves.toEqual({ status: 'expired', code: 'SESSION_LOCK_TIMEOUT' }); + }); + + it('rotates the opaque token when a password unlock succeeds', async () => { + const service = new SessionService(); + const created = await service.create('user-1', 'admin', 2); + await service.lock(created.token); + const result = await service.unlock(created.token); + expect(result.status).toBe('active'); + if (result.status === 'active' && 'token' in result) { + expect(result.token).not.toBe(created.token); + await expect(service.validate(created.token, false)).resolves.toEqual({ status: 'expired', code: 'SESSION_INVALID' }); + await expect(service.validate(result.token, false)).resolves.toEqual(expect.objectContaining({ status: 'active' })); + } + }); +}); diff --git a/api/src/auth/session.service.ts b/api/src/auth/session.service.ts new file mode 100644 index 0000000..ef41eed --- /dev/null +++ b/api/src/auth/session.service.ts @@ -0,0 +1,202 @@ +import { createHash, randomBytes } from 'node:crypto'; +import { Injectable, OnModuleDestroy, ServiceUnavailableException } from '@nestjs/common'; +import IORedis from 'ioredis'; + +export type SessionPortal = 'admin' | 'client'; + +export type AuthSessionRecord = { + userId: string; + portal: SessionPortal; + sessionVersion: number; + createdAt: number; + lastActivityAt: number; + lastAuthenticatedAt: number; + absoluteExpiresAt: number; + lockedAt?: number; +}; + +export type SessionValidationResult = + | { status: 'active'; record: AuthSessionRecord } + | { status: 'locked'; record: AuthSessionRecord; newlyLocked?: boolean } + | { status: 'expired'; code: 'SESSION_INVALID' | 'SESSION_ABSOLUTE_TIMEOUT' | 'SESSION_LOCK_TIMEOUT' }; + +const SESSION_PREFIX = 'cmpp:auth:session:'; +export const SESSION_COOKIE_NAME = '__Host-cmpp_session'; +export const DEVELOPMENT_SESSION_COOKIE_NAME = 'cmpp_session'; + +@Injectable() +export class SessionService implements OnModuleDestroy { + private redis?: IORedis; + + onModuleDestroy() { + this.redis?.disconnect(); + } + + async create(userId: string, portal: SessionPortal, sessionVersion: number) { + const now = Date.now(); + const token = randomBytes(32).toString('base64url'); + const record: AuthSessionRecord = { + userId, + portal, + sessionVersion, + createdAt: now, + lastActivityAt: now, + lastAuthenticatedAt: now, + absoluteExpiresAt: now + this.absoluteTimeoutMs, + }; + await this.write(token, record); + return { token, record }; + } + + async validate(token: string, markActivity: boolean): Promise { + const record = await this.read(token); + if (!record) return { status: 'expired', code: 'SESSION_INVALID' }; + + const now = Date.now(); + if (now >= record.absoluteExpiresAt) { + await this.remove(token); + return { status: 'expired', code: 'SESSION_ABSOLUTE_TIMEOUT' }; + } + if (record.lockedAt && now - record.lockedAt >= this.lockRecoveryMs) { + await this.remove(token); + return { status: 'expired', code: 'SESSION_LOCK_TIMEOUT' }; + } + if (!record.lockedAt && now - record.lastActivityAt >= this.idleTimeoutMs(record.portal)) { + record.lockedAt = now; + await this.write(token, record); + return { status: 'locked', record, newlyLocked: true }; + } + if (record.lockedAt) return { status: 'locked', record, newlyLocked: false }; + + if (markActivity && now - record.lastActivityAt >= 30_000) { + record.lastActivityAt = now; + await this.write(token, record); + } + return { status: 'active', record }; + } + + async lock(token: string) { + const record = await this.read(token); + if (!record) return null; + record.lockedAt = record.lockedAt ?? Date.now(); + await this.write(token, record); + return record; + } + + async touch(token: string) { + const result = await this.validate(token, false); + if (result.status !== 'active') return result; + result.record.lastActivityAt = Date.now(); + await this.write(token, result.record); + return result; + } + + async unlock(token: string) { + const result = await this.validate(token, false); + if (result.status === 'expired') return result; + const now = Date.now(); + const nextToken = randomBytes(32).toString('base64url'); + const record = { + ...result.record, + lockedAt: undefined, + lastActivityAt: now, + lastAuthenticatedAt: now, + }; + await this.write(nextToken, record); + await this.remove(token); + return { status: 'active' as const, token: nextToken, record }; + } + + async markReauthenticated(token: string) { + const result = await this.validate(token, false); + if (result.status !== 'active') return result; + result.record.lastAuthenticatedAt = Date.now(); + await this.write(token, result.record); + return result; + } + + remove(token: string) { + return this.client.del(this.key(token)); + } + + isRecentlyAuthenticated(record: AuthSessionRecord) { + return Date.now() - record.lastAuthenticatedAt < this.recentAuthenticationMs; + } + + publicSession(record: AuthSessionRecord) { + return { + idleTimeoutSeconds: Math.floor(this.idleTimeoutMs(record.portal) / 1000), + lockRecoverySeconds: Math.floor(this.lockRecoveryMs / 1000), + absoluteExpiresAt: new Date(record.absoluteExpiresAt).toISOString(), + lastActivityAt: new Date(record.lastActivityAt).toISOString(), + recentAuthenticationExpiresAt: new Date(record.lastAuthenticatedAt + this.recentAuthenticationMs).toISOString(), + }; + } + + get cookieSecure() { + return process.env.SESSION_COOKIE_SECURE === 'true' || (process.env.NODE_ENV === 'production' && process.env.SESSION_COOKIE_SECURE !== 'false'); + } + + get cookieName() { + return this.cookieSecure ? SESSION_COOKIE_NAME : DEVELOPMENT_SESSION_COOKIE_NAME; + } + + get absoluteTimeoutMs() { + return this.duration('SESSION_ABSOLUTE_TIMEOUT_MS', 12 * 60 * 60 * 1000); + } + + get lockRecoveryMs() { + return this.duration('SESSION_LOCK_RECOVERY_MS', 4 * 60 * 60 * 1000); + } + + get recentAuthenticationMs() { + return this.duration('SESSION_RECENT_AUTH_MS', 30 * 60 * 1000); + } + + idleTimeoutMs(portal: SessionPortal) { + return portal === 'admin' + ? this.duration('ADMIN_SESSION_IDLE_TIMEOUT_MS', 60 * 60 * 1000) + : this.duration('CLIENT_SESSION_IDLE_TIMEOUT_MS', 120 * 60 * 1000); + } + + private duration(name: string, fallback: number) { + const value = Number(process.env[name]); + return Number.isFinite(value) && value > 0 ? value : fallback; + } + + private async read(token: string): Promise { + try { + const value = await this.client.get(this.key(token)); + return value ? JSON.parse(value) as AuthSessionRecord : null; + } catch { + throw new ServiceUnavailableException('登录会话服务暂不可用'); + } + } + + private async write(token: string, record: AuthSessionRecord) { + const ttl = record.absoluteExpiresAt - Date.now(); + if (ttl <= 0) { + await this.remove(token); + return; + } + try { + await this.client.set(this.key(token), JSON.stringify(record), 'PX', ttl); + } catch { + throw new ServiceUnavailableException('登录会话服务暂不可用'); + } + } + + private key(token: string) { + return `${SESSION_PREFIX}${createHash('sha256').update(token).digest('hex')}`; + } + + private get client() { + if (!this.redis) { + this.redis = new IORedis(process.env.REDIS_URL ?? 'redis://127.0.0.1:6379', { + enableReadyCheck: true, + maxRetriesPerRequest: 1, + }); + } + return this.redis; + } +} diff --git a/api/src/billing/billing.controller.ts b/api/src/billing/billing.controller.ts index 90cf418..70f7af1 100644 --- a/api/src/billing/billing.controller.ts +++ b/api/src/billing/billing.controller.ts @@ -1,6 +1,7 @@ import { Body, Controller, Get, Post } from '@nestjs/common'; import { ApiTags } from '@nestjs/swagger'; import { TenantId } from '../common/tenant-id.decorator'; +import { RequireRecentAuthentication } from '../auth/require-recent-authentication.decorator'; import { BillingService, BillingActionDto, @@ -54,6 +55,7 @@ export class BillingController { } @Post('manual-recharges') + @RequireRecentAuthentication() createManualRecharge(@Body() body: CreateManualRechargeDto) { return this.billing.createManualRecharge(body); } @@ -84,11 +86,13 @@ export class BillingController { } @Post('refund') + @RequireRecentAuthentication() refund(@Body() body: BillingActionDto) { return this.billing.refund(body); } @Post('adjust') + @RequireRecentAuthentication() adjust(@Body() body: BillingActionDto) { return this.billing.adjust(body); } diff --git a/api/src/channels/channels.controller.ts b/api/src/channels/channels.controller.ts index ea92b54..0b6a932 100644 --- a/api/src/channels/channels.controller.ts +++ b/api/src/channels/channels.controller.ts @@ -1,5 +1,6 @@ import { Body, Controller, Delete, Get, Param, Post, Put, Query } from '@nestjs/common'; import { ApiTags } from '@nestjs/swagger'; +import { RequireRecentAuthentication } from '../auth/require-recent-authentication.decorator'; import { ChannelsService, ChangeChannelStatusDto, @@ -31,31 +32,37 @@ export class ChannelsController { } @Post('channels') + @RequireRecentAuthentication() createChannel(@Body() body: CreateChannelDto) { return this.channels.createChannel(body); } @Put('channels/:id') + @RequireRecentAuthentication() updateChannel(@Param('id') channelId: string, @Body() body: UpdateChannelDto) { return this.channels.updateChannel(channelId, body); } @Post('channels/:id/test') + @RequireRecentAuthentication() testChannel(@Param('id') channelId: string, @Body() body: TestChannelDto) { return this.channels.testChannel(channelId, body); } @Post('channels/:id/status') + @RequireRecentAuthentication() changeChannelStatus(@Param('id') channelId: string, @Body() body: ChangeChannelStatusDto) { return this.channels.changeChannelStatus(channelId, body); } @Post('channels/:id/copy') + @RequireRecentAuthentication() copyChannel(@Param('id') channelId: string, @Body() body: CopyChannelDto) { return this.channels.copyChannel(channelId, body); } @Delete('channels/:id') + @RequireRecentAuthentication() deleteChannel(@Param('id') channelId: string, @Body() body: ChangeChannelStatusDto) { return this.channels.deleteChannel(channelId, body); } @@ -96,21 +103,25 @@ export class ChannelsController { } @Post('channel-groups') + @RequireRecentAuthentication() createGroup(@Body() body: CreateChannelGroupDto) { return this.channels.createGroup(body); } @Put('channel-groups/:id') + @RequireRecentAuthentication() updateGroup(@Param('id') groupId: string, @Body() body: UpdateChannelGroupDto) { return this.channels.updateGroup(groupId, body); } @Delete('channel-groups/:id') + @RequireRecentAuthentication() deleteGroup(@Param('id') groupId: string) { return this.channels.deleteGroup(groupId); } @Post('channel-groups/items') + @RequireRecentAuthentication() addGroupItem(@Body() body: CreateChannelGroupItemDto) { return this.channels.addGroupItem(body); } @@ -121,6 +132,7 @@ export class ChannelsController { } @Post('channel-route-rules') + @RequireRecentAuthentication() createRouteRule(@Body() body: CreateRouteRuleDto) { return this.channels.createRouteRule(body); } @@ -156,6 +168,7 @@ export class ChannelsController { } @Post('report-tasks/status-change') + @RequireRecentAuthentication() changeReportTaskStatuses(@Body() body: ChangeReportTaskStatusesDto) { return this.channels.changeReportTaskStatuses(body); } diff --git a/api/src/operations/operations.service.spec.ts b/api/src/operations/operations.service.spec.ts index 479636b..9868751 100644 --- a/api/src/operations/operations.service.spec.ts +++ b/api/src/operations/operations.service.spec.ts @@ -396,9 +396,13 @@ describe('OperationsService', () => { prisma.cmppDownstreamDelivery.count = jest.fn() .mockResolvedValueOnce(12) .mockResolvedValueOnce(3) + .mockResolvedValueOnce(0) .mockResolvedValueOnce(8) .mockResolvedValueOnce(1) + .mockResolvedValueOnce(0) + .mockResolvedValueOnce(0) .mockResolvedValueOnce(1) + .mockResolvedValueOnce(0) .mockResolvedValueOnce(1) .mockResolvedValueOnce(2) .mockResolvedValueOnce(1) @@ -428,15 +432,19 @@ describe('OperationsService', () => { summary: { total: 12, pending: 3, + awaitingAck: 0, delivered: 8, failed: 1, + unconfirmed: 0, + rejected: 0, stalledPending: 1, + stalledAck: 0, recentFailed: 1, alertCount: 2, }, typeBreakdown: [ - { deliveryType: 'receipt', total: 9, pending: 2, delivered: 6, failed: 1 }, - { deliveryType: 'uplink', total: 3, pending: 1, delivered: 2, failed: 0 }, + { deliveryType: 'receipt', total: 9, pending: 2, awaitingAck: 0, delivered: 6, failed: 1, unconfirmed: 0, rejected: 0 }, + { deliveryType: 'uplink', total: 3, pending: 1, awaitingAck: 0, delivered: 2, failed: 0, unconfirmed: 0, rejected: 0 }, ], retryBuckets: [ { label: '0次', count: 2 }, @@ -444,8 +452,8 @@ describe('OperationsService', () => { { label: '4次及以上', count: 0 }, ], topApplications: [ - { applicationId: 'app-1', name: '应用A', pending: 2, failed: 1, delivered: 5, alertCount: 3 }, - { applicationId: 'app-2', name: '应用B', pending: 1, failed: 0, delivered: 3, alertCount: 1 }, + { applicationId: 'app-1', name: '应用A', pending: 2, awaitingAck: 0, failed: 1, unconfirmed: 0, rejected: 0, delivered: 5, alertCount: 3 }, + { applicationId: 'app-2', name: '应用B', pending: 1, awaitingAck: 0, failed: 0, unconfirmed: 0, rejected: 0, delivered: 3, alertCount: 1 }, ], }); }); diff --git a/api/src/operations/operations.service.ts b/api/src/operations/operations.service.ts index bf3ce84..6156aaf 100644 --- a/api/src/operations/operations.service.ts +++ b/api/src/operations/operations.service.ts @@ -408,11 +408,14 @@ export class OperationsService { const scopedWhere = downstreamDeliveryScopedWhere(query); const stalledPendingAt = new Date(Date.now() - downstreamAlertPendingMinutes() * 60_000); const recentFailedAt = new Date(Date.now() - downstreamAlertRecentFailedHours() * 60 * 60_000); - const [total, pending, delivered, failed, stalledPending, recentFailed, typeGroups, applicationGroups, retryZero, retryLow, retryHigh] = await Promise.all([ + const [total, pending, awaitingAck, delivered, failed, unconfirmed, rejected, stalledPending, stalledAck, recentFailed, typeGroups, applicationGroups, retryZero, retryLow, retryHigh] = await Promise.all([ this.prisma.cmppDownstreamDelivery.count({ where: scopedWhere }), this.prisma.cmppDownstreamDelivery.count({ where: { ...scopedWhere, status: 'pending' } }), + this.prisma.cmppDownstreamDelivery.count({ where: { ...scopedWhere, status: 'awaiting_ack' } }), this.prisma.cmppDownstreamDelivery.count({ where: { ...scopedWhere, status: 'delivered' } }), this.prisma.cmppDownstreamDelivery.count({ where: { ...scopedWhere, status: 'failed' } }), + this.prisma.cmppDownstreamDelivery.count({ where: { ...scopedWhere, status: 'unconfirmed' } }), + this.prisma.cmppDownstreamDelivery.count({ where: { ...scopedWhere, status: 'rejected' } }), this.prisma.cmppDownstreamDelivery.count({ where: { ...scopedWhere, @@ -420,10 +423,13 @@ export class OperationsService { createdAt: { lte: stalledPendingAt }, }, }), + this.prisma.cmppDownstreamDelivery.count({ + where: { ...scopedWhere, status: 'awaiting_ack', ackDeadlineAt: { lte: new Date() } }, + }), this.prisma.cmppDownstreamDelivery.count({ where: { ...scopedWhere, - status: 'failed', + status: { in: ['failed', 'unconfirmed', 'rejected'] }, updatedAt: { gte: recentFailedAt }, }, }), @@ -440,21 +446,21 @@ export class OperationsService { this.prisma.cmppDownstreamDelivery.count({ where: { ...scopedWhere, - status: { in: ['pending', 'failed'] }, + status: { in: ['pending', 'failed', 'unconfirmed', 'rejected'] }, retryCount: 0, }, }), this.prisma.cmppDownstreamDelivery.count({ where: { ...scopedWhere, - status: { in: ['pending', 'failed'] }, + status: { in: ['pending', 'failed', 'unconfirmed', 'rejected'] }, retryCount: { gte: 1, lte: 3 }, }, }), this.prisma.cmppDownstreamDelivery.count({ where: { ...scopedWhere, - status: { in: ['pending', 'failed'] }, + status: { in: ['pending', 'failed', 'unconfirmed', 'rejected'] }, retryCount: { gte: 4 }, }, }), @@ -474,18 +480,25 @@ export class OperationsService { summary: { total, pending, + awaitingAck, delivered, failed, + unconfirmed, + rejected, stalledPending, + stalledAck, recentFailed, - alertCount: stalledPending + recentFailed, + alertCount: stalledPending + stalledAck + recentFailed, }, typeBreakdown: ['receipt', 'uplink'].map((deliveryType) => ({ deliveryType, total: groupedByType[deliveryType]?.total ?? 0, pending: groupedByType[deliveryType]?.pending ?? 0, + awaitingAck: groupedByType[deliveryType]?.awaitingAck ?? 0, delivered: groupedByType[deliveryType]?.delivered ?? 0, failed: groupedByType[deliveryType]?.failed ?? 0, + unconfirmed: groupedByType[deliveryType]?.unconfirmed ?? 0, + rejected: groupedByType[deliveryType]?.rejected ?? 0, })), retryBuckets: [ { label: '0次', count: retryZero }, @@ -899,15 +912,21 @@ function summarizeMessageGroups(groups: Array<{ status: string; _count: { _all: function groupDownstreamByType( groups: Array<{ deliveryType: string; status: string; _count: { _all: number } }>, ) { - return groups.reduce>((accumulator, item) => { - const current = accumulator[item.deliveryType] ?? { total: 0, pending: 0, delivered: 0, failed: 0 }; + return groups.reduce>((accumulator, item) => { + const current = accumulator[item.deliveryType] ?? { total: 0, pending: 0, awaitingAck: 0, delivered: 0, failed: 0, unconfirmed: 0, rejected: 0 }; current.total += item._count._all; if (item.status === 'pending') { current.pending += item._count._all; + } else if (item.status === 'awaiting_ack') { + current.awaitingAck += item._count._all; } else if (item.status === 'delivered') { current.delivered += item._count._all; } else if (item.status === 'failed') { current.failed += item._count._all; + } else if (item.status === 'unconfirmed') { + current.unconfirmed += item._count._all; + } else if (item.status === 'rejected') { + current.rejected += item._count._all; } accumulator[item.deliveryType] = current; return accumulator; @@ -918,24 +937,33 @@ function groupDownstreamByApplication( groups: Array<{ applicationId: string; status: string; _count: { _all: number } }>, applicationMap: Map, ) { - const summaryMap = new Map(); + const summaryMap = new Map(); groups.forEach((item) => { const current = summaryMap.get(item.applicationId) ?? { applicationId: item.applicationId, name: applicationMap.get(item.applicationId) ?? item.applicationId, pending: 0, + awaitingAck: 0, failed: 0, + unconfirmed: 0, + rejected: 0, delivered: 0, alertCount: 0, }; if (item.status === 'pending') { current.pending += item._count._all; + } else if (item.status === 'awaiting_ack') { + current.awaitingAck += item._count._all; } else if (item.status === 'failed') { current.failed += item._count._all; + } else if (item.status === 'unconfirmed') { + current.unconfirmed += item._count._all; + } else if (item.status === 'rejected') { + current.rejected += item._count._all; } else if (item.status === 'delivered') { current.delivered += item._count._all; } - current.alertCount = current.pending + current.failed; + current.alertCount = current.pending + current.failed + current.unconfirmed + current.rejected; summaryMap.set(item.applicationId, current); }); return [...summaryMap.values()]; diff --git a/api/src/send-chain/gateway-events.controller.ts b/api/src/send-chain/gateway-events.controller.ts index 3cb98be..baedbf6 100644 --- a/api/src/send-chain/gateway-events.controller.ts +++ b/api/src/send-chain/gateway-events.controller.ts @@ -4,6 +4,9 @@ import { GatewayInboundAuthDto, GatewayInboundSubmitDto, GatewayPendingDeliveryQueryDto, + GatewayDownstreamAcknowledgedDto, + GatewayDownstreamFailureType, + GatewayDownstreamSentDto, GatewayDownstreamRecoveryStatusDto, GatewayReceiptEventDto, GatewaySubmitDeadLetterDto, @@ -66,9 +69,19 @@ export class GatewayEventsController { return this.sendChain.markDownstreamDeliveryDelivered(body.id); } + @Post('downstream/sent') + downstreamSent(@Body() body: GatewayDownstreamSentDto) { + return this.sendChain.markDownstreamDeliverySent(body); + } + + @Post('downstream/acknowledged') + downstreamAcknowledged(@Body() body: GatewayDownstreamAcknowledgedDto) { + return this.sendChain.acknowledgeDownstreamDelivery(body); + } + @Post('downstream/failed') - downstreamFailed(@Body() body: { id: string; errorMessage?: string }) { - return this.sendChain.markDownstreamDeliveryFailed(body.id, body.errorMessage); + downstreamFailed(@Body() body: { id: string; errorMessage?: string; failureType?: GatewayDownstreamFailureType }) { + return this.sendChain.markDownstreamDeliveryFailed(body.id, body.errorMessage, body.failureType); } @Post('downstream/recovery-status') diff --git a/api/src/send-chain/send-chain.service.spec.ts b/api/src/send-chain/send-chain.service.spec.ts index f11915d..b0a8a4a 100644 --- a/api/src/send-chain/send-chain.service.spec.ts +++ b/api/src/send-chain/send-chain.service.spec.ts @@ -213,6 +213,7 @@ function createPrismaMock() { }), findMany: jest.fn().mockResolvedValue([]), update: jest.fn().mockImplementation(({ data }) => Promise.resolve({ id: 'delivery-1', tenantId: 'tenant-1', applicationId: 'app-1', messageId: 'MSG-1', deliveryType: 'receipt', ...data })), + updateMany: jest.fn().mockResolvedValue({ count: 1 }), }, gatewaySubmitDeadLetter: { upsert: jest.fn().mockResolvedValue({ id: 'dead-1' }), @@ -1454,6 +1455,49 @@ describe('SendChainService', () => { }); }); + it('only marks downstream delivery delivered after a successful CMPP_DELIVER_RESP', async () => { + const { service, prisma } = createService(); + + await service.markDownstreamDeliverySent({ + id: 'delivery-1', + connectionId: 'conn-1', + sequenceId: '37', + messageId: '9016479179509871733', + sentAt: '2026-07-14T03:40:18.030Z', + ackDeadlineAt: '2026-07-14T03:40:48.030Z', + }); + expect(prisma.cmppDownstreamDelivery.updateMany).toHaveBeenLastCalledWith(expect.objectContaining({ + where: { id: 'delivery-1', status: { not: 'delivered' } }, + data: expect.objectContaining({ status: 'awaiting_ack', ackSequenceId: '37', connectionId: 'conn-1' }), + })); + + await service.acknowledgeDownstreamDelivery({ + id: 'delivery-1', + connectionId: 'conn-1', + sequenceId: '37', + messageId: '9016479179509871733', + result: 0, + acknowledgedAt: '2026-07-14T03:40:18.060Z', + }); + expect(prisma.cmppDownstreamDelivery.updateMany).toHaveBeenLastCalledWith(expect.objectContaining({ + data: expect.objectContaining({ status: 'delivered', ackResult: 0, deliveredAt: new Date('2026-07-14T03:40:18.060Z') }), + })); + }); + + it('does not automatically retry an unacknowledged delivery when its policy snapshot is disabled', async () => { + const { service, prisma } = createService(); + prisma.cmppDownstreamDelivery.findUnique.mockResolvedValue({ + id: 'delivery-1', tenantId: 'tenant-1', applicationId: 'app-1', messageId: 'MSG-1', + deliveryType: 'receipt', status: 'awaiting_ack', retryEnabled: false, retryCount: 0, + }); + + await service.markDownstreamDeliveryFailed('delivery-1', 'CMPP_DELIVER_RESP timeout', 'ack_timeout'); + + expect(prisma.cmppDownstreamDelivery.update).toHaveBeenCalledWith(expect.objectContaining({ + data: expect.objectContaining({ status: 'unconfirmed', retryCount: 1, nextRetryAt: null }), + })); + }); + it('uses exponential backoff for downstream delivery retries before final failure', async () => { const { service, prisma } = createService(); const previousBase = process.env.CMPP_DOWNSTREAM_RETRY_DELAY_MS; diff --git a/api/src/send-chain/send-chain.service.ts b/api/src/send-chain/send-chain.service.ts index ace2973..11fb5ed 100644 --- a/api/src/send-chain/send-chain.service.ts +++ b/api/src/send-chain/send-chain.service.ts @@ -105,6 +105,32 @@ export interface GatewayPendingDeliveryQueryDto { limit?: number; } +export interface GatewayDownstreamSentDto { + id: string; + connectionId?: string; + sequenceId?: string; + messageId?: string; + sentAt?: string; + ackDeadlineAt?: string; +} + +export interface GatewayDownstreamAcknowledgedDto extends GatewayDownstreamSentDto { + result: number; + acknowledgedAt?: string; +} + +export type GatewayDownstreamFailureType = 'send_failed' | 'ack_timeout' | 'ack_rejected' | 'connection_lost'; + +type GatewayControlDeliveryResult = { + sent?: boolean; + delivered?: boolean; + connectionId?: string; + sequenceId?: string; + messageId?: string; + sentAt?: string; + ackDeadlineAt?: string; +}; + export interface GatewaySubmitDeadLetterDto { streamMessageId: string; traceId?: string; @@ -910,6 +936,14 @@ export class SendChainService implements OnModuleInit, OnModuleDestroy { if (!application || application.status !== 'active' || application.tenant.status !== 'active') { throw new BadRequestException('CMPP account is invalid or disabled'); } + const expiredAcknowledgements = await this.prisma.cmppDownstreamDelivery.findMany({ + where: { applicationId: application.id, status: 'awaiting_ack', ackDeadlineAt: { lte: new Date() } }, + select: { id: true }, + take: 500, + }); + for (const expired of expiredAcknowledgements) { + await this.markDownstreamDeliveryFailed(expired.id, 'CMPP_DELIVER_RESP timeout recovered after Gateway restart', 'ack_timeout'); + } return this.prisma.cmppDownstreamDelivery.findMany({ where: { applicationId: application.id, @@ -932,19 +966,78 @@ export class SendChainService implements OnModuleInit, OnModuleDestroy { }); } - async markDownstreamDeliveryFailed(id: string, errorMessage?: string) { + async markDownstreamDeliverySent(data: GatewayDownstreamSentDto) { + const sentAt = asDateOrNull(data.sentAt) ?? new Date(); + const ackDeadlineAt = asDateOrNull(data.ackDeadlineAt) ?? new Date(sentAt.getTime() + downstreamAckTimeoutMs()); + await this.prisma.cmppDownstreamDelivery.updateMany({ + where: { id: data.id, status: { not: 'delivered' } }, + data: { + status: 'awaiting_ack', + sentAt, + ackDeadlineAt, + ackSequenceId: data.sequenceId, + ackMessageId: data.messageId, + connectionId: data.connectionId, + nextRetryAt: null, + lastError: null, + }, + }); + return this.prisma.cmppDownstreamDelivery.findUnique({ where: { id: data.id } }); + } + + async acknowledgeDownstreamDelivery(data: GatewayDownstreamAcknowledgedDto) { + const acknowledgedAt = asDateOrNull(data.acknowledgedAt) ?? new Date(); + if (data.result === 0) { + await this.prisma.cmppDownstreamDelivery.updateMany({ + where: { id: data.id, status: { not: 'delivered' } }, + data: { + status: 'delivered', + acknowledgedAt, + deliveredAt: acknowledgedAt, + ackDeadlineAt: null, + ackResult: data.result, + ackSequenceId: data.sequenceId, + ackMessageId: data.messageId, + connectionId: data.connectionId, + nextRetryAt: null, + lastError: null, + }, + }); + return this.prisma.cmppDownstreamDelivery.findUnique({ where: { id: data.id } }); + } + await this.prisma.cmppDownstreamDelivery.updateMany({ + where: { id: data.id, status: { not: 'delivered' } }, + data: { + acknowledgedAt, + ackResult: data.result, + ackSequenceId: data.sequenceId, + ackMessageId: data.messageId, + connectionId: data.connectionId, + }, + }); + return this.markDownstreamDeliveryFailed(data.id, `downstream CMPP_DELIVER_RESP result=${data.result}`, 'ack_rejected'); + } + + async markDownstreamDeliveryFailed(id: string, errorMessage?: string, failureType: GatewayDownstreamFailureType = 'send_failed') { const delivery = await this.prisma.cmppDownstreamDelivery.findUnique({ where: { id } }); if (!delivery) { throw new NotFoundException('Downstream delivery not found'); } + if (delivery.status === 'delivered') { + return delivery; + } const retryCount = (delivery.retryCount ?? 0) + 1; - const finalFailure = retryCount >= downstreamMaxRetries(); + const acknowledgementFailure = failureType === 'ack_timeout' || failureType === 'ack_rejected' || failureType === 'connection_lost'; + const retryAllowed = !acknowledgementFailure || delivery.retryEnabled !== false; + const finalFailure = !retryAllowed || retryCount >= downstreamMaxRetries(); + const finalStatus = failureType === 'ack_rejected' ? 'rejected' : acknowledgementFailure ? 'unconfirmed' : 'failed'; const updated = await this.prisma.cmppDownstreamDelivery.update({ where: { id }, data: { - status: finalFailure ? 'failed' : 'pending', + status: finalFailure ? finalStatus : 'pending', retryCount, nextRetryAt: finalFailure ? null : new Date(Date.now() + downstreamRetryDelayMs(retryCount)), + ackDeadlineAt: null, lastError: errorMessage ?? 'downstream delivery failed', }, }); @@ -960,6 +1053,8 @@ export class SendChainService implements OnModuleInit, OnModuleDestroy { applicationId: updated.applicationId, messageId: updated.messageId, retryCount, + failureType, + retryEnabled: updated.retryEnabled, errorMessage: updated.lastError, }, }, @@ -1168,12 +1263,16 @@ export class SendChainService implements OnModuleInit, OnModuleDestroy { account: String(payload.account ?? delivery.application?.cmppAccount ?? ''), ...payload, }; + await this.prisma.cmppDownstreamDelivery.update({ + where: { id: delivery.id }, + data: { status: 'pending', retryCount: 0, nextRetryAt: null, ackDeadlineAt: null, lastError: null }, + }); try { - const result = await this.postGatewayControl(path, requestPayload) as { delivered?: boolean }; - if (result.delivered) { - return this.markDownstreamDeliveryDelivered(delivery.id); + const result = await this.postGatewayControl(path, requestPayload) as GatewayControlDeliveryResult; + if (result.sent || result.delivered) { + return this.markDownstreamDeliverySent({ id: delivery.id, ...result }); } - return this.markDownstreamDeliveryFailed(delivery.id, 'downstream client is not connected'); + return this.prisma.cmppDownstreamDelivery.findUnique({ where: { id: delivery.id } }); } catch (error) { return this.markDownstreamDeliveryFailed( delivery.id, @@ -1326,7 +1425,7 @@ export class SendChainService implements OnModuleInit, OnModuleDestroy { } const application = await this.prisma.smsApplication.findUnique({ where: { id: data.applicationId }, - select: { cmppAccount: true }, + select: { cmppAccount: true, downstreamReceiptRetryEnabled: true, downstreamUplinkRetryEnabled: true }, }); const payload = { account: application?.cmppAccount, applicationId: data.applicationId, ...data.payload }; const delivery = await this.prisma.cmppDownstreamDelivery.create({ @@ -1337,6 +1436,9 @@ export class SendChainService implements OnModuleInit, OnModuleDestroy { messageId: data.messageId, deliveryType: data.deliveryType, payload, + retryEnabled: data.deliveryType === 'uplink' + ? application?.downstreamUplinkRetryEnabled ?? true + : application?.downstreamReceiptRetryEnabled ?? true, status: 'pending', }, }); @@ -1344,11 +1446,9 @@ export class SendChainService implements OnModuleInit, OnModuleDestroy { const result = await this.postGatewayControl( data.deliveryType === 'receipt' ? '/downstream/receipt' : '/downstream/uplink', { deliveryId: delivery.id, ...payload }, - ) as { delivered?: boolean }; - if (result.delivered) { - await this.markDownstreamDeliveryDelivered(delivery.id); - } else { - await this.markDownstreamDeliveryFailed(delivery.id, 'downstream client is not connected'); + ) as GatewayControlDeliveryResult; + if (result.sent || result.delivered) { + await this.markDownstreamDeliverySent({ id: delivery.id, ...result }); } } catch (error) { await this.markDownstreamDeliveryFailed(delivery.id, error instanceof Error ? error.message : 'Gateway control delivery failed'); @@ -2646,6 +2746,11 @@ function downstreamRetryDelayMs(retryCount = 1) { return Math.min(delay, max); } +function downstreamAckTimeoutMs() { + const configured = Number(process.env.CMPP_DOWNSTREAM_ACK_TIMEOUT_SECONDS ?? 30); + return Math.max(5, Number.isFinite(configured) ? configured : 30) * 1000; +} + function downstreamRetryBaseDelayMs() { const value = Number(process.env.CMPP_DOWNSTREAM_RETRY_DELAY_MS ?? DEFAULT_DOWNSTREAM_RETRY_DELAY_MS); return Number.isFinite(value) && value > 0 ? value : DEFAULT_DOWNSTREAM_RETRY_DELAY_MS; diff --git a/api/src/sms-config/admin-sms-config.controller.ts b/api/src/sms-config/admin-sms-config.controller.ts index 2c5e085..38d06a8 100644 --- a/api/src/sms-config/admin-sms-config.controller.ts +++ b/api/src/sms-config/admin-sms-config.controller.ts @@ -1,5 +1,6 @@ import { Body, Controller, Get, Param, Post, Put, Query } from '@nestjs/common'; import { ApiTags } from '@nestjs/swagger'; +import { RequireRecentAuthentication } from '../auth/require-recent-authentication.decorator'; import { CreateSmsApplicationDto, CreateSmsDrainageInfoDto, CreateSmsSignatureDto, CreateSmsTemplateDto, ReplaceApplicationRouteRulesDto, ReviewDto, SmsConfigService, StatusChangeDto, UpdateSmsApplicationDto, UpdateSmsDrainageInfoDto, UpdateSmsSignatureDto, UpdateSmsTemplateDto } from './sms-config.service'; @ApiTags('admin-sms-config') @@ -23,16 +24,19 @@ export class AdminSmsConfigController { } @Post('enterprise-applications') + @RequireRecentAuthentication() createApplication(@Body() body: CreateSmsApplicationDto) { return this.smsConfig.createApplication(body); } @Put('enterprise-applications/:id') + @RequireRecentAuthentication() updateApplication(@Param('id') applicationId: string, @Body() body: UpdateSmsApplicationDto) { return this.smsConfig.updateApplication(applicationId, body); } @Put('enterprise-applications/:id/route-rules') + @RequireRecentAuthentication() replaceApplicationRouteRules(@Param('id') applicationId: string, @Body() body: ReplaceApplicationRouteRulesDto) { return this.smsConfig.replaceApplicationRouteRules(applicationId, body); } @@ -78,16 +82,19 @@ export class AdminSmsConfigController { } @Post('drainage-infos/:id/approve') + @RequireRecentAuthentication() approveDrainageInfo(@Param('id') itemId: string, @Body() body: ReviewDto) { return this.smsConfig.approveDrainageInfo(itemId, body); } @Post('drainage-infos/:id/reject') + @RequireRecentAuthentication() rejectDrainageInfo(@Param('id') itemId: string, @Body() body: ReviewDto) { return this.smsConfig.rejectDrainageInfo(itemId, body); } @Post('drainage-infos/:id/status') + @RequireRecentAuthentication() changeDrainageInfoStatus(@Param('id') itemId: string, @Body() body: StatusChangeDto) { return this.smsConfig.changeDrainageInfoStatus(itemId, body); } @@ -113,36 +120,43 @@ export class AdminSmsConfigController { } @Post('signatures/:id/approve') + @RequireRecentAuthentication() approveSignature(@Param('id') signatureId: string, @Body() body: ReviewDto) { return this.smsConfig.approveSignature(signatureId, body); } @Post('signatures/:id/reject') + @RequireRecentAuthentication() rejectSignature(@Param('id') signatureId: string, @Body() body: ReviewDto) { return this.smsConfig.rejectSignature(signatureId, body); } @Post('templates/:id/approve') + @RequireRecentAuthentication() approveTemplate(@Param('id') templateId: string, @Body() body: ReviewDto) { return this.smsConfig.approveTemplate(templateId, body); } @Post('templates/:id/reject') + @RequireRecentAuthentication() rejectTemplate(@Param('id') templateId: string, @Body() body: ReviewDto) { return this.smsConfig.rejectTemplate(templateId, body); } @Post('enterprise-applications/:id/status') + @RequireRecentAuthentication() changeApplicationStatus(@Param('id') applicationId: string, @Body() body: StatusChangeDto) { return this.smsConfig.changeApplicationStatus(applicationId, body); } @Post('enterprise-signatures/:id/status') + @RequireRecentAuthentication() changeSignatureStatus(@Param('id') signatureId: string, @Body() body: StatusChangeDto) { return this.smsConfig.changeSignatureStatus(signatureId, body); } @Post('enterprise-templates/:id/status') + @RequireRecentAuthentication() changeTemplateStatus(@Param('id') templateId: string, @Body() body: StatusChangeDto) { return this.smsConfig.changeTemplateStatus(templateId, body); } diff --git a/api/src/sms-config/client-sms-config.controller.ts b/api/src/sms-config/client-sms-config.controller.ts index e1e2425..6da8a59 100644 --- a/api/src/sms-config/client-sms-config.controller.ts +++ b/api/src/sms-config/client-sms-config.controller.ts @@ -1,6 +1,7 @@ import { Body, Controller, Get, Param, Post, Put } from '@nestjs/common'; import { ApiTags } from '@nestjs/swagger'; import { TenantId } from '../common/tenant-id.decorator'; +import { RequireRecentAuthentication } from '../auth/require-recent-authentication.decorator'; import { CreateSignatureMaterialDto, CreateSmsApplicationDto, @@ -39,11 +40,13 @@ export class ClientSmsConfigController { } @Post('applications/:id/secret/reset') + @RequireRecentAuthentication() resetApplicationSecret(@Param('id') applicationId: string, @Body() body: StatusChangeDto) { return this.smsConfig.resetApplicationSecret(applicationId, body); } @Post('applications/:id/status') + @RequireRecentAuthentication() changeApplicationStatus(@Param('id') applicationId: string, @Body() body: StatusChangeDto) { return this.smsConfig.changeApplicationStatus(applicationId, body); } diff --git a/api/src/sms-config/sms-config.service.spec.ts b/api/src/sms-config/sms-config.service.spec.ts index d1574cb..dbb84e6 100644 --- a/api/src/sms-config/sms-config.service.spec.ts +++ b/api/src/sms-config/sms-config.service.spec.ts @@ -271,6 +271,8 @@ describe('SmsConfigService', () => { interfaceEnabled: false, interfaceType: 'cmpp20', queuePriority: 'priority', + downstreamReceiptRetryEnabled: true, + downstreamUplinkRetryEnabled: true, ipAllowlist: { create: [{ ipCidr: '10.0.0.1/32' }] }, }), })); diff --git a/api/src/sms-config/sms-config.service.ts b/api/src/sms-config/sms-config.service.ts index 04ebb6b..7d5e790 100644 --- a/api/src/sms-config/sms-config.service.ts +++ b/api/src/sms-config/sms-config.service.ts @@ -20,6 +20,8 @@ export interface CreateSmsApplicationDto { queuePriority?: string; maxPhonesPerTask?: number; templateMismatchMode?: string; + downstreamReceiptRetryEnabled?: boolean; + downstreamUplinkRetryEnabled?: boolean; ipAllowlist?: string[]; } @@ -314,6 +316,8 @@ export class SmsConfigService { queuePriority, maxPhonesPerTask: data.maxPhonesPerTask ?? 1000000, templateMismatchMode: data.templateMismatchMode ?? 'reject', + downstreamReceiptRetryEnabled: data.downstreamReceiptRetryEnabled ?? true, + downstreamUplinkRetryEnabled: data.downstreamUplinkRetryEnabled ?? true, ipAllowlist: { create: (data.ipAllowlist ?? []).map((ipCidr) => ({ ipCidr })), }, @@ -365,6 +369,8 @@ export class SmsConfigService { queuePriority, maxPhonesPerTask: data.maxPhonesPerTask, templateMismatchMode: data.templateMismatchMode, + downstreamReceiptRetryEnabled: data.downstreamReceiptRetryEnabled, + downstreamUplinkRetryEnabled: data.downstreamUplinkRetryEnabled, status: data.status, ipAllowlist: data.ipAllowlist ? { create: data.ipAllowlist.map((ipCidr) => ({ ipCidr })), diff --git a/api/src/tenants/tenants.controller.ts b/api/src/tenants/tenants.controller.ts index 64c5cb8..44df7b5 100644 --- a/api/src/tenants/tenants.controller.ts +++ b/api/src/tenants/tenants.controller.ts @@ -1,5 +1,6 @@ import { Body, Controller, Delete, Get, Param, Post, Put } from '@nestjs/common'; import { ApiTags } from '@nestjs/swagger'; +import { RequireRecentAuthentication } from '../auth/require-recent-authentication.decorator'; import { CreateTenantDto, TenantsService, UpdateTenantDto } from './tenants.service'; @ApiTags('tenants') @@ -23,21 +24,25 @@ export class TenantsController { } @Post() + @RequireRecentAuthentication() create(@Body() body: CreateTenantDto) { return this.tenants.create(body); } @Put(':id') + @RequireRecentAuthentication() update(@Param('id') id: string, @Body() body: UpdateTenantDto) { return this.tenants.update(id, body); } @Post(':id/status') + @RequireRecentAuthentication() changeStatus(@Param('id') id: string, @Body() body: { status: string }) { return this.tenants.changeStatus(id, body.status); } @Delete(':id') + @RequireRecentAuthentication() delete(@Param('id') id: string) { return this.tenants.delete(id); } diff --git a/api/src/users/users.controller.ts b/api/src/users/users.controller.ts index e499021..b71af9b 100644 --- a/api/src/users/users.controller.ts +++ b/api/src/users/users.controller.ts @@ -1,6 +1,7 @@ import { Body, Controller, Delete, Get, Param, Patch, Post, Put, Query } from '@nestjs/common'; import { ApiTags } from '@nestjs/swagger'; import { TenantId } from '../common/tenant-id.decorator'; +import { RequireRecentAuthentication } from '../auth/require-recent-authentication.decorator'; import { AssignPermissionDto, AssignRoleDto, @@ -24,31 +25,37 @@ export class UsersController { } @Post('admin/users') + @RequireRecentAuthentication() create(@Body() body: CreateUserDto) { return this.users.create(body); } @Put('admin/users/:id') + @RequireRecentAuthentication() update(@Param('id') id: string, @Body() body: UpdateUserDto) { return this.users.update(id, body); } @Patch('admin/users/:id') + @RequireRecentAuthentication() patch(@Param('id') id: string, @Body() body: UpdateUserDto) { return this.users.update(id, body); } @Post('admin/users/:id/status') + @RequireRecentAuthentication() changeStatus(@Param('id') id: string, @Body() body: ChangeUserStatusDto) { return this.users.changeStatus(id, body); } @Post('admin/users/:id/password') + @RequireRecentAuthentication() changePassword(@Param('id') id: string, @Body() body: ChangePasswordDto) { return this.users.changePassword(id, body); } @Delete('admin/users/:id') + @RequireRecentAuthentication() remove(@Param('id') id: string, @Body('operatorId') operatorId?: string) { return this.users.remove(id, operatorId); } @@ -59,26 +66,31 @@ export class UsersController { } @Post('client/users') + @RequireRecentAuthentication() createClient(@TenantId() tenantId: string | undefined, @Body() body: CreateUserDto) { return this.users.create({ ...body, roleCode: 'enterprise_admin' }, tenantId); } @Put('client/users/:id') + @RequireRecentAuthentication() updateClient(@TenantId() tenantId: string | undefined, @Param('id') id: string, @Body() body: UpdateUserDto) { return this.users.update(id, { ...body, roleCode: 'enterprise_admin' }, tenantId); } @Post('client/users/:id/status') + @RequireRecentAuthentication() changeClientStatus(@TenantId() tenantId: string | undefined, @Param('id') id: string, @Body() body: ChangeUserStatusDto) { return this.users.changeStatus(id, body, tenantId); } @Post('client/users/:id/password') + @RequireRecentAuthentication() changeClientPassword(@TenantId() tenantId: string | undefined, @Param('id') id: string, @Body() body: ChangePasswordDto) { return this.users.changePassword(id, body, tenantId); } @Delete('client/users/:id') + @RequireRecentAuthentication() removeClient(@TenantId() tenantId: string | undefined, @Param('id') id: string, @Body('operatorId') operatorId?: string) { return this.users.remove(id, operatorId, tenantId); } @@ -89,6 +101,7 @@ export class UsersController { } @Post('admin/users/roles') + @RequireRecentAuthentication() createRole(@Body() body: CreateRoleDto) { return this.users.createRole(body); } @@ -99,16 +112,19 @@ export class UsersController { } @Post('admin/users/permissions') + @RequireRecentAuthentication() createPermission(@Body() body: CreatePermissionDto) { return this.users.createPermission(body); } @Post('admin/users/roles/assign') + @RequireRecentAuthentication() assignRole(@Body() body: AssignRoleDto) { return this.users.assignRole(body); } @Post('admin/users/permissions/assign') + @RequireRecentAuthentication() assignPermission(@Body() body: AssignPermissionDto) { return this.users.assignPermission(body); } diff --git a/api/src/users/users.service.spec.ts b/api/src/users/users.service.spec.ts index 96284aa..5c6611f 100644 --- a/api/src/users/users.service.spec.ts +++ b/api/src/users/users.service.spec.ts @@ -84,4 +84,29 @@ describe('UsersService', () => { data: expect.objectContaining({ action: 'user.created', resource: 'user' }), })); }); + + it('revokes the affected user session when a role is assigned', async () => { + const tx = { + userRole: { upsert: jest.fn().mockResolvedValue({ userId: 'user-1', roleId: 'role-1' }) }, + user: { update: jest.fn().mockResolvedValue({ id: 'user-1' }) }, + }; + const prisma = { $transaction: jest.fn((callback) => callback(tx)) }; + const service = new UsersService(prisma as never); + await service.assignRole({ userId: 'user-1', roleId: 'role-1' }); + expect(tx.user.update).toHaveBeenCalledWith({ where: { id: 'user-1' }, data: { sessionVersion: { increment: 1 } } }); + }); + + it('revokes every affected user session when role permissions change', async () => { + const tx = { + rolePermission: { upsert: jest.fn().mockResolvedValue({ roleId: 'role-1', permissionId: 'permission-1' }) }, + user: { updateMany: jest.fn().mockResolvedValue({ count: 2 }) }, + }; + const prisma = { $transaction: jest.fn((callback) => callback(tx)) }; + const service = new UsersService(prisma as never); + await service.assignPermission({ roleId: 'role-1', permissionId: 'permission-1' }); + expect(tx.user.updateMany).toHaveBeenCalledWith({ + where: { roles: { some: { roleId: 'role-1' } } }, + data: { sessionVersion: { increment: 1 } }, + }); + }); }); diff --git a/api/src/users/users.service.ts b/api/src/users/users.service.ts index 331e356..93e53f2 100644 --- a/api/src/users/users.service.ts +++ b/api/src/users/users.service.ts @@ -128,6 +128,8 @@ export class UsersService { async update(id: string, data: UpdateUserDto, scopeTenantId?: string) { const current = await this.getExisting(id, scopeTenantId); const roleCode = data.roleCode ?? current.roles[0]?.role.code as UserRoleCode | undefined; + const roleChanged = data.roleCode !== undefined && data.roleCode !== current.roles[0]?.role.code; + const tenantChanged = data.tenantId !== undefined && data.tenantId !== current.tenantId; this.assertUserInput({ tenantId: data.tenantId ?? current.tenantId ?? undefined, username: data.username ?? current.username, @@ -153,7 +155,7 @@ export class UsersService { phone: data.phone === undefined ? undefined : normalizeOptional(data.phone), displayName: data.displayName ?? current.displayName, status: data.status ?? current.status, - ...(data.status === 'disabled' && current.status !== 'disabled' ? { sessionVersion: { increment: 1 } } : {}), + ...((data.status === 'disabled' && current.status !== 'disabled') || roleChanged || tenantChanged ? { sessionVersion: { increment: 1 } } : {}), }, include: { tenant: true, roles: { include: { role: true } } }, }); @@ -235,6 +237,17 @@ export class UsersService { return updated; } + async verifyCurrentPassword(id: string, password: string) { + if (!password) { + throw new BadRequestException('请输入当前密码'); + } + const current = await this.getExisting(id); + if (current.passwordHash !== hashPassword(password)) { + throw new BadRequestException('当前密码不正确'); + } + return current; + } + listRoles() { return this.prisma.role.findMany({ include: { permissions: { include: { permission: true } } }, @@ -262,18 +275,26 @@ export class UsersService { } assignRole(data: AssignRoleDto) { - return this.prisma.userRole.upsert({ - where: { userId_roleId: { userId: data.userId, roleId: data.roleId } }, - update: {}, - create: data, + return this.prisma.$transaction(async (tx) => { + const assignment = await tx.userRole.upsert({ + where: { userId_roleId: { userId: data.userId, roleId: data.roleId } }, + update: {}, + create: data, + }); + await tx.user.update({ where: { id: data.userId }, data: { sessionVersion: { increment: 1 } } }); + return assignment; }); } assignPermission(data: AssignPermissionDto) { - return this.prisma.rolePermission.upsert({ - where: { roleId_permissionId: { roleId: data.roleId, permissionId: data.permissionId } }, - update: {}, - create: data, + return this.prisma.$transaction(async (tx) => { + const assignment = await tx.rolePermission.upsert({ + where: { roleId_permissionId: { roleId: data.roleId, permissionId: data.permissionId } }, + update: {}, + create: data, + }); + await tx.user.updateMany({ where: { roles: { some: { roleId: data.roleId } } }, data: { sessionVersion: { increment: 1 } } }); + return assignment; }); } diff --git a/docs/first-version-development-requirements.md b/docs/first-version-development-requirements.md index 29ee214..189d6ed 100644 --- a/docs/first-version-development-requirements.md +++ b/docs/first-version-development-requirements.md @@ -278,6 +278,8 @@ - 已实现多 Gateway 恢复抢占协调第一版：恢复锁从单纯实例名升级为 Redis token 租约，状态记录 `lockOwner/lockExpiresAt`；恢复完成时必须通过 Lua 原子校验锁 token，只有持锁实例才能写入最终恢复状态并释放锁，避免旧实例超时后误删新实例锁或覆盖新实例恢复结果；运营端详情/列表可查看锁持有实例。 - 已实现长短信分片审计第一版：Gateway `SubmitResult` 回传真实 `segments[]`，包含 `segmentTotal/segmentIndex/sequenceId/gatewayMessageId/submitStatus/submittedAt`；NestJS 写入 Prisma/PostgreSQL `SmsMessageSegmentAudit`，回执按 `gatewayMessageId` 回填分片回执状态，补偿归因可记录 `compensationType`；运营端短信记录详情可查看真实分片提交、回执和补偿审计。 - 下游投递重试已改为指数退避第一版：首次失败后按基础间隔重试，随后按 2 倍递增，并受最大退避上限约束，避免客户长时间离线时平台每分钟机械重试。 +- 下游状态必须以客户确认作为终态：Gateway `SendPkt` 成功后只能写 `awaiting_ack`，仅收到匹配连接、`Sequence_Id`、`Msg_Id` 且 `CMPP_DELIVER_RESP.Result=0` 后才能写 `delivered`；超时、非零 Result 和历史未留存 ACK 的记录分别按未确认、拒绝或历史未确认展示，不能再把 TCP 写出冒充客户已收到。 +- 企业应用必须分别提供“回执自动重试投递”和“上行短信自动重试投递”开关，默认开启。开关按投递创建时快照保存；关闭只阻止已经写出但未获 ACK/被拒绝后的自动重发，不阻止离线队列在客户首次上线时完成首次投递。手工重投不受开关限制，但必须提示重复业务处理风险并二次确认。 - 已实现客户侧最终 Deliver 推送的第一版能力：Gateway 在下游 Submit 被接受后记录 messageId 到客户连接的内存映射；NestJS 收到最终 receipt/uplink 并入库后调用 Gateway `/downstream/receipt`、`/downstream/uplink`，Gateway 向仍在线的客户 CMPP 连接下发 Deliver Receipt 或普通 Deliver。 - 已实现客户侧 Deliver 持久化第一版能力：NestJS 收到最终 receipt/uplink 后写入 `CmppDownstreamDelivery` 待投递记录；在线推送成功标记 delivered，客户断线或 Gateway 不可达时保留 pending 并记录 retry 信息；客户重新 bind 后 Gateway 按账号拉取 pending 记录补发。 - 已实现普通上行匹配与人工认领第一版：优先按 messageId 精确匹配；无 messageId 时按接入号匹配应用路由；仍无唯一应用时按手机号和最近下发时间窗口匹配；多候选标记 ambiguous 并写入 `SmsUplinkMatchCandidate` 候选，运营端可人工认领候选应用/下发记录；认领后更新上行记录、保留候选审计，并创建真实客户侧上行 Deliver 投递记录。 @@ -1370,6 +1372,13 @@ - 两端登录均输入用户名、邮箱或手机号，外加密码和图形验证码。 - 登录接口必须调用真实 NestJS API，不允许前端静态用户、localStorage mock 或纯前端验证码作为验收依据。 - 运营端登录仅允许平台管理员；客户端登录仅允许已关联企业的企业管理员。 +- 浏览器认证凭据必须改为服务端随机生成的不可预测会话标识，通过 `HttpOnly`、`SameSite=Lax` Cookie 传输；前端 `localStorage` 只可保留非敏感展示信息，不得保存访问令牌。生产 Cookie 必须启用 `Secure`，因此正式部署本功能前必须先为页面和 API 配置 HTTPS。 +- 会话真实状态保存在 Redis，至少包含用户、入口、`sessionVersion`、创建时间、最后有效操作时间、最近密码认证时间、锁定时间和绝对到期时间。Redis 不可用或记录不存在时必须失败关闭，不得退回纯前端会话。 +- 运营端连续 60 分钟、客户端连续 120 分钟无用户操作后进入安全锁定，提前 5 分钟提醒。锁定后只需验证当前密码即可生成新会话标识并继续使用，不重复输入账号和图形验证码；锁定超过 4 小时必须完整登录。 +- 自动轮询、Dashboard 定时刷新、健康检查和后台标签页请求不得延长无操作期限。前端计时器只负责提醒和锁屏，NestJS/Redis 必须在每次受保护请求前独立执行空闲、锁定和绝对期限判断。 +- 单次登录绝对时长为 12 小时，无论是否持续操作均不得自动续期；到期必须使用账号、密码和图形验证码完整登录。修改密码、禁用/删除用户、角色变化和管理员强制下线必须通过 `sessionVersion` 和 Redis 会话立即撤销现有会话。 +- 用户、权限、企业状态、应用密钥、通道配置、路由、报备状态和资金调整等敏感操作要求最近 30 分钟内验证过当前密码。超时后由后端返回 `RECENT_AUTHENTICATION_REQUIRED`，前端验证当前密码后自动重试原操作；不能只依赖前端弹窗判断。 +- 主动退出、空闲锁定、密码解锁、敏感操作再认证和会话创建均需写真实系统日志；多标签页使用浏览器消息同步锁定、解锁和退出。普通网络错误、400、403 业务拒绝或 5xx 不得被误判为自动退出。 ### 2. 用户类型和企业关联 diff --git a/docs/production-deployment.md b/docs/production-deployment.md index c774879..fade90c 100644 --- a/docs/production-deployment.md +++ b/docs/production-deployment.md @@ -29,6 +29,13 @@ REPO_URL=http://175.27.255.91:3000/hectorzhao/lislgosms.git BRANCH=main PUBLIC_HTTP_PORT=12026 API_PORT=3000 +ADMIN_SESSION_IDLE_TIMEOUT_MS=3600000 +CLIENT_SESSION_IDLE_TIMEOUT_MS=7200000 +SESSION_LOCK_RECOVERY_MS=14400000 +SESSION_ABSOLUTE_TIMEOUT_MS=43200000 +SESSION_RECENT_AUTH_MS=1800000 +SESSION_COOKIE_SECURE=true +CMPP_DOWNSTREAM_ACK_TIMEOUT_SECONDS=30 GATEWAY_CMPP_ADDR=0.0.0.0:17890 OBJECT_STORAGE_DRIVER=minio OBJECT_STORAGE_LOCAL_ROOT=/var/lib/cmpp-platform/object-storage @@ -37,6 +44,8 @@ PROD_ADMIN_USERNAME=prod_admin PROD_ADMIN_PASSWORD='change-me' ``` +安全会话使用 HttpOnly Cookie，正式生产必须先为页面和 `/api` 配置 HTTPS，并保持 `SESSION_COOKIE_SECURE=true`。仅在用户明确授权的 HTTP 生产验证环境中，允许临时显式设置 `SESSION_COOKIE_SECURE=false` 维持验证可用性；该例外必须记录在发布验收中，不能替代正式环境 TLS。 + 脚本会安装 Node.js、Go、PostgreSQL、Redis、MinIO、Nginx，创建 systemd 服务，执行 Prisma migrate，构建前端/API/Gateway，并创建平台管理员。Node.js、Go 和 MinIO 下载会按服务器架构自动选择 x64/amd64 或 arm64。 如生产验证服务器临时无法稳定下载 MinIO，可显式传入 `OBJECT_STORAGE_DRIVER=local`，文件会通过真实 API 保存到服务器本地目录 `OBJECT_STORAGE_LOCAL_ROOT`，`cmpp-minio` 服务会跳过安装和启动。该模式只建议用于验证环境；正式生产建议恢复 `OBJECT_STORAGE_DRIVER=minio`。 diff --git a/docs/system-functional-test-cases.md b/docs/system-functional-test-cases.md index 1f3ccc3..2302e1b 100644 --- a/docs/system-functional-test-cases.md +++ b/docs/system-functional-test-cases.md @@ -3251,6 +3251,12 @@ npm run verify:phase8 | TC-AUTH-004 | 使用平台管理员或未关联企业的用户登录 `/client/login`。 | 登录失败；不进入客户端。 | | TC-AUTH-005 | 同一用户连续输错密码 5 次。 | 第 5 次后用户锁定 24 小时；锁定期内正确密码也被拒绝；系统记录失败次数和锁定时间。 | | TC-AUTH-006 | 输入错误或过期图形验证码登录。 | 返回 400 可读错误；必须刷新验证码后重试。 | +| TC-AUTH-007 | 使用运营管理员登录，确认响应、浏览器存储和 Redis；不携带 Cookie 访问运营 API。 | 登录响应和 localStorage 不含访问令牌；浏览器仅持有 HttpOnly 会话 Cookie；Redis 存在哈希会话记录；无 Cookie 请求返回 `401/SESSION_INVALID`。 | +| TC-AUTH-008 | 将运营端无操作参数缩短后等待超时，同时保持 Dashboard 自动轮询；再输入当前密码解锁。 | 自动轮询不续期；服务端返回 `401/SESSION_LOCKED`；页面锁屏但无需账号和验证码；当前密码正确时轮换 Session ID，旧标识失效，多标签同步解锁。 | +| TC-AUTH-009 | 将客户端无操作参数缩短并验证其阈值独立于运营端；锁定后超过锁定恢复期限。 | 客户端使用独立 120 分钟默认阈值；超过 4 小时恢复期限后密码快速解锁被拒绝，必须完整登录。 | +| TC-AUTH-010 | 持续操作至绝对期限，将默认 12 小时在测试环境缩短验证。 | 用户活动只能刷新空闲时间，不能延长绝对期限；到期返回 `401/SESSION_ABSOLUTE_TIMEOUT`，必须重新输入账号、密码和验证码。 | +| TC-AUTH-011 | 登录超过最近认证窗口后执行用户禁用、手工充值、通道修改、路由修改或报备状态修改。 | 后端先返回 `403/RECENT_AUTHENTICATION_REQUIRED`，输入当前密码后 30 分钟内自动重试；错误密码不执行原操作，数据库无副作用。 | +| TC-AUTH-012 | 分别执行主动退出、修改密码、禁用、删除和角色变更，并在另一标签页继续请求。 | Redis 会话删除或 `sessionVersion` 失效；所有标签页同步退出；旧 Cookie 均返回 401；系统日志可查询创建、锁定、解锁、再认证和退出事件。 | | TC-USER-ADMIN-001 | 运营端新增平台管理员，填写用户名/登录账号、邮箱或手机号、初始密码。 | 创建成功；用户无 `tenantId`；可登录运营端；写 `user.created` 日志。 | | TC-USER-ADMIN-002 | 运营端新增企业管理员但不选择企业。 | 返回 400；不创建用户。 | | TC-USER-ADMIN-003 | 运营端新增企业管理员并选择企业。 | 创建成功；用户关联企业；可登录客户端；客户端数据按该企业隔离。 | @@ -3276,3 +3282,12 @@ npm run verify:phase8 | TC-MOCK-CLEAN-006 | 运营端短信记录按手机号、状态、日期和内容查询，打开详情。 | 数据来自 `sms_message_records`；详情展示真实 messageId、状态、失败原因；无数据时为空态。 | | TC-MOCK-CLEAN-007 | 访问明确标注待开发的彩信菜单。 | 可以显示待开发/空态；不得作为第一版短信真实功能通过依据。 | | TC-PHONE-SEGMENT-001 | 生产库存在 50 万级手机号段时打开手机号段库，连续点击下一页、上一页，并按号段、省份、城市或运营商搜索。 | API 使用 `prefix` 游标分页并返回 `hasMore/nextCursor`；页面数据来自真实数据库，可稳定前后翻页和搜索；接口不执行全表总数统计，页面不展示号段总条数。 | + +### 17.11 下游投递 ACK 与应用级重试策略 + +| 用例编号 | 操作 | 预期结果 | +| --- | --- | --- | +| TC-GW-ACK-001 | 客户在线时分别触发一条状态回执和一条上行短信，Gateway `SendPkt` 成功后延迟返回 `CMPP_DELIVER_RESP`。 | 写出后 `CmppDownstreamDelivery.status=awaiting_ack`；只有匹配连接、Sequence_Id、Msg_Id 且 Result=0 后才变为 `delivered`，并保存写出时间、确认时间、ACK 字段和连接 ID。 | +| TC-GW-ACK-002 | 分别返回非零 Result、不返回响应直到超时、重启 Gateway 后让旧 `awaiting_ack` 超时。 | 非零 Result 和超时不会误记为 delivered；Gateway 重启后 NestJS 能恢复过期 ACK，按策略进入退避重试或最终 `rejected/unconfirmed`。 | +| TC-GW-ACK-003 | 在企业应用中分别关闭“回执自动重试”和“上行短信自动重试”，各制造一次 ACK 超时，再重新开启并创建新投递。 | 关闭只影响对应类型的新投递策略快照；离线后的首次投递仍会在重连时执行；已写出未确认的记录不自动重发；重新开启后新记录按退避策略重试。 | +| TC-GW-ACK-004 | 对 `unconfirmed/rejected/failed` 记录执行单条和批量手工重投。 | 页面提示重复处理风险并二次确认；真实调用 Gateway；`awaiting_ack/delivered` 不允许重投；重发复用同一业务 Msg_Id。 | diff --git a/docs/testing-progress.md b/docs/testing-progress.md index 875ceb1..47ac731 100644 --- a/docs/testing-progress.md +++ b/docs/testing-progress.md @@ -1708,3 +1708,19 @@ git diff --check - 新增真实故障形态回归用例：完整括号签名、审核通过、全局 reporting 时可进入模板不匹配人工审核聚合，并验证查询不再携带全局报备条件、不产生签名失败回执。定向 API 测试 1 suite、41 项通过；API 全量 13 suites、142 项通过，API build、前端 build、Gateway 全量 Go 测试和 `git diff --check` 通过，前端仅有既有 Vite chunk size warning。 - 功能提交 `f08a73e1` 已 push 并部署生产。部署前 PostgreSQL 备份为 `/opt/cmpp-platform/backups/cmpp-20260714-112012.sql`（约 64MB），运行源码备份为 `/opt/cmpp-platform/backups/source-20260714-112012.tar.gz`（约 15MB）；发布包本地与服务器 SHA-256 均为 `38b3fa6282a06648dabc048fbebca7a6579a6a7ef1f8eee514ecf14fb9d69790`。 - 生产 36 条 migration 无待执行项，`.deployed-commit=f08a73e19129f5249a5a9e0035c7ab63b80422d4`；`cmpp-api`、`cmpp-gateway`、Nginx、MinIO 均 active，`12026/17890/8090/3000` 监听，API/Gateway health 和外部首页、运营端登录页 HTTP 200。线上源码确认按完整括号 `match[0]` 查询且不带全局 `reportStatus` 条件，部署后 API/Gateway 近期日志无 error。未擅自向客户号码重发短信；新提交将按应用现有模板不匹配人工审核策略进入真实审核与后续通道路由。 + +## 2026-07-14 服务端安全会话与自动锁定（未提交、未部署） + +- 将可预测的 `dev-token:userId:sessionVersion` 和 localStorage 访问令牌替换为 256 位随机会话标识；浏览器只通过 HttpOnly、SameSite Cookie 携带，Redis 使用会话标识 SHA-256 键保存真实状态。生产模式 Cookie 默认 `Secure`，当前生产仍为 HTTP，因此本功能在完成 HTTPS 配置前不得部署。 +- 运营端/客户端无操作阈值分别为 60/120 分钟，提前 5 分钟提醒；超时进入密码锁屏，4 小时内可用当前密码解锁并轮换会话标识，超过后完整登录。绝对会话时长 12 小时不可滑动续期；敏感操作最近密码认证窗口为 30 分钟。 +- NestJS 中间件对运营端和客户端受保护 API 强制要求 Redis 会话，逐次校验用户状态和 `sessionVersion`；Gateway 回调和 health 保持原内部链路，不被浏览器会话门禁拦截。自动轮询只有检测到近期真实浏览器操作时才携带活动标识，不能长期保活无人值守会话。 +- 用户/权限、企业状态、应用密钥、通道和路由、报备状态、手工充值/退款/调整已接后端最近认证 Guard。前端收到 `RECENT_AUTHENTICATION_REQUIRED` 后要求当前密码，成功后自动重试；普通 JSON、Blob 和文件上传统一处理会话 401。会话创建、锁定、解锁、再认证和退出写 `OperationLog`，多标签页同步状态。 +- API 自动化测试当前 15 suites、151 项通过；新增 Redis 会话空闲锁定、客户端独立阈值、4 小时恢复期限、解锁轮换旧标识失效、`sessionVersion`/角色权限变更撤销和敏感操作 Guard 用例。真实本地 PostgreSQL + Redis + NestJS 短阈值验收通过：登录响应无访问令牌且收到 Cookie、无 Cookie 访问返回 `401/SESSION_INVALID`、空闲后返回 `401/SESSION_LOCKED`、密码解锁后恢复查询、认证窗口过期后敏感操作返回 `403/RECENT_AUTHENTICATION_REQUIRED`、重新认证和登出成功；临时用户与会话已清理。本轮按要求不提交、不推送、不部署。 + +## 2026-07-14 下游 CMPP_DELIVER_RESP 精确确认与双重试开关 + +- 生产只读排查 11:40 两条余额失败短信确认：NestJS 已生成 `REJECTD` 回执，Gateway 对同一 CMPP2.0 会话写出后分别收到 Sequence_Id 37/38 的 `CMPP_DELIVER_RESP`。原 `CmppDownstreamDelivery.status=delivered` 仅代表 `SendPkt` 成功，无法证明客户确认，属于观测口径缺陷。 +- Gateway 新增按连接和 Sequence_Id 跟踪投递，记录实际下发 Msg_Id；写出后回调 `downstream/sent`，收到 CMPP2.0/2.1/3.0 `DELIVER_RESP` 后校验 Sequence_Id、Msg_Id 和 Result，再回调 `downstream/acknowledged`。ACK 超时回调真实失败类型；重启后 NestJS 在客户恢复拉取 pending 时补偿处理过期 `awaiting_ack`。 +- Prisma `SmsApplication` 新增回执、上行两个自动重试开关，默认开启；`CmppDownstreamDelivery` 保存策略快照、写出/确认/截止时间、ACK Result/Sequence_Id/Msg_Id 和连接 ID。关闭开关后首次投递仍保留，已写出未确认或被拒绝的对应类型不自动重发；手工重投保留并增加重复处理二次确认。 +- 运营端企业应用编辑页增加两个独立开关；下游投递页区分待首次投递、等待客户端确认、客户端已确认、未确认、拒绝和最终失败，Dashboard 和告警改为真实 ACK 口径。历史 7 条仅确认写出的记录迁移为 `unconfirmed`，不伪造 ACK。 +- 新增 API ACK 状态与关闭重试策略单测、Gateway 真实 CMPP 会话 ACK 回调/超时单测；同步更新需求和 TC-GW-ACK-001～004。API 全量 15 suites、153 项通过，Gateway 全量 Go 测试、Prisma validate 与本地真实 PostgreSQL migration、API build、前端 build 和 `git diff --check` 通过；前端仅有既有 chunk size warning。应用内浏览器确认本地构建标题、登录 DOM、无框架覆盖及 console 无 error/warn；因真实图形验证码未获授权代解，受保护页面内的两个开关点击验收未执行。待提交并部署后补充最终提交、备份、migration 和生产验收结果。 diff --git a/gateway/internal/control/server.go b/gateway/internal/control/server.go index 48d8a1e..7c8d44a 100644 --- a/gateway/internal/control/server.go +++ b/gateway/internal/control/server.go @@ -139,13 +139,13 @@ func (s Server) handleDownstreamReceipt(w http.ResponseWriter, r *http.Request) http.Error(w, fmt.Sprintf("invalid downstream receipt: %v", err), http.StatusBadRequest) return } - delivered, err := inbound.PushReceipt(event) + result, err := inbound.PushReceiptWithResult(event) if err != nil { http.Error(w, err.Error(), http.StatusBadRequest) return } w.Header().Set("Content-Type", "application/json") - _ = json.NewEncoder(w).Encode(map[string]any{"delivered": delivered}) + _ = json.NewEncoder(w).Encode(result) } func (s Server) handleDownstreamUplink(w http.ResponseWriter, r *http.Request) { @@ -158,13 +158,13 @@ func (s Server) handleDownstreamUplink(w http.ResponseWriter, r *http.Request) { http.Error(w, fmt.Sprintf("invalid downstream uplink: %v", err), http.StatusBadRequest) return } - delivered, err := inbound.PushUplink(event) + result, err := inbound.PushUplinkWithResult(event) if err != nil { http.Error(w, err.Error(), http.StatusBadRequest) return } w.Header().Set("Content-Type", "application/json") - _ = json.NewEncoder(w).Encode(map[string]any{"delivered": delivered}) + _ = json.NewEncoder(w).Encode(result) } func (s Server) handleDownstreamRecoveryCandidates(w http.ResponseWriter, r *http.Request) { diff --git a/gateway/internal/inbound/server.go b/gateway/internal/inbound/server.go index 5b009c0..9d47824 100644 --- a/gateway/internal/inbound/server.go +++ b/gateway/internal/inbound/server.go @@ -11,6 +11,8 @@ import ( "log" "net" "net/http" + "os" + "strconv" "strings" "sync" "time" @@ -20,6 +22,7 @@ import ( ) const defaultHTTPTimeout = 10 * time.Second +const defaultDownstreamAckTimeout = 30 * time.Second type Server struct { Addr string @@ -86,15 +89,46 @@ type DownstreamUplink struct { ReceivedAt string `json:"receivedAt,omitempty"` } +type DownstreamSendResult struct { + Sent bool `json:"sent"` + ConnectionID string `json:"connectionId,omitempty"` + SequenceID string `json:"sequenceId,omitempty"` + MessageID string `json:"messageId,omitempty"` + SentAt string `json:"sentAt,omitempty"` + AckDeadlineAt string `json:"ackDeadlineAt,omitempty"` +} + +type downstreamDeliveryLifecycleEvent struct { + Kind string + DeliveryID string + ConnectionID string + SequenceID uint32 + MessageID uint64 + Result uint32 + ObservedAt time.Time + AckDeadlineAt time.Time + FailureType string + ErrorMessage string +} + +type downstreamAckTracker struct { + deliveryID string + connectionID string + sequenceID uint32 + messageID uint64 + session *downstreamSession + timer *time.Timer +} + type downstreamConnectionEvent struct { - Account string `json:"account"` - ConnectionID string `json:"connectionId"` - Status string `json:"status"` - RemoteIP string `json:"remoteIp,omitempty"` - Protocol string `json:"protocol,omitempty"` - ConnectedAt string `json:"connectedAt,omitempty"` - ObservedAt string `json:"observedAt,omitempty"` - ErrorMessage string `json:"errorMessage,omitempty"` + Account string `json:"account"` + ConnectionID string `json:"connectionId"` + Status string `json:"status"` + RemoteIP string `json:"remoteIp,omitempty"` + Protocol string `json:"protocol,omitempty"` + ConnectedAt string `json:"connectedAt,omitempty"` + ObservedAt string `json:"observedAt,omitempty"` + ErrorMessage string `json:"errorMessage,omitempty"` } type downstreamSession struct { @@ -113,6 +147,7 @@ type downstreamSession struct { presence PresenceStore instanceID string report func(*downstreamSession, string, string) + deliveryReport func(downstreamDeliveryLifecycleEvent) } var downstreamRegistry = struct { @@ -126,6 +161,11 @@ var downstreamRegistry = struct { byConn: make(map[*cmpp.Conn]*downstreamSession), } +var downstreamAckRegistry = struct { + sync.Mutex + items map[string]*downstreamAckTracker +}{items: make(map[string]*downstreamAckTracker)} + func (s Server) ListenAndServe() error { addr := s.Addr if addr == "" { @@ -176,6 +216,7 @@ func (s Server) handleLogin(response *cmpp.Response, packet *cmpp.Packet, logger presence: s.PresenceStore, instanceID: s.gatewayInstanceID(), report: s.reportConnection, + deliveryReport: s.reportDownstreamDelivery, } rememberAccount(session) go s.reportConnection(&session, "connected", "") @@ -273,6 +314,7 @@ func (s Server) handleSubmit(response *cmpp.Response, packet *cmpp.Packet, logge presence: s.PresenceStore, instanceID: s.gatewayInstanceID(), report: session.report, + deliveryReport: session.deliveryReport, }) if current := findSessionByConn(packet.Conn); current != nil && current.report != nil { go current.report(current, "submit", "") @@ -289,14 +331,20 @@ func (s Server) handleSubmit(response *cmpp.Response, packet *cmpp.Packet, logge return false, nil } -func (s Server) handleActivity(_ *cmpp.Response, packet *cmpp.Packet, _ *log.Logger) (bool, error) { +func (s Server) handleActivity(_ *cmpp.Response, packet *cmpp.Packet, logger *log.Logger) (bool, error) { session := findSessionByConn(packet.Conn) - if session == nil || session.report == nil { + if session == nil { return true, nil } - switch packet.Packer.(type) { + switch response := packet.Packer.(type) { case *cmpp.CmppActiveTestReqPkt, *cmpp.CmppActiveTestRspPkt: - go session.report(session, "heartbeat", "") + if session.report != nil { + go session.report(session, "heartbeat", "") + } + case *cmpp.Cmpp2DeliverRspPkt: + handleDownstreamAcknowledgement(packet.Conn, response.SeqId, response.MsgId, uint32(response.Result), logger) + case *cmpp.Cmpp3DeliverRspPkt: + handleDownstreamAcknowledgement(packet.Conn, response.SeqId, response.MsgId, response.Result, logger) } return true, nil } @@ -316,6 +364,31 @@ func (s Server) reportConnection(session *downstreamSession, status string, erro } } +func (s Server) reportDownstreamDelivery(event downstreamDeliveryLifecycleEvent) { + if strings.TrimSpace(event.DeliveryID) == "" { + return + } + payload := map[string]any{ + "id": event.DeliveryID, "connectionId": event.ConnectionID, + "sequenceId": strconv.FormatUint(uint64(event.SequenceID), 10), + "messageId": strconv.FormatUint(event.MessageID, 10), + } + switch event.Kind { + case "sent": + payload["sentAt"] = formatRFC3339Nano(event.ObservedAt) + payload["ackDeadlineAt"] = formatRFC3339Nano(event.AckDeadlineAt) + _ = s.post(context.Background(), "/gateway/events/downstream/sent", payload, nil) + case "acknowledged": + payload["result"] = event.Result + payload["acknowledgedAt"] = formatRFC3339Nano(event.ObservedAt) + _ = s.post(context.Background(), "/gateway/events/downstream/acknowledged", payload, nil) + case "failed": + payload["failureType"] = event.FailureType + payload["errorMessage"] = event.ErrorMessage + _ = s.post(context.Background(), "/gateway/events/downstream/failed", payload, nil) + } +} + type inboundSubmitPacket struct { protocol string pkTotal uint8 @@ -444,19 +517,19 @@ func (s Server) flushPending(account string, logger *log.Logger) (pendingFlushRe } result.Deliveries = len(deliveries) for _, delivery := range deliveries { - delivered, err := s.pushPendingDelivery(account, delivery) + sendResult, err := s.pushPendingDelivery(account, delivery) if err != nil { result.FailedCount++ result.LastError = err.Error() _ = s.post(context.Background(), "/gateway/events/downstream/failed", map[string]string{ "id": delivery.ID, "errorMessage": err.Error(), + "failureType": "send_failed", }, nil) continue } - if delivered { + if sendResult.Sent { result.DeliveredCount++ - _ = s.post(context.Background(), "/gateway/events/downstream/delivered", map[string]string{"id": delivery.ID}, nil) continue } result.WaitingCount++ @@ -464,26 +537,26 @@ func (s Server) flushPending(account string, logger *log.Logger) (pendingFlushRe return result, nil } -func (s Server) pushPendingDelivery(account string, delivery pendingDelivery) (bool, error) { +func (s Server) pushPendingDelivery(account string, delivery pendingDelivery) (DownstreamSendResult, error) { switch delivery.DeliveryType { case "receipt": var event DownstreamReceipt if err := json.Unmarshal(delivery.Payload, &event); err != nil { - return false, err + return DownstreamSendResult{}, err } event.DeliveryID = delivery.ID event.Account = defaultString(event.Account, account) - return PushReceipt(event) + return PushReceiptWithResult(event) case "uplink": var event DownstreamUplink if err := json.Unmarshal(delivery.Payload, &event); err != nil { - return false, err + return DownstreamSendResult{}, err } event.DeliveryID = delivery.ID event.Account = defaultString(event.Account, account) - return PushUplink(event) + return PushUplinkWithResult(event) default: - return false, fmt.Errorf("unsupported downstream delivery type %q", delivery.DeliveryType) + return DownstreamSendResult{}, fmt.Errorf("unsupported downstream delivery type %q", delivery.DeliveryType) } } @@ -770,9 +843,14 @@ func onlineAccounts() []string { } func PushReceipt(event DownstreamReceipt) (bool, error) { + result, err := PushReceiptWithResult(event) + return result.Sent, err +} + +func PushReceiptWithResult(event DownstreamReceipt) (DownstreamSendResult, error) { session := findSession(event.MessageID, event.Account) if session == nil { - return false, nil + return DownstreamSendResult{}, nil } stat := strings.TrimSpace(event.RawStatus) if stat == "" { @@ -794,20 +872,25 @@ func PushReceipt(event DownstreamReceipt) (bool, error) { } receiptBytes, err := receipt.Pack() if err != nil { - return false, err + return DownstreamSendResult{}, err } deliver := downstreamDeliverPacket(session, session.gatewayMsgID, session.srcID, defaultString(event.PhoneNumber, session.phoneNumber), 0, 1, string(receiptBytes)) - return sendDownstream(session, deliver) + return sendDownstream(session, deliver, event.DeliveryID) } func PushUplink(event DownstreamUplink) (bool, error) { + result, err := PushUplinkWithResult(event) + return result.Sent, err +} + +func PushUplinkWithResult(event DownstreamUplink) (DownstreamSendResult, error) { session := findSession(event.MessageID, event.Account) if session == nil { - return false, nil + return DownstreamSendResult{}, nil } content, err := cmpputils.Utf8ToUcs2(event.Content) if err != nil { - return false, err + return DownstreamSendResult{}, err } deliver := downstreamDeliverPacket( session, @@ -818,7 +901,7 @@ func PushUplink(event DownstreamUplink) (bool, error) { 0, content, ) - return sendDownstream(session, deliver) + return sendDownstream(session, deliver, event.DeliveryID) } func downstreamDeliverPacket(session *downstreamSession, messageID uint64, destID string, sourceTerminalID string, msgFmt uint8, registerDelivery uint8, content string) cmpp.Packer { @@ -850,21 +933,134 @@ func findSession(messageID string, account string) *downstreamSession { return nil } -func sendDownstream(session *downstreamSession, deliver cmpp.Packer) (bool, error) { +func sendDownstream(session *downstreamSession, deliver cmpp.Packer, deliveryID string) (DownstreamSendResult, error) { session.mu.Lock() defer session.mu.Unlock() - if err := session.conn.SendPkt(deliver, <-session.conn.SeqId); err != nil { + sequenceID := <-session.conn.SeqId + messageID := downstreamDeliverMessageID(deliver) + sentAt := time.Now().UTC() + ackDeadlineAt := sentAt.Add(downstreamAckTimeout()) + tracker := registerDownstreamAck(session, deliveryID, sequenceID, messageID, ackDeadlineAt) + if err := session.conn.SendPkt(deliver, sequenceID); err != nil { + removeDownstreamAck(tracker) if session.report != nil { go session.report(session, "disconnected", err.Error()) } forgetDownstream(session) - return false, err + return DownstreamSendResult{}, err + } + result := DownstreamSendResult{ + Sent: true, ConnectionID: session.connectionID, + SequenceID: strconv.FormatUint(uint64(sequenceID), 10), MessageID: strconv.FormatUint(messageID, 10), + SentAt: formatRFC3339Nano(sentAt), AckDeadlineAt: formatRFC3339Nano(ackDeadlineAt), + } + if deliveryID != "" && session.deliveryReport != nil { + go session.deliveryReport(downstreamDeliveryLifecycleEvent{ + Kind: "sent", DeliveryID: deliveryID, ConnectionID: session.connectionID, + SequenceID: sequenceID, MessageID: messageID, ObservedAt: sentAt, AckDeadlineAt: ackDeadlineAt, + }) } session.touchPresence("connected", false, true) if session.report != nil { go session.report(session, "deliver", "") } - return true, nil + return result, nil +} + +func downstreamDeliverMessageID(deliver cmpp.Packer) uint64 { + switch packet := deliver.(type) { + case *cmpp.Cmpp2DeliverReqPkt: + return packet.MsgId + case *cmpp.Cmpp3DeliverReqPkt: + return packet.MsgId + default: + return 0 + } +} + +func downstreamAckKey(conn *cmpp.Conn, sequenceID uint32) string { + return fmt.Sprintf("%p:%d", conn, sequenceID) +} + +func registerDownstreamAck(session *downstreamSession, deliveryID string, sequenceID uint32, messageID uint64, deadline time.Time) *downstreamAckTracker { + if session == nil || session.conn == nil || strings.TrimSpace(deliveryID) == "" { + return nil + } + tracker := &downstreamAckTracker{ + deliveryID: deliveryID, connectionID: session.connectionID, + sequenceID: sequenceID, messageID: messageID, session: session, + } + key := downstreamAckKey(session.conn, sequenceID) + downstreamAckRegistry.Lock() + downstreamAckRegistry.items[key] = tracker + downstreamAckRegistry.Unlock() + tracker.timer = time.AfterFunc(time.Until(deadline), func() { + timedOut := takeDownstreamAck(session.conn, sequenceID) + if timedOut == nil || timedOut.session == nil || timedOut.session.deliveryReport == nil { + return + } + timedOut.session.deliveryReport(downstreamDeliveryLifecycleEvent{ + Kind: "failed", DeliveryID: timedOut.deliveryID, ConnectionID: timedOut.connectionID, + SequenceID: timedOut.sequenceID, MessageID: timedOut.messageID, ObservedAt: time.Now().UTC(), + FailureType: "ack_timeout", ErrorMessage: "CMPP_DELIVER_RESP timeout", + }) + }) + return tracker +} + +func takeDownstreamAck(conn *cmpp.Conn, sequenceID uint32) *downstreamAckTracker { + key := downstreamAckKey(conn, sequenceID) + downstreamAckRegistry.Lock() + tracker := downstreamAckRegistry.items[key] + delete(downstreamAckRegistry.items, key) + downstreamAckRegistry.Unlock() + if tracker != nil && tracker.timer != nil { + tracker.timer.Stop() + } + return tracker +} + +func removeDownstreamAck(tracker *downstreamAckTracker) { + if tracker == nil || tracker.session == nil { + return + } + _ = takeDownstreamAck(tracker.session.conn, tracker.sequenceID) +} + +func handleDownstreamAcknowledgement(conn *cmpp.Conn, sequenceID uint32, messageID uint64, result uint32, logger *log.Logger) { + tracker := takeDownstreamAck(conn, sequenceID) + if tracker == nil { + if logger != nil { + logger.Printf("cmpp inbound event=deliver_ack_unmatched seq=%d message_id=%d result=%d", sequenceID, messageID, result) + } + return + } + if tracker.messageID != messageID { + if logger != nil { + logger.Printf("cmpp inbound event=deliver_ack_message_mismatch delivery_id=%s seq=%d expected_message_id=%d actual_message_id=%d", tracker.deliveryID, sequenceID, tracker.messageID, messageID) + } + result = 1 + } + if logger != nil { + logger.Printf("cmpp inbound event=deliver_acknowledged delivery_id=%s connection_id=%s seq=%d message_id=%d result=%d", tracker.deliveryID, tracker.connectionID, sequenceID, messageID, result) + } + if tracker.session != nil && tracker.session.deliveryReport != nil { + go tracker.session.deliveryReport(downstreamDeliveryLifecycleEvent{ + Kind: "acknowledged", DeliveryID: tracker.deliveryID, ConnectionID: tracker.connectionID, + SequenceID: sequenceID, MessageID: messageID, Result: result, ObservedAt: time.Now().UTC(), + }) + } +} + +func downstreamAckTimeout() time.Duration { + configured, err := strconv.Atoi(strings.TrimSpace(os.Getenv("CMPP_DOWNSTREAM_ACK_TIMEOUT_SECONDS"))) + if err != nil || configured <= 0 { + return defaultDownstreamAckTimeout + } + if configured < 5 { + configured = 5 + } + return time.Duration(configured) * time.Second } func (s Server) pendingFlushInterval() time.Duration { diff --git a/gateway/internal/inbound/server_test.go b/gateway/internal/inbound/server_test.go index 54a6425..f520508 100644 --- a/gateway/internal/inbound/server_test.go +++ b/gateway/internal/inbound/server_test.go @@ -100,6 +100,7 @@ func TestInboundServerAuthenticatesAndSubmits(t *testing.T) { var gotAuth authRequest var gotSubmit submitRequest connectionEvents := make(chan downstreamConnectionEvent, 8) + acknowledgements := make(chan map[string]any, 1) api := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { switch r.URL.Path { case "/api/gateway/events/inbound/authenticate": @@ -121,6 +122,15 @@ func TestInboundServerAuthenticatesAndSubmits(t *testing.T) { } connectionEvents <- event w.WriteHeader(http.StatusOK) + case "/api/gateway/events/downstream/sent": + w.WriteHeader(http.StatusOK) + case "/api/gateway/events/downstream/acknowledged": + var event map[string]any + if err := json.NewDecoder(r.Body).Decode(&event); err != nil { + t.Fatalf("decode acknowledgement: %v", err) + } + acknowledgements <- event + w.WriteHeader(http.StatusOK) default: t.Fatalf("unexpected api path: %s", r.URL.Path) } @@ -176,14 +186,15 @@ func TestInboundServerAuthenticatesAndSubmits(t *testing.T) { if rsp.Result != 0 || rsp.MsgId == 0 { t.Fatalf("unexpected submit response: %+v", rsp) } - delivered, err := PushReceipt(DownstreamReceipt{ + sendResult, err := PushReceiptWithResult(DownstreamReceipt{ + DeliveryID: "delivery-1", MessageID: "MSG-1", PhoneNumber: "13500002696", ReceiptStatus: "delivered", DeliveredAt: time.Now().UTC().Format(time.RFC3339Nano), }) - if err != nil || !delivered { - t.Fatalf("push receipt delivered=%v err=%v", delivered, err) + if err != nil || !sendResult.Sent { + t.Fatalf("push receipt sent=%v err=%v", sendResult.Sent, err) } deliver := recvDeliver(t, client) if deliver.RegisterDelivery != 1 { @@ -196,6 +207,17 @@ func TestInboundServerAuthenticatesAndSubmits(t *testing.T) { if receipt.Stat != "DELIVRD" || receipt.DestTerminalId != "13500002696" { t.Fatalf("unexpected pushed receipt: %+v", receipt) } + if err := client.SendRspPkt(&cmpp.Cmpp3DeliverRspPkt{MsgId: deliver.MsgId, Result: 0}, deliver.SeqId); err != nil { + t.Fatalf("send deliver response: %v", err) + } + select { + case event := <-acknowledgements: + if event["id"] != "delivery-1" || event["result"] != float64(0) { + t.Fatalf("unexpected acknowledgement callback: %+v", event) + } + case <-time.After(2 * time.Second): + t.Fatal("expected downstream acknowledgement callback") + } if gotAuth.Account != account || gotAuth.AuthSource == "" || gotAuth.RemoteIP == "" { t.Fatalf("unexpected auth payload: %+v", gotAuth) } @@ -537,6 +559,50 @@ func TestRememberAndForgetAccountUpdatesPresenceStore(t *testing.T) { } } +func TestDownstreamDeliveryRequiresAcknowledgement(t *testing.T) { + resetDownstreamRegistry() + defer resetDownstreamRegistry() + + events := make(chan downstreamDeliveryLifecycleEvent, 1) + conn := &cmpp.Conn{} + session := &downstreamSession{ + conn: conn, connectionID: "conn-1", + deliveryReport: func(event downstreamDeliveryLifecycleEvent) { events <- event }, + } + registerDownstreamAck(session, "delivery-1", 37, 9016479179509871733, time.Now().Add(time.Second)) + handleDownstreamAcknowledgement(conn, 37, 9016479179509871733, 0, log.Default()) + + select { + case event := <-events: + if event.Kind != "acknowledged" || event.DeliveryID != "delivery-1" || event.Result != 0 || event.SequenceID != 37 { + t.Fatalf("unexpected acknowledgement event: %+v", event) + } + case <-time.After(time.Second): + t.Fatal("timed out waiting acknowledgement event") + } +} + +func TestDownstreamDeliveryReportsAckTimeout(t *testing.T) { + resetDownstreamRegistry() + defer resetDownstreamRegistry() + + events := make(chan downstreamDeliveryLifecycleEvent, 1) + session := &downstreamSession{ + conn: &cmpp.Conn{}, connectionID: "conn-1", + deliveryReport: func(event downstreamDeliveryLifecycleEvent) { events <- event }, + } + registerDownstreamAck(session, "delivery-timeout", 38, 9017467844344255865, time.Now().Add(20*time.Millisecond)) + + select { + case event := <-events: + if event.Kind != "failed" || event.FailureType != "ack_timeout" || event.DeliveryID != "delivery-timeout" { + t.Fatalf("unexpected timeout event: %+v", event) + } + case <-time.After(time.Second): + t.Fatal("timed out waiting acknowledgement timeout") + } +} + func recvDeliver(t *testing.T, client *cmpp.Client) *cmpp.Cmpp3DeliverReqPkt { t.Helper() deadline := time.Now().Add(2 * time.Second) @@ -575,6 +641,14 @@ func resetDownstreamRegistry() { downstreamRegistry.byAccount = make(map[string]*downstreamSession) downstreamRegistry.byMessageID = make(map[string]*downstreamSession) downstreamRegistry.byConn = make(map[*cmpp.Conn]*downstreamSession) + downstreamAckRegistry.Lock() + for _, tracker := range downstreamAckRegistry.items { + if tracker.timer != nil { + tracker.timer.Stop() + } + } + downstreamAckRegistry.items = make(map[string]*downstreamAckTracker) + downstreamAckRegistry.Unlock() } func reserveTCPAddr(t *testing.T) string { diff --git a/src/api/adminApi.ts b/src/api/adminApi.ts index 9e80152..fe8c6b7 100644 --- a/src/api/adminApi.ts +++ b/src/api/adminApi.ts @@ -1,9 +1,22 @@ -import { clearSession, getSessionTenantId, readSession, type LoginSession } from './session'; +import { clearSession, dispatchSessionEvent, getSessionTenantId, hasRecentUserActivity, readSession, requestReauthentication, type LoginSession } from './session'; type RequestOptions = RequestInit & { tenantId?: string; + reauthenticationAttempted?: boolean; }; +type ApiErrorBody = { message?: string | string[]; error?: string; code?: string }; + +async function readErrorBody(response: Response): Promise { + const text = await response.text(); + if (!text) return {}; + try { + return JSON.parse(text) as ApiErrorBody; + } catch { + return { message: text }; + } +} + export const DEFAULT_CLIENT_TENANT_ID = 'tenant-a'; async function readErrorMessage(response: Response) { @@ -27,19 +40,30 @@ async function request(path: string, options: RequestOptions = {}): Promise(path, { ...options, reauthenticationAttempted: true }); + } + } if (!response.ok) { throw new Error(await readErrorMessage(response)); } @@ -49,19 +73,30 @@ async function request(path: string, options: RequestOptions = {}): Promise { const headers = new Headers(options.headers); const session = readSession(); - if (session?.accessToken) { - headers.set('Authorization', `${session.tokenType} ${session.accessToken}`); - } + if (session && hasRecentUserActivity()) headers.set('x-session-activity', 'user'); const tenantId = options.tenantId ?? (path.startsWith('/client') ? getSessionTenantId() : undefined); if (tenantId) { headers.set('x-tenant-id', tenantId); } - const response = await fetch(`/api${path}`, { ...options, headers }); + const response = await fetch(`/api${path}`, { ...options, headers, credentials: 'same-origin' }); if (response.status === 401 && session) { + const body = await readErrorBody(response.clone()); + if (body.code === 'SESSION_LOCKED') { + dispatchSessionEvent('locked', { message: body.message }); + throw new Error(typeof body.message === 'string' ? body.message : '由于长时间未操作，会话已安全锁定'); + } clearSession(); + dispatchSessionEvent('logout', { code: body.code, message: body.message }); window.location.assign(session.portal === 'admin' ? '/admin/login' : '/client/login'); throw new Error('登录会话已失效，请重新登录'); } + if (response.status === 403 && session && !options.reauthenticationAttempted) { + const body = await readErrorBody(response.clone()); + if (body.code === 'RECENT_AUTHENTICATION_REQUIRED') { + await requestReauthentication(); + return requestBlob(path, { ...options, reauthenticationAttempted: true }); + } + } if (!response.ok) { throw new Error(await readErrorMessage(response)); } @@ -689,6 +724,8 @@ export type EnterpriseApplication = { queuePriority?: 'normal' | 'priority' | string | null; maxPhonesPerTask?: number | null; templateMismatchMode?: string | null; + downstreamReceiptRetryEnabled?: boolean | null; + downstreamUplinkRetryEnabled?: boolean | null; cmppAccount?: string | null; cmppEnterpriseCode?: string | null; interfaceEnabled?: boolean | null; @@ -775,7 +812,15 @@ export type DownstreamDeliveryRecord = { status: string; payload: Record; retryCount: number; + retryEnabled: boolean; nextRetryAt?: string | null; + sentAt?: string | null; + acknowledgedAt?: string | null; + ackDeadlineAt?: string | null; + ackResult?: number | null; + ackSequenceId?: string | null; + ackMessageId?: string | null; + connectionId?: string | null; deliveredAt?: string | null; lastError?: string | null; createdAt: string; @@ -796,9 +841,13 @@ export type DownstreamDeliveryDashboard = { summary: { total: number; pending: number; + awaitingAck: number; delivered: number; failed: number; + unconfirmed: number; + rejected: number; stalledPending: number; + stalledAck: number; recentFailed: number; alertCount: number; }; @@ -806,8 +855,11 @@ export type DownstreamDeliveryDashboard = { deliveryType: string; total: number; pending: number; + awaitingAck: number; delivered: number; failed: number; + unconfirmed: number; + rejected: number; }>; retryBuckets: Array<{ label: string; @@ -817,7 +869,10 @@ export type DownstreamDeliveryDashboard = { applicationId: string; name: string; pending: number; + awaitingAck: number; failed: number; + unconfirmed: number; + rejected: number; delivered: number; alertCount: number; }>; @@ -881,6 +936,11 @@ export const adminApi = { getCaptcha: () => request('/admin/auth/captcha'), login: (body: { login: string; password: string; captchaId: string; captchaText: string }) => request('/admin/auth/login', { method: 'POST', body: JSON.stringify(body) }), + touchSession: () => request>('/auth/session/touch', { method: 'POST', body: '{}' }), + lockSession: () => request<{ locked: boolean }>('/auth/session/lock', { method: 'POST', body: '{}' }), + unlockSession: (password: string) => request>('/auth/session/unlock', { method: 'POST', body: JSON.stringify({ password }) }), + reauthenticate: (password: string) => request>('/auth/reauthenticate', { method: 'POST', body: JSON.stringify({ password }), reauthenticationAttempted: true }), + logout: () => request<{ success: boolean }>('/auth/logout', { method: 'POST', body: '{}' }), changeOwnPassword: (body: { currentPassword: string; password: string }) => request('/auth/password', { method: 'POST', body: JSON.stringify(body) }), listTenants: () => request('/admin/tenants'), @@ -912,9 +972,9 @@ export const adminApi = { request(withQuery('/admin/enterprise-applications', query)), getEnterpriseApplication: (id: string) => request(`/admin/enterprise-applications/${id}`), - createEnterpriseApplication: (body: { tenantId: string; name: string; scene?: string; dailyLimit?: number; customerUnitPrice?: number; queuePriority?: 'normal' | 'priority'; maxPhonesPerTask?: number; templateMismatchMode?: string; cmppAccount?: string; cmppEnterpriseCode?: string; passwordCipher?: string; interfaceEnabled?: boolean; interfaceType?: 'cmpp20'; cmppMaxConnections?: number; cmppWindowSize?: number; ipAllowlist?: string[] }) => + createEnterpriseApplication: (body: { tenantId: string; name: string; scene?: string; dailyLimit?: number; customerUnitPrice?: number; queuePriority?: 'normal' | 'priority'; maxPhonesPerTask?: number; templateMismatchMode?: string; downstreamReceiptRetryEnabled?: boolean; downstreamUplinkRetryEnabled?: boolean; cmppAccount?: string; cmppEnterpriseCode?: string; passwordCipher?: string; interfaceEnabled?: boolean; interfaceType?: 'cmpp20'; cmppMaxConnections?: number; cmppWindowSize?: number; ipAllowlist?: string[] }) => request('/admin/enterprise-applications', { method: 'POST', body: JSON.stringify(body) }), - updateEnterpriseApplication: (id: string, body: { name?: string; scene?: string; dailyLimit?: number; customerUnitPrice?: number; queuePriority?: 'normal' | 'priority'; maxPhonesPerTask?: number; templateMismatchMode?: string; cmppAccount?: string; cmppEnterpriseCode?: string; passwordCipher?: string; interfaceEnabled?: boolean; interfaceType?: 'cmpp20'; cmppMaxConnections?: number; cmppWindowSize?: number; ipAllowlist?: string[] }) => + updateEnterpriseApplication: (id: string, body: { name?: string; scene?: string; dailyLimit?: number; customerUnitPrice?: number; queuePriority?: 'normal' | 'priority'; maxPhonesPerTask?: number; templateMismatchMode?: string; downstreamReceiptRetryEnabled?: boolean; downstreamUplinkRetryEnabled?: boolean; cmppAccount?: string; cmppEnterpriseCode?: string; passwordCipher?: string; interfaceEnabled?: boolean; interfaceType?: 'cmpp20'; cmppMaxConnections?: number; cmppWindowSize?: number; ipAllowlist?: string[] }) => request(`/admin/enterprise-applications/${id}`, { method: 'PUT', body: JSON.stringify(body) }), changeApplicationStatus: (id: string, status: string, reason?: string) => request(`/admin/enterprise-applications/${id}/status`, { @@ -1103,13 +1163,21 @@ export const adminApi = { } const headers = new Headers(); const session = readSession(); - if (session?.accessToken) { - headers.set('Authorization', `${session.tokenType} ${session.accessToken}`); - } + if (session && hasRecentUserActivity()) headers.set('x-session-activity', 'user'); if (tenantId) { headers.set('x-tenant-id', tenantId); } - const response = await fetch('/api/admin/files/upload', { method: 'POST', headers, body: form }); + const response = await fetch('/api/admin/files/upload', { method: 'POST', headers, body: form, credentials: 'same-origin' }); + if (response.status === 401 && session) { + const error = await readErrorBody(response.clone()); + if (error.code === 'SESSION_LOCKED') { + dispatchSessionEvent('locked', { message: error.message }); + } else { + clearSession(); + dispatchSessionEvent('logout', { code: error.code, message: error.message }); + window.location.assign(session.portal === 'admin' ? '/admin/login' : '/client/login'); + } + } if (!response.ok) { throw new Error(await response.text()); } @@ -1210,13 +1278,21 @@ export const clientApi = { } const headers = new Headers(); const session = readSession(); - if (session?.accessToken) { - headers.set('Authorization', `${session.tokenType} ${session.accessToken}`); - } + if (session && hasRecentUserActivity()) headers.set('x-session-activity', 'user'); if (tenantId) { headers.set('x-tenant-id', tenantId); } - const response = await fetch('/api/admin/files/upload', { method: 'POST', headers, body: form }); + const response = await fetch('/api/admin/files/upload', { method: 'POST', headers, body: form, credentials: 'same-origin' }); + if (response.status === 401 && session) { + const error = await readErrorBody(response.clone()); + if (error.code === 'SESSION_LOCKED') { + dispatchSessionEvent('locked', { message: error.message }); + } else { + clearSession(); + dispatchSessionEvent('logout', { code: error.code, message: error.message }); + window.location.assign(session.portal === 'admin' ? '/admin/login' : '/client/login'); + } + } if (!response.ok) { throw new Error(await response.text()); } diff --git a/src/api/session.ts b/src/api/session.ts index 7f9147d..d3a49dc 100644 --- a/src/api/session.ts +++ b/src/api/session.ts @@ -12,10 +12,13 @@ export type SessionUser = { }; export type LoginSession = { - accessToken: string; - tokenType: string; portal: Portal; user: SessionUser; + idleTimeoutSeconds: number; + lockRecoverySeconds: number; + absoluteExpiresAt: string; + lastActivityAt: string; + recentAuthenticationExpiresAt: string; }; const sessionKey = 'cmpp-auth-session'; @@ -37,6 +40,46 @@ export function clearSession() { window.localStorage.removeItem(sessionKey); } +export function updateSessionTiming(timing: Partial>) { + const current = readSession(); + if (current) writeSession({ ...current, ...timing }); +} + +let lastUserActivityAt = Date.now(); +let reauthenticationHandler: (() => Promise) | undefined; + +export function markUserActivity() { + lastUserActivityAt = Date.now(); +} + +export function getLastUserActivityAt() { + return lastUserActivityAt; +} + +export function hasRecentUserActivity() { + return Date.now() - lastUserActivityAt < 60_000; +} + +export function setReauthenticationHandler(handler?: () => Promise) { + reauthenticationHandler = handler; +} + +export function requestReauthentication() { + if (!reauthenticationHandler) return Promise.reject(new Error('请重新验证当前密码后再操作')); + return reauthenticationHandler(); +} + +export function dispatchSessionEvent(type: 'locked' | 'unlocked' | 'logout', detail?: Record) { + window.dispatchEvent(new CustomEvent(`cmpp-session-${type}`, { detail })); + try { + const channel = new BroadcastChannel('cmpp-session'); + channel.postMessage({ type, detail }); + channel.close(); + } catch { + // BroadcastChannel is an enhancement; the current tab still receives the DOM event. + } +} + export function getSessionTenantId() { return readSession()?.user.tenantId ?? undefined; } diff --git a/src/apps/admin/AdminDownstreamDeliveriesPage.tsx b/src/apps/admin/AdminDownstreamDeliveriesPage.tsx index 3478447..f35f378 100644 --- a/src/apps/admin/AdminDownstreamDeliveriesPage.tsx +++ b/src/apps/admin/AdminDownstreamDeliveriesPage.tsx @@ -5,8 +5,20 @@ import { Breadcrumb, Button, Input, Modal, Pagination, Select, Table, Tag, type const statusTone: Record = { pending: 'warning', + awaiting_ack: 'info', delivered: 'success', failed: 'danger', + unconfirmed: 'warning', + rejected: 'danger', +}; + +const statusLabel: Record = { + pending: '待首次投递', + awaiting_ack: '等待客户端确认', + delivered: '客户端已确认', + failed: '投递失败', + unconfirmed: '客户端未确认', + rejected: '客户端拒绝', }; const deliveryTypeLabel: Record = { @@ -30,11 +42,17 @@ function DeliveryDetailModal({ record, onClose }: { record: DownstreamDeliveryRe
企业{record.tenant?.name ?? record.tenantId}
应用{record.application?.name ?? record.applicationId}
投递类型{deliveryTypeLabel[record.deliveryType] ?? record.deliveryType}
-
当前状态{record.status}
+
当前状态{statusLabel[record.status] ?? record.status}
消息 ID{record.messageId ?? '-'}
重试次数{record.retryCount}
下次重试{record.nextRetryAt ?? '-'}
-
已投递时间{record.deliveredAt ?? '-'}
+
自动重试{record.retryEnabled ? '开启' : '关闭'}
+
写出时间{record.sentAt ?? '-'}
+
确认时间{record.acknowledgedAt ?? '-'}
+
ACK Result{record.ackResult ?? '-'}
+
ACK Sequence_Id{record.ackSequenceId ?? '-'}
+
ACK Msg_Id{record.ackMessageId ?? '-'}
+
连接 ID{record.connectionId ?? '-'}
最后错误{record.lastError ?? '-'}
@@ -96,7 +114,7 @@ export function AdminDownstreamDeliveriesPage() { }, [loadData]); const selectableIds = useMemo( - () => records.filter((item) => item.status !== 'delivered').map((item) => item.id), + () => records.filter((item) => ['pending', 'failed', 'unconfirmed', 'rejected'].includes(item.status)).map((item) => item.id), [records], ); const allSelected = selectableIds.length > 0 && selectableIds.every((id) => selectedIds.includes(id)); @@ -114,7 +132,7 @@ export function AdminDownstreamDeliveriesPage() { render: (record) => ( { setSelectedIds((current) => @@ -132,7 +150,7 @@ export function AdminDownstreamDeliveriesPage() { { key: 'application', title: '应用', width: '180px', render: (record) => record.application?.name ?? '-' }, { key: 'type', title: '类型', width: '110px', render: (record) => deliveryTypeLabel[record.deliveryType] ?? record.deliveryType }, { key: 'messageId', title: '消息 ID', width: '180px', render: (record) => {record.messageId ?? '-'} }, - { key: 'status', title: '状态', width: '110px', render: (record) => {record.status} }, + { key: 'status', title: '状态', width: '150px', render: (record) => {statusLabel[record.status] ?? record.status} }, { key: 'retry', title: '重试', width: '90px', align: 'center', render: (record) => record.retryCount }, { key: 'error', title: '最后错误', render: (record) => record.lastError ?? '-' }, { @@ -144,8 +162,10 @@ export function AdminDownstreamDeliveriesPage() {
+ +
+
+ + 首次投递始终保留；关闭后，已写出但未收到 CMPP_DELIVER_RESP 的消息不会自动重发，仍可在下游投递记录中手工重投。 +
+
diff --git a/src/layouts/AppShell.tsx b/src/layouts/AppShell.tsx index 57faead..2c82a3b 100644 --- a/src/layouts/AppShell.tsx +++ b/src/layouts/AppShell.tsx @@ -1,5 +1,5 @@ import type { ComponentType } from 'react'; -import { useEffect, useMemo, useState } from 'react'; +import { useEffect, useMemo, useRef, useState } from 'react'; import { Bell, ChevronDown, @@ -12,7 +12,15 @@ import { } from 'lucide-react'; import { NavLink, Outlet, useNavigate } from 'react-router-dom'; import { adminApi } from '@/api/adminApi'; -import { clearSession } from '@/api/session'; +import { + clearSession, + dispatchSessionEvent, + getLastUserActivityAt, + markUserActivity, + readSession, + setReauthenticationHandler, + updateSessionTiming, +} from '@/api/session'; import { Button, Input, Modal } from '@/components/ui'; export type ShellNavItem = { @@ -64,6 +72,18 @@ export function AppShell({ const [confirmPassword, setConfirmPassword] = useState(''); const [passwordError, setPasswordError] = useState(''); const [passwordSaving, setPasswordSaving] = useState(false); + const [idleWarningSeconds, setIdleWarningSeconds] = useState(null); + const [locked, setLocked] = useState(false); + const [unlockPassword, setUnlockPassword] = useState(''); + const [unlockError, setUnlockError] = useState(''); + const [unlocking, setUnlocking] = useState(false); + const [reauthenticationOpen, setReauthenticationOpen] = useState(false); + const [reauthenticationPassword, setReauthenticationPassword] = useState(''); + const [reauthenticationError, setReauthenticationError] = useState(''); + const [reauthenticating, setReauthenticating] = useState(false); + const reauthenticationResolve = useRef<(() => void) | null>(null); + const reauthenticationReject = useRef<((error: Error) => void) | null>(null); + const lockRequested = useRef(false); const navigate = useNavigate(); const ToggleIcon = collapsed ? PanelLeftOpen : PanelLeftClose; const auditTotal = useMemo( @@ -85,6 +105,7 @@ export function AppShell({ try { await adminApi.changeOwnPassword({ currentPassword, password: newPassword }); clearSession(); + dispatchSessionEvent('logout', { message: '密码修改成功，请重新登录' }); navigate(loginPath, { replace: true }); } catch (error) { setPasswordError(error instanceof Error ? error.message : '修改密码失败'); @@ -93,6 +114,139 @@ export function AppShell({ } } + async function logout() { + try { + await adminApi.logout(); + } finally { + clearSession(); + dispatchSessionEvent('logout'); + navigate(loginPath, { replace: true }); + } + } + + async function continueSession() { + try { + const timing = await adminApi.touchSession(); + updateSessionTiming(timing); + markUserActivity(); + setIdleWarningSeconds(null); + } catch { + setLocked(true); + } + } + + async function unlockSession() { + if (!unlockPassword) { + setUnlockError('请输入当前密码'); + return; + } + setUnlocking(true); + setUnlockError(''); + try { + const timing = await adminApi.unlockSession(unlockPassword); + updateSessionTiming(timing); + markUserActivity(); + setLocked(false); + lockRequested.current = false; + setUnlockPassword(''); + setIdleWarningSeconds(null); + dispatchSessionEvent('unlocked'); + } catch (error) { + setUnlockError(error instanceof Error ? error.message : '解锁失败'); + } finally { + setUnlocking(false); + } + } + + async function confirmReauthentication() { + if (!reauthenticationPassword) { + setReauthenticationError('请输入当前密码'); + return; + } + setReauthenticating(true); + setReauthenticationError(''); + try { + const timing = await adminApi.reauthenticate(reauthenticationPassword); + updateSessionTiming(timing); + reauthenticationResolve.current?.(); + reauthenticationResolve.current = null; + reauthenticationReject.current = null; + setReauthenticationOpen(false); + setReauthenticationPassword(''); + } catch (error) { + setReauthenticationError(error instanceof Error ? error.message : '身份验证失败'); + } finally { + setReauthenticating(false); + } + } + + useEffect(() => { + const activityEvents = ['pointerdown', 'keydown', 'touchstart', 'scroll'] as const; + const onActivity = () => markUserActivity(); + activityEvents.forEach((eventName) => window.addEventListener(eventName, onActivity, { passive: true })); + + const onLocked = () => setLocked(true); + const onUnlocked = () => { setLocked(false); markUserActivity(); }; + const onLogout = () => { clearSession(); navigate(loginPath, { replace: true }); }; + window.addEventListener('cmpp-session-locked', onLocked); + window.addEventListener('cmpp-session-unlocked', onUnlocked); + window.addEventListener('cmpp-session-logout', onLogout); + + let channel: BroadcastChannel | undefined; + try { + channel = new BroadcastChannel('cmpp-session'); + channel.onmessage = (event: MessageEvent<{ type?: string }>) => { + if (event.data?.type === 'locked') onLocked(); + if (event.data?.type === 'unlocked') onUnlocked(); + if (event.data?.type === 'logout') onLogout(); + }; + } catch { + channel = undefined; + } + + setReauthenticationHandler(() => new Promise((resolve, reject) => { + reauthenticationResolve.current = resolve; + reauthenticationReject.current = reject; + setReauthenticationPassword(''); + setReauthenticationError(''); + setReauthenticationOpen(true); + })); + + const timer = window.setInterval(() => { + const session = readSession(); + if (!session) return; + const now = Date.now(); + if (now >= Date.parse(session.absoluteExpiresAt)) { + clearSession(); + dispatchSessionEvent('logout', { code: 'SESSION_ABSOLUTE_TIMEOUT' }); + navigate(loginPath, { replace: true }); + return; + } + const remaining = session.idleTimeoutSeconds * 1000 - (now - getLastUserActivityAt()); + if (remaining <= 0 && !lockRequested.current) { + lockRequested.current = true; + setLocked(true); + setIdleWarningSeconds(null); + void adminApi.lockSession().catch(() => undefined); + } else if (remaining <= 5 * 60 * 1000) { + setIdleWarningSeconds(Math.ceil(remaining / 1000)); + } else { + setIdleWarningSeconds(null); + } + }, 1000); + + return () => { + activityEvents.forEach((eventName) => window.removeEventListener(eventName, onActivity)); + window.removeEventListener('cmpp-session-locked', onLocked); + window.removeEventListener('cmpp-session-unlocked', onUnlocked); + window.removeEventListener('cmpp-session-logout', onLogout); + channel?.close(); + window.clearInterval(timer); + setReauthenticationHandler(undefined); + reauthenticationReject.current?.(new Error('身份验证已取消')); + }; + }, [loginPath, navigate]); + useEffect(() => { if (auditTotal <= 0 || typeof window === 'undefined') { return; @@ -228,9 +382,8 @@ export function AppShell({ 修改密码 } + onClose={() => undefined} + open={locked} + title="会话已安全锁定" + > +

由于长时间未操作，请输入当前密码继续使用。锁定超过 4 小时后需要完整登录。

+ setUnlockPassword(event.target.value)} type="password" value={unlockPassword} /> + {unlockError ?

{unlockError}

: null} + + } + onClose={() => setIdleWarningSeconds(null)} + open={!locked && idleWarningSeconds !== null} + title="会话即将锁定" + > +

长时间未操作，系统将在 {formatCountdown(idleWarningSeconds ?? 0)} 后锁定。点击“继续使用”可保持当前会话。

+ + } + onClose={() => { reauthenticationReject.current?.(new Error('已取消敏感操作')); reauthenticationResolve.current = null; reauthenticationReject.current = null; setReauthenticationOpen(false); }} + open={reauthenticationOpen} + title="敏感操作身份验证" + > +

该操作影响账号、通道或资金安全，请输入当前登录用户密码。验证通过后 30 分钟内无需重复输入。

+ setReauthenticationPassword(event.target.value)} type="password" value={reauthenticationPassword} /> + {reauthenticationError ?

{reauthenticationError}

: null} + ); } + +function formatCountdown(seconds: number) { + const minutes = Math.floor(seconds / 60); + return `${String(minutes).padStart(2, '0')}:${String(seconds % 60).padStart(2, '0')}`; +} diff --git a/src/styles/global.css b/src/styles/global.css index f5fc047..dc52351 100644 --- a/src/styles/global.css +++ b/src/styles/global.css @@ -1059,6 +1059,11 @@ h3 { grid-template-columns: minmax(180px, 1.4fr) repeat(4, minmax(72px, 0.65fr)); } +.downstream-breakdown-table__head--ack, +.downstream-breakdown-table__row--ack { + grid-template-columns: minmax(120px, 1.2fr) repeat(6, minmax(72px, 0.65fr)); +} + .downstream-breakdown-table__row { border-bottom: 1px solid var(--color-border); }