distribute

-module(distribute@telemetry).
-compile([no_auto_import, nowarn_unused_vars, nowarn_unused_function, nowarn_nomatch, inline]).
-define(FILEPATH, "src/distribute/telemetry.gleam").
-export([install/1, reset/0, emit/1]).
-export_type([event/0, atom_budget_origin/0, payload_origin/0, decode_origin/0]).

-if(?OTP_RELEASE >= 27).
-define(MODULEDOC(Str), -moduledoc(Str)).
-define(DOC(Str), -doc(Str)).
-else.
-define(MODULEDOC(Str), -compile([])).
-define(DOC(Str), -compile([])).
-endif.

?MODULEDOC(
    " Single-event observability sink.\n"
    "\n"
    " `distribute` emits structured `Event` values at every load-bearing\n"
    " boundary: registry operations, atom-budget exhaustion, payload\n"
    " rejection, call lifecycle, decode failures, orphan cleanup. The\n"
    " sink is **opt-in**: with no sink installed, every emit point is a\n"
    " branch on a missing ETS entry and a return. Microseconds, lock-\n"
    " free, zero allocation.\n"
    "\n"
    " ## Why one global sink instead of `_observed` variants\n"
    "\n"
    " `_observed` start variants already cover decode errors at the\n"
    " per-actor boundary. They are kept for that use case, where the\n"
    " hook needs to capture the exact handler context. The telemetry\n"
    " sink is for cross-cutting observability: a single subscriber that\n"
    " wants visibility into *all* registry/cluster/payload/call events\n"
    " without instrumenting every call site.\n"
    "\n"
    " ## No structured error types in `Event`\n"
    "\n"
    " `Event` variants carry **stringified** failure reasons (via\n"
    " `*_error_to_string`) instead of structured `RegisterError` /\n"
    " `DecodeError` / `UnregisterError` references. Two reasons:\n"
    "\n"
    " 1. **No import cycles**. `telemetry` is a leaf module that every\n"
    "    other module can depend on; the moment we re-exported\n"
    "    `registry.RegisterError` here, `registry → telemetry → registry`\n"
    "    would close. Stringified reasons keep `telemetry` independent.\n"
    " 2. **Operationally, downstream sinks log/meter strings anyway**.\n"
    "    A `Prometheus` exporter increments a counter labelled with the\n"
    "    failure kind; a structured logger writes the rendered message.\n"
    "    Pattern-matching on a typed sum is rarely the right shape for\n"
    "    observability output.\n"
    "\n"
    " ## Building on top\n"
    "\n"
    " Downstream packages (`distribute_telemetry`, `distribute_audit`,\n"
    " `distribute_metrics`, `distribute_otel`) install a single sink and\n"
    " fan out from there. The contract is intentionally narrow: `Event`\n"
    " is a pure data sum, the sink is `fn(Event) -> Nil`. No subscription\n"
    " manager, no filtering, no async fan-out. Compose with whatever\n"
    " your stack already has.\n"
    "\n"
    " ## Example\n"
    "\n"
    " ```gleam\n"
    " import distribute/telemetry\n"
    " import gleam/io\n"
    " import gleam/int\n"
    "\n"
    " pub fn main() {\n"
    "   telemetry.install(fn(event) {\n"
    "     case event {\n"
    "       telemetry.ActorRegistered(name) ->\n"
    "         io.println(\"registered: \" <> name)\n"
    "       telemetry.PayloadRejected(size, _, _) ->\n"
    "         io.println(\"payload rejected: \" <> int.to_string(size))\n"
    "       _ -> Nil\n"
    "     }\n"
    "   })\n"
    "   // ... rest of your app\n"
    " }\n"
    " ```\n"
    "\n"
    " ## Contract for downstream consumers\n"
    "\n"
    " **Stability**. `Event` variants are append-only within a major\n"
    " version: a 4.x release will never remove or rename a variant, but\n"
    " it may add new ones. Downstream sinks must include a `_ -> Nil`\n"
    " catch-all to stay forward-compatible across minor releases. New\n"
    " variants will appear in `CHANGELOG.md` under \"Added\".\n"
    "\n"
    " **Field semantics**, by event:\n"
    "\n"
    " - `ActorRegistered(name)`, `ActorRegistrationFailed(name, reason)`,\n"
    "   `ActorUnregistered(name, removed)`: `name` is the binary name\n"
    "   string the caller passed; `reason` is the rendered\n"
    "   `register_error_to_string`; `removed` is `True` only when\n"
    "   `:global` actually removed the entry on this VM (i.e. local\n"
    "   ownership ACL passed and `whereis` succeeded).\n"
    " - `AtomBudgetExhausted(attempted_input, where)`:`attempted_input`\n"
    "   is the raw binary the caller passed (treat as untrusted: log\n"
    "   bytes, never feed back to atom creation). `where` distinguishes\n"
    "   `connect` / `start_node` / `ping`. All three paths emit on\n"
    "   budget refusal; `ping`'s emit is wired at the FFI layer\n"
    "   (`cluster_ffi:ping/1`) because its public Gleam return is `Bool`\n"
    "   with no error variant to carry the typed refusal.\n"
    " - `PayloadRejected(size_bytes, cap_bytes, where)`:`size_bytes`\n"
    "   is the **encoded** payload size (post-codec), not the source\n"
    "   value's heap size. `cap_bytes` is the active\n"
    "   `config.max_payload_size_bytes` at the moment of rejection.\n"
    " - `DecodeFailed(error_message, where)`: `error_message` is the\n"
    "   rendered `decode_error_to_string`. `where` distinguishes\n"
    "   one-shot `receive` / `call` reply / actor-inbound paths.\n"
    " - `CallTimedOut(elapsed_ms)`: `elapsed_ms` is the **effective**\n"
    "   timeout used by the receive path (user input clamped to\n"
    "   non-negative), not the real wall-clock elapsed (the BEAM\n"
    "   does not expose that without monotonic measurement at the call\n"
    "   site). For real elapsed measurement, time around the call in\n"
    "   the sink itself. For `call_isolated`, treat this as\n"
    "   at-least-once telemetry: in a timeout race you may observe two\n"
    "   `CallTimedOut` emits for one user call (inner `call` timeout +\n"
    "   outer caller-side timeout).\n"
    " - `CallTargetDown` carries no payload. Two emit paths fold here:\n"
    "   the caller observed a dead target at call time, or the monitor\n"
    "   fired during the wait.\n"
    " - `OrphanKillEscalated(pid)`: the carried Pid is the orphan\n"
    "   actor that ignored `exit(_, shutdown)` past the grace window\n"
    "   and was force-killed. Operationally significant: log and alert.\n"
    "\n"
    " **Concurrency model**. `emit/1` runs the installed sink **inline\n"
    " in the calling process**. There is no buffering, no async\n"
    " dispatch, and no retry. If the sink raises, the FFI catches the\n"
    " exception and logs it via `logger:warning/2` (event tag plus\n"
    " stack trace, no payload, so sensitive `attempted_input` strings\n"
    " do not leak into shared logs). The library-internal process that\n"
    " emitted the event survives. This protects actors, the call\n"
    " selector, and the `call_isolated` proxy from being taken down by\n"
    " a buggy sink, while still surfacing the bug operationally.\n"
    "\n"
    " Production sinks should still be written fail-fast and total:\n"
    " the catch is a safety net for shipped bugs, not a license to\n"
    " raise. Async fan-out (routing to a metrics aggregator) is best\n"
    " done by `process.send`-ing the event to a background worker\n"
    " from inside the sink itself.\n"
    "\n"
    " **Visibility model**. `install/1` is last-wins, the most recent\n"
    " caller replaces the previous sink. There is no subscription list,\n"
    " no priority, no filter chain. A library that wants multiple\n"
    " downstream consumers (e.g. logger + metrics + audit) installs\n"
    " **one** sink that fans out internally to whatever it cares about.\n"
    "\n"
    " **Storage**. Sink reference is held in an ETS table\n"
    " (`distribute_telemetry_sink_table`, `public`, `read_concurrency`).\n"
    " Reads on the hot path are lock-free; writes (`install`/`reset`)\n"
    " trigger no global GC. Production code is expected to call\n"
    " `install` exactly once at boot.\n"
).

