Changeset

--- a/mod_pep_plus/mod_pep_plus.lua	Tue Oct 17 22:39:37 2017 +0200
+++ b/mod_pep_plus/mod_pep_plus.lua	Wed Oct 18 09:56:29 2017 +0200
@@ -1,18 +1,19 @@
-local pubsub = require "util.pubsub";
+local pubsub = module:require "util_pubsub";
 local jid_bare = require "util.jid".bare;
 local jid_split = require "util.jid".split;
+local jid_join = require "util.jid".join;
 local set_new = require "util.set".new;
 local st = require "util.stanza";
 local calculate_hash = require "util.caps".calculate_hash;
 local is_contact_subscribed = require "core.rostermanager".is_contact_subscribed;
+local cache = require "util.cache";
+local set = require "util.set";
 
 local xmlns_pubsub = "http://jabber.org/protocol/pubsub";
 local xmlns_pubsub_event = "http://jabber.org/protocol/pubsub#event";
 local xmlns_pubsub_owner = "http://jabber.org/protocol/pubsub#owner";
 
 local lib_pubsub = module:require "pubsub";
-local handlers = lib_pubsub.handlers;
-local pubsub_error_reply = lib_pubsub.pubsub_error_reply;
 
 local empty_set = set_new();
 
@@ -20,6 +21,11 @@
 local recipients = {};
 local hash_map = {};
 
+local host = module.host;
+
+local known_nodes_map = module:open_store("pep", "map");
+local known_nodes = module:open_store("pep");
+
 function module.save()
 	return { services = services };
 end
@@ -28,25 +34,41 @@
 	services = data.services;
 end
 
-local function subscription_presence(user_bare, recipient)
+local function subscription_presence(username, recipient)
+	local user_bare = jid_join(username, host);
 	local recipient_bare = jid_bare(recipient);
 	if (recipient_bare == user_bare) then return true; end
-	local username, host = jid_split(user_bare);
 	return is_contact_subscribed(username, host, recipient_bare);
 end
 
-local function get_broadcaster(name)
+local function simple_itemstore(username)
+	return function (config, node)
+		if config["persist_items"] then
+			module:log("debug", "Creating new persistent item store for user %s, node %q", username, node);
+			known_nodes_map:set(username, node, true);
+			local archive = module:open_store("pep_"..node, "archive");
+			return lib_pubsub.archive_itemstore(archive, config, username, node, false);
+		else
+			module:log("debug", "Creating new ephemeral item store for user %s, node %q", username, node);
+			known_nodes_map:set(username, node, nil);
+			return cache.new(tonumber(config["max_items"]));
+		end
+	end
+end
+
+local function get_broadcaster(username)
+	local user_bare = jid_join(username, host);
 	local function simple_broadcast(kind, node, jids, item)
 		if item then
 			item = st.clone(item);
 			item.attr.xmlns = nil; -- Clear the pubsub namespace
 		end
-		local message = st.message({ from = name, type = "headline" })
+		local message = st.message({ from = user_bare, type = "headline" })
 			:tag("event", { xmlns = xmlns_pubsub_event })
 				:tag(kind, { node = node })
 					:add_child(item);
 		for jid in pairs(jids) do
-			module:log("debug", "Sending notification to %s from %s: %s", jid, name, tostring(item));
+			module:log("debug", "Sending notification to %s from %s: %s", jid, user_bare, tostring(item));
 			message.attr.to = jid;
 			module:send(message);
 		end
@@ -54,8 +76,10 @@
 	return simple_broadcast;
 end
 
-function get_pep_service(name)
-	local service = services[name];
+function get_pep_service(username)
+	module:log("debug", "get_pep_service(%q)", username);
+	local user_bare = jid_join(username, host);
+	local service = services[username];
 	if service then
 		return service;
 	end
@@ -155,42 +179,51 @@
 		};
 
 		node_defaults = {
-			["pubsub#max_items"] = "1";
+			["max_items"] = 1;
+			["persist_items"] = true;
 		};
 
 		autocreate_on_publish = true;
 		autocreate_on_subscribe = true;
 
-		broadcaster = get_broadcaster(name);
+		itemstore = simple_itemstore(username);
+		broadcaster = get_broadcaster(username);
 		get_affiliation = function (jid)
-			if jid_bare(jid) == name then
+			if jid_bare(jid) == user_bare then
 				return "owner";
-			elseif subscription_presence(name, jid) then
+			elseif subscription_presence(username, jid) then
 				return "subscriber";
 			end
 		end;
 
 		normalize_jid = jid_bare;
 	});
-	services[name] = service;
-	module:add_item("pep-service", { service = service, jid = name });
+	local nodes, err = known_nodes:get(username);
+	if nodes then
+		module:log("debug", "Restoring nodes for user %s", username);
+		for node in pairs(nodes) do
+			module:log("debug", "Restoring node %q", node);
+			service:create(node, true);
+		end
+	elseif err then
+		module:log("error", "Could not restore nodes for %s: %s", username, err);
+	else
+		module:log("debug", "No known nodes");
+	end
+	services[username] = service;
+	module:add_item("pep-service", { service = service, jid = user_bare });
 	return service;
 end
 
 function handle_pubsub_iq(event)
 	local origin, stanza = event.origin, event.stanza;
-	local pubsub = stanza.tags[1];
-	local action = pubsub.tags[1];
-	if not action then
-		return origin.send(st.error_reply(stanza, "cancel", "bad-request"));
+	local service_name = origin.username;
+	if stanza.attr.to ~= nil then
+		service_name = jid_split(stanza.attr.to);
 	end
