Repository: trustwallet/blockatlas
Pull request #1113 (2020-05-20 11:49:00)

[BulkInsert] Use gorm-bulk-insert instead of BulkCreate

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

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

```diff
@@          Coverage Diff           @@
##           master   #1113   +/-   ##
======================================
- Coverage      38%     38%   -1%     
======================================
  Files         162     162           
  Lines        5571    5588   +17     
======================================
  Hits         2168    2168           
- Misses       3259    3276   +17     
  Partials      144     144           
```


| [Impacted Files](https://codecov.io/gh/trustwallet/blockatlas/pull/1113?src=pr&el=tree) | Coverage Δ | |
|---|---|---|
| [db/db.go](https://codecov.io/gh/trustwallet/blockatlas/pull/1113/diff?src=pr&el=tree#diff-ZGIvZGIuZ28=) | `0% <0%> (ø)` | |
| [db/subscriptions.go](https://codecov.io/gh/trustwallet/blockatlas/pull/1113/diff?src=pr&el=tree#diff-ZGIvc3Vic2NyaXB0aW9ucy5nbw==) | `12% <0%> (+1%)` | :arrow_up: |

------

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

EnoRage: https://github.com/trustwallet/blockatlas/issues/1107

Patch:
diff --git a/db/db.go b/db/db.go
index fe79c96..6d62678 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
@@ -60,3 +65,48 @@ func RestoreConnectionWorker(database *Instance, timeout time.Duration, uri stri
 		time.Sleep(timeout)
 	}
 }
+
+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 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
+}
+
+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
+}
diff --git a/db/models/subscriptions.go b/db/models/subscriptions.go
index 016e862..56eafb1 100644
--- a/db/models/subscriptions.go
+++ b/db/models/subscriptions.go
@@ -11,6 +11,6 @@ type Subscription struct {
 type SubscriptionData struct {
 	ID             uint   `gorm:"primary_key;"`
 	SubscriptionId uint   `gorm:"primary_key; column:subscription_id; auto_increment:false"`
-	Coin           uint   `gorm:"primary_key; column:coin; auto_increment:false"`
+	Coin           *uint  `gorm:"primary_key; column:coin; auto_increment:false"`
 	Address        string `gorm:"primary_key; column:address; type:varchar(128)"`
 }
diff --git a/db/subscriptions.go b/db/subscriptions.go
index 917fc40..e5c0c53 100644
--- a/db/subscriptions.go
+++ b/db/subscriptions.go
@@ -2,17 +2,13 @@ package db
 
 import (
 	"context"
-	"fmt"
 	"github.com/trustwallet/blockatlas/db/models"
 	"github.com/trustwallet/blockatlas/pkg/errors"
 	"go.elastic.co/apm/module/apmgorm"
 	"strconv"
-	"strings"
 	"time"
 )
 
-const rawBulkInsert = `INSERT INTO subscription_data(subscription_id, coin, address) VALUES %s ON CONFLICT DO NOTHING`
-
 func (i *Instance) GetSubscriptionData(coin uint, addresses []string, ctx context.Context) ([]models.SubscriptionData, error) {
 	if len(addresses) == 0 {
 		return nil, errors.E("Empty addresses")
@@ -65,7 +61,10 @@ func (i *Instance) AddSubscriptions(id uint, subscriptions []models.Subscription
 			txInstance.Gorm.Rollback()
 			return err
 		}
-		err = txInstance.BulkCreate(subscriptions)
+
+		db := txInstance.Gorm.Set("gorm:insert_option", "ON CONFLICT DO NOTHING")
+		err = bulkInsert(db, subscriptions)
+
 	} else {
 		err = txInstance.AddToExistingSubscription(id, subscriptions)
 	}
@@ -91,7 +90,8 @@ func (i *Instance) AddToExistingSubscription(id uint, subscriptions []models.Sub
 
 	updateList, deleteList := getSubscriptionsToDeleteAndUpdate(existingData, subscriptions)
 	if len(updateList) > 0 {
-		if err := i.BulkCreate(updateList); err != nil {
+		db := i.Gorm.Set("gorm:insert_option", "ON CONFLICT DO NOTHING")
+		if err := bulkInsert(db, updateList); err != nil {
 			return err
 		}
 	}
@@ -136,29 +136,6 @@ func (i *Instance) DeleteSubscriptions(subscriptions []models.SubscriptionData)
 	return nil
 }
 
-func (i *Instance) BulkCreate(dataList []models.SubscriptionData) error {
-	var (
-		valueStrings []string
-		valueArgs    []interface{}
-	)
-
-	for _, d := range dataList {
-		valueStrings = append(valueStrings, "(?, ?, ?)")
-
-		valueArgs = append(valueArgs, d.SubscriptionId)
-		valueArgs = append(valueArgs, d.Coin)
-		valueArgs = append(valueArgs, d.Address)
-	}
-
-	smt := fmt.Sprintf(rawBulkInsert, strings.Join(valueStrings, ","))
-
-	if err := i.Gorm.Exec(smt, valueArgs...).Error; err != nil {
-		return err
-	}
-
-	return nil
-}
-
 func getSubscriptionsToDeleteAndUpdate(existing, new []models.SubscriptionData) (subToUpdate, subToDelete []models.SubscriptionData) {
 	for _, n := range new {
 		if !containSubscription(n, existing) {
diff --git a/go.mod b/go.mod
index 443ac60..c00f0de 100644
--- a/go.mod
+++ b/go.mod
@@ -36,6 +36,7 @@ require (
 	github.com/stretchr/testify v1.5.1
 	github.com/swaggo/gin-swagger v1.2.0
 	github.com/swaggo/swag v1.6.5
+	github.com/t-tiger/gorm-bulk-insert v1.3.0
 	github.com/trustwallet/ens-coincodec v1.0.5
 	go.elastic.co/apm v1.8.0
 	go.elastic.co/apm/module/apmgin v1.8.0
diff --git a/go.sum b/go.sum
index 4d2b753..287d794 100644
--- a/go.sum
+++ b/go.sum
@@ -236,6 +236,7 @@ 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 h1:Drgk1clyWT9t9ERbzHza6Mj/8FY/CqMyVzOiHviMo6Q=
 github.com/jinzhu/gorm v1.9.12/go.mod h1:vhTjlKSJUTWNtcbQtrMBFCxy7eXTzeCAzfL5fBZT/Qs=
 github.com/jinzhu/inflection v1.0.0 h1:K317FqzuhWc8YvSVlFMCCUb36O/S9MCKRDI7QkRKD/E=
@@ -286,6 +287,7 @@ 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 v2.0.1+incompatible h1:xQ15muvnzGBHpIpdrNi1DA5x0+TcBZzsIDwmw9uTHzw=
 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=
@@ -419,6 +421,8 @@ github.com/swaggo/gin-swagger v1.2.0/go.mod h1:qlH2+W7zXGZkczuL+r2nEBR2JTT+/lX05
 github.com/swaggo/swag v1.5.1/go.mod h1:1Bl9F/ZBpVWh22nY0zmYyASPO1lI/zIwRDrpZU+tv8Y=
 github.com/swaggo/swag v1.6.5 h1:2C+t+xyK6p1sujqncYO/VnMvPZcBJjNdKKyxbOdAW8o=
 github.com/swaggo/swag v1.6.5/go.mod h1:Y7ZLSS0d0DdxhWGVhQdu+Bu1QhaF5k0RD7FKdiAykeY=
+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/ens-coincodec v1.0.5 h1:rdebDQYHqNjv+yUdtMf3DXyMgJpEJSI2DSbC1GUFuyk=
 github.com/trustwallet/ens-coincodec v1.0.5/go.mod h1:hTZCauB3S2Gt9M3FiRNva+xpLJXFqLyODKFF+UvH1lw=
diff --git a/services/observer/notifier/notifier.go b/services/observer/notifier/notifier.go
index 898242b..be4834d 100644
--- a/services/observer/notifier/notifier.go
+++ b/services/observer/notifier/notifier.go
@@ -61,7 +61,7 @@ func RunNotifier(database *db.Instance, delivery amqp.Delivery) {
 	for _, data := range subscriptionsDataList {
 		go buildAndPostMessage(
 			blockTransactions,
-			blockatlas.Subscription{Coin: data.Coin, Address: data.Address, Id: data.SubscriptionId},
+			blockatlas.Subscription{Coin: *data.Coin, Address: data.Address, Id: data.SubscriptionId},
 			&wg, ctx)
 	}
 	wg.Wait()
diff --git a/services/observer/subscriber/subscriber.go b/services/observer/subscriber/subscriber.go
index 1100b43..d5f9e88 100644
--- a/services/observer/subscriber/subscriber.go
+++ b/services/observer/subscriber/subscriber.go
@@ -57,7 +57,8 @@ func RunSubscriber(database *db.Instance, delivery amqp.Delivery) {
 func ToSubscriptionData(sub []blockatlas.Subscription) []models.SubscriptionData {
 	data := make([]models.SubscriptionData, 0, len(sub))
 	for _, s := range sub {
-		data = append(data, models.SubscriptionData{Coin: s.Coin, Address: s.Address, SubscriptionId: s.Id})
+		coin := s.Coin
+		data = append(data, models.SubscriptionData{Coin: &coin, Address: s.Address, SubscriptionId: s.Id})
 	}
 	return data
 }


Test patch:
diff --git a/db/subscritions_test.go b/db/subscritions_test.go
index 274ceec..8f15008 100644
--- a/db/subscritions_test.go
+++ b/db/subscritions_test.go
@@ -3,31 +3,32 @@ package db
 import (
 	"github.com/stretchr/testify/assert"
 	"github.com/trustwallet/blockatlas/db/models"
+	"github.com/trustwallet/blockatlas/tests/integration/setup/testdata"
 	"testing"
 )
 
 func TestGetSubscriptionsToDeleteAndUpdate(t *testing.T) {
 	oldSubscriptions := []models.SubscriptionData{{
 		SubscriptionId: 1,
-		Coin:           60,
+		Coin:           &testdata.EthCoin.ID,
 		Address:        "A",
 	}, {
 		SubscriptionId: 1,
-		Coin:           60,
+		Coin:           &testdata.EthCoin.ID,
 		Address:        "B",
 	}, {
 		SubscriptionId: 1,
-		Coin:           60,
+		Coin:           &testdata.EthCoin.ID,
 		Address:        "C",
 	}}
 
 	newSubscription := []models.SubscriptionData{{
 		SubscriptionId: 1,
-		Coin:           60,
+		Coin:           &testdata.EthCoin.ID,
 		Address:        "B",
 	}, {
 		SubscriptionId: 1,
-		Coin:           60,
+		Coin:           &testdata.EthCoin.ID,
 		Address:        "C",
 	}}
 
@@ -38,25 +39,25 @@ func TestGetSubscriptionsToDeleteAndUpdate(t *testing.T) {
 
 	oldSubscriptions = []models.SubscriptionData{{
 		SubscriptionId: 1,
-		Coin:           60,
+		Coin:           &testdata.EthCoin.ID,
 		Address:        "A",
 	}, {
 		SubscriptionId: 1,
-		Coin:           60,
+		Coin:           &testdata.EthCoin.ID,
 		Address:        "B",
 	}, {
 		SubscriptionId: 1,
-		Coin:           60,
+		Coin:           &testdata.EthCoin.ID,
 		Address:        "C",
 	}, {
 		SubscriptionId: 1,
-		Coin:           60,
+		Coin:           &testdata.EthCoin.ID,
 		Address:        "D",
 	}}
 
 	newSubscription = []models.SubscriptionData{{
 		SubscriptionId: 1,
-		Coin:           60,
+		Coin:           &testdata.EthCoin.ID,
 		Address:        "E",
 	}}
 
@@ -66,21 +67,21 @@ func TestGetSubscriptionsToDeleteAndUpdate(t *testing.T) {
 
 	oldSubscriptions = []models.SubscriptionData{{
 		SubscriptionId: 1,
-		Coin:           60,
+		Coin:           &testdata.EthCoin.ID,
 		Address:        "A",
 	}, {
 		SubscriptionId: 1,
-		Coin:           60,
+		Coin:           &testdata.EthCoin.ID,
 		Address:        "B",
 	}}
 
 	newSubscription = []models.SubscriptionData{{
 		SubscriptionId: 1,
-		Coin:           60,
+		Coin:           &testdata.EthCoin.ID,
 		Address:        "A",
 	}, {
 		SubscriptionId: 1,
-		Coin:           60,
+		Coin:           &testdata.EthCoin.ID,
 		Address:        "B",
 	}}
 
diff --git a/services/observer/subscriber/subscriber_test.go b/services/observer/subscriber/subscriber_test.go
index 713b879..b8aefe5 100644
--- a/services/observer/subscriber/subscriber_test.go
+++ b/services/observer/subscriber/subscriber_test.go
@@ -4,29 +4,30 @@ import (
 	"github.com/stretchr/testify/assert"
 	"github.com/trustwallet/blockatlas/db/models"
 	"github.com/trustwallet/blockatlas/pkg/blockatlas"
+	"github.com/trustwallet/blockatlas/tests/integration/setup/testdata"
 	"testing"
 )
 
 func TestToSubscriptionData(t *testing.T) {
 	sub := blockatlas.Subscription{
-		Coin:    60,
+		Coin:    testdata.EthCoin.ID,
 		Address: "A",
 		Id:      1,
 	}
 	sub2 := blockatlas.Subscription{
-		Coin:    60,
+		Coin:    testdata.EthCoin.ID,
 		Address: "B",
 		Id:      2,
 	}
 
 	expectedModel := models.SubscriptionData{
 		SubscriptionId: 1,
-		Coin:           60,
+		Coin:           &testdata.EthCoin.ID,
 		Address:        "A",
 	}
 	expectedModel1 := models.SubscriptionData{
 		SubscriptionId: 2,
-		Coin:           60,
+		Coin:           &testdata.EthCoin.ID,
 		Address:        "B",
 	}
 
diff --git a/tests/integration/db_test/subscriptions_test.go b/tests/integration/db_test/subscriptions_test.go
index 0cafda6..27ef7db 100644
--- a/tests/integration/db_test/subscriptions_test.go
+++ b/tests/integration/db_test/subscriptions_test.go
@@ -9,6 +9,8 @@ import (
 	"github.com/trustwallet/blockatlas/pkg/blockatlas"
 	"github.com/trustwallet/blockatlas/services/observer/subscriber"
 	"github.com/trustwallet/blockatlas/tests/integration/setup"
+	"github.com/trustwallet/blockatlas/tests/integration/setup/testdata"
+	"sort"
 	"testing"
 	"time"
 )
@@ -20,9 +22,10 @@ func TestDb_AddSubscriptionsBulk(t *testing.T) {
 	id := uint(1)
 
 	for i := 0; i < 100; i++ {
+		coin := uint(i)
 		subscriptions = append(subscriptions, models.SubscriptionData{
 			SubscriptionId: id,
-			Coin:           uint(i),
+			Coin:           &coin,
 			Address:        "testAddrtestAddrtestAddrtestAddrtestAddrtestAddrtestAddrtestAddrtestAddrtestAddr",
 		})
 	}
@@ -41,21 +44,25 @@ func TestDb_AddSubscriptions(t *testing.T) {
 	var subscriptions []models.SubscriptionData
 
 	id := uint(1)
+
+	c1 := uint(60)
+	c2 := uint(61)
+	c3 := uint(62)
 	subscriptions = append(subscriptions, models.SubscriptionData{
 		SubscriptionId: id,
-		Coin:           60,
+		Coin:           &c1,
 		Address:        "testAddr",
 	})
 
 	subscriptions = append(subscriptions, models.SubscriptionData{
 		SubscriptionId: id,
-		Coin:           61,
+		Coin:           &c2,
 		Address:        "testAddr2",
 	})
 
 	subscriptions = append(subscriptions, models.SubscriptionData{
 		SubscriptionId: id,
-		Coin:           62,
+		Coin:           &c3,
 		Address:        "testAddr3",
 	})
 
@@ -93,19 +100,19 @@ func TestDb_AddSubscriptionsWithRewrite(t *testing.T) {
 	var subscriptions []models.SubscriptionData
 	subscriptions = append(subscriptions, models.SubscriptionData{
 		SubscriptionId: id,
-		Coin:           60,
+		Coin:           &testdata.EthCoin.ID,
 		Address:        "testAddr",
 	})
 
 	subscriptions = append(subscriptions, models.SubscriptionData{
 		SubscriptionId: id,
-		Coin:           714,
+		Coin:           &testdata.BnbCoin.ID,
 		Address:        "testAddr",
 	})
 
 	subscriptions = append(subscriptions, models.SubscriptionData{
 		SubscriptionId: id,
-		Coin:           144,
+		Coin:           &testdata.XrpCoin.ID,
 		Address:        "testAddr",
 	})
 
@@ -137,19 +144,19 @@ func TestDb_AddSubscriptionsWithRewrite(t *testing.T) {
 
 	subscriptions = append(subscriptions, models.SubscriptionData{
 		SubscriptionId: id,
-		Coin:           60,
+		Coin:           &testdata.EthCoin.ID,
 		Address:        "testAddr2",
 	})
 
 	subscriptions = append(subscriptions, models.SubscriptionData{
 		SubscriptionId: id,
-		Coin:           714,
+		Coin:           &testdata.BnbCoin.ID,
 		Address:        "testAddr2",
 	})
 
 	subscriptions = append(subscriptions, models.SubscriptionData{
 		SubscriptionId: id,
-		Coin:           144,
+		Coin:           &testdata.XrpCoin.ID,
 		Address:        "testAddr2",
 	})
 
@@ -256,19 +263,19 @@ func TestDb_DeleteSubscriptions(t *testing.T) {
 	id := uint(1)
 	subscriptions = append(subscriptions, models.SubscriptionData{
 		SubscriptionId: id,
-		Coin:           60,
+		Coin:           &testdata.EthCoin.ID,
 		Address:        "testAddr",
 	})
 
 	subscriptions = append(subscriptions, models.SubscriptionData{
 		SubscriptionId: id,
-		Coin:           714,
+		Coin:           &testdata.BnbCoin.ID,
 		Address:        "testAddr2",
 	})
 
 	subscriptions = append(subscriptions, models.SubscriptionData{
 		SubscriptionId: id,
-		Coin:           144,
+		Coin:           &testdata.XrpCoin.ID,
 		Address:        "testAddr3",
 	})
 
@@ -331,19 +338,19 @@ func TestDeleteAll(t *testing.T) {
 	id := uint(1)
 	subscriptions = append(subscriptions, models.SubscriptionData{
 		SubscriptionId: id,
-		Coin:           60,
+		Coin:           &testdata.EthCoin.ID,
 		Address:        "testAddr",
 	})
 
 	subscriptions = append(subscriptions, models.SubscriptionData{
 		SubscriptionId: id,
-		Coin:           714,
+		Coin:           &testdata.BnbCoin.ID,
 		Address:        "testAddr2",
 	})
 
 	subscriptions = append(subscriptions, models.SubscriptionData{
 		SubscriptionId: id,
-		Coin:           144,
+		Coin:           &testdata.XrpCoin.ID,
 		Address:        "testAddr3",
 	})
 	assert.Nil(t, database.AddSubscriptions(id, subscriptions, context.Background()))
@@ -390,7 +397,7 @@ func TestDb_DuplicateEntries(t *testing.T) {
 	for i := 0; i < 10; i++ {
 		subscriptions = append(subscriptions, models.SubscriptionData{
 			SubscriptionId: id,
-			Coin:           60,
+			Coin:           &testdata.EthCoin.ID,
 			Address:        "testAddr",
 		})
 	}
@@ -407,8 +414,9 @@ func TestDb_DuplicateEntries(t *testing.T) {
 func TestDb_FindSubscriptions_Multiple(t *testing.T) {
 	setup.CleanupPgContainer(database.Gorm)
 	var subscriptions []models.SubscriptionData
+
 	subscriptions = append(subscriptions, models.SubscriptionData{
-		Coin:    60,
+		Coin:    &testdata.EthCoin.ID,
 		Address: "testAddr",
 	})
 
@@ -424,6 +432,10 @@ func TestDb_FindSubscriptions_Multiple(t *testing.T) {
 	assert.Nil(t, err)
 	assert.Equal(t, 5, len(subs))
 
+	sort.Slice(subs, func(i, j int) bool {
+		return subs[i].SubscriptionId < subs[j].SubscriptionId
+	})
+
 	for i := 0; i < 5; i++ {
 		assert.Equal(t, uint(i)+1, subs[i].SubscriptionId)
 	}
@@ -433,7 +445,7 @@ func TestDb_AddToExisting(t *testing.T) {
 	setup.CleanupPgContainer(database.Gorm)
 	var subscriptions []models.SubscriptionData
 	subscriptions = append(subscriptions, models.SubscriptionData{
-		Coin:    60,
+		Coin:    &testdata.EthCoin.ID,
 		Address: "testAddr",
 	})
 
@@ -467,11 +479,11 @@ func TestDb_UpdatedAt(t *testing.T) {
 	setup.CleanupPgContainer(database.Gorm)
 	var subscriptions []models.SubscriptionData
 	subscriptions = append(subscriptions, models.SubscriptionData{
-		Coin:    60,
-		Address: "testAddr",
+		SubscriptionId: 1,
+		Coin:           &testdata.EthCoin.ID,
+		Address:        "testAddr",
 	})
 
-	subscriptions[0].SubscriptionId = uint(1)
 	assert.Nil(t, database.AddSubscriptions(uint(1), subscriptions, context.Background()))
 	subs, err := database.GetSubscriptionData(60, []string{"testAddr"}, context.Background())
 	assert.Nil(t, err)
@@ -485,8 +497,9 @@ func TestDb_UpdatedAt(t *testing.T) {
 	assert.Greater(t, existingSub.UpdatedAt.Unix(), time.Now().Unix()-120)
 
 	subscriptions = append(subscriptions, models.SubscriptionData{
-		Coin:    714,
-		Address: "newtestAddr",
+		SubscriptionId: 1,
+		Coin:           &testdata.BnbCoin.ID,
+		Address:        "newtestAddr",
 	})
 
 	assert.Nil(t, database.AddSubscriptions(uint(1), subscriptions, context.Background()))
@@ -499,12 +512,11 @@ func TestDb_UpdatedAt(t *testing.T) {
 	assert.Greater(t, time.Now().Unix(), existingSub2.UpdatedAt.Unix())
 	assert.Greater(t, existingSub2.UpdatedAt.Unix(), time.Now().Unix()-120)
 	assert.Greater(t, existingSub2.UpdatedAt.Unix(), existingSub.UpdatedAt.Unix())
-
 }
 
 func containSub(sub models.SubscriptionData, list []models.SubscriptionData) bool {
 	for _, s := range list {
-		if sub.Address == s.Address && sub.Coin == s.Coin && sub.SubscriptionId == s.SubscriptionId {
+		if sub.Address == s.Address && *sub.Coin == *s.Coin && sub.SubscriptionId == s.SubscriptionId {
 			return true
 		}
 	}
diff --git a/tests/integration/observer_test/full_flow_test.go b/tests/integration/observer_test/full_flow_test.go
index 29cce3a..24b4173 100644
--- a/tests/integration/observer_test/full_flow_test.go
+++ b/tests/integration/observer_test/full_flow_test.go
@@ -14,6 +14,7 @@ import (
 	"github.com/trustwallet/blockatlas/services/observer/notifier"
 	"github.com/trustwallet/blockatlas/services/observer/parser"
 	"github.com/trustwallet/blockatlas/tests/integration/setup"
+	"github.com/trustwallet/blockatlas/tests/integration/setup/testdata"
 	"go.uber.org/atomic"
 	"testing"
 	"time"
@@ -25,7 +26,8 @@ var (
 
 func TestFullFlow(t *testing.T) {
 	setup.CleanupPgContainer(database.Gorm)
-	err := database.AddSubscriptions(1, []models.SubscriptionData{{Coin: 60, Address: "testAddress", SubscriptionId: 1}}, context.Background())
+
+	err := database.AddSubscriptions(1, []models.SubscriptionData{{Coin: &testdata.EthCoin.ID, Address: "testAddress", SubscriptionId: 1}}, context.Background())
 	assert.Nil(t, err)
 
 	ctx, cancel := context.WithCancel(context.Background())
diff --git a/tests/integration/observer_test/notifier_test.go b/tests/integration/observer_test/notifier_test.go
index 0f6a1ab..9203544 100644
--- a/tests/integration/observer_test/notifier_test.go
+++ b/tests/integration/observer_test/notifier_test.go
@@ -13,6 +13,7 @@ import (
 	"github.com/trustwallet/blockatlas/pkg/blockatlas"
 	"github.com/trustwallet/blockatlas/services/observer/notifier"
 	"github.com/trustwallet/blockatlas/tests/integration/setup"
+	"github.com/trustwallet/blockatlas/tests/integration/setup/testdata"
 	"testing"
 	"time"
 )
@@ -44,7 +45,7 @@ var (
 func TestNotifier(t *testing.T) {
 	setup.CleanupPgContainer(database.Gorm)
 
-	err := database.AddSubscriptions(1, []models.SubscriptionData{{Coin: 714, Address: "tbnb1ttyn4csghfgyxreu7lmdu3lcplhqhxtzced45a", SubscriptionId: 1}}, context.Background())
+	err := database.AddSubscriptions(1, []models.SubscriptionData{{Coin: &testdata.BnbCoin.ID, Address: "tbnb1ttyn4csghfgyxreu7lmdu3lcplhqhxtzced45a", SubscriptionId: 1}}, context.Background())
 	assert.Nil(t, err)
 
 	err = produceTxs(txs)
diff --git a/tests/integration/observer_test/subscriber_test.go b/tests/integration/observer_test/subscriber_test.go
index b7e9ac1..80eb667 100644
--- a/tests/integration/observer_test/subscriber_test.go
+++ b/tests/integration/observer_test/subscriber_test.go
@@ -60,7 +60,7 @@ func TestSubscriberAddSubscription(t *testing.T) {
 		result, err := database.GetSubscriptionData(wanted.Coin, []string{wanted.Address}, context.Background())
 		assert.Nil(t, err)
 		assert.Equal(t, result[0].SubscriptionId, wanted.Id)
-		assert.Equal(t, result[0].Coin, wanted.Coin)
+		assert.Equal(t, wanted.Coin, *result[0].Coin)
 		assert.Equal(t, result[0].Address, wanted.Address)
 	}
 }
@@ -91,18 +91,21 @@ func TestSubscriber_UpdateSubscription(t *testing.T) {
 	if err := json.Unmarshal(testFileWanted, &wantedEvents); err != nil {
 		t.Fatal(err)
 	}
-
+	c1 := uint(61)
+	c2 := uint(62)
+	c3 := uint(63)
+	c4 := uint(64)
 	database.AddSubscriptions(10, []models.SubscriptionData{
-		{Coin: 61, Address: "0x0000000000000000000000000000000000000000", SubscriptionId: 10},
+		{Coin: &c1, Address: "0x0000000000000000000000000000000000000000", SubscriptionId: 10},
 	}, context.Background())
 	database.AddSubscriptions(1, []models.SubscriptionData{
-		{Coin: 62, Address: "0x0000000000000000000000000000000000000000", SubscriptionId: 1},
+		{Coin: &c2, Address: "0x0000000000000000000000000000000000000000", SubscriptionId: 1},
 	}, context.Background())
 	database.AddSubscriptions(2, []models.SubscriptionData{
-		{Coin: 63, Address: "0x0000000000000000000000000000000000000000", SubscriptionId: 2},
+		{Coin: &c3, Address: "0x0000000000000000000000000000000000000000", SubscriptionId: 2},
 	}, context.Background())
 	database.AddSubscriptions(3, []models.SubscriptionData{
-		{Coin: 64, Address: "0x0000000000000000000000000000000000000000", SubscriptionId: 3},
+		{Coin: &c4, Address: "0x0000000000000000000000000000000000000000", SubscriptionId: 3},
 	}, context.Background())
 
 	for _, event := range givenEvents {
@@ -123,7 +126,7 @@ func TestSubscriber_UpdateSubscription(t *testing.T) {
 		result, err := database.GetSubscriptionData(wanted.Coin, []string{wanted.Address}, context.Background())
 		assert.Nil(t, err)
 		assert.Equal(t, result[0].SubscriptionId, wanted.Id)
-		assert.Equal(t, result[0].Coin, wanted.Coin)
+		assert.Equal(t, wanted.Coin, *result[0].Coin)
 		assert.Equal(t, result[0].Address, wanted.Address)
 
 	}
diff --git a/tests/integration/setup/testdata/coins.go b/tests/integration/setup/testdata/coins.go
new file mode 100644
index 0000000..610fd16
--- /dev/null
+++ b/tests/integration/setup/testdata/coins.go
@@ -0,0 +1,9 @@
+package testdata
+
+import "github.com/trustwallet/blockatlas/coin"
+
+var (
+	EthCoin = coin.Ethereum()
+	BnbCoin = coin.Binance()
+	XrpCoin = coin.Ripple()
+)