Adds blockstore, core, metrics, router (#7)

- Add `blockstore`
- Add `core`, adds `router` into this package
- Add `metrics`

blockstore/blockstore.go

@@ -0,0 +1,117 @@
+// Copyright 2020 ChainSafe Systems
+// SPDX-License-Identifier: LGPL-3.0-only
+
+package blockstore
+
+import (
+	"fmt"
+	"io/ioutil"
+	"math/big"
+	"os"
+	"path/filepath"
+
+	"github.com/ChainSafe/chainbridge-utils/msg"
+)
+
+const PathPostfix = ".chainbridge/blockstore"
+
+type Blockstorer interface {
+	StoreBlock(*big.Int) error
+}
+
+var _ Blockstorer = &EmptyStore{}
+var _ Blockstorer = &Blockstore{}
+
+// Dummy store for testing only
+type EmptyStore struct{}
+
+func (s *EmptyStore) StoreBlock(_ *big.Int) error { return nil }
+
+// Blockstore implements Blockstorer.
+type Blockstore struct {
+	path     string // Path excluding filename
+	fullPath string
+	chain    msg.ChainId
+	relayer  string
+}
+
+func NewBlockstore(path string, chain msg.ChainId, relayer string) (*Blockstore, error) {
+	fileName := getFileName(chain, relayer)
+	if path == "" {
+		def, err := getDefaultPath()
+		if err != nil {
+			return nil, err
+		}
+		path = def
+	}
+
+	return &Blockstore{
+		path:     path,
+		fullPath: filepath.Join(path, fileName),
+		chain:    chain,
+		relayer:  relayer,
+	}, nil
+}
+
+// StoreBlock writes the block number to disk.
+func (b *Blockstore) StoreBlock(block *big.Int) error {
+	// Create dir if it does not exist
+	if _, err := os.Stat(b.path); os.IsNotExist(err) {
+		errr := os.MkdirAll(b.path, os.ModePerm)
+		if errr != nil {
+			return errr
+		}
+	}
+
+	// Write bytes to file
+	data := []byte(block.String())
+	err := ioutil.WriteFile(b.fullPath, data, 0600)
+	if err != nil {
+		return err
+	}
+	return nil
+}
+
+// TryLoadLatestBlock will attempt to load the latest block for the chain/relayer pair, returning 0 if not found.
+// Passing an empty string for path will cause it to use the home directory.
+func (b *Blockstore) TryLoadLatestBlock() (*big.Int, error) {
+	// If it exists, load and return
+	exists, err := fileExists(b.fullPath)
+	if err != nil {
+		return nil, err
+	}
+	if exists {
+		dat, err := ioutil.ReadFile(b.fullPath)
+		if err != nil {
+			return nil, err
+		}
+		block, _ := big.NewInt(0).SetString(string(dat), 10)
+		return block, nil
+	}
+	// Otherwise just return 0
+	return big.NewInt(0), nil
+}
+
+func getFileName(chain msg.ChainId, relayer string) string {
+	return fmt.Sprintf("%s-%d.block", relayer, chain)
+}
+
+// getHomePath returns the home directory joined with PathPostfix
+func getDefaultPath() (string, error) {
+	home, err := os.UserHomeDir()
+	if err != nil {
+		return "", err
+	}
+
+	return filepath.Join(home, PathPostfix), nil
+}
+
+func fileExists(fileName string) (bool, error) {
+	_, err := os.Stat(fileName)
+	if os.IsNotExist(err) {
+		return false, nil
+	} else if err != nil {
+		return false, err
+	}
+	return true, nil
+}

blockstore/blockstore_test.go

@@ -0,0 +1,73 @@
+// Copyright 2020 ChainSafe Systems
+// SPDX-License-Identifier: LGPL-3.0-only
+
+package blockstore
+
+import (
+	"io/ioutil"
+	"math/big"
+	"os"
+	"testing"
+
+	"github.com/ChainSafe/chainbridge-utils/keystore"
+	"github.com/ChainSafe/chainbridge-utils/msg"
+)
+
+func TestSaveAndLoad(t *testing.T) {
+	dir, err := ioutil.TempDir(os.TempDir(), "blockstore")
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer os.RemoveAll(dir)
+
+	chain := msg.ChainId(10)
+	relayer := keystore.AliceSr25519.Address()
+
+	bs, err := NewBlockstore(dir, chain, relayer)
+	if err != nil {
+		t.Fatal(err)
+	}
+	// Load non-existent dir/file
+	block, err := bs.TryLoadLatestBlock()
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	if block.Uint64() != uint64(0) {
+		t.Fatalf("Expected: %d got: %d", 0, block.Uint64())
+	}
+
+	// Save block number
+	block = big.NewInt(999)
+	err = bs.StoreBlock(block)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	// Load block number
+	latest, err := bs.TryLoadLatestBlock()
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	if block.Uint64() != latest.Uint64() {
+		t.Fatalf("Expected: %d got: %d", block.Uint64(), latest.Uint64())
+	}
+
+	// Save block number again
+	block = big.NewInt(1234)
+	err = bs.StoreBlock(block)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	// Load block number
+	latest, err = bs.TryLoadLatestBlock()
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	if block.Uint64() != latest.Uint64() {
+		t.Fatalf("Expected: %d got: %d", block.Uint64(), latest.Uint64())
+	}
+}

core/chain.go

@@ -0,0 +1,31 @@
+// Copyright 2020 ChainSafe Systems
+// SPDX-License-Identifier: LGPL-3.0-only
+
+package core
+
+import (
+	metrics "github.com/ChainSafe/chainbridge-utils/metrics/types"
+	"github.com/ChainSafe/chainbridge-utils/msg"
+)
+
+type Chain interface {
+	Start() error // Start chain
+	SetRouter(*Router)
+	Id() msg.ChainId
+	Name() string
+	LatestBlock() metrics.LatestBlock
+	Stop()
+}
+
+type ChainConfig struct {
+	Name           string            // Human-readable chain name
+	Id             msg.ChainId       // ChainID
+	Endpoint       string            // url for rpc endpoint
+	From           string            // address of key to use
+	KeystorePath   string            // Location of key files
+	Insecure       bool              // Indicated whether the test keyring should be used
+	BlockstorePath string            // Location of blockstore
+	FreshStart     bool              // If true, blockstore is ignored at start.
+	LatestBlock    bool              // If true, overrides blockstore or latest block in config and starts from current block
+	Opts           map[string]string // Per chain options
+}

core/core.go

@@ -0,0 +1,72 @@
+// Copyright 2020 ChainSafe Systems
+// SPDX-License-Identifier: LGPL-3.0-only
+
+package core
+
+import (
+	"fmt"
+	"os"
+	"os/signal"
+	"syscall"
+
+	"github.com/ChainSafe/log15"
+)
+
+type Core struct {
+	Registry []Chain
+	route    *Router
+	log      log15.Logger
+	sysErr   <-chan error
+}
+
+func NewCore(sysErr <-chan error) *Core {
+	return &Core{
+		Registry: make([]Chain, 0),
+		route:    NewRouter(log15.New("system", "router")),
+		log:      log15.New("system", "core"),
+		sysErr:   sysErr,
+	}
+}
+
+// AddChain registers the chain in the Registry and calls Chain.SetRouter()
+func (c *Core) AddChain(chain Chain) {
+	c.Registry = append(c.Registry, chain)
+	chain.SetRouter(c.route)
+}
+
+// Start will call all registered chains' Start methods and block forever (or until signal is received)
+func (c *Core) Start() {
+	for _, chain := range c.Registry {
+		err := chain.Start()
+		if err != nil {
+			c.log.Error(
+				"failed to start chain",
+				"chain", chain.Id(),
+				"err", err,
+			)
+			return
+		}
+		c.log.Info(fmt.Sprintf("Started %s chain", chain.Name()))
+	}
+
+	sigc := make(chan os.Signal, 1)
+	signal.Notify(sigc, syscall.SIGINT, syscall.SIGTERM)
+	defer signal.Stop(sigc)
+
+	// Block here and wait for a signal
+	select {
+	case err := <-c.sysErr:
+		c.log.Error("FATAL ERROR. Shutting down.", "err", err)
+	case <-sigc:
+		c.log.Warn("Interrupt received, shutting down now.")
+	}
+
+	// Signal chains to shutdown
+	for _, chain := range c.Registry {
+		chain.Stop()
+	}
+}
+
+func (c *Core) Errors() <-chan error {
+	return c.sysErr
+}

core/router.go

@@ -0,0 +1,55 @@
+// Copyright 2020 ChainSafe Systems
+// SPDX-License-Identifier: LGPL-3.0-only
+
+package core
+
+import (
+	"fmt"
+	"sync"
+
+	"github.com/ChainSafe/chainbridge-utils/msg"
+	log "github.com/ChainSafe/log15"
+)
+
+// Writer consumes a message and makes the requried on-chain interactions.
+type Writer interface {
+	ResolveMessage(message msg.Message) bool
+}
+
+// Router forwards messages from their source to their destination
+type Router struct {
+	registry map[msg.ChainId]Writer
+	lock     *sync.RWMutex
+	log      log.Logger
+}
+
+func NewRouter(log log.Logger) *Router {
+	return &Router{
+		registry: make(map[msg.ChainId]Writer),
+		lock:     &sync.RWMutex{},
+		log:      log,
+	}
+}
+
+// Send passes a message to the destination Writer if it exists
+func (r *Router) Send(msg msg.Message) error {
+	r.lock.Lock()
+	defer r.lock.Unlock()
+
+	r.log.Trace("Routing message", "src", msg.Source, "dest", msg.Destination, "nonce", msg.DepositNonce, "rId", msg.ResourceId.Hex())
+	w := r.registry[msg.Destination]
+	if w == nil {
+		return fmt.Errorf("unknown destination chainId: %d", msg.Destination)
+	}
+
+	go w.ResolveMessage(msg)
+	return nil
+}
+
+// Listen registers a Writer with a ChainId which Router.Send can then use to propagate messages
+func (r *Router) Listen(id msg.ChainId, w Writer) {
+	r.lock.Lock()
+	defer r.lock.Unlock()
+	r.log.Debug("Registering new chain in router", "id", id)
+	r.registry[id] = w
+}