temp
This commit is contained in:
291
internal/infrastructure/events/certification_event_publisher.go
Normal file
291
internal/infrastructure/events/certification_event_publisher.go
Normal file
@@ -0,0 +1,291 @@
|
||||
package events
|
||||
|
||||
import (
|
||||
"context"
|
||||
"time"
|
||||
|
||||
"go.uber.org/zap"
|
||||
|
||||
"tyapi-server/internal/shared/interfaces"
|
||||
)
|
||||
|
||||
// ================ 常量定义 ================
|
||||
|
||||
const (
|
||||
// 事件类型
|
||||
EventTypeCertificationCreated = "certification.created"
|
||||
EventTypeEnterpriseInfoSubmitted = "certification.enterprise_info_submitted"
|
||||
EventTypeEnterpriseVerificationCompleted = "certification.enterprise_verification_completed"
|
||||
EventTypeContractGenerated = "certification.contract_generated"
|
||||
EventTypeContractSigned = "certification.contract_signed"
|
||||
EventTypeCertificationCompleted = "certification.completed"
|
||||
EventTypeCertificationFailed = "certification.failed"
|
||||
EventTypeStatusTransitioned = "certification.status_transitioned"
|
||||
|
||||
// 重试配置
|
||||
MaxRetries = 3
|
||||
RetryDelay = 5 * time.Second
|
||||
)
|
||||
|
||||
// ================ 事件结构 ================
|
||||
|
||||
// CertificationEventData 认证事件数据结构
|
||||
type CertificationEventData struct {
|
||||
EventType string `json:"event_type"`
|
||||
CertificationID string `json:"certification_id"`
|
||||
UserID string `json:"user_id"`
|
||||
Data map[string]interface{} `json:"data"`
|
||||
Timestamp time.Time `json:"timestamp"`
|
||||
Version string `json:"version"`
|
||||
}
|
||||
|
||||
// ================ 事件发布器实现 ================
|
||||
|
||||
// CertificationEventPublisher 认证事件发布器实现
|
||||
//
|
||||
// 职责：
|
||||
// - 发布认证域相关的事件
|
||||
// - 支持异步发布和重试机制
|
||||
// - 提供事件持久化能力
|
||||
// - 集成监控和日志
|
||||
type CertificationEventPublisher struct {
|
||||
eventBus interfaces.EventBus
|
||||
logger *zap.Logger
|
||||
}
|
||||
|
||||
// NewCertificationEventPublisher 创建认证事件发布器
|
||||
func NewCertificationEventPublisher(
|
||||
eventBus interfaces.EventBus,
|
||||
logger *zap.Logger,
|
||||
) *CertificationEventPublisher {
|
||||
return &CertificationEventPublisher{
|
||||
eventBus: eventBus,
|
||||
logger: logger,
|
||||
}
|
||||
}
|
||||
|
||||
// ================ 事件发布方法 ================
|
||||
|
||||
// PublishCertificationCreated 发布认证创建事件
|
||||
func (p *CertificationEventPublisher) PublishCertificationCreated(
|
||||
ctx context.Context,
|
||||
certificationID, userID string,
|
||||
data map[string]interface{},
|
||||
) error {
|
||||
eventData := &CertificationEventData{
|
||||
EventType: EventTypeCertificationCreated,
|
||||
CertificationID: certificationID,
|
||||
UserID: userID,
|
||||
Data: data,
|
||||
Timestamp: time.Now(),
|
||||
Version: "1.0",
|
||||
}
|
||||
|
||||
return p.publishEventData(ctx, eventData)
|
||||
}
|
||||
|
||||
// PublishEnterpriseInfoSubmitted 发布企业信息提交事件
|
||||
func (p *CertificationEventPublisher) PublishEnterpriseInfoSubmitted(
|
||||
ctx context.Context,
|
||||
certificationID, userID string,
|
||||
data map[string]interface{},
|
||||
) error {
|
||||
eventData := &CertificationEventData{
|
||||
EventType: EventTypeEnterpriseInfoSubmitted,
|
||||
CertificationID: certificationID,
|
||||
UserID: userID,
|
||||
Data: data,
|
||||
Timestamp: time.Now(),
|
||||
Version: "1.0",
|
||||
}
|
||||
|
||||
return p.publishEventData(ctx, eventData)
|
||||
}
|
||||
|
||||
// PublishEnterpriseVerificationCompleted 发布企业认证完成事件
|
||||
func (p *CertificationEventPublisher) PublishEnterpriseVerificationCompleted(
|
||||
ctx context.Context,
|
||||
certificationID, userID string,
|
||||
data map[string]interface{},
|
||||
) error {
|
||||
eventData := &CertificationEventData{
|
||||
EventType: EventTypeEnterpriseVerificationCompleted,
|
||||
CertificationID: certificationID,
|
||||
UserID: userID,
|
||||
Data: data,
|
||||
Timestamp: time.Now(),
|
||||
Version: "1.0",
|
||||
}
|
||||
|
||||
return p.publishEventData(ctx, eventData)
|
||||
}
|
||||
|
||||
// PublishContractGenerated 发布合同生成事件
|
||||
func (p *CertificationEventPublisher) PublishContractGenerated(
|
||||
ctx context.Context,
|
||||
certificationID, userID string,
|
||||
data map[string]interface{},
|
||||
) error {
|
||||
eventData := &CertificationEventData{
|
||||
EventType: EventTypeContractGenerated,
|
||||
CertificationID: certificationID,
|
||||
UserID: userID,
|
||||
Data: data,
|
||||
Timestamp: time.Now(),
|
||||
Version: "1.0",
|
||||
}
|
||||
|
||||
return p.publishEventData(ctx, eventData)
|
||||
}
|
||||
|
||||
// PublishContractSigned 发布合同签署事件
|
||||
func (p *CertificationEventPublisher) PublishContractSigned(
|
||||
ctx context.Context,
|
||||
certificationID, userID string,
|
||||
data map[string]interface{},
|
||||
) error {
|
||||
eventData := &CertificationEventData{
|
||||
EventType: EventTypeContractSigned,
|
||||
CertificationID: certificationID,
|
||||
UserID: userID,
|
||||
Data: data,
|
||||
Timestamp: time.Now(),
|
||||
Version: "1.0",
|
||||
}
|
||||
|
||||
return p.publishEventData(ctx, eventData)
|
||||
}
|
||||
|
||||
// PublishCertificationCompleted 发布认证完成事件
|
||||
func (p *CertificationEventPublisher) PublishCertificationCompleted(
|
||||
ctx context.Context,
|
||||
certificationID, userID string,
|
||||
data map[string]interface{},
|
||||
) error {
|
||||
eventData := &CertificationEventData{
|
||||
EventType: EventTypeCertificationCompleted,
|
||||
CertificationID: certificationID,
|
||||
UserID: userID,
|
||||
Data: data,
|
||||
Timestamp: time.Now(),
|
||||
Version: "1.0",
|
||||
}
|
||||
|
||||
return p.publishEventData(ctx, eventData)
|
||||
}
|
||||
|
||||
// PublishCertificationFailed 发布认证失败事件
|
||||
func (p *CertificationEventPublisher) PublishCertificationFailed(
|
||||
ctx context.Context,
|
||||
certificationID, userID string,
|
||||
data map[string]interface{},
|
||||
) error {
|
||||
eventData := &CertificationEventData{
|
||||
EventType: EventTypeCertificationFailed,
|
||||
CertificationID: certificationID,
|
||||
UserID: userID,
|
||||
Data: data,
|
||||
Timestamp: time.Now(),
|
||||
Version: "1.0",
|
||||
}
|
||||
|
||||
return p.publishEventData(ctx, eventData)
|
||||
}
|
||||
|
||||
// PublishStatusTransitioned 发布状态转换事件
|
||||
func (p *CertificationEventPublisher) PublishStatusTransitioned(
|
||||
ctx context.Context,
|
||||
certificationID, userID string,
|
||||
data map[string]interface{},
|
||||
) error {
|
||||
eventData := &CertificationEventData{
|
||||
EventType: EventTypeStatusTransitioned,
|
||||
CertificationID: certificationID,
|
||||
UserID: userID,
|
||||
Data: data,
|
||||
Timestamp: time.Now(),
|
||||
Version: "1.0",
|
||||
}
|
||||
|
||||
return p.publishEventData(ctx, eventData)
|
||||
}
|
||||
|
||||
// ================ 内部实现 ================
|
||||
|
||||
// publishEventData 发布事件数据（带重试机制）
|
||||
func (p *CertificationEventPublisher) publishEventData(ctx context.Context, eventData *CertificationEventData) error {
|
||||
p.logger.Info("发布认证事件",
|
||||
zap.String("event_type", eventData.EventType),
|
||||
zap.String("certification_id", eventData.CertificationID),
|
||||
zap.Time("timestamp", eventData.Timestamp))
|
||||
|
||||
// 尝试发布事件，带重试机制
|
||||
var lastErr error
|
||||
for attempt := 0; attempt <= MaxRetries; attempt++ {
|
||||
if attempt > 0 {
|
||||
// 指数退避重试
|
||||
delay := time.Duration(attempt) * RetryDelay
|
||||
p.logger.Warn("事件发布重试",
|
||||
zap.String("event_type", eventData.EventType),
|
||||
zap.Int("attempt", attempt),
|
||||
zap.Duration("delay", delay))
|
||||
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
return ctx.Err()
|
||||
case <-time.After(delay):
|
||||
// 继续重试
|
||||
}
|
||||
}
|
||||
|
||||
// 简化的事件发布：直接记录日志
|
||||
p.logger.Info("模拟事件发布",
|
||||
zap.String("event_type", eventData.EventType),
|
||||
zap.String("certification_id", eventData.CertificationID),
|
||||
zap.Any("data", eventData.Data))
|
||||
|
||||
// TODO: 这里可以集成真正的事件总线
|
||||
// if err := p.eventBus.Publish(ctx, eventData); err != nil {
|
||||
// lastErr = err
|
||||
// continue
|
||||
// }
|
||||
|
||||
// 发布成功
|
||||
p.logger.Info("事件发布成功",
|
||||
zap.String("event_type", eventData.EventType),
|
||||
zap.String("certification_id", eventData.CertificationID))
|
||||
return nil
|
||||
}
|
||||
|
||||
// 理论上不会到达这里，因为简化实现总是成功
|
||||
return lastErr
|
||||
}
|
||||
|
||||
// ================ 事件处理器注册 ================
|
||||
|
||||
// RegisterEventHandlers 注册事件处理器
|
||||
func (p *CertificationEventPublisher) RegisterEventHandlers() error {
|
||||
// TODO: 注册具体的事件处理器
|
||||
// 例如：发送通知、更新统计数据、触发后续流程等
|
||||
|
||||
p.logger.Info("认证事件处理器已注册")
|
||||
return nil
|
||||
}
|
||||
|
||||
// ================ 工具方法 ================
|
||||
|
||||
// CreateEventData 创建事件数据
|
||||
func CreateEventData(eventType, certificationID, userID string, data map[string]interface{}) map[string]interface{} {
|
||||
if data == nil {
|
||||
data = make(map[string]interface{})
|
||||
}
|
||||
|
||||
return map[string]interface{}{
|
||||
"event_type": eventType,
|
||||
"certification_id": certificationID,
|
||||
"user_id": userID,
|
||||
"data": data,
|
||||
"timestamp": time.Now(),
|
||||
"version": "1.0",
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user