Blockchain events subscription

blockchain/client.go

@@ -1,14 +1,29 @@
 package blockchain
 
 import (
+	"fmt"
+	"math"
+	"sync"
+	"sync/atomic"
+
 	gsrpc "github.com/centrifuge/go-substrate-rpc-client/v4"
+	"github.com/centrifuge/go-substrate-rpc-client/v4/types"
+	"github.com/centrifuge/go-substrate-rpc-client/v4/types/codec"
 
 	"github.com/cerebellum-network/cere-ddc-sdk-go/blockchain/pallets"
 )
 
+type EventsListener func(*pallets.Events)
+
 type Client struct {
 	*gsrpc.SubstrateAPI
 
+	eventsListeners map[int]EventsListener
+	mu              sync.Mutex
+	isListening     uint32
+	stopListening   func()
+	errsListening   chan error
+
 	DdcClusters  pallets.DdcClustersApi
 	DdcCustomers pallets.DdcCustomersApi
 	DdcNodes     pallets.DdcNodesApi
@@ -26,10 +41,97 @@ func NewClient(url string) (*Client, error) {
 	}
 
 	return &Client{
-		SubstrateAPI: substrateApi,
-		DdcClusters:  pallets.NewDdcClustersApi(substrateApi),
-		DdcCustomers: pallets.NewDdcCustomersApi(substrateApi, meta),
-		DdcNodes:     pallets.NewDdcNodesApi(substrateApi, meta),
-		DdcPayouts:   pallets.NewDdcPayoutsApi(substrateApi, meta),
+		SubstrateAPI:    substrateApi,
+		eventsListeners: make(map[int]EventsListener),
+		DdcClusters:     pallets.NewDdcClustersApi(substrateApi),
+		DdcCustomers:    pallets.NewDdcCustomersApi(substrateApi, meta),
+		DdcNodes:        pallets.NewDdcNodesApi(substrateApi, meta),
+		DdcPayouts:      pallets.NewDdcPayoutsApi(substrateApi, meta),
 	}, nil
 }
+
+func (c *Client) StartEventsListening() (func(), <-chan error, error) {
+	if !atomic.CompareAndSwapUint32(&c.isListening, 0, 1) {
+		return c.stopListening, c.errsListening, nil
+	}
+
+	meta, err := c.RPC.State.GetMetadataLatest()
+	if err != nil {
+		return nil, nil, err
+	}
+	key, err := types.CreateStorageKey(meta, "System", "Events", nil)
+	if err != nil {
+		return nil, nil, err
+	}
+	sub, err := c.RPC.State.SubscribeStorageRaw([]types.StorageKey{key})
+	if err != nil {
+		return nil, nil, err
+	}
+
+	done := make(chan struct{})
+	c.errsListening = make(chan error)
+
+	go func() {
+		for {
+			select {
+			case <-done:
+				return
+			case set := <-sub.Chan():
+				for _, change := range set.Changes {
+					if !codec.Eq(change.StorageKey, key) || !change.HasStorageData {
+						continue
+					}
+
+					events := &pallets.Events{}
+					err = types.EventRecordsRaw(change.StorageData).DecodeEventRecords(meta, events)
+					if err != nil {
+						c.errsListening <- fmt.Errorf("events decoder: %w", err)
+					}
+
+					for _, callback := range c.eventsListeners {
+						go callback(events)
+					}
+				}
+			}
+		}
+	}()
+
+	once := sync.Once{}
+	c.stopListening = func() {
+		once.Do(func() {
+			done <- struct{}{}
+			sub.Unsubscribe()
+			c.isListening = 0
+		})
+	}
+
+	return c.stopListening, c.errsListening, nil
+}
+
+func (c *Client) RegisterEventsListener(callback EventsListener) (func(), error) {
+	var idx int
+	for i := 0; i <= math.MaxInt; i++ {
+		if _, ok := c.eventsListeners[i]; !ok {
+			idx = i
+			break
+		}
+		if i == math.MaxInt {
+			return nil, fmt.Errorf("too many events listeners")
+		}
+	}
+
+	c.mu.Lock()
+	c.eventsListeners[idx] = callback
+	c.mu.Unlock()
+
+	once := sync.Once{}
+	stop := func() {
+		once.Do(func() {
+			c.mu.Lock()
+			delete(c.eventsListeners, idx)
+			c.mu.Unlock()
+		})
+	}
+
+	return stop, nil
+}

blockchain/go.mod

@@ -2,7 +2,7 @@ module github.com/cerebellum-network/cere-ddc-sdk-go/blockchain
 
 go 1.18
 
