xmpp.chapril.org-ejabberd/src/mod_pubsub/pubsub_odbc.patch

--- mod_pubsub.erl	2010-05-28 13:16:05.000000000 +0200
+++ mod_pubsub_odbc.erl	2010-05-28 13:17:04.000000000 +0200
@@ -42,7 +42,7 @@
 %%% 6.2.3.1, 6.2.3.5, and 6.3. For information on subscription leases see
 %%% XEP-0060 section 12.18.
 
--module(mod_pubsub).
+-module(mod_pubsub_odbc).
 -author('christophe.romain@process-one.net').
 -version('1.13-0').
 
@@ -55,9 +55,9 @@
 -include("adhoc.hrl").
 -include("pubsub.hrl").
 
--define(STDTREE, "tree").
--define(STDNODE, "flat").
--define(PEPNODE, "pep").
+-define(STDTREE, "tree_odbc").
+-define(STDNODE, "flat_odbc").
+-define(PEPNODE, "pep_odbc").
 
 %% exports for hooks
 -export([presence_probe/3,
@@ -105,7 +105,7 @@
 	 string_to_affiliation/1,
 	 extended_error/2,
 	 extended_error/3,
-	 rename_default_nodeplugin/0
+	 escape/1
 	]).
 
 %% API and gen_server callbacks
@@ -124,7 +124,7 @@
 -export([send_loop/1
 	]).
 
--define(PROCNAME, ejabberd_mod_pubsub).
+-define(PROCNAME, ejabberd_mod_pubsub_odbc).
 -define(LOOPNAME, ejabberd_mod_pubsub_loop).
 -define(PLUGIN_PREFIX, "node_").
 -define(TREE_PREFIX, "nodetree_").
@@ -224,8 +224,6 @@
 	    ok
     end,
     ejabberd_router:register_route(Host),
-    update_node_database(Host, ServerHost),
-    update_state_database(Host, ServerHost),
     init_nodes(Host, ServerHost, NodeTree, Plugins),
     State = #state{host = Host,
 		server_host = ServerHost,
@@ -284,206 +282,15 @@
 
 init_nodes(Host, ServerHost, _NodeTree, Plugins) ->
     %% TODO, this call should be done plugin side
-    case lists:member("hometree", Plugins) of
+    case lists:member("hometree_odbc", Plugins) of
     true ->
-	create_node(Host, ServerHost, string_to_node("/home"), service_jid(Host), "hometree"),
-	create_node(Host, ServerHost, string_to_node("/home/" ++ ServerHost), service_jid(Host), "hometree");
+	create_node(Host, ServerHost, string_to_node("/home"), service_jid(Host), "hometree_odbc"),
+	create_node(Host, ServerHost, string_to_node("/home/" ++  ServerHost), service_jid(Host), "hometree_odbc");
     false ->
 	ok
     end.
 
