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