laconicd/ethereum/rpc/websockets.go

package rpc

import (
	"bytes"
	"encoding/json"
	"fmt"
	"io/ioutil"
	"math/big"
	"net/http"
	"sync"

	"github.com/ethereum/go-ethereum/eth/filters"

	rpcfilters "github.com/tharsis/ethermint/ethereum/rpc/namespaces/eth/filters"

	"github.com/gorilla/mux"
	"github.com/gorilla/websocket"
	"github.com/pkg/errors"
	log "github.com/xlab/suplog"

	coretypes "github.com/tendermint/tendermint/rpc/core/types"
	rpcclient "github.com/tendermint/tendermint/rpc/jsonrpc/client"
	tmtypes "github.com/tendermint/tendermint/types"

	"github.com/ethereum/go-ethereum/common"
	"github.com/ethereum/go-ethereum/rpc"

	"github.com/tharsis/ethermint/ethereum/rpc/types"
	evmtypes "github.com/tharsis/ethermint/x/evm/types"
)

type WebsocketsServer interface {
	Start()
}

type SubscriptionResponseJSON struct {
	Jsonrpc string      `json:"jsonrpc"`
	Result  interface{} `json:"result"`
	ID      float64     `json:"id"`
}

type SubscriptionNotification struct {
	Jsonrpc string              `json:"jsonrpc"`
	Method  string              `json:"method"`
	Params  *SubscriptionResult `json:"params"`
}

type SubscriptionResult struct {
	Subscription rpc.ID      `json:"subscription"`
	Result       interface{} `json:"result"`
}

type ErrorResponseJSON struct {
	Jsonrpc string            `json:"jsonrpc"`
	Error   *ErrorMessageJSON `json:"error"`
	ID      *big.Int          `json:"id"`
}

type ErrorMessageJSON struct {
	Code    *big.Int `json:"code"`
	Message string   `json:"message"`
}

type websocketsServer struct {
	rpcAddr string // listen address of rest-server
	wsAddr  string // listen address of ws server
	api     *pubSubAPI
	logger  log.Logger
}

func NewWebsocketsServer(tmWSClient *rpcclient.WSClient, rpcAddr, wsAddr string) WebsocketsServer {
	return &websocketsServer{
		rpcAddr: rpcAddr,
		wsAddr:  wsAddr,
		api:     newPubSubAPI(tmWSClient),
		logger:  log.WithField("module", "websocket-server"),
	}
}

func (s *websocketsServer) Start() {
	ws := mux.NewRouter()
	ws.Handle("/", s)

	go func() {
		err := http.ListenAndServe(s.wsAddr, ws)
		if err != nil {
			if err == http.ErrServerClosed {
				return
			}

			s.logger.WithError(err).Errorln("failed to start HTTP server for WS")
		}
	}()
}

func (s *websocketsServer) ServeHTTP(w http.ResponseWriter, r *http.Request) {
	var upgrader = websocket.Upgrader{
		CheckOrigin: func(r *http.Request) bool {
			return true
		},
	}

	conn, err := upgrader.Upgrade(w, r, nil)
	if err != nil {
		s.logger.WithError(err).Warningln("websocket upgrade failed")
		return
	}

	s.readLoop(&wsConn{
		mux:  new(sync.Mutex),
		conn: conn,
	})
}

func (s *websocketsServer) sendErrResponse(wsConn *wsConn, msg string) {
	res := &ErrorResponseJSON{
		Jsonrpc: "2.0",
		Error: &ErrorMessageJSON{
			Code:    big.NewInt(-32600),
			Message: msg,
		},
		ID: nil,
	}

	_ = wsConn.WriteJSON(res)
}

type wsConn struct {
	conn *websocket.Conn
	mux  *sync.Mutex
}

func (w *wsConn) WriteJSON(v interface{}) error {
	w.mux.Lock()
	defer w.mux.Unlock()

	return w.conn.WriteJSON(v)
}

func (w *wsConn) Close() error {
	w.mux.Lock()
	defer w.mux.Unlock()

	return w.conn.Close()
}

func (w *wsConn) ReadMessage() (messageType int, p []byte, err error) {
	// not protected by write mutex

	return w.conn.ReadMessage()
}