-	local service_name = stanza.attr.to or origin.username.."@"..origin.host
 	local service = get_pep_service(service_name);
-	local handler = handlers[stanza.attr.type.."_"..action.name];
-	if handler then
-		handler(origin, stanza, action, service);
-		return true;
-	end
+
+	return lib_pubsub.handle_pubsub_iq(event, service)
 end
 
 module:hook("iq/bare/"..xmlns_pubsub..":pubsub", handle_pubsub_iq);
@@ -225,7 +258,7 @@
 local function resend_last_item(jid, node, service)
 	local ok, items = service:get_items(node, jid);
 	if not ok then return; end
-	for i, id in ipairs(items) do
+	for _, id in ipairs(items) do
 		service.config.broadcaster("items", node, { [jid] = true }, items[id]);
 	end
 end
@@ -268,30 +301,34 @@
 module:hook("presence/bare", function(event)
 	-- inbound presence to bare JID recieved
 	local origin, stanza = event.origin, event.stanza;
-	local user = stanza.attr.to or (origin.username..'@'..origin.host);
 	local t = stanza.attr.type;
-	local self = not stanza.attr.to;
-	local service = get_pep_service(user);
+	local is_self = not stanza.attr.to;
+	local username = jid_split(stanza.attr.to);
+	local user_bare = jid_bare(stanza.attr.to);
+	if is_self then
+		username = origin.username;
+		user_bare = jid_join(username, host);
+	end
 
 	if not t then -- available presence
-		if self or subscription_presence(user, stanza.attr.from) then
+		if is_self or subscription_presence(username, stanza.attr.from) then
 			local recipient = stanza.attr.from;
-			local current = recipients[user] and recipients[user][recipient];
+			local current = recipients[username] and recipients[username][recipient];
 			local hash, query_node = get_caps_hash_from_presence(stanza, current);
 			if current == hash or (current and current == hash_map[hash]) then return; end
 			if not hash then
-				update_subscriptions(recipient, user);
+				update_subscriptions(recipient, username);
 			else
-				recipients[user] = recipients[user] or {};
+				recipients[username] = recipients[username] or {};
 				if hash_map[hash] then
-					update_subscriptions(recipient, user, hash_map[hash]);
+					update_subscriptions(recipient, username, hash_map[hash]);
 				else
-					recipients[user][recipient] = hash;
+					recipients[username][recipient] = hash;
 					local from_bare = origin.type == "c2s" and origin.username.."@"..origin.host;
-					if self or origin.type ~= "c2s" or (recipients[from_bare] and recipients[from_bare][origin.full_jid]) ~= hash then
+					if is_self or origin.type ~= "c2s" or (recipients[from_bare] and recipients[from_bare][origin.full_jid]) ~= hash then
 						-- COMPAT from ~= stanza.attr.to because OneTeam can't deal with missing from attribute
 						origin.send(
-							st.stanza("iq", {from=user, to=stanza.attr.from, id="disco", type="get"})
+							st.stanza("iq", {from=user_bare, to=stanza.attr.from, id="disco", type="get"})
 								:tag("query", {xmlns = "http://jabber.org/protocol/disco#info", node = query_node})
 						);
 					end
@@ -299,14 +336,14 @@
 			end
 		end
 	elseif t == "unavailable" then
-		update_subscriptions(stanza.attr.from, user);
-	elseif not self and t == "unsubscribe" then
+		update_subscriptions(stanza.attr.from, username);
+	elseif not is_self and t == "unsubscribe" then
 		local from = jid_bare(stanza.attr.from);
-		local subscriptions = recipients[user];
+		local subscriptions = recipients[username];
 		if subscriptions then
 			for subscriber in pairs(subscriptions) do
 				if jid_bare(subscriber) == from then
-					update_subscriptions(subscriber, user);
+					update_subscriptions(subscriber, username);
 				end
 			end
 		end
@@ -321,10 +358,15 @@
 	end
 
 	-- Process disco response
-	local self = not stanza.attr.to;
-	local user = stanza.attr.to or (origin.username..'@'..origin.host);
+	local is_self = stanza.attr.to == nil;
+	local user_bare = jid_bare(stanza.attr.to);
+	local username = jid_split(stanza.attr.to);
+	if is_self then
+		username = origin.username;
+		user_bare = jid_join(username, host);
+	end
 	local contact = stanza.attr.from;
-	local current = recipients[user] and recipients[user][contact];
+	local current = recipients[username] and recipients[username][contact];
 	if type(current) ~= "string" then return; end -- check if waiting for recipient's response
 	local ver = current;
 	if not string.find(current, "#") then
@@ -338,20 +380,26 @@
 		end
 	end
 	hash_map[ver] = notify; -- update hash map
-	if self then
+	if is_self then
+		-- Optimization: Fiddle with other local users
 		for jid, item in pairs(origin.roster) do -- for all interested contacts
-			if item.subscription == "both" or item.subscription == "from" then
-				if not recipients[jid] then recipients[jid] = {}; end
-				update_subscriptions(contact, jid, notify);
+			if jid then
+				local contact_node, contact_host = jid_split(jid);
+				if contact_host == host and item.subscription == "both" or item.subscription == "from" then
+					update_subscriptions(user_bare, contact_node, notify);
+				end
 			end
 		end
 	end
-	update_subscriptions(contact, user, notify);
+	update_subscriptions(contact, username, notify);
 end);
 
 module:hook("account-disco-info-node", function(event)
 	local reply, stanza, origin = event.reply, event.stanza, event.origin;
-	local service_name = stanza.attr.to or origin.username.."@"..origin.host
+	local service_name = origin.username;
+	if stanza.attr.to ~= nil then
+		service_name = jid_split(stanza.attr.to);
+	end
 	local service = get_pep_service(service_name);
 	local node = event.node;
 	local ok = service:get_items(node, jid_bare(stanza.attr.from) or true);
@@ -361,33 +409,64 @@
 end);
 
 module:hook("account-disco-info", function(event)
-	local reply = event.reply;
+	local origin, reply = event.origin, event.reply;
+
 	reply:tag('identity', {category='pubsub', type='pep'}):up();
-	reply:tag('feature', {var='http://jabber.org/protocol/pubsub#publish'}):up();
+
+	local username = jid_split(reply.attr.from) or origin.username;
+	local service = get_pep_service(username);
+
+	local suppored_features = lib_pubsub.get_feature_set(service) + set.new{
+		-- Features not covered by the above
+		"access-presence",
+		"auto-subscribe",
+		"filtered-notifications",
+		"last-published",
+		"persistent-items",
+		"presence-notifications",
+		"presence-subscribe",
+	};
+
+	for feature in suppored_features do
+		reply:tag('feature', {var=xmlns_pubsub.."#"..feature}):up();
+	end
 end);
 
 module:hook("account-disco-items-node", function(event)
 	local reply, stanza, origin = event.reply, event.stanza, event.origin;
 	local node = event.node;
-	local service_name = stanza.attr.to or origin.username.."@"..origin.host
-	local service = get_pep_service(service_name);
+	local is_self = stanza.attr.to == nil;
+	local user_bare = jid_bare(stanza.attr.to);
+	local username = jid_split(stanza.attr.to);
+	if is_self then
+		username = origin.username;
+		user_bare = jid_join(username, host);
+	end
+	local service = get_pep_service(username);
 	local ok, ret = service:get_items(node, jid_bare(stanza.attr.from) or true);
 	if not ok then return; end
 	event.exists = true;
 	for _, id in ipairs(ret) do
-		reply:tag("item", { jid = service_name, name = id }):up();
+		reply:tag("item", { jid = user_bare, name = id }):up();
 	end
 end);
 
 module:hook("account-disco-items", function(event)
 	local reply, stanza, origin = event.reply, event.stanza, event.origin;
 
-	local service_name = reply.attr.from or origin.username.."@"..origin.host
-	local service = get_pep_service(service_name);
+	local is_self = stanza.attr.to == nil;
+	local user_bare = jid_bare(stanza.attr.to);
+	local username = jid_split(stanza.attr.to);
+	if is_self then
+		username = origin.username;
+		user_bare = jid_join(username, host);
+	end
+	local service = get_pep_service(username);
+
 	local ok, ret = service:get_nodes(jid_bare(stanza.attr.from));
 	if not ok then return; end
 
 	for node, node_obj in pairs(ret) do
-		reply:tag("item", { jid = service_name, node = node, name = node_obj.config.name }):up();
+		reply:tag("item", { jid = user_bare, node = node, name = node_obj.config.name }):up();
 	end
 end);

