diff --git a/lib/fact/event_ledger.ex b/lib/fact/event_ledger.ex index ebaacb6..7e2dbf3 100644 --- a/lib/fact/event_ledger.ex +++ b/lib/fact/event_ledger.ex @@ -1,6 +1,6 @@ defmodule Fact.EventLedger do @moduledoc """ - This is the Judge Judy of the system, it manages the event ledger for a Fact database instance, + This is the Judge Judy of the system, it manages the event ledger for a Fact database instance, handling all commits, enforcing optimistic concurrency control via append conditions. `Fact.EventLedger` is a GenServer responsible for: @@ -28,8 +28,8 @@ defmodule Fact.EventLedger do @typedoc """ Defines the options that can be specified when committing. - - * `:timeout` (default: 5000) - specifies how long the caller should wait for the commit operation to complete in milliseconds + + * `:timeout` (default: 5000) - specifies how long the caller should wait for the commit operation to complete in milliseconds """ @typedoc since: "0.1.0" @type commit_option :: {:timeout, timeout()} @@ -67,7 +67,7 @@ defmodule Fact.EventLedger do Each event is enriched with additional metadata, including a timestamp, and assigning a store position. Events are then written to the configured storage and a reference to the event is appended to the ledger file. - If the commit operation is successful, an `{:appended, {record_id, event}}` message will be published by the + If the commit operation is successful, an `{:appended, {record_id, event}}` message will be published by the `Fact.EventPublisher` for each event that was written. """ @doc since: "0.2.0" @@ -290,6 +290,10 @@ defmodule Fact.EventLedger do event_with_renamed_keys = rename_keys(event, replacements) + deduplicated_tags = + Map.get(event_with_renamed_keys, schema.event_tags, []) + |> Enum.uniq() + enriched_event = %{ schema.event_data => %{}, @@ -300,6 +304,7 @@ defmodule Fact.EventLedger do schema.event_store_timestamp => timestamp } |> Map.merge(event_with_renamed_keys) + |> Map.put(schema.event_tags, deduplicated_tags) {enriched_event, next} end) diff --git a/test/fact_test.exs b/test/fact_test.exs index 3ecdb11..ab5c18a 100644 --- a/test/fact_test.exs +++ b/test/fact_test.exs @@ -32,6 +32,17 @@ defmodule FactTest do assert position == record["store_position"] end + test "append event with duplicate tags are deduplicated", %{db: db} do + event = %{type: "turtle_mutated", tags: ["mutant:1", "mutant:1"]} + assert {:ok, position} = Fact.append(db, event) + + record = + Fact.read(db, :all, position: position - 1, count: 1) + |> List.first() + + assert length(record["event_tags"]) == 1 + end + test "append and read it back via a type query", %{db: db} do event = %{type: "pizza_ordered"} {:ok, position} = Fact.append(db, event) diff --git a/test/test_helper.exs b/test/test_helper.exs index 6c72452..71d74ed 100644 --- a/test/test_helper.exs +++ b/test/test_helper.exs @@ -60,8 +60,8 @@ defmodule TestHelper do @doc """ Subscribes to the database and waits for all the indexers to index events - up to the specified position. - + up to the specified position. + This uses a really naive approach when waiting, it just expects to receive some message with the timeout window, then waits again resetting the time window, a continues until the expected message is received.