func (s *websocketsServer) readLoop(wsConn *wsConn) {
	for {
		_, mb, err := wsConn.ReadMessage()
		if err != nil {
			_ = wsConn.Close()
			return
		}

		var msg map[string]interface{}
		err = json.Unmarshal(mb, &msg)
		if err != nil {
			s.sendErrResponse(wsConn, "invalid request")
			continue
		}

		// check if method == eth_subscribe or eth_unsubscribe
		method, ok := msg["method"].(string)
		if !ok {
			// otherwise, call the usual rpc server to respond
			err = s.tcpGetAndSendResponse(wsConn, mb)
			if err != nil {
				s.sendErrResponse(wsConn, err.Error())
			}

			continue
		}

		connID := msg["id"].(float64)
		if method == "eth_subscribe" {
			params := msg["params"].([]interface{})
			if len(params) == 0 {
				s.sendErrResponse(wsConn, "invalid parameters")
				continue
			}

			id, err := s.api.subscribe(wsConn, params)
			if err != nil {
				s.sendErrResponse(wsConn, err.Error())
				continue
			}

			res := &SubscriptionResponseJSON{
				Jsonrpc: "2.0",
				ID:      connID,
				Result:  id,
			}

			err = wsConn.WriteJSON(res)
			if err != nil {
				continue
			}

			continue
		} else if method == "eth_unsubscribe" {
			ids, ok := msg["params"].([]interface{})
			if _, idok := ids[0].(string); !ok || !idok {
				s.sendErrResponse(wsConn, "invalid parameters")
				continue
			}

			ok = s.api.unsubscribe(rpc.ID(ids[0].(string)))
			res := &SubscriptionResponseJSON{
				Jsonrpc: "2.0",
				ID:      connID,
				Result:  ok,
			}

			err = wsConn.WriteJSON(res)
			if err != nil {
				continue
			}

			continue
		}

		// otherwise, call the usual rpc server to respond
		err = s.tcpGetAndSendResponse(wsConn, mb)
		if err != nil {
			s.sendErrResponse(wsConn, err.Error())
		}
	}
}

// tcpGetAndSendResponse connects to the rest-server over tcp, posts a JSON-RPC request, and sends the response
// to the client over websockets
func (s *websocketsServer) tcpGetAndSendResponse(wsConn *wsConn, mb []byte) error {
	req, err := http.NewRequest("POST", "http://"+s.rpcAddr, bytes.NewBuffer(mb))
	if err != nil {
		return errors.Wrap(err, "Could not build request")
	}

	req.Header.Set("Content-Type", "application/json")
	client := &http.Client{}
	resp, err := client.Do(req)
	if err != nil {
		return errors.Wrap(err, "Could not perform request")
	}

	defer resp.Body.Close()

	body, err := ioutil.ReadAll(resp.Body)
	if err != nil {
		return errors.Wrap(err, "could not read body from response")
	}

	var wsSend interface{}
	err = json.Unmarshal(body, &wsSend)
	if err != nil {
		return errors.Wrap(err, "failed to unmarshal rest-server response")
	}

	return wsConn.WriteJSON(wsSend)
}

type wsSubscription struct {
	sub          *rpcfilters.Subscription
	unsubscribed chan struct{} // closed when unsubscribing
	wsConn       *wsConn
	query        string
}

// pubSubAPI is the eth_ prefixed set of APIs in the Web3 JSON-RPC spec
type pubSubAPI struct {
	events    *rpcfilters.EventSystem
	filtersMu *sync.RWMutex
	filters   map[rpc.ID]*wsSubscription
	logger    log.Logger
}

// newPubSubAPI creates an instance of the ethereum PubSub API.
func newPubSubAPI(tmWSClient *rpcclient.WSClient) *pubSubAPI {
	return &pubSubAPI{
		events:    rpcfilters.NewEventSystem(tmWSClient),
		filtersMu: new(sync.RWMutex),
		filters:   make(map[rpc.ID]*wsSubscription),
		logger:    log.WithField("module", "websocket-client"),
	}
}

