StoreTable.hpp

Go to the documentation of this file.

#ifndef _IDENTT_STORE_STORE_TABLE_HPP_
#define _IDENTT_STORE_STORE_TABLE_HPP_

#include <set>
#include <google/protobuf/reflection.h>
#include <google/protobuf/map.h>
#include "StoreLevel.hpp"

namespace identt {
namespace store {
template <class T>
class StoreTable : public StoreLevel {
public:
    using StoreLevel::dbpointer;
    using KeySetT=std::set<KeyTypeE>;
    using MapT=google::protobuf::Map<uint64_t,T>;

    StoreTable(dbpointer trydb, const KeyTypeE pkey, const KeySetT ukeys, const KeySetT ikeys)
        :
        StoreLevel(trydb),
        PrimaryKey(pkey),
        unique_keys(ukeys),
        index_keys(ikeys)
    {}

    virtual ~StoreTable() {}

    StoreTable() = delete;
    StoreTable(const StoreTable&) = delete;
    StoreTable& operator=(const StoreTable&) = delete;
    // StoreTable(const std::string, const size_t) = delete;
    // void Initialize(const std::string , const size_t , uint64_t& ,uint64_t& )=delete;

    bool GetOne(T* record, KeyTypeE keytype)
    {
        bool key_found=false;
        std::string key,value;
        usemydb::Status s ;
        // primary key is direct
        if (keytype==PrimaryKey) {
            key = GetKey(record,PrimaryKey,false);
            key_found=true;
            // check if unique key or index key
        } else if ( unique_keys.find(keytype)!=unique_keys.end() || index_keys.find(keytype)!=index_keys.end()) {
            s = getDB()->Get(usemydb::ReadOptions(), GetKey(record,keytype,false), &value);
            if (s.ok()) {
                NodeT node;
                if ( node.ParseFromString(value) ) {
                    record->set_id( node.id() );
                    key = GetKey(record,PrimaryKey,false);
                    key_found=true;
                }
            }
        } else
            throw identt::BadDataException("Keytype invalid");

        if (!key_found) {
            return false;
        }
        s = getDB()->Get(usemydb::ReadOptions(), key, &value);
        if (!s.ok()) {
            return false;
        }

        if (!record->ParseFromString(value))
            throw identt::BadDataException("Record cannot be parsed");

        return true;
    }


    bool GetMany(T* refer, MapT* records, KeyTypeE keytype, size_t skip, size_t limit)
    {
        if (keytype==PrimaryKey) {
            throw identt::BadCodeException("This cannot be used for primary key");
        }
        NodeListT nodes;
        bool key_found=GetKeysFromIndex(keytype,refer, &nodes,skip,limit);
        if (!key_found) return false;
        std::string key,value;
        usemydb::Status s ;
        T record;

        for (size_t i=0; i<nodes.node_size(); ++i) {
            if (nodes.mutable_node(i)->keytype()!=PrimaryKey)
                throw identt::BadDataException("Record is bad ");
            record.set_id( nodes.mutable_node(i)->id() );
            key = GetKey(&record,PrimaryKey,false);
            s = getDB()->Get(usemydb::ReadOptions(), key, &value);
            if (!s.ok())
                throw identt::BadDataException("Record is missing ");
            if (!record.ParseFromString(value))
                throw identt::BadDataException("Record cannot be parsed");
            (*records)[record.id()]=record;
        }

        return true;
    }

    void AddTrans(TransactionT* trans, std::string key, std::string& value, bool to_del)
    {
        TransItemT* item = trans->add_item();
        item->set_key(key);
        if (!to_del) item->set_value(value);
        item->set_to_del(to_del);
    }


    virtual std::string GetKey(T* record, KeyTypeE keytype, bool pre)=0;

    bool GetSecondaryValue(T* record, KeyTypeE keytype, NodeT* node)
    {
        if ( unique_keys.find(keytype)==unique_keys.end() && index_keys.find(keytype)==index_keys.end() )
            return false;
        std::string key = GetKey(record,keytype,false);
        std::string value;
        usemydb::Status s = getDB()->Get(usemydb::ReadOptions(), key, &value);
        if (!s.ok()) return false;
        return node->ParseFromString(value);
    }

