事件驱动模块实现教程（Kafka）

package biz

import "context"

// EventPublisher 定义业务层发布异步事件的接口，由 event 包实现。
// 通过接口注入避免 biz 与 event 包之间的循环依赖。
type EventPublisher interface {
	Publish(ctx context.Context, eventType string, payload interface{}) error
}

// 异步事件类型常量（共 7 种，覆盖文件 / 分享 / 回收站 / 上传四大业务域）。
const (
	EventFileDeleted    = "file.deleted"
	EventFileMoved      = "file.moved"
	EventFileTrashed    = "file.trashed"
	EventShareCreated   = "share.created"
	EventShareCancelled = "share.cancelled"
	EventRecycleClean   = "recycle.clean"
	EventUploadCompleted = "upload.completed"
)

// FileChangedPayload 文件变更事件负载
type FileChangedPayload struct {
	UserID    uint64   `json:"user_id"`
	FileIDs   []uint64 `json:"file_ids,omitempty"`
	FolderIDs []uint64 `json:"folder_ids,omitempty"`
	Action    string   `json:"action"`
	ParentID  *uint64  `json:"parent_id,omitempty"`
}

// SharePayload 分享事件负载
type SharePayload struct {
	UserID  uint64 `json:"user_id"`
	ShareID uint64 `json:"share_id"`
	UUID    string `json:"uuid"`
	Action  string `json:"action"`
}

// RecycleCleanPayload 回收站清理事件负载
type RecycleCleanPayload struct {
	Count int `json:"count"`
}

// UploadCompletedPayload 上传完成事件负载
type UploadCompletedPayload struct {
	UserID   uint64 `json:"user_id"`
	FileID   uint64 `json:"file_id"`
	FileName string `json:"file_name"`
	FileSize int64  `json:"file_size"`
	Hash     string `json:"hash,omitempty"`
}

package event

import "time"

const (
	SourceFileUC    = "file_uc"
	SourceShareUC   = "share_uc"
	SourceRecycleUC = "recycle_uc"
	SourceUploadUC  = "upload_uc"
	SourceSystem    = "system"
)

// Event 是所有异步事件的统一消息格式。
type Event struct {
	// ID 事件唯一标识，由生产者自动生成（evt_ + 16字节随机hex）
	ID string `json:"id"`
	// IdempotencyKey 唯一消息ID（默认 "类型:事件ID"），用于消费端去重（Redis SetNX）
	IdempotencyKey string `json:"idempotency_key"`
	// PartitionKey 修正点：业务实体键（如文件ID/分享ID），专用于 Kafka 分区路由与保序。
	// 与 IdempotencyKey 解耦：前者控"顺序"，后者控"去重"。
	PartitionKey string `json:"partition_key,omitempty"`
	Type         string `json:"type"`
	Source       string `json:"source"`
	Timestamp    string `json:"timestamp"`
	Payload      interface{} `json:"payload"`
}

// NewEvent 创建一个新的事件，自动填充时间戳。
func NewEvent(eventType, source string, payload interface{}) *Event {
	return &Event{
		Type:      eventType,
		Source:    source,
		Timestamp: time.Now().UTC().Format(time.RFC3339),
		Payload:   payload,
	}
}

package event

import (
	"cloud-disk/internal/conf"
	"context"
	"crypto/rand"
	"encoding/hex"
	"encoding/json"
	"fmt"
	"time"

	"github.com/go-kratos/kratos/v3/log"
	"github.com/segmentio/kafka-go"
)

type Producer interface {
	Publish(ctx context.Context, topic string, event *Event) error
	Close() error
}

type kafkaProducer struct {
	writer *kafka.Writer
}

func NewProducer(cfg *conf.Data_Kafka, writer *kafka.Writer) Producer {
	if cfg == nil || len(cfg.Addrs) == 0 {
		return noopProducer{}
	}
	return &kafkaProducer{writer: writer}
}

