initial commit

cmd/indexer/internal/clients/block.go

@@ -0,0 +1,7 @@
+package clients
+
+import "context"
+
+type BlockByNumber interface {
+	Get(ctx context.Context, blockNo uint64) (blk interface{}, err error)
+}

cmd/indexer/internal/datastore/public_key.go

@@ -1,10 +1,12 @@
 package datastore
 
 import (
+	"context"
+
 	"github.com/mailchain/mailchain/crypto"
 )
 
 type PublicKeyStore interface {
-	PutPublicKey(protocol, network string, address []byte, pubKey crypto.PublicKey) error
-	GetPublicKey(protocol, network string, address []byte) (pubKey crypto.PublicKey, err error)
+	PutPublicKey(ctx context.Context, protocol, network string, address []byte, pubKey crypto.PublicKey) error
+	GetPublicKey(ctx context.Context, protocol, network string, address []byte) (pubKey crypto.PublicKey, err error)
 }

cmd/indexer/internal/datastore/sync.go

@@ -1,6 +1,8 @@
 package datastore
 
+import "context"
+
 type SyncStore interface {
-	GetBlockNumber(protocol, network string) (blockNo uint64, err error)
-	PutBlockNumber(protocol, network string, blockNo uint64) error
+	GetBlockNumber(ctx context.Context, protocol, network string) (blockNo uint64, err error)
+	PutBlockNumber(ctx context.Context, protocol, network string, blockNo uint64) error
 }

cmd/indexer/internal/datastore/transaction.go

@@ -0,0 +1,27 @@
+package datastore
+
+import (
+	"context"
+	"math/big"
+)
+
+type Transaction struct {
+	From      []byte
+	To        []byte
+	Data      []byte
+	BlockHash []byte
+	Hash      []byte
+	Value     big.Int
+	GasUsed   big.Int
+	GasPrice  big.Int
+}
+
+type TransactionStore interface {
+	PutTransaction(ctx context.Context, protocol, network string, hash []byte, tx *Transaction) error
+	GetTransactionsFrom(ctx context.Context, protocol, network string, address []byte) ([]Transaction, error)
+	GetTransactionsTo(ctx context.Context, protocol, network string, address []byte) ([]Transaction, error)
+}
+
+type RawTransactionStore interface {
+	PutRawTransaction(ctx context.Context, protocol, network string, hash []byte, tx interface{}) error
+}

cmd/indexer/internal/ethereum/block.go

@@ -0,0 +1,29 @@
+package ethereum
+
+import (
+	"context"
+	"errors"
+
+	"github.com/ethereum/go-ethereum/core/types"
+	"github.com/mailchain/mailchain/cmd/indexer/internal/processor"
+)
+
+type Block struct {
+	txProcessor processor.Transaction
+}
+
+func (b *Block) Run(ctx context.Context, protocol, network string, blk interface{}) error {
+	ethBlk, ok := blk.(*types.Block)
+	if !ok {
+		return errors.New("tx must be go-ethereum/core/types.Block")
+	}
+
+	txs := ethBlk.Transactions()
+	for i := range txs {
+		if err := b.txProcessor.Run(ctx, protocol, network, txs[i], txOptions{block: ethBlk}); err != nil {
+			return err
+		}
+	}
+
+	return nil
+}

cmd/indexer/internal/ethereum/client.go

@@ -0,0 +1,24 @@
+package ethereum
+
+import (
+	"context"
+	"math/big"
+
+	"github.com/ethereum/go-ethereum/ethclient"
+)
+
+func NewRPC(address string) (*Client, error) {
+	client, err := ethclient.Dial(address)
+	if err != nil {
+		return nil, err
+	}
+	return &Client{client: client}, nil
+}
+
+type Client struct {
+	client *ethclient.Client
+}
+
+func (c *Client) Get(ctx context.Context, blockNo uint64) (blk interface{}, err error) {
+	return c.client.BlockByNumber(ctx, big.NewInt(int64(blockNo)))
+}

cmd/indexer/internal/ethereum/transaction.go

