Mercurial > prosody-modules
view mod_muc_rai/mod_muc_rai.lua @ 5696:66986f5271c3
mod_storage_s3: Skip archive items matching on date but not full datetime
Since it only encodes dates in paths, it would have returned items from
outside the specified start..end range if they were from earlier or
later in the same (UTC) day.
author | Kim Alvefur <zash@zash.se> |
---|---|
date | Sat, 11 Nov 2023 22:26:39 +0100 |
parents | 9377b5593cc7 |
children |
line wrap: on
line source
local cache = require "util.cache"; local jid = require "util.jid"; local st = require "util.stanza"; local max_subscribers = module:get_option_number("muc_rai_max_subscribers", 1024); local muc_affiliation_store = module:open_store("config", "map"); local muc_archive = module:open_store("muc_log", "archive"); local xmlns_rai = "urn:xmpp:rai:0"; local muc_markers = module:depends("muc_markers"); -- subscriber_jid -> { [room_jid] = interested } local subscribed_users = cache.new(max_subscribers, false); -- room_jid -> { [user_jid] = interested } local interested_users = {}; -- room_jid -> last_id local room_activity_cache = cache.new(1024); -- Send a single notification for a room, updating data structures as needed local function send_single_notification(user_jid, room_jid) local notification = st.message({ to = user_jid, from = module.host }) :tag("rai", { xmlns = xmlns_rai }) :text_tag("activity", room_jid) :up(); local interested_room_users = interested_users[room_jid]; if interested_room_users then interested_room_users[user_jid] = nil; end local interested_rooms = subscribed_users:get(user_jid); if interested_rooms then interested_rooms[room_jid] = nil; end module:log("debug", "Sending notification from %s to %s", room_jid, user_jid); return module:send(notification); end local function subscribe_room(user_jid, room_jid) local interested_rooms = subscribed_users:get(user_jid); if not interested_rooms then return nil, "not-subscribed"; end module:log("debug", "Subscribed %s to %s", user_jid, room_jid); interested_rooms[room_jid] = true; local interested_room_users = interested_users[room_jid]; if not interested_room_users then interested_room_users = {}; interested_users[room_jid] = interested_room_users; end interested_room_users[user_jid] = true; return true; end local function unsubscribe_room(user_jid, room_jid) local interested_rooms = subscribed_users:get(user_jid); if not interested_rooms then return nil, "not-subscribed"; end interested_rooms[room_jid] = nil; local interested_room_users = interested_users[room_jid]; if not interested_room_users then return true; end interested_room_users[user_jid] = nil; return true; end local function notify_interested_users(room_jid) module:log("warn", "NOTIFYING FOR %s", room_jid) local interested_room_users = interested_users[room_jid]; if not interested_room_users then module:log("debug", "Nobody interested in %s", room_jid); return; end for user_jid in pairs(interested_room_users) do send_single_notification(user_jid, room_jid); end return true; end local function unsubscribe_user_from_all_rooms(user_jid) local interested_rooms = subscribed_users:get(user_jid); if not interested_rooms then return nil, "not-subscribed"; end for room_jid in pairs(interested_rooms) do unsubscribe_room(user_jid, room_jid); end return true; end local function get_last_room_message_id(room_jid) local last_room_message_id = room_activity_cache:get(room_jid); if last_room_message_id then module:log("debug", "Last room message: %s (cached)", last_room_message_id); return last_room_message_id; end -- Load all the data! local query = { limit = 1; reverse = true; with = "message<groupchat"; } local data, err = muc_archive:find(jid.node(room_jid), query); if not data then module:log("error", "Could not fetch history: %s", err); return nil; end local id = data(); room_activity_cache:set(room_jid, id); module:log("debug", "Last room message: %s (queried)", id); return id; end local function update_room_activity(room_jid, last_id) room_activity_cache:set(room_jid, last_id); end local function get_last_user_read_id(user_jid, room_jid) return muc_markers.get_user_read_marker(jid.bare(user_jid), room_jid); end local function has_new_activity(room_jid, user_jid) local last_room_message_id = get_last_room_message_id(room_jid); local last_user_read_id = get_last_user_read_id(user_jid, room_jid); module:log("debug", "Checking activity in <%s> (%s) for <%s> (%s): %s", room_jid, last_room_message_id, user_jid, last_user_read_id, tostring(last_room_message_id ~= last_user_read_id) ); return last_room_message_id ~= last_user_read_id; end -- Returns a set of rooms that a user is interested in local function get_interested_rooms(user_jid) -- Use affiliation as an indication of interest, return -- all rooms a user is affiliated return muc_affiliation_store:get_all(jid.bare(user_jid)); end local function is_subscribed(user_jid) return not not subscribed_users:get(user_jid); end -- Subscribes to all rooms that the user has an interest in -- Returns a set of room JIDs that have already had activity (thus no subscription) local function subscribe_all_rooms(user_jid) if is_subscribed(user_jid) then return nil; end -- Send activity notifications for all relevant rooms local interested_rooms, err = get_interested_rooms(user_jid); if not interested_rooms then if err then return nil, "internal-server-error"; end interested_rooms = {}; end if not subscribed_users:set(user_jid, {}) then module:log("warn", "Subscriber limit (%d) reached, rejecting subscription from %s", max_subscribers, user_jid); return nil, "resource-constraint"; end local rooms_with_activity; for room_name in pairs(interested_rooms) do local room_jid = room_name.."@"..module.host; if has_new_activity(room_jid, user_jid) then -- There has already been activity, include this room -- in the response if not rooms_with_activity then rooms_with_activity = {}; end rooms_with_activity[room_jid] = true; else -- Subscribe to any future activity subscribe_room(user_jid, room_jid); end end return rooms_with_activity; end module:hook("muc-occupant-pre-join", function(event) local room_jid, user_jid = event.room.jid, event.stanza.attr.from; local ok, _ = unsubscribe_room(user_jid, room_jid); if ok then module:log("debug", "Unsubscribed %s to %s Reason: muc-occupant-joined", user_jid, room_jid) end end, -500); module:hook("muc-occupant-pre-leave", function(event) local room_jid, user_jid = event.room.jid, event.stanza.attr.from; local ok, _ = subscribe_room(user_jid, room_jid); if ok then module:log("debug", "Subscribed %s to %s Reason: muc-occupant-left", user_jid, room_jid) end end, -500); module:hook("presence/host", function (event) local origin, stanza = event.origin, event.stanza; local user_jid = stanza.attr.from; if stanza.attr.type == "unavailable" then -- User going offline unsubscribe_user_from_all_rooms(user_jid); return true; end if not stanza:get_child("rai", xmlns_rai) then return; -- Ignore, no <rai/> tag end local rooms_with_activity, err = subscribe_all_rooms(user_jid); if not rooms_with_activity then if not err then module:log("debug", "No activity to notify"); return true; else return origin.send(st.error_reply(stanza, "wait", "resource-constraint")); end end local reply = st.message({ to = stanza.attr.from, from = module.host }) :tag("rai", { xmlns = xmlns_rai }); for room_jid in pairs(rooms_with_activity) do reply:text_tag("activity", room_jid); end return origin.send(reply); end); module:hook("muc-broadcast-message", function (event) local room, stanza = event.room, event.stanza; local archive_id = stanza:get_child("stanza-id", "urn:xmpp:sid:0"); if archive_id and archive_id.attr.id then -- Remember the id of the last message so we can compare it -- to the per-user marker (managed by mod_muc_markers) update_room_activity(room.jid, archive_id.attr.id); -- Notify any users that need to be notified notify_interested_users(room.jid); end end, -1);