Rewriting the Reward Application to Use Event Sourcing

To rewrite the Reward application using Event Sourcing, we will adjust the way we handle state changes and communication with other services. Instead of persisting the current state directly, we'll record all the events that represent state changes. This approach ensures that we can reconstruct the application's state by replaying these events.
Key Aspects:
Event Store: Implement an event store to persist events.
Domain Events: Use events as the primary source of truth for state changes.
Event Handlers: Handle events to update read models (projections) for querying.
Consistency: Ensure that all functionalities are maintained, and the application remains production-ready.

1. Updating the Reward Domain Model
We will modify the Reward entity to use events for all state changes.
1.1. Define Domain Events
Create events representing actions in the Reward domain.
RewardCreatedEvent
package com.parkingexchange.reward.domain.event;

import java.math.BigDecimal;
import java.time.Instant;
import java.util.UUID;

public class RewardCreatedEvent {

    private final UUID rewardId;
    private final UUID fromUserId;
    private final UUID toUserId;
    private final UUID availabilityId;
    private final BigDecimal amount;
    private final Instant occurredOn;

    public RewardCreatedEvent(UUID rewardId, UUID fromUserId, UUID toUserId,
                              UUID availabilityId, BigDecimal amount) {
        this.rewardId = rewardId;
        this.fromUserId = fromUserId;
        this.toUserId = toUserId;
        this.availabilityId = availabilityId;
        this.amount = amount;
        this.occurredOn = Instant.now();
    }

    // Getters ...

}


RewardCompletedEvent
package com.parkingexchange.reward.domain.event;

import java.time.Instant;
import java.util.UUID;

public class RewardCompletedEvent {

    private final UUID rewardId;
    private final Instant occurredOn;

    public RewardCompletedEvent(UUID rewardId) {
        this.rewardId = rewardId;
        this.occurredOn = Instant.now();
    }

    // Getters ...

}


RewardFailedEvent
package com.parkingexchange.reward.domain.event;

import java.time.Instant;
import java.util.UUID;

public class RewardFailedEvent {

    private final UUID rewardId;
    private final String reason;
    private final Instant occurredOn;

    public RewardFailedEvent(UUID rewardId, String reason) {
        this.rewardId = rewardId;
        this.reason = reason;
        this.occurredOn = Instant.now();
    }

    // Getters ...

}


1.2. Modify the Reward Aggregate
Update the Reward aggregate to apply events.
package com.parkingexchange.reward.domain.model;

import com.parkingexchange.reward.domain.event.RewardCompletedEvent;
import com.parkingexchange.reward.domain.event.RewardCreatedEvent;
import com.parkingexchange.reward.domain.event.RewardFailedEvent;
import java.math.BigDecimal;
import java.time.LocalDateTime;
import java.util.ArrayList;
import java.util.List;
import java.util.UUID;

public class Reward {

    private UUID rewardId;
    private UUID fromUserId;
    private UUID toUserId;
    private UUID availabilityId;
    private BigDecimal amount;
    private RewardStatus status;
    private LocalDateTime createdAt;
    private LocalDateTime completedAt;

    // Events to be persisted
    private final List<Object> pendingEvents = new ArrayList<>();

    // Event sourcing constructor
    public Reward(List<Object> eventStream) {
        for (Object event : eventStream) {
            applyEvent(event);
        }
    }

    // Factory method to create a new Reward
    public static Reward create(UUID fromUserId, UUID toUserId, UUID availabilityId,
                                BigDecimal amount) {
        Reward reward = new Reward();
        reward.applyChange(new RewardCreatedEvent(
                UUID.randomUUID(),
                fromUserId,
                toUserId,
                availabilityId,
                amount
        ));
        return reward;
    }

