consolereader.go

package firecore

import (
	"errors"
	"fmt"
	"io"
	"regexp"
	"strconv"
	"strings"
	"sync"
	"time"

	"github.com/emmansun/base64" // benchmarked 2.4x faster than standard encoding/base64
	"github.com/streamingfast/bstream"
	pbbstream "github.com/streamingfast/bstream/pb/sf/bstream/v1"
	"github.com/streamingfast/dmetrics"
	"github.com/streamingfast/firehose-core/node-manager/consoleline"
	"github.com/streamingfast/firehose-core/node-manager/mindreader"
	"github.com/streamingfast/logging"
	"go.uber.org/zap"
	"google.golang.org/protobuf/types/known/anypb"
	"google.golang.org/protobuf/types/known/timestamppb"
)

const FirePrefix = "FIRE "
const FirePrefixLen = len(FirePrefix)
const InitLogPrefix = "INIT "
const InitLogPrefixLen = len(InitLogPrefix)
const BlockLogPrefix = "BLOCK "
const BlockLogPrefixLen = len(BlockLogPrefix)

// ethereumBlockType is the protobuf fully qualified block type used by every EVM Firehose
// tracer (geth and its forks: Injective, BSC, Polygon/bor, etc). Those tracers print
// "FIRE INIT <version> <node_variant> <node_version>" instead of a protobuf type name,
// because firehose-ethereum's own console reader hardcodes the block type on its side and
// only uses the variant/version tail for logging and (for the deprecated event-stream
// protocol only, which firecore does not support) normalization.
const ethereumBlockType = "sf.ethereum.type.v2.Block"

var protoFullyQualifiedNameRegexp = regexp.MustCompile(`^[A-Za-z_]\w*(\.[A-Za-z_]\w*)+$`)

// resolveBlockType interprets the token(s) following the reader protocol version in a
// "FIRE INIT" line. The first token is either a protobuf fully qualified type name
// (e.g. "sf.cosmos.type.v2.Block", optionally "type.googleapis.com/"-prefixed) or a bare
// node-variant word (e.g. "geth", "polygon"). Three shapes follow from that:
//   - a lone FQN (e.g. "sf.cosmos.type.v2.Block"): used as-is, nodeVariant/nodeVersion empty.
//   - "<fqn> <node_name> <node_version...>" (e.g. "sf.ethereum.type.v2.Block geth 1.20.4"):
//     the FQN is trusted as given; node_name/node_version are informational only.
//   - "<node_variant> <node_version...>" (e.g. "geth 1.20.4", "polygon 1.10.17-fh+hotfix
//     (deadbeef)"): every EVM Firehose tracer (geth and forks: Injective, BSC, Polygon/bor,
//     etc) hardcodes its block type to sf.ethereum.type.v2.Block on its own side and only
//     prints this tail for logging, so firecore does the same mapping here.
//
// Anything else (an empty or malformed first token, a trailing token/space with no
// node_version to go with it) fails loudly instead of producing a wrong block type.
func resolveBlockType(typeOrVariant string) (blockType, nodeVariant, nodeVersion string, err error) {
	invalidTypeErr := fmt.Errorf("invalid type %q: expected a protobuf fully qualified message type (e.g. %q), that type followed by a node name and version (e.g. %q), or a node variant and version (e.g. %q)", typeOrVariant, "sf.ethereum.type.v2.Block", "sf.ethereum.type.v2.Block geth 1.20.4", "geth 1.20.4")

	firstToken, rest, hasRest := strings.Cut(typeOrVariant, " ")

	// A type sent as `type.googleapis.com/<fqn>` is valid as-is, 'setProtoMessageType' handles it natively.
	bareFirstToken := strings.TrimPrefix(firstToken, "type.googleapis.com/")
	firstTokenIsFQN := protoFullyQualifiedNameRegexp.MatchString(bareFirstToken)

	if !hasRest {
		if !firstTokenIsFQN {
			return "", "", "", invalidTypeErr
		}

		return firstToken, "", "", nil
	}

	if firstTokenIsFQN {
		nodeVariant, nodeVersion, hasVersion := strings.Cut(rest, " ")
		nodeVersion = strings.TrimSpace(nodeVersion)
		if nodeVariant == "" || !hasVersion || nodeVersion == "" {
			return "", "", "", invalidTypeErr
		}

		return firstToken, nodeVariant, nodeVersion, nil
	}

	nodeVersion = strings.TrimSpace(rest)
	if firstToken == "" || nodeVersion == "" {
		return "", "", "", invalidTypeErr
	}

	return ethereumBlockType, firstToken, nodeVersion, nil
}

