[WIP] Mock store

code/go/0chain.net/blobber/main.go

@@ -143,7 +143,7 @@ func checkForDBConnection() {
 	retries := 0
 	var err error
 	for retries < 600 {
-		err = datastore.GetStore().Open()
+		err = datastore.TheStore.Open()
 		if err != nil {
 			time.Sleep(1 * time.Second)
 			retries++

code/go/0chain.net/blobbercore/allocation/allocationchange.go

@@ -69,14 +69,14 @@ func (AllocationChange) TableName() string {
 
 func GetAllocationChanges(ctx context.Context, connectionID string, allocationID string, clientID string) (*AllocationChangeCollector, error) {
 	cc := &AllocationChangeCollector{}
-	db := datastore.GetStore().GetTransaction(ctx)
+	db := datastore.GetTransaction(ctx)
 	err := db.Where(&AllocationChangeCollector{
 		ConnectionID: connectionID,
 		AllocationID: allocationID,
 		ClientID:     clientID,
 	}).Not(&AllocationChangeCollector{
 		Status: DeletedConnection,
-	}).Preload("Changes").First(cc).Error
+	}).Preload("Changes").First(cc).Error()
 
 	if err == nil {
 		cc.ComputeProperties()
@@ -101,13 +101,13 @@ func (cc *AllocationChangeCollector) AddChange(allocationChange *AllocationChang
 
 func (cc *AllocationChangeCollector) Save(ctx context.Context) error {
 
-	db := datastore.GetStore().GetTransaction(ctx)
+	db := datastore.GetTransaction(ctx)
 	if cc.Status == NewConnection {
 		cc.Status = InProgressConnection
-		err := db.Create(cc).Error
+		err := db.Create(cc).Error()
 		return err
 	} else {
-		err := db.Save(cc).Error
+		err := db.Save(cc).Error()
 		return err
 	}
 }

code/go/0chain.net/blobbercore/allocation/deletefilechange.go

@@ -5,10 +5,10 @@ import (
 	"encoding/json"
 	"path/filepath"
 
-	"0chain.net/blobbercore/datastore"
-	"0chain.net/blobbercore/filestore"
 	"0chain.net/blobbercore/reference"
 	"0chain.net/core/common"
+	"0chain.net/blobbercore/datastore"
+	"0chain.net/blobbercore/filestore"
 	. "0chain.net/core/logging"
 
 	"go.uber.org/zap"
@@ -118,7 +118,7 @@ func (nf *DeleteFileChange) DeleteTempFile() error {
 }
 
 func (nf *DeleteFileChange) CommitToFileStore(ctx context.Context) error {
-	db := datastore.GetStore().GetTransaction(ctx)
+	db := datastore.GetTransaction(ctx)
 	for contenthash := range nf.ContentHash {
 		var count int64
 		err := db.Table((&reference.Ref{}).TableName()).Where(&reference.Ref{ThumbnailHash: contenthash}).Or(&reference.Ref{ContentHash: contenthash}).Count(&count).Error

code/go/0chain.net/blobbercore/allocation/entity.go

@@ -1,6 +1,7 @@
 package allocation
 
 import (
+	"0chain.net/blobbercore/datastore"
 	"errors"
 	"time"
 
@@ -110,7 +111,7 @@ func (a *Allocation) WantWrite(blobberID string, size int64,
 }
 
 // ReadPools from DB cache.
-func ReadPools(tx *gorm.DB, clientID, allocID, blobberID string,
+func ReadPools(tx datastore.Transaction, clientID, allocID, blobberID string,
 	until common.Timestamp) (rps []*ReadPool, err error) {
 
 	const query = `client_id = ? AND
@@ -120,7 +121,7 @@ func ReadPools(tx *gorm.DB, clientID, allocID, blobberID string,
 
 	err = tx.Model(&ReadPool{}).
 		Where(query, clientID, allocID, blobberID, until).
-		Find(&rps).Error
+		Find(&rps).Error()
 	return
 }
 
@@ -149,7 +150,7 @@ func (*Pending) TableName() string {
 	return "pendings"
 }
 
-func GetPending(tx *gorm.DB, clientID, allocationID, blobberID string) (
+func GetPending(tx datastore.Transaction, clientID, allocationID, blobberID string) (
 	p *Pending, err error) {
 
 	const query = `client_id = ? AND
@@ -159,12 +160,12 @@ func GetPending(tx *gorm.DB, clientID, allocationID, blobberID string) (
 	p = new(Pending)
 	err = tx.Model(&Pending{}).
 		Where(query, clientID, allocationID, blobberID).
-		First(&p).Error
+		First(&p).Error()
 	if errors.Is(err, gorm.ErrRecordNotFound) {
 		p.ClientID = clientID
 		p.AllocationID = allocationID
 		p.BlobberID = blobberID
-		err = tx.Create(p).Error
+		err = tx.Create(p).Error()
 	}
 	return
 }
@@ -179,7 +180,7 @@ func (p *Pending) SubPendingWrite(size int64) {
 	}
 }
 
-func (p *Pending) WritePools(tx *gorm.DB, blobberID string,
+func (p *Pending) WritePools(tx datastore.Transaction, blobberID string,
 	until common.Timestamp) (wps []*WritePool, err error) {
 
 	const query = `client_id = ? AND
@@ -189,7 +190,7 @@ func (p *Pending) WritePools(tx *gorm.DB, blobberID string,
 
 	err = tx.Model(&WritePool{}).
 		Where(query, p.ClientID, p.AllocationID, blobberID, until).
-		Find(&wps).Error
+		Find(&wps).Error()
 	return
 }
 
@@ -202,11 +203,11 @@ func (p *Pending) HaveWrite(wps []*WritePool, ww WantWriter,
 	return have - ww.WantWrite(p.BlobberID, p.PendingWrite, wmt)
 }
 
-func (p *Pending) Save(tx *gorm.DB) error {
+func (p *Pending) Save(tx datastore.Transaction) error {
 	if p.ID == 0 {
-		return tx.Create(p).Error
+		return tx.Create(p).Error()
 	}
-	return tx.Save(p).Error
+	return tx.Save(p).Error()
 }
 
 // Terms for allocation by its Tx.
@@ -253,7 +254,7 @@ func (*WritePool) TableName() string {
 	return "write_pools"
 }
 
-func SetReadPools(db *gorm.DB, clientID, allocationID, blobberID string,
+func SetReadPools(db datastore.Transaction, clientID, allocationID, blobberID string,
 	rps []*ReadPool) (err error) {
 
 	// cleanup and batch insert (remove old pools, add / update new)
@@ -265,7 +266,7 @@ func SetReadPools(db *gorm.DB, clientID, allocationID, blobberID string,
 	var stub []*ReadPool
 	err = db.Model(&ReadPool{}).
 		Where(query, clientID, allocationID, blobberID).
-		Delete(&stub).Error
+		Delete(&stub).Error()
 	if err != nil {
 		return
 	}
@@ -277,11 +278,11 @@ func SetReadPools(db *gorm.DB, clientID, allocationID, blobberID string,
 	err = db.Model(&ReadPool{}).Clauses(clause.OnConflict{
 		Columns:   []clause.Column{{Name: "pool_id"}},
 		DoUpdates: clause.AssignmentColumns([]string{"balance"}),
-	}).Create(rps).Error
+	}).Create(rps).Error()
 	return
 }
 
-func SetWritePools(db *gorm.DB, clientID, allocationID, blobberID string,
+func SetWritePools(db datastore.Transaction, clientID, allocationID, blobberID string,
 	wps []*WritePool) (err error) {
 
 	const query = `client_id = ? AND
@@ -291,7 +292,7 @@ func SetWritePools(db *gorm.DB, clientID, allocationID, blobberID string,
 	var stub []*WritePool
 	err = db.Model(&WritePool{}).
 		Where(query, clientID, allocationID, blobberID).
-		Delete(&stub).Error
+		Delete(&stub).Error()
 	if err != nil {
 		return
 	}
@@ -303,7 +304,7 @@ func SetWritePools(db *gorm.DB, clientID, allocationID, blobberID string,
 	err = db.Model(&WritePool{}).Clauses(clause.OnConflict{
 		Columns:   []clause.Column{{Name: "pool_id"}},
 		DoUpdates: clause.AssignmentColumns([]string{"balance"}),
-	}).Create(wps).Error
+	}).Create(wps).Error()
 	return
 }
 

code/go/0chain.net/blobbercore/allocation/protocol.go

@@ -20,12 +20,12 @@ import (
 func GetAllocationByID(ctx context.Context, allocID string) (
 	a *Allocation, err error) {
 
-	var tx = datastore.GetStore().GetTransaction(ctx)
+	var tx = datastore.GetTransaction(ctx)
 
 	a = new(Allocation)
 	err = tx.Model(&Allocation{}).
 		Where(&Allocation{ID: allocID}).
-		First(a).Error
+		First(a).Error()
 	return
 }
 
@@ -34,12 +34,12 @@ func GetAllocationByID(ctx context.Context, allocID string) (
 // loads the Terms for an allocation.
 func (a *Allocation) LoadTerms(ctx context.Context) (err error) {
 	// get transaction
-	var tx = datastore.GetStore().GetTransaction(ctx)
+	var tx = datastore.GetTransaction(ctx)
 	// load related terms
 	var terms []*Terms
 	err = tx.Model(terms).
 		Where("allocation_id = ?", a.ID).
-		Find(&terms).Error
+		Find(&terms).Error()
 	if err != nil {
 		// unexpected DB error, including a RecordNotFoundError, since
 		// an allocation can't be without its terms (the terms must exist)
@@ -52,12 +52,12 @@ func (a *Allocation) LoadTerms(ctx context.Context) (err error) {
 func VerifyAllocationTransaction(ctx context.Context, allocationTx string,
 	readonly bool) (a *Allocation, err error) {
 
-	var tx = datastore.GetStore().GetTransaction(ctx)
+	var tx = datastore.GetTransaction(ctx)
 
 	a = new(Allocation)
 	err = tx.Model(&Allocation{}).
 		Where(&Allocation{Tx: allocationTx}).
-		First(a).Error
+		First(a).Error()
 
 	if err != nil && !errors.Is(err, gorm.ErrRecordNotFound) {
 		return nil, err // unexpected DB error
@@ -68,7 +68,7 @@ func VerifyAllocationTransaction(ctx context.Context, allocationTx string,
 		var terms []*Terms
 		err = tx.Model(terms).
 			Where("allocation_id = ?", a.ID).
-			Find(&terms).Error
+			Find(&terms).Error()
 		if err != nil {
 			return // unexpected DB error
 		}
@@ -91,7 +91,7 @@ func VerifyAllocationTransaction(ctx context.Context, allocationTx string,
 	var isExist bool
 	err = tx.Model(&Allocation{}).
 		Where("id = ?", sa.ID).
-		First(a).Error
+		First(a).Error()
 	if err != nil && !errors.Is(err, gorm.ErrRecordNotFound) {
 		return nil, err // unexpected
 	}
@@ -146,9 +146,9 @@ func VerifyAllocationTransaction(ctx context.Context, allocationTx string,
 	Logger.Info("Saving the allocation to DB")
 
 	if isExist {
-		err = tx.Save(a).Error
+		err = tx.Save(a).Error()
 	} else {
-		err = tx.Create(a).Error
+		err = tx.Create(a).Error()
 	}
 
 	if err != nil {
@@ -158,9 +158,9 @@ func VerifyAllocationTransaction(ctx context.Context, allocationTx string,
 	// save/update related terms
 	for _, t := range a.Terms {
 		if isExist {
-			err = tx.Save(t).Error
+			err = tx.Save(t).Error()
 		} else {
-			err = tx.Create(t).Error
+			err = tx.Create(t).Error()
 		}
 		if err != nil {
 			return nil, err