/**
 * 领域事件基类
 */
public abstract class DomainEvent {
    private final String eventId;
    private final LocalDateTime occurredOn;
    
    protected DomainEvent() {
        this.eventId = UUID.randomUUID().toString();
        this.occurredOn = LocalDateTime.now();
    }
    
    protected DomainEvent(String eventId, LocalDateTime occurredOn) {
        this.eventId = eventId;
        this.occurredOn = occurredOn;
    }
    
    public String getEventId() { return eventId; }
    public LocalDateTime getOccurredOn() { return occurredOn; }
}

/**
 * 订单已创建事件
 */
public class OrderCreatedEvent extends DomainEvent {
    private final OrderId orderId;
    private final CustomerId customerId;
    private final Money totalAmount;
    private final List<OrderItemSnapshot> items;
    
    public OrderCreatedEvent(OrderId orderId, CustomerId customerId, 
                            Money totalAmount, List<OrderItemSnapshot> items) {
        super();
        this.orderId = orderId;
        this.customerId = customerId;
        this.totalAmount = totalAmount;
        this.items = Collections.unmodifiableList(items);
    }
    
    // Getters
    public OrderId getOrderId() { return orderId; }
    public CustomerId getCustomerId() { return customerId; }
    public Money getTotalAmount() { return totalAmount; }
    public List<OrderItemSnapshot> getItems() { return items; }
}

/**
 * 订单项快照（值对象）
 */
public class OrderItemSnapshot {
    private final ProductId productId;
    private final String productName;
    private final Money unitPrice;
    private final int quantity;
    
    public OrderItemSnapshot(ProductId productId, String productName, 
                            Money unitPrice, int quantity) {
        this.productId = productId;
        this.productName = productName;
        this.unitPrice = unitPrice;
        this.quantity = quantity;
    }
    
    // Getters
    public ProductId getProductId() { return productId; }
    public String getProductName() { return productName; }
    public Money getUnitPrice() { return unitPrice; }
    public int getQuantity() { return quantity; }
}

/**
 * 订单已取消事件
 */
public class OrderCancelledEvent extends DomainEvent {
    private final OrderId orderId;
    private final CancellationReason reason;
    private final LocalDateTime cancelledAt;
    
    public OrderCancelledEvent(OrderId orderId, CancellationReason reason) {
        super();
        this.orderId = orderId;
        this.reason = reason;
        this.cancelledAt = LocalDateTime.now();
    }
    
    // Getters
    public OrderId getOrderId() { return orderId; }
    public CancellationReason getReason() { return reason; }
    public LocalDateTime getCancelledAt() { return cancelledAt; }
}

/**
 * 领域事件发布器接口
 */
public interface DomainEventPublisher {
    void publish(DomainEvent event);
    void publishAll(List<DomainEvent> events);
}

/**
 * 简单的内存事件发布器（用于演示）
 */
public class InMemoryEventPublisher implements DomainEventPublisher {
    private final List<EventSubscriber> subscribers = new CopyOnWriteArrayList<>();
    
    @Override
    public void publish(DomainEvent event) {
        for (EventSubscriber subscriber : subscribers) {
            if (subscriber.supports(event.getClass())) {
                try {
                    subscriber.handle(event);
                } catch (Exception e) {
                    // 记录错误日志
                    System.err.println("Event handling failed: " + e.getMessage());
                }
            }
        }
    }
    
    @Override
    public void publishAll(List<DomainEvent> events) {
        for (DomainEvent event : events) {
            publish(event);
        }
    }
    
    public void subscribe(EventSubscriber subscriber) {
        subscribers.add(subscriber);
    }
}

/**
 * 事件订阅者接口
 */
public interface EventSubscriber {
    boolean supports(Class<? extends DomainEvent> eventType);
    void handle(DomainEvent event);
}

/**
 * 订单创建事件订阅者
 */
public class OrderCreatedSubscriber implements EventSubscriber {
    private final InventoryService inventoryService;
    private final NotificationService notificationService;
    
    public OrderCreatedSubscriber(InventoryService inventoryService,
                                 NotificationService notificationService) {
        this.inventoryService = inventoryService;
        this.notificationService = notificationService;
    }
    
