23044468042

Committed 13 Mar 2026 09:24AM UTC coverage: 85.152% (+0.003%) from 85.149%

Build # 23044468042

Build Type

push

github

Committed by

web-flow

Commit Message

Merge pull request #4656 from esl/traffic-monitor

MIM-2088 GraphQL subscription to receive all stanzas

Coverage Stats

70 of 74 new or added lines in 7 files covered. (94.59%)

6 existing lines in 4 files now uncovered.

28239 of 33163 relevant lines covered (85.15%)

41387.25 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

70.15

/src/inbox/mod_inbox_rdbms_async.erl

-module(mod_inbox_rdbms_async).

-include("mod_inbox.hrl").
-include("mongoose_logger.hrl").

-behaviour(mod_inbox_backend).
-behaviour(mongoose_aggregator_worker).

-type box() :: binary().
-type task() ::
    {set_inbox, mod_inbox:entry_key(), exml:element(), pos_integer(), id(), integer(), box()} |
    {set_inbox_incr_unread, mod_inbox:entry_key(), exml:element(), id(), integer(), pos_integer(), box()} |
    {remove_inbox_row, mod_inbox:entry_key()} |
    {reset_unread, mod_inbox:entry_key(), id(), integer()}.

%% API
-export([init/2,
         set_inbox/6,
         set_inbox_incr_unread/5,
         reset_unread/4,
         remove_inbox_row/2,
         empty_user_bin/4,
         empty_domain_bin/3,
         empty_global_bin/2,
         remove_domain/2,
         clear_inbox/3,
         get_inbox/4,
         get_inbox_unread/2,
         get_full_entry/2,
         get_entry_properties/2,
         set_entry_properties/3]).
-export([stop/1]).

%% Worker callbacks
-export([request/2, aggregate/3, verify/3]).

%% Initialisation
-spec init(mongooseim:host_type(), gen_mod:module_opts()) -> ok.
init(HostType, Opts) ->
    AsyncOpts = prepare_pool_opts(Opts),
    mod_inbox_rdbms:init(HostType, Opts),
    prepare_deletes(HostType, Opts),
    start_pool(HostType, AsyncOpts),
    ok.

stop(HostType) ->
    mongoose_async_pools:stop_pool(HostType, inbox).

prepare_pool_opts(#{async_writer := AsyncOpts}) ->
    AsyncOpts#{pool_type => aggregate,
               request_callback => fun ?MODULE:request/2,
               aggregate_callback => fun ?MODULE:aggregate/3,
               verify_callback => fun ?MODULE:verify/3}.

prepare_deletes(_HostType, _Opts) ->
    mongoose_rdbms:prepare(inbox_move_conversation_to_bin, inbox,
                           [luser, lserver, remote_bare_jid],
                           <<"UPDATE inbox SET box='bin'",
                             " WHERE luser = ? AND lserver = ? AND remote_bare_jid = ?">>),
    ok.

-spec start_pool(mongooseim:host_type(), mongoose_async_pools:pool_opts()) -> term().
start_pool(HostType, Opts) ->
    mongoose_async_pools:start_pool(HostType, inbox, Opts).

