// ? 錯(cuò)誤示范
natsPublisher.publish(task); // 1. 消息發(fā)出，Worker 開始下載轉(zhuǎn)存
messageRepository.save(message); // 2. 數(shù)據(jù)庫(kù)報(bào)錯(cuò)（如字段超長(zhǎng)、唯一鍵沖突）
// 后果：DB 回滾，業(yè)務(wù)無(wú)記錄，但 OSS 中產(chǎn)生了一個(gè)永遠(yuǎn)無(wú)法被引用的“孤兒文件”。

// ? 錯(cuò)誤示范
@Transactional
public void handle() {
    messageRepository.save(message);
    natsPublisher.publish(task); 
    // 3. 代碼執(zhí)行完畢，但在事務(wù)提交（Commit）的一瞬間數(shù)據(jù)庫(kù)連接斷開
}
// 后果：Worker 收到任務(wù)并完成處理，但在回調(diào)更新狀態(tài)時(shí)發(fā)現(xiàn) DB 中不存在該記錄。

// 1. 準(zhǔn)備階段：預(yù)生成任務(wù)（純內(nèi)存操作，無(wú)副作用）
// 此時(shí)并沒(méi)有真正發(fā)送 NATS 消息，只是構(gòu)建了對(duì)象
List<MediaTransferTask> mediaTasks = prepareMediaTransferTasks(msg, ids.sessionId());
// 2. 構(gòu)建消息實(shí)體
WxMessage message = buildMessage(msg, accountId, ids.sessionId(), ids.senderId());
try {
    // 【核心步驟 A】數(shù)據(jù)庫(kù)落庫(kù) (Source of Truth)
    // 這是唯一的“事實(shí)來(lái)源”。如果這里失敗，后續(xù)一切都不應(yīng)發(fā)生。
    messageRepository.save(message);
    log.info("Message saved: id={}, wxid={}", message.getId(), message.getWxid());
    // 3. 發(fā)布會(huì)話更新事件 (內(nèi)存事件或 MQ)
    SessionUpdateEvent event = SessionUpdateEvent.builder()
        .accountId(accountId)
        // ... build params
        .build();
    sessionEventPublisher.publishSessionUpdate(event);
    // 【核心步驟 B】注冊(cè)事務(wù)回調(diào)
    // 關(guān)鍵點(diǎn)：這里不是立即發(fā)送，而是“預(yù)約”發(fā)送
    publishMediaTransferTasksAfterCommit(mediaTasks);
    return ProcessResult.success();
} catch (DataIntegrityViolationException e) {
    // 【并發(fā)場(chǎng)景的保護(hù)】
    // 如果兩個(gè)線程同時(shí)處理同一條消息（如網(wǎng)絡(luò)重放或客戶端重試），
    // 數(shù)據(jù)庫(kù)的唯一索引會(huì)拋出此異常。
    // 由于消息發(fā)送邏輯在事務(wù)提交后執(zhí)行，失敗的線程事務(wù)回滾，
    // 因此“afterCommit”鉤子不會(huì)被觸發(fā)，完美避免了 Worker 重復(fù)搬運(yùn)文件。
    log.debug("Duplicate message (concurrent): {}", msg.getMessageId());
    return ProcessResult.duplicate();
}

publishMediaTransferTasksAfterCommit 方法利用了 Spring 的 TransactionSynchronizationManager 來(lái)掛載回調(diào)。
private void publishMediaTransferTasksAfterCommit(List<MediaTransferTask> tasks) {
    if (CollectionUtils.isEmpty(tasks)) {
        return;
    }
    // 判斷當(dāng)前是否在事務(wù)中
    if (TransactionSynchronizationManager.isActualTransactionActive()) {
        // 注冊(cè)同步器
        TransactionSynchronizationManager.registerSynchronization(new TransactionSynchronization() {
            @Override
            public void afterCommit() {
                // 【真正發(fā)送的時(shí)機(jī)】
                // 只有當(dāng) DB 事務(wù)成功 Commit 后，這一行才會(huì)執(zhí)行
                // 此時(shí) DB 里一定有數(shù)據(jù)，Worker 回調(diào)一定能成功
                tasks.forEach(mediaTransferPublisher::publishMediaTransfer);
                log.debug("Async media tasks published after commit: size={}", tasks.size());
            }
        });
    } else {
        // 如果不在事務(wù)中（比如非事務(wù)方法調(diào)用），則立即發(fā)送（降級(jí)策略）
        tasks.forEach(mediaTransferPublisher::publishMediaTransfer);
    }
}