-module(lightspeed@pipeline).
-compile([no_auto_import, nowarn_unused_vars, nowarn_unused_function, nowarn_nomatch, inline]).
-define(FILEPATH, "src/lightspeed/pipeline.gleam").
-export([source_stage/2, transform_stage/3, sink_stage/2, pipeline/3, new/1, checkpoints/1, valid_pipeline/1, valid/1, schedule/2, trigger/3, start/2, process/6, record_retry/2, record_dead_letter/2, fail/2, crash/2, resume_from_latest_checkpoint/2, complete/2, runtime_pipeline/1, lifecycle/1, runtime_telemetry/1, sink_idempotency_keys/1, stage_kind_label/1, schedule_label/1, lifecycle_label/1, process_result_label/1, pipeline_signature/1, signature/1]).
-export_type([stage_kind/0, boundary/0, stage/0, schedule/0, lifecycle/0, pipeline/0, process_result/0, runtime/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(" Integrated data-pipeline core contracts for M31.\n").

-type stage_kind() :: source | transform | sink.

-type boundary() :: {boundary, binary(), binary()}.

-type stage() :: {stage, binary(), stage_kind(), boundary()}.

-type schedule() :: manual | {interval, integer()} | {cron, binary()}.

-type lifecycle() :: idle |
    {scheduled, integer()} |
    {triggered, binary(), integer()} |
    {running, binary(), integer()} |
    {failed, binary(), binary()} |
    {crashed, binary(), binary()} |
    {replaying, binary(), binary(), integer()} |
    {completed, binary(), integer()}.

-type pipeline() :: {pipeline, binary(), schedule(), list(stage())}.

-type process_result() :: {applied, lightspeed@pipeline@checkpoint:checkpoint()} |
    {duplicate_suppressed, binary(), binary()} |
    {rejected, binary()}.

-opaque runtime() :: {runtime,
        pipeline(),
        lifecycle(),
        integer(),
        list(lightspeed@pipeline@checkpoint:checkpoint()),
        list(binary()),
        lightspeed@pipeline@telemetry:snapshot()}.

-file("src/lightspeed/pipeline.gleam", 70).
?DOC(" Build a source stage.\n").
-spec source_stage(binary(), binary()) -> stage().
source_stage(Name, Output_contract) ->
    {stage, Name, source, {boundary, <<"none"/utf8>>, Output_contract}}.

-file("src/lightspeed/pipeline.gleam", 79).
?DOC(" Build a transform stage.\n").
-spec transform_stage(binary(), binary(), binary()) -> stage().
transform_stage(Name, Input_contract, Output_contract) ->
    {stage, Name, transform, {boundary, Input_contract, Output_contract}}.

-file("src/lightspeed/pipeline.gleam", 95).
?DOC(" Build a sink stage.\n").
-spec sink_stage(binary(), binary()) -> stage().
sink_stage(Name, Input_contract) ->
    {stage, Name, sink, {boundary, Input_contract, <<"none"/utf8>>}}.

-file("src/lightspeed/pipeline.gleam", 104).
?DOC(" Build one pipeline contract.\n").
-spec pipeline(binary(), schedule(), list(stage())) -> pipeline().
pipeline(Name, Schedule, Stages) ->
    {pipeline, Name, Schedule, Stages}.

-file("src/lightspeed/pipeline.gleam", 113).
?DOC(" Build one runtime.\n").
-spec new(pipeline()) -> runtime().
new(Pipeline) ->
    {runtime, Pipeline, idle, 1, [], [], lightspeed@pipeline@telemetry:zero()}.

-file("src/lightspeed/pipeline.gleam", 348).
?DOC(" Stable checkpoints in emit order.\n").
-spec checkpoints(runtime()) -> list(lightspeed@pipeline@checkpoint:checkpoint()).
checkpoints(Runtime) ->
    lists:reverse(erlang:element(5, Runtime)).

-file("src/lightspeed/pipeline.gleam", 538).
-spec checkpoints_valid(list(lightspeed@pipeline@checkpoint:checkpoint())) -> boolean().
checkpoints_valid(Entries) ->
    case Entries of
        [] ->
            true;

        [Entry | Rest] ->
            lightspeed@pipeline@checkpoint:valid(Entry) andalso checkpoints_valid(
                Rest
            )
    end.

-file("src/lightspeed/pipeline.gleam", 450).
-spec stages_valid(list(stage())) -> boolean().
stages_valid(Stages) ->
    case Stages of
        [] ->
            true;

        [Stage | Rest] ->
            (((erlang:element(2, Stage) /= <<""/utf8>>) andalso (erlang:element(
                2,
                erlang:element(4, Stage)
            )
            /= <<""/utf8>>))
            andalso (erlang:element(3, erlang:element(4, Stage)) /= <<""/utf8>>))
            andalso stages_valid(Rest)
    end.

-file("src/lightspeed/pipeline.gleam", 520).
-spec has_key(list(binary()), binary()) -> boolean().
has_key(Keys, Key) ->
    case Keys of
        [] ->
            false;

        [Entry | Rest] ->
            case Entry =:= Key of
                true ->
                    true;

                false ->
                    has_key(Rest, Key)
            end
    end.

-file("src/lightspeed/pipeline.gleam", 461).
-spec stage_names_unique(list(stage()), list(binary())) -> boolean().
stage_names_unique(Stages, Seen) ->
    case Stages of
        [] ->
            true;

        [Stage | Rest] ->
            case has_key(Seen, erlang:element(2, Stage)) of
                true ->
                    false;

                false ->
                    stage_names_unique(Rest, [erlang:element(2, Stage) | Seen])
            end
    end.

-file("src/lightspeed/pipeline.gleam", 442).
-spec schedule_valid(schedule()) -> boolean().
schedule_valid(Schedule) ->
    case Schedule of
        manual ->
            true;

        {interval, Interval_ms} ->
            Interval_ms > 0;

        {cron, Expression} ->
            Expression /= <<""/utf8>>
    end.

-file("src/lightspeed/pipeline.gleam", 479).
-spec last_stage_kind(list(stage())) -> gleam@option:option(stage_kind()).
last_stage_kind(Stages) ->
    case lists:reverse(Stages) of
        [] ->
            none;

        [Stage | _] ->
            {some, erlang:element(3, Stage)}
    end.

-file("src/lightspeed/pipeline.gleam", 472).
-spec first_stage_kind(list(stage())) -> gleam@option:option(stage_kind()).
first_stage_kind(Stages) ->
    case Stages of
        [] ->
            none;

        [Stage | _] ->
            {some, erlang:element(3, Stage)}
    end.

-file("src/lightspeed/pipeline.gleam", 133).
?DOC(" Validate one pipeline contract.\n").
-spec valid_pipeline(pipeline()) -> boolean().
valid_pipeline(Pipeline) ->
    case (erlang:element(2, Pipeline) =:= <<""/utf8>>) orelse (erlang:element(
        4,
        Pipeline
    )
    =:= []) of
        true ->
            false;

        false ->
            First_kind = first_stage_kind(erlang:element(4, Pipeline)),
            Last_kind = last_stage_kind(erlang:element(4, Pipeline)),
            (((schedule_valid(erlang:element(3, Pipeline)) andalso (First_kind
            =:= {some, source}))
            andalso (Last_kind =:= {some, sink}))
            andalso stage_names_unique(erlang:element(4, Pipeline), []))
            andalso stages_valid(erlang:element(4, Pipeline))
    end.

-file("src/lightspeed/pipeline.gleam", 125).
?DOC(" Validate pipeline and runtime invariants.\n").
-spec valid(runtime()) -> boolean().
valid(Runtime) ->
    ((valid_pipeline(erlang:element(2, Runtime)) andalso (erlang:element(
        4,
        Runtime
    )
    > 0))
    andalso checkpoints_valid(checkpoints(Runtime)))
    andalso lightspeed@pipeline@telemetry:valid(erlang:element(7, Runtime)).

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

        false ->
            Right
    end.

-file("src/lightspeed/pipeline.gleam", 150).
?DOC(" Schedule one runtime.\n").
-spec schedule(runtime(), integer()) -> runtime().
schedule(Runtime, At_ms) ->
    {runtime,
        erlang:element(2, Runtime),
        {scheduled, max(0, At_ms)},
        erlang:element(4, Runtime),
        erlang:element(5, Runtime),
        erlang:element(6, Runtime),
        erlang:element(7, Runtime)}.

-file("src/lightspeed/pipeline.gleam", 155).
?DOC(" Trigger one runtime.\n").
-spec trigger(runtime(), binary(), integer()) -> runtime().
trigger(Runtime, Reason, At_ms) ->
    {runtime,
        erlang:element(2, Runtime),
        {triggered, Reason, max(0, At_ms)},
        erlang:element(4, Runtime),
        erlang:element(5, Runtime),
        erlang:element(6, Runtime),
        erlang:element(7, Runtime)}.

-file("src/lightspeed/pipeline.gleam", 505).
-spec next_run_id(integer()) -> binary().
next_run_id(Seq) ->
    <<"run-"/utf8, (erlang:integer_to_binary(Seq))/binary>>.

-file("src/lightspeed/pipeline.gleam", 160).
?DOC(" Start one run.\n").
-spec start(runtime(), integer()) -> runtime().
start(Runtime, Now_ms) ->
    Run_id = next_run_id(erlang:element(4, Runtime)),
    {runtime,
        erlang:element(2, Runtime),
        {running, Run_id, max(0, Now_ms)},
        erlang:element(4, Runtime) + 1,
        erlang:element(5, Runtime),
        erlang:element(6, Runtime),
        erlang:element(7, Runtime)}.

-file("src/lightspeed/pipeline.gleam", 497).
-spec active_started_ms(lifecycle()) -> integer().
active_started_ms(Lifecycle) ->
    case Lifecycle of
        {running, _, Started_ms} ->
            Started_ms;

        {replaying, _, _, Started_ms@1} ->
            Started_ms@1;

        _ ->
            0
    end.

-file("src/lightspeed/pipeline.gleam", 531).
-spec total_offset(list(lightspeed@pipeline@checkpoint:checkpoint())) -> integer().
total_offset(Entries_rev) ->
    case Entries_rev of
        [] ->
            0;

        [Entry | _] ->
            erlang:element(2, erlang:element(5, Entry))
    end.

-file("src/lightspeed/pipeline.gleam", 509).
-spec find_stage(list(stage()), binary()) -> gleam@option:option(stage()).
find_stage(Stages, Name) ->
    case Stages of
        [] ->
            none;

        [Stage | Rest] ->
            case erlang:element(2, Stage) =:= Name of
                true ->
                    {some, Stage};

                false ->
                    find_stage(Rest, Name)
            end
    end.

-file("src/lightspeed/pipeline.gleam", 486).
-spec active_run(lifecycle()) -> gleam@option:option(binary()).
active_run(Lifecycle) ->
    case Lifecycle of
        {running, Run_id, _} ->
            {some, Run_id};

        {replaying, Run_id@1, _, _} ->
            {some, Run_id@1};

        {failed, Run_id@2, _} ->
            {some, Run_id@2};

        {crashed, Run_id@3, _} ->
            {some, Run_id@3};

        {completed, Run_id@4, _} ->
            {some, Run_id@4};

        _ ->
            none
    end.

-file("src/lightspeed/pipeline.gleam", 170).
?DOC(" Process one stage event.\n").
-spec process(runtime(), binary(), integer(), integer(), binary(), integer()) -> {runtime(),
    process_result()}.
process(Runtime, Stage_name, Processed_records, Lag_ms, Idempotency_key, Now_ms) ->
    case active_run(erlang:element(3, Runtime)) of
        none ->
            {Runtime, {rejected, <<"not_running"/utf8>>}};

        {some, Run_id} ->
            case find_stage(
                erlang:element(4, erlang:element(2, Runtime)),
                Stage_name
            ) of
                none ->
                    {Runtime,
                        {rejected, <<"unknown_stage:"/utf8, Stage_name/binary>>}};

                {some, Stage} ->
                    case Processed_records =< 0 of
                        true ->
                            {Runtime,
                                {rejected,
                                    <<"invalid_processed_records:"/utf8,
                                        (erlang:integer_to_binary(
                                            Processed_records
                                        ))/binary>>}};

                        false ->
                            case (erlang:element(3, Stage) =:= sink) andalso has_key(
                                erlang:element(6, Runtime),
                                Idempotency_key
                            ) of
                                true ->
                                    {Runtime,
                                        {duplicate_suppressed,
                                            erlang:element(2, Stage),
                                            Idempotency_key}};

                                false ->
                                    Sequence = erlang:length(
                                        erlang:element(5, Runtime)
                                    )
                                    + 1,
                                    Watermark = lightspeed@pipeline@checkpoint:watermark(
                                        total_offset(erlang:element(5, Runtime))
                                        + Processed_records,
                                        max(0, Now_ms)
                                    ),
                                    Entry = lightspeed@pipeline@checkpoint:checkpoint(
                                        Run_id,
                                        erlang:element(2, Stage),
                                        Sequence,
                                        Watermark,
                                        Idempotency_key,
                                        max(0, Now_ms)
                                    ),
                                    Started_ms = active_started_ms(
                                        erlang:element(3, Runtime)
                                    ),
                                    Snapshot = lightspeed@pipeline@telemetry:apply(
                                        erlang:element(7, Runtime),
                                        lightspeed@pipeline@telemetry:delta(
                                            Processed_records,
                                            max(0, Lag_ms),
                                            0,
                                            0,
                                            max(1, max(0, Now_ms) - Started_ms)
                                        )
                                    ),
                                    Keys = case erlang:element(3, Stage) =:= sink of
                                        true ->
                                            [Idempotency_key |
                                                erlang:element(6, Runtime)];

                                        false ->
                                            erlang:element(6, Runtime)
                                    end,
                                    {{runtime,
                                            erlang:element(2, Runtime),
                                            erlang:element(3, Runtime),
                                            erlang:element(4, Runtime),
                                            [Entry | erlang:element(5, Runtime)],
                                            Keys,
                                            Snapshot},
                                        {applied, Entry}}
                            end
                    end
            end
    end.

-file("src/lightspeed/pipeline.gleam", 257).
?DOC(" Record retries for the active run.\n").
-spec record_retry(runtime(), integer()) -> runtime().
record_retry(Runtime, Count) ->
    {runtime,
        erlang:element(2, Runtime),
        erlang:element(3, Runtime),
        erlang:element(4, Runtime),
        erlang:element(5, Runtime),
        erlang:element(6, Runtime),
        lightspeed@pipeline@telemetry:apply(
            erlang:element(7, Runtime),
            lightspeed@pipeline@telemetry:delta(
                0,
                erlang:element(2, erlang:element(7, Runtime)),
                max(0, Count),
                0,
                1
            )
        )}.

-file("src/lightspeed/pipeline.gleam", 268).
?DOC(" Record dead-letter increments for the active run.\n").
-spec record_dead_letter(runtime(), integer()) -> runtime().
record_dead_letter(Runtime, Count) ->
    {runtime,
        erlang:element(2, Runtime),
        erlang:element(3, Runtime),
        erlang:element(4, Runtime),
        erlang:element(5, Runtime),
        erlang:element(6, Runtime),
        lightspeed@pipeline@telemetry:apply(
            erlang:element(7, Runtime),
            lightspeed@pipeline@telemetry:delta(
                0,
                erlang:element(2, erlang:element(7, Runtime)),
                0,
                max(0, Count),
                1
            )
        )}.

-file("src/lightspeed/pipeline.gleam", 279).
?DOC(" Mark active run as failed.\n").
-spec fail(runtime(), binary()) -> runtime().
fail(Runtime, Reason) ->
    case active_run(erlang:element(3, Runtime)) of
        none ->
            Runtime;

        {some, Run_id} ->
            {runtime,
                erlang:element(2, Runtime),
                {failed, Run_id, Reason},
                erlang:element(4, Runtime),
                erlang:element(5, Runtime),
                erlang:element(6, Runtime),
                erlang:element(7, Runtime)}
    end.

-file("src/lightspeed/pipeline.gleam", 288).
?DOC(" Mark active run as crashed.\n").
-spec crash(runtime(), binary()) -> runtime().
crash(Runtime, Reason) ->
    case active_run(erlang:element(3, Runtime)) of
        none ->
            Runtime;

        {some, Run_id} ->
            {runtime,
                erlang:element(2, Runtime),
                {crashed, Run_id, Reason},
                erlang:element(4, Runtime),
                erlang:element(5, Runtime),
                erlang:element(6, Runtime),
                erlang:element(7, Runtime)}
    end.

-file("src/lightspeed/pipeline.gleam", 297).
?DOC(" Resume from the latest checkpoint.\n").
-spec resume_from_latest_checkpoint(runtime(), integer()) -> {ok, runtime()} |
    {error, binary()}.
resume_from_latest_checkpoint(Runtime, Now_ms) ->
    case lightspeed@pipeline@checkpoint:latest(checkpoints(Runtime)) of
        none ->
            {error, <<"missing_checkpoint"/utf8>>};

        {some, Entry} ->
            Run_id = <<(next_run_id(erlang:element(4, Runtime)))/binary,
                "-replay"/utf8>>,
            {ok,
                {runtime,
                    erlang:element(2, Runtime),
                    {replaying,
                        Run_id,
                        erlang:element(3, Entry),
                        max(0, Now_ms)},
                    erlang:element(4, Runtime) + 1,
                    erlang:element(5, Runtime),
                    erlang:element(6, Runtime),
                    erlang:element(7, Runtime)}}
    end.

-file("src/lightspeed/pipeline.gleam", 321).
?DOC(" Complete the active run.\n").
-spec complete(runtime(), integer()) -> runtime().
complete(Runtime, Now_ms) ->
    case active_run(erlang:element(3, Runtime)) of
        none ->
            Runtime;

        {some, Run_id} ->
            {runtime,
                erlang:element(2, Runtime),
                {completed, Run_id, max(0, Now_ms)},
                erlang:element(4, Runtime),
                erlang:element(5, Runtime),
                erlang:element(6, Runtime),
                erlang:element(7, Runtime)}
    end.

-file("src/lightspeed/pipeline.gleam", 333).
?DOC(" Runtime pipeline accessor.\n").
-spec runtime_pipeline(runtime()) -> pipeline().
runtime_pipeline(Runtime) ->
    erlang:element(2, Runtime).

-file("src/lightspeed/pipeline.gleam", 338).
?DOC(" Runtime lifecycle accessor.\n").
-spec lifecycle(runtime()) -> lifecycle().
lifecycle(Runtime) ->
    erlang:element(3, Runtime).

-file("src/lightspeed/pipeline.gleam", 343).
?DOC(" Runtime telemetry accessor.\n").
-spec runtime_telemetry(runtime()) -> lightspeed@pipeline@telemetry:snapshot().
runtime_telemetry(Runtime) ->
    erlang:element(7, Runtime).

-file("src/lightspeed/pipeline.gleam", 353).
?DOC(" Sink idempotency keys in application order.\n").
-spec sink_idempotency_keys(runtime()) -> list(binary()).
sink_idempotency_keys(Runtime) ->
    lists:reverse(erlang:element(6, Runtime)).

-file("src/lightspeed/pipeline.gleam", 358).
?DOC(" Stage kind label.\n").
-spec stage_kind_label(stage_kind()) -> binary().
stage_kind_label(Kind) ->
    case Kind of
        source ->
            <<"source"/utf8>>;

        transform ->
            <<"transform"/utf8>>;

        sink ->
            <<"sink"/utf8>>
    end.

-file("src/lightspeed/pipeline.gleam", 367).
?DOC(" Schedule label.\n").
-spec schedule_label(schedule()) -> binary().
schedule_label(Schedule) ->
    case Schedule of
        manual ->
            <<"manual"/utf8>>;

        {interval, Interval_ms} ->
            <<"interval:"/utf8, (erlang:integer_to_binary(Interval_ms))/binary>>;

        {cron, Expression} ->
            <<"cron:"/utf8, Expression/binary>>
    end.

-file("src/lightspeed/pipeline.gleam", 376).
?DOC(" Lifecycle label.\n").
-spec lifecycle_label(lifecycle()) -> binary().
lifecycle_label(Lifecycle) ->
    case Lifecycle of
        idle ->
            <<"idle"/utf8>>;

        {scheduled, At_ms} ->
            <<"scheduled:"/utf8, (erlang:integer_to_binary(At_ms))/binary>>;

        {triggered, Reason, At_ms@1} ->
            <<<<<<"triggered:"/utf8, Reason/binary>>/binary, ":"/utf8>>/binary,
                (erlang:integer_to_binary(At_ms@1))/binary>>;

        {running, Run_id, Started_ms} ->
            <<<<<<"running:"/utf8, Run_id/binary>>/binary, ":"/utf8>>/binary,
                (erlang:integer_to_binary(Started_ms))/binary>>;

        {failed, Run_id@1, Reason@1} ->
            <<<<<<"failed:"/utf8, Run_id@1/binary>>/binary, ":"/utf8>>/binary,
                Reason@1/binary>>;

        {crashed, Run_id@2, Reason@2} ->
            <<<<<<"crashed:"/utf8, Run_id@2/binary>>/binary, ":"/utf8>>/binary,
                Reason@2/binary>>;

        {replaying, Run_id@3, From_stage, Started_ms@1} ->
            <<<<<<<<<<"replaying:"/utf8, Run_id@3/binary>>/binary, ":"/utf8>>/binary,
                        From_stage/binary>>/binary,
                    ":"/utf8>>/binary,
                (erlang:integer_to_binary(Started_ms@1))/binary>>;

        {completed, Run_id@4, Finished_ms} ->
            <<<<<<"completed:"/utf8, Run_id@4/binary>>/binary, ":"/utf8>>/binary,
                (erlang:integer_to_binary(Finished_ms))/binary>>
    end.

-file("src/lightspeed/pipeline.gleam", 399).
?DOC(" Stable process-result label.\n").
-spec process_result_label(process_result()) -> binary().
process_result_label(Result) ->
    case Result of
        {applied, Entry} ->
            <<"applied:"/utf8,
                (lightspeed@pipeline@checkpoint:checkpoint_label(Entry))/binary>>;

        {duplicate_suppressed, Stage, Key} ->
            <<<<<<"duplicate_suppressed:"/utf8, Stage/binary>>/binary,
                    ":"/utf8>>/binary,
                Key/binary>>;

        {rejected, Reason} ->
            <<"rejected:"/utf8, Reason/binary>>
    end.

-file("src/lightspeed/pipeline.gleam", 545).
-spec join_with(binary(), list(binary())) -> binary().
join_with(Separator, Values) ->
    case Values of
        [] ->
            <<""/utf8>>;

        [Value] ->
            Value;

        [Value@1 | Rest] ->
            <<<<Value@1/binary, Separator/binary>>/binary,
                (join_with(Separator, Rest))/binary>>
    end.

-file("src/lightspeed/pipeline.gleam", 432).
-spec stage_signature(stage()) -> binary().
stage_signature(Stage) ->
    <<<<<<<<<<<<(erlang:element(2, Stage))/binary, ":"/utf8>>/binary,
                        (stage_kind_label(erlang:element(3, Stage)))/binary>>/binary,
                    ":"/utf8>>/binary,
                (erlang:element(2, erlang:element(4, Stage)))/binary>>/binary,
            "->"/utf8>>/binary,
        (erlang:element(3, erlang:element(4, Stage)))/binary>>.

-file("src/lightspeed/pipeline.gleam", 423).
?DOC(" Stable pipeline signature.\n").
-spec pipeline_signature(pipeline()) -> binary().
pipeline_signature(Pipeline) ->
    <<<<<<<<<<"name="/utf8, (erlang:element(2, Pipeline))/binary>>/binary,
                    "|schedule="/utf8>>/binary,
                (schedule_label(erlang:element(3, Pipeline)))/binary>>/binary,
            "|stages="/utf8>>/binary,
        (join_with(
            <<","/utf8>>,
            gleam@list:map(erlang:element(4, Pipeline), fun stage_signature/1)
        ))/binary>>.

-file("src/lightspeed/pipeline.gleam", 409).
?DOC(" Stable runtime signature.\n").
-spec signature(runtime()) -> binary().
signature(Runtime) ->
    <<<<<<<<<<<<<<<<<<"pipeline="/utf8,
                                        (pipeline_signature(
                                            erlang:element(2, Runtime)
                                        ))/binary>>/binary,
                                    "|lifecycle="/utf8>>/binary,
                                (lifecycle_label(erlang:element(3, Runtime)))/binary>>/binary,
                            "|checkpoints="/utf8>>/binary,
                        (lightspeed@pipeline@checkpoint:chain_signature(
                            checkpoints(Runtime)
                        ))/binary>>/binary,
                    "|sink_keys="/utf8>>/binary,
                (join_with(<<","/utf8>>, sink_idempotency_keys(Runtime)))/binary>>/binary,
            "|telemetry="/utf8>>/binary,
        (lightspeed@pipeline@telemetry:label(erlang:element(7, Runtime)))/binary>>.