processone / ejabberd / 511

Build # 511

Build Type

push

github

Committed by Badlop

Commit Message

Set version to 23.04

Run Details

13516 of 40931 relevant lines covered (33.02%)

672.99 hits per line

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

52.84

/src/mod_mam_sql.erl

%%%-------------------------------------------------------------------
%%% File    : mod_mam_sql.erl
%%% Author  : Evgeny Khramtsov <ekhramtsov@process-one.net>
%%% Created : 15 Apr 2016 by Evgeny Khramtsov <ekhramtsov@process-one.net>
%%%
%%%
%%% ejabberd, Copyright (C) 2002-2023   ProcessOne
%%%
%%% This program is free software; you can redistribute it and/or
%%% modify it under the terms of the GNU General Public License as
%%% published by the Free Software Foundation; either version 2 of the
%%% License, or (at your option) any later version.
%%%
%%% This program is distributed in the hope that it will be useful,
%%% but WITHOUT ANY WARRANTY; without even the implied warranty of
%%% MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
%%% General Public License for more details.
%%%
%%% You should have received a copy of the GNU General Public License along
%%% with this program; if not, write to the Free Software Foundation, Inc.,
%%% 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.
%%%
%%%----------------------------------------------------------------------

-module(mod_mam_sql).


-behaviour(mod_mam).

%% API
-export([init/2, remove_user/2, remove_room/3, delete_old_messages/3,
         extended_fields/0, store/8, write_prefs/4, get_prefs/2, select/7, export/1, remove_from_archive/3,
         is_empty_for_user/2, is_empty_for_room/3, select_with_mucsub/6,
         delete_old_messages_batch/4, count_messages_to_delete/3]).

-include_lib("stdlib/include/ms_transform.hrl").
-include_lib("xmpp/include/xmpp.hrl").
-include("mod_mam.hrl").
-include("logger.hrl").
-include("ejabberd_sql_pt.hrl").
-include("mod_muc_room.hrl").

%%%===================================================================
%%% API
%%%===================================================================
init(_Host, _Opts) ->
    ok.

remove_user(LUser, LServer) ->
    ejabberd_sql:sql_query(
      LServer,
      ?SQL("delete from archive where username=%(LUser)s and %(LServer)H")),
    ejabberd_sql:sql_query(
      LServer,
      ?SQL("delete from archive_prefs where username=%(LUser)s and %(LServer)H")).

remove_room(LServer, LName, LHost) ->
    LUser = jid:encode({LName, LHost, <<>>}),
    remove_user(LUser, LServer).

remove_from_archive(LUser, LServer, none) ->
    case ejabberd_sql:sql_query(LServer,
                                ?SQL("delete from archive where username=%(LUser)s and %(LServer)H")) of
        {error, Reason} -> {error, Reason};
        _ -> ok
    end;
remove_from_archive(LUser, LServer, #jid{} = WithJid) ->
    Peer = jid:encode(jid:remove_resource(WithJid)),
    case ejabberd_sql:sql_query(LServer,
                                ?SQL("delete from archive where username=%(LUser)s and %(LServer)H and bare_peer=%(Peer)s")) of
        {error, Reason} -> {error, Reason};
        _ -> ok
    end;
remove_from_archive(LUser, LServer, StanzaId) ->
    case ejabberd_sql:sql_query(LServer,
                                ?SQL("delete from archive where username=%(LUser)s and %(LServer)H and timestamp=%(StanzaId)d")) of
        {error, Reason} -> {error, Reason};
        _ -> ok
    end.

count_messages_to_delete(ServerHost, TimeStamp, Type) ->
    TS = misc:now_to_usec(TimeStamp),
    Res =
    case Type of
        all ->
            ejabberd_sql:sql_query(
                ServerHost,
                ?SQL("select count(*) from archive"
                     " where timestamp < %(TS)d and %(ServerHost)H"));
        _ ->
            SType = misc:atom_to_binary(Type),
            ejabberd_sql:sql_query(
                ServerHost,
                ?SQL("select @(count(*))d from archive"
                     " where timestamp < %(TS)d"
                     " and kind=%(SType)s"
                     " and %(ServerHost)H"))
    end,
    case Res of
        {selected, [Count]} ->
            {ok, Count};
        _ ->
            error
    end.

delete_old_messages_batch(ServerHost, TimeStamp, Type, Batch) ->
    TS = misc:now_to_usec(TimeStamp),
    Res =
    case Type of
        all ->
            ejabberd_sql:sql_query(
                ServerHost,
                ?SQL("delete from archive"
                     " where timestamp < %(TS)d and %(ServerHost)H limit %(Batch)d"));
        _ ->
            SType = misc:atom_to_binary(Type),
            ejabberd_sql:sql_query(
                ServerHost,
                ?SQL("delete from archive"
                     " where timestamp < %(TS)d"
                     " and kind=%(SType)s"
                     " and %(ServerHost)H limit %(Batch)d"))
    end,
    case Res of
        {updated, Count} ->
            {ok, Count};
        {error, _} = Error ->
            Error
    end.

delete_old_messages(ServerHost, TimeStamp, Type) ->
    TS = misc:now_to_usec(TimeStamp),
    case Type of
        all ->
            ejabberd_sql:sql_query(
              ServerHost,
              ?SQL("delete from archive"
                   " where timestamp < %(TS)d and %(ServerHost)H"));
        _ ->
            SType = misc:atom_to_binary(Type),
            ejabberd_sql:sql_query(
              ServerHost,
              ?SQL("delete from archive"
                   " where timestamp < %(TS)d"
                   " and kind=%(SType)s"
                   " and %(ServerHost)H"))
    end,
    ok.

extended_fields() ->
    [{withtext, <<"">>}].

store(Pkt, LServer, {LUser, LHost}, Type, Peer, Nick, _Dir, TS) ->
    SUser = case Type of
                chat -> LUser;
                groupchat -> jid:encode({LUser, LHost, <<>>})
            end,
    BarePeer = jid:encode(
                 jid:tolower(
                   jid:remove_resource(Peer))),
    LPeer = jid:encode(
              jid:tolower(Peer)),
    Body = fxml:get_subtag_cdata(Pkt, <<"body">>),
    SType = misc:atom_to_binary(Type),
    SqlType = ejabberd_option:sql_type(LServer),
    XML = case mod_mam_opt:compress_xml(LServer) of
              true ->
                  J1 = case Type of
                              chat -> jid:encode({LUser, LHost, <<>>});
                              groupchat -> SUser
                          end,
                  xml_compress:encode(Pkt, J1, LPeer);
              _ ->
                  fxml:element_to_binary(Pkt)
          end,
        case SqlType of
          mssql -> case ejabberd_sql:sql_query(
                   LServer,
                   ?SQL_INSERT(
                      "archive",
                      ["username=%(SUser)s",
                       "server_host=%(LServer)s",
                       "timestamp=%(TS)d",
                       "peer=%(LPeer)s",
                       "bare_peer=%(BarePeer)s",
                       "xml=N%(XML)s",
                       "txt=N%(Body)s",
                       "kind=%(SType)s",
                       "nick=%(Nick)s"])) of
                {updated, _} ->
                    ok;
                Err ->
                    Err
            end;
            _ -> case ejabberd_sql:sql_query(
                   LServer,
                   ?SQL_INSERT(
                      "archive",
                      ["username=%(SUser)s",
                       "server_host=%(LServer)s",
                       "timestamp=%(TS)d",
                       "peer=%(LPeer)s",
                       "bare_peer=%(BarePeer)s",
                       "xml=%(XML)s",
                       "txt=%(Body)s",
                       "kind=%(SType)s",
                       "nick=%(Nick)s"])) of
                {updated, _} ->
                    ok;
                Err ->
                    Err
            end
        end.

write_prefs(LUser, _LServer, #archive_prefs{default = Default,
                                           never = Never,
                                           always = Always},
            ServerHost) ->
    SDefault = erlang:atom_to_binary(Default, utf8),
    SAlways = misc:term_to_expr(Always),
    SNever = misc:term_to_expr(Never),
    case ?SQL_UPSERT(
            ServerHost,
            "archive_prefs",
            ["!username=%(LUser)s",
             "!server_host=%(ServerHost)s",
             "def=%(SDefault)s",
             "always=%(SAlways)s",
             "never=%(SNever)s"]) of
        ok ->
            ok;
        Err ->
            Err
    end.

