Repository: trustwallet/blockatlas
Pull request #1119 (2020-05-29 16:34:15)

Fix issues with notifier
- Remove legacy codebase with weird maps with pointers as key
- Cover new codebase with unit tests
- Solve issue with multiple transactions duplicates that have occured because of (1) 


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

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

```diff
@@          Coverage Diff           @@
##           master   #1119   +/-   ##
======================================
- Coverage      39%     38%   -1%     
======================================
  Files         162     161    -1     
  Lines        5510    5473   -37     
======================================
- Hits         2154    2124   -30     
+ Misses       3212    3205    -7     
  Partials      144     144           
```


| [Impacted Files](https://codecov.io/gh/trustwallet/blockatlas/pull/1119?src=pr&el=tree) | Coverage Δ | |
|---|---|---|
| [pkg/blockatlas/tx.go](https://codecov.io/gh/trustwallet/blockatlas/pull/1119/diff?src=pr&el=tree#diff-cGtnL2Jsb2NrYXRsYXMvdHguZ28=) | `81% <ø> (-2%)` | :arrow_down: |

------

[Continue to review full report at Codecov](https://codecov.io/gh/trustwallet/blockatlas/pull/1119?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/1119?src=pr&el=footer). Last update [e71a841...855d841](https://codecov.io/gh/trustwallet/blockatlas/pull/1119?src=pr&el=lastupdated). Read the [comment docs](https://docs.codecov.io/docs/pull-request-comments).


Patch:
diff --git a/cmd/observer_parser/main.go b/cmd/observer_parser/main.go
index e828283..ccbd384 100644
--- a/cmd/observer_parser/main.go
+++ b/cmd/observer_parser/main.go
@@ -9,7 +9,6 @@ import (
 	"github.com/trustwallet/blockatlas/mq"
 	"github.com/trustwallet/blockatlas/pkg/logger"
 	"github.com/trustwallet/blockatlas/platform"
-	"github.com/trustwallet/blockatlas/services/observer/notifier"
 	"github.com/trustwallet/blockatlas/services/observer/parser"
 	"os"
 	"os/signal"
@@ -86,7 +85,7 @@ func main() {
 	for _, api := range platform.BlockAPIs {
 		time.Sleep(time.Millisecond * 5)
 		coin := api.Coin()
-		pollInterval := notifier.GetInterval(coin.BlockTime, minInterval, maxInterval)
+		pollInterval := parser.GetInterval(coin.BlockTime, minInterval, maxInterval)
 
 		var backlogCount int
 		if coin.BlockTime == 0 {
diff --git a/pkg/blockatlas/tx.go b/pkg/blockatlas/tx.go
index d0ec8a6..a555d6f 100644
--- a/pkg/blockatlas/tx.go
+++ b/pkg/blockatlas/tx.go
@@ -228,21 +228,6 @@ func (t Txs) SortByDate() Txs {
 	return t
 }
 
-func (t Txs) GetTransactionsMap() TxSetMap {
-	txSetMap := TxSetMap{Map: make(map[string]*TxSet)}
-	for i := 0; i < len(t); i++ {
-		addresses := t[i].GetAddresses()
-		addresses = append(addresses, t[i].GetUtxoAddresses()...)
-		for _, address := range addresses {
-			if txSetMap.Map[address] == nil {
-				txSetMap.Map[address] = new(TxSet)
-			}
-			txSetMap.Map[address].Add(&t[i])
-		}
-	}
-	return txSetMap
-}
-
 func (t *Tx) GetUtxoAddresses() (addresses []string) {
 	for _, input := range t.Inputs {
 		addresses = append(addresses, input.Address)
diff --git a/pkg/blockatlas/txset.go b/pkg/blockatlas/txset.go
deleted file mode 100644
index 1819a14..0000000
--- a/pkg/blockatlas/txset.go
+++ /dev/null
@@ -1,55 +0,0 @@
-package blockatlas
-
-import "sync"
-
-type (
-	TxSetMap struct {
-		Map map[string]*TxSet
-	}
-
-	TxSet struct {
-		items map[*Tx]bool
-		lock  sync.RWMutex
-	}
-)
-
-// Add adds a new element to the Map. Returns a pointer to the Map.
-func (s *TxSet) Add(t *Tx) *TxSet {
-	s.lock.Lock()
-	defer s.lock.Unlock()
-	if s.items == nil {
-		s.items = make(map[*Tx]bool)
-	}
-	_, ok := s.items[t]
-	if !ok {
-		s.items[t] = true
-	}
-	return s
-}
-
-func (s *TxSet) Txs() []Tx {
-	s.lock.RLock()
-	defer s.lock.RUnlock()
-	items := []Tx{}
-	for i := range s.items {
-		items = append(items, *i)
-	}
-	return items
-}
-
-func (s *TxSet) Size() int {
-	s.lock.RLock()
-	defer s.lock.RUnlock()
-	return len(s.items)
-}
-
-func (s TxSetMap) GetUniqueAddresses() []string {
-	var addresses []string
-	for address := range s.Map {
-		if len(address) == 0 {
-			continue
-		}
-		addresses = append(addresses, address)
-	}
-	return addresses
-}
diff --git a/services/observer/notifier/base.go b/services/observer/notifier/base.go
new file mode 100644
index 0000000..2e56ae4
--- /dev/null
+++ b/services/observer/notifier/base.go
@@ -0,0 +1,55 @@
+package notifier
+
+import (
+	"context"
+	"github.com/streadway/amqp"
+	"github.com/trustwallet/blockatlas/db"
+	"github.com/trustwallet/blockatlas/pkg/logger"
+
+	"go.elastic.co/apm"
+)
+
+const DefaultPushNotificationsBatchLimit = 50
+
+var MaxPushNotificationsBatchLimit uint = DefaultPushNotificationsBatchLimit
+
+func RunNotifier(database *db.Instance, delivery amqp.Delivery) {
+	tx := apm.DefaultTracer.StartTransaction("RunNotifier", "app")
+	defer tx.End()
+	ctx := apm.ContextWithTransaction(context.Background(), tx)
+
+	defer func() {
+		if err := delivery.Ack(false); err != nil {
+			logger.Error(err)
+		}
+	}()
+
+	txs, err := getTransactionsFromDelivery(delivery, ctx)
+	if err != nil {
+		logger.Error("failed to get transactions", err)
+	}
+
+	allAddresses := make([]string, 0)
+	for _, tx := range txs {
+		allAddresses = append(allAddresses, tx.GetAddresses()...)
+	}
+
+	addresses := toUniqueAddresses(allAddresses)
+
+	subscriptionsDataList, err := database.GetSubscriptions(txs[0].Coin, addresses, ctx)
+	if err != nil || len(subscriptionsDataList) == 0 {
+		return
+	}
+
+	notifications := make([]TransactionNotification, 0)
+	for _, sub := range subscriptionsDataList {
+		notificationsForAddress := buildNotificationsByAddress(sub.Address, txs, ctx)
+		notifications = append(notifications, notificationsForAddress...)
+	}
+
+	batches := getNotificationBatches(notifications, MaxPushNotificationsBatchLimit, ctx)
+
+	for _, batch := range batches {
+		publishNotificationBatch(batch, ctx)
+	}
+}
diff --git a/services/observer/notifier/delivery.go b/services/observer/notifier/delivery.go
new file mode 100644
index 0000000..9a1c3ca
--- /dev/null
+++ b/services/observer/notifier/delivery.go
@@ -0,0 +1,48 @@
+package notifier
+
+import (
+	"context"
+	"encoding/json"
+	"github.com/streadway/amqp"
+	"github.com/trustwallet/blockatlas/mq"
+	"github.com/trustwallet/blockatlas/pkg/blockatlas"
+	"github.com/trustwallet/blockatlas/pkg/errors"
+	"github.com/trustwallet/blockatlas/pkg/logger"
+	"go.elastic.co/apm"
+)
+
+func getTransactionsFromDelivery(delivery amqp.Delivery, ctx context.Context) (blockatlas.Txs, error) {
+	var txs blockatlas.Txs
+
+	span, _ := apm.StartSpan(ctx, "getTransactionsFromDelivery", "app")
+	defer span.End()
+
+	if err := json.Unmarshal(delivery.Body, &txs); err != nil {
+		return nil, err
+	}
+
+	logger.Info("Consumed", logger.Params{"txs": len(txs), "coin": txs[0].Coin})
+
+	if len(txs) == 0 {
+		return nil, errors.E("empty txs list")
+	}
+	return txs, nil
+}
+
+func publishNotificationBatch(batch []TransactionNotification, ctx context.Context) {
+	span, _ := apm.StartSpan(ctx, "getNotificationBatches", "app")
+	defer span.End()
+
+	raw, err := json.Marshal(batch)
+	if err != nil {
+		err = errors.E(err, " failed to dispatch event")
+		logger.Fatal(err)
+	}
+	err = mq.TxNotifications.Publish(raw)
+	if err != nil {
+		err = errors.E(err, " failed to dispatch event")
+		logger.Fatal(err)
+	}
+
+	logger.Info("Txs batch dispatched", logger.Params{"txs": len(batch)})
+}
diff --git a/services/observer/notifier/models.go b/services/observer/notifier/models.go
new file mode 100644
index 0000000..f33054c
--- /dev/null
+++ b/services/observer/notifier/models.go
@@ -0,0 +1,91 @@
+package notifier
+
+import (
+	"context"
+	"github.com/trustwallet/blockatlas/pkg/blockatlas"
+	"go.elastic.co/apm"
+)
+
+type TransactionNotification struct {
+	Action blockatlas.TransactionType `json:"action"`
+	Result blockatlas.Tx              `json:"result"`
+}
+
+func getNotificationBatches(notifications []TransactionNotification, sizeUint uint, ctx context.Context) [][]TransactionNotification {
+	span, _ := apm.StartSpan(ctx, "getNotificationBatches", "app")
+	defer span.End()
+	size := int(sizeUint)
+	resultLength := (len(notifications) + size - 1) / size
+	result := make([][]TransactionNotification, resultLength)
+	lo, hi := 0, size
+	for i := range result {
+		if hi > len(notifications) {
+			hi = len(notifications)
+		}
+		result[i] = notifications[lo:hi:hi]
+		lo, hi = hi, hi+size
+	}
+	return result
+}
+
+func buildNotificationsByAddress(address string, txs blockatlas.Txs, ctx context.Context) []TransactionNotification {
+	span, _ := apm.StartSpan(ctx, "buildNotification", "app")
+	defer span.End()
+
+	transactionsByAddress := toUniqueTransactions(findTransactionsByAddress(txs, address))
+
+	result := make([]TransactionNotification, 0, len(transactionsByAddress))
+	for _, tx := range transactionsByAddress {
+		tx.Direction = tx.GetTransactionDirection(address)
+		tx.InferUtxoValue(address, tx.Coin)
+		result = append(result, TransactionNotification{Action: tx.Type, Result: tx})
+	}
+
+	return result
+}
+
+func toUniqueAddresses(addresses []string) []string {
+	keys := make(map[string]bool)
+	var list []string
+	for _, entry := range addresses {
+		if _, value := keys[entry]; !value {
+			keys[entry] = true
+			list = append(list, entry)
+		}
+	}
+	return list
+}
+
+func toUniqueTransactions(txs []blockatlas.Tx) []blockatlas.Tx {
+	keys := make(map[string]bool)
+	var list []blockatlas.Tx
+	for _, entry := range txs {
+		key := entry.ID + string(entry.Direction)
+		if _, value := keys[key]; !value {
+			keys[key] = true
+			list = append(list, entry)
+		}
+	}
+	return list
+}
+
+func findTransactionsByAddress(txs blockatlas.Txs, address string) []blockatlas.Tx {
+	result := make([]blockatlas.Tx, 0)
+	for _, tx := range txs {
+		if containsAddress(tx, address) {
+			result = append(result, tx)
+		}
+	}
+	return result
+}
+
+func containsAddress(tx blockatlas.Tx, address string) bool {
+	allAddresses := tx.GetAddresses()
+	txAddresses := toUniqueAddresses(allAddresses)
+	for _, a := range txAddresses {
+		if a == address {
+			return true
+		}
+	}
+	return false
+}
diff --git a/services/observer/notifier/notifier.go b/services/observer/notifier/notifier.go
deleted file mode 100644
index e1a0193..0000000
--- a/services/observer/notifier/notifier.go
+++ /dev/null
@@ -1,140 +0,0 @@
-package notifier
-
-import (
-	"context"
-	"encoding/json"
-	"github.com/streadway/amqp"
-	"github.com/trustwallet/blockatlas/db"
-	"github.com/trustwallet/blockatlas/mq"
-	"github.com/trustwallet/blockatlas/pkg/blockatlas"
-	"github.com/trustwallet/blockatlas/pkg/errors"
-	"github.com/trustwallet/blockatlas/pkg/logger"
-	"github.com/trustwallet/blockatlas/pkg/numbers"
-	"go.elastic.co/apm"
-	"sync"
-	"time"
-)
-
-const DefaultPushNotificationsBatchLimit = 50
-
-var MaxPushNotificationsBatchLimit uint = DefaultPushNotificationsBatchLimit
-
-type TransactionNotification struct {
-	Action blockatlas.TransactionType `json:"action"`
-	Result *blockatlas.Tx             `json:"result"`
-}
-
-func RunNotifier(database *db.Instance, delivery amqp.Delivery) {
-	tx := apm.DefaultTracer.StartTransaction("RunNotifier", "app")
-	defer tx.End()
-	ctx := apm.ContextWithTransaction(context.Background(), tx)
-	defer func() {
-		if err := delivery.Ack(false); err != nil {
-			logger.Error(err)
-		}
-	}()
-	var txs blockatlas.Txs
-	if err := json.Unmarshal(delivery.Body, &txs); err != nil {
-		logger.Error(err)
-		return
-	}
-	if len(txs) == 0 {
-		return
-	}
-
-	logger.Info("Consumed", logger.Params{"txs": len(txs), "coin": txs[0].Coin})
-
-	blockTransactions := txs.GetTransactionsMap()
-	if len(blockTransactions.Map) == 0 {
-		return
-	}
-
-	addresses := blockTransactions.GetUniqueAddresses()
-	subscriptionsDataList, err := database.GetSubscriptions(txs[0].Coin, addresses, ctx)
-	if err != nil || len(subscriptionsDataList) == 0 {
-		return
-	}
-
-	var wg sync.WaitGroup
-	wg.Add(len(subscriptionsDataList))
-	for _, data := range subscriptionsDataList {
-		go buildAndPostMessage(
-			blockTransactions,
-			blockatlas.Subscription{Coin: data.Coin, Address: data.Address},
-			&wg, ctx)
-	}
-	wg.Wait()
-}
-
-func buildAndPostMessage(blockTransactions blockatlas.TxSetMap, sub blockatlas.Subscription, wg *sync.WaitGroup, ctx context.Context) {
-	defer wg.Done()
-
-	span, ctx := apm.StartSpan(ctx, "buildAndPostMessage", "app")
-	defer span.End()
-
-	tx, ok := blockTransactions.Map[sub.Address]
-	if !ok {
-		return
-	}
-	notifications := make([]TransactionNotification, 0, len(tx.Txs()))
-	for _, tx := range tx.Txs() {
-		tx.Direction = tx.GetTransactionDirection(sub.Address)
-		tx.InferUtxoValue(sub.Address, tx.Coin)
-		notification := TransactionNotification{
-			Action: tx.Type,
-			Result: &tx,
-		}
-
-		logger.Info("Notification ready", logger.Params{"coin": sub.Coin, "txID": tx.ID})
-
-		notifications = append(notifications, notification)
-	}
-
-	batches := getNotificationBatches(notifications, MaxPushNotificationsBatchLimit, ctx)
-
-	for _, batch := range batches {
-		publishNotificationBatch(batch, ctx)
-	}
-}
-
-func publishNotificationBatch(batch []TransactionNotification, ctx context.Context) {
-	span, _ := apm.StartSpan(ctx, "getNotificationBatches", "app")
-	defer span.End()
-	raw, err := json.Marshal(batch)
-	if err != nil {
-		err = errors.E(err, " failed to dispatch event")
-		logger.Fatal(err)
-	}
-
-	err = mq.TxNotifications.Publish(raw)
-	if err != nil {
-		err = errors.E(err, " failed to dispatch event")
-		logger.Fatal(err)
-	}
-
-	logger.Info("Txs batch dispatched", logger.Params{"txs": len(batch)})
-}
-
-func GetInterval(value int, minInterval, maxInterval time.Duration) time.Duration {
-	interval := time.Duration(value) * time.Millisecond
-	pMin := numbers.Max(minInterval.Nanoseconds(), interval.Nanoseconds())
-	pMax := numbers.Min(int(maxInterval.Nanoseconds()), int(pMin))
-	return time.Duration(pMax)
-}
-
-func getNotificationBatches(notifications []TransactionNotification, sizeUint uint, ctx context.Context) [][]TransactionNotification {
-	span, _ := apm.StartSpan(ctx, "getNotificationBatches", "app")
-	defer span.End()
-	size := int(sizeUint)
-	resultLength := (len(notifications) + size - 1) / size
-	result := make([][]TransactionNotification, resultLength)
-	lo, hi := 0, size
-	for i := range result {
-		if hi > len(notifications) {
-			hi = len(notifications)
-		}
-		result[i] = notifications[lo:hi:hi]
-		lo, hi = hi, hi+size
-	}
-	return result
-}
diff --git a/services/observer/parser/parser.go b/services/observer/parser/parser.go
index e5e103c..0e793c6 100644
--- a/services/observer/parser/parser.go
+++ b/services/observer/parser/parser.go
@@ -8,6 +8,7 @@ import (
 	"github.com/trustwallet/blockatlas/mq"
 	"github.com/trustwallet/blockatlas/pkg/blockatlas"
 	"github.com/trustwallet/blockatlas/pkg/errors"
+	"github.com/trustwallet/blockatlas/pkg/numbers"
 	"go.elastic.co/apm"
 	"strconv"
 	"sync/atomic"
@@ -61,6 +62,13 @@ func RunParser(params Params) {
 	}
 }
 
+func GetInterval(value int, minInterval, maxInterval time.Duration) time.Duration {
+	interval := time.Duration(value) * time.Millisecond
+	pMin := numbers.Max(minInterval.Nanoseconds(), interval.Nanoseconds())
+	pMax := numbers.Min(int(maxInterval.Nanoseconds()), int(pMin))
+	return time.Duration(pMax)
+}
+
 func parse(params Params) {
 	tx := apm.DefaultTracer.StartTransaction("parse", "app")
 	ctx := apm.ContextWithTransaction(context.Background(), tx)


Test patch:
diff --git a/pkg/blockatlas/tx_test.go b/pkg/blockatlas/tx_test.go
index 688f733..1f1fc8d 100644
--- a/pkg/blockatlas/tx_test.go
+++ b/pkg/blockatlas/tx_test.go
@@ -109,17 +109,6 @@ var utxoTransferDst2 = Tx{
 	Memo:   "test",
 }
 
-func TestTxSet_Add(t *testing.T) {
-	set := TxSet{}
-	set.Add(&transferDst1)
-	var txs = set.Txs()
-	assert.Equal(t, txs[0].ID, transferDst1.ID)
-	set.Add(&transferDst1)
-	assert.Equal(t, set.Size(), 1)
-	set.Add(&nativeTransferDst1)
-	assert.Equal(t, set.Size(), 2)
-}
-
 func TestTx_GetAddresses(t *testing.T) {
 	assert.Equal(t, transferDst1.GetAddresses(), []string{"tbnb1fhr04azuhcj0dulm7ka40y0cqjlafwae9k9gk2", "tbnb1sylyjw032eajr9cyllp26n04300qzzre38qyv5"})
 	assert.Equal(t, nativeTransferDst1.GetAddresses(), []string{"tbnb1ttyn4csghfgyxreu7lmdu3lcplhqhxtzced45a", "tbnb12hlquylu78cjylk5zshxpdj6hf3t0tahwjt3ex"})
@@ -403,97 +392,6 @@ func TestInferDirection(t *testing.T) {
 	}
 }
 
-var (
-	transferDstOne = Tx{
-		ID:     "1681EE543FB4B5A628EF21D746E031F018E226D127044A4F9BA5EE2542A44555",
-		Coin:   coin.BNB,
-		From:   "tbnb1fhr04azuhcj0dulm7ka40y0cqjlafwae9k9gk2",
-		To:     "tbnb1sylyjw032eajr9cyllp26n04300qzzre38qyv5",
-		Fee:    "125000",
-		Date:   1555049867,
-		Block:  7761368,
-		Status: StatusCompleted,
-		Memo:   "test",
-		Meta: Transfer{
-			Value:    "10000000000000",
-			Decimals: 8,
-			Symbol:   "BNB",
-		},
-	}
-
-	transferDst2 = Tx{
-		ID:     "1681EE543FB4B5A628EF21D746E031F018E226D127044A4F9BA5EE2542A44556",
-		Coin:   coin.BNB,
-		From:   "tbnb1fhr04azuhcj0dulm7ka40y0cqjlafwae9k9gk2",
-		To:     "tbnb1fhr04azuhcj0dulm7ka40y0cqjlafwae9k9gk2",
-		Fee:    "125000",
-		Date:   1555049867,
-		Block:  7761368,
-		Status: StatusCompleted,
-		Memo:   "test",
-		Meta: Transfer{
-			Value:    "10000000000000",
-			Decimals: 8,
-			Symbol:   "BNB",
-		},
-	}
-
-	nativeTransferDstOne = Tx{
-		ID:     "95CF63FAA27579A9B6AF84EF8B2DFEAC29627479E9C98E7F5AE4535E213FA4C9",
-		Coin:   coin.BNB,
-		From:   "tbnb1ttyn4csghfgyxreu7lmdu3lcplhqhxtzced45a",
-		To:     "tbnb12hlquylu78cjylk5zshxpdj6hf3t0tahwjt3ex",
-		Fee:    "125000",
-		Date:   1555117625,
-		Block:  7928667,
-		Status: StatusCompleted,
-		Memo:   "test",
-		Meta: NativeTokenTransfer{
-			TokenID:  "YLC-D8B",
-			Symbol:   "YLC",
-			Value:    "210572645",
-			Decimals: 8,
-			From:     "tbnb1ttyn4csghfgyxreu7lmdu3lcplhqhxtzced45a",
-			To:       "tbnb12hlquylu78cjylk5zshxpdj6hf3t0tahwjt3ex",
-		},
-	}
-
-	nativeTransferDst2 = Tx{
-		ID:     "95CF63FAA27579A9B6AF84EF8B2DFEAC29627479E9C98E7F5AE4535E213FA4D0",
-		Coin:   coin.BNB,
-		From:   "tbnb1ttyn4csghfgyxreu7lmdu3lcplhqhxtzced45a",
-		To:     "tbnb12hlquylu78cjylk5zshxpdj6hf3t0tahwjt3ex",
-		Fee:    "125000",
-		Date:   1555117625,
-		Block:  7928667,
-		Status: StatusCompleted,
-		Memo:   "test",
-		Meta: NativeTokenTransfer{
-			TokenID:  "YLC-D8B",
-			Symbol:   "YLC",
-			Value:    "210572645",
-			Decimals: 8,
-			From:     "tbnb1ttyn4csghfgyxreu7lmdu3lcplhqhxtzced45a",
-			To:       "tbnb12hlquylu78cjylk5zshxpdj6hf3t0tahwjt3ex",
-		},
-	}
-)
-
-func TestGetTxsTx(t *testing.T) {
-	tx := Txs{
-		transferDstOne,
-		transferDst2,
-		nativeTransferDstOne,
-		nativeTransferDst2,
-	}
-	txs := tx.GetTransactionsMap()
-	assert.Equal(t, len(txs.Map), 4)
-	assert.Equal(t, txs.Map["tbnb1fhr04azuhcj0dulm7ka40y0cqjlafwae9k9gk2"].Size(), 2)
-	assert.Equal(t, txs.Map["tbnb1sylyjw032eajr9cyllp26n04300qzzre38qyv5"].Size(), 1)
-	assert.Equal(t, txs.Map["tbnb1ttyn4csghfgyxreu7lmdu3lcplhqhxtzced45a"].Size(), 2)
-	assert.Equal(t, txs.Map["tbnb12hlquylu78cjylk5zshxpdj6hf3t0tahwjt3ex"].Size(), 2)
-}
-
 func TestTx_GetTransactionDirection(t *testing.T) {
 	txMeta := TokenTransfer{
 		Name:     "Kyber Network Crystal",
diff --git a/services/observer/notifier/models_test.go b/services/observer/notifier/models_test.go
new file mode 100644
index 0000000..0569906
--- /dev/null
+++ b/services/observer/notifier/models_test.go
@@ -0,0 +1,146 @@
+package notifier
+
+import (
+	"context"
+	"github.com/stretchr/testify/assert"
+	"github.com/trustwallet/blockatlas/coin"
+	"github.com/trustwallet/blockatlas/pkg/blockatlas"
+	"sort"
+	"testing"
+)
+
+var (
+	nativeTokenTransfer = blockatlas.Tx{
+		ID:     "95CF63FAA27579A9B6AF84EF8B2DFEAC29627479E9C98E7F5AE4535E213FA4C9",
+		Coin:   coin.BNB,
+		From:   "tbnb1ttyn4csghfgyxreu7lmdu3lcplhqhxtzced45a",
+		To:     "tbnb12hlquylu78cjylk5zshxpdj6hf3t0tahwjt3ex",
+		Fee:    "125000",
+		Date:   1555117625,
+		Block:  7928667,
+		Status: blockatlas.StatusCompleted,
+		Memo:   "test",
+		Meta: blockatlas.NativeTokenTransfer{
+			TokenID:  "YLC-D8B",
+			Symbol:   "YLC",
+			Value:    "210572645",
+			Decimals: 8,
+			From:     "tbnb1ttyn4csghfgyxreu7lmdu3lcplhqhxtzced45a",
+			To:       "tbnb12hlquylu78cjylk5zshxpdj6hf3t0tahwjt3ex",
+		},
+	}
+	tokenTransfer = blockatlas.Tx{
+		ID:       "0xbcd1a43e796de4035e5e2991d8db332958e36031d54cb1d3a08d2cb790e338c4",
+		Coin:     60,
+		From:     "0x08777CB1e80F45642752662B04886Df2d271E049",
+		To:       "0xdd974D5C2e2928deA5F71b9825b8b646686BD200",
+		Fee:      "52473000000000",
+		Date:     1585169424,
+		Block:    9742705,
+		Status:   "completed",
+		Sequence: 149,
+		Type:     "token_transfer",
+		Meta: blockatlas.TokenTransfer{
+			Name:     "Kyber Network Crystal",
+			Symbol:   "KNC",
+			TokenID:  "0xdd974D5C2e2928deA5F71b9825b8b646686BD200",
+			Decimals: 18,
+			Value:    "100000000000000",
+			From:     "0x08777CB1e80F45642752662B04886Df2d271E049",
+			To:       "0x38d45371993eEc84f38FEDf93C646aA2D2267CEA",
+		},
+	}
+	transfer = blockatlas.Tx{
+		ID:     "1681EE543FB4B5A628EF21D746E031F018E226D127044A4F9BA5EE2542A44556",
+		Coin:   coin.BNB,
+		From:   "tbnb1fhr04azuhcj0dulm7ka40y0cqjlafwae9k9gk2",
+		To:     "tbnb1fhr04azuhcj0dulm7ka40y0cqjlafwae9k9gk2",
+		Fee:    "125000",
+		Date:   1555049867,
+		Block:  7761368,
+		Status: blockatlas.StatusCompleted,
+		Memo:   "test",
+		Meta: blockatlas.Transfer{
+			Value:    "10000000000000",
+			Decimals: 8,
+			Symbol:   "BNB",
+		},
+	}
+	utxoTransfer = blockatlas.Tx{
+		ID:   "zpub6ruK9k6YGm8BRHWvTiQcrEPnFkuRDJhR7mPYzV2LDvjpLa5CuGgrhCYVZjMGcLcFqv9b2WvsFtY2Gb3xq8NVq8qhk9veozrA2W9QaWtihrC",
+		Coin: coin.BTC,
+		Inputs: []blockatlas.TxOutput{
+			{
+				Address: "bc1qhn03cww757mnnlpkdvvfkaydxqygm86nvkm92h",
+				Value:   "1",
+			},
+			{
+				Address: "bc1qc7ekqf2t0elfsmtgr2mgd7da2up4vgq8uqk2nh",
+				Value:   "1",
+			},
+			{
+				Address: "bc1qv454wacvnenr3hzzldjqn8cgfltdlxwe96h737",
+				Value:   "1",
+			},
+		},
+		Outputs: []blockatlas.TxOutput{
+			{
+				Address: "bc1qjcslq88cht8llqmh3aqshjx9we9msv386jvxl6",
+				Value:   "3",
+			},
+		},
+		From:   "bc1qhn03cww757mnnlpkdvvfkaydxqygm86nvkm92h",
+		To:     "bc1qjcslq88cht8llqmh3aqshjx9we9msv386jvxl6",
+		Fee:    "125000",
+		Date:   1555117625,
+		Block:  592400,
+		Status: blockatlas.StatusCompleted,
+		Memo:   "test",
+		Meta: blockatlas.Transfer{
+			Value:    "10000000000000",
+			Decimals: 8,
+			Symbol:   "BNB",
+		},
+	}
+)
+
+func Test_containsAddress(t *testing.T) {
+	assert.True(t, containsAddress(tokenTransfer, "0x08777CB1e80F45642752662B04886Df2d271E049"))
+	assert.False(t, containsAddress(tokenTransfer, "0xdd974D5C2e2928deA5F71b9825b8b646686BD200"))
+	assert.True(t, containsAddress(tokenTransfer, "0x38d45371993eEc84f38FEDf93C646aA2D2267CEA"))
+	assert.False(t, containsAddress(tokenTransfer, "0xdd974D5C2e2928deA5F71b9825b8b646686BD200"))
+
+	assert.True(t, containsAddress(transfer, "tbnb1fhr04azuhcj0dulm7ka40y0cqjlafwae9k9gk2"))
+	assert.False(t, containsAddress(transfer, "1681EE543FB4B5A628EF21D746E031F018E226D127044A4F9BA5EE2542A44556"))
+	assert.True(t, containsAddress(transfer, "tbnb1fhr04azuhcj0dulm7ka40y0cqjlafwae9k9gk2"))
+
+	assert.True(t, containsAddress(nativeTokenTransfer, "tbnb1ttyn4csghfgyxreu7lmdu3lcplhqhxtzced45a"))
+	assert.False(t, containsAddress(nativeTokenTransfer, "95CF63FAA27579A9B6AF84EF8B2DFEAC29627479E9C98E7F5AE4535E213FA4C9"))
+	assert.True(t, containsAddress(nativeTokenTransfer, "tbnb12hlquylu78cjylk5zshxpdj6hf3t0tahwjt3ex"))
+
+	assert.True(t, containsAddress(utxoTransfer, "bc1qhn03cww757mnnlpkdvvfkaydxqygm86nvkm92h"))
+	assert.False(t, containsAddress(utxoTransfer, "bc1qc7ekqf2t0elfsmtgr2mgd7da2up4vgq8uqk2nh"))
+	assert.False(t, containsAddress(utxoTransfer, "bc1qv454wacvnenr3hzzldjqn8cgfltdlxwe96h737"))
+	assert.False(t, containsAddress(utxoTransfer, "zpub6ruK9k6YGm8BRHWvTiQcrEPnFkuRDJhR7mPYzV2LDvjpLa5CuGgrhCYVZjMGcLcFqv9b2WvsFtY2Gb3xq8NVq8qhk9veozrA2W9QaWtihrC"))
+	assert.True(t, containsAddress(utxoTransfer, "bc1qjcslq88cht8llqmh3aqshjx9we9msv386jvxl6"))
+}
+
+func Test_findTransactionsByAddress(t *testing.T) {
+	res := findTransactionsByAddress([]blockatlas.Tx{nativeTokenTransfer, tokenTransfer}, "tbnb1ttyn4csghfgyxreu7lmdu3lcplhqhxtzced45a")
+	sort.Slice(res, func(i, j int) bool {
+		return res[i].ID < res[j].ID
+	})
+	assert.Equal(t, []blockatlas.Tx{nativeTokenTransfer}, res)
+
+	resFail := findTransactionsByAddress([]blockatlas.Tx{nativeTokenTransfer, tokenTransfer}, "tbnb1ttyn4csghfgyxreu7lmdu3lcplhqhxtzced")
+	assert.Equal(t, []blockatlas.Tx{}, resFail)
+}
+
+func Test_buildNotificationsByAddress(t *testing.T) {
+	notifications := buildNotificationsByAddress("tbnb1ttyn4csghfgyxreu7lmdu3lcplhqhxtzced45a", []blockatlas.Tx{nativeTokenTransfer, tokenTransfer}, context.Background())
+	sort.Slice(notifications, func(i, j int) bool {
+		return notifications[i].Action < notifications[j].Action
+	})
+	nativeTokenTransfer.Direction = blockatlas.DirectionOutgoing
+	assert.Equal(t, nativeTokenTransfer, notifications[0].Result)
+}
diff --git a/services/observer/notifier/notifier_test.go b/services/observer/notifier/notifier_test.go
deleted file mode 100644
index 882343e..0000000
--- a/services/observer/notifier/notifier_test.go
+++ /dev/null
@@ -1,54 +0,0 @@
-package notifier
-
-import (
-	"github.com/stretchr/testify/assert"
-	"testing"
-	"time"
-)
-
-func TestGetInterval(t *testing.T) {
-	min, _ := time.ParseDuration("2s")
-	max, _ := time.ParseDuration("30s")
-	type args struct {
-		blockTime   int
-		minInterval time.Duration
-		maxInterval time.Duration
-	}
-	tests := []struct {
-		name string
-		args args
-		want time.Duration
-	}{
-		{
-			"test minimum",
-			args{
-				blockTime:   100,
-				minInterval: min,
-				maxInterval: max,
-			},
-			min,
-		}, {
-			"test maximum",
-			args{
-				blockTime:   600000,
-				minInterval: min,
-				maxInterval: max,
-			},
-			max,
-		}, {
-			"test right blocktime",
-			args{
-				blockTime:   5000,
-				minInterval: min,
-				maxInterval: max,
-			},
-			5000 * time.Millisecond,
-		},
-	}
-	for _, tt := range tests {
-		t.Run(tt.name, func(t *testing.T) {
-			got := GetInterval(tt.args.blockTime, tt.args.minInterval, tt.args.maxInterval)
-			assert.EqualValues(t, tt.want, got)
-		})
-	}
-}
diff --git a/services/observer/parser/parser_test.go b/services/observer/parser/parser_test.go
index a6e87a3..e4a8e78 100644
--- a/services/observer/parser/parser_test.go
+++ b/services/observer/parser/parser_test.go
@@ -174,3 +174,50 @@ func TestGetTxBatches(t *testing.T) {
 	batches = getTxsBatches(txs, 5000, context.Background())
 	assert.Len(t, batches, 200)
 }
+
+func TestGetInterval(t *testing.T) {
+	min, _ := time.ParseDuration("2s")
+	max, _ := time.ParseDuration("30s")
+	type args struct {
+		blockTime   int
+		minInterval time.Duration
+		maxInterval time.Duration
+	}
+	tests := []struct {
+		name string
+		args args
+		want time.Duration
+	}{
+		{
+			"test minimum",
+			args{
+				blockTime:   100,
+				minInterval: min,
+				maxInterval: max,
+			},
+			min,
+		}, {
+			"test maximum",
+			args{
+				blockTime:   600000,
+				minInterval: min,
+				maxInterval: max,
+			},
+			max,
+		}, {
+			"test right blocktime",
+			args{
+				blockTime:   5000,
+				minInterval: min,
+				maxInterval: max,
+			},
+			5000 * time.Millisecond,
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			got := GetInterval(tt.args.blockTime, tt.args.minInterval, tt.args.maxInterval)
+			assert.EqualValues(t, tt.want, got)
+		})
+	}
+}
diff --git a/tests/integration/observer_test/full_flow_test.go b/tests/integration/observer_test/full_flow_test.go
index d859ab1..031d424 100644
--- a/tests/integration/observer_test/full_flow_test.go
+++ b/tests/integration/observer_test/full_flow_test.go
@@ -121,7 +121,7 @@ func ConsumerToTestTransactionsFull(delivery amqp.Delivery, t *testing.T, cancel
 
 	assert.Equal(t, notifier.TransactionNotification{
 		Action: blockatlas.TxNativeTokenTransfer,
-		Result: &blockatlas.Tx{
+		Result: blockatlas.Tx{
 			Type:      blockatlas.TxNativeTokenTransfer,
 			Direction: "incoming",
 			ID:        "95CF63FAA27579A9B6AF84EF8B2DFEAC29627479E9C98E7F5AE4535E213FA4C9",
@@ -147,7 +147,7 @@ func setupParserFull(stopChan chan<- struct{}) parser.Params {
 	maxTime := time.Second * 2
 	maxBatchBlocksAmount := 1
 
-	pollInterval := notifier.GetInterval(0, minTime, maxTime)
+	pollInterval := parser.GetInterval(0, minTime, maxTime)
 
 	backlogCount := 1
 
diff --git a/tests/integration/observer_test/notifier_test.go b/tests/integration/observer_test/notifier_test.go
index dda3052..1f64e5d 100644
--- a/tests/integration/observer_test/notifier_test.go
+++ b/tests/integration/observer_test/notifier_test.go
@@ -82,7 +82,7 @@ func ConsumerToTestTransactions(delivery amqp.Delivery, t *testing.T) {
 
 	assert.Equal(t, notifier.TransactionNotification{
 		Action: blockatlas.TxNativeTokenTransfer,
-		Result: &blockatlas.Tx{
+		Result: blockatlas.Tx{
 			Type:      blockatlas.TxNativeTokenTransfer,
 			Direction: "outgoing",
 			ID:        "95CF63FAA27579A9B6AF84EF8B2DFEAC29627479E9C98E7F5AE4535E213FA4C9",
diff --git a/tests/integration/observer_test/parser_test.go b/tests/integration/observer_test/parser_test.go
index a4f9a29..1c5d6b7 100644
--- a/tests/integration/observer_test/parser_test.go
+++ b/tests/integration/observer_test/parser_test.go
@@ -11,7 +11,6 @@ import (
 	"github.com/trustwallet/blockatlas/mq"
 	"github.com/trustwallet/blockatlas/pkg/blockatlas"
 	"github.com/trustwallet/blockatlas/pkg/logger"
-	"github.com/trustwallet/blockatlas/services/observer/notifier"
 	"github.com/trustwallet/blockatlas/services/observer/parser"
 	"github.com/trustwallet/blockatlas/tests/integration/setup"
 	"testing"
@@ -105,7 +104,7 @@ func setupParser(stopChan chan struct{}) parser.Params {
 	maxTime := time.Second * 2
 	maxBatchBlocksAmount := 100
 
-	pollInterval := notifier.GetInterval(0, minTime, maxTime)
+	pollInterval := parser.GetInterval(0, minTime, maxTime)
 
 	backlogCount := 50