-- mod_storage_gdbm

-- Copyright (C) 2014-2015 Kim Alvefur

--

-- This file is MIT/X11 licensed.

-- 

-- Depends on lgdbm:

-- http://webserver2.tecgraf.puc-rio.br/~lhf/ftp/lua/#lgdbm

--

-- luacheck: globals open purge

local gdbm = require"gdbm";

local path = require"util.paths";

local lfs = require"lfs";

local st = require"util.stanza";

local uuid = require"util.uuid".generate;

local serialization = require"util.serialization";

local serialize = serialization.serialize;

local deserialize = serialization.deserialize;

local g_set, g_get, g_del = gdbm.replace, gdbm.fetch, gdbm.delete;

local g_first, g_next = gdbm.firstkey, gdbm.nextkey;

local empty = {};

local function id(v) return v; end

local function is_stanza(s)

	return getmetatable(s) == st.stanza_mt;

end

local function t(c, a, b)

	if c then return a; end return b;

end

local base_path = path.resolve_relative_path(prosody.paths.data, module.host);

lfs.mkdir(base_path);

local cache = {};

local keyval = {};

local keyval_mt = { __index = keyval, suffix = ".db" };

function keyval:set(user, value)

	if type(value) == "table" and next(value) == nil then

		value = nil;

	end

	if value ~= nil then

		value = serialize(value);

	end

	local ok, err = (value and g_set or g_del)(self._db, user or "@", value);

	if not ok then return nil, err; end

	return true;

end

function keyval:get(user)

	local data, err = g_get(self._db, user or "@");

	if not data then return nil, err; end

	return deserialize(data);

end

local function g_keys(db, key)

	return (key == nil and g_first or g_next)(db, key);

end

function keyval:users()

	return g_keys, self._db, nil;

end

local archive = {};

local archive_mt = { __index = archive, suffix = ".adb" };

archive.get = keyval.get;

archive.set = keyval.set;

function archive:append(username, key, value, when, with)

	if type(when) ~= "number" then

		when, with, value = value, when, with;

	end

	key = key or uuid();

	local meta = self:get(username);

	if not meta then

		meta = {};

	end

	local i = meta[key] or #meta+1;

	local type;

	if is_stanza(value) then

		type, value = "stanza", st.preserialize(value);

	end

	meta[i] = { key = key, when = when, with = with, type = type };

	meta[key] = i;

	local prefix = (username or "@") .. "#";

	local ok, err = self:set(prefix..key, value);

	if not ok then return nil, err; end

	ok, err = self:set(username, meta);

	if not ok then return nil, err; end

	return key;

end

local deserialize_map = {

	stanza = st.deserialize;

};

function archive:find(username, query)

	query = query or empty;

	local meta = self:get(username) or empty;

	local prefix = (username or "@") .. "#";

	local r = query.reverse;

	local d = t(r, -1, 1);

	local s = meta[t(r, query.before, query.after)];

	local limit = query.limit;

	if s then

		s = s + d;

	else

		s = t(r, #meta, 1)

	end

	local e = t(r, 1, #meta);

	local c = 0;

	return function ()

		if limit and c >= limit then return end

		local item, value;

		for i = s, e, d do

			item = meta[i];

			if (not query.with or item.with == query.with)

			and (not query.start or item.when >= query.start)

			and (not query["end"] or item.when <= query["end"]) then

				s = i + d; c = c + 1;

				value = self:get(prefix..item.key);

				return item.key, (deserialize_map[item.type] or id)(value), item.when, item.with;

			end

		end

	end

end

local drivers = {

	keyval = keyval_mt;

	archive = archive_mt;

}

function open(_, store, typ)

	typ = typ or "keyval";

	local driver_mt = drivers[typ];

	if not driver_mt then

		return nil, "unsupported-store";

	end

	local db_path = path.join(base_path, store) .. driver_mt.suffix;

	local db = cache[db_path];

	if not db then

		db = assert(gdbm.open(db_path, "c"));

		cache[db_path] = db;

	end

	return setmetatable({ _db = db; _path = db_path; store = store, type = typ }, driver_mt);

end

function purge(_, user)

	for dir in lfs.dir(base_path) do

		local name, ext = dir:match("^(.-)%.a?db$");

		if ext == ".db" then

			open(_, name, "keyval"):set(user, nil);

		elseif ext == ".adb" then

			open(_, name, "archive"):delete(user);

		end

	end

	return true;

end

function module.unload()

	for db_path, db in pairs(cache) do

		module:log("debug", "Closing db at %q", db_path);

		gdbm.reorganize(db);

		gdbm.sync(db);

		gdbm.close(db);

	end

end

module:provides"storage";