Event Sourcing Made Simple

Philippe Creux – RailsConf 2019

pcreux.com

1 / 93

Let's build a TODO™ app together🤓2 / 93

rails new todoz3 / 93

class Task < ApplicationRecord
end

4 / 93

class TasksController < ApplicationController
  def index
    # ...
  end
  def show
    # ...
  end
  def create
    # ...
  end
end

5 / 93

✅ Complete a task6 / 93

✅ Complete a task

class TasksController < ApplicationController
  # ...
  def complete
    task = Task.find(params[:id])
    task.update!(completed: true)
    render json: task
  end
  # ...
end

7 / 93

Let's record the completion date8 / 93

Let's record the completion date

class TasksController < ApplicationController
  # ...
  def complete
    task = Task.find(params[:id])
    task.update!(completed_at: Time.now)
    render json: task
  end
  # ...
end

9 / 93

Let's record the completion date

  class AddCompletedAtToTasks < ActiveRecord::Migration[6.0]
    def up
      add_column 'tasks', :completed_at, :datetime
      Task.where(completed: true).find_each do |task|
        task.update!(completed_at: ???)
      end
      remove_column 'tasks', :completed
    end
  end

10 / 93

I want to know who completed a given task11 / 93

I want to know who completed a given task

class TasksController < ApplicationController
  # ...
  def complete
    task = Task.find(params[:id])
    task.update!(completed_at: Time.now, completed_by: current_user)
    render json: task
  end
  # ...
end

12 / 93

I want to know who completed a given task

  class AddCompletedByToTasks < ActiveRecord::Migration[6.0]
    def up
      add_column 'tasks', :completed_by_id, :integer
      Task.completed.find_each do |task|
        task.update!(completed_by: ???)
      end
    end
  end

13 / 93

Let's add some event tracking

class TasksController < ApplicationController
  # ...
  def complete
    task = Task.find(params[:id])
    task.update!(completed_at: Time.now, completed_by: current_user)
    track_event("Task completed", task_id: task.id, user_id: current_user.id)
    render json: task
  end
  # ...
end

14 / 93

... and email notifications

class TasksController < ApplicationController
  # ...
  def complete
    task = Task.find(params[:id])
    task.update!(completed_at: Time.now, completed_by: current_user)
    if task.created_by != current_user
      TaskMailer.queue_task_completed(task: task, recipient: task.created_by)
    end
    track_event("Task completed", task_id: task.id, user_id: current_user.id)
    render json: task
  end
  # ...
end

15 / 93

... and an activity feed

class TasksController < ApplicationController
  # ...
  def complete
    task = Task.find(params[:id])
    task.update!(completed_at: Time.now, completed_by: current_user)
    if task.created_by != current_user
      TaskMailer.queue_task_completed(task: task, recipient: task.created_by)
    end
    Activity.create!(subject: task, who: current_user, action: :completed)
    track_event("Task completed", task_id: task.id, user_id: current_user.id)
    render json: task
  end
  # ...
end

16 / 93

👋

Philippe Creux

17 / 93

Kickstarter18 / 93

Kickstarter is turning 10 today!🎂19 / 93

🎂 Kickstarter

162,000+ creative projects brought to life
thanks to 16,000,000+ backers
who've raised $4,250,000,000+

20 / 93

Event Sourcing Made Simple

Philippe Creux – RailsConf 2019

pcreux.com

21 / 93

What is Event Sourcing?22 / 93

The application state is the result of a sequence of events

Application State

Events

23 / 93

Application State

Events

Task Created
- title: "Prepare sides"

24 / 93

Application State

□ Task #123: "Prepare sides"

Events

Task Created
- title: "Prepare sides"

25 / 93

Application State

□ Task #123: "Prepare sides"

Events

Task Created
- title: "Prepare sides"
Task Due Date Set
- date: "2019-04-20"

26 / 93

Application State

□ Task #123: "Prepare sides"
  Due April 20th, 2019.

Events

Task Created
- title: "Prepare sides"
Task Due Date Set
- date: "2019-04-20"

27 / 93

Application State