-type event() :: {actor_registered, binary()} |
    {actor_registration_failed, binary(), binary()} |
    {actor_unregistered, binary(), boolean()} |
    {atom_budget_exhausted, binary(), atom_budget_origin()} |
    {payload_rejected, integer(), integer(), payload_origin()} |
    {decode_failed, binary(), decode_origin()} |
    {call_timed_out, integer()} |
    call_target_down |
    call_proxy_crashed |
    {orphan_kill_escalated, gleam@erlang@process:pid_()} |
    {conflict_resolved,
        binary(),
        gleam@option:option(gleam@erlang@process:pid_())} |
    {conflict_resolver_failed, binary(), binary()}.

-type atom_budget_origin() :: atom_budget_on_connect |
    atom_budget_on_ping |
    atom_budget_on_start_node.

-type payload_origin() :: payload_on_send |
    payload_on_call_request |
    payload_on_call_response |
    payload_on_reply |
    payload_on_receive |
    payload_on_actor_inbound.

-type decode_origin() :: decode_on_receive |
    decode_on_call_reply |
    decode_on_actor_inbound.

-file("src/distribute/telemetry.gleam", 272).
?DOC(
    " Install (or replace) the telemetry sink. Last-wins, a second\n"
    " `install` call replaces the previous sink without warning. This is\n"
    " intentional: tests that swap sinks between cases stay terse, and\n"
    " production code should call `install` exactly once at boot.\n"
).
-spec install(fun((event()) -> nil)) -> nil.
install(Sink) ->
    telemetry_ffi:install(Sink).

-file("src/distribute/telemetry.gleam", 279).
?DOC(false).
-spec reset() -> nil.
reset() ->
    telemetry_ffi:reset().

-file("src/distribute/telemetry.gleam", 288).
?DOC(false).
-spec emit(event()) -> nil.
emit(Event) ->
    telemetry_ffi:emit(Event).