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 new file mode 100644 index 0000000..17e2cdf --- /dev/null +++ b/src/modules/qqbot/core/application/message-push/system-message-delivery-coordinator.service.ts @@ -0,0 +1,169 @@ +import { + Injectable, + Logger, + type OnModuleDestroy, + type OnModuleInit, +} from '@nestjs/common'; +import { DataSource, In } from 'typeorm'; +import { isIP } from 'node:net'; +import { KtDateTime } from '@/common'; +import { QqbotMessageDelivery } from '../../infrastructure/persistence/message-push/qqbot-message-delivery.entity'; +import { QqbotMessageEvent } from '../../infrastructure/persistence/message-push/qqbot-message-event.entity'; +import { QqbotMessageSubscription } from '../../infrastructure/persistence/message-push/qqbot-message-subscription.entity'; +import { + SYSTEM_MESSAGE_BATCH_SIZE, + SYSTEM_MESSAGE_SCAN_INTERVAL_MS, +} from './system-message-runner.constants'; +import { SystemMessageDeliveryRunnerService } from './system-message-delivery-runner.service'; +import { SystemMessageFanoutService } from './system-message-fanout.service'; + +const NETWORK_STUN_SOURCE = 'network.stun.mapping-port-changed'; + +/** Coordinates durable fan-out and delivery recovery without coupling Core to Admin providers. */ +@Injectable() +export class SystemMessageDeliveryCoordinatorService + implements OnModuleInit, OnModuleDestroy +{ + private readonly logger = new Logger( + SystemMessageDeliveryCoordinatorService.name, + ); + private destroyed = false; + private drainRequested = false; + private drainPromise: null | Promise = null; + private scanInterval?: NodeJS.Timeout; + private startupTimer?: NodeJS.Timeout; + + /** Creates the Core-owned coordinator and both bounded persistence runners. */ + constructor( + private readonly dataSource: DataSource, + private readonly fanoutRunner: SystemMessageFanoutService, + private readonly deliveryRunner: SystemMessageDeliveryRunnerService, + ) {} + + /** Starts one post-init wake and an unref'ed durable five-second recovery scan. */ + onModuleInit(): void { + if (this.destroyed || this.scanInterval) return; + this.startupTimer = setTimeout(() => { + this.startupTimer = undefined; + this.requestDrain(); + }, 0); + this.startupTimer.unref?.(); + this.scanInterval = setInterval( + () => this.requestDrain(), + SYSTEM_MESSAGE_SCAN_INTERVAL_MS, + ); + this.scanInterval.unref?.(); + } + + /** Stops timers and waits for the active bounded pass without allowing a replacement pass. */ + async onModuleDestroy(): Promise { + this.destroyed = true; + this.drainRequested = false; + if (this.startupTimer) clearTimeout(this.startupTimer); + if (this.scanInterval) clearInterval(this.scanInterval); + this.startupTimer = undefined; + this.scanInterval = undefined; + await this.drainPromise; + } + + /** Coalesces one or many post-commit wakeups into at most one active drain loop. */ + requestDrain(): void { + if (this.destroyed) return; + this.drainRequested = true; + if (this.drainPromise) return; + this.drainPromise = this.drainLoop() + .catch((error: unknown) => + this.logger.warn( + 'System message drain failed', + error instanceof Error ? error.message : undefined, + ), + ) + .finally(() => { + this.drainPromise = null; + if (!this.destroyed && this.drainRequested) this.requestDrain(); + }); + } + + /** Wakes only address-relevant DDNS-blocked rows after the DDNS transition has committed. */ + async notifyDdnsSynced(input: { + appliedAddress: string; + ddnsRecordId: string; + }): Promise { + if ( + this.destroyed || + isIP(input.appliedAddress) !== 4 || + !input.ddnsRecordId + ) + return; + const advanced = await this.dataSource.transaction(async (manager) => { + const subscriptions = await manager + .getRepository(QqbotMessageSubscription) + .find({ + where: { + enabled: true, + isDeleted: false, + sourceKey: NETWORK_STUN_SOURCE, + }, + }); + const subscriptionIds = subscriptions + .filter( + (subscription) => + typeof subscription.sourceConfig?.ddnsRecordId === 'string' && + subscription.sourceConfig.ddnsRecordId === input.ddnsRecordId, + ) + .map((subscription) => subscription.id); + if (!subscriptionIds.length) return 0; + const events = await manager + .getRepository(QqbotMessageEvent) + .find({ where: { sourceKey: NETWORK_STUN_SOURCE } }); + const eventIds = events + .filter((event) => event.payload.publicIpv4 === input.appliedAddress) + .map((event) => event.id); + if (!eventIds.length) return 0; + const result = await manager.getRepository(QqbotMessageDelivery).update( + { + messageEventId: In(eventIds), + status: 'waiting_ddns', + subscriptionId: In(subscriptionIds), + }, + { nextAttemptAt: new KtDateTime() }, + ); + return result.affected || 0; + }); + if (advanced > 0) this.requestDrain(); + } + + /** Runs ordered fan-out then delivery passes, isolating each runner's failure and backlog signal. */ + private async drainLoop(): Promise { + while (!this.destroyed && this.drainRequested) { + this.drainRequested = false; + const fanout = await this.runBounded('fan-out', () => + this.fanoutRunner.runOnce(), + ); + const delivery = await this.runBounded('delivery', () => + this.deliveryRunner.runOnce(), + ); + if ( + fanout === SYSTEM_MESSAGE_BATCH_SIZE || + delivery === SYSTEM_MESSAGE_BATCH_SIZE + ) + this.drainRequested = true; + } + } + + /** Executes one runner without allowing its rejection to starve the other durable queue. */ + private async runBounded( + name: string, + runner: () => Promise, + ): Promise { + try { + return await runner(); + } catch (error) { + this.logger.warn( + `System message ${name} scan failed`, + error instanceof Error ? error.message : undefined, + ); + return 0; + } + } +} diff --git a/src/modules/qqbot/core/application/message-push/system-message-delivery-runner.service.ts b/src/modules/qqbot/core/application/message-push/system-message-delivery-runner.service.ts new file mode 100644 index 0000000..5ea1d01 --- /dev/null +++ b/src/modules/qqbot/core/application/message-push/system-message-delivery-runner.service.ts @@ -0,0 +1,438 @@ +import { Injectable } from '@nestjs/common'; +import { Brackets, DataSource } from 'typeorm'; +import { KtDateTime } from '@/common'; +import { + SystemMessageContractError, + type SystemMessageDeliveryReadiness, +} from '../../contract/message-push/qqbot-message-push.types'; +import { QqbotAccount } from '../../infrastructure/persistence/account/qqbot-account.entity'; +import { QqbotMessageDelivery } from '../../infrastructure/persistence/message-push/qqbot-message-delivery.entity'; +import { QqbotMessageEvent } from '../../infrastructure/persistence/message-push/qqbot-message-event.entity'; +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 { QqbotMessageSubscription } from '../../infrastructure/persistence/message-push/qqbot-message-subscription.entity'; +import { QqbotSendAttemptError } from '../send/qqbot-send.error'; +import { QqbotSendService } from '../send/qqbot-send.service'; +import { + SYSTEM_MESSAGE_BATCH_SIZE, + SYSTEM_MESSAGE_DDNS_RECHECK_MS, + SYSTEM_MESSAGE_LEASE_MS, + SYSTEM_MESSAGE_RETRY_BASE_MS, + SYSTEM_MESSAGE_RETRY_MAX_MS, +} from './system-message-runner.constants'; +import { SystemMessageSourceRegistry } from './system-message-source.registry'; +import { SystemMessageTemplateRendererService } from './system-message-template-renderer.service'; + +const DELIVERY_EXPIRED = 'delivery_expired'; +const TRANSIENT_ERROR = 'delivery_transient_error'; + +type ClaimToken = { + attempt: number; + delivery: QqbotMessageDelivery; + leaseUntil: KtDateTime; +}; + +type PreparedDelivery = + | { kind: 'send'; delivery: QqbotMessageDelivery } + | { kind: 'stale' } + | { + code: string; + kind: 'finish'; + status: 'cancelled' | 'failed' | 'superseded' | 'waiting_ddns'; + }; + +/** Calculates the bounded delay for a one-based durable delivery attempt. */ +export function deliveryRetryDelayMs(attemptCount: number): number { + return Math.min( + SYSTEM_MESSAGE_RETRY_BASE_MS * 2 ** Math.max(0, attemptCount - 1), + SYSTEM_MESSAGE_RETRY_MAX_MS, + ); +} + +/** Claims, rechecks, and sends a bounded set of durable system-message deliveries. */ +@Injectable() +export class SystemMessageDeliveryRunnerService { + /** Creates the runner from Core persistence, source, renderer, and strict-send boundaries. */ + constructor( + private readonly dataSource: DataSource, + private readonly sourceRegistry: SystemMessageSourceRegistry, + private readonly templateRenderer: SystemMessageTemplateRendererService, + private readonly sendService: QqbotSendService, + ) {} + + /** Processes up to one bounded batch of due deliveries and returns its claim count. */ + async runOnce(now: Date = new Date()): Promise { + let claimed = 0; + for (let index = 0; index < SYSTEM_MESSAGE_BATCH_SIZE; index += 1) { + const token = await this.claimOne(now); + if (!token) break; + claimed += 1; + try { + await this.processClaim(token, now); + } catch { + await this.handleUnexpectedClaimFailure(token, now); + } + } + return claimed; + } + + /** Claims one oldest due row in a short transaction and returns its exact owner token. */ + private async claimOne(now: Date): Promise { + return this.dataSource.transaction(async (manager) => { + const deliveries = manager.getRepository(QqbotMessageDelivery); + const delivery = await deliveries + .createQueryBuilder('delivery') + .setLock('pessimistic_write') + .setOnLocked('skip_locked') + .where( + new Brackets((where) => { + where + .where( + 'delivery.status IN (:...due) AND delivery.nextAttemptAt <= :now', + { due: ['pending', 'retry', 'waiting_ddns'], now }, + ) + .orWhere( + 'delivery.status = :processing AND delivery.processingLeaseUntil <= :now', + { processing: 'processing', now }, + ); + }), + ) + .orderBy('delivery.nextAttemptAt', 'ASC') + .addOrderBy('delivery.id', 'ASC') + .take(1) + .getOne(); + if (!delivery) return null; + const leaseUntil = new KtDateTime( + now.getTime() + SYSTEM_MESSAGE_LEASE_MS, + ); + delivery.attemptCount += 1; + delivery.nextAttemptAt = null; + delivery.processingLeaseUntil = leaseUntil; + delivery.status = 'processing'; + await deliveries.save(delivery); + return { attempt: delivery.attemptCount, delivery, leaseUntil }; + }); + } + + /** Rechecks one claimed delivery before doing external I/O and then performs its final CAS. */ + private async processClaim(token: ClaimToken, now: Date): Promise { + if (now.getTime() >= token.delivery.expiresAt.getTime()) { + await this.finish( + token, + 'failed', + DELIVERY_EXPIRED, + 'delivery deadline reached', + null, + ); + return; + } + const prepared = await this.prepare(token, now); + if (prepared.kind === 'stale') return; + if (prepared.kind === 'finish') { + if (prepared.status === 'waiting_ddns') { + await this.finish( + token, + 'waiting_ddns', + prepared.code, + 'DDNS is not synchronized', + null, + new KtDateTime(now.getTime() + SYSTEM_MESSAGE_DDNS_RECHECK_MS), + ); + } else { + await this.finish( + token, + prepared.status, + prepared.code, + prepared.code, + null, + ); + } + return; + } + try { + const result = await this.sendService.sendStrictPlainText({ + attemptNumber: token.attempt, + deliveryId: token.delivery.id, + message: prepared.delivery.renderedMessage, + selfId: prepared.delivery.selfId, + targetId: prepared.delivery.targetId, + targetType: prepared.delivery.targetType, + }); + await this.finish(token, 'success', null, null, String(result.logId)); + } catch (error) { + if (error instanceof QqbotSendAttemptError) { + if (!error.retryable) { + await this.finish( + token, + 'failed', + error.code, + this.safeMessage(error), + error.sendLogId, + ); + return; + } + await this.retryOrFail( + token, + now, + error.code, + this.safeMessage(error), + error.sendLogId, + ); + return; + } + await this.retryOrFail( + token, + now, + TRANSIENT_ERROR, + 'delivery transport unavailable', + null, + ); + } + } + + /** Locks current configuration before locking the delivery and returns a safe next action. */ + private async prepare( + token: ClaimToken, + now: Date, + ): Promise { + return this.dataSource.transaction(async (manager) => { + const event = await manager.getRepository(QqbotMessageEvent).findOne({ + where: { id: token.delivery.messageEventId }, + lock: { mode: 'pessimistic_read' }, + }); + const subscription = await manager + .getRepository(QqbotMessageSubscription) + .findOne({ + where: { id: token.delivery.subscriptionId }, + lock: { mode: 'pessimistic_read' }, + }); + const binding = await manager + .getRepository(QqbotMessagePublishBinding) + .findOne({ + where: { id: token.delivery.bindingId }, + lock: { mode: 'pessimistic_read' }, + }); + const target = await manager + .getRepository(QqbotMessagePublishTarget) + .findOne({ + where: { id: token.delivery.publishTargetId }, + lock: { mode: 'pessimistic_read' }, + }); + const account = binding + ? await manager.getRepository(QqbotAccount).findOne({ + where: { id: binding.accountId }, + lock: { mode: 'pessimistic_read' }, + }) + : null; + const delivery = await manager + .getRepository(QqbotMessageDelivery) + .findOne({ + where: { id: token.delivery.id }, + lock: { mode: 'pessimistic_write' }, + }); + if (!delivery || !this.owns(delivery, token)) return { kind: 'stale' }; + if (!event) + return { code: 'event_invalid', kind: 'finish', status: 'failed' }; + if (delivery.targetType !== 'group' && delivery.targetType !== 'private') + return { + code: 'invalid_target_type', + kind: 'finish', + status: 'failed', + }; + if ( + !subscription || + !binding || + !target || + !account || + subscription.isDeleted || + binding.isDeleted || + target.isDeleted || + account.isDeleted || + !subscription.enabled || + !binding.enabled || + !target.enabled || + !account.enabled || + subscription.sourceKey !== event.sourceKey || + binding.subscriptionId !== delivery.subscriptionId || + target.bindingId !== delivery.bindingId || + target.targetId !== delivery.targetId || + target.targetType !== delivery.targetType || + binding.selfId !== delivery.selfId || + account.selfId !== delivery.selfId + ) { + return { + code: 'delivery_configuration_cancelled', + kind: 'finish', + status: 'cancelled', + }; + } + const adapter = this.sourceRegistry.get(event.sourceKey); + let eventPayload: Record; + try { + eventPayload = adapter.validateEventPayload(event.payload); + this.validateFrozen(delivery, event.sourceKey); + } catch (error) { + if (!(error instanceof SystemMessageContractError)) throw error; + return { + code: error.code, + kind: 'finish', + status: 'failed', + }; + } + const readiness: SystemMessageDeliveryReadiness = + await adapter.resolveDelivery({ + eventPayload, + subscriptionConfig: subscription.sourceConfig, + }); + if (readiness.status === 'waiting_ddns') { + if ( + now.getTime() + SYSTEM_MESSAGE_DDNS_RECHECK_MS >= + delivery.expiresAt.getTime() + ) + return { code: DELIVERY_EXPIRED, kind: 'finish', status: 'failed' }; + return { + code: readiness.reasonCode, + kind: 'finish', + status: 'waiting_ddns', + }; + } + if (readiness.status === 'cancelled' || readiness.status === 'superseded') + return { + code: readiness.reasonCode, + kind: 'finish', + status: readiness.status, + }; + return { delivery, kind: 'send' }; + }); + } + + /** Validates only immutable frozen content; current templates never replace historical work. */ + private validateFrozen( + delivery: QqbotMessageDelivery, + sourceKey: string, + ): void { + const definitions = this.sourceRegistry.get(sourceKey).definition.variables; + const allowed = definitions.map((item) => item.key); + const snapshot = delivery.variableSnapshot; + if ( + !snapshot || + typeof snapshot !== 'object' || + Array.isArray(snapshot) || + Object.keys(snapshot).length !== definitions.length || + definitions.some(({ key, type }) => { + if (!Object.prototype.hasOwnProperty.call(snapshot, key)) return true; + const value = snapshot[key]; + return ( + value === null || + typeof value !== type || + (type === 'number' && + (typeof value !== 'number' || !Number.isFinite(value))) + ); + }) || + Object.keys(snapshot).some((key) => !allowed.includes(key)) + ) + throw new SystemMessageContractError('template_invalid'); + this.templateRenderer.validate(delivery.templateContent, allowed); + if ( + this.templateRenderer.render( + delivery.templateContent, + snapshot as Record, + ) !== delivery.renderedMessage + ) + throw new SystemMessageContractError('rendered_message_mismatch'); + } + + /** Best-effort fences an unexpected dependency failure without starving later claims. */ + private async handleUnexpectedClaimFailure( + token: ClaimToken, + now: Date, + ): Promise { + try { + await this.retryOrFail( + token, + now, + TRANSIENT_ERROR, + 'delivery dependency unavailable', + null, + ); + } catch { + // A later scan recovers the still-processing row after its lease expires. + } + } + + /** Persists a terminal, wait, or successful state only while this worker owns its lease. */ + private async finish( + token: ClaimToken, + status: 'cancelled' | 'failed' | 'success' | 'superseded' | 'waiting_ddns', + code: null | string, + message: null | string, + sendLogId: null | string, + nextAttemptAt: KtDateTime | null = null, + ): Promise { + const result = await this.dataSource + .getRepository(QqbotMessageDelivery) + .update(this.ownerWhere(token), { + lastErrorCode: code, + lastErrorMessage: message, + nextAttemptAt, + processingLeaseUntil: null, + sendLogId: sendLogId ?? token.delivery.sendLogId, + status, + }); + if (result.affected !== 1) return; + } + + /** Retries until the event-derived expiration boundary and otherwise writes a final failure. */ + private async retryOrFail( + token: ClaimToken, + now: Date, + code: string, + message: string, + sendLogId: null | string, + ): Promise { + const next = new KtDateTime( + now.getTime() + deliveryRetryDelayMs(token.attempt), + ); + if (next.getTime() >= token.delivery.expiresAt.getTime()) { + await this.finish(token, 'failed', code, message, sendLogId); + return; + } + const result = await this.dataSource + .getRepository(QqbotMessageDelivery) + .update(this.ownerWhere(token), { + lastErrorCode: code, + lastErrorMessage: message, + nextAttemptAt: next, + processingLeaseUntil: null, + sendLogId: sendLogId ?? token.delivery.sendLogId, + status: 'retry', + }); + if (result.affected !== 1) return; + } + + /** Builds the exact attempt-and-lease owner fence used by every post-claim mutation. */ + private ownerWhere(token: ClaimToken) { + return { + attemptCount: token.attempt, + id: token.delivery.id, + processingLeaseUntil: token.leaseUntil, + status: 'processing' as const, + }; + } + + /** Tests whether a locked delivery still belongs to the detached claim owner. */ + private owns(delivery: QqbotMessageDelivery, token: ClaimToken): boolean { + return ( + delivery.status === 'processing' && + delivery.attemptCount === token.attempt && + delivery.processingLeaseUntil?.getTime() === token.leaseUntil.getTime() + ); + } + + /** Converts unexpected error text into a bounded persistence-safe summary. */ + private safeMessage(error: Error): string { + return String(error.message || 'delivery failed') + .replace(/[\r\n]/g, ' ') + .slice(0, 500); + } +} diff --git a/src/modules/qqbot/core/qqbot-core.module.ts b/src/modules/qqbot/core/qqbot-core.module.ts index 2cb03d4..abfc727 100644 --- a/src/modules/qqbot/core/qqbot-core.module.ts +++ b/src/modules/qqbot/core/qqbot-core.module.ts @@ -50,12 +50,17 @@ import { QqbotMessageTemplate } from '@/modules/qqbot/core/infrastructure/persis import { SystemMessageSourceRegistry } from './application/message-push/system-message-source.registry'; import { SystemMessageEventStagerService } from './application/message-push/system-message-event-stager.service'; import { SystemMessageFanoutService } from './application/message-push/system-message-fanout.service'; +import { SystemMessageDeliveryRunnerService } from './application/message-push/system-message-delivery-runner.service'; +import { SystemMessageDeliveryCoordinatorService } from './application/message-push/system-message-delivery-coordinator.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'; +import { + SYSTEM_MESSAGE_DELIVERY_COORDINATOR, + SYSTEM_MESSAGE_EVENT_STAGER, +} from './contract/message-push/qqbot-message-push.types'; export { QQBOT_CORE_DOMAIN_CONTRACT } from './contract/qqbot-core.contract'; @@ -94,10 +99,16 @@ export const QQBOT_CORE_PROVIDERS = [ SystemMessageSourceRegistry, SystemMessageEventStagerService, SystemMessageFanoutService, + SystemMessageDeliveryRunnerService, + SystemMessageDeliveryCoordinatorService, { provide: SYSTEM_MESSAGE_EVENT_STAGER, useExisting: SystemMessageEventStagerService, }, + { + provide: SYSTEM_MESSAGE_DELIVERY_COORDINATOR, + useExisting: SystemMessageDeliveryCoordinatorService, + }, SystemMessageTemplateRendererService, QqbotMessageSubscriptionService, QqbotMessageTemplateService, @@ -124,6 +135,7 @@ export const QQBOT_CORE_PROVIDERS = [ export const QQBOT_CORE_EXPORTS = [ SYSTEM_MESSAGE_EVENT_STAGER, + SYSTEM_MESSAGE_DELIVERY_COORDINATOR, SystemMessageSourceRegistry, SystemMessageTemplateRendererService, QqbotMessageSubscriptionService, diff --git a/test/modules/qqbot/core/qqbot-core-module-contract.spec.ts b/test/modules/qqbot/core/qqbot-core-module-contract.spec.ts index 73af88d..8f3a497 100644 --- a/test/modules/qqbot/core/qqbot-core-module-contract.spec.ts +++ b/test/modules/qqbot/core/qqbot-core-module-contract.spec.ts @@ -1,4 +1,4 @@ -import { existsSync } from 'fs'; +import { existsSync, readFileSync } from 'fs'; import { join } from 'path'; import { MODULE_METADATA } from '@nestjs/common/constants'; import { ConfigModule } from '@nestjs/config'; @@ -16,7 +16,12 @@ import { QqbotRuleController } from '../../../../src/modules/qqbot/core/contract 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 { SystemMessageFanoutService } from '../../../../src/modules/qqbot/core/application/message-push/system-message-fanout.service'; -import { SYSTEM_MESSAGE_EVENT_STAGER } from '../../../../src/modules/qqbot/core/contract/message-push/qqbot-message-push.types'; +import { SystemMessageDeliveryCoordinatorService } from '../../../../src/modules/qqbot/core/application/message-push/system-message-delivery-coordinator.service'; +import { SystemMessageDeliveryRunnerService } from '../../../../src/modules/qqbot/core/application/message-push/system-message-delivery-runner.service'; +import { + SYSTEM_MESSAGE_DELIVERY_COORDINATOR, + SYSTEM_MESSAGE_EVENT_STAGER, +} from '../../../../src/modules/qqbot/core/contract/message-push/qqbot-message-push.types'; import { QQBOT_CORE_CONTROLLERS, QQBOT_CORE_ENTITIES, @@ -173,10 +178,16 @@ describe('QQBot core module contract', () => { expect.arrayContaining([ SystemMessageEventStagerService, SystemMessageFanoutService, + SystemMessageDeliveryRunnerService, + SystemMessageDeliveryCoordinatorService, { provide: SYSTEM_MESSAGE_EVENT_STAGER, useExisting: SystemMessageEventStagerService, }, + { + provide: SYSTEM_MESSAGE_DELIVERY_COORDINATOR, + useExisting: SystemMessageDeliveryCoordinatorService, + }, ]), ); expect( @@ -184,8 +195,45 @@ describe('QQBot core module contract', () => { (provider) => provider === SystemMessageFanoutService, ), ).toHaveLength(1); + expect( + QQBOT_CORE_PROVIDERS.filter( + (provider) => provider === SystemMessageDeliveryRunnerService, + ), + ).toHaveLength(1); + expect( + QQBOT_CORE_PROVIDERS.filter( + (provider) => provider === SystemMessageDeliveryCoordinatorService, + ), + ).toHaveLength(1); expect(QQBOT_CORE_EXPORTS).toEqual( - expect.arrayContaining([SYSTEM_MESSAGE_EVENT_STAGER]), + expect.arrayContaining([ + SYSTEM_MESSAGE_EVENT_STAGER, + SYSTEM_MESSAGE_DELIVERY_COORDINATOR, + ]), + ); + expect(QQBOT_CORE_EXPORTS).not.toContain( + SystemMessageDeliveryCoordinatorService, + ); + expect( + QQBOT_CORE_PROVIDERS.filter( + (provider) => + typeof provider === 'object' && + provider !== null && + 'provide' in provider && + provider.provide === SYSTEM_MESSAGE_DELIVERY_COORDINATOR, + ), + ).toEqual([ + { + provide: SYSTEM_MESSAGE_DELIVERY_COORDINATOR, + useExisting: SystemMessageDeliveryCoordinatorService, + }, + ]); + const coreModuleSource = readFileSync( + join(process.cwd(), 'src/modules/qqbot/core/qqbot-core.module.ts'), + 'utf8', + ); + expect(coreModuleSource).not.toMatch( + /network-(?:agent-mqtt|ddns|stun-message-source)/, ); }); diff --git a/test/modules/qqbot/message-push/system-message-delivery-runner.spec.ts b/test/modules/qqbot/message-push/system-message-delivery-runner.spec.ts new file mode 100644 index 0000000..58ebce0 --- /dev/null +++ b/test/modules/qqbot/message-push/system-message-delivery-runner.spec.ts @@ -0,0 +1,1455 @@ +import { KtDateTime } from '../../../../src/common'; +import { + deliveryRetryDelayMs, + SystemMessageDeliveryRunnerService, +} from '../../../../src/modules/qqbot/core/application/message-push/system-message-delivery-runner.service'; +import { SystemMessageDeliveryCoordinatorService } from '../../../../src/modules/qqbot/core/application/message-push/system-message-delivery-coordinator.service'; +import { + SYSTEM_MESSAGE_BATCH_SIZE, + SYSTEM_MESSAGE_DDNS_RECHECK_MS, + SYSTEM_MESSAGE_LEASE_MS, + SYSTEM_MESSAGE_SCAN_INTERVAL_MS, +} from '../../../../src/modules/qqbot/core/application/message-push/system-message-runner.constants'; +import { SystemMessageSourceRegistry } from '../../../../src/modules/qqbot/core/application/message-push/system-message-source.registry'; +import { SystemMessageTemplateRendererService } from '../../../../src/modules/qqbot/core/application/message-push/system-message-template-renderer.service'; +import { QqbotSendAttemptError } from '../../../../src/modules/qqbot/core/application/send/qqbot-send.error'; +import { + SystemMessageContractError, + type StrictPlainTextSendInput, + type SystemMessageSourceAdapter, +} from '../../../../src/modules/qqbot/core/contract/message-push/qqbot-message-push.types'; +import { QqbotAccount } from '../../../../src/modules/qqbot/core/infrastructure/persistence/account/qqbot-account.entity'; +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 { 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 { QqbotMessageSubscription } from '../../../../src/modules/qqbot/core/infrastructure/persistence/message-push/qqbot-message-subscription.entity'; + +const NOW = new Date('2026-07-24T00:00:00.000Z'); +const SOURCE_KEY = 'network.stun.mapping-port-changed'; +const RESOURCE_KEY = '9007199254740993'; + +type Store = { + accounts: QqbotAccount[]; + bindings: QqbotMessagePublishBinding[]; + deliveries: QqbotMessageDelivery[]; + events: QqbotMessageEvent[]; + subscriptions: QqbotMessageSubscription[]; + targets: QqbotMessagePublishTarget[]; +}; + +type QueryRecord = { + lock: string; + onLocked: string; + orders: string[]; + predicates: Array<{ + expression: string; + parameters: Record; + }>; + take: number; + topLevelBrackets: boolean; +}; + +/** Creates a deferred promise used to force claim and lifecycle interleavings. */ +function deferred() { + let resolve!: (value: T | PromiseLike) => void; + let reject!: (reason?: unknown) => void; + const promise = new Promise((resolvePromise, rejectPromise) => { + resolve = resolvePromise; + reject = rejectPromise; + }); + return { promise, reject, resolve }; +} + +/** Builds one immutable source event with Snowflake identities retained as strings. */ +function event(overrides: Partial = {}) { + return Object.assign(new QqbotMessageEvent(), { + id: '201', + occurredAt: new KtDateTime(NOW), + payload: { + currentPort: 38213, + portForwardId: RESOURCE_KEY, + previousPort: 8213, + publicIpv4: '203.0.113.10', + }, + resourceKey: RESOURCE_KEY, + sourceKey: SOURCE_KEY, + ...overrides, + }); +} + +/** Builds one active source subscription. */ +function subscription(overrides: Partial = {}) { + return Object.assign(new QqbotMessageSubscription(), { + enabled: true, + id: '301', + isDeleted: false, + sourceConfig: { + ddnsRecordId: '9007199254740995', + portForwardId: RESOURCE_KEY, + }, + sourceKey: SOURCE_KEY, + ...overrides, + }); +} + +/** Builds one enabled administrative account; runtime online state is deliberately irrelevant. */ +function account(overrides: Partial = {}) { + return Object.assign(new QqbotAccount(), { + enabled: true, + id: '401', + isDeleted: false, + selfId: 'bot-a', + ...overrides, + }); +} + +/** Builds one enabled account-to-subscription publishing binding. */ +function binding(overrides: Partial = {}) { + return Object.assign(new QqbotMessagePublishBinding(), { + accountId: '401', + enabled: true, + id: '501', + isDeleted: false, + selfId: 'bot-a', + subscriptionId: '301', + templateId: '601', + ...overrides, + }); +} + +/** Builds one enabled frozen group target. */ +function target(overrides: Partial = {}) { + return Object.assign(new QqbotMessagePublishTarget(), { + bindingId: '501', + enabled: true, + id: '701', + isDeleted: false, + targetId: 'group-1', + targetType: 'group', + ...overrides, + }); +} + +/** Builds one due delivery with a complete immutable template and variable snapshot. */ +function delivery(overrides: Partial = {}) { + return Object.assign(new QqbotMessageDelivery(), { + attemptCount: 0, + bindingId: '501', + expiresAt: new KtDateTime(NOW.getTime() + 24 * 60 * 60_000), + id: '801', + lastErrorCode: 'old_error', + lastErrorMessage: 'old message', + messageEventId: '201', + nextAttemptAt: new KtDateTime(NOW), + processingLeaseUntil: null, + publishTargetId: '701', + renderedMessage: 'endpoint=pal.example.com:38213 port=38213', + selfId: 'bot-a', + sendLogId: 'old-log', + status: 'pending', + subscriptionId: '301', + targetId: 'group-1', + targetType: 'group', + templateContent: 'endpoint=${{endpoint}} port=${{port}}', + templateId: '601', + variableSnapshot: { + endpoint: 'pal.example.com:38213', + port: 38213, + }, + ...overrides, + }); +} + +/** Clones all rows for an isolated transaction draft while preserving entity prototypes. */ +function cloneStore(value: Store): Store { + const clone = (items: T[]) => + items.map((item) => + Object.assign( + Object.create(Object.getPrototypeOf(item)), + structuredClone(item), + ), + ); + return { + accounts: clone(value.accounts), + bindings: clone(value.bindings), + deliveries: clone(value.deliveries), + events: clone(value.events), + subscriptions: clone(value.subscriptions), + targets: clone(value.targets), + }; +} + +/** Compares scalar and date fields used by the repository condition harness. */ +function sameValue(actual: unknown, expected: unknown): boolean { + if (actual instanceof Date && expected instanceof Date) + return actual.getTime() === expected.getTime(); + return actual === expected; +} + +/** Expands TypeORM `In()` operators without interpreting production control flow. */ +function conditionValues(value: unknown): unknown[] | null { + if ( + value && + typeof value === 'object' && + '_type' in value && + (value as { _type?: unknown })._type === 'in' + ) + return (value as unknown as { _value: unknown[] })._value; + return null; +} + +/** Matches the small equality/IN repository condition subset exercised by the runner. */ +function matches( + row: Record, + where: Record = {}, +): boolean { + return Object.entries(where).every(([key, expected]) => { + const values = conditionValues(expected); + return values + ? values.some((value) => sameValue(row[key], value)) + : sameValue(row[key], expected); + }); +} + +/** Creates an authoritative-store runner harness with commit/rollback, locks, and exact CAS. */ +function setup(seed: Partial = {}) { + let state: Store = { + accounts: seed.accounts ?? [account()], + bindings: seed.bindings ?? [binding()], + deliveries: seed.deliveries ?? [delivery()], + events: seed.events ?? [event()], + subscriptions: seed.subscriptions ?? [subscription()], + targets: seed.targets ?? [target()], + }; + const registry = new SystemMessageSourceRegistry(); + const adapter: jest.Mocked = { + definition: { + description: 'test', + displayName: 'test', + sourceKey: SOURCE_KEY, + subscriptionFields: [], + variables: [ + { + description: 'endpoint', + example: 'pal.example.com:38213', + key: 'endpoint', + label: 'endpoint', + type: 'string', + }, + { + description: 'port', + example: '38213', + key: 'port', + label: 'port', + type: 'number', + }, + ], + version: 1, + }, + inspectSubscription: jest.fn(), + listSubscriptionOptions: jest.fn(), + normalizeSubscriptionConfig: jest.fn(), + resolveDelivery: jest.fn( + async ( + _input: Parameters[0], + ) => { + void _input; + return { + reasonCode: null, + status: 'ready' as const, + variables: { endpoint: 'current.example.com:39999', port: 39999 }, + }; + }, + ), + validateEventPayload: jest.fn( + (payload) => payload as Record, + ), + }; + registry.register(adapter); + const operations: string[] = []; + const lockOrder: string[] = []; + const queries: QueryRecord[] = []; + const senderDepths: number[] = []; + const preparationClaimLocks: boolean[] = []; + const activeLocks = new Set(); + let transactionDepth = 0; + let externalUpdateFailures = 0; + let beforeExternalUpdate: + | null + | ((authoritative: Store, where: Record) => void) = null; + let beforePreparationDeliveryRead: null | ((draft: Store) => void) = null; + let pauseNextClaim: null | { entered: () => void; resume: Promise } = + null; + + const sender = { + sendStrictPlainText: jest.fn( + async (_input: StrictPlainTextSendInput): Promise<{ logId: string }> => { + void _input; + senderDepths.push(transactionDepth); + operations.push('send'); + return { logId: 'send-log-1' }; + }, + ), + }; + + /** Merges only locally changed draft rows so overlapping transactions retain both commits. */ + const mergeRows = ( + authoritative: T[], + draft: T[], + baseline: T[], + ): T[] => { + const merged = new Map(authoritative.map((row) => [row.id, row])); + const baselineById = new Map(baseline.map((row) => [row.id, row])); + for (const row of draft) { + if ( + !baselineById.has(row.id) || + JSON.stringify(row) !== JSON.stringify(baselineById.get(row.id)) + ) + merged.set(row.id, row); + } + return [...merged.values()]; + }; + + /** Applies a committed draft without replacing unrelated concurrent authoritative rows. */ + const mergeCommitted = (draft: Store, baseline: Store): void => { + state = { + accounts: mergeRows(state.accounts, draft.accounts, baseline.accounts), + bindings: mergeRows(state.bindings, draft.bindings, baseline.bindings), + deliveries: mergeRows( + state.deliveries, + draft.deliveries, + baseline.deliveries, + ), + events: mergeRows(state.events, draft.events, baseline.events), + subscriptions: mergeRows( + state.subscriptions, + draft.subscriptions, + baseline.subscriptions, + ), + targets: mergeRows(state.targets, draft.targets, baseline.targets), + }; + }; + + /** Returns the entity array belonging to one authoritative or transaction-local store. */ + const rowsFor = (entity: unknown, store: Store) => { + if (entity === QqbotAccount) return store.accounts; + if (entity === QqbotMessageEvent) return store.events; + if (entity === QqbotMessageSubscription) return store.subscriptions; + if (entity === QqbotMessagePublishBinding) return store.bindings; + if (entity === QqbotMessagePublishTarget) return store.targets; + return store.deliveries; + }; + + /** Creates a repository facade over one store and transaction lock scope. */ + const repository = ( + entity: unknown, + store: Store, + context?: { claim: boolean; preparation: boolean }, + transactionLocks?: Set, + ) => { + const rows = rowsFor(entity, store) as unknown as Array< + Record + >; + return { + createQueryBuilder: (_alias: string) => { + void _alias; + const record: QueryRecord = { + lock: '', + onLocked: '', + orders: [], + predicates: [], + take: 0, + topLevelBrackets: false, + }; + queries.push(record); + let queryNow = NOW; + const builder = { + addOrderBy: (expression: string, order: string) => { + record.orders.push(`${expression} ${order}`); + return builder; + }, + getOne: async () => { + const due = (rows as unknown as QqbotMessageDelivery[]) + .filter((item) => { + const scheduled = + ['pending', 'retry', 'waiting_ddns'].includes(item.status) && + !!item.nextAttemptAt && + item.nextAttemptAt.getTime() <= queryNow.getTime(); + const expiredLease = + item.status === 'processing' && + !!item.processingLeaseUntil && + item.processingLeaseUntil.getTime() <= queryNow.getTime(); + return scheduled || expiredLease; + }) + .sort((left, right) => { + const leftTime = left.nextAttemptAt?.getTime() ?? -Infinity; + const rightTime = right.nextAttemptAt?.getTime() ?? -Infinity; + if (leftTime !== rightTime) return leftTime - rightTime; + return BigInt(left.id) < BigInt(right.id) ? -1 : 1; + }) + .find((item) => !activeLocks.has(`delivery:${item.id}`)); + if (!due) return null; + const lockKey = `delivery:${due.id}`; + activeLocks.add(lockKey); + transactionLocks?.add(lockKey); + if (context) context.claim = true; + const pause = pauseNextClaim; + pauseNextClaim = null; + if (pause) { + pause.entered(); + await pause.resume; + } + return due; + }, + orderBy: (expression: string, order: string) => { + record.orders.push(`${expression} ${order}`); + return builder; + }, + setLock: (mode: string) => { + record.lock = mode; + return builder; + }, + setOnLocked: (mode: string) => { + record.onLocked = mode; + return builder; + }, + take: (count: number) => { + record.take = count; + return builder; + }, + where: (condition: unknown) => { + record.topLevelBrackets = + !!condition && + typeof condition === 'object' && + 'whereFactory' in condition; + if (record.topLevelBrackets) { + const recorder = { + orWhere: ( + expression: string, + parameters: Record = {}, + ) => { + record.predicates.push({ expression, parameters }); + if (parameters.now instanceof Date) queryNow = parameters.now; + return recorder; + }, + where: ( + expression: string, + parameters: Record = {}, + ) => { + record.predicates.push({ expression, parameters }); + if (parameters.now instanceof Date) queryNow = parameters.now; + return recorder; + }, + }; + ( + condition as { + whereFactory: (builderValue: typeof recorder) => void; + } + ).whereFactory(recorder); + } + return builder; + }, + }; + return builder; + }, + find: async ( + options: { + where?: Record; + } = {}, + ) => rows.filter((row) => matches(row, options.where)), + findOne: async (options: { + lock?: { mode: string }; + where: Record; + }) => { + if (options.lock) { + lockOrder.push((entity as { name?: string }).name || 'unknown'); + if (entity === QqbotMessageDelivery) { + if (context) context.preparation = true; + beforePreparationDeliveryRead?.(store); + preparationClaimLocks.push( + activeLocks.has(`delivery:${String(options.where.id)}`), + ); + } + } + return rows.find((row) => matches(row, options.where)) ?? null; + }, + save: async (item: Record) => item, + update: async ( + where: Record, + values: Record, + ) => { + if (!context) { + beforeExternalUpdate?.(state, where); + if (externalUpdateFailures > 0) { + externalUpdateFailures -= 1; + throw new Error('simulated persistence failure'); + } + } + let affected = 0; + for (const row of rowsFor( + entity, + context ? store : state, + ) as unknown as Array>) { + if (!matches(row, where)) continue; + Object.assign(row, values); + affected += 1; + } + return { affected }; + }, + }; + }; + + const dataSource = { + getRepository: (entity: unknown) => repository(entity, state), + transaction: async ( + callback: (manager: { + getRepository: (entity: unknown) => ReturnType; + }) => Promise, + ) => { + const baseline = cloneStore(state); + const draft = cloneStore(state); + const transactionLocks = new Set(); + const context = { claim: false, preparation: false }; + transactionDepth += 1; + try { + const result = await callback({ + getRepository: (entity: unknown) => + repository(entity, draft, context, transactionLocks), + }); + mergeCommitted(draft, baseline); + operations.push( + context.claim + ? 'claim:commit' + : context.preparation + ? 'prepare:commit' + : 'transaction:commit', + ); + return result; + } catch (error) { + operations.push('transaction:rollback'); + throw error; + } finally { + for (const key of transactionLocks) activeLocks.delete(key); + transactionDepth -= 1; + } + }, + }; + + const runner = new SystemMessageDeliveryRunnerService( + dataSource as never, + registry, + new SystemMessageTemplateRendererService(), + sender as never, + ); + return { + adapter, + beforeExternalUpdate: ( + callback: + | null + | ((authoritative: Store, where: Record) => void), + ) => { + beforeExternalUpdate = callback; + }, + beforePreparationDeliveryRead: ( + callback: null | ((draft: Store) => void), + ) => { + beforePreparationDeliveryRead = callback; + }, + failExternalUpdates: (count: number) => { + externalUpdateFailures = count; + }, + getState: () => state, + lockOrder, + operations, + pauseClaim: (entered: () => void, resume: Promise) => { + pauseNextClaim = { entered, resume }; + }, + queries, + preparationClaimLocks, + registry, + runner, + sender, + senderDepths, + }; +} + +/** Flushes coordinator promise continuations without reading its private fields. */ +async function flushPromises(rounds = 8): Promise { + for (let index = 0; index < rounds; index += 1) await Promise.resolve(); +} + +describe('System message delivery runner direct preflight contracts', () => { + it('1 records the exact top-level due/expired predicate, order, limit, and skip-locked write lock', async () => { + const harness = setup(); + await harness.runner.runOnce(NOW); + expect(harness.queries[0]).toMatchObject({ + lock: 'pessimistic_write', + onLocked: 'skip_locked', + orders: ['delivery.nextAttemptAt ASC', 'delivery.id ASC'], + take: 1, + topLevelBrackets: true, + }); + expect(harness.queries[0].predicates).toEqual([ + { + expression: + 'delivery.status IN (:...due) AND delivery.nextAttemptAt <= :now', + parameters: { + due: ['pending', 'retry', 'waiting_ddns'], + now: NOW, + }, + }, + { + expression: + 'delivery.status = :processing AND delivery.processingLeaseUntil <= :now', + parameters: { now: NOW, processing: 'processing' }, + }, + ]); + }); + + it('2 bounds claims at 50 and excludes no-due, future, and live-lease rows', async () => { + const due = Array.from( + { length: SYSTEM_MESSAGE_BATCH_SIZE + 2 }, + (_, index) => + delivery({ + id: `${1000 + index}`, + publishTargetId: `${2000 + index}`, + }), + ); + const harness = setup({ + deliveries: [ + ...due, + delivery({ + id: '3001', + nextAttemptAt: new KtDateTime(NOW.getTime() + 1), + status: 'retry', + }), + delivery({ + id: '3002', + nextAttemptAt: null, + processingLeaseUntil: new KtDateTime(NOW.getTime() + 1), + status: 'processing', + }), + ], + targets: due.map((item) => + target({ + id: item.publishTargetId, + targetId: item.targetId, + }), + ), + }); + expect(await harness.runner.runOnce(NOW)).toBe(SYSTEM_MESSAGE_BATCH_SIZE); + expect( + harness.getState().deliveries.filter((item) => item.status === 'success'), + ).toHaveLength(SYSTEM_MESSAGE_BATCH_SIZE); + expect( + harness.getState().deliveries.find((item) => item.id === '3001')?.status, + ).toBe('retry'); + expect( + harness.getState().deliveries.find((item) => item.id === '3002')?.status, + ).toBe('processing'); + expect(await setup({ deliveries: [] }).runner.runOnce(NOW)).toBe(0); + }); + + it('3 skips an overlapping claim lock and never claims the same live lease twice', async () => { + const firstEntered = deferred(); + const releaseFirst = deferred(); + const harness = setup({ + deliveries: [ + delivery({ id: '801', publishTargetId: '701' }), + delivery({ id: '802', publishTargetId: '702' }), + ], + targets: [target(), target({ id: '702' })], + }); + harness.pauseClaim(firstEntered.resolve, releaseFirst.promise); + const firstRun = harness.runner.runOnce(NOW); + await firstEntered.promise; + const secondRun = harness.runner.runOnce(NOW); + await flushPromises(); + releaseFirst.resolve(); + expect(await Promise.all([firstRun, secondRun])).toEqual([1, 1]); + expect(harness.sender).toHaveProperty( + 'sendStrictPlainText', + expect.any(Function), + ); + expect(harness.sender.sendStrictPlainText).toHaveBeenCalledTimes(2); + expect( + new Set( + harness.sender.sendStrictPlainText.mock.calls.map( + ([input]) => input.deliveryId, + ), + ).size, + ).toBe(2); + }); + + it('4 recovers only an expired processing lease with a new attempt and lease', async () => { + const harness = setup({ + deliveries: [ + delivery({ + attemptCount: 4, + id: '801', + nextAttemptAt: null, + processingLeaseUntil: new KtDateTime(NOW.getTime() - 1), + status: 'processing', + }), + delivery({ + attemptCount: 8, + id: '802', + nextAttemptAt: null, + processingLeaseUntil: new KtDateTime(NOW.getTime() + 1), + status: 'processing', + }), + ], + }); + expect(await harness.runner.runOnce(NOW)).toBe(1); + expect(harness.sender.sendStrictPlainText).toHaveBeenCalledWith( + expect.objectContaining({ attemptNumber: 5, deliveryId: '801' }), + ); + expect( + harness.getState().deliveries.find((item) => item.id === '802'), + ).toMatchObject({ attemptCount: 8, status: 'processing' }); + expect(harness.operations.slice(0, 3)).toEqual([ + 'claim:commit', + 'prepare:commit', + 'send', + ]); + }); + + it.each([ + 'success', + 'retry', + 'cancelled', + 'failed', + 'superseded', + 'waiting_ddns', + ] as const)( + '5 fences a stale owner %s result with exact id/status/attempt/lease CAS', + async (resultKind) => { + const harness = setup(); + if (resultKind === 'retry') { + harness.sender.sendStrictPlainText.mockRejectedValueOnce( + new QqbotSendAttemptError({ + code: 'onebot_timeout', + message: 'timeout', + retryable: true, + sendLogId: 'new-log', + }), + ); + } + if (resultKind === 'cancelled') { + harness.getState().targets[0].enabled = false; + } + if (resultKind === 'failed') { + harness.getState().deliveries[0].templateContent = + 'endpoint=${{endpoint}'; + } + if (resultKind === 'superseded') { + harness.adapter.resolveDelivery.mockResolvedValueOnce({ + reasonCode: 'endpoint_superseded', + status: 'superseded', + }); + } + if (resultKind === 'waiting_ddns') { + harness.adapter.resolveDelivery.mockResolvedValueOnce({ + reasonCode: 'ddns_not_synced', + status: 'waiting_ddns', + variables: { endpoint: 'current', port: 38213 }, + }); + } + harness.beforeExternalUpdate((state, where) => { + expect(Object.keys(where).sort()).toEqual([ + 'attemptCount', + 'id', + 'processingLeaseUntil', + 'status', + ]); + expect(where).toEqual({ + attemptCount: 1, + id: '801', + processingLeaseUntil: new KtDateTime( + NOW.getTime() + SYSTEM_MESSAGE_LEASE_MS, + ), + status: 'processing', + }); + const row = state.deliveries[0]; + row.attemptCount += 1; + row.processingLeaseUntil = new KtDateTime(NOW.getTime() + 99_000); + }); + await harness.runner.runOnce(NOW); + expect(harness.getState().deliveries[0]).toMatchObject({ + attemptCount: 2, + status: 'processing', + }); + }, + ); + + it('5 treats stale preparation ownership as a no-op instead of semantic cancellation', async () => { + const harness = setup(); + harness.beforePreparationDeliveryRead((draft) => { + draft.deliveries[0].attemptCount = 2; + draft.deliveries[0].processingLeaseUntil = new KtDateTime( + NOW.getTime() + 99_000, + ); + }); + await harness.runner.runOnce(NOW); + expect(harness.sender.sendStrictPlainText).not.toHaveBeenCalled(); + expect(harness.getState().deliveries[0]).toMatchObject({ + attemptCount: 2, + lastErrorCode: 'old_error', + status: 'processing', + }); + }); + + it('6-8 sends exact frozen input after both transactions commit and ignores current template identity', async () => { + const harness = setup({ + bindings: [binding({ templateId: 'new-template' })], + }); + await harness.runner.runOnce(NOW); + expect(harness.sender.sendStrictPlainText).toHaveBeenCalledWith({ + attemptNumber: 1, + deliveryId: '801', + message: 'endpoint=pal.example.com:38213 port=38213', + selfId: 'bot-a', + targetId: 'group-1', + targetType: 'group', + }); + expect(harness.senderDepths).toEqual([0]); + expect(harness.operations.slice(0, 3)).toEqual([ + 'claim:commit', + 'prepare:commit', + 'send', + ]); + expect(harness.preparationClaimLocks).toEqual([false]); + expect(harness.lockOrder).toEqual([ + 'QqbotMessageEvent', + 'QqbotMessageSubscription', + 'QqbotMessagePublishBinding', + 'QqbotMessagePublishTarget', + 'QqbotAccount', + 'QqbotMessageDelivery', + ]); + expect(harness.getState().deliveries[0]).toMatchObject({ + lastErrorCode: null, + lastErrorMessage: null, + nextAttemptAt: null, + processingLeaseUntil: null, + sendLogId: 'send-log-1', + status: 'success', + }); + }); + + it.each([ + ['unknown variable', { endpoint: 'x', extra: true, port: 1 }], + ['missing variable', { endpoint: 'x' }], + ['null variable', { endpoint: 'x', port: null }], + ['wrong variable type', { endpoint: 'x', port: '1' }], + ])( + '9 permanently rejects a %s in the frozen snapshot', + async (_name, snapshot) => { + const harness = setup({ + deliveries: [ + delivery({ + ...(String(_name) === 'missing variable' + ? { + renderedMessage: 'endpoint=x', + templateContent: 'endpoint=${{endpoint}}', + } + : {}), + variableSnapshot: snapshot, + }), + ], + }); + await harness.runner.runOnce(NOW); + expect(harness.sender.sendStrictPlainText).not.toHaveBeenCalled(); + expect(harness.getState().deliveries[0]).toMatchObject({ + lastErrorCode: 'template_invalid', + status: 'failed', + }); + }, + ); + + it.each([ + ['invalid syntax', 'endpoint=${{endpoint}', 'irrelevant'], + [ + 'render mismatch', + 'endpoint=${{endpoint}} port=${{port}}', + 'rewritten current content', + ], + ])( + '9 permanently rejects frozen %s', + async (_name, templateContent, renderedMessage) => { + const harness = setup({ + deliveries: [delivery({ renderedMessage, templateContent })], + }); + await harness.runner.runOnce(NOW); + expect(harness.sender.sendStrictPlainText).not.toHaveBeenCalled(); + expect(harness.getState().deliveries[0].status).toBe('failed'); + }, + ); + + it('9 permanently rejects a corrupt frozen target type before configuration mismatch handling', async () => { + const harness = setup({ + deliveries: [delivery({ targetType: 'channel' as never })], + }); + await harness.runner.runOnce(NOW); + expect(harness.sender.sendStrictPlainText).not.toHaveBeenCalled(); + expect(harness.getState().deliveries[0]).toMatchObject({ + lastErrorCode: 'invalid_target_type', + status: 'failed', + }); + }); + + it('9 validates immutable event payload before readiness and fails corrupt payload permanently', async () => { + const harness = setup(); + harness.adapter.validateEventPayload.mockImplementationOnce(() => { + throw new SystemMessageContractError('invalid_source_config'); + }); + await harness.runner.runOnce(NOW); + expect(harness.adapter.resolveDelivery).not.toHaveBeenCalled(); + expect(harness.getState().deliveries[0]).toMatchObject({ + lastErrorCode: 'invalid_source_config', + status: 'failed', + }); + }); + + it('10 waits 60 seconds for DDNS while preserving the latest log', async () => { + const harness = setup(); + harness.adapter.resolveDelivery.mockResolvedValueOnce({ + reasonCode: 'ddns_not_synced', + status: 'waiting_ddns', + variables: { endpoint: 'current', port: 38213 }, + }); + await harness.runner.runOnce(NOW); + expect(harness.sender.sendStrictPlainText).not.toHaveBeenCalled(); + expect(harness.getState().deliveries[0]).toMatchObject({ + lastErrorCode: 'ddns_not_synced', + nextAttemptAt: new KtDateTime( + NOW.getTime() + SYSTEM_MESSAGE_DDNS_RECHECK_MS, + ), + processingLeaseUntil: null, + sendLogId: 'old-log', + status: 'waiting_ddns', + }); + }); + + it('11 sends a due waiting row without replacing frozen variables with readiness variables', async () => { + const harness = setup({ + deliveries: [delivery({ status: 'waiting_ddns' })], + }); + await harness.runner.runOnce(NOW); + expect(harness.sender.sendStrictPlainText).toHaveBeenCalledWith( + expect.objectContaining({ + message: 'endpoint=pal.example.com:38213 port=38213', + }), + ); + expect(harness.getState().deliveries[0].status).toBe('success'); + }); + + it.each([ + ['endpoint_port_changed', 'superseded'], + ['endpoint_ipv4_changed', 'superseded'], + ['endpoint_lease_expired', 'superseded'], + ] as const)( + '12 persists adapter %s as superseded', + async (reasonCode, status) => { + const harness = setup(); + harness.adapter.resolveDelivery.mockResolvedValueOnce({ + reasonCode, + status, + }); + await harness.runner.runOnce(NOW); + expect(harness.sender.sendStrictPlainText).not.toHaveBeenCalled(); + expect(harness.getState().deliveries[0]).toMatchObject({ + lastErrorCode: reasonCode, + nextAttemptAt: null, + processingLeaseUntil: null, + status: 'superseded', + }); + }, + ); + + it.each([ + ['missing subscription', { subscriptions: [] }], + [ + 'disabled subscription', + { subscriptions: [subscription({ enabled: false })] }, + ], + ['deleted binding', { bindings: [binding({ isDeleted: true })] }], + [ + 'mismatched binding', + { bindings: [binding({ subscriptionId: 'other' })] }, + ], + ['disabled target', { targets: [target({ enabled: false })] }], + ['changed target', { targets: [target({ targetId: 'other' })] }], + ])('13 cancels %s configuration', async (_name, seed) => { + const harness = setup(seed); + await harness.runner.runOnce(NOW); + expect(harness.sender.sendStrictPlainText).not.toHaveBeenCalled(); + expect(harness.getState().deliveries[0].status).toBe('cancelled'); + }); + + it.each([ + ['missing account', { accounts: [] }], + ['disabled account', { accounts: [account({ enabled: false })] }], + ['deleted account', { accounts: [account({ isDeleted: true })] }], + ['binding self mismatch', { bindings: [binding({ selfId: 'bot-b' })] }], + ['account self mismatch', { accounts: [account({ selfId: 'bot-b' })] }], + ])( + '14 cancels %s without a default-account fallback', + async (_name, seed) => { + const harness = setup(seed); + await harness.runner.runOnce(NOW); + expect(harness.sender.sendStrictPlainText).not.toHaveBeenCalled(); + expect(harness.getState().deliveries[0].status).toBe('cancelled'); + }, + ); + + it('15 sends with the exact enabled account despite runtime-offline fields and retries disconnect', async () => { + const harness = setup({ + accounts: [ + account({ + connectStatus: 'offline', + oneBotStatus: 'offline', + }), + ], + }); + harness.sender.sendStrictPlainText.mockRejectedValueOnce( + new QqbotSendAttemptError({ + code: 'onebot_disconnected', + message: 'runtime offline', + retryable: true, + sendLogId: null, + }), + ); + await harness.runner.runOnce(NOW); + expect(harness.sender.sendStrictPlainText).toHaveBeenCalledWith( + expect.objectContaining({ selfId: 'bot-a' }), + ); + expect(harness.getState().deliveries[0].status).toBe('retry'); + }); + + it.each([ + ['onebot_rejected', 'latest-log'], + ['invalid_target_type', null], + ])( + '16 permanently fails typed %s and retains the latest available log', + async (code, sendLogId) => { + const harness = setup(); + harness.sender.sendStrictPlainText.mockRejectedValueOnce( + new QqbotSendAttemptError({ + code, + message: 'safe rejection', + retryable: false, + sendLogId, + }), + ); + await harness.runner.runOnce(NOW); + expect(harness.getState().deliveries[0]).toMatchObject({ + lastErrorCode: code, + sendLogId: sendLogId ?? 'old-log', + status: 'failed', + }); + }, + ); + + it('17 retries a typed first failure after 10 seconds with exact safe details', async () => { + const harness = setup(); + harness.sender.sendStrictPlainText.mockRejectedValueOnce( + new QqbotSendAttemptError({ + code: 'onebot_timeout', + message: 'OneBot action timeout', + retryable: true, + sendLogId: '90001', + }), + ); + await harness.runner.runOnce(NOW); + expect(harness.getState().deliveries[0]).toMatchObject({ + attemptCount: 1, + lastErrorCode: 'onebot_timeout', + lastErrorMessage: 'OneBot action timeout', + nextAttemptAt: new KtDateTime(NOW.getTime() + 10_000), + processingLeaseUntil: null, + sendLogId: '90001', + status: 'retry', + }); + }); + + it('18 caps exponential backoff at fifteen minutes', async () => { + const harness = setup({ + deliveries: [delivery({ attemptCount: 19 })], + }); + harness.sender.sendStrictPlainText.mockRejectedValueOnce( + new QqbotSendAttemptError({ + code: 'onebot_timeout', + message: 'timeout', + retryable: true, + sendLogId: null, + }), + ); + await harness.runner.runOnce(NOW); + expect(harness.getState().deliveries[0].nextAttemptAt).toEqual( + new KtDateTime(NOW.getTime() + 15 * 60_000), + ); + expect(deliveryRetryDelayMs(20)).toBe(15 * 60_000); + }); + + it.each([ + ['exact expiry', NOW.getTime(), 'failed'], + ['after expiry', NOW.getTime() - 1, 'failed'], + ['retry reaches expiry', NOW.getTime() + 10_000, 'failed'], + ['retry remains inside', NOW.getTime() + 10_001, 'retry'], + ] as const)( + '19 enforces the event-based 24h boundary: %s', + async (_name, expiresAt, status) => { + const harness = setup({ + deliveries: [delivery({ expiresAt: new KtDateTime(expiresAt) })], + }); + harness.sender.sendStrictPlainText.mockRejectedValueOnce( + new QqbotSendAttemptError({ + code: 'onebot_timeout', + message: 'timeout', + retryable: true, + sendLogId: null, + }), + ); + await harness.runner.runOnce(NOW); + expect(harness.getState().deliveries[0].status).toBe(status); + if (expiresAt <= NOW.getTime()) + expect(harness.sender.sendStrictPlainText).not.toHaveBeenCalled(); + }, + ); + + it('20 retries an ambiguous timeout with the same delivery ID and next attempt number', async () => { + const harness = setup(); + harness.sender.sendStrictPlainText + .mockRejectedValueOnce( + new QqbotSendAttemptError({ + code: 'onebot_timeout', + message: 'ambiguous external result', + retryable: true, + sendLogId: 'attempt-log-1', + }), + ) + .mockResolvedValueOnce({ logId: 'attempt-log-2' }); + await harness.runner.runOnce(NOW); + await harness.runner.runOnce(new Date(NOW.getTime() + 10_000)); + expect(harness.sender.sendStrictPlainText.mock.calls).toEqual([ + [expect.objectContaining({ attemptNumber: 1, deliveryId: '801' })], + [expect.objectContaining({ attemptNumber: 2, deliveryId: '801' })], + ]); + expect(harness.getState().deliveries[0]).toMatchObject({ + attemptCount: 2, + sendLogId: 'attempt-log-2', + status: 'success', + }); + }); + + it('21 sanitizes an untyped infrastructure failure without persisting raw error content', async () => { + const harness = setup(); + harness.sender.sendStrictPlainText.mockRejectedValueOnce( + new Error('password=secret\nSELECT * FROM private_table'), + ); + await harness.runner.runOnce(NOW); + expect(harness.getState().deliveries[0]).toMatchObject({ + lastErrorCode: 'delivery_transient_error', + lastErrorMessage: 'delivery transport unavailable', + status: 'retry', + }); + expect(JSON.stringify(harness.getState().deliveries[0])).not.toContain( + 'secret', + ); + }); + + it('21 retries registry and adapter infrastructure errors instead of classifying them as corrupt payload', async () => { + const harness = setup(); + harness.adapter.resolveDelivery.mockRejectedValueOnce( + new SystemMessageContractError('adapter_dependency_unavailable'), + ); + await harness.runner.runOnce(NOW); + expect(harness.getState().deliveries[0]).toMatchObject({ + lastErrorCode: 'delivery_transient_error', + status: 'retry', + }); + + const unavailable = setup(); + unavailable.registry.unregister(SOURCE_KEY, unavailable.adapter); + await unavailable.runner.runOnce(NOW); + expect(unavailable.getState().deliveries[0]).toMatchObject({ + lastErrorCode: 'delivery_transient_error', + status: 'retry', + }); + }); + + it('22 lets a sibling succeed after another target receives a typed send failure', async () => { + const harness = setup({ + deliveries: [ + delivery({ id: '801', publishTargetId: '701' }), + delivery({ id: '802', publishTargetId: '702' }), + ], + targets: [target(), target({ id: '702' })], + }); + harness.sender.sendStrictPlainText.mockRejectedValueOnce( + new QqbotSendAttemptError({ + code: 'onebot_timeout', + message: 'timeout', + retryable: true, + sendLogId: 'failed-log', + }), + ); + expect(await harness.runner.runOnce(NOW)).toBe(2); + expect(harness.sender.sendStrictPlainText).toHaveBeenCalledTimes(2); + expect(harness.getState().deliveries.map((item) => item.status)).toEqual([ + 'retry', + 'success', + ]); + }); + + it('22 isolates a failed final persistence path so a sibling target still succeeds', async () => { + const harness = setup({ + deliveries: [ + delivery({ id: '801', publishTargetId: '701' }), + delivery({ id: '802', publishTargetId: '702' }), + ], + targets: [target(), target({ id: '702' })], + }); + harness.failExternalUpdates(3); + expect(await harness.runner.runOnce(NOW)).toBe(2); + expect(harness.sender.sendStrictPlainText).toHaveBeenCalledTimes(2); + expect( + harness.getState().deliveries.find((item) => item.id === '802')?.status, + ).toBe('success'); + }); + + it('uses the exact 30-second claim lease and clears it on every persisted result', async () => { + const harness = setup(); + const claimLease: Array = []; + harness.beforeExternalUpdate((state) => { + claimLease.push(state.deliveries[0].processingLeaseUntil); + }); + await harness.runner.runOnce(NOW); + expect(claimLease[0]).toEqual( + new KtDateTime(NOW.getTime() + SYSTEM_MESSAGE_LEASE_MS), + ); + expect(harness.getState().deliveries[0].processingLeaseUntil).toBeNull(); + }); +}); + +describe('System message coordinator direct lifecycle contracts', () => { + afterEach(() => { + jest.useRealTimers(); + jest.restoreAllMocks(); + }); + + it('33 coalesces wakes without overlap and performs exactly one necessary follow-up', async () => { + const first = deferred(); + let active = 0; + let maximumActive = 0; + const fanout = { + runOnce: jest.fn(async () => { + active += 1; + maximumActive = Math.max(maximumActive, active); + if (fanout.runOnce.mock.calls.length === 1) await first.promise; + active -= 1; + return 0; + }), + }; + const deliveryRunner = { runOnce: jest.fn().mockResolvedValue(0) }; + const coordinator = new SystemMessageDeliveryCoordinatorService( + {} as never, + fanout as never, + deliveryRunner as never, + ); + coordinator.requestDrain(); + await flushPromises(); + coordinator.requestDrain(); + coordinator.requestDrain(); + first.resolve(); + await flushPromises(); + expect(maximumActive).toBe(1); + expect(fanout.runOnce).toHaveBeenCalledTimes(2); + expect(deliveryRunner.runOnce).toHaveBeenCalledTimes(2); + }); + + it('34 runs fan-out before delivery on every pass and isolates either rejection', async () => { + const calls: string[] = []; + const fanout = { + runOnce: jest + .fn() + .mockImplementationOnce(async () => { + calls.push('fanout-1'); + throw new Error('fanout failed'); + }) + .mockImplementationOnce(async () => { + calls.push('fanout-2'); + return 0; + }), + }; + const deliveryRunner = { + runOnce: jest + .fn() + .mockImplementationOnce(async () => { + calls.push('delivery-1'); + return 0; + }) + .mockImplementationOnce(async () => { + calls.push('delivery-2'); + throw new Error('delivery failed'); + }), + }; + const coordinator = new SystemMessageDeliveryCoordinatorService( + {} as never, + fanout as never, + deliveryRunner as never, + ); + coordinator.requestDrain(); + await flushPromises(); + coordinator.requestDrain(); + await flushPromises(); + expect(calls).toEqual(['fanout-1', 'delivery-1', 'fanout-2', 'delivery-2']); + }); + + it('35 immediately drains a full bounded batch and stops below 50 without another wake', async () => { + const fanout = { + runOnce: jest + .fn() + .mockResolvedValueOnce(SYSTEM_MESSAGE_BATCH_SIZE) + .mockResolvedValueOnce(SYSTEM_MESSAGE_BATCH_SIZE - 1), + }; + const deliveryRunner = { runOnce: jest.fn().mockResolvedValue(0) }; + const coordinator = new SystemMessageDeliveryCoordinatorService( + {} as never, + fanout as never, + deliveryRunner as never, + ); + coordinator.requestDrain(); + await flushPromises(); + expect(fanout.runOnce).toHaveBeenCalledTimes(2); + expect(deliveryRunner.runOnce).toHaveBeenCalledTimes(2); + }); + + it('36-37 installs one unref interval and a post-module-init zero-delay startup wake', async () => { + jest.useFakeTimers(); + const fanout = { runOnce: jest.fn().mockResolvedValue(0) }; + const deliveryRunner = { runOnce: jest.fn().mockResolvedValue(0) }; + const intervalSpy = jest.spyOn(global, 'setInterval'); + const timeoutSpy = jest.spyOn(global, 'setTimeout'); + const coordinator = new SystemMessageDeliveryCoordinatorService( + {} as never, + fanout as never, + deliveryRunner as never, + ); + coordinator.onModuleInit(); + expect(fanout.runOnce).not.toHaveBeenCalled(); + expect(intervalSpy).toHaveBeenCalledTimes(1); + expect(intervalSpy).toHaveBeenCalledWith( + expect.any(Function), + SYSTEM_MESSAGE_SCAN_INTERVAL_MS, + ); + expect(timeoutSpy).toHaveBeenCalledWith(expect.any(Function), 0); + expect( + (intervalSpy.mock.results[0].value as NodeJS.Timeout).hasRef?.(), + ).toBe(false); + expect( + (timeoutSpy.mock.results[0].value as NodeJS.Timeout).hasRef?.(), + ).toBe(false); + expect(jest.getTimerCount()).toBe(2); + jest.advanceTimersByTime(0); + await flushPromises(); + expect(fanout.runOnce).toHaveBeenCalledTimes(1); + jest.advanceTimersByTime(SYSTEM_MESSAGE_SCAN_INTERVAL_MS); + await flushPromises(); + expect(fanout.runOnce).toHaveBeenCalledTimes(2); + await coordinator.onModuleDestroy(); + expect(jest.getTimerCount()).toBe(0); + }); + + it.each([ + 'wake before destroy', + 'wake during active pass', + 'wake after destroy', + ])( + '38 clears timers, awaits active work, and suppresses post-destroy calls: %s', + async (race) => { + jest.useFakeTimers(); + const active = deferred(); + const fanout = { + runOnce: jest.fn(async () => { + if (fanout.runOnce.mock.calls.length === 1) await active.promise; + return 0; + }), + }; + const deliveryRunner = { runOnce: jest.fn().mockResolvedValue(0) }; + const coordinator = new SystemMessageDeliveryCoordinatorService( + {} as never, + fanout as never, + deliveryRunner as never, + ); + coordinator.onModuleInit(); + if (race === 'wake after destroy') { + await coordinator.onModuleDestroy(); + coordinator.requestDrain(); + jest.runAllTimers(); + await flushPromises(); + expect(fanout.runOnce).not.toHaveBeenCalled(); + expect(jest.getTimerCount()).toBe(0); + return; + } + coordinator.requestDrain(); + if (race !== 'wake before destroy') { + await flushPromises(); + if (race === 'wake during active pass') coordinator.requestDrain(); + } + const destroyed = coordinator.onModuleDestroy(); + let destroySettled = false; + void destroyed.then(() => { + destroySettled = true; + }); + await Promise.resolve(); + if (fanout.runOnce.mock.calls.length > 0) + expect(destroySettled).toBe(false); + active.resolve(); + await destroyed; + coordinator.requestDrain(); + jest.runAllTimers(); + await flushPromises(); + expect(fanout.runOnce).toHaveBeenCalledTimes(1); + expect(jest.getTimerCount()).toBe(0); + }, + ); + + it('38 suppresses a wake queued after loop completion but before promise finally', async () => { + jest.useFakeTimers(); + let destroyPromise: Promise | undefined; + const fanout = { runOnce: jest.fn().mockResolvedValue(0) }; + const deliveryRunner = { + runOnce: jest.fn( + () => + new Promise((resolve) => { + queueMicrotask(() => { + resolve(0); + queueMicrotask(() => { + queueMicrotask(() => { + coordinator.requestDrain(); + destroyPromise = coordinator.onModuleDestroy(); + }); + }); + }); + }), + ), + }; + const coordinator = new SystemMessageDeliveryCoordinatorService( + {} as never, + fanout as never, + deliveryRunner as never, + ); + coordinator.onModuleInit(); + coordinator.requestDrain(); + await flushPromises(16); + await destroyPromise; + jest.runAllTicks(); + await flushPromises(); + expect(fanout.runOnce).toHaveBeenCalledTimes(1); + expect(deliveryRunner.runOnce).toHaveBeenCalledTimes(1); + expect(jest.getTimerCount()).toBe(0); + }); +});