thalos/internal/server/ship_processor.go

package server

import (
	"bytes"
	"encoding/hex"

	"github.com/eosswedenorg/thalos/api"
	"github.com/eosswedenorg/thalos/api/message"
	"github.com/eosswedenorg/thalos/internal/abi"
	"github.com/eosswedenorg/thalos/internal/driver"

	log "github.com/sirupsen/logrus"

	shipclient "github.com/eosswedenorg-go/antelope-ship-client"
	"github.com/pnx/antelope-go/chain"
	"github.com/pnx/antelope-go/ship"
)

// logDecoratedEncoder decorates a message.Encoder and logs any error.
func logDecoratedEncoder(encoder message.Encoder) message.Encoder {
	return func(v interface{}) ([]byte, error) {
		payload, err := encoder(v)
		if err != nil {
			log.WithError(err).
				WithField("v", v).
				Warn("Failed to encode message")
		}
		return payload, err
	}
}

// A ShipProcessor will consume messages from a ship stream, convert the messages into
// thalos specific ones, encode them and finally post them to an api.Writer
type ShipProcessor struct {
	// The ship stream to process.
	shipStream *shipclient.Stream

	// Abi manager used for cacheing
	abi *abi.AbiManager

	// Writer to send messages to.
	writer driver.Writer

	// Encoder used to encode messages
	encode message.Encoder

	// Function for saving state.
	saver StateSaver

	// Internal state
	state State

	// System contract ("eosio" per default)
	syscontract chain.Name

	// ABI Returned from SHIP
	shipABI *chain.Abi
}

// SpawnProcessor creates a new ShipProccessor that consumes the shipclient.Stream passed to it.
func SpawnProccessor(shipStream *shipclient.Stream, loader StateLoader, saver StateSaver, writer driver.Writer, abi *abi.AbiManager, codec message.Codec) *ShipProcessor {
	processor := &ShipProcessor{
		saver:       saver,
		abi:         abi,
		writer:      writer,
		shipStream:  shipStream,
		encode:      logDecoratedEncoder(codec.Encoder),
		syscontract: chain.N("eosio"),
	}

	loader(&processor.state)

	// Attach handlers
	shipStream.BlockHandler = processor.processBlock
	shipStream.InitHandler = processor.initHandler

	// Needed because if nil, traces/table deltas will not be included in the response from ship.
	shipStream.TraceHandler = func(*ship.TransactionTraceArray) {}
	shipStream.TableDeltaHandler = func(*ship.TableDeltaArray) {}

	return processor
}

func (processor *ShipProcessor) initHandler(abi *chain.Abi) {
	processor.shipABI = abi
}

func (processor *ShipProcessor) queueMessage(channel api.Channel, payload []byte) bool {
	err := processor.writer.Write(channel, payload)
	if err != nil {
		log.WithError(err).Errorf("Failed to post to channel '%s'", channel)
		return false
	}
	return true
}

func (processor *ShipProcessor) encodeQueue(channel api.Channel, v interface{}) bool {
	if payload, err := processor.encode(v); err == nil {
		return processor.queueMessage(channel, payload)
	}
	return false
}

// updateAbiFromAction updates the contract abi based on the ship.Action passed.
func (processor *ShipProcessor) updateAbiFromAction(act *chain.Action) error {
	set_abi := struct {
		Abi     string
		Account chain.Name
	}{}

	if err := act.DecodeInto(&set_abi); err != nil {
		return err
	}

	binary_abi, err := hex.DecodeString(set_abi.Abi)
	if err != nil {
		return err
	}

	contract_abi := chain.Abi{}
	err = chain.NewDecoder(bytes.NewReader(binary_abi)).Decode(&contract_abi)
	if err != nil {
		return err
	}
	return processor.abi.SetAbi(set_abi.Account, &contract_abi)
}

// Get the current block.
func (processor *ShipProcessor) GetCurrentBlock() uint32 {
	return processor.state.CurrentBlock
}

