./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 46 register_prepared_queries(Opts),
54 46 ok.
55
56 -spec stop(host_type()) -> ok.
57 stop(_HostType) ->
58 46 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 50 [];
87 false ->
88 42 [{mam_muc_archive_message, HostType, fun ?MODULE:archive_message/3, #{}, 50}]
89 92 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 46 prepare_insert(insert_mam_muc_message, 1),
103 46 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 46 prepare_remove_domain(Opts),
109
110 46 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 46 {LimitSQL, LimitMSSQL} = rdbms_queries:get_db_specific_limits_binaries(1),
114 46 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 46 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 46 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 44 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 44 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 1931 [#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 367 [#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 3469 #{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 1295 mam_decoder:decode_muc_row(Row, Env).
194
195 -spec index_hint_sql(env_vars()) -> string().
196 25 index_hint_sql(_) -> "".
197
198 16 columns_sql(lookup) -> "id, nick_name, message";
199 9 columns_sql(count) -> "COUNT(*)".
200
201 266 column_to_id(id) -> "i";
202 945 column_to_id(room_id) -> "u";
203 10 column_to_id(nick_name) -> "n";
204 15 column_to_id(search_body) -> "s".
205
206 column_names(Mappings) ->
207 88 [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 1840 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 493 Filter = [{equal, room_id, ArcID}],
225 493 Env = env_vars(HostType, ArcJID),
226 493 Result = lookup_query(count, Env, Filter, unordered, all),
227 493 {ok, mongoose_rdbms:selected_to_integer(Result)}.
228
229 extend_params_with_sender_id(HostType, Params = #{remote_jid := SenderJID}) ->
230 1843 BareSenderJID = jid:to_bare(SenderJID),
231 1843 SenderID = mod_mam_pm:archive_id_int(HostType, BareSenderJID),
232 1843 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 1124 try
240 1124 Params = extend_params_with_sender_id(HostType, Params0),
241 1124 Env = env_vars(HostType, ArcJID),
242 1124 do_archive_message(HostType, Params, Env),
243 1121 retract_message(HostType, Params, Env),
244 1121 {ok, ok}
245 catch error:Reason:StackTrace ->
246 3 mongoose_instrument:execute(mod_mam_muc_dropped, #{host_type => HostType}, #{count => 1}),
247 3 ?LOG_ERROR(#{what => archive_message_failed,
248 host_type => HostType, mam_params => Params0,
249 
:-(
 reason => Reason, stacktrace => StackTrace}),
250 3 erlang:raise(error, Reason, StackTrace)
251 end.
252
253 do_archive_message(HostType, Params, Env) ->
254 1124 Row = mam_encoder:encode_message(Params, Env, db_mappings()),
255 1124 {updated, 1} = mongoose_rdbms:execute_successfully(HostType, insert_mam_muc_message, Row).
256
257 %% Retraction logic
258 %% Called after inserting a new message
259 -spec retract_message(mongooseim:host_type(), mod_mam:archive_message_params()) -> ok.
260 retract_message(HostType, #{local_jid := ArcJID} = Params) ->
261 719 Env = env_vars(HostType, ArcJID),
262 719 retract_message(HostType, Params, Env).
263
264 -spec retract_message(mongooseim:host_type(), mod_mam:archive_message_params(), env_vars()) -> ok.
265 retract_message(HostType, #{archive_id := ArcID, sender_id := SenderID,
266 packet := Packet} = Params, Env) ->
267 1840 case get_retract_id(Packet, Env) of
268 1825 none -> ok;
269 RetractionId ->
270 15 Info = get_retraction_info(HostType, ArcID, SenderID, RetractionId, Env),
271 15 make_tombstone(HostType, ArcID, RetractionId, Info, Params, Env)
272 end.
273
274 get_retraction_info(HostType, ArcID, SenderID, RetractionId, Env) ->
275 15 {selected, Rows} =
276 execute_select_messages_to_retract(HostType, ArcID, SenderID, RetractionId),
277 15 mam_decoder:decode_retraction_info(Env, Rows, RetractionId).
278
279 make_tombstone(_HostType, ArcID, RetractionId, skip, _Params, _Env) ->
280 5 ?LOG_INFO(#{what => make_tombstone_failed,
281 text => <<"Message to retract was not found">>,
282 5 user_id => ArcID, retraction_context => RetractionId});
283 make_tombstone(HostType, ArcID, _RetractionId,
284 RetractionInfo = #{message_id := MessID}, Params,
285 #{archive_jid := ArcJID} = Env) ->
286 10 RetractionInfo1 = mongoose_hooks:mam_muc_retraction(HostType, RetractionInfo, Params),
287 10 Tombstone = mod_mam_utils:tombstone(RetractionInfo1, ArcJID),
288 10 TombstoneData = mam_encoder:encode_packet(Tombstone, Env),
289 10 execute_make_tombstone(HostType, TombstoneData, ArcID, MessID).
290
291 execute_select_messages_to_retract(HostType, ArcID, SenderID, {origin_id, OriginID}) ->
292 10 mongoose_rdbms:execute_successfully(HostType, mam_muc_select_messages_to_retract_on_origin_id,
293 [ArcID, SenderID, OriginID]);
294 execute_select_messages_to_retract(HostType, ArcID, SenderID, {stanza_id, BinStanzaId}) ->
295 5 StanzaId = mod_mam_utils:external_binary_to_mess_id(BinStanzaId),
296 5 mongoose_rdbms:execute_successfully(HostType, mam_muc_select_messages_to_retract_on_stanza_id,
297 [ArcID, SenderID, StanzaId]).
298
299 execute_make_tombstone(HostType, TombstoneData, ArcID, MessID) ->
300 10 mongoose_rdbms:execute_successfully(HostType, mam_muc_make_tombstone,
301 [TombstoneData, ArcID, MessID]).
302
303 %% Insert logic
304 -spec prepare_message(mongooseim:host_type(), mod_mam:archive_message_params()) -> list().
305 prepare_message(HostType, Params = #{local_jid := ArcJID}) ->
306 719 Env = env_vars(HostType, ArcJID),
307 719 mam_encoder:encode_message(Params, Env, db_mappings()).
308
309 -spec prepare_insert(Name :: atom(), NumRows :: pos_integer()) -> ok.
310 prepare_insert(Name, NumRows) ->
311 88 Table = mam_muc_message,
312 88 Fields = column_names(db_mappings()),
313 88 {Query, Fields2} = rdbms_queries:create_bulk_insert_query(Table, Fields, NumRows),
314 88 mongoose_rdbms:prepare(Name, Table, Fields2, Query),
315 88 ok.
316
317 %% Removal logic
318 -spec remove_archive(Acc, Params, Extra) -> {ok, Acc} when
319 Acc :: mongoose_acc:t(),
320 Params :: #{archive_id := mod_mam:archive_id() | undefined, room := jid:jid()},
321 Extra :: gen_hook:extra().
322 remove_archive(Acc, #{archive_id := ArcID}, #{host_type := HostType}) ->
323 438 mongoose_rdbms:execute_successfully(HostType, mam_muc_archive_remove, [ArcID]),
324 438 {ok, Acc}.
325
326 -spec remove_domain(Acc, Params, Extra) -> {ok | stop, Acc} when
327 Acc :: mongoose_domain_api:remove_domain_acc(),
328 Params :: map(),
329 Extra :: gen_hook:extra().
330 remove_domain(Acc, #{domain := Domain}, #{host_type := HostType}) ->
331 5 F = fun() ->
332 5 case gen_mod:get_module_opt(HostType, ?MODULE, delete_domain_limit) of
333 3 infinity -> remove_domain_all(HostType, Domain);
334 2 Limit -> remove_domain_batch(HostType, Domain, Limit)
335 end,
336 5 Acc
337 end,
338 5 mongoose_domain_api:remove_domain_wrapper(Acc, F, ?MODULE).
339
340 -spec remove_domain_all(host_type(), jid:lserver()) -> any().
341 remove_domain_all(HostType, Domain) ->
342 3 SubHosts = get_subhosts(HostType, Domain),
343 3 {atomic, _} = mongoose_rdbms:sql_transaction(HostType, fun() ->
344 3 [remove_domain_trans(HostType, SubHost) || SubHost <- SubHosts]
345 end).
346
347 -spec remove_domain_batch(host_type(), jid:lserver(), non_neg_integer()) -> any().
348 remove_domain_batch(HostType, Domain, Limit) ->
349 2 SubHosts = get_subhosts(HostType, Domain),
350 2 DeleteQueries = [mam_muc_incr_remove_domain, mam_muc_incr_remove_domain_users],
351 2 DelSubHost = [ mod_mam_utils:incremental_delete_domain(HostType, SubHost, Limit, DeleteQueries, 0)
352 2 || SubHost <- SubHosts],
353 2 TotalDeleted = lists:sum(DelSubHost),
354 2 ?LOG_INFO(#{what => mam_muc_domain_removal_completed, total_records_deleted => TotalDeleted,
355 2 domain => Domain, host_type => HostType}).
356
357 remove_domain_trans(HostType, MucHost) ->
358 3 mongoose_rdbms:execute_successfully(HostType, mam_muc_remove_domain, [MucHost]),
359 3 mongoose_rdbms:execute_successfully(HostType, mam_muc_remove_domain_users, [MucHost]).
360
361 get_subhosts(HostType, Domain) ->
362 5 lists:usort(
363 10 lists:flatmap(fun(Module) -> get_subhosts_for_module(HostType, Domain, Module) end,
364 [mod_muc, mod_muc_light])).
365
366 get_subhosts_for_module(HostType, Domain, Module) ->
367 10 case gen_mod:get_module_opts(HostType, Module) of
368 #{host := HostPattern} ->
369 5 [mongoose_subdomain_utils:get_fqdn(HostPattern, Domain)];
370 #{} ->
371 5 []
372 end.
373
374 %% GDPR logic
375 extract_gdpr_messages(HostType, SenderID) ->
376 47 mongoose_rdbms:execute_successfully(HostType, mam_muc_extract_gdpr_messages, [SenderID]).
377
378 %% Lookup logic
379 -spec lookup_messages(Acc, Params, Extra) -> {ok, Acc} when
380 Acc :: {ok, mod_mam:lookup_result()},
381 Params :: mam_iq:lookup_params(),
382 Extra :: gen_hook:extra().
383 lookup_messages(_Result, #{owner_jid := ArcJID} = Params, #{host_type := HostType}) ->
384 367 Env = env_vars(HostType, ArcJID),
385 367 ExdParams = mam_encoder:extend_lookup_params(Params, Env),
386 367 Filter = mam_filter:produce_filter(ExdParams, lookup_fields()),
387 367 {ok, mam_lookup:lookup(Env, Filter, ExdParams)}.
388
389 lookup_query(QueryType, Env, Filters, Order, OffsetLimit) ->
390 945 mam_lookup_sql:lookup_query(QueryType, Env, Filters, Order, OffsetLimit).
Line Hits Source