@spec append(
  database_id(),
  event() | [event(), ...],
  append_condition(),
  append_options()
) :: {:ok, event_position()} | {:error, term()}

Appends one or more events to the ledger.

The ledger represents the full, ordered event history for a database. In addition to standard optimistic concurrency mechanisms at the stream level, Fact also supports conditional appends at the ledger level to help prevent duplicate or conflicting writes.

These conditions are expressed through the append_condition argument.

When appending, you may provide one of the following:

• nil (default) — no condition is applied; the events are always appended

• a Fact.fail_if_match/0 value — the append will be rejected if the query matches any existing events anywhere in the ledger

• {fail_if_match, after_position} — the append will be rejected if the query matches any events whose position is strictly greater than after_position.

If the condition is violated, the append is rejected and an error tuple including a Fact.ConcurrencyError is returned.

On success, each appended event record is enriched with the :event_id, :event_store_position, :event_store_timestamp fields defined by the configured event schema. In addition the field keys used to define events (type, data, metatadata and tags) are renamed according to the configured schema.

The function returns {:ok, last_position}, where last_position is the store position of the last appended event.

Examples

Append without conditions:

iex> {:ok, pos} = Fact.append(db, %{type: "user_registered", data: %{id: 123}})
{:ok, 42}

Append a duplicate:

iex> {:ok, pos} = Fact.append(db, %{type: "user_registered", data: %{id: 123}})
{:ok, 43}

Prevent a third duplicate using a fail_if_match query:

iex> import Fact.QueryItem
iex> fail_if_match = types("user_registered") |> data(id: "123")
iex> Fact.append(db, %{type: "user_registered", data: %{id: 123}}, fail_if_match)
{:error, %Fact.ConcurrencyError{source: :all, expected: 0, actual: 42}}

Allow the append only if no matching events exist after a given position:

iex> Fact.append(db, %{type: "user_registered"}, {fail_if_match, last_pos})
{:ok, 44}

The final example was intentionally contrived; in practice, append conditions are best applied to model explicit business invariants. For deeper guidance and real-world usage patterns, see the Dynamic Consistency Boundary website.

@spec append_stream(
  database_id(),
  event() | [event(), ...],
  event_stream_id(),
  append_stream_expectation(),
  append_stream_options()
) :: {:ok, event_position()} | {:error, term()}

Appends one or more events to a stream.

Event streams define a consistency boundary for a set of related events. To preserve this boundary, Fact, like many event stores uses stream position expectations to provide optimistic concurrency control.

When appending to a stream, you may provide an expected position value:

• A non negative integer - verifies that the stream is currently at that exact position before appending
• :any (default) - no check is performed, the events are just appended
• :none - ensures the stream does not yet exist (equivalent to 0)
• :exists - ensures the stream already exists (i.e. position ≥ 1)

If the expectation is not met, the append is rejected and a {:error, %Fact.ConcurrencyError{}} is returned.

On success, each appended event record is enriched with the stream fields as defined by :event_stream_id and :event_stream_position of the configured schema (see Fact.Event.Schema). The function returns {:ok, last_stream_position}, where the last_stream_position refers to the stream position of the final event written to the stream.

It is strongly recommnded that you persis this returned position in your application state and reuse it in subsequent calls to append_stream/5 to ensure consistency.

Examples

Append a single event.

iex> {:ok, db} = Fact.open("data/turtle")
{:ok, "TURTLE4F7Q6Y2X3VQKBJ5M7P4Z"}

iex> Fact.append_stream(db, %{type: "egg_hatched", data: %{name: "Turts"}}, "turtle-1")
{:ok, 1}

iex> Fact.read(db, {:stream, "turtle-1"}) |> Enum.to_list()
[
  %{
    "event_data" => %{"name" => "Turts"},
    "event_id" => "3bb4808303c847fd9ceb0a1251ef95da",                                                                                                                                                                                                                                                               
    "event_metadata" => %{},                                                                                                                                                                                                                                                                                        
    "event_tags" => [],                                                                                                                                                                                                                                                                                             
    "event_type" => "egg_hatched",                                                                                                                                                                                                                                                                                  
    "store_position" => 2,                                                                                                                                                                                                                                                                                          
    "store_timestamp" => 1765039106962264,                                                                                                                                                                                                                                                                          
    "stream_id" => "turtle-1",                                                                                                                                                                                                                                                                                  
    "stream_position" => 1                                                                                                                                                                                                                                                                                          
  }                                                                                                                                                                                                                                                                                                                 
]  

Append another event, expecting correct stream position.

iex> Fact.append_stream(db, %{type: "MySecondEvent"}, "myteststream", 1)
{:ok, 2}

Append a third event, but provide an invalid expected position.

iex> Fact.append_stream(:mydb, %{type: "MyThirdEvent"}, "myteststream", 1)
{:error, %Fact.ConcurrencyError{source: "myteststream", actual: 2, expected: 1}}

Append multiple events to a new stream.

iex> Fact.append_stream(db, [%{type: "foo"}, %{type: "bar"}, %{type: "baz"}], "foobarbaz-1", :none)
{:ok, 3}

