Event Sourcing Pipeline

// Event Sourcing Implementation
@Entity
public class CustomerAggregate {
    
    @Id
    private String customerId;
    
    @Version
    private Long version;
    
    private String name;
    private String email;
    private CustomerStatus status;
    
    @OneToMany(cascade = CascadeType.ALL, fetch = FetchType.EAGER)
    private List<DomainEvent> uncommittedEvents = new ArrayList<>();
    
    public void createCustomer(String name, String email) {
        if (this.customerId != null) {
            throw new IllegalStateException("Customer already exists");
        }
        
        CustomerCreatedEvent event = new CustomerCreatedEvent(
            UUID.randomUUID().toString(),
            name,
            email,
            Instant.now()
        );
        
        apply(event);
        uncommittedEvents.add(event);
    }
    
    public void updateEmail(String newEmail) {
        if (this.customerId == null) {
            throw new IllegalStateException("Customer does not exist");
        }
        
        EmailUpdatedEvent event = new EmailUpdatedEvent(
            this.customerId,
            this.email,
            newEmail,
            Instant.now()
        );
        
        apply(event);
        uncommittedEvents.add(event);
    }
    
    private void apply(DomainEvent event) {
        if (event instanceof CustomerCreatedEvent) {
            CustomerCreatedEvent e = (CustomerCreatedEvent) event;
            this.customerId = e.getCustomerId();
            this.name = e.getName();
            this.email = e.getEmail();
            this.status = CustomerStatus.ACTIVE;
        } else if (event instanceof EmailUpdatedEvent) {
            EmailUpdatedEvent e = (EmailUpdatedEvent) event;
            this.email = e.getNewEmail();
        }
        
        this.version++;
    }
    
    public List<DomainEvent> getUncommittedEvents() {
        return new ArrayList<>(uncommittedEvents);
    }
    
    public void markEventsAsCommitted() {
        uncommittedEvents.clear();
    }
}

// Event Store Implementation
@Repository
public class EventStore {
    
    @Autowired
    private JdbcTemplate jdbcTemplate;
    
    public void appendEvents(String aggregateId, List<DomainEvent> events, long expectedVersion) {
        jdbcTemplate.execute(status -> {
            // Check version for optimistic concurrency
            Long currentVersion = jdbcTemplate.queryForObject(
                "SELECT version FROM aggregates WHERE aggregate_id = ?",
                Long.class,
                aggregateId
            );
            
            if (currentVersion != null && currentVersion != expectedVersion) {
                throw new ConcurrencyException("Version mismatch");
            }
            
            // Insert events
            for (DomainEvent event : events) {
                jdbcTemplate.update(
                    "INSERT INTO events (event_id, aggregate_id, event_type, event_data, version, timestamp) VALUES (?, ?, ?, ?, ?, ?)",
                    event.getEventId(),
                    aggregateId,
                    event.getClass().getSimpleName(),
                    serializeEvent(event),
                    expectedVersion + 1,
                    event.getTimestamp()
                );
            }
            
            // Update aggregate version
            if (currentVersion == null) {
                jdbcTemplate.update(
                    "INSERT INTO aggregates (aggregate_id, version) VALUES (?, ?)",
                    aggregateId,
                    expectedVersion + events.size()
                );
            } else {
                jdbcTemplate.update(
                    "UPDATE aggregates SET version = ? WHERE aggregate_id = ?",
                    expectedVersion + events.size(),
                    aggregateId
                );
            }
            
            return null;
        });
    }
    
    public List<DomainEvent> getEvents(String aggregateId, long fromVersion) {
        return jdbcTemplate.query(
            "SELECT event_type, event_data, version, timestamp FROM events WHERE aggregate_id = ? AND version > ? ORDER BY version",
            new Object[]{aggregateId, fromVersion},
            (rs, rowNum) -> deserializeEvent(
                rs.getString("event_type"),
                rs.getString("event_data"),
                rs.getLong("version"),
                rs.getTimestamp("timestamp").toInstant()
            )
        );
    }
    
    private String serializeEvent(DomainEvent event) {
        try {
            ObjectMapper mapper = new ObjectMapper();
            return mapper.writeValueAsString(event);
        } catch (JsonProcessingException e) {
            throw new RuntimeException("Failed to serialize event", e);
        }
    }
    
    private DomainEvent deserializeEvent(String eventType, String eventData, long version, Instant timestamp) {
        try {
            ObjectMapper mapper = new ObjectMapper();
            Class<?> eventClass = Class.forName("com.example.events." + eventType);
            DomainEvent event = (DomainEvent) mapper.readValue(eventData, eventClass);
            event.setVersion(version);
            event.setTimestamp(timestamp);
            return event;
        } catch (Exception e) {
            throw new RuntimeException("Failed to deserialize event", e);
        }
    }
}