// Publish 序列化事件并发送到 Kafka，支持自动重试。
func (p *kafkaProducer) Publish(ctx context.Context, topic string, event *Event) error {
	// 步骤 1：唯一消息 ID + 幂等键（用于去重，与分区无关）
	event.ID = generateEventID()
	if event.IdempotencyKey == "" {
		event.IdempotencyKey = fmt.Sprintf("%s:%s", event.Type, event.ID)
	}
	if event.Timestamp == "" {
		event.Timestamp = time.Now().UTC().Format(time.RFC3339)
	}

	data, err := json.Marshal(event)
	if err != nil {
		return fmt.Errorf("event: failed to marshal event: %w", err)
	}

	// 修正点：Kafka Key 用 PartitionKey（保序），缺省时回退到幂等键，避免随机打散
	key := event.IdempotencyKey
	if event.PartitionKey != "" {
		key = event.PartitionKey
	}

	msg := kafka.Message{
		Topic: topic,
		Key:   []byte(key), // 同一业务实体 → 同一分区 → 单分区内有序
		Value: data,
		Headers: []kafka.Header{
			{Key: "event_type", Value: []byte(event.Type)},
			{Key: "event_source", Value: []byte(event.Source)},
		},
	}

	// 步骤 2：带退避的重试（共 3 次），且可被 ctx 中断
	var lastErr error
	for i := 0; i < 3; i++ {
		select {
		case <-ctx.Done():
			return ctx.Err()
		default:
		}
		if err := p.writer.WriteMessages(ctx, msg); err != nil {
			lastErr = err
			log.Warn("event: publish failed, retrying",
				"topic", topic, "type", event.Type, "attempt", i+1, "err", err)
			select {
			case <-ctx.Done():
				return ctx.Err()
			case <-time.After(time.Duration(100*(i+1)) * time.Millisecond):
			}
			continue
		}
		return nil
	}
	return fmt.Errorf("event: failed to publish after 3 retries: %w", lastErr)
}

func generateEventID() string {
	b := make([]byte, 16)
	if _, err := rand.Read(b); err != nil {
		return fmt.Sprintf("evt_%d", time.Now().UnixNano())
	}
	return "evt_" + hex.EncodeToString(b)
}

// businessKeyOf 根据事件类型与负载推导分区键（业务实体），用于保序。
// 修正点：原实现没有这一层，导致 Kafka Key 是随机消息ID，无法按实体保序。
func businessKeyOf(eventType string, payload interface{}) string {
	switch eventType {
	case biz.EventFileDeleted, biz.EventFileMoved, biz.EventFileTrashed:
		if p, ok := payload.(*biz.FileChangedPayload); ok && len(p.FileIDs) > 0 {
			return fmt.Sprintf("file:%d", p.FileIDs[0]) // 同一文件的多事件稳定同分区
		}
	case biz.EventShareCreated, biz.EventShareCancelled:
		if p, ok := payload.(*biz.SharePayload); ok {
			return fmt.Sprintf("share:%d", p.ShareID)
		}
	case biz.EventUploadCompleted:
		if p, ok := payload.(*biz.UploadCompletedPayload); ok {
			return fmt.Sprintf("file:%d", p.FileID)
		}
	}
	return "" // 空 → 生产者回退到幂等键（顺序性不保证，但可用）
}

func (a *eventPublisherAdapter) Publish(ctx context.Context, eventType string, payload interface{}) error {
	topic, ok := a.topics[eventType]
	if !ok {
		return nil
	}
	evt := NewEvent(eventType, SourceSystem, payload)
	evt.PartitionKey = businessKeyOf(eventType, payload) // 修正点：注入分区键
	// 修正点：不再吞错，错误向上返回（或交给 Outbox 兜底，见 5.6）
	return a.producer.Publish(ctx, topic, evt)
}

writer := &kafka.Writer{
	Addr:                   kafka.TCP(addrs...),
	Balancer:               &kafka.Hash{},          // 按 Key 哈希分区（保序需要）
	RequiredAcks:           kafka.RequireAll,       // 修正点：多数副本确认，防丢
	AllowAutoTopicCreation: false,                  // 修正点：生产禁用自动建Topic，由运维预建
	BatchTimeout:           5 * time.Millisecond,   // 优化：批量刷新
	BatchSize:              100,                     // 优化：批量大小
	Compression:            kafka.Snappy,            // 优化：压缩降带宽
}

type EventHandler func(ctx context.Context, event *Event) error

type Consumer interface {
	RegisterHandler(eventType string, handler EventHandler)
	Start(ctx context.Context) error
	Stop() error
}

// IdempotencyStore 幂等检查接口（抽象便于替换 Redis / 内存 / DB）
type IdempotencyStore interface {
	CheckAndSet(ctx context.Context, key string, ttl time.Duration) (bool, error)
}

// DLQSender 死信队列发送接口（修正点：新增）
type DLQSender interface {
	Send(ctx context.Context, event *Event, reason string) error
}