    @Override
    public boolean supports(Class<? extends DomainEvent> eventType) {
        return OrderCreatedEvent.class.isAssignableFrom(eventType);
    }
    
    @Override
    public void handle(DomainEvent event) {
        OrderCreatedEvent orderEvent = (OrderCreatedEvent) event;
        
        // 扣减库存
        for (OrderItemSnapshot item : orderEvent.getItems()) {
            inventoryService.reduceStock(item.getProductId(), item.getQuantity());
        }
        
        // 发送通知
        notificationService.sendOrderConfirmation(
            orderEvent.getOrderId(),
            orderEvent.getCustomerId()
        );
    }
}

from abc import ABC, abstractmethod
from dataclasses import dataclass, field
from datetime import datetime
from typing import List, Dict, Type, Callable
import uuid


@dataclass(frozen=True)
class DomainEvent:
    """领域事件基类"""
    event_id: str = field(default_factory=lambda: str(uuid.uuid4()))
    occurred_on: datetime = field(default_factory=datetime.now)


@dataclass(frozen=True)
class UserRegisteredEvent(DomainEvent):
    """用户已注册事件"""
    user_id: str
    email: str
    name: str


@dataclass(frozen=True)
class UserDeactivatedEvent(DomainEvent):
    """用户已停用事件"""
    user_id: str
    reason: str


@dataclass(frozen=True)
class OrderPlacedEvent(DomainEvent):
    """订单已下单事件"""
    order_id: str
    customer_id: str
    total_amount: float
    items: List[Dict]


class EventSubscriber(ABC):
    """事件订阅者抽象类"""
    
    @abstractmethod
    def handles(self) -> Type[DomainEvent]:
        """返回该订阅者处理的事件类型"""
        pass
    
    @abstractmethod
    def handle(self, event: DomainEvent):
        """处理事件"""
        pass


class EventBus:
    """事件总线
    
    负责事件的发布和订阅
    """
    
    def __init__(self):
        self._subscribers: Dict[Type[DomainEvent], List[EventSubscriber]] = {}
    
    def subscribe(self, subscriber: EventSubscriber):
        """订阅事件"""
        event_type = subscriber.handles()
        if event_type not in self._subscribers:
            self._subscribers[event_type] = []
        self._subscribers[event_type].append(subscriber)
    
    def publish(self, event: DomainEvent):
        """发布事件"""
        event_type = type(event)
        subscribers = self._subscribers.get(event_type, [])
        
        for subscriber in subscribers:
            try:
                subscriber.handle(event)
            except Exception as e:
                # 记录错误
                print(f"Event handling failed: {e}")
    
    def publish_all(self, events: List[DomainEvent]):
        """发布多个事件"""
        for event in events:
            self.publish(event)


# 具体订阅者
class WelcomeEmailSubscriber(EventSubscriber):
    """欢迎邮件订阅者"""
    
    def handles(self) -> Type[DomainEvent]:
        return UserRegisteredEvent
    
    def handle(self, event: DomainEvent):
        register_event = event
        print(f"Sending welcome email to {register_event.email}")
        # 实际发送邮件的逻辑


class AnalyticsSubscriber(EventSubscriber):
    """分析数据订阅者"""
    
    def handles(self) -> Type[DomainEvent]:
        return UserRegisteredEvent
    
    def handle(self, event: DomainEvent):
        register_event = event
        print(f"Recording user registration: {register_event.user_id}")
        # 记录分析数据


class InventorySubscriber(EventSubscriber):
    """库存扣减订阅者"""
    
    def handles(self) -> Type[DomainEvent]:
        return OrderPlacedEvent
    
    def handle(self, event: DomainEvent):
        order_event = event
        print(f"Reducing inventory for order {order_event.order_id}")
        for item in order_event.items:
            print(f"  Product {item['product_id']}: -{item['quantity']}")


