tools/migration/*: Numerous changes and restructuring, and the addition of a Makefile
authorMatthew Wild <mwild1@gmail.com>
Sat, 26 Feb 2011 00:23:48 +0000 (00:23 +0000)
committerMatthew Wild <mwild1@gmail.com>
Sat, 26 Feb 2011 00:23:48 +0000 (00:23 +0000)
tools/migration/Makefile [new file with mode: 0644]
tools/migration/main.lua
tools/migration/migrator/mtools.lua [new file with mode: 0644]
tools/migration/migrator/prosody_files.lua [new file with mode: 0644]
tools/migration/migrator/prosody_sql.lua [new file with mode: 0644]
tools/migration/mtools.lua [deleted file]
tools/migration/prosody_files.lua [deleted file]
tools/migration/prosody_sql.lua [deleted file]

diff --git a/tools/migration/Makefile b/tools/migration/Makefile
new file mode 100644 (file)
index 0000000..839f2d8
--- /dev/null
@@ -0,0 +1,38 @@
+
+include ../../config.unix
+
+BIN = $(DESTDIR)$(PREFIX)/bin
+CONFIG = $(DESTDIR)$(SYSCONFDIR)
+SOURCE = $(DESTDIR)$(PREFIX)/lib/prosody
+DATA = $(DESTDIR)$(DATADIR)
+MAN = $(DESTDIR)$(PREFIX)/share/man
+
+INSTALLEDSOURCE = $(PREFIX)/lib/prosody
+INSTALLEDCONFIG = $(SYSCONFDIR)
+INSTALLEDMODULES = $(PREFIX)/lib/prosody/modules
+INSTALLEDDATA = $(DATADIR)
+
+SOURCE_FILES = main.lua migrator/*.lua
+
+all: prosody-migrator.install migrator.cfg.lua.install $(SOURCE_FILES)
+
+install: prosody-migrator.install migrator.cfg.lua.install
+       install -d $(BIN) $(CONFIG) $(SOURCE) $(SOURCE)/migrator
+       install -d $(MAN)/man1
+       install -d $(SOURCE)/migrator
+       install -m755 ./prosody-migrator.install $(BIN)/prosody-migrator
+       install -m644 $(SOURCE_FILES) $(SOURCE)/migrator
+       test -e $(CONFIG)/migrator.cfg.lua || install -m644 migrator.cfg.lua.install $(CONFIG)/migrator.cfg.lua
+
+clean:
+       rm -f prosody-migrator.install
+       rm -f migrator.cfg.lua.install
+
+prosody-migrator.install: main.lua
+       sed "s|^CFG_SOURCEDIR=.*;$$|CFG_SOURCEDIR='$(INSTALLEDSOURCE)';|; \
+               s|^CFG_CONFIGDIR=.*;$$|CFG_CONFIGDIR='$(INSTALLEDCONFIG)';|;" \
+                       < main.lua > prosody-migrator.install
+
+migrator.cfg.lua.install: migrator.cfg.lua
+       sed "s|^local data_path = .*;$$|local data_path = '$(INSTALLEDDATA)';|;" \
+               < migrator.cfg.lua > migrator.cfg.lua.install
index e4891966d481b98abb42e3a618d6dc8b2a015894..82eeab9d90d527465e71a5ec6f70d34fd4b7be1b 100644 (file)
@@ -1,4 +1,9 @@
-local default_config = "./migrator.cfg.lua";
+#!/usr/bin/env lua
+
+CFG_SOURCEDIR=os.getenv("PROSODY_SRCDIR");
+CFG_CONFIGDIR=os.getenv("PROSODY_CFGDIR");
+
+local default_config = (CFG_CONFIGDIR or ".").."/migrator.cfg.lua";
 
 -- Command-line parsing
 local options = {};
@@ -45,9 +50,12 @@ end
 
 config_chunk();
 
-if not package.loaded["util.json"] then
+if CFG_SOURCEDIR then
+       package.path = CFG_SOURCEDIR.."/?.lua;"..package.path;
+       package.cpath = CFG_SOURCEDIR.."/?.so;"..package.cpath;
+elseif not package.loaded["util.json"] then
        package.path = "../../?.lua;"..package.path
-       package.cpath = "../../?.dll;"..package.cpath
+       package.cpath = "../../?.so;"..package.cpath
 end
 
 local have_err;
@@ -66,14 +74,14 @@ end
 if not config[from_store].type then
        have_err = true;
        print("Error: Input store type not specified in the config file");
-elseif not pcall(require, config[from_store].type) then
+elseif not pcall(require, "migrator."..config[from_store].type) then
        have_err = true;
        print("Error: Unrecognised store type for '"..from_store.."': "..config[from_store].type);
 end
 if not config[to_store].type then
        have_err = true;
        print("Error: Output store type not specified in the config file");
-elseif not pcall(require, config[to_store].type) then
+elseif not pcall(require, "migrator."..config[to_store].type) then
        have_err = true;
        print("Error: Unrecognised store type for '"..to_store.."': "..config[to_store].type);
 end
@@ -88,13 +96,14 @@ if have_err then
        for store in pairs(config) do
                print("", store);
        end
+       print("");
        os.exit(1);
 end
        
 local itype = config[from_store].type;
 local otype = config[to_store].type;
-local reader = require(itype).reader(config[from_store]);
-local writer = require(otype).writer(config[to_store]);
+local reader = require("migrator."..itype).reader(config[from_store]);
+local writer = require("migrator."..otype).writer(config[to_store]);
 
 local json = require "util.json";
 
diff --git a/tools/migration/migrator/mtools.lua b/tools/migration/migrator/mtools.lua
new file mode 100644 (file)
index 0000000..e7b774b
--- /dev/null
@@ -0,0 +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;
diff --git a/tools/migration/migrator/prosody_files.lua b/tools/migration/migrator/prosody_files.lua
new file mode 100644 (file)
index 0000000..0a610d0
--- /dev/null
@@ -0,0 +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 "migrator.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;
diff --git a/tools/migration/migrator/prosody_sql.lua b/tools/migration/migrator/prosody_sql.lua
new file mode 100644 (file)
index 0000000..3a9172f
--- /dev/null
@@ -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;
diff --git a/tools/migration/mtools.lua b/tools/migration/mtools.lua
deleted file mode 100644 (file)
index e7b774b..0000000
+++ /dev/null
@@ -1,56 +0,0 @@
-
-
-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;
diff --git a/tools/migration/prosody_files.lua b/tools/migration/prosody_files.lua
deleted file mode 100644 (file)
index df67c24..0000000
+++ /dev/null
@@ -1,133 +0,0 @@
-
-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;
diff --git a/tools/migration/prosody_sql.lua b/tools/migration/prosody_sql.lua
deleted file mode 100644 (file)
index ff33652..0000000
+++ /dev/null
@@ -1,174 +0,0 @@
-
-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;