gem-hunter

Experiment using PostgreSQL as a natively event sourcing database.

Uses triggers and functions to manage projections transactionally.

The basic flow of action is:

event -> after insert trigger -> trigger function -> projection function -> projection

The advantage of this model is that triggers ensure the projections are always up to date, but we do not lose the ability to replay the event stream with the same logic.

Event Sourcing ensures that all changes to application state are stored as a sequence of events.

Events are stored in an events table.

We assume that all objects/entities in the system have a globally unique identifier.

CREATE TABLE "events" (
  "id" serial primary key not null,
  "uuid" uuid NOT NULL,
  "type" text NOT NULL,
  "body" jsonb NOT NULL,
  "inserted_at" timestamp(6) NOT NULL DEFAULT statement_timestamp()
);

An example event, tracking an update to the name of the user identified by a uuid:

insert into events (type, uuid, body)
values ('user_update', '11111111-1111-1111-1111-111111111111', '{"name": "blah"}');

Use after insert triggers on the events table to handle the incoming event actions.

In order to replay the events outside of the trigger mechanism, we wrap a general projection function inside the trigger. This will make more sense in a moment.

Below we create a trigger function and a trigger to execute. The trigger uses a conditional to only fire when the appropriate event type has been inserted.

create or replace function fn_trigger_user_create() returns trigger
  security definer
  language plpgsql
as $$
  begin
    perform fn_project_user_create(new.uuid, new.body);
    return new;
  end;
$$;

create trigger event_insert_user_create after insert on events
  for each row
  when (new.type = 'user_create')
  execute procedure fn_trigger_user_insert();

A projection function does the actual work of handling the event data and mapping to the appropriate projection.

Multiple triggers and multiple functions can be added to handle different aspects of the same event type if required.

Assuming a users table with a name and uuid, the following function inserts a new user record into the table based on the user_create event.

create or replace function fn_project_user_create(uuid uuid, body jsonb) returns integer
  security definer
  language plpgsql as $$
  declare result int;
  begin
    insert into users(uuid, name, inserted_at, updated_at)
      values(uuid, body->>'name', NOW(), NOW())
      returning id into result;
    return result;
  end;
$$;

JSON can be referenced using the native operators in PostgreSQL 9.5. body->>'name' extracts the value of the name field from the body JSON.

Any constraints on the table will also be enforced, ensuring referential integrity.

Using projection functions means that at any point the events can be replayed, simply by calling the function and passing the correct identifier and data.

The following code replays all user_create events in order

do language plpgsql $$
  declare
    e record;
  begin
    for e in select uuid body from events where type = 'user_create' order by inserted_at asc loop
      perform fn_project_user_create(e.uuid, e.body);
    end loop;
  end;
$$;

Any valid query can be used as the basis for the replay loop, and any combination of valid events.

The following code replays all events for the user identified by the specified uuid:

do language plpgsql $$
  declare
    e record;
  begin
    for e in select type, uuid, body from events where uuid = '11111111-1111-1111-1111-111111111111' order by inserted_at asc loop
    case e.type
      when 'user_create' then
        perform fn_project_user_create(e.uuid, e.body);
	   when 'user_update' then
        perform fn_project_user_update(e.uuid, e.body);
	  end case;
    end loop;
  end;
$$;

All of these functions will be executed in the same transaction block. This doesn't particularly matter in an event sourced system, but it is good to know.

PostgreSQL is not just limited to processing events iteratively.

Below is an example of using a materialized view to project the user data.

create materialized view users_view as
  with t as (
      select *, row_number() over(partition by uuid order by inserted_at desc) as row_number
      from events
      where type = 'user_update'
  )
  select uuid, body->>'name' as name, inserted_at from t where row_number = 1;

select * from users_view;

In this case we assume that the most recent update event contains the correct user data, and we query to find the most recent update_user event for each user identified.