%%%------------------------------------------------------------------- %%% @doc %%% Macula SDK connection manager. %%% %%% Manages individual HTTP/3 (QUIC) connections to Macula mesh nodes. %%% Handles connection lifecycle, authentication, message sending/receiving, %%% and subscription management. %%% @end %%%------------------------------------------------------------------- -module(macula_connection). -behaviour(gen_server). %% API -export([ start_link/2, stop/1, publish/3, publish/4, subscribe/3, unsubscribe/2, call/3, call/4, advertise/4, unadvertise/2 ]). %% gen_server callbacks -export([ init/1, handle_call/3, handle_cast/2, handle_info/2, terminate/2, code_change/3 ]). %% Exported for testing -ifdef(TEST). -export([ next_message_id/1, topic_matches/3, decode_messages/2, process_message/2 ]). -endif. -include_lib("kernel/include/logger.hrl"). -include("macula_config.hrl"). -record(state, { url :: binary(), opts :: map(), realm :: binary(), node_id :: binary(), %% Supervision tree (Phase 7) supervisor_pid :: undefined | pid(), connection_manager_pid :: undefined | pid(), pubsub_handler_pid :: undefined | pid(), rpc_handler_pid :: undefined | pid(), advertisement_manager_pid :: undefined | pid(), %% Legacy fields - will be removed after Phase 7 complete %% QUIC connection and stream connection :: undefined | pid(), stream :: undefined | pid(), %% Connection state status :: connecting | connected | disconnected | error, %% Subscriptions: #{SubscriptionRef => {Topic, Callback}} subscriptions :: #{reference() => {binary(), fun((map()) -> ok)}}, %% Pending RPC calls: #{CallId => {From, Timer}} pending_calls :: #{binary() => {term(), reference()}}, %% Pending DHT queries: #{QueryKey => {From, Timer}} pending_queries :: #{binary() => {term(), reference()}}, %% Pending publish acknowledgments for QoS 1: #{MsgId => {Topic, Payload, QoS, RetryCount, TimerRef}} pending_pubacks :: #{binary() => {binary(), binary(), integer(), integer(), reference()}}, %% Message ID counter msg_id_counter :: non_neg_integer(), %% Receive buffer for partial messages recv_buffer :: binary(), %% Service registry for DHT advertisement service_registry :: macula_service_registry:registry(), %% Advertised services with re-advertisement timers %% #{Procedure => #{handler, metadata, ttl, timer_ref}} advertised_services :: #{binary() => #{ handler := fun((map()) -> {ok, term()} | {error, term()}), metadata := map(), ttl := pos_integer(), timer_ref := reference() }}, %% Advertised subscriptions with re-advertisement timers %% #{Topic => #{sub_ref, ttl, timer_ref}} advertised_subscriptions :: #{binary() => #{ sub_ref := reference(), ttl := pos_integer(), timer_ref := reference() }}, %% Topic pattern matching configuration topic_separator :: binary(), % Default: <<".">> topic_wildcard_single :: binary(), % Default: <<"*">> (matches one segment) topic_wildcard_multi :: binary(), % Default: <<"**">> (matches zero or more segments) %% Provider selection strategy state provider_selector_state :: map(), %% Connection cache for multi-endpoint RPC %% #{Endpoint => #{connection, stream, last_used}} endpoint_connections :: #{binary() => #{ connection := pid(), stream := pid(), last_used := integer() }}, %% Pending subscriber discovery queries (for mesh-wide pub/sub) %% #{QueryId => {Topic, Payload, Qos, Opts}} pending_subscriber_queries :: #{binary() => {binary(), binary(), integer(), map()}} }). -define(CONNECT_RETRY_DELAY, 1000). %%%=================================================================== %%% API %%%=================================================================== %% @doc Start a client connection to a Macula mesh. -spec start_link(binary(), map()) -> {ok, pid()} | {error, term()}. start_link(Url, Opts) -> gen_server:start_link(?MODULE, {Url, Opts}, []). %% @doc Stop the client connection. -spec stop(pid()) -> ok. stop(Client) -> gen_server:stop(Client). %% @doc Publish an event through this client (no options). -spec publish(pid(), binary(), map() | binary()) -> ok | {error, term()}. publish(Client, Topic, Data) -> publish(Client, Topic, Data, #{}). %% @doc Publish an event through this client with options. -spec publish(pid(), binary(), map() | binary(), map()) -> ok | {error, term()}. publish(Client, Topic, Data, Opts) -> gen_server:call(Client, {publish, Topic, Data, Opts}, ?DEFAULT_TIMEOUT). %% @doc Subscribe to a topic through this client. -spec subscribe(pid(), binary(), fun((map()) -> ok)) -> {ok, reference()} | {error, term()}. subscribe(Client, Topic, Callback) -> gen_server:call(Client, {subscribe, Topic, Callback}, ?DEFAULT_TIMEOUT). %% @doc Unsubscribe from a topic. -spec unsubscribe(pid(), reference()) -> ok | {error, term()}. unsubscribe(Client, SubRef) -> gen_server:call(Client, {unsubscribe, SubRef}, ?DEFAULT_TIMEOUT). %% @doc Make an RPC call through this client (default timeout). -spec call(pid(), binary(), map() | list()) -> {ok, term()} | {error, term()}. call(Client, Procedure, Args) -> call(Client, Procedure, Args, #{}). %% @doc Make an RPC call through this client with options. -spec call(pid(), binary(), map() | list(), map()) -> {ok, term()} | {error, term()}. call(Client, Procedure, Args, Opts) -> Timeout = maps:get(timeout, Opts, ?CALL_TIMEOUT), gen_server:call(Client, {call, Procedure, Args, Opts}, Timeout + 1000). %% @doc Advertise a service that this client provides. %% %% Registers a handler function locally and publishes the service to the DHT %% so other clients can discover and call it. -spec advertise(pid(), binary(), macula_service_registry:handler_fn(), map()) -> {ok, reference()} | {error, term()}. advertise(Client, Procedure, Handler, Opts) -> gen_server:call(Client, {advertise, Procedure, Handler, Opts}, ?DEFAULT_TIMEOUT). %% @doc Stop advertising a service. %% %% Removes the local handler and stops advertising to the DHT. -spec unadvertise(pid(), binary()) -> ok | {error, term()}. unadvertise(Client, Procedure) -> gen_server:call(Client, {unadvertise, Procedure}, ?DEFAULT_TIMEOUT). %%%=================================================================== %%% gen_server callbacks %%%=================================================================== %% @private init({Url, Opts}) -> %% Parse URL to extract host and port {Host, Port} = macula_utils:parse_url(Url), %% Get realm (required) Realm = get_realm_from_opts(Opts), %% Generate or get node ID NodeId = maps:get(node_id, Opts, macula_utils:generate_node_id()), %% Get provider selection strategy from options (default: random) SelectionStrategy = maps:get(provider_selection_strategy, Opts, random), %% Get topic pattern matching configuration (defaults: dot-separated with * wildcards) TopicSeparator = maps:get(topic_separator, Opts, <<".">>), TopicWildcardSingle = maps:get(topic_wildcard_single, Opts, <<"*">>), TopicWildcardMulti = maps:get(topic_wildcard_multi, Opts, <<"**">>), %% Phase 7: Start supervision tree ?LOG_INFO("[Phase 7] Starting supervision tree for ~s", [Url]), %% Prepare opts for handlers (includes node_id, realm, url) HandlerOpts = Opts#{ node_id => NodeId, realm => Realm, url => Url, host => Host, port => Port, provider_strategy => SelectionStrategy }, %% Start supervision tree {ok, SupPid} = macula_connection_sup:start_link(Url, HandlerOpts), %% Look up child PIDs from supervisor Children = supervisor:which_children(SupPid), ConnMgrPid = find_child_pid(Children, connection_manager), PubSubPid = find_child_pid(Children, pubsub_handler), RpcPid = find_child_pid(Children, rpc_handler), AdvMgrPid = find_child_pid(Children, advertisement_manager), ?LOG_INFO("[Phase 7] Supervision tree started - ConnMgr: ~p, PubSub: ~p, RPC: ~p, AdvMgr: ~p", [ConnMgrPid, PubSubPid, RpcPid, AdvMgrPid]), State = #state{ url = Url, opts = Opts#{host => Host, port => Port}, realm = Realm, node_id = NodeId, %% Phase 7: New supervision tree fields supervisor_pid = SupPid, connection_manager_pid = ConnMgrPid, pubsub_handler_pid = PubSubPid, rpc_handler_pid = RpcPid, advertisement_manager_pid = AdvMgrPid, %% Legacy fields (still needed for backward compatibility during Phase 7) connection = undefined, stream = undefined, status = connecting, subscriptions = #{}, pending_calls = #{}, pending_queries = #{}, pending_pubacks = #{}, msg_id_counter = 0, recv_buffer = <<>>, service_registry = macula_service_registry:new(), advertised_services = #{}, advertised_subscriptions = #{}, topic_separator = TopicSeparator, topic_wildcard_single = TopicWildcardSingle, topic_wildcard_multi = TopicWildcardMulti, provider_selector_state = #{ strategy => SelectionStrategy, counters => #{} }, endpoint_connections = #{}, pending_subscriber_queries = #{} }, %% Register this connection process with gproc so the gateway can find it gproc:reg({p, l, macula_connection}), %% Schedule periodic cache pruning (every 60 seconds) erlang:send_after(60000, self(), prune_caches), %% NOTE: Connection is now managed by connection_manager, not here %% The old "self() ! connect" is removed {ok, State}. %% @private %% Phase 7.3: Delegate to pubsub_handler handle_call({publish, Topic, Data, Opts}, _From, State) -> Result = macula_pubsub_handler:publish(State#state.pubsub_handler_pid, Topic, Data, Opts), {reply, Result, State}; %% Phase 7.3: Delegate to pubsub_handler handle_call({subscribe, Topic, Callback}, _From, State) -> Result = macula_pubsub_handler:subscribe(State#state.pubsub_handler_pid, Topic, Callback), {reply, Result, State}; %% Phase 7.3: Delegate to pubsub_handler handle_call({unsubscribe, SubRef}, _From, State) -> Result = macula_pubsub_handler:unsubscribe(State#state.pubsub_handler_pid, SubRef), {reply, Result, State}; %% Phase 7.3: Delegate to rpc_handler handle_call({call, Procedure, Args, Opts}, _From, State) -> Result = macula_rpc_handler:call(State#state.rpc_handler_pid, Procedure, Args, Opts), {reply, Result, State}; %% Phase 7.3: Delegate to advertisement_manager handle_call({advertise, Procedure, Handler, Opts}, _From, State) -> Result = macula_advertisement_manager:advertise_service(State#state.advertisement_manager_pid, Procedure, Handler, Opts), {reply, Result, State}; %% Phase 7.3: Delegate to advertisement_manager handle_call({unadvertise, Procedure}, _From, State) -> Result = macula_advertisement_manager:unadvertise_service(State#state.advertisement_manager_pid, Procedure), {reply, Result, State}; handle_call(_Request, _From, State) -> {reply, {error, unknown_request}, State}. %% @private handle_cast({do_publish, PublishMsg, Qos, BinaryTopic, Payload, _Opts, MsgId}, State) -> %% Spawn the publish send to avoid blocking the gen_server %% Even async_send can block at NIF level or due to flow control Stream = State#state.stream, NodeId = State#state.node_id, spawn(fun() -> case send_message_raw(publish, PublishMsg, Stream) of ok -> ok; {error, Reason} -> ?LOG_WARNING("[~s] Failed to send publish: ~p", [NodeId, Reason]) end end), %% If QoS 1, track for acknowledgment State2 = track_qos_if_needed(Qos, MsgId, BinaryTopic, Payload, State), %% Trigger async discovery of remote subscribers (mesh-wide pub/sub) %% Using cast to make the entire discovery process asynchronous gen_server:cast(self(), {discover_subscribers, BinaryTopic, Payload, Qos, _Opts}), {noreply, State2}; %% @private %% Handle async discovery of remote subscribers (mesh-wide pub/sub) %% Spawn the entire discovery process to avoid any blocking handle_cast({discover_subscribers, Topic, Payload, Qos, Opts}, State) -> %% Spawn the entire discovery process to completely avoid blocking spawn(fun() -> _State2 = discover_remote_subscribers(Topic, Payload, Qos, Opts, State), %% Send updated state back to gen_server if needed (for now, discovery is fire-and-forget) ok end), {noreply, State}; %% @doc Handle routed RPC REPLY from local gateway handle_cast({rpc_route_reply, RpcRouteMsg}, State) -> io:format("[Connection ~s] Received routed REPLY from gateway~n", [State#state.node_id]), %% Process the rpc_route message using existing handler NewState = process_message({rpc_route, RpcRouteMsg}, State), {noreply, NewState}; handle_cast(_Msg, State) -> {noreply, State}. %% @private handle_info(connect, State) -> case do_connect(State) of {ok, State2} -> {noreply, State2}; {error, Reason} -> ?LOG_ERROR("Connection failed: ~p, retrying in ~p ms", [Reason, ?CONNECT_RETRY_DELAY]), erlang:send_after(?CONNECT_RETRY_DELAY, self(), connect), {noreply, State#state{status = error}} end; handle_info({quic, Data, Stream, _Props}, State) when is_binary(Data) -> %% Received data from QUIC stream (main or endpoint connection) MainStream = State#state.stream, EndpointConnections = State#state.endpoint_connections, %% Check if this is the main stream or an endpoint stream IsMainStream = (Stream =:= MainStream), IsEndpointStream = lists:any(fun({_Endpoint, #{stream := S}}) -> S =:= Stream end, maps:to_list(EndpointConnections)), StreamType = if IsMainStream -> "main"; IsEndpointStream -> "endpoint"; true -> "unknown" end, ?LOG_INFO("Received ~p bytes on ~s stream ~p", [byte_size(Data), StreamType, Stream]), %% Valid stream - process the data, otherwise ignore handle_stream_data(IsMainStream orelse IsEndpointStream, Data, Stream, State); %% Handle QUIC control messages (non-binary Data) handle_info({quic, ControlMsg, _Stream, _Props}, State) when is_atom(ControlMsg) -> %% QUIC control message (send_shutdown_complete, etc.) - ignore ?LOG_DEBUG("Ignoring QUIC control message: ~p", [ControlMsg]), {noreply, State}; handle_info({call_timeout, CallId}, State) -> %% RPC call timed out case maps:get(CallId, State#state.pending_calls, undefined) of undefined -> {noreply, State}; {From, _Timer} -> %% Simple timeout without failover context ?LOG_WARNING("RPC call ~s timed out (no failover available)", [CallId]), gen_server:reply(From, {error, timeout}), PendingCalls = maps:remove(CallId, State#state.pending_calls), {noreply, State#state{pending_calls = PendingCalls}}; {From, _Timer, FailoverContext} -> %% Timeout with failover context - try next provider #{ procedure := Procedure, args := Args, opts := Opts, all_providers := AllProviders, excluded_providers := ExcludedProviders, attempt := Attempt } = FailoverContext, ?LOG_WARNING("RPC call to ~s timed out on attempt ~p, trying next provider", [Procedure, Attempt]), %% Remove this call from pending PendingCalls = maps:remove(CallId, State#state.pending_calls), State2 = State#state{pending_calls = PendingCalls}, %% Retry with next provider case do_remote_call_with_failover(Procedure, Args, Opts, From, AllProviders, ExcludedProviders, Attempt + 1, State2) of {noreply, State3} -> {noreply, State3}; {reply, Error, State3} -> %% All providers exhausted or max attempts reached gen_server:reply(From, Error), {noreply, State3} end end; handle_info({find_value_timeout, ServiceKey}, State) -> %% DHT FIND_VALUE query timed out case maps:get(ServiceKey, State#state.pending_queries, undefined) of undefined -> %% Already handled or cancelled {noreply, State}; {From, Procedure, Args, Opts, _Registry, _Timer} -> %% Query timed out - fall back to direct call ?LOG_WARNING("FIND_VALUE query for service ~s timed out. Attempting direct call.", [Procedure]), PendingQueries = maps:remove(ServiceKey, State#state.pending_queries), State2 = State#state{pending_queries = PendingQueries}, do_direct_call(Procedure, Args, Opts, From, State2) end; handle_info({readvertise, Procedure}, State) -> %% Re-advertisement timer fired - re-publish to DHT AdvertisedServices = State#state.advertised_services, case maps:get(Procedure, AdvertisedServices, undefined) of undefined -> %% Service was unadvertised, timer should have been cancelled but wasn't ?LOG_WARNING("Re-advertisement timer fired for unadvertised service: ~s", [Procedure]), {noreply, State}; #{metadata := Metadata, ttl := TTL} = ServiceInfo -> %% Re-publish to DHT ProviderInfo = #{ node_id => State#state.node_id, endpoint => State#state.url, metadata => Metadata }, case macula_service_registry:publish_to_dht(macula_routing_server, Procedure, ProviderInfo, TTL, 20) of ok -> ?LOG_DEBUG("Re-advertised service ~s to DHT", [Procedure]); {error, Reason} -> ?LOG_WARNING("Failed to re-advertise service ~s to DHT: ~p", [Procedure, Reason]) end, %% Schedule next re-advertisement ReadvInterval = max(10, TTL - 60) * 1000, TimerRef = erlang:send_after(ReadvInterval, self(), {readvertise, Procedure}), ?LOG_DEBUG("Scheduled next re-advertisement for ~s in ~p seconds", [Procedure, ReadvInterval div 1000]), %% Update timer reference ServiceInfo2 = ServiceInfo#{timer_ref => TimerRef}, AdvertisedServices2 = AdvertisedServices#{Procedure => ServiceInfo2}, {noreply, State#state{advertised_services = AdvertisedServices2}} end; %% Re-advertise subscription in DHT (periodic refresh) handle_info({resubscribe, Topic}, State) -> case maps:get(Topic, State#state.advertised_subscriptions, undefined) of undefined -> %% Subscription was removed, don't re-advertise ?LOG_DEBUG("[~s] Skipping re-subscription for ~s (no longer subscribed)", [State#state.node_id, Topic]), {noreply, State}; SubInfo -> %% Cancel old timer OldTimerRef = maps:get(timer_ref, SubInfo), erlang:cancel_timer(OldTimerRef), %% Re-advertise (same as initial advertisement) SubRef = maps:get(sub_ref, SubInfo), State2 = advertise_subscription_in_dht(Topic, SubRef, State), ?LOG_DEBUG("[~s] Re-advertised subscription for topic ~s", [State#state.node_id, Topic]), {noreply, State2} end; handle_info({puback_timeout, MsgId}, State) -> %% Handle PUBACK timeout - retry or give up PendingPubacks = State#state.pending_pubacks, case maps:get(MsgId, PendingPubacks, undefined) of undefined -> %% Already acknowledged or removed {noreply, State}; {Topic, Payload, Qos, RetryCount, _OldTimerRef} -> NewRetryCount = RetryCount + 1, if NewRetryCount >= ?PUBACK_MAX_RETRIES -> %% Max retries reached, give up ?LOG_ERROR("[~s] PUBACK timeout for message ~p (topic: ~s) after ~p retries - giving up", [State#state.node_id, MsgId, Topic, NewRetryCount]), {noreply, State#state{pending_pubacks = maps:remove(MsgId, PendingPubacks)}}; true -> %% Retry sending the message ?LOG_WARNING("[~s] PUBACK timeout for message ~p (topic: ~s) - retry ~p/~p", [State#state.node_id, MsgId, Topic, NewRetryCount, ?PUBACK_MAX_RETRIES]), %% Rebuild and resend publish message PublishMsg = #{ topic => Topic, payload => Payload, qos => Qos, retain => false, message_id => MsgId }, case send_message(publish, PublishMsg, State) of ok -> %% Start new timer NewTimerRef = erlang:send_after(?PUBACK_TIMEOUT, self(), {puback_timeout, MsgId}), %% Update pending with new retry count and timer PendingPubacks2 = PendingPubacks#{ MsgId => {Topic, Payload, Qos, NewRetryCount, NewTimerRef} }, {noreply, State#state{pending_pubacks = PendingPubacks2}}; {error, Reason} -> ?LOG_ERROR("[~s] Failed to retry PUBACK message ~p: ~p", [State#state.node_id, MsgId, Reason]), {noreply, State#state{pending_pubacks = maps:remove(MsgId, PendingPubacks)}} end end end; %% Periodic cache pruning handle_info(prune_caches, State) -> Registry = State#state.service_registry, %% Prune expired entries from both service cache and subscriber cache {Registry2, ServicesPruned} = macula_service_registry:prune_expired_services(Registry), {Registry3, SubscribersPruned} = macula_service_registry:prune_expired_subscribers(Registry2), TotalPruned = ServicesPruned + SubscribersPruned, case TotalPruned of 0 -> ok; % No entries expired _ -> ?LOG_DEBUG("[~s] Cache pruning: removed ~p service entries, ~p subscriber entries", [State#state.node_id, ServicesPruned, SubscribersPruned]) end, %% Schedule next pruning in 60 seconds erlang:send_after(60000, self(), prune_caches), {noreply, State#state{service_registry = Registry3}}; handle_info(Info, State) -> %% Ignore unexpected messages ?LOG_DEBUG("Unhandled handle_info message: ~p", [Info]), {noreply, State}. %% @private terminate(_Reason, #state{stream = Stream, connection = Conn, endpoint_connections = EndpointConns}) -> %% Clean up QUIC resources catch macula_quic:close(Stream), catch macula_quic:close(Conn), %% Close all endpoint connections lists:foreach(fun({_Endpoint, #{connection := C, stream := S}}) -> catch macula_quic:close(S), catch macula_quic:close(C) end, maps:to_list(EndpointConns)), ok. %% @private code_change(_OldVsn, State, _Extra) -> {ok, State}. %%%=================================================================== %%% Internal functions %%%=================================================================== %% @doc Extract and normalize realm from options (pattern matching on type). -spec get_realm_from_opts(map()) -> binary(). get_realm_from_opts(Opts) -> normalize_realm(maps:get(realm, Opts, undefined)). normalize_realm(undefined) -> error({missing_required_option, realm}); normalize_realm(Realm) when is_binary(Realm) -> Realm; normalize_realm(Realm) when is_list(Realm) -> list_to_binary(Realm); normalize_realm(Realm) when is_atom(Realm) -> atom_to_binary(Realm). %% @doc Find child PID from supervisor children list (Phase 7 helper). -spec find_child_pid(list(), atom()) -> pid(). find_child_pid(Children, ChildId) -> extract_child_pid(lists:keyfind(ChildId, 1, Children), ChildId). extract_child_pid({_, Pid, _Type, _Modules}, _ChildId) when is_pid(Pid) -> Pid; extract_child_pid({_, undefined, _Type, _Modules}, ChildId) -> error({child_not_started, ChildId}); extract_child_pid(false, ChildId) -> error({child_not_found, ChildId}). %% @doc Establish QUIC connection and perform handshake. -spec do_connect(#state{}) -> {ok, #state{}} | {error, term()}. do_connect(State) -> #{host := Host, port := Port} = State#state.opts, %% Connect via QUIC QuicOpts = [ {alpn, ["macula"]}, {verify, none} %% TODO: Add proper TLS verification ], ConnectResult = try macula_quic:connect(Host, Port, QuicOpts, ?DEFAULT_TIMEOUT) catch _:Error -> {error, Error} end, case ConnectResult of {ok, Conn} -> %% Open bidirectional stream case macula_quic:open_stream(Conn) of {ok, Stream} -> %% Set stream to active mode so we receive messages case quicer:setopt(Stream, active, true) of ok -> ?LOG_DEBUG("Stream set to active mode"); {error, SetOptErr} -> ?LOG_WARNING("Failed to set stream active: ~p", [SetOptErr]) end, %% Send CONNECT message %% Include advertise endpoint for peer-to-peer connections LocalEndpoint = case application:get_env(macula, advertise_endpoint) of {ok, Endpoint} when is_binary(Endpoint) -> Endpoint; _ -> %% Construct from NODE_HOST env var NodeHost = list_to_binary(os:getenv("NODE_HOST", "localhost")), <<"https://", NodeHost/binary, ":9443">> end, ConnectMsg = #{ version => <<"1.0">>, node_id => State#state.node_id, realm_id => State#state.realm, capabilities => [pubsub, rpc], endpoint => LocalEndpoint }, case send_message_raw(connect, ConnectMsg, Stream) of ok -> ?LOG_INFO("Connected to Macula mesh: ~s:~p", [Host, Port]), %% Add connected server node to DHT routing table try %% Parse the server's endpoint URL to get address ServerAddress = parse_server_endpoint(State#state.url), %% Use correct node_info structure with address field ServerNodeInfo = #{ node_id => crypto:strong_rand_bytes(32), %% We'll get actual node_id from CONNECTED response address => ServerAddress }, macula_routing_server:add_node(macula_routing_server, ServerNodeInfo), ?LOG_DEBUG("Added server node to DHT routing table with address: ~p", [ServerAddress]) catch _:E -> ?LOG_WARNING("Failed to add server node to DHT: ~p", [E]) end, {ok, State#state{ connection = Conn, stream = Stream, status = connected }}; {error, Reason} -> macula_quic:close(Stream), macula_quic:close(Conn), {error, {handshake_failed, Reason}} end; {error, Reason} -> macula_quic:close(Conn), {error, {stream_open_failed, Reason}} end; {error, Reason} -> {error, {connection_failed, Reason}}; {error, Type, Details} -> {error, {connection_failed, {Type, Details}}}; Other -> {error, {connection_failed, Other}} end. %% @doc Send a protocol message through the established stream. -spec send_message(atom(), map(), #state{}) -> ok | {error, term()}. send_message(Type, Msg, #state{stream = Stream}) -> send_message_raw(Type, Msg, Stream). %% @doc Send a protocol message through a stream (raw). -spec send_message_raw(atom(), map(), pid()) -> ok | {error, term()}. send_message_raw(Type, Msg, Stream) -> try io:format("[send_message_raw] Type=~p, Msg=~p~n", [Type, Msg]), Binary = macula_protocol_encoder:encode(Type, Msg), macula_quic:async_send(Stream, Binary) catch error:Reason:Stacktrace -> io:format("[send_message_raw] ERROR: ~p~nStacktrace: ~p~n", [Reason, Stacktrace]), {error, {encode_error, Reason}} end. %% @doc Dispatch stream data based on validity (pattern matching on boolean). handle_stream_data(true, Data, _Stream, State) -> handle_received_data(Data, State); handle_stream_data(false, _Data, Stream, State) -> ?LOG_WARNING("Received data from unknown stream: ~p", [Stream]), {noreply, State}. %% @doc Track QoS 1 message for acknowledgment (pattern matching on QoS level). track_qos_if_needed(1, MsgId, BinaryTopic, Payload, State) -> %% Start timeout timer TimerRef = erlang:send_after(?PUBACK_TIMEOUT_MS, self(), {puback_timeout, MsgId}), %% Store pending acknowledgment PendingPubacks = State#state.pending_pubacks, State#state{ pending_pubacks = PendingPubacks#{ MsgId => {BinaryTopic, Payload, 1, 0, TimerRef} } }; track_qos_if_needed(_Qos, _MsgId, _BinaryTopic, _Payload, State) -> State. %% @doc Handle received data from the stream. -spec handle_received_data(binary(), #state{}) -> {noreply, #state{}}. handle_received_data(Data, State) -> %% Append to receive buffer Buffer = <<(State#state.recv_buffer)/binary, Data/binary>>, %% Try to decode messages {Messages, RemainingBuffer} = decode_messages(Buffer, []), %% Process each message State2 = lists:foldl(fun process_message/2, State, Messages), {noreply, State2#state{recv_buffer = RemainingBuffer}}. %% @doc Decode all complete messages from buffer. -spec decode_messages(binary(), list()) -> {list(), binary()}. decode_messages(Buffer, Acc) when byte_size(Buffer) < 8 -> %% Not enough for header {lists:reverse(Acc), Buffer}; decode_messages(<<_Version:8, _TypeId:8, _Flags:8, _Reserved:8, PayloadLen:32/big-unsigned, Rest/binary>> = Buffer, Acc) -> case byte_size(Rest) of ActualLen when ActualLen >= PayloadLen -> %% We have a complete message case macula_protocol_decoder:decode(Buffer) of {ok, {Type, Msg}} -> %% Skip this message and continue <<_:8/binary, _Payload:PayloadLen/binary, Remaining/binary>> = Buffer, decode_messages(Remaining, [{Type, Msg} | Acc]); {error, _Reason} -> %% Decode error, skip this message <<_:8/binary, _Payload:PayloadLen/binary, Remaining/binary>> = Buffer, decode_messages(Remaining, Acc) end; _ -> %% Incomplete message, wait for more data {lists:reverse(Acc), Buffer} end. %% @doc Process a received message. -spec process_message({atom(), map()}, #state{}) -> #state{}. process_message({publish, Msg}, State) -> %% Handle incoming publish (for subscriptions) %% Support both atom and binary keys from MessagePack decoding Topic = case maps:get(topic, Msg, undefined) of undefined -> maps:get(<<"topic">>, Msg); T -> T end, Payload = case maps:get(payload, Msg, undefined) of undefined -> maps:get(<<"payload">>, Msg); P -> P end, ?LOG_DEBUG("[~s] Received PUBLISH message: topic=~s, payload_size=~p bytes", [State#state.node_id, Topic, byte_size(Payload)]), %% Find matching subscriptions and invoke callbacks SubscriptionMatches = lists:filter( fun({_SubRef, {SubTopic, _Callback}}) -> topic_matches(SubTopic, Topic, State) end, maps:to_list(State#state.subscriptions) ), case SubscriptionMatches of [] -> ?LOG_DEBUG("[~s] No matching subscriptions for topic: ~s", [State#state.node_id, Topic]); Matches -> ?LOG_INFO("[~s] Found ~p subscription(s) for topic: ~s", [State#state.node_id, length(Matches), Topic]), %% Invoke each matching callback lists:foreach(fun({SubRef, {SubTopic, Callback}}) -> spawn(fun() -> try %% Decode payload Event = macula_utils:decode_json(Payload), ?LOG_DEBUG("[~s] Invoking callback for subscription ~p (topic: ~s)", [State#state.node_id, SubRef, SubTopic]), %% Invoke callback Callback(Event), ?LOG_DEBUG("[~s] Callback completed successfully for subscription ~p", [State#state.node_id, SubRef]) catch Class:Reason:Stacktrace -> ?LOG_ERROR("[~s] Callback error for subscription ~p: ~p:~p~nStacktrace: ~p", [State#state.node_id, SubRef, Class, Reason, Stacktrace]) end end) end, Matches) end, State; process_message({call, Msg}, State) -> %% Handle incoming RPC call (for locally advertised services) #{procedure := Procedure, args := Args, call_id := CallId} = Msg, %% Look up local handler case macula_service_registry:get_local_handler(State#state.service_registry, Procedure) of {ok, Handler} -> %% Decode args DecodedArgs = macula_utils:decode_json(Args), %% Execute handler (spawn to avoid blocking) spawn(fun() -> try Result = Handler(DecodedArgs), send_rpc_reply(CallId, Result, State) catch _:Error -> send_rpc_error(CallId, Error, State) end end), State; not_found -> %% Service not found locally spawn(fun() -> send_rpc_error(CallId, service_not_found, State) end), State end; process_message({reply, Msg}, State) -> %% Handle RPC reply %% MessagePack decoder returns binary keys #{<<"call_id">> := CallId} = Msg, case maps:get(CallId, State#state.pending_calls, undefined) of undefined -> ?LOG_WARNING("[~s] Received REPLY for unknown call_id: ~p", [State#state.node_id, binary:encode_hex(CallId)]), State; {From, Timer} -> %% Simple reply without failover context erlang:cancel_timer(Timer), case maps:get(<<"result">>, Msg, undefined) of undefined -> %% Error response Error = maps:get(<<"error">>, Msg, <<"Unknown error">>), gen_server:reply(From, {error, Error}), State#state{pending_calls = maps:remove(CallId, State#state.pending_calls)}; Result -> %% Success response DecodedResult = macula_utils:decode_json(Result), gen_server:reply(From, {ok, DecodedResult}), State#state{pending_calls = maps:remove(CallId, State#state.pending_calls)} end; {From, Timer, FailoverContext} -> %% Reply with failover context - check for errors and retry if needed erlang:cancel_timer(Timer), case maps:get(<<"result">>, Msg, undefined) of undefined -> %% Error response - try failover #{ procedure := Procedure, args := Args, opts := Opts, all_providers := AllProviders, excluded_providers := ExcludedProviders, attempt := Attempt } = FailoverContext, Error = maps:get(<<"error">>, Msg, <<"Unknown error">>), ?LOG_WARNING("RPC call to ~s failed with error: ~p (attempt ~p), trying next provider", [Procedure, Error, Attempt]), %% Remove this call from pending PendingCalls = maps:remove(CallId, State#state.pending_calls), State2 = State#state{pending_calls = PendingCalls}, %% Retry with next provider case do_remote_call_with_failover(Procedure, Args, Opts, From, AllProviders, ExcludedProviders, Attempt + 1, State2) of {noreply, State3} -> State3; {reply, ErrorReply, State3} -> %% All providers exhausted or max attempts reached gen_server:reply(From, ErrorReply), State3 end; Result -> %% Success response DecodedResult = macula_utils:decode_json(Result), gen_server:reply(From, {ok, DecodedResult}), State#state{pending_calls = maps:remove(CallId, State#state.pending_calls)} end end; process_message({pong, _Msg}, State) -> %% TODO: Handle ping/pong for keepalive State; process_message({rpc_route, RpcRouteMsg}, State) -> %% Handle routed RPC messages (CALL or REPLY) LocalNodeId = State#state.node_id, case macula_rpc_routing:should_deliver_locally(LocalNodeId, RpcRouteMsg) of true -> %% Message is for us - unwrap and deliver %% MessagePack decoder returns binary keys #{<<"payload_type">> := PayloadType, <<"payload">> := Payload} = RpcRouteMsg, case PayloadType of <<"reply">> -> %% Routed REPLY coming back from provider ?LOG_INFO("[~s] Received routed REPLY", [State#state.node_id]), process_message({reply, Payload}, State); <<"call">> -> %% Routed CALL - forward to gateway for handler processing ?LOG_INFO("[~s] Connection received routed CALL, forwarding to gateway", [State#state.node_id]), case whereis(macula_gateway) of undefined -> ?LOG_WARNING("[~s] No gateway running, cannot process routed CALL", [State#state.node_id]), State; GatewayPid -> %% Send the entire rpc_route message to gateway for processing gen_server:cast(GatewayPid, {process_rpc_route, RpcRouteMsg}), State end end; false -> %% Not for us - this shouldn't happen at connection level %% (gateway does the forwarding) ?LOG_WARNING("[~s] Connection received rpc_route not for us (destination: ~p)", [State#state.node_id, binary:encode_hex(maps:get(<<"destination_node_id">>, RpcRouteMsg))]), State end; process_message({find_value_reply, Msg}, State) -> %% Handle FIND_VALUE_REPLY messages from DHT io:format("[~s] ==> RECEIVED FIND_VALUE_REPLY MESSAGE <=~n", [State#state.node_id]), handle_find_value_reply(Msg, State); process_message({store_reply, _Msg}, State) -> %% STORE replies don't need handling on client side State; process_message({puback, Msg}, State) -> %% Handle PUBACK acknowledgment for QoS 1 MsgId = maps:get(message_id, Msg), PendingPubacks = State#state.pending_pubacks, case maps:get(MsgId, PendingPubacks, undefined) of undefined -> ?LOG_WARNING("[~s] Received PUBACK for unknown message_id: ~p", [State#state.node_id, MsgId]), State; {Topic, _Payload, _Qos, _RetryCount, TimerRef} -> %% Cancel timeout timer erlang:cancel_timer(TimerRef), %% Remove from pending ?LOG_DEBUG("[~s] PUBACK received for message ~p (topic: ~s)", [State#state.node_id, MsgId, Topic]), State#state{pending_pubacks = maps:remove(MsgId, PendingPubacks)} end; process_message(OtherMsg, State) -> %% Ignore unknown message types io:format("[~s] Unknown message: ~p~n", [State#state.node_id, OtherMsg]), State. %% @doc Handle incoming FIND_VALUE_REPLY from DHT. -spec handle_find_value_reply(map(), #state{}) -> #state{}. handle_find_value_reply(Msg, State) -> %% Check if this is for subscriber discovery or RPC service discovery HasSubscriberQuery = maps:size(State#state.pending_subscriber_queries) > 0, %% Decode the DHT protocol message case macula_routing_protocol:decode_find_value_reply(Msg) of {ok, {value, Provider}} when is_map(Provider) -> %% Single provider returned - normalize to list and convert binary keys to atoms Providers = [macula_utils:normalize_provider(Provider)], ?LOG_INFO("[~s] Received FIND_VALUE_REPLY with ~p provider(s)", [State#state.node_id, length(Providers)]), %% Log provider details for debugging lists:foreach(fun(P) -> NodeId = maps:get(node_id, P, <<"unknown">>), Endpoint = maps:get(endpoint, P, <<"unknown">>), ?LOG_INFO("[~s] Provider: node_id=~s, endpoint=~s", [State#state.node_id, NodeId, Endpoint]) end, Providers), %% Route to appropriate handler case HasSubscriberQuery of true -> handle_subscriber_discovery_reply(Providers, State); false -> continue_pending_query_with_providers(Providers, State) end; {ok, {value, Providers}} when is_list(Providers) -> %% List of providers returned - normalize all providers NormalizedProviders = lists:map(fun macula_utils:normalize_provider/1, Providers), ?LOG_INFO("[~s] Received FIND_VALUE_REPLY with ~p provider(s)", [State#state.node_id, length(NormalizedProviders)]), %% Log provider details for debugging lists:foreach(fun(P) -> NodeId = maps:get(node_id, P, <<"unknown">>), Endpoint = maps:get(endpoint, P, <<"unknown">>), ?LOG_INFO("[~s] Provider: node_id=~s, endpoint=~s", [State#state.node_id, NodeId, Endpoint]) end, NormalizedProviders), %% Route to appropriate handler case HasSubscriberQuery of true -> handle_subscriber_discovery_reply(NormalizedProviders, State); false -> continue_pending_query_with_providers(NormalizedProviders, State) end; {ok, {nodes, _Nodes}} -> %% Service/subscribers not found ?LOG_INFO("[~s] Received FIND_VALUE_REPLY with nodes (not found in DHT)", [State#state.node_id]), case HasSubscriberQuery of true -> handle_subscriber_discovery_reply([], State); false -> continue_pending_query_with_no_providers(State) end; {error, Reason} -> ?LOG_WARNING("[~s] Failed to decode FIND_VALUE_REPLY: ~p", [State#state.node_id, Reason]), case HasSubscriberQuery of true -> handle_subscriber_discovery_reply([], State); false -> continue_pending_query_with_no_providers(State) end end. %% @doc Continue pending query with discovered providers. -spec continue_pending_query_with_providers(list(), #state{}) -> #state{}. continue_pending_query_with_providers(Providers, State) -> io:format("[~s] *** continue_pending_query_with_providers called with ~p providers ***~n", [State#state.node_id, length(Providers)]), case maps:to_list(State#state.pending_queries) of [{ServiceKey, {From, Procedure, Args, Opts, Registry, Timer}} | _] -> io:format("[~s] Found pending query for procedure ~s~n", [State#state.node_id, Procedure]), %% Cancel timeout timer erlang:cancel_timer(Timer), %% Cache providers in service registry CacheTTL = 60, % Cache for 1 minute UpdatedRegistry = macula_service_registry:cache_service(Registry, Procedure, Providers, CacheTTL), State2 = State#state{ service_registry = UpdatedRegistry, pending_queries = maps:remove(ServiceKey, State#state.pending_queries) }, %% Continue with the RPC call io:format("[~s] About to call do_remote_call for ~s with ~p providers~n", [State#state.node_id, Procedure, length(Providers)]), case do_remote_call(Procedure, Args, Opts, From, Providers, State2) of {noreply, State3} -> io:format("[~s] do_remote_call returned {noreply, State}~n", [State#state.node_id]), State3; {reply, Reply, State3} -> io:format("[~s] do_remote_call returned {reply, ~p, State}~n", [State#state.node_id, Reply]), gen_server:reply(From, Reply), State3 end; [] -> %% No pending query - this is unexpected but not fatal ?LOG_WARNING("Received FIND_VALUE_REPLY with providers but no pending query"), State end. %% @doc Continue pending query with no providers (fall back to direct call). -spec continue_pending_query_with_no_providers(#state{}) -> #state{}. continue_pending_query_with_no_providers(State) -> case maps:to_list(State#state.pending_queries) of [{ServiceKey, {From, Procedure, Args, Opts, _Registry, Timer}} | _] -> %% Cancel timeout timer erlang:cancel_timer(Timer), %% Remove from pending queries State2 = State#state{pending_queries = maps:remove(ServiceKey, State#state.pending_queries)}, %% Fall back to direct call ?LOG_WARNING("Service ~s not found in DHT. Attempting direct call to connected endpoint.", [Procedure]), case do_direct_call(Procedure, Args, Opts, From, State2) of {noreply, State3} -> State3; {reply, Reply, State3} -> gen_server:reply(From, Reply), State3 end; [] -> %% No pending query - this is unexpected but not fatal ?LOG_WARNING("Received FIND_VALUE_REPLY with no providers and no pending query"), State end. %% @doc Handle subscriber discovery reply from DHT. -spec handle_subscriber_discovery_reply(list(), #state{}) -> #state{}. handle_subscriber_discovery_reply(Subscribers, State) -> %% Get the first pending subscriber query (FIFO - should only be one at a time) case maps:to_list(State#state.pending_subscriber_queries) of [{QueryId, {Topic, Payload, Qos, _Opts}} | _] -> ?LOG_DEBUG("[~s] Handling subscriber discovery reply for topic ~s (query_id: ~s, ~p subscribers)", [State#state.node_id, Topic, QueryId, length(Subscribers)]), %% Remove from pending queries PendingQueries = maps:remove(QueryId, State#state.pending_subscriber_queries), State2 = State#state{pending_subscriber_queries = PendingQueries}, %% Cache discovered subscribers (TTL: 60 seconds) CacheTTL = 60, Registry = State2#state.service_registry, Registry2 = macula_service_registry:cache_subscribers(Registry, Topic, Subscribers, CacheTTL), State3 = State2#state{service_registry = Registry2}, %% Route to discovered remote subscribers route_to_remote_subscribers(Topic, Payload, Qos, Subscribers, State3), State3; [] -> %% No pending subscriber query - this is unexpected but not fatal ?LOG_WARNING("[~s] Received subscriber discovery reply but no pending query", [State#state.node_id]), State end. %% @doc Route a published message to remote subscribers. -spec route_to_remote_subscribers(binary(), binary(), integer(), list(), #state{}) -> ok. route_to_remote_subscribers(_Topic, _Payload, _Qos, [], State) -> ?LOG_DEBUG("[~s] No remote subscribers found for topic", [State#state.node_id]), ok; route_to_remote_subscribers(Topic, Payload, Qos, Subscribers, State) -> ?LOG_INFO("[~s] Routing message to ~p remote subscriber(s) for topic ~s", [State#state.node_id, length(Subscribers), Topic]), %% Route message to each remote subscriber lists:foreach(fun(Subscriber) -> Endpoint = maps:get(endpoint, Subscriber, <<"unknown">>), NodeId = maps:get(node_id, Subscriber, <<"unknown">>), %% Send the message to this subscriber case send_to_remote_subscriber(Topic, Payload, Qos, Endpoint, NodeId, State) of ok -> ?LOG_INFO("[~s] Successfully sent message to remote subscriber ~s at ~s", [State#state.node_id, NodeId, Endpoint]); {error, Reason} -> ?LOG_WARNING("[~s] Failed to send message to remote subscriber ~s: ~p", [State#state.node_id, NodeId, Reason]) end end, Subscribers), ok. %% @doc Send a publish message to a single remote subscriber endpoint. -spec send_to_remote_subscriber(binary(), binary(), integer(), binary(), binary(), #state{}) -> ok | {error, term()}. send_to_remote_subscriber(Topic, Payload, Qos, Endpoint, NodeId, State) -> %% Get or create connection to subscriber endpoint using connection pool Pool = State#state.endpoint_connections, case macula_connection_pool:get_or_create_connection( Endpoint, State#state.node_id, State#state.realm, Pool ) of {ok, _Conn, Stream, _UpdatedPool} -> %% Note: Not updating state with pool - maintains existing behavior %% where connections weren't cached across calls (pre-existing issue) %% Generate message ID for this publish {MsgId, _State2} = next_message_id(State), %% Build publish message PublishMsg = #{ topic => Topic, payload => Payload, qos => Qos, message_id => MsgId, retain => false }, %% Send via the subscriber's stream case send_message_raw(publish, PublishMsg, Stream) of ok -> ?LOG_DEBUG("[~s] Sent PUBLISH to remote subscriber ~s (msg_id: ~p)", [State#state.node_id, NodeId, MsgId]), ok; {error, Reason} -> ?LOG_WARNING("[~s] Failed to send PUBLISH to ~s: ~p", [State#state.node_id, Endpoint, Reason]), {error, {send_failed, Reason}} end; {error, Reason, _Pool} -> ?LOG_WARNING("[~s] Failed to connect to remote subscriber ~s: ~p", [State#state.node_id, Endpoint, Reason]), {error, {connection_failed, Reason}} end. %% @doc Get next message ID. -spec next_message_id(#state{}) -> {binary(), #state{}}. next_message_id(State) -> Counter = State#state.msg_id_counter, {MsgId, NewCounter} = macula_utils:next_message_id(Counter), {MsgId, State#state{msg_id_counter = NewCounter}}. %% @doc Check if a published topic matches a subscription topic pattern. %% Supports configurable wildcards (defaults: dot-separated with * and **): %% - '*' matches a single segment (configurable via topic_wildcard_single) %% - '**' matches multiple segments (configurable via topic_wildcard_multi) %% - '.' as separator (configurable via topic_separator) %% - exact match otherwise -spec topic_matches(binary(), binary(), #state{}) -> boolean(). topic_matches(Pattern, Topic, State) -> Separator = State#state.topic_separator, WildcardSingle = State#state.topic_wildcard_single, WildcardMulti = State#state.topic_wildcard_multi, macula_utils:topic_matches(Pattern, Topic, Separator, WildcardSingle, WildcardMulti). %% @doc Send RPC reply for successful handler execution. -spec send_rpc_reply(binary(), term(), #state{}) -> ok | {error, term()}. send_rpc_reply(CallId, {ok, Result}, State) -> %% Success result EncodedResult = macula_utils:encode_json(Result), ReplyMsg = #{ call_id => CallId, result => EncodedResult }, send_message(reply, ReplyMsg, State); send_rpc_reply(CallId, {error, Reason}, State) -> %% Error result send_rpc_error(CallId, Reason, State). %% @doc Send RPC error reply. -spec send_rpc_error(binary(), term(), #state{}) -> ok | {error, term()}. send_rpc_error(CallId, Reason, State) -> ErrorMsg = #{ call_id => CallId, error => macula_utils:ensure_binary(io_lib:format("~p", [Reason])) }, send_message(error, ErrorMsg, State). %% @doc Make RPC call to a remote provider (with service discovery). %% Supports automatic failover to alternate providers on failure. -spec do_remote_call(binary(), term(), map(), term(), list(), #state{}) -> {noreply, #state{}} | {reply, {error, term()}, #state{}}. do_remote_call(Procedure, Args, Opts, From, Providers, State) -> io:format("[~s] do_remote_call ENTER~n", [State#state.node_id]), Result = do_remote_call_with_failover(Procedure, Args, Opts, From, Providers, [], 1, State), io:format("[~s] do_remote_call EXIT with ~p~n", [State#state.node_id, element(1, Result)]), Result. %% @doc Make RPC call with failover support. %% ExcludedProviders is a list of node_ids that have already been tried and failed. %% Attempt is the current attempt number (starts at 1). -spec do_remote_call_with_failover(binary(), term(), map(), term(), list(), list(), pos_integer(), #state{}) -> {noreply, #state{}} | {reply, {error, term()}, #state{}}. do_remote_call_with_failover(Procedure, Args, Opts, From, Providers, ExcludedProviders, Attempt, State) -> io:format("[~s] do_remote_call_with_failover ENTER: ~p providers, attempt ~p~n", [State#state.node_id, length(Providers), Attempt]), %% Get max attempts from options (default: 3 or number of providers, whichever is smaller) MaxAttempts = maps:get(max_attempts, Opts, min(3, length(Providers))), io:format("[~s] MaxAttempts: ~p~n", [State#state.node_id, MaxAttempts]), %% Filter out excluded providers AvailableProviders = case ExcludedProviders of [] -> Providers; _ -> lists:filter(fun(#{node_id := NodeId}) -> not lists:member(NodeId, ExcludedProviders) end, Providers) end, io:format("[~s] AvailableProviders: ~p~n", [State#state.node_id, length(AvailableProviders)]), %% Check if we have any providers left case AvailableProviders of [] -> ?LOG_ERROR("All providers exhausted for service ~s after ~p attempts", [Procedure, Attempt - 1]), {reply, {error, all_providers_failed}, State}; _ when Attempt > MaxAttempts -> ?LOG_ERROR("Max attempts (~p) reached for service ~s", [MaxAttempts, Procedure]), {reply, {error, max_attempts_exceeded}, State}; _ -> %% Use provider selection strategy to pick a provider SelectorState = State#state.provider_selector_state, Strategy = maps:get(strategy, SelectorState, random), %% Set the current service ID for round-robin tracking SelectorState2 = SelectorState#{current_service_id => Procedure}, case macula_provider_selector:select_provider(AvailableProviders, Strategy, SelectorState2) of {ok, Provider, SelectorState3} -> #{node_id := NodeId} = Provider, io:format("[~s] Provider selected: ~s~n", [State#state.node_id, binary:encode_hex(NodeId)]), AttemptLog = case Attempt of 1 -> ""; N -> io_lib:format(" (attempt ~p/~p)", [N, MaxAttempts]) end, ?LOG_INFO("Calling service ~s via DHT routing to provider ~s~s (strategy: ~p, ~p providers available)", [Procedure, binary:encode_hex(NodeId), AttemptLog, Strategy, length(AvailableProviders)]), %% Update state with new selector state State2 = State#state{provider_selector_state = SelectorState3}, io:format("[~s] About to call do_remote_call_to_provider~n", [State#state.node_id]), %% Attempt the call and store failover context case do_remote_call_to_provider(Procedure, Args, Opts, From, Provider, Providers, ExcludedProviders, Attempt, State2) of {noreply, State3} -> io:format("[~s] do_remote_call_to_provider returned {noreply, State}~n", [State#state.node_id]), {noreply, State3}; {reply, Error, State3} -> io:format("[~s] do_remote_call_to_provider returned {reply, ~p, State}~n", [State#state.node_id, Error]), {reply, Error, State3} end; {error, no_providers} -> ?LOG_ERROR("No providers available for service ~s", [Procedure]), {reply, {error, no_providers}, State} end end. %% @doc Make direct RPC call to currently connected endpoint. -spec do_direct_call(binary(), term(), map(), term(), #state{}) -> {noreply, #state{}} | {reply, {error, term()}, #state{}}. do_direct_call(Procedure, Args, Opts, From, State) -> %% Generate call ID {CallId, State2} = next_message_id(State), %% Encode args EncodedArgs = case Args of A when is_binary(A) -> A; A when is_map(A) -> macula_utils:encode_json(A); A when is_list(A) -> macula_utils:encode_json(A) end, %% Build call message (RPC call) - use CallId directly as binary CallMsg = #{ procedure => macula_utils:ensure_binary(Procedure), args => EncodedArgs, call_id => CallId }, case send_message(call, CallMsg, State2) of ok -> %% Set up timeout timer Timeout = maps:get(timeout, Opts, ?CALL_TIMEOUT), Timer = erlang:send_after(Timeout, self(), {call_timeout, CallId}), %% Store pending call (without failover context) PendingCalls = maps:put(CallId, {From, Timer}, State2#state.pending_calls), State3 = State2#state{pending_calls = PendingCalls}, {noreply, State3}; {error, Reason} -> {reply, {error, Reason}, State2} end. %% @doc Make RPC call to a specific provider endpoint with failover context. -spec do_remote_call_to_provider(binary(), term(), map(), term(), map(), list(), list(), pos_integer(), #state{}) -> {noreply, #state{}} | {reply, {error, term()}, #state{}}. do_remote_call_to_provider(Procedure, Args, Opts, From, Provider, AllProviders, ExcludedProviders, Attempt, State) -> io:format("[~s] do_remote_call_to_provider ENTER~n", [State#state.node_id]), %% Extract provider info #{node_id := ProviderNodeId} = Provider, io:format("[~s] Provider node_id: ~s~n", [State#state.node_id, binary:encode_hex(ProviderNodeId)]), %% ------------------------------------------------------------------- %% NEW: DHT-routed RPC using rpc_route message %% Instead of creating direct connection, route through mesh %% ------------------------------------------------------------------- %% Generate call ID {CallId, State2} = next_message_id(State), io:format("[~s] Generated CallId: ~p~n", [State#state.node_id, CallId]), %% Encode args EncodedArgs = case Args of A when is_binary(A) -> A; A when is_map(A) -> macula_utils:encode_json(A); A when is_list(A) -> macula_utils:encode_json(A) end, %% Build call message CallMsg = #{ procedure => macula_utils:ensure_binary(Procedure), args => EncodedArgs, call_id => CallId }, %% Wrap in rpc_route envelope for DHT routing LocalNodeId = State2#state.node_id, MaxHops = 10, io:format("[~s] About to wrap in rpc_route envelope~n", [State#state.node_id]), RpcRouteMsg = macula_rpc_routing:wrap_call(LocalNodeId, ProviderNodeId, CallMsg, MaxHops), io:format("[~s] RPC route envelope created~n", [State#state.node_id]), ?LOG_INFO("Sending routed RPC call to provider ~p (call_id: ~p, max_hops: ~p)", [binary:encode_hex(ProviderNodeId), CallId, MaxHops]), io:format("[~s] About to send_message_raw rpc_route~n", [State#state.node_id]), %% Send via main mesh stream (not direct endpoint connection) case send_message_raw(rpc_route, RpcRouteMsg, State2#state.stream) of ok -> io:format("[~s] send_message_raw returned ok~n", [State#state.node_id]), ?LOG_INFO("Sent routed RPC call (call_id: ~p)", [CallId]), %% Set up timeout timer Timeout = maps:get(timeout, Opts, ?CALL_TIMEOUT), Timer = erlang:send_after(Timeout, self(), {call_timeout, CallId}), %% Store pending call WITH failover context FailoverContext = #{ procedure => Procedure, args => Args, opts => Opts, all_providers => AllProviders, excluded_providers => [ProviderNodeId | ExcludedProviders], attempt => Attempt, provider_node_id => ProviderNodeId % Store for reply routing }, PendingCalls = maps:put(CallId, {From, Timer, FailoverContext}, State2#state.pending_calls), State3 = State2#state{pending_calls = PendingCalls}, io:format("[~s] do_remote_call_to_provider EXIT with {noreply, State}~n", [State#state.node_id]), {noreply, State3}; {error, Reason} -> io:format("[~s] send_message_raw returned error: ~p~n", [State#state.node_id, Reason]), ?LOG_WARNING("Failed to send routed RPC call: ~p", [Reason]), {reply, {error, {send_failed, Reason}}, State2} end. %%%=================================================================== %%% Multi-Endpoint Connection Management %%%=================================================================== %% Note: Endpoint connection management now handled by macula_connection_pool module %%%=================================================================== %%% DHT-Based Subscriber Discovery %%%=================================================================== %% @doc Query DHT for remote subscribers to a topic (exact match only) %% Note: This implementation supports exact-match topics only. Wildcard subscription %% discovery across the mesh is deferred to a future enhancement. -spec discover_remote_subscribers(binary(), binary(), integer(), map(), #state{}) -> #state{}. discover_remote_subscribers(Topic, Payload, Qos, _Opts, State) -> NodeId = State#state.node_id, Registry = State#state.service_registry, %% Check subscriber cache first (pattern matching for cache hit/miss) case macula_service_registry:discover_subscribers(Registry, Topic) of {ok, Subscribers, Registry2} -> %% Cache hit - route to cached subscribers immediately ?LOG_DEBUG("[~s] Cache hit for subscribers to topic: ~s (~p subscribers)", [NodeId, Topic, length(Subscribers)]), route_to_remote_subscribers(Topic, Payload, Qos, Subscribers, State), State#state{service_registry = Registry2}; {cache_miss, Registry2} -> %% Cache miss - query DHT asynchronously State2 = State#state{service_registry = Registry2}, query_dht_for_subscribers(Topic, Payload, Qos, State2) end. %% @doc Query DHT for subscribers (extracted for clarity). -spec query_dht_for_subscribers(binary(), binary(), integer(), #state{}) -> #state{}. query_dht_for_subscribers(Topic, Payload, Qos, State) -> NodeId = State#state.node_id, %% Create DHT key from topic (same as subscription advertisement) TopicKey = crypto:hash(sha256, Topic), %% Generate message ID for tracking the query MsgId = next_message_id(State), State2 = State#state{msg_id_counter = MsgId}, ?LOG_DEBUG("[~s] Querying DHT for remote subscribers to topic: ~s (MsgId: ~s)", [NodeId, Topic, MsgId]), %% Send FIND_VALUE query to DHT asynchronously %% The response will come back via handle_info({quic, Data, ...}) spawn(fun() -> try FindValueMsg = macula_routing_protocol:encode_find_value(TopicKey), case send_message(find_value, FindValueMsg, State2) of ok -> ?LOG_DEBUG("[~s] Sent FIND_VALUE for topic ~s", [NodeId, Topic]); {error, SendError} -> ?LOG_WARNING("[~s] Failed to send FIND_VALUE for topic ~s: ~p", [NodeId, Topic, SendError]) end catch _:QueryError:Stack -> ?LOG_WARNING("[~s] Failed to query DHT for topic ~s: ~p~nStack: ~p", [NodeId, Topic, QueryError, Stack]) end end), %% Track the pending query with topic/payload/qos so we can route when response arrives PendingQueries = State2#state.pending_subscriber_queries, PendingQueries2 = PendingQueries#{MsgId => {Topic, Payload, Qos, #{}}}, State2#state{pending_subscriber_queries = PendingQueries2}. %%%=================================================================== %%% DHT Subscription Advertisement %%%=================================================================== %% @doc Advertise a topic subscription in the DHT for mesh-wide pub/sub routing -spec advertise_subscription_in_dht(binary(), reference(), #state{}) -> #state{}. advertise_subscription_in_dht(Topic, SubRef, State) -> %% Default TTL for subscriptions is 300 seconds (5 minutes) TTL = 300, %% Create DHT key by hashing the topic (same as service registry pattern) TopicKey = crypto:hash(sha256, Topic), %% Create subscriber value with endpoint info for routing messages back SubscriberValue = #{ node_id => State#state.node_id, endpoint => State#state.url, ttl => TTL }, %% Send STORE message to DHT ?LOG_INFO("[~s] Advertising subscription to topic ~s in DHT", [State#state.node_id, Topic]), try StoreMsg = macula_routing_protocol:encode_store(TopicKey, SubscriberValue), case send_message(store, StoreMsg, State) of ok -> ?LOG_DEBUG("[~s] Successfully stored subscription for ~s in DHT", [State#state.node_id, Topic]); {error, SendError} -> ?LOG_WARNING("[~s] Failed to send STORE for subscription ~s: ~p", [State#state.node_id, Topic, SendError]) end catch _:DhtError -> ?LOG_WARNING("[~s] Failed to advertise subscription ~s in DHT: ~p (continuing)", [State#state.node_id, Topic, DhtError]) end, %% Schedule re-advertisement (TTL - 60 seconds, minimum 10 seconds) ResubInterval = max(10, TTL - 60) * 1000, % milliseconds TimerRef = erlang:send_after(ResubInterval, self(), {resubscribe, Topic}), ?LOG_DEBUG("[~s] Scheduled re-subscription for ~s in ~p seconds", [State#state.node_id, Topic, ResubInterval div 1000]), %% Store subscription info for re-advertisement AdvertisedSubscriptions = State#state.advertised_subscriptions, SubInfo = #{ sub_ref => SubRef, ttl => TTL, timer_ref => TimerRef }, AdvertisedSubscriptions2 = AdvertisedSubscriptions#{Topic => SubInfo}, State#state{advertised_subscriptions = AdvertisedSubscriptions2}. %% @doc Parse server endpoint URL to {IP, Port} tuple. %% Uses idiomatic Erlang pattern matching on function heads. -spec parse_server_endpoint(binary()) -> {{byte(), byte(), byte(), byte()}, inet:port_number()}. parse_server_endpoint(Endpoint) when is_binary(Endpoint) -> try case uri_string:parse(Endpoint) of #{host := Host, port := Port} when is_integer(Port) -> HostStr = binary_to_list(Host), case inet:getaddr(HostStr, inet) of {ok, IPTuple} -> {IPTuple, Port}; {error, _Reason} -> %% Fallback to placeholder if DNS fails {{0,0,0,0}, Port} end; #{host := Host} -> %% No port specified, use default 9443 HostStr = binary_to_list(Host), case inet:getaddr(HostStr, inet) of {ok, IPTuple} -> {IPTuple, 9443}; {error, _Reason} -> {{0,0,0,0}, 9443} end; _ -> %% Invalid URL format {{0,0,0,0}, 0} end catch _:_ -> %% Parse error, return placeholder {{0,0,0,0}, 0} end; parse_server_endpoint(_Other) -> {{0,0,0,0}, 0}.