重构：迁移到 NATS

2026-08-04 17:29:02 +08:00
1 parent 6c88fd9f50
commit a940d10592
@@ -51,7 +51,7 @@
 - 数据库：`PostgreSQL` + `Drizzle ORM`
 - 认证：`Better Auth`
 - 缓存 / 分布式协调：`Redis`
- 领域事件：`NATS JetStream`
+- 消息与实时广播：`NATS JetStream` + `NATS Core`
 - AI 能力：`AI SDK` + `DeepSeek`

 ## 当前目录结构
@@ -77,7 +77,7 @@

 - `src/react-app` 使用 `BrowserRouter` 管理前端路由
 - `src/server/app.ts` 提供 `/api/*` 和 `/internal/*` 接口
- `src/index.ts` 在生产环境会注册 Bun 内置 cron，用于单容器内直接触发定时任务
+- `src/index.ts` 在生产环境注册 Bun 内置 cron；Cron 只向 NATS WorkQueue 发布后台 command
 - 前端 SPA 独立部署到 Cloudflare Pages；后端 Docker 不再服务 `index.html` 或静态资源

 当前路由约定：
@@ -93,7 +93,7 @@
 - `Bun 1.3+`
 - `PostgreSQL`
 - `Redis`：不是进程启动硬依赖，但排行榜、世界聊天、部分定时任务等功能会用到
- `NATS JetStream`：进程启动硬依赖，负责消息持久化投递；当前主要承载领域事件和异步投影
+- `NATS`：进程启动硬依赖；JetStream 承载领域事件、异步投影和后台 command，Core 承载跨实例实时广播

 说明：

@@ -302,7 +302,7 @@ docker compose up -d

 当前仓库默认采用两层设计：

- 生产环境单容器运行时，`src/index.ts` 会直接注册 Bun 内置 cron
+- 生产环境中 `src/index.ts` 注册 Bun 内置 cron，Cron 只发布 JetStream command，durable Worker consumer 执行 job runner
 - `/internal/cron/*` 仍然保留，便于手动触发、联调，或后续切回外部调度器

 - `GET /internal/cron/auction-expire`
@@ -325,10 +325,9 @@ docker compose up -d

 说明：

- Bun 内置 cron 运行在 Web 进程内，适合当前“单 Docker 容器先跑起来”的场景
- 这类进程内调度不保证跨重启继续执行；如果后续要更强保证，仍建议切回宿主机 cron、云定时任务或 K8s CronJob
+- Bun 内置 cron 仍运行在 Web 进程内，但实际任务已与调度回调解耦；发布成功的 command 由 JetStream 持久化并可跨应用重启继续执行
 - Bun 的 cron 表达式按 `UTC` 解释，所以 `rank-rewards` 在代码里配置为 `0 16 * * *`，对应北京时间次日 `00:00`
- 内置任务直接调用 job runner，不走 HTTP
+- 内置调度不直接调用 job runner，也不走 HTTP；它发布 `daoyou.command.cron.>` command
 - `/internal/cron/*` 接口继续要求 `Authorization: Bearer ${CRON_SECRET}`，适合人工补跑或外部调度
 - 这些任务内部带 Redis 分布式锁与幂等保护，重复触发会返回 `skipped`

@@ -1,9 +1,13 @@
-# NATS JetStream 领域事件运行手册
+# NATS 消息基础设施运行手册

 ## 运行模型

- `DAOYOU_DOMAIN_EVENTS` 使用文件持久化和 Limits retention，保存 `daoyou.domain.>` 事件 14 天，最大占用 3 GiB。
- `DAOYOU_DOMAIN_EVENT_DLQ` 保存终止失败消息 30 天，最大占用 1 GiB。
+- `DAOYOU_DOMAIN_EVENTS` 使用文件持久化和 Limits retention，保存 `daoyou.domain.>` 事件 14 天，最大占用 2.5 GiB。
+- `DAOYOU_DOMAIN_EVENT_DLQ` 保存终止失败消息 30 天，最大占用 768 MiB。
+- `DAOYOU_BACKGROUND_COMMANDS` 使用 WorkQueue retention 保存 `daoyou.command.cron.>`，最大占用 384 MiB；一个 command 只由一个 Worker 实例成功处理。
+- `DAOYOU_BACKGROUND_COMMAND_DLQ` 保存后台 command 的终止失败消息，最大占用 128 MiB。
+- 四个 Stream 的上限合计为 3.75 GiB，为生产 NATS 的 `max_file_store: 5GB` 留出约 20% 的文件与元数据余量。
+- `daoyou.realtime.>` 使用 NATS Core，不进入 JetStream；用于资源变更、世界聊天和宗门聊天的跨实例实时广播。
 - 应用启动时以幂等方式校验并创建 Stream 和 durable consumer。
 - 业务事务将待发布消息写入通用 PostgreSQL 事务消息表 `wanjiedaoyou_transactional_messages`；领域事件只是当前消息类型之一。
 - 事务消息表只保存通用的 `message_key`、`destination`、`payload` 与发布状态，不持有 aggregate、event type 等领域事件专属列。
@@ -11,7 +15,7 @@
 - 消费者在同一 PostgreSQL 事务中写入通用消费幂等记录并更新业务数据，提交后才 ACK JetStream。
 - 单条事件处理失败 10 次后转入 DLQ；DLQ 发布失败时继续重试原事件。

-`wanjiedaoyou_local_transaction_messages` 是 BullMQ 时代“本地执行消息”的旧模型。迁移只对它添加 deprecated 数据库注释，不删除、不改列，也不再产生新写入。后续应在确认历史数据无保留价值后使用独立迁移清理。
+`wanjiedaoyou_local_transaction_messages` 是 BullMQ 时代“本地执行消息”的旧模型，已确认由迁移删除。新代码只使用通用事务消息表 `wanjiedaoyou_transactional_messages`。

 ## 当前事件

@@ -21,6 +25,41 @@
 | `alchemy.craft.completed` | `daoyou.domain.activity.alchemy-craft-completed.v1` | `task-projector-v1` |
 | `ranking.challenge.completed` | `daoyou.domain.activity.ranking-challenge-completed.v1` | `task-projector-v1` |
 | `dungeon.run.settled` | `daoyou.domain.activity.dungeon-run-settled.v1` | `task-projector-v1` |
+| `yield.claimed` | `daoyou.domain.activity.yield-claimed.v1` | `yield-reward-projector-v1` |
+| `cultivator.realm.changed` | `daoyou.domain.gameplay.cultivator-realm-changed.v1` | `world-rumor-projector-v1`、`ranking-realm-projector-v1` |
+| `mail.created` | `daoyou.domain.communication.mail-created.v1` | `mail-notification-projector-v1` |
+| `craft.item.created` | `daoyou.domain.gameplay.craft-item-created.v1` | `world-rumor-projector-v1` |
+| `market.material.revealed` | `daoyou.domain.gameplay.market-material-revealed.v1` | `world-rumor-projector-v1` |
+| `bet-battle.created` | `daoyou.domain.gameplay.bet-battle-created.v1` | `world-rumor-projector-v1` |
+| `bet-battle.settled` | `daoyou.domain.gameplay.bet-battle-settled.v1` | `world-rumor-projector-v1` |
+| `ranking.position.changed` | `daoyou.domain.gameplay.ranking-position-changed.v1` | `world-rumor-projector-v1` |
+
+`ranking.challenge.completed` 仍保留为战斗完成事实，但当前任务定义没有对应计数目标，任务投影器会显式忽略它。
+
+## 后台 Command
+
+Bun Cron 只负责把以下 command 发布到 WorkQueue Stream，实际 job runner 由 durable consumer `background-command-worker-v1` 执行：
+
+- `auction.expire`
+- `bet-battle.expire`
+- `ranking.rewards.distribute`
+- `market.refresh`
+- `tower.enemy-sets.refresh`
+- `resource-replay.cleanup`
+- `expired-data.cleanup`
+- `material-library.generate`
+
+调度器按 command 类型和时间槽生成 JetStream `Nats-Msg-Id`。多个应用实例同时注册相同 Cron 时，同一时间槽只保留一个 command；业务 job 原有 Redis 锁和幂等检查继续作为执行层保护。
+
+JetStream 只能保证“已经发布成功”的 command 持久化；如果所有应用实例在 Cron 触发时都处于停机状态，则不会自动补发该时间槽。每日任务需要通过 `/internal/cron/*` 人工补跑，或在需要自动补偿时改由外部持久化调度器发布 command。
+
+## 实时 Subject
+
+- `daoyou.realtime.resource-state.<scope-kind>.<base64url-scope-id>`
+- `daoyou.realtime.world-chat`
+- `daoyou.realtime.sect-chat.<base64url-sect-id>`
+
+这些消息只负责唤醒当前在线连接。聊天历史仍由原 Redis list 保存，玩家资源变化仍以 PostgreSQL resource event/replay log 为权威数据，因此 NATS Core 丢失一条实时消息不会造成业务数据丢失。

 ## 必需环境变量

@@ -37,8 +76,8 @@ NATS_PASSWORD=replace-with-production-secret
 1. 停止旧应用，确保不会再产生 BullMQ 消息。
 2. 确认旧 BullMQ 队列已经处理完需要保留的作业。
 3. 在生产 env 文件中配置全部 `NATS_*` 变量。
-4. 执行 `bunx drizzle-kit migrate`。迁移会新增通用事务消息表和消费幂等表，并把旧 `wanjiedaoyou_local_transaction_messages` 标记为 deprecated；不会删除旧表或旧数据。
-5. 启动新应用；启动成功意味着 NATS 连接、Stream 和 consumer 初始化成功。
+4. 执行 `bunx drizzle-kit migrate`。迁移会新增通用事务消息表和消费幂等表，并删除已确认废弃的 `wanjiedaoyou_local_transaction_messages`。注意现有 `0025` 还会删除 `wanjiedaoyou_sect_contribution_ledger` 与 `wanjiedaoyou_sect_daily_commissions`，上线前应单独确认这两张宗门旧表的数据无需保留。
+5. 启动新应用；启动成功意味着 NATS 连接、领域事件 Stream、Command WorkQueue 和全部 consumer 初始化成功。
 6. 检查 `/api/health-check` 返回 `redis: up` 和 `nats: up`。
 7. 检查 NATS `8222` 监控端点与应用日志，确认无 Outbox/consumer 错误。

@@ -56,6 +95,7 @@ docker compose -f docker-compose.nats.yml ps
 - 事务消息积压：查询 `wanjiedaoyou_transactional_messages` 中 `published_at IS NULL` 的数量、最早 `created_at` 和 `last_publish_error`。
 - Consumer 积压：查看 JetStream consumer 的 `num_pending`、 `num_ack_pending` 和 `num_redelivered`。
 - 毒消息：查看 `DAOYOU_DOMAIN_EVENT_DLQ`，subject 为 `daoyou.dead-letter.<consumer-name>`。
+- 后台命令失败：查看 `DAOYOU_BACKGROUND_COMMAND_DLQ`，subject 为 `daoyou.command-dead-letter.background-command-worker-v1`。
 - 手工重放前必须确认 `wanjiedaoyou_message_consumptions` 中对应消费记录是否仍存在；消费记录保留 30 天，长于主 Stream 的 14 天保留期。

 ## 事件演进规则
@@ -2,19 +2,19 @@ import { websocket } from 'hono/bun';
 import app from './server/app';
 import { registerInternalCronJobs } from './server/lib/jobs/internalCronScheduler';
 import {
-  registerDomainEventInfrastructure,
-  shutdownDomainEventInfrastructure,
+  registerMessageInfrastructure,
+  shutdownMessageInfrastructure,
 } from './server/lib/mq/domainEventRegistry';

+await registerMessageInfrastructure();
 registerInternalCronJobs({ enabled: import.meta.env.PROD });
-await registerDomainEventInfrastructure();

 let shuttingDown = false;
 async function shutdown(signal: NodeJS.Signals) {
  if (shuttingDown) return;
  shuttingDown = true;
  console.info('[runtime] graceful shutdown started', { signal });
-  await shutdownDomainEventInfrastructure();
+  await shutdownMessageInfrastructure();
  process.exit(0);
 }

@@ -1,13 +0,0 @@
-export function json(body: unknown, init?: ResponseInit): Response {
-  return Response.json(body, init);
-}
-
-export function runDetached(task: () => Promise<void> | void): void {
-  queueMicrotask(() => {
-    void Promise.resolve()
-      .then(task)
-      .catch((error) => {
-        console.error('[runDetached] task failed:', error);
-      });
-  });
-}
@@ -200,11 +200,12 @@ export async function runBetBattleExpireJob(): Promise<CronJobResult> {
  });
 }

