Packages
macula
13.0.0
13.0.1
13.0.0
12.12.0
12.11.1
12.11.0
12.10.0
12.9.1
12.9.0
12.8.0
12.7.0
12.6.0
12.5.1
12.5.0
12.4.0
12.3.0
12.2.1
12.2.0
12.1.0
12.0.0
11.5.0
11.4.0
11.3.1
11.3.0
11.2.0
11.1.0
11.0.0
10.25.0
10.24.0
10.23.0
10.22.0
10.21.0
10.20.3
10.20.2
10.20.0
10.19.2
10.19.1
10.19.0
10.18.0
10.17.0
10.16.0
10.15.0
10.14.5
10.14.4
10.14.2
10.14.1
10.14.0
10.13.2
10.13.1
10.11.0
10.10.2
10.10.1
10.10.0
10.9.1
10.9.0
10.8.0
10.7.0
10.5.8
10.5.7
10.5.6
10.5.5
10.5.4
10.5.3
10.5.2
10.5.1
10.5.0
10.4.0
10.2.0
10.1.1
10.1.0
10.0.2
10.0.1
10.0.0
9.13.8
9.8.2
9.8.1
9.8.0
9.5.0
9.4.0
9.3.1
9.3.0
9.2.0
9.1.1
9.1.0
9.0.0
8.7.0
8.6.0
8.5.0
8.4.1
8.4.0
8.3.0
8.2.0
8.1.0
8.0.2
8.0.1
8.0.0
7.1.0
7.0.0
6.0.0
5.2.2
5.2.1
5.2.0
5.1.0
5.0.0
4.8.0
4.7.1
4.7.0
4.6.0
4.5.0
4.4.10
4.4.9
4.4.8
4.4.7
4.4.6
4.4.5
4.4.4
4.4.3
4.4.2
4.4.1
4.4.0
4.3.1
4.3.0
4.2.9
4.2.8
4.2.7
4.2.6
4.2.5
4.2.4
4.2.3
4.2.2
4.2.1
4.2.0
4.1.1
4.1.0
4.0.0
3.16.0
3.15.3
3.15.2
3.15.1
3.14.0
3.13.0
3.12.1
3.12.0
3.11.1
3.11.0
3.10.3
3.10.2
3.10.1
3.9.0
3.8.0
3.7.0
3.5.0
3.4.0
3.3.0
3.2.0
3.1.0
3.0.0
2.1.1
2.1.0
2.0.0
1.5.2
1.5.1
1.4.30
1.4.29
1.4.28
1.4.27
1.4.26
1.4.25
1.4.24
1.4.23
1.4.22
1.4.21
1.4.20
1.4.19
1.4.18
1.4.17
1.4.16
1.4.15
1.4.14
1.4.13
1.4.11
1.4.10
1.4.9
1.4.8
1.4.7
1.4.6
1.4.5
1.4.4
1.4.3
1.4.2
1.4.1
1.4.0
1.3.1
1.3.0
1.2.0
1.1.0
1.0.10
1.0.9
1.0.8
1.0.7
1.0.6
1.0.5
1.0.4
1.0.3
1.0.2
1.0.1
1.0.0
0.48.6
0.48.5
0.48.4
0.48.3
0.48.2
0.48.1
0.48.0
0.47.1
0.47.0
0.46.3
0.46.1
0.46.0
0.45.3
0.45.2
0.45.1
0.45.0
0.44.2
0.44.1
0.44.0
0.43.3
0.43.2
0.43.1
0.43.0
0.42.9
0.42.8
0.42.7
0.42.6
0.42.5
0.42.4
0.42.3
0.42.2
0.42.1
0.42.0
0.41.1
0.41.0
0.40.1
0.40.0
0.39.9
0.39.8
0.39.7
0.39.6
0.39.5
0.39.4
0.39.3
0.39.2
0.39.1
0.39.0
0.38.8
0.38.7
0.38.6
0.38.5
0.38.4
0.38.3
0.38.2
0.38.1
0.38.0
0.37.7
0.37.6
0.37.5
0.37.4
0.37.3
0.37.2
0.37.1
0.37.0
0.36.6
0.36.5
0.36.4
0.36.3
0.36.2
0.36.1
0.36.0
0.35.4
0.35.3
0.35.2
0.35.1
0.35.0
0.34.1
0.34.0
0.33.1
0.33.0
0.32.5
0.32.4
0.32.3
0.32.2
0.32.1
0.32.0
0.31.9
0.31.8
0.31.7
0.31.6
0.31.5
0.31.4
0.31.3
0.31.2
0.31.1
0.31.0
0.30.10
0.30.9
0.30.8
0.30.7
0.30.6
0.30.5
0.30.4
0.30.3
0.30.2
0.30.1
0.30.0
0.29.0
0.28.3
0.28.2
0.28.1
0.28.0
0.27.1
0.27.0
0.26.1
0.26.0
0.25.6
0.25.5
0.25.4
0.25.3
0.25.2
0.25.1
0.25.0
0.24.6
0.24.5
0.24.4
0.24.3
0.24.2
0.24.1
0.24.0
0.23.3
0.23.2
0.23.1
0.23.0
0.22.12
0.22.11
0.22.10
0.22.9
0.22.8
0.22.7
0.22.6
0.22.5
0.22.4
0.22.3
0.22.2
0.22.1
0.22.0
0.21.7
0.21.6
0.21.5
0.21.4
0.21.2
0.21.1
0.21.0
0.20.25
0.20.24
0.20.23
0.20.22
0.20.21
0.20.20
0.20.19
0.20.18
0.20.17
0.20.16
0.20.15
0.20.14
0.20.13
0.20.12
0.20.11
0.20.10
0.20.9
0.20.8
0.20.7
0.20.6
0.20.5
0.20.3
0.20.2
0.20.1
0.20.0
0.19.2
0.19.1
0.19.0
0.18.1
0.18.0
0.17.4
0.17.3
0.17.2
0.17.1
0.17.0
0.16.6
0.16.5
0.16.4
0.16.3
0.16.2
0.16.1
0.16.0
0.15.1
0.15.0
0.14.3
0.14.2
0.14.1
0.14.0
0.12.6
0.12.5
0.12.3
0.11.3
0.10.2
0.10.1
0.10.0
0.9.2
0.9.1
0.9.0
0.8.25
0.8.24
0.8.23
0.8.22
0.8.21
0.8.20
0.8.19
0.8.18
0.8.17
0.8.16
0.8.15
0.8.14
0.8.13
0.8.12
0.8.11
0.8.10
0.8.9
0.8.8
0.8.7
0.8.6
0.8.5
0.8.4
0.8.3
0.8.2
0.8.1
0.8.0
0.7.30
0.7.29
0.7.28
0.7.27
0.7.26
0.7.25
0.7.24
0.7.23
0.7.22
0.7.21
0.7.20
0.7.19
0.7.18
0.7.17
0.7.16
0.7.15
0.7.14
0.7.13
0.7.12
0.7.11
0.7.10
0.7.9
0.7.8
0.7.7
0.7.6
0.7.5
0.7.4
0.7.3
0.7.2
0.7.1
0.7.0
0.6.7
0.6.6
0.6.5
0.6.4
0.6.3
0.6.2
0.6.1
0.6.0
0.5.0
0.4.4
0.4.3
0.4.2
0.4.1
0.4.0
0.3.4
0.3.3
0.3.2
0.3.1
Macula HTTP/3 Mesh SDK — connect, subscribe, publish, call, advertise
Current section
Files
Jump to
Current section
Files
src/macula.erl
%%%-------------------------------------------------------------------
%%% @doc Macula SDK — Public API for mesh applications.
%%%
%%% This is the main entry point for applications using the Macula SDK.
%%%
%%% Apps connect via `connect/2', which returns a `macula_client'
%%% pool that internally wraps N peering links to N stations.
%%% `publish/4,5', `subscribe/4,5', `unsubscribe/2', `call/5,6', `providers/3,4',
%%% `advertise/5', `unadvertise/3', `call_stream/5',
%%% `advertise_stream/5', and `unadvertise_stream/3' route through
%%% the pool with realm-per-call semantics. See `macula_pubsub' for
%%% the slice module of the publish/subscribe surface.
%%%
%%% LOCAL streaming (`call_stream/2,3', `open_stream/3,4',
%%% `advertise_stream/2,3', `unadvertise_stream/1') dispatches
%%% in-process via `macula_stream_local' — for unit tests and
%%% same-BEAM pairs.
%%%
%%% Erlang distribution over the mesh ships via `join_mesh/1' (V2
%%% pool carrier) or `join_dist_relay/1' (dedicated dist relay). See
%%% `macula_dist_pool' / `macula_dist_relay_client'.
%%%
%%% @end
%%%-------------------------------------------------------------------
-module(macula).
-include_lib("kernel/include/logger.hrl").
%% Connection
-export([connect/2, close/1, child_spec/3, status/1, links/1]).
%% Reading fields of peer-supplied maps (D26)
-export([field/2, field/3, text/1]).
%% Pub/Sub — realm-per-call against a V2 pool
-export([subscribe/4, subscribe/5,
subscribe_callback/4,
unsubscribe/2,
publish/4, publish/5]).
%% RPC — realm-per-call against a V2 pool
-export([call/5,
call/6,
providers/3,
providers/4,
call_station/7,
call_station/8,
advertise/5,
unadvertise/3]).
%% The pool's own D25 provider authorization, resolved from the DHT and
%% verified against the realm key the pool pins — the `authorization'
%% opt a direct-dial record publish needs.
-export([provider_authorization/3, provider_authorization/4]).
%% The pool renews an advertised chain through this (D32); internal.
-export([renew_authorization/4, advertise_confidentiality/1, call_seal/5]).
%% Signed DHT records — realm-agnostic infrastructure procedures
%% (`_dht.put_record', `_dht.find_record', `_dht.find_records_by_type',
%% `_dht.records.<type>.stored'). The all-zeros realm is the SDK
%% convention for protocol-internal traffic.
-export([put_record/2,
find_record/2,
find_record/3,
find_records/2,
find_records/3,
find_records_by_type/2,
subscribe_records/3,
unsubscribe_records/2]).
-export([sign_node_record/2, sign_node_record/3, sign_domain_record/2,
withdraw_node_record/3, parse_stations/1]).
%% Node-served content (D27): the node that shares content keeps it and
%% serves it; stations only relay. A content id (MCID) is 50 bytes: the hash
%% tag (2, SHA-384), the codec (16#55 raw, 16#56 manifest), then the 48-byte
%% hash.
-export([share_content/3, share_content/4, unshare_content/3,
get_content/3, get_content/4]).
%% A pool link to one station, and the overlay frames a consumer drives it
%% with (macula-realm's peer overlay).
-export([ensure_station_link/4, overlay_subscribe/3, overlay_unsubscribe/2,
send_overlay_frame/2, send_overlay_frame/3]).
%% Streaming RPC (LOCAL in-process + V2 pool, see PLAN_MACULA_STREAMING.md)
-export([
call_stream/2, call_stream/3, call_stream/5, call_stream_station/7,
open_stream/3, open_stream/4,
advertise_stream/2, advertise_stream/3, advertise_stream/5,
advertise_stream/6,
unadvertise_stream/1, unadvertise_stream/3,
send/2, send/3,
recv/1, recv/2,
close_stream/1, close_send/1,
await_reply/1, await_reply/2,
set_reply/2, abort/3
]).
%% Cluster (LAN)
-export([ensure_distributed/0, monitor_nodes/0, unmonitor_nodes/0]).
%% Mesh Distribution
-export([join_mesh/1, join_dist_relay/1, dist_relay_client/0]).
-ifdef(TEST).
-export([join_pool_args/1, renewal/3]).
-endif.
%% Types
-export_type([pool/0, realm/0, provider_io/0,
topic/0, procedure/0,
stream/0, stream_mode/0, stream_handler/0,
m_record/0, record_type/0, record_key/0,
mcid/0]).
-type pool() :: macula_client:pool().
-type realm() :: <<_:256>>. %% 32-byte realm tag.
-type topic() :: binary().
-type procedure() :: binary().
%% The I/O a provider-advertisement resolution runs on, overrideable per
%% call in `advertise/5''s and `provider_authorization/4''s `Opts' (the
%% same seam discipline as `macula_direct_dial:dial_io/0'): each entry
%% defaults to the facade's own function. Tests inject stubs here rather
%% than replacing a module other processes call.
-type provider_io() :: #{
status => fun((pool()) -> {ok, macula_client:status()} | {error, term()}),
find_record => fun((pool(), record_key()) -> {ok, m_record()} | {error, term()}),
sign_node_record => fun((pool(), m_record(), #{not_after := integer()}) ->
{ok, m_record()} | {error, term()}),
realm_key => fun((pool(), realm()) -> {ok, binary()} | none)
}.
-type stream() :: pid().
-type stream_mode() :: server_stream | client_stream | bidi.
-type stream_handler() :: fun((stream(), term()) -> any()).
-type m_record() :: macula_record:m_record().
-type record_type() :: macula_record:type_tag().
-type record_key() :: <<_:256>>. %% DHT storage key — `macula_record:storage_key/1' output.
%%%===================================================================
%%% Connection — V2 (pool, since 3.11.0)
%%%===================================================================
%% @doc Connect to the Macula relay mesh and return a pool handle.
%%
%% `Seeds' is a list of station endpoints: `#{host, port, expected_node_id}'
%% maps, or URL binaries or strings. Every seed names the node_id it
%% expects, in the seed or in the `expected_node_id' option; otherwise no
%% pool starts and `{error, {seeds, expected_node_id_required}}' is
%% returned. The pool spawns one peering link per seed and routes ops with
%% replication, replay, and event dedup. Returns immediately; link
%% handshakes complete asynchronously.
%%
%% Honored opts (full reference: `macula_client:opts()'):
%% <ul>
%% <li>`node_identity': the pool's node identity key, in the node's crypto
%% profile. If absent, the pool uses the node's one stored identity
%% (`macula_node_keys:node_identity/1', at the `node_identity_path'
%% application env): loaded if stored, ground and stored once if not,
%% and refused, never replaced, if the file exists and will not load.
%% Every pool on the node, and every restart, is then the same node.</li>
%% <li>`realm_trust': the realm keys the pool pins, one per realm id, as
%% `#{RealmId => RealmKey}', each realm's public key as carried. A call
%% trusts an org namespaced advertisement only through the key pinned
%% for its realm. Refused as `{error, {realm_trust, invalid}}' unless
%% every id is 32 bytes and every key is well formed for the node's
%% crypto profile, and as `{error, {realm_trust, profile_mismatch}}'
%% for a key of the other profile.</li>
%% <li>`replication_factor' — links per PUBLISH (default 2, since 10.19.0).</li>
%% <li>`capabilities' — per-link bitfield (default 0).</li>
%% <li>`alpn' — QUIC ALPN list (default `[<<"macula">>]').</li>
%% <li>`connect_timeout_ms' — per-link CONNECT/HELLO deadline (default 30_000).</li>
%% <li>`dedup_sweep_ms': how often the inbound publication dedup table is swept.</li>
%% <li>`admission_sweep_ms': how often the request admission is swept for
%% entries past their deadline plus 5 minutes (default 30_000).</li>
%% <li>`renew_backoff_ms' / `renew_recheck_ms': the retry delay of a failed
%% renewal of an advertised chain (default 5_000, doubling, never past
%% its `not_after') and how often a chain past it is asked for again
%% (default 300_000). See `advertise/5' and D32.</li>
%% <li>`verify' — ⚠ REFUSED in any value, here and on every seed in
%% `Seeds', with `{error, {refused, {verify, one_verification_mode}}}'.
%% A link trusts a station in one way only: its handshake signature
%% under the key of its ML-DSA-87 certificate, and its identity
%% through the handshake. There is no chain to check.</li>
%% <li>`expected_node_id' — the station node_id the handshake must
%% prove, for every link this pool dials.</li>
%% <li>`pin_tls_cert' — ⚠ `true' is REFUSED, here and on every seed in
%% `Seeds', with
%% `{error, {refused, {pin_tls_cert, no_pin_primitive_for_mldsa87_identity}}}'.
%% No pin primitive can express an ML-DSA-87 identity, so the option
%% could never be honoured; it pinned nothing at any value. `false'
%% and an absent key pass and change nothing. See macula#15.</li>
%% </ul>
%%
%% Legacy opts silently dropped (with a one-shot `logger:notice'):
%% `relays' (use the `Seeds' positional argument), `realm' (V2 is
%% realm-per-call), `site' (no V2 analog), `connections' (one link
%% per seed; add more seeds to grow the pool).
%%
%% See `macula_client' for the canonical pool implementation and
%% `macula_pubsub' for the slice module.
-spec connect([macula_client:seed()], macula_client:opts()) ->
{ok, pool()} | {error, term()}.
connect(Seeds, Opts) when is_list(Seeds), is_map(Opts) ->
refused(seeds_checked(Seeds, Opts),
fun() -> macula_client:connect(Seeds, Opts) end).
%% @doc Stop a V2 pool. Every subscriber receives a final
%% `{macula_event_gone, SubRef, pool_closed}' message.
-spec close(pool()) -> ok.
close(Pool) when is_pid(Pool) ->
macula_client:close(Pool).
%% @doc OTP child spec to drop a V2 pool into a caller's supervision
%% tree. Give `node_identity' as a loader `{Module, Function, Args}' that
%% returns `{ok, Key}'. Its `Args' say where the key is and never hold the
%% key, because a supervisor logs them when a start fails. See
%% `macula_client:child_spec/3'.
-spec child_spec(term(), [macula_client:seed()], macula_client:opts()) ->
supervisor:child_spec().
child_spec(Id, Seeds, Opts) ->
macula_client:child_spec(Id, Seeds, Opts).
%% @doc Aggregate health snapshot of a V2 pool. Suitable for
%% `/health' or `/status' endpoints; not for hot-loop polling. See
%% `macula_client:status/1' for the full shape.
-spec status(pool()) -> {ok, macula_client:status()}.
status(Pool) when is_pid(Pool) ->
macula_client:status(Pool).
%% @doc Per-link snapshot of a V2 pool — one entry per spawned link
%% with its peer station `node_id' (pubkey), dial `host', `pid', and
%% `connected' flag. Use this to resolve a specific station (by pubkey
%% or hostname) to its link for targeted, per-station operations. See
%% `macula_client:links/1' for the `link_info()' shape.
-spec links(pool()) -> {ok, [macula_client:link_info()]}.
links(Pool) when is_pid(Pool) ->
macula_client:links(Pool).
%%%===================================================================
%%% Pub/Sub — realm-per-call against a V2 pool
%%%===================================================================
%% @doc Publish to `(Realm, Topic)' on `Pool'. Equivalent to
%% `publish/5' with empty opts.
-spec publish(pool(), realm(), topic(), term()) -> ok | {error, term()}.
publish(Pool, Realm, Topic, Payload) ->
macula_pubsub:publish(Pool, Realm, Topic, Payload).
%% @doc Publish to `(Realm, Topic)' on `Pool' with options. See
%% `macula_pubsub:publish/5' for honored opts.
-spec publish(pool(), realm(), topic(), term(), map()) ->
ok | {error, term()}.
publish(Pool, Realm, Topic, Payload, Opts) ->
macula_pubsub:publish(Pool, Realm, Topic, Payload, Opts).
%% @doc Subscribe `Subscriber' to `(Realm, Topic)' on `Pool'.
%% Equivalent to `subscribe/5' with empty opts.
-spec subscribe(pool(), realm(), topic(), pid()) -> {ok, reference()} | {error, {text_too_long | invalid_text, topic}}.
subscribe(Pool, Realm, Topic, Subscriber) ->
macula_pubsub:subscribe(Pool, Realm, Topic, Subscriber).
%% @doc Subscribe `Subscriber' to `(Realm, Topic)' on `Pool' with
%% options. The `delivery' option chooses how a single publisher's
%% out-of-order arrivals are handled:
%% <ul>
%% <li>`ordered' (default) — per-publisher FIFO by seq; out-of-order
%% arrivals are buffered and released in order, a genuinely
%% missing seq skipped after `order_timeout_ms' (a `connect/2'
%% option, default 250ms). A new publisher's first facts are
%% held for up to `order_timeout_ms', so its order starts at the
%% lowest seq seen.</li>
%% <li>`latest_only' — deliver only seqs newer than the highest seen
%% for that publisher (drop stale); no buffering, no delay.</li>
%% <li>`as_arrives' — deliver in raw arrival order; the consumer
%% orders it itself.</li>
%% </ul>
%% Ordering state is kept per publisher, and apart for EVENTs whose
%% publisher signature did not verify.
%% See `macula_pubsub:subscribe/5'.
-spec subscribe(pool(), realm(), topic(), pid(), map()) ->
{ok, reference()} | {error, {text_too_long | invalid_text, topic}}.
subscribe(Pool, Realm, Topic, Subscriber, Opts) ->
macula_pubsub:subscribe(Pool, Realm, Topic, Subscriber, Opts).
%% @doc Subscribe with a callback function. The SDK spawns a small
%% receiver process internally and invokes the callback once per
%% inbound event. See `macula_pubsub:subscribe_callback/4'.
-spec subscribe_callback(pool(), realm(), topic(),
macula_pubsub:callback()) ->
{ok, reference()} | {error, term()}.
subscribe_callback(Pool, Realm, Topic, Callback) ->
macula_pubsub:subscribe_callback(Pool, Realm, Topic, Callback).
%% @doc Drop a pool subscription. Idempotent.
-spec unsubscribe(pool(), reference()) -> ok.
unsubscribe(Pool, SubRef) when is_pid(Pool), is_reference(SubRef) ->
macula_pubsub:unsubscribe(Pool, SubRef).
%%%===================================================================
%%% RPC — realm-per-call against a V2 pool
%%%===================================================================
%% @doc Call `Procedure' in `Realm' at the provider that serves it: resolve
%% the procedure's verified advertisements, authorized against the realm key
%% the pool pinned for `Realm', reach the station a candidate names
%% directly, and call its provider there. See
%% `macula_direct_dial:call/5'. A procedure the pool's linked stations serve
%% themselves, such as `_dht.*', goes through
%% `macula_client:call_linked_station/5'.
-spec call(pool(), realm(), procedure(), term(), 1..600_000) ->
{ok, term()} | {error, term()}.
call(Pool, Realm, Procedure, Payload, TimeoutMs) ->
macula_direct_dial:call(Pool, Realm, Procedure, Payload, TimeoutMs).
%% @doc As `call/5', with `Opts'. `#{provider => NodeId}' calls THAT
%% provider of `Procedure' and no other: the procedure is resolved and its
%% advertisements trust-checked exactly as `call/5' does, and only the named
%% provider's are tried. A provider with no trusted advertisement by the
%% deadline is `{error, {unresolved, provider_not_advertised}}'. To have
%% every provider of a procedure answer, list them with `providers/3,4' and
%% make one call per provider. See `macula_direct_dial:call/6'.
-spec call(pool(), realm(), procedure(), term(), 1..600_000,
#{provider => <<_:256>>}) ->
{ok, term()} | {error, term()}.
call(Pool, Realm, Procedure, Payload, TimeoutMs, Opts) when is_map(Opts) ->
macula_direct_dial:call(Pool, Realm, Procedure, Payload, TimeoutMs, Opts).
%% @doc As `providers/4', within 5 seconds.
-spec providers(pool(), realm(), procedure()) ->
{ok, [#{provider := <<_:256>>, station := <<_:256>>}]} | {error, term()}.
providers(Pool, Realm, Procedure) ->
providers(Pool, Realm, Procedure, 5_000).
%% @doc Who provides `Procedure' in `Realm': each provider whose
%% advertisement passes the same trust check `call/5' applies, with the
%% station it serves from, in one DHT lookup bounded by `TimeoutMs'. A
%% provider whose record has not replicated yet is not listed; ask again
%% for a fresher answer. See `macula_direct_dial:providers/4'.
-spec providers(pool(), realm(), procedure(), 1..600_000) ->
{ok, [#{provider := <<_:256>>, station := <<_:256>>}]} | {error, term()}.
providers(Pool, Realm, Procedure, TimeoutMs) ->
macula_direct_dial:providers(Pool, Realm, Procedure, TimeoutMs).
%% @doc Issue a CALL to `Target', a provider's node_id, at ONE specific
%% station, dialing it directly even if it is not in the pool's seed set.
%% `Station' is a seed URL (e.g. `<<"quic://[::1]:4433">>'). The pool reuses
%% an existing link or dials and monitors a new one, waits for the
%% handshake, and calls there. This is the direct-dial data path: resolve a
%% provider's serving_station and its endpoint, then reach it in one hop.
%% See `macula_client:call_station/11'. With no options it carries no signed state to decide on sealing from, so it is
%% refused as `{error, {confidentiality, no_signed_state}}' (macula 13.0.0): use `call_station/8' with
%% `advertisement', `confidential => required' or `confidential => off' (see `call_seal/5').
-spec call_station(pool(), macula_client:seed(), <<_:256>>, realm(), procedure(),
term(), 1..600_000) -> {ok, term()} | {error, term()}.
call_station(Pool, Station, Target, Realm, Procedure, Payload, TimeoutMs) ->
call_station(Pool, Station, Target, Realm, Procedure, Payload, TimeoutMs, #{}).
%% @doc As `call_station/7', presenting a capability token to a gated
%% provider via `Opts' (`#{ucan_token => Token}'). Empty/absent = none.
%% Slice 7b dual-trust. `Opts' also carries the station this dial must
%% prove, `expected_node_id' (see `macula_client:call_station/11'), and may
%% set `dial_timeout_ms', how much of `TimeoutMs' the wait for a fresh
%% link's handshake may take (default: all of it). `pin_tls_cert => true'
%% is REFUSED with
%% `{error, {refused, {pin_tls_cert, no_pin_primitive_for_mldsa87_identity}}}',
%% and `verify' in any value with
%% `{error, {refused, {verify, one_verification_mode}}}' (see
%% `trust_options_checked/1').
-spec call_station(pool(), macula_client:seed(), <<_:256>>, realm(), procedure(),
term(), 1..600_000, map()) ->
{ok, term()} | {error, term()}.
call_station(Pool, Station, Target, Realm, Procedure, Payload, TimeoutMs, Opts) ->
refused(target_checked(Station, Opts),
fun() -> do_call_station(Pool, Station, Target, Realm, Procedure,
Payload, TimeoutMs, Opts) end).
do_call_station(Pool, Station, Target, Realm, Procedure, Payload, TimeoutMs, Opts) ->
Ucan = maps:get(ucan_token, Opts, <<>>),
LinkOpts = maps:with([expected_node_id], Opts),
DialTimeoutMs = maps:get(dial_timeout_ms, Opts, TimeoutMs),
Resolve = fun() -> find_records(Pool, macula_record:procedure_key(Realm, Procedure)) end,
sealed_or_refused(call_seal(Target, Realm, Procedure, Opts, Resolve), fun(Seal) ->
macula_client:call_station(Pool, Station, Target, Realm, Procedure, Payload,
TimeoutMs, Ucan, LinkOpts, DialTimeoutMs, Seal)
end).
sealed_or_refused({ok, Seal}, Call) -> Call(Seal);
sealed_or_refused({error, _} = Refused, _Call) -> Refused.
%% @private
%% @doc How a call to the explicit `Target' is sent (E2E design §8.1, Amendment A1), decided from signed state only, so
%% a lookup can deny the call but never downgrade it:
%%
%% - `advertisement', a verified procedure advertisement of `Target' for this realm and procedure (as `find_records/2'
%% or `providers/3' hand one over): sealed to its KEM key when it names one, in the clear when it names none, unless
%% `confidential => required', which then fails with `no_kem_key';
%% - `confidential => off': in the clear, the application's own decision;
%% - `confidential => required' without an advertisement: `Resolve()' looks up the procedure's advertisements, and
%% only one `Target' signed is used; with none, or one naming no key, the call fails with `no_kem_key';
%% - none of these: `{error, {confidentiality, no_signed_state}}'.
-spec call_seal(<<_:256>>, realm(), procedure(), map(), fun(() -> {ok, [map()]} | {error, term()})) ->
{ok, clear | {sealed_to, binary()}} | {error, {confidentiality, atom()}}.
call_seal(_Target, _Realm, _Procedure, #{confidential := off}, _Resolve) ->
{ok, clear};
call_seal(Target, Realm, Procedure, #{advertisement := Ad} = Opts, _Resolve) ->
sealed_by(advertised_key(Ad, Target, Realm, Procedure), maps:get(confidential, Opts, preferred));
call_seal(Target, Realm, Procedure, #{confidential := required}, Resolve) ->
sealed_by(resolved_key(Resolve(), Target, Realm, Procedure), required);
call_seal(_Target, _Realm, _Procedure, _Opts, _Resolve) ->
{error, {confidentiality, no_signed_state}}.
%% The KEM key an advertisement of `Target' for this realm and procedure names, `none' when it names none.
advertised_key(Ad, Target, Realm, Procedure) ->
advertised_by(macula_record:read_procedure_advertisement(Ad), Target, Realm, Procedure).
advertised_by(#{advertiser_node := Target, realm_id := Realm, procedure := Procedure} = Read, Target, Realm, Procedure) ->
{ok, maps:get(kem_key, Read, none)};
advertised_by(#{advertiser_node := Target, realm_id := Realm}, Target, Realm, _Procedure) ->
{error, {confidentiality, not_the_procedure}};
advertised_by(#{advertiser_node := Target}, Target, _Realm, _Procedure) ->
{error, {confidentiality, not_the_realm}};
advertised_by(_AnotherNodes, _Target, _Realm, _Procedure) ->
{error, {confidentiality, not_the_target}}.
%% The key an advertisement `Target' signed names, among those resolved: a keyed one wins over a stale keyless one
%% listed before it; anything else seals nothing.
resolved_key({ok, Ads}, Target, Realm, Procedure) ->
first_key([advertised_key(Ad, Target, Realm, Procedure) || Ad <- Ads,
macula_record:type(Ad) =:= macula_record:type_procedure_advertisement()]);
resolved_key(_NothingResolved, _Target, _Realm, _Procedure) ->
{ok, none}.
first_key(Found) ->
keyed_first([KemKey || {ok, KemKey} <- Found, is_binary(KemKey)]).
keyed_first([KemKey | _]) -> {ok, KemKey};
keyed_first([]) -> {ok, none}.
sealed_by({ok, KemKey}, _Mode) when is_binary(KemKey) -> {ok, {sealed_to, KemKey}};
sealed_by({ok, none}, required) -> {error, {confidentiality, no_kem_key}};
sealed_by({ok, none}, _Preferred) -> {ok, clear};
sealed_by({error, _} = Refused, _Mode) -> Refused.
%% @doc Register a procedure handler on a V2 pool and advertise it:
%% the pool resolves its own D25 provider authorization — the
%% realm-signed `org_directory' and the org-signed
%% `procedure_delegation' that names the pool's node id, both fetched
%% from the DHT — verifies it against the pinned realm key, and hands
%% every link the advertisement spec: for each link the pool signs an
%% advertisement naming the station that link is connected to as
%% `serving_station', sent as an ADVERTISE frame (signed again on
%% reconnect and on link respawn, never past the chain's earlier
%% expiry). The pool renews the chain at a third of its remaining life
%% (`macula_client:advertise/8', D32): a delegation lives 30 minutes, so
%% a provider stays advertised while the realm keeps reissuing, and
%% drops out once it stops. The frame registers the procedure at that station; it
%% does not put a record in the DHT, which is what a direct-dialling
%% caller resolves (`macula_response:advertise_direct/6,7' publishes
%% that one). The
%% procedure must carry an org namespace, and the pool must run a
%% provisioned identity whose delegation the org has published, and
%% pin the realm's key (`realm_trust' at connect); a missing piece
%% fails fast with `{error, {provider_authorization, _}}'.
%%
%% `Opts' `auth' sets the procedure's policy: `open' (default, serve
%% any identified caller), `{ucan_required, IssuerNodeId}' (gated to
%% tokens from one known node), or `{realm_member_required, RealmKeyId,
%% RequiredCan}' (gated to realm membership at a specific tier) -- see
%% `macula_client:auth_policy()' for the full set.
%%
%% `stations => [StationNodeId]' registers the procedure on the links to
%% those stations only, each named by the node_id its link pins, and a link
%% respawned later registers it again only when its station is one of them.
%% A station the pool holds no link to is refused as `{error,
%% {station_not_linked, StationNodeId}}', an empty list as `{error,
%% {stations, empty}}' and anything but a list of 32-byte node ids as
%% `{error, {stations, malformed}}'; nothing is registered or kept. Without
%% it every link registers the procedure.
%%
%% `Opts' may also carry `advertise', an arity-6 override for the pool
%% fan-out, which then decides where the procedure registers (see
%% `macula_response:advertise_opts()'), and the `provider_io/0' seam
%% entries the resolution reads its DHT calls from.
-spec advertise(pool(), realm(), procedure(),
macula_client:handler(), map()) ->
ok | {error, term()}.
advertise(Pool, Realm, Procedure, Handler, Opts)
when is_pid(Pool), is_binary(Realm), byte_size(Realm) =:= 32 ->
on_stations(stations_opt(Opts), fun(Stations) ->
Policy = maps:get(auth, Opts, open),
Advertise = maps:get(advertise, Opts,
fun(P, R, Pr, H, Po, Spec) ->
macula_client:advertise(P, R, Pr, H, Po, Spec, Stations,
renewal(R, Pr, Opts))
end),
advertise_authorized(Pool, Realm, Procedure, Opts, fun(Spec) ->
Advertise(Pool, Realm, Procedure, Handler, Policy, Spec)
end)
end).
%% The `stations' an advertisement names, checked before anything is resolved
%% or registered: `all' when it names none.
stations_opt(#{stations := []}) -> {error, {stations, empty}};
stations_opt(#{stations := Stations}) -> node_ids(Stations, Stations);
stations_opt(#{}) -> {ok, all}.
node_ids([], Stations) -> {ok, Stations};
node_ids([<<_:256>> | Rest], Stations) -> node_ids(Rest, Stations);
node_ids(_NotANodeIdList, _Stations) -> {error, {stations, malformed}}.
on_stations({ok, Stations}, Fun) -> Fun(Stations);
on_stations({error, _} = Refused, _Fun) -> Refused.
%% @doc Stop advertising a procedure on a V2 pool.
-spec unadvertise(pool(), realm(), procedure()) -> ok.
unadvertise(Pool, Realm, Procedure) ->
macula_client:unadvertise(Pool, Realm, Procedure).
%%%===================================================================
%%% Signed DHT records (v3.3.0)
%%%===================================================================
%%%
%%% Records are typed payloads signed in the signed-object format of
%%% DESIGN_PQ_SIGNED_FRAMES_AND_RECORDS.md and stored in the relay
%%% mesh's distributed hash table. A record travels as its wire form:
%%% stations store and forward its bytes as received. A record reaches
%%% a caller only after it verifies under the node's crypto profile
%%% (`macula_record:verify/2'): `find_record/2' returns the refusal of
%%% one that does not, and the list functions and `subscribe_records/3'
%%% drop it.
%%%
%%% See `macula_record' for the record shape, the typed
%%% constructors (`node_record/3', `realm_directory/3',
%%% `realm_stations/2', `procedure_advertisement/4',
%%% `content_announcement/3', `tombstone/2', and the foundation_*
%%% constructors), and `storage_key/1' for the DHT addressing rule.
%%%
%%% Two complementary retrieval paths:
%%%
%%% - `find_record/2' — fetch one record by its
%%% `storage_key/1' output
%%% - `find_records_by_type/2' — list every record of a given
%%% type tag
%%%
%%% Plus a live-update channel:
%%%
%%% - `subscribe_records/3' — receive new records of a type
%%% as they are stored
%% Procedure + topic shape — hidden from API consumers but exposed
%% as documentation. The relay backend (hecate-station and successors)
%% MUST advertise these procedures and publish on the per-type
%% record-stored topic for the SDK to function. DHT traffic travels
%% under the all-zeros realm tag (protocol-internal infrastructure;
%% the same convention `macula_dist_pool' uses for tunnel frames).
-define(DHT_REALM, <<0:256>>).
-define(DHT_PUT_RECORD_PROC, <<"_dht.put_record">>).
-define(DHT_FIND_RECORD_PROC, <<"_dht.find_record">>).
-define(DHT_FIND_RECORDS_PROC, <<"_dht.find_records">>).
-define(DHT_FIND_RECORDS_BY_TYPE_PROC, <<"_dht.find_records_by_type">>).
-define(DHT_RECORD_TIMEOUT_MS, 5_000).
%% @doc Store a signed record in the mesh DHT via a V2 pool.
%%
%% Build the record via the typed constructors in `macula_record'
%% (`node_record/3,4', `content_announcement/3,4', `tombstone/2,3',
%% `realm_directory/3,4', `procedure_advertisement/4,5', etc.) and sign
%% it. A record this node signs about itself is signed by its pool,
%% which holds the node identity key: through
%% `macula_client:sign_node_record/2', a domain record (tags 0x20 to
%% 0xFF) through `macula_client:sign_domain_record/2', and the
%% tombstone of either through `macula_client:withdraw_node_record/3'.
%% A realm-, org- or
%% foundation-signed record is signed with `macula_record:sign/2' and
%% its signer's key. Pass the signed record or its wire form
%% (`macula_record:encode/1'). The record travels as its
%% wire form; the station verifies it on receipt and stores it under
%% `macula_record:storage_key/1', propagating to the K-nearest peers.
-spec put_record(pool(), m_record() | binary()) -> ok | {error, term()}.
put_record(Pool, #{key := _, tbs := _, signature := _} = Signed) when is_pid(Pool) ->
put_record(Pool, macula_record:encode(Signed));
put_record(Pool, Wire) when is_pid(Pool), is_binary(Wire) ->
%% Record bytes paced per pool, so a bulk writer stays under the
%% station's STORE allowance (D28, 3.5) instead of running into stored 0.
%% One bucket per pool is conservative for a pool of several links: its
%% calls fan out one link at a time, and the per-link pacer in
%% macula_station_link:put_record/3 paces a caller that writes straight
%% through a link.
ok = macula_store_pacer:await(Pool, byte_size(Wire)),
classify_put(macula_client:call_linked_station(Pool, ?DHT_REALM,
?DHT_PUT_RECORD_PROC,
Wire, ?DHT_RECORD_TIMEOUT_MS)).
classify_put({ok, ok}) -> ok;
%% The reply's atom crossed the wire as text and decoded back to the
%% codec's {text, _} marker (D26 makes no atom on the wire).
classify_put({ok, {text, <<"ok">>}}) -> ok;
classify_put({ok, Reply}) -> {error, {unexpected_reply, Reply}};
classify_put({error, _} = E) -> E.
%% @doc Fetch a record from the mesh DHT by its
%% `macula_record:storage_key/1'.
%%
%% Returns `{error, not_found}' when no record exists at the key. The
%% record is verified under the node's crypto profile before it is
%% returned; one that does not verify returns its refusal from
%% `macula_record:verify/2', such as `{error, expired}'.
-spec find_record(pool(), record_key()) ->
{ok, m_record()} | {error, not_found | term()}.
find_record(Pool, Key) ->
find_record(Pool, Key, ?DHT_RECORD_TIMEOUT_MS).
%% @doc As `find_record/2', waiting at most `TimeoutMs' for the reply, for a
%% caller that bounds its work by a deadline of its own.
-spec find_record(pool(), record_key(), pos_integer()) ->
{ok, m_record()} | {error, not_found | term()}.
find_record(Pool, Key, TimeoutMs)
when is_pid(Pool), is_binary(Key), byte_size(Key) =:= 32,
is_integer(TimeoutMs), TimeoutMs > 0 ->
classify_find(macula_client:call_linked_station(Pool, ?DHT_REALM,
?DHT_FIND_RECORD_PROC,
#{key => Key}, TimeoutMs)).
classify_find({ok, Wire}) when is_binary(Wire) ->
with_profile(fun(Profile) -> macula_record:verify(Wire, Profile) end);
classify_find({ok, not_found}) -> {error, not_found};
%% The reply's atom crossed the wire as text and decoded back to the
%% codec's {text, _} marker (D26 makes no atom on the wire).
classify_find({ok, {text, <<"not_found">>}}) -> {error, not_found};
classify_find({ok, Reply}) -> {error, {unexpected_reply, Reply}};
classify_find({error, _} = E) -> E.
%% @doc Fetch EVERY record stored at `Key' — the full multi-value
%% set, e.g. every `procedure_advertisement' under one procedure's
%% storage key. Where `find_record/2' returns the first record (or
%% `not_found'), this returns the whole list, empty when none.
%%
%% The relay's local store is a signer-deduped multiset: one record
%% per signing key at a storage key, so N providers of one procedure
%% return N records. Each record is verified under the node's crypto
%% profile, and a record that does not verify is dropped.
-spec find_records(pool(), record_key()) ->
{ok, [m_record()]} | {error, term()}.
find_records(Pool, Key) ->
find_records(Pool, Key, ?DHT_RECORD_TIMEOUT_MS).
%% @doc As `find_records/2', waiting at most `TimeoutMs' for the reply, for a
%% caller that bounds its work by a deadline of its own.
-spec find_records(pool(), record_key(), pos_integer()) ->
{ok, [m_record()]} | {error, term()}.
find_records(Pool, Key, TimeoutMs)
when is_pid(Pool), is_binary(Key), byte_size(Key) =:= 32,
is_integer(TimeoutMs), TimeoutMs > 0 ->
classify_find_list(macula_client:call_linked_station(Pool, ?DHT_REALM,
?DHT_FIND_RECORDS_PROC,
#{key => Key}, TimeoutMs)).
classify_find_list({ok, Wires}) when is_list(Wires) -> verified_records(Wires);
classify_find_list({ok, Reply}) -> {error, {unexpected_reply, Reply}};
classify_find_list({error, _} = E) -> E.
%% @doc Return every record of a given type currently visible from
%% the pool's connected stations, each verified under the node's
%% crypto profile; a record that does not verify is dropped.
%%
%% Coverage depends on each station's view of the DHT — a single
%% station sees its local replicas plus whatever its peers have
%% gossiped. Aggregating across the full mesh requires querying
%% multiple stations and deduplicating by record key.
-spec find_records_by_type(pool(), record_type()) ->
{ok, [m_record()]} | {error, term()}.
find_records_by_type(Pool, Type)
when is_pid(Pool), is_integer(Type), Type >= 0, Type =< 255 ->
classify_list(macula_client:call_linked_station(Pool, ?DHT_REALM,
?DHT_FIND_RECORDS_BY_TYPE_PROC,
#{type => Type},
?DHT_RECORD_TIMEOUT_MS)).
classify_list({ok, Wires}) when is_list(Wires) -> verified_records(Wires);
classify_list({ok, Reply}) -> {error, {unexpected_reply, Reply}};
classify_list({error, _} = E) -> E.
%% The records among Wires that verify under the node's crypto profile;
%% the rest are dropped.
verified_records(Wires) ->
with_profile(fun(Profile) ->
{ok, [Record || Wire <- Wires, {ok, Record} <- [macula_record:verify(Wire, Profile)]]}
end).
%% Verify with the node's crypto profile, or return the refusal of a
%% node that has none.
with_profile(Verify) ->
profile_verified(macula_crypto_profile:configured(), Verify).
profile_verified({ok, Profile}, Verify) -> Verify(Profile);
profile_verified({error, _} = Refusal, _Verify) -> Refusal.
%% @doc Subscribe to live record-stored events filtered by type.
%%
%% The callback receives each newly-stored record of the given type
%% that verifies under the node's crypto profile. Returns a
%% subscription reference for `unsubscribe_records/2'.
%% Topic shape is `_dht.records.<type>.stored', rendered with the
%% type tag as a decimal integer for log friendliness.
-spec subscribe_records(pool(), record_type(),
fun((m_record()) -> any())) ->
{ok, reference()} | {error, term()}.
subscribe_records(Pool, Type, Callback)
when is_pid(Pool), is_integer(Type), Type >= 0, Type =< 255,
is_function(Callback, 1) ->
Topic = record_stored_topic(Type),
macula_pubsub:subscribe_callback(Pool, ?DHT_REALM, Topic,
wrap_record_callback(Callback)).
%% @doc Cancel a `subscribe_records/3' subscription.
-spec unsubscribe_records(pool(), reference()) -> ok.
unsubscribe_records(Pool, Ref)
when is_pid(Pool), is_reference(Ref) ->
macula_pubsub:unsubscribe(Pool, Ref).
%%%===================================================================
%%% Signing records this node signs about itself — the pool holds the key
%%%===================================================================
%% @doc Sign a record this node signs about itself with the pool's node
%% identity key, in the pool's own process: a node record, a procedure
%% advertisement or a content announcement that names this node. See
%% `macula_client:sign_node_record/2' for the record checks and
%% refusals. `Opts' `not_after' bounds the signed record's expiry: the
%% pool judges it on its own clock, refuses a bound already passed as
%% `{error, not_after_passed}', ends the record at the bound when the
%% bound comes before the record's lifetime runs out, and keeps the
%% built lifetime otherwise.
-spec sign_node_record(pool(), m_record()) ->
{ok, m_record()} | {error, term()}.
sign_node_record(Pool, Record) when is_pid(Pool) ->
macula_client:sign_node_record(Pool, Record).
-spec sign_node_record(pool(), m_record(), #{not_after := integer()}) ->
{ok, m_record()} | {error, term()}.
sign_node_record(Pool, Record, Opts) when is_pid(Pool), is_map(Opts) ->
macula_client:sign_node_record(Pool, Record, Opts).
%% @doc Sign a domain record (tags 0x20 to 0xFF) as this node, with the
%% pool's node identity key, in the pool's own process. See
%% `macula_client:sign_domain_record/2'.
-spec sign_domain_record(pool(), m_record()) ->
{ok, m_record()} | {error, term()}.
sign_domain_record(Pool, Record) when is_pid(Pool) ->
macula_client:sign_domain_record(Pool, Record).
%% @doc Withdraw a record this node signed — a node record, a procedure
%% advertisement, a content announcement or a domain record — with a
%% tombstone signed by the pool's node identity key, in the pool's own
%% process. See `macula_client:withdraw_node_record/3'.
-spec withdraw_node_record(pool(), m_record() | binary(), macula_record:reason()) ->
{ok, m_record()} | {error, term()}.
withdraw_node_record(Pool, Withdrawn, Reason) when is_pid(Pool) ->
macula_client:withdraw_node_record(Pool, Withdrawn, Reason).
%% @doc Parse a `MACULA_STATIONS' seed list: comma-separated
%% `<node id>@<host>:<port>' entries, the node id as 64 lowercase hex
%% characters, the host a DNS name, an IPv4 address or a bracketed IPv6
%% address, the port 1 to 65535. Returns the seeds in order, each pinned
%% to its node id. A refusal names the entry's position and what is
%% wrong with it, and carries neither the value, nor a node id, nor a
%% host. See `macula_stations:parse/1'.
-spec parse_stations(binary()) ->
{ok, [macula_stations:seed()]} | {error, term()}.
parse_stations(Value) when is_binary(Value) ->
macula_stations:parse(Value).
record_stored_topic(Type) ->
iolist_to_binary([<<"_dht.records.">>,
integer_to_binary(Type),
<<".stored">>]).
%% Adapt a 1-arg `(Record) -> any()' user callback to the 3-arg
%% `(Topic, Payload, Meta) -> any()' shape `macula_pubsub' delivers.
%%
%% PubSub delivers the payload as the record's wire form (the
%% substrate's `record_fanout' publishes `macula_record:encode/1'
%% output on the `_dht.records.<type>.stored' topic). The record is
%% verified here under the node's crypto profile, so the callback
%% receives only verified records. A payload that does not verify is
%% dropped silently: surfacing it would force every user to handle
%% refusals on a protocol-internal channel.
wrap_record_callback(Fun) ->
fun(_Topic, Payload, _Meta) -> apply_callback_with_verified(Fun, Payload) end.
apply_callback_with_verified(Fun, Payload) ->
callback_with(with_profile(fun(Profile) -> macula_record:verify(Payload, Profile) end), Fun).
callback_with({ok, Record}, Fun) -> Fun(Record), ok;
callback_with({error, _}, _Fun) -> ok.
%%%===================================================================
%%% Node-served content (D27)
%%%===================================================================
-type mcid() :: <<_:400>>.
%% @doc Share `Bytes' in `Realm', as `share_content/4' with no options.
-spec share_content(pool(), realm(), binary()) -> {ok, mcid()} | {error, term()}.
share_content(Pool, Realm, Bytes) ->
share_content(Pool, Realm, Bytes, #{}).
%% @doc Share `Bytes' in `Realm' from this node (D27): the pool's sharer keeps
%% them, serves them on the node's own content procedure and announces their
%% root in the DHT, naming the realm, the station the node is reachable
%% through and that procedure (`macula_content_sharer'). Returns the root's
%% content id: bytes of at most 256 KiB are one raw block, larger bytes a
%% manifest over their chunks. The content is available while this node is
%% online. `Opts': `name', and `org', the org whose delegation serves the
%% content (`<org>/content_v1_<node id>'); without `org' the node serves in
%% its own namespace (`~<node id>/content_v1').
-spec share_content(pool(), realm(), binary(), map()) -> {ok, mcid()} | {error, term()}.
share_content(Pool, Realm, Bytes, Opts)
when is_pid(Pool), is_binary(Realm), byte_size(Realm) =:= 32, is_binary(Bytes), is_map(Opts) ->
shared_by(macula_content_sharer_sup:sharer(Pool), Realm, Bytes, Opts).
shared_by({ok, Sharer}, Realm, Bytes, Opts) -> macula_content_sharer:share(Sharer, Realm, Bytes, Opts);
shared_by({error, _} = Error, _Realm, _Bytes, _Opts) -> Error.
%% @doc Stop sharing the root `MCID' in `Realm': its announcement is withdrawn
%% and its bytes dropped. Idempotent.
-spec unshare_content(pool(), realm(), mcid()) -> ok.
unshare_content(Pool, Realm, MCID) when is_pid(Pool) ->
unshared_by(macula_content_sharer_sup:sharer(Pool), Realm, MCID).
unshared_by({ok, Sharer}, Realm, MCID) -> macula_content_sharer:unshare(Sharer, Realm, MCID);
unshared_by({error, _}, _Realm, _MCID) -> ok.
%% @doc Fetch `MCID' in `Realm', as `get_content/4' with no options.
-spec get_content(pool(), realm(), mcid()) -> {ok, binary()} | {error, term()}.
get_content(Pool, Realm, MCID) ->
get_content(Pool, Realm, MCID, #{}).
%% @doc Fetch `MCID' from a node that shares it in `Realm' (D27): find its
%% announcements, open a stream to a sharer through the station it
%% announced, and verify every block and the manifest against their content
%% ids (`macula_content_fetch'). A sharer that fails moves the fetch to the
%% next; `{error, {unavailable, [{Sharer, Reason}]}}' when all fail,
%% `{error, not_shared}' when none announces it, `{error, invalid_mcid}' for
%% an id that is not tag 2. `Opts': `max_bytes' (256 MiB), `max_chunks',
%% `chunk_timeout_ms', `parallel'. No realm key is needed: content verifies
%% itself.
-spec get_content(pool(), realm(), mcid(), map()) -> {ok, binary()} | {error, term()}.
get_content(Pool, Realm, MCID, Opts)
when is_pid(Pool), is_binary(Realm), byte_size(Realm) =:= 32, is_map(Opts) ->
macula_content_fetch:get(Pool, MCID, Opts#{realm => Realm}).
%%%===================================================================
%%% Streaming RPC (v1.5.0+)
%%%===================================================================
%%%
%%% Streaming RPC ships in two phases (see PLAN_MACULA_STREAMING.md):
%%%
%%% Phase 1 (this release) — LOCAL dispatch only. Client and server
%%% halves both live in the same BEAM and are paired in-process. The
%%% public surface below is what cross-node streaming will use; only
%%% the transport behind the scenes changes in Phase 2.
%%%
%%% Phase 2 — wire STREAM_OPEN / STREAM_DATA / STREAM_END / STREAM_ERROR
%%% / STREAM_REPLY frames through QUIC, one QUIC stream per call_id.
%%%
%%% Patterns supported (gRPC taxonomy):
%%% server_stream — single Args, streamed reply
%%% client_stream — streamed Args, single reply
%%% bidi — duplex
%%%
%%% Example (server-stream):
%%%
%%% ok = macula:advertise_stream(<<"foo.count">>, server_stream,
%%% fun(Stream, #{n := N}) ->
%%% [ok = macula:send(Stream, integer_to_binary(I))
%%% || I <- lists:seq(1, N)],
%%% macula:close_stream(Stream)
%%% end),
%%% {ok, S} = macula:call_stream(<<"foo.count">>, #{n => 5}),
%%% drain(S).
%%%
%%% drain(S) ->
%%% case macula:recv(S) of
%%% {chunk, Bin} -> io:format("~s~n", [Bin]), drain(S);
%%% eof -> ok
%%% end.
%% @doc Open a LOCAL in-process server-stream call. Used for unit
%% tests and same-BEAM dispatch via `macula_stream_local'.
-spec call_stream(procedure(), term()) -> {ok, stream()} | {error, term()}.
call_stream(Procedure, Args) when is_binary(Procedure) ->
call_stream(Procedure, Args, #{}).
%% @doc Open a LOCAL in-process server-stream call with options.
-spec call_stream(procedure(), term(), map()) ->
{ok, stream()} | {error, term()}.
call_stream(Procedure, Args, Opts) when is_binary(Procedure), is_map(Opts) ->
macula_stream_local:call_stream(Procedure, Args, Opts).
%% @doc Open a streaming RPC to `Procedure''s provider in `Realm': resolve
%% the provider through its `procedure_advertisement' and open the stream
%% at its serving station, naming the provider as the target, as
%% `call/5' does for a single-reply call. Same as
%% `macula_direct_dial:call_stream/5'. The returned stream is bound to
%% that station's link (errors with `peer_down' if the link dies; caller
%% re-opens).
%% `Opts' `ucan_token' presents a UCAN to a streaming procedure
%% advertised with an `auth' policy (see `advertise_stream/6').
%% An open whose signed STREAM_OPEN would be longer than
%% `max_stream_open_bytes' (1 MiB by default) returns
%% `{error, {open_too_large, Limit}}' without sending anything.
-spec call_stream(pool(), realm(), procedure(), term(), map()) ->
{ok, stream()} | {error, term()}.
call_stream(Pool, Realm, Procedure, Args, Opts)
when is_pid(Pool), is_binary(Realm), byte_size(Realm) =:= 32,
is_binary(Procedure), is_map(Opts) ->
macula_direct_dial:call_stream(Pool, Realm, Procedure, Args, Opts).
%% @doc Open a streaming RPC to `Target', a provider's node_id, by DIALING
%% a specific station directly (direct-dial): the streaming analogue of
%% `call_station/7'. Compose it with DHT resolution
%% (`find_records' -> `read_procedure_advertisement' -> `station_endpoint')
%% to reach a stream provider in one hop, exactly as a unary caller does.
%% `Opts' may set `dial_timeout_ms' (default 10_000) and a `mode'.
%% `Opts' also names the station this dial must prove, `expected_node_id'
%% (see `macula_client:call_station/11'); `pin_tls_cert => true' and `verify'
%% are REFUSED as `call_station/8' describes.
%%
%% Whether the STREAM_OPEN and every later frame is sealed is decided from
%% signed state only, as for `call_station/8' (`call_seal/5', E2E design
%% §8.1): `advertisement', `confidential => off' or `confidential =>
%% required'. With none of them the open is refused
%% `{error, {confidentiality, no_signed_state}}' before anything is sent.
-spec call_stream_station(pool(), macula_client:seed(), <<_:256>>, realm(), procedure(),
term(), map()) -> {ok, stream()} | {error, term()}.
call_stream_station(Pool, Station, Target, Realm, Procedure, Args, Opts)
when is_pid(Pool), is_binary(Target), byte_size(Target) =:= 32,
is_binary(Realm), byte_size(Realm) =:= 32,
is_binary(Procedure), is_map(Opts) ->
refused(target_checked(Station, Opts),
fun() -> do_call_stream_station(Pool, Station, Target, Realm, Procedure, Args, Opts) end).
%% The pool is handed the seal decided, not the signed state it was decided from.
do_call_stream_station(Pool, Station, Target, Realm, Procedure, Args, Opts) ->
Resolve = fun() -> find_records(Pool, macula_record:procedure_key(Realm, Procedure)) end,
sealed_or_refused(call_seal(Target, Realm, Procedure, Opts, Resolve), fun(Seal) ->
macula_client:call_stream_station(Pool, Station, Target, Realm, Procedure, Args,
maps:without([advertisement, confidential], Opts), Seal)
end).
%% @doc A pool link to `Station', for a consumer that must address one
%% station through one link, as an overlay does: a live link the pool
%% already holds to that station, else one it dials. The station is
%% pinned by the `expected_node_id' in its seed map or in `Opts', which a
%% link requires (D5, D16); `pin_tls_cert => true' and `verify' are
%% refused as `call_station/8' describes. Answers `{ok, Link}' once the
%% link's handshake has completed, within `TimeoutMs', or
%% `{error, not_connected}'. The pool owns and monitors the link: it
%% respawns it and ends it with the pool, so a consumer holds the pid and
%% never stops it. Drive it with `overlay_subscribe/3' and
%% `send_overlay_frame/2,3'.
-spec ensure_station_link(pool(), macula_client:seed(), map(), pos_integer()) ->
{ok, pid()} | {error, term()}.
ensure_station_link(Pool, Station, Opts, TimeoutMs)
when is_pid(Pool), is_map(Opts), is_integer(TimeoutMs), TimeoutMs > 0 ->
refused(target_checked(Station, Opts),
fun() -> macula_client:ensure_station_link(Pool, Station, Opts, TimeoutMs) end).
%% @doc Take the overlay frames for `Realm' arriving on `Link' (from
%% `ensure_station_link/4'), delivered to `Subscriber' as
%% `{macula_overlay_frame, SubRef, Frame, Meta}', where `Meta' names the
%% frame's authenticated `sender' and, for a relayed frame, the station it
%% came `via'. A frame from the connected peer, or a GOSSIP, for a realm
%% nobody on this link subscribed to is dropped.
-spec overlay_subscribe(pid(), realm(), pid()) -> {ok, reference()} | {error, term()}.
overlay_subscribe(Link, Realm, Subscriber) ->
macula_station_link:overlay_subscribe(Link, Realm, Subscriber).
%% @doc Drop an overlay subscription. Idempotent.
-spec overlay_unsubscribe(pid(), reference()) -> ok.
overlay_unsubscribe(Link, SubRef) ->
macula_station_link:overlay_unsubscribe(Link, SubRef).
%% @doc Send a pre-built, pre-signed overlay frame to the peer at the other
%% end of `Link'. `{error, not_connected}' until its handshake completes.
-spec send_overlay_frame(pid(), macula_frame:frame()) -> ok | {error, term()}.
send_overlay_frame(Link, Frame) ->
macula_station_link:send_overlay_frame(Link, Frame).
%% @doc Send a pre-built, pre-signed overlay frame to `TargetPeer' (its
%% node_id), relayed by the station `Link' is connected to.
%% `{error, not_connected}' until the link's handshake completes.
-spec send_overlay_frame(pid(), <<_:256>>, macula_frame:frame()) -> ok | {error, term()}.
send_overlay_frame(Link, TargetPeer, Frame) ->
macula_station_link:send_overlay_frame(Link, TargetPeer, Frame).
%% @doc Open a LOCAL in-process client-stream or bidi call. Used
%% for unit tests and same-BEAM dispatch via `macula_stream_local'.
-spec open_stream(procedure(), term(), map()) ->
{ok, stream()} | {error, term()}.
open_stream(Procedure, Args, Opts)
when is_binary(Procedure), is_map(Opts) ->
macula_stream_local:open_stream(Procedure, Args, Opts).
%% @doc Open a LOCAL in-process stream with explicit mode.
-spec open_stream(procedure(), term(), map(), stream_mode()) ->
{ok, stream()} | {error, term()}.
open_stream(Procedure, Args, Opts, Mode)
when is_binary(Procedure), is_map(Opts), is_atom(Mode) ->
macula_stream_local:open_stream(Procedure, Args, Opts#{mode => Mode}).
%% @doc Advertise a LOCAL in-process streaming procedure
%% (default: server_stream).
-spec advertise_stream(procedure(), stream_handler()) -> ok | {error, term()}.
advertise_stream(Procedure, Handler)
when is_binary(Procedure), is_function(Handler, 2) ->
advertise_stream(Procedure, server_stream, Handler).
%% @doc Advertise a LOCAL in-process streaming procedure with mode.
-spec advertise_stream(procedure(), stream_mode(), stream_handler()) ->
ok | {error, term()}.
advertise_stream(Procedure, Mode, Handler)
when is_binary(Procedure), is_atom(Mode), is_function(Handler, 2) ->
macula_stream_local:advertise(Procedure, Mode, Handler).
%% @doc Register a streaming procedure handler on a V2 pool. Fans out to
%% every healthy link and stores in pool state for replay on link
%% respawn. A caller reaches this provider only through a
%% `procedure_advertisement' record that names it; registering the
%% handler publishes none. See `macula_client:advertise_stream/5'.
-spec advertise_stream(pool(), realm(), procedure(),
stream_mode(), stream_handler()) ->
ok | {error, term()}.
advertise_stream(Pool, Realm, Procedure, Mode, Handler)
when is_pid(Pool), is_binary(Realm), byte_size(Realm) =:= 32,
is_binary(Procedure),
(Mode =:= server_stream orelse Mode =:= client_stream
orelse Mode =:= bidi),
is_function(Handler, 2) ->
advertise_stream(Pool, Realm, Procedure, Mode, Handler, #{}).
%% @doc As `advertise_stream/5', with `Opts'. `auth' sets the streaming
%% procedure's policy, the same set `advertise/5' takes: `open' (default),
%% `{ucan_required, IssuerNodeId}' or `{realm_member_required, RealmKeyId,
%% RequiredCan}'. A consumer presents its token with `call_stream/5''s
%% `ucan_token' opt. `stations' registers it on the links to those stations
%% only, as `advertise/5' does.
-spec advertise_stream(pool(), realm(), procedure(),
stream_mode(), stream_handler(), map()) ->
ok | {error, term()}.
advertise_stream(Pool, Realm, Procedure, Mode, Handler, Opts)
when is_pid(Pool), is_binary(Realm), byte_size(Realm) =:= 32,
is_binary(Procedure),
(Mode =:= server_stream orelse Mode =:= client_stream
orelse Mode =:= bidi),
is_function(Handler, 2), is_map(Opts) ->
on_stations(stations_opt(Opts), fun(Stations) ->
Policy = maps:get(auth, Opts, open),
advertise_authorized(Pool, Realm, Procedure, Opts, fun(Spec) ->
macula_client:advertise_stream(Pool, Realm, Procedure, Mode, Handler,
Policy, Spec, Stations, renewal(Realm, Procedure, Opts))
end)
end).
%% The provider-authorization resolution behind `advertise' and
%% `advertise_stream': the pool's own D25 chain, fetched from the DHT,
%% signed and verified before a single frame goes out. Any missing
%% piece fails fast under `provider_authorization'. What reaches the
%% pool is the advertisement spec, not the signed check: each link
%% signs its own advertisement naming the station it is connected to,
%% bounded by the chain (macula_station_link:advertisement_spec()).
advertise_authorized(Pool, Realm, Procedure, Opts, Fun) ->
confidential(advertise_confidentiality(Opts), fun(Confidentiality) ->
signed_provider_advertisement(Pool, Realm, Procedure, provider_io(Opts),
fun(Spec) -> Fun(maps:merge(Spec, Confidentiality)) end)
end).
%% Whether an advertisement names this node's KEM key (E2E design §8.2, Amendment A1), as spec fields: `kem' names
%% it, and `confidential => required' also refuses every clear call. A key is named only once the node is switched on
%% with the `kem_advertise' application setting, which stays `disabled' until every station runs the release that
%% stores a keyed advertisement, and only when `confidential' is not `off'. `required' while switched off is refused:
%% the provider would name no key and refuse every clear call, and so be unreachable.
%% @private
%% @doc What an advertise's `confidential' option and this node's `kem_advertise' switch decide, as the spec fields an
%% advertisement carries: `kem' when it names the KEM key, `confidential => required' when it also refuses clear
%% calls. The pool's ADVERTISE and a direct-dial DHT record decide the same way (`macula_direct_dial').
-spec advertise_confidentiality(map()) -> {ok, #{kem => true, confidential => required}} | {error, term()}.
advertise_confidentiality(Opts) ->
confidentiality(maps:get(confidential, Opts, preferred), kem_advertise()).
confidentiality(Mode, _Switch) when Mode =/= preferred, Mode =/= required, Mode =/= off ->
{error, {confidentiality, {not_a_mode, Mode}}};
confidentiality(required, disabled) -> {error, {confidentiality, kem_advertise_disabled}};
confidentiality(off, _Switch) -> {ok, #{}};
confidentiality(preferred, disabled) -> {ok, #{}};
confidentiality(preferred, enabled) -> {ok, #{kem => true}};
confidentiality(required, enabled) -> {ok, #{kem => true, confidential => required}}.
confidential({ok, Confidentiality}, Fun) -> Fun(Confidentiality);
confidential({error, _} = Refused, _Fun) -> Refused.
%% The node's `kem_advertise' switch: `disabled' unless set to `enabled'. A value outside the two raises, loudly, rather
%% than quietly meaning off.
kem_advertise() ->
kem_switch(application:get_env(macula, kem_advertise, disabled)).
kem_switch(enabled) -> enabled;
kem_switch(disabled) -> disabled;
kem_switch(Other) -> erlang:error({kem_advertise, {not_a_switch, Other}}).
%% How the pool renews the chain an advertisement carries (D32): the same
%% resolution, over the same `provider_io/0' seams the advertise was given,
%% named as an MFA the pool applies with itself prepended. Only the seam
%% entries travel, and in production there are none.
renewal(Realm, Procedure, Opts) ->
{?MODULE, renew_authorization,
[Realm, Procedure, maps:with([status, find_record, sign_node_record, realm_key, confidential], Opts)]}.
%% @private
%% @doc The pool's renewal of an advertised chain (D32, macula#38): the chain
%% resolved, signed and verified again as `advertise/5' does, answered as the
%% fresh advertisement spec, or the same refusal `advertise/5' would give. A
%% `procedure_delegation' lives 30 minutes, so the pool calls this at a third
%% of the spec's remaining life, from a worker of its own. Internal: called
%% through the MFA `advertise/5' registers.
-spec renew_authorization(pool(), realm(), procedure(), map()) ->
{ok, macula_station_link:advertisement_spec()} | {error, term()}.
renew_authorization(Pool, Realm, Procedure, Seams) ->
advertise_authorized(Pool, Realm, Procedure, Seams, fun(Spec) -> {ok, Spec} end).
%% @doc Resolve this pool's own D25 provider authorization for an
%% org-namespaced `Procedure' under `Realm' — the realm-signed
%% `org_directory' and the org-signed `procedure_delegation' naming the
%% pool's node id, both fetched from the DHT and verified against the
%% realm key the pool pinned for `Realm' at connect — as the
%% `authorization' opt `macula_direct_dial:publish_advertisement/5'
%% and `macula_response:advertise_direct/6,7' take: a map of the two
%% records' encoded wire forms. `advertise/5' resolves this same chain
%% itself for the wire frame; a caller that publishes the direct-dial
%% record needs it explicitly, since the station refuses an
%% org-namespaced record without one (`no_authorization').
%%
%% Fails fast with `{error, {provider_authorization, _}}' when the
%% procedure has no org namespace, a chain piece is missing from the
%% DHT, the pool pinned no key for `Realm', or the chain does not
%% verify against that key — the same failures `advertise/5' reports.
%%
%% A procedure in this node's own namespace, `~<own node_id>/<name>', needs
%% no chain (D25 item 6, revised 2026-09-24): the answer is `{ok, undefined}',
%% which `publish_advertisement/5' reads as no authorization. Another node's
%% namespace is `{error, {provider_authorization, not_own_namespace}}'.
-spec provider_authorization(pool(), realm(), procedure()) ->
{ok, #{org_directory := binary(), procedure_delegation := binary()} | undefined} |
{error, term()}.
provider_authorization(Pool, Realm, Procedure) ->
provider_authorization(Pool, Realm, Procedure, #{}).
%% @doc As `provider_authorization/3', reading the resolution's DHT
%% calls from the `provider_io/0' seam entries in `Opts' (defaults to
%% the facade's own functions) — the seam `advertise/5' accepts too.
-spec provider_authorization(pool(), realm(), procedure(), provider_io()) ->
{ok, #{org_directory := binary(), procedure_delegation := binary()} | undefined} |
{error, term()}.
provider_authorization(Pool, Realm, Procedure, Opts) ->
case signed_provider_advertisement(Pool, Realm, Procedure,
provider_io(Opts),
fun(#{authorization := A}) -> A;
(#{}) -> undefined
end) of
{error, _} = E -> E;
Authorization -> {ok, Authorization}
end.
provider_io(Opts) ->
#{status => maps:get(status, Opts, fun status/1),
find_record => maps:get(find_record, Opts, fun find_record/2),
sign_node_record => maps:get(sign_node_record, Opts,
fun sign_node_record/3),
realm_key => maps:get(realm_key, Opts,
fun macula_client:realm_key/2)}.
%% The pool's own D25 chain, resolved from the DHT, signed and verified
%% against the realm key the pool pins, passed to `Fun' as the
%% advertisement spec: the verified authorization and the bound no
%% advertisement carrying it may pass. Any missing piece fails fast
%% under `provider_authorization'.
signed_provider_advertisement(Pool, Realm, Procedure, Io, Fun) ->
case macula_record:procedure_org(Procedure) of
none ->
{error, {provider_authorization, no_org_namespace}};
{error, malformed} ->
{error, {provider_authorization, malformed_procedure}};
{org, <<"~", _/binary>>} ->
case (maps:get(status, Io))(Pool) of
{ok, #{self_node_id := NodeId}} ->
own_namespace_spec(Realm, Procedure, NodeId, Fun);
{error, _} = E ->
E
end;
{org, Org} ->
case (maps:get(status, Io))(Pool) of
{ok, #{self_node_id := NodeId}} ->
resolve_org_directory(Io, Pool, Realm, Org, Procedure,
NodeId, Fun);
{error, _} = E ->
E
end
end.
%% A procedure in this node's own namespace, `~<own node_id>/<name>', is
%% authorized by the advertisement's own signature (D25 item 6, revised
%% 2026-09-24): no org directory, no delegation, no bound but the
%% advertisement's own lifetime. What each link signs is checked by the
%% same rule every verifier uses, before any spec reaches the pool; another
%% node's namespace is refused here.
own_namespace_spec(Realm, Procedure, NodeId, Fun) ->
own_namespace_checked(
macula_record:own_namespace(
macula_record:procedure_advertisement(NodeId, Realm, Procedure, NodeId)),
Fun).
own_namespace_checked(ok, Fun) ->
Fun(#{});
own_namespace_checked({error, Reason}, _Fun) ->
{error, {provider_authorization, Reason}}.
resolve_org_directory(Io, Pool, Realm, Org, Procedure, NodeId, Fun) ->
Find = maps:get(find_record, Io),
case Find(Pool, macula_record:org_directory_key(Realm, Org)) of
{ok, OrgDir} ->
#{org_key := OrgKeyId} =
macula_record:read_org_directory(OrgDir),
resolve_delegation(Io, Pool, Realm, Procedure, OrgDir, OrgKeyId,
NodeId, Fun);
{error, not_found} ->
{error, {provider_authorization, {org_directory, not_found}}};
{error, _} = E ->
{error, {provider_authorization, E}}
end.
resolve_delegation(Io, Pool, Realm, Procedure, OrgDir, OrgKeyId, NodeId,
Fun) ->
Find = maps:get(find_record, Io),
case Find(Pool, macula_record:procedure_delegation_key(OrgKeyId, NodeId)) of
{ok, Deleg} ->
sign_provider_advertisement(Io, Pool, Realm, Procedure, OrgDir,
Deleg, NodeId, Fun);
{error, not_found} ->
{error, {provider_authorization,
{procedure_delegation, not_found}}};
{error, _} = E ->
{error, {provider_authorization, E}}
end.
%% The advertisement is signed under a bound: it must not outlive what
%% authorizes it. An advertisement is built with its type's own
%% lifetime, which knows nothing about the chain it carries, and both
%% verifiers refuse one that ends after the earlier of the org
%% directory's and the delegation's expiry —
%% `macula_record:delegation_matched/3' here, and macula-station's at
%% admission. Without the bound a provider spends the last stretch of
%% every authorization window signing advertisements its own pool
%% refuses, and goes dark before its delegation expires.
sign_provider_advertisement(Io, Pool, Realm, Procedure, OrgDir, Deleg,
NodeId, Fun) ->
Authorization = #{org_directory => macula_record:encode(OrgDir),
procedure_delegation => macula_record:encode(Deleg)},
Unsigned = macula_record:procedure_advertisement(
NodeId, Realm, Procedure, NodeId,
#{authorization => Authorization}),
NotAfter = min(macula_record:expires_at(OrgDir),
macula_record:expires_at(Deleg)),
case (maps:get(sign_node_record, Io))(Pool, Unsigned,
#{not_after => NotAfter}) of
{ok, Signed} ->
trusted_provider_advertisement(
Io, Pool, Realm, Signed,
fun(Verified) ->
#{authorization := A} =
macula_record:read_procedure_advertisement(Verified),
Fun(#{authorization => A, not_after => NotAfter})
end);
{error, _} = E ->
{error, {provider_authorization, E}}
end.
%% The advertisement goes out only after its authorization verifies
%% against the realm key this pool pins — a chain the pool cannot
%% check never reaches a station.
trusted_provider_advertisement(Io, Pool, Realm, Signed, Fun) ->
case (maps:get(realm_key, Io))(Pool, Realm) of
{ok, RealmKey} ->
{ok, Profile} = macula_crypto_profile:configured(),
Trust = #{profile => Profile, realm_key => RealmKey},
case macula_record:verify_authorization(
Signed, Trust, erlang:system_time(millisecond)) of
ok ->
Fun(Signed);
{error, _} = E ->
{error, {provider_authorization, E}}
end;
none ->
{error, {provider_authorization, no_realm_key}}
end.
%% @doc Stop advertising a LOCAL streaming procedure.
-spec unadvertise_stream(procedure()) -> ok.
unadvertise_stream(Procedure) when is_binary(Procedure) ->
macula_stream_local:unadvertise(Procedure).
%% @doc Stop advertising a streaming procedure on a V2 pool.
-spec unadvertise_stream(pool(), realm(), procedure()) -> ok.
unadvertise_stream(Pool, Realm, Procedure)
when is_pid(Pool), is_binary(Realm), byte_size(Realm) =:= 32,
is_binary(Procedure) ->
macula_client:unadvertise_stream(Pool, Realm, Procedure).
%% @doc Send a binary chunk on the stream.
-spec send(stream(), binary()) -> ok | {error, term()}.
send(Stream, Bin) when is_pid(Stream), is_binary(Bin) ->
macula_stream:send(Stream, Bin).
%% @doc Send a chunk with explicit encoding.
-spec send(stream(), binary() | term(), raw | msgpack) -> ok | {error, term()}.
send(Stream, Body, Encoding) when is_pid(Stream) ->
macula_stream:send(Stream, Body, Encoding).
%% @doc Receive the next chunk (blocks).
-spec recv(stream()) -> {chunk, binary()}
| {data, term()}
| eof
| {error, term()}.
recv(Stream) when is_pid(Stream) ->
macula_stream:recv(Stream).
-spec recv(stream(), timeout()) -> {chunk, binary()}
| {data, term()}
| eof
| {error, term()}.
recv(Stream, Timeout) when is_pid(Stream) ->
macula_stream:recv(Stream, Timeout).
%% @doc Close a V1 stream (both sides). Renamed from `close/1' in
%% 3.11.0 because `close/1' now refers to the V2 pool surface.
-spec close_stream(stream()) -> ok.
close_stream(Stream) when is_pid(Stream) ->
macula_stream:close(Stream).
%% @doc Half-close the write side; recv still drains.
-spec close_send(stream()) -> ok.
close_send(Stream) when is_pid(Stream) ->
macula_stream:close_send(Stream).
%% @doc Wait for the terminal reply (client-stream / bidi).
-spec await_reply(stream()) -> {ok, term()} | {error, term()}.
await_reply(Stream) when is_pid(Stream) ->
macula_stream:await_reply(Stream).
-spec await_reply(stream(), timeout()) -> {ok, term()} | {error, term()}.
await_reply(Stream, Timeout) when is_pid(Stream) ->
macula_stream:await_reply(Stream, Timeout).
%% @doc Server-side: emit the terminal reply value.
-spec set_reply(stream(), term()) -> ok.
set_reply(Stream, Result) when is_pid(Stream) ->
macula_stream:set_reply(Stream, Result).
%% @doc Abort the stream with an error frame.
-spec abort(stream(), binary(), binary()) -> ok | {error, {text_too_long | invalid_text, code}}.
abort(Stream, Code, Message)
when is_pid(Stream), is_binary(Code), is_binary(Message) ->
macula_stream:abort(Stream, Code, Message).
%%%===================================================================
%%% Cluster (LAN)
%%%===================================================================
%% @doc Ensure this node is running in distributed mode.
-spec ensure_distributed() -> ok | {error, term()}.
ensure_distributed() -> macula_cluster:ensure_distributed().
%% @doc Subscribe to node up/down events.
-spec monitor_nodes() -> ok.
monitor_nodes() -> macula_cluster:monitor_nodes().
%% @doc Unsubscribe from node up/down events.
-spec unmonitor_nodes() -> ok.
unmonitor_nodes() -> macula_cluster:unmonitor_nodes().
%%%===================================================================
%%% Mesh Distribution
%%%===================================================================
%% @doc Join the Macula relay mesh with Erlang distribution.
%%
%% After calling this, standard OTP distribution works across firewalls.
%% `Opts' takes:
%% <ul>
%% <li>`relays' (required): the V2 pool's seeds, each a map with `host',
%% `port' and `expected_node_id', the relay's 32-byte node_id, which
%% every dial checks (D16). A relay without one refuses the join with
%% `{error, {relays, expected_node_id_required}}' before any pool
%% starts.</li>
%% <li>`node_identity': the V2 pool's node identity key,
%% `macula_node_keys:node_key()'. Default: the node's one stored
%% identity, as for `connect/2'.</li>
%% </ul>
%%
%% Internally builds a V2 `macula_client:pool()' and registers it
%% with `macula_dist_pool' as the carrier for `_dist.tunnel.*'
%% traffic. Dist tunnel frames travel under the all-zeros realm
%% (protocol-internal infrastructure, not bound to any user realm).
-spec join_mesh(map()) -> ok | {error, term()}.
join_mesh(Opts) ->
joined(join_pool_args(Opts)).
joined({ok, Relays, PoolOpts}) ->
on_pool_for_join(macula_client:connect(Relays, PoolOpts));
joined({error, _} = Refusal) ->
Refusal.
%% The seeds and options of the pool a join starts: the relays, when every one names the node_id it expects, and the
%% node identity key when one is given.
join_pool_args(#{relays := Relays} = Opts) ->
joinable(seeds_checked(Relays, Opts), Relays, Opts).
%% A relay carrying `pin_tls_cert => true' or `verify' is refused here
%% rather than at link start, so a join reports it instead of leaving a pool respawning a
%% seed that can never start.
joinable(ok, Relays, Opts) ->
pinned_relays(lists:all(fun pinned_relay/1, Relays), Relays,
maps:with([node_identity], Opts));
joinable({error, _} = Refusal, _Relays, _Opts) ->
Refusal.
pinned_relays(true, Relays, PoolOpts) -> {ok, Relays, PoolOpts};
pinned_relays(false, _Relays, _PoolOpts) -> {error, {relays, expected_node_id_required}}.
pinned_relay(#{host := _, port := _, expected_node_id := <<_:256>>}) -> true;
pinned_relay(_Relay) -> false.
on_pool_for_join({ok, Pool}) ->
wait_for_pool(Pool, 30),
os:putenv("MACULA_DIST_MODE", "relay"),
macula_dist_pool:register_mesh_pool(Pool),
macula_dist_pool:advertise_dist_accept(),
?LOG_INFO("[macula] Joined mesh — distribution enabled"),
ok;
on_pool_for_join({error, Reason}) ->
?LOG_ERROR("[macula] Failed to join mesh: ~p", [Reason]),
{error, Reason}.
%% @doc Enable Erlang distribution over a dedicated dist relay
%% (`macula-io/macula-dist-relay').
%%
%% Different from `join_mesh/1':
%% - Connects to a dist relay (port 4434, ALPN `macula-dist'), NOT the
%% pub/sub station mesh
%% - No mesh_client, no pub/sub subscriptions — only dist traffic
%% - Uses raw QUIC stream routing with no MessagePack overhead
%%
%% Options:
%% - `url' (required): `<<"quic://relay.example.com:4434">>'
%%
%% After this returns `ok', standard OTP distribution (`rpc:call/4',
%% `gen_server:call/3' across nodes, `pg' groups, etc.) works across
%% firewalls via the dist relay.
%%
%% The relay client runs as a temporary child of the macula application
%% supervisor. It does not reconnect: when the relay closes the
%% connection the client ends and is not restarted. Monitor the pid from
%% `dist_relay_client/0' to learn that, then call this function again.
%% Returns `{error, macula_not_started}' when the macula application is
%% not running.
-spec join_dist_relay(map()) -> ok | {error, term()}.
join_dist_relay(Opts) ->
Url = maps:get(url, Opts),
NodeName = atom_to_binary(node()),
start_dist_relay_client(whereis(macula_root), Url, NodeName).
start_dist_relay_client(undefined, _Url, _NodeName) ->
{error, macula_not_started};
start_dist_relay_client(_Root, Url, NodeName) ->
ChildSpec = macula_dist_relay_client:child_spec(Url, NodeName),
case supervisor:start_child(macula_root, ChildSpec) of
{ok, _Pid} ->
os:putenv("MACULA_DIST_MODE", "dist_relay"),
?LOG_INFO("[macula] Joined dist relay ~s, distribution enabled", [Url]),
ok;
{error, {already_started, _Pid}} ->
os:putenv("MACULA_DIST_MODE", "dist_relay"),
?LOG_INFO("[macula] dist_relay_client already running, mode set"),
ok;
{error, Reason} ->
Refusal = child_start_refusal(Reason),
?LOG_ERROR("[macula] Failed to join dist relay: ~p", [Refusal]),
{error, Refusal}
end.
%% `supervisor:start_child/2' wraps a child's own refusal with the child
%% spec that failed to start. A caller of `join_dist_relay/1' wants the
%% refusal, not the spec it already gave us: the client refuses when its
%% operator has not accepted an unidentified peer
%% (`macula_dist:unidentified_peer_refusal/0'), and that term says so.
child_start_refusal({Refusal, Child}) when is_tuple(Child), element(1, Child) =:= child ->
Refusal;
child_start_refusal(Reason) ->
Reason.
%% @doc The dist relay client that `join_dist_relay/1' started, if it
%% is running.
%%
%% The client exits with `{relay_closed, Reason}' when the relay closes
%% the connection and is not restarted. Monitor the returned pid and call
%% `join_dist_relay/1' again after it goes down.
-spec dist_relay_client() -> {ok, pid()} | {error, not_joined}.
dist_relay_client() ->
dist_relay_client_result(macula_dist_relay_client:whereis_client()).
dist_relay_client_result(undefined) -> {error, not_joined};
dist_relay_client_result(Pid) -> {ok, Pid}.
%% @private Wait until the V2 pool has at least one healthy
%% station_link (CONNECT/HELLO completed). One-second polling, capped
%% at `Retries' iterations. A handshake, not a route: see `healthy_links'
%% in `macula_client:status/1'.
wait_for_pool(_Pool, 0) ->
?LOG_WARNING("[macula] Mesh pool not ready after timeout");
wait_for_pool(Pool, Retries) ->
on_pool_status(macula_client:status(Pool), Pool, Retries).
on_pool_status({ok, #{healthy_links := N}}, _Pool, _Retries) when N > 0 ->
?LOG_INFO("[macula] Mesh pool connected (~p healthy link(s))", [N]),
ok;
on_pool_status(_Other, Pool, Retries) ->
timer:sleep(1000),
wait_for_pool(Pool, Retries - 1).
%%%===================================================================
%%% Peer-supplied maps (D26)
%%%===================================================================
%% @doc A field of a map a peer supplied, or `undefined' when it is absent. See `field/3'.
-spec field(atom() | binary(), map()) -> term().
field(Name, Map) ->
field(Name, Map, undefined).
%% @doc A field of a map a peer supplied (D26), or `Default' when it is absent. A map from the codec carries its text
%% keys as `{text, Bin}'; a map handed over in process may carry atom or binary keys. The lookup tries `{text, Name}',
%% then the atom, then the binary, so a handler reads both kinds of map the same way. Looking up a binary name never
%% creates an atom.
-spec field(atom() | binary(), map(), term()) -> term().
field(Name, Map, Default) when (is_atom(Name) orelse is_binary(Name)), is_map(Map) ->
first_found(field_keys(Name), Map, Default).
%% @doc The binary of a text value a peer supplied (D26): the binary of `{text, Bin}', a binary unchanged, and
%% `badarg' for anything else.
-spec text({text, binary()} | binary()) -> binary().
text({text, Bin}) when is_binary(Bin) -> Bin;
text(Bin) when is_binary(Bin) -> Bin;
text(Other) -> erlang:error(badarg, [Other]).
field_keys(Name) when is_atom(Name) ->
Bin = atom_to_binary(Name),
[{text, Bin}, Name, Bin];
field_keys(Name) when is_binary(Name) ->
[{text, Name}] ++ existing_atom(Name) ++ [Name].
existing_atom(Name) ->
try [binary_to_existing_atom(Name)]
catch error:badarg -> []
end.
first_found([], _Map, Default) -> Default;
first_found([Key | Keys], Map, Default) -> found(maps:find(Key, Map), Keys, Map, Default).
found({ok, Value}, _Keys, _Map, _Default) -> Value;
found(error, Keys, Map, Default) -> first_found(Keys, Map, Default).
%%------------------------------------------------------------------
%% `pin_tls_cert' and `verify' are refused, not accepted and ignored
%%------------------------------------------------------------------
%% Every public entry point that takes a seed, a station or a per-dial TLS
%% trust map runs this before it does anything else: `connect/2',
%% `call_station/7,8', `call_stream_station/7' and `join_mesh/1'.
%% `pin_tls_cert => true' is refused, and `verify' in any value;
%% `pin_tls_cert => false' and absence
%% pass through.
%%
%% WHY NEITHER CAN BE HONOURED, so the next reader does not re-litigate it.
%% A dial trusts a station in exactly one way (plan decisions D12 and D16):
%% the station presents one self-signed ML-DSA-87 certificate on its TLS
%% key, its handshake signature must verify under that key, and its
%% identity is proved by the signed handshake against `expected_node_id',
%% which is mandatory on a client dial. `pin_tls_cert' once mapped
%% `expected_node_id' onto a pinned Ed25519 key, which stopped meaning
%% anything when a node_id became a hash over an ML-DSA-87 identity key and
%% the TLS key became a key of its own; the pin primitive is gone from the
%% QUIC NIF. `verify' chose between a webpki chain check, which no ML-DSA
%% certificate can pass because no authority issues one, and no check at
%% all, which the NIF no longer offers.
%%
%% So the choice is between accepting a value that does nothing and saying
%% so. An inert security option is worse than an absent one, because a
%% reader takes it for a check that is happening: that is macula#15, and it
%% reached hexdocs as a documented default of `true' that was never true.
%% `pin_tls_cert => false' still passes, as it did through 11.x; `verify'
%% is refused in any value, since 12.0.0 is where it went.
%%
%% ⚠ EVERY map a caller can put the keys in, not just `Opts'. The seed and
%% station maps are where the options historically lived, so checking only
%% `Opts' advertises a check that does not happen on the path a reader of
%% our own CHANGELOG would take.
%%
%% The refusal is wrapped in `refused' because nothing is sent and every
%% candidate refuses it identically, which is `request' scope in the dial
%% taxonomy. Left bare it would fall to a catch-all and be classified
%% `provider', meaning "may have reached a provider", which is the wrong
%% scope for a refusal that sent nothing.
%%
%% ⚠ MAPS ONLY. A seed may be a charlist (`seed()' includes `string()'),
%% so an `is_list' clause here would walk `"quic://..."' character by
%% character and be correct only by accident, every integer falling to the
%% catch-all. The list is mapped over by `seeds_checked/2', which knows it
%% has a list OF seeds rather than a seed that happens to be a list.
-spec trust_options_checked(map() | macula_client:seed()) ->
ok | {error, {refused, {pin_tls_cert | verify, atom()}}}.
trust_options_checked(#{pin_tls_cert := true}) ->
{error, {refused, {pin_tls_cert, no_pin_primitive_for_mldsa87_identity}}};
trust_options_checked(#{verify := _}) ->
{error, {refused, {verify, one_verification_mode}}};
trust_options_checked(_NoRefusedOption) ->
ok.
%% `connect/2' and `join_mesh/1': a LIST of seeds, plus the option map.
-spec seeds_checked([macula_client:seed()], map()) ->
ok | {error, {refused, {pin_tls_cert | verify, atom()}}}.
seeds_checked(Seeds, Opts) when is_list(Seeds) ->
first_error([trust_options_checked(S) || S <- Seeds] ++
[trust_options_checked(Opts)]).
%% One station or seed, plus the option map, in the order the public
%% functions receive them.
-spec target_checked(macula_client:seed(), map()) ->
ok | {error, {refused, {pin_tls_cert | verify, atom()}}}.
target_checked(Target, Opts) ->
first_error([trust_options_checked(Target), trust_options_checked(Opts)]).
%% Named `first_error' and not `first_refusal': `macula_station_link' has a
%% `first_refusal/1' of its own with a different shape (it folds `none' and
%% a refusal term, not `ok' and an `{error, _}'). Two functions of the same
%% name meaning different things in one module family is the small version
%% of a contract in two places.
first_error([{error, _} = Refusal | _Rest]) -> Refusal;
first_error([ok | Rest]) -> first_error(Rest);
first_error([]) -> ok.
%% Run the call, or return the refusal that stopped it.
refused(ok, Call) -> Call();
refused({error, _} = Refusal, _Call) -> Refusal.