MucSub: push notifications do not work with user_mucsub_from_muc_archive: true

[marc0s]

We are using muc_sub and have enabled user_mucsub_from_muc_archive: true to avoid rapidly increasing the archive table size. What we noticed is that push notifications for those subscribed to the MUC stopped to work. They worked via the mam_archived hook of the mucsub message. Now, as those mucsub messages are not stored in the archive, pushes are gone. Is this the expected behaviour or a bug? We came up with a patch that adds a new offline_message/1 implementation that matches those with mam_archived := true, in_muc_mam := true that calls notify/5.  In case it is a bug, does relying on offline seems like the suitable approach or is there any other that fits better? Thanks!

[prefiks]

We don't call store_mam_message is this case because last handler for it do the actual storing in db (and we don't want to do it in this case). But we can add extra parameter to this callback that can trigger skipping store, but notify other listeners, and since mod_push seems to be only other consumer of this, doing this shouldn't too disruptive.

[prefiks]

This should be implemented by 9fd1ed3

[marc0s]

Thanks! Will report back when I give it a try

[prefiks]

Try one commit later (current master), as.i missed one place where this hook was called

[marc0s]

Confirmed that it works 🙂

[marc0s]

Sorry to bring this up again, but we may have hit a similar case were messages with a no-store hint are not push-notified (mediated MUC invites in this case, probably happening with other messages?). Following your line of thought from your commits fixing the above mucsub issue, I have come up with this diff that seems to work. What are your thoughts about it, @prefiks? Thanks in advance!