get_prefs(LUser, LServer) ->
    case ejabberd_sql:sql_query(
           LServer,
           ?SQL("select @(def)s, @(always)s, @(never)s from archive_prefs"
                " where username=%(LUser)s and %(LServer)H")) of
        {selected, [{SDefault, SAlways, SNever}]} ->
            Default = erlang:binary_to_existing_atom(SDefault, utf8),
            Always = ejabberd_sql:decode_term(SAlways),
            Never = ejabberd_sql:decode_term(SNever),
            {ok, #archive_prefs{us = {LUser, LServer},
                    default = Default,
                    always = Always,
                    never = Never}};
        _ ->
            error
    end.

select(LServer, JidRequestor, #jid{luser = LUser} = JidArchive,
       MAMQuery, RSM, MsgType, Flags) ->
    User = case MsgType of
               chat -> LUser;
               _ -> jid:encode(JidArchive)
           end,
    {Query, CountQuery} = make_sql_query(User, LServer, MAMQuery, RSM, none),
    do_select_query(LServer, JidRequestor, JidArchive, RSM, MsgType, Query, CountQuery, Flags).

-spec select_with_mucsub(binary(), jid(), jid(), mam_query:result(),
                             #rsm_set{} | undefined, all | only_count | only_messages) ->
                                {[{binary(), non_neg_integer(), xmlel()}], boolean(), non_neg_integer()} |
                                {error, db_failure}.
select_with_mucsub(LServer, JidRequestor, #jid{luser = LUser} = JidArchive,
                   MAMQuery, RSM, Flags) ->
    Extra = case gen_mod:db_mod(LServer, mod_muc) of
                mod_muc_sql ->
                    subscribers_table;
                _ ->
                    SubRooms = case mod_muc_admin:find_hosts(LServer) of
                                   [First|_] ->
                                       case mod_muc:get_subscribed_rooms(First, JidRequestor) of
                                           {ok, L} -> L;
                                           {error, _} -> []
                                       end;
                                   _ ->
                                       []
                               end,
                    [jid:encode(Jid) || {Jid, _, _} <- SubRooms]
            end,
    {Query, CountQuery} = make_sql_query(LUser, LServer, MAMQuery, RSM, Extra),
    do_select_query(LServer, JidRequestor, JidArchive, RSM, chat, Query, CountQuery, Flags).

do_select_query(LServer, JidRequestor, #jid{luser = LUser} = JidArchive, RSM,
                MsgType, Query, CountQuery, Flags) ->
    % TODO from XEP-0313 v0.2: "To conserve resources, a server MAY place a
    % reasonable limit on how many stanzas may be pushed to a client in one
    % request. If a query returns a number of stanzas greater than this limit
    % and the client did not specify a limit using RSM then the server should
    % return a policy-violation error to the client." We currently don't do this
    % for v0.2 requests, but we do limit #rsm_in.max for v0.3 and newer.
    QRes = case Flags of
                   all ->
                       {ejabberd_sql:sql_query(LServer, Query), ejabberd_sql:sql_query(LServer, CountQuery)};
                   only_messages ->
                       {ejabberd_sql:sql_query(LServer, Query), {selected, ok, [[<<"0">>]]}};
                   only_count ->
                       {{selected, ok, []}, ejabberd_sql:sql_query(LServer, CountQuery)}
               end,
    case QRes of
        {{selected, _, Res}, {selected, _, [[Count]]}} ->
            {Max, Direction, _} = get_max_direction_id(RSM),
            {Res1, IsComplete} =
            if Max >= 0 andalso Max /= undefined andalso length(Res) > Max ->
                if Direction == before ->
                    {lists:nthtail(1, Res), false};
                    true ->
                        {lists:sublist(Res, Max), false}
                end;
                true ->
                    {Res, true}
            end,
            MucState = #state{config = #config{anonymous = true}},
            JidArchiveS = jid:encode(jid:remove_resource(JidArchive)),
            {lists:flatmap(
                fun([TS, XML, PeerBin, Kind, Nick]) ->
                    case make_archive_el(JidArchiveS, TS, XML, PeerBin, Kind, Nick,
                                         MsgType, JidRequestor, JidArchive) of
                        {ok, El} ->
                            [{TS, binary_to_integer(TS), El}];
                        {error, _} ->
                            []
                    end;
                   ([User, TS, XML, PeerBin, Kind, Nick]) when User == LUser ->
                       case make_archive_el(JidArchiveS, TS, XML, PeerBin, Kind, Nick,
                                            MsgType, JidRequestor, JidArchive) of
                           {ok, El} ->
                               [{TS, binary_to_integer(TS), El}];
                           {error, _} ->
                               []
                       end;
                   ([User, TS, XML, PeerBin, Kind, Nick]) ->
                       case make_archive_el(User, TS, XML, PeerBin, Kind, Nick,
                                            {groupchat, member, MucState}, JidRequestor,
                                            jid:decode(User)) of
                           {ok, El} ->
                               mod_mam:wrap_as_mucsub([{TS, binary_to_integer(TS), El}],
                                                      JidRequestor);
                           {error, _} ->
                               []
                       end
                end, Res1), IsComplete, binary_to_integer(Count)};
        _ ->
            {[], false, 0}
    end.

export(_Server) ->
    [{archive_prefs,
      fun(Host, #archive_prefs{us =
                {LUser, LServer},
                default = Default,
                always = Always,
                never = Never})
          when LServer == Host ->
                SDefault = erlang:atom_to_binary(Default, utf8),
                SAlways = misc:term_to_expr(Always),
                SNever = misc:term_to_expr(Never),
                [?SQL_INSERT(
                    "archive_prefs",
                    ["username=%(LUser)s",
                     "server_host=%(LServer)s",
                     "def=%(SDefault)s",
                     "always=%(SAlways)s",
                     "never=%(SNever)s"])];
          (_Host, _R) ->
              []
      end},
     {archive_msg,
      fun([Host | HostTail], #archive_msg{us ={LUser, LServer},
                id = _ID, timestamp = TS, peer = Peer,
                type = Type, nick = Nick, packet = Pkt})
          when (LServer == Host) or ([LServer] == HostTail)  ->
                TStmp = misc:now_to_usec(TS),
                SUser = case Type of
                      chat -> LUser;
                      groupchat -> jid:encode({LUser, LServer, <<>>})
                    end,
                BarePeer = jid:encode(jid:tolower(jid:remove_resource(Peer))),
                LPeer = jid:encode(jid:tolower(Peer)),
                XML = fxml:element_to_binary(Pkt),
                Body = fxml:get_subtag_cdata(Pkt, <<"body">>),
                SType = misc:atom_to_binary(Type),
                SqlType = ejabberd_option:sql_type(Host),
                case SqlType of
                        mssql -> [?SQL_INSERT(
                            "archive",
                            ["username=%(SUser)s",
                             "server_host=%(LServer)s",
                             "timestamp=%(TStmp)d",
                             "peer=%(LPeer)s",
                             "bare_peer=%(BarePeer)s",
                             "xml=N%(XML)s",
                             "txt=N%(Body)s",
                             "kind=%(SType)s",
                             "nick=%(Nick)s"])];
                        _ -> [?SQL_INSERT(
                            "archive",
                            ["username=%(SUser)s",
                             "server_host=%(LServer)s",
                             "timestamp=%(TStmp)d",
                             "peer=%(LPeer)s",
                             "bare_peer=%(BarePeer)s",
                             "xml=%(XML)s",
                             "txt=%(Body)s",
                             "kind=%(SType)s",
                             "nick=%(Nick)s"])]
                            end;
         (_Host, _R) ->
              []
      end}].

