socket.io.go

// Copyright (c) 2018, The Bitum developers
// See LICENSE for details.

package insight

import (
	"encoding/json"
	"regexp"
	"sync"

	"github.com/bitum-project/bitumd/chaincfg"
	"github.com/bitum-project/bitumd/bitumjson"
	"github.com/bitum-project/bitumd/bitumutil"
	"github.com/bitum-project/bitumd/txscript"
	"github.com/bitum-project/bitumd/wire"
	"github.com/bitum-project/bitumdata/blockdata"
	"github.com/bitum-project/bitumdata/txhelpers"
	socketio "github.com/googollee/go-socket.io"
)

var isAlphaNumeric = regexp.MustCompile(`^[a-zA-Z0-9]+$`).MatchString

type roomSubscriptionCounter struct {
	*sync.RWMutex
	c map[string]int
}

// SocketServer wraps the socket.io server with the watched address list.
type SocketServer struct {
	socketio.Server
	params           *chaincfg.Params
	watchedAddresses roomSubscriptionCounter
	txGetter         txhelpers.RawTransactionGetter
}

// InsightSocketVin represents a single vin for the Insight "vin" JSON object
// that appears in a "tx" message from the "inv" room.
type InsightSocketVin struct {
	TxID      string   `json:"txid,omitempty"`
	Vout      *uint32  `json:"vout,omitempty"`
	Addresses []string `json:"addresses,omitempty"`
	Value     *int64   `json:"value,omitempty"`
}

func newInt64Ptr(i int64) *int64 {
	ii := i
	return &ii
}

func newUint32Ptr(i uint32) *uint32 {
	ii := i
	return &ii
}

// InsightSocketVout represents a single vout for the Insight "vout" JSON object
// that appears in a "tx" message from the "inv" room.
type InsightSocketVout struct {
	Address string
	Value   int64
}

// MarshalJSON implements json.Marshaler so that an InsightSocketVout will
// marshal to JSON like:
//	{
//	  "DsZQaCQES5vh3JmcyyFokJYz3aSw8Sm1dsQ": 13741789
//	}
func (v *InsightSocketVout) MarshalJSON() ([]byte, error) {
	vout := map[string]int64{
		v.Address: v.Value,
	}
	return json.Marshal(vout)
}

// WebSocketTx models the JSON data sent as the tx event in the inv room.
type WebSocketTx struct {
	Hash     string              `json:"txid"`
	Size     int                 `json:"size"`
	TotalOut int64               `json:"valueOut"`
	Vins     []InsightSocketVin  `json:"vins,omitempty"`
	Vouts    []InsightSocketVout `json:"vout,omitempty"`
}

// NewSocketServer constructs a new SocketServer, registering handlers for the
// "connection", "disconnection", and "subscribe" events.
func NewSocketServer(params *chaincfg.Params, txGetter txhelpers.RawTransactionGetter) (*SocketServer, error) {
	server, err := socketio.NewServer(nil)
	if err != nil {
		apiLog.Errorf("Could not create socket.io server: %v", err)
		return nil, err
	}

	// Each address subscription uses its own room, which has the same name as
	// the address. The number of subscribers for each room is tracked.
	addrs := roomSubscriptionCounter{
		RWMutex: new(sync.RWMutex),
		c:       make(map[string]int),
	}

	server.On("connection", func(so socketio.Socket) {
		apiLog.Debug("New socket.io connection")
		// New connections automatically join the inv and sync rooms.
		so.Join("inv")
		so.Join("sync")

		// Disconnection decrements or deletes the subscriber counter for each
		// address room to which the client was subscribed.
		so.On("disconnection", func() {
			apiLog.Debug("socket.io client disconnected")
			addrs.Lock()
			for _, str := range so.Rooms() {
				if c, ok := addrs.c[str]; ok {
					if c == 1 {
						delete(addrs.c, str)
					} else {
						addrs.c[str]--
					}
				}
			}
			addrs.Unlock()
		})

		// Subscription to a room checks the room name is as expected for an
		// address (TODO: do this better), joins the room, and increments the
		// room's subscriber count.
		so.On("subscribe", func(room string) {
			if len(room) > 64 || !isAlphaNumeric(room) {
				return
			}
			if addr, err := bitumutil.DecodeAddress(room); err == nil {
				if addr.IsForNet(params) {
					so.Join(room)
					apiLog.Debugf("socket.io client joining room: %s", room)

					addrs.Lock()
					addrs.c[room]++
					addrs.Unlock()
				}
			}
		})
	})

	server.On("error", func(_ socketio.Socket, err error) {
		apiLog.Errorf("Insight socket.io server error: %v", err)
	})

	sockServ := SocketServer{
		Server:           *server,
		params:           params,
		watchedAddresses: addrs,
		txGetter:         txGetter,
	}
	return &sockServ, nil
}

