xmpp.chapril.org-ejabberd/src/mod_sm.erl

%%%-------------------------------------------------------------------
%%% Author  : Holger Weiss <holger@zedat.fu-berlin.de>
%%% Created : 25 Dec 2016 by Evgeny Khramtsov <ekhramtsov@process-one.net>
%%%
%%%
%%% ejabberd, Copyright (C) 2002-2016   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_sm).
-behaviour(gen_mod).
-author('holger@zedat.fu-berlin.de').
-protocol({xep, 198, '1.5.2'}).

%% gen_mod API
-export([start/2, stop/1, depends/2, mod_opt_type/1]).
%% hooks
-export([c2s_stream_init/2, c2s_stream_started/2, c2s_stream_features/2,
	 c2s_authenticated_packet/2, c2s_unauthenticated_packet/2,
	 c2s_unbinded_packet/2, c2s_closed/2, c2s_terminated/2,
	 c2s_handle_send/3, c2s_handle_info/2, c2s_handle_call/3,
	 c2s_handle_recv/3]).

-include("xmpp.hrl").
-include("logger.hrl").

-define(is_sm_packet(Pkt),
	is_record(Pkt, sm_enable) or
	is_record(Pkt, sm_resume) or
	is_record(Pkt, sm_a) or
	is_record(Pkt, sm_r)).

-type state() :: ejabberd_c2s:state().
-type lqueue() :: {non_neg_integer(), queue:queue()}.

%%%===================================================================
%%% API
%%%===================================================================
start(Host, _Opts) ->
    ejabberd_hooks:add(c2s_init, ?MODULE, c2s_stream_init, 50),
    ejabberd_hooks:add(c2s_stream_started, Host, ?MODULE,
		       c2s_stream_started, 50),
    ejabberd_hooks:add(c2s_post_auth_features, Host, ?MODULE,
		       c2s_stream_features, 50),
    ejabberd_hooks:add(c2s_unauthenticated_packet, Host, ?MODULE,
		       c2s_unauthenticated_packet, 50),
    ejabberd_hooks:add(c2s_unbinded_packet, Host, ?MODULE,
		       c2s_unbinded_packet, 50),
    ejabberd_hooks:add(c2s_authenticated_packet, Host, ?MODULE,
		       c2s_authenticated_packet, 50),
    ejabberd_hooks:add(c2s_handle_send, Host, ?MODULE, c2s_handle_send, 50),
    ejabberd_hooks:add(c2s_handle_recv, Host, ?MODULE, c2s_handle_recv, 50),
    ejabberd_hooks:add(c2s_handle_info, Host, ?MODULE, c2s_handle_info, 50),
    ejabberd_hooks:add(c2s_handle_call, Host, ?MODULE, c2s_handle_call, 50),
    ejabberd_hooks:add(c2s_closed, Host, ?MODULE, c2s_closed, 50),
    ejabberd_hooks:add(c2s_terminated, Host, ?MODULE, c2s_terminated, 50).

stop(Host) ->
    %% TODO: do something with global 'c2s_init' hook
    ejabberd_hooks:delete(c2s_stream_started, Host, ?MODULE,
			  c2s_stream_started, 50),
    ejabberd_hooks:delete(c2s_post_auth_features, Host, ?MODULE,
			  c2s_stream_features, 50),
    ejabberd_hooks:delete(c2s_unauthenticated_packet, Host, ?MODULE,
			  c2s_unauthenticated_packet, 50),
    ejabberd_hooks:delete(c2s_unbinded_packet, Host, ?MODULE,
			  c2s_unbinded_packet, 50),
    ejabberd_hooks:delete(c2s_authenticated_packet, Host, ?MODULE,
			  c2s_authenticated_packet, 50),
    ejabberd_hooks:delete(c2s_handle_send, Host, ?MODULE, c2s_handle_send, 50),
    ejabberd_hooks:delete(c2s_handle_recv, Host, ?MODULE, c2s_handle_recv, 50),
    ejabberd_hooks:delete(c2s_handle_info, Host, ?MODULE, c2s_handle_info, 50),
    ejabberd_hooks:delete(c2s_handle_call, Host, ?MODULE, c2s_handle_call, 50),
    ejabberd_hooks:delete(c2s_closed, Host, ?MODULE, c2s_closed, 50),
    ejabberd_hooks:delete(c2s_terminated, Host, ?MODULE, c2s_terminated, 50).