is_empty_for_user(LUser, LServer) ->
    case ejabberd_sql:sql_query(
           LServer,
           ?SQL("select @(1)d from archive"
                " where username=%(LUser)s and %(LServer)H limit 1")) of
        {selected, [{1}]} ->
            false;
        _ ->
            true
    end.

is_empty_for_room(LServer, LName, LHost) ->
    LUser = jid:encode({LName, LHost, <<>>}),
    is_empty_for_user(LUser, LServer).

%%%===================================================================
%%% Internal functions
%%%===================================================================
make_sql_query(User, LServer, MAMQuery, RSM, ExtraUsernames) ->
    Start = proplists:get_value(start, MAMQuery),
    End = proplists:get_value('end', MAMQuery),
    With = proplists:get_value(with, MAMQuery),
    WithText = proplists:get_value(withtext, MAMQuery),
    {Max, Direction, ID} = get_max_direction_id(RSM),
    ODBCType = ejabberd_option:sql_type(LServer),
    ToString = fun(S) -> ejabberd_sql:to_string_literal(ODBCType, S) end,
    LimitClause = if is_integer(Max), Max >= 0, ODBCType /= mssql ->
                          [<<" limit ">>, integer_to_binary(Max+1)];
                     true ->
                          []
                  end,
    TopClause = if is_integer(Max), Max >= 0, ODBCType == mssql ->
                          [<<" TOP ">>, integer_to_binary(Max+1)];
                     true ->
                          []
                  end,
    SubOrderClause = if LimitClause /= []; TopClause /= [] ->
                          <<" ORDER BY timestamp DESC ">>;
                     true ->
                          []
                  end,
    WithTextClause = if is_binary(WithText), WithText /= <<>> ->
                             [<<" and match (txt) against (">>,
                              ToString(WithText), <<")">>];
                        true ->
                             []
                     end,
    WithClause = case catch jid:tolower(With) of
                     {_, _, <<>>} ->
                         [<<" and bare_peer=">>,
                          ToString(jid:encode(With))];
                     {_, _, _} ->
                         [<<" and peer=">>,
                          ToString(jid:encode(With))];
                     _ ->
                         []
                 end,
    PageClause = case catch binary_to_integer(ID) of
                     I when is_integer(I), I >= 0 ->
                         case Direction of
                             before ->
                                 [<<" AND timestamp < ">>, ID];
                             'after' ->
                                 [<<" AND timestamp > ">>, ID];
                             _ ->
                                 []
                         end;
                     _ ->
                         []
                 end,
    StartClause = case Start of
                      {_, _, _} ->
                          [<<" and timestamp >= ">>,
                           integer_to_binary(misc:now_to_usec(Start))];
                      _ ->
                          []
                  end,
    EndClause = case End of
                    {_, _, _} ->
                        [<<" and timestamp <= ">>,
                         integer_to_binary(misc:now_to_usec(End))];
                    _ ->
                        []
                end,
    SUser = ToString(User),
    SServer = ToString(LServer),

    HostMatch = case ejabberd_sql:use_new_schema() of
                    true ->
                        [<<" and server_host=", SServer/binary>>];
                    _ ->
                        <<"">>
                end,

    {UserSel, UserWhere} = case ExtraUsernames of
                               Users when is_list(Users) ->
                                   EscUsers = [ToString(U) || U <- [User | Users]],
                                   {<<" username,">>,
                                    [<<" username in (">>, str:join(EscUsers, <<",">>), <<")">>]};
                               subscribers_table ->
                                   SJid = ToString(jid:encode({User, LServer, <<>>})),
                                   RoomName = case ODBCType of
                                                  sqlite ->
                                                      <<"room || '@' || host">>;
                                                  _ ->
                                                      <<"concat(room, '@', host)">>
                                              end,
                                   {<<" username,">>,
                                    [<<" (username = ">>, SUser,
                                        <<" or username in (select ">>, RoomName,
                                          <<" from muc_room_subscribers where jid=">>, SJid, HostMatch, <<"))">>]};
                               _ ->
                                   {<<>>, [<<" username=">>, SUser]}
                           end,

    Query = [<<"SELECT ">>, TopClause, UserSel,
             <<" timestamp, xml, peer, kind, nick"
               " FROM archive WHERE">>, UserWhere, HostMatch,
             WithClause, WithTextClause,
             StartClause, EndClause, PageClause],

    QueryPage =
        case Direction of
            before ->
                % ID can be empty because of
                % XEP-0059: Result Set Management
                % 2.5 Requesting the Last Page in a Result Set
                [<<"SELECT">>, UserSel, <<" timestamp, xml, peer, kind, nick FROM (">>,
                 Query, SubOrderClause,
                 LimitClause, <<") AS t ORDER BY timestamp ASC;">>];
            _ ->
                [Query, <<" ORDER BY timestamp ASC ">>,
                 LimitClause, <<";">>]
        end,
    {QueryPage,
     [<<"SELECT COUNT(*) FROM archive WHERE ">>, UserWhere,
      HostMatch, WithClause, WithTextClause,
      StartClause, EndClause, <<";">>]}.