    // Apply events to rebuild state
    private void applyEvent(Object event) {
        if (event instanceof RewardCreatedEvent) {
            apply((RewardCreatedEvent) event);
        } else if (event instanceof RewardCompletedEvent) {
            apply((RewardCompletedEvent) event);
        } else if (event instanceof RewardFailedEvent) {
            apply((RewardFailedEvent) event);
        }
    }

    // Apply changes and record events
    private void applyChange(Object event) {
        applyEvent(event);
        pendingEvents.add(event);
    }

    private void apply(RewardCreatedEvent event) {
        this.rewardId = event.getRewardId();
        this.fromUserId = event.getFromUserId();
        this.toUserId = event.getToUserId();
        this.availabilityId = event.getAvailabilityId();
        this.amount = event.getAmount();
        this.status = RewardStatus.PENDING;
        this.createdAt = LocalDateTime.now();
    }

    private void apply(RewardCompletedEvent event) {
        this.status = RewardStatus.COMPLETED;
        this.completedAt = LocalDateTime.now();
    }

    private void apply(RewardFailedEvent event) {
        this.status = RewardStatus.FAILED;
    }

    // Business methods

    public void markAsCompleted() {
        applyChange(new RewardCompletedEvent(this.rewardId));
    }

    public void markAsFailed(String reason) {
        applyChange(new RewardFailedEvent(this.rewardId, reason));
    }

    // Getters and other methods...

    public List<Object> getPendingEvents() {
        return pendingEvents;
    }

    public void clearPendingEvents() {
        pendingEvents.clear();
    }
}



2. Implementing the Event Store
We need a way to persist events and retrieve them for replaying.
2.1. Event Store Interface
package com.parkingexchange.reward.infrastructure.eventstore;

import java.util.List;
import java.util.UUID;

public interface EventStore {

    void saveEvents(UUID aggregateId, List<Object> events, int expectedVersion);

    List<Object> getEvents(UUID aggregateId);
}


2.2. Event Store Implementation
Implement the event store using your preferred database (e.g., PostgreSQL, MySQL).
Event Entity
package com.parkingexchange.reward.infrastructure.eventstore;

import jakarta.persistence.*;

import java.time.Instant;
import java.util.UUID;

@Entity
@Table(name = "event_stream")
public class EventEntity {

    @Id
    @GeneratedValue(strategy = GenerationType.IDENTITY)
    private Long id;

    private UUID aggregateId;
    private String eventType;

    @Lob
    private String eventData;

    private Instant occurredOn;

    private int version;

    // Getters and setters...

}


Event Store Implementation
package com.parkingexchange.reward.infrastructure.eventstore;

import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;
import org.springframework.stereotype.Repository;
import org.springframework.transaction.annotation.Transactional;

import java.util.List;
import java.util.UUID;
import java.util.stream.Collectors;

@Repository
public class JpaEventStore implements EventStore {

    private final EventEntityRepository eventRepository;
    private final ObjectMapper objectMapper;

    public JpaEventStore(EventEntityRepository eventRepository,
                         ObjectMapper objectMapper) {
        this.eventRepository = eventRepository;
        this.objectMapper = objectMapper;
    }

    @Override
    @Transactional
    public void saveEvents(UUID aggregateId, List<Object> events, int expectedVersion) {
        int currentVersion = eventRepository.countByAggregateId(aggregateId);
        if (currentVersion != expectedVersion) {
            throw new RuntimeException("Concurrency exception");
        }

        int version = expectedVersion;
        for (Object event : events) {
            version++;
            EventEntity eventEntity = new EventEntity();
            eventEntity.setAggregateId(aggregateId);
            eventEntity.setEventType(event.getClass().getName());
            try {
                eventEntity.setEventData(objectMapper.writeValueAsString(event));
            } catch (JsonProcessingException e) {
                throw new RuntimeException("Failed to serialize event", e);
            }
            eventEntity.setOccurredOn(Instant.now());
            eventEntity.setVersion(version);
            eventRepository.save(eventEntity);
        }
    }