func (processor *ShipProcessor) broadcastAction(act *message.ActionTrace) {
	payload, err := processor.encode(*act)
	if err != nil {
		log.WithField("act", act).Warn("failed to encode action")
		return
	}

	channels := []api.Channel{
		api.ActionChannel{}.Channel(),
		api.ActionChannel{Name: act.Name}.Channel(),
		api.ActionChannel{Contract: act.Contract}.Channel(),
		api.ActionChannel{Name: act.Name, Contract: act.Contract}.Channel(),
	}

	for _, channel := range channels {
		processor.queueMessage(channel, payload)
	}
}

func (processor *ShipProcessor) processTransactionTrace(log *log.Entry, blockNumber uint32, block *ship.SignedBlock, trace *ship.TransactionTraceV0) {
	logger := log.WithField("type", "trace").WithField("tx_id", trace.ID.String()).Dup()

	timestamp := block.BlockHeader.Timestamp.Time().UTC()

	transaction := message.TransactionTrace{
		ID:            trace.ID.String(),
		BlockNum:      blockNumber,
		Timestamp:     timestamp,
		Status:        trace.Status.String(),
		CPUUsageUS:    trace.CPUUsageUS,
		NetUsage:      trace.NetUsage,
		NetUsageWords: uint32(trace.NetUsageWords),
		Elapsed:       int64(trace.Elapsed),
		Scheduled:     trace.Scheduled,
		Except:        trace.Except,
		Error:         trace.ErrorCode,
	}

	// Actions
	for _, actionTraceVar := range trace.ActionTraces {

		actionTrace := toActionTraceV1(actionTraceVar)
		actMsg := processor.proccessActionTrace(logger, actionTrace)
		if actMsg != nil {
			actMsg.TxID = trace.ID.String()
			actMsg.BlockNum = blockNumber
			actMsg.Timestamp = timestamp

			processor.broadcastAction(actMsg)

			transaction.ActionTraces = append(transaction.ActionTraces, *actMsg)
		}
	}

	processor.encodeQueue(api.TransactionChannel, transaction)
}

func (processor *ShipProcessor) proccessActionTrace(logger *log.Entry, trace *ship.ActionTraceV1) *message.ActionTrace {
	// Check if actions updates an abi.
	if trace.Act.Account == processor.syscontract && trace.Act.Name == chain.N("setabi") {
		err := processor.updateAbiFromAction(&trace.Act)
		if err != nil {
			logger.WithError(err).Warn("Failed to update abi")
		}
	}

	act := &message.ActionTrace{
		Name:          trace.Act.Name.String(),
		Contract:      trace.Act.Account.String(),
		Receiver:      trace.Receiver.String(),
		FirstReceiver: trace.Act.Account.String() == trace.Receiver.String(),
	}

	if trace.Receipt != nil {
		receipt := trace.Receipt.V0
		act.Receipt = &message.ActionReceipt{
			Receiver:       receipt.Receiver.String(),
			ActDigest:      receipt.ActDigest.String(),
			GlobalSequence: receipt.GlobalSequence,
			RecvSequence:   receipt.RecvSequence,
			CodeSequence:   uint32(receipt.CodeSequence),
			ABISequence:    uint32(receipt.ABISequence),
		}

		for _, auth := range receipt.AuthSequence {
			act.Receipt.AuthSequence = append(act.Receipt.AuthSequence, message.AccountAuthSequence{
				Account:  auth.Account.String(),
				Sequence: auth.Sequence,
			})
		}
	}

	for _, auth := range trace.Act.Authorization {
		act.Authorization = append(act.Authorization, message.PermissionLevel{
			Actor:      auth.Actor.String(),
			Permission: auth.Permission.String(),
		})
	}

	ABI, err := processor.abi.GetAbi(trace.Act.Account)
	if err == nil {
		if act.Data, err = trace.Act.Decode(ABI); err != nil {
			logger.WithFields(log.Fields{
				"contract": trace.Act.Account,
				"action":   trace.Act.Name,
			}).WithError(err).Warn("Failed to decode action")
		}
	} else {
		logger.WithField("contract", trace.Act.Account).
			WithError(err).Error("Failed to get abi for contract")
	}

	return act
}

