feat: substrate end to end messages (#607)

* end to end substrate message

* fix tests

* fix comment

* doc update

DEVELOPMENT_GUIDE.md

@@ -19,6 +19,10 @@ To compile and run the package locally, you need to run the following command in
 1. Navigate into the directory of the repository
 2. Run: `go run cmd/mailchain/main.go account add --protocol=ethereum --key-type="secp256k1" --private-key=YOUR_PRIVATE_KEY`
 
+**To run substrate**
+
+1. Run: `docker-compose -f docker-compose.substrate.yml up --build`
+
 ### Inbox vs Inbox Staging
 
 From time to time, there may be pending breaking changes that have not been released. This means that if you use the web inbox, it may not work as desired.

Makefile

@@ -76,3 +76,6 @@ lint:
 	golangci-lint run --fix
 
 .FORCE:
+
+indexer-database-up:
+	go run cmd/indexer/main.go database up --master-postgres-password=mailchain --master-postgres-user=mailchain --indexer-postgres-password=indexer --envelope-postgres-password=envelope --pubkey-postgres-password=pubkey

cmd/indexer/commands/block-num.go

@@ -0,0 +1,16 @@
+package commands
+
+import (
+	"context"
+	"strconv"
+
+	"github.com/mailchain/mailchain/cmd/indexer/internal/clients"
+)
+
+func getBlockNumber(blockNumber string, client clients.Block) (uint64, error) {
+	if blockNumber == "latest" {
+		return client.LatestBlockNumber(context.Background())
+	}
+
+	return strconv.ParseUint(blockNumber, 0, 64)
+}

cmd/indexer/commands/ethereum.go

@@ -23,7 +23,7 @@ func ethereumCmd() *cobra.Command {
 		RunE: func(cmd *cobra.Command, args []string) error {
 			network, _ := cmd.Flags().GetString("network")
 			protocol, _ := cmd.Flags().GetString("protocol")
-			blockNumber, _ := cmd.Flags().GetUint64("start-block")
+			blockNumber, _ := cmd.Flags().GetString("start-block")
 
 			addressRPC, _ := cmd.Flags().GetString("rpc-address")
 			if addressRPC == "" {
@@ -62,22 +62,27 @@ func ethereumCmd() *cobra.Command {
 		},
 	}
 
-	cmd.Flags().Uint64("start-block", 0, "Block number from which the indexer will start")
+	cmd.Flags().String("start-block", "latest", "Block number from which the indexer will start, e.g. 10000, or 'latest'")
 	cmd.Flags().String("protocol", protocols.Ethereum, "Protocol to run against")
 	cmd.Flags().String("network", ethereum.Mainnet, "Network to run against")
 	cmd.Flags().String("rpc-address", "", "Ethereum RPC-JSON address")
 
 	return cmd
 }
 
-func createEthereumProcessor(connIndexer, connPublicKey, connEnvelope *sqlx.DB, blockNumber uint64, protocol, network, rawStorePath, addressRPC string) (*processor.Sequential, error) {
+func createEthereumProcessor(connIndexer, connPublicKey, connEnvelope *sqlx.DB, blockNumber, protocol, network, rawStorePath, addressRPC string) (*processor.Sequential, error) {
 	ctx := context.Background()
 
 	ethClient, err := eth.NewRPC(addressRPC)
 	if err != nil {
 		return nil, err
 	}
 
+	blockNo, err := getBlockNumber(blockNumber, ethClient)
+	if err != nil {
+		return nil, err
+	}
+
 	networkID, err := ethClient.NetworkID(ctx)
 	if err != nil {
 		return nil, err
@@ -119,7 +124,7 @@ func createEthereumProcessor(connIndexer, connPublicKey, connEnvelope *sqlx.DB,
 		chainCfg,
 	)
 
-	if err := syncStore.PutBlockNumber(ctx, protocol, network, blockNumber); err != nil {
+	if err := syncStore.PutBlockNumber(ctx, protocol, network, blockNo); err != nil {
 		return nil, err
 	}
 

cmd/indexer/commands/sequenial.go

@@ -4,16 +4,24 @@ import (
 	"context"
 	"fmt"
 
+	"github.com/cenkalti/backoff"
 	"github.com/mailchain/mailchain/cmd/indexer/internal/processor"
 	"github.com/spf13/cobra"
 )
 
 func doSequential(cmd *cobra.Command, p *processor.Sequential) {
 	for {
-		err := p.NextBlock(context.Background())
+		operation := func() error {
+			err := p.NextBlock(context.Background())
+			if err != nil {
+				fmt.Fprintf(cmd.ErrOrStderr(), "%+v\n", err)
+			}
 
-		if err != nil {
-			fmt.Fprintf(cmd.ErrOrStderr(), "%+v", err)
+			return err
+		}
+
+		if err := backoff.Retry(operation, backoff.NewExponentialBackOff()); err != nil {
+			fmt.Fprintf(cmd.ErrOrStderr(), "%+v\n", err)
 		}
 	}
 }

cmd/indexer/commands/substrate.go

@@ -6,7 +6,7 @@ import (
 	"github.com/jmoiron/sqlx"
 	"github.com/mailchain/mailchain/cmd/indexer/internal/processor"
 	sub "github.com/mailchain/mailchain/cmd/indexer/internal/substrate"
-	"github.com/mailchain/mailchain/cmd/internal/datastore/os"
+	"github.com/mailchain/mailchain/cmd/internal/datastore/noop"
 	"github.com/mailchain/mailchain/cmd/internal/datastore/pq"
 	"github.com/mailchain/mailchain/internal/protocols"
 	"github.com/mailchain/mailchain/internal/protocols/substrate"
@@ -22,15 +22,13 @@ func substrateCmd() *cobra.Command {
 		RunE: func(cmd *cobra.Command, args []string) error {
 			network, _ := cmd.Flags().GetString("network")
 			protocol, _ := cmd.Flags().GetString("protocol")
-			blockNumber, _ := cmd.Flags().GetUint64("start-block")
+			blockNumber, _ := cmd.Flags().GetString("start-block")
 
 			addressRPC, _ := cmd.Flags().GetString("rpc-address")
 			if addressRPC == "" {
 				return errors.Errorf("rpc-address must not be empty")
 			}
 
-			rawStorePath, _ := cmd.Flags().GetString("raw-store-path")
-
 			connIndexer, err := newPostgresConnection(cmd, "indexer")
 			if err != nil {
 				return err
@@ -50,7 +48,7 @@ func substrateCmd() *cobra.Command {
 			defer connPublicKey.Close()
 			defer connEnvelope.Close()
 
-			seqProcessor, err := createSubstrateProcessor(connIndexer, connPublicKey, connEnvelope, blockNumber, protocol, network, rawStorePath, addressRPC)
+			seqProcessor, err := createSubstrateProcessor(connIndexer, connPublicKey, connEnvelope, blockNumber, protocol, network, addressRPC)
 			if err != nil {
 				return err
 			}
@@ -61,22 +59,27 @@ func substrateCmd() *cobra.Command {
 		},
 	}
 
-	cmd.Flags().Uint64("start-block", 0, "Block number from which the indexer will start")
+	cmd.Flags().String("start-block", "latest", "Block number from which the indexer will start, e.g. 10000, or 'latest'")
 	cmd.Flags().String("protocol", protocols.Substrate, "Protocol to run against")
 	cmd.Flags().String("network", substrate.EdgewareMainnet, "Network to run against")
 	cmd.Flags().String("rpc-address", "", "Substrate RPC-JSON address")
 
 	return cmd
 }
 
-func createSubstrateProcessor(connIndexer, connPublicKey, connEnvelope *sqlx.DB, blockNumber uint64, protocol, network, rawStorePath, addressRPC string) (*processor.Sequential, error) {
+func createSubstrateProcessor(connIndexer, connPublicKey, connEnvelope *sqlx.DB, blockNumber, protocol, network, addressRPC string) (*processor.Sequential, error) {
 	ctx := context.Background()
 
 	subClient, err := sub.NewRPC(addressRPC)
 	if err != nil {
 		return nil, err
 	}
 
+	blockNo, err := getBlockNumber(blockNumber, subClient)
+	if err != nil {
+		return nil, err
+	}
+
 	syncStore, err := pq.NewSyncStore(connIndexer)
 	if err != nil {
 		return nil, err
@@ -92,18 +95,14 @@ func createSubstrateProcessor(connIndexer, connPublicKey, connEnvelope *sqlx.DB,
 		return nil, err
 	}
 
-	rawStore, err := os.NewRawTransactionStore(rawStorePath)
+	rawStore, err := noop.NewRawTransactionStore()
 	if err != nil {
 		return nil, err
 	}
 
-	processorTransaction := sub.NewExtrinsicProcessor(
-		transactionStore,
-		rawStore,
-		pubKeyStore,
-	)
+	processorTransaction := sub.NewExtrinsicProcessor(transactionStore, rawStore, pubKeyStore)
 
-	if err := syncStore.PutBlockNumber(ctx, protocol, network, blockNumber); err != nil {
+	if err := syncStore.PutBlockNumber(ctx, protocol, network, blockNo); err != nil {
 		return nil, err
 	}
 

cmd/indexer/internal/substrate/extrinsic.go

@@ -53,15 +53,6 @@ func (t *Extrinsic) Run(ctx context.Context, protocol, network string, tx interf
 	return actions.StoreTransaction(ctx, t.txStore, t.rawTxStore, protocol, network, storeTx, subEx)
 }
 
-// func (t *Transaction) From(blockNo *big.Int, tx *types.Transaction) ([]byte, error) {
-// 	msg, err := tx.AsMessage(types.MakeSigner(t.chainConfig, blockNo))
-// 	if err != nil {
-// 		return nil, err
-// 	}
-
-// 	return msg.From().Bytes(), nil
-// }
-
 func getFromPublicKey(sig *types.ExtrinsicSignatureV4) (crypto.PublicKey, error) {
 	if !sig.Signer.IsAccountID {
 		return nil, errors.Errorf("must be signed by account ID")
@@ -119,11 +110,6 @@ func (t *Extrinsic) ToTransaction(network string, blk *types.Block, ex *types.Ex
 
 	txInfo, data := getParts(w.Bytes())
 
-	decodedData, err := encoding.DecodeHex(string(data))
-	if err != nil {
-		return nil, err
-	}
-
 	from, err := getFromAddress(network, &ex.Signature)
 	if err != nil {
 		return nil, err
@@ -143,7 +129,7 @@ func (t *Extrinsic) ToTransaction(network string, blk *types.Block, ex *types.Ex
 		From: from,
 		// BlockHash: blk.Hash().Bytes(),
 		Hash: hash,
-		Data: decodedData,
+		Data: data,
 		To:   to,
 		// Value:    *value,
 		// GasUsed:  *gasUsed,
@@ -154,13 +140,13 @@ func (t *Extrinsic) ToTransaction(network string, blk *types.Block, ex *types.Ex
 func getParts(data []byte) (txInfo, dataField []byte) {
 	// mailchain hex encoded -> 0x6d61696c636861696e
 	// 0x6d61696c636861696e hex encoded -> 3078366436313639366336333638363136393665
-	pieces := bytes.Split(data, []byte{0x30, 0x78, 0x36, 0x64, 0x36, 0x31, 0x36, 0x39, 0x36, 0x63, 0x36, 0x33, 0x36, 0x38, 0x36, 0x31, 0x36, 0x39, 0x36, 0x65})
+	pieces := bytes.Split(data, encoding.DataPrefix())
 	txInfo = pieces[0]
 
 	if len(pieces) == 2 {
 		dataField = append(
 			// 0x30, 0x78 -> 0x
-			[]byte{0x36, 0x64, 0x36, 0x31, 0x36, 0x39, 0x36, 0x63, 0x36, 0x33, 0x36, 0x38, 0x36, 0x31, 0x36, 0x39, 0x36, 0x65},
+			encoding.DataPrefix(),
 			pieces[1]...,
 		)
 	}

cmd/indexer/internal/substrate/extrinsic_test.go

@@ -60,7 +60,7 @@ func TestExtrinsic_Run(t *testing.T) {
 				"substrate",
 				"edgeware-berlin",
 				func() *types.Extrinsic {
-					b := getBlock(t, "0x7ce3d93396dac53f1ae5fba268afcaa623e224e359507d581439dea791bab971")
+					b := getBlock(t, "0x61c83ac28eec9ac530df6f117edc761e3c3a0861f73dc8534f2e0682f1d9ef75")
 					e := b.Extrinsics[2]
 					return &e
 				}(),
@@ -98,36 +98,36 @@ func Test_ExtrinsicHash(t *testing.T) {
 			"success-timestamp-set",
 			args{
 				func() *types.Extrinsic {
-					b := getBlock(t, "0x7ce3d93396dac53f1ae5fba268afcaa623e224e359507d581439dea791bab971")
+					b := getBlock(t, "0x61c83ac28eec9ac530df6f117edc761e3c3a0861f73dc8534f2e0682f1d9ef75")
 					e := b.Extrinsics[0]
 					return &e
 				}(),
 			},
-			[]byte{0x94, 0x6c, 0x96, 0xef, 0x50, 0x2b, 0x4f, 0x4, 0x6e, 0x4f, 0xae, 0x52, 0x9, 0x52, 0x3a, 0xe3, 0xaa, 0x59, 0xa5, 0x66, 0x49, 0x2d, 0xf6, 0x4e, 0xdf, 0x38, 0x10, 0xd0, 0x35, 0x7c, 0x35, 0x51},
+			[]byte{0xc7, 0x6e, 0xcb, 0x62, 0x2e, 0xca, 0x6e, 0x4d, 0xdb, 0xab, 0xd3, 0xfc, 0x50, 0xe, 0x59, 0x9c, 0x7e, 0xce, 0xb, 0x87, 0x16, 0x9d, 0x39, 0x2f, 0xa1, 0xab, 0xc6, 0xce, 0x8f, 0xca, 0x20, 0x32},
 			false,
 		},
 		{
 			"success-final-hint",
 			args{
 				func() *types.Extrinsic {
-					b := getBlock(t, "0x7ce3d93396dac53f1ae5fba268afcaa623e224e359507d581439dea791bab971")
+					b := getBlock(t, "0x61c83ac28eec9ac530df6f117edc761e3c3a0861f73dc8534f2e0682f1d9ef75")
 					e := b.Extrinsics[1]
 					return &e
 				}(),
 			},
-			[]byte{0x49, 0xf0, 0xb1, 0x75, 0x20, 0xe9, 0x77, 0x4f, 0x8d, 0x86, 0xf4, 0xba, 0xae, 0x6, 0x15, 0xe9, 0x2c, 0x54, 0xbd, 0x9f, 0x87, 0x35, 0xd6, 0xe8, 0xa7, 0xab, 0xeb, 0x8c, 0x76, 0x2a, 0x67, 0xd4},
+			[]byte{0x18, 0xd0, 0x2a, 0x17, 0x69, 0x9d, 0x50, 0xba, 0xe, 0x42, 0xa2, 0x7a, 0xa3, 0x52, 0x6d, 0x36, 0x4a, 0x9, 0xbe, 0x30, 0x54, 0xe8, 0x49, 0xa5, 0x74, 0x47, 0x8d, 0xa5, 0x50, 0x5a, 0x4c, 0xc2},
 			false,
 		},
 		{
 			"success-contract",
 			args{
 				func() *types.Extrinsic {
-					b := getBlock(t, "0x7ce3d93396dac53f1ae5fba268afcaa623e224e359507d581439dea791bab971")
+					b := getBlock(t, "0x61c83ac28eec9ac530df6f117edc761e3c3a0861f73dc8534f2e0682f1d9ef75")
 					e := b.Extrinsics[2]
 					return &e
 				}(),
 			},
-			[]byte{0xdc, 0xd9, 0x48, 0x65, 0x81, 0x56, 0x13, 0x1f, 0x43, 0xb7, 0x8a, 0x4c, 0x85, 0xe1, 0x63, 0xa8, 0xa1, 0xaa, 0x33, 0x65, 0xa3, 0x92, 0xf4, 0x9e, 0x51, 0x43, 0x65, 0x20, 0x57, 0xfc, 0xf5, 0xe9},
+			[]byte{0x3f, 0xcb, 0x9c, 0x48, 0x7d, 0x1a, 0xc, 0x28, 0xfe, 0x27, 0xa7, 0x73, 0xcd, 0x9d, 0x20, 0x12, 0x48, 0x25, 0x33, 0xd9, 0x6, 0x4a, 0xed, 0x91, 0xb2, 0x98, 0x3, 0xaa, 0xcc, 0x3d, 0x4a, 0x9e},
 			false,
 		},
 	}

cmd/indexer/internal/substrate/testdata/blocks/0x61c83ac28eec9ac530df6f117edc761e3c3a0861f73dc8534f2e0682f1d9ef75.json

@@ -0,0 +1,26 @@
+{
+    "jsonrpc": "2.0",
+    "result": {
+        "block": {
+            "extrinsics": [
+                "0x280402000b40f1a9737301",
+                "0x1c040b00eada8e00",
+                "0x490484ff169a11721851f5dff3541dd5c4b0b478ac1cd092c9d5976e83daa0d03f26620c0164abcbfe0a217090472c5c173222e9f9f35dc17c8cce8db860a29571eef574288545e2d09772f043aa6e2858bd0421640b25f666af8ff16535ccc6af398f01850030000e02ff84623e7252e41138af6904e1b02304c941625f39e5762589125dc1a2f2cf2e300002f4010009026d61696c636861696e010a6e2ae344da21829b0b22ab9cfead2ada180ee30318ae9279988275688b059de8d2e6487a7000bf9de65744ccf32dfa2a1f834a6e4f444b45f1257f922ec726b19485a5bbb83bee0c206a786516110d05f583b1c4dd8a3826369c999611705be7ffb419363e2c9797ec5f32b6661a4612062204a3b3ba0d"
+            ],
+            "header": {
+                "digest": {
+                    "logs": [
+                        "0x0661757261208c3ed90f00000000",
+                        "0x0561757261010149a7f5bdc38daac2f75942f0c543eda761241284f4b737c9c98125ac52187c5fdfc017a24564aed5018a75fc131e3e9f21f35a419befd128d7abe642a126780e"
+                    ]
+                },
+                "extrinsicsRoot": "0x607f22bb3237dc3aa4e705402f346f661c0a479004e9ca50f3e0a26e5f70a34f",
+                "number": "0x23b6bd",
+                "parentHash": "0x52626c79f0e7008112ab337595597950174e08cbae0cef1a65fbf80e8698cde9",
+                "stateRoot": "0x08554277fafec2afe218c66f29b2986cd000e438b36b65168471353221b5134a"
+            }
+        },
+        "justification": null
+    },
+    "id": 22
+}