Support flexible schemas for LLO

core/services/llo/codecs.go

@@ -14,6 +14,7 @@ func NewReportCodecs(lggr logger.Logger, donID uint32) map[llotypes.ReportFormat
 
 	codecs[llotypes.ReportFormatJSON] = llo.JSONReportCodec{}
 	codecs[llotypes.ReportFormatEVMPremiumLegacy] = evm.NewReportCodecPremiumLegacy(lggr, donID)
+	codecs[llotypes.ReportFormatEVMABIEncodeUnpacked] = evm.NewReportCodecEVMABIEncodeUnpacked(lggr, donID)
 
 	return codecs
 }

core/services/llo/evm/fees.go

@@ -16,8 +16,9 @@ const Precision int32 = 18
 
 // CalculateFee outputs a fee in wei according to the formula: baseUSDFee / tokenPriceInUSD
 func CalculateFee(tokenPriceInUSD decimal.Decimal, baseUSDFee decimal.Decimal) *big.Int {
-	if tokenPriceInUSD.IsZero() || baseUSDFee.IsZero() {
+	if baseUSDFee.IsZero() || baseUSDFee.IsNegative() || tokenPriceInUSD.IsZero() || tokenPriceInUSD.IsNegative() {
 		// zero fee if token price or base fee is zero
+		// if either fee should somehow be negative, also, return zero
 		return big.NewInt(0)
 	}
 

core/services/llo/evm/fees_test.go

@@ -38,8 +38,27 @@ func Test_Fees(t *testing.T) {
 
 	t.Run("with base fee == 0", func(t *testing.T) {
 		tokenPriceInUSD := decimal.NewFromInt32(123)
-		BaseUSDFee = decimal.NewFromInt32(0)
-		fee := CalculateFee(tokenPriceInUSD, BaseUSDFee)
+		baseUSDFee := decimal.NewFromInt32(0)
+		fee := CalculateFee(tokenPriceInUSD, baseUSDFee)
+		assert.Equal(t, big.NewInt(0), fee)
+	})
+
+	t.Run("negative fee rounds up to zero", func(t *testing.T) {
+		tokenPriceInUSD := decimal.NewFromInt32(-123)
+		baseUSDFee := decimal.NewFromInt32(1)
+		fee := CalculateFee(tokenPriceInUSD, baseUSDFee)
+		assert.Equal(t, big.NewInt(0), fee)
+
+		tokenPriceInUSD = decimal.NewFromInt32(123)
+		baseUSDFee = decimal.NewFromInt32(-1)
+		fee = CalculateFee(tokenPriceInUSD, baseUSDFee)
+		assert.Equal(t, big.NewInt(0), fee)
+
+		// Multiple negative values also return a zero fee since negative
+		// prices are always nonsensical
+		tokenPriceInUSD = decimal.NewFromInt32(-123)
+		baseUSDFee = decimal.NewFromInt32(-1)
+		fee = CalculateFee(tokenPriceInUSD, baseUSDFee)
 		assert.Equal(t, big.NewInt(0), fee)
 	})
 

core/services/llo/evm/report_codec_evm_abi_encode_unpacked.go

@@ -0,0 +1,275 @@
+package evm
+
+import (
+	"context"
+	"encoding/json"
+	"errors"
+	"fmt"
+	"math/big"
+
+	"github.com/ethereum/go-ethereum/accounts/abi"
+	"github.com/ethereum/go-ethereum/common"
+	"github.com/shopspring/decimal"
+
+	"github.com/smartcontractkit/chainlink-common/pkg/logger"
+	llotypes "github.com/smartcontractkit/chainlink-common/pkg/types/llo"
+	"github.com/smartcontractkit/chainlink-data-streams/llo"
+	ubig "github.com/smartcontractkit/chainlink/v2/core/chains/evm/utils/big"
+)
+
+var (
+	_ llo.ReportCodec = ReportCodecEVMABIEncodeUnpacked{}
+
+	zero = big.NewInt(0)
+)
+
+type ReportCodecEVMABIEncodeUnpacked struct {
+	logger.Logger
+	donID uint32
+}
+
+func NewReportCodecEVMABIEncodeUnpacked(lggr logger.Logger, donID uint32) ReportCodecEVMABIEncodeUnpacked {
+	return ReportCodecEVMABIEncodeUnpacked{logger.Sugared(lggr).Named("ReportCodecEVMABIEncodeUnpacked"), donID}
+}
+
+// Opts format remains unchanged
+type ReportFormatEVMABIEncodeOpts struct {
+	// BaseUSDFee is the cost on-chain of verifying a report
+	BaseUSDFee decimal.Decimal `json:"baseUSDFee"`
+	// Expiration window is the length of time in seconds the report is valid
+	// for, from the observation timestamp
+	ExpirationWindow uint32 `json:"expirationWindow"`
+	// FeedID is for compatibility with existing on-chain verifiers
+	FeedID common.Hash `json:"feedID"`
+	// ABI defines the encoding of the payload. Each element maps to exactly
+	// one stream (although sub-arrays may be specified for streams that
+	// produce a composite data type).
+	//
+	// EXAMPLE
+	//
+	// [{"streamID":123,"multiplier":"10000","type":"uint192"}, ...]
+	//
+	// See definition of ABIEncoder struct for more details.
+	//
+	// The total number of streams must be 2+n, where n is the number of
+	// top-level elements in this ABI array (stream 0 is always the native
+	// token price and stream 1 is the link token price).
+	ABI []ABIEncoder `json:"abi"`
+}
+
+func (r *ReportFormatEVMABIEncodeOpts) Decode(opts []byte) error {
+	return json.Unmarshal(opts, r)
+}
+
+func (r *ReportFormatEVMABIEncodeOpts) Encode() ([]byte, error) {
+	return json.Marshal(r)
+}
+
+type EVMBaseReportFields struct {
+	FeedID             common.Hash
+	ValidFromTimestamp uint32
+	Timestamp          uint32
+	NativeFee          *big.Int
+	LinkFee            *big.Int
+	ExpiresAt          uint32
+}
+
+// TODO: Add VerifyOpts public function and add to interface for chainlink-data-streams?
+// Or just Verify(channelDefinitions) ? to handle e.g. unique feed IDs
+
+func (r ReportCodecEVMABIEncodeUnpacked) Encode(ctx context.Context, report llo.Report, cd llotypes.ChannelDefinition) ([]byte, error) {
+	if report.Specimen {
+		return nil, errors.New("ReportCodecEVMABIEncodeUnpacked does not support encoding specimen reports")
+	}
+	if len(report.Values) < 2 {
+		return nil, fmt.Errorf("ReportCodecEVMABIEncodeUnpacked requires at least 2 values (NativePrice, LinkPrice, ...); got report.Values: %v", report.Values)
+	}
+	nativePrice, err := extractPrice(report.Values[0])
+	if err != nil {
+		return nil, fmt.Errorf("ReportCodecEVMABIEncodeUnpacked failed to extract native price: %w", err)
+	}
+	linkPrice, err := extractPrice(report.Values[1])
+	if err != nil {
+		return nil, fmt.Errorf("ReportCodecEVMABIEncodeUnpacked failed to extract link price: %w", err)
+	}
+
+	// NOTE: It seems suboptimal to have to parse the opts on every encode but
+	// not sure how to avoid it. Should be negligible performance hit as long
+	// as Opts is small.
+	opts := ReportFormatEVMABIEncodeOpts{}
+	if err := (&opts).Decode(cd.Opts); err != nil {
+		return nil, fmt.Errorf("failed to decode opts; got: '%s'; %w", cd.Opts, err)
+	}
+
+	rf := EVMBaseReportFields{
+		FeedID:             opts.FeedID,
+		ValidFromTimestamp: report.ValidAfterSeconds + 1,
+		Timestamp:          report.ObservationTimestampSeconds,
+		NativeFee:          CalculateFee(nativePrice, opts.BaseUSDFee),
+		LinkFee:            CalculateFee(linkPrice, opts.BaseUSDFee),
+		ExpiresAt:          report.ObservationTimestampSeconds + opts.ExpirationWindow,
+	}
+
+	// TODO: Enable with verbose logging?
+	// r.Logger.Debugw("Encoding report", "report", report, "opts", opts, "nativePrice", nativePrice, "linkPrice", linkPrice, "quote", quote, "multiplier", multiplier, "rf", rf)
+
+	header, err := r.buildHeader(ctx, rf)
+	if err != nil {
+		return nil, fmt.Errorf("failed to build base report; %w", err)
+	}
+
+	payload, err := r.buildPayload(ctx, opts.ABI, report.Values[2:])
+	if err != nil {
+		return nil, fmt.Errorf("failed to build payload; %w", err)
+	}
+
+	return append(header, payload...), nil
+}
+
+// BaseSchema represents the fixed base schema that remains unchanged for all
+// EVMABIEncodeUnpacked reports.
+//
+// An arbitrary payload will be appended to this.
+var BaseSchema = getBaseSchema()
+
+func getBaseSchema() abi.Arguments {
+	mustNewType := func(t string) abi.Type {
+		result, err := abi.NewType(t, "", []abi.ArgumentMarshaling{})
+		if err != nil {
+			panic(fmt.Sprintf("Unexpected error during abi.NewType: %s", err))
+		}
+		return result
+	}
+	return abi.Arguments([]abi.Argument{
+		{Name: "feedId", Type: mustNewType("bytes32")},
+		{Name: "validFromTimestamp", Type: mustNewType("uint32")},
+		{Name: "observationsTimestamp", Type: mustNewType("uint32")},
+		{Name: "nativeFee", Type: mustNewType("uint192")},
+		{Name: "linkFee", Type: mustNewType("uint192")},
+		{Name: "expiresAt", Type: mustNewType("uint32")},
+	})
+}
+
+func (r ReportCodecEVMABIEncodeUnpacked) buildHeader(ctx context.Context, rf EVMBaseReportFields) ([]byte, error) {
+	var merr error
+	if rf.LinkFee == nil {
+		merr = errors.Join(merr, errors.New("linkFee may not be nil"))
+	} else if rf.LinkFee.Cmp(zero) < 0 {
+		merr = errors.Join(merr, fmt.Errorf("linkFee may not be negative (got: %s)", rf.LinkFee))
+	}
+	if rf.NativeFee == nil {
+		merr = errors.Join(merr, errors.New("nativeFee may not be nil"))
+	} else if rf.NativeFee.Cmp(zero) < 0 {
+		merr = errors.Join(merr, fmt.Errorf("nativeFee may not be negative (got: %s)", rf.NativeFee))
+	}
+	if merr != nil {
+		return nil, merr
+	}
+	b, err := BaseSchema.Pack(rf.FeedID, rf.ValidFromTimestamp, rf.Timestamp, rf.NativeFee, rf.LinkFee, rf.ExpiresAt)
+	if err != nil {
+		return nil, fmt.Errorf("failed to pack base report blob; %w", err)
+	}
+	return b, nil
+}
+
+func (r ReportCodecEVMABIEncodeUnpacked) buildPayload(ctx context.Context, encoders []ABIEncoder, values []llo.StreamValue) (payload []byte, merr error) {
+	if len(encoders) != len(values) {
+		return nil, fmt.Errorf("ABI and values length mismatch; ABI: %d, Values: %d", len(encoders), len(values))
+	}
+
+	for i, encoder := range encoders {
+		b, err := encoder.Encode(values[i])
+		if err != nil {
+			var vStr []byte
+			if values[i] == nil {
+				vStr = []byte("<nil>")
+			} else {
+				var marshalErr error
+				vStr, marshalErr = values[i].MarshalText()
+				if marshalErr != nil {
+					vStr = []byte(fmt.Sprintf("%v(failed to marshal: %s)", values[i], marshalErr))
+				}
+			}
+			merr = errors.Join(merr, fmt.Errorf("failed to encode stream value %s at index %d with abi %q; %w", string(vStr), i, encoder.Type, err))
+			continue
+		}
+		payload = append(payload, b...)
+	}
+
+	return payload, merr
+}
+
+// An ABIEncoder encodes exactly one stream value into a byte slice
+type ABIEncoder struct {
+	// StreamID is the ID of the stream that this encoder is responsible for.
+	// MANDATORY
+	StreamID llotypes.StreamID `json:"streamID"`
+	// Type is the ABI type of the stream value. E.g. "uint192", "int256", "bool", "string" etc.
+	// MANDATORY
+	Type string `json:"type"`
+	// Multiplier, if provided, will be multiplied with the stream value before
+	// encoding.
+	// OPTIONAL
+	Multiplier *ubig.Big `json:"multiplier"`
+}
+
+// getNormalizedMultiplier returns the multiplier as a decimal.Decimal, defaulting
+// to 1 if the multiplier is nil.
+// TODO: Verify its not negative
+func (a ABIEncoder) getNormalizedMultiplier() (multiplier decimal.Decimal) {
+	if a.Multiplier == nil {
+		multiplier = decimal.NewFromInt(1)
+	} else {
+		multiplier = decimal.NewFromBigInt(a.Multiplier.ToInt(), 0)
+	}
+	return
+}
+
+func (a ABIEncoder) applyMultiplier(d decimal.Decimal) *big.Int {
+	return d.Mul(a.getNormalizedMultiplier()).BigInt()
+}
+
+func (a ABIEncoder) Encode(value llo.StreamValue) ([]byte, error) {
+	switch sv := value.(type) {
+	case *llo.Decimal:
+		if sv == nil {
+			return nil, fmt.Errorf("expected non-nil *Decimal; got: %v", sv)
+		}
+		return packBigInt(a.applyMultiplier(sv.Decimal()), a.Type)
+	default:
+		return nil, fmt.Errorf("unhandled type; supported types are: *llo.Decimal; got: %T", value)
+	}
+}
+
+// TODO: How to handle overflow?
+
+// packBigInt encodes a *big.Int as a byte slice according to the given ABI type
+func packBigInt(val *big.Int, t string) (b []byte, err error) {
+	abiType, err := abi.NewType(t, "", []abi.ArgumentMarshaling{})
+	if err != nil {
+		return nil, fmt.Errorf("invalid ABI type %q; %w", abiType, err)
+	}
+
+	// Pack the value using ABI type
+	arguments := abi.Arguments{
+		{
+			Type: abiType,
+		},
+	}
+
+	switch t {
+	case "uint32":
+		// packing uint32 expects uint32 as argument
+		if val.BitLen() > 32 {
+			return nil, fmt.Errorf("value %v is too large for uint32", val)
+		}
+		b, err = arguments.Pack(uint32(val.Uint64()))
+	default:
+		b, err = arguments.Pack(val)
+	}
+	if err != nil {
+		return nil, fmt.Errorf("failed to pack value %v as %q: %w", val, t, err)
+	}
+
+	return b, nil
+}