diff --git a/src/mod_mam.erl b/src/mod_mam.erl
index b79966cb3..4950ca797 100644
--- a/src/mod_mam.erl
+++ b/src/mod_mam.erl
@@ -426,14 +426,17 @@ user_receive_packet({#message{from = Peer} = Pkt, #{jid := JID} = C2SState}) ->
     LUser = JID#jid.luser,
     LServer = JID#jid.lserver,
     Pkt1 = case should_archive(Pkt, LServer) of
-              true ->
-                  case store_msg(Pkt, LUser, LServer, Peer, recv) of
+              store ->
+                  case store_msg(Pkt, LUser, LServer, Peer, recv, false) of
                       ok ->
                           mark_stored_msg(Pkt, JID);
                       _ ->
                           Pkt
                   end;
-              _ ->
+              notify_only ->
+                  _ = store_msg(Pkt, LUser, LServer, Peer, recv, true),
+                  Pkt;
+              skip ->
                   Pkt
           end,
     {Pkt1, C2SState};
@@ -447,15 +450,19 @@ user_send_packet({#message{to = Peer} = Pkt, #{jid := JID} = C2SState}) ->
     LServer = JID#jid.lserver,
     Pkt1 = init_stanza_id(Pkt, LServer),
     Pkt2 = case should_archive(Pkt1, LServer) of
-              true ->
+              store ->
                   case store_msg(xmpp:set_from_to(Pkt1, JID, Peer),
-                                 LUser, LServer, Peer, send) of
+                                 LUser, LServer, Peer, send, false) of
                       ok ->
                           mark_stored_msg(Pkt1, JID);
                       _ ->
                           Pkt1
                   end;
-              false ->
+              notify_only ->
+                  _ = store_msg(xmpp:set_from_to(Pkt1, JID, Peer),
+                                LUser, LServer, Peer, send, true),
+                  Pkt1;
+              skip ->
                   Pkt1
           end,
     {Pkt2, C2SState};
@@ -477,14 +484,17 @@ offline_message({_Action, #message{from = Peer, to = To} = Pkt} = Acc) ->
     LUser = To#jid.luser,
     LServer = To#jid.lserver,
     case should_archive(Pkt, LServer) of
-       true ->
-          case store_msg(Pkt, LUser, LServer, Peer, recv) of
+       store ->
+          case store_msg(Pkt, LUser, LServer, Peer, recv, false) of
               ok ->
                   {archived, mark_stored_msg(Pkt, To)};
               _ ->
                   Acc
           end;
-       false ->
+       notify_only ->
+          _ = store_msg(Pkt, LUser, LServer, Peer, recv, true),
+          Acc;
+       skip ->
           Acc
     end.
 
@@ -917,38 +927,41 @@ process_iq(LServer, #iq{from = #jid{luser = LUser}, lang = Lang,
            xmpp:make_error(IQ, xmpp:err_internal_server_error(Txt, Lang))
     end.
 
--spec should_archive(message(), binary()) -> boolean().
+-spec should_archive(message(), binary()) -> store | notify_only | skip.
 should_archive(#message{type = error}, _LServer) ->
-    false;
+    skip;
 should_archive(#message{type = groupchat, meta = #{is_muc_subscriber := true}} = Msg, LServer) ->
        case mod_mam_opt:archive_muc_as_mucsub(LServer) of
                true ->
                    should_archive(Msg#message{type = chat}, LServer);
                false ->
-                       false
+                       skip
        end;
 should_archive(#message{type = groupchat}, _LServer) ->
-    false;
+    skip;
 should_archive(#message{meta = #{from_offline := true}}, _LServer) ->
-    false;
+    skip;
 should_archive(#message{body = Body, subject = Subject,
                        type = Type} = Pkt, LServer) ->
     case is_archived(Pkt, LServer) of
        true ->
-           false;
+           skip;
        false ->
            case check_store_hint(Pkt) of
                store ->
-                   true;
+                   store;
                no_store ->
-                   false;
+                   % Not persisted, but store_mam_message is still invoked
+                   % (with a skip-persist flag) so listeners like mod_push
+                   % can still react, e.g. to notify about MUC invites.
+                   notify_only;
                none when Type == headline ->
-                   false;
+                   skip;
                none ->
                    case xmpp:get_text(Body) /= <<>> orelse
                         xmpp:get_text(Subject) /= <<>> of
                        true ->
-                           true;
+                           store;
                        _ ->
                            case misc:unwrap_mucsub_message(Pkt) of
                                #message{type = groupchat} = Msg ->
@@ -956,13 +969,16 @@ should_archive(#message{body = Body, subject = Subject,
                                #message{} = Msg ->
                                    should_archive(Msg, LServer);
                                _ ->
-                                   misc:is_mucsub_message(Pkt)
+                                   case misc:is_mucsub_message(Pkt) of
+                                       true -> store;
+                                       false -> skip
+                                   end
                            end
                    end
            end
     end;
 should_archive(_, _LServer) ->
-    false.
+    skip.
 
 -spec strip_my_stanza_id(stanza(), binary()) -> stanza().
 strip_my_stanza_id(Pkt, LServer) ->
@@ -1112,9 +1128,9 @@ may_enter_room(From,
 may_enter_room(From, MUCState) ->
     mod_muc_room:is_occupant_or_admin(From, MUCState).
 
--spec store_msg(message(), binary(), binary(), jid(), send | recv)
+-spec store_msg(message(), binary(), binary(), jid(), send | recv, boolean())
       -> ok | pass | any().
-store_msg(#message{type = groupchat, from = From, to = To, meta = #{is_muc_subscriber := true}} = Pkt, LUser, LServer, _Peer, Dir) ->
+store_msg(#message{type = groupchat, from = From, to = To, meta = #{is_muc_subscriber := true}} = Pkt, LUser, LServer, _Peer, Dir, SkipStore) ->
        BarePeer = jid:remove_resource(From),
        StanzaId = xmpp:get_subtag(Pkt, #stanza_id{by = #jid{}}),
     Id = case StanzaId of
@@ -1138,18 +1154,19 @@ store_msg(#message{type = groupchat, from = From, to = To, meta = #{is_muc_subsc
                          }]
             },
        Pkt3 = xmpp:put_meta(Pkt2, stanza_id, binary_to_integer(Id)),
-       store_msg(Pkt3, LUser, LServer, BarePeer, Dir);
-store_msg(Pkt, LUser, LServer, Peer, Dir) ->
+       store_msg(Pkt3, LUser, LServer, BarePeer, Dir, SkipStore);
+store_msg(Pkt, LUser, LServer, Peer, Dir, SkipStore) ->
     case get_prefs(LUser, LServer) of
        {ok, Prefs} ->
            UseMucArchive = mod_mam_opt:user_mucsub_from_muc_archive(LServer),
            StoredInMucMam = UseMucArchive andalso xmpp:get_meta(Pkt, in_muc_mam, false),
+           NoStore = SkipStore orelse StoredInMucMam,
            case {should_archive_peer(LUser, LServer, Prefs, Peer), Pkt} of
                {true, #message{meta = #{sm_copy := true}}} ->
                    ok; % Already stored.
                {true, _} ->
                    case ejabberd_hooks:run_fold(store_mam_message, LServer, Pkt,
-                                                [LUser, LServer, Peer, <<"">>, chat, Dir, StoredInMucMam]) of
+                                                [LUser, LServer, Peer, <<"">>, chat, Dir, NoStore]) of
                        #message{} -> ok;
                        _ -> pass
                    end;

[marc0s]

Got any chance to take a look at it, @prefiks? Thanks!

[badlop]

You may want to publish your diff as a PR, so that it gets its own issue number, comment section, people can pull it and test easily...