func (api *pubSubAPI) subscribe(wsConn *wsConn, params []interface{}) (rpc.ID, error) {
	method, ok := params[0].(string)
	if !ok {
		return "0", errors.New("invalid parameters")
	}

	switch method {
	case "newHeads":
		// TODO: handle extra params
		return api.subscribeNewHeads(wsConn)
	case "logs":
		if len(params) > 1 {
			return api.subscribeLogs(wsConn, params[1])
		}
		return api.subscribeLogs(wsConn, nil)
	case "newPendingTransactions":
		return api.subscribePendingTransactions(wsConn)
	case "syncing":
		return api.subscribeSyncing(wsConn)
	default:
		return "0", errors.Errorf("unsupported method %s", method)
	}
}

func (api *pubSubAPI) unsubscribe(id rpc.ID) bool {
	api.filtersMu.Lock()
	defer api.filtersMu.Unlock()

	wsSub, ok := api.filters[id]
	if !ok {
		return false
	}

	wsSub.sub.Unsubscribe(api.events)
	close(api.filters[id].unsubscribed)
	delete(api.filters, id)
	return true
}

func (api *pubSubAPI) subscribeNewHeads(wsConn *wsConn) (rpc.ID, error) {
	var query = "subscribeNewHeads"
	var subID = rpc.NewID()

	sub, _, err := api.events.SubscribeNewHeads()
	if err != nil {
		return "", errors.Wrap(err, "error creating block filter")
	}

	unsubscribed := make(chan struct{})
	api.filtersMu.Lock()
	api.filters[subID] = &wsSubscription{
		sub:          sub,
		wsConn:       wsConn,
		unsubscribed: unsubscribed,
		query:        query,
	}
	api.filtersMu.Unlock()

	go func(headersCh <-chan coretypes.ResultEvent, errCh <-chan error) {
		for {
			select {
			case event, ok := <-headersCh:
				if !ok {
					api.unsubscribe(subID)
					return
				}

				data, ok := event.Data.(tmtypes.EventDataNewBlockHeader)
				if !ok {
					log.WithFields(log.Fields{
						"expected": "tmtypes.EventDataNewBlockHeader",
						"actual":   fmt.Sprintf("%T", event.Data),
					}).Warningln("event Data type mismatch")
					continue
				}

				header := types.EthHeaderFromTendermint(data.Header)

				api.filtersMu.RLock()
				for subID, wsSub := range api.filters {
					subID := subID
					wsSub := wsSub
					if wsSub.query != query {
						continue
					}
					// write to ws conn
					res := &SubscriptionNotification{
						Jsonrpc: "2.0",
						Method:  "eth_subscription",
						Params: &SubscriptionResult{
							Subscription: subID,
							Result:       header,
						},
					}

					err = wsSub.wsConn.WriteJSON(res)
					if err != nil {
						api.logger.WithError(err).Error("error writing header, will drop peer")

						try(func() {
							api.filtersMu.RUnlock()
							api.filtersMu.Lock()
							defer func() {
								api.filtersMu.Unlock()
								api.filtersMu.RLock()
							}()

							if err != websocket.ErrCloseSent {
								_ = wsSub.wsConn.Close()
							}

							delete(api.filters, subID)
							close(wsSub.unsubscribed)
						}, api.logger, "closing websocket peer sub")
					}
				}
				api.filtersMu.RUnlock()
			case err, ok := <-errCh:
				if !ok {
					api.unsubscribe(subID)
					return
				}
				log.WithError(err).Debugln("dropping NewHeads WebSocket subscription", subID)
				api.unsubscribe(subID)
			case <-unsubscribed:
				return
			}
		}
	}(sub.Event(), sub.Err())

	return subID, nil
}

func try(fn func(), l log.Logger, desc string) {
	if l == nil {
		l = log.DefaultLogger
	}

	defer func() {
		if x := recover(); x != nil {
			if err, ok := x.(error); ok {
				// debug.PrintStack()
				l.WithError(err).Debugln("panic during", desc)
				return
			}

			// debug.PrintStack()
			l.Debugf("panic during %s: %+v", desc, x)
			return
		}
	}()

	fn()
}

