locus

%% Copyright (c) 2017-2019 Guilherme Andrade
%%
%% Permission is hereby granted, free of charge, to any person obtaining a
%% copy  of this software and associated documentation files (the "Software"),
%% to deal in the Software without restriction, including without limitation
%% the rights to use, copy, modify, merge, publish, distribute, sublicense,
%% and/or sell copies of the Software, and to permit persons to whom the
%% Software is furnished to do so, subject to the following conditions:
%%
%% The above copyright notice and this permission notice shall be included in
%% all copies or substantial portions of the Software.
%%
%% THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
%% IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
%% FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
%% AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
%% LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING
%% FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER
%% DEALINGS IN THE SOFTWARE.
%%
%% locus is an independent project and has not been authorized, sponsored,
%% or otherwise approved by MaxMind.

-module(locus_loader).
-behaviour(gen_server).

%% ------------------------------------------------------------------
%% API Function Exports
%% ------------------------------------------------------------------

-export(
   [validate_opts/2,
    start_link/4
   ]).

-ignore_xref(
   [start_link/4
   ]).

-ifdef(TEST).
-export([cached_database_path_for_url/1]).
-endif.

%% ------------------------------------------------------------------
%% gen_server Function Exports
%% ------------------------------------------------------------------

-export(
   [init/1,
    handle_call/3,
    handle_cast/2,
    handle_info/2,
    terminate/2,
    code_change/3
   ]).

%% ------------------------------------------------------------------
%% Macro Definitions
%% ------------------------------------------------------------------

-ifndef(NO_GEN_SERVER_HIBERNATE_AFTER).
-define(HIBERNATE_AFTER, (timer:seconds(5))).
-endif.

-define(DEFAULT_HTTP_UNREADY_UPDATE_PERIOD, (timer:minutes(1))).
-define(DEFAULT_HTTP_READY_UPDATE_PERIOD, (timer:hours(6))).

-define(DEFAULT_FS_UNREADY_UPDATE_PEROID, (timer:seconds(5))).
-define(DEFAULT_FS_READY_UPDATE_PERIOD, (timer:seconds(30))).

-define(is_pos_integer(V), ((is_integer((V)) andalso ((V) >= 1)))).

