DB schema rework for GDPR MAM MUC retrieval

big_tests/tests/mam_SUITE.erl

@@ -689,7 +689,7 @@ init_modules(rdbms, muc_all, Config) ->
     init_module(host(), mod_mam_muc, [{host, muc_domain(Config)}]),
     Config;
 init_modules(rdbms_simple, muc_all, Config) ->
-    init_module(host(), mod_mam_muc_rdbms_arch, [muc, simple]),
+    init_module(host(), mod_mam_muc_rdbms_arch, [muc, rdbms_simple_opts()]),
     init_module(host(), mod_mam_rdbms_prefs, [muc]),
     init_module(host(), mod_mam_rdbms_user, [muc]),
     init_module(host(), mod_mam_muc, [{host, muc_domain(Config)}]),
@@ -750,7 +750,7 @@ init_modules(rdbms, C, Config) ->
     Config;
 init_modules(rdbms_simple, C, Config) ->
     init_module(host(), mod_mam, addin_mam_options(C, Config)),
-    init_module(host(), mod_mam_rdbms_arch, [pm, simple]),
+    init_module(host(), mod_mam_rdbms_arch, [pm, rdbms_simple_opts()]),
     init_module(host(), mod_mam_rdbms_prefs, [pm]),
     init_module(host(), mod_mam_rdbms_user, [pm]),
     Config;
@@ -815,6 +815,9 @@ init_modules(rdbms_mnesia_cache, C, Config) ->
     init_module(host(), mod_mam_cache_user, [pm]),
     Config.
 
+rdbms_simple_opts() ->
+    [{db_jid_format, mam_jid_rfc}, {db_message_format, mam_message_xml}].
+
 init_modules_for_muc_light(BackendType, Config) ->
     dynamic_modules:start(host(), mod_muc_light, [{host, binary_to_list(muc_light_host())}]),
     Config1 = init_modules(BackendType, muc_all, [{muc_domain, "muclight.@HOST@"} | Config]),
@@ -1493,7 +1496,7 @@ muc_querying_for_all_messages_with_jid(Config) ->
     P = ?config(props, Config),
     F = fun(Alice, Bob) ->
         Room = ?config(room, Config),
-        BWithJID = room_address(Room, nick(Bob)),
+        BWithJID = room_address(Room, nick(bob)),
 
         MucMsgs = ?config(pre_generated_muc_msgs, Config),
         WithJID = [1 || {_, _, {JID, _, _}, _, _} <- MucMsgs, JID == BWithJID],

big_tests/tests/muc_SUITE.erl

@@ -379,7 +379,7 @@ init_per_group(G, Config) when G =:= http_auth_no_server;
 init_per_group(hibernation, Config) ->
     case mam_helper:backend() of
         rdbms ->
-    dynamic_modules:start(domain(), mod_mam_muc_rdbms_arch, [muc, simple]),
+    dynamic_modules:start(domain(), mod_mam_muc_rdbms_arch, [muc]),
     dynamic_modules:start(domain(), mod_mam_rdbms_prefs, [muc]),
     dynamic_modules:start(domain(), mod_mam_rdbms_user, [muc]),
             dynamic_modules:start(domain(), mod_mam_muc, [{host, "muc.@HOST@"}]);

big_tests/tests/muc_helper.erl

@@ -70,7 +70,62 @@ muc_host() ->
 start_room(Config, User, Room, Nick, Opts) ->
     From = generate_rpc_jid(User),
     create_instant_room(<<"localhost">>, Room, From, Nick, Opts),