func (api *pubSubAPI) subscribeLogs(wsConn *wsConn, extra interface{}) (rpc.ID, error) {
	crit := filters.FilterCriteria{}

	if extra != nil {
		params, ok := extra.(map[string]interface{})
		if !ok {
			err := errors.New("invalid criteria")
			log.WithFields(log.Fields{
				"criteria": fmt.Sprintf("expected: map[string]interface{}, got %T", extra),
			}).Errorln(err)
			return "", err
		}

		if params["address"] != nil {
			address, ok := params["address"].(string)
			addresses, sok := params["address"].([]interface{})
			if !ok && !sok {
				err := errors.New("invalid address; must be address or array of addresses")
				log.WithFields(log.Fields{
					"address":   ok,
					"addresses": sok,
					"type":      fmt.Sprintf("%T", params["address"]),
				}).Errorln(err)
				return "", err
			}

			if ok {
				crit.Addresses = []common.Address{common.HexToAddress(address)}
			}

			if sok {
				crit.Addresses = []common.Address{}
				for _, addr := range addresses {
					address, ok := addr.(string)
					if !ok {
						err := errors.New("invalid address")
						log.WithField("type", fmt.Sprintf("%T", addr)).Errorln(err)
						return "", err
					}

					crit.Addresses = append(crit.Addresses, common.HexToAddress(address))
				}
			}
		}

		if params["topics"] != nil {
			topics, ok := params["topics"].([]interface{})
			if !ok {
				err := errors.New("invalid topics")

				log.WithFields(log.Fields{
					"type": fmt.Sprintf("expected []interface{}, got %T", params["topics"]),
				}).Errorln(err)

				return "", err
			}

			crit.Topics = make([][]common.Hash, len(topics))

			addCritTopic := func(topicIdx int, topic interface{}) error {
				tstr, ok := topic.(string)
				if !ok {
					err := errors.Errorf("invalid topic: %s", topic)
					log.WithField("type", fmt.Sprintf("expected string, got %T", topic)).Errorln(err)
					return err
				}

				crit.Topics[topicIdx] = []common.Hash{common.HexToHash(tstr)}
				return nil
			}

			for topicIdx, subtopics := range topics {
				if subtopics == nil {
					continue
				}

				// in case we don't have list, but a single topic value
				if topic, ok := subtopics.(string); ok {
					if err := addCritTopic(topicIdx, topic); err != nil {
						return "", err
					}

					continue
				}

				// in case we actually have a list of subtopics
				subtopicsList, ok := subtopics.([]interface{})
				if !ok {
					err := errors.New("invalid subtopics")

					log.WithFields(log.Fields{
						"type": fmt.Sprintf("expected []interface{}, got %T", subtopics),
					}).Errorln(err)

					return "", err
				}

				subtopicsCollect := make([]common.Hash, len(subtopicsList))
				for idx, subtopic := range subtopicsList {
					tstr, ok := subtopic.(string)
					if !ok {
						err := errors.Errorf("invalid subtopic: %s", subtopic)
						log.WithField("type", fmt.Sprintf("expected string, got %T", subtopic)).Errorln(err)
						return "", err
					}

					subtopicsCollect[idx] = common.HexToHash(tstr)
				}

				crit.Topics[topicIdx] = subtopicsCollect
			}
		}
	}

	critBz, err := json.Marshal(crit)
	if err != nil {
		log.WithError(err).Errorln("failed to JSON marshal criteria")
		return rpc.ID(""), err
	}

	var query = "subscribeLogs" + string(critBz)
	var subID = rpc.NewID()

	sub, _, err := api.events.SubscribeLogs(crit)
	if err != nil {
		log.WithError(err).Errorln("failed to subscribe logs")
		return rpc.ID(""), err
	}

	unsubscribed := make(chan struct{})
	api.filtersMu.Lock()
	api.filters[subID] = &wsSubscription{
		sub:          sub,
		wsConn:       wsConn,
		unsubscribed: unsubscribed,
		query:        query,
	}
	api.filtersMu.Unlock()

	go func(ch <-chan coretypes.ResultEvent, errCh <-chan error, subID rpc.ID) {
		for {
			select {
			case event, ok := <-ch:
				if !ok {
					api.unsubscribe(subID)
					return
				}

				dataTx, ok := event.Data.(tmtypes.EventDataTx)
				if !ok {
					log.WithFields(log.Fields{
						"expected": "tmtypes.EventDataTx",
						"actual":   fmt.Sprintf("%T", event.Data),
					}).Warningln("event Data type mismatch")
					continue
				}

				txResponse, err := evmtypes.DecodeTxResponse(dataTx.TxResult.Result.Data)
				if err != nil {
					log.WithError(err).Errorln("failed to decode tx response")
					return
				}

				logs := rpcfilters.FilterLogs(evmtypes.LogsToEthereum(txResponse.Logs), crit.FromBlock, crit.ToBlock, crit.Addresses, crit.Topics)
				if len(logs) == 0 {
					continue
				}

				api.filtersMu.RLock()
				wsSub, ok := api.filters[subID]
				if !ok {
					log.Warningln("subID not in filters", subID)
					return
				}
				api.filtersMu.RUnlock()

				for _, ethLog := range logs {
					res := &SubscriptionNotification{
						Jsonrpc: "2.0",
						Method:  "eth_subscription",
						Params: &SubscriptionResult{
							Subscription: subID,
							Result:       ethLog,
						},
					}

					err = wsSub.wsConn.WriteJSON(res)
					if err != nil {
						try(func() {
							api.filtersMu.Lock()
							defer api.filtersMu.Unlock()

							if err != websocket.ErrCloseSent {
								_ = wsSub.wsConn.Close()
							}

							delete(api.filters, subID)
							close(wsSub.unsubscribed)
						}, api.logger, "closing websocket peer sub")
					}
				}
			case err, ok := <-errCh:
				if !ok {
					api.unsubscribe(subID)
					return
				}
				log.WithError(err).Warningln("dropping Logs WebSocket subscription", subID)
				api.unsubscribe(subID)
			case <-unsubscribed:
				return
			}
		}
	}(sub.Event(), sub.Err(), subID)

	return subID, nil
}

