lmdbal/src/cache.hpp

/*
 * LMDB Abstraction Layer.
 * Copyright (C) 2023  Yury Gubich <blue@macaw.me>
 *
 * This program is free software: you can redistribute it and/or modify
 * it under the terms of the GNU General Public License as published by
 * the Free Software Foundation, either version 3 of the License, or
 * (at your option) any later version.
 *
 * This program is distributed in the hope that it will be useful,
 * but WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 * GNU General Public License for more details.
 *
 * You should have received a copy of the GNU General Public License
 * along with this program.  If not, see <http://www.gnu.org/licenses/>.
 */

#ifndef LMDBAL_CACHE_HPP
#define LMDBAL_CACHE_HPP

#include "cache.h"
#include "exceptions.h"

/**
 * \class LMDBAL::Cache
 * \brief Storage with additional caching in std::map
 *
 * \tparam K type of the keys of the cache
 * \tparam V type of the values of the cache
 *
 * You can receive an instance of this class calling LMDBAL::Base::addCache(const std::string&)
 * if the database is yet closed and you're defining the storages you're going to need.
 * Or you can call LMDBAL::Base::getCache(const std::string&) if you didn't save a pointer to the cache at first
 *
 * You are not supposed to instantiate or destory instances of this class yourself!
 */

/**
 * \brief Creates a cache
 *
 * \param[in] _name     - name of the new cache
 * \param[in] parent    - parent database pointed (borrowed)
 */
template<class K, class V>
LMDBAL::Cache<K, V>::Cache(const std::string& _name, Base* parent):
    Storage<K, V>(_name, parent),
    mode(Mode::nothing),
    cache(new std::map<K, V>()),
    abscent(new std::set<K>()),
    sizeDifference(0),
    transactionCache(new TransactionCache) {}

/**
 * \brief Destroys a cache
 */
template<class K, class V>
LMDBAL::Cache<K, V>::~Cache() {
    delete transactionCache;
    delete cache;
    delete abscent;
}

template<class K, class V>
void LMDBAL::Cache<K, V>::addRecord(const K& key, const V& value) {
    iStorage::ensureOpened(iStorage::addRecordMethodName);

    if (cache->count(key) > 0)
        iStorage::throwDuplicate(iStorage::toString(key));

    Storage<K, V>::addRecord(key, value);
    handleAddRecord(key, value);
}

template<class K, class V>
void LMDBAL::Cache<K, V>::addRecord(const K& key, const V& value, TransactionID txn) {
    iStorage::ensureOpened(iStorage::addRecordMethodName);

    if (cache->count(key) > 0)
        iStorage::throwDuplicate(iStorage::toString(key));

    Storage<K, V>::addRecord(key, value, txn);

    typename TransactionCache::iterator tc = transactionCache->find(txn);
    if (tc != transactionCache->end()) {
        std::pair<K, V>* pair = new std::pair<K, V>(key, value);
        tc->second.emplace_back(Operation::add, pair);
    }
}

template<class K, class V>
void LMDBAL::Cache<K, V>::handleAddRecord(const K& key, const V& value) {
    cache->insert(std::make_pair(key, value));

    if (mode != Mode::full)
        abscent->erase(key);
}

template<class K, class V>
bool LMDBAL::Cache<K, V>::forceRecord(const K& key, const V& value) {
    iStorage::ensureOpened(iStorage::forceRecordMethodName);

    bool added = Storage<K, V>::forceRecord(key, value);
    handleForceRecord(key, value, added);

    return added;
}

template<class K, class V>
bool LMDBAL::Cache<K, V>::forceRecord(const K& key, const V& value, TransactionID txn) {
    iStorage::ensureOpened(iStorage::forceRecordMethodName);

    bool added = Storage<K, V>::forceRecord(key, value, txn);

    typename TransactionCache::iterator tc = transactionCache->find(txn);
    if (tc != transactionCache->end()) {
        std::tuple<bool, K, V>* t = new std::tuple<bool, K, V>(added, key, value);
        tc->second.emplace_back(Operation::force, t);
    }

    return added;
}