// Store broadcasts the lastest block hash to the the inv room. The coinbase
// transaction is also relayed to the new Tx channel where it is included in tx
// and address broadcasts.
func (soc *SocketServer) Store(blockData *blockdata.BlockData, msgBlock *wire.MsgBlock) error {
	apiLog.Debugf("Sending new websocket block %s", blockData.Header.Hash)
	soc.BroadcastTo("inv", "block", blockData.Header.Hash)

	// Since the coinbase transaction is generated by the miner, it will never
	// hit mempool. It must be processed now, with the new block.
	return soc.sendNewMsgTx(msgBlock.Transactions[0])
}

// SendNewTx prepares a bitumd mempool tx for broadcast. This method satisfies
// notification.TxHandler and is registered as a handler in main.go.
func (soc *SocketServer) SendNewTx(rawTx *bitumjson.TxRawResult) error {
	msgTx, err := txhelpers.MsgTxFromHex(rawTx.Hex)
	if err != nil {
		return err
	}
	return soc.sendNewTx(msgTx, rawTx.Vout)
}

// sendNewMsgTx processes and broadcasts a msgTx to subscribers.
func (soc *SocketServer) sendNewMsgTx(msgTx *wire.MsgTx) error {
	return soc.sendNewTx(msgTx, nil)
}

// sendNewTx processes and broadcasts a msgTx to subscribers, using an existing
// []Vout, if it is available. If vouts is zero-length, the output addresses are
// decoded from their pkScripts.
func (soc *SocketServer) sendNewTx(msgTx *wire.MsgTx, vouts []bitumjson.Vout) error {
	var vins []InsightSocketVin
	for _, v := range msgTx.TxIn {
		txid := v.PreviousOutPoint.Hash.String()
		idx := v.PreviousOutPoint.Index
		tree := v.PreviousOutPoint.Tree
		var addrs []string
		var amt bitumutil.Amount
		if txhelpers.IsZeroHashStr(txid) {
			// Coinbase and stake base inputs need to be "{}".
			vins = append(vins, InsightSocketVin{})
			continue
		} else {
			var err error
			addrs, amt, err = txhelpers.OutPointAddressesFromString(
				txid, idx, tree, soc.txGetter, soc.params)
			if err != nil {
				apiLog.Warnf("failed to get outpoint address from txid: %v", err)
				// Still must append this vin to maintain valid implicit
				// indexing of vins array.
			}
		}
		vins = append(vins, InsightSocketVin{
			TxID:      txid,
			Vout:      newUint32Ptr(idx),
			Addresses: addrs,
			Value:     newInt64Ptr(int64(amt)),
		})
	}

	var voutAddrs [][]string
	for i, v := range msgTx.TxOut {
		// Allow Vouts to be nil or empty, extracting the addresses from the
		// pkScripts here.
		if len(vouts) == 0 {
			_, scriptAddrs, _, err := txscript.ExtractPkScriptAddrs(
				v.Version, v.PkScript, soc.params)
			if err != nil {
				apiLog.Warnf("failed to decode pkScript: %v", err)
				// still need to append an empty array to voutAddrs.
			}
			var addrs []string
			for i := range scriptAddrs {
				addrs = append(addrs, scriptAddrs[i].EncodeAddress())
			}
			voutAddrs = append(voutAddrs, addrs)
		} else {
			voutAddrs = append(voutAddrs, vouts[i].ScriptPubKey.Addresses)
		}
	}

	hash := msgTx.TxHash().String()
	var voutsInsight []InsightSocketVout
	var total int64
	for i, v := range msgTx.TxOut {
		total += v.Value
		if len(voutAddrs[i]) == 0 {
			continue
		}

		soc.watchedAddresses.RLock()
		for _, address := range voutAddrs[i] {
			if _, ok := soc.watchedAddresses.c[address]; ok {
				soc.BroadcastTo(address, address, hash)
			}
			voutsInsight = append(voutsInsight, InsightSocketVout{
				Address: address,
				Value:   v.Value,
			})
		}
		soc.watchedAddresses.RUnlock()
	}

	tx := WebSocketTx{
		Hash:     hash,
		Size:     msgTx.SerializeSize(),
		TotalOut: total,
		Vins:     vins,
		Vouts:    voutsInsight,
	}
	apiLog.Tracef("Sending new websocket tx %s", hash)
	soc.BroadcastTo("inv", "tx", tx)
	return nil
}