The second release of mod_mam

apps/ejabberd/include/jlib.hrl

@@ -69,6 +69,7 @@
 -define(NS_BYTESTREAMS, <<"http://jabber.org/protocol/bytestreams">>).
 -define(NS_ADMIN,       <<"http://jabber.org/protocol/admin">>).
 -define(NS_SERVERINFO,  <<"http://jabber.org/network/serverinfo">>).
+-define(NS_MAM,         <<"urn:xmpp:mam:tmp">>).
 
 -define(NS_RSM,         <<"http://jabber.org/protocol/rsm">>).
 -define(NS_EJABBERD_CONFIG,<<"ejabberd:config">>).

apps/ejabberd/priv/cassandra.cql

@@ -0,0 +1,25 @@
+CREATE KEYSPACE mam
+         WITH REPLICATION = {'class' : 'SimpleStrategy', 'replication_factor': 3};
+USE mam;
+
+CREATE TABLE mam_muc_message (
+  id bigint,
+  room_id int,
+  nick_name varchar,
+  message blob,
+  PRIMARY KEY (room_id, id));
+
+CREATE TABLE mam_con_message(
+  id bigint,
+  lower_jid varchar,
+  upper_jid varchar,
+  is_from_lower boolean,
+  message blob,
+  PRIMARY KEY ((lower_jid, upper_jid), id));
+
+CREATE TABLE mam_con_user(
+  local_jid varchar,
+  remote_jid varchar,
+  last_message_id bigint,
+  PRIMARY KEY (local_jid, remote_jid));
+

apps/ejabberd/priv/mysql.sql

@@ -266,10 +266,20 @@ CREATE INDEX i_mam_config USING HASH ON mam_config(user_id, remote_jid);
 CREATE TABLE mam_user(
   id INT UNSIGNED NOT NULL AUTO_INCREMENT,
   user_name varchar(250) CHARACTER SET binary NOT NULL,
-  PRIMARY KEY(id) USING HASH
+  PRIMARY KEY(id) USING HASH,
+  CONSTRAINT uc_mam_user_name UNIQUE (user_name)
 );
 CREATE INDEX i_mam_user_name USING HASH ON mam_user(user_name);
 
