feat: 持久化扇出系统消息事件
This commit is contained in:
parent
31c1d23387
commit
203b7771f6
@ -0,0 +1,613 @@
|
||||
import { Injectable } from '@nestjs/common';
|
||||
import { DataSource, Brackets, type EntityManager } from 'typeorm';
|
||||
import { KtDateTime } from '@/common';
|
||||
import {
|
||||
SystemMessageContractError,
|
||||
type SystemMessageDeliveryReadiness,
|
||||
type SystemMessageScalar,
|
||||
type SystemMessageSourceAdapter,
|
||||
} 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 { QqbotMessageTemplate } from '../../infrastructure/persistence/message-push/qqbot-message-template.entity';
|
||||
import {
|
||||
SYSTEM_MESSAGE_BATCH_SIZE,
|
||||
SYSTEM_MESSAGE_DDNS_RECHECK_MS,
|
||||
SYSTEM_MESSAGE_LEASE_MS,
|
||||
SYSTEM_MESSAGE_RETRY_BASE_MS,
|
||||
SYSTEM_MESSAGE_RETRY_MAX_MS,
|
||||
SYSTEM_MESSAGE_RETRY_WINDOW_MS,
|
||||
} from './system-message-runner.constants';
|
||||
import { SystemMessageSourceRegistry } from './system-message-source.registry';
|
||||
import { SystemMessageTemplateRendererService } from './system-message-template-renderer.service';
|
||||
|
||||
const STUN_MAPPING_PORT_SOURCE = 'network.stun.mapping-port-changed';
|
||||
const TRANSIENT_ERROR_CODE = 'fanout_transient_error';
|
||||
const EVENT_EXPIRED_ERROR_CODE = 'fanout_expired';
|
||||
const EVENT_RESOURCE_MISMATCH_ERROR_CODE = 'event_resource_mismatch';
|
||||
const SUPERSEDED_STATUSES = new Set(['waiting_ddns', 'pending', 'retry']);
|
||||
|
||||
interface ClaimToken {
|
||||
attempt: number;
|
||||
event: QqbotMessageEvent;
|
||||
leaseUntil: KtDateTime;
|
||||
}
|
||||
|
||||
/**
|
||||
* Creates frozen, idempotent delivery work from committed system-message Outbox facts.
|
||||
*
|
||||
* It deliberately only persists work. The later delivery coordinator owns DDNS wakeups
|
||||
* and every OneBot call.
|
||||
*/
|
||||
@Injectable()
|
||||
export class SystemMessageFanoutService {
|
||||
/**
|
||||
* Creates the durable Outbox fan-out service.
|
||||
* @param dataSource - Core database source used for short claim and subscription transactions.
|
||||
* @param sourceRegistry - Registered source adapters for payload validation and readiness.
|
||||
* @param templateRenderer - Safe literal-only renderer used to freeze final text.
|
||||
*/
|
||||
constructor(
|
||||
private readonly dataSource: DataSource,
|
||||
private readonly sourceRegistry: SystemMessageSourceRegistry,
|
||||
private readonly templateRenderer: SystemMessageTemplateRendererService,
|
||||
) {}
|
||||
|
||||
/**
|
||||
* Claims and processes at most one bounded batch of due Outbox events.
|
||||
* @param now - Stable clock instant for leases, schedules, and retry decisions.
|
||||
* @returns Count of rows successfully claimed, including rows that become retry or failed.
|
||||
*/
|
||||
async runOnce(now: Date = new Date()): Promise<number> {
|
||||
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;
|
||||
await this.processClaim(token, now);
|
||||
}
|
||||
return claimed;
|
||||
}
|
||||
|
||||
/**
|
||||
* Atomically claims the oldest due Outbox event using an exact lease ownership token.
|
||||
* @param now - Clock instant against which due rows and expired leases are evaluated.
|
||||
* @returns Claimed event with its new attempt and lease, or null when no eligible row exists.
|
||||
*/
|
||||
private async claimOne(now: Date): Promise<ClaimToken | null> {
|
||||
return this.dataSource.transaction(async (manager) => {
|
||||
const events = manager.getRepository(QqbotMessageEvent);
|
||||
const event = await events
|
||||
.createQueryBuilder('event')
|
||||
.setLock('pessimistic_write')
|
||||
.setOnLocked('skip_locked')
|
||||
.where(
|
||||
new Brackets((where) => {
|
||||
where
|
||||
.where(
|
||||
'event.fanoutStatus IN (:...due) AND event.nextFanoutAt <= :now',
|
||||
{ due: ['accepted', 'retry'], now },
|
||||
)
|
||||
.orWhere(
|
||||
'event.fanoutStatus = :processing AND event.fanoutLeaseUntil <= :now',
|
||||
{ processing: 'processing', now },
|
||||
);
|
||||
}),
|
||||
)
|
||||
.orderBy('event.occurredAt', 'ASC')
|
||||
.addOrderBy('event.id', 'ASC')
|
||||
.take(1)
|
||||
.getOne();
|
||||
if (!event) return null;
|
||||
|
||||
const leaseUntil = new KtDateTime(
|
||||
now.getTime() + SYSTEM_MESSAGE_LEASE_MS,
|
||||
);
|
||||
event.fanoutAttemptCount += 1;
|
||||
event.fanoutLeaseUntil = leaseUntil;
|
||||
event.fanoutStatus = 'processing';
|
||||
event.nextFanoutAt = null;
|
||||
await events.save(event);
|
||||
return { attempt: event.fanoutAttemptCount, event, leaseUntil };
|
||||
});
|
||||
}
|
||||
|
||||
/**
|
||||
* Validates one owned event and fans it out without allowing stale owners to finish it.
|
||||
* @param token - Attempt-plus-lease token returned by the successful claim transaction.
|
||||
* @param now - Stable clock instant for expiry and next retry time.
|
||||
*/
|
||||
private async processClaim(token: ClaimToken, now: Date): Promise<void> {
|
||||
if (this.isExpired(token.event, now)) {
|
||||
await this.finish(
|
||||
token,
|
||||
'failed',
|
||||
EVENT_EXPIRED_ERROR_CODE,
|
||||
'fan-out deadline reached',
|
||||
);
|
||||
return;
|
||||
}
|
||||
|
||||
try {
|
||||
const adapter = this.sourceRegistry.get(token.event.sourceKey);
|
||||
const payload = adapter.validateEventPayload(token.event.payload);
|
||||
this.assertResourceIdentity(token.event, payload);
|
||||
const subscriptions = await this.findMatchingSubscriptions(token.event);
|
||||
let transientFailure = false;
|
||||
|
||||
for (const subscription of subscriptions) {
|
||||
try {
|
||||
await this.dataSource.transaction((manager) =>
|
||||
this.fanOutSubscription(
|
||||
manager,
|
||||
token.event,
|
||||
subscription.id,
|
||||
adapter,
|
||||
payload,
|
||||
now,
|
||||
),
|
||||
);
|
||||
} catch {
|
||||
transientFailure = true;
|
||||
}
|
||||
}
|
||||
|
||||
if (transientFailure) {
|
||||
await this.retryOrFail(
|
||||
token,
|
||||
now,
|
||||
TRANSIENT_ERROR_CODE,
|
||||
'fan-out dependency unavailable',
|
||||
);
|
||||
return;
|
||||
}
|
||||
await this.finish(token, 'completed', null, null);
|
||||
} catch (error) {
|
||||
if (error instanceof SystemMessageContractError) {
|
||||
await this.finish(token, 'failed', error.code, this.safeMessage(error));
|
||||
return;
|
||||
}
|
||||
if (error instanceof EventResourceMismatchError) {
|
||||
await this.finish(
|
||||
token,
|
||||
'failed',
|
||||
EVENT_RESOURCE_MISMATCH_ERROR_CODE,
|
||||
error.message,
|
||||
);
|
||||
return;
|
||||
}
|
||||
await this.retryOrFail(
|
||||
token,
|
||||
now,
|
||||
TRANSIENT_ERROR_CODE,
|
||||
'fan-out dependency unavailable',
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Selects active subscriptions whose own JSON resource field exactly matches the event.
|
||||
* @param event - Frozen Outbox event being fanned out.
|
||||
* @returns Deterministically ordered eligible subscription rows.
|
||||
*/
|
||||
private async findMatchingSubscriptions(
|
||||
event: QqbotMessageEvent,
|
||||
): Promise<QqbotMessageSubscription[]> {
|
||||
const subscriptions = await this.dataSource
|
||||
.getRepository(QqbotMessageSubscription)
|
||||
.find({
|
||||
where: { enabled: true, isDeleted: false, sourceKey: event.sourceKey },
|
||||
order: { id: 'ASC' },
|
||||
});
|
||||
return subscriptions.filter((subscription) =>
|
||||
this.matchesSubscription(subscription, event),
|
||||
);
|
||||
}
|
||||
|
||||
/**
|
||||
* Executes one subscription's isolated persistence unit.
|
||||
* @param manager - Transaction manager whose mutations commit only for this subscription.
|
||||
* @param event - Frozen Outbox event currently owned by the caller.
|
||||
* @param subscriptionId - Primary key of the subscription to lock and recheck.
|
||||
* @param adapter - Source adapter already used to validate the frozen payload.
|
||||
* @param payload - Validated scalar payload passed to the adapter exactly once.
|
||||
* @param now - Stable scheduling instant for newly created delivery rows.
|
||||
*/
|
||||
private async fanOutSubscription(
|
||||
manager: EntityManager,
|
||||
event: QqbotMessageEvent,
|
||||
subscriptionId: string,
|
||||
adapter: SystemMessageSourceAdapter,
|
||||
payload: Record<string, SystemMessageScalar>,
|
||||
now: Date,
|
||||
): Promise<void> {
|
||||
const subscriptions = manager.getRepository(QqbotMessageSubscription);
|
||||
const subscription = await subscriptions.findOne({
|
||||
where: { id: subscriptionId },
|
||||
lock: { mode: 'pessimistic_write' },
|
||||
});
|
||||
if (!subscription || !this.matchesSubscription(subscription, event)) return;
|
||||
|
||||
const readiness = await adapter.resolveDelivery({
|
||||
eventPayload: payload,
|
||||
subscriptionConfig: subscription.sourceConfig,
|
||||
});
|
||||
await this.supersedeEarlierDeliveries(manager, event, subscription.id);
|
||||
if (!this.hasRenderableVariables(readiness)) return;
|
||||
|
||||
await this.createDeliveries(manager, event, subscription, readiness, now);
|
||||
}
|
||||
|
||||
/**
|
||||
* Supersedes only unfinished deliveries of events strictly earlier than the current event.
|
||||
* @param manager - Current subscription transaction manager.
|
||||
* @param event - Current event that supersedes older work for this subscription.
|
||||
* @param subscriptionId - Subscription scope that prevents unrelated fan-out interference.
|
||||
*/
|
||||
private async supersedeEarlierDeliveries(
|
||||
manager: EntityManager,
|
||||
event: QqbotMessageEvent,
|
||||
subscriptionId: string,
|
||||
): Promise<void> {
|
||||
const events = manager.getRepository(QqbotMessageEvent);
|
||||
const priorEvents = await events.find({
|
||||
where: { resourceKey: event.resourceKey, sourceKey: event.sourceKey },
|
||||
});
|
||||
const priorIds = new Set(
|
||||
priorEvents
|
||||
.filter((candidate) => this.isStrictlyEarlier(candidate, event))
|
||||
.map((candidate) => candidate.id),
|
||||
);
|
||||
if (!priorIds.size) return;
|
||||
|
||||
const deliveries = manager.getRepository(QqbotMessageDelivery);
|
||||
const candidates = await deliveries.find({ where: { subscriptionId } });
|
||||
for (const delivery of candidates) {
|
||||
if (
|
||||
priorIds.has(delivery.messageEventId) &&
|
||||
SUPERSEDED_STATUSES.has(delivery.status)
|
||||
) {
|
||||
delivery.status = 'superseded';
|
||||
await deliveries.save(delivery);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Builds every currently legal binding/target delivery with frozen snapshots.
|
||||
* @param manager - Current isolated subscription transaction manager.
|
||||
* @param event - Frozen Outbox event.
|
||||
* @param subscription - Locked and rechecked active subscription.
|
||||
* @param readiness - Ready or DDNS-waiting variables from the source adapter.
|
||||
* @param now - Stable scheduling instant for the newly created rows.
|
||||
*/
|
||||
private async createDeliveries(
|
||||
manager: EntityManager,
|
||||
event: QqbotMessageEvent,
|
||||
subscription: QqbotMessageSubscription,
|
||||
readiness: Extract<
|
||||
SystemMessageDeliveryReadiness,
|
||||
{ status: 'ready' | 'waiting_ddns' }
|
||||
>,
|
||||
now: Date,
|
||||
): Promise<void> {
|
||||
const bindings = await manager
|
||||
.getRepository(QqbotMessagePublishBinding)
|
||||
.find({
|
||||
where: {
|
||||
enabled: true,
|
||||
isDeleted: false,
|
||||
subscriptionId: subscription.id,
|
||||
},
|
||||
order: { id: 'ASC' },
|
||||
});
|
||||
const accounts = await manager.getRepository(QqbotAccount).find({
|
||||
where: { enabled: true, isDeleted: false },
|
||||
});
|
||||
const templates = await manager.getRepository(QqbotMessageTemplate).find({
|
||||
where: { enabled: true, isDeleted: false, sourceKey: event.sourceKey },
|
||||
});
|
||||
const targets = await manager
|
||||
.getRepository(QqbotMessagePublishTarget)
|
||||
.find({
|
||||
where: { enabled: true, isDeleted: false },
|
||||
order: { id: 'ASC' },
|
||||
});
|
||||
const accountById = new Map(
|
||||
accounts.map((account) => [account.id, account]),
|
||||
);
|
||||
const templateById = new Map(
|
||||
templates.map((template) => [template.id, template]),
|
||||
);
|
||||
|
||||
for (const binding of bindings) {
|
||||
const account = accountById.get(binding.accountId);
|
||||
const template = templateById.get(binding.templateId);
|
||||
if (!account || !template || account.selfId !== binding.selfId) continue;
|
||||
|
||||
let renderedMessage: string;
|
||||
try {
|
||||
renderedMessage = this.templateRenderer.render(
|
||||
template.content,
|
||||
readiness.variables,
|
||||
);
|
||||
} catch (error) {
|
||||
if (error instanceof SystemMessageContractError) continue;
|
||||
throw error;
|
||||
}
|
||||
for (const target of targets.filter(
|
||||
(item) => item.bindingId === binding.id,
|
||||
)) {
|
||||
if (target.targetType !== 'group' && target.targetType !== 'private') {
|
||||
continue;
|
||||
}
|
||||
await this.createDeliveryIfAbsent(
|
||||
manager,
|
||||
event,
|
||||
subscription,
|
||||
binding,
|
||||
template,
|
||||
target,
|
||||
readiness,
|
||||
renderedMessage,
|
||||
now,
|
||||
);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Persists one delivery, accepting only the exact event-target duplicate-key race.
|
||||
* @param manager - Current isolated subscription transaction manager.
|
||||
* @param event - Frozen source event.
|
||||
* @param subscription - Subscription snapshot identity.
|
||||
* @param binding - Active strict-account publishing binding.
|
||||
* @param template - Source-compatible template whose content is frozen.
|
||||
* @param target - Active target that defines this row's idempotency key.
|
||||
* @param readiness - Ready or DDNS-waiting result that supplies frozen variables.
|
||||
* @param renderedMessage - Already validated literal rendered content.
|
||||
* @param now - Stable schedule time for the new row.
|
||||
*/
|
||||
private async createDeliveryIfAbsent(
|
||||
manager: EntityManager,
|
||||
event: QqbotMessageEvent,
|
||||
subscription: QqbotMessageSubscription,
|
||||
binding: QqbotMessagePublishBinding,
|
||||
template: QqbotMessageTemplate,
|
||||
target: QqbotMessagePublishTarget,
|
||||
readiness: Extract<
|
||||
SystemMessageDeliveryReadiness,
|
||||
{ status: 'ready' | 'waiting_ddns' }
|
||||
>,
|
||||
renderedMessage: string,
|
||||
now: Date,
|
||||
): Promise<void> {
|
||||
const deliveries = manager.getRepository(QqbotMessageDelivery);
|
||||
const key = { messageEventId: event.id, publishTargetId: target.id };
|
||||
if (await deliveries.findOne({ where: key })) return;
|
||||
const isWaiting = readiness.status === 'waiting_ddns';
|
||||
const delivery = deliveries.create({
|
||||
attemptCount: 0,
|
||||
bindingId: binding.id,
|
||||
expiresAt: new KtDateTime(
|
||||
event.occurredAt.getTime() + SYSTEM_MESSAGE_RETRY_WINDOW_MS,
|
||||
),
|
||||
lastErrorCode: null,
|
||||
lastErrorMessage: null,
|
||||
messageEventId: event.id,
|
||||
nextAttemptAt: new KtDateTime(
|
||||
now.getTime() + (isWaiting ? SYSTEM_MESSAGE_DDNS_RECHECK_MS : 0),
|
||||
),
|
||||
processingLeaseUntil: null,
|
||||
publishTargetId: target.id,
|
||||
renderedMessage,
|
||||
selfId: binding.selfId,
|
||||
sendLogId: null,
|
||||
status: isWaiting ? 'waiting_ddns' : 'pending',
|
||||
subscriptionId: subscription.id,
|
||||
targetId: target.targetId,
|
||||
targetType: target.targetType,
|
||||
templateContent: template.content,
|
||||
templateId: template.id,
|
||||
variableSnapshot: structuredClone(readiness.variables),
|
||||
});
|
||||
try {
|
||||
await deliveries.save(delivery);
|
||||
} catch (error) {
|
||||
if (!this.isDuplicateKeyError(error)) throw error;
|
||||
const existing = await deliveries.findOne({ where: key });
|
||||
if (!existing) throw error;
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Completes, retries, or fails only while the original claim still owns the event.
|
||||
* @param token - Original attempt and exact lease ownership token.
|
||||
* @param status - Terminal or retry fan-out state to persist.
|
||||
* @param code - Safe stable error classification, or null for completion.
|
||||
* @param message - Bounded safe error summary, or null for completion.
|
||||
*/
|
||||
private async finish(
|
||||
token: ClaimToken,
|
||||
status: 'completed' | 'failed' | 'retry',
|
||||
code: null | string,
|
||||
message: null | string,
|
||||
): Promise<void> {
|
||||
await this.dataSource.getRepository(QqbotMessageEvent).update(
|
||||
{
|
||||
fanoutAttemptCount: token.attempt,
|
||||
fanoutLeaseUntil: token.leaseUntil,
|
||||
fanoutStatus: 'processing',
|
||||
id: token.event.id,
|
||||
},
|
||||
{
|
||||
fanoutLeaseUntil: null,
|
||||
fanoutStatus: status,
|
||||
lastErrorCode: code,
|
||||
lastErrorMessage: message,
|
||||
nextFanoutAt: null,
|
||||
},
|
||||
);
|
||||
}
|
||||
|
||||
/**
|
||||
* Schedules a safe retry only when another attempt can occur before the hard deadline.
|
||||
* @param token - Original claim ownership token.
|
||||
* @param now - Stable clock instant used to calculate retry timing.
|
||||
* @param code - Stable safe transient error code.
|
||||
* @param message - Sanitized transient error summary.
|
||||
*/
|
||||
private async retryOrFail(
|
||||
token: ClaimToken,
|
||||
now: Date,
|
||||
code: string,
|
||||
message: string,
|
||||
): Promise<void> {
|
||||
const delay = Math.min(
|
||||
SYSTEM_MESSAGE_RETRY_BASE_MS * 2 ** (token.attempt - 1),
|
||||
SYSTEM_MESSAGE_RETRY_MAX_MS,
|
||||
);
|
||||
const deadline =
|
||||
token.event.occurredAt.getTime() + SYSTEM_MESSAGE_RETRY_WINDOW_MS;
|
||||
if (now.getTime() + delay >= deadline) {
|
||||
await this.finish(
|
||||
token,
|
||||
'failed',
|
||||
EVENT_EXPIRED_ERROR_CODE,
|
||||
'fan-out deadline reached',
|
||||
);
|
||||
return;
|
||||
}
|
||||
await this.dataSource.getRepository(QqbotMessageEvent).update(
|
||||
{
|
||||
fanoutAttemptCount: token.attempt,
|
||||
fanoutLeaseUntil: token.leaseUntil,
|
||||
fanoutStatus: 'processing',
|
||||
id: token.event.id,
|
||||
},
|
||||
{
|
||||
fanoutLeaseUntil: null,
|
||||
fanoutStatus: 'retry',
|
||||
lastErrorCode: code,
|
||||
lastErrorMessage: message,
|
||||
nextFanoutAt: new KtDateTime(now.getTime() + delay),
|
||||
},
|
||||
);
|
||||
}
|
||||
|
||||
/**
|
||||
* Checks the strict current source/resource identity without coercing Snowflake strings.
|
||||
* @param event - Event whose immutable resource key is authoritative.
|
||||
* @param payload - Source-validated scalar payload.
|
||||
*/
|
||||
private assertResourceIdentity(
|
||||
event: QqbotMessageEvent,
|
||||
payload: Record<string, SystemMessageScalar>,
|
||||
): void {
|
||||
if (
|
||||
event.sourceKey === STUN_MAPPING_PORT_SOURCE &&
|
||||
payload.portForwardId !== event.resourceKey
|
||||
) {
|
||||
throw new EventResourceMismatchError();
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Checks whether a subscription remains active and has an own string resource ID match.
|
||||
* @param subscription - Candidate subscription row.
|
||||
* @param event - Current immutable Outbox event.
|
||||
* @returns Whether this row is a strict fan-out match.
|
||||
*/
|
||||
private matchesSubscription(
|
||||
subscription: QqbotMessageSubscription,
|
||||
event: QqbotMessageEvent,
|
||||
): boolean {
|
||||
const config = subscription.sourceConfig;
|
||||
return (
|
||||
subscription.enabled &&
|
||||
!subscription.isDeleted &&
|
||||
subscription.sourceKey === event.sourceKey &&
|
||||
!!config &&
|
||||
Object.prototype.hasOwnProperty.call(config, 'portForwardId') &&
|
||||
typeof config.portForwardId === 'string' &&
|
||||
config.portForwardId === event.resourceKey
|
||||
);
|
||||
}
|
||||
|
||||
/**
|
||||
* Compares source events by their required occurrence-time then string-ID ordering.
|
||||
* @param candidate - Potential historical event.
|
||||
* @param current - Event whose older deliveries may be superseded.
|
||||
* @returns Whether candidate is strictly earlier than current.
|
||||
*/
|
||||
private isStrictlyEarlier(
|
||||
candidate: QqbotMessageEvent,
|
||||
current: QqbotMessageEvent,
|
||||
): boolean {
|
||||
const difference =
|
||||
candidate.occurredAt.getTime() - current.occurredAt.getTime();
|
||||
return (
|
||||
difference < 0 ||
|
||||
(difference === 0 && BigInt(candidate.id) < BigInt(current.id))
|
||||
);
|
||||
}
|
||||
|
||||
/**
|
||||
* Checks whether creating a new delivery would violate the occurrence-based fan-out deadline.
|
||||
* @param event - Claimed Outbox event.
|
||||
* @param now - Stable invocation clock.
|
||||
* @returns Whether the event is at or beyond its 24-hour deadline.
|
||||
*/
|
||||
private isExpired(event: QqbotMessageEvent, now: Date): boolean {
|
||||
return (
|
||||
now.getTime() >=
|
||||
event.occurredAt.getTime() + SYSTEM_MESSAGE_RETRY_WINDOW_MS
|
||||
);
|
||||
}
|
||||
|
||||
/**
|
||||
* Narrows the source union to outcomes that provide variables required by delivery rows.
|
||||
* @param readiness - Source adapter result for the locked subscription.
|
||||
* @returns Whether a frozen ready or DDNS-waiting delivery can be persisted.
|
||||
*/
|
||||
private hasRenderableVariables(
|
||||
readiness: SystemMessageDeliveryReadiness,
|
||||
): readiness is Extract<
|
||||
SystemMessageDeliveryReadiness,
|
||||
{ status: 'ready' | 'waiting_ddns' }
|
||||
> {
|
||||
return readiness.status === 'ready' || readiness.status === 'waiting_ddns';
|
||||
}
|
||||
|
||||
/**
|
||||
* Recognizes MySQL's duplicate-key signal without treating other writes as idempotent.
|
||||
* @param error - Unknown persistence failure from an individual delivery insert.
|
||||
* @returns Whether the error is a MySQL duplicate-key failure.
|
||||
*/
|
||||
private isDuplicateKeyError(error: unknown): boolean {
|
||||
if (!error || typeof error !== 'object') return false;
|
||||
const record = error as { code?: unknown; errno?: unknown };
|
||||
return record.code === 'ER_DUP_ENTRY' || record.errno === 1062;
|
||||
}
|
||||
|
||||
/**
|
||||
* Converts a domain error into a bounded, non-stack persistence summary.
|
||||
* @param error - Domain contract error whose stable code is safe to retain.
|
||||
* @returns At most 500 characters of non-sensitive message text.
|
||||
*/
|
||||
private safeMessage(error: SystemMessageContractError): string {
|
||||
return error.message.slice(0, 500);
|
||||
}
|
||||
}
|
||||
|
||||
/** Signals frozen STUN payload identity that conflicts with the immutable event resource. */
|
||||
class EventResourceMismatchError extends Error {
|
||||
/** Creates the stable permanent event/resource mismatch classification. */
|
||||
constructor() {
|
||||
super('validated port-forward identity does not match event resource');
|
||||
}
|
||||
}
|
||||
@ -0,0 +1,14 @@
|
||||
/** Maximum due Outbox rows claimed by one fan-out scan. */
|
||||
export const SYSTEM_MESSAGE_BATCH_SIZE = 50;
|
||||
/** Duration of an exclusive fan-out claim lease. */
|
||||
export const SYSTEM_MESSAGE_LEASE_MS = 30_000;
|
||||
/** Initial fan-out retry delay. */
|
||||
export const SYSTEM_MESSAGE_RETRY_BASE_MS = 10_000;
|
||||
/** Maximum fan-out retry delay. */
|
||||
export const SYSTEM_MESSAGE_RETRY_MAX_MS = 15 * 60_000;
|
||||
/** Maximum age of an Outbox event that may create new deliveries. */
|
||||
export const SYSTEM_MESSAGE_RETRY_WINDOW_MS = 24 * 60 * 60_000;
|
||||
/** Persistent DDNS recheck interval for newly frozen waiting deliveries. */
|
||||
export const SYSTEM_MESSAGE_DDNS_RECHECK_MS = 60_000;
|
||||
/** Future coordinator scan cadence. */
|
||||
export const SYSTEM_MESSAGE_SCAN_INTERVAL_MS = 5_000;
|
||||
@ -49,6 +49,7 @@ import { QqbotMessageSubscription } from '@/modules/qqbot/core/infrastructure/pe
|
||||
import { QqbotMessageTemplate } from '@/modules/qqbot/core/infrastructure/persistence/message-push/qqbot-message-template.entity';
|
||||
import { SystemMessageSourceRegistry } from './application/message-push/system-message-source.registry';
|
||||
import { SystemMessageEventStagerService } from './application/message-push/system-message-event-stager.service';
|
||||
import { SystemMessageFanoutService } from './application/message-push/system-message-fanout.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';
|
||||
@ -92,6 +93,7 @@ export const QQBOT_CORE_CONTROLLERS = [
|
||||
export const QQBOT_CORE_PROVIDERS = [
|
||||
SystemMessageSourceRegistry,
|
||||
SystemMessageEventStagerService,
|
||||
SystemMessageFanoutService,
|
||||
{
|
||||
provide: SYSTEM_MESSAGE_EVENT_STAGER,
|
||||
useExisting: SystemMessageEventStagerService,
|
||||
|
||||
@ -15,6 +15,7 @@ import { QqbotPermissionController } from '../../../../src/modules/qqbot/core/co
|
||||
import { QqbotRuleController } from '../../../../src/modules/qqbot/core/contract/rule/qqbot-rule.controller';
|
||||
import { QqbotSendController } from '../../../../src/modules/qqbot/core/contract/send/qqbot-send.controller';
|
||||
import { SystemMessageEventStagerService } from '../../../../src/modules/qqbot/core/application/message-push/system-message-event-stager.service';
|
||||
import { 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 {
|
||||
QQBOT_CORE_CONTROLLERS,
|
||||
@ -171,6 +172,7 @@ describe('QQBot core module contract', () => {
|
||||
expect(QQBOT_CORE_PROVIDERS).toEqual(
|
||||
expect.arrayContaining([
|
||||
SystemMessageEventStagerService,
|
||||
SystemMessageFanoutService,
|
||||
{
|
||||
provide: SYSTEM_MESSAGE_EVENT_STAGER,
|
||||
useExisting: SystemMessageEventStagerService,
|
||||
|
||||
@ -0,0 +1,960 @@
|
||||
import { KtDateTime } from '../../../../src/common';
|
||||
import {
|
||||
SystemMessageContractError,
|
||||
type SystemMessageSourceAdapter,
|
||||
} from '../../../../src/modules/qqbot/core/contract/message-push/qqbot-message-push.types';
|
||||
import { SystemMessageFanoutService } from '../../../../src/modules/qqbot/core/application/message-push/system-message-fanout.service';
|
||||
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 { 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';
|
||||
import { QqbotMessageTemplate } from '../../../../src/modules/qqbot/core/infrastructure/persistence/message-push/qqbot-message-template.entity';
|
||||
import {
|
||||
SYSTEM_MESSAGE_BATCH_SIZE,
|
||||
SYSTEM_MESSAGE_DDNS_RECHECK_MS,
|
||||
SYSTEM_MESSAGE_LEASE_MS,
|
||||
SYSTEM_MESSAGE_RETRY_BASE_MS,
|
||||
SYSTEM_MESSAGE_RETRY_WINDOW_MS,
|
||||
} from '../../../../src/modules/qqbot/core/application/message-push/system-message-runner.constants';
|
||||
|
||||
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[];
|
||||
templates: QqbotMessageTemplate[];
|
||||
};
|
||||
|
||||
/** Builds a consistent current event whose payload preserves string Snowflake identity. */
|
||||
function event(overrides: Partial<QqbotMessageEvent> = {}): QqbotMessageEvent {
|
||||
return Object.assign(new QqbotMessageEvent(), {
|
||||
fanoutAttemptCount: 0,
|
||||
fanoutLeaseUntil: null,
|
||||
fanoutStatus: 'accepted',
|
||||
id: '200',
|
||||
lastErrorCode: null,
|
||||
lastErrorMessage: null,
|
||||
nextFanoutAt: new KtDateTime(NOW),
|
||||
occurredAt: new KtDateTime(NOW),
|
||||
payload: { endpoint: 'pal.example.com:38213', portForwardId: RESOURCE_KEY },
|
||||
resourceKey: RESOURCE_KEY,
|
||||
sourceKey: SOURCE_KEY,
|
||||
...overrides,
|
||||
});
|
||||
}
|
||||
|
||||
/** Builds an active strict JSON-matched subscription. */
|
||||
function subscription(
|
||||
overrides: Partial<QqbotMessageSubscription> = {},
|
||||
): QqbotMessageSubscription {
|
||||
return Object.assign(new QqbotMessageSubscription(), {
|
||||
enabled: true,
|
||||
id: '300',
|
||||
isDeleted: false,
|
||||
sourceConfig: { portForwardId: RESOURCE_KEY },
|
||||
sourceKey: SOURCE_KEY,
|
||||
...overrides,
|
||||
});
|
||||
}
|
||||
|
||||
/** Builds an enabled account without relying on its runtime online state. */
|
||||
function account(overrides: Partial<QqbotAccount> = {}): QqbotAccount {
|
||||
return Object.assign(new QqbotAccount(), {
|
||||
enabled: true,
|
||||
id: '400',
|
||||
isDeleted: false,
|
||||
selfId: 'bot-a',
|
||||
...overrides,
|
||||
});
|
||||
}
|
||||
|
||||
/** Builds an active account-scoped publishing binding. */
|
||||
function binding(
|
||||
overrides: Partial<QqbotMessagePublishBinding> = {},
|
||||
): QqbotMessagePublishBinding {
|
||||
return Object.assign(new QqbotMessagePublishBinding(), {
|
||||
accountId: '400',
|
||||
enabled: true,
|
||||
id: '500',
|
||||
isDeleted: false,
|
||||
selfId: 'bot-a',
|
||||
subscriptionId: '300',
|
||||
templateId: '600',
|
||||
...overrides,
|
||||
});
|
||||
}
|
||||
|
||||
/** Builds an active source-compatible literal template. */
|
||||
function template(
|
||||
overrides: Partial<QqbotMessageTemplate> = {},
|
||||
): QqbotMessageTemplate {
|
||||
return Object.assign(new QqbotMessageTemplate(), {
|
||||
content: 'endpoint=${{endpoint}}',
|
||||
enabled: true,
|
||||
id: '600',
|
||||
isDeleted: false,
|
||||
sourceKey: SOURCE_KEY,
|
||||
...overrides,
|
||||
});
|
||||
}
|
||||
|
||||
/** Builds one active group/private target under a binding. */
|
||||
function target(
|
||||
overrides: Partial<QqbotMessagePublishTarget> = {},
|
||||
): QqbotMessagePublishTarget {
|
||||
return Object.assign(new QqbotMessagePublishTarget(), {
|
||||
bindingId: '500',
|
||||
enabled: true,
|
||||
id: '700',
|
||||
isDeleted: false,
|
||||
targetId: 'group-1',
|
||||
targetType: 'group',
|
||||
...overrides,
|
||||
});
|
||||
}
|
||||
|
||||
/** Builds a controllable source adapter while retaining a real source registry boundary. */
|
||||
function sourceAdapter(): jest.Mocked<SystemMessageSourceAdapter> {
|
||||
return {
|
||||
definition: {
|
||||
description: 'test',
|
||||
displayName: 'test',
|
||||
sourceKey: SOURCE_KEY,
|
||||
subscriptionFields: [],
|
||||
variables: [],
|
||||
version: 1,
|
||||
},
|
||||
inspectSubscription: jest.fn(),
|
||||
listSubscriptionOptions: jest.fn(),
|
||||
normalizeSubscriptionConfig: jest.fn(),
|
||||
resolveDelivery: jest.fn(async (_input) => {
|
||||
void _input;
|
||||
return {
|
||||
reasonCode: null,
|
||||
status: 'ready' as const,
|
||||
variables: { endpoint: 'pal.example.com:38213' },
|
||||
};
|
||||
}),
|
||||
validateEventPayload: jest.fn(
|
||||
(payload) => payload as Record<string, boolean | null | number | string>,
|
||||
),
|
||||
};
|
||||
}
|
||||
|
||||
/** Clones transaction-visible rows while preserving Date objects and entity prototypes. */
|
||||
function cloneStore(value: Store): Store {
|
||||
const clone = <T>(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),
|
||||
templates: clone(value.templates),
|
||||
};
|
||||
}
|
||||
|
||||
/** Matches the intentionally small TypeORM `where` subset used by the fan-out service. */
|
||||
function matches(
|
||||
row: Record<string, unknown>,
|
||||
where: Record<string, unknown> = {},
|
||||
): boolean {
|
||||
return Object.entries(where).every(([key, value]) => {
|
||||
const actual = row[key];
|
||||
if (
|
||||
actual &&
|
||||
value &&
|
||||
typeof actual === 'object' &&
|
||||
typeof value === 'object' &&
|
||||
'getTime' in actual &&
|
||||
'getTime' in value &&
|
||||
typeof actual.getTime === 'function' &&
|
||||
typeof value.getTime === 'function'
|
||||
)
|
||||
return actual.getTime() === value.getTime();
|
||||
return actual === value;
|
||||
});
|
||||
}
|
||||
|
||||
/** Creates a transaction-faithful in-memory persistence harness rather than fan-out logic. */
|
||||
function setup(seed: Partial<Store> = {}) {
|
||||
let state: Store = {
|
||||
accounts: seed.accounts ?? [account()],
|
||||
bindings: seed.bindings ?? [binding()],
|
||||
deliveries: seed.deliveries ?? [],
|
||||
events: seed.events ?? [event()],
|
||||
subscriptions: seed.subscriptions ?? [subscription()],
|
||||
targets: seed.targets ?? [target()],
|
||||
templates: seed.templates ?? [template()],
|
||||
};
|
||||
const adapter = sourceAdapter();
|
||||
const registry = new SystemMessageSourceRegistry();
|
||||
registry.register(adapter);
|
||||
const query = {
|
||||
brackets: false,
|
||||
lock: '',
|
||||
onLocked: '',
|
||||
order: [] as string[],
|
||||
take: 0,
|
||||
};
|
||||
const transactions: string[] = [];
|
||||
const locked = new Set<string>();
|
||||
let deliverySequence = 1000;
|
||||
let duplicateRace: 'exact' | 'wrong' | null = null;
|
||||
let failSubscription: null | string = null;
|
||||
|
||||
/** Returns a repository facade bound to exactly one authoritative or transaction-local store. */
|
||||
const repository = (entity: unknown, store: Store) => {
|
||||
const rows = (entity === QqbotMessageEvent
|
||||
? store.events
|
||||
: entity === QqbotMessageSubscription
|
||||
? store.subscriptions
|
||||
: entity === QqbotMessagePublishBinding
|
||||
? store.bindings
|
||||
: entity === QqbotAccount
|
||||
? store.accounts
|
||||
: entity === QqbotMessageTemplate
|
||||
? store.templates
|
||||
: entity === QqbotMessagePublishTarget
|
||||
? store.targets
|
||||
: store.deliveries) as unknown as Array<
|
||||
Record<string, unknown>
|
||||
>;
|
||||
return {
|
||||
create: (input: object) =>
|
||||
Object.assign(new (entity as new () => object)(), input),
|
||||
createQueryBuilder: () => {
|
||||
const builder = {
|
||||
addOrderBy: (field: string) => {
|
||||
query.order.push(field);
|
||||
return builder;
|
||||
},
|
||||
getOne: async () =>
|
||||
rows
|
||||
.filter((row) => {
|
||||
const item = row as unknown as QqbotMessageEvent;
|
||||
return (
|
||||
((item.fanoutStatus === 'accepted' ||
|
||||
item.fanoutStatus === 'retry') &&
|
||||
!!item.nextFanoutAt &&
|
||||
item.nextFanoutAt.getTime() <= NOW.getTime()) ||
|
||||
(item.fanoutStatus === 'processing' &&
|
||||
!!item.fanoutLeaseUntil &&
|
||||
item.fanoutLeaseUntil.getTime() <= NOW.getTime())
|
||||
);
|
||||
})
|
||||
.filter(
|
||||
(row) => !locked.has((row as unknown as QqbotMessageEvent).id),
|
||||
)
|
||||
.sort((left, right) => {
|
||||
const a = left as unknown as QqbotMessageEvent;
|
||||
const b = right as unknown as QqbotMessageEvent;
|
||||
return (
|
||||
a.occurredAt.getTime() - b.occurredAt.getTime() ||
|
||||
(BigInt(a.id) < BigInt(b.id)
|
||||
? -1
|
||||
: BigInt(a.id) > BigInt(b.id)
|
||||
? 1
|
||||
: 0)
|
||||
);
|
||||
})[0] ?? null,
|
||||
orderBy: (field: string) => {
|
||||
query.order.push(field);
|
||||
return builder;
|
||||
},
|
||||
setLock: (value: string) => {
|
||||
query.lock = value;
|
||||
return builder;
|
||||
},
|
||||
setOnLocked: (value: string) => {
|
||||
query.onLocked = value;
|
||||
return builder;
|
||||
},
|
||||
take: (value: number) => {
|
||||
query.take = value;
|
||||
return builder;
|
||||
},
|
||||
where: (value: unknown) => {
|
||||
query.brackets = value.constructor.name === 'Brackets';
|
||||
return builder;
|
||||
},
|
||||
};
|
||||
return builder;
|
||||
},
|
||||
find: async ({
|
||||
order,
|
||||
where,
|
||||
}: { order?: { id?: 'ASC' }; where?: Record<string, unknown> } = {}) => {
|
||||
const result = rows.filter((row) => matches(row, where));
|
||||
return order?.id
|
||||
? result.sort((left, right) =>
|
||||
`${(left as { id: string }).id}`.localeCompare(
|
||||
`${(right as { id: string }).id}`,
|
||||
),
|
||||
)
|
||||
: result;
|
||||
},
|
||||
findOne: async ({ where }: { where: Record<string, unknown> }) =>
|
||||
rows.find((row) => matches(row, where)) ?? null,
|
||||
save: async (item: Record<string, unknown>) => {
|
||||
if (
|
||||
entity === QqbotMessageDelivery &&
|
||||
failSubscription === item.bindingId
|
||||
)
|
||||
throw new Error('repository offline');
|
||||
if (entity === QqbotMessageDelivery) {
|
||||
const pair = rows.find((row) =>
|
||||
matches(row, {
|
||||
messageEventId: item.messageEventId,
|
||||
publishTargetId: item.publishTargetId,
|
||||
}),
|
||||
);
|
||||
if (!item.id && duplicateRace) {
|
||||
if (duplicateRace === 'exact')
|
||||
rows.push(
|
||||
Object.assign(new QqbotMessageDelivery(), item, {
|
||||
id: 'race',
|
||||
}) as unknown as Record<string, unknown>,
|
||||
);
|
||||
duplicateRace = null;
|
||||
throw { errno: 1062 };
|
||||
}
|
||||
if (pair && pair !== item) throw { errno: 1062 };
|
||||
if (!item.id) item.id = `${deliverySequence++}`;
|
||||
}
|
||||
const index = rows.findIndex(
|
||||
(row) => (row as { id?: string }).id === item.id,
|
||||
);
|
||||
if (index >= 0) Object.assign(rows[index] as object, item);
|
||||
else rows.push(item as never);
|
||||
return item;
|
||||
},
|
||||
update: async (
|
||||
where: Record<string, unknown>,
|
||||
values: Record<string, unknown>,
|
||||
) => {
|
||||
const row = rows.find((candidate) => matches(candidate, where));
|
||||
if (!row) return { affected: 0 };
|
||||
Object.assign(row as object, values);
|
||||
return { affected: 1 };
|
||||
},
|
||||
};
|
||||
};
|
||||
const dataSource = {
|
||||
getRepository: (entity: unknown) => repository(entity, state),
|
||||
transaction: async (
|
||||
callback: (manager: {
|
||||
getRepository: (entity: unknown) => ReturnType<typeof repository>;
|
||||
}) => Promise<unknown>,
|
||||
) => {
|
||||
const draft = cloneStore(state);
|
||||
transactions.push('begin');
|
||||
try {
|
||||
const result = await callback({
|
||||
getRepository: (entity) => repository(entity, draft),
|
||||
});
|
||||
state = draft;
|
||||
transactions.push('commit');
|
||||
return result;
|
||||
} catch (error) {
|
||||
transactions.push('rollback');
|
||||
throw error;
|
||||
}
|
||||
},
|
||||
};
|
||||
const service = new SystemMessageFanoutService(
|
||||
dataSource as never,
|
||||
registry,
|
||||
new SystemMessageTemplateRendererService(),
|
||||
);
|
||||
return {
|
||||
adapter,
|
||||
events: () => state.events,
|
||||
deliveries: () => state.deliveries,
|
||||
failSubscription: (id: null | string) => {
|
||||
failSubscription = id;
|
||||
},
|
||||
lock: (id: string) => locked.add(id),
|
||||
query,
|
||||
setDuplicateRace: (value: 'exact' | 'wrong' | null) => {
|
||||
duplicateRace = value;
|
||||
},
|
||||
service,
|
||||
transactions,
|
||||
};
|
||||
}
|
||||
|
||||
describe('SystemMessageFanoutService', () => {
|
||||
it('creates exact multi-account target cardinality and freezes strict snapshots', async () => {
|
||||
const fixture = setup({
|
||||
accounts: [account(), account({ id: '401', selfId: 'bot-b' })],
|
||||
bindings: [
|
||||
binding(),
|
||||
binding({ accountId: '401', id: '501', selfId: 'bot-b' }),
|
||||
],
|
||||
targets: [
|
||||
target(),
|
||||
target({ id: '701', targetId: 'private-1', targetType: 'private' }),
|
||||
target({ bindingId: '501', id: '702', targetId: 'group-2' }),
|
||||
],
|
||||
});
|
||||
|
||||
await expect(fixture.service.runOnce(NOW)).resolves.toBe(1);
|
||||
expect(fixture.deliveries()).toHaveLength(3);
|
||||
expect(
|
||||
new Set(fixture.deliveries().map((item) => item.publishTargetId)).size,
|
||||
).toBe(3);
|
||||
expect(fixture.deliveries()[0]).toMatchObject({
|
||||
attemptCount: 0,
|
||||
bindingId: '500',
|
||||
renderedMessage: 'endpoint=pal.example.com:38213',
|
||||
selfId: 'bot-a',
|
||||
status: 'pending',
|
||||
subscriptionId: '300',
|
||||
templateContent: 'endpoint=${{endpoint}}',
|
||||
templateId: '600',
|
||||
variableSnapshot: { endpoint: 'pal.example.com:38213' },
|
||||
});
|
||||
expect(fixture.deliveries()[0].expiresAt.getTime()).toBe(
|
||||
NOW.getTime() + SYSTEM_MESSAGE_RETRY_WINDOW_MS,
|
||||
);
|
||||
});
|
||||
|
||||
it('schedules ready and waiting-DDNS snapshots without sending OneBot', async () => {
|
||||
const fixture = setup();
|
||||
fixture.adapter.resolveDelivery.mockResolvedValueOnce({
|
||||
reasonCode: 'ddns_not_synced',
|
||||
status: 'waiting_ddns',
|
||||
variables: { endpoint: 'wait.example:1' },
|
||||
});
|
||||
|
||||
await fixture.service.runOnce(NOW);
|
||||
expect(fixture.deliveries()[0]).toMatchObject({
|
||||
renderedMessage: 'endpoint=wait.example:1',
|
||||
status: 'waiting_ddns',
|
||||
});
|
||||
expect(fixture.deliveries()[0].nextAttemptAt.getTime()).toBe(
|
||||
NOW.getTime() + SYSTEM_MESSAGE_DDNS_RECHECK_MS,
|
||||
);
|
||||
expect(fixture.events()[0].fanoutStatus).toBe('completed');
|
||||
});
|
||||
|
||||
it('requires an active own string JSON resource match before resolving a subscription', async () => {
|
||||
const inherited = Object.create({ portForwardId: RESOURCE_KEY }) as Record<
|
||||
string,
|
||||
unknown
|
||||
>;
|
||||
const fixture = setup({
|
||||
subscriptions: [
|
||||
subscription(),
|
||||
subscription({
|
||||
id: '301',
|
||||
sourceConfig: { portForwardId: Number(RESOURCE_KEY) },
|
||||
}),
|
||||
subscription({ id: '302', sourceConfig: inherited }),
|
||||
subscription({ enabled: false, id: '303' }),
|
||||
subscription({ id: '304', isDeleted: true }),
|
||||
],
|
||||
});
|
||||
|
||||
await fixture.service.runOnce(NOW);
|
||||
expect(fixture.adapter.resolveDelivery).toHaveBeenCalledTimes(1);
|
||||
});
|
||||
|
||||
it('isolates disabled/deleted or mismatched account, binding, template and target rows', async () => {
|
||||
const fixture = setup({
|
||||
accounts: [
|
||||
account(),
|
||||
account({ enabled: false, id: '401', selfId: 'bad' }),
|
||||
],
|
||||
bindings: [
|
||||
binding(),
|
||||
binding({ accountId: '401', id: '501', selfId: 'bad' }),
|
||||
binding({ enabled: false, id: '502' }),
|
||||
binding({ id: '503', selfId: 'wrong' }),
|
||||
],
|
||||
targets: [
|
||||
target(),
|
||||
target({ bindingId: '501', id: '701' }),
|
||||
target({ bindingId: '500', enabled: false, id: '702' }),
|
||||
target({ bindingId: '500', id: '703', isDeleted: true }),
|
||||
],
|
||||
});
|
||||
|
||||
await fixture.service.runOnce(NOW);
|
||||
expect(fixture.deliveries().map((item) => item.publishTargetId)).toEqual([
|
||||
'700',
|
||||
]);
|
||||
});
|
||||
|
||||
it.each(['cancelled', 'superseded'] as const)(
|
||||
'completes %s readiness without current delivery or render',
|
||||
async (status) => {
|
||||
const fixture = setup();
|
||||
fixture.adapter.resolveDelivery.mockResolvedValueOnce({
|
||||
reasonCode: status,
|
||||
status,
|
||||
});
|
||||
|
||||
await fixture.service.runOnce(NOW);
|
||||
expect(fixture.deliveries()).toHaveLength(0);
|
||||
expect(fixture.events()[0].fanoutStatus).toBe('completed');
|
||||
},
|
||||
);
|
||||
|
||||
it('supersedes only strictly earlier unfinished deliveries for the same subscription', async () => {
|
||||
const older = event({
|
||||
id: '100',
|
||||
occurredAt: new KtDateTime(NOW.getTime() - 1),
|
||||
});
|
||||
const newer = event({
|
||||
id: '201',
|
||||
occurredAt: new KtDateTime(NOW.getTime() + 1),
|
||||
nextFanoutAt: new KtDateTime(NOW.getTime() + 1),
|
||||
});
|
||||
const makeDelivery = (
|
||||
id: string,
|
||||
messageEventId: string,
|
||||
status: QqbotMessageDelivery['status'],
|
||||
) =>
|
||||
Object.assign(new QqbotMessageDelivery(), {
|
||||
id,
|
||||
messageEventId,
|
||||
publishTargetId: id,
|
||||
status,
|
||||
subscriptionId: '300',
|
||||
});
|
||||
const fixture = setup({
|
||||
events: [older, event(), newer],
|
||||
deliveries: [
|
||||
makeDelivery('1', '100', 'pending'),
|
||||
makeDelivery('2', '100', 'retry'),
|
||||
makeDelivery('3', '100', 'processing'),
|
||||
makeDelivery('4', '200', 'pending'),
|
||||
makeDelivery('5', '201', 'pending'),
|
||||
],
|
||||
});
|
||||
|
||||
await fixture.service.runOnce(NOW);
|
||||
expect(
|
||||
fixture
|
||||
.deliveries()
|
||||
.filter((item) => item.messageEventId === '100')
|
||||
.map((item) => item.status),
|
||||
).toEqual(['superseded', 'superseded', 'processing', 'superseded']);
|
||||
expect(
|
||||
fixture.deliveries().find((item) => item.messageEventId === '201')
|
||||
?.status,
|
||||
).toBe('pending');
|
||||
expect(
|
||||
fixture.deliveries().find((item) => item.messageEventId === '200')
|
||||
?.status,
|
||||
).toBe('pending');
|
||||
});
|
||||
|
||||
it('accepts exact event-target replay and verifies only an exact duplicate-key race', async () => {
|
||||
const existing = Object.assign(new QqbotMessageDelivery(), {
|
||||
id: 'old',
|
||||
messageEventId: '200',
|
||||
publishTargetId: '700',
|
||||
status: 'pending',
|
||||
});
|
||||
const replay = setup({ deliveries: [existing] });
|
||||
await replay.service.runOnce(NOW);
|
||||
expect(replay.deliveries()).toHaveLength(1);
|
||||
|
||||
const race = setup();
|
||||
race.setDuplicateRace('exact');
|
||||
await race.service.runOnce(NOW);
|
||||
expect(race.deliveries()).toHaveLength(1);
|
||||
|
||||
const wrong = setup();
|
||||
wrong.setDuplicateRace('wrong');
|
||||
await wrong.service.runOnce(NOW);
|
||||
expect(wrong.events()[0]).toMatchObject({
|
||||
fanoutStatus: 'retry',
|
||||
lastErrorCode: 'fanout_transient_error',
|
||||
});
|
||||
});
|
||||
|
||||
it('records bounded SQL claim contract including top-level Brackets, locking, ordering and limit', async () => {
|
||||
const fixture = setup({ events: [] });
|
||||
await fixture.service.runOnce(NOW);
|
||||
expect(fixture.query).toMatchObject({
|
||||
brackets: true,
|
||||
lock: 'pessimistic_write',
|
||||
onLocked: 'skip_locked',
|
||||
take: 1,
|
||||
});
|
||||
expect(fixture.query.order).toEqual(['event.occurredAt', 'event.id']);
|
||||
});
|
||||
|
||||
it('skips a locked oldest due event and reclaims expired processing with an exact new lease', async () => {
|
||||
const oldest = event({
|
||||
id: '100',
|
||||
occurredAt: new KtDateTime(NOW.getTime() - 2),
|
||||
});
|
||||
const expired = event({
|
||||
fanoutAttemptCount: 4,
|
||||
fanoutLeaseUntil: new KtDateTime(NOW.getTime() - 1),
|
||||
fanoutStatus: 'processing',
|
||||
id: '101',
|
||||
nextFanoutAt: null,
|
||||
occurredAt: new KtDateTime(NOW.getTime() - 1),
|
||||
});
|
||||
const fixture = setup({ events: [oldest, expired] });
|
||||
fixture.lock('100');
|
||||
|
||||
await fixture.service.runOnce(NOW);
|
||||
expect(fixture.events().find((item) => item.id === '101')).toMatchObject({
|
||||
fanoutAttemptCount: 5,
|
||||
fanoutStatus: 'completed',
|
||||
});
|
||||
expect(fixture.events().find((item) => item.id === '100')).toMatchObject({
|
||||
fanoutStatus: 'accepted',
|
||||
});
|
||||
});
|
||||
|
||||
it('uses a lease-and-attempt ownership fence so stale completion cannot clobber a new owner', async () => {
|
||||
const fixture = setup({ events: [] });
|
||||
const staleEvent = event({
|
||||
fanoutAttemptCount: 1,
|
||||
fanoutLeaseUntil: new KtDateTime(NOW.getTime() + SYSTEM_MESSAGE_LEASE_MS),
|
||||
fanoutStatus: 'processing',
|
||||
});
|
||||
const current = event({
|
||||
fanoutAttemptCount: 2,
|
||||
fanoutLeaseUntil: new KtDateTime(
|
||||
NOW.getTime() + SYSTEM_MESSAGE_LEASE_MS + 1,
|
||||
),
|
||||
fanoutStatus: 'processing',
|
||||
});
|
||||
const finish = (
|
||||
fixture.service as unknown as {
|
||||
finish: (
|
||||
token: object,
|
||||
status: 'completed',
|
||||
code: null,
|
||||
message: null,
|
||||
) => Promise<void>;
|
||||
}
|
||||
).finish;
|
||||
(fixture as unknown as { events: () => QqbotMessageEvent[] })
|
||||
.events()
|
||||
.push(current);
|
||||
|
||||
await finish.call(
|
||||
fixture.service,
|
||||
{
|
||||
attempt: 1,
|
||||
event: staleEvent,
|
||||
leaseUntil: staleEvent.fanoutLeaseUntil,
|
||||
},
|
||||
'completed',
|
||||
null,
|
||||
null,
|
||||
);
|
||||
expect(fixture.events()[0].fanoutStatus).toBe('processing');
|
||||
});
|
||||
|
||||
it('rolls back one failing subscription, retains valid work, then retries and recovers idempotently', async () => {
|
||||
const fixture = setup({
|
||||
subscriptions: [subscription(), subscription({ id: '301' })],
|
||||
bindings: [binding(), binding({ id: '501', subscriptionId: '301' })],
|
||||
targets: [target(), target({ bindingId: '501', id: '701' })],
|
||||
});
|
||||
fixture.failSubscription('501');
|
||||
await fixture.service.runOnce(NOW);
|
||||
expect(fixture.deliveries().map((item) => item.publishTargetId)).toEqual([
|
||||
'700',
|
||||
]);
|
||||
expect(fixture.events()[0]).toMatchObject({
|
||||
fanoutStatus: 'retry',
|
||||
nextFanoutAt: new KtDateTime(
|
||||
NOW.getTime() + SYSTEM_MESSAGE_RETRY_BASE_MS,
|
||||
),
|
||||
});
|
||||
fixture.failSubscription(null);
|
||||
fixture.events()[0].nextFanoutAt = new KtDateTime(NOW);
|
||||
await fixture.service.runOnce(NOW);
|
||||
expect(
|
||||
new Set(fixture.deliveries().map((item) => item.publishTargetId)),
|
||||
).toEqual(new Set(['700', '701']));
|
||||
expect(fixture.events()[0].fanoutStatus).toBe('completed');
|
||||
expect(fixture.transactions).toContain('rollback');
|
||||
});
|
||||
|
||||
it('fails at the occurrence deadline and permanently rejects malformed source facts', async () => {
|
||||
const deadline = setup({
|
||||
events: [
|
||||
event({
|
||||
occurredAt: new KtDateTime(
|
||||
NOW.getTime() - SYSTEM_MESSAGE_RETRY_WINDOW_MS,
|
||||
),
|
||||
}),
|
||||
],
|
||||
});
|
||||
await deadline.service.runOnce(NOW);
|
||||
expect(deadline.events()[0]).toMatchObject({
|
||||
fanoutStatus: 'failed',
|
||||
lastErrorCode: 'fanout_expired',
|
||||
});
|
||||
expect(deadline.deliveries()).toHaveLength(0);
|
||||
|
||||
const malformed = setup({
|
||||
events: [event({ payload: { portForwardId: 'other' } })],
|
||||
});
|
||||
await malformed.service.runOnce(NOW);
|
||||
expect(malformed.events()[0]).toMatchObject({
|
||||
fanoutStatus: 'failed',
|
||||
lastErrorCode: 'event_resource_mismatch',
|
||||
});
|
||||
});
|
||||
|
||||
it('bounds a pass to fifty oldest claims and leaves not-due rows untouched', async () => {
|
||||
const events = Array.from(
|
||||
{ length: SYSTEM_MESSAGE_BATCH_SIZE + 1 },
|
||||
(_, index) =>
|
||||
event({
|
||||
id: `${1000 + index}`,
|
||||
occurredAt: new KtDateTime(
|
||||
NOW.getTime() - SYSTEM_MESSAGE_BATCH_SIZE + index,
|
||||
),
|
||||
}),
|
||||
);
|
||||
const fixture = setup({ events });
|
||||
await expect(fixture.service.runOnce(NOW)).resolves.toBe(
|
||||
SYSTEM_MESSAGE_BATCH_SIZE,
|
||||
);
|
||||
expect(
|
||||
fixture.events().filter((item) => item.fanoutStatus === 'completed'),
|
||||
).toHaveLength(SYSTEM_MESSAGE_BATCH_SIZE);
|
||||
expect(fixture.events().at(-1)?.fanoutStatus).toBe('accepted');
|
||||
});
|
||||
|
||||
it('marks source contract errors as permanent failures rather than transient adapter retries', async () => {
|
||||
const fixture = setup();
|
||||
fixture.adapter.validateEventPayload.mockImplementation(() => {
|
||||
throw new SystemMessageContractError('payload_invalid');
|
||||
});
|
||||
await fixture.service.runOnce(NOW);
|
||||
expect(fixture.events()[0]).toMatchObject({
|
||||
fanoutStatus: 'failed',
|
||||
lastErrorCode: 'payload_invalid',
|
||||
});
|
||||
});
|
||||
|
||||
it('completes a matched event with no legal targets without resolving a default account', async () => {
|
||||
const fixture = setup({ targets: [] });
|
||||
await fixture.service.runOnce(NOW);
|
||||
expect(fixture.deliveries()).toHaveLength(0);
|
||||
expect(fixture.events()[0].fanoutStatus).toBe('completed');
|
||||
});
|
||||
|
||||
it('does not resolve when no subscription matches the frozen source/resource', async () => {
|
||||
const fixture = setup({
|
||||
subscriptions: [
|
||||
subscription({ sourceConfig: { portForwardId: 'other' } }),
|
||||
],
|
||||
});
|
||||
await fixture.service.runOnce(NOW);
|
||||
expect(fixture.adapter.resolveDelivery).not.toHaveBeenCalled();
|
||||
expect(fixture.events()[0].fanoutStatus).toBe('completed');
|
||||
});
|
||||
|
||||
it('leaves future retries and live processing leases unclaimed', async () => {
|
||||
const fixture = setup({
|
||||
events: [
|
||||
event({
|
||||
fanoutStatus: 'retry',
|
||||
nextFanoutAt: new KtDateTime(NOW.getTime() + 1),
|
||||
}),
|
||||
event({
|
||||
fanoutLeaseUntil: new KtDateTime(NOW.getTime() + 1),
|
||||
fanoutStatus: 'processing',
|
||||
id: '201',
|
||||
nextFanoutAt: null,
|
||||
}),
|
||||
],
|
||||
});
|
||||
await expect(fixture.service.runOnce(NOW)).resolves.toBe(0);
|
||||
expect(fixture.events().map((item) => item.fanoutStatus)).toEqual([
|
||||
'retry',
|
||||
'processing',
|
||||
]);
|
||||
});
|
||||
|
||||
it('keeps a runtime-offline but enabled strict account eligible at fan-out', async () => {
|
||||
const fixture = setup({
|
||||
accounts: [
|
||||
account({ connectStatus: 'offline', oneBotStatus: 'offline' }),
|
||||
],
|
||||
});
|
||||
await fixture.service.runOnce(NOW);
|
||||
expect(fixture.deliveries()).toHaveLength(1);
|
||||
});
|
||||
|
||||
it('isolates an invalid template render while allowing another legal binding', async () => {
|
||||
const fixture = setup({
|
||||
bindings: [binding(), binding({ id: '501', templateId: '601' })],
|
||||
templates: [
|
||||
template({ content: '${{unknown}}' }),
|
||||
template({ id: '601' }),
|
||||
],
|
||||
targets: [target(), target({ bindingId: '501', id: '701' })],
|
||||
});
|
||||
await fixture.service.runOnce(NOW);
|
||||
expect(fixture.deliveries().map((item) => item.publishTargetId)).toEqual([
|
||||
'701',
|
||||
]);
|
||||
});
|
||||
|
||||
it('does not supersede current-event rows during an idempotent replay', async () => {
|
||||
const fixture = setup({
|
||||
deliveries: [
|
||||
Object.assign(new QqbotMessageDelivery(), {
|
||||
id: 'old',
|
||||
messageEventId: '200',
|
||||
publishTargetId: '700',
|
||||
status: 'retry',
|
||||
subscriptionId: '300',
|
||||
}),
|
||||
],
|
||||
});
|
||||
await fixture.service.runOnce(NOW);
|
||||
expect(fixture.deliveries()[0].status).toBe('retry');
|
||||
});
|
||||
|
||||
it('uses numeric BIGINT ordering when occurrence times are equal', async () => {
|
||||
const older = event({ id: '9', occurredAt: new KtDateTime(NOW) });
|
||||
const current = event({ id: '10', occurredAt: new KtDateTime(NOW) });
|
||||
const fixture = setup({
|
||||
events: [current, older],
|
||||
deliveries: [
|
||||
Object.assign(new QqbotMessageDelivery(), {
|
||||
id: 'old',
|
||||
messageEventId: '9',
|
||||
publishTargetId: 'old-target',
|
||||
status: 'pending',
|
||||
subscriptionId: '300',
|
||||
}),
|
||||
],
|
||||
});
|
||||
await fixture.service.runOnce(NOW);
|
||||
expect(fixture.deliveries()[0].status).toBe('superseded');
|
||||
});
|
||||
|
||||
it('marks a transient event failure retryable with the one-based ten-second delay', async () => {
|
||||
const fixture = setup();
|
||||
fixture.adapter.resolveDelivery.mockRejectedValueOnce(
|
||||
new Error('adapter unavailable'),
|
||||
);
|
||||
await fixture.service.runOnce(NOW);
|
||||
expect(fixture.events()[0]).toMatchObject({
|
||||
fanoutStatus: 'retry',
|
||||
lastErrorCode: 'fanout_transient_error',
|
||||
});
|
||||
expect(fixture.events()[0].nextFanoutAt.getTime()).toBe(
|
||||
NOW.getTime() + SYSTEM_MESSAGE_RETRY_BASE_MS,
|
||||
);
|
||||
});
|
||||
|
||||
it('fails immediately when the next transient retry would fall outside the hard deadline', async () => {
|
||||
const fixture = setup({
|
||||
events: [
|
||||
event({
|
||||
occurredAt: new KtDateTime(
|
||||
NOW.getTime() - SYSTEM_MESSAGE_RETRY_WINDOW_MS + 1,
|
||||
),
|
||||
}),
|
||||
],
|
||||
});
|
||||
fixture.adapter.resolveDelivery.mockRejectedValueOnce(
|
||||
new Error('adapter unavailable'),
|
||||
);
|
||||
await fixture.service.runOnce(NOW);
|
||||
expect(fixture.events()[0]).toMatchObject({
|
||||
fanoutStatus: 'failed',
|
||||
lastErrorCode: 'fanout_expired',
|
||||
nextFanoutAt: null,
|
||||
});
|
||||
});
|
||||
|
||||
it('fails an unknown source permanently without mutating subscriptions or deliveries', async () => {
|
||||
const fixture = setup({ events: [event({ sourceKey: 'unknown.source' })] });
|
||||
await fixture.service.runOnce(NOW);
|
||||
expect(fixture.events()[0]).toMatchObject({
|
||||
fanoutStatus: 'failed',
|
||||
lastErrorCode: 'unknown_message_source',
|
||||
});
|
||||
expect(fixture.deliveries()).toHaveLength(0);
|
||||
});
|
||||
|
||||
it('calls the adapter exactly once per matched subscription rather than once per target', async () => {
|
||||
const fixture = setup({ targets: [target(), target({ id: '701' })] });
|
||||
await fixture.service.runOnce(NOW);
|
||||
expect(fixture.adapter.resolveDelivery).toHaveBeenCalledTimes(1);
|
||||
});
|
||||
|
||||
it('keeps processing and terminal historical deliveries outside supersession', async () => {
|
||||
const older = event({
|
||||
id: '100',
|
||||
occurredAt: new KtDateTime(NOW.getTime() - 1),
|
||||
});
|
||||
const statuses: QqbotMessageDelivery['status'][] = [
|
||||
'processing',
|
||||
'success',
|
||||
'failed',
|
||||
'superseded',
|
||||
'cancelled',
|
||||
];
|
||||
const fixture = setup({
|
||||
events: [older, event()],
|
||||
deliveries: statuses.map((status, index) =>
|
||||
Object.assign(new QqbotMessageDelivery(), {
|
||||
id: `${index}`,
|
||||
messageEventId: '100',
|
||||
publishTargetId: `${index}`,
|
||||
status,
|
||||
subscriptionId: '300',
|
||||
}),
|
||||
),
|
||||
});
|
||||
await fixture.service.runOnce(NOW);
|
||||
expect(
|
||||
fixture
|
||||
.deliveries()
|
||||
.filter((item) => ['0', '1', '2', '3', '4'].includes(item.id))
|
||||
.map((item) => item.status),
|
||||
).toEqual(statuses);
|
||||
});
|
||||
|
||||
it('preserves snapshots after later in-memory configuration edits', async () => {
|
||||
const fixture = setup();
|
||||
await fixture.service.runOnce(NOW);
|
||||
const frozen = fixture.deliveries()[0];
|
||||
frozen.templateContent = 'endpoint=${{endpoint}}';
|
||||
expect(frozen).toMatchObject({
|
||||
renderedMessage: 'endpoint=pal.example.com:38213',
|
||||
selfId: 'bot-a',
|
||||
targetId: 'group-1',
|
||||
});
|
||||
});
|
||||
});
|
||||
Loading…
Reference in New Issue
Block a user