macula

%%%-------------------------------------------------------------------
%%% @doc
%%% Macula Gateway - HTTP/3 Message Router
%%%
%%% Main API module for the Macula Gateway.
%%% The gateway can be embedded in applications or run standalone.
%%%
%%% Architecture:
%%% - QUIC Listener: Accepts HTTP/3 connections from SDK clients
%%% - Router: Routes pub/sub messages between clients
%%% - RPC: Handles remote procedure calls
%%% - Realm Manager: Manages multiple realms
%%%
%%% Usage (Embedded):
%%% ```
%%% {ok, Pid} = macula_gateway:start_link([
%%%     {port, 9443},
%%%     {realm, <<"be.cortexiq.energy">>}
%%% ]).
%%% '''
%%%
%%% Usage (Standalone):
%%% ```
%%% application:start(macula_gateway).
%%% '''
%%% @end
%%%-------------------------------------------------------------------
-module(macula_gateway).

-behaviour(gen_server).

%% API
-export([
    start_link/0,
    start_link/1,
    stop/1,
    get_stats/1
]).

%% gen_server callbacks
-export([
    init/1,
    handle_call/3,
    handle_cast/2,
    handle_info/2,
    terminate/2
]).

-record(state, {
    port :: inet:port_number(),
    realm :: binary(),
    listener :: pid() | undefined,
    clients :: #{pid() => client_info()},
    subscriptions :: #{binary() => [pid()]},  % topic => [client_pids]
    registrations :: #{binary() => pid()}     % procedure => client_pid
}).

-type client_info() :: #{
    realm := binary(),
    node_id := binary(),
    capabilities := [atom()]
}.

-define(DEFAULT_PORT, 9443).
-define(DEFAULT_REALM, <<"macula.default">>).

%%%===================================================================
%%% API Functions
%%%===================================================================

%% @doc Start the gateway with default options.
-spec start_link() -> {ok, pid()} | {error, term()}.
start_link() ->
    start_link([]).

%% @doc Start the gateway with custom options.
%% Options:
%%   {port, Port} - Listen port (default: 9443)
%%   {realm, Realm} - Default realm (default: "macula.default")
-spec start_link(proplists:proplist()) -> {ok, pid()} | {error, term()}.
start_link(Opts) ->
    gen_server:start_link(?MODULE, Opts, []).

%% @doc Stop the gateway.
-spec stop(pid()) -> ok.
stop(Gateway) ->
    gen_server:stop(Gateway).

%% @doc Get gateway statistics.
-spec get_stats(pid()) -> map().
get_stats(Gateway) ->
    gen_server:call(Gateway, get_stats).

%%%===================================================================
%%% gen_server callbacks
%%%===================================================================

init(Opts) ->
    Port = proplists:get_value(port, Opts, ?DEFAULT_PORT),
    Realm = proplists:get_value(realm, Opts, ?DEFAULT_REALM),

    %% Get TLS certificates
    %% Priority: 1. Environment variables (production)
    %%           2. Pre-generated certs (development)
    {CertFile, KeyFile} = get_tls_certificates(),

    io:format("Using TLS certificates:~n"),
    io:format("  Cert: ~s~n", [CertFile]),
    io:format("  Key:  ~s~n", [KeyFile]),

    %% Validate certificate files exist
    case macula_quic_cert:validate_files(CertFile, KeyFile) of
        ok ->
            start_quic_listener(Port, Realm, CertFile, KeyFile);
        {error, Reason} ->
            io:format("Certificate validation failed: ~p~n", [Reason]),
            {stop, {cert_validation_failed, Reason}}
    end.

%% @private
%% @doc Get TLS certificate paths from environment or use pre-generated ones.
get_tls_certificates() ->
    case {os:getenv("TLS_CERT_FILE"), os:getenv("TLS_KEY_FILE")} of
        {false, false} ->
            %% No env vars, use pre-generated certs
            io:format("Using pre-generated TLS certificates~n"),
            {"/opt/macula/certs/cert.pem", "/opt/macula/certs/key.pem"};
        {CertEnv, KeyEnv} when CertEnv =/= false andalso KeyEnv =/= false ->
            %% Use mounted certificates (production)
            io:format("Using mounted TLS certificates from environment~n"),
            {CertEnv, KeyEnv};
        _ ->
            %% Partial configuration, log warning and use defaults
            io:format("WARNING: Partial TLS environment config, using pre-generated certs~n"),
            {"/opt/macula/certs/cert.pem", "/opt/macula/certs/key.pem"}
    end.

%% @private
%% @doc Start the QUIC listener with given certificates.
start_quic_listener(Port, Realm, CertFile, KeyFile) ->
    %% Start QUIC listener
    ListenOpts = [
        {cert, CertFile},
        {key, KeyFile},
        {alpn, ["macula"]},
        {peer_unidi_stream_count, 3}
    ],

    case macula_quic:listen(Port, ListenOpts) of
        {ok, Listener} ->
            io:format("Macula Gateway listening on port ~p (realm: ~s)~n", [Port, Realm]),

            %% Mark health server as ready
            try
                macula_gateway_health:set_ready(true)
            catch
                _:_ -> ok  % Health server might not be running in embedded mode
            end,

            %% Register diagnostics procedures
            try
                macula_gateway_diagnostics:register_procedures(self())
            catch
                _:_ -> ok  % Diagnostics service might not be running in embedded mode
            end,

            %% Start accepting connections
            self() ! accept,

            State = #state{
                port = Port,
                realm = Realm,
                listener = Listener,
                clients = #{},
                subscriptions = #{},
                registrations = #{}
            },

            {ok, State};

        {error, Reason} ->
            io:format("QUIC listen failed: ~p~n", [Reason]),
            {stop, {listen_failed, Reason}};

        {error, Type, Details} ->
            io:format("QUIC listen failed: ~p ~p~n", [Type, Details]),
            {stop, {listen_failed, {Type, Details}}};

        Other ->
            io:format("QUIC listen unexpected result: ~p~n", [Other]),
            {stop, {listen_failed, Other}}
    end.

handle_call(get_stats, _From, State) ->
    Stats = #{
        port => State#state.port,
        realm => State#state.realm,
        clients => maps:size(State#state.clients),
        subscriptions => maps:size(State#state.subscriptions),
        registrations => maps:size(State#state.registrations)
    },
    {reply, Stats, State};

handle_call(_Request, _From, State) ->
    {reply, {error, unknown_request}, State}.

handle_cast(_Request, State) ->
    {noreply, State}.

handle_info(accept, #state{listener = Listener} = State) ->
    %% Accept incoming connection
    case macula_quic:accept(Listener, 5000) of
        {ok, Conn} ->
            %% Spawn handler for this connection
            spawn_link(fun() -> handle_client_connection(self(), Conn, State) end),

            %% Continue accepting
            self() ! accept,

            {noreply, State};

        {error, timeout} ->
            %% No connection, keep accepting
            self() ! accept,
            {noreply, State};

        {error, Reason} ->
            io:format("Accept error: ~p~n", [Reason]),
            {stop, {accept_error, Reason}, State}
    end;

%% Client registered
handle_info({client_connected, ClientPid, ClientInfo}, State) ->
    io:format("Client connected: ~p~n", [ClientInfo]),

    %% Monitor client
    erlang:monitor(process, ClientPid),

    Clients = maps:put(ClientPid, ClientInfo, State#state.clients),
    {noreply, State#state{clients = Clients}};

%% Client disconnected
handle_info({'DOWN', _Ref, process, ClientPid, _Reason}, State) ->
    io:format("Client disconnected: ~p~n", [ClientPid]),

    %% Remove client from all subscriptions
    Subscriptions = maps:map(fun(_Topic, Subscribers) ->
        lists:delete(ClientPid, Subscribers)
    end, State#state.subscriptions),

    %% Remove client registrations
    Registrations = maps:filter(fun(_Proc, Pid) ->
        Pid =/= ClientPid
    end, State#state.registrations),

    %% Remove client
    Clients = maps:remove(ClientPid, State#state.clients),

    {noreply, State#state{
        clients = Clients,
        subscriptions = Subscriptions,
        registrations = Registrations
    }};

%% Publish message (from client)
handle_info({publish, FromPid, Topic, Payload}, State) ->
    %% Get all subscribers to this topic
    Subscribers = maps:get(Topic, State#state.subscriptions, []),

    %% Send to all subscribers (except sender)
    [SubPid ! {event, Topic, Payload} || SubPid <- Subscribers, SubPid =/= FromPid],

    {noreply, State};

%% Subscribe request
handle_info({subscribe, ClientPid, Topic}, State) ->
    %% Add client to topic subscribers
    Subscribers = maps:get(Topic, State#state.subscriptions, []),
    NewSubscribers = [ClientPid | Subscribers],
    Subscriptions = maps:put(Topic, NewSubscribers, State#state.subscriptions),

    %% Acknowledge subscription
    ClientPid ! {subscribed, Topic},

    {noreply, State#state{subscriptions = Subscriptions}};

%% Unsubscribe request
handle_info({unsubscribe, ClientPid, Topic}, State) ->
    %% Remove client from topic subscribers
    Subscribers = maps:get(Topic, State#state.subscriptions, []),
    NewSubscribers = lists:delete(ClientPid, Subscribers),
    Subscriptions = maps:put(Topic, NewSubscribers, State#state.subscriptions),

    %% Acknowledge unsubscription
    ClientPid ! {unsubscribed, Topic},

    {noreply, State#state{subscriptions = Subscriptions}};

%% RPC Call request
handle_info({call, FromPid, CallId, Procedure, Args}, State) ->
    %% Find registered handler for procedure
    case maps:get(Procedure, State#state.registrations, undefined) of
        undefined ->
            %% No handler registered
            FromPid ! {call_error, CallId, <<"wamp.error.no_such_procedure">>};
        HandlerPid ->
            %% Forward to handler
            HandlerPid ! {invoke, FromPid, CallId, Procedure, Args}
    end,

    {noreply, State};

%% Register procedure
handle_info({register, ClientPid, Procedure}, State) ->
    %% Register procedure handler
    Registrations = maps:put(Procedure, ClientPid, State#state.registrations),

    %% Acknowledge registration
    ClientPid ! {registered, Procedure},

    {noreply, State#state{registrations = Registrations}};

handle_info(_Info, State) ->
    {noreply, State}.

terminate(_Reason, #state{listener = Listener}) ->
    %% Close listener
    case Listener of
        undefined -> ok;
        _ -> macula_quic:close(Listener)
    end,
    ok.

%%%===================================================================
%%% Internal Functions
%%%===================================================================

%% @doc Handle a client connection.
handle_client_connection(GatewayPid, Conn, State) ->
    accept_client_stream(GatewayPid, Conn, State).

%% @doc Accept stream from client connection.
accept_client_stream(GatewayPid, Conn, State) ->
    case macula_quic:accept_stream(Conn, 5000) of
        {ok, Stream} ->
            receive_connect_message(GatewayPid, Stream, Conn, State);
        {error, Reason} ->
            io:format("Stream accept error: ~p~n", [Reason]),
            macula_quic:close(Conn)
    end.

%% @doc Receive and decode CONNECT message from client.
receive_connect_message(GatewayPid, Stream, Conn, State) ->
    case macula_quic:recv(Stream, 5000) of
        {ok, Data} ->
            decode_connect_message(GatewayPid, Stream, Conn, State, Data);
        {error, Reason} ->
            io:format("Recv error: ~p~n", [Reason]),
            macula_quic:close(Stream),
            macula_quic:close(Conn)
    end.

%% @doc Decode CONNECT message.
decode_connect_message(GatewayPid, Stream, Conn, State, Data) ->
    case macula_protocol_decoder:decode(Data) of
        {ok, {connect, ConnectMsg}} ->
            validate_client_realm(GatewayPid, Stream, Conn, State, ConnectMsg);
        {error, Reason} ->
            io:format("Decode error: ~p~n", [Reason]),
            macula_quic:close(Stream),
            macula_quic:close(Conn)
    end.

%% @doc Validate client realm and register if valid.
validate_client_realm(GatewayPid, Stream, Conn, State, ConnectMsg) ->
    RealmId = maps:get(realm_id, ConnectMsg),
    check_realm_match(GatewayPid, Stream, Conn, State, RealmId, ConnectMsg).

%% @doc Check if client realm matches gateway realm.
check_realm_match(GatewayPid, Stream, Conn, State, RealmId, ConnectMsg)
  when RealmId =:= State#state.realm ->
    %% Register client
    ClientInfo = #{
        realm => RealmId,
        node_id => maps:get(node_id, ConnectMsg),
        capabilities => maps:get(capabilities, ConnectMsg, [])
    },
    GatewayPid ! {client_connected, self(), ClientInfo},
    %% Handle client messages
    client_message_loop(GatewayPid, Stream, Conn);
check_realm_match(_GatewayPid, Stream, Conn, _State, _OtherRealm, _ConnectMsg) ->
    %% Wrong realm, reject
    io:format("Client rejected: wrong realm~n"),
    macula_quic:close(Stream),
    macula_quic:close(Conn).

%% @doc Message loop for connected client.
client_message_loop(GatewayPid, Stream, Conn) ->
    receive
        %% Event from gateway (subscribed topic)
        {event, Topic, Payload} ->
            %% Send event to client
            EventMsg = #{
                topic => Topic,
                payload => Payload
            },
            case macula_protocol_encoder:encode(publish, EventMsg) of
                {ok, Data} ->
                    macula_quic:send(Stream, Data);
                {error, Reason} ->
                    io:format("Encode error: ~p~n", [Reason])
            end,
            client_message_loop(GatewayPid, Stream, Conn);

        %% Subscription confirmed
        {subscribed, Topic} ->
            io:format("Subscription confirmed: ~s~n", [Topic]),
            client_message_loop(GatewayPid, Stream, Conn);

        %% Unsubscription confirmed
        {unsubscribed, Topic} ->
            io:format("Unsubscription confirmed: ~s~n", [Topic]),
            client_message_loop(GatewayPid, Stream, Conn);

        %% RPC invocation (this client is the handler)
        {invoke, CallerPid, CallId, Procedure, Args} ->
            %% Send invocation to client
            InvokeMsg = #{
                call_id => CallId,
                procedure => Procedure,
                args => Args
            },
            case macula_protocol_encoder:encode(call, InvokeMsg) of
                {ok, Data} ->
                    macula_quic:send(Stream, Data),
                    %% Wait for result from client
                    %% (Simplified - should track pending invocations)
                    CallerPid ! {call_result, CallId, <<"dummy result">>};
                {error, Reason} ->
                    io:format("Encode error: ~p~n", [Reason]),
                    CallerPid ! {call_error, CallId, <<"encoding_failed">>}
            end,
            client_message_loop(GatewayPid, Stream, Conn);

        %% Call result
        {call_result, _CallId, Result} ->
            %% Forward result (simplified)
            io:format("Call result: ~p~n", [Result]),
            client_message_loop(GatewayPid, Stream, Conn);

        %% Call error
        {call_error, _CallId, Error} ->
            %% Forward error
            io:format("Call error: ~p~n", [Error]),
            client_message_loop(GatewayPid, Stream, Conn)

    after 100 ->
        %% Check for incoming messages from client
        case macula_quic:recv(Stream, 100) of
            {ok, Data} ->
                handle_client_message(GatewayPid, Data),
                client_message_loop(GatewayPid, Stream, Conn);

            {error, timeout} ->
                client_message_loop(GatewayPid, Stream, Conn);

            {error, Reason} ->
                io:format("Client recv error: ~p~n", [Reason]),
                macula_quic:close(Stream),
                macula_quic:close(Conn)
        end
    end.

%% @doc Handle message from client.
handle_client_message(GatewayPid, Data) ->
    case macula_protocol_decoder:decode(Data) of
        {ok, {publish, PubMsg}} ->
            Topic = maps:get(topic, PubMsg),
            Payload = maps:get(payload, PubMsg),
            GatewayPid ! {publish, self(), Topic, Payload};

        {ok, {subscribe, SubMsg}} ->
            Topics = maps:get(topics, SubMsg, []),
            lists:foreach(fun(Topic) ->
                GatewayPid ! {subscribe, self(), Topic}
            end, Topics);

        {ok, {unsubscribe, UnsubMsg}} ->
            Topics = maps:get(topics, UnsubMsg, []),
            lists:foreach(fun(Topic) ->
                GatewayPid ! {unsubscribe, self(), Topic}
            end, Topics);

        {ok, {call, CallMsg}} ->
            CallId = maps:get(call_id, CallMsg),
            Procedure = maps:get(procedure, CallMsg),
            Args = maps:get(args, CallMsg),
            GatewayPid ! {call, self(), CallId, Procedure, Args};

        {error, Reason} ->
            io:format("Decode error: ~p~n", [Reason])
    end.