@@ -0,0 +1,87 @@
+package ethereum
+
+import (
+	"context"
+	"errors"
+	"math/big"
+
+	"github.com/ethereum/go-ethereum/core/types"
+	"github.com/ethereum/go-ethereum/params"
+	"github.com/mailchain/mailchain/cmd/indexer/internal/datastore"
+	"github.com/mailchain/mailchain/cmd/indexer/internal/processor"
+	"github.com/mailchain/mailchain/crypto/secp256k1"
+	"github.com/mailchain/mailchain/internal/protocols/ethereum"
+)
+
+type Transaction struct {
+	txStore    datastore.TransactionStore
+	rawTxStore datastore.RawTransactionStore
+	pkStore    datastore.PublicKeyStore
+
+	networkID *big.Int
+}
+
+type txOptions struct {
+	block *types.Block
+}
+
+func (t *Transaction) Run(ctx context.Context, protocol, network string, tx interface{}, txOpts processor.TransactionOptions) error {
+	// blk *types.Block, ethTx *types.Transaction
+	ethTx, ok := tx.(*types.Transaction)
+	if !ok {
+		return errors.New("tx must be go-ethereum/core/types.Transaction")
+	}
+
+	opts, ok := txOpts.(*txOptions)
+	if !ok {
+		return errors.New("tx must be ethereum.txOptions")
+	}
+
+	storeTx, err := t.toTransaction(opts.block, ethTx)
+	if err != nil {
+		return err
+	}
+
+	v, r, s := ethTx.RawSignatureValues()
+	pubKeyBytes, err := ethereum.GetPublicKeyFromTransaction(r, s, v,
+		ethTx.To().Bytes(),
+		ethTx.Data(),
+		ethTx.Nonce(),
+		ethTx.GasPrice(),
+		ethTx.Gas(),
+		ethTx.Value())
+	if err != nil {
+		return err
+	}
+	pubKey, err := secp256k1.PublicKeyFromBytes(pubKeyBytes)
+	if err != nil {
+		return err
+	}
+
+	if err := t.pkStore.PutPublicKey(ctx, protocol, network, storeTx.From, pubKey); err != nil {
+		return err
+	}
+
+	return processor.StoreTransaction(ctx, t.txStore, t.rawTxStore, protocol, network, storeTx, ethTx)
+}
+
+func (t *Transaction) toTransaction(blk *types.Block, tx *types.Transaction) (*datastore.Transaction, error) {
+	msg, err := tx.AsMessage(types.MakeSigner(&params.ChainConfig{ChainID: t.networkID}, blk.Number()))
+	if err != nil {
+		return nil, err
+	}
+	gasPrice := tx.GasPrice()
+	value := tx.Value()
+	gasUsed := big.NewInt(int64(tx.Gas()))
+
+	return &datastore.Transaction{
+		From:      msg.From().Bytes(),
+		BlockHash: blk.Hash().Bytes(),
+		Hash:      tx.Hash().Bytes(),
+		Data:      tx.Data(),
+		To:        tx.To().Bytes(),
+		Value:     *value,
+		GasUsed:   *gasUsed,
+		GasPrice:  *gasPrice,
+	}, nil
+}

cmd/indexer/internal/pq/mappings.go

@@ -0,0 +1,34 @@
+package pq
+
+import (
+	"github.com/mailchain/mailchain/internal/protocols"
+	"github.com/mailchain/mailchain/internal/protocols/ethereum"
+)
+
+var protocolUint8 = map[string]uint8{ //nolint:gochecknoglobals
+	protocols.Ethereum: 1,
+}
+
+var uint8Protocol = map[uint8]string{ //nolint:gochecknoglobals
+	1: protocols.Ethereum,
+}
+
+var protocolNetworkUint8 = map[string]map[string]uint8{ //nolint:gochecknoglobals
+	protocols.Ethereum: {
+		ethereum.Mainnet: 1,
+		ethereum.Goerli:  2,
+		ethereum.Kovan:   3,
+		ethereum.Rinkeby: 4,
+		ethereum.Ropsten: 5,
+	},
+}
+
+var uint8ProtocolNetwork = map[string]map[uint8]string{ //nolint:gochecknoglobals
+	protocols.Ethereum: {
+		1: ethereum.Mainnet,
+		2: ethereum.Goerli,
+		3: ethereum.Kovan,
+		4: ethereum.Rinkeby,
+		5: ethereum.Ropsten,
+	},
+}

cmd/indexer/internal/pq/sync.go

