Postgres database

.github/workflows/ci.yml

@@ -42,6 +42,9 @@ jobs:
       - name: Get dependencies
         run: go get -v -t -d ./...
 
+      - name: Setup Pg test db
+        run: make pgcreatetestdb
+
       - name: Run Tests
         run: |
           export TEST_ELEMENTS_ENDPOINT='http://admin1:123@localhost:18884'

Makefile

@@ -82,4 +82,85 @@ integrationtest:
 ## mock: generates mocks for unit tests
 mock:
 	@echo "Generating mocks for unit tests..."
-	@mockery --dir=internal/core/domain --name=SwapParser --structname=MockSwapParser --filename=swap.go --output=internal/core/domain/mocks
+	@mockery --dir=internal/core/domain --name=SwapParser --structname=MockSwapParser --filename=swap.go --output=internal/core/domain/mocks
+
+######## PG_DB ########
+## pg: starts postgres db inside docker container
+pg:
+	@echo "Starting postgres container..."
+	@docker run --name tdexd-pg -p 5432:5432 -e POSTGRES_USER=root -e POSTGRES_PASSWORD=secret -e POSTGRES_DB=tdexd -d postgres
+
+## droppg: stop and remove postgres container
+droppg:
+	@echo "Stopping and removing postgres container..."
+	@docker stop tdexd-pg
+	@docker rm tdexd-pg
+
+## createdb: create db inside docker container
+createdb:
+	@echo "Creating db..."
+	@docker exec tdexd-pg createdb --username=root --owner=root tdexd
+
+## createtestdb: create test db inside docker container
+createtestdb:
+	@echo "Creating test db..."
+	@docker exec tdexd-pg createdb --username=root --owner=root tdexd-test
+
+## recreatedb: drop and create main db
+recreatedb: dropdb createdb
+
+## recreatetestdb: drop and create main and test db
+recreatetestdb: droptestdb createtestdb
+
+## pgcreatetestdb: starts docker container and creates test db, used in CI
+pgcreatetestdb: pg sleep createtestdb
+	@echo "Starting postgres container with test db..."
+
+## dropdb: drops db inside docker container
+dropdb:
+	@echo "Dropping db..."
+	@docker exec tdexd-pg dropdb tdexd
+
+## droptestdb: drops test db inside docker container
+droptestdb:
+	@echo "Dropping test db..."
+	@docker exec tdexd-pg dropdb tdexd-test
+
+## mig_file: creates pg migration file(eg. make FILE=init mig_file)
+mig_file:
+	@echo "creating migration file..."
+	@migrate create -ext sql -dir ./internal/infrastructure/storage/db/pg/migration $(FILE)
+
+## mig_up_test: creates test db schema
+mig_up_test:
+	@echo "creating test db schema..."
+	@echo "creating db schema..."
+	@migrate -database "postgres://root:secret@localhost:5432/tdexd-test?sslmode=disable" -path ./internal/infrastructure/storage/db/pg/migration up
+
+## mig_up: creates db schema
+mig_up:
+	@echo "creating db schema..."
+	@migrate -database "postgres://root:secret@localhost:5432/tdexd?sslmode=disable" -path ./internal/infrastructure/storage/db/pg/migration up
+
+## mig_down_test: apply down migration on test db
+mig_down_test:
+	@echo "migration down on test db..."
+	@migrate -database "postgres://root:secret@localhost:5432/tdexd-test?sslmode=disable" -path ./internal/infrastructure/storage/db/pg/migration down
+
+## mig_down: apply down migration without prompt
+mig_down:
+	@echo "migration down..."
+	@"yes" | migrate -database "postgres://root:secret@localhost:5432/tdexd?sslmode=disable" -path ./internal/infrastructure/storage/db/pg/migration down
+
+## vet_db: check if mig_up and mig_down are ok
+vet_db: recreatedb mig_up mig_down
+	@echo "vet db migration scripts..."
+
+## sqlc: gen sql
+sqlc:
+	@echo "gen sql..."
+	@cd ./internal/infrastructure/storage/db/pg; sqlc generate
+
+sleep:
+	@echo "sleeping for 3 seconds..."
+	@sleep 3

go.mod

