From 01dac4a88be67b6839fbbefaf0b95fdedf8c1bd1 Mon Sep 17 00:00:00 2001 From: sunlei Date: Fri, 24 Jul 2026 09:32:17 +0800 Subject: [PATCH] =?UTF-8?q?feat:=20=E5=8E=9F=E5=AD=90=E6=9A=82=E5=AD=98STU?= =?UTF-8?q?N=E7=AB=AF=E5=8F=A3=E5=8F=98=E6=9B=B4=E4=BA=8B=E4=BB=B6?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../network-agent-mqtt.service.ts | 61 ++++ .../system-message-event-stager.service.ts | 70 +++++ src/modules/qqbot/core/qqbot-core.module.ts | 8 + .../network-agent-mqtt.service.spec.ts | 267 +++++++++++++++++- .../core/qqbot-core-module-contract.spec.ts | 14 + .../system-message-event-stager.spec.ts | 126 +++++++++ 6 files changed, 539 insertions(+), 7 deletions(-) create mode 100644 src/modules/qqbot/core/application/message-push/system-message-event-stager.service.ts create mode 100644 test/modules/qqbot/message-push/system-message-event-stager.spec.ts diff --git a/src/modules/admin/platform-config/network-management/network-agent-mqtt.service.ts b/src/modules/admin/platform-config/network-management/network-agent-mqtt.service.ts index 5a4fce1..415bde0 100644 --- a/src/modules/admin/platform-config/network-management/network-agent-mqtt.service.ts +++ b/src/modules/admin/platform-config/network-management/network-agent-mqtt.service.ts @@ -11,6 +11,10 @@ import { DataSource } from 'typeorm'; import * as mqtt from 'mqtt'; import type { IClientOptions, MqttClient } from 'mqtt'; import { KtDateTime } from '@/common'; +import { + SYSTEM_MESSAGE_EVENT_STAGER, + type SystemMessageEventStager, +} from '@/modules/qqbot/core/contract/message-push/qqbot-message-push.types'; import { NetworkAgentState } from './network-agent-state.entity'; import { NetworkEndpointHistory } from './network-endpoint-history.entity'; import { NetworkDdnsService } from './network-ddns.service'; @@ -60,6 +64,7 @@ export class NetworkAgentMqttService implements OnModuleInit, OnModuleDestroy { * @param configService - Runtime broker, Agent, and client identity settings. * @param dataSource - Transaction boundary for publish acknowledgements and inbound state. * @param eventStream - SSE fan-out notified only after accepted inbound commits. + * @param eventStager - Core-owned Outbox port that shares endpoint-history transactions. * @param clientFactory - Optional deterministic MQTT client factory used by tests. * @param ddnsService - Optional automatic-DDNS reconciler notified after address semantics commit. */ @@ -67,6 +72,8 @@ export class NetworkAgentMqttService implements OnModuleInit, OnModuleDestroy { private readonly configService: ConfigService, private readonly dataSource: DataSource, private readonly eventStream: NetworkManagementEventStreamService, + @Inject(SYSTEM_MESSAGE_EVENT_STAGER) + private readonly eventStager: SystemMessageEventStager, @Optional() @Inject(NETWORK_MQTT_CLIENT_FACTORY) private readonly clientFactory?: NetworkMqttClientFactory, @@ -654,6 +661,10 @@ export class NetworkAgentMqttService implements OnModuleInit, OnModuleDestroy { if (await repository.findOne({ where: { eventId: event.eventId } })) { return false; } + const previousHistory = await repository.findOne({ + order: { id: 'DESC', occurredAt: 'DESC' }, + where: { mappingId: event.mappingId }, + }); const history = repository.create({ eventId: event.eventId, eventType: event.type, @@ -667,6 +678,21 @@ export class NetworkAgentMqttService implements OnModuleInit, OnModuleDestroy { }); try { await repository.save(history); + if (this.shouldStagePortChange(event, previousHistory)) { + await this.eventStager.stage(manager, { + eventId: event.eventId, + occurredAt: event.occurredAt, + payload: { + changedAt: event.occurredAt, + currentPort: event.endpoint.publicPort, + portForwardId: event.mappingId, + previousPort: previousHistory.publicPort, + publicIpv4: event.endpoint.publicIpv4, + }, + resourceKey: event.mappingId, + sourceKey: 'network.stun.mapping-port-changed', + }); + } return true; } catch (error) { if (!this.isDuplicateKeyError(error)) throw error; @@ -675,6 +701,41 @@ export class NetworkAgentMqttService implements OnModuleInit, OnModuleDestroy { }); } + /** + * Determines whether adjacent endpoint-history rows prove a direct valid port transition. + * @param event - Current validated Agent endpoint event. + * @param previousHistory - Newest history row for the same mapping before this event. + * @returns Whether exactly one Outbox fact must be staged in the current transaction. + */ + private shouldStagePortChange( + event: NetworkEndpointEvent, + previousHistory: NetworkEndpointHistory | null, + ): boolean { + const previousPort = previousHistory?.publicPort; + const currentPort = event.endpoint.publicPort; + return ( + event.type === 'changed' && + previousHistory?.eventType !== 'withdrawn' && + this.isValidPort(previousPort) && + this.isValidPort(currentPort) && + previousPort !== currentPort + ); + } + + /** + * Checks whether a candidate is a concrete transport port suitable for a message payload. + * @param value - Persisted or incoming endpoint port. + * @returns Whether the value is an integer in the legal TCP/UDP port range. + */ + private isValidPort(value: unknown): value is number { + return ( + typeof value === 'number' && + Number.isInteger(value) && + value >= 1 && + value <= 65_535 + ); + } + /** * Serializes every report-owned mapping field that must be persisted. * @param mapping - Current persisted mapping before or after one report application. diff --git a/src/modules/qqbot/core/application/message-push/system-message-event-stager.service.ts b/src/modules/qqbot/core/application/message-push/system-message-event-stager.service.ts new file mode 100644 index 0000000..9861b65 --- /dev/null +++ b/src/modules/qqbot/core/application/message-push/system-message-event-stager.service.ts @@ -0,0 +1,70 @@ +import { Injectable } from '@nestjs/common'; +import type { EntityManager } from 'typeorm'; +import { KtDateTime } from '@/common'; +import { + type SystemMessageEventInput, + type SystemMessageEventStager, +} from '../../contract/message-push/qqbot-message-push.types'; +import { QqbotMessageEvent } from '../../infrastructure/persistence/message-push/qqbot-message-event.entity'; +import { SystemMessageSourceRegistry } from './system-message-source.registry'; + +/** Validates and atomically stages one producer-owned system-message Outbox fact. */ +@Injectable() +export class SystemMessageEventStagerService implements SystemMessageEventStager { + /** + * Creates the transaction-bound Outbox stager. + * @param sourceRegistry - Registry that owns source-specific event payload validation. + */ + constructor(private readonly sourceRegistry: SystemMessageSourceRegistry) {} + + /** + * Validates and inserts one Outbox fact through the caller's transaction manager. + * @param manager - Active caller-owned transaction manager; never replaced with a new connection. + * @param input - Producer event identity, occurrence time, source, resource, and scalar payload. + * @returns `accepted` for a new event or `duplicate` for the same persisted event ID. + */ + async stage( + manager: EntityManager, + input: SystemMessageEventInput, + ): Promise<'accepted' | 'duplicate'> { + const payload = this.sourceRegistry + .get(input.sourceKey) + .validateEventPayload(input.payload); + const repository = manager.getRepository(QqbotMessageEvent); + if (await repository.findOne({ where: { eventId: input.eventId } })) { + return 'duplicate'; + } + try { + await repository.save( + repository.create({ + eventId: input.eventId, + fanoutAttemptCount: 0, + fanoutLeaseUntil: null, + fanoutStatus: 'accepted', + lastErrorCode: null, + lastErrorMessage: null, + nextFanoutAt: new KtDateTime(input.occurredAt), + occurredAt: new KtDateTime(input.occurredAt), + payload, + resourceKey: input.resourceKey, + sourceKey: input.sourceKey, + }), + ); + return 'accepted'; + } catch (error) { + if (this.isDuplicateKeyError(error)) return 'duplicate'; + throw error; + } + } + + /** + * Recognizes only MySQL duplicate-key failures as idempotent races. + * @param error - Unknown persistence error caught from the caller's manager. + * @returns Whether the error is MySQL's duplicate key signal. + */ + private isDuplicateKeyError(error: unknown): boolean { + if (!error || typeof error !== 'object') return false; + const record = error as { code?: unknown; errno?: unknown }; + return record.code === 'ER_DUP_ENTRY' || record.errno === 1062; + } +} diff --git a/src/modules/qqbot/core/qqbot-core.module.ts b/src/modules/qqbot/core/qqbot-core.module.ts index 7b3e229..d4c6181 100644 --- a/src/modules/qqbot/core/qqbot-core.module.ts +++ b/src/modules/qqbot/core/qqbot-core.module.ts @@ -48,11 +48,13 @@ import { QqbotMessagePublishTarget } from '@/modules/qqbot/core/infrastructure/p import { QqbotMessageSubscription } from '@/modules/qqbot/core/infrastructure/persistence/message-push/qqbot-message-subscription.entity'; import { QqbotMessageTemplate } from '@/modules/qqbot/core/infrastructure/persistence/message-push/qqbot-message-template.entity'; import { SystemMessageSourceRegistry } from './application/message-push/system-message-source.registry'; +import { SystemMessageEventStagerService } from './application/message-push/system-message-event-stager.service'; import { SystemMessageTemplateRendererService } from './application/message-push/system-message-template-renderer.service'; import { QqbotMessageSubscriptionService } from './application/message-push/qqbot-message-subscription.service'; import { QqbotMessageTemplateService } from './application/message-push/qqbot-message-template.service'; import { QqbotAccountMessagePushService } from './application/message-push/qqbot-account-message-push.service'; import { QqbotMessageTargetOptionsService } from './application/message-push/qqbot-message-target-options.service'; +import { SYSTEM_MESSAGE_EVENT_STAGER } from './contract/message-push/qqbot-message-push.types'; export { QQBOT_CORE_DOMAIN_CONTRACT } from './contract/qqbot-core.contract'; @@ -89,6 +91,11 @@ export const QQBOT_CORE_CONTROLLERS = [ export const QQBOT_CORE_PROVIDERS = [ SystemMessageSourceRegistry, + SystemMessageEventStagerService, + { + provide: SYSTEM_MESSAGE_EVENT_STAGER, + useExisting: SystemMessageEventStagerService, + }, SystemMessageTemplateRendererService, QqbotMessageSubscriptionService, QqbotMessageTemplateService, @@ -114,6 +121,7 @@ export const QQBOT_CORE_PROVIDERS = [ ]; export const QQBOT_CORE_EXPORTS = [ + SYSTEM_MESSAGE_EVENT_STAGER, SystemMessageSourceRegistry, SystemMessageTemplateRendererService, QqbotMessageSubscriptionService, diff --git a/test/admin/network-management/network-agent-mqtt.service.spec.ts b/test/admin/network-management/network-agent-mqtt.service.spec.ts index 9d5cac0..f00187b 100644 --- a/test/admin/network-management/network-agent-mqtt.service.spec.ts +++ b/test/admin/network-management/network-agent-mqtt.service.spec.ts @@ -11,6 +11,10 @@ import { NetworkAgentState } from '../../../src/modules/admin/platform-config/ne import { NetworkEndpointHistory } from '../../../src/modules/admin/platform-config/network-management/network-endpoint-history.entity'; import type { NetworkManagementEventStreamService } from '../../../src/modules/admin/platform-config/network-management/network-management-event-stream.service'; import { NetworkPortForward } from '../../../src/modules/admin/platform-config/network-management/network-management.entity'; +import type { + SystemMessageEventInput, + SystemMessageEventStager, +} from '../../../src/modules/qqbot/core/contract/message-push/qqbot-message-push.types'; import { buildDesiredSnapshot, desiredSnapshotDigest, @@ -20,12 +24,16 @@ type MqttHarness = { client: MqttClient & EventEmitter; clientOptions: () => IClientOptions; histories: NetworkEndpointHistory[]; + historyFindOne: jest.Mock; + historySave: jest.Mock; mappingSave: jest.Mock; publishCommitted: jest.Mock; mapping: NetworkPortForward; publishCallback: () => (error?: Error) => void; requestDdnsReconcile: jest.Mock; service: NetworkAgentMqttService; + stagedEvents: SystemMessageEventInput[]; + stager: jest.Mocked; state: NetworkAgentState; stateSave: jest.Mock; transactionCalls: () => number; @@ -75,14 +83,31 @@ function createHarness(): MqttHarness { findOne: async ({ where }) => (where.id === mapping.id ? mapping : null), save: mappingSave, } as unknown as Repository; + const historyFindOne = jest.fn(async ({ order, where }) => { + if (where.eventId) { + return histories.find((item) => item.eventId === where.eventId) || null; + } + const matches = histories.filter( + (item) => item.mappingId === where.mappingId, + ); + if (!order) return matches[0] || null; + return ( + [...matches].sort((left, right) => { + const occurredAt = + right.occurredAt.getTime() - left.occurredAt.getTime(); + return occurredAt || String(right.id).localeCompare(String(left.id)); + })[0] || null + ); + }); + const historySave = jest.fn(async (value: NetworkEndpointHistory) => { + value.id ||= String(histories.length + 1); + histories.push(value); + return value; + }); const historyRepository = { create: (input) => Object.assign(new NetworkEndpointHistory(), input), - findOne: async ({ where }) => - histories.find((item) => item.eventId === where.eventId) || null, - save: async (value) => { - histories.push(value); - return value; - }, + findOne: historyFindOne, + save: historySave, } as unknown as Repository; const manager = { getRepository: (entity) => { @@ -92,12 +117,27 @@ function createHarness(): MqttHarness { throw new Error('unexpected repository'); }, } as unknown as EntityManager; + const stagedEvents: SystemMessageEventInput[] = []; + const stager = { + stage: jest.fn(async (_manager, input) => { + stagedEvents.push(input); + return 'accepted' as const; + }), + } as jest.Mocked; let transactionCallCount = 0; const dataSource = { getRepository: manager.getRepository.bind(manager), transaction: async (work) => { transactionCallCount += 1; - return await work(manager); + const historiesBefore = [...histories]; + const stagedEventsBefore = [...stagedEvents]; + try { + return await work(manager); + } catch (error) { + histories.splice(0, histories.length, ...historiesBefore); + stagedEvents.splice(0, stagedEvents.length, ...stagedEventsBefore); + throw error; + } }, } as unknown as DataSource; const configService = { @@ -143,6 +183,7 @@ function createHarness(): MqttHarness { configService, dataSource, eventStream, + stager, factory, { requestReconcile: requestDdnsReconcile } as never, ); @@ -150,18 +191,60 @@ function createHarness(): MqttHarness { client, clientOptions: () => options, histories, + historyFindOne, + historySave, mappingSave, mapping, publishCallback: () => publishAck, publishCommitted, requestDdnsReconcile, service, + stagedEvents, state, stateSave, + stager, transactionCalls: () => transactionCallCount, }; } +/** Builds one endpoint event payload with a valid STUN endpoint transition shape. */ +function endpointEvent(overrides: Record = {}): Buffer { + return Buffer.from( + JSON.stringify({ + agentId: 'nas-main', + endpoint: { + observedAt: '2026-07-22T01:02:04.000Z', + publicIpv4: '8.8.4.4', + publicPort: 38213, + validUntil: '2026-07-22T01:04:04.000Z', + }, + eventId: 'endpoint-event-2', + mappingId: '100', + occurredAt: '2026-07-22T01:02:05.000Z', + revision: 7, + schemaVersion: 1, + type: 'changed', + ...overrides, + }), + ); +} + +/** Builds a persisted endpoint-history fixture for direct prior-port comparisons. */ +function endpointHistory( + overrides: Partial = {}, +): NetworkEndpointHistory { + return Object.assign(new NetworkEndpointHistory(), { + eventId: 'endpoint-event-1', + eventType: 'changed', + id: '1', + mappingId: '100', + occurredAt: new KtDateTime('2026-07-22T01:02:03.000Z'), + publicIpv4: '8.8.8.8', + publicPort: 8213, + ...overrides, + }); +} + /** Waits for promise continuations scheduled by the MQTT bridge. */ async function flushPromises(): Promise { await new Promise((resolve) => setImmediate(resolve)); @@ -762,6 +845,176 @@ describe('NetworkAgentMqttService', () => { expect(harness.publishCommitted).toHaveBeenCalledWith('events'); }); + it('stages one fact for a direct valid port change with the same transaction manager', async () => { + const harness = createHarness(); + harness.histories.push(endpointHistory()); + const topic = 'kt/network/v1/agents/nas-main/events'; + + await harness.service.consumeMessage(topic, endpointEvent()); + + expect(harness.stager.stage).toHaveBeenCalledTimes(1); + expect(harness.stager.stage).toHaveBeenCalledWith( + expect.objectContaining({ getRepository: expect.any(Function) }), + { + eventId: 'endpoint-event-2', + occurredAt: '2026-07-22T01:02:05.000Z', + payload: { + changedAt: '2026-07-22T01:02:05.000Z', + currentPort: 38213, + portForwardId: '100', + previousPort: 8213, + publicIpv4: '8.8.4.4', + }, + resourceKey: '100', + sourceKey: 'network.stun.mapping-port-changed', + }, + ); + expect(harness.histories).toHaveLength(2); + expect(harness.stagedEvents).toHaveLength(1); + }); + + it.each(['published', 'withdrawn', 'restored'])( + 'does not stage a %s endpoint event', + async (type) => { + const harness = createHarness(); + harness.histories.push(endpointHistory()); + + await harness.service.consumeMessage( + 'kt/network/v1/agents/nas-main/events', + endpointEvent({ type }), + ); + + expect(harness.stager.stage).not.toHaveBeenCalled(); + }, + ); + + it.each([ + [ + 'an IP-only change', + endpointEvent({ + endpoint: { + observedAt: '2026-07-22T01:02:04.000Z', + publicIpv4: '1.1.1.1', + publicPort: 8213, + validUntil: '2026-07-22T01:04:04.000Z', + }, + }), + ], + ['a first event', endpointEvent()], + ])('does not stage %s', async (_name, payload) => { + const harness = createHarness(); + if (_name === 'an IP-only change') + harness.histories.push(endpointHistory()); + + await harness.service.consumeMessage( + 'kt/network/v1/agents/nas-main/events', + payload, + ); + + expect(harness.stager.stage).not.toHaveBeenCalled(); + }); + + it('does not stage a changed event after a withdrawn history row or invalid prior port', async () => { + const topic = 'kt/network/v1/agents/nas-main/events'; + const withdrawn = createHarness(); + withdrawn.histories.push(endpointHistory({ eventType: 'withdrawn' })); + await withdrawn.service.consumeMessage(topic, endpointEvent()); + expect(withdrawn.stager.stage).not.toHaveBeenCalled(); + + const invalidPort = createHarness(); + invalidPort.histories.push(endpointHistory({ publicPort: null })); + await invalidPort.service.consumeMessage(topic, endpointEvent()); + expect(invalidPort.stager.stage).not.toHaveBeenCalled(); + }); + + it('orders prior history by occurredAt then id before deciding the previous port', async () => { + const harness = createHarness(); + harness.histories.push( + endpointHistory({ id: '2', publicPort: 8213 }), + endpointHistory({ + eventId: 'endpoint-event-older', + id: '1', + publicPort: 38213, + }), + endpointHistory({ + eventId: 'endpoint-event-newest', + id: '3', + publicPort: 45000, + }), + ); + + await harness.service.consumeMessage( + 'kt/network/v1/agents/nas-main/events', + endpointEvent(), + ); + + expect(harness.historyFindOne).toHaveBeenCalledWith({ + order: { id: 'DESC', occurredAt: 'DESC' }, + where: { mappingId: '100' }, + }); + expect(harness.stagedEvents[0]?.payload).toMatchObject({ + previousPort: 45000, + }); + }); + + it('does not save or stage a duplicate endpoint event ID', async () => { + const harness = createHarness(); + harness.histories.push(endpointHistory({ eventId: 'endpoint-event-2' })); + + await harness.service.consumeMessage( + 'kt/network/v1/agents/nas-main/events', + endpointEvent(), + ); + + expect(harness.historySave).not.toHaveBeenCalled(); + expect(harness.stager.stage).not.toHaveBeenCalled(); + }); + + it('commits history when the stager reports an outbox duplicate', async () => { + const harness = createHarness(); + harness.histories.push(endpointHistory()); + harness.stager.stage.mockResolvedValueOnce('duplicate'); + + await harness.service.consumeMessage( + 'kt/network/v1/agents/nas-main/events', + endpointEvent(), + ); + + expect(harness.histories).toHaveLength(2); + }); + + it('rolls back history and staged events when staging fails', async () => { + const harness = createHarness(); + harness.histories.push(endpointHistory()); + harness.stager.stage.mockImplementationOnce(async (_manager, input) => { + harness.stagedEvents.push(input); + throw new Error('outbox unavailable'); + }); + + await expect( + harness.service.consumeMessage( + 'kt/network/v1/agents/nas-main/events', + endpointEvent(), + ), + ).rejects.toThrow('outbox unavailable'); + expect(harness.histories).toHaveLength(1); + expect(harness.stagedEvents).toHaveLength(0); + }); + + it('leaves no staged event when history persistence fails', async () => { + const harness = createHarness(); + harness.histories.push(endpointHistory()); + harness.historySave.mockRejectedValueOnce(new Error('history unavailable')); + + await expect( + harness.service.consumeMessage( + 'kt/network/v1/agents/nas-main/events', + endpointEvent(), + ), + ).rejects.toThrow('history unavailable'); + expect(harness.stagedEvents).toHaveLength(0); + }); + /** Proves liveness persistence does not become a periodic browser refresh. */ it('suppresses Admin events for status heartbeat-only timestamp renewal', async () => { const harness = createHarness(); diff --git a/test/modules/qqbot/core/qqbot-core-module-contract.spec.ts b/test/modules/qqbot/core/qqbot-core-module-contract.spec.ts index 5604cd1..a35a6ea 100644 --- a/test/modules/qqbot/core/qqbot-core-module-contract.spec.ts +++ b/test/modules/qqbot/core/qqbot-core-module-contract.spec.ts @@ -14,6 +14,8 @@ import { QqbotMessageController } from '../../../../src/modules/qqbot/core/contr import { QqbotPermissionController } from '../../../../src/modules/qqbot/core/contract/permission/qqbot-permission.controller'; import { QqbotRuleController } from '../../../../src/modules/qqbot/core/contract/rule/qqbot-rule.controller'; import { QqbotSendController } from '../../../../src/modules/qqbot/core/contract/send/qqbot-send.controller'; +import { SystemMessageEventStagerService } from '../../../../src/modules/qqbot/core/application/message-push/system-message-event-stager.service'; +import { SYSTEM_MESSAGE_EVENT_STAGER } from '../../../../src/modules/qqbot/core/contract/message-push/qqbot-message-push.types'; import { QQBOT_CORE_CONTROLLERS, QQBOT_CORE_ENTITIES, @@ -166,6 +168,18 @@ describe('QQBot core module contract', () => { expect(getModuleMetadata(QqbotCoreModule, MODULE_METADATA.EXPORTS)).toEqual( expect.arrayContaining(QQBOT_CORE_EXPORTS), ); + expect(QQBOT_CORE_PROVIDERS).toEqual( + expect.arrayContaining([ + SystemMessageEventStagerService, + { + provide: SYSTEM_MESSAGE_EVENT_STAGER, + useExisting: SystemMessageEventStagerService, + }, + ]), + ); + expect(QQBOT_CORE_EXPORTS).toEqual( + expect.arrayContaining([SYSTEM_MESSAGE_EVENT_STAGER]), + ); }); it('makes the legacy QQBot controllers, providers and entities explicit', () => { diff --git a/test/modules/qqbot/message-push/system-message-event-stager.spec.ts b/test/modules/qqbot/message-push/system-message-event-stager.spec.ts new file mode 100644 index 0000000..a88093f --- /dev/null +++ b/test/modules/qqbot/message-push/system-message-event-stager.spec.ts @@ -0,0 +1,126 @@ +import type { EntityManager, Repository } from 'typeorm'; +import { SystemMessageEventStagerService } from '../../../../src/modules/qqbot/core/application/message-push/system-message-event-stager.service'; +import { SystemMessageSourceRegistry } from '../../../../src/modules/qqbot/core/application/message-push/system-message-source.registry'; +import { QqbotMessageEvent } from '../../../../src/modules/qqbot/core/infrastructure/persistence/message-push/qqbot-message-event.entity'; + +const SOURCE_KEY = 'network.stun.mapping-port-changed'; + +/** Registers a source whose payload normalizer proves the stager uses the registry. */ +function createRegistry() { + const registry = new SystemMessageSourceRegistry(); + const validateEventPayload = jest.fn((payload: Record) => ({ + ...payload, + normalized: true, + })); + registry.register({ + definition: { + description: '测试来源', + displayName: '测试来源', + sourceKey: SOURCE_KEY, + subscriptionFields: [], + variables: [], + version: 1, + }, + inspectSubscription: jest.fn(), + listSubscriptionOptions: jest.fn(), + normalizeSubscriptionConfig: jest.fn(), + resolveDelivery: jest.fn(), + validateEventPayload, + }); + return { registry, validateEventPayload }; +} + +/** Builds one valid producer input for idempotent outbox staging tests. */ +function input(overrides: Record = {}) { + return { + eventId: 'endpoint-event-1', + occurredAt: '2026-07-24T00:00:00.000Z', + payload: { currentPort: 38213 }, + resourceKey: '100', + sourceKey: SOURCE_KEY, + ...overrides, + }; +} + +/** Creates a manager fake that exposes only the caller-owned outbox repository. */ +function createManager(events: QqbotMessageEvent[] = [], saveError?: unknown) { + const repository = { + create: jest.fn((value) => Object.assign(new QqbotMessageEvent(), value)), + findOne: jest.fn( + async ({ where: { eventId } }) => + events.find((event) => event.eventId === eventId) ?? null, + ), + save: jest.fn(async (event) => { + if (saveError) throw saveError; + events.push(event); + return event; + }), + } as unknown as jest.Mocked>; + const manager = { + getRepository: jest.fn((entity) => { + if (entity === QqbotMessageEvent) return repository; + throw new Error('unexpected repository'); + }), + } as unknown as jest.Mocked; + return { events, manager, repository }; +} + +describe('SystemMessageEventStagerService', () => { + it('validates, normalizes, and accepts a new event through the supplied manager', async () => { + const { registry, validateEventPayload } = createRegistry(); + const { events, manager, repository } = createManager(); + const service = new SystemMessageEventStagerService(registry); + + await expect(service.stage(manager, input())).resolves.toBe('accepted'); + + expect(validateEventPayload).toHaveBeenCalledWith({ currentPort: 38213 }); + expect(manager.getRepository).toHaveBeenCalledWith(QqbotMessageEvent); + expect(repository.save).toHaveBeenCalledWith( + expect.objectContaining({ + eventId: 'endpoint-event-1', + fanoutAttemptCount: 0, + fanoutLeaseUntil: null, + fanoutStatus: 'accepted', + lastErrorCode: null, + lastErrorMessage: null, + payload: { currentPort: 38213, normalized: true }, + resourceKey: '100', + sourceKey: SOURCE_KEY, + }), + ); + expect(events).toHaveLength(1); + expect(events[0].occurredAt).toBeInstanceOf(Date); + expect(events[0].nextFanoutAt).toBeInstanceOf(Date); + }); + + it('returns duplicate without saving when the event already exists', async () => { + const { registry } = createRegistry(); + const { manager, repository } = createManager([ + Object.assign(new QqbotMessageEvent(), { eventId: 'endpoint-event-1' }), + ]); + const service = new SystemMessageEventStagerService(registry); + + await expect(service.stage(manager, input())).resolves.toBe('duplicate'); + expect(repository.save).not.toHaveBeenCalled(); + }); + + it.each([{ code: 'ER_DUP_ENTRY' }, { errno: 1062 }])( + 'treats a MySQL save race with %o as duplicate', + async (error) => { + const { registry } = createRegistry(); + const { manager } = createManager([], error); + const service = new SystemMessageEventStagerService(registry); + + await expect(service.stage(manager, input())).resolves.toBe('duplicate'); + }, + ); + + it('rethrows a non-duplicate save error', async () => { + const { registry } = createRegistry(); + const failure = new Error('database unavailable'); + const { manager } = createManager([], failure); + const service = new SystemMessageEventStagerService(registry); + + await expect(service.stage(manager, input())).rejects.toBe(failure); + }); +});