    bool AddRecord(T* record, TransactionT* trans, bool upsert)
    {

        if (record->notfound()) return false; // bad data
        if (record->id()==0) return false; // bad data

        T old;
        old.set_id(record->id());
        bool old_record_exists = GetOne(&old,PrimaryKey);
        if (old_record_exists && !upsert) return false; // old data no upsert

        NodeT node;
        bool status;

        // check unique key for clash
        for (auto& keytype : unique_keys) {
            status = GetSecondaryValue(record,keytype,&node);
            if (status && ( record->id() != node.id()) ) return false; // unique key clash
        }

        // node record for dump
        std::string dumpid;
        node.set_id(record->id());
        node.set_keytype(PrimaryKey);
        node.SerializeToString(&dumpid);

        // check unique keys , delete old and update new , keep if same
        for (auto& keytype : unique_keys) {
            std::string u_new = GetKey(record,keytype,false);
            if (old_record_exists) {
                std::string u_old = GetKey(&old,keytype,false);
                if (u_old!=u_new) {
                    AddTrans(trans,u_old,dumpid,true); // dumpid ignored
                    AddTrans(trans,u_new,dumpid,false);
                }
            } else {
                AddTrans(trans,u_new,dumpid,false);
            }
        }

        // check index keys , delete old and update new , keep if same
        for (auto& keytype : index_keys) {
            std::string u_new = GetKey(record,keytype,false);
            if (old_record_exists) {
                std::string u_old = GetKey(&old,keytype,false);
                if (u_old!=u_new) {
                    AddTrans(trans,u_old,dumpid,true); // dumpid ignored
                    AddTrans(trans,u_new,dumpid,false);
                }
            } else {
                AddTrans(trans,u_new,dumpid,false);
            }
        }

        // add the record
        record->SerializeToString(&dumpid);
        std::string pkey = GetKey(record,PrimaryKey,false);
        AddTrans(trans,pkey,dumpid,false);

        return true;
    }

    bool DelRecord(T* record, TransactionT* trans)
    {
        if (record->notfound()) return false; // bad data

        // insert record for deletion
        std::string value;
        AddTrans(trans,EncodePrimaryKey(PrimaryKey,record->id()),value,true);
        for (auto& keytype : unique_keys) {
            AddTrans(trans,GetKey(record,keytype,false),value,true);
        }
        for (auto& keytype : index_keys) {
            AddTrans(trans,GetKey(record,keytype,false),value,true);
        }
        return true;
    }

    bool GetKeysFromIndex(KeyTypeE keytype, T* record, NodeListT* nodes,size_t skip, size_t limit)
    {
        if ( unique_keys.find(keytype)==unique_keys.end() && index_keys.find(keytype)==index_keys.end() )
            return false;
        std::string key = GetKey(record,keytype,true);
        if (key.length() <= IDENTT_KEYID_LEN ) return false;
        std::shared_ptr<usemydb::Iterator> it(getDB()->NewIterator(usemydb::ReadOptions()));
        usemydb::Slice start = usemydb::Slice ( key );
        usemydb::Slice match = usemydb::Slice ( key );
        uint64_t counter=0;
        bool at_least_one=false;
        for (it->Seek(start); it->Valid() && it->key().starts_with(match) ; it->Next()) {
            if (++counter<=skip) continue; // ignore skip
            NodeT* node = nodes->add_node();
            node->ParseFromString(it->value().ToString());
            at_least_one=true;
            if (counter==skip+limit) break; // to handle the id==0 this comes here
        }
        return at_least_one;
    }

    void ScanTable(uint64_t startid, size_t limit, std::function<void(T*)> Func)
    {
        std::shared_ptr<usemydb::Iterator> it(getDB()->NewIterator(usemydb::ReadOptions()));
        std::string keymatch = EncodeKeyType(PrimaryKey);
        usemydb::Slice match = usemydb::Slice ( keymatch );
        startid = (startid>0) ? startid : 1;
        std::string keystart = EncodePrimaryKey(PrimaryKey,startid);
        usemydb::Slice start = usemydb::Slice ( keystart );

        uint64_t counter=0;
        for (it->Seek(start); it->Valid() && it->key().starts_with(match) ; it->Next()) {
            T record;
            if (!record.ParseFromString(it->value().ToString()) )
                throw identt::BadDataException("Record cannot be parsed");
            Func(&record);
            if (++counter==limit) break;
        }
    }

protected:
    const KeyTypeE PrimaryKey;
    const KeySetT unique_keys;
    const KeySetT index_keys;


};
} // namespace store
} // namespace identt
#endif /* _IDENTT_STORE_STORE_TABLE_HPP_ */