%% Worker callbacks
-spec request(task(), mongoose_async_pools:pool_extra()) -> gen_server:request_id().
request(Task, _Extra = #{host_type := HostType}) ->
    request_one(HostType, Task).

request_one(HostType, {set_inbox, {LUser, LServer, LToBareJid}, Packet, Count, MsgId, Timestamp, Box}) ->
    Content = exml:to_binary(Packet),
    Update = [MsgId, Box, Content, Count, Timestamp],
    Insert = [LUser, LServer, LToBareJid, MsgId, Box, Content, Count, Timestamp],
    rdbms_queries:request_upsert(HostType, inbox_upsert, Insert, Update);
request_one(HostType, {set_inbox_incr_unread, {LUser, LServer, LToBareJid}, Packet, MsgId, Timestamp, Incrs, Box}) ->
    Content = exml:to_binary(Packet),
    Update = [MsgId, Box, Content, Incrs, Timestamp],
    Insert = [LUser, LServer, LToBareJid, MsgId, Box, Content, Incrs, Timestamp],
    rdbms_queries:request_upsert(HostType, inbox_upsert_incr_unread, Insert, Update);
request_one(HostType, {reset_unread, {LUser, LServer, LToBareJid}, undefined, TS}) ->
    mongoose_rdbms:execute_request(HostType, inbox_reset_unread, [LUser, LServer, LToBareJid, TS]);
request_one(HostType, {reset_unread, {LUser, LServer, LToBareJid}, MsgId, TS}) ->
    mongoose_rdbms:execute_request(HostType, inbox_reset_unread_msg, [LUser, LServer, LToBareJid, MsgId, TS]);
request_one(HostType, {remove_inbox_row, {LUser, LServer, LToBareJid}}) ->
    mongoose_rdbms:execute_request(HostType, inbox_move_conversation_to_bin, [LUser, LServer, LToBareJid]).

-spec aggregate(task(), task(), mongoose_async_pools:pool_extra()) -> {ok, task()}.
aggregate(Current, NewTask, _Extra) ->
    {ok, aggregate(Current, NewTask)}.

-spec verify(term(), task(), mongoose_async_pools:pool_extra()) -> ok | {error, term()}.
verify(Answer, InboxTask, _Extra) ->
    case mod_inbox_rdbms:check_result(Answer) of
        {error, Reason} ->
            {LU, LS, LRem} = element(2, InboxTask),
            ?LOG_WARNING(#{what => inbox_process_message_failed, reason => Reason,
                           from_jid => jid:to_binary({LU, LS}), to_jid => LRem}),
            {error, Reason};
        _ -> ok
    end.

%% async callbacks
-spec set_inbox(mongooseim:host_type(), mod_inbox:entry_key(),
                exml:element(), Count :: integer(), id(), Timestamp :: integer()) ->
    mod_inbox:write_res().
set_inbox(HostType, Entry, Packet, Count, MsgId, Timestamp) ->
    Params = {set_inbox, Entry, Packet, Count, MsgId, Timestamp, <<"inbox">>},
    mongoose_async_pools:put_task(HostType, inbox, Entry, Params).

-spec set_inbox_incr_unread(mongooseim:host_type(), mod_inbox:entry_key(),
                            exml:element(), MsgId :: binary(), Timestamp :: integer()) ->
    mod_inbox:count_res().
set_inbox_incr_unread(HostType, Entry, Packet, MsgId, Timestamp) ->
    Params = {set_inbox_incr_unread, Entry, Packet, MsgId, Timestamp, 1, <<"inbox">>},
    mongoose_async_pools:put_task(HostType, inbox, Entry, Params).

-spec reset_unread(mongooseim:host_type(), mod_inbox:entry_key(), binary() | undefined, integer()) ->
    mod_inbox:write_res().
reset_unread(HostType, Entry, MsgId, TS) ->
    Params = {reset_unread, Entry, MsgId, TS},
    mongoose_async_pools:put_task(HostType, inbox, Entry, Params).

-spec remove_inbox_row(mongooseim:host_type(), mod_inbox:entry_key()) -> mod_inbox:write_res().
remove_inbox_row(HostType, Entry) ->
    Params = {remove_inbox_row, Entry},
    mongoose_async_pools:put_task(HostType, inbox, Entry, Params).

%% synchronous callbacks
-spec get_inbox(mongooseim:host_type(), jid:luser(), jid:lserver(), mod_inbox:get_inbox_params()) ->
    [mod_inbox:inbox_res()].
get_inbox(HostType, LUser, LServer, Params) ->
    mod_inbox_rdbms:get_inbox(HostType, LUser, LServer, Params).

-spec get_inbox_unread(mongooseim:host_type(), mod_inbox:entry_key()) ->
    {ok, integer()}.
get_inbox_unread(HostType, Entry) ->
    mod_inbox_rdbms:get_inbox_unread(HostType, Entry).

-spec remove_domain(mongooseim:host_type(), jid:lserver()) -> term().
remove_domain(HostType, LServer) ->
    mod_inbox_rdbms:remove_domain(HostType, LServer).

-spec clear_inbox(mongooseim:host_type(), jid:luser(), jid:lserver()) ->
    mod_inbox:write_res().
clear_inbox(HostType, LUser, LServer) ->
    mod_inbox_rdbms:clear_inbox(HostType, LUser, LServer).

-spec get_full_entry(mongooseim:host_type(), mod_inbox:entry_key()) ->
    inbox_res() | nil().
get_full_entry(HostType, Entry) ->
    mod_inbox_rdbms:get_full_entry(HostType, Entry).

