Repository: trustwallet/blockatlas
Pull request #1197 (2020-08-17 22:10:13)

[General] Implement token searcher 
https://github.com/trustwallet/blockatlas/issues/1172

codecov[bot]: # [Codecov](https://codecov.io/gh/trustwallet/blockatlas/pull/1197?src=pr&el=h1) Report
> Merging [#1197](https://codecov.io/gh/trustwallet/blockatlas/pull/1197?src=pr&el=desc) into [master](https://codecov.io/gh/trustwallet/blockatlas/commit/768eb7b83925d29891989bb0cddbbaecaa912cec&el=desc) will **decrease** coverage by `0%`.
> The diff coverage is `0%`.

[![Impacted file tree graph](https://codecov.io/gh/trustwallet/blockatlas/pull/1197/graphs/tree.svg?width=650&height=150&src=pr&token=jnrjoEQAqv)](https://codecov.io/gh/trustwallet/blockatlas/pull/1197?src=pr&el=tree)

```diff
@@           Coverage Diff           @@
##           master   #1197    +/-   ##
=======================================
- Coverage      38%     37%    -1%     
=======================================
  Files         168     169     +1     
  Lines        5997    6136   +139     
=======================================
  Hits         2318    2318            
- Misses       3508    3647   +139     
  Partials      171     171            
```


| [Impacted Files](https://codecov.io/gh/trustwallet/blockatlas/pull/1197?src=pr&el=tree) | Coverage Δ | |
|---|---|---|
| [db/db.go](https://codecov.io/gh/trustwallet/blockatlas/pull/1197/diff?src=pr&el=tree#diff-ZGIvZGIuZ28=) | `0% <0%> (ø)` | |
| [db/tokenassociations.go](https://codecov.io/gh/trustwallet/blockatlas/pull/1197/diff?src=pr&el=tree#diff-ZGIvdG9rZW5hc3NvY2lhdGlvbnMuZ28=) | `0% <0%> (ø)` | |
| [pkg/blockatlas/tx.go](https://codecov.io/gh/trustwallet/blockatlas/pull/1197/diff?src=pr&el=tree#diff-cGtnL2Jsb2NrYXRsYXMvdHguZ28=) | `68% <0%> (-14%)` | :arrow_down: |

------

[Continue to review full report at Codecov](https://codecov.io/gh/trustwallet/blockatlas/pull/1197?src=pr&el=continue).
> **Legend** - [Click here to learn more](https://docs.codecov.io/docs/codecov-delta)
> `Δ = absolute <relative> (impact)`, `ø = not affected`, `? = missing data`
> Powered by [Codecov](https://codecov.io/gh/trustwallet/blockatlas/pull/1197?src=pr&el=footer). Last update [bea0c7f...d15851e](https://codecov.io/gh/trustwallet/blockatlas/pull/1197?src=pr&el=lastupdated). Read the [comment docs](https://docs.codecov.io/docs/pull-request-comments).

todo[bot]: ## fix GetAssetsMapByAddresses

https://github.com/trustwallet/blockatlas/blob/bb0e8464df707c32d2ca26856828022c7b631c8c/tests/integration/db_test/tokenassociations_test.go#L24-L29

---

###### This comment was generated by [todo](https://todo.jasonet.co) based on a `TODO` comment in bb0e8464df707c32d2ca26856828022c7b631c8c in #1197. cc @trustwallet.
todo[bot]: ## Fix

https://github.com/trustwallet/blockatlas/blob/157e462411459c421665a592fd8aa5e78a9b01cb/tests/integration/db_test/subscriptions_test.go#L201-L206

---

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

Patch:
diff --git a/.github/workflows/master.yml b/.github/workflows/master.yml
index 3414c5b..75861de 100644
--- a/.github/workflows/master.yml
+++ b/.github/workflows/master.yml
@@ -112,15 +112,18 @@ jobs:
            PARSER_IMAGE: ${{ secrets.REGISTRY }}:parser-${{ steps.hash.outputs.sha7 }}
            OBSERVER_SUBSCRIBER_IMAGE: ${{ secrets.REGISTRY }}:subscriber-${{ steps.hash.outputs.sha7 }}
            OBSERVER_NOTIFIER_IMAGE: ${{ secrets.REGISTRY }}:notifier-${{ steps.hash.outputs.sha7 }}
+           SEARCHER_IMAGE: ${{ secrets.REGISTRY }}:searcher-${{ steps.hash.outputs.sha7 }}
          run: |
            docker build --build-arg SERVICE=api/api -f Dockerfile.runner -t $API_IMAGE .
            docker build --build-arg SERVICE=parser/parser -f Dockerfile.runner -t $PARSER_IMAGE .
            docker build --build-arg SERVICE=subscriber/subscriber -f Dockerfile.runner -t $OBSERVER_SUBSCRIBER_IMAGE .
            docker build --build-arg SERVICE=notifier/notifier -f Dockerfile.runner -t $OBSERVER_NOTIFIER_IMAGE .
+           docker build --build-arg SERVICE=searcher/searcher -f Dockerfile.runner -t $SEARCHER_IMAGE .
            docker push $API_IMAGE
            docker push $PARSER_IMAGE
            docker push $OBSERVER_SUBSCRIBER_IMAGE
            docker push $OBSERVER_NOTIFIER_IMAGE
+           docker push $SEARCHER_IMAGE
 
   deploy:
     name: CD
diff --git a/.github/workflows/pr-build.yml b/.github/workflows/pr-build.yml
index 1b81a1b..1759a51 100644
--- a/.github/workflows/pr-build.yml
+++ b/.github/workflows/pr-build.yml
@@ -64,12 +64,15 @@ jobs:
            PARSER_IMAGE: ${{ secrets.REGISTRY }}:parser-${{ steps.hash.outputs.sha7 }}
            OBSERVER_SUBSCRIBER_IMAGE: ${{ secrets.REGISTRY }}:subscriber-${{ steps.hash.outputs.sha7 }}
            OBSERVER_NOTIFIER_IMAGE: ${{ secrets.REGISTRY }}:notifier-${{ steps.hash.outputs.sha7 }}
+           SEARCHER_IMAGE: ${{ secrets.REGISTRY }}:searcher-${{ steps.hash.outputs.sha7 }}
          run: |
            docker build --build-arg SERVICE=api/api -f Dockerfile.runner -t $API_IMAGE .
            docker build --build-arg SERVICE=parser/parser -f Dockerfile.runner -t $PARSER_IMAGE .
            docker build --build-arg SERVICE=subscriber/subscriber -f Dockerfile.runner -t $OBSERVER_SUBSCRIBER_IMAGE .
            docker build --build-arg SERVICE=notifier/notifier -f Dockerfile.runner -t $OBSERVER_NOTIFIER_IMAGE .
+           docker build --build-arg SERVICE=searcher/searcher -f Dockerfile.runner -t $SEARCHER_IMAGE .
            docker push $API_IMAGE
            docker push $PARSER_IMAGE
            docker push $OBSERVER_SUBSCRIBER_IMAGE
            docker push $OBSERVER_NOTIFIER_IMAGE
+           docker push $SEARCHER_IMAGE
diff --git a/Makefile b/Makefile
index cb3c7b9..d630b42 100644
--- a/Makefile
+++ b/Makefile
@@ -10,6 +10,7 @@ API := api
 NOTIFIER := notifier
 PARSER := parser
 SUBSCRIBER := subscriber
+SEARCHER := searcher
 COIN_FILE := coin/coins.yml
 COIN_GO_FILE := coin/coins.go
 GEN_COIN_FILE := coin/gen.go
@@ -42,6 +43,7 @@ STDERR := /tmp/.$(PROJECT_NAME)-stderr.txt
 
 # PID file will keep the process id of the server
 PID_API := /tmp/.$(PROJECT_NAME).$(API).pid
+PID_SEARCHER := /tmp/.$(PROJECT_NAME).$(SEARCHER).pid
 PID_NOTIFIER := /tmp/.$(PROJECT_NAME).$(NOTIFIER).pid
 PID_PARSER := /tmp/.$(PROJECT_NAME).$(PARSER).pid
 PID_SUBSCRIBER := /tmp/.$(PROJECT_NAME).$(SUBSCRIBER).pid
@@ -54,52 +56,60 @@ install: go-get
 
 ## start: Start API, Observer and Sync in development mode.
 start:
-	@bash -c "$(MAKE) clean compile start-api start-parser start-notifier start-subscriber"
+	@bash -c "$(MAKE) clean compile start-api start-parser start-notifier start-subscriber start-searcher"
 
 ## start-api: Start platform api in development mode.
 start-api: stop
 	@echo "  >  Starting $(PROJECT_NAME)"
 	@-$(GOBIN)/$(API)/api -c $(CONFIG_FILE) 2>&1 & echo $$! > $(PID_API)
-	@cat $(PID_API) | sed "/^/s/^/  \>  API PID: /"
+	@cat $(PID_API) | sed "/^/s/^/  \>  Api PID: /"
 	@echo "  >  Error log: $(STDERR)"
 
 # start-platform-api-mock: Start API.  Similar to start-platform-api, but uses config file with mock URLs, and port 8437.
 start-api-mock: stop start-mockserver
 	@echo "  >  Starting $(PROJECT_NAME) API"
 	@-$(GOBIN)/$(API)/api -p 8437 -c $(CONFIG_MOCK_FILE) 2>&1 & echo $$! > $(PID_API)
-	@cat $(PID_API) | sed "/^/s/^/  \>  API PID: /"
+	@cat $(PID_API) | sed "/^/s/^/  \>  Mock PID: /"
 	@echo "  >  Error log: $(STDERR)"
 
 ## start-observer-parser: Start observer-parser in development mode.
 start-parser: stop
 	@echo "  >  Starting $(PROJECT_NAME)"
 	@-$(GOBIN)/$(PARSER)/parser -c $(CONFIG_FILE) 2>&1 & echo $$! > $(PID_PARSER)
-	@cat $(PID_PARSER) | sed "/^/s/^/  \>  Sync PID: /"
+	@cat $(PID_PARSER) | sed "/^/s/^/  \>  Parser PID: /"
 	@echo "  >  Error log: $(STDERR)"
 
 ## start-observer-notifier: Start observer-notifier in development mode.
 start-notifier: stop
 	@echo "  >  Starting $(PROJECT_NAME)"
 	@-$(GOBIN)/$(NOTIFIER)/notifier -c $(CONFIG_FILE) 2>&1 & echo $$! > $(PID_NOTIFIER)
-	@cat $(PID_NOTIFIER) | sed "/^/s/^/  \>  Sync PID: /"
+	@cat $(PID_NOTIFIER) | sed "/^/s/^/  \>  Notifier PID: /"
 	@echo "  >  Error log: $(STDERR)"
 
 ## start-observer-subscriber: Start observer-subscriber in development mode.
 start-subscriber: stop
 	@echo "  >  Starting $(PROJECT_NAME)"
 	@-$(GOBIN)/$(SUBSCRIBER)/subscriber -c $(CONFIG_FILE) 2>&1 & echo $$! > $(PID_SUBSCRIBER)
-	@cat $(PID_SUBSCRIBER) | sed "/^/s/^/  \>  Sync PID: /"
+	@cat $(PID_SUBSCRIBER) | sed "/^/s/^/  \>  Subscriber PID: /"
+	@echo "  >  Error log: $(STDERR)"
+
+## start-api: Start searcher in development mode.
+start-searcher: stop
+	@echo "  >  Starting $(PROJECT_NAME)"
+	@-$(GOBIN)/$(SEARCHER)/searcher -c $(CONFIG_FILE) 2>&1 & echo $$! > $(PID_SEARCHER)
+	@cat $(PID_SEARCHER) | sed "/^/s/^/  \>  Searcher PID: /"
 	@echo "  >  Error log: $(STDERR)"
 
 ## stop: Stop development mode.
 stop:
-	@-touch $(PID_API) $(PID_NOTIFIER) $(PID_PARSER) $(PID_SUBSCRIBER) $(PID_MOCKSERVER)
+	@-touch $(PID_API) $(PID_NOTIFIER) $(PID_PARSER) $(PID_SUBSCRIBER) $(PID_MOCKSERVER) $(PID_SEARCHER)
 	@-kill `cat $(PID_API)` 2> /dev/null || true
 	@-kill `cat $(PID_NOTIFIER)` 2> /dev/null || true
 	@-kill `cat $(PID_PARSER)` 2> /dev/null || true
 	@-kill `cat $(PID_SUBSCRIBER)` 2> /dev/null || true
 	@-kill `cat $(PID_MOCKSERVER)` 2> /dev/null || true
-	@-rm $(PID_API) $(PID_NOTIFIER) $(PID_PARSER) $(PID_SUBSCRIBER) $(PID_MOCKSERVER)
+	@-kill `cat $(PID_SEARCHER)` 2> /dev/null || true
+	@-rm $(PID_API) $(PID_NOTIFIER) $(PID_PARSER) $(PID_SUBSCRIBER) $(PID_MOCKSERVER) $(PID_SEARCHER)
 
 stop-mockserver:
 	@-touch $(PID_MOCKSERVER)
@@ -219,7 +229,7 @@ endif
 
 go-compile: go-get go-build
 
-go-build: go-build-api go-build-notifier go-build-parser go-build-subscriber
+go-build: go-build-api go-build-notifier go-build-parser go-build-subscriber go-build-searcher
 
 docker-shutdown:
 	@echo "  >  Shutdown docker containers..."
@@ -247,6 +257,10 @@ go-build-subscriber:
 	@echo "  >  Building subscriber binary..."
 	GOBIN=$(GOBIN) go build $(LDFLAGS) -o $(GOBIN)/$(SUBSCRIBER)/subscriber ./cmd/$(SUBSCRIBER)
 
+go-build-searcher:
+	@echo "  >  Building searcher binary..."
+	GOBIN=$(GOBIN) go build $(LDFLAGS) -o $(GOBIN)/$(SEARCHER)/searcher ./cmd/$(SEARCHER)
+
 go-generate:
 	@echo "  >  Generating dependency files..."
 	GOBIN=$(GOBIN) go generate $(generate)
diff --git a/api/api.go b/api/api.go
index 32bc850..01ed507 100644
--- a/api/api.go
+++ b/api/api.go
@@ -6,6 +6,7 @@ import (
 	"github.com/swaggo/gin-swagger/swaggerFiles"
 	_ "github.com/trustwallet/blockatlas/docs"
 	"github.com/trustwallet/blockatlas/platform"
+	"github.com/trustwallet/blockatlas/services/tokensearcher"
 )
 
 func SetupPlatformAPI(router gin.IRouter) {
@@ -23,6 +24,10 @@ func SetupPlatformAPI(router gin.IRouter) {
 	RegisterBasicAPI(router)
 }
 
+func SetupTokensIndexAPI(router gin.IRouter, instance tokensearcher.Instance) {
+	RegisterTokensIndexAPI(router, instance)
+}
+
 func SetupSwaggerAPI(router gin.IRouter) {
 	router.GET("swagger/*any", ginSwagger.WrapHandler(swaggerFiles.Handler))
 }
diff --git a/api/endpoint/token.go b/api/endpoint/token.go
index 84da5b7..6c0c6a1 100644
--- a/api/endpoint/token.go
+++ b/api/endpoint/token.go
@@ -3,6 +3,7 @@ package endpoint
 import (
 	"github.com/gin-gonic/gin"
 	"github.com/trustwallet/blockatlas/pkg/blockatlas"
+	"github.com/trustwallet/blockatlas/services/tokensearcher"
 	"net/http"
 	"strconv"
 	"sync"
@@ -123,3 +124,17 @@ func getTokens(tokenAPI blockatlas.TokensAPI, addresses []string, data *tokensRe
 	}
 	data.mu.Unlock()
 }
+
+func GetTokensByAddressIndexer(c *gin.Context, instance tokensearcher.Instance) {
+	var query tokensearcher.Request
+	if err := c.Bind(&query); err != nil {
+		c.AbortWithStatusJSON(http.StatusBadRequest, errorResponse(err))
+		return
+	}
+	result, err := instance.HandleTokensRequest(query, c.Request.Context())
+	if err != nil {
+		c.AbortWithStatusJSON(http.StatusInternalServerError, errorResponse(err))
+		return
+	}
+	c.JSON(http.StatusOK, result)
+}
diff --git a/api/registry.go b/api/registry.go
index 2346249..fdd4eff 100644
--- a/api/registry.go
+++ b/api/registry.go
@@ -8,6 +8,7 @@ import (
 	"github.com/trustwallet/blockatlas/api/middleware"
 	"github.com/trustwallet/blockatlas/pkg/blockatlas"
 	"github.com/trustwallet/blockatlas/platform"
+	"github.com/trustwallet/blockatlas/services/tokensearcher"
 	"time"
 )
 
@@ -106,3 +107,9 @@ func RegisterBasicAPI(router gin.IRouter) {
 	router.GET("/", endpoint.GetStatus)
 	router.GET("/metrics", ginprom.PromHandler(promhttp.Handler()))
 }
+
+func RegisterTokensIndexAPI(router gin.IRouter, instance tokensearcher.Instance) {
+	router.POST("/v3/tokens", func(c *gin.Context) {
+		endpoint.GetTokensByAddressIndexer(c, instance)
+	})
+}
diff --git a/cmd/api/main.go b/cmd/api/main.go
index b840f24..ab6f604 100644
--- a/cmd/api/main.go
+++ b/cmd/api/main.go
@@ -4,20 +4,28 @@ import (
 	"github.com/gin-gonic/gin"
 	"github.com/spf13/viper"
 	"github.com/trustwallet/blockatlas/api"
+	"github.com/trustwallet/blockatlas/db"
 	_ "github.com/trustwallet/blockatlas/docs"
 	"github.com/trustwallet/blockatlas/internal"
+	"github.com/trustwallet/blockatlas/mq"
 	"github.com/trustwallet/blockatlas/pkg/logger"
 	"github.com/trustwallet/blockatlas/platform"
+	"github.com/trustwallet/blockatlas/services/tokensearcher"
+	"time"
 )
 
 const (
 	defaultPort       = "8420"
 	defaultConfigPath = "../../config.yml"
+	prod              = "prod"
 )
 
 var (
 	port, confPath string
 	engine         *gin.Engine
+	database       *db.Instance
+	t              tokensearcher.Instance
+	restAPI        string
 )
 
 func init() {
@@ -26,18 +34,42 @@ func init() {
 	internal.InitConfig(confPath)
 	logger.InitLogger()
 
+	restAPI = viper.GetString("rest_api")
 	engine = internal.InitEngine(viper.GetString("gin.mode"))
 
+	if restAPI == "tokens" || restAPI == "all" {
+		pgUri := viper.GetString("postgres.uri")
+
+		var err error
+		database, err = db.New(pgUri, prod)
+		if err != nil {
+			logger.Fatal(err)
+		}
+
+		mqHost := viper.GetString("observer.rabbitmq.uri")
+		prefetchCount := viper.GetInt("observer.rabbitmq.consumer.prefetch_count")
+		internal.InitRabbitMQ(mqHost, prefetchCount)
+		if err := mq.TokensRegistration.Declare(); err != nil {
+			logger.Fatal(err)
+		}
+		t = tokensearcher.Init(database, platform.TokensAPIs, mq.TokensRegistration)
+
+		go db.RestoreConnectionWorker(database, time.Second*10, pgUri)
+		go mq.FatalWorker(time.Second * 10)
+	}
 	platform.Init(viper.GetStringSlice("platform"))
 }
 
 func main() {
-	switch viper.GetString("rest_api") {
+	switch restAPI {
 	case "swagger":
 		api.SetupSwaggerAPI(engine)
 	case "platform":
 		api.SetupPlatformAPI(engine)
+	case "tokens":
+		api.SetupTokensIndexAPI(engine, t)
 	default:
+		api.SetupTokensIndexAPI(engine, t)
 		api.SetupSwaggerAPI(engine)
 		api.SetupPlatformAPI(engine)
 	}
diff --git a/cmd/notifier/main.go b/cmd/notifier/main.go
index b71fe18..5bee2ce 100644
--- a/cmd/notifier/main.go
+++ b/cmd/notifier/main.go
@@ -7,7 +7,7 @@ import (
 	"github.com/trustwallet/blockatlas/internal"
 	"github.com/trustwallet/blockatlas/mq"
 	"github.com/trustwallet/blockatlas/pkg/logger"
-	"github.com/trustwallet/blockatlas/services/observer/notifier"
+	"github.com/trustwallet/blockatlas/services/notifier"
 	"time"
 )
 
diff --git a/cmd/parser/main.go b/cmd/parser/main.go
index ccbd384..e9b721c 100644
--- a/cmd/parser/main.go
+++ b/cmd/parser/main.go
@@ -9,7 +9,7 @@ import (
 	"github.com/trustwallet/blockatlas/mq"
 	"github.com/trustwallet/blockatlas/pkg/logger"
 	"github.com/trustwallet/blockatlas/platform"
-	"github.com/trustwallet/blockatlas/services/observer/parser"
+	"github.com/trustwallet/blockatlas/services/parser"
 	"os"
 	"os/signal"
 	"sync"
@@ -107,7 +107,7 @@ func main() {
 		params := parser.Params{
 			Ctx:                   ctx,
 			Api:                   api,
-			Queue:                 mq.RawTransactions,
+			Queue:                 []mq.Queue{mq.RawTransactions, mq.TokensRegistration},
 			ParsingBlocksInterval: pollInterval,
 			FetchBlocksTimeout:    fetchBlocksInterval,
 			BacklogCount:          backlogCount,
diff --git a/cmd/searcher/main.go b/cmd/searcher/main.go
new file mode 100644
index 0000000..b58faa1
--- /dev/null
+++ b/cmd/searcher/main.go
@@ -0,0 +1,74 @@
+package main
+
+import (
+	"context"
+	"github.com/spf13/viper"
+	"github.com/trustwallet/blockatlas/db"
+	"github.com/trustwallet/blockatlas/internal"
+	"github.com/trustwallet/blockatlas/mq"
+	"github.com/trustwallet/blockatlas/pkg/logger"
+	"github.com/trustwallet/blockatlas/services/notifier"
+	"github.com/trustwallet/blockatlas/services/tokensearcher"
+	"time"
+)
+
+const (
+	defaultConfigPath = "../../config.yml"
+	prod              = "prod"
+)
+
+var (
+	confPath string
+	database *db.Instance
+)
+
+func init() {
+	_, confPath = internal.ParseArgs("", defaultConfigPath)
+
+	internal.InitConfig(confPath)
+	logger.InitLogger()
+
+	mqHost := viper.GetString("observer.rabbitmq.uri")
+	prefetchCount := viper.GetInt("observer.rabbitmq.consumer.prefetch_count")
+	maxPushNotificationsBatchLimit := viper.GetUint("observer.push_notifications_batch_limit")
+	pgUri := viper.GetString("postgres.uri")
+
+	internal.InitRabbitMQ(mqHost, prefetchCount)
+
+	if err := mq.RawTransactions.Declare(); err != nil {
+		logger.Fatal(err)
+	}
+
+	if err := mq.TxNotifications.Declare(); err != nil {
+		logger.Fatal(err)
+	}
+
+	var err error
+	database, err = db.New(pgUri, prod)
+	if err != nil {
+		logger.Fatal(err)
+	}
+
+	if maxPushNotificationsBatchLimit == 0 {
+		notifier.MaxPushNotificationsBatchLimit = notifier.DefaultPushNotificationsBatchLimit
+	} else {
+		notifier.MaxPushNotificationsBatchLimit = maxPushNotificationsBatchLimit
+	}
+
+	logger.Info("maxPushNotificationsBatchLimit ", logger.Params{"limit": maxPushNotificationsBatchLimit})
+
+	go mq.FatalWorker(time.Second * 10)
+	go db.RestoreConnectionWorker(database, time.Second*10, pgUri)
+
+	time.Sleep(time.Millisecond)
+}
+
+func main() {
+	defer mq.Close()
+
+	ctx, cancel := context.WithCancel(context.Background())
+
+	go mq.TokensRegistration.RunConsumerWithCancelAndDbConn(tokensearcher.Run, database, ctx)
+
+	internal.SetupGracefulShutdownForObserver(cancel)
+}
diff --git a/cmd/subscriber/main.go b/cmd/subscriber/main.go
index 38d8dfd..ee130d0 100644
--- a/cmd/subscriber/main.go
+++ b/cmd/subscriber/main.go
@@ -8,7 +8,7 @@ import (
 	"github.com/trustwallet/blockatlas/internal"
 	"github.com/trustwallet/blockatlas/mq"
 	"github.com/trustwallet/blockatlas/pkg/logger"
-	"github.com/trustwallet/blockatlas/services/observer/subscriber"
+	"github.com/trustwallet/blockatlas/services/subscriber"
 	"time"
 )
 
@@ -51,9 +51,20 @@ func main() {
 	if err := mq.Subscriptions.Declare(); err != nil {
 		logger.Fatal(err)
 	}
+	if err := mq.TokensRegistration.Declare(); err != nil {
+		logger.Fatal(err)
+	}
 	ctx, cancel := context.WithCancel(context.Background())
 
-	go mq.Subscriptions.RunConsumerWithCancelAndDbConn(subscriber.RunSubscriber, database, ctx)
+	subscriberType := subscriber.Subscriber(viper.GetString("subscriber"))
+	switch subscriberType {
+	case subscriber.Tokens:
+		go mq.TokensRegistration.RunConsumerWithCancelAndDbConn(subscriber.RunTokensSubscriber, database, ctx)
+	case subscriber.Notifications:
+		go mq.Subscriptions.RunConsumerWithCancelAndDbConn(subscriber.RunTransactionsSubscriber, database, ctx)
+	default:
+		logger.Fatal("bad subscriber: " + subscriberType)
+	}
 
 	internal.SetupGracefulShutdownForObserver(cancel)
 }
diff --git a/config.yml b/config.yml
index 9e00143..c9c2b9b 100644
--- a/config.yml
+++ b/config.yml
@@ -11,9 +11,12 @@ gin:
 # You can see all the coin handles at coins/coins.yml file
 platform: [all]
 
-# Can be platform or swagger
+# Can be platform or swagger, or tokens, or all
 rest_api: all
 
+# Can be tokens or notifications
+subscriber: tokens
+
 # The transaction watcher
 observer:
   # Don't request blocks older than this
@@ -69,7 +72,7 @@ ethereum:
   api: https://localhost:4567 #(Trust-Ray API)
   blockbook_api: https://eth1.trezor.io
   collections_api: https://api.opensea.io
-#  collections_api_key: [opensea_api_key]
+  #  collections_api_key: [opensea_api_key]
   rpc: https://main-rpc.linkpool.io
 
 # [ETC] Ethereum Classic: https://ethereumclassic.org (Trust-Ray API)
diff --git a/configmock.yml b/configmock.yml
index 4e030b4..db71678 100644
--- a/configmock.yml
+++ b/configmock.yml
@@ -13,8 +13,8 @@ gin:
 # You can see all the coin handles at coins/coins.yml file
 platform: all
 
-# Can be platform or swagger
-rest_api: all
+# Can be platform or swagger, or tokens, or all
+rest_api: platform
 
 # The transaction watcher
 observer:
diff --git a/db/addresstoasset.go b/db/addresstoasset.go
new file mode 100644
index 0000000..924a40d
--- /dev/null
+++ b/db/addresstoasset.go
@@ -0,0 +1,279 @@
+package db
+
+import (
+	"context"
+	"github.com/jinzhu/gorm"
+	"github.com/trustwallet/blockatlas/db/models"
+	"go.elastic.co/apm/module/apmgorm"
+	"time"
+)
+
+func (i Instance) GetSubscribedAddressesForAssets(ctx context.Context, addresses []string) ([]models.Address, error) {
+	db := apmgorm.WithContext(ctx, i.Gorm)
+
+	addressesSubQuery := db.
+		Table("addresses").
+		Select("id").
+		Where("address in (?)", addresses).
+		QueryExpr()
+
+	var assetSubs []models.AssetSubscription
+	err := db.
+		Preload("Address").
+		Where("address_id in (?)", addressesSubQuery).
+		Find(&assetSubs).
+		Error
+	if err != nil {
+		return nil, err
+	}
+
+	var result []models.Address
+	for _, a := range assetSubs {
+		result = append(result, a.Address)
+	}
+	return result, nil
+}
+
+func (i Instance) GetAssetsMapByAddresses(addresses []string, ctx context.Context) (map[string][]string, error) {
+	db := apmgorm.WithContext(ctx, i.Gorm)
+
+	var dbAddresses []models.Address
+	err := db.Where("address in (?)", addresses).Find(&dbAddresses).Error
+	if err != nil {
+		return nil, err
+	}
+
+	addressesIDs := make([]uint, 0, len(dbAddresses))
+	for _, a := range dbAddresses {
+		addressesIDs = append(addressesIDs, a.ID)
+	}
+
+	var associations []models.AddressToAssetAssociation
+	err = db.
+		Preload("Address").
+		Preload("Asset").
+		Where("address_id in (?)", addressesIDs).
+		Find(&associations).
+		Error
+	if err != nil {
+		return nil, err
+	}
+
+	result := make(map[string][]string)
+	for _, a := range associations {
+		assets := result[a.Address.Address]
+		result[a.Address.Address] = append(assets, a.Asset.AssetID)
+	}
+	return result, nil
+}
+
+func (i Instance) GetAssetsMapByAddressesFromTime(addresses []string, from time.Time, ctx context.Context) (map[string][]string, error) {
+	db := apmgorm.WithContext(ctx, i.Gorm)
+
+	var dbAddresses []models.Address
+	err := db.Where("address in (?)", addresses).Find(&dbAddresses).Error
+	if err != nil {
+		return nil, err
+	}
+
+	addressesIDs := make([]uint, 0, len(dbAddresses))
+	for _, a := range dbAddresses {
+		addressesIDs = append(addressesIDs, a.ID)
+	}
+
+	var associations []models.AddressToAssetAssociation
+	err = db.
+		Preload("Address").
+		Preload("Asset").
+		Where("address_id in (?)", addressesIDs).
+		Where("updated_at > ?", from).
+		Find(&associations).
+		Error
+	if err != nil {
+		return nil, err
+	}
+
+	result := make(map[string][]string)
+	for _, a := range associations {
+		assets := result[a.Address.Address]
+		result[a.Address.Address] = append(assets, a.Asset.AssetID)
+	}
+	return result, nil
+}
+
+func (i *Instance) GetAssociationsByAddresses(addresses []string, ctx context.Context) ([]models.AddressToAssetAssociation, error) {
+	db := apmgorm.WithContext(ctx, i.Gorm)
+
+	addressesSubQuery := db.Table("addresses").
+		Select("id").
+		Where("address in (?)", addresses).
+		QueryExpr()
+
+	var result []models.AddressToAssetAssociation
+	err := db.
+		Preload("Address").
+		Preload("Asset").
+		Where("address_id in (?)", addressesSubQuery).
+		Find(&result).
+		Error
+	return result, err
+}
+
+func (i *Instance) GetAssociationsByAddressesFromTime(addresses []string, from time.Time, ctx context.Context) ([]models.AddressToAssetAssociation, error) {
+	db := apmgorm.WithContext(ctx, i.Gorm)
+
+	addressesSubQuery := db.Table("addresses").
+		Select("id").
+		Where("address in (?)", addresses).
+		QueryExpr()
+
+	var result []models.AddressToAssetAssociation
+	err := db.
+		Preload("Address").
+		Preload("Asset").
+		Where("address_id in (?)", addressesSubQuery).
+		Where("updated_at > ?", from).
+		Find(&result).
+		Error
+	return result, err
+}
+
+func (i *Instance) AddAssociationsForAddress(address string, assets []string, ctx context.Context) error {
+	db := apmgorm.WithContext(ctx, i.Gorm)
+	return db.Transaction(func(tx *gorm.DB) error {
+		uniqueAssets := getUniqueStrings(assets)
+		uniqueAssetsModel := make([]models.Asset, 0, len(uniqueAssets))
+		for _, l := range uniqueAssets {
+			uniqueAssetsModel = append(uniqueAssetsModel, models.Asset{
+				AssetID: l,
+			})
+		}
+
+		err := BulkInsert(db.Set("gorm:insert_option", "ON CONFLICT DO NOTHING"), uniqueAssetsModel)
+		if err != nil {
+			return err
+		}
+
+		var dbAssets []models.Asset
+		err = db.Where("asset_id in (?)", uniqueAssets).Find(&dbAssets).Error
+		if err != nil {
+			return err
+		}
+
+		dbAddress := models.Address{Address: address}
+		err = db.Where("address = ?", address).FirstOrCreate(&dbAddress).Error
+		if err != nil {
+			return err
+		}
+
+		assetsSub := models.AssetSubscription{AddressID: dbAddress.ID}
+		err = db.Set("gorm:insert_option", "ON CONFLICT DO NOTHING").Create(&assetsSub).Error
+		if err != nil {
+			return err
+		}
+
+		result := make([]models.AddressToAssetAssociation, 0, len(dbAssets))
+		for _, asset := range dbAssets {
+			result = append(result, models.AddressToAssetAssociation{
+				AddressID: dbAddress.ID,
+				AssetID:   asset.ID,
+			})
+		}
+		return BulkInsert(db.Set("gorm:insert_option", "ON CONFLICT DO NOTHING"), result)
+	})
+}
+
+func (i *Instance) UpdateAssociationsForExistingAddresses(associations map[string][]string, ctx context.Context) error {
+	db := apmgorm.WithContext(ctx, i.Gorm)
+	return db.Transaction(func(tx *gorm.DB) error {
+		var assets []string
+
+		for _, v := range associations {
+			assets = append(assets, v...)
+		}
+
+		uniqueAssets := getUniqueStrings(assets)
+		uniqueAssetsModel := make([]models.Asset, 0, len(uniqueAssets))
+		for _, l := range uniqueAssets {
+			uniqueAssetsModel = append(uniqueAssetsModel, models.Asset{
+				AssetID: l,
+			})
+		}
+
+		err := BulkInsert(db.Set("gorm:insert_option", "ON CONFLICT DO NOTHING"), uniqueAssetsModel)
+		if err != nil {
+			return err
+		}
+
+		var dbAssets []models.Asset
+		err = db.Where("asset_id in (?)", uniqueAssets).Find(&dbAssets).Error
+		if err != nil {
+			return err
+		}
+
+		assetsMap := makeMapAssets(dbAssets)
+
+		var addresses []string
+		for k := range associations {
+			addresses = append(addresses, k)
+		}
+
+		var dbAddresses []models.Address
+		if err := db.Where("address in (?)", addresses).Find(&dbAddresses).Error; err != nil {
+			return err
+		}
+
+		var addressSubs []models.AssetSubscription
+		for _, a := range dbAddresses {
+			sub := models.AssetSubscription{AddressID: a.ID}
+			addressSubs = append(addressSubs, sub)
+		}
+
+		err = BulkInsert(db.Set("gorm:insert_option", "ON CONFLICT DO NOTHING"), addressSubs)
+		if err != nil {
+			return err
+		}
+
+		addressesMap := makeMapAddress(dbAddresses)
+
+		var result []models.AddressToAssetAssociation
+		for address, assets := range associations {
+			for _, asset := range assets {
+				r := models.AddressToAssetAssociation{
+					AddressID: addressesMap[address],
+					AssetID:   assetsMap[asset],
+				}
+				result = append(result, r)
+			}
+		}
+		return BulkInsert(db.Set("gorm:insert_option", "ON CONFLICT DO NOTHING"), result)
+	})
+}
+
+func makeMapAssets(addresses []models.Asset) map[string]uint {
+	result := make(map[string]uint)
+	for _, a := range addresses {
+		result[a.AssetID] = a.ID
+	}
+	return result
+}
+
+func makeMapAddress(addresses []models.Address) map[string]uint {
+	result := make(map[string]uint)
+	for _, a := range addresses {
+		result[a.Address] = a.ID
+	}
+	return result
+}
+
+func getUniqueStrings(values []string) []string {
+	keys := make(map[string]bool)
+	var list []string
+	for _, entry := range values {
+		if _, value := keys[entry]; !value {
+			keys[entry] = true
+			list = append(list, entry)
+		}
+	}
+	return list
+}
diff --git a/db/db.go b/db/db.go
index 5eed3e7..dc56c92 100644
--- a/db/db.go
+++ b/db/db.go
@@ -1,11 +1,14 @@
 package db
 
 import (
+	"errors"
 	"github.com/jinzhu/gorm"
+	gormbulk "github.com/t-tiger/gorm-bulk-insert"
 	"github.com/trustwallet/blockatlas/db/models"
 	"github.com/trustwallet/blockatlas/pkg/logger"
 	"go.elastic.co/apm/module/apmgorm"
 	_ "go.elastic.co/apm/module/apmgorm/dialects/postgres"
+	"reflect"
 	"time"
 )
 
@@ -13,6 +16,8 @@ type Instance struct {
 	Gorm *gorm.DB
 }
 
+const batchCount = 3000
+
 func New(uri, env string) (*Instance, error) {
 	var (
 		g   *gorm.DB
@@ -29,8 +34,12 @@ func New(uri, env string) (*Instance, error) {
 	}
 
 	g.AutoMigrate(
-		&models.Subscription{},
+		&models.NotificationSubscription{},
 		&models.Tracker{},
+		&models.AddressToAssetAssociation{},
+		&models.Asset{},
+		&models.AssetSubscription{},
+		&models.Address{},
 	)
 
 	i := &Instance{Gorm: g}
@@ -59,3 +68,50 @@ func RestoreConnectionWorker(database *Instance, timeout time.Duration, uri stri
 		time.Sleep(timeout)
 	}
 }
+
+// Example:
+// postgres.BulkInsert(DBWrite, []models.User{...})
+func BulkInsert(db *gorm.DB, dbModels interface{}) error {
+	interfaceSlice, err := getInterfaceSlice(dbModels)
+	if err != nil {
+		return err
+	}
+	batchList := getInterfaceSliceBatch(interfaceSlice, batchCount)
+	for _, batch := range batchList {
+		err := gormbulk.BulkInsert(db, batch, len(batch))
+		if err != nil {
+			return err
+		}
+	}
+	return nil
+}
+
+func getInterfaceSliceBatch(values []interface{}, sizeUint uint) [][]interface{} {
+	size := int(sizeUint)
+	resultLength := (len(values) + size - 1) / size
+	result := make([][]interface{}, resultLength)
+	lo, hi := 0, size
+	for i := range result {
+		if hi > len(values) {
+			hi = len(values)
+		}
+		result[i] = values[lo:hi:hi]
+		lo, hi = hi, hi+size
+	}
+	return result
+}
+
+func getInterfaceSlice(slice interface{}) ([]interface{}, error) {
+	s := reflect.ValueOf(slice)
+	if s.Kind() != reflect.Slice {
+		return nil, errors.New("InterfaceSlice() given a non-slice type")
+	}
+
+	ret := make([]interface{}, s.Len())
+
+	for i := 0; i < s.Len(); i++ {
+		ret[i] = s.Index(i).Interface()
+	}
+
+	return ret, nil
+}
diff --git a/db/models/address.go b/db/models/address.go
new file mode 100644
index 0000000..85b211f
--- /dev/null
+++ b/db/models/address.go
@@ -0,0 +1,8 @@
+package models
+
+import "github.com/jinzhu/gorm"
+
+type Address struct {
+	gorm.Model
+	Address string `gorm:"type:varchar(128); unique" sql:"index"`
+}
diff --git a/db/models/addresstoasset.go b/db/models/addresstoasset.go
new file mode 100644
index 0000000..034ddfe
--- /dev/null
+++ b/db/models/addresstoasset.go
@@ -0,0 +1,17 @@
+package models
+
+import (
+	"time"
+)
+
+type AddressToAssetAssociation struct {
+	CreatedAt time.Time
+	UpdatedAt time.Time
+	DeletedAt *time.Time `sql:"index"`
+
+	Address   Address `gorm:"ForeignKey:AddressID; not null"`
+	AddressID uint    `sql:"index"`
+
+	Asset   Asset `gorm:"ForeignKey:AssetID; not null"`
+	AssetID uint  `sql:"index"`
+}
diff --git a/db/models/asset.go b/db/models/asset.go
new file mode 100644
index 0000000..79441c4
--- /dev/null
+++ b/db/models/asset.go
@@ -0,0 +1,8 @@
+package models
+
+import "github.com/jinzhu/gorm"
+
+type Asset struct {
+	gorm.Model
+	AssetID string `gorm:"type:varchar(128); unique_index"`
+}
diff --git a/db/models/subscriptions.go b/db/models/subscriptions.go
index 019159e..909df39 100644
--- a/db/models/subscriptions.go
+++ b/db/models/subscriptions.go
@@ -1,11 +1,19 @@
 package models
 
 import (
-	"time"
+	"github.com/jinzhu/gorm"
 )
 
-type Subscription struct {
-	CreatedAt time.Time  `gorm:"default:CURRENT_TIMESTAMP"`
-	Coin      uint       `gorm:"primary_key; column:coin; auto_increment:false" sql:"index"`
-	Address   string     `gorm:"primary_key; column:address; type:varchar(128)" sql:"index"`
-}
+type (
+	NotificationSubscription struct {
+		gorm.Model
+		Address   Address `gorm:"ForeignKey:AddressID; not null"`
+		AddressID uint    `gorm:"unique" sql:"index"`
+	}
+
+	AssetSubscription struct {
+		gorm.Model
+		Address   Address `gorm:"ForeignKey:AddressID; not null"`
+		AddressID uint    `gorm:"unique" sql:"index"`
+	}
+)
diff --git a/db/notification.go b/db/notification.go
new file mode 100644
index 0000000..9455c88
--- /dev/null
+++ b/db/notification.go
@@ -0,0 +1,86 @@
+package db
+
+import (
+	"context"
+	"github.com/jinzhu/gorm"
+	"github.com/trustwallet/blockatlas/db/models"
+	"github.com/trustwallet/blockatlas/pkg/errors"
+	"go.elastic.co/apm/module/apmgorm"
+)
+
+func (i *Instance) GetSubscriptionsForNotifications(addresses []string, ctx context.Context) ([]models.NotificationSubscription, error) {
+	if len(addresses) == 0 {
+		return nil, errors.E("Empty addresses")
+	}
+	db := apmgorm.WithContext(ctx, i.Gorm)
+
+	addressesSubQuery := db.
+		Table("addresses").
+		Select("id").
+		Where("address in (?)", addresses).
+		QueryExpr()
+
+	var subscriptionsDataList []models.NotificationSubscription
+	err := db.
+		Preload("Address").
+		Where("address_id in (?)", addressesSubQuery).
+		Find(&subscriptionsDataList).
+		Error
+	if err != nil {
+		return nil, err
+	}
+	return subscriptionsDataList, nil
+}
+
+func (i *Instance) AddSubscriptionsForNotifications(addresses []string, ctx context.Context) error {
+	if len(addresses) == 0 {
+		return errors.E("Empty subscriptions")
+	}
+	db := apmgorm.WithContext(ctx, i.Gorm)
+
+	return db.Transaction(func(tx *gorm.DB) error {
+		uniqueAddresses := getUniqueStrings(addresses)
+		uniqueAddressesModel := make([]models.Address, 0, len(uniqueAddresses))
+		for _, a := range uniqueAddresses {
+			uniqueAddressesModel = append(uniqueAddressesModel, models.Address{
+				Address: a,
+			})
+		}
+
+		err := BulkInsert(db.Set("gorm:insert_option", "ON CONFLICT DO NOTHING"), uniqueAddressesModel)
+		if err != nil {
+			return err
+		}
+
+		var dbAddresses []models.Address
+		err = db.Where("address in (?)", uniqueAddresses).Find(&dbAddresses).Error
+		if err != nil {
+			return err
+		}
+
+		result := make([]models.NotificationSubscription, 0, len(dbAddresses))
+		for _, a := range dbAddresses {
+			result = append(result, models.NotificationSubscription{
+				AddressID: a.ID,
+			})
+		}
+		return BulkInsert(db.Set("gorm:insert_option", "ON CONFLICT (address_id) DO UPDATE SET deleted_at = null, updated_at = now()"), result)
+	})
+}
+
+func (i *Instance) DeleteSubscriptionsForNotifications(addresses []string, ctx context.Context) error {
+	if len(addresses) == 0 {
+		return errors.E("Empty subscriptions")
+	}
+	db := apmgorm.WithContext(ctx, i.Gorm)
+
+	addressSubQuery := db.Table("addresses").
+		Select("id").
+		Where("address in (?)", addresses).
+		QueryExpr()
+
+	return db.
+		Where("address_id in (?)", addressSubQuery).
+		Delete(&models.NotificationSubscription{}).
+		Error
+}
diff --git a/db/subscriptions.go b/db/subscriptions.go
deleted file mode 100644
index 121ce34..0000000
--- a/db/subscriptions.go
+++ /dev/null
@@ -1,105 +0,0 @@
-package db
-
-import (
-	"context"
-	"fmt"
-	"github.com/jinzhu/gorm"
-	"github.com/trustwallet/blockatlas/db/models"
-	"github.com/trustwallet/blockatlas/pkg/errors"
-	"go.elastic.co/apm"
-	"go.elastic.co/apm/module/apmgorm"
-	"strings"
-)
-
-func (i *Instance) GetSubscriptions(coin uint, addresses []string, ctx context.Context) ([]models.Subscription, error) {
-	if len(addresses) == 0 {
-		return nil, errors.E("Empty addresses")
-	}
-	g := apmgorm.WithContext(ctx, i.Gorm)
-	var subscriptionsDataList []models.Subscription
-	err := g.
-		Model(&models.Subscription{}).
-		Where("address in (?) AND coin = ?", addresses, coin).
-		Find(&subscriptionsDataList).Error
-
-	if err != nil {
-		return nil, err
-	}
-	return subscriptionsDataList, nil
-}
-
-func (i *Instance) AddSubscriptions(subscriptions []models.Subscription, ctx context.Context) error {
-	if len(subscriptions) == 0 {
-		return errors.E("Empty subscriptions")
-	}
-
-	subscriptionsBatch := toSubscriptionBatch(subscriptions, batchLimit, ctx)
-	g := apmgorm.WithContext(ctx, i.Gorm)
-
-	for _, s := range subscriptionsBatch {
-		if err := bulkCreate(g, s); err != nil {
-			return err
-		}
-	}
-
-	return nil
-}
-
-func (i *Instance) DeleteSubscriptions(subscriptions []models.Subscription, ctx context.Context) error {
-	if len(subscriptions) == 0 {
-		return errors.E("Empty subscriptions")
-	}
-
-	g := apmgorm.WithContext(ctx, i.Gorm)
-	for _, s := range subscriptions {
-		err := g.Where("coin = ? and address = ?", s.Coin, s.Address).Delete(&models.Subscription{}).Error
-		if err != nil {
-			return err
-		}
-	}
-	return nil
-}
-
-const (
-	batchLimit    = 3000
-	rawBulkInsert = `INSERT INTO subscriptions(coin,address) VALUES %s ON CONFLICT DO NOTHING`
-)
-
-func bulkCreate(db *gorm.DB, dataList []models.Subscription) error {
-	var (
-		valueStrings []string
-		valueArgs    []interface{}
-	)
-
-	for _, d := range dataList {
-		valueStrings = append(valueStrings, "(?, ?)")
-
-		valueArgs = append(valueArgs, d.Coin)
-		valueArgs = append(valueArgs, d.Address)
-	}
-
-	smt := fmt.Sprintf(rawBulkInsert, strings.Join(valueStrings, ","))
-
-	if err := db.Exec(smt, valueArgs...).Error; err != nil {
-		return err
-	}
-
-	return nil
-}
-
-func toSubscriptionBatch(txs []models.Subscription, sizeUint uint, ctx context.Context) [][]models.Subscription {
-	span, _ := apm.StartSpan(ctx, "toSubscriptionBatch", "app")
-	defer span.End()
-	size := int(sizeUint)
-	resultLength := (len(txs) + size - 1) / size
-	result := make([][]models.Subscription, resultLength)
-	lo, hi := 0, size
-	for i := range result {
-		if hi > len(txs) {
-			hi = len(txs)
-		}
-		result[i] = txs[lo:hi:hi]
-		lo, hi = hi, hi+size
-	}
-	return result
-}
diff --git a/go.mod b/go.mod
index a174c85..94cb232 100644
--- a/go.mod
+++ b/go.mod
@@ -3,31 +3,18 @@ module github.com/trustwallet/blockatlas
 go 1.14
 
 require (
-	github.com/Azure/go-ansiterm v0.0.0-20170929234023-d6e3b3328b78 // indirect
 	github.com/DATA-DOG/go-sqlmock v1.5.0
-	github.com/Microsoft/go-winio v0.4.14 // indirect
-	github.com/Nvveen/Gotty v0.0.0-20120604004816-cd527374f1e5 // indirect
 	github.com/alecthomas/template v0.0.0-20190718012654-fb15b899a751
 	github.com/btcsuite/btcutil v1.0.2
-	github.com/cenkalti/backoff v2.2.1+incompatible // indirect
 	github.com/chenjiandongx/ginprom v0.0.0-20200410120253-7cfb22707fa6
-	github.com/containerd/continuity v0.0.0-20200413184840-d3ef23f19fbb // indirect
 	github.com/deckarep/golang-set v1.7.1
-	github.com/docker/go-connections v0.4.0 // indirect
-	github.com/docker/go-units v0.4.0 // indirect
-	github.com/elastic/go-sysinfo v1.3.0 // indirect
-	github.com/elastic/go-windows v1.0.1 // indirect
 	github.com/gin-gonic/gin v1.6.3
-	github.com/gotestyourself/gotestyourself v2.2.0+incompatible // indirect
 	github.com/imroc/req v0.3.0
 	github.com/jinzhu/gorm v1.9.16
 	github.com/mitchellh/mapstructure v1.3.3
 	github.com/mr-tron/base58 v1.2.0
-	github.com/opencontainers/image-spec v1.0.1 // indirect
-	github.com/opencontainers/runc v0.1.1 // indirect
 	github.com/ory/dockertest v3.3.5+incompatible
 	github.com/patrickmn/go-cache v2.1.0+incompatible
-	github.com/pkg/errors v0.9.1 // indirect
 	github.com/prometheus/client_golang v1.7.1
 	github.com/shopspring/decimal v1.2.0
 	github.com/sirupsen/logrus v1.6.0
@@ -36,18 +23,16 @@ require (
 	github.com/stretchr/testify v1.6.1
 	github.com/swaggo/gin-swagger v1.2.0
 	github.com/swaggo/swag v1.6.7
+	github.com/t-tiger/gorm-bulk-insert v1.3.0
 	github.com/trustwallet/ens-coincodec v1.0.6
+	github.com/trustwallet/watchmarket v1.1.0
 	go.elastic.co/apm v1.8.0
 	go.elastic.co/apm/module/apmgin v1.8.0
 	go.elastic.co/apm/module/apmgorm v1.8.0
 	go.elastic.co/apm/module/apmhttp v1.8.0
 	go.elastic.co/apm/module/apmlogrus v1.8.0
-	go.elastic.co/fastjson v1.1.0 // indirect
 	go.uber.org/atomic v1.6.0
 	golang.org/x/crypto v0.0.0-20200510223506-06a226fb4e37
 	golang.org/x/net v0.0.0-20200506145744-7e3656a0809f
-	golang.org/x/tools v0.0.0-20200513175351-0951661448da // indirect
 	gopkg.in/yaml.v2 v2.3.0
-	gotest.tools v2.2.0+incompatible // indirect
-	howett.net/plist v0.0.0-20200419221736-3b63eb3a43b5 // indirect
 )
diff --git a/go.sum b/go.sum
index 5c5a6c4..ae48691 100644
--- a/go.sum
+++ b/go.sum
@@ -44,6 +44,8 @@ github.com/alecthomas/template v0.0.0-20190718012654-fb15b899a751 h1:JYp7IbQjafo
 github.com/alecthomas/template v0.0.0-20190718012654-fb15b899a751/go.mod h1:LOuyumcjzFXgccqObfd/Ljyb9UuFJ6TxHnclSeseNhc=
 github.com/alecthomas/units v0.0.0-20151022065526-2efee857e7cf/go.mod h1:ybxpYRFXyAe+OPACYpWeL0wqObRcbAqCMya13uyzqw0=
 github.com/alecthomas/units v0.0.0-20190717042225-c3de453c63f4/go.mod h1:ybxpYRFXyAe+OPACYpWeL0wqObRcbAqCMya13uyzqw0=
+github.com/alicebob/gopher-json v0.0.0-20200520072559-a9ecdc9d1d3a/go.mod h1:SGnFV6hVsYE877CKEZ6tDNTjaSXYUk6QqoIK6PrAtcc=
+github.com/alicebob/miniredis/v2 v2.13.1/go.mod h1:0UIBNuf97uxrWhdVBpJvPtafKyGpL2NS2pYe0tYM97k=
 github.com/andybalholm/cascadia v1.1.0/go.mod h1:GsXiBklL0woXo1j/WYWtSYYC4ouU9PqHO0sqidkEA4Y=
 github.com/apache/thrift v0.12.0/go.mod h1:cp2SuWMxlEZw2r+iP2GNCdIi4C1qmUzdZFSVb+bacwQ=
 github.com/armon/circbuf v0.0.0-20150827004946-bbbad097214e/go.mod h1:3U/XgcO3hCbHZ8TKRvWD2dDTCfh9M9ya+I9JpbB7O8o=
@@ -77,6 +79,9 @@ github.com/cespare/xxhash/v2 v2.1.1 h1:6MnRN8NT7+YBpUIWxHtefFZOKTAPgGjpQSxqLNn0+
 github.com/cespare/xxhash/v2 v2.1.1/go.mod h1:VGX0DQ3Q6kWi7AoAeZDth3/j3BFtOZR5XLFGgcrjCOs=
 github.com/chenjiandongx/ginprom v0.0.0-20200410120253-7cfb22707fa6 h1:3Wv1E0CqL45hBs4bykb586wvZMRGKfFITHrN3ilm4FE=
 github.com/chenjiandongx/ginprom v0.0.0-20200410120253-7cfb22707fa6/go.mod h1:lINNCb1ZH3c0uL/9ApaQ8muR4QILsi0STj8Ojt8ZmwU=
+github.com/chzyer/logex v1.1.10/go.mod h1:+Ywpsq7O8HXn0nuIou7OrIPyXbp3wmkHB+jjWRnGsAI=
+github.com/chzyer/readline v0.0.0-20180603132655-2972be24d48e/go.mod h1:nSuG5e5PlCu98SY8svDHJxuZscDgtXS6KTTbou5AhLI=
+github.com/chzyer/test v0.0.0-20180213035817-a1ea475d72b1/go.mod h1:Q3SI9o4m/ZMnBNeIyt5eFwwo7qiLfzFZmjNmxjkiQlU=
 github.com/client9/misspell v0.3.4/go.mod h1:qj6jICC3Q7zFZvVWo7KLAzC3yx5G7kyvSDkc90ppPyw=
 github.com/containerd/continuity v0.0.0-20200413184840-d3ef23f19fbb h1:nXPkFq8X1a9ycY3GYQpFNxHh3j2JgY7zDZfq2EXMIzk=
 github.com/containerd/continuity v0.0.0-20200413184840-d3ef23f19fbb/go.mod h1:Dq467ZllaHgAtVp4p1xUQWBrFXR9s/wyoTpG8zOJGkY=
@@ -162,6 +167,8 @@ github.com/go-playground/universal-translator v0.17.0 h1:icxd5fm+REJzpZx7ZfpaD87
 github.com/go-playground/universal-translator v0.17.0/go.mod h1:UkSxE5sNxxRwHyU+Scu5vgOQjsIJAF8j9muTVoKLVtA=
 github.com/go-playground/validator/v10 v10.2.0 h1:KgJ0snyC2R9VXYN2rneOtQcw5aHQB1Vv0sFl1UcHBOY=
 github.com/go-playground/validator/v10 v10.2.0/go.mod h1:uOYAAleCW8F/7oMFd6aG0GOhaH6EGOAJShg8Id5JGkI=
+github.com/go-redis/redis v6.15.3-0.20190424063336-97e6ed817821+incompatible/go.mod h1:NAIEuMOZ/fxfXJIrKDQDz8wamY7mA7PouImQ2Jvg6kA=
+github.com/go-redis/redis v6.15.8+incompatible/go.mod h1:NAIEuMOZ/fxfXJIrKDQDz8wamY7mA7PouImQ2Jvg6kA=
 github.com/go-sql-driver/mysql v1.4.1 h1:g24URVg0OFbNUTx9qqY1IRZ9D9z3iPyi5zKhQZpNwpA=
 github.com/go-sql-driver/mysql v1.4.1/go.mod h1:zAC/RDZ24gD3HViQzih4MyKcchzm+sOG5ZlKdlhCg5w=
 github.com/go-sql-driver/mysql v1.5.0 h1:ozyZYNQW3x3HtqT1jira07DN2PArx2v7/mN66gGcHOs=
@@ -190,6 +197,7 @@ github.com/golang/protobuf v1.4.0/go.mod h1:jodUvKwWbYaEsadDk5Fwe5c77LiNKVO9IDvq
 github.com/golang/protobuf v1.4.2 h1:+Z5KGCizgyZCbGh1KZqA0fcLLkwbsjIzS4aV2v7wJX0=
 github.com/golang/protobuf v1.4.2/go.mod h1:oDoupMAO8OvCJWAcko0GGGIgR6R6ocIYbsSw735rRwI=
 github.com/golang/snappy v0.0.0-20180518054509-2e65f85255db/go.mod h1:/XxbfmMg8lxefKM7IXC3fBNl/7bRcc72aCRzEWrmP2Q=
+github.com/gomodule/redigo v1.8.1/go.mod h1:P9dn9mFrCBvWhGE1wpxx6fgq7BAeLBk+UUUzlpkBYO0=
 github.com/google/btree v0.0.0-20180813153112-4030bb1f1f0c/go.mod h1:lNA+9X1NB3Zf8V7Ke586lFgjr2dZNuvo3lPJSGZ5JPQ=
 github.com/google/btree v1.0.0/go.mod h1:lNA+9X1NB3Zf8V7Ke586lFgjr2dZNuvo3lPJSGZ5JPQ=
 github.com/google/go-cmp v0.2.0/go.mod h1:oXzfMopK8JAjlY9xF4vHSVASa0yLyX7SntLO5aqRK0M=
@@ -242,6 +250,8 @@ github.com/inconshreveable/mousetrap v1.0.0/go.mod h1:PxqpIevigyE2G7u3NXJIT2ANyt
 github.com/jessevdk/go-flags v0.0.0-20141203071132-1679536dcc89/go.mod h1:4FA24M0QyGHXBuZZK/XkWh8h0e1EYbRYJSGM75WSRxI=
 github.com/jessevdk/go-flags v1.4.0/go.mod h1:4FA24M0QyGHXBuZZK/XkWh8h0e1EYbRYJSGM75WSRxI=
 github.com/jinzhu/gorm v1.9.10/go.mod h1:Kh6hTsSGffh4ui079FHrR5Gg+5D0hgihqDcsDN2BBJY=
+github.com/jinzhu/gorm v1.9.11/go.mod h1:bu/pK8szGZ2puuErfU0RwyeNdsf3e6nCX/noXaVxkfw=
+github.com/jinzhu/gorm v1.9.12/go.mod h1:vhTjlKSJUTWNtcbQtrMBFCxy7eXTzeCAzfL5fBZT/Qs=
 github.com/jinzhu/gorm v1.9.15 h1:OdR1qFvtXktlxk73XFYMiYn9ywzTwytqe4QkuMRqc38=
 github.com/jinzhu/gorm v1.9.15/go.mod h1:G3LB3wezTOWM2ITLzPxEXgSkOXAntiLHS7UdBefADcs=
 github.com/jinzhu/gorm v1.9.16 h1:+IyIjPEABKRpsu/F8OvDPy9fyQlgsg2luMV2ZIH5i5o=
@@ -296,8 +306,10 @@ github.com/mattn/go-isatty v0.0.8/go.mod h1:Iq45c/XA43vh69/j3iqttzPXn0bhXyGjM0Hd
 github.com/mattn/go-isatty v0.0.12 h1:wuysRhFDzyxgEmMf5xjvJ2M9dZoWAXNNr5LSBS7uHXY=
 github.com/mattn/go-isatty v0.0.12/go.mod h1:cbi8OIDigv2wuxKPP5vlRcQ1OAZbq2CE4Kysco4FUpU=
 github.com/mattn/go-sqlite3 v1.10.0/go.mod h1:FPy6KqzDD04eiIsT53CuJW3U88zkxoIYsOqkbpncsNc=
+github.com/mattn/go-sqlite3 v1.11.0/go.mod h1:FPy6KqzDD04eiIsT53CuJW3U88zkxoIYsOqkbpncsNc=
 github.com/mattn/go-sqlite3 v1.14.0 h1:mLyGNKR8+Vv9CAU7PphKa2hkEqxxhn8i32J6FPj1/QA=
 github.com/mattn/go-sqlite3 v1.14.0/go.mod h1:JIl7NbARA7phWnGvh0LKTyg7S9BA+6gx71ShQilpsus=
+github.com/mattn/go-sqlite3 v2.0.1+incompatible/go.mod h1:FPy6KqzDD04eiIsT53CuJW3U88zkxoIYsOqkbpncsNc=
 github.com/matttproud/golang_protobuf_extensions v1.0.1 h1:4hp9jkHxhMHkqkrB3Ix0jegS5sx/RkqARlsWZ6pIwiU=
 github.com/matttproud/golang_protobuf_extensions v1.0.1/go.mod h1:D8He9yQNgCq6Z5Ld7szi9bcBfOoFv/3dc6xSMkL2PC0=
 github.com/miekg/dns v1.0.14/go.mod h1:W1PPwlIAgtquWBMBEV9nkV9Cazfe8ScdGz/Lj7v3Nrg=
@@ -308,6 +320,7 @@ github.com/mitchellh/gox v0.4.0/go.mod h1:Sd9lOJ0+aimLBi73mGofS1ycjY8lL3uZM3JPS4
 github.com/mitchellh/iochan v1.0.0/go.mod h1:JwYml1nuB7xOzsp52dPpHFffvOCDupsG0QubkSMEySY=
 github.com/mitchellh/mapstructure v0.0.0-20160808181253-ca63d7c062ee/go.mod h1:FVVH3fgwuzCH5S8UJGiWEs2h04kUh9fWfEaFds41c1Y=
 github.com/mitchellh/mapstructure v1.1.2/go.mod h1:FVVH3fgwuzCH5S8UJGiWEs2h04kUh9fWfEaFds41c1Y=
+github.com/mitchellh/mapstructure v1.3.0/go.mod h1:bFUtVrKA4DC2yAKiSyO/QUcy7e+RRV2QTWOzhPopBRo=
 github.com/mitchellh/mapstructure v1.3.3 h1:SzB1nHZ2Xi+17FP0zVQBHIZqvwRN9408fJO8h+eeNA8=
 github.com/mitchellh/mapstructure v1.3.3/go.mod h1:bFUtVrKA4DC2yAKiSyO/QUcy7e+RRV2QTWOzhPopBRo=
 github.com/modern-go/concurrent v0.0.0-20180228061459-e0a39a4cb421/go.mod h1:6dJC0mAP4ikYIbvyc7fijjWJddQyLn8Ig3JB5CqoB9Q=
@@ -316,14 +329,17 @@ github.com/modern-go/concurrent v0.0.0-20180306012644-bacd9c7ef1dd/go.mod h1:6dJ
 github.com/modern-go/reflect2 v0.0.0-20180701023420-4b7aa43c6742/go.mod h1:bx2lNnkwVCuqBIxFjflWJWanXIb3RllmbCylyMrvgv0=
 github.com/modern-go/reflect2 v1.0.1 h1:9f412s+6RmYXLWZSEzVVgPGK7C2PphHj5RJrvfx9AWI=
 github.com/modern-go/reflect2 v1.0.1/go.mod h1:bx2lNnkwVCuqBIxFjflWJWanXIb3RllmbCylyMrvgv0=
+github.com/mr-tron/base58 v1.1.3/go.mod h1:BinMc/sQntlIE1frQmRFPUoPA1Zkr8VRgBdjWI2mNwc=
 github.com/mr-tron/base58 v1.2.0 h1:T/HDJBh4ZCPbU39/+c3rRvE0uKBQlU27+QI8LJ4t64o=
 github.com/mr-tron/base58 v1.2.0/go.mod h1:BinMc/sQntlIE1frQmRFPUoPA1Zkr8VRgBdjWI2mNwc=
 github.com/mwitkow/go-conntrack v0.0.0-20161129095857-cc309e4a2223/go.mod h1:qRWi+5nqEBWmkhHvq77mSJWrCKwh8bxhgT7d/eI7P4U=
 github.com/oklog/ulid v1.3.1/go.mod h1:CirwcVhetQ6Lv90oh/F+FBtV6XMibvdAFo93nm5qn4U=
 github.com/onsi/ginkgo v1.6.0/go.mod h1:lLunBs/Ym6LB5Z9jYTR76FiuTmxDTDusOGeTQH+WWjE=
 github.com/onsi/ginkgo v1.7.0/go.mod h1:lLunBs/Ym6LB5Z9jYTR76FiuTmxDTDusOGeTQH+WWjE=
+github.com/onsi/ginkgo v1.8.0/go.mod h1:lLunBs/Ym6LB5Z9jYTR76FiuTmxDTDusOGeTQH+WWjE=
 github.com/onsi/ginkgo v1.10.1/go.mod h1:lLunBs/Ym6LB5Z9jYTR76FiuTmxDTDusOGeTQH+WWjE=
 github.com/onsi/gomega v1.4.3/go.mod h1:ex+gbHU/CVuBBDIJjb2X0qEXbFg53c61hWP/1CpauHY=
+github.com/onsi/gomega v1.5.0/go.mod h1:ex+gbHU/CVuBBDIJjb2X0qEXbFg53c61hWP/1CpauHY=
 github.com/onsi/gomega v1.7.0/go.mod h1:ex+gbHU/CVuBBDIJjb2X0qEXbFg53c61hWP/1CpauHY=
 github.com/opencontainers/go-digest v1.0.0-rc1 h1:WzifXhOVOEOuFYOJAW6aQqW0TooG2iki3E3Ii+WN7gQ=
 github.com/opencontainers/go-digest v1.0.0-rc1/go.mod h1:cMLVZDEM3+U2I4VmLI6N8jQYUd2OVphdqWwCJHrFt2s=
@@ -353,6 +369,7 @@ github.com/prometheus/client_golang v0.9.1/go.mod h1:7SWBe2y4D6OKWSNQJUaRYU/AaXP
 github.com/prometheus/client_golang v0.9.3-0.20190127221311-3c4408c8b829/go.mod h1:p2iRAGwDERtqlqzRXnrOVns+ignqQo//hLXqYxZYVNs=
 github.com/prometheus/client_golang v0.9.3/go.mod h1:/TN21ttK/J9q6uSwhBd54HahCDft0ttaMvbicHlPoso=
 github.com/prometheus/client_golang v1.0.0/go.mod h1:db9x61etRT2tGnBNRi70OPL5FsnadC4Ky3P0J6CfImo=
+github.com/prometheus/client_golang v1.6.0/go.mod h1:ZLOG9ck3JLRdB5MgO8f+lLTe83AXG6ro35rLTxvnIl4=
 github.com/prometheus/client_golang v1.7.1 h1:NTGy1Ja9pByO+xAeH/qiWnLrKtr3hJPNjaVUwnjpdpA=
 github.com/prometheus/client_golang v1.7.1/go.mod h1:PY5Wy2awLA44sXw4AOSfFBetzPP4j5+D6mVACh+pe2M=
 github.com/prometheus/client_model v0.0.0-20180712105110-5c3871d89910/go.mod h1:MbSGuTsp3dbXC40dX6PRTWyKYBIrTGTE9sqQNg2J8bo=
@@ -364,6 +381,7 @@ github.com/prometheus/common v0.0.0-20181113130724-41aa239b4cce/go.mod h1:daVV7q
 github.com/prometheus/common v0.2.0/go.mod h1:TNfzLD0ON7rHzMJeJkieUDPYmFC7Snx/y86RQel1bk4=
 github.com/prometheus/common v0.4.0/go.mod h1:TNfzLD0ON7rHzMJeJkieUDPYmFC7Snx/y86RQel1bk4=
 github.com/prometheus/common v0.4.1/go.mod h1:TNfzLD0ON7rHzMJeJkieUDPYmFC7Snx/y86RQel1bk4=
+github.com/prometheus/common v0.9.1/go.mod h1:yhUN8i9wzaXS3w1O07YhxHEBxD+W35wd8bs7vj7HSQ4=
 github.com/prometheus/common v0.10.0 h1:RyRA7RzGXQZiW+tGMr7sxa85G1z0yOpM1qq5c8lNawc=
 github.com/prometheus/common v0.10.0/go.mod h1:Tlit/dnDKsSWFlCLTWaA1cyBgKHSMdTB80sz/V91rCo=
 github.com/prometheus/procfs v0.0.0-20181005140218-185b4288413d/go.mod h1:c3At6R/oaqEKCNdg8wHV1ftS6bRYblBhIjjI8uT2IGk=
@@ -372,10 +390,12 @@ github.com/prometheus/procfs v0.0.0-20190425082905-87a4384529e0/go.mod h1:TjEm7z
 github.com/prometheus/procfs v0.0.0-20190507164030-5867b95ac084/go.mod h1:TjEm7ze935MbeOT/UhFTIMYKhuLP4wbCsTZCD3I8kEA=
 github.com/prometheus/procfs v0.0.2/go.mod h1:TjEm7ze935MbeOT/UhFTIMYKhuLP4wbCsTZCD3I8kEA=
 github.com/prometheus/procfs v0.0.3/go.mod h1:4A/X28fw3Fc593LaREMrKMqOKvUAntwMDaekg4FpcdQ=
+github.com/prometheus/procfs v0.0.11/go.mod h1:lV6e/gmhEcM9IjHGsFOCxxuZ+z1YqCvr4OA4YeYWdaU=
 github.com/prometheus/procfs v0.1.3 h1:F0+tqvhOksq22sc6iCHF5WGlWjdwj92p0udFh1VFBS8=
 github.com/prometheus/procfs v0.1.3/go.mod h1:lV6e/gmhEcM9IjHGsFOCxxuZ+z1YqCvr4OA4YeYWdaU=
 github.com/prometheus/tsdb v0.7.1/go.mod h1:qhTCs0VvXwvX/y3TZrWD7rabWM+ijKTux40TwIPHuXU=
 github.com/rcrowley/go-metrics v0.0.0-20181016184325-3113b8401b8a/go.mod h1:bCqnVzQkZxMG4s8nGwiZ5l3QUCyqpo9Y+/ZMZ9VjZe4=
+github.com/robfig/cron/v3 v3.0.1/go.mod h1:eQICP3HwyT7UooqI/z+Ov+PtYAWygg1TEWWzGIFLtro=
 github.com/rogpeppe/fastuuid v0.0.0-20150106093220-6724a57986af/go.mod h1:XWv6SoW27p1b0cqNHllgS5HIMJraePCO15w5zCzIWYg=
 github.com/rogpeppe/go-internal v1.3.0/go.mod h1:M8bDsm7K2OlrFYOpmOWEs/qY81heoFRclV5y23lUDJ4=
 github.com/russross/blackfriday/v2 v2.0.1/go.mod h1:+Rmxgy9KzJVeS9/2gXHxylqXiyQDYRxCVz55jmeOWTM=
@@ -422,6 +442,7 @@ github.com/stretchr/objx v0.2.0/go.mod h1:qt09Ya8vawLte6SNmTgCsAVtYtaKzEcn8ATUoH
 github.com/stretchr/testify v1.2.2/go.mod h1:a8OnRcib4nhh0OaRAV+Yts87kKdq0PP7pXfy6kDkUVs=
 github.com/stretchr/testify v1.3.0/go.mod h1:M5WIy9Dh21IEIfnGCwXGc5bZfKNJtfHm1UVUgZn+9EI=
 github.com/stretchr/testify v1.4.0/go.mod h1:j7eGeouHqKxXV5pUuKE4zz7dFj8WfuZ+81PSLYec5m4=
+github.com/stretchr/testify v1.5.1/go.mod h1:5W2xD1RspED5o8YsWQXVCued0rvSQ+mT+I5cxcmMvtA=
 github.com/stretchr/testify v1.6.1 h1:hDPOHmpOpP40lSULcqw7IrRb/u7w6RpDC9399XyoNd0=
 github.com/stretchr/testify v1.6.1/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/h/Wwjteg=
 github.com/subosito/gotenv v1.2.0 h1:Slr1R9HxAlEKefgq5jn9U+DnETlIUa6HfgEzj0g5d7s=
@@ -430,11 +451,18 @@ github.com/swaggo/files v0.0.0-20190704085106-630677cd5c14/go.mod h1:gxQT6pBGRuI
 github.com/swaggo/gin-swagger v1.2.0 h1:YskZXEiv51fjOMTsXrOetAjrMDfFaXD79PEoQBOe2W0=
 github.com/swaggo/gin-swagger v1.2.0/go.mod h1:qlH2+W7zXGZkczuL+r2nEBR2JTT+/lX05Nn6vPhc7OI=
 github.com/swaggo/swag v1.5.1/go.mod h1:1Bl9F/ZBpVWh22nY0zmYyASPO1lI/zIwRDrpZU+tv8Y=
+github.com/swaggo/swag v1.6.5/go.mod h1:Y7ZLSS0d0DdxhWGVhQdu+Bu1QhaF5k0RD7FKdiAykeY=
 github.com/swaggo/swag v1.6.7 h1:e8GC2xDllJZr3omJkm9YfmK0Y56+rMO3cg0JBKNz09s=
 github.com/swaggo/swag v1.6.7/go.mod h1:xDhTyuFIujYiN3DKWC/H/83xcfHp+UE/IzWWampG7Zc=
+github.com/t-tiger/gorm-bulk-insert v1.3.0 h1:9k7BaVEhw/3fsvh6GTOBwJ2RXk3asc5xs5m6hwozq20=
+github.com/t-tiger/gorm-bulk-insert v1.3.0/go.mod h1:ruDlk8xDl+8sX4bA7PQuYly9YEb3pbp1eP2LCyeRrFY=
 github.com/tmc/grpc-websocket-proxy v0.0.0-20190109142713-0ad062ec5ee5/go.mod h1:ncp9v5uamzpCO7NfCPTXjqaC+bZgJeR0sMTm6dMHP7U=
+github.com/trustwallet/blockatlas v1.1.3-0.20200518232337-7495506f457c/go.mod h1:5Ev/iawD12+qgmtMkGgjYCKmGLF724YaxYfEbfY8lKc=
+github.com/trustwallet/ens-coincodec v1.0.5/go.mod h1:hTZCauB3S2Gt9M3FiRNva+xpLJXFqLyODKFF+UvH1lw=
 github.com/trustwallet/ens-coincodec v1.0.6 h1:ZPvFDJwvv4s+XCnUM7pzub8wYHy37tyJBNtKvgx9FDs=
 github.com/trustwallet/ens-coincodec v1.0.6/go.mod h1:hTZCauB3S2Gt9M3FiRNva+xpLJXFqLyODKFF+UvH1lw=
+github.com/trustwallet/watchmarket v1.1.0 h1:tD+dlnKRONYr2ljsirM0Gx90ahzlfL1ceNSJlpTxINQ=
+github.com/trustwallet/watchmarket v1.1.0/go.mod h1:VKVAqqClA5Oc6jaJvNorvn9wRnp9BKV04jVbjTGx0Zw=
 github.com/ugorji/go v1.1.4/go.mod h1:uQMGLiO92mf5W77hV/PUCpI3pbzQx3CRekS0kk+RGrc=
 github.com/ugorji/go v1.1.5-pre/go.mod h1:FwP/aQVg39TXzItUBMwnWp9T9gPQnXw4Poh4/oBQZ/0=
 github.com/ugorji/go v1.1.7 h1:/68gy2h+1mWMrwZFeD1kQialdSzAb432dtpeJ42ovdo=
@@ -444,15 +472,18 @@ github.com/ugorji/go/codec v1.1.5-pre/go.mod h1:tULtS6Gy1AE1yCENaw4Vb//HLH5njI2t
 github.com/ugorji/go/codec v1.1.7 h1:2SvQaVZ1ouYrrKKwoSk2pzd4A9evlKJb9oTL+OaLUSs=
 github.com/ugorji/go/codec v1.1.7/go.mod h1:Ax+UKWsSmolVDwsd+7N3ZtXu+yMGCf907BLYF3GoBXY=
 github.com/urfave/cli v1.20.0/go.mod h1:70zkFmudgCuE/ngEzBv17Jvp/497gISqfk5gWijbERA=
+github.com/urfave/cli v1.22.2/go.mod h1:Gos4lmkARVdJ6EkW0WaNv/tZAAMe9V7XWyB60NtXRu0=
 github.com/urfave/cli/v2 v2.1.1/go.mod h1:SE9GqnLQmjVa0iPEY0f1w3ygNIYcIJ0OKPMoW2caLfQ=
 github.com/wealdtech/go-slip44 v1.0.0 h1:g0Wi5EufdD/ULZ4hcZKQ3TGlA6FZhGTCkmVzYmf+7Is=
 github.com/wealdtech/go-slip44 v1.0.0/go.mod h1://S3V9M6iN1cCflLXMjJYciRPakCDubG9YpWVyW1AvM=
 github.com/xiang90/probing v0.0.0-20190116061207-43a291ad63a2/go.mod h1:UETIi67q53MR2AWcXfiuqkDkRtnGDLqkBTpCHuJHxtU=
 github.com/yuin/goldmark v1.1.27/go.mod h1:3hX8gzYuyVAZsxl0MRgGTJEmQBFcNTphYh9decYSb74=
+github.com/yuin/gopher-lua v0.0.0-20191220021717-ab39c6098bdb/go.mod h1:gqRgreBUhTSL0GeU64rtZ3Uq3wtjOa/TB2YfrtkCbVQ=
 go.elastic.co/apm v1.8.0 h1:AWEKpHwRal0yCMd4K8Oxy1HAa7xid+xq1yy+XjgoVU0=
 go.elastic.co/apm v1.8.0/go.mod h1:tCw6CkOJgkWnzEthFN9HUP1uL3Gjc/Ur6m7gRPLaoH0=
 go.elastic.co/apm/module/apmgin v1.8.0 h1:QcXVtFhqAcSGqC/ywzqZBHdWzJG3wyrBxgvCvAZ98do=
 go.elastic.co/apm/module/apmgin v1.8.0/go.mod h1:heH2rXaOluBb+ucwlG3FBCGrx56qHqoxb1KI7uBLlAk=
+go.elastic.co/apm/module/apmgoredis v1.8.0/go.mod h1:J8mXOp8sOJG1/Vuz0gPB7+ovcQY6vYqpabuRsk9beBM=
 go.elastic.co/apm/module/apmgorm v1.8.0 h1:1vKuD8hsbJcChcUQTBb1N8Zy1tILR5pGTExLtijyhko=
 go.elastic.co/apm/module/apmgorm v1.8.0/go.mod h1:18ttEKPQr0cvRSeG95WMXoPUy2BT8Jp1EzhPU5NQYTY=
 go.elastic.co/apm/module/apmhttp v1.8.0 h1:5AJPefWJzWDLX/47XIDfaloGiYWkkOQEULvlrI6Ieaw=
@@ -556,6 +587,7 @@ golang.org/x/sys v0.0.0-20181107165924-66b7b1311ac8/go.mod h1:STP8DvDyc/dI5b8T5h
 golang.org/x/sys v0.0.0-20181116152217-5ac8a444bdc5/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
 golang.org/x/sys v0.0.0-20181122145206-62eef0e2fa9b/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
 golang.org/x/sys v0.0.0-20181228144115-9a3f9b0469bb/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
+golang.org/x/sys v0.0.0-20190204203706-41f3e6584952/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
 golang.org/x/sys v0.0.0-20190215142949-d0b11bdaac8a/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
 golang.org/x/sys v0.0.0-20190222072716-a9d3bda3a223/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
 golang.org/x/sys v0.0.0-20190312061237-fead79001313/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
@@ -573,6 +605,8 @@ golang.org/x/sys v0.0.0-20191204072324-ce4227a45e2e/go.mod h1:h1NjWce9XRLGQEsW7w
 golang.org/x/sys v0.0.0-20200106162015-b016eb3dc98e/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
 golang.org/x/sys v0.0.0-20200116001909-b77594299b42/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
 golang.org/x/sys v0.0.0-20200323222414-85ca7c5b95cd/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
+golang.org/x/sys v0.0.0-20200420163511-1957bb5e6d1f/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
+golang.org/x/sys v0.0.0-20200515095857-1151b9dac4a9/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
 golang.org/x/sys v0.0.0-20200615200032-f1bc736245b1 h1:ogLJMz+qpzav7lGMh10LMvAkM/fAoGlaiiHYiFYdm80=
 golang.org/x/sys v0.0.0-20200615200032-f1bc736245b1/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
 golang.org/x/text v0.3.0/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ=
diff --git a/mq/mq.go b/mq/mq.go
index e2c2a60..431ed8c 100644
--- a/mq/mq.go
+++ b/mq/mq.go
@@ -22,9 +22,10 @@ type (
 )
 
 const (
-	TxNotifications Queue = "txNotifications"
-	Subscriptions   Queue = "subscriptions"
-	RawTransactions Queue = "rawTransactions"
+	TxNotifications    Queue = "txNotifications"
+	Subscriptions      Queue = "subscriptions"
+	RawTransactions    Queue = "rawTransactions"
+	TokensRegistration Queue = "tokensRegistration"
 )
 
 func Init(uri string) (err error) {
diff --git a/pkg/address/address.go b/pkg/address/address.go
index dce3240..34a08f7 100644
--- a/pkg/address/address.go
+++ b/pkg/address/address.go
@@ -8,6 +8,7 @@ import (
 	"github.com/trustwallet/blockatlas/pkg/errors"
 	"github.com/trustwallet/blockatlas/pkg/logger"
 	"golang.org/x/crypto/sha3"
+	"strconv"
 	"strings"
 )
 
@@ -115,3 +116,20 @@ func FormatAddress(address string, coinID uint) string {
 		return address
 	}
 }
+
+func PrefixedAddress(coinID uint, address string) string {
+	return strconv.Itoa(int(coinID)) + "_" + address
+}
+
+func UnprefixedAddress(address string) (string, uint, bool) {
+	result := strings.Split(address, "_")
+	if len(result) != 2 {
+		return "", 0, false
+	}
+	id, err := strconv.Atoi(result[0])
+	if err != nil {
+		return "", 0, false
+	}
+	return result[1], uint(id), true
+
+}
diff --git a/pkg/blockatlas/tx.go b/pkg/blockatlas/tx.go
index e5768c3..02b83d8 100644
--- a/pkg/blockatlas/tx.go
+++ b/pkg/blockatlas/tx.go
@@ -269,6 +269,29 @@ func (t *Tx) GetAddresses() []string {
 	}
 }
 
+func (t *Tx) TokenID() (string, bool) {
+	var tokenID string
+	switch t.Meta.(type) {
+	case Transfer, *Transfer, CollectibleTransfer, *CollectibleTransfer, ContractCall, *ContractCall, MultiCurrencyTransfer, *MultiCurrencyTransfer:
+		return "", false
+	case NativeTokenTransfer:
+		tokenID = t.Meta.(NativeTokenTransfer).TokenID
+	case *NativeTokenTransfer:
+		tokenID = t.Meta.(*NativeTokenTransfer).TokenID
+	case TokenTransfer:
+		tokenID = t.Meta.(TokenTransfer).TokenID
+	case *TokenTransfer:
+		tokenID = t.Meta.(*TokenTransfer).TokenID
+	case AnyAction:
+		tokenID = t.Meta.(AnyAction).TokenID
+	case *AnyAction:
+		tokenID = t.Meta.(*AnyAction).TokenID
+	default:
+		return "", false
+	}
+	return tokenID, true
+}
+
 func (t *Tx) GetTransactionDirection(address string) Direction {
 	if t.Direction != "" {
 		return t.Direction
diff --git a/platform/tron/token.go b/platform/tron/token.go
index 8b15aa6..29a18d2 100644
--- a/platform/tron/token.go
+++ b/platform/tron/token.go
@@ -57,7 +57,7 @@ func (p *Platform) getTokens(ids []string) chan blockatlas.Token {
 			time.Sleep(time.Millisecond)
 			err := p.getTokensChannel(i, c)
 			if err != nil {
-				logger.Error("TRON getTokens token: " + i)
+				logger.Error("tron getTokens: " + i)
 			}
 		}(id, tkChan)
 	}
@@ -69,7 +69,6 @@ func (p *Platform) getTokens(ids []string) chan blockatlas.Token {
 func (p *Platform) getTokensChannel(id string, tkChan chan blockatlas.Token) error {
 	info, err := p.client.fetchTokenInfo(id)
 	if err != nil || len(info.Data) == 0 {
-		logger.Error(err, "fetchTokenInfo: invalid token")
 		return err
 	}
 	asset := NormalizeToken(info.Data[0])
diff --git a/services/observer/notifier/base.go b/services/notifier/base.go
similarity index 70%
rename from services/observer/notifier/base.go
rename to services/notifier/base.go
index 082c154..dc1d85e 100644
--- a/services/observer/notifier/base.go
+++ b/services/notifier/base.go
@@ -4,7 +4,9 @@ import (
 	"context"
 	"github.com/streadway/amqp"
 	"github.com/trustwallet/blockatlas/db"
+	"github.com/trustwallet/blockatlas/pkg/address"
 	"github.com/trustwallet/blockatlas/pkg/logger"
+	"strconv"
 
 	"go.elastic.co/apm"
 )
@@ -24,7 +26,7 @@ func RunNotifier(database *db.Instance, delivery amqp.Delivery) {
 		}
 	}()
 
-	txs, err := getTransactionsFromDelivery(delivery, ctx)
+	txs, err := GetTransactionsFromDelivery(delivery, ctx)
 	if err != nil {
 		logger.Error("failed to get transactions", err)
 	}
@@ -34,19 +36,26 @@ func RunNotifier(database *db.Instance, delivery amqp.Delivery) {
 		allAddresses = append(allAddresses, tx.GetAddresses()...)
 	}
 
-	addresses := toUniqueAddresses(allAddresses)
+	addresses := ToUniqueAddresses(allAddresses)
+	for i := range addresses {
+		addresses[i] = strconv.Itoa(int(txs[0].Coin)) + "_" + addresses[i]
+	}
 
 	if len(txs) < 1 {
 		return
 	}
-	subscriptionsDataList, err := database.GetSubscriptions(txs[0].Coin, addresses, ctx)
+	subscriptionsDataList, err := database.GetSubscriptionsForNotifications(addresses, ctx)
 	if err != nil || len(subscriptionsDataList) == 0 {
 		return
 	}
 
 	notifications := make([]TransactionNotification, 0)
 	for _, sub := range subscriptionsDataList {
-		notificationsForAddress := buildNotificationsByAddress(sub.Address, txs, ctx)
+		ua, _, ok := address.UnprefixedAddress(sub.Address.Address)
+		if !ok {
+			continue
+		}
+		notificationsForAddress := buildNotificationsByAddress(ua, txs, ctx)
 		notifications = append(notifications, notificationsForAddress...)
 	}
 
diff --git a/services/observer/notifier/delivery.go b/services/notifier/delivery.go
similarity index 89%
rename from services/observer/notifier/delivery.go
rename to services/notifier/delivery.go
index 9a1c3ca..cc5c5c1 100644
--- a/services/observer/notifier/delivery.go
+++ b/services/notifier/delivery.go
@@ -11,10 +11,10 @@ import (
 	"go.elastic.co/apm"
 )
 
-func getTransactionsFromDelivery(delivery amqp.Delivery, ctx context.Context) (blockatlas.Txs, error) {
+func GetTransactionsFromDelivery(delivery amqp.Delivery, ctx context.Context) (blockatlas.Txs, error) {
 	var txs blockatlas.Txs
 
-	span, _ := apm.StartSpan(ctx, "getTransactionsFromDelivery", "app")
+	span, _ := apm.StartSpan(ctx, "GetTransactionsFromDelivery", "app")
 	defer span.End()
 
 	if err := json.Unmarshal(delivery.Body, &txs); err != nil {
diff --git a/services/observer/notifier/models.go b/services/notifier/models.go
similarity index 95%
rename from services/observer/notifier/models.go
rename to services/notifier/models.go
index f33054c..9227ab5 100644
--- a/services/observer/notifier/models.go
+++ b/services/notifier/models.go
@@ -44,7 +44,7 @@ func buildNotificationsByAddress(address string, txs blockatlas.Txs, ctx context
 	return result
 }
 
-func toUniqueAddresses(addresses []string) []string {
+func ToUniqueAddresses(addresses []string) []string {
 	keys := make(map[string]bool)
 	var list []string
 	for _, entry := range addresses {
@@ -81,7 +81,7 @@ func findTransactionsByAddress(txs blockatlas.Txs, address string) []blockatlas.
 
 func containsAddress(tx blockatlas.Tx, address string) bool {
 	allAddresses := tx.GetAddresses()
-	txAddresses := toUniqueAddresses(allAddresses)
+	txAddresses := ToUniqueAddresses(allAddresses)
 	for _, a := range txAddresses {
 		if a == address {
 			return true
diff --git a/services/observer/notifier/models_test.go b/services/notifier/models_test.go
similarity index 100%
rename from services/observer/notifier/models_test.go
rename to services/notifier/models_test.go
diff --git a/services/observer/parser/parser.go b/services/parser/parser.go
similarity index 96%
rename from services/observer/parser/parser.go
rename to services/parser/parser.go
index 331d25f..0778519 100644
--- a/services/observer/parser/parser.go
+++ b/services/parser/parser.go
@@ -23,7 +23,7 @@ type (
 	Params struct {
 		Ctx                                       context.Context
 		Api                                       blockatlas.BlockAPI
-		Queue                                     mq.Queue
+		Queue                                     []mq.Queue
 		ParsingBlocksInterval, FetchBlocksTimeout time.Duration
 		BacklogCount                              int
 		MaxBacklogBlocks                          int64
@@ -296,10 +296,12 @@ func publish(params Params, txs blockatlas.Txs, ctx context.Context) {
 		logger.Error(err, logger.Params{"coin": params.Api.Coin().Handle})
 		return
 	}
-	err = params.Queue.Publish(body)
-	if err != nil {
-		logger.Error(err, logger.Params{"coin": params.Api.Coin().Handle})
-		return
+	for _, q := range params.Queue {
+		err = q.Publish(body)
+		if err != nil {
+			logger.Error(err, logger.Params{"coin": params.Api.Coin().Handle})
+			return
+		}
 	}
 }
 
@@ -318,12 +320,7 @@ func getBlockByNumberWithRetry(attempts int, sleep time.Duration, getBlockByNumb
 			sleep = sleep + jitter/2
 
 			logger.Info("retry GetBlockByNumber",
-				logger.Params{
-					"number":   n,
-					"attempts": attempts,
-					"sleep":    sleep.String(),
-					"symbol":   symbol,
-				},
+				logger.Params{"number": n, "attempts": attempts, "sleep": sleep.String(), "symbol": symbol},
 			)
 
 			time.Sleep(sleep)
diff --git a/services/subscriber/subscriber.go b/services/subscriber/subscriber.go
new file mode 100644
index 0000000..7d9a0aa
--- /dev/null
+++ b/services/subscriber/subscriber.go
@@ -0,0 +1,3 @@
+package subscriber
+
+type Subscriber string
diff --git a/services/subscriber/tokens.go b/services/subscriber/tokens.go
new file mode 100644
index 0000000..80ffe78
--- /dev/null
+++ b/services/subscriber/tokens.go
@@ -0,0 +1,37 @@
+package subscriber
+
+import (
+	"context"
+	"encoding/json"
+	"github.com/streadway/amqp"
+	"github.com/trustwallet/blockatlas/db"
+	"github.com/trustwallet/blockatlas/pkg/logger"
+	"go.elastic.co/apm"
+	"strconv"
+)
+
+const Tokens Subscriber = "tokens"
+
+func RunTokensSubscriber(database *db.Instance, delivery amqp.Delivery) {
+	tx := apm.DefaultTracer.StartTransaction("RunTokensSubscriber", "app")
+	defer tx.End()
+
+	ctx := apm.ContextWithTransaction(context.Background(), tx)
+
+	event := make(map[string][]string)
+	if err := json.Unmarshal(delivery.Body, &event); err != nil {
+		if err := delivery.Ack(false); err != nil {
+			logger.Fatal(err, err)
+		}
+	}
+
+	for address, assets := range event {
+		if err := database.AddAssociationsForAddress(address, assets, ctx); err != nil {
+			logger.Error("Failed to AddAssociationsForAddress: " + err.Error())
+		}
+	}
+	logger.Info("Subscribed " + strconv.Itoa(len(event)))
+	if err := delivery.Ack(false); err != nil {
+		logger.Fatal(err, err)
+	}
+}
diff --git a/services/observer/subscriber/subscriber.go b/services/subscriber/transactions.go
similarity index 66%
rename from services/observer/subscriber/subscriber.go
rename to services/subscriber/transactions.go
index d4ced17..05626ca 100644
--- a/services/observer/subscriber/subscriber.go
+++ b/services/subscriber/transactions.go
@@ -5,20 +5,21 @@ import (
 	"encoding/json"
 	"github.com/streadway/amqp"
 	"github.com/trustwallet/blockatlas/db"
-	"github.com/trustwallet/blockatlas/db/models"
 	"github.com/trustwallet/blockatlas/pkg/blockatlas"
 	"github.com/trustwallet/blockatlas/pkg/logger"
 	"go.elastic.co/apm"
+	"strconv"
 )
 
 const (
+	Notifications      Subscriber                       = "notifications"
 	AddSubscription    blockatlas.SubscriptionOperation = "AddSubscription"
 	DeleteSubscription blockatlas.SubscriptionOperation = "DeleteSubscription"
 	UpdateSubscription blockatlas.SubscriptionOperation = "UpdateSubscription"
 )
 
-func RunSubscriber(database *db.Instance, delivery amqp.Delivery) {
-	tx := apm.DefaultTracer.StartTransaction("RunSubscriber", "app")
+func RunTransactionsSubscriber(database *db.Instance, delivery amqp.Delivery) {
+	tx := apm.DefaultTracer.StartTransaction("RunTransactionsSubscriber", "app")
 	defer tx.End()
 
 	ctx := apm.ContextWithTransaction(context.Background(), tx)
@@ -35,13 +36,13 @@ func RunSubscriber(database *db.Instance, delivery amqp.Delivery) {
 
 	switch event.Operation {
 	case AddSubscription, UpdateSubscription:
-		err = database.AddSubscriptions(ToSubscriptionData(subscriptions), ctx)
+		err = database.AddSubscriptionsForNotifications(ToSubscriptionData(subscriptions), ctx)
 		if err != nil {
 			logger.Error(err, params)
 		}
 		logger.Info("Added", params)
 	case DeleteSubscription:
-		err := database.DeleteSubscriptions(ToSubscriptionData(subscriptions), ctx)
+		err := database.DeleteSubscriptionsForNotifications(ToSubscriptionData(subscriptions), ctx)
 		if err != nil {
 			logger.Error(err, params)
 		}
@@ -54,10 +55,12 @@ func RunSubscriber(database *db.Instance, delivery amqp.Delivery) {
 	}
 }
 
-func ToSubscriptionData(sub []blockatlas.Subscription) []models.Subscription {
-	data := make([]models.Subscription, 0, len(sub))
+func ToSubscriptionData(sub []blockatlas.Subscription) []string {
+	data := make([]string, 0, len(sub))
 	for _, s := range sub {
-		data = append(data, models.Subscription{Coin: s.Coin, Address: s.Address})
+		coinStr := strconv.FormatUint(uint64(s.Coin), 10)
+		address := coinStr + "_" + s.Address
+		data = append(data, address)
 	}
 	return data
 }
diff --git a/services/tokensearcher/api.go b/services/tokensearcher/api.go
new file mode 100644
index 0000000..cad07cd
--- /dev/null
+++ b/services/tokensearcher/api.go
@@ -0,0 +1,183 @@
+package tokensearcher
+
+import (
+	"context"
+	"encoding/json"
+	"github.com/trustwallet/blockatlas/db"
+	"github.com/trustwallet/blockatlas/mq"
+	"github.com/trustwallet/blockatlas/pkg/address"
+	"github.com/trustwallet/blockatlas/pkg/blockatlas"
+	"github.com/trustwallet/blockatlas/pkg/logger"
+	"sync"
+	"time"
+)
+
+type (
+	AddressesByCoin map[uint][]string
+	AssetsByAddress map[string][]string
+	Request         struct {
+		AddressesByCoin map[string][]string `json:"addresses"`
+		From            uint                `json:"from"`
+	}
+)
+
+type Instance struct {
+	database *db.Instance
+	apis     map[uint]blockatlas.TokensAPI
+	queue    mq.Queue
+}
+
+func Init(database *db.Instance, apis map[uint]blockatlas.TokensAPI, queue mq.Queue) Instance {
+	return Instance{database: database, apis: apis, queue: queue}
+}
+
+func (i Instance) HandleTokensRequest(request Request, ctx context.Context) (AssetsByAddress, error) {
+	addresses := getAddressesFromRequest(request)
+
+	subscribedAddresses, err := getSubscribedAddresses(i.database, addresses, ctx)
+	if err != nil {
+		return nil, err
+	}
+	unsubscribedAddresses := getUnsubscribedAddresses(subscribedAddresses, addresses)
+
+	assetsFromDB, err := i.database.GetAssetsMapByAddressesFromTime(
+		subscribedAddresses,
+		time.Unix(int64(request.From), 0),
+		ctx)
+	if err != nil {
+		return nil, err
+	}
+
+	assetsFromNodes := make(AssetsByAddress)
+	if len(unsubscribedAddresses) != 0 {
+		assetsFromNodes = getAssetsByAddressFromNodes(unsubscribedAddresses, i.apis)
+		err = publishNewAddressesToQueue(i.queue, assetsFromNodes)
+		if err != nil {
+			logger.Error(err)
+		}
+	}
+
+	return getAssetsToResponse(assetsFromDB, assetsFromNodes, addresses), nil
+}
+
+func getSubscribedAddresses(database *db.Instance, addresses []string, ctx context.Context) ([]string, error) {
+	subscribedAddressesModel, err := database.GetSubscribedAddressesForAssets(ctx, addresses)
+	if err != nil {
+		return nil, err
+	}
+
+	subscribedAddresses := make([]string, 0, len(subscribedAddressesModel))
+	for _, a := range subscribedAddressesModel {
+		subscribedAddresses = append(subscribedAddresses, a.Address)
+	}
+	return subscribedAddresses, nil
+}
+
+func getAddressesFromRequest(request Request) []string {
+	var addresses []string
+	for coinID, requestAddresses := range request.AddressesByCoin {
+		for _, a := range requestAddresses {
+			addresses = append(addresses, coinID+"_"+a)
+		}
+	}
+	return addresses
+}
+
+func getUnsubscribedAddresses(subscribed []string, all []string) AddressesByCoin {
+	addressesByCoin := make(AddressesByCoin)
+	subscribedMap := make(map[string]bool)
+	for _, a := range subscribed {
+		subscribedMap[a] = true
+	}
+	for _, a := range all {
+		_, ok := subscribedMap[a]
+		if !ok {
+			ua, coinID, ok := address.UnprefixedAddress(a)
+			if !ok {
+				continue
+			}
+			currentAddresses := addressesByCoin[coinID]
+			addressesByCoin[coinID] = append(currentAddresses, ua)
+		}
+	}
+	return addressesByCoin
+}
+
+func getAddressesToRegisterByCoin(assetsByAddresses AssetsByAddress, addresses []string) AddressesByCoin {
+	addressesByCoin := make(AddressesByCoin)
+	addressesFromRequestMap := make(map[string]bool)
+	for _, a := range addresses {
+		addressesFromRequestMap[a] = true
+	}
+	for _, a := range addresses {
+		_, ok := assetsByAddresses[a]
+		if !ok {
+			ua, coinID, ok := address.UnprefixedAddress(a)
+			if !ok {
+				continue
+			}
+			currentAddresses := addressesByCoin[coinID]
+			addressesByCoin[coinID] = append(currentAddresses, ua)
+		}
+	}
+	return addressesByCoin
+}
+
+func getAssetsByAddressFromNodes(addressesByCoin AddressesByCoin, apis map[uint]blockatlas.TokensAPI) AssetsByAddress {
+	a := NodesResponse{AssetsByAddress: make(AssetsByAddress)}
+	var wg sync.WaitGroup
+	for coinID, addresses := range addressesByCoin {
+		api, ok := apis[coinID]
+		if !ok {
+			continue
+		}
+		wg.Add(1)
+		go fetchAssetsByAddresses(api, addresses, &a, &wg)
+	}
+	wg.Wait()
+	return a.AssetsByAddress
+}
+
+func fetchAssetsByAddresses(tokenAPI blockatlas.TokensAPI, addresses []string, result *NodesResponse, wg *sync.WaitGroup) {
+	defer wg.Done()
+
+	var tWg sync.WaitGroup
+	tWg.Add(len(addresses))
+	for _, a := range addresses {
+		go func(address string, tWg *sync.WaitGroup) {
+			defer tWg.Done()
+			tokens, err := tokenAPI.GetTokenListByAddress(address)
+			if err != nil {
+				logger.Error("Chain: " + tokenAPI.Coin().Handle + " Address: " + address)
+				return
+			}
+			result.UpdateAssetsByAddress(tokens, int(tokenAPI.Coin().ID), address)
+		}(a, &tWg)
+	}
+	tWg.Wait()
+}
+
+func publishNewAddressesToQueue(queue mq.Queue, message AssetsByAddress) error {
+	body, err := json.Marshal(message)
+	if err != nil {
+		return err
+	}
+	return queue.Publish(body)
+}
+
+func getAssetsToResponse(dbAssetsMap, nodesAssetsMap AssetsByAddress, addresses []string) map[string][]string {
+	result := make(map[string][]string)
+	for _, address := range addresses {
+		dbAddresses, ok := dbAssetsMap[address]
+		if !ok {
+			nodesAssets, ok := nodesAssetsMap[address]
+			if !ok {
+				continue
+			}
+			result[address] = nodesAssets
+			continue
+		}
+		result[address] = dbAddresses
+	}
+	return result
+}
diff --git a/services/tokensearcher/association.go b/services/tokensearcher/association.go
new file mode 100644
index 0000000..948c146
--- /dev/null
+++ b/services/tokensearcher/association.go
@@ -0,0 +1,62 @@
+package tokensearcher
+
+import (
+	"github.com/trustwallet/blockatlas/db/models"
+	"github.com/trustwallet/blockatlas/pkg/blockatlas"
+	"github.com/trustwallet/watchmarket/pkg/watchmarket"
+	"strings"
+)
+
+func assetsMap(txs blockatlas.Txs, coinID string) map[string][]string {
+	result := make(map[string][]string)
+	prefix := coinID + "_"
+	for _, tx := range txs {
+		addresses := tx.GetAddresses()
+		tokenID, ok := tx.TokenID()
+		if !ok {
+			continue
+		}
+		assetID := watchmarket.BuildID(tx.Coin, tokenID)
+		for _, a := range addresses {
+			assetIDs := result[prefix+a]
+			result[prefix+a] = append(assetIDs, assetID)
+		}
+	}
+	return result
+}
+
+func associationsToAdd(oldAssociations map[string][]string, newAssociations map[string][]string) map[string][]string {
+	result := make(map[string][]string)
+	for oldAddresses, oldAssets := range oldAssociations {
+		for newAddresses, newAssets := range newAssociations {
+			if strings.EqualFold(oldAddresses, newAddresses) {
+				m := result[newAddresses]
+				result[newAddresses] = append(m, newAssociationsForAddress(oldAssets, newAssets)...)
+			}
+		}
+	}
+	return result
+}
+
+func newAssociationsForAddress(oldAssociations []string, newAssociations []string) []string {
+	var result []string
+	oldM := make(map[string]bool)
+	for _, o := range oldAssociations {
+		oldM[o] = true
+	}
+	for _, n := range newAssociations {
+		if ok := oldM[n]; !ok {
+			result = append(result, n)
+		}
+	}
+	return result
+}
+
+func fromModelToAssociation(associations []models.AddressToAssetAssociation) map[string][]string {
+	result := make(map[string][]string)
+	for _, a := range associations {
+		m := result[a.Address.Address]
+		result[a.Address.Address] = append(m, a.Asset.AssetID)
+	}
+	return result
+}
diff --git a/services/tokensearcher/models.go b/services/tokensearcher/models.go
new file mode 100644
index 0000000..0897e3b
--- /dev/null
+++ b/services/tokensearcher/models.go
@@ -0,0 +1,23 @@
+package tokensearcher
+
+import (
+	"github.com/trustwallet/blockatlas/pkg/address"
+	"github.com/trustwallet/blockatlas/pkg/blockatlas"
+	"github.com/trustwallet/watchmarket/pkg/watchmarket"
+	"sync"
+)
+
+type NodesResponse struct {
+	sync.Mutex
+	AssetsByAddress AssetsByAddress
+}
+
+func (nr *NodesResponse) UpdateAssetsByAddress(tokens blockatlas.TokenPage, coin int, a string) {
+	nr.Lock()
+	for _, t := range tokens {
+		key := address.PrefixedAddress(uint(coin), a)
+		r := nr.AssetsByAddress[key]
+		nr.AssetsByAddress[key] = append(r, watchmarket.BuildID(t.Coin, t.TokenID))
+	}
+	nr.Unlock()
+}
diff --git a/services/tokensearcher/tokensearcher.go b/services/tokensearcher/tokensearcher.go
new file mode 100644
index 0000000..29716ab
--- /dev/null
+++ b/services/tokensearcher/tokensearcher.go
@@ -0,0 +1,53 @@
+package tokensearcher
+
+import (
+	"context"
+	"github.com/streadway/amqp"
+	"github.com/trustwallet/blockatlas/db"
+	"github.com/trustwallet/blockatlas/pkg/logger"
+	"github.com/trustwallet/blockatlas/services/notifier"
+	"go.elastic.co/apm"
+	"strconv"
+)
+
+func Run(database *db.Instance, delivery amqp.Delivery) {
+	tx := apm.DefaultTracer.StartTransaction("RunNotifier", "app")
+	defer tx.End()
+	ctx := apm.ContextWithTransaction(context.Background(), tx)
+
+	txs, err := notifier.GetTransactionsFromDelivery(delivery, ctx)
+	if err != nil {
+		logger.Error("failed to get transactions", err)
+		if err := delivery.Ack(false); err != nil {
+			logger.Error(err)
+		}
+	}
+	if len(txs) == 0 {
+		return
+	}
+	coinID := strconv.Itoa(int(txs[0].Coin))
+	var addresses []string
+	for _, tx := range txs {
+		addresses = append(addresses, tx.GetAddresses()...)
+	}
+	for i := range addresses {
+		addresses[i] = coinID + "_" + addresses[i]
+	}
+
+	associationsFromTransactions, err := database.GetAssociationsByAddresses(notifier.ToUniqueAddresses(addresses), ctx)
+	if err != nil {
+		logger.Error(err)
+		return
+	}
+
+	associationsToAdd := associationsToAdd(fromModelToAssociation(associationsFromTransactions), assetsMap(txs, coinID))
+	err = database.UpdateAssociationsForExistingAddresses(associationsToAdd, ctx)
+	if err != nil {
+		logger.Error(err)
+		return
+	}
+
+	if err := delivery.Ack(false); err != nil {
+		logger.Error(err)
+	}
+}


Test patch:
diff --git a/pkg/address/address_test.go b/pkg/address/address_test.go
index f968523..e1c8cda 100644
--- a/pkg/address/address_test.go
+++ b/pkg/address/address_test.go
@@ -156,3 +156,15 @@ func TestFormatAddress(t *testing.T) {
 		}
 	})
 }
+
+func TestUnprefixedAddress(t *testing.T) {
+	address, id, ok := UnprefixedAddress("60_a")
+	assert.Equal(t, "a", address)
+	assert.Equal(t, uint(60), id)
+	assert.True(t, ok)
+}
+
+func TestPrefixedAddress(t *testing.T) {
+	address := PrefixedAddress(60, "a")
+	assert.Equal(t, "60_a", address)
+}
diff --git a/pkg/blockatlas/tx_test.go b/pkg/blockatlas/tx_test.go
index 1f1fc8d..5b7db6e 100644
--- a/pkg/blockatlas/tx_test.go
+++ b/pkg/blockatlas/tx_test.go
@@ -521,3 +521,47 @@ func TestTxs_SortByDate(t *testing.T) {
 	})
 	assert.True(t, isSorted)
 }
+
+func TestTx_TokenID(t *testing.T) {
+	tx1 := Tx{
+		Coin: 60,
+		From: "A",
+		To:   "B",
+		Meta: NativeTokenTransfer{
+			TokenID: "ABC",
+			From:    "A",
+			To:      "C",
+		},
+	}
+
+	tx2 := Tx{
+		Coin: 60,
+		From: "D",
+		To:   "V",
+		Meta: TokenTransfer{
+			TokenID: "EFG",
+			From:    "D",
+			To:      "F",
+		},
+	}
+
+	tx3 := Tx{
+		Coin: 60,
+		From: "Q",
+		To:   "L",
+		Meta: AnyAction{
+			TokenID: "HIJ",
+		},
+	}
+
+	token1, ok1 := tx1.TokenID()
+	assert.True(t, ok1)
+	assert.Equal(t, token1, "ABC")
+	token2, ok2 := tx2.TokenID()
+	assert.True(t, ok2)
+	assert.Equal(t, token2, "EFG")
+	token3, ok3 := tx3.TokenID()
+	assert.Equal(t, token3, "HIJ")
+	assert.True(t, ok3)
+
+}
diff --git a/services/observer/parser/parser_test.go b/services/parser/parser_test.go
similarity index 98%
rename from services/observer/parser/parser_test.go
rename to services/parser/parser_test.go
index e4a8e78..6b9eaee 100644
--- a/services/observer/parser/parser_test.go
+++ b/services/parser/parser_test.go
@@ -5,6 +5,7 @@ import (
 	"errors"
 	"github.com/stretchr/testify/assert"
 	"github.com/trustwallet/blockatlas/coin"
+	"github.com/trustwallet/blockatlas/mq"
 	"github.com/trustwallet/blockatlas/pkg/blockatlas"
 	"sync"
 	"testing"
@@ -45,7 +46,7 @@ func TestFetchBlocks(t *testing.T) {
 	params := Params{
 		Ctx:                   nil,
 		Api:                   getMockedBlockAPI(),
-		Queue:                 "",
+		Queue:                 []mq.Queue{""},
 		ParsingBlocksInterval: 0,
 		FetchBlocksTimeout:    0,
 		BacklogCount:          0,
diff --git a/services/observer/subscriber/subscriber_test.go b/services/subscriber/transactions_test.go
similarity index 59%
rename from services/observer/subscriber/subscriber_test.go
rename to services/subscriber/transactions_test.go
index bb75f7c..ab27df7 100644
--- a/services/observer/subscriber/subscriber_test.go
+++ b/services/subscriber/transactions_test.go
@@ -2,7 +2,6 @@ package subscriber
 
 import (
 	"github.com/stretchr/testify/assert"
-	"github.com/trustwallet/blockatlas/db/models"
 	"github.com/trustwallet/blockatlas/pkg/blockatlas"
 	"testing"
 )
@@ -17,17 +16,10 @@ func TestToSubscriptionData(t *testing.T) {
 		Address: "B",
 	}
 
-	expectedModel := models.Subscription{
-		Coin:    60,
-		Address: "A",
-	}
-	expectedModel1 := models.Subscription{
-		Coin:    60,
-		Address: "B",
-	}
-
+	expected := "60_A"
+	expected1 := "60_B"
 	res := ToSubscriptionData([]blockatlas.Subscription{sub, sub2})
 	assert.Equal(t, 2, len(res))
-	assert.Equal(t, expectedModel, res[0])
-	assert.Equal(t, expectedModel1, res[1])
+	assert.Equal(t, expected, res[0])
+	assert.Equal(t, expected1, res[1])
 }
diff --git a/services/tokensearcher/api_test.go b/services/tokensearcher/api_test.go
new file mode 100644
index 0000000..5524a3c
--- /dev/null
+++ b/services/tokensearcher/api_test.go
@@ -0,0 +1,93 @@
+package tokensearcher
+
+import (
+	"errors"
+	"github.com/stretchr/testify/assert"
+	"github.com/trustwallet/blockatlas/coin"
+	"github.com/trustwallet/blockatlas/pkg/blockatlas"
+	"testing"
+)
+
+func Test_getAddressesFromRequest(t *testing.T) {
+	request := Request{
+		AddressesByCoin: make(map[string][]string),
+		From:            0,
+	}
+	request.AddressesByCoin["60"] = []string{"1", "2", "3"}
+	r := getAddressesFromRequest(request)
+	assert.Equal(t, []string{"60_1", "60_2", "60_3"}, r)
+}
+
+func Test_getUnsubscribedAddresses(t *testing.T) {
+	r := getUnsubscribedAddresses([]string{"60_1"}, []string{"60_1", "714_2", "714_22", "118_3"})
+	assert.Equal(t, []string{"2", "22"}, r[714])
+	assert.Equal(t, []string{"3"}, r[118])
+}
+
+func Test_getAddressesToRegisterByCoin(t *testing.T) {
+	addressFromDB := make(map[string][]string)
+	addressFromDB["60_a"] = []string{"1", "2", "3"}
+	addressFromDB["714_b"] = []string{"1", "3"}
+
+	addressesFromRequest := []string{"60_a", "714_b", "118_c"}
+	result := getAddressesToRegisterByCoin(addressFromDB, addressesFromRequest)
+
+	c, ok := result[118]
+	assert.True(t, ok)
+	assert.Equal(t, []string{"c"}, c)
+}
+
+func Test_getAssetsToResponse(t *testing.T) {
+	addressFromDB := make(map[string][]string)
+	addressFromDB["60_a"] = []string{"1", "2", "3"}
+	addressFromDB["714_b"] = []string{"1", "3"}
+
+	addressFromNodes := make(map[string][]string)
+	addressFromNodes["118_c"] = []string{"1", "2", "3"}
+
+	addressesFromRequest := []string{"60_a", "714_b", "118_c"}
+
+	result := getAssetsToResponse(addressFromDB, addressFromNodes, addressesFromRequest)
+	assert.NotNil(t, result)
+
+	assert.Equal(t, []string{"1", "2", "3"}, result["60_a"])
+	assert.Equal(t, []string{"1", "3"}, result["714_b"])
+	assert.Equal(t, []string{"1", "2", "3"}, result["118_c"])
+}
+
+func Test_getAssetsForAddressesFromNodes(t *testing.T) {
+	apis := make(map[uint]blockatlas.TokensAPI)
+	mock0 := mockedTokenAPI{WantedToken: "ABC", WantedCoin: 0}
+	mock60 := mockedTokenAPI{WantedToken: "XYZ", WantedCoin: 60}
+	apis[0] = mock0
+	apis[60] = mock60
+
+	addresses := make(map[uint][]string)
+	addresses[0] = []string{"A", "B", "C"}
+	addresses[60] = []string{"X", "Y", "Z"}
+	result := getAssetsByAddressFromNodes(addresses, apis)
+	assert.NotNil(t, result)
+}
+
+type mockedTokenAPI struct {
+	WantedCoin  uint
+	WantedToken string
+}
+
+func (m mockedTokenAPI) GetTokenListByAddress(address string) (blockatlas.TokenPage, error) {
+	if address == "Y" {
+		return nil, errors.New("failed")
+	}
+	tk := blockatlas.Token{
+		Name:     "",
+		Symbol:   "",
+		Decimals: 0,
+		TokenID:  m.WantedToken,
+		Coin:     m.WantedCoin,
+		Type:     "",
+	}
+	return blockatlas.TokenPage{tk}, nil
+}
+func (m mockedTokenAPI) Coin() coin.Coin {
+	return coin.Coin{ID: m.WantedCoin}
+}
diff --git a/services/tokensearcher/association_test.go b/services/tokensearcher/association_test.go
new file mode 100644
index 0000000..b861b67
--- /dev/null
+++ b/services/tokensearcher/association_test.go
@@ -0,0 +1,100 @@
+package tokensearcher
+
+import (
+	"github.com/stretchr/testify/assert"
+	"github.com/trustwallet/blockatlas/db/models"
+	"github.com/trustwallet/blockatlas/pkg/blockatlas"
+	"testing"
+)
+
+func Test_assetsMap(t *testing.T) {
+	tx1 := blockatlas.Tx{
+		Coin: 60,
+		From: "A",
+		To:   "B",
+		Meta: blockatlas.NativeTokenTransfer{
+			TokenID: "ABC",
+			From:    "A",
+			To:      "C",
+		},
+	}
+
+	tx2 := blockatlas.Tx{
+		Coin: 60,
+		From: "D",
+		To:   "V",
+		Meta: blockatlas.TokenTransfer{
+			TokenID: "EFG",
+			From:    "D",
+			To:      "F",
+		},
+	}
+
+	tx3 := blockatlas.Tx{
+		Coin: 60,
+		From: "Q",
+		To:   "L",
+		Meta: blockatlas.AnyAction{
+			TokenID: "HIJ",
+		},
+	}
+
+	result := assetsMap(blockatlas.Txs{tx1, tx2, tx3}, "60")
+	assert.Equal(t, result["60_A"], []string{"c60_tABC"})
+	assert.Equal(t, result["60_C"], []string{"c60_tABC"})
+	assert.Equal(t, result["60_D"], []string{"c60_tEFG"})
+	assert.Equal(t, result["60_F"], []string{"c60_tEFG"})
+	assert.Equal(t, result["60_Q"], []string{"c60_tHIJ"})
+	assert.Equal(t, result["60_L"], []string{"c60_tHIJ"})
+}
+
+func Test_associationsToAdd(t *testing.T) {
+	o := make(map[string][]string)
+	n := make(map[string][]string)
+
+	o["A"] = []string{"1", "2", "3"}
+	o["B"] = []string{"3", "4", "5"}
+
+	n["A"] = []string{"1", "2", "5"}
+	n["B"] = []string{"3", "9", "8"}
+
+	result := associationsToAdd(o, n)
+
+	assert.Equal(t, result["A"], []string{"5"})
+	assert.Equal(t, result["B"], []string{"9", "8"})
+}
+
+func Test_newAssociationsForAddress(t *testing.T) {
+	o := []string{"1", "2", "3"}
+	n := []string{"1", "2", "3", "4", "5"}
+
+	result := newAssociationsForAddress(o, n)
+	assert.Equal(t, result, []string{"4", "5"})
+
+	o = []string{"1", "2", "3"}
+	n = []string{"1", "2", "3"}
+
+	result = newAssociationsForAddress(o, n)
+	assert.Equal(t, len(result), len([]string{}))
+
+	o = []string{"1", "2", "3"}
+	n = []string{"1", "2"}
+
+	result = newAssociationsForAddress(o, n)
+	assert.Equal(t, len(result), len([]string{}))
+}
+
+func Test_fromModelToAssociation(t *testing.T) {
+	a := []models.AddressToAssetAssociation{
+		{Address: models.Address{Address: "A"}, Asset: models.Asset{AssetID: "1"}},
+		{Address: models.Address{Address: "A"}, Asset: models.Asset{AssetID: "2"}},
+		{Address: models.Address{Address: "A"}, Asset: models.Asset{AssetID: "3"}},
+		{Address: models.Address{Address: "B"}, Asset: models.Asset{AssetID: "2"}},
+		{Address: models.Address{Address: "B"}, Asset: models.Asset{AssetID: "3"}},
+		{Address: models.Address{Address: "B"}, Asset: models.Asset{AssetID: "4"}},
+	}
+
+	result := fromModelToAssociation(a)
+	assert.Equal(t, result["A"], []string{"1", "2", "3"})
+	assert.Equal(t, result["B"], []string{"2", "3", "4"})
+}
diff --git a/tests/integration/db_test/subscriptions_test.go b/tests/integration/db_test/subscriptions_test.go
index fdf2fad..c35cbef 100644
--- a/tests/integration/db_test/subscriptions_test.go
+++ b/tests/integration/db_test/subscriptions_test.go
@@ -7,7 +7,7 @@ import (
 	"github.com/stretchr/testify/assert"
 	"github.com/trustwallet/blockatlas/db/models"
 	"github.com/trustwallet/blockatlas/pkg/blockatlas"
-	"github.com/trustwallet/blockatlas/services/observer/subscriber"
+	"github.com/trustwallet/blockatlas/services/subscriber"
 	"github.com/trustwallet/blockatlas/tests/integration/setup"
 	"testing"
 	"time"
@@ -15,155 +15,49 @@ import (
 
 func TestDb_AddSubscriptionsBulk(t *testing.T) {
 	setup.CleanupPgContainer(database.Gorm)
-	var subscriptions []models.Subscription
+	var subscriptions []string
 
 	for i := 0; i < 100; i++ {
-		subscriptions = append(subscriptions, models.Subscription{
-			Coin:    uint(i),
-			Address: "testAddrtestAddrtestAddrtestAddrtestAddrtestAddrtestAddrtestAddrtestAddrtestAddr",
-		})
+		subscriptions = append(subscriptions, "testAddrtestAddrtestAddrtestAddrtestAddrtestAddrtestAddrtestAddrtestAddrtestAddr")
 	}
 
-	assert.Nil(t, database.AddSubscriptions(subscriptions, context.Background()))
+	assert.Nil(t, database.AddSubscriptionsForNotifications(subscriptions, context.Background()))
 	for i := 0; i < 100; i++ {
-		s, err := database.GetSubscriptions(uint(i), []string{"testAddrtestAddrtestAddrtestAddrtestAddrtestAddrtestAddrtestAddrtestAddrtestAddr"}, context.Background())
+		s, err := database.GetSubscriptionsForNotifications([]string{"testAddrtestAddrtestAddrtestAddrtestAddrtestAddrtestAddrtestAddrtestAddrtestAddr"}, context.Background())
 		assert.Nil(t, err)
 		assert.NotNil(t, s)
 	}
-
 }
 
 func TestDb_AddSubscriptions(t *testing.T) {
 	setup.CleanupPgContainer(database.Gorm)
-	var subscriptions []models.Subscription
-
-	subscriptions = append(subscriptions, models.Subscription{
-		Coin:    60,
-		Address: "testAddr",
-	})
 
-	subscriptions = append(subscriptions, models.Subscription{
-		Coin:    61,
-		Address: "testAddr2",
-	})
-
-	subscriptions = append(subscriptions, models.Subscription{
-		Coin:    62,
-		Address: "testAddr3",
-	})
-
-	assert.Nil(t, database.AddSubscriptions(subscriptions, context.Background()))
+	assert.Nil(t, database.AddSubscriptionsForNotifications([]string{"60_testAddr", "60_testAddr2", "60_testAddr3"}, context.Background()))
 
-	subs, err := database.GetSubscriptions(60, []string{"testAddr"}, context.Background())
+	subs, err := database.GetSubscriptionsForNotifications([]string{"60_testAddr"}, context.Background())
 	assert.Nil(t, err)
 	assert.NotNil(t, subs)
 	assert.Equal(t, 1, len(subs))
-	assert.Equal(t, subscriptions[0].Coin, subs[0].Coin)
-	assert.Equal(t, subscriptions[0].Address, subs[0].Address)
+	assert.Equal(t, "60_testAddr", subs[0].Address.Address)
 
-	subs, err = database.GetSubscriptions(61, []string{"testAddr2"}, context.Background())
+	subs, err = database.GetSubscriptionsForNotifications([]string{"60_testAddr2"}, context.Background())
 	assert.Nil(t, err)
 	assert.NotNil(t, subs)
 	assert.Equal(t, 1, len(subs))
-	assert.Equal(t, subscriptions[1].Coin, subs[0].Coin)
-	assert.Equal(t, subscriptions[1].Address, subs[0].Address)
+	assert.Equal(t, "60_testAddr2", subs[0].Address.Address)
 
-	subs, err = database.GetSubscriptions(62, []string{"testAddr3"}, context.Background())
+	subs, err = database.GetSubscriptionsForNotifications([]string{"60_testAddr3"}, context.Background())
 	assert.Nil(t, err)
 	assert.NotNil(t, subs)
 	assert.Equal(t, 1, len(subs))
-	assert.Equal(t, subscriptions[2].Coin, subs[0].Coin)
-	assert.Equal(t, subscriptions[2].Address, subs[0].Address)
-}
-
-func TestDb_AddSubscriptionsWithRewrite(t *testing.T) {
-	setup.CleanupPgContainer(database.Gorm)
-
-	var subscriptions []models.Subscription
-	subscriptions = append(subscriptions, models.Subscription{
-		Coin:    60,
-		Address: "testAddr",
-	})
-
-	subscriptions = append(subscriptions, models.Subscription{
-		Coin:    714,
-		Address: "testAddr",
-	})
-
-	subscriptions = append(subscriptions, models.Subscription{
-		Coin:    144,
-		Address: "testAddr",
-	})
-
-	assert.Nil(t, database.AddSubscriptions(subscriptions, context.Background()))
-
-	subs60, err := database.GetSubscriptions(60, []string{"testAddr"}, context.Background())
-	assert.Nil(t, err)
-	assert.NotNil(t, subs60)
-	assert.Equal(t, 1, len(subs60))
-	assert.Equal(t, subscriptions[0].Coin, subs60[0].Coin)
-	assert.Equal(t, subscriptions[0].Address, subs60[0].Address)
-
-	subs714, err := database.GetSubscriptions(714, []string{"testAddr"}, context.Background())
-	assert.Nil(t, err)
-	assert.NotNil(t, subs714)
-	assert.Equal(t, 1, len(subs714))
-	assert.Equal(t, subscriptions[1].Coin, subs714[0].Coin)
-	assert.Equal(t, subscriptions[1].Address, subs714[0].Address)
-
-	subs144, err := database.GetSubscriptions(144, []string{"testAddr"}, context.Background())
-	assert.Nil(t, err)
-	assert.NotNil(t, subs144)
-	assert.Equal(t, 1, len(subs144))
-	assert.Equal(t, subscriptions[2].Coin, subs144[0].Coin)
-	assert.Equal(t, subscriptions[2].Address, subs144[0].Address)
-
-	subscriptions = append(subscriptions, models.Subscription{
-		Coin:    60,
-		Address: "testAddr2",
-	})
-
-	subscriptions = append(subscriptions, models.Subscription{
-		Coin:    714,
-		Address: "testAddr2",
-	})
-
-	subscriptions = append(subscriptions, models.Subscription{
-		Coin:    144,
-		Address: "testAddr2",
-	})
-
-	assert.Nil(t, database.AddSubscriptions(subscriptions, context.Background()))
-
-	subs2N60, err := database.GetSubscriptions(60, []string{"testAddr2"}, context.Background())
-	assert.Nil(t, err)
-	assert.Nil(t, err)
-	assert.NotNil(t, subs2N60)
-	assert.Equal(t, 1, len(subs2N60))
-	assert.Equal(t, subscriptions[3].Coin, subs2N60[0].Coin)
-	assert.Equal(t, subscriptions[3].Address, subs2N60[0].Address)
-
-	subs2N714, err := database.GetSubscriptions(714, []string{"testAddr2"}, context.Background())
-	assert.Nil(t, err)
-	assert.Nil(t, err)
-	assert.NotNil(t, subs2N714)
-	assert.Equal(t, 1, len(subs2N714))
-	assert.Equal(t, subscriptions[4].Coin, subs2N714[0].Coin)
-	assert.Equal(t, subscriptions[4].Address, subs2N714[0].Address)
-
-	subs2N114, err := database.GetSubscriptions(144, []string{"testAddr2"}, context.Background())
-	assert.Nil(t, err)
-	assert.Nil(t, err)
-	assert.NotNil(t, subs2N114)
-	assert.Equal(t, 1, len(subs2N114))
-	assert.Equal(t, subscriptions[5].Coin, subs2N114[0].Coin)
-	assert.Equal(t, subscriptions[5].Address, subs2N114[0].Address)
+	assert.Equal(t, "60_testAddr3", subs[0].Address.Address)
 }
 
 func TestDb_FindSubscriptions(t *testing.T) {
 	setup.CleanupPgContainer(database.Gorm)
 
 	var subscriptionsA []blockatlas.Subscription
+
 	subscriptionsA = append(subscriptionsA, blockatlas.Subscription{
 		Coin:    60,
 		Address: "etherAddress",
@@ -189,224 +83,151 @@ func TestDb_FindSubscriptions(t *testing.T) {
 		Address: "ETCAddress",
 	})
 
-	assert.Nil(t, database.AddSubscriptions(subscriber.ToSubscriptionData(subscriptionsA), context.Background()))
+	assert.Nil(t, database.AddSubscriptionsForNotifications(subscriber.ToSubscriptionData(subscriptionsA), context.Background()))
 
 	var subscriptionsB []blockatlas.Subscription
 
 	for _, sub := range subscriptionsA {
 		subscriptionsB = append(subscriptionsB, sub)
 	}
-	assert.Nil(t, database.AddSubscriptions(subscriber.ToSubscriptionData(subscriptionsB), context.Background()))
+	assert.Nil(t, database.AddSubscriptionsForNotifications(subscriber.ToSubscriptionData(subscriptionsB), context.Background()))
 
-	returnedSubs, err := database.GetSubscriptions(60, []string{"etherAddress"}, context.Background())
+	returnedSubs, err := database.GetSubscriptionsForNotifications([]string{"60_etherAddress"}, context.Background())
 	assert.Nil(t, err)
 	assert.Equal(t, 1, len(returnedSubs))
 
-	returnedSubs, err = database.GetSubscriptions(714, []string{"binanceAddress"}, context.Background())
+	returnedSubs, err = database.GetSubscriptionsForNotifications([]string{"714_binanceAddress"}, context.Background())
 	assert.Nil(t, err)
 	assert.Equal(t, 1, len(returnedSubs))
 
-	returnedSubs, err = database.GetSubscriptions(144, []string{"XLMAddress"}, context.Background())
+	returnedSubs, err = database.GetSubscriptionsForNotifications([]string{"144_XLMAddress"}, context.Background())
 	assert.Nil(t, err)
 	assert.Equal(t, 1, len(returnedSubs))
 
-	returnedSubs, err = database.GetSubscriptions(148, []string{"AtomAddress"}, context.Background())
+	returnedSubs, err = database.GetSubscriptionsForNotifications([]string{"148_AtomAddress"}, context.Background())
 	assert.Nil(t, err)
 	assert.Equal(t, 1, len(returnedSubs))
 
-	returnedSubs, err = database.GetSubscriptions(61, []string{"ETCAddress"}, context.Background())
+	returnedSubs, err = database.GetSubscriptionsForNotifications([]string{"61_ETCAddress"}, context.Background())
 	assert.Nil(t, err)
 	assert.Equal(t, 1, len(returnedSubs))
 }
 
 func TestDb_DeleteSubscriptions(t *testing.T) {
 	setup.CleanupPgContainer(database.Gorm)
-	var subscriptions []models.Subscription
-
-	subscriptions = append(subscriptions, models.Subscription{
-		Coin:    60,
-		Address: "testAddr",
-	})
-
-	subscriptions = append(subscriptions, models.Subscription{
-		Coin:    714,
-		Address: "testAddr2",
-	})
-
-	subscriptions = append(subscriptions, models.Subscription{
-		Coin:    144,
-		Address: "testAddr3",
-	})
+	subscriptions := []string{
+		"60_testAddr",
+		"714_testAddr2",
+		"144_testAddr3",
+	}
 
-	assert.Nil(t, database.AddSubscriptions(subscriptions, context.Background()))
+	assert.Nil(t, database.AddSubscriptionsForNotifications(subscriptions, context.Background()))
 
-	subs60, err := database.GetSubscriptions(60, []string{"testAddr"}, context.Background())
+	subs60, err := database.GetSubscriptionsForNotifications([]string{"60_testAddr"}, context.Background())
 	assert.Nil(t, err)
 	assert.NotNil(t, subs60)
 	assert.Equal(t, 1, len(subs60))
-	assert.Equal(t, subscriptions[0].Coin, subs60[0].Coin)
-	assert.Equal(t, subscriptions[0].Address, subs60[0].Address)
 
-	subs714, err := database.GetSubscriptions(714, []string{"testAddr2"}, context.Background())
+	subs714, err := database.GetSubscriptionsForNotifications([]string{"714_testAddr2"}, context.Background())
 	assert.Nil(t, err)
 	assert.NotNil(t, subs714)
 	assert.Equal(t, 1, len(subs714))
-	assert.Equal(t, subscriptions[1].Coin, subs714[0].Coin)
-	assert.Equal(t, subscriptions[1].Address, subs714[0].Address)
 
-	subs144, err := database.GetSubscriptions(144, []string{"testAddr3"}, context.Background())
+	subs144, err := database.GetSubscriptionsForNotifications([]string{"144_testAddr3"}, context.Background())
 	assert.Nil(t, err)
 	assert.NotNil(t, subs144)
 	assert.Equal(t, 1, len(subs144))
-	assert.Equal(t, subscriptions[2].Coin, subs144[0].Coin)
-	assert.Equal(t, subscriptions[2].Address, subs144[0].Address)
-
-	subsToDel := []models.Subscription{subscriptions[0]}
 
-	assert.Nil(t, database.DeleteSubscriptions(subsToDel, context.Background()))
+	assert.Nil(t, database.DeleteSubscriptionsForNotifications([]string{subscriptions[0]}, context.Background()))
 
-	subs714N2, err := database.GetSubscriptions(714, []string{"testAddr2"}, context.Background())
+	subs714N2, err := database.GetSubscriptionsForNotifications([]string{"714_testAddr2"}, context.Background())
 	assert.Nil(t, err)
 	assert.NotNil(t, subs714N2)
 	assert.Equal(t, 1, len(subs714N2))
-	assert.Equal(t, subscriptions[1].Coin, subs714N2[0].Coin)
-	assert.Equal(t, subscriptions[1].Address, subs714N2[0].Address)
 
-	subs144N2, err := database.GetSubscriptions(144, []string{"testAddr3"}, context.Background())
+	subs144N2, err := database.GetSubscriptionsForNotifications([]string{"144_testAddr3"}, context.Background())
 	assert.Nil(t, err)
 	assert.NotNil(t, subs144N2)
 	assert.Equal(t, 1, len(subs144N2))
-	assert.Equal(t, subscriptions[2].Coin, subs144N2[0].Coin)
-	assert.Equal(t, subscriptions[2].Address, subs144N2[0].Address)
 
-	subs60N2, err := database.GetSubscriptions(60, []string{"testAddr"}, context.Background())
+	subs60N2, err := database.GetSubscriptionsForNotifications([]string{"60_testAddr"}, context.Background())
 	assert.Nil(t, err)
 	assert.Len(t, subs60N2, 0)
 }
 
-func TestDeleteAll(t *testing.T) {
-	setup.CleanupPgContainer(database.Gorm)
-	var subscriptions []models.Subscription
-
-	subscriptions = append(subscriptions, models.Subscription{
-		Coin:    60,
-		Address: "testAddr",
-	})
-
-	subscriptions = append(subscriptions, models.Subscription{
-		Coin:    714,
-		Address: "testAddr2",
-	})
-
-	subscriptions = append(subscriptions, models.Subscription{
-		Coin:    144,
-		Address: "testAddr3",
-	})
-	assert.Nil(t, database.AddSubscriptions(subscriptions, context.Background()))
-
-	subs60, err := database.GetSubscriptions(60, []string{"testAddr"}, context.Background())
-	assert.Nil(t, err)
-	assert.Len(t, subs60, 1)
-
-	subs714, err := database.GetSubscriptions(714, []string{"testAddr2"}, context.Background())
-	assert.Nil(t, err)
-	assert.Len(t, subs714, 1)
-
-	subs144, err := database.GetSubscriptions(144, []string{"testAddr3"}, context.Background())
-	assert.Nil(t, err)
-	assert.Len(t, subs144, 1)
-}
-
 func TestDb_DuplicateEntries(t *testing.T) {
 	setup.CleanupPgContainer(database.Gorm)
-	var subscriptions []models.Subscription
+	var subscriptions []blockatlas.Subscription
 
 	for i := 0; i < 10; i++ {
-		subscriptions = append(subscriptions, models.Subscription{
+		subscriptions = append(subscriptions, blockatlas.Subscription{
 			Coin:    60,
 			Address: "testAddr",
 		})
 	}
 
-	assert.Nil(t, database.AddSubscriptions(subscriptions, context.Background()))
+	assert.Nil(t, database.AddSubscriptionsForNotifications(subscriber.ToSubscriptionData(subscriptions), context.Background()))
 
-	subs, err := database.GetSubscriptions(60, []string{"testAddr"}, context.Background())
+	subs, err := database.GetSubscriptionsForNotifications([]string{"60_testAddr"}, context.Background())
 	assert.Nil(t, err)
 	assert.NotNil(t, subs)
 	assert.Equal(t, 1, len(subs))
-	assert.True(t, containSub(subscriptions[0], subs))
 }
 
 func TestDb_CreateDeleteCreate(t *testing.T) {
 	setup.CleanupPgContainer(database.Gorm)
-	var subscriptions []models.Subscription
-	subscriptions = append(subscriptions, models.Subscription{
+	var subscriptions []blockatlas.Subscription
+	subscriptions = append(subscriptions, blockatlas.Subscription{
 		Coin:    60,
 		Address: "testAddr",
 	})
 
-	assert.Nil(t, database.AddSubscriptions(subscriptions, context.Background()))
-	subs, err := database.GetSubscriptions(60, []string{"testAddr"}, context.Background())
+	assert.Nil(t, database.AddSubscriptionsForNotifications(subscriber.ToSubscriptionData(subscriptions), context.Background()))
+	subs, err := database.GetSubscriptionsForNotifications([]string{"60_testAddr"}, context.Background())
 	assert.Nil(t, err)
 	assert.Equal(t, 1, len(subs))
 
-	time.Sleep(time.Second)
+	assert.Nil(t, database.DeleteSubscriptionsForNotifications([]string{"60_testAddr"}, context.Background()))
 
-	assert.Nil(t, database.DeleteSubscriptions(subs, context.Background()))
-
-	subs2, err := database.GetSubscriptions(60, []string{"testAddr"}, context.Background())
+	subs2, err := database.GetSubscriptionsForNotifications([]string{"60_testAddr"}, context.Background())
 	assert.Nil(t, err)
 	assert.Equal(t, 0, len(subs2))
 
-	assert.Nil(t, database.AddSubscriptions(subscriptions, context.Background()))
+	assert.Nil(t, database.AddSubscriptionsForNotifications(subscriber.ToSubscriptionData(subscriptions), context.Background()))
 
-	subs3, err := database.GetSubscriptions(60, []string{"testAddr"}, context.Background())
+	subs3, err := database.GetSubscriptionsForNotifications([]string{"60_testAddr"}, context.Background())
 	assert.Nil(t, err)
 	assert.Equal(t, 1, len(subs3))
 }
 
 func TestDb_UpdatedAt(t *testing.T) {
 	setup.CleanupPgContainer(database.Gorm)
-	var subscriptions []models.Subscription
-	subscriptions = append(subscriptions, models.Subscription{
+	var subscriptions []blockatlas.Subscription
+	subscriptions = append(subscriptions, blockatlas.Subscription{
 		Coin:    60,
 		Address: "testAddr",
 	})
 
-	assert.Nil(t, database.AddSubscriptions(subscriptions, context.Background()))
-	subs, err := database.GetSubscriptions(60, []string{"testAddr"}, context.Background())
+	assert.Nil(t, database.AddSubscriptionsForNotifications(subscriber.ToSubscriptionData(subscriptions), context.Background()))
+	subs, err := database.GetSubscriptionsForNotifications([]string{"60_testAddr"}, context.Background())
 	assert.Nil(t, err)
 	assert.Equal(t, 1, len(subs))
 
-	time.Sleep(time.Second)
+	var existingSub models.NotificationSubscription
+	var existingAddr models.Address
 
-	var existingSub models.Subscription
-	assert.False(t, database.Gorm.Where(models.Subscription{Address: "testAddr"}).First(&existingSub).RecordNotFound())
-	assert.Greater(t, time.Now().Unix(), existingSub.CreatedAt.Unix())
-	assert.Greater(t, existingSub.CreatedAt.Unix(), time.Now().Unix()-120)
+	assert.False(t, database.Gorm.Where("address = ?", "60_testAddr").First(&existingAddr).RecordNotFound())
+	assert.False(t, database.Gorm.Where("address_id = ?", existingAddr.ID).First(&existingSub).RecordNotFound())
+	assert.Greater(t, existingSub.UpdatedAt.Unix(), time.Now().Unix()-120)
 
-	subscriptions = append(subscriptions, models.Subscription{
+	subscriptions = append(subscriptions, blockatlas.Subscription{
 		Coin:    714,
 		Address: "newtestAddr",
 	})
 
-	assert.Nil(t, database.AddSubscriptions(subscriptions, context.Background()))
-
-	time.Sleep(time.Second)
-
-	var existingSub2 models.Subscription
-	assert.False(t, database.Gorm.Where(models.Subscription{Address: "testAddr"}).First(&existingSub2).RecordNotFound())
+	assert.Nil(t, database.AddSubscriptionsForNotifications(subscriber.ToSubscriptionData(subscriptions), context.Background()))
 
-	assert.Greater(t, time.Now().Unix(), existingSub2.CreatedAt.Unix())
-	assert.Greater(t, existingSub2.CreatedAt.Unix(), time.Now().Unix()-120)
-	assert.GreaterOrEqual(t, existingSub2.CreatedAt.Unix(), existingSub.CreatedAt.Unix())
-}
-
-func containSub(sub models.Subscription, list []models.Subscription) bool {
-	for _, s := range list {
-		if sub.Address == s.Address && sub.Coin == s.Coin {
-			return true
-		}
-	}
-	return false
+	var existingSub2 models.NotificationSubscription
+	assert.False(t, database.Gorm.Where("address_id = ?", existingSub.ID).First(&existingSub2).RecordNotFound())
+	assert.Greater(t, existingSub2.UpdatedAt.Unix(), time.Now().Unix()-120)
 }
diff --git a/tests/integration/db_test/tokenassociations_test.go b/tests/integration/db_test/tokenassociations_test.go
new file mode 100644
index 0000000..39bd5c4
--- /dev/null
+++ b/tests/integration/db_test/tokenassociations_test.go
@@ -0,0 +1,179 @@
+// +build integration
+
+package db_test
+
+import (
+	"context"
+	"github.com/stretchr/testify/assert"
+	"github.com/trustwallet/blockatlas/tests/integration/setup"
+	"sort"
+	"testing"
+	"time"
+)
+
+func Test_GetAssetsMapByAddresses(t *testing.T) {
+	setup.CleanupPgContainer(database.Gorm)
+
+	assets := []string{"aa", "bbb", "cccc"}
+
+	err := database.AddAssociationsForAddress("a", assets, context.Background())
+	assert.Nil(t, err)
+
+	err = database.AddAssociationsForAddress("b", nil, context.Background())
+	assert.Nil(t, err)
+
+	m, err := database.GetAssetsMapByAddresses([]string{"a", "b"}, context.Background())
+	assert.Nil(t, err)
+	wantedMap := make(map[string][]string)
+	wantedMap["a"] = assets
+	assert.Equal(t, wantedMap, m)
+}
+
+func Test_GetAssetsMapByAddressesFromTime(t *testing.T) {
+	setup.CleanupPgContainer(database.Gorm)
+
+	assets := []string{"aa", "bbb", "cccc"}
+
+	err := database.AddAssociationsForAddress("a", assets, context.Background())
+	assert.Nil(t, err)
+
+	err = database.AddAssociationsForAddress("b", nil, context.Background())
+	assert.Nil(t, err)
+	tm := time.Now().Unix() - 100
+	m, err := database.GetAssetsMapByAddressesFromTime([]string{"a", "b"}, time.Unix(tm, 0), context.Background())
+	assert.Nil(t, err)
+	wantedMap := make(map[string][]string)
+	wantedMap["a"] = assets
+	assert.Equal(t, wantedMap, m)
+
+	m, err = database.GetAssetsMapByAddressesFromTime([]string{"a", "b"}, time.Unix(tm+101, 0), context.Background())
+	assert.Nil(t, err)
+	assert.Equal(t, 0, len(m))
+}
+
+func Test_GetSubscribedAddressesForAssets(t *testing.T) {
+	setup.CleanupPgContainer(database.Gorm)
+
+	assets := []string{"aa", "bbb", "cccc"}
+
+	err := database.AddAssociationsForAddress("a", assets, context.Background())
+	assert.Nil(t, err)
+
+	err = database.AddAssociationsForAddress("b", nil, context.Background())
+	assert.Nil(t, err)
+
+	m, err := database.GetSubscribedAddressesForAssets(context.Background(), []string{"a", "b"})
+	assert.Nil(t, err)
+	assert.Equal(t, 2, len(m))
+}
+
+func Test_AddNewAssociationForAddress(t *testing.T) {
+	setup.CleanupPgContainer(database.Gorm)
+	assets := []string{"aa", "bbb", "cccc"}
+
+	err := database.AddAssociationsForAddress("a", assets, context.Background())
+	assert.Nil(t, err)
+
+	associations, err := database.GetAssociationsByAddresses([]string{"a"}, context.Background())
+	assert.Nil(t, err)
+
+	var assetIDsFromDB []string
+	for _, a := range associations {
+		assetIDsFromDB = append(assetIDsFromDB, a.Asset.AssetID)
+	}
+
+	sort.Slice(assets, func(i, j int) bool {
+		return len(assets[i]) > len(assets[j])
+	})
+
+	sort.Slice(assetIDsFromDB, func(i, j int) bool {
+		return len(assetIDsFromDB[i]) > len(assetIDsFromDB[j])
+	})
+
+	assert.Equal(t, assetIDsFromDB, assets)
+
+	err = database.AddAssociationsForAddress("b", nil, context.Background())
+	assert.Nil(t, err)
+
+	associations2, err := database.GetAssociationsByAddresses([]string{"b"}, context.Background())
+	assert.Nil(t, err)
+	assert.NotNil(t, associations2)
+}
+
+func Test_UpdateAssociationsForExistingAddresses(t *testing.T) {
+	setup.CleanupPgContainer(database.Gorm)
+	assets := []string{"f"}
+
+	err := database.AddAssociationsForAddress("A", assets, context.Background())
+	assert.Nil(t, err)
+
+	err = database.AddAssociationsForAddress("B", assets, context.Background())
+	assert.Nil(t, err)
+
+	assetsForA := []string{"aa", "bbb", "cccc"}
+	assetsForB := []string{"as", "bbb", "cccc"}
+
+	updateMap := make(map[string][]string)
+	updateMap["A"] = assetsForA
+	updateMap["B"] = assetsForB
+
+	err = database.UpdateAssociationsForExistingAddresses(updateMap, context.Background())
+	assert.Nil(t, err)
+
+	associationsA, err := database.GetAssociationsByAddresses([]string{"A"}, context.Background())
+	assert.Nil(t, err)
+
+	var assetIDsFromDBA []string
+	for _, a := range associationsA {
+		assetIDsFromDBA = append(assetIDsFromDBA, a.Asset.AssetID)
+	}
+	assetsA := []string{"aa", "bbb", "cccc", "f"}
+
+	sort.Slice(assetsA, func(i, j int) bool {
+		return len(assetsA[i]) > len(assetsA[j])
+	})
+
+	sort.Slice(assetIDsFromDBA, func(i, j int) bool {
+		return len(assetIDsFromDBA[i]) > len(assetIDsFromDBA[j])
+	})
+
+	assert.Equal(t, assetIDsFromDBA, assetsA)
+
+	associationsB, err := database.GetAssociationsByAddresses([]string{"B"}, context.Background())
+	assert.Nil(t, err)
+
+	var assetIDsFromDBB []string
+	for _, a := range associationsB {
+		assetIDsFromDBB = append(assetIDsFromDBB, a.Asset.AssetID)
+	}
+	assetsB := []string{"as", "bbb", "cccc", "f"}
+
+	sort.Slice(assetsB, func(i, j int) bool {
+		return len(assetsB[i]) > len(assetsB[j])
+	})
+
+	sort.Slice(assetIDsFromDBB, func(i, j int) bool {
+		return len(assetIDsFromDBB[i]) > len(assetIDsFromDBB[j])
+	})
+
+	assert.Equal(t, assetIDsFromDBB, assetsB)
+
+	associationsAB, err := database.GetAssociationsByAddresses([]string{"A", "B"}, context.Background())
+	assert.Nil(t, err)
+
+	var assetIDsFromDBAB []string
+	for _, a := range associationsAB {
+		assetIDsFromDBAB = append(assetIDsFromDBAB, a.Asset.AssetID)
+	}
+	assetsAB := []string{"cccc", "cccc", "bbb", "bbb", "aa", "as", "f", "f"}
+
+	sort.Slice(assetsAB, func(i, j int) bool {
+		return len(assetsAB[i]) > len(assetsAB[j])
+	})
+
+	sort.Slice(assetIDsFromDBAB, func(i, j int) bool {
+		return len(assetIDsFromDBAB[i]) > len(assetIDsFromDBAB[j])
+	})
+
+	assert.Equal(t, assetIDsFromDBAB, assetsAB)
+}
diff --git a/tests/integration/observer_test/data/wanted_subscriptions_added.json b/tests/integration/observer_test/data/wanted_subscriptions_added.json
index 051f5d6..febd627 100644
--- a/tests/integration/observer_test/data/wanted_subscriptions_added.json
+++ b/tests/integration/observer_test/data/wanted_subscriptions_added.json
@@ -1,18 +1,18 @@
 [
   {
     "coin": 60,
-    "address": "0x0000000000000000000000000000000000000000"
+    "address": "60_0x0000000000000000000000000000000000000000"
   },
   {
     "coin": 118,
-    "address": "0x0000000000000000000000000000000000000001"
+    "address": "118_0x0000000000000000000000000000000000000001"
   },
   {
     "coin": 714,
-    "address": "0x0000000000000000000000000000000000000002"
+    "address": "714_0x0000000000000000000000000000000000000002"
   },
   {
     "coin": 144,
-    "address": "0x0000000000000000000000000000000000000003"
+    "address": "144_0x0000000000000000000000000000000000000003"
   }
 ]
diff --git a/tests/integration/observer_test/full_flow_test.go b/tests/integration/observer_test/full_flow_test.go
index 031d424..d2fddf6 100644
--- a/tests/integration/observer_test/full_flow_test.go
+++ b/tests/integration/observer_test/full_flow_test.go
@@ -8,11 +8,10 @@ import (
 	"github.com/streadway/amqp"
 	"github.com/stretchr/testify/assert"
 	"github.com/trustwallet/blockatlas/coin"
-	"github.com/trustwallet/blockatlas/db/models"
 	"github.com/trustwallet/blockatlas/mq"
 	"github.com/trustwallet/blockatlas/pkg/blockatlas"
-	"github.com/trustwallet/blockatlas/services/observer/notifier"
-	"github.com/trustwallet/blockatlas/services/observer/parser"
+	"github.com/trustwallet/blockatlas/services/notifier"
+	"github.com/trustwallet/blockatlas/services/parser"
 	"github.com/trustwallet/blockatlas/tests/integration/setup"
 	"go.uber.org/atomic"
 	"testing"
@@ -25,7 +24,7 @@ var (
 
 func TestFullFlow(t *testing.T) {
 	setup.CleanupPgContainer(database.Gorm)
-	err := database.AddSubscriptions([]models.Subscription{{Coin: 60, Address: "testAddress"}}, context.Background())
+	err := database.AddSubscriptionsForNotifications([]string{"60_testAddress"}, context.Background())
 	assert.Nil(t, err)
 
 	ctx, cancel := context.WithCancel(context.Background())
@@ -35,7 +34,7 @@ func TestFullFlow(t *testing.T) {
 	params := setupParserFull(stopChan)
 	params.Database = database
 	params.Ctx = ctx
-	params.Queue = mq.RawTransactions
+	params.Queue = []mq.Queue{mq.RawTransactions}
 
 	go parser.RunParser(params)
 	time.Sleep(time.Second * 2)
diff --git a/tests/integration/observer_test/notifier_test.go b/tests/integration/observer_test/notifier_test.go
index 1f64e5d..39d00b0 100644
--- a/tests/integration/observer_test/notifier_test.go
+++ b/tests/integration/observer_test/notifier_test.go
@@ -8,10 +8,9 @@ import (
 	"github.com/streadway/amqp"
 	"github.com/stretchr/testify/assert"
 	"github.com/trustwallet/blockatlas/coin"
-	"github.com/trustwallet/blockatlas/db/models"
 	"github.com/trustwallet/blockatlas/mq"
 	"github.com/trustwallet/blockatlas/pkg/blockatlas"
-	"github.com/trustwallet/blockatlas/services/observer/notifier"
+	"github.com/trustwallet/blockatlas/services/notifier"
 	"github.com/trustwallet/blockatlas/tests/integration/setup"
 	"testing"
 	"time"
@@ -44,7 +43,7 @@ var (
 func TestNotifier(t *testing.T) {
 	setup.CleanupPgContainer(database.Gorm)
 
-	err := database.AddSubscriptions([]models.Subscription{{Coin: 714, Address: "tbnb1ttyn4csghfgyxreu7lmdu3lcplhqhxtzced45a"}}, context.Background())
+	err := database.AddSubscriptionsForNotifications([]string{"714_tbnb1ttyn4csghfgyxreu7lmdu3lcplhqhxtzced45a"}, context.Background())
 	assert.Nil(t, err)
 
 	err = produceTxs(txs)
diff --git a/tests/integration/observer_test/parser_test.go b/tests/integration/observer_test/parser_test.go
index 1c5d6b7..860b552 100644
--- a/tests/integration/observer_test/parser_test.go
+++ b/tests/integration/observer_test/parser_test.go
@@ -11,7 +11,7 @@ import (
 	"github.com/trustwallet/blockatlas/mq"
 	"github.com/trustwallet/blockatlas/pkg/blockatlas"
 	"github.com/trustwallet/blockatlas/pkg/logger"
-	"github.com/trustwallet/blockatlas/services/observer/parser"
+	"github.com/trustwallet/blockatlas/services/parser"
 	"github.com/trustwallet/blockatlas/tests/integration/setup"
 	"testing"
 	"time"
@@ -27,7 +27,7 @@ func TestParserFetchAndPublishBlock_NormalCase(t *testing.T) {
 	ctx, cancel := context.WithCancel(context.Background())
 
 	params.Ctx = ctx
-	params.Queue = mq.RawTransactions
+	params.Queue = []mq.Queue{mq.RawTransactions}
 
 	go parser.RunParser(params)
 
diff --git a/tests/integration/observer_test/subscriber_test.go b/tests/integration/observer_test/subscriber_test.go
index 540d715..fa804f0 100644
--- a/tests/integration/observer_test/subscriber_test.go
+++ b/tests/integration/observer_test/subscriber_test.go
@@ -6,10 +6,9 @@ import (
 	"context"
 	"encoding/json"
 	"github.com/stretchr/testify/assert"
-	"github.com/trustwallet/blockatlas/db/models"
 	"github.com/trustwallet/blockatlas/mq"
 	"github.com/trustwallet/blockatlas/pkg/blockatlas"
-	"github.com/trustwallet/blockatlas/services/observer/subscriber"
+	"github.com/trustwallet/blockatlas/services/subscriber"
 	"github.com/trustwallet/blockatlas/tests/integration/setup"
 	"io/ioutil"
 	"path/filepath"
@@ -51,16 +50,15 @@ func TestSubscriberAddSubscription(t *testing.T) {
 
 		ctx, cancel := context.WithCancel(context.Background())
 
-		go mq.RunConsumerForChannelWithCancelAndDbConn(subscriber.RunSubscriber, subscriptionChannel, database, ctx)
+		go mq.RunConsumerForChannelWithCancelAndDbConn(subscriber.RunTransactionsSubscriber, subscriptionChannel, database, ctx)
 		time.Sleep(time.Second * 2)
 		cancel()
 	}
 
 	for _, wanted := range wantedEvents {
-		result, err := database.GetSubscriptions(wanted.Coin, []string{wanted.Address}, context.Background())
+		result, err := database.GetSubscriptionsForNotifications([]string{wanted.Address}, context.Background())
 		assert.Nil(t, err)
-		assert.Equal(t, result[0].Coin, wanted.Coin)
-		assert.Equal(t, result[0].Address, wanted.Address)
+		assert.Equal(t, result[0].Address.Address, wanted.Address)
 	}
 }
 
@@ -91,14 +89,17 @@ func TestSubscriber_UpdateSubscription(t *testing.T) {
 		t.Fatal(err)
 	}
 
-	database.AddSubscriptions([]models.Subscription{
-		{Coin: 61, Address: "0x0000000000000000000000000000000000000000"}}, context.Background())
-	database.AddSubscriptions([]models.Subscription{
-		{Coin: 62, Address: "0x0000000000000000000000000000000000000000"}}, context.Background())
-	database.AddSubscriptions([]models.Subscription{
-		{Coin: 63, Address: "0x0000000000000000000000000000000000000000"}}, context.Background())
-	database.AddSubscriptions([]models.Subscription{
-		{Coin: 64, Address: "0x0000000000000000000000000000000000000000"}}, context.Background())
+	database.AddSubscriptionsForNotifications(
+		[]string{"61_0x0000000000000000000000000000000000000000"}, context.Background())
+
+	database.AddSubscriptionsForNotifications(
+		[]string{"62_0x0000000000000000000000000000000000000000"}, context.Background())
+
+	database.AddSubscriptionsForNotifications(
+		[]string{"63_0x0000000000000000000000000000000000000000"}, context.Background())
+
+	database.AddSubscriptionsForNotifications(
+		[]string{"64_0x0000000000000000000000000000000000000000"}, context.Background())
 
 	for _, event := range givenEvents {
 		body, err := json.Marshal(event)
@@ -109,36 +110,34 @@ func TestSubscriber_UpdateSubscription(t *testing.T) {
 
 		ctx, cancel := context.WithCancel(context.Background())
 
-		go mq.RunConsumerForChannelWithCancelAndDbConn(subscriber.RunSubscriber, subscriptionChannel, database, ctx)
+		go mq.RunConsumerForChannelWithCancelAndDbConn(subscriber.RunTransactionsSubscriber, subscriptionChannel, database, ctx)
 		time.Sleep(time.Second)
 		cancel()
 	}
 
 	for _, wanted := range wantedEvents {
-		result, err := database.GetSubscriptions(wanted.Coin, []string{wanted.Address}, context.Background())
+		result, err := database.GetSubscriptionsForNotifications([]string{wanted.Address}, context.Background())
 		assert.Nil(t, err)
-		assert.Equal(t, result[0].Coin, wanted.Coin)
-		assert.Equal(t, result[0].Address, wanted.Address)
-
+		assert.Len(t, result, 1)
 	}
 
-	abs61, err := database.GetSubscriptions(61, []string{"0x0000000000000000000000000000000000000000"}, context.Background())
+	abs61, err := database.GetSubscriptionsForNotifications([]string{"61_0x0000000000000000000000000000000000000000"}, context.Background())
 	assert.Nil(t, err)
 	assert.Len(t, abs61, 1)
 
-	abs62, err := database.GetSubscriptions(62, []string{"0x0000000000000000000000000000000000000000"}, context.Background())
+	abs62, err := database.GetSubscriptionsForNotifications([]string{"62_0x0000000000000000000000000000000000000000"}, context.Background())
 	assert.Nil(t, err)
 	assert.Len(t, abs62, 1)
 
-	abs63, err := database.GetSubscriptions(63, []string{"0x0000000000000000000000000000000000000000"}, context.Background())
+	abs63, err := database.GetSubscriptionsForNotifications([]string{"63_0x0000000000000000000000000000000000000000"}, context.Background())
 	assert.Nil(t, err)
 	assert.Len(t, abs63, 1)
 
-	abs64, err := database.GetSubscriptions(64, []string{"0x0000000000000000000000000000000000000000"}, context.Background())
+	abs64, err := database.GetSubscriptionsForNotifications([]string{"64_0x0000000000000000000000000000000000000000"}, context.Background())
 	assert.Nil(t, err)
 	assert.Len(t, abs64, 1)
 
-	abs65, err := database.GetSubscriptions(65, []string{"0x0000000000000000000000000000000000000000"}, context.Background())
+	abs65, err := database.GetSubscriptionsForNotifications([]string{"65_0x0000000000000000000000000000000000000000"}, context.Background())
 	assert.Nil(t, err)
 	assert.Len(t, abs65, 0)
 }
diff --git a/tests/integration/setup/postgres.go b/tests/integration/setup/postgres.go
index ec09821..d300c16 100644
--- a/tests/integration/setup/postgres.go
+++ b/tests/integration/setup/postgres.go
@@ -24,8 +24,12 @@ var (
 	}
 
 	tables = []interface{}{
-		&models.Subscription{},
+		&models.AssetSubscription{},
+		&models.NotificationSubscription{},
 		&models.Tracker{},
+		&models.AddressToAssetAssociation{},
+		&models.Asset{},
+		&models.Address{},
 	}
 
 	uri string