    @Override
    public List<Object> getEvents(UUID aggregateId) {
        List<EventEntity> eventEntities = eventRepository.findByAggregateIdOrderByVersion(aggregateId);
        return eventEntities.stream()
                .map(entity -> {
                    try {
                        return objectMapper.readValue(entity.getEventData(),
                                Class.forName(entity.getEventType()));
                    } catch (Exception e) {
                        throw new RuntimeException("Failed to deserialize event", e);
                    }
                })
                .collect(Collectors.toList());
    }
}


Event Repository
package com.parkingexchange.reward.infrastructure.eventstore;

import org.springframework.data.jpa.repository.JpaRepository;

import java.util.List;
import java.util.UUID;

public interface EventEntityRepository extends JpaRepository<EventEntity, Long> {

    List<EventEntity> findByAggregateIdOrderByVersion(UUID aggregateId);

    int countByAggregateId(UUID aggregateId);
}



3. Adjusting the Reward Application Service
Modify the RewardApplicationService to work with the event store.
package com.parkingexchange.reward.application.service;

import com.parkingexchange.reward.application.dto.RewardResponseDto;
import com.parkingexchange.reward.application.mapper.RewardMapper;
import com.parkingexchange.reward.domain.model.Reward;
import com.parkingexchange.reward.domain.repository.RewardRepository;
import com.parkingexchange.reward.infrastructure.eventstore.EventStore;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;

import java.math.BigDecimal;
import java.util.List;
import java.util.UUID;

@Service
public class RewardApplicationServiceImpl implements RewardApplicationService {

    private final RewardRepository rewardRepository;
    private final RewardCalculationService rewardCalculationService;
    private final RewardProcessingService rewardProcessingService;
    private final RewardMapper rewardMapper;
    private final EventStore eventStore;

    public RewardApplicationServiceImpl(
            RewardRepository rewardRepository,
            RewardCalculationService rewardCalculationService,
            RewardProcessingService rewardProcessingService,
            RewardMapper rewardMapper,
            EventStore eventStore
    ) {
        this.rewardRepository = rewardRepository;
        this.rewardCalculationService = rewardCalculationService;
        this.rewardProcessingService = rewardProcessingService;
        this.rewardMapper = rewardMapper;
        this.eventStore = eventStore;
    }

    @Transactional
    @Override
    public RewardResponseDto processReward(UUID availabilityId, UUID fromUserId, UUID toUserId) {
        try {
            // Check if reward for this availabilityId already exists
            Reward reward = rewardRepository.findByAvailabilityId(availabilityId);
            if (reward != null) {
                return rewardMapper.toDto(reward);
            }

            // Calculate reward amount
            BigDecimal amount = rewardCalculationService.calculateRewardAmount(availabilityId);

            // Create reward
            reward = Reward.create(fromUserId, toUserId, availabilityId, amount);

            // Save events to event store
            eventStore.saveEvents(reward.getRewardId(), reward.getPendingEvents(), 0);
            reward.clearPendingEvents();

            // Process payment
            boolean success = rewardProcessingService.processPayment(reward);

            if (success) {
                reward.markAsCompleted();
            } else {
                reward.markAsFailed("Payment processing failed");
            }

            // Save new events
            eventStore.saveEvents(reward.getRewardId(), reward.getPendingEvents(), 1);
            reward.clearPendingEvents();

            // Update projections
            rewardRepository.save(reward);

            return rewardMapper.toDto(reward);

        } catch (Exception e) {
            throw e; // handle exception appropriately
        }
    }
}



4. Updating the Reward Repository
The repository now acts as a projection of the event store for querying.
package com.parkingexchange.reward.domain.repository;

import com.parkingexchange.reward.domain.model.Reward;

import java.util.UUID;

public interface RewardRepository {

    Reward save(Reward reward);

    Reward findByRewardId(UUID rewardId);

    Reward findByAvailabilityId(UUID availabilityId);
}



