Syntropy.EventRecorder (syntropy v0.2.0)

Copy Markdown

Retains lattice runtime events and broadcasts them on Phoenix PubSub.

Recording is asynchronous: record/2 builds the event in the caller, casts it to the recorder, and returns immediately. The recorder broadcasts the event on "lattice:events" as soon as it is received and persists events in batches (Repo.insert_all) on a short flush interval, so producers such as the task scheduler never block on Postgres writes.

The recorder also subscribes to "lattice:events" itself: events that originate on other cluster nodes (PG PubSub fans them out cluster-wide) are merged into the in-memory history with uid-based deduplication, so recent/1 — and therefore GET /api/events and the cockpit event panel — reflect cluster-wide history. Only locally-recorded events are persisted; each event reaches the shared database exactly once, from its origin node.

Summary

Types

event()

Functions

child_spec(init_arg)

Returns a specification to start this module under a supervisor.

flush()

Synchronously persists any batched events that have not been flushed yet.

recent(limit \\ 100)
record(event_id, payload)
reset()
start_link(opts \\ [])

Types

event()

@type event() :: %{
  id: String.t(),
  uid: String.t(),
  node_id: String.t(),
  node_name: String.t(),
  payload: map(),
  timestamp: DateTime.t()
}

Functions

child_spec(init_arg)

Returns a specification to start this module under a supervisor.

See Supervisor.

flush()

@spec flush() :: :ok

Synchronously persists any batched events that have not been flushed yet.

Intended for tests and shutdown paths that need write durability before inspecting the database.

recent(limit \\ 100)

@spec recent(non_neg_integer()) :: [event()]

record(event_id, payload)

@spec record(String.t(), map()) :: event()

reset()

@spec reset() :: :ok

start_link(opts \\ [])

@spec start_link(keyword()) :: GenServer.on_start()