Bug fixes for hybrid model (#162)

* tweaks to logs and updating sockets

* fixing bugs found while testing hybrid model

* adding compat mode to tidy command

Makefile

@@ -81,4 +81,4 @@ stop-compose:
 	@docker-compose down
 
 vendor-deps:
-	@go mod tidy && go mod vendor
+	@go mod tidy -compat=1.17 && go mod vendor

cmd/internal/dependencies.go

@@ -51,11 +51,11 @@ func SetupRestDeps(cfg *config.Config, l log.Logger, db *sqlx.DB, c *client.Clie
 	sqlLiteStore := paydSQL.NewSQLiteStore(db)
 	proofSvc := service.NewProofsService(sqlLiteStore)
 
-	pcSvc := service.NewPeerChannelsSvc(sqlLiteStore, cfg.PeerChannels)
+	pcSvc := service.NewPeerChannelsSvc(sqlLiteStore, cfg.PeerChannels, &paydSQL.Transacter{})
 	pcNotifSvc := service.NewPeerChannelsNotifyService(cfg.PeerChannels, pcSvc)
 	pcNotifSvc.RegisterHandler(payd.PeerChannelHandlerTypeProof, proofSvc)
 
-	mapiStore := mapi.NewMapi(cfg.Mapi, mapiCli)
+	mapiStore := mapi.NewMapi(cfg.Mapi, mapiCli, l)
 	spvv, err := spv.NewPaymentVerifier(dataHttp.NewHeaderSVConnection(&http.Client{Timeout: time.Duration(cfg.HeadersClient.Timeout) * time.Second}, cfg.HeadersClient.Address))
 	if err != nil {
 		l.Fatal(err, "failed to create spv client")
@@ -138,10 +138,10 @@ func SetupSocketDeps(cfg *config.Config, l log.Logger, db *sqlx.DB, c *client.Cl
 	}
 	sqlLiteStore := paydSQL.NewSQLiteStore(db)
 	proofSvc := service.NewProofsService(sqlLiteStore)
-	pcSvc := service.NewPeerChannelsSvc(sqlLiteStore, cfg.PeerChannels)
+	pcSvc := service.NewPeerChannelsSvc(sqlLiteStore, cfg.PeerChannels, &paydSQL.Transacter{})
 	pcNotifSvc := service.NewPeerChannelsNotifyService(cfg.PeerChannels, pcSvc)
 	pcNotifSvc.RegisterHandler(payd.PeerChannelHandlerTypeProof, proofSvc)
-	mapiStore := mapi.NewMapi(cfg.Mapi, mapiCli)
+	mapiStore := mapi.NewMapi(cfg.Mapi, mapiCli, l)
 	spvv, err := spv.NewPaymentVerifier(dataHttp.NewHeaderSVConnection(&http.Client{Timeout: time.Duration(cfg.HeadersClient.Timeout) * time.Second}, cfg.HeadersClient.Address))
 	if err != nil {
 		l.Fatal(err, "failed to create spv client")

cmd/internal/internal.go

@@ -24,19 +24,24 @@ import (
 	"github.com/pkg/errors"
 	"github.com/spf13/viper"
 	echoSwagger "github.com/swaggo/echo-swagger"
+	"github.com/theflyingcodr/sockets"
 	"github.com/theflyingcodr/sockets/client"
 	smw "github.com/theflyingcodr/sockets/middleware"
 	"github.com/theflyingcodr/sockets/server"
 )
 
 // SetupEcho will set up and return an echo server.
-func SetupEcho(l log.Logger) *echo.Echo {
+func SetupEcho(cfg *config.Config, l log.Logger) *echo.Echo {
 	e := echo.New()
 	e.HideBanner = true
 
 	// Middleware
 	e.Use(middleware.Recover())
-	e.Use(middleware.Logger())
+	e.Use(middleware.LoggerWithConfig(middleware.LoggerConfig{
+		Skipper: func(c echo.Context) bool {
+			return cfg.Logging.Level != config.LogDebug
+		},
+	}))
 	e.Use(middleware.RequestID())
 	e.Use(middleware.CORSWithConfig(middleware.CORSConfig{
 		AllowOrigins: []string{"*"},
@@ -72,10 +77,14 @@ func SetupHTTPEndpoints(cfg config.Config, services *RestDeps, g *echo.Group) {
 
 // SetupSocketClient will setup handlers and socket server.
 func SetupSocketClient(cfg config.Config, deps *SocketDeps, c *client.Client) {
+	lcfg := smw.NewLoggerConfig()
+	lcfg.AddSkipper(func(msg *sockets.Message) bool {
+		return cfg.Logging.Level != config.LogDebug
+	})
 	c.WithMiddleware(smw.PanicHandler,
 		smw.Timeout(smw.NewTimeoutConfig()),
 		smw.Metrics(),
-		smw.Logger(smw.NewLoggerConfig()),
+		smw.Logger(lcfg),
 		socMiddleware.IgnoreMyMessages(cfg.Socket),
 		socMiddleware.WithAppIDPayD()).
 		WithErrorHandler(socMiddleware.ErrorHandler).

cmd/server/main.go

@@ -75,7 +75,7 @@ func main() {
 	// nolint:errcheck // dont care about error.
 	defer db.Close()
 
-	e := internal.SetupEcho(log)
+	e := internal.SetupEcho(cfg, log)
 
 	if cfg.Server.SwaggerEnabled {
 		internal.SetupSwagger(*cfg.Server, e)

config/defaults.go

@@ -27,7 +27,7 @@ func SetupDefaults() {
 
 	// db
 	viper.SetDefault(EnvDb, "sqlite")
-	viper.SetDefault(EnvDbDsn, "file:data/wallet.db?_foreign_keys=true&pooled=true")
+	viper.SetDefault(EnvDbDsn, "file:data/wallet.db?_foreign_keys=true&pooling=true")
 	viper.SetDefault(EnvDbSchema, "data/sqlite/migrations")
 	viper.SetDefault(EnvDbMigrate, true)
 

data/http/dpp.go

@@ -5,6 +5,7 @@ import (
 	"context"
 	"encoding/json"
 	"errors"
+	"fmt"
 	"net/http"
 
 	"github.com/libsv/go-dpp"
@@ -68,15 +69,17 @@ func (p *dppClient) PaymentSend(ctx context.Context, args payd.PayRequest, req d
 		_ = resp.Body.Close()
 	}()
 
-	if resp.StatusCode != http.StatusCreated && resp.StatusCode != http.StatusNoContent {
+	if resp.StatusCode != http.StatusCreated && resp.StatusCode != http.StatusNoContent && resp.StatusCode != http.StatusUnprocessableEntity {
 		return nil, p.handleErr(resp)
 	}
 
 	var ack dpp.PaymentACK
 	if err := json.NewDecoder(resp.Body).Decode(&ack); err != nil {
 		return nil, err
 	}
-
+	if ack.Error > 0 {
+		return nil, errs.NewErrUnprocessable(fmt.Sprintf("%d", ack.Error), ack.Memo)
+	}
 	return &ack, nil
 }
 
@@ -87,11 +90,9 @@ func (p *dppClient) handleErr(resp *http.Response) error {
 		Title   string `json:"title"`
 		Message string `json:"message"`
 	}{}
-
 	if err := json.NewDecoder(resp.Body).Decode(&errResp); err != nil {
 		return err
 	}
-
 	switch resp.StatusCode {
 	case http.StatusUnauthorized:
 		return errs.NewErrNotAuthenticated(errResp.Code, errResp.Message)

data/mapi/mapi.go

@@ -10,17 +10,19 @@ import (
 
 	"github.com/libsv/payd"
 	"github.com/libsv/payd/config"
+	"github.com/libsv/payd/log"
 )
 
 type minercraftMapi struct {
 	client *minercraft.Client
 	cfg    *config.MApi
 	fq     *bt.FeeQuote
+	l      log.Logger
 }
 
 // NewMapi will setup and return a new MAPI minercraftMapi data store.
-func NewMapi(cfg *config.MApi, client *minercraft.Client) *minercraftMapi {
-	return &minercraftMapi{client: client, cfg: cfg, fq: bt.NewFeeQuote()}
+func NewMapi(cfg *config.MApi, client *minercraft.Client, l log.Logger) *minercraftMapi {
+	return &minercraftMapi{client: client, cfg: cfg, fq: bt.NewFeeQuote(), l: l}
 }
 
 // Broadcast will submit a transaction to mapi for inclusion in a block.
@@ -52,6 +54,7 @@ func (m *minercraftMapi) Broadcast(ctx context.Context, args payd.BroadcastArgs,
 		// This should be fixed in MAPI not here, long term.
 		return nil
 	}
+	m.l.Debugf("failed to submit transaction with hex: %s", tx.String())
 	return errors.Errorf("failed to submit transaction %s", resp.Results.ResultDescription)
 }
 

data/sqlite/peerchannels.go

@@ -103,13 +103,20 @@ func (s *sqliteStore) PeerChannelsOpened(ctx context.Context, channelType payd.P
 }
 
 func (s *sqliteStore) PeerChannelCloseChannel(ctx context.Context, channelID string) error {
-	if _, err := s.db.NamedExecContext(ctx, sqlPeerChannelsCloseUpdate, struct {
+	tx, err := s.newTx(ctx)
+	if err != nil {
+		return errors.Wrapf(err, "failed to close peer channel sql")
+	}
+	defer func() {
+		_ = rollback(ctx, tx)
+	}()
+	if _, err := tx.NamedExecContext(ctx, sqlPeerChannelsCloseUpdate, struct {
 		ChannelID string `db:"channel_id"`
 	}{
 		ChannelID: channelID,
 	}); err != nil {
 		return errors.Wrapf(err, "failed to close channel %s", channelID)
 	}
 
-	return nil
+	return errors.Wrap(commit(ctx, tx), "failed to commit transaction for peerChannelClose")
 }

data/sqlite/sqlite.go

@@ -43,7 +43,6 @@ func commit(ctx context.Context, tx *sqlx.Tx) error {
 			return nil
 		}
 	}
-
 	return tx.Commit()
 }
 
@@ -119,6 +118,10 @@ type Tx struct {
 
 // WithTxContext will add a new empty transaction to the provided context.
 func WithTxContext(ctx context.Context) context.Context {
+	tx := TxFromContext(ctx)
+	if tx != nil {
+		return ctx
+	}
 	return context.WithValue(ctx, exec, &Tx{})
 }
 

data/sqlite/transaction.go

@@ -70,7 +70,7 @@ func (s *sqliteStore) TransactionCreate(ctx context.Context, req payd.Transactio
 	// insert tx and utxos
 	if err := handleNamedExec(tx, sqlTransactionCreate, req); err != nil {
 		var sqlErr sqlite3.Error
-		if ok := errors.As(err, sqlErr); ok {
+		if ok := errors.As(err, &sqlErr); ok {
 			if sqlErr.Code == sqlite3.ErrConstraint {
 				return lathos.NewErrDuplicate("D001", "transaction has already been stored")
 			}

data/sqlite/utxo.go

@@ -47,7 +47,9 @@ func (s *sqliteStore) UTXOReserve(ctx context.Context, req payd.UTXOReserve) ([]
 	if err != nil {
 		return nil, errors.Wrap(err, "error creation transaction to get utxos")
 	}
-
+	defer func() {
+		_ = rollback(ctx, tx)
+	}()
 	timestamp := time.Now().UTC()
 	var utxos []payd.UTXO
 	for total := uint64(0); total <= req.Satoshis; {
@@ -79,7 +81,9 @@ func (s *sqliteStore) UTXOUnreserve(ctx context.Context, req payd.UTXOUnreserve)
 	if err != nil {
 		return errors.Wrap(err, "error creation transaction to get utxos")
 	}
-
+	defer func() {
+		_ = rollback(ctx, tx)
+	}()
 	if _, err = tx.ExecContext(ctx, sqlUTXOUnreserve, time.Now().UTC(), req.ReservedFor); err != nil {
 		return errors.Wrap(err, "failed to unreserve utxos")
 	}
@@ -93,7 +97,9 @@ func (s *sqliteStore) UTXOSpend(ctx context.Context, req payd.UTXOSpend) error {
 	if err != nil {
 		return errors.Wrap(err, "failed to create transaction to spend utxos")
 	}
-
+	defer func() {
+		_ = rollback(ctx, tx)
+	}()
 	req.Timestamp = time.Now().UTC()
 	if err := handleNamedExec(tx, sqlUTXOSpend, req); err != nil {
 		return errors.Wrap(err, "failed to mark utxos as spent")

docker-compose.faucet.yml

@@ -8,7 +8,7 @@ services:
       LOG_LEVEL: "info"
       DPP_HOST: "wss://faucet.bitcoinsv.io/dpp/ws"
       MAPI_CALLBACK_HOST: "http://faucet.bitcoinsv.io/dpp"
-      PEERCHANNELS_HOST: "faucet.bitcoinsv.io/peerchannels"
+      PEERCHANNELS_HOST: 'faucet.bitcoinsv.io/peerchannels'
       WALLET_SPVREQUIRED: 'false'
     volumes:
       - ./run/regtest/payd:/paydb

docker-compose.yml

@@ -5,7 +5,7 @@ services:
     container_name: payd
     image: local.payd
     environment:
-      DB_DSN: "file:paydb/wallet.db?cache=shared&_foreign_keys=true"
+      DB_DSN: "file:paydb/wallet.db?_foreign_keys=true&pooling=true"
       DB_SCHEMA_PATH: "migrations"
       LOG_LEVEL: "info"
       DPP_HOST: "ws://dpp:8445/ws"
@@ -27,13 +27,13 @@ services:
     container_name: payd-merchant
     image: local.payd
     environment:
-      DB_DSN: "file:paydb/merchant-wallet.db?cache=shared&_foreign_keys=true"
+      DB_DSN: "file:paydb/merchant-wallet.db?_foreign_keys=true&pooling=true"
       DB_SCHEMA_PATH: "migrations"
       SERVER_HOST: payd-merchant:28443
       SERVER_PORT: :28443
       LOG_LEVEL: "info"
-      DPP_HOST: "ws://dpp-merchant:28445/ws"
-      MAPI_CALLBACK_HOST: "http://dpp-merchant:28445"
+      DPP_HOST: "ws://dpp:8445/ws"
+      MAPI_CALLBACK_HOST: "http://dpp:8445"
       PEERCHANNELS_HOST: "peerchannels:25009"
       WALLET_SPVREQUIRED: 'true'
     healthcheck:

go.mod

@@ -33,7 +33,7 @@ require (
 	github.com/gorilla/websocket v1.5.0
 	github.com/libsv/go-bc v0.1.10
 	github.com/rs/zerolog v1.27.0
-	github.com/theflyingcodr/sockets v0.0.11-beta.0.20220225103542-c6eecb16f586
+	github.com/theflyingcodr/sockets v0.0.12-beta
 )
 
 require (