Repository: trustwallet/blockatlas
Pull request #941 (2020-03-01 00:17:23)

Change subscription api. Now we will subscribe users by using MQ


Patch:
diff --git a/Makefile b/Makefile
index 49451de..b05a97f 100644
--- a/Makefile
+++ b/Makefile
@@ -7,8 +7,8 @@ BUILD := $(shell git rev-parse --short HEAD)
 DATETIME := $(shell date +"%Y.%m.%d-%H:%M:%S")
 PROJECT_NAME := $(shell basename "$(PWD)")
 API_SERVICE := platform_api
-OBSERVER_SERVICE := platform_observer
-OBSERVER_API := observer_api
+OBSERVER_SERVICE := observer_worker
+OBSERVER_SUBSCRIBER := observer_subscriber
 SWAGGER_API := swagger_api
 COIN_FILE := coin/coins.yml
 COIN_GO_FILE := coin/coins.go
@@ -34,7 +34,7 @@ STDERR := /tmp/.$(PROJECT_NAME)-stderr.txt
 # PID file will keep the process id of the server
 PID_API := /tmp/.$(PROJECT_NAME).$(API_SERVICE).pid
 PID_OBSERVER := /tmp/.$(PROJECT_NAME).$(OBSERVER_SERVICE).pid
-PID_OBSERVER_API := /tmp/.$(PROJECT_NAME).$(OBSERVER_API).pid
+PID_OBSERVER_SUBSCRIBER := /tmp/.$(PROJECT_NAME).$(OBSERVER_SUBSCRIBER).pid
 PID_SWAGGER_API := /tmp/.$(PROJECT_NAME).$(SWAGGER_API).pid
 # Make is verbose in Linux. Make it silent.
 MAKEFLAGS += --silent
@@ -63,8 +63,8 @@ start-platform-observer: stop
 ## start-observer: Start Observer in development mode.
 start-observer-api: stop
 	@echo "  >  Starting $(PROJECT_NAME) Observer"
-	@-$(GOBIN)/$(OBSERVER_API)/observer_api -c $(CONFIG_FILE) 2>&1 & echo $$! > $(PID_OBSERVER_API)
-	@cat $(PID_OBSERVER_API) | sed "/^/s/^/  \>  Observer PID: /"
+	@-$(GOBIN)/$(OBSERVER_SUBSCRIBER)/observer_api -c $(CONFIG_FILE) 2>&1 & echo $$! > $(PID_OBSERVER_SUBSCRIBER)
+	@cat $(PID_OBSERVER_SUBSCRIBER) | sed "/^/s/^/  \>  Observer PID: /"
 	@echo "  >  Error log: $(STDERR)"
 
 ## start-sync-market-api: Start Sync market api in development mode.
@@ -76,12 +76,12 @@ start-swagger-api: stop
 
 ## stop: Stop development mode.
 stop:
-	@-touch $(PID_API) $(PID_OBSERVER) $(PID_OBSERVER_API) $(PID_SWAGGER_API)
+	@-touch $(PID_API) $(PID_OBSERVER) $(PID_OBSERVER_SUBSCRIBER) $(PID_SWAGGER_API)
 	@-kill `cat $(PID_API)` 2> /dev/null || true
 	@-kill `cat $(PID_OBSERVER)` 2> /dev/null || true
-	@-kill `cat $(PID_OBSERVER_API)` 2> /dev/null || true
+	@-kill `cat $(PID_OBSERVER_SUBSCRIBER)` 2> /dev/null || true
 	@-kill `cat $(PID_SWAGGER_API)` 2> /dev/null || true
-	@-rm $(PID_API) $(PID_OBSERVER) $(PID_OBSERVER_API) $(PID_SWAGGER_API)
+	@-rm $(PID_API) $(PID_OBSERVER) $(PID_OBSERVER_SUBSCRIBER) $(PID_SWAGGER_API)
 
 ## compile: Compile the project.
 compile:
@@ -152,7 +152,6 @@ ifeq (,$(test))
 	@bash -c "$(MAKE) newman test=collection host=$(host)"
 	@bash -c "$(MAKE) newman test=domain host=$(host)"
 	@bash -c "$(MAKE) newman test=healthcheck host=$(host)"
-	@bash -c "$(MAKE) newman test=observer host=$(host)"
 else
 	@newman run tests/postman/Blockatlas.postman_collection.json --folder $(test) -d tests/postman/$(test)_data.json --env-var "host=$(host)"
 endif