-update_node_database(Host, ServerHost) ->
-    mnesia:del_table_index(pubsub_node, type),
-    mnesia:del_table_index(pubsub_node, parentid),
-    case catch mnesia:table_info(pubsub_node, attributes) of
-	[host_node, host_parent, info] ->
-	    ?INFO_MSG("upgrade node pubsub tables",[]),
-	    F = fun() ->
-			{Result, LastIdx} = lists:foldl(
-			  fun({pubsub_node, NodeId, ParentId, {nodeinfo, Items, Options, Entities}}, {RecList, NodeIdx}) ->
-				  ItemsList =
-				      lists:foldl(
-					fun({item, IID, Publisher, Payload}, Acc) ->
-						C = {unknown, Publisher},
-						M = {now(), Publisher},
-						mnesia:write(
-						  #pubsub_item{itemid = {IID, NodeIdx},
-							       creation = C,
-							       modification = M,
-							       payload = Payload}),
-						[{Publisher, IID} | Acc]
-					end, [], Items),
-				  Owners =
-				      dict:fold(
-					fun(JID, {entity, Aff, Sub}, Acc) ->
-						UsrItems =
-						    lists:foldl(
-						      fun({P, I}, IAcc) ->
-							      case P of
-								  JID -> [I | IAcc];
-								  _ -> IAcc
-							      end
-						      end, [], ItemsList),
-						mnesia:write({pubsub_state,
-							      {JID, NodeIdx},
-							      UsrItems,
-							      Aff,
-							      Sub}),
-						case Aff of
-						    owner -> [JID | Acc];
-						    _ -> Acc
-						end
-					end, [], Entities),
-				  mnesia:delete({pubsub_node, NodeId}),
-				  {[#pubsub_node{nodeid = NodeId,
-						id = NodeIdx,
-						parents = [element(2, ParentId)],
-						owners = Owners,
-						options = Options} |
-				   RecList], NodeIdx + 1}
-			  end, {[], 1},
-			  mnesia:match_object(
-			    {pubsub_node, {Host, '_'}, '_', '_'})),
-			mnesia:write(#pubsub_index{index = node, last = LastIdx, free = []}),
-			Result
-		end,
-	    {atomic, NewRecords} = mnesia:transaction(F),
-	    {atomic, ok} = mnesia:delete_table(pubsub_node),
-	    {atomic, ok} = mnesia:create_table(pubsub_node,
-					       [{disc_copies, [node()]},
-						{attributes, record_info(fields, pubsub_node)}]),
-	    FNew = fun() -> lists:foreach(fun(Record) ->
-						  mnesia:write(Record)
-					  end, NewRecords)
-		   end,
-	    case mnesia:transaction(FNew) of
-		{atomic, Result} ->
-		    ?INFO_MSG("Pubsub node tables updated correctly: ~p", [Result]);
-		{aborted, Reason} ->
-		    ?ERROR_MSG("Problem updating Pubsub node tables:~n~p", [Reason])
-	    end;
-	[nodeid, parentid, type, owners, options] ->
-	    F = fun({pubsub_node, NodeId, {_, Parent}, Type, Owners, Options}) ->
-		    #pubsub_node{
-			nodeid = NodeId,
-			id = 0,
-			parents = [Parent],
-			type = Type,
-			owners = Owners,
-			options = Options}
-		end,
-	    mnesia:transform_table(pubsub_node, F, [nodeid, id, parents, type, owners, options]),
-	    FNew = fun() ->
-		LastIdx = lists:foldl(fun(#pubsub_node{nodeid = NodeId} = PubsubNode, NodeIdx) ->
-		    mnesia:write(PubsubNode#pubsub_node{id = NodeIdx}),
-		    lists:foreach(fun(#pubsub_state{stateid = StateId} = State) ->
-			{JID, _} = StateId,
-			mnesia:delete({pubsub_state, StateId}),
-			mnesia:write(State#pubsub_state{stateid = {JID, NodeIdx}})
-		    end, mnesia:match_object(#pubsub_state{stateid = {'_', NodeId}, _ = '_'})),
-		    lists:foreach(fun(#pubsub_item{itemid = ItemId} = Item) ->
-			{IID, _} = ItemId,
-			{M1, M2} = Item#pubsub_item.modification,
-			{C1, C2} = Item#pubsub_item.creation,
-			mnesia:delete({pubsub_item, ItemId}),
-			mnesia:write(Item#pubsub_item{itemid = {IID, NodeIdx},
-						    modification = {M2, M1},
-						    creation = {C2, C1}})
-		    end, mnesia:match_object(#pubsub_item{itemid = {'_', NodeId}, _ = '_'})),
-		    NodeIdx + 1
-		end, 1, mnesia:match_object(
-			{pubsub_node, {Host, '_'}, '_', '_', '_', '_', '_'})
-		    ++  mnesia:match_object(
-			{pubsub_node, {{'_', ServerHost, '_'}, '_'}, '_', '_', '_', '_', '_'})),
-		mnesia:write(#pubsub_index{index = node, last = LastIdx, free = []})
-		end,
-	    case mnesia:transaction(FNew) of
-		{atomic, Result} ->
-		    rename_default_nodeplugin(),
-		    ?INFO_MSG("Pubsub node tables updated correctly: ~p", [Result]);
-		{aborted, Reason} ->
-		    ?ERROR_MSG("Problem updating Pubsub node tables:~n~p", [Reason])
-	    end;
-	[nodeid, id, parent, type, owners, options] ->
-	    F = fun({pubsub_node, NodeId, Id, Parent, Type, Owners, Options}) ->
-		    #pubsub_node{
-			nodeid = NodeId,
-			id = Id,
-			parents = [Parent],
-			type = Type,
-			owners = Owners,
-			options = Options}
-		end,
-	    mnesia:transform_table(pubsub_node, F, [nodeid, id, parents, type, owners, options]),
-	    rename_default_nodeplugin();
-	_ ->
-	    ok
-    end,
-    mnesia:transaction(fun() ->
-	    case catch mnesia:first(pubsub_node) of
-	    {_, L} when is_list(L) ->
-            lists:foreach(
-                    fun({H, N}) when is_list(N) ->
-                        [Node] = mnesia:read({pubsub_node, {H, N}}),
-                         Type = Node#pubsub_node.type,
-                         BN = element(2, node_call(Type, path_to_node, [N])),
-                         BP = case [element(2, node_call(Type, path_to_node, [P])) || P <- Node#pubsub_node.parents] of
-                            [<<>>] -> [];
-                            Parents -> Parents
-                         end,
-                         mnesia:write(Node#pubsub_node{nodeid={H, BN}, parents=BP}),
-                         mnesia:delete({pubsub_node, {H, N}});
-                      (_) ->
-                        ok
-	        end, mnesia:all_keys(pubsub_node));
-	    _ ->
-	        ok
-    	end
-    end). 
-
-rename_default_nodeplugin() ->
-    lists:foreach(fun(Node) ->
-	mnesia:dirty_write(Node#pubsub_node{type = "hometree"})
-    end, mnesia:dirty_match_object(#pubsub_node{type = "default", _ = '_'})).
-
-update_state_database(_Host, _ServerHost) ->
-    case catch mnesia:table_info(pubsub_state, attributes) of
-	[stateid, items, affiliation, subscription] ->
-	    ?INFO_MSG("upgrade state pubsub tables", []),
-	    F = fun ({pubsub_state, {JID, NodeID}, Items, Aff, Sub}, Acc) ->
-			Subs = case Sub of
-				   none ->
-				       [];
-				   _ ->
-				       {result, SubID} = pubsub_subscription:subscribe_node(JID, NodeID, []),
-				       [{Sub, SubID}]
-			       end,
-			NewState = #pubsub_state{stateid       = {JID, NodeID},
-						 items	 = Items,
-						 affiliation   = Aff,
-						 subscriptions = Subs},
-			[NewState | Acc]
-		end,
-	    {atomic, NewRecs} = mnesia:transaction(fun mnesia:foldl/3,
-						   [F, [], pubsub_state]),
-	    {atomic, ok} = mnesia:delete_table(pubsub_state),
-	    {atomic, ok} = mnesia:create_table(pubsub_state,
-					       [{disc_copies, [node()]},
-						{attributes, record_info(fields, pubsub_state)}]),
-	    FNew = fun () ->
-			   lists:foreach(fun mnesia:write/1, NewRecs)
-		   end,
-	    case mnesia:transaction(FNew) of
-		{atomic, Result} ->
-		    ?INFO_MSG("Pubsub state tables updated correctly: ~p",
-			      [Result]);
-		{aborted, Reason} ->
-		    ?ERROR_MSG("Problem updating Pubsub state tables:~n~p",
-			       [Reason])
-	    end;
-	_ ->
-	    ok
-    end.
+  
 
 send_loop(State) ->
     receive
@@ -495,7 +302,10 @@
 	%% for each node From is subscribed to
 	%% and if the node is so configured, send the last published item to From
 	lists:foreach(fun(PType) ->
-	    {result, Subscriptions} = node_action(Host, PType, get_entity_subscriptions, [Host, JID]),
+	    Subscriptions = case catch node_action(Host, PType, get_entity_subscriptions_for_send_last, [Host, JID]) of
+		{result, S} -> S;
+		_ -> []
+	    end,
 	    lists:foreach(
 		fun({Node, subscribed, _, SubJID}) -> 
 		    if (SubJID == LJID) or (SubJID == BJID) ->
@@ -758,10 +568,10 @@
 	lists:foreach(fun(PType) ->
 	    {result, Subscriptions} = node_action(Host, PType, get_entity_subscriptions, [Host, Entity]),
 	    lists:foreach(fun
-		({#pubsub_node{options = Options, owners = Owners, id = NodeId}, subscribed, _, JID}) ->
+		({#pubsub_node{options = Options, id = NodeId}, subscribed, _, JID}) ->
 		    case get_option(Options, access_model) of
 			presence ->
-			    case lists:member(BJID, Owners) of
+			    case lists:member(BJID, node_owners(Host, PType, NodeId)) of
 				true ->
 				    node_action(Host, PType, unsubscribe_node, [NodeId, Entity, JID, all]);
 				false ->
@@ -939,10 +749,11 @@
 				  end,
 			    ejabberd_router:route(To, From, Res);
 			#iq{type = get, ns = ?NS_DISCO_ITEMS,
-			    payload = SubEl} ->
+			    payload = SubEl} = IQ ->
 			    QAttrs = SubEl#xmlel.attrs,
 			    Node = exmpp_xml:get_attribute_from_list_as_list(QAttrs, 'node', ""),
-			    Res = case iq_disco_items(Host, Node, From) of
+			    Rsm = jlib:rsm_decode(IQ),
+			    Res = case iq_disco_items(Host, Node, From, Rsm) of
 				      {result, IQRes} ->
 					  Result = #xmlel{ns = ?NS_DISCO_ITEMS,
 					    name = 'query', attrs = QAttrs,
@@ -1085,7 +896,7 @@
 				    [] ->
 					["leaf"]; %% No sub-nodes: it's a leaf node
 				    _ ->
-					case node_call(Type, get_items, [NodeId, From]) of
+					case node_call(Type, get_items, [NodeId, From, none]) of
 					    {result, []} -> ["collection"];
 					    {result, _} -> ["leaf", "collection"];
 					    _ -> []
@@ -1101,8 +912,9 @@
 			    [];
 			true ->
 			    [#xmlel{ns = ?NS_DISCO_INFO, name = 'feature', attrs = [?XMLATTR('var', ?NS_PUBSUB_s)]} |
-			     lists:map(fun(T) ->
-					       #xmlel{ns = ?NS_DISCO_INFO, name = 'feature', attrs = [?XMLATTR('var', ?NS_PUBSUB_s++"#"++T)]}
+			     lists:map(fun
+                            ("rsm") -> #xmlel{ns = ?NS_DISCO_INFO, name = 'feature', attrs = [?XMLATTR('var', ?NS_RSM_s)]};
+                            (T) ->  #xmlel{ns = ?NS_DISCO_INFO, name = 'feature', attrs = [?XMLATTR('var', ?NS_PUBSUB_s++"#"++T)]}
 				       end, features(Type))]
 		    end,
 		%% TODO: add meta-data info (spec section 5.4)
@@ -1131,8 +943,9 @@
 		#xmlel{ns = ?NS_DISCO_INFO, name = 'feature', attrs = [?XMLATTR('var', ?NS_PUBSUB_s)]},
 		#xmlel{ns = ?NS_DISCO_INFO, name = 'feature', attrs = [?XMLATTR('var', ?NS_ADHOC_s)]},
 		#xmlel{ns = ?NS_DISCO_INFO, name = 'feature', attrs = [?XMLATTR('var', ?NS_VCARD_s)]}] ++
-	     lists:map(fun(Feature) ->
-		 #xmlel{ns = ?NS_DISCO_INFO, name = 'feature', attrs = [?XMLATTR('var', ?NS_PUBSUB_s++"#"++Feature)]}
+	     lists:map(fun
+		("rsm") -> #xmlel{ns = ?NS_DISCO_INFO, name = 'feature', attrs = [?XMLATTR('var', ?NS_RSM_s)]};
+		(Feature) -> #xmlel{ns = ?NS_DISCO_INFO, name = 'feature', attrs = [?XMLATTR('var', ?NS_PUBSUB_s++"#"++Feature)]}
 	     end, features(Host, Node))};
         ?NS_ADHOC_b ->
             command_disco_info(Host, Node, From);
@@ -1142,7 +955,7 @@
 	    node_disco_info(Host, Node, From)
     end.
 
-iq_disco_items(Host, [], From) ->
+iq_disco_items(Host, [], From, _RSM) ->
     case tree_action(Host, get_subnodes, [Host, <<>>, From]) of
 	Nodes when is_list(Nodes) ->
 	    {result, lists:map(
@@ -1159,7 +972,7 @@
 	Other ->
 	    Other
     end;
-iq_disco_items(Host, ?NS_ADHOC_s, _From) ->
+iq_disco_items(Host, ?NS_ADHOC_s, _From, _RSM) ->
     %% TODO: support localization of this string
     CommandItems = [
 	  #xmlel{ns = ?NS_DISCO_ITEMS, name = 'item', 
@@ -1168,10 +981,10 @@
 		  ?XMLATTR('name', "Get Pending") 
 	      ]}],
     {result, CommandItems};
-iq_disco_items(_Host, ?NS_PUBSUB_GET_PENDING, _From) ->
+iq_disco_items(_Host, ?NS_PUBSUB_GET_PENDING, _From, _RSM) ->
     CommandItems = [],
     {result, CommandItems};
-iq_disco_items(Host, Item, From) ->
+iq_disco_items(Host, Item, From, RSM) ->
     case string:tokens(Item, "!") of
 	[_SNode, _ItemID] ->
 	    {result, []};
@@ -1180,9 +993,9 @@
 	    Action =
 		fun(#pubsub_node{type = Type, id = NodeId}) ->
 			% TODO call get_items/6 instead for access control (EJAB-1033)
-			NodeItems = case node_call(Type, get_items, [NodeId, From]) of
+			{NodeItems, RsmOut} = case node_call(Type, get_items, [NodeId, From, RSM]) of
 					{result, I} -> I;
-					_ -> []
+					_ -> {[], none}
 				    end,
 			Nodes = lists:map(
 				  fun(#pubsub_node{nodeid = {_, SubNode}, options = Options}) ->
@@ -1200,7 +1013,7 @@
 					  {result, Name} = node_call(Type, get_item_name, [Host, Node, RN]),
 					  #xmlel{ns = ?NS_DISCO_ITEMS, name = 'item', attrs = [?XMLATTR('jid', Host), ?XMLATTR('name', Name)]}
 				  end, NodeItems),
-			{result, Nodes ++ Items}
+			{result, Nodes ++ Items ++ jlib:rsm_encode(RsmOut)}
 		end,
 	    case transaction(Host, Node, Action, sync_dirty) of
 		{result, {_, Result}} -> {result, Result};
@@ -1332,7 +1145,8 @@
 			(_, Acc) ->
 			    Acc
 			end, [], exmpp_xml:remove_cdata_from_list(Els)),
-		    get_items(Host, Node, From, SubId, MaxItems, ItemIDs);
+		    RSM = jlib:rsm_decode(SubEl),
+            get_items(Host, Node, From, SubId, MaxItems, ItemIDs, RSM);
 		{get, 'subscriptions'} ->
 		    get_subscriptions(Host, Node, From, Plugins);
 		{get, 'affiliations'} ->
@@ -1489,7 +1303,8 @@
 		    _	       -> []
 		end
 	end,
-    case transaction(fun () -> {result, lists:flatmap(Tr, Plugins)} end,
+    case transaction(Host, 
+                     fun () -> {result, lists:flatmap(Tr, Plugins)} end,
 		     sync_dirty) of
 	{result, Res} -> Res;
 	Err	   -> Err
@@ -1533,7 +1348,7 @@
 
 %%% authorization handling
 
-send_authorization_request(#pubsub_node{owners = Owners, nodeid = {Host, Node}}, Subscriber) ->
+send_authorization_request(#pubsub_node{nodeid = {Host, Node}, type = Type, id = NodeId}, Subscriber) ->
     Lang = "en", %% TODO fix
     {U, S, R} = Subscriber,
     Stanza = #xmlel{ns = ?NS_JABBER_CLIENT, name = 'message', children =
@@ -1563,7 +1378,7 @@
     lists:foreach(fun(Owner) ->
     	{U, S, R} = Owner,
 	ejabberd_router:route(service_jid(Host), exmpp_jid:make(U, S, R), Stanza)
-    end, Owners).
+    end, node_owners(Host, Type, NodeId)).
 
 find_authorization_response(Packet) ->
     Els = Packet#xmlel.children,
@@ -1622,8 +1437,8 @@
 			"true" -> true;
 			_ -> false
 		    end,
-	    Action = fun(#pubsub_node{type = Type, owners = Owners, id = NodeId}) ->
-			     IsApprover = lists:member(jlib:short_prepd_bare_jid(From), Owners),
+	    Action = fun(#pubsub_node{type = Type, id = NodeId}) ->
+			     IsApprover = lists:member(jlib:short_prepd_bare_jid(From), node_owners_call(Type, NodeId)),
 			     {result, Subscriptions} = node_call(Type, get_subscriptions, [NodeId, Subscriber]),
 			     if
 				 not IsApprover ->
@@ -1818,7 +1633,7 @@
 		end,
 	    Reply = #xmlel{ns = ?NS_PUBSUB, name = 'pubsub', children =
 		      [#xmlel{ns = ?NS_PUBSUB, name = 'create', attrs = nodeAttr(Node)}]},
-	    case transaction(CreateNode, transaction) of
+	    case transaction(Host, CreateNode, transaction) of
 		{result, {Result, broadcast}} ->
 		    %%Lang = "en", %% TODO: fix
 		    %%OwnerKey = jlib:jid_tolower(jlib:jid_remove_resource(Owner)),
@@ -1927,7 +1742,7 @@
 %%<li>The node does not exist.</li>
 %%</ul>
 subscribe_node(Host, Node, From, JID, Configuration) ->
-    SubOpts = case pubsub_subscription:parse_options_xform(Configuration) of
+    SubOpts = case pubsub_subscription_odbc:parse_options_xform(Configuration) of
 	{result, GoodSubOpts} -> GoodSubOpts;
 	_ -> invalid
     end,
@@ -1937,7 +1752,7 @@
 	_:_ ->
 	    {undefined, undefined, undefined}
     end,
-    Action = fun(#pubsub_node{options = Options, owners = [Owner|_], type = Type, id = NodeId}) ->
+    Action = fun(#pubsub_node{options = Options, type = Type, id = NodeId}) ->
 		    Features = features(Type),
 		    SubscribeFeature = lists:member("subscribe", Features),
 		    OptionsFeature = lists:member("subscription-options", Features),
@@ -1956,9 +1771,12 @@
 				    {"", "", ""} ->
 					{false, false};
 				    _ ->
-					{OU, OS, _} = Owner,
-					get_roster_info(OU, OS,
-							Subscriber, AllowedGroups)
+					case node_owners_call(Type, NodeId) of
+					    [{OU, OS, _} | _] ->
+						get_roster_info(OU, OS, Subscriber, AllowedGroups);
+					    _ ->
+						{false, false}
+					end
 				end
 			end,
 		    if
@@ -2304,7 +2122,7 @@
 %% <p>The permission are not checked in this function.</p>
 %% @todo We probably need to check that the user doing the query has the right
 %% to read the items.
-get_items(Host, Node, From, SubId, SMaxItems, ItemIDs) ->
+get_items(Host, Node, From, SubId, SMaxItems, ItemIDs, RSM) ->
     MaxItems =
 	if
 	    SMaxItems == "" -> get_max_items_node(Host);
@@ -2343,11 +2161,11 @@
 			     node_call(Type, get_items,
 				       [NodeId, From,
 					AccessModel, PresenceSubscription, RosterGroup,
-					SubId])
+					SubId, RSM])
 		     end
 	     end,
 	     case transaction(Host, Node, Action, sync_dirty) of
-		{result, {_, Items}} ->
+		{result, {_, Items, RSMOut}} ->
 		    SendItems = case ItemIDs of
 			[] -> 
 			    Items;
@@ -2360,7 +2178,7 @@
 		    %% number of items sent to MaxItems:
 		    {result, #xmlel{ns = ?NS_PUBSUB, name = 'pubsub', children =
 				[#xmlel{ns = ?NS_PUBSUB, name = 'items', attrs = nodeAttr(Node), children =
-				    itemsEls(lists:sublist(SendItems, MaxItems))}]}};
+				    itemsEls(lists:sublist(SendItems, MaxItems))} | jlib:rsm_encode(RSMOut)]}};
 		Error ->
 		    Error
 	    end
@@ -2392,16 +2210,29 @@
 %% @doc <p>Resend the items of a node to the user.</p>
 %% @todo use cache-last-item feature
 send_items(Host, Node, NodeId, Type, LJID, last) ->
-    case get_cached_item(Host, NodeId) of
+    Stanza = case get_cached_item(Host, NodeId) of
 	undefined ->
-	    send_items(Host, Node, NodeId, Type, LJID, 1);
+	    % special ODBC optimization, works only with node_hometree_odbc, node_flat_odbc and node_pep_odbc
+	    case node_action(Host, Type, get_last_items, [NodeId, LJID, 1]) of
+		{result, [LastItem]} ->
+		    {ModifNow, ModifUSR} = LastItem#pubsub_item.modification,
+		    event_stanza_with_delay([#xmlel{ns = ?NS_PUBSUB_EVENT, name = 'items',
+					attrs = nodeAttr(Node),
+					children = itemsEls([LastItem])}],
+				ModifNow, ModifUSR);
+		_ ->
+		    event_stanza([#xmlel{ns = ?NS_PUBSUB_EVENT, name = 'items',
+					attrs = nodeAttr(Node),
+					children = itemsEls([])}])
+	    end;
 	LastItem ->
 	    {ModifNow, ModifUSR} = LastItem#pubsub_item.modification,
-	    Stanza = event_stanza_with_delay(
+	    event_stanza_with_delay(
 	    	[#xmlel{ns = ?NS_PUBSUB_EVENT, name = 'items', attrs = nodeAttr(Node),
-			children = itemsEls([LastItem])}], ModifNow, ModifUSR),
-	    ejabberd_router:route(service_jid(Host), exmpp_jid:make(LJID), Stanza)
-    end;
+			children = itemsEls([LastItem])}], ModifNow, ModifUSR)
+    end,
+    {U, S, R} = LJID,
+    ejabberd_router:route(service_jid(Host), exmpp_jid:make(U, S, R), Stanza);
 send_items(Host, Node, NodeId, Type, {LU, LS, LR} = LJID, Number) ->
     ToSend = case node_action(Host, Type, get_items, [NodeId, LJID]) of
 	{result, []} -> 
@@ -2528,7 +2359,8 @@
 	error ->
 	    {error, 'bad-request'};
 	_ ->
-	    Action = fun(#pubsub_node{owners = Owners, type = Type, id = NodeId}=N) ->
+	    Action = fun(#pubsub_node{type = Type, id = NodeId}) ->
+			Owners = node_owners_call(Type, NodeId),
 			case lists:member(Owner, Owners) of
 			    true ->
 				OwnerJID = exmpp_jid:make(Owner),
@@ -2538,24 +2370,8 @@
 				    end,
 				lists:foreach(
 				    fun({JID, Affiliation}) ->
-					{result, _} = node_call(Type, set_affiliation, [NodeId, JID, Affiliation]),
-					case Affiliation of
-					    owner ->
-						NewOwner = jlib:short_prepd_bare_jid(JID),
-						NewOwners = [NewOwner|Owners],
-						tree_call(Host, set_node, [N#pubsub_node{owners = NewOwners}]);
-					    none ->
-						OldOwner = jlib:short_prepd_bare_jid(JID),
-						case lists:member(OldOwner, Owners) of
-						    true ->
-							NewOwners = Owners--[OldOwner],
-							tree_call(Host, set_node, [N#pubsub_node{owners = NewOwners}]);
-						    _ ->
-							ok
-						end;
-					    _ ->
-						ok
-					end
+					    % TODO, check if nothing missing here about new owners
+					    node_call(Type, set_affiliation, [NodeId, JID, Affiliation])
 				    end, FilteredEntities),
 				{result, []};
 			    _ ->
@@ -2610,11 +2426,11 @@
     end.
 
 read_sub(Subscriber, Node, NodeID, SubID, Lang) ->
-    case pubsub_subscription:get_subscription(Subscriber, NodeID, SubID) of
+    case pubsub_subscription_odbc:get_subscription(Subscriber, NodeID, SubID) of
 	{error, notfound} ->
 	    {error, extended_error('not-acceptable', "invalid-subid")};
 	{result, #pubsub_subscription{options = Options}} ->
-            {result, XdataEl} = pubsub_subscription:get_options_xform(Lang, Options),
+            {result, XdataEl} = pubsub_subscription_odbc:get_options_xform(Lang, Options),
             OptionsEl = #xmlel{ns = ?NS_PUBSUB, name = 'options',
 			       attrs = [ ?XMLATTR('jid', exmpp_jid:to_binary(Subscriber)),
 					?XMLATTR('subid', SubID) | nodeAttr(Node)],
@@ -2641,7 +2457,7 @@
     end.
 
 set_options_helper(Configuration, JID, NodeID, SubID, Type) ->
-    SubOpts = case pubsub_subscription:parse_options_xform(Configuration) of
+    SubOpts = case pubsub_subscription_odbc:parse_options_xform(Configuration) of
 	{result, GoodSubOpts} -> GoodSubOpts;
 	_ -> invalid
     end,
@@ -2671,7 +2487,7 @@
 write_sub(_Subscriber, _NodeID, _SubID, invalid) ->
     {error, extended_error('bad-request', "invalid-options")};
 write_sub(Subscriber, NodeID, SubID, Options) ->
-    case pubsub_subscription:set_subscription(Subscriber, NodeID, SubID, Options) of
+    case pubsub_subscription_odbc:set_subscription(Subscriber, NodeID, SubID, Options) of
 	{error, notfound} ->
 	    {error, extended_error('not-acceptable', "invalid-subid")};
 	{result, _} ->
@@ -2844,8 +2660,8 @@
 						 ?XMLATTR('subsription', subscription_to_string(Sub)) | nodeAttr(Node)]}]}]},
 		ejabberd_router:route(service_jid(Host), JID, Stanza)
 	    end,
-	    Action = fun(#pubsub_node{owners = Owners, type = Type, id = NodeId}) ->
-			    case lists:member(Owner, Owners) of
+	    Action = fun(#pubsub_node{type = Type, id = NodeId}) ->
+			    case lists:member(Owner, node_owners_call(Type, NodeId)) of
 				true ->
 				    Result = lists:foldl(fun({JID, Subscription, SubId}, Acc) ->
 
@@ -3186,7 +3002,7 @@
 			{Depth, [{N, get_node_subs(N)} || N <- Nodes]}
 	    end, tree_call(Host, get_parentnodes_tree, [Host, Node, service_jid(Host)]))}
 	end,
-    case transaction(Action, sync_dirty) of
+    case transaction(Host, Action, sync_dirty) of
 	{result, CollSubs} -> CollSubs;
 	_ -> []
     end.
@@ -3200,9 +3016,9 @@
 
 get_options_for_subs(NodeID, Subs) ->
     lists:foldl(fun({JID, subscribed, SubID}, Acc) ->
-			case pubsub_subscription:read_subscription(JID, NodeID, SubID) of
+			case pubsub_subscription_odbc:get_subscription(JID, NodeID, SubID) of
 			    {error, notfound} -> [{JID, SubID, []} | Acc];
-			    #pubsub_subscription{options = Options} -> [{JID, SubID, Options} | Acc];
+			    {result, #pubsub_subscription{options = Options}} -> [{JID, SubID, Options} | Acc];
 			    _ -> Acc
 			end;
 		    (_, Acc) ->
@@ -3424,6 +3240,30 @@
 	    Result
     end.
 
+%% @spec (Host, Type, NodeId) -> [ljid()]
+%%    NodeId = pubsubNodeId()
+%% @doc <p>Return list of node owners.</p>
+node_owners(Host, Type, NodeId) ->
+    case node_action(Host, Type, get_node_affiliations, [NodeId]) of
+	{result, Affiliations} ->
+	    lists:foldl(
+		fun({LJID, owner}, Acc) -> [LJID|Acc];
+		   (_, Acc) -> Acc
+	    end, [], Affiliations);
+	_ ->
+	    []
+    end.
+node_owners_call(Type, NodeId) ->
+    case node_call(Type, get_node_affiliations, [NodeId]) of
+	{result, Affiliations} ->
+	    lists:foldl(
+		fun({LJID, owner}, Acc) -> [LJID|Acc];
+		   (_, Acc) -> Acc
+	    end, [], Affiliations);
+	_ ->
+	    []
+    end.
+
 %% @spec (Host, Options) -> MaxItems
 %%	 Host = host()
 %%	 Options = [Option]
@@ -3827,7 +3667,13 @@
 tree_action(Host, Function, Args) ->
     ?DEBUG("tree_action ~p ~p ~p",[Host,Function,Args]),
     Fun = fun() -> tree_call(Host, Function, Args) end,
-    catch mnesia:sync_dirty(Fun).
+    case catch ejabberd_odbc:sql_bloc(odbc_conn(Host), Fun) of
+    {atomic, Result} -> 
+    	Result;
+    {aborted, Reason} -> 
+	    ?ERROR_MSG("transaction return internal error: ~p~n",[{aborted, Reason}]),
+    	{error, 'internal-server-error'}
+    end.
 
 %% @doc <p>node plugin call.</p>
 node_call(Type, Function, Args) ->
@@ -3847,13 +3693,13 @@
 
 node_action(Host, Type, Function, Args) ->
     ?DEBUG("node_action ~p ~p ~p ~p",[Host,Type,Function,Args]),
-    transaction(fun() ->
+    transaction(Host, fun() ->
 			node_call(Type, Function, Args)
 		end, sync_dirty).
 
 %% @doc <p>plugin transaction handling.</p>
 transaction(Host, Node, Action, Trans) ->
-    transaction(fun() ->
+    transaction(Host, fun() ->
 			case tree_call(Host, get_node, [Host, Node]) of
 			    N when is_record(N, pubsub_node) ->
 				case Action(N) of
@@ -3866,8 +3712,15 @@
 			end
 		end, Trans).
 
-transaction(Fun, Trans) ->
-    case catch mnesia:Trans(Fun) of
+transaction(Host, Fun, Trans) ->
+    transaction_retry(Host, Fun, Trans, 2).
+
+transaction_retry(Host, Fun, Trans, Count) ->
+    SqlFun = case Trans of
+                transaction -> sql_transaction;
+                _ -> sql_bloc
+             end,
+    case catch ejabberd_odbc:SqlFun(odbc_conn(Host), Fun) of
 	{result, Result} -> {result, Result};
 	{error, Error} -> {error, Error};
 	{atomic, {result, Result}} -> {result, Result};
@@ -3875,6 +3728,15 @@
 	{aborted, Reason} ->
 	    ?ERROR_MSG("transaction return internal error: ~p~n", [{aborted, Reason}]),
 	    {error, 'internal-server-error'};
+	{'EXIT', {timeout, _} = Reason} ->
+	    case Count of
+		0 ->
+		    ?ERROR_MSG("transaction return internal error: ~p~n", [{'EXIT', Reason}]),
+		    {error, 'internal-server-error'};
+		N ->
+		    erlang:yield(),
+		    transaction_retry(Host, Fun, Trans, N-1)
+	    end;
 	{'EXIT', Reason} ->
 	    ?ERROR_MSG("transaction return internal error: ~p~n", [{'EXIT', Reason}]),
 	    {error, 'internal-server-error'};
@@ -3883,6 +3745,16 @@
 	    {error, 'internal-server-error'}
     end.
 
+odbc_conn({_U, Host, _R})->
+    Host;
+odbc_conn(Host) ->
+    lists:dropwhile(fun(A) -> A/=$. end, Host) -- ".".
+
+%% escape value for database storage
+escape({_U, _H, _R}=JID)->
+    ejabberd_odbc:escape(exmpp_jid:to_list(JID));
+escape(Value)->
+    ejabberd_odbc:escape(Value).
 %%%% helpers
 
 %% Add pubsub-specific error element