Merge pull request #9 from Zondax/ezequiel/refactor

Code Refactor

Author: raynaudoe

.circleci/config.yml

@@ -3,7 +3,7 @@ version: 2.1
 jobs:
   build:
     docker:
-      - image: golang:1.15
+      - image: golang:1.16
     working_directory: /project
     steps:
       - checkout
@@ -12,7 +12,7 @@ jobs:
 
   checks:
     docker:
-      - image: golang:1.15
+      - image: golang:1.16
     steps:
       - checkout
       - run: make install_lint
@@ -22,7 +22,7 @@ jobs:
 
   integration:
     docker:
-      - image: golang:1.15
+      - image: golang:1.16
     steps:
       - checkout
       - run:

connections/data_source.go

@@ -0,0 +1,82 @@
+package connections
+
+import (
+	"context"
+	"fmt"
+	ds "github.com/Zondax/zindexer/connections/data_store"
+	"github.com/coinbase/rosetta-sdk-go/client"
+	"gorm.io/gorm"
+	"time"
+)
+
+const (
+	defaultRetryDelay = 30 * time.Second
+)
+
+type DataSource struct {
+	// data sources
+	DbConn        *gorm.DB
+	RosettaClient *client.APIClient
+	NodeClient    interface{}
+	DataStore     ds.DataStoreClient
+	// common
+	Ctx        context.Context
+	RetryDelay time.Duration
+}
+
+type SourceOption func(*DataSource)
+
+func NewDataSource(opts ...SourceOption) DataSource {
+	d := DataSource{
+		Ctx:        context.Background(),
+		RetryDelay: defaultRetryDelay,
+	}
+	for _, opt := range opts {
+		opt(&d)
+	}
+
+	return d
+}
+
+func WithContext(ctx context.Context) SourceOption {
+	return func(w *DataSource) {
+		w.Ctx = ctx
+	}
+}
+
+func WithRetryDelay(delay time.Duration) SourceOption {
+	return func(w *DataSource) {
+		w.RetryDelay = delay
+	}
+}
+
+func WithDBConnection(dbConn *gorm.DB) SourceOption {
+	return func(w *DataSource) {
+		w.DbConn = dbConn
+	}
+}
+
+func WithRosettaClient(client *client.APIClient) SourceOption {
+	return func(w *DataSource) {
+		w.RosettaClient = client
+	}
+}
+
+func WithNodeClient(node interface{}) SourceOption {
+	return func(w *DataSource) {
+		w.NodeClient = node
+	}
+}
+
+func WithDataStore(cfg ds.DataStoreConfig) SourceOption {
+	return func(w *DataSource) {
+		switch cfg.Service {
+		case ds.MinIOStorage:
+			c, _ := ds.NewMinioClient(cfg)
+			w.DataStore = ds.DataStoreClient{Client: c}
+			return
+		default:
+			panic(fmt.Errorf("DataStore with service %s, is not available", cfg.Service))
+		}
+	}
+}

connections/data_store/interface.go

@@ -0,0 +1,18 @@
+package data_store
+
+type IDataStoreClient interface {
+	GetFile(name string, location string) (*[]byte, error)
+	UploadFile(name string, dest string) error
+	List(dir string, prefix string) *[]string
+}
+
+type DataStoreClient struct {
+	Client IDataStoreClient
+}
+
+type DataStoreConfig struct {
+	Url      string
+	User     string
+	Password string
+	Service  string
+}

connections/data_store/minio.go

@@ -0,0 +1,78 @@
+package data_store
+
+import (
+	"bytes"
+	"context"
+	"github.com/minio/minio-go/v7"
+	"github.com/minio/minio-go/v7/pkg/credentials"
+	"go.uber.org/zap"
+	"os"
+)
+
+const MinIOStorage = "minio"
+
+type MinioClient struct {
+	client *minio.Client
+}
+
+func NewMinioClient(config DataStoreConfig) (MinioClient, error) {
+	minioClient, err := minio.New(config.Url, &minio.Options{
+		Creds:  credentials.NewStaticV4(config.User, config.Password, ""),
+		Secure: true,
+	})
+	if err != nil {
+		zap.S().Error(err.Error())
+		return MinioClient{}, err
+	}
+
+	return MinioClient{client: minioClient}, nil
+}
+
+func (c MinioClient) GetClient() *minio.Client {
+	return c.client
+}
+
+func (c MinioClient) GetFile(object string, bucket string) (*[]byte, error) {
+	obj, err := c.GetClient().GetObject(context.Background(), bucket, object, minio.GetObjectOptions{})
+	if err != nil {
+		return nil, err
+	}
+
+	buf := new(bytes.Buffer)
+	_, err = buf.ReadFrom(obj)
+	if err != nil {
+		return nil, err
+	}
+
+	data := buf.Bytes()
+	return &data, nil
+}
+
+func (c MinioClient) List(bucket string, prefix string) *[]string {
+	var list []string
+	for object := range c.client.ListObjects(context.Background(), bucket, minio.ListObjectsOptions{Prefix: prefix}) {
+		list = append(list, object.Key)
+	}
+
+	return &list
+}
+
+func (c MinioClient) UploadFile(name string, dest string) error {
+	file, err := os.Open(name)
+	if err != nil {
+		return err
+	}
+	defer file.Close()
+
+	fileStat, err := file.Stat()
+	if err != nil {
+		return err
+	}
+
+	_, err = c.client.PutObject(context.Background(), dest, file.Name(), file,
+		fileStat.Size(), minio.PutObjectOptions{ContentType: "application/octet-stream"})
+	if err != nil {
+		return err
+	}
+	return nil
+}