□ Task #123: "Prepare sides"
  Due April 20th, 2019.

Events

Task Created
- title: "Prepare sides"
Task Due Date Set
- date: "2019-04-20"
Task Title Updated
- title: "Prepare slides"

28 / 93

Application State

□ Task #123: "Prepare slides"
  Due April 20th, 2019.

Events

Task Created
- title: "Prepare sides"
Task Due Date Set
- date: "2019-04-20"
Task Title Updated
- title: "Prepare slides"

29 / 93

Application State

□ Task #123: "Prepare slides"
  Due April 20th, 2019.

Events

Task Created
- title: "Prepare sides"
Task Due Date Set
- date: "2019-04-20"
Task Title Updated
- title: "Prepare slides"
Task Assigned
- assignee: "Philippe"

30 / 93

Application State

□ Task #123: "Prepare slides"
  Assigned to Philippe.
  Due April 20th, 2019.

Events

Task Created
- title: "Prepare sides"
Task Due Date Set
- date: "2019-04-20"
Task Title Updated
- title: "Prepare slides"
Task Assigned
- assignee: "Philippe"

31 / 93

Application State

□ Task #123: "Prepare slides"
  Assigned to Philippe.
  Due April 20th, 2019.

Events

Task Created
- title: "Prepare sides"
Task Due Date Set
- date: "2019-04-20"
Task Title Updated
- title: "Prepare slides"
Task Assigned
- assignee: "Philippe"
Task Due Date Set
- date: "2019-04-24"

32 / 93

Application State

□ Task #123: "Prepare slides"
  Assigned to Philippe.
  Due April 24th, 2019.

Events

Task Created
- title: "Prepare sides"
Task Due Date Set
- date: "2019-04-20"
Task Title Updated
- title: "Prepare slides"
Task Assigned
- assignee: "Philippe"
Task Due Date Set
- date: "2019-04-24"

33 / 93

Application State

□ Task #123: "Prepare slides"
  Assigned to Philippe.
  Due April 24th, 2019.

Events

Task Created
- title: "Prepare sides"
Task Due Date Set
- date: "2019-04-20"
Task Title Updated
- title: "Prepare slides"
Task Assigned
- assignee: "Philippe"
Task Due Date Set
- date: "2019-04-24"
Task Completed

34 / 93

Application State

✅ Task #123: "Prepare slides"
  Assigned to Philippe.
  Due April 24th, 2019.

Events

Task Created
- title: "Prepare sides"
Task Due Date Set
- date: "2019-04-20"
Task Title Updated
- title: "Prepare slides"
Task Assigned
- assignee: "Philippe"
Task Due Date Set
- date: "2019-04-24"
Task Completed

35 / 93

Nice Properties:

you can reconstruct the state at any point in time.
you can reconstruct the state up to a certain point in time. Time machine!
full history

Events Metadata

Task Created
- title: "Prepare sides"
- event.created_at: "2019-04-15"
- event.user: "Caroline"
- event.device: "Desktop"
- event.ip: "92.43.x.x"
Task Due Date Set
- date: "2019-04-20"
- event.created_at: "2019-04-15"
- event.device: "Desktop"
- event.ip: "92.43.x.x"

...

Task Completed
- event.created_at: "2019-04-29"
- event.user: "Philippe"
- event.device: "Pixel 3"
- event.ip: "82.23.x.x"

36 / 93

Events have metadata associated to them.
Thanks to the disconnect between the Application State Data and the Event Data we can store data that we don't need right now but might be useful in the future.

You use Event Sourcing everyday!37 / 93

git working directory is the current state commits are events ** travel back and forth, branch out (what if scenarios...?)
business people accounting ledger bank statement: current balance full history

Key Benefits38 / 93

Full history / audit logWe understand how we got there39 / 93

Make supports easier
Make debugging easier. ex: unbounce subscription

Replay events...to backfill a new attribute40 / 93

events has a lot of data and metadata
when adding a new attribute to the application state
you can replay events to backfill that attribute with data

Replay events...to recover lost records41 / 93

a record has been deleted by mistake?
you can replay events to get it back!

