Merge pull request #3895 from esl/graphql/import-users

Add importUsers to GraphQL

big_tests/tests/graphql_account_SUITE.erl

@@ -1,6 +1,7 @@
 -module(graphql_account_SUITE).
 
 -include_lib("eunit/include/eunit.hrl").
+-include_lib("common_test/include/ct.hrl").
 
 -compile([export_all, nowarn_export_all]).
 
@@ -26,8 +27,8 @@ all() ->
 
 groups() ->
     [{user_account, [parallel], user_account_tests()},
-     {admin_account_http, [], admin_account_tests()},
-     {admin_account_cli, [], admin_account_tests()},
+     {admin_account_http, [], admin_account_tests() ++ admin_account_http_tests()},
+     {admin_account_cli, [], admin_account_tests() ++ admin_account_cli_tests()},
      {domain_admin_account, [], domain_admin_tests()}].
 
 user_account_tests() ->
@@ -55,6 +56,12 @@ admin_account_tests() ->
      admin_change_user_password,
      admin_change_non_existing_user_password].
 
+admin_account_http_tests() ->
+    [admin_import_users_http].
+
+admin_account_cli_tests() ->
+    [admin_import_users_cli].
+
 domain_admin_tests() ->
     [admin_list_users,
      domain_admin_list_users_no_permission,
@@ -168,6 +175,11 @@ end_per_testcase(domain_admin_register_user = C, Config) ->
     {Username, Domain} = proplists:get_value(user, Config),
     rpc(mim(), mongoose_account_api, unregister_user, [Username, Domain]),
     escalus:end_per_testcase(C, Config);
+end_per_testcase(CaseName, Config)
+      when CaseName == admin_import_users_http; CaseName == admin_import_users_cli ->
+    Domain = domain_helper:domain(),
+    rpc(mim(), mongoose_account_api, unregister_user, [<<"john">>, Domain]),
+    escalus:end_per_testcase(CaseName, Config);
 end_per_testcase(CaseName, Config) ->
     escalus:end_per_testcase(CaseName, Config).
 
@@ -401,6 +413,57 @@ admin_change_non_existing_user_password(Config) ->
     Resp3 = ban_user(?EMPTY_NAME_JID, NewPassword, Config),
     get_coercion_err_msg(Resp3).
 
+admin_import_users_cli(Config) ->
+    escalus:fresh_story(Config, [{alice, 1}], fun(_Alice) ->
+        % Non-existing file
+        Resp = import_users(<<"nonexisting.csv">>, Config),
+        ?assertEqual(<<"File not found">>, get_err_msg(Resp)),
+        % Summary
+        Path = filename:join(?config(mim_data_dir, Config), "users.csv"),
+        Path2 = replace_hosts_in_file(Path),
+        Resp2 = import_users(list_to_binary(Path2), Config),
+        Domain = domain_helper:domain(),
+        ?assertEqual(#{<<"status">> => <<"Completed">>,
+                       <<"created">> => [<<"john@", Domain/binary>>],
+                       <<"emptyPassword">> => [<<"elise@", Domain/binary>>],
+                       <<"existing">> => [<<"alice@", Domain/binary>>],
+                       <<"invalidJID">> => [<<",", Domain/binary, ",password">>],
+                       <<"invalidRecord">> => [<<"elise,elise,", Domain/binary, ",esile">>],
+                       <<"notAllowed">> => null},
+                     get_ok_value([data, account, importUsers], Resp2))
+    end).
+
+admin_import_users_http(Config) ->
+    escalus:fresh_story(Config, [{alice, 1}], fun(_Alice) ->
+        % Summary
+        Path = filename:join(?config(mim_data_dir, Config), "users.csv"),
+        Path2 = replace_hosts_in_file(Path),
+        Resp2 = import_users(list_to_binary(Path2), Config),
+        ?assertEqual(#{<<"status">> => <<"ImportUsers scheduled">>,
+                       <<"created">> => null,
+                       <<"emptyPassword">> => null,
+                       <<"existing">> => null,
+                       <<"invalidJID">> => null,
+                       <<"invalidRecord">> => null,
+                       <<"notAllowed">> => null},
+                     get_ok_value([data, account, importUsers], Resp2)),
+        Domain = domain_helper:domain(),
+        mongoose_helper:wait_until(fun() ->
+                                       rpc(mim(), mongoose_account_api, check_account, [<<"john">>, Domain])
+                                   end,
+                                   {ok, io_lib:format("User ~s exists", [<<"john@", Domain/binary>>])},
+                                   #{time_left => timer:seconds(20),
+                                     sleep_time => 1000,
+                                     name => verify_account_created})
+    end).
+
+replace_hosts_in_file(Path) ->
+    {ok, Content} = file:read_file(Path),
+    Content2 = binary:replace(Content, <<"$host$">>, domain_helper:domain(), [global]),
+    Path2 = Path ++ ".tmp",
+    ok = file:write_file(Path2, Content2),
+    Path2.
+
 domain_admin_list_users_no_permission(Config) ->
     % An unknown domain
     Resp1 = list_users(<<"unknown-domain">>, Config),
