Repository: trustwallet/blockatlas
Pull request #932 (2020-02-27 18:23:25)

Adopt to new trust-ray token response
Adopt to trust ray new response where no contract and unnecessary nested structure.


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/README.md b/README.md
index 1a1eab1..1d77263 100644
--- a/README.md
+++ b/README.md
@@ -13,7 +13,7 @@ BlockAtlas connects to nodes or explorer APIs of the supported coins and maps tr
 account transaction history into a generic, easy to work with JSON format.
 It is in production use at the [Trust Wallet app](https://trustwallet.com/), 
 the official cryptocurrency wallet of Binance. Also is in production at the [BUTTON Wallet](https://buttonwallet.com), Telegram based non-custodial wallet.
-The observer API watches the chain for new transactions and generates notifications by webhooks.
+The observer API watches the chain for new transactions and generates notifications by guids.
 
 #### Supported Coins
 
@@ -51,12 +51,12 @@ The observer API watches the chain for new transactions and generates notificati
 There are multiple services:
 
 1. Platform API - to get transactions, staking, tokens, domain lookup for supported coins in common format
-2. Observer API - to subscribe several addresses on different supported coins and receive webhook
+2. Observer API - to subscribe several addresses on different supported coins and receive message
 3. Swagger API - swagger for all handlers of 1-3 APIs. You need to route requests to them on you own (nginx)
 
 There are workers that are linked with Observer API and Market API:
 
-5. Platform Observer - fetching latest blocks, parse them to common block specification, check subscribed addresses - send webhook. We use Redis to get information about subscribed addresses per coin with webhooks and caching latest block that was processed by observer
+5. Platform Observer - fetching latest blocks, parse them to common block specification, check subscribed addresses - send message to queue. We use Redis to get information about subscribed addresses per coin with webhooks and caching latest block that was processed by observer
 
 Observer API <-> Redis <-> Platform Observer
 
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..8713bc4
--- /dev/null
+++ b/cmd/observer_subscriber/main.go
@@ -0,0 +1,44 @@
+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/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})
+	}
+
+}
+
+func main() {
+	defer mq.Close()
+	mq.Subscriptions.RunConsumer(subscription.Consume, cache)
+	<-make(chan struct{})
+	mq.Close()
+}
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..e126abe 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"))
 }
 
@@ -90,6 +96,6 @@ func main() {
 	}
 
 	wg.Wait()
-
+	mq.Close()
 	logger.Info("Exiting cleanly")
 }
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/docker-compose.yml b/docker-compose.yml
index 915d746..cae42de 100644
--- a/docker-compose.yml
+++ b/docker-compose.yml
@@ -11,18 +11,6 @@ services:
       - observer_api
       - platform_api
       - swagger_api
- 
-  observer_api:
-    container_name: observer_api
-    build:
-      context: .
-      args:
-        - SERVICE=observer_api
-    ports:
-      - 8422:8422
-    links:
-      - redis
-    restart: on-failure
 
   platform_api:
     container_name: platform_api
@@ -31,7 +19,7 @@ services:
       args:
         - SERVICE=platform_api
     ports:
-    - 8420:8420
+      - 8420:8420
 
   swagger_api:
     container_name: swagger_api
@@ -42,11 +30,22 @@ services:
     ports:
       - 8423:8423
 
-  platform_observer:
+  observer_worker:
+    build:
+      context: .
+      args:
+        - SERVICE=observer_worker
+    links:
+      - redis
+      - rabbit
+    restart: on-failure
+
+  observer_subscriber:
+    container_name: observer_subscriber
     build:
       context: .
       args:
-        - SERVICE=platform_observer
+        - SERVICE=observer_subscriber
     links:
       - redis
       - rabbit
@@ -67,4 +66,4 @@ services:
     container_name: redis
     image: redis
     ports:
-    - 6379:6379
+      - 6379:6379
diff --git a/docs/docs.go b/docs/docs.go
index 99b46c3..247a273 100644
--- a/docs/docs.go
+++ b/docs/docs.go
@@ -1,6 +1,6 @@
 // GENERATED BY THE COMMAND ABOVE; DO NOT EDIT
 // This file was generated by swaggo/swag at
-// 2020-02-26 03:37:11.056244 +0300 MSK m=+0.121112696
+// 2020-03-01 04:39:26.356879 +0300 MSK m=+1.896562991
 
 package docs
 
@@ -67,124 +67,6 @@ var doc = `{
                 }
             }
         },
-        "/observer/v1/status": {
-            "get": {
-                "description": "Get coin status",
-                "consumes": [
-                    "application/json"
-                ],
-                "produces": [
-                    "application/json"
-                ],
-                "tags": [
-                    "Observer"
-                ],
-                "summary": "Get coin status",
-                "operationId": "coin_status",
-                "parameters": [
-                    {
-                        "type": "string",
-                        "default": "Bearer test",
-                        "description": "Bearer authorization header",
-                        "name": "Authorization",
-                        "in": "header",
-                        "required": true
-                    }
-                ],
-                "responses": {
-                    "200": {
-                        "description": "OK",
-                        "schema": {
-                            "$ref": "#/definitions/blockatlas.CoinStatus"
-                        }
-                    }
-                }
-            }
-        },
-        "/observer/v1/webhook/register": {
-            "post": {
-                "description": "Create a webhook for addresses transactions",
-                "consumes": [
-                    "application/json"
-                ],
-                "produces": [
-                    "application/json"
-                ],
-                "tags": [
-                    "Observer"
-                ],
-                "summary": "Create a webhook",
-                "operationId": "create_webhook",
-                "parameters": [
-                    {
-                        "description": "Accounts subscriptions",
-                        "name": "subscriptions",
-                        "in": "body",
-                        "required": true,
-                        "schema": {
-                            "$ref": "#/definitions/blockatlas.Webhook"
-                        }
-                    },
-                    {
-                        "type": "string",
-                        "default": "Bearer test",
-                        "description": "Bearer authorization header",
-                        "name": "Authorization",
-                        "in": "header",
-                        "required": true
-                    }
-                ],
-                "responses": {
-                    "200": {
-                        "description": "OK",
-                        "schema": {
-                            "$ref": "#/definitions/blockatlas.Observer"
-                        }
-                    }
-                }
-            },
-            "delete": {
-                "description": "Delete a webhook for addresses transactions",
-                "consumes": [
-                    "application/json"
-                ],
-                "produces": [
-                    "application/json"
-                ],
-                "tags": [
-                    "Observer"
-                ],
-                "summary": "Delete a webhook",
-                "operationId": "delete_webhook",
-                "parameters": [
-                    {
-                        "description": "Accounts subscriptions",
-                        "name": "subscriptions",
-                        "in": "body",
-                        "required": true,
-                        "schema": {
-                            "$ref": "#/definitions/blockatlas.Webhook"
-                        }
-                    },
-                    {
-                        "type": "string",
-                        "default": "Bearer test",
-                        "description": "Bearer authorization header",
-                        "name": "Authorization",
-                        "in": "header",
-                        "required": true
-                    }
-                ],
-                "responses": {
-                    "200": {
-                        "description": "OK",
-                        "schema": {
-                            "$ref": "#/definitions/blockatlas.Observer"
-                        }
-                    }
-                }
-            }
-        },
         "/v1/market/charts": {
             "get": {
                 "description": "Get the charts data from an market and coin/token",
@@ -240,7 +122,7 @@ var doc = `{
                     "200": {
                         "description": "OK",
                         "schema": {
-                            "$ref": "#/definitions/blockatlas.ChartData"
+                            "$ref": "#/definitions/watchmarket.ChartData"
                         }
                     }
                 }
@@ -294,7 +176,7 @@ var doc = `{
                     "200": {
                         "description": "OK",
                         "schema": {
-                            "$ref": "#/definitions/blockatlas.ChartCoinInfo"
+                            "$ref": "#/definitions/watchmarket.ChartCoinInfo"
                         }
                     }
                 }
@@ -329,7 +211,7 @@ var doc = `{
                     "200": {
                         "description": "OK",
                         "schema": {
-                            "$ref": "#/definitions/blockatlas.Tickers"
+                            "$ref": "#/definitions/api.Tickers"
                         }
                     }
                 }
@@ -1123,113 +1005,68 @@ var doc = `{
                 }
             }
         },
-        "api.TickerRequest": {
+        "api.Ticker": {
             "type": "object",
             "properties": {
-                "assets": {
-                    "type": "array",
-                    "items": {
-                        "$ref": "#/definitions/api.Coin"
-                    }
+                "coin": {
+                    "type": "integer"
                 },
-                "currency": {
+                "coin_name": {
                     "type": "string"
-                }
-            }
-        },
-        "blockatlas.ChartCoinInfo": {
-            "type": "object",
-            "properties": {
-                "circulating_supply": {
-                    "type": "number"
                 },
-                "info": {
-                    "type": "object",
-                    "$ref": "#/definitions/blockatlas.CoinInfo"
+                "error": {
+                    "type": "string"
                 },
-                "market_cap": {
-                    "type": "number"
+                "last_update": {
+                    "type": "string"
                 },
-                "total_supply": {
-                    "type": "number"
+                "price": {
+                    "type": "object",
+                    "$ref": "#/definitions/api.TickerPrice"
                 },
-                "volume_24": {
-                    "type": "number"
-                }
-            }
-        },
-        "blockatlas.ChartData": {
-            "type": "object",
-            "properties": {
-                "error": {
+                "token_id": {
                     "type": "string"
                 },
-                "prices": {
-                    "type": "array",
-                    "items": {
-                        "$ref": "#/definitions/blockatlas.ChartPrice"
-                    }
+                "type": {
+                    "type": "string"
                 }
             }
         },
-        "blockatlas.ChartPrice": {
+        "api.TickerPrice": {
             "type": "object",
             "properties": {
-                "date": {
-                    "type": "integer"
+                "change_24h": {
+                    "type": "number"
                 },
-                "price": {
+                "currency": {
+                    "type": "string"
+                },
+                "provider": {
+                    "type": "string"
+                },
+                "value": {
                     "type": "number"
                 }
             }
         },
-        "blockatlas.CoinInfo": {
+        "api.TickerRequest": {
             "type": "object",
             "properties": {
-                "data_source": {
-                    "type": "string"
-                },
-                "description": {
-                    "type": "string"
-                },
-                "explorers": {
-                    "type": "array",
-                    "items": {
-                        "$ref": "#/definitions/blockatlas.Link"
-                    }
-                },
-                "name": {
-                    "type": "string"
-                },
-                "short_description": {
-                    "type": "string"
-                },
-                "socials": {
+                "assets": {
                     "type": "array",
                     "items": {
-                        "$ref": "#/definitions/blockatlas.SocialLink"
+                        "$ref": "#/definitions/api.Coin"
                     }
                 },
-                "source_code": {
-                    "type": "string"
-                },
-                "website": {
-                    "type": "string"
-                },
-                "white_paper": {
+                "currency": {
                     "type": "string"
                 }
             }
         },
-        "blockatlas.CoinStatus": {
-            "type": "object",
-            "properties": {
-                "error": {
-                    "type": "string"
-                },
-                "height": {
-                    "type": "integer"
-                }
+        "api.Tickers": {
+            "type": "array",
+            "items": {
+                "$ref": "#/definitions/api.Ticker"
             }
         },
         "blockatlas.Collection": {
@@ -1344,28 +1181,6 @@ var doc = `{
                 }
             }
         },
-        "blockatlas.Link": {
-            "type": "object",
-            "properties": {
-                "name": {
-                    "type": "string"
-                },
-                "url": {
-                    "type": "string"
-                }
-            }
-        },
-        "blockatlas.Observer": {
-            "type": "object",
-            "properties": {
-                "message": {
-                    "type": "string"
-                },
-                "status": {
-                    "type": "boolean"
-                }
-            }
-        },
         "blockatlas.Resolved": {
             "type": "object",
             "properties": {
@@ -1377,20 +1192,6 @@ var doc = `{
                 }
             }
         },
-        "blockatlas.SocialLink": {
-            "type": "object",
-            "properties": {
-                "handle": {
-                    "type": "string"
-                },
-                "name": {
-                    "type": "string"
-                },
-                "url": {
-                    "type": "string"
-                }
-            }
-        },
         "blockatlas.StakeValidator": {
             "type": "object",
             "properties": {
@@ -1453,65 +1254,6 @@ var doc = `{
                 }
             }
         },