depends(_Host, _Opts) ->
    [].

c2s_stream_init({ok, State}, Opts) ->
    MgmtOpts = lists:filter(
		 fun({stream_management, _}) -> true;
		    ({max_ack_queue, _}) -> true;
		    ({resume_timeout, _}) -> true;
		    ({max_resume_timeout, _}) -> true;
		    ({ack_timeout, _}) -> true;
		    ({resend_on_timeout, _}) -> true;
		    (_) -> false
		 end, Opts),
    {ok, State#{mgmt_options => MgmtOpts}};
c2s_stream_init(Acc, _Opts) ->
    Acc.

c2s_stream_started(#{lserver := LServer, mgmt_options := Opts} = State,
		   _StreamStart) ->
    State1 = maps:remove(mgmt_options, State),
    ResumeTimeout = get_resume_timeout(LServer, Opts),
    MaxResumeTimeout = get_max_resume_timeout(LServer, Opts, ResumeTimeout),
    State1#{mgmt_state => inactive,
	    mgmt_max_queue => get_max_ack_queue(LServer, Opts),
	    mgmt_timeout => ResumeTimeout,
	    mgmt_max_timeout => MaxResumeTimeout,
	    mgmt_ack_timeout => get_ack_timeout(LServer, Opts),
	    mgmt_resend => get_resend_on_timeout(LServer, Opts),
	    mgmt_stanzas_in => 0,
	    mgmt_stanzas_out => 0,
	    mgmt_stanzas_req => 0};
c2s_stream_started(State, _StreamStart) ->
    State.

