-module(lightspeed@pipeline@telemetry).
-compile([no_auto_import, nowarn_unused_vars, nowarn_unused_function, nowarn_nomatch, inline]).
-define(FILEPATH, "src/lightspeed/pipeline/telemetry.gleam").
-export([zero/0, delta/5, apply/2, valid/1, label/1, metrics/3]).
-export_type([snapshot/0, delta/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(" Pipeline telemetry contracts for lag/throughput/retry/dead-letter tracking.\n").

-type snapshot() :: {snapshot, integer(), integer(), integer(), integer()}.

-type delta() :: {delta, integer(), integer(), integer(), integer(), integer()}.

-file("src/lightspeed/pipeline/telemetry.gleam", 28).
?DOC(" Build a zeroed telemetry snapshot.\n").
-spec zero() -> snapshot().
zero() ->
    {snapshot, 0, 0, 0, 0}.

-file("src/lightspeed/pipeline/telemetry.gleam", 38).
?DOC(" Build one telemetry delta.\n").
-spec delta(integer(), integer(), integer(), integer(), integer()) -> delta().
delta(Processed_records, Lag_ms, Retry_count, Dead_letter_count, Elapsed_ms) ->
    {delta,
        Processed_records,
        Lag_ms,
        Retry_count,
        Dead_letter_count,
        Elapsed_ms}.

-file("src/lightspeed/pipeline/telemetry.gleam", 132).
-spec max(integer(), integer()) -> integer().
max(Left, Right) ->
    case Left >= Right of
        true ->
            Left;

        false ->
            Right
    end.

-file("src/lightspeed/pipeline/telemetry.gleam", 117).
-spec throughput_from_delta(delta()) -> integer().
throughput_from_delta(Delta) ->
    case erlang:element(6, Delta) =< 0 of
        true ->
            max(0, erlang:element(2, Delta));

        false ->
            max(0, case max(1, erlang:element(6, Delta)) of
                    0 -> 0;
                    Gleam@denominator -> (erlang:element(2, Delta) * 1000) div Gleam@denominator
                end)
    end.

-file("src/lightspeed/pipeline/telemetry.gleam", 55).
?DOC(" Apply one delta and return a new snapshot.\n").
-spec apply(snapshot(), delta()) -> snapshot().
apply(Snapshot, Delta) ->
    {snapshot,
        max(erlang:element(2, Snapshot), erlang:element(3, Delta)),
        throughput_from_delta(Delta),
        max(0, erlang:element(4, Snapshot) + erlang:element(4, Delta)),
        max(0, erlang:element(5, Snapshot) + erlang:element(5, Delta))}.

-file("src/lightspeed/pipeline/telemetry.gleam", 68).
?DOC(" Snapshot invariants.\n").
-spec valid(snapshot()) -> boolean().
valid(Snapshot) ->
    (((erlang:element(2, Snapshot) >= 0) andalso (erlang:element(3, Snapshot) >= 0))
    andalso (erlang:element(4, Snapshot) >= 0))
    andalso (erlang:element(5, Snapshot) >= 0).

-file("src/lightspeed/pipeline/telemetry.gleam", 76).
?DOC(" Stable snapshot label.\n").
-spec label(snapshot()) -> binary().
label(Snapshot) ->
    <<<<<<<<<<<<<<"lag_ms="/utf8,
                                (erlang:integer_to_binary(
                                    erlang:element(2, Snapshot)
                                ))/binary>>/binary,
                            "|throughput_rps="/utf8>>/binary,
                        (erlang:integer_to_binary(erlang:element(3, Snapshot)))/binary>>/binary,
                    "|retries="/utf8>>/binary,
                (erlang:integer_to_binary(erlang:element(4, Snapshot)))/binary>>/binary,
            "|dead_letters="/utf8>>/binary,
        (erlang:integer_to_binary(erlang:element(5, Snapshot)))/binary>>.

-file("src/lightspeed/pipeline/telemetry.gleam", 125).
-spec metric_tags(binary(), binary()) -> list(lightspeed@ops@telemetry:tag()).
metric_tags(Pipeline_name, Run_id) ->
    [{tag, <<"pipeline"/utf8>>, Pipeline_name},
        {tag, <<"run_id"/utf8>>, Run_id}].

-file("src/lightspeed/pipeline/telemetry.gleam", 88).
?DOC(" Convert one snapshot to metric points.\n").
-spec metrics(snapshot(), binary(), binary()) -> list(lightspeed@ops@telemetry:metric()).
metrics(Snapshot, Pipeline_name, Run_id) ->
    [{gauge,
            <<"lightspeed.pipeline.lag_ms"/utf8>>,
            erlang:element(2, Snapshot),
            metric_tags(Pipeline_name, Run_id)},
        {gauge,
            <<"lightspeed.pipeline.throughput_records_per_sec"/utf8>>,
            erlang:element(3, Snapshot),
            metric_tags(Pipeline_name, Run_id)},
        {counter,
            <<"lightspeed.pipeline.retry_total"/utf8>>,
            erlang:element(4, Snapshot),
            metric_tags(Pipeline_name, Run_id)},
        {counter,
            <<"lightspeed.pipeline.dead_letter_total"/utf8>>,
            erlang:element(5, Snapshot),
            metric_tags(Pipeline_name, Run_id)}].