Add stanza-id to every archived message (#1477)

src/mod_mam.erl

@@ -750,14 +750,19 @@ select(_LServer, JidRequestor, JidArchive, Query, RSM,
 		  case match_interval(Now, Start, End) and
 		      match_rsm(Now, RSM) of
 		      true ->
-			  [{integer_to_binary(TS), TS,
+			  case msg_to_el(#archive_msg{
-			    msg_to_el(#archive_msg{
+					    id = integer_to_binary(TS),
 					    type = groupchat,
 					    timestamp = Now,
 					    peer = undefined,
 					    nick = Nick,
 					    packet = Pkt},
-				      MsgType, JidRequestor, JidArchive)}];
+					 MsgType, JidRequestor, JidArchive) of
+			      {ok, Msg} ->
+				  [{integer_to_binary(TS), TS, Msg}];
+			      {error, _} ->
+				  []
+			  end;
 		      false ->
 			  []
 		  end
@@ -776,25 +781,23 @@ select(LServer, JidRequestor, JidArchive, Query, RSM, MsgType) ->
     Mod = gen_mod:db_mod(LServer, ?MODULE),
     Mod:select(LServer, JidRequestor, JidArchive, Query, RSM, MsgType).
 
-msg_to_el(#archive_msg{timestamp = TS, packet = Pkt1, nick = Nick, peer = Peer},
+msg_to_el(#archive_msg{timestamp = TS, packet = El, nick = Nick,
+		       peer = Peer, id = ID},
 	  MsgType, JidRequestor, #jid{lserver = LServer} = JidArchive) ->
-    Pkt2 = maybe_update_from_to(Pkt1, JidRequestor, JidArchive, Peer, MsgType,
+    try xmpp:decode(El, ?NS_CLIENT, [ignore_els]) of
-				Nick),
+	Pkt1 ->
-    El = case Pkt2 of
+	    Pkt2 = set_stanza_id(Pkt1, JidArchive, ID),
-	     #xmlel{attrs = Attrs} ->
+	    Pkt3 = maybe_update_from_to(
-		 Attrs1 = lists:keystore(<<"xmlns">>, 1, Attrs,
+		     Pkt2, JidRequestor, JidArchive, Peer, MsgType, Nick),
-					 {<<"xmlns">>, ?NS_CLIENT}),
+	    Delay = #delay{stamp = TS, from = jid:make(LServer)},
-		 Pkt2#xmlel{attrs = Attrs1};
+	    {ok, #forwarded{xml_els = [xmpp:encode(Pkt3)], delay = Delay}}
-	     _ ->
+    catch _:{xmpp_codec, Why} ->
-		 xmpp:encode(Pkt2)
+	    ?ERROR_MSG("Failed to decode raw element ~p from message "
-	 end,
+		       "archive of user ~s: ~s",
-    #forwarded{xml_els = [El],
+		       [El, jid:to_string(JidArchive), xmpp:format_error(Why)]),
-	       delay = #delay{stamp = TS, from = jid:make(LServer)}}.
+	    {error, invalid_xml}
+    end.
 
-maybe_update_from_to(#xmlel{} = El, JidRequestor, JidArchive, Peer,
-		     {groupchat, _, _} = MsgType, Nick) ->
-    Pkt = xmpp:decode(El, ?NS_CLIENT, [ignore_els]),
-    maybe_update_from_to(Pkt, JidRequestor, JidArchive, Peer, MsgType, Nick);
 maybe_update_from_to(#message{sub_els = Els} = Pkt, JidRequestor, JidArchive,
 		     Peer, {groupchat, Role,
 			    #state{config = #config{anonymous = Anon}}},

src/mod_mam_mnesia.erl

@@ -160,12 +160,17 @@ select(_LServer, JidRequestor,
     SortedMsgs = lists:keysort(#archive_msg.timestamp, Msgs),
     {FilteredMsgs, IsComplete} = filter_by_rsm(SortedMsgs, RSM),
     Count = length(Msgs),
-    Result = {lists:map(
+    Result = {lists:flatmap(
 		fun(Msg) ->
-			{Msg#archive_msg.id,
+			case mod_mam:msg_to_el(
+			       Msg, MsgType, JidRequestor, JidArchive) of
+			    {ok, El} ->
+				[{Msg#archive_msg.id,
 				  binary_to_integer(Msg#archive_msg.id),
-			 mod_mam:msg_to_el(Msg, MsgType, JidRequestor,
+				  El}];
-					   JidArchive)}
+			    {error, _} ->
+				[]
+			end
 		end, FilteredMsgs), IsComplete, Count},
     erlang:garbage_collect(),
     Result.

src/mod_mam_sql.erl

@@ -168,28 +168,12 @@ select(LServer, JidRequestor, #jid{luser = LUser} = JidArchive,
 		end,
 	    {lists:flatmap(
 	       fun([TS, XML, PeerBin, Kind, Nick]) ->
-		       try
+		       case make_archive_el(
-			   #xmlel{} = El = fxml_stream:parse_element(XML),
+			      TS, XML, PeerBin, Kind, Nick,
-			   Now = usec_to_now(binary_to_integer(TS)),
+			      MsgType, JidRequestor, JidArchive) of
-			   PeerJid = jid:tolower(jid:from_string(PeerBin)),
+			   {ok, El} ->
-			   T = case Kind of
+			       [{TS, binary_to_integer(TS), El}];
-				   <<"">> -> chat;
+			   {error, _} ->
-				   null -> chat;
-				   _ -> jlib:binary_to_atom(Kind)
-			       end,
-			   [{TS, binary_to_integer(TS),
-			     mod_mam:msg_to_el(#archive_msg{timestamp = Now,
-							    packet = El,
-							    type = T,
-							    nick = Nick,
-							    peer = PeerJid},
-					       MsgType, JidRequestor, JidArchive)}]
-		       catch _:Err ->
-			       ?ERROR_MSG("failed to parse data from SQL: ~p. "
-					  "The data was: "
-					  "timestamp = ~s, xml = ~s, "
-					  "peer = ~s, kind = ~s, nick = ~s",
-					  [Err, TS, XML, PeerBin, Kind, Nick]),
 			       []
 		       end
 	       end, Res1), IsComplete, binary_to_integer(Count)};
@@ -319,3 +303,51 @@ get_max_direction_id(RSM) ->
 	_ ->
 	    {undefined, undefined, <<>>}
     end.
+
+-spec make_archive_el(binary(), binary(), binary(), binary(),
+		      binary(), _, jid(), jid()) ->
+			     {ok, xmpp_element()} | {error, invalid_jid |
+						     invalid_timestamp |
+						     invalid_xml}.
+make_archive_el(TS, XML, Peer, Kind, Nick, MsgType, JidRequestor, JidArchive) ->
+    case fxml_stream:parse_element(XML) of
+	#xmlel{} = El ->
+	    try binary_to_integer(TS) of
+		TSInt ->
+		    case jid:from_string(Peer) of
+			#jid{} = PeerJID ->
+			    Now = usec_to_now(TSInt),
+			    PeerLJID = jid:tolower(PeerJID),
+			    T = case Kind of
+				    <<"">> -> chat;
+				    null -> chat;
+				    _ -> jlib:binary_to_atom(Kind)
+				end,
+			    mod_mam:msg_to_el(
+			      #archive_msg{timestamp = Now,
+					   id = TS,
+					   packet = El,
+					   type = T,
+					   nick = Nick,
+					   peer = PeerLJID},
+			      MsgType, JidRequestor, JidArchive);
+			error ->
+			    ?ERROR_MSG("Malformed 'peer' field with value "
+				       "'~s' detected for user ~s in table "
+				       "'archive': invalid JID",
+				       [Peer, jid:to_string(JidArchive)]),
+			    {error, invalid_jid}
+		    end
+	    catch _:_ ->
+		    ?ERROR_MSG("Malformed 'timestamp' field with value '~s' "
+			       "detected for user ~s in table 'archive': "
+			       "not an integer",
+			       [TS, jid:to_string(JidArchive)]),
+		    {error, invalid_timestamp}
+	    end;
+	{error, {_, Reason}} ->
+	    ?ERROR_MSG("Malformed 'xml' field with value '~s' detected "
+		       "for user ~s in table 'archive': ~s",
+		       [XML, jid:to_string(JidArchive), Reason]),
+	    {error, invalid_xml}
+    end.