%%%-------------------------------------------------------------------
%%% @doc Relay-routed Erlang distribution.
%%%
%%% When MACULA_DIST_MODE=relay, Erlang distribution connections
%%% are tunneled through the Macula relay mesh instead of direct QUIC.
%%% This enables distribution across firewalls and NATs — nodes only
%%% need outbound connectivity to a relay.
%%%
%%% The tunnel works by:
%%% 1. Node A publishes a DIST_CONNECT request to the mesh
%%% 2. The relay forwards it to Node B's handler
%%% 3. Node B opens a distribution stream back through the relay
%%% 4. Raw ETF bytes flow through the relay tunnel
%%%
%%% This is EXPERIMENTAL. Supports Pid ! Msg, gen_server:call,
%%% pg groups, process monitoring. Does NOT guarantee Mnesia or
%%% global module compatibility over WAN latency.
%%% @end
%%%-------------------------------------------------------------------
-module(macula_dist_relay).

-include_lib("kernel/include/logger.hrl").

-export([connect/2, accept_dist/2]).
-export([is_relay_mode/0, get_mesh_client/0]).

-define(DIST_TOPIC, <<"_dist.connect">>).
-define(DIST_TIMEOUT, 10000).

%% @doc Check if relay distribution mode is enabled.
-spec is_relay_mode() -> boolean().
is_relay_mode() ->
    case os:getenv("MACULA_DIST_MODE") of
        "relay" -> true;
        _ -> false
    end.

%% @doc Connect to a remote node via relay mesh.
%% Returns {ok, Conn, Stream} where the stream carries ETF bytes
%% through the relay tunnel.
-spec connect(string(), integer()) -> {ok, reference(), reference()} | {error, term()}.
connect(Host, Port) ->
    NodeName = lists:flatten(io_lib:format("~b@~s", [Port, Host])),
    ?LOG_INFO("[dist_relay] Connecting to ~s via mesh", [NodeName]),

    %% Step 1: Find which relay the target node is connected to
    %% For now, use the same relay we're connected to (they share the mesh)
    case get_mesh_client() of
        undefined ->
            {error, no_mesh_connection};
        MeshClient ->
            %% Step 2: Request a distribution tunnel via mesh RPC
            Procedure = <<"_dist.tunnel.", (list_to_binary(NodeName))/binary>>,
            case macula_relay_client:call(MeshClient, Procedure, #{
                <<"from_node">> => atom_to_binary(node()),
                <<"target_node">> => list_to_binary(NodeName)
            }, ?DIST_TIMEOUT) of
                {ok, #{<<"tunnel_id">> := TunnelId}} ->
                    ?LOG_INFO("[dist_relay] Tunnel established: ~s", [TunnelId]),
                    %% Step 3: The tunnel_id identifies a relay-side bridge.
                    %% We use the mesh QUIC connection's stream for ETF bytes.
                    %% For the experimental version, we reuse the existing
                    %% relay_client connection and multiplex dist traffic
                    %% as pub/sub messages on a private topic.
                    create_dist_socket(MeshClient, TunnelId);
                {error, Reason} ->
                    ?LOG_WARNING("[dist_relay] Tunnel request failed: ~p", [Reason]),
                    {error, {tunnel_failed, Reason}}
            end
    end.

%% @doc Accept an incoming distribution connection via relay mesh.
%% Called when a remote node requests a tunnel to us.
-spec accept_dist(binary(), map()) -> {ok, pid()} | {error, term()}.
accept_dist(TunnelId, Opts) ->
    ?LOG_INFO("[dist_relay] Accepting dist tunnel: ~s", [TunnelId]),
    %% The accepting side creates a process that bridges
    %% mesh pub/sub messages to the dist_util handshake
    FromNode = maps:get(<<"from_node">>, Opts, <<>>),
    spawn_link(fun() -> dist_tunnel_process(TunnelId, FromNode) end),
    {ok, TunnelId}.

%%====================================================================
%% Internal
%%====================================================================

get_mesh_client() ->
    case whereis(hecate_mesh_client) of
        undefined ->
            %% Try macula_multi_relay directly
            case whereis(macula_multi_relay) of
                undefined -> undefined;
                Pid -> Pid
            end;
        _Pid ->
            case hecate_mesh_client:get_client() of
                {ok, Client} -> Client;
                _ -> undefined
            end
    end.

create_dist_socket(MeshClient, TunnelId) ->
    %% Create a pair of linked processes that act as a socket:
    %% - Send: publish ETF bytes to tunnel topic
    %% - Recv: subscribe to tunnel topic, deliver as {quic, Data, ...}
    SendTopic = <<"_dist.data.", TunnelId/binary, ".out">>,
    RecvTopic = <<"_dist.data.", TunnelId/binary, ".in">>,
    Self = self(),

    %% Subscribe to incoming data
    {ok, _SubRef} = macula_relay_client:subscribe(MeshClient, RecvTopic,
        fun(#{payload := Data}) ->
            Self ! {dist_data, Data}
        end),

    {ok, MeshClient, {tunnel, TunnelId, SendTopic, RecvTopic}}.

dist_tunnel_process(_TunnelId, _FromNode) ->
    %% This process lives for the duration of the distribution connection
    receive
        stop -> ok
    end.