-spec get_max_direction_id(rsm_set() | undefined) ->
                                  {integer() | undefined,
                                   before | 'after' | undefined,
                                   binary()}.
get_max_direction_id(RSM) ->
    case RSM of
        #rsm_set{max = Max, before = Before} when is_binary(Before) ->
            {Max, before, Before};
        #rsm_set{max = Max, 'after' = After} when is_binary(After) ->
            {Max, 'after', After};
        #rsm_set{max = Max} ->
            {Max, undefined, <<>>};
        _ ->
            {undefined, undefined, <<>>}
    end.

-spec make_archive_el(binary(), binary(), binary(), binary(), binary(),
                      binary(), _, jid(), jid()) ->
                             {ok, xmpp_element()} | {error, invalid_jid |
                                                     invalid_timestamp |
                                                     invalid_xml}.
make_archive_el(User, TS, XML, Peer, Kind, Nick, MsgType, JidRequestor, JidArchive) ->
    case xml_compress:decode(XML, User, Peer) of
        #xmlel{} = El ->
            try binary_to_integer(TS) of
                TSInt ->
                    try jid:decode(Peer) of
                        PeerJID ->
                            Now = misc:usec_to_now(TSInt),
                            PeerLJID = jid:tolower(PeerJID),
                            T = case Kind of
                                    <<"">> -> chat;
                                    null -> chat;
                                    _ -> misc: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)
                    catch _:{bad_jid, _} ->
                            ?ERROR_MSG("Malformed 'peer' field with value "
                                       "'~ts' detected for user ~ts in table "
                                       "'archive': invalid JID",
                                       [Peer, jid:encode(JidArchive)]),
                            {error, invalid_jid}
                    end
            catch _:_ ->
                    ?ERROR_MSG("Malformed 'timestamp' field with value '~ts' "
                               "detected for user ~ts in table 'archive': "
                               "not an integer",
                               [TS, jid:encode(JidArchive)]),
                    {error, invalid_timestamp}
            end;
        {error, {_, Reason}} ->
            ?ERROR_MSG("Malformed 'xml' field with value '~ts' detected "
                       "for user ~ts in table 'archive': ~ts",
                       [XML, jid:encode(JidArchive), Reason]),
            {error, invalid_xml}
    end.