-spec get_entry_properties(mongooseim:host_type(), mod_inbox:entry_key()) ->
    entry_properties() | nil().
get_entry_properties(HostType, Entry) ->
    mod_inbox_rdbms:get_entry_properties(HostType, Entry).

-spec set_entry_properties(mongooseim:host_type(), mod_inbox:entry_key(), entry_properties()) ->
    entry_properties() | {error, binary()}.
set_entry_properties(HostType, Entry, Properties) ->
    mod_inbox_rdbms:set_entry_properties(HostType, Entry, Properties).

-spec empty_user_bin(HostType :: mongooseim:host_type(),
                     LServer :: jid:lserver(),
                     LUser :: jid:luser(),
                     TS :: integer()) -> non_neg_integer().
empty_user_bin(HostType, LServer, LUser, TS) ->
    mod_inbox_rdbms:empty_user_bin(HostType, LServer, LUser, TS).

-spec empty_domain_bin(HostType :: mongooseim:host_type(),
                       LServer :: jid:lserver(),
                       TS :: integer()) -> non_neg_integer().
empty_domain_bin(HostType, LServer, TS) ->
    mod_inbox_rdbms:empty_domain_bin(HostType, LServer, TS).

-spec empty_global_bin(HostType :: mongooseim:host_type(),
                       TS :: integer()) -> non_neg_integer().
empty_global_bin(HostType, TS) ->
    mod_inbox_rdbms:empty_global_bin(HostType, TS).

-spec aggregate(CurrentlyAccumulatedTask :: task(), NewTask :: task()) -> FinalTask :: task().
%%% if new task is remove_row, do the previous with an updated box
aggregate({set_inbox, Entry, Content, Count, MsgId, Timestamp, _},
          {remove_inbox_row, _}) ->
    {set_inbox, Entry, Content, Count, MsgId, Timestamp, <<"bin">>};
aggregate({set_inbox_incr_unread, Entry, Content, MsgId, Timestamp, Incrs, _},
          {remove_inbox_row, _}) ->
    {set_inbox_incr_unread, Entry, Content, MsgId, Timestamp, Incrs, <<"bin">>};
aggregate(_, {remove_inbox_row, _} = NewTask) ->
    NewTask;

%%% if the last task was remove_row, this task should now only be an insert
aggregate({remove_inbox_row, _} = OldTask, {reset_unread, _, _, _}) ->
    OldTask;
aggregate({remove_inbox_row, _},
          {set_inbox, Entry, Content, Count, MsgId, Timestamp, _}) ->
    {set_inbox, Entry, Content, Count, MsgId, Timestamp, <<"bin">>};
aggregate({remove_inbox_row, _},
          {set_inbox_incr_unread, Entry, Content, MsgId, Timestamp, Incrs, _}) ->
    {set_inbox_incr_unread, Entry, Content, MsgId, Timestamp, Incrs, <<"bin">>};

%%% If the last task was a reset_unread,
%   we prefer explicit resets,
%   then adhoc newer resets,
%   then we accumulate inserts
%% an undefined means an explicit request to reset, it has priority
aggregate({reset_unread, _, _, _}, {reset_unread, _, undefined, _} = NewTask) ->
    NewTask;
%% an undefined means an explicit request to reset, it has priority
aggregate({reset_unread, _, undefined, _} = OldTask, {reset_unread, _, _, _}) ->
    OldTask;
%% both are adhoc, we prefer the newer
aggregate({reset_unread, _, _, _}, {reset_unread, _, _, _} = NewTask) ->
    NewTask;
aggregate({reset_unread, _, _, _}, {set_inbox, _, _, _, _, _, _} = NewTask) ->
    NewTask;
%% Here `Count` becomes an absolute value instead of an increment
aggregate({reset_unread, _, _, _},
          {set_inbox_incr_unread, Entry, Content, MsgId, Timestamp, Incrs, Box}) ->
    {set_inbox, Entry, Content, Incrs, MsgId, Timestamp, Box};

%%% If the last task was a set_inbox
%% Reset is an explicit reset-to-zero, so do reset the counter
aggregate({set_inbox, Entry, Content, _, MsgId, Timestamp, Box},
          {reset_unread, _, undefined, _}) ->
    {set_inbox, Entry, Content, 0, MsgId, Timestamp, Box};
%% Reset refers to that same set_inbox
aggregate({set_inbox, Entry, Content, _, MsgId, Timestamp, Box},
          {reset_unread, _, MsgId, _}) ->
    {set_inbox, Entry, Content, 0, MsgId, Timestamp, Box};