c2s_stream_features(Acc, Host) ->
    case gen_mod:is_loaded(Host, ?MODULE) of
	true ->
	    [#feature_sm{xmlns = ?NS_STREAM_MGMT_2},
	     #feature_sm{xmlns = ?NS_STREAM_MGMT_3}|Acc];
	false ->
	    Acc
    end.

c2s_unauthenticated_packet(State, Pkt) when ?is_sm_packet(Pkt) ->
    %% XEP-0198 says: "For client-to-server connections, the client MUST NOT
    %% attempt to enable stream management until after it has completed Resource
    %% Binding unless it is resuming a previous session".  However, it also
    %% says: "Stream management errors SHOULD be considered recoverable", so we
    %% won't bail out.
    Err = #sm_failed{reason = 'unexpected-request', xmlns = ?NS_STREAM_MGMT_3},
    {stop, send(State, Err)};
c2s_unauthenticated_packet(State, _Pkt) ->
    State.

c2s_unbinded_packet(State, #sm_resume{} = Pkt) ->
    case handle_resume(State, Pkt) of
	{ok, ResumedState} ->
	    {stop, ResumedState};
	{error, State1} ->
	    {stop, State1}
    end;
c2s_unbinded_packet(State, Pkt) when ?is_sm_packet(Pkt) ->
    c2s_unauthenticated_packet(State, Pkt);
c2s_unbinded_packet(State, _Pkt) ->
    State.

c2s_authenticated_packet(#{mgmt_state := MgmtState} = State, Pkt)
  when ?is_sm_packet(Pkt) ->
    if MgmtState == pending; MgmtState == active ->
	    {stop, perform_stream_mgmt(Pkt, State)};
       true ->
	    {stop, negotiate_stream_mgmt(Pkt, State)}
    end;
c2s_authenticated_packet(State, Pkt) ->
    update_num_stanzas_in(State, Pkt).

c2s_handle_recv(#{lang := Lang} = State, El, {error, Why}) ->
    Xmlns = xmpp:get_ns(El),
    if Xmlns == ?NS_STREAM_MGMT_2; Xmlns == ?NS_STREAM_MGMT_3 ->
	    Txt = xmpp:io_format_error(Why),
	    Err = #sm_failed{reason = 'bad-request',
			     text = xmpp:mk_text(Txt, Lang),
			     xmlns = Xmlns},
	    send(State, Err);
       true ->
	    State
    end;
c2s_handle_recv(State, _, _) ->
    State.

c2s_handle_send(#{mgmt_state := MgmtState, mod := Mod,
		  lang := Lang} = State, Pkt, SendResult)
  when MgmtState == pending; MgmtState == active ->
    case xmpp:is_stanza(Pkt) of
	true ->
	    case mgmt_queue_add(State, Pkt) of
		#{mgmt_max_queue := exceeded} = State1 ->
		    State2 = State1#{mgmt_resend => false},
		    case MgmtState of
			active ->
			    Err = xmpp:serr_policy_violation(
				    <<"Too many unacked stanzas">>, Lang),
			    send(State2, Err);
			_ ->
			    Mod:stop(State2)
		    end;
		State1 when SendResult == ok ->
		    send_rack(State1);
		State1 ->
		    State1
	    end;
	false ->
	    State
    end;
c2s_handle_send(State, _Pkt, _Result) ->
    State.

c2s_handle_call(#{sid := {Time, _}, mod := Mod} = State,
		{resume_session, Time}, From) ->
    Mod:reply(From, {resume, State}),
    {stop, State#{mgmt_state => resumed}};
c2s_handle_call(#{mod := Mod} = State, {resume_session, _}, From) ->
    Mod:reply(From, {error, <<"Previous session not found">>}),
    {stop, State};
c2s_handle_call(State, _Call, _From) ->
    State.

c2s_handle_info(#{mgmt_ack_timer := TRef, jid := JID, mod := Mod} = State,
		{timeout, TRef, ack_timeout}) ->
    ?DEBUG("Timed out waiting for stream management acknowledgement of ~s",
	   [jid:to_string(JID)]),
    State1 = State#{stop_reason => {socket, timeout}},
    State2 = Mod:close(State1, _SendTrailer = false),
    {stop, transition_to_pending(State2)};
c2s_handle_info(#{mgmt_state := pending, jid := JID, mod := Mod} = State,
		{timeout, _, pending_timeout}) ->
    ?DEBUG("Timed out waiting for resumption of stream for ~s",
	   [jid:to_string(JID)]),
    Mod:stop(State#{mgmt_state => timeout});
c2s_handle_info(#{jid := JID} = State, {_Ref, {resume, OldState}}) ->
    %% This happens if the resume_session/1 request timed out; the new session
    %% now receives the late response.
    ?DEBUG("Received old session state for ~s after failed resumption",
	   [jid:to_string(JID)]),
    route_unacked_stanzas(OldState#{mgmt_resend => false}),
    State;
c2s_handle_info(State, _) ->
    State.

c2s_closed(State, {stream, _}) ->
    State;
c2s_closed(#{mgmt_state := active} = State, _Reason) ->
    {stop, transition_to_pending(State)};
c2s_closed(State, _Reason) ->
    State.

c2s_terminated(#{mgmt_state := resumed, jid := JID} = State, _Reason) ->
    ?INFO_MSG("Closing former stream of resumed session for ~s",
	      [jid:to_string(JID)]),
    bounce_message_queue(),
    {stop, State};
c2s_terminated(#{mgmt_state := MgmtState, mgmt_stanzas_in := In, sid := SID,
		 user := U, server := S, resource := R} = State, _Reason) ->
    case MgmtState of
	timeout ->
	    Info = [{num_stanzas_in, In}],
	    ejabberd_sm:set_offline_info(SID, U, S, R, Info);
	_ ->
	    ok
    end,
    route_unacked_stanzas(State),
    State;
c2s_terminated(State, _Reason) ->
    State.

%%%===================================================================
%%% Internal functions
%%%===================================================================
-spec negotiate_stream_mgmt(xmpp_element(), state()) -> state().
negotiate_stream_mgmt(Pkt, State) ->
    Xmlns = xmpp:get_ns(Pkt),
    case Pkt of
	#sm_enable{} ->
	    handle_enable(State#{mgmt_xmlns => Xmlns}, Pkt);
	_ when is_record(Pkt, sm_a);
	       is_record(Pkt, sm_r);
	       is_record(Pkt, sm_resume) ->
	    Err = #sm_failed{reason = 'unexpected-request', xmlns = Xmlns},
	    send(State, Err);
	_ ->
	    Err = #sm_failed{reason = 'bad-request', xmlns = Xmlns},
	    send(State, Err)
    end.

-spec perform_stream_mgmt(xmpp_element(), state()) -> state().
perform_stream_mgmt(Pkt, #{mgmt_xmlns := Xmlns} = State) ->
    case xmpp:get_ns(Pkt) of
	Xmlns ->
	    case Pkt of
		#sm_r{} ->
		    handle_r(State);
		#sm_a{} ->
		    handle_a(State, Pkt);
		_ when is_record(Pkt, sm_enable);
		       is_record(Pkt, sm_resume) ->
		    send(State, #sm_failed{reason = 'unexpected-request',
					   xmlns = Xmlns});
		_ ->
		    send(State, #sm_failed{reason = 'bad-request',
					   xmlns = Xmlns})
	    end;
	_ ->
	    send(State, #sm_failed{reason = 'unsupported-version', xmlns = Xmlns})
    end.

-spec handle_enable(state(), sm_enable()) -> state().
handle_enable(#{mgmt_timeout := DefaultTimeout,
		mgmt_max_timeout := MaxTimeout,
		mgmt_xmlns := Xmlns, jid := JID} = State,
	      #sm_enable{resume = Resume, max = Max}) ->
    Timeout = if Resume == false ->
		      0;
		 Max /= undefined, Max > 0, Max =< MaxTimeout ->
		      Max;
		 true ->
		      DefaultTimeout
	      end,
    Res = if Timeout > 0 ->
		  ?INFO_MSG("Stream management with resumption enabled for ~s",
			    [jid:to_string(JID)]),
		  #sm_enabled{xmlns = Xmlns,
			      id = make_resume_id(State),
			      resume = true,
			      max = Timeout};
	     true ->
		  ?INFO_MSG("Stream management without resumption enabled for ~s",
			    [jid:to_string(JID)]),
		  #sm_enabled{xmlns = Xmlns}
	  end,
    State1 = State#{mgmt_state => active,
		    mgmt_queue => queue_new(),
		    mgmt_timeout => Timeout},
    send(State1, Res).

-spec handle_r(state()) -> state().
handle_r(#{mgmt_xmlns := Xmlns, mgmt_stanzas_in := H} = State) ->
    Res = #sm_a{xmlns = Xmlns, h = H},
    send(State, Res).

-spec handle_a(state(), sm_a()) -> state().
handle_a(State, #sm_a{h = H}) ->
    State1 = check_h_attribute(State, H),
    resend_rack(State1).

-spec handle_resume(state(), sm_resume()) -> {ok, state()} | {error, state()}.
handle_resume(#{user := User, lserver := LServer, sockmod := SockMod,
		lang := Lang, socket := Socket} = State,
	      #sm_resume{h = H, previd = PrevID, xmlns = Xmlns}) ->
    R = case inherit_session_state(State, PrevID) of
	    {ok, InheritedState} ->
		{ok, InheritedState, H};
	    {error, Err, InH} ->
		{error, #sm_failed{reason = 'item-not-found',
				   text = xmpp:mk_text(Err, Lang),
				   h = InH, xmlns = Xmlns}, Err};
	    {error, Err} ->
		{error, #sm_failed{reason = 'item-not-found',
				   text = xmpp:mk_text(Err, Lang),
				   xmlns = Xmlns}, Err}
	end,
    case R of
	{ok, #{jid := JID} = ResumedState, NumHandled} ->
	    State1 = check_h_attribute(ResumedState, NumHandled),
	    #{mgmt_xmlns := AttrXmlns, mgmt_stanzas_in := AttrH} = State1,
	    AttrId = make_resume_id(State1),
	    State2 = send(State1, #sm_resumed{xmlns = AttrXmlns,
					      h = AttrH,
					      previd = AttrId}),
	    State3 = resend_unacked_stanzas(State2),
	    State4 = send(State3, #sm_r{xmlns = AttrXmlns}),
	    %% TODO: move this to mod_client_state
	    %% csi_flush_queue(State4),
	    State5 = ejabberd_hooks:run_fold(c2s_session_resumed, LServer, State4, []),
	    ?INFO_MSG("(~s) Resumed session for ~s",
		      [SockMod:pp(Socket), jid:to_string(JID)]),
	    {ok, State5};
	{error, El, Msg} ->
	    ?INFO_MSG("Cannot resume session for ~s@~s: ~s",
		      [User, LServer, Msg]),
	    {error, send(State, El)}
    end.

-spec transition_to_pending(state()) -> state().
transition_to_pending(#{mgmt_state := active, mod := Mod,
			mgmt_timeout := 0} = State) ->
    Mod:stop(State);
transition_to_pending(#{mgmt_state := active, jid := JID,
			lserver := LServer, mgmt_timeout := Timeout} = State) ->
    State1 = cancel_ack_timer(State),
    ?INFO_MSG("Waiting for resumption of stream for ~s", [jid:to_string(JID)]),
    erlang:start_timer(timer:seconds(Timeout), self(), pending_timeout),
    State2 = State1#{mgmt_state => pending},
    ejabberd_hooks:run_fold(c2s_session_pending, LServer, State2, []);
transition_to_pending(State) ->
    State.

-spec check_h_attribute(state(), non_neg_integer()) -> state().
check_h_attribute(#{mgmt_stanzas_out := NumStanzasOut, jid := JID} = State, H)
  when H > NumStanzasOut ->
    ?DEBUG("~s acknowledged ~B stanzas, but only ~B were sent",
	   [jid:to_string(JID), H, NumStanzasOut]),
    mgmt_queue_drop(State#{mgmt_stanzas_out => H}, NumStanzasOut);
check_h_attribute(#{mgmt_stanzas_out := NumStanzasOut, jid := JID} = State, H) ->
    ?DEBUG("~s acknowledged ~B of ~B stanzas",
	   [jid:to_string(JID), H, NumStanzasOut]),
    mgmt_queue_drop(State, H).

-spec update_num_stanzas_in(state(), xmpp_element()) -> state().
update_num_stanzas_in(#{mgmt_state := MgmtState,
			mgmt_stanzas_in := NumStanzasIn} = State, El)
  when MgmtState == active; MgmtState == pending ->
    NewNum = case {xmpp:is_stanza(El), NumStanzasIn} of
		 {true, 4294967295} ->
		     0;
		 {true, Num} ->
		     Num + 1;
		 {false, Num} ->
		     Num
	     end,
    State#{mgmt_stanzas_in => NewNum};
update_num_stanzas_in(State, _El) ->
    State.

send_rack(#{mgmt_ack_timer := _} = State) ->
    State;
send_rack(#{mgmt_xmlns := Xmlns,
	    mgmt_stanzas_out := NumStanzasOut,
	    mgmt_ack_timeout := AckTimeout} = State) ->
    TRef = erlang:start_timer(AckTimeout, self(), ack_timeout),
    State1 = State#{mgmt_ack_timer => TRef, mgmt_stanzas_req => NumStanzasOut},
    send(State1, #sm_r{xmlns = Xmlns}).

resend_rack(#{mgmt_ack_timer := _,
	      mgmt_queue := Queue,
	      mgmt_stanzas_out := NumStanzasOut,
	      mgmt_stanzas_req := NumStanzasReq} = State) ->
    State1 = cancel_ack_timer(State),
    case NumStanzasReq < NumStanzasOut andalso not queue_is_empty(Queue) of
	true -> send_rack(State1);
	false -> State1
    end;
resend_rack(State) ->
    State.

-spec mgmt_queue_add(state(), xmpp_element()) -> state().
mgmt_queue_add(#{mgmt_stanzas_out := NumStanzasOut,
		 mgmt_queue := Queue} = State, Pkt) ->
    NewNum = case NumStanzasOut of
		 4294967295 -> 0;
		 Num -> Num + 1
	     end,
    Queue1 = queue_in({NewNum, p1_time_compat:timestamp(), Pkt}, Queue),
    State1 = State#{mgmt_queue => Queue1, mgmt_stanzas_out => NewNum},
    check_queue_length(State1).

-spec mgmt_queue_drop(state(), non_neg_integer()) -> state().
mgmt_queue_drop(#{mgmt_queue := Queue} = State, NumHandled) ->
    NewQueue = queue_dropwhile(
		 fun({N, _T, _E}) -> N =< NumHandled end, Queue),
    State#{mgmt_queue => NewQueue}.

-spec check_queue_length(state()) -> state().
check_queue_length(#{mgmt_max_queue := Limit} = State)
  when Limit == infinity; Limit == exceeded ->
    State;
check_queue_length(#{mgmt_queue := Queue, mgmt_max_queue := Limit} = State) ->
    case queue_len(Queue) > Limit of
	true ->
	    State#{mgmt_max_queue => exceeded};
	false ->
	    State
    end.

-spec resend_unacked_stanzas(state()) -> state().
resend_unacked_stanzas(#{mgmt_state := MgmtState,
			 mgmt_queue := {QueueLen, _} = Queue,
			 jid := JID} = State)
  when (MgmtState == active orelse
	MgmtState == pending orelse
	MgmtState == timeout) andalso QueueLen > 0 ->
    ?DEBUG("Resending ~B unacknowledged stanza(s) to ~s",
	   [QueueLen, jid:to_string(JID)]),
    queue_foldl(
      fun({_, Time, Pkt}, AccState) ->
	      NewPkt = add_resent_delay_info(AccState, Pkt, Time),
	      send(AccState, NewPkt)
      end, State, Queue);
resend_unacked_stanzas(State) ->
    State.

-spec route_unacked_stanzas(state()) -> ok.
route_unacked_stanzas(#{mgmt_state := MgmtState,
			mgmt_resend := MgmtResend,
			lang := Lang, user := User,
			jid := JID, lserver := LServer,
			mgmt_queue := {QueueLen, _} = Queue,
			resource := Resource} = State)
  when (MgmtState == active orelse
	MgmtState == pending orelse
	MgmtState == timeout) andalso QueueLen > 0 ->
    ResendOnTimeout = case MgmtResend of
			  Resend when is_boolean(Resend) ->
			      Resend;
			  if_offline ->
			      case ejabberd_sm:get_user_resources(User, Resource) of
				  [Resource] ->
				      %% Same resource opened new session
				      true;
				  [] -> true;
				  _ -> false
			      end
		      end,
    ?DEBUG("Re-routing ~B unacknowledged stanza(s) to ~s",
	   [QueueLen, jid:to_string(JID)]),
    queue_foreach(
      fun({_, _Time, #presence{from = From}}) ->
	      ?DEBUG("Dropping presence stanza from ~s", [jid:to_string(From)]);
	 ({_, _Time, #iq{} = El}) ->
	      Txt = <<"User session terminated">>,
	      route_error(El, xmpp:err_service_unavailable(Txt, Lang));
	 ({_, _Time, #message{from = From, meta = #{carbon_copy := true}}}) ->
	      %% XEP-0280 says: "When a receiving server attempts to deliver a
	      %% forked message, and that message bounces with an error for
	      %% any reason, the receiving server MUST NOT forward that error
	      %% back to the original sender."  Resending such a stanza could
	      %% easily lead to unexpected results as well.
	      ?DEBUG("Dropping forwarded message stanza from ~s",
		     [jid:to_string(From)]);
	 ({_, Time, #message{} = Msg}) ->
	      case ejabberd_hooks:run_fold(message_is_archived,
					   LServer, false,
					   [State, Msg]) of
		  true ->
		      ?DEBUG("Dropping archived message stanza from ~s",
			     [jid:to_string(xmpp:get_from(Msg))]);
		  false when ResendOnTimeout ->
		      NewEl = add_resent_delay_info(State, Msg, Time),
		      route(NewEl);
		  false ->
		      Txt = <<"User session terminated">>,
		      route_error(Msg, xmpp:err_service_unavailable(Txt, Lang))
	      end;
	 ({_, _Time, El}) ->
	      %% Raw element of type 'error' resulting from a validation error
	      %% We cannot pass it to the router, it will generate an error
	      ?DEBUG("Do not route raw element from ack queue: ~p", [El])
      end, Queue);
route_unacked_stanzas(_State) ->
    ok.

-spec inherit_session_state(state(), binary()) -> {ok, state()} |
						  {error, binary()} |
						  {error, binary(), non_neg_integer()}.
inherit_session_state(#{user := U, server := S} = State, ResumeID) ->
    case jlib:base64_to_term(ResumeID) of
	{term, {R, Time}} ->
	    case ejabberd_sm:get_session_pid(U, S, R) of
		none ->
		    case ejabberd_sm:get_offline_info(Time, U, S, R) of
			none ->
			    {error, <<"Previous session PID not found">>};
			Info ->
			    case proplists:get_value(num_stanzas_in, Info) of
				undefined ->
				    {error, <<"Previous session timed out">>};
				H ->
				    {error, <<"Previous session timed out">>, H}
			    end
		    end;
		OldPID ->
		    OldSID = {Time, OldPID},
		    try resume_session(OldSID, State) of
			{resume, #{mgmt_xmlns := Xmlns,
				   mgmt_queue := Queue,
				   mgmt_timeout := Timeout,
				   mgmt_stanzas_in := NumStanzasIn,
				   mgmt_stanzas_out := NumStanzasOut} = OldState} ->
			    State1 = ejabberd_c2s:copy_state(State, OldState),
			    State2 = State1#{mgmt_xmlns => Xmlns,
					     mgmt_queue => Queue,
					     mgmt_timeout => Timeout,
					     mgmt_stanzas_in => NumStanzasIn,
					     mgmt_stanzas_out => NumStanzasOut,
					     mgmt_state => active},
			    ejabberd_sm:close_session(OldSID, U, S, R),
			    State3 = ejabberd_c2s:open_session(State2),
			    ejabberd_c2s:stop(OldPID),
			    {ok, State3};
			{error, Msg} ->
			    {error, Msg}
		    catch exit:{noproc, _} ->
			    {error, <<"Previous session PID is dead">>};
			  exit:{timeout, _} ->
			    {error, <<"Session state copying timed out">>}
		    end
	    end;
	_ ->
	    {error, <<"Invalid 'previd' value">>}
    end.

-spec resume_session({integer(), pid()}, state()) -> {resume, state()} |
						     {error, binary()}.
resume_session({Time, Pid}, _State) ->
    ejabberd_c2s:call(Pid, {resume_session, Time}, timer:seconds(15)).

-spec make_resume_id(state()) -> binary().
make_resume_id(#{sid := {Time, _}, resource := Resource}) ->
    jlib:term_to_base64({Resource, Time}).

-spec add_resent_delay_info(state(), stanza(), erlang:timestamp()) -> stanza();
			   (state(), xmlel(), erlang:timestamp()) -> xmlel().
add_resent_delay_info(#{lserver := LServer}, El, Time)
  when is_record(El, message); is_record(El, presence) ->
    xmpp_util:add_delay_info(El, jid:make(LServer), Time, <<"Resent">>);
add_resent_delay_info(_State, El, _Time) ->
    El.

-spec route(stanza()) -> ok.
route(Pkt) ->
    From = xmpp:get_from(Pkt),
    To = xmpp:get_to(Pkt),
    ejabberd_router:route(From, To, Pkt).

-spec route_error(stanza(), stanza_error()) -> ok.
route_error(Pkt, Err) ->
    From = xmpp:get_from(Pkt),
    To = xmpp:get_to(Pkt),
    ejabberd_router:route_error(To, From, Pkt, Err).

-spec send(state(), xmpp_element()) -> state().
send(#{mod := Mod} = State, Pkt) ->
    Mod:send(State, Pkt).

-spec queue_new() -> lqueue().
queue_new() ->
    {0, queue:new()}.

-spec queue_in(term(), lqueue()) -> lqueue().
queue_in(Elem, {N, Q}) ->
    {N+1, queue:in(Elem, Q)}.

-spec queue_len(lqueue()) -> non_neg_integer().
queue_len({N, _}) ->
    N.

-spec queue_foldl(fun((term(), T) -> T), T, lqueue()) -> T.
queue_foldl(F, Acc, {_N, Q}) ->
    jlib:queue_foldl(F, Acc, Q).

-spec queue_foreach(fun((_) -> _), lqueue()) -> ok.
queue_foreach(F, {_N, Q}) ->
    jlib:queue_foreach(F, Q).

-spec queue_dropwhile(fun((term()) -> boolean()), lqueue()) -> lqueue().
queue_dropwhile(F, {N, Q}) ->
    case queue:peek(Q) of
	{value, Item} ->
	    case F(Item) of
		true ->
		    queue_dropwhile(F, {N-1, queue:drop(Q)});
		false ->
		    {N, Q}
	    end;
	empty ->
	    {N, Q}
    end.

-spec queue_is_empty(lqueue()) -> boolean().
queue_is_empty({N, _Q}) ->
    N == 0.

-spec cancel_ack_timer(state()) -> state().
cancel_ack_timer(#{mgmt_ack_timer := TRef} = State) ->
    case erlang:cancel_timer(TRef) of
        false -> 
            receive {timeout, TRef, _} -> ok
            after 0 -> ok
            end;
        _ ->
            ok
    end,
    maps:remove(mgmt_ack_timer, State);
cancel_ack_timer(State) ->
    State.

-spec bounce_message_queue() -> ok.
bounce_message_queue() ->
    receive {route, From, To, Pkt} ->
	    ejabberd_router:route(From, To, Pkt),
	    bounce_message_queue()
    after 0 ->
	    ok
    end.

%%%===================================================================
%%% Configuration processing
%%%===================================================================
get_max_ack_queue(Host, Opts) ->
    VFun = mod_opt_type(max_ack_queue),
    case gen_mod:get_module_opt(Host, ?MODULE, max_ack_queue, VFun) of
	undefined -> gen_mod:get_opt(max_ack_queue, Opts, VFun, 1000);
	Limit -> Limit
    end.

get_resume_timeout(Host, Opts) ->
    VFun = mod_opt_type(resume_timeout),
    case gen_mod:get_module_opt(Host, ?MODULE, resume_timeout, VFun) of
	undefined -> gen_mod:get_opt(resume_timeout, Opts, VFun, 300);
	Timeout -> Timeout
    end.

get_max_resume_timeout(Host, Opts, ResumeTimeout) ->
    VFun = mod_opt_type(max_resume_timeout),
    case gen_mod:get_module_opt(Host, ?MODULE, max_resume_timeout, VFun) of
	undefined ->
	    case gen_mod:get_opt(max_resume_timeout, Opts, VFun) of
		undefined -> ResumeTimeout;
		Max when Max >= ResumeTimeout -> Max;
		_ -> ResumeTimeout
	    end;
	Max when Max >= ResumeTimeout -> Max;
	_ -> ResumeTimeout
    end.

get_ack_timeout(Host, Opts) ->
    VFun = mod_opt_type(ack_timeout),
    T = case gen_mod:get_module_opt(Host, ?MODULE, ack_timeout, VFun) of
	    undefined -> gen_mod:get_opt(ack_timeout, Opts, VFun, 60);
	    AckTimeout -> AckTimeout
	end,
    case T of
	infinity -> infinity;
	_ -> timer:seconds(T)
    end.

get_resend_on_timeout(Host, Opts) ->
    VFun = mod_opt_type(resend_on_timeout),
    case gen_mod:get_module_opt(Host, ?MODULE, resend_on_timeout, VFun) of
	undefined -> gen_mod:get_opt(resend_on_timeout, Opts, VFun, false);
	Resend -> Resend
    end.

mod_opt_type(max_ack_queue) ->
    fun(I) when is_integer(I), I > 0 -> I;
       (infinity) -> infinity
    end;
mod_opt_type(resume_timeout) ->
    fun(I) when is_integer(I), I >= 0 -> I end;
mod_opt_type(max_resume_timeout) ->
    fun(I) when is_integer(I), I >= 0 -> I end;
mod_opt_type(ack_timeout) ->
    fun(I) when is_integer(I), I > 0 -> I;
       (infinity) -> infinity
    end;
mod_opt_type(resend_on_timeout) ->
    fun(B) when is_boolean(B) -> B;
       (if_offline) -> if_offline
    end;
mod_opt_type(_) ->
    [max_ack_queue, resume_timeout, max_resume_timeout, ack_timeout,
     resend_on_timeout].