-require github.com/centrifuge/go-substrate-rpc-client/v4 v4.0.8
+require github.com/centrifuge/go-substrate-rpc-client/v4 v4.2.1
 
 require (
 	github.com/ChainSafe/go-schnorrkel v1.0.0 // indirect
@@ -18,7 +18,7 @@ require (
 	github.com/mimoo/StrobeGo v0.0.0-20220103164710-9a04d6ca976b // indirect
 	github.com/pierrec/xxHash v0.1.5 // indirect
 	github.com/rs/cors v1.8.2 // indirect
-	github.com/vedhavyas/go-subkey v1.0.3 // indirect
+	github.com/vedhavyas/go-subkey/v2 v2.0.0 // indirect
 	golang.org/x/crypto v0.7.0 // indirect
 	golang.org/x/sys v0.6.0 // indirect
 	gopkg.in/natefinch/npipe.v2 v2.0.0-20160621034901-c1b8fa8bdcce // indirect

blockchain/go.sum

@@ -1,11 +1,10 @@
 github.com/ChainSafe/go-schnorrkel v1.0.0 h1:3aDA67lAykLaG1y3AOjs88dMxC88PgUuHRrLeDnvGIM=
 github.com/ChainSafe/go-schnorrkel v1.0.0/go.mod h1:dpzHYVxLZcp8pjlV+O+UR8K0Hp/z7vcchBSbMBEhCw4=
 github.com/StackExchange/wmi v0.0.0-20180116203802-5d049714c4a6 h1:fLjPD/aNc3UIOA6tDi6QXUemppXK3P9BI7mr2hd6gx8=
-github.com/btcsuite/btcd v0.22.0-beta h1:LTDpDKUM5EeOFBPM8IXpinEcmZ6FWfNZbE3lfrfdnWo=
 github.com/btcsuite/btcd/btcec/v2 v2.2.0 h1:fzn1qaOt32TuLjFlkzYSsBC35Q3KUjT1SwPxiMSCF5k=
 github.com/btcsuite/btcutil v1.0.3-0.20201208143702-a53e38424cce h1:YtWJF7RHm2pYCvA5t0RPmAaLUhREsKuKd+SLhxFbFeQ=
-github.com/centrifuge/go-substrate-rpc-client/v4 v4.0.8 h1:gHLD5S81As9u5DbefLahw1enVO6OdBSS8gBI2R6KNEQ=
-github.com/centrifuge/go-substrate-rpc-client/v4 v4.0.8/go.mod h1:5g1oM4Zu3BOaLpsKQ+O8PAv2kNuq+kPcA1VzFbsSqxE=
+github.com/centrifuge/go-substrate-rpc-client/v4 v4.2.1 h1:io49TJ8IOIlzipioJc9pJlrjgdJvqktpUWYxVY5AUjE=
+github.com/centrifuge/go-substrate-rpc-client/v4 v4.2.1/go.mod h1:k61SBXqYmnZO4frAJyH3iuqjolYrYsq79r8EstmklDY=
 github.com/cosmos/go-bip39 v0.0.0-20180819234021-555e2067c45d/go.mod h1:tSxLoYXyBmiFeKpvmq4dzayMdCjCnu8uqmCysIGBT2Y=
 github.com/cosmos/go-bip39 v1.0.0 h1:pcomnQdrdH22njcAatO0yWojsUnCO3y2tNoV1cb6hHY=
 github.com/cosmos/go-bip39 v1.0.0/go.mod h1:RNJv0H/pOIVgxw6KS7QeX2a0Uo0aKUlfhZ4xuwvCdJw=
@@ -47,8 +46,8 @@ github.com/stretchr/testify v1.6.1/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/
 github.com/stretchr/testify v1.7.2 h1:4jaiDzPyXQvSd7D0EjG45355tLlV3VOECpq10pLC+8s=
 github.com/tklauser/go-sysconf v0.3.5 h1:uu3Xl4nkLzQfXNsWn15rPc/HQCJKObbt1dKJeWp3vU4=
 github.com/tklauser/numcpus v0.2.2 h1:oyhllyrScuYI6g+h/zUvNXNp1wy7x8qQy3t/piefldA=
-github.com/vedhavyas/go-subkey v1.0.3 h1:iKR33BB/akKmcR2PMlXPBeeODjWLM90EL98OrOGs8CA=
-github.com/vedhavyas/go-subkey v1.0.3/go.mod h1:CloUaFQSSTdWnINfBRFjVMkWXZANW+nd8+TI5jYcl6Y=
+github.com/vedhavyas/go-subkey/v2 v2.0.0 h1:LemDIsrVtRSOkp0FA8HxP6ynfKjeOj3BY2U9UNfeDMA=
+github.com/vedhavyas/go-subkey/v2 v2.0.0/go.mod h1:95aZ+XDCWAUUynjlmi7BtPExjXgXxByE0WfBwbmIRH4=
 golang.org/x/crypto v0.0.0-20190308221718-c2843e01d9a2/go.mod h1:djNgcEr1/C05ACkg1iLfiJU5Ep61QUkGW8qpdssI0+w=
 golang.org/x/crypto v0.0.0-20191206172530-e9b2fee46413/go.mod h1:LzIPMQfyMNhhGPhUkYOs5KpL4U8rLKemX1yGLhDgUto=
 golang.org/x/crypto v0.0.0-20200728195943-123391ffb6de/go.mod h1:LzIPMQfyMNhhGPhUkYOs5KpL4U8rLKemX1yGLhDgUto=

blockchain/pallets/ddcclusters.go

@@ -21,6 +21,41 @@ type ClusterProps struct {
 	NodeProviderAuthContract types.AccountID
 }
 
+// Events
+type (
+	EventDdcClustersClusterCreated struct {
+		Phase     types.Phase
+		ClusterId ClusterId
+		Topics    []types.Hash
+	}
+
+	EventDdcClustersClusterNodeAdded struct {
+		Phase      types.Phase
+		ClusterId  ClusterId
+		NodePubKey NodePubKey
+		Topics     []types.Hash
+	}
+
+	EventDdcClustersClusterNodeRemoved struct {
+		Phase      types.Phase
+		ClusterId  ClusterId
+		NodePubKey NodePubKey
+		Topics     []types.Hash
+	}
+
+	EventDdcClustersClusterParamsSet struct {
+		Phase     types.Phase
+		ClusterId ClusterId
+		Topics    []types.Hash
+	}
+
+	EventDdcClustersClusterGovParamsSet struct {
+		Phase     types.Phase
+		ClusterId ClusterId
+		Topics    []types.Hash
+	}
+)
+
 type DdcClustersApi interface {
 	GetClustersNodes(clusterId ClusterId) ([]NodePubKey, error)
 }

blockchain/pallets/ddccustomers.go

@@ -25,6 +25,44 @@ type UnlockChunk struct {
 	Block types.BlockNumber
 }
 
+// Events
+type (
+	EventDdcCustomersDeposited struct {
+		Phase  types.Phase
+		Owner  types.AccountID
+		Amount types.U128
+		Topics []types.Hash
+	}
+	EventDdcCustomersInitialDepositUnlock struct {
+		Phase  types.Phase
+		Owner  types.AccountID
+		Amount types.U128
+		Topics []types.Hash
+	}
+	EventDdcCustomersWithdrawn struct {
+		Phase  types.Phase
+		Owner  types.AccountID
+		Amount types.U128
+		Topics []types.Hash
+	}
+	EventDdcCustomersCharged struct {
+		Phase  types.Phase
+		Owner  types.AccountID
+		Amount types.U128
+		Topics []types.Hash
+	}
+	EventDdcCustomersBucketCreated struct {
+		Phase    types.Phase
+		BucketId BucketId
+		Topics   []types.Hash
+	}
+	EventDdcCustomersBucketUpdated struct {
+		Phase    types.Phase
+		BucketId BucketId
+		Topics   []types.Hash
+	}
+)
+
 type DdcCustomersApi interface {
 	GetBuckets(bucketId BucketId) (types.Option[Bucket], error)
 	GetBucketsCount() (types.U64, error)
@@ -36,9 +74,9 @@ type ddcCustomersApi struct {
 	meta         *types.Metadata
 }
 
-func NewDdcCustomersApi(substrateAPI *gsrpc.SubstrateAPI, meta *types.Metadata) DdcCustomersApi {
+func NewDdcCustomersApi(substrateApi *gsrpc.SubstrateAPI, meta *types.Metadata) DdcCustomersApi {
 	return &ddcCustomersApi{
-		substrateAPI,
+		substrateApi,
 		meta,
 	}
 }

blockchain/pallets/ddcnodes.go

@@ -40,9 +40,9 @@ type ddcNodesApi struct {
 	meta         *types.Metadata
 }
 
-func NewDdcNodesApi(substrateAPI *gsrpc.SubstrateAPI, meta *types.Metadata) DdcNodesApi {
+func NewDdcNodesApi(substrateApi *gsrpc.SubstrateAPI, meta *types.Metadata) DdcNodesApi {
 	return &ddcNodesApi{
-		substrateAPI,
+		substrateApi,
 		meta,
 	}
 }