%%%------------------------------------------------------------------- %%% @doc Macula Mesh Client — connects a node to the relay mesh via QUIC. %%% %%% Single gen_server managing ONE persistent QUIC connection to a relay %%% with automatic failover across multiple relays. On disconnect, cycles %%% to the next relay in the list with exponential backoff + jitter. %%% Replays all subscriptions and procedure registrations on reconnect. %%% %%% This is the SDK client module. For relay-to-relay peering connections, %%% see macula_relay_client in the macula-relay repository. %%% %%% Usage: %%% %%% ``` %%% {ok, Client} = macula_mesh_client:start_link(#{ %%% relays => [<<"quic://boot.macula.io:443">>], %%% realm => <<"io.macula">> %%% }). %%% ''' %%% @end %%%------------------------------------------------------------------- -module(macula_mesh_client). -behaviour(gen_server). -include_lib("kernel/include/logger.hrl"). -export([start_link/1, stop/1]). -export([subscribe/3, unsubscribe/2, publish/3]). -export([advertise/3, unadvertise/2, call/4, call/5, async_call/7]). -export([is_connected/1]). -export([init/1, handle_call/3, handle_cast/2, handle_info/2, terminate/2]). %% Test exports -ifdef(TEST). -export([backoff_ms/1, parse_url/1]). -endif. -define(RECONNECT_BASE_MS, 1000). -define(RECONNECT_MAX_MS, 30000). -define(RECONNECT_JITTER, 0.3). -define(CALL_TIMEOUT, 5000). -define(PING_INTERVAL_MS, 30000). -define(PING_TOPIC, <<"_mesh.relay.ping">>). -record(state, { relays :: [binary()], %% all configured relay URLs relay_index :: non_neg_integer(), %% current relay (0-based) url :: binary(), %% current relay URL host :: string(), port :: integer(), realm :: binary(), identity :: binary(), geo_identity :: map(), %% geo metadata for CONNECT (city, country, lat, lng) site :: map() | undefined, %% site metadata for CONNECT (site_id, name, city, lat, lng, site_type) client_type = <<"node">> :: binary(), %% always "node" for mesh clients tls_verify :: verify_peer | none, %% TLS verification mode (none for relay peering) preferred_host :: string(), %% first relay hostname (for target_relay in CONNECT) previous_host :: string() | undefined, %% host before failover (for reroute detection) conn :: reference() | undefined, stream :: reference() | undefined, status :: connecting | connected | disconnected, recv_buffer :: binary(), reconnect_attempt :: non_neg_integer(), %% for exponential backoff %% Application state (survives reconnects) subscriptions :: #{reference() => {binary(), fun()}}, %% ref => {topic, callback} procedures :: #{binary() => fun()}, %% procedure => handler pending_calls :: #{binary() => {pid(), reference()}}, %% call_id => {from, timer_ref} ping_sent_at :: integer() | undefined, %% erlang:monotonic_time(millisecond) when PING sent last_rtt_ms :: non_neg_integer() | undefined %% most recent RTT measurement }). %%==================================================================== %% API %%==================================================================== -spec start_link(map()) -> {ok, pid()} | {error, term()}. start_link(Opts) -> gen_server:start_link(?MODULE, Opts, []). -spec stop(pid()) -> ok. stop(Pid) -> gen_server:stop(Pid). -spec is_connected(pid()) -> boolean(). is_connected(Pid) -> gen_server:call(Pid, is_connected). -spec subscribe(pid(), binary(), fun((map()) -> ok)) -> {ok, reference()}. subscribe(Pid, Topic, Callback) -> gen_server:call(Pid, {subscribe, Topic, Callback}). -spec unsubscribe(pid(), reference()) -> ok. unsubscribe(Pid, Ref) -> gen_server:call(Pid, {unsubscribe, Ref}). -spec publish(pid(), binary(), binary() | map()) -> ok. publish(Pid, Topic, Payload) -> gen_server:cast(Pid, {publish, Topic, Payload}). -spec advertise(pid(), binary(), fun((map()) -> {ok, term()} | {error, term()})) -> {ok, reference()}. advertise(Pid, Procedure, Handler) -> gen_server:call(Pid, {advertise, Procedure, Handler}). -spec unadvertise(pid(), binary()) -> ok. unadvertise(Pid, Procedure) -> gen_server:call(Pid, {unadvertise, Procedure}). -spec call(pid(), binary(), map(), timeout()) -> {ok, term()} | {error, term()}. call(Pid, Procedure, Args, Timeout) -> call(Pid, Procedure, Args, Timeout, #{}). -spec call(pid(), binary(), map(), timeout(), map()) -> {ok, term()} | {error, term()}. call(Pid, Procedure, Args, Timeout, Opts) -> gen_server:call(Pid, {rpc_call, Procedure, Args, Timeout, Opts}, Timeout + 1000). %% @doc Fire-and-forget RPC call. Sends CALL to the relay and delivers the reply %% as {relay_reply, CorrelationId, Result, Trace} to CallbackPid. %% Does NOT block the caller. Used for cross-relay RPC forwarding. -spec async_call(pid(), binary(), map(), timeout(), map(), pid(), binary()) -> ok. async_call(Pid, Procedure, Args, Timeout, Opts, CallbackPid, CorrelationId) -> gen_server:cast(Pid, {async_rpc_call, Procedure, Args, Timeout, Opts, CallbackPid, CorrelationId}). %%==================================================================== %% gen_server callbacks %%==================================================================== init(Opts) -> Relays = case maps:find(relays, Opts) of {ok, List} when is_list(List), length(List) > 0 -> List; _ -> [maps:get(url, Opts, <<"https://localhost:4433">>)] end, %% First relay is the preferred target (e.g., nearest virtual relay identity). %% Remaining entries are ordered fallbacks for failover — not load balancing. Index = 0, Url = hd(Relays), {Host, Port} = parse_url(Url), Realm = maps:get(realm, Opts, <<"io.macula">>), Identity = maps:get(identity, Opts, <<"anonymous">>), GeoIdentity = maps:get(geo_identity, Opts, #{}), Site = maps:get(site, Opts, undefined), %% Mesh clients are always "node" type (relay peering uses macula_relay_client) TlsVerify = maps:get(tls_verify, Opts, verify_peer), State = #state{ relays = Relays, relay_index = Index, url = Url, host = Host, preferred_host = Host, port = Port, realm = Realm, identity = Identity, geo_identity = GeoIdentity, site = Site, %% client_type defaults to <<"node">> in record def tls_verify = TlsVerify, conn = undefined, stream = undefined, status = connecting, recv_buffer = <<>>, reconnect_attempt = 0, subscriptions = #{}, procedures = #{}, pending_calls = #{} }, self() ! connect, {ok, State}. %%==================================================================== %% Status %%==================================================================== handle_call(is_connected, _From, #state{status = connected, stream = S} = State) when S =/= undefined -> {reply, true, State}; handle_call(is_connected, _From, State) -> {reply, false, State}; %%==================================================================== %% Subscribe / Unsubscribe %%==================================================================== handle_call({subscribe, Topic, Callback}, _From, State) -> Ref = make_ref(), Subs = maps:put(Ref, {Topic, Callback}, State#state.subscriptions), %% Send SUBSCRIBE if connected maybe_send(subscribe, #{<<"topics">> => [Topic], <<"qos">> => 0}, State), {reply, {ok, Ref}, State#state{subscriptions = Subs}}; handle_call({unsubscribe, Ref}, _From, State) -> case maps:get(Ref, State#state.subscriptions, undefined) of undefined -> {reply, {error, not_found}, State}; {Topic, _Callback} -> Subs = maps:remove(Ref, State#state.subscriptions), maybe_send(unsubscribe, #{<<"topics">> => [Topic]}, State), {reply, ok, State#state{subscriptions = Subs}} end; %%==================================================================== %% Advertise / Unadvertise %%==================================================================== handle_call({advertise, Procedure, Handler}, _From, State) -> Procs = maps:put(Procedure, Handler, State#state.procedures), maybe_send(register_procedure, #{<<"procedure">> => Procedure}, State), {reply, {ok, make_ref()}, State#state{procedures = Procs}}; handle_call({unadvertise, Procedure}, _From, State) -> Procs = maps:remove(Procedure, State#state.procedures), %% No protocol message for unadvertise yet — procedure removed on disconnect {reply, ok, State#state{procedures = Procs}}; %%==================================================================== %% RPC Call %%==================================================================== handle_call({rpc_call, Procedure, Args, Timeout}, From, State) -> handle_call({rpc_call, Procedure, Args, Timeout, #{}}, From, State); handle_call({rpc_call, Procedure, Args, Timeout, Opts}, From, State) -> CallId = base64:encode(crypto:strong_rand_bytes(12)), TimerRef = erlang:send_after(Timeout, self(), {call_timeout, CallId}), PendingCalls = maps:put(CallId, {From, TimerRef}, State#state.pending_calls), maybe_send(call, build_call_msg(CallId, Procedure, Args, Opts), State), {noreply, State#state{pending_calls = PendingCalls}}; handle_call(_Request, _From, State) -> {reply, {error, unknown}, State}. %%==================================================================== %% Publish (async) %%==================================================================== handle_cast({publish, Topic, Payload}, State) -> BinPayload = to_binary_payload(Payload), maybe_send(publish, #{ <<"topic">> => Topic, <<"payload">> => BinPayload, <<"qos">> => 0, <<"retain">> => false, <<"message_id">> => crypto:strong_rand_bytes(16) }, State), {noreply, State}; %%==================================================================== %% Async RPC (non-blocking cross-relay forwarding) %%==================================================================== handle_cast({async_rpc_call, Procedure, Args, Timeout, Opts, CallbackPid, CorrelationId}, State) -> CallId = base64:encode(crypto:strong_rand_bytes(12)), TimerRef = erlang:send_after(Timeout, self(), {call_timeout, CallId}), PendingCalls = maps:put(CallId, {async, CallbackPid, CorrelationId, TimerRef}, State#state.pending_calls), maybe_send(call, build_call_msg(CallId, Procedure, Args, Opts), State), {noreply, State#state{pending_calls = PendingCalls}}; handle_cast(_Msg, State) -> {noreply, State}. %%==================================================================== %% Connection lifecycle %%==================================================================== handle_info(connect, State) -> QuicOpts = [{alpn, ["macula"]} | build_tls_opts(State#state.tls_verify)], case macula_quic:connect(State#state.host, State#state.port, QuicOpts, 10000) of {ok, Conn} -> case macula_quic:open_stream(Conn) of {ok, Stream} -> macula_quic:setopt(Stream, active, true), ?LOG_INFO("[relay_client] Connected to ~s", [State#state.url]), State2 = State#state{conn = Conn, stream = Stream, status = connected, recv_buffer = <<>>, reconnect_attempt = 0}, send_connect(State2), replay_state(State2), schedule_ping(), {noreply, State2}; {error, StreamErr} -> ?LOG_WARNING("[relay_client] Stream open failed on ~s: ~p", [State#state.url, StreamErr]), catch macula_quic:close(Conn), {noreply, schedule_failover(State)} end; {error, Reason} -> ?LOG_WARNING("[relay_client] Connect to ~s failed: ~p", [State#state.url, Reason]), {noreply, schedule_failover(State)}; {error, Type, Detail} -> ?LOG_WARNING("[relay_client] Connect to ~s failed: ~p ~p", [State#state.url, Type, Detail]), {noreply, schedule_failover(State)} end; handle_info({call_timeout, CallId}, State) -> case maps:get(CallId, State#state.pending_calls, undefined) of undefined -> {noreply, State}; {From, _TimerRef} -> ?LOG_WARNING("[mesh_client] RPC call ~s TIMED OUT (no reply from relay)", [CallId]), gen_server:reply(From, {error, timeout}), {noreply, State#state{pending_calls = maps:remove(CallId, State#state.pending_calls)}}; {async, CallbackPid, CorrelationId, _TimerRef} -> ?LOG_WARNING("[mesh_client] Async RPC ~s TIMED OUT", [CallId]), CallbackPid ! {relay_reply, CorrelationId, #{<<"error">> => #{<<"code">> => <<"timeout">>, <<"message">> => <<"Cross-relay RPC timed out">>}}, undefined}, {noreply, State#state{pending_calls = maps:remove(CallId, State#state.pending_calls)}} end; %%==================================================================== %% QUIC data received %%==================================================================== handle_info({quic, Data, _Stream, _Flags}, State) when is_binary(Data) -> Buffer = <<(State#state.recv_buffer)/binary, Data/binary>>, {NewBuffer, State2} = process_buffer(Buffer, State), {noreply, State2#state{recv_buffer = NewBuffer}}; %%==================================================================== %% QUIC lifecycle — reconnect on any closure %%==================================================================== handle_info({quic, peer_send_shutdown, _Stream, _}, State) -> handle_disconnect(State); handle_info({quic, peer_send_aborted, _Stream, _}, State) -> handle_disconnect(State); handle_info({quic, send_shutdown_complete, _Stream, _}, State) -> handle_disconnect(State); handle_info({quic, shutdown, _Ref, _Reason}, State) -> handle_disconnect(State); handle_info({quic, closed, _Ref, _Reason}, State) -> handle_disconnect(State); handle_info({quic, transport_shutdown, _Ref, _Reason}, State) -> handle_disconnect(State); handle_info({error, transport_down, _Detail}, State) -> handle_disconnect(State); handle_info({quic, streams_available, _Conn, _Info}, State) -> {noreply, State}; handle_info({quic, peer_needs_streams, _Conn, _Info}, State) -> {noreply, State}; handle_info(send_ping, #state{status = connected} = State) -> send_protocol_ping(State), schedule_ping(), {noreply, State#state{ping_sent_at = erlang:monotonic_time(millisecond)}}; handle_info(send_ping, State) -> %% Not connected — skip, next ping will fire after reconnect {noreply, State}; handle_info(Info, State) -> ?LOG_DEBUG("[relay_client] Unhandled: ~p", [Info]), {noreply, State}. terminate(_Reason, #state{conn = Conn, stream = Stream}) -> catch macula_quic:close(Stream), catch macula_quic:close(Conn), ok. %%==================================================================== %% Mesh ping — RTT measurement + publish %%==================================================================== schedule_ping() -> erlang:send_after(?PING_INTERVAL_MS, self(), send_ping). send_protocol_ping(State) -> Ts = erlang:system_time(millisecond), maybe_send(ping, #{timestamp => Ts}, State). publish_mesh_ping(RttMs, State) -> RelayHost = list_to_binary(State#state.host), NodeName = State#state.identity, Site = State#state.site, Lat = site_field(lat, Site), Lng = site_field(lng, Site), Payload = json:encode(#{ relay => RelayHost, node => NodeName, rtt_ms => RttMs, lat => Lat, lng => Lng, ts => erlang:system_time(second) }), maybe_send(publish, #{ <<"topic">> => ?PING_TOPIC, <<"payload">> => iolist_to_binary(Payload), <<"qos">> => 0, <<"retain">> => false, <<"message_id">> => base64:encode(crypto:strong_rand_bytes(8)) }, State). notify_discovery(RttMs, State) -> RelayHost = list_to_binary(State#state.host), try macula_relay_discovery ! {ping_rtt, RelayHost, RttMs} catch _:_ -> ok end. site_field(Key, Site) when is_map(Site) -> maps:get(Key, Site, 0.0); site_field(_, _) -> 0.0. %%==================================================================== %% Internal: message processing %%==================================================================== process_buffer(Buffer, State) when byte_size(Buffer) < 8 -> {Buffer, State}; process_buffer(<<_Version:8, _TypeId:8, _Flags:8, _Reserved:8, PayloadLen:32/big-unsigned, Rest/binary>> = Buffer, State) when byte_size(Rest) >= PayloadLen -> MsgBytes = binary:part(Buffer, 0, 8 + PayloadLen), Remaining = binary:part(Buffer, 8 + PayloadLen, byte_size(Buffer) - 8 - PayloadLen), State2 = handle_message(macula_protocol_decoder:decode(MsgBytes), State), process_buffer(Remaining, State2); process_buffer(Buffer, State) -> {Buffer, State}. %% Incoming SWIM messages — sent directly by remote relay handler on the inbound %% stream as a PUBLISH. Route to SWIM gen_server + extract piggybacked Bloom. handle_message({ok, {publish, #{<<"topic">> := <<"_swim.", _/binary>> = Topic, <<"payload">> := Payload}}}, State) -> Decoded = safe_decode_json(Payload), route_swim_message(Topic, Payload, Decoded), State; %% Incoming PUBLISH from relay handle_message({ok, {publish, Msg}}, State) -> Topic = maps:get(<<"topic">>, Msg, <<>>), Payload = maps:get(<<"payload">>, Msg, <<>>), Trace = maps:get(<<"_trace">>, Msg, undefined), invoke_matching_callbacks(Topic, Payload, Trace, State#state.subscriptions), State; %% Incoming CALL from relay (we're the provider) handle_message({ok, {call, Msg}}, State) -> Procedure = maps:get(<<"procedure">>, Msg, <<>>), CallId = maps:get(<<"call_id">>, Msg, <<>>), Args = maps:get(<<"args">>, Msg, #{}), Trace = maps:get(<<"_trace">>, Msg, undefined), dispatch_incoming_call(Procedure, CallId, Args, Trace, State), State; %% Incoming REPLY from relay (we're the caller) handle_message({ok, {reply, Msg}}, State) -> CallId = maps:get(<<"call_id">>, Msg, <<>>), case maps:get(CallId, State#state.pending_calls, undefined) of undefined -> ?LOG_DEBUG("[relay_client] Reply for unknown call_id ~s", [CallId]), State; {From, TimerRef} -> %% Synchronous call — unblock the caller erlang:cancel_timer(TimerRef), Trace = maps:get(<<"_trace">>, Msg, undefined), Reply = decode_reply(Msg, Trace), gen_server:reply(From, Reply), State#state{pending_calls = maps:remove(CallId, State#state.pending_calls)}; {async, CallbackPid, CorrelationId, TimerRef} -> %% Async call — deliver reply to callback PID erlang:cancel_timer(TimerRef), Trace = maps:get(<<"_trace">>, Msg, undefined), Reply = decode_reply(Msg, Trace), CallbackPid ! {relay_reply, CorrelationId, unwrap_reply(Reply), Trace}, State#state{pending_calls = maps:remove(CallId, State#state.pending_calls)} end; %% PONG — measure RTT and publish mesh ping handle_message({ok, {pong, _Msg}}, #state{ping_sent_at = undefined} = State) -> State; handle_message({ok, {pong, _Msg}}, State) -> RttMs = erlang:monotonic_time(millisecond) - State#state.ping_sent_at, publish_mesh_ping(RttMs, State), notify_discovery(RttMs, State), State#state{ping_sent_at = undefined, last_rtt_ms = RttMs}; handle_message({ok, {Type, _Msg}}, State) -> ?LOG_DEBUG("[relay_client] Ignoring message type: ~p", [Type]), State; handle_message({error, Reason}, State) -> ?LOG_WARNING("[relay_client] Decode error: ~p", [Reason]), State. %%==================================================================== %%==================================================================== %% RPC call/reply helpers — decomposed from handle_message %%==================================================================== %% Dispatch an incoming CALL to the registered handler or reply not_found. dispatch_incoming_call(Procedure, CallId, Args, Trace, State) -> case maps:get(Procedure, State#state.procedures, undefined) of undefined -> Reply = #{<<"call_id">> => CallId, <<"error">> => #{<<"code">> => <<"not_found">>}}, maybe_send(reply, maybe_add_trace(Reply, Trace), State); Handler -> Stream = State#state.stream, spawn(fun() -> execute_and_reply(Handler, Args, CallId, Trace, Stream) end) end. execute_and_reply(Handler, Args, CallId, Trace, Stream) -> Result = try Handler(Args) catch _:Err -> {error, Err} end, ReplyBase = format_call_result(Result, CallId), Binary = macula_protocol_encoder:encode(reply, maybe_add_trace(ReplyBase, Trace)), macula_quic:async_send(Stream, Binary). format_call_result({ok, R}, CallId) -> #{<<"call_id">> => CallId, <<"result">> => R}; format_call_result({error, R}, CallId) -> #{<<"call_id">> => CallId, <<"error">> => #{<<"code">> => <<"handler_error">>, <<"message">> => iolist_to_binary(io_lib:format("~p", [R]))}}. %% Decode a REPLY message into the caller-facing result tuple. decode_reply(Msg, Trace) -> Result = case maps:get(<<"result">>, Msg, undefined) of undefined -> {error, maps:get(<<"error">>, Msg, #{})}; R -> {ok, R} end, maybe_attach_trace(Result, Trace). maybe_attach_trace(Result, undefined) -> Result; maybe_attach_trace({ok, R}, Trace) -> {ok, R, #{trace => Trace}}; maybe_attach_trace(Other, _Trace) -> Other. %% @private Build a CALL message map with optional target and trace fields. build_call_msg(CallId, Procedure, Args, Opts) -> Msg0 = #{<<"call_id">> => CallId, <<"procedure">> => Procedure, <<"args">> => Args}, Msg1 = case maps:get(target, Opts, undefined) of undefined -> Msg0; Target -> Msg0#{<<"target">> => Target} end, case maps:get(<<"_trace">>, Opts, undefined) of undefined -> Msg1; Trace -> Msg1#{<<"_trace">> => Trace} end. %% @private Unwrap decoded reply to raw result for async callbacks. %% relay_reply expects the raw result map, not {ok, Result} tuples. unwrap_reply({ok, Result}) -> Result; unwrap_reply({ok, Result, _TraceInfo}) -> Result; unwrap_reply({error, ErrorMap}) -> #{<<"error">> => ErrorMap}; unwrap_reply(Other) -> Other. %% Internal helpers %%==================================================================== send_connect(State) -> NodeId = crypto:strong_rand_bytes(32), %% Use identity as node_name when set (multi-tenant stubs). %% Falls back to Erlang node name, then random hex. NodeName = case State#state.identity of Id when is_binary(Id), byte_size(Id) > 0 -> Id; _ -> case net_kernel:nodename() of nonode@nohost -> binary:encode_hex(crypto:strong_rand_bytes(8)); Name -> atom_to_binary(Name) end end, %% target_relay = the PREFERRED relay (first in list), not the actual connected host. %% When failover connects to a box URL, target_relay still identifies the intended %% virtual relay identity so the node registers under the correct city. TargetRelay = list_to_binary(State#state.preferred_host), Base = #{ version => <<"1.0">>, node_id => NodeId, node_name => NodeName, realm_id => State#state.realm, capabilities => [<<"pubsub">>, <<"rpc">>], endpoint => State#state.url, type => State#state.client_type, target_relay => TargetRelay }, Msg0 = case State#state.geo_identity of GeoId when is_map(GeoId), map_size(GeoId) > 0 -> Base#{identity => GeoId}; _ -> %% macula_connection is in macula-relay; build identity from env case build_env_identity() of EnvId when map_size(EnvId) > 0 -> Base#{identity => EnvId}; _ -> Base end end, %% Include site metadata if configured Msg1 = case State#state.site of SiteMap when is_map(SiteMap), map_size(SiteMap) > 0 -> Msg0#{site => SiteMap}; _ -> Msg0 end, %% Include previous_relay if this is a failover reconnect (different host) Msg = case State#state.previous_host of PrevHost when is_list(PrevHost), PrevHost =/= State#state.host -> Msg1#{previous_relay => list_to_binary(PrevHost)}; _ -> Msg1 end, maybe_send(connect, Msg, State). %% Replay all subscriptions and procedure registrations after reconnect replay_state(#state{subscriptions = Subs, procedures = Procs} = State) -> %% Batch all topics in one SUBSCRIBE Topics = lists:usort([T || {_Ref, {T, _Cb}} <- maps:to_list(Subs)]), case Topics of [] -> ok; _ -> ?LOG_INFO("[relay_client] Replaying ~p subscription(s)", [length(Topics)]), maybe_send(subscribe, #{<<"topics">> => Topics, <<"qos">> => 0}, State) end, %% Register all procedures maps:foreach(fun(Proc, _Handler) -> ?LOG_INFO("[relay_client] Replaying procedure: ~s", [Proc]), maybe_send(register_procedure, #{<<"procedure">> => Proc}, State) end, Procs). maybe_send(Type, _Msg, #state{stream = undefined}) -> ?LOG_DEBUG("[mesh_client] Dropped ~p — no stream", [Type]); maybe_send(Type, Msg, #state{stream = Stream}) -> Binary = macula_protocol_encoder:encode(Type, Msg), macula_quic:async_send(Stream, Binary). %% Ensure payload is a flat binary before passing to msgpack. %% Callers may pass: map, iolist (from json:encode), or binary. to_binary_payload(Payload) when is_binary(Payload) -> Payload; to_binary_payload(Payload) when is_map(Payload) -> iolist_to_binary(json:encode(Payload)); to_binary_payload(Payload) when is_list(Payload) -> iolist_to_binary(Payload); to_binary_payload(Payload) -> iolist_to_binary(json:encode(Payload)). handle_disconnect(#state{status = disconnected} = State) -> {noreply, State}; handle_disconnect(State) -> ?LOG_WARNING("[relay_client] Disconnected from ~s", [State#state.url]), catch macula_quic:close(State#state.stream), catch macula_quic:close(State#state.conn), State2 = State#state{conn = undefined, stream = undefined, status = disconnected, recv_buffer = <<>>}, {noreply, schedule_failover(State2)}. %% Pick next relay and schedule reconnect with exponential backoff + jitter. %% Uses geographic discovery if available, falls back to round-robin. schedule_failover(#state{reconnect_attempt = Attempt} = State) -> {NextUrl, NextState} = select_failover_relay(State), {NextHost, NextPort} = parse_url(NextUrl), BackoffMs = backoff_ms(Attempt), erlang:send_after(BackoffMs, self(), connect), ?LOG_INFO("[relay_client] Failover to ~s in ~bms (attempt ~b)", [NextUrl, BackoffMs, Attempt + 1]), NextState#state{url = NextUrl, host = NextHost, port = NextPort, status = disconnected, reconnect_attempt = Attempt + 1, previous_host = State#state.host}. %% Try geographic discovery first, fall back to round-robin. select_failover_relay(#state{host = CurrentHost} = State) -> CurrentHostname = list_to_binary(CurrentHost), case try_discovery_failover(CurrentHostname) of {ok, Url} -> {Url, State}; {error, _} -> round_robin_failover(State) end. try_discovery_failover(CurrentHostname) -> try macula_relay_discovery:nearest_except(CurrentHostname) catch _:_ -> {error, discovery_unavailable} end. round_robin_failover(#state{relays = Relays, relay_index = CurrentIdx} = State) -> NumRelays = length(Relays), NextIdx = case NumRelays > 1 of true -> (CurrentIdx + 1) rem NumRelays; false -> CurrentIdx end, NextUrl = lists:nth(NextIdx + 1, Relays), {NextUrl, State#state{relay_index = NextIdx}}. backoff_ms(Attempt) -> Base = min(?RECONNECT_BASE_MS * (1 bsl min(Attempt, 10)), ?RECONNECT_MAX_MS), Jitter = round(Base * ?RECONNECT_JITTER), Base + rand:uniform(2 * Jitter + 1) - Jitter - 1. parse_url(Url) when is_binary(Url) -> parse_url(binary_to_list(Url)); parse_url("https://" ++ Rest) -> parse_host_port(Rest); parse_url(Rest) -> parse_host_port(Rest). parse_host_port(HostPort) -> case string:split(HostPort, ":", trailing) of [Host, PortStr] -> {Host, list_to_integer(PortStr)}; [Host] -> {Host, 4433} end. invoke_matching_callbacks(Topic, Payload, Trace, Subscriptions) -> DecodedPayload = safe_decode_json(Payload), Base = #{topic => Topic, payload => DecodedPayload}, Event = case Trace of undefined -> Base; _ -> Base#{'_trace' => Trace} end, maps:foreach(fun(_Ref, {SubTopic, Callback}) -> case topic_matches(SubTopic, Topic) of true -> spawn(fun() -> try Callback(Event) catch C:E -> ?LOG_WARNING("[relay_client] Callback error: ~p:~p", [C, E]) end end); false -> ok end end, Subscriptions). topic_matches(Pattern, Topic) when Pattern =:= Topic -> true; topic_matches(<<"**">>, _Topic) -> true; topic_matches(Pattern, Topic) -> PatternParts = binary:split(Pattern, <<".">>, [global]), TopicParts = binary:split(Topic, <<".">>, [global]), parts_match(PatternParts, TopicParts). parts_match([], []) -> true; parts_match([<<"**">>], _) -> true; parts_match([<<"*">> | PR], [_ | TR]) -> parts_match(PR, TR); parts_match([P | PR], [T | TR]) when P =:= T -> parts_match(PR, TR); parts_match(_, _) -> false. safe_decode_json(Payload) when is_binary(Payload) -> try json:decode(Payload) catch _:_ -> Payload end; safe_decode_json(Payload) -> Payload. %% SWIM message routing — decomposed from handle_message to avoid nesting. route_swim_message(Topic, RawPayload, Decoded) -> From = swim_from(Decoded), forward_to_swim(Topic, From, RawPayload), store_piggybacked_bloom(From, Decoded). swim_from(#{<<"from">> := F}) when is_binary(F) -> F; swim_from(_) -> <<>>. forward_to_swim(<<"_swim.ack">>, From, Payload) -> swim_cast({ack_received, From, Payload}); forward_to_swim(<<"_swim.ping">>, From, Payload) -> swim_cast({ping_received, From, Payload}); forward_to_swim(_, _, _) -> ok. swim_cast(Msg) -> case erlang:whereis(macula_relay_swim) of undefined -> ok; Pid -> gen_server:cast(Pid, Msg) end. %% Mesh clients ignore bloom filters — those are relay-to-relay peering data. store_piggybacked_bloom(_, _) -> ok. maybe_add_trace(Msg, undefined) -> Msg; maybe_add_trace(Msg, Trace) -> Msg#{<<"_trace">> => Trace}. %% Build TLS options for QUIC client connection. %% verify_peer: use global TLS mode (production or development). %% none: force no-verify opts regardless of global mode. build_tls_opts(none) -> [{verify, none}]; build_tls_opts(_) -> macula_tls:quic_client_opts(). %% @private Build node identity from environment variables. %% Reads HECATE_GEO_* / MACULA_GEO_* env vars for geo metadata. build_env_identity() -> Pairs = [ {city, "HECATE_GEO_CITY"}, {country, "HECATE_GEO_COUNTRY"}, {owner, "HECATE_OWNER_NAME"}, {site, "HECATE_SITE_NAME"} ], Floats = [ {lat, "HECATE_GEO_LAT"}, {lng, "HECATE_GEO_LNG"} ], M1 = lists:foldl(fun({K, Env}, Acc) -> case os:getenv(Env) of false -> Acc; Val -> Acc#{K => list_to_binary(Val)} end end, #{}, Pairs), lists:foldl(fun({K, Env}, Acc) -> case os:getenv(Env) of false -> Acc; Val -> try Acc#{K => list_to_float(Val)} catch _:_ -> Acc end end end, M1, Floats).