-export async function runRankRewardsJob(): Promise<RankRewardsJobResult> {
+export async function runRankRewardsJob(
+  scheduledAt = new Date(),
+): Promise<RankRewardsJobResult> {
  return withJobLock('rank-rewards', async () => {
-    const now = new Date();
-    const settlementDate = getSettlementDateCN(now);
-    if (!isSettlementMondayCN(now)) {
+    const settlementDate = getSettlementDateCN(scheduledAt);
+    if (!isSettlementMondayCN(scheduledAt)) {
      return {
        success: true,
        processed: 0,
@@ -299,7 +300,9 @@ export async function runMarketRefreshCronJob(): Promise<CronJobResult> {
  });
 }

-export async function runMaterialLibraryDailyGenerationJob(): Promise<CronJobResult> {
+export async function runMaterialLibraryDailyGenerationJob(
+  scheduledAt = new Date(),
+): Promise<CronJobResult> {
  return withJobLock(
    'material-library-daily-generation',
    async () => {
@@ -317,7 +320,7 @@ export async function runMaterialLibraryDailyGenerationJob(): Promise<CronJobRes
        count: settings.count,
        userId: ITEM_LIBRARY_SYSTEM_USER_ID,
        source: 'daily_cron',
-        seed: `daily_cron:${new Date().toISOString()}`,
+        seed: `daily_cron:${getSettlementDateCN(scheduledAt)}`,
      });

      return {
@@ -1,13 +1,5 @@
-import {
-  runAuctionExpireJob,
-  runBetBattleExpireJob,
-  runExpiredDataCleanupJob,
-  runMarketRefreshCronJob,
-  runMaterialLibraryDailyGenerationJob,
-  runResourceReplayCleanupJob,
-  runRankRewardsJob,
-  runTowerEnemySetRefreshJob,
-} from './internalCron';
+import { publishScheduledBackgroundCommand } from '@server/lib/mq/backgroundCommandPublisher';
+import type { BackgroundCommandType } from '@shared/contracts/backgroundCommands';

 const AUCTION_EXPIRE_SCHEDULE = '*/2 * * * *';
 const BET_BATTLE_EXPIRE_SCHEDULE = '*/2 * * * *';
@@ -25,13 +17,12 @@ let schedulerRegistered = false;
 let scheduledTasks: Bun.CronJob[] = [];

 async function runScheduledJob(
-  jobName: string,
-  runJob: () => Promise<unknown>,
+  commandType: BackgroundCommandType,
 ): Promise<void> {
  try {
-    await runJob();
+    await publishScheduledBackgroundCommand(commandType);
  } catch (error) {
-    console.error(`[cron] scheduled ${jobName} failed`, error);
+    console.error(`[cron] publish ${commandType} failed`, error);
  }
 }

@@ -47,35 +38,25 @@ export function registerInternalCronJobs(
  }

  scheduledTasks = [
-    Bun.cron(AUCTION_EXPIRE_SCHEDULE, () =>
-      runScheduledJob('auction-expire', runAuctionExpireJob),
-    ),
+    Bun.cron(AUCTION_EXPIRE_SCHEDULE, () => runScheduledJob('auction.expire')),
    Bun.cron(BET_BATTLE_EXPIRE_SCHEDULE, () =>
-      runScheduledJob('bet-battle-expire', runBetBattleExpireJob),
+      runScheduledJob('bet-battle.expire'),
    ),
    Bun.cron(RANK_REWARDS_SCHEDULE, () =>
-      runScheduledJob('rank-rewards', runRankRewardsJob),
-    ),
-    Bun.cron(MARKET_REFRESH_SCHEDULE, () =>
-      runScheduledJob('market-refresh', runMarketRefreshCronJob),
+      runScheduledJob('ranking.rewards.distribute'),
    ),
+    Bun.cron(MARKET_REFRESH_SCHEDULE, () => runScheduledJob('market.refresh')),
    Bun.cron(TOWER_ENEMY_SETS_SCHEDULE, () =>
-      runScheduledJob('tower-enemy-sets', runTowerEnemySetRefreshJob),
+      runScheduledJob('tower.enemy-sets.refresh'),
    ),
    Bun.cron(RESOURCE_REPLAY_CLEANUP_SCHEDULE, () =>
-      runScheduledJob(
-        'resource-replay-cleanup',
-        runResourceReplayCleanupJob,
-      ),
+      runScheduledJob('resource-replay.cleanup'),
    ),
    Bun.cron(EXPIRED_DATA_CLEANUP_SCHEDULE, () =>
-      runScheduledJob('expired-data-cleanup', runExpiredDataCleanupJob),
+      runScheduledJob('expired-data.cleanup'),
    ),
    Bun.cron(MATERIAL_LIBRARY_DAILY_GENERATION_SCHEDULE, () =>
-      runScheduledJob(
-        'material-library-daily-generation',
-        runMaterialLibraryDailyGenerationJob,
-      ),
+      runScheduledJob('material-library.generate'),
    ),
  ];
  schedulerRegistered = true;
@@ -0,0 +1,169 @@
+import { getJetStreamClient } from '@server/lib/nats';
+import {
+  parseBackgroundCommandEnvelope,
+  type BackgroundCommandEnvelope,
+  type BackgroundCommandType,
+} from '@shared/contracts/backgroundCommands';
+import { JSONCodec, type ConsumerMessages, type JsMsg } from 'nats';
+import {
+  runAuctionExpireJob,
+  runBetBattleExpireJob,
+  runExpiredDataCleanupJob,
+  runMarketRefreshCronJob,
+  runMaterialLibraryDailyGenerationJob,
+  runRankRewardsJob,
+  runResourceReplayCleanupJob,
+  runTowerEnemySetRefreshJob,
+} from '../jobs/internalCron';
+import {
+  BACKGROUND_COMMAND_CONSUMER,
+  COMMAND_DEAD_LETTER_STREAM,
+  COMMAND_DEAD_LETTER_SUBJECT_PREFIX,
+  consumerRetryDelayMs,
+} from './natsTopology';
+
+const MAX_PROCESSING_ATTEMPTS = 10;
+const WORKING_INTERVAL_MS = 30_000;
+const codec = JSONCodec();
+let runningMessages: ConsumerMessages | undefined;
+const activeHandlers = new Set<Promise<void>>();
+
+const handlers = {
+  'auction.expire': () => runAuctionExpireJob(),
+  'bet-battle.expire': () => runBetBattleExpireJob(),
+  'ranking.rewards.distribute': (command) =>
+    runRankRewardsJob(new Date(command.requestedAt)),
+  'market.refresh': () => runMarketRefreshCronJob(),
+  'tower.enemy-sets.refresh': () => runTowerEnemySetRefreshJob(),
+  'resource-replay.cleanup': () => runResourceReplayCleanupJob(),
+  'expired-data.cleanup': () => runExpiredDataCleanupJob(),
+  'material-library.generate': (command) =>
+    runMaterialLibraryDailyGenerationJob(new Date(command.requestedAt)),
+} satisfies Record<
+  BackgroundCommandType,
+  (command: BackgroundCommandEnvelope) => Promise<unknown>
+>;
+
+async function publishDeadLetter(
+  message: JsMsg,
+  command: BackgroundCommandEnvelope | null,
+  error: unknown,
+): Promise<void> {
+  const jetStream = await getJetStreamClient();
+  await jetStream.publish(
+    `${COMMAND_DEAD_LETTER_SUBJECT_PREFIX}.${BACKGROUND_COMMAND_CONSUMER.name}`,
+    codec.encode({
+      consumerName: BACKGROUND_COMMAND_CONSUMER.name,
+      command,
+      originalSubject: message.subject,
+      streamSequence: message.info.streamSequence,
+      deliveryCount: message.info.deliveryCount,
+      failedAt: new Date().toISOString(),
+      error: (error instanceof Error ? error.message : String(error)).slice(
+        0,
+        2_000,
+      ),
+      payload: message.string(),
+    }),
+    {
+      msgID: `${BACKGROUND_COMMAND_CONSUMER.name}:${message.info.streamSequence}`,
+      expect: { streamName: COMMAND_DEAD_LETTER_STREAM },
+      timeout: 5_000,
+    },
+  );
+}
+
+async function processMessage(message: JsMsg): Promise<void> {
+  let command: BackgroundCommandEnvelope | null = null;
+  const workingTimer = setInterval(() => {
+    try {
+      message.working();
+    } catch (error) {
+      console.warn('[background-command-consumer] working heartbeat failed', {
+        streamSequence: message.info.streamSequence,
+        error,
+      });
+    }
+  }, WORKING_INTERVAL_MS);
+  workingTimer.unref();
+  try {
+    command = parseBackgroundCommandEnvelope(message.json());
+    if (command.subject !== message.subject) {
+      throw new Error(
+        `后台命令 subject 不一致: ${command.subject} != ${message.subject}`,
+      );
+    }
+    await handlers[command.type](command);
+    const acknowledged = await message.ackAck({ timeout: 5_000 });
+    if (!acknowledged) throw new Error('后台命令 JetStream 双向 ACK 未确认');
+  } catch (error) {
+    console.error('[background-command-consumer] processing failed', {
+      commandType: command?.type,
+      subject: message.subject,
+      deliveryCount: message.info.deliveryCount,
+      streamSequence: message.info.streamSequence,
+      error,
+    });
+    if (message.info.deliveryCount >= MAX_PROCESSING_ATTEMPTS) {
+      try {
+        await publishDeadLetter(message, command, error);
+        message.term();
+      } catch (deadLetterError) {
+        console.error('[background-command-consumer] dead-letter failed', {
+          streamSequence: message.info.streamSequence,
+          deadLetterError,
+        });
+        message.nak(60_000);
+      }
+    } else {
+      message.nak(consumerRetryDelayMs(message.info.deliveryCount));
+    }
+  } finally {
+    clearInterval(workingTimer);
+  }
+}
+
+export async function startBackgroundCommandConsumer(): Promise<void> {
+  if (runningMessages) return;
+  const jetStream = await getJetStreamClient();
+  const consumer = await jetStream.consumers.get(
+    BACKGROUND_COMMAND_CONSUMER.stream,
+    BACKGROUND_COMMAND_CONSUMER.name,
+  );
+  const messages = await consumer.consume({
+    max_messages: BACKGROUND_COMMAND_CONSUMER.concurrency,
+  });
+  runningMessages = messages;
+  void (async () => {
+    const handlers = new Set<Promise<void>>();
+    try {
+      for await (const message of messages) {
+        const handler = processMessage(message).finally(() => {
+          handlers.delete(handler);
+          activeHandlers.delete(handler);
+        });
+        handlers.add(handler);
+        activeHandlers.add(handler);
+        if (handlers.size >= BACKGROUND_COMMAND_CONSUMER.concurrency) {
+          await Promise.race(handlers);
+        }
+      }
+    } catch (error) {
+      console.error('[background-command-consumer] loop stopped', error);
+    } finally {
+      if (runningMessages === messages) runningMessages = undefined;
+    }
+  })();
+  console.info('[background-command-consumer] started', {
+    consumerName: BACKGROUND_COMMAND_CONSUMER.name,
+    concurrency: BACKGROUND_COMMAND_CONSUMER.concurrency,
+  });
+}
+
+export async function stopBackgroundCommandConsumer(): Promise<void> {
+  const messages = runningMessages;
+  runningMessages = undefined;
+  await messages?.close();
+  await Promise.allSettled([...activeHandlers]);
+  activeHandlers.clear();
+}
@@ -0,0 +1,39 @@
+import { getJetStreamClient } from '@server/lib/nats';
+import {
+  BACKGROUND_COMMAND_DEFINITIONS,
+  BACKGROUND_COMMAND_STREAM,
+  parseBackgroundCommandEnvelope,
+  type BackgroundCommandType,
+} from '@shared/contracts/backgroundCommands';
+import { JSONCodec } from 'nats';
+import { randomUUID } from 'node:crypto';
+
+const codec = JSONCodec();
+
+export async function publishScheduledBackgroundCommand(
+  type: BackgroundCommandType,
+  requestedAt = new Date(),
+): Promise<{ id: string; deduplicationKey: string }> {
+  const definition = BACKGROUND_COMMAND_DEFINITIONS[type];
+  const scheduleBucketStartedAt = new Date(
+    Math.floor(requestedAt.getTime() / definition.scheduleBucketMs) *
+      definition.scheduleBucketMs,
+  ).toISOString();
+  const deduplicationKey = `${type}:${scheduleBucketStartedAt}`;
+  const command = parseBackgroundCommandEnvelope({
+    id: randomUUID(),
+    type,
+    version: definition.version,
+    subject: definition.subject,
+    requestedAt: requestedAt.toISOString(),
+    scheduleBucketStartedAt,
+    deduplicationKey,
+  });
+  const jetStream = await getJetStreamClient();
+  await jetStream.publish(command.subject, codec.encode(command), {
+    msgID: deduplicationKey,
+    expect: { streamName: BACKGROUND_COMMAND_STREAM },
+    timeout: 5_000,
+  });
+  return { id: command.id, deduplicationKey };
+}
@@ -6,9 +6,14 @@ import {
  type DomainEventType,
 } from '@shared/contracts/domainEvents';
 import { JSONCodec, type ConsumerMessages, type JsMsg } from 'nats';
-import { DEAD_LETTER_STREAM, DEAD_LETTER_SUBJECT_PREFIX } from './natsTopology';
+import {
+  consumerRetryDelayMs,
+  DEAD_LETTER_STREAM,
+  DEAD_LETTER_SUBJECT_PREFIX,
+} from './natsTopology';

 const MAX_PROCESSING_ATTEMPTS = 10;
+const WORKING_INTERVAL_MS = 30_000;
 const codec = JSONCodec();
 const runningConsumers = new Set<ConsumerMessages>();
 const activeHandlers = new Set<Promise<void>>();
@@ -51,6 +56,18 @@ async function processMessage(
  registration: DomainEventConsumerRegistration,
  message: JsMsg,
 ): Promise<void> {
+  const workingTimer = setInterval(() => {
+    try {
+      message.working();
+    } catch (error) {
+      console.warn('[domain-event-consumer] working heartbeat failed', {
+        consumerName: registration.consumerName,
+        streamSequence: message.info.streamSequence,
+        error,
+      });
+    }
+  }, WORKING_INTERVAL_MS);
+  workingTimer.unref();
  try {
    const event = parseDomainEventEnvelope(message.json());
    if (event.subject !== message.subject) {
@@ -89,7 +106,9 @@ async function processMessage(
      }
      return;
    }
-    message.nak();
+    message.nak(consumerRetryDelayMs(message.info.deliveryCount));
+  } finally {
+    clearInterval(workingTimer);
  }
 }

@@ -1,19 +1,27 @@
 import { closeNatsConnection, getNatsConnection } from '@server/lib/nats';
+import { projectMailCreated } from '@server/lib/services/MailDomainEventProjector';
+import { projectRealmChangedRanking } from '@server/lib/services/RealmChangedDomainEventProjector';
 import { projectSectConstructionDonation } from '@server/lib/services/sect-organization/SectConstructionSettlementService';
 import { projectTaskDomainEvent } from '@server/lib/services/TaskDomainEventProjector';
+import { projectWorldRumorDomainEvent } from '@server/lib/services/WorldRumorDomainEventProjector';
+import {
+  generateYieldRewardAttachments,
+  projectYieldReward,
+} from '@server/lib/services/YieldDomainEventProjector';
 import {
  isDomainEventType,
  type DomainEventEnvelope,
 } from '@shared/contracts/domainEvents';
+import {
+  startBackgroundCommandConsumer,
+  stopBackgroundCommandConsumer,
+} from './backgroundCommandConsumer';
 import {
  startDomainEventConsumer,
  stopDomainEventConsumers,
 } from './domainEventConsumer';
 import { executeDomainEvent } from './DomainEventExecutor';
-import {
-  DOMAIN_EVENT_CONSUMERS,
-  ensureDomainEventTopology,
-} from './natsTopology';
+import { DOMAIN_EVENT_CONSUMERS, ensureMessageTopology } from './natsTopology';
 import {
  startTransactionalMessageRelay,
  stopTransactionalMessageRelay,
@@ -21,12 +29,13 @@ import {

 let registered = false;

-export async function registerDomainEventInfrastructure(): Promise<void> {
+export async function registerMessageInfrastructure(): Promise<void> {
  if (registered) return;

  await getNatsConnection();
-  await ensureDomainEventTopology();
+  await ensureMessageTopology();
  await Promise.all([
+    startBackgroundCommandConsumer(),
    startDomainEventConsumer({
      consumerName: DOMAIN_EVENT_CONSUMERS.sectFacilityProjector.name,
      concurrency: DOMAIN_EVENT_CONSUMERS.sectFacilityProjector.concurrency,
@@ -40,9 +49,41 @@ export async function registerDomainEventInfrastructure(): Promise<void> {
        'alchemy.craft.completed',
        'ranking.challenge.completed',
        'dungeon.run.settled',
+        'yield.claimed',
      ],
      handle: handleTaskEvent,
    }),
+    startDomainEventConsumer({
+      consumerName: DOMAIN_EVENT_CONSUMERS.yieldRewardProjector.name,
+      concurrency: DOMAIN_EVENT_CONSUMERS.yieldRewardProjector.concurrency,
+      acceptedTypes: ['yield.claimed'],
+      handle: handleYieldRewardEvent,
+    }),
+    startDomainEventConsumer({
+      consumerName: DOMAIN_EVENT_CONSUMERS.worldRumorProjector.name,
+      concurrency: DOMAIN_EVENT_CONSUMERS.worldRumorProjector.concurrency,
+      acceptedTypes: [
+        'cultivator.realm.changed',
+        'craft.item.created',
+        'market.material.revealed',
+        'bet-battle.created',
+        'bet-battle.settled',
+        'ranking.position.changed',
+      ],
+      handle: handleWorldRumorEvent,
+    }),
+    startDomainEventConsumer({
+      consumerName: DOMAIN_EVENT_CONSUMERS.rankingRealmProjector.name,
+      concurrency: DOMAIN_EVENT_CONSUMERS.rankingRealmProjector.concurrency,
+      acceptedTypes: ['cultivator.realm.changed'],
+      handle: handleRankingRealmEvent,
+    }),
+    startDomainEventConsumer({
+      consumerName: DOMAIN_EVENT_CONSUMERS.mailNotificationProjector.name,
+      concurrency: DOMAIN_EVENT_CONSUMERS.mailNotificationProjector.concurrency,
+      acceptedTypes: ['mail.created'],
+      handle: handleMailCreatedEvent,
+    }),
  ]);
  startTransactionalMessageRelay();
  registered = true;
@@ -69,10 +110,57 @@ async function handleTaskEvent(event: DomainEventEnvelope) {
  });
 }

-export async function shutdownDomainEventInfrastructure(): Promise<void> {
+async function handleYieldRewardEvent(event: DomainEventEnvelope) {
+  if (!isDomainEventType(event, 'yield.claimed')) {
+    throw new Error(`历练奖励投影不支持领域事件: ${event.type}`);
+  }
+  const attachments = await generateYieldRewardAttachments(event);
+  await executeDomainEvent({
+    consumerName: DOMAIN_EVENT_CONSUMERS.yieldRewardProjector.name,
+    source: 'yield_reward_domain_event',
+    event,
+    handle: (message, tx) => projectYieldReward(message, attachments, tx),
+  });
+}
+
+async function handleWorldRumorEvent(event: DomainEventEnvelope) {
+  await executeDomainEvent({
+    consumerName: DOMAIN_EVENT_CONSUMERS.worldRumorProjector.name,
+    source: 'world_rumor_domain_event',
+    event,
+    handle: projectWorldRumorDomainEvent,
+  });
+}
+
+async function handleRankingRealmEvent(event: DomainEventEnvelope) {
+  if (!isDomainEventType(event, 'cultivator.realm.changed')) {
+    throw new Error(`境界榜单投影不支持领域事件: ${event.type}`);
+  }
+  await executeDomainEvent({
+    consumerName: DOMAIN_EVENT_CONSUMERS.rankingRealmProjector.name,
+    source: 'ranking_realm_domain_event',
+    event,
+    handle: projectRealmChangedRanking,
+  });
+}
+
+async function handleMailCreatedEvent(event: DomainEventEnvelope) {
+  if (!isDomainEventType(event, 'mail.created')) {
+    throw new Error(`邮件通知投影不支持领域事件: ${event.type}`);
+  }
+  await executeDomainEvent({
+    consumerName: DOMAIN_EVENT_CONSUMERS.mailNotificationProjector.name,
+    source: 'mail_notification_domain_event',
+    event,
+    handle: projectMailCreated,
+  });
+}
+
+export async function shutdownMessageInfrastructure(): Promise<void> {
  if (!registered) return;
  registered = false;
  stopTransactionalMessageRelay();
+  await stopBackgroundCommandConsumer();
  await stopDomainEventConsumers();
  await closeNatsConnection();
 }
@@ -1,4 +1,8 @@
 import { getJetStreamManager } from '@server/lib/nats';
+import {
+  BACKGROUND_COMMAND_STREAM,
+  BACKGROUND_COMMAND_SUBJECT_PREFIX,
+} from '@shared/contracts/backgroundCommands';
 import {
  DOMAIN_EVENT_STREAM,
  DOMAIN_EVENT_SUBJECT_PREFIX,
@@ -18,6 +22,15 @@ import {

 export const DEAD_LETTER_STREAM = 'DAOYOU_DOMAIN_EVENT_DLQ';
 export const DEAD_LETTER_SUBJECT_PREFIX = 'daoyou.dead-letter';
+export const COMMAND_DEAD_LETTER_STREAM = 'DAOYOU_BACKGROUND_COMMAND_DLQ';
+export const COMMAND_DEAD_LETTER_SUBJECT_PREFIX = 'daoyou.command-dead-letter';
+
+export const BACKGROUND_COMMAND_CONSUMER = {
+  stream: BACKGROUND_COMMAND_STREAM,
+  name: 'background-command-worker-v1',
+  filterSubject: `${BACKGROUND_COMMAND_SUBJECT_PREFIX}.>`,
+  concurrency: 4,
+} as const;

 export const DOMAIN_EVENT_CONSUMERS = {
  sectFacilityProjector: {
@@ -30,6 +43,26 @@ export const DOMAIN_EVENT_CONSUMERS = {
    filterSubject: `${DOMAIN_EVENT_SUBJECT_PREFIX}.activity.*.v1`,
    concurrency: 8,
  },
+  yieldRewardProjector: {
+    name: 'yield-reward-projector-v1',
+    filterSubject: `${DOMAIN_EVENT_SUBJECT_PREFIX}.activity.yield-claimed.v1`,
+    concurrency: 2,
+  },
+  worldRumorProjector: {
+    name: 'world-rumor-projector-v1',
+    filterSubject: `${DOMAIN_EVENT_SUBJECT_PREFIX}.gameplay.*.v1`,
+    concurrency: 4,
+  },
+  rankingRealmProjector: {
+    name: 'ranking-realm-projector-v1',
+    filterSubject: `${DOMAIN_EVENT_SUBJECT_PREFIX}.gameplay.cultivator-realm-changed.v1`,
+    concurrency: 4,
+  },
+  mailNotificationProjector: {
+    name: 'mail-notification-projector-v1',
+    filterSubject: `${DOMAIN_EVENT_SUBJECT_PREFIX}.communication.mail-created.v1`,
+    concurrency: 8,
+  },
 } as const;

 const DOMAIN_EVENT_STREAM_CONFIG: Partial<StreamConfig> = {
@@ -40,7 +73,7 @@ const DOMAIN_EVENT_STREAM_CONFIG: Partial<StreamConfig> = {
  storage: StorageType.File,
  discard: DiscardPolicy.Old,
  max_age: nanos(14 * 24 * 60 * 60 * 1_000),
-  max_bytes: 3 * 1_024 * 1_024 * 1_024,
+  max_bytes: 2_560 * 1_024 * 1_024,
  max_msg_size: 256 * 1_024,
  duplicate_window: nanos(2 * 60 * 1_000),
  num_replicas: 1,
@@ -55,21 +88,59 @@ const DEAD_LETTER_STREAM_CONFIG: Partial<StreamConfig> = {
  storage: StorageType.File,
  discard: DiscardPolicy.Old,
  max_age: nanos(30 * 24 * 60 * 60 * 1_000),
-  max_bytes: 1 * 1_024 * 1_024 * 1_024,
+  max_bytes: 768 * 1_024 * 1_024,
  max_msg_size: 512 * 1_024,
  duplicate_window: nanos(2 * 60 * 1_000),
  num_replicas: 1,
  allow_direct: true,
 };

-const CONSUMER_BACKOFF = [
+const BACKGROUND_COMMAND_STREAM_CONFIG: Partial<StreamConfig> = {
+  name: BACKGROUND_COMMAND_STREAM,
+  description: 'Daoyou durable one-to-one background commands',
+  subjects: [`${BACKGROUND_COMMAND_SUBJECT_PREFIX}.>`],
+  retention: RetentionPolicy.Workqueue,
+  storage: StorageType.File,
+  discard: DiscardPolicy.Old,
+  max_age: nanos(7 * 24 * 60 * 60 * 1_000),
+  max_bytes: 384 * 1_024 * 1_024,
+  max_msg_size: 256 * 1_024,
+  duplicate_window: nanos(10 * 60 * 1_000),
+  num_replicas: 1,
+  allow_direct: true,
+};
+
+const COMMAND_DEAD_LETTER_STREAM_CONFIG: Partial<StreamConfig> = {
+  name: COMMAND_DEAD_LETTER_STREAM,
+  description: 'Daoyou terminal background command failures',
+  subjects: [`${COMMAND_DEAD_LETTER_SUBJECT_PREFIX}.>`],
+  retention: RetentionPolicy.Limits,
+  storage: StorageType.File,
+  discard: DiscardPolicy.Old,
+  max_age: nanos(30 * 24 * 60 * 60 * 1_000),
+  max_bytes: 128 * 1_024 * 1_024,
+  max_msg_size: 512 * 1_024,
+  duplicate_window: nanos(10 * 60 * 1_000),
+  num_replicas: 1,
+  allow_direct: true,
+};
+
+export const CONSUMER_RETRY_DELAYS_MS = [
  1_000,
  5_000,
  15_000,
  60_000,
  5 * 60_000,
  15 * 60_000,
-].map(nanos);
+] as const;
+
+export function consumerRetryDelayMs(deliveryCount: number): number {
+  const index = Math.min(
+    Math.max(0, deliveryCount - 1),
+    CONSUMER_RETRY_DELAYS_MS.length - 1,
+  );
+  return CONSUMER_RETRY_DELAYS_MS[index]!;
+}

 function isNotFoundError(error: unknown): boolean {
  return (
@@ -106,7 +177,10 @@ async function ensureConsumer(input: {
    max_deliver: -1,
    max_ack_pending: input.concurrency,
    max_batch: input.concurrency,
-    backoff: CONSUMER_BACKOFF,
+    // Keep ACK timeout independent from business retry delays. JetStream uses
+    // the first backoff value as ack_wait, so an empty list is required here
+    // to clear older consumer configs that accidentally reduced it to 1s.
+    backoff: [],
    filter_subject: input.filterSubject,
  };
  try {
@@ -128,18 +202,66 @@ async function ensureConsumer(input: {
  }
 }

-export async function ensureDomainEventTopology(): Promise<void> {
+async function ensureBackgroundCommandConsumer() {
+  const manager = await getJetStreamManager();
+  const mutableConfig: Partial<ConsumerUpdateConfig> = {
+    description: `Daoyou command consumer ${BACKGROUND_COMMAND_CONSUMER.name}`,
+    ack_wait: nanos(15 * 60 * 1_000),
+    max_deliver: -1,
+    max_ack_pending: BACKGROUND_COMMAND_CONSUMER.concurrency,
+    max_batch: BACKGROUND_COMMAND_CONSUMER.concurrency,
+    backoff: [],
+    filter_subject: BACKGROUND_COMMAND_CONSUMER.filterSubject,
+  };
+  try {
+    await manager.consumers.info(
+      BACKGROUND_COMMAND_STREAM,
+      BACKGROUND_COMMAND_CONSUMER.name,
+    );
+    await manager.consumers.update(
+      BACKGROUND_COMMAND_STREAM,
+      BACKGROUND_COMMAND_CONSUMER.name,
+      mutableConfig,
+    );
+  } catch (error) {
+    if (!isNotFoundError(error)) throw error;
+    await manager.consumers.add(BACKGROUND_COMMAND_STREAM, {
+      ...mutableConfig,
+      durable_name: BACKGROUND_COMMAND_CONSUMER.name,
+      ack_policy: AckPolicy.Explicit,
+      deliver_policy: DeliverPolicy.All,
+      replay_policy: ReplayPolicy.Instant,
+    } satisfies Partial<ConsumerConfig>);
+  }
+}
+
+export async function ensureMessageTopology(): Promise<void> {
  await ensureStream(
    DOMAIN_EVENT_STREAM_CONFIG as Partial<StreamConfig> & { name: string },
  );
  await ensureStream(
    DEAD_LETTER_STREAM_CONFIG as Partial<StreamConfig> & { name: string },
  );
-  await Promise.all(Object.values(DOMAIN_EVENT_CONSUMERS).map(ensureConsumer));
+  await ensureStream(
+    BACKGROUND_COMMAND_STREAM_CONFIG as Partial<StreamConfig> & {
+      name: string;
+    },
+  );
+  await ensureStream(
+    COMMAND_DEAD_LETTER_STREAM_CONFIG as Partial<StreamConfig> & {
+      name: string;
+    },
+  );
+  await Promise.all([
+    ...Object.values(DOMAIN_EVENT_CONSUMERS).map(ensureConsumer),
+    ensureBackgroundCommandConsumer(),
+  ]);
  console.info('[nats] JetStream topology ready', {
    stream: DOMAIN_EVENT_STREAM,
    consumers: Object.values(DOMAIN_EVENT_CONSUMERS).map(
      (consumer) => consumer.name,
    ),
+    commandStream: BACKGROUND_COMMAND_STREAM,
+    commandConsumer: BACKGROUND_COMMAND_CONSUMER.name,
  });
 }
@@ -84,7 +84,18 @@ function parseStoredMessage(raw: unknown): WorldChatMessageDTO | null {
  return null;
 }

+function newestFirst(
+  left: WorldChatMessageDTO,
+  right: WorldChatMessageDTO,
+): number {
+  const difference =
+    Date.parse(right.createdAt) - Date.parse(left.createdAt);
+  return Number.isFinite(difference) ? difference : 0;
+}
+
 export async function createMessage(data: {
+  id?: string;
+  createdAt?: string;
  senderUserId: string;
  senderCultivatorId: string | null;
  senderName: string;
@@ -96,7 +107,7 @@ export async function createMessage(data: {
  payload: WorldChatPayload;
 }): Promise<WorldChatMessageDTO> {
  const message: WorldChatMessageDTO = {
-    id: randomUUID(),
+    id: data.id ?? randomUUID(),
    channel: data.channel ?? 'world',
    sectId: null,
    senderUserId: data.senderUserId,
@@ -108,11 +119,16 @@ export async function createMessage(data: {
    textContent: data.textContent ?? null,
    payload: data.payload,
    status: 'active',
-    createdAt: new Date().toISOString(),
+    createdAt: data.createdAt ?? new Date().toISOString(),
  };

-  await redis.lpush(WORLD_CHAT_LIST_KEY, JSON.stringify(message));
-  await redis.ltrim(WORLD_CHAT_LIST_KEY, 0, WORLD_CHAT_MAX_MESSAGES - 1);
+  const encoded = JSON.stringify(message);
+  await redis
+    .multi()
+    .lrem(WORLD_CHAT_LIST_KEY, 0, encoded)
+    .lpush(WORLD_CHAT_LIST_KEY, encoded)
+    .ltrim(WORLD_CHAT_LIST_KEY, 0, WORLD_CHAT_MAX_MESSAGES - 1)
+    .exec();
  publishWorldChatMessage(message);

  return message;
@@ -137,9 +153,9 @@ export async function listMessages(options: {
    .map((raw) => parseStoredMessage(raw))
    .filter((item): item is WorldChatMessageDTO => Boolean(item))
    .filter(
-      (item) =>
-        options.channel === 'all' || item.channel === options.channel,
-    );
+      (item) => options.channel === 'all' || item.channel === options.channel,
+    )
+    .sort(newestFirst);
  const pageRows = parsedRows.slice(start, end);
  const hasMore = pageRows.length > options.pageSize;
  const trimmedRows = hasMore ? pageRows.slice(0, options.pageSize) : pageRows;
@@ -163,5 +179,6 @@ export async function listLatestMessages(
    .map((raw) => parseStoredMessage(raw))
    .filter((item): item is WorldChatMessageDTO => Boolean(item))
    .filter((item) => channel === 'all' || item.channel === channel)
+    .sort(newestFirst)
    .slice(0, limit);
 }
@@ -1,8 +1,9 @@
 import type { DbTransaction } from '@server/lib/drizzle/db';
-import { cultivators, mails } from '@server/lib/drizzle/schema';
+import { cultivators } from '@server/lib/drizzle/schema';
 import { redisLockKeys, withRedisLock } from '@server/lib/redis/lock';
+import { getPlayerLoadoutByCultivatorId } from '@server/lib/services/cultivator/CultivatorLoadoutReader';
 import type { ResourceChangeDescriptor } from '@shared/contracts/resources';
-import { and, eq, sql } from 'drizzle-orm';
+import { eq } from 'drizzle-orm';
 import {
  buyItem,
  cancelListing,
@@ -10,9 +11,6 @@ import {
  listItem,
 } from './AuctionService';
 import { playerCommandExecutor } from './CommandExecutors';
-import {
-  getPlayerLoadoutByCultivatorId,
-} from '@server/lib/services/cultivator/CultivatorLoadoutReader';
 import { readCultivatorName } from './cultivator/CultivatorFactsReader';

 export async function executeAuctionBuyCommand(args: {
@@ -34,15 +32,6 @@ export async function executeAuctionBuyCommand(args: {
    .from(cultivators)
    .where(eq(cultivators.id, args.buyerCultivatorId))
    .limit(1);
-  const [mailSummary] = await args.tx
-    .select({ unreadCount: sql<number>`count(*)::int` })
-    .from(mails)
-    .where(
-      and(
-        eq(mails.cultivatorId, args.buyerCultivatorId),
-        eq(mails.isRead, false),
-      ),
-    );
  if (!currency) throw new Error('拍卖结算后角色不存在');
  return {
    result: { message: '成功购入物品，请查收邮件' },
@@ -53,12 +42,6 @@ export async function executeAuctionBuyCommand(args: {
        operation: 'merge',
        payload: { spiritStones: currency.spiritStones },
      },
-      {
-        resourceTopic: 'player.mail-summary',
-        eventType: 'mail.auction.purchase.created',
-        operation: 'merge',
-        payload: { unreadCount: Number(mailSummary?.unreadCount ?? 0) },
-      },
    ] satisfies ResourceChangeDescriptor[],
  };
 }
@@ -108,25 +91,9 @@ export async function executeAuctionCancelCommand(args: {
    tx: args.tx,
    deferCacheClear: true,
  });
-  const [mailSummary] = await args.tx
-    .select({ unreadCount: sql<number>`count(*)::int` })
-    .from(mails)
-    .where(
-      and(
-        eq(mails.cultivatorId, args.cultivatorId),
-        eq(mails.isRead, false),
-      ),
-    );
  return {
    result: { message: '物品已下架，将通过邮件返还' },
-    resourceChanges: [
-      {
-        resourceTopic: 'player.mail-summary',
-        eventType: 'mail.auction.cancel.created',
-        operation: 'merge',
-        payload: { unreadCount: Number(mailSummary?.unreadCount ?? 0) },
-      },
-    ] satisfies ResourceChangeDescriptor[],
+    resourceChanges: [],
  };
 }

@@ -195,10 +162,7 @@ export async function listAuctionItem(args: {
      timeoutMs: 10_000,
    },
    command: async (tx) => {
-      const { name } = await readCultivatorName(
-        args.actor.cultivatorId,
-        tx,
-      );
+      const { name } = await readCultivatorName(args.actor.cultivatorId, tx);
      return executeAuctionListCommand({
        userId: args.actor.userId,
        cultivatorId: args.actor.cultivatorId,
@@ -237,6 +201,7 @@ export async function cancelAuctionListing(args: {
        userId: args.actor.userId,
        cultivatorId: args.actor.cultivatorId,
        source: 'auction_cancel',
+        allowEmpty: true,
        idempotency: {
          key: `auction-cancel:${args.listingId}`,
          fingerprint: `${args.actor.cultivatorId}:${args.listingId}`,
@@ -1,22 +1,20 @@
 import type { DbTransaction } from '@server/lib/drizzle/db';
 import { cultivators } from '@server/lib/drizzle/schema';
-import * as betBattleRepository from '@server/lib/repositories/betBattleRepository';
-import { createMessage } from '@server/lib/repositories/worldChatRepository';
+import { createDomainEvent } from '@server/lib/mq/domainEventWriter';
+import { publishTransactionalMessageBestEffort } from '@server/lib/mq/transactionalMessagePublisher';
 import { redisLockKeys, withRedisLock } from '@server/lib/redis/lock';
+import * as betBattleRepository from '@server/lib/repositories/betBattleRepository';
+import { getPlayerLoadoutByCultivatorId } from '@server/lib/services/cultivator/CultivatorLoadoutReader';
 import type { ResourceChangeDescriptor } from '@shared/contracts/resources';
 import type { RealmType } from '@shared/types/constants';
 import { eq } from 'drizzle-orm';
-import { playerCommandExecutor } from './CommandExecutors';
 import {
  cancelBetBattle,
  challengeBetBattle,
  createBetBattle,
  type BetStakeInventoryChange,
 } from './BetBattleService';
-import {
-  getPlayerLoadoutByCultivatorId,
-} from '@server/lib/services/cultivator/CultivatorLoadoutReader';
-import { readPlayerMailSummary } from './PlayerResourceReaderService';
+import { playerCommandExecutor } from './CommandExecutors';
 import { readCultivatorName } from './cultivator/CultivatorFactsReader';

 type Stake = {
@@ -81,9 +79,12 @@ async function stakeResourceChanges(
 }

 export async function executeBetBattleCreateCommand(
-  args: Parameters<typeof createBetBattle>[0] & { tx: DbTransaction },
+  args: Parameters<typeof createBetBattle>[0] & {
+    creatorUserId: string;
+    tx: DbTransaction;
+  },
 ) {
-  const { tx, ...input } = args;
+  const { tx, creatorUserId, ...input } = args;
  const created = await createBetBattle(input, { tx });
  const resourceChanges = await stakeResourceChanges({
    ...input,
@@ -96,6 +97,21 @@ export async function executeBetBattleCreateCommand(
    tx,
  );
  if (!listing) throw new Error('赌战创建后记录不存在');
+  const event = await createDomainEvent(
+    {
+      type: 'bet-battle.created',
+      aggregate: { type: 'bet-battle', id: created.battleId },
+      data: {
+        userId: creatorUserId,
+        cultivatorId: input.creatorId,
+        cultivatorName: input.creatorName,
+        battleId: created.battleId,
+        ...(input.taunt?.trim() ? { taunt: input.taunt.trim() } : {}),
+      },
+      deduplicationKey: created.battleId,
+    },
+    tx,
+  );
  return {
    result: {
      battleId: created.battleId,
@@ -103,6 +119,7 @@ export async function executeBetBattleCreateCommand(
      listing,
    },
    resourceChanges,
+    domainEventId: event.id,
  };
 }

@@ -117,23 +134,12 @@ export async function executeBetBattleCancelCommand(args: {
    args.tx,
  );
  if (!listing) throw new Error('赌战取消后记录不存在');
-  const mailSummary = await readPlayerMailSummary(
-    args.cultivatorId,
-    args.tx,
-  );
  return {
    result: {
      message: '赌战已取消，押注将通过邮件返还',
      listing,
    },
-    resourceChanges: [
-      {
-        resourceTopic: 'player.mail-summary',
-        eventType: 'mail.bet_battle.cancel.created',
-        operation: 'replace',
-        payload: mailSummary,
-      },
-    ] satisfies ResourceChangeDescriptor[],
+    resourceChanges: [],
  };
 }

@@ -149,7 +155,20 @@ export async function executeBetBattleChallengeCommand(
    tx,
    inventoryChange: challenge.stakeInventoryChange,
  });
-  const winnerMailSummary = await readPlayerMailSummary(challenge.winnerId, tx);
+  const event = await createDomainEvent(
+    {
+      type: 'bet-battle.settled',
+      aggregate: { type: 'bet-battle', id: challenge.battleId },
+      data: {
+        userId: input.challengerUserId,
+        cultivatorId: input.challengerId,
+        battleId: challenge.battleId,
+        rumor: challenge.rumor,
+      },
+      deduplicationKey: challenge.battleId,
+    },
+    tx,
+  );
  return {
    result: {
      type: 'battle_result',
@@ -165,16 +184,8 @@ export async function executeBetBattleChallengeCommand(
        rumor: challenge.rumor,
      },
    },
-    resourceChanges: [
-      ...stakeChanges,
-      {
-        scope: { kind: 'cultivator', id: challenge.winnerId },
-        resourceTopic: 'player.mail-summary',
-        eventType: 'mail.bet_battle.settlement.created',
-        operation: 'replace',
-        payload: winnerMailSummary,
-      },
-    ] satisfies ResourceChangeDescriptor[],
+    resourceChanges: stakeChanges,
+    domainEventId: event.id,
  };
 }

@@ -193,45 +204,26 @@ type BetStake = {
  } | null;
 };

-async function publishRumor(args: {
-  userId: string;
-  messageType: 'duel_invite' | 'text';
-  text: string;
-  payload: Record<string, unknown>;
-}): Promise<void> {
-  try {
-    await createMessage({
-      senderUserId: args.userId,
-      senderCultivatorId: null,
-      senderName: '修仙界传闻',
-      senderRealm: '炼气',
-      senderRealmStage: '系统',
-      channel: 'system',
-      messageType: args.messageType,
-      textContent: args.text,
-      payload: args.payload,
-    });
-  } catch (error) {
-    console.error('赌战结算成功但世界频道广播失败:', error);
-  }
-}
-
-export async function createBetBattleCommand(args: {
-  actor: BetBattleActor;
-  minRealm: RealmType;
-  maxRealm: RealmType;
-  taunt?: string;
-} & BetStake) {
+export async function createBetBattleCommand(
+  args: {
+    actor: BetBattleActor;
+    minRealm: RealmType;
+    maxRealm: RealmType;
+    taunt?: string;
+  } & BetStake,
+) {
  const { name: cultivatorName } = await readCultivatorName(
    args.actor.cultivatorId,
  );
+  let domainEventId: string | undefined;
  const committed = await playerCommandExecutor.executeWithLock({
    userId: args.actor.userId,
    cultivatorId: args.actor.cultivatorId,
    source: 'bet_battle_create',
    lock: { context: 'bet-battle-create', timeoutMs: 10_000 },
-    command: (tx) =>
-      executeBetBattleCreateCommand({
+    command: async (tx) => {
+      const command = await executeBetBattleCreateCommand({
+        creatorUserId: args.actor.userId,
        creatorId: args.actor.cultivatorId,
        creatorName: cultivatorName,
        minRealm: args.minRealm,
@@ -241,23 +233,15 @@ export async function createBetBattleCommand(args: {
        spiritStones: args.spiritStones,
        stakeItem: args.stakeItem,
        tx,
-      }),
-  });
-  const taunt = args.taunt?.trim();
-  const rumor = taunt
-    ? `${cultivatorName}在赌战台放话：${taunt} 有胆便来应战！`
-    : `${cultivatorName}在赌战台摆下战帖，静候各路道友应战！`;
-  await publishRumor({
-    userId: args.actor.userId,
-    messageType: 'duel_invite',
-    text: rumor,
-    payload: {
-      battleId: committed.result.battleId,
-      routePath: '/game/bet-battle',
-      taunt: taunt || undefined,
-      expiresAt: undefined,
+      });
+      domainEventId = command.domainEventId;
+      return command;
    },
  });
+  publishTransactionalMessageBestEffort(domainEventId, {
+    source: 'bet_battle_create',
+    cultivatorId: args.actor.cultivatorId,
+  });
  return committed;
 }

@@ -281,6 +265,7 @@ export async function cancelBetBattleCommand(args: {
        userId: args.actor.userId,
        cultivatorId: args.actor.cultivatorId,
        source: 'bet_battle_cancel',
+        allowEmpty: true,
        idempotency: {
          key: `bet-battle-cancel:${args.battleId}`,
          fingerprint: `${args.actor.cultivatorId}:${args.battleId}`,
@@ -295,14 +280,17 @@ export async function cancelBetBattleCommand(args: {
  );
 }

-export async function challengeBetBattleCommand(args: {
-  actor: BetBattleActor;
-  battleId: string;
-  requestFingerprint: string;
-} & BetStake) {
+export async function challengeBetBattleCommand(
+  args: {
+    actor: BetBattleActor;
+    battleId: string;
+    requestFingerprint: string;
+  } & BetStake,
+) {
  const { name: cultivatorName } = await readCultivatorName(
    args.actor.cultivatorId,
  );
+  let domainEventId: string | undefined;
  const committed = await withRedisLock(
    {
      keys: [
@@ -323,8 +311,8 @@ export async function challengeBetBattleCommand(args: {
          key: `bet-battle-challenge:${args.battleId}`,
          fingerprint: args.requestFingerprint,
        },
-        command: (tx) =>
-          executeBetBattleChallengeCommand({
+        command: async (tx) => {
+          const command = await executeBetBattleChallengeCommand({
            battleId: args.battleId,
            challengerId: args.actor.cultivatorId,
            challengerName: cultivatorName,
@@ -333,16 +321,15 @@ export async function challengeBetBattleCommand(args: {
            spiritStones: args.spiritStones,
            stakeItem: args.stakeItem,
            tx,
-          }),
+          });
+          domainEventId = command.domainEventId;
+          return command;
+        },
      }),
  );
-  if (!committed.state.replayed) {
-    await publishRumor({
-      userId: args.actor.userId,
-      messageType: 'text',
-      text: committed.result.settlement.rumor,
-      payload: { text: committed.result.settlement.rumor },
-    });
-  }
+  publishTransactionalMessageBestEffort(domainEventId, {
+    source: 'bet_battle_challenge',
+    cultivatorId: args.actor.cultivatorId,
+  });
  return committed;
 }
@@ -2,7 +2,6 @@ import type { DbTransaction } from '@server/lib/drizzle/db';
 import { createDomainEvent } from '@server/lib/mq/domainEventWriter';
 import { publishTransactionalMessageBestEffort } from '@server/lib/mq/transactionalMessagePublisher';
 import { redisLockKeys, withRedisLock } from '@server/lib/redis/lock';
-import { createMessage } from '@server/lib/repositories/worldChatRepository';
 import { getPlayerLoadoutByCultivatorId } from '@server/lib/services/cultivator/CultivatorLoadoutReader';
 import { normalizeFreeformLlmInput } from '@server/utils/llmPayload';
 import type { QiAction } from '@shared/config/qiSystem';
@@ -39,8 +38,6 @@ import {
 } from './QiResourceChanges';
 import { QiService } from './QiService';

-const WORLD_CHAT_BROADCAST_QUALITY_FLOOR: Quality = '天品';
-
 type CraftTargetPolicy = {
  team: 'enemy' | 'ally' | 'self' | 'any';
  scope: 'single' | 'aoe' | 'random';
@@ -110,7 +107,7 @@ export async function executeCraftCommand(args: {
            input.analysisId,
          );
    let afterCommit: (() => Promise<void>) | undefined;
-    let domainEventId: string | undefined;
+    const domainEventIds: string[] = [];
    const actionInstanceId = randomUUID();
    const committed = await playerCommandExecutor.executeWithLock({
      userId: args.userId,
@@ -140,7 +137,7 @@ export async function executeCraftCommand(args: {
          metadata: { committedAt: new Date().toISOString() },
          tx,
        });
-        domainEventId = (
+        domainEventIds.push(
          await createDomainEvent(
            {
              type: 'alchemy.craft.completed',
@@ -156,8 +153,20 @@ export async function executeCraftCommand(args: {
              deduplicationKey: `${args.cultivatorId}:alchemy:${actionInstanceId}`,
            },
            tx,
-          )
-        ).id;
+          ).then((event) => event.id),
+        );
+        const rumorEventId = await createAlchemyItemCreatedEvent(
+          {
+            userId: args.userId,
+            cultivatorId: args.cultivatorId,
+            cultivatorName,
+            actionInstanceId,
+            consumable: (preparedCommit.result as { consumable?: Consumable })
+              .consumable,
+          },
+          tx,
+        );
+        if (rumorEventId) domainEventIds.push(rumorEventId);
        return {
          result: preparedCommit.result,
          resourceChanges: settleAlchemyCraft({
@@ -168,15 +177,12 @@ export async function executeCraftCommand(args: {
      },
    });
    await runAfterCommit(afterCommit, args.cultivatorId, `alchemy_${mode}`);
-    publishTransactionalMessageBestEffort(domainEventId, {
-      source: `alchemy_${mode}`,
-      cultivatorId: args.cultivatorId,
-    });
-    await broadcastAlchemyRumor({
-      userId: args.userId,
-      cultivatorName: cultivatorName,
-      consumable: (committed.result as { consumable?: Consumable }).consumable,
-    });
+    for (const domainEventId of domainEventIds) {
+      publishTransactionalMessageBestEffort(domainEventId, {
+        source: `alchemy_${mode}`,
+        cultivatorId: args.cultivatorId,
+      });
+    }
    return committed;
  }

@@ -193,6 +199,7 @@ export async function executeCraftCommand(args: {
    },
  );
  let afterCommit: (() => Promise<void>) | undefined;
+  let domainEventId: string | undefined;
  const actionInstanceId = randomUUID();
  const committed = await playerCommandExecutor.executeWithLock({
    userId: args.userId,
@@ -221,6 +228,16 @@ export async function executeCraftCommand(args: {
        metadata: { committedAt: new Date().toISOString() },
        tx,
      });
+      domainEventId = await createCreationItemCreatedEvent(
+        {
+          userId: args.userId,
+          cultivatorId: args.cultivatorId,
+          cultivatorName,
+          actionInstanceId,
+          item: preparedCommit.result as BroadcastableCreationResult,
+        },
+        tx,
+      );
      return {
        result: preparedCommit.result,
        resourceChanges: await settleCreationCraft({
@@ -239,10 +256,9 @@ export async function executeCraftCommand(args: {
    args.cultivatorId,
    `creation_${creationCraftType}`,
  );
-  await broadcastCreationRumor({
-    userId: args.userId,
-    cultivatorName: cultivatorName,
-    item: committed.result as BroadcastableCreationResult,
+  publishTransactionalMessageBestEffort(domainEventId, {
+    source: `creation_${creationCraftType}`,
+    cultivatorId: args.cultivatorId,
  });
  return committed;
 }
@@ -286,6 +302,8 @@ export async function executeCreationConfirmationCommand(args: {
    args.replaceId ?? null,
  );
  let afterCommit: (() => Promise<void>) | undefined;
+  let domainEventId: string | undefined;
+  const actionInstanceId = randomUUID();
  const committed = await playerCommandExecutor.executeWithLock({
    userId: args.userId,
    cultivatorId: args.cultivatorId,
@@ -294,6 +312,16 @@ export async function executeCreationConfirmationCommand(args: {
    command: async (tx) => {
      const preparedCommit = await prepared.commit(tx);
      afterCommit = preparedCommit.afterCommit;
+      domainEventId = await createCreationItemCreatedEvent(
+        {
+          userId: args.userId,
+          cultivatorId: args.cultivatorId,
+          cultivatorName,
+          actionInstanceId,
+          item: preparedCommit.result as BroadcastableCreationResult,
+        },
+        tx,
+      );
      return {
        result: {
          message: '领悟成功，已纳入道基',
@@ -309,10 +337,9 @@ export async function executeCreationConfirmationCommand(args: {
    },
  });
  await runAfterCommit(afterCommit, args.cultivatorId, 'creation_confirm');
-  await broadcastCreationRumor({
-    userId: args.userId,
-    cultivatorName: cultivatorName,
-    item: committed.result.item as BroadcastableCreationResult,
+  publishTransactionalMessageBestEffort(domainEventId, {
+    source: 'creation_confirm',
+    cultivatorId: args.cultivatorId,
  });
  return { kind: 'committed', committed };
 }
@@ -463,13 +490,8 @@ async function runAfterCommit(
  }
 }

-function isBroadcastQuality(quality: string | null): quality is Quality {
-  return (
-    typeof quality === 'string' &&
-    quality in QUALITY_ORDER &&
-    QUALITY_ORDER[quality as Quality] >=
-      QUALITY_ORDER[WORLD_CHAT_BROADCAST_QUALITY_FLOOR]
-  );
+function isKnownQuality(quality: string | null): quality is Quality {
+  return typeof quality === 'string' && quality in QUALITY_ORDER;
 }

 function buildCreationShowcaseSnapshot(
@@ -510,79 +532,86 @@ function buildCreationShowcaseSnapshot(
  };
 }

-async function broadcastCreationRumor(args: {
-  userId: string;
-  cultivatorName: string;
-  item: BroadcastableCreationResult;
-}): Promise<void> {
+async function createCreationItemCreatedEvent(
+  args: {
+    userId: string;
+    cultivatorId: string;
+    cultivatorName: string;
+    actionInstanceId: string;
+    item: BroadcastableCreationResult;
+  },
+  tx: DbTransaction,
+): Promise<string | undefined> {
  if (
    !args.item.id ||
    args.item.needs_replace ||
-    !isBroadcastQuality(args.item.quality)
+    !isKnownQuality(args.item.quality)
  ) {
-    return;
+    return undefined;
  }
-  const text = `由${args.cultivatorName}炼成，品阶已入${args.item.quality}，灵韵自生，足令诸修侧目。`;
-  try {
-    await createMessage({
-      senderUserId: args.userId,
-      senderCultivatorId: null,
-      senderName: '修仙界传闻',
-      senderRealm: '炼气',
-      senderRealmStage: '系统',
-      channel: 'system',
-      messageType: 'item_showcase',
-      textContent: text,
-      payload: {
+  const event = await createDomainEvent(
+    {
+      type: 'craft.item.created',
+      aggregate: { type: 'creation-product', id: args.item.id },
+      data: {
+        userId: args.userId,
+        cultivatorId: args.cultivatorId,
+        cultivatorName: args.cultivatorName,
        itemType: args.item.productType,
        itemId: args.item.id,
-        snapshot: buildCreationShowcaseSnapshot(args.item),
-        text,
+        itemName: args.item.name,
+        quality: args.item.quality,
+        snapshot: buildCreationShowcaseSnapshot(args.item) as unknown as Record<
+          string,
+          unknown
+        >,
      },
-    });
-  } catch (error) {
-    console.error('造物传闻发送失败:', error);
-  }
+      deduplicationKey: `${args.cultivatorId}:craft-item:${args.actionInstanceId}`,
+    },
+    tx,
+  );
+  return event.id;
 }

-async function broadcastAlchemyRumor(args: {
-  userId: string;
-  cultivatorName: string;
-  consumable?: Consumable;
-}): Promise<void> {
-  if (
-    !args.consumable?.id ||
-    !isBroadcastQuality(args.consumable.quality ?? null)
-  ) {
-    return;
+async function createAlchemyItemCreatedEvent(
+  args: {
+    userId: string;
+    cultivatorId: string;
+    cultivatorName: string;
+    actionInstanceId: string;
+    consumable?: Consumable;
+  },
+  tx: DbTransaction,
+): Promise<string | undefined> {
+  const quality = args.consumable?.quality ?? null;
+  if (!args.consumable?.id || !isKnownQuality(quality)) {
+    return undefined;
  }
-  const text = `由${args.cultivatorName}炼成，丹品已入${args.consumable.quality}，药香化霞，足令诸修侧目。`;
-  try {
-    await createMessage({
-      senderUserId: args.userId,
-      senderCultivatorId: null,
-      senderName: '修仙界传闻',
-      senderRealm: '炼气',
-      senderRealmStage: '系统',
-      channel: 'system',
-      messageType: 'item_showcase',
-      textContent: text,
-      payload: {
+  const event = await createDomainEvent(
+    {
+      type: 'craft.item.created',
+      aggregate: { type: 'consumable', id: args.consumable.id },
+      data: {
+        userId: args.userId,
+        cultivatorId: args.cultivatorId,
+        cultivatorName: args.cultivatorName,
        itemType: 'consumable',
        itemId: args.consumable.id,
+        itemName: args.consumable.name,
+        quality,
        snapshot: {
          id: args.consumable.id,
          name: args.consumable.name,
          type: args.consumable.type,
-          quality: args.consumable.quality,
+          quality,
          quantity: args.consumable.quantity,
          description: args.consumable.description,
          spec: args.consumable.spec,
        },
-        text,
      },
-    });
-  } catch (error) {
-    console.error('造物传闻发送失败:', error);
-  }
+      deduplicationKey: `${args.cultivatorId}:craft-item:${args.actionInstanceId}`,
+    },
+    tx,
+  );
+  return event.id;
 }
@@ -0,0 +1,23 @@
+import type { DbTransaction } from '@server/lib/drizzle/db';
+import type { DomainEventEnvelope } from '@shared/contracts/domainEvents';
+import type { ResourceChangeDescriptor } from '@shared/contracts/resources';
+import { readPlayerMailSummary } from './PlayerResourceReaderService';
+
+export async function projectMailCreated(
+  event: DomainEventEnvelope<'mail.created'>,
+  tx: DbTransaction,
+) {
+  const summary = await readPlayerMailSummary(event.data.cultivatorId, tx);
+  return {
+    result: { status: 'applied' as const },
+    resourceChanges: [
+      {
+        scope: { kind: 'cultivator', id: event.data.cultivatorId },
+        resourceTopic: 'player.mail-summary',
+        eventType: 'mail.created',
+        operation: 'replace',
+        payload: summary,
+      },
+    ] satisfies ResourceChangeDescriptor[],
+  };
+}
@@ -1,8 +1,9 @@
-import { getExecutor } from '@server/lib/drizzle/db';
+import { db, getExecutor, type DbTransaction } from '@server/lib/drizzle/db';
 import { mails } from '@server/lib/drizzle/schema';
+import { createDomainEvent } from '@server/lib/mq/domainEventWriter';
+import { publishTransactionalMessageBestEffort } from '@server/lib/mq/transactionalMessagePublisher';
 import type { MailAttachment } from '@shared/types/mail';
 import { eq } from 'drizzle-orm';
-import type { DbTransaction } from '../drizzle/db';

 export type { MailAttachment, MailAttachmentType } from '@shared/types/mail';

@@ -21,20 +22,44 @@ export class MailService {
    // If there are attachments, force type to reward
    const mailType = attachments.length > 0 ? 'reward' : type;

-    const q = getExecutor(tx);
-    const [mail] = await q
-      .insert(mails)
-      .values({
-      cultivatorId,
-      title,
-      content,
-      type: mailType,
-      attachments,
-      isRead: false,
-      isClaimed: false,
-      })
-      .returning({ id: mails.id });
+    const persist = async (q: DbTransaction) => {
+      const [mail] = await q
+        .insert(mails)
+        .values({
+          cultivatorId,
+          title,
+          content,
+          type: mailType,
+          attachments,
+          isRead: false,
+          isClaimed: false,
+        })
+        .returning({ id: mails.id });
+      if (!mail) throw new Error('邮件创建失败');
+      const event = await createDomainEvent(
+        {
+          type: 'mail.created',
+          aggregate: { type: 'mail', id: mail.id },
+          data: {
+            mailId: mail.id,
+            cultivatorId,
+            mailType,
+            attachmentCount: attachments.length,
+          },
+          deduplicationKey: mail.id,
+        },
+        q,
+      );
+      return { ...mail, domainEventId: event.id };
+    };

+    if (tx) return persist(tx);
+    const mail = await db.transaction(persist);
+    publishTransactionalMessageBestEffort(mail.domainEventId, {
+      source: 'mail_created',
+      cultivatorId,
+      mailId: mail.id,
+    });
    return mail;
  }

@@ -47,7 +72,7 @@ export class MailService {
    content: string,
    tx?: DbTransaction,
  ) {
-    await this.sendMail(cultivatorId, title, content, [], 'system', tx);
+    return this.sendMail(cultivatorId, title, content, [], 'system', tx);
  }

  /**
@@ -1,5 +1,7 @@
 import { getExecutor, type DbTransaction } from '@server/lib/drizzle/db';
 import { cultivators, materials } from '@server/lib/drizzle/schema';
+import { createDomainEvent } from '@server/lib/mq/domainEventWriter';
+import { publishTransactionalMessageBestEffort } from '@server/lib/mq/transactionalMessagePublisher';
 import { redis } from '@server/lib/redis';
 import { parseRedisJson } from '@server/lib/redis/json';
 import {
@@ -7,7 +9,6 @@ import {
  redisLockKeys,
  withRedisLock,
 } from '@server/lib/redis/lock';
-import { createMessage } from '@server/lib/repositories/worldChatRepository';
 import {
  BASE_PRICES,
  QUALITY_CHANCE_MAP,
@@ -1472,49 +1473,45 @@ export async function prepareMysteryMaterialIdentification(
            : delta <= -2
              ? 'big_loss'
              : 'normal';
-      const isHeavenOrAbove = realOrder >= QUALITY_ORDER['天品'];
+      const [sender] = await tx
+        .select({ userId: cultivators.userId, name: cultivators.name })
+        .from(cultivators)
+        .where(eq(cultivators.id, cultivatorId))
+        .limit(1);
+      if (!sender) throw new Error('鉴定完成后角色不存在');
+      const rumorEvent = await createDomainEvent(
+        {
+          type: 'market.material.revealed',
+          aggregate: { type: 'material', id: revealedMaterialId },
+          data: {
+            userId: sender.userId,
+            cultivatorId,
+            cultivatorName: sender.name,
+            materialId: revealedMaterialId,
+            materialName: revealedMaterial.name,
+            quality: revealedMaterial.rank,
+            snapshot: {
+              id: revealedMaterialId,
+              name: revealedMaterial.name,
+              type: revealedMaterial.type,
+              rank: revealedMaterial.rank,
+              element: revealedMaterial.element,
+              description: revealedMaterial.description,
+              quantity: 1,
+            },
+          },
+          deduplicationKey: `${cultivatorId}:market-reveal:${mystery.mysteryId}`,
+        },
+        tx,
+      );

      const afterCommit = async () => {
        await redis.del(mysteryKey);
-        if (!isHeavenOrAbove) {
-          return;
-        }
-        const [sender] = await getExecutor()
-          .select({ userId: cultivators.userId, name: cultivators.name })
-          .from(cultivators)
-          .where(eq(cultivators.id, cultivatorId))
-          .limit(1);
-        if (sender) {
-          const rumorText = `鉴宝司金光冲霄，${sender.name}鉴出${revealedMaterial.rank}「${revealedMaterial.name}」，天降异象，诸界皆闻。`;
-          try {
-            await createMessage({
-              senderUserId: sender.userId,
-              senderCultivatorId: null,
-              senderName: '修仙界传闻',
-              senderRealm: '炼气',
-              senderRealmStage: '系统',
-              channel: 'system',
-              messageType: 'item_showcase',
-              textContent: rumorText,
-              payload: {
-                itemType: 'material',
-                itemId: revealedMaterialId,
-                snapshot: {
-                  id: revealedMaterialId,
-                  name: revealedMaterial.name,
-                  type: revealedMaterial.type,
-                  rank: revealedMaterial.rank,
-                  element: revealedMaterial.element,
-                  description: revealedMaterial.description,
-                  quantity: 1,
-                },
-                text: rumorText,
-              },
-            });
-          } catch (chatError) {
-            console.error('鉴定传闻发送失败:', chatError);
-          }
-        }
+        publishTransactionalMessageBestEffort(rumorEvent.id, {
+          source: 'market_material_reveal',
+          cultivatorId,
+          materialId: revealedMaterialId,
+        });
      };

      return {
@@ -213,6 +213,7 @@ export function materialLibraryEntryToMaterial(
 export async function generateMaterialLibraryEntries(input: {
  request: ItemLibraryMaterialGenerateInput;
  userId: string;
+  ignoreItemIdConflicts?: boolean;
 }): Promise<ItemLibraryEntry[]> {
  const skeletons: MaterialSkeleton[] = Array.from(
    { length: input.request.count },
@@ -268,10 +269,12 @@ export async function generateMaterialLibraryEntries(input: {
    };
  });

-  const rows = await getExecutor()
-    .insert(itemLibrary)
-    .values(values)
-    .returning();
+  const insert = getExecutor().insert(itemLibrary).values(values);
+  const rows = input.ignoreItemIdConflicts
+    ? await insert
+        .onConflictDoNothing({ target: itemLibrary.itemId })
+        .returning()
+    : await insert.returning();
  return rows.map(parseRow);
 }

@@ -351,6 +354,7 @@ export async function generateDailyMarketMaterialLibraryEntries(input: {
    const entries = await generateMaterialLibraryEntries({
      request,
      userId: input.userId,
+      ignoreItemIdConflicts: true,
    });
    generated.push(...entries);
  }
@@ -13,17 +13,13 @@ import {
  updateRanking,
 } from '@server/lib/redis/rankings';
 import { createBattleRecordV3 } from '@server/lib/repositories/battleRecordV3Repository';
-import { createMessage } from '@server/lib/repositories/worldChatRepository';
+import { loadCultivatorCombatInput } from '@server/lib/services/cultivator/CultivatorCombatProjectionReader';
 import { prepareStandardFullBattle } from '@shared/engine/battle-v5/setup/BattleStateStrategy';
 import type { RealmType } from '@shared/types/constants';
 import { playerCommandExecutor } from './CommandExecutors';
-import {
-  loadCultivatorCombatInput,
-} from '@server/lib/services/cultivator/CultivatorCombatProjectionReader';
-import { MailService, type MailAttachment } from './MailService';
-import { readPlayerMailSummary } from './PlayerResourceReaderService';
-import { simulateBattleV5 } from './simulateBattleV5';
 import { readCultivatorPublicIdentity } from './cultivator/CultivatorFactsReader';
+import { MailService, type MailAttachment } from './MailService';
+import { simulateBattleV5 } from './simulateBattleV5';

 export function settleRankingDirectEntry<T>(result: T) {
  return {
@@ -84,37 +80,6 @@ export class RankingCommandError extends Error {
  }
 }

-async function broadcastRankingChange(params: {
-  userId: string;
-  changeType: Exclude<RankingChangeType, null> | 'direct_entry';
-  challengerName: string;
-  targetName?: string;
-  realm: RealmType;
-  rank: number;
-}): Promise<void> {
-  const rumor =
-    params.changeType === 'direct_entry'
-      ? `万界金榜初开，${params.challengerName}登临${params.realm}天骄榜第${params.rank}名。`
-      : params.changeType === 'vacancy_entry'
-        ? `万界金榜有感，${params.challengerName}虽挑战${params.targetName ?? '榜上修士'}未胜，仍补入${params.realm}天骄榜第${params.rank}名。`
-        : `万界金榜有感，${params.challengerName}击败${params.targetName ?? '榜上修士'}，登临${params.realm}天骄榜第${params.rank}名。`;
-  try {
-    await createMessage({
-      senderUserId: params.userId,
-      senderCultivatorId: null,
-      senderName: '修仙界传闻',
-      senderRealm: '炼气',
-      senderRealmStage: '系统',
-      channel: 'system',
-      messageType: 'text',
-      textContent: rumor,
-      payload: { text: rumor },
-    });
-  } catch (error) {
-    console.error('天骄榜名次变动传音发送失败:', error);
-  }
-}
-
 export async function runRankingBattleCommand(args: {
  userId: string;
  cultivatorId: string;
@@ -125,10 +90,7 @@ export async function runRankingBattleCommand(args: {
  const affectsRanking = args.rankingRealm === identity.realm;
  const challengeCheck = await checkDailyChallenges(args.cultivatorId);
  if (!challengeCheck.success) {
-    throw new RankingCommandError(
-      '今日挑战次数已用完（每日限10次）',
-      403,
-    );
+    throw new RankingCommandError('今日挑战次数已用完（每日限10次）', 403);
  }
  const [isEmpty, challengerRank] = await Promise.all([
    isRankingEmpty(args.rankingRealm),
@@ -137,25 +99,42 @@ export async function runRankingBattleCommand(args: {
  const targetId = args.targetId?.trim() || null;
  if (!targetId && affectsRanking && isEmpty && challengerRank === null) {
    await addToRanking(args.rankingRealm, args.cultivatorId, args.userId, 1);
+    let domainEventId: string | undefined;
    const committed = await playerCommandExecutor.executeWithLock({
      userId: args.userId,
      cultivatorId: args.cultivatorId,
      source: 'ranking_challenge_direct_entry',
      allowEmpty: true,
-      command: async () =>
-        settleRankingDirectEntry({
+      command: async (tx) => {
+        const settled = settleRankingDirectEntry({
          type: 'direct_entry' as const,
          realm: args.rankingRealm,
          rank: 1,
          remainingChallenges: challengeCheck.remaining,
-        }),
+        });
+        domainEventId = (
+          await createDomainEvent(
+            {
+              type: 'ranking.position.changed',
+              aggregate: { type: 'cultivator', id: args.cultivatorId },
+              data: {
+                userId: args.userId,
+                cultivatorId: args.cultivatorId,
+                challengerName: identity.name,
+                realm: args.rankingRealm,
+                rank: 1,
+                changeType: 'direct_entry',
+              },
+            },
+            tx,
+          )
+        ).id;
+        return settled;
+      },
    });
-    await broadcastRankingChange({
-      userId: args.userId,
-      changeType: 'direct_entry',
-      challengerName: identity.name,
-      realm: args.rankingRealm,
-      rank: 1,
+    publishTransactionalMessageBestEffort(domainEventId, {
+      source: 'ranking_challenge_direct_entry',
+      cultivatorId: args.cultivatorId,
    });
    return committed;
  }
@@ -217,7 +196,7 @@ export async function runRankingBattleCommand(args: {
    const remainingChallenges = await incrementDailyChallenges(
      args.cultivatorId,
    );
-    let domainEventId: string | undefined;
+    const domainEventIds: string[] = [];
    const committed = await playerCommandExecutor.executeWithLock({
      userId: args.userId,
      cultivatorId: args.cultivatorId,
@@ -244,22 +223,34 @@ export async function runRankingBattleCommand(args: {
          battleResult,
          tx,
        });
-        domainEventId = command.domainEventId;
+        domainEventIds.push(command.domainEventId);
+        if (rankChangeType && newChallengerRank !== null) {
+          const positionEvent = await createDomainEvent(
+            {
+              type: 'ranking.position.changed',
+              aggregate: { type: 'cultivator', id: args.cultivatorId },
+              data: {
+                userId: args.userId,
+                cultivatorId: args.cultivatorId,
+                challengerName: challengerRecord.cultivator.name,
+                targetName: targetRecord.cultivator.name,
+                realm: args.rankingRealm,
+                rank: newChallengerRank,
+                changeType: rankChangeType,
+              },
+              deduplicationKey: `${args.cultivatorId}:${args.rankingRealm}:${command.domainEventId}`,
+            },
+            tx,
+          );
+          domainEventIds.push(positionEvent.id);
+        }
        return command;
      },
    });
-    publishTransactionalMessageBestEffort(domainEventId, {
-      source: 'ranking_challenge_battle',
-      cultivatorId: args.cultivatorId,
-    });
-    if (rankChangeType && newChallengerRank !== null) {
-      await broadcastRankingChange({
-        userId: args.userId,
-        changeType: rankChangeType,
-        challengerName: challengerRecord.cultivator.name,
-        targetName: targetRecord.cultivator.name,
-        realm: args.rankingRealm,
-        rank: newChallengerRank,
+    for (const domainEventId of domainEventIds) {
+      publishTransactionalMessageBestEffort(domainEventId, {
+        source: 'ranking_challenge_battle',
+        cultivatorId: args.cultivatorId,
      });
    }
    return committed;
@@ -281,6 +272,7 @@ export async function sendWeeklyRankingRewardCommand(args: {
    userId: args.userId,
    cultivatorId: args.cultivatorId,
    source: 'rank_weekly_rewards',
+    allowEmpty: true,
    idempotency: {
      key: args.requestKey,
      fingerprint: args.requestFingerprint,
@@ -297,17 +289,9 @@ export async function sendWeeklyRankingRewardCommand(args: {
      if (!mail) {
        throw new Error('排行榜奖励邮件创建失败');
      }
-      const mailSummary = await readPlayerMailSummary(args.cultivatorId, tx);
      return {
        result: { mailId: mail.id },
-        resourceChanges: [
-          {
-            resourceTopic: 'player.mail-summary',
-            eventType: 'mail.rank_weekly_reward.created',
-            operation: 'replace',
-            payload: mailSummary,
-          },
-        ],
+        resourceChanges: [],
      };
    },
  });
@@ -0,0 +1,16 @@
+import { removeFromAllRankingRealmsExcept } from '@server/lib/redis/rankings';
+import type { DomainEventEnvelope } from '@shared/contracts/domainEvents';
+import type { FeatureCommandResult } from './CommandExecutors';
+
+export async function projectRealmChangedRanking(
+  event: DomainEventEnvelope<'cultivator.realm.changed'>,
+): Promise<FeatureCommandResult<{ status: 'ignored' | 'applied' }>> {
+  if (!event.data.major) {
+    return { result: { status: 'ignored' as const }, resourceChanges: [] };
+  }
+  await removeFromAllRankingRealmsExcept(
+    event.data.cultivatorId,
+    event.data.toRealm,
+  );
+  return { result: { status: 'applied' as const }, resourceChanges: [] };
+}
@@ -1,12 +1,8 @@
-import {
-  redeemCodeClaims,
-  redeemCodes,
-} from '@server/lib/drizzle/schema';
+import { redeemCodeClaims, redeemCodes } from '@server/lib/drizzle/schema';
 import { resolveRedeemCodeRewardAttachments } from '@server/lib/redeem/reward';
 import { and, eq, sql } from 'drizzle-orm';
 import { playerCommandExecutor } from './CommandExecutors';
 import { MailService } from './MailService';
-import { readPlayerMailSummary } from './PlayerResourceReaderService';

 export class RedeemClaimError extends Error {
  constructor(
@@ -26,6 +22,7 @@ export function claimRedeemCode(args: {
    userId: args.userId,
    cultivatorId: args.cultivatorId,
    source: 'redeem_code_claim',
+    allowEmpty: true,
    command: async (tx) => {
      const redeemCode = await tx.query.redeemCodes.findFirst({
        where: eq(redeemCodes.code, args.code),
@@ -86,23 +83,12 @@ export function claimRedeemCode(args: {
        cultivatorId: args.cultivatorId,
        mailId: mail.id,
      });
-      const mailSummary = await readPlayerMailSummary(
-        args.cultivatorId,
-        tx,
-      );
      return {
        result: {
          message: '兑换成功，奖励已通过传音玉简发放',
          mailId: mail.id,
        },
-        resourceChanges: [
-          {
-            resourceTopic: 'player.mail-summary',
-            eventType: 'mail.redeem_code.created',
-            operation: 'replace',
-            payload: mailSummary,
-          },
-        ],
+        resourceChanges: [],
      };
    },
  });
@@ -1,11 +1,11 @@
 import { consumeLifespanAndHandleDepletion } from '@server/lib/lifespan/handleLifespan';
+import { createDomainEvent } from '@server/lib/mq/domainEventWriter';
+import { publishTransactionalMessageBestEffort } from '@server/lib/mq/transactionalMessagePublisher';
 import {
  redisLockKeys,
  withRedisLock,
  type RedisLeaseContext,
 } from '@server/lib/redis/lock';
-import { removeFromAllRankingRealmsExcept } from '@server/lib/redis/rankings';
-import { createMessage } from '@server/lib/repositories/worldChatRepository';
 import { loadPlayerRetreatFacts } from '@server/lib/services/cultivator/CultivatorConditionFactsReader';
 import {
  addBreakthroughHistoryEntry,
@@ -23,7 +23,6 @@ import {
  attemptBreakthrough,
  performCultivation,
 } from '@shared/engine/cultivation/CultivationEngine';
-import type { RealmType } from '@shared/types/constants';
 import type { BreakthroughHistoryEntry } from '@shared/types/cultivator';
 import { randomUUID } from 'crypto';
 import { playerCommandExecutor } from './CommandExecutors';
@@ -48,22 +47,6 @@ export class RetreatCommandError extends Error {
  }
 }

-function buildMajorBreakthroughRumor(
-  cultivatorName: string,
-  toRealm?: string,
-  toStage?: string,
-): string {
-  const target = `${toRealm ?? '未知境界'}${toStage ?? ''}`;
-  const templates = [
-    `${cultivatorName}闭关洞府霞光冲霄，竟一举破境，踏入「${target}」！`,
-    `有修士夜观天象见异光东来，传闻${cultivatorName}已至「${target}」！`,
-    `${cultivatorName}冲关成功，道音震荡八方，自此迈入「${target}」！`,
-    `灵潮翻涌，雷声隐隐，${cultivatorName}于万众传闻中晋升「${target}」！`,
-    `${cultivatorName}破开桎梏，境界再上一重楼，正式踏入「${target}」！`,
-  ];
-  return templates[Math.floor(Math.random() * templates.length)];
-}
-
 function buildLifespanHistoryEntry(
  payload: LifespanExhaustedStoryPayload,
  story: string,
@@ -185,53 +168,22 @@ export function executeRetreatCommand(args: {
            },
          }
        : null;
-      const isMajorBreakthrough =
-        result.summary.success &&
-        result.summary.toRealm &&
-        result.summary.toRealm !== result.summary.fromRealm;
      const retreatResult: RetreatResultData = {
        summary: result.summary,
        storyType: storySource ? 'breakthrough' : null,
        action: 'breakthrough',
      };
-      const committed = await commitBreakthroughRetreat({
+      const { committed, domainEventId } = await commitBreakthroughRetreat({
        userId: args.userId,
        cultivatorId: args.cultivatorId,
        result,
        retreatResult,
        lease,
      });
-
-      if (isMajorBreakthrough) {
-        const rumor = buildMajorBreakthroughRumor(
-          result.cultivator.name,
-          result.summary.toRealm,
-          result.summary.toStage,
-        );
-        try {
-          await createMessage({
-            senderUserId: args.userId,
-            senderCultivatorId: null,
-            senderName: '修仙界传闻',
-            senderRealm: '炼气',
-            senderRealmStage: '系统',
-            channel: 'system',
-            messageType: 'text',
-            textContent: rumor,
-            payload: { text: rumor },
-          });
-        } catch (error) {
-          console.error('突破传闻发送失败:', error);
-        }
-        try {
-          await removeFromAllRankingRealmsExcept(
-            args.cultivatorId,
-            result.summary.toRealm as RealmType,
-          );
-        } catch (error) {
-          console.error('大境界突破后清理天骄榜旧境界失败:', error);
-        }
-      }
+      publishTransactionalMessageBestEffort(domainEventId, {
+        source: 'retreat_breakthrough',
+        cultivatorId: args.cultivatorId,
+      });
      return {
        committed,
        storySource,
@@ -356,7 +308,7 @@ export async function commitCultivationRetreat(args: {
  return { committed, lifespanStoryPayload };
 }

-export function commitBreakthroughRetreat(args: {
+export async function commitBreakthroughRetreat(args: {
  userId: string;
  cultivatorId: string;
  result: ReturnType<typeof attemptBreakthrough>;
@@ -364,7 +316,8 @@ export function commitBreakthroughRetreat(args: {
  lease: RedisLeaseContext;
 }) {
  const actionInstanceId = randomUUID();
-  return playerCommandExecutor.execute({
+  let domainEventId: string | undefined;
+  const committed = await playerCommandExecutor.execute({
    coordination: { mode: 'redis', lease: args.lease },
    userId: args.userId,
    cultivatorId: args.cultivatorId,
@@ -410,6 +363,36 @@ export function commitBreakthroughRetreat(args: {
          tx,
        });
      }
+      if (args.result.summary.success) {
+        const fromRealm = args.result.summary.fromRealm;
+        const fromStage = args.result.summary.fromStage;
+        const toRealm = args.result.summary.toRealm;
+        const toStage = args.result.summary.toStage;
+        if (!toRealm || !toStage) {
+          throw new Error('突破成功但缺少目标境界');
+        }
+        domainEventId = (
+          await createDomainEvent(
+            {
+              type: 'cultivator.realm.changed',
+              aggregate: { type: 'cultivator', id: args.cultivatorId },
+              data: {
+                userId: args.userId,
+                cultivatorId: args.cultivatorId,
+                actionInstanceId,
+                cultivatorName: next.name,
+                fromRealm,
+                fromStage,
+                toRealm,
+                toStage,
+                major: toRealm !== fromRealm,
+              },
+              deduplicationKey: `${args.cultivatorId}:realm:${actionInstanceId}`,
+            },
+            tx,
+          )
+        ).id;
+      }
      return {
        result: args.retreatResult,
        resourceChanges: breakthroughChanges({
@@ -432,4 +415,5 @@ export function commitBreakthroughRetreat(args: {
      };
    },
  });
+  return { committed, domainEventId };
 }
@@ -1,12 +1,9 @@
 import type { DbTransaction } from '@server/lib/drizzle/db';
 import type { ResourceChangeDescriptor } from '@shared/contracts/resources';
 import { RESOURCE_DATA_SCHEMAS } from '@shared/contracts/resources';
-import {
-  readPlayerMailSummary,
-  readPlayerTaskSummary,
-} from './PlayerResourceReaderService';
-import { TaskService } from './TaskService';
 import { playerCommandExecutor } from './CommandExecutors';
+import { readPlayerTaskSummary } from './PlayerResourceReaderService';
+import { TaskService } from './TaskService';

 export function claimTaskRewardCommand(args: {
  userId: string;
@@ -87,14 +84,7 @@ export async function executeTaskRewardClaimCommand(args: {
    args.taskId,
    args.tx,
  );
-  const mailSummary = await readPlayerMailSummary(
-    args.cultivatorId,
-    args.tx,
-  );
-  const taskSummary = await readPlayerTaskSummary(
-    args.cultivatorId,
-    args.tx,
-  );
+  const taskSummary = await readPlayerTaskSummary(args.cultivatorId, args.tx);

  return {
    result,
@@ -114,12 +104,6 @@ export async function executeTaskRewardClaimCommand(args: {
          items: [result.task],
        },
      },
-      {
-        resourceTopic: 'player.mail-summary',
-        eventType: 'mail.reward_created',
-        payload: mailSummary,
-        operation: 'merge',
-      },
    ],
  };
 }
@@ -5,13 +5,14 @@ import {
  type DomainEventEnvelope,
 } from '@shared/contracts/domainEvents';
 import type { ResourceChangeDescriptor } from '@shared/contracts/resources';
+import type { FeatureCommandResult } from './CommandExecutors';
 import { readPlayerTaskSummary } from './PlayerResourceReaderService';
 import { TaskService } from './TaskService';

 export async function projectTaskDomainEvent(
  event: DomainEventEnvelope,
  tx: DbTransaction,
-) {
+): Promise<FeatureCommandResult<{ status: 'applied' | 'ignored' }>> {
  if (isDomainEventType(event, 'alchemy.craft.completed')) {
    await lockCultivatorForStateMutation(tx, event.data.cultivatorId);
    await TaskService.recordTaskEvent(
@@ -23,13 +24,10 @@ export async function projectTaskDomainEvent(
  }

  if (isDomainEventType(event, 'ranking.challenge.completed')) {
-    await lockCultivatorForStateMutation(tx, event.data.cultivatorId);
-    await TaskService.recordTaskEvent(
-      event.data.cultivatorId,
-      'ranking_challenge_battled',
-      { tx },
-    );
-    return taskProjectionResult(event.data.cultivatorId, tx);
+    return {
+      result: { status: 'ignored' as const },
+      resourceChanges: [],
+    };
  }

  if (isDomainEventType(event, 'dungeon.run.settled')) {
@@ -49,10 +47,20 @@ export async function projectTaskDomainEvent(
    return taskProjectionResult(event.data.cultivatorId, tx);
  }

+  if (isDomainEventType(event, 'yield.claimed')) {
+    return {
+      result: { status: 'ignored' as const },
+      resourceChanges: [],
+    };
+  }
+
  throw new Error(`任务投影不支持领域事件: ${event.type}`);
 }

-async function taskProjectionResult(cultivatorId: string, tx: DbTransaction) {
+async function taskProjectionResult(
+  cultivatorId: string,
+  tx: DbTransaction,
+): Promise<FeatureCommandResult<{ status: 'applied' | 'ignored' }>> {
  const taskSummary = await readPlayerTaskSummary(cultivatorId, tx);
  const scope = { kind: 'cultivator' as const, id: cultivatorId };
  return {
@@ -7,10 +7,8 @@ import {
  towerService,
  type TowerBattleRuntimeCommit,
 } from '@server/lib/tower/service';
-import type { ResourceChangeDescriptor } from '@shared/contracts/resources';
 import type { TowerBlessingId } from '@shared/lib/tower';
 import { playerCommandExecutor } from './CommandExecutors';
-import { readPlayerMailSummary } from './PlayerResourceReaderService';
 import { toPlayerStateMutationResponse } from './ResourceMutationResponse';

 const TOWER_COMMAND_TIMEOUT_MS = 240_000;
@@ -20,9 +18,7 @@ type TowerBattlePublicResult = {
    ReturnType<typeof towerService.executeBattle>
  >['battleResult'];
  callbackData: {
-    towerState: Awaited<
-      ReturnType<typeof towerService.executeBattle>
-    >['state'];
+    towerState: Awaited<ReturnType<typeof towerService.executeBattle>>['state'];
    isFinished: boolean;
    settlement: Awaited<
      ReturnType<typeof towerService.executeBattle>
@@ -101,21 +97,12 @@ export function executeTowerBattleCommand(args: {
                milestoneReward: battle.milestoneReward,
              },
            };
-            const resourceChanges: ResourceChangeDescriptor[] = [];
-            if (battle.milestoneReward) {
-              resourceChanges.push({
-                resourceTopic: 'player.mail-summary',
-                eventType: 'mail.tower_milestone.created',
-                operation: 'replace',
-                payload: await readPlayerMailSummary(args.cultivatorId, tx),
-              });
-            }
            return {
              result: {
                response,
                runtimeCommit: battle.runtimeCommit,
              },
-              resourceChanges,
+              resourceChanges: [],
            };
          },
        });
@@ -0,0 +1,123 @@
+import { createMessage } from '@server/lib/repositories/worldChatRepository';
+import {
+  isDomainEventType,
+  type DomainEventEnvelope,
+} from '@shared/contracts/domainEvents';
+import { QUALITY_ORDER } from '@shared/types/constants';
+import type { WorldChatPayload } from '@shared/types/world-chat';
+import type { FeatureCommandResult } from './CommandExecutors';
+
+type RumorProjectionResult = FeatureCommandResult<{
+  status: 'ignored' | 'created';
+}>;
+
+function deterministicTemplate(eventId: string, templates: readonly string[]) {
+  const hash = [...eventId].reduce(
+    (value, character) => (value * 31 + character.charCodeAt(0)) >>> 0,
+    0,
+  );
+  return templates[hash % templates.length];
+}
+
+export async function projectWorldRumorDomainEvent(
+  event: DomainEventEnvelope,
+): Promise<RumorProjectionResult> {
+  if (isDomainEventType(event, 'cultivator.realm.changed')) {
+    if (!event.data.major) return ignored();
+    const target = `${event.data.toRealm}${event.data.toStage}`;
+    const text = deterministicTemplate(event.id, [
+      `${event.data.cultivatorName}闭关洞府霞光冲霄，竟一举破境，踏入「${target}」！`,
+      `有修士夜观天象见异光东来，传闻${event.data.cultivatorName}已至「${target}」！`,
+      `${event.data.cultivatorName}冲关成功，道音震荡八方，自此迈入「${target}」！`,
+      `灵潮翻涌，雷声隐隐，${event.data.cultivatorName}于万众传闻中晋升「${target}」！`,
+      `${event.data.cultivatorName}破开桎梏，境界再上一重楼，正式踏入「${target}」！`,
+    ]);
+    return createRumor(event, event.data.userId, 'text', text, { text });
+  }
+
+  if (isDomainEventType(event, 'craft.item.created')) {
+    if (QUALITY_ORDER[event.data.quality] < QUALITY_ORDER['天品']) {
+      return ignored();
+    }
+    const noun = event.data.itemType === 'consumable' ? '丹品' : '品阶';
+    const flourish =
+      event.data.itemType === 'consumable' ? '药香化霞' : '灵韵自生';
+    const text = `由${event.data.cultivatorName}炼成，${noun}已入${event.data.quality}，${flourish}，足令诸修侧目。`;
+    return createRumor(event, event.data.userId, 'item_showcase', text, {
+      itemType: event.data.itemType,
+      itemId: event.data.itemId,
+      snapshot: event.data.snapshot,
+      text,
+    } as WorldChatPayload);
+  }
+
+  if (isDomainEventType(event, 'market.material.revealed')) {
+    if (QUALITY_ORDER[event.data.quality] < QUALITY_ORDER['天品']) {
+      return ignored();
+    }
+    const text = `鉴宝司金光冲霄，${event.data.cultivatorName}鉴出${event.data.quality}「${event.data.materialName}」，天降异象，诸界皆闻。`;
+    return createRumor(event, event.data.userId, 'item_showcase', text, {
+      itemType: 'material',
+      itemId: event.data.materialId,
+      snapshot: event.data.snapshot,
+      text,
+    } as WorldChatPayload);
+  }
+
+  if (isDomainEventType(event, 'bet-battle.created')) {
+    const text = event.data.taunt
+      ? `${event.data.cultivatorName}在赌战台放话：${event.data.taunt} 有胆便来应战！`
+      : `${event.data.cultivatorName}在赌战台摆下战帖，静候各路道友应战！`;
+    return createRumor(event, event.data.userId, 'duel_invite', text, {
+      battleId: event.data.battleId,
+      routePath: '/game/bet-battle',
+      taunt: event.data.taunt,
+      expiresAt: undefined,
+    });
+  }
+
+  if (isDomainEventType(event, 'bet-battle.settled')) {
+    return createRumor(event, event.data.userId, 'text', event.data.rumor, {
+      text: event.data.rumor,
+    });
+  }
+
+  if (isDomainEventType(event, 'ranking.position.changed')) {
+    const text =
+      event.data.changeType === 'direct_entry'
+        ? `万界金榜初开，${event.data.challengerName}登临${event.data.realm}天骄榜第${event.data.rank}名。`
+        : event.data.changeType === 'vacancy_entry'
+          ? `万界金榜有感，${event.data.challengerName}虽挑战${event.data.targetName ?? '榜上修士'}未胜，仍补入${event.data.realm}天骄榜第${event.data.rank}名。`
+          : `万界金榜有感，${event.data.challengerName}击败${event.data.targetName ?? '榜上修士'}，登临${event.data.realm}天骄榜第${event.data.rank}名。`;
+    return createRumor(event, event.data.userId, 'text', text, { text });
+  }
+
+  throw new Error(`世界传闻投影不支持领域事件: ${event.type}`);
+}
+
+function ignored(): RumorProjectionResult {
+  return { result: { status: 'ignored' }, resourceChanges: [] };
+}
+
+async function createRumor(
+  event: DomainEventEnvelope,
+  senderUserId: string,
+  messageType: 'text' | 'item_showcase' | 'duel_invite',
+  text: string,
+  payload: WorldChatPayload,
+): Promise<RumorProjectionResult> {
+  await createMessage({
+    id: event.id,
+    createdAt: event.occurredAt,
+    senderUserId,
+    senderCultivatorId: null,
+    senderName: '修仙界传闻',
+    senderRealm: '炼气',
+    senderRealmStage: '系统',
+    channel: 'system',
+    messageType,
+    textContent: text,
+    payload,
+  });
+  return { result: { status: 'created' }, resourceChanges: [] };
+}
@@ -1,21 +1,20 @@
 import { cultivators } from '@server/lib/drizzle/schema';
-import { runDetached } from '@server/lib/http/response';
+import { createDomainEvent } from '@server/lib/mq/domainEventWriter';
+import { publishTransactionalMessageBestEffort } from '@server/lib/mq/transactionalMessagePublisher';
 import { redisLockKeys, withRedisLock } from '@server/lib/redis/lock';
 import {
  updateCultivationExp,
  updateSpiritStones,
 } from '@server/lib/services/cultivator/CultivatorStateRepository';
 import { getOrInitCultivationProgress } from '@server/utils/cultivationUtils';
-import { MaterialGenerator } from '@shared/engine/material/creation/MaterialGenerator';
 import type { GeneratedMaterial } from '@shared/engine/material/creation/types';
 import { YieldCalculator } from '@shared/engine/yield/YieldCalculator';
 import type { RealmStage, RealmType } from '@shared/types/constants';
 import type { CultivationProgress } from '@shared/types/cultivator';
 import { and, eq } from 'drizzle-orm';
+import { randomUUID } from 'node:crypto';
 import { getExecutor } from '../drizzle/db';
 import { playerCommandExecutor } from './CommandExecutors';
-import { MailService, type MailAttachment } from './MailService';
-import { readPlayerMailSummary } from './PlayerResourceReaderService';

 export class YieldCommandError extends Error {
  constructor(
@@ -78,6 +77,8 @@ export async function executeYieldCommand(args: {
  userId: string;
  cultivatorId: string;
 }) {
+  const actionInstanceId = randomUUID();
+  let domainEventId: string | undefined;
  const prepared = await withRedisLock(
    {
      key: redisLockKeys.cultivatorMutation(args.cultivatorId),
@@ -172,6 +173,24 @@ export async function executeYieldCommand(args: {
            .update(cultivators)
            .set({ last_yield_at: claimedAt })
            .where(eq(cultivators.id, args.cultivatorId));
+          if (materialCount > 0) {
+            domainEventId = (
+              await createDomainEvent(
+                {
+                  type: 'yield.claimed',
+                  aggregate: { type: 'cultivator', id: args.cultivatorId },
+                  data: {
+                    cultivatorId: args.cultivatorId,
+                    actionInstanceId,
+                    realm: facts.realm,
+                    materialCount,
+                  },
+                  deduplicationKey: `${args.cultivatorId}:yield:${actionInstanceId}`,
+                },
+                tx,
+              )
+            ).id;
+          }
          return {
            result,
            resourceChanges: [
@@ -204,78 +223,9 @@ export async function executeYieldCommand(args: {
      return { committed, result, realm: facts.realm, materialCount };
    },
  );
-
-  if (prepared.materialCount > 0) {
-    runDetached(() =>
-      generateAndSendYieldMaterials({
-        userId: args.userId,
-        cultivatorId: args.cultivatorId,
-        realm: prepared.realm,
-        materialCount: prepared.materialCount,
-      }),
-    );
-  }
+  publishTransactionalMessageBestEffort(domainEventId, {
+    source: 'yield_claim',
+    cultivatorId: args.cultivatorId,
+  });
  return prepared;
 }
-
-async function generateAndSendYieldMaterials(args: {
-  userId: string;
-  cultivatorId: string;
-  realm: RealmType;
-  materialCount: number;
-}): Promise<void> {
-  let materials: GeneratedMaterial[];
-  try {
-    materials = await MaterialGenerator.generateRandom(args.materialCount, {
-      qualityChanceMap: YieldCalculator.getMaterialQualityChanceMap(args.realm),
-    });
-  } catch (error) {
-    console.error('[Yield] 材料生成异常:', error);
-    materials = [];
-  }
-  if (materials.length === 0) return;
-  const attachments: MailAttachment[] = materials.map((material) => ({
-    type: 'material',
-    name: material.name,
-    quantity: material.quantity,
-    data: material,
-  }));
-  for (let attempt = 1; attempt <= 3; attempt += 1) {
-    try {
-      await playerCommandExecutor.executeWithLock({
-        userId: args.userId,
-        cultivatorId: args.cultivatorId,
-        source: 'yield_material_mail',
-        command: async (tx) => {
-          await MailService.sendMail(
-            args.cultivatorId,
-            '历练机缘',
-            '道友历练途中，偶得天材地宝，特以此传音玉简送达。',
-            attachments,
-            'reward',
-            tx,
-          );
-          const summary = await readPlayerMailSummary(args.cultivatorId, tx);
-          return {
-            result: null,
-            resourceChanges: [
-              {
-                resourceTopic: 'player.mail-summary',
-                eventType: 'mail.yield_material.created',
-                operation: 'replace',
-                payload: summary,
-              },
-            ],
-          };
-        },
-      });
-      return;
-    } catch (error) {
-      if (attempt === 3) {
-        console.error('[Yield] 材料奖励邮件发送失败:', error);
-        return;
-      }
-      await new Promise((resolve) => setTimeout(resolve, attempt * 3000));
-    }
-  }
-}
@@ -0,0 +1,46 @@
+import type { DbTransaction } from '@server/lib/drizzle/db';
+import type { DomainEventEnvelope } from '@shared/contracts/domainEvents';
+import { MaterialGenerator } from '@shared/engine/material/creation/MaterialGenerator';
+import { YieldCalculator } from '@shared/engine/yield/YieldCalculator';
+import { MailService, type MailAttachment } from './MailService';
+
+export async function generateYieldRewardAttachments(
+  event: DomainEventEnvelope<'yield.claimed'>,
+): Promise<MailAttachment[]> {
+  const materials = await MaterialGenerator.generateRandom(
+    event.data.materialCount,
+    {
+      qualityChanceMap: YieldCalculator.getMaterialQualityChanceMap(
+        event.data.realm,
+      ),
+    },
+  );
+  if (materials.length === 0) {
+    throw new Error(`历练奖励未生成材料: ${event.id}`);
+  }
+  return materials.map((material) => ({
+    type: 'material',
+    name: material.name,
+    quantity: material.quantity,
+    data: material,
+  }));
+}
+
+export async function projectYieldReward(
+  event: DomainEventEnvelope<'yield.claimed'>,
+  attachments: MailAttachment[],
+  tx: DbTransaction,
+) {
+  await MailService.sendMail(
+    event.data.cultivatorId,
+    '历练机缘',
+    '道友历练途中，偶得天材地宝，特以此传音玉简送达。',
+    attachments,
+    'reward',
+    tx,
+  );
+  return {
+    result: { status: 'created' as const },
+    resourceChanges: [],
+  };
+}
@@ -0,0 +1,74 @@
+import { getNatsConnection } from '@server/lib/nats';
+import { StringCodec, type Subscription } from 'nats';
+
+type NatsMessageHandler = (message: string) => void;
+
+type SharedSubscription = {
+  handlers: Set<NatsMessageHandler>;
+  subscription?: Subscription;
+  closed: boolean;
+  ready: Promise<void>;
+};
+
+const codec = StringCodec();
+const subscriptions = new Map<string, SharedSubscription>();
+
+function createSharedSubscription(subject: string): SharedSubscription {
+  const shared: SharedSubscription = {
+    handlers: new Set(),
+    closed: false,
+    ready: Promise.resolve(),
+  };
+  shared.ready = getNatsConnection()
+    .then(async (connection) => {
+      if (shared.closed) return;
+      const subscription = connection.subscribe(subject);
+      shared.subscription = subscription;
+      for await (const message of subscription) {
+        const decoded = codec.decode(message.data);
+        for (const handler of shared.handlers) handler(decoded);
+      }
+    })
+    .catch((error) => {
+      if (!shared.closed) {
+        console.warn('[nats-core] subscription stopped', { subject, error });
+      }
+    });
+  subscriptions.set(subject, shared);
+  return shared;
+}
+
+export function encodeNatsSubjectToken(value: string): string {
+  return Buffer.from(value, 'utf8').toString('base64url');
+}
+
+export async function publishNatsCoreMessage(
+  subject: string,
+  message: string,
+): Promise<void> {
+  try {
+    const connection = await getNatsConnection();
+    connection.publish(subject, codec.encode(message));
+  } catch (error) {
+    console.warn('[nats-core] publish failed', { subject, error });
+  }
+}
+
+export function subscribeNatsCoreSubject(
+  subject: string,
+  handler: NatsMessageHandler,
+): () => void {
+  const shared =
+    subscriptions.get(subject) ?? createSharedSubscription(subject);
+  shared.handlers.add(handler);
+
+  return () => {
+    const current = subscriptions.get(subject);
+    if (!current) return;
+    current.handlers.delete(handler);
+    if (current.handlers.size > 0) return;
+    subscriptions.delete(subject);
+    current.closed = true;
+    current.subscription?.unsubscribe();
+  };
+}
@@ -4,23 +4,24 @@ import type {
 } from '@shared/contracts/resources';
 import { ResourceChangeSchema } from '@shared/contracts/resources';
 import {
-  createPubSubEnvelope,
-  parsePubSubEnvelope,
-} from './pubSubEnvelope';
-import { publishRedisMessage, subscribeRedisChannel } from './redisPubSub';
+  encodeNatsSubjectToken,
+  publishNatsCoreMessage,
+  subscribeNatsCoreSubject,
+} from './natsCorePubSub';
+import { createPubSubEnvelope, parsePubSubEnvelope } from './pubSubEnvelope';

 type Listener = (changes: ResourceChange[]) => void;

-const RESOURCE_CHANNEL_PREFIX = 'resource-state:';
+const RESOURCE_SUBJECT_PREFIX = 'daoyou.realtime.resource-state';
 const listeners = new Map<string, Set<Listener>>();
-const redisSubscriptions = new Map<string, () => void>();
+const natsSubscriptions = new Map<string, () => void>();

 function scopeKey(scope: ResourceScope): string {
  return `${scope.kind}:${scope.id}`;
 }

-function channelForScope(scope: ResourceScope): string {
-  return `${RESOURCE_CHANNEL_PREFIX}${scopeKey(scope)}`;
+function subjectForScope(scope: ResourceScope): string {
+  return `${RESOURCE_SUBJECT_PREFIX}.${scope.kind}.${encodeNatsSubjectToken(scope.id)}`;
 }

 function isResourceChanges(value: unknown): value is ResourceChange[] {
@@ -32,14 +33,17 @@ function parseEvents(raw: string): ResourceChange[] {
  return changes ? ResourceChangeSchema.array().parse(changes) : [];
 }

-function ensureRedisSubscription(scope: ResourceScope): void {
+function ensureNatsSubscription(scope: ResourceScope): void {
  const key = scopeKey(scope);
-  if (redisSubscriptions.has(key)) return;
-  const unsubscribe = subscribeRedisChannel(channelForScope(scope), (raw) => {
-    const events = parseEvents(raw);
-    if (events.length > 0) notifyLocalListeners(scope, events);
-  });
-  redisSubscriptions.set(key, unsubscribe);
+  if (natsSubscriptions.has(key)) return;
+  const unsubscribe = subscribeNatsCoreSubject(
+    subjectForScope(scope),
+    (raw) => {
+      const events = parseEvents(raw);
+      if (events.length > 0) notifyLocalListeners(scope, events);
+    },
+  );
+  natsSubscriptions.set(key, unsubscribe);
 }

 function notifyLocalListeners(
@@ -60,13 +64,13 @@ export function subscribeResourceEvents(
    const set = listeners.get(key) ?? new Set<Listener>();
    set.add(listener);
    listeners.set(key, set);
-    ensureRedisSubscription(scope);
+    ensureNatsSubscription(scope);
    return () => {
      set.delete(listener);
      if (set.size === 0) {
        listeners.delete(key);
-        redisSubscriptions.get(key)?.();
-        redisSubscriptions.delete(key);
+        natsSubscriptions.get(key)?.();
+        natsSubscriptions.delete(key);
      }
    };
  });
@@ -88,8 +92,8 @@ export function publishResourceEvents(events: ResourceChange[]): void {
  }
  for (const group of grouped.values()) {
    notifyLocalListeners(group.scope, group.events);
-    void publishRedisMessage(
-      channelForScope(group.scope),
+    void publishNatsCoreMessage(
+      subjectForScope(group.scope),
      JSON.stringify(createPubSubEnvelope(group.events)),
    );
  }
@@ -1,152 +0,0 @@
-import { getRedisClient } from '@server/lib/redis';
-
-type RedisMessageHandler = (message: string) => void;
-type RedisReconnectTimer = ReturnType<typeof setTimeout>;
-
-type ChannelSubscription = {
-  client: ReturnType<typeof getRedisClient>;
-  handlers: Set<RedisMessageHandler>;
-  ready: Promise<void>;
-  onMessage: (channel: string, message: string) => void;
-  reconnectAttempts: number;
-  reconnectTimer: RedisReconnectTimer | null;
-  closed: boolean;
-};
-
-const subscriptions = new Map<string, ChannelSubscription>();
-const REDIS_RECONNECT_DELAYS_MS = [1_000, 2_000, 5_000, 10_000, 30_000] as const;
-
-function isRedisConfigured(): boolean {
-  return Boolean(process.env.REDIS_URL);
-}
-
-function createSubscription(channel: string): ChannelSubscription {
-  const client = getRedisClient().duplicate();
-  const subscription: ChannelSubscription = {
-    client,
-    handlers: new Set(),
-    ready: Promise.resolve(),
-    onMessage(receivedChannel, message) {
-      if (receivedChannel !== channel) {
-        return;
-      }
-
-      for (const handler of subscription.handlers) {
-        handler(message);
-      }
-    },
-    reconnectAttempts: 0,
-    reconnectTimer: null,
-    closed: false,
-  };
-
-  client.on('message', subscription.onMessage);
-  client.on('error', (error) => {
-    console.warn('[redis-pubsub] subscriber error', { channel, error });
-  });
-
-  subscription.ready = client
-    .subscribe(channel)
-    .then(() => {
-      subscription.reconnectAttempts = 0;
-    })
-    .catch((error) => {
-      if (subscription.closed) {
-        return;
-      }
-      client.off('message', subscription.onMessage);
-      client.disconnect(false);
-      scheduleReconnect(channel, subscription);
-      throw error;
-    });
-
-  subscriptions.set(channel, subscription);
-  return subscription;
-}
-
-function scheduleReconnect(channel: string, subscription: ChannelSubscription) {
-  if (
-    subscription.closed ||
-    subscription.handlers.size === 0 ||
-    subscription.reconnectTimer
-  ) {
-    return;
-  }
-
-  const delay =
-    REDIS_RECONNECT_DELAYS_MS[
-      Math.min(subscription.reconnectAttempts, REDIS_RECONNECT_DELAYS_MS.length - 1)
-    ];
-  subscription.reconnectAttempts += 1;
-  subscription.reconnectTimer = setTimeout(() => {
-    subscription.reconnectTimer = null;
-    if (subscription.closed || subscription.handlers.size === 0) {
-      return;
-    }
-
-    const next = createSubscription(channel);
-    for (const handler of subscription.handlers) {
-      next.handlers.add(handler);
-    }
-  }, delay);
-}
-
-export async function publishRedisMessage(
-  channel: string,
-  message: string,
-): Promise<void> {
-  if (!isRedisConfigured()) {
-    return;
-  }
-
-  try {
-    await getRedisClient().publish(channel, message);
-  } catch (error) {
-    console.warn('[redis-pubsub] publish failed', { channel, error });
-  }
-}
-
-export function subscribeRedisChannel(
-  channel: string,
-  handler: RedisMessageHandler,
-): () => void {
-  if (!isRedisConfigured()) {
-    return () => {};
-  }
-
-  let subscription = subscriptions.get(channel);
-  if (!subscription) {
-    try {
-      subscription = createSubscription(channel);
-    } catch (error) {
-      console.warn('[redis-pubsub] subscribe failed', { channel, error });
-      return () => {};
-    }
-  }
-
-  subscription.handlers.add(handler);
-  subscription.ready.catch((error) => {
-    console.warn('[redis-pubsub] subscribe failed', { channel, error });
-  });
-
-  return () => {
-    const current = subscriptions.get(channel);
-    if (!current) {
-      return;
-    }
-
-    current.handlers.delete(handler);
-    if (current.handlers.size > 0) {
-      return;
-    }
-
-    subscriptions.delete(channel);
-    current.closed = true;
-    if (current.reconnectTimer) {
-      clearTimeout(current.reconnectTimer);
-    }
-    current.client.off('message', current.onMessage);
-    current.client.unsubscribe(channel).catch(() => undefined);
-    current.client.disconnect(false);
-  };
-}
@@ -1,32 +1,33 @@
+import {
+  encodeNatsSubjectToken,
+  publishNatsCoreMessage,
+  subscribeNatsCoreSubject,
+} from '@server/lib/services/natsCorePubSub';
 import {
  createPubSubEnvelope,
  parsePubSubEnvelope,
 } from '@server/lib/services/pubSubEnvelope';
-import {
-  publishRedisMessage,
-  subscribeRedisChannel,
-} from '@server/lib/services/redisPubSub';
 import type { WorldChatMessageDTO } from '@shared/types/world-chat';

 type Listener = (message: WorldChatMessageDTO) => void;
 type SectSubscription = {
  listeners: Set<Listener>;
-  unsubscribeRedis: () => void;
+  unsubscribeNats: () => void;
 };

 const subscriptions = new Map<string, SectSubscription>();

-function channelForSect(sectId: string) {
-  return `sect-chat:${sectId}`;
+function subjectForSect(sectId: string) {
+  return `daoyou.realtime.sect-chat.${encodeNatsSubjectToken(sectId)}`;
 }

 function isSectChatMessage(value: unknown): value is WorldChatMessageDTO {
  return Boolean(
    value &&
-      typeof value === 'object' &&
-      (value as { channel?: unknown }).channel === 'sect' &&
-      typeof (value as { id?: unknown }).id === 'string' &&
-      typeof (value as { sectId?: unknown }).sectId === 'string',
+    typeof value === 'object' &&
+    (value as { channel?: unknown }).channel === 'sect' &&
+    typeof (value as { id?: unknown }).id === 'string' &&
+    typeof (value as { sectId?: unknown }).sectId === 'string',
  );
 }

@@ -45,12 +46,15 @@ export function subscribeSectChatMessages(
    const listeners = new Set<Listener>();
    const created: SectSubscription = {
      listeners,
-      unsubscribeRedis: subscribeRedisChannel(channelForSect(sectId), (raw) => {
-        const message = parsePubSubEnvelope(raw, isSectChatMessage);
-        if (message?.sectId === sectId) {
-          notify(created, message);
-        }
-      }),
+      unsubscribeNats: subscribeNatsCoreSubject(
+        subjectForSect(sectId),
+        (raw) => {
+          const message = parsePubSubEnvelope(raw, isSectChatMessage);
+          if (message?.sectId === sectId) {
+            notify(created, message);
+          }
+        },
+      ),
    };
    subscription = created;
    subscriptions.set(sectId, created);
@@ -62,7 +66,7 @@ export function subscribeSectChatMessages(
    if (!current) return;
    current.listeners.delete(listener);
    if (current.listeners.size > 0) return;
-    current.unsubscribeRedis();
+    current.unsubscribeNats();
    subscriptions.delete(sectId);
  };
 }
@@ -75,8 +79,8 @@ export function publishSectChatMessage(
  if (subscription) {
    notify(subscription, message);
  }
-  void publishRedisMessage(
-    channelForSect(sectId),
+  void publishNatsCoreMessage(
+    subjectForSect(sectId),
    JSON.stringify(createPubSubEnvelope(message)),
  );
 }
@@ -1,4 +1,7 @@
-import { publishRedisMessage, subscribeRedisChannel } from '@server/lib/services/redisPubSub';
+import {
+  publishNatsCoreMessage,
+  subscribeNatsCoreSubject,
+} from '@server/lib/services/natsCorePubSub';
 import {
  createPubSubEnvelope,
  parsePubSubEnvelope,
@@ -7,24 +10,28 @@ import type { WorldChatMessageDTO } from '@shared/types/world-chat';

 type Listener = (message: WorldChatMessageDTO) => void;

-const WORLD_CHAT_CHANNEL = 'world-chat';
+const WORLD_CHAT_SUBJECT = 'daoyou.realtime.world-chat';
 const localListeners = new Set<Listener>();
-let unsubscribeRedis: (() => void) | null = null;
+let unsubscribeNats: (() => void) | null = null;

 function isWorldChatMessage(value: unknown): value is WorldChatMessageDTO {
-  return Boolean(value && typeof value === 'object' && typeof (value as { id?: unknown }).id === 'string');
+  return Boolean(
+    value &&
+    typeof value === 'object' &&
+    typeof (value as { id?: unknown }).id === 'string',
+  );
 }

 function parseMessage(raw: string): WorldChatMessageDTO | null {
  return parsePubSubEnvelope(raw, isWorldChatMessage);
 }

-function ensureRedisSubscription() {
-  if (unsubscribeRedis) {
+function ensureNatsSubscription() {
+  if (unsubscribeNats) {
    return;
  }

-  unsubscribeRedis = subscribeRedisChannel(WORLD_CHAT_CHANNEL, (raw) => {
+  unsubscribeNats = subscribeNatsCoreSubject(WORLD_CHAT_SUBJECT, (raw) => {
    const message = parseMessage(raw);
    if (message) {
      notifyLocalWorldChatListeners(message);
@@ -40,21 +47,21 @@ function notifyLocalWorldChatListeners(message: WorldChatMessageDTO) {

 export function subscribeWorldChatMessages(listener: Listener): () => void {
  localListeners.add(listener);
-  ensureRedisSubscription();
+  ensureNatsSubscription();

  return () => {
    localListeners.delete(listener);
    if (localListeners.size === 0) {
-      unsubscribeRedis?.();
-      unsubscribeRedis = null;
+      unsubscribeNats?.();
+      unsubscribeNats = null;
    }
  };
 }

 export function publishWorldChatMessage(message: WorldChatMessageDTO): void {
  notifyLocalWorldChatListeners(message);
-  void publishRedisMessage(
-    WORLD_CHAT_CHANNEL,
+  void publishNatsCoreMessage(
+    WORLD_CHAT_SUBJECT,
    JSON.stringify(createPubSubEnvelope(message)),
  );
 }
@@ -0,0 +1,55 @@
+import {
+  BACKGROUND_COMMAND_DEFINITIONS,
+  BACKGROUND_COMMAND_TYPES,
+  parseBackgroundCommandEnvelope,
+} from './backgroundCommands';
+
+describe('background command contracts', () => {
+  it('parses a versioned scheduled command', () => {
+    const type = 'market.refresh';
+    const definition = BACKGROUND_COMMAND_DEFINITIONS[type];
+    const command = parseBackgroundCommandEnvelope({
+      id: '11111111-1111-4111-8111-111111111111',
+      type,
+      version: definition.version,
+      subject: definition.subject,
+      requestedAt: '2026-08-03T08:01:00.000Z',
+      scheduleBucketStartedAt: '2026-08-03T08:00:00.000Z',
+      deduplicationKey: 'market.refresh:2026-08-03T08:00:00.000Z',
+    });
+
+    expect(command.type).toBe(type);
+  });
+
+  it.each(BACKGROUND_COMMAND_TYPES)(
+    'keeps %s definition and envelope aligned',
+    (type) => {
+      const definition = BACKGROUND_COMMAND_DEFINITIONS[type];
+      const command = parseBackgroundCommandEnvelope({
+        id: '11111111-1111-4111-8111-111111111111',
+        type,
+        version: definition.version,
+        subject: definition.subject,
+        requestedAt: '2026-08-03T08:01:00.000Z',
+        scheduleBucketStartedAt: '2026-08-03T08:00:00.000Z',
+        deduplicationKey: `${type}:2026-08-03T08:00:00.000Z`,
+      });
+
+      expect(command.subject).toBe(definition.subject);
+    },
+  );
+
+  it('rejects a command sent to another subject', () => {
+    expect(() =>
+      parseBackgroundCommandEnvelope({
+        id: '11111111-1111-4111-8111-111111111111',
+        type: 'auction.expire',
+        version: 1,
+        subject: 'daoyou.command.cron.wrong.v1',
+        requestedAt: '2026-08-03T08:01:00.000Z',
+        scheduleBucketStartedAt: '2026-08-03T08:00:00.000Z',
+        deduplicationKey: 'auction.expire:2026-08-03T08:00:00.000Z',
+      }),
+    ).toThrow('后台命令定义不匹配');
+  });
+});
@@ -0,0 +1,95 @@
+import { z } from 'zod';
+
+export const BACKGROUND_COMMAND_STREAM = 'DAOYOU_BACKGROUND_COMMANDS';
+export const BACKGROUND_COMMAND_SUBJECT_PREFIX = 'daoyou.command.cron';
+
+export const BACKGROUND_COMMAND_TYPES = [
+  'auction.expire',
+  'bet-battle.expire',
+  'ranking.rewards.distribute',
+  'market.refresh',
+  'tower.enemy-sets.refresh',
+  'resource-replay.cleanup',
+  'expired-data.cleanup',
+  'material-library.generate',
+] as const;
+
+export type BackgroundCommandType = (typeof BACKGROUND_COMMAND_TYPES)[number];
+
+export const BACKGROUND_COMMAND_DEFINITIONS = {
+  'auction.expire': {
+    version: 1,
+    subject: `${BACKGROUND_COMMAND_SUBJECT_PREFIX}.auction-expire.v1`,
+    scheduleBucketMs: 2 * 60_000,
+  },
+  'bet-battle.expire': {
+    version: 1,
+    subject: `${BACKGROUND_COMMAND_SUBJECT_PREFIX}.bet-battle-expire.v1`,
+    scheduleBucketMs: 2 * 60_000,
+  },
+  'ranking.rewards.distribute': {
+    version: 1,
+    subject: `${BACKGROUND_COMMAND_SUBJECT_PREFIX}.ranking-rewards-distribute.v1`,
+    scheduleBucketMs: 24 * 60 * 60_000,
+  },
+  'market.refresh': {
+    version: 1,
+    subject: `${BACKGROUND_COMMAND_SUBJECT_PREFIX}.market-refresh.v1`,
+    scheduleBucketMs: 5 * 60_000,
+  },
+  'tower.enemy-sets.refresh': {
+    version: 1,
+    subject: `${BACKGROUND_COMMAND_SUBJECT_PREFIX}.tower-enemy-sets-refresh.v1`,
+    scheduleBucketMs: 60 * 60_000,
+  },
+  'resource-replay.cleanup': {
+    version: 1,
+    subject: `${BACKGROUND_COMMAND_SUBJECT_PREFIX}.resource-replay-cleanup.v1`,
+    scheduleBucketMs: 24 * 60 * 60_000,
+  },
+  'expired-data.cleanup': {
+    version: 1,
+    subject: `${BACKGROUND_COMMAND_SUBJECT_PREFIX}.expired-data-cleanup.v1`,
+    scheduleBucketMs: 24 * 60 * 60_000,
+  },
+  'material-library.generate': {
+    version: 1,
+    subject: `${BACKGROUND_COMMAND_SUBJECT_PREFIX}.material-library-generate.v1`,
+    scheduleBucketMs: 24 * 60 * 60_000,
+  },
+} as const satisfies Record<
+  BackgroundCommandType,
+  { version: number; subject: string; scheduleBucketMs: number }
+>;
+
+const BackgroundCommandEnvelopeSchema = z
+  .object({
+    id: z.uuid(),
+    type: z.enum(BACKGROUND_COMMAND_TYPES),
+    version: z.number().int().positive(),
+    subject: z.string().min(1).max(160),
+    requestedAt: z.string().datetime(),
+    scheduleBucketStartedAt: z.string().datetime(),
+    deduplicationKey: z.string().min(1).max(256),
+  })
+  .strict();
+
+export type BackgroundCommandEnvelope = z.infer<
+  typeof BackgroundCommandEnvelopeSchema
+>;
+
+export function parseBackgroundCommandEnvelope(
+  input: unknown,
+): BackgroundCommandEnvelope {
+  const command = BackgroundCommandEnvelopeSchema.parse(input);
+  const definition = BACKGROUND_COMMAND_DEFINITIONS[command.type];
+  if (
+    command.version !== definition.version ||
+    command.subject !== definition.subject
+  ) {
+    throw new Error(
+      `后台命令定义不匹配: ${command.type}@v${command.version} subject=${command.subject}`,
+    );
+  }
+  return command;
+}
@@ -5,6 +5,7 @@ import {

 const EVENT_ID = '11111111-1111-4111-8111-111111111111';
 const CULTIVATOR_ID = '22222222-2222-4222-8222-222222222222';
+const USER_ID = '55555555-5555-4555-8555-555555555555';

 describe('domain event contracts', () => {
  it('parses a versioned event envelope', () => {
@@ -63,4 +64,107 @@ describe('domain event contracts', () => {
      }),
    ).toThrow();
  });
+
+  it.each([
+    {
+      type: 'yield.claimed' as const,
+      data: {
+        cultivatorId: CULTIVATOR_ID,
+        actionInstanceId: '33333333-3333-4333-8333-333333333333',
+        realm: '筑基',
+        materialCount: 2,
+      },
+    },
+    {
+      type: 'cultivator.realm.changed' as const,
+      data: {
+        userId: USER_ID,
+        cultivatorId: CULTIVATOR_ID,
+        actionInstanceId: '33333333-3333-4333-8333-333333333333',
+        cultivatorName: '玄真',
+        fromRealm: '炼气',
+        fromStage: '圆满',
+        toRealm: '筑基',
+        toStage: '初期',
+        major: true,
+      },
+    },
+    {
+      type: 'mail.created' as const,
+      data: {
+        mailId: '66666666-6666-4666-8666-666666666666',
+        cultivatorId: CULTIVATOR_ID,
+        mailType: 'reward',
+        attachmentCount: 1,
+      },
+    },
+    {
+      type: 'craft.item.created' as const,
+      data: {
+        userId: USER_ID,
+        cultivatorId: CULTIVATOR_ID,
+        cultivatorName: '玄真',
+        itemType: 'artifact',
+        itemId: '77777777-7777-4777-8777-777777777777',
+        itemName: '玄光剑',
+        quality: '天品',
+        snapshot: { id: '77777777-7777-4777-8777-777777777777' },
+      },
+    },
+    {
+      type: 'market.material.revealed' as const,
+      data: {
+        userId: USER_ID,
+        cultivatorId: CULTIVATOR_ID,
+        cultivatorName: '玄真',
+        materialId: '77777777-7777-4777-8777-777777777777',
+        materialName: '星辰砂',
+        quality: '天品',
+        snapshot: { id: '77777777-7777-4777-8777-777777777777' },
+      },
+    },
+    {
+      type: 'bet-battle.created' as const,
+      data: {
+        userId: USER_ID,
+        cultivatorId: CULTIVATOR_ID,
+        cultivatorName: '玄真',
+        battleId: '88888888-8888-4888-8888-888888888888',
+        taunt: '可敢一战',
+      },
+    },
+    {
+      type: 'bet-battle.settled' as const,
+      data: {
+        userId: USER_ID,
+        cultivatorId: CULTIVATOR_ID,
+        battleId: '88888888-8888-4888-8888-888888888888',
+        rumor: '玄真于赌战台取胜。',
+      },
+    },
+    {
+      type: 'ranking.position.changed' as const,
+      data: {
+        userId: USER_ID,
+        cultivatorId: CULTIVATOR_ID,
+        challengerName: '玄真',
+        targetName: '赤霄',
+        realm: '筑基',
+        rank: 3,
+        changeType: 'challenge_win',
+      },
+    },
+  ])('parses $type integration event data', ({ type, data }) => {
+    const definition = DOMAIN_EVENT_DEFINITIONS[type];
+    const event = parseDomainEventEnvelope({
+      id: EVENT_ID,
+      type,
+      version: definition.version,
+      subject: definition.subject,
+      occurredAt: '2026-08-03T08:00:00.000Z',
+      aggregate: { type: 'cultivator', id: CULTIVATOR_ID },
+      data,
+    });
+    expect(event.type).toBe(type);
+  });
 });
@@ -1,3 +1,8 @@
+import {
+  QUALITY_VALUES,
+  REALM_STAGE_VALUES,
+  REALM_VALUES,
+} from '@shared/types/constants';
 import { ALCHEMY_MODE_VALUES } from '@shared/types/consumable';
 import { z } from 'zod';

@@ -9,6 +14,14 @@ export const DOMAIN_EVENT_TYPES = [
  'alchemy.craft.completed',
  'ranking.challenge.completed',
  'dungeon.run.settled',
+  'yield.claimed',
+  'cultivator.realm.changed',
+  'mail.created',
+  'craft.item.created',
+  'market.material.revealed',
+  'bet-battle.created',
+  'bet-battle.settled',
+  'ranking.position.changed',
 ] as const;

 export type DomainEventType = (typeof DOMAIN_EVENT_TYPES)[number];
@@ -53,6 +66,86 @@ export const DomainEventDataSchemas = {
      ]),
    })
    .strict(),
+  'yield.claimed': z
+    .object({
+      cultivatorId: z.uuid(),
+      actionInstanceId: z.uuid(),
+      realm: z.enum(REALM_VALUES),
+      materialCount: z.number().int().positive().max(100),
+    })
+    .strict(),
+  'cultivator.realm.changed': z
+    .object({
+      userId: z.uuid(),
+      cultivatorId: z.uuid(),
+      actionInstanceId: z.uuid(),
+      cultivatorName: z.string().min(1).max(100),
+      fromRealm: z.enum(REALM_VALUES),
+      fromStage: z.enum(REALM_STAGE_VALUES),
+      toRealm: z.enum(REALM_VALUES),
+      toStage: z.enum(REALM_STAGE_VALUES),
+      major: z.boolean(),
+    })
+    .strict(),
+  'mail.created': z
+    .object({
+      mailId: z.uuid(),
+      cultivatorId: z.uuid(),
+      mailType: z.enum(['system', 'reward']),
+      attachmentCount: z.number().int().nonnegative().max(100),
+    })
+    .strict(),
+  'craft.item.created': z
+    .object({
+      userId: z.uuid(),
+      cultivatorId: z.uuid(),
+      cultivatorName: z.string().min(1).max(100),
+      itemType: z.enum(['artifact', 'skill', 'gongfa', 'consumable']),
+      itemId: z.uuid(),
+      itemName: z.string().min(1).max(200),
+      quality: z.enum(QUALITY_VALUES),
+      snapshot: z.record(z.string(), z.unknown()),
+    })
+    .strict(),
+  'market.material.revealed': z
+    .object({
+      userId: z.uuid(),
+      cultivatorId: z.uuid(),
+      cultivatorName: z.string().min(1).max(100),
+      materialId: z.uuid(),
+      materialName: z.string().min(1).max(200),
+      quality: z.enum(QUALITY_VALUES),
+      snapshot: z.record(z.string(), z.unknown()),
+    })
+    .strict(),
+  'bet-battle.created': z
+    .object({
+      userId: z.uuid(),
+      cultivatorId: z.uuid(),
+      cultivatorName: z.string().min(1).max(100),
+      battleId: z.uuid(),
+      taunt: z.string().min(1).max(500).optional(),
+    })
+    .strict(),
+  'bet-battle.settled': z
+    .object({
+      userId: z.uuid(),
+      cultivatorId: z.uuid(),
+      battleId: z.uuid(),
+      rumor: z.string().min(1).max(1_000),
+    })
+    .strict(),
+  'ranking.position.changed': z
+    .object({
+      userId: z.uuid(),
+      cultivatorId: z.uuid(),
+      challengerName: z.string().min(1).max(100),
+      targetName: z.string().min(1).max(100).optional(),
+      realm: z.enum(REALM_VALUES),
+      rank: z.number().int().positive(),
+      changeType: z.enum(['direct_entry', 'challenge_win', 'vacancy_entry']),
+    })
+    .strict(),
 } as const;

 export type DomainEventData<TType extends DomainEventType> = z.infer<
@@ -76,6 +169,38 @@ export const DOMAIN_EVENT_DEFINITIONS = {
    version: 1,
    subject: `${DOMAIN_EVENT_SUBJECT_PREFIX}.activity.dungeon-run-settled.v1`,
  },
+  'yield.claimed': {
+    version: 1,
+    subject: `${DOMAIN_EVENT_SUBJECT_PREFIX}.activity.yield-claimed.v1`,
+  },
+  'cultivator.realm.changed': {
+    version: 1,
+    subject: `${DOMAIN_EVENT_SUBJECT_PREFIX}.gameplay.cultivator-realm-changed.v1`,
+  },
+  'mail.created': {
+    version: 1,
+    subject: `${DOMAIN_EVENT_SUBJECT_PREFIX}.communication.mail-created.v1`,
+  },
+  'craft.item.created': {
+    version: 1,
+    subject: `${DOMAIN_EVENT_SUBJECT_PREFIX}.gameplay.craft-item-created.v1`,
+  },
+  'market.material.revealed': {
+    version: 1,
+    subject: `${DOMAIN_EVENT_SUBJECT_PREFIX}.gameplay.market-material-revealed.v1`,
+  },
+  'bet-battle.created': {
+    version: 1,
+    subject: `${DOMAIN_EVENT_SUBJECT_PREFIX}.gameplay.bet-battle-created.v1`,
+  },
+  'bet-battle.settled': {
+    version: 1,
+    subject: `${DOMAIN_EVENT_SUBJECT_PREFIX}.gameplay.bet-battle-settled.v1`,
+  },
+  'ranking.position.changed': {
+    version: 1,
+    subject: `${DOMAIN_EVENT_SUBJECT_PREFIX}.gameplay.ranking-position-changed.v1`,
+  },
 } as const satisfies Record<
  DomainEventType,
  { version: number; subject: string }
@@ -14,7 +14,6 @@ export type TaskDailyKind = 'alchemy' | 'dungeon' | 'ranking';
 export const TASK_EVENT_VALUES = [
  'alchemy_crafted',
  'dungeon_completed',
-  'ranking_challenge_battled',
 ] as const;

 export type TaskEvent = (typeof TASK_EVENT_VALUES)[number];