emqx4.3/apps/emqx_plugins/src/emqx_plugins.erl

%%--------------------------------------------------------------------
%% Copyright (c) 2017-2024 EMQ Technologies Co., Ltd. All Rights Reserved.
%%
%% Licensed under the Apache License, Version 2.0 (the "License");
%% you may not use this file except in compliance with the License.
%% You may obtain a copy of the License at
%%
%%     http://www.apache.org/licenses/LICENSE-2.0
%%
%% Unless required by applicable law or agreed to in writing, software
%% distributed under the License is distributed on an "AS IS" BASIS,
%% WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
%% See the License for the specific language governing permissions and
%% limitations under the License.
%%--------------------------------------------------------------------

-module(emqx_plugins).

-include_lib("emqx/include/logger.hrl").
-include("emqx_plugins.hrl").

-ifdef(TEST).
-include_lib("eunit/include/eunit.hrl").
-endif.

-export([
    describe/1,
    plugin_avsc/1,
    plugin_i18n/1,
    plugin_avro/1,
    parse_name_vsn/1,
    make_name_vsn_string/2
]).

%% Package operations
-export([
    ensure_installed/1,
    ensure_uninstalled/1,
    ensure_enabled/1,
    ensure_enabled/2,
    ensure_enabled/3,
    ensure_disabled/1,
    purge/1,
    delete_package/1
]).

%% Plugin runtime management
-export([
    ensure_started/0,
    ensure_started/1,
    ensure_stopped/0,
    ensure_stopped/1,
    restart/1,
    list/0
]).

%% Plugin config APIs
-export([
    get_config/1,
    get_config/2,
    get_config/3,
    get_config/4,
    put_config/3
]).

%% Package utils
-export([
    decode_plugin_avro_config/2,
    install_dir/0,
    avsc_file_path/1
]).

