Add DB Connection and Logging
* Utilize LogRus * Create a DB connection using PGX. * Create an internal boot package for starting the application.
This commit is contained in:
Abdul Rabbani
@@ -0,0 +1,22 @@
|
||||
package sql
|
||||
|
||||
import (
|
||||
"fmt"
|
||||
)
|
||||
|
||||
const (
|
||||
DbConnectionFailedMsg = "db connection failed"
|
||||
SettingNodeFailedMsg = "unable to set db node"
|
||||
)
|
||||
|
||||
func ErrDBConnectionFailed(connectErr error) error {
|
||||
return formatError(DbConnectionFailedMsg, connectErr.Error())
|
||||
}
|
||||
|
||||
func ErrUnableToSetNode(setErr error) error {
|
||||
return formatError(SettingNodeFailedMsg, setErr.Error())
|
||||
}
|
||||
|
||||
func formatError(msg, err string) error {
|
||||
return fmt.Errorf("%s: %s", msg, err)
|
||||
}
|
||||
@@ -0,0 +1,54 @@
|
||||
package sql
|
||||
|
||||
import (
|
||||
"context"
|
||||
"io"
|
||||
"time"
|
||||
)
|
||||
|
||||
// Database interfaces required by the sql indexer
|
||||
type Database interface {
|
||||
Driver
|
||||
}
|
||||
|
||||
// Driver interface has all the methods required by a driver implementation to support the sql indexer
|
||||
type Driver interface {
|
||||
QueryRow(ctx context.Context, sql string, args ...interface{}) ScannableRow
|
||||
Exec(ctx context.Context, sql string, args ...interface{}) (Result, error)
|
||||
Select(ctx context.Context, dest interface{}, query string, args ...interface{}) error
|
||||
Get(ctx context.Context, dest interface{}, query string, args ...interface{}) error
|
||||
Begin(ctx context.Context) (Tx, error)
|
||||
Stats() Stats
|
||||
Context() context.Context
|
||||
io.Closer
|
||||
}
|
||||
|
||||
// Tx interface to accommodate different concrete SQL transaction types
|
||||
type Tx interface {
|
||||
QueryRow(ctx context.Context, sql string, args ...interface{}) ScannableRow
|
||||
Exec(ctx context.Context, sql string, args ...interface{}) (Result, error)
|
||||
Commit(ctx context.Context) error
|
||||
Rollback(ctx context.Context) error
|
||||
}
|
||||
|
||||
// ScannableRow interface to accommodate different concrete row types
|
||||
type ScannableRow interface {
|
||||
Scan(dest ...interface{}) error
|
||||
}
|
||||
|
||||
// Result interface to accommodate different concrete result types
|
||||
type Result interface {
|
||||
RowsAffected() (int64, error)
|
||||
}
|
||||
|
||||
// Stats interface to accommodate different concrete sql stats types
|
||||
type Stats interface {
|
||||
MaxOpen() int64
|
||||
Open() int64
|
||||
InUse() int64
|
||||
Idle() int64
|
||||
WaitCount() int64
|
||||
WaitDuration() time.Duration
|
||||
MaxIdleClosed() int64
|
||||
MaxLifetimeClosed() int64
|
||||
}
|
||||
@@ -0,0 +1,71 @@
|
||||
package postgres
|
||||
|
||||
import (
|
||||
"fmt"
|
||||
"strings"
|
||||
"time"
|
||||
)
|
||||
|
||||
// DriverType to explicitly type the kind of sql driver we are using
|
||||
type DriverType string
|
||||
|
||||
const (
|
||||
PGX DriverType = "PGX"
|
||||
SQLX DriverType = "SQLX"
|
||||
Unknown DriverType = "Unknown"
|
||||
)
|
||||
|
||||
// DefaultConfig are default parameters for connecting to a Postgres sql
|
||||
var DefaultConfig = Config{
|
||||
Hostname: "localhost",
|
||||
Port: 8077,
|
||||
DatabaseName: "vulcanize_testing",
|
||||
Username: "vdbm",
|
||||
Password: "password",
|
||||
}
|
||||
|
||||
// ResolveDriverType resolves a DriverType from a provided string
|
||||
func ResolveDriverType(str string) (DriverType, error) {
|
||||
switch strings.ToLower(str) {
|
||||
case "pgx", "pgxpool":
|
||||
return PGX, nil
|
||||
case "sqlx":
|
||||
return SQLX, nil
|
||||
default:
|
||||
return Unknown, fmt.Errorf("unrecognized driver type string: %s", str)
|
||||
}
|
||||
}
|
||||
|
||||
// Config holds params for a Postgres db
|
||||
type Config struct {
|
||||
// conn string params
|
||||
Hostname string
|
||||
Port int
|
||||
DatabaseName string
|
||||
Username string
|
||||
Password string
|
||||
|
||||
// conn settings
|
||||
MaxConns int
|
||||
MaxIdle int
|
||||
MinConns int
|
||||
MaxConnIdleTime time.Duration
|
||||
MaxConnLifetime time.Duration
|
||||
ConnTimeout time.Duration
|
||||
|
||||
// driver type
|
||||
Driver DriverType
|
||||
}
|
||||
|
||||
// DbConnectionString constructs and returns the connection string from the config
|
||||
func (c Config) DbConnectionString() string {
|
||||
if len(c.Username) > 0 && len(c.Password) > 0 {
|
||||
return fmt.Sprintf("postgresql://%s:%s@%s:%d/%s?sslmode=disable",
|
||||
c.Username, c.Password, c.Hostname, c.Port, c.DatabaseName)
|
||||
}
|
||||
if len(c.Username) > 0 && len(c.Password) == 0 {
|
||||
return fmt.Sprintf("postgresql://%s@%s:%d/%s?sslmode=disable",
|
||||
c.Username, c.Hostname, c.Port, c.DatabaseName)
|
||||
}
|
||||
return fmt.Sprintf("postgresql://%s:%d/%s?sslmode=disable", c.Hostname, c.Port, c.DatabaseName)
|
||||
}
|
||||
@@ -0,0 +1,50 @@
|
||||
package postgres
|
||||
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
|
||||
log "github.com/sirupsen/logrus"
|
||||
"github.com/vulcanize/ipld-ethcl-indexer/pkg/database/sql"
|
||||
)
|
||||
|
||||
var _ sql.Database = &DB{}
|
||||
|
||||
// TODO: Make NewPostgresDB accept a string and Config. IT should
|
||||
// Create a driver of its own.
|
||||
// This will make sure that if you want a driver, it conforms to the interface.
|
||||
|
||||
// NewPostgresDB returns a postgres.DB using the provided Config and driver type.
|
||||
func NewPostgresDB(c Config) (*DB, error) {
|
||||
var driver *pgxDriver
|
||||
|
||||
driver, err := createDriver(c)
|
||||
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
return &DB{driver}, nil
|
||||
}
|
||||
|
||||
func createDriver(c Config) (*pgxDriver, error) {
|
||||
switch c.Driver {
|
||||
case PGX:
|
||||
log.Debug("Creating New Driver")
|
||||
driver, err := newPGXDriver(context.Background(), c)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("Error Creating Driver, err: %e", err)
|
||||
}
|
||||
log.Info("Successfully created a driver for PGX")
|
||||
return driver, nil
|
||||
default:
|
||||
log.Fatal("Couldnt find a driver to create for: ", c.Driver)
|
||||
return nil, fmt.Errorf("Can't find a driver to create")
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
// DB implements sql.Database using a configured driver and Postgres statement syntax
|
||||
type DB struct {
|
||||
sql.Driver
|
||||
}
|
||||
@@ -0,0 +1,190 @@
|
||||
package postgres
|
||||
|
||||
import (
|
||||
"context"
|
||||
"time"
|
||||
|
||||
"github.com/georgysavva/scany/pgxscan"
|
||||
"github.com/jackc/pgconn"
|
||||
"github.com/jackc/pgx/v4"
|
||||
"github.com/jackc/pgx/v4/pgxpool"
|
||||
"github.com/vulcanize/ipld-ethcl-indexer/pkg/database/sql"
|
||||
)
|
||||
|
||||
// pgxDriver driver, implements sql.Driver
|
||||
type pgxDriver struct {
|
||||
ctx context.Context
|
||||
pool *pgxpool.Pool
|
||||
}
|
||||
|
||||
// newPGXDriver returns a new pgx driver.
|
||||
// It initializes the connection pool.
|
||||
func newPGXDriver(ctx context.Context, config Config) (*pgxDriver, error) {
|
||||
pgConf, err := makeConfig(config)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
dbPool, err := pgxpool.ConnectConfig(ctx, pgConf)
|
||||
if err != nil {
|
||||
return nil, sql.ErrDBConnectionFailed(err)
|
||||
}
|
||||
pg := &pgxDriver{ctx: ctx, pool: dbPool}
|
||||
return pg, nil
|
||||
}
|
||||
|
||||
// makeConfig creates a pgxpool.Config from the provided Config
|
||||
func makeConfig(config Config) (*pgxpool.Config, error) {
|
||||
conf, err := pgxpool.ParseConfig("")
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
//conf.ConnConfig.BuildStatementCache = nil
|
||||
conf.ConnConfig.Config.Host = config.Hostname
|
||||
conf.ConnConfig.Config.Port = uint16(config.Port)
|
||||
conf.ConnConfig.Config.Database = config.DatabaseName
|
||||
conf.ConnConfig.Config.User = config.Username
|
||||
conf.ConnConfig.Config.Password = config.Password
|
||||
|
||||
if config.ConnTimeout != 0 {
|
||||
conf.ConnConfig.Config.ConnectTimeout = config.ConnTimeout
|
||||
}
|
||||
if config.MaxConns != 0 {
|
||||
conf.MaxConns = int32(config.MaxConns)
|
||||
}
|
||||
if config.MinConns != 0 {
|
||||
conf.MinConns = int32(config.MinConns)
|
||||
}
|
||||
if config.MaxConnLifetime != 0 {
|
||||
conf.MaxConnLifetime = config.MaxConnLifetime
|
||||
}
|
||||
if config.MaxConnIdleTime != 0 {
|
||||
conf.MaxConnIdleTime = config.MaxConnIdleTime
|
||||
}
|
||||
return conf, nil
|
||||
}
|
||||
|
||||
// QueryRow satisfies sql.Database
|
||||
func (pgx *pgxDriver) QueryRow(ctx context.Context, sql string, args ...interface{}) sql.ScannableRow {
|
||||
return pgx.pool.QueryRow(ctx, sql, args...)
|
||||
}
|
||||
|
||||
// Exec satisfies sql.Database
|
||||
func (pgx *pgxDriver) Exec(ctx context.Context, sql string, args ...interface{}) (sql.Result, error) {
|
||||
res, err := pgx.pool.Exec(ctx, sql, args...)
|
||||
return resultWrapper{ct: res}, err
|
||||
}
|
||||
|
||||
// Select satisfies sql.Database
|
||||
func (pgx *pgxDriver) Select(ctx context.Context, dest interface{}, query string, args ...interface{}) error {
|
||||
return pgxscan.Select(ctx, pgx.pool, dest, query, args...)
|
||||
}
|
||||
|
||||
// Get satisfies sql.Database
|
||||
func (pgx *pgxDriver) Get(ctx context.Context, dest interface{}, query string, args ...interface{}) error {
|
||||
return pgxscan.Get(ctx, pgx.pool, dest, query, args...)
|
||||
}
|
||||
|
||||
// Begin satisfies sql.Database
|
||||
func (pgx *pgxDriver) Begin(ctx context.Context) (sql.Tx, error) {
|
||||
tx, err := pgx.pool.Begin(ctx)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return pgxTxWrapper{tx: tx}, nil
|
||||
}
|
||||
|
||||
func (pgx *pgxDriver) Stats() sql.Stats {
|
||||
stats := pgx.pool.Stat()
|
||||
return pgxStatsWrapper{stats: stats}
|
||||
}
|
||||
|
||||
// Close satisfies sql.Database/io.Closer
|
||||
func (pgx *pgxDriver) Close() error {
|
||||
pgx.pool.Close()
|
||||
return nil
|
||||
}
|
||||
|
||||
// Context satisfies sql.Database
|
||||
func (pgx *pgxDriver) Context() context.Context {
|
||||
return pgx.ctx
|
||||
}
|
||||
|
||||
type resultWrapper struct {
|
||||
ct pgconn.CommandTag
|
||||
}
|
||||
|
||||
// RowsAffected satisfies sql.Result
|
||||
func (r resultWrapper) RowsAffected() (int64, error) {
|
||||
return r.ct.RowsAffected(), nil
|
||||
}
|
||||
|
||||
type pgxStatsWrapper struct {
|
||||
stats *pgxpool.Stat
|
||||
}
|
||||
|
||||
// MaxOpen satisfies sql.Stats
|
||||
func (s pgxStatsWrapper) MaxOpen() int64 {
|
||||
return int64(s.stats.MaxConns())
|
||||
}
|
||||
|
||||
// Open satisfies sql.Stats
|
||||
func (s pgxStatsWrapper) Open() int64 {
|
||||
return int64(s.stats.TotalConns())
|
||||
}
|
||||
|
||||
// InUse satisfies sql.Stats
|
||||
func (s pgxStatsWrapper) InUse() int64 {
|
||||
return int64(s.stats.AcquiredConns())
|
||||
}
|
||||
|
||||
// Idle satisfies sql.Stats
|
||||
func (s pgxStatsWrapper) Idle() int64 {
|
||||
return int64(s.stats.IdleConns())
|
||||
}
|
||||
|
||||
// WaitCount satisfies sql.Stats
|
||||
func (s pgxStatsWrapper) WaitCount() int64 {
|
||||
return s.stats.EmptyAcquireCount()
|
||||
}
|
||||
|
||||
// WaitDuration satisfies sql.Stats
|
||||
func (s pgxStatsWrapper) WaitDuration() time.Duration {
|
||||
return s.stats.AcquireDuration()
|
||||
}
|
||||
|
||||
// MaxIdleClosed satisfies sql.Stats
|
||||
func (s pgxStatsWrapper) MaxIdleClosed() int64 {
|
||||
// this stat isn't supported by pgxpool, but we don't want to panic
|
||||
return 0
|
||||
}
|
||||
|
||||
// MaxLifetimeClosed satisfies sql.Stats
|
||||
func (s pgxStatsWrapper) MaxLifetimeClosed() int64 {
|
||||
return s.stats.CanceledAcquireCount()
|
||||
}
|
||||
|
||||
type pgxTxWrapper struct {
|
||||
tx pgx.Tx
|
||||
}
|
||||
|
||||
// QueryRow satisfies sql.Tx
|
||||
func (t pgxTxWrapper) QueryRow(ctx context.Context, sql string, args ...interface{}) sql.ScannableRow {
|
||||
return t.tx.QueryRow(ctx, sql, args...)
|
||||
}
|
||||
|
||||
// Exec satisfies sql.Tx
|
||||
func (t pgxTxWrapper) Exec(ctx context.Context, sql string, args ...interface{}) (sql.Result, error) {
|
||||
res, err := t.tx.Exec(ctx, sql, args...)
|
||||
return resultWrapper{ct: res}, err
|
||||
}
|
||||
|
||||
// Commit satisfies sql.Tx
|
||||
func (t pgxTxWrapper) Commit(ctx context.Context) error {
|
||||
return t.tx.Commit(ctx)
|
||||
}
|
||||
|
||||
// Rollback satisfies sql.Tx
|
||||
func (t pgxTxWrapper) Rollback(ctx context.Context) error {
|
||||
return t.tx.Rollback(ctx)
|
||||
}
|
||||
@@ -0,0 +1,98 @@
|
||||
package postgres
|
||||
|
||||
import (
|
||||
"context"
|
||||
"math/big"
|
||||
"strings"
|
||||
"testing"
|
||||
|
||||
"github.com/jackc/pgx/v4/pgxpool"
|
||||
"github.com/vulcanize/ipld-ethcl-indexer/pkg/database/sql"
|
||||
"github.com/vulcanize/ipld-ethcl-indexer/pkg/testhelpers"
|
||||
)
|
||||
|
||||
var (
|
||||
pgConfig, _ = makeConfig(DefaultConfig)
|
||||
ctx = context.Background()
|
||||
)
|
||||
|
||||
func expectContainsSubstring(t *testing.T, full string, sub string) {
|
||||
if !strings.Contains(full, sub) {
|
||||
t.Fatalf("Expected \"%v\" to contain substring \"%v\"\n", full, sub)
|
||||
}
|
||||
}
|
||||
|
||||
func TestPostgresPGX(t *testing.T) {
|
||||
t.Run("connects to the sql", func(t *testing.T) {
|
||||
dbPool, err := pgxpool.ConnectConfig(context.Background(), pgConfig)
|
||||
if err != nil {
|
||||
t.Fatalf("failed to connect to db with connection string: %s err: %v", pgConfig.ConnString(), err)
|
||||
}
|
||||
if dbPool == nil {
|
||||
t.Fatal("DB pool is nil")
|
||||
}
|
||||
dbPool.Close()
|
||||
})
|
||||
|
||||
t.Run("serializes big.Int to db", func(t *testing.T) {
|
||||
// postgres driver doesn't support go big.Int type
|
||||
// various casts in golang uint64, int64, overflow for
|
||||
// transaction value (in wei) even though
|
||||
// postgres numeric can handle an arbitrary
|
||||
// sized int, so use string representation of big.Int
|
||||
// and cast on insert
|
||||
|
||||
dbPool, err := pgxpool.ConnectConfig(context.Background(), pgConfig)
|
||||
if err != nil {
|
||||
t.Fatalf("failed to connect to db with connection string: %s err: %v", pgConfig.ConnString(), err)
|
||||
}
|
||||
defer dbPool.Close()
|
||||
|
||||
bi := new(big.Int)
|
||||
bi.SetString("34940183920000000000", 10)
|
||||
testhelpers.ExpectEqual(t, bi.String(), "34940183920000000000")
|
||||
|
||||
defer dbPool.Exec(ctx, `DROP TABLE IF EXISTS example`)
|
||||
_, err = dbPool.Exec(ctx, "CREATE TABLE example ( id INTEGER, data NUMERIC )")
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
||||
sqlStatement := `
|
||||
INSERT INTO example (id, data)
|
||||
VALUES (1, cast($1 AS NUMERIC))`
|
||||
_, err = dbPool.Exec(ctx, sqlStatement, bi.String())
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
||||
var data string
|
||||
err = dbPool.QueryRow(ctx, `SELECT cast(data AS TEXT) FROM example WHERE id = 1`).Scan(&data)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
||||
testhelpers.ExpectEqual(t, data, bi.String())
|
||||
actual := new(big.Int)
|
||||
actual.SetString(data, 10)
|
||||
testhelpers.ExpectEqual(t, actual, bi)
|
||||
})
|
||||
|
||||
t.Run("throws error when can't connect to the database", func(t *testing.T) {
|
||||
_, err := NewPostgresDB(Config{}, "PGX")
|
||||
if err == nil {
|
||||
t.Fatal("Expected an error")
|
||||
}
|
||||
|
||||
expectContainsSubstring(t, err.Error(), sql.DbConnectionFailedMsg)
|
||||
})
|
||||
t.Run("Connect to the database", func(t *testing.T) {
|
||||
driver, err := NewPostgresDB(DefaultConfig, "pgx")
|
||||
defer driver.Close()
|
||||
|
||||
if err != nil {
|
||||
t.Fatal("Error creating the postgres driver")
|
||||
}
|
||||
|
||||
})
|
||||
}
|
||||
Reference in New Issue
Block a user