feat: add protocol interaction observability

api/src/app.module.ts

@@ -15,6 +15,7 @@ import { HealthController } from './health.controller';
 import { OperationsModule } from './operations/operations.module';
 import { OpenApiModule } from './open-api/open-api.module';
 import { PrismaModule } from './prisma/prisma.module';
+import { ProtocolLogsModule } from './protocol-logs/protocol-logs.module';
 import { RiskReviewModule } from './risk-review/risk-review.module';
 import { ReportsModule } from './reports/reports.module';
 import { ReportMaterialsModule } from './report-materials/report-materials.module';
@@ -30,6 +31,7 @@ import { UsersModule } from './users/users.module';
       envFilePath: ['.env.local', '.env'],
     }),
     PrismaModule,
+    ProtocolLogsModule,
     AuthModule,
     TenantsModule,
     UsersModule,

api/src/open-api/open-api.service.ts

@@ -1,4 +1,4 @@
-import { BadRequestException, ConflictException, ForbiddenException, forwardRef, HttpException, Inject, Injectable, NotFoundException, OnModuleDestroy, OnModuleInit, UnprocessableEntityException } from '@nestjs/common';
+import { BadRequestException, ConflictException, ForbiddenException, forwardRef, HttpException, Inject, Injectable, NotFoundException, OnModuleDestroy, OnModuleInit, Optional, UnprocessableEntityException } from '@nestjs/common';
 import { Prisma } from '@prisma/client';
 import { Queue, Worker } from 'bullmq';
 import { createHash, createHmac, randomBytes, randomUUID } from 'node:crypto';
@@ -10,6 +10,7 @@ import { PrismaService } from '../prisma/prisma.service';
 import { SendChainService } from '../send-chain/send-chain.service';
 import { decryptSecret, encryptSecret } from './open-api.crypto';
 import type { OpenApiAuthContext } from './open-api.types';
+import { ProtocolLogsService } from '../protocol-logs/protocol-logs.service';
 
 const WEBHOOK_QUEUE = 'http-webhook-delivery';
 const DELIVERY_MODES = ['cmpp', 'http', 'both', 'none'] as const;
@@ -45,7 +46,11 @@ export class OpenApiService implements OnModuleInit, OnModuleDestroy {
   private queue?: Queue<{ deliveryId: string }>;
   private worker?: Worker<{ deliveryId: string }>;
 
-  constructor(private readonly prisma: PrismaService, @Inject(forwardRef(() => SendChainService)) private readonly sendChain: SendChainService) {}
+  constructor(
+    private readonly prisma: PrismaService,
+    @Inject(forwardRef(() => SendChainService)) private readonly sendChain: SendChainService,
+    @Optional() private readonly protocolLogs?: ProtocolLogsService,
+  ) {}
 
   onModuleInit() {
     const connection = bullmqConnection();
@@ -203,6 +208,21 @@ export class OpenApiService implements OnModuleInit, OnModuleDestroy {
       }
       const response = { code: 'ACCEPTED', requestId, messageId: message?.messageId, clientMessageId: input.clientMessageId ?? null, status: message?.status ?? task.status, acceptedAt: new Date().toISOString() };
       await this.prisma.openApiRequest.update({ where: { id: request.id }, data: { status: 'completed', httpStatus: 202, businessCode: 'ACCEPTED', responseBody: response, messageRecordId: message?.id, durationMs: Date.now() - startedAt, completedAt: new Date() } });
+      this.protocolLogs?.record({
+        protocol: 'http',
+        direction: 'client_to_platform',
+        eventType: 'send_request',
+        status: 'accepted',
+        tenantId: auth.application.tenantId,
+        applicationId: auth.application.id,
+        messageId: message?.messageId,
+        requestId,
+        phone: mobile,
+        resultCode: 'ACCEPTED',
+        durationMs: Date.now() - startedAt,
+        payloadBytes: Buffer.byteLength(content, 'utf8'),
+        detail: { clientMessageId: input.clientMessageId },
+      });
       return response;
     } catch (error) {
       let outwardError = error;
@@ -213,6 +233,19 @@ export class OpenApiService implements OnModuleInit, OnModuleDestroy {
       }
       const failure = normalizeOpenApiFailure(outwardError);
       await this.prisma.openApiRequest.update({ where: { id: request.id }, data: { status: 'failed', httpStatus: failure.httpStatus, businessCode: failure.code, responseBody: failure.responseBody, durationMs: Date.now() - startedAt, completedAt: new Date() } });
+      this.protocolLogs?.record({
+        protocol: 'http',
+        direction: 'client_to_platform',
+        eventType: 'send_request',
+        status: 'failed',
+        tenantId: auth.application.tenantId,
+        applicationId: auth.application.id,
+        requestId,
+        phone: mobile,
+        resultCode: failure.code,
+        durationMs: Date.now() - startedAt,
+        payloadBytes: Buffer.byteLength(content, 'utf8'),
+      });
       throw outwardError;
     }
   }
