func (s *Store) GetValuesWrittenByUser(userID string) ([]*types.KVWithMetadata, error) {
	s.mutex.RLock()
	defer s.mutex.RUnlock()
	txIDs, err := s.GetTxIDsSubmittedByUser(userID)
	if err != nil {
		return nil, err
	}
	return s.outEdgesFrom(txIDs, WRITES)
}
GetValuesWrittenByUser returns all values written by a given user
package provenance
import (
	"context"
	"encoding/json"
	"fmt"
	"sort"
	"strings"
	interrors "github.com/hyperledger-labs/orion-server/internal/errors"
	"github.com/hyperledger-labs/orion-server/pkg/types"
	"github.com/cayleygraph/cayley"
	"github.com/cayleygraph/cayley/graph"
	"github.com/cayleygraph/quad"
	"github.com/pkg/errors"
)
const (
	SUBMITTED = "s"
	INCLUDES = "i"
	WRITES = "w"
	READS = "r"
	DELETES = "d"
	NEXT = "n"
	PREVIOUS = "p"
)
type TxDataForProvenance struct {
	IsValid            bool
	DBName             string
	UserID             string
	TxID               string
	Reads              []*KeyWithVersion
	Writes             []*types.KVWithMetadata
	Deletes            map[string]*types.Version
	OldVersionOfWrites map[string]*types.Version
}
type KeyWithVersion struct {
	Key     string
	Version *types.Version
}
type TxIDLocation struct {
	BlockNum uint64 `json:"block_num"`
	TxIndex  int    `json:"tx_index"`
}
func (s *Store) Commit(blockNum uint64, txsData []*TxDataForProvenance) error {
	s.mutex.Lock()
	defer s.mutex.Unlock()
	batch := graph.NewWriter(s.cayleyGraph.QuadWriter)
	for txNum, tx := range txsData {
		loc, err := json.Marshal(&TxIDLocation{blockNum, txNum})
		if err != nil {
			return errors.WithMessage(err, "error while marshaling txID location")
		}
		s.logger.Debugf("loc[%s]]---(includes)--->txID[%s]", loc, tx.TxID)
		batch.WriteQuad(quad.Make(string(loc), INCLUDES, tx.TxID, ""))
		if !tx.IsValid {
			s.logger.Debugf("as txID [%s] is invalid, we created vertex and edge to represent only the relation [location--(includes)-->txID]", tx.TxID)
			continue
		}
		s.logger.Debugf("userID[%s]---(submitted)--->txID[%s]", tx.UserID, tx.TxID)
		batch.WriteQuad(quad.Make(tx.UserID, SUBMITTED, tx.TxID, ""))
		if err := s.addReads(tx, batch); err != nil {
			return err
		}
		if err := s.addWrites(tx, batch); err != nil {
			return err
		}
		if err := s.addDeletes(tx, batch); err != nil {
			return err
		}
	}
	return batch.Close()
}
func (s *Store) addReads(tx *TxDataForProvenance, batch graph.BatchWriter) error {
	for _, read := range tx.Reads {
		value, err := s.getValueVertex(tx.DBName, read.Key, read.Version)
		if err != nil {
			return err
		}
		s.logger.Debugf("txID[%s]---(reads)--->value[%s]", tx.TxID, quad.NativeOf(value))
		batch.WriteQuad(quad.Make(tx.TxID, READS, value, ""))
	}
	return nil
}
func (s *Store) addWrites(tx *TxDataForProvenance, batch graph.BatchWriter) error {
	for _, write := range tx.Writes {
		actualKey := write.Key
		write.Key = constructCompositeKey(tx.DBName, write.Key)
		newValue, err := json.Marshal(write)
		if err != nil {
			return err
		}
		newVersion, err := json.Marshal(write.Metadata.Version)
		if err != nil {
			return err
		}
		s.logger.Debugf("key[%s]---(version[%s])--->value[%s]", write.Key, string(newVersion), string(newValue))
		batch.WriteQuad(quad.Make(write.Key, string(newVersion), string(newValue), ""))
		s.logger.Debugf("txID[%s]---(writes)--->value[%s]", tx.TxID, string(newValue))
		batch.WriteQuad(quad.Make(tx.TxID, WRITES, string(newValue), ""))
		oldVersion, ok := tx.OldVersionOfWrites[actualKey]
		if !ok {
			s.logger.Debug("fetching last deleted version of key [" + actualKey + "] from db [" + tx.DBName + "]")
			lastVer, err := s.getLastDeletedVersion(tx.DBName, write.Key)
			if err != nil {
				return err
			}
			if lastVer == nil {
				s.logger.Debug("previous version of key [" + actualKey + "] does not exist in db [" + tx.DBName + "]")
				continue
			}
			oldVersion = lastVer
		}
		oldValue, err := s.getValueVertex(tx.DBName, actualKey, oldVersion)
		if err != nil {
			return err
		}
		if oldValue == nil {
			s.logger.Debugf("key [%s] version [%d,%d] for which oldValue is not found", actualKey, oldVersion.BlockNum, oldVersion.TxNum)
			return errors.Errorf("error while finding the previous version of the key[%s]", write.Key)
		}
		s.logger.Debugf("oldValue[%s]<---(previous)---newValue[%s]", quad.NativeOf(oldValue), string(newValue))
		batch.WriteQuad(quad.Make(string(newValue), PREVIOUS, oldValue, ""))
		s.logger.Debugf("oldValue[%s]---(next)--->newValue[%s]", quad.NativeOf(oldValue), string(newValue))
		batch.WriteQuad(quad.Make(oldValue, NEXT, string(newValue), ""))
	}
	return nil
}
func (s *Store) addDeletes(tx *TxDataForProvenance, batch graph.BatchWriter) error {
	for k, v := range tx.Deletes {
		s.logger.Debugf("fetch value of key [%s] at version (%d, %d)", k, v.BlockNum, v.TxNum)
		value, err := s.getValueVertex(tx.DBName, k, v)
		if err != nil {
			return err
		}
		if value == nil {
			continue
		}
		s.logger.Debugf("txID[%s]---(deletes)--->value[%s]", tx.TxID, quad.NativeOf(value))
		batch.WriteQuad(quad.Make(tx.TxID, DELETES, value, ""))
	}
	return nil
}
func (s *Store) GetValues(dbName, key string) ([]*types.ValueWithMetadata, error) {
	s.mutex.RLock()
	defer s.mutex.RUnlock()
	s.logger.Debugf("fetch all historical values associated with the key [%s] in db [%s]", key, dbName)
	cKey := constructCompositeKey(dbName, key)
	p := cayley.StartPath(s.cayleyGraph, quad.String(cKey)).Out()
	valueVertices, err := p.Iterate(context.Background()).AllValues(s.cayleyGraph)
	if err != nil {
		return nil, err
	}
	return verticesToValues(valueVertices)
}
func (s *Store) GetPreviousValues(dbName, key string, version *types.Version, limit int) ([]*types.ValueWithMetadata, error) {
	s.mutex.RLock()
	defer s.mutex.RUnlock()
	return s.getValuesRecursively(dbName, key, version, PREVIOUS, limit)
}
func (s *Store) GetNextValues(dbName, key string, version *types.Version, limit int) ([]*types.ValueWithMetadata, error) {
	s.mutex.RLock()
	defer s.mutex.RUnlock()
	return s.getValuesRecursively(dbName, key, version, NEXT, limit)
}
func (s *Store) GetValueAt(dbName, key string, version *types.Version) (*types.ValueWithMetadata, error) {
	s.mutex.RLock()
	defer s.mutex.RUnlock()
	s.logger.Debugf("fetch value of key [%s] at version (%d, %d)", key, version.BlockNum, version.TxNum)
	valueVertex, err := s.getValueVertex(dbName, key, version)
	if err != nil {
		return nil, err
	}
	if valueVertex == nil {
		return nil, nil
	}
	return vertexToValue(valueVertex)
}
func (s *Store) GetValuesReadByUser(userID string) ([]*types.KVWithMetadata, error) {
	s.mutex.RLock()
	defer s.mutex.RUnlock()
	txIDs, err := s.GetTxIDsSubmittedByUser(userID)
	if err != nil {
		return nil, err
	}
	return s.outEdgesFrom(txIDs, READS)
}