Merge pull request #563 from BuxOrg/feat/368-callbacks-v2

feat(BUX-368): Arc callbacks

.gitignore

@@ -21,6 +21,9 @@ go.list
 # Jetbrains
 .idea/
 
+#VSCode
+.vscode/
+
 # Eclipse
 .project
 

action_destination.go

@@ -2,15 +2,16 @@ package bux
 
 import (
 	"context"
+
 	"github.com/mrz1836/go-datastore"
 )
 
 // NewDestination will get a new destination for an existing xPub
 //
 // xPubKey is the raw public xPub
 func (c *Client) NewDestination(ctx context.Context, xPubKey string, chain uint32,
-	destinationType string, opts ...ModelOps) (*Destination, error) {
-
+	destinationType string, opts ...ModelOps,
+) (*Destination, error) {
 	// Check for existing NewRelic transaction
 	ctx = c.GetOrStartTxn(ctx, "new_destination")
 
@@ -46,8 +47,8 @@ func (c *Client) NewDestination(ctx context.Context, xPubKey string, chain uint3
 
 // NewDestinationForLockingScript will create a new destination based on a locking script
 func (c *Client) NewDestinationForLockingScript(ctx context.Context, xPubID, lockingScript string,
-	opts ...ModelOps) (*Destination, error) {
-
+	opts ...ModelOps,
+) (*Destination, error) {
 	// Check for existing NewRelic transaction
 	ctx = c.GetOrStartTxn(ctx, "new_destination_for_locking_script")
 
@@ -66,18 +67,6 @@ func (c *Client) NewDestinationForLockingScript(ctx context.Context, xPubID, loc
 		return nil, ErrUnknownLockingScript
 	}
 
-<<<<<<< HEAD
-	// set the monitoring, passed down from the initiating function
-	// this will be set when calling NewDestination from http, but not for instance paymail
-	if monitor {
-		destination.Monitor = customTypes.NullTime{NullTime: sql.NullTime{
-			Valid: true,
-			Time:  time.Now(),
-		}}
-	}
-
-=======
->>>>>>> 06feaba (feat(BUX-417): remove monitor, ITC flag and IncomingTransaction (#532))
 	// Save the destination
 	if err := destination.Save(ctx); err != nil {
 		return nil, err
@@ -89,8 +78,8 @@ func (c *Client) NewDestinationForLockingScript(ctx context.Context, xPubID, loc
 
 // GetDestinations will get all the destinations from the Datastore
 func (c *Client) GetDestinations(ctx context.Context, metadataConditions *Metadata,
-	conditions *map[string]interface{}, queryParams *datastore.QueryParams, opts ...ModelOps) ([]*Destination, error) {
-
+	conditions *map[string]interface{}, queryParams *datastore.QueryParams, opts ...ModelOps,
+) ([]*Destination, error) {
 	// Check for existing NewRelic transaction
 	ctx = c.GetOrStartTxn(ctx, "get_destinations")
 
@@ -108,8 +97,8 @@ func (c *Client) GetDestinations(ctx context.Context, metadataConditions *Metada
 
 // GetDestinationsCount will get a count of all the destinations from the Datastore
 func (c *Client) GetDestinationsCount(ctx context.Context, metadataConditions *Metadata,
-	conditions *map[string]interface{}, opts ...ModelOps) (int64, error) {
-
+	conditions *map[string]interface{}, opts ...ModelOps,
+) (int64, error) {
 	// Check for existing NewRelic transaction
 	ctx = c.GetOrStartTxn(ctx, "get_destinations_count")
 
@@ -129,8 +118,8 @@ func (c *Client) GetDestinationsCount(ctx context.Context, metadataConditions *M
 //
 // metadataConditions are the search criteria used to find destinations
 func (c *Client) GetDestinationsByXpubID(ctx context.Context, xPubID string, metadataConditions *Metadata,
-	conditions *map[string]interface{}, queryParams *datastore.QueryParams) ([]*Destination, error) {
-
+	conditions *map[string]interface{}, queryParams *datastore.QueryParams,
+) ([]*Destination, error) {
 	// Check for existing NewRelic transaction
 	ctx = c.GetOrStartTxn(ctx, "get_destinations")
 
@@ -147,8 +136,8 @@ func (c *Client) GetDestinationsByXpubID(ctx context.Context, xPubID string, met
 
 // GetDestinationsByXpubIDCount will get a count of all destinations based on an xPub
 func (c *Client) GetDestinationsByXpubIDCount(ctx context.Context, xPubID string, metadataConditions *Metadata,
-	conditions *map[string]interface{}) (int64, error) {
-
+	conditions *map[string]interface{},
+) (int64, error) {
 	// Check for existing NewRelic transaction
 	ctx = c.GetOrStartTxn(ctx, "get_destinations")
 
@@ -165,7 +154,6 @@ func (c *Client) GetDestinationsByXpubIDCount(ctx context.Context, xPubID string
 
 // GetDestinationByID will get a destination by id
 func (c *Client) GetDestinationByID(ctx context.Context, xPubID, id string) (*Destination, error) {
-
 	// Check for existing NewRelic transaction
 	ctx = c.GetOrStartTxn(ctx, "get_destination_by_id")
 
@@ -187,7 +175,6 @@ func (c *Client) GetDestinationByID(ctx context.Context, xPubID, id string) (*De
 
 // GetDestinationByLockingScript will get a destination for a locking script
 func (c *Client) GetDestinationByLockingScript(ctx context.Context, xPubID, lockingScript string) (*Destination, error) {
-
 	// Check for existing NewRelic transaction
 	ctx = c.GetOrStartTxn(ctx, "get_destination_by_locking_script")
 
@@ -209,7 +196,6 @@ func (c *Client) GetDestinationByLockingScript(ctx context.Context, xPubID, lock
 
 // GetDestinationByAddress will get a destination for an address
 func (c *Client) GetDestinationByAddress(ctx context.Context, xPubID, address string) (*Destination, error) {
-
 	// Check for existing NewRelic transaction
 	ctx = c.GetOrStartTxn(ctx, "get_destination_by_address")
 
@@ -231,8 +217,8 @@ func (c *Client) GetDestinationByAddress(ctx context.Context, xPubID, address st
 
 // UpdateDestinationMetadataByID will update the metadata in an existing destination by id
 func (c *Client) UpdateDestinationMetadataByID(ctx context.Context, xPubID, id string,
-	metadata Metadata) (*Destination, error) {
-
+	metadata Metadata,
+) (*Destination, error) {
 	// Check for existing NewRelic transaction
 	ctx = c.GetOrStartTxn(ctx, "update_destination_by_id")
 
@@ -253,8 +239,8 @@ func (c *Client) UpdateDestinationMetadataByID(ctx context.Context, xPubID, id s
 
 // UpdateDestinationMetadataByLockingScript will update the metadata in an existing destination by locking script
 func (c *Client) UpdateDestinationMetadataByLockingScript(ctx context.Context, xPubID,
-	lockingScript string, metadata Metadata) (*Destination, error) {
-
+	lockingScript string, metadata Metadata,
+) (*Destination, error) {
 	// Check for existing NewRelic transaction
 	ctx = c.GetOrStartTxn(ctx, "update_destination_by_locking_script")
 
@@ -275,8 +261,8 @@ func (c *Client) UpdateDestinationMetadataByLockingScript(ctx context.Context, x
 
 // UpdateDestinationMetadataByAddress will update the metadata in an existing destination by address
 func (c *Client) UpdateDestinationMetadataByAddress(ctx context.Context, xPubID, address string,
-	metadata Metadata) (*Destination, error) {
-
+	metadata Metadata,
+) (*Destination, error) {
 	// Check for existing NewRelic transaction
 	ctx = c.GetOrStartTxn(ctx, "update_destination_by_address")
 

action_transaction.go

@@ -9,6 +9,8 @@ import (
 
 	"github.com/BuxOrg/bux/chainstate"
 	"github.com/BuxOrg/bux/utils"
+	"github.com/bitcoin-sv/go-broadcast-client/broadcast"
+	"github.com/libsv/go-bc"
 	"github.com/libsv/go-bt"
 	"github.com/mrz1836/go-datastore"
 )
@@ -381,6 +383,39 @@ func (c *Client) RevertTransaction(ctx context.Context, id string) error {
 	return err
 }
 
+// UpdateTransaction will update the broadcast callback transaction info, like: block height, block hash, status, bump.
+func (c *Client) UpdateTransaction(ctx context.Context, callbackResp *broadcast.SubmittedTx) error {
+	bump, err := bc.NewBUMPFromStr(callbackResp.MerklePath)
+	if err != nil {
+		c.options.logger.Err(err).Msgf("failed to parse merkle path from broadcast callback - tx: %v", callbackResp)
+		return err
+	}
+
+	txInfo := &chainstate.TransactionInfo{
+		BlockHash:   callbackResp.BlockHash,
+		BlockHeight: callbackResp.BlockHeight,
+		ID:          callbackResp.TxID,
+		TxStatus:    callbackResp.TxStatus,
+		BUMP:        bump,
+		// it's not possible to get confirmations from broadcast client; zero would be treated as "not confirmed" that's why -1
+		Confirmations: -1,
+	}
+
+	tx, err := c.GetTransaction(ctx, "", txInfo.ID)
+	if err != nil {
+		c.options.logger.Err(err).Msgf("failed to get transaction by id: %v", txInfo.ID)
+		return err
+	}
+
+	syncTx, err := GetSyncTransactionByTxID(ctx, txInfo.ID, c.DefaultModelOptions()...)
+	if err != nil {
+		c.options.logger.Err(err).Msgf("failed to get sync transaction by tx id: %v", txInfo.ID)
+		return err
+	}
+
+	return processSyncTxSave(ctx, txInfo, syncTx, tx)
+}
+
 func generateTxIDFilterConditions(txIDs []string) *map[string]interface{} {
 	orConditions := make([]map[string]interface{}, len(txIDs))
 

chainstate/broadcast_providers.go

@@ -94,14 +94,19 @@ func (provider broadcastClientProvider) broadcast(ctx context.Context, c *Client
 	return broadcastWithBroadcastClient(ctx, c, provider.txID, provider.txHex)
 }
 
-func broadcastWithBroadcastClient(ctx context.Context, client ClientInterface, txID, hex string) error {
+func broadcastWithBroadcastClient(ctx context.Context, client *Client, txID, hex string) error {
 	debugLog(client, txID, "executing broadcast request for "+ProviderBroadcastClient)
 
 	tx := broadcast.Transaction{
 		Hex: hex,
 	}
 
-	result, err := client.BroadcastClient().SubmitTransaction(ctx, &tx, broadcast.WithRawFormat())
+	result, err := client.BroadcastClient().SubmitTransaction(
+		ctx,
+		&tx,
+		broadcast.WithRawFormat(),
+		broadcast.WithCallback(client.options.config.callbackURL, client.options.config.callbackToken),
+	)
 	if err != nil {
 		debugLog(client, txID, "error broadcast request for "+ProviderBroadcastClient+" failed: "+err.Error())
 		return err

chainstate/client.go

@@ -32,6 +32,8 @@ type (
 
 	// syncConfig holds all the configuration about the different sync processes
 	syncConfig struct {
+		callbackURL       string                     // Broadcast callback URL
+		callbackToken     string                     // Broadcast callback access token
 		excludedProviders []string                   // List of provider names
 		httpClient        HTTPInterface              // Custom HTTP client (Minercraft, WOC)
 		minercraftConfig  *minercraftConfig          // minercraftConfig configuration

chainstate/client_options.go

@@ -165,3 +165,11 @@ func WithConnectionToPulse(url, authToken string) ClientOps {
 		c.config.pulseClient = newPulseClientProvider(url, authToken)
 	}
 }
+
+// WithCallback will set broadcast callback settings
+func WithCallback(callbackURL, callbackAuthToken string) ClientOps {
+	return func(c *clientOptions) {
+		c.config.callbackURL = callbackURL
+		c.config.callbackToken = callbackAuthToken
+	}
+}

client_options.go

@@ -662,3 +662,10 @@ func WithBroadcastClient(broadcastClient broadcast.Client) ClientOps {
 		c.chainstate.options = append(c.chainstate.options, chainstate.WithBroadcastClient(broadcastClient))
 	}
 }
+
+// WithCallback set callback settings
+func WithCallback(callbackURL string, callbackToken string) ClientOps {
+	return func(c *clientOptions) {
+		c.chainstate.options = append(c.chainstate.options, chainstate.WithCallback(callbackURL, callbackToken))
+	}
+}

cron_job_declarations.go

@@ -35,7 +35,7 @@ func (c *Client) cronJobs() taskmanager.CronJobs {
 			Handler: handler(taskBroadcastTransactions),
 		},
 		CronJobNameSyncTransactionSync: {
-			Period:  120 * time.Second,
+			Period:  600 * time.Second,
 			Handler: handler(taskSyncTransactions),
 		},
 	}

definitions.go

@@ -20,7 +20,7 @@ const (
 	dustLimit                  = uint64(1)         // Dust limit
 	mongoTestVersion           = "6.0.4"           // Mongo Testing Version
 	sqliteTestVersion          = "3.37.0"          // SQLite Testing Version (dummy version for now)
-	version                    = "v0.13.0"         // bux version
+	version                    = "v0.14.2"         // bux version
 )
 
 // All the base models

errors.go

@@ -104,7 +104,7 @@ var ErrUtxoNotReserved = errors.New("transaction utxo has not been reserved for
 var ErrDraftIDMismatch = errors.New("transaction draft id does not match utxo draft reservation id")
 
 // ErrMissingTxHex is when the hex is missing or invalid and creates an empty id
-var ErrMissingTxHex = errors.New("transaction hex is invalid or id is missing")
+var ErrMissingTxHex = errors.New("transaction hex is empty or id is missing")
 
 // ErrUtxoAlreadySpent is when the utxo is already spent, but is trying to be used
 var ErrUtxoAlreadySpent = errors.New("utxo has already been spent")

interface.go

@@ -8,6 +8,7 @@ import (
 	"github.com/BuxOrg/bux/cluster"
 	"github.com/BuxOrg/bux/notifications"
 	"github.com/BuxOrg/bux/taskmanager"
+	"github.com/bitcoin-sv/go-broadcast-client/broadcast"
 	"github.com/bitcoin-sv/go-paymail"
 	"github.com/mrz1836/go-cachestore"
 	"github.com/mrz1836/go-datastore"
@@ -132,6 +133,7 @@ type TransactionService interface {
 	RecordTransaction(ctx context.Context, xPubKey, txHex, draftID string,
 		opts ...ModelOps) (*Transaction, error)
 	RecordRawTransaction(ctx context.Context, txHex string, opts ...ModelOps) (*Transaction, error)
+	UpdateTransaction(ctx context.Context, txInfo *broadcast.SubmittedTx) error
 	UpdateTransactionMetadata(ctx context.Context, xPubID, id string, metadata Metadata) (*Transaction, error)
 	RevertTransaction(ctx context.Context, id string) error
 }

sync_tx_repository.go

@@ -30,6 +30,25 @@ func GetSyncTransactionByID(ctx context.Context, id string, opts ...ModelOps) (*
 	return txs[0], nil
 }
 
+// GetSyncTransactionByTxID will get a sync transaction by it's transaction id.
+func GetSyncTransactionByTxID(ctx context.Context, txID string, opts ...ModelOps) (*SyncTransaction, error) {
+	// Get the records by status
+	txs, err := _getSyncTransactionsByConditions(ctx,
+		map[string]interface{}{
+			idField: txID,
+		},
+		nil, opts...,
+	)
+	if err != nil {
+		return nil, err
+	}
+	if len(txs) != 1 {
+		return nil, nil
+	}
+
+	return txs[0], nil
+}
+
 /*** /exported funcs ***/
 
 /*** public unexported funcs ***/