From 2b4811c8430dcbec09c67b73e8f939b4047f1aec Mon Sep 17 00:00:00 2001 From: Matthew Wild Date: Sat, 26 Feb 2011 00:23:48 +0000 Subject: tools/migration/*: Numerous changes and restructuring, and the addition of a Makefile --- tools/migration/migrator/prosody_sql.lua | 174 +++++++++++++++++++++++++++++++ 1 file changed, 174 insertions(+) create mode 100644 tools/migration/migrator/prosody_sql.lua (limited to 'tools/migration/migrator/prosody_sql.lua') diff --git a/tools/migration/migrator/prosody_sql.lua b/tools/migration/migrator/prosody_sql.lua new file mode 100644 index 00000000..3a9172ff --- /dev/null +++ b/tools/migration/migrator/prosody_sql.lua @@ -0,0 +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 "migrator.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; -- cgit v1.2.3