@@ -162,10 +161,10 @@ go-compile: go-get go-build
 go-build:
 	@echo "  >  Building platform_api binary..."
 	GOBIN=$(GOBIN) go build $(LDFLAGS) -o $(GOBIN)/$(API_SERVICE)/platform_api ./cmd/$(API_SERVICE)
-	@echo "  >  Building platform_observer binary..."
-	GOBIN=$(GOBIN) go build $(LDFLAGS) -o $(GOBIN)/$(OBSERVER_SERVICE)/platform_observer ./cmd/$(OBSERVER_SERVICE)
-	@echo "  >  Building observer_api binary..."
-	GOBIN=$(GOBIN) go build $(LDFLAGS) -o $(GOBIN)/$(OBSERVER_API)/observer_api ./cmd/$(OBSERVER_API)
+	@echo "  >  Building observer_worker binary..."
+	GOBIN=$(GOBIN) go build $(LDFLAGS) -o $(GOBIN)/$(OBSERVER_SERVICE)/observer_worker ./cmd/$(OBSERVER_SERVICE)
+	@echo "  >  Building observer_subscriber binary..."
+	GOBIN=$(GOBIN) go build $(LDFLAGS) -o $(GOBIN)/$(observer_subscriber)/observer_subscriber ./cmd/$(OBSERVER_SUBSCRIBER)
 	@echo "  >  Building swagger_api binary..."
 	GOBIN=$(GOBIN) go build $(LDFLAGS) -o $(GOBIN)/$(SWAGGER_API)/swagger_api ./cmd/$(SWAGGER_API)
 
diff --git a/api/observer.go b/api/observer.go
deleted file mode 100644
index 160832f..0000000
--- a/api/observer.go
+++ /dev/null
@@ -1,106 +0,0 @@
-package api
-
-import (
-	"github.com/gin-gonic/gin"
-	"github.com/trustwallet/blockatlas/pkg/blockatlas"
-	"github.com/trustwallet/blockatlas/pkg/ginutils"
-	"github.com/trustwallet/blockatlas/platform"
-	"github.com/trustwallet/blockatlas/storage"
-)
-
-// @Summary Create a webhook
-// @ID create_webhook
-// @Description Create a webhook for addresses transactions
-// @Accept json
-// @Produce json
-// @Tags Observer
-// @Param subscriptions body blockatlas.Webhook true "Accounts subscriptions"
-// @Param Authorization header string true "Bearer authorization header" default(Bearer test)
-// @Header 200 {string} Authorization {token}
-// @Success 200 {object} blockatlas.Observer
-// @Router /observer/v1/webhook/register [post]
-func addCall(storage storage.Addresses) func(c *gin.Context) {
-	if storage == nil {
-		return nil
-	}
-	return func(c *gin.Context) {
-		var req blockatlas.Webhook
-		if c.BindJSON(&req) != nil {
-			return
-		}
-
-		if len(req.Subscriptions) == 0 {
-			ginutils.RenderSuccess(c, blockatlas.Observer{Message: "Added", Status: true})
-			return
-		}
-		subs := req.ParseSubscriptions()
-		go storage.AddSubscriptions(subs)
-
-		ginutils.RenderSuccess(c, blockatlas.Observer{Message: "Added", Status: true})
-	}
-}
-
-// @Summary Delete a webhook
-// @ID delete_webhook
-// @Description Delete a webhook for addresses transactions
-// @Accept json
-// @Produce json
-// @Tags Observer
-// @Param subscriptions body blockatlas.Webhook true "Accounts subscriptions"
-// @Param Authorization header string true "Bearer authorization header" default(Bearer test)
-// @Header 200 {string} Authorization {token}
-// @Success 200 {object} blockatlas.Observer
-// @Router /observer/v1/webhook/register [delete]
-func deleteCall(storage storage.Addresses) func(c *gin.Context) {
-	if storage == nil {
-		return nil
-	}
-	return func(c *gin.Context) {
-		var req blockatlas.Webhook
-		if c.BindJSON(&req) != nil {
-			return
-		}
-
-		if len(req.Subscriptions) == 0 {
-			ginutils.RenderSuccess(c, blockatlas.Observer{Message: "Deleted", Status: true})
-			return
-		}
-
-		subs := req.ParseSubscriptions()
-		go storage.DeleteSubscriptions(subs)
-		ginutils.RenderSuccess(c, blockatlas.Observer{Message: "Deleted", Status: true})
-	}
-}
-
-// @Summary Get coin status
-// @ID coin_status
-// @Description Get coin status
-// @Accept json
-// @Produce json
-// @Tags Observer
-// @Param Authorization header string true "Bearer authorization header" default(Bearer test)
-// @Header 200 {string} Authorization {token}
-// @Success 200 {object} blockatlas.CoinStatus
-// @Router /observer/v1/status [get]
-func statusCall(storage storage.Tracker) func(c *gin.Context) {
-	if storage == nil {
-		return nil
-	}
-	return func(c *gin.Context) {
-		result := make(map[string]blockatlas.CoinStatus)
-		for _, api := range platform.BlockAPIs {
-			coin := api.Coin()
-			num, err := storage.GetBlockNumber(coin.ID)
-			var status blockatlas.CoinStatus
-			if err != nil {
-				status = blockatlas.CoinStatus{Error: err.Error()}
-			} else if num == 0 {
-				status = blockatlas.CoinStatus{Error: "no blocks"}
-			} else {
-				status = blockatlas.CoinStatus{Height: num}
-			}
-			result[coin.Handle] = status
-		}
-		ginutils.RenderSuccess(c, result)
-	}
-}
diff --git a/api/routes.go b/api/routes.go
index c1ce1ba..c843a21 100644
--- a/api/routes.go
+++ b/api/routes.go
@@ -3,27 +3,14 @@ package api
 import (
 	"fmt"
 	"github.com/gin-gonic/gin"
-	"github.com/spf13/viper"
 	"github.com/trustwallet/blockatlas/internal"
 	"github.com/trustwallet/blockatlas/pkg/ginutils"
 	"github.com/trustwallet/blockatlas/pkg/logger"
 	"github.com/trustwallet/blockatlas/platform"
-	"github.com/trustwallet/blockatlas/storage"
 )
 
 var routers = make(map[string]gin.IRouter)
 