@@ -326,6 +359,21 @@ export class OpenApiService implements OnModuleInit, OnModuleDestroy {
     const success = responseStatus !== undefined && responseStatus >= 200 && responseStatus < 300;
     const retryable = errorMessage !== undefined || responseStatus === 408 || responseStatus === 429 || (responseStatus !== undefined && responseStatus >= 500);
     await this.prisma.httpWebhookAttempt.create({ data: { deliveryId, attemptNo, responseStatus, responseSummary, errorMessage, durationMs: Date.now() - startedAt, requestHeaders: { 'x-event-id': delivery.event.eventId, 'x-event-type': delivery.event.eventType, 'x-timestamp': timestamp, 'x-signature': 'sha256=***' } } });
+    this.protocolLogs?.record({
+      protocol: 'http',
+      direction: 'platform_to_client',
+      eventType: `${delivery.event.eventType}_webhook`,
+      status: success ? 'success' : retryable ? 'retrying' : 'failed',
+      tenantId: delivery.event.tenantId,
+      applicationId: delivery.event.applicationId,
+      messageId: delivery.event.messageId,
+      requestId: delivery.event.eventId,
+      resultCode: responseStatus ?? 'NETWORK_ERROR',
+      durationMs: Date.now() - startedAt,
+      payloadBytes: Buffer.byteLength(body, 'utf8'),
+      retryCount: attemptNo - 1,
+      detail: { deliveryId, attemptNo, error: errorMessage },
+    });
     if (success) {
       await this.prisma.httpWebhookDelivery.update({ where: { id: deliveryId }, data: { status: 'delivered', attemptCount: attemptNo, lastHttpStatus: responseStatus, lastError: null, deliveredAt: new Date(), nextRetryAt: null } });
       return;

api/src/operations/admin-operations.controller.ts

@@ -4,6 +4,7 @@ import { CurrentSessionUserId } from '../auth/current-session-user.decorator';
 import { RequireRecentAuthentication } from '../auth/require-recent-authentication.decorator';
 import { SendChainService } from '../send-chain/send-chain.service';
 import { OperationsService } from './operations.service';
+import { ProtocolLogsService } from '../protocol-logs/protocol-logs.service';
 
 type DownloadResponse = {
   setHeader(name: string, value: number | string): void;
@@ -259,7 +260,24 @@ export class AdminOperationsController {
 @ApiTags('admin-system-logs')
 @Controller('admin/system-logs')
 export class AdminSystemLogsController {
-  constructor(private readonly operations: OperationsService) {}
+  constructor(
+    private readonly operations: OperationsService,
+    private readonly protocolLogs: ProtocolLogsService,
+  ) {}
+
+  @Get('protocol-interactions')
+  protocolInteractions(
+    @Query('protocol') protocol?: string,
+    @Query('direction') direction?: string,
+    @Query('eventType') eventType?: string,
+    @Query('status') status?: string,
+    @Query('keyword') keyword?: string,
+    @Query('range') range?: string,
+    @Query('page') page?: string,
+    @Query('pageSize') pageSize?: string,
+  ) {
+    return this.protocolLogs.list({ protocol, direction, eventType, status, keyword, range, page: Number(page), pageSize: Number(pageSize) });
+  }
 
   @Get()
   list(

api/src/protocol-logs/protocol-logs.module.ts

@@ -0,0 +1,9 @@
+import { Global, Module } from '@nestjs/common';
+import { ProtocolLogsService } from './protocol-logs.service';
+
+@Global()
+@Module({
+  providers: [ProtocolLogsService],
+  exports: [ProtocolLogsService],
+})
+export class ProtocolLogsModule {}

api/src/protocol-logs/protocol-logs.service.spec.ts

@@ -0,0 +1,68 @@
+import { ProtocolLogsService } from './protocol-logs.service';
+
+describe('ProtocolLogsService', () => {
+  const prisma = {
+    protocolInteractionLog: {
+      createMany: jest.fn(),
+      deleteMany: jest.fn(),
+      findMany: jest.fn(),
+      count: jest.fn(),
+      groupBy: jest.fn(),
+    },
+  };
+
+  beforeEach(() => {
+    jest.clearAllMocks();
+    prisma.protocolInteractionLog.createMany.mockResolvedValue({ count: 1 });
+    prisma.protocolInteractionLog.deleteMany.mockResolvedValue({ count: 0 });
+    prisma.protocolInteractionLog.findMany.mockResolvedValue([]);
+    prisma.protocolInteractionLog.count.mockResolvedValue(0);
+    prisma.protocolInteractionLog.groupBy.mockResolvedValue([]);
+  });
+
+  it('buffers a masked and secret-free business event', async () => {
+    const service = new ProtocolLogsService(prisma as never);
+    service.record({
+      protocol: 'cmpp',
+      direction: 'channel_to_platform',
+      eventType: 'deliver_receipt',
+      status: 'received',
+      phone: '18821203795',
+      gatewayMessageId: 123n,
+      detail: { sequenceId: 7, content: 'must not persist', authorization: 'secret' },
+    });
+
+    await service.onModuleDestroy();
+
+    expect(prisma.protocolInteractionLog.createMany).toHaveBeenCalledWith({
+      data: [expect.objectContaining({
+        phoneMasked: '188****3795',
+        gatewayMessageId: '123',
+        detail: { sequenceId: 7 },
+      })],
+    });
+  });
+
+  it('applies protocol, direction, keyword and time filters', async () => {
+    const service = new ProtocolLogsService(prisma as never);
+    await service.list({
+      protocol: 'cmpp',
+      direction: 'channel_to_platform',
+      keyword: 'MSG-1',
+      range: '7d',
+      page: 2,
+      pageSize: 20,
+    });
+
+    expect(prisma.protocolInteractionLog.findMany).toHaveBeenCalledWith(expect.objectContaining({
+      skip: 20,
+      take: 20,
+      where: expect.objectContaining({
+        protocol: 'cmpp',
+        direction: 'channel_to_platform',
+        createdAt: { gte: expect.any(Date) },
+        OR: expect.arrayContaining([{ messageId: { contains: 'MSG-1' } }]),
+      }),
+    }));
+  });
+});

api/src/protocol-logs/protocol-logs.service.ts

@@ -0,0 +1,189 @@
+import { Injectable, Logger, OnModuleDestroy, OnModuleInit } from '@nestjs/common';
+import { Prisma } from '@prisma/client';
+import { PrismaService } from '../prisma/prisma.service';
+
+export type ProtocolLogInput = {
+  protocol: 'cmpp' | 'http';
+  direction: 'client_to_platform' | 'platform_to_channel' | 'channel_to_platform' | 'platform_to_client';
+  eventType: string;
+  status: 'received' | 'accepted' | 'success' | 'failed' | 'retrying';
+  tenantId?: string | null;
+  applicationId?: string | null;
+  channelId?: string | null;
+  account?: string | null;
+  messageId?: string | null;
+  gatewayMessageId?: string | number | bigint | null;
+  traceId?: string | null;
+  requestId?: string | null;
+  phone?: string | null;
+  resultCode?: string | number | null;
+  durationMs?: number | null;
+  payloadBytes?: number | null;
+  retryCount?: number | null;
+  detail?: Record<string, unknown> | null;
+};
+
+export type ProtocolLogQuery = {
+  protocol?: string;
+  direction?: string;
+  eventType?: string;
+  status?: string;
+  keyword?: string;
+  range?: string;
+  page?: number;
+  pageSize?: number;
+};
+
+@Injectable()
+export class ProtocolLogsService implements OnModuleInit, OnModuleDestroy {
+  private readonly logger = new Logger(ProtocolLogsService.name);
+  private readonly buffer: Prisma.ProtocolInteractionLogCreateManyInput[] = [];
+  private flushTimer?: ReturnType<typeof setInterval>;
+  private retentionTimer?: ReturnType<typeof setInterval>;
+  private flushing = false;
+
+  constructor(private readonly prisma: PrismaService) {}
+
+  onModuleInit() {
+    this.flushTimer = setInterval(() => void this.flush(), positiveEnv('PROTOCOL_LOG_FLUSH_INTERVAL_MS', 500));
+    this.flushTimer.unref?.();
+    this.retentionTimer = setInterval(() => void this.purgeExpired(), positiveEnv('PROTOCOL_LOG_RETENTION_INTERVAL_MS', 86_400_000));
+    this.retentionTimer.unref?.();
+    setTimeout(() => void this.purgeExpired(), 30_000).unref?.();
+  }
+
+  async onModuleDestroy() {
+    if (this.flushTimer) clearInterval(this.flushTimer);
+    if (this.retentionTimer) clearInterval(this.retentionTimer);
+    await this.flush();
+  }
+
+  record(input: ProtocolLogInput) {
+    const maxQueue = positiveEnv('PROTOCOL_LOG_MAX_BUFFER', 10_000);
+    if (this.buffer.length >= maxQueue) {
+      this.logger.warn(`Protocol log buffer full; dropping event ${input.protocol}/${input.eventType}`);
+      return;
+    }
+    this.buffer.push({
+      protocol: input.protocol,
+      direction: input.direction,
+      eventType: clean(input.eventType, 64) ?? 'unknown',
+      status: input.status,
+      tenantId: clean(input.tenantId),
+      applicationId: clean(input.applicationId),
+      channelId: clean(input.channelId),
+      account: clean(input.account, 128),
+      messageId: clean(input.messageId, 128),
+      gatewayMessageId: input.gatewayMessageId == null ? null : String(input.gatewayMessageId).slice(0, 128),
+      traceId: clean(input.traceId, 128),
+      requestId: clean(input.requestId, 128),
+      phoneMasked: maskPhone(input.phone),
+      resultCode: input.resultCode == null ? null : String(input.resultCode).slice(0, 64),
+      durationMs: safeInteger(input.durationMs),
+      payloadBytes: safeInteger(input.payloadBytes),
+      retryCount: safeInteger(input.retryCount),
+      detail: sanitizeDetail(input.detail),
+    });
+    if (this.buffer.length >= positiveEnv('PROTOCOL_LOG_BATCH_SIZE', 100)) void this.flush();
+  }
+
+  async list(query: ProtocolLogQuery) {
+    const page = Math.max(1, Number(query.page) || 1);
+    const pageSize = Math.min(100, Math.max(1, Number(query.pageSize) || 20));
+    const where: Prisma.ProtocolInteractionLogWhereInput = {
+      protocol: selected(query.protocol),
+      direction: selected(query.direction),
+      eventType: selected(query.eventType),
+      status: selected(query.status),
+      createdAt: rangeWhere(query.range),
+      OR: query.keyword ? [
+        { messageId: { contains: query.keyword } },
+        { gatewayMessageId: { contains: query.keyword } },
+        { requestId: { contains: query.keyword } },
+        { traceId: { contains: query.keyword } },
+        { account: { contains: query.keyword } },
+        { phoneMasked: { contains: query.keyword } },
+        { resultCode: { contains: query.keyword } },
+      ] : undefined,
+    };
+    const [items, total, eventTypes] = await Promise.all([
+      this.prisma.protocolInteractionLog.findMany({
+        where,
+        orderBy: [{ createdAt: 'desc' }, { id: 'desc' }],
+        skip: (page - 1) * pageSize,
+        take: pageSize,
+      }),
+      this.prisma.protocolInteractionLog.count({ where }),
+      this.prisma.protocolInteractionLog.groupBy({ by: ['eventType'], orderBy: { eventType: 'asc' } }),
+    ]);
+    return { items, total, page, pageSize, eventTypes: eventTypes.map((item) => item.eventType) };
+  }
+
+  private async flush() {
+    if (this.flushing || this.buffer.length === 0) return;
+    this.flushing = true;
+    const batch = this.buffer.splice(0, positiveEnv('PROTOCOL_LOG_BATCH_SIZE', 100));
+    try {
+      await this.prisma.protocolInteractionLog.createMany({ data: batch });
+    } catch (error) {
+      this.logger.error(`Protocol log batch write failed (${batch.length} events)`, error instanceof Error ? error.stack : String(error));
+    } finally {
+      this.flushing = false;
+      if (this.buffer.length > 0) setImmediate(() => void this.flush());
+    }
+  }
+
+  private async purgeExpired() {
+    const retentionDays = positiveEnv('PROTOCOL_LOG_RETENTION_DAYS', 30);
+    try {
+      const result = await this.prisma.protocolInteractionLog.deleteMany({
+        where: { createdAt: { lt: new Date(Date.now() - retentionDays * 86_400_000) } },
+      });
+      if (result.count > 0) this.logger.log(`Purged ${result.count} protocol logs older than ${retentionDays} days`);
+    } catch (error) {
+      this.logger.error('Protocol log retention failed', error instanceof Error ? error.stack : String(error));
+    }
+  }
+}
+
+function clean(value: unknown, max = 191) {
+  const text = String(value ?? '').trim();
+  return text ? text.slice(0, max) : null;
+}
+
+function maskPhone(value: unknown) {
+  const text = String(value ?? '').replace(/\D/g, '');
+  if (!text) return null;
+  return text.length >= 7 ? `${text.slice(0, 3)}****${text.slice(-4)}` : `***${text.slice(-2)}`;
+}
+
+function safeInteger(value: unknown) {
+  const number = Number(value);
+  return Number.isSafeInteger(number) && number >= 0 ? number : null;
+}
+
+function selected(value?: string) {
+  return value && value !== 'all' ? value : undefined;
+}
+
+function positiveEnv(name: string, fallback: number) {
+  const value = Number(process.env[name]);
+  return Number.isFinite(value) && value > 0 ? Math.floor(value) : fallback;
+}
+
+function rangeWhere(range?: string): Prisma.DateTimeFilter | undefined {
+  const now = new Date();
+  if (range === 'today') return { gte: new Date(now.getFullYear(), now.getMonth(), now.getDate()) };
+  if (range === '7d') return { gte: new Date(now.getTime() - 7 * 86_400_000) };
+  if (range === '30d') return { gte: new Date(now.getTime() - 30 * 86_400_000) };
+  return undefined;
+}
+
+function sanitizeDetail(detail?: Record<string, unknown> | null): Prisma.InputJsonValue | undefined {
+  if (!detail) return undefined;
+  const blocked = /password|secret|token|signature|authorization|content|raw|body/i;
+  return Object.fromEntries(Object.entries(detail).filter(([key, value]) => !blocked.test(key) && value !== undefined).map(([key, value]) => [
+    key.slice(0, 64),
+    typeof value === 'string' ? value.slice(0, 500) : typeof value === 'number' || typeof value === 'boolean' || value === null ? value : String(value).slice(0, 500),
+  ])) as Prisma.InputJsonValue;
+}

api/src/send-chain/gateway-events.controller.ts

@@ -15,6 +15,7 @@ import {
   SendChainService,
 } from './send-chain.service';
 import { GatewayDownstreamConnectionEventDto, SmsConfigService } from '../sms-config/sms-config.service';
+import { ProtocolLogsService, type ProtocolLogInput } from '../protocol-logs/protocol-logs.service';
 
 @ApiTags('gateway-events')
 @Controller('gateway/events')
@@ -22,21 +23,22 @@ export class GatewayEventsController {
   constructor(
     private readonly sendChain: SendChainService,
     private readonly smsConfig: SmsConfigService,
+    private readonly protocolLogs: ProtocolLogsService,
   ) {}
 
   @Post('submit-result')
   submitResult(@Body() body: GatewaySubmitResultDto) {
-    return this.sendChain.handleSubmitResult(body);
+    return this.trackGatewayEvent('submit_resp', body, () => this.sendChain.handleSubmitResult(body));
   }
 
   @Post('receipt')
   receipt(@Body() body: GatewayReceiptEventDto) {
-    return this.sendChain.handleReceipt(body);
+    return this.trackGatewayEvent('deliver_receipt', body, () => this.sendChain.handleReceipt(body));
   }
 
   @Post('uplink')
   uplink(@Body() body: GatewayUplinkEventDto) {
-    return this.sendChain.handleUplink(body);
+    return this.trackGatewayEvent('deliver_uplink', body, () => this.sendChain.handleUplink(body));
   }
 
   @Post('dead-letter')
@@ -46,12 +48,12 @@ export class GatewayEventsController {
 
   @Post('inbound/authenticate')
   authenticateInbound(@Body() body: GatewayInboundAuthDto) {
-    return this.sendChain.authenticateInboundApplication(body);
+    return this.trackGatewayEvent('connect', body, () => this.sendChain.authenticateInboundApplication(body), 'client_to_platform');
   }
 
   @Post('inbound/submit')
   submitInbound(@Body() body: GatewayInboundSubmitDto) {
-    return this.sendChain.submitInboundMessage(body);
+    return this.trackGatewayEvent('submit', body, () => this.sendChain.submitInboundMessage(body), 'client_to_platform');
   }
 
   @Post('inbound/connection')
@@ -88,4 +90,46 @@ export class GatewayEventsController {
   downstreamRecoveryStatus(@Body() body: GatewayDownstreamRecoveryStatusDto) {
     return this.sendChain.recordGatewayDownstreamRecoveryStatus(body);
   }
+
+  private async trackGatewayEvent<T>(
+    eventType: string,
+    body: object,
+    action: () => Promise<T> | T,
+    direction: ProtocolLogInput['direction'] = 'channel_to_platform',
+  ) {
+    const startedAt = Date.now();
+    const value = body as Record<string, unknown>;
+    const common: Omit<ProtocolLogInput, 'status'> = {
+      protocol: 'cmpp',
+      direction,
+      eventType,
+      tenantId: value.tenantId as string,
+      applicationId: value.applicationId as string,
+      channelId: value.channelId as string,
+      account: (value.account ?? value.loginAccount) as string,
+      messageId: (value.messageId ?? value.platformMessageId) as string,
+      gatewayMessageId: (value.gatewayMessageId ?? value.msgId ?? value.upstreamMessageId) as string,
+      phone: (value.phoneNumber ?? value.srcTerminalId ?? value.destinationId) as string,
+      resultCode: (value.result ?? value.status ?? value.stat) as string,
+      detail: { sequenceId: value.sequenceId, connectionId: value.connectionId },
+    };
+    this.protocolLogs.record({ ...common, status: 'received', durationMs: 0 });
+    try {
+      const result = await action();
+      this.protocolLogs.record({
+        ...common,
+        status: 'success',
+        durationMs: Date.now() - startedAt,
+      });
+      return result;
+    } catch (error) {
+      this.protocolLogs.record({
+        ...common,
+        status: 'failed',
+        durationMs: Date.now() - startedAt,
+        detail: { error: error instanceof Error ? error.message : 'unknown error' },
+      });
+      throw error;
+    }
+  }
 }