// processMessage 处理单条消息：反序列化 → 幂等 → 分发 → 重试 → (成功提交 | 发DLQ)
func (c *kafkaConsumer) processMessage(ctx context.Context, msg kafka.Message) {
	var event Event
	if err := json.Unmarshal(msg.Value, &event); err != nil {
		c.logger.Error("event: failed to unmarshal, skipping", "err", err, "key", string(msg.Key))
		c.commit(ctx, msg) // 毒丸消息：提交跳过，避免无限重试
		return
	}

	// 幂等检查（按幂等键）。Redis 故障 fail-open：记录日志但继续处理（宁可重复）
	if c.idempotencyStore != nil {
		ok, err := c.idempotencyStore.CheckAndSet(ctx, "idempotency:"+event.IdempotencyKey, 24*time.Hour)
		if err != nil {
			c.logger.Error("event: idempotency check failed, continue(fail-open)", "err", err)
		}
		if !ok {
			c.logger.Debug("event: duplicate, skip", "id", event.ID)
			c.commit(ctx, msg)
			return
		}
	}

	c.mu.RLock()
	handler, ok := c.handlers[event.Type]
	c.mu.RUnlock()
	if !ok {
		c.logger.Warn("event: no handler for type", "type", event.Type)
		c.commit(ctx, msg)
		return
	}

	// 重试退避（修正点：sleep 可被 ctx 中断，停机时不空等）
	var lastErr error
	for i := 0; i < 3; i++ {
		if err := handler(ctx, &event); err != nil {
			lastErr = err
			c.logger.Error("event: handler failed, retrying",
				"type", event.Type, "id", event.ID, "attempt", i+1, "err", err)
			select {
			case <-ctx.Done():
				return // 停机：交给优雅退出逻辑，不强行提交
			case <-time.After(time.Duration(200*(i+1)) * time.Millisecond):
			}
			continue
		}
		c.commit(ctx, msg)
		return
	}

	// 修正点：重试耗尽 → 发往 DLQ（不再静默丢弃），再提交原消息让主队列前进
	c.logger.Error("event: handler failed after retries, to DLQ",
		"type", event.Type, "id", event.ID, "err", lastErr)
	if c.dlq != nil {
		if derr := c.dlq.Send(ctx, &event, lastErr.Error()); derr != nil {
			c.logger.Error("event: send to DLQ failed", "err", derr)
		}
	}
	c.commit(ctx, msg)
}

func (c *kafkaConsumer) commit(ctx context.Context, msg kafka.Message) {
	if err := c.reader.CommitMessages(ctx, msg); err != nil {
		c.logger.Error("event: commit failed", "err", err, "id", string(msg.Key))
	}
}

// RedisIdempotencyStore 基于 Redis SetNX 的原子幂等检查。
// 返回 true = 首次（可处理）；false = 重复（应跳过）。TTL 防无限膨胀。
type RedisIdempotencyStore struct{ rdb *redis.Client }

func NewRedisIdempotencyStore(rdb *redis.Client) *RedisIdempotencyStore {
	return &RedisIdempotencyStore{rdb: rdb}
}

func (s *RedisIdempotencyStore) CheckAndSet(ctx context.Context, key string, ttl time.Duration) (bool, error) {
	result, err := s.rdb.SetNX(ctx, key, "1", ttl).Result()
	if err != nil {
		return false, fmt.Errorf("event: idempotency check failed: %w", err)
	}
	return result, nil
}

var _ IdempotencyStore = (*RedisIdempotencyStore)(nil)

type EventHandlerService struct {
	fileHandler    func(context.Context, *Event) error
	shareHandler   func(context.Context, *Event) error
	recycleHandler func(context.Context, *Event) error
	uploadHandler  func(context.Context, *Event) error
	mu             sync.RWMutex
}

func NewEventHandlerService(shareRepo biz.ShareRepo) *EventHandlerService {
	svc := &EventHandlerService{}
	svc.SetFileHandler(NewFileEventHandler(shareRepo))
	svc.SetShareHandler(NewShareEventHandler())
	svc.SetRecycleHandler(NewRecycleEventHandler())
	svc.SetUploadHandler(NewUploadEventHandler())
	return svc
}

// Handle 统一分发入口
func (s *EventHandlerService) Handle(ctx context.Context, event *Event) error {
	s.mu.RLock()
	defer s.mu.RUnlock()
	switch event.Type {
	case biz.EventFileDeleted, biz.EventFileMoved, biz.EventFileTrashed:
		if s.fileHandler != nil {
			return s.fileHandler(ctx, event)
		}
	case biz.EventShareCreated, biz.EventShareCancelled:
		if s.shareHandler != nil {
			return s.shareHandler(ctx, event)
		}
	case biz.EventRecycleClean:
		if s.recycleHandler != nil {
			return s.recycleHandler(ctx, event)
		}
	case biz.EventUploadCompleted:
		if s.uploadHandler != nil {
			return s.uploadHandler(ctx, event)
		}
	default:
		return fmt.Errorf("event: unknown event type: %s", event.Type)
	}
	return nil
}

