./ct_report/coverage/mod_mam_cassandra_arch.COVER.html

1 %%%-------------------------------------------------------------------
2 %%% @author Uvarov Michael <arcusfelis@gmail.com>
3 %%% @copyright (C) 2013, Uvarov Michael
4 %%% @doc RDBMS backend for Message Archive Management.
5 %%% @end
6 %%%-------------------------------------------------------------------
7 -module(mod_mam_cassandra_arch).
8 -behaviour(mongoose_cassandra).
9 -behaviour(ejabberd_gen_mam_archive).
10 -behaviour(gen_mod).
11 -behaviour(mongoose_module_metrics).
12
13 %% ----------------------------------------------------------------------
14 %% Exports
15
16 %% gen_mod handlers
17 -export([start/2, stop/1, hooks/1]).
18
19 %% MAM hook handlers
20 -export([archive_size/3,
21 archive_message/3,
22 lookup_messages/3,
23 remove_archive/3]).
24
25 %% mongoose_cassandra callbacks
26 -export([prepared_queries/0]).
27
28 %gdpr
29 -export([get_mam_pm_gdpr_data/3]).
30
31 %% ----------------------------------------------------------------------
32 %% Imports
33
34 %% Other
35 -import(mod_mam_utils,
36 [maybe_min/2,
37 maybe_max/2,
38 bare_jid/1,
39 full_jid/1
40 ]).
41
42 -include("mongoose.hrl").
43 -include("jlib.hrl").
44 -include("mongoose_rsm.hrl").
45
46 -callback encode(binary()) -> binary().
47 -callback decode(binary()) -> binary().
48
49 -record(mam_ca_filter, {
50 user_jid,
51 with_jid,
52 remote_jid,
53 start_id,
54 end_id
55 }).
56
57 -record(mam_message, {
58 id :: non_neg_integer(),
59 user_jid :: binary(),
60 remote_jid :: binary() | undefined,
61 from_jid :: binary() | undefined,
62 with_jid = <<>> :: binary(),
63 message :: binary() | undefined
64 }).
65
66 %% ----------------------------------------------------------------------
67 %% Types
68
69 -type filter() :: #mam_ca_filter{}.
70 -type message_id() :: non_neg_integer().
71 -type host_type() :: mongooseim:host_type().
72
73 %% ----------------------------------------------------------------------
74 %% gen_mod callbacks
75 %% Starting and stopping functions for users' archives
76
77 -spec start(host_type(), gen_mod:module_opts()) -> ok.
78 start(_HostType, _Opts) ->
79
:-(
ok.
80
81 -spec stop(host_type()) -> ok.
82 stop(_HostType) ->
83
:-(
ok.
84
85 %% ----------------------------------------------------------------------
86 %% Add hooks for mod_mam_pm
87
88 -spec hooks(mongooseim:host_type()) -> gen_hook:hook_list().
89 hooks(HostType) ->
90
:-(
[{mam_archive_message, HostType, fun ?MODULE:archive_message/3, #{}, 50},
91 {mam_archive_size, HostType, fun ?MODULE:archive_size/3, #{}, 50},
92 {mam_lookup_messages, HostType, fun ?MODULE:lookup_messages/3, #{}, 50},
93 {mam_remove_archive, HostType, fun ?MODULE:remove_archive/3, #{}, 50},
94 {get_mam_pm_gdpr_data, HostType, fun ?MODULE:get_mam_pm_gdpr_data/3, #{}, 50}].
95
96 %% ----------------------------------------------------------------------
97 %% mongoose_cassandra_worker callbacks
98
99 prepared_queries() ->
100 [
101 {insert_offset_hint_query, insert_offset_hint_query_cql()},
102 {prev_offset_query, prev_offset_query_cql()},
103 {insert_query, insert_query_cql()},
104 {fetch_user_messages_query, fetch_user_messages_cql()},
105 {select_for_removal_query, select_for_removal_query_cql()},
106 {remove_archive_query, remove_archive_query_cql()},
107 {remove_archive_offsets_query, remove_archive_offsets_query_cql()}
108 ]
109
:-(
++ extract_messages_queries()
110 ++ extract_messages_r_queries()
111 ++ calc_count_queries()
112 ++ list_message_ids_queries().
113
114 %% ----------------------------------------------------------------------
115 %% Internal functions and callbacks
116
117 -spec archive_size(Acc, Params, Extra) -> {ok, Acc} when
118 Acc :: integer(),
119 Params :: #{archive_id := mod_mam:archive_id() | undefined, owner := jid:jid()},
120 Extra :: gen_hook:extra().
121 archive_size(Size, #{owner := UserJID}, #{host_type := HostType}) when is_integer(Size) ->
122
:-(
Borders = Start = End = WithJID = undefined,
123
:-(
Filter = prepare_filter(UserJID, Borders, Start, End, WithJID, undefined),
124
:-(
{ok, calc_count(pool_name(HostType), UserJID, HostType, Filter)}.
125
126
127 %% ----------------------------------------------------------------------
128 %% INSERT MESSAGE
129
130 insert_query_cql() ->
131
:-(
"INSERT INTO mam_message "
132 "(id, user_jid, from_jid, remote_jid, with_jid, message) "
133 "VALUES (?, ?, ?, ?, ?, ?)".
134
135 -spec archive_message(Acc, Params, Extra) -> {ok, Acc} when
136 Acc :: ok,
137 Params :: mod_mam:archive_message_params(),
138 Extra :: gen_hook:extra().
139 archive_message(_Result, Params, #{host_type := HostType}) ->
140
:-(
try
141
:-(
{ok, archive_message2(Params, HostType)}
142 catch _Type:Reason:StackTrace ->
143
:-(
?LOG_ERROR(#{what => archive_message_failed,
144 host_type => HostType, mam_params => Params,
145
:-(
reason => Reason, stacktrace => StackTrace}),
146
:-(
mongoose_instrument:execute(mod_mam_pm_dropped, #{host_type => HostType}, #{count => 1}),
147
:-(
{ok, {error, Reason}}
148 end.
149
150 archive_message2(#{message_id := MessID,
151 local_jid := LocJID,
152 remote_jid := RemJID,
153 source_jid := SrcJID,
154 packet := Packet}, HostType) ->
155
:-(
BLocJID = bare_jid(LocJID),
156
:-(
BRemBareJID = bare_jid(RemJID),
157
:-(
BRemFullJID = full_jid(RemJID),
158
:-(
BSrcJID = full_jid(SrcJID),
159
:-(
BPacket = packet_to_stored_binary(HostType, Packet),
160
:-(
Message = #mam_message{
161 id = MessID,
162 user_jid = BLocJID,
163 from_jid = BSrcJID,
164 remote_jid = BRemFullJID,
165 message = BPacket
166 },
167
:-(
WithJIDs = lists:usort([<<>>, BRemFullJID, BRemBareJID]),
168
:-(
Messages = [Message#mam_message{with_jid = BWithJID} || BWithJID <- WithJIDs],
169
:-(
ok = write_messages(HostType, LocJID, Messages).
170
171 write_messages(HostType, UserJID, Messages) ->
172
:-(
MultiParams = [message_to_params(M) || M <- Messages],
173
:-(
mongoose_cassandra:cql_write_async(pool_name(HostType), UserJID,
174 ?MODULE, insert_query, MultiParams).
175
176 message_to_params(#mam_message{
177 id = MessID,
178 user_jid = BLocJID,
179 from_jid = BSrcJID,
180 remote_jid = BRemJID,
181 with_jid = BWithJID,
182 message = BPacket
183 }) ->
184
:-(
#{id => MessID, user_jid => BLocJID, from_jid => BSrcJID,
185 remote_jid => BRemJID, with_jid => BWithJID, message => BPacket}.
186
187 %% ----------------------------------------------------------------------
188 %% REMOVE ARCHIVE
189
190 remove_archive_query_cql() ->
191
:-(
"DELETE FROM mam_message WHERE user_jid = ? AND with_jid = ?".
192
193 remove_archive_offsets_query_cql() ->
194
:-(
"DELETE FROM mam_message WHERE user_jid = ? AND with_jid = ?".
195
196 select_for_removal_query_cql() ->
197
:-(
"SELECT DISTINCT user_jid, with_jid FROM mam_message WHERE user_jid = ?".
198
199 -spec remove_archive(Acc, Params, Extra) -> {ok, Acc} when
200 Acc :: term(),
201 Params :: #{archive_id := mod_mam:archive_id() | undefined, owner := jid:jid()},
202 Extra :: gen_hook:extra().
203 remove_archive(Acc, #{owner := UserJID}, #{host_type := HostType}) ->
204
:-(
remove_archive(HostType, UserJID),
205
:-(
{ok, Acc}.
206
207 remove_archive(HostType, UserJID) ->
208
:-(
PoolName = pool_name(HostType),
209
:-(
BUserJID = bare_jid(UserJID),
210
:-(
Params = #{user_jid => BUserJID},
211 %% Wait until deleted
212
:-(
DeleteFun =
213 fun(Rows, _AccIn) ->
214
:-(
mongoose_cassandra:cql_write(PoolName, UserJID, ?MODULE,
215 remove_archive_query, Rows),
216
:-(
mongoose_cassandra:cql_write(PoolName, UserJID, ?MODULE,
217 remove_archive_offsets_query, Rows)
218 end,
219
220
:-(
mongoose_cassandra:cql_foldl(PoolName, UserJID, ?MODULE,
221 select_for_removal_query, Params, DeleteFun, []).
222 %% ----------------------------------------------------------------------
223 %% SELECT MESSAGES
224
225 -spec lookup_messages(Acc, Params, Extra) -> {ok, Acc} when
226 Acc :: {ok, mod_mam:lookup_result()} | {error, term()},
227 Params :: mam_iq:lookup_params(),
228 Extra :: gen_hook:extra().
229 lookup_messages({error, _Reason} = Result, _Params, _Extra) ->
230
:-(
{ok, Result};
231 lookup_messages(_Result, #{search_text := <<_/binary>>}, _Extra) ->
232
:-(
{ok, {error, 'not-supported'}};
233 lookup_messages(_Result,
234 #{owner_jid := UserJID, rsm := RSM, borders := Borders,
235 start_ts := Start, end_ts := End, with_jid := WithJID,
236 search_text := undefined, page_size := PageSize,
237 is_simple := IsSimple, message_id := MsgID},
238 #{host_type := HostType}) ->
239
:-(
try
240
:-(
{ok, lookup_messages2(pool_name(HostType), HostType,
241 UserJID, RSM, Borders,
242 Start, End, WithJID,
243 PageSize, MsgID, IsSimple)}
244 catch _Type:Reason:S ->
245
:-(
{ok, {error, {Reason, S}}}
246 end.
247
248 lookup_messages2(PoolName, HostType,
249 UserJID = #jid{}, RSM, Borders,
250 Start, End, WithJID,
251 PageSize, MsgID, _IsSimple = true) ->
252 %% Simple query without calculating offset and total count
253
:-(
Filter = prepare_filter(UserJID, Borders, Start, End, WithJID, MsgID),
254
:-(
lookup_messages_simple(PoolName, HostType, UserJID, RSM, PageSize, Filter);
255 lookup_messages2(PoolName, HostType,
256 UserJID = #jid{}, RSM, Borders,
257 Start, End, WithJID,
258 PageSize, MsgID, _IsSimple) ->
259 %% Query with offset calculation
260 %% We cannot just use RDBMS code because "LIMIT X, Y" is not supported by cassandra
261 %% Not all queries are optimal. You would like to disable something for production
262 %% once you know how you will call bd
263
:-(
Strategy = rsm_to_strategy(RSM),
264
:-(
Filter = prepare_filter(UserJID, Borders, Start, End, WithJID, MsgID),
265
:-(
case Strategy of
266 last_page ->
267
:-(
lookup_messages_last_page(PoolName, HostType, UserJID, RSM, PageSize, Filter);
268 by_offset ->
269
:-(
lookup_messages_by_offset(PoolName, HostType, UserJID, RSM, PageSize, Filter);
270 first_page ->
271
:-(
lookup_messages_first_page(PoolName, HostType, UserJID, RSM, PageSize, Filter);
272 before_id ->
273
:-(
lookup_messages_before_id(PoolName, HostType, UserJID, RSM, PageSize, Filter);
274 after_id ->
275
:-(
lookup_messages_after_id(PoolName, HostType, UserJID, RSM, PageSize, Filter)
276 end.
277
278 rsm_to_strategy(#rsm_in{direction = before, id = undefined}) ->
279
:-(
last_page;
280 rsm_to_strategy(#rsm_in{direction = undefined, index = 0}) ->
281
:-(
first_page;
282 rsm_to_strategy(#rsm_in{direction = undefined, index = Offset}) when is_integer(Offset) ->
283
:-(
by_offset;
284 rsm_to_strategy(#rsm_in{direction = before, id = Id}) when is_integer(Id) ->
285
:-(
before_id;
286 rsm_to_strategy(#rsm_in{direction = aft, id = Id}) when is_integer(Id) ->
287
:-(
after_id;
288 rsm_to_strategy(#rsm_in{}) ->
289
:-(
first_page;
290 rsm_to_strategy(undefined) ->
291
:-(
first_page.
292
293 lookup_messages_simple(PoolName, HostType, UserJID,
294 #rsm_in{direction = aft, id = ID},
295 PageSize, Filter) ->
296 %% Get last rows from result set
297
:-(
MessageRows = extract_messages(PoolName, UserJID, HostType, after_id(ID, Filter), PageSize, false),
298
:-(
{ok, {undefined, undefined, rows_to_uniform_format(HostType, MessageRows)}};
299 lookup_messages_simple(PoolName, HostType, UserJID,
300 #rsm_in{direction = before, id = ID},
301 PageSize, Filter) ->
302
:-(
MessageRows = extract_messages(PoolName, UserJID, HostType, before_id(ID, Filter), PageSize, true),
303
:-(
{ok, {undefined, undefined, rows_to_uniform_format(HostType, MessageRows)}};
304 lookup_messages_simple(PoolName, HostType, UserJID,
305 #rsm_in{direction = undefined, index = Offset},
306 PageSize, Filter) ->
307 %% Apply offset
308
:-(
StartId = offset_to_start_id(PoolName, UserJID, Filter,
309 Offset), %% POTENTIALLY SLOW AND NOT SIMPLE :)
310
:-(
MessageRows = extract_messages(PoolName, UserJID, HostType, from_id(StartId, Filter), PageSize,
311 false),
312
:-(
{ok, {undefined, undefined, rows_to_uniform_format(HostType, MessageRows)}};
313 lookup_messages_simple(PoolName, HostType, UserJID,
314 _,
315 PageSize, Filter) ->
316
:-(
MessageRows = extract_messages(PoolName, UserJID, HostType, Filter, PageSize, false),
317
:-(
{ok, {undefined, undefined, rows_to_uniform_format(HostType, MessageRows)}}.
318
319 lookup_messages_last_page(PoolName, HostType, UserJID,
320 #rsm_in{direction = before, id = undefined},
321 0, Filter) ->
322 %% Last page
323
:-(
TotalCount = calc_count(PoolName, UserJID, HostType, Filter),
324
:-(
{ok, {TotalCount, TotalCount, []}};
325 lookup_messages_last_page(PoolName, HostType, UserJID,
326 #rsm_in{direction = before, id = undefined},
327 PageSize, Filter) ->
328 %% Last page
329
:-(
MessageRows = extract_messages(PoolName, UserJID, HostType, Filter, PageSize, true),
330
:-(
MessageRowsCount = length(MessageRows),
331
:-(
case MessageRowsCount < PageSize of
332 true ->
333
:-(
{ok, {MessageRowsCount, 0,
334 rows_to_uniform_format(HostType, MessageRows)}};
335 false ->
336
:-(
FirstID = row_to_message_id(hd(MessageRows)),
337
:-(
Offset = calc_count(PoolName, UserJID, HostType, before_id(FirstID, Filter)),
338
:-(
{ok, {Offset + MessageRowsCount, Offset,
339 rows_to_uniform_format(HostType, MessageRows)}}
340 end.
341
342 lookup_messages_by_offset(PoolName, HostType, UserJID,
343 #rsm_in{direction = undefined, index = Offset},
344 0, Filter) when is_integer(Offset) ->
345 %% By offset
346
:-(
TotalCount = calc_count(PoolName, UserJID, HostType, Filter),
347
:-(
{ok, {TotalCount, Offset, []}};
348 lookup_messages_by_offset(PoolName, HostType, UserJID,
349 #rsm_in{direction = undefined, index = Offset},
350 PageSize, Filter) when is_integer(Offset) ->
351 %% By offset
352
:-(
StartId = offset_to_start_id(PoolName, UserJID, Filter, Offset), %% POTENTIALLY SLOW
353
:-(
MessageRows = extract_messages(PoolName, UserJID, HostType, from_id(StartId, Filter), PageSize,
354 false),
355
:-(
MessageRowsCount = length(MessageRows),
356
:-(
case MessageRowsCount < PageSize of
357 true ->
358
:-(
{ok, {Offset + MessageRowsCount, Offset,
359 rows_to_uniform_format(HostType, MessageRows)}};
360 false ->
361
:-(
LastID = row_to_message_id(lists:last(MessageRows)),
362
:-(
CountAfterLastID = calc_count(PoolName, UserJID, HostType, after_id(LastID, Filter)),
363
:-(
{ok, {Offset + MessageRowsCount + CountAfterLastID, Offset,
364 rows_to_uniform_format(HostType, MessageRows)}}
365 end.
366
367 lookup_messages_first_page(PoolName, HostType, UserJID, _, 0, Filter) ->
368 %% First page, just count
369
:-(
TotalCount = calc_count(PoolName, UserJID, HostType, Filter),
370
:-(
{ok, {TotalCount, 0, []}};
371 lookup_messages_first_page(PoolName, HostType, UserJID, _, PageSize, Filter) ->
372 %% First page
373
:-(
MessageRows = extract_messages(PoolName, UserJID, HostType, Filter, PageSize, false),
374
:-(
MessageRowsCount = length(MessageRows),
375
:-(
case MessageRowsCount < PageSize of
376 true ->
377 %% Total number of messages is less than one page
378
:-(
{ok, {MessageRowsCount, 0,
379 rows_to_uniform_format(HostType, MessageRows)}};
380 false ->
381
:-(
LastID = row_to_message_id(lists:last(MessageRows)),
382
:-(
CountAfterLastID = calc_count(PoolName, UserJID, HostType, after_id(LastID, Filter)),
383
:-(
{ok, {MessageRowsCount + CountAfterLastID, 0,
384 rows_to_uniform_format(HostType, MessageRows)}}
385 end.
386
387 lookup_messages_before_id(PoolName, HostType, UserJID,
388 RSM = #rsm_in{direction = before, id = ID},
389 PageSize, Filter) ->
390
:-(
TotalCount = calc_count(PoolName, UserJID, HostType, Filter),
391
:-(
Offset = calc_offset(PoolName, UserJID, HostType, Filter, PageSize, TotalCount, RSM),
392
:-(
MessageRows = extract_messages(PoolName, UserJID, HostType, to_id(ID, Filter),
393 PageSize + 1, true),
394
:-(
Result = {TotalCount, Offset, rows_to_uniform_format(HostType, MessageRows)},
395
:-(
mod_mam_utils:check_for_item_not_found(RSM, PageSize, Result).
396
397 lookup_messages_after_id(PoolName, HostType, UserJID,
398 RSM = #rsm_in{direction = aft, id = ID},
399 PageSize, Filter) ->
400
:-(
TotalCount = calc_count(PoolName, UserJID, HostType, Filter),
401
:-(
Offset = calc_offset(PoolName, UserJID, HostType, Filter, PageSize, TotalCount, RSM),
402
:-(
MessageRows = extract_messages(PoolName, UserJID, HostType, from_id(ID, Filter),
403 PageSize + 1, false),
404
:-(
Result = {TotalCount, Offset, rows_to_uniform_format(HostType, MessageRows)},
405
:-(
mod_mam_utils:check_for_item_not_found(RSM, PageSize, Result).
406
407
408 after_id(ID, Filter = #mam_ca_filter{start_id = AfterID}) ->
409
:-(
Filter#mam_ca_filter{start_id = maybe_max(ID + 1, AfterID)}.
410
411 before_id(undefined, Filter) ->
412
:-(
Filter;
413 before_id(ID, Filter = #mam_ca_filter{end_id = BeforeID}) ->
414
:-(
Filter#mam_ca_filter{end_id = maybe_min(ID - 1, BeforeID)}.
415
416 to_id(ID, Filter = #mam_ca_filter{end_id = BeforeID}) ->
417
:-(
Filter#mam_ca_filter{end_id = maybe_min(ID, BeforeID)}.
418
419 from_id(ID, Filter = #mam_ca_filter{start_id = AfterID}) ->
420
:-(
Filter#mam_ca_filter{start_id = maybe_max(ID, AfterID)}.
421
422
423 rows_to_uniform_format(HostType, MessageRows) ->
424
:-(
[row_to_uniform_format(HostType, Row) || Row <- MessageRows].
425
426 row_to_uniform_format(HostType, #{from_jid := FromJID, message := Msg, id := MsgID}) ->
427
:-(
SrcJID = jid:from_binary(FromJID),
428
:-(
Packet = stored_binary_to_packet(HostType, Msg),
429
:-(
#{id => MsgID, jid => SrcJID, packet => Packet}.
430
431 row_to_message_id(#{id := MsgID}) ->
432
:-(
MsgID.
433
434 -spec get_mam_pm_gdpr_data(Acc, Params, Extra) -> {ok, Acc} when
435 Acc :: ejabberd_gen_mam_archive:mam_pm_gdpr_data(),
436 Params :: #{jid := jid:jid()},
437 Extra :: gen_hook:extra().
438 get_mam_pm_gdpr_data(Acc, #{jid := JID}, #{host_type := HostType}) ->
439
:-(
BinJID = jid:to_binary(jid:to_lower(JID)),
440
:-(
FilterMap = #{user_jid => BinJID, with_jid => <<"">>},
441
:-(
Rows = fetch_user_messages(pool_name(HostType), JID, FilterMap),
442
:-(
Messages = [rows_to_gdpr_mam_message(HostType, Row) || Row <- Rows],
443
:-(
{ok, Messages ++ Acc}.
444
445 rows_to_gdpr_mam_message(HostType, #{message := Data, id:= Id, from_jid:=FromJid}) ->
446
:-(
{Id, FromJid, exml:to_binary(stored_binary_to_packet(HostType, Data))}.
447
448 %% Offset is not supported
449 %% Each record is a tuple of form
450 %% `{<<"13663125233">>, <<"bob@localhost">>, <<"res1">>, <<binary>>}'.
451 %% Columns are `["id", "from_jid", "message"]'.
452 -spec extract_messages(PoolName, UserJID, HostType, Filter, IMax, ReverseLimit) ->
453 [Row] when
454 PoolName :: mongoose_cassandra:pool_name(),
455 UserJID :: jid:jid(),
456 HostType :: host_type(),
457 Filter :: filter(),
458 IMax :: pos_integer(),
459 ReverseLimit :: boolean(),
460 Row :: mongoose_cassandra:row().
461 extract_messages(_Worker, _UserJID, _HostType, _Filter, 0, _) ->
462
:-(
[];
463 extract_messages(PoolName, UserJID, _HostType, Filter, IMax, false) ->
464
:-(
QueryName = {extract_messages_query, select_filter(Filter)},
465
:-(
Params = maps:put('[limit]', IMax, eval_filter_params(Filter)),
466
:-(
{ok, Rows} = mongoose_cassandra:cql_read(PoolName, UserJID, ?MODULE, QueryName, Params),
467
:-(
Rows;
468 extract_messages(PoolName, UserJID, _HostType, Filter, IMax, true) ->
469
:-(
QueryName = {extract_messages_r_query, select_filter(Filter)},
470
:-(
Params = maps:put('[limit]', IMax, eval_filter_params(Filter)),
471
:-(
{ok, Rows} = mongoose_cassandra:cql_read(PoolName, UserJID, ?MODULE, QueryName, Params),
472
:-(
lists:reverse(Rows).
473
474 fetch_user_messages(PoolName, UserJID, FilterMap) ->
475
:-(
QueryName = fetch_user_messages_query,
476
:-(
{ok, Rows} = mongoose_cassandra:cql_read(PoolName, UserJID, ?MODULE, QueryName, FilterMap),
477
:-(
Rows.
478
479
480 %% @doc Calculate a zero-based index of the row with UID in the result test.
481 %%
482 %% If the element does not exists, the ID of the next element will
483 %% be returned instead.
484 %% @end
485 -spec calc_index(PoolName, UserJID, HostType, Filter, MessID) -> Count
486 when
487 PoolName :: mongoose_cassandra:pool_name(),
488 UserJID :: jid:jid(),
489 HostType :: host_type(),
490 Filter :: filter(),
491 MessID :: message_id(),
492 Count :: non_neg_integer().
493 calc_index(PoolName, UserJID, HostType, Filter, MessID) ->
494
:-(
calc_count(PoolName, UserJID, HostType, to_id(MessID, Filter)).
495
496 %% @doc Count of elements in RSet before the passed element.
497 %%
498 %% The element with the passed UID can be already deleted.
499 %% @end
500 -spec calc_before(PoolName, UserJID, HostType, Filter, MessID) -> Count
501 when
502 PoolName :: mongoose_cassandra:pool_name(),
503 UserJID :: jid:jid(),
504 HostType :: host_type(),
505 Filter :: filter(),
506 MessID :: message_id(),
507 Count :: non_neg_integer().
508 calc_before(PoolName, UserJID, HostType, Filter, MessID) ->
509
:-(
calc_count(PoolName, UserJID, HostType, before_id(MessID, Filter)).
510
511
512 %% @doc Get the total result set size.
513 %% "SELECT COUNT(*) as "count" FROM mam_message WHERE "
514 -spec calc_count(PoolName, UserJID, HostType, Filter) -> Count
515 when
516 PoolName :: mongoose_cassandra:pool_name(),
517 UserJID :: jid:jid(),
518 HostType :: host_type(),
519 Filter :: filter(),
520 Count :: non_neg_integer().
521 calc_count(PoolName, UserJID, _HostType, Filter) ->
522
:-(
QueryName = {calc_count_query, select_filter(Filter)},
523
:-(
Params = eval_filter_params(Filter),
524
:-(
{ok, [#{count := Count}]} = mongoose_cassandra:cql_read(PoolName, UserJID, ?MODULE, QueryName, Params),
525
:-(
Count.
526
527 %% @doc Convert offset to index of the first entry
528 %% Returns undefined if not there are not enough rows
529 %% Uses previously calculated offsets to speed up queries
530 -spec offset_to_start_id(PoolName, UserJID, Filter, Offset) -> Id
531 when
532 PoolName :: mongoose_cassandra:pool_name(),
533 UserJID :: jid:jid(),
534 Offset :: non_neg_integer(),
535 Filter :: filter(),
536 Id :: non_neg_integer() | undefined.
537 offset_to_start_id(PoolName, UserJID, Filter, Offset) when is_integer(Offset), Offset >= 0,
538 Offset =< 100 ->
539
:-(
calc_offset_to_start_id(PoolName, UserJID, Filter, Offset);
540 offset_to_start_id(PoolName, UserJID, Filter, Offset) when is_integer(Offset), Offset >= 0 ->
541
:-(
Params = maps:put(offset, Offset, eval_filter_params(Filter)),
542 %% Try to find already calculated nearby offset to reduce query size
543
:-(
case mongoose_cassandra:cql_read(PoolName, UserJID, ?MODULE, prev_offset_query, Params) of
544 {ok, []} -> %% No hints, just calculate offset sloooowly
545
:-(
StartId = calc_offset_to_start_id(PoolName, UserJID, Filter, Offset),
546
:-(
maybe_save_offset_hint(PoolName, UserJID, Filter, 0, Offset, StartId);
547 {ok, [#{offset := PrevOffset, id := PrevId}]} ->
548 %% Offset hint found, use it to reduce query size
549
:-(
case Offset of
550
:-(
PrevOffset -> PrevId;
551 _ ->
552
:-(
StartId = calc_offset_to_start_id(PoolName, UserJID,
553 Filter#mam_ca_filter{start_id = PrevId},
554 Offset - PrevOffset + 1),
555
:-(
maybe_save_offset_hint(PoolName, UserJID, Filter, PrevOffset, Offset, StartId)
556 end
557 end.
558
559 %% @doc Saves offset hint for future use in order to speed up queries with similar offset
560 %% Hint is save only if previous offset hint was 50+ entires from current query
561 %% This function returns given StartId as passthrough for convenience
562 -spec maybe_save_offset_hint(PoolName :: mongoose_cassandra:pool_name(), UserJID :: jid:jid(),
563 Filter :: filter(), HintOffset :: non_neg_integer(),
564 NewOffset :: non_neg_integer(),
565 StartId :: non_neg_integer() | undefined) ->
566 StartId :: non_neg_integer() | undefined.
567 maybe_save_offset_hint(_PoolName, _UserJID, _Filter, _HintOffset, _NewOffset,
568 StartId = undefined) ->
569
:-(
StartId;
570 maybe_save_offset_hint(PoolName, UserJID, Filter, HintOffset, NewOffset, StartId) ->
571
:-(
case abs(NewOffset - HintOffset) > 50 of
572 true ->
573
:-(
#mam_ca_filter{user_jid = FUserJID, with_jid = FWithJID} = Filter,
574
:-(
Row = #{user_jid => FUserJID, with_jid => FWithJID, offset => NewOffset, id => StartId},
575
:-(
mongoose_cassandra:cql_write(PoolName, UserJID, ?MODULE,
576 insert_offset_hint_query, [Row]);
577 false ->
578
:-(
skip
579 end,
580
:-(
StartId.
581
582 %% @doc Convert offset to index of the first entry
583 %% Returns undefined if not there are not enough rows
584 -spec calc_offset_to_start_id(PoolName, UserJID, Filter, Offset) -> Id
585 when
586 PoolName :: mongoose_cassandra:pool_name(),
587 UserJID :: jid:jid(),
588 Offset :: non_neg_integer(),
589 Filter :: filter(),
590 Id :: non_neg_integer() | undefined.
591 calc_offset_to_start_id(PoolName, UserJID, Filter, Offset) when is_integer(Offset), Offset >= 0 ->
592
:-(
QueryName = {list_message_ids_query, select_filter(Filter)},
593
:-(
Params = maps:put('[limit]', Offset + 1, eval_filter_params(Filter)),
594
:-(
{ok, RowsIds} = mongoose_cassandra:cql_read(PoolName, UserJID, ?MODULE, QueryName, Params),
595
:-(
case RowsIds of
596
:-(
[] -> undefined;
597 [_ | _] ->
598
:-(
maps:get(id, lists:last(RowsIds))
599 end.
600
601 %% @doc Get closest offset -> message id 'hint' for specified offset
602 prev_offset_query_cql() ->
603
:-(
"SELECT id, offset FROM mam_message_offset WHERE user_jid = ? and with_jid = ? "
604 "and offset <= ? LIMIT 1".
605
606 %% @doc Insert offset -> message id 'hint'
607 insert_offset_hint_query_cql() ->
608
:-(
"INSERT INTO mam_message_offset(user_jid, with_jid, id, offset) VALUES(?, ?, ?, ?)".
609
610 prepare_filter(UserJID, Borders, Start, End, WithJID, MsgID) ->
611
:-(
BUserJID = bare_jid(UserJID),
612 %% In Cassandra, a column cannot be restricted by both an equality and an inequality relation.
613 %% When MsgID is defined, it is used as both StartID and EndID to comply with this limitation.
614 %% This means that the `ids` filter effectively overrides any "before" or "after" filters.
615
:-(
{StartID, EndID} = case MsgID of
616 undefined ->
617
:-(
mod_mam_utils:calculate_msg_id_borders(Borders, Start, End);
618 ID ->
619
:-(
{ID, ID}
620 end,
621
:-(
BWithJID = maybe_full_jid(WithJID), %% it's NOT optional field
622
:-(
prepare_filter_params(BUserJID, BWithJID, StartID, EndID).
623
624 prepare_filter_params(BUserJID, BWithJID, StartID, EndID) ->
625
:-(
#mam_ca_filter{
626 user_jid = BUserJID,
627 with_jid = BWithJID,
628 start_id = StartID,
629 end_id = EndID
630 }.
631
632 eval_filter_params(#mam_ca_filter{
633 user_jid = BUserJID,
634 with_jid = BWithJID,
635 start_id = StartID,
636 end_id = EndID
637 }) ->
638
:-(
Optional = maps:filter(fun(_K, V) -> V =/= undefined end,
639 #{start_id => StartID, end_id =>EndID}),
640
:-(
maps:merge(#{user_jid => BUserJID, with_jid => BWithJID}, Optional).
641
642 select_filter(#mam_ca_filter{
643 start_id = StartID,
644 end_id = EndID
645 }) ->
646
:-(
select_filter(StartID, EndID).
647
648
649 -spec select_filter(StartID, EndID) ->
650 all | 'end' | start | start_end when
651 StartID :: integer() | undefined,
652 EndID :: integer() | undefined.
653 select_filter(undefined, undefined) ->
654
:-(
all;
655 select_filter(undefined, _) ->
656
:-(
'end';
657 select_filter(_, undefined) ->
658
:-(
start;
659 select_filter(_, _) ->
660
:-(
start_end.
661
662 prepare_filter_cql(StartID, EndID) ->
663 case StartID of
664
:-(
undefined -> "";
665
:-(
_ -> " AND id >= :start_id"
666
:-(
end ++
667 case EndID of
668
:-(
undefined -> "";
669
:-(
_ -> " AND id <= :end_id"
670 end.
671
672 filter_to_cql() ->
673
:-(
[{select_filter(StartID, EndID), prepare_filter_cql(StartID, EndID)}
674
:-(
|| StartID <- [undefined, 0], EndID <- [undefined, 0]].
675
676 -spec calc_offset(PoolName, UserJID, HostType, Filter, PageSize, TotalCount, RSM) -> Offset
677 when
678 PoolName :: mongoose_cassandra:pool_name(),
679 UserJID :: jid:jid(),
680 HostType :: host_type(),
681 Filter :: filter(),
682 PageSize :: non_neg_integer(),
683 TotalCount :: non_neg_integer(),
684 RSM :: jlib:rsm_in() | undefined,
685 Offset :: non_neg_integer().
686 %% Requesting the Last Page in a Result Set
687 calc_offset(_W, _UserJID, _LS, _F, PS, TC, #rsm_in{direction = before, id = undefined}) ->
688
:-(
max(0, TC - PS);
689 calc_offset(PoolName, UserJID, HostType, F, PS, _TC, #rsm_in{direction = before, id = ID})
690 when is_integer(ID) ->
691
:-(
max(0, calc_before(PoolName, UserJID, HostType, F, ID) - PS);
692 calc_offset(PoolName, UserJID, HostType, F, _PS, _TC, #rsm_in{direction = aft, id = ID})
693 when is_integer(ID) ->
694
:-(
calc_index(PoolName, UserJID, HostType, F, ID);
695 calc_offset(_W, _UserJID, _LS, _F, _PS, _TC, _RSM) ->
696
:-(
0.
697
698 -spec maybe_full_jid(undefined | jid:jid()) -> undefined | binary().
699
:-(
maybe_full_jid(undefined) -> <<>>;
700 maybe_full_jid(JID) ->
701
:-(
jid:to_binary(jid:to_lower(JID)).
702
703 %%====================================================================
704 %% Internal SQL part
705 %%====================================================================
706
707 extract_messages_queries() ->
708
:-(
[{{extract_messages_query, FilterName}, extract_messages_cql(Filter)}
709
:-(
|| {FilterName, Filter} <- filter_to_cql()].
710
711 extract_messages_r_queries() ->
712
:-(
[{{extract_messages_r_query, FilterName}, extract_messages_r_cql(Filter)}
713
:-(
|| {FilterName, Filter} <- filter_to_cql()].
714
715 calc_count_queries() ->
716
:-(
[{{calc_count_query, FilterName}, calc_count_cql(Filter)}
717
:-(
|| {FilterName, Filter} <- filter_to_cql()].
718
719 list_message_ids_queries() ->
720
:-(
[{{list_message_ids_query, FilterName}, list_message_ids_cql(Filter)}
721
:-(
|| {FilterName, Filter} <- filter_to_cql()].
722
723 extract_messages_cql(Filter) ->
724 "SELECT id, from_jid, message FROM mam_message "
725
:-(
"WHERE user_jid = ? AND with_jid = ? " ++
726 Filter ++ " ORDER BY id LIMIT ?".
727
728 extract_messages_r_cql(Filter) ->
729 "SELECT id, from_jid, message FROM mam_message "
730
:-(
"WHERE user_jid = ? AND with_jid = ? " ++
731 Filter ++ " ORDER BY id DESC LIMIT ?".
732
733 calc_count_cql(Filter) ->
734 "SELECT COUNT(*) FROM mam_message "
735
:-(
"WHERE user_jid = ? AND with_jid = ? " ++ Filter.
736
737 list_message_ids_cql(Filter) ->
738 "SELECT id FROM mam_message "
739
:-(
"WHERE user_jid = ? AND with_jid = ? " ++ Filter ++
740 " ORDER BY id LIMIT ?".
741
742 fetch_user_messages_cql() ->
743
:-(
"SELECT id, from_jid, message FROM mam_message "
744 "WHERE user_jid = ? AND with_jid = ? "
745 "ORDER BY id".
746
747
748
749 %% ----------------------------------------------------------------------
750 %% Optimizations
751
752 packet_to_stored_binary(HostType, Packet) ->
753 %% Module implementing mam_message behaviour
754
:-(
Module = db_message_format(HostType),
755
:-(
mam_message:encode(Module, Packet).
756
757 stored_binary_to_packet(HostType, Bin) ->
758 %% Module implementing mam_message behaviour
759
:-(
Module = db_message_format(HostType),
760
:-(
mam_message:decode(Module, Bin).
761
762 %% ----------------------------------------------------------------------
763 %% Params getters
764
765 -spec db_message_format(HostType :: host_type()) -> module().
766 db_message_format(HostType) ->
767
:-(
gen_mod:get_module_opt(HostType, ?MODULE, db_message_format).
768
769 -spec pool_name(HostType :: host_type()) -> term().
770 pool_name(_HostType) ->
771
:-(
default.
Line Hits Source