5. Implementing the Projection
Implement a projection that rebuilds the Reward read model from events.
package com.parkingexchange.reward.infrastructure.repository;

import com.parkingexchange.reward.domain.model.Reward;
import com.parkingexchange.reward.domain.repository.RewardRepository;
import com.parkingexchange.reward.infrastructure.eventstore.EventStore;
import org.springframework.stereotype.Repository;

import java.util.List;
import java.util.UUID;

@Repository
public class RewardRepositoryImpl implements RewardRepository {

    private final EventStore eventStore;
    private final JpaRewardProjectionRepository rewardProjectionRepository;

    public RewardRepositoryImpl(EventStore eventStore,
                                JpaRewardProjectionRepository rewardProjectionRepository) {
        this.eventStore = eventStore;
        this.rewardProjectionRepository = rewardProjectionRepository;
    }

    @Override
    public Reward save(Reward reward) {
        // The reward is already persisted via events
        // Here, we update the projection for faster queries
        RewardProjection projection = new RewardProjection(reward);
        rewardProjectionRepository.save(projection);
        return reward;
    }

    @Override
    public Reward findByRewardId(UUID rewardId) {
        List<Object> events = eventStore.getEvents(rewardId);
        if (events.isEmpty()) {
            return null;
        }
        return new Reward(events);
    }

    @Override
    public Reward findByAvailabilityId(UUID availabilityId) {
        RewardProjection projection = rewardProjectionRepository.findByAvailabilityId(availabilityId);
        if (projection == null) {
            return null;
        }
        return findByRewardId(projection.getRewardId());
    }
}


Reward Projection Entity
package com.parkingexchange.reward.infrastructure.repository;

import com.parkingexchange.reward.domain.model.Reward;
import com.parkingexchange.reward.domain.model.RewardStatus;
import jakarta.persistence.*;

import java.math.BigDecimal;
import java.time.LocalDateTime;
import java.util.UUID;

@Entity
@Table(name = "reward_projection")
public class RewardProjection {

    @Id
    @Column(name = "reward_id")
    private UUID rewardId;

    @Column(name = "from_user_id")
    private UUID fromUserId;

    @Column(name = "to_user_id")
    private UUID toUserId;

    @Column(name = "availability_id")
    private UUID availabilityId;

    private BigDecimal amount;

    @Enumerated(EnumType.STRING)
    private RewardStatus status;

    private LocalDateTime createdAt;

    private LocalDateTime completedAt;

    protected RewardProjection() {
        // JPA constructor
    }

    public RewardProjection(Reward reward) {
        this.rewardId = reward.getRewardId();
        this.fromUserId = reward.getFromUserId();
        this.toUserId = reward.getToUserId();
        this.availabilityId = reward.getAvailabilityId();
        this.amount = reward.getAmount();
        this.status = reward.getStatus();
        this.createdAt = reward.getCreatedAt();
        this.completedAt = reward.getCompletedAt();
    }

    // Getters and setters...

}


Reward Projection Repository
package com.parkingexchange.reward.infrastructure.repository;

import org.springframework.data.jpa.repository.JpaRepository;

import java.util.UUID;

public interface JpaRewardProjectionRepository extends JpaRepository<RewardProjection, UUID> {

    RewardProjection findByAvailabilityId(UUID availabilityId);

}



6. Updating the Reward Processing Service
This service remains largely the same but ensure any state changes emit events.
package com.parkingexchange.reward.application.service;

import com.parkingexchange.reward.domain.model.Reward;
import org.springframework.stereotype.Service;

@Service
public class RewardProcessingServiceImpl implements RewardProcessingService {

    @Override
    public boolean processPayment(Reward reward) {
        // Implement payment processing logic

        // For the purposes of this example, we'll assume payment succeeds
        return true;
    }
}



7. Listening to Events from Parking Service
Implement a listener to handle SpotTakenConfirmedEvent.
Event Listener
package com.parkingexchange.reward.infrastructure.messaging;

