From d2fd59dcd495e84a724550d17cf9e9f63bce383a Mon Sep 17 00:00:00 2001 From: sunlei Date: Fri, 24 Jul 2026 13:10:03 +0800 Subject: [PATCH] =?UTF-8?q?feat:=20=E6=8E=A5=E5=85=A5=E6=B6=88=E6=81=AF?= =?UTF-8?q?=E6=8A=95=E9=80=92=E5=8F=96=E6=B6=88=E4=B8=8E=E5=94=A4=E9=86=92?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../network-agent-mqtt.service.ts | 55 ++- .../network-ddns.service.ts | 25 +- .../account/qqbot-account.service.ts | 99 +++-- .../qqbot-account-message-push.service.ts | 51 ++- .../qqbot-message-subscription.service.ts | 39 +- ...em-message-delivery-coordinator.service.ts | 4 +- .../network-agent-mqtt.service.spec.ts | 121 +++++- .../network-ddns.service.spec.ts | 207 ++++++++- ...qqbot-account-message-push.service.spec.ts | 331 +++++++++++++++ ...qqbot-message-subscription.service.spec.ts | 216 ++++++++++ ...e-delivery-coordinator-integration.spec.ts | 328 +++++++++++++++ .../account/qqbot-account.service.spec.ts | 392 +++++++++++++++++- 12 files changed, 1812 insertions(+), 56 deletions(-) create mode 100644 test/modules/qqbot/message-push/system-message-delivery-coordinator-integration.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 ee80152..87361d4 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 @@ -12,7 +12,9 @@ import * as mqtt from 'mqtt'; import type { IClientOptions, MqttClient } from 'mqtt'; import { KtDateTime } from '@/common'; import { + SYSTEM_MESSAGE_DELIVERY_COORDINATOR, SYSTEM_MESSAGE_EVENT_STAGER, + type SystemMessageDeliveryCoordinator, type SystemMessageEventStager, } from '@/modules/qqbot/core/contract/message-push/qqbot-message-push.types'; import { NetworkAgentState } from './network-agent-state.entity'; @@ -65,6 +67,7 @@ export class NetworkAgentMqttService implements OnModuleInit, OnModuleDestroy { * @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 deliveryCoordinator - Core-owned durable delivery wake port used only after commit. * @param clientFactory - Optional deterministic MQTT client factory used by tests. * @param ddnsService - Optional automatic-DDNS reconciler notified after address semantics commit. */ @@ -74,6 +77,8 @@ export class NetworkAgentMqttService implements OnModuleInit, OnModuleDestroy { private readonly eventStream: NetworkManagementEventStreamService, @Inject(SYSTEM_MESSAGE_EVENT_STAGER) private readonly eventStager: SystemMessageEventStager, + @Inject(SYSTEM_MESSAGE_DELIVERY_COORDINATOR) + private readonly deliveryCoordinator: SystemMessageDeliveryCoordinator, @Optional() @Inject(NETWORK_MQTT_CLIENT_FACTORY) private readonly clientFactory?: NetworkMqttClientFactory, @@ -200,7 +205,17 @@ export class NetworkAgentMqttService implements OnModuleInit, OnModuleDestroy { ddnsSourceChanged = result.ddnsSourceChanged; } else if (topic === this.topic('events')) { source = 'events'; - changed = await this.appendEndpointEvent(parseEndpointEvent(parsed)); + const endpointResult = await this.appendEndpointEvent( + parseEndpointEvent(parsed), + ); + changed = endpointResult.changed; + if (endpointResult.deliveryAccepted) { + try { + this.deliveryCoordinator.requestDrain(); + } catch { + this.logger.warn('System message delivery wake failed'); + } + } } else { throw new NetworkMessageValidationError('Unexpected network MQTT topic'); } @@ -638,11 +653,11 @@ export class NetworkAgentMqttService implements OnModuleInit, OnModuleDestroy { /** * Appends one endpoint change event exactly once by event ID. * @param event - Strict endpoint transition parsed from the events topic. - * @returns True only when a new history row committed. + * @returns Committed history visibility plus whether the same transaction accepted Outbox work. */ private async appendEndpointEvent( event: NetworkEndpointEvent, - ): Promise { + ): Promise<{ changed: boolean; deliveryAccepted: boolean }> { this.assertAgentId(event.agentId); return await this.dataSource.transaction(async (manager) => { const state = await manager.getRepository(NetworkAgentState).findOne({ @@ -660,7 +675,7 @@ export class NetworkAgentMqttService implements OnModuleInit, OnModuleDestroy { } const repository = manager.getRepository(NetworkEndpointHistory); if (await repository.findOne({ where: { eventId: event.eventId } })) { - return false; + return { changed: false, deliveryAccepted: false }; } const previousHistory = await repository.findOne({ lock: { mode: 'pessimistic_read' }, @@ -680,25 +695,27 @@ export class NetworkAgentMqttService implements OnModuleInit, OnModuleDestroy { }); try { await repository.save(history); + let deliveryAccepted = false; 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', - }); + deliveryAccepted = + (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', + })) === 'accepted'; } - return true; + return { changed: true, deliveryAccepted }; } catch (error) { if (!this.isDuplicateKeyError(error)) throw error; - return false; + return { changed: false, deliveryAccepted: false }; } }); } diff --git a/src/modules/admin/platform-config/network-management/network-ddns.service.ts b/src/modules/admin/platform-config/network-management/network-ddns.service.ts index 2aeccef..1baff66 100644 --- a/src/modules/admin/platform-config/network-management/network-ddns.service.ts +++ b/src/modules/admin/platform-config/network-management/network-ddns.service.ts @@ -1,7 +1,9 @@ import { isIP } from 'node:net'; import { HttpStatus, + Inject, Injectable, + Logger, type OnModuleDestroy, type OnModuleInit, } from '@nestjs/common'; @@ -9,6 +11,10 @@ import { ConfigService } from '@nestjs/config'; import { InjectRepository } from '@nestjs/typeorm'; import { IsNull, Repository } from 'typeorm'; import { KtDateTime, throwVbenError } from '@/common'; +import { + SYSTEM_MESSAGE_DELIVERY_COORDINATOR, + type SystemMessageDeliveryCoordinator, +} from '@/modules/qqbot/core/contract/message-push/qqbot-message-push.types'; import { NetworkAgentState } from './network-agent-state.entity'; import { NetworkDdnsRecord } from './network-ddns.entity'; import { @@ -72,6 +78,7 @@ const PROVIDER_ERROR_CODES: Record = { */ @Injectable() export class NetworkDdnsService implements OnModuleInit, OnModuleDestroy { + private readonly logger = new Logger(NetworkDdnsService.name); private destroyed = false; private reconcileInterval?: NodeJS.Timeout; private reconcileRequestTimer?: NodeJS.Timeout; @@ -88,6 +95,7 @@ export class NetworkDdnsService implements OnModuleInit, OnModuleDestroy { * @param configService - Reconcile cadence and Agent identity configuration. * @param dnsPodClient - Redacted Tencent Cloud DNS provider boundary. * @param eventStream - Committed semantic-change notification stream. + * @param deliveryCoordinator - Core-owned wake port notified only after authoritative DDNS sync. */ constructor( @InjectRepository(NetworkDdnsRecord) @@ -99,6 +107,8 @@ export class NetworkDdnsService implements OnModuleInit, OnModuleDestroy { private readonly configService: ConfigService, private readonly dnsPodClient: NetworkDnsPodClient, private readonly eventStream: NetworkManagementEventStreamService, + @Inject(SYSTEM_MESSAGE_DELIVERY_COORDINATOR) + private readonly deliveryCoordinator: SystemMessageDeliveryCoordinator, ) {} /** @@ -593,11 +603,19 @@ export class NetworkDdnsService implements OnModuleInit, OnModuleDestroy { current.retryCount = 0; current.sourceAddress = targetAddress; current.syncStatus = 'synced'; - await this.saveReconcileState( + const saved = await this.saveReconcileState( current, beforeSynced, expectedProviderRecordId, ); + if (saved && current.syncStatus === 'synced' && current.appliedAddress) { + void this.deliveryCoordinator + .notifyDdnsSynced({ + appliedAddress: current.appliedAddress, + ddnsRecordId: String(current.id), + }) + .catch(() => this.loggerWarnDeliveryWake()); + } } /** @@ -1209,6 +1227,11 @@ export class NetworkDdnsService implements OnModuleInit, OnModuleDestroy { } } + /** Logs a non-authoritative delivery wake failure without changing DDNS commit success. */ + private loggerWarnDeliveryWake(): void { + this.logger.warn('System message delivery wake failed after DDNS sync'); + } + /** * Captures user-visible fields while excluding retry counts and timestamp-only churn. * @param record - Current persisted or pending entity state. diff --git a/src/modules/qqbot/core/application/account/qqbot-account.service.ts b/src/modules/qqbot/core/application/account/qqbot-account.service.ts index 7a51430..0dbe731 100644 --- a/src/modules/qqbot/core/application/account/qqbot-account.service.ts +++ b/src/modules/qqbot/core/application/account/qqbot-account.service.ts @@ -1,7 +1,7 @@ import { Inject, Injectable, Optional } from '@nestjs/common'; import { ConfigService } from '@nestjs/config'; import { InjectRepository } from '@nestjs/typeorm'; -import { Repository } from 'typeorm'; +import { In, Repository, type EntityManager } from 'typeorm'; import { SYSTEM_NOTICE_PUBLISHER, SystemNoticePublisher, @@ -15,6 +15,8 @@ import { } from './qqbot-account-napcat-runtime.port'; import { QqbotAccountAbility } from '../../infrastructure/persistence/account/qqbot-account-ability.entity'; import { QqbotAccount } from '../../infrastructure/persistence/account/qqbot-account.entity'; +import { QqbotMessageDelivery } from '../../infrastructure/persistence/message-push/qqbot-message-delivery.entity'; +import { QqbotMessagePublishBinding } from '../../infrastructure/persistence/message-push/qqbot-message-publish-binding.entity'; import type { QqbotAccountBodyDto, QqbotAccountQueryDto, @@ -402,13 +404,27 @@ export class QqbotAccountService { if (!body.accessToken) { delete payload.accessToken; } - await this.accountRepository.update({ id: body.id }, payload); - if (payload.selfId) { - await this.accountAbilityRepository.update( - { accountId: body.id }, - { selfId: payload.selfId }, - ); - } + await this.accountRepository.manager.transaction(async (manager) => { + const accounts = manager.getRepository(QqbotAccount); + const current = await accounts.findOne({ + lock: { mode: 'pessimistic_write' }, + where: { id: body.id, isDeleted: false }, + }); + if (!current) { + throwVbenError('QQBot 账号不存在或已删除'); + } + const selfIdChanged = + typeof payload.selfId === 'string' && payload.selfId !== current.selfId; + await accounts.update({ id: body.id }, payload); + if (selfIdChanged) { + await manager + .getRepository(QqbotAccountAbility) + .update({ accountId: body.id }, { selfId: payload.selfId! }); + } + if (selfIdChanged || payload.enabled === false) { + await this.cancelAccountDeliveries(manager, body.id); + } + }); return true; } @@ -432,22 +448,32 @@ export class QqbotAccountService { )) || { deletedContainers: 0, }; - await this.accountRepository.update( - { id }, - { - containerStatus: 'unknown', - connectStatus: 'offline', - enabled: false, - isDeleted: true, - oneBotStatus: 'offline', - qqLoginStatus: 'unknown', - webuiStatus: 'unknown', - }, - ); - await this.accountAbilityRepository.update( - { accountId: id }, - { isDeleted: true }, - ); + await this.accountRepository.manager.transaction(async (manager) => { + const accounts = manager.getRepository(QqbotAccount); + const current = await accounts.findOne({ + lock: { mode: 'pessimistic_write' }, + where: { id, isDeleted: false }, + }); + if (!current) { + throwVbenError('QQBot 账号不存在或已删除'); + } + await accounts.update( + { id }, + { + containerStatus: 'unknown', + connectStatus: 'offline', + enabled: false, + isDeleted: true, + oneBotStatus: 'offline', + qqLoginStatus: 'unknown', + webuiStatus: 'unknown', + }, + ); + await manager + .getRepository(QqbotAccountAbility) + .update({ accountId: id }, { isDeleted: true }); + await this.cancelAccountDeliveries(manager, id); + }); return { deletedContainers: containerResult.deletedContainers, }; @@ -478,6 +504,27 @@ export class QqbotAccountService { await this.accountRepository.update({ selfId }, payload); } + /** Cancels only claimable rows through an account's bindings in the caller transaction. */ + private async cancelAccountDeliveries( + manager: EntityManager, + accountId: string, + ): Promise { + const bindings = await manager + .getRepository(QqbotMessagePublishBinding) + .find({ + select: { id: true }, + where: { accountId: String(accountId) }, + }); + if (!bindings.length) return; + await manager.getRepository(QqbotMessageDelivery).update( + { + bindingId: In(bindings.map((binding) => binding.id)), + status: In(['waiting_ddns', 'pending', 'retry']), + }, + { nextAttemptAt: null, processingLeaseUntil: null, status: 'cancelled' }, + ); + } + /** * 执行 QQBot 核心流程。 * @param selfId - 账号 ID；定位本次读取、更新、删除或关联的账号。 @@ -634,9 +681,7 @@ export class QqbotAccountService { * Persists computed NapCat split statuses back to qqbot_account when the entity exposes the v3 status columns. * @param account - Enriched account list row; its `napcat` snapshot is the latest runtime evidence produced for Admin. */ - private async syncPersistedNapcatSplitStatus( - account: QqbotAccountListItem, - ) { + private async syncPersistedNapcatSplitStatus(account: QqbotAccountListItem) { if (!this.hasPersistedNapcatSplitStatus(account)) return; const payload = this.buildNapcatSplitStatusPayload(account); diff --git a/src/modules/qqbot/core/application/message-push/qqbot-account-message-push.service.ts b/src/modules/qqbot/core/application/message-push/qqbot-account-message-push.service.ts index 63d642e..8e6687c 100644 --- a/src/modules/qqbot/core/application/message-push/qqbot-account-message-push.service.ts +++ b/src/modules/qqbot/core/application/message-push/qqbot-account-message-push.service.ts @@ -1,7 +1,7 @@ import { HttpStatus, Injectable } from '@nestjs/common'; import { InjectRepository } from '@nestjs/typeorm'; import { throwVbenError } from '@/common'; -import { Repository, type EntityManager } from 'typeorm'; +import { In, Repository, type EntityManager } from 'typeorm'; import { SystemMessageContractError, type QqbotMessagePublishBindingInput, @@ -13,6 +13,7 @@ import { import { QqbotAccountService } from '../account/qqbot-account.service'; import { QqbotMessagePublishBinding } from '../../infrastructure/persistence/message-push/qqbot-message-publish-binding.entity'; import { QqbotMessagePublishTarget } from '../../infrastructure/persistence/message-push/qqbot-message-publish-target.entity'; +import { QqbotMessageDelivery } from '../../infrastructure/persistence/message-push/qqbot-message-delivery.entity'; import { QqbotMessageSubscription } from '../../infrastructure/persistence/message-push/qqbot-message-subscription.entity'; import { QqbotMessageTemplate } from '../../infrastructure/persistence/message-push/qqbot-message-template.entity'; import { QqbotMessageSubscriptionService } from './qqbot-message-subscription.service'; @@ -197,6 +198,14 @@ export class QqbotAccountMessagePushService { ); const saved = await bindings.save(current); await this.synchronizeTargets(manager, saved, normalizedTargets); + if ( + !saved.enabled || + snapshot!.subscriptionId !== saved.subscriptionId + ) { + await this.cancelUnfinishedDeliveries(manager, { + bindingId: saved.id, + }); + } return saved; }, ); @@ -240,7 +249,13 @@ export class QqbotAccountMessagePushService { ); this.assertStableBindingSnapshot(current, snapshot!); current.enabled = enabled; - return bindings.save(current); + const saved = await bindings.save(current); + if (!saved.enabled) { + await this.cancelUnfinishedDeliveries(manager, { + bindingId: saved.id, + }); + } + return saved; }, ); return this.toView(binding); @@ -267,6 +282,7 @@ export class QqbotAccountMessagePushService { binding.enabled = false; binding.isDeleted = true; await bindings.save(binding); + await this.cancelUnfinishedDeliveries(manager, { bindingId: binding.id }); }); return true; } @@ -362,6 +378,22 @@ export class QqbotAccountMessagePushService { saves.push(target); }); if (saves.length > 0) await targets.save(saves); + const removedIds = saves + .filter((target) => target.isDeleted) + .map((target) => target.id); + if (removedIds.length > 0) { + await manager.getRepository(QqbotMessageDelivery).update( + { + publishTargetId: In(removedIds), + status: In(['waiting_ddns', 'pending', 'retry']), + }, + { + nextAttemptAt: null, + processingLeaseUntil: null, + status: 'cancelled', + }, + ); + } } /** Builds a detached binding view using live dependency availability and active targets only. */ @@ -590,4 +622,19 @@ export class QqbotAccountMessagePushService { const value = error as { code?: unknown; errno?: unknown }; return value.code === 'ER_DUP_ENTRY' || value.errno === 1062; } + + /** Cancels only not-yet-processing deliveries in the active binding mutation transaction. */ + private async cancelUnfinishedDeliveries( + manager: EntityManager, + where: Pick, + ): Promise { + await manager.getRepository(QqbotMessageDelivery).update( + { ...where, status: In(['waiting_ddns', 'pending', 'retry']) }, + { + nextAttemptAt: null, + processingLeaseUntil: null, + status: 'cancelled', + }, + ); + } } diff --git a/src/modules/qqbot/core/application/message-push/qqbot-message-subscription.service.ts b/src/modules/qqbot/core/application/message-push/qqbot-message-subscription.service.ts index 7cb15a6..d5c91c9 100644 --- a/src/modules/qqbot/core/application/message-push/qqbot-message-subscription.service.ts +++ b/src/modules/qqbot/core/application/message-push/qqbot-message-subscription.service.ts @@ -4,6 +4,7 @@ import { InjectRepository } from '@nestjs/typeorm'; import { throwVbenError } from '@/common'; import { Like, + In, Repository, type EntityManager, type FindOptionsWhere, @@ -16,6 +17,7 @@ import { } from '../../contract/message-push/qqbot-message-push.types'; import { QqbotMessageSubscription } from '../../infrastructure/persistence/message-push/qqbot-message-subscription.entity'; import { QqbotMessagePublishBinding } from '../../infrastructure/persistence/message-push/qqbot-message-publish-binding.entity'; +import { QqbotMessageDelivery } from '../../infrastructure/persistence/message-push/qqbot-message-delivery.entity'; import { SystemMessageSourceRegistry } from './system-message-source.registry'; const DEFAULT_PAGE_NO = 1; @@ -164,8 +166,17 @@ export class QqbotMessageSubscriptionService { if (conflict && conflict.id !== current.id) { this.throwNaturalKeyConflict(); } + const sourceIdentityChanged = + current.sourceKey !== normalized.sourceKey || + current.sourceConfigDigest !== normalized.sourceConfigDigest; Object.assign(current, normalized); - return repository.save(current); + const saved = await repository.save(current); + if (!saved.enabled || sourceIdentityChanged) { + await this.cancelUnfinishedDeliveries(manager, { + subscriptionId: saved.id, + }); + } + return saved; }, ); return this.toView(saved); @@ -202,7 +213,13 @@ export class QqbotMessageSubscriptionService { } } current.enabled = enabled; - return repository.save(current); + const saved = await repository.save(current); + if (!saved.enabled) { + await this.cancelUnfinishedDeliveries(manager, { + subscriptionId: saved.id, + }); + } + return saved; }, ); return this.toView(saved); @@ -224,6 +241,9 @@ export class QqbotMessageSubscriptionService { current.enabled = false; current.isDeleted = true; await repository.save(current); + await this.cancelUnfinishedDeliveries(manager, { + subscriptionId: current.id, + }); return true; }, ); @@ -345,6 +365,21 @@ export class QqbotMessageSubscriptionService { } } + /** Cancels only claimable historical deliveries in the caller's configuration transaction. */ + private async cancelUnfinishedDeliveries( + manager: EntityManager, + where: Pick, + ): Promise { + await manager.getRepository(QqbotMessageDelivery).update( + { ...where, status: In(['waiting_ddns', 'pending', 'retry']) }, + { + status: 'cancelled', + nextAttemptAt: null, + processingLeaseUntil: null, + }, + ); + } + /** * Maps duplicate natural keys to the existing Vben-compatible HTTP conflict response. * @returns Never returns because it always throws an HTTP 409 exception. diff --git a/src/modules/qqbot/core/application/message-push/system-message-delivery-coordinator.service.ts b/src/modules/qqbot/core/application/message-push/system-message-delivery-coordinator.service.ts index 17e2cdf..995599f 100644 --- a/src/modules/qqbot/core/application/message-push/system-message-delivery-coordinator.service.ts +++ b/src/modules/qqbot/core/application/message-push/system-message-delivery-coordinator.service.ts @@ -91,8 +91,10 @@ export class SystemMessageDeliveryCoordinatorService }): Promise { if ( this.destroyed || + typeof input.appliedAddress !== 'string' || isIP(input.appliedAddress) !== 4 || - !input.ddnsRecordId + typeof input.ddnsRecordId !== 'string' || + input.ddnsRecordId.length === 0 ) return; const advanced = await this.dataSource.transaction(async (manager) => { 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 ca2d7a5..61172f2 100644 --- a/test/admin/network-management/network-agent-mqtt.service.spec.ts +++ b/test/admin/network-management/network-agent-mqtt.service.spec.ts @@ -23,10 +23,12 @@ import { type MqttHarness = { client: MqttClient & EventEmitter; clientOptions: () => IClientOptions; + deliveryCoordinator: { requestDrain: jest.Mock }; histories: NetworkEndpointHistory[]; historyFindOne: jest.Mock; historySave: jest.Mock; mappingSave: jest.Mock; + operations: string[]; publishCommitted: jest.Mock; mapping: NetworkPortForward; publishCallback: () => (error?: Error) => void; @@ -64,6 +66,7 @@ function createDeferred(): Deferred { /** Creates a fake MQTT client plus in-memory TypeORM state for bridge tests. */ function createHarness(): MqttHarness { + const operations: string[] = []; const state = Object.assign(new NetworkAgentState(), { agentId: 'nas-main', appliedRevision: '0', @@ -150,6 +153,7 @@ function createHarness(): MqttHarness { const stagedEvents: SystemMessageEventInput[] = []; const stager = { stage: jest.fn(async (_manager, input) => { + operations.push('stager:accepted'); stagedEvents.push(input); return 'accepted' as const; }), @@ -159,13 +163,17 @@ function createHarness(): MqttHarness { getRepository: manager.getRepository.bind(manager), transaction: async (work) => { transactionCallCount += 1; + operations.push('transaction:start'); const historiesBefore = [...histories]; const stagedEventsBefore = [...stagedEvents]; try { - return await work(manager); + const result = await work(manager); + operations.push('transaction:commit'); + return result; } catch (error) { histories.splice(0, histories.length, ...historiesBefore); stagedEvents.splice(0, stagedEvents.length, ...stagedEventsBefore); + operations.push('transaction:rollback'); throw error; } }, @@ -209,22 +217,31 @@ function createHarness(): MqttHarness { publishCommitted, } as unknown as NetworkManagementEventStreamService; const requestDdnsReconcile = jest.fn(); + const deliveryCoordinator = { + notifyDdnsSynced: jest.fn().mockResolvedValue(undefined), + requestDrain: jest.fn(() => { + operations.push('delivery:wake'); + }), + }; const service = new NetworkAgentMqttService( configService, dataSource, eventStream, stager, + deliveryCoordinator, factory, { requestReconcile: requestDdnsReconcile } as never, ); return { client, clientOptions: () => options, + deliveryCoordinator, histories, historyFindOne, historySave, mappingSave, mapping, + operations, publishCallback: () => publishAck, publishCommitted, requestDdnsReconcile, @@ -389,6 +406,10 @@ function createConcurrentEndpointHarness(): ConcurrentEndpointHarness { dataSource, { publishCommitted: jest.fn() } as never, stager, + { + notifyDdnsSynced: jest.fn().mockResolvedValue(undefined), + requestDrain: jest.fn(), + }, ); return { historyStageEntered: firstStageEntered.promise, @@ -1064,6 +1085,52 @@ describe('NetworkAgentMqttService', () => { ); expect(harness.histories).toHaveLength(2); expect(harness.stagedEvents).toHaveLength(1); + expect(harness.deliveryCoordinator.requestDrain).toHaveBeenCalledTimes(1); + expect(harness.operations).toEqual([ + 'transaction:start', + 'stager:accepted', + 'transaction:commit', + 'delivery:wake', + ]); + }); + + it('wakes accepted work only after commit and never for duplicate or non-port transitions', async () => { + const accepted = createHarness(); + accepted.histories.push(endpointHistory()); + await accepted.service.consumeMessage( + 'kt/network/v1/agents/nas-main/events', + endpointEvent(), + ); + expect(accepted.operations.indexOf('transaction:commit')).toBeLessThan( + accepted.operations.indexOf('delivery:wake'), + ); + + const duplicate = createHarness(); + duplicate.histories.push(endpointHistory()); + duplicate.stager.stage.mockImplementationOnce(async () => { + duplicate.operations.push('stager:duplicate'); + return 'duplicate'; + }); + await duplicate.service.consumeMessage( + 'kt/network/v1/agents/nas-main/events', + endpointEvent(), + ); + expect(duplicate.deliveryCoordinator.requestDrain).not.toHaveBeenCalled(); + + const nonPort = createHarness(); + nonPort.histories.push(endpointHistory()); + await nonPort.service.consumeMessage( + 'kt/network/v1/agents/nas-main/events', + endpointEvent({ + endpoint: { + observedAt: '2026-07-22T01:02:04.000Z', + publicIpv4: '8.8.4.4', + publicPort: 8213, + validUntil: '2026-07-22T01:04:04.000Z', + }, + }), + ); + expect(nonPort.deliveryCoordinator.requestDrain).not.toHaveBeenCalled(); }); it.each(['published', 'withdrawn', 'restored'])( @@ -1252,6 +1319,58 @@ describe('NetworkAgentMqttService', () => { ).rejects.toThrow('outbox unavailable'); expect(harness.histories).toHaveLength(1); expect(harness.stagedEvents).toHaveLength(0); + expect(harness.deliveryCoordinator.requestDrain).not.toHaveBeenCalled(); + expect(harness.operations).toContain('transaction:rollback'); + expect(harness.operations).not.toContain('transaction:commit'); + }); + + it('keeps a committed event ACK-successful when the post-commit wake throws synchronously', async () => { + const harness = createHarness(); + harness.histories.push(endpointHistory()); + harness.deliveryCoordinator.requestDrain.mockImplementationOnce(() => { + harness.operations.push('delivery:wake-throw'); + throw new Error('wake failed'); + }); + const loggerWarn = jest + .spyOn((harness.service as any).logger, 'warn') + .mockImplementation(); + const callback = jest.fn(); + + await (harness.service as any).acknowledgeIncoming( + 'kt/network/v1/agents/nas-main/events', + endpointEvent(), + callback, + ); + + expect(harness.histories).toHaveLength(2); + expect(harness.stagedEvents).toHaveLength(1); + expect(harness.operations.indexOf('transaction:commit')).toBeLessThan( + harness.operations.indexOf('delivery:wake-throw'), + ); + expect(callback).toHaveBeenCalledWith(0); + expect(loggerWarn).toHaveBeenCalledWith( + 'System message delivery wake failed', + ); + }); + + it('keeps staging failures ACK-negative and never wakes after rollback', async () => { + const harness = createHarness(); + harness.histories.push(endpointHistory()); + harness.stager.stage.mockRejectedValueOnce(new Error('outbox unavailable')); + const callback = jest.fn(); + + await (harness.service as any).acknowledgeIncoming( + 'kt/network/v1/agents/nas-main/events', + endpointEvent(), + callback, + ); + + expect(callback).toHaveBeenCalledWith(expect.any(Error)); + expect(callback.mock.calls[0][0]).toMatchObject({ + message: 'outbox unavailable', + }); + expect(harness.deliveryCoordinator.requestDrain).not.toHaveBeenCalled(); + expect(harness.operations).toContain('transaction:rollback'); }); it('leaves no staged event when history persistence fails', async () => { diff --git a/test/admin/network-management/network-ddns.service.spec.ts b/test/admin/network-management/network-ddns.service.spec.ts index 9a87455..19849c0 100644 --- a/test/admin/network-management/network-ddns.service.spec.ts +++ b/test/admin/network-management/network-ddns.service.spec.ts @@ -14,6 +14,7 @@ import { NetworkPortForward } from '../../../src/modules/admin/platform-config/n type Harness = { client: jest.Mocked>; + deliveryCoordinator: { notifyDdnsSynced: jest.Mock }; mapping: NetworkPortForward; recordUpdate: jest.Mock; records: NetworkDdnsRecord[]; @@ -161,6 +162,10 @@ function createHarness(): Harness { const eventStream = { publishCommitted: jest.fn(), } as unknown as NetworkManagementEventStreamService; + const deliveryCoordinator = { + notifyDdnsSynced: jest.fn().mockResolvedValue(undefined), + requestDrain: jest.fn(), + }; const service = new NetworkDdnsService( recordRepository, mappingRepository, @@ -168,8 +173,32 @@ function createHarness(): Harness { config, client as unknown as NetworkDnsPodClient, eventStream, + deliveryCoordinator, ); - return { client, mapping, recordUpdate, records, service, state }; + return { + client, + deliveryCoordinator, + mapping, + recordUpdate, + records, + service, + state, + }; +} + +/** Persists one enabled pending IPv4 DDNS row for reconcile-path assertions. */ +async function prepareEnabledA(harness: Harness): Promise { + await harness.service.create({ + domain: 'kwitsukasa.top', + enabled: false, + name: 'Pal A', + portForwardId: '100', + recordType: 'A', + sourceType: 'port_forward_ipv4', + subDomain: 'pal', + }); + harness.records[0].enabled = true; + harness.records[0].syncStatus = 'pending'; } /** Creates the fluent list query subset used by the service. */ @@ -347,6 +376,181 @@ describe('NetworkDdnsService', () => { 'synced', 'synced', ]); + expect(harness.deliveryCoordinator.notifyDdnsSynced).toHaveBeenCalledTimes( + 2, + ); + expect(harness.deliveryCoordinator.notifyDdnsSynced).toHaveBeenCalledWith( + expect.objectContaining({ + appliedAddress: '8.8.8.8', + ddnsRecordId: '200', + }), + ); + }); + + it('notifies exactly once only after the final guarded synced update commits', async () => { + const harness = createHarness(); + let providerStarted!: () => void; + let resolveProvider!: (result: { + appliedAddress: string; + changed: boolean; + providerRecordId: string; + }) => void; + const started = new Promise((resolve) => { + providerStarted = resolve; + }); + const providerResult = new Promise<{ + appliedAddress: string; + changed: boolean; + providerRecordId: string; + }>((resolve) => { + resolveProvider = resolve; + }); + harness.client.reconcile.mockImplementation(async () => { + providerStarted(); + return providerResult; + }); + await prepareEnabledA(harness); + + const reconcile = harness.service.reconcileNow('200', true); + await started; + expect(harness.deliveryCoordinator.notifyDdnsSynced).not.toHaveBeenCalled(); + resolveProvider({ + appliedAddress: '8.8.8.8', + changed: true, + providerRecordId: '300', + }); + await reconcile; + await Promise.resolve(); + + expect(harness.recordUpdate).toHaveBeenCalledTimes(2); + expect(harness.records[0]).toMatchObject({ + appliedAddress: '8.8.8.8', + syncStatus: 'synced', + }); + expect(harness.deliveryCoordinator.notifyDdnsSynced).toHaveBeenCalledTimes( + 1, + ); + expect(harness.deliveryCoordinator.notifyDdnsSynced).toHaveBeenCalledWith({ + appliedAddress: '8.8.8.8', + ddnsRecordId: '200', + }); + expect(harness.recordUpdate.mock.invocationCallOrder[1]).toBeLessThan( + harness.deliveryCoordinator.notifyDdnsSynced.mock.invocationCallOrder[0], + ); + }); + + it('does not notify when the final guarded synced update affects zero rows', async () => { + const harness = createHarness(); + const update = harness.recordUpdate.getMockImplementation()!; + harness.recordUpdate + .mockImplementationOnce(update) + .mockResolvedValueOnce({ affected: 0, generatedMaps: [], raw: [] }); + harness.client.reconcile.mockResolvedValue({ + appliedAddress: '8.8.8.8', + changed: true, + providerRecordId: '300', + }); + await prepareEnabledA(harness); + + await harness.service.reconcileNow('200', true); + await Promise.resolve(); + + expect(harness.recordUpdate).toHaveBeenCalledTimes(2); + expect(harness.deliveryCoordinator.notifyDdnsSynced).not.toHaveBeenCalled(); + expect(harness.records[0].syncStatus).toBe('syncing'); + }); + + it('does not notify on pre-provider stale CAS, provider failure, disabled reread, or synced no-op', async () => { + const stale = createHarness(); + stale.recordUpdate.mockResolvedValueOnce({ + affected: 0, + generatedMaps: [], + raw: [], + }); + await prepareEnabledA(stale); + await stale.service.reconcileNow('200', true); + expect(stale.client.reconcile).not.toHaveBeenCalled(); + expect(stale.deliveryCoordinator.notifyDdnsSynced).not.toHaveBeenCalled(); + + const failed = createHarness(); + failed.client.reconcile.mockRejectedValueOnce( + new NetworkDnsPodClientError('DNSPOD_RATE_LIMITED', 'rate limited', true), + ); + await prepareEnabledA(failed); + await failed.service.reconcileNow('200', true); + expect(failed.deliveryCoordinator.notifyDdnsSynced).not.toHaveBeenCalled(); + + const disabled = createHarness(); + let providerStarted!: () => void; + let resolveProvider!: (result: { + appliedAddress: string; + changed: boolean; + providerRecordId: string; + }) => void; + const started = new Promise((resolve) => { + providerStarted = resolve; + }); + disabled.client.reconcile.mockImplementation( + () => + new Promise((resolve) => { + resolveProvider = resolve; + providerStarted(); + }), + ); + await prepareEnabledA(disabled); + const disabledRun = disabled.service.reconcileNow('200', true); + await started; + disabled.records[0].enabled = false; + resolveProvider({ + appliedAddress: '8.8.8.8', + changed: true, + providerRecordId: '300', + }); + await disabledRun; + expect( + disabled.deliveryCoordinator.notifyDdnsSynced, + ).not.toHaveBeenCalled(); + + const synced = createHarness(); + await prepareEnabledA(synced); + synced.records[0].appliedAddress = '8.8.8.8'; + synced.records[0].providerRecordId = '300'; + synced.records[0].sourceAddress = '8.8.8.8'; + synced.records[0].syncStatus = 'synced'; + await synced.service.reconcileNow('200'); + expect(synced.client.reconcile).not.toHaveBeenCalled(); + expect(synced.deliveryCoordinator.notifyDdnsSynced).not.toHaveBeenCalled(); + }); + + it('logs notification rejection without changing authoritative DDNS success', async () => { + const harness = createHarness(); + const wakeFailure = new Error('wake unavailable'); + harness.deliveryCoordinator.notifyDdnsSynced.mockRejectedValueOnce( + wakeFailure, + ); + const loggerWarn = jest + .spyOn((harness.service as any).logger, 'warn') + .mockImplementation(); + harness.client.reconcile.mockResolvedValue({ + appliedAddress: '8.8.8.8', + changed: true, + providerRecordId: '300', + }); + await prepareEnabledA(harness); + + await expect( + harness.service.reconcileNow('200', true), + ).resolves.toBeUndefined(); + await Promise.resolve(); + + expect(harness.records[0]).toMatchObject({ + appliedAddress: '8.8.8.8', + providerRecordId: '300', + syncStatus: 'synced', + }); + expect(loggerWarn).toHaveBeenCalledWith( + 'System message delivery wake failed after DDNS sync', + ); }); it('waits without calling DNSPod when the IPv6 source is stale', async () => { @@ -373,6 +577,7 @@ describe('NetworkDdnsService', () => { syncStatus: 'waiting_source', }); expect(harness.client.reconcile).not.toHaveBeenCalled(); + expect(harness.deliveryCoordinator.notifyDdnsSynced).not.toHaveBeenCalled(); }); it('waits without calling the provider when an IPv4 source keeps an ineligible residual lease', async () => { diff --git a/test/modules/qqbot/message-push/qqbot-account-message-push.service.spec.ts b/test/modules/qqbot/message-push/qqbot-account-message-push.service.spec.ts index e29c7a1..7e98b1d 100644 --- a/test/modules/qqbot/message-push/qqbot-account-message-push.service.spec.ts +++ b/test/modules/qqbot/message-push/qqbot-account-message-push.service.spec.ts @@ -3,6 +3,7 @@ import { QqbotAccountMessagePushService } from '../../../../src/modules/qqbot/co import { SystemMessageContractError } from '../../../../src/modules/qqbot/core/contract/message-push/qqbot-message-push.types'; import { QqbotMessagePublishBinding } from '../../../../src/modules/qqbot/core/infrastructure/persistence/message-push/qqbot-message-publish-binding.entity'; import { QqbotMessagePublishTarget } from '../../../../src/modules/qqbot/core/infrastructure/persistence/message-push/qqbot-message-publish-target.entity'; +import { QqbotMessageDelivery } from '../../../../src/modules/qqbot/core/infrastructure/persistence/message-push/qqbot-message-delivery.entity'; const ACCOUNT = { id: '2041700000000000000', @@ -150,6 +151,9 @@ function setupBindingFixture(options: BindingFixtureOptions = {}) { getRepository: jest.fn((entity) => { if (entity === QqbotMessagePublishBinding) return bindingStore; if (entity === QqbotMessagePublishTarget) return targetStore; + if (entity === QqbotMessageDelivery) { + return { update: jest.fn(async () => ({ affected: 0 })) }; + } throw new Error('unexpected repository'); }), }; @@ -250,6 +254,333 @@ describe('QqbotAccountMessagePushService', () => { ).toThrow(new SystemMessageContractError('invalid_target_type')); }); + describe('binding and target delivery cancellation transaction', () => { + /** Builds isolated binding/target/delivery drafts and commits them only on transaction success. */ + function cancellationHarness() { + const subscription = { + enabled: true, + id: '2041700000000000001', + isDeleted: false, + name: '订阅', + sourceConfig: {}, + sourceKey: SOURCE_KEY, + }; + const template = { + content: '正文', + enabled: true, + id: '2041700000000000002', + isDeleted: false, + name: '模板', + sourceKey: SOURCE_KEY, + }; + const bindings = [bindingRow()]; + const targets = [ + targetRow({ id: 'target-retained' }), + targetRow({ + id: 'target-removed', + targetId: '30000000000000001', + targetType: 'private', + }), + ]; + const deliveries = [ + ...['waiting_ddns', 'pending', 'retry', 'processing', 'success'].map( + (status) => ({ + bindingId: bindings[0].id, + frozen: `binding-${status}`, + id: `binding-${status}`, + nextAttemptAt: `schedule-${status}`, + processingLeaseUntil: `lease-${status}`, + publishTargetId: 'target-retained', + status, + }), + ), + { + bindingId: bindings[0].id, + frozen: 'removed', + id: 'removed-target-delivery', + nextAttemptAt: 'schedule-removed', + processingLeaseUntil: 'lease-removed', + publishTargetId: 'target-removed', + status: 'pending', + }, + { + bindingId: 'other-binding', + frozen: 'other-binding', + id: 'other-binding-delivery', + nextAttemptAt: 'schedule-other', + processingLeaseUntil: 'lease-other', + publishTargetId: 'other-target', + status: 'pending', + }, + ]; + const deliveryUpdates: Array> = []; + let failCancellation = false; + let nextTargetId = 1; + const transaction = jest.fn( + async (work: (manager: any) => Promise) => { + const draftBindings = structuredClone(bindings); + const draftTargets = structuredClone(targets); + const draftDeliveries = structuredClone(deliveries); + const bindingStore = { + create: (value: Record) => bindingRow(value), + findOne: jest.fn( + async ({ where }: { where: Record }) => + structuredClone( + draftBindings.find((row) => + Object.entries(where).every( + ([key, value]) => row[key] === value, + ), + ) ?? null, + ), + ), + save: jest.fn(async (row: Record) => { + const index = draftBindings.findIndex( + (candidate) => candidate.id === row.id, + ); + if (index >= 0) draftBindings[index] = structuredClone(row); + else draftBindings.push(structuredClone(row)); + return row; + }), + }; + const targetStore = { + create: (value: Record) => + targetRow({ id: `target-new-${nextTargetId++}`, ...value }), + find: jest.fn( + async ({ where }: { where: Record }) => + structuredClone( + draftTargets.filter((row) => + Object.entries(where).every( + ([key, value]) => row[key] === value, + ), + ), + ), + ), + save: jest.fn(async (rows: Array>) => { + rows.forEach((row) => { + const index = draftTargets.findIndex( + (candidate) => candidate.id === row.id, + ); + if (index >= 0) draftTargets[index] = structuredClone(row); + else draftTargets.push(structuredClone(row)); + }); + return rows; + }), + }; + const manager = { + getRepository: (entity: unknown) => { + if (entity === QqbotMessagePublishBinding) return bindingStore; + if (entity === QqbotMessagePublishTarget) return targetStore; + if (entity === QqbotMessageDelivery) { + return { + update: async ( + where: Record, + patch: Record, + ) => { + deliveryUpdates.push(where); + const statuses = where.status._value as string[]; + const targetIds = where.publishTargetId?._value as + | string[] + | undefined; + let affected = 0; + draftDeliveries.forEach((row) => { + const identityMatches = + (where.bindingId && + row.bindingId === where.bindingId) || + (targetIds && + targetIds.includes(String(row.publishTargetId))); + if (identityMatches && statuses.includes(row.status)) { + Object.assign(row, patch); + affected += 1; + } + }); + if (failCancellation) { + throw new Error('delivery cancellation failed'); + } + return { affected }; + }, + }; + } + throw new Error('unexpected cancellation repository'); + }, + }; + const result = await work(manager); + bindings.splice(0, bindings.length, ...draftBindings); + targets.splice(0, targets.length, ...draftTargets); + deliveries.splice(0, deliveries.length, ...draftDeliveries); + return result; + }, + ); + const outerBindingRepository = { + find: async () => bindings, + manager: { transaction }, + }; + const service = new QqbotAccountMessagePushService( + outerBindingRepository as never, + { + find: async ({ where }: { where: Record }) => + targets.filter((row) => + Object.entries(where).every(([key, value]) => row[key] === value), + ), + } as never, + { findOne: async () => subscription } as never, + { findOne: async () => template } as never, + { findBySelfId: async () => ACCOUNT } as never, + { + requireAvailableForBinding: async ( + _manager: unknown, + subscriptionId: string, + ) => ({ ...subscription, id: subscriptionId }), + } as never, + { + requireAvailableForBinding: async ( + _manager: unknown, + templateId: string, + ) => ({ ...template, id: templateId }), + } as never, + { + get: () => ({ + definition: { displayName: 'STUN 映射', variables: [] }, + inspectSubscription: async () => ({ valid: true }), + }), + } as never, + { validate: jest.fn() } as never, + ); + return { + bindings, + deliveries, + deliveryUpdates, + failCancellation: () => { + failCancellation = true; + }, + service, + targets, + }; + } + + it.each([ + [ + 'disable', + async (service: QqbotAccountMessagePushService) => + service.setBindingEnabled(ACCOUNT.selfId, bindingRow().id, false), + ], + [ + 'remove', + async (service: QqbotAccountMessagePushService) => + service.removeBinding(ACCOUNT.selfId, bindingRow().id), + ], + [ + 'subscription change', + async (service: QqbotAccountMessagePushService) => + service.updateBinding(ACCOUNT.selfId, bindingRow().id, { + enabled: true, + subscriptionId: '2041700000000000099', + targets: [ + { targetId: '20000000000000001', targetType: 'group' }, + { targetId: '30000000000000001', targetType: 'private' }, + ], + templateId: bindingRow().templateId, + }), + ], + ])( + '%s cancels only exact binding unfinished rows', + async (_name, mutate) => { + const harness = cancellationHarness(); + const before = structuredClone(harness.deliveries); + + await mutate(harness.service); + + expect(harness.deliveries).toEqual( + before.map((delivery) => + delivery.bindingId === bindingRow().id && + ['waiting_ddns', 'pending', 'retry'].includes(delivery.status) + ? { + ...delivery, + nextAttemptAt: null, + processingLeaseUntil: null, + status: 'cancelled', + } + : delivery, + ), + ); + }, + ); + + it('keeps frozen deliveries on template-only switch', async () => { + const harness = cancellationHarness(); + const before = structuredClone(harness.deliveries); + + await harness.service.updateBinding(ACCOUNT.selfId, bindingRow().id, { + enabled: true, + subscriptionId: bindingRow().subscriptionId, + targets: [ + { targetId: '20000000000000001', targetType: 'group' }, + { targetId: '30000000000000001', targetType: 'private' }, + ], + templateId: '2041700000000000098', + }); + + expect(harness.deliveries).toEqual(before); + expect(harness.deliveryUpdates).toHaveLength(0); + }); + + it('target synchronization cancels only removed target IDs and retains new/sibling scopes', async () => { + const harness = cancellationHarness(); + const before = structuredClone(harness.deliveries); + + await harness.service.updateBinding(ACCOUNT.selfId, bindingRow().id, { + enabled: true, + subscriptionId: bindingRow().subscriptionId, + targets: [ + { targetId: '20000000000000001', targetType: 'group' }, + { targetId: '40000000000000001', targetType: 'private' }, + ], + templateId: bindingRow().templateId, + }); + + expect(harness.deliveries).toEqual( + before.map((delivery) => + delivery.id === 'removed-target-delivery' + ? { + ...delivery, + nextAttemptAt: null, + processingLeaseUntil: null, + status: 'cancelled', + } + : delivery, + ), + ); + expect( + harness.targets.find((target) => target.id === 'target-retained'), + ).toMatchObject({ enabled: true, isDeleted: false }); + expect( + harness.targets.find((target) => target.id === 'target-new-1'), + ).toMatchObject({ enabled: true, isDeleted: false }); + }); + + it('rolls back binding, target, and cancellation drafts on a cancellation write failure', async () => { + const harness = cancellationHarness(); + const before = { + bindings: structuredClone(harness.bindings), + deliveries: structuredClone(harness.deliveries), + targets: structuredClone(harness.targets), + }; + harness.failCancellation(); + + await expect( + harness.service.updateBinding(ACCOUNT.selfId, bindingRow().id, { + enabled: false, + subscriptionId: bindingRow().subscriptionId, + targets: [{ targetId: '20000000000000001', targetType: 'group' }], + templateId: bindingRow().templateId, + }), + ).rejects.toThrow('delivery cancellation failed'); + + expect(harness.bindings).toEqual(before.bindings); + expect(harness.targets).toEqual(before.targets); + expect(harness.deliveries).toEqual(before.deliveries); + }); + }); + it('holds subscription then template gates through multi-target persistence without coercing IDs', async () => { const events: string[] = []; const targetRows: any[] = []; diff --git a/test/modules/qqbot/message-push/qqbot-message-subscription.service.spec.ts b/test/modules/qqbot/message-push/qqbot-message-subscription.service.spec.ts index b3497ec..72bb9b6 100644 --- a/test/modules/qqbot/message-push/qqbot-message-subscription.service.spec.ts +++ b/test/modules/qqbot/message-push/qqbot-message-subscription.service.spec.ts @@ -13,6 +13,7 @@ import { NetworkDdnsRecord } from '../../../../src/modules/admin/platform-config import { NetworkPortForward } from '../../../../src/modules/admin/platform-config/network-management/network-management.entity'; import { QqbotMessageSubscription } from '../../../../src/modules/qqbot/core/infrastructure/persistence/message-push/qqbot-message-subscription.entity'; import { QqbotMessagePublishBinding } from '../../../../src/modules/qqbot/core/infrastructure/persistence/message-push/qqbot-message-publish-binding.entity'; +import { QqbotMessageDelivery } from '../../../../src/modules/qqbot/core/infrastructure/persistence/message-push/qqbot-message-delivery.entity'; const SOURCE_KEY = 'network.stun.mapping-port-changed'; const NOW = new KtDateTime('2026-07-24 08:09:10'); @@ -211,6 +212,9 @@ function setup( getRepository: jest.fn((entity) => { if (entity === QqbotMessageSubscription) return subscriptionRepository; if (entity === QqbotMessagePublishBinding) return bindingRepository; + if (entity === QqbotMessageDelivery) { + return { update: jest.fn(async () => ({ affected: 0 })) }; + } throw new Error('unexpected repository'); }), } as unknown as jest.Mocked; @@ -741,6 +745,215 @@ describe('QqbotMessageSubscriptionService', () => { } }); + describe('delivery cancellation transaction', () => { + /** + * Builds authoritative subscription/delivery stores whose transaction drafts commit only + * when the complete lifecycle callback resolves. + */ + function cancellationHarness() { + const subscriptions = [subscription()]; + const deliveries = [ + ...[ + 'waiting_ddns', + 'pending', + 'retry', + 'processing', + 'success', + 'failed', + 'superseded', + 'cancelled', + ].map((status, index) => ({ + frozen: `exact-${status}`, + id: `delivery-${index}`, + nextAttemptAt: `schedule-${status}`, + processingLeaseUntil: `lease-${status}`, + status, + subscriptionId: '100', + })), + { + frozen: 'other', + id: 'delivery-other', + nextAttemptAt: 'schedule-other', + processingLeaseUntil: 'lease-other', + status: 'pending', + subscriptionId: '101', + }, + ]; + let failCancellation = false; + const deliveryUpdates: Array> = []; + const transaction = jest.fn( + async (work: (manager: EntityManager) => Promise) => { + const draftSubscriptions = subscriptions.map((row) => ({ + ...row, + sourceConfig: { ...row.sourceConfig }, + })) as QqbotMessageSubscription[]; + const draftDeliveries = structuredClone(deliveries); + const subscriptionStore = { + findOne: jest.fn( + async ({ where }: { where: Record }) => + draftSubscriptions.find((row) => + Object.entries(where).every( + ([key, value]) => row[key] === value, + ), + ) ?? null, + ), + save: jest.fn(async (row: QqbotMessageSubscription) => row), + }; + const manager = { + getRepository: jest.fn((entity) => { + if (entity === QqbotMessageSubscription) return subscriptionStore; + if (entity === QqbotMessagePublishBinding) { + return { count: jest.fn().mockResolvedValue(0) }; + } + if (entity === QqbotMessageDelivery) { + return { + update: jest.fn( + async ( + where: Record, + patch: Record, + ) => { + deliveryUpdates.push(where); + const statuses = (where.status as { _value?: unknown[] }) + ._value; + let affected = 0; + draftDeliveries.forEach((row) => { + if ( + row.subscriptionId === where.subscriptionId && + statuses?.includes(row.status) + ) { + Object.assign(row, patch); + affected += 1; + } + }); + if (failCancellation) { + throw new Error('cancellation persistence failed'); + } + return { affected }; + }, + ), + }; + } + throw new Error('unexpected cancellation repository'); + }), + } as unknown as EntityManager; + const result = await work(manager); + subscriptions.splice(0, subscriptions.length, ...draftSubscriptions); + deliveries.splice(0, deliveries.length, ...draftDeliveries); + return result; + }, + ); + const source = adapter(); + source.normalizeSubscriptionConfig.mockImplementation(async (input) => { + const canonicalConfig = input as Record; + return { + canonicalConfig, + resourceKey: canonicalConfig.portForwardId, + sourceSummary: 'transaction fixture', + }; + }); + const service = new QqbotMessageSubscriptionService( + { + manager: { transaction }, + } as unknown as Repository, + registry(source), + ); + return { + deliveries, + deliveryUpdates, + failCancellation: () => { + failCancellation = true; + }, + service, + subscriptions, + }; + } + + it.each([ + [ + 'disable', + async (service: QqbotMessageSubscriptionService) => + service.setEnabled('100', false), + ], + [ + 'soft delete', + async (service: QqbotMessageSubscriptionService) => + service.remove('100'), + ], + [ + 'canonical config change', + async (service: QqbotMessageSubscriptionService) => + service.update('100', { + enabled: true, + name: '端口提醒', + remark: null, + sourceConfig: { + ddnsRecordId: '2041700000000000099', + portForwardId: CONFIG.portForwardId, + }, + sourceKey: SOURCE_KEY, + }), + ], + ])( + '%s commits config and cancels only exact claimable rows without rewriting snapshots', + async (_name, mutate) => { + const harness = cancellationHarness(); + const before = structuredClone(harness.deliveries); + + await mutate(harness.service); + + expect(harness.deliveries).toEqual( + before.map((delivery) => + delivery.subscriptionId === '100' && + ['waiting_ddns', 'pending', 'retry'].includes(delivery.status) + ? { + ...delivery, + nextAttemptAt: null, + processingLeaseUntil: null, + status: 'cancelled', + } + : delivery, + ), + ); + expect( + (harness.deliveryUpdates[0].status as { _value: string[] })._value, + ).toEqual(['waiting_ddns', 'pending', 'retry']); + }, + ); + + it('does not cancel for name/remark-only edits or the same canonical identity', async () => { + const harness = cancellationHarness(); + const before = structuredClone(harness.deliveries); + + await harness.service.update('100', { + enabled: true, + name: '新名称', + remark: '新备注', + sourceConfig: { ...CONFIG }, + sourceKey: SOURCE_KEY, + }); + + expect(harness.deliveries).toEqual(before); + expect(harness.deliveryUpdates).toHaveLength(0); + }); + + it('rolls back both subscription mutation and draft cancellations when cancellation persistence fails', async () => { + const harness = cancellationHarness(); + const subscriptionsBefore = harness.subscriptions.map((row) => ({ + ...row, + sourceConfig: { ...row.sourceConfig }, + })); + const deliveriesBefore = structuredClone(harness.deliveries); + harness.failCancellation(); + + await expect(harness.service.setEnabled('100', false)).rejects.toThrow( + 'cancellation persistence failed', + ); + + expect(harness.subscriptions).toEqual(subscriptionsBefore); + expect(harness.deliveries).toEqual(deliveriesBefore); + }); + }); + it('pages only undeleted records with real filters and returns detached, real-time source status', async () => { const source = adapter({ invalidReasonCode: 'ddns_mapping_mismatch', @@ -975,6 +1188,9 @@ describe('QqbotMessageSubscriptionService', () => { if (entity === QqbotMessagePublishBinding) { return { count: jest.fn(async () => 0) }; } + if (entity === QqbotMessageDelivery) { + return { update: jest.fn(async () => ({ affected: 0 })) }; + } throw new Error('unexpected repository'); }), } as unknown as EntityManager; diff --git a/test/modules/qqbot/message-push/system-message-delivery-coordinator-integration.spec.ts b/test/modules/qqbot/message-push/system-message-delivery-coordinator-integration.spec.ts new file mode 100644 index 0000000..bda829d --- /dev/null +++ b/test/modules/qqbot/message-push/system-message-delivery-coordinator-integration.spec.ts @@ -0,0 +1,328 @@ +import type { DataSource, EntityManager } from 'typeorm'; +import { KtDateTime } from '../../../../src/common'; +import { SystemMessageDeliveryCoordinatorService } from '../../../../src/modules/qqbot/core/application/message-push/system-message-delivery-coordinator.service'; +import { QqbotMessageDelivery } from '../../../../src/modules/qqbot/core/infrastructure/persistence/message-push/qqbot-message-delivery.entity'; +import { QqbotMessageEvent } from '../../../../src/modules/qqbot/core/infrastructure/persistence/message-push/qqbot-message-event.entity'; +import { QqbotMessageSubscription } from '../../../../src/modules/qqbot/core/infrastructure/persistence/message-push/qqbot-message-subscription.entity'; + +const SOURCE_KEY = 'network.stun.mapping-port-changed'; +const NOW = new Date('2026-07-24T03:04:05.000Z'); + +type CoordinatorStore = { + deliveries: Array>; + events: Array>; + subscriptions: Array>; +}; + +type CoordinatorHarness = { + coordinator: SystemMessageDeliveryCoordinatorService; + failUpdate: (error: Error | null) => void; + operations: string[]; + requestDrain: jest.SpyInstance; + store: CoordinatorStore; + transaction: jest.Mock; +}; + +/** Deep-clones the JSON-shaped coordinator fixture while preserving project timestamps. */ +function cloneStore(store: CoordinatorStore): CoordinatorStore { + return structuredClone(store); +} + +/** Reads the value carried by TypeORM's in-memory `In` find operator. */ +function inValues(value: unknown): unknown[] { + if (!value || typeof value !== 'object') return []; + const values = (value as { _value?: unknown })._value; + return Array.isArray(values) ? values : []; +} + +/** Builds a transaction-faithful Core harness with commit-on-resolve and discard-on-reject. */ +function createCoordinatorHarness(): CoordinatorHarness { + const store: CoordinatorStore = { + deliveries: [ + { + id: 'delivery-match', + messageEventId: 'event-match', + nextAttemptAt: 'future-match', + status: 'waiting_ddns', + subscriptionId: 'subscription-match', + }, + { + id: 'delivery-old-address', + messageEventId: 'event-old-address', + nextAttemptAt: 'future-old', + status: 'waiting_ddns', + subscriptionId: 'subscription-match', + }, + { + id: 'delivery-pending', + messageEventId: 'event-match', + nextAttemptAt: 'future-pending', + status: 'pending', + subscriptionId: 'subscription-match', + }, + { + id: 'delivery-inactive-subscription', + messageEventId: 'event-match', + nextAttemptAt: 'future-inactive', + status: 'waiting_ddns', + subscriptionId: 'subscription-disabled', + }, + { + id: 'delivery-numeric-config', + messageEventId: 'event-match', + nextAttemptAt: 'future-numeric', + status: 'waiting_ddns', + subscriptionId: 'subscription-numeric', + }, + { + id: 'delivery-other-source', + messageEventId: 'event-other-source', + nextAttemptAt: 'future-source', + status: 'waiting_ddns', + subscriptionId: 'subscription-match', + }, + ], + events: [ + { + id: 'event-match', + payload: { publicIpv4: '8.8.8.8' }, + sourceKey: SOURCE_KEY, + }, + { + id: 'event-old-address', + payload: { publicIpv4: '1.1.1.1' }, + sourceKey: SOURCE_KEY, + }, + { + id: 'event-other-source', + payload: { publicIpv4: '8.8.8.8' }, + sourceKey: 'other.source', + }, + { + id: 'event-numeric-address', + payload: { publicIpv4: 8_888 }, + sourceKey: SOURCE_KEY, + }, + ], + subscriptions: [ + { + enabled: true, + id: 'subscription-match', + isDeleted: false, + sourceConfig: { ddnsRecordId: '9007199254740993' }, + sourceKey: SOURCE_KEY, + }, + { + enabled: false, + id: 'subscription-disabled', + isDeleted: false, + sourceConfig: { ddnsRecordId: '9007199254740993' }, + sourceKey: SOURCE_KEY, + }, + { + enabled: true, + id: 'subscription-numeric', + isDeleted: false, + sourceConfig: { ddnsRecordId: 9_007_199_254_740_993 }, + sourceKey: SOURCE_KEY, + }, + { + enabled: true, + id: 'subscription-wrong', + isDeleted: false, + sourceConfig: { ddnsRecordId: 'other' }, + sourceKey: SOURCE_KEY, + }, + { + enabled: true, + id: 'subscription-deleted', + isDeleted: true, + sourceConfig: { ddnsRecordId: '9007199254740993' }, + sourceKey: SOURCE_KEY, + }, + ], + }; + const operations: string[] = []; + let updateFailure: Error | null = null; + const transaction = jest.fn( + async (work: (manager: EntityManager) => Promise) => { + operations.push('transaction:start'); + const draft = cloneStore(store); + const manager = { + getRepository: (entity: unknown) => { + if (entity === QqbotMessageSubscription) { + return { + find: async ({ where }: { where: Record }) => + draft.subscriptions.filter((row) => + Object.entries(where).every( + ([key, value]) => row[key] === value, + ), + ), + }; + } + if (entity === QqbotMessageEvent) { + return { + find: async ({ where }: { where: Record }) => + draft.events.filter((row) => + Object.entries(where).every( + ([key, value]) => row[key] === value, + ), + ), + }; + } + if (entity === QqbotMessageDelivery) { + return { + update: async ( + where: Record, + patch: Record, + ) => { + operations.push('delivery:update'); + if (updateFailure) throw updateFailure; + const eventIds = inValues(where.messageEventId); + const subscriptionIds = inValues(where.subscriptionId); + let affected = 0; + draft.deliveries.forEach((delivery) => { + if ( + eventIds.includes(delivery.messageEventId) && + subscriptionIds.includes(delivery.subscriptionId) && + (where.status === undefined || + delivery.status === where.status) + ) { + Object.assign(delivery, patch); + affected += 1; + } + }); + return { affected }; + }, + }; + } + throw new Error('unexpected coordinator repository'); + }, + } as unknown as EntityManager; + const result = await work(manager); + store.deliveries = draft.deliveries; + store.events = draft.events; + store.subscriptions = draft.subscriptions; + operations.push('transaction:commit'); + return result; + }, + ); + const coordinator = new SystemMessageDeliveryCoordinatorService( + { transaction } as unknown as DataSource, + { runOnce: jest.fn().mockResolvedValue(0) } as never, + { runOnce: jest.fn().mockResolvedValue(0) } as never, + ); + const requestDrain = jest + .spyOn(coordinator, 'requestDrain') + .mockImplementation(() => { + operations.push('drain'); + }); + return { + coordinator, + failUpdate: (error) => { + updateFailure = error; + }, + operations, + requestDrain, + store, + transaction, + }; +} + +describe('SystemMessageDeliveryCoordinatorService DDNS integration', () => { + beforeEach(() => { + jest.useFakeTimers().setSystemTime(NOW); + }); + + afterEach(() => { + jest.useRealTimers(); + }); + + it('requires an exact non-empty string DDNS ID and IPv4 address', async () => { + const harness = createCoordinatorHarness(); + + for (const input of [ + { appliedAddress: '8.8.8.8', ddnsRecordId: 9_007_199_254_740_993 }, + { appliedAddress: '8.8.8.8', ddnsRecordId: '' }, + { appliedAddress: 'not-an-ip', ddnsRecordId: '9007199254740993' }, + { appliedAddress: '::1', ddnsRecordId: '9007199254740993' }, + { appliedAddress: 8_888, ddnsRecordId: '9007199254740993' }, + ]) { + await harness.coordinator.notifyDdnsSynced(input as never); + } + + expect(harness.transaction).not.toHaveBeenCalled(); + expect(harness.requestDrain).not.toHaveBeenCalled(); + }); + + it('advances only exact active-subscription and address-relevant waiting rows after commit', async () => { + const harness = createCoordinatorHarness(); + const before = cloneStore(harness.store); + + await harness.coordinator.notifyDdnsSynced({ + appliedAddress: '8.8.8.8', + ddnsRecordId: '9007199254740993', + }); + + expect(harness.store.deliveries).toEqual( + before.deliveries.map((delivery) => + delivery.id === 'delivery-match' + ? { ...delivery, nextAttemptAt: new KtDateTime(NOW) } + : delivery, + ), + ); + expect(harness.operations).toEqual([ + 'transaction:start', + 'delivery:update', + 'transaction:commit', + 'drain', + ]); + expect(harness.requestDrain).toHaveBeenCalledTimes(1); + }); + + it('keeps the write status-fenced and does not drain when the update advances zero rows', async () => { + const harness = createCoordinatorHarness(); + harness.store.deliveries.find( + (delivery) => delivery.id === 'delivery-match', + )!.status = 'processing'; + const before = cloneStore(harness.store); + + await harness.coordinator.notifyDdnsSynced({ + appliedAddress: '8.8.8.8', + ddnsRecordId: '9007199254740993', + }); + + expect(harness.store).toEqual(before); + expect(harness.requestDrain).not.toHaveBeenCalled(); + }); + + it('discards transaction drafts and never drains after an update failure', async () => { + const harness = createCoordinatorHarness(); + const before = cloneStore(harness.store); + harness.failUpdate(new Error('delivery update failed')); + + await expect( + harness.coordinator.notifyDdnsSynced({ + appliedAddress: '8.8.8.8', + ddnsRecordId: '9007199254740993', + }), + ).rejects.toThrow('delivery update failed'); + + expect(harness.store).toEqual(before); + expect(harness.requestDrain).not.toHaveBeenCalled(); + expect(harness.operations).not.toContain('transaction:commit'); + }); + + it('does not query or drain after coordinator destruction', async () => { + const harness = createCoordinatorHarness(); + await harness.coordinator.onModuleDestroy(); + + await harness.coordinator.notifyDdnsSynced({ + appliedAddress: '8.8.8.8', + ddnsRecordId: '9007199254740993', + }); + + expect(harness.transaction).not.toHaveBeenCalled(); + expect(harness.requestDrain).not.toHaveBeenCalled(); + }); +}); diff --git a/test/qqbot/account/qqbot-account.service.spec.ts b/test/qqbot/account/qqbot-account.service.spec.ts index a53af26..df60352 100644 --- a/test/qqbot/account/qqbot-account.service.spec.ts +++ b/test/qqbot/account/qqbot-account.service.spec.ts @@ -1,6 +1,10 @@ import { ToolsService } from '@/common'; import type { QqbotAccountNapcatRuntimePort } from '@/modules/qqbot/core/application/account/qqbot-account-napcat-runtime.port'; import { QqbotAccountService } from '@/modules/qqbot/core/application/account/qqbot-account.service'; +import { QqbotAccountAbility } from '@/modules/qqbot/core/infrastructure/persistence/account/qqbot-account-ability.entity'; +import { QqbotAccount } from '@/modules/qqbot/core/infrastructure/persistence/account/qqbot-account.entity'; +import { QqbotMessageDelivery } from '@/modules/qqbot/core/infrastructure/persistence/message-push/qqbot-message-delivery.entity'; +import { QqbotMessagePublishBinding } from '@/modules/qqbot/core/infrastructure/persistence/message-push/qqbot-message-publish-binding.entity'; import { QqbotNapcatAccountRuntimeService } from '@/modules/qqbot/napcat/application/account-runtime/qqbot-napcat-account-runtime.service'; /** @@ -74,6 +78,369 @@ const createAccountServiceWithNapcatRuntime = (input: { }; describe('QqbotAccountService', () => { + it('fences administrative account cancellation to unfinished delivery statuses', async () => { + const deliveryUpdate = jest.fn().mockResolvedValue({ affected: 2 }); + const manager = { + getRepository: (entity: unknown) => { + if (entity === QqbotMessagePublishBinding) { + return { find: jest.fn().mockResolvedValue([{ id: 'binding-1' }]) }; + } + if (entity === QqbotMessageDelivery) return { update: deliveryUpdate }; + throw new Error('unexpected repository'); + }, + }; + const service = createAccountService({ accountRepository: {} }); + + await (service as any).cancelAccountDeliveries(manager, 'account-1'); + + expect(deliveryUpdate).toHaveBeenCalledWith( + expect.objectContaining({ + bindingId: expect.anything(), + status: expect.anything(), + }), + expect.objectContaining({ status: 'cancelled' }), + ); + const [where] = deliveryUpdate.mock.calls[0]; + expect(where.status.value).toEqual(['waiting_ddns', 'pending', 'retry']); + }); + + describe('administrative delivery cancellation transaction', () => { + /** Builds account/ability/delivery authority with isolated commit-on-resolve drafts. */ + function cancellationHarness() { + const accounts = [ + { + accessToken: null, + connectionMode: 'reverse-ws', + enabled: true, + id: 'account-1', + isDeleted: false, + name: 'Primary', + remark: '', + selfId: '1914728559', + }, + ]; + const abilities = [ + { + accountId: 'account-1', + id: 'ability-1', + isDeleted: false, + selfId: '1914728559', + }, + ]; + const bindings = [ + { accountId: 'account-1', id: 'binding-1' }, + { accountId: 'account-1', id: 'binding-2' }, + { accountId: 'account-2', id: 'binding-other' }, + ]; + const deliveries = [ + ...['waiting_ddns', 'pending', 'retry', 'processing', 'success'].map( + (status) => ({ + bindingId: 'binding-1', + frozen: `binding-1-${status}`, + id: `binding-1-${status}`, + nextAttemptAt: `schedule-${status}`, + processingLeaseUntil: `lease-${status}`, + status, + }), + ), + { + bindingId: 'binding-2', + frozen: 'binding-2', + id: 'binding-2-pending', + nextAttemptAt: 'schedule-binding-2', + processingLeaseUntil: 'lease-binding-2', + status: 'pending', + }, + { + bindingId: 'binding-other', + frozen: 'other', + id: 'binding-other-pending', + nextAttemptAt: 'schedule-other', + processingLeaseUntil: 'lease-other', + status: 'pending', + }, + ]; + const operations: string[] = []; + let failCancellation = false; + const transaction = jest.fn( + async (work: (manager: any) => Promise) => { + operations.push('transaction:start'); + const draftAccounts = structuredClone(accounts); + const draftAbilities = structuredClone(abilities); + const draftDeliveries = structuredClone(deliveries); + const accountStore = { + findOne: jest.fn( + async ({ + lock, + where, + }: { + lock?: { mode: string }; + where: Record; + }) => { + if (lock) operations.push('account:lock'); + const row = draftAccounts.find((candidate) => + Object.entries(where).every( + ([key, value]) => candidate[key] === value, + ), + ); + return row ? structuredClone(row) : null; + }, + ), + update: jest.fn( + async ( + where: Record, + patch: Record, + ) => { + operations.push('account:update'); + const row = draftAccounts.find((candidate) => + Object.entries(where).every( + ([key, value]) => candidate[key] === value, + ), + ); + if (!row) return { affected: 0 }; + Object.assign(row, patch); + return { affected: 1 }; + }, + ), + }; + const manager = { + getRepository: (entity: unknown) => { + if (entity === QqbotAccount) return accountStore; + if (entity === QqbotAccountAbility) { + return { + update: async ( + where: Record, + patch: Record, + ) => { + operations.push('ability:update'); + draftAbilities + .filter((row) => + Object.entries(where).every( + ([key, value]) => row[key] === value, + ), + ) + .forEach((row) => Object.assign(row, patch)); + return { affected: 1 }; + }, + }; + } + if (entity === QqbotMessagePublishBinding) { + return { + find: async ({ + where, + }: { + where: Record; + }) => { + operations.push('bindings:read'); + return bindings.filter((row) => + Object.entries(where).every( + ([key, value]) => row[key] === value, + ), + ); + }, + }; + } + if (entity === QqbotMessageDelivery) { + return { + update: async ( + where: Record, + patch: Record, + ) => { + operations.push('delivery:update'); + const bindingIds = where.bindingId._value as string[]; + const statuses = where.status._value as string[]; + draftDeliveries.forEach((row) => { + if ( + bindingIds.includes(row.bindingId) && + statuses.includes(row.status) + ) { + Object.assign(row, patch); + } + }); + if (failCancellation) { + throw new Error('account cancellation failed'); + } + return { affected: 4 }; + }, + }; + } + throw new Error('unexpected account repository'); + }, + }; + const result = await work(manager); + accounts.splice(0, accounts.length, ...draftAccounts); + abilities.splice(0, abilities.length, ...draftAbilities); + deliveries.splice(0, deliveries.length, ...draftDeliveries); + operations.push('transaction:commit'); + return result; + }, + ); + const accountRepository = { + findOne: jest.fn( + async ({ where }: { where: Record }) => { + const row = accounts.find((candidate) => + Object.entries(where).every( + ([key, value]) => candidate[key] === value, + ), + ); + return row ? structuredClone(row) : null; + }, + ), + manager: { transaction }, + update: jest.fn(), + }; + const removeAccountContainers = jest.fn(async () => { + operations.push('external:remove'); + return { deletedContainers: 1 }; + }); + const service = createAccountService({ + accountAbilityRepository: { update: jest.fn() }, + accountRepository, + napcatRuntime: { removeAccountContainers } as never, + }); + return { + abilities, + accounts, + deliveries, + failCancellation: () => { + failCancellation = true; + }, + operations, + removeAccountContainers, + service, + transaction, + }; + } + + /** Asserts only account-owned claimable delivery rows entered cancellation state. */ + function expectAccountDeliveriesCancelled( + before: Array>, + after: Array>, + ): void { + expect(after).toEqual( + before.map((delivery) => + ['binding-1', 'binding-2'].includes(String(delivery.bindingId)) && + ['waiting_ddns', 'pending', 'retry'].includes(String(delivery.status)) + ? { + ...delivery, + nextAttemptAt: null, + processingLeaseUntil: null, + status: 'cancelled', + } + : delivery, + ), + ); + } + + it.each([ + [ + 'administrative disable', + async (service: QqbotAccountService) => + service.update({ enabled: false, id: 'account-1' }), + ], + [ + 'actual selfId replacement', + async (service: QqbotAccountService) => + service.update({ + id: 'account-1', + selfId: '1914728560', + }), + ], + [ + 'administrative delete', + async (service: QqbotAccountService) => service.remove('account-1'), + ], + ])( + '%s commits account state and exact delivery cancellations', + async (_name, mutate) => { + const harness = cancellationHarness(); + const before = structuredClone(harness.deliveries); + + await mutate(harness.service); + + expectAccountDeliveriesCancelled(before, harness.deliveries); + expect(harness.operations.indexOf('account:lock')).toBeLessThan( + harness.operations.indexOf('bindings:read'), + ); + expect(harness.operations.indexOf('bindings:read')).toBeLessThan( + harness.operations.indexOf('delivery:update'), + ); + expect(harness.operations.at(-1)).toBe('transaction:commit'); + }, + ); + + it('same selfId and metadata-only updates do not cancel frozen deliveries', async () => { + const sameSelfId = cancellationHarness(); + const sameBefore = structuredClone(sameSelfId.deliveries); + await sameSelfId.service.update({ + id: 'account-1', + name: 'Renamed', + selfId: '1914728559', + }); + expect(sameSelfId.deliveries).toEqual(sameBefore); + expect(sameSelfId.operations).not.toContain('delivery:update'); + + const metadata = cancellationHarness(); + const metadataBefore = structuredClone(metadata.deliveries); + await metadata.service.update({ id: 'account-1', remark: 'metadata' }); + expect(metadata.deliveries).toEqual(metadataBefore); + expect(metadata.operations).not.toContain('delivery:update'); + }); + + it('runtime online/offline changes stay outside administrative cancellation', async () => { + const harness = cancellationHarness(); + const before = structuredClone(harness.deliveries); + + await harness.service.markOnline('1914728559', 'Universal'); + await harness.service.markOffline('1914728559'); + await harness.service.markHeartbeat('1914728559'); + + expect(harness.transaction).not.toHaveBeenCalled(); + expect(harness.deliveries).toEqual(before); + }); + + it('rolls back account, ability, and delivery drafts together when cancellation fails', async () => { + const harness = cancellationHarness(); + const before = { + abilities: structuredClone(harness.abilities), + accounts: structuredClone(harness.accounts), + deliveries: structuredClone(harness.deliveries), + }; + harness.failCancellation(); + + await expect( + harness.service.update({ + id: 'account-1', + selfId: '1914728560', + }), + ).rejects.toThrow('account cancellation failed'); + + expect(harness.accounts).toEqual(before.accounts); + expect(harness.abilities).toEqual(before.abilities); + expect(harness.deliveries).toEqual(before.deliveries); + expect(harness.operations).not.toContain('transaction:commit'); + }); + + it('keeps external removal outside the DB transaction while rolling back authoritative delete state', async () => { + const harness = cancellationHarness(); + const before = { + abilities: structuredClone(harness.abilities), + accounts: structuredClone(harness.accounts), + deliveries: structuredClone(harness.deliveries), + }; + harness.failCancellation(); + + await expect(harness.service.remove('account-1')).rejects.toThrow( + 'account cancellation failed', + ); + + expect(harness.operations[0]).toBe('external:remove'); + expect(harness.accounts).toEqual(before.accounts); + expect(harness.abilities).toEqual(before.abilities); + expect(harness.deliveries).toEqual(before.deliveries); + expect(harness.removeAccountContainers).toHaveBeenCalledTimes(1); + }); + }); it('stores NapCat login password as encrypted secret and never persists the transport field', async () => { const toolsService = new ToolsService(); const accountRepository = { @@ -208,8 +575,29 @@ describe('QqbotAccountService', () => { }); it('does not update NapCat login password when edit leaves the password blank', async () => { + const accountUpdate = jest.fn().mockResolvedValue({ affected: 1 }); + const account = { + enabled: true, + id: 'account-1', + isDeleted: false, + selfId: '1914728559', + }; const accountRepository = { - findOne: jest.fn().mockResolvedValue({ id: 'account-1' }), + findOne: jest.fn().mockResolvedValue(account), + manager: { + transaction: async (work: (manager: any) => Promise) => + work({ + getRepository: (entity: unknown) => { + if (entity === QqbotAccount) { + return { + findOne: jest.fn().mockResolvedValue({ ...account }), + update: accountUpdate, + }; + } + throw new Error('unexpected repository'); + }, + }), + }, update: jest.fn(), }; const service = createAccountService({ @@ -223,7 +611,7 @@ describe('QqbotAccountService', () => { selfId: '1914728559', }); - expect(accountRepository.update).toHaveBeenCalledWith( + expect(accountUpdate).toHaveBeenCalledWith( { id: 'account-1' }, expect.not.objectContaining({ encryptedLoginPassword: expect.anything(),