Replay events...to fix inconsistent state42 / 93

a bug in your code
a race condition
or a data migration that went bad can lead to an inconsistent state
by replaying events you can get back to a sane state

Replay events...to travel back in time43 / 93

replay events up to a certain point in time... and you have a time machine.

Forward events...to your datawarehouse44 / 93

events are first class citizens of you system
rather than writing custom code to send events to a datawarehouse
forward them as is

Forward events...to your monitoring solution45 / 93

track application level metrics
raise alerts when such metric (checkouts!) is low

Apply events...

to build tailored projections

ActivityFeed, Daily Report etc.

46 / 93

We call the Main representation of an entity an aggregate: Task, Subscription, User, Post
Other represenations that are tailored to a specific usage or a specific type of user. Great high read / fast read application

Existing event can be used to...backfill your datawarehouse tables47 / 93

Not only you can backfill new attributes / columns
you can backfill the datawarehouse tables as well
it's simple function to transform event into a datawarehouse event
so you can also re-ingest all the events if something went wrong in the transformation

Existing event can be used to...backfill your new ActivityFeed feature from Day 1.48 / 93

When you release a new feature such as an activity feed, you tend to only log activity items from the day that you released the feature...
or have to write custom code

Great fit for distributed systems49 / 93

evnent as source of truth - first class citizens are great for distributed systems
async communication
eventual consitency
micro services and the like.
But you don't have to have such an architecture to benefit from such an architecture.

Great fit for distributed systemsbut you don't have to build one50 / 93

Great fit for monoliths51 / 93

Event Sourcing at Kickstartermaking it simple52 / 93

https://d.rip

A subscription platform to support creators

53 / 93

Event Sourcing FrameworkRequirementsMust be simple to learn and use
Must be easy to remove
SolutionCustom "framework" ~150 LOC
Events are regular ActiveRecord models
Only change the way we mutate data
54 / 93

Let's build a TODO™ app together🤓55 / 93

✅ Complete a task

class TasksController < ApplicationController
  def complete
    event = Events::Task::Completed.create!(
      task: Task.find(params[:id]),
      metadata: { user_id: current_user&.id }
    )
    render json: event.task
  end
end

56 / 93

✅ Complete a task

class TasksController < ApplicationController
  def complete
    event = Events::Task::Completed.create!(
      task: Task.find(params[:id]),
      metadata: { user_id: current_user&.id }
    )
    render json: event.task
  end
end

class Events::Task::Completed < Events::Task::BaseEvent
  # task is an instance of a Task model (aka "aggregate")
  def apply(task)
    task.completed = true
    task
  end
end

57 / 93

✅ Complete a task

class TasksController < ApplicationController
  def complete
    event = Events::Task::Completed.create!(
      task: Task.find(params[:id]),
      metadata: { user_id: current_user&.id }
    )
    render json: event.task
  end
end

class Events::Task::Completed < Events::Task::BaseEvent
  # task is an instance of a Task model (aka "aggregate")
  def apply(task)
    task.completed = true
    task
  end
end

| task_events                                                      |
| ---------------------------------------------------------------- |
| id | task_id | type      | data | metadata          | created_at |
| ---------------------------------------------------------------- |
|  6 |      10 | Completed | {}   | { "user_id":  3 } | 2019-04-01 |

58 / 93

New requirement: record completion date

class TasksController < ApplicationController
  def complete
    event = Events::Task::Completed.create!(
      task: Task.find(params[:id]),
      metadata: { user_id: current_user&.id }
    )
    render json: event.task
  end
end

class Events::Task::Completed < Events::Task::BaseEvent
  def apply(task)
    task.completed_at = created_at
    task
  end
end

| task_events                                                       |
| ----------------------------------------------------------------- |
| id | task_id | type       | data | metadata          | created_at |
| ----------------------------------------------------------------- |
|  6 |      10 | Completed  | {}   | { "user_id":  3 } | 2019-04-01 |

59 / 93

Controller doesn't change.

New requirement: record completion date

class Events::Task::Completed < Events::Task::BaseEvent
  def apply(task)
    task.completed_at = created_at
    task
  end
