factos

//// Store-independent event-sourcing domain primitives.
////
//// Factos keeps the domain model in the application. The core module models
//// facts, command contexts, pure decision components, and pure views. Concrete
//// storage concerns live in backend packages such as `factos_sqlight` and
//// `factos_kurrentdb_erlang`.
////
//// This package follows a context-first reading of Event Sourcing: accepted
//// facts are the authoritative state of the system, and the facts relevant to a
//// command are considered before new facts are accepted. Aggregates, stream-per-
//// object storage, CQRS, projections, and message brokers are implementation
//// choices rather than prerequisites.
////
//// The central flow is:
////
//// 1. Select a command context with `Query`.
//// 2. Fold the matching recorded events into a temporary decision state.
//// 3. Run a pure `Decider`.
//// 4. Append the produced facts only if the context is still stable.
////
//// Backends implement the storage-specific parts of that flow. This module keeps
//// the shared types and pure computations small and portable.

import gleam/list
import gleam/option.{type Option, None, Some}
import gleam/result

/// A store-visible event type name.
///
/// Event types are part of the query contract. A backend may use them for
/// efficient context reads, and applications should keep names stable enough
/// for stored history to remain decodable.
pub opaque type EventType {
  EventType(String)
}

/// A store-visible tag value.
///
/// Tags expose selected payload information to the event store so commands can
/// query the facts relevant to a decision. For example, an event payload may
/// contain `username: "renata"`, while the stored event also carries the tag
/// `username:renata`.
pub opaque type Tag {
  Tag(String)
}