import com.parkingexchange.reward.application.service.RewardApplicationService;
import com.parkingexchange.reward.domain.event.SpotTakenConfirmedEvent;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;

import java.util.UUID;

@Component
public class SpotTakenConfirmedEventListener {

    private final RewardApplicationService rewardApplicationService;

    public SpotTakenConfirmedEventListener(RewardApplicationService rewardApplicationService) {
        this.rewardApplicationService = rewardApplicationService;
    }

    @KafkaListener(topics = "spot-taken-confirmed-topic", groupId = "reward-group")
    public void handleSpotTakenConfirmedEvent(SpotTakenConfirmedEvent event) {
        UUID availabilityId = event.getAvailabilityId();
        UUID arrivingUserId = event.getArrivingUserId();
        UUID leavingUserId = event.getLeavingUserId();

        rewardApplicationService.processReward(availabilityId, arrivingUserId, leavingUserId);
    }
}



8. Configuring Event Serialization
Use spring-kafka along with JSON serialization for events.
Configure Kafka Consumer
spring:
  kafka:
    consumer:
      bootstrap-servers: localhost:9092
      group-id: reward-group
      key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
      value-deserializer: org.springframework.kafka.support.serializer.JsonDeserializer
      properties:
        spring.json.trusted.packages: com.parkingexchange.*


Configure Kafka Producer (for any outgoing events)
spring:
  kafka:
    producer:
      bootstrap-servers: localhost:9092
      key-serializer: org.apache.kafka.common.serialization.StringSerializer
      value-serializer: org.springframework.kafka.support.serializer.JsonSerializer



9. Maintaining All Functionalities
By switching to Event Sourcing:
State Changes via Events: All changes to the Reward aggregate are represented as events.
Rebuilding State: The Reward aggregate can be reconstructed by replaying events.
Querying Data: Use projections (the RewardProjection entity) to support queries efficiently without compromising performance.
Inter-Service Communication: Continue to listen to events from the Parking service to trigger reward processing.

10. Ensuring Production Readiness
Concurrency Handling: Implement optimistic concurrency control in the event store using event versioning.
Event Serialization: Use reliable serializers (e.g., Jackson) and handle versioning of events.
Error Handling: Wrap operations in transactions and implement proper exception handling.
Monitoring: Use monitoring tools to track event processing and identify bottlenecks.
Testing: Write unit and integration tests to ensure correctness of event sourcing implementation.

Final Notes
By applying Event Sourcing to the Reward application:
Consistency is Maintained: Events are the single source of truth, ensuring consistency.
Functionalities are Preserved: All existing functionalities are maintained with the new architecture.
Scalability: Event Sourcing can enhance scalability and performance, especially for complex domains.
Auditability: Complete history of state changes is available for audit purposes.

Please note: The provided code snippets focus on the key aspects of implementing Event Sourcing in your Reward application. Depending on your specific setup and additional functionalities, you may need to adjust configurations, add necessary imports, and handle additional edge cases.

Let me know if you need further clarification or assistance with any part of the implementation.
what are the advantages of using an event store
Advantages of Using an Event Store
An event store is a database optimized for storing events in an append-only fashion. When implementing Event Sourcing, an event store becomes the primary source of truth for the system's state. Utilizing an event store offers numerous advantages, especially in applications that benefit from capturing state changes as a sequence of events.

