Mercurial > prosody-hg

comparison plugins/mod_storage_sql.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 51f145094648
children 3098eac31139
comparison
equal deleted inserted replaced
10410:659b577f280c 10411:db2a06b9ff98
1 1
2 -- luacheck: ignore 212/self 2 -- luacheck: ignore 212/self
3 3
4 local cache = require "util.cache";
4 local json = require "util.json"; 5 local json = require "util.json";
5 local sql = require "util.sql"; 6 local sql = require "util.sql";
6 local xml_parse = require "util.xml".parse; 7 local xml_parse = require "util.xml".parse;
7 local uuid = require "util.uuid"; 8 local uuid = require "util.uuid";
8 local resolve_relative_path = require "util.paths".resolve_relative_path; 9 local resolve_relative_path = require "util.paths".resolve_relative_path;
10 local jid_join = require "util.jid".join;
9 11
10 local is_stanza = require"util.stanza".is_stanza; 12 local is_stanza = require"util.stanza".is_stanza;
11 local t_concat = table.concat; 13 local t_concat = table.concat;
12 14
13 local noop = function() end 15 local noop = function() end
14 local unpack = table.unpack or unpack; 16 local unpack = table.unpack or unpack; -- luacheck: ignore 113
15 local function iterator(result) 17 local function iterator(result)
16 return function(result_) 18 return function(result_)
17 local row = result_(); 19 local row = result_();
18 if row ~= nil then 20 if row ~= nil then
19 return unpack(row); 21 return unpack(row);
146 return iterator(result); 148 return iterator(result);
147 end 149 end
148 150
149 --- Archive store API 151 --- Archive store API
150 152
151 -- luacheck: ignore 512 431/user 431/store 153 local archive_item_limit = module:get_option_number("storage_archive_item_limit");
154 local archive_item_count_cache = cache.new(module:get_option("storage_archive_item_limit_cache_size", 1000));
155
156 -- luacheck: ignore 512 431/user 431/store 431/err
152 local map_store = {}; 157 local map_store = {};
153 map_store.__index = map_store; 158 map_store.__index = map_store;
154 map_store.remove = {}; 159 map_store.remove = {};
155 function map_store:get(username, key) 160 function map_store:get(username, key)
156 local ok, result = engine:transaction(function() 161 local ok, result = engine:transaction(function()
226 end 231 end
227 232
228 local archive_store = {} 233 local archive_store = {}
229 archive_store.caps = { 234 archive_store.caps = {
230 total = true; 235 total = true;
236 quota = archive_item_limit;
237 truncate = true;
231 }; 238 };
232 archive_store.__index = archive_store 239 archive_store.__index = archive_store
233 function archive_store:append(username, key, value, when, with) 240 function archive_store:append(username, key, value, when, with)
234 local user,store = username,self.store; 241 local user,store = username,self.store;
242 local cache_key = jid_join(username, host, store);
243 local item_count = archive_item_count_cache:get(cache_key);
244 if not item_count then
245 local ok, ret = engine:transaction(function()
246 local count_sql = [[
247 SELECT COUNT(*) FROM "prosodyarchive"
248 WHERE "host"=? AND "user"=? AND "store"=?;
249 ]];
250 local result = engine:select(count_sql, host, user, store);
251 if result then
252 for row in result do
253 item_count = row[1];
254 end
255 end
256 end);
257 if not ok or not item_count then
258 module:log("error", "Failed while checking quota for %s: %s", username, ret);
259 return nil, "Failure while checking quota";
260 end
261 archive_item_count_cache:set(cache_key, item_count);
262 end
263
264 if archive_item_limit then
265 module:log("debug", "%s has %d items out of %d limit", username, item_count, archive_item_limit);
266 if item_count >= archive_item_limit then
267 return nil, "quota-limit";
268 end
269 end
270
235 when = when or os.time(); 271 when = when or os.time();
236 with = with or ""; 272 with = with or "";
237 local ok, ret = engine:transaction(function() 273 local ok, ret = engine:transaction(function()
238 local delete_sql = [[ 274 local delete_sql = [[
239 DELETE FROM "prosodyarchive" 275 DELETE FROM "prosodyarchive"
243 INSERT INTO "prosodyarchive" 279 INSERT INTO "prosodyarchive"
244 ("host", "user", "store", "when", "with", "key", "type", "value") 280 ("host", "user", "store", "when", "with", "key", "type", "value")
245 VALUES (?,?,?,?,?,?,?,?); 281 VALUES (?,?,?,?,?,?,?,?);
246 ]]; 282 ]];
247 if key then 283 if key then
248 engine:delete(delete_sql, host, user or "", store, key); 284 local result, err = engine:delete(delete_sql, host, user or "", store, key);
285 if result then
286 item_count = item_count - result:affected();
287 end
249 else 288 else
250 key = uuid.generate(); 289 key = uuid.generate();
251 end 290 end
252 local t, encoded_value = assert(serialize(value)); 291 local t, encoded_value = assert(serialize(value));
253 engine:insert(insert_sql, host, user or "", store, when, with, key, t, encoded_value); 292 engine:insert(insert_sql, host, user or "", store, when, with, key, t, encoded_value);
293 archive_item_count_cache:set(cache_key, item_count+1);
254 return key; 294 return key;
255 end); 295 end);
256 if not ok then return ok, ret; end 296 if not ok then return ok, ret; end
257 return ret; -- the key 297 return ret; -- the key
258 end 298 end
285 where[#where+1] = "\"key\" = ?"; 325 where[#where+1] = "\"key\" = ?";
286 args[#args+1] = query.key 326 args[#args+1] = query.key
287 end 327 end
288 end 328 end
289 local function archive_where_id_range(query, args, where) 329 local function archive_where_id_range(query, args, where)
290 local args_len = #args
291 -- Before or after specific item, exclusive 330 -- Before or after specific item, exclusive
331 local id_lookup_sql = [[
332 SELECT "sort_id"
333 FROM "prosodyarchive"
334 WHERE "key" = ? AND "host" = ? AND "user" = ? AND "store" = ?
335 LIMIT 1;
336 ]];
292 if query.after then -- keys better be unique! 337 if query.after then -- keys better be unique!
293 where[#where+1] = [[ 338 local after_id = nil;
294 "sort_id" > COALESCE( 339 for row in engine:select(id_lookup_sql, query.after, args[1], args[2], args[3]) do
295 ( 340 after_id = row[1];
296 SELECT "sort_id" 341 end
297 FROM "prosodyarchive" 342 if not after_id then
298 WHERE "key" = ? AND "host" = ? AND "user" = ? AND "store" = ? 343 return nil, "item-not-found";
299 LIMIT 1 344 end
300 ), 0) 345 where[#where+1] = '"sort_id" > ?';
301 ]]; 346 args[#args+1] = after_id;
302 args[args_len+1], args[args_len+2], args[args_len+3], args[args_len+4] = query.after, args[1], args[2], args[3];
303 args_len = args_len + 4
304 end 347 end
305 if query.before then 348 if query.before then
306 where[#where+1] = [[ 349 local before_id = nil;
307 "sort_id" < COALESCE( 350 for row in engine:select(id_lookup_sql, query.after, args[1], args[2], args[3]) do
308 ( 351 before_id = row[1];
309 SELECT "sort_id" 352 end
310 FROM "prosodyarchive" 353 if not before_id then
311 WHERE "key" = ? AND "host" = ? AND "user" = ? AND "store" = ? 354 return nil, "item-not-found";
312 LIMIT 1 355 end
313 ), 356 where[#where+1] = '"sort_id" < ?';
314 ( 357 args[#args+1] = before_id;
315 SELECT MAX("sort_id")+1 358 end
316 FROM "prosodyarchive" 359 return true;
317 )
318 )
319 ]]
320 args[args_len+1], args[args_len+2], args[args_len+3], args[args_len+4] = query.before, args[1], args[2], args[3];
321 end
322 end 360 end
323 361
324 function archive_store:find(username, query) 362 function archive_store:find(username, query)
325 query = query or {}; 363 query = query or {};
326 local user,store = username,self.store; 364 local user,store = username,self.store;
327 local total; 365 local cache_key = jid_join(username, host, self.store);
328 local ok, result = engine:transaction(function() 366 local total = archive_item_count_cache:get(cache_key);
367 if total ~= nil and query.limit == 0 and query.start == nil and query.with == nil and query["end"] == nil and query.key == nil then
368 return noop, total;
369 end
370 local ok, result, err = engine:transaction(function()
329 local sql_query = [[ 371 local sql_query = [[
330 SELECT "key", "type", "value", "when", "with" 372 SELECT "key", "type", "value", "when", "with"
331 FROM "prosodyarchive" 373 FROM "prosodyarchive"
332 WHERE %s 374 WHERE %s
333 ORDER BY "sort_id" %s%s; 375 ORDER BY "sort_id" %s%s;
344 if stats then 386 if stats then
345 for row in stats do 387 for row in stats do
346 total = row[1]; 388 total = row[1];
347 end 389 end
348 end 390 end
391 if query.start == nil and query.with == nil and query["end"] == nil and query.key == nil then
392 archive_item_count_cache:set(cache_key, total);
393 end
349 if query.limit == 0 then -- Skip the real query 394 if query.limit == 0 then -- Skip the real query
350 return noop, total; 395 return noop, total;
351 end 396 end
352 end 397 end
353 398
354 archive_where_id_range(query, args, where); 399 local ok, err = archive_where_id_range(query, args, where);
400 if not ok then return ok, err; end
355 401
356 if query.limit then 402 if query.limit then
357 args[#args+1] = query.limit; 403 args[#args+1] = query.limit;
358 end 404 end
359 405
360 sql_query = sql_query:format(t_concat(where, " AND "), query.reverse 406 sql_query = sql_query:format(t_concat(where, " AND "), query.reverse
361 and "DESC" or "ASC", query.limit and " LIMIT ?" or ""); 407 and "DESC" or "ASC", query.limit and " LIMIT ?" or "");
362 return engine:select(sql_query, unpack(args)); 408 return engine:select(sql_query, unpack(args));
363 end); 409 end);
364 if not ok then return ok, result end 410 if not ok then return ok, result; end
411 if not result then return nil, err; end
365 return function() 412 return function()
366 local row = result(); 413 local row = result();
367 if row ~= nil then 414 if row ~= nil then
368 local value, err = deserialize(row[2], row[3]); 415 local value, err = deserialize(row[2], row[3]);
369 assert(value ~= nil, err); 416 assert(value ~= nil, err);
370 return row[1], value, row[4], row[5]; 417 return row[1], value, row[4], row[5];
371 end 418 end
372 end, total; 419 end, total;
420 end
421
422 function archive_store:summary(username, query)
423 query = query or {};
424 local user,store = username,self.store;
425 local ok, result = engine:transaction(function()
426 local sql_query = [[
427 SELECT DISTINCT "with", COUNT(*), MIN("when"), MAX("when")
428 FROM "prosodyarchive"
429 WHERE %s
430 GROUP BY "with"
431 ORDER BY "sort_id" %s%s;
432 ]];
433 local args = { host, user or "", store, };
434 local where = { "\"host\" = ?", "\"user\" = ?", "\"store\" = ?", };
435
436 archive_where(query, args, where);
437
438 archive_where_id_range(query, args, where);
439
440 if query.limit then
441 args[#args+1] = query.limit;
442 end
443
444 sql_query = sql_query:format(t_concat(where, " AND "), query.reverse
445 and "DESC" or "ASC", query.limit and " LIMIT ?" or "");
446 return engine:select(sql_query, unpack(args));
447 end);
448 if not ok then return ok, result end
449 local counts = {};
450 local earliest, latest = {}, {};
451 for row in result do
452 local with, count = row[1], row[2];
453 counts[with] = count;
454 earliest[with] = row[3];
455 latest[with] = row[4];
456 end
457 return {
458 counts = counts;
459 earliest = earliest;
460 latest = latest;
461 };
373 end 462 end
374 463
375 function archive_store:delete(username, query) 464 function archive_store:delete(username, query)
376 query = query or {}; 465 query = query or {};
377 local user,store = username,self.store; 466 local user,store = username,self.store;
382 if user == true then 471 if user == true then
383 table.remove(args, 2); 472 table.remove(args, 2);
384 table.remove(where, 2); 473 table.remove(where, 2);
385 end 474 end
386 archive_where(query, args, where); 475 archive_where(query, args, where);
387 archive_where_id_range(query, args, where); 476 local ok, err = archive_where_id_range(query, args, where);
477 if not ok then return ok, err; end
388 if query.truncate == nil then 478 if query.truncate == nil then
389 sql_query = sql_query:format(t_concat(where, " AND ")); 479 sql_query = sql_query:format(t_concat(where, " AND "));
390 else 480 else
391 args[#args+1] = query.truncate; 481 args[#args+1] = query.truncate;
392 local unlimited = "ALL"; 482 local unlimited = "ALL";
421 sql_query = string.format(sql_query, t_concat(where, " AND "), 511 sql_query = string.format(sql_query, t_concat(where, " AND "),
422 query.reverse and "ASC" or "DESC", unlimited); 512 query.reverse and "ASC" or "DESC", unlimited);
423 end 513 end
424 return engine:delete(sql_query, unpack(args)); 514 return engine:delete(sql_query, unpack(args));
425 end); 515 end);
516 local cache_key = jid_join(username, host, self.store);
517 archive_item_count_cache:set(cache_key, nil);
426 return ok and stmt:affected(), stmt; 518 return ok and stmt:affected(), stmt;
519 end
520
521 function archive_store:users()
522 local ok, result = engine:transaction(function()
523 local select_sql = [[
524 SELECT DISTINCT "user"
525 FROM "prosodyarchive"
526 WHERE "host"=? AND "store"=?;
527 ]];
528 return engine:select(select_sql, host, self.store);
529 end);
530 if not ok then error(result); end
531 return iterator(result);
427 end 532 end
428 533
429 local stores = { 534 local stores = {
430 keyval = keyval_store; 535 keyval = keyval_store;
431 map = map_store; 536 map = map_store;