% https://en.wikipedia.org/wiki/Gzip
-define(GZIP_MAGIC_BYTES, 16#1f,16#8b).

%% ------------------------------------------------------------------
%% Record and Type Definitions
%% ------------------------------------------------------------------

-type opt() :: loader_opt() | fetcher_opt().
-export_type([opt/0]).

-type loader_opt() ::
    {pre_readiness_update_period, pos_integer()} |
    {post_readiness_update_period, pos_integer()} |
    no_cache.
-export_type([loader_opt/0]).

-type fetcher_opt() :: locus_http_download:opt().
-export_type([fetcher_opt/0]).

-type fetcher_msg() :: locus_http_download:msg() | locus_filesystem_load:msg().
-type fetcher_success() :: locus_http_download:success() | locus_filesystem_load:success().
-type cacher_msg() :: locus_filesystem_store:msg().

-record(state, {
          owner_pid :: pid(),
          database_id :: atom(),
          origin :: origin(),
          settings :: settings(),
          fetcher_opts :: [fetcher_opt()],

          update_timer :: reference() | undefined,
          last_modified :: calendar:datetime() | undefined,

          fetcher_pid :: pid() | undefined,
          fetcher_source :: source() | undefined,

          cacher_pid :: pid() | undefined,
          cacher_path :: locus_filesystem_store:path() | undefined,
          cacher_source :: source() | undefined
         }).
-type state() :: #state{}.

-record(settings, {
          unready_update_period :: pos_integer(),
          ready_update_period :: pos_integer(),
          use_cache :: boolean()
         }).
-type settings() :: #settings{}.

-type blob_format() :: tgz | tarball | gzip | gzipped_mmdb | mmdb | unknown.

-type origin() ::
    {http, locus_http_download:url()} |
    {filesystem, locus_filesystem_load:path()}.
-export_type([origin/0]).

-type event() ::
    locus_http_download:event() |
    locus_filesystem_load:event() |
    event_cache_attempt_finished().
-export_type([event/0]).

-type event_cache_attempt_finished() ::
    {cache_attempt_finished, locus_filesystem_store:path(), ok} |
    {cache_attempt_finished, locus_filesystem_store:path(), {error, term()}}.
-export_type([event_cache_attempt_finished/0]).

-type source() ::
    {remote, locus_http_download:url()} |
    locus_filesystem_load:source().
-export_type([source/0]).

-type msg() ::
    {event, event()} |
    {load_success, source(), calendar:datetime(), locus_mmdb:parts()} |
    {load_failure, source(), Reason :: term()}.
-export_type([msg/0]).

%% ------------------------------------------------------------------
%% API Function Definitions
%% ------------------------------------------------------------------

-spec validate_opts(origin(), proplists:proplist())
        -> {ok, {[opt()], [fetcher_opt()], proplists:proplist()}} |
           {error, BadOpt :: term()}.
%% @private
validate_opts(Origin, MixedOpts) ->
    case validate_fetcher_opts(Origin, MixedOpts) of
        {ok, {FetcherOpts, RemainingMixedOpts}} ->
            validate_loader_opts(RemainingMixedOpts, FetcherOpts);
        {error, BadOpt} ->
            {error, BadOpt}
    end.

-spec start_link(atom(), origin(), [loader_opt()], [fetcher_opt()]) -> {ok, pid()}.
%% @private
start_link(DatabaseId, Origin, LoaderOpts, FetcherOpts) ->
    Opts = [self(), DatabaseId, Origin, LoaderOpts, FetcherOpts],
    ServerOpts = server_opts(),
    gen_server:start_link(?MODULE, Opts, ServerOpts).

%% ------------------------------------------------------------------
%% gen_server Function Definitions
%% ------------------------------------------------------------------

-spec init([pid() | atom() | origin() | [loader_opt()] | [fetcher_opt()], ...])
    -> {ok, state()}.
%% @private
init([OwnerPid, DatabaseId, Origin, LoaderOpts, FetcherOpts]) ->
    _ = process_flag(trap_exit, true),
    DefaultSettings = default_settings(Origin),
    Settings = customized_settings(DefaultSettings, LoaderOpts),
    State =
        #state{
           owner_pid = OwnerPid,
           database_id = DatabaseId,
           origin = Origin,
           settings = Settings,
           fetcher_opts = FetcherOpts
          },
    self() ! finish_initialization,
    {ok, State}.

-spec handle_call(term(), {pid(),reference()}, state())
        -> {stop, unexpected_call, state()}.
%% @private
handle_call(_Call, _From, State) ->
    {stop, unexpected_call, State}.

-spec handle_cast(term(), state())
        -> {stop, unexpected_cast, state()}.
%% @private
handle_cast(_Cast, State) ->
    {stop, unexpected_cast, State}.

-spec handle_info(term(), state())
        -> {noreply, state()} |
           {stop, term(), state()}.
%% @private
handle_info(finish_initialization, State)
  when State#state.update_timer =:= undefined ->
    UpdatedState = finish_initialization(State),
    {noreply, UpdatedState};
