eventsourcing_sqlite

import eventsourcing
import gleam/dynamic
import gleam/int
import gleam/io
import gleam/list
import gleam/pair
import gleam/result
import sqlight

// CONSTANTS ----

const insert_event_query = "
  INSERT INTO event 
  (aggregate_type, aggregate_id, sequence, event_type, event_version, payload)
  VALUES 
  ($1, $2, $3, $4, $5, $6)
  "

const select_events_query = "
  SELECT aggregate_type, aggregate_id, sequence, event_type, event_version, payload
  FROM event
  WHERE aggregate_type = $1 AND aggregate_id = $2
  ORDER BY sequence
  "

const create_event_table_query = "
  CREATE TABLE IF NOT EXISTS event
  (
    aggregate_type text                         NOT NULL,
    aggregate_id   text                         NOT NULL,
    sequence       bigint CHECK (sequence >= 0) NOT NULL,
    event_type     text                         NOT NULL,
    event_version  text                         NOT NULL,
    payload        text                         NOT NULL,
    PRIMARY KEY (aggregate_type, aggregate_id, sequence)
  );
  "

// TYPES ----

pub opaque type SqliteStore(entity, command, event, error) {
  SqliteStore(
    db: sqlight.Connection,
    empty_aggregate: eventsourcing.Aggregate(entity, command, event, error),
    event_encoder: fn(event) -> String,
    event_decoder: fn(String) -> Result(event, List(dynamic.DecodeError)),
    event_type: String,
    event_version: String,
    aggregate_type: String,
  )
}

// CONSTRUCTORS ----

pub fn new(
  sqlight_connection sqlight_connection: sqlight.Connection,
  empty_entity empty_entity: entity,
  handle_command_function handle: eventsourcing.Handle(
    entity,
    command,
    event,
    error,
  ),
  apply_function apply: eventsourcing.Apply(entity, event),
  event_encoder event_encoder: fn(event) -> String,
  event_decoder event_decoder: fn(String) ->
    Result(event, List(dynamic.DecodeError)),
  event_type event_type: String,
  event_version event_version: String,
  aggregate_type aggregate_type: String,
) -> eventsourcing.EventStore(
  SqliteStore(entity, command, event, error),
  entity,
  command,
  event,
  error,
) {
  let eventstore =
    SqliteStore(
      db: sqlight_connection,
      empty_aggregate: eventsourcing.Aggregate(empty_entity, handle, apply),
      event_encoder:,
      event_decoder:,
      event_type:,
      event_version:,
      aggregate_type:,
    )

  eventsourcing.EventStore(
    eventstore:,
    commit: commit,
    load_aggregate: load_aggregate,
  )
}

pub fn create_event_table(
  sqlite_store: SqliteStore(entity, command, event, error),
) {
  sqlight.query(
    create_event_table_query,
    on: sqlite_store.db,
    with: [],
    expecting: dynamic.dynamic,
  )
}

pub fn load_aggregate_entity(
  sqlite_store: SqliteStore(entity, command, event, error),
  aggregate_id: eventsourcing.AggregateId,
) -> entity {
  load_aggregate(sqlite_store, aggregate_id).aggregate.entity
}

pub fn load_events(
  sqlite_store: SqliteStore(entity, command, event, error),
  aggregate_id: eventsourcing.AggregateId,
) {
  use resulted <- result.map(sqlight.query(
    select_events_query,
    on: sqlite_store.db,
    with: [
      sqlight.text(sqlite_store.aggregate_type),
      sqlight.text(aggregate_id),
    ],
    expecting: dynamic.decode6(
      eventsourcing.SerializedEventEnvelop,
      dynamic.element(1, dynamic.string),
      dynamic.element(2, dynamic.int),
      dynamic.element(5, fn(dyn) {
        let assert Ok(payload) =
          dynamic.string(dyn) |> result.map(sqlite_store.event_decoder)
        payload
      }),
      dynamic.element(3, dynamic.string),
      dynamic.element(4, dynamic.string),
      dynamic.element(0, dynamic.string),
    ),
  ))
  resulted
}

fn load_aggregate(
  sqlite_store: SqliteStore(entity, command, event, error),
  aggregate_id: eventsourcing.AggregateId,
) -> eventsourcing.AggregateContext(entity, command, event, error) {
  let assert Ok(commited_events) = load_events(sqlite_store, aggregate_id)

  let #(aggregate, sequence) =
    list.fold(
      over: commited_events,
      from: #(sqlite_store.empty_aggregate, 0),
      with: fn(aggregate_and_sequence, event_envelop) {
        let #(aggregate, _) = aggregate_and_sequence
        #(
          eventsourcing.Aggregate(
            ..aggregate,
            entity: aggregate.apply(aggregate.entity, event_envelop.payload),
          ),
          event_envelop.sequence,
        )
      },
    )
  eventsourcing.AggregateContext(aggregate_id:, aggregate:, sequence:)
}

fn commit(
  sqlite_store: SqliteStore(entity, command, event, error),
  context: eventsourcing.AggregateContext(entity, command, event, error),
  events: List(event),
) {
  let eventsourcing.AggregateContext(aggregate_id, _, sequence) = context
  let wrapped_events = wrap_events(sqlite_store, aggregate_id, events, sequence)
  persist_events(sqlite_store, wrapped_events)
  io.println(
    "storing: "
    <> wrapped_events |> list.length |> int.to_string
    <> " events for Aggregate ID '"
    <> aggregate_id
    <> "'",
  )
  wrapped_events
}

fn wrap_events(
  postgres_store: SqliteStore(entity, command, event, error),
  aggregate_id: eventsourcing.AggregateId,
  events: List(event),
  sequence: Int,
) -> List(eventsourcing.EventEnvelop(event)) {
  list.map_fold(
    over: events,
    from: sequence,
    with: fn(sequence: Int, event: event) {
      let next_sequence = sequence + 1
      #(
        next_sequence,
        eventsourcing.SerializedEventEnvelop(
          aggregate_id:,
          sequence: sequence + 1,
          payload: event,
          event_type: postgres_store.event_type,
          event_version: postgres_store.event_version,
          aggregate_type: postgres_store.aggregate_type,
        ),
      )
    },
  )
  |> pair.second
}

fn persist_events(
  sqlite_store: SqliteStore(entity, command, event, error),
  wrapped_events: List(eventsourcing.EventEnvelop(event)),
) {
  wrapped_events
  |> list.map(fn(event) {
    let assert eventsourcing.SerializedEventEnvelop(
      aggregate_id,
      sequence,
      payload,
      event_type,
      event_version,
      aggregate_type,
    ) = event

    sqlight.query(
      insert_event_query,
      on: sqlite_store.db,
      with: [
        sqlight.text(aggregate_type),
        sqlight.text(aggregate_id),
        sqlight.int(sequence),
        sqlight.text(event_type),
        sqlight.text(event_version),
        sqlight.text(payload |> sqlite_store.event_encoder),
      ],
      expecting: dynamic.dynamic,
    )
  })
}