%%%------------------------------------------------------------------- %%% @doc %%% `embryo' - Library for managing Emergence configuration and embryo objects %%% %%% This module provides functions to: %%% - Create and manipulate embryo objects %%% - Read Emergence configuration files %%% - Retrieve URLs from the discovery service %%% - Merge lists of embryos %%% - Aggregate data from multiple filters %%% - Discover and manage filters automatically %%% %%% @author Steve Roques %%% @version 0.3.0 %%% @end %%%------------------------------------------------------------------- -module(embryo). -include("embryo.hrl"). %% Public API -export([ new/1, to_map/1, merge_lists_by_url/2, read_emergence_conf/0, get_em_disco_url/0, aggregate/1, aggregate/2, add_filter/1, remove_filter/1, add_discovery_source/1, remove_discovery_source/1, discover_filters/0, start_discovery/0, stop_discovery/0, get_port_from_env/2, get_url_from_env/2, get_config_value/3, safe_string_to_integer/2, decode_html_entities/1 ]). -define(TIMEOUT, 30000). -define(DISCOVERY_INTERVAL, 120000). %%==================================================================== %% API Functions %%==================================================================== -spec new(map()) -> embryo(). new(Properties) -> #embryo{properties = Properties}. -spec to_map(embryo()) -> map(). to_map(#embryo{properties = Properties}) -> #{properties => Properties}. -spec read_emergence_conf() -> config_map() | undefined. read_emergence_conf() -> ConfigPath = get_config_path(), case ConfigPath of undefined -> undefined; Path -> case filelib:is_file(Path) of true -> read_file(Path); false -> undefined end end. -spec get_em_disco_url() -> string(). get_em_disco_url() -> case os:getenv("server_url") of false -> ConfigMap = read_emergence_conf(), get_url_from_config(ConfigMap); Url -> Url end. -spec merge_lists_by_url([embryo()], [embryo()]) -> [embryo()]. merge_lists_by_url(UriList, OtherList) -> CombinedList = UriList ++ OtherList, {Result, _} = lists:foldl(fun merge_embryo/2, {[], sets:new()}, CombinedList), lists:reverse(Result). %%==================================================================== %% Aggregation Functions %%==================================================================== -spec aggregate(binary()) -> [embryo()]. aggregate(Body) -> aggregate(Body, []). -spec aggregate(binary(), [binary()]) -> [embryo()]. aggregate(Body, AdditionalFilters) -> io:format("[INFO] Starting aggregation with body: ~p~n", [Body]), FilterUrls = get_filter_urls() ++ AdditionalFilters, case FilterUrls of [] -> io:format("[WARN] No filters available for aggregation~n"), []; _ -> Results = call_filters(FilterUrls, Body), MergedResults = merge_results(Results), io:format("[INFO] Aggregation completed with ~p results~n", [length(MergedResults)]), MergedResults end. -spec add_filter(binary()) -> ok. add_filter(Url) when is_binary(Url) -> io:format("[INFO] Adding filter: ~p~n", [Url]), ensure_filter_registry(), ets:insert(filter_registry, {Url, true}), io:format("[SUCCESS] Filter added: ~p~n", [Url]), ok. -spec remove_filter(binary()) -> ok. remove_filter(Url) when is_binary(Url) -> io:format("[INFO] Removing filter: ~p~n", [Url]), ensure_filter_registry(), ets:delete(filter_registry, Url), io:format("[SUCCESS] Filter removed: ~p~n", [Url]), ok. %%==================================================================== %% Discovery Functions %%==================================================================== -spec add_discovery_source(binary()) -> ok. add_discovery_source(Source) when is_binary(Source) -> io:format("[INFO] Adding discovery source: ~p~n", [Source]), ensure_discovery_sources(), ets:insert(discovery_sources, {Source, true}), io:format("[SUCCESS] Discovery source added: ~p~n", [Source]), % Trigger immediate discovery spawn(fun() -> discover_filters() end), ok. -spec remove_discovery_source(binary()) -> ok. remove_discovery_source(Source) when is_binary(Source) -> io:format("[INFO] Removing discovery source: ~p~n", [Source]), ensure_discovery_sources(), ets:delete(discovery_sources, Source), io:format("[SUCCESS] Discovery source removed: ~p~n", [Source]), ok. -spec discover_filters() -> ok. discover_filters() -> io:format("[INFO] Running filter discovery...~n"), Sources = get_discovery_sources(), % Discover from configured sources lists:foreach(fun(Source) -> discover_from_source(Source) end, Sources), % Discover local Erlang nodes discover_local_erlang_nodes(), io:format("[INFO] Filter discovery completed. Current filters: ~p~n", [get_filter_urls()]), ok. -spec start_discovery() -> ok. start_discovery() -> io:format("[INFO] Starting automatic filter discovery...~n"), ensure_filter_registry(), ensure_discovery_sources(), case whereis(embryo_discovery_loop) of undefined -> Pid = spawn_link(fun() -> discovery_loop() end), register(embryo_discovery_loop, Pid), io:format("[SUCCESS] Discovery loop started~n"); _ -> io:format("[INFO] Discovery loop already running~n") end, ok. -spec stop_discovery() -> ok. stop_discovery() -> io:format("[INFO] Stopping automatic filter discovery...~n"), case whereis(embryo_discovery_loop) of undefined -> io:format("[INFO] Discovery loop not running~n"); Pid -> unregister(embryo_discovery_loop), exit(Pid, normal), io:format("[SUCCESS] Discovery loop stopped~n") end, ok. %%==================================================================== %% Configuration Helper Functions %%==================================================================== -spec get_port_from_env(string(), integer()) -> integer(). get_port_from_env(EnvVar, DefaultPort) -> case os:getenv(EnvVar) of false -> ConfigMap = read_emergence_conf(), SectionName = string:replace(EnvVar, "_port", ""), get_port_from_config(ConfigMap, SectionName, DefaultPort); PortStr -> safe_string_to_integer(PortStr, DefaultPort) end. -spec get_url_from_env(string(), string()) -> string(). get_url_from_env(EnvVar, DefaultUrl) -> case os:getenv(EnvVar) of false -> ConfigMap = read_emergence_conf(), SectionName = string:replace(EnvVar, "_url", ""), get_url_from_config_section(ConfigMap, SectionName, DefaultUrl); Url -> Url end. -spec get_config_value(string(), string(), string()) -> string(). get_config_value(Section, Key, DefaultValue) -> ConfigMap = read_emergence_conf(), case ConfigMap of undefined -> DefaultValue; Map -> case maps:get(Section, Map, undefined) of undefined -> DefaultValue; SectionMap -> maps:get(Key, SectionMap, DefaultValue) end end. -spec safe_string_to_integer(string(), integer()) -> integer(). safe_string_to_integer(Str, DefaultValue) -> try list_to_integer(Str) catch _:_ -> DefaultValue end. %%==================================================================== %% Internal Configuration Functions %%==================================================================== -spec get_port_from_config(map() | undefined, string(), integer()) -> integer(). get_port_from_config(undefined, _, DefaultPort) -> DefaultPort; get_port_from_config(ConfigMap, SectionName, DefaultPort) -> case maps:get(SectionName, ConfigMap, undefined) of undefined -> DefaultPort; Section -> PortStr = maps:get("port", Section, integer_to_list(DefaultPort)), safe_string_to_integer(PortStr, DefaultPort) end. -spec get_url_from_config_section(map() | undefined, string(), string()) -> string(). get_url_from_config_section(undefined, _, DefaultUrl) -> DefaultUrl; get_url_from_config_section(ConfigMap, SectionName, DefaultUrl) -> case maps:get(SectionName, ConfigMap, undefined) of undefined -> DefaultUrl; Section -> maps:get("server_url", Section, DefaultUrl) end. %%==================================================================== %% Internal Functions %%==================================================================== -spec get_config_path() -> string() | undefined. get_config_path() -> case os:getenv("HOME") of false -> case os:getenv("APPDATA") of false -> undefined; AppData -> filename:join([AppData, "emergence", "emergence.conf"]) end; Home -> case os:type() of {unix, _} -> filename:join([Home, ".config", "emergence", "emergence.conf"]); {win32, _} -> filename:join([Home, "AppData", "Roaming", "emergence", "emergence.conf"]) end end. -spec read_file(string()) -> map(). read_file(Path) -> case file:read_file(Path) of {ok, Binary} -> Lines = binary:split(Binary, <<"\n">>, [global, trim_all]), parse_config_lines(Lines, #{}, ""); {error, _} -> #{} end. -spec parse_config_lines([binary()], map(), string()) -> map(). parse_config_lines([], Map, _) -> Map; parse_config_lines([Line|Rest], Map, CurrentSection) -> case Line of <<"[", Rest1/binary>> -> Section = string:trim(binary_to_list(binary:part(Rest1, 0, byte_size(Rest1)-1)), both, "]"), parse_config_lines(Rest, Map#{Section => #{}}, Section); <<>> -> parse_config_lines(Rest, Map, CurrentSection); _ -> case binary:split(Line, <<"=">>) of [Key, Value] -> TrimmedKey = string:trim(binary_to_list(Key)), TrimmedValue = string:trim(binary_to_list(Value)), NewMap = case maps:get(CurrentSection, Map, undefined) of undefined -> Map; SectionMap -> Map#{CurrentSection => SectionMap#{TrimmedKey => TrimmedValue}} end, parse_config_lines(Rest, NewMap, CurrentSection); _ -> parse_config_lines(Rest, Map, CurrentSection) end end. -spec get_url_from_config(map() | undefined) -> string(). get_url_from_config(undefined) -> "http://localhost:8080"; get_url_from_config(ConfigMap) -> case maps:get("em_disco", ConfigMap, undefined) of undefined -> "http://localhost:8080"; EmDisco -> maps:get("server_url", EmDisco, "http://localhost:8080") end. -spec merge_embryo(embryo(), {[embryo()], sets:set()}) -> {[embryo()], sets:set()}. merge_embryo(Embryo, {Acc, Seen}) -> Properties = Embryo#embryo.properties, case maps:get(<<"url">>, Properties, undefined) of undefined -> {[Embryo | Acc], Seen}; Url -> case sets:is_element(Url, Seen) of true -> {Acc, Seen}; false -> {[Embryo | Acc], sets:add_element(Url, Seen)} end end. %%==================================================================== %% Internal Aggregation Functions %%==================================================================== ensure_filter_registry() -> case ets:info(filter_registry) of undefined -> ets:new(filter_registry, [set, public, named_table]); _ -> ok end. ensure_discovery_sources() -> case ets:info(discovery_sources) of undefined -> ets:new(discovery_sources, [set, public, named_table]); _ -> ok end. get_filter_urls() -> case ets:info(filter_registry) of undefined -> []; _ -> FilterUrls = ets:tab2list(filter_registry), lists:map(fun({Url, _}) -> Url end, FilterUrls) end. call_filters(Urls, Body) -> Parent = self(), Refs = [begin Ref = make_ref(), spawn(fun() -> Result = call_filter(Url, Body), Parent ! {Ref, Result} end), Ref end || Url <- Urls], collect_results(Refs, []). call_filter(Url, RequestBody) -> StartTime = erlang:system_time(millisecond), io:format("[INFO] Calling filter ~p with body ~p~n", [Url, RequestBody]), UrlStr = binary_to_list(Url), TimeoutStr = integer_to_binary(?TIMEOUT div 1000 - 1), JsonBody = jsx:encode(#{ <<"value">> => RequestBody, <<"timeout">> => TimeoutStr }), Headers = [{"content-type", "application/json"}], HttpOptions = [{timeout, ?TIMEOUT}, {connect_timeout, ?TIMEOUT}], try case httpc:request(post, {UrlStr, Headers, "application/json", JsonBody}, HttpOptions, []) of {ok, {{_, 200, _}, _RespHeaders, ResponseBody}} -> ElapsedTime = erlang:system_time(millisecond) - StartTime, io:format("[SUCCESS] Filter ~p responded in ~p ms~n", [UrlStr, ElapsedTime]), parse_response(ResponseBody, Url); {ok, {{_, StatusCode, _}, _, _}} -> io:format("[ERROR] Filter ~p returned HTTP status code ~p~n", [UrlStr, StatusCode]), {error, {bad_status_code, StatusCode}}; {error, Reason} -> io:format("[ERROR] HTTP request to filter ~p failed with reason ~p~n", [UrlStr, Reason]), % Remove unresponsive filter from registry ets:delete(filter_registry, Url), {error, Reason} end catch E:R:_S -> io:format("[ERROR] Exception occurred while calling filter ~p. Error: ~p:~p~n", [UrlStr, E, R]), {error, {exception, {E, R}}} end. parse_response(ResponseBody, _Url) -> case safe_json_decode(ResponseBody) of {ok, EmbryoList} -> {ok, EmbryoList}; {error, _} -> io:format("[WARN] JSON decode failed, attempting regex extraction~n"), ExtractedData = extract_embryos(ResponseBody), {ok, ExtractedData} end. safe_json_decode(ResponseBody) -> try Normalized = normalize_json_string(ResponseBody), DecodedResponse = jsx:decode(Normalized, [return_maps]), case maps:get(<<"embryo_list">>, DecodedResponse, undefined) of EmbryoList when is_list(EmbryoList) -> {ok, EmbryoList}; _ -> {error, invalid_response} end catch error:badarg -> {error, json_decode_error}; E:R:_S -> io:format("[ERROR] Unexpected error during JSON decode: ~p:~p~n", [E, R]), {error, {unexpected_error, {E, R}}} end. normalize_json_string(Binary) -> try case unicode:characters_to_binary(Binary) of {error, _, _} -> safe_binary_sanitize(Binary); {incomplete, _, _} -> safe_binary_sanitize(Binary); ValidUtf8 -> ValidUtf8 end catch _:_ -> safe_binary_sanitize(Binary) end. safe_binary_sanitize(Binary) -> binary:replace(Binary, [<<194, 160>>, % non-breaking space <<195>>, % common part of many UTF-8 accented chars <<194>>], % common part of many UTF-8 special chars <<32>>, % regular space [global]). extract_embryos(ResponseBody) -> try Pattern = <<"\"embryo_list\":\\s*\\[(.*?)\\]">>, case re:run(ResponseBody, Pattern, [dotall, {capture, [1], binary}]) of {match, [ListContent]} -> extract_embryo_objects(ListContent); nomatch -> EmbrPattern = <<"\\{\"properties\":\\s*\\{[^{}]*\\}\\}">>, case re:run(ResponseBody, EmbrPattern, [dotall, {capture, all, binary}, global]) of {match, Matches} -> [parse_single_embryo(Match) || [Match] <- Matches]; nomatch -> [] end end catch E:R:_S -> io:format("[ERROR] Failed to extract embryos: ~p:~p~n", [E, R]), [] end. extract_embryo_objects(ListContent) -> Pattern = <<"\\{\"properties\":[^{}]*\\{[^{}]*\\}[^{}]*\\}">>, case re:run(ListContent, Pattern, [dotall, {capture, all, binary}, global]) of {match, Matches} -> [parse_single_embryo(Match) || [Match] <- Matches]; nomatch -> [] end. parse_single_embryo(EmbryoJson) -> UrlPattern = <<"\"url\":\\s*\"([^\"]+)\"">>, ResumePattern = <<"\"resume\":\\s*\"([^\"]*)\"">>, Url = case re:run(EmbryoJson, UrlPattern, [{capture, [1], binary}]) of {match, [UrlValue]} -> UrlValue; nomatch -> <<"">> end, Resume = case re:run(EmbryoJson, ResumePattern, [{capture, [1], binary}]) of {match, [ResumeValue]} -> ResumeValue; nomatch -> <<"">> end, #embryo{properties = #{ <<"url">> => unescape_json_string(Url), <<"resume">> => unescape_json_string(Resume) }}. unescape_json_string(Str) -> S1 = binary:replace(Str, <<"\\\\">>, <<"\\">>, [global]), S2 = binary:replace(S1, <<"\\\/">>, <<"/">>, [global]), S3 = binary:replace(S2, <<"\\\"">>, <<"\"">>, [global]), S4 = binary:replace(S3, <<"\\n">>, <<"\n">>, [global]), S5 = binary:replace(S4, <<"\\r">>, <<"\r">>, [global]), S6 = binary:replace(S5, <<"\\t">>, <<"\t">>, [global]), handle_unicode_escapes(S6). handle_unicode_escapes(Str) -> Pattern = <<"\\\\u([0-9a-fA-F]{4})">>, case re:run(Str, Pattern, [{capture, all, binary}, global]) of {match, Matches} -> lists:foldl(fun([Full, HexCode], Acc) -> try Code = binary_to_integer(HexCode, 16), Char = unicode:characters_to_binary([Code], unicode, utf8), binary:replace(Acc, Full, Char, [global]) catch _:_ -> Acc end end, Str, Matches); nomatch -> Str end. collect_results([], Acc) -> Acc; collect_results([Ref|Refs], Acc) -> receive {Ref, {ok, Result}} -> collect_results(Refs, [Result|Acc]); {Ref, {error, Error}} -> io:format("[WARN] Received error from a filter: ~p~n", [Error]), collect_results(Refs, Acc) after ?TIMEOUT -> io:format("[ERROR] Timeout occurred while waiting for filter responses~n"), collect_results(Refs, Acc) end. merge_results(Lists) -> CombinedList = lists:flatten(Lists), {Result, _} = lists:foldl(fun merge_embryo_result/2, {[], sets:new()}, CombinedList), lists:reverse(Result). merge_embryo_result(Data, {Acc, Seen}) -> Properties = case Data of #{<<"properties">> := Props} -> Props; #embryo{properties = Props} -> Props; _ -> #{} end, Key = case maps:find(<<"url">>, Properties) of {ok, Url} when is_binary(Url) -> Url; _ -> case maps:find(<<"resume">>, Properties) of {ok, Resume} when is_binary(Resume) -> Resume; _ -> make_ref() end end, case sets:is_element(Key, Seen) of true -> {Acc, Seen}; false -> {[Data | Acc], sets:add_element(Key, Seen)} end. %%==================================================================== %% Discovery Internal Functions %%==================================================================== discovery_loop() -> discover_filters(), timer:sleep(?DISCOVERY_INTERVAL), discovery_loop(). get_discovery_sources() -> case ets:info(discovery_sources) of undefined -> []; _ -> SourceList = ets:tab2list(discovery_sources), lists:map(fun({Source, _}) -> Source end, SourceList) end. discover_from_source(Source) -> io:format("[INFO] Discovering filters from source: ~p~n", [Source]), try SourceStr = binary_to_list(Source), HttpOptions = [{timeout, 5000}, {connect_timeout, 5000}], case httpc:request(get, {SourceStr, []}, HttpOptions, []) of {ok, {{_, 200, _}, _, ResponseBody}} -> Filters = parse_discovery_response(ResponseBody), lists:foreach(fun(FilterUrl) -> add_filter(FilterUrl) end, Filters); {ok, {{_, StatusCode, _}, _, _}} -> io:format("[ERROR] Discovery source ~p returned HTTP status code ~p~n", [SourceStr, StatusCode]); {error, Reason} -> io:format("[ERROR] HTTP request to discovery source ~p failed with reason ~p~n", [SourceStr, Reason]) end catch E:R:_S -> io:format("[ERROR] Exception occurred while discovering from source ~p. Error: ~p:~p~n", [Source, E, R]) end. parse_discovery_response(ResponseBody) -> try Normalized = normalize_json_string(ResponseBody), DecodedResponse = jsx:decode(Normalized, [return_maps]), case maps:get(<<"filters">>, DecodedResponse, undefined) of FilterList when is_list(FilterList) -> FilterList; _ -> io:format("[ERROR] Invalid discovery response format. Expected 'filters' array~n"), [] end catch error:badarg -> io:format("[ERROR] Failed to parse discovery response as JSON~n"), extract_filter_urls_from_text(ResponseBody); E:R:_S -> io:format("[ERROR] Unexpected error during discovery response parsing: ~p:~p~n", [E, R]), [] end. extract_filter_urls_from_text(ResponseBody) -> try Pattern = <<"https?://[^\\s\"']+">>, case re:run(ResponseBody, Pattern, [global, {capture, all, binary}]) of {match, Matches} -> [Url || [Url] <- Matches]; nomatch -> [] end catch _:_ -> [] end. discover_local_erlang_nodes() -> io:format("[INFO] Discovering local Erlang nodes responding to HTTP...~n"), BaseUrl = "http://localhost:", TestValue = <<"test">>, Ports = get_potential_erlang_ports(), lists:foreach( fun(Port) -> spawn(fun() -> Url = list_to_binary(BaseUrl ++ integer_to_list(Port) ++ "/query"), case try_call_filter(Url, TestValue) of {ok, _} -> io:format("[INFO] Found Erlang node responding at: ~s~n", [Url]), add_filter(Url); {error, _Reason} -> ok end end) end, Ports), ok. try_call_filter(Url, RequestBody) -> UrlStr = binary_to_list(Url), TimeoutStr = integer_to_binary(?TIMEOUT), JsonBody = jsx:encode(#{ <<"value">> => RequestBody, <<"timeout">> => TimeoutStr }), Headers = [{"content-type", "application/json"}], HttpOptions = [{timeout, ?TIMEOUT}, {connect_timeout, ?TIMEOUT}], try case httpc:request(post, {UrlStr, Headers, "application/json", JsonBody}, HttpOptions, []) of {ok, {{_, 200, _}, _RespHeaders, ResponseBody}} -> {ok, ResponseBody}; {ok, {{_, StatusCode, _}, _, _}} -> {error, {bad_status_code, StatusCode}}; {error, Reason} -> {error, Reason} end catch E:R:_ -> {error, {exception, {E, R}}} end. get_potential_erlang_ports() -> lists:seq(8081, 8100). %%------------------------------------------------------------------ %% HTML Entity Decoding %%------------------------------------------------------------------ decode_html_entities(Text) -> Entities = [ {"&", "&"}, {"<", "<"}, {">", ">"}, {""", "\""}, {"'", "'"}, {"à", "à"}, {"á", "á"}, {"é", "é"}, {"è", "è"}, {"ê", "ê"}, {"î", "î"}, {"ô", "ô"}, {"ù", "ù"}, {"ç", "ç"}, {" ", " "} ], lists:foldl(fun({Entity, Char}, Acc) -> re:replace(Acc, Entity, Char, [global, {return, list}]) end, Text, Entities).