%% `emqx_config_handler' API
-export([
    post_config_update/5
]).

%% RPC call
-export([get_tar/1]).

%% Internal export
-export([do_ensure_started/1]).
%% for test cases
-export([put_config_internal/2]).

-ifdef(TEST).
-compile(export_all).
-compile(nowarn_export_all).
-endif.

%% Defines
-define(PLUGIN_PERSIS_CONFIG_KEY(NameVsn), {?MODULE, NameVsn}).

-define(RAW_BIN, binary).
-define(JSON_MAP, json_map).

%% "my_plugin-0.1.0"
-type name_vsn() :: binary() | string().
%% the parse result of the JSON info file
-type plugin() :: map().
-type schema_json() :: map().
-type i18n_json() :: map().
-type avro_binary() :: binary().
-type plugin_config() :: map().
-type position() :: no_move | front | rear | {before, name_vsn()} | {behind, name_vsn()}.

%%--------------------------------------------------------------------
%% APIs
%%--------------------------------------------------------------------

%% @doc Describe a plugin.
-spec describe(name_vsn()) -> {ok, plugin()} | {error, any()}.
describe(NameVsn) ->
    read_plugin_info(NameVsn, #{fill_readme => true}).

-spec plugin_avsc(name_vsn()) -> {ok, schema_json()} | {error, any()}.
plugin_avsc(NameVsn) ->
    read_plugin_avsc(NameVsn).

-spec plugin_i18n(name_vsn()) -> {ok, i18n_json()} | {error, any()}.
plugin_i18n(NameVsn) ->
    read_plugin_i18n(NameVsn).

-spec plugin_avro(name_vsn()) -> {ok, avro_binary()} | {error, any()}.
plugin_avro(NameVsn) ->
    read_plugin_avro(NameVsn).

parse_name_vsn(NameVsn) when is_binary(NameVsn) ->
    parse_name_vsn(binary_to_list(NameVsn));
parse_name_vsn(NameVsn) when is_list(NameVsn) ->
    case lists:splitwith(fun(X) -> X =/= $- end, NameVsn) of
        {AppName, [$- | Vsn]} -> {ok, list_to_atom(AppName), Vsn};
        _ -> {error, "bad_name_vsn"}
    end.

make_name_vsn_string(Name, Vsn) ->
    binary_to_list(iolist_to_binary([Name, "-", Vsn])).

%%--------------------------------------------------------------------
%% Package operations

%% @doc Install a .tar.gz package placed in install_dir.
-spec ensure_installed(name_vsn()) -> ok | {error, map()}.
ensure_installed(NameVsn) ->
    case read_plugin_info(NameVsn, #{}) of
        {ok, _} ->
            ok;
        {error, _} ->
            ok = purge(NameVsn),
            do_ensure_installed(NameVsn)
    end.

%% @doc Ensure files and directories for the given plugin are being deleted.
%% If a plugin is running, or enabled, an error is returned.
-spec ensure_uninstalled(name_vsn()) -> ok | {error, any()}.
ensure_uninstalled(NameVsn) ->
    case read_plugin_info(NameVsn, #{}) of
        {ok, #{running_status := RunningSt}} when RunningSt =/= stopped ->
            {error, #{
                error_msg => "bad_plugin_running_status",
                hint => "stop_the_plugin_first"
            }};
        {ok, #{config_status := enabled}} ->
            {error, #{
                error_msg => "bad_plugin_config_status",
                hint => "disable_the_plugin_first"
            }};
        _ ->
            purge(NameVsn),
            ensure_delete(NameVsn)
    end.

%% @doc Ensure a plugin is enabled to the end of the plugins list.
-spec ensure_enabled(name_vsn()) -> ok | {error, any()}.
ensure_enabled(NameVsn) ->
    ensure_enabled(NameVsn, no_move).

%% @doc Ensure a plugin is enabled at the given position of the plugin list.
-spec ensure_enabled(name_vsn(), position()) -> ok | {error, any()}.
ensure_enabled(NameVsn, Position) ->
    ensure_state(NameVsn, Position, _Enabled = true, _ConfLocation = local).

-spec ensure_enabled(name_vsn(), position(), local | global) -> ok | {error, any()}.
ensure_enabled(NameVsn, Position, ConfLocation) when
    ConfLocation =:= local; ConfLocation =:= global
->
    ensure_state(NameVsn, Position, _Enabled = true, ConfLocation).

%% @doc Ensure a plugin is disabled.
-spec ensure_disabled(name_vsn()) -> ok | {error, any()}.
ensure_disabled(NameVsn) ->
    ensure_state(NameVsn, no_move, false, _ConfLocation = local).

%% @doc Delete extracted dir
%% In case one lib is shared by multiple plugins.
%% it might be the case that purging one plugin's install dir
%% will cause deletion of loaded beams.
%% It should not be a problem, because shared lib should
%% reside in all the plugin install dirs.
-spec purge(name_vsn()) -> ok.
purge(NameVsn) ->
    _ = maybe_purge_plugin_config(NameVsn),
    purge_plugin(NameVsn).

%% @doc Delete the package file.
-spec delete_package(name_vsn()) -> ok.
delete_package(NameVsn) ->
    File = pkg_file_path(NameVsn),
    _ = emqx_plugins_serde:delete_schema(NameVsn),
    case file:delete(File) of
        ok ->
            ?SLOG(info, #{msg => "purged_plugin_dir", path => File}),
            ok;
        {error, enoent} ->
            ok;
        {error, Reason} ->
            ?SLOG(error, #{
                msg => "failed_to_delete_package_file",
                path => File,
                reason => Reason
            }),
            {error, Reason}
    end.

%%--------------------------------------------------------------------
%% Plugin runtime management

%% @doc Start all configured plugins are started.
-spec ensure_started() -> ok.
ensure_started() ->
    ok = for_plugins(fun ?MODULE:do_ensure_started/1).

%% @doc Start a plugin from Management API or CLI.
%% the input is a <name>-<vsn> string.
-spec ensure_started(name_vsn()) -> ok | {error, term()}.
ensure_started(NameVsn) ->
    case do_ensure_started(NameVsn) of
        ok ->
            ok;
        {error, Reason} ->
            ?SLOG(alert, Reason#{msg => "failed_to_start_plugin"}),
            {error, Reason}
    end.

%% @doc Stop all plugins before broker stops.
-spec ensure_stopped() -> ok.
ensure_stopped() ->
    for_plugins(fun ?MODULE:ensure_stopped/1).

%% @doc Stop a plugin from Management API or CLI.
-spec ensure_stopped(name_vsn()) -> ok | {error, term()}.
ensure_stopped(NameVsn) ->
    tryit(
        "stop_plugin",
        fun() ->
            Plugin = do_read_plugin(NameVsn),
            ensure_apps_stopped(Plugin)
        end
    ).

get_config(Name, Vsn, Options, Default) ->
    get_config(make_name_vsn_string(Name, Vsn), Options, Default).

-spec get_config(name_vsn()) ->
    {ok, plugin_config()}
    | {error, term()}.
get_config(NameVsn) ->
    get_config(bin(NameVsn), #{format => ?CONFIG_FORMAT_MAP}).

-spec get_config(name_vsn(), Options :: map()) ->
    {ok, avro_binary() | plugin_config()}
    | {error, term()}.
get_config(NameVsn, #{format := ?CONFIG_FORMAT_AVRO}) ->
    %% no default value when get raw binary config
    case read_plugin_avro(NameVsn) of
        {ok, _AvroJson} = Res -> Res;
        {error, _Reason} = Err -> Err
    end;
get_config(NameVsn, Options = #{format := ?CONFIG_FORMAT_MAP}) ->
    get_config(NameVsn, Options, #{}).

get_config(NameVsn, #{format := ?CONFIG_FORMAT_MAP}, Default) ->
    {ok, persistent_term:get(?PLUGIN_PERSIS_CONFIG_KEY(NameVsn), Default)}.

%% @doc Update plugin's config.
%% RPC call from Management API or CLI.
%% the avro Json Map and plugin config ALWAYS be valid before calling this function.
put_config(NameVsn, AvroJsonMap, _DecodedPluginConfig) ->
    AvroJsonBin = emqx_utils_json:encode(AvroJsonMap),
    ok = write_avro_bin(NameVsn, AvroJsonBin),
    ok = persistent_term:put(?PLUGIN_PERSIS_CONFIG_KEY(NameVsn), AvroJsonMap),
    ok.

%% @doc Stop and then start the plugin.
restart(NameVsn) ->
    case ensure_stopped(NameVsn) of
        ok -> ensure_started(NameVsn);
        {error, Reason} -> {error, Reason}
    end.

%% @doc List all installed plugins.
%% Including the ones that are installed, but not enabled in config.
-spec list() -> [plugin()].
list() ->
    Pattern = filename:join([install_dir(), "*", "release.json"]),
    All = lists:filtermap(
        fun(JsonFilePath) ->
            [_, NameVsn | _] = lists:reverse(filename:split(JsonFilePath)),
            case read_plugin_info(NameVsn, #{}) of
                {ok, Info} ->
                    {true, Info};
                {error, Reason} ->
                    ?SLOG(warning, Reason),
                    false
            end
        end,
        filelib:wildcard(Pattern)
    ),
    do_list(configured(), All).

%%--------------------------------------------------------------------
%% Package utils

-spec decode_plugin_avro_config(name_vsn(), map() | binary()) -> {ok, map()} | {error, any()}.
decode_plugin_avro_config(NameVsn, AvroJsonMap) when is_map(AvroJsonMap) ->
    decode_plugin_avro_config(NameVsn, emqx_utils_json:encode(AvroJsonMap));
decode_plugin_avro_config(NameVsn, AvroJsonBin) ->
    case emqx_plugins_serde:decode(NameVsn, AvroJsonBin) of
        {ok, Config} -> {ok, Config};
        {error, ReasonMap} -> {error, ReasonMap}
    end.

get_config_interal(Key, Default) when is_atom(Key) ->
    get_config_interal([Key], Default);
get_config_interal(Path, Default) ->
    emqx_conf:get([?CONF_ROOT | Path], Default).

put_config_internal(Key, Value) ->
    do_put_config_internal(Key, Value, _ConfLocation = local).

-spec get_tar(name_vsn()) -> {ok, binary()} | {error, any}.
get_tar(NameVsn) ->
    TarGz = pkg_file_path(NameVsn),
    case file:read_file(TarGz) of
        {ok, Content} ->
            {ok, Content};
        {error, _} ->
            case maybe_create_tar(NameVsn, TarGz, install_dir()) of
                ok ->
                    file:read_file(TarGz);
                Err ->
                    Err
            end
    end.

%%--------------------------------------------------------------------
%% Internal
%%--------------------------------------------------------------------

maybe_create_tar(NameVsn, TarGzName, InstallDir) when is_binary(InstallDir) ->
    maybe_create_tar(NameVsn, TarGzName, binary_to_list(InstallDir));
maybe_create_tar(NameVsn, TarGzName, InstallDir) ->
    case filelib:wildcard(filename:join(plugin_dir(NameVsn), "**")) of
        [_ | _] = PluginFiles ->
            InstallDir1 = string:trim(InstallDir, trailing, "/") ++ "/",
            PluginFiles1 = [{string:prefix(F, InstallDir1), F} || F <- PluginFiles],
            erl_tar:create(TarGzName, PluginFiles1, [compressed]);
        _ ->
            {error, plugin_not_found}
    end.

write_tar_file_content(BaseDir, TarContent) ->
    lists:foreach(
        fun({Name, Bin}) ->
            Filename = filename:join(BaseDir, Name),
            ok = filelib:ensure_dir(Filename),
            ok = file:write_file(Filename, Bin)
        end,
        TarContent
    ).

delete_tar_file_content(BaseDir, TarContent) ->
    lists:foreach(
        fun({Name, _}) ->
            Filename = filename:join(BaseDir, Name),
            case filelib:is_file(Filename) of
                true ->
                    TopDirOrFile = top_dir(BaseDir, Filename),
                    ok = file:del_dir_r(TopDirOrFile);
                false ->
                    %% probably already deleted
                    ok
            end
        end,
        TarContent
    ).

top_dir(BaseDir0, DirOrFile) ->
    BaseDir = normalize_dir(BaseDir0),
    case filename:dirname(DirOrFile) of
        RockBottom when RockBottom =:= "/" orelse RockBottom =:= "." ->
            throw({out_of_bounds, DirOrFile});
        BaseDir ->
            DirOrFile;
        Parent ->
            top_dir(BaseDir, Parent)
    end.

normalize_dir(Dir) ->
    %% Get rid of possible trailing slash
    filename:join([Dir, ""]).

-ifdef(TEST).
normalize_dir_test_() ->
    [
        ?_assertEqual("foo", normalize_dir("foo")),
        ?_assertEqual("foo", normalize_dir("foo/")),
        ?_assertEqual("/foo", normalize_dir("/foo")),
        ?_assertEqual("/foo", normalize_dir("/foo/"))
    ].

top_dir_test_() ->
    [
        ?_assertEqual("base/foo", top_dir("base", filename:join(["base", "foo", "bar"]))),
        ?_assertEqual("/base/foo", top_dir("/base", filename:join(["/", "base", "foo", "bar"]))),
        ?_assertEqual("/base/foo", top_dir("/base/", filename:join(["/", "base", "foo", "bar"]))),
        ?_assertThrow({out_of_bounds, _}, top_dir("/base", filename:join(["/", "base"]))),
        ?_assertThrow({out_of_bounds, _}, top_dir("/base", filename:join(["/", "foo", "bar"])))
    ].
-endif.

do_ensure_installed(NameVsn) ->
    TarGz = pkg_file_path(NameVsn),
    case erl_tar:extract(TarGz, [compressed, memory]) of
        {ok, TarContent} ->
            ok = write_tar_file_content(install_dir(), TarContent),
            case read_plugin_info(NameVsn, #{}) of
                {ok, _} ->
                    ok = maybe_post_op_after_install(NameVsn),
                    ok;
                {error, Reason} ->
                    ?SLOG(warning, Reason#{msg => "failed_to_read_after_install"}),
                    ok = delete_tar_file_content(install_dir(), TarContent),
                    {error, Reason}
            end;
        {error, {_, enoent}} ->
            {error, #{
                error_msg => "failed_to_extract_plugin_package",
                path => TarGz,
                reason => not_found
            }};
        {error, Reason} ->
            {error, #{
                error_msg => "bad_plugin_package",
                path => TarGz,
                reason => Reason
            }}
    end.

ensure_delete(NameVsn0) ->
    NameVsn = bin(NameVsn0),
    List = configured(),
    put_configured(lists:filter(fun(#{name_vsn := N1}) -> bin(N1) =/= NameVsn end, List)),
    ok.

ensure_state(NameVsn, Position, State, ConfLocation) when is_binary(NameVsn) ->
    ensure_state(binary_to_list(NameVsn), Position, State, ConfLocation);
ensure_state(NameVsn, Position, State, ConfLocation) ->
    case read_plugin_info(NameVsn, #{}) of
        {ok, _} ->
            Item = #{
                name_vsn => NameVsn,
                enable => State
            },
            tryit(
                "ensure_state",
                fun() -> ensure_configured(Item, Position, ConfLocation) end
            );
        {error, Reason} ->
            {error, Reason}
    end.

ensure_configured(#{name_vsn := NameVsn} = Item, Position, ConfLocation) ->
    Configured = configured(),
    SplitFun = fun(#{name_vsn := Nv}) -> bin(Nv) =/= bin(NameVsn) end,
    {Front, Rear} = lists:splitwith(SplitFun, Configured),
    NewConfigured =
        case Rear of
            [_ | More] when Position =:= no_move ->
                Front ++ [Item | More];
            [_ | More] ->
                add_new_configured(Front ++ More, Position, Item);
            [] ->
                add_new_configured(Configured, Position, Item)
        end,
    ok = put_configured(NewConfigured, ConfLocation).

add_new_configured(Configured, no_move, Item) ->
    %% default to rear
    add_new_configured(Configured, rear, Item);
add_new_configured(Configured, front, Item) ->
    [Item | Configured];
add_new_configured(Configured, rear, Item) ->
    Configured ++ [Item];
add_new_configured(Configured, {Action, NameVsn}, Item) ->
    SplitFun = fun(#{name_vsn := Nv}) -> bin(Nv) =/= bin(NameVsn) end,
    {Front, Rear} = lists:splitwith(SplitFun, Configured),
    Rear =:= [] andalso
        throw(#{
            error_msg => "position_anchor_plugin_not_configured",
            hint => "maybe_install_and_configure",
            name_vsn => NameVsn
        }),
    case Action of
        before ->
            Front ++ [Item | Rear];
        behind ->
            [Anchor | Rear0] = Rear,
            Front ++ [Anchor, Item | Rear0]
    end.

maybe_purge_plugin_config(NameVsn) ->
    _ = persistent_term:erase(?PLUGIN_PERSIS_CONFIG_KEY(NameVsn)),
    ok.

purge_plugin(NameVsn) ->
    Dir = plugin_dir(NameVsn),
    purge_plugin_dir(Dir).

purge_plugin_dir(Dir) ->
    case file:del_dir_r(Dir) of
        ok ->
            ?SLOG(info, #{
                msg => "purged_plugin_dir",
                dir => Dir
            });
        {error, enoent} ->
            ok;
        {error, Reason} ->
            ?SLOG(error, #{
                msg => "failed_to_purge_plugin_dir",
                dir => Dir,
                reason => Reason
            }),
            {error, Reason}
    end.

%% Make sure configured ones are ordered in front.
do_list([], All) ->
    All;
do_list([#{name_vsn := NameVsn} | Rest], All) ->
    SplitF = fun(#{<<"name">> := Name, <<"rel_vsn">> := Vsn}) ->
        bin([Name, "-", Vsn]) =/= bin(NameVsn)
    end,
    case lists:splitwith(SplitF, All) of
        {_, []} ->
            ?SLOG(warning, #{
                msg => "configured_plugin_not_installed",
                name_vsn => NameVsn
            }),
            do_list(Rest, All);
        {Front, [I | Rear]} ->
            [I | do_list(Rest, Front ++ Rear)]
    end.

do_ensure_started(NameVsn) ->
    tryit(
        "start_plugins",
        fun() ->
            case ensure_exists_and_installed(NameVsn) of
                ok ->
                    Plugin = do_read_plugin(NameVsn),
                    ok = load_code_start_apps(NameVsn, Plugin);
                {error, plugin_not_found} ->
                    ?SLOG(error, #{
                        error_msg => "plugin_not_found",
                        name_vsn => NameVsn
                    }),
                    ok
            end
        end
    ).

%%--------------------------------------------------------------------

%% try the function, catch 'throw' exceptions as normal 'error' return
%% other exceptions with stacktrace logged.
tryit(WhichOp, F) ->
    try
        F()
    catch
        throw:ReasonMap ->
            %% thrown exceptions are known errors
            %% translate to a return value without stacktrace
            {error, ReasonMap};
        error:Reason:Stacktrace ->
            %% unexpected errors, log stacktrace
            ?SLOG(warning, #{
                msg => "plugin_op_failed",
                which_op => WhichOp,
                exception => Reason,
                stacktrace => Stacktrace
            }),
            {error, {failed, WhichOp}}
    end.

%% read plugin info from the JSON file
%% returns {ok, Info} or {error, Reason}
read_plugin_info(NameVsn, Options) ->
    tryit(
        atom_to_list(?FUNCTION_NAME),
        fun() -> {ok, do_read_plugin2(NameVsn, Options)} end
    ).

do_read_plugin(NameVsn) ->
    do_read_plugin2(NameVsn, #{}).

do_read_plugin2(NameVsn, Option) ->
    do_read_plugin3(NameVsn, info_file_path(NameVsn), Option).

do_read_plugin3(NameVsn, InfoFilePath, Options) ->
    {ok, PlainMap} = (read_file_fun(InfoFilePath, "bad_info_file", #{read_mode => ?JSON_MAP}))(),
    Info0 = check_plugin(PlainMap, NameVsn, InfoFilePath),
    Info1 = plugins_readme(NameVsn, Options, Info0),
    plugin_status(NameVsn, Info1).

read_plugin_avsc(NameVsn) ->
    read_plugin_avsc(NameVsn, #{read_mode => ?JSON_MAP}).
read_plugin_avsc(NameVsn, Options) ->
    tryit(
        atom_to_list(?FUNCTION_NAME),
        read_file_fun(avsc_file_path(NameVsn), "bad_avsc_file", Options)
    ).

read_plugin_i18n(NameVsn) ->
    read_plugin_i18n(NameVsn, #{read_mode => ?JSON_MAP}).
read_plugin_i18n(NameVsn, Options) ->
    tryit(
        atom_to_list(?FUNCTION_NAME),
        read_file_fun(i18n_file_path(NameVsn), "bad_i18n_file", Options)
    ).

read_plugin_avro(NameVsn) ->
    read_plugin_avro(NameVsn, #{read_mode => ?RAW_BIN}).
read_plugin_avro(NameVsn, Options) ->
    tryit(
        atom_to_list(?FUNCTION_NAME),
        read_file_fun(avro_config_file(NameVsn), "bad_avro_file", Options)
    ).

ensure_exists_and_installed(NameVsn) ->
    case filelib:is_dir(plugin_dir(NameVsn)) of
        true ->
            ok;
        false ->
            %% Do we have the package, but it's not extracted yet?
            case get_tar(NameVsn) of
                {ok, TarContent} ->
                    ok = file:write_file(pkg_file_path(NameVsn), TarContent),
                    ok = do_ensure_installed(NameVsn);
                _ ->
                    %% If not, try to get it from the cluster.
                    do_get_from_cluster(NameVsn)
            end
    end.

do_get_from_cluster(NameVsn) ->
    Nodes = [N || N <- mria:running_nodes(), N /= node()],
    case get_from_any_node(Nodes, NameVsn, []) of
        {ok, TarContent} ->
            ok = file:write_file(pkg_file_path(NameVsn), TarContent),
            ok = do_ensure_installed(NameVsn);
        {error, NodeErrors} when Nodes =/= [] ->
            ?SLOG(error, #{
                msg => "failed_to_copy_plugin_from_other_nodes",
                name_vsn => NameVsn,
                node_errors => NodeErrors
            }),
            {error, plugin_not_found};
        {error, _} ->
            ?SLOG(error, #{
                msg => "no_nodes_to_copy_plugin_from",
                name_vsn => NameVsn
            }),
            {error, plugin_not_found}
    end.

get_from_any_node([], _NameVsn, Errors) ->
    {error, Errors};
get_from_any_node([Node | T], NameVsn, Errors) ->
    case emqx_plugins_proto_v1:get_tar(Node, NameVsn, infinity) of
        {ok, _} = Res ->
            Res;
        Err ->
            get_from_any_node(T, NameVsn, [{Node, Err} | Errors])
    end.

plugins_readme(NameVsn, #{fill_readme := true}, Info) ->
    case file:read_file(readme_file(NameVsn)) of
        {ok, Bin} -> Info#{readme => Bin};
        _ -> Info#{readme => <<>>}
    end;
plugins_readme(_NameVsn, _Options, Info) ->
    Info.

plugin_status(NameVsn, Info) ->
    {ok, AppName, _AppVsn} = parse_name_vsn(NameVsn),
    RunningSt =
        case application:get_key(AppName, vsn) of
            {ok, _} ->
                case lists:keyfind(AppName, 1, running_apps()) of
                    {AppName, _} -> running;
                    _ -> loaded
                end;
            undefined ->
                stopped
        end,
    Configured = lists:filtermap(
        fun(#{name_vsn := Nv, enable := St}) ->
            case bin(Nv) =:= bin(NameVsn) of
                true -> {true, St};
                false -> false
            end
        end,
        configured()
    ),
    ConfSt =
        case Configured of
            [] -> not_configured;
            [true] -> enabled;
            [false] -> disabled
        end,
    Info#{
        running_status => RunningSt,
        config_status => ConfSt
    }.

check_plugin(
    #{
        <<"name">> := Name,
        <<"rel_vsn">> := Vsn,
        <<"rel_apps">> := Apps,
        <<"description">> := _
    } = Info,
    NameVsn,
    FilePath
) ->
    case bin(NameVsn) =:= bin([Name, "-", Vsn]) of
        true ->
            try
                %% assert
                [_ | _] = Apps,
                %% validate if the list is all <app>-<vsn> strings
                lists:foreach(fun(App) -> {ok, _, _} = parse_name_vsn(App) end, Apps)
            catch
                _:_ ->
                    throw(#{
                        error_msg => "bad_rel_apps",
                        rel_apps => Apps,
                        hint => "A non-empty string list of app_name-app_vsn format"
                    })
            end,
            Info;
        false ->
            throw(#{
                error_msg => "name_vsn_mismatch",
                name_vsn => NameVsn,
                path => FilePath,
                name => Name,
                rel_vsn => Vsn
            })
    end;
check_plugin(_What, NameVsn, File) ->
    throw(#{
        error_msg => "bad_info_file_content",
        mandatory_fields => [rel_vsn, name, rel_apps, description],
        name_vsn => NameVsn,
        path => File
    }).

load_code_start_apps(RelNameVsn, #{<<"rel_apps">> := Apps}) ->
    LibDir = filename:join([install_dir(), RelNameVsn]),
    RunningApps = running_apps(),
    %% load plugin apps and beam code
    AppNames =
        lists:map(
            fun(AppNameVsn) ->
                {ok, AppName, AppVsn} = parse_name_vsn(AppNameVsn),
                EbinDir = filename:join([LibDir, AppNameVsn, "ebin"]),
                ok = load_plugin_app(AppName, AppVsn, EbinDir, RunningApps),
                AppName
            end,
            Apps
        ),
    lists:foreach(fun start_app/1, AppNames).

load_plugin_app(AppName, AppVsn, Ebin, RunningApps) ->
    case lists:keyfind(AppName, 1, RunningApps) of
        false ->
            do_load_plugin_app(AppName, Ebin);
        {_, Vsn} ->
            case bin(Vsn) =:= bin(AppVsn) of
                true ->
                    %% already started on the exact version
                    ok;
                false ->
                    %% running but a different version
                    ?SLOG(warning, #{
                        msg => "plugin_app_already_running",
                        name => AppName,
                        running_vsn => Vsn,
                        loading_vsn => AppVsn
                    })
            end
    end.

do_load_plugin_app(AppName, Ebin) when is_binary(Ebin) ->
    do_load_plugin_app(AppName, binary_to_list(Ebin));
do_load_plugin_app(AppName, Ebin) ->
    _ = code:add_patha(Ebin),
    Modules = filelib:wildcard(filename:join([Ebin, "*.beam"])),
    lists:foreach(
        fun(BeamFile) ->
            Module = list_to_atom(filename:basename(BeamFile, ".beam")),
            _ = code:purge(Module),
            case code:load_file(Module) of
                {module, _} ->
                    ok;
                {error, Reason} ->
                    throw(#{
                        error_msg => "failed_to_load_plugin_beam",
                        path => BeamFile,
                        reason => Reason
                    })
            end
        end,
        Modules
    ),
    case application:load(AppName) of
        ok ->
            ok;
        {error, {already_loaded, _}} ->
            ok;
        {error, Reason} ->
            throw(#{
                error_msg => "failed_to_load_plugin_app",
                name => AppName,
                reason => Reason
            })
    end.

start_app(App) ->
    case application:ensure_all_started(App) of
        {ok, Started} ->
            case Started =/= [] of
                true -> ?SLOG(debug, #{msg => "started_plugin_apps", apps => Started});
                false -> ok
            end,
            ?SLOG(debug, #{msg => "started_plugin_app", app => App}),
            ok;
        {error, {ErrApp, Reason}} ->
            throw(#{
                error_msg => "failed_to_start_plugin_app",
                app => App,
                err_app => ErrApp,
                reason => Reason
            })
    end.

%% Stop all apps installed by the plugin package,
%% but not the ones shared with others.
ensure_apps_stopped(#{<<"rel_apps">> := Apps}) ->
    %% load plugin apps and beam code
    AppsToStop =
        lists:map(
            fun(NameVsn) ->
                {ok, AppName, _AppVsn} = parse_name_vsn(NameVsn),
                AppName
            end,
            Apps
        ),
    case tryit("stop_apps", fun() -> stop_apps(AppsToStop) end) of
        {ok, []} ->
            %% all apps stopped
            ok;
        {ok, Left} ->
            ?SLOG(warning, #{
                msg => "unabled_to_stop_plugin_apps",
                apps => Left,
                reason => "running_apps_still_depends_on_this_apps"
            }),
            ok;
        {error, Reason} ->
            {error, Reason}
    end.

stop_apps(Apps) ->
    RunningApps = running_apps(),
    case do_stop_apps(Apps, [], RunningApps) of
        %% all stopped
        {ok, []} -> {ok, []};
        %% no progress
        {ok, Remain} when Remain =:= Apps -> {ok, Apps};
        %% try again
        {ok, Remain} -> stop_apps(Remain)
    end.

do_stop_apps([], Remain, _AllApps) ->
    {ok, lists:reverse(Remain)};
do_stop_apps([App | Apps], Remain, RunningApps) ->
    case is_needed_by_any(App, RunningApps) of
        true ->
            do_stop_apps(Apps, [App | Remain], RunningApps);
        false ->
            ok = stop_app(App),
            do_stop_apps(Apps, Remain, RunningApps)
    end.

stop_app(App) ->
    case application:stop(App) of
        ok ->
            ?SLOG(debug, #{msg => "stop_plugin_successfully", app => App}),
            ok = unload_moudle_and_app(App);
        {error, {not_started, App}} ->
            ?SLOG(debug, #{msg => "plugin_not_started", app => App}),
            ok = unload_moudle_and_app(App);
        {error, Reason} ->
            throw(#{error_msg => "failed_to_stop_app", app => App, reason => Reason})
    end.

unload_moudle_and_app(App) ->
    case application:get_key(App, modules) of
        {ok, Modules} -> lists:foreach(fun code:soft_purge/1, Modules);
        _ -> ok
    end,
    _ = application:unload(App),
    ok.

is_needed_by_any(AppToStop, RunningApps) ->
    lists:any(
        fun({RunningApp, _RunningAppVsn}) ->
            is_needed_by(AppToStop, RunningApp)
        end,
        RunningApps
    ).

is_needed_by(AppToStop, AppToStop) ->
    false;
is_needed_by(AppToStop, RunningApp) ->
    case application:get_key(RunningApp, applications) of
        {ok, Deps} -> lists:member(AppToStop, Deps);
        undefined -> false
    end.

do_put_config_internal(Key, Value, ConfLocation) when is_atom(Key) ->
    do_put_config_internal([Key], Value, ConfLocation);
do_put_config_internal(Path, Values, _ConfLocation = local) when is_list(Path) ->
    Opts = #{rawconf_with_defaults => true, override_to => cluster},
    %% Already in cluster_rpc, don't use emqx_conf:update, dead calls
    case emqx:update_config([?CONF_ROOT | Path], bin_key(Values), Opts) of
        {ok, _} -> ok;
        Error -> Error
    end;
do_put_config_internal(Path, Values, _ConfLocation = global) when is_list(Path) ->
    Opts = #{rawconf_with_defaults => true, override_to => cluster},
    case emqx_conf:update([?CONF_ROOT | Path], bin_key(Values), Opts) of
        {ok, _} -> ok;
        Error -> Error
    end.

%%--------------------------------------------------------------------
%% `emqx_config_handler' API
%%--------------------------------------------------------------------

