Baseledger p2p client

cmd/statsdaemon/main_test.go

@@ -1,17 +1,135 @@
 package main
 
 import (
+	"encoding/json"
+	"fmt"
 	"testing"
 
+	uuid "github.com/kthomas/go.uuid"
 	. "github.com/onsi/ginkgo"
 	. "github.com/onsi/gomega"
+
+	"github.com/provideplatform/ident/common"
+	"github.com/provideplatform/nchain/network"
+	ident "github.com/provideplatform/provide-go/api/ident"
+	"github.com/provideplatform/provide-go/api/nchain"
 )
 
-func TestNChainStatsdaemon(t *testing.T) {
+type chainSpecConfig struct{}
+
+type chainSpec struct {
+	Config *chainSpecConfig `json:"config"`
+}
+
+type chainConfig struct {
+	NativeCurrency      *string    `json:"native_currency"`
+	IsBaseledgerNetwork bool       `json:"is_baseledger_network"`
+	Client              *string    `json:"client"`
+	BlockExplorerUrl    *string    `json:"block_explorer_url"`
+	JsonRpcUrl          *string    `json:"json_rpc_url"`
+	WebsocketUrl        *string    `json:"websocket_url"`
+	Platform            *string    `json:"platform"`
+	EngineID            *string    `json:"engine_id"`
+	Chain               *string    `json:"chain"`
+	ProtocolID          *string    `json:"protocol_id"`
+	NetworkID           int        `json:"network_id"`
+	ChainSpec           *chainSpec `json:"chainspec"`
+}
+
+type chainDef struct {
+	Name    *string      `json:"name"`
+	Enabled bool         `json:"enabled"`
+	Config  *chainConfig `json:"config"`
+}
+
+func SetupBaseledgerTestNetwork() (*network.Network, error) {
+	testID, _ := uuid.NewV4()
+
+	email := "prvd" + testID.String() + "@email.com"
+	pwd := "super_secret"
+	_, err := ident.CreateUser("", map[string]interface{}{
+		"first_name": "statsdaemon first name" + testID.String(),
+		"last_name":  "statsdaemon last name" + testID.String(),
+		"email":      email,
+		"password":   pwd,
+	})
+	if err != nil {
+		return nil, fmt.Errorf("error creating user. Error: %s", err.Error())
+	}
+
+	authResponse, _ := ident.Authenticate(email, pwd)
+	if err != nil {
+		return nil, fmt.Errorf("error authenticating user. Error: %s", err.Error())
+	}
+
+	chainySpecConfig := chainSpecConfig{}
+	chainySpec := chainSpec{
+		Config: &chainySpecConfig,
+	}
+	chainyConfig := chainConfig{
+		NativeCurrency:      common.StringOrNil("token"),
+		Platform:            common.StringOrNil("tendermint"),
+		Client:              common.StringOrNil("baseledger"),
+		NetworkID:           3,
+		IsBaseledgerNetwork: true,
+		Chain:               common.StringOrNil("peachtree"),
+		WebsocketUrl:        common.StringOrNil("ws://genesis.peachtree.baseledger.provide.network:1337/websocket"),
+		ChainSpec:           &chainySpec,
+	}
+
+	chainName := fmt.Sprintf("Baseledger Testnet %s", testID.String())
+
+	chainyChain := chainDef{
+		Name:   common.StringOrNil(chainName),
+		Config: &chainyConfig,
+	}
+
+	chainyChainJSON, _ := json.Marshal(chainyChain)
+
+	params := map[string]interface{}{}
+	json.Unmarshal(chainyChainJSON, &params)
+
+	testNetwork, err := nchain.CreateNetwork(*authResponse.Token.AccessToken, params)
+	if err != nil {
+		return nil, fmt.Errorf("error creating network. Error: %s", err.Error())
+	}
+
+	return &network.Network{
+		ApplicationID: testNetwork.ApplicationID,
+		UserID:        testNetwork.UserID,
+		Name:          testNetwork.Name,
+		Description:   testNetwork.Description,
+		Enabled:       testNetwork.Enabled,
+		ChainID:       testNetwork.ChainID,
+		NetworkID:     testNetwork.NetworkID,
+		Config:        testNetwork.Config,
+	}, nil
+}
+
+func TestNChainBaseledgerStatsdaemon(t *testing.T) {
 	RegisterFailHandler(Fail)
 	RunSpecs(t, "NChain statsdaemon Suite")
 }
 
 var _ = Describe("Main", func() {
+	It("Should parse one successful baseledger block header event", func() {
+		testNetwork, err := SetupBaseledgerTestNetwork()
+		if err != nil {
+			Fail("Failed to set up baseledger test network")
+		}
+
+		statsDaemon := RequireNetworkStatsDaemon(testNetwork)
+		// get one result and shutdown statsdaemon and check result
+		sampleResult := <-statsDaemon.queue
+		EvictNetworkStatsDaemon(testNetwork)
+
+		jsonSampleResult, _ := json.Marshal(sampleResult.Meta["last_block_header"])
+		formattedSampleHeaderResult := nchain.BaseledgerBlockHeaderResponse{}.Value.Header
+		err = json.Unmarshal(jsonSampleResult, &formattedSampleHeaderResult)
+		if err != nil {
+			Fail("Failed to unmarshall header response")
+		}
 
+		Expect(formattedSampleHeaderResult).NotTo(BeNil())
+	})
 })

cmd/statsdaemon/stats_daemon.go

@@ -26,6 +26,7 @@ import (
 	uuid "github.com/kthomas/go.uuid"
 	"github.com/provideplatform/nchain/common"
 	"github.com/provideplatform/nchain/network"
+	"github.com/provideplatform/provide-go/api/nchain"
 	provide "github.com/provideplatform/provide-go/api/nchain"
 	providecrypto "github.com/provideplatform/provide-go/crypto"
 )
@@ -333,6 +334,77 @@ func EthereumNetworkStatsDataSourceFactory(network *network.Network) *NetworkSta
 	}
 }
 