handle_info(begin_update, State) ->
    false = erlang:cancel_timer(State#state.update_timer),
    State2 = State#state{ update_timer = undefined },
    State3 = begin_update(State2),
    {noreply, State3};
handle_info({FetcherPid, Msg}, State)
  when FetcherPid =:= State#state.fetcher_pid ->
    handle_fetcher_msg(Msg, State);
handle_info({CacherPid, Msg}, State)
  when CacherPid =:= State#state.cacher_pid ->
    handle_cacher_msg(Msg, State);
handle_info({'EXIT', Pid, Reason}, State) ->
    handle_linked_process_death(Pid, Reason, State);
handle_info(_Info, State) ->
    {stop, unexpected_info, State}.

-spec terminate(term(), state()) -> ok.
%% @private
terminate(_Reason, _State) ->
    ok.

-spec code_change(term(), state(), term()) -> {ok, state()}.
%% @private
code_change(_OldVsn, #state{} = State, _Extra) ->
    {ok, State}.

%% ------------------------------------------------------------------
%% Internal Function Definitions - Initialization
%% ------------------------------------------------------------------

-ifndef(NO_GEN_SERVER_HIBERNATE_AFTER).
server_opts() -> [{locus_util:dialyzer_opaque_atom(hibernate_after), ?HIBERNATE_AFTER}].
-else.
server_opts() -> [].
-endif.

-spec validate_fetcher_opts(origin(), list())
        -> {ok, {[fetcher_opt()], list()}} |
           {error, term()}.
validate_fetcher_opts({http,_}, MixedOpts) ->
    locus_http_download:validate_opts(MixedOpts);
validate_fetcher_opts({filesystem,_}, MixedOpts) ->
    {ok, {[], MixedOpts}}.

-spec validate_loader_opts(list(), [fetcher_opt()])
        -> {ok, {[loader_opt()], [fetcher_opt()], list()}} |
           {error, term()}.
validate_loader_opts(MixedOpts, FetcherOpts) ->
    try
        lists:partition(
          fun ({pre_readiness_update_period, Interval} = Opt) ->
                  ?is_pos_integer(Interval) orelse error({badopt,Opt});
              ({post_readiness_update_period, Interval} = Opt) ->
                  ?is_pos_integer(Interval) orelse error({badopt,Opt});
              (no_cache) ->
                  true;
              (_) ->
                  false
          end,
          MixedOpts)
    of
        {LoaderOpts, OtherOpts} ->
            {ok, {LoaderOpts, FetcherOpts, OtherOpts}}
    catch
        error:{badopt,BadOpt} ->
            {error, BadOpt}
    end.

-spec default_settings(origin()) -> settings().
default_settings({http,_}) ->
    #settings{
       unready_update_period = ?DEFAULT_HTTP_UNREADY_UPDATE_PERIOD,
       ready_update_period = ?DEFAULT_HTTP_READY_UPDATE_PERIOD,
       use_cache = true
      };
default_settings({filesystem,_}) ->
    #settings{
       unready_update_period = ?DEFAULT_FS_UNREADY_UPDATE_PEROID,
       ready_update_period = ?DEFAULT_FS_READY_UPDATE_PERIOD,
       use_cache = false
      }.

-spec customized_settings(settings(), [loader_opt()]) -> settings().
customized_settings(Settings, LoaderOpts) ->
    lists:foldl(
      fun ({pre_readiness_update_period, Interval}, Acc) ->
              Acc#settings{ unready_update_period = Interval };
          ({post_readiness_update_period, Interval}, Acc) ->
              Acc#settings{ ready_update_period = Interval };
          (no_cache, Acc) ->
              Acc#settings{ use_cache = false}
      end,
      Settings, LoaderOpts).

-spec finish_initialization(state()) -> state().
finish_initialization(State)
  when (State#state.settings)#settings.use_cache ->
    CachedDatabasePath = cached_database_path(State),
    Source = {cache,CachedDatabasePath},
    {ok, FetcherPid} = locus_filesystem_load:start_link(Source, undefined),
    State#state{ fetcher_pid = FetcherPid, fetcher_source = Source };
finish_initialization(State) ->
    schedule_update(0, State).

-spec schedule_update(non_neg_integer(), state()) -> state().
schedule_update(Interval, State)
  when State#state.update_timer =:= undefined ->
    NewTimer = erlang:send_after(Interval, self(), begin_update),
    State#state{ update_timer = NewTimer }.

-spec cached_database_path(state()) -> nonempty_string().
cached_database_path(State) ->
    #state{origin = Origin} = State,
    {http, URL} = Origin,
    cached_database_path_for_url(URL).

-spec cached_database_path_for_url(string()) -> nonempty_string().
%% @private
cached_database_path_for_url(URL) ->
    Hash = crypto:hash(sha256, URL),
    HexHash = locus_util:bin_to_hex_str(Hash),
    Filename = HexHash ++ ".mmdb.gz",
    UserCachePath = filename:basedir(user_cache, "locus_erlang"),
    filename:join(UserCachePath, Filename).

%% ------------------------------------------------------------------
%% Internal Function Definitions - Database Updates
%% ------------------------------------------------------------------

-spec begin_update(state()) -> state().
begin_update(State)
  when State#state.fetcher_pid =:= undefined ->
    #state{origin = Origin,
           fetcher_opts = FetcherOpts,
           last_modified = LastModified} = State,

    case Origin of
        {http, URL} ->
            Headers = http_request_headers(LastModified),
            {ok, FetcherPid} = locus_http_download:start_link(URL, Headers, FetcherOpts),
            State#state{ fetcher_pid = FetcherPid, fetcher_source = {remote,URL} };
        {filesystem, _} = Source ->
            {ok, FetcherPid} = locus_filesystem_load:start_link(Source, LastModified),
            State#state{ fetcher_pid = FetcherPid, fetcher_source = Source }
    end.