Expect a stream to not exist, yet it does.

iex> Fact.append_stream(db, %{type: "foo"}, "foobarbaz-1", :none)
{:ok, %Fact.ConcurrencyError{source: "foobarbaz-1", expected: :none, actual: 3}}

Expect a stream to exist, when it does not exist.

iex> Fact.append_stream(db, %{type: "foo"}, "foo-1", :exists)
{:ok, %Fact.ConcurrencyError{source: "foo-1", expected: :exists, actual: 0}}

@spec open(Path.t()) :: {:ok, database_id()} | {:error, term()}

Initializes a Fact database at the given filesystem path.

This function ensures that the Fact.Supervisor is running and then starts the database supervision tree as a child process. Once the Fact.DatabaseSupervisor is running, the database id is returned, and you may use it as a handle for appending events, reading and subscribing to event sources.

Examples

Opens a new database.

iex> {:ok, db} = Fact.open("data/turtles")
{:ok, "EF73AQJ6S5HHZE5PMX7ZP254QQ"}

Subsequent calls to the same path return the same database id...with the same BEAM.

iex> {:ok, db2} = Fact.open("data/turtles")
{:ok, "EF73AQJ6S5HHZE5PMX7ZP254QQ"}

Keep that database running. Try to open the database again in another instance of IEx. You'll get a database locked error similar to the following:

iex> Fact.open("data/turtles")
{:error, :database_locked,
 %{
   "locked_at" => "2026-01-07T06:18:57.669109Z",
   "mode" => "run",
   "node" => "nonode@nohost",
   "os_pid" => "933078",
   "vm_pid" => "#PID<0.232.0>"
 }}

Remember to use mix fact.create -p <path> to create a database before attempting to open it.

iex> Fact.open("does/not/exist")
{:error, :database_not_found}

@spec read(database_id(), read_source(), read_options()) :: read_result()

Read from an event source.

Event Sources

Like most event stores, you can read from the global stream, or an individual event stream. Fact provides a few more options...

• :none - the empty stream

• :all - the all stream (a.k.a. the global stream; a.k.a. the ledger)

• {:stream, stream_id} - an individual event stream

• {:index, indexer_id, index} - an event index

• {:query, query_items} - an event query

Options

You may provide a keyword list with the following options to craft the results to fit your need. If any of the options are not specified, sensible defaults will be provided.

direction:

• :forward (default) - events are read in increasing position order (e.g. 1, 2, 3, ...)
• :backward - events are read in decreasing position order (e.g. 100, 99, 98, ...)

position:

Set the position to begin reading the event source.

• :start (default) - the position immediately before the first item in the event source
• :end - the position immediately after the last item in the event source
• Or a non-negative integer representing the absolute position within the source

count:

Control the maximum size of the result set.

• :all (default) - reads everything in the event source
• Or a non-negative integer.

result:

Control the shape of the elements in result set.

• :event (default) - each element is a map containing the event details.
• :record - each element is a 2-tuple {record_id, event} containing both the record id and the event.
• :record_id - each element is the record id of the event.

eager:

Controls if the result set will be enumerated or lazy.

• true (default) - The internal Stream is enumerated and a List is returned.
• false - The result is returned as Stream

@spec subscribe(database_id(), subscribe_source(), subscribe_options()) ::
  {:ok, pid()}

Subscribe a process to an event source.

A subscription streams new events to the subscriber process as they are appended to the event store. The subscriber process receives one message per event in the form:

• {:record, record} - where record is a 2-tuple {record_id, event_record} (see t:Fact.record())

The subscription begins by replaying events from the specified source, starting at the configured position. Once all historical events have been delivered, the processes receives a :caught_up message and the subscription shifts into live mode, where it waits for and delivers new events as they arrive.

Event Sources

Subscriptions support most of the same sources as read/3, with a few exceptions.

Subscribing to the empty stream (i.e. :none) would create a very lonely Fact.CatchUpSubscription process that never delivers any messages, and I don't believe that would be useful. If you have a legitimate use case, please make your case. It would be straightforward to implement and support.

Subscribing to query sources requires Fact.QueryItem.t/0; a Fact.Query.t/0 won't work here. A Fact.Query.t/0 is just a function, and that function depends on data produced by a combination of the Fact.EventTypeIndexer, Fact.EventTagsIndexer, and any number of Fact.EventDataIndexer processes. A Fact.QueryItem.t/0 contains the metadata needed to subscribe to the correct indexers so the subscription and coordinate when events become "visible" to the subscriber.

Could a Fact.Query.t/0 be decompiled back into an AST so we could reconstruct that information? Probably! But for now, I'd rather spend my time building other things. If you're like my good buddy Tim and love spelunking through abstract syntax trees, I'll happily take your pull request.

Options

position:

Set the position to begin reading the event source.

• :start (default) - the position immediately before the first item in the event source
• :end - the position immediately after the last item in the event source
• Or a non-negative integer representing the absolute position within the source

subscriber:

Specifies the PID of the process that will receive subscription message. Defaults to self().