Packages
macula
0.8.21
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_peer_system/macula_peer_discovery.erl
%%%-------------------------------------------------------------------
%%% @doc Peer Discovery - DHT-based gateway discovery and P2P mesh formation.
%%%
%%% This module implements automatic peer discovery to enable true P2P mesh:
%%% 1. Gateways register themselves in the bootstrap DHT
%%% 2. Peers periodically query DHT to discover other gateways
%%% 3. Peers establish direct QUIC connections to discovered gateways
%%% 4. Cross-peer relay works automatically via these mesh connections
%%%
%%% Architecture:
%%% Peer1 Gateway - Peer2 Gateway - Peer3 Gateway
%%% | | |
%%% Local Clients Local Clients Local Clients
%%%
%%% DHT Storage:
%%% Key pattern: "peer.gateway." + NodeID
%%% Value: map with node_id, host, port, realm fields
%%%
%%% @end
%%%-------------------------------------------------------------------
-module(macula_peer_discovery).
-behaviour(gen_server).
%% API
-export([start_link/1, register_gateway/0, discover_peers/0]).
%% gen_server callbacks
-export([init/1, handle_call/3, handle_cast/2, handle_info/2, terminate/2]).
-record(state, {
node_id :: binary(),
host :: binary(),
port :: integer(),
realm :: binary(),
discovery_interval :: integer(), % milliseconds
timer_ref :: reference() | undefined
}).
%%==============================================================================
%% API
%%==============================================================================
%% @doc Start the peer discovery process
-spec start_link(map()) -> {ok, pid()} | {error, term()}.
start_link(Config) ->
gen_server:start_link({local, ?MODULE}, ?MODULE, Config, []).
%% @doc Register this gateway in the bootstrap DHT
-spec register_gateway() -> ok | {error, term()}.
register_gateway() ->
gen_server:call(?MODULE, register_gateway).
%% @doc Discover other peer gateways from DHT
-spec discover_peers() -> {ok, [map()]} | {error, term()}.
discover_peers() ->
gen_server:call(?MODULE, discover_peers).
%%==============================================================================
%% gen_server callbacks
%%==============================================================================
init(Config) ->
NodeID = maps:get(node_id, Config),
Host = maps:get(host, Config, <<"localhost">>),
Port = maps:get(port, Config),
Realm = maps:get(realm, Config, <<"default">>),
DiscoveryInterval = maps:get(discovery_interval, Config, 30000), % 30s default
io:format("[PeerDiscovery] Initializing for ~s:~p~n", [Host, Port]),
State = #state{
node_id = NodeID,
host = Host,
port = Port,
realm = Realm,
discovery_interval = DiscoveryInterval
},
%% Register ourselves in DHT after short delay (let bootstrap connect first)
timer:send_after(3000, self(), register_self),
%% Start periodic peer discovery
TimerRef = erlang:send_after(5000, self(), discover_and_connect),
{ok, State#state{timer_ref = TimerRef}}.
handle_call(register_gateway, _From, State) ->
Result = do_register_gateway(State),
{reply, Result, State};
handle_call(discover_peers, _From, State) ->
Result = do_discover_peers(State),
{reply, Result, State};
handle_call(_Request, _From, State) ->
{reply, {error, unknown_request}, State}.
handle_cast(_Msg, State) ->
{noreply, State}.
handle_info(register_self, State) ->
case do_register_gateway(State) of
ok ->
io:format("[PeerDiscovery] Successfully registered gateway in DHT~n");
{error, Reason} ->
io:format("[PeerDiscovery] Failed to register gateway: ~p~n", [Reason])
end,
{noreply, State};
handle_info(discover_and_connect, State) ->
#state{discovery_interval = Interval} = State,
%% Discover peers from DHT
case do_discover_peers(State) of
{ok, Peers} when length(Peers) > 0 ->
io:format("[PeerDiscovery] Discovered ~p peer(s)~n", [length(Peers)]),
lists:foreach(fun(Peer) ->
connect_to_peer(Peer, State)
end, Peers);
{ok, []} ->
ok; % No peers discovered yet
{error, Reason} ->
io:format("[PeerDiscovery] Discovery failed: ~p~n", [Reason])
end,
%% Schedule next discovery
TimerRef = erlang:send_after(Interval, self(), discover_and_connect),
{noreply, State#state{timer_ref = TimerRef}};
handle_info(_Info, State) ->
{noreply, State}.
terminate(_Reason, #state{timer_ref = TimerRef}) ->
case TimerRef of
undefined -> ok;
Ref -> erlang:cancel_timer(Ref)
end,
ok.
%%==============================================================================
%% Internal functions
%%==============================================================================
%% @private Register this gateway in the DHT
do_register_gateway(State) ->
#state{node_id = NodeID, host = Host, port = Port, realm = Realm} = State,
Key = <<"peer.gateway.", NodeID/binary>>,
Value = #{
node_id => NodeID,
host => Host,
port => Port,
realm => Realm,
registered_at => erlang:system_time(second)
},
%% Store in local routing server's DHT
case whereis(macula_routing_server) of
undefined ->
{error, routing_server_not_found};
RoutingServer ->
macula_routing_server:store_local(RoutingServer, Key, Value),
ok
end.
%% @private Discover peer gateways from DHT
do_discover_peers(State) ->
#state{node_id = MyNodeID} = State,
case whereis(macula_routing_server) of
undefined ->
{error, routing_server_not_found};
RoutingServer ->
%% Query DHT for all peer.gateway.* keys
%% Note: This is a simplified approach - in production we'd use proper DHT iteration
AllPeers = discover_all_gateway_peers(RoutingServer),
%% Filter out ourselves
OtherPeers = lists:filter(fun(#{node_id := NodeID}) ->
NodeID =/= MyNodeID
end, AllPeers),
{ok, OtherPeers}
end.
%% @private Discover all registered gateways
%% Note: This is a naive implementation - we should use DHT prefix queries
discover_all_gateway_peers(RoutingServer) ->
%% For now, we'll rely on the fact that keys are stored locally
%% In a real implementation, we'd iterate through the DHT keyspace
case macula_routing_server:get_all_keys(RoutingServer) of
{ok, Keys} ->
GatewayKeys = lists:filter(fun(Key) ->
case Key of
<<"peer.gateway.", _/binary>> -> true;
_ -> false
end
end, Keys),
lists:filtermap(fun(Key) ->
case macula_routing_server:get_local(RoutingServer, Key) of
{ok, [Value|_]} -> {true, Value}; % Extract first element from list
{ok, []} -> false; % Empty list
_ -> false
end
end, GatewayKeys);
_ ->
[]
end.
%% @private Connect to a discovered peer gateway
connect_to_peer(PeerInfo, State) ->
#{node_id := PeerNodeID, host := Host, port := Port, realm := Realm} = PeerInfo,
#state{node_id = MyNodeID} = State,
%% Don't connect to ourselves
case PeerNodeID of
MyNodeID ->
ok;
_ ->
%% Build peer URL
PeerUrl = iolist_to_binary([<<"https://">>, Host, <<":">>, integer_to_binary(Port)]),
%% Check if already connected
case is_already_connected(PeerNodeID) of
true ->
ok; % Already connected
false ->
io:format("[PeerDiscovery] Connecting to peer ~s at ~s~n",
[binary:encode_hex(PeerNodeID), PeerUrl]),
%% Start peer connection
case macula_peers_sup:start_peer(PeerUrl, #{realm => Realm}) of
{ok, _PeerPid} ->
io:format("[PeerDiscovery] Connected to peer ~s~n",
[binary:encode_hex(PeerNodeID)]);
{error, {already_started, _}} ->
ok; % Already connected
{error, Reason} ->
io:format("[PeerDiscovery] Failed to connect to ~s: ~p~n",
[binary:encode_hex(PeerNodeID), Reason])
end
end
end.
%% @private Check if already connected to a peer
is_already_connected(_PeerNodeID) ->
%% Check if we have an active peer connection to this node
case whereis(macula_peers_sup) of
undefined ->
false;
_ ->
%% Query peer supervisor for active connections
%% This is a simplified check - in production we'd have a proper registry
Children = supervisor:which_children(macula_peers_sup),
lists:any(fun({_Id, Pid, _Type, _Modules}) ->
case Pid of
undefined -> false;
_ ->
%% Check if this peer process is for our target node
%% This is simplified - we'd need to query the peer for its node_id
false % For now, allow reconnections
end
end, Children)
end.