/// Application metadata attached to a recorded event.
///
/// Metadata is not part of query matching. It is intended for operational and
/// audit context such as correlation ids, causation ids, actors, and timestamps.
pub opaque type Metadata {
  Metadata(List(#(String, String)))
}

pub type Query {
  /// Match every recorded event.
  AllEvents

  /// Match events using one or more query items.
  ///
  /// Query items are OR-combined. See `QueryItem` for the matching rules inside
  /// each item.
  Query(items: List(QueryItem))
}

pub type QueryItem {
  /// One branch of a command-context query.
  ///
  /// Within an item, event types are OR-combined and tags are AND-combined. Empty
  /// `types` means any event type matches. Empty `tags` means no tag constraint.
  ///
  /// A query item with `types: [UserRegistered, UsernameReserved]` and
  /// `tags: [username:renata]` means: events of either type that also have the
  /// `username:renata` tag.
  QueryItem(types: List(EventType), tags: List(Tag))
}

pub type SequencePosition {
  /// No global position was observed.
  NoPosition

  /// A backend-specific global sequence position.
  ///
  /// Positions are used by context append conditions to express "after this
  /// observed point in history". They are not stream revisions.
  SequencePosition(Int)
}

pub type AppendCondition {
  /// Append without an additional context condition.
  NoAppendCondition

  /// Append only if no event matching `query` appeared after `after`.
  ///
  /// This models Command Context Consistency. The decision was made from the
  /// matching facts visible at `after`, so the append must fail if that relevant
  /// context changed before the new facts are recorded.
  FailIfEventsMatch(query: Query, after: SequencePosition)
}

pub type Decider(command, state, event, domain_error) {
  /// A pure command-side domain component.
  ///
  /// `initial` is the empty decision state. `evolve` folds accepted events into
  /// state. `decide` applies a command to the folded state and either returns new
  /// events or a domain error.
  ///
  /// A decider has no dependency on storage, transactions, codecs, projections,
  /// subscriptions, or transports.
  Decider(
    initial: state,
    decide: fn(state, command) -> Result(List(event), domain_error),
    evolve: fn(state, event) -> state,
  )
}

pub type View(state, event) {
  /// A pure projection fold.
  ///
  /// Views derive read-side state from events. They are intentionally only the
  /// computation; persistence, delivery, rebuilds, and subscription management are
  /// outside the core library.
  View(initial: state, evolve: fn(state, event) -> state)
}

pub type Reactor(event, effect) {
  /// A pure event reaction.
  ///
  /// Reactors inspect committed recorded events and produce application-owned
  /// effect values. They do not run IO. Applications or backend adapters decide
  /// whether to execute effects immediately, persist them durably, retry them, or
  /// ignore them during replay.
  Reactor(react: fn(Recorded(event)) -> List(effect))
}

pub type Revision {
  /// A stream has no events.
  NoEvents

  /// The last known revision of a stream.
  CurrentRevision(Int)
}

pub type Decoded(event) {
  /// A domain event decoded from backend storage.
  ///
  /// Backends use codecs supplied by the application. The decoded value includes
  /// the domain event plus the event type and tags that should participate in
  /// query matching, along with non-query event metadata.
  Decoded(
    event: event,
    type_: EventType,
    version: Int,
    tags: List(Tag),
    metadata: Metadata,
  )
}

pub type Recorded(event) {
  /// A stored event with backend metadata.
  ///
  /// `revision` is the per-stream revision. `position` is the global sequence
  /// position used for context consistency. `type_` and `tags` are store-visible
  /// query metadata.
  Recorded(
    id: String,
    stream: String,
    revision: Int,
    position: SequencePosition,
    type_: EventType,
    version: Int,
    tags: List(Tag),
    metadata: Metadata,
    event: event,
  )
}

pub type Context(event, state) {
  /// A command context read from history.
  ///
  /// The context contains the query that selected the relevant facts, the folded
  /// decision state, the matching recorded events, the highest observed position,
  /// and the append condition needed to protect the decision.
  Context(
    query: Query,
    state: state,
    events: List(Recorded(event)),
    position: SequencePosition,
    append_condition: AppendCondition,
  )
}

pub type LoadedStream(event, state) {
  /// A stream read from history.
  ///
  /// This is the stream-consistency counterpart to `Context`. It contains the
  /// folded state for one stream and its current revision.
  LoadedStream(
    stream: String,
    state: state,
    events: List(Recorded(event)),
    revision: Revision,
  )
}

/// Wrap an event type name.
pub fn event_type(name: String) -> EventType {
  EventType(name)
}

/// Unwrap an event type name.
pub fn event_type_name(event_type: EventType) -> String {
  let EventType(name) = event_type
  name
}

/// Wrap a tag value.
pub fn tag(value: String) -> Tag {
  Tag(value)
}

/// Unwrap a tag value.
pub fn tag_value(tag: Tag) -> String {
  let Tag(value) = tag
  value
}

/// No event metadata.
pub fn empty_metadata() -> Metadata {
  Metadata([])
}

/// Build event metadata from key/value pairs.
pub fn metadata(entries: List(#(String, String))) -> Metadata {
  Metadata(entries)
}

/// Unwrap event metadata entries.
pub fn metadata_entries(metadata: Metadata) -> List(#(String, String)) {
  let Metadata(entries) = metadata
  entries
}

/// Build a query from query items.
///
/// An empty list becomes `AllEvents`; otherwise the query contains the supplied
/// items. Query items are OR-combined by `matches_query`.
pub fn query(items: List(QueryItem)) -> Query {
  case items {
    [] -> AllEvents
    [_, ..] -> Query(items)
  }
}

/// Build one command-context query branch.
///
/// Event types are OR-combined. Tags are AND-combined. Empty lists act as wildcards
/// for that part of the item.
pub fn query_item(
  types types: List(EventType),
  tags tags: List(Tag),
) -> QueryItem {
  QueryItem(types:, tags:)
}

/// Build a pure command-side decider.
///
/// The supplied functions remain owned by the application domain. Factos only
/// stores them together so backends and tests can run the same read-decide-append
/// flow consistently.
pub fn decider(
  initial initial: state,
  decide decide: fn(state, command) -> Result(List(event), domain_error),
  evolve evolve: fn(state, event) -> state,
) -> Decider(command, state, event, domain_error) {
  Decider(initial:, decide:, evolve:)
}

/// Build a pure projection view.
///
/// A view folds events into read-side state. It does not prescribe where that
/// state is stored or how events are delivered.
pub fn view(
  initial initial: state,
  evolve evolve: fn(state, event) -> state,
) -> View(state, event) {
  View(initial:, evolve:)
}

/// Build a pure event reactor.
///
/// Reactors are the side-effect planning counterpart to views: they consume
/// committed recorded events and return application-owned effect values without
/// executing IO.
pub fn reactor(
  react react: fn(Recorded(event)) -> List(effect),
) -> Reactor(event, effect) {
  Reactor(react:)
}

/// Fold events with a decider and decide which new events a command produces.
///
/// This is useful for unit tests and for in-memory command handling. It does not
/// perform any append or consistency check.
pub fn compute_events(
  decider decider: Decider(command, state, event, domain_error),
  events events: List(event),
  command command: command,
) -> Result(List(event), domain_error) {
  let Decider(initial, decide, evolve) = decider
  decide(fold_events(initial, events, evolve), command)
}

/// Decide from an optional current state and return the state after produced events.
///
/// If `current` is `None`, the decider's initial state is used. The function first
/// runs the decider, then folds the produced events into the decision state.
pub fn compute_state(
  decider decider: Decider(command, state, event, domain_error),
  current current: Option(state),
  command command: command,
) -> Result(state, domain_error) {
  let Decider(initial, decide, evolve) = decider
  let state = case current {
    Some(state) -> state
    None -> initial
  }

  use events <- result.try(decide(state, command))
  Ok(fold_events(state, events, evolve))
}

/// Fold recorded events into state using a domain evolution function.
///
/// Backends use this after decoding stored events. Only the domain event payload is
/// passed to `evolve`; storage metadata is ignored for state computation.
pub fn evolve_recorded(
  initial initial: state,
  events events: List(Recorded(event)),
  evolve evolve: fn(state, event) -> state,
) -> state {
  use state, recorded <- list.fold(events, initial)
  evolve(state, recorded.event)
}

/// Project events from a view's initial state.
pub fn project(
  view view: View(state, event),
  events events: List(event),
) -> state {
  let View(initial, evolve) = view
  fold_events(initial, events, evolve)
}

/// Project events starting from an already materialized view state.
pub fn project_from(
  view view: View(state, event),
  state state: state,
  events events: List(event),
) -> state {
  let View(_, evolve) = view
  fold_events(state, events, evolve)
}

/// Produce effect values for one committed recorded event.
pub fn react(
  reactor reactor: Reactor(event, effect),
  event event: Recorded(event),
) -> List(effect) {
  let Reactor(react) = reactor
  react(event)
}

/// Produce effect values for committed recorded events, preserving event order.
pub fn react_all(
  reactor reactor: Reactor(event, effect),
  events events: List(Recorded(event)),
) -> List(effect) {
  use event <- list.flat_map(events)
  react(reactor, event)
}

/// Merge two reactors that consume the same event type.
///
/// The resulting reactor runs both reactors for each event and concatenates their
/// produced effects in argument order.
pub fn merge_reactors(
  first first: Reactor(event, effect),
  second second: Reactor(event, effect),
) -> Reactor(event, effect) {
  use event <- Reactor
  list.append(react(first, event), react(second, event))
}

/// Merge two views that consume the same event type.
///
/// The resulting view keeps both states in a tuple and evolves both for every
/// event. This is a convenience for composing small pure projections.
pub fn merge_views(
  first first: View(first_state, event),
  second second: View(second_state, event),
) -> View(#(first_state, second_state), event) {
  let View(first_initial, first_evolve) = first
  let View(second_initial, second_evolve) = second

  use state, event <- View(initial: #(first_initial, second_initial))
  let #(first_state, second_state) = state
  #(first_evolve(first_state, event), second_evolve(second_state, event))
}

/// Run a command against a previously read context.
///
/// The returned tuple preserves the original context alongside the newly produced
/// events so a backend can append them with `context.append_condition`.
pub fn decide_context(
  context: Context(event, state),
  command: command,
  decider: Decider(command, state, event, domain_error),
) -> Result(#(Context(event, state), List(event)), domain_error) {
  use events <- result.try(decider.decide(context.state, command))
  Ok(#(context, events))
}

/// Test whether a recorded event belongs to a query-defined context.
pub fn matches_query(recorded: Recorded(event), query: Query) -> Bool {
  case query {
    AllEvents -> True
    Query(items) -> list.any(items, matches_item(recorded, _))
  }
}

/// Return the later of two sequence positions.
///
/// `NoPosition` acts as absence of an observed position. If both positions are
/// concrete, the larger integer is returned.
pub fn highest_position(
  left: SequencePosition,
  right: SequencePosition,
) -> SequencePosition {
  case left, right {
    NoPosition, position -> position
    position, NoPosition -> position
    SequencePosition(left), SequencePosition(right) ->
      case left >= right {
        True -> SequencePosition(left)
        False -> SequencePosition(right)
      }
  }
}

fn fold_events(
  initial: state,
  events: List(event),
  evolve: fn(state, event) -> state,
) -> state {
  use state, event <- list.fold(events, initial)
  evolve(state, event)
}

fn matches_item(recorded: Recorded(event), item: QueryItem) -> Bool {
  let QueryItem(types, tags) = item
  matches_types(recorded.type_, types) && matches_tags(recorded.tags, tags)
}

fn matches_types(event_type: EventType, types: List(EventType)) -> Bool {
  case types {
    [] -> True
    [_, ..] -> {
      use required <- list.any(types)
      event_type == required
    }
  }
}

fn matches_tags(event_tags: List(Tag), required_tags: List(Tag)) -> Bool {
  case required_tags {
    [] -> True
    [_, ..] -> {
      use required <- list.all(required_tags)
      use tag <- list.any(event_tags)
      tag == required
    }
  }
}