+CREATE TABLE mam_server_user(
+  id INT UNSIGNED NOT NULL AUTO_INCREMENT,
+  server    varchar(250) CHARACTER SET binary NOT NULL,
+  user_name varchar(250) CHARACTER SET binary NOT NULL,
+  PRIMARY KEY(id) USING HASH,
+  CONSTRAINT uc_mam_server_user_name UNIQUE (server, user_name)
+);
+CREATE INDEX i_mam_server_user_name USING HASH ON mam_server_user(server, user_name);
+
 
 CREATE TABLE mam_muc_message(
   -- Message UID

apps/ejabberd/src/ejabberd_odbc.erl

@@ -210,6 +210,10 @@ to_bool(_) -> false.
 %%% Callback functions from gen_fsm
 %%%----------------------------------------------------------------------
 init([Host, StartInterval, ParentPid, Dedicated]) ->
+    %% For debugging and introspection only.
+    put(mim_host, Host),
+    put(mim_process_type, odbc_worker),
+    put(mim_odbc_dedicated, Dedicated),
     case ejabberd_config:get_local_option({odbc_keepalive_interval, Host}) of
 	KeepaliveInterval when is_integer(KeepaliveInterval) ->
 	    timer:apply_interval(KeepaliveInterval*1000, ?MODULE,

apps/ejabberd/src/ejabberd_router.erl

@@ -222,6 +222,7 @@ code_change(_OldVsn, State, _Extra) ->
 do_route(OrigFrom, OrigTo, OrigPacket) ->
     ?DEBUG("route~n\tfrom ~p~n\tto ~p~n\tpacket ~p~n",
            [OrigFrom, OrigTo, OrigPacket]),
+    %% Filter globally
     case ejabberd_hooks:run_fold(filter_packet,
                                  {OrigFrom, OrigTo, OrigPacket}, []) of
         {From, To, Packet} ->
@@ -230,12 +231,25 @@ do_route(OrigFrom, OrigTo, OrigPacket) ->
                 [] ->
                     ejabberd_s2s:route(From, To, Packet);
                 [#route{handler=Handler}] ->
-                    case Handler of
-                        {apply_fun, Fun} ->
-                            Fun(From, To, Packet);
-                        {apply, Module, Function} ->
-                            Module:Function(From, To, Packet)
-                    end
+                    do_local_route(OrigFrom, OrigTo, OrigPacket, LDstDomain, Handler)
+            end;
+        drop ->
+            ejabberd_hooks:run(xmpp_stanza_dropped, 
+                               OrigFrom#jid.lserver,
+                               [OrigFrom, OrigTo, OrigPacket]),
+            ok
+    end.
+
+do_local_route(OrigFrom, OrigTo, OrigPacket, LDstDomain, Handler) ->
+    %% Filter locally
+    case ejabberd_hooks:run_fold(filter_local_packet, LDstDomain,
+                                 {OrigFrom, OrigTo, OrigPacket}, []) of
+        {From, To, Packet} ->
+            case Handler of
+                {apply_fun, Fun} ->
+                    Fun(From, To, Packet);
+                {apply, Module, Function} ->
+                    Module:Function(From, To, Packet)
             end;
         drop ->
             ejabberd_hooks:run(xmpp_stanza_dropped, 

apps/ejabberd/src/mod_mam.erl

@@ -114,10 +114,6 @@
 %% ----------------------------------------------------------------------
 %% Constants
 
-mam_ns_string() -> "urn:xmpp:mam:tmp".
-
-mam_ns_binary() -> <<"urn:xmpp:mam:tmp">>.
-
 default_result_limit() -> 50.
 
 max_result_limit() -> 50.
@@ -236,21 +232,21 @@ start(Host, Opts) ->
     ejabberd_users:start(Host),
     %% `parallel' is the only one recommended here.
     IQDisc = gen_mod:get_opt(iqdisc, Opts, parallel), %% Type
-    mod_disco:register_feature(Host, mam_ns_binary()),
-    gen_iq_handler:add_iq_handler(ejabberd_sm, Host, mam_ns_binary(),
+    mod_disco:register_feature(Host, ?NS_MAM),
+    gen_iq_handler:add_iq_handler(ejabberd_sm, Host, ?NS_MAM,
                                   ?MODULE, process_mam_iq, IQDisc),
     ejabberd_hooks:add(user_send_packet, Host, ?MODULE, user_send_packet, 90),
-    ejabberd_hooks:add(filter_packet, global, ?MODULE, filter_packet, 90),
+    ejabberd_hooks:add(filter_local_packet, Host, ?MODULE, filter_packet, 90),
     ejabberd_hooks:add(remove_user, Host, ?MODULE, remove_user, 50),
     ok.
 
 stop(Host) ->
     ?DEBUG("mod_mam stopping", []),
     ejabberd_hooks:delete(user_send_packet, Host, ?MODULE, user_send_packet, 90),
-    ejabberd_hooks:delete(filter_packet, global, ?MODULE, filter_packet, 90),
+    ejabberd_hooks:delete(filter_local_packet, Host, ?MODULE, filter_packet, 90),
     ejabberd_hooks:delete(remove_user, Host, ?MODULE, remove_user, 50),
-    gen_iq_handler:remove_iq_handler(ejabberd_sm, Host, mam_ns_string()),
-    mod_disco:unregister_feature(Host, mam_ns_binary()),
+    gen_iq_handler:remove_iq_handler(ejabberd_sm, Host, ?NS_MAM),
+    mod_disco:unregister_feature(Host, ?NS_MAM),
     ok.
 
 %% ----------------------------------------------------------------------
@@ -271,14 +267,8 @@ process_mam_iq(From=#jid{lserver=Host}, To, IQ) ->
         true  -> 
             case wait_shaper(Host, Action, From) of
                 ok ->
-                    try
-                        handle_mam_iq(Action, From, To, IQ)
-                    catch error:Reason ->
-                        ejabberd_hooks:run(mam_drop_iq, Host,
-                            [Host, To, IQ, Action, Reason]),
-                        lager:error("Action ~p failed ~p", [Action, Reason]),
-                        return_error_iq(IQ, Reason)
-                    end;
+                    handle_error_iq(Host, To, Action,
+                        handle_mam_iq(Action, From, To, IQ));
                 {error, max_delay_reached} ->
                     ejabberd_hooks:run(mam_drop_iq, Host,
                         [Host, To, IQ, Action, max_delay_reached]),
@@ -399,19 +389,29 @@ handle_set_prefs(ArcJID=#jid{},
               [DefaultMode, AlwaysJIDs, NeverJIDs]),
     Host = server_host(ArcJID),
     ArcID = archive_id_int(Host, ArcJID),
-    set_prefs(Host, ArcID, ArcJID, DefaultMode, AlwaysJIDs, NeverJIDs),
+    Res = set_prefs(Host, ArcID, ArcJID, DefaultMode, AlwaysJIDs, NeverJIDs),
+    handle_set_prefs_result(Res, DefaultMode, AlwaysJIDs, NeverJIDs, IQ).
+
+handle_set_prefs_result(ok, DefaultMode, AlwaysJIDs, NeverJIDs, IQ) ->
     ResultPrefsEl = result_prefs(DefaultMode, AlwaysJIDs, NeverJIDs),
-    IQ#iq{type = result, sub_el = [ResultPrefsEl]}.
+    IQ#iq{type = result, sub_el = [ResultPrefsEl]};
+handle_set_prefs_result({error, Reason},
+                        _DefaultMode, _AlwaysJIDs, _NeverJIDs, IQ) ->
+    return_error_iq(IQ, Reason).
 
 handle_get_prefs(ArcJID=#jid{}, IQ=#iq{}) ->
     Host = server_host(ArcJID),
     ArcID = archive_id_int(Host, ArcJID),
-    {DefaultMode, AlwaysJIDs, NeverJIDs} =
-        get_prefs(Host, ArcID, ArcJID, always),
+    Res = get_prefs(Host, ArcID, ArcJID, always),
+    handle_get_prefs_result(Res, IQ).
+
+handle_get_prefs_result({DefaultMode, AlwaysJIDs, NeverJIDs}, IQ) ->
     ?DEBUG("Extracted data~n\tDefaultMode ~p~n\tAlwaysJIDs ~p~n\tNeverJIDS ~p~n",
               [DefaultMode, AlwaysJIDs, NeverJIDs]),
     ResultPrefsEl = result_prefs(DefaultMode, AlwaysJIDs, NeverJIDs),
-    IQ#iq{type = result, sub_el = [ResultPrefsEl]}.
+    IQ#iq{type = result, sub_el = [ResultPrefsEl]};
+handle_get_prefs_result({error, Reason}, IQ) ->
+    return_error_iq(IQ, Reason).
 
 handle_lookup_messages(
         From=#jid{},
@@ -442,6 +442,8 @@ handle_lookup_messages(
         ErrorEl = ?STANZA_ERRORT(<<"">>, <<"modify">>, <<"policy-violation">>,
                                  <<"en">>, <<"Too many results">>),          
         IQ#iq{type = error, sub_el = [ErrorEl]};
+    {error, Reason} ->
+        return_error_iq(IQ, Reason);
     {ok, {TotalCount, Offset, MessageRows}} ->
         {FirstMessID, LastMessID} =
             case MessageRows of
@@ -473,8 +475,9 @@ handle_purge_multiple_messages(ArcJID=#jid{},
     Borders = borders_decode(PurgeEl),
     %% Filtering by contact.
     With  = elem_to_with_jid(PurgeEl),
-    purge_multiple_messages(Host, ArcID, ArcJID, Borders, Start, End, Now, With),
-    return_purge_success(IQ).
+    Res = purge_multiple_messages(Host, ArcID, ArcJID, Borders,
+                                  Start, End, Now, With),
+    return_purge_multiple_message_iq(IQ, Res).
 
 handle_purge_single_message(ArcJID=#jid{},
                             IQ=#iq{sub_el = PurgeEl}) ->
@@ -550,9 +553,10 @@ set_prefs(Host, ArcID, ArcJID, DefaultMode, AlwaysJIDs, NeverJIDs) ->
     ArcJID      :: jid(),
     DefaultMode :: archive_behaviour(),
     GlobalDefaultMode :: archive_behaviour(),
-    Result      :: {DefaultMode, AlwaysJIDs, NeverJIDs},
+    Result      :: {DefaultMode, AlwaysJIDs, NeverJIDs} | {error, Reason},
     AlwaysJIDs  :: [literal_jid()],
-    NeverJIDs   :: [literal_jid()].
+    NeverJIDs   :: [literal_jid()],
+    Reason      :: term().
 get_prefs(Host, ArcID, ArcJID, GlobalDefaultMode) ->
     ejabberd_hooks:run_fold(mam_get_prefs, Host,
         {GlobalDefaultMode, [], []},
@@ -565,7 +569,9 @@ remove_archive(Host, ArcID, ArcJID=#jid{}) ->
 -spec lookup_messages(Host, ArcID, ArcJID, RSM, Borders,
                       Start, End, Now, WithJID,
                       PageSize, LimitPassed, MaxResultLimit, IsSimple) ->
-    {ok, {TotalCount, Offset, MessageRows}} | {error, 'policy-violation'}
+    {ok, {TotalCount, Offset, MessageRows}}
+    | {error, 'policy-violation'}
+    | {error, Reason}
     when
     Host    :: server_host(),
     ArcID   :: archive_id(),
@@ -582,7 +588,8 @@ remove_archive(Host, ArcID, ArcJID=#jid{}) ->
     IsSimple :: boolean() | opt_count,
     TotalCount :: non_neg_integer(),
     Offset  :: non_neg_integer(),
-    MessageRows :: list(tuple()).
+    MessageRows :: list(tuple()),
+    Reason :: term().
 lookup_messages(Host, ArcID, ArcJID, RSM, Borders, Start, End, Now,
                 WithJID, PageSize, LimitPassed, MaxResultLimit, IsSimple) ->
     ejabberd_hooks:run_fold(mam_lookup_messages, Host, {ok, {0, 0, []}},
@@ -604,20 +611,20 @@ archive_message(Host, MessID, ArcID, LocJID, RemJID, SrcJID, Dir, Packet) ->
     ejabberd_hooks:run_fold(mam_archive_message, Host, ok,
         [Host, MessID, ArcID, LocJID, RemJID, SrcJID, Dir, Packet]).
 
-
 -spec purge_single_message(Host, MessID, ArcID, ArcJID, Now) ->
-    ok | {error, 'not-found'} when
+    ok | {error, 'not-found'} | {error, Reason} when
     Host   :: server_host(),
     MessID :: message_id(),
     ArcID  :: archive_id(),
     ArcJID :: jid(),
-    Now :: unix_timestamp().
+    Now    :: unix_timestamp(),
+    Reason :: term().
 purge_single_message(Host, MessID, ArcID, ArcJID, Now) ->
     ejabberd_hooks:run_fold(mam_purge_single_message, Host, ok,
         [Host, MessID, ArcID, ArcJID, Now]).
 
 -spec purge_multiple_messages(Host, ArcID, ArcJID, Borders,
-                              Start, End, Now, WithJID) -> ok
+                              Start, End, Now, WithJID) -> ok | {error, Reason}
     when
     Host    :: server_host(),
     ArcID   :: archive_id(),
@@ -626,7 +633,8 @@ purge_single_message(Host, MessID, ArcID, ArcJID, Now) ->
     Start   :: unix_timestamp() | undefined,
     End     :: unix_timestamp() | undefined,
     Now     :: unix_timestamp(),
-    WithJID :: jid() | undefined.
+    WithJID :: jid() | undefined,
+    Reason  :: term().
 purge_multiple_messages(Host, ArcID, ArcJID, Borders, Start, End, Now, WithJID) ->
     ejabberd_hooks:run_fold(mam_purge_multiple_messages, Host, ok,
         [Host, ArcID, ArcJID, Borders, Start, End, Now, WithJID]).
@@ -688,14 +696,38 @@ elem_to_limit(QueryEl) ->
         [{elem, <<"set">>}, {elem, <<"limit">>}, cdata]
     ]).
 
-return_purge_success(IQ) ->
-    IQ#iq{type = result, sub_el = []}.
+handle_error_iq(Host, To, Action, IQ=#iq{type = {error, Reason}}) ->
+    ejabberd_hooks:run(mam_drop_iq, Host,
+        [Host, To, IQ, Action, Reason]),
+    IQ#iq{type = error};
+handle_error_iq(_Host, _To, _Action, IQ) ->
+    IQ.
+
+return_error_iq(IQ, timeout) ->
+    IQ#iq{type = {error, timeout}, sub_el = [?ERR_SERVICE_UNAVAILABLE]};
+return_error_iq(IQ, Reason) ->
+    IQ#iq{type = {error, Reason}, sub_el = [?ERR_INTERNAL_SERVER_ERROR]}.
 
 return_action_not_allowed_error_iq(IQ) ->
     ErrorEl = ?STANZA_ERRORT(<<"">>, <<"cancel">>, <<"not-allowed">>,
          <<"en">>, <<"The action is not allowed.">>),
     IQ#iq{type = error, sub_el = [ErrorEl]}.
 
+return_purge_multiple_message_iq(IQ, ok) ->
+    return_purge_success(IQ);
+return_purge_multiple_message_iq(IQ, {error, Reason}) ->
+    return_error_iq(IQ, Reason).
+
+return_purge_single_message_iq(IQ, ok) ->
+    return_purge_success(IQ);
+return_purge_single_message_iq(IQ, {error, 'not-found'}) ->
+    return_purge_not_found_error_iq(IQ);
+return_purge_single_message_iq(IQ, {error, Reason}) ->
+    return_error_iq(IQ, Reason).
+
+return_purge_success(IQ) ->
+    IQ#iq{type = result, sub_el = []}.
+
 return_purge_not_found_error_iq(IQ) ->
     %% Message not found.
     ErrorEl = ?STANZA_ERRORT(<<"">>, <<"cancel">>, <<"item-not-found">>,
@@ -707,14 +739,3 @@ return_max_delay_reached_error_iq(IQ) ->
     ErrorEl = ?ERRT_RESOURCE_CONSTRAINT(
         <<"en">>, <<"The action is cancelled because of flooding.">>),
     IQ#iq{type = error, sub_el = [ErrorEl]}.
-
-return_error_iq(IQ, timeout) ->
-    IQ#iq{type = error, sub_el = [?ERR_SERVICE_UNAVAILABLE]};
-return_error_iq(IQ, _Reason) ->
-    IQ#iq{type = error, sub_el = [?ERR_INTERNAL_SERVER_ERROR]}.
-
-return_purge_single_message_iq(IQ, ok) ->
-    return_purge_success(IQ);
-return_purge_single_message_iq(IQ, {error, 'not-found'}) ->
-    return_purge_not_found_error_iq(IQ).
-

apps/ejabberd/src/mod_mam_cache_user.erl

@@ -5,7 +5,9 @@
 %%% This module is a proxy for `mod_mam_odbc_user' (it should be started).
 %%%
 %%% There are 2 hooks for `mam_archive_id':
-%%% `cache_archive_id/3' and `store_archive_id/3'.
+%%% `cached_archive_id/3' and `store_archive_id/3'.
+%%%
+%%% This module supports several hosts.
 %%% 
 %%% @end
 %%%-------------------------------------------------------------------
@@ -53,7 +55,7 @@ su_key(#jid{lserver = LServer, luser = LUser}) ->
 %% Starting and stopping functions for users' archives
 
 start(Host, Opts) ->
-    supervisor:start_child(ejabberd_sup, writer_child_spec()),
+    start_server(Host),
     case gen_mod:get_module_opt(Host, ?MODULE, pm, false) of
         true ->
             start_pm(Host, Opts);
@@ -68,6 +70,7 @@ start(Host, Opts) ->
     end.
 
 stop(Host) ->
+    stop_server(Host),
     case gen_mod:get_module_opt(Host, ?MODULE, pm, false) of
         true ->
             stop_pm(Host);
@@ -82,12 +85,18 @@ stop(Host) ->
     end.
 
 writer_child_spec() ->
-    {mod_mam_cache_user,
-     {mod_mam_cache_user, start_link, []},
+    {?MODULE,
+     {?MODULE, start_link, []},
      permanent,
      5000,
      worker,
-     [mod_mam_cache_user]}.
+     [?MODULE]}.
+
+start_server(_Host) ->
+    supervisor:start_child(ejabberd_sup, writer_child_spec()).
+
+stop_server(_Host) ->
+    ok.
 
 %% ----------------------------------------------------------------------
 %% Add hooks for mod_mam