Packages
macula
3.16.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.
%%%
%%% == V2 (canonical, since 3.11.0) ==
%%%
%%% Apps connect via `connect/2', which returns a `macula_client'
%%% pool that internally wraps N peering links to N stations.
%%% `publish/4', `publish/5', `subscribe/4', `subscribe/5', and
%%% `unsubscribe/2' route through the pool with realm-per-call
%%% semantics. See `macula_pubsub' for the slice module of the same
%%% surface.
%%%
%%% == V1 (legacy, retired at 4.0.0) ==
%%%
%%% Older apps used a single-connection client (`macula_mesh_client').
%%% V1 facade surfaces still here untouched: `subscribe/3',
%%% `publish/3', `call/3,4', `advertise/3,4', `unadvertise/2',
%%% `put_record/2', plus all stream and directed-RPC operations.
%%% V1 callers wishing to keep V1 semantics for connect/publish/4/
%%% unsubscribe/close after upgrading to 3.11.0 should call
%%% `macula_mesh_client' / `macula_stream_v1' directly — see the
%%% migration guide.
%%%
%%% @end
%%%-------------------------------------------------------------------
-module(macula).
-include_lib("kernel/include/logger.hrl").
%% Connection — V2 (pool)
-export([connect/2, close/1, child_spec/3, status/1]).
%% Connection — V1 (legacy)
-export([disconnect/1]).
%% Pub/Sub — V2 (realm-per-call) + V1 (no-realm)
-export([subscribe/3, subscribe/4, subscribe/5,
subscribe_callback/4,
unsubscribe/2,
publish/3, publish/4, publish/5]).
%% RPC — V1 (single-conn client) + V2 (pool, since 3.16.0)
-export([call/3, call/4, call/5,
advertise/3, advertise/4, advertise/5,
unadvertise/2, unadvertise/3]).
%% Signed DHT records (v3.2.0)
-export([put_record/2,
find_record/2,
find_records_by_type/2,
subscribe_records/3,
unsubscribe_records/2]).
%% Streaming RPC (v1.5.0+ — see PLAN_MACULA_STREAMING.md)
%% NOTE: V1 stream `close/1' renamed to `close_stream/1' in 3.11.0
%% to free `close/1' for the V2 pool surface.
-export([
call_stream/2, call_stream/3, call_stream/4,
open_stream/3, open_stream/4,
advertise_stream/2, advertise_stream/3, advertise_stream/4,
unadvertise_stream/1,
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
]).
%% Directed RPC (Mesh Name Service)
-export([call_node/4, call_node/5, resolve/2, list_nodes/1, list_nodes/2]).
%% Cluster (LAN)
-export([ensure_distributed/0, get_cookie/0, set_cookie/1,
monitor_nodes/0, unmonitor_nodes/0]).
%% Mesh Distribution
-export([join_mesh/1, join_dist_relay/1]).
%% Types
-export_type([client/0, pool/0, realm/0,
topic/0, procedure/0,
stream/0, stream_mode/0, stream_handler/0,
record/0, record_type/0, record_key/0]).
-type client() :: pid(). %% V1 client (`macula_mesh_client').
-type pool() :: macula_client:pool(). %% V2 pool (`macula_client').
-type realm() :: <<_:256>>. %% V2 32-byte realm tag.
-type topic() :: binary().
-type procedure() :: binary().
-type stream() :: pid().
-type stream_mode() :: server_stream | client_stream | bidi.
-type stream_handler() :: fun((stream(), term()) -> any()).
-type record() :: macula_record: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 relay endpoints (URL binaries/strings or
%% `#{host, port}' maps). 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>`identity' — pool's Ed25519 keypair; auto-generated if absent.</li>
%% <li>`replication_factor' — links per PUBLISH (default 1).</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_window_ms', `dedup_sweep_ms' — inbound-EVENT dedup tunables.</li>
%% </ul>
%%
%% V1-only opts that have NO V2 equivalent and are silently ignored
%% (with a `logger:notice') for callers migrating from
%% `macula_multi_relay':
%% <ul>
%% <li>`relays' — superseded by the `Seeds' positional argument.</li>
%% <li>`realm' — V2 is realm-per-call; pass realm to each
%% `publish/4' / `subscribe/4' / `call/5' invocation instead.</li>
%% <li>`site' — V1 routing-affinity hint with no V2 analog.</li>
%% <li>`connections' — V1 was N connections per relay; V2 is one
%% link per seed (add more seeds to grow the pool).</li>
%% </ul>
%%
%% See `macula_client' for the canonical pool implementation and
%% `macula_pubsub' for the slice module.
%%
%% **Breaking change since 3.11.0**: this function previously
%% returned a `macula_mesh_client' single-connection client. V1
%% callers must now call `macula_mesh_client:start_link/1' directly
%% to retain V1 semantics. See `CHANGELOG.md' / migration guide.
-spec connect([macula_client:seed()], macula_client:opts()) ->
{ok, pool()} | {error, term()}.
connect(Seeds, Opts) when is_list(Seeds), is_map(Opts) ->
macula_client:connect(Seeds, Opts).
%% @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.
-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).
%%%===================================================================
%%% Connection — V1 (legacy)
%%%===================================================================
%% @doc Disconnect a V1 `macula_mesh_client' client. **Legacy** —
%% V2 pools use `close/1'.
-spec disconnect(client()) -> ok.
disconnect(Client) when is_pid(Client) ->
macula_mesh_client:stop(Client).
%%%===================================================================
%%% Pub/Sub — V1 (legacy)
%%%===================================================================
%% @doc V1 subscribe — single-connection client, no realm. **Legacy**.
%% V2 callers use `subscribe/4' or `subscribe/5'.
-spec subscribe(client(), topic(), fun((term()) -> ok) | pid()) ->
{ok, reference()} | {error, term()}.
subscribe(Client, Topic, Callback) when is_pid(Client), is_binary(Topic) ->
macula_mesh_client:subscribe(Client, Topic, Callback).
%% @doc V1 publish — single-connection client, no realm. **Legacy**.
%% V2 callers use `publish/4' or `publish/5'.
-spec publish(client(), topic(), term()) -> ok.
publish(Client, Topic, Data) when is_pid(Client), is_binary(Topic) ->
macula_mesh_client:publish(Client, Topic, Data).
%%%===================================================================
%%% Pub/Sub — V2 (pool, realm-per-call, since 3.11.0)
%%%===================================================================
%% @doc Publish to `(Realm, Topic)' on `Pool'. Equivalent to
%% `publish/5' with empty opts.
%%
%% **Breaking change since 3.11.0**: this signature previously was
%% `publish(Client, Topic, Data, Opts)' (V1). V1 callers using the
%% 4-arg form must call `macula_mesh_client:publish/3' directly to
%% retain V1 semantics.
-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()}.
subscribe(Pool, Realm, Topic, Subscriber) ->
macula_pubsub:subscribe(Pool, Realm, Topic, Subscriber).
%% @doc Subscribe `Subscriber' to `(Realm, Topic)' on `Pool' with
%% options. See `macula_pubsub:subscribe/5'.
-spec subscribe(pool(), realm(), topic(), pid(), map()) ->
{ok, reference()}.
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 V2 pool subscription. Idempotent.
%%
%% **Breaking change since 3.11.0**: this routes to the V2 pool
%% (`macula_client:unsubscribe/2'). V1 callers must call
%% `macula_mesh_client:unsubscribe/2' directly to drop a V1 sub.
-spec unsubscribe(pool(), reference()) -> ok.
unsubscribe(Pool, SubRef) when is_pid(Pool), is_reference(SubRef) ->
macula_pubsub:unsubscribe(Pool, SubRef).
%%%===================================================================
%%% RPC
%%%===================================================================
%% @doc Call a remote procedure (default 5s timeout).
-spec call(client(), procedure(), term()) -> {ok, term()} | {error, term()}.
call(Client, Procedure, Args) ->
call(Client, Procedure, Args, 5000).
%% @doc Call a remote procedure with timeout.
-spec call(client(), procedure(), term(), pos_integer()) ->
{ok, term()} | {error, term()}.
call(Client, Procedure, Args, Timeout) when is_pid(Client), is_binary(Procedure) ->
macula_mesh_client:call(Client, Procedure, Args, Timeout).
%% @doc Advertise an RPC procedure handler.
-spec advertise(client(), procedure(), fun()) -> {ok, reference()} | {error, term()}.
advertise(Client, Procedure, Handler) ->
advertise(Client, Procedure, Handler, #{}).
%% @doc Advertise with options.
-spec advertise(client(), procedure(), fun(), map()) -> {ok, reference()} | {error, term()}.
advertise(Client, Procedure, Handler, _Opts) when is_pid(Client), is_binary(Procedure) ->
macula_mesh_client:advertise(Client, Procedure, Handler).
%% @doc Stop advertising a V1 procedure.
-spec unadvertise(client(), procedure()) -> ok | {error, term()}.
unadvertise(Client, Procedure) when is_pid(Client), is_binary(Procedure) ->
macula_mesh_client:unadvertise(Client, Procedure).
%%%===================================================================
%%% RPC — V2 (pool, since 3.16.0)
%%%===================================================================
%% @doc Issue a CALL frame against a V2 pool. First-success across
%% the pool's healthy links. See `macula_client:call/5'.
-spec call(pool(), realm(), procedure(), term(), pos_integer()) ->
{ok, term()} | {error, term()}.
call(Pool, Realm, Procedure, Payload, TimeoutMs) ->
macula_client:call(Pool, Realm, Procedure, Payload, TimeoutMs).
%% @doc Advertise a procedure handler on a V2 pool. Fans out to every
%% healthy link and stores in pool state for replay on link respawn.
%% See `macula_client:advertise/4'. Equivalent to `advertise/5' with
%% empty opts.
%%
%% Arity 5 (not 4) so V2 calls do not collide with the legacy V1
%% `advertise(Client, Procedure, Handler, Opts)' surface.
-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 ->
macula_client:advertise(Pool, Realm, Procedure, Handler).
%% @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, signed payloads stored in the relay mesh's
%%% distributed hash table. The record format follows Macula V2
%%% spec Part 6 §9 (PKARR-compatible CBOR with single-letter keys
%%% `t', `k', `v', `c', `x', `p', `s'), Part 6 §10.2 (signing
%%% domain `"macula-v2-record\\0" || canonical_cbor(unsigned)'),
%%% and Part 3 §3.3 (domain-separated storage keys).
%%%
%%% See `macula_record' for the record shape, the typed
%%% constructors (`node_record/3', `realm_directory/3',
%%% `realm_stations/2', `procedure_advertisement/3',
%%% `content_announcement/3', `tombstone/3', 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.
-define(DHT_PUT_RECORD_PROC, <<"_dht.put_record">>).
-define(DHT_FIND_RECORD_PROC, <<"_dht.find_record">>).
-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.
%%
%% Build the record via the typed constructors in `macula_record'
%% (`node_record/3,4', `content_announcement/3,4', `tombstone/3,4',
%% `realm_directory/3,4', `procedure_advertisement/3,4', etc.) then
%% sign it with `macula_record:sign/2'. The relay validates the
%% signature on receipt; an invalid signature returns
%% `{error, bad_signature}'. Successful stores propagate to the
%% K-nearest peers in the DHT under the record's
%% `macula_record:storage_key/1'.
-spec put_record(client(), record()) -> ok | {error, term()}.
put_record(Client, Record) when is_pid(Client), is_map(Record) ->
classify_put(macula_mesh_client:call(Client, ?DHT_PUT_RECORD_PROC,
Record, ?DHT_RECORD_TIMEOUT_MS)).
classify_put({ok, ok}) -> ok;
classify_put({ok, Reply}) -> {error, {unexpected_reply, Reply}};
classify_put({error, _} = E) -> E.
%% @doc Fetch a record by its `macula_record:storage_key/1'.
%%
%% Returns `{error, not_found}' when no record exists at the key.
%% The returned record's signature should be verified via
%% `macula_record:verify/1' before its payload is trusted.
-spec find_record(client(), record_key()) ->
{ok, record()} | {error, not_found | term()}.
find_record(Client, Key)
when is_pid(Client), is_binary(Key), byte_size(Key) =:= 32 ->
classify_find(macula_mesh_client:call(Client, ?DHT_FIND_RECORD_PROC,
#{key => Key},
?DHT_RECORD_TIMEOUT_MS)).
classify_find({ok, #{type := _, payload := _, sig := _} = Record}) ->
{ok, Record};
classify_find({ok, not_found}) -> {error, not_found};
classify_find({ok, Reply}) -> {error, {unexpected_reply, Reply}};
classify_find({error, _} = E) -> E.
%% @doc Return every record of a given type currently visible from
%% the connected relay.
%%
%% Coverage depends on the relay's view of the DHT — a single relay
%% sees its local replicas plus whatever its peers have gossiped.
%% Aggregating across the full mesh requires querying multiple
%% relays and deduplicating by record key.
-spec find_records_by_type(client(), record_type()) ->
{ok, [record()]} | {error, term()}.
find_records_by_type(Client, Type)
when is_pid(Client), is_integer(Type), Type >= 0, Type =< 255 ->
classify_list(macula_mesh_client:call(Client,
?DHT_FIND_RECORDS_BY_TYPE_PROC,
#{type => Type},
?DHT_RECORD_TIMEOUT_MS)).
classify_list({ok, Records}) when is_list(Records) -> {ok, Records};
classify_list({ok, Reply}) -> {error, {unexpected_reply, Reply}};
classify_list({error, _} = E) -> E.
%% @doc Subscribe to live record-stored events filtered by type.
%%
%% The callback (or pid) receives each newly-stored record of the
%% given type as `{record, Record}' (pid form) or via direct
%% invocation (fun form). 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(client(), record_type(),
fun((record()) -> any()) | pid()) ->
{ok, reference()} | {error, term()}.
subscribe_records(Client, Type, Callback)
when is_pid(Client), is_integer(Type), Type >= 0, Type =< 255 ->
Topic = record_stored_topic(Type),
macula_mesh_client:subscribe(Client, Topic,
wrap_record_callback(Callback)).
%% @doc Cancel a `subscribe_records/3' subscription.
-spec unsubscribe_records(client(), reference()) -> ok | {error, term()}.
unsubscribe_records(Client, Ref)
when is_pid(Client), is_reference(Ref) ->
macula_mesh_client:unsubscribe(Client, Ref).
record_stored_topic(Type) ->
iolist_to_binary([<<"_dht.records.">>,
integer_to_binary(Type),
<<".stored">>]).
wrap_record_callback(Pid) when is_pid(Pid) ->
fun(Record) -> Pid ! {record, Record}, ok end;
wrap_record_callback(Fun) when is_function(Fun, 1) ->
fun(Record) -> Fun(Record), ok end.
%%%===================================================================
%%% 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 server-stream call (default mode).
%%
%% Phase 1 shortcut — dispatches in-process via the local registry.
%% For cross-node streaming use the `(Client, Procedure, Args)' form.
-spec call_stream(procedure(), term()) -> {ok, stream()} | {error, term()}.
call_stream(Procedure, Args) when is_binary(Procedure) ->
call_stream(Procedure, Args, #{}).
%% @doc Open a server-stream call with options.
%%
%% Two shapes:
%% call_stream(Procedure, Args, Opts) — LOCAL dispatch only
%% call_stream(Client, Procedure, Args) — REMOTE via mesh client
-spec call_stream(procedure() | client(), procedure() | term(), 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);
call_stream(Client, Procedure, Args) when is_pid(Client), is_binary(Procedure) ->
macula_mesh_client:call_stream(Client, Procedure, Args, #{}).
%% @doc Open a remote server-stream call against a mesh client, with opts.
-spec call_stream(client(), procedure(), term(), map()) ->
{ok, stream()} | {error, term()}.
call_stream(Client, Procedure, Args, Opts)
when is_pid(Client), is_binary(Procedure), is_map(Opts) ->
macula_mesh_client:call_stream(Client, Procedure, Args, Opts).
%% @doc Open a client-stream or bidi call.
%%
%% Two shapes:
%% open_stream(Procedure, Args, Opts) — LOCAL dispatch
%% open_stream(Client, Procedure, Args) — REMOTE, default mode bidi
-spec open_stream(procedure() | client(), procedure() | term(), 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);
open_stream(Client, Procedure, Args) when is_pid(Client), is_binary(Procedure) ->
macula_mesh_client:open_stream(Client, Procedure, Args, #{}).
%% @doc Open a stream with explicit mode.
%%
%% Two shapes:
%% open_stream(Procedure, Args, Opts, Mode) — LOCAL dispatch, explicit mode
%% open_stream(Client, Procedure, Args, Opts) — REMOTE via mesh client
-spec open_stream(procedure() | client(), procedure() | term(),
term() | map(), stream_mode() | map()) ->
{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});
open_stream(Client, Procedure, Args, Opts)
when is_pid(Client), is_binary(Procedure), is_map(Opts) ->
macula_mesh_client:open_stream(Client, Procedure, Args, Opts).
%% @doc Advertise a streaming procedure (default: server_stream).
%% LOCAL dispatch only — see `advertise_stream/3,4' for the Client form.
-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 streaming procedure.
%%
%% Two shapes:
%% advertise_stream(Procedure, Mode, Handler) — LOCAL dispatch
%% advertise_stream(Client, Procedure, Handler) — REMOTE, default
%% mode server_stream
-spec advertise_stream(procedure() | client(),
stream_mode() | procedure(),
stream_handler()) ->
ok | {error, term()}.
advertise_stream(Procedure, Mode, Handler)
when is_binary(Procedure), is_atom(Mode), is_function(Handler, 2) ->
advertise_stream(Procedure, Mode, Handler, #{});
advertise_stream(Client, Procedure, Handler)
when is_pid(Client), is_binary(Procedure), is_function(Handler, 2) ->
macula_mesh_client:advertise_stream(Client, Procedure, server_stream,
Handler).
%% @doc Advertise with explicit mode / options.
%%
%% Two shapes:
%% advertise_stream(Procedure, Mode, Handler, Opts) — LOCAL
%% advertise_stream(Client, Procedure, Mode, Handler) — REMOTE
-spec advertise_stream(procedure() | client(),
stream_mode() | procedure(),
stream_handler() | stream_mode(),
map() | stream_handler()) ->
ok | {error, term()}.
advertise_stream(Procedure, Mode, Handler, Opts)
when is_binary(Procedure), is_function(Handler, 2), is_map(Opts) ->
macula_stream_local:advertise(Procedure, Mode, Handler);
advertise_stream(Client, Procedure, Mode, Handler)
when is_pid(Client), is_binary(Procedure), is_atom(Mode),
is_function(Handler, 2) ->
macula_mesh_client:advertise_stream(Client, Procedure, Mode, Handler).
%% @doc Stop advertising a streaming procedure.
-spec unadvertise_stream(procedure()) -> ok.
unadvertise_stream(Procedure) when is_binary(Procedure) ->
macula_stream_local:unadvertise(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_v1: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_v1: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_v1:recv(Stream).
-spec recv(stream(), timeout()) -> {chunk, binary()}
| {data, term()}
| eof
| {error, term()}.
recv(Stream, Timeout) when is_pid(Stream) ->
macula_stream_v1: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_v1: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_v1: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_v1:await_reply(Stream).
-spec await_reply(stream(), timeout()) -> {ok, term()} | {error, term()}.
await_reply(Stream, Timeout) when is_pid(Stream) ->
macula_stream_v1: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_v1:set_reply(Stream, Result).
%% @doc Abort the stream with an error frame.
-spec abort(stream(), binary(), binary()) -> ok.
abort(Stream, Code, Message)
when is_pid(Stream), is_binary(Code), is_binary(Message) ->
macula_stream_v1:abort(Stream, Code, Message).
%%%===================================================================
%%% Directed RPC (Mesh Name Service)
%%%===================================================================
%% @doc Call a remote procedure on a specific target node (default 5s timeout).
%%
%% Target can be a mesh name, site_id, or node_id (all binaries).
%% The relay resolves the target and routes the CALL directly to that node.
-spec call_node(client(), binary(), procedure(), term()) ->
{ok, term()} | {error, term()}.
call_node(Client, Target, Procedure, Args) ->
call_node(Client, Target, Procedure, Args, 5000).
%% @doc Call a remote procedure on a specific target node with timeout.
-spec call_node(client(), binary(), procedure(), term(), pos_integer()) ->
{ok, term()} | {error, term()}.
call_node(Client, Target, Procedure, Args, Timeout)
when is_pid(Client), is_binary(Target), is_binary(Procedure) ->
macula_mesh_client:call(Client, Procedure, Args, Timeout, #{target => Target}).
%% @doc Resolve a mesh name to node identity information.
%%
%% Returns the node's identity including name, site_id, city, endpoint,
%% and connected_at timestamp. Works for names, site_ids, or node_ids.
-spec resolve(client(), binary()) -> {ok, map()} | {error, term()}.
resolve(Client, Name) when is_pid(Client), is_binary(Name) ->
macula_mesh_client:call(Client, <<"_mesh.resolve">>,
#{<<"name">> => Name}, 5000).
%% @doc List all nodes connected to the relay (default 5s timeout).
-spec list_nodes(client()) -> {ok, map()} | {error, term()}.
list_nodes(Client) ->
list_nodes(Client, #{}).
%% @doc List all nodes connected to the relay with options.
-spec list_nodes(client(), map()) -> {ok, map()} | {error, term()}.
list_nodes(Client, _Opts) when is_pid(Client) ->
macula_mesh_client:call(Client, <<"_mesh.list_nodes">>, #{}, 5000).
%%%===================================================================
%%% Cluster (LAN)
%%%===================================================================
%% @doc Ensure this node is running in distributed mode.
-spec ensure_distributed() -> ok | {error, term()}.
ensure_distributed() -> macula_cluster:ensure_distributed().
%% @doc Get the Erlang cluster cookie.
-spec get_cookie() -> atom().
get_cookie() -> macula_cluster:get_cookie().
%% @doc Set the Erlang cluster cookie.
-spec set_cookie(atom() | binary()) -> ok.
set_cookie(Cookie) -> macula_cluster:set_cookie(Cookie).
%% @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.
%% Options: relays (required list), realm, identity, site, tls_verify.
-spec join_mesh(map()) -> ok | {error, term()}.
join_mesh(Opts) ->
Relays = maps:get(relays, Opts),
ClientOpts = #{
relays => Relays,
realm => maps:get(realm, Opts, <<"io.macula">>),
identity => maps:get(identity, Opts, atom_to_binary(node())),
tls_verify => maps:get(tls_verify, Opts, none),
site => maps:get(site, Opts, undefined)
},
case macula_mesh_client:start_link(ClientOpts) of
{ok, Client} ->
%% Wait for relay connection before advertising
wait_for_relay(Client, 30),
os:putenv("MACULA_DIST_MODE", "relay"),
macula_dist_relay:register_mesh_client(Client),
macula_dist_relay:advertise_dist_accept(),
?LOG_INFO("[macula] Joined mesh — distribution enabled"),
ok;
{error, Reason} ->
?LOG_ERROR("[macula] Failed to join mesh: ~p", [Reason]),
{error, Reason}
end.
%% @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.
-spec join_dist_relay(map()) -> ok | {error, term()}.
join_dist_relay(Opts) ->
Url = maps:get(url, Opts),
NodeName = atom_to_binary(node()),
case macula_dist_system:start_dist_relay_client(Url, NodeName) 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} = Err ->
?LOG_ERROR("[macula] Failed to join dist relay: ~p", [Reason]),
Err
end.
%% @private Wait until the mesh client has an active QUIC stream.
wait_for_relay(_Client, 0) ->
?LOG_WARNING("[macula] Relay connection not ready after timeout");
wait_for_relay(Client, Retries) ->
case catch macula_mesh_client:is_connected(Client) of
true ->
?LOG_INFO("[macula] Relay connected"),
ok;
_ ->
timer:sleep(1000),
wait_for_relay(Client, Retries - 1)
end.