I'm aware that this question has been asked several times, but I have some concerns regarding querying from the write side that I don't see addressed in the already existing questions, more especifically regarding eventual consistency in the command model.
I have a simple CQRS + ES architecture for an application. Customers can buy stuff from my site, but there is a hardcoded requirement: A customer cannot purchase more than 500$ of products from our store. If they try, the purchase should not be accepted.
So, this is how my command handler looks like (in python, and simplified from concerns like currencies, injection for the sake of simplicity):
class NewPurchaseCommand:
customer_id: int
product_ids: List[int]
class PurchasesCommandHandler:
purchase_repository: PurchaseRepository
product_repository: ProductRepository
customer_query_service: CustomerQueryService
def handle(self, cmd: NewPurchaseCommand):
current_amount_purchased = self.customer_query_service.get_total(cmd.customer_id)
purchase_amount = 0
for product_id in cmd.product_ids:
product = self.product_repository.get(product_id)
purchase_amount += product.amount
if current_amount_purchase + purchase_amount > 500:
raise Exception('You cannot purchase over 500$')
new_purchase = Purchase.create(cmd.customer_id, cmd.product_ids)
self.purchase_repository.save(new_purchase)
# Then, after the purchase is saved, a PurchaseCreated event is persisted,
# sent to a queue which will then update several read projections, which one
# of them is the underlying table that the customer_query_service uses.
The CustomerQueryService uses an underlying table to quickly retrieve the amount that the user has purchased at the time being, and this table is exclusively used by the write side, and updated eventually:
CustomerPurchasedAmount table
CustomerId | Amount
10 | 480
While my command handler works on simple scenarios, I want to know how to handle the possible edge cases that might happen:
- This user 10, which is a malicious one, makes two purchases at the same time of 20$. But since the CustomerPurchasedAmount table is updated eventually, both requests will succeed (this is the case I'm most concerned)
- There might exist the possibility that some product price might change while processing the request (unlikely, but then again, it can happen).
My questions are:
- How can I protect the command from the concurrency case exposed before?
- How should read models specifically tailored for the write side be updated? Synchronously? Asynchronously like I'm doing right now?
- And in general, how should command validation happen if the information you are querying in order to validate might be stale?