end

  class AddCompletedAtToTasks < ActiveRecord::Migration[6.0]
    def up
      add_column 'tasks', :completed_at, :datetime
      # Replay events
      Task.lock.find_each do |id|
        task.events.reduce(task) do |task, event|
          event.apply(task)
        end
        task.save!
      end
      remove_column 'tasks', :completed
    end
  end

60 / 93

On a real / live system, this would be a 5 steps process:

add new column
make the event write to both the old and the new column
replay events
stop writing to the old column
remove the old column

New requirement: record who completed a task

class TasksController < ApplicationController
  def complete
    event = Events::Task::Completed.create!(
      task: Task.find(params[:id]),
      metadata: { user_id: current_user&.id }
    )
    render json: event.task
  end
end

class Events::Task::Completed < Events::Task::BaseEvent
  def apply(task)
    task.completed_at = created_at
    task.completed_by_id = metadata["user_id"]
    task
  end
end

| task_events                                                       |
| ----------------------------------------------------------------- |
| id | task_id | type       | data | metadata          | created_at |
| ----------------------------------------------------------------- |
|  6 |      10 | Completed  | {}   | { "user_id":  3 } | 2019-04-01 |

61 / 93

New requirement: record who completed a task

class Events::Task::Completed < Events::Task::BaseEvent
  def apply(task)
    task.completed_at = created_at
    task.completed_by_id = metadata["user_id"]
    task
  end
end

  class AddCompletedByIdToTasks < ActiveRecord::Migration[6.0]
    def up
      add_column 'tasks', :completed_by_id, :integer
      # Replay events
      Task.lock.find_each do |id|
        task.events.reduce(task) do |task, event|
          event.apply(task)
        end
        task.save!
      end
    end
  end

62 / 93

Controller doesn't change

Side-effects performed by Reactors63 / 93

Introducing Reactors

Reactors are triggered when an event is created
They can be associated with one or many event types
They can run synchronously or async

Usage:

They are used to trigger side effects
They can create other events in turn

64 / 93

Let's add email notifications

class Events::Dispatcher < BaseDispatcher
  # ...
  on Events::Task::Completed, trigger: Reactors::Task::NotifyCompleted
  # ...
end

module Reactors
  module Task
    NotifyCompleted = ->(event) do
      if event.user != event.task.author
        TaskMailer.queue_task_completed(task: event.task)
      end
    end
  end
end

65 / 93

... and some event tracking

class Events::Dispatcher < BaseDispatcher
  # ...
  on Events::Task::Completed, trigger: Reactors::Task::NotifyCompleted
  on Events::Task::Completed, trigger: Reactors::ForwardEvent
  # ...
end

module Reactors
  ForwardEvent = ->(event) do
    attributes = event.data.merge(event.metadata)
    EventBus.track_event(event.class.name, attributes)
  end
end

66 / 93

... and an activity feed

class Events::Dispatcher < BaseDispatcher
  # ...
  on Events::Task::Completed, trigger: Reactors::Task::NotifyCompleted
  on Events::Task::Completed, trigger: Reactors::ForwardEvent
  on Events::Task::Completed, trigger: Reactors::CreateActivityEntry
  # ...
end

module Reactors
  CreateActivityEntry = ->(event) do
    Activity.create!(
      subject: event.aggregate,
      user_id: event.metadata["user_id"],
      recorded_at: event.created_at,
      action: event.class.name.demodulize.underscore
    )
  end
end

67 / 93

Tracking all the things

# Inheritance
Events::BaseEvent <- Events::Task::BaseEvent <- Events::Task::Completed

class Events::Dispatcher < BaseDispatcher
  # ...
  on Events::Task::Completed, trigger: Reactors::Task::NotifyCompleted
  on Events::BaseEvent, trigger: Reactors::ForwardEvent
  on Events::Task::BaseEvent, trigger: Reactors::CreateActivityEntry
  # ...
end

68 / 93

Run reactors asynchronously via ActiveJob