--- /dev/null	Thu Jan 01 00:00:00 1970 +0000
+++ b/mod_pep_plus/pubsub.lib.lua	Wed Oct 18 09:56:29 2017 +0200
@@ -0,0 +1,497 @@
+local t_unpack = table.unpack or unpack; -- luacheck: ignore 113
+local time_now = os.time;
+
+local set = require "util.set";
+local st = require "util.stanza";
+local it = require "util.iterators";
+local uuid_generate = require "util.uuid".generate;
+local dataform = require"util.dataforms".new;
+
+local xmlns_pubsub = "http://jabber.org/protocol/pubsub";
+local xmlns_pubsub_errors = "http://jabber.org/protocol/pubsub#errors";
+local xmlns_pubsub_owner = "http://jabber.org/protocol/pubsub#owner";
+
+local _M = {};
+
+local handlers = {};
+_M.handlers = handlers;
+
+local pubsub_errors = {
+	["conflict"] = { "cancel", "conflict" };
+	["invalid-jid"] = { "modify", "bad-request", nil, "invalid-jid" };
+	["jid-required"] = { "modify", "bad-request", nil, "jid-required" };
+	["nodeid-required"] = { "modify", "bad-request", nil, "nodeid-required" };
+	["item-not-found"] = { "cancel", "item-not-found" };
+	["not-subscribed"] = { "modify", "unexpected-request", nil, "not-subscribed" };
+	["forbidden"] = { "auth", "forbidden" };
+	["not-allowed"] = { "cancel", "not-allowed" };
+};
+local function pubsub_error_reply(stanza, error)
+	local e = pubsub_errors[error];
+	local reply = st.error_reply(stanza, t_unpack(e, 1, 3));
+	if e[4] then
+		reply:tag(e[4], { xmlns = xmlns_pubsub_errors }):up();
+	end
+	return reply;
+end
+_M.pubsub_error_reply = pubsub_error_reply;
+
+local node_config_form = dataform {
+	{
+		type = "hidden";
+		name = "FORM_TYPE";
+		value = "http://jabber.org/protocol/pubsub#node_config";
+	};
+	{
+		type = "text-single";
+		name = "pubsub#max_items";
+		label = "Max # of items to persist";
+	};
+	{
+		type = "boolean";
+		name = "pubsub#persist_items";
+		label = "Persist items to storage";
+	};
+};
+
+local service_method_feature_map = {
+	add_subscription = { "subscribe" };
+	create = { "create-nodes", "instant-nodes", "item-ids", "create-and-configure" };
+	delete = { "delete-nodes" };
+	get_items = { "retrieve-items" };
+	get_subscriptions = { "retrieve-subscriptions" };
+	node_defaults = { "retrieve-default" };
+	publish = { "publish" };
+	purge = { "purge-nodes" };
+	retract = { "delete-items", "retract-items" };
+	set_node_config = { "config-node" };
+};
+local service_config_feature_map = {
+	autocreate_on_publish = { "auto-create" };
+};
+
+function _M.get_feature_set(service)
+	local supported_features = set.new();
+
+	for method, features in pairs(service_method_feature_map) do
+		if service[method] then
+			for _, feature in ipairs(features) do
+				if feature then
+					supported_features:add(feature);
+				end
+			end
+		end
+	end
+
+	for option, features in pairs(service_config_feature_map) do
+		if service.config[option] then
+			for _, feature in ipairs(features) do
+				if feature then
+					supported_features:add(feature);
+				end
+			end
+		end
+	end
+
+	for affiliation in pairs(service.config.capabilities) do
+		if affiliation ~= "none" and affiliation ~= "owner" then
+			supported_features:add(affiliation.."-affiliation");
+		end
+	end
+
+	return supported_features;
+end
+
+function _M.handle_pubsub_iq(event, service)
+	local origin, stanza = event.origin, event.stanza;
+	local pubsub_tag = stanza.tags[1];
+	local action = pubsub_tag.tags[1];
+	if not action then
+		return origin.send(st.error_reply(stanza, "cancel", "bad-request"));
+	end
+	local prefix = "";
+	if pubsub_tag.attr.xmlns == xmlns_pubsub_owner then
+		prefix = "owner_";
+	end
+	local handler = handlers[prefix..stanza.attr.type.."_"..action.name];
+	if handler then
+		handler(origin, stanza, action, service);
+		return true;
+	end
+end
+
+function handlers.get_items(origin, stanza, items, service)
+	local node = items.attr.node;
+	local item = items:get_child("item");
+	local item_id = item and item.attr.id;
+
+	if not node then
+		origin.send(pubsub_error_reply(stanza, "nodeid-required"));
+		return true;
+	end
+	local ok, results = service:get_items(node, stanza.attr.from, item_id);
+	if not ok then
+		origin.send(pubsub_error_reply(stanza, results));
+		return true;
+	end
+
+	local data = st.stanza("items", { node = node });
+	for _, id in ipairs(results) do
+		data:add_child(results[id]);
+	end
+	local reply;
+	if data then
+		reply = st.reply(stanza)
+			:tag("pubsub", { xmlns = xmlns_pubsub })
+				:add_child(data);
+	else
+		reply = pubsub_error_reply(stanza, "item-not-found");
+	end
+	origin.send(reply);
+	return true;
+end
+
+function handlers.get_subscriptions(origin, stanza, subscriptions, service)
+	local node = subscriptions.attr.node;
+	local ok, ret = service:get_subscriptions(node, stanza.attr.from, stanza.attr.from);
+	if not ok then
+		origin.send(pubsub_error_reply(stanza, ret));
+		return true;
+	end
+	local reply = st.reply(stanza)
+		:tag("pubsub", { xmlns = xmlns_pubsub })
+			:tag("subscriptions");
+	for _, sub in ipairs(ret) do
+		reply:tag("subscription", { node = sub.node, jid = sub.jid, subscription = 'subscribed' }):up();
+	end
+	origin.send(reply);
+	return true;
+end
+
+function handlers.set_create(origin, stanza, create, service)
+	local node = create.attr.node;
+	local ok, ret, reply;
+	local config;
+	local configure = stanza.tags[1]:get_child("configure");
+	if configure then
+		local config_form = configure:get_child("x", "jabber:x:data");
+		if not config_form then
+			origin.send(st.error_reply(stanza, "modify", "bad-request", "Missing dataform"));
+			return true;
+		end
+		local form_data, err = node_config_form:data(config_form);
+		if not form_data then
+			origin.send(st.error_reply(stanza, "modify", "bad-request", err));
+			return true;
+		end
+		config = {
+			["max_items"] = tonumber(form_data["pubsub#max_items"]);
+			["persist_items"] = form_data["pubsub#persist_items"];
+		};
+	end
+	if node then
+		ok, ret = service:create(node, stanza.attr.from, config);
+		if ok then
+			reply = st.reply(stanza);
+		else
+			reply = pubsub_error_reply(stanza, ret);
+		end
+	else
+		repeat
+			node = uuid_generate();
+			ok, ret = service:create(node, stanza.attr.from, config);
+		until ok or ret ~= "conflict";
+		if ok then
+			reply = st.reply(stanza)
+				:tag("pubsub", { xmlns = xmlns_pubsub })
+					:tag("create", { node = node });
+		else
+			reply = pubsub_error_reply(stanza, ret);
+		end
+	end
+	origin.send(reply);
+	return true;
+end
+
+function handlers.owner_set_delete(origin, stanza, delete, service)
+	local node = delete.attr.node;
+
+	local reply;
+	if not node then
+		origin.send(pubsub_error_reply(stanza, "nodeid-required"));
+		return true;
+	end
+	local ok, ret = service:delete(node, stanza.attr.from);
+	if ok then
+		reply = st.reply(stanza);
+	else
+		reply = pubsub_error_reply(stanza, ret);
+	end
+	origin.send(reply);
+	return true;
+end
+
+function handlers.set_subscribe(origin, stanza, subscribe, service)
+	local node, jid = subscribe.attr.node, subscribe.attr.jid;
+	if not (node and jid) then
+		origin.send(pubsub_error_reply(stanza, jid and "nodeid-required" or "invalid-jid"));
+		return true;
+	end
+	--[[
+	local options_tag, options = stanza.tags[1]:get_child("options"), nil;
+	if options_tag then
+		options = options_form:data(options_tag.tags[1]);
+	end
+	--]]
+	local options_tag, options; -- FIXME
+	local ok, ret = service:add_subscription(node, stanza.attr.from, jid, options);
+	local reply;
+	if ok then
+		reply = st.reply(stanza)
+			:tag("pubsub", { xmlns = xmlns_pubsub })
+				:tag("subscription", {
+					node = node,
+					jid = jid,
+					subscription = "subscribed"
+				}):up();
+		if options_tag then
+			reply:add_child(options_tag);
+		end
+	else
+		reply = pubsub_error_reply(stanza, ret);
+	end
+	origin.send(reply);
+end
+
+function handlers.set_unsubscribe(origin, stanza, unsubscribe, service)
+	local node, jid = unsubscribe.attr.node, unsubscribe.attr.jid;
+	if not (node and jid) then
+		origin.send(pubsub_error_reply(stanza, jid and "nodeid-required" or "invalid-jid"));
+		return true;
+	end
+	local ok, ret = service:remove_subscription(node, stanza.attr.from, jid);
+	local reply;
+	if ok then
+		reply = st.reply(stanza);
+	else
+		reply = pubsub_error_reply(stanza, ret);
+	end
+	origin.send(reply);
+	return true;
+end
+
+function handlers.set_publish(origin, stanza, publish, service)
+	local node = publish.attr.node;
+	if not node then
+		origin.send(pubsub_error_reply(stanza, "nodeid-required"));
+		return true;
+	end
+	local item = publish:get_child("item");
+	local id = (item and item.attr.id);
+	if not id then
+		id = uuid_generate();
+		if item then
+			item.attr.id = id;
+		end
+	end
+	local ok, ret = service:publish(node, stanza.attr.from, id, item);
+	local reply;
+	if ok then
+		reply = st.reply(stanza)
+			:tag("pubsub", { xmlns = xmlns_pubsub })
+				:tag("publish", { node = node })
+					:tag("item", { id = id });
+	else
+		reply = pubsub_error_reply(stanza, ret);
+	end
+	origin.send(reply);
+	return true;
+end
+
+function handlers.set_retract(origin, stanza, retract, service)
+	local node, notify = retract.attr.node, retract.attr.notify;
+	notify = (notify == "1") or (notify == "true");
+	local item = retract:get_child("item");
+	local id = item and item.attr.id
+	if not (node and id) then
+		origin.send(pubsub_error_reply(stanza, node and "item-not-found" or "nodeid-required"));
+		return true;
+	end
+	local reply, notifier;
+	if notify then
+		notifier = st.stanza("retract", { id = id });
+	end
+	local ok, ret = service:retract(node, stanza.attr.from, id, notifier);
+	if ok then
+		reply = st.reply(stanza);
+	else
+		reply = pubsub_error_reply(stanza, ret);
+	end
+	origin.send(reply);
+	return true;
+end
+
+function handlers.owner_set_purge(origin, stanza, purge, service)
+	local node, notify = purge.attr.node, purge.attr.notify;
+	notify = (notify == "1") or (notify == "true");
+	local reply;
+	if not node then
+		origin.send(pubsub_error_reply(stanza, "nodeid-required"));
+		return true;
+	end
+	local ok, ret = service:purge(node, stanza.attr.from, notify);
+	if ok then
+		reply = st.reply(stanza);
+	else
+		reply = pubsub_error_reply(stanza, ret);
+	end
+	origin.send(reply);
+	return true;
+end
+
+function handlers.owner_get_configure(origin, stanza, config, service)
+	local node = config.attr.node;
+	if not node then
+		origin.send(pubsub_error_reply(stanza, "nodeid-required"));
+		return true;
+	end
+
+	if not service:may(node, stanza.attr.from, "configure") then
+		origin.send(pubsub_error_reply(stanza, "forbidden"));
+		return true;
+	end
+
+	local node_obj = service.nodes[node];
+	if not node_obj then
+		origin.send(pubsub_error_reply(stanza, "item-not-found"));
+		return true;
+	end
+
+	local node_config = node_obj.config;
+	local pubsub_form_data = {
+		["pubsub#max_items"] = tostring(node_config["max_items"]);
+		["pubsub#persist_items"] = node_config["persist_items"]
+	}
+	local reply = st.reply(stanza)
+		:tag("pubsub", { xmlns = xmlns_pubsub_owner })
+			:tag("configure", { node = node })
+				:add_child(node_config_form:form(pubsub_form_data));
+	origin.send(reply);
+	return true;
+end
+
+function handlers.owner_set_configure(origin, stanza, config, service)
+	local node = config.attr.node;
+	if not node then
+		origin.send(pubsub_error_reply(stanza, "nodeid-required"));
+		return true;
+	end
+	if not service:may(node, stanza.attr.from, "configure") then
+		origin.send(pubsub_error_reply(stanza, "forbidden"));
+		return true;
+	end
+	local config_form = config:get_child("x", "jabber:x:data");
+	if not config_form then
+		origin.send(st.error_reply(stanza, "modify", "bad-request", "Missing dataform"));
+		return true;
+	end
+	local form_data, err = node_config_form:data(config_form);
+	if not form_data then
+		origin.send(st.error_reply(stanza, "modify", "bad-request", err));
+		return true;
+	end
+	local new_config = {
+		["max_items"] = tonumber(form_data["pubsub#max_items"]);
+		["persist_items"] = form_data["pubsub#persist_items"];
+	};
+	local ok, err = service:set_node_config(node, stanza.attr.from, new_config);
+	if not ok then
+		origin.send(pubsub_error_reply(stanza, err));
+		return true;
+	end
+	origin.send(st.reply(stanza));
+	return true;
+end
+
+function handlers.owner_get_default(origin, stanza, default, service) -- luacheck: ignore 212/default
+	local pubsub_form_data = {
+		["pubsub#max_items"] = tostring(service.node_defaults["max_items"]);
+		["pubsub#persist_items"] = service.node_defaults["persist_items"]
+	}
+	local reply = st.reply(stanza)
+		:tag("pubsub", { xmlns = xmlns_pubsub_owner })
+			:tag("default")
+				:add_child(node_config_form:form(pubsub_form_data));
+	origin.send(reply);
+	return true;
+end
+
+local function create_encapsulating_item(id, payload)
+	local item = st.stanza("item", { id = id, xmlns = xmlns_pubsub });
+	item:add_child(payload);
+	return item;
+end
+
+local function archive_itemstore(archive, config, user, node)
+	module:log("debug", "Creation of itemstore for node %s with config %s", node, config);
+	local get_set = {};
+	function get_set:items() -- luacheck: ignore 212/self
+		local data, err = archive:find(user, {
+			limit = tonumber(config["pubsub#max_items"]);
+			reverse = true;
+		});
+		if not data then
+			module:log("error", "Unable to get items: %s", err);
+			return true;
+		end
+		module:log("debug", "Listed items %s", data);
+		return it.reverse(function()
+			local id, payload, when, publisher = data();
+			if id == nil then
+				return;
+			end
+			local item = create_encapsulating_item(id, payload, publisher);
+			return id, item;
+		end);
+	end
+	function get_set:get(key) -- luacheck: ignore 212/self
+		local data, err = archive:find(user, {
+			key = key;
+			-- Get the last item with that key, if the archive doesn't deduplicate
+			reverse = true,
+			limit = 1;
+		});
+		if not data then
+			module:log("error", "Unable to get item: %s", err);
+			return nil, err;
+		end
+		local id, payload, when, publisher = data();
+		module:log("debug", "Get item %s (published at %s by %s)", id, when, publisher);
+		if id == nil then
+			return nil;
+		end
+		return create_encapsulating_item(id, payload, publisher);
+	end
+	function get_set:set(key, value) -- luacheck: ignore 212/self
+		local data, err;
+		if value ~= nil then
+			local publisher = value.attr.publisher;
+			local payload = value.tags[1];
+			data, err = archive:append(user, key, payload, time_now(), publisher);
+		else
+			data, err = archive:delete(user, { key = key; });
+		end
+		if not data then
+			module:log("error", "Unable to set item: %s", err);
+			return nil, err;
+		end
+		return data;
+	end
+	function get_set:clear() -- luacheck: ignore 212/self
+		return archive:delete(user);
+	end
+	return setmetatable(get_set, archive);
+end
+_M.archive_itemstore = archive_itemstore;
+
+return _M;

