Packages
macula
0.6.1
7.0.0
6.0.0
5.2.2
5.2.1
5.2.0
5.1.0
5.0.0
4.8.0
4.7.1
4.7.0
4.6.0
4.5.0
4.4.10
4.4.9
4.4.8
4.4.7
4.4.6
4.4.5
4.4.4
4.4.3
4.4.2
4.4.1
4.4.0
4.3.1
4.3.0
4.2.9
4.2.8
4.2.7
4.2.6
4.2.5
4.2.4
4.2.3
4.2.2
4.2.1
4.2.0
4.1.1
4.1.0
4.0.0
3.16.0
3.15.3
3.15.2
3.15.1
3.14.0
3.13.0
3.12.1
3.12.0
3.11.1
3.11.0
3.10.3
3.10.2
3.10.1
3.9.0
3.8.0
3.7.0
3.5.0
3.4.0
3.3.0
3.2.0
3.1.0
3.0.0
2.1.1
2.1.0
2.0.0
1.5.2
1.5.1
1.4.30
1.4.29
1.4.28
1.4.27
1.4.26
1.4.25
1.4.24
1.4.23
1.4.22
1.4.21
1.4.20
1.4.19
1.4.18
1.4.17
1.4.16
1.4.15
1.4.14
1.4.13
1.4.11
1.4.10
1.4.9
1.4.8
1.4.7
1.4.6
1.4.5
1.4.4
1.4.3
1.4.2
1.4.1
1.4.0
1.3.1
1.3.0
1.2.0
1.1.0
1.0.10
1.0.9
1.0.8
1.0.7
1.0.6
1.0.5
1.0.4
1.0.3
1.0.2
1.0.1
1.0.0
0.48.6
0.48.5
0.48.4
0.48.3
0.48.2
0.48.1
0.48.0
0.47.1
0.47.0
0.46.3
0.46.1
0.46.0
0.45.3
0.45.2
0.45.1
0.45.0
0.44.2
0.44.1
0.44.0
0.43.3
0.43.2
0.43.1
0.43.0
0.42.9
0.42.8
0.42.7
0.42.6
0.42.5
0.42.4
0.42.3
0.42.2
0.42.1
0.42.0
0.41.1
0.41.0
0.40.1
0.40.0
0.39.9
0.39.8
0.39.7
0.39.6
0.39.5
0.39.4
0.39.3
0.39.2
0.39.1
0.39.0
0.38.8
0.38.7
0.38.6
0.38.5
0.38.4
0.38.3
0.38.2
0.38.1
0.38.0
0.37.7
0.37.6
0.37.5
0.37.4
0.37.3
0.37.2
0.37.1
0.37.0
0.36.6
0.36.5
0.36.4
0.36.3
0.36.2
0.36.1
0.36.0
0.35.4
0.35.3
0.35.2
0.35.1
0.35.0
0.34.1
0.34.0
0.33.1
0.33.0
0.32.5
0.32.4
0.32.3
0.32.2
0.32.1
0.32.0
0.31.9
0.31.8
0.31.7
0.31.6
0.31.5
0.31.4
0.31.3
0.31.2
0.31.1
0.31.0
0.30.10
0.30.9
0.30.8
0.30.7
0.30.6
0.30.5
0.30.4
0.30.3
0.30.2
0.30.1
0.30.0
0.29.0
0.28.3
0.28.2
0.28.1
0.28.0
0.27.1
0.27.0
0.26.1
0.26.0
0.25.6
0.25.5
0.25.4
0.25.3
0.25.2
0.25.1
0.25.0
0.24.6
0.24.5
0.24.4
0.24.3
0.24.2
0.24.1
0.24.0
0.23.3
0.23.2
0.23.1
0.23.0
0.22.12
0.22.11
0.22.10
0.22.9
0.22.8
0.22.7
0.22.6
0.22.5
0.22.4
0.22.3
0.22.2
0.22.1
0.22.0
0.21.7
0.21.6
0.21.5
0.21.4
0.21.2
0.21.1
0.21.0
0.20.25
0.20.24
0.20.23
0.20.22
0.20.21
0.20.20
0.20.19
0.20.18
0.20.17
0.20.16
0.20.15
0.20.14
0.20.13
0.20.12
0.20.11
0.20.10
0.20.9
0.20.8
0.20.7
0.20.6
0.20.5
0.20.3
0.20.2
0.20.1
0.20.0
0.19.2
0.19.1
0.19.0
0.18.1
0.18.0
0.17.4
0.17.3
0.17.2
0.17.1
0.17.0
0.16.6
0.16.5
0.16.4
0.16.3
0.16.2
0.16.1
0.16.0
0.15.1
0.15.0
0.14.3
0.14.2
0.14.1
0.14.0
0.12.6
0.12.5
0.12.3
0.11.3
0.10.2
0.10.1
0.10.0
0.9.2
0.9.1
0.9.0
0.8.25
0.8.24
0.8.23
0.8.22
0.8.21
0.8.20
0.8.19
0.8.18
0.8.17
0.8.16
0.8.15
0.8.14
0.8.13
0.8.12
0.8.11
0.8.10
0.8.9
0.8.8
0.8.7
0.8.6
0.8.5
0.8.4
0.8.3
0.8.2
0.8.1
0.8.0
0.7.30
0.7.29
0.7.28
0.7.27
0.7.26
0.7.25
0.7.24
0.7.23
0.7.22
0.7.21
0.7.20
0.7.19
0.7.18
0.7.17
0.7.16
0.7.15
0.7.14
0.7.13
0.7.12
0.7.11
0.7.10
0.7.9
0.7.8
0.7.7
0.7.6
0.7.5
0.7.4
0.7.3
0.7.2
0.7.1
0.7.0
0.6.7
0.6.6
0.6.5
0.6.4
0.6.3
0.6.2
0.6.1
0.6.0
0.5.0
0.4.4
0.4.3
0.4.2
0.4.1
0.4.0
0.3.4
0.3.3
0.3.2
0.3.1
Macula HTTP/3 Mesh SDK — connect, subscribe, publish, call, advertise
Current section
Files
Jump to
Current section
Files
src/macula_gateway.erl
%%%-------------------------------------------------------------------
%%% @doc
%%% Macula Gateway - HTTP/3 Message Router & Orchestrator
%%%
%%% Main API module and coordinator for the Macula Gateway.
%%% The gateway can be embedded in applications or run standalone.
%%%
%%% Architecture (Modular Design - Refactored Jan 2025):
%%% ========================================================
%%%
%%% Gateway (this module):
%%% - QUIC Listener Management
%%% - Message Decoding & Routing
%%% - Supervisor Coordination
%%% - API Facade
%%%
%%% Child Modules (managed via macula_gateway_sup):
%%% - macula_gateway_client_manager: Client lifecycle management
%%% - macula_gateway_pubsub: Pub/Sub message routing with wildcards
%%% - macula_gateway_rpc: RPC handler registration & invocation
%%% - macula_gateway_mesh: Mesh connection pooling
%%%
%%% Stateless Delegation Modules:
%%% - macula_gateway_dht: DHT query forwarding to routing server
%%% - macula_gateway_rpc_router: Multi-hop RPC routing via DHT
%%%
%%% Single Responsibility Principle:
%%% Each module has one clear purpose and delegates to specialized
%%% child modules. Gateway acts as orchestrator, not implementer.
%%%
%%% Usage (Embedded):
%%% ```
%%% {ok, Pid} = macula_gateway:start_link([
%%% {port, 9443},
%%% {realm, <<"com.example.realm">>}
%%% ]).
%%%
%%% %% Register RPC handler
%%% macula_gateway:register_handler(<<"add">>, fun(#{a := A, b := B}) ->
%%% #{result => A + B}
%%% end).
%%% '''
%%%
%%% Usage (Standalone):
%%% ```
%%% application:start(macula_gateway).
%%% '''
%%%
%%% @end
%%%-------------------------------------------------------------------
-module(macula_gateway).
-behaviour(gen_server).
-include_lib("kernel/include/logger.hrl").
-include("macula_config.hrl").
%% API
-export([
start_link/0,
start_link/1,
stop/1,
get_stats/1,
register_handler/2,
unregister_handler/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(),
node_id :: binary(), % 32-byte local node ID
listener :: pid() | undefined,
supervisor :: pid() | undefined, % Supervisor PID
client_manager :: pid() | undefined, % Client manager child PID
pubsub :: pid() | undefined, % Pub/Sub child PID
rpc :: pid() | undefined, % RPC child PID
mesh :: pid() | undefined, % Mesh connection manager child PID
client_streams :: #{binary() => pid()} % node_id => stream for bidirectional communication
}).
%%%===================================================================
%%% 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({local, macula_gateway}, ?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).
%% @doc Register a handler for a procedure.
-spec register_handler(binary(), fun()) -> ok | {error, term()}.
register_handler(Procedure, Handler) ->
case whereis(macula_gateway) of
undefined ->
{error, no_gateway};
Pid ->
gen_server:call(Pid, {register_handler, Procedure, Handler})
end.
%% @doc Unregister a handler for a procedure.
-spec unregister_handler(binary()) -> ok.
unregister_handler(Procedure) ->
case whereis(macula_gateway) of
undefined ->
ok;
Pid ->
gen_server:call(Pid, {unregister_handler, Procedure})
end.
%%%===================================================================
%%% 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 Get node ID from NODE_NAME environment variable or generate from realm/port.
%% Uses idiomatic Erlang pattern matching on function heads.
-spec get_node_id(binary(), inet:port_number()) -> binary().
get_node_id(Realm, Port) ->
case os:getenv("NODE_NAME") of
false ->
%% No NODE_NAME set, generate hash for backward compatibility
io:format("[Gateway] No NODE_NAME env var, generating node ID from realm/port~n"),
crypto:hash(sha256, term_to_binary({Realm, Port}));
NodeName when is_list(NodeName) ->
%% Use NODE_NAME from environment (converted to binary)
NodeNameBin = list_to_binary(NodeName),
io:format("[Gateway] Using NODE_NAME from environment: ~s~n", [NodeName]),
NodeNameBin
end.
%% @private
%% @doc Start the QUIC listener with given certificates.
start_quic_listener(Port, Realm, CertFile, KeyFile) ->
%% Start QUIC listener using simple quicer API
ListenOpts = [
{cert, CertFile},
{key, KeyFile},
{alpn, ["macula"]},
{peer_unidi_stream_count, 3},
{peer_bidi_stream_count, 100} % Allow clients to create bidirectional streams
],
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 (if running)
notify_health_server_ready(),
%% Register diagnostics procedures (if running)
register_diagnostics_procedures(self()),
%% Start async accept to receive connection events
case quicer:async_accept(Listener, #{}) of
{ok, Listener} ->
io:format("[Gateway] Async accept registered, ready for connections~n"),
ok;
{error, AcceptErr} ->
io:format("[Gateway] WARNING: async_accept failed: ~p~n", [AcceptErr]),
ok % Continue anyway, maybe we can still work
end,
%% Start routing server for DHT operations
%% Use NODE_NAME from environment if available, otherwise generate hash
LocalNodeId = get_node_id(Realm, Port),
io:format("[Gateway] Using node ID: ~p (binary: ~p)~n",
[binary:encode_hex(LocalNodeId), LocalNodeId]),
RoutingConfig = #{
k => 20, % Kademlia k-bucket size
alpha => 3 % Kademlia concurrency parameter
},
case macula_routing_server:start_link(LocalNodeId, RoutingConfig) of
{ok, _RoutingPid} ->
io:format("[Gateway] DHT routing server started with node ID: ~p~n",
[binary:encode_hex(LocalNodeId)]),
ok;
{error, {already_started, _}} ->
io:format("[Gateway] DHT routing server already running~n"),
ok;
{error, RoutingErr} ->
io:format("[Gateway] WARNING: Failed to start routing server: ~p~n", [RoutingErr]),
ok % Continue without routing server
end,
%% Start supervisor with configuration
Config = #{
port => Port,
realm => Realm,
node_id => LocalNodeId
},
case macula_gateway_sup:start_link(Config) of
{ok, SupPid} ->
io:format("[Gateway] Supervisor started: ~p~n", [SupPid]),
%% Get child PIDs from supervisor
{ok, ClientMgrPid} = macula_gateway_sup:get_client_manager(SupPid),
{ok, PubSubPid} = macula_gateway_sup:get_pubsub(SupPid),
{ok, RpcPid} = macula_gateway_sup:get_rpc(SupPid),
{ok, MeshPid} = macula_gateway_sup:get_mesh(SupPid),
io:format("[Gateway] Child modules started:~n"),
io:format("[Gateway] - Client Manager: ~p~n", [ClientMgrPid]),
io:format("[Gateway] - Pub/Sub: ~p~n", [PubSubPid]),
io:format("[Gateway] - RPC: ~p~n", [RpcPid]),
io:format("[Gateway] - Mesh: ~p~n", [MeshPid]),
State = #state{
port = Port,
realm = Realm,
node_id = LocalNodeId,
listener = Listener,
supervisor = SupPid,
client_manager = ClientMgrPid,
pubsub = PubSubPid,
rpc = RpcPid,
mesh = MeshPid,
client_streams = #{}
},
{ok, State};
{error, SupErr} ->
io:format("[Gateway] Failed to start supervisor: ~p~n", [SupErr]),
{stop, {supervisor_failed, SupErr}}
end;
{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) ->
ClientMgr = State#state.client_manager,
Rpc = State#state.rpc,
%% Query child modules for their stats
{ok, AllClients} = macula_gateway_client_manager:get_all_clients(ClientMgr),
{ok, AllHandlers} = macula_gateway_rpc:list_handlers(Rpc),
Stats = #{
port => State#state.port,
realm => State#state.realm,
clients => length(AllClients),
registrations => length(AllHandlers)
},
{reply, Stats, State};
handle_call({register_handler, Procedure, Handler}, _From, State) ->
io:format("[Gateway] Registering handler for procedure: ~s (delegating to rpc)~n", [Procedure]),
Rpc = State#state.rpc,
ok = macula_gateway_rpc:register_handler(Rpc, Procedure, Handler),
{reply, ok, State};
handle_call({unregister_handler, Procedure}, _From, State) ->
io:format("[Gateway] Unregistering handler for procedure: ~s (delegating to rpc)~n", [Procedure]),
Rpc = State#state.rpc,
ok = macula_gateway_rpc:unregister_handler(Rpc, Procedure),
{reply, ok, State};
handle_call(_Request, _From, State) ->
{reply, {error, unknown_request}, State}.
%% @doc Handle rpc_route message forwarded from connection
handle_cast({process_rpc_route, RpcRouteMsg}, State) ->
io:format("[Gateway] Processing rpc_route message from connection~n"),
%% Extract the CALL payload from rpc_route envelope
#{<<"payload">> := CallMsg} = RpcRouteMsg,
%% Call handle_rpc_call_routed with Stream=undefined, CallMsg, RpcRouteMsg, State
handle_rpc_call_routed(undefined, CallMsg, RpcRouteMsg, State);
handle_cast(_Request, State) ->
{noreply, State}.
%% Handle new stream created by peer (quicer message)
handle_info({quic, new_stream, Stream, StreamProps}, State) ->
io:format("[Gateway] ========================================~n"),
io:format("[Gateway] NEW STREAM RECEIVED!~n"),
io:format("[Gateway] Stream: ~p~n", [Stream]),
io:format("[Gateway] StreamProps: ~p~n", [StreamProps]),
io:format("[Gateway] ========================================~n"),
%% Set stream to active mode to receive data automatically
case quicer:setopt(Stream, active, true) of
ok ->
io:format("[Gateway] Stream set to active mode~n"),
{noreply, State};
{error, Reason} ->
io:format("[Gateway] Failed to set stream active: ~p~n", [Reason]),
{noreply, State}
end;
%% Handle data from QUIC stream (active mode)
handle_info({quic, Data, Stream, _Flags}, State) when is_binary(Data) ->
io:format("[Gateway] ===== Received ~p bytes from stream ~p =====~n", [byte_size(Data), Stream]),
io:format("[Gateway] Raw data (first 100 bytes): ~p~n", [binary:part(Data, 0, min(100, byte_size(Data)))]),
DecodeResult = macula_protocol_decoder:decode(Data),
handle_decoded_message(DecodeResult, Stream, State);
%% Client connected - delegate to client_manager
handle_info({client_connected, ClientPid, ClientInfo}, State) ->
io:format("[Gateway] Client connected: ~p (delegating to client_manager)~n", [ClientInfo]),
ClientMgr = State#state.client_manager,
ok = macula_gateway_client_manager:client_connected(ClientMgr, ClientPid, ClientInfo),
{noreply, State};
%% Client disconnected - delegate cleanup to all child modules
handle_info({'DOWN', _Ref, process, ClientPid, _Reason}, State) ->
io:format("[Gateway] Client disconnected: ~p (delegating cleanup)~n", [ClientPid]),
ClientMgr = State#state.client_manager,
PubSub = State#state.pubsub,
Rpc = State#state.rpc,
%% Client manager handles client removal
macula_gateway_client_manager:client_disconnected(ClientMgr, ClientPid),
%% Clean up subscriptions in pubsub
case macula_gateway_pubsub:get_stream_topics(PubSub, ClientPid) of
{ok, Topics} ->
lists:foreach(fun(Topic) ->
macula_gateway_pubsub:unsubscribe(PubSub, ClientPid, Topic)
end, Topics);
not_found ->
ok
end,
%% Clean up RPC registrations
{ok, Handlers} = macula_gateway_rpc:list_handlers(Rpc),
lists:foreach(fun
({Proc, HandlerPid}) when HandlerPid =:= ClientPid ->
macula_gateway_rpc:unregister_handler(Rpc, Proc);
(_) ->
ok
end, Handlers),
{noreply, State};
%% QUIC control event: peer_needs_streams
handle_info({quic, peer_needs_streams, _Conn, _StreamType}, State) ->
%% Peer is signaling it wants to open more streams (bidi_streams or uni_streams)
%% This is normal - just acknowledge
{noreply, State};
%% QUIC event: new_conn (connection established)
handle_info({quic, new_conn, Conn, ConnInfo}, State) ->
io:format("[Gateway] ========================================~n"),
io:format("[Gateway] NEW CONNECTION RECEIVED!~n"),
io:format("[Gateway] Connection: ~p~n", [Conn]),
io:format("[Gateway] Connection Info: ~p~n", [ConnInfo]),
io:format("[Gateway] ========================================~n"),
%% Complete TLS handshake to accept the connection
case quicer:handshake(Conn) of
ok ->
io:format("[Gateway] Connection handshake completed successfully~n"),
%% Start accepting streams on this connection
case quicer:async_accept_stream(Conn, #{}) of
{ok, Conn} ->
io:format("[Gateway] Ready to accept streams on connection~n"),
ok;
{error, StreamAcceptErr} ->
io:format("[Gateway] WARNING: async_accept_stream failed: ~p~n", [StreamAcceptErr]),
ok
end;
{ok, _} ->
io:format("[Gateway] Connection handshake completed successfully~n"),
%% Start accepting streams on this connection
case quicer:async_accept_stream(Conn, #{}) of
{ok, Conn} ->
io:format("[Gateway] Ready to accept streams on connection~n"),
ok;
{error, StreamAcceptErr} ->
io:format("[Gateway] WARNING: async_accept_stream failed: ~p~n", [StreamAcceptErr]),
ok
end;
{error, Reason} ->
io:format("[Gateway] Handshake failed: ~p~n", [Reason]),
ok
end,
%% Register for next connection
case quicer:async_accept(State#state.listener, #{}) of
{ok, _} ->
io:format("[Gateway] Ready for next connection~n"),
ok;
{error, AcceptErr} ->
io:format("[Gateway] WARNING: async_accept failed: ~p~n", [AcceptErr]),
ok
end,
{noreply, State};
%% QUIC event: shutdown (connection shutting down)
handle_info({quic, shutdown, Conn, Reason}, State) ->
io:format("[Gateway] QUIC shutdown: Conn=~p, Reason=~p~n", [Conn, Reason]),
{noreply, State};
%% QUIC event: transport_shutdown (transport layer shutting down)
handle_info({quic, transport_shutdown, Conn, Reason}, State) ->
io:format("[Gateway] QUIC transport_shutdown: Conn=~p, Reason=~p~n", [Conn, Reason]),
{noreply, State};
%% DHT query (find_node, find_value, store)
handle_info({dht_query, FromPid, QueryType, QueryData}, State) ->
%% Delegate to DHT module
_Result = macula_gateway_dht:handle_query(FromPid, QueryType, QueryData),
{noreply, State};
handle_info(Info, State) ->
io:format("[Gateway] WARNING: Unhandled handle_info message: ~p~n", [Info]),
{noreply, State}.
terminate(_Reason, #state{listener = Listener, supervisor = SupPid}) ->
%% Close listener
case Listener of
undefined -> ok;
_ -> macula_quic:close(Listener)
end,
%% Stop supervisor (will stop all children: client_manager, pubsub, rpc)
case SupPid of
undefined -> ok;
_ ->
erlang:unlink(SupPid),
exit(SupPid, shutdown)
end,
ok.
%%%===================================================================
%%% Internal Functions
%%%===================================================================
%% @doc Handle CONNECT message from client.
handle_connect(Stream, ConnectMsg, #state{realm = Realm} = State) ->
RealmId = maps:get(<<"realm_id">>, ConnectMsg),
handle_connect_realm(RealmId =:= Realm, Stream, ConnectMsg, State).
%% @doc Process CONNECT with valid realm (pattern matching on boolean).
handle_connect_realm(true, Stream, ConnectMsg, State) ->
RealmId = maps:get(<<"realm_id">>, ConnectMsg),
NodeId = maps:get(<<"node_id">>, ConnectMsg),
%% Extract and parse endpoint from CONNECT message for peer-to-peer connections
Endpoint = maps:get(<<"endpoint">>, ConnectMsg, undefined),
Address = parse_endpoint(Endpoint),
ClientInfo = #{
realm => RealmId,
node_id => NodeId,
capabilities => maps:get(<<"capabilities">>, ConnectMsg, []),
endpoint => Endpoint,
address => Address
},
io:format("[Gateway] Client connected: ~p~n", [ClientInfo]),
%% HTTP/3 streams are bidirectional - we can send messages back on the same stream
%% Store the client's incoming stream for sending replies (enables client-only mode)
%% Also store endpoint address for creating mesh connections when needed
ClientStreams = State#state.client_streams,
NewClientStreams = ClientStreams#{NodeId => Stream},
io:format("[Gateway] Stored client stream for node ~p (bidirectional communication)~n",
[binary:encode_hex(NodeId)]),
%% Connection management now handled by macula_gateway_mesh module
NewState = State#state{client_streams = NewClientStreams},
%% Add peer to DHT routing table with real address
NodeInfo = #{
node_id => NodeId,
address => Address % Use real parsed address instead of placeholder
},
case whereis(macula_routing_server) of
undefined ->
io:format("[Gateway] WARNING: Routing server not running, cannot add peer~n");
RoutingServerPid ->
io:format("[Gateway] Adding peer to routing table: ~p~n", [NodeId]),
macula_routing_server:add_node(RoutingServerPid, NodeInfo),
io:format("[Gateway] Peer added to routing table~n")
end,
%% Send PONG acknowledgment back to keep stream alive for bidirectional communication
PongMsg = #{
timestamp => erlang:system_time(millisecond),
server_time => erlang:system_time(millisecond)
},
PongBinary = macula_protocol_encoder:encode(pong, PongMsg),
case macula_quic:send(Stream, PongBinary) of
ok ->
io:format("[Gateway] Sent PONG acknowledgment to client ~p~n",
[binary:encode_hex(NodeId)]);
{error, PongErr} ->
io:format("[Gateway] WARNING: Failed to send PONG: ~p~n", [PongErr])
end,
{noreply, NewState};
handle_connect_realm(false, Stream, ConnectMsg, State) ->
RealmId = maps:get(<<"realm_id">>, ConnectMsg),
io:format("[Gateway] Realm mismatch: ~p != ~p~n", [RealmId, State#state.realm]),
macula_quic:close(Stream),
{noreply, State}.
%% @doc Handle DHT STORE message.
handle_dht_store(Stream, StoreMsg, State) ->
%% Delegate to DHT module
_Result = macula_gateway_dht:handle_store(Stream, StoreMsg),
{noreply, State}.
%% @doc Handle DHT FIND_VALUE message.
handle_dht_find_value(Stream, FindValueMsg, State) ->
%% Delegate to DHT module
_Result = macula_gateway_dht:handle_find_value(Stream, FindValueMsg),
{noreply, State}.
%% @doc Handle DHT FIND_NODE message.
handle_dht_find_node(Stream, FindNodeMsg, State) ->
%% Delegate to DHT module
_Result = macula_gateway_dht:handle_find_node(Stream, FindNodeMsg),
{noreply, State}.
%% @doc Handle RPC call message received via DHT routing.
%% Processes the call locally and sends REPLY back via DHT routing.
handle_rpc_call_routed(_Stream, CallMsg, RpcRouteMsg, State) ->
%% Delegate to RPC router module
_Result = macula_gateway_rpc_router:handle_routed_call(
CallMsg, RpcRouteMsg, State#state.node_id, State#state.rpc, State#state.mesh
),
{noreply, State}.
%% @doc Handle routed REPLY message delivered locally.
%% Forward the REPLY to the local connection process.
handle_rpc_reply_routed(ReplyMsg, RpcRouteMsg, State) ->
%% Delegate to RPC router module
_Result = macula_gateway_rpc_router:handle_routed_reply(
ReplyMsg, RpcRouteMsg, State#state.node_id, State#state.client_streams
),
{noreply, State}.
%% @doc Handle RPC call message (legacy direct handling).
handle_rpc_call(Stream, CallMsg, State) ->
io:format("[Gateway] Processing RPC CALL message: ~p~n", [CallMsg]),
%% Extract call data (CallMsg has binary keys from msgpack)
Procedure = maps:get(<<"procedure">>, CallMsg),
CallId = maps:get(<<"call_id">>, CallMsg),
ArgsJson = maps:get(<<"args">>, CallMsg),
io:format("[Gateway] Procedure: ~s, CallId: ~p~n", [Procedure, CallId]),
%% Look up handler (delegate to rpc module)
Rpc = State#state.rpc,
case macula_gateway_rpc:get_handler(Rpc, Procedure) of
not_found ->
%% No handler registered
io:format("[Gateway] No handler found for procedure: ~s~n", [Procedure]),
ErrorReply = #{
call_id => CallId,
error => #{
code => <<"no_such_procedure">>,
message => <<"No handler registered for ", Procedure/binary>>
}
},
ReplyBinary = macula_protocol_encoder:encode(reply, ErrorReply),
macula_quic:send(Stream, ReplyBinary),
{noreply, State};
{ok, Handler} ->
%% Decode args from JSON
try
Args = json:decode(ArgsJson),
io:format("[Gateway] Decoded args: ~p~n", [Args]),
%% Invoke handler
io:format("[Gateway] Invoking handler~n"),
Result = Handler(Args),
io:format("[Gateway] Handler result: ~p~n", [Result]),
%% Unwrap result tuple if needed
ResultMap = unwrap_result_to_map(Result),
%% Send reply (don't close stream - let connection idle timeout handle cleanup)
Reply = #{
call_id => CallId,
result => encode_json(ResultMap)
},
SuccessReplyBinary = macula_protocol_encoder:encode(reply, Reply),
SendResult = macula_quic:send(Stream, SuccessReplyBinary),
io:format("[Gateway] Sent reply (result: ~p, stream: ~p, size: ~p bytes)~n",
[SendResult, Stream, byte_size(SuccessReplyBinary)]),
{noreply, State}
catch
Class:Reason:Stacktrace ->
io:format("[Gateway] Handler error: ~p:~p~n~p~n", [Class, Reason, Stacktrace]),
ErrorReply = #{
call_id => CallId,
error => #{
code => <<"handler_error">>,
message => iolist_to_binary(io_lib:format("~p:~p", [Class, Reason]))
}
},
ErrorReplyBinary = macula_protocol_encoder:encode(reply, ErrorReply),
macula_quic:send(Stream, ErrorReplyBinary),
{noreply, State}
end
end.
%% @doc Encode map/list to JSON binary.
-spec encode_json(map() | list()) -> binary().
encode_json(Data) ->
macula_utils:encode_json(Data).
%%%===================================================================
%%% Mesh Connection Management
%%%===================================================================
%% Get or create a QUIC connection to a peer node for message forwarding.
%% Opens a NEW stream for each message (QUIC best practice).
%% Handle successful connection acceptance.
%% Register for incoming streams with active mode enabled.
%% Manual accept functions removed - quicer_server handles connections automatically
%% @doc Handle decoded protocol messages.
%% Dispatches to appropriate handlers based on message type.
handle_decoded_message({ok, {connect, ConnectMsg}}, Stream, State) ->
io:format("[Gateway] Decoded CONNECT message~n"),
handle_connect(Stream, ConnectMsg, State);
%% Handle decoded STORE message.
handle_decoded_message({ok, {store, StoreMsg}}, Stream, State) ->
io:format("[Gateway] *** RECEIVED STORE MESSAGE ***~n"),
io:format("[Gateway] STORE message: ~p~n", [StoreMsg]),
handle_dht_store(Stream, StoreMsg, State);
%% Handle decoded FIND_VALUE message.
handle_decoded_message({ok, {find_value, FindValueMsg}}, Stream, State) ->
io:format("[Gateway] *** RECEIVED FIND_VALUE MESSAGE ***~n"),
io:format("[Gateway] FIND_VALUE message: ~p~n", [FindValueMsg]),
handle_dht_find_value(Stream, FindValueMsg, State);
%% Handle decoded FIND_NODE message.
handle_decoded_message({ok, {find_node, FindNodeMsg}}, Stream, State) ->
io:format("[Gateway] *** RECEIVED FIND_NODE MESSAGE ***~n"),
io:format("[Gateway] FIND_NODE message: ~p~n", [FindNodeMsg]),
handle_dht_find_node(Stream, FindNodeMsg, State);
%% Handle RPC route message (multi-hop DHT routing).
handle_decoded_message({ok, {rpc_route, RpcRouteMsg}}, Stream, State) ->
io:format("[Gateway] *** RECEIVED RPC_ROUTE MESSAGE ***~n"),
io:format("[Gateway] RPC route message: ~p~n", [RpcRouteMsg]),
LocalNodeId = State#state.node_id,
RoutingServerPid = whereis(macula_routing_server),
case macula_rpc_routing:route_or_deliver(LocalNodeId, RpcRouteMsg, RoutingServerPid) of
{deliver, <<"call">>, CallMsg} ->
%% Unwrap and process CALL locally (MessagePack returns binary, not atom)
io:format("[Gateway] RPC route: delivering CALL locally~n"),
handle_rpc_call_routed(Stream, CallMsg, RpcRouteMsg, State);
{deliver, <<"reply">>, ReplyMsg} ->
%% Routed REPLY delivered - forward to connection for matching with pending call
io:format("[Gateway] Routed REPLY delivered locally, forwarding to connection~n"),
handle_rpc_reply_routed(ReplyMsg, RpcRouteMsg, State);
{forward, NextHopNodeInfo, UpdatedRpcRouteMsg} ->
%% Forward to next hop through mesh (delegate to RPC router)
io:format("[Gateway] RPC route: forwarding to next hop~n"),
_Result = macula_gateway_rpc_router:forward_rpc_route(
NextHopNodeInfo, UpdatedRpcRouteMsg, State#state.mesh
),
{noreply, State};
{error, Reason} ->
%% Routing error (max hops, no route, etc.)
io:format("[Gateway] RPC route error: ~p~n", [Reason]),
{noreply, State}
end;
%% Handle RPC call message (legacy direct call - will be deprecated).
handle_decoded_message({ok, {call, CallMsg}}, Stream, State) ->
io:format("[Gateway] *** RECEIVED RPC CALL (DIRECT) ***~n"),
io:format("[Gateway] Call message: ~p~n", [CallMsg]),
handle_rpc_call(Stream, CallMsg, State);
%% Handle SUBSCRIBE message.
handle_decoded_message({ok, {subscribe, SubMsg}}, Stream, State) ->
io:format("[Gateway] *** RECEIVED SUBSCRIBE MESSAGE ***~n"),
io:format("[Gateway] Subscribe message: ~p~n", [SubMsg]),
handle_subscribe(Stream, SubMsg, State);
%% Handle UNSUBSCRIBE message.
handle_decoded_message({ok, {unsubscribe, UnsubMsg}}, Stream, State) ->
io:format("[Gateway] *** RECEIVED UNSUBSCRIBE MESSAGE ***~n"),
io:format("[Gateway] Unsubscribe message: ~p~n", [UnsubMsg]),
handle_unsubscribe(Stream, UnsubMsg, State);
%% Handle PUBLISH message.
handle_decoded_message({ok, {publish, PubMsg}}, Stream, State) ->
io:format("[Gateway] *** RECEIVED PUBLISH MESSAGE ***~n"),
io:format("[Gateway] Publish message: ~p~n", [PubMsg]),
handle_publish(Stream, PubMsg, State);
%% Handle other decoded message types.
handle_decoded_message({ok, {Type, Other}}, _Stream, State) ->
io:format("[Gateway] Received message type ~p: ~p~n", [Type, Other]),
{noreply, State};
%% Handle decode error.
handle_decoded_message({error, DecodeErr}, _Stream, State) ->
io:format("[Gateway] !!! DECODE ERROR: ~p !!!~n", [DecodeErr]),
{noreply, State}.
%%%===================================================================
%%% Pub/Sub Handlers
%%%===================================================================
%% @doc Handle subscription to topics - delegate to pubsub module.
handle_subscribe(Stream, SubMsg, State) ->
Topics = maps:get(<<"topics">>, SubMsg, []),
io:format("[Gateway] Stream ~p subscribing to topics: ~p (delegating to pubsub)~n", [Stream, Topics]),
PubSub = State#state.pubsub,
%% Subscribe to each topic via pubsub module
lists:foreach(fun(Topic) ->
macula_gateway_pubsub:subscribe(PubSub, Stream, Topic)
end, Topics),
{noreply, State}.
%% @doc Handle unsubscribe from topics - delegate to pubsub module.
handle_unsubscribe(Stream, UnsubMsg, State) ->
Topics = maps:get(<<"topics">>, UnsubMsg, []),
io:format("[Gateway] Stream ~p unsubscribing from topics: ~p (delegating to pubsub)~n", [Stream, Topics]),
PubSub = State#state.pubsub,
%% Unsubscribe from each topic via pubsub module
lists:foreach(fun(Topic) ->
macula_gateway_pubsub:unsubscribe(PubSub, Stream, Topic)
end, Topics),
{noreply, State}.
%% @doc Handle publish message - distribute to all topic subscribers via pubsub module.
handle_publish(_PublisherStream, PubMsg, State) ->
Topic = maps:get(<<"topic">>, PubMsg),
io:format("[Gateway] Publishing message to topic: ~s (delegating to pubsub)~n", [Topic]),
PubSub = State#state.pubsub,
%% Get matching subscribers from pubsub module
{ok, Subscribers} = macula_gateway_pubsub:get_subscribers(PubSub, Topic),
io:format("[Gateway] Found ~p subscribers for topic ~s~n", [length(Subscribers), Topic]),
%% Encode and send to each subscriber
PubBinary = macula_protocol_encoder:encode(publish, PubMsg),
lists:foreach(
fun(SubscriberStream) ->
case macula_quic:send(SubscriberStream, PubBinary) of
ok ->
io:format("[Gateway] Successfully sent to stream ~p~n", [SubscriberStream]);
{error, Reason} ->
io:format("[Gateway] Failed to send to stream ~p: ~p~n",
[SubscriberStream, Reason])
end
end,
Subscribers
),
io:format("[Gateway] Finished distributing message to ~p subscribers~n", [length(Subscribers)]),
{noreply, State}.
%%%===================================================================
%%% Internal Helper Functions
%%%===================================================================
%% @private
%% @doc Notify health server that gateway is ready (if health server is running).
notify_health_server_ready() ->
check_and_notify_health(whereis(macula_gateway_health)).
check_and_notify_health(undefined) ->
ok; % Health server not running (embedded mode)
check_and_notify_health(_Pid) ->
macula_gateway_health:set_ready(true).
%% @private
%% @doc Register diagnostics procedures (if diagnostics service is running).
register_diagnostics_procedures(GatewayPid) ->
check_and_register_diagnostics(whereis(macula_gateway_diagnostics), GatewayPid).
check_and_register_diagnostics(undefined, _GatewayPid) ->
ok; % Diagnostics service not running (embedded mode)
check_and_register_diagnostics(_Pid, GatewayPid) ->
macula_gateway_diagnostics:register_procedures(GatewayPid).
%% @doc Parse endpoint URL to address tuple.
%% Converts "https://host:port" to {{IP_tuple}, Port}
%% Crashes on parsing errors - indicates invalid endpoint configuration.
-spec parse_endpoint(undefined | binary()) -> {{byte(), byte(), byte(), byte()}, inet:port_number()}.
parse_endpoint(undefined) ->
{{0,0,0,0}, 0};
parse_endpoint(Endpoint) when is_binary(Endpoint) ->
%% Parse URL using uri_string (let it crash on invalid URLs)
case uri_string:parse(Endpoint) of
#{host := Host, port := Port} when is_integer(Port) ->
%% Resolve hostname to IP address (let it crash on resolution errors)
HostStr = binary_to_list(Host),
{ok, IPTuple} = inet:getaddr(HostStr, inet),
{IPTuple, Port};
#{host := Host} ->
%% No port specified, use default 9443
HostStr = binary_to_list(Host),
{ok, IPTuple} = inet:getaddr(HostStr, inet),
{IPTuple, 9443};
_ ->
io:format("[Gateway] Invalid endpoint URL format: ~s~n", [Endpoint]),
error({invalid_endpoint_format, Endpoint})
end.
%% @doc Unwrap result to map format.
unwrap_result_to_map({ok, Map}) when is_map(Map) -> Map;
unwrap_result_to_map(Map) when is_map(Map) -> Map;
unwrap_result_to_map(Other) -> #{value => Other}.