-func SetupObserverAPI(router gin.IRouter, db *storage.Storage) {
-	router.GET("/", GetRoot)
-	router.GET("/status", GetStatus)
-
-	observerAPI := router.Group("/observer/v1")
-	observerAPI.Use(ginutils.TokenAuthMiddleware(viper.GetString("observer.auth")))
-	observerAPI.POST("/webhook/register", addCall(db))
-	observerAPI.DELETE("/webhook/register", deleteCall(db))
-	observerAPI.GET("/status", statusCall(db))
-}
-
 func SetupPlatformAPI(root gin.IRouter) {
 	root.GET("/", GetRoot)
 	root.GET("/status", GetStatus)
diff --git a/cmd/observer_api/main.go b/cmd/observer_api/main.go
deleted file mode 100644
index ce72ea5..0000000
--- a/cmd/observer_api/main.go
+++ /dev/null
@@ -1,39 +0,0 @@
-package main
-
-import (
-	"github.com/gin-gonic/gin"
-	"github.com/spf13/viper"
-	"github.com/trustwallet/blockatlas/api"
-	_ "github.com/trustwallet/blockatlas/docs"
-	"github.com/trustwallet/blockatlas/internal"
-	"github.com/trustwallet/blockatlas/pkg/ginutils"
-	"github.com/trustwallet/blockatlas/storage"
-)
-
-const (
-	defaultPort       = "8422"
-	defaultConfigPath = "../../config.yml"
-)
-
-var (
-	port, confPath string
-	cache          *storage.Storage
-	sg             *gin.HandlerFunc
-)
-
-func init() {
-	port, confPath, sg, cache = internal.InitAPIWithRedis(defaultPort, defaultConfigPath)
-}
-
-func main() {
-	gin.SetMode(viper.GetString("gin.mode"))
-
-	engine := gin.New()
-	engine.Use(ginutils.CheckReverseProxy, *sg)
-	engine.Use(ginutils.CORSMiddleware())
-	engine.Use(gin.Logger())
-	engine.OPTIONS("/*path", ginutils.CORSMiddleware())
-
-	api.SetupObserverAPI(engine, cache)
-	internal.SetupGracefulShutdown(port, engine)
-}
diff --git a/cmd/observer_subscriber/main.go b/cmd/observer_subscriber/main.go
new file mode 100644
index 0000000..7e32970
--- /dev/null
+++ b/cmd/observer_subscriber/main.go
@@ -0,0 +1,45 @@
+package main
+
+import (
+	"github.com/gin-gonic/gin"
+	"github.com/spf13/viper"
+	_ "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/platform"
+	"github.com/trustwallet/blockatlas/services/subscription"
+	"github.com/trustwallet/blockatlas/storage"
+)
+
+const (
+	defaultConfigPath = "../../config.yml"
+)
+
+var (
+	port, confPath string
+	cache          *storage.Storage
+	sg             *gin.HandlerFunc
+)
+
+func init() {
+	_, confPath, _, cache = internal.InitAPIWithRedis("", defaultConfigPath)
+
+	uri := viper.GetString("observer.rabbitmq.uri")
+	err := mq.Init(uri)
+	if err != nil {
+		logger.Fatal("Failed to init Rabbit MQ", logger.Params{"uri": uri})
+	}
+	err = mq.Transactions.Declare()
+	if err != nil {
+		logger.Fatal("Failed to init Rabbit MQ", logger.Params{"uri": uri})
+	}
+
+	platform.Init(viper.GetString("platform"))
+}
+
+func main() {
+	defer mq.Close()
+	mq.Subscriptions.RunConsumer(subscription.Consume, cache)
+	<-make(chan struct{})
+}
diff --git a/cmd/platform_observer/main.go b/cmd/observer_worker/main.go
similarity index 93%
rename from cmd/platform_observer/main.go
rename to cmd/observer_worker/main.go
index a92a72c..2809ee8 100644
--- a/cmd/platform_observer/main.go
+++ b/cmd/observer_worker/main.go
@@ -18,16 +18,22 @@ const (
 
 var (
 	confPath string
-	cache                      *storage.Storage
+	cache    *storage.Storage
 )
 
 func init() {
 	_, confPath, _, cache = internal.InitAPIWithRedis("", defaultConfigPath)
+
 	uri := viper.GetString("observer.rabbitmq.uri")
 	err := mq.Init(uri)
 	if err != nil {
 		logger.Fatal("Failed to init Rabbit MQ", logger.Params{"uri": uri})
 	}
+	err = mq.Transactions.Declare()
+	if err != nil {
+		logger.Fatal("Failed to init Rabbit MQ", logger.Params{"uri": uri})
+	}
+
 	platform.Init(viper.GetString("platform"))
 }
 
diff --git a/cmd/platform_api/main.go b/cmd/platform_api/main.go
index 5072204..26ac439 100644
--- a/cmd/platform_api/main.go
+++ b/cmd/platform_api/main.go
@@ -32,7 +32,6 @@ func main() {
 	engine.Use(ginutils.CheckReverseProxy, *sg)
 	engine.Use(ginutils.CORSMiddleware())
 	engine.Use(gin.Logger())
-	engine.OPTIONS("/*path", ginutils.CORSMiddleware())
 
 	api.SetupPlatformAPI(engine)
 	internal.SetupGracefulShutdown(port, engine)
diff --git a/cmd/swagger_api/main.go b/cmd/swagger_api/main.go
index 660821c..2d2c4b4 100644
--- a/cmd/swagger_api/main.go
+++ b/cmd/swagger_api/main.go
@@ -32,7 +32,6 @@ func main() {
 	engine.Use(ginutils.CheckReverseProxy, *sg)
 	engine.Use(ginutils.CORSMiddleware())
 
-	engine.OPTIONS("/*path", ginutils.CORSMiddleware())
 	engine.GET("/", api.GetRoot)
 	engine.GET("/status", api.GetStatus)
 	engine.GET("swagger/*any", ginSwagger.WrapHandler(swaggerFiles.Handler))
diff --git a/docs/docs.go b/docs/docs.go
index 99b46c3..b3caf16 100644
--- a/docs/docs.go
+++ b/docs/docs.go
@@ -122,7 +122,7 @@ var doc = `{
                         "in": "body",
                         "required": true,
                         "schema": {
-                            "$ref": "#/definitions/blockatlas.Webhook"
+                            "$ref": "#/definitions/blockatlas.GUID"
                         }
                     },
                     {
@@ -163,7 +163,7 @@ var doc = `{
                         "in": "body",
                         "required": true,
                         "schema": {
-                            "$ref": "#/definitions/blockatlas.Webhook"
+                            "$ref": "#/definitions/blockatlas.GUID"
                         }
                     },
                     {
@@ -1603,7 +1603,7 @@ var doc = `{
                 "$ref": "#/definitions/blockatlas.Tx"
             }
         },
-        "blockatlas.Webhook": {
+        "blockatlas.GUID": {
             "type": "object",
             "properties": {
                 "subscriptions": {
diff --git a/mq/mq.go b/mq/mq.go
index 3538222..4aa00c6 100644
--- a/mq/mq.go
+++ b/mq/mq.go
@@ -2,6 +2,8 @@ package mq
 
 import (
 	"github.com/streadway/amqp"
+	"github.com/trustwallet/blockatlas/pkg/logger"
+	"github.com/trustwallet/blockatlas/storage"
 )
 
 var (
@@ -10,9 +12,15 @@ var (
 	queue    amqp.Queue
 )
 
-type QueueName string
+type (
+	Queue    string
+	Consumer func(amqp.Delivery, storage.Addresses)
+)
 
-const Notifications QueueName = "transactions"
+const (
+	Transactions  Queue = "transactions"
+	Subscriptions Queue = "subscriptions"
+)
 
 func Init(uri string) (err error) {
 	conn, err = amqp.Dial(uri)
@@ -23,7 +31,6 @@ func Init(uri string) (err error) {
 	if err != nil {
 		return
 	}
-	queue, err = amqpChan.QueueDeclare(string(Notifications), true, false, false, false, nil)
 	return
 }
 
@@ -32,10 +39,35 @@ func Close() {
 	conn.Close()
 }
 
-func Publish(body []byte) error {
-	return amqpChan.Publish("", queue.Name, false, false, amqp.Publishing{
+func (q Queue) Declare() error {
+	_, err := amqpChan.QueueDeclare(string(q), true, false, false, false, nil)
+	return err
+}
+
+func (q Queue) Publish(body []byte) error {
+	return amqpChan.Publish("", string(q), false, false, amqp.Publishing{
 		DeliveryMode: amqp.Persistent,
 		ContentType:  "text/plain",
 		Body:         body,
 	})
 }
+
+func (q Queue) RunConsumer(consumer Consumer, cache storage.Addresses) {
+	messageChannel, err := amqpChan.Consume(
+		string(q),
+		"",
+		false,
+		false,
+		false,
+		false,
+		nil,
+	)
+	if err != nil {
+		logger.Error(err)
+		return
+	}
+
+	for data := range messageChannel {
+		consumer(data, cache)
+	}
+}
diff --git a/observer/dispatcher.go b/observer/dispatcher.go
index 397073b..e68c40a 100644
--- a/observer/dispatcher.go
+++ b/observer/dispatcher.go
@@ -51,7 +51,7 @@ func (d *Dispatcher) dispatch(event Event) {
 }
 
 func (d *Dispatcher) postMessageToQueue(message string, rawMessage []byte, logParams logger.Params) {
-	err := mq.Publish(rawMessage)
+	err := mq.Transactions.Publish(rawMessage)
 	if err != nil {
 		err = errors.E(err, "Failed to dispatch event", errors.Params{"message": message}, logParams)
 		logger.Error(err, logger.Params{"message": message}, logParams)
diff --git a/pkg/blockatlas/observer.go b/pkg/blockatlas/observer.go
index f46cc11..abb1f1f 100644
--- a/pkg/blockatlas/observer.go
+++ b/pkg/blockatlas/observer.go
@@ -4,9 +4,12 @@ import "strconv"
 
 type Subscriptions map[string][]string
 
-type Webhook struct {
-	Subscriptions Subscriptions `json:"subscriptions"`
-	Webhook       string        `json:"webhook"`
+type SubscriptionOperation string
+
+type SubscriptionEvent struct {
+	Subscriptions Subscriptions         `json:"subscriptions"`
+	GUID          string                `json:"guid"`
+	Operation     SubscriptionOperation `json:"operation"`
 }
 
 type CoinStatus struct {
@@ -19,9 +22,9 @@ type Observer struct {
 	Message string `json:"message"`
 }
 
-func (w *Webhook) ParseSubscriptions() []Subscription {
+func (e *SubscriptionEvent) ParseSubscriptions() []Subscription {
 	subs := make([]Subscription, 0)
-	for coinStr, perCoin := range w.Subscriptions {
+	for coinStr, perCoin := range e.Subscriptions {
 		coin, err := strconv.Atoi(coinStr)
 		if err != nil {
 			continue
@@ -30,7 +33,7 @@ func (w *Webhook) ParseSubscriptions() []Subscription {
 			subs = append(subs, Subscription{
 				Coin:    uint(coin),
 				Address: addr,
-				Webhook: w.Webhook,
+				Webhook: e.GUID,
 			})
 		}
 	}
diff --git a/services/subscription/subscribtion.go b/services/subscription/subscribtion.go
new file mode 100644
index 0000000..6f79d6c
--- /dev/null
+++ b/services/subscription/subscribtion.go
@@ -0,0 +1,36 @@
+package subscription
+
+import (
+	"encoding/json"
+	"github.com/streadway/amqp"
+	"github.com/trustwallet/blockatlas/pkg/blockatlas"
+	"github.com/trustwallet/blockatlas/pkg/logger"
+	"github.com/trustwallet/blockatlas/storage"
+)
+
+const (
+	addSubscription    blockatlas.SubscriptionOperation = "AddSubscription"
+	deleteSubscription blockatlas.SubscriptionOperation = "DeleteSubscription"
+)
+
+func Consume(delivery amqp.Delivery, storage storage.Addresses) {
+	var event blockatlas.SubscriptionEvent
+	err := json.Unmarshal(delivery.Body, &event)
+	if err != nil {
+		logger.Fatal(err)
+	}
+	subscription := event.ParseSubscriptions()
+
+	switch event.Operation {
+	case addSubscription:
+		err := storage.AddSubscriptions(subscription)
+		if err != nil {
+			logger.Fatal(err, logger.Params{"operation": event.Operation, "guid": event.GUID})
+		}
+	case deleteSubscription:
+		err := storage.DeleteSubscriptions(nil)
+		if err != nil {
+			logger.Fatal(logger.Params{"operation": event.Operation, "guid": event.GUID})
+		}
+	}
+}
diff --git a/storage/addresses.go b/storage/addresses.go
index 760bebc..cc7e54f 100644
--- a/storage/addresses.go
+++ b/storage/addresses.go
@@ -4,7 +4,6 @@ import (
 	"fmt"
 	"github.com/trustwallet/blockatlas/pkg/blockatlas"
 	"github.com/trustwallet/blockatlas/pkg/errors"
-	"github.com/trustwallet/blockatlas/pkg/logger"
 )
 
 const (
@@ -30,11 +29,11 @@ func (s *Storage) Lookup(coin uint, addresses []string) ([]blockatlas.Subscripti
 	return observers, nil
 }
 
-func (s *Storage) AddSubscriptions(subscriptions []blockatlas.Subscription) {
+func (s *Storage) AddSubscriptions(subscriptions []blockatlas.Subscription) error {
 	for _, sub := range subscriptions {
 		key := getSubscriptionKey(sub.Coin, sub.Address)
 		var webhooks []string
-		_ = s.GetHMValue(ATLAS_OBSERVER, key, &webhooks)
+		s.GetHMValue(ATLAS_OBSERVER, key, &webhooks)
 		if webhooks == nil {
 			webhooks = make([]string, 0)
 		}
@@ -44,12 +43,13 @@ func (s *Storage) AddSubscriptions(subscriptions []blockatlas.Subscription) {
 		webhooks = append(webhooks, sub.Webhook)
 		err := s.AddHM(ATLAS_OBSERVER, key, webhooks)
 		if err != nil {
-			logger.Error(err, "AddSubscriptions error", errors.Params{"webhooks": webhooks, "address": sub.Address, "coin": sub.Coin})
+			return err
 		}
 	}
+	return nil
 }
 
-func (s *Storage) DeleteSubscriptions(subscriptions []blockatlas.Subscription) {
+func (s *Storage) DeleteSubscriptions(subscriptions []blockatlas.Subscription) error {
 	for _, sub := range subscriptions {
 		key := getSubscriptionKey(sub.Coin, sub.Address)
 		var webhooks []string
@@ -69,10 +69,9 @@ func (s *Storage) DeleteSubscriptions(subscriptions []blockatlas.Subscription) {
 			continue
 		}
 		err = s.AddHM(ATLAS_OBSERVER, key, newHooks)
-		if err != nil {
-			logger.Error(err, "DeleteSubscriptions - AddHM", errors.Params{"webhook": newHooks, "address": sub.Address, "coin": sub.Coin})
-		}
+		return err
 	}
+	return nil
 }
 
 func getSubscriptionKey(coin uint, address string) string {
diff --git a/storage/storage.go b/storage/storage.go
index 40b8afd..9182843 100644
--- a/storage/storage.go
+++ b/storage/storage.go
@@ -23,6 +23,6 @@ type Tracker interface {
 
 type Addresses interface {
 	Lookup(coin uint, addresses []string) ([]blockatlas.Subscription, error)
-	AddSubscriptions(subscriptions []blockatlas.Subscription)
-	DeleteSubscriptions(subscriptions []blockatlas.Subscription)
+	AddSubscriptions(subscriptions []blockatlas.Subscription) error
+	DeleteSubscriptions(subscriptions []blockatlas.Subscription) error
 }


Test patch:
diff --git a/pkg/blockatlas/observer_test.go b/pkg/blockatlas/observer_test.go
index 00d73b2..7d488c1 100644
--- a/pkg/blockatlas/observer_test.go
+++ b/pkg/blockatlas/observer_test.go
@@ -9,13 +9,13 @@ import (
 func Test_parseSubscriptions(t *testing.T) {
 	tests := []struct {
 		name          string
-		subscriptions Webhook
+		subscriptions SubscriptionEvent
 		wantSubs      []Subscription
 	}{
 		{
 			name: "webhook with 1 coin",
-			subscriptions: Webhook{
-				Webhook: "http://127.0.0.1:8080",
+			subscriptions: SubscriptionEvent{
+				GUID: "http://127.0.0.1:8080",
 				Subscriptions: Subscriptions{
 					"0": {"xpub6BpYi6J1GZzfY3yY7DbhLLccF3efQa18nQngM3jaehgtNSoEgk6UtPULpC3oK5oA3trczY8Ld34LFw1USMPfGHwTEizdD5QyGcMyuh2UoBA", "xpub6CYwPfnPJLPquufPkb98coSb3mdy1CgaZrWUtYWGJTJ4VWZUbzH9HLGy7nHpP7DG4UdTkYYpirkTWQSP7pWHsrk24Nos5oYNHpfr4BgPVTL"},
 				},
@@ -33,8 +33,8 @@ func Test_parseSubscriptions(t *testing.T) {
 		},
 		{
 			name: "webhook with 2 coins",
-			subscriptions: Webhook{
-				Webhook: "http://127.0.0.1:8080",
+			subscriptions: SubscriptionEvent{
+				GUID: "http://127.0.0.1:8080",
 				Subscriptions: Subscriptions{
 					"2": {"zpub6rH4MwgyTmuexAX6HAraks5cKv5BbtmwdLirvnU5845ovUJb4abgjt9DtXK4ZEaToRrNj8dQznuLC6Nka4eMviGMinCVMUxKLpuyddcG9Vc"},
 					"0": {"xpub6BpYi6J1GZzfY3yY7DbhLLccF3efQa18nQngM3jaehgtNSoEgk6UtPULpC3oK5oA3trczY8Ld34LFw1USMPfGHwTEizdD5QyGcMyuh2UoBA", "xpub6CYwPfnPJLPquufPkb98coSb3mdy1CgaZrWUtYWGJTJ4VWZUbzH9HLGy7nHpP7DG4UdTkYYpirkTWQSP7pWHsrk24Nos5oYNHpfr4BgPVTL"},
diff --git a/services/subscription/subscription_test.go b/services/subscription/subscription_test.go
new file mode 100644
index 0000000..a52d129
--- /dev/null
+++ b/services/subscription/subscription_test.go
@@ -0,0 +1,32 @@
+package subscription
+
+import (
+	"testing"
+)
+
+func TestGetInterval(t *testing.T) {
+	//cache := storage.New()
+	//err := cache.Init("redis://localhost:6379")
+	//if err != nil {
+	//	logger.Fatal(err)
+	//}
+	//
+	//var subscriptions = blockatlas.Subscriptions{
+	//	"2": {"COINADDRESS 0"},
+	//	"0": {"ADDRES 1", "aDDRESS 2"},
+	//}
+	//
+	//var mockedData = blockatlas.SubscriptionEvent{
+	//	Subscriptions: subscriptions,
+	//	GUID:          "EXAMPLE",
+	//	Operation:     addSubscription,
+	//}
+
+	//	b, _ := json.Marshal(mockedData)
+	//
+	//	//success := Consume(b, cache)
+	//	if !success {
+	//		t.Fatal()
+	//	}
+	//}
+}