@@ -0,0 +1,93 @@
+package pq
+
+import (
+	"context"
+	"time"
+
+	"github.com/Masterminds/squirrel"
+	"github.com/jmoiron/sqlx"
+	"github.com/mailchain/mailchain/cmd/indexer/internal/datastore"
+	"github.com/pkg/errors"
+)
+
+// SyncStore database connection object
+type SyncStore struct {
+	db  *sqlx.DB
+	now func() time.Time
+}
+
+// NewSyncStore create new postgres database
+func NewSyncStore(db *sqlx.DB, now func() time.Time) (datastore.SyncStore, error) {
+	return &SyncStore{db: db}, nil
+}
+
+type sync struct {
+	Protocol uint8 `db:"protocol"`
+	Network  uint8 `db:"network"`
+
+	BlockNo uint64 `db:"block_no"`
+
+	CreatedAt time.Time `db:"created_at"`
+	UpdatedAt time.Time `db:"updated_at"`
+}
+
+func (s SyncStore) GetBlockNumber(ctx context.Context, protocol, network string) (blockNo uint64, err error) {
+	uProtocol, ok := protocolUint8[protocol]
+	if !ok {
+		return 0, errors.Errorf("unknown protocol: %q", protocol)
+	}
+	uNetwork, ok := protocolNetworkUint8[protocol][network]
+	if !ok {
+		return 0, errors.Errorf("unknown protocol.network: \"%s.%s\"", protocol, network)
+	}
+	sql, args, err := s.selectBlockNumberQuery(uProtocol, uNetwork)
+
+	//  // You can also get a single result, a la QueryRow
+	state := sync{}
+	err = s.db.Get(&state, sql, args)
+	if err != nil {
+		return 0, err
+	}
+
+	return state.BlockNo, nil
+}
+
+func (s SyncStore) PutBlockNumber(ctx context.Context, protocol, network string, blockNo uint64) error {
+	uProtocol, ok := protocolUint8[protocol]
+	if !ok {
+		return errors.Errorf("unknown protocol: %q", protocol)
+	}
+	uNetwork, ok := protocolNetworkUint8[protocol][network]
+	if !ok {
+		return errors.Errorf("unknown protocol.network: \"%s.%s\"", protocol, network)
+	}
+
+	sql, args, err := s.updateBlockNumberQuery(uProtocol, uNetwork, blockNo)
+	if err != nil {
+		return errors.WithStack(err)
+	}
+	_, err = s.db.Exec(sql, args...)
+	if err != nil {
+		return errors.WithStack(err)
+	}
+	return nil
+}
+
+func (s SyncStore) updateBlockNumberQuery(protocol, network uint8, blockNo uint64) (sql string, args []interface{}, err error) {
+	return squirrel.Update("sync").
+		Set("block_no", blockNo).
+		Set("updated_at", s.now()).
+		PlaceholderFormat(squirrel.Dollar).
+		Where(squirrel.Eq{"protocol": protocol}).
+		Where(squirrel.Eq{"network": network}).
+		ToSql()
+}
+
+func (s SyncStore) selectBlockNumberQuery(protocol, network uint8) (sql string, args []interface{}, err error) {
+	return squirrel.Select("block_no").
+		From("sync").
+		PlaceholderFormat(squirrel.Dollar).
+		Where(squirrel.Eq{"protocol": protocol}).
+		Where(squirrel.Eq{"network": network}).
+		ToSql()
+}

cmd/indexer/internal/processor/block.go

@@ -0,0 +1,7 @@
+package processor
+
+import "context"
+
+type Block interface {
+	Run(ctx context.Context, protocol, network string, blk interface{}) error
+}

cmd/indexer/internal/processor/process.go

@@ -0,0 +1 @@
+package processor

cmd/indexer/internal/processor/sequential.go

@@ -0,0 +1,41 @@
+package processor
+
+import (
+	"context"
+
+	"github.com/mailchain/mailchain/cmd/indexer/internal/clients"
+	"github.com/mailchain/mailchain/cmd/indexer/internal/datastore"
+)
+
+type Sequential struct {
+	syncStore datastore.SyncStore
+	protocol  string
+	network   string
+
+	blockProcessor Block
+	blockClient    clients.BlockByNumber
+}
+
+func (s *Sequential) NextBlock(ctx context.Context) error {
+	blkNo, err := s.syncStore.GetBlockNumber(ctx, s.protocol, s.network)
+	if err != nil {
+		return err
+	}
+	nextBlockNo := blkNo + 1
+
+	// big.NewInt(int64(nextBlockNo))
+	blk, err := s.blockClient.Get(ctx, nextBlockNo)
+	if err != nil {
+		return err
+	}
+
+	if err := s.blockProcessor.Run(ctx, s.protocol, s.network, blk); err != nil {
+		return err
+	}
+
+	if err := s.syncStore.PutBlockNumber(ctx, s.protocol, s.network, nextBlockNo); err != nil {
+		return err
+	}
+
+	return nil
+}