1	%%%-------------------------------------------------------------------
2	%%% File : mod_mam_sql.erl
3	%%% Author : Evgeny Khramtsov <ekhramtsov@process-one.net>
4	%%% Created : 15 Apr 2016 by Evgeny Khramtsov <ekhramtsov@process-one.net>
5	%%%
6	%%%
7	%%% ejabberd, Copyright (C) 2002-2023 ProcessOne
8	%%%
9	%%% This program is free software; you can redistribute it and/or
10	%%% modify it under the terms of the GNU General Public License as
11	%%% published by the Free Software Foundation; either version 2 of the
12	%%% License, or (at your option) any later version.
13	%%%
14	%%% This program is distributed in the hope that it will be useful,
15	%%% but WITHOUT ANY WARRANTY; without even the implied warranty of
16	%%% MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
17	%%% General Public License for more details.
18	%%%
19	%%% You should have received a copy of the GNU General Public License along
20	%%% with this program; if not, write to the Free Software Foundation, Inc.,
21	%%% 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.
22	%%%
23	%%%----------------------------------------------------------------------
24
25	-module(mod_mam_sql).
26
27
28	-behaviour(mod_mam).
29
30	%% API
31	-export([init/2, remove_user/2, remove_room/3, delete_old_messages/3,
32	extended_fields/0, store/8, write_prefs/4, get_prefs/2, select/7, export/1, remove_from_archive/3,
33	is_empty_for_user/2, is_empty_for_room/3, select_with_mucsub/6,
34	delete_old_messages_batch/4, count_messages_to_delete/3]).
35
36	-include_lib("stdlib/include/ms_transform.hrl").
37	-include_lib("xmpp/include/xmpp.hrl").
38	-include("mod_mam.hrl").
39	-include("logger.hrl").
40	-include("ejabberd_sql_pt.hrl").
41	-include("mod_muc_room.hrl").
42
43	%%%===================================================================
44	%%% API
45	%%%===================================================================
46	init(_Host, _Opts) ->
47	ok.	3✔
48
49	remove_user(LUser, LServer) ->
50	ejabberd_sql:sql_query(	354✔
51	LServer,
52	?SQL("delete from archive where username=%(LUser)s and %(LServer)H")),	354✔
53	ejabberd_sql:sql_query(	354✔
54	LServer,
55	?SQL("delete from archive_prefs where username=%(LUser)s and %(LServer)H")).	354✔
56
57	remove_room(LServer, LName, LHost) ->
58	LUser = jid:encode({LName, LHost, <<>>}),	264✔
59	remove_user(LUser, LServer).	264✔
60
61	remove_from_archive(LUser, LServer, none) ->
62	case ejabberd_sql:sql_query(LServer,	×
63	?SQL("delete from archive where username=%(LUser)s and %(LServer)H")) of	×
64	{error, Reason} -> {error, Reason};	×
65	_ -> ok	×
66	end;
67	remove_from_archive(LUser, LServer, #jid{} = WithJid) ->
68	Peer = jid:encode(jid:remove_resource(WithJid)),	×
69	case ejabberd_sql:sql_query(LServer,	×
70	?SQL("delete from archive where username=%(LUser)s and %(LServer)H and bare_peer=%(Peer)s")) of	×
71	{error, Reason} -> {error, Reason};	×
72	_ -> ok	×
73	end;
74	remove_from_archive(LUser, LServer, StanzaId) ->
75	case ejabberd_sql:sql_query(LServer,	×
76	?SQL("delete from archive where username=%(LUser)s and %(LServer)H and timestamp=%(StanzaId)d")) of	×
77	{error, Reason} -> {error, Reason};	×
78	_ -> ok	×
79	end.
80
81	count_messages_to_delete(ServerHost, TimeStamp, Type) ->
82	TS = misc:now_to_usec(TimeStamp),	×
83	Res =	×
84	case Type of
85	all ->
86	ejabberd_sql:sql_query(	×
87	ServerHost,
88	?SQL("select count(*) from archive"	×
89	" where timestamp < %(TS)d and %(ServerHost)H"));
90	_ ->
91	SType = misc:atom_to_binary(Type),	×
92	ejabberd_sql:sql_query(	×
93	ServerHost,
94	?SQL("select @(count(*))d from archive"	×
95	" where timestamp < %(TS)d"
96	" and kind=%(SType)s"
97	" and %(ServerHost)H"))
98	end,
99	case Res of	×
100	{selected, [Count]} ->
101	{ok, Count};	×
102	_ ->
103	error	×
104	end.
105
106	delete_old_messages_batch(ServerHost, TimeStamp, Type, Batch) ->
107	TS = misc:now_to_usec(TimeStamp),	×
108	Res =	×
109	case Type of
110	all ->
111	ejabberd_sql:sql_query(	×
112	ServerHost,
113	?SQL("delete from archive"	×
114	" where timestamp < %(TS)d and %(ServerHost)H limit %(Batch)d"));
115	_ ->
116	SType = misc:atom_to_binary(Type),	×
117	ejabberd_sql:sql_query(	×
118	ServerHost,
119	?SQL("delete from archive"	×
120	" where timestamp < %(TS)d"
121	" and kind=%(SType)s"
122	" and %(ServerHost)H limit %(Batch)d"))
123	end,
124	case Res of	×
125	{updated, Count} ->
126	{ok, Count};	×
127	{error, _} = Error ->
128	Error	×
129	end.
130
131	delete_old_messages(ServerHost, TimeStamp, Type) ->
132	TS = misc:now_to_usec(TimeStamp),	×
133	case Type of	×
134	all ->
135	ejabberd_sql:sql_query(	×
136	ServerHost,
137	?SQL("delete from archive"	×
138	" where timestamp < %(TS)d and %(ServerHost)H"));
139	_ ->
140	SType = misc:atom_to_binary(Type),	×
141	ejabberd_sql:sql_query(	×
142	ServerHost,
143	?SQL("delete from archive"	×
144	" where timestamp < %(TS)d"
145	" and kind=%(SType)s"
146	" and %(ServerHost)H"))
147	end,
148	ok.	×
149
150	extended_fields() ->
151	[{withtext, <<"">>}].	18✔
152
153	store(Pkt, LServer, {LUser, LHost}, Type, Peer, Nick, _Dir, TS) ->
154	SUser = case Type of	789✔
155	chat -> LUser;	723✔
156	groupchat -> jid:encode({LUser, LHost, <<>>})	66✔
157	end,
158	BarePeer = jid:encode(	789✔
159	jid:tolower(
160	jid:remove_resource(Peer))),
161	LPeer = jid:encode(	789✔
162	jid:tolower(Peer)),
163	Body = fxml:get_subtag_cdata(Pkt, <<"body">>),	789✔
164	SType = misc:atom_to_binary(Type),	789✔
165	SqlType = ejabberd_option:sql_type(LServer),	789✔
166	XML = case mod_mam_opt:compress_xml(LServer) of	789✔
167	true ->
168	J1 = case Type of	×
169	chat -> jid:encode({LUser, LHost, <<>>});	×
170	groupchat -> SUser	×
171	end,
172	xml_compress:encode(Pkt, J1, LPeer);	×
173	_ ->
174	fxml:element_to_binary(Pkt)	789✔
175	end,
176	case SqlType of	789✔
177	mssql -> case ejabberd_sql:sql_query(	×
178	LServer,
179	?SQL_INSERT(	×
180	"archive",
181	["username=%(SUser)s",
182	"server_host=%(LServer)s",
183	"timestamp=%(TS)d",
184	"peer=%(LPeer)s",
185	"bare_peer=%(BarePeer)s",
186	"xml=N%(XML)s",
187	"txt=N%(Body)s",
188	"kind=%(SType)s",
189	"nick=%(Nick)s"])) of
190	{updated, _} ->
191	ok;	×
192	Err ->
193	Err	×
194	end;
195	_ -> case ejabberd_sql:sql_query(	789✔
196	LServer,
197	?SQL_INSERT(	789✔
198	"archive",
199	["username=%(SUser)s",
200	"server_host=%(LServer)s",
201	"timestamp=%(TS)d",
202	"peer=%(LPeer)s",
203	"bare_peer=%(BarePeer)s",
204	"xml=%(XML)s",
205	"txt=%(Body)s",
206	"kind=%(SType)s",
207	"nick=%(Nick)s"])) of
208	{updated, _} ->
209	ok;	789✔
210	Err ->
211	Err	×
212	end
213	end.
214
215	write_prefs(LUser, _LServer, #archive_prefs{default = Default,
216	never = Never,
217	always = Always},
218	ServerHost) ->
219	SDefault = erlang:atom_to_binary(Default, utf8),	483✔
220	SAlways = misc:term_to_expr(Always),	483✔
221	SNever = misc:term_to_expr(Never),	483✔
222	case ?SQL_UPSERT(	483✔
223	ServerHost,	483✔
224	"archive_prefs",
225	["!username=%(LUser)s",
226	"!server_host=%(ServerHost)s",
227	"def=%(SDefault)s",
228	"always=%(SAlways)s",
229	"never=%(SNever)s"]) of
230	ok ->
231	ok;	483✔
232	Err ->
233	Err	×
234	end.
235
236	get_prefs(LUser, LServer) ->
237	case ejabberd_sql:sql_query(	360✔
238	LServer,
239	?SQL("select @(def)s, @(always)s, @(never)s from archive_prefs"	360✔
240	" where username=%(LUser)s and %(LServer)H")) of
241	{selected, [{SDefault, SAlways, SNever}]} ->
242	Default = erlang:binary_to_existing_atom(SDefault, utf8),	339✔
243	Always = ejabberd_sql:decode_term(SAlways),	339✔
244	Never = ejabberd_sql:decode_term(SNever),	339✔
245	{ok, #archive_prefs{us = {LUser, LServer},	339✔
246	default = Default,
247	always = Always,
248	never = Never}};
249	_ ->
250	error	21✔
251	end.
252
253	select(LServer, JidRequestor, #jid{luser = LUser} = JidArchive,
254	MAMQuery, RSM, MsgType, Flags) ->
255	User = case MsgType of	570✔
256	chat -> LUser;	558✔
257	_ -> jid:encode(JidArchive)	12✔
258	end,
259	{Query, CountQuery} = make_sql_query(User, LServer, MAMQuery, RSM, none),	570✔
260	do_select_query(LServer, JidRequestor, JidArchive, RSM, MsgType, Query, CountQuery, Flags).	570✔
261
262	-spec select_with_mucsub(binary(), jid(), jid(), mam_query:result(),
263	#rsm_set{} \| undefined, all \| only_count \| only_messages) ->
264	{[{binary(), non_neg_integer(), xmlel()}], boolean(), non_neg_integer()} \|
265	{error, db_failure}.
266	select_with_mucsub(LServer, JidRequestor, #jid{luser = LUser} = JidArchive,
267	MAMQuery, RSM, Flags) ->
268	Extra = case gen_mod:db_mod(LServer, mod_muc) of	9✔
269	mod_muc_sql ->
270	subscribers_table;	9✔
271	_ ->
272	SubRooms = case mod_muc_admin:find_hosts(LServer) of	×
273	[First\|_] ->
274	case mod_muc:get_subscribed_rooms(First, JidRequestor) of	×
275	{ok, L} -> L;	×
276	{error, _} -> []	×
277	end;
278	_ ->
279	[]	×
280	end,
281	[jid:encode(Jid) \|\| {Jid, _, _} <- SubRooms]	×
282	end,
283	{Query, CountQuery} = make_sql_query(LUser, LServer, MAMQuery, RSM, Extra),	9✔
284	do_select_query(LServer, JidRequestor, JidArchive, RSM, chat, Query, CountQuery, Flags).	9✔
285
286	do_select_query(LServer, JidRequestor, #jid{luser = LUser} = JidArchive, RSM,
287	MsgType, Query, CountQuery, Flags) ->
288	% TODO from XEP-0313 v0.2: "To conserve resources, a server MAY place a
289	% reasonable limit on how many stanzas may be pushed to a client in one
290	% request. If a query returns a number of stanzas greater than this limit
291	% and the client did not specify a limit using RSM then the server should
292	% return a policy-violation error to the client." We currently don't do this
293	% for v0.2 requests, but we do limit #rsm_in.max for v0.3 and newer.
294	QRes = case Flags of	579✔
295	all ->
296	{ejabberd_sql:sql_query(LServer, Query), ejabberd_sql:sql_query(LServer, CountQuery)};	573✔
297	only_messages ->
298	{ejabberd_sql:sql_query(LServer, Query), {selected, ok, [[<<"0">>]]}};	6✔
299	only_count ->
300	{{selected, ok, []}, ejabberd_sql:sql_query(LServer, CountQuery)}	×
301	end,
302	case QRes of	579✔
303	{{selected, _, Res}, {selected, _, [[Count]]}} ->
304	{Max, Direction, _} = get_max_direction_id(RSM),	579✔
305	{Res1, IsComplete} =	579✔
306	if Max >= 0 andalso Max /= undefined andalso length(Res) > Max ->
307	if Direction == before ->	144✔
308	{lists:nthtail(1, Res), false};	24✔
309	true ->
310	{lists:sublist(Res, Max), false}	120✔
311	end;
312	true ->
313	{Res, true}	435✔
314	end,
315	MucState = #state{config = #config{anonymous = true}},	579✔
316	JidArchiveS = jid:encode(jid:remove_resource(JidArchive)),	579✔
317	{lists:flatmap(	579✔
318	fun([TS, XML, PeerBin, Kind, Nick]) ->
319	case make_archive_el(JidArchiveS, TS, XML, PeerBin, Kind, Nick,	2,196✔
320	MsgType, JidRequestor, JidArchive) of
321	{ok, El} ->
322	[{TS, binary_to_integer(TS), El}];	2,196✔
323	{error, _} ->
324	[]	×
325	end;
326	([User, TS, XML, PeerBin, Kind, Nick]) when User == LUser ->
327	case make_archive_el(JidArchiveS, TS, XML, PeerBin, Kind, Nick,	×
328	MsgType, JidRequestor, JidArchive) of
329	{ok, El} ->
330	[{TS, binary_to_integer(TS), El}];	×
331	{error, _} ->
332	[]	×
333	end;
334	([User, TS, XML, PeerBin, Kind, Nick]) ->
335	case make_archive_el(User, TS, XML, PeerBin, Kind, Nick,	33✔
336	{groupchat, member, MucState}, JidRequestor,
337	jid:decode(User)) of
338	{ok, El} ->
339	mod_mam:wrap_as_mucsub([{TS, binary_to_integer(TS), El}],	33✔
340	JidRequestor);
341	{error, _} ->
342	[]	×
343	end
344	end, Res1), IsComplete, binary_to_integer(Count)};
345	_ ->
346	{[], false, 0}	×
347	end.
348
349	export(_Server) ->
350	[{archive_prefs,	×
351	fun(Host, #archive_prefs{us =
352	{LUser, LServer},
353	default = Default,
354	always = Always,
355	never = Never})
356	when LServer == Host ->
357	SDefault = erlang:atom_to_binary(Default, utf8),	×
358	SAlways = misc:term_to_expr(Always),	×
359	SNever = misc:term_to_expr(Never),	×
360	[?SQL_INSERT(	×
361	"archive_prefs",
362	["username=%(LUser)s",
363	"server_host=%(LServer)s",
364	"def=%(SDefault)s",
365	"always=%(SAlways)s",
366	"never=%(SNever)s"])];
367	(_Host, _R) ->
368	[]	×
369	end},
370	{archive_msg,
371	fun([Host \| HostTail], #archive_msg{us ={LUser, LServer},
372	id = _ID, timestamp = TS, peer = Peer,
373	type = Type, nick = Nick, packet = Pkt})
374	when (LServer == Host) or ([LServer] == HostTail) ->
375	TStmp = misc:now_to_usec(TS),	×
376	SUser = case Type of	×
377	chat -> LUser;	×
378	groupchat -> jid:encode({LUser, LServer, <<>>})	×
379	end,
380	BarePeer = jid:encode(jid:tolower(jid:remove_resource(Peer))),	×
381	LPeer = jid:encode(jid:tolower(Peer)),	×
382	XML = fxml:element_to_binary(Pkt),	×
383	Body = fxml:get_subtag_cdata(Pkt, <<"body">>),	×
384	SType = misc:atom_to_binary(Type),	×
385	SqlType = ejabberd_option:sql_type(Host),	×
386	case SqlType of	×
387	mssql -> [?SQL_INSERT(	×
388	"archive",
389	["username=%(SUser)s",
390	"server_host=%(LServer)s",
391	"timestamp=%(TStmp)d",
392	"peer=%(LPeer)s",
393	"bare_peer=%(BarePeer)s",
394	"xml=N%(XML)s",
395	"txt=N%(Body)s",
396	"kind=%(SType)s",
397	"nick=%(Nick)s"])];
398	_ -> [?SQL_INSERT(	×
399	"archive",
400	["username=%(SUser)s",
401	"server_host=%(LServer)s",
402	"timestamp=%(TStmp)d",
403	"peer=%(LPeer)s",
404	"bare_peer=%(BarePeer)s",
405	"xml=%(XML)s",
406	"txt=%(Body)s",
407	"kind=%(SType)s",
408	"nick=%(Nick)s"])]
409	end;
410	(_Host, _R) ->
411	[]	×
412	end}].
413
414	is_empty_for_user(LUser, LServer) ->
415	case ejabberd_sql:sql_query(	×
416	LServer,
417	?SQL("select @(1)d from archive"	×
418	" where username=%(LUser)s and %(LServer)H limit 1")) of
419	{selected, [{1}]} ->
420	false;	×
421	_ ->
422	true	×
423	end.
424
425	is_empty_for_room(LServer, LName, LHost) ->
426	LUser = jid:encode({LName, LHost, <<>>}),	×
427	is_empty_for_user(LUser, LServer).	×
428
429	%%%===================================================================
430	%%% Internal functions
431	%%%===================================================================
432	make_sql_query(User, LServer, MAMQuery, RSM, ExtraUsernames) ->
433	Start = proplists:get_value(start, MAMQuery),	579✔
434	End = proplists:get_value('end', MAMQuery),	579✔
435	With = proplists:get_value(with, MAMQuery),	579✔
436	WithText = proplists:get_value(withtext, MAMQuery),	579✔
437	{Max, Direction, ID} = get_max_direction_id(RSM),	579✔
438	ODBCType = ejabberd_option:sql_type(LServer),	579✔
439	ToString = fun(S) -> ejabberd_sql:to_string_literal(ODBCType, S) end,	579✔
440	LimitClause = if is_integer(Max), Max >= 0, ODBCType /= mssql ->	579✔
441	[<<" limit ">>, integer_to_binary(Max+1)];	483✔
442	true ->
443	[]	96✔
444	end,
445	TopClause = if is_integer(Max), Max >= 0, ODBCType == mssql ->	579✔
446	[<<" TOP ">>, integer_to_binary(Max+1)];	×
447	true ->
448	[]	579✔
449	end,
450	SubOrderClause = if LimitClause /= []; TopClause /= [] ->	579✔
451	<<" ORDER BY timestamp DESC ">>;	483✔
452	true ->
453	[]	96✔
454	end,
455	WithTextClause = if is_binary(WithText), WithText /= <<>> ->	579✔
456	[<<" and match (txt) against (">>,	×
457	ToString(WithText), <<")">>];
458	true ->
459	[]	579✔
460	end,
461	WithClause = case catch jid:tolower(With) of	579✔
462	{_, _, <<>>} ->
463	[<<" and bare_peer=">>,	24✔
464	ToString(jid:encode(With))];
465	{_, _, _} ->
466	[<<" and peer=">>,	24✔
467	ToString(jid:encode(With))];
468	_ ->
469	[]	531✔
470	end,
471	PageClause = case catch binary_to_integer(ID) of	579✔
472	I when is_integer(I), I >= 0 ->
473	case Direction of	246✔
474	before ->
475	[<<" AND timestamp < ">>, ID];	126✔
476	'after' ->
477	[<<" AND timestamp > ">>, ID];	120✔
478	_ ->
479	[]	×
480	end;
481	_ ->
482	[]	333✔
483	end,
484	StartClause = case Start of	579✔
485	{_, _, _} ->
486	[<<" and timestamp >= ">>,	6✔
487	integer_to_binary(misc:now_to_usec(Start))];
488	_ ->
489	[]	573✔
490	end,
491	EndClause = case End of	579✔
492	{_, _, _} ->
493	[<<" and timestamp <= ">>,	×
494	integer_to_binary(misc:now_to_usec(End))];
495	_ ->
496	[]	579✔
497	end,
498	SUser = ToString(User),	579✔
499	SServer = ToString(LServer),	579✔
500
501	HostMatch = case ejabberd_sql:use_new_schema() of	579✔
502	true ->
503	[<<" and server_host=", SServer/binary>>];	×
504	_ ->
505	<<"">>	579✔
506	end,
507
508	{UserSel, UserWhere} = case ExtraUsernames of	579✔
509	Users when is_list(Users) ->
510	EscUsers = [ToString(U) \|\| U <- [User \| Users]],	×
511	{<<" username,">>,	×
512	[<<" username in (">>, str:join(EscUsers, <<",">>), <<")">>]};
513	subscribers_table ->
514	SJid = ToString(jid:encode({User, LServer, <<>>})),	9✔
515	RoomName = case ODBCType of	9✔
516	sqlite ->
517	<<"room \|\| '@' \|\| host">>;	3✔
518	_ ->
519	<<"concat(room, '@', host)">>	6✔
520	end,
521	{<<" username,">>,	9✔
522	[<<" (username = ">>, SUser,
523	<<" or username in (select ">>, RoomName,
524	<<" from muc_room_subscribers where jid=">>, SJid, HostMatch, <<"))">>]};
525	_ ->
526	{<<>>, [<<" username=">>, SUser]}	570✔
527	end,
528
529	Query = [<<"SELECT ">>, TopClause, UserSel,	579✔
530	<<" timestamp, xml, peer, kind, nick"
531	" FROM archive WHERE">>, UserWhere, HostMatch,
532	WithClause, WithTextClause,
533	StartClause, EndClause, PageClause],
534
535	QueryPage =	579✔
536	case Direction of
537	before ->
538	% ID can be empty because of
539	% XEP-0059: Result Set Management
540	% 2.5 Requesting the Last Page in a Result Set
541	[<<"SELECT">>, UserSel, <<" timestamp, xml, peer, kind, nick FROM (">>,	174✔
542	Query, SubOrderClause,
543	LimitClause, <<") AS t ORDER BY timestamp ASC;">>];
544	_ ->
545	[Query, <<" ORDER BY timestamp ASC ">>,	405✔
546	LimitClause, <<";">>]
547	end,
548	{QueryPage,	579✔
549	[<<"SELECT COUNT(*) FROM archive WHERE ">>, UserWhere,
550	HostMatch, WithClause, WithTextClause,
551	StartClause, EndClause, <<";">>]}.
552
553	-spec get_max_direction_id(rsm_set() \| undefined) ->
554	{integer() \| undefined,
555	before \| 'after' \| undefined,
556	binary()}.
557	get_max_direction_id(RSM) ->
558	case RSM of	1,158✔
559	#rsm_set{max = Max, before = Before} when is_binary(Before) ->
560	{Max, before, Before};	348✔
561	#rsm_set{max = Max, 'after' = After} when is_binary(After) ->
562	{Max, 'after', After};	240✔
563	#rsm_set{max = Max} ->
564	{Max, undefined, <<>>};	534✔
565	_ ->
566	{undefined, undefined, <<>>}	36✔
567	end.
568
569	-spec make_archive_el(binary(), binary(), binary(), binary(), binary(),
570	binary(), _, jid(), jid()) ->
571	{ok, xmpp_element()} \| {error, invalid_jid \|
572	invalid_timestamp \|
573	invalid_xml}.
574	make_archive_el(User, TS, XML, Peer, Kind, Nick, MsgType, JidRequestor, JidArchive) ->
575	case xml_compress:decode(XML, User, Peer) of	2,229✔
576	#xmlel{} = El ->
577	try binary_to_integer(TS) of	2,229✔
578	TSInt ->
579	try jid:decode(Peer) of	2,229✔
580	PeerJID ->
581	Now = misc:usec_to_now(TSInt),	2,229✔
582	PeerLJID = jid:tolower(PeerJID),	2,229✔
583	T = case Kind of	2,229✔
584	<<"">> -> chat;	×
585	null -> chat;	×
586	_ -> misc:binary_to_atom(Kind)	2,229✔
587	end,
588	mod_mam:msg_to_el(	2,229✔
589	#archive_msg{timestamp = Now,
590	id = TS,
591	packet = El,
592	type = T,
593	nick = Nick,
594	peer = PeerLJID},
595	MsgType, JidRequestor, JidArchive)
596	catch _:{bad_jid, _} ->
597	?ERROR_MSG("Malformed 'peer' field with value "	×
598	"'~ts' detected for user ~ts in table "
599	"'archive': invalid JID",
600	[Peer, jid:encode(JidArchive)]),	×
601	{error, invalid_jid}	×
602	end
603	catch _:_ ->
604	?ERROR_MSG("Malformed 'timestamp' field with value '~ts' "	×
605	"detected for user ~ts in table 'archive': "
606	"not an integer",
607	[TS, jid:encode(JidArchive)]),	×
608	{error, invalid_timestamp}	×
609	end;
610	{error, {_, Reason}} ->
611	?ERROR_MSG("Malformed 'xml' field with value '~ts' detected "	×
612	"for user ~ts in table 'archive': ~ts",
613	[XML, jid:encode(JidArchive), Reason]),	×
614	{error, invalid_xml}	×
615	end.

processone / ejabberd / 511

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