func (api *pubSubAPI) subscribePendingTransactions(wsConn *wsConn) (rpc.ID, error) {
	var query = "subscribePendingTransactions"
	var subID = rpc.NewID()

	sub, _, err := api.events.SubscribePendingTxs()
	if err != nil {
		return "", errors.Wrap(err, "error creating block filter: %s")
	}

	unsubscribed := make(chan struct{})
	api.filtersMu.Lock()
	api.filters[subID] = &wsSubscription{
		sub:          sub,
		wsConn:       wsConn,
		unsubscribed: unsubscribed,
		query:        query,
	}
	api.filtersMu.Unlock()

	go func(txsCh <-chan coretypes.ResultEvent, errCh <-chan error) {
		for {
			select {
			case ev := <-txsCh:
				data, _ := ev.Data.(tmtypes.EventDataTx)
				txHash := common.BytesToHash(tmtypes.Tx(data.Tx).Hash())

				api.filtersMu.RLock()
				for subID, wsSub := range api.filters {
					subID := subID
					wsSub := wsSub
					if wsSub.query != query {
						continue
					}
					// write to ws conn
					res := &SubscriptionNotification{
						Jsonrpc: "2.0",
						Method:  "eth_subscription",
						Params: &SubscriptionResult{
							Subscription: subID,
							Result:       txHash,
						},
					}

					err = wsSub.wsConn.WriteJSON(res)
					if err != nil {
						api.logger.WithError(err).Warningln("error writing header, will drop peer")

						try(func() {
							api.filtersMu.Lock()
							defer api.filtersMu.Unlock()

							if err != websocket.ErrCloseSent {
								_ = wsSub.wsConn.Close()
							}

							delete(api.filters, subID)
							close(wsSub.unsubscribed)
						}, api.logger, "closing websocket peer sub")
					}
				}
				api.filtersMu.RUnlock()
			case err, ok := <-errCh:
				if !ok {
					api.unsubscribe(subID)
					return
				}
				log.WithError(err).Warningln("dropping PendingTransactions WebSocket subscription", subID)
				api.unsubscribe(subID)
			case <-unsubscribed:
				return
			}
		}
	}(sub.Event(), sub.Err())

	return subID, nil
}

func (api *pubSubAPI) subscribeSyncing(wsConn *wsConn) (rpc.ID, error) {
	return "", nil
}