class Events::Dispatcher < BaseDispatcher
  # ...
  on Events::Task::Completed, trigger: Reactors::Task::NotifyCompleted
  on Events::BaseEvent, async: Reactors::ForwardEvent
  on Events::Task::BaseEvent, async: Reactors::CreateActivityEntry
  # ...
end

69 / 93

70 / 93

71 / 93

72 / 93

1 year of Event Sourcing on https://d.rip

Code

15 aggregates
135 events
60 reactors

Data

~100k aggregate records
~500k event records

73 / 93

Might want to remove those.

Ah ha!74 / 93

Replaying events is awesome!75 / 93

Replaying events is awesome!To backfill new attributes76 / 93

subscription.deactivated_at
subscription.duration
device.type (ipad, iphone, etc)

Replaying events is awesome!To recover hard-destroyed records77 / 93

when we decided to switch from hard-destroy to soft-destroy we were able to recover to destroyed aggregates

Audit all the things!

Application state

Subscription #123 – Unpaid

Events

Event Type	Data	Created At
Unpaid		2019-03-06 22:09:24
FailedToCollect	invoice_key: 535	2019-03-06 22:09:22
FailedToCollect	invoice_key: 535	2019-03-05 22:06:15
FailedToCollect	invoice_key: 535	2019-03-02 22:01:04
PastDue		2019-03-02 22:00:48
InvoiceCreated	invoice_key: 535	2019-03-02 21:00:28
Collected	invoice_key: 234	2019-02-02 20:52:47
InvoiceCreated	invoice_key: 234	2019-02-02 20:52:46
Activated		2019-02-02 20:52:43
Created	stripe_key: 123	2019-02-02 20:52:35

78 / 93

Have a full audit-log such as this one is great for:

customer support
debugging on production
debugging in dev or test environment
track down race conditions

Ex: subscription status mismatch

Monitor all the things!

79 / 93

We use a combination of Statsd + Grafana + InfluxDB for monitoring.
All events are sent to Statsd so that
We can start monitoring any event - the data is there

Report on all the things!

80 / 93

We use a combination of Kinesis + Redshift + Looker for business analytics.
All events are sent to our Redshift cluster
started to send events to the Redshift Cluster 6 month after going private-beta
we were able to backfill the datawarehouse with all the events since the beginning of time

Side effects at a glance

81 / 93

DSL to connect reactors to events
Reactors only create other events in turn
All that chain is logged in events metadata so you can trace the chain of effects that one event has introduced

Side effects at a glance

82 / 93

DSL to connect reactors to events
Reactors only create other events in turn
All that chain is logged in events metadata so you can trace the chain of effects that one event has introduced

Data you write != Data you readPersist everything in EventsSurface what you need in Aggregates83 / 93

You don't need to make trade offs anymore when designing

The hard parts84 / 93

Naming is hardWith immutable events, names are final.85 / 93

Updating event schemas is (surprisingly) not a big deal86 / 93

Updating event schemas is (surprisingly) not a big dealNew attributes tend to have an implicit default value87 / 93

For example, adding currency. Default to USD.

Destructuring a request into multiple events is complex

88 / 93

Destructuring a request into multiple events is complex

89 / 93

Destructuring a request into multiple events is complex

90 / 93

Event Sourcing Made Simple91 / 93

Event SourcingThe application state is the result of a sequence of events
Events hold data and metadata available at the time of creation
Reactors trigger side effects - and can create events
Key BenefitsFull history and audit log
Backfilling
Fixing bugs is easier - less risk of losing data
Events can be forwarded at (almost) no additional cost
92 / 93

Event Sourcing Made Simple

🙏

Philippe Creux – pcreux.com

👉 https://kickstarter.engineering/event-sourcing-made-simple 👀

93 / 93

Help

Keyboard shortcuts

↑, ←, Pg Up, k

Go to previous slide

↓, →, Pg Dn, Space, j

Go to next slide

Home

Go to first slide

End

Go to last slide

Number + Return

Go to specific slide

b / m / f

Toggle blackout / mirrored / fullscreen mode

Clone slideshow

Toggle presenter mode

Restart the presentation timer

?, h

Toggle this help