-    [{nick, Nick}, {room, Room} | Config].
+    RoomJID = room_address(Room),
+    [{nick, Nick}, {room, Room}, {room_jid, RoomJID}, {muc_host, muc_host()} | Config].
+
+start_fresh_room(Config, User, Nick, Opts) ->
+    Room = fresh_room_name(),
+    start_room(Config, User, Room, Nick, Opts).
+
+enter_room(Config, UsersAndNicks) ->
+    [Room, RoomJid] = [proplists:get_value(Key, Config) || Key <- [room, room_jid]],
+    lists:foldl(fun({User, Nick}, Acc) ->
+                        escalus:send(User, stanza_muc_enter_room(Room, Nick)),
+                        wait_for_presence(RoomJid, User, length(Acc)),
+                        foreach_recipient([User | Acc],
+                                          fun(Stanza) ->
+                                                  validate_presence(Stanza,
+                                                                    RoomJid,
+                                                                    Nick)
+                                          end),
+                        Subject = escalus:wait_for_stanza(User),
+                        validate_subject_message(Subject, RoomJid),
+                        [User | Acc]
+                end,
+                [],UsersAndNicks).
+
+wait_for_presence(_, _, 0) ->
+    ok;
+wait_for_presence(RoomJid, User, N) ->
+    Stanza = escalus:wait_for_stanza(User),
+    validate_presence(Stanza, RoomJid),
+    wait_for_presence(RoomJid, User, N - 1).
+
+validate_presence(Stanza, RoomJid) ->
+    escalus:assert(is_presence, [], Stanza),
+    [RoomJid, _] = binary:split(exml_query:attr(Stanza, <<"from">>), <<"/">>).
+
+validate_presence(Stanza, RoomJid, Nick) ->
+    [RoomJid, Nick] = validate_presence(Stanza, RoomJid).
+
+validate_subject_message(Stanza, RoomJid) ->
+    RoomJid = exml_query:attr(Stanza, <<"from">>),
+    #xmlel{} = exml_query:subelement(Stanza, <<"subject">>).
+
+verify_message_received(Config, Users, Nick, TextBody) ->
+    RoomJid = proplists:get_value(room_jid, Config),
+    foreach_recipient(Users, muc_msg_verify(RoomJid, Nick, TextBody)).
+
+muc_msg_verify(RoomBareJID, NickName, MsgText) ->
+    fun(Msg) ->
+        escalus:assert(is_groupchat_message, [MsgText], Msg),
+        [RoomBareJID, NickName] = binary:split(exml_query:attr(Msg, <<"from">>), <<"/">>)
+    end.
+
+send_to_room(RoomCfg, User, TextBody) ->
+    RoomJid = proplists:get_value(room_jid, RoomCfg),
+    Stanza = escalus_stanza:groupchat_to(RoomJid, TextBody),
+    escalus:send(User, Stanza).
 
 generate_rpc_jid({_,User}) ->
     {username, Username} = lists:keyfind(username, 1, User),

big_tests/tests/rest_helper.erl

@@ -274,7 +274,7 @@ to_list(V) when is_list(V) ->
     V.
 
 maybe_enable_mam(rdbms, Host, Config) ->
-    init_module(Host, mod_mam_rdbms_arch, [muc, pm, simple]),
+    init_module(Host, mod_mam_rdbms_arch, [muc, pm]),
     init_module(Host, mod_mam_rdbms_prefs, [muc, pm]),
     init_module(Host, mod_mam_rdbms_user, [muc, pm]),
     init_module(Host, mod_mam, []),

priv/cassandra.cql

@@ -26,6 +26,10 @@ CREATE TABLE mam_message_offset(
 CREATE TABLE mam_muc_message(
   id bigint,
   room_jid varchar,
+  // One of:
+  // - Sender JID (if with_nick is empty)
+  // - empty
+  from_jid varchar,
   nick_name varchar,
   // One of:
   // - nick_name
@@ -53,4 +57,6 @@ CREATE TABLE mam_config(
 
 CREATE INDEX ON mongooseim.mam_message (user_jid);
 CREATE INDEX ON mongooseim.mam_muc_message (room_jid);
+CREATE INDEX ON mongooseim.mam_muc_message (from_jid);
+
 

priv/elasticsearch/muc.json

@@ -8,6 +8,9 @@
 				"room": {
 					"type": "keyword"
 				},
+				"from_jid" : {
+					"type": "keyword"
+				},
 				"source_jid": {
 					"type": "keyword"
 				},

rel/files/mongooseim.cfg

@@ -664,7 +664,8 @@
 {services,
     [
         {service_admin_extra, [{submods, [node, accounts, sessions, vcard, gdpr,
-                                          roster, last, private, stanza, stats]}]}
+                                          roster, last, private, stanza, stats]}]},
+        {service_cache, []}
     ]
 }.
 

src/admin_extra/service_admin_extra.erl

@@ -29,7 +29,7 @@
 
 -behaviour(mongoose_service).
 