func RegisterAllHandlers(consumer Consumer, handlerSvc *EventHandlerService) {
	consumer.RegisterHandler(biz.EventFileDeleted, handlerSvc.Handle)
	consumer.RegisterHandler(biz.EventFileMoved, handlerSvc.Handle)
	consumer.RegisterHandler(biz.EventFileTrashed, handlerSvc.Handle)
	consumer.RegisterHandler(biz.EventShareCreated, handlerSvc.Handle)
	consumer.RegisterHandler(biz.EventShareCancelled, handlerSvc.Handle)
	consumer.RegisterHandler(biz.EventRecycleClean, handlerSvc.Handle)
	consumer.RegisterHandler(biz.EventUploadCompleted, handlerSvc.Handle)
}

// parsePayload 通过 JSON 往返将 interface{} 负载转为强类型
func parsePayload(payload interface{}, out interface{}) error {
	data, err := json.Marshal(payload)
	if err != nil {
		return err
	}
	return json.Unmarshal(data, out)
}

// NewFileEventHandler 文件删除/移回收站时异步取消相关分享（最终一致）。
func NewFileEventHandler(shareRepo biz.ShareRepo) func(context.Context, *Event) error {
	return func(ctx context.Context, event *Event) error {
		var payload biz.FileChangedPayload
		if err := parsePayload(event.Payload, &payload); err != nil {
			log.Warn("event: parse file payload failed", "err", err)
			return nil // 负载解析失败不重试（非瞬态）
		}
		// DeleteByItemID 本身幂等：重复取消无害，配合幂等键保证副作用只生效一次
		for _, id := range payload.FileIDs {
			if err := shareRepo.DeleteByItemID(ctx, "file", id); err != nil {
				log.Error("event: cancel share for file failed", "file_id", id, "err", err)
			}
		}
		for _, id := range payload.FolderIDs {
			if err := shareRepo.DeleteByItemID(ctx, "folder", id); err != nil {
				log.Error("event: cancel share for folder failed", "folder_id", id, "err", err)
			}
		}
		log.Info("event: file event handled", "type", event.Type, "id", event.ID,
			"files", len(payload.FileIDs), "folders", len(payload.FolderIDs))
		return nil
	}
}

func NewShareEventHandler() func(context.Context, *Event) error {
	return func(ctx context.Context, event *Event) error {
		log.Info("event: share event handled", "type", event.Type, "id", event.ID)
		return nil
	}
}
func NewRecycleEventHandler() func(context.Context, *Event) error {
	return func(ctx context.Context, event *Event) error {
		log.Info("event: recycle event handled", "type", event.Type, "id", event.ID)
		return nil
	}
}
func NewUploadEventHandler() func(context.Context, *Event) error {
	return func(ctx context.Context, event *Event) error {
		log.Info("event: upload completed handled", "type", event.Type, "id", event.ID)
		return nil
	}
}

var _ transport.Server = (*ConsumerServer)(nil)

type ConsumerServer struct {
	consumer       Consumer
	handlerService *EventHandlerService
	idempotency    IdempotencyStore
	dlq            DLQSender // 修正点：注入死信发送器
	topics         []string
}

func NewConsumerServer(consumer Consumer, handlerService *EventHandlerService,
	idempotency IdempotencyStore, dlq DLQSender, topics []string) *ConsumerServer {
	return &ConsumerServer{consumer: consumer, handlerService: handlerService,
		idempotency: idempotency, dlq: dlq, topics: topics}
}

func (s *ConsumerServer) Start(ctx context.Context) error {
	if s.handlerService != nil {
		RegisterAllHandlers(s.consumer, s.handlerService)
	}
	if s.idempotency != nil {
		if c, ok := s.consumer.(*kafkaConsumer); ok {
			c.SetIdempotencyStore(s.idempotency)
			c.dlq = s.dlq // 修正点：注入 DLQ
		}
	}
	if err := s.consumer.Start(ctx); err != nil {
		log.Error("event: failed to start consumer", "err", err)
		return err
	}
	log.Info("event: consumer server started", "topics", s.topics)
	return nil
}

func (s *ConsumerServer) Stop(ctx context.Context) error {
	if err := s.consumer.Stop(); err != nil {
		log.Error("event: failed to stop consumer", "err", err)
		return err
	}
	log.Info("event: consumer server stopped")
	return nil
}

const defaultConsumerGroup = "cloud-disk-consumer"

