./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 1913 [#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 449 [#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
174 -spec env_vars(host_type(), jid:jid()) -> env_vars().
175 env_vars(HostType, ArcJID) ->
176 %% Please, minimize the usage of the host field.
177 %% It's only for passing into RDBMS.
178 3355 #{host_type => HostType,
179 archive_jid => ArcJID,
180 table => mam_muc_message,
181 index_hint_fn => fun index_hint_sql/1,
182 columns_sql_fn => fun columns_sql/1,
183 column_to_id_fn => fun column_to_id/1,
184 lookup_fn => fun lookup_query/5,
185 decode_row_fn => fun row_to_uniform_format/2,
186 has_message_retraction => mod_mam_utils:has_message_retraction(mod_mam_muc, HostType),
187 has_full_text_search => mod_mam_utils:has_full_text_search(mod_mam_muc, HostType),
188 db_jid_codec => mod_mam_utils:db_jid_codec(HostType, ?MODULE),
189 db_message_codec => mod_mam_utils:db_message_codec(HostType, ?MODULE)}.
190
191 row_to_uniform_format(Row, Env) ->
192 1759 mam_decoder:decode_muc_row(Row, Env).
193
194 -spec index_hint_sql(env_vars()) -> string().
195 24 index_hint_sql(_) -> "".
196
197 14 columns_sql(lookup) -> "id, nick_name, message";
198 10 columns_sql(count) -> "COUNT(*)".
199
200 299 column_to_id(id) -> "i";
201 1104 column_to_id(room_id) -> "u";
202 14 column_to_id(nick_name) -> "n";
203 21 column_to_id(search_body) -> "s".
204
205 column_names(Mappings) ->
206 102 [Column || #db_mapping{column = Column} <- Mappings].
207
208 %% ----------------------------------------------------------------------
209 %% Options
210
211 -spec get_retract_id(exml:element(), env_vars()) -> none | mod_mam_utils:retraction_id().
212 get_retract_id(Packet, #{has_message_retraction := Enabled}) ->
213 1811 mod_mam_utils:get_retract_id(Enabled, Packet).
214
215 %% ----------------------------------------------------------------------
216 %% Internal functions and callbacks
217
218 -spec archive_size(Acc, Params, Extra) -> {ok, Acc} when
219 Acc :: integer(),
220 Params :: #{archive_id := mod_mam:archive_id() | undefined, room := jid:jid()},
221 Extra :: gen_hook:extra().
222 archive_size(Size, #{archive_id := ArcID, room := ArcJID}, #{host_type := HostType}) when is_integer(Size) ->
223 536 Filter = [{equal, room_id, ArcID}],
224 536 Env = env_vars(HostType, ArcJID),
225 536 Result = lookup_query(count, Env, Filter, unordered, all),
226 536 {ok, mongoose_rdbms:selected_to_integer(Result)}.
227
228 extend_params_with_sender_id(HostType, Params = #{remote_jid := SenderJID}) ->
229 1811 BareSenderJID = jid:to_bare(SenderJID),
230 1811 SenderID = mod_mam_pm:archive_id_int(HostType, BareSenderJID),
231 1811 Params#{sender_id => SenderID}.
232
233 -spec archive_message(Acc, Params, Extra) -> {ok, Acc} when
234 Acc :: ok,
235 Params :: mod_mam:archive_message_params(),
236 Extra :: gen_hook:extra().
237 archive_message(_Result, #{local_jid := ArcJID} = Params0, #{host_type := HostType}) ->
238 1299 try
239 1299 Params = extend_params_with_sender_id(HostType, Params0),
240 1299 Env = env_vars(HostType, ArcJID),
241 1299 do_archive_message(HostType, Params, Env),
242 1299 retract_message(HostType, Params, Env),
243 1299 {ok, ok}
244 catch error:Reason:StackTrace ->
245 
:-(
 ?LOG_ERROR(#{what => archive_message_failed,
246 host_type => HostType, mam_params => Params0,
247 
:-(
 reason => Reason, stacktrace => StackTrace}),
248 
:-(
 erlang:raise(error, Reason, StackTrace)
249 end.
250
251 do_archive_message(HostType, Params, Env) ->
252 1299 Row = mam_encoder:encode_message(Params, Env, db_mappings()),
253 1299 {updated, 1} = mongoose_rdbms:execute_successfully(HostType, insert_mam_muc_message, Row).
254
255 %% Retraction logic
256 %% Called after inserting a new message
257 -spec retract_message(mongooseim:host_type(), mod_mam:archive_message_params()) -> ok.
258 retract_message(HostType, #{local_jid := ArcJID} = Params) ->
259 512 Env = env_vars(HostType, ArcJID),
260 512 retract_message(HostType, Params, Env).
261
262 -spec retract_message(mongooseim:host_type(), mod_mam:archive_message_params(), env_vars()) -> ok.
263 retract_message(HostType, #{archive_id := ArcID, sender_id := SenderID,
264 packet := Packet} = Params, Env) ->
265 1811 case get_retract_id(Packet, Env) of
266 1790 none -> ok;
267 RetractionId ->
268 21 Info = get_retraction_info(HostType, ArcID, SenderID, RetractionId, Env),
269 21 make_tombstone(HostType, ArcID, RetractionId, Info, Params, Env)
270 end.
271
272 get_retraction_info(HostType, ArcID, SenderID, RetractionId, Env) ->
273 21 {selected, Rows} =
274 execute_select_messages_to_retract(HostType, ArcID, SenderID, RetractionId),
275 21 mam_decoder:decode_retraction_info(Env, Rows, RetractionId).
276
277 make_tombstone(_HostType, ArcID, RetractionId, skip, _Params, _Env) ->
278 7 ?LOG_INFO(#{what => make_tombstone_failed,
279 text => <<"Message to retract was not found">>,
280 7 user_id => ArcID, retraction_context => RetractionId});
281 make_tombstone(HostType, ArcID, _RetractionId,
282 RetractionInfo = #{message_id := MessID}, Params,
283 #{archive_jid := ArcJID} = Env) ->
284 14 RetractionInfo1 = mongoose_hooks:mam_muc_retraction(HostType, RetractionInfo, Params),
285 14 Tombstone = mod_mam_utils:tombstone(RetractionInfo1, ArcJID),
286 14 TombstoneData = mam_encoder:encode_packet(Tombstone, Env),
287 14 execute_make_tombstone(HostType, TombstoneData, ArcID, MessID).
288
289 execute_select_messages_to_retract(HostType, ArcID, SenderID, {origin_id, OriginID}) ->
290 14 mongoose_rdbms:execute_successfully(HostType, mam_muc_select_messages_to_retract_on_origin_id,
291 [ArcID, SenderID, OriginID]);
292 execute_select_messages_to_retract(HostType, ArcID, SenderID, {stanza_id, BinStanzaId}) ->
293 7 StanzaId = mod_mam_utils:external_binary_to_mess_id(BinStanzaId),
294 7 mongoose_rdbms:execute_successfully(HostType, mam_muc_select_messages_to_retract_on_stanza_id,
295 [ArcID, SenderID, StanzaId]).
296
297 execute_make_tombstone(HostType, TombstoneData, ArcID, MessID) ->
298 14 mongoose_rdbms:execute_successfully(HostType, mam_muc_make_tombstone,
299 [TombstoneData, ArcID, MessID]).
300
301 %% Insert logic
302 -spec prepare_message(mongooseim:host_type(), mod_mam:archive_message_params()) -> list().
303 prepare_message(HostType, Params = #{local_jid := ArcJID}) ->
304 512 Env = env_vars(HostType, ArcJID),
305 512 mam_encoder:encode_message(Params, Env, db_mappings()).
306
307 -spec prepare_insert(Name :: atom(), NumRows :: pos_integer()) -> ok.
308 prepare_insert(Name, NumRows) ->
309 102 Table = mam_muc_message,
310 102 Fields = column_names(db_mappings()),
311 102 {Query, Fields2} = rdbms_queries:create_bulk_insert_query(Table, Fields, NumRows),
312 102 mongoose_rdbms:prepare(Name, Table, Fields2, Query),
313 102 ok.
314
315 %% Removal logic
316 -spec remove_archive(Acc, Params, Extra) -> {ok, Acc} when
317 Acc :: mongoose_acc:t(),
318 Params :: #{archive_id := mod_mam:archive_id() | undefined, room := jid:jid()},
319 Extra :: gen_hook:extra().
320 remove_archive(Acc, #{archive_id := ArcID}, #{host_type := HostType}) ->
321 506 mongoose_rdbms:execute_successfully(HostType, mam_muc_archive_remove, [ArcID]),
322 506 {ok, Acc}.
323
324 -spec remove_domain(Acc, Params, Extra) -> {ok | stop, Acc} when
325 Acc :: mongoose_domain_api:remove_domain_acc(),
326 Params :: map(),
327 Extra :: gen_hook:extra().
328 remove_domain(Acc, #{domain := Domain}, #{host_type := HostType}) ->
329 5 F = fun() ->
330 5 case gen_mod:get_module_opt(HostType, ?MODULE, delete_domain_limit) of
331 3 infinity -> remove_domain_all(HostType, Domain);
332 2 Limit -> remove_domain_batch(HostType, Domain, Limit)
333 end,
334 5 Acc
335 end,
336 5 mongoose_domain_api:remove_domain_wrapper(Acc, F, ?MODULE).
337
338 -spec remove_domain_all(host_type(), jid:lserver()) -> any().
339 remove_domain_all(HostType, Domain) ->
340 3 SubHosts = get_subhosts(HostType, Domain),
341 3 {atomic, _} = mongoose_rdbms:sql_transaction(HostType, fun() ->
342 3 [remove_domain_trans(HostType, SubHost) || SubHost <- SubHosts]
343 end).
344
345 -spec remove_domain_batch(host_type(), jid:lserver(), non_neg_integer()) -> any().
346 remove_domain_batch(HostType, Domain, Limit) ->
347 2 SubHosts = get_subhosts(HostType, Domain),
348 2 DeleteQueries = [mam_muc_incr_remove_domain, mam_muc_incr_remove_domain_users],
349 2 DelSubHost = [ mod_mam_utils:incremental_delete_domain(HostType, SubHost, Limit, DeleteQueries, 0)
350 2 || SubHost <- SubHosts],
351 2 TotalDeleted = lists:sum(DelSubHost),
352 2 ?LOG_INFO(#{what => mam_muc_domain_removal_completed, total_records_deleted => TotalDeleted,
353 2 domain => Domain, host_type => HostType}).
354
355 remove_domain_trans(HostType, MucHost) ->
356 3 mongoose_rdbms:execute_successfully(HostType, mam_muc_remove_domain, [MucHost]),
357 3 mongoose_rdbms:execute_successfully(HostType, mam_muc_remove_domain_users, [MucHost]).
358
359 get_subhosts(HostType, Domain) ->
360 5 lists:usort(
361 10 lists:flatmap(fun(Module) -> get_subhosts_for_module(HostType, Domain, Module) end,
362 [mod_muc, mod_muc_light])).
363
364 get_subhosts_for_module(HostType, Domain, Module) ->
365 10 case gen_mod:get_module_opts(HostType, Module) of
366 #{host := HostPattern} ->
367 5 [mongoose_subdomain_utils:get_fqdn(HostPattern, Domain)];
368 #{} ->
369 5 []
370 end.
371
372 %% GDPR logic
373 extract_gdpr_messages(HostType, SenderID) ->
374 47 mongoose_rdbms:execute_successfully(HostType, mam_muc_extract_gdpr_messages, [SenderID]).
375
376 %% Lookup logic
377 -spec lookup_messages(Acc, Params, Extra) -> {ok, Acc} when
378 Acc :: {ok, mod_mam:lookup_result()},
379 Params :: mam_iq:lookup_params(),
380 Extra :: gen_hook:extra().
381 lookup_messages(_Result, #{owner_jid := ArcJID} = Params, #{host_type := HostType}) ->
382 449 Env = env_vars(HostType, ArcJID),
383 449 ExdParams = mam_encoder:extend_lookup_params(Params, Env),
384 449 Filter = mam_filter:produce_filter(ExdParams, lookup_fields()),
385 449 {ok, mam_lookup:lookup(Env, Filter, ExdParams)}.
386
387 lookup_query(QueryType, Env, Filters, Order, OffsetLimit) ->
388 1104 mam_lookup_sql:lookup_query(QueryType, Env, Filters, Order, OffsetLimit).
Line Hits Source