diff --git a/src/modules/qqbot/core/application/message-push/qqbot-message-subscription.service.ts b/src/modules/qqbot/core/application/message-push/qqbot-message-subscription.service.ts
new file mode 100644
index 0000000..64ace60
--- /dev/null
+++ b/src/modules/qqbot/core/application/message-push/qqbot-message-subscription.service.ts
@@ -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);
+  }
+}
diff --git a/src/modules/qqbot/core/qqbot-core.module.ts b/src/modules/qqbot/core/qqbot-core.module.ts
index b994892..48b6853 100644
--- a/src/modules/qqbot/core/qqbot-core.module.ts
+++ b/src/modules/qqbot/core/qqbot-core.module.ts
@@ -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,
diff --git a/test/modules/qqbot/message-push/qqbot-message-subscription.service.spec.ts b/test/modules/qqbot/message-push/qqbot-message-subscription.service.spec.ts
new file mode 100644
index 0000000..ca5fa6d
--- /dev/null
+++ b/test/modules/qqbot/message-push/qqbot-message-subscription.service.spec.ts
@@ -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,
+      }),
+    );
+  });
+});