+// BaseledgerNetworkStatsDataSourceFactory builds and returns a JSON-RPC and streaming websocket
+// data source which is used by stats daemon instances to consume EVM-based network statistics
+func BaseledgerNetworkStatsDataSourceFactory(network *network.Network) *NetworkStatsDataSource {
+	return &NetworkStatsDataSource{
+		Network: network,
+
+		Poll: func(ch chan *provide.NetworkStatus) error {
+			return new(jsonRpcNotSupported)
+		},
+
+		Stream: func(ch chan *provide.NetworkStatus) error {
+			websocketURL := network.WebsocketURL()
+			if websocketURL == "" {
+				err := new(websocketNotSupported)
+				return *err
+			}
+			var wsDialer websocket.Dialer
+			wsConn, _, err := wsDialer.Dial(websocketURL, nil)
+			if err != nil {
+				common.Log.Errorf("Failed to establish network stats websocket connection to %s; %s", websocketURL, err.Error())
+			} else {
+				defer wsConn.Close()
+				// { "jsonrpc": "2.0", "method": "subscribe", "params": ["tm.event='NewBlock'"], "id": 1 }
+				payload := map[string]interface{}{
+					"method":  "subscribe",
+					"params":  []string{"tm.event='NewBlockHeader'"},
+					"id":      1,
+					"jsonrpc": "2.0",
+				}
+				if err := wsConn.WriteJSON(payload); err != nil {
+					common.Log.Errorf("Failed to write subscribe message to network stats websocket connection")
+				} else {
+					common.Log.Debugf("Subscribed to network stats websocket: %s", websocketURL)
+					for {
+						_, message, err := wsConn.ReadMessage()
+						if err != nil {
+							common.Log.Errorf("Failed to receive message on network stats websocket; %s", err)
+							break
+						} else {
+							common.Log.Debugf("Received %d-byte message on network stats websocket for network: %s", len(message), *network.Name)
+							response := &provide.BaseledgerSubscriptionResponse{}
+							err := json.Unmarshal(message, response)
+							if err != nil {
+								common.Log.Warningf("Failed to unmarshal message received on network stats websocket: %s; %s", message, err.Error())
+							} else {
+								if result, ok := response.Result["data"].(map[string]interface{}); ok {
+									if resultJSON, err := json.Marshal(result); err == nil {
+										headerResponse := &nchain.BaseledgerBlockHeaderResponse{}
+										err := json.Unmarshal(resultJSON, headerResponse)
+										if err != nil {
+											common.Log.Warningf("Failed to stringify result JSON in otherwise valid message received on network stats websocket: %s; %s", response, err.Error())
+										} else if headerResponse != nil {
+											common.Log.Debugf("Block header received %v\n", headerResponse.Value.Header)
+											ch <- &provide.NetworkStatus{
+												Meta: map[string]interface{}{
+													"last_block_header": headerResponse.Value.Header,
+												},
+											}
+										}
+									}
+								}
+							}
+						}
+					}
+				}
+			}
+			return err
+		},
+	}
+}
+
 // Consume the websocket stream; attempts to fallback to JSON-RPC if websocket stream fails or is not available for the network
 func (sd *StatsDaemon) consume() []error {
 	errs := make([]error, 0)
@@ -700,6 +772,8 @@ func NewNetworkStatsDaemon(lg *logger.Logger, network *network.Network) *StatsDa
 		sd.dataSource = BcoinNetworkStatsDataSourceFactory(network)
 	} else if network.IsEthereumNetwork() {
 		sd.dataSource = EthereumNetworkStatsDataSourceFactory(network)
+	} else if network.IsBaseledgerNetwork() {
+		sd.dataSource = BaseledgerNetworkStatsDataSourceFactory(network)
 	}
 	//sd.handleSignals()
 

go.mod