%% Reset refers to some other set_inbox
aggregate({set_inbox, _, _, _, _, _, _} = OldTask,
          {reset_unread, _, _, _}) ->
    OldTask;
aggregate({set_inbox, _, _, Count, _, _, _, _},
          {set_inbox_incr_unread, Entry, Content, MsgId, Timestamp, Incrs, Box}) ->
    {set_inbox_incr_unread, Entry, Content, MsgId, Timestamp, Count + Incrs, Box};

%%% If the last task was a set_inbox_incr_unread
% we're resetting on this message:
aggregate({set_inbox_incr_unread, Entry, Content, MsgId, Timestamp, _, Box},
          {reset_unread, _, MsgId, _}) ->
    {set_inbox, Entry, Content, 0, MsgId, Timestamp, Box};
aggregate({set_inbox_incr_unread, _, _, _, _, _, _} = OldTask,
          {reset_unread, _, _, _}) ->
    OldTask;
% prefer newest row, but accumulate increment
aggregate({set_inbox_incr_unread, _, _, _, _, Incrs2, _},
          {set_inbox_incr_unread, Entry, Content, MsgId, Timestamp, Incrs1, Box}) ->
    {set_inbox_incr_unread, Entry, Content, MsgId, Timestamp, Incrs1 + Incrs2, Box};

aggregate({set_inbox_incr_unread, _, _, MsgId, _, _, _},
          {set_inbox, _, _, _, MsgId, _, _} = NewTask) ->
    NewTask;

aggregate(_OldTask, NewTask) ->
    NewTask.

