KTeam/kt-template-online-api
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: 原子暂存STUN端口变更事件

Browse Source
This commit is contained in:
sunlei 2026-07-24 09:32:17 +08:00
parent 0517586c53
commit 01dac4a88b
6 changed files with 539 additions and 7 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

61
src/modules/admin/platform-config/network-management/network-agent-mqtt.service.ts
View File

@ -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.

70
src/modules/qqbot/core/application/message-push/system-message-event-stager.service.ts Normal file
View File

@ -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;
}
}

8
src/modules/qqbot/core/qqbot-core.module.ts
View File

@ -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,

265
test/admin/network-management/network-agent-mqtt.service.spec.ts
View File

@ -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<SystemMessageEventStager>;
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<NetworkPortForward>;
const historyRepository = {
create: (input) => Object.assign(new NetworkEndpointHistory(), input),
findOne: async ({ where }) =>
histories.find((item) => item.eventId === where.eventId) || null,
save: async (value) => {
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: historyFindOne,
save: historySave,
} as unknown as Repository<NetworkEndpointHistory>;
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<SystemMessageEventStager>;
let transactionCallCount = 0;
const dataSource = {
getRepository: manager.getRepository.bind(manager),
transaction: async (work) => {
transactionCallCount += 1;
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<string, unknown> = {}): 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> = {},
): 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<void> {
await new Promise<void>((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();

14
test/modules/qqbot/core/qqbot-core-module-contract.spec.ts
View File

@ -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', () => {

126
test/modules/qqbot/message-push/system-message-event-stager.spec.ts Normal file
View File

@ -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<string, unknown>) => ({
...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<string, unknown> = {}) {
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<Repository<QqbotMessageEvent>>;
const manager = {
getRepository: jest.fn((entity) => {
if (entity === QqbotMessageEvent) return repository;
throw new Error('unexpected repository');
}),
} as unknown as jest.Mocked<EntityManager>;
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);
});
});