Repository: trustwallet/blockatlas
Pull request #975 (2020-03-25 18:29:48)

Migration from Redis to Postgres DB

hewigovens: @vikmeup I remember Danilo had tried PG? no idea why it didn't work for him
todo[bot]: ## rewrite

https://github.com/trustwallet/blockatlas/blob/9c879aeaccd37973de9a3683c04b001be33b4751/db/subscriptions.go#L65-L70

---

###### This comment was generated by [todo](https://todo.jasonet.co) based on a `TODO` comment in 9c879aeaccd37973de9a3683c04b001be33b4751 in #975. cc @trustwallet.

Patch:
diff --git a/.github/ISSUE_TEMPLATE/IMPROVEMENT_TEMPLATE b/.github/ISSUE_TEMPLATE/IMPROVEMENT_TEMPLATE
new file mode 100644
index 0000000..f1e0f32
--- /dev/null
+++ b/.github/ISSUE_TEMPLATE/IMPROVEMENT_TEMPLATE
@@ -0,0 +1,21 @@
+---
+name: Enhancement
+about: Improve existing architecture
+title: ''
+labels: 'Type: Enhancement'
+assignees: ''
+
+---
+
+<!-- < < < < < < < < < < < < < < < < < < < < < < < < < < < < < < < < < ☺ 
+v                 ✰  Thanks for opening an issue! ✰                    v
+v    Before smashing the submit button please review the template.     v
+v               Don't forget to set the priority label.                v
+v       Please also ensure that this is not a duplicate issue :)       v  
+☺ > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > >  -->
+
+## Current Functionality
+
+
+## Possible Improvement
+
diff --git a/.github/ISSUE_TEMPLATE/MAINTANCE_TEMPLATE b/.github/ISSUE_TEMPLATE/MAINTANCE_TEMPLATE
new file mode 100644
index 0000000..38e8634
--- /dev/null
+++ b/.github/ISSUE_TEMPLATE/MAINTANCE_TEMPLATE
@@ -0,0 +1,16 @@
+---
+name: Maintance
+about: Deprecate old functionality or routes
+title: ''
+labels: 'Type: Maintance'
+assignees: ''
+
+---
+
+<!-- < < < < < < < < < < < < < < < < < < < < < < < < < < < < < < < < < ☺ 
+v                 ✰  Thanks for opening an issue! ✰                    v
+v    Before smashing the submit button please review the template.     v
+v               Don't forget to set the priority label.                v
+v       Please also ensure that this is not a duplicate issue :)       v  
+☺ > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > >  -->
+
diff --git a/Makefile b/Makefile
index 696a1ff..0c92805 100644
--- a/Makefile
+++ b/Makefile
@@ -248,7 +248,7 @@ go-test:
 
 go-integration:
 	@echo "  >  Running integration tests"
-	GOBIN=$(GOBIN) TEST_CONFIG=$(CONFIG_FILE) go test -race -tags=integration -v ./tests/integration ./tests/docker_test
+	GOBIN=$(GOBIN) TEST_CONFIG=$(CONFIG_FILE) go test -race -tags=integration -v ./tests/integration/...
 
 go-fmt:
 	@echo "  >  Format all go files"
diff --git a/api/observer_healthcheck.go b/api/observer_healthcheck.go
deleted file mode 100644
index dafe734..0000000
--- a/api/observer_healthcheck.go
+++ /dev/null
@@ -1,12 +0,0 @@
-package api
-
-import (
-	"github.com/gin-gonic/gin"
-	"github.com/trustwallet/blockatlas/services/observer/healthcheck"
-	"net/http"
-)
-
-func GetObserverStatus(c *gin.Context) {
-	exclude := splitParam(c.Query("exclude"))
-	c.JSON(http.StatusOK, healthcheck.GetStatus(exclude))
-}
diff --git a/api/routes.go b/api/routes.go
index 8766293..e029d5a 100644
--- a/api/routes.go
+++ b/api/routes.go
@@ -75,10 +75,6 @@ func SetupPlatformAPI(root gin.IRouter) {
 	logger.Info("Routes set up", logger.Params{"routes": len(routers)})
 }
 
-func SetupHealthCheckApi(root gin.IRouter) {
-	root.GET("/", GetObserverStatus)
-}
-
 // getRouter lazy loads routers
 func getRouter(router *gin.RouterGroup, handle string) gin.IRouter {
 	key := fmt.Sprintf("%s/%s", router.BasePath(), handle)
@@ -118,4 +114,4 @@ func splitParam(param string) []string {
 		return c == ','
 	}
 	return strings.FieldsFunc(param, splitFn)
-}
\ No newline at end of file
+}
diff --git a/cmd/observer_healthcheck/main.go b/cmd/observer_healthcheck/main.go
deleted file mode 100644
index b716972..0000000
--- a/cmd/observer_healthcheck/main.go
+++ /dev/null
@@ -1,50 +0,0 @@
-package main
-
-import (
-	sentrygin "github.com/getsentry/sentry-go/gin"
-	"github.com/gin-gonic/gin"
-	"github.com/spf13/viper"
-	"github.com/trustwallet/blockatlas/api"
-	"github.com/trustwallet/blockatlas/internal"
-	"github.com/trustwallet/blockatlas/pkg/logger"
-	"github.com/trustwallet/blockatlas/platform"
-	"github.com/trustwallet/blockatlas/services/observer/healthcheck"
-	"github.com/trustwallet/blockatlas/storage"
-	"time"
-)
-
-const (
-	defaultPort       = "3333"
-	defaultConfigPath = "../../config.yml"
-)
-
-var (
-	port, confPath string
-	engine         *gin.Engine
-	cache          *storage.Storage
-)
-
-func init() {
-	port, confPath = internal.ParseArgs(defaultPort, defaultConfigPath)
-
-	internal.InitConfig(confPath)
-	redisHost := viper.GetString("storage.redis")
-	cache = internal.InitRedis(redisHost)
-	logger.InitLogger()
-	tmp := sentrygin.New(sentrygin.Options{})
-	sg := &tmp
-
-	engine = internal.InitEngine(sg, viper.GetString("gin.mode"))
-
-	platform.Init(viper.GetString("platform"))
-}
-
-func main() {
-	for _, blockAPI := range platform.BlockAPIs {
-		go healthcheck.Worker(cache, blockAPI)
-		time.Sleep(time.Millisecond * 200)
-	}
-
-	api.SetupHealthCheckApi(engine)
-	internal.SetupGracefulShutdown(port, engine)
-}
diff --git a/cmd/observer_notifier/main.go b/cmd/observer_notifier/main.go
index d3032ea..6acfc2e 100644
--- a/cmd/observer_notifier/main.go
+++ b/cmd/observer_notifier/main.go
@@ -3,11 +3,11 @@ package main
 import (
 	"context"
 	"github.com/spf13/viper"
+	"github.com/trustwallet/blockatlas/db"
 	"github.com/trustwallet/blockatlas/internal"
 	"github.com/trustwallet/blockatlas/mq"
 	"github.com/trustwallet/blockatlas/pkg/logger"
 	"github.com/trustwallet/blockatlas/services/observer/notifier"
-	"github.com/trustwallet/blockatlas/storage"
 	"time"
 )
 