post_config_update([?CONF_ROOT], _Req, #{states := NewStates}, #{states := OldStates}, _Envs) ->
    NewStatesIndex = maps:from_list([{NV, S} || S = #{name_vsn := NV} <- NewStates]),
    OldStatesIndex = maps:from_list([{NV, S} || S = #{name_vsn := NV} <- OldStates]),
    #{changed := Changed} = emqx_utils_maps:diff_maps(NewStatesIndex, OldStatesIndex),
    maps:foreach(fun enable_disable_plugin/2, Changed),
    ok;
post_config_update(_Path, _Req, _NewConf, _OldConf, _Envs) ->
    ok.

enable_disable_plugin(NameVsn, {#{enable := true}, #{enable := false}}) ->
    %% errors are already logged in this fn
    _ = ensure_stopped(NameVsn),
    ok;
enable_disable_plugin(NameVsn, {#{enable := false}, #{enable := true}}) ->
    %% errors are already logged in this fn
    _ = ensure_started(NameVsn),
    ok;
enable_disable_plugin(_NameVsn, _Diff) ->
    ok.

%%--------------------------------------------------------------------
%% Helper functions
%%--------------------------------------------------------------------

install_dir() ->
    get_config_interal(install_dir, "").

put_configured(Configured) ->
    put_configured(Configured, _ConfLocation = local).

put_configured(Configured, ConfLocation) ->
    ok = do_put_config_internal(states, bin_key(Configured), ConfLocation).

configured() ->
    get_config_interal(states, []).

for_plugins(ActionFun) ->
    case lists:flatmap(fun(I) -> for_plugin(I, ActionFun) end, configured()) of
        [] -> ok;
        Errors -> erlang:error(#{function => ActionFun, errors => Errors})
    end.

for_plugin(#{name_vsn := NameVsn, enable := true}, Fun) ->
    case Fun(NameVsn) of
        ok -> [];
        {error, Reason} -> [{NameVsn, Reason}]
    end;
for_plugin(#{name_vsn := NameVsn, enable := false}, _Fun) ->
    ?SLOG(debug, #{
        msg => "plugin_disabled",
        name_vsn => NameVsn
    }),
    [].

maybe_post_op_after_install(NameVsn) ->
    _ = maybe_load_config_schema(NameVsn),
    _ = maybe_create_config_dir(NameVsn),
    ok.

maybe_load_config_schema(NameVsn) ->
    AvscPath = avsc_file_path(NameVsn),
    filelib:is_regular(AvscPath) andalso
        do_load_config_schema(NameVsn, AvscPath).

do_load_config_schema(NameVsn, AvscPath) ->
    case emqx_plugins_serde:add_schema(NameVsn, AvscPath) of
        ok -> ok;
        {error, already_exists} -> ok;
        {error, _Reason} -> ok
    end.

maybe_create_config_dir(NameVsn) ->
    ConfigDir = plugin_config_dir(NameVsn),
    case filelib:ensure_path(ConfigDir) of
        ok ->
            ok;
        {error, Reason} ->
            ?SLOG(warning, #{
                msg => "failed_to_create_plugin_config_dir",
                dir => ConfigDir,
                reason => Reason
            }),
            {error, {mkdir_failed, ConfigDir, Reason}}
    end.

write_avro_bin(NameVsn, AvroBin) ->
    ok = file:write_file(avro_config_file(NameVsn), AvroBin).

read_file_fun(Path, ErrMsg, #{read_mode := ?RAW_BIN}) ->
    fun() ->
        case file:read_file(Path) of
            {ok, Bin} ->
                {ok, Bin};
            {error, Reason} ->
                ErrMeta = #{error_msg => ErrMsg, reason => Reason},
                throw(ErrMeta)
        end
    end;
read_file_fun(Path, ErrMsg, #{read_mode := ?JSON_MAP}) ->
    fun() ->
        case hocon:load(Path, #{format => richmap}) of
            {ok, RichMap} ->
                {ok, hocon_maps:ensure_plain(RichMap)};
            {error, Reason} ->
                ErrMeta = #{error_msg => ErrMsg, reason => Reason},
                throw(ErrMeta)
        end
    end.

%% Directorys
plugin_dir(NameVsn) ->
    filename:join([install_dir(), NameVsn]).

plugin_config_dir(NameVsn) ->
    filename:join([plugin_dir(NameVsn), "data", "configs"]).

%% Files
pkg_file_path(NameVsn) ->
    filename:join([install_dir(), bin([NameVsn, ".tar.gz"])]).

info_file_path(NameVsn) ->
    filename:join([plugin_dir(NameVsn), "release.json"]).

avsc_file_path(NameVsn) ->
    filename:join([plugin_dir(NameVsn), "config_schema.avsc"]).

avro_config_file(NameVsn) ->
    filename:join([plugin_config_dir(NameVsn), "config.avro"]).

i18n_file_path(NameVsn) ->
    filename:join([plugin_dir(NameVsn), "config_i18n.json"]).

readme_file(NameVsn) ->
    filename:join([plugin_dir(NameVsn), "README.md"]).

running_apps() ->
    lists:map(
        fun({N, _, V}) ->
            {N, V}
        end,
        application:which_applications(infinity)
    ).

bin_key(Map) when is_map(Map) ->
    maps:fold(fun(K, V, Acc) -> Acc#{bin(K) => V} end, #{}, Map);
bin_key(List = [#{} | _]) ->
    lists:map(fun(M) -> bin_key(M) end, List);
bin_key(Term) ->
    Term.

bin(A) when is_atom(A) -> atom_to_binary(A, utf8);
bin(L) when is_list(L) -> unicode:characters_to_binary(L, utf8);
bin(B) when is_binary(B) -> B.