1	-module(mod_inbox_rdbms_async).
2
3	-include("mod_inbox.hrl").
4	-include("mongoose_logger.hrl").
5
6	-behaviour(mod_inbox_backend).
7	-behaviour(mongoose_aggregator_worker).
8
9	-type box() :: binary().
10	-type task() ::
11	{set_inbox, mod_inbox:entry_key(), exml:element(), pos_integer(), id(), integer(), box()} \|
12	{set_inbox_incr_unread, mod_inbox:entry_key(), exml:element(), id(), integer(), pos_integer(), box()} \|
13	{remove_inbox_row, mod_inbox:entry_key()} \|
14	{reset_unread, mod_inbox:entry_key(), id(), integer()}.
15
16	%% API
17	-export([init/2,
18	set_inbox/6,
19	set_inbox_incr_unread/5,
20	reset_unread/4,
21	remove_inbox_row/2,
22	empty_user_bin/4,
23	empty_domain_bin/3,
24	empty_global_bin/2,
25	remove_domain/2,
26	clear_inbox/3,
27	get_inbox/4,
28	get_inbox_unread/2,
29	get_full_entry/2,
30	get_entry_properties/2,
31	set_entry_properties/3]).
32	-export([stop/1]).
33
34	%% Worker callbacks
35	-export([request/2, aggregate/3, verify/3]).
36
37	%% Initialisation
38	-spec init(mongooseim:host_type(), gen_mod:module_opts()) -> ok.
39	init(HostType, Opts) ->
40	AsyncOpts = prepare_pool_opts(Opts),	138✔
41	mod_inbox_rdbms:init(HostType, Opts),	138✔
42	prepare_deletes(HostType, Opts),	138✔
43	start_pool(HostType, AsyncOpts),	138✔
44	ok.	138✔
45
46	stop(HostType) ->
47	mongoose_async_pools:stop_pool(HostType, inbox).	138✔
48
49	prepare_pool_opts(#{async_writer := AsyncOpts}) ->
50	AsyncOpts#{pool_type => aggregate,	138✔
51	request_callback => fun ?MODULE:request/2,
52	aggregate_callback => fun ?MODULE:aggregate/3,
53	verify_callback => fun ?MODULE:verify/3}.
54
55	prepare_deletes(_HostType, _Opts) ->
56	mongoose_rdbms:prepare(inbox_move_conversation_to_bin, inbox,	138✔
57	[luser, lserver, remote_bare_jid],
58	<<"UPDATE inbox SET box='bin'",
59	" WHERE luser = ? AND lserver = ? AND remote_bare_jid = ?">>),
60	ok.	138✔
61
62	-spec start_pool(mongooseim:host_type(), mongoose_async_pools:pool_opts()) -> term().
63	start_pool(HostType, Opts) ->
64	mongoose_async_pools:start_pool(HostType, inbox, Opts).	138✔
65
66	%% Worker callbacks
67	-spec request(task(), mongoose_async_pools:pool_extra()) -> gen_server:request_id().
68	request(Task, _Extra = #{host_type := HostType}) ->
69	request_one(HostType, Task).	3,813✔
70
71	request_one(HostType, {set_inbox, {LUser, LServer, LToBareJid}, Packet, Count, MsgId, Timestamp, Box}) ->
72	Content = exml:to_binary(Packet),	908✔
73	Update = [MsgId, Box, Content, Count, Timestamp],	908✔
74	Insert = [LUser, LServer, LToBareJid, MsgId, Box, Content, Count, Timestamp],	908✔
75	rdbms_queries:request_upsert(HostType, inbox_upsert, Insert, Update);	908✔
76	request_one(HostType, {set_inbox_incr_unread, {LUser, LServer, LToBareJid}, Packet, MsgId, Timestamp, Incrs, Box}) ->
77	Content = exml:to_binary(Packet),	1,874✔
78	Update = [MsgId, Box, Content, Incrs, Timestamp],	1,874✔
79	Insert = [LUser, LServer, LToBareJid, MsgId, Box, Content, Incrs, Timestamp],	1,874✔
80	rdbms_queries:request_upsert(HostType, inbox_upsert_incr_unread, Insert, Update);	1,874✔
81	request_one(HostType, {reset_unread, {LUser, LServer, LToBareJid}, undefined, TS}) ->
82	mongoose_rdbms:execute_request(HostType, inbox_reset_unread, [LUser, LServer, LToBareJid, TS]);	18✔
83	request_one(HostType, {reset_unread, {LUser, LServer, LToBareJid}, MsgId, TS}) ->
84	mongoose_rdbms:execute_request(HostType, inbox_reset_unread_msg, [LUser, LServer, LToBareJid, MsgId, TS]);	688✔
85	request_one(HostType, {remove_inbox_row, {LUser, LServer, LToBareJid}}) ->
86	mongoose_rdbms:execute_request(HostType, inbox_move_conversation_to_bin, [LUser, LServer, LToBareJid]).	325✔
87
88	-spec aggregate(task(), task(), mongoose_async_pools:pool_extra()) -> {ok, task()}.
89	aggregate(Current, NewTask, _Extra) ->
90	{ok, aggregate(Current, NewTask)}.	129✔
91
92	-spec verify(term(), task(), mongoose_async_pools:pool_extra()) -> ok \| {error, term()}.
93	verify(Answer, InboxTask, _Extra) ->
94	case mod_inbox_rdbms:check_result(Answer) of	3,813✔
95	{error, Reason} ->
96	{LU, LS, LRem} = element(2, InboxTask),	×
97	?LOG_WARNING(#{what => inbox_process_message_failed, reason => Reason,	×
98	from_jid => jid:to_binary({LU, LS}), to_jid => LRem}),	×
99	{error, Reason};	×
100	_ -> ok	3,813✔
101	end.
102
103	%% async callbacks
104	-spec set_inbox(mongooseim:host_type(), mod_inbox:entry_key(),
105	exml:element(), Count :: integer(), id(), Timestamp :: integer()) ->
106	mod_inbox:write_res().
107	set_inbox(HostType, Entry, Packet, Count, MsgId, Timestamp) ->
108	Params = {set_inbox, Entry, Packet, Count, MsgId, Timestamp, <<"inbox">>},	834✔
109	mongoose_async_pools:put_task(HostType, inbox, Entry, Params).	834✔
110
111	-spec set_inbox_incr_unread(mongooseim:host_type(), mod_inbox:entry_key(),
112	exml:element(), MsgId :: binary(), Timestamp :: integer()) ->
113	mod_inbox:count_res().
114	set_inbox_incr_unread(HostType, Entry, Packet, MsgId, Timestamp) ->
115	Params = {set_inbox_incr_unread, Entry, Packet, MsgId, Timestamp, 1, <<"inbox">>},	1,974✔
116	mongoose_async_pools:put_task(HostType, inbox, Entry, Params).	1,974✔
117
118	-spec reset_unread(mongooseim:host_type(), mod_inbox:entry_key(), binary() \| undefined, integer()) ->
119	mod_inbox:write_res().
120	reset_unread(HostType, Entry, MsgId, TS) ->
121	Params = {reset_unread, Entry, MsgId, TS},	792✔
122	mongoose_async_pools:put_task(HostType, inbox, Entry, Params).	792✔
123
124	-spec remove_inbox_row(mongooseim:host_type(), mod_inbox:entry_key()) -> mod_inbox:write_res().
125	remove_inbox_row(HostType, Entry) ->
126	Params = {remove_inbox_row, Entry},	342✔
127	mongoose_async_pools:put_task(HostType, inbox, Entry, Params).	342✔
128
129	%% synchronous callbacks
130	-spec get_inbox(mongooseim:host_type(), jid:luser(), jid:lserver(), mod_inbox:get_inbox_params()) ->
131	[mod_inbox:inbox_res()].
132	get_inbox(HostType, LUser, LServer, Params) ->
133	mod_inbox_rdbms:get_inbox(HostType, LUser, LServer, Params).	4,390✔
134
135	-spec get_inbox_unread(mongooseim:host_type(), mod_inbox:entry_key()) ->
136	{ok, integer()}.
137	get_inbox_unread(HostType, Entry) ->
138	mod_inbox_rdbms:get_inbox_unread(HostType, Entry).	×
139
140	-spec remove_domain(mongooseim:host_type(), jid:lserver()) -> term().
141	remove_domain(HostType, LServer) ->
142	mod_inbox_rdbms:remove_domain(HostType, LServer).	×
143
144	-spec clear_inbox(mongooseim:host_type(), jid:luser(), jid:lserver()) ->
145	mod_inbox:write_res().
146	clear_inbox(HostType, LUser, LServer) ->
147	mod_inbox_rdbms:clear_inbox(HostType, LUser, LServer).	2,286✔
148
149	-spec get_full_entry(mongooseim:host_type(), mod_inbox:entry_key()) ->
150	inbox_res() \| nil().
151	get_full_entry(HostType, Entry) ->
152	mod_inbox_rdbms:get_full_entry(HostType, Entry).	6✔
153
154	-spec get_entry_properties(mongooseim:host_type(), mod_inbox:entry_key()) ->
155	entry_properties() \| nil().
156	get_entry_properties(HostType, Entry) ->
157	mod_inbox_rdbms:get_entry_properties(HostType, Entry).	2,814✔
158
159	-spec set_entry_properties(mongooseim:host_type(), mod_inbox:entry_key(), entry_properties()) ->
160	entry_properties() \| {error, binary()}.
161	set_entry_properties(HostType, Entry, Properties) ->
162	mod_inbox_rdbms:set_entry_properties(HostType, Entry, Properties).	306✔
163
164	-spec empty_user_bin(HostType :: mongooseim:host_type(),
165	LServer :: jid:lserver(),
166	LUser :: jid:luser(),
167	TS :: integer()) -> non_neg_integer().
168	empty_user_bin(HostType, LServer, LUser, TS) ->
169	mod_inbox_rdbms:empty_user_bin(HostType, LServer, LUser, TS).	42✔
170
171	-spec empty_domain_bin(HostType :: mongooseim:host_type(),
172	LServer :: jid:lserver(),
173	TS :: integer()) -> non_neg_integer().
174	empty_domain_bin(HostType, LServer, TS) ->
175	mod_inbox_rdbms:empty_domain_bin(HostType, LServer, TS).	42✔
176
177	-spec empty_global_bin(HostType :: mongooseim:host_type(),
178	TS :: integer()) -> non_neg_integer().
179	empty_global_bin(HostType, TS) ->
180	mod_inbox_rdbms:empty_global_bin(HostType, TS).	126✔
181
182	-spec aggregate(CurrentlyAccumulatedTask :: task(), NewTask :: task()) -> FinalTask :: task().
183	%%% if new task is remove_row, do the previous with an updated box
184	aggregate({set_inbox, Entry, Content, Count, MsgId, Timestamp, _},
185	{remove_inbox_row, _}) ->
186	{set_inbox, Entry, Content, Count, MsgId, Timestamp, <<"bin">>};	×
187	aggregate({set_inbox_incr_unread, Entry, Content, MsgId, Timestamp, Incrs, _},
188	{remove_inbox_row, _}) ->
189	{set_inbox_incr_unread, Entry, Content, MsgId, Timestamp, Incrs, <<"bin">>};	14✔
190	aggregate(_, {remove_inbox_row, _} = NewTask) ->
191	NewTask;	×
192
193	%%% if the last task was remove_row, this task should now only be an insert
194	aggregate({remove_inbox_row, _} = OldTask, {reset_unread, _, _, _}) ->
195	OldTask;	×
196	aggregate({remove_inbox_row, _},
197	{set_inbox, Entry, Content, Count, MsgId, Timestamp, _}) ->
198	{set_inbox, Entry, Content, Count, MsgId, Timestamp, <<"bin">>};	×
199	aggregate({remove_inbox_row, _},
200	{set_inbox_incr_unread, Entry, Content, MsgId, Timestamp, Incrs, _}) ->
201	{set_inbox_incr_unread, Entry, Content, MsgId, Timestamp, Incrs, <<"bin">>};	3✔
202
203	%%% If the last task was a reset_unread,
204	% we prefer explicit resets,
205	% then adhoc newer resets,
206	% then we accumulate inserts
207	%% an undefined means an explicit request to reset, it has priority
208	aggregate({reset_unread, _, _, _}, {reset_unread, _, undefined, _} = NewTask) ->
209	NewTask;	×
210	%% an undefined means an explicit request to reset, it has priority
211	aggregate({reset_unread, _, undefined, _} = OldTask, {reset_unread, _, _, _}) ->
212	OldTask;	×
213	%% both are adhoc, we prefer the newer
214	aggregate({reset_unread, _, _, _}, {reset_unread, _, _, _} = NewTask) ->
215	NewTask;	×
216	aggregate({reset_unread, _, _, _}, {set_inbox, _, _, _, _, _, _} = NewTask) ->
217	NewTask;	×
218	%% Here `Count` becomes an absolute value instead of an increment
219	aggregate({reset_unread, _, _, _},
220	{set_inbox_incr_unread, Entry, Content, MsgId, Timestamp, Incrs, Box}) ->
221	{set_inbox, Entry, Content, Incrs, MsgId, Timestamp, Box};	10✔
222
223	%%% If the last task was a set_inbox
224	%% Reset is an explicit reset-to-zero, so do reset the counter
225	aggregate({set_inbox, Entry, Content, _, MsgId, Timestamp, Box},
226	{reset_unread, _, undefined, _}) ->
227	{set_inbox, Entry, Content, 0, MsgId, Timestamp, Box};	×
228	%% Reset refers to that same set_inbox
229	aggregate({set_inbox, Entry, Content, _, MsgId, Timestamp, Box},
230	{reset_unread, _, MsgId, _}) ->
UNCOV 231	{set_inbox, Entry, Content, 0, MsgId, Timestamp, Box};	×
232	%% Reset refers to some other set_inbox
233	aggregate({set_inbox, _, _, _, _, _, _} = OldTask,
234	{reset_unread, _, _, _}) ->
235	OldTask;	×
236	aggregate({set_inbox, _, _, Count, _, _, _, _},
237	{set_inbox_incr_unread, Entry, Content, MsgId, Timestamp, Incrs, Box}) ->
238	{set_inbox_incr_unread, Entry, Content, MsgId, Timestamp, Count + Incrs, Box};	×
239
240	%%% If the last task was a set_inbox_incr_unread
241	% we're resetting on this message:
242	aggregate({set_inbox_incr_unread, Entry, Content, MsgId, Timestamp, _, Box},
243	{reset_unread, _, MsgId, _}) ->
244	{set_inbox, Entry, Content, 0, MsgId, Timestamp, Box};	76✔
245	aggregate({set_inbox_incr_unread, _, _, _, _, _, _} = OldTask,
246	{reset_unread, _, _, _}) ->
247	OldTask;	×
248	% prefer newest row, but accumulate increment
249	aggregate({set_inbox_incr_unread, _, _, _, _, Incrs2, _},
250	{set_inbox_incr_unread, Entry, Content, MsgId, Timestamp, Incrs1, Box}) ->
251	{set_inbox_incr_unread, Entry, Content, MsgId, Timestamp, Incrs1 + Incrs2, Box};	12✔
252
253	aggregate({set_inbox_incr_unread, _, _, MsgId, _, _, _},
254	{set_inbox, _, _, _, MsgId, _, _} = NewTask) ->
255	NewTask;	×
256
257	aggregate(_OldTask, NewTask) ->
258	NewTask.	14✔

esl / MongooseIM / 23044468042

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous