Version: 0.6.1.dev.260316

♻️ refactor(outbox): 抽离通用事件总线，并完成 event_type-only 收口 - ✨ 新增 `infra` 层通用 `EventBus` / `EventContract`，统一事件发布与消费协议 - 🔄 将聊天持久化链路调整为通过 `service/events` 注册 handler 并发布事件，进一步解耦业务逻辑与异步处理流程 - 🧹 移除 `chat_history_async` 旧适配实现，以及基于 `biz_type` 的兼容分发逻辑 - 📝 更新 Outbox 异步持久化决策记录，明确保留方案 A，并正式启用方案 B - 📚 同步更新 README 中关于 Outbox + Kafka 可靠异步链路的说明 - 🚚 当前 `outbox + kafka` 已与项目业务链路完全解耦，沉淀为通用、可靠性更强的消息队列能力；后续将参考消息队列的典型使用方式，逐步扩展到更多业务场景 - ✨ 补充跨不同分类事务管理器中的 `agent dao` 注册与接入支持
2026-03-16 13:00:26 +08:00
parent 712bcd3605
commit 626fc700d2
17 changed files with 782 additions and 422 deletions
--- a/backend/infra/outbox/engine.go
+++ b/backend/infra/outbox/engine.go
@@ -6,6 +6,8 @@ import (
 	"errors"
 	"fmt"
 	"log"
+	"strconv"
+	"strings"
 	"sync"
 	"time"

@@ -15,24 +17,35 @@ import (
 	"gorm.io/gorm"
 )

-// MessageHandler 是 outbox 消费分发处理器。
+// MessageHandler 是事件消费处理器。
 //
-// 设计约束：
-// 1) 入参 envelope 已经完成最外层解析（含 outbox_id、biz_type、payload）；
-// 2) 若返回 nil，表示业务处理成功，框架将继续提交 Kafka offset；
-// 3) 若返回 error，框架会按“可重试错误”处理：回写 outbox 失败状态并进入重试窗口。
+// 语义约束：
+// 1. 入参 envelope 已完成最外层解析；
+// 2. 返回 nil 表示处理成功，框架提交 offset；
+// 3. 返回 error 表示可重试失败，框架回写 retry 后提交 offset。
 type MessageHandler func(ctx context.Context, envelope kafkabus.Envelope) error

-// Engine 是 Outbox + Kafka 的通用异步引擎。
+// PublishRequest 是通用事件发布入参。
+//
+// 设计目标：
+// 1. 业务只描述“要发什么事件”，不关心 outbox/kafka 细节；
+// 2. 统一收敛事件元数据（event_type/version/aggregate_id）；
+// 3. payload 支持任意 DTO，由 infra 统一 JSON 序列化。
+type PublishRequest struct {
+	EventType    string
+	EventVersion string
+	MessageKey   string
+	AggregateID  string
+	EventID      string
+	Payload      any
+}
+
+// Engine 是 Outbox + Kafka 通用异步引擎。
 //
 // 职责边界：
-// 1) 负责 outbox 扫描、Kafka 投递、Kafka 消费与统一状态机流转；
-// 2) 负责 biz_type 到处理器的分发；
-// 3) 不关心具体业务含义（例如“聊天记录落库”），业务语义由 handler 提供。
-//
-// 状态流转口径：
-// pending -> published -> consumed（成功）；
-// pending/published --失败--> pending(带 next_retry_at) 或 dead（达到最大重试）。
+// 1. 负责 outbox 扫描、kafka 投递、kafka 消费、状态机推进；
+// 2. 负责 event_type -> handler 路由；
+// 3. 不负责任何业务语义（业务由 handler 承担）。
 type Engine struct {
 	repo     *Repository
 	producer *kafkabus.Producer
@@ -48,23 +61,19 @@ type Engine struct {
 	handlers   map[string]MessageHandler
 }

-// NewEngine 创建 outbox 异步引擎。
+// NewEngine 创建异步引擎。
 //
-// 说明：
-// 1) cfg.Enabled=false 时返回 nil，调用方可按“异步关闭”处理；
-// 2) producer/consumer 初始化失败时会确保资源回收，避免半初始化泄漏。
+// 规则：
+// 1. kafka.enabled=false 时返回 nil，调用方可降级同步；
+// 2. producer/consumer 任一步失败都会回收已创建资源。
 func NewEngine(repo *Repository, cfg kafkabus.Config) (*Engine, error) {
-	// 1. 配置关闭时直接返回 nil，让上层可以“无侵入降级为同步模式”。
 	if !cfg.Enabled {
 		return nil, nil
 	}
-	// 2. 仓储缺失属于启动期配置错误，直接返回。
 	if repo == nil {
 		return nil, errors.New("outbox repository is nil")
 	}

-	// 3. 先初始化 producer，再初始化 consumer。
-	//    如果第二步失败，要主动回收第一步资源，避免泄漏。
 	producer, err := kafkabus.NewProducer(cfg)
 	if err != nil {
 		return nil, err
@@ -75,7 +84,6 @@ func NewEngine(repo *Repository, cfg kafkabus.Config) (*Engine, error) {
 		return nil, err
 	}

-	// 4. 汇总配置，构造引擎实例。
 	return &Engine{
 		repo:      repo,
 		producer:  producer,
@@ -89,78 +97,62 @@ func NewEngine(repo *Repository, cfg kafkabus.Config) (*Engine, error) {
 	}, nil
 }

-// RegisterHandler 注册某个 biz_type 的消费处理器。
-//
-// 设计要求：
-// 1) biz_type 必须唯一，重复注册会覆盖旧值（并打印提示日志）；
-// 2) handler 不能为空；
-// 3) 建议在 Start 前完成注册，减少运行时热更新复杂度。
-func (e *Engine) RegisterHandler(bizType string, handler MessageHandler) error {
-	// 1. 参数校验：防止业务侧在启动链路上把 nil 引擎继续往下用。
+// RegisterHandler 是历史别名（等价 RegisterEventHandler）。
+func (e *Engine) RegisterHandler(eventType string, handler MessageHandler) error {
+	return e.RegisterEventHandler(eventType, handler)
+}
+
+// RegisterEventHandler 注册事件处理器。
+func (e *Engine) RegisterEventHandler(eventType string, handler MessageHandler) error {
 	if e == nil {
 		return errors.New("outbox engine is nil")
 	}
-	// 2. biz_type 为空会导致无法分发，提前拦截。
-	if bizType == "" {
-		return errors.New("bizType is empty")
+	eventType = strings.TrimSpace(eventType)
+	if eventType == "" {
+		return errors.New("eventType is empty")
 	}
-	// 3. handler 为空会在消费时 panic，必须提前拒绝。
 	if handler == nil {
 		return errors.New("handler is nil")
 	}

-	// 4. 加写锁更新 handler 映射，保证并发注册时 map 安全。
 	e.handlersMu.Lock()
 	defer e.handlersMu.Unlock()
-	if _, exists := e.handlers[bizType]; exists {
-		log.Printf("outbox handler 覆盖注册: biz_type=%s", bizType)
+	if _, exists := e.handlers[eventType]; exists {
+		log.Printf("outbox handler 覆盖注册: event_type=%s", eventType)
 	}
-	e.handlers[bizType] = handler
+	e.handlers[eventType] = handler
 	return nil
 }

-func (e *Engine) getHandler(bizType string) (MessageHandler, bool) {
-	// 读锁足够满足并发读取需求，避免无谓阻塞。
+func (e *Engine) getHandler(eventType string) (MessageHandler, bool) {
 	e.handlersMu.RLock()
 	defer e.handlersMu.RUnlock()
-	h, ok := e.handlers[bizType]
+	h, ok := e.handlers[eventType]
 	return h, ok
 }

-// Start 启动 outbox 异步引擎。
-//
-// 会启动两个后台循环：
-// 1) dispatch loop：扫描 due outbox 并投递到 Kafka；
-// 2) consume loop：消费 Kafka 并按 biz_type 分发处理。
+// Start 启动 dispatch + consume 两个后台循环。
 func (e *Engine) Start(ctx context.Context) {
 	if e == nil {
 		return
 	}

-	// 1. 启动日志：把关键运行参数打出来，便于排查“为什么没消费/没扫描”。
 	log.Printf("outbox engine starting: topic=%s brokers=%v retry_scan=%s batch=%d", e.topic, e.brokers, e.scanEvery, e.scanBatch)
-
-	// 2. 启动前探活 topic 是否可用。
-	//    注意：即使探活失败也不会阻断引擎启动，后续循环会继续重试。
 	if err := kafkabus.WaitTopicReady(ctx, e.brokers, e.topic, 30*time.Second); err != nil {
 		log.Printf("Kafka topic not ready before consume loop start: %v", err)
 	} else {
 		log.Printf("Kafka topic is ready: %s", e.topic)
 	}

-	// 3. 并行启动两条核心循环：
-	//    - dispatch loop：负责 outbox -> Kafka；
-	//    - consume loop：负责 Kafka -> handler -> outbox 状态推进。
 	go e.startDispatchLoop(ctx)
 	go e.startConsumeLoop(ctx)
 }

-// Close 关闭 producer/consumer 资源。
+// Close 关闭 kafka 资源。
 func (e *Engine) Close() {
 	if e == nil {
 		return
 	}
-	// 逐个关闭并记录错误，避免某个 close 失败导致后续资源无法回收。
 	if err := e.producer.Close(); err != nil {
 		log.Printf("关闭 Kafka producer 失败: %v", err)
 	}
@@ -169,35 +161,65 @@ func (e *Engine) Close() {
 	}
 }

-// Enqueue 把业务消息写入 outbox（请求路径调用）。
+// Enqueue 是历史别名（等价 Publish）。
+func (e *Engine) Enqueue(ctx context.Context, eventType, messageKey string, payload any) error {
+	return e.Publish(ctx, PublishRequest{
+		EventType:   eventType,
+		MessageKey:  messageKey,
+		AggregateID: messageKey,
+		Payload:     payload,
+	})
+}
+
+// Publish 发布事件到 outbox。
 //
-// 注意：
-// 1) 该方法不做 Kafka 网络写入，只有数据库写入；
-// 2) messageKey 建议使用业务幂等键（如 conversation_id）以提升分区稳定性；
-// 3) payload 需要可 JSON 序列化。
-func (e *Engine) Enqueue(ctx context.Context, bizType, messageKey string, payload any) error {
+// 步骤：
+// 1. 标准化 event_type/version/key；
+// 2. payload 序列化；
+// 3. 写入 outbox（仅本地写库，不做 kafka 网络 IO）。
+func (e *Engine) Publish(ctx context.Context, req PublishRequest) error {
 	if e == nil {
 		return errors.New("outbox engine is nil")
 	}
-	// 这里故意只写数据库，不做 Kafka 网络 IO，
-	// 目的是把请求耗时稳定在“单次写库”的可控范围。
-	_, err := e.repo.CreateMessage(ctx, bizType, e.topic, messageKey, payload, e.maxRetry)
+
+	eventType := strings.TrimSpace(req.EventType)
+	if eventType == "" {
+		return errors.New("eventType is empty")
+	}
+	eventVersion := strings.TrimSpace(req.EventVersion)
+	if eventVersion == "" {
+		eventVersion = DefaultEventVersion
+	}
+	messageKey := strings.TrimSpace(req.MessageKey)
+	aggregateID := strings.TrimSpace(req.AggregateID)
+	if aggregateID == "" {
+		aggregateID = messageKey
+	}
+
+	payloadJSON, err := json.Marshal(req.Payload)
+	if err != nil {
+		return err
+	}
+
+	_, err = e.repo.CreateMessage(ctx, eventType, e.topic, messageKey, OutboxEventPayload{
+		EventID:      strings.TrimSpace(req.EventID),
+		EventType:    eventType,
+		EventVersion: eventVersion,
+		AggregateID:  aggregateID,
+		Payload:      payloadJSON,
+	}, e.maxRetry)
 	return err
 }

 func (e *Engine) startDispatchLoop(ctx context.Context) {
-	// 1. 定时扫描 due outbox 记录。
-	//    扫描间隔由 scanEvery 控制，避免每次请求都主动触发投递造成抖动。
 	ticker := time.NewTicker(e.scanEvery)
 	defer ticker.Stop()

 	for {
 		select {
 		case <-ctx.Done():
-			// 2. 收到退出信号后优雅停止循环。
 			return
 		case <-ticker.C:
-			// 3. 拉取当前窗口内可投递消息。
 			pendingMessages, err := e.repo.ListDueMessages(ctx, e.scanBatch)
 			if err != nil {
 				log.Printf("扫描 outbox 失败: %v", err)
@@ -207,7 +229,6 @@ func (e *Engine) startDispatchLoop(ctx context.Context) {
 				log.Printf("outbox due messages=%d, start dispatch", len(pendingMessages))
 			}

-			// 4. 逐条投递，单条失败不影响同批后续消息。
 			for _, msg := range pendingMessages {
 				if err = e.dispatchOne(ctx, msg.ID); err != nil {
 					log.Printf("重试投递 outbox 消息失败(id=%d): %v", msg.ID, err)
@@ -218,29 +239,39 @@ func (e *Engine) startDispatchLoop(ctx context.Context) {
 }

 func (e *Engine) dispatchOne(ctx context.Context, outboxID int64) error {
-	// 1. 投递前重新按 ID 读取最新状态，避免用到过期快照。
 	outboxMsg, err := e.repo.GetByID(ctx, outboxID)
 	if err != nil {
 		if errors.Is(err, gorm.ErrRecordNotFound) {
-			// 1.1 记录已不存在（可能被清理），按幂等成功处理。
 			return nil
 		}
 		return err
 	}
-	// 1.2 最终态直接跳过，避免重复投递。
 	if outboxMsg.Status == model.OutboxStatusConsumed || outboxMsg.Status == model.OutboxStatusDead {
 		return nil
 	}

-	// 2. 组装 Kafka 包装体，统一带上 outbox_id 供消费端做状态回写。
+	eventPayload, payloadErr := parseOutboxEventPayload(outboxMsg.Payload)
+	if payloadErr != nil {
+		markErr := e.repo.MarkDead(ctx, outboxMsg.ID, "解析 outbox 事件包失败: "+payloadErr.Error())
+		if markErr != nil {
+			log.Printf("标记 outbox 死信失败(id=%d): %v", outboxMsg.ID, markErr)
+		}
+		return payloadErr
+	}
+	if eventPayload.EventID == "" {
+		eventPayload.EventID = strconv.FormatInt(outboxMsg.ID, 10)
+	}
+
 	envelope := kafkabus.Envelope{
-		OutboxID: outboxMsg.ID,
-		BizType:  outboxMsg.BizType,
-		Payload:  json.RawMessage(outboxMsg.Payload),
+		OutboxID:     outboxMsg.ID,
+		EventID:      eventPayload.EventID,
+		EventType:    eventPayload.EventType,
+		EventVersion: eventPayload.EventVersion,
+		AggregateID:  eventPayload.AggregateID,
+		Payload:      eventPayload.PayloadJSON,
 	}
 	raw, err := json.Marshal(envelope)
 	if err != nil {
-		// 2.1 包装层序列化失败通常不可恢复，直接标 dead。
 		markErr := e.repo.MarkDead(ctx, outboxMsg.ID, "序列化 outbox 包装失败: "+err.Error())
 		if markErr != nil {
 			log.Printf("标记 outbox 死信失败(id=%d): %v", outboxMsg.ID, markErr)
@@ -248,8 +279,6 @@ func (e *Engine) dispatchOne(ctx context.Context, outboxID int64) error {
 		return err
 	}

-	// 3. 先投 Kafka，再把 outbox 状态推进到 published。
-	//    任一步骤失败都回写 retry，让扫描器后续重试。
 	if err = e.producer.Enqueue(ctx, outboxMsg.Topic, outboxMsg.MessageKey, raw); err != nil {
 		_ = e.repo.MarkFailedForRetry(ctx, outboxMsg.ID, "投递 Kafka 失败: "+err.Error())
 		return err
@@ -262,29 +291,23 @@ func (e *Engine) dispatchOne(ctx context.Context, outboxID int64) error {
 }

 func (e *Engine) startConsumeLoop(ctx context.Context) {
-	// 消费循环采用“拉取 -> 处理 -> 提交 offset”的标准模型。
 	for {
 		select {
 		case <-ctx.Done():
-			// 1. 收到退出信号后终止循环。
 			return
 		default:
 		}

-		// 2. 拉取下一条 Kafka 消息。
 		msg, err := e.consumer.Dequeue(ctx)
 		if err != nil {
 			if errors.Is(err, context.Canceled) {
-				// 2.1 context 主动取消时，不记错误日志，直接退出。
 				return
 			}
-			// 2.2 临时错误短暂退避后继续，避免空转刷日志。
 			log.Printf("Kafka 消费拉取失败(topic=%s): %v", e.topic, err)
 			time.Sleep(300 * time.Millisecond)
 			continue
 		}

-		// 3. 单条消息处理失败仅记录日志，不阻断消费循环。
 		if err = e.handleMessage(ctx, msg); err != nil {
 			log.Printf("处理 Kafka 消息失败(topic=%s, partition=%d, offset=%d): %v", msg.Topic, msg.Partition, msg.Offset, err)
 		}
@@ -292,33 +315,35 @@ func (e *Engine) startConsumeLoop(ctx context.Context) {
 }

 func (e *Engine) handleMessage(ctx context.Context, msg segmentkafka.Message) error {
-	// 1. 先解析最外层 envelope，拿到 outbox_id + biz_type + payload。
 	var envelope kafkabus.Envelope
 	if err := json.Unmarshal(msg.Value, &envelope); err != nil {
-		// 1.1 包装层损坏时无法恢复，直接提交 offset 防止无限重放。
 		_ = e.consumer.Commit(ctx, msg)
 		return fmt.Errorf("解析 Kafka 包装失败: %w", err)
 	}
 	if envelope.OutboxID <= 0 {
-		// 1.2 缺少 outbox_id 无法回写状态，同样提交 offset 跳过。
 		_ = e.consumer.Commit(ctx, msg)
 		return errors.New("Kafka 包装缺少 outbox_id")
 	}

-	// 2. 根据 biz_type 查找业务处理器。
-	handler, ok := e.getHandler(envelope.BizType)
-	if !ok {
-		// 2.1 未注册处理器是配置错误，标记 dead 并提交 offset，避免重复消费。
-		_ = e.repo.MarkDead(ctx, envelope.OutboxID, "未知业务类型: "+envelope.BizType)
+	eventType := strings.TrimSpace(envelope.EventType)
+	if eventType == "" {
+		_ = e.repo.MarkDead(ctx, envelope.OutboxID, "消息缺少事件类型")
+		if err := e.consumer.Commit(ctx, msg); err != nil {
+			return err
+		}
+		return nil
+	}
+
+	handler, ok := e.getHandler(eventType)
+	if !ok {
+		_ = e.repo.MarkDead(ctx, envelope.OutboxID, "未知事件类型: "+eventType)
 		if err := e.consumer.Commit(ctx, msg); err != nil {
 			return err
 		}
 		return nil
 	}

-	// 3. 调用业务处理器。
 	if err := handler(ctx, envelope); err != nil {
-		// 统一按“可重试错误”处理，回写 retry 状态后提交 offset，避免同一条消息在 Kafka 侧死循环。
 		if markErr := e.repo.MarkFailedForRetry(ctx, envelope.OutboxID, "消费处理失败: "+err.Error()); markErr != nil {
 			return markErr
 		}
@@ -328,6 +353,5 @@ func (e *Engine) handleMessage(ctx context.Context, msg segmentkafka.Message) er
 		return err
 	}

-	// 4. 业务处理成功后提交 offset。
 	return e.consumer.Commit(ctx, msg)
 }