connections/graphql.go connections/database/graphql.goconnections/graphql.go renamed to connections/database/graphql.go

@@ -1,4 +1,4 @@
-package connections
+package database
 
 import (
 	"context"

config.go connections/database/interface.goconfig.go renamed to connections/database/interface.go

@@ -1,10 +1,23 @@
-package zindexer
+package database
 
 import (
 	"fmt"
 	"gorm.io/gorm"
 )
 
+type DBConnection interface {
+	GetDB() *gorm.DB
+}
+
+type IDBQueryClient interface {
+	Connect() error
+	GetDB() interface{}
+}
+
+type DBQueryClient struct {
+	Client IDBQueryClient
+}
+
 type DBConnectionParams struct {
 	User     string
 	Password string
@@ -18,10 +31,6 @@ type GraphqlClientParams struct {
 	Token string
 }
 
-type DBConnectionConfig struct {
-	Gorm *gorm.Config
-}
-
 func (p *DBConnectionParams) GetDSN() (string, error) {
 	return fmt.Sprintf(
 		"user=%s password=%s dbname=%s host=%s port=%s sslmode=disable",

connections/postgres.go connections/database/postgres.goconnections/postgres.go renamed to connections/database/postgres.go

@@ -1,13 +1,20 @@
-package connections
+package database
 
 import (
 	"fmt"
-	"github.com/Zondax/zindexer"
 	"gorm.io/driver/postgres"
 	"gorm.io/gorm"
 )
 
-func NewPostgresConnection(params *zindexer.DBConnectionParams, config *zindexer.DBConnectionConfig) (*GormConnection, error) {
+type GormConnection struct {
+	db *gorm.DB
+}
+
+type DBConnectionConfig struct {
+	Gorm *gorm.Config
+}
+
+func NewPostgresConnection(params *DBConnectionParams, config *DBConnectionConfig) (*GormConnection, error) {
 	dsn, err := params.GetDSN()
 	if err != nil {
 		return nil, fmt.Errorf("failed to retrieve dsn")
@@ -28,7 +35,7 @@ func (c *GormConnection) GetDB() *gorm.DB {
 	return c.db
 }
 
-func ConnectDB(params zindexer.DBConnectionParams, config zindexer.DBConnectionConfig) (*gorm.DB, error) {
+func ConnectDB(params DBConnectionParams, config DBConnectionConfig) (*gorm.DB, error) {
 	conn, err := NewPostgresConnection(&params, &config)
 	if err != nil {
 		return nil, err

connections/pub_sub/interface.go

@@ -0,0 +1,26 @@
+package pub_sub
+
+import (
+	"encoding/json"
+	"github.com/ThreeDotsLabs/watermill/message"
+)
+
+type IDBSubscriptionClient interface {
+	Subscribe(query interface{}, handler func(message *json.RawMessage, err error) error) error
+	Unsubscribe() error
+	Start() error
+	Stop() error
+}
+
+type ITopicPubSubClient interface {
+	Subscribe(string, func(messages <-chan *message.Message)) error
+	Publish(string, *message.Message) error
+}
+
+type DBSubscriptionClient struct {
+	Client IDBSubscriptionClient
+}
+
+type TopicPubSubClient struct {
+	Client ITopicPubSubClient
+}

connections/pubsub_goChannel.go connections/pub_sub/pubsub_goChannel.goconnections/pubsub_goChannel.go renamed to connections/pub_sub/pubsub_goChannel.go

@@ -1,4 +1,4 @@
-package connections
+package pub_sub
 
 import (
 	"context"