--- /dev/null	Thu Jan 01 00:00:00 1970 +0000
+++ b/mod_pep_plus/util_pubsub.lib.lua	Wed Oct 18 09:56:29 2017 +0200
@@ -0,0 +1,450 @@
+local events = require "util.events";
+local cache = require "util.cache";
+
+local service = {};
+local service_mt = { __index = service };
+
+local default_config = { __index = {
+	itemstore = function (config, _) return cache.new(config["max_items"]) end;
+	broadcaster = function () end;
+	get_affiliation = function () end;
+	capabilities = {};
+} };
+local default_node_config = { __index = {
+	["persist_items"] = false;
+	["max_items"] = 20;
+} };
+
+local function new(config)
+	config = config or {};
+	return setmetatable({
+		config = setmetatable(config, default_config);
+		node_defaults = setmetatable(config.node_defaults or {}, default_node_config);
+		affiliations = {};
+		subscriptions = {};
+		nodes = {};
+		data = {};
+		events = events.new();
+	}, service_mt);
+end
+
+function service:jids_equal(jid1, jid2)
+	local normalize = self.config.normalize_jid;
+	return normalize(jid1) == normalize(jid2);
+end
+
+function service:may(node, actor, action)
+	if actor == true then return true; end
+
+	local node_obj = self.nodes[node];
+	local node_aff = node_obj and node_obj.affiliations[actor];
+	local service_aff = self.affiliations[actor]
+	                 or self.config.get_affiliation(actor, node, action)
+	                 or "none";
+
+	-- Check if node allows/forbids it
+	local node_capabilities = node_obj and node_obj.capabilities;
+	if node_capabilities then
+		local caps = node_capabilities[node_aff or service_aff];
+		if caps then
+			local can = caps[action];
+			if can ~= nil then
+				return can;
+			end
+		end
+	end
+
+	-- Check service-wide capabilities instead
+	local service_capabilities = self.config.capabilities;
+	local caps = service_capabilities[node_aff or service_aff];
+	if caps then
+		local can = caps[action];
+		if can ~= nil then
+			return can;
+		end
+	end
+
+	return false;
+end
+
+function service:set_affiliation(node, actor, jid, affiliation)
+	-- Access checking
+	if not self:may(node, actor, "set_affiliation") then
+		return false, "forbidden";
+	end
+	--
+	local node_obj = self.nodes[node];
+	if not node_obj then
+		return false, "item-not-found";
+	end
+	node_obj.affiliations[jid] = affiliation;
+	local _, jid_sub = self:get_subscription(node, true, jid);
+	if not jid_sub and not self:may(node, jid, "be_unsubscribed") then
+		local ok, err = self:add_subscription(node, true, jid);
+		if not ok then
+			return ok, err;
+		end
+	elseif jid_sub and not self:may(node, jid, "be_subscribed") then
+		local ok, err = self:add_subscription(node, true, jid);
+		if not ok then
+			return ok, err;
+		end
+	end
+	return true;
+end
+
+function service:add_subscription(node, actor, jid, options)
+	-- Access checking
+	local cap;
+	if actor == true or jid == actor or self:jids_equal(actor, jid) then
+		cap = "subscribe";
+	else
+		cap = "subscribe_other";
+	end
+	if not self:may(node, actor, cap) then
+		return false, "forbidden";
+	end
+	if not self:may(node, jid, "be_subscribed") then
+		return false, "forbidden";
+	end
+	--
+	local node_obj = self.nodes[node];
+	if not node_obj then
+		if not self.config.autocreate_on_subscribe then
+			return false, "item-not-found";
+		else
+			local ok, err = self:create(node, true);
+			if not ok then
+				return ok, err;
+			end
+			node_obj = self.nodes[node];
+		end
+	end
+	node_obj.subscribers[jid] = options or true;
+	local normal_jid = self.config.normalize_jid(jid);
+	local subs = self.subscriptions[normal_jid];
+	if subs then
+		if not subs[jid] then
+			subs[jid] = { [node] = true };
+		else
+			subs[jid][node] = true;
+		end
+	else
+		self.subscriptions[normal_jid] = { [jid] = { [node] = true } };
+	end
+	self.events.fire_event("subscription-added", { node = node, jid = jid, normalized_jid = normal_jid, options = options });
+	return true;
+end
+
+function service:remove_subscription(node, actor, jid)
+	-- Access checking
+	local cap;
+	if actor == true or jid == actor or self:jids_equal(actor, jid) then
+		cap = "unsubscribe";
+	else
+		cap = "unsubscribe_other";
+	end
+	if not self:may(node, actor, cap) then
+		return false, "forbidden";
+	end
+	if not self:may(node, jid, "be_unsubscribed") then
+		return false, "forbidden";
+	end
+	--
+	local node_obj = self.nodes[node];
+	if not node_obj then
+		return false, "item-not-found";
+	end
+	if not node_obj.subscribers[jid] then
+		return false, "not-subscribed";
+	end
+	node_obj.subscribers[jid] = nil;
+	local normal_jid = self.config.normalize_jid(jid);
+	local subs = self.subscriptions[normal_jid];
+	if subs then
+		local jid_subs = subs[jid];
+		if jid_subs then
+			jid_subs[node] = nil;
+			if next(jid_subs) == nil then
+				subs[jid] = nil;
+			end
+		end
+		if next(subs) == nil then
+			self.subscriptions[normal_jid] = nil;
+		end
+	end
+	self.events.fire_event("subscription-removed", { node = node, jid = jid, normalized_jid = normal_jid });
+	return true;
+end
+
+function service:remove_all_subscriptions(actor, jid)
+	local normal_jid = self.config.normalize_jid(jid);
+	local subs = self.subscriptions[normal_jid]
+	subs = subs and subs[jid];
+	if subs then
+		for node in pairs(subs) do
+			self:remove_subscription(node, true, jid);
+		end
+	end
+	return true;
+end
+
+function service:get_subscription(node, actor, jid)
+	-- Access checking
+	local cap;
+	if actor == true or jid == actor or self:jids_equal(actor, jid) then
+		cap = "get_subscription";
+	else
+		cap = "get_subscription_other";
+	end
+	if not self:may(node, actor, cap) then
+		return false, "forbidden";
+	end
+	--
+	local node_obj = self.nodes[node];
+	if not node_obj then
+		return false, "item-not-found";
+	end
+	return true, node_obj.subscribers[jid];
+end
+
+function service:create(node, actor, options)
+	-- Access checking
+	if not self:may(node, actor, "create") then
+		return false, "forbidden";
+	end
+	--
+	if self.nodes[node] then
+		return false, "conflict";
+	end
+
+	self.nodes[node] = {
+		name = node;
+		subscribers = {};
+		config = setmetatable(options or {}, {__index=self.node_defaults});
+		affiliations = {};
+	};
+	self.data[node] = self.config.itemstore(self.nodes[node].config, node);
+	self.events.fire_event("node-created", { node = node, actor = actor });
+	local ok, err = self:set_affiliation(node, true, actor, "owner");
+	if not ok then
+		self.nodes[node] = nil;
+		self.data[node] = nil;
+	end
+	return ok, err;
+end
+
+function service:delete(node, actor)
+	-- Access checking
+	if not self:may(node, actor, "delete") then
+		return false, "forbidden";
+	end
+	--
+	local node_obj = self.nodes[node];
+	if not node_obj then
+		return false, "item-not-found";
+	end
+	self.nodes[node] = nil;
+	if self.data[node] and self.data[node].clear then
+		self.data[node]:clear();
+	end
+	self.data[node] = nil;
+	self.events.fire_event("node-deleted", { node = node, actor = actor });
+	self.config.broadcaster("delete", node, node_obj.subscribers);
+	return true;
+end
+
+function service:publish(node, actor, id, item)
+	-- Access checking
+	if not self:may(node, actor, "publish") then
+		return false, "forbidden";
+	end
+	--
+	local node_obj = self.nodes[node];
+	if not node_obj then
+		if not self.config.autocreate_on_publish then
+			return false, "item-not-found";
+		end
+		local ok, err = self:create(node, true);
+		if not ok then
+			return ok, err;
+		end
+		node_obj = self.nodes[node];
+	end
+	local node_data = self.data[node];
+	local ok = node_data:set(id, item);
+	if not ok then
+		return nil, "internal-server-error";
+	end
+	if type(ok) == "string" then id = ok; end
+	self.events.fire_event("item-published", { node = node, actor = actor, id = id, item = item });
+	self.config.broadcaster("items", node, node_obj.subscribers, item, actor);
+	return true;
+end
+
+function service:retract(node, actor, id, retract)
+	-- Access checking
+	if not self:may(node, actor, "retract") then
+		return false, "forbidden";
+	end
+	--
+	local node_obj = self.nodes[node];
+	if (not node_obj) or (not self.data[node]:get(id)) then
+		return false, "item-not-found";
+	end
+	local ok = self.data[node]:set(id, nil);
+	if not ok then
+		return nil, "internal-server-error";
+	end
+	self.events.fire_event("item-retracted", { node = node, actor = actor, id = id });
+	if retract then
+		self.config.broadcaster("items", node, node_obj.subscribers, retract);
+	end
+	return true
+end
+
+function service:purge(node, actor, notify)
+	-- Access checking
+	if not self:may(node, actor, "retract") then
+		return false, "forbidden";
+	end
+	--
+	local node_obj = self.nodes[node];
+	if not node_obj then
+		return false, "item-not-found";
+	end
+	if self.data[node] and self.data[node].clear then
+		self.data[node]:clear()
+	else
+		self.data[node] = self.config.itemstore(self.nodes[node].config, node);
+	end
+	self.events.fire_event("node-purged", { node = node, actor = actor });
+	if notify then
+		self.config.broadcaster("purge", node, node_obj.subscribers);
+	end
+	return true
+end
+
+function service:get_items(node, actor, id)
+	-- Access checking
+	if not self:may(node, actor, "get_items") then
+		return false, "forbidden";
+	end
+	--
+	local node_obj = self.nodes[node];
+	if not node_obj then
+		return false, "item-not-found";
+	end
+	if id then -- Restrict results to a single specific item
+		local with_id = self.data[node]:get(id);
+		if not with_id then
+			return false, "item-not-found";
+		end
+		return true, { id, [id] = with_id };
+	else
+		local data = {}
+		for key, value in self.data[node]:items() do
+			data[#data+1] = key;
+			data[key] = value;
+		end
+		return true, data;
+	end
+end
+
+function service:get_nodes(actor)
+	-- Access checking
+	if not self:may(nil, actor, "get_nodes") then
+		return false, "forbidden";
+	end
+	--
+	return true, self.nodes;
+end
+
+function service:get_subscriptions(node, actor, jid)
+	-- Access checking
+	local cap;
+	if actor == true or jid == actor or self:jids_equal(actor, jid) then
+		cap = "get_subscriptions";
+	else
+		cap = "get_subscriptions_other";
+	end
+	if not self:may(node, actor, cap) then
+		return false, "forbidden";
+	end
+	--
+	local node_obj;
+	if node then
+		node_obj = self.nodes[node];
+		if not node_obj then
+			return false, "item-not-found";
+		end
+	end
+	local normal_jid = self.config.normalize_jid(jid);
+	local subs = self.subscriptions[normal_jid];
+	-- We return the subscription object from the node to save
+	-- a get_subscription() call for each node.
+	local ret = {};
+	if subs then
+		for subscribed_jid, subscribed_nodes in pairs(subs) do
+			if node then -- Return only subscriptions to this node
+				if subscribed_nodes[node] then
+					ret[#ret+1] = {
+						node = node;
+						jid = subscribed_jid;
+						subscription = node_obj.subscribers[subscribed_jid];
+					};
+				end
+			else -- Return subscriptions to all nodes
+				local nodes = self.nodes;
+				for subscribed_node in pairs(subscribed_nodes) do
+					ret[#ret+1] = {
+						node = subscribed_node;
+						jid = subscribed_jid;
+						subscription = nodes[subscribed_node].subscribers[subscribed_jid];
+					};
+				end
+			end
+		end
+	end
+	return true, ret;
+end
+
+-- Access models only affect 'none' affiliation caps, service/default access level...
+function service:set_node_capabilities(node, actor, capabilities)
+	-- Access checking
+	if not self:may(node, actor, "configure") then
+		return false, "forbidden";
+	end
+	--
+	local node_obj = self.nodes[node];
+	if not node_obj then
+		return false, "item-not-found";
+	end
+	node_obj.capabilities = capabilities;
+	return true;
+end
+
+function service:set_node_config(node, actor, new_config)
+	if not self:may(node, actor, "configure") then
+		return false, "forbidden";
+	end
+
+	local node_obj = self.nodes[node];
+	if not node_obj then
+		return false, "item-not-found";
+	end
+
+	for k,v in pairs(new_config) do
+		node_obj.config[k] = v;
+	end
+	local new_data = self.config.itemstore(self.nodes[node].config, node);
+	for key, value in self.data[node]:items() do
+		new_data:set(key, value);
+	end
+	self.data[node] = new_data;
+	return true;
+end
+
+return {
+	new = new;
+};