type ParsingStats struct {
}

type ConsoleReader struct {
	lines        chan string
	consoleLines <-chan consoleline.Line
	done         chan interface{}
	closeOnce    sync.Once
	logger       *zap.Logger
	tracer       logging.Tracer

	// Parsing context
	readerProtocolVersion string
	protoMessageType      string
	lastBlock             bstream.BlockRef
	lastParentBlock       bstream.BlockRef
	lastBlockTimestamp    time.Time
	readPartialBlockIndex bool

	lib uint64

	blockRate *dmetrics.AvgRatePromCounter
}

func NewConsoleReader(lines chan string, blockEncoder BlockEncoder, logger *zap.Logger, tracer logging.Tracer) (mindreader.ConsolerReader, error) {
	reader := newConsoleReader(lines, logger, tracer)

	delayBetweenStats := 30 * time.Second
	if tracer.Enabled() {
		delayBetweenStats = 5 * time.Second
	}

	go func() {
		defer reader.blockRate.Stop()

		for {
			select {
			case <-reader.done:
				return
			case <-time.After(delayBetweenStats):
				reader.printStats()
			}
		}
	}()

	return reader, nil
}

func newConsoleReader(lines chan string, logger *zap.Logger, tracer logging.Tracer) *ConsoleReader {
	return &ConsoleReader{
		lines:  lines,
		done:   make(chan interface{}),
		logger: logger,
		tracer: tracer,

		blockRate: dmetrics.MustNewAvgRateFromPromCounter(ConsoleReaderBlockReadCount, 1*time.Second, 30*time.Second, "blocks"),
	}
}

func (r *ConsoleReader) Done() <-chan interface{} {
	return r.done
}

func (r *ConsoleReader) Close() error {
	r.closeOnce.Do(func() {
		r.blockRate.SyncNow()
		r.printStats()

		r.logger.Info("console reader done")
		close(r.done)
	})

	return nil
}

type blockRefView struct {
	ref bstream.BlockRef
}

func (v blockRefView) String() string {
	if v.ref == nil {
		return "<unset>"
	}

	return v.ref.String()
}

type blockRefViewTimestamp struct {
	ref       bstream.BlockRef
	timestamp time.Time
}

func (v blockRefViewTimestamp) String() string {
	return fmt.Sprintf("%s @ %s", blockRefView{v.ref}, v.timestamp.Local().Format(time.RFC822Z))
}

func (r *ConsoleReader) printStats() {
	r.logger.Info("console reader stats",
		zap.Stringer("block_rate", r.blockRate),
		zap.Stringer("last_block", blockRefViewTimestamp{r.lastBlock, r.lastBlockTimestamp}),
		zap.Stringer("last_parent_block", blockRefView{r.lastParentBlock}),
		zap.Uint64("lib", r.lib),
	)
}

func (r *ConsoleReader) ReadBlock() (out *pbbstream.Block, err error) {
	out, err = r.next()
	if err != nil {
		return nil, err
	}

	return out, nil
}

// ReadLines makes the reader read its lines from the given channel instead of the one
// it was created with, "FIRE BLOCK" lines on it can have their payload already decoded.
// It implements [mindreader.BlockLineConsoleReader].
func (r *ConsoleReader) ReadLines(lines <-chan consoleline.Line) {
	r.consoleLines = lines
}

func (r *ConsoleReader) next() (out *pbbstream.Block, err error) {
	if r.consoleLines != nil {
		for line := range r.consoleLines {
			if line.Block != nil {
				out, err = r.readDecodedBlock(line.Block)
			} else {
				out, err = r.readLine(line.Text)
			}

			if err != nil || out != nil {
				return out, err
			}
		}
	} else {
		for line := range r.lines {
			out, err = r.readLine(line)
			if err != nil || out != nil {
				return out, err
			}
		}
	}

	r.Close()

	return nil, io.EOF
}

