macula

%%%-------------------------------------------------------------------
%%% @doc Direct-dial resolve-and-call: shared internals for
%%% `macula_request'/`macula_response', `macula_streamer'/
%%% `macula_stream_sink', and `macula_feeder'/`macula_download'.
%%%
%%% Not a public API on its own — `macula_request:start_link_direct/6,7,8',
%%% `macula_response:advertise_direct/6,7',
%%% `macula_stream_sink:start_link_direct/5,6',
%%% `macula_streamer:advertise_direct/6,7',
%%% `macula_download:start_link_direct/4,5', and
%%% `macula_feeder:start_link_direct/5,6' are the entry points.
%%% Factored out because RPC, streaming, and content-download all need
%%% the same shape of resolve sequence (`find_records' -> verify -> read
%%% the record -> build a `quic://' dial URL, retrying past DHT
%%% propagation lag throughout). Streaming and RPC share the IDENTICAL
%%% discovery mechanism — a `procedure_advertisement' does not
%%% distinguish RPC from streaming, only the eventual dial
%%% (`call_station/7' vs `call_stream_station/6') does — so
%%% `publish_advertisement/4,5' is reused as-is by both providers, and
%%% `call/6'/`call_stream/6' share the same `resolve_dial_url/4' and
%%% "Trust model" below. Content has no publish step here at all — see
%%% "Content" further down.
%%%
%%% The resolve side retries: a record just published on the provider's
%%% station has not necessarily replicated to the caller's station yet,
%%% and treating the first miss as failure would make every direct-dial
%%% call racy against DHT propagation lag. `macula_station_cert_chain_SUITE'
%%% (macula-station) proved this exact resolve+dial sequence works
%%% cross-station against the live fleet; this module is that sequence,
%%% lifted out of the test and made reusable.
%%%
%%% == Trust model ==
%%%
%%% Two independent checks, both mandatory, cover what the QUIC/TLS
%%% layer cannot: (1) every candidate `procedure_advertisement' must
%%% carry a valid Ed25519 signature before its `serving_station' is
%%% trusted at all — otherwise any identity able to sign SOME record
%%% could name a real, legitimate station as the server for a
%%% procedure it has no authority over, and the station_endpoint check
%%% below would still pass (it only proves we reached the station we
%%% were told to reach, not that whoever told us so was authorized to);
%%% (2) the resolved `station_endpoint' must be signed by the station
%%% itself (`verify_and_build/2', unchanged). The actual QUIC dial
%%% trusts NEITHER the TLS certificate (`pin_tls_cert => false' — a
%%% production station's TLS is terminated by an unrelated PKI, e.g.
%%% Let's Encrypt, so pinning the cert key can never succeed) NOR
%%% nothing (`verify => none' alone would trust whoever answers): trust
%%% is enforced at the application layer instead, via the
%%% cryptographically signed CONNECT/HELLO handshake
%%% (`macula_peering_conn:bind_peer_identity/2') checked against the
%%% exact pubkey the signed DHT chain above resolved.
%%%
%%% A third check is available but OPT-IN, via `Opts' (managed realms
%%% only — see `macula_record:verify_advertisement_cert_chain/3'):
%%% `verify_cert_chain => {RealmCaPem, Org}' additionally requires the
%%% advertisement's embedded X.509 service-cert chain to verify to
%%% `RealmCaPem' under `Org' (Slice 7c Direction B), proving the
%%% ADVERTISER — not just the station it names — is an org/realm-
%%% authorized identity. Without it, (1) above still rejects an
%%% unsigned or badly-signed advertisement, but not one signed by an
%%% unauthorized (if self-consistent) identity; unmanaged realms have
%%% no realm CA to check against, so this stays opt-in rather than
%%% mandatory.
%%%
%%% == Content ==
%%%
%%% `put_content/4' has no resolve step at all — unlike a GET, a PUT
%%% names its OWN target: the caller already knows (or is choosing)
%%% which station to seed, so it takes `Station' directly and resolves
%%% only that station's own `station_endpoint' (`resolve_station_endpoint/2',
%%% the same machinery `call/6' uses internally for `serving_station').
%%%
%%% `get_content/3' resolves and fetches deliberately WITHOUT the
%%% cert-chain machinery above — content's threat model genuinely
%%% differs from RPC's. An RPC reply is opaque and unverifiable except
%%% by trusting whoever answered, so proving the ADVERTISER is
%%% authorized matters. Content is content-addressed: the fetched bytes
%%% are independently re-hashed against the MCID client-side
%%% (`macula:verify_block_hash/2' for single-block,
%%% `macula_manifest:verify/2' for chunked) regardless of which peer
%%% served them, so a rogue or unauthorized announcer can at most
%%% refuse to serve or waste a dial — it cannot make a caller accept
%%% content that does not hash to the MCID it asked for. What still
%%% matters, and is still mandatory, is (1)'s analogue for
%%% `content_announcement': the signer must equal the `announcer_node'
%%% it claims (`macula:decode_provider/1'), so an attacker cannot at
%%% least misattribute who is claiming to serve what.
%%% @end
%%%-------------------------------------------------------------------
-module(macula_direct_dial).

-export([call/5, call/6, call_stream/5, call_stream/6,
        publish_advertisement/4, publish_advertisement/5,
        get_content/3, resolve_content_provider/2,
        put_content/4, resolve_station_endpoint/2]).

-ifdef(TEST).
%% Exports for unit tests — pure helpers that are otherwise private.
-export([advertisement_trusted/2]).
-endif.

-define(RESOLVE_RETRIES, 50).
-define(RESOLVE_RETRY_MS, 100).

%% @doc As `call/6' with no cert-chain verification.
-spec call(macula:pool(), macula:realm(), macula:procedure(), term(),
          pos_integer()) -> {ok, term()} | {error, term()}.
call(Pool, Realm, Procedure, Payload, TimeoutMs) ->
    call(Pool, Realm, Procedure, Payload, TimeoutMs, #{}).

%% @doc Resolve `Procedure''s provider and call it there directly. Same
%% return shape as `macula:call/5'; resolve failures surface as
%% `{error, {unresolved, Reason}}' so a caller can tell "nobody has
%% advertised this via direct-dial yet" apart from a real call failure.
%% `Opts' may include `verify_cert_chain => {RealmCaPem, Org}' — see
%% the module doc's "Trust model" section.
-spec call(macula:pool(), macula:realm(), macula:procedure(), term(),
          pos_integer(), map()) -> {ok, term()} | {error, term()}.
call(Pool, Realm, Procedure, Payload, TimeoutMs, Opts) ->
    case resolve_dial_url(Pool, Realm, Procedure, Opts) of
        {ok, {Station, DialUrl}} ->
            %% See the module doc's "Trust model" section: trust is
            %% pinned to the exact pubkey the signed DHT chain resolved,
            %% but enforced at the application layer, not the TLS
            %% layer — a production station's TLS certificate has no
            %% relationship to its macula identity.
            macula:call_station(Pool, DialUrl, Realm, Procedure, Payload,
                                TimeoutMs, #{expected_node_id => Station,
                                             pin_tls_cert => false,
                                             verify => none});
        {error, Reason} ->
            {error, {unresolved, Reason}}
    end.

%% @doc As `call_stream/6' with no cert-chain verification.
-spec call_stream(macula:pool(), macula:realm(), macula:procedure(), term(),
                  map()) -> {ok, macula:stream()} | {error, term()}.
call_stream(Pool, Realm, Procedure, Args, StreamOpts) ->
    call_stream(Pool, Realm, Procedure, Args, StreamOpts, #{}).

%% @doc As `call/6', but opens a stream (`macula:call_stream_station/6''s
%% shape) instead of making a single-reply call, built on the exact
%% same resolve+trust machinery — see the module doc. `StreamOpts' is
%% forwarded to `call_stream_station/6' alongside the resolved trust
%% override (`mode', `owner', `dial_timeout_ms', etc); `Opts' is the
%% resolve-side `verify_cert_chain' opt, same as `call/6'.
-spec call_stream(macula:pool(), macula:realm(), macula:procedure(), term(),
                  map(), map()) -> {ok, macula:stream()} | {error, term()}.
call_stream(Pool, Realm, Procedure, Args, StreamOpts, Opts) ->
    case resolve_dial_url(Pool, Realm, Procedure, Opts) of
        {ok, {Station, DialUrl}} ->
            macula:call_stream_station(Pool, DialUrl, Realm, Procedure, Args,
                                       StreamOpts#{expected_node_id => Station,
                                                   pin_tls_cert => false,
                                                   verify => none});
        {error, Reason} ->
            {error, {unresolved, Reason}}
    end.

%% @doc As `publish_advertisement/5' with no cert chain embedded.
-spec publish_advertisement(macula:pool(), macula:realm(), macula:procedure(),
                            macula_identity:key_pair()) -> ok | {error, term()}.
publish_advertisement(Pool, Realm, Procedure, Identity) ->
    publish_advertisement(Pool, Realm, Procedure, Identity, #{}).

%% @doc Publish a signed `procedure_advertisement' for `Procedure',
%% naming `Pool''s currently-connected station as the serving station.
%% `Identity' signs it — the pool itself has no identity to sign with,
%% so the caller supplies one (reuse the same one across re-advertises;
%% a fresh identity per call is valid but needless DHT churn). `Opts'
%% may include `cert_chain => ChainPem' (leaf ++ org CA, PEM) so a
%% verifying consumer's `verify_cert_chain' opt (see `call/6') can
%% check this advertiser's org/realm authorization — Slice 7c
%% Direction B, managed realms only.
-spec publish_advertisement(macula:pool(), macula:realm(), macula:procedure(),
                            macula_identity:key_pair(), map()) ->
    ok | {error, term()}.
publish_advertisement(Pool, Realm, Procedure, Identity, Opts) ->
    case macula:links(Pool) of
        {ok, Links} -> on_links(connected_station(Links), Pool, Realm,
                                Procedure, Identity, Opts);
        {error, _} = Error -> Error
    end.

on_links({ok, Station}, Pool, Realm, Procedure, Identity, Opts) ->
    AdvPub = macula_identity:public(Identity),
    Uri = discovery_uri(Realm, Procedure),
    Ad = macula_record:sign(
           macula_record:procedure_advertisement(AdvPub, Uri, Station,
                                                 adv_opts(Opts)),
           Identity),
    macula:put_record(Pool, Ad);
on_links({error, _} = Error, _Pool, _Realm, _Procedure, _Identity, _Opts) ->
    Error.

adv_opts(#{cert_chain := ChainPem}) when is_binary(ChainPem) ->
    #{cert_chain => ChainPem};
adv_opts(_Opts) ->
    #{}.

%% The first CONNECTED link with a known peer pubkey — `links/1' can
%% include configured-but-not-yet-spawned or dead entries, and taking
%% one of those blindly would publish an advertisement pointing at a
%% station this pool cannot currently prove it can reach.
connected_station(Links) ->
    case [S || #{connected := true, node_id := S} <- Links, is_binary(S)] of
        [Station | _] -> {ok, Station};
        [] -> {error, no_healthy_link}
    end.

%% @doc Resolve `MCID''s provider via its signed `content_announcement'
%% and fetch it there directly, retrying past DHT propagation lag the
%% same way `call/6' does for procedures. Same return shape as
%% `macula:get_content/2'; resolve failures surface as
%% `{error, {unresolved, Reason}}'. `TimeoutMs' bounds only the QUIC
%% handshake if a fresh link must be dialed
%% (`macula:get_content_station/5') — the underlying block/manifest
%% transfer has its own internal timeouts. See the module doc's
%% "Content" section for why this has no `verify_cert_chain'-equivalent
%% opt, unlike `call/6'. Only chunked content is discoverable this way
%% — see `macula:find_content_providers/2'.
-spec get_content(macula:pool(), macula:mcid(), pos_integer()) ->
    {ok, binary()} | {error, term()}.
get_content(Pool, MCID, TimeoutMs) ->
    case resolve_content_provider(Pool, MCID) of
        {ok, #{announcer_node := Node, endpoint := Endpoint}} ->
            macula:get_content_station(Pool, Endpoint, MCID, TimeoutMs,
                                       #{expected_node_id => Node,
                                         pin_tls_cert => false,
                                         verify => none});
        {error, Reason} ->
            {error, {unresolved, Reason}}
    end.

%% @doc Resolve `MCID''s provider via a signed `content_announcement',
%% retrying past a not-yet-replicated announcement. Returns the first
%% candidate `macula:find_content_providers/2' finds — that function
%% already discards unsigned or signer-mismatched announcements before
%% this ever sees them.
-spec resolve_content_provider(macula:pool(), macula:mcid()) ->
    {ok, map()} | {error, term()}.
resolve_content_provider(Pool, MCID) ->
    resolve_content_provider(Pool, MCID, ?RESOLVE_RETRIES).

resolve_content_provider(_Pool, _MCID, 0) ->
    {error, content_not_announced};
resolve_content_provider(Pool, MCID, N) ->
    on_providers_found(macula:find_content_providers(Pool, MCID), Pool, MCID, N).

on_providers_found({ok, [Provider | _]}, _Pool, _MCID, _N) ->
    {ok, Provider};
on_providers_found({ok, []}, Pool, MCID, N) ->
    timer:sleep(?RESOLVE_RETRY_MS),
    resolve_content_provider(Pool, MCID, N - 1);
on_providers_found({error, _} = Error, _Pool, _MCID, _N) ->
    Error.

%% @doc Resolve `Station''s dialable `quic://' URL from its own signed
%% `station_endpoint' record and put `Bytes' there directly. Same
%% return shape as `macula:put_content/2'; resolve failures surface as
%% `{error, {unresolved, Reason}}'. `TimeoutMs' bounds only the QUIC
%% handshake if a fresh link must be dialed
%% (`macula:put_content_station/5') — the underlying block/manifest
%% transfer has its own internal timeouts.
-spec put_content(macula:pool(), macula_identity:pubkey(), binary(),
                  pos_integer()) -> {ok, macula:mcid()} | {error, term()}.
put_content(Pool, Station, Bytes, TimeoutMs) ->
    case resolve_station_endpoint(Pool, Station) of
        {ok, DialUrl} ->
            macula:put_content_station(Pool, DialUrl, Bytes, TimeoutMs,
                                       #{expected_node_id => Station,
                                         pin_tls_cert => false,
                                         verify => none});
        {error, Reason} ->
            {error, {unresolved, Reason}}
    end.

%% @doc Resolve `Station''s dialable `quic://' URL from its own signed
%% `station_endpoint' record, verifying the record's signer is exactly
%% `Station' and retrying past a stale/expired replica — the same
%% discipline `call/6' applies internally once it has resolved a
%% procedure's `serving_station'.
-spec resolve_station_endpoint(macula:pool(), macula_identity:pubkey()) ->
    {ok, binary()} | {error, term()}.
resolve_station_endpoint(Pool, Station) ->
    case resolve_endpoint(Pool, Station) of
        {ok, {Station, DialUrl}} -> {ok, DialUrl};
        {error, _} = Error -> Error
    end.

%%%===================================================================
%%% Internal
%%%===================================================================

resolve_dial_url(Pool, Realm, Procedure, Opts) ->
    Uri = discovery_uri(Realm, Procedure),
    Key = macula_record:procedure_key(Uri),
    case find_records_retry(Pool, Key, ?RESOLVE_RETRIES) of
        {ok, [_ | _] = Recs} ->
            resolve_station(Pool, trusted_advertisements(Recs, Opts));
        {ok, []} -> {error, procedure_not_advertised};
        {error, _} = Error -> Error
    end.

%% Only a record that passes trust filtering is a candidate at all —
%% see the module doc's "Trust model" section. Base signature check is
%% mandatory; `verify_cert_chain' additionally requires the embedded
%% X.509 chain when the caller opted in.
trusted_advertisements(Recs, Opts) ->
    [Rec || Rec <- Recs, advertisement_trusted(Rec, Opts)].

advertisement_trusted(Rec, #{verify_cert_chain := {RealmCaPem, Org}}) ->
    ok =:= macula_record:verify_advertisement_cert_chain(RealmCaPem, Rec, Org);
advertisement_trusted(Rec, _Opts) ->
    signature_ok(macula_record:verify(Rec)).

signature_ok({ok, _})    -> true;
signature_ok({error, _}) -> false.

resolve_station(_Pool, []) ->
    {error, no_trusted_advertisement};
resolve_station(Pool, [Rec | Rest]) ->
    case macula_record:read_procedure_advertisement(Rec) of
        #{serving_station := Station} ->
            resolve_endpoint(Pool, Station);
        _ ->
            resolve_station(Pool, Rest)
    end.

%% Retries past a resolved-but-stale record, not just an absent one:
%% the DHT can hand back a replica that hasn't been evicted or
%% refreshed yet even though the station's own current publish is
%% live, and giving up on the first stale hit would make an otherwise
%% healthy station unreachable via direct-dial until that one replica
%% happens to age out on its own.
resolve_endpoint(Pool, Station) ->
    resolve_endpoint(Pool, Station, ?RESOLVE_RETRIES).

resolve_endpoint(_Pool, _Station, 0) ->
    {error, station_endpoint_not_found};
resolve_endpoint(Pool, Station, N) ->
    Key = macula_record:station_endpoint_key(Station),
    on_endpoint_fetch(find_record_retry(Pool, Key, 1), Pool, Station, N).

on_endpoint_fetch({ok, EpRec}, Pool, Station, N) ->
    on_endpoint_verified(verify_and_build(Station, EpRec), Pool, Station, N);
on_endpoint_fetch({error, not_found}, Pool, Station, N) ->
    timer:sleep(?RESOLVE_RETRY_MS),
    resolve_endpoint(Pool, Station, N - 1).

on_endpoint_verified({ok, _} = Ok, _Pool, _Station, _N) -> Ok;
on_endpoint_verified({error, expired}, Pool, Station, N) ->
    timer:sleep(?RESOLVE_RETRY_MS),
    resolve_endpoint(Pool, Station, N - 1);
on_endpoint_verified({error, _} = Error, _Pool, _Station, _N) -> Error.

%% The `station_endpoint' record for `Station' must be SIGNED BY
%% `Station' itself (macula_station_announcer publishes it self-signed
%% — the station describing its own reachable address). Checking the
%% signature AND that the signer is exactly `Station', not just any
%% valid signature, is what makes pinning `expected_node_id => Station'
%% on the dial meaningful: without it, a record merely stored under the
%% right DHT key (but signed, or not, by someone else) would still be
%% trusted, and per-call pinning would authenticate the wrong thing.
verify_and_build(Station, #{key := Station} = EpRec) ->
    case macula_record:verify(EpRec) of
        {ok, Verified} -> build_dial_url(Station, Verified);
        {error, _} = Error -> Error
    end;
verify_and_build(_Station, _EpRec) ->
    {error, station_endpoint_signer_mismatch}.

build_dial_url(Station, EpRec) ->
    case macula_record:read_station_endpoint(EpRec) of
        #{quic_port := Port, host_advertised := [Host | _]} ->
            {ok, {Station, <<"quic://[", Host/binary, "]:",
                            (integer_to_binary(Port))/binary>>}};
        _ ->
            {error, malformed_station_endpoint}
    end.

find_records_retry(_Pool, _Key, 0) -> {ok, []};
find_records_retry(Pool, Key, N) ->
    on_find_records(macula:find_records(Pool, Key), Pool, Key, N).

on_find_records({ok, [_ | _] = Recs}, _Pool, _Key, _N) -> {ok, Recs};
on_find_records(_Other, Pool, Key, N) ->
    timer:sleep(?RESOLVE_RETRY_MS),
    find_records_retry(Pool, Key, N - 1).

find_record_retry(_Pool, _Key, 0) -> {error, not_found};
find_record_retry(Pool, Key, N) ->
    on_find_record(macula:find_record(Pool, Key), Pool, Key, N).

on_find_record({ok, Rec}, _Pool, _Key, _N) -> {ok, Rec};
on_find_record(_Other, Pool, Key, N) ->
    timer:sleep(?RESOLVE_RETRY_MS),
    find_record_retry(Pool, Key, N - 1).

%% No `Org' segment in the discovery URI: `Org' is only consulted
%% post-resolve, as the `verify_cert_chain' opt's expected leaf-cert
%% organization (see `advertisement_trusted/2') — it does not affect
%% how a `procedure_advertisement' is keyed or found.
discovery_uri(Realm, Procedure) ->
    <<(binary:encode_hex(Realm))/binary, "/", Procedure/binary>>.