--export([start/1, stop/0]).
+-export([start/1, stop/0, deps/0]).
 
 -define(SUBMODS, [node, accounts, sessions, vcard, roster, last,
                   private, stanza, stats, gdpr
@@ -40,6 +40,8 @@
 %%% gen_mod
 %%%
 
+deps() -> [service_cache].
+
 start(Opts) ->
     Submods = gen_mod:get_opt(submods, Opts, ?SUBMODS),
     lists:foreach(fun(Submod) ->

src/mam/mod_mam_meta.erl

@@ -102,25 +102,24 @@ handle_nested_opts(Key, RootOpts, Default, Deps) ->
 -spec parse_opts(Type :: pm | muc, Opts :: proplists:proplist(), deps()) -> deps().
 parse_opts(Type, Opts, Deps) ->
     CoreMod = mam_type_to_core_mod(Type),
-
-    CoreModOpts =
-        lists:filtermap(
-          fun(Key) ->
-                  case proplists:lookup(Key, Opts) of
-                      none -> false;
-                      Opt -> {true, Opt}
-                  end
-          end, valid_core_mod_opts(CoreMod)),
-
+    CoreModOpts = filter_opts(Opts, valid_core_mod_opts(CoreMod)),
     WithCoreDeps = add_dep(CoreMod, CoreModOpts, Deps),
     Backend = proplists:get_value(backend, Opts, rdbms),
-
     parse_backend_opts(Backend, Type, Opts, WithCoreDeps).
 
 -spec mam_type_to_core_mod(atom()) -> module().
 mam_type_to_core_mod(pm) -> mod_mam;
 mam_type_to_core_mod(muc) -> mod_mam_muc.
 
+filter_opts(Opts, ValidOpts) ->
+    lists:filtermap(
+        fun(Key) ->
+            case proplists:lookup(Key, Opts) of
+                none -> false;
+                Opt -> {true, Opt}
+            end
+        end, ValidOpts).
+
 -spec valid_core_mod_opts(module()) -> [atom()].
 valid_core_mod_opts(mod_mam) ->
     [
@@ -152,7 +151,8 @@ parse_backend_opts(cassandra, Type, Opts, Deps0) ->
             muc -> mod_mam_muc_cassandra_arch
         end,
 
-    Deps = add_dep(ModArch, Deps0),
+    Opts1 = filter_opts(Opts, [db_message_format, pool_name, simple]),
+    Deps = add_dep(ModArch, Opts1, Deps0),
 
     case proplists:get_value(user_prefs_store, Opts, false) of
         cassandra -> add_dep(mod_mam_cassandra_prefs, [Type], Deps);
@@ -161,7 +161,8 @@ parse_backend_opts(cassandra, Type, Opts, Deps0) ->
     end;
 
 parse_backend_opts(riak, Type, Opts, Deps0) ->
-    Deps = add_dep(mod_mam_riak_timed_arch_yz, [Type], Deps0),
+    Opts1 = filter_opts(Opts, [db_message_format]),
+    Deps = add_dep(mod_mam_riak_timed_arch_yz, [Type | Opts1], Deps0),
 
     case proplists:get_value(user_prefs_store, Opts, false) of
         mnesia -> add_dep(mod_mam_mnesia_prefs, [Type], Deps);
@@ -181,25 +182,17 @@ parse_backend_opts(rdbms, Type, Opts0, Deps0) ->
     Deps = add_dep(mod_mam_rdbms_user, [Type], Deps1),
 
     lists:foldl(
-      pa:bind(fun parse_backend_opt/5, Type, ModRDBMSArch, ModAsyncWriter),
+      pa:bind(fun parse_rdbms_opt/5, Type, ModRDBMSArch, ModAsyncWriter),
       Deps, Opts);
 
 parse_backend_opts(elasticsearch, Type, Opts, Deps0) ->
-    ExtraOpts =
-        case proplists:get_value(elasticsearch_index_name, Opts) of
-            IndexName when is_binary(IndexName) ->
-                [{index_name, IndexName}];
-            _ ->
-                []
-        end,
-
     ModArch =
         case Type of
             pm -> mod_mam_elasticsearch_arch;
             muc -> mod_mam_muc_elasticsearch_arch
         end,
 
-    Deps = add_dep(ModArch, ExtraOpts, Deps0),
+    Deps = add_dep(ModArch, Deps0),
 
     case proplists:get_value(user_prefs_store, Opts, false) of
         mnesia -> add_dep(mod_mam_mnesia_prefs, [Type], Deps);
@@ -219,7 +212,7 @@ add_dep(Dep, Deps) ->
 -spec add_dep(Dep :: module(), Args :: proplists:proplist(), deps()) -> deps().
 add_dep(Dep, Args, Deps) ->
     PrevArgs = maps:get(Dep, Deps, []),
-    NewArgs = Args ++ PrevArgs,
+    NewArgs = lists:usort(Args ++ PrevArgs),
     maps:put(Dep, NewArgs, Deps).
 
 
@@ -236,9 +229,9 @@ add_default_rdbms_opts(Opts) ->
       [{cache_users, true}, {async_writer, true}]).
 
 
--spec parse_backend_opt(Type :: pm | muc, module(), module(),
+-spec parse_rdbms_opt(Type :: pm | muc, module(), module(),
                         Option :: {module(), term()}, deps()) -> deps().
-parse_backend_opt(Type, ModRDBMSArch, ModAsyncWriter, Option, Deps) ->
+parse_rdbms_opt(Type, ModRDBMSArch, ModAsyncWriter, Option, Deps) ->
     case Option of
         {cache_users, true} ->
             add_dep(mod_mam_cache_user, [Type], Deps);

src/mam/mod_mam_muc.erl

@@ -242,7 +242,7 @@ archive_room_packet(Packet, FromNick, FromJID=#jid{}, RoomJID=#jid{}, Role, Affi
             MessID = generate_message_id(),
             Packet1 = replace_x_user_element(FromJID, Role, Affiliation, Packet),
             Result = archive_message(Host, MessID, ArcID,
-                                     RoomJID, SrcJID, SrcJID, incoming, Packet1),
+                                     RoomJID, FromJID, SrcJID, incoming, Packet1),
             %% Packet2 goes to archive, Packet to other users
             case Result of
                 ok ->