# 使用示例
def event_bus_example():
    """演示事件总线的使用"""
    # 创建事件总线
    event_bus = EventBus()
    
    # 订阅事件
    event_bus.subscribe(WelcomeEmailSubscriber())
    event_bus.subscribe(AnalyticsSubscriber())
    event_bus.subscribe(InventorySubscriber())
    
    # 发布事件
    user_event = UserRegisteredEvent(
        user_id="U001",
        email="john@example.com",
        name="John Doe"
    )
    event_bus.publish(user_event)
    
    order_event = OrderPlacedEvent(
        order_id="ORD001",
        customer_id="C001",
        total_amount=99.99,
        items=[
            {"product_id": "P001", "quantity": 2},
            {"product_id": "P002", "quantity": 1}
        ]
    )
    event_bus.publish(order_event)

/**
 * Outbox 实体
 * 用于保证事件发布的可靠性
 */
@Entity
@Table(name = "event_outbox")
public class OutboxEvent {
    @Id
    private String id;
    
    private String eventType;
    
    @Column(columnDefinition = "TEXT")
    private String payload;
    
    private LocalDateTime createdAt;
    
    private LocalDateTime processedAt;
    
    private String status;  // PENDING, PROCESSED, FAILED
    
    // 构造函数
    public OutboxEvent(String id, String eventType, String payload) {
        this.id = id;
        this.eventType = eventType;
        this.payload = payload;
        this.createdAt = LocalDateTime.now();
        this.status = "PENDING";
    }
    
    public void markAsProcessed() {
        this.processedAt = LocalDateTime.now();
        this.status = "PROCESSED";
    }
    
    public void markAsFailed() {
        this.status = "FAILED";
    }
    
    // Getters
    public String getId() { return id; }
    public String getEventType() { return eventType; }
    public String getPayload() { return payload; }
    public LocalDateTime getCreatedAt() { return createdAt; }
    public LocalDateTime getProcessedAt() { return processedAt; }
    public String getStatus() { return status; }
}

/**
 * Outbox 仓储
 */
public interface OutboxRepository {
    void save(OutboxEvent event);
    List<OutboxEvent> findPendingEvents(int limit);
    void markAsProcessed(String eventId);
    void markAsFailed(String eventId);
}

/**
 * Outbox 事件发布器
 * 
 * 先将事件保存到 Outbox 表，再由后台进程处理
 */
public class OutboxEventPublisher implements DomainEventPublisher {
    private final OutboxRepository outboxRepository;
    private final ObjectMapper objectMapper;
    
    public OutboxEventPublisher(OutboxRepository outboxRepository,
                               ObjectMapper objectMapper) {
        this.outboxRepository = outboxRepository;
        this.objectMapper = objectMapper;
    }
    
    @Override
    public void publish(DomainEvent event) {
        try {
            String payload = objectMapper.writeValueAsString(event);
            OutboxEvent outboxEvent = new OutboxEvent(
                UUID.randomUUID().toString(),
                event.getClass().getSimpleName(),
                payload
            );
            outboxRepository.save(outboxEvent);
        } catch (JsonProcessingException e) {
            throw new RuntimeException("Failed to serialize event", e);
        }
    }
    
    @Override
    public void publishAll(List<DomainEvent> events) {
        for (DomainEvent event : events) {
            publish(event);
        }
    }
}

/**
 * Outbox 处理器
 * 后台进程，定期从 Outbox 表读取事件并发布
 */
@Component
public class OutboxProcessor {
    private final OutboxRepository outboxRepository;
    private final MessageBroker messageBroker;
    
    @Scheduled(fixedDelay = 5000)  // 每5秒执行一次
    public void processPendingEvents() {
        List<OutboxEvent> pendingEvents = outboxRepository.findPendingEvents(100);
        
        for (OutboxEvent event : pendingEvents) {
            try {
                messageBroker.send(event.getEventType(), event.getPayload());
                outboxRepository.markAsProcessed(event.getId());
            } catch (Exception e) {
                outboxRepository.markAsFailed(event.getId());
                // 记录错误日志
            }
        }
    }
}

public class OrderEvent {
    private Long orderId;
    private Long userId;
    private String status;
    private Date timestamp;
    // 包含整个订单对象
    private Order order;
}