1. Complete Audit Trail
An event store records every change to the system state as an immutable event, creating a comprehensive audit log.
Traceability: Every action taken in the system is recorded, allowing for complete traceability of operations.
Compliance: Essential for industries where regulatory compliance requires detailed records (e.g., finance, healthcare).
Debugging: Facilitates pinpointing when and why a particular state change occurred.
2. Event-Driven Architecture Support
Event stores are integral to event-driven architectures, promoting decoupled and scalable systems.
Loose Coupling: Services communicate via events, reducing direct dependencies.
Asynchronous Processing: Events enable asynchronous workflows, improving responsiveness.
Scalability: Easily scales with the volume of events without impacting performance.
3. State Reconstruction and Temporal Queries
By storing events, the system can reconstruct its state at any point in time.
State Reconstruction: Rebuild the current state by replaying events from the beginning or a snapshot.
Time Travel: Inspect the state of the system at any historical point for analysis or debugging.
What-If Scenarios: Replay events in different orders or apply alternative business logic for testing.
4. Flexibility and Adaptability
Event sourcing with an event store provides flexibility in adapting to new requirements.
New Projections: Create new read models or projections without changing the event store.
Feature Evolution: Implement new features that leverage historical data without altering existing data structures.
Event Versioning: Supports evolving event schemas over time while maintaining backward compatibility.
5. Improved Reliability and Consistency
The use of an event store enhances the reliability of state changes.
Atomic Operations: Writing events is an atomic operation, ensuring consistency.
Immutability: Immutable events prevent accidental overwrites or deletions, maintaining data integrity.
Failure Recovery: In case of failures, the system can recover by replaying events.
6. Enhanced Integration Capabilities
Event stores facilitate integration with other systems and services.
Event Publishing: Acts as a source of events that can be published to message brokers for other services to consume.
Reactive Extensions: External systems can react to events without impacting the core application.
Audit Feeds: Provide external auditors or monitoring tools with real-time or batch event data.
7. Performance Optimization
Event stores can optimize performance for specific workloads.
Write Performance: Append-only operations are fast and efficient.
Read Models: Use projections to create denormalized views optimized for read performance.
Horizontal Scaling: Scale out by partitioning events, for instance, by aggregate IDs.
8. Simplified Concurrency Handling
Event sourcing simplifies dealing with concurrent updates.
Optimistic Concurrency: Versioning of events allows for optimistic concurrency control, reducing locking overhead.
Conflict Resolution: Conflicts can be detected and handled at the event level, providing more granular control.
9. Facilitating Analytics and Machine Learning
Historical event data is valuable for analytics.
Data Mining: Analyze sequences of events to identify patterns or trends.
Machine Learning: Use event data to train models for predictions or anomaly detection.
Business Insights: Gain insights into user behavior and system usage over time.
10. Error Correction Without Data Loss
Mistakes can be corrected by appending compensating events.
Non-Destructive Fixes: Instead of deleting or altering incorrect data, append events that correct the state.
Audit Integrity: Maintains the integrity of the audit trail by preserving all original events.

In the Context of Your Reward Application
Applying these advantages to your Reward application:
Reliable Reward Processing: Ensures that reward transactions are consistently recorded and can be audited.
Eventual Consistency with Other Services: Harmonizes state between the Reward and Parking services through event-driven updates.
Enhanced Debugging: Facilitates troubleshooting reward-related issues by reviewing the sequence of events leading to a state.
Future Extensibility: Allows adding new features that leverage historical reward events without significant database schema changes.
Improved Integration: Other services can consume reward events for purposes like notifications, reporting, or additional business logic.

Conclusion
Using an event store as part of an event-sourced architecture provides numerous benefits, from maintaining a complete and tamper-proof audit trail to enhancing scalability and adaptability. For applications where understanding the sequence of state changes is critical, such as financial transactions, inventory management, or, in your case, reward processing, an event store offers a robust and flexible foundation.

Considerations
While there are significant advantages, it's also important to be aware of potential challenges:
Complexity: Event sourcing can introduce complexity in understanding and implementing the architecture.
Storage Requirements: Over time, the event store can grow large, requiring strategies for archives or snapshots.
Learning Curve: Developers and team members may need training to work effectively with event-sourced systems.
Planning and addressing these considerations will help you fully leverage the benefits of an event store in your application.




I don't undestand how reward projection works, when do I look at the projection and when i look at the events?