http_request_headers(undefined) ->
    http_base_request_headers();
http_request_headers(LastModified) ->
    LocalLastModified = calendar:universal_time_to_local_time(LastModified),
    [{"if-modified-since", httpd_util:rfc1123_date(LocalLastModified)}
     | http_base_request_headers()].

http_base_request_headers() ->
    [{"accept", join_http_header_values(
                  ["application/gzip",
                   "application/x-gzip",
                   "application/x-gtar",
                   "application/x-tgz",
                   "application/x-tar",
                   "application/octet-stream"
                  ])},
     {"connection", "close"}].

join_http_header_values(Values) ->
    string:join(Values, "; ").

-spec handle_fetcher_msg(fetcher_msg(), state()) -> {noreply, state()}.
handle_fetcher_msg({event,Event}, State) ->
    #state{fetcher_source = Source} = State,
    case Source of
        {cache,_} ->
            {noreply, State};
        _ ->
            report_event(Event, State),
            {noreply, State}
    end;
handle_fetcher_msg({finished,Status}, State) ->
    #state{fetcher_pid = FetcherPid, fetcher_source = Source} = State,
    expect_linked_process_termination(FetcherPid),
    UpdatedState = State#state{ fetcher_pid = undefined, fetcher_source = undefined },
    case Status of
        dismissed ->
            handle_database_fetch_dismissal(Source, UpdatedState);
        {success, Success} ->
            handle_database_fetch_success(Source, Success, UpdatedState);
        {error, Reason} ->
            handle_database_fetch_error(Source, Reason, UpdatedState)
    end.

-spec handle_cacher_msg(cacher_msg(), state()) -> {noreply, state()}.
handle_cacher_msg({finished,Status}, State) ->
    #state{cacher_pid = CacherPid, cacher_path = CacherPath, cacher_source = Source} = State,

    expect_linked_process_termination(CacherPid),
    UpdatedState = State#state{ cacher_pid = undefined,
                                cacher_path = undefined,
                                cacher_source = undefined },

    case Status of
        success ->
            report_event({cache_attempt_finished, CacherPath, ok}, UpdatedState),
            handle_update_conclusion(Source, UpdatedState);
        {error, Reason} ->
            report_event({cache_attempt_finished, CacherPath, {error,Reason}}, UpdatedState),
            handle_update_conclusion(Source, UpdatedState)
    end.

-spec handle_database_fetch_dismissal(source(), state()) -> {noreply, state()}.
handle_database_fetch_dismissal(Source, State)
  when State#state.last_modified =/= undefined -> % sanity check
    handle_update_conclusion(Source, State).

-spec handle_database_fetch_success(source(), fetcher_success(), state()) -> {noreply, state()}.
handle_database_fetch_success(Source, Success, State) ->
    {BlobFormat, Blob} = fetched_database_format_and_blob(Source, Success),
    case decode_database_from_blob(Source, BlobFormat, Blob) of
        {ok, Version, Parts, BinDatabase} ->
            LastModified = fetched_database_modification_datetime(Source, Success),
            handle_database_decode_success(Source, Version, Parts, BinDatabase, LastModified, State);
        {error, Reason} ->
            handle_database_decode_error(Source, Reason, State)
    end.

-spec handle_database_decode_success(source(), calendar:datetime(), locus_mmdb:parts(),
                                     binary(), calendar:datetime(), state()) -> {noreply, state()}.
handle_database_decode_success(Source, Version, Parts, BinDatabase, LastModified, State) ->
    State2 = State#state{ last_modified = LastModified },
    notify_owner({load_success, Source, Version, Parts}, State2),

    case Source of
        {remote,_} when (State2#state.settings)#settings.use_cache ->
            CachedDatabasePath = cached_database_path(State2),
            CachedDatabaseBlob = make_cached_database_blob(CachedDatabasePath, BinDatabase),
            {ok, CacherPid} = locus_filesystem_store:start_link(CachedDatabasePath, CachedDatabaseBlob,
                                                                LastModified),
            State3 = State2#state{ cacher_pid = CacherPid,
                                   cacher_path = CachedDatabasePath,
                                   cacher_source = Source },
            {noreply, State3};
        _ ->
            handle_update_conclusion(Source, State2)
    end.

