-module(lightspeed@cluster@durable_session).
-compile([no_auto_import, nowarn_unused_vars, nowarn_unused_function, nowarn_nomatch, inline]).
-define(FILEPATH, "src/lightspeed/cluster/durable_session.gleam").
-export([supported_backends/0, expected_token/3, start/8, append_counter_delta/3, crash/2, restart/2, request_takeover/5, reconnect/4, journal/1, continuity_report/3, backend_label/1, resume_policy_label/1, lifecycle_label/1, journal_entry_label/1, ownership_error_label/1, fence_signature/1, continuity_signature/1, snapshot_signature/1, signature/1, valid/1, backend/1, fence/1, snapshot/1]).
-export_type([persistence_backend/0, resume_policy/0, lifecycle_state/0, journal_entry/0, snapshot/0, ownership_fence/0, durable_session/0, ownership_error/0, continuity_report/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(" Durable cluster-session contracts and ownership fencing for M27.\n").

-type persistence_backend() :: {snapshot_only, binary(), integer()} |
    {journal_only, binary()} |
    {snapshot_journal, binary(), binary(), integer()}.

-type resume_policy() :: resume | rehydrate | remount.

-type lifecycle_state() :: live | disconnected | crashed.

-type journal_entry() :: {counter_delta, integer()} |
    {crash_marked, binary()} |
    {restarted, integer()} |
    {reconnected, resume_policy(), binary(), integer()}.

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

-type ownership_fence() :: {ownership_fence, binary(), integer(), binary()}.

-type durable_session() :: {durable_session,
        persistence_backend(),
        resume_policy(),
        ownership_fence(),
        snapshot(),
        list(journal_entry()),
        integer(),
        integer()}.

-type ownership_error() :: {stale_owner, integer(), integer()} |
    {split_brain, binary(), binary(), integer()} |
    {invalid_token, binary(), integer()}.

-type continuity_report() :: {continuity_report,
        integer(),
        integer(),
        boolean()}.

-file("src/lightspeed/cluster/durable_session.gleam", 86).
?DOC(" Supported backend matrix for M27 certification.\n").
-spec supported_backends() -> list(persistence_backend()).
supported_backends() ->
    [{snapshot_only, <<"snapshot_store_a"/utf8>>, 2},
        {journal_only, <<"journal_store_a"/utf8>>},
        {snapshot_journal,
            <<"snapshot_store_b"/utf8>>,
            <<"journal_store_b"/utf8>>,
            2}].

-file("src/lightspeed/cluster/durable_session.gleam", 274).
?DOC(" Expected deterministic fence token.\n").
-spec expected_token(binary(), binary(), integer()) -> binary().
expected_token(Session_id, Owner, Epoch) ->
    <<<<<<<<<<"fence:"/utf8, Session_id/binary>>/binary, ":"/utf8>>/binary,
                Owner/binary>>/binary,
            ":"/utf8>>/binary,
        (erlang:integer_to_binary(Epoch))/binary>>.

-file("src/lightspeed/cluster/durable_session.gleam", 99).
?DOC(" Start a durable session contract.\n").
-spec start(
    binary(),
    binary(),
    binary(),
    persistence_backend(),
    resume_policy(),
    integer(),
    integer(),
    integer()
) -> durable_session().
start(
    Session_id,
    Owner,
    Route,
    Backend,
    Resume_policy,
    Now_ms,
    Continuity_slo_ms,
    Max_data_loss_events
) ->
    Fence = {ownership_fence, Owner, 1, expected_token(Session_id, Owner, 1)},
    {durable_session,
        Backend,
        Resume_policy,
        Fence,
        {snapshot, Session_id, Route, 0, disconnected, 1, Now_ms},
        [],
        Continuity_slo_ms,
        Max_data_loss_events}.

-file("src/lightspeed/cluster/durable_session.gleam", 135).
?DOC(" Append one counter-delta entry and update the snapshot.\n").
-spec append_counter_delta(durable_session(), integer(), integer()) -> durable_session().
append_counter_delta(Session, Delta, Now_ms) ->
    {durable_session,
        erlang:element(2, Session),
        erlang:element(3, Session),
        erlang:element(4, Session),
        begin
            _record = erlang:element(5, Session),
            {snapshot,
                erlang:element(2, _record),
                erlang:element(3, _record),
                erlang:element(4, erlang:element(5, Session)) + Delta,
                erlang:element(5, _record),
                erlang:element(6, _record),
                Now_ms}
        end,
        [{counter_delta, Delta} | erlang:element(6, Session)],
        erlang:element(7, Session),
        erlang:element(8, Session)}.

-file("src/lightspeed/cluster/durable_session.gleam", 152).
?DOC(" Mark a crash in snapshot + journal.\n").
-spec crash(durable_session(), binary()) -> durable_session().
crash(Session, Reason) ->
    {durable_session,
        erlang:element(2, Session),
        erlang:element(3, Session),
        erlang:element(4, Session),
        begin
            _record = erlang:element(5, Session),
            {snapshot,
                erlang:element(2, _record),
                erlang:element(3, _record),
                erlang:element(4, _record),
                crashed,
                erlang:element(6, _record),
                erlang:element(7, _record)}
        end,
        [{crash_marked, Reason} | erlang:element(6, Session)],
        erlang:element(7, Session),
        erlang:element(8, Session)}.

-file("src/lightspeed/cluster/durable_session.gleam", 161).
?DOC(" Mark a restart in snapshot + journal.\n").
-spec restart(durable_session(), integer()) -> durable_session().
restart(Session, Now_ms) ->
    {durable_session,
        erlang:element(2, Session),
        erlang:element(3, Session),
        erlang:element(4, Session),
        begin
            _record = erlang:element(5, Session),
            {snapshot,
                erlang:element(2, _record),
                erlang:element(3, _record),
                erlang:element(4, _record),
                disconnected,
                erlang:element(6, _record),
                Now_ms}
        end,
        [{restarted, Now_ms} | erlang:element(6, Session)],
        erlang:element(7, Session),
        erlang:element(8, Session)}.

-file("src/lightspeed/cluster/durable_session.gleam", 174).
?DOC(" Attempt ownership takeover under fencing rules.\n").
-spec request_takeover(
    durable_session(),
    binary(),
    integer(),
    binary(),
    integer()
) -> {ok, durable_session()} | {error, ownership_error()}.
request_takeover(Session, Owner, Epoch, Token, Now_ms) ->
    Current_owner = erlang:element(2, erlang:element(4, Session)),
    Current_epoch = erlang:element(3, erlang:element(4, Session)),
    Expected = expected_token(
        erlang:element(2, erlang:element(5, Session)),
        Owner,
        Epoch
    ),
    case Epoch < Current_epoch of
        true ->
            {error, {stale_owner, Current_epoch, Epoch}};

        false ->
            case (Epoch =:= Current_epoch) andalso (Owner /= Current_owner) of
                true ->
                    {error, {split_brain, Current_owner, Owner, Epoch}};

                false ->
                    case Token =:= Expected of
                        false ->
                            {error, {invalid_token, Owner, Epoch}};

                        true ->
                            {ok,
                                {durable_session,
                                    erlang:element(2, Session),
                                    erlang:element(3, Session),
                                    {ownership_fence, Owner, Epoch, Token},
                                    begin
                                        _record = erlang:element(5, Session),
                                        {snapshot,
                                            erlang:element(2, _record),
                                            erlang:element(3, _record),
                                            erlang:element(4, _record),
                                            disconnected,
                                            Epoch,
                                            Now_ms}
                                    end,
                                    erlang:element(6, Session),
                                    erlang:element(7, Session),
                                    erlang:element(8, Session)}}
                    end
            end
    end.

-file("src/lightspeed/cluster/durable_session.gleam", 222).
?DOC(" Reconnect using one recovery policy.\n").
-spec reconnect(durable_session(), resume_policy(), binary(), integer()) -> durable_session().
reconnect(Session, Policy, Route, Now_ms) ->
    Counter = case Policy of
        resume ->
            erlang:element(4, erlang:element(5, Session));

        rehydrate ->
            erlang:element(4, erlang:element(5, Session));

        remount ->
            0
    end,
    {durable_session,
        erlang:element(2, Session),
        erlang:element(3, Session),
        erlang:element(4, Session),
        begin
            _record = erlang:element(5, Session),
            {snapshot,
                erlang:element(2, _record),
                Route,
                Counter,
                live,
                erlang:element(6, _record),
                Now_ms}
        end,
        [{reconnected, Policy, Route, Now_ms} | erlang:element(6, Session)],
        erlang:element(7, Session),
        erlang:element(8, Session)}.

-file("src/lightspeed/cluster/durable_session.gleam", 397).
?DOC(" Journal entries in append order.\n").
-spec journal(durable_session()) -> list(journal_entry()).
journal(Session) ->
    lists:reverse(erlang:element(6, Session)).

-file("src/lightspeed/cluster/durable_session.gleam", 439).
-spec estimate_data_loss_events(persistence_backend(), list(journal_entry())) -> integer().
estimate_data_loss_events(Backend, Entries) ->
    case Backend of
        {snapshot_only, _, Checkpoint_every} ->
            case erlang:length(Entries) > Checkpoint_every of
                true ->
                    1;

                false ->
                    0
            end;

        {journal_only, _} ->
            0;

        {snapshot_journal, _, _, _} ->
            0
    end.

-file("src/lightspeed/cluster/durable_session.gleam", 251).
?DOC(" Deterministic continuity SLO report for one failover timeline.\n").
-spec continuity_report(durable_session(), integer(), integer()) -> continuity_report().
continuity_report(Session, Failover_at_ms, Reconnect_at_ms) ->
    Reconnect_latency_ms = case Reconnect_at_ms < Failover_at_ms of
        true ->
            0;

        false ->
            Reconnect_at_ms - Failover_at_ms
    end,
    Estimated_data_loss_events = estimate_data_loss_events(
        erlang:element(2, Session),
        journal(Session)
    ),
    Met = (Reconnect_latency_ms =< erlang:element(7, Session)) andalso (Estimated_data_loss_events
    =< erlang:element(8, Session)),
    {continuity_report, Reconnect_latency_ms, Estimated_data_loss_events, Met}.

-file("src/lightspeed/cluster/durable_session.gleam", 279).
?DOC(" Backend label.\n").
-spec backend_label(persistence_backend()) -> binary().
backend_label(Backend) ->
    case Backend of
        {snapshot_only, Store, Checkpoint_every} ->
            <<<<<<"snapshot_only:"/utf8, Store/binary>>/binary, ":"/utf8>>/binary,
                (erlang:integer_to_binary(Checkpoint_every))/binary>>;

        {journal_only, Store@1} ->
            <<"journal_only:"/utf8, Store@1/binary>>;

        {snapshot_journal, Snapshot_store, Journal_store, Checkpoint_every@1} ->
            <<<<<<<<<<"snapshot_journal:"/utf8, Snapshot_store/binary>>/binary,
                            ":"/utf8>>/binary,
                        Journal_store/binary>>/binary,
                    ":"/utf8>>/binary,
                (erlang:integer_to_binary(Checkpoint_every@1))/binary>>
    end.

-file("src/lightspeed/cluster/durable_session.gleam", 295).
?DOC(" Resume policy label.\n").
-spec resume_policy_label(resume_policy()) -> binary().
resume_policy_label(Policy) ->
    case Policy of
        resume ->
            <<"resume"/utf8>>;

        rehydrate ->
            <<"rehydrate"/utf8>>;

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

-file("src/lightspeed/cluster/durable_session.gleam", 304).
?DOC(" Lifecycle label.\n").
-spec lifecycle_label(lifecycle_state()) -> binary().
lifecycle_label(Lifecycle) ->
    case Lifecycle of
        live ->
            <<"live"/utf8>>;

        disconnected ->
            <<"disconnected"/utf8>>;

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

-file("src/lightspeed/cluster/durable_session.gleam", 313).
?DOC(" Journal entry label.\n").
-spec journal_entry_label(journal_entry()) -> binary().
journal_entry_label(Entry) ->
    case Entry of
        {counter_delta, Delta} ->
            <<"counter_delta:"/utf8, (erlang:integer_to_binary(Delta))/binary>>;

        {crash_marked, Reason} ->
            <<"crash_marked:"/utf8, Reason/binary>>;

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

        {reconnected, Policy, Route, Now_ms@1} ->
            <<<<<<<<<<"reconnected:"/utf8,
                                (resume_policy_label(Policy))/binary>>/binary,
                            ":"/utf8>>/binary,
                        Route/binary>>/binary,
                    ":"/utf8>>/binary,
                (erlang:integer_to_binary(Now_ms@1))/binary>>
    end.

-file("src/lightspeed/cluster/durable_session.gleam", 329).
?DOC(" Ownership error label.\n").
-spec ownership_error_label(ownership_error()) -> binary().
ownership_error_label(Error) ->
    case Error of
        {stale_owner, Current_epoch, Requested_epoch} ->
            <<<<<<"stale_owner:"/utf8,
                        (erlang:integer_to_binary(Current_epoch))/binary>>/binary,
                    ":"/utf8>>/binary,
                (erlang:integer_to_binary(Requested_epoch))/binary>>;

        {split_brain, Current_owner, Requested_owner, Epoch} ->
            <<<<<<<<<<"split_brain:"/utf8, Current_owner/binary>>/binary,
                            ":"/utf8>>/binary,
                        Requested_owner/binary>>/binary,
                    ":"/utf8>>/binary,
                (erlang:integer_to_binary(Epoch))/binary>>;

        {invalid_token, Owner, Epoch@1} ->
            <<<<<<"invalid_token:"/utf8, Owner/binary>>/binary, ":"/utf8>>/binary,
                (erlang:integer_to_binary(Epoch@1))/binary>>
    end.

-file("src/lightspeed/cluster/durable_session.gleam", 349).
?DOC(" Ownership-fence signature.\n").
-spec fence_signature(ownership_fence()) -> binary().
fence_signature(Fence) ->
    <<<<<<<<(erlang:element(2, Fence))/binary, ":"/utf8>>/binary,
                (erlang:integer_to_binary(erlang:element(3, Fence)))/binary>>/binary,
            ":"/utf8>>/binary,
        (erlang:element(4, Fence))/binary>>.

-file("src/lightspeed/cluster/durable_session.gleam", 454).
-spec bool_label(boolean()) -> binary().
bool_label(Value) ->
    case Value of
        true ->
            <<"true"/utf8>>;

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

-file("src/lightspeed/cluster/durable_session.gleam", 354).
?DOC(" Continuity-report signature.\n").
-spec continuity_signature(continuity_report()) -> binary().
continuity_signature(Report) ->
    <<<<<<<<<<"latency_ms="/utf8,
                        (erlang:integer_to_binary(erlang:element(2, Report)))/binary>>/binary,
                    "|data_loss_events="/utf8>>/binary,
                (erlang:integer_to_binary(erlang:element(3, Report)))/binary>>/binary,
            "|met="/utf8>>/binary,
        (bool_label(erlang:element(4, Report)))/binary>>.

-file("src/lightspeed/cluster/durable_session.gleam", 461).
-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/cluster/durable_session.gleam", 382).
?DOC(" Snapshot signature.\n").
-spec snapshot_signature(snapshot()) -> binary().
snapshot_signature(Snapshot) ->
    <<<<<<<<<<<<<<<<<<<<(erlang:element(2, Snapshot))/binary, ":"/utf8>>/binary,
                                        (erlang:element(3, Snapshot))/binary>>/binary,
                                    ":counter="/utf8>>/binary,
                                (erlang:integer_to_binary(
                                    erlang:element(4, Snapshot)
                                ))/binary>>/binary,
                            ":lifecycle="/utf8>>/binary,
                        (lifecycle_label(erlang:element(5, Snapshot)))/binary>>/binary,
                    ":epoch="/utf8>>/binary,
                (erlang:integer_to_binary(erlang:element(6, Snapshot)))/binary>>/binary,
            ":checkpoint_ms="/utf8>>/binary,
        (erlang:integer_to_binary(erlang:element(7, Snapshot)))/binary>>.

-file("src/lightspeed/cluster/durable_session.gleam", 364).
?DOC(" Deterministic durable-session signature.\n").
-spec signature(durable_session()) -> binary().
signature(Session) ->
    <<<<<<<<<<<<<<<<<<<<<<<<<<"backend="/utf8,
                                                        (backend_label(
                                                            erlang:element(
                                                                2,
                                                                Session
                                                            )
                                                        ))/binary>>/binary,
                                                    "|resume_policy="/utf8>>/binary,
                                                (resume_policy_label(
                                                    erlang:element(3, Session)
                                                ))/binary>>/binary,
                                            "|fence="/utf8>>/binary,
                                        (fence_signature(
                                            erlang:element(4, Session)
                                        ))/binary>>/binary,
                                    "|snapshot="/utf8>>/binary,
                                (snapshot_signature(erlang:element(5, Session)))/binary>>/binary,
                            "|journal="/utf8>>/binary,
                        (join_with(
                            <<","/utf8>>,
                            gleam@list:map(
                                journal(Session),
                                fun journal_entry_label/1
                            )
                        ))/binary>>/binary,
                    "|continuity_slo_ms="/utf8>>/binary,
                (erlang:integer_to_binary(erlang:element(7, Session)))/binary>>/binary,
            "|max_data_loss_events="/utf8>>/binary,
        (erlang:integer_to_binary(erlang:element(8, Session)))/binary>>.

-file("src/lightspeed/cluster/durable_session.gleam", 430).
-spec valid_backend(persistence_backend()) -> boolean().
valid_backend(Backend) ->
    case Backend of
        {snapshot_only, Store, Checkpoint_every} ->
            (Store /= <<""/utf8>>) andalso (Checkpoint_every > 0);

        {journal_only, Store@1} ->
            Store@1 /= <<""/utf8>>;

        {snapshot_journal, Snapshot_store, Journal_store, Checkpoint_every@1} ->
            ((Snapshot_store /= <<""/utf8>>) andalso (Journal_store /= <<""/utf8>>))
            andalso (Checkpoint_every@1 > 0)
    end.

-file("src/lightspeed/cluster/durable_session.gleam", 402).
?DOC(" Validate one durable-session contract.\n").
-spec valid(durable_session()) -> boolean().
valid(Session) ->
    (((valid_backend(erlang:element(2, Session)) andalso (erlang:element(
        3,
        erlang:element(5, Session)
    )
    /= <<""/utf8>>))
    andalso (erlang:element(7, Session) > 0))
    andalso (erlang:element(8, Session) >= 0))
    andalso (erlang:element(4, erlang:element(4, Session)) =:= expected_token(
        erlang:element(2, erlang:element(5, Session)),
        erlang:element(2, erlang:element(4, Session)),
        erlang:element(3, erlang:element(4, Session))
    )).

-file("src/lightspeed/cluster/durable_session.gleam", 416).
?DOC(" Durable-state backend accessor.\n").
-spec backend(durable_session()) -> persistence_backend().
backend(Session) ->
    erlang:element(2, Session).

-file("src/lightspeed/cluster/durable_session.gleam", 421).
?DOC(" Durable-state fence accessor.\n").
-spec fence(durable_session()) -> ownership_fence().
fence(Session) ->
    erlang:element(4, Session).

-file("src/lightspeed/cluster/durable_session.gleam", 426).
?DOC(" Durable-state snapshot accessor.\n").
-spec snapshot(durable_session()) -> snapshot().
snapshot(Session) ->
    erlang:element(5, Session).