template<class K, class V>
void LMDBAL::Cache<K, V>::handleForceRecord(const K& key, const V& value, bool added) {
    if (mode == Mode::full) {
        (*cache)[key] = value;
    } else {
        if (added)
            abscent->erase(key);

        std::pair<typename std::map<K, V>::iterator, bool> result =
            cache->insert(std::make_pair(key, value));
        if (!result.second)
            result.first->second = value;
        else if (!added)    //this way database had value but cache didn't, so, need to decrease sizeDifference
            handleMode();
    }
}

template<class K, class V>
void LMDBAL::Cache<K, V>::changeRecord(const K& key, const V& value) {
    iStorage::ensureOpened(iStorage::changeRecordMethodName);

    if (mode == Mode::full) {
        typename std::map<K, V>::iterator itr = cache->find(key);
        if (itr == cache->end())
            iStorage::throwNotFound(iStorage::toString(key));

        Storage<K, V>::changeRecord(key, value);
        itr->second = value;
    } else {
        if (abscent->count(key) > 0)
            iStorage::throwNotFound(iStorage::toString(key));

        try {
            Storage<K, V>::changeRecord(key, value);
            typename std::pair<typename std::map<K, V>::iterator, bool> res =
                cache->insert(std::make_pair(key, value));
            if (!res.second)
                res.first->second = value;
            else
                handleMode();
        } catch (const NotFound& error) {
            abscent->insert(key);
            throw error;
        }
    }
}

template<class K, class V>
void LMDBAL::Cache<K, V>::changeRecord(const K& key, const V& value, TransactionID txn) {
    iStorage::ensureOpened(iStorage::changeRecordMethodName);

    if (mode == Mode::full) {
        typename std::map<K, V>::iterator itr = cache->find(key);
        if (itr == cache->end())
            iStorage::throwNotFound(iStorage::toString(key));

        Storage<K, V>::changeRecord(key, value, txn);
    } else {
        if (abscent->count(key) > 0)
            iStorage::throwNotFound(iStorage::toString(key));

        try {
            Storage<K, V>::changeRecord(key, value, txn);
        } catch (const NotFound& error) {
            abscent->insert(key);
            throw error;
        }
    }

    typename TransactionCache::iterator tc = transactionCache->find(txn);
    if (tc != transactionCache->end()) {
        std::pair<K, V>* pair = new std::pair<K, V>(key, value);
        tc->second.emplace_back(Operation::add, pair);
    }
}

template<class K, class V>
void LMDBAL::Cache<K, V>::handleChangeRecord(const K& key, const V& value) {
    if (mode == Mode::full) {
        cache->at(key) = value;
    } else {
        typename std::pair<typename std::map<K, V>::iterator, bool> res =
            cache->insert(std::make_pair(key, value));
        if (!res.second)
            res.first->second = value;
        else
            handleMode();
    }
}

template<class K, class V>
V LMDBAL::Cache<K, V>::getRecord(const K& key) const {
    iStorage::ensureOpened(iStorage::getRecordMethodName);

    V value;
    Cache<K, V>::getRecord(key, value);
    return value;
}

template<class K, class V>
void LMDBAL::Cache<K, V>::getRecord(const K& key, V& out) const {
    iStorage::ensureOpened(iStorage::getRecordMethodName);

    typename std::map<K, V>::const_iterator itr = cache->find(key);
    if (itr != cache->end()) {
        out = itr->second;
        return;
    }

    if (mode == Mode::full || abscent->count(key) != 0)
        iStorage::throwNotFound(iStorage::toString(key));

    try {
        Storage<K, V>::getRecord(key, out);
        cache->insert(std::make_pair(key, out));
        handleMode();
        return;
    } catch (const NotFound& error) {
        if (mode != Mode::full)
            abscent->insert(key);

        throw error;
    }
}

template<class K, class V>
V LMDBAL::Cache<K, V>::getRecord(const K& key, TransactionID txn) const {
    iStorage::ensureOpened(iStorage::getRecordMethodName);

    V value;
    Cache<K, V>::getRecord(key, value, txn);
    return value;
}