-spec handle_database_decode_error(source(), term(), state()) -> {noreply, state()}.
handle_database_decode_error(Source, Reason, State) ->
    notify_owner({load_failure, Source, Reason}, State),
    handle_update_conclusion(Source, State).

-spec handle_database_fetch_error(source(), term(), state()) -> {noreply, state()}.
handle_database_fetch_error(Source, Reason, State) ->
    notify_owner({load_failure, Source, Reason}, State),
    handle_update_conclusion(Source, State).

-spec handle_update_conclusion(source(), state()) -> {noreply, state()}.
handle_update_conclusion(Source, State) ->
    TimeToNextUpdate = time_to_next_update(Source, State),
    UpdatedState = schedule_update(TimeToNextUpdate, State),
    {noreply, UpdatedState}.

-spec time_to_next_update(source(), state()) -> non_neg_integer().
time_to_next_update(LastFetchSource, State) ->
    #state{settings = Settings, last_modified = LastModified} = State,
    #settings{ready_update_period = ReadyPeriod, unready_update_period = UnreadyPeriod} = Settings,
    HasVersionLoaded = (LastModified =/= undefined),

    case LastFetchSource of
        {cache,_} ->
            0; % XXX document this exception
        _ when HasVersionLoaded ->
            ReadyPeriod;
        _ ->
            UnreadyPeriod
    end.

-spec decode_database_from_blob(source(), blob_format(), binary())
        -> {ok, calendar:datetime(), locus_mmdb:parts(), binary()} |
           {error, {decode_database_from_tgz_blob, {atom(), term(), [term()]}}} |
           {error, {decode_database_from_tarball_blob, {atom(), term(), [term()]}}} |
           {error, {decode_database_from_mmdb_blob, {atom(), term(), [term()]}}}.
decode_database_from_blob(Source, tgz, Blob) ->
    decode_database_from_tgz_blob(Source, Blob);
decode_database_from_blob(Source, tarball, Blob) ->
    decode_database_from_tarball_blob(Source, Blob);
decode_database_from_blob(Source, gzip, Blob) ->
    decode_database_from_gzip_blob(Source, Blob);
decode_database_from_blob(Source, gzipped_mmdb, Blob) ->
    decode_database_from_gzipped_mmdb_blob(Source, Blob);
decode_database_from_blob(Source, mmdb, Blob) ->
    decode_database_from_mmdb_blob(Source, Blob);
decode_database_from_blob(Source, unknown, Blob) ->
    decode_database_from_unknown_blob(Source, Blob).

-spec decode_database_from_tgz_blob(source(), binary())
        -> {ok, calendar:datetime(), locus_mmdb:parts(), binary()} |
           {error, {decode_database_from_tarball_blob, {atom(), term(), [term()]}}} |
           {error, {decode_database_from_mmdb_blob, {atom(), term(), [term()]}}}.
decode_database_from_tgz_blob(Source, Blob) ->
    try zlib:gunzip(Blob) of
        Tarball ->
            decode_database_from_tarball_blob(Source, Tarball)
    catch
        Class:Reason ->
            Stacktrace = erlang:get_stacktrace(),
            {error, {decode_database_from_tgz_blob, {Class, Reason, Stacktrace}}}
    end.

-spec decode_database_from_gzip_blob(source(), binary())
        -> {ok, calendar:datetime(), locus_mmdb:parts(), binary()} |
           {error, {decode_database_from_gzip_blob, {atom(), term(), [term()]}}} |
           {error, {decode_database_from_tarball_blob, {atom(), term(), [term()]}}} |
           {error, {decode_database_from_mmdb_blob, {atom(), term(), [term()]}}}.
decode_database_from_gzip_blob(Source, Blob) ->
    try zlib:gunzip(Blob) of
        Uncompressed ->
            decode_database_from_unknown_blob(Source, Uncompressed)
    catch
        Class:Reason ->
            Stacktrace = erlang:get_stacktrace(),
            {error, {decode_database_from_gzip_blob, {Class, Reason, Stacktrace}}}
    end.