// Callback function called by shipclient.Stream when a new block arrives.
func (processor *ShipProcessor) processBlock(blockResult *ship.GetBlocksResultV0) {
	block := ship.SignedBlock{}
	blockResult.Block.Unpack(&block)
	timestamp := block.BlockHeader.Timestamp.Time().UTC()
	blockNumber := blockResult.ThisBlock.BlockNum

	// Check to see if we have a microfork and post a message to
	// the rollback channel in that case.
	if processor.state.CurrentBlock > 0 && blockNumber < processor.state.CurrentBlock {
		log.WithField("old_block", processor.state.CurrentBlock).
			WithField("new_block", blockResult.ThisBlock.BlockNum).
			Warn("Fork detected, old_block is greater than new_block")

		processor.encodeQueue(api.RollbackChannel, message.RollbackMessage{
			OldBlockNum: processor.state.CurrentBlock,
			NewBlockNum: blockResult.ThisBlock.BlockNum,
		})
	}

	processor.state.CurrentBlock = blockNumber

	if blockResult.ThisBlock.BlockNum%100 == 0 {
		log.Infof("Current: %d, Head: %d", processor.state.CurrentBlock, blockResult.Head.BlockNum)
	}

	if blockResult.ThisBlock.BlockNum%10 == 0 {
		hb := message.HeartBeat{
			BlockNum:                 blockNumber,
			LastIrreversibleBlockNum: blockResult.LastIrreversible.BlockNum,
			HeadBlockNum:             blockResult.Head.BlockNum,
		}

		processor.encodeQueue(api.HeartbeatChannel, hb)
	}

	mainLogger := log.WithField("block", blockNumber).Dup()

	// Process traces
	if blockResult.Traces != nil {
		unpacked := []ship.TransactionTrace{}
		if err := blockResult.Traces.Unpack(&unpacked); err != nil {
			mainLogger.WithError(err).Error("Failed to unpack transaction traces")
		} else {
			for _, trace := range unpacked {
				processor.processTransactionTrace(mainLogger, blockNumber, &block, trace.V0)
			}
		}
	}

	// Process deltas
	deltas := []ship.TableDelta{}
	if err := blockResult.Deltas.Unpack(&deltas); err != nil {
		mainLogger.WithError(err).Error("Failed to unpack table deltas")
	} else {
		for _, delta := range deltas {

			logger := mainLogger.WithField("type", "table_delta").Dup()

			rows := []message.TableDeltaRow{}
			for _, row := range delta.V0.Rows {

				msg := message.TableDeltaRow{
					Present: row.Present,
					RawData: row.Data,
				}

				if processor.shipABI != nil {

					v, err := processor.shipABI.Decode(bytes.NewReader(row.Data), delta.V0.Name)
					if err == nil {
						var ok bool
						if msg.Data, ok = v.(map[string]any); !ok {
							// logger.Error("Failed to cast table data")
						}
					} else {
						logger.Error("Failed to decode table delta")
					}
				} else {
					logger.Warn("No SHIP ABI present")
				}
				rows = append(rows, msg)
			}

			message := message.TableDelta{
				BlockNum:  blockNumber,
				Timestamp: timestamp,
				Name:      delta.V0.Name,
				Rows:      rows,
			}

			channels := []api.Channel{
				api.TableDeltaChannel{}.Channel(),
				api.TableDeltaChannel{Name: delta.V0.Name}.Channel(),
			}

			for _, channel := range channels {
				processor.encodeQueue(channel, message)
			}
		}
	}

	err := processor.writer.Flush()
	if err != nil {
		log.WithError(err).Error("Failed to send messages")
	}

	err = processor.saver(processor.state)
	if err != nil {
		log.WithError(err).Error("Failed to save state")
	}
}

// Close closes the writer associated with the processor.
func (processor *ShipProcessor) Close() error {
	return processor.writer.Close()
}