func KafkaConsumerGroup() string { return defaultConsumerGroup }

func KafkaTopics(cfg *conf.Data_Kafka) []string {
	if cfg == nil || cfg.Topics == nil {
		return []string{"file-events", "share-events", "recycle-events", "upload-events"}
	}
	prefix := cfg.TopicPrefix
	if prefix == "" {
		prefix = "cloud_disk_"
	}
	return []string{
		prefix + cfg.Topics.FileEvents,
		prefix + cfg.Topics.ShareEvents,
		prefix + cfg.Topics.RecycleEvents,
		prefix + cfg.Topics.UploadEvents,
	}
}

var ProviderSet = wire.NewSet(
	NewProducer,
	NewConsumer,
	NewEventHandlerService,
	NewRedisIdempotencyStore,
	NewEventPublisherAdapter,
	NewConsumerServer,
	KafkaConsumerGroup,
	KafkaTopics,
	wire.Bind(new(IdempotencyStore), new(*RedisIdempotencyStore)),
	// 修正点：绑定 DLQ 发送器
	wire.Bind(new(DLQSender), new(*KafkaDLQSender)),
)

type noopProducer struct{}

func (noopProducer) Publish(ctx context.Context, topic string, event *Event) error { return nil }
func (noopProducer) Close() error                                                { return nil }

type noopConsumer struct{}

func (noopConsumer) RegisterHandler(eventType string, handler EventHandler) {}
func (noopConsumer) Start(ctx context.Context) error                        { return nil }
func (noopConsumer) Stop() error                                            { return nil }

type noopEventPublisher struct{}

func (noopEventPublisher) Publish(ctx context.Context, eventType string, payload interface{}) error {
	return nil
}

// Outbox 事件发件箱表：与业务在同一 DB 事务写入，保证原子性。
type Outbox struct {
	ID        uint64 `gorm:"primaryKey"`
	EventType string `gorm:"index"`
	Payload   string // JSON 序列化后的事件（含 PartitionKey）
	Status    int    // 0=待发送 1=已发送
	CreatedAt time.Time
}

// OutboxRepo 发件箱仓储
type OutboxRepo interface {
	// Save 在业务事务内调用，把事件作为"待发送"写入
	Save(ctx context.Context, eventType string, payload []byte) error
	// Pending 取出待发送事件（中继者轮询）
	Pending(ctx context.Context, limit int) ([]*Outbox, error)
	// MarkSent 标记已发送
	MarkSent(ctx context.Context, id uint64) error
}

// 原错误写法（已修正）：_ = uc.eventPublisher.Publish(...)  // 吞错 + 非原子
//
// 修正点：改为在 DB 事务内写入 outbox，由中继者异步投递，保证"业务落库=事件落库"
func (uc *ShareUsecase) CreateShare(ctx context.Context, userID uint64, itemType string, itemID uint64, ...) (...) {
	// ... 构造 share 并 shareRepo.Create（事务内）...
	// 修正点：同一事务内落 outbox
	payload, _ := json.Marshal(&SharePayload{UserID: userID, ShareID: created.ID, UUID: created.UUID, Action: "created"})
	if err := uc.outbox.Save(ctx, EventShareCreated, payload); err != nil {
		return nil, "", err // 落 outbox 失败 = 整笔事务回滚，绝不会"分享了却没发事件"
	}
	return created, url, nil
}

// OutboxRelay 后台 goroutine：轮询 outbox，投递到 Kafka，成功则标记已发。
func (r *outboxRelay) Run(ctx context.Context) {
	ticker := time.NewTicker(500 * time.Millisecond)
	defer ticker.Stop()
	for {
		select {
		case <-ctx.Done():
			return
		case <-ticker.C:
			batch, err := r.outbox.Pending(ctx, 100)
			if err != nil || len(batch) == 0 {
				continue
			}
			for _, row := range batch {
				// 构造事件并发布（带分区键）
				evt, _ := rebuildEvent(row) // 从 payload 还原 Event + PartitionKey
				if err := r.producer.Publish(ctx, r.topicOf(row.EventType), evt); err != nil {
					log.Error("outbox: publish failed", "id", row.ID, "err", err)
					continue // 不标记，下次轮询重试
				}
				_ = r.outbox.MarkSent(ctx, row.ID)
			}
		}
	}
}

// 示例：在重试耗尽发 DLQ 处计数（伪代码）
eventDLQTotal.WithLabelValues(event.Type).Inc()
// 在 processMessage 入口记录耗时
defer func(start time.Time) {
	eventProcessDuration.WithLabelValues(event.Type).Observe(time.Since(start).Seconds())
}(time.Now())