-spec decode_database_from_gzipped_mmdb_blob(source(), binary())
        -> {ok, calendar:datetime(), locus_mmdb:parts(), binary()} |
           {error, {decode_database_from_gzipped_mmdb_blob, {atom(), term(), [term()]}}} |
           {error, {decode_database_from_mmdb_blob, {atom(), term(), [term()]}}}.
decode_database_from_gzipped_mmdb_blob(Source, Blob) ->
    try zlib:gunzip(Blob) of
        Uncompressed ->
            decode_database_from_mmdb_blob(Source, Uncompressed)
    catch
        Class:Reason ->
            Stacktrace = erlang:get_stacktrace(),
            {error, {decode_database_from_gzipped_mmdb_blob, {Class, Reason, Stacktrace}}}
    end.

-spec decode_database_from_unknown_blob(source(), binary())
        -> {ok, calendar:datetime(), locus_mmdb:parts(), binary()} |
           {error, {decode_database_from_tarball_blob, {atom(), term(), [term()]}}} |
           {error, {decode_database_from_mmdb_blob, {atom(), term(), [term()]}}}.
decode_database_from_unknown_blob(Source, Blob) ->
    case decode_database_from_tarball_blob(Source, Blob) of
        {ok, _, _, _} = Success ->
            Success;
        _ ->
            decode_database_from_mmdb_blob(Source, Blob)
    end.

-spec decode_database_from_tarball_blob(source(), binary())
        -> {ok, calendar:datetime(), locus_mmdb:parts(), binary()} |
           {error, {decode_database_from_tarball_blob, {atom(), term(), [term()]}}} |
           {error, {decode_database_from_mmdb_blob, {atom(), term(), [term()]}}}.
decode_database_from_tarball_blob(Source, Tarball) ->
    try extract_mmdb_from_tarball_blob(Tarball) of
        BinDatabase ->
            decode_database_from_mmdb_blob(Source, BinDatabase)
    catch
        Class:Reason ->
            Stacktrace = erlang:get_stacktrace(),
            {error, {decode_database_from_tarball_blob, {Class, Reason, Stacktrace}}}
    end.

-spec decode_database_from_mmdb_blob(source(), binary())
        -> {ok, calendar:datetime(), locus_mmdb:parts(), binary()} |
           {error, {decode_database_from_mmdb_blob, {atom(), term(), [term()]}}}.
decode_database_from_mmdb_blob(Source, BinDatabase) ->
    try locus_mmdb:decode_database_parts(Source, BinDatabase) of
        {Version, Parts} ->
            {ok, Version, Parts, BinDatabase}
    catch
        Class:Reason ->
            Stacktrace = erlang:get_stacktrace(),
            {error, {decode_database_from_mmdb_blob, {Class, Reason, Stacktrace}}}
    end.

-spec extract_mmdb_from_tarball_blob(binary()) -> binary().
extract_mmdb_from_tarball_blob(Tarball) ->
    {ok, ContainedPaths} = erl_tar:table({binary,Tarball}),
    {true, DatabasePath} = locus_util:lists_anymap(fun has_mmdb_extension/1, ContainedPaths),
    {ok, [{DatabasePath, BinDatabase}]} =
        erl_tar:extract({binary,Tarball}, [{files, [DatabasePath]}, memory]),
    BinDatabase.

%-spec has_mmdb_extension(nonempty_string()) -> boolean().
has_mmdb_extension({Filename, _Type, _Size, _MTime, _Mode, _Uid, _Gid}) ->
    % FIXME: this a placeholder for OTP 20; due to the incomplete spec
    % of erl_tar:table/2, Dialyzer comes to believe that no strings
    % can be returned, only the above tuple, which in fact is only returned
    % if the 'verbose' option is picked, something that we are definitely
    % not doing.
    case filename_extension_parts(Filename) of
        ["mmdb"|_] -> {true, Filename};
        _ -> false
    end;
has_mmdb_extension(Filename) ->
    case filename_extension_parts(Filename) of
        ["mmdb"|_] -> true;
        _ -> false
    end.

filename_extension_parts(Filename) ->
    filename_extension_parts_recur(Filename, []).

filename_extension_parts_recur(Filename, Acc) ->
    case filename:extension(Filename) of
        "." ++ RawExtension ->
            Extension = string:to_lower(RawExtension),
            UpdatedFilename = lists:sublist(Filename, length(Filename) - length(Extension) - 1),
            UpdatedAcc = [Extension | Acc],
            filename_extension_parts_recur(UpdatedFilename, UpdatedAcc);
        _ ->
            lists:reverse(Acc)
    end.

