CQRS & Event Sourcing

Event Sourcing, Stream processing, CQRS: what are those buzzwords? How my app can benefit from it? This talk aims to clarify these concepts through examples in PHP, and see why this totally changes how to structure the data. We will see how to take full advantage of the philosophy of NoSQL to build a scalable & robust app which is will be easier to maintain.

Transcript

1. Event Sourcing @MattKetmo CQRS & #PHPCon 2015
2. How can it help to maintain your app? How can it help to scale horizontally?
3. Matthieu Moquet @MattKetmo @ BlaBlaCar for 4 years
4. {Disclaimer Oriented for long lived apps.
 Overkill for RAD / CRUD / TODO apps. Use those patterns in moderation.
 There is no silver bullet. This talk aims to open your mind.
 Think different!
5. I/O
6. 90% ViewModels / Flexible Eventually consistent 10% Validation / Business rules Coherence (ACID) Read Write Fast & Scale
7. Use case bank accounts Account #1 1337€ Account #2 42€ Account #3 10€ + Transfer money
8. Facade (Controller) Models (ORM) DB
9. //  Open  new  account   $account  =  new  Account();   $account-­‐>setAmount(0,  'EUR');   $account-­‐>setUser($user);   $em-­‐>persist($account);   $em-­‐>flush();   //  Credit  100  EUR   $account-­‐>setAmount(100,  'EUR');   $em-­‐>flush();   //  Debit  20  EUR   $account-­‐>setAmount(80,  'EUR');   $em-­‐>flush();
10. does not scale well CRUD
11. Facade (Controller) Models (ORM) DB Service Layer (Business)
12. interface  BankAccountService     {          /**  @return  int  */          public  function  open();          /**  @return  Account  */          public  function  get($accountId);          /**  @return  Account[]  */          public  function  findAll($userId);          /**  @return  void  */          public  function  credit($accountId,  $balance);          /**  @return  void  */          public  function  debit($accountId,  $balance);   }  
13. CQS Command Query Separation }Each operation is either a Command or a Query, not both
14. Commands Queries Change the state of the system. Do not return anything. Does not change the state of the system. Return data. Write only Read only
15. Commands /**  @return  Account  */   public  function  get($accountId);   /**  @return  Account[]  */   public  function  findAll($userId); /**  @return  int  */   public  function  open();   /**  @return  void  */   public  function  credit($accountId,  $balance);   /**  @return  void  */   public  function  debit($accountId,  $balance); Should be void Queries
16. /**      *  @param  Uuid  $accountId    *    *  @return  void      */   public  function  open(Uuid  $accountId);   Let the client generate the identifier, not the infrastructure (database)
17. CQRS Command Query Responsibility Segregation }Like CQS but separate queries and commands in different classes
18. CQRS challenges the assumption that reading and writing are sharing the same abstractions Databases Models Apps
19. Command Facade Query Facade Command Handler Query Repository Command Models DB Read Models Command DTO Segregation of read and write is a radical form of decoupling
21. Now that the workflow of Read & Write has been separated. We can maintain & optimize each side separately.
22. asynchronously Since Command Handlers do NOT return anything, 
 then do the execution