template<class K, class V>
void LMDBAL::Cache<K, V>::getRecord(const K& key, V& out, TransactionID txn) const {
    iStorage::ensureOpened(iStorage::getRecordMethodName);

    //if there are any changes made within this transaction
    //I will be able to see them among pending changes
    //so, I'm going to go through them in reverse order
    //and check every key. If it has anything to do this requested key
    //there is a way to tell...
    bool currentTransaction = false;
    typename TransactionCache::const_iterator tc = transactionCache->find(txn);
    if (tc != transactionCache->end()) {
        currentTransaction = true;
        const Queue& queue = tc->second;
        for (typename Queue::const_reverse_iterator i = queue.rbegin(), end = queue.rend(); i != end; ++i) {
            const Entry& entry = *i;

            switch (entry.first) {
                case Operation::add:
                    if (static_cast<std::pair<K, V>*>(entry.second)->first == key) {
                        out = static_cast<std::pair<K, V>*>(entry.second)->second;
                        return;
                    }
                    break;
                case Operation::remove:
                        iStorage::throwNotFound(iStorage::toString(key));
                    break;
                case Operation::change:
                    if (static_cast<std::pair<K, V>*>(entry.second)->first == key) {
                        out = static_cast<std::pair<K, V>*>(entry.second)->second;
                        return;
                    }

                    break;
                case Operation::force:
                    if (std::get<1>(*static_cast<std::tuple<bool, K, V>*>(entry.second)) == key) {
                        out = std::get<2>(*static_cast<std::tuple<bool, K, V>*>(entry.second));
                        return;
                    }
                    break;
                case Operation::drop:
                    iStorage::throwNotFound(iStorage::toString(key));
                    break;
                case Operation::replace: {
                    std::map<K, V>* newMap = static_cast<std::map<K, V>*>(entry.second);
                    typename std::map<K, V>::const_iterator vitr = newMap->find(key);
                    if (vitr != newMap->end()) {
                        out = vitr->second;
                        return;
                    } else {
                        iStorage::throwNotFound(iStorage::toString(key));
                    }
                }
                    break;
                case Operation::addMany: {
                    const std::tuple<bool, SizeType, std::map<K, V>>& tuple =
                        *static_cast<std::tuple<bool, SizeType, std::map<K, V>>*>(entry.second);
                    const std::map<K, V>& newElements = std::get<2>(tuple);
                    typename std::map<K, V>::const_iterator vitr = newElements.find(key);
                    if (vitr != newElements.end()) {
                        out = vitr->second;
                        return;
                    }
                }
                    break;
            }
        }
    }
    //... but if nothing was found or if the transaction is not the one
    //which caused the changes i just need to check it among local cache

    typename std::map<K, V>::const_iterator itr = cache->find(key);
    if (itr != cache->end()) {
        out = itr->second;
        return;
    }

    if (mode == Mode::full || abscent->count(key) != 0)
        iStorage::throwNotFound(iStorage::toString(key));

    try {
        Storage<K, V>::getRecord(key, out, txn);
        if (!currentTransaction) {
            cache->insert(std::make_pair(key, out));
            handleMode();
        }
        return;
    } catch (const NotFound& error) {
        if (!currentTransaction && mode != Mode::full)
            abscent->insert(key);

        throw error;
    }
}

template<class K, class V>
bool LMDBAL::Cache<K, V>::checkRecord(const K& key) const {
    iStorage::ensureOpened(iStorage::checkRecordMethodName);

    typename std::map<K, V>::const_iterator itr = cache->find(key);
    if (itr != cache->end())
        return true;

    if (mode == Mode::full || abscent->count(key) != 0)
        return false;

    try {
        V value = Storage<K, V>::getRecord(key);
        cache->insert(std::make_pair(key, value));
        handleMode();
        return true;
    } catch (const NotFound& error) {
        if (mode != Mode::full)
            abscent->insert(key);

        return false;
    }
}

template<class K, class V>
bool LMDBAL::Cache<K, V>::checkRecord(const K& key, TransactionID txn) const {
    iStorage::ensureOpened(iStorage::checkRecordMethodName);

    //if there are any changes made within this transaction
    //I will be able to see them among pending changes
    //so, I'm going to go through them in reverse order
    //and check every key. If it has anything to do this requested key
    //there is a way to tell...
    bool currentTransaction = false;
    typename TransactionCache::const_iterator tc = transactionCache->find(txn);
    if (tc != transactionCache->end()) {
        currentTransaction = true;
        const Queue& queue = tc->second;
        for (typename Queue::const_reverse_iterator i = queue.rbegin(), end = queue.rend(); i != end; ++i) {
            const Entry& entry = *i;

            switch (entry.first) {
                case Operation::add:
                    if (static_cast<std::pair<K, V>*>(entry.second)->first == key)
                        return true;
                    break;
                case Operation::remove:
                    if (*static_cast<K*>(entry.second) == key)
                        return false;
                    break;
                case Operation::change:
                    if (static_cast<std::pair<K, V>*>(entry.second)->first == key)
                        return true;
                    break;
                case Operation::force:
                    if (std::get<1>(*static_cast<std::tuple<bool, K, V>*>(entry.second)) == key)
                        return true;
                    break;
                case Operation::drop:
                    return false;
                    break;
                case Operation::replace:
                    if (static_cast<std::map<K, V>*>(entry.second)->count(key) > 0)
                        return true;
                    else
                        return false;
                    break;
                case Operation::addMany:
                    if (std::get<2>(
                            *static_cast<std::tuple<bool, SizeType, std::map<K, V>>*>(entry.second)
                        ).count(key) > 0)
                        return true;
                    break;
            }
        }
    }
    //... but if nothing was found or if the transaction is not the one
    //which caused the changes i just need to check it among local cache

    typename std::map<K, V>::const_iterator itr = cache->find(key);
    if (itr != cache->end())
        return true;

    if (mode == Mode::full || abscent->count(key) != 0)
        return false;

    try {
        V value = Storage<K, V>::getRecord(key, txn);
        if (!currentTransaction) {
            cache->insert(std::make_pair(key, value));
            handleMode();
        }
        return true;
    } catch (const NotFound& error) {
        if (!currentTransaction && mode != Mode::full)
            abscent->insert(key);

        return false;
    }
}

template<class K, class V>
std::map<K, V> LMDBAL::Cache<K, V>::readAll() const {
    iStorage::ensureOpened(iStorage::readAllMethodName);

    if (mode != Mode::full) {               //there is a room for optimization
        mode = Mode::full;                  //I can read and deserialize only those values
        *cache = Storage<K, V>::readAll();  //that are missing in the cache
        abscent->clear();
        sizeDifference = 0;
    }

    return *cache;
}

template<class K, class V>
void LMDBAL::Cache<K, V>::readAll(std::map<K, V>& out) const {
    iStorage::ensureOpened(iStorage::readAllMethodName);

    if (mode != Mode::full) {               //there is a room for optimization
        mode = Mode::full;                  //I can read and deserialize only those values
        Storage<K, V>::readAll(out);        //that are missing in the cache
        *cache = out;
        abscent->clear();
        sizeDifference = 0;
    }
}

template<class K, class V>
std::map<K, V> LMDBAL::Cache<K, V>::readAll(TransactionID txn) const {
    iStorage::ensureOpened(iStorage::readAllMethodName);

    std::map<K, V> out;
    readAll(out, txn);

    return out;
}

template<class K, class V>
void LMDBAL::Cache<K, V>::readAll(std::map<K, V>& out, TransactionID txn) const {
    iStorage::ensureOpened(iStorage::readAllMethodName);

    typename TransactionCache::iterator tc = transactionCache->find(txn);
    if (tc != transactionCache->end()) {
        Queue& queue = tc->second;
        if (mode != Mode::full) {
            out = *cache;

            for (typename Queue::const_iterator i = queue.begin(), end = queue.end(); i != end; ++i) {
                const Entry& entry = *i;
                switch (entry.first) {
                    case Operation::add:
                        out.insert(*static_cast<std::pair<K, V>*>(entry.second));
                        break;
                    case Operation::remove:
                        out.erase(*static_cast<K*>(entry.second));
                        break;
                    case Operation::change: {
                        std::pair<K, V>* pair = static_cast<std::pair<K, V>*>(entry.second);
                        out.at(pair->first) = pair->second;
                    }
                        break;
                    case Operation::force:{
                        const std::tuple<bool, K, V>& tuple =
                            *static_cast<std::tuple<bool, K, V>*>(entry.second);
                        out[std::get<1>(tuple)] = std::get<2>(tuple);
                    }
                        break;
                    case Operation::drop:
                        out.clear();
                        break;
                    case Operation::replace:
                        out = *static_cast<std::map<K, V>*>(entry.second);
                        break;
                    case Operation::addMany: {
                        const std::tuple<bool, SizeType, std::map<K, V>>& t =
                            *static_cast<std::tuple<bool, SizeType, std::map<K, V>>*>(entry.second);
                        const std::map<K, V>& added = std::get<2>(t);
                        bool overwrite = std::get<0>(t);
                        for (const std::pair<const K, V>& pair : added) {
                            if (overwrite)
                                out[pair.first] = pair.second;
                            else
                                out.insert(pair);
                        }
                    }
                        break;
                }
            }
        } else {
            Storage<K, V>::readAll(out, txn);

            //queue.clear();                                                    //since I'm getting a complete state of the database
            queue.emplace_back(Operation::replace, new std::map<K, V>(out));    //I can as well erase all previous cache entries
        }
    } else {
        if (mode != Mode::full) {               //there is a room for optimization
            mode = Mode::full;                  //I can read and deserialize only those values
            Storage<K, V>::readAll(out);        //that are missing in the cache
            *cache = out;
            abscent->clear();
            sizeDifference = 0;
        }
    }
}

template<class K, class V>
void LMDBAL::Cache<K, V>::replaceAll(const std::map<K, V>& data) {
    Storage<K, V>::replaceAll(data);
    *cache = data;

    if (mode != Mode::full) {
        mode = Mode::full;
        abscent->clear();
        sizeDifference = 0;
    }
}

template<class K, class V>
void LMDBAL::Cache<K, V>::replaceAll(const std::map<K, V>& data, TransactionID txn) {
    Storage<K, V>::replaceAll(data, txn);

    typename TransactionCache::iterator tc = transactionCache->find(txn);
    if (tc != transactionCache->end()) {
        //queue.clear();
        std::map<K, V>* map = new std::map<K, V>(data);     //since I'm getting a complete state of the database
        tc->second.emplace_back(Operation::replace, map);   //I can as well erase all previous cache entries
    }
}

template<class K, class V>
void LMDBAL::Cache<K, V>::handleReplaceAll(std::map<K, V>* data) {
    delete cache;
    cache = data;

    if (mode != Mode::full) {
        mode = Mode::full;
        abscent->clear();
        sizeDifference = 0;
    }
}

template<class K, class V>
LMDBAL::SizeType LMDBAL::Cache<K, V>::addRecords(const std::map<K, V>& data, bool overwrite) {
    SizeType newSize = Storage<K, V>::addRecords(data, overwrite);
    handleAddRecords(data, overwrite, newSize);

    return newSize;
}

template<class K, class V>
LMDBAL::SizeType LMDBAL::Cache<K, V>::addRecords(const std::map<K, V>& data, TransactionID txn, bool overwrite) {
    SizeType newSize = Storage<K, V>::addRecords(data, txn, overwrite);

    typename TransactionCache::iterator tc = transactionCache->find(txn);
    if (tc != transactionCache->end()) {
        std::tuple<bool, SizeType, std::map<K, V>>* tuple =
            new std::tuple<bool, SizeType, std::map<K, V>>(overwrite, newSize, data);
        tc->second.emplace_back(Operation::addMany, tuple);
    }

    return newSize;
}

template<class K, class V>
void LMDBAL::Cache<K, V>::handleAddRecords(const std::map<K, V>& data, bool overwrite, SizeType newSize) {
    if (mode == Mode::nothing)
        mode = Mode::size;

    std::map<K, V>& c = *cache;
    std::set<K>& a = *abscent;
    for (const std::pair<const K, V>& pair : data) {
        std::pair<typename std::map<K, V>::iterator, bool> res = c.insert(pair);
        if (!res.second) {
            if (overwrite)
                res.first->second = pair.second;
        } else if (mode != Mode::full) {
            a.erase(pair.first);
        }
    }

    if (mode != Mode::full) {
        sizeDifference = newSize - c.size();
        if (sizeDifference == 0) {
            mode = Mode::full;
            abscent->clear();
        }
    }
}

template<class K, class V>
void LMDBAL::Cache<K, V>::removeRecord(const K& key) {
    iStorage::ensureOpened(iStorage::removeRecordMethodName);

    bool noKey = false;
    if (mode != Mode::full)
        noKey = cache->count(key) == 0;
    else
        noKey = abscent->count(key) > 0;

    if (noKey)
        iStorage::throwNotFound(iStorage::toString(key));

    Storage<K, V>::removeRecord(key);
    handleRemoveRecord(key);
}

template<class K, class V>
void LMDBAL::Cache<K, V>::removeRecord(const K& key, TransactionID txn) {
    iStorage::ensureOpened(iStorage::removeRecordMethodName);

    bool noKey = false;
    if (mode != Mode::full)
        noKey = cache->count(key) == 0;
    else
        noKey = abscent->count(key) > 0;

    if (noKey)
        iStorage::throwNotFound(iStorage::toString(key));

    Storage<K, V>::removeRecord(key, txn);

    typename TransactionCache::iterator tc = transactionCache->find(txn);
    if (tc != transactionCache->end())
        tc->second.emplace_back(Operation::remove, new K(key));
}

template<class K, class V>
void LMDBAL::Cache<K, V>::handleRemoveRecord(const K& key)  {
    if (cache->erase(key) == 0)     //if it was not cached and we are now in size mode then the sizeDifference would decrease
        handleMode();

    if (mode != Mode::full)
        abscent->insert(key);
}

template<class K, class V>
uint32_t LMDBAL::Cache<K, V>::count() const {
    switch (mode) {
        case Mode::nothing:
        {
            uint32_t sz = Storage<K, V>::count();
            sizeDifference = sz - cache->size();
            if (sz == 0) {
                mode = Mode::full;
                abscent->clear();
            } else {
                mode = Mode::size;
            }
            return sz;
        }
        case Mode::size:
            return cache->size() + sizeDifference;
        case Mode::full:
            return cache->size();
        default:
            return 0;   //unreachable, no such state, just to suppress the waring
    }
}

template<class K, class V>
uint32_t LMDBAL::Cache<K, V>::count(TransactionID txn) const {

    int32_t diff = 0;
    bool currentTransaction = false;
    typename TransactionCache::const_iterator tc = transactionCache->find(txn);
    if (tc != transactionCache->end()) {
        currentTransaction = true;
        const Queue& queue = tc->second;
        for (typename Queue::const_reverse_iterator i = queue.rbegin(), end = queue.rend(); i != end; ++i) {
            const Entry& entry = *i;

            switch (entry.first) {
                case Operation::add:
                    ++diff;
                    break;
                case Operation::remove:
                    --diff;
                    break;
                case Operation::change:
                    break;
                case Operation::force:
                    if (std::get<0>(*static_cast<std::tuple<bool, K, V>*>(entry.second)))
                        ++diff;
                    break;
                case Operation::drop:
                    return false;
                    break;
                case Operation::replace:
                    return static_cast<std::map<K, V>*>(entry.second)->size() + diff;
                    break;
                case Operation::addMany:
                    return std::get<1>(*static_cast<std::tuple<bool, SizeType, std::map<K, V>>*>(entry.second)) + diff;
                    break;
            }
        }
    }

    switch (mode) {
        case Mode::nothing: {
            uint32_t sz = Storage<K, V>::count(txn);
            if (!currentTransaction) {
                sizeDifference = sz - cache->size();
                if (sz == 0) {
                    mode = Mode::full;
                    abscent->clear();
                } else {
                    mode = Mode::size;
                }
            }
            return sz;
        }
        case Mode::size:
            return cache->size() + sizeDifference + diff;
        case Mode::full:
            return cache->size() + diff;
        default:
            return 0;   //unreachable, no such state, just to suppress the waring
    }
}

template<class K, class V>
void LMDBAL::Cache<K, V>::handleMode() const {
    if (mode == Mode::size) {
        --sizeDifference;
        if (sizeDifference == 0) {
            mode = Mode::full;
            abscent->clear();
        }
    }
}

template<class K, class V>
int LMDBAL::Cache<K, V>::drop(TransactionID transaction) {
    int res = Storage<K, V>::drop(transaction);

    typename TransactionCache::iterator tc = transactionCache->find(transaction);
    if (tc != transactionCache->end())
        tc->second.emplace_back(Operation::drop, nullptr);

    return res;
}

template<class K, class V>
void LMDBAL::Cache<K, V>::handleDrop() {
    cache->clear();
    abscent->clear();
    mode = Mode::full;
    sizeDifference = 0;
}

template<class K, class V>
void LMDBAL::Cache<K, V>::transactionStarted(TransactionID txn, bool readOnly) const {
    if (!readOnly)
        transactionCache->emplace(txn, Queue());
}

template<class K, class V>
void LMDBAL::Cache<K, V>::transactionCommited(TransactionID txn) {
    typename TransactionCache::iterator itr = transactionCache->find(txn);
    if (itr != transactionCache->end()) {
        Queue& queue = itr->second;
        for (const Entry& entry : queue)
            handleTransactionEntry(entry);

        transactionCache->erase(itr);
    }
}

template<class K, class V>
void LMDBAL::Cache<K, V>::transactionAborted(TransactionID txn) const {
    typename TransactionCache::iterator itr = transactionCache->find(txn);
    if (itr != transactionCache->end()) {
        Queue& queue = itr->second;
        for (const Entry& entry : queue)
            destroyTransactionEntry(entry);

        transactionCache->erase(itr);
    }
}

template<class K, class V>
void LMDBAL::Cache<K, V>::handleTransactionEntry(const Entry& entry) {
    switch (entry.first) {
        case Operation::add: {
            std::pair<K, V>* pair = static_cast<std::pair<K, V>*>(entry.second);
            handleAddRecord(pair->first, pair->second);
            delete pair;
        }
            break;
        case Operation::remove: {
            K* key = static_cast<K*>(entry.second);
            handleRemoveRecord(*key);
            delete key;
        }

            break;
        case Operation::change: {
            std::pair<K, V>* pair = static_cast<std::pair<K, V>*>(entry.second);
            handleChangeRecord(pair->first, pair->second);
            delete pair;
        }
        case Operation::force: {
            std::tuple<bool, K, V>* tuple = static_cast<std::tuple<bool, K, V>*>(entry.second);
            const std::tuple<bool, K, V>& t = *tuple;
            handleForceRecord(std::get<1>(t), std::get<2>(t), std::get<0>(t));
            delete tuple;
        }
            break;
        case Operation::drop:
            handleDrop();
            break;
        case Operation::replace:
            handleReplaceAll(static_cast<std::map<K, V>*>(entry.second));   //I take ownership, no need to delete
            break;
        case Operation::addMany: {
            std::tuple<bool, SizeType, std::map<K, V>>* tuple = static_cast<std::tuple<bool, SizeType, std::map<K, V>>*>(entry.second);
            const std::tuple<bool, SizeType, std::map<K, V>>& t = * tuple;
            handleAddRecords(std::get<2>(t), std::get<0>(t), std::get<1>(t));
            delete tuple;
        }
            break;
    }
}

template<class K, class V>
void LMDBAL::Cache<K, V>::destroyTransactionEntry(const Entry& entry) const {
    switch (entry.first) {
        case Operation::add:
            delete static_cast<std::pair<K, V>*>(entry.second);
            break;
        case Operation::remove:
            delete static_cast<K*>(entry.second);
            break;
        case Operation::change:
            delete static_cast<std::pair<K, V>*>(entry.second);
            break;
        case Operation::force:
            delete static_cast<std::tuple<bool, K, V>*>(entry.second);
            break;
        case Operation::drop:
            break;
        case Operation::replace:
            delete static_cast<std::map<K, V>*>(entry.second);
            break;
        case Operation::addMany:
            delete static_cast<std::tuple<bool, SizeType, std::map<K, V>>*>(entry.second);
            break;
    }
}

#endif //LMDBAL_CACHE_HPP