AshR2RML.Telemetry.FlyClient (AshR2RML v26.8.22)

Copy Markdown View Source

Telemetry and Fly Control Plane Client for AshR2RML.

Signs and batches IEEE OCEL 2.0 event streams into standard chatgpt-cloud-ocel/1 envelopes, computes deterministic SHA-256 hash-chained batch digests, and dispatches them to FLY_CONTROL_PLANE_URL (POST /api/v1/ocel/events).

Provides:

  • Deterministic batch digest computation & cryptographic hash chaining (previous_digest -> digest).
  • Monotonic sequencing.
  • Client-side event buffering with configurable batch size and auto-flush timer.
  • HTTP dispatching via :httpc with configurable adapters/mock dispatchers.
  • Offline fallback persistence to disk (NDJSON) and replay functionality.
  • Cryptographic envelope signature verification & chain validation.

Summary

Types

envelope()
producer()

Functions

build_envelope(events, sequence, prev_digest, producer, opts \\ [])

Constructs a standard chatgpt-cloud-ocel/1 envelope map with deterministic SHA-256 digest and monotonic sequence metadata.

child_spec(init_arg)

Returns a specification to start this module under a supervisor.

compute_digest(envelope)

Computes a deterministic SHA-256 hex digest from an envelope map.

compute_digest(events, sequence, prev_digest, producer)

Computes a deterministic SHA-256 hex digest for an envelope or batch components.

default_producer(opts \\ [])

Returns a standard producer map populated with system & runtime information.

dispatch_envelope(envelope, opts \\ [])

Dispatches an envelope map to the Fly Control Plane endpoint.

flush(client \\ __MODULE__)

Explicitly flushes buffered events into an envelope and dispatches to the control plane.

get_state(client \\ __MODULE__)

Retrieves the current introspection state of the FlyClient.

push_event(client \\ __MODULE__, event)

Pushes a single IEEE OCEL 2.0 event to the client buffer.

push_events(client \\ __MODULE__, events)

Pushes a batch of IEEE OCEL 2.0 events to the client buffer.

replay_offline(client \\ __MODULE__, opts \\ [])

Replays all offline-buffered envelopes from disk and memory fallback queues.

sign_digest(digest, secret)

Signs a digest using HMAC-SHA256.

start_link(opts \\ [])

Starts the FlyClient GenServer process.

stop(client \\ __MODULE__)

Stops the FlyClient GenServer gracefully after flushing any remaining events.

verify_chain(envelopes)

Verifies that a list of envelopes forms an unbroken, monotonically sequenced cryptographic hash chain.

verify_digest(envelope)

Verifies that an envelope's internal digest matches its computed deterministic digest.

Types

envelope()

@type envelope() :: %{required(String.t()) => any()}

producer()

@type producer() :: %{
  required(String.t()) => String.t(),
  optional(atom()) => String.t()
}

Functions

build_envelope(events, sequence, prev_digest, producer, opts \\ [])

@spec build_envelope(
  [map()],
  non_neg_integer(),
  String.t() | nil,
  map() | keyword(),
  keyword()
) ::
  envelope()

Constructs a standard chatgpt-cloud-ocel/1 envelope map with deterministic SHA-256 digest and monotonic sequence metadata.

child_spec(init_arg)

Returns a specification to start this module under a supervisor.

See Supervisor.

compute_digest(envelope)

@spec compute_digest(envelope()) :: String.t()

Computes a deterministic SHA-256 hex digest from an envelope map.

compute_digest(events, sequence, prev_digest, producer)

@spec compute_digest([map()], non_neg_integer(), String.t() | nil, map()) ::
  String.t()

Computes a deterministic SHA-256 hex digest for an envelope or batch components.

default_producer(opts \\ [])

@spec default_producer(keyword()) :: map()

Returns a standard producer map populated with system & runtime information.

dispatch_envelope(envelope, opts \\ [])

@spec dispatch_envelope(
  envelope(),
  keyword()
) :: {:ok, map()} | {:error, {:offline_buffered, any()}} | {:error, any()}

Dispatches an envelope map to the Fly Control Plane endpoint.

flush(client \\ __MODULE__)

@spec flush(GenServer.server()) :: {:ok, envelope()} | {:ok, :empty} | {:error, any()}

Explicitly flushes buffered events into an envelope and dispatches to the control plane.

get_state(client \\ __MODULE__)

@spec get_state(GenServer.server()) :: map()

Retrieves the current introspection state of the FlyClient.

push_event(client \\ __MODULE__, event)

@spec push_event(GenServer.server(), map()) :: :ok

Pushes a single IEEE OCEL 2.0 event to the client buffer.

push_events(client \\ __MODULE__, events)

@spec push_events(GenServer.server(), [map()]) :: :ok

Pushes a batch of IEEE OCEL 2.0 events to the client buffer.

replay_offline(client \\ __MODULE__, opts \\ [])

@spec replay_offline(
  GenServer.server(),
  keyword()
) :: {:ok, non_neg_integer()} | {:error, any()}

Replays all offline-buffered envelopes from disk and memory fallback queues.

sign_digest(digest, secret)

@spec sign_digest(String.t(), String.t()) :: String.t()

Signs a digest using HMAC-SHA256.

start_link(opts \\ [])

Starts the FlyClient GenServer process.

stop(client \\ __MODULE__)

@spec stop(GenServer.server()) :: :ok

Stops the FlyClient GenServer gracefully after flushing any remaining events.

verify_chain(envelopes)

@spec verify_chain([envelope()]) ::
  {:ok, non_neg_integer()} | {:error, {atom(), map()}}

Verifies that a list of envelopes forms an unbroken, monotonically sequenced cryptographic hash chain.

verify_digest(envelope)

@spec verify_digest(envelope()) :: boolean()

Verifies that an envelope's internal digest matches its computed deterministic digest.