Export IPLD encoders

indexer/constructor.go

@@ -70,12 +70,12 @@ func NewStateDiffIndexer(
 		var driver sql.Driver
 		switch pgc.Driver {
 		case postgres.PGX:
-			driver, err = postgres.ConnectPGXDriver(ctx, pgc, nodeInfo)
+			driver, err = postgres.NewPGXDriver(ctx, pgc, nodeInfo)
 			if err != nil {
 				return nil, nil, err
 			}
 		case postgres.SQLX:
-			driver, err = postgres.ConnectSQLXDriver(ctx, pgc, nodeInfo)
+			driver, err = postgres.NewSQLXDriver(ctx, pgc, nodeInfo)
 			if err != nil {
 				return nil, nil, err
 			}

indexer/database/file/csv_indexer_legacy_test.go

@@ -67,7 +67,7 @@ func dumpCSVFileData(t *testing.T) {
 
 	localOutputDir := filepath.Join(workingDir, file.CSVTestConfig.OutputDir)
 
-	for _, tbl := range schema.EthTables {
+	for _, tbl := range schema.Tables {
 		err := test_helpers.DedupFile(file.TableFilePath(localOutputDir, tbl.Name))
 		require.NoError(t, err)
 

indexer/database/file/csv_writer.go

@@ -120,7 +120,7 @@ func NewCSVWriter(path string, watchedAddressesFilePath string, diff bool) (*CSV
 		return nil, fmt.Errorf("unable to create directory '%s': %w", path, err)
 	}
 
-	writers, err := makeFileWriters(path, schema.EthTables)
+	writers, err := makeFileWriters(path, schema.Tables)
 	if err != nil {
 		return nil, err
 	}

indexer/database/sql/postgres/pgx.go

@@ -37,6 +37,7 @@ type PGXDriver struct {
 	ctx      context.Context
 	pool     *pgxpool.Pool
 	nodeInfo node.Info
+	nodeID   string
 	config   Config
 }
 
@@ -49,25 +50,21 @@ func ConnectPGX(ctx context.Context, config Config) (*pgxpool.Pool, error) {
 	return pgxpool.ConnectConfig(ctx, pgConf)
 }
 
-// ConnectPGXDriver returns a new pgx driver
+// NewPGXDriver returns a new pgx driver
 // it initializes the connection pool and creates the node info table
-func ConnectPGXDriver(ctx context.Context, config Config, node node.Info) (*PGXDriver, error) {
+func NewPGXDriver(ctx context.Context, config Config, node node.Info) (*PGXDriver, error) {
 	dbPool, err := ConnectPGX(ctx, config)
 	if err != nil {
 		return nil, ErrDBConnectionFailed(err)
 	}
-	pg := NewPGXDriver(ctx, dbPool, config)
-	nodeErr := pg.createNode(node)
+	pg := &PGXDriver{ctx: ctx, pool: dbPool, nodeInfo: node, config: config}
+	nodeErr := pg.createNode()
 	if nodeErr != nil {
 		return &PGXDriver{}, ErrUnableToSetNode(nodeErr)
 	}
 	return pg, nil
 }
 
-func NewPGXDriver(ctx context.Context, pool *pgxpool.Pool, config Config) *PGXDriver {
-	return &PGXDriver{ctx: ctx, pool: pool, config: config}
-}
-
 // MakeConfig creates a pgxpool.Config from the provided Config
 func MakeConfig(config Config) (*pgxpool.Config, error) {
 	conf, err := pgxpool.ParseConfig("")
@@ -105,19 +102,19 @@ func MakeConfig(config Config) (*pgxpool.Config, error) {
 	return conf, nil
 }
 
-func (pgx *PGXDriver) createNode(nodeInfo node.Info) error {
+func (pgx *PGXDriver) createNode() error {
 	_, err := pgx.pool.Exec(
 		pgx.ctx,
 		createNodeStm,
-		nodeInfo.GenesisBlock,
-		nodeInfo.NetworkID,
-		nodeInfo.ID,
-		nodeInfo.ClientName,
-		nodeInfo.ChainID)
+		pgx.nodeInfo.GenesisBlock,
+		pgx.nodeInfo.NetworkID,
+		pgx.nodeInfo.ID,
+		pgx.nodeInfo.ClientName,
+		pgx.nodeInfo.ChainID)
 	if err != nil {
 		return ErrUnableToSetNode(err)
 	}
-	pgx.nodeInfo = nodeInfo
+	pgx.nodeID = pgx.nodeInfo.ID
 	return nil
 }
 
@@ -158,7 +155,7 @@ func (pgx *PGXDriver) Stats() metrics.DbStats {
 
 // NodeID satisfies sql.Database
 func (pgx *PGXDriver) NodeID() string {
-	return pgx.nodeInfo.ID
+	return pgx.nodeID
 }
 
 // Close satisfies sql.Database/io.Closer

indexer/database/sql/postgres/pgx_test.go

@@ -94,7 +94,7 @@ func TestPostgresPGX(t *testing.T) {
 
 	t.Run("throws error when can't connect to the database", func(t *testing.T) {
 		goodInfo := node.Info{GenesisBlock: "GENESIS", NetworkID: "1", ID: "x123", ClientName: "geth"}
-		_, err := postgres.ConnectPGXDriver(ctx, postgres.Config{}, goodInfo)
+		_, err := postgres.NewPGXDriver(ctx, postgres.Config{}, goodInfo)
 		if err == nil {
 			t.Fatal("Expected an error")
 		}
@@ -106,7 +106,7 @@ func TestPostgresPGX(t *testing.T) {
 		badHash := fmt.Sprintf("x %s", strings.Repeat("1", 100))
 		badInfo := node.Info{GenesisBlock: badHash, NetworkID: "1", ID: "x123", ClientName: "geth"}
 
-		_, err := postgres.ConnectPGXDriver(ctx, pgConfig, badInfo)
+		_, err := postgres.NewPGXDriver(ctx, pgConfig, badInfo)
 		if err == nil {
 			t.Fatal("Expected an error")
 		}

indexer/database/sql/postgres/sqlx.go

@@ -34,6 +34,7 @@ type SQLXDriver struct {
 	ctx      context.Context
 	db       *sqlx.DB
 	nodeInfo node.Info
+	nodeID   string
 }
 
 // ConnectSQLX initializes and returns a SQLX connection pool for postgres
@@ -52,36 +53,32 @@ func ConnectSQLX(ctx context.Context, config Config) (*sqlx.DB, error) {
 	return db, nil
 }
 
-// ConnectSQLXDriver returns a new sqlx driver for Postgres
+// NewSQLXDriver returns a new sqlx driver for Postgres
 // it initializes the connection pool and creates the node info table
-func ConnectSQLXDriver(ctx context.Context, config Config, node node.Info) (*SQLXDriver, error) {
+func NewSQLXDriver(ctx context.Context, config Config, node node.Info) (*SQLXDriver, error) {
 	db, err := ConnectSQLX(ctx, config)
 	if err != nil {
 		return nil, err
 	}
-	driver := NewSQLXDriver(ctx, db)
-	if err := driver.createNode(node); err != nil {
+	driver := &SQLXDriver{ctx: ctx, db: db, nodeInfo: node}
+	if err := driver.createNode(); err != nil {
 		return nil, err
 	}
 	return driver, nil
 }
 
-func NewSQLXDriver(ctx context.Context, db *sqlx.DB) *SQLXDriver {
-	return &SQLXDriver{ctx: ctx, db: db}
-}
-
-func (driver *SQLXDriver) createNode(nodeInfo node.Info) error {
+func (driver *SQLXDriver) createNode() error {
 	_, err := driver.db.Exec(
 		createNodeStm,
-		nodeInfo.GenesisBlock,
-		nodeInfo.NetworkID,
-		nodeInfo.ID,
-		nodeInfo.ClientName,
-		nodeInfo.ChainID)
+		driver.nodeInfo.GenesisBlock,
+		driver.nodeInfo.NetworkID,
+		driver.nodeInfo.ID,
+		driver.nodeInfo.ClientName,
+		driver.nodeInfo.ChainID)
 	if err != nil {
 		return ErrUnableToSetNode(err)
 	}
-	driver.nodeInfo = nodeInfo
+	driver.nodeID = driver.nodeInfo.ID
 	return nil
 }
 
@@ -121,7 +118,7 @@ func (driver *SQLXDriver) Stats() metrics.DbStats {
 
 // NodeID satisfies sql.Database
 func (driver *SQLXDriver) NodeID() string {
-	return driver.nodeInfo.ID
+	return driver.nodeID
 }
 
 // Close satisfies sql.Database/io.Closer

indexer/database/sql/postgres/sqlx_test.go

@@ -97,7 +97,7 @@ func TestPostgresSQLX(t *testing.T) {
 
 	t.Run("throws error when can't connect to the database", func(t *testing.T) {
 		goodInfo := node.Info{GenesisBlock: "GENESIS", NetworkID: "1", ID: "x123", ClientName: "geth"}
-		_, err := postgres.ConnectSQLXDriver(ctx, postgres.Config{}, goodInfo)
+		_, err := postgres.NewSQLXDriver(ctx, postgres.Config{}, goodInfo)
 		if err == nil {
 			t.Fatal("Expected an error")
 		}
@@ -109,7 +109,7 @@ func TestPostgresSQLX(t *testing.T) {
 		badHash := fmt.Sprintf("x %s", strings.Repeat("1", 100))
 		badInfo := node.Info{GenesisBlock: badHash, NetworkID: "1", ID: "x123", ClientName: "geth"}
 
-		_, err := postgres.ConnectSQLXDriver(ctx, pgConfig, badInfo)
+		_, err := postgres.NewSQLXDriver(ctx, pgConfig, badInfo)
 		if err == nil {
 			t.Fatal("Expected an error")
 		}

indexer/database/sql/postgres/test_helpers.go

@@ -30,7 +30,7 @@ func SetupSQLXDB() (sql.Database, error) {
 		return nil, err
 	}
 	conf.MaxIdle = 0
-	driver, err := ConnectSQLXDriver(context.Background(), conf, node.Info{})
+	driver, err := NewSQLXDriver(context.Background(), conf, node.Info{})
 	if err != nil {
 		return nil, err
 	}
@@ -39,7 +39,7 @@ func SetupSQLXDB() (sql.Database, error) {
 
 // SetupPGXDB is used to setup a pgx db for tests
 func SetupPGXDB(config Config) (sql.Database, error) {
-	driver, err := ConnectPGXDriver(context.Background(), config, node.Info{})
+	driver, err := NewPGXDriver(context.Background(), config, node.Info{})
 	if err != nil {
 		return nil, err
 	}

indexer/shared/schema/schema.go

@@ -16,7 +16,7 @@
 
 package schema
 
-var EthTables = []*Table{
+var Tables = []*Table{
 	&TableIPLDBlock,
 	&TableNodeInfo,
 	&TableHeader,
@@ -29,11 +29,6 @@ var EthTables = []*Table{
 	&TableWithdrawal,
 }
 
-var AllTables = append(
-	EthTables,
-	&TableWatchedAddresses,
-)
-
 var TableIPLDBlock = Table{
 	Name: `ipld.blocks`,
 	Columns: []Column{

indexer/test_helpers/test_helpers.go

@@ -19,14 +19,10 @@ package test_helpers
 import (
 	"bufio"
 	"context"
-	"fmt"
 	"os"
 	"testing"
 
 	"github.com/cerc-io/plugeth-statediff/indexer/database/sql"
-	"github.com/cerc-io/plugeth-statediff/indexer/database/sql/postgres"
-	"github.com/cerc-io/plugeth-statediff/indexer/shared/schema"
-	"github.com/jmoiron/sqlx"
 )
 
 // DedupFile removes duplicates from the given file
@@ -43,6 +39,9 @@ func DedupFile(filePath string) error {
 		s := sc.Text()
 		stmts[s] = struct{}{}
 	}
+	if err != nil {
+		return err
+	}
 
 	f.Close()
 
@@ -61,30 +60,31 @@ func DedupFile(filePath string) error {
 
 // TearDownDB is used to tear down the watcher dbs after tests
 func TearDownDB(t *testing.T, db sql.Database) {
-	err := ClearDB(db)
-	if err != nil {
-		t.Fatal(err)
-	}
-}
-
-func ClearSqlxDB(sqlxdb *sqlx.DB) error {
-	driver := postgres.NewSQLXDriver(context.Background(), sqlxdb)
-	db := postgres.NewPostgresDB(driver, false)
-	return ClearDB(db)
-}
-
-func ClearDB(db sql.Database) error {
 	ctx := context.Background()
 	tx, err := db.Begin(ctx)
 	if err != nil {
-		return err
+		t.Fatal(err)
 	}
 
-	for _, tbl := range schema.AllTables {
-		stm := fmt.Sprintf("TRUNCATE %s", tbl.Name)
+	statements := []string{
+		`TRUNCATE nodes`,
+		`TRUNCATE ipld.blocks`,
+		`TRUNCATE eth.header_cids`,
+		`TRUNCATE eth.uncle_cids`,
+		`TRUNCATE eth.transaction_cids`,
+		`TRUNCATE eth.receipt_cids`,
+		`TRUNCATE eth.state_cids`,
+		`TRUNCATE eth.storage_cids`,
+		`TRUNCATE eth.log_cids`,
+		`TRUNCATE eth.withdrawal_cids`,
+		`TRUNCATE eth_meta.watched_addresses`,
+	}
+	for _, stm := range statements {
 		if _, err = tx.Exec(ctx, stm); err != nil {
-			return fmt.Errorf("error executing `%s`: %w", stm, err)
+			t.Fatal(err)
 		}
 	}
-	return tx.Commit(ctx)
+	if err = tx.Commit(ctx); err != nil {
+		t.Fatal(err)
+	}
 }

test_helpers/db.go

@@ -0,0 +1,34 @@
+package test_helpers
+
+import (
+	"fmt"
+
+	"github.com/jmoiron/sqlx"
+)
+
+// ClearDB is used to empty the IPLD-ETH tables after tests
+func ClearDB(db *sqlx.DB) error {
+	tx, err := db.Beginx()
+	if err != nil {
+		return err
+	}
+	statements := []string{
+		`TRUNCATE nodes`,
+		`TRUNCATE ipld.blocks`,
+		`TRUNCATE eth.header_cids`,
+		`TRUNCATE eth.uncle_cids`,
+		`TRUNCATE eth.transaction_cids`,
+		`TRUNCATE eth.receipt_cids`,
+		`TRUNCATE eth.state_cids`,
+		`TRUNCATE eth.storage_cids`,
+		`TRUNCATE eth.log_cids`,
+		`TRUNCATE eth.withdrawal_cids`,
+		`TRUNCATE eth_meta.watched_addresses`,
+	}
+	for _, stm := range statements {
+		if _, err = tx.Exec(stm); err != nil {
+			return fmt.Errorf("error executing `%s`: %w", stm, err)
+		}
+	}
+	return tx.Commit()
+}