Mercurial > prosody-hg

comparison plugins/mod_storage_internal.lua @ 10411:db2a06b9ff98

Find changesets by keywords (author, files, the commit message), revision number or hash, or revset expression.
Merge 0.11->trunk
author Kim Alvefur <zash@zash.se>
date Sat, 16 Nov 2019 16:52:31 +0100
parents d185c4961ee0
children f23363380599
comparison
equal deleted inserted replaced
10410:659b577f280c 10411:db2a06b9ff98
1 local cache = require "util.cache";
1 local datamanager = require "core.storagemanager".olddm; 2 local datamanager = require "core.storagemanager".olddm;
2 local array = require "util.array"; 3 local array = require "util.array";
3 local datetime = require "util.datetime"; 4 local datetime = require "util.datetime";
4 local st = require "util.stanza"; 5 local st = require "util.stanza";
5 local now = require "util.time".now; 6 local now = require "util.time".now;
6 local id = require "util.id".medium; 7 local id = require "util.id".medium;
8 local jid_join = require "util.jid".join;
7 9
8 local host = module.host; 10 local host = module.host;
11
12 local archive_item_limit = module:get_option_number("storage_archive_item_limit", 10000);
13 local archive_item_count_cache = cache.new(module:get_option("storage_archive_item_limit_cache_size", 1000));
9 14
10 local driver = {}; 15 local driver = {};
11 16
12 function driver:open(store, typ) 17 function driver:open(store, typ)
13 local mt = self[typ or "keyval"] 18 local mt = self[typ or "keyval"]
40 return datamanager.users(host, self.store, self.type); 45 return datamanager.users(host, self.store, self.type);
41 end 46 end
42 47
43 local archive = {}; 48 local archive = {};
44 driver.archive = { __index = archive }; 49 driver.archive = { __index = archive };
50
51 archive.caps = {
52 total = true;
53 quota = archive_item_limit;
54 truncate = true;
55 };
45 56
46 function archive:append(username, key, value, when, with) 57 function archive:append(username, key, value, when, with)
47 when = when or now(); 58 when = when or now();
48 if not st.is_stanza(value) then 59 if not st.is_stanza(value) then
49 return nil, "unsupported-datatype"; 60 return nil, "unsupported-datatype";
52 value.when = when; 63 value.when = when;
53 value.with = with; 64 value.with = with;
54 value.attr.stamp = datetime.datetime(when); 65 value.attr.stamp = datetime.datetime(when);
55 value.attr.stamp_legacy = datetime.legacy(when); 66 value.attr.stamp_legacy = datetime.legacy(when);
56 67
68 local cache_key = jid_join(username, host, self.store);
69 local item_count = archive_item_count_cache:get(cache_key);
70
57 if key then 71 if key then
58 local items, err = datamanager.list_load(username, host, self.store); 72 local items, err = datamanager.list_load(username, host, self.store);
59 if not items and err then return items, err; end 73 if not items and err then return items, err; end
74
75 -- Check the quota
76 item_count = items and #items or 0;
77 archive_item_count_cache:set(cache_key, item_count);
78 if item_count >= archive_item_limit then
79 module:log("debug", "%s reached or over quota, not adding to store", username);
80 return nil, "quota-limit";
81 end
82
60 if items then 83 if items then
84 -- Filter out any item with the same key as the one being added
61 items = array(items); 85 items = array(items);
62 items:filter(function (item) 86 items:filter(function (item)
63 return item.key ~= key; 87 return item.key ~= key;
64 end); 88 end);
89
65 value.key = key; 90 value.key = key;
66 items:push(value); 91 items:push(value);
67 local ok, err = datamanager.list_store(username, host, self.store, items); 92 local ok, err = datamanager.list_store(username, host, self.store, items);
68 if not ok then return ok, err; end 93 if not ok then return ok, err; end
94 archive_item_count_cache:set(cache_key, #items);
69 return key; 95 return key;
70 end 96 end
71 else 97 else
98 if not item_count then -- Item count not cached?
99 -- We need to load the list to get the number of items currently stored
100 local items, err = datamanager.list_load(username, host, self.store);
101 if not items and err then return items, err; end
102 item_count = items and #items or 0;
103 archive_item_count_cache:set(cache_key, item_count);
104 end
105 if item_count >= archive_item_limit then
106 module:log("debug", "%s reached or over quota, not adding to store", username);
107 return nil, "quota-limit";
108 end
72 key = id(); 109 key = id();
73 end 110 end
111
112 module:log("debug", "%s has %d items out of %d limit in store %s", username, item_count, archive_item_limit, self.store);
74 113
75 value.key = key; 114 value.key = key;
76 115
77 local ok, err = datamanager.list_append(username, host, self.store, value); 116 local ok, err = datamanager.list_append(username, host, self.store, value);
78 if not ok then return ok, err; end 117 if not ok then return ok, err; end
118 archive_item_count_cache:set(cache_key, item_count+1);
79 return key; 119 return key;
80 end 120 end
81 121
82 function archive:find(username, query) 122 function archive:find(username, query)
83 local items, err = datamanager.list_load(username, host, self.store); 123 local items, err = datamanager.list_load(username, host, self.store);
84 if not items then 124 if not items then
85 if err then 125 if err then
86 return items, err; 126 return items, err;
87 else 127 elseif query then
88 return function () end, 0; 128 if query.before or query.after then
89 end 129 return nil, "item-not-found";
90 end 130 end
91 local count = #items; 131 if query.total then
132 return function () end, 0;
133 end
134 end
135 return function () end;
136 end
137 local count = nil;
92 local i = 0; 138 local i = 0;
93 if query then 139 if query then
94 items = array(items); 140 items = array(items);
95 if query.key then 141 if query.key then
96 items:filter(function (item) 142 items:filter(function (item)
110 if query["end"] then 156 if query["end"] then
111 items:filter(function (item) 157 items:filter(function (item)
112 return item.when <= query["end"]; 158 return item.when <= query["end"];
113 end); 159 end);
114 end 160 end
115 count = #items; 161 if query.total then
162 count = #items;
163 end
116 if query.reverse then 164 if query.reverse then
117 items:reverse(); 165 items:reverse();
118 if query.before then 166 if query.before then
119 for j = 1, count do 167 local found = false;
168 for j = 1, #items do
120 if (items[j].key or tostring(j)) == query.before then 169 if (items[j].key or tostring(j)) == query.before then
170 found = true;
121 i = j; 171 i = j;
122 break; 172 break;
123 end 173 end
124 end 174 end
175 if not found then
176 return nil, "item-not-found";
177 end
125 end 178 end
126 elseif query.after then 179 elseif query.after then
127 for j = 1, count do 180 local found = false;
181 for j = 1, #items do
128 if (items[j].key or tostring(j)) == query.after then 182 if (items[j].key or tostring(j)) == query.after then
183 found = true;
129 i = j; 184 i = j;
130 break; 185 break;
131 end 186 end
187 end
188 if not found then
189 return nil, "item-not-found";
132 end 190 end
133 end 191 end
134 if query.limit and #items - i > query.limit then 192 if query.limit and #items - i > query.limit then
135 items[i+query.limit+1] = nil; 193 items[i+query.limit+1] = nil;
136 end 194 end
154 local items, err = datamanager.list_load(username, host, self.store); 212 local items, err = datamanager.list_load(username, host, self.store);
155 if not items then return items, err; end 213 if not items then return items, err; end
156 return array(items):pluck("when"):map(datetime.date):unique(); 214 return array(items):pluck("when"):map(datetime.date):unique();
157 end 215 end
158 216
217 function archive:summary(username, query)
218 local iter, err = self:find(username, query)
219 if not iter then return iter, err; end
220 local counts = {};
221 local earliest = {};
222 local latest = {};
223 local body = {};
224 for _, stanza, when, with in iter do
225 counts[with] = (counts[with] or 0) + 1;
226 if earliest[with] == nil then
227 earliest[with] = when;
228 end
229 latest[with] = when;
230 body[with] = stanza:get_child_text("body") or body[with];
231 end
232 return {
233 counts = counts;
234 earliest = earliest;
235 latest = latest;
236 body = body;
237 };
238 end
239
240 function archive:users()
241 return datamanager.users(host, self.store, "list");
242 end
243
159 function archive:delete(username, query) 244 function archive:delete(username, query)
245 local cache_key = jid_join(username, host, self.store);
160 if not query or next(query) == nil then 246 if not query or next(query) == nil then
247 archive_item_count_cache:set(cache_key, nil);
161 return datamanager.list_store(username, host, self.store, nil); 248 return datamanager.list_store(username, host, self.store, nil);
162 end 249 end
163 local items, err = datamanager.list_load(username, host, self.store); 250 local items, err = datamanager.list_load(username, host, self.store);
164 if not items then 251 if not items then
165 if err then 252 if err then
166 return items, err; 253 return items, err;
167 end 254 end
255 archive_item_count_cache:set(cache_key, 0);
168 -- Store is empty 256 -- Store is empty
169 return 0; 257 return 0;
170 end 258 end
171 items = array(items); 259 items = array(items);
172 local count_before = #items; 260 local count_before = #items;
212 if count == 0 then 300 if count == 0 then
213 return 0; -- No changes, skip write 301 return 0; -- No changes, skip write
214 end 302 end
215 local ok, err = datamanager.list_store(username, host, self.store, items); 303 local ok, err = datamanager.list_store(username, host, self.store, items);
216 if not ok then return ok, err; end 304 if not ok then return ok, err; end
305 archive_item_count_cache:set(cache_key, #items);
217 return count; 306 return count;
218 end 307 end
219 308
220 module:provides("storage", driver); 309 module:provides("storage", driver);