@@ -12,16 +12,18 @@ require (
 	github.com/btcsuite/btcd v0.23.2
 	github.com/btcsuite/btcd/btcec/v2 v2.2.0
 	github.com/btcsuite/btcd/btcutil v1.1.0
-	github.com/cpuguy83/go-md2man/v2 v2.0.0 // indirect
 	github.com/desertbit/timer v0.0.0-20180107155436-c41aec40b27f // indirect
 	github.com/dgraph-io/badger/v3 v3.2103.1
 	github.com/gogo/protobuf v1.3.2
 	github.com/golang-jwt/jwt v3.2.1+incompatible
+	github.com/golang-migrate/migrate/v4 v4.15.2
 	github.com/google/uuid v1.3.0
-	github.com/gorilla/websocket v1.5.0 // indirect
+	github.com/gorilla/websocket v1.5.0
 	github.com/grpc-ecosystem/go-grpc-middleware v1.3.0
 	github.com/grpc-ecosystem/grpc-gateway/v2 v2.15.2
 	github.com/improbable-eng/grpc-web v0.13.0
+	github.com/jackc/pgconn v1.14.0
+	github.com/jackc/pgx/v4 v4.18.1
 	github.com/prometheus/client_golang v1.11.1
 	github.com/rs/cors v1.7.0 // indirect
 	github.com/shopspring/decimal v1.2.0

internal/infrastructure/storage/db/pg/db_manager.go

@@ -0,0 +1,174 @@
+package postgresdb
+
+import (
+	"context"
+	"errors"
+	"fmt"
+
+	"github.com/jackc/pgx/v4"
+	log "github.com/sirupsen/logrus"
+	"github.com/tdex-network/tdex-daemon/internal/infrastructure/storage/db/pg/sqlc/queries"
+
+	"github.com/tdex-network/tdex-daemon/internal/core/domain"
+	"github.com/tdex-network/tdex-daemon/internal/core/ports"
+
+	"github.com/golang-migrate/migrate/v4"
+	"github.com/golang-migrate/migrate/v4/database/postgres"
+	_ "github.com/golang-migrate/migrate/v4/source/file"
+	"github.com/jackc/pgx/v4/pgxpool"
+)
+
+const (
+	postgresDriver             = "pgx"
+	insecureDataSourceTemplate = "postgresql://%s:%s@%s:%d/%s?sslmode=disable"
+
+	uniqueViolation = "23505"
+	pgxNoRows       = "no rows in result set"
+)
+
+type repoManager struct {
+	pgxPool *pgxpool.Pool
+	querier *queries.Queries
+
+	marketRepository     domain.MarketRepository
+	tradeRepository      domain.TradeRepository
+	depositRepository    domain.DepositRepository
+	withdrawalRepository domain.WithdrawalRepository
+}
+
+func NewService(dbConfig DbConfig) (ports.RepoManager, error) {
+	dataSource := insecureDataSourceStr(dbConfig)
+
+	pgxPool, err := connect(dataSource)
+	if err != nil {
+		return nil, err
+	}
+
+	if err = migrateDb(dataSource, dbConfig.MigrationSourceURL); err != nil {
+		return nil, err
+	}
+
+	rm := &repoManager{
+		pgxPool: pgxPool,
+		querier: queries.New(pgxPool),
+	}
+
+	marketRepository := NewMarketRepositoryImpl(rm.querier, rm.execTx)
+	tradeRepository := NewTradeRepositoryImpl(rm.querier, rm.execTx)
+	depositRepository := NewDepositRepositoryImpl(rm.querier, rm.execTx)
+	withdrawalRepository := NewWithdrawalRepositoryImpl(rm.querier, rm.execTx)
+
+	rm.marketRepository = marketRepository
+	rm.tradeRepository = tradeRepository
+	rm.depositRepository = depositRepository
+	rm.withdrawalRepository = withdrawalRepository
+
+	return rm, nil
+}
+
+func (r *repoManager) MarketRepository() domain.MarketRepository {
+	return r.marketRepository
+}
+
+func (r *repoManager) TradeRepository() domain.TradeRepository {
+	return r.tradeRepository
+}
+
+func (r *repoManager) DepositRepository() domain.DepositRepository {
+	return r.depositRepository
+}
+
+func (r *repoManager) WithdrawalRepository() domain.WithdrawalRepository {
+	return r.withdrawalRepository
+}
+
+func (r *repoManager) Close() {
+	r.pgxPool.Close()
+}
+
+func (r *repoManager) execTx(
+	ctx context.Context,
+	txBody func(*queries.Queries) error,
+) error {
+	conn, err := r.pgxPool.Acquire(ctx)
+	if err != nil {
+		return err
+	}
+	defer conn.Release()
+
+	tx, err := conn.Begin(ctx)
+	if err != nil {
+		return err
+	}
+
+	// Rollback is safe to call even if the tx is already closed, so if
+	// the tx commits successfully, this is a no-op.
+	defer func() {
+		err := tx.Rollback(ctx)
+		switch {
+		// If the tx was already closed (it was successfully executed)
+		// we do not need to log that error.
+		case errors.Is(err, pgx.ErrTxClosed):
+			return
+
+		// If this is an unexpected error, log it.
+		case err != nil:
+			log.Errorf("unable to rollback db tx: %v", err)
+		}
+	}()
+
+	if err := txBody(r.querier.WithTx(tx)); err != nil {
+		return err
+	}
+
+	// Commit transaction.
+	return tx.Commit(ctx)
+}
+
+type DbConfig struct {
+	DbUser             string
+	DbPassword         string
+	DbHost             string
+	DbPort             int
+	DbName             string
+	MigrationSourceURL string
+}
+
+func connect(dataSource string) (*pgxpool.Pool, error) {
+	return pgxpool.Connect(context.Background(), dataSource)
+}
+
+func migrateDb(dataSource, migrationSourceUrl string) error {
+	pg := postgres.Postgres{}
+
+	d, err := pg.Open(dataSource)
+	if err != nil {
+		return err
+	}
+
+	m, err := migrate.NewWithDatabaseInstance(
+		migrationSourceUrl,
+		postgresDriver,
+		d,
+	)
+	if err != nil {
+		return err
+	}
+
+	if err := m.Up(); err != nil && err != migrate.ErrNoChange {
+		return err
+	}
+
+	return nil
+}
+
+func insecureDataSourceStr(dbConfig DbConfig) string {
+	return fmt.Sprintf(
+		insecureDataSourceTemplate,
+		dbConfig.DbUser,
+		dbConfig.DbPassword,
+		dbConfig.DbHost,
+		dbConfig.DbPort,
+		dbConfig.DbName,
+	)
+}