23. Command Facade Command Handlers Commands Queueing Ack
24. $accoundId  =  AccountId::generate();   $command  =  new  OpenAccount($accountId);   if  (/*  user  is  allowed  to  open  account  */)  {          $commandBus-­‐>handle($command);          return  new  Response(                  null,                  $async  ?  202  :  201,  //  Accepted  :  Created                  ['Location'  =>  "/accounts/$accountId"]          );     }
25. $command  =  new  Debit(          $accountId,            new  Money(20,  'EUR')   );   if  (/*  account  has  enough  money  */)  {          $commandBus-­‐>handle($command);          return  new  Response(null,  204);   }
26. Command tracking Scale workers Horizontally by instantiating more workers (subscribers) Generate an identifier per command to let the client tracks its status Rabbitmq, Kafka, Gearman, ……… make your choice on queues.io
27. Projections On the Read side you should prepare you data as it will be requested
28. Having only 1 data store for reads & writes does not scale (well) different usage different needs
29. ………and use different data stores Denormalize your data... UI View models Statistic models Search index models API Read models ... Read models are faster than JOIN different usage different needs
31. {      "user_id":  133742      "from":  "Paris",      "to":  "Luxembourg",      "by":  [          "Reims",            "..."      ],      "date":  "2015-­‐05-­‐11",   } Input (mysql)
32. [{      "trip":  {          "from":  "Paris",          "to":  "Luxembourg",          "date":  "2015-­‐05-­‐11",          "..."      },      "user":  {          "name":  "John  D",          "age":  29,          "grade":  "beginner",          "..."      }   }] Output (Elasticsearch) denormalized searchable
34. SELECT  tweets.*,  users.*      FROM  tweets   JOIN  users      ON  users.id  =  tweet.sender_id   JOIN  follows      ON  follows.followee_id  =  user.id   WHERE  follows.follower_id  =  $userId   ORDER  BY  tweets.time  DESC   LIMIT  100
36. Tweets stream DB Aggs Timelines {      "user_id":  1234567890      "status":  "Hello  World"      "timestamp":  1430491773   } [{      "tweet_id":  1234567890876543,      "username":  "MattKetmo",      "name":  "Matthieu  Moquet",      "timestamp":  1430491773,      "status":  "Hello  World",   },  {      "tweet_id":  1234567890886445,      ...   }]  
37. SELECT  tweets      FROM  timelines    WHERE  timeline_id  =  $userId    LIMIT  500
38. PageViewEvents DB Aggs Increment counters /month Google Analytics /day /hour total {      "eventType":  "PageViewEvent"      "timestamp":  1430491773,      "ipAddress":  "12.34.56.78",      "sessionId":  "abcd1234567890",      "pageUrl":  "/hello-­‐world",      "..."   }
39. Using Cassandra you need to: ‣ know the read requests before creating your data models ‣ create as many tables (ie. KeySpaces) than you have views ‣ denormalize the data (no join allowed) ’s keypoints
40. C* C* C* Primary Data C* Read Data User post message Endpoint
41. Multi Data Center Replication + Sharding + Cache
42. Command Command Handler Primary DB Projections View Models Message Bus Write Read Eventual Consistency {...} {...} {...} {...}
43. Event Sourcing change the way we store our primary data
44. Event Sourcing Register only a series of events. Reconstitute the state of current "entity" by reading the past events. If we know the events of the past we can reconstitute the present
45. the standard way Something happen State A State B Something happen (delta) (dropped) (stored) (delta) Time
46. the event sourcing way Something happen State A State B Something happen (stored) (reconstituted) (reconstituted) (stored) Time
47. [{        "uuid":  "110e8400-­‐e29b-­‐11d4-­‐a716-­‐446655440000",        "type":  "AccountWasOpen",        "recorded_on":  "2015-­‐05-­‐11T13:37:00Z",        "payload":  {}   },  {        "uuid":  "110e8400-­‐e29b-­‐11d4-­‐a716-­‐446655440000",        "type":  "AccountWasCredited",        "recorded_on":  "2015-­‐05-­‐11T14:42:00Z",        "payload":  {  "amount":  100,  "currency":  "EUR"  }   },  {      "..."   }] Event Store
48. Events are the of the system
49. Aggregates Isolated domain models which run the business logic (via an ubiquitous language) through events
50. //  Open  a  new  account   $accountId  =  AccountId::generate();   $account  =  BankAccount::open($accountId);   //  Add  some  money   $account-­‐>credit(new  Money(100,  'EUR'));
51. Records events Aggregates class  BankAccount  implements  AggregateRootInterface   {          //  ...   }
52. interface  AggregateRoot   {          /**            *  @return  DomainEventStream            */          public  function  getUncommittedEvents();          /**            *  @return  string            */          public  function  getAggregateRootId();   }
53. $accountId  =  AccountId::generate();   $account  =  BankAccount::open($accountId);   $account-­‐>credit(new  Money(100,  'EUR'));   //  Retrieve  event  stream   $events  =  $account-­‐>getUncommittedEvents();   //  -­‐  AccountWasOpen   //  -­‐  AccountWasCredited
54. class  BankAccount  extends  EventSourcedAggregateRoot   {          private  $accountId;          public  function  getAggregateRootId()          {                  return  $this-­‐>accountId;          }          public  static  function  open(AccountId  $accountId)          {                  $account  =  new  self();                  $account-­‐>apply(new  AccountWasOpen($accountId));                  return  $account;          }          public  function  credit(Money  $balance)          {                  $this-­‐>apply(new  AccountWasCredited(                          $this-­‐>accountId,                          $balance-­‐>getAmount(),                          $balance-­‐>getCurrency()                  );          }   }
55. Reconstitute from events Aggregates
56. class  BankAccount  extends  EventSourcedAggregateRoot   {          private  $amount;          //  ...              protected  function  applyAccountWasOpen(AccountWasOpen  $event)          {                  $this-­‐>accountId  =  $event-­‐>getAccountId();          }          protected  function  applyAccountWasCredited(AccountWasCredited  $event)          {                  $this-­‐>amount  +=  $event-­‐>getBalance();          }          protected  function  applyAccountWasDebited(AccountWasDebited  $event)          {                  $this-­‐>amount  -­‐=  $event-­‐>getBalance();          }   }
57. Aggregates Protect invariants
58. class  BankAccount  extends  EventSourcedAggregateRoot   {          //  ...          public  function  debit(Money  $balance)          {                  if  ($this-­‐>amount  <  $balance-­‐>getAmount())  {                          throw  new  NotEnoughMoneyException(                                  'Cannot  debit  more  than  current  amount'                          );                  }                  return  $this-­‐>apply(new  AccountWasCredited(                          $this-­‐>accountId,                          $balance-­‐>getAmount(),                          $balance-­‐>getCurrency()                  ));                        }   }
59. Command + Event Sourcing
60. Command User intent Events Event Store Apply changes on system Primary DB Transactional append
61. class  BankAccountCommandHandler  extends  CommandHandler   {          public  function  handleOpenAccount(OpenAccount  $command)          {                  $account  =  BankAccount::open($command-­‐>getAccountId());                  $this-­‐>repository-­‐>save($account);          }          public  function  handleCreditAccount(CreditAccount  $command)          {                  $account  =  $this-­‐>repository-­‐>load($command-­‐>getAccountId());                  $account-­‐>credit($command-­‐>getBalance());                  $this-­‐>repository-­‐>save($account);          }   }
62. class  EventSourcingRepository  implements  RepositoryInterface   {          public  function  save(AggregateRoot  $aggregate)          {                  $events  =  $aggregate-­‐>getUncommittedEvents();                  $this-­‐>eventStore-­‐>append(                          $aggregate-­‐>getAggregateRootId(),                            $events                  );          }          public  function  load($id)          {                  try  {                          $events  =  $this-­‐>eventStore-­‐>load($id);                          return  $this-­‐>aggregateFactory-­‐>create($events);                  }  catch  (EventStreamNotFoundException  $e)  {                          throw  AggregateNotFoundException::create($id,  $e);                  }          }   }
63. Event Sourcing PROS
64. Append Only Very fast data update Deleting data = new events
65. Immutable Infinite caching
66. No Database migration when updating the data DBA will love it
67. Sharding Aggregates are self-reliant and don’t have external references No Joins
68. Complete historical data Being able to replay history is a major benefit both technically and for the business BI team will love it
69. Event Sourcing CONS Well, not really
70. Ensure consistency How to avoid concurrent updates?
71. Load aggregate from EventStore: 100€ Create AccountWasDebited(100) event Append event in datastore Example Two debits of 100€ must not be accepted if only 100€ left Load aggregate from EventStore: 100€ Create AccountWasDebited(100) event Append event in datastore Event must NOT be appended twice
72. abstract  class  EventSourcedAggregateRoot  implements  AggregateRootInterface   {          private  $uncommittedEvents  =  array();          private  $playhead  =  -­‐1;            /**          *  Applies  an  event.            *  The  event  is  added  to  the  list  of  uncommited  events.          */          public  function  apply($event)          {                  $this-­‐>playhead++;                  $this-­‐>uncommittedEvents[]  =  DomainMessage::recordNow(                          $this-­‐>getAggregateRootId(),                          $this-­‐>playhead,                          new  Metadata(array()),                          $event                  );          }   }
73. DBAL EventStore (MySQL) CREATE  TABLE  EventStore  (      `id`                    INT(11)  AUTO_INCREMENT,      `uuid`                BINARY(16),      `playhead`        INT(11),      `type`                TEXT,      `payload`          TEXT,      `metadata`        TEXT,      `recorded_on`  DATETIME,      PRIMARY  KEY  (`id`),      UNIQUE  KEY  `UNIQ_PLAYHEAD`  (`uuid`,  `playhead`)   );
74. E_TOO_MANY_EVENTS When your Aggregates are "long lived" it may be slow to read the full history eg. BankAccount
75. Snapshotting is a solution
76. Snapshotting Store the state of the aggregate and read only the events created after
77. Snapshotting Snapshot Snapshot Events States
78. TL;DR
79. Command Handler Events Event Store Command Auditing Read Model Event Listener Event Bus Projector Write Read Business rules Historical data Read / search index Side effects User intent View models
80. Frameworks C# / Java nCQRS Fohjin NEventStore LiteCQRS Lokad.CQRS Agr.CQRS Axon Framework JDON PHP Broadway Predaddy Prooph EventCentric.Core litecqrs-php ...?
81. qandidate-labs/broadway qandidate-labs/broadway-demo
82. cqrs.nu #MustRead
83. dddinphp.org Google Groups
84. Thank You! Slides available at moquet.net/talks/phpcon-2015 Leave feedbacks @MattKetmo