/*
   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&) {}
        size_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(rS_p.size()) {}
        size_t size() const { return m_size; }
        size_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, size_t /*size_p*/) :
            m_cptr((const char*)&rVal_p)
        {
        }

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

        size_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());
        }

        size_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(size_t) + rVal_p.size()),
            m_cptr((char*)malloc(m_size)),
            m_alloc(true)
        {
            *((size_t*)m_cptr) = rVal_p.size();
            memcpy((void*)(m_cptr + sizeof(size_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;
        }

        size_t size() const
        {
            return m_size;
        }

    private:
        const char* m_cptr;
        size_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
        {
            size_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());
        }

        size_t size() const
        {
            return m_size;
        }

    private:
        size_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(rDb_p.GetThreadCount()),
            m_Pidx(new std::pair<uint64_t, uint32_t>[m_ThreadCount])
        {
            for (size_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(size_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;
        size_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 (size_t idx(1); idx < m_Pidx.second; ++idx)
            {
                fromstream<T> fromStream(&m_P->Db(), &readInfo);
                const size_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()));
        }
    };
}
