Packages
macula
0.10.1
7.1.0
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_system/macula_gateway_dht.erl
%%%-------------------------------------------------------------------
%%% @doc
%%% DHT Query Handler Module - handles DHT message forwarding to routing server.
%%%
%%% Responsibilities:
%%% - Forward DHT STORE messages to routing server
%%% - Forward DHT FIND_VALUE messages to routing server, send encoded replies
%%% - Forward DHT FIND_NODE messages to routing server, send encoded replies
%%% - Handle DHT queries from process messages
%%% - Encode replies using protocol encoder
%%% - Handle errors gracefully
%%%
%%% Pattern: Stateless delegation module
%%% - No GenServer (no state to manage)
%%% - Pure functions forwarding to routing server
%%% - Consistent error handling ({ok, Result} | {error, Reason})
%%%
%%% Extracted from macula_gateway.erl (Phase 10)
%%% @end
%%%-------------------------------------------------------------------
-module(macula_gateway_dht).
%% API
-export([
handle_store/2,
handle_find_value/2,
handle_find_node/2,
handle_query/3,
lookup_value/1,
forward_publish_to_bootstrap/1,
send_to_peer/3,
query_peer/3
]).
%%%===================================================================
%%% API
%%%===================================================================
%% @doc Handle DHT STORE message.
%% Forwards to routing server (no reply sent back).
%% Crashes on routing server failures - this exposes DHT issues immediately.
-spec handle_store(pid(), map()) -> ok.
handle_store(_Stream, StoreMsg) ->
%% Forward to routing server (let it crash on errors)
io:format("[DHT] Processing STORE message, forwarding to routing_server~n"),
Reply = macula_routing_server:handle_message(macula_routing_server, StoreMsg),
io:format("[DHT] STORE reply from routing_server: ~p~n", [Reply]),
%% STORE doesn't require a response to be sent back
ok.
%% @doc Handle DHT FIND_VALUE message.
%% Forwards to routing server and sends encoded reply over stream.
%% Crashes on routing server or encoding failures - exposes DHT/protocol bugs.
-spec handle_find_value(pid(), map()) -> ok.
handle_find_value(Stream, FindValueMsg) ->
%% Forward to routing server (let it crash on errors)
Reply = macula_routing_server:handle_message(macula_routing_server, FindValueMsg),
%% Send reply back over stream (let it crash on errors)
ReplyBinary = macula_protocol_encoder:encode(find_value_reply, Reply),
macula_quic:send(Stream, ReplyBinary),
ok.
%% @doc Handle DHT FIND_NODE message.
%% Forwards to routing server and sends encoded reply over stream.
%% Crashes on routing server or encoding failures - exposes DHT/protocol bugs.
-spec handle_find_node(pid(), map()) -> ok.
handle_find_node(Stream, FindNodeMsg) ->
%% Forward to routing server (let it crash on errors)
Reply = macula_routing_server:handle_message(macula_routing_server, FindNodeMsg),
%% Send reply back over stream (let it crash on errors)
ReplyBinary = macula_protocol_encoder:encode(find_node_reply, Reply),
macula_quic:send(Stream, ReplyBinary),
ok.
%% @doc Handle DHT query from process message.
%% Decodes query, forwards to routing server, encodes reply, sends to requesting process.
%% Crashes on decode or routing failures - exposes protocol/DHT bugs.
-spec handle_query(pid(), atom(), binary()) -> ok.
handle_query(FromPid, _QueryType, QueryData) ->
%% Decode the query message (let it crash on decode errors)
{ok, {MessageType, Message}} = macula_protocol_decoder:decode(QueryData),
%% Forward to DHT routing server (let it crash on errors)
Reply = macula_routing_server:handle_message(macula_routing_server, Message),
%% Encode reply based on message type (let it crash on encoding errors)
ReplyData = encode_reply_by_type(MessageType, Reply),
%% Send reply back to requesting process
FromPid ! {dht_reply, ReplyData},
ok.
%% @doc Look up a value from the DHT by key.
%% Synchronous lookup from local DHT storage.
%% NOTE: For pubsub, use forward_publish_to_bootstrap/2 instead of looking up
%% subscribers locally - the bootstrap has the complete subscriber list.
%% Returns list of subscribers for the given key.
-spec lookup_value(binary()) -> {ok, list()} | {error, not_found}.
lookup_value(Key) ->
io:format("[DHT] lookup_value called with key hash: ~p~n", [Key]),
case whereis(macula_routing_server) of
undefined ->
io:format("[DHT] ERROR: routing_server not found!~n"),
{error, not_found};
RoutingServerPid ->
%% K=20 is the standard Kademlia replication factor
io:format("[DHT] Calling routing_server:find_value...~n"),
Result = macula_routing_server:find_value(RoutingServerPid, Key, 20),
io:format("[DHT] find_value result: ~p~n", [Result]),
case Result of
{ok, []} ->
{error, not_found};
{ok, Value} when is_list(Value) ->
{ok, Value};
{ok, Value} ->
%% Single value, wrap in list
{ok, [Value]};
{error, Reason} ->
{error, Reason}
end
end.
%% @doc Forward a PUBLISH message to the bootstrap gateway for distribution.
%% The bootstrap has all DHT subscriptions and can distribute to all subscribers.
%% This avoids the problem of local DHT not having remote peer subscriptions.
-spec forward_publish_to_bootstrap(map()) -> ok | {error, term()}.
forward_publish_to_bootstrap(PubMsg) ->
%% Find the connection PID (connects to bootstrap)
Realm = application:get_env(macula, realm, <<"default">>),
case gproc:lookup_local_name({connection, Realm}) of
undefined ->
{error, no_connection};
ConnPid ->
case macula_connection:send_message(ConnPid, publish, PubMsg) of
ok ->
ok;
{error, Reason} ->
{error, Reason}
end
end.
%%%===================================================================
%%% Internal functions
%%%===================================================================
%% @doc Send DHT message to remote peer (fire-and-forget).
%% Used for STORE operations that don't need a response.
-spec send_to_peer(map(), atom(), map()) -> ok | {error, term()}.
send_to_peer(NodeInfo, MessageType, Message) ->
%% Extract endpoint - either from endpoint field or construct from address
Endpoint = case maps:get(endpoint, NodeInfo, undefined) of
undefined ->
%% No endpoint, try to construct from address tuple
case maps:get(address, NodeInfo, undefined) of
undefined ->
undefined;
{Host, Port} when is_integer(Port) ->
%% Construct "host:port" string from address tuple
%% Note: peer_connector expects "host:port" format, NOT URL
HostBin = case Host of
{_, _, _, _} -> list_to_binary(inet:ntoa(Host)); % IPv4
{_, _, _, _, _, _, _, _} -> list_to_binary(inet:ntoa(Host)); % IPv6
_ when is_list(Host) -> list_to_binary(Host);
_ when is_binary(Host) -> Host
end,
PortBin = integer_to_binary(Port),
<<HostBin/binary, ":", PortBin/binary>>;
HostPortStr when is_binary(HostPortStr) ->
%% Address is already a "host:port" string
HostPortStr;
HostPortStr when is_list(HostPortStr) ->
%% Address is a "host:port" string (as list), convert to binary
list_to_binary(HostPortStr);
_Other ->
undefined
end;
Ep -> Ep
end,
case Endpoint of
undefined ->
{error, no_endpoint};
_ ->
%% Send directly via peer connector (establishes QUIC connection)
macula_peer_connector:send_message(Endpoint, MessageType, Message)
end.
%% @doc Query remote peer and wait for response.
%% Used for FIND_NODE and FIND_VALUE operations.
-spec query_peer(map(), atom(), map()) -> {ok, term()} | {error, term()}.
query_peer(NodeInfo, MessageType, Message) ->
%% For now, use send_to_peer (fire-and-forget)
%% TODO(v0.9.0): Implement request/response pattern with timeout - see TODO.md
send_to_peer(NodeInfo, MessageType, Message).
%% @private
%% @doc Encode reply based on message type.
encode_reply_by_type(find_node, Reply) ->
macula_protocol_encoder:encode(find_node_reply, Reply);
encode_reply_by_type(find_value, Reply) ->
macula_protocol_encoder:encode(find_value_reply, Reply);
encode_reply_by_type(store, Reply) ->
macula_protocol_encoder:encode(store, Reply);
encode_reply_by_type(_UnknownType, _Reply) ->
macula_protocol_encoder:encode(reply, #{error => <<"Unknown DHT message type">>}).