@@ -17,7 +17,7 @@ const (
 
 var (
 	confPath string
-	cache    *storage.Storage
+	database *db.Instance
 )
 
 func init() {
@@ -26,11 +26,10 @@ func init() {
 	internal.InitConfig(confPath)
 	logger.InitLogger()
 
-	redisHost := viper.GetString("storage.redis")
 	mqHost := viper.GetString("observer.rabbitmq.uri")
 	prefetchCount := viper.GetInt("observer.rabbitmq.consumer.prefetch_count")
+	pgUri := viper.GetString("postgres.uri")
 
-	cache = internal.InitRedis(redisHost)
 	internal.InitRabbitMQ(mqHost, prefetchCount)
 
 	if err := mq.RawTransactions.Declare(); err != nil {
@@ -41,8 +40,15 @@ func init() {
 		logger.Fatal(err)
 	}
 
-	go storage.RestoreConnectionWorker(cache, redisHost, time.Second*10)
+	var err error
+	database, err = db.New(pgUri)
+	if err != nil {
+		logger.Fatal(err)
+	}
+
 	go mq.RestoreConnectionWorker(mqHost, mq.RawTransactions, time.Second*10)
+	go db.RestoreConnectionWorker(database.DB, time.Second*10, pgUri)
+	time.Sleep(time.Millisecond)
 }
 
 func main() {
@@ -50,7 +56,7 @@ func main() {
 
 	ctx, cancel := context.WithCancel(context.Background())
 
-	go mq.RawTransactions.RunConsumerWithCancel(notifier.RunNotifier, cache, ctx)
+	go mq.RawTransactions.RunConsumerWithCancelAndDbConn(notifier.RunNotifier, database, ctx)
 
 	internal.SetupGracefulShutdownForObserver(cancel)
 }
diff --git a/cmd/observer_parser/main.go b/cmd/observer_parser/main.go
index 978418d..6aed486 100644
--- a/cmd/observer_parser/main.go
+++ b/cmd/observer_parser/main.go
@@ -4,13 +4,13 @@ import (
 	"context"
 	"fmt"
 	"github.com/spf13/viper"
+	"github.com/trustwallet/blockatlas/db"
 	"github.com/trustwallet/blockatlas/internal"
 	"github.com/trustwallet/blockatlas/mq"
 	"github.com/trustwallet/blockatlas/pkg/logger"
 	"github.com/trustwallet/blockatlas/platform"
 	"github.com/trustwallet/blockatlas/services/observer/notifier"
 	"github.com/trustwallet/blockatlas/services/observer/parser"
-	"github.com/trustwallet/blockatlas/storage"
 	"os"
 	"os/signal"
 	"sync"
@@ -24,10 +24,10 @@ const (
 
 var (
 	confPath                              string
-	cache                                 *storage.Storage
 	backlogTime, minInterval, maxInterval time.Duration
 	maxBackLogBlocks                      int64
 	txsBatchLimit                         uint
+	database                              *db.Instance
 )
 
 func init() {
@@ -36,12 +36,10 @@ func init() {
 	internal.InitConfig(confPath)
 	logger.InitLogger()
 
-	redisHost := viper.GetString("storage.redis")
 	mqHost := viper.GetString("observer.rabbitmq.uri")
 	prefetchCount := viper.GetInt("observer.rabbitmq.consumer.prefetch_count")
 	platformHandle := viper.GetString("platform")
 
-	cache = internal.InitRedis(redisHost)
 	internal.InitRabbitMQ(mqHost, prefetchCount)
 	platform.Init(platformHandle)
 
@@ -52,6 +50,9 @@ func init() {
 	if len(platform.BlockAPIs) == 0 {
 		logger.Fatal("No APIs to observe")
 	}
+
+	pgUri := viper.GetString("postgres.uri")
+
 	txsBatchLimit = viper.GetUint("observer.txs_batch_limit")
 	backlogTime = viper.GetDuration("observer.backlog")
 	minInterval = viper.GetDuration("observer.block_poll.min")
@@ -60,10 +61,14 @@ func init() {
 	if minInterval >= maxInterval {
 		logger.Fatal("minimum block polling interval cannot be greater or equal than maximum")
 	}
+	var err error
+	database, err = db.New(pgUri)
+	if err != nil {
+		logger.Fatal(err)
+	}
 
 	go mq.FatalWorker(time.Second * 10)
-	time.Sleep(time.Millisecond)
-	go storage.RestoreConnectionWorker(cache, redisHost, time.Second*10)
+	go db.RestoreConnectionWorker(database.DB, time.Second*10, pgUri)
 	time.Sleep(time.Millisecond)
 }
 
@@ -100,13 +105,13 @@ func main() {
 		params := parser.Params{
 			Ctx:                   ctx,
 			Api:                   api,
-			Storage:               cache,
 			Queue:                 mq.RawTransactions,
 			ParsingBlocksInterval: pollInterval,
 			BacklogCount:          backlogCount,
 			MaxBacklogBlocks:      maxBackLogBlocks,
 			StopChannel:           stopChannel,
 			TxBatchLimit:          txsBatchLimit,
+			Database:              database,
 		}
 
 		go parser.RunParser(params)
diff --git a/cmd/observer_subscriber/main.go b/cmd/observer_subscriber/main.go
index 3a16999..9c9bf9f 100644
--- a/cmd/observer_subscriber/main.go
+++ b/cmd/observer_subscriber/main.go
@@ -3,12 +3,12 @@ package main
 import (
 	"context"
 	"github.com/spf13/viper"
+	"github.com/trustwallet/blockatlas/db"
 	_ "github.com/trustwallet/blockatlas/docs"
 	"github.com/trustwallet/blockatlas/internal"
 	"github.com/trustwallet/blockatlas/mq"
 	"github.com/trustwallet/blockatlas/pkg/logger"
 	"github.com/trustwallet/blockatlas/services/observer/subscriber"
-	"github.com/trustwallet/blockatlas/storage"
 	"time"
 )
 
@@ -18,7 +18,7 @@ const (
 
 var (
 	confPath string
-	cache    *storage.Storage
+	database *db.Instance
 )
 
 func init() {
@@ -27,16 +27,22 @@ func init() {
 	internal.InitConfig(confPath)
 	logger.InitLogger()
 
-	redisHost := viper.GetString("storage.redis")
+	pgUri := viper.GetString("postgres.uri")
+
 	mqHost := viper.GetString("observer.rabbitmq.uri")
 	prefetchCount := viper.GetInt("observer.rabbitmq.consumer.prefetch_count")
 
-	cache = internal.InitRedis(redisHost)
-
 	internal.InitRabbitMQ(mqHost, prefetchCount)
 
+	var err error
+	database, err = db.New(pgUri)
+	if err != nil {
+		logger.Fatal(err)
+	}
+
 	go mq.FatalWorker(time.Second * 10)
-	go storage.RestoreConnectionWorker(cache, redisHost, time.Second*10)
+	go db.RestoreConnectionWorker(database.DB, time.Second*10, pgUri)
+	time.Sleep(time.Millisecond)
 }
 
 func main() {
@@ -46,7 +52,7 @@ func main() {
 	}
 	ctx, cancel := context.WithCancel(context.Background())
 
-	go mq.Subscriptions.RunConsumerWithCancel(subscriber.RunSubscriber, cache, ctx)
+	go mq.Subscriptions.RunConsumerWithCancelAndDbConn(subscriber.RunSubscriber, database, ctx)
 
 	internal.SetupGracefulShutdownForObserver(cancel)
 }
diff --git a/config.yml b/config.yml
index 9456cc0..6430bb6 100644
--- a/config.yml
+++ b/config.yml
@@ -9,7 +9,7 @@ gin:
 # If all - run all platforms in one binary. You can pick specific coin handle to run binary only with specific coin
 # Example: ethereum
 # You can see all the coin handles at coins/coins.yml file
-platform: all
+platform: binance
 
 # Sentry error tracking
 #sentry:
@@ -32,8 +32,8 @@ observer:
     consumer:
       prefetch_count: 10
 
-storage:
-  redis: redis://localhost:6379
+postgres:
+  uri: postgresql://user:pass@localhost/my_db?sslmode=disable
 
 # [BNB] Binance DEX: https://wallet.binance.org
 #       Binance Chain: https://explorer.binance.org
diff --git a/db/db.go b/db/db.go
new file mode 100644
index 0000000..98b4f41
--- /dev/null
+++ b/db/db.go
@@ -0,0 +1,52 @@
+package db
+
+import (
+	"github.com/jinzhu/gorm"
+	_ "github.com/jinzhu/gorm/dialects/postgres"
+	"github.com/trustwallet/blockatlas/db/models"
+
+	"github.com/trustwallet/blockatlas/pkg/logger"
+	"time"
+)
+
+type Instance struct {
+	DB *gorm.DB
+}
+
+func New(uri string) (*Instance, error) {
+	dbConn, err := gorm.Open("postgres", uri)
+	if err != nil {
+		return nil, err
+	}
+
+	dbConn.AutoMigrate(
+		&models.Subscription{},
+		&models.SubscriptionData{},
+		&models.Tracker{},
+	)
+	i := &Instance{DB: dbConn}
+
+	return i, nil
+}
+
+func RestoreConnectionWorker(dbConn *gorm.DB, timeout time.Duration, uri string) {
+	logger.Info("Run PG RestoreConnectionWorker")
+	for {
+		if err := dbConn.DB().Ping(); err != nil {
+			for {
+				logger.Warn("PG is not available now")
+				logger.Warn("Trying to connect to PG...")
+				dbConn, err = gorm.Open("postgres", uri)
+				if err != nil {
+					logger.Warn("PG is still unavailable:", err.Error())
+					time.Sleep(timeout)
+					continue
+				} else {
+					logger.Info("PG connection restored")
+					break
+				}
+			}
+		}
+		time.Sleep(timeout)
+	}
+}
diff --git a/db/models/subscriptions.go b/db/models/subscriptions.go
new file mode 100644
index 0000000..a98ab60
--- /dev/null
+++ b/db/models/subscriptions.go
@@ -0,0 +1,24 @@
+package models
+
+import (
+	"time"
+)
+
+type TimeModel struct {
+	CreatedAt time.Time
+	UpdatedAt time.Time
+}
+
+type Subscription struct {
+	TimeModel
+	SubscriptionId uint               `gorm:"primary_key:true"`
+	Data           []SubscriptionData `gorm:"foreignkey:SubscriptionId"`
+}
+
+type SubscriptionData struct {
+	TimeModel
+	ID             uint   `gorm:"primary_key:true"`
+	SubscriptionId uint   `sql:"index"`
+	Coin           uint   `sql:"index"`
+	Address        string `sql:"index"`
+}
diff --git a/db/models/tracker.go b/db/models/tracker.go
new file mode 100644
index 0000000..10e62a2
--- /dev/null
+++ b/db/models/tracker.go
@@ -0,0 +1,6 @@
+package models
+
+type Tracker struct {
+	Coin   uint `gorm:"primary_key:true"`
+	Height int64
+}
diff --git a/db/subscriptions.go b/db/subscriptions.go
new file mode 100644
index 0000000..8ef21dc
--- /dev/null
+++ b/db/subscriptions.go
@@ -0,0 +1,152 @@
+package db
+
+import (
+	"github.com/trustwallet/blockatlas/db/models"
+	"github.com/trustwallet/blockatlas/pkg/errors"
+)
+
+func (i *Instance) GetSubscriptionData(coin uint, addresses []string) ([]models.SubscriptionData, error) {
+	if len(addresses) == 0 {
+		return nil, errors.E("Empty addresses")
+	}
+
+	var subscriptionsDataList []models.SubscriptionData
+	err := i.DB.
+		Model(&models.SubscriptionData{}).
+		Where("address in (?) AND coin = ?", addresses, coin).
+		Find(&subscriptionsDataList).Error
+
+	if err != nil {
+		return nil, err
+	}
+	return subscriptionsDataList, nil
+}
+
+func (i *Instance) AddSubscriptions(id uint, subscriptions []models.SubscriptionData) error {
+	txInstance := Instance{DB: i.DB.Begin()}
+	defer func() {
+		if r := recover(); r != nil {
+			txInstance.DB.Rollback()
+		}
+	}()
+
+	if err := txInstance.DB.Error; err != nil {
+		return err
+	}
+	if len(subscriptions) == 0 {
+		return errors.E("Empty subscriptions")
+	}
+	var (
+		existingSub models.Subscription
+		err         error
+	)
+
+	recordNotFound := txInstance.DB.
+		Where(models.Subscription{SubscriptionId: id}).
+		First(&existingSub).
+		RecordNotFound()
+
+	subscriptions = removeSubscriptionDuplicates(subscriptions)
+	if recordNotFound {
+		err = txInstance.AddSubscription(id, subscriptions)
+	} else {
+		err = txInstance.AddToExistingSubscription(id, subscriptions)
+	}
+	if err != nil {
+		txInstance.DB.Rollback()
+		return err
+	}
+	return txInstance.DB.Commit().Error
+}
+
+func (i *Instance) AddSubscription(id uint, data []models.SubscriptionData) error {
+	return i.DB.Create(&models.Subscription{SubscriptionId: id, Data: data}).Error
+}
+
+func (i *Instance) AddToExistingSubscription(id uint, subscriptions []models.SubscriptionData) error {
+	var (
+		existingData []models.SubscriptionData
+		association  = i.DB.Model(&models.Subscription{SubscriptionId: id}).Association("Data")
+	)
+	if err := association.Error; err != nil {
+		return err
+	}
+	if err := association.Find(&existingData).Error; err != nil {
+		return err
+	}
+
+	updateList, deleteList := getSubscriptionsToDeleteAndUpdate(existingData, subscriptions)
+	if len(updateList) > 0 {
+		if err := association.Append(updateList).Error; err != nil {
+			return err
+		}
+	}
+	if len(deleteList) > 0 {
+		if err := i.DeleteSubscriptions(deleteList); err != nil {
+			return err
+		}
+	}
+	return nil
+}
+
+func (i *Instance) DeleteAllSubscriptions(id uint) error {
+	request := i.DB.Where("subscription_id = ?", id)
+	if err := request.Error; err != nil {
+		return err
+	}
+	return request.Delete(&models.SubscriptionData{}).Error
+}
+
+func (i *Instance) DeleteSubscriptions(subscriptions []models.SubscriptionData) error {
+	var (
+		errorsList = make([]error, 0)
+		errDetails string
+	)
+	for _, sub := range subscriptions {
+		if err := i.DB.Delete(&models.SubscriptionData{}, sub).Error; err != nil {
+			errorsList = append(errorsList, err)
+		}
+	}
+	if len(errorsList) != 0 {
+		for _, err := range errorsList {
+			errDetails += err.Error() + " "
+		}
+		return errors.E(errDetails)
+	}
+	return nil
+}
+
+func getSubscriptionsToDeleteAndUpdate(existing, new []models.SubscriptionData) (subToUpdate, subToDelete []models.SubscriptionData) {
+	for _, n := range new {
+		if !containSubscription(n, existing) {
+			subToUpdate = append(subToUpdate, n)
+		}
+	}
+	for _, e := range existing {
+		if !containSubscription(e, new) {
+			subToDelete = append(subToDelete, e)
+		}
+	}
+	return subToUpdate, subToDelete
+}
+
+func containSubscription(sub models.SubscriptionData, list []models.SubscriptionData) bool {
+	for _, s := range list {
+		if s.Address == sub.Address && sub.Coin == s.Coin && s.SubscriptionId == sub.SubscriptionId {
+			return true
+		}
+	}
+	return false
+}
+
+func removeSubscriptionDuplicates(sub []models.SubscriptionData) []models.SubscriptionData {
+	keys := make(map[models.SubscriptionData]bool)
+	result := make([]models.SubscriptionData, 0)
+	for _, entry := range sub {
+		if _, value := keys[entry]; !value {
+			keys[entry] = true
+			result = append(result, entry)
+		}
+	}
+	return result
+}
diff --git a/db/tracker.go b/db/tracker.go
new file mode 100644
index 0000000..60098a0
--- /dev/null
+++ b/db/tracker.go
@@ -0,0 +1,55 @@
+package db
+
+import (
+	"github.com/trustwallet/blockatlas/db/models"
+	"sync"
+)
+
+var memoryCache heightBlockMap
+
+func init() {
+	memoryCache.m = make(map[uint]int64)
+}
+
+type heightBlockMap struct {
+	m map[uint]int64
+	sync.RWMutex
+}
+
+func (hbm *heightBlockMap) SetHeight(coin uint, b int64) {
+	hbm.Lock()
+	defer hbm.Unlock()
+	hbm.m[coin] = b
+}
+
+func (hbm *heightBlockMap) GetHeight(coin uint) (int64, bool) {
+	hbm.RLock()
+	defer hbm.RUnlock()
+	b, ok := hbm.m[coin]
+	return b, ok
+}
+
+func (i *Instance) GetLastParsedBlockNumber(coin uint) (int64, error) {
+	height, ok := memoryCache.GetHeight(coin)
+	if ok {
+		return height, nil
+	}
+	var tracker models.Tracker
+	if err := i.DB.Where(models.Tracker{Coin: coin}).Find(&tracker).Error; err != nil {
+		return 0, nil
+	}
+	return tracker.Height, nil
+}
+
+func (i *Instance) SetLastParsedBlockNumber(coin uint, num int64) error {
+	memoryCache.SetHeight(coin, num)
+	tracker := models.Tracker{
+		Coin:   coin,
+		Height: num,
+	}
+
+	return i.DB.
+		Set("gorm:insert_option", "ON CONFLICT (coin) DO UPDATE SET height = excluded.height").
+		Where(models.Tracker{Coin: coin}).
+		Create(&tracker).Error
+}
diff --git a/go.mod b/go.mod
index de25d5f..a91e3f6 100644
--- a/go.mod
+++ b/go.mod
@@ -23,11 +23,12 @@ require (
 	github.com/google/go-cmp v0.4.0 // indirect
 	github.com/gorilla/websocket v1.4.1 // indirect
 	github.com/gotestyourself/gotestyourself v2.2.0+incompatible // indirect
+	github.com/hewigovens/go-coincodec v1.0.4
+	github.com/jinzhu/gorm v1.9.12
 	github.com/json-iterator/go v1.1.9 // indirect
 	github.com/klauspost/compress v1.10.1 // indirect
 	github.com/konsorten/go-windows-terminal-sequences v1.0.2 // indirect
 	github.com/kr/pretty v0.2.0 // indirect
-	github.com/lib/pq v1.1.1 // indirect
 	github.com/mitchellh/mapstructure v1.1.2
 	github.com/mr-tron/base58 v1.1.3
 	github.com/opencontainers/image-spec v1.0.1 // indirect
diff --git a/go.sum b/go.sum
index 29bdd1f..8b9d371 100644
--- a/go.sum
+++ b/go.sum
@@ -78,6 +78,7 @@ github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c
 github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
 github.com/deckarep/golang-set v1.7.1 h1:SCQV0S6gTtp6itiFrTqI+pfmJ4LN85S1YzhDf9rTHJQ=
 github.com/deckarep/golang-set v1.7.1/go.mod h1:93vsz/8Wt4joVM7c2AVqh+YRMiUSc14yDtF28KmMOgQ=
+github.com/denisenkom/go-mssqldb v0.0.0-20191124224453-732737034ffd/go.mod h1:xbL0rPBG9cCiLr28tMa8zpbdarY27NDyej4t/EjAShU=
 github.com/dgraph-io/badger v1.6.0/go.mod h1:zwt7syl517jmP8s94KqSxTlM6IMsdhYy6psNgSztDR4=
 github.com/dgrijalva/jwt-go v3.2.0+incompatible/go.mod h1:E3ru+11k8xSBh+hMPgOLZmtrrCbhqsmaPHjLKYnJCaQ=
 github.com/dgryski/go-farm v0.0.0-20190423205320-6a90982ecee2/go.mod h1:SqUrOPUnsFjfmXRMNPybcSiG0BgUW2AuFH8PAnS2iTw=
@@ -89,10 +90,9 @@ github.com/docker/go-units v0.4.0/go.mod h1:fgPhTUdO+D/Jk86RDLlptpiXQzgHJF7gydDD
 github.com/dustin/go-humanize v0.0.0-20171111073723-bb3d318650d4/go.mod h1:HtrtbFcZ19U5GC7JDqmcUSB87Iq5E25KnS6fMYU6eOk=
 github.com/dustin/go-humanize v1.0.0/go.mod h1:HtrtbFcZ19U5GC7JDqmcUSB87Iq5E25KnS6fMYU6eOk=
 github.com/eknkc/amber v0.0.0-20171010120322-cdade1c07385/go.mod h1:0vRUJqYpeSZifjYj7uP3BG/gKcuzL9xWVV/Y+cK33KM=
+github.com/erikstmartin/go-testdb v0.0.0-20160219214506-8d10e4a1bae5/go.mod h1:a2zkGnVExMxdzMo3M0Hi/3sEU+cWnZpSni0O6/Yb/P0=
 github.com/etcd-io/bbolt v1.3.3/go.mod h1:ZF2nL25h33cCyBtcyWeZ2/I3HQOfTP+0PIEvHjkjCrw=
-github.com/fasthttp-contrib/websocket v0.0.0-20160511215533-1f3b11f56072 h1:DddqAaWDpywytcG8w/qoQ5sAN8X12d3Z3koB0C3Rxsc=
 github.com/fasthttp-contrib/websocket v0.0.0-20160511215533-1f3b11f56072/go.mod h1:duJ4Jxv5lDcvg4QuQr0oowTf7dz4/CR8NtyCooz9HL8=
-github.com/fatih/structs v1.1.0 h1:Q7juDM0QtcnhCpeyLGQKyg4TOIghuNXrkL32pHAUMxo=
 github.com/fatih/structs v1.1.0/go.mod h1:9NiDSp5zOcgEDl+j00MP/WkGVPOlPRLejGD8Ga6PJ7M=
 github.com/flosch/pongo2 v0.0.0-20190707114632-bbf5a6c351f4/go.mod h1:T9YF2M40nIgbVgp3rreNmTged+9HrbNTIQf1PsaIiTA=
 github.com/fsnotify/fsnotify v1.4.7 h1:IXs+QLmnXW2CcXuY+8Mzv/fWEsPGWxqefPtCP5CnV9I=
@@ -101,7 +101,6 @@ github.com/gavv/httpexpect v2.0.0+incompatible/go.mod h1:x+9tiU1YnrOvnB725RkpoLv
 github.com/getsentry/sentry-go v0.4.0 h1:WqRI2/7EiALbdG9qGB47c0Aks1tdznG5DZd6GSQ1y/8=
 github.com/getsentry/sentry-go v0.4.0/go.mod h1:xkGcb82SipKQloDNa5b7hTV4VdEyc2bhwd1/UczP52k=
 github.com/ghodss/yaml v1.0.0/go.mod h1:4dBDuWmgqj2HViK6kFavaiC9ZROes6MMH2rRYeMEF04=
-github.com/gin-contrib/gzip v0.0.1 h1:ezvKOL6jH+jlzdHNE4h9h8q8uMpDQjyl0NN0Jd7jozc=
 github.com/gin-contrib/gzip v0.0.1/go.mod h1:fGBJBCdt6qCZuCAOwWuFhBB4OOq9EFqlo5dEaFhhu5w=
 github.com/gin-contrib/sse v0.0.0-20170109093832-22d885f9ecc7/go.mod h1:VJ0WA2NBN22VlZ2dKZQPAPnyWw5XTlK1KymzLKsr59s=
 github.com/gin-contrib/sse v0.0.0-20190301062529-5545eab6dad3/go.mod h1:VJ0WA2NBN22VlZ2dKZQPAPnyWw5XTlK1KymzLKsr59s=
@@ -112,11 +111,9 @@ github.com/gin-gonic/gin v1.4.0/go.mod h1:OW2EZn3DO8Ln9oIKOvM++LBO+5UPHJJDH72/q/
 github.com/gin-gonic/gin v1.5.0 h1:fi+bqFAx/oLK54somfCtEZs9HeH1LHVoEPUgARpTqyc=
 github.com/gin-gonic/gin v1.5.0/go.mod h1:Nd6IXA8m5kNZdNEHMBd93KT+mdY3+bewLgRvmCsR2Do=
 github.com/go-check/check v0.0.0-20180628173108-788fd7840127/go.mod h1:9ES+weclKsC9YodN5RgxqK/VD9HM9JsCSh7rNhMZE98=
-github.com/go-errors/errors v1.0.1 h1:LUHzmkK3GUKUrL/1gfBUxAHzcev3apQlezX/+O7ma6w=
 github.com/go-errors/errors v1.0.1/go.mod h1:f4zRHt4oKfwPJE5k8C9vpYG+aDHdBFUsgrm6/TyX73Q=
 github.com/go-kit/kit v0.8.0/go.mod h1:xBxKIO96dXMWWy0MnWVtmwkA9/13aqxPnvrjFYMA2as=
 github.com/go-logfmt/logfmt v0.3.0/go.mod h1:Qt1PoO58o5twSAckw1HlFXLmHsOX5/0LbT9GBnD5lWE=
-github.com/go-logfmt/logfmt v0.4.0 h1:MP4Eh7ZCb31lleYCFuwm0oe4/YGak+5l1vA2NOE80nA=
 github.com/go-logfmt/logfmt v0.4.0/go.mod h1:3RMwSq7FuexP4Kalkev3ejPJsZTpXXBr9+V4qmtdjCk=
 github.com/go-martini/martini v0.0.0-20170121215854-22fa46961aab/go.mod h1:/P9AEU963A2AYjv4d1V5eVL1CQbEJq6aCNHDDjibzu8=
 github.com/go-openapi/jsonpointer v0.17.0/go.mod h1:cOnomiV+CVVwFLk0A/MExoFMjwdsUdVpsRhURCKh+3M=
@@ -139,40 +136,34 @@ github.com/go-playground/locales v0.12.1 h1:2FITxuFt/xuCNP1Acdhv62OzaCiviiE4kotf
 github.com/go-playground/locales v0.12.1/go.mod h1:IUMDtCfWo/w/mtMfIE/IG2K+Ey3ygWanZIBtBW0W2TM=
 github.com/go-playground/universal-translator v0.16.0 h1:X++omBR/4cE2MNg91AoC3rmGrCjJ8eAeUP/K/EKx4DM=
 github.com/go-playground/universal-translator v0.16.0/go.mod h1:1AnU7NaIRDWWzGEKwgtJRd2xk99HeFyHw3yid4rvQIY=
-github.com/go-redis/redis v6.15.6+incompatible h1:H9evprGPLI8+ci7fxQx6WNZHJSb7be8FqJQRhdQZ5Sg=
 github.com/go-redis/redis v6.15.6+incompatible/go.mod h1:NAIEuMOZ/fxfXJIrKDQDz8wamY7mA7PouImQ2Jvg6kA=
-github.com/go-stack/stack v1.8.0 h1:5SgMzNM5HxrEjV0ww2lTmX6E2Izsfxas4+YHWRs3Lsk=
+github.com/go-sql-driver/mysql v1.4.1/go.mod h1:zAC/RDZ24gD3HViQzih4MyKcchzm+sOG5ZlKdlhCg5w=
 github.com/go-stack/stack v1.8.0/go.mod h1:v0f6uXyyMGvRgIKkXu+yp6POWl0qKG85gN/melR3HDY=
 github.com/gobwas/httphead v0.0.0-20180130184737-2c6c146eadee/go.mod h1:L0fX3K22YWvt/FAX9NnzrNzcI4wNYi9Yku4O0LKYflo=
 github.com/gobwas/pool v0.2.0/go.mod h1:q8bcK0KcYlCgd9e7WYLm9LpyS+YeLd8JVDW6WezmKEw=
 github.com/gobwas/ws v1.0.2/go.mod h1:szmBTxLgaFppYjEmNtny/v3w89xOydFnnZMcgRRu/EM=
 github.com/gogo/protobuf v1.1.1/go.mod h1:r8qH/GZQm5c6nD/R0oafs1akxWv10x8SbQlK7atdtwQ=
 github.com/gogo/protobuf v1.2.1/go.mod h1:hp+jE20tsWTFYpLwKvXlhS1hjn+gTNwPg2I6zVXpSg4=
+github.com/golang-sql/civil v0.0.0-20190719163853-cb61b32ac6fe/go.mod h1:8vg3r2VgvsThLBIFL93Qb5yWzgyZWhEmBwUJWevAkK0=
 github.com/golang/glog v0.0.0-20160126235308-23def4e6c14b/go.mod h1:SBH7ygxi8pfUlaOkMMuAQtPIUF8ecWP5IEl/CR7VP2Q=
 github.com/golang/groupcache v0.0.0-20190129154638-5b532d6fd5ef/go.mod h1:cIg4eruTrX1D+g88fzRXU5OdNfaM+9IcxsU14FzY7Hc=
 github.com/golang/mock v1.1.1/go.mod h1:oTYuIxOrZwtPieC+H1uAHpcLFnEyAGVDL/k47Jfbm0A=
 github.com/golang/protobuf v1.2.0/go.mod h1:6lQm79b+lXiMfvg/cZm0SGofjICqVBUtrP5yJMmIC1U=
 github.com/golang/protobuf v1.3.1/go.mod h1:6lQm79b+lXiMfvg/cZm0SGofjICqVBUtrP5yJMmIC1U=
-github.com/golang/protobuf v1.3.2 h1:6nsPYzhq5kReh6QImI3k5qWzO4PEbvbIW2cwSfR/6xs=
 github.com/golang/protobuf v1.3.2/go.mod h1:6lQm79b+lXiMfvg/cZm0SGofjICqVBUtrP5yJMmIC1U=
 github.com/golang/protobuf v1.3.3 h1:gyjaxf+svBWX08ZjK86iN9geUJF0H6gp2IRKX6Nf6/I=
 github.com/golang/protobuf v1.3.3/go.mod h1:vzj43D7+SQXF/4pzW/hwtAqwc6iTitCiVSaWz5lYuqw=
-github.com/gomodule/redigo v1.7.1-0.20190724094224-574c33c3df38 h1:y0Wmhvml7cGnzPa9nocn/fMraMH/lMDdeG+rkx4VgYY=
 github.com/gomodule/redigo v1.7.1-0.20190724094224-574c33c3df38/go.mod h1:B4C85qUVwatsJoIUNIfCRsp7qO0iAmpGFZ4EELWSbC4=
 github.com/google/btree v1.0.0/go.mod h1:lNA+9X1NB3Zf8V7Ke586lFgjr2dZNuvo3lPJSGZ5JPQ=
 github.com/google/go-cmp v0.2.0/go.mod h1:oXzfMopK8JAjlY9xF4vHSVASa0yLyX7SntLO5aqRK0M=
 github.com/google/go-cmp v0.3.1/go.mod h1:8QqcDgzrUqlUb/G2PQTWiueGozuR1884gddMywk6iLU=
 github.com/google/go-cmp v0.4.0 h1:xsAVV57WRhGj6kEIi8ReJzQlHHqcBYCElAvkovg3B/4=
 github.com/google/go-cmp v0.4.0/go.mod h1:v8dTdLbMG2kIc/vJvl+f65V22dbkXbowE6jgT/gNBxE=
-github.com/google/go-querystring v1.0.0 h1:Xkwi/a1rcvNg1PPYe5vI8GbeBY/jrVuDX5ASuANWTrk=
 github.com/google/go-querystring v1.0.0/go.mod h1:odCYkC5MyYFN7vkCjXpyrEuKhc/BUO6wN/zVPAxq5ck=
 github.com/google/gofuzz v1.0.0/go.mod h1:dBl0BpW6vV/+mYPU4Po3pmUjxk6FQPldtuIdl/M65Eg=
-github.com/gopherjs/gopherjs v0.0.0-20181017120253-0766667cb4d1 h1:EGx4pi6eqNxGaHF6qqu48+N2wcFQ5qg5FXgOdqsJ5d8=
 github.com/gopherjs/gopherjs v0.0.0-20181017120253-0766667cb4d1/go.mod h1:wJfORRmW1u3UXTncJ5qlYoELFm8eSnnEO6hX4iZ3EWY=
 github.com/gorilla/websocket v1.4.0/go.mod h1:E7qHFY5m1UJ88s3WnNqhKjPHQ0heANvMoAMk2YaljkQ=
-github.com/gorilla/websocket v1.4.1 h1:q7AeDBpnBk8AogcD4DSag/Ukw/KV+YhzLj2bP5HvKCM=
 github.com/gorilla/websocket v1.4.1/go.mod h1:YR8l580nyteQvAITg2hZ9XVh4b55+EU/adAjf1fMHhE=
-github.com/gotestyourself/gotestyourself v2.2.0+incompatible h1:AQwinXlbQR2HvPjQZOmDhRqsv5mZf+Jb1RnSLxcqZcI=
 github.com/gotestyourself/gotestyourself v2.2.0+incompatible/go.mod h1:zZKM6oeNM8k+FRljX1mnzVYeS8wiGgQyvST1/GafPbY=
 github.com/grpc-ecosystem/go-grpc-middleware v1.0.0/go.mod h1:FiyG127CGDf3tlThmgyCl78X/SZQqEOJBCDaAfeWzPs=
 github.com/grpc-ecosystem/go-grpc-prometheus v1.2.0/go.mod h1:8NvIoxWQoOIhqOTXgfV/d3M/q6VIi02HzZEHgUlZvzk=
@@ -180,9 +171,8 @@ github.com/grpc-ecosystem/grpc-gateway v1.9.0/go.mod h1:vNeuVxBJEsws4ogUvrchl83t
 github.com/hashicorp/go-version v1.2.0/go.mod h1:fltr4n8CU8Ke44wwGCBoEymUuxUHl09ZGVZPK5anwXA=
 github.com/hashicorp/hcl v1.0.0 h1:0Anlzjpi4vEasTeNFn2mLJgTSwt0+6sfsiTG8qcWGx4=
 github.com/hashicorp/hcl v1.0.0/go.mod h1:E5yfLk+7swimpb2L/Alb/PJmXilQ/rhwaUYs4T20WEQ=
-github.com/hpcloud/tail v1.0.0 h1:nfCOvKYfkgYP8hkirhJocXT2+zOD8yUNjXaWfTlyFKI=
+github.com/hewigovens/go-coincodec v1.0.4/go.mod h1:+LTdzScnu782gMt0J3ZccPY9S2uyrGe0R1LUYGWecfc=
 github.com/hpcloud/tail v1.0.0/go.mod h1:ab1qPbhIpdTxEkNHXyeSf5vhxWSCs/tWer42PpOxQnU=
-github.com/imkira/go-interpol v1.1.0 h1:KIiKr0VSG2CUW1hl1jpiyuzuJeKUUpC8iM1AIE7N1Vk=
 github.com/imkira/go-interpol v1.1.0/go.mod h1:z0h2/2T3XF8kyEPpRgJ3kmNv+C43p+I/CoI+jC3w2iA=
 github.com/inconshreveable/mousetrap v1.0.0/go.mod h1:PxqpIevigyE2G7u3NXJIT2ANytuPF1OarO4DADm73n8=
 github.com/iris-contrib/blackfriday v2.0.0+incompatible/go.mod h1:UzZ2bDEoaSGPbkg6SAB4att1aAwTmVIx/5gCVqeyUdI=
@@ -190,20 +180,22 @@ github.com/iris-contrib/go.uuid v2.0.0+incompatible/go.mod h1:iz2lgM/1UnEf1kP0L/
 github.com/iris-contrib/i18n v0.0.0-20171121225848-987a633949d0/go.mod h1:pMCz62A0xJL6I+umB2YTlFRwWXaDFA0jy+5HzGiJjqI=
 github.com/iris-contrib/schema v0.0.1/go.mod h1:urYA3uvUNG1TIIjOSCzHr9/LmbQo8LrOcOqfqxa4hXw=
 github.com/jessevdk/go-flags v0.0.0-20141203071132-1679536dcc89/go.mod h1:4FA24M0QyGHXBuZZK/XkWh8h0e1EYbRYJSGM75WSRxI=
+github.com/jinzhu/gorm v1.9.12 h1:Drgk1clyWT9t9ERbzHza6Mj/8FY/CqMyVzOiHviMo6Q=
+github.com/jinzhu/gorm v1.9.12/go.mod h1:vhTjlKSJUTWNtcbQtrMBFCxy7eXTzeCAzfL5fBZT/Qs=
+github.com/jinzhu/inflection v1.0.0 h1:K317FqzuhWc8YvSVlFMCCUb36O/S9MCKRDI7QkRKD/E=
+github.com/jinzhu/inflection v1.0.0/go.mod h1:h+uFLlag+Qp1Va5pdKtLDYj+kHp5pxUVkryuEj+Srlc=
+github.com/jinzhu/now v1.0.1/go.mod h1:d3SSVoowX0Lcu0IBviAWJpolVfI5UJVZZ7cO71lE/z8=
 github.com/jonboulle/clockwork v0.1.0/go.mod h1:Ii8DK3G1RaLaWxj9trq07+26W01tbo22gdxWY5EU2bo=
 github.com/jrick/logrotate v1.0.0/go.mod h1:LNinyqDIJnpAur+b8yyulnQw/wDuN1+BYKlTRt3OuAQ=
 github.com/json-iterator/go v1.1.5/go.mod h1:+SdeFBvtyEkXs7REEP0seUULqWtbJapLOCVDaaPEHmU=
 github.com/json-iterator/go v1.1.6/go.mod h1:+SdeFBvtyEkXs7REEP0seUULqWtbJapLOCVDaaPEHmU=
 github.com/json-iterator/go v1.1.7/go.mod h1:KdQUCv79m/52Kvf8AW2vK1V8akMuk1QjK/uOdHXbAo4=
-github.com/json-iterator/go v1.1.9 h1:9yzud/Ht36ygwatGx56VwCZtlI/2AD15T1X2sjSuGns=
 github.com/json-iterator/go v1.1.9/go.mod h1:KdQUCv79m/52Kvf8AW2vK1V8akMuk1QjK/uOdHXbAo4=
-github.com/jtolds/gls v4.20.0+incompatible h1:xdiiI2gbIgH/gLH7ADydsJ1uDOEzR8yvV7C0MuV77Wo=
 github.com/jtolds/gls v4.20.0+incompatible/go.mod h1:QJZ7F/aHp+rZTRtaJ1ow/lLfFfVYBRgL+9YlvaHOwJU=
 github.com/juju/errors v0.0.0-20181118221551-089d3ea4e4d5/go.mod h1:W54LbzXuIE0boCoNJfwqpmkKJ1O4TCTZMetAt6jGk7Q=
 github.com/juju/loggo v0.0.0-20180524022052-584905176618/go.mod h1:vgyd7OREkbtVEN/8IXZe5Ooef3LQePvuBm9UWj6ZL8U=
 github.com/juju/testing v0.0.0-20180920084828-472a3e8b2073/go.mod h1:63prj8cnj0tU0S9OHjGJn+b1h0ZghCndfnbQolrYTwA=
 github.com/julienschmidt/httprouter v1.2.0/go.mod h1:SYymIcj16QtmaHHD7aYtjjsJG7VTCxuUUipMqKk8s4w=
-github.com/k0kubun/colorstring v0.0.0-20150214042306-9440f1994b88 h1:uC1QfSlInpQF+M0ao65imhwqKnz3Q2z/d8PWZRMQvDM=
 github.com/k0kubun/colorstring v0.0.0-20150214042306-9440f1994b88/go.mod h1:3w7q1U84EfirKl04SVQ/s7nPm1ZPhiXd34z40TNz36k=
 github.com/kataras/golog v0.0.9/go.mod h1:12HJgwBIZFNGL0EJnMRhmvGA0PQGx8VFwrZtM4CqbAk=
 github.com/kataras/iris/v12 v12.0.1/go.mod h1:udK4vLQKkdDqMGJJVd/msuMtN6hpYJhg/lSzuxjhO+U=
@@ -213,24 +205,16 @@ github.com/kisielk/errcheck v1.1.0/go.mod h1:EZBBE59ingxPouuu3KfxchcWSUPOHkagtvW
 github.com/kisielk/gotool v1.0.0/go.mod h1:XhKaO+MFFWcvkIS/tQcRk01m1F5IRFswLeQ+oQHNcck=
 github.com/kkdai/bstream v0.0.0-20161212061736-f391b8402d23/go.mod h1:J+Gs4SYgM6CZQHDETBtE9HaSEkGmuNXF86RwHhHUvq4=
 github.com/klauspost/compress v1.8.2/go.mod h1:RyIbtBH6LamlWaDj8nUwkbUhJ87Yi3uG0guNDohfE1A=
-github.com/klauspost/compress v1.9.0 h1:GhthINjveNZAdFUD8QoQYfjxnOONZgztK/Yr6M23UTY=
 github.com/klauspost/compress v1.9.0/go.mod h1:RyIbtBH6LamlWaDj8nUwkbUhJ87Yi3uG0guNDohfE1A=
-github.com/klauspost/compress v1.10.1 h1:a/QY0o9S6wCi0XhxaMX/QmusicNUqCqFugR6WKPOSoQ=
 github.com/klauspost/compress v1.10.1/go.mod h1:aoV0uJVorq1K+umq18yTdKaF57EivdYsUV+/s2qKfXs=
 github.com/klauspost/cpuid v1.2.1/go.mod h1:Pj4uuM528wm8OyEC2QMXAi2YiTZ96dNQPGgoMS4s3ek=
-github.com/konsorten/go-windows-terminal-sequences v1.0.1 h1:mweAR1A6xJ3oS2pRaGiHgQ4OO8tzTaLawm8vnODuwDk=
 github.com/konsorten/go-windows-terminal-sequences v1.0.1/go.mod h1:T0+1ngSBFLxvqU3pZ+m/2kptfBszLMUkC4ZK/EgS/cQ=
-github.com/konsorten/go-windows-terminal-sequences v1.0.2 h1:DB17ag19krx9CFsz4o3enTrPXyIXCl+2iCXH/aMAp9s=
 github.com/konsorten/go-windows-terminal-sequences v1.0.2/go.mod h1:T0+1ngSBFLxvqU3pZ+m/2kptfBszLMUkC4ZK/EgS/cQ=
-github.com/kr/logfmt v0.0.0-20140226030751-b84e30acd515 h1:T+h1c/A9Gawja4Y9mFVWj2vyii2bbUNDw3kt9VxK2EY=
 github.com/kr/logfmt v0.0.0-20140226030751-b84e30acd515/go.mod h1:+0opPa2QZZtGFBFZlji/RkVcI2GknAs/DXo4wKdlNEc=
-github.com/kr/pretty v0.1.0 h1:L/CwN0zerZDmRFUapSPitk6f+Q3+0za1rQkzVuMiMFI=
 github.com/kr/pretty v0.1.0/go.mod h1:dAy3ld7l9f0ibDNOQOHHMYYIIbhfbHSm3C4ZsoJORNo=
-github.com/kr/pretty v0.2.0 h1:s5hAObm+yFO5uHYt5dYjxi2rXrsnmRpJx4OYvIWUaQs=
 github.com/kr/pretty v0.2.0/go.mod h1:ipq/a2n7PKx3OHsz4KJII5eveXtPO4qwEXGdVfWzfnI=
 github.com/kr/pty v1.1.1/go.mod h1:pFQYn66WHrOpPYNljwOMqo10TkYh1fy3cYio2l3bCsQ=
 github.com/kr/pty v1.1.5/go.mod h1:9r2w37qlBe7rQ6e1fg1S/9xpWHSnaqNdHD3WcMdbPDA=
-github.com/kr/text v0.1.0 h1:45sCR5RtlFHMR4UwH9sdQ5TC8v0qDQCHnXt+kaKSTVE=
 github.com/kr/text v0.1.0/go.mod h1:4Jbv+DJW3UT/LiOwJeYQe1efqtUx/iVham/4vfdArNI=
 github.com/labstack/echo/v4 v4.1.11/go.mod h1:i541M3Fj6f76NZtHSj7TXnyM8n2gaodfvfxNnFqi74g=
 github.com/labstack/gommon v0.3.0/go.mod h1:MULnywXg0yavhxWKc+lOruYdAhDwPK9wf0OL7NoOu+k=
@@ -245,15 +229,14 @@ github.com/mailru/easyjson v0.0.0-20180823135443-60711f1a8329/go.mod h1:C1wdFJiN
 github.com/mailru/easyjson v0.0.0-20190614124828-94de47d64c63/go.mod h1:C1wdFJiN94OJF2b5HbByQZoLdCWB1Yqtg26g4irojpc=
 github.com/mailru/easyjson v0.0.0-20190626092158-b2ccc519800e h1:hB2xlXdHp/pmPZq0y3QnmWAArdw9PqbmotexnWx/FU8=
 github.com/mailru/easyjson v0.0.0-20190626092158-b2ccc519800e/go.mod h1:C1wdFJiN94OJF2b5HbByQZoLdCWB1Yqtg26g4irojpc=
-github.com/mattn/go-colorable v0.1.2 h1:/bC9yWikZXAL9uJdulbSfyVNIR3n3trXl+v8+1sx8mU=
 github.com/mattn/go-colorable v0.1.2/go.mod h1:U0ppj6V5qS13XJ6of8GYAs25YV2eR4EVcfRqFIhoBtE=
 github.com/mattn/go-isatty v0.0.4/go.mod h1:M+lRXTBqGeGNdLjl/ufCoiOlB5xdOkqRJdNxMWT7Zi4=
 github.com/mattn/go-isatty v0.0.7/go.mod h1:Iq45c/XA43vh69/j3iqttzPXn0bhXyGjM0Hdxcsrc5s=
 github.com/mattn/go-isatty v0.0.8/go.mod h1:Iq45c/XA43vh69/j3iqttzPXn0bhXyGjM0Hdxcsrc5s=
 github.com/mattn/go-isatty v0.0.9 h1:d5US/mDsogSGW37IV293h//ZFaeajb69h+EHFsv2xGg=
 github.com/mattn/go-isatty v0.0.9/go.mod h1:YNRxwqDuOph6SZLI9vUUz6OYw3QyUt7WiY2yME+cCiQ=
+github.com/mattn/go-sqlite3 v2.0.1+incompatible/go.mod h1:FPy6KqzDD04eiIsT53CuJW3U88zkxoIYsOqkbpncsNc=
 github.com/mattn/goveralls v0.0.2/go.mod h1:8d1ZMHsd7fW6IRPKQh46F2WRpyib5/X4FOpevwGNQEw=
-github.com/matttproud/golang_protobuf_extensions v1.0.1 h1:4hp9jkHxhMHkqkrB3Ix0jegS5sx/RkqARlsWZ6pIwiU=
 github.com/matttproud/golang_protobuf_extensions v1.0.1/go.mod h1:D8He9yQNgCq6Z5Ld7szi9bcBfOoFv/3dc6xSMkL2PC0=
 github.com/mediocregopher/mediocre-go-lib v0.0.0-20181029021733-cb65787f37ed/go.mod h1:dSsfyI2zABAdhcbvkXqgxOxrCsbYeHCPgrZkku60dSg=
 github.com/mediocregopher/radix/v3 v3.3.0/go.mod h1:EmfVyvspXz1uZEyPBMyGK+kjWiKQGvsUt6O3Pj+LDCQ=
@@ -262,12 +245,9 @@ github.com/mitchellh/go-homedir v1.1.0/go.mod h1:SfyaCUpYCn1Vlf4IUYiD9fPX4A5wJrk
 github.com/mitchellh/mapstructure v1.1.2 h1:fmNYVwqnSfB9mZU6OS2O6GsXM+wcskZDuKQzvN1EDeE=
 github.com/mitchellh/mapstructure v1.1.2/go.mod h1:FVVH3fgwuzCH5S8UJGiWEs2h04kUh9fWfEaFds41c1Y=
 github.com/modern-go/concurrent v0.0.0-20180228061459-e0a39a4cb421/go.mod h1:6dJC0mAP4ikYIbvyc7fijjWJddQyLn8Ig3JB5CqoB9Q=
-github.com/modern-go/concurrent v0.0.0-20180306012644-bacd9c7ef1dd h1:TRLaZ9cD/w8PVh93nsPXa1VrQ6jlwL5oN8l14QlcNfg=
 github.com/modern-go/concurrent v0.0.0-20180306012644-bacd9c7ef1dd/go.mod h1:6dJC0mAP4ikYIbvyc7fijjWJddQyLn8Ig3JB5CqoB9Q=
 github.com/modern-go/reflect2 v0.0.0-20180701023420-4b7aa43c6742/go.mod h1:bx2lNnkwVCuqBIxFjflWJWanXIb3RllmbCylyMrvgv0=
-github.com/modern-go/reflect2 v1.0.1 h1:9f412s+6RmYXLWZSEzVVgPGK7C2PphHj5RJrvfx9AWI=
 github.com/modern-go/reflect2 v1.0.1/go.mod h1:bx2lNnkwVCuqBIxFjflWJWanXIb3RllmbCylyMrvgv0=
-github.com/moul/http2curl v1.0.0 h1:dRMWoAtb+ePxMlLkrCbAqh4TlPHXvoGUSQ323/9Zahs=
 github.com/moul/http2curl v1.0.0/go.mod h1:8UbvGypXm98wA/IqH45anm5Y2Z6ep6O31QGOAZ3H0fQ=
 github.com/mr-tron/base58 v1.1.3 h1:v+sk57XuaCKGXpWtVBX8YJzO7hMGx4Aajh4TQbdEFdc=
 github.com/mr-tron/base58 v1.1.3/go.mod h1:BinMc/sQntlIE1frQmRFPUoPA1Zkr8VRgBdjWI2mNwc=
@@ -279,11 +259,9 @@ github.com/oklog/ulid v1.3.1/go.mod h1:CirwcVhetQ6Lv90oh/F+FBtV6XMibvdAFo93nm5qn
 github.com/onsi/ginkgo v1.6.0/go.mod h1:lLunBs/Ym6LB5Z9jYTR76FiuTmxDTDusOGeTQH+WWjE=
 github.com/onsi/ginkgo v1.7.0/go.mod h1:lLunBs/Ym6LB5Z9jYTR76FiuTmxDTDusOGeTQH+WWjE=
 github.com/onsi/ginkgo v1.10.1/go.mod h1:lLunBs/Ym6LB5Z9jYTR76FiuTmxDTDusOGeTQH+WWjE=
-github.com/onsi/ginkgo v1.10.3 h1:OoxbjfXVZyod1fmWYhI7SEyaD8B00ynP3T+D5GiyHOY=
 github.com/onsi/ginkgo v1.10.3/go.mod h1:lLunBs/Ym6LB5Z9jYTR76FiuTmxDTDusOGeTQH+WWjE=
 github.com/onsi/gomega v1.4.3/go.mod h1:ex+gbHU/CVuBBDIJjb2X0qEXbFg53c61hWP/1CpauHY=
 github.com/onsi/gomega v1.7.0/go.mod h1:ex+gbHU/CVuBBDIJjb2X0qEXbFg53c61hWP/1CpauHY=
-github.com/onsi/gomega v1.7.1 h1:K0jcRCwNQM3vFGh1ppMtDh/+7ApJrjldlX8fA0jDTLQ=
 github.com/onsi/gomega v1.7.1/go.mod h1:XdKZgCCFLUoM/7CFJVPcG8C1xQ1AJ0vpAezJrB7JYyY=
 github.com/opencontainers/go-digest v1.0.0-rc1 h1:WzifXhOVOEOuFYOJAW6aQqW0TooG2iki3E3Ii+WN7gQ=
 github.com/opencontainers/go-digest v1.0.0-rc1/go.mod h1:cMLVZDEM3+U2I4VmLI6N8jQYUd2OVphdqWwCJHrFt2s=
@@ -297,11 +275,9 @@ github.com/patrickmn/go-cache v2.1.0+incompatible h1:HRMgzkcYKYpi3C8ajMPV8OFXaaR
 github.com/patrickmn/go-cache v2.1.0+incompatible/go.mod h1:3Qf8kWWT7OJRJbdiICTKqZju1ZixQ/KpMGzzAfe6+WQ=
 github.com/pelletier/go-toml v1.2.0 h1:T5zMGML61Wp+FlcbWjRDT7yAxhJNAiPPLOFECq181zc=
 github.com/pelletier/go-toml v1.2.0/go.mod h1:5z9KED0ma1S8pY6P1sdut58dfprrGBbd/94hg7ilaic=
-github.com/pingcap/errors v0.11.4 h1:lFuQV/oaUMGcD2tqt+01ROSmJs75VG1ToEOkZIZ4nE4=
 github.com/pingcap/errors v0.11.4/go.mod h1:Oi8TUi2kEtXXLMJk9l1cGmz20kV3TaQ0usTwv5KuLY8=
 github.com/pkg/errors v0.8.0/go.mod h1:bwawxfHBFNV+L2hUp1rHADufV3IMtnDRdf1r5NINEl0=
 github.com/pkg/errors v0.8.1-0.20171018195549-f15c970de5b7/go.mod h1:bwawxfHBFNV+L2hUp1rHADufV3IMtnDRdf1r5NINEl0=
-github.com/pkg/errors v0.8.1 h1:iURUrRGxPUNPdy5/HRSm+Yj6okJ6UtLINN0Q9M4+h3I=
 github.com/pkg/errors v0.8.1/go.mod h1:bwawxfHBFNV+L2hUp1rHADufV3IMtnDRdf1r5NINEl0=
 github.com/pkg/errors v0.9.1 h1:FEBLx1zS214owpjy7qsBeixbURkuhQAwrK5UwLGTwt4=
 github.com/pkg/errors v0.9.1/go.mod h1:bwawxfHBFNV+L2hUp1rHADufV3IMtnDRdf1r5NINEl0=
@@ -321,7 +297,6 @@ github.com/russross/blackfriday v1.5.2/go.mod h1:JO/DiYxRf+HjHt06OyowR9PTA263kcR
 github.com/russross/blackfriday/v2 v2.0.1/go.mod h1:+Rmxgy9KzJVeS9/2gXHxylqXiyQDYRxCVz55jmeOWTM=
 github.com/ryanuber/columnize v2.1.0+incompatible/go.mod h1:sm1tb6uqfes/u+d4ooFouqFdy9/2g9QGwK3SQygK0Ts=
 github.com/satori/go.uuid v1.2.0/go.mod h1:dA0hQrYB0VpLJoorglMZABFdXlWrHn1NEOzdhQKdks0=
-github.com/sergi/go-diff v1.0.0 h1:Kpca3qRNrduNnOQeazBd0ysaKrUJiIuISHxogkT9RPQ=
 github.com/sergi/go-diff v1.0.0/go.mod h1:0CfEIISq7TuYL3j771MWULgwwjU+GofnZX9QAmXWZgo=
 github.com/shopspring/decimal v0.0.0-20200105231215-408a2507e114 h1:Pm6R878vxWWWR+Sa3ppsLce/Zq+JNTs6aVvRu13jv9A=
 github.com/shopspring/decimal v0.0.0-20200105231215-408a2507e114/go.mod h1:DKyhrW/HYNuLGql+MJL6WCR6knT2jwCFRcu2hWCYk4o=
@@ -331,9 +306,7 @@ github.com/sirupsen/logrus v1.2.0/go.mod h1:LxeOpSwHxABJmUn/MG1IvRgCAasNZTLOkJPx
 github.com/sirupsen/logrus v1.4.1/go.mod h1:ni0Sbl8bgC9z8RoU9G6nDWqqs/fq4eDPysMBDgk/93Q=
 github.com/sirupsen/logrus v1.4.2 h1:SPIRibHv4MatM3XXNO2BJeFLZwZ2LvZgfQ5+UNI2im4=
 github.com/sirupsen/logrus v1.4.2/go.mod h1:tLMulIdttU9McNUspp0xgXVQah82FyeX6MwdIuYE2rE=
-github.com/smartystreets/assertions v0.0.0-20180927180507-b2de0cb4f26d h1:zE9ykElWQ6/NYmHa3jpm/yHnI4xSofP+UP6SpjHcSeM=
 github.com/smartystreets/assertions v0.0.0-20180927180507-b2de0cb4f26d/go.mod h1:OnSkiWE9lh6wB0YB77sQom3nweQdgAjqCqsofrRNTgc=
-github.com/smartystreets/goconvey v1.6.4 h1:fv0U8FUIMPNf1L9lnHLvLhgicrIVChEkdzIKYqbNC9s=
 github.com/smartystreets/goconvey v1.6.4/go.mod h1:syvi0/a8iFYH4r/RixwvyeAJjdLS9QV7WQ/tjFTllLA=
 github.com/soheilhy/cmux v0.1.4/go.mod h1:IM3LyeVVIOuxMH7sFAkER9+bJ4dT7Ms6E4xg4kGIyLM=
 github.com/spaolacci/murmur3 v0.0.0-20180118202830-f09979ecbc72/go.mod h1:JwIasOWyU6f++ZhiEuf87xNszmSA2myDM2Kzu9HwQUA=
@@ -384,31 +357,21 @@ github.com/ugorji/go/codec v1.1.7/go.mod h1:Ax+UKWsSmolVDwsd+7N3ZtXu+yMGCf907BLY
 github.com/urfave/cli v1.20.0/go.mod h1:70zkFmudgCuE/ngEzBv17Jvp/497gISqfk5gWijbERA=
 github.com/urfave/cli v1.22.2/go.mod h1:Gos4lmkARVdJ6EkW0WaNv/tZAAMe9V7XWyB60NtXRu0=
 github.com/urfave/negroni v1.0.0/go.mod h1:Meg73S6kFm/4PpbYdq35yYWoCZ9mS/YSx+lKnmiohz4=
-github.com/valyala/bytebufferpool v1.0.0 h1:GqA5TC/0021Y/b9FG4Oi9Mr3q7XYx6KllzawFIhcdPw=
 github.com/valyala/bytebufferpool v1.0.0/go.mod h1:6bBcMArwyJ5K/AmCkWv1jt77kVWyCJ6HpOuEn7z0Csc=
-github.com/valyala/fasthttp v1.6.0 h1:uWF8lgKmeaIewWVPwi4GRq2P6+R46IgYZdxWtM+GtEY=
 github.com/valyala/fasthttp v1.6.0/go.mod h1:FstJa9V+Pj9vQ7OJie2qMHdwemEDaDiSdBnvPM1Su9w=
 github.com/valyala/fasttemplate v1.0.1/go.mod h1:UQGH1tvbgY+Nz5t2n7tXsz52dQxojPUpymEIMZ47gx8=
 github.com/valyala/tcplisten v0.0.0-20161114210144-ceec8f93295a/go.mod h1:v3UYOV9WzVtRmSR+PDvWpU/qWl4Wa5LApYYX4ZtKbio=
 github.com/wealdtech/go-slip44 v1.0.0 h1:g0Wi5EufdD/ULZ4hcZKQ3TGlA6FZhGTCkmVzYmf+7Is=
 github.com/wealdtech/go-slip44 v1.0.0/go.mod h1://S3V9M6iN1cCflLXMjJYciRPakCDubG9YpWVyW1AvM=
-github.com/xeipuuv/gojsonpointer v0.0.0-20180127040702-4e3ac2762d5f h1:J9EGpcZtP0E/raorCMxlFGSTBrsSlaDGf3jU/qvAE2c=
 github.com/xeipuuv/gojsonpointer v0.0.0-20180127040702-4e3ac2762d5f/go.mod h1:N2zxlSyiKSe5eX1tZViRH5QA0qijqEDrYZiPEAiq3wU=
-github.com/xeipuuv/gojsonreference v0.0.0-20180127040603-bd5ef7bd5415 h1:EzJWgHovont7NscjpAxXsDA8S8BMYve8Y5+7cuRE7R0=
 github.com/xeipuuv/gojsonreference v0.0.0-20180127040603-bd5ef7bd5415/go.mod h1:GwrjFmJcFw6At/Gs6z4yjiIwzuJ1/+UwLxMQDVQXShQ=
-github.com/xeipuuv/gojsonschema v1.2.0 h1:LhYJRs+L4fBtjZUfuSZIKGeVu0QRy8e5Xi7D17UxZ74=
 github.com/xeipuuv/gojsonschema v1.2.0/go.mod h1:anYRn/JVcOK2ZgGU+IjEV4nwlhoK5sQluxsYJ78Id3Y=
 github.com/xiang90/probing v0.0.0-20190116061207-43a291ad63a2/go.mod h1:UETIi67q53MR2AWcXfiuqkDkRtnGDLqkBTpCHuJHxtU=
 github.com/xordataexchange/crypt v0.0.3-0.20170626215501-b2862e3d0a77/go.mod h1:aYKd//L2LvnjZzWKhF00oedf4jCCReLcmhLdhm1A27Q=
-github.com/yalp/jsonpath v0.0.0-20180802001716-5cc68e5049a0 h1:6fRhSjgLCkTD3JnJxvaJ4Sj+TYblw757bqYgZaOq5ZY=
 github.com/yalp/jsonpath v0.0.0-20180802001716-5cc68e5049a0/go.mod h1:/LWChgwKmvncFJFHJ7Gvn9wZArjbV5/FppcK2fKk/tI=
-github.com/yudai/gojsondiff v1.0.0 h1:27cbfqXLVEJ1o8I6v3y9lg8Ydm53EKqHXAOMxEGlCOA=
 github.com/yudai/gojsondiff v1.0.0/go.mod h1:AY32+k2cwILAkW1fbgxQ5mUmMiZFgLIV+FBNExI05xg=
-github.com/yudai/golcs v0.0.0-20170316035057-ecda9a501e82 h1:BHyfKlQyqbsFN5p3IfnEUduWvb9is428/nNb5L3U01M=
 github.com/yudai/golcs v0.0.0-20170316035057-ecda9a501e82/go.mod h1:lgjkn3NuSvDfVJdfcVVdX+jpBxNmX4rDAzaS45IcYoM=
-github.com/yudai/pp v2.0.1+incompatible h1:Q4//iY4pNF6yPLZIigmvcl7k/bPgrcTPIFIcmawg5bI=
 github.com/yudai/pp v2.0.1+incompatible/go.mod h1:PuxR/8QJ7cyCkFp/aUDS+JY727OFEZkTdatxwunjIkc=
-github.com/yuin/gopher-lua v0.0.0-20191220021717-ab39c6098bdb h1:ZkM6LRnq40pR1Ox0hTHlnpkcOTuFIDQpZ1IN8rKKhX0=
 github.com/yuin/gopher-lua v0.0.0-20191220021717-ab39c6098bdb/go.mod h1:gqRgreBUhTSL0GeU64rtZ3Uq3wtjOa/TB2YfrtkCbVQ=
 go.etcd.io/bbolt v1.3.2/go.mod h1:IbVyRI1SCnLcuJnV2u8VeU0CEYM7e686BmAb1XKL+uU=
 go.uber.org/atomic v1.4.0 h1:cxzIVoETapQEqDhQu3QfnvXAV4AlzcvUCxkVUFw3+EU=
@@ -420,17 +383,19 @@ golang.org/x/crypto v0.0.0-20171113213409-9f005a07e0d3/go.mod h1:6SG95UA2DQfeDnf
 golang.org/x/crypto v0.0.0-20180904163835-0709b304e793/go.mod h1:6SG95UA2DQfeDnfUPMdvaQW0Q7yPrPDi9nlGo2tz2b4=
 golang.org/x/crypto v0.0.0-20181203042331-505ab145d0a9/go.mod h1:6SG95UA2DQfeDnfUPMdvaQW0Q7yPrPDi9nlGo2tz2b4=
 golang.org/x/crypto v0.0.0-20190308221718-c2843e01d9a2/go.mod h1:djNgcEr1/C05ACkg1iLfiJU5Ep61QUkGW8qpdssI0+w=
+golang.org/x/crypto v0.0.0-20190325154230-a5d413f7728c/go.mod h1:djNgcEr1/C05ACkg1iLfiJU5Ep61QUkGW8qpdssI0+w=
 golang.org/x/crypto v0.0.0-20190611184440-5c40567a22f8/go.mod h1:yigFU9vqHzYiE8UmvKecakEJjdnWj3jj499lnFckfCI=
 golang.org/x/crypto v0.0.0-20190701094942-4def268fd1a4/go.mod h1:yigFU9vqHzYiE8UmvKecakEJjdnWj3jj499lnFckfCI=
 golang.org/x/crypto v0.0.0-20191011191535-87dc89f01550/go.mod h1:yigFU9vqHzYiE8UmvKecakEJjdnWj3jj499lnFckfCI=
 golang.org/x/crypto v0.0.0-20191112222119-e1110fd1c708/go.mod h1:LzIPMQfyMNhhGPhUkYOs5KpL4U8rLKemX1yGLhDgUto=
+golang.org/x/crypto v0.0.0-20191205180655-e7c4368fe9dd/go.mod h1:LzIPMQfyMNhhGPhUkYOs5KpL4U8rLKemX1yGLhDgUto=
 golang.org/x/crypto v0.0.0-20200115085410-6d4e4cb37c7d/go.mod h1:LzIPMQfyMNhhGPhUkYOs5KpL4U8rLKemX1yGLhDgUto=
 golang.org/x/crypto v0.0.0-20200221231518-2aa609cf4a9d h1:1ZiEyfaQIg3Qh0EoqpwAakHVhecoE5wlSg5GjnafJGw=
 golang.org/x/crypto v0.0.0-20200221231518-2aa609cf4a9d/go.mod h1:LzIPMQfyMNhhGPhUkYOs5KpL4U8rLKemX1yGLhDgUto=
 golang.org/x/lint v0.0.0-20181026193005-c67002cb31c3/go.mod h1:UVdnD1Gm6xHRNCYTkRU2/jEulfH38KcIWyp/GAMgvoE=
 golang.org/x/lint v0.0.0-20190313153728-d0100b6bd8b3/go.mod h1:6SW0HCj/g11FgYtHlgUYUwCkIfeOF89ocIRzGO/8vkc=
-golang.org/x/mod v0.1.1-0.20191105210325-c90efee705ee h1:WG0RUwxtNT4qqaXX3DPA8zHFNm/D9xaBpxzHt1WcA/E=
 golang.org/x/mod v0.1.1-0.20191105210325-c90efee705ee/go.mod h1:QqPTAvyqsEbceGzBzNggFXnrqF1CaUcvgkdR5Ot7KZg=
+golang.org/x/net v0.0.0-20180724234803-3673e40ba225/go.mod h1:mL1N/T3taQHkDXs73rZJwtUhF3w3ftmwwsq0BUmARs4=
 golang.org/x/net v0.0.0-20180826012351-8a410e7b638d/go.mod h1:mL1N/T3taQHkDXs73rZJwtUhF3w3ftmwwsq0BUmARs4=
 golang.org/x/net v0.0.0-20180906233101-161cd47e91fd/go.mod h1:mL1N/T3taQHkDXs73rZJwtUhF3w3ftmwwsq0BUmARs4=
 golang.org/x/net v0.0.0-20181005035420-146acd28ed58/go.mod h1:mL1N/T3taQHkDXs73rZJwtUhF3w3ftmwwsq0BUmARs4=
@@ -444,7 +409,6 @@ golang.org/x/net v0.0.0-20190522155817-f3200d17e092/go.mod h1:HSz+uSET+XFnRR8LxR
 golang.org/x/net v0.0.0-20190611141213-3f473d35a33a/go.mod h1:z5CRVTTTmAJ677TzLLGU+0bjPO0LkuOLi4/5GtJWs/s=
 golang.org/x/net v0.0.0-20190613194153-d28f0bde5980/go.mod h1:z5CRVTTTmAJ677TzLLGU+0bjPO0LkuOLi4/5GtJWs/s=
 golang.org/x/net v0.0.0-20190620200207-3b0461eec859/go.mod h1:z5CRVTTTmAJ677TzLLGU+0bjPO0LkuOLi4/5GtJWs/s=
-golang.org/x/net v0.0.0-20190827160401-ba9fcec4b297 h1:k7pJ2yAPLPgbskkFdhRCsA77k2fySZ1zf2zCjvQCiIM=
 golang.org/x/net v0.0.0-20190827160401-ba9fcec4b297/go.mod h1:z5CRVTTTmAJ677TzLLGU+0bjPO0LkuOLi4/5GtJWs/s=
 golang.org/x/net v0.0.0-20200301022130-244492dfa37a h1:GuSPYbZzB5/dcLNCwLQLsg3obCJtX9IJhpXkvY7kzk0=
 golang.org/x/net v0.0.0-20200301022130-244492dfa37a/go.mod h1:z5CRVTTTmAJ677TzLLGU+0bjPO0LkuOLi4/5GtJWs/s=
@@ -485,14 +449,13 @@ golang.org/x/tools v0.0.0-20190327201419-c70d86f8b7cf/go.mod h1:LCzVGOaR6xXOjkQ3
 golang.org/x/tools v0.0.0-20190328211700-ab21143f2384/go.mod h1:LCzVGOaR6xXOjkQ3onu1FJEFr0SW1gC7cKk1uF8kGRs=
 golang.org/x/tools v0.0.0-20190606050223-4d9ae51c2468/go.mod h1:/rFqwRUd4F7ZHNgwSSTFct+R/Kf4OFW1sUzUTQQTgfc=
 golang.org/x/tools v0.0.0-20190611222205-d73e1c7e250b/go.mod h1:/rFqwRUd4F7ZHNgwSSTFct+R/Kf4OFW1sUzUTQQTgfc=
-golang.org/x/tools v0.0.0-20190614205625-5aca471b1d59 h1:QjA/9ArTfVTLfEhClDCG7SGrZkZixxWpwNCDiwJfh88=
 golang.org/x/tools v0.0.0-20190614205625-5aca471b1d59/go.mod h1:/rFqwRUd4F7ZHNgwSSTFct+R/Kf4OFW1sUzUTQQTgfc=
 golang.org/x/tools v0.0.0-20200221224223-e1da425f72fd h1:hHkvGJK23seRCflePJnVa9IMv8fsuavSCWKd11kDQFs=
 golang.org/x/tools v0.0.0-20200221224223-e1da425f72fd/go.mod h1:TB2adYChydJhpapKDTa4BR/hXlZSLoq2Wpct/0txZ28=
 golang.org/x/xerrors v0.0.0-20191011141410-1b5146add898/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
-golang.org/x/xerrors v0.0.0-20191204190536-9bdfabe68543 h1:E7g+9GITq07hpfrRu66IVDexMakfv52eLZ2CXBWiKr4=
 golang.org/x/xerrors v0.0.0-20191204190536-9bdfabe68543/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
 google.golang.org/appengine v1.1.0/go.mod h1:EbEs0AVv82hx2wNQdGPgUI5lhzA/G0D9YwlJXL52JkM=
+google.golang.org/appengine v1.4.0/go.mod h1:xpcJRLb0r/rnEns0DIKYYv+WjYCduHsrkT7/EB5XEv4=
 google.golang.org/genproto v0.0.0-20180817151627-c66870c02cf8/go.mod h1:JiN7NxoALGmiZfu7CAH4rXhgtRTLTxftemlI0sWmxmc=
 google.golang.org/grpc v1.19.0/go.mod h1:mqu4LbDTu4XGKhr4mRzUsmM4RtVoemTSY81AxZiDr8c=
 google.golang.org/grpc v1.21.0/go.mod h1:oYelfM1adQP15Ek0mdvEgi9Df8B9CZIaU1084ijfRaM=
@@ -500,12 +463,9 @@ gopkg.in/airbrake/gobrake.v2 v2.0.9/go.mod h1:/h5ZAUhDkGaJfjzjKLSjv6zCL6O0LLBxU4
 gopkg.in/alecthomas/kingpin.v2 v2.2.6/go.mod h1:FMv+mEhP44yOT+4EoQTLFTRgOQ1FBLkstjWtayDeSgw=
 gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
 gopkg.in/check.v1 v1.0.0-20180628173108-788fd7840127/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
-gopkg.in/check.v1 v1.0.0-20190902080502-41f04d3bba15 h1:YR8cESwS4TdDjEe65xsg0ogRM/Nc3DYOhEAlW+xobZo=
 gopkg.in/check.v1 v1.0.0-20190902080502-41f04d3bba15/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
-gopkg.in/fsnotify.v1 v1.4.7 h1:xOHLXZwVvI9hhs+cLKq5+I5onOuwQLhQwiu63xxlHs4=
 gopkg.in/fsnotify.v1 v1.4.7/go.mod h1:Tz8NjZHkW78fSQdbUxIjBTcgA1z1m8ZHf0WmKUhAMys=
 gopkg.in/gemnasium/logrus-airbrake-hook.v2 v2.1.2/go.mod h1:Xk6kEKp8OKb+X14hQBKWaSkCsqBpgog8nAV2xsGOxlo=
-gopkg.in/go-playground/assert.v1 v1.2.1 h1:xoYuJVE7KT85PYWrN730RguIQO0ePzVRfFMXadIrXTM=
 gopkg.in/go-playground/assert.v1 v1.2.1/go.mod h1:9RXL0bg/zibRAgZUYszZSwO/z8Y/a8bDuhia5mkpMnE=
 gopkg.in/go-playground/validator.v8 v8.18.2/go.mod h1:RX2a/7Ha8BgOhfk7j780h4/u/RRjR0eouCJSH80/M2Y=
 gopkg.in/go-playground/validator.v9 v9.29.1 h1:SvGtYmN60a5CVKTOzMSyfzWDeZRxRuGvRQyEAKbw1xc=
@@ -514,7 +474,6 @@ gopkg.in/ini.v1 v1.51.0 h1:AQvPpx3LzTDM0AjnIRlVFwFFGC+npRopjZxLJj6gdno=
 gopkg.in/ini.v1 v1.51.0/go.mod h1:pNLf8WUiyNEtQjuu5G5vTm06TEv9tsIgeAvK8hOrP4k=
 gopkg.in/mgo.v2 v2.0.0-20180705113604-9856a29383ce/go.mod h1:yeKp02qBN3iKW1OzL3MGk2IdtZzaj7SFntXj72NppTA=
 gopkg.in/resty.v1 v1.12.0/go.mod h1:mDo4pnntr5jdWRML875a/NmxYqAlA73dVijT2AXvQQo=
-gopkg.in/tomb.v1 v1.0.0-20141024135613-dd632973f1e7 h1:uRGJdciOHaEIrze2W8Q3AKkepLTh2hOroT7a+7czfdQ=
 gopkg.in/tomb.v1 v1.0.0-20141024135613-dd632973f1e7/go.mod h1:dt/ZhP58zS4L8KSrWDmTeBkI65Dw0HsyUHuEVlX15mw=
 gopkg.in/yaml.v2 v2.0.0-20170812160011-eb3733d160e7/go.mod h1:JAlM8MvJe8wmxCU4Bli9HhUf9+ttbYbLASfIpnQbh74=
 gopkg.in/yaml.v2 v2.2.1/go.mod h1:hI93XBmqTisBFMUTm0b8Fm+jr3Dg1NNxqwp+5A1VGuI=
diff --git a/internal/init.go b/internal/init.go
index 699a7ac..bce4270 100644
--- a/internal/init.go
+++ b/internal/init.go
@@ -8,7 +8,7 @@ import (
 	"github.com/trustwallet/blockatlas/mq"
 	"github.com/trustwallet/blockatlas/pkg/ginutils"
 	"github.com/trustwallet/blockatlas/pkg/logger"
-	"github.com/trustwallet/blockatlas/storage"
+
 	"path/filepath"
 	"runtime"
 	"time"
@@ -31,15 +31,6 @@ func ParseArgs(defaultPort, defaultConfigPath string) (string, string) {
 	return port, confPath
 }
 
-func InitRedis(host string) *storage.Storage {
-	cache := storage.New()
-	err := cache.Init(host)
-	if err != nil {
-		logger.Fatal(err)
-	}
-	return cache
-}
-
 func InitConfig(confPath string) {
 	confPath, err := filepath.Abs(confPath)
 	if err != nil {
diff --git a/mq/mq.go b/mq/mq.go
index 1ad87c0..812fd90 100644
--- a/mq/mq.go
+++ b/mq/mq.go
@@ -3,8 +3,8 @@ package mq
 import (
 	"context"
 	"github.com/streadway/amqp"
+	"github.com/trustwallet/blockatlas/db"
 	"github.com/trustwallet/blockatlas/pkg/logger"
-	"github.com/trustwallet/blockatlas/storage"
 	"time"
 )
 
@@ -15,9 +15,10 @@ var (
 )
 
 type (
-	Queue          string
-	Consumer       func(amqp.Delivery, storage.Addresses)
-	MessageChannel <-chan amqp.Delivery
+	Queue              string
+	Consumer           func(amqp.Delivery)
+	ConsumerWithDbConn func(*db.Instance, amqp.Delivery)
+	MessageChannel     <-chan amqp.Delivery
 )
 
 const (
@@ -31,13 +32,10 @@ const (
 func Init(uri string) (err error) {
 	conn, err = amqp.Dial(uri)
 	if err != nil {
-		return
+		return err
 	}
 	amqpChan, err = conn.Channel()
-	if err != nil {
-		return
-	}
-	return
+	return err
 }
 
 func Close() {
@@ -45,6 +43,10 @@ func Close() {
 	conn.Close()
 }
 
+func (mc MessageChannel) GetMessage() amqp.Delivery {
+	return <-mc
+}
+
 func (q Queue) Declare() error {
 	_, err := amqpChan.QueueDeclare(string(q), true, false, false, false, nil)
 	return err
@@ -58,7 +60,7 @@ func (q Queue) Publish(body []byte) error {
 	})
 }
 
-func (q Queue) RunConsumerForChannelWithCancel(consumer Consumer, messageChannel MessageChannel, cache storage.Addresses, ctx context.Context) {
+func RunConsumerForChannelWithCancelAndDbConn(consumer ConsumerWithDbConn, messageChannel MessageChannel, database *db.Instance, ctx context.Context) {
 	for {
 		select {
 		case <-ctx.Done():
@@ -68,12 +70,12 @@ func (q Queue) RunConsumerForChannelWithCancel(consumer Consumer, messageChannel
 			if message.Body == nil {
 				continue
 			}
-			go consumer(message, cache)
+			go consumer(database, message)
 		}
 	}
 }
 
-func (q Queue) RunConsumer(consumer Consumer, cache storage.Addresses) {
+func (q Queue) GetMessageChannel() MessageChannel {
 	messageChannel, err := amqpChan.Consume(
 		string(q),
 		"",
@@ -85,12 +87,6 @@ func (q Queue) RunConsumer(consumer Consumer, cache storage.Addresses) {
 	)
 	if err != nil {
 		logger.Fatal("MQ issue " + err.Error())
-		return
-	}
-
-	if PrefetchCount < minPrefetchCount {
-		logger.Info("Change prefetch count to default")
-		PrefetchCount = defaultPrefetchCount
 	}
 
 	err = amqpChan.Qos(
@@ -98,46 +94,22 @@ func (q Queue) RunConsumer(consumer Consumer, cache storage.Addresses) {
 		0,
 		true,
 	)
-
 	if err != nil {
-		logger.Error("no qos limit ", err)
+		logger.Fatal("No qos limit ", err)
 	}
 
-	for data := range messageChannel {
-		go consumer(data, cache)
-	}
+	return messageChannel
 }
 
-func (q Queue) RunConsumerWithCancel(consumer Consumer, cache storage.Addresses, ctx context.Context) {
-	messageChannel, err := amqpChan.Consume(
-		string(q),
-		"",
-		false,
-		false,
-		false,
-		false,
-		nil,
-	)
-	if err != nil {
-		logger.Fatal("MQ issue " + err.Error())
-		return
-	}
-
-	if PrefetchCount < minPrefetchCount {
-		logger.Info("Change prefetch count to default")
-		PrefetchCount = defaultPrefetchCount
-	}
-
-	err = amqpChan.Qos(
-		PrefetchCount,
-		0,
-		true,
-	)
-
-	if err != nil {
-		logger.Error("no qos limit ", err)
+func (q Queue) RunConsumer(consumer Consumer) {
+	messageChannel := q.GetMessageChannel()
+	for data := range messageChannel {
+		go consumer(data)
 	}
+}
 
+func (q Queue) RunConsumerWithCancel(consumer Consumer, ctx context.Context) {
+	messageChannel := q.GetMessageChannel()
 	for {
 		select {
 		case <-ctx.Done():
@@ -147,30 +119,25 @@ func (q Queue) RunConsumerWithCancel(consumer Consumer, cache storage.Addresses,
 			if message.Body == nil {
 				continue
 			}
-			go consumer(message, cache)
+			go consumer(message)
 		}
 	}
 }
 
-func (q Queue) GetMessageChannel() MessageChannel {
-	messageChannel, err := amqpChan.Consume(
-		string(q),
-		"",
-		false,
-		false,
-		false,
-		false,
-		nil,
-	)
-	if err != nil {
-		logger.Fatal("MQ issue " + err.Error())
+func (q Queue) RunConsumerWithCancelAndDbConn(consumer ConsumerWithDbConn, database *db.Instance, ctx context.Context) {
+	messageChannel := q.GetMessageChannel()
+	for {
+		select {
+		case <-ctx.Done():
+			logger.Info("Consumer stopped")
+			return
+		case message := <-messageChannel:
+			if message.Body == nil {
+				continue
+			}
+			go consumer(database, message)
+		}
 	}
-
-	return messageChannel
-}
-
-func (mc MessageChannel) GetMessage() amqp.Delivery {
-	return <-mc
 }
 
 func RestoreConnectionWorker(uri string, queue Queue, timeout time.Duration) {
@@ -200,7 +167,7 @@ func RestoreConnectionWorker(uri string, queue Queue, timeout time.Duration) {
 }
 
 func FatalWorker(timeout time.Duration) {
-	logger.Info("Run FatalWorker")
+	logger.Info("Run MQ FatalWorker")
 	for {
 		if conn.IsClosed() {
 			logger.Fatal("MQ is not available now")
diff --git a/pkg/blockatlas/clientcache.go b/pkg/blockatlas/clientcache.go
index 12ace94..6c376f4 100644
--- a/pkg/blockatlas/clientcache.go
+++ b/pkg/blockatlas/clientcache.go
@@ -7,7 +7,6 @@ import (
 	"github.com/patrickmn/go-cache"
 	"github.com/trustwallet/blockatlas/pkg/errors"
 	"github.com/trustwallet/blockatlas/pkg/logger"
-	"github.com/trustwallet/blockatlas/pkg/storage/util"
 	"net/url"
 	"strings"
 	"sync"
@@ -85,7 +84,7 @@ func (mc *memCache) getCache(key string, value interface{}) error {
 	}
 	err := json.Unmarshal(r, value)
 	if err != nil {
-		return errors.E(err, util.ErrNotFound)
+		return errors.E(err, "not found")
 	}
 	return nil
 }
diff --git a/pkg/blockatlas/observer.go b/pkg/blockatlas/observer.go
index 168c94e..da61b1b 100644
--- a/pkg/blockatlas/observer.go
+++ b/pkg/blockatlas/observer.go
@@ -8,16 +8,15 @@ type (
 	SubscriptionOperation string
 
 	SubscriptionEvent struct {
-		NewSubscriptions Subscriptions         `json:"new_subscriptions"`
-		OldSubscriptions Subscriptions         `json:"old_subscriptions"`
-		GUID             string                `json:"guid"`
-		Operation        SubscriptionOperation `json:"operation"`
+		Subscriptions Subscriptions         `json:"subscriptions"`
+		Id            uint                  `json:"id"`
+		Operation     SubscriptionOperation `json:"operation"`
 	}
 
 	Subscription struct {
 		Coin    uint   `json:"coin"`
 		Address string `json:"address"`
-		GUID    string `json:"guid"`
+		Id      uint   `json:"id"`
 	}
 
 	CoinStatus struct {
@@ -42,7 +41,7 @@ func (e *SubscriptionEvent) ParseSubscriptions(s Subscriptions) []Subscription {
 			subs = append(subs, Subscription{
 				Coin:    uint(coin),
 				Address: addr,
-				GUID:    e.GUID,
+				Id:      e.Id,
 			})
 		}
 	}
diff --git a/pkg/blockatlas/tx.go b/pkg/blockatlas/tx.go
index 49f435c..09cbc57 100644
--- a/pkg/blockatlas/tx.go
+++ b/pkg/blockatlas/tx.go
@@ -56,12 +56,12 @@ const (
 
 type (
 	// Types of transaction statuses
-	Direction        string
-	Status           string
-	TokenType        string
-	TransactionType  string
-	KeyType          string
-	KeyTitle         string
+	Direction       string
+	Status          string
+	TokenType       string
+	TransactionType string
+	KeyType         string
+	KeyTitle        string
 
 	Block struct {
 		Number int64  `json:"number"`
diff --git a/pkg/ginutils/gincache/cache.go b/pkg/ginutils/gincache/cache.go
index 1de5fca..43474e1 100644
--- a/pkg/ginutils/gincache/cache.go
+++ b/pkg/ginutils/gincache/cache.go
@@ -10,7 +10,6 @@ import (
 	"github.com/patrickmn/go-cache"
 	"github.com/trustwallet/blockatlas/pkg/errors"
 	"github.com/trustwallet/blockatlas/pkg/logger"
-	"github.com/trustwallet/blockatlas/pkg/storage/util"
 	"io/ioutil"
 	"net/http"
 	"sync"
@@ -135,7 +134,7 @@ func (mc *memCache) getCache(key string) (cacheResponse, error) {
 	}
 	err := json.Unmarshal(r, &result)
 	if err != nil {
-		return result, errors.E(err, util.ErrNotFound)
+		return result, errors.E(err, "not found")
 	}
 	return result, nil
 }
diff --git a/pkg/storage/redis/hmap.go b/pkg/storage/redis/hmap.go
deleted file mode 100644
index a5c4691..0000000
--- a/pkg/storage/redis/hmap.go
+++ /dev/null
@@ -1,51 +0,0 @@
-package redis
-
-import (
-	"encoding/json"
-	"github.com/trustwallet/blockatlas/pkg/errors"
-	"github.com/trustwallet/blockatlas/pkg/storage/util"
-)
-
-func (db *Redis) GetAllHM(entity string) (map[string]string, error) {
-	cmd := db.client.HGetAll(entity)
-	if cmd.Err() != nil {
-		return nil, errors.E(cmd.Err(), util.ErrNotFound)
-	}
-	return cmd.Val(), nil
-}
-
-func (db *Redis) GetHMValue(entity, key string, value interface{}) error {
-	cmd := db.client.HMGet(entity, key)
-	if cmd.Err() != nil {
-		return errors.E(cmd.Err(), util.ErrNotFound)
-	}
-	val, ok := cmd.Val()[0].(string)
-	if !ok {
-		return errors.E(util.ErrNotFound)
-	}
-	err := json.Unmarshal([]byte(val), value)
-	if err != nil {
-		return errors.E(err, util.ErrNotFound)
-	}
-	return nil
-}
-
-func (db *Redis) AddHM(entity, key string, value interface{}) error {
-	j, err := json.Marshal(value)
-	if err != nil {
-		return errors.E(err, errors.Params{"key": key})
-	}
-	cmd := db.client.HMSet(entity, map[string]interface{}{key: j})
-	if cmd.Err() != nil {
-		return errors.E(cmd.Err(), util.ErrNotStored)
-	}
-	return nil
-}
-
-func (db *Redis) DeleteHM(entity, key string) error {
-	cmd := db.client.HDel(entity, key)
-	if cmd.Err() != nil {
-		return errors.E(cmd.Err(), util.ErrNotDeleted)
-	}
-	return nil
-}
diff --git a/pkg/storage/redis/redis.go b/pkg/storage/redis/redis.go
deleted file mode 100644
index 979ae2a..0000000
--- a/pkg/storage/redis/redis.go
+++ /dev/null
@@ -1,64 +0,0 @@
-package redis
-
-import (
-	"encoding/json"
-	"github.com/go-redis/redis"
-	"github.com/trustwallet/blockatlas/pkg/errors"
-	"github.com/trustwallet/blockatlas/pkg/storage/util"
-)
-
-type Redis struct {
-	client *redis.Client
-}
-
-func (db *Redis) Init(host string) error {
-	options, err := redis.ParseURL(host)
-	if err != nil {
-		return errors.E(err, "Cannot connect to Redis")
-	}
-	client := redis.NewClient(options)
-	if err := client.Ping().Err(); err != nil {
-		return errors.E(err, "Redis connection test failed")
-	}
-	db.client = client
-	return nil
-}
-
-func (db *Redis) GetValue(key string, value interface{}) error {
-	cmd := db.client.Get(key)
-	if cmd.Err() != nil {
-		return errors.E(cmd.Err(), util.ErrNotFound)
-	}
-	err := json.Unmarshal([]byte(cmd.Val()), value)
-	if err != nil {
-		return errors.E(err, util.ErrNotFound)
-	}
-	return nil
-}
-
-func (db *Redis) Add(key string, value interface{}) error {
-	j, err := json.Marshal(value)
-	if err != nil {
-		return errors.E(err, errors.Params{"key": key})
-	}
-	cmd := db.client.Set(key, j, 0)
-	if cmd.Err() != nil {
-		return errors.E(cmd.Err(), util.ErrNotStored)
-	}
-	return nil
-}
-
-func (db *Redis) Delete(key string) error {
-	cmd := db.client.Del(key)
-	if cmd.Err() != nil {
-		return errors.E(cmd.Err(), util.ErrNotDeleted)
-	}
-	return nil
-}
-
-func (db *Redis) IsReady() bool {
-	if db.client == nil {
-		return false
-	}
-	return db.client.Ping().Err() == nil
-}
diff --git a/pkg/storage/util/errors.go b/pkg/storage/util/errors.go
deleted file mode 100644
index 162401f..0000000
--- a/pkg/storage/util/errors.go
+++ /dev/null
@@ -1,14 +0,0 @@
-package util
-
-import (
-	"github.com/trustwallet/blockatlas/pkg/errors"
-)
-
-var (
-	ErrNotFound     = errors.E("storage: obj not found")
-	ErrNotStored    = errors.E("storage: not stored")
-	ErrNotUpdated   = errors.E("storage: not updated")
-	ErrNotDeleted   = errors.E("storage: not deleted")
-	ErrInvalidType  = errors.E("storage: invalid type")
-	ErrAlreadyExist = errors.E("storage: object already exist")
-)
diff --git a/platform/binance/model.go b/platform/binance/model.go
index d54d9e5..8523d09 100644
--- a/platform/binance/model.go
+++ b/platform/binance/model.go
@@ -104,7 +104,7 @@ func (subTxs *SubTxs) getTxs() (txs []Tx) {
 		if err != nil {
 			subTxValue = 0
 		}
-		value := numbers.Float64toString(txValue+subTxValue)
+		value := numbers.Float64toString(txValue + subTxValue)
 		tx.Value = json.Number(value)
 		mapTx[key] = tx
 	}
diff --git a/platform/binance/transaction.go b/platform/binance/transaction.go
index 289fe6f..ba8ab21 100644
--- a/platform/binance/transaction.go
+++ b/platform/binance/transaction.go
@@ -65,7 +65,7 @@ func normalizeTransfer(tx blockatlas.Tx, srcTx Tx, token, address string) (block
 		txs := make(blockatlas.TxPage, 0)
 		// Parse all assets as a transaction
 		for _, subTx := range srcTx.SubTxsDto.SubTxDtoList.getTxs() {
-			// If this is not called from a block observer, only get the user txs/assets
+			// If this is not called from a block observer_test, only get the user txs/assets
 			if !subTx.containAddress(address) {
 				continue
 			}
diff --git a/platform/tezos/model.go b/platform/tezos/model.go
index 970c08c..451b39c 100644
--- a/platform/tezos/model.go
+++ b/platform/tezos/model.go
@@ -109,3 +109,11 @@ func (t *Transaction) Direction(address string) blockatlas.Direction {
 
 	return blockatlas.DirectionIncoming
 }
+
+func (t *Transaction) GetReceiver() string {
+	if t.Receiver != "" {
+		return t.Receiver
+	} else {
+		return t.Delegate
+	}
+}
diff --git a/platform/tezos/transaction.go b/platform/tezos/transaction.go
index 415b771..73347a0 100644
--- a/platform/tezos/transaction.go
+++ b/platform/tezos/transaction.go
@@ -36,17 +36,18 @@ func NormalizeTx(srcTx Transaction, address string) (blockatlas.Tx, bool) {
 		Coin:   coin.XTZ,
 		Date:   srcTx.BlockTimestamp(),
 		Error:  srcTx.ErrorMsg(),
-		Fee:    blockatlas.Amount(numbers.Float64toString(srcTx.Fee)),
+		Fee:    blockatlas.Amount(numbers.DecimalExp(numbers.Float64toString(srcTx.Fee), 6)),
 		From:   srcTx.Sender,
 		ID:     srcTx.Hash,
 		Status: srcTx.Status(),
-		To:     srcTx.Receiver,
+		To:     srcTx.GetReceiver(),
 		Type:   srcTx.TransferType(),
 	}
 	if address != "" {
 		tx.Direction = srcTx.Direction(address)
 	}
 
+	value := blockatlas.Amount(numbers.DecimalExp(numbers.Float64toString(srcTx.Volume), 6))
 	switch srcTx.TransferType() {
 	case blockatlas.TxAnyAction:
 		tx.Meta = blockatlas.AnyAction{
@@ -56,10 +57,11 @@ func NormalizeTx(srcTx Transaction, address string) (blockatlas.Tx, bool) {
 			Name:     coin.Tezos().Name,
 			Symbol:   coin.Tezos().Symbol,
 			Decimals: coin.Tezos().Decimals,
+			Value:    value,
 		}
 	case blockatlas.TxTransfer:
 		tx.Meta = blockatlas.Transfer{
-			Value:    blockatlas.Amount(numbers.Float64toString(srcTx.Volume)),
+			Value:    value,
 			Symbol:   coin.Coins[coin.XTZ].Symbol,
 			Decimals: coin.Coins[coin.XTZ].Decimals,
 		}
diff --git a/scripts/.goreleaser.yml b/scripts/.goreleaser.yml
index a9ea728..f183aab 100644
--- a/scripts/.goreleaser.yml
+++ b/scripts/.goreleaser.yml
@@ -12,9 +12,9 @@ builds:
       - amd64
     env:
       - CGO_ENABLED=0
-  - id: observer
-    binary: observer
-    main: "./cmd/observer/main.go"
+  - id: observer_test
+    binary: observer_test
+    main: "./cmd/observer_test/main.go"
     goos:
       - linux
     goarch:
diff --git a/services/observer/healthcheck/healthcheck.go b/services/observer/healthcheck/healthcheck.go
deleted file mode 100644
index 1fbde8a..0000000
--- a/services/observer/healthcheck/healthcheck.go
+++ /dev/null
@@ -1,112 +0,0 @@
-package healthcheck
-
-import (
-	"github.com/trustwallet/blockatlas/pkg/blockatlas"
-	"github.com/trustwallet/blockatlas/pkg/logger"
-	"github.com/trustwallet/blockatlas/storage"
-	"sync"
-	"time"
-)
-
-var observerStatus status
-
-func init() {
-	observerStatus = status{
-		m: make(map[string]Info),
-	}
-}
-
-type status struct {
-	m map[string]Info
-	sync.RWMutex
-}
-
-func (cr *status) update(coin string, info Info) {
-	cr.Lock()
-	cr.m[coin] = info
-	cr.Unlock()
-}
-
-func (cr *status) get() map[string]Info {
-	cr.RLock()
-	defer cr.RUnlock()
-	return cr.m
-}
-
-func Worker(storage storage.Tracker, api blockatlas.BlockAPI) {
-	var (
-		lastParsedBlock, currentBlock int64
-		err                           error
-	)
-
-	lastParsedBlock, err = storage.GetLastParsedBlockNumber(api.Coin().ID)
-	if err != nil {
-		logger.Fatal(err)
-	}
-
-	var duration time.Duration
-	t := api.Coin().BlockTime / 1000
-
-	if t > 30 {
-		duration = time.Duration(int64(time.Second) * int64(t))
-	} else {
-		duration = time.Minute * 11
-	}
-	logger.Info("Setting update duration", logger.Params{"duration": duration})
-
-	for {
-		currentBlock, err = storage.GetLastParsedBlockNumber(api.Coin().ID)
-		if err != nil {
-			logger.Error(err)
-			continue
-		}
-
-		latestBlock, err := api.CurrentBlockNumber()
-		if err != nil {
-			logger.Error(err)
-			continue
-		}
-
-		logger.Info("fetched", logger.Params{"currentBlock": currentBlock, "lastParsedBlock": lastParsedBlock, "latestBlock": latestBlock, "handle": api.Coin().Handle})
-
-		if currentBlock > lastParsedBlock || latestBlock == currentBlock {
-			lastParsedBlock = currentBlock
-			observerStatus.update(api.Coin().Handle, Info{
-				LastParsedBlock: lastParsedBlock,
-				Healthy:         true,
-			})
-		} else {
-			observerStatus.update(api.Coin().Handle, Info{
-				LastParsedBlock: lastParsedBlock,
-				Healthy:         false,
-			})
-		}
-
-		time.Sleep(duration)
-	}
-}
-
-func GetStatus(exclude []string) map[string]interface{} {
-	total := true
-	result := make(map[string]interface{})
-	status := observerStatus.get()
-	for k, v := range status{
-		if !contain(k, exclude){
-			if v.Healthy == false{
-				total = false
-			}
-			result[k] = v
-		}
-	}
-	result["total"] = total
-	return result
-}
-
-func contain(elem string, list []string) bool{
-	for _, v := range list{
-		if v == elem{
-			return true
-		}
-	}
-	return false
-}
diff --git a/services/observer/healthcheck/model.go b/services/observer/healthcheck/model.go
deleted file mode 100644
index 6a4fa94..0000000
--- a/services/observer/healthcheck/model.go
+++ /dev/null
@@ -1,6 +0,0 @@
-package healthcheck
-
-type Info struct {
-	LastParsedBlock int64 `json:"lastParsedBlock"`
-	Healthy         bool  `json:"healthy"`
-}
diff --git a/services/observer/notifier/notifier.go b/services/observer/notifier/notifier.go
index 0f75260..16e97ff 100644
--- a/services/observer/notifier/notifier.go
+++ b/services/observer/notifier/notifier.go
@@ -3,12 +3,12 @@ package notifier
 import (
 	"encoding/json"
 	"github.com/streadway/amqp"
+	"github.com/trustwallet/blockatlas/db"
 	"github.com/trustwallet/blockatlas/mq"
 	"github.com/trustwallet/blockatlas/pkg/blockatlas"
 	"github.com/trustwallet/blockatlas/pkg/errors"
 	"github.com/trustwallet/blockatlas/pkg/logger"
 	"github.com/trustwallet/blockatlas/pkg/numbers"
-	"github.com/trustwallet/blockatlas/storage"
 	"sync"
 	"time"
 )
@@ -16,10 +16,10 @@ import (
 type DispatchEvent struct {
 	Action blockatlas.TransactionType `json:"action"`
 	Result *blockatlas.Tx             `json:"result"`
-	GUID   string                     `json:"guid"`
+	Id     uint                       `json:"id"`
 }
 
-func RunNotifier(delivery amqp.Delivery, s storage.Addresses) {
+func RunNotifier(database *db.Instance, delivery amqp.Delivery) {
 	defer func() {
 		if err := delivery.Ack(false); err != nil {
 			logger.Error(err)
@@ -42,15 +42,18 @@ func RunNotifier(delivery amqp.Delivery, s storage.Addresses) {
 	}
 
 	addresses := blockTransactions.GetUniqueAddresses()
-	subs, err := s.FindSubscriptions(txs[0].Coin, addresses)
-	if err != nil || len(subs) == 0 {
+	subscriptionsDataList, err := database.GetSubscriptionData(txs[0].Coin, addresses)
+	if err != nil || len(subscriptionsDataList) == 0 {
 		return
 	}
 
 	var wg sync.WaitGroup
-	wg.Add(len(subs))
-	for _, sub := range subs {
-		go buildAndPostMessage(blockTransactions, sub, &wg)
+	wg.Add(len(subscriptionsDataList))
+	for _, data := range subscriptionsDataList {
+		go buildAndPostMessage(
+			blockTransactions,
+			blockatlas.Subscription{Coin: data.Coin, Address: data.Address, Id: data.SubscriptionId},
+			&wg)
 	}
 	wg.Wait()
 }
@@ -67,30 +70,30 @@ func buildAndPostMessage(blockTransactions blockatlas.TxSetMap, sub blockatlas.S
 		action := DispatchEvent{
 			Action: tx.Type,
 			Result: &tx,
-			GUID:   sub.GUID,
+			Id:     sub.Id,
 		}
 		txJson, err := json.Marshal(action)
 		if err != nil {
-			logger.Panic(err)
+			logger.Panic(err, logger.Params{"coin": tx.Coin})
 		}
 
 		logParams := logger.Params{
-			"guid": sub.GUID,
+			"Id":   sub.Id,
 			"coin": sub.Coin,
 			"txID": tx.ID,
 		}
 
-		publishTransaction(sub.GUID, txJson, logParams)
+		publishTransaction(sub.Id, txJson, logParams)
 	}
 }
 
-func publishTransaction(message string, rawMessage []byte, logParams logger.Params) {
+func publishTransaction(id uint, rawMessage []byte, logParams logger.Params) {
 	err := mq.Transactions.Publish(rawMessage)
 	if err != nil {
-		err = errors.E(err, "Failed to dispatch event", errors.Params{"message": message}, logParams)
-		logger.Fatal(err, logger.Params{"message": message}, logParams)
+		err = errors.E(err, "Failed to dispatch event", errors.Params{"id": id}, logParams)
+		logger.Fatal(err, logger.Params{"id": id}, logParams)
 	}
-	logger.Info("Message dispatched", logger.Params{"message": message}, logParams)
+	logger.Info("Message dispatched", logger.Params{"id": id}, logParams)
 }
 
 func GetInterval(value int, minInterval, maxInterval time.Duration) time.Duration {
diff --git a/services/observer/parser/parser.go b/services/observer/parser/parser.go
index bcec764..10d71a2 100644
--- a/services/observer/parser/parser.go
+++ b/services/observer/parser/parser.go
@@ -4,13 +4,13 @@ import (
 	"context"
 	"encoding/json"
 	"fmt"
+	"github.com/trustwallet/blockatlas/db"
 	"github.com/trustwallet/blockatlas/mq"
 	"github.com/trustwallet/blockatlas/pkg/blockatlas"
 	"github.com/trustwallet/blockatlas/pkg/errors"
 	"sync/atomic"
 
 	"github.com/trustwallet/blockatlas/pkg/logger"
-	"github.com/trustwallet/blockatlas/storage"
 	"math/rand"
 	"sort"
 	"sync"
@@ -21,13 +21,13 @@ type (
 	Params struct {
 		Ctx                   context.Context
 		Api                   blockatlas.BlockAPI
-		Storage               storage.Tracker
 		Queue                 mq.Queue
 		ParsingBlocksInterval time.Duration
 		BacklogCount          int
 		MaxBacklogBlocks      int64
 		StopChannel           chan<- struct{}
 		TxBatchLimit          uint
+		Database              *db.Instance
 	}
 
 	GetBlockByNumber func(num int64) (*blockatlas.Block, error)
@@ -79,7 +79,7 @@ func RunParser(params Params) {
 }
 
 func GetBlocksIntervalToFetch(params Params) (int64, int64, error) {
-	lastParsedBlock, err := params.Storage.GetLastParsedBlockNumber(params.Api.Coin().ID)
+	lastParsedBlock, err := params.Database.GetLastParsedBlockNumber(params.Api.Coin().ID)
 	if err != nil {
 		return 0, 0, errors.E(err, "Polling failed: tracker didn't return last known block number")
 	}
@@ -174,8 +174,7 @@ func SaveLastParsedBlock(params Params, blocks []blockatlas.Block) error {
 	})
 
 	lastBlockNumber := blocks[len(blocks)-1].Number
-
-	err := params.Storage.SetLastParsedBlockNumber(params.Api.Coin().ID, lastBlockNumber)
+	err := params.Database.SetLastParsedBlockNumber(params.Api.Coin().ID, lastBlockNumber)
 	if err != nil {
 		return err
 	}
diff --git a/services/observer/subscriber/subscriber.go b/services/observer/subscriber/subscriber.go
index dc9cc4a..40ab290 100644
--- a/services/observer/subscriber/subscriber.go
+++ b/services/observer/subscriber/subscriber.go
@@ -3,9 +3,10 @@ package subscriber
 import (
 	"encoding/json"
 	"github.com/streadway/amqp"
+	"github.com/trustwallet/blockatlas/db"
+	"github.com/trustwallet/blockatlas/db/models"
 	"github.com/trustwallet/blockatlas/pkg/blockatlas"
 	"github.com/trustwallet/blockatlas/pkg/logger"
-	"github.com/trustwallet/blockatlas/storage"
 )
 
 const (
@@ -14,36 +15,34 @@ const (
 	UpdateSubscription blockatlas.SubscriptionOperation = "UpdateSubscription"
 )
 
-func RunSubscriber(delivery amqp.Delivery, storage storage.Addresses) {
+func RunSubscriber(database *db.Instance, delivery amqp.Delivery) {
 	var event blockatlas.SubscriptionEvent
 	err := json.Unmarshal(delivery.Body, &event)
 	if err != nil {
 		logger.Fatal(err)
 	}
-	newSubscriptions := event.ParseSubscriptions(event.NewSubscriptions)
-	oldSubscriptions := event.ParseSubscriptions(event.OldSubscriptions)
 
-	params := logger.Params{"operation": event.Operation, "guid": event.GUID, "new_subscriptions_len": len(newSubscriptions), "old_subscriptions_len": len(oldSubscriptions)}
+	subscriptions := event.ParseSubscriptions(event.Subscriptions)
+
+	params := logger.Params{"operation": event.Operation, "id": event.Id, "subscriptions_len": len(subscriptions)}
+
+	id := event.Id
 
 	switch event.Operation {
 	case UpdateSubscription:
-		err := storage.DeleteSubscriptions(oldSubscriptions)
-		if err != nil {
-			logger.Error(err, params)
-		}
-		err = storage.AddSubscriptions(newSubscriptions)
+		err := database.AddToExistingSubscription(id, ToSubscriptionData(subscriptions))
 		if err != nil {
 			logger.Error(err, params)
 		}
 		logger.Info("Updated", params)
 	case AddSubscription:
-		err = storage.AddSubscriptions(newSubscriptions)
+		err = database.AddSubscriptions(id, ToSubscriptionData(subscriptions))
 		if err != nil {
 			logger.Error(err, params)
 		}
 		logger.Info("Added", params)
 	case DeleteSubscription:
-		err := storage.DeleteSubscriptions(oldSubscriptions)
+		err := database.DeleteAllSubscriptions(id)
 		if err != nil {
 			logger.Error(err, params)
 		}
@@ -55,3 +54,11 @@ func RunSubscriber(delivery amqp.Delivery, storage storage.Addresses) {
 		logger.Error(err, params)
 	}
 }
+
+func ToSubscriptionData(sub []blockatlas.Subscription) []models.SubscriptionData {
+	data := make([]models.SubscriptionData, 0, len(sub))
+	for _, s := range sub {
+		data = append(data, models.SubscriptionData{Coin: s.Coin, Address: s.Address, SubscriptionId: s.Id})
+	}
+	return data
+}
diff --git a/storage/addresses.go b/storage/addresses.go
deleted file mode 100644
index 8a3a1cf..0000000
--- a/storage/addresses.go
+++ /dev/null
@@ -1,147 +0,0 @@
-package storage
-
-import (
-	"fmt"
-	"github.com/trustwallet/blockatlas/pkg/blockatlas"
-	"github.com/trustwallet/blockatlas/pkg/errors"
-	"sync"
-)
-
-const (
-	ATLAS_OBSERVER = "ATLAS_OBSERVER"
-)
-
-type SubscriptionOperation func(sub blockatlas.Subscription, wg *sync.WaitGroup, errorsChan chan<- error)
-
-func (s *Storage) RunOperation(subscriptions []blockatlas.Subscription, operation SubscriptionOperation) error {
-	var (
-		errorsChan = make(chan error, len(subscriptions))
-		wg         sync.WaitGroup
-	)
-
-	for _, sub := range subscriptions {
-		wg.Add(1)
-		go operation(sub, &wg, errorsChan)
-	}
-	wg.Wait()
-	close(errorsChan)
-
-	if len(errorsChan) != 0 {
-		var errorsStr string
-		for err := range errorsChan {
-			errorsStr += err.Error() + " "
-		}
-		return errors.E(errorsStr)
-	}
-
-	return nil
-}
-
-func (s *Storage) DeleteSubscriptions(subscriptions []blockatlas.Subscription) error {
-	return s.RunOperation(subscriptions, s.deleteSubscriptions)
-}
-
-func (s *Storage) AddSubscriptions(subscriptions []blockatlas.Subscription) error {
-	return s.RunOperation(subscriptions, s.addSubscriptions)
-}
-
-func (s *Storage) FindSubscriptions(coin uint, addresses []string) ([]blockatlas.Subscription, error) {
-	if len(addresses) == 0 {
-		return nil, errors.E("cannot look up an empty list")
-	}
-
-	observersSliceChan := make(chan []blockatlas.Subscription, len(addresses))
-	observers := make([]blockatlas.Subscription, 0, len(addresses))
-
-	var wg sync.WaitGroup
-	wg.Add(len(addresses))
-
-	for _, address := range addresses {
-		go s.findSubscriptionsByAddress(coin, address, observersSliceChan, &wg)
-	}
-	wg.Wait()
-	close(observersSliceChan)
-
-	for slice := range observersSliceChan {
-		for _, v := range slice {
-			observers = append(observers, v)
-		}
-	}
-
-	return observers, nil
-}
-
-func (s *Storage) findSubscriptionsByAddress(coin uint, address string, sub chan<- []blockatlas.Subscription, wg *sync.WaitGroup) {
-	defer wg.Done()
-	key := getSubscriptionKey(coin, address)
-	var guids []string
-	err := s.GetHMValue(ATLAS_OBSERVER, key, &guids)
-	if err != nil {
-		return
-	}
-
-	result := make([]blockatlas.Subscription, 0, len(guids))
-
-	for _, guid := range guids {
-		result = append(result, blockatlas.Subscription{Coin: coin, Address: address, GUID: guid})
-	}
-
-	sub <- result
-}
-
-func (s *Storage) deleteSubscriptions(sub blockatlas.Subscription, wg *sync.WaitGroup, errorsChan chan<- error) {
-	defer wg.Done()
-	key := getSubscriptionKey(sub.Coin, sub.Address)
-	var guids []string
-	err := s.GetHMValue(ATLAS_OBSERVER, key, &guids)
-	if err != nil {
-		errorsChan <- err
-		return
-	}
-	newHooks := make([]string, 0)
-	for _, guid := range guids {
-		if guid == sub.GUID {
-			continue
-		}
-		newHooks = append(newHooks, guid)
-	}
-	if len(newHooks) == 0 {
-		_ = s.DeleteHM(ATLAS_OBSERVER, key)
-		return
-	}
-	err = s.AddHM(ATLAS_OBSERVER, key, newHooks)
-	if err != nil {
-		errorsChan <- err
-	}
-}
-
-func (s *Storage) addSubscriptions(sub blockatlas.Subscription, wg *sync.WaitGroup, errorsChan chan<- error) {
-	defer wg.Done()
-	key := getSubscriptionKey(sub.Coin, sub.Address)
-	var guids []string
-	s.GetHMValue(ATLAS_OBSERVER, key, &guids)
-	if guids == nil {
-		guids = make([]string, 0)
-	}
-	if hasObject(guids, sub.GUID) {
-		return
-	}
-	guids = append(guids, sub.GUID)
-	err := s.AddHM(ATLAS_OBSERVER, key, guids)
-	if err != nil {
-		errorsChan <- err
-	}
-}
-
-func getSubscriptionKey(coin uint, address string) string {
-	return fmt.Sprintf("%d-%s", coin, address)
-}
-
-func hasObject(array []string, obj string) bool {
-	for _, temp := range array {
-		if temp == obj {
-			return true
-		}
-	}
-	return false
-}
diff --git a/storage/storage.go b/storage/storage.go
deleted file mode 100644
index 2849776..0000000
--- a/storage/storage.go
+++ /dev/null
@@ -1,51 +0,0 @@
-package storage
-
-import (
-	"github.com/trustwallet/blockatlas/pkg/blockatlas"
-	"github.com/trustwallet/blockatlas/pkg/logger"
-	"github.com/trustwallet/blockatlas/pkg/storage/redis"
-	"time"
-)
-
-type Storage struct {
-	redis.Redis
-	blockHeights BlockMap
-}
-
-func New() *Storage {
-	s := new(Storage)
-	s.blockHeights.heights = make(map[uint]int64)
-	return s
-}
-
-func RestoreConnectionWorker(storage *Storage, uri string, timeout time.Duration) {
-	logger.Info("Run Redis RestoreConnectionWorker")
-	for {
-		if !storage.IsReady() {
-			for {
-				logger.Warn("Redis is not available now")
-				logger.Warn("Trying to connect to MQ...")
-				if err := storage.Init(uri); err != nil {
-					logger.Warn("Redis is still unavailable")
-					time.Sleep(timeout)
-					continue
-				} else {
-					logger.Info("Redis connection restored")
-					break
-				}
-			}
-		}
-		time.Sleep(timeout)
-	}
-}
-
-type Tracker interface {
-	GetLastParsedBlockNumber(coin uint) (int64, error)
-	SetLastParsedBlockNumber(coin uint, num int64) error
-}
-
-type Addresses interface {
-	FindSubscriptions(coin uint, addresses []string) ([]blockatlas.Subscription, error)
-	AddSubscriptions(subscriptions []blockatlas.Subscription) error
-	DeleteSubscriptions(subscriptions []blockatlas.Subscription) error
-}
diff --git a/storage/tracker.go b/storage/tracker.go
deleted file mode 100644
index 78ab2b6..0000000
--- a/storage/tracker.go
+++ /dev/null
@@ -1,56 +0,0 @@
-package storage
-
-import (
-	"fmt"
-	"sync"
-)
-
-const (
-	ATLAS_BLOCK_NUMBER = "ATLAS_BLOCK_NUMBER_%d"
-)
-
-type BlockMap struct {
-	heights map[uint]int64
-	lock    sync.RWMutex
-}
-
-func (bm *BlockMap) SetBlock(coin uint, b int64) {
-	bm.lock.Lock()
-	defer bm.lock.Unlock()
-	bm.heights[coin] = b
-}
-
-func (bm *BlockMap) GetBlock(coin uint) (int64, bool) {
-	bm.lock.RLock()
-	defer bm.lock.RUnlock()
-	b, ok := bm.heights[coin]
-	return b, ok
-}
-
-func (bm *BlockMap) GetHeights() map[uint]int64 {
-	bm.lock.RLock()
-	defer bm.lock.RUnlock()
-	return bm.heights
-}
-
-func (s *Storage) GetLastParsedBlockNumber(coin uint) (int64, error) {
-	b, ok := s.blockHeights.GetBlock(coin)
-	if ok {
-		return b, nil
-	}
-
-	err := s.GetValue(getBlockKey(coin), &b)
-	if err != nil {
-		return 0, nil
-	}
-	return b, nil
-}
-
-func (s *Storage) SetLastParsedBlockNumber(coin uint, num int64) error {
-	s.blockHeights.SetBlock(coin, num)
-	return s.Add(getBlockKey(coin), num)
-}
-
-func getBlockKey(coin uint) string {
-	return fmt.Sprintf(ATLAS_BLOCK_NUMBER, coin)
-}


Test patch:
diff --git a/db/subscritions_test.go b/db/subscritions_test.go
new file mode 100644
index 0000000..274ceec
--- /dev/null
+++ b/db/subscritions_test.go
@@ -0,0 +1,91 @@
+package db
+
+import (
+	"github.com/stretchr/testify/assert"
+	"github.com/trustwallet/blockatlas/db/models"
+	"testing"
+)
+
+func TestGetSubscriptionsToDeleteAndUpdate(t *testing.T) {
+	oldSubscriptions := []models.SubscriptionData{{
+		SubscriptionId: 1,
+		Coin:           60,
+		Address:        "A",
+	}, {
+		SubscriptionId: 1,
+		Coin:           60,
+		Address:        "B",
+	}, {
+		SubscriptionId: 1,
+		Coin:           60,
+		Address:        "C",
+	}}
+
+	newSubscription := []models.SubscriptionData{{
+		SubscriptionId: 1,
+		Coin:           60,
+		Address:        "B",
+	}, {
+		SubscriptionId: 1,
+		Coin:           60,
+		Address:        "C",
+	}}
+
+	update, delete := getSubscriptionsToDeleteAndUpdate(oldSubscriptions, newSubscription)
+	assert.Len(t, update, 0)
+	assert.Len(t, delete, 1)
+	assert.Equal(t, "A", delete[0].Address)
+
+	oldSubscriptions = []models.SubscriptionData{{
+		SubscriptionId: 1,
+		Coin:           60,
+		Address:        "A",
+	}, {
+		SubscriptionId: 1,
+		Coin:           60,
+		Address:        "B",
+	}, {
+		SubscriptionId: 1,
+		Coin:           60,
+		Address:        "C",
+	}, {
+		SubscriptionId: 1,
+		Coin:           60,
+		Address:        "D",
+	}}
+
+	newSubscription = []models.SubscriptionData{{
+		SubscriptionId: 1,
+		Coin:           60,
+		Address:        "E",
+	}}
+
+	update, delete = getSubscriptionsToDeleteAndUpdate(oldSubscriptions, newSubscription)
+	assert.Len(t, update, 1)
+	assert.Len(t, delete, 4)
+
+	oldSubscriptions = []models.SubscriptionData{{
+		SubscriptionId: 1,
+		Coin:           60,
+		Address:        "A",
+	}, {
+		SubscriptionId: 1,
+		Coin:           60,
+		Address:        "B",
+	}}
+
+	newSubscription = []models.SubscriptionData{{
+		SubscriptionId: 1,
+		Coin:           60,
+		Address:        "A",
+	}, {
+		SubscriptionId: 1,
+		Coin:           60,
+		Address:        "B",
+	}}
+
+	update, delete = getSubscriptionsToDeleteAndUpdate(oldSubscriptions, newSubscription)
+	assert.Len(t, update, 0)
+	assert.Len(t, delete, 0)
+
+}
diff --git a/pkg/blockatlas/observer_test.go b/pkg/blockatlas/observer_test.go
index 1f80e0e..1cd29b1 100644
--- a/pkg/blockatlas/observer_test.go
+++ b/pkg/blockatlas/observer_test.go
@@ -15,18 +15,18 @@ func Test_parseSubscriptions(t *testing.T) {
 		{
 			name: "guid with 1 coin",
 			subscriptions: SubscriptionEvent{
-				GUID: "http://127.0.0.1:8080",
-				OldSubscriptions: Subscriptions{
+				Id: 1,
+				Subscriptions: Subscriptions{
 					"0": {"xpub6BpYi6J1GZzfY3yY7DbhLLccF3efQa18nQngM3jaehgtNSoEgk6UtPULpC3oK5oA3trczY8Ld34LFw1USMPfGHwTEizdD5QyGcMyuh2UoBA", "xpub6CYwPfnPJLPquufPkb98coSb3mdy1CgaZrWUtYWGJTJ4VWZUbzH9HLGy7nHpP7DG4UdTkYYpirkTWQSP7pWHsrk24Nos5oYNHpfr4BgPVTL"},
 				},
 			},
 			wantSubs: []Subscription{
 				{
-					Coin: 0, GUID: "http://127.0.0.1:8080",
+					Coin: 0, Id: 1,
 					Address: "xpub6BpYi6J1GZzfY3yY7DbhLLccF3efQa18nQngM3jaehgtNSoEgk6UtPULpC3oK5oA3trczY8Ld34LFw1USMPfGHwTEizdD5QyGcMyuh2UoBA",
 				},
 				{
-					Coin: 0, GUID: "http://127.0.0.1:8080",
+					Coin: 0, Id: 1,
 					Address: "xpub6CYwPfnPJLPquufPkb98coSb3mdy1CgaZrWUtYWGJTJ4VWZUbzH9HLGy7nHpP7DG4UdTkYYpirkTWQSP7pWHsrk24Nos5oYNHpfr4BgPVTL",
 				},
 			},
@@ -34,23 +34,23 @@ func Test_parseSubscriptions(t *testing.T) {
 		{
 			name: "guid with 2 coins",
 			subscriptions: SubscriptionEvent{
-				GUID: "http://127.0.0.1:8080",
-				OldSubscriptions: Subscriptions{
+				Id: 1,
+				Subscriptions: Subscriptions{
 					"2": {"zpub6rH4MwgyTmuexAX6HAraks5cKv5BbtmwdLirvnU5845ovUJb4abgjt9DtXK4ZEaToRrNj8dQznuLC6Nka4eMviGMinCVMUxKLpuyddcG9Vc"},
 					"0": {"xpub6BpYi6J1GZzfY3yY7DbhLLccF3efQa18nQngM3jaehgtNSoEgk6UtPULpC3oK5oA3trczY8Ld34LFw1USMPfGHwTEizdD5QyGcMyuh2UoBA", "xpub6CYwPfnPJLPquufPkb98coSb3mdy1CgaZrWUtYWGJTJ4VWZUbzH9HLGy7nHpP7DG4UdTkYYpirkTWQSP7pWHsrk24Nos5oYNHpfr4BgPVTL"},
 				},
 			},
 			wantSubs: []Subscription{
 				{
-					Coin: 2, GUID: "http://127.0.0.1:8080",
+					Coin: 2, Id: 1,
 					Address: "zpub6rH4MwgyTmuexAX6HAraks5cKv5BbtmwdLirvnU5845ovUJb4abgjt9DtXK4ZEaToRrNj8dQznuLC6Nka4eMviGMinCVMUxKLpuyddcG9Vc",
 				},
 				{
-					Coin: 0, GUID: "http://127.0.0.1:8080",
+					Coin: 0, Id: 1,
 					Address: "xpub6BpYi6J1GZzfY3yY7DbhLLccF3efQa18nQngM3jaehgtNSoEgk6UtPULpC3oK5oA3trczY8Ld34LFw1USMPfGHwTEizdD5QyGcMyuh2UoBA",
 				},
 				{
-					Coin: 0, GUID: "http://127.0.0.1:8080",
+					Coin: 0, Id: 1,
 					Address: "xpub6CYwPfnPJLPquufPkb98coSb3mdy1CgaZrWUtYWGJTJ4VWZUbzH9HLGy7nHpP7DG4UdTkYYpirkTWQSP7pWHsrk24Nos5oYNHpfr4BgPVTL",
 				},
 			},
@@ -58,7 +58,7 @@ func Test_parseSubscriptions(t *testing.T) {
 	}
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
-			gotSubs := tt.subscriptions.ParseSubscriptions(tt.subscriptions.OldSubscriptions)
+			gotSubs := tt.subscriptions.ParseSubscriptions(tt.subscriptions.Subscriptions)
 			sort.Slice(gotSubs, func(i, j int) bool {
 				return gotSubs[i].Coin > gotSubs[j].Coin
 			})
diff --git a/platform/tezos/model_test.go b/platform/tezos/model_test.go
index 773b11d..7243c1a 100644
--- a/platform/tezos/model_test.go
+++ b/platform/tezos/model_test.go
@@ -66,7 +66,7 @@ func TestTransaction_Status(t *testing.T) {
 		in   Transaction
 		out  int64
 	}{
-		{"Delegation", Transaction{Time: "2020-02-04T12:27:59Z",}, 1580819279},
+		{"Delegation", Transaction{Time: "2020-02-04T12:27:59Z"}, 1580819279},
 	}
 
 	for _, tt := range testsBlockTimestamp {
@@ -80,10 +80,10 @@ func TestTransaction_Status(t *testing.T) {
 		in   Transaction
 		out  blockatlas.TransactionType
 	}{
-		{"Type should be transaction", Transaction{Type: "transaction",}, blockatlas.TxTransfer},
-		{"Type should be delegation", Transaction{Type: "delegation",}, blockatlas.TxAnyAction},
-		{"Type unsupported", Transaction{Type: "bake",}, ""},
-		{"Type endorsement", Transaction{Type: "endorsement",}, ""},
+		{"Type should be transaction", Transaction{Type: "transaction"}, blockatlas.TxTransfer},
+		{"Type should be delegation", Transaction{Type: "delegation"}, blockatlas.TxAnyAction},
+		{"Type unsupported", Transaction{Type: "bake"}, ""},
+		{"Type endorsement", Transaction{Type: "endorsement"}, ""},
 	}
 
 	for _, tt := range testsKind {
@@ -128,4 +128,19 @@ func TestTransaction_Status(t *testing.T) {
 		})
 	}
 
+	testsGetReceiver := []struct {
+		name string
+		in   Transaction
+		out  string
+	}{
+		{"Should get receiver when no delegate", Transaction{Receiver: addr1, Delegate: ""}, addr1},
+		{"Should get receiver when delegate", Transaction{Receiver: "", Delegate: addr1}, addr1},
+	}
+
+	for _, tt := range testsGetReceiver {
+		t.Run(tt.name, func(t *testing.T) {
+			assert.Equal(t, tt.out, tt.in.GetReceiver())
+		})
+	}
+
 }
diff --git a/platform/tezos/transaction_test.go b/platform/tezos/transaction_test.go
index cc8febf..aa98ce7 100644
--- a/platform/tezos/transaction_test.go
+++ b/platform/tezos/transaction_test.go
@@ -26,7 +26,7 @@ var (
 		Coin:      1729,
 		From:      addr1,
 		To:        addr1,
-		Fee:       "0.0015",
+		Fee:       "1500",
 		Date:      1582894746,
 		Block:     843988,
 		Status:    "completed",
@@ -34,7 +34,7 @@ var (
 		Direction: "yourself",
 		Memo:      "",
 		Meta: blockatlas.Transfer{
-			Value:    "0.000001",
+			Value:    "1",
 			Symbol:   "XTZ",
 			Decimals: 6,
 		},
diff --git a/scripts/run_tests_postman.sh b/scripts/run_tests_postman.sh
index 0877438..e18f114 100644
--- a/scripts/run_tests_postman.sh
+++ b/scripts/run_tests_postman.sh
@@ -8,4 +8,4 @@ make newman test=token host=$1
 make newman test=staking host=$1
 make newman test=collection host=$1
 make newman test=domain host=$1
-make newman test=observer host=$1
+make newman test=observer_test host=$1
diff --git a/services/observer/parser/parser_test.go b/services/observer/parser/parser_test.go
index 38457c6..6b6d369 100644
--- a/services/observer/parser/parser_test.go
+++ b/services/observer/parser/parser_test.go
@@ -2,13 +2,9 @@ package parser
 
 import (
 	"errors"
-	"fmt"
-	"github.com/alicebob/miniredis"
 	"github.com/stretchr/testify/assert"
 	"github.com/trustwallet/blockatlas/coin"
 	"github.com/trustwallet/blockatlas/pkg/blockatlas"
-	"github.com/trustwallet/blockatlas/pkg/logger"
-	"github.com/trustwallet/blockatlas/storage"
 	"sync"
 	"testing"
 	"time"
@@ -66,48 +62,11 @@ var (
 	}
 )
 
-func Test_GetBlocksIntervalToFetch(t *testing.T) {
-
-	params := Params{
-		ParsingBlocksInterval: time.Minute,
-		BacklogCount:          10,
-		MaxBacklogBlocks:      100,
-		Storage:               getMockedRedis(t),
-		Api:                   getMockedBlockAPI(),
-	}
-	latestParsedBlock := int64(100)
-	wantedMockedNumber = 110
-	lastParsedBlock, currentBlock, err := GetBlocksIntervalToFetch(params)
-	assert.Nil(t, err)
-	assert.Equal(t, wantedMockedNumber, currentBlock)
-	assert.Equal(t, latestParsedBlock, lastParsedBlock)
-}
-
 func TestFetchBlocks(t *testing.T) {
 	blocks := FetchBlocks(getMockedBlockAPI(), 0, 100)
 	assert.Equal(t, len(blocks), 100)
 }
 
-func TestSaveLastParsedBlock(t *testing.T) {
-	blocks := make([]blockatlas.Block, 0)
-	blocks = append(blocks, block)
-	s := getMockedRedis(t)
-
-	params := Params{
-		ParsingBlocksInterval: time.Minute,
-		BacklogCount:          10,
-		MaxBacklogBlocks:      100,
-		Storage:               s,
-		Api:                   getMockedBlockAPI(),
-	}
-	err := SaveLastParsedBlock(params, blocks)
-	assert.Nil(t, err)
-
-	lastParsedBlock, err := s.GetLastParsedBlockNumber(getMockedBlockAPI().Coin().ID)
-	assert.Nil(t, err)
-	assert.Equal(t, lastParsedBlock, int64(110))
-}
-
 func TestParser_ConvertToBatch(t *testing.T) {
 	blocks := []blockatlas.Block{block, block, block, block}
 	txs := ConvertToBatch(blocks)
@@ -174,20 +133,6 @@ func getMockedBlockAPI() blockatlas.BlockAPI {
 	return &p
 }
 
-func getMockedRedis(t *testing.T) *storage.Storage {
-	s, err := miniredis.Run()
-	if err != nil {
-		t.Fatal(err)
-	}
-
-	cache := storage.New()
-	err = cache.Init(fmt.Sprintf("redis://%s", s.Addr()))
-	if err != nil {
-		logger.Fatal(err)
-	}
-	return cache
-}
-
 type Platform struct {
 	CoinIndex uint
 }
diff --git a/storage/addresses_test.go b/storage/addresses_test.go
deleted file mode 100644
index 52ef385..0000000
--- a/storage/addresses_test.go
+++ /dev/null
@@ -1,347 +0,0 @@
-package storage
-
-import (
-	"fmt"
-	"github.com/alicebob/miniredis"
-	"github.com/stretchr/testify/assert"
-	"github.com/trustwallet/blockatlas/pkg/blockatlas"
-	"github.com/trustwallet/blockatlas/pkg/logger"
-	"testing"
-)
-
-func TestStorage_Lookup(t *testing.T) {
-	s := initStorage(t)
-	assert.NotNil(t, s)
-
-	type fields struct {
-		coin      int
-		addresses []string
-	}
-	tests := []struct {
-		name          string
-		fields        fields
-		wantData      []blockatlas.Subscription
-		wantCondition bool
-	}{
-		{"test all guids found",
-			fields{coin: 60, addresses: []string{"1", "2", "3"}},
-			[]blockatlas.Subscription{
-				{Coin: 60, Address: "1", GUID: "1"},
-				{Coin: 60, Address: "2", GUID: "2"},
-				{Coin: 60, Address: "3", GUID: "3"},
-			},
-			true,
-		},
-		{"test not found",
-			fields{coin: 60, addresses: []string{"1", "4", "3"}},
-			[]blockatlas.Subscription{
-				{Coin: 60, Address: "1", GUID: "1"},
-				{Coin: 60, Address: "2", GUID: "2"},
-				{Coin: 60, Address: "3", GUID: "3"},
-			},
-			false,
-		},
-	}
-
-	for _, tt := range tests {
-		for i, a := range tt.fields.addresses {
-			key := getSubscriptionKey(uint(tt.fields.coin), a)
-			err := s.AddHM(ATLAS_OBSERVER, key, []string{tt.wantData[i].GUID})
-			if err != nil {
-				t.Fatal(err)
-			}
-		}
-
-		t.Run(tt.name, func(t *testing.T) {
-			if got, err := s.FindSubscriptions(uint(tt.fields.coin), tt.fields.addresses); !(isEqualSubscriptions(got, tt.wantData) == tt.wantCondition) || err != nil {
-				t.Fatal(got)
-			}
-		})
-	}
-}
-
-func TestStorage_Lookup_MultipleGUIDs(t *testing.T) {
-	s := initStorage(t)
-	assert.NotNil(t, s)
-
-	want := []blockatlas.Subscription{
-		{Coin: 60, Address: "1", GUID: "1"},
-		{Coin: 60, Address: "2", GUID: "2"},
-		{Coin: 60, Address: "2", GUID: "3"},
-		{Coin: 60, Address: "3", GUID: "3"},
-	}
-
-	key1 := getSubscriptionKey(uint(60), "1")
-	err := s.AddHM(ATLAS_OBSERVER, key1, []string{"1"})
-	if err != nil {
-		t.Fatal(err)
-	}
-
-	key2 := getSubscriptionKey(uint(60), "2")
-	err = s.AddHM(ATLAS_OBSERVER, key2, []string{"2", "3"})
-	if err != nil {
-		t.Fatal(err)
-	}
-
-	key3 := getSubscriptionKey(uint(60), "3")
-	err = s.AddHM(ATLAS_OBSERVER, key3, []string{"3"})
-	if err != nil {
-		t.Fatal(err)
-	}
-
-	given, err := s.FindSubscriptions(uint(60), []string{"1", "2", "3"})
-	assert.Nil(t, err)
-	assert.True(t, isEqualSubscriptions(given, want))
-}
-
-func TestStorage_Lookup_NotFoundSeveral(t *testing.T) {
-	s := initStorage(t)
-	assert.NotNil(t, s)
-
-	want := []blockatlas.Subscription{
-		{Coin: 60, Address: "1", GUID: "1"},
-	}
-
-	key1 := getSubscriptionKey(uint(60), "1")
-	err := s.AddHM(ATLAS_OBSERVER, key1, []string{"1"})
-	if err != nil {
-		t.Fatal(err)
-	}
-
-	key2 := getSubscriptionKey(uint(60), "2")
-	err = s.AddHM(ATLAS_OBSERVER, key2, []string{"2", "3"})
-	if err != nil {
-		t.Fatal(err)
-	}
-
-	key3 := getSubscriptionKey(uint(60), "3")
-	err = s.AddHM(ATLAS_OBSERVER, key3, []string{"3"})
-	if err != nil {
-		t.Fatal(err)
-	}
-
-	given, err := s.FindSubscriptions(uint(60), []string{"1", "4", "5"})
-	assert.Nil(t, err)
-	assert.True(t, isEqualSubscriptions(given, want))
-}
-
-func TestStorage_AddSubscriptions(t *testing.T) {
-	s := initStorage(t)
-	assert.NotNil(t, s)
-
-	subs := []blockatlas.Subscription{
-		{Coin: 60, Address: "1", GUID: "1"},
-		{Coin: 144, Address: "11212", GUID: "112"},
-		{Coin: 144, Address: "112121", GUID: "112"},
-	}
-	var want []string
-	for _, sub := range subs {
-		key := getSubscriptionKey(sub.Coin, sub.Address)
-		var guids []string
-
-		err := s.GetHMValue(ATLAS_OBSERVER, key, &guids)
-		assert.NotNil(t, err)
-		assert.Equal(t, want, guids)
-	}
-
-	err := s.AddSubscriptions(subs)
-	assert.Nil(t, err)
-
-	for _, sub := range subs {
-		key := getSubscriptionKey(sub.Coin, sub.Address)
-		var guids []string
-		err := s.GetHMValue(ATLAS_OBSERVER, key, &guids)
-		assert.Nil(t, err)
-		assert.NotNil(t, guids)
-
-		var counter int
-		for _, g := range guids {
-			if g == sub.GUID {
-				counter++
-			}
-		}
-		assert.True(t, counter == 1)
-	}
-
-	NewSubs := []blockatlas.Subscription{
-		{Coin: 714, Address: "2", GUID: "2"},
-		{Coin: 148, Address: "21", GUID: "21"},
-		{Coin: 148, Address: "21", GUID: "21"},
-	}
-
-	err = s.AddSubscriptions(NewSubs)
-	assert.Nil(t, err)
-
-	for _, sub := range subs {
-		key := getSubscriptionKey(sub.Coin, sub.Address)
-		var guids []string
-		err := s.GetHMValue(ATLAS_OBSERVER, key, &guids)
-		assert.Nil(t, err)
-		assert.NotNil(t, guids)
-
-		var counter int
-		for _, g := range guids {
-			if g == sub.GUID {
-				counter++
-			}
-		}
-		assert.True(t, counter == 1)
-	}
-
-	for _, sub := range NewSubs {
-		key := getSubscriptionKey(sub.Coin, sub.Address)
-		var guids []string
-		err := s.GetHMValue(ATLAS_OBSERVER, key, &guids)
-		assert.Nil(t, err)
-		assert.NotNil(t, guids)
-
-		var counter int
-		for _, g := range guids {
-			if g == sub.GUID {
-				counter++
-			}
-		}
-		assert.True(t, counter == 1)
-	}
-
-	NotExistingSubs := []blockatlas.Subscription{
-		{Coin: 111, Address: "2", GUID: "2"},
-		{Coin: 222, Address: "21", GUID: "21"},
-	}
-
-	for _, sub := range NotExistingSubs {
-		key := getSubscriptionKey(sub.Coin, sub.Address)
-		var guids []string
-		err := s.GetHMValue(ATLAS_OBSERVER, key, &guids)
-		assert.NotNil(t, err)
-		assert.Nil(t, guids)
-
-		var counter int
-		for _, g := range guids {
-			if g == sub.GUID {
-				counter++
-			}
-		}
-		assert.False(t, counter == 1)
-	}
-}
-
-func TestStorage_DeleteSubscriptions(t *testing.T) {
-	s := initStorage(t)
-	assert.NotNil(t, s)
-
-	subs := []blockatlas.Subscription{
-		{Coin: 60, Address: "1", GUID: "1"},
-		{Coin: 144, Address: "11212", GUID: "112"},
-		{Coin: 255, Address: "112121", GUID: "1121"},
-	}
-	var want []string
-	for _, sub := range subs {
-		key := getSubscriptionKey(sub.Coin, sub.Address)
-		var guids []string
-
-		err := s.GetHMValue(ATLAS_OBSERVER, key, &guids)
-		assert.NotNil(t, err)
-		assert.Equal(t, want, guids)
-	}
-
-	err := s.AddSubscriptions(subs)
-	assert.Nil(t, err)
-
-	for _, sub := range subs {
-		key := getSubscriptionKey(sub.Coin, sub.Address)
-		var guids []string
-		err := s.GetHMValue(ATLAS_OBSERVER, key, &guids)
-		assert.Nil(t, err)
-		assert.NotNil(t, guids)
-
-		var counter int
-		for _, g := range guids {
-			if g == sub.GUID {
-				counter++
-			}
-		}
-		assert.True(t, counter == 1)
-	}
-
-	deleted_subs := []blockatlas.Subscription{
-		{Coin: 60, Address: "1", GUID: "1"},
-		{Coin: 144, Address: "11212", GUID: "112"},
-	}
-
-	err = s.DeleteSubscriptions(deleted_subs)
-	assert.Nil(t, err)
-
-	for _, sub := range deleted_subs {
-		key := getSubscriptionKey(sub.Coin, sub.Address)
-		var guids []string
-		err := s.GetHMValue(ATLAS_OBSERVER, key, &guids)
-		assert.NotNil(t, err)
-		assert.Nil(t, guids)
-
-		var counter int
-		for _, g := range guids {
-			if g == sub.GUID {
-				counter++
-			}
-		}
-		assert.True(t, counter == 0)
-	}
-
-	existing_subs := []blockatlas.Subscription{
-		{Coin: 255, Address: "112121", GUID: "1121"},
-	}
-
-	for _, sub := range existing_subs {
-		key := getSubscriptionKey(sub.Coin, sub.Address)
-		var guids []string
-		err := s.GetHMValue(ATLAS_OBSERVER, key, &guids)
-		assert.Nil(t, err)
-		assert.NotNil(t, guids)
-
-		var counter int
-		for _, g := range guids {
-			if g == sub.GUID {
-				counter++
-			}
-		}
-		assert.True(t, counter == 1)
-	}
-}
-
-func isEqualSubscriptions(given, want []blockatlas.Subscription) bool {
-	if len(given) != len(want) {
-		return false
-	}
-	var givenCounter int
-	for _, g := range given {
-		var wantCounter int
-		for _, w := range want {
-			if w == g {
-				wantCounter++
-			}
-		}
-		if wantCounter > 0 {
-			givenCounter++
-		}
-	}
-
-	if givenCounter == len(want) {
-		return true
-	}
-	return false
-}
-
-func initStorage(t *testing.T) *Storage {
-	s, err := miniredis.Run()
-	if err != nil {
-		t.Fatal(err)
-	}
-
-	storage := New()
-	err = storage.Init(fmt.Sprintf("redis://%s", s.Addr()))
-	if err != nil {
-		logger.Fatal(err)
-	}
-	return storage
-}
diff --git a/storage/tracker_test.go b/storage/tracker_test.go
deleted file mode 100644
index 56a7a30..0000000
--- a/storage/tracker_test.go
+++ /dev/null
@@ -1,24 +0,0 @@
-package storage
-
-import (
-	"github.com/stretchr/testify/assert"
-	"testing"
-)
-
-func TestBlockMap(t *testing.T) {
-	s := initStorage(t)
-	assert.NotNil(t, s)
-
-	block, err := s.GetLastParsedBlockNumber(60)
-	assert.Nil(t, err)
-	assert.Equal(t, int64(0), block)
-
-	newBlock := int64(1400)
-
-	err = s.SetLastParsedBlockNumber(60, newBlock)
-	assert.Nil(t, err)
-
-	current, err := s.GetLastParsedBlockNumber(60)
-	assert.Nil(t, err)
-	assert.Equal(t, newBlock, current)
-}
diff --git a/tests/docker_test/setup/redis.go b/tests/docker_test/setup/redis.go
deleted file mode 100644
index 5ae759e..0000000
--- a/tests/docker_test/setup/redis.go
+++ /dev/null
@@ -1,43 +0,0 @@
-package setup
-
-import (
-	"fmt"
-	"github.com/ory/dockertest"
-	"github.com/trustwallet/blockatlas/storage"
-	"log"
-)
-
-var (
-	Cache         *storage.Storage
-	redisResource *dockertest.Resource
-)
-
-func runRedisContainer() error {
-	var err error
-	pool, err := dockertest.NewPool("")
-	if err != nil {
-		log.Fatalf("Could not connect to docker: %s", err)
-	}
-
-	redisResource, err = pool.Run("redis", "latest", nil)
-	if err != nil {
-		log.Fatalf("Could not start resource: %s", err)
-	}
-
-	if err = pool.Retry(func() error {
-		var err error
-		Cache = storage.New()
-		err = Cache.Init(fmt.Sprintf("redis://localhost:%s", redisResource.GetPort("6379/tcp")))
-		if err != nil {
-			return err
-		}
-		return nil
-	}); err != nil {
-		return err
-	}
-	return nil
-}
-
-func stopRedisContainer() error {
-	return redisResource.Close()
-}
diff --git a/tests/docker_test/setup/setup.go b/tests/docker_test/setup/setup.go
deleted file mode 100644
index 53210ff..0000000
--- a/tests/docker_test/setup/setup.go
+++ /dev/null
@@ -1,27 +0,0 @@
-package setup
-
-import "log"
-
-func RunRedisContainer() {
-	if err := runRedisContainer(); err != nil {
-		log.Fatal(err)
-	}
-}
-
-func StopRedisContainer() {
-	if err := stopRedisContainer(); err != nil {
-		log.Fatal(err)
-	}
-}
-
-func RunMQContainer() {
-	if err := runMQContainer(); err != nil {
-		log.Fatal(err)
-	}
-}
-
-func StopMQContainer() {
-	if err := stopMQContainer(); err != nil {
-		log.Fatal(err)
-	}
-}
diff --git a/tests/docker_test/subscriber_test.go b/tests/docker_test/subscriber_test.go
deleted file mode 100644
index ca65276..0000000
--- a/tests/docker_test/subscriber_test.go
+++ /dev/null
@@ -1,66 +0,0 @@
-// +build integration
-
-package docker_test
-
-import (
-	"context"
-	"encoding/json"
-	"github.com/stretchr/testify/assert"
-	"github.com/trustwallet/blockatlas/mq"
-	"github.com/trustwallet/blockatlas/pkg/blockatlas"
-	"github.com/trustwallet/blockatlas/services/observer/subscriber"
-	"github.com/trustwallet/blockatlas/storage"
-	"github.com/trustwallet/blockatlas/tests/docker_test/setup"
-	"io/ioutil"
-	"path/filepath"
-	"runtime"
-	"testing"
-	"time"
-)
-
-func TestSubscriberAddSubscription(t *testing.T) {
-	_, goFile, _, _ := runtime.Caller(0)
-	testFilePathGiven := filepath.Join(filepath.Dir(goFile), "data", "given_subscriptions_added.json")
-	testFileGiven, err := ioutil.ReadFile(testFilePathGiven)
-	if err != nil {
-		t.Fatal(err)
-	}
-	var givenEvents []blockatlas.SubscriptionEvent
-	if err := json.Unmarshal(testFileGiven, &givenEvents); err != nil {
-		t.Fatal(err)
-	}
-
-	testFilePathWanted := filepath.Join(filepath.Dir(goFile), "data", "wanted_subscriptions_added.json")
-	testFileWanted, err := ioutil.ReadFile(testFilePathWanted)
-	if err != nil {
-		t.Fatal(err)
-	}
-	var wantedEvents []blockatlas.Subscription
-	if err := json.Unmarshal(testFileWanted, &wantedEvents); err != nil {
-		t.Fatal(err)
-	}
-
-	for _, event := range givenEvents {
-		body, err := json.Marshal(event)
-		assert.Nil(t, err)
-
-		err = mq.Subscriptions.Publish(body)
-		assert.Nil(t, err)
-
-		ctx, cancel := context.WithCancel(context.Background())
-
-		go mq.Subscriptions.RunConsumerForChannelWithCancel(subscriber.RunSubscriber, subscriptionChannel, setup.Cache, ctx)
-		time.Sleep(time.Second / 5)
-		cancel()
-	}
-
-	result, err := setup.Cache.GetAllHM(storage.ATLAS_OBSERVER)
-	assert.Nil(t, err)
-	assert.NotNil(t, result)
-
-	for _, wanted := range wantedEvents {
-		result, err := setup.Cache.FindSubscriptions(wanted.Coin, []string{wanted.Address})
-		assert.Nil(t, err)
-		assert.Equal(t, result[0], wanted)
-	}
-}
diff --git a/tests/integration/bitcoin/block.go b/tests/integration/bitcoin/block.go
deleted file mode 100755
index 3237b43..0000000
--- a/tests/integration/bitcoin/block.go
+++ /dev/null
@@ -1,44 +0,0 @@
-// +build integration
-
-package bitcoin
-
-import (
-	"github.com/trustwallet/blockatlas/coin"
-	"github.com/trustwallet/blockatlas/platform"
-	"github.com/trustwallet/blockatlas/platform/bitcoin"
-	"testing"
-)
-
-const (
-	blockNum = 616070
-)
-
-func TestBitcoin(t *testing.T) {
-	t.Run("test bitcoin", func(t *testing.T) {
-		p := bitcoin.Init(coin.BTC, platform.GetApiVar(coin.BTC))
-		testCurrentBlockNumber(p, t)
-		testGetBlockByNumber(p, t)
-	})
-}
-
-func testCurrentBlockNumber(p *bitcoin.Platform, t *testing.T) {
-	resp, err := p.CurrentBlockNumber()
-	if err != nil {
-		t.Error(err)
-	}
-	if resp < 0 {
-		t.Error("block is < 0")
-	}
-}
-
-func testGetBlockByNumber(p *bitcoin.Platform, t *testing.T) {
-	resp, err := p.GetBlockByNumber(int64(blockNum))
-	if err != nil {
-		t.Error(err)
-	}
-
-	expectedTxs := 2129
-	if len(resp.Txs) != expectedTxs {
-		t.Errorf("invalid tx count for block %d, expected %d", len(resp.Txs), expectedTxs)
-	}
-}
diff --git a/tests/integration/db_test/db_run_test.go b/tests/integration/db_test/db_run_test.go
new file mode 100644
index 0000000..061c11b
--- /dev/null
+++ b/tests/integration/db_test/db_run_test.go
@@ -0,0 +1,25 @@
+// +build integration
+
+package db_test
+
+import (
+	"github.com/stretchr/testify/assert"
+	"github.com/trustwallet/blockatlas/db"
+	"github.com/trustwallet/blockatlas/tests/integration/setup"
+	"os"
+	"testing"
+)
+
+var database *db.Instance
+
+func TestMain(m *testing.M) {
+	database = setup.RunPgContainer()
+	code := m.Run()
+	setup.StopPgContainer()
+	os.Exit(code)
+}
+
+func TestPgSetup(t *testing.T) {
+	assert.NotNil(t, database)
+	assert.NotNil(t, database.DB)
+}
diff --git a/tests/integration/db_test/subscriptions_test.go b/tests/integration/db_test/subscriptions_test.go
new file mode 100644
index 0000000..97e66c6
--- /dev/null
+++ b/tests/integration/db_test/subscriptions_test.go
@@ -0,0 +1,448 @@
+// +build integration
+
+package db_test
+
+import (
+	"github.com/stretchr/testify/assert"
+	"github.com/trustwallet/blockatlas/db/models"
+	"github.com/trustwallet/blockatlas/pkg/blockatlas"
+	"github.com/trustwallet/blockatlas/services/observer/subscriber"
+	"github.com/trustwallet/blockatlas/tests/integration/setup"
+	"testing"
+)
+
+func TestDb_AddSubscriptions(t *testing.T) {
+	setup.CleanupPgContainer(database.DB)
+	var subscriptions []models.SubscriptionData
+
+	id := uint(1)
+	subscriptions = append(subscriptions, models.SubscriptionData{
+		SubscriptionId: id,
+		Coin:           60,
+		Address:        "testAddr",
+	})
+
+	subscriptions = append(subscriptions, models.SubscriptionData{
+		SubscriptionId: id,
+		Coin:           61,
+		Address:        "testAddr2",
+	})
+
+	subscriptions = append(subscriptions, models.SubscriptionData{
+		SubscriptionId: id,
+		Coin:           62,
+		Address:        "testAddr3",
+	})
+
+	assert.Nil(t, database.AddSubscriptions(id, subscriptions))
+
+	subs, err := database.GetSubscriptionData(60, []string{"testAddr"})
+	assert.Nil(t, err)
+	assert.NotNil(t, subs)
+	assert.Equal(t, 1, len(subs))
+	assert.Equal(t, subscriptions[0].SubscriptionId, subs[0].SubscriptionId)
+	assert.Equal(t, subscriptions[0].Coin, subs[0].Coin)
+	assert.Equal(t, subscriptions[0].Address, subs[0].Address)
+
+	subs, err = database.GetSubscriptionData(61, []string{"testAddr2"})
+	assert.Nil(t, err)
+	assert.NotNil(t, subs)
+	assert.Equal(t, 1, len(subs))
+	assert.Equal(t, subscriptions[1].SubscriptionId, subs[0].SubscriptionId)
+	assert.Equal(t, subscriptions[1].Coin, subs[0].Coin)
+	assert.Equal(t, subscriptions[1].Address, subs[0].Address)
+
+	subs, err = database.GetSubscriptionData(62, []string{"testAddr3"})
+	assert.Nil(t, err)
+	assert.NotNil(t, subs)
+	assert.Equal(t, 1, len(subs))
+	assert.Equal(t, subscriptions[2].SubscriptionId, subs[0].SubscriptionId)
+	assert.Equal(t, subscriptions[2].Coin, subs[0].Coin)
+	assert.Equal(t, subscriptions[2].Address, subs[0].Address)
+}
+
+func TestDb_AddSubscriptionsWithRewrite(t *testing.T) {
+	setup.CleanupPgContainer(database.DB)
+	id := uint(1)
+
+	var subscriptions []models.SubscriptionData
+	subscriptions = append(subscriptions, models.SubscriptionData{
+		SubscriptionId: id,
+		Coin:           60,
+		Address:        "testAddr",
+	})
+
+	subscriptions = append(subscriptions, models.SubscriptionData{
+		SubscriptionId: id,
+		Coin:           714,
+		Address:        "testAddr",
+	})
+
+	subscriptions = append(subscriptions, models.SubscriptionData{
+		SubscriptionId: id,
+		Coin:           144,
+		Address:        "testAddr",
+	})
+
+	assert.Nil(t, database.AddSubscriptions(id, subscriptions))
+
+	subs60, err := database.GetSubscriptionData(60, []string{"testAddr"})
+	assert.Nil(t, err)
+	assert.NotNil(t, subs60)
+	assert.Equal(t, 1, len(subs60))
+	assert.Equal(t, subscriptions[0].SubscriptionId, subs60[0].SubscriptionId)
+	assert.Equal(t, subscriptions[0].Coin, subs60[0].Coin)
+	assert.Equal(t, subscriptions[0].Address, subs60[0].Address)
+
+	subs714, err := database.GetSubscriptionData(714, []string{"testAddr"})
+	assert.Nil(t, err)
+	assert.NotNil(t, subs714)
+	assert.Equal(t, 1, len(subs714))
+	assert.Equal(t, subscriptions[1].SubscriptionId, subs714[0].SubscriptionId)
+	assert.Equal(t, subscriptions[1].Coin, subs714[0].Coin)
+	assert.Equal(t, subscriptions[1].Address, subs714[0].Address)
+
+	subs144, err := database.GetSubscriptionData(144, []string{"testAddr"})
+	assert.Nil(t, err)
+	assert.NotNil(t, subs144)
+	assert.Equal(t, 1, len(subs144))
+	assert.Equal(t, subscriptions[2].SubscriptionId, subs144[0].SubscriptionId)
+	assert.Equal(t, subscriptions[2].Coin, subs144[0].Coin)
+	assert.Equal(t, subscriptions[2].Address, subs144[0].Address)
+
+	subscriptions = append(subscriptions, models.SubscriptionData{
+		SubscriptionId: id,
+		Coin:           60,
+		Address:        "testAddr2",
+	})
+
+	subscriptions = append(subscriptions, models.SubscriptionData{
+		SubscriptionId: id,
+		Coin:           714,
+		Address:        "testAddr2",
+	})
+
+	subscriptions = append(subscriptions, models.SubscriptionData{
+		SubscriptionId: id,
+		Coin:           144,
+		Address:        "testAddr2",
+	})
+
+	assert.Nil(t, database.AddSubscriptions(id, subscriptions))
+
+	subs2N60, err := database.GetSubscriptionData(60, []string{"testAddr2"})
+	assert.Nil(t, err)
+	assert.Nil(t, err)
+	assert.NotNil(t, subs2N60)
+	assert.Equal(t, 1, len(subs2N60))
+	assert.Equal(t, subscriptions[3].SubscriptionId, subs2N60[0].SubscriptionId)
+	assert.Equal(t, subscriptions[3].Coin, subs2N60[0].Coin)
+	assert.Equal(t, subscriptions[3].Address, subs2N60[0].Address)
+
+	subs2N714, err := database.GetSubscriptionData(714, []string{"testAddr2"})
+	assert.Nil(t, err)
+	assert.Nil(t, err)
+	assert.NotNil(t, subs2N714)
+	assert.Equal(t, 1, len(subs2N714))
+	assert.Equal(t, subscriptions[4].SubscriptionId, subs2N714[0].SubscriptionId)
+	assert.Equal(t, subscriptions[4].Coin, subs2N714[0].Coin)
+	assert.Equal(t, subscriptions[4].Address, subs2N714[0].Address)
+
+	subs2N114, err := database.GetSubscriptionData(144, []string{"testAddr2"})
+	assert.Nil(t, err)
+	assert.Nil(t, err)
+	assert.NotNil(t, subs2N114)
+	assert.Equal(t, 1, len(subs2N114))
+	assert.Equal(t, subscriptions[5].SubscriptionId, subs2N114[0].SubscriptionId)
+	assert.Equal(t, subscriptions[5].Coin, subs2N114[0].Coin)
+	assert.Equal(t, subscriptions[5].Address, subs2N114[0].Address)
+}
+
+func TestDb_FindSubscriptions(t *testing.T) {
+	setup.CleanupPgContainer(database.DB)
+
+	var subscriptionsA []blockatlas.Subscription
+	id := uint(1)
+	subscriptionsA = append(subscriptionsA, blockatlas.Subscription{
+		Id:      id,
+		Coin:    60,
+		Address: "etherAddress",
+	})
+
+	subscriptionsA = append(subscriptionsA, blockatlas.Subscription{
+		Id:      id,
+		Coin:    714,
+		Address: "binanceAddress",
+	})
+
+	subscriptionsA = append(subscriptionsA, blockatlas.Subscription{
+		Id:      id,
+		Coin:    148,
+		Address: "AtomAddress",
+	})
+
+	subscriptionsA = append(subscriptionsA, blockatlas.Subscription{
+		Id:      id,
+		Coin:    144,
+		Address: "XLMAddress",
+	})
+
+	subscriptionsA = append(subscriptionsA, blockatlas.Subscription{
+		Id:      id,
+		Coin:    61,
+		Address: "ETCAddress",
+	})
+
+	assert.Nil(t, database.AddSubscriptions(id, subscriber.ToSubscriptionData(subscriptionsA)))
+
+	var subscriptionsB []blockatlas.Subscription
+
+	for _, sub := range subscriptionsA {
+		sub.Id = uint(2)
+		subscriptionsB = append(subscriptionsB, sub)
+	}
+	assert.Nil(t, database.AddSubscriptions(2, subscriber.ToSubscriptionData(subscriptionsB)))
+
+	returnedSubs, err := database.GetSubscriptionData(60, []string{"etherAddress"})
+	assert.Nil(t, err)
+	assert.Equal(t, 2, len(returnedSubs))
+
+	returnedSubs, err = database.GetSubscriptionData(714, []string{"binanceAddress"})
+	assert.Nil(t, err)
+	assert.Equal(t, 2, len(returnedSubs))
+
+	returnedSubs, err = database.GetSubscriptionData(144, []string{"XLMAddress"})
+	assert.Nil(t, err)
+	assert.Equal(t, 2, len(returnedSubs))
+
+	returnedSubs, err = database.GetSubscriptionData(148, []string{"AtomAddress"})
+	assert.Nil(t, err)
+	assert.Equal(t, 2, len(returnedSubs))
+
+	returnedSubs, err = database.GetSubscriptionData(61, []string{"ETCAddress"})
+	assert.Nil(t, err)
+	assert.Equal(t, 2, len(returnedSubs))
+}
+
+func TestDb_DeleteSubscriptions(t *testing.T) {
+	setup.CleanupPgContainer(database.DB)
+	var subscriptions []models.SubscriptionData
+
+	id := uint(1)
+	subscriptions = append(subscriptions, models.SubscriptionData{
+		SubscriptionId: id,
+		Coin:           60,
+		Address:        "testAddr",
+	})
+
+	subscriptions = append(subscriptions, models.SubscriptionData{
+		SubscriptionId: id,
+		Coin:           714,
+		Address:        "testAddr2",
+	})
+
+	subscriptions = append(subscriptions, models.SubscriptionData{
+		SubscriptionId: id,
+		Coin:           144,
+		Address:        "testAddr3",
+	})
+
+	assert.Nil(t, database.AddSubscriptions(id, subscriptions))
+
+	subs60, err := database.GetSubscriptionData(60, []string{"testAddr"})
+	assert.Nil(t, err)
+	assert.NotNil(t, subs60)
+	assert.Equal(t, 1, len(subs60))
+	assert.Equal(t, subscriptions[0].SubscriptionId, subs60[0].SubscriptionId)
+	assert.Equal(t, subscriptions[0].Coin, subs60[0].Coin)
+	assert.Equal(t, subscriptions[0].Address, subs60[0].Address)
+
+	subs714, err := database.GetSubscriptionData(714, []string{"testAddr2"})
+	assert.Nil(t, err)
+	assert.NotNil(t, subs714)
+	assert.Equal(t, 1, len(subs714))
+	assert.Equal(t, subscriptions[1].SubscriptionId, subs714[0].SubscriptionId)
+	assert.Equal(t, subscriptions[1].Coin, subs714[0].Coin)
+	assert.Equal(t, subscriptions[1].Address, subs714[0].Address)
+
+	subs144, err := database.GetSubscriptionData(144, []string{"testAddr3"})
+	assert.Nil(t, err)
+	assert.NotNil(t, subs144)
+	assert.Equal(t, 1, len(subs144))
+	assert.Equal(t, subscriptions[2].SubscriptionId, subs144[0].SubscriptionId)
+	assert.Equal(t, subscriptions[2].Coin, subs144[0].Coin)
+	assert.Equal(t, subscriptions[2].Address, subs144[0].Address)
+
+	subsToDel := []models.SubscriptionData{subscriptions[0]}
+
+	assert.Nil(t, database.DeleteSubscriptions(subsToDel))
+
+	subs714N2, err := database.GetSubscriptionData(714, []string{"testAddr2"})
+	assert.Nil(t, err)
+	assert.NotNil(t, subs714N2)
+	assert.Equal(t, 1, len(subs714N2))
+	assert.Equal(t, subscriptions[1].SubscriptionId, subs714N2[0].SubscriptionId)
+	assert.Equal(t, subscriptions[1].Coin, subs714N2[0].Coin)
+	assert.Equal(t, subscriptions[1].Address, subs714N2[0].Address)
+
+	subs144N2, err := database.GetSubscriptionData(144, []string{"testAddr3"})
+	assert.Nil(t, err)
+	assert.NotNil(t, subs144N2)
+	assert.Equal(t, 1, len(subs144N2))
+	assert.Equal(t, subscriptions[2].SubscriptionId, subs144N2[0].SubscriptionId)
+	assert.Equal(t, subscriptions[2].Coin, subs144N2[0].Coin)
+	assert.Equal(t, subscriptions[2].Address, subs144N2[0].Address)
+
+	subs60N2, err := database.GetSubscriptionData(60, []string{"testAddr"})
+	assert.Nil(t, err)
+	assert.Len(t, subs60N2, 0)
+}
+
+func TestDeleteAll(t *testing.T) {
+	setup.CleanupPgContainer(database.DB)
+	var subscriptions []models.SubscriptionData
+
+	id := uint(1)
+	subscriptions = append(subscriptions, models.SubscriptionData{
+		SubscriptionId: id,
+		Coin:           60,
+		Address:        "testAddr",
+	})
+
+	subscriptions = append(subscriptions, models.SubscriptionData{
+		SubscriptionId: id,
+		Coin:           714,
+		Address:        "testAddr2",
+	})
+
+	subscriptions = append(subscriptions, models.SubscriptionData{
+		SubscriptionId: id,
+		Coin:           144,
+		Address:        "testAddr3",
+	})
+	assert.Nil(t, database.AddSubscriptions(id, subscriptions))
+
+	subs60, err := database.GetSubscriptionData(60, []string{"testAddr"})
+	assert.Nil(t, err)
+	assert.Len(t, subs60, 1)
+
+	subs714, err := database.GetSubscriptionData(714, []string{"testAddr2"})
+	assert.Nil(t, err)
+	assert.Len(t, subs714, 1)
+
+	subs144, err := database.GetSubscriptionData(144, []string{"testAddr3"})
+	assert.Nil(t, err)
+	assert.Len(t, subs144, 1)
+
+	assert.Nil(t, database.DeleteAllSubscriptions(1))
+
+	subs60, err = database.GetSubscriptionData(60, []string{"testAddr"})
+	assert.Nil(t, err)
+	assert.Len(t, subs60, 0)
+
+	subs714, err = database.GetSubscriptionData(714, []string{"testAddr2"})
+	assert.Nil(t, err)
+	assert.Len(t, subs714, 0)
+
+	subs144, err = database.GetSubscriptionData(144, []string{"testAddr3"})
+	assert.Nil(t, err)
+	assert.Len(t, subs144, 0)
+
+	assert.Nil(t, database.AddSubscriptions(id, subscriptions))
+
+	subs60, err = database.GetSubscriptionData(60, []string{"testAddr"})
+	assert.Nil(t, err)
+	assert.Len(t, subs60, 1)
+}
+
+func TestDb_DuplicateEntries(t *testing.T) {
+	setup.CleanupPgContainer(database.DB)
+	var subscriptions []models.SubscriptionData
+
+	id := uint(1)
+
+	for i := 0; i < 10; i++ {
+		subscriptions = append(subscriptions, models.SubscriptionData{
+			SubscriptionId: id,
+			Coin:           60,
+			Address:        "testAddr",
+		})
+	}
+
+	assert.Nil(t, database.AddSubscriptions(id, subscriptions))
+
+	subs, err := database.GetSubscriptionData(60, []string{"testAddr"})
+	assert.Nil(t, err)
+	assert.NotNil(t, subs)
+	assert.Equal(t, 1, len(subs))
+	assert.True(t, containSub(subscriptions[0], subs))
+}
+
+func TestDb_FindSubscriptions_Multiple(t *testing.T) {
+	setup.CleanupPgContainer(database.DB)
+	var subscriptions []models.SubscriptionData
+	subscriptions = append(subscriptions, models.SubscriptionData{
+		Coin:    60,
+		Address: "testAddr",
+	})
+
+	for i := 1; i < 6; i++ {
+		subscriptions[0].SubscriptionId = uint(i)
+		assert.Nil(t, database.AddSubscriptions(uint(i), subscriptions))
+	}
+
+	subscriptions[0].SubscriptionId = uint(1)
+	assert.Nil(t, database.AddSubscriptions(uint(1), subscriptions))
+
+	subs, err := database.GetSubscriptionData(60, []string{"testAddr"})
+	assert.Nil(t, err)
+	assert.Equal(t, 5, len(subs))
+
+	for i := 0; i < 5; i++ {
+		assert.Equal(t, uint(i)+1, subs[i].SubscriptionId)
+	}
+}
+
+func TestDb_AddToExisting(t *testing.T) {
+	setup.CleanupPgContainer(database.DB)
+	var subscriptions []models.SubscriptionData
+	subscriptions = append(subscriptions, models.SubscriptionData{
+		Coin:    60,
+		Address: "testAddr",
+	})
+
+	subscriptions[0].SubscriptionId = uint(1)
+	assert.Nil(t, database.AddSubscriptions(uint(1), subscriptions))
+
+	subs, err := database.GetSubscriptionData(60, []string{"testAddr"})
+	assert.Nil(t, err)
+	assert.Equal(t, 1, len(subs))
+
+	assert.Equal(t, uint(1), subs[0].SubscriptionId)
+
+	assert.Nil(t, database.AddToExistingSubscription(uint(1), subscriptions))
+
+	subs2, err2 := database.GetSubscriptionData(60, []string{"testAddr"})
+	assert.Nil(t, err2)
+	assert.Equal(t, 1, len(subs2))
+
+	assert.Equal(t, uint(1), subs2[0].SubscriptionId)
+
+	assert.Nil(t, database.AddToExistingSubscription(uint(2), subscriptions))
+	assert.Nil(t, database.AddToExistingSubscription(uint(1), subscriptions))
+
+	for i := 1; i < 2; i++ {
+		assert.Nil(t, database.AddToExistingSubscription(uint(i), subscriptions))
+	}
+	assert.NotNil(t, database.AddToExistingSubscription(uint(0), subscriptions))
+
+}
+
+func containSub(sub models.SubscriptionData, list []models.SubscriptionData) bool {
+	for _, s := range list {
+		if sub.Address == s.Address && sub.Coin == s.Coin && sub.SubscriptionId == s.SubscriptionId {
+			return true
+		}
+	}
+	return false
+}
diff --git a/tests/integration/db_test/tracker_test.go b/tests/integration/db_test/tracker_test.go
new file mode 100644
index 0000000..a05f1c3
--- /dev/null
+++ b/tests/integration/db_test/tracker_test.go
@@ -0,0 +1,25 @@
+// +build integration
+
+package db_test
+
+import (
+	"github.com/stretchr/testify/assert"
+	"github.com/trustwallet/blockatlas/tests/integration/setup"
+	"testing"
+)
+
+func TestDb_SetBlock(t *testing.T) {
+	setup.CleanupPgContainer(database.DB)
+
+	assert.Nil(t, database.SetLastParsedBlockNumber(60, 0))
+
+	block, err := database.GetLastParsedBlockNumber(60)
+	assert.Nil(t, err)
+	assert.Equal(t, block, int64(0))
+
+	assert.Nil(t, database.SetLastParsedBlockNumber(60, 110))
+
+	newBlock, err := database.GetLastParsedBlockNumber(60)
+	assert.Nil(t, err)
+	assert.Equal(t, newBlock, int64(110))
+}
diff --git a/tests/integration/domains/domains.go b/tests/integration/domains/domains.go
deleted file mode 100644
index a4568b0..0000000
--- a/tests/integration/domains/domains.go
+++ /dev/null
@@ -1,129 +0,0 @@
-// +build integration
-
-package domains
-
-import (
-	"testing"
-
-	"github.com/stretchr/testify/assert"
-	"github.com/trustwallet/blockatlas/coin"
-	"github.com/trustwallet/blockatlas/pkg/blockatlas"
-	"github.com/trustwallet/blockatlas/services/domains"
-)
-
-func TestDomains(t *testing.T) {
-	tests := []struct {
-		name     string
-		domain   string
-		coins    []uint64
-		Expected []blockatlas.Resolved
-		wantErr  bool
-	}{
-		{
-			"test .eth domain",
-			"vitalik.eth",
-			[]uint64{coin.ETH},
-			[]blockatlas.Resolved{{Result: "0xd8dA6BF26964aF9D7eEd9e03E53415D37aA96045", Coin: coin.ETH}},
-			false,
-		},
-		{
-			"test .luxe domain",
-			"vitalik.luxe",
-			[]uint64{coin.ETH},
-			[]blockatlas.Resolved{{Result: "0xD8A667312D5260F12a306Ae7730C754d938da86c", Coin: coin.ETH}},
-			false,
-		},
-		{
-			"test .xyz domain",
-			"ourxyzwallet.xyz",
-			[]uint64{coin.ETH},
-			[]blockatlas.Resolved{{Result: "0x0C54eEAd78d555bE3cbCD451424F9A27a7843935", Coin: coin.ETH}},
-			false,
-		},
-		{
-			"test .kred domain",
-			"Cameron.Kred",
-			[]uint64{coin.ETH},
-			[]blockatlas.Resolved{{Result: "0xfc9FB438c456C771c0B38a57Cf0e8A3eaB4704fE", Coin: coin.ETH}},
-			false,
-		},
-		{
-			"test .zil domain",
-			"dpantani.zil",
-			[]uint64{coin.ZIL},
-			[]blockatlas.Resolved{{Result: "zil1vdntvlk47j9kh9a85klqcd9rvgze06ruhmna64", Coin: coin.ZIL}},
-			false,
-		},
-		{
-			"test batch .zil domains",
-			"dpantani.zil",
-			[]uint64{coin.BTC, coin.ETH, coin.ZIL, coin.LTC, coin.BNB, coin.BCH, coin.DOGE, coin.XRP},
-			[]blockatlas.Resolved{
-				{Result: "bc1qd7eystu9xl53hkyxm4kyg7h5yk4p436sqx6f27", Coin: coin.BTC},
-				{Result: "0x5574Cd97432cEd0D7Caf58ac3c4fEDB2061C98fB", Coin: coin.ETH},
-				{Result: "zil1vdntvlk47j9kh9a85klqcd9rvgze06ruhmna64", Coin: coin.ZIL},
-				{Result: "ltc1qz6nd472gx5gl3urfeldkrhg3h83c8tp2m7m6sd", Coin: coin.LTC},
-				{Result: "bnb1h4vyuuytu4rm86ust29wwlevt95du52383cctm", Coin: coin.BNB},
-				{Result: "qzpjlfnzudeu83krv0yk0r2kys67qptj6ys6eg6dms", Coin: coin.BCH},
-				{Result: "DP9VmQyDMyB1TWwgXkyRpBa7rTfPYgMvjy", Coin: coin.DOGE},
-				{Result: "rUvXBttEXhdwaKjEM2MxbtswHU6AMhUTgJ", Coin: coin.XRP},
-			},
-			false,
-		},
-		{
-			"test .crypto domain",
-			"dpantani.crypto",
-			[]uint64{coin.ZIL},
-			[]blockatlas.Resolved{
-				{Result: "zil1vdntvlk47j9kh9a85klqcd9rvgze06ruhmna64", Coin: coin.ZIL},
-			},
-			false,
-		},
-		//{
-		//	"test @fiotestnet domain",
-		//	"adam@fiotestnet",
-		//	[]uint64{coin.ETH},
-		//	[]blockatlas.Resolved{
-		//		{Result: "0xce5cB6c92Da37bbBa91Bd40D4C9D4D724A3a8F51", Coin: coin.ETH},
-		//	},
-		//	false,
-		//},
-		{
-			"test batch .crypto domains",
-			"dpantani.crypto",
-			[]uint64{coin.BTC, coin.ETH, coin.ZIL, coin.LTC, coin.BNB, coin.BCH, coin.DOGE, coin.XRP},
-			[]blockatlas.Resolved{
-				{Result: "bc1qd7eystu9xl53hkyxm4kyg7h5yk4p436sqx6f27", Coin: coin.BTC},
-				{Result: "0x5574Cd97432cEd0D7Caf58ac3c4fEDB2061C98fB", Coin: coin.ETH},
-				{Result: "zil1vdntvlk47j9kh9a85klqcd9rvgze06ruhmna64", Coin: coin.ZIL},
-				{Result: "ltc1qz6nd472gx5gl3urfeldkrhg3h83c8tp2m7m6sd", Coin: coin.LTC},
-				{Result: "bnb1h4vyuuytu4rm86ust29wwlevt95du52383cctm", Coin: coin.BNB},
-				{Result: "qzpjlfnzudeu83krv0yk0r2kys67qptj6ys6eg6dms", Coin: coin.BCH},
-				{Result: "DP9VmQyDMyB1TWwgXkyRpBa7rTfPYgMvjy", Coin: coin.DOGE},
-				{Result: "rUvXBttEXhdwaKjEM2MxbtswHU6AMhUTgJ", Coin: coin.XRP},
-			},
-			false,
-		},
-		{
-			"test batch with invalid coin",
-			"vitalik.eth",
-			[]uint64{44, coin.ETH},
-			[]blockatlas.Resolved{{Result: "0xd8dA6BF26964aF9D7eEd9e03E53415D37aA96045", Coin: coin.ETH}},
-			false,
-		},
-		{"test invalid coin", "vitalik.eth", []uint64{44}, nil, false},
-		{"test invalid name", "z9z9z900s982jidhwallet.eth", []uint64{coin.ALGO}, nil, true},
-		{"test invalid domain", "vitalik.blabla", []uint64{coin.ALGO}, nil, true},
-	}
-	for _, test := range tests {
-		t.Run(test.name, func(t *testing.T) {
-			got, err := domains.HandleLookup(test.domain, test.coins)
-			if test.wantErr {
-				assert.NotNil(t, err)
-				return
-			}
-			assert.Nil(t, err)
-			assert.EqualValues(t, test.Expected, got)
-		})
-	}
-}
diff --git a/tests/integration/integration_test.go b/tests/integration/integration_test.go
deleted file mode 100755
index 218e490..0000000
--- a/tests/integration/integration_test.go
+++ /dev/null
@@ -1,30 +0,0 @@
-// +build integration
-
-package integration
-
-import (
-	"github.com/spf13/viper"
-	"github.com/trustwallet/blockatlas/config"
-	"github.com/trustwallet/blockatlas/tests/integration/bitcoin"
-	"github.com/trustwallet/blockatlas/tests/integration/domains"
-	"github.com/trustwallet/blockatlas/tests/integration/ontology"
-
-	"github.com/trustwallet/blockatlas/platform"
-	"os"
-	"testing"
-)
-
-func Test(t *testing.T) {
-	configPath := os.Getenv("TEST_CONFIG")
-	if configPath == "" {
-		config.LoadConfig("../../../config.yml")
-	} else {
-		config.LoadConfig(configPath)
-	}
-	platform.Init(viper.GetString("platform"))
-
-	// Add your integration tests here
-	ontology.TestOntology(t)
-	bitcoin.TestBitcoin(t)
-	domains.TestDomains(t)
-}
diff --git a/tests/docker_test/data/given_subscriptions_added.json b/tests/integration/observer_test/data/given_subscriptions_added.json
similarity index 62%
rename from tests/docker_test/data/given_subscriptions_added.json
rename to tests/integration/observer_test/data/given_subscriptions_added.json
index 6d5efbc..87fdbc9 100644
--- a/tests/docker_test/data/given_subscriptions_added.json
+++ b/tests/integration/observer_test/data/given_subscriptions_added.json
@@ -1,42 +1,38 @@
 [
   {
-    "new_subscriptions": {
+    "subscriptions": {
       "60": [
         "0x0000000000000000000000000000000000000000"
       ]
     },
-    "old_subscriptions": null,
-    "guid": "0",
+    "id": 10,
     "operation": "AddSubscription"
   },
   {
-    "new_subscriptions": {
+    "subscriptions": {
       "118": [
         "0x0000000000000000000000000000000000000001"
       ]
     },
-    "old_subscriptions": null,
-    "guid": "1",
+    "id": 1,
     "operation": "AddSubscription"
   },
   {
-    "new_subscriptions": {
+    "subscriptions": {
       "714": [
         "0x0000000000000000000000000000000000000002"
       ]
     },
-    "old_subscriptions": null,
-    "guid": "2",
+    "id": 2,
     "operation": "AddSubscription"
   },
   {
-    "new_subscriptions": {
+    "subscriptions": {
       "144": [
         "0x0000000000000000000000000000000000000003"
       ]
     },
-    "old_subscriptions": null,
-    "guid": "3",
+    "id": 3,
     "operation": "AddSubscription"
   }
 ]
diff --git a/tests/integration/observer_test/data/given_subscriptions_deleted.json b/tests/integration/observer_test/data/given_subscriptions_deleted.json
new file mode 100644
index 0000000..7c88737
--- /dev/null
+++ b/tests/integration/observer_test/data/given_subscriptions_deleted.json
@@ -0,0 +1,38 @@
+[
+  {
+    "subscriptions": {
+      "60": [
+        "0x0000000000000000000000000000000000000000"
+      ]
+    },
+    "id": 10,
+    "operation": "UpdateSubscription"
+  },
+  {
+    "subscriptions": {
+      "118": [
+        "0x0000000000000000000000000000000000000001"
+      ]
+    },
+    "id": 1,
+    "operation": "UpdateSubscription"
+  },
+  {
+    "subscriptions": {
+      "714": [
+        "0x0000000000000000000000000000000000000002"
+      ]
+    },
+    "id": 2,
+    "operation": "UpdateSubscription"
+  },
+  {
+    "subscriptions": {
+      "144": [
+        "0x0000000000000000000000000000000000000003"
+      ]
+    },
+    "id": 3,
+    "operation": "UpdateSubscription"
+  }
+]
diff --git a/tests/docker_test/data/wanted_subscriptions_added.json b/tests/integration/observer_test/data/wanted_subscriptions_added.json
similarity index 84%
rename from tests/docker_test/data/wanted_subscriptions_added.json
rename to tests/integration/observer_test/data/wanted_subscriptions_added.json
index c89a4b5..7007019 100644
--- a/tests/docker_test/data/wanted_subscriptions_added.json
+++ b/tests/integration/observer_test/data/wanted_subscriptions_added.json
@@ -2,21 +2,21 @@
   {
     "coin": 60,
     "address": "0x0000000000000000000000000000000000000000",
-    "guid": "0"
+    "id": 10
   },
   {
     "coin": 118,
     "address": "0x0000000000000000000000000000000000000001",
-    "guid": "1"
+    "id": 1
   },
   {
     "coin": 714,
     "address": "0x0000000000000000000000000000000000000002",
-    "guid": "2"
+    "id": 2
   },
   {
     "coin": 144,
     "address": "0x0000000000000000000000000000000000000003",
-    "guid": "3"
+    "id": 3
   }
 ]
diff --git a/tests/docker_test/observer_test.go b/tests/integration/observer_test/full_flow_test.go
similarity index 86%
rename from tests/docker_test/observer_test.go
rename to tests/integration/observer_test/full_flow_test.go
index 8c60b23..0dbe863 100644
--- a/tests/docker_test/observer_test.go
+++ b/tests/integration/observer_test/full_flow_test.go
@@ -1,6 +1,6 @@
 // +build integration
 
-package docker_test
+package observer_test
 
 import (
 	"context"
@@ -8,21 +8,24 @@ import (
 	"github.com/streadway/amqp"
 	"github.com/stretchr/testify/assert"
 	"github.com/trustwallet/blockatlas/coin"
+	"github.com/trustwallet/blockatlas/db/models"
 	"github.com/trustwallet/blockatlas/mq"
 	"github.com/trustwallet/blockatlas/pkg/blockatlas"
 	"github.com/trustwallet/blockatlas/services/observer/notifier"
 	"github.com/trustwallet/blockatlas/services/observer/parser"
-	"github.com/trustwallet/blockatlas/tests/docker_test/setup"
+	"github.com/trustwallet/blockatlas/tests/integration/setup"
 	"go.uber.org/atomic"
 	"testing"
 	"time"
 )
 
-var counter atomic.Int32
-var counterBlock atomic.Int32
+var (
+	currentBlockNumberCounter atomic.Int32
+)
 
 func TestFullFlow(t *testing.T) {
-	err := setup.Cache.AddSubscriptions([]blockatlas.Subscription{{Coin: 60, Address: "testAddress", GUID: "guid_test"}})
+	setup.CleanupPgContainer(database.DB)
+	err := database.AddSubscriptions(1, []models.SubscriptionData{{Coin: 60, Address: "testAddress", SubscriptionId: 1}})
 	assert.Nil(t, err)
 
 	ctx, cancel := context.WithCancel(context.Background())
@@ -30,13 +33,14 @@ func TestFullFlow(t *testing.T) {
 	stopChan := make(chan struct{}, 1)
 
 	params := setupParserFull(stopChan)
+	params.Database = database
 	params.Ctx = ctx
 	params.Queue = mq.RawTransactions
 
 	go parser.RunParser(params)
 	time.Sleep(time.Second * 2)
 
-	go mq.RawTransactions.RunConsumerForChannelWithCancel(notifier.RunNotifier, rawTransactionsChannel, setup.Cache, ctx)
+	go mq.RunConsumerForChannelWithCancelAndDbConn(notifier.RunNotifier, rawTransactionsChannel, database, ctx)
 	time.Sleep(time.Second * 5)
 
 	for i := 0; i < 11; i++ {
@@ -56,8 +60,8 @@ type PlatformFullFlow struct {
 }
 
 func (p *PlatformFullFlow) CurrentBlockNumber() (int64, error) {
-	i := counterBlock.Load()
-	counterBlock.Add(1)
+	i := currentBlockNumberCounter.Load()
+	currentBlockNumberCounter.Add(1)
 	return int64(i), nil
 }
 
@@ -131,7 +135,7 @@ func ConsumerToTestTransactionsFull(delivery amqp.Delivery, t *testing.T, cancel
 			Memo:      "test",
 			Meta:      &memo,
 		},
-		GUID: "guid_test",
+		Id: 1,
 	}, event)
 
 	if counter == 10 {
@@ -150,7 +154,6 @@ func setupParserFull(stopChan chan<- struct{}) parser.Params {
 
 	return parser.Params{
 		Api:                   getMockedBlockAPIFull(),
-		Storage:               setup.Cache,
 		ParsingBlocksInterval: pollInterval,
 		BacklogCount:          backlogCount,
 		MaxBacklogBlocks:      int64(maxBatchBlocksAmount),
diff --git a/tests/docker_test/notifier_test.go b/tests/integration/observer_test/notifier_test.go
similarity index 87%
rename from tests/docker_test/notifier_test.go
rename to tests/integration/observer_test/notifier_test.go
index ea9fa95..3151047 100644
--- a/tests/docker_test/notifier_test.go
+++ b/tests/integration/observer_test/notifier_test.go
@@ -1,6 +1,6 @@
 // +build integration
 
-package docker_test
+package observer_test
 
 import (
 	"context"
@@ -8,10 +8,11 @@ import (
 	"github.com/streadway/amqp"
 	"github.com/stretchr/testify/assert"
 	"github.com/trustwallet/blockatlas/coin"
+	"github.com/trustwallet/blockatlas/db/models"
 	"github.com/trustwallet/blockatlas/mq"
 	"github.com/trustwallet/blockatlas/pkg/blockatlas"
 	"github.com/trustwallet/blockatlas/services/observer/notifier"
-	"github.com/trustwallet/blockatlas/tests/docker_test/setup"
+	"github.com/trustwallet/blockatlas/tests/integration/setup"
 	"testing"
 	"time"
 )
@@ -41,7 +42,9 @@ var (
 )
 
 func TestNotifier(t *testing.T) {
-	err := setup.Cache.AddSubscriptions([]blockatlas.Subscription{{Coin: 714, Address: "tbnb1ttyn4csghfgyxreu7lmdu3lcplhqhxtzced45a", GUID: "guid_test"}})
+	setup.CleanupPgContainer(database.DB)
+
+	err := database.AddSubscriptions(1, []models.SubscriptionData{{Coin: 714, Address: "tbnb1ttyn4csghfgyxreu7lmdu3lcplhqhxtzced45a", SubscriptionId: 1}})
 	assert.Nil(t, err)
 
 	err = produceTxs(txs)
@@ -49,7 +52,7 @@ func TestNotifier(t *testing.T) {
 
 	ctx, cancel := context.WithCancel(context.Background())
 
-	go mq.RawTransactions.RunConsumerForChannelWithCancel(notifier.RunNotifier, rawTransactionsChannel, setup.Cache, ctx)
+	go mq.RunConsumerForChannelWithCancelAndDbConn(notifier.RunNotifier, rawTransactionsChannel, database, ctx)
 	time.Sleep(time.Second * 3)
 	msg := transactionsChannel.GetMessage()
 	ConsumerToTestTransactions(msg, t)
@@ -93,7 +96,7 @@ func ConsumerToTestTransactions(delivery amqp.Delivery, t *testing.T) {
 			Memo:      "test",
 			Meta:      &memo,
 		},
-		GUID: "guid_test",
+		Id: 1,
 	}, event)
 
 	return
diff --git a/tests/docker_test/environment_run_test.go b/tests/integration/observer_test/observer_test.go
similarity index 64%
rename from tests/docker_test/environment_run_test.go
rename to tests/integration/observer_test/observer_test.go
index 35e38f3..dc75b91 100644
--- a/tests/docker_test/environment_run_test.go
+++ b/tests/integration/observer_test/observer_test.go
@@ -1,21 +1,23 @@
 // +build integration
 
-package docker_test
+package observer_test
 
 import (
+	"github.com/trustwallet/blockatlas/db"
 	"github.com/trustwallet/blockatlas/mq"
-	"github.com/trustwallet/blockatlas/tests/docker_test/setup"
+	"github.com/trustwallet/blockatlas/tests/integration/setup"
 	"log"
 	"os"
 	"testing"
 )
 
 var (
-	rawTransactionsChannel,
-	transactionsChannel, subscriptionChannel mq.MessageChannel
+	rawTransactionsChannel, transactionsChannel, subscriptionChannel mq.MessageChannel
+	database                                                         *db.Instance
 )
 
 func TestMain(m *testing.M) {
+	database = setup.RunPgContainer()
 	setup.RunMQContainer()
 	if err := mq.RawTransactions.Declare(); err != nil {
 		log.Fatal(err)
@@ -30,9 +32,9 @@ func TestMain(m *testing.M) {
 	subscriptionChannel = mq.Subscriptions.GetMessageChannel()
 	transactionsChannel = mq.Transactions.GetMessageChannel()
 
-	setup.RunRedisContainer()
 	code := m.Run()
+
 	setup.StopMQContainer()
-	setup.StopRedisContainer()
+	setup.StopPgContainer()
 	os.Exit(code)
 }
diff --git a/tests/docker_test/parser_test.go b/tests/integration/observer_test/parser_test.go
similarity index 95%
rename from tests/docker_test/parser_test.go
rename to tests/integration/observer_test/parser_test.go
index 1c1005a..59e7a85 100644
--- a/tests/docker_test/parser_test.go
+++ b/tests/integration/observer_test/parser_test.go
@@ -1,6 +1,6 @@
 // +build integration
 
-package docker_test
+package observer_test
 
 import (
 	"context"
@@ -13,15 +13,17 @@ import (
 	"github.com/trustwallet/blockatlas/pkg/logger"
 	"github.com/trustwallet/blockatlas/services/observer/notifier"
 	"github.com/trustwallet/blockatlas/services/observer/parser"
-	"github.com/trustwallet/blockatlas/tests/docker_test/setup"
+	"github.com/trustwallet/blockatlas/tests/integration/setup"
 	"testing"
 	"time"
 )
 
 func TestParserFetchAndPublishBlock_NormalCase(t *testing.T) {
+	setup.CleanupPgContainer(database.DB)
 	stopChan := make(chan struct{}, 1)
 
 	params := setupParser(stopChan)
+	params.Database = database
 
 	ctx, cancel := context.WithCancel(context.Background())
 
@@ -109,7 +111,6 @@ func setupParser(stopChan chan struct{}) parser.Params {
 
 	return parser.Params{
 		Api:                   getMockedBlockAPI(),
-		Storage:               setup.Cache,
 		ParsingBlocksInterval: pollInterval,
 		BacklogCount:          backlogCount,
 		MaxBacklogBlocks:      int64(maxBatchBlocksAmount),
diff --git a/tests/integration/observer_test/subscriber_test.go b/tests/integration/observer_test/subscriber_test.go
new file mode 100644
index 0000000..ef52c74
--- /dev/null
+++ b/tests/integration/observer_test/subscriber_test.go
@@ -0,0 +1,146 @@
+// +build integration
+
+package observer_test
+
+import (
+	"context"
+	"encoding/json"
+	"github.com/stretchr/testify/assert"
+	"github.com/trustwallet/blockatlas/db/models"
+	"github.com/trustwallet/blockatlas/mq"
+	"github.com/trustwallet/blockatlas/pkg/blockatlas"
+	"github.com/trustwallet/blockatlas/services/observer/subscriber"
+	"github.com/trustwallet/blockatlas/tests/integration/setup"
+	"io/ioutil"
+	"path/filepath"
+	"runtime"
+	"testing"
+	"time"
+)
+
+func TestSubscriberAddSubscription(t *testing.T) {
+	setup.CleanupPgContainer(database.DB)
+
+	_, goFile, _, _ := runtime.Caller(0)
+	testFilePathGiven := filepath.Join(filepath.Dir(goFile), "data", "given_subscriptions_added.json")
+	testFileGiven, err := ioutil.ReadFile(testFilePathGiven)
+	if err != nil {
+		t.Fatal(err)
+	}
+	var givenEvents []blockatlas.SubscriptionEvent
+	if err := json.Unmarshal(testFileGiven, &givenEvents); err != nil {
+		t.Fatal(err)
+	}
+
+	testFilePathWanted := filepath.Join(filepath.Dir(goFile), "data", "wanted_subscriptions_added.json")
+	testFileWanted, err := ioutil.ReadFile(testFilePathWanted)
+	if err != nil {
+		t.Fatal(err)
+	}
+	var wantedEvents []blockatlas.Subscription
+	if err := json.Unmarshal(testFileWanted, &wantedEvents); err != nil {
+		t.Fatal(err)
+	}
+
+	for _, event := range givenEvents {
+		body, err := json.Marshal(event)
+		assert.Nil(t, err)
+
+		err = mq.Subscriptions.Publish(body)
+		assert.Nil(t, err)
+
+		ctx, cancel := context.WithCancel(context.Background())
+
+		go mq.RunConsumerForChannelWithCancelAndDbConn(subscriber.RunSubscriber, subscriptionChannel, database, ctx)
+		time.Sleep(time.Second * 2)
+		cancel()
+	}
+
+	for _, wanted := range wantedEvents {
+		result, err := database.GetSubscriptionData(wanted.Coin, []string{wanted.Address})
+		assert.Nil(t, err)
+		assert.Equal(t, result[0].SubscriptionId, wanted.Id)
+		assert.Equal(t, result[0].Coin, wanted.Coin)
+		assert.Equal(t, result[0].Address, wanted.Address)
+	}
+}
+
+func TestSubscriber_UpdateSubscription(t *testing.T) {
+	setup.CleanupPgContainer(database.DB)
+	_, goFile, _, _ := runtime.Caller(0)
+	testFilePathGiven := filepath.Join(filepath.Dir(goFile), "data", "given_subscriptions_deleted.json")
+	testFileGiven, err := ioutil.ReadFile(testFilePathGiven)
+	if err != nil {
+		t.Fatal(err)
+	}
+	var givenEvents []blockatlas.SubscriptionEvent
+	if err := json.Unmarshal(testFileGiven, &givenEvents); err != nil {
+		t.Fatal(err)
+	}
+
+	if err := json.Unmarshal(testFileGiven, &givenEvents); err != nil {
+		t.Fatal(err)
+	}
+
+	testFilePathWanted := filepath.Join(filepath.Dir(goFile), "data", "wanted_subscriptions_added.json")
+	testFileWanted, err := ioutil.ReadFile(testFilePathWanted)
+	if err != nil {
+		t.Fatal(err)
+	}
+	var wantedEvents []blockatlas.Subscription
+	if err := json.Unmarshal(testFileWanted, &wantedEvents); err != nil {
+		t.Fatal(err)
+	}
+
+	database.AddSubscriptions(10, []models.SubscriptionData{
+		{Coin: 61, Address: "0x0000000000000000000000000000000000000000", SubscriptionId: 10},
+	})
+	database.AddSubscriptions(1, []models.SubscriptionData{
+		{Coin: 62, Address: "0x0000000000000000000000000000000000000000", SubscriptionId: 1},
+	})
+	database.AddSubscriptions(2, []models.SubscriptionData{
+		{Coin: 63, Address: "0x0000000000000000000000000000000000000000", SubscriptionId: 2},
+	})
+	database.AddSubscriptions(3, []models.SubscriptionData{
+		{Coin: 64, Address: "0x0000000000000000000000000000000000000000", SubscriptionId: 3},
+	})
+
+	for _, event := range givenEvents {
+		body, err := json.Marshal(event)
+		assert.Nil(t, err)
+
+		err = mq.Subscriptions.Publish(body)
+		assert.Nil(t, err)
+
+		ctx, cancel := context.WithCancel(context.Background())
+
+		go mq.RunConsumerForChannelWithCancelAndDbConn(subscriber.RunSubscriber, subscriptionChannel, database, ctx)
+		time.Sleep(time.Second)
+		cancel()
+	}
+
+	for _, wanted := range wantedEvents {
+		result, err := database.GetSubscriptionData(wanted.Coin, []string{wanted.Address})
+		assert.Nil(t, err)
+		assert.Equal(t, result[0].SubscriptionId, wanted.Id)
+		assert.Equal(t, result[0].Coin, wanted.Coin)
+		assert.Equal(t, result[0].Address, wanted.Address)
+
+	}
+
+	abs61, err := database.GetSubscriptionData(61, []string{"0x0000000000000000000000000000000000000000"})
+	assert.Nil(t, err)
+	assert.Len(t, abs61, 0)
+
+	abs62, err := database.GetSubscriptionData(62, []string{"0x0000000000000000000000000000000000000000"})
+	assert.Nil(t, err)
+	assert.Len(t, abs62, 0)
+
+	abs63, err := database.GetSubscriptionData(63, []string{"0x0000000000000000000000000000000000000000"})
+	assert.Nil(t, err)
+	assert.Len(t, abs63, 0)
+
+	abs64, err := database.GetSubscriptionData(64, []string{"0x0000000000000000000000000000000000000000"})
+	assert.Nil(t, err)
+	assert.Len(t, abs64, 0)
+}
diff --git a/tests/integration/ontology/block.go b/tests/integration/ontology/block.go
deleted file mode 100755
index 7ffa7d2..0000000
--- a/tests/integration/ontology/block.go
+++ /dev/null
@@ -1,128 +0,0 @@
-// build integration
-
-package ontology
-
-import (
-	"github.com/trustwallet/blockatlas/coin"
-	"github.com/trustwallet/blockatlas/pkg/blockatlas"
-	"github.com/trustwallet/blockatlas/platform"
-	"github.com/trustwallet/blockatlas/platform/ontology"
-	"testing"
-)
-
-var (
-	testBlock = blockatlas.Block{
-		Number: 7707834,
-		Txs: []blockatlas.Tx{
-			{
-				From:   "AbEeCHUWpzQaxUN7G1a83N3P2XtVLuMLaE",
-				To:     "ASLbwuar3ZTbUbLPnCgjGUw2WHhMfvJJtx",
-				ID:     "266d9d7282a5601bf6cb8fc5368a76a2aa54f45731a063a699a692487bcbd0cb",
-				Fee:    "10000000",
-				Block:  7707834,
-				Status: blockatlas.StatusCompleted,
-				Date:   1580481541,
-				Coin:   coin.Ontology().ID,
-				Type:   blockatlas.TxNativeTokenTransfer,
-				Meta: blockatlas.AnyAction{
-					Name:     "Claim Rewards",
-					Symbol:   "ONG",
-					TokenID:  "ong",
-					Decimals: 9,
-					Value:    "51000000000000",
-					Title:    blockatlas.AnyActionClaimRewards,
-					Key:      blockatlas.KeyStakeClaimRewards,
-				},
-			}, {
-				From:   "ANdrA47zDXUu8MCkMdD3FYPmpSNGYeAvKz",
-				To:     "ASLbwuar3ZTbUbLPnCgjGUw2WHhMfvJJtx",
-				ID:     "2935268c5715f1f2015ba828681c39399dedbe7a24ed628ef7b85d9aac8045fd",
-				Fee:    "10000000",
-				Block:  7707834,
-				Status: blockatlas.StatusCompleted,
-				Date:   1580481541,
-				Coin:   coin.Ontology().ID,
-				Type:   blockatlas.TxNativeTokenTransfer,
-				Meta: blockatlas.AnyAction{
-					Name:     "Claim Rewards",
-					Symbol:   "ONG",
-					TokenID:  "ong",
-					Decimals: 9,
-					Value:    "113200000000",
-					Title:    blockatlas.AnyActionClaimRewards,
-					Key:      blockatlas.KeyStakeClaimRewards,
-				},
-			}, {
-				From:   "Abg2gs6pfpQu82jXbm8EYGiipRBvf9ktVS",
-				To:     "ASLbwuar3ZTbUbLPnCgjGUw2WHhMfvJJtx",
-				ID:     "40976edc1306b0e5f55b90c8d3ca248bb544e5ebbadb02be6146ba0a0de402c3",
-				Fee:    "10000000",
-				Block:  7707834,
-				Status: blockatlas.StatusCompleted,
-				Date:   1580481541,
-				Coin:   coin.Ontology().ID,
-				Type:   blockatlas.TxNativeTokenTransfer,
-				Meta: blockatlas.AnyAction{
-					Name:     "Claim Rewards",
-					Symbol:   "ONG",
-					TokenID:  "ong",
-					Decimals: 9,
-					Value:    "10949000000000",
-					Title:    blockatlas.AnyActionClaimRewards,
-					Key:      blockatlas.KeyStakeClaimRewards,
-				},
-			},
-		},
-	}
-)
-
-const (
-	blockNum = 7707834
-)
-
-func TestOntology(t *testing.T) {
-	t.Run("test ontology", func(t *testing.T) {
-		p := ontology.Init(platform.GetVar("ontology.api"))
-		testCurrentBlockNumber(p, t)
-		testGetBlockByNumber(p, t)
-	})
-}
-
-func testCurrentBlockNumber(p *ontology.Platform, t *testing.T) {
-	resp, err := p.CurrentBlockNumber()
-	if err != nil {
-		t.Error(err)
-	}
-	if resp < 0 {
-		t.Error("block is < 0")
-	}
-}
-
-func testGetBlockByNumber(p *ontology.Platform, t *testing.T) {
-	resp, err := p.GetBlockByNumber(int64(blockNum))
-	if err != nil {
-		t.Error(err)
-	}
-
-	isSame := resp.Number == testBlock.Number &&
-		resp.Txs[0].Block == testBlock.Txs[0].Block &&
-		resp.Txs[0].Status == testBlock.Txs[0].Status &&
-		resp.Txs[0].Date == testBlock.Txs[0].Date &&
-		resp.Txs[0].Coin == testBlock.Txs[0].Coin
-	if !isSame {
-		t.Errorf("Block is not the same")
-	}
-
-	// check that we have tx hashes of parsed block
-	txMap := map[string]bool{}
-	for _, tx := range resp.Txs {
-		txMap[tx.ID] = true
-	}
-	delete(txMap, "40976edc1306b0e5f55b90c8d3ca248bb544e5ebbadb02be6146ba0a0de402c3")
-	delete(txMap, "266d9d7282a5601bf6cb8fc5368a76a2aa54f45731a063a699a692487bcbd0cb")
-	delete(txMap, "2935268c5715f1f2015ba828681c39399dedbe7a24ed628ef7b85d9aac8045fd")
-
-	if len(txMap) > 0 {
-		t.Errorf("Block is not the same: %v", txMap)
-	}
-}
diff --git a/tests/docker_test/setup/mq.go b/tests/integration/setup/mq.go
similarity index 100%
rename from tests/docker_test/setup/mq.go
rename to tests/integration/setup/mq.go
diff --git a/tests/integration/setup/postgres.go b/tests/integration/setup/postgres.go
new file mode 100644
index 0000000..7390437
--- /dev/null
+++ b/tests/integration/setup/postgres.go
@@ -0,0 +1,81 @@
+package setup
+
+import (
+	"fmt"
+	"github.com/jinzhu/gorm"
+	"github.com/ory/dockertest"
+	"github.com/trustwallet/blockatlas/db"
+	"github.com/trustwallet/blockatlas/db/models"
+	"log"
+)
+
+const (
+	pgUser = "user"
+	pgPass = "pass"
+	pgDB   = "my_db"
+)
+
+var (
+	pgResource     *dockertest.Resource
+	pgContainerENV = []string{
+		"POSTGRES_USER=" + pgUser,
+		"POSTGRES_PASSWORD=" + pgPass,
+		"POSTGRES_DB=" + pgDB,
+	}
+
+	tables = []interface{}{
+		&models.Subscription{},
+		&models.SubscriptionData{},
+		&models.Tracker{},
+	}
+
+	uri string
+)
+
+func runPgContainerAndInitConnection() (*db.Instance, error) {
+	pool := runPgContainer()
+	var (
+		dbConn *db.Instance
+		err    error
+	)
+	if err := pool.Retry(func() error {
+		dbConn, err = db.New(uri)
+		return err
+	}); err != nil {
+		return nil, err
+	}
+	autoMigrate(dbConn.DB)
+
+	return dbConn, nil
+}
+
+func CleanupPgContainer(dbConn *gorm.DB) {
+	dbConn.DropTable(tables...)
+	autoMigrate(dbConn)
+}
+
+func autoMigrate(dbConn *gorm.DB) {
+	dbConn.AutoMigrate(tables...)
+}
+
+func stopPgContainer() error {
+	return pgResource.Close()
+}
+
+func runPgContainer() *dockertest.Pool {
+	var err error
+	pool, err := dockertest.NewPool("")
+	if err != nil {
+		log.Fatalf("Could not connect to docker: %s", err)
+	}
+
+	pgResource, err = pool.Run("postgres", "latest", pgContainerENV)
+	if err != nil {
+		log.Fatalf("Could not start resource: %s", err)
+	}
+
+	uri = fmt.Sprintf("postgres://%s:%s@localhost:%s/%s?sslmode=disable",
+		pgUser, pgPass, pgResource.GetPort("5432/tcp"), pgDB,
+	)
+	return pool
+}
diff --git a/tests/integration/setup/setup.go b/tests/integration/setup/setup.go
new file mode 100644
index 0000000..0e916c8
--- /dev/null
+++ b/tests/integration/setup/setup.go
@@ -0,0 +1,32 @@
+package setup
+
+import (
+	"github.com/trustwallet/blockatlas/db"
+	"log"
+)
+
+func RunMQContainer() {
+	if err := runMQContainer(); err != nil {
+		log.Fatal(err)
+	}
+}
+
+func StopMQContainer() {
+	if err := stopMQContainer(); err != nil {
+		log.Fatal(err)
+	}
+}
+
+func RunPgContainer() *db.Instance {
+	dbConn, err := runPgContainerAndInitConnection()
+	if err != nil {
+		log.Fatal(err)
+	}
+	return dbConn
+}
+
+func StopPgContainer() {
+	if err := stopPgContainer(); err != nil {
+		log.Fatal(err)
+	}
+}