-spec fetched_database_format_and_blob(source(), fetcher_success()) -> {blob_format(), binary()}.
fetched_database_format_and_blob({remote,_}, #{headers := Headers, body := Body}) ->
    case {lists:keyfind("content-type", 1, Headers), Body} of
        {{_,"application/gzip"}, _} ->
            {gzip, Body};
        {{_,"application/x-gzip"}, _} ->
            {gzip, Body};
        {{_,"application/x-gtar"}, <<?GZIP_MAGIC_BYTES,_/bytes>>} ->
            {tgz, Body};
        {{_,"application/x-tgz"}, _} ->
            {tgz, Body};
        {{_,"application/x-tar"}, _} ->
            {tarball, Body};
        {_, <<?GZIP_MAGIC_BYTES,_/bytes>>} ->
            {gzip, Body};
        _ ->
            {unknown, Body}
    end;
fetched_database_format_and_blob({SourceType,Path}, #{content := Content})
  when SourceType =:= cache;
       SourceType =:= filesystem ->
    case {filename_extension_parts(Path), Content} of
        {["tgz"|_], _} ->
            {tgz, Content};
        {["gz","tar"|_], _} ->
            {tgz, Content};
        {["tar"|_], _} ->
            {tarball, Content};
        {["mmdb"|_], _} ->
            {mmdb, Content};
        {["gz","mmdb"|_], _} ->
            {gzipped_mmdb, Content};
        {_, <<?GZIP_MAGIC_BYTES,_/bytes>>} ->
            {gzip, Content};
        _ ->
            {unknown, Content}
    end.

-spec fetched_database_modification_datetime(source(), fetcher_success()) -> calendar:datetime().
fetched_database_modification_datetime({remote,_}, #{headers := Headers}) ->
    case lists:keyfind("last-modified", 1, Headers) of
        {"last-modified", LastModified} ->
            ({_,_} = ModificationDate) = httpd_util:convert_request_date(LastModified),
            ModificationDate;
        false ->
            {{1970,1,1}, {0,0,0}}
    end;
fetched_database_modification_datetime({cache,_}, #{modified_on := ModificationDate}) ->
    ModificationDate;
fetched_database_modification_datetime({filesystem,_}, #{modified_on := ModificationDate}) ->
    ModificationDate.

-spec expect_linked_process_termination(pid()) -> boolean().
expect_linked_process_termination(Pid) ->
    case locus_util:flush_link_exit(Pid, 5000) of
        true -> true;
        false ->
            exit(Pid, kill),
            locus_util:flush_link_exit(Pid, 1000)
    end.

%-spec make_cached_database_blob(nonempty_string(), binary()) -> binary().
make_cached_database_blob(CachedTarballPath, BinDatabase) ->
    case filename_extension_parts(CachedTarballPath) of
        ["gz","mmdb"|_] -> zlib:gzip(BinDatabase)
    end.

%% ------------------------------------------------------------------
%% Internal Function Definitions - Monitoring and Events
%% ------------------------------------------------------------------

-spec handle_linked_process_death(pid(), term(), state())
        -> {stop, Reason, state()}
    when Reason :: normal | FetcherStopped | CacherStopped,
         FetcherStopped :: {fetched_stopped, pid(), term()},
         CacherStopped :: {cached_stopped, pid(), term()}.
handle_linked_process_death(Pid, _, State)
  when Pid =:= State#state.owner_pid ->
    {stop, normal, State};
handle_linked_process_death(Pid, Reason, State)
  when Pid =:= State#state.fetcher_pid ->
    {stop, {fetcher_stopped, Pid, Reason}, State};
handle_linked_process_death(Pid, Reason, State)
  when Pid =:= State#state.cacher_pid ->
    {stop, {cacher_stopped, Pid, Reason}, State}.

-spec report_event(event(), state()) -> ok.
report_event(Event, State) ->
    notify_owner({event,Event}, State).

-spec notify_owner(msg(), state()) -> ok.
notify_owner(Msg, State) ->
    #state{owner_pid = OwnerPid} = State,
    _ = erlang:send(OwnerPid, {self(),Msg}, [noconnect]),
    ok.