-        "blockatlas.Subscriptions": {
-            "type": "object",
-            "additionalProperties": {
-                "type": "array",
-                "items": {
-                    "type": "string"
-                }
-            }
-        },
-        "blockatlas.Ticker": {
-            "type": "object",
-            "properties": {
-                "coin": {
-                    "type": "integer"
-                },
-                "coin_name": {
-                    "type": "string"
-                },
-                "error": {
-                    "type": "string"
-                },
-                "last_update": {
-                    "type": "string"
-                },
-                "price": {
-                    "type": "object",
-                    "$ref": "#/definitions/blockatlas.TickerPrice"
-                },
-                "token_id": {
-                    "type": "string"
-                },
-                "type": {
-                    "type": "string"
-                }
-            }
-        },
-        "blockatlas.TickerPrice": {
-            "type": "object",
-            "properties": {
-                "change_24h": {
-                    "type": "number"
-                },
-                "currency": {
-                    "type": "string"
-                },
-                "provider": {
-                    "type": "string"
-                },
-                "value": {
-                    "type": "number"
-                }
-            }
-        },
-        "blockatlas.Tickers": {
-            "type": "array",
-            "items": {
-                "$ref": "#/definitions/blockatlas.Ticker"
-            }
-        },
         "blockatlas.Tx": {
             "type": "object",
             "properties": {
@@ -1603,18 +1345,6 @@ var doc = `{
                 "$ref": "#/definitions/blockatlas.Tx"
             }
         },
-        "blockatlas.Webhook": {
-            "type": "object",
-            "properties": {
-                "subscriptions": {
-                    "type": "object",
-                    "$ref": "#/definitions/blockatlas.Subscriptions"
-                },
-                "webhook": {
-                    "type": "string"
-                }
-            }
-        },
         "coin.ExternalCoin": {
             "type": "object",
             "properties": {
@@ -1642,6 +1372,40 @@ var doc = `{
                     "type": "string"
                 }
             }
+        },
+        "watchmarket.ChartCoinInfo": {
+            "type": "object",
+            "properties": {
+                "circulating_supply": {
+                    "type": "number"
+                },
+                "info": {
+                    "type": "CoinInfo"
+                },
+                "market_cap": {
+                    "type": "number"
+                },
+                "total_supply": {
+                    "type": "number"
+                },
+                "volume_24": {
+                    "type": "number"
+                }
+            }
+        },
+        "watchmarket.ChartData": {
+            "type": "object",
+            "properties": {
+                "error": {
+                    "type": "string"
+                },
+                "prices": {
+                    "type": "array",
+                    "items": {
+                        "type": "ChartPrice"
+                    }
+                }
+            }
         }
     }
 }`
diff --git a/docs/swagger.json b/docs/swagger.json
index e348326..dc8104c 100644
--- a/docs/swagger.json
+++ b/docs/swagger.json
@@ -46,124 +46,6 @@
                 }
             }
         },
-        "/observer/v1/status": {
-            "get": {
-                "description": "Get coin status",
-                "consumes": [
-                    "application/json"
-                ],
-                "produces": [
-                    "application/json"
-                ],
-                "tags": [
-                    "Observer"
-                ],
-                "summary": "Get coin status",
-                "operationId": "coin_status",
-                "parameters": [
-                    {
-                        "type": "string",
-                        "default": "Bearer test",
-                        "description": "Bearer authorization header",
-                        "name": "Authorization",
-                        "in": "header",
-                        "required": true
-                    }
-                ],
-                "responses": {
-                    "200": {
-                        "description": "OK",
-                        "schema": {
-                            "$ref": "#/definitions/blockatlas.CoinStatus"
-                        }
-                    }
-                }
-            }
-        },
-        "/observer/v1/webhook/register": {
-            "post": {
-                "description": "Create a webhook for addresses transactions",
-                "consumes": [
-                    "application/json"
-                ],
-                "produces": [
-                    "application/json"
-                ],
-                "tags": [
-                    "Observer"
-                ],
-                "summary": "Create a webhook",
-                "operationId": "create_webhook",
-                "parameters": [
-                    {
-                        "description": "Accounts subscriptions",
-                        "name": "subscriptions",
-                        "in": "body",
-                        "required": true,
-                        "schema": {
-                            "$ref": "#/definitions/blockatlas.Webhook"
-                        }
-                    },
-                    {
-                        "type": "string",
-                        "default": "Bearer test",
-                        "description": "Bearer authorization header",
-                        "name": "Authorization",
-                        "in": "header",
-                        "required": true
-                    }
-                ],
-                "responses": {
-                    "200": {
-                        "description": "OK",
-                        "schema": {
-                            "$ref": "#/definitions/blockatlas.Observer"
-                        }
-                    }
-                }
-            },
-            "delete": {
-                "description": "Delete a webhook for addresses transactions",
-                "consumes": [
-                    "application/json"
-                ],
-                "produces": [
-                    "application/json"
-                ],
-                "tags": [
-                    "Observer"
-                ],
-                "summary": "Delete a webhook",
-                "operationId": "delete_webhook",
-                "parameters": [
-                    {
-                        "description": "Accounts subscriptions",
-                        "name": "subscriptions",
-                        "in": "body",
-                        "required": true,
-                        "schema": {
-                            "$ref": "#/definitions/blockatlas.Webhook"
-                        }
-                    },
-                    {
-                        "type": "string",
-                        "default": "Bearer test",
-                        "description": "Bearer authorization header",
-                        "name": "Authorization",
-                        "in": "header",
-                        "required": true
-                    }
-                ],
-                "responses": {
-                    "200": {
-                        "description": "OK",
-                        "schema": {
-                            "$ref": "#/definitions/blockatlas.Observer"
-                        }
-                    }
-                }
-            }
-        },
         "/v1/market/charts": {
             "get": {
                 "description": "Get the charts data from an market and coin/token",
@@ -219,7 +101,7 @@
                     "200": {
                         "description": "OK",
                         "schema": {
-                            "$ref": "#/definitions/blockatlas.ChartData"
+                            "$ref": "#/definitions/watchmarket.ChartData"
                         }
                     }
                 }
@@ -273,7 +155,7 @@
                     "200": {
                         "description": "OK",
                         "schema": {
-                            "$ref": "#/definitions/blockatlas.ChartCoinInfo"
+                            "$ref": "#/definitions/watchmarket.ChartCoinInfo"
                         }
                     }
                 }
@@ -308,7 +190,7 @@
                     "200": {
                         "description": "OK",
                         "schema": {
-                            "$ref": "#/definitions/blockatlas.Tickers"
+                            "$ref": "#/definitions/api.Tickers"
                         }
                     }
                 }
@@ -1102,113 +984,68 @@
                 }
             }
         },
-        "api.TickerRequest": {
+        "api.Ticker": {
             "type": "object",
             "properties": {
-                "assets": {
-                    "type": "array",
-                    "items": {
-                        "$ref": "#/definitions/api.Coin"
-                    }
+                "coin": {
+                    "type": "integer"
                 },
-                "currency": {
+                "coin_name": {
                     "type": "string"
-                }
-            }
-        },
-        "blockatlas.ChartCoinInfo": {
-            "type": "object",
-            "properties": {
-                "circulating_supply": {
-                    "type": "number"
                 },
-                "info": {
-                    "type": "object",
-                    "$ref": "#/definitions/blockatlas.CoinInfo"
+                "error": {
+                    "type": "string"
                 },
-                "market_cap": {
-                    "type": "number"
+                "last_update": {
+                    "type": "string"
                 },
-                "total_supply": {
-                    "type": "number"
+                "price": {
+                    "type": "object",
+                    "$ref": "#/definitions/api.TickerPrice"
                 },
-                "volume_24": {
-                    "type": "number"
-                }
-            }
-        },
-        "blockatlas.ChartData": {
-            "type": "object",
-            "properties": {
-                "error": {
+                "token_id": {
                     "type": "string"
                 },
-                "prices": {
-                    "type": "array",
-                    "items": {
-                        "$ref": "#/definitions/blockatlas.ChartPrice"
-                    }
+                "type": {
+                    "type": "string"
                 }
             }
         },
-        "blockatlas.ChartPrice": {
+        "api.TickerPrice": {
             "type": "object",
             "properties": {
-                "date": {
-                    "type": "integer"
+                "change_24h": {
+                    "type": "number"
                 },
-                "price": {
+                "currency": {
+                    "type": "string"
+                },
+                "provider": {
+                    "type": "string"
+                },
+                "value": {
                     "type": "number"
                 }
             }
         },
-        "blockatlas.CoinInfo": {
+        "api.TickerRequest": {
             "type": "object",
             "properties": {
-                "data_source": {
-                    "type": "string"
-                },
-                "description": {
-                    "type": "string"
-                },
-                "explorers": {
-                    "type": "array",
-                    "items": {
-                        "$ref": "#/definitions/blockatlas.Link"
-                    }
-                },
-                "name": {
-                    "type": "string"
-                },
-                "short_description": {
-                    "type": "string"
-                },
-                "socials": {
+                "assets": {
                     "type": "array",
                     "items": {
-                        "$ref": "#/definitions/blockatlas.SocialLink"
+                        "$ref": "#/definitions/api.Coin"
                     }
                 },
-                "source_code": {
-                    "type": "string"
-                },
-                "website": {
-                    "type": "string"
-                },
-                "white_paper": {
+                "currency": {
                     "type": "string"
                 }
             }
         },
-        "blockatlas.CoinStatus": {
-            "type": "object",
-            "properties": {
-                "error": {
-                    "type": "string"
-                },
-                "height": {
-                    "type": "integer"
-                }
+        "api.Tickers": {
+            "type": "array",
+            "items": {
+                "$ref": "#/definitions/api.Ticker"
             }
         },
         "blockatlas.Collection": {
@@ -1323,28 +1160,6 @@
                 }
             }
         },
-        "blockatlas.Link": {
-            "type": "object",
-            "properties": {
-                "name": {
-                    "type": "string"
-                },
-                "url": {
-                    "type": "string"
-                }
-            }
-        },
-        "blockatlas.Observer": {
-            "type": "object",
-            "properties": {
-                "message": {
-                    "type": "string"
-                },
-                "status": {
-                    "type": "boolean"
-                }
-            }
-        },
         "blockatlas.Resolved": {
             "type": "object",
             "properties": {
@@ -1356,20 +1171,6 @@
                 }
             }
         },
-        "blockatlas.SocialLink": {
-            "type": "object",
-            "properties": {
-                "handle": {
-                    "type": "string"
-                },
-                "name": {
-                    "type": "string"
-                },
-                "url": {
-                    "type": "string"
-                }
-            }
-        },
         "blockatlas.StakeValidator": {
             "type": "object",
             "properties": {
@@ -1432,65 +1233,6 @@
                 }
             }
         },
-        "blockatlas.Subscriptions": {
-            "type": "object",
-            "additionalProperties": {
-                "type": "array",
-                "items": {
-                    "type": "string"
-                }
-            }
-        },
-        "blockatlas.Ticker": {
-            "type": "object",
-            "properties": {
-                "coin": {
-                    "type": "integer"
-                },
-                "coin_name": {
-                    "type": "string"
-                },
-                "error": {
-                    "type": "string"
-                },
-                "last_update": {
-                    "type": "string"
-                },
-                "price": {
-                    "type": "object",
-                    "$ref": "#/definitions/blockatlas.TickerPrice"
-                },
-                "token_id": {
-                    "type": "string"
-                },
-                "type": {
-                    "type": "string"
-                }
-            }
-        },
-        "blockatlas.TickerPrice": {
-            "type": "object",
-            "properties": {
-                "change_24h": {
-                    "type": "number"
-                },
-                "currency": {
-                    "type": "string"
-                },
-                "provider": {
-                    "type": "string"
-                },
-                "value": {
-                    "type": "number"
-                }
-            }
-        },
-        "blockatlas.Tickers": {
-            "type": "array",
-            "items": {
-                "$ref": "#/definitions/blockatlas.Ticker"
-            }
-        },
         "blockatlas.Tx": {
             "type": "object",
             "properties": {
@@ -1582,18 +1324,6 @@
                 "$ref": "#/definitions/blockatlas.Tx"
             }
         },
-        "blockatlas.Webhook": {
-            "type": "object",
-            "properties": {
-                "subscriptions": {
-                    "type": "object",
-                    "$ref": "#/definitions/blockatlas.Subscriptions"
-                },
-                "webhook": {
-                    "type": "string"
-                }
-            }
-        },
         "coin.ExternalCoin": {
             "type": "object",
             "properties": {
@@ -1621,6 +1351,40 @@
                     "type": "string"
                 }
             }
+        },
+        "watchmarket.ChartCoinInfo": {
+            "type": "object",
+            "properties": {
+                "circulating_supply": {
+                    "type": "number"
+                },
+                "info": {
+                    "type": "CoinInfo"
+                },
+                "market_cap": {
+                    "type": "number"
+                },
+                "total_supply": {
+                    "type": "number"
+                },
+                "volume_24": {
+                    "type": "number"
+                }
+            }
+        },
+        "watchmarket.ChartData": {
+            "type": "object",
+            "properties": {
+                "error": {
+                    "type": "string"
+                },
+                "prices": {
+                    "type": "array",
+                    "items": {
+                        "type": "ChartPrice"
+                    }
+                }
+            }
         }
     }
 }
\ No newline at end of file
diff --git a/docs/swagger.yaml b/docs/swagger.yaml
index 428a6fa..4ddb42c 100644
--- a/docs/swagger.yaml
+++ b/docs/swagger.yaml
@@ -19,77 +19,48 @@ definitions:
       type:
         type: string
     type: object
-  api.TickerRequest:
-    properties:
-      assets:
-        items:
-          $ref: '#/definitions/api.Coin'
-        type: array
-      currency:
-        type: string
-    type: object
-  blockatlas.ChartCoinInfo:
+  api.Ticker:
     properties:
-      circulating_supply:
-        type: number
-      info:
-        $ref: '#/definitions/blockatlas.CoinInfo'
-        type: object
-      market_cap:
-        type: number
-      total_supply:
-        type: number
-      volume_24:
-        type: number
-    type: object
-  blockatlas.ChartData:
-    properties:
-      error:
-        type: string
-      prices:
-        items:
-          $ref: '#/definitions/blockatlas.ChartPrice'
-        type: array
-    type: object
-  blockatlas.ChartPrice:
-    properties:
-      date:
+      coin:
         type: integer
-      price:
-        type: number
-    type: object
-  blockatlas.CoinInfo:
-    properties:
-      data_source:
+      coin_name:
         type: string
-      description:
+      error:
         type: string
-      explorers:
-        items:
-          $ref: '#/definitions/blockatlas.Link'
-        type: array
-      name:
+      last_update:
         type: string
-      short_description:
+      price:
+        $ref: '#/definitions/api.TickerPrice'
+        type: object
+      token_id:
         type: string
-      socials:
-        items:
-          $ref: '#/definitions/blockatlas.SocialLink'
-        type: array
-      source_code:
+      type:
         type: string
-      website:
+    type: object
+  api.TickerPrice:
+    properties:
+      change_24h:
+        type: number
+      currency:
         type: string
-      white_paper:
+      provider:
         type: string
+      value:
+        type: number
     type: object
-  blockatlas.CoinStatus:
+  api.TickerRequest:
     properties:
-      error:
+      assets:
+        items:
+          $ref: '#/definitions/api.Coin'
+        type: array
+      currency:
         type: string
-      height:
-        type: integer
     type: object
+  api.Tickers:
+    items:
+      $ref: '#/definitions/api.Ticker'
+    type: array
   blockatlas.Collection:
     properties:
       address:
@@ -165,20 +136,6 @@ definitions:
       docs:
         type: object
     type: object
-  blockatlas.Link:
-    properties:
-      name:
-        type: string
-      url:
-        type: string
-    type: object
-  blockatlas.Observer:
-    properties:
-      message:
-        type: string
-      status:
-        type: boolean
-    type: object
   blockatlas.Resolved:
     properties:
       coin:
@@ -186,15 +143,6 @@ definitions:
       result:
         type: string
     type: object
-  blockatlas.SocialLink:
-    properties:
-      handle:
-        type: string
-      name:
-        type: string
-      url:
-        type: string
-    type: object
   blockatlas.StakeValidator:
     properties:
       details:
@@ -236,45 +184,6 @@ definitions:
       annual:
         type: number
     type: object
-  blockatlas.Subscriptions:
-    additionalProperties:
-      items:
-        type: string
-      type: array
-    type: object
-  blockatlas.Ticker:
-    properties:
-      coin:
-        type: integer
-      coin_name:
-        type: string
-      error:
-        type: string
-      last_update:
-        type: string
-      price:
-        $ref: '#/definitions/blockatlas.TickerPrice'
-        type: object
-      token_id:
-        type: string
-      type:
-        type: string
-    type: object
-  blockatlas.TickerPrice:
-    properties:
-      change_24h:
-        type: number
-      currency:
-        type: string
-      provider:
-        type: string
-      value:
-        type: number
-    type: object
-  blockatlas.Tickers:
-    items:
-      $ref: '#/definitions/blockatlas.Ticker'
-    type: array
   blockatlas.Tx:
     properties:
       block:
@@ -342,14 +251,6 @@ definitions:
     items:
       $ref: '#/definitions/blockatlas.Tx'
     type: array
-  blockatlas.Webhook:
-    properties:
-      subscriptions:
-        $ref: '#/definitions/blockatlas.Subscriptions'
-        type: object
-      webhook:
-        type: string
-    type: object
   coin.ExternalCoin:
     properties:
       coin:
@@ -368,6 +269,28 @@ definitions:
       status_message:
         type: string
     type: object
+  watchmarket.ChartCoinInfo:
+    properties:
+      circulating_supply:
+        type: number
+      info:
+        type: CoinInfo
+      market_cap:
+        type: number
+      total_supply:
+        type: number
+      volume_24:
+        type: number
+    type: object
+  watchmarket.ChartData:
+    properties:
+      error:
+        type: string
+      prices:
+        items:
+          type: ChartPrice
+        type: array
+    type: object
 info:
   contact: {}
   license: {}
@@ -399,86 +322,6 @@ paths:
       summary: Lookup .eth / .zil addresses
       tags:
       - Naming
-  /observer/v1/status:
-    get:
-      consumes:
-      - application/json
-      description: Get coin status
-      operationId: coin_status
-      parameters:
-      - default: Bearer test
-        description: Bearer authorization header
-        in: header
-        name: Authorization
-        required: true
-        type: string
-      produces:
-      - application/json
-      responses:
-        "200":
-          description: OK
-          schema:
-            $ref: '#/definitions/blockatlas.CoinStatus'
-      summary: Get coin status
-      tags:
-      - Observer
-  /observer/v1/webhook/register:
-    delete:
-      consumes:
-      - application/json
-      description: Delete a webhook for addresses transactions
-      operationId: delete_webhook
-      parameters:
-      - description: Accounts subscriptions
-        in: body
-        name: subscriptions
-        required: true
-        schema:
-          $ref: '#/definitions/blockatlas.Webhook'
-      - default: Bearer test
-        description: Bearer authorization header
-        in: header
-        name: Authorization
-        required: true
-        type: string
-      produces:
-      - application/json
-      responses:
-        "200":
-          description: OK
-          schema:
-            $ref: '#/definitions/blockatlas.Observer'
-      summary: Delete a webhook
-      tags:
-      - Observer
-    post:
-      consumes:
-      - application/json
-      description: Create a webhook for addresses transactions
-      operationId: create_webhook
-      parameters:
-      - description: Accounts subscriptions
-        in: body
-        name: subscriptions
-        required: true
-        schema:
-          $ref: '#/definitions/blockatlas.Webhook'
-      - default: Bearer test
-        description: Bearer authorization header
-        in: header
-        name: Authorization
-        required: true
-        type: string
-      produces:
-      - application/json
-      responses:
-        "200":
-          description: OK
-          schema:
-            $ref: '#/definitions/blockatlas.Observer'
-      summary: Create a webhook
-      tags:
-      - Observer
   /v1/{coin}/{address}:
     get:
       consumes:
@@ -575,7 +418,7 @@ paths:
         "200":
           description: OK
           schema:
-            $ref: '#/definitions/blockatlas.ChartData'
+            $ref: '#/definitions/watchmarket.ChartData'
       summary: Get charts data for a specific coin
       tags:
       - Market
@@ -612,7 +455,7 @@ paths:
         "200":
           description: OK
           schema:
-            $ref: '#/definitions/blockatlas.ChartCoinInfo'
+            $ref: '#/definitions/watchmarket.ChartCoinInfo'
       summary: Get charts coin info data for a specific coin
       tags:
       - Market
@@ -635,7 +478,7 @@ paths:
         "200":
           description: OK
           schema:
-            $ref: '#/definitions/blockatlas.Tickers'
+            $ref: '#/definitions/api.Tickers'
       summary: Get ticker values for a specific market
       tags:
       - Market
diff --git a/mq/mq.go b/mq/mq.go
index 3538222..3ee3ab8 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,44 @@ 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
+	}
+
+	err = amqpChan.Qos(
+		5,
+		0,
+		true,
+	)
+	if err != nil {
+		logger.Error("no qos limit ", err)
+	}
+
+	for data := range messageChannel {
+		go consumer(data, cache)
+	}
+}
diff --git a/nginx.conf b/nginx.conf
index 77b5be9..8fd11f9 100644
--- a/nginx.conf
+++ b/nginx.conf
@@ -5,11 +5,7 @@ server {
         proxy_pass http://swagger_api:8423;
     }
 
-    location ~* /observer.* {
-        proxy_pass http://observer_api:8422;
-    }
-
     location / {
         proxy_pass http://platform_api:8420;
     }
-}
\ No newline at end of file
+}
diff --git a/observer/dispatcher.go b/observer/dispatcher.go
index 397073b..1e11e67 100644
--- a/observer/dispatcher.go
+++ b/observer/dispatcher.go
@@ -14,9 +14,9 @@ type Dispatcher struct {
 }
 
 type DispatchEvent struct {
-	Action  blockatlas.TransactionType `json:"action"`
-	Result  *blockatlas.Tx             `json:"result"`
-	Webhook string                     `json:"webhook"`
+	Action blockatlas.TransactionType `json:"action"`
+	Result *blockatlas.Tx             `json:"result"`
+	GUID   string                     `json:"guid"`
 }
 
 func (d *Dispatcher) Run(events <-chan Event) {
@@ -26,12 +26,12 @@ func (d *Dispatcher) Run(events <-chan Event) {
 }
 
 func (d *Dispatcher) dispatch(event Event) {
-	webhook := event.Subscription.Webhook
+	guid := event.Subscription.GUID
 
 	action := DispatchEvent{
-		Action:  event.Tx.Type,
-		Result:  event.Tx,
-		Webhook: webhook,
+		Action: event.Tx.Type,
+		Result: event.Tx,
+		GUID:   guid,
 	}
 
 	txJson, err := json.Marshal(action)
@@ -40,18 +40,18 @@ func (d *Dispatcher) dispatch(event Event) {
 	}
 
 	logParams := logger.Params{
-		"webhook": webhook,
-		"coin":    event.Subscription.Coin,
-		"txID":    event.Tx.ID,
+		"guid": guid,
+		"coin": event.Subscription.Coin,
+		"txID": event.Tx.ID,
 	}
 
-	go d.postMessageToQueue(webhook, txJson, logParams)
+	go d.postMessageToQueue(guid, txJson, logParams)
 
 	logger.Info("Dispatching messages...", logParams)
 }
 
 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/block.go b/pkg/blockatlas/block.go
index e11e8e2..14e54f0 100644
--- a/pkg/blockatlas/block.go
+++ b/pkg/blockatlas/block.go
@@ -9,5 +9,5 @@ type Block struct {
 type Subscription struct {
 	Coin    uint   `json:"coin"`
 	Address string `json:"address"`
-	Webhook string `json:"webhook"`
+	GUID    string `json:"guid"`
 }
diff --git a/pkg/blockatlas/observer.go b/pkg/blockatlas/observer.go
index f46cc11..7b7e79e 100644
--- a/pkg/blockatlas/observer.go
+++ b/pkg/blockatlas/observer.go
@@ -4,9 +4,13 @@ import "strconv"
 
 type Subscriptions map[string][]string
 
-type Webhook struct {
-	Subscriptions Subscriptions `json:"subscriptions"`
-	Webhook       string        `json:"webhook"`
+type SubscriptionOperation string
+
+type SubscriptionEvent struct {
+	NewSubscriptions Subscriptions         `json:"new_subscriptions"`
+	OldSubscriptions Subscriptions         `json:"old_subscriptions"`
+	GUID             string                `json:"guid"`
+	Operation        SubscriptionOperation `json:"operation"`
 }
 
 type CoinStatus struct {
@@ -19,9 +23,9 @@ type Observer struct {
 	Message string `json:"message"`
 }
 
-func (w *Webhook) ParseSubscriptions() []Subscription {
+func (e *SubscriptionEvent) ParseSubscriptions(s Subscriptions) []Subscription {
 	subs := make([]Subscription, 0)
-	for coinStr, perCoin := range w.Subscriptions {
+	for coinStr, perCoin := range s {
 		coin, err := strconv.Atoi(coinStr)
 		if err != nil {
 			continue
@@ -30,7 +34,7 @@ func (w *Webhook) ParseSubscriptions() []Subscription {
 			subs = append(subs, Subscription{
 				Coin:    uint(coin),
 				Address: addr,
-				Webhook: w.Webhook,
+				GUID:    e.GUID,
 			})
 		}
 	}
diff --git a/pkg/blockatlas/tx.go b/pkg/blockatlas/tx.go
index 6e40b64..7ae77f4 100644
--- a/pkg/blockatlas/tx.go
+++ b/pkg/blockatlas/tx.go
@@ -11,7 +11,7 @@ type KeyTitle string
 const (
 	StatusCompleted Status = "completed"
 	StatusPending   Status = "pending"
-	StatusFailed    Status = "failed"
+	StatusError     Status = "error"
 
 	DirectionOutgoing Direction = "outgoing"
 	DirectionIncoming Direction = "incoming"
diff --git a/platform/cosmos/client.go b/platform/cosmos/client.go
index eee6f4b..2378061 100644
--- a/platform/cosmos/client.go
+++ b/platform/cosmos/client.go
@@ -15,10 +15,10 @@ type Client struct {
 }
 
 // GetAddrTxs - get all ATOM transactions for a given address
-func (c *Client) GetAddrTxs(address string, tag string) (txs TxPage, err error) {
+func (c *Client) GetAddrTxs(address, tag string, page int) (txs TxPage, err error) {
 	query := url.Values{
 		tag:     {address},
-		"page":  {"1"},
+		"page":  {strconv.Itoa(page)},
 		"limit": {"25"},
 	}
 	err = c.Get(&txs, "txs", query)
diff --git a/platform/cosmos/model.go b/platform/cosmos/model.go
index 09d7c2e..c9b1e6e 100644
--- a/platform/cosmos/model.go
+++ b/platform/cosmos/model.go
@@ -48,7 +48,8 @@ type Tx struct {
 }
 
 type TxPage struct {
-	Txs []Tx `json:"txs"`
+	PageTotal string `json:"page_total"`
+	Txs       []Tx   `json:"txs"`
 }
 
 // Events
diff --git a/platform/cosmos/transaction.go b/platform/cosmos/transaction.go
index 04f7521..5a08c1f 100644
--- a/platform/cosmos/transaction.go
+++ b/platform/cosmos/transaction.go
@@ -12,24 +12,43 @@ import (
 func (p *Platform) GetTxsByAddress(address string) (blockatlas.TxPage, error) {
 	tagsList := []string{"transfer.recipient", "message.sender"}
 	var wg sync.WaitGroup
-	out := make(chan TxPage, len(tagsList))
+	out := make(chan []Tx, len(tagsList))
 	wg.Add(len(tagsList))
 	for _, t := range tagsList {
 		go func(tag, addr string, wg *sync.WaitGroup) {
 			defer wg.Done()
-			txs, err := p.client.GetAddrTxs(addr, tag)
+			page := 1
+			txs, err := p.client.GetAddrTxs(addr, tag, page)
 			if err != nil {
 				logger.Error("GetAddrTxs", err, logger.Params{"address": tag, "tag": tag})
 				return
 			}
-			out <- txs
+			// Condition when no more pages to paginate
+			if txs.PageTotal == "1" {
+				out <- txs.Txs
+				return
+			}
+
+			totalPages, err := strconv.Atoi(txs.PageTotal)
+			if err != nil {
+				logger.Error("GetAddrTxs", err, logger.Params{"totalPages": totalPages})
+				return
+			}
+			// gaia does support sort option, paginate to get latest transactions by passing total pages page
+			// https://github.com/cosmos/gaia/blob/f61b391aee5d04364d2b5539692bbb187ad9b946/docs/resources/gaiacli.md#query-transactions
+			txs2, err := p.client.GetAddrTxs(addr, tag, totalPages)
+			if err != nil {
+				logger.Error("GetAddrTxs", err, logger.Params{"address": tag, "tag": tag})
+				return
+			}
+			out <- txs2.Txs
 		}(t, address, &wg)
 	}
 	wg.Wait()
 	close(out)
 	srcTxs := make([]Tx, 0)
 	for r := range out {
-		srcTxs = append(srcTxs, r.Txs...)
+		srcTxs = append(srcTxs, r...)
 	}
 	return p.NormalizeTxs(srcTxs), nil
 }
@@ -77,7 +96,7 @@ func (p *Platform) Normalize(srcTx *Tx) (tx blockatlas.Tx, ok bool) {
 	status := blockatlas.StatusCompleted
 	// https://github.com/cosmos/cosmos-sdk/blob/95ddc242ad024ca78a359a13122dade6f14fd676/types/errors/errors.go#L19
 	if srcTx.Code > 0 {
-		status = blockatlas.StatusFailed
+		status = blockatlas.StatusError
 	}
 
 	tx = blockatlas.Tx{
diff --git a/platform/ethereum/model.go b/platform/ethereum/model.go
index e4f2e33..d8464d0 100644
--- a/platform/ethereum/model.go
+++ b/platform/ethereum/model.go
@@ -11,13 +11,8 @@ type Page struct {
 }
 
 type TokenPage struct {
-	Total uint    `json:"total"`
-	Docs  []Token `json:"docs"`
-}
-
-type Token struct {
-	Balance  string   `json:"balance"`
-	Contract Contract `json:"contract"`
+	Total uint       `json:"total"`
+	Docs  []Contract `json:"docs"`
 }
 
 type Doc struct {
@@ -54,7 +49,6 @@ type Contract struct {
 	Decimals    uint   `json:"decimals"`
 	TotalSupply string `json:"totalSupply,omitempty"`
 	Name        string `json:"name"`
-	Contract    string `json:"contract,omitempty"`
 }
 
 type NodeInfo struct {
diff --git a/platform/ethereum/token.go b/platform/ethereum/token.go
index cce551a..78b15f8 100644
--- a/platform/ethereum/token.go
+++ b/platform/ethereum/token.go
@@ -13,13 +13,13 @@ func (p *Platform) GetTokenListByAddress(address string) (blockatlas.TokenPage,
 }
 
 // NormalizeToken converts a Ethereum token into the generic model
-func NormalizeToken(srcToken *Token, coinIndex uint) (t blockatlas.Token, ok bool) {
+func NormalizeToken(srcToken *Contract, coinIndex uint) (t blockatlas.Token, ok bool) {
 	t = blockatlas.Token{
-		Name:     srcToken.Contract.Name,
-		Symbol:   srcToken.Contract.Symbol,
-		TokenID:  srcToken.Contract.Contract,
+		Name:     srcToken.Name,
+		Symbol:   srcToken.Symbol,
+		TokenID:  srcToken.Address,
 		Coin:     coinIndex,
-		Decimals: srcToken.Contract.Decimals,
+		Decimals: srcToken.Decimals,
 		Type:     blockatlas.TokenTypeERC20,
 	}
 
@@ -27,7 +27,7 @@ func NormalizeToken(srcToken *Token, coinIndex uint) (t blockatlas.Token, ok boo
 }
 
 // NormalizeTxs converts multiple Ethereum tokens
-func NormalizeTokens(srcTokens []Token, p Platform) []blockatlas.Token {
+func NormalizeTokens(srcTokens []Contract, p Platform) []blockatlas.Token {
 	tokenPage := make([]blockatlas.Token, 0)
 	for _, srcToken := range srcTokens {
 		token, ok := NormalizeToken(&srcToken, p.CoinIndex)
diff --git a/platform/ethereum/transaction.go b/platform/ethereum/transaction.go
index 882fa83..e3dfbd0 100644
--- a/platform/ethereum/transaction.go
+++ b/platform/ethereum/transaction.go
@@ -49,7 +49,7 @@ func extractBase(srcTx *Doc, coinIndex uint) (base blockatlas.Tx, ok bool) {
 	if srcTx.Error == "" {
 		status = blockatlas.StatusCompleted
 	} else {
-		status = blockatlas.StatusFailed
+		status = blockatlas.StatusError
 		errReason = srcTx.Error
 	}
 
diff --git a/platform/nebulas/transaction.go b/platform/nebulas/transaction.go
index d31df4d..c22ce5a 100644
--- a/platform/nebulas/transaction.go
+++ b/platform/nebulas/transaction.go
@@ -41,7 +41,7 @@ func NormalizeTxs(txs []Transaction) []blockatlas.Tx {
 func NormalizeTx(srcTx Transaction) blockatlas.Tx {
 	var status = blockatlas.StatusCompleted
 	if srcTx.Status == 0 {
-		status = blockatlas.StatusFailed
+		status = blockatlas.StatusError
 	}
 	return blockatlas.Tx{
 		ID:       srcTx.Hash,
diff --git a/platform/ontology/transaction.go b/platform/ontology/transaction.go
index c2fb048..5721640 100644
--- a/platform/ontology/transaction.go
+++ b/platform/ontology/transaction.go
@@ -34,7 +34,7 @@ func Normalize(srcTx *Tx, assetName AssetType) (tx blockatlas.Tx, ok bool) {
 	fee := numbers.DecimalExp(srcTx.Fee, ONGDecimals)
 	status := blockatlas.StatusCompleted
 	if srcTx.ConfirmFlag != 1 {
-		status = blockatlas.StatusFailed
+		status = blockatlas.StatusError
 	}
 	tx = blockatlas.Tx{
 		ID:     srcTx.Hash,
diff --git a/platform/polkadot/transaction.go b/platform/polkadot/transaction.go
index 1685c71..5a1387c 100644
--- a/platform/polkadot/transaction.go
+++ b/platform/polkadot/transaction.go
@@ -35,7 +35,7 @@ func (p *Platform) NormalizeTransfer(srcTx *Transfer) blockatlas.Tx {
 	amount := strings.Split(numbers.DecimalExp(srcTx.Amount, int(decimals)), ".")[0]
 	status := blockatlas.StatusCompleted
 	if !srcTx.Success {
-		status = blockatlas.StatusFailed
+		status = blockatlas.StatusError
 	}
 	result := blockatlas.Tx{
 		ID:     srcTx.Hash,
@@ -75,7 +75,7 @@ func (p *Platform) NormalizeExtrinsic(srcTx *Extrinsic) *blockatlas.Tx {
 
 	var status blockatlas.Status
 	if !srcTx.Success {
-		status = blockatlas.StatusFailed
+		status = blockatlas.StatusError
 	} else {
 		status = blockatlas.StatusCompleted
 	}
diff --git a/platform/ripple/client.go b/platform/ripple/client.go
index b2d0800..6900f4a 100644
--- a/platform/ripple/client.go
+++ b/platform/ripple/client.go
@@ -11,19 +11,35 @@ type Client struct {
 }
 
 func (c *Client) GetTxsOfAddress(address string) ([]Tx, error) {
+	res, err := c.fetchTransactions(address, "true")
+	if err != nil {
+		return nil, err
+	}
+
+	if res.Result == "error" {
+		res, err = c.fetchTransactions(address, "false")
+		if err != nil {
+			return nil, err
+		}
+	}
+
+	return res.Transactions, nil
+}
+
+func (c *Client) fetchTransactions(address, descending string) (Response, error) {
 	query := url.Values{
 		"type":       {"Payment"},
-		"descending": {"false"},
-		"limit":      {"200"},
+		"descending": {descending},
+		"limit":      {"25"},
 	}
 	uri := fmt.Sprintf("accounts/%s/transactions", url.PathEscape(address))
 
 	var res Response
 	err := c.Get(&res, uri, query)
 	if err != nil {
-		return nil, err
+		return Response{}, err
 	}
-	return res.Transactions, nil
+	return res, nil
 }
 
 func (c *Client) GetCurrentBlock() (int64, error) {
diff --git a/platform/ripple/transaction.go b/platform/ripple/transaction.go
index 84f0fd8..68e054a 100644
--- a/platform/ripple/transaction.go
+++ b/platform/ripple/transaction.go
@@ -55,7 +55,7 @@ func NormalizeTx(srcTx *Tx) (blockatlas.Tx, bool) {
 
 	status := blockatlas.StatusCompleted
 	if srcTx.Meta.TransactionResult != transactionResultSuccess {
-		status = blockatlas.StatusFailed
+		status = blockatlas.StatusError
 	}
 
 	result := blockatlas.Tx{
diff --git a/platform/tezos/transaction.go b/platform/tezos/transaction.go
index 354c795..d2615e4 100644
--- a/platform/tezos/transaction.go
+++ b/platform/tezos/transaction.go
@@ -49,7 +49,7 @@ func NormalizeTx(srcTx Transaction) (blockatlas.Tx, bool) {
 	status := blockatlas.StatusCompleted
 	if srcTx.Status() != TxStatusApplied {
 		errMsg = "transaction failed"
-		status = blockatlas.StatusFailed
+		status = blockatlas.StatusError
 	}
 	tx := blockatlas.Tx{
 		ID:     srcTx.Op.OpHash,
diff --git a/platform/zilliqa/transaction.go b/platform/zilliqa/transaction.go
index cb52d1e..c18bda5 100644
--- a/platform/zilliqa/transaction.go
+++ b/platform/zilliqa/transaction.go
@@ -41,7 +41,7 @@ func Normalize(srcTx *Tx) (tx blockatlas.Tx) {
 		},
 	}
 	if !srcTx.ReceiptSuccess {
-		tx.Status = blockatlas.StatusFailed
+		tx.Status = blockatlas.StatusError
 	}
 	return tx
 }
diff --git a/services/domains/domains.go b/services/domains/domains.go
index 05adb51..e814230 100644
--- a/services/domains/domains.go
+++ b/services/domains/domains.go
@@ -1,12 +1,13 @@
 package domains
 
 import (
+	"math"
+	"strings"
+
 	CoinType "github.com/trustwallet/blockatlas/coin"
 	"github.com/trustwallet/blockatlas/pkg/blockatlas"
 	"github.com/trustwallet/blockatlas/pkg/errors"
 	"github.com/trustwallet/blockatlas/platform"
-	"math"
-	"strings"
 )
 
 // TLDMapping Mapping of name TLD's to coin where they are handled
@@ -14,6 +15,7 @@ var TLDMapping = map[string]uint64{
 	".eth":        CoinType.ETH,
 	".xyz":        CoinType.ETH,
 	".luxe":       CoinType.ETH,
+	".kred":       CoinType.ETH,
 	".zil":        CoinType.ZIL,
 	".crypto":     CoinType.ZIL,
 	"@fiotestnet": CoinType.FIO,
diff --git a/services/subscription/subscribtion.go b/services/subscription/subscribtion.go
new file mode 100644
index 0000000..a381fd5
--- /dev/null
+++ b/services/subscription/subscribtion.go
@@ -0,0 +1,64 @@
+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"
+	UpdateSubscription blockatlas.SubscriptionOperation = "UpdateSubscription"
+)
+
+func Consume(delivery amqp.Delivery, storage storage.Addresses) {
+	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)}
+
+	switch event.Operation {
+	case UpdateSubscription:
+		err := storage.DeleteSubscriptions(oldSubscriptions)
+		if err != nil {
+			logger.Error(err, params)
+		}
+		err = storage.AddSubscriptions(newSubscriptions)
+		if err != nil {
+			logger.Error(err, params)
+		}
+		err = delivery.Ack(false)
+		if err != nil {
+			logger.Error(err, params)
+		}
+		logger.Info("Updated", params)
+	case AddSubscription:
+		err = storage.AddSubscriptions(newSubscriptions)
+		if err != nil {
+			logger.Error(err, params)
+		}
+		err = delivery.Ack(false)
+		if err != nil {
+			logger.Error(err, params)
+		}
+		logger.Info("Added", params)
+	case DeleteSubscription:
+		err := storage.DeleteSubscriptions(oldSubscriptions)
+		if err != nil {
+			logger.Error(err, params)
+		}
+		err = delivery.Ack(false)
+		if err != nil {
+			logger.Error(err, params)
+		}
+		logger.Info("Deleted", params)
+	}
+}
diff --git a/storage/addresses.go b/storage/addresses.go
index 760bebc..97e8cb2 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 (
@@ -18,51 +17,52 @@ func (s *Storage) Lookup(coin uint, addresses []string) ([]blockatlas.Subscripti
 	observers := make([]blockatlas.Subscription, 0)
 	for _, address := range addresses {
 		key := getSubscriptionKey(coin, address)
-		var webhooks []string
-		err := s.GetHMValue(ATLAS_OBSERVER, key, &webhooks)
+		var guids []string
+		err := s.GetHMValue(ATLAS_OBSERVER, key, &guids)
 		if err != nil {
 			continue
 		}
-		for _, webhook := range webhooks {
-			observers = append(observers, blockatlas.Subscription{Coin: coin, Address: address, Webhook: webhook})
+		for _, guid := range guids {
+			observers = append(observers, blockatlas.Subscription{Coin: coin, Address: address, GUID: guid})
 		}
 	}
 	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)
-		if webhooks == nil {
-			webhooks = make([]string, 0)
+		var guids []string
+		s.GetHMValue(ATLAS_OBSERVER, key, &guids)
+		if guids == nil {
+			guids = make([]string, 0)
 		}
-		if hasObject(webhooks, sub.Webhook) {
+		if hasObject(guids, sub.GUID) {
 			continue
 		}
-		webhooks = append(webhooks, sub.Webhook)
-		err := s.AddHM(ATLAS_OBSERVER, key, webhooks)
+		guids = append(guids, sub.GUID)
+		err := s.AddHM(ATLAS_OBSERVER, key, guids)
 		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
-		err := s.GetHMValue(ATLAS_OBSERVER, key, &webhooks)
+		var guids []string
+		err := s.GetHMValue(ATLAS_OBSERVER, key, &guids)
 		if err != nil {
 			continue
 		}
 		newHooks := make([]string, 0)
-		for _, webhook := range webhooks {
-			if webhook == sub.Webhook {
+		for _, guid := range guids {
+			if guid == sub.GUID {
 				continue
 			}
-			newHooks = append(newHooks, webhook)
+			newHooks = append(newHooks, guid)
 		}
 		if len(newHooks) == 0 {
 			_ = s.DeleteHM(ATLAS_OBSERVER, key)
@@ -70,9 +70,10 @@ func (s *Storage) DeleteSubscriptions(subscriptions []blockatlas.Subscription) {
 		}
 		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..1f80e0e 100644
--- a/pkg/blockatlas/observer_test.go
+++ b/pkg/blockatlas/observer_test.go
@@ -9,48 +9,48 @@ 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: Subscriptions{
+			name: "guid with 1 coin",
+			subscriptions: SubscriptionEvent{
+				GUID: "http://127.0.0.1:8080",
+				OldSubscriptions: Subscriptions{
 					"0": {"xpub6BpYi6J1GZzfY3yY7DbhLLccF3efQa18nQngM3jaehgtNSoEgk6UtPULpC3oK5oA3trczY8Ld34LFw1USMPfGHwTEizdD5QyGcMyuh2UoBA", "xpub6CYwPfnPJLPquufPkb98coSb3mdy1CgaZrWUtYWGJTJ4VWZUbzH9HLGy7nHpP7DG4UdTkYYpirkTWQSP7pWHsrk24Nos5oYNHpfr4BgPVTL"},
 				},
 			},
 			wantSubs: []Subscription{
 				{
-					Coin: 0, Webhook: "http://127.0.0.1:8080",
+					Coin: 0, GUID: "http://127.0.0.1:8080",
 					Address: "xpub6BpYi6J1GZzfY3yY7DbhLLccF3efQa18nQngM3jaehgtNSoEgk6UtPULpC3oK5oA3trczY8Ld34LFw1USMPfGHwTEizdD5QyGcMyuh2UoBA",
 				},
 				{
-					Coin: 0, Webhook: "http://127.0.0.1:8080",
+					Coin: 0, GUID: "http://127.0.0.1:8080",
 					Address: "xpub6CYwPfnPJLPquufPkb98coSb3mdy1CgaZrWUtYWGJTJ4VWZUbzH9HLGy7nHpP7DG4UdTkYYpirkTWQSP7pWHsrk24Nos5oYNHpfr4BgPVTL",
 				},
 			},
 		},
 		{
-			name: "webhook with 2 coins",
-			subscriptions: Webhook{
-				Webhook: "http://127.0.0.1:8080",
-				Subscriptions: Subscriptions{
+			name: "guid with 2 coins",
+			subscriptions: SubscriptionEvent{
+				GUID: "http://127.0.0.1:8080",
+				OldSubscriptions: Subscriptions{
 					"2": {"zpub6rH4MwgyTmuexAX6HAraks5cKv5BbtmwdLirvnU5845ovUJb4abgjt9DtXK4ZEaToRrNj8dQznuLC6Nka4eMviGMinCVMUxKLpuyddcG9Vc"},
 					"0": {"xpub6BpYi6J1GZzfY3yY7DbhLLccF3efQa18nQngM3jaehgtNSoEgk6UtPULpC3oK5oA3trczY8Ld34LFw1USMPfGHwTEizdD5QyGcMyuh2UoBA", "xpub6CYwPfnPJLPquufPkb98coSb3mdy1CgaZrWUtYWGJTJ4VWZUbzH9HLGy7nHpP7DG4UdTkYYpirkTWQSP7pWHsrk24Nos5oYNHpfr4BgPVTL"},
 				},
 			},
 			wantSubs: []Subscription{
 				{
-					Coin: 2, Webhook: "http://127.0.0.1:8080",
+					Coin: 2, GUID: "http://127.0.0.1:8080",
 					Address: "zpub6rH4MwgyTmuexAX6HAraks5cKv5BbtmwdLirvnU5845ovUJb4abgjt9DtXK4ZEaToRrNj8dQznuLC6Nka4eMviGMinCVMUxKLpuyddcG9Vc",
 				},
 				{
-					Coin: 0, Webhook: "http://127.0.0.1:8080",
+					Coin: 0, GUID: "http://127.0.0.1:8080",
 					Address: "xpub6BpYi6J1GZzfY3yY7DbhLLccF3efQa18nQngM3jaehgtNSoEgk6UtPULpC3oK5oA3trczY8Ld34LFw1USMPfGHwTEizdD5QyGcMyuh2UoBA",
 				},
 				{
-					Coin: 0, Webhook: "http://127.0.0.1:8080",
+					Coin: 0, GUID: "http://127.0.0.1:8080",
 					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()
+			gotSubs := tt.subscriptions.ParseSubscriptions(tt.subscriptions.OldSubscriptions)
 			sort.Slice(gotSubs, func(i, j int) bool {
 				return gotSubs[i].Coin > gotSubs[j].Coin
 			})
diff --git a/platform/cosmos/transaction_test.go b/platform/cosmos/transaction_test.go
index c571ec9..32570c7 100644
--- a/platform/cosmos/transaction_test.go
+++ b/platform/cosmos/transaction_test.go
@@ -664,7 +664,7 @@ var failedTransferDst = blockatlas.Tx{
 	Fee:    "2000",
 	Date:   1576120902,
 	Block:  5552,
-	Status: blockatlas.StatusFailed,
+	Status: blockatlas.StatusError,
 	Type:   blockatlas.TxTransfer,
 	Memo:   "UniCoins registration rewards",
 	Meta: blockatlas.Transfer{
diff --git a/platform/ethereum/token_test.go b/platform/ethereum/token_test.go
index 1a87cdf..a2e9fdd 100644
--- a/platform/ethereum/token_test.go
+++ b/platform/ethereum/token_test.go
@@ -10,21 +10,17 @@ import (
 
 const tokenSrc = `
 {
-	"balance": "0",
-	"contract": {
-		"contract": "0xa14839c9837657efcde754ebeaf5cbecdd801b2a",
-		"address": "0xa14839c9837657efcde754ebeaf5cbecdd801b2a",
-		"name": "FusChain",
-		"decimals": 18,
-		"symbol": "FUS"
-	}
+	"address": "0xa14839c9837657EFcDE754EbEAF5cbECDd801B2A",
+	"name": "FusChain",
+	"decimals": 18,
+	"symbol": "FUS"
 }`
 
 var tokenDst = blockatlas.Token{
 	Name:     "FusChain",
 	Symbol:   "FUS",
 	Decimals: 18,
-	TokenID:  "0xa14839c9837657efcde754ebeaf5cbecdd801b2a",
+	TokenID:  "0xa14839c9837657EFcDE754EbEAF5cbECDd801B2A",
 	Coin:     coin.ETH,
 	Type:     blockatlas.TokenTypeERC20,
 }
@@ -44,7 +40,7 @@ func TestNormalizeToken(t *testing.T) {
 }
 
 func testNormalizeToken(t *testing.T, _test *testToken) {
-	var token Token
+	var token Contract
 	err := json.Unmarshal([]byte(_test.apiResponse), &token)
 	if err != nil {
 		t.Error(err)
diff --git a/platform/ethereum/transaction_test.go b/platform/ethereum/transaction_test.go
index c26a50f..eef7313 100644
--- a/platform/ethereum/transaction_test.go
+++ b/platform/ethereum/transaction_test.go
@@ -172,7 +172,7 @@ var failedDst = blockatlas.Tx{
 	Date:     1554662399,
 	Block:    7522678,
 	Sequence: 1,
-	Status:   blockatlas.StatusFailed,
+	Status:   blockatlas.StatusError,
 	Error:    "Error",
 	Meta: blockatlas.Transfer{
 		Value:    "59859820000000000",
diff --git a/platform/ripple/transaction_test.go b/platform/ripple/transaction_test.go
index ef640be..8cbc06a 100644
--- a/platform/ripple/transaction_test.go
+++ b/platform/ripple/transaction_test.go
@@ -199,7 +199,7 @@ var failedPaymentDst = blockatlas.Tx{
 	Date:   1581590872,
 	Block:  53401154,
 	Memo:   "",
-	Status: blockatlas.StatusFailed,
+	Status: blockatlas.StatusError,
 	Meta: blockatlas.Transfer{
 		Value:    "24999750000",
 		Symbol:   "XRP",
diff --git a/platform/tezos/transaction_test.go b/platform/tezos/transaction_test.go
index 125acb9..42e56ca 100644
--- a/platform/tezos/transaction_test.go
+++ b/platform/tezos/transaction_test.go
@@ -274,7 +274,7 @@ var transfer3 = blockatlas.Tx{
 		Symbol:   coin.Coins[coin.XTZ].Symbol,
 		Decimals: coin.Coins[coin.XTZ].Decimals,
 	},
-	Status: blockatlas.StatusFailed,
+	Status: blockatlas.StatusError,
 	Error:  "transaction failed",
 }
 
diff --git a/services/domains/domains_test.go b/services/domains/domains_test.go
index 3dc2f45..9786527 100644
--- a/services/domains/domains_test.go
+++ b/services/domains/domains_test.go
@@ -1,12 +1,15 @@
 package domains
 
 import (
+	"strings"
+	"testing"
+
 	"github.com/stretchr/testify/assert"
 	"github.com/trustwallet/blockatlas/pkg/errors"
-	"testing"
 )
 
 func checkGetTLD(t *testing.T, name string, expectedTLD string, expectedError error) {
+	name = strings.ToLower(name)
 	tld, err := getTLD(name)
 	assert.Equal(t, expectedTLD, tld)
 	if expectedError == nil {
@@ -20,6 +23,7 @@ func Test_getTLD(t *testing.T) {
 	checkGetTLD(t, "vitalik.eth", ".eth", nil)
 	checkGetTLD(t, "vitalik.ens", ".ens", nil)
 	checkGetTLD(t, "ourxyzwallet.xyz", ".xyz", nil)
+	checkGetTLD(t, "Cameron.Kred", ".kred", nil)
 	checkGetTLD(t, "btc.zil", ".zil", nil)
 	checkGetTLD(t, "btc.crypto", ".crypto", nil)
 	checkGetTLD(t, "nick@fiotestnet", "@fiotestnet", nil)
diff --git a/tests/functional/testdata/exclude.json b/tests/functional/testdata/exclude.json
index 87d3f8a..7143270 100644
--- a/tests/functional/testdata/exclude.json
+++ b/tests/functional/testdata/exclude.json
@@ -28,6 +28,8 @@
   "/v2/tomochain/collections/:owner/collection/:collection_id",
   "/v2/gochain/collections/:owner/collection/:collection_id",
   "/v2/ethereum/collections/:owner/collection/:collection_id",
+  "/v3/ethereum/collections/:owner/collection/:collection_id",
+  "/v4/ethereum/collections/:owner/collection/:collection_id",
   "/v3/classic/collections/:owner",
   "/v3/callisto/collections/:owner",
   "/v3/poa/collections/:owner",
diff --git a/tests/integration/domains/domains.go b/tests/integration/domains/domains.go
index 7dcb3f1..a5dac9e 100644
--- a/tests/integration/domains/domains.go
+++ b/tests/integration/domains/domains.go
@@ -3,11 +3,12 @@
 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"
-	"testing"
 )
 
 func TestDomains(t *testing.T) {
@@ -39,6 +40,13 @@ func TestDomains(t *testing.T) {
 			[]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",
diff --git a/tests/postman/Blockatlas.postman_collection.json b/tests/postman/Blockatlas.postman_collection.json
index 383850d..a0b5f81 100644
--- a/tests/postman/Blockatlas.postman_collection.json
+++ b/tests/postman/Blockatlas.postman_collection.json
@@ -1,6 +1,6 @@
 {
 	"info": {
-		"_postman_id": "63af4255-85a1-4505-87cb-13f8de0453af",
+		"_postman_id": "60e7e382-2f40-487f-a291-3e62daadadd8",
 		"name": "Blockatlas",
 		"schema": "https://schema.getpostman.com/json/collection/v2.1.0/collection.json"
 	},
@@ -2380,214 +2380,6 @@
 			],
 			"protocolProfileBehavior": {}
 		},
-		{
-			"name": "observer",
-			"item": [
-				{
-					"name": "add",
-					"event": [
-						{
-							"listen": "prerequest",
-							"script": {
-								"id": "7312c9e7-5330-421a-9ca8-171101050e38",
-								"exec": [
-									"",
-									"var body = {\"subscriptions\": {}, \"webhook\": \"\"}",
-									"",
-									"let subscriptions = pm.variables.get(\"subscriptions\");",
-									"let webhook = pm.variables.get(\"webhook\");",
-									"",
-									"body[\"subscriptions\"] = subscriptions",
-									"body[\"webhook\"] = webhook",
-									"",
-									"pm.environment.set(\"request_body\", JSON.stringify(body));",
-									""
-								],
-								"type": "text/javascript"
-							}
-						},
-						{
-							"listen": "test",
-							"script": {
-								"id": "acec8baa-0ef0-4703-ab83-c555260deffa",
-								"exec": [
-									"let result = {",
-									"    \"status\": true,",
-									"    \"message\": \"Added\"",
-									"};",
-									"var jsonData = pm.response.json();",
-									"",
-									"pm.test(\"response must be valid and have a body\", function () {",
-									"    pm.response.to.have.status(200);",
-									"    pm.response.to.be.ok;",
-									"    pm.response.to.not.be.error;",
-									"    pm.response.to.be.withBody;",
-									"    pm.response.to.be.json;",
-									"});",
-									"",
-									"pm.test(\"response is valid\", function() {",
-									"    pm.expect(jsonData).to.eql(result);",
-									"});",
-									""
-								],
-								"type": "text/javascript"
-							}
-						}
-					],
-					"request": {
-						"method": "POST",
-						"header": [
-							{
-								"key": "Authorization",
-								"type": "text",
-								"value": "Bearer {{observer_auth}}"
-							},
-							{
-								"key": "Content-Type",
-								"name": "Content-Type",
-								"value": "application/json",
-								"type": "text"
-							}
-						],
-						"body": {
-							"mode": "raw",
-							"raw": "{{request_body}}",
-							"options": {
-								"raw": {
-									"language": "json"
-								}
-							}
-						},
-						"url": {
-							"raw": "{{host}}/observer/v1/webhook/register",
-							"host": [
-								"{{host}}"
-							],
-							"path": [
-								"observer",
-								"v1",
-								"webhook",
-								"register"
-							]
-						}
-					},
-					"response": []
-				},
-				{
-					"name": "delete",
-					"event": [
-						{
-							"listen": "prerequest",
-							"script": {
-								"id": "7312c9e7-5330-421a-9ca8-171101050e38",
-								"exec": [
-									"",
-									"var body = {\"subscriptions\": {}, \"webhook\": \"\"}",
-									"",
-									"let subscriptions = pm.variables.get(\"subscriptions\");",
-									"let webhook = pm.variables.get(\"webhook\");",
-									"",
-									"body[\"subscriptions\"] = subscriptions",
-									"body[\"webhook\"] = webhook",
-									"",
-									"pm.environment.set(\"request_body\", JSON.stringify(body));",
-									""
-								],
-								"type": "text/javascript"
-							}
-						},
-						{
-							"listen": "test",
-							"script": {
-								"id": "acec8baa-0ef0-4703-ab83-c555260deffa",
-								"exec": [
-									"let result = {",
-									"    \"status\": true,",
-									"    \"message\": \"Deleted\"",
-									"};",
-									"var jsonData = pm.response.json();",
-									"",
-									"pm.test(\"response must be valid and have a body\", function () {",
-									"    pm.response.to.have.status(200);",
-									"    pm.response.to.be.ok;",
-									"    pm.response.to.not.be.error;",
-									"    pm.response.to.be.withBody;",
-									"    pm.response.to.be.json;",
-									"});",
-									"",
-									"pm.test(\"response is valid\", function() {",
-									"    pm.expect(jsonData).to.eql(result);",
-									"});",
-									""
-								],
-								"type": "text/javascript"
-							}
-						}
-					],
-					"request": {
-						"method": "DELETE",
-						"header": [
-							{
-								"key": "Authorization",
-								"type": "text",
-								"value": "Bearer {{observer_auth}}"
-							},
-							{
-								"key": "Content-Type",
-								"name": "Content-Type",
-								"type": "text",
-								"value": "application/json"
-							}
-						],
-						"body": {
-							"mode": "raw",
-							"raw": "{{request_body}}",
-							"options": {
-								"raw": {
-									"language": "json"
-								}
-							}
-						},
-						"url": {
-							"raw": "{{host}}/observer/v1/webhook/register",
-							"host": [
-								"{{host}}"
-							],
-							"path": [
-								"observer",
-								"v1",
-								"webhook",
-								"register"
-							]
-						}
-					},
-					"response": []
-				}
-			],
-			"event": [
-				{
-					"listen": "prerequest",
-					"script": {
-						"id": "50b8bc65-9acc-4652-8ac7-722f45767478",
-						"type": "text/javascript",
-						"exec": [
-							""
-						]
-					}
-				},
-				{
-					"listen": "test",
-					"script": {
-						"id": "88a2b78a-643e-449b-9ec0-f7c7a62ad50a",
-						"type": "text/javascript",
-						"exec": [
-							""
-						]
-					}
-				}
-			],
-			"protocolProfileBehavior": {}
-		},
 		{
 			"name": "healthcheck",
 			"item": [
diff --git a/tests/postman/observer_data.json b/tests/postman/observer_data.json
deleted file mode 100644
index 9c5394e..0000000
--- a/tests/postman/observer_data.json
+++ /dev/null
@@ -1,72 +0,0 @@
-[
-    {
-        "subscriptions": {
-            "0": [
-                "bc1qrfr44n2j4czd5c9txwlnw0yj2h82x9566fglqj"
-            ],
-            "2": [
-                "ltc1qpm594ntjq6ayqjngf6t9td2dxtey9d7985eept"
-            ],
-            "60": [
-                "0x0875BCab22dE3d02402bc38aEe4104e1239374a7"
-            ],
-            "459": [
-                "kava1l8va9zyl50cpzv447c694k3jndelc9ygtfll2m"
-            ],
-            "714": [
-                "bnb1jxfh2g85q3v0tdq56fnevx6xcxtcnhtsmcu64m"
-            ]
-        },
-        "webhook": "http://localhost"
-    },
-    {
-        "subscriptions": {
-            "0": [
-                "bc1qrfr44n2j4czd5c9txwlnw0yj2h82x9566fglqj"
-            ],
-            "2": [
-                "ltc1qpm594ntjq6ayqjngf6t9td2dxtey9d7985eept"
-            ],
-            "60": [
-                "0x0875BCab22dE3d02402bc38aEe4104e1239374a7"
-            ],
-            "459": [
-                "kava1l8va9zyl50cpzv447c694k3jndelc9ygtfll2m"
-            ]
-        },
-        "webhook": "http://localhost"
-    },
-    {
-        "subscriptions": {
-            "0": [
-                "bc1qrfr44n2j4czd5c9txwlnw0yj2h82x9566fglqj"
-            ],
-            "2": [
-                "ltc1qpm594ntjq6ayqjngf6t9td2dxtey9d7985eept"
-            ],
-            "60": [
-                "0x0875BCab22dE3d02402bc38aEe4104e1239374a7"
-            ]
-        },
-        "webhook": "http://localhost"
-    },
-    {
-        "subscriptions": {
-            "0": [
-                "bc1qrfr44n2j4czd5c9txwlnw0yj2h82x9566fglqj"
-            ],
-            "2": [
-                "ltc1qpm594ntjq6ayqjngf6t9td2dxtey9d7985eept"
-            ]
-        },
-        "webhook": "http://localhost"
-    },
-    {
-        "subscriptions": {
-            "0": [
-                "bc1qrfr44n2j4czd5c9txwlnw0yj2h82x9566fglqj"
-            ]
-        },
-        "webhook": "http://localhost"
-    }
-]
\ No newline at end of file