Packages
macula
0.39.9
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_content_system/macula_content_dht.erl
%%%-------------------------------------------------------------------
%%% @doc
%%% DHT integration for Macula content-addressed storage.
%%%
%%% Handles content announcement and discovery through the DHT:
%%% - Announce manifest availability to mesh peers
%%% - Discover providers of content by MCID
%%% - TTL-based re-announcement for liveness
%%%
%%% == DHT Key Structure ==
%%% DHT keys are SHA-256 hashes of the MCID. This ensures:
%%% - Uniform distribution in the DHT keyspace
%%% - Same content always maps to same key
%%% - 32-byte keys match DHT node ID size
%%%
%%% == Provider Discovery ==
%%% When content is published, the provider announces to the DHT:
%%% ```
%%% Key: hash(MCID)
%%% Value: #{node_id, endpoint, metadata, advertised_at, ttl}
%%% '''
%%% Multiple providers can announce the same content.
%%%
%%% == Example Usage ==
%%% ```
%%% %% Announce content availability
%%% {Key, Value} = macula_content_dht:create_announcement(MCID, NodeId, Endpoint, Info),
%%% macula_routing_server:store(RoutingPid, Key, Value).
%%%
%%% %% Discover providers
%%% case macula_routing_server:find_value(RoutingPid, Key, 20) of
%%% {ok, Providers} -> handle_providers(Providers);
%%% {nodes, _} -> {error, not_found}
%%% end.
%%% '''
%%% @end
%%%-------------------------------------------------------------------
-module(macula_content_dht).
-include_lib("kernel/include/logger.hrl").
%% API - Key generation
-export([
dht_key/1
]).
%% API - Provider info
-export([
create_provider_info/3,
format_providers/1
]).
%% API - Announcement
-export([
create_announcement/4,
create_removal/1
]).
%% API - TTL
-export([
default_ttl/0,
get_ttl/1,
reannounce_interval/1
]).
%% API - High-level operations (use with routing server)
-export([
announce_manifest/4,
announce_manifest/5,
unannounce_manifest/3,
locate_providers/1,
locate_providers/2
]).
%% Constants
-define(DEFAULT_TTL, 300). % 5 minutes
-define(MIN_REANNOUNCE, 30). % Minimum 30 seconds before TTL expires
%%%===================================================================
%%% Types
%%%===================================================================
-type mcid() :: <<_:272>>. %% 34 bytes: version + codec + hash
-type provider_info() :: #{
node_id := binary(),
endpoint := binary(),
metadata := map(),
advertised_at := integer(),
ttl => pos_integer()
}.
-export_type([mcid/0, provider_info/0]).
%%%===================================================================
%%% Key Generation
%%%===================================================================
%% @doc Generate DHT key from MCID.
%% Uses SHA-256 to hash the MCID, ensuring uniform distribution
%% in the DHT keyspace.
-spec dht_key(mcid()) -> binary().
dht_key(MCID) when is_binary(MCID) ->
crypto:hash(sha256, MCID).
%%%===================================================================
%%% Provider Info
%%%===================================================================
%% @doc Create provider info map with all required fields.
-spec create_provider_info(binary(), binary(), map()) -> provider_info().
create_provider_info(NodeId, Endpoint, Metadata) ->
#{
node_id => NodeId,
endpoint => Endpoint,
metadata => Metadata,
advertised_at => erlang:system_time(second)
}.
%% @doc Format raw provider values into standardized list.
%% Handles both single provider and list of providers.
-spec format_providers([provider_info()] | provider_info()) -> [provider_info()].
format_providers([]) ->
[];
format_providers(Providers) when is_list(Providers) ->
[format_single_provider(P) || P <- Providers];
format_providers(SingleProvider) when is_map(SingleProvider) ->
[format_single_provider(SingleProvider)].
%% @private Format single provider, ensuring consistent structure.
format_single_provider(#{node_id := NodeId, endpoint := Endpoint} = Provider) ->
#{
node_id => NodeId,
endpoint => Endpoint,
metadata => maps:get(metadata, Provider, #{}),
advertised_at => maps:get(advertised_at, Provider, 0)
};
format_single_provider(Provider) when is_map(Provider) ->
%% Handle legacy or incomplete provider info
#{
node_id => maps:get(node_id, Provider, <<>>),
endpoint => maps:get(endpoint, Provider, <<>>),
metadata => maps:get(metadata, Provider, #{}),
advertised_at => maps:get(advertised_at, Provider, 0)
}.
%%%===================================================================
%%% Announcement Functions
%%%===================================================================
%% @doc Create DHT announcement for content availability.
%% Returns {DHTKey, ProviderValue} tuple ready for storage.
-spec create_announcement(mcid(), binary(), binary(), map()) -> {binary(), provider_info()}.
create_announcement(MCID, NodeId, Endpoint, ManifestInfo) ->
Key = dht_key(MCID),
Value = #{
node_id => NodeId,
endpoint => Endpoint,
metadata => ManifestInfo,
advertised_at => erlang:system_time(second),
ttl => ?DEFAULT_TTL
},
{Key, Value}.
%% @doc Create removal marker for DHT.
%% When a provider stops hosting content, they announce removal.
-spec create_removal(binary()) -> map().
create_removal(NodeId) ->
#{
node_id => NodeId,
removed => true,
removed_at => erlang:system_time(second)
}.
%%%===================================================================
%%% TTL Functions
%%%===================================================================
%% @doc Get default TTL for content announcements.
-spec default_ttl() -> pos_integer().
default_ttl() ->
?DEFAULT_TTL.
%% @doc Get TTL from options or use default.
-spec get_ttl(map()) -> pos_integer().
get_ttl(Opts) when is_map(Opts) ->
maps:get(ttl, Opts, ?DEFAULT_TTL).
%% @doc Calculate interval for re-announcement.
%% Re-announces slightly before TTL expires to maintain presence.
%% Returns interval in seconds.
-spec reannounce_interval(pos_integer()) -> pos_integer().
reannounce_interval(TTL) when TTL > ?MIN_REANNOUNCE + 60 ->
TTL - 60; % 60 seconds before expiry
reannounce_interval(_TTL) ->
?MIN_REANNOUNCE. % Minimum 30 seconds
%%%===================================================================
%%% High-Level DHT Operations
%%%===================================================================
%% @doc Announce manifest availability to DHT.
%% Stores provider info at DHT key derived from MCID.
-spec announce_manifest(mcid(), binary(), binary(), map()) ->
ok | {error, term()}.
announce_manifest(MCID, NodeId, Endpoint, ManifestInfo) ->
announce_manifest(MCID, NodeId, Endpoint, ManifestInfo, #{}).
%% @doc Announce manifest with options.
%% Options:
%% - ttl: Custom TTL (default: 300 seconds)
-spec announce_manifest(mcid(), binary(), binary(), map(), map()) ->
ok | {error, term()}.
announce_manifest(MCID, NodeId, Endpoint, ManifestInfo, Opts) ->
{Key, BaseValue} = create_announcement(MCID, NodeId, Endpoint, ManifestInfo),
TTL = get_ttl(Opts),
Value = BaseValue#{ttl => TTL},
?LOG_INFO("[content_dht] Announcing manifest ~s from node ~s",
[format_mcid(MCID), short_id(NodeId)]),
case whereis(macula_routing_server) of
undefined ->
?LOG_WARNING("[content_dht] Routing server not available"),
{error, routing_server_unavailable};
RoutingPid ->
case macula_routing_server:store(RoutingPid, Key, Value) of
ok ->
?LOG_DEBUG("[content_dht] Successfully announced manifest"),
ok;
{error, Reason} = Error ->
?LOG_WARNING("[content_dht] Failed to announce: ~p", [Reason]),
Error
end
end.
%% @doc Remove manifest announcement from DHT.
-spec unannounce_manifest(mcid(), binary(), map()) -> ok | {error, term()}.
unannounce_manifest(MCID, NodeId, _Opts) ->
Key = dht_key(MCID),
RemovalValue = create_removal(NodeId),
?LOG_INFO("[content_dht] Unannouncing manifest ~s from node ~s",
[format_mcid(MCID), short_id(NodeId)]),
case whereis(macula_routing_server) of
undefined ->
{error, routing_server_unavailable};
RoutingPid ->
%% Store removal marker (or delete if DHT supports it)
macula_routing_server:store(RoutingPid, Key, RemovalValue)
end.
%% @doc Locate providers of content by MCID.
%% Queries DHT and returns list of provider endpoints.
-spec locate_providers(mcid()) -> {ok, [provider_info()]} | {error, term()}.
locate_providers(MCID) ->
locate_providers(MCID, #{}).
%% @doc Locate providers with options.
%% Options:
%% - k: Number of closest nodes to query (default: 20)
%% - timeout: Query timeout in milliseconds
-spec locate_providers(mcid(), map()) -> {ok, [provider_info()]} | {error, term()}.
locate_providers(MCID, Opts) ->
Key = dht_key(MCID),
K = maps:get(k, Opts, 20),
?LOG_DEBUG("[content_dht] Locating providers for ~s", [format_mcid(MCID)]),
case whereis(macula_routing_server) of
undefined ->
{error, routing_server_unavailable};
RoutingPid ->
case macula_routing_server:find_value(RoutingPid, Key, K) of
{ok, Value} when is_map(Value) ->
%% Single provider
filter_active_providers([Value]);
{ok, Values} when is_list(Values) ->
%% Multiple providers
filter_active_providers(Values);
{nodes, _ClosestNodes} ->
%% Value not found
{ok, []};
{error, Reason} ->
{error, Reason}
end
end.
%%%===================================================================
%%% Internal Functions
%%%===================================================================
%% @private Filter out removed providers and format result.
filter_active_providers(Providers) ->
Active = [P || P <- Providers,
is_map(P),
maps:get(removed, P, false) =:= false],
{ok, format_providers(Active)}.
%% @private Format MCID for logging.
format_mcid(<<_Version:8, _Codec:8, Hash:32/binary>>) ->
HexHash = macula_content_hasher:hex_encode(Hash),
<<Short:16/binary, _/binary>> = HexHash,
<<"mcid:", Short/binary, "...">>;
format_mcid(_) ->
<<"mcid:invalid">>.
%% @private Format node ID for logging (first 8 chars).
short_id(NodeId) when is_binary(NodeId), byte_size(NodeId) >= 4 ->
HexId = macula_content_hasher:hex_encode(binary:part(NodeId, 0, 4)),
<<Short:8/binary, _/binary>> = HexId,
Short;
short_id(_) ->
<<"unknown">>.