@@ -50,7 +50,7 @@ require (
 	github.com/onsi/gomega v1.10.1
 	github.com/prometheus/tsdb v0.10.0 // indirect
 	github.com/provideplatform/ident v0.9.10-0.20210727215953-cdd9940805ce
-	github.com/provideplatform/provide-go v0.0.0-20210701150626-bff9948fbd9a
+	github.com/provideplatform/provide-go v0.0.0-20210822183020-1c080cdc4a77
 	github.com/spaolacci/murmur3 v1.1.1-0.20190317074736-539464a789e9 // indirect
 	github.com/status-im/keycard-go v0.0.0-20191119114148-6dd40a46baa0 // indirect
 	go.mongodb.org/mongo-driver v1.5.1

go.sum

@@ -576,8 +576,8 @@ github.com/prometheus/tsdb v0.10.0/go.mod h1:oi49uRhEe9dPUTlS3JRZOwJuVi6tmh10QSg
 github.com/provideplatform/ident v0.9.10-0.20210727215953-cdd9940805ce h1:0JZPWGnkBdFRGN/ptm49gMebcmMkLqnRukExazrvwCQ=
 github.com/provideplatform/ident v0.9.10-0.20210727215953-cdd9940805ce/go.mod h1:A0IzIOtqzS+MgpSTTKSQITY4kPH48IsGXw5pDCcegTU=
 github.com/provideplatform/provide-go v0.0.0-20210624064849-d7328258f0d8/go.mod h1:q0/Q8KaZxYg84rdwBIIE7ZwHluzM5zw7zJJoJOqAbzg=
-github.com/provideplatform/provide-go v0.0.0-20210701150626-bff9948fbd9a h1:9NekZCQm1lPSlh76Vu+mSlrr99ofxYMzdzltBCHcu9Q=
-github.com/provideplatform/provide-go v0.0.0-20210701150626-bff9948fbd9a/go.mod h1:q0/Q8KaZxYg84rdwBIIE7ZwHluzM5zw7zJJoJOqAbzg=
+github.com/provideplatform/provide-go v0.0.0-20210822183020-1c080cdc4a77 h1:+/LcHOVu85ySEFunR/6f8k4GcBBb/xug86pGHkohc6I=
+github.com/provideplatform/provide-go v0.0.0-20210822183020-1c080cdc4a77/go.mod h1:q0/Q8KaZxYg84rdwBIIE7ZwHluzM5zw7zJJoJOqAbzg=
 github.com/provideservices/provide-go v0.0.0-20210409104111-70ad008e4ae8 h1:QaP1R8tsO69Pt5FQ2y8JZNpH5Dwk5xeNRcoethNcx3U=
 github.com/provideservices/provide-go v0.0.0-20210409104111-70ad008e4ae8/go.mod h1:zMvfECorm5+hb9aewWEVpxkaFGko3r5Qk3/N/aIuLeY=
 github.com/rcrowley/go-metrics v0.0.0-20181016184325-3113b8401b8a/go.mod h1:bCqnVzQkZxMG4s8nGwiZ5l3QUCyqpo9Y+/ZMZ9VjZe4=

network/network.go

@@ -55,6 +55,7 @@ const networkConfigIsHandshakeNetwork = "is_handshake_network"
 const networkConfigIsHyperledgerBesuNetwork = "is_hyperledger_besu_network"
 const networkConfigIsHyperledgerFabricNetwork = "is_hyperledger_fabric_network"
 const networkConfigIsQuorumNetwork = "is_quorum_network"
+const networkConfigIsBaseledgerNetwork = "is_baseledger_network"
 
 const networkConfigEnvBootnodes = "BOOTNODES"
 const networkConfigEnvClient = "CLIENT"
@@ -868,6 +869,16 @@ func (n *Network) IsHandshakeNetwork() bool {
 	return false
 }
 
+func (n *Network) IsBaseledgerNetwork() bool {
+	cfg := n.ParseConfig()
+	if cfg != nil {
+		if isBaseledgerNetwork, ok := cfg[networkConfigIsBaseledgerNetwork].(bool); ok {
+			return isBaseledgerNetwork
+		}
+	}
+	return false
+}
+
 // P2PAPIClient returns an instance of the network's underlying p2p.API, if that is possible given the network config
 func (n *Network) P2PAPIClient() (p2p.API, error) {
 	cfg := n.ParseConfig()
@@ -897,6 +908,8 @@ func (n *Network) P2PAPIClient() (p2p.API, error) {
 		apiClient = p2p.InitParityP2PProvider(common.StringOrNil(rpcURL), n.ID.String(), n)
 	case p2p.ProviderQuorum:
 		apiClient = p2p.InitQuorumP2PProvider(common.StringOrNil(rpcURL), n.ID.String(), n)
+	case p2p.ProviderBaseledger:
+		apiClient = p2p.InitBaseledgerP2PProvider(common.StringOrNil(rpcURL), n.ID.String(), n)
 	default:
 		return nil, fmt.Errorf("Failed to resolve p2p provider for network %s; unsupported client", n.ID)
 	}

network/node.go

@@ -735,6 +735,8 @@ func (n *Node) P2PAPIClient() (p2p.API, error) {
 	var apiClient p2p.API
 
 	switch client {
+	case p2p.ProviderBaseledger:
+		apiClient = p2p.InitBaseledgerP2PProvider(rpcURL, n.NetworkID.String(), n.Network)
 	case p2p.ProviderBcoin:
 		return nil, fmt.Errorf("Bcoin p2p provider not yet implemented")
 	case p2p.ProviderGeth: