feat: 实现系统消息订阅管理
This commit is contained in:
parent
1a19730c18
commit
986615ac32
@ -0,0 +1,377 @@
|
||||
import { createHash } from 'node:crypto';
|
||||
import { HttpStatus, Injectable } from '@nestjs/common';
|
||||
import { InjectRepository } from '@nestjs/typeorm';
|
||||
import { throwVbenError } from '@/common';
|
||||
import {
|
||||
Like,
|
||||
Repository,
|
||||
type EntityManager,
|
||||
type FindOptionsWhere,
|
||||
} from 'typeorm';
|
||||
import {
|
||||
SystemMessageContractError,
|
||||
type MessageSubscriptionInput,
|
||||
type MessageSubscriptionListQuery,
|
||||
type MessageSubscriptionView,
|
||||
} from '../../contract/message-push/qqbot-message-push.types';
|
||||
import { QqbotMessageSubscription } from '../../infrastructure/persistence/message-push/qqbot-message-subscription.entity';
|
||||
import { SystemMessageSourceRegistry } from './system-message-source.registry';
|
||||
|
||||
const DEFAULT_PAGE_NO = 1;
|
||||
const DEFAULT_PAGE_SIZE = 10;
|
||||
|
||||
type NormalizedSubscriptionInput = {
|
||||
activeKey: string;
|
||||
enabled: boolean;
|
||||
name: string;
|
||||
remark: null | string;
|
||||
sourceConfig: Record<string, string>;
|
||||
sourceConfigDigest: string;
|
||||
sourceKey: string;
|
||||
};
|
||||
|
||||
/**
|
||||
* Manages source-normalized global system-message subscriptions and their binding lock gate.
|
||||
*
|
||||
* Database errors other than MySQL duplicate-key conflicts deliberately propagate so callers
|
||||
* cannot treat incomplete persistence as a valid subscription lifecycle transition.
|
||||
*/
|
||||
@Injectable()
|
||||
export class QqbotMessageSubscriptionService {
|
||||
/**
|
||||
* Initializes subscription persistence and the live process-local source registry.
|
||||
* @param subscriptionRepository - Persistence root used to open lifecycle transactions.
|
||||
* @param sourceRegistry - Current registered source definitions and adapters.
|
||||
*/
|
||||
constructor(
|
||||
@InjectRepository(QqbotMessageSubscription)
|
||||
private readonly subscriptionRepository: Repository<QqbotMessageSubscription>,
|
||||
private readonly sourceRegistry: SystemMessageSourceRegistry,
|
||||
) {}
|
||||
|
||||
/**
|
||||
* Pages non-deleted subscriptions with current source validity rather than stale snapshots.
|
||||
* @param query - Optional name, source key, enabled state, and pagination filters.
|
||||
* @returns Deterministically ordered detached subscription views and their total count.
|
||||
*/
|
||||
async page(query: MessageSubscriptionListQuery): Promise<{
|
||||
items: MessageSubscriptionView[];
|
||||
total: number;
|
||||
}> {
|
||||
const pageNo = Math.max(
|
||||
DEFAULT_PAGE_NO,
|
||||
Math.floor(query.pageNo ?? DEFAULT_PAGE_NO),
|
||||
);
|
||||
const pageSize = Math.max(
|
||||
1,
|
||||
Math.floor(query.pageSize ?? DEFAULT_PAGE_SIZE),
|
||||
);
|
||||
const where: FindOptionsWhere<QqbotMessageSubscription> = {
|
||||
isDeleted: false,
|
||||
};
|
||||
if (query.name) where.name = Like(`%${query.name}%`);
|
||||
if (query.sourceKey) where.sourceKey = query.sourceKey;
|
||||
if (query.enabled !== undefined) where.enabled = query.enabled;
|
||||
|
||||
const [subscriptions, total] =
|
||||
await this.subscriptionRepository.findAndCount({
|
||||
order: { createTime: 'DESC', id: 'DESC' },
|
||||
skip: (pageNo - 1) * pageSize,
|
||||
take: pageSize,
|
||||
where,
|
||||
});
|
||||
return {
|
||||
items: await Promise.all(subscriptions.map((item) => this.toView(item))),
|
||||
total,
|
||||
};
|
||||
}
|
||||
|
||||
/**
|
||||
* Creates a source-normalized subscription or revives its most recently updated history row.
|
||||
* @param input - Untrusted metadata and source configuration submitted by management UI.
|
||||
* @returns The new or revived subscription view.
|
||||
* @throws {HttpException} HTTP 409 when another active row owns the same natural key.
|
||||
*/
|
||||
async create(
|
||||
input: MessageSubscriptionInput,
|
||||
): Promise<MessageSubscriptionView> {
|
||||
const normalized = await this.normalizeInput(input);
|
||||
try {
|
||||
const saved = await this.subscriptionRepository.manager.transaction(
|
||||
/** Serializes active-key ownership and deleted-row revival in one transaction. */
|
||||
async (manager) => {
|
||||
const repository = manager.getRepository(QqbotMessageSubscription);
|
||||
const active = await repository.findOne({
|
||||
lock: { mode: 'pessimistic_write' },
|
||||
where: { activeKey: normalized.activeKey, isDeleted: false },
|
||||
});
|
||||
if (active) this.throwNaturalKeyConflict();
|
||||
|
||||
const historical = await repository.findOne({
|
||||
lock: { mode: 'pessimistic_write' },
|
||||
order: { updateTime: 'DESC', id: 'DESC' },
|
||||
where: {
|
||||
isDeleted: true,
|
||||
sourceConfigDigest: normalized.sourceConfigDigest,
|
||||
sourceKey: normalized.sourceKey,
|
||||
},
|
||||
});
|
||||
if (historical) {
|
||||
Object.assign(historical, normalized, { isDeleted: false });
|
||||
return repository.save(historical);
|
||||
}
|
||||
return repository.save(
|
||||
repository.create({ ...normalized, isDeleted: false }),
|
||||
);
|
||||
},
|
||||
);
|
||||
return this.toView(saved);
|
||||
} catch (error) {
|
||||
if (this.isDuplicateKeyError(error)) this.throwNaturalKeyConflict();
|
||||
throw error;
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Replaces one active subscription's metadata and canonical source configuration.
|
||||
* @param id - String Snowflake identity of the subscription being changed.
|
||||
* @param input - Complete replacement metadata and untrusted source configuration.
|
||||
* @returns The updated view for the same subscription ID.
|
||||
* @throws {HttpException} HTTP 409 when another active subscription owns the new natural key.
|
||||
*/
|
||||
async update(
|
||||
id: string,
|
||||
input: MessageSubscriptionInput,
|
||||
): Promise<MessageSubscriptionView> {
|
||||
const normalized = await this.normalizeInput(input);
|
||||
const saved = await this.subscriptionRepository.manager.transaction(
|
||||
/** Holds both target and prospective active-key rows through the update save. */
|
||||
async (manager) => {
|
||||
const repository = manager.getRepository(QqbotMessageSubscription);
|
||||
const current = await this.findActiveForWrite(repository, id);
|
||||
const conflict = await repository.findOne({
|
||||
lock: { mode: 'pessimistic_write' },
|
||||
where: { activeKey: normalized.activeKey, isDeleted: false },
|
||||
});
|
||||
if (conflict && conflict.id !== current.id) {
|
||||
this.throwNaturalKeyConflict();
|
||||
}
|
||||
Object.assign(current, normalized);
|
||||
return repository.save(current);
|
||||
},
|
||||
);
|
||||
return this.toView(saved);
|
||||
}
|
||||
|
||||
/**
|
||||
* Changes an active subscription's enabled state after a live source validation when enabling.
|
||||
* @param id - String Snowflake identity of the active subscription.
|
||||
* @param enabled - Requested future enabled state.
|
||||
* @returns The updated detached subscription view.
|
||||
* @throws {SystemMessageContractError} With the current source reason when enabling is unsafe.
|
||||
*/
|
||||
async setEnabled(
|
||||
id: string,
|
||||
enabled: boolean,
|
||||
): Promise<MessageSubscriptionView> {
|
||||
const saved = await this.subscriptionRepository.manager.transaction(
|
||||
/** Holds the subscription row while current source validity is checked and persisted. */
|
||||
async (manager) => {
|
||||
const repository = manager.getRepository(QqbotMessageSubscription);
|
||||
const current = await this.findActiveForWrite(repository, id);
|
||||
if (enabled) {
|
||||
const inspection = await this.sourceRegistry
|
||||
.get(current.sourceKey)
|
||||
.inspectSubscription(current.sourceConfig);
|
||||
if (!inspection.valid) {
|
||||
throw new SystemMessageContractError(
|
||||
inspection.invalidReasonCode || 'invalid_source_config',
|
||||
);
|
||||
}
|
||||
}
|
||||
current.enabled = enabled;
|
||||
return repository.save(current);
|
||||
},
|
||||
);
|
||||
return this.toView(saved);
|
||||
}
|
||||
|
||||
/**
|
||||
* Soft-deletes one active subscription and releases its active natural key atomically.
|
||||
* @param id - String Snowflake identity of the subscription to remove.
|
||||
* @returns `true` once disabled/deleted state and null active key were persisted.
|
||||
*/
|
||||
async remove(id: string): Promise<boolean> {
|
||||
return this.subscriptionRepository.manager.transaction(
|
||||
/** Holds the active row until its deletion-safe fields are stored together. */
|
||||
async (manager) => {
|
||||
const repository = manager.getRepository(QqbotMessageSubscription);
|
||||
const current = await this.findActiveForWrite(repository, id);
|
||||
current.activeKey = null;
|
||||
current.enabled = false;
|
||||
current.isDeleted = true;
|
||||
await repository.save(current);
|
||||
return true;
|
||||
},
|
||||
);
|
||||
}
|
||||
|
||||
/**
|
||||
* Enforces the subscription half of the binding lock order before any live binding save.
|
||||
*
|
||||
* Callers must pass their current binding-write transaction manager, invoke this before
|
||||
* saving every new, updated, or revived non-deleted binding, and keep that transaction open
|
||||
* through the binding save and commit. Disabled bindings require this same lock because they
|
||||
* still reference the subscription and must not race a concurrent removal.
|
||||
* @param manager - Current binding transaction manager; never substitute a global manager.
|
||||
* @param subscriptionId - Candidate subscription's string Snowflake identity.
|
||||
* @param bindingEnabled - Whether the binding will be enabled for message delivery.
|
||||
* @returns The locked active subscription available to the requested binding state.
|
||||
* @throws {SystemMessageContractError} For deleted/missing subscriptions, disabled enabled-bindings, or invalid sources.
|
||||
*/
|
||||
async requireAvailableForBinding(
|
||||
manager: EntityManager,
|
||||
subscriptionId: string,
|
||||
bindingEnabled: boolean,
|
||||
): Promise<QqbotMessageSubscription> {
|
||||
const subscription = await manager
|
||||
.getRepository(QqbotMessageSubscription)
|
||||
.findOne({
|
||||
lock: { mode: 'pessimistic_write' },
|
||||
where: { id: subscriptionId, isDeleted: false },
|
||||
});
|
||||
if (!subscription) {
|
||||
throw new SystemMessageContractError('invalid_source_config');
|
||||
}
|
||||
if (!bindingEnabled) return subscription;
|
||||
if (!subscription.enabled) {
|
||||
throw new SystemMessageContractError('subscription_disabled');
|
||||
}
|
||||
const inspection = await this.sourceRegistry
|
||||
.get(subscription.sourceKey)
|
||||
.inspectSubscription(subscription.sourceConfig);
|
||||
if (!inspection.valid) {
|
||||
throw new SystemMessageContractError(
|
||||
inspection.invalidReasonCode || 'invalid_source_config',
|
||||
);
|
||||
}
|
||||
return subscription;
|
||||
}
|
||||
|
||||
/**
|
||||
* Normalizes user input through the exact source adapter before any transaction begins.
|
||||
* @param input - Untrusted management payload with a requested source key.
|
||||
* @returns Detached canonical persistence fields and their stable SHA-256 natural key.
|
||||
*/
|
||||
private async normalizeInput(
|
||||
input: MessageSubscriptionInput,
|
||||
): Promise<NormalizedSubscriptionInput> {
|
||||
const adapter = this.sourceRegistry.get(input.sourceKey);
|
||||
const normalized = await adapter.normalizeSubscriptionConfig(
|
||||
input.sourceConfig,
|
||||
);
|
||||
const sourceConfig = this.sortConfig(normalized.canonicalConfig);
|
||||
const sourceConfigDigest = createHash('sha256')
|
||||
.update(JSON.stringify(sourceConfig))
|
||||
.digest('hex');
|
||||
return {
|
||||
activeKey: `${input.sourceKey}:${sourceConfigDigest}`,
|
||||
enabled: input.enabled,
|
||||
name: input.name.trim(),
|
||||
remark: input.remark?.trim() || null,
|
||||
sourceConfig,
|
||||
sourceConfigDigest,
|
||||
sourceKey: input.sourceKey,
|
||||
};
|
||||
}
|
||||
|
||||
/**
|
||||
* Sorts a canonical configuration into a detached ordinary object for stable JSON hashing.
|
||||
* @param config - Adapter-owned allowlisted scalar configuration.
|
||||
* @returns New key-sorted object that cannot mutate the adapter-owned input object.
|
||||
*/
|
||||
private sortConfig(config: Record<string, string>): Record<string, string> {
|
||||
return Object.fromEntries(
|
||||
Object.entries(config).sort(([left], [right]) =>
|
||||
left.localeCompare(right),
|
||||
),
|
||||
);
|
||||
}
|
||||
|
||||
/**
|
||||
* Locks and loads one active subscription for a direct lifecycle mutation.
|
||||
* @param repository - Repository obtained from the caller's current transaction manager.
|
||||
* @param id - String Snowflake identity of the expected active subscription.
|
||||
* @returns Locked active subscription row.
|
||||
* @throws {SystemMessageContractError} When the row is absent or already soft-deleted.
|
||||
*/
|
||||
private async findActiveForWrite(
|
||||
repository: Repository<QqbotMessageSubscription>,
|
||||
id: string,
|
||||
): Promise<QqbotMessageSubscription> {
|
||||
const current = await repository.findOne({
|
||||
lock: { mode: 'pessimistic_write' },
|
||||
where: { id, isDeleted: false },
|
||||
});
|
||||
if (!current) throw new SystemMessageContractError('invalid_source_config');
|
||||
return current;
|
||||
}
|
||||
|
||||
/**
|
||||
* Maps duplicate natural keys to the existing Vben-compatible HTTP conflict response.
|
||||
* @returns Never returns because it always throws an HTTP 409 exception.
|
||||
*/
|
||||
private throwNaturalKeyConflict(): never {
|
||||
return throwVbenError('相同消息源配置的订阅已存在', HttpStatus.CONFLICT);
|
||||
}
|
||||
|
||||
/**
|
||||
* Recognizes only MySQL duplicate-key failures that are the database's concurrency authority.
|
||||
* @param error - Unknown exception returned by transaction, repository, or driver layers.
|
||||
* @returns Whether the error represents MySQL `ER_DUP_ENTRY` / errno 1062.
|
||||
*/
|
||||
private isDuplicateKeyError(error: unknown): boolean {
|
||||
if (!error || typeof error !== 'object') return false;
|
||||
const value = error as { code?: unknown; errno?: unknown };
|
||||
return value.code === 'ER_DUP_ENTRY' || value.errno === 1062;
|
||||
}
|
||||
|
||||
/**
|
||||
* Maps a persistence row to a detached view using its current source definition and validity.
|
||||
* @param subscription - Persisted subscription row whose source configuration must not leak mutability.
|
||||
* @returns Serializable management view with live source summary and validity fields.
|
||||
*/
|
||||
private async toView(
|
||||
subscription: QqbotMessageSubscription,
|
||||
): Promise<MessageSubscriptionView> {
|
||||
const adapter = this.sourceRegistry.get(subscription.sourceKey);
|
||||
const inspection = await adapter.inspectSubscription(
|
||||
subscription.sourceConfig,
|
||||
);
|
||||
return {
|
||||
createTime: this.serializeTime(subscription.createTime),
|
||||
enabled: subscription.enabled,
|
||||
id: String(subscription.id),
|
||||
invalidReasonCode: inspection.invalidReasonCode,
|
||||
name: subscription.name,
|
||||
remark: subscription.remark?.trim() || null,
|
||||
sourceConfig: structuredClone(
|
||||
subscription.sourceConfig,
|
||||
) as unknown as MessageSubscriptionView['sourceConfig'],
|
||||
sourceKey: subscription.sourceKey,
|
||||
sourceName: adapter.definition.displayName,
|
||||
sourceSummary: inspection.sourceSummary,
|
||||
updateTime: this.serializeTime(subscription.updateTime),
|
||||
valid: inspection.valid,
|
||||
};
|
||||
}
|
||||
|
||||
/**
|
||||
* Serializes project datetime values without silently converting them to UTC ISO text.
|
||||
* @param value - Entity date value transformed by the project's KtDateTime column type.
|
||||
* @returns Project-formatted datetime text used by the management contract.
|
||||
*/
|
||||
private serializeTime(value: QqbotMessageSubscription['createTime']): string {
|
||||
return String(value);
|
||||
}
|
||||
}
|
||||
@ -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 { 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';
|
||||
|
||||
export { QQBOT_CORE_DOMAIN_CONTRACT } from './contract/qqbot-core.contract';
|
||||
@ -87,6 +88,7 @@ export const QQBOT_CORE_CONTROLLERS = [
|
||||
export const QQBOT_CORE_PROVIDERS = [
|
||||
SystemMessageSourceRegistry,
|
||||
SystemMessageTemplateRendererService,
|
||||
QqbotMessageSubscriptionService,
|
||||
QqbotMessageTemplateService,
|
||||
QqbotAccountService,
|
||||
QqbotBusService,
|
||||
@ -110,6 +112,7 @@ export const QQBOT_CORE_PROVIDERS = [
|
||||
export const QQBOT_CORE_EXPORTS = [
|
||||
SystemMessageSourceRegistry,
|
||||
SystemMessageTemplateRendererService,
|
||||
QqbotMessageSubscriptionService,
|
||||
QqbotMessageTemplateService,
|
||||
QqbotAccountService,
|
||||
QqbotConfigService,
|
||||
|
||||
@ -0,0 +1,626 @@
|
||||
import { createHash } from 'node:crypto';
|
||||
import { HttpException, HttpStatus } from '@nestjs/common';
|
||||
import type { EntityManager, FindOptionsWhere, Repository } from 'typeorm';
|
||||
import { KtDateTime } from '../../../../src/common';
|
||||
import {
|
||||
SystemMessageContractError,
|
||||
type SystemMessageSourceAdapter,
|
||||
} from '../../../../src/modules/qqbot/core/contract/message-push/qqbot-message-push.types';
|
||||
import { SystemMessageSourceRegistry } from '../../../../src/modules/qqbot/core/application/message-push/system-message-source.registry';
|
||||
import { QqbotMessageSubscriptionService } from '../../../../src/modules/qqbot/core/application/message-push/qqbot-message-subscription.service';
|
||||
import { NetworkStunMessageSourceAdapter } from '../../../../src/modules/admin/platform-config/network-management/network-stun-message-source.adapter';
|
||||
import { NetworkDdnsRecord } from '../../../../src/modules/admin/platform-config/network-management/network-ddns.entity';
|
||||
import { NetworkPortForward } from '../../../../src/modules/admin/platform-config/network-management/network-management.entity';
|
||||
import { QqbotMessageSubscription } from '../../../../src/modules/qqbot/core/infrastructure/persistence/message-push/qqbot-message-subscription.entity';
|
||||
|
||||
const SOURCE_KEY = 'network.stun.mapping-port-changed';
|
||||
const NOW = new KtDateTime('2026-07-24 08:09:10');
|
||||
const CONFIG = {
|
||||
ddnsRecordId: '2041700000000000002',
|
||||
portForwardId: '2041700000000000001',
|
||||
};
|
||||
|
||||
/** Creates a subscription fixture with deterministic string IDs and project timestamps. */
|
||||
function subscription(
|
||||
overrides: Partial<QqbotMessageSubscription> = {},
|
||||
): QqbotMessageSubscription {
|
||||
const sourceConfig = { ...CONFIG };
|
||||
const digest = digestFor(sourceConfig);
|
||||
return {
|
||||
activeKey: `${SOURCE_KEY}:${digest}`,
|
||||
createId: jest.fn(),
|
||||
createTime: NOW,
|
||||
enabled: true,
|
||||
id: '100',
|
||||
isDeleted: false,
|
||||
name: '端口提醒',
|
||||
remark: null,
|
||||
sourceConfig,
|
||||
sourceConfigDigest: digest,
|
||||
sourceKey: SOURCE_KEY,
|
||||
updateTime: NOW,
|
||||
...overrides,
|
||||
};
|
||||
}
|
||||
|
||||
/** Calculates the exact stable digest that production natural-key persistence requires. */
|
||||
function digestFor(config: Record<string, string>): string {
|
||||
const json = JSON.stringify(
|
||||
Object.fromEntries(
|
||||
Object.entries(config).sort(([left], [right]) =>
|
||||
left.localeCompare(right),
|
||||
),
|
||||
),
|
||||
);
|
||||
return createHash('sha256').update(json).digest('hex');
|
||||
}
|
||||
|
||||
/** Reads TypeORM Like's internal test value for high-fidelity in-memory filtering. */
|
||||
function likeValue(value: unknown): string | undefined {
|
||||
if (!value || typeof value !== 'object') return undefined;
|
||||
const candidate = value as { _value?: unknown };
|
||||
return typeof candidate._value === 'string' ? candidate._value : undefined;
|
||||
}
|
||||
|
||||
/** Builds a promise gate whose release is controlled by a concurrency test. */
|
||||
function deferred(): {
|
||||
promise: Promise<void>;
|
||||
resolve: () => void;
|
||||
} {
|
||||
let resolve!: () => void;
|
||||
const promise = new Promise<void>((next) => {
|
||||
resolve = next;
|
||||
});
|
||||
return { promise, resolve };
|
||||
}
|
||||
|
||||
/** Builds a controllable adapter whose current validity can change after persistence. */
|
||||
function adapter(
|
||||
inspect = {
|
||||
invalidReasonCode: null,
|
||||
sourceSummary: '帕鲁新世界 · pal.example.com',
|
||||
valid: true,
|
||||
},
|
||||
): jest.Mocked<SystemMessageSourceAdapter> {
|
||||
return {
|
||||
definition: {
|
||||
description: 'STUN 映射端口变化',
|
||||
displayName: 'STUN 端口变化',
|
||||
sourceKey: SOURCE_KEY,
|
||||
subscriptionFields: [],
|
||||
variables: [],
|
||||
version: 1,
|
||||
},
|
||||
inspectSubscription: jest.fn(async (config: Record<string, unknown>) => {
|
||||
void config;
|
||||
return inspect;
|
||||
}),
|
||||
listSubscriptionOptions: jest.fn(),
|
||||
normalizeSubscriptionConfig: jest.fn(async (input: unknown) => {
|
||||
void input;
|
||||
return {
|
||||
canonicalConfig: { ...CONFIG },
|
||||
resourceKey: CONFIG.portForwardId,
|
||||
sourceSummary: '忽略：服务必须使用实时 inspect 结果',
|
||||
};
|
||||
}),
|
||||
resolveDelivery: jest.fn(),
|
||||
validateEventPayload: jest.fn(),
|
||||
};
|
||||
}
|
||||
|
||||
/** Registers one test source adapter in a fresh process-local source registry. */
|
||||
function registry(
|
||||
source: SystemMessageSourceAdapter,
|
||||
): SystemMessageSourceRegistry {
|
||||
const value = new SystemMessageSourceRegistry();
|
||||
value.register(source);
|
||||
return value;
|
||||
}
|
||||
|
||||
/** Builds a transaction-aware repository fake backed by mutable subscription rows. */
|
||||
function setup(
|
||||
items: QqbotMessageSubscription[] = [],
|
||||
source: SystemMessageSourceAdapter = adapter(),
|
||||
) {
|
||||
let createSequence = 1000;
|
||||
const subscriptionRepository = {
|
||||
create: jest.fn((input) =>
|
||||
subscription({ id: String(createSequence++), ...input }),
|
||||
),
|
||||
findAndCount: jest.fn(
|
||||
async ({
|
||||
skip = 0,
|
||||
take = items.length,
|
||||
where,
|
||||
}: {
|
||||
skip?: number;
|
||||
take?: number;
|
||||
where: FindOptionsWhere<QqbotMessageSubscription>;
|
||||
}) => {
|
||||
const name = likeValue(where.name)?.slice(1, -1);
|
||||
const matched = items.filter((item) => {
|
||||
if (item.isDeleted !== where.isDeleted) return false;
|
||||
if (where.enabled !== undefined && item.enabled !== where.enabled) {
|
||||
return false;
|
||||
}
|
||||
if (where.sourceKey && item.sourceKey !== where.sourceKey)
|
||||
return false;
|
||||
return !name || item.name.includes(name);
|
||||
});
|
||||
return [matched.slice(skip, skip + take), matched.length];
|
||||
},
|
||||
),
|
||||
findOne: jest.fn(async (options) => {
|
||||
const where = options.where as FindOptionsWhere<QqbotMessageSubscription>;
|
||||
const candidates = items.filter((item) => {
|
||||
if (where.id !== undefined && item.id !== where.id) return false;
|
||||
if (
|
||||
where.activeKey !== undefined &&
|
||||
item.activeKey !== where.activeKey
|
||||
) {
|
||||
return false;
|
||||
}
|
||||
if (
|
||||
where.sourceKey !== undefined &&
|
||||
item.sourceKey !== where.sourceKey
|
||||
) {
|
||||
return false;
|
||||
}
|
||||
if (
|
||||
where.sourceConfigDigest !== undefined &&
|
||||
item.sourceConfigDigest !== where.sourceConfigDigest
|
||||
) {
|
||||
return false;
|
||||
}
|
||||
return (
|
||||
where.isDeleted === undefined || item.isDeleted === where.isDeleted
|
||||
);
|
||||
});
|
||||
if (options.order?.updateTime === 'DESC') {
|
||||
candidates.sort((left, right) => {
|
||||
const time = String(right.updateTime).localeCompare(
|
||||
String(left.updateTime),
|
||||
);
|
||||
return time || right.id.localeCompare(left.id);
|
||||
});
|
||||
}
|
||||
return candidates[0] ?? null;
|
||||
}),
|
||||
save: jest.fn(async (item: QqbotMessageSubscription) => {
|
||||
const existing = items.find((candidate) => candidate.id === item.id);
|
||||
if (existing) Object.assign(existing, item);
|
||||
else items.push(item);
|
||||
return item;
|
||||
}),
|
||||
} as unknown as jest.Mocked<Repository<QqbotMessageSubscription>>;
|
||||
const manager = {
|
||||
getRepository: jest.fn((entity) => {
|
||||
if (entity === QqbotMessageSubscription) return subscriptionRepository;
|
||||
throw new Error('unexpected repository');
|
||||
}),
|
||||
} as unknown as jest.Mocked<EntityManager>;
|
||||
Object.assign(subscriptionRepository, {
|
||||
manager: { transaction: jest.fn((callback) => callback(manager)) },
|
||||
});
|
||||
const service = new QqbotMessageSubscriptionService(
|
||||
subscriptionRepository,
|
||||
registry(source),
|
||||
);
|
||||
return { items, manager, service, source, subscriptionRepository };
|
||||
}
|
||||
|
||||
describe('QqbotMessageSubscriptionService', () => {
|
||||
it('uses sorted allowlisted config JSON for the active natural key', async () => {
|
||||
const source = adapter();
|
||||
source.normalizeSubscriptionConfig.mockResolvedValue({
|
||||
canonicalConfig: {
|
||||
portForwardId: CONFIG.portForwardId,
|
||||
ddnsRecordId: CONFIG.ddnsRecordId,
|
||||
},
|
||||
resourceKey: CONFIG.portForwardId,
|
||||
sourceSummary: 'ignored',
|
||||
});
|
||||
const { service, subscriptionRepository } = setup([], source);
|
||||
|
||||
await expect(
|
||||
service.create({
|
||||
enabled: true,
|
||||
name: '帕鲁端口变更',
|
||||
remark: '',
|
||||
sourceConfig: { ...CONFIG, ignored: 'discarded' },
|
||||
sourceKey: SOURCE_KEY,
|
||||
}),
|
||||
).resolves.toEqual(expect.objectContaining({ sourceConfig: CONFIG }));
|
||||
|
||||
const saved = subscriptionRepository.save.mock.calls[0][0];
|
||||
const expectedDigest = digestFor(CONFIG);
|
||||
expect(saved.sourceConfig).toEqual(CONFIG);
|
||||
expect(saved.sourceConfigDigest).toBe(expectedDigest);
|
||||
expect(saved.activeKey).toBe(`${SOURCE_KEY}:${expectedDigest}`);
|
||||
expect(source.normalizeSubscriptionConfig).toHaveBeenCalledWith({
|
||||
...CONFIG,
|
||||
ignored: 'discarded',
|
||||
});
|
||||
});
|
||||
|
||||
it('returns Vben HTTP 409 for an active duplicate and a duplicate-key create race', async () => {
|
||||
const existing = subscription();
|
||||
const { service } = setup([existing]);
|
||||
|
||||
await expect(
|
||||
service.create({
|
||||
enabled: true,
|
||||
name: '重复',
|
||||
sourceConfig: CONFIG,
|
||||
sourceKey: SOURCE_KEY,
|
||||
}),
|
||||
).rejects.toMatchObject({ status: HttpStatus.CONFLICT });
|
||||
await expect(
|
||||
service.create({
|
||||
enabled: true,
|
||||
name: '重复',
|
||||
sourceConfig: CONFIG,
|
||||
sourceKey: SOURCE_KEY,
|
||||
}),
|
||||
).rejects.not.toBeInstanceOf(SystemMessageContractError);
|
||||
|
||||
const race = setup();
|
||||
race.subscriptionRepository.save.mockRejectedValueOnce({
|
||||
code: 'ER_DUP_ENTRY',
|
||||
errno: 1062,
|
||||
});
|
||||
await expect(
|
||||
race.service.create({
|
||||
enabled: true,
|
||||
name: '竞态',
|
||||
sourceConfig: CONFIG,
|
||||
sourceKey: SOURCE_KEY,
|
||||
}),
|
||||
).rejects.toBeInstanceOf(HttpException);
|
||||
race.subscriptionRepository.save.mockRejectedValueOnce(
|
||||
new Error('database unavailable'),
|
||||
);
|
||||
await expect(
|
||||
race.service.create({
|
||||
enabled: true,
|
||||
name: '故障',
|
||||
sourceConfig: CONFIG,
|
||||
sourceKey: SOURCE_KEY,
|
||||
}),
|
||||
).rejects.toThrow('database unavailable');
|
||||
});
|
||||
|
||||
it('soft deletes safely and revives the newest matching historical row with its original ID', async () => {
|
||||
const older = subscription({
|
||||
activeKey: null,
|
||||
id: '101',
|
||||
isDeleted: true,
|
||||
updateTime: new KtDateTime('2026-07-23 08:09:10'),
|
||||
});
|
||||
const newest = subscription({
|
||||
activeKey: null,
|
||||
enabled: false,
|
||||
id: '102',
|
||||
isDeleted: true,
|
||||
updateTime: NOW,
|
||||
});
|
||||
const { items, service, subscriptionRepository } = setup([older, newest]);
|
||||
|
||||
await expect(
|
||||
service.create({
|
||||
enabled: true,
|
||||
name: '复活',
|
||||
remark: '说明',
|
||||
sourceConfig: CONFIG,
|
||||
sourceKey: SOURCE_KEY,
|
||||
}),
|
||||
).resolves.toEqual(
|
||||
expect.objectContaining({ id: '102', name: '复活', remark: '说明' }),
|
||||
);
|
||||
expect(newest).toEqual(
|
||||
expect.objectContaining({ enabled: true, isDeleted: false }),
|
||||
);
|
||||
expect(older.isDeleted).toBe(true);
|
||||
|
||||
await expect(service.remove('102')).resolves.toBe(true);
|
||||
expect(items.find((item) => item.id === '102')).toEqual(
|
||||
expect.objectContaining({
|
||||
activeKey: null,
|
||||
enabled: false,
|
||||
isDeleted: true,
|
||||
}),
|
||||
);
|
||||
expect(subscriptionRepository.findOne).toHaveBeenLastCalledWith({
|
||||
lock: { mode: 'pessimistic_write' },
|
||||
where: { id: '102', isDeleted: false },
|
||||
});
|
||||
});
|
||||
|
||||
it('updates the same ID while rejecting another active natural key', async () => {
|
||||
const first = subscription({ id: '100' });
|
||||
const secondConfig = {
|
||||
ddnsRecordId: '2041700000000000004',
|
||||
portForwardId: '2041700000000000003',
|
||||
};
|
||||
const second = subscription({
|
||||
activeKey: `${SOURCE_KEY}:${digestFor(secondConfig)}`,
|
||||
id: '101',
|
||||
sourceConfig: secondConfig,
|
||||
sourceConfigDigest: digestFor(secondConfig),
|
||||
});
|
||||
const source = adapter();
|
||||
const { service } = setup([first, second], source);
|
||||
|
||||
source.normalizeSubscriptionConfig.mockResolvedValueOnce({
|
||||
canonicalConfig: secondConfig,
|
||||
resourceKey: secondConfig.portForwardId,
|
||||
sourceSummary: 'other',
|
||||
});
|
||||
await expect(
|
||||
service.update('100', {
|
||||
enabled: true,
|
||||
name: '冲突',
|
||||
sourceConfig: secondConfig,
|
||||
sourceKey: SOURCE_KEY,
|
||||
}),
|
||||
).rejects.toMatchObject({ status: HttpStatus.CONFLICT });
|
||||
|
||||
source.normalizeSubscriptionConfig.mockResolvedValueOnce({
|
||||
canonicalConfig: CONFIG,
|
||||
resourceKey: CONFIG.portForwardId,
|
||||
sourceSummary: 'current',
|
||||
});
|
||||
await expect(
|
||||
service.update('100', {
|
||||
enabled: false,
|
||||
name: '更新',
|
||||
remark: ' 新说明 ',
|
||||
sourceConfig: CONFIG,
|
||||
sourceKey: SOURCE_KEY,
|
||||
}),
|
||||
).resolves.toEqual(
|
||||
expect.objectContaining({
|
||||
enabled: false,
|
||||
id: '100',
|
||||
name: '更新',
|
||||
remark: '新说明',
|
||||
}),
|
||||
);
|
||||
});
|
||||
|
||||
it('pages only undeleted records with real filters and returns detached, real-time source status', async () => {
|
||||
const source = adapter({
|
||||
invalidReasonCode: 'ddns_mapping_mismatch',
|
||||
sourceSummary: '已失效',
|
||||
valid: false,
|
||||
});
|
||||
const first = subscription();
|
||||
const { service, subscriptionRepository } = setup(
|
||||
[
|
||||
first,
|
||||
subscription({ id: '101', isDeleted: true }),
|
||||
subscription({ enabled: false, id: '102' }),
|
||||
subscription({ id: '103', name: '不匹配' }),
|
||||
subscription({ id: '104', sourceKey: 'other.source' }),
|
||||
],
|
||||
source,
|
||||
);
|
||||
|
||||
const page = await service.page({
|
||||
enabled: true,
|
||||
name: '端口',
|
||||
pageNo: 1,
|
||||
pageSize: 10,
|
||||
sourceKey: SOURCE_KEY,
|
||||
});
|
||||
expect(page).toEqual({
|
||||
items: [
|
||||
expect.objectContaining({
|
||||
invalidReasonCode: 'ddns_mapping_mismatch',
|
||||
sourceName: 'STUN 端口变化',
|
||||
sourceSummary: '已失效',
|
||||
valid: false,
|
||||
}),
|
||||
],
|
||||
total: 1,
|
||||
});
|
||||
page.items[0].sourceConfig.portForwardId = 'mutated';
|
||||
expect(first.sourceConfig.portForwardId).toBe(CONFIG.portForwardId);
|
||||
expect(subscriptionRepository.findAndCount).toHaveBeenCalledWith(
|
||||
expect.objectContaining({
|
||||
skip: 0,
|
||||
take: 10,
|
||||
where: {
|
||||
enabled: true,
|
||||
isDeleted: false,
|
||||
name: expect.objectContaining({ _type: 'like', _value: '%端口%' }),
|
||||
sourceKey: SOURCE_KEY,
|
||||
},
|
||||
}),
|
||||
);
|
||||
});
|
||||
|
||||
it('keeps a structurally valid source valid without a current Keeper lease and rejects only enabling an invalid relationship', async () => {
|
||||
const leaseMissing = adapter({
|
||||
invalidReasonCode: null,
|
||||
sourceSummary: '映射关系合法，当前无租约',
|
||||
valid: true,
|
||||
});
|
||||
const valid = setup([subscription({ enabled: false })], leaseMissing);
|
||||
await expect(valid.service.page({})).resolves.toEqual(
|
||||
expect.objectContaining({
|
||||
items: [expect.objectContaining({ valid: true })],
|
||||
}),
|
||||
);
|
||||
await expect(valid.service.setEnabled('100', true)).resolves.toEqual(
|
||||
expect.objectContaining({ enabled: true }),
|
||||
);
|
||||
|
||||
const invalid = adapter({
|
||||
invalidReasonCode: 'ddns_mapping_mismatch',
|
||||
sourceSummary: '失效',
|
||||
valid: false,
|
||||
});
|
||||
const blocked = setup([subscription({ enabled: false })], invalid);
|
||||
await expect(blocked.service.setEnabled('100', true)).rejects.toMatchObject(
|
||||
{ code: 'ddns_mapping_mismatch' },
|
||||
);
|
||||
await expect(blocked.service.setEnabled('100', false)).resolves.toEqual(
|
||||
expect.objectContaining({ enabled: false }),
|
||||
);
|
||||
});
|
||||
|
||||
it('uses the real STUN adapter to keep a structurally valid subscription valid without a lease', async () => {
|
||||
const mapping = {
|
||||
desiredPresence: 'present',
|
||||
externalPort: 8213,
|
||||
id: CONFIG.portForwardId,
|
||||
internalPort: 8213,
|
||||
isDeleted: false,
|
||||
keeperDesiredEnabled: true,
|
||||
name: '帕鲁新世界',
|
||||
protocol: 'udp',
|
||||
} as NetworkPortForward;
|
||||
const record = {
|
||||
domain: 'example.com',
|
||||
enabled: true,
|
||||
id: CONFIG.ddnsRecordId,
|
||||
isDeleted: false,
|
||||
portForwardId: CONFIG.portForwardId,
|
||||
recordType: 'A',
|
||||
sourceType: 'port_forward_ipv4',
|
||||
subDomain: 'pal',
|
||||
} as NetworkDdnsRecord;
|
||||
const source = new NetworkStunMessageSourceAdapter(
|
||||
{
|
||||
findOne: jest.fn(async () => mapping),
|
||||
} as unknown as Repository<NetworkPortForward>,
|
||||
{
|
||||
findOne: jest.fn(async () => record),
|
||||
} as unknown as Repository<NetworkDdnsRecord>,
|
||||
new SystemMessageSourceRegistry(),
|
||||
);
|
||||
const { service } = setup([subscription()], source);
|
||||
|
||||
await expect(service.page({})).resolves.toEqual({
|
||||
items: [
|
||||
expect.objectContaining({
|
||||
invalidReasonCode: null,
|
||||
sourceSummary: '帕鲁新世界 · pal.example.com',
|
||||
valid: true,
|
||||
}),
|
||||
],
|
||||
total: 1,
|
||||
});
|
||||
});
|
||||
|
||||
it('uses the caller transaction lock for binding availability and permits disabled bindings only for non-deleted subscriptions', async () => {
|
||||
const source = adapter({
|
||||
invalidReasonCode: 'ddns_mapping_mismatch',
|
||||
sourceSummary: '失效',
|
||||
valid: false,
|
||||
});
|
||||
const current = subscription();
|
||||
const { manager, service, subscriptionRepository } = setup(
|
||||
[current],
|
||||
source,
|
||||
);
|
||||
|
||||
await expect(
|
||||
service.requireAvailableForBinding(manager, '100', false),
|
||||
).resolves.toBe(current);
|
||||
await expect(
|
||||
service.requireAvailableForBinding(manager, '100', true),
|
||||
).rejects.toMatchObject({ code: 'ddns_mapping_mismatch' });
|
||||
current.enabled = false;
|
||||
await expect(
|
||||
service.requireAvailableForBinding(manager, '100', true),
|
||||
).rejects.toMatchObject({ code: 'subscription_disabled' });
|
||||
current.isDeleted = true;
|
||||
await expect(
|
||||
service.requireAvailableForBinding(manager, '100', false),
|
||||
).rejects.toMatchObject({ code: 'invalid_source_config' });
|
||||
expect(manager.getRepository).toHaveBeenCalledWith(
|
||||
QqbotMessageSubscription,
|
||||
);
|
||||
expect(subscriptionRepository.findOne).toHaveBeenCalledWith({
|
||||
lock: { mode: 'pessimistic_write' },
|
||||
where: { id: '100', isDeleted: false },
|
||||
});
|
||||
});
|
||||
|
||||
it('makes a binding transaction observe deletion after the shared subscription row lock commits', async () => {
|
||||
const current = subscription();
|
||||
const deleteHasLock = deferred();
|
||||
const allowDeleteCommit = deferred();
|
||||
let lockTail = Promise.resolve();
|
||||
|
||||
/** Creates a manager whose pessimistic row lock lasts until its transaction callback settles. */
|
||||
const transactionManager = (() => {
|
||||
let releaseLock: (() => void) | undefined;
|
||||
const acquireLock = async (): Promise<void> => {
|
||||
const previous = lockTail;
|
||||
const next = deferred();
|
||||
lockTail = next.promise;
|
||||
await previous;
|
||||
releaseLock = next.resolve;
|
||||
};
|
||||
const repository = {
|
||||
findOne: jest.fn(async (options) => {
|
||||
if (options.lock?.mode === 'pessimistic_write') {
|
||||
await acquireLock();
|
||||
deleteHasLock.resolve();
|
||||
}
|
||||
return current.isDeleted ? null : current;
|
||||
}),
|
||||
save: jest.fn(async (item: QqbotMessageSubscription) => {
|
||||
if (item.isDeleted) await allowDeleteCommit.promise;
|
||||
return item;
|
||||
}),
|
||||
} as unknown as Repository<QqbotMessageSubscription>;
|
||||
const manager = {
|
||||
getRepository: jest.fn((entity) => {
|
||||
if (entity === QqbotMessageSubscription) return repository;
|
||||
throw new Error('unexpected repository');
|
||||
}),
|
||||
} as unknown as EntityManager;
|
||||
Object.assign(manager, {
|
||||
transaction: async <T>(
|
||||
callback: (currentManager: EntityManager) => Promise<T>,
|
||||
) => {
|
||||
try {
|
||||
return await callback(manager);
|
||||
} finally {
|
||||
releaseLock?.();
|
||||
}
|
||||
},
|
||||
});
|
||||
return manager;
|
||||
})();
|
||||
const service = new QqbotMessageSubscriptionService(
|
||||
{ manager: transactionManager } as Repository<QqbotMessageSubscription>,
|
||||
registry(adapter()),
|
||||
);
|
||||
|
||||
const deletion = service.remove('100');
|
||||
await deleteHasLock.promise;
|
||||
const binding = transactionManager.transaction((manager) =>
|
||||
service.requireAvailableForBinding(manager, '100', false),
|
||||
);
|
||||
await Promise.resolve();
|
||||
allowDeleteCommit.resolve();
|
||||
|
||||
await expect(deletion).resolves.toBe(true);
|
||||
await expect(binding).rejects.toMatchObject({
|
||||
code: 'invalid_source_config',
|
||||
});
|
||||
expect(current).toEqual(
|
||||
expect.objectContaining({
|
||||
activeKey: null,
|
||||
enabled: false,
|
||||
isDeleted: true,
|
||||
}),
|
||||
);
|
||||
});
|
||||
});
|
||||
Loading…
Reference in New Issue
Block a user