Mercurial > prosody-hg

comparison plugins/mod_storage_sql.lua @ 11200:bf8f2da84007

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 Thu, 05 Nov 2020 22:31:25 +0100
parents b5e7f4d533e2
children 41a962b72a6e
comparison
equal deleted inserted replaced
11199:6c7c50a4de32 11200:bf8f2da84007
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 local item_count_cache_hit = module:measure("item_count_cache_hit", "rate");
157 local item_count_cache_miss = module:measure("item_count_cache_miss", "rate")
158
159 -- luacheck: ignore 512 431/user 431/store 431/err
152 local map_store = {}; 160 local map_store = {};
153 map_store.__index = map_store; 161 map_store.__index = map_store;
154 map_store.remove = {}; 162 map_store.remove = {};
155 function map_store:get(username, key) 163 function map_store:get(username, key)
156 local ok, result = engine:transaction(function() 164 local ok, result = engine:transaction(function()
223 end); 231 end);
224 if not ok then return nil, result; end 232 if not ok then return nil, result; end
225 return result; 233 return result;
226 end 234 end
227 235
236 function map_store:get_all(key)
237 if type(key) ~= "string" or key == "" then
238 return nil, "get_all only supports non-empty string keys";
239 end
240 local ok, result = engine:transaction(function()
241 local query = [[
242 SELECT "user", "type", "value"
243 FROM "prosody"
244 WHERE "host"=? AND "store"=? AND "key"=?
245 ]];
246
247 local data;
248 for row in engine:select(query, host, self.store, key) do
249 local key_data, err = deserialize(row[2], row[3]);
250 assert(key_data ~= nil, err);
251 if data == nil then
252 data = {};
253 end
254 data[row[1]] = key_data;
255 end
256
257 return data;
258
259 end);
260 if not ok then return nil, result; end
261 return result;
262 end
263
264 function map_store:delete_all(key)
265 if type(key) ~= "string" or key == "" then
266 return nil, "delete_all only supports non-empty string keys";
267 end
268 local ok, result = engine:transaction(function()
269 local delete_sql = [[
270 DELETE FROM "prosody"
271 WHERE "host"=? AND "store"=? AND "key"=?;
272 ]];
273 engine:delete(delete_sql, host, self.store, key);
274 return true;
275 end);
276 if not ok then return nil, result; end
277 return result;
278 end
279
228 local archive_store = {} 280 local archive_store = {}
229 archive_store.caps = { 281 archive_store.caps = {
230 total = true; 282 total = true;
283 quota = archive_item_limit;
284 truncate = true;
231 }; 285 };
232 archive_store.__index = archive_store 286 archive_store.__index = archive_store
233 function archive_store:append(username, key, value, when, with) 287 function archive_store:append(username, key, value, when, with)
234 local user,store = username,self.store; 288 local user,store = username,self.store;
289 local cache_key = jid_join(username, host, store);
290 local item_count = archive_item_count_cache:get(cache_key);
291 if not item_count then
292 item_count_cache_miss();
293 local ok, ret = engine:transaction(function()
294 local count_sql = [[
295 SELECT COUNT(*) FROM "prosodyarchive"
296 WHERE "host"=? AND "user"=? AND "store"=?;
297 ]];
298 local result = engine:select(count_sql, host, user, store);
299 if result then
300 for row in result do
301 item_count = row[1];
302 end
303 end
304 end);
305 if not ok or not item_count then
306 module:log("error", "Failed while checking quota for %s: %s", username, ret);
307 return nil, "Failure while checking quota";
308 end
309 archive_item_count_cache:set(cache_key, item_count);
310 else
311 item_count_cache_hit();
312 end
313
314 if archive_item_limit then
315 module:log("debug", "%s has %d items out of %d limit", username, item_count, archive_item_limit);
316 if item_count >= archive_item_limit then
317 return nil, "quota-limit";
318 end
319 end
320
235 when = when or os.time(); 321 when = when or os.time();
236 with = with or ""; 322 with = with or "";
237 local ok, ret = engine:transaction(function() 323 local ok, ret = engine:transaction(function()
238 local delete_sql = [[ 324 local delete_sql = [[
239 DELETE FROM "prosodyarchive" 325 DELETE FROM "prosodyarchive"
243 INSERT INTO "prosodyarchive" 329 INSERT INTO "prosodyarchive"
244 ("host", "user", "store", "when", "with", "key", "type", "value") 330 ("host", "user", "store", "when", "with", "key", "type", "value")
245 VALUES (?,?,?,?,?,?,?,?); 331 VALUES (?,?,?,?,?,?,?,?);
246 ]]; 332 ]];
247 if key then 333 if key then
248 engine:delete(delete_sql, host, user or "", store, key); 334 local result = engine:delete(delete_sql, host, user or "", store, key);
335 if result then
336 item_count = item_count - result:affected();
337 end
249 else 338 else
250 key = uuid.generate(); 339 key = uuid.generate();
251 end 340 end
252 local t, encoded_value = assert(serialize(value)); 341 local t, encoded_value = assert(serialize(value));
253 engine:insert(insert_sql, host, user or "", store, when, with, key, t, encoded_value); 342 engine:insert(insert_sql, host, user or "", store, when, with, key, t, encoded_value);
343 archive_item_count_cache:set(cache_key, item_count+1);
254 return key; 344 return key;
255 end); 345 end);
256 if not ok then return ok, ret; end 346 if not ok then return ok, ret; end
257 return ret; -- the key 347 return ret; -- the key
258 end 348 end
285 where[#where+1] = "\"key\" = ?"; 375 where[#where+1] = "\"key\" = ?";
286 args[#args+1] = query.key 376 args[#args+1] = query.key
287 end 377 end
288 end 378 end
289 local function archive_where_id_range(query, args, where) 379 local function archive_where_id_range(query, args, where)
290 local args_len = #args
291 -- Before or after specific item, exclusive 380 -- Before or after specific item, exclusive
381 local id_lookup_sql = [[
382 SELECT "sort_id"
383 FROM "prosodyarchive"
384 WHERE "key" = ? AND "host" = ? AND "user" = ? AND "store" = ?
385 LIMIT 1;
386 ]];
292 if query.after then -- keys better be unique! 387 if query.after then -- keys better be unique!
293 where[#where+1] = [[ 388 local after_id = nil;
294 "sort_id" > COALESCE( 389 for row in engine:select(id_lookup_sql, query.after, args[1], args[2], args[3]) do
295 ( 390 after_id = row[1];
296 SELECT "sort_id" 391 end
297 FROM "prosodyarchive" 392 if not after_id then
298 WHERE "key" = ? AND "host" = ? AND "user" = ? AND "store" = ? 393 return nil, "item-not-found";
299 LIMIT 1 394 end
300 ), 0) 395 where[#where+1] = '"sort_id" > ?';
301 ]]; 396 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 397 end
305 if query.before then 398 if query.before then
306 where[#where+1] = [[ 399 local before_id = nil;
307 "sort_id" < COALESCE( 400 for row in engine:select(id_lookup_sql, query.before, args[1], args[2], args[3]) do
308 ( 401 before_id = row[1];
309 SELECT "sort_id" 402 end
310 FROM "prosodyarchive" 403 if not before_id then
311 WHERE "key" = ? AND "host" = ? AND "user" = ? AND "store" = ? 404 return nil, "item-not-found";
312 LIMIT 1 405 end
313 ), 406 where[#where+1] = '"sort_id" < ?';
314 ( 407 args[#args+1] = before_id;
315 SELECT MAX("sort_id")+1 408 end
316 FROM "prosodyarchive" 409 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 410 end
323 411
324 function archive_store:find(username, query) 412 function archive_store:find(username, query)
325 query = query or {}; 413 query = query or {};
326 local user,store = username,self.store; 414 local user,store = username,self.store;
327 local total; 415 local cache_key = jid_join(username, host, self.store);
328 local ok, result = engine:transaction(function() 416 local total = archive_item_count_cache:get(cache_key);
417 (total and item_count_cache_hit or item_count_cache_miss)();
418 if total ~= nil and query.limit == 0 and query.start == nil and query.with == nil and query["end"] == nil and query.key == nil then
419 return noop, total;
420 end
421 local ok, result, err = engine:transaction(function()
329 local sql_query = [[ 422 local sql_query = [[
330 SELECT "key", "type", "value", "when", "with" 423 SELECT "key", "type", "value", "when", "with"
331 FROM "prosodyarchive" 424 FROM "prosodyarchive"
332 WHERE %s 425 WHERE %s
333 ORDER BY "sort_id" %s%s; 426 ORDER BY "sort_id" %s%s;
344 if stats then 437 if stats then
345 for row in stats do 438 for row in stats do
346 total = row[1]; 439 total = row[1];
347 end 440 end
348 end 441 end
442 if query.start == nil and query.with == nil and query["end"] == nil and query.key == nil then
443 archive_item_count_cache:set(cache_key, total);
444 end
349 if query.limit == 0 then -- Skip the real query 445 if query.limit == 0 then -- Skip the real query
350 return noop, total; 446 return noop, total;
351 end 447 end
352 end 448 end
353 449
354 archive_where_id_range(query, args, where); 450 local ok, err = archive_where_id_range(query, args, where);
451 if not ok then return ok, err; end
355 452
356 if query.limit then 453 if query.limit then
357 args[#args+1] = query.limit; 454 args[#args+1] = query.limit;
358 end 455 end
359 456
360 sql_query = sql_query:format(t_concat(where, " AND "), query.reverse 457 sql_query = sql_query:format(t_concat(where, " AND "), query.reverse
361 and "DESC" or "ASC", query.limit and " LIMIT ?" or ""); 458 and "DESC" or "ASC", query.limit and " LIMIT ?" or "");
362 return engine:select(sql_query, unpack(args)); 459 return engine:select(sql_query, unpack(args));
363 end); 460 end);
364 if not ok then return ok, result end 461 if not ok then return ok, result; end
462 if not result then return nil, err; end
365 return function() 463 return function()
366 local row = result(); 464 local row = result();
367 if row ~= nil then 465 if row ~= nil then
368 local value, err = deserialize(row[2], row[3]); 466 local value, err = deserialize(row[2], row[3]);
369 assert(value ~= nil, err); 467 assert(value ~= nil, err);
370 return row[1], value, row[4], row[5]; 468 return row[1], value, row[4], row[5];
371 end 469 end
372 end, total; 470 end, total;
471 end
472
473 function archive_store:summary(username, query)
474 query = query or {};
475 local user,store = username,self.store;
476 local ok, result = engine:transaction(function()
477 local sql_query = [[
478 SELECT DISTINCT "with", COUNT(*), MIN("when"), MAX("when")
479 FROM "prosodyarchive"
480 WHERE %s
481 GROUP BY "with"
482 ORDER BY "sort_id" %s%s;
483 ]];
484 local args = { host, user or "", store, };
485 local where = { "\"host\" = ?", "\"user\" = ?", "\"store\" = ?", };
486
487 archive_where(query, args, where);
488
489 archive_where_id_range(query, args, where);
490
491 if query.limit then
492 args[#args+1] = query.limit;
493 end
494
495 sql_query = sql_query:format(t_concat(where, " AND "), query.reverse
496 and "DESC" or "ASC", query.limit and " LIMIT ?" or "");
497 return engine:select(sql_query, unpack(args));
498 end);
499 if not ok then return ok, result end
500 local counts = {};
501 local earliest, latest = {}, {};
502 for row in result do
503 local with, count = row[1], row[2];
504 counts[with] = count;
505 earliest[with] = row[3];
506 latest[with] = row[4];
507 end
508 return {
509 counts = counts;
510 earliest = earliest;
511 latest = latest;
512 };
373 end 513 end
374 514
375 function archive_store:delete(username, query) 515 function archive_store:delete(username, query)
376 query = query or {}; 516 query = query or {};
377 local user,store = username,self.store; 517 local user,store = username,self.store;
382 if user == true then 522 if user == true then
383 table.remove(args, 2); 523 table.remove(args, 2);
384 table.remove(where, 2); 524 table.remove(where, 2);
385 end 525 end
386 archive_where(query, args, where); 526 archive_where(query, args, where);
387 archive_where_id_range(query, args, where); 527 local ok, err = archive_where_id_range(query, args, where);
528 if not ok then return ok, err; end
388 if query.truncate == nil then 529 if query.truncate == nil then
389 sql_query = sql_query:format(t_concat(where, " AND ")); 530 sql_query = sql_query:format(t_concat(where, " AND "));
390 else 531 else
391 args[#args+1] = query.truncate; 532 args[#args+1] = query.truncate;
392 local unlimited = "ALL"; 533 local unlimited = "ALL";
421 sql_query = string.format(sql_query, t_concat(where, " AND "), 562 sql_query = string.format(sql_query, t_concat(where, " AND "),
422 query.reverse and "ASC" or "DESC", unlimited); 563 query.reverse and "ASC" or "DESC", unlimited);
423 end 564 end
424 return engine:delete(sql_query, unpack(args)); 565 return engine:delete(sql_query, unpack(args));
425 end); 566 end);
567 local cache_key = jid_join(username, host, self.store);
568 archive_item_count_cache:set(cache_key, nil);
426 return ok and stmt:affected(), stmt; 569 return ok and stmt:affected(), stmt;
570 end
571
572 function archive_store:users()
573 local ok, result = engine:transaction(function()
574 local select_sql = [[
575 SELECT DISTINCT "user"
576 FROM "prosodyarchive"
577 WHERE "host"=? AND "store"=?;
578 ]];
579 return engine:select(select_sql, host, self.store);
580 end);
581 if not ok then error(result); end
582 return iterator(result);
427 end 583 end
428 584
429 local stores = { 585 local stores = {
430 keyval = keyval_store; 586 keyval = keyval_store;
431 map = map_store; 587 map = map_store;
608 764
609 function module.load() 765 function module.load()
610 if prosody.prosodyctl then return; end 766 if prosody.prosodyctl then return; end
611 local engines = module:shared("/*/sql/connections"); 767 local engines = module:shared("/*/sql/connections");
612 local params = normalize_params(module:get_option("sql", default_params)); 768 local params = normalize_params(module:get_option("sql", default_params));
613 engine = engines[sql.db2uri(params)]; 769 local db_uri = sql.db2uri(params);
770 engine = engines[db_uri];
614 if not engine then 771 if not engine then
615 module:log("debug", "Creating new engine"); 772 module:log("debug", "Creating new engine %s", db_uri);
616 engine = sql:create_engine(params, function (engine) -- luacheck: ignore 431/engine 773 engine = sql:create_engine(params, function (engine) -- luacheck: ignore 431/engine
617 if module:get_option("sql_manage_tables", true) then 774 if module:get_option("sql_manage_tables", true) then
618 -- Automatically create table, ignore failure (table probably already exists) 775 -- Automatically create table, ignore failure (table probably already exists)
619 -- FIXME: we should check in information_schema, etc. 776 -- FIXME: we should check in information_schema, etc.
620 create_table(engine); 777 create_table(engine);