./ct_report/coverage/mod_mam_muc_rdbms_arch.COVER.html

1 %%%-------------------------------------------------------------------
2 %%% @author Uvarov Michael <arcusfelis@gmail.com>
3 %%% @copyright (C) 2013, Uvarov Michael
4 %%% @doc RDBMS backend for MUC Message Archive Management.
5 %%% @end
6 %%%-------------------------------------------------------------------
7 -module(mod_mam_muc_rdbms_arch).
8
9 %% ----------------------------------------------------------------------
10 %% Exports
11
12 %% gen_mod handlers
13 -export([start/2, stop/1, hooks/1, supported_features/0]).
14
15 %% MAM hook handlers
16 -behaviour(ejabberd_gen_mam_archive).
17 -behaviour(gen_mod).
18 -behaviour(mongoose_module_metrics).
19
20 -callback encode(term()) -> binary().
21 -callback decode(binary()) -> term().
22
23 -export([archive_size/3,
24 archive_message/3,
25 lookup_messages/3,
26 remove_archive/3,
27 remove_domain/3,
28 get_mam_muc_gdpr_data/3]).
29
30 %% Called from mod_mam_muc_rdbms_async_pool_writer
31 -export([prepare_message/2, retract_message/2, prepare_insert/2]).
32 -export([extend_params_with_sender_id/2]).
33
34 %% ----------------------------------------------------------------------
35 %% Imports
36
37 -include("mongoose.hrl").
38 -include("jlib.hrl").
39 -include("mongoose_mam.hrl").
40
41 %% ----------------------------------------------------------------------
42 %% Types
43
44 -type env_vars() :: mod_mam_rdbms_arch:env_vars().
45 -type host_type() :: mongooseim:host_type().
46
47 %% ----------------------------------------------------------------------
48 %% gen_mod callbacks
49 %% Starting and stopping functions for users' archives
50
51 -spec start(host_type(), gen_mod:module_opts()) -> ok.
52 start(_HostType, Opts) ->
53 56 register_prepared_queries(Opts),
54 56 ok.
55
56 -spec stop(host_type()) -> ok.
57 stop(_HostType) ->
58 56 ok.
59
60 -spec supported_features() -> [atom()].
61 supported_features() ->
62 
:-(
 [dynamic_domains].
63
64 -spec get_mam_muc_gdpr_data(Acc, Params, Extra) -> {ok, Acc} when
65 Acc :: ejabberd_gen_mam_archive:mam_muc_gdpr_data(),
66 Params :: #{jid := jid:jid()},
67 Extra :: gen_hook:extra().
68 get_mam_muc_gdpr_data(Acc, #{jid := #jid{luser = LUser, lserver = LServer}}, #{host_type := HostType}) ->
69 47 case mod_mam_pm:archive_id(LServer, LUser) of
70 undefined ->
71 
:-(
 {ok, Acc};
72 SenderID ->
73 %% We don't know the real room JID here, use FakeEnv
74 47 FakeEnv = env_vars(HostType, jid:make(<<>>, <<>>, <<>>)),
75 47 {selected, Rows} = extract_gdpr_messages(HostType, SenderID),
76 47 {ok, [mam_decoder:decode_muc_gdpr_row(Row, FakeEnv) || Row <- Rows] ++ Acc}
77 end.
78
79 %% ----------------------------------------------------------------------
80 %% Add hooks for mod_mam_pm
81
82 -spec hooks(mongooseim:host_type()) -> gen_hook:hook_list().
83 hooks(HostType) ->
84 case gen_mod:get_module_opt(HostType, ?MODULE, no_writer) of
85 true ->
86 54 [];
87 false ->
88 58 [{mam_muc_archive_message, HostType, fun ?MODULE:archive_message/3, #{}, 50}]
89 112 end ++
90 [
91 {remove_domain, HostType, fun ?MODULE:remove_domain/3, #{}, 50},
92 {mam_muc_archive_size, HostType, fun ?MODULE:archive_size/3, #{}, 50},
93 {mam_muc_lookup_messages, HostType, fun ?MODULE:lookup_messages/3, #{}, 50},
94 {mam_muc_remove_archive, HostType, fun ?MODULE:remove_archive/3, #{}, 50},
95 {get_mam_muc_gdpr_data, HostType, fun ?MODULE:get_mam_muc_gdpr_data/3, #{}, 50}
96 ].
97
98 %% ----------------------------------------------------------------------
99 %% SQL queries
100
101 register_prepared_queries(Opts) ->
102 56 prepare_insert(insert_mam_muc_message, 1),
103 56 mongoose_rdbms:prepare(mam_muc_archive_remove, mam_muc_message, [room_id],
104 <<"DELETE FROM mam_muc_message "
105 "WHERE room_id = ?">>),
106
107 %% Domain Removal
108 56 prepare_remove_domain(Opts),
109
110 56 mongoose_rdbms:prepare(mam_muc_make_tombstone, mam_muc_message, [message, room_id, id],
111 <<"UPDATE mam_muc_message SET message = ?, search_body = '' "
112 "WHERE room_id = ? AND id = ?">>),
113 56 {LimitSQL, LimitMSSQL} = rdbms_queries:get_db_specific_limits_binaries(1),
114 56 mongoose_rdbms:prepare(mam_muc_select_messages_to_retract_on_origin_id, mam_muc_message,
115 [room_id, sender_id, origin_id],
116 <<"SELECT ", LimitMSSQL/binary,
117 " id, message FROM mam_muc_message"
118 " WHERE room_id = ? AND sender_id = ? "
119 " AND origin_id = ?"
120 " ORDER BY id DESC ", LimitSQL/binary>>),
121 56 mongoose_rdbms:prepare(mam_muc_select_messages_to_retract_on_stanza_id, mam_muc_message,
122 [room_id, sender_id, id],
123 <<"SELECT ", LimitMSSQL/binary,
124 " origin_id, message FROM mam_muc_message"
125 " WHERE room_id = ? AND sender_id = ? "
126 " AND id = ?"
127 " ORDER BY id DESC ", LimitSQL/binary>>),
128 56 mongoose_rdbms:prepare(mam_muc_extract_gdpr_messages, mam_muc_message, [sender_id],
129 <<"SELECT id, message FROM mam_muc_message "
130 " WHERE sender_id = ? ORDER BY id">>).
131
132 prepare_remove_domain(#{delete_domain_limit := infinity}) ->
133 54 mongoose_rdbms:prepare(mam_muc_remove_domain, mam_muc_message, ['mam_server_user.server'],
134 <<"DELETE FROM mam_muc_message "
135 "WHERE room_id IN (SELECT id FROM mam_server_user where server = ?)">>),
136 54 mongoose_rdbms:prepare(mam_muc_remove_domain_users, mam_server_user, [server],
137 <<"DELETE FROM mam_server_user WHERE server = ? ">>);
138 prepare_remove_domain(#{delete_domain_limit := Limit}) ->
139 2 LimitSQL = case mongoose_rdbms:db_type() of
140 
:-(
 mssql -> throw(delete_domain_limit_not_supported_for_mssql);
141 2 _ -> {MaybeLimitSQL, _} = rdbms_queries:get_db_specific_limits_binaries(Limit),
142 2 MaybeLimitSQL
143 end,
144 2 IdTable = <<"(SELECT * FROM ",
145 "(SELECT msg.room_id, msg.id FROM mam_muc_message msg",
146 " INNER JOIN mam_server_user msu ON msu.id=msg.room_id",
147 " WHERE msu.server = ? ", LimitSQL/binary, ") AS T)">>,
148 2 mongoose_rdbms:prepare(mam_muc_incr_remove_domain, mam_muc_message, ['mam_server_user.server'],
149 <<"DELETE FROM mam_muc_message WHERE (room_id, id) IN ", IdTable/binary>>),
150 2 ServerTable = <<"(SELECT * FROM",
151 "(SELECT id FROM mam_server_user WHERE server = ? ", LimitSQL/binary, ") as t)">>,
152 2 mongoose_rdbms:prepare(mam_muc_incr_remove_domain_users, mam_server_user, [server],
153 <<"DELETE FROM mam_server_user WHERE id IN ", ServerTable/binary>>).
154
155 %% ----------------------------------------------------------------------
156 %% Declarative logic
157
158 db_mappings() ->
159 2627 [#db_mapping{column = id, param = message_id, format = int},
160 #db_mapping{column = room_id, param = archive_id, format = int},
161 #db_mapping{column = sender_id, param = sender_id, format = int},
162 #db_mapping{column = nick_name, param = source_jid, format = jid_resource},
163 #db_mapping{column = origin_id, param = origin_id, format = maybe_string},
164 #db_mapping{column = message, param = packet, format = xml},
165 #db_mapping{column = search_body, param = packet, format = search}].
166
167 lookup_fields() ->
168 498 [#lookup_field{op = equal, column = room_id, param = archive_id, required = true},
169 #lookup_field{op = ge, column = id, param = start_id},
170 #lookup_field{op = le, column = id, param = end_id},
171 #lookup_field{op = equal, column = nick_name, param = remote_resource},
172 #lookup_field{op = like, column = search_body, param = norm_search_text, value_maker = search_words},
173 #lookup_field{op = equal, column = id, param = message_id}].
174
175 -spec env_vars(host_type(), jid:jid()) -> env_vars().
176 env_vars(HostType, ArcJID) ->
177 %% Please, minimize the usage of the host field.
178 %% It's only for passing into RDBMS.
179 4364 #{host_type => HostType,
180 archive_jid => ArcJID,
181 table => mam_muc_message,
182 index_hint_fn => fun index_hint_sql/1,
183 columns_sql_fn => fun columns_sql/1,
184 column_to_id_fn => fun column_to_id/1,
185 lookup_fn => fun lookup_query/5,
186 decode_row_fn => fun row_to_uniform_format/2,
187 has_message_retraction => mod_mam_utils:has_message_retraction(mod_mam_muc, HostType),
188 has_full_text_search => mod_mam_utils:has_full_text_search(mod_mam_muc, HostType),
189 db_jid_codec => mod_mam_utils:db_jid_codec(HostType, ?MODULE),
190 db_message_codec => mod_mam_utils:db_message_codec(HostType, ?MODULE)}.
191
192 row_to_uniform_format(Row, Env) ->
193 1801 mam_decoder:decode_muc_row(Row, Env).
194
195 -spec index_hint_sql(env_vars()) -> string().
196 26 index_hint_sql(_) -> "".
197
198 16 columns_sql(lookup) -> "id, nick_name, message";
199 10 columns_sql(count) -> "COUNT(*)".
200
201 362 column_to_id(id) -> "i";
202 1195 column_to_id(room_id) -> "u";
203 14 column_to_id(nick_name) -> "n";
204 21 column_to_id(search_body) -> "s".
205
206 column_names(Mappings) ->
207 102 [Column || #db_mapping{column = Column} <- Mappings].
208
209 %% ----------------------------------------------------------------------
210 %% Options
211
212 -spec get_retract_id(exml:element(), env_vars()) -> none | mod_mam_utils:retraction_id().
213 get_retract_id(Packet, #{has_message_retraction := Enabled}) ->
214 2525 mod_mam_utils:get_retract_id(Enabled, Packet).
215
216 %% ----------------------------------------------------------------------
217 %% Internal functions and callbacks
218
219 -spec archive_size(Acc, Params, Extra) -> {ok, Acc} when
220 Acc :: integer(),
221 Params :: #{archive_id := mod_mam:archive_id() | undefined, room := jid:jid()},
222 Extra :: gen_hook:extra().
223 archive_size(Size, #{archive_id := ArcID, room := ArcJID}, #{host_type := HostType}) when is_integer(Size) ->
224 578 Filter = [{equal, room_id, ArcID}],
225 578 Env = env_vars(HostType, ArcJID),
226 578 Result = lookup_query(count, Env, Filter, unordered, all),
227 578 {ok, mongoose_rdbms:selected_to_integer(Result)}.
228
229 extend_params_with_sender_id(HostType, Params = #{remote_jid := SenderJID}) ->
230 2525 BareSenderJID = jid:to_bare(SenderJID),
231 2525 SenderID = mod_mam_pm:archive_id_int(HostType, BareSenderJID),
232 2525 Params#{sender_id => SenderID}.
233
234 -spec archive_message(Acc, Params, Extra) -> {ok, Acc} when
235 Acc :: ok,
236 Params :: mod_mam:archive_message_params(),
237 Extra :: gen_hook:extra().
238 archive_message(_Result, #{local_jid := ArcJID} = Params0, #{host_type := HostType}) ->
239 1809 try
240 1809 Params = extend_params_with_sender_id(HostType, Params0),
241 1809 Env = env_vars(HostType, ArcJID),
242 1809 do_archive_message(HostType, Params, Env),
243 1809 retract_message(HostType, Params, Env),
244 1809 {ok, ok}
245 catch error:Reason:StackTrace ->
246 
:-(
 ?LOG_ERROR(#{what => archive_message_failed,
247 host_type => HostType, mam_params => Params0,
248 
:-(
 reason => Reason, stacktrace => StackTrace}),
249 
:-(
 erlang:raise(error, Reason, StackTrace)
250 end.
251
252 do_archive_message(HostType, Params, Env) ->
253 1809 Row = mam_encoder:encode_message(Params, Env, db_mappings()),
254 1809 {updated, 1} = mongoose_rdbms:execute_successfully(HostType, insert_mam_muc_message, Row).
255
256 %% Retraction logic
257 %% Called after inserting a new message
258 -spec retract_message(mongooseim:host_type(), mod_mam:archive_message_params()) -> ok.
259 retract_message(HostType, #{local_jid := ArcJID} = Params) ->
260 716 Env = env_vars(HostType, ArcJID),
261 716 retract_message(HostType, Params, Env).
262
263 -spec retract_message(mongooseim:host_type(), mod_mam:archive_message_params(), env_vars()) -> ok.
264 retract_message(HostType, #{archive_id := ArcID, sender_id := SenderID,
265 packet := Packet} = Params, Env) ->
266 2525 case get_retract_id(Packet, Env) of
267 2504 none -> ok;
268 RetractionId ->
269 21 Info = get_retraction_info(HostType, ArcID, SenderID, RetractionId, Env),
270 21 make_tombstone(HostType, ArcID, RetractionId, Info, Params, Env)
271 end.
272
273 get_retraction_info(HostType, ArcID, SenderID, RetractionId, Env) ->
274 21 {selected, Rows} =
275 execute_select_messages_to_retract(HostType, ArcID, SenderID, RetractionId),
276 21 mam_decoder:decode_retraction_info(Env, Rows, RetractionId).
277
278 make_tombstone(_HostType, ArcID, RetractionId, skip, _Params, _Env) ->
279 7 ?LOG_INFO(#{what => make_tombstone_failed,
280 text => <<"Message to retract was not found">>,
281 7 user_id => ArcID, retraction_context => RetractionId});
282 make_tombstone(HostType, ArcID, _RetractionId,
283 RetractionInfo = #{message_id := MessID}, Params,
284 #{archive_jid := ArcJID} = Env) ->
285 14 RetractionInfo1 = mongoose_hooks:mam_muc_retraction(HostType, RetractionInfo, Params),
286 14 Tombstone = mod_mam_utils:tombstone(RetractionInfo1, ArcJID),
287 14 TombstoneData = mam_encoder:encode_packet(Tombstone, Env),
288 14 execute_make_tombstone(HostType, TombstoneData, ArcID, MessID).
289
290 execute_select_messages_to_retract(HostType, ArcID, SenderID, {origin_id, OriginID}) ->
291 14 mongoose_rdbms:execute_successfully(HostType, mam_muc_select_messages_to_retract_on_origin_id,
292 [ArcID, SenderID, OriginID]);
293 execute_select_messages_to_retract(HostType, ArcID, SenderID, {stanza_id, BinStanzaId}) ->
294 7 StanzaId = mod_mam_utils:external_binary_to_mess_id(BinStanzaId),
295 7 mongoose_rdbms:execute_successfully(HostType, mam_muc_select_messages_to_retract_on_stanza_id,
296 [ArcID, SenderID, StanzaId]).
297
298 execute_make_tombstone(HostType, TombstoneData, ArcID, MessID) ->
299 14 mongoose_rdbms:execute_successfully(HostType, mam_muc_make_tombstone,
300 [TombstoneData, ArcID, MessID]).
301
302 %% Insert logic
303 -spec prepare_message(mongooseim:host_type(), mod_mam:archive_message_params()) -> list().
304 prepare_message(HostType, Params = #{local_jid := ArcJID}) ->
305 716 Env = env_vars(HostType, ArcJID),
306 716 mam_encoder:encode_message(Params, Env, db_mappings()).
307
308 -spec prepare_insert(Name :: atom(), NumRows :: pos_integer()) -> ok.
309 prepare_insert(Name, NumRows) ->
310 102 Table = mam_muc_message,
311 102 Fields = column_names(db_mappings()),
312 102 {Query, Fields2} = rdbms_queries:create_bulk_insert_query(Table, Fields, NumRows),
313 102 mongoose_rdbms:prepare(Name, Table, Fields2, Query),
314 102 ok.
315
316 %% Removal logic
317 -spec remove_archive(Acc, Params, Extra) -> {ok, Acc} when
318 Acc :: mongoose_acc:t(),
319 Params :: #{archive_id := mod_mam:archive_id() | undefined, room := jid:jid()},
320 Extra :: gen_hook:extra().
321 remove_archive(Acc, #{archive_id := ArcID}, #{host_type := HostType}) ->
322 548 mongoose_rdbms:execute_successfully(HostType, mam_muc_archive_remove, [ArcID]),
323 548 {ok, Acc}.
324
325 -spec remove_domain(Acc, Params, Extra) -> {ok | stop, Acc} when
326 Acc :: mongoose_domain_api:remove_domain_acc(),
327 Params :: map(),
328 Extra :: gen_hook:extra().
329 remove_domain(Acc, #{domain := Domain}, #{host_type := HostType}) ->
330 5 F = fun() ->
331 5 case gen_mod:get_module_opt(HostType, ?MODULE, delete_domain_limit) of
332 3 infinity -> remove_domain_all(HostType, Domain);
333 2 Limit -> remove_domain_batch(HostType, Domain, Limit)
334 end,
335 5 Acc
336 end,
337 5 mongoose_domain_api:remove_domain_wrapper(Acc, F, ?MODULE).
338
339 -spec remove_domain_all(host_type(), jid:lserver()) -> any().
340 remove_domain_all(HostType, Domain) ->
341 3 SubHosts = get_subhosts(HostType, Domain),
342 3 {atomic, _} = mongoose_rdbms:sql_transaction(HostType, fun() ->
343 3 [remove_domain_trans(HostType, SubHost) || SubHost <- SubHosts]
344 end).
345
346 -spec remove_domain_batch(host_type(), jid:lserver(), non_neg_integer()) -> any().
347 remove_domain_batch(HostType, Domain, Limit) ->
348 2 SubHosts = get_subhosts(HostType, Domain),
349 2 DeleteQueries = [mam_muc_incr_remove_domain, mam_muc_incr_remove_domain_users],
350 2 DelSubHost = [ mod_mam_utils:incremental_delete_domain(HostType, SubHost, Limit, DeleteQueries, 0)
351 2 || SubHost <- SubHosts],
352 2 TotalDeleted = lists:sum(DelSubHost),
353 2 ?LOG_INFO(#{what => mam_muc_domain_removal_completed, total_records_deleted => TotalDeleted,
354 2 domain => Domain, host_type => HostType}).
355
356 remove_domain_trans(HostType, MucHost) ->
357 3 mongoose_rdbms:execute_successfully(HostType, mam_muc_remove_domain, [MucHost]),
358 3 mongoose_rdbms:execute_successfully(HostType, mam_muc_remove_domain_users, [MucHost]).
359
360 get_subhosts(HostType, Domain) ->
361 5 lists:usort(
362 10 lists:flatmap(fun(Module) -> get_subhosts_for_module(HostType, Domain, Module) end,
363 [mod_muc, mod_muc_light])).
364
365 get_subhosts_for_module(HostType, Domain, Module) ->
366 10 case gen_mod:get_module_opts(HostType, Module) of
367 #{host := HostPattern} ->
368 5 [mongoose_subdomain_utils:get_fqdn(HostPattern, Domain)];
369 #{} ->
370 5 []
371 end.
372
373 %% GDPR logic
374 extract_gdpr_messages(HostType, SenderID) ->
375 47 mongoose_rdbms:execute_successfully(HostType, mam_muc_extract_gdpr_messages, [SenderID]).
376
377 %% Lookup logic
378 -spec lookup_messages(Acc, Params, Extra) -> {ok, Acc} when
379 Acc :: {ok, mod_mam:lookup_result()},
380 Params :: mam_iq:lookup_params(),
381 Extra :: gen_hook:extra().
382 lookup_messages(_Result, #{owner_jid := ArcJID} = Params, #{host_type := HostType}) ->
383 498 Env = env_vars(HostType, ArcJID),
384 498 ExdParams = mam_encoder:extend_lookup_params(Params, Env),
385 498 Filter = mam_filter:produce_filter(ExdParams, lookup_fields()),
386 498 {ok, mam_lookup:lookup(Env, Filter, ExdParams)}.
387
388 lookup_query(QueryType, Env, Filters, Order, OffsetLimit) ->
389 1195 mam_lookup_sql:lookup_query(QueryType, Env, Filters, Order, OffsetLimit).
Line Hits Source