mirror of
https://github.com/bjc/prosody.git
synced 2025-04-04 05:37:39 +03:00
tools/migration/*.lua: Convert to unix line endings
This commit is contained in:
Matthew Wild
parent
6bac77b06f
commit
ba01859e1c
5 changed files with 416 additions and 416 deletions
|
|
@ -1,26 +1,26 @@
|
|||
input {
|
||||
type = "prosody_sql";
|
||||
driver = "SQLite3";
|
||||
database = "out.sqlite";
|
||||
}
|
||||
output {
|
||||
type = "prosody_files";
|
||||
path = "out";
|
||||
}
|
||||
|
||||
--[[
|
||||
|
||||
input {
|
||||
path = "../../data";
|
||||
type = "prosody_files";
|
||||
driver = "SQLite3";
|
||||
database = "../../prosody.sqlite";
|
||||
}
|
||||
output {
|
||||
type = "prosody_sql";
|
||||
driver = "SQLite3";
|
||||
database = "out.sqlite";
|
||||
path = "out";
|
||||
}
|
||||
|
||||
]]
|
||||
input {
|
||||
type = "prosody_sql";
|
||||
driver = "SQLite3";
|
||||
database = "out.sqlite";
|
||||
}
|
||||
output {
|
||||
type = "prosody_files";
|
||||
path = "out";
|
||||
}
|
||||
|
||||
--[[
|
||||
|
||||
input {
|
||||
path = "../../data";
|
||||
type = "prosody_files";
|
||||
driver = "SQLite3";
|
||||
database = "../../prosody.sqlite";
|
||||
}
|
||||
output {
|
||||
type = "prosody_sql";
|
||||
driver = "SQLite3";
|
||||
database = "out.sqlite";
|
||||
path = "out";
|
||||
}
|
||||
|
||||
]]
|
||||
|
|
|
|||
|
|
@ -1,27 +1,27 @@
|
|||
|
||||
|
||||
|
||||
local function loadfilein(file, env) return loadin and loadin(env, io.open(file):read("*a")) or setfenv(loadfile(file), env); end
|
||||
config = {};
|
||||
local config_env = setmetatable({}, { __index = function(t, k) return function(tbl) config[k] = tbl; end; end });
|
||||
loadfilein("config.lua", config_env)();
|
||||
|
||||
package.path = "../../?.lua;"..package.path
|
||||
package.cpath = "../../?.dll;"..package.cpath
|
||||
|
||||
|
||||
assert(config.input, "no input specified")
|
||||
assert(config.output, "no output specified")
|
||||
local itype = assert(config.input.type, "no input.type specified");
|
||||
local otype = assert(config.output.type, "no output.type specified");
|
||||
local reader = require(itype).reader(config.input);
|
||||
local writer = require(otype).writer(config.output);
|
||||
|
||||
local json = require "util.json";
|
||||
|
||||
for x in reader do
|
||||
--print(json.encode(x))
|
||||
writer(x);
|
||||
end
|
||||
writer(nil); -- close
|
||||
|
||||
|
||||
|
||||
|
||||
local function loadfilein(file, env) return loadin and loadin(env, io.open(file):read("*a")) or setfenv(loadfile(file), env); end
|
||||
config = {};
|
||||
local config_env = setmetatable({}, { __index = function(t, k) return function(tbl) config[k] = tbl; end; end });
|
||||
loadfilein("config.lua", config_env)();
|
||||
|
||||
package.path = "../../?.lua;"..package.path
|
||||
package.cpath = "../../?.dll;"..package.cpath
|
||||
|
||||
|
||||
assert(config.input, "no input specified")
|
||||
assert(config.output, "no output specified")
|
||||
local itype = assert(config.input.type, "no input.type specified");
|
||||
local otype = assert(config.output.type, "no output.type specified");
|
||||
local reader = require(itype).reader(config.input);
|
||||
local writer = require(otype).writer(config.output);
|
||||
|
||||
local json = require "util.json";
|
||||
|
||||
for x in reader do
|
||||
--print(json.encode(x))
|
||||
writer(x);
|
||||
end
|
||||
writer(nil); -- close
|
||||
|
||||
|
|
|
|||
|
|
@ -1,56 +1,56 @@
|
|||
|
||||
|
||||
local print = print;
|
||||
local t_insert = table.insert;
|
||||
local t_sort = table.sort;
|
||||
|
||||
module "mtools"
|
||||
|
||||
function sorted(params)
|
||||
|
||||
local reader = params.reader; -- iterator to get items from
|
||||
local sorter = params.sorter; -- sorting function
|
||||
local filter = params.filter; -- filter function
|
||||
|
||||
local cache = {};
|
||||
for item in reader do
|
||||
if filter then item = filter(item); end
|
||||
if item then t_insert(cache, item); end
|
||||
end
|
||||
if sorter then
|
||||
t_sort(cache, sorter);
|
||||
end
|
||||
local i = 0;
|
||||
return function()
|
||||
i = i + 1;
|
||||
return cache[i];
|
||||
end;
|
||||
|
||||
end
|
||||
|
||||
function merged(reader, merger)
|
||||
|
||||
local item1 = reader();
|
||||
local merged = { item1 };
|
||||
return function()
|
||||
while true do
|
||||
if not item1 then return nil; end
|
||||
local item2 = reader();
|
||||
if not item2 then item1 = nil; return merged; end
|
||||
if merger(item1, item2) then
|
||||
--print("merged")
|
||||
item1 = item2;
|
||||
t_insert(merged, item1);
|
||||
else
|
||||
--print("unmerged", merged)
|
||||
item1 = item2;
|
||||
local tmp = merged;
|
||||
merged = { item1 };
|
||||
return tmp;
|
||||
end
|
||||
end
|
||||
end;
|
||||
|
||||
end
|
||||
|
||||
return _M;
|
||||
|
||||
|
||||
local print = print;
|
||||
local t_insert = table.insert;
|
||||
local t_sort = table.sort;
|
||||
|
||||
module "mtools"
|
||||
|
||||
function sorted(params)
|
||||
|
||||
local reader = params.reader; -- iterator to get items from
|
||||
local sorter = params.sorter; -- sorting function
|
||||
local filter = params.filter; -- filter function
|
||||
|
||||
local cache = {};
|
||||
for item in reader do
|
||||
if filter then item = filter(item); end
|
||||
if item then t_insert(cache, item); end
|
||||
end
|
||||
if sorter then
|
||||
t_sort(cache, sorter);
|
||||
end
|
||||
local i = 0;
|
||||
return function()
|
||||
i = i + 1;
|
||||
return cache[i];
|
||||
end;
|
||||
|
||||
end
|
||||
|
||||
function merged(reader, merger)
|
||||
|
||||
local item1 = reader();
|
||||
local merged = { item1 };
|
||||
return function()
|
||||
while true do
|
||||
if not item1 then return nil; end
|
||||
local item2 = reader();
|
||||
if not item2 then item1 = nil; return merged; end
|
||||
if merger(item1, item2) then
|
||||
--print("merged")
|
||||
item1 = item2;
|
||||
t_insert(merged, item1);
|
||||
else
|
||||
--print("unmerged", merged)
|
||||
item1 = item2;
|
||||
local tmp = merged;
|
||||
merged = { item1 };
|
||||
return tmp;
|
||||
end
|
||||
end
|
||||
end;
|
||||
|
||||
end
|
||||
|
||||
return _M;
|
||||
|
|
|
|||
|
|
@ -1,133 +1,133 @@
|
|||
|
||||
local print = print;
|
||||
local assert = assert;
|
||||
local setmetatable = setmetatable;
|
||||
local tonumber = tonumber;
|
||||
local char = string.char;
|
||||
local coroutine = coroutine;
|
||||
local lfs = require "lfs";
|
||||
local loadfile = loadfile;
|
||||
local setfenv = setfenv;
|
||||
local pcall = pcall;
|
||||
local mtools = require "mtools";
|
||||
local next = next;
|
||||
local pairs = pairs;
|
||||
local json = require "util.json";
|
||||
|
||||
prosody = {};
|
||||
local dm = require "util.datamanager"
|
||||
|
||||
module "prosody_files"
|
||||
|
||||
local function is_dir(path) return lfs.attributes(path, "mode") == "directory"; end
|
||||
local function is_file(path) return lfs.attributes(path, "mode") == "file"; end
|
||||
local function clean_path(path)
|
||||
return path:gsub("\\", "/"):gsub("//+", "/");
|
||||
end
|
||||
local encode, decode; do
|
||||
local urlcodes = setmetatable({}, { __index = function (t, k) t[k] = char(tonumber("0x"..k)); return t[k]; end });
|
||||
decode = function (s) return s and (s:gsub("+", " "):gsub("%%([a-fA-F0-9][a-fA-F0-9])", urlcodes)); end
|
||||
encode = function (s) return s and (s:gsub("%W", function (c) return format("%%%02x", c:byte()); end)); end
|
||||
end
|
||||
local function decode_dir(x)
|
||||
if x:gsub("%%%x%x", ""):gsub("[a-zA-Z0-9]", "") == "" then
|
||||
return decode(x);
|
||||
end
|
||||
end
|
||||
local function decode_file(x)
|
||||
if x:match(".%.dat$") and x:gsub("%.dat$", ""):gsub("%%%x%x", ""):gsub("[a-zA-Z0-9]", "") == "" then
|
||||
return decode(x:gsub("%.dat$", ""));
|
||||
end
|
||||
end
|
||||
local function prosody_dir(path, ondir, onfile, ...)
|
||||
for x in lfs.dir(path) do
|
||||
local xpath = path.."/"..x;
|
||||
if decode_dir(x) and is_dir(xpath) then
|
||||
ondir(xpath, x, ...);
|
||||
elseif decode_file(x) and is_file(xpath) then
|
||||
onfile(xpath, x, ...);
|
||||
end
|
||||
end
|
||||
end
|
||||
|
||||
local function handle_root_file(path, name)
|
||||
--print("root file: ", decode_file(name))
|
||||
coroutine.yield { user = nil, host = nil, store = decode_file(name) };
|
||||
end
|
||||
local function handle_host_file(path, name, host)
|
||||
--print("host file: ", decode_dir(host).."/"..decode_file(name))
|
||||
coroutine.yield { user = nil, host = decode_dir(host), store = decode_file(name) };
|
||||
end
|
||||
local function handle_store_file(path, name, store, host)
|
||||
--print("store file: ", decode_file(name).."@"..decode_dir(host).."/"..decode_dir(store))
|
||||
coroutine.yield { user = decode_file(name), host = decode_dir(host), store = decode_dir(store) };
|
||||
end
|
||||
local function handle_host_store(path, name, host)
|
||||
prosody_dir(path, function() end, handle_store_file, name, host);
|
||||
end
|
||||
local function handle_host_dir(path, name)
|
||||
prosody_dir(path, handle_host_store, handle_host_file, name);
|
||||
end
|
||||
local function handle_root_dir(path)
|
||||
prosody_dir(path, handle_host_dir, handle_root_file);
|
||||
end
|
||||
|
||||
local function decode_user(item)
|
||||
local userdata = {
|
||||
user = item[1].user;
|
||||
host = item[1].host;
|
||||
stores = {};
|
||||
};
|
||||
for i=1,#item do -- loop over stores
|
||||
local result = {};
|
||||
local store = item[i];
|
||||
userdata.stores[store.store] = store.data;
|
||||
store.user = nil; store.host = nil; store.store = nil;
|
||||
end
|
||||
return userdata;
|
||||
end
|
||||
|
||||
function reader(input)
|
||||
local path = clean_path(assert(input.path, "no input.path specified"));
|
||||
assert(is_dir(path), "input.path is not a directory");
|
||||
local iter = coroutine.wrap(function()handle_root_dir(path);end);
|
||||
-- get per-user stores, sorted
|
||||
local iter = mtools.sorted {
|
||||
reader = function()
|
||||
local x = iter();
|
||||
if x then
|
||||
dm.set_data_path(path);
|
||||
x.data = assert(dm.load(x.user, x.host, x.store));
|
||||
return x;
|
||||
end
|
||||
end;
|
||||
sorter = function(a, b)
|
||||
local a_host, a_user, a_store = a.host or "", a.user or "", a.store or "";
|
||||
local b_host, b_user, b_store = b.host or "", b.user or "", b.store or "";
|
||||
return a_host > b_host or (a_host==b_host and a_user > b_user) or (a_host==b_host and a_user==b_user and a_store > b_store);
|
||||
end;
|
||||
};
|
||||
-- merge stores to get users
|
||||
iter = mtools.merged(iter, function(a, b)
|
||||
return (a.host == b.host and a.user == b.user);
|
||||
end);
|
||||
|
||||
return function()
|
||||
local x = iter();
|
||||
return x and decode_user(x);
|
||||
end
|
||||
end
|
||||
|
||||
function writer(output)
|
||||
local path = clean_path(assert(output.path, "no output.path specified"));
|
||||
assert(is_dir(path), "output.path is not a directory");
|
||||
return function(item)
|
||||
if not item then return; end -- end of input
|
||||
dm.set_data_path(path);
|
||||
for store, data in pairs(item.stores) do
|
||||
assert(dm.store(item.user, item.host, store, data));
|
||||
end
|
||||
end
|
||||
end
|
||||
|
||||
return _M;
|
||||
|
||||
local print = print;
|
||||
local assert = assert;
|
||||
local setmetatable = setmetatable;
|
||||
local tonumber = tonumber;
|
||||
local char = string.char;
|
||||
local coroutine = coroutine;
|
||||
local lfs = require "lfs";
|
||||
local loadfile = loadfile;
|
||||
local setfenv = setfenv;
|
||||
local pcall = pcall;
|
||||
local mtools = require "mtools";
|
||||
local next = next;
|
||||
local pairs = pairs;
|
||||
local json = require "util.json";
|
||||
|
||||
prosody = {};
|
||||
local dm = require "util.datamanager"
|
||||
|
||||
module "prosody_files"
|
||||
|
||||
local function is_dir(path) return lfs.attributes(path, "mode") == "directory"; end
|
||||
local function is_file(path) return lfs.attributes(path, "mode") == "file"; end
|
||||
local function clean_path(path)
|
||||
return path:gsub("\\", "/"):gsub("//+", "/");
|
||||
end
|
||||
local encode, decode; do
|
||||
local urlcodes = setmetatable({}, { __index = function (t, k) t[k] = char(tonumber("0x"..k)); return t[k]; end });
|
||||
decode = function (s) return s and (s:gsub("+", " "):gsub("%%([a-fA-F0-9][a-fA-F0-9])", urlcodes)); end
|
||||
encode = function (s) return s and (s:gsub("%W", function (c) return format("%%%02x", c:byte()); end)); end
|
||||
end
|
||||
local function decode_dir(x)
|
||||
if x:gsub("%%%x%x", ""):gsub("[a-zA-Z0-9]", "") == "" then
|
||||
return decode(x);
|
||||
end
|
||||
end
|
||||
local function decode_file(x)
|
||||
if x:match(".%.dat$") and x:gsub("%.dat$", ""):gsub("%%%x%x", ""):gsub("[a-zA-Z0-9]", "") == "" then
|
||||
return decode(x:gsub("%.dat$", ""));
|
||||
end
|
||||
end
|
||||
local function prosody_dir(path, ondir, onfile, ...)
|
||||
for x in lfs.dir(path) do
|
||||
local xpath = path.."/"..x;
|
||||
if decode_dir(x) and is_dir(xpath) then
|
||||
ondir(xpath, x, ...);
|
||||
elseif decode_file(x) and is_file(xpath) then
|
||||
onfile(xpath, x, ...);
|
||||
end
|
||||
end
|
||||
end
|
||||
|
||||
local function handle_root_file(path, name)
|
||||
--print("root file: ", decode_file(name))
|
||||
coroutine.yield { user = nil, host = nil, store = decode_file(name) };
|
||||
end
|
||||
local function handle_host_file(path, name, host)
|
||||
--print("host file: ", decode_dir(host).."/"..decode_file(name))
|
||||
coroutine.yield { user = nil, host = decode_dir(host), store = decode_file(name) };
|
||||
end
|
||||
local function handle_store_file(path, name, store, host)
|
||||
--print("store file: ", decode_file(name).."@"..decode_dir(host).."/"..decode_dir(store))
|
||||
coroutine.yield { user = decode_file(name), host = decode_dir(host), store = decode_dir(store) };
|
||||
end
|
||||
local function handle_host_store(path, name, host)
|
||||
prosody_dir(path, function() end, handle_store_file, name, host);
|
||||
end
|
||||
local function handle_host_dir(path, name)
|
||||
prosody_dir(path, handle_host_store, handle_host_file, name);
|
||||
end
|
||||
local function handle_root_dir(path)
|
||||
prosody_dir(path, handle_host_dir, handle_root_file);
|
||||
end
|
||||
|
||||
local function decode_user(item)
|
||||
local userdata = {
|
||||
user = item[1].user;
|
||||
host = item[1].host;
|
||||
stores = {};
|
||||
};
|
||||
for i=1,#item do -- loop over stores
|
||||
local result = {};
|
||||
local store = item[i];
|
||||
userdata.stores[store.store] = store.data;
|
||||
store.user = nil; store.host = nil; store.store = nil;
|
||||
end
|
||||
return userdata;
|
||||
end
|
||||
|
||||
function reader(input)
|
||||
local path = clean_path(assert(input.path, "no input.path specified"));
|
||||
assert(is_dir(path), "input.path is not a directory");
|
||||
local iter = coroutine.wrap(function()handle_root_dir(path);end);
|
||||
-- get per-user stores, sorted
|
||||
local iter = mtools.sorted {
|
||||
reader = function()
|
||||
local x = iter();
|
||||
if x then
|
||||
dm.set_data_path(path);
|
||||
x.data = assert(dm.load(x.user, x.host, x.store));
|
||||
return x;
|
||||
end
|
||||
end;
|
||||
sorter = function(a, b)
|
||||
local a_host, a_user, a_store = a.host or "", a.user or "", a.store or "";
|
||||
local b_host, b_user, b_store = b.host or "", b.user or "", b.store or "";
|
||||
return a_host > b_host or (a_host==b_host and a_user > b_user) or (a_host==b_host and a_user==b_user and a_store > b_store);
|
||||
end;
|
||||
};
|
||||
-- merge stores to get users
|
||||
iter = mtools.merged(iter, function(a, b)
|
||||
return (a.host == b.host and a.user == b.user);
|
||||
end);
|
||||
|
||||
return function()
|
||||
local x = iter();
|
||||
return x and decode_user(x);
|
||||
end
|
||||
end
|
||||
|
||||
function writer(output)
|
||||
local path = clean_path(assert(output.path, "no output.path specified"));
|
||||
assert(is_dir(path), "output.path is not a directory");
|
||||
return function(item)
|
||||
if not item then return; end -- end of input
|
||||
dm.set_data_path(path);
|
||||
for store, data in pairs(item.stores) do
|
||||
assert(dm.store(item.user, item.host, store, data));
|
||||
end
|
||||
end
|
||||
end
|
||||
|
||||
return _M;
|
||||
|
|
|
|||
|
|
@ -1,174 +1,174 @@
|
|||
|
||||
local assert = assert;
|
||||
local DBI = require "DBI";
|
||||
local print = print;
|
||||
local type = type;
|
||||
local next = next;
|
||||
local pairs = pairs;
|
||||
local t_sort = table.sort;
|
||||
local json = require "util.json";
|
||||
local mtools = require "mtools";
|
||||
local tostring = tostring;
|
||||
local tonumber = tonumber;
|
||||
|
||||
module "prosody_sql"
|
||||
|
||||
local function create_table(connection, params)
|
||||
local create_sql = "CREATE TABLE `prosody` (`host` TEXT, `user` TEXT, `store` TEXT, `key` TEXT, `type` TEXT, `value` TEXT);";
|
||||
if params.driver == "PostgreSQL" then
|
||||
create_sql = create_sql:gsub("`", "\"");
|
||||
end
|
||||
|
||||
local stmt = connection:prepare(create_sql);
|
||||
if stmt then
|
||||
local ok = stmt:execute();
|
||||
local commit_ok = connection:commit();
|
||||
if ok and commit_ok then
|
||||
local index_sql = "CREATE INDEX `prosody_index` ON `prosody` (`host`, `user`, `store`, `key`)";
|
||||
if params.driver == "PostgreSQL" then
|
||||
index_sql = index_sql:gsub("`", "\"");
|
||||
elseif params.driver == "MySQL" then
|
||||
index_sql = index_sql:gsub("`([,)])", "`(20)%1");
|
||||
end
|
||||
local stmt, err = connection:prepare(index_sql);
|
||||
local ok, commit_ok, commit_err;
|
||||
if stmt then
|
||||
ok, err = assert(stmt:execute());
|
||||
commit_ok, commit_err = assert(connection:commit());
|
||||
end
|
||||
end
|
||||
end
|
||||
end
|
||||
|
||||
local function serialize(value)
|
||||
local t = type(value);
|
||||
if t == "string" or t == "boolean" or t == "number" then
|
||||
return t, tostring(value);
|
||||
elseif t == "table" then
|
||||
local value,err = json.encode(value);
|
||||
if value then return "json", value; end
|
||||
return nil, err;
|
||||
end
|
||||
return nil, "Unhandled value type: "..t;
|
||||
end
|
||||
local function deserialize(t, value)
|
||||
if t == "string" then return value;
|
||||
elseif t == "boolean" then
|
||||
if value == "true" then return true;
|
||||
elseif value == "false" then return false; end
|
||||
elseif t == "number" then return tonumber(value);
|
||||
elseif t == "json" then
|
||||
return json.decode(value);
|
||||
end
|
||||
end
|
||||
|
||||
local function decode_user(item)
|
||||
local userdata = {
|
||||
user = item[1][1].user;
|
||||
host = item[1][1].host;
|
||||
stores = {};
|
||||
};
|
||||
for i=1,#item do -- loop over stores
|
||||
local result = {};
|
||||
local store = item[i];
|
||||
for i=1,#store do -- loop over store data
|
||||
local row = store[i];
|
||||
local k = row.key;
|
||||
local v = deserialize(row.type, row.value);
|
||||
if k and v then
|
||||
if k ~= "" then result[k] = v; elseif type(v) == "table" then
|
||||
for a,b in pairs(v) do
|
||||
result[a] = b;
|
||||
end
|
||||
end
|
||||
end
|
||||
userdata.stores[store[1].store] = result;
|
||||
end
|
||||
end
|
||||
return userdata;
|
||||
end
|
||||
|
||||
function reader(input)
|
||||
local dbh = assert(DBI.Connect(
|
||||
assert(input.driver, "no input.driver specified"),
|
||||
assert(input.database, "no input.database specified"),
|
||||
input.username, input.password,
|
||||
input.host, input.port
|
||||
));
|
||||
assert(dbh:ping());
|
||||
local stmt = assert(dbh:prepare("SELECT * FROM prosody"));
|
||||
assert(stmt:execute());
|
||||
local keys = {"host", "user", "store", "key", "type", "value"};
|
||||
local f,s,val = stmt:rows(true);
|
||||
-- get SQL rows, sorted
|
||||
local iter = mtools.sorted {
|
||||
reader = function() val = f(s, val); return val; end;
|
||||
filter = function(x)
|
||||
for i=1,#keys do
|
||||
if not x[keys[i]] then return false; end -- TODO log error, missing field
|
||||
end
|
||||
if x.host == "" then x.host = nil; end
|
||||
if x.user == "" then x.user = nil; end
|
||||
if x.store == "" then x.store = nil; end
|
||||
return x;
|
||||
end;
|
||||
sorter = function(a, b)
|
||||
local a_host, a_user, a_store = a.host or "", a.user or "", a.store or "";
|
||||
local b_host, b_user, b_store = b.host or "", b.user or "", b.store or "";
|
||||
return a_host > b_host or (a_host==b_host and a_user > b_user) or (a_host==b_host and a_user==b_user and a_store > b_store);
|
||||
end;
|
||||
};
|
||||
-- merge rows to get stores
|
||||
iter = mtools.merged(iter, function(a, b)
|
||||
return (a.host == b.host and a.user == b.user and a.store == b.store);
|
||||
end);
|
||||
-- merge stores to get users
|
||||
iter = mtools.merged(iter, function(a, b)
|
||||
return (a[1].host == b[1].host and a[1].user == b[1].user);
|
||||
end);
|
||||
return function()
|
||||
local x = iter();
|
||||
return x and decode_user(x);
|
||||
end;
|
||||
end
|
||||
|
||||
function writer(output, iter)
|
||||
local dbh = assert(DBI.Connect(
|
||||
assert(output.driver, "no output.driver specified"),
|
||||
assert(output.database, "no output.database specified"),
|
||||
output.username, output.password,
|
||||
output.host, output.port
|
||||
));
|
||||
assert(dbh:ping());
|
||||
create_table(dbh, output);
|
||||
local stmt = assert(dbh:prepare("SELECT * FROM prosody"));
|
||||
assert(stmt:execute());
|
||||
local stmt = assert(dbh:prepare("DELETE FROM prosody"));
|
||||
assert(stmt:execute());
|
||||
local insert = assert(dbh:prepare("INSERT INTO `prosody` (`host`,`user`,`store`,`key`,`type`,`value`) VALUES (?,?,?,?,?,?)"));
|
||||
|
||||
return function(item)
|
||||
if not item then assert(dbh:commit()) return dbh:close(); end -- end of input
|
||||
local host = item.host or "";
|
||||
local user = item.user or "";
|
||||
for store, data in pairs(item.stores) do
|
||||
-- TODO transactions
|
||||
local extradata = {};
|
||||
for key, value in pairs(data) do
|
||||
if type(key) == "string" and key ~= "" then
|
||||
local t, value = assert(serialize(value));
|
||||
local ok, err = assert(insert:execute(host, user, store, key, t, value));
|
||||
else
|
||||
extradata[key] = value;
|
||||
end
|
||||
end
|
||||
if next(extradata) ~= nil then
|
||||
local t, extradata = assert(serialize(extradata));
|
||||
local ok, err = assert(insert:execute(host, user, store, "", t, extradata));
|
||||
end
|
||||
end
|
||||
end;
|
||||
end
|
||||
|
||||
|
||||
return _M;
|
||||
|
||||
local assert = assert;
|
||||
local DBI = require "DBI";
|
||||
local print = print;
|
||||
local type = type;
|
||||
local next = next;
|
||||
local pairs = pairs;
|
||||
local t_sort = table.sort;
|
||||
local json = require "util.json";
|
||||
local mtools = require "mtools";
|
||||
local tostring = tostring;
|
||||
local tonumber = tonumber;
|
||||
|
||||
module "prosody_sql"
|
||||
|
||||
local function create_table(connection, params)
|
||||
local create_sql = "CREATE TABLE `prosody` (`host` TEXT, `user` TEXT, `store` TEXT, `key` TEXT, `type` TEXT, `value` TEXT);";
|
||||
if params.driver == "PostgreSQL" then
|
||||
create_sql = create_sql:gsub("`", "\"");
|
||||
end
|
||||
|
||||
local stmt = connection:prepare(create_sql);
|
||||
if stmt then
|
||||
local ok = stmt:execute();
|
||||
local commit_ok = connection:commit();
|
||||
if ok and commit_ok then
|
||||
local index_sql = "CREATE INDEX `prosody_index` ON `prosody` (`host`, `user`, `store`, `key`)";
|
||||
if params.driver == "PostgreSQL" then
|
||||
index_sql = index_sql:gsub("`", "\"");
|
||||
elseif params.driver == "MySQL" then
|
||||
index_sql = index_sql:gsub("`([,)])", "`(20)%1");
|
||||
end
|
||||
local stmt, err = connection:prepare(index_sql);
|
||||
local ok, commit_ok, commit_err;
|
||||
if stmt then
|
||||
ok, err = assert(stmt:execute());
|
||||
commit_ok, commit_err = assert(connection:commit());
|
||||
end
|
||||
end
|
||||
end
|
||||
end
|
||||
|
||||
local function serialize(value)
|
||||
local t = type(value);
|
||||
if t == "string" or t == "boolean" or t == "number" then
|
||||
return t, tostring(value);
|
||||
elseif t == "table" then
|
||||
local value,err = json.encode(value);
|
||||
if value then return "json", value; end
|
||||
return nil, err;
|
||||
end
|
||||
return nil, "Unhandled value type: "..t;
|
||||
end
|
||||
local function deserialize(t, value)
|
||||
if t == "string" then return value;
|
||||
elseif t == "boolean" then
|
||||
if value == "true" then return true;
|
||||
elseif value == "false" then return false; end
|
||||
elseif t == "number" then return tonumber(value);
|
||||
elseif t == "json" then
|
||||
return json.decode(value);
|
||||
end
|
||||
end
|
||||
|
||||
local function decode_user(item)
|
||||
local userdata = {
|
||||
user = item[1][1].user;
|
||||
host = item[1][1].host;
|
||||
stores = {};
|
||||
};
|
||||
for i=1,#item do -- loop over stores
|
||||
local result = {};
|
||||
local store = item[i];
|
||||
for i=1,#store do -- loop over store data
|
||||
local row = store[i];
|
||||
local k = row.key;
|
||||
local v = deserialize(row.type, row.value);
|
||||
if k and v then
|
||||
if k ~= "" then result[k] = v; elseif type(v) == "table" then
|
||||
for a,b in pairs(v) do
|
||||
result[a] = b;
|
||||
end
|
||||
end
|
||||
end
|
||||
userdata.stores[store[1].store] = result;
|
||||
end
|
||||
end
|
||||
return userdata;
|
||||
end
|
||||
|
||||
function reader(input)
|
||||
local dbh = assert(DBI.Connect(
|
||||
assert(input.driver, "no input.driver specified"),
|
||||
assert(input.database, "no input.database specified"),
|
||||
input.username, input.password,
|
||||
input.host, input.port
|
||||
));
|
||||
assert(dbh:ping());
|
||||
local stmt = assert(dbh:prepare("SELECT * FROM prosody"));
|
||||
assert(stmt:execute());
|
||||
local keys = {"host", "user", "store", "key", "type", "value"};
|
||||
local f,s,val = stmt:rows(true);
|
||||
-- get SQL rows, sorted
|
||||
local iter = mtools.sorted {
|
||||
reader = function() val = f(s, val); return val; end;
|
||||
filter = function(x)
|
||||
for i=1,#keys do
|
||||
if not x[keys[i]] then return false; end -- TODO log error, missing field
|
||||
end
|
||||
if x.host == "" then x.host = nil; end
|
||||
if x.user == "" then x.user = nil; end
|
||||
if x.store == "" then x.store = nil; end
|
||||
return x;
|
||||
end;
|
||||
sorter = function(a, b)
|
||||
local a_host, a_user, a_store = a.host or "", a.user or "", a.store or "";
|
||||
local b_host, b_user, b_store = b.host or "", b.user or "", b.store or "";
|
||||
return a_host > b_host or (a_host==b_host and a_user > b_user) or (a_host==b_host and a_user==b_user and a_store > b_store);
|
||||
end;
|
||||
};
|
||||
-- merge rows to get stores
|
||||
iter = mtools.merged(iter, function(a, b)
|
||||
return (a.host == b.host and a.user == b.user and a.store == b.store);
|
||||
end);
|
||||
-- merge stores to get users
|
||||
iter = mtools.merged(iter, function(a, b)
|
||||
return (a[1].host == b[1].host and a[1].user == b[1].user);
|
||||
end);
|
||||
return function()
|
||||
local x = iter();
|
||||
return x and decode_user(x);
|
||||
end;
|
||||
end
|
||||
|
||||
function writer(output, iter)
|
||||
local dbh = assert(DBI.Connect(
|
||||
assert(output.driver, "no output.driver specified"),
|
||||
assert(output.database, "no output.database specified"),
|
||||
output.username, output.password,
|
||||
output.host, output.port
|
||||
));
|
||||
assert(dbh:ping());
|
||||
create_table(dbh, output);
|
||||
local stmt = assert(dbh:prepare("SELECT * FROM prosody"));
|
||||
assert(stmt:execute());
|
||||
local stmt = assert(dbh:prepare("DELETE FROM prosody"));
|
||||
assert(stmt:execute());
|
||||
local insert = assert(dbh:prepare("INSERT INTO `prosody` (`host`,`user`,`store`,`key`,`type`,`value`) VALUES (?,?,?,?,?,?)"));
|
||||
|
||||
return function(item)
|
||||
if not item then assert(dbh:commit()) return dbh:close(); end -- end of input
|
||||
local host = item.host or "";
|
||||
local user = item.user or "";
|
||||
for store, data in pairs(item.stores) do
|
||||
-- TODO transactions
|
||||
local extradata = {};
|
||||
for key, value in pairs(data) do
|
||||
if type(key) == "string" and key ~= "" then
|
||||
local t, value = assert(serialize(value));
|
||||
local ok, err = assert(insert:execute(host, user, store, key, t, value));
|
||||
else
|
||||
extradata[key] = value;
|
||||
end
|
||||
end
|
||||
if next(extradata) ~= nil then
|
||||
local t, extradata = assert(serialize(extradata));
|
||||
local ok, err = assert(insert:execute(host, user, store, "", t, extradata));
|
||||
end
|
||||
end
|
||||
end;
|
||||
end
|
||||
|
||||
|
||||
return _M;
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue