/*
   Copyright (c) 2019 by The ThreadDB Project
   All Rights Reserved.

   ThreadDB undergoes the BSD License 2.0. You should have received a copy along with this program; if not, write to the ThreadDB Project.
   To obtain a full unlimited version contact thethreaddbproject(at)gmail.com.

   threaddbSTLWraper.h - standard library container support wrapper classes
*/

#pragma once

#include "threaddbCPP.h"
#include <atomic>
#include <mutex>
#include <vector>
#include <deque>
#include <memory>
#include <stdio.h>
#include <string.h>

/**
   \file
   \brief         Implementation of the threadDB integration for std:: container utilization.
   \details       This module provides the necessary classes to directly utilize the std:: containers.
                  To handle different scenarios of fixed and dynamic data objects efficiently, specialized
                  implementations are available.
                  The additional pool class is responsible for keeping track of the open (registered) link
                  handles not yet processed. Since these handles are updated delayed it is important that
                  they are not deleted prior unregistering.
*/

namespace tdb
{
    /**
    *  @brief Trait class for size of fixed length data items.
    */

    template<class T>
    struct size_
    {
        size_() {}
        size_(const T&) {}
        uint32_t size() const { return sizeof(T); }
    };

    /**
    *  @brief Trait class for size of variable length data items (std::string).
    */

    template<>
    struct size_<std::string>
    {
        size_() : m_size(0) {}
        size_(const std::string& rS_p) : m_size((uint32_t)rS_p.size()) {}
        uint32_t size() const { return m_size; }
        uint32_t m_size;
    };

    /**
    *  @brief Converter of fixed size data elements into a byte stream representation.
    */

    template<class T>
    class tostream
    {
    public:
        tostream(const T& rVal_p) :
            m_cptr((const char*)&rVal_p)
        {
        }

        tostream(const T& rVal_p, uint32_t /*size_p*/) :
            m_cptr((const char*)&rVal_p)
        {
        }

        const char* get() const
        {
            return m_cptr;
        }

        uint32_t size() const
        {
            return sizeof(T);
        }

    private:
        const char* m_cptr;
    };

    /**
    *  @brief Converter of byte stream into fixed data item representation.
    */

    template<class T>
    class fromstream
    {
    public:
        fromstream()
        {
        }

        fromstream(size_<std::string> size_p)
        {
        }

        fromstream(database* /*pDb_p*/, ReadInfo* /*pReadInfo_p*/)
        {
        }

        std::unique_ptr<T> get(database* pDb_p, ItemInfo* pItemInfo_p, uint64_t Package_p) const
        {
            std::unique_ptr<T> pT(new T());
            pDb_p->Recover(sizeof(T), (char*)pT.get(), *pItemInfo_p, Package_p);
            return pT;
        }

        std::unique_ptr<T> get(database* pDb_p, ItemInfo* pItemInfo_p, uint64_t Package_p, size_<T> /*Size_p*/) const
        {
            return get(pDb_p, pItemInfo_p, Package_p);
        }

        std::unique_ptr<T> get(database* pDb_p, ReadInfo* pReadInfo_p) const
        {
            std::unique_ptr<T> pT(new T());
            pDb_p->Recover(sizeof(T), (char*)pT.get(), *pReadInfo_p);
            return pT;
        }

        std::unique_ptr<std::string> get() const
        {
            return std::unique_ptr<std::string>(new std::string());
        }

        uint32_t size() const
        {
            return sizeof(T);
        }
    };

    /**
    *  @brief Converter of variable length data items into byte stream representation.
    */

    template<>
    class tostream<std::string>
    {
    public:
        tostream(const std::string& rVal_p) :
            m_size(sizeof(uint32_t) + (uint32_t)rVal_p.size()),
            m_cptr((char*)malloc(m_size)),
            m_alloc(true)
        {
            *((uint32_t*)m_cptr) = (uint32_t)rVal_p.size();
            memcpy((void*)(m_cptr + sizeof(uint32_t)), (void*)rVal_p.c_str(), rVal_p.size());
        }

        tostream(const std::string& rVal_p, size_<std::string> size_p) :
            m_size(size_p.size()),
            m_cptr(rVal_p.c_str()),
            m_alloc(false)
        {
        }

        ~tostream()
        {
            if (m_alloc)
                free((void*)m_cptr);
        }

        const char* get() const
        {
            return m_cptr;
        }

        uint32_t size() const
        {
            return m_size;
        }

    private:
        const char* m_cptr;
        uint32_t m_size;
        bool m_alloc;
    };

    /**
    *  @brief Converter of byte stream into variable length data item representation.
    */

    template<>
    class fromstream<std::string>
    {
    public:
        fromstream() :
            m_size()
        {
        }

        fromstream(size_<std::string> size_p) :
            m_size(size_p.size())
        {
        }

        fromstream(database* pDb_p, ReadInfo* pReadInfo_p) :
            m_size()
        {
            pDb_p->Recover(sizeof(m_size), (char*)&m_size, *pReadInfo_p);
        }

        std::unique_ptr<std::string> get(database* pDb_p, ItemInfo* pItemInfo_p, uint64_t Package_p) const
        {
            uint32_t size_ = m_size + sizeof(m_size);

            std::vector<char> cbuf(size_);
            pDb_p->Recover(size_, (char*)&cbuf[0], *pItemInfo_p, Package_p);

            std::unique_ptr<std::string> pstr(new std::string());
            (*pstr).insert((*pstr).begin(), &cbuf[sizeof(m_size)], &cbuf[size_]);

            return pstr;
        }

        std::unique_ptr<std::string> get(database* pDb_p, ItemInfo* pItemInfo_p, uint64_t Package_p, size_<std::string> Size_p) const
        {
            std::unique_ptr<std::string> pstr(new std::string());
            pstr->resize(Size_p.size());
            pDb_p->Recover(Size_p.size(), (char*)pstr->c_str(), *pItemInfo_p, Package_p);
            return pstr;
        }

        std::unique_ptr<std::string> get(database* pDb_p, ReadInfo* pReadInfo_p) const
        {
            std::unique_ptr<std::string> pstr(new std::string());
            pstr->resize(m_size);
            pDb_p->Recover(m_size, (char*)pstr->c_str(), *pReadInfo_p);
            return pstr;
        }

        std::unique_ptr<std::string> get() const
        {
            return std::unique_ptr<std::string>(new std::string());
        }

        uint32_t size() const
        {
            return m_size;
        }

    private:
        uint32_t m_size;
    };

    template<class T>
    class ritem;

    template<class T>
    class sitem;

    typedef std::pair<ItemInfo, std::atomic<uint8_t>> ItemHandle;

    /**
    *  @brief Pool datastructure for keeping track and recycling open ItemHandles.
    */

    template<class T>
    class pool
    {
    public:
        pool(database& rDb_p, uint32_t ItemLimit_p = std::numeric_limits<uint32_t>::max()) :
            m_DeleteLock(),
            m_DeletedItems(),
            m_rDb(rDb_p),
            m_ItemIndex(0),
            m_ItemLimit(ItemLimit_p),
            m_ThreadCount((uint32_t)rDb_p.GetThreadCount()),
            m_Pidx(new std::pair<uint64_t, uint32_t>[m_ThreadCount])
        {
            for (uint32_t idx(0); idx < m_ThreadCount; ++idx)
            {
                m_Pidx[idx] = std::make_pair(m_rDb.NewPackage(), 0);
            }
        }

        ~pool()
        {
            std::lock_guard<std::mutex> deleteLock(m_DeleteLock);
            do
            {
                for (auto iter(m_DeletedItems.begin()); iter != m_DeletedItems.end(); )
                {
                    if ((*iter).first->first.Unregistered())
                    {
                        delete (*iter).first;
                        iter = m_DeletedItems.erase(iter);
                        continue;
                    }
                    ++iter;
                }

                if (!m_DeletedItems.empty())
                {
                    m_rDb.Synchronize(m_DeletedItems.front().second);
                }
            } while (m_DeletedItems.empty() == false);

            delete[] m_Pidx;
        }

    private:
        friend class ritem<T>;
        friend class sitem<T>;

        const std::pair<uint64_t, uint32_t>&
            Store(uint32_t Size_p, const char* pData_p, ItemInfo* pItemInfo_p = 0)
        {
            std::pair<uint64_t, uint32_t>& rPidx(m_Pidx[increment()]);
            m_rDb.Store(rPidx.first, Size_p, pData_p, pItemInfo_p);
            return rPidx;
        }

        ItemHandle* Alloc()
        {
            ItemHandle* pItem = 0;

            std::lock_guard<std::mutex> deleteLock(m_DeleteLock);

            auto iter = m_DeletedItems.begin();
            if (iter != m_DeletedItems.end())
            {
                if ((*iter).first->first.m_PackageSize != std::numeric_limits<uint32_t>::max())
                {
                    pItem = (*iter).first;
                    iter = m_DeletedItems.erase(iter);
                }
            }

            return pItem != 0 ? pItem : new ItemHandle();
        }

        void Delete(ItemHandle* pItemInfo_p, uint64_t PackageID_p)
        {
            std::lock_guard<std::mutex> deleteLock(m_DeleteLock);
            if (pItemInfo_p != 0)
            {
                if (pItemInfo_p->first.m_PackageSize != std::numeric_limits<uint32_t>::max())
                    m_DeletedItems.push_front(std::make_pair(pItemInfo_p, PackageID_p));
                else
                    m_DeletedItems.push_back(std::make_pair(pItemInfo_p, PackageID_p));
            }
        }

        database& Db() { return m_rDb; }

        uint32_t increment()
        {
            ++m_ItemIndex;

            const uint32_t packageIdx = m_ItemIndex % m_ThreadCount;
            std::pair<uint64_t, uint32_t>& rPackageIdx(m_Pidx[packageIdx]);

            ++(rPackageIdx.second);
            if (rPackageIdx.second > m_ItemLimit)
            {
                rPackageIdx = std::make_pair(m_rDb.NewPackage(), 1);
            }

            return packageIdx;
        }

        std::mutex m_DeleteLock;
        std::deque< std::pair<ItemHandle*, uint64_t> > m_DeletedItems;

        database& m_rDb;
        uint32_t m_ItemIndex;
        uint32_t m_ItemLimit;
        uint32_t m_ThreadCount;
        std::pair<uint64_t, uint32_t>* m_Pidx;
    };

    /**
    *  @brief Wrapper class for storing data items in a random fashion.
    */

    template<class T>
    class ritem
    {
    public:
        typedef T value_type;

        ritem() :
            m_T(),
            m_P(),
            m_Pidx(),
            m_Size()
        {
            m_Pidx = 0;
        }

        ritem(const T& rVal_p, pool<T>& rPool_p) :
            m_T(rPool_p.Alloc()),
            m_P(&rPool_p),
            m_Pidx(),
            m_Size(size_<T>(rVal_p))
        {
            ++m_T->second;
            const tostream<T> byteStream(rVal_p, m_Size);
            m_Pidx = m_P->Store(byteStream.size(), byteStream.get(), &m_T->first).first;
        }

        ritem(const ritem& rhs_p) :
            m_T(rhs_p.m_T),
            m_P(rhs_p.m_P),
            m_Pidx(rhs_p.m_Pidx),
            m_Size(rhs_p.m_Size)
        {
            ++m_T->second;
        }

        void operator=(const ritem &rhs_p)
        {
            deref();

            m_T = rhs_p.m_T;
            ++m_T->second;
            m_P = rhs_p.m_P;
            m_Pidx = rhs_p.m_Pidx;
        }

        ~ritem()
        {
            deref();
        }

        void deref()
        {
            --m_T->second;
            if (m_T->second == 0)
            {
                m_P->Delete(m_T, m_Pidx);
                m_T = 0;
            }
        }

        std::unique_ptr<T> get() const
        {
            return fromstream<T>(m_Size).get(&m_P->Db(), &m_T->first, m_Pidx, m_Size);
        }

    protected:
        ItemHandle* m_T;

    private:
        pool<T>* m_P;
        uint64_t m_Pidx;
        size_<T> m_Size;
    };

    /**
    *  @brief Wrapper class for storing data items in a sequential fashion.
    */

    template<class T>
    class sitem
    {
    public:
        typedef T value_type;

        sitem() :
            m_P(),
            m_Pidx()
        {
        }

        sitem(const T& rVal_p, pool<T>& rPool_p) :
            m_P(&rPool_p),
            m_Pidx()
        {
            const tostream<T> byteStream(rVal_p);
            m_Pidx = m_P->Store(byteStream.size(), byteStream.get());
        }

        void operator=(const sitem &rhs_p)
        {
            m_P = rhs_p.m_P;
            m_Pidx = rhs_p.m_Pidx;
        }

        std::unique_ptr<T> get() const
        {
            if (!m_P) fromstream<T>().get();

            m_P->Db().Synchronize(m_Pidx.first);

            ReadInfo readInfo = m_P->Db().Open(m_Pidx.first);
            for (uint32_t idx(1); idx < m_Pidx.second; ++idx)
            {
                fromstream<T> fromStream(&m_P->Db(), &readInfo);
                const uint32_t bytesToRead = fromStream.size();
                if (m_P->Db().Recover(bytesToRead, 0, readInfo) != bytesToRead)
                    throw std::runtime_error("Invalid read operation in item data recovery.");
            }
            return fromstream<T>(&m_P->Db(), &readInfo).get(&m_P->Db(), &readInfo);
        }

    private:
        pool<T>* m_P;
        std::pair<uint64_t, uint32_t> m_Pidx;
    };

    /**
    *  @brief Wrapper class for storing key data items.
    */

    template<class T>
    class key : public T
    {
    public:
        key(const typename T::value_type& rVal_p, pool<typename T::value_type>& rPool_p) :
            T(rVal_p, rPool_p)
        {
        }

        ~key()
        {
        }

        bool operator<(const T& rhs_p) const
        {
            return (*this->get()) < (*rhs_p.get());
        }

        bool operator>(const T& rhs_p) const
        {
            return ((*rhs_p.get() < *this->get()));
        }
    };
}