@@ -551,3 +614,7 @@ ban_user(JID, Reason, Config) ->
 change_user_password(JID, NewPassword, Config) ->
     Vars = #{<<"user">> => JID, <<"newPassword">> => NewPassword},
     execute_command(<<"account">>, <<"changeUserPassword">>, Vars, Config).
+
+import_users(Filename, Config) ->
+    Vars = #{<<"filename">> => Filename},
+    execute_command(<<"account">>, <<"importUsers">>, Vars, Config).

big_tests/tests/graphql_account_SUITE_data/users.csv

@@ -0,0 +1,5 @@
+john,$host$,smith
+alice,$host$,smith
+elise,$host$,
+,$host$,password
+elise,elise,$host$,esile

priv/graphql/schemas/admin/account.gql

@@ -35,6 +35,8 @@ type AccountAdminMutation @protected{
   "Change the password of a user"
   changeUserPassword(user: JID!, newPassword: String!): UserPayload
     @protected(type: DOMAIN, args: ["user"])
+  "Import users from a CSV file"
+  importUsers(filename: NonEmptyString!): ImportPayload
 }
 
 "Modify user payload"
@@ -45,6 +47,24 @@ type UserPayload{
   message: String!
 }
 
+"Import users payload"
+type ImportPayload{
+  "Status"
+  status: String!
+  "Users created"
+  created: [JID!]
+  "Users that were already existing"
+  existing: [JID!]
+  "Users there were not allowed to be created"
+  notAllowed: [JID!]
+  "Users with invalid JIDs"
+  invalidJID: [String!]
+  "Users with empty passwords"
+  emptyPassword: [JID!]
+  "Invalid records"
+  invalidRecord: [String!]
+}
+
 "Check password correctness payload"
 type CheckPasswordPayload{
   "Status of the password correctness"

src/ejabberd_admin.erl

@@ -26,8 +26,6 @@
 -module(ejabberd_admin).
 -author('[email protected]').
 
--define(REGISTER_WORKERS_NUM, 10).
-
 -export([start/0, stop/0,
          %% Server
          %% Accounts
@@ -38,8 +36,6 @@
          delete_expired_messages/1, delete_old_messages/2,
          remove_from_cluster/1]).
 
--export([registrator_proc/1]).
-
 -ignore_xref([
     backup_mnesia/1, delete_expired_messages/1, delete_old_messages/2,
     dump_mnesia/1, dump_table/2,
@@ -104,9 +100,9 @@ commands() ->
                         desc = "Import users from CSV file",
                         module = ?MODULE, function = import_users,
                         args = [{file, string}],
-                        result = {users, {list, {res, {tuple,
-                                                       [{result, atom},
-                                                        {user, binary}]}}}}},
+                        result = {summary, {list, {res, {tuple,
+                                                         [{reason, binary},
+                                                          {users, {list, {user, binary}}}]}}}}},
      #ejabberd_commands{name = delete_expired_messages, tags = [purge],
                         desc = "Delete expired offline messages from database",
                         module = ?MODULE, function = delete_expired_messages,
@@ -245,90 +241,10 @@ unregister(User, Host) ->
 registered_users(Host) ->
     mongoose_account_api:list_users(Host).
 
--spec import_users(Filename :: string()) -> [{ok, jid:user()} |
-                                             {exists, jid:user()} |
-                                             {not_allowed, jid:user()} |
-                                             {invalid_jid, jid:user()} |
-                                             {null_password, jid:user()} |
-                                             {bad_csv, binary()}].
-import_users(File) ->
-    {ok, CsvStream} = erl_csv:decode_new_s(File),
-    Workers = spawn_link_workers(),
-    WorkersQueue = queue:from_list(Workers),
-    do_import(CsvStream, WorkersQueue).
-
--spec do_import(erl_csv:csv_stream(), Workers :: queue:queue()) ->
-    [{ok, jid:user()} |
-     {exists, jid:user()} |
-     {not_allowed, jid:user()} |
-     {invalid_jid, jid:user()} |
-     {null_password, jid:user()} |
-     {bad_csv, binary()}].
-do_import(stream_end, WQueue) ->
-    Workers = queue:to_list(WQueue),
-    lists:flatmap(fun get_results_from_registrator/1, Workers);
-do_import(Stream, WQueue) ->
-    {ok, Decoded, MoreStream} = erl_csv:decode_s(Stream),
-    WQueue1 = send_job_to_next_worker(Decoded, WQueue),
-    do_import(MoreStream, WQueue1).
-
--spec spawn_link_workers() -> [pid()].
-spawn_link_workers() ->
-    [ spawn_link(?MODULE, registrator_proc, [self()]) ||
-      _ <- lists:seq(1, ?REGISTER_WORKERS_NUM)].
-
--spec get_results_from_registrator(Worker :: pid()) ->
-    [{ok, jid:user()} |
-     {exists, jid:user()} |
-     {not_allowed, jid:user()} |
-     {invalid_jid, jid:user()} |
-     {null_password, jid:user()} |
-     {bad_csv, binary()}].
-get_results_from_registrator(Pid) ->
-    Pid ! get_result,
-    receive
-        {result, Result} -> Result
-    end.
-
-send_job_to_next_worker([], WQueue) ->
-    WQueue;
-send_job_to_next_worker([Record], WQueue) ->
-    {{value, Worker}, Q1} = queue:out(WQueue),
-    Worker ! {proccess, Record},
-    queue:in(Worker, Q1).
-
--spec registrator_proc(Manager :: pid()) -> ok.
-registrator_proc(Manager) ->
-    registrator_proc(Manager, []).
-
--spec registrator_proc(Manager :: pid(), any()) -> ok.
-registrator_proc(Manager, Result) ->
-    receive
-        {proccess, Data} ->
-            RegisterResult = do_register(Data),
-            registrator_proc(Manager, [RegisterResult | Result]);
-        get_result -> Manager ! {result, Result}
-    end,
-    ok.
-
--spec do_register([binary()]) -> {ok, jid:user()} |
-                                 {exists, jid:user()} |
-                                 {not_allowed, jid:user()} |
-                                 {invalid_jid, jid:user()} |
-                                 {null_password, jid:user()} |
-                                 {bad_csv, binary()}.
-do_register([User, Host, Password]) ->
-    case ejabberd_auth:try_register(jid:make(User, Host, <<>>), Password) of
-        {error, Reason} -> {Reason, User};
-        _ -> {ok, User}
-    end;
-
-do_register(List) ->
-    JoinBinary = fun(Elem, <<"">>) -> Elem;
-                    (Elem, Acc) -> <<Elem/binary, ",", Acc/binary>>
-                 end,
-    Info = lists:foldr(JoinBinary, <<"">>, List),
-    {bad_csv, Info}.
+-spec import_users(file:filename()) -> [{binary(), jid:user() | binary()}].
+import_users(Filename) ->
+    {ok, Result} = mongoose_import_users:run(Filename),
+    maps:to_list(Result).
 
 %%%
 %%% Purge DB

src/ejabberd_ctl.erl

@@ -370,7 +370,7 @@ format_error(Error) ->
 -spec format_result(In :: tuple() | atom() | integer() | string() | binary(),
                     {_, 'atom'|'integer'|'string'|'binary'}
                     ) -> string() | {string(), _}.
-format_result({Atom, Error}, _) when Atom =/= ok ->
+format_result({Atom, Error}, _) when is_atom(Atom), Atom =/= ok ->
     {io_lib:format("Error: ~ts", [format_error(Error)]), make_status(error)};
 format_result(Atom, {_Name, atom}) ->
     io_lib:format("~p", [Atom]);

src/graphql/admin/mongoose_graphql_account_admin_mutation.erl

@@ -2,21 +2,27 @@
 -behaviour(mongoose_graphql).
 
 -export([execute/4]).
+-export([await_execution/4]).
 
 -ignore_xref([execute/4]).
 
 -include("../mongoose_graphql_types.hrl").
 
 -import(mongoose_graphql_helper, [format_result/2, make_error/2]).
 
-execute(_Ctx, _Obj, <<"registerUser">>, Args) ->
+execute(_Ctx, account, <<"registerUser">>, Args) ->
     register_user(Args);
-execute(_Ctx, _Obj, <<"removeUser">>, Args) ->
+execute(_Ctx, account, <<"removeUser">>, Args) ->
     remove_user(Args);
-execute(_Ctx, _Obj, <<"banUser">>, Args) ->
+execute(_Ctx, account, <<"banUser">>, Args) ->
     ban_user(Args);
-execute(_Ctx, _Obj, <<"changeUserPassword">>, Args) ->
-    change_user_password(Args).
+execute(_Ctx, account, <<"changeUserPassword">>, Args) ->
+    change_user_password(Args);
+execute(#{method := cli}, account, <<"importUsers">>, Args) ->
+    import_users(Args);
+execute(#{method := http}, account, <<"importUsers">>, #{<<"filename">> := Filename}) ->
+    spawn(?MODULE, await_execution, [1000, mongoose_account_api, import_users, [Filename]]),
+    {ok, #{<<"status">> => <<"ImportUsers scheduled">>}}.
 
 % Internal
 
@@ -49,6 +55,13 @@ change_user_password(#{<<"user">> := JID, <<"newPassword">> := Password}) ->
     Result = mongoose_account_api:change_password(JID, Password),
     format_user_payload(Result, JID).
 
+-spec import_users(map()) -> {ok, map()} | {error, resolver_error()}.
+import_users(#{<<"filename">> := Filename}) ->
+    case mongoose_account_api:import_users(Filename) of
+        {ok, _} = Result -> Result;
+        Error -> make_error(Error, #{filename => Filename})
+    end.
+
 -spec format_user_payload({atom(), string()}, jid:jid()) -> {ok, map()} | {error, resolver_error()}.
 format_user_payload(InResult, JID) ->
     case InResult of
@@ -61,3 +74,7 @@ format_user_payload(InResult, JID) ->
 -spec make_user_payload(string(), jid:literal_jid()) -> map().
 make_user_payload(Msg, JID) ->
     #{<<"message">> => iolist_to_binary(Msg), <<"jid">> => JID}.
+
+await_execution(Timeout, Module, Fun, Args) ->
+    timer:sleep(Timeout),
+    apply(Module, Fun, Args).

src/mongoose_account_api.erl

@@ -16,7 +16,8 @@
          check_password/2,
          check_password/3,
          check_password_hash/3,
-         check_password_hash/4]).
+         check_password_hash/4,
+         import_users/1]).
 
 -type register_result() :: {ok | exists | invalid_jid | cannot_register, iolist()}.
 
@@ -184,6 +185,30 @@ check_password_hash(JID, PasswordHash, HashMethod) ->
             {incorrect, "Password hash is incorrect"}
     end.
 
+-spec import_users(file:filename()) -> {ok, #{binary() => [{ok, jid:jid() | binary()}]}}
+                                     | {file_not_found, binary()}.
+import_users(Filename) ->
+    case mongoose_import_users:run(Filename) of
+        {ok, Summary} ->
+            {ok, maps:fold(
+                fun(Reason, List, Map) ->
+                    List2 = [{ok, El} || El <- List],
+                    maps:put(from_reason(Reason), List2, Map)
+                end,
+                #{<<"status">> => <<"Completed">>},
+                Summary)};
+        {error, file_not_found} ->
+            {file_not_found, <<"File not found">>}
+    end.
+
+-spec from_reason(mongoose_import_users:reason()) -> binary().
+from_reason(ok) -> <<"created">>;
+from_reason(exists) -> <<"existing">>;
+from_reason(not_allowed) -> <<"notAllowed">>;
+from_reason(invalid_jid) -> <<"invalidJID">>;
+from_reason(null_password) -> <<"emptyPassword">>;
+from_reason(bad_csv) -> <<"invalidRecord">>.
+
 -spec ban_account(jid:user(), jid:server(), binary()) -> change_password_result().
 ban_account(User, Host, ReasonText) ->
     JID = jid:make(User, Host, <<>>),