// readLine returns the block of a "FIRE BLOCK" line, or nil for any other line.
func (r *ConsoleReader) readLine(line string) (out *pbbstream.Block, err error) {
	if !strings.HasPrefix(line, "FIRE ") {
		return nil, nil
	}

	line = line[FirePrefixLen:]

	switch {
	case strings.HasPrefix(line, BlockLogPrefix):
		out, err = r.readBlock(line[BlockLogPrefixLen:])

	case strings.HasPrefix(line, InitLogPrefix):
		err = r.readInit(line[InitLogPrefixLen:])
	default:
		if r.tracer.Enabled() {
			r.logger.Debug("skipping unknown Firehose log line", zap.String("line", line))
		}
		return nil, nil
	}

	if err != nil {
		chunks := strings.SplitN(line, " ", 2)
		return nil, fmt.Errorf("%s: %s (line %q)", chunks[0], err, line)
	}

	return out, nil
}

// Formats
// [READER_PROTOCOL_VERSION] sf.ethereum.type.v2.Block
// [READER_PROTOCOL_VERSION] sf.ethereum.type.v2.Block [NODE_NAME] [NODE_VERSION...]
// [READER_PROTOCOL_VERSION] [NODE_VARIANT] [NODE_VERSION...]
func (r *ConsoleReader) readInit(line string) error {
	chunks, err := splitInBoundedChunks(line, 2)
	if err != nil {
		return fmt.Errorf("split: %s", err)
	}

	r.readerProtocolVersion = chunks[0]

	switch r.readerProtocolVersion {
	// Implementation of RPC poller were set to use 1.0 so we keep support for it for now
	case "1.0", "3.0":
	// Supported
	//
	case "3.1":
		r.readPartialBlockIndex = true
	default:
		return fmt.Errorf("major version of Firehose exchange protocol is unsupported (expected: one of [1.0, 3.0], found %s), you are most probably running an incompatible version of the Firehose aware node client/node poller", r.readerProtocolVersion)
	}

	typeOrVariant := chunks[1]
	if typeOrVariant == "" {
		return fmt.Errorf("protobuf fully qualified name is empty, it must be set to a valid Protobuf fully qualified message type representing your block format")
	}

	blockType, nodeVariant, nodeVersion, err := resolveBlockType(typeOrVariant)
	if err != nil {
		return err
	}

	r.setProtoMessageType(blockType)

	r.logger.Info("console reader protocol version init",
		zap.String("version", r.readerProtocolVersion),
		zap.String("protobuf_fully_qualified_name", blockType),
		zap.String("node_variant", nodeVariant),
		zap.String("node_version", nodeVersion),
	)

	return nil
}

// Formats
// [block_num:342342342] [block_hash] [parent_num] [parent_hash] [lib:123123123] [timestamp:unix_nano] B64ENCODED_any // 1.0 or 3.0
// [block_num:342342342] [partial_block_idx] [block_hash] [parent_num] [parent_hash] [lib:123123123] [timestamp:unix_nano] B64ENCODED_any // 3.1
//
//	if [partial_block_idx] is greater than 1000, it means that it is the final partial
//	of a particular block and you must substract 1000 from it to get the actual index
func (r *ConsoleReader) readBlock(line string) (out *pbbstream.Block, err error) {
	if r.readerProtocolVersion == "" {
		return nil, errReaderProtocolVersionNotSet
	}

	chunks, err := splitInBoundedChunks(line, r.blockHeaderFieldCount()+1)
	if err != nil {
		return nil, fmt.Errorf("splitting block log line: %w", err)
	}

	encodedPayload := chunks[len(chunks)-1]
	return r.buildBlock(chunks[:len(chunks)-1], func() ([]byte, error) {
		payload, err := base64.StdEncoding.DecodeString(encodedPayload)
		if err != nil {
			return nil, fmt.Errorf("decoding payload %q: %w", encodedPayload, err)
		}

		return payload, nil
	})
}

// readDecodedBlock is readBlock for a "FIRE BLOCK" line whose payload was decoded while it
// was read.
func (r *ConsoleReader) readDecodedBlock(line *consoleline.Block) (out *pbbstream.Block, err error) {
	defer func() {
		if err != nil {
			err = fmt.Errorf("BLOCK: %s (line header %q)", err, line.Header)
		}
	}()

	if r.readerProtocolVersion == "" {
		return nil, errReaderProtocolVersionNotSet
	}

	fields, err := splitInBoundedChunks(line.Header, r.blockHeaderFieldCount())
	if err != nil {
		return nil, fmt.Errorf("splitting block log line: %w", err)
	}

	return r.buildBlock(fields, func() ([]byte, error) {
		return line.Payload, line.Err
	})
}

var errReaderProtocolVersionNotSet = errors.New("reader protocol version not set, did you forget to send the 'FIRE INIT <reader_protocol_version> <protobuf_fully_qualified_type>' line?")

// blockHeaderFieldCount is the number of fields before the payload in a "FIRE BLOCK" line.
func (r *ConsoleReader) blockHeaderFieldCount() int {
	if r.readPartialBlockIndex {
		return 7
	}
	return 6
}

// buildBlock builds the block out of the header fields of a "FIRE BLOCK" line, payload is
// called once the fields are validated.
func (r *ConsoleReader) buildBlock(chunks []string, payload func() ([]byte, error)) (out *pbbstream.Block, err error) {
	i := 0
	blockNum, err := strconv.ParseUint(chunks[i], 10, 64)
	if err != nil {
		return nil, fmt.Errorf("parsing block num %q: %w", chunks[0], err)
	}
	i++

	var partialBlockIndex int64
	var lastPartial bool
	if r.readPartialBlockIndex {
		partialBlockIndex, err = strconv.ParseInt(chunks[i], 10, 32)
		if err != nil {
			return nil, fmt.Errorf("parsing partial block index %q: %w", chunks[i], err)
		}
		if partialBlockIndex >= 1000 {
			partialBlockIndex -= 1000
			lastPartial = true
		}
		i++
	}

	blockHash := chunks[i]
	i++

	parentNum, err := strconv.ParseUint(chunks[i], 10, 64)
	if err != nil {
		return nil, fmt.Errorf("parsing parent num %q: %w", chunks[i], err)
	}
	i++

	parentHash := chunks[i]
	i++

	libNum, err := strconv.ParseUint(chunks[i], 10, 64)
	if err != nil {
		return nil, fmt.Errorf("parsing lib num %q: %w", chunks[i], err)
	}
	i++

	timestampUnixNano, err := strconv.ParseUint(chunks[i], 10, 64)
	if err != nil {
		return nil, fmt.Errorf("parsing timestamp %q: %w", chunks[i], err)
	}
	i++

	timestamp := time.Unix(0, int64(timestampUnixNano))

	payloadBytes, err := payload()
	if err != nil {
		return nil, err
	}

	blockPayload := &anypb.Any{
		TypeUrl: r.protoMessageType,
		Value:   payloadBytes,
	}

	block := &pbbstream.Block{
		Id:           blockHash,
		Number:       blockNum,
		ParentId:     parentHash,
		ParentNum:    parentNum,
		Timestamp:    timestamppb.New(timestamp),
		LibNum:       libNum,
		Payload:      blockPayload,
		PartialIndex: int32(partialBlockIndex),
		LastPartial:  lastPartial,
	}

	ConsoleReaderBlockReadCount.Inc()
	r.lastBlock = bstream.NewBlockRef(blockHash, blockNum)
	r.lastParentBlock = bstream.NewBlockRef(parentHash, parentNum)
	r.lastBlockTimestamp = timestamp
	r.lib = libNum

	return block, nil
}

func (r *ConsoleReader) setProtoMessageType(typeURL string) {
	if strings.HasPrefix(typeURL, "type.googleapis.com/") {
		r.protoMessageType = typeURL
		return
	}

	if strings.Contains(typeURL, "/") {
		panic(fmt.Sprintf("invalid type url %q, expecting type.googleapis.com/", typeURL))
	}

	r.protoMessageType = "type.googleapis.com/" + typeURL
}

// splitInBoundedChunks splits the line in `count` chunks and returns the slice `chunks[1:count]` (so exclusive end),
// but will accumulate all trailing chunks within the last (for free-form strings, or JSON objects)
func splitInBoundedChunks(line string, count int) ([]string, error) {
	chunks := strings.SplitN(line, " ", count)
	if len(chunks) != count {
		return nil, fmt.Errorf("%d fields required but found %d fields for line %q", count, len(chunks), line)
	}

	return chunks, nil
}