integration_test.go

package llo_test

import (
	"crypto/ed25519"
	"encoding/binary"
	"encoding/hex"
	"encoding/json"
	"fmt"
	"math/big"
	"net/http"
	"net/http/httptest"
	"sort"
	"strings"
	"testing"
	"time"

	"github.com/ethereum/go-ethereum"
	"github.com/ethereum/go-ethereum/accounts/abi"
	"github.com/ethereum/go-ethereum/accounts/abi/bind"
	"github.com/ethereum/go-ethereum/common"
	gethtypes "github.com/ethereum/go-ethereum/core/types"
	"github.com/ethereum/go-ethereum/eth/ethconfig"
	"github.com/shopspring/decimal"
	"github.com/stretchr/testify/assert"
	"github.com/stretchr/testify/require"
	"go.uber.org/zap/zapcore"
	"golang.org/x/crypto/sha3"
	"google.golang.org/grpc/peer"
	"google.golang.org/protobuf/proto"

	"github.com/smartcontractkit/freeport"

	"github.com/smartcontractkit/libocr/offchainreporting2/types"
	"github.com/smartcontractkit/libocr/offchainreporting2plus/confighelper"
	"github.com/smartcontractkit/libocr/offchainreporting2plus/ocr3confighelper"
	ocr2types "github.com/smartcontractkit/libocr/offchainreporting2plus/types"

	"github.com/smartcontractkit/chainlink-common/pkg/sqlutil"
	llotypes "github.com/smartcontractkit/chainlink-common/pkg/types/llo"
	"github.com/smartcontractkit/chainlink-common/pkg/utils"
	datastreamsllo "github.com/smartcontractkit/chainlink-data-streams/llo"
	lloevm "github.com/smartcontractkit/chainlink-data-streams/llo/reportcodecs/evm"
	"github.com/smartcontractkit/chainlink-evm/gethwrappers/generated/link_token_interface"
	"github.com/smartcontractkit/chainlink-evm/gethwrappers/llo-feeds/generated/channel_config_store"
	"github.com/smartcontractkit/chainlink-evm/gethwrappers/llo-feeds/generated/configurator"
	"github.com/smartcontractkit/chainlink-evm/gethwrappers/llo-feeds/generated/destination_verifier"
	"github.com/smartcontractkit/chainlink-evm/gethwrappers/llo-feeds/generated/destination_verifier_proxy"
	"github.com/smartcontractkit/chainlink-evm/gethwrappers/llo-feeds/generated/fee_manager"
	"github.com/smartcontractkit/chainlink-evm/gethwrappers/llo-feeds/generated/reward_manager"
	"github.com/smartcontractkit/chainlink-evm/gethwrappers/llo-feeds/generated/verifier"
	"github.com/smartcontractkit/chainlink-evm/gethwrappers/llo-feeds/generated/verifier_proxy"
	"github.com/smartcontractkit/chainlink-evm/pkg/assets"
	"github.com/smartcontractkit/chainlink-evm/pkg/llo"
	"github.com/smartcontractkit/chainlink-evm/pkg/mercury"
	reportcodecv3 "github.com/smartcontractkit/chainlink-evm/pkg/mercury/v3/reportcodec"
	mercuryverifier "github.com/smartcontractkit/chainlink-evm/pkg/mercury/verifier"
	evmtestutils "github.com/smartcontractkit/chainlink-evm/pkg/testutils"
	evmtypes "github.com/smartcontractkit/chainlink-evm/pkg/types"
	evmutils "github.com/smartcontractkit/chainlink-evm/pkg/utils"

	"github.com/smartcontractkit/chainlink-common/keystore/corekeys/csakey"
	"github.com/smartcontractkit/chainlink/v2/core/config"
	"github.com/smartcontractkit/chainlink/v2/core/config/toml"
	"github.com/smartcontractkit/chainlink/v2/core/internal/cltest"
	"github.com/smartcontractkit/chainlink/v2/core/internal/testutils"
	"github.com/smartcontractkit/chainlink/v2/core/services/chainlink"
	"github.com/smartcontractkit/chainlink/v2/core/services/relay/evm"
)

var (
	fNodes        = uint8(1)
	nNodes        = 4 // number of nodes (not including bootstrap)
	reportTimeout = time.Second * 60
)

func setupBlockchain(t *testing.T, adders ...*bind.TransactOpts) (
	*bind.TransactOpts,
	evmtypes.Backend,
	*configurator.Configurator,
	common.Address,
	*destination_verifier.DestinationVerifier,
	common.Address,
	*destination_verifier_proxy.DestinationVerifierProxy,
	common.Address,
	*channel_config_store.ChannelConfigStore,
	common.Address,
	*verifier.Verifier,
	common.Address,
	*verifier_proxy.VerifierProxy,
	common.Address,
) {
	steve := evmtestutils.MustNewSimTransactor(t) // config contract deployer and owner
	genesisData := gethtypes.GenesisAlloc{steve.From: {Balance: assets.Ether(1000).ToInt()}}
	for _, adder := range adders {
		genesisData[adder.From] = gethtypes.Account{Balance: assets.Ether(1000).ToInt()}
	}
	backend := cltest.NewSimulatedBackend(t, genesisData, ethconfig.Defaults.Miner.GasCeil)
	backend.Commit()
	backend.Commit() // ensure starting block number at least 1

	// Configurator
	configuratorAddress, _, configurator, err := configurator.DeployConfigurator(steve, backend.Client())
	require.NoError(t, err)
	backend.Commit()

	// DestinationVerifierProxy
	destinationVerifierProxyAddr, _, verifierProxy, err := destination_verifier_proxy.DeployDestinationVerifierProxy(steve, backend.Client())
	require.NoError(t, err)
	backend.Commit()
	// DestinationVerifier
	destinationVerifierAddr, _, destinationVerifier, err := destination_verifier.DeployDestinationVerifier(steve, backend.Client(), destinationVerifierProxyAddr)
	require.NoError(t, err)
	backend.Commit()
	// AddVerifier
	_, err = verifierProxy.SetVerifier(steve, destinationVerifierAddr)
	require.NoError(t, err)
	backend.Commit()

	// Legacy mercury verifier
	legacyVerifier, legacyVerifierAddr, legacyVerifierProxy, legacyVerifierProxyAddr := setupLegacyMercuryVerifier(t, steve, backend)

	// ChannelConfigStore
	configStoreAddress, _, configStore, err := channel_config_store.DeployChannelConfigStore(steve, backend.Client())
	require.NoError(t, err)

	backend.Commit()

	return steve, backend, configurator, configuratorAddress, destinationVerifier, destinationVerifierAddr, verifierProxy, destinationVerifierProxyAddr, configStore, configStoreAddress, legacyVerifier, legacyVerifierAddr, legacyVerifierProxy, legacyVerifierProxyAddr
}

func setupLegacyMercuryVerifier(t *testing.T, steve *bind.TransactOpts, backend evmtypes.Backend) (*verifier.Verifier, common.Address, *verifier_proxy.VerifierProxy, common.Address) {
	linkTokenAddress, _, linkToken, err := link_token_interface.DeployLinkToken(steve, backend.Client())
	require.NoError(t, err)
	backend.Commit()
	_, err = linkToken.Transfer(steve, steve.From, big.NewInt(1000))
	require.NoError(t, err)
	backend.Commit()
	nativeTokenAddress, _, nativeToken, err := link_token_interface.DeployLinkToken(steve, backend.Client())
	require.NoError(t, err)
	backend.Commit()
	_, err = nativeToken.Transfer(steve, steve.From, big.NewInt(1000))
	require.NoError(t, err)
	backend.Commit()
	verifierProxyAddr, _, verifierProxy, err := verifier_proxy.DeployVerifierProxy(steve, backend.Client(), common.Address{}) // zero address for access controller disables access control
	require.NoError(t, err)
	backend.Commit()
	verifierAddress, _, verifier, err := verifier.DeployVerifier(steve, backend.Client(), verifierProxyAddr)
	require.NoError(t, err)
	backend.Commit()
	_, err = verifierProxy.InitializeVerifier(steve, verifierAddress)
	require.NoError(t, err)
	backend.Commit()
	rewardManagerAddr, _, rewardManager, err := reward_manager.DeployRewardManager(steve, backend.Client(), linkTokenAddress)
	require.NoError(t, err)
	backend.Commit()
	feeManagerAddr, _, _, err := fee_manager.DeployFeeManager(steve, backend.Client(), linkTokenAddress, nativeTokenAddress, verifierProxyAddr, rewardManagerAddr)
	require.NoError(t, err)
	backend.Commit()
	_, err = verifierProxy.SetFeeManager(steve, feeManagerAddr)
	require.NoError(t, err)
	backend.Commit()
	_, err = rewardManager.SetFeeManager(steve, feeManagerAddr)
	require.NoError(t, err)
	backend.Commit()
	return verifier, verifierAddress, verifierProxy, verifierProxyAddr
}

type Stream struct {
	id                 uint32
	baseBenchmarkPrice decimal.Decimal
	baseBid            decimal.Decimal
	baseAsk            decimal.Decimal
}

const (
	ethStreamID    = 52
	linkStreamID   = 53
	quoteStreamID1 = 55
	quoteStreamID2 = 56
)

var (
	quoteStreamFeedID1 = common.HexToHash(`0x0003111111111111111111111111111111111111111111111111111111111111`)
	quoteStreamFeedID2 = common.HexToHash(`0x0003222222222222222222222222222222222222222222222222222222222222`)
	ethStream          = Stream{
		id:                 ethStreamID,
		baseBenchmarkPrice: decimal.NewFromFloat32(2_976.39),
	}
	linkStream = Stream{
		id:                 linkStreamID,
		baseBenchmarkPrice: decimal.NewFromFloat32(13.25),
	}
	quoteStream1 = Stream{
		id:                 quoteStreamID1,
		baseBenchmarkPrice: decimal.NewFromFloat32(1000.1212),
		baseBid:            decimal.NewFromFloat32(998.5431),
		baseAsk:            decimal.NewFromFloat32(1001.6999),
	}
	quoteStream2 = Stream{
		id:                 quoteStreamID2,
		baseBenchmarkPrice: decimal.NewFromFloat32(500.1212),
		baseBid:            decimal.NewFromFloat32(499.5431),
		baseAsk:            decimal.NewFromFloat32(502.6999),
	}
)

// see: https://github.com/smartcontractkit/offchain-reporting/blob/master/lib/offchainreporting2plus/internal/config/ocr3config/public_config.go
type OCRConfig struct {
	DeltaProgress                           time.Duration
	DeltaResend                             time.Duration
	DeltaInitial                            time.Duration
	DeltaRound                              time.Duration
	DeltaGrace                              time.Duration
	DeltaCertifiedCommitRequest             time.Duration
	DeltaStage                              time.Duration
	RMax                                    uint64
	S                                       []int
	Oracles                                 []confighelper.OracleIdentityExtra
	ReportingPluginConfig                   []byte
	MaxDurationInitialization               *time.Duration
	MaxDurationQuery                        time.Duration
	MaxDurationObservation                  time.Duration
	MaxDurationShouldAcceptAttestedReport   time.Duration
	MaxDurationShouldTransmitAcceptedReport time.Duration
	F                                       int
	OnchainConfig                           []byte
}

func makeDefaultOCRConfig() *OCRConfig {
	defaultOnchainConfig, err := (&datastreamsllo.EVMOnchainConfigCodec{}).Encode(datastreamsllo.OnchainConfig{
		Version:                 1,
		PredecessorConfigDigest: nil,
	})
	if err != nil {
		panic(err)
	}
	return &OCRConfig{
		DeltaProgress:                           2 * time.Second,
		DeltaResend:                             20 * time.Second,
		DeltaInitial:                            400 * time.Millisecond,
		DeltaRound:                              500 * time.Millisecond,
		DeltaGrace:                              250 * time.Millisecond,
		DeltaCertifiedCommitRequest:             300 * time.Millisecond,
		DeltaStage:                              1 * time.Minute,
		RMax:                                    100,
		ReportingPluginConfig:                   []byte{},
		MaxDurationInitialization:               nil,
		MaxDurationQuery:                        0,
		MaxDurationObservation:                  250 * time.Millisecond,
		MaxDurationShouldAcceptAttestedReport:   0,
		MaxDurationShouldTransmitAcceptedReport: 0,
		F:                                       int(fNodes),
		OnchainConfig:                           defaultOnchainConfig,
	}
}

func WithPredecessorConfigDigest(predecessorConfigDigest ocr2types.ConfigDigest) OCRConfigOption {
	return func(cfg *OCRConfig) {
		onchainConfig, err := (&datastreamsllo.EVMOnchainConfigCodec{}).Encode(datastreamsllo.OnchainConfig{
			Version:                 1,
			PredecessorConfigDigest: &predecessorConfigDigest,
		})
		if err != nil {
			panic(err)
		}
		cfg.OnchainConfig = onchainConfig
	}
}

func WithOffchainConfig(offchainConfig datastreamsllo.OffchainConfig) OCRConfigOption {
	return func(cfg *OCRConfig) {
		offchainConfigEncoded, err := offchainConfig.Encode()
		if err != nil {
			panic(err)
		}
		cfg.ReportingPluginConfig = offchainConfigEncoded
	}
}

func WithOracles(oracles []confighelper.OracleIdentityExtra) OCRConfigOption {
	return func(cfg *OCRConfig) {
		cfg.Oracles = oracles
		cfg.S = []int{len(oracles)} // all oracles transmit by default
	}
}

type OCRConfigOption func(*OCRConfig)

func generateConfig(t *testing.T, opts ...OCRConfigOption) (signers []types.OnchainPublicKey, transmitters []types.Account, f uint8, outOnchainConfig []byte, offchainConfigVersion uint64, offchainConfig []byte) {
	cfg := makeDefaultOCRConfig()

	for _, opt := range opts {
		opt(cfg)
	}
	t.Logf("Using OCR config: %+v\n", cfg)
	var err error
	signers, transmitters, f, outOnchainConfig, offchainConfigVersion, offchainConfig, err = ocr3confighelper.ContractSetConfigArgsForTests(
		cfg.DeltaProgress,
		cfg.DeltaResend,
		cfg.DeltaInitial,
		cfg.DeltaRound,
		cfg.DeltaGrace,
		cfg.DeltaCertifiedCommitRequest,
		cfg.DeltaStage,
		cfg.RMax,
		cfg.S,
		cfg.Oracles,
		cfg.ReportingPluginConfig,
		cfg.MaxDurationInitialization,
		cfg.MaxDurationQuery,
		cfg.MaxDurationObservation,
		cfg.MaxDurationShouldAcceptAttestedReport,
		cfg.MaxDurationShouldTransmitAcceptedReport,
		cfg.F,
		cfg.OnchainConfig,
	)

	require.NoError(t, err)

	return
}

func setLegacyConfig(t *testing.T, donID uint32, steve *bind.TransactOpts, backend evmtypes.Backend, legacyVerifier *verifier.Verifier, legacyVerifierAddr common.Address, nodes []Node, oracles []confighelper.OracleIdentityExtra, inOffchainConfig datastreamsllo.OffchainConfig) ocr2types.ConfigDigest {
	signers, _, _, onchainConfig, offchainConfigVersion, offchainConfig := generateConfig(t, WithOracles(oracles), WithOffchainConfig(inOffchainConfig))

	signerAddresses, err := evm.OnchainPublicKeyToAddress(signers)
	require.NoError(t, err)
	offchainTransmitters := make([][32]byte, nNodes)
	for i := range nNodes {
		offchainTransmitters[i] = nodes[i].ClientPubKey
	}
	donIDPadded := llo.DonIDToBytes32(donID)
	_, err = legacyVerifier.SetConfig(steve, donIDPadded, signerAddresses, offchainTransmitters, fNodes, onchainConfig, offchainConfigVersion, offchainConfig, nil)
	require.NoError(t, err)

	// libocr requires a few confirmations to accept the config
	backend.Commit()
	backend.Commit()
	backend.Commit()
	backend.Commit()

	l, err := legacyVerifier.LatestConfigDigestAndEpoch(&bind.CallOpts{}, donIDPadded)
	require.NoError(t, err)

	return l.ConfigDigest
}

func setStagingConfig(t *testing.T, donID uint32, steve *bind.TransactOpts, backend evmtypes.Backend, configurator *configurator.Configurator, configuratorAddress common.Address, nodes []Node, opts ...OCRConfigOption) ocr2types.ConfigDigest {
	return setBlueGreenConfig(t, donID, steve, backend, configurator, configuratorAddress, nodes, opts...)
}

func setProductionConfig(t *testing.T, donID uint32, steve *bind.TransactOpts, backend evmtypes.Backend, configurator *configurator.Configurator, configuratorAddress common.Address, nodes []Node, opts ...OCRConfigOption) ocr2types.ConfigDigest {
	return setBlueGreenConfig(t, donID, steve, backend, configurator, configuratorAddress, nodes, opts...)
}

func setBlueGreenConfig(t *testing.T, donID uint32, steve *bind.TransactOpts, backend evmtypes.Backend, configurator *configurator.Configurator, configuratorAddress common.Address, nodes []Node, opts ...OCRConfigOption) ocr2types.ConfigDigest {
	signers, _, _, onchainConfig, offchainConfigVersion, offchainConfig := generateConfig(t, opts...)

	var onchainPubKeys [][]byte
	for _, signer := range signers {
		onchainPubKeys = append(onchainPubKeys, signer)
	}
	offchainTransmitters := make([][32]byte, nNodes)
	for i := range nNodes {
		offchainTransmitters[i] = nodes[i].ClientPubKey
	}
	donIDPadded := llo.DonIDToBytes32(donID)
	var isProduction bool
	{
		cfg, err := (&datastreamsllo.EVMOnchainConfigCodec{}).Decode(onchainConfig)
		require.NoError(t, err)
		isProduction = cfg.PredecessorConfigDigest == nil
	}
	var err error
	if isProduction {
		_, err = configurator.SetProductionConfig(steve, donIDPadded, onchainPubKeys, offchainTransmitters, fNodes, onchainConfig, offchainConfigVersion, offchainConfig)
	} else {
		_, err = configurator.SetStagingConfig(steve, donIDPadded, onchainPubKeys, offchainTransmitters, fNodes, onchainConfig, offchainConfigVersion, offchainConfig)
	}
	require.NoError(t, err)

	// libocr requires a few confirmations to accept the config
	backend.Commit()
	backend.Commit()
	backend.Commit()
	backend.Commit()

	var topic common.Hash
	if isProduction {
		topic = llo.ProductionConfigSet
	} else {
		topic = llo.StagingConfigSet
	}
	logs, err := backend.Client().FilterLogs(testutils.Context(t), ethereum.FilterQuery{Addresses: []common.Address{configuratorAddress}, Topics: [][]common.Hash{[]common.Hash{topic, donIDPadded}}})
	require.NoError(t, err)
	require.GreaterOrEqual(t, len(logs), 1)

	cfg, err := mercury.ConfigFromLog(logs[len(logs)-1].Data)
	require.NoError(t, err)

	return cfg.ConfigDigest
}

func promoteStagingConfig(t *testing.T, donID uint32, steve *bind.TransactOpts, backend evmtypes.Backend, configurator *configurator.Configurator, configuratorAddress common.Address, isGreenProduction bool) {
	donIDPadded := llo.DonIDToBytes32(donID)
	_, err := configurator.PromoteStagingConfig(steve, donIDPadded, isGreenProduction)
	require.NoError(t, err)

	// libocr requires a few confirmations to accept the config
	backend.Commit()
	backend.Commit()
	backend.Commit()
	backend.Commit()
}

func TestIntegration_LLO_evm_premium_legacy(t *testing.T) {
	t.Parallel()
	offchainConfigs := []datastreamsllo.OffchainConfig{
		{
			ProtocolVersion:                     0,
			DefaultMinReportIntervalNanoseconds: 0,
		},
		{
			ProtocolVersion:                     1,
			DefaultMinReportIntervalNanoseconds: 1,
		},
	}
	for _, offchainConfig := range offchainConfigs {
		t.Run(fmt.Sprintf("offchainConfig=%+v", offchainConfig), func(t *testing.T) {
			t.Parallel()
			testIntegrationLLOEVMPremiumLegacy(t, offchainConfig)
		})
	}
}

func testIntegrationLLOEVMPremiumLegacy(t *testing.T, offchainConfig datastreamsllo.OffchainConfig) {
	testStartTimeStamp := time.Now()
	multiplier := decimal.New(1, 18)
	expirationWindow := time.Hour / time.Second

	const salt = 100

	clientCSAKeys := make([]csakey.KeyV2, nNodes)
	clientPubKeys := make([]ed25519.PublicKey, nNodes)
	for i := range nNodes {
		k := big.NewInt(int64(salt + i))
		key := csakey.MustNewV2XXXTestingOnly(k)
		clientCSAKeys[i] = key
		clientPubKeys[i] = key.PublicKey
	}

	steve, backend, _, _, verifier, _, verifierProxy, _, configStore, configStoreAddress, legacyVerifier, legacyVerifierAddr, _, _ := setupBlockchain(t)
	fromBlock := 1

	// Setup bootstrap
	bootstrapCSAKey := csakey.MustNewV2XXXTestingOnly(big.NewInt(salt - 1))
	bootstrapNodePort := freeport.GetOne(t)
	appBootstrap, bootstrapPeerID, _, bootstrapKb, _ := setupNode(t, bootstrapNodePort, "bootstrap_llo", backend, bootstrapCSAKey, nil)
	bootstrapNode := Node{App: appBootstrap, KeyBundle: bootstrapKb}

	t.Run("using legacy verifier configuration contract, produces reports in v0.3 format", func(t *testing.T) {
		reqs := make(chan *packet, 100000)
		serverKey := csakey.MustNewV2XXXTestingOnly(big.NewInt(salt - 2))
		serverPubKey := serverKey.PublicKey
		srv := NewMercuryServer(t, serverKey, reqs)

		serverURL := startMercuryServer(t, srv, clientPubKeys)

		donID := uint32(995544)
		streams := []Stream{ethStream, linkStream, quoteStream1, quoteStream2}
		streamMap := make(map[uint32]Stream)
		for _, strm := range streams {
			streamMap[strm.id] = strm
		}

		// Setup oracle nodes
		oracles, nodes := setupNodes(t, nNodes, backend, clientCSAKeys, func(c *chainlink.Config) {
			c.Mercury.Transmitter.Protocol = ptr(config.MercuryTransmitterProtocolGRPC)
		})

		chainID := testutils.SimulatedChainID
		relayType := "evm"
		relayConfig := fmt.Sprintf(`
chainID = "%s"
fromBlock = %d
lloDonID = %d
lloConfigMode = "mercury"
`, chainID, fromBlock, donID)
		addBootstrapJob(t, bootstrapNode, legacyVerifierAddr, "job-2", relayType, relayConfig)

		// Channel definitions
		channelDefinitions := llotypes.ChannelDefinitions{
			1: {
				ReportFormat: llotypes.ReportFormatEVMPremiumLegacy,
				Streams: []llotypes.Stream{
					{
						StreamID:   ethStreamID,
						Aggregator: llotypes.AggregatorMedian,
					},
					{
						StreamID:   linkStreamID,
						Aggregator: llotypes.AggregatorMedian,
					},
					{
						StreamID:   quoteStreamID1,
						Aggregator: llotypes.AggregatorQuote,
					},
				},
				Opts: llotypes.ChannelOpts(fmt.Appendf(nil, `{"baseUSDFee":"0.1","expirationWindow":%d,"feedId":"0x%x","multiplier":"%s"}`, expirationWindow, quoteStreamFeedID1, multiplier.String())),
			},
			2: {
				ReportFormat: llotypes.ReportFormatEVMPremiumLegacy,
				Streams: []llotypes.Stream{
					{
						StreamID:   ethStreamID,
						Aggregator: llotypes.AggregatorMedian,
					},
					{
						StreamID:   linkStreamID,
						Aggregator: llotypes.AggregatorMedian,
					},
					{
						StreamID:   quoteStreamID2,
						Aggregator: llotypes.AggregatorQuote,
					},
				},
				Opts: llotypes.ChannelOpts(fmt.Appendf(nil, `{"baseUSDFee":"0.1","expirationWindow":%d,"feedId":"0x%x","multiplier":"%s"}`, expirationWindow, quoteStreamFeedID2, multiplier.String())),
			},
		}

		url, sha := newChannelDefinitionsServer(t, channelDefinitions)

		// Set channel definitions
		_, err := configStore.SetChannelDefinitions(steve, donID, url, sha)
		require.NoError(t, err)
		backend.Commit()

		pluginConfig := fmt.Sprintf(`servers = { "%s" = "%x" }
donID = %d
channelDefinitionsContractAddress = "0x%x"
channelDefinitionsContractFromBlock = %d`, serverURL, serverPubKey, donID, configStoreAddress, fromBlock)
		addOCRJobsEVMPremiumLegacy(t, streams, serverPubKey, serverURL, legacyVerifierAddr, bootstrapPeerID, bootstrapNodePort, nodes, configStoreAddress, clientPubKeys, pluginConfig, relayType, relayConfig)

		// Set config on configurator
		setLegacyConfig(
			t, donID, steve, backend, legacyVerifier, legacyVerifierAddr, nodes, oracles, offchainConfig,
		)

		// Set config on the destination verifier
		signerAddresses := make([]common.Address, len(oracles))
		for i, oracle := range oracles {
			signerAddresses[i] = common.BytesToAddress(oracle.OnchainPublicKey)
		}
		{
			recipientAddressesAndWeights := []destination_verifier.CommonAddressAndWeight{}

			_, err := verifier.SetConfig(steve, signerAddresses, fNodes, recipientAddressesAndWeights)
			require.NoError(t, err)
			backend.Commit()
		}

		t.Run("receives at least one report per channel from each oracle when EAs are at 100% reliability", func(t *testing.T) {
			// Expect at least one report per feed from each oracle
			seen := make(map[[32]byte]map[string]struct{})
			for _, cd := range channelDefinitions {
				var opts lloevm.ReportFormatEVMPremiumLegacyOpts
				err := json.Unmarshal(cd.Opts, &opts)
				require.NoError(t, err)
				// feedID will be deleted when all n oracles have reported
				seen[opts.FeedID] = make(map[string]struct{}, nNodes)
			}
			for {
				req, err := receiveWithTimeout(t, reqs, reportTimeout)
				require.NoError(t, err)
				assert.Equal(t, uint32(llotypes.ReportFormatEVMPremiumLegacy), req.req.ReportFormat)
				v := make(map[string]any)
				err = mercury.PayloadTypes.UnpackIntoMap(v, req.req.Payload)
				require.NoError(t, err)
				report, exists := v["report"]
				if !exists {
					t.Fatalf("expected payload %#v to contain 'report'", v)
				}
				reportElems := make(map[string]any)
				err = reportcodecv3.ReportTypes.UnpackIntoMap(reportElems, report.([]byte))
				require.NoError(t, err)

				feedID := reportElems["feedId"].([32]uint8)

				if _, exists := seen[feedID]; !exists {
					continue // already saw all oracles for this feed
				}

				var expectedBm, expectedBid, expectedAsk *big.Int
				if feedID == quoteStreamFeedID1 {
					expectedBm = quoteStream1.baseBenchmarkPrice.Mul(multiplier).BigInt()
					expectedBid = quoteStream1.baseBid.Mul(multiplier).BigInt()
					expectedAsk = quoteStream1.baseAsk.Mul(multiplier).BigInt()
				} else if feedID == quoteStreamFeedID2 {
					expectedBm = quoteStream2.baseBenchmarkPrice.Mul(multiplier).BigInt()
					expectedBid = quoteStream2.baseBid.Mul(multiplier).BigInt()
					expectedAsk = quoteStream2.baseAsk.Mul(multiplier).BigInt()
				} else {
					t.Fatalf("unrecognized feedID: 0x%x", feedID)
				}

				assert.GreaterOrEqual(t, reportElems["validFromTimestamp"].(uint32), uint32(testStartTimeStamp.Unix()))
				assert.GreaterOrEqual(t, int(reportElems["observationsTimestamp"].(uint32)), int(testStartTimeStamp.Unix()))
				assert.Equal(t, "33597747607000", reportElems["nativeFee"].(*big.Int).String())
				assert.Equal(t, "7547169811320755", reportElems["linkFee"].(*big.Int).String())
				assert.Equal(t, reportElems["observationsTimestamp"].(uint32)+uint32(expirationWindow), reportElems["expiresAt"].(uint32))
				assert.Equal(t, expectedBm.String(), reportElems["benchmarkPrice"].(*big.Int).String())
				assert.Equal(t, expectedBid.String(), reportElems["bid"].(*big.Int).String())
				assert.Equal(t, expectedAsk.String(), reportElems["ask"].(*big.Int).String())

				// emulate mercury server verifying report (local verification)
				{
					rv := mercuryverifier.NewVerifier()

					reportSigners, err := rv.Verify(mercuryverifier.SignedReport{
						RawRs:         v["rawRs"].([][32]byte),
						RawSs:         v["rawSs"].([][32]byte),
						RawVs:         v["rawVs"].([32]byte),
						ReportContext: v["reportContext"].([3][32]byte),
						Report:        v["report"].([]byte),
					}, fNodes, signerAddresses)
					require.NoError(t, err)
					assert.GreaterOrEqual(t, len(reportSigners), int(fNodes+1))
					assert.Subset(t, signerAddresses, reportSigners)
				}

				// test on-chain verification
				t.Run("on-chain verification", func(t *testing.T) {
					t.Skip("SKIP - MERC-6637")
					// Disabled because it flakes, sometimes returns "execution reverted"
					// No idea why
					// https://smartcontract-it.atlassian.net/browse/MERC-6637
					_, err = verifierProxy.Verify(steve, req.req.Payload, []byte{})
					require.NoError(t, err)
				})

				pr, ok := peer.FromContext(req.ctx)
				require.True(t, ok)
				t.Logf("oracle %x reported for 0x%x", pr.String(), feedID[:])

				seen[feedID][pr.String()] = struct{}{}
				if len(seen[feedID]) == nNodes {
					t.Logf("all oracles reported for 0x%x", feedID[:])
					delete(seen, feedID)
					if len(seen) == 0 {
						break // saw all oracles; success!
					}
				}
			}
		})
	})
}

func TestIntegration_LLO_multi_formats(t *testing.T) {
	t.Parallel()
	offchainConfigs := []datastreamsllo.OffchainConfig{
		{
			ProtocolVersion:                     0,
			DefaultMinReportIntervalNanoseconds: 0,
		},
		{
			ProtocolVersion:                     1,
			DefaultMinReportIntervalNanoseconds: 1,
		},
	}
	for _, offchainConfig := range offchainConfigs {
		t.Run(fmt.Sprintf("offchainConfig=%+v", offchainConfig), func(t *testing.T) {
			t.Parallel()
			testIntegrationLLOMultiFormats(t, offchainConfig)
		})
	}
}

func testIntegrationLLOMultiFormats(t *testing.T, offchainConfig datastreamsllo.OffchainConfig) {
	testStartTimeStamp := time.Now()
	expirationWindow := uint32(3600)

	const salt = 200

	clientCSAKeys := make([]csakey.KeyV2, nNodes)
	clientPubKeys := make([]ed25519.PublicKey, nNodes)
	for i := range nNodes {
		k := big.NewInt(int64(salt + i))
		key := csakey.MustNewV2XXXTestingOnly(k)
		clientCSAKeys[i] = key
		clientPubKeys[i] = key.PublicKey
	}

	steve, backend, configurator, configuratorAddress, _, _, _, _, configStore, configStoreAddress, _, _, _, _ := setupBlockchain(t)
	fromBlock := 1

	// Setup bootstrap
	bootstrapCSAKey := csakey.MustNewV2XXXTestingOnly(big.NewInt(salt - 1))
	bootstrapNodePort := freeport.GetOne(t)
	appBootstrap, bootstrapPeerID, _, bootstrapKb, _ := setupNode(t, bootstrapNodePort, "bootstrap_llo", backend, bootstrapCSAKey, nil)
	bootstrapNode := Node{App: appBootstrap, KeyBundle: bootstrapKb}

	t.Run("generates reports using multiple formats", func(t *testing.T) {
		packetCh := make(chan *packet, 100000)
		serverKey := csakey.MustNewV2XXXTestingOnly(big.NewInt(salt - 2))
		serverPubKey := serverKey.PublicKey
		srv := NewMercuryServer(t, serverKey, packetCh)

		serverURL := startMercuryServer(t, srv, clientPubKeys)

		donID := uint32(888333)
		streams := []Stream{ethStream, linkStream}
		streamMap := make(map[uint32]Stream)
		for _, strm := range streams {
			streamMap[strm.id] = strm
		}

		// Setup oracle nodes
		oracles, nodes := setupNodes(t, nNodes, backend, clientCSAKeys, func(c *chainlink.Config) {
			c.Mercury.Transmitter.Protocol = ptr(config.MercuryTransmitterProtocolGRPC)
		})

		chainID := testutils.SimulatedChainID
		relayType := "evm"
		relayConfig := fmt.Sprintf(`
chainID = "%s"
fromBlock = %d
lloDonID = %d
lloConfigMode = "bluegreen"
`, chainID, fromBlock, donID)
		addBootstrapJob(t, bootstrapNode, configuratorAddress, "job-4", relayType, relayConfig)

		dexBasedAssetPriceStreamID := uint32(1)
		marketStatusStreamID := uint32(2)
		baseMarketDepthStreamID := uint32(3)
		quoteMarketDepthStreamID := uint32(4)
		benchmarkPriceStreamID := uint32(5)
		binanceFundingRateStreamID := uint32(6)
		binanceFundingTimeStreamID := uint32(7)
		binanceFundingIntervalHoursStreamID := uint32(8)
		deribitFundingRateStreamID := uint32(9)
		deribitFundingTimeStreamID := uint32(10)
		deribitFundingIntervalHoursStreamID := uint32(11)
		timestampedStonkPriceStreamID := uint32(12)
		nullTimestampPriceStreamID := uint32(13)
		missingTimestampPriceStreamID := uint32(14)
		timestampedStreamValueValueStreamID := uint32(15)
		timestampedStreamValueTimestampStreamID := uint32(16)

		mustEncodeOpts := func(opts any) []byte {
			encoded, err := json.Marshal(opts)
			require.NoError(t, err)
			return encoded
		}

		standardMultiplier := sqlutil.NewI(1e18)
		millisToNanosMultiplier := sqlutil.NewI(1e6)

		const simpleStreamlinedChannelID = 5
		const complexStreamlinedChannelID = 6
		const sampleTimestampsStockPriceChannelID = 7
		const sampleTimestampedStreamValueChannelID = 8

		dexBasedAssetFeedID := evmutils.NewHash()
		rwaFeedID := evmutils.NewHash()
		benchmarkPriceFeedID := evmutils.NewHash()
		timestampedStreamValueFeedID := evmutils.NewHash()
		fundingRateFeedID := evmutils.NewHash()
		simpleStreamlinedFeedID := pad32bytes(simpleStreamlinedChannelID)
		complexStreamlinedFeedID := pad32bytes(complexStreamlinedChannelID)
		sampleTimestampsStockPriceFeedID := pad32bytes(sampleTimestampsStockPriceChannelID)

		// Channel definitions
		channelDefinitions := llotypes.ChannelDefinitions{
			// Sample DEX-based asset schema
			1: {
				ReportFormat: llotypes.ReportFormatEVMABIEncodeUnpacked,
				Streams: []llotypes.Stream{
					{
						StreamID:   ethStreamID,
						Aggregator: llotypes.AggregatorMedian,
					},
					{
						StreamID:   linkStreamID,
						Aggregator: llotypes.AggregatorMedian,
					},
					{
						StreamID:   dexBasedAssetPriceStreamID,
						Aggregator: llotypes.AggregatorMedian,
					},
					{
						StreamID:   baseMarketDepthStreamID,
						Aggregator: llotypes.AggregatorMedian,
					},
					{
						StreamID:   quoteMarketDepthStreamID,
						Aggregator: llotypes.AggregatorMedian,
					},
				},
				Opts: mustEncodeOpts(&lloevm.ReportFormatEVMABIEncodeOpts{
					BaseUSDFee:       decimal.NewFromFloat32(0.1),
					ExpirationWindow: expirationWindow,
					FeedID:           dexBasedAssetFeedID,
					ABI: []lloevm.ABIEncoder{
						newSingleABIEncoder("int192", standardMultiplier),
						newSingleABIEncoder("int192", nil),
						newSingleABIEncoder("int192", nil),
					},
				}),
			},
			// Sample RWA schema
			2: {
				ReportFormat: llotypes.ReportFormatEVMABIEncodeUnpacked,
				Streams: []llotypes.Stream{
					{
						StreamID:   ethStreamID,
						Aggregator: llotypes.AggregatorMedian,
					},
					{
						StreamID:   linkStreamID,
						Aggregator: llotypes.AggregatorMedian,
					},
					{
						StreamID:   marketStatusStreamID,
						Aggregator: llotypes.AggregatorMode,
					},
				},
				Opts: mustEncodeOpts(&lloevm.ReportFormatEVMABIEncodeOpts{
					BaseUSDFee:       decimal.NewFromFloat32(0.1),
					ExpirationWindow: expirationWindow,
					FeedID:           rwaFeedID,
					ABI: []lloevm.ABIEncoder{
						newSingleABIEncoder("uint32", nil),
					},
				}),
			},
			// Sample Benchmark price schema
			3: {
				ReportFormat: llotypes.ReportFormatEVMABIEncodeUnpacked,
				Streams: []llotypes.Stream{
					{
						StreamID:   ethStreamID,
						Aggregator: llotypes.AggregatorMedian,
					},
					{
						StreamID:   linkStreamID,
						Aggregator: llotypes.AggregatorMedian,
					},
					{
						StreamID:   benchmarkPriceStreamID,
						Aggregator: llotypes.AggregatorMedian,
					},
				},
				Opts: mustEncodeOpts(&lloevm.ReportFormatEVMABIEncodeOpts{
					BaseUSDFee:       decimal.NewFromFloat32(0.1),
					ExpirationWindow: expirationWindow,
					FeedID:           benchmarkPriceFeedID,
					ABI: []lloevm.ABIEncoder{
						newSingleABIEncoder("int192", standardMultiplier),
					},
				}),
			},
			// Sample funding rate scheam
			4: {
				ReportFormat: llotypes.ReportFormatEVMABIEncodeUnpacked,
				Streams: []llotypes.Stream{
					{
						StreamID:   ethStreamID,
						Aggregator: llotypes.AggregatorMedian,
					},
					{
						StreamID:   linkStreamID,
						Aggregator: llotypes.AggregatorMedian,
					},
					{
						StreamID:   binanceFundingRateStreamID,
						Aggregator: llotypes.AggregatorMedian,
					},
					{
						StreamID:   binanceFundingTimeStreamID,
						Aggregator: llotypes.AggregatorMedian,
					},
					{
						StreamID:   binanceFundingIntervalHoursStreamID,
						Aggregator: llotypes.AggregatorMedian,
					},
					{
						StreamID:   deribitFundingRateStreamID,
						Aggregator: llotypes.AggregatorMedian,
					},
					{
						StreamID:   deribitFundingTimeStreamID,
						Aggregator: llotypes.AggregatorMedian,
					},
					{
						StreamID:   deribitFundingIntervalHoursStreamID,
						Aggregator: llotypes.AggregatorMedian,
					},
				},
				Opts: mustEncodeOpts(&lloevm.ReportFormatEVMABIEncodeOpts{
					BaseUSDFee:       decimal.NewFromFloat32(0.1),
					ExpirationWindow: expirationWindow,
					FeedID:           fundingRateFeedID,
					ABI: []lloevm.ABIEncoder{
						newSingleABIEncoder("int192", nil),
						newSingleABIEncoder("int192", nil),
						newSingleABIEncoder("int192", nil),
						newSingleABIEncoder("int192", nil),
						newSingleABIEncoder("int192", nil),
						newSingleABIEncoder("int192", nil),
					},
				}),
			},
			// Simple sample streamlined schema
			simpleStreamlinedChannelID: {
				ReportFormat: llotypes.ReportFormatEVMStreamlined,
				Streams: []llotypes.Stream{
					{
						StreamID:   ethStreamID,
						Aggregator: llotypes.AggregatorMedian,
					},
				},
				Opts: mustEncodeOpts(&lloevm.ReportFormatEVMStreamlinedOpts{
					ABI: []lloevm.ABIEncoder{
						newSingleABIEncoder("int128", standardMultiplier),
					},
				}),
			},
			// Complex sample streamlined schema
			complexStreamlinedChannelID: {
				ReportFormat: llotypes.ReportFormatEVMStreamlined,
				Streams: []llotypes.Stream{
					{
						StreamID:   ethStreamID,
						Aggregator: llotypes.AggregatorMedian,
					},
					{
						StreamID:   linkStreamID,
						Aggregator: llotypes.AggregatorMedian,
					},
					{
						StreamID:   dexBasedAssetPriceStreamID,
						Aggregator: llotypes.AggregatorMode,
					},
				},
				Opts: mustEncodeOpts(&lloevm.ReportFormatEVMStreamlinedOpts{
					ABI: []lloevm.ABIEncoder{
						newSingleABIEncoder("int192", standardMultiplier),
						newSingleABIEncoder("int8", sqlutil.NewI(1)),
						newSingleABIEncoder("uint64", sqlutil.NewI(100)),
					},
				}),
			},
			// Sample timestamped stock price schema/RWA
			sampleTimestampsStockPriceChannelID: {
				ReportFormat: llotypes.ReportFormatJSON,
				Streams: []llotypes.Stream{
					{
						StreamID:   timestampedStonkPriceStreamID,
						Aggregator: llotypes.AggregatorMedian,
					},
					{
						StreamID:   nullTimestampPriceStreamID,
						Aggregator: llotypes.AggregatorMedian,
					},
					{
						StreamID:   missingTimestampPriceStreamID,
						Aggregator: llotypes.AggregatorMedian,
					},
				},
			},
			// Sample timestamped stream value schema
			sampleTimestampedStreamValueChannelID: {
				ReportFormat: llotypes.ReportFormatEVMABIEncodeUnpacked,
				Streams: []llotypes.Stream{
					{
						StreamID:   ethStreamID,
						Aggregator: llotypes.AggregatorMedian,
					},
					{
						StreamID:   linkStreamID,
						Aggregator: llotypes.AggregatorMedian,
					},
					{
						StreamID:   timestampedStreamValueValueStreamID,
						Aggregator: llotypes.AggregatorMedian,
					},
					{
						StreamID:   timestampedStreamValueTimestampStreamID,
						Aggregator: llotypes.AggregatorMedian,
					},
				},
				Opts: mustEncodeOpts(&lloevm.ReportFormatEVMABIEncodeOpts{
					BaseUSDFee:       decimal.NewFromFloat32(0.1),
					ExpirationWindow: expirationWindow,
					FeedID:           timestampedStreamValueFeedID,
					ABI: []lloevm.ABIEncoder{
						newSingleABIEncoder("int192", standardMultiplier),
						newSingleABIEncoder("uint64", millisToNanosMultiplier),
					},
				}),
			},
		}
		url, sha := newChannelDefinitionsServer(t, channelDefinitions)

		// Set channel definitions
		_, err := configStore.SetChannelDefinitions(steve, donID, url, sha)
		require.NoError(t, err)
		backend.Commit()

		pluginConfig := fmt.Sprintf(`servers = { "%s" = "%x" }
donID = %d
channelDefinitionsContractAddress = "0x%x"
channelDefinitionsContractFromBlock = %d`, serverURL, serverPubKey, donID, configStoreAddress, fromBlock)

		bridgeName := "superbridge"

		responseJSON := `{
	"data": {
		"benchmarkPrice": "111.22",
		"marketStatus": 1
	},
	"result": {
		"benchmarkPrice": "2976.39",
		"benchmarkPrice2": "3156.79",
		"baseMarketDepth": "1000.1212",
		"quoteMarketDepth": "998.5431",
		"binanceFundingRate": "1234.5678",
		"binanceFundingTime": "1630000000",
		"binanceFundingIntervalHours": "8",
		"deribitFundingRate": "5432.2345",
		"deribitFundingTime": "1630000000",
		"deribitFundingIntervalHours": "8",
		"ethPrice": "3976.39",
		"linkPrice": "23.45"
	},
	"timestamps": {
		"providerIndicatedTimeUnixMs": 1742314713000,
		"providerIndicatedTimeUnixMs2": 1742314717000,
		"providerIndicatedTimeUnixMs_TestNull": null,
		"providerDataReceivedUnixMs": 1742314713050
	}
}`

		pricePipeline := fmt.Sprintf(`
dp          [type=bridge name="%s" requestData="{\\"data\\":{\\"data\\":\\"foo\\"}}"];
eth_parse    					[type=jsonparse path="result,ethPrice"];
eth_decimal 					[type=multiply times=1 streamID=%d];
link_parse    				[type=jsonparse path="result,linkPrice"];
link_decimal 				[type=multiply times=1 streamID=%d];
dp -> eth_parse -> eth_decimal;
dp -> link_parse -> link_decimal;
`, bridgeName, ethStreamID, linkStreamID)

		dexBasedAssetPipeline := fmt.Sprintf(`
dp          [type=bridge name="%s" requestData="{\\"data\\":{\\"data\\":\\"foo\\"}}"];

bp_parse    				[type=jsonparse path="result,benchmarkPrice"];
base_market_depth_parse   	[type=jsonparse path="result,baseMarketDepth"];
quote_market_depth_parse    [type=jsonparse path="result,quoteMarketDepth"];

bp_decimal 					[type=multiply times=1 streamID=%d];
base_market_depth_decimal   [type=multiply times=1 streamID=%d];
quote_market_depth_decimal  [type=multiply times=1 streamID=%d];

dp -> bp_parse -> bp_decimal;
dp -> base_market_depth_parse -> base_market_depth_decimal;
dp -> quote_market_depth_parse -> quote_market_depth_decimal;
`, bridgeName, dexBasedAssetPriceStreamID, baseMarketDepthStreamID, quoteMarketDepthStreamID)

		// Don't use a multiply task so that the task result has int64 type.
		rwaPipeline := fmt.Sprintf(`
dp [type=bridge name="%s" requestData="{\\"data\\":{\\"data\\":\\"foo\\"}}"];

market_status_parse [type=jsonparse path="data,marketStatus" streamID=%d];
stonk_price_parse [type=jsonparse path="data,benchmarkPrice"];

dp -> market_status_parse;

provider_indicated_time_parse [type=jsonparse lax=true path="timestamps,providerIndicatedTimeUnixMs"];
provider_data_received_parse [type=jsonparse lax=true path="timestamps,providerDataReceivedUnixMs"];
provider_indicated_time [type=median lax=true];
provider_data_received [type=median lax=true];
stonk_price_timestamped [type=merge left="{}" right="{\\"streamValueType\\": %d, \\"timestamps\\":{\\"providerIndicatedTimeUnixMs\\":$(provider_indicated_time),\\"providerDataReceivedUnixMs\\":$(provider_data_received)}, \\"result\\": $(stonk_price_parse)}" streamID=%d];

dp -> provider_indicated_time_parse -> provider_indicated_time;
dp -> provider_data_received_parse -> provider_data_received;
dp -> stonk_price_parse -> stonk_price_timestamped;

# test null providerIndicatedTimeUnixMs
null_provider_indicated_time_parse [type=jsonparse lax=true path="timestamps,providerIndicatedTimeUnixMs_TestNull"];
null_provider_indicated_time [type=median lax=true];
stonk_price_timestamped_null_indicated_time [type=merge left="{}" right="{\\"streamValueType\\": %d, \\"timestamps\\":{\\"providerIndicatedTimeUnixMs\\":$(null_provider_indicated_time),\\"providerDataReceivedUnixMs\\":$(provider_data_received)}, \\"result\\": $(stonk_price_parse)}" streamID=%d];

dp -> null_provider_indicated_time_parse -> null_provider_indicated_time;
dp -> stonk_price_parse -> stonk_price_timestamped_null_indicated_time;

# test missing providerIndicatedTimeUnixMs
missing_provider_indicated_time_parse [type=jsonparse lax=true path="timestamps,providerIndicatedTimeUnixMs_TestMissing"];
missing_provider_indicated_time [type=median lax=true];
stonk_price_timestamped_missing_indicated_time [type=merge left="{}" right="{\\"streamValueType\\": %d, \\"timestamps\\":{\\"providerIndicatedTimeUnixMs\\":$(missing_provider_indicated_time),\\"providerDataReceivedUnixMs\\":$(provider_data_received)}, \\"result\\": $(stonk_price_parse)}" streamID=%d];

dp -> missing_provider_indicated_time_parse -> missing_provider_indicated_time;
dp -> stonk_price_parse -> stonk_price_timestamped_missing_indicated_time;
`, bridgeName, marketStatusStreamID,
			datastreamsllo.LLOStreamValue_TimestampedStreamValue, timestampedStonkPriceStreamID,
			datastreamsllo.LLOStreamValue_TimestampedStreamValue, nullTimestampPriceStreamID,
			datastreamsllo.LLOStreamValue_TimestampedStreamValue, missingTimestampPriceStreamID,
		)

		benchmarkPricePipeline := fmt.Sprintf(`
dp          [type=bridge name="%s" requestData="{\\"data\\":{\\"data\\":\\"foo\\"}}"];

bp_parse    				[type=jsonparse path="result,benchmarkPrice"];
bp_decimal 					[type=multiply times=1 streamID=%d];

dp -> bp_parse -> bp_decimal;
`, bridgeName, benchmarkPriceStreamID)

		timestampedStreamValuePipeline := fmt.Sprintf(`
ds1_payload [type=bridge name="%s" requestData="{\\"data\\":{\\"data\\":\\"foo\\"}}"];
ds1_benchmark [type=jsonparse path="result,benchmarkPrice"];
ds1_payload -> ds1_benchmark -> benchmark_price;
ds1_provider_indicated_time [type=jsonparse lax=true path="timestamps,providerIndicatedTimeUnixMs"];
ds1_payload -> ds1_provider_indicated_time -> provider_indicated_time;
ds1_data_received_time [type=jsonparse lax=true path="timestamps,providerDataReceivedUnixMs"];
ds1_payload -> ds1_data_received_time -> data_received_time;

ds2_payload [type=bridge name="%s" requestData="{\\"data\\":{\\"data\\":\\"foo\\"}}"];
ds2_benchmark [type=jsonparse path="result,benchmarkPrice2"];
ds2_payload -> ds2_benchmark -> benchmark_price;
ds2_provider_indicated_time [type=jsonparse lax=true path="timestamps,providerIndicatedTimeUnixMs2"];
ds2_payload -> ds2_provider_indicated_time -> provider_indicated_time;
ds2_data_received_time [type=jsonparse lax=true path="timestamps,providerDataReceivedUnixMs"];
ds2_payload -> ds2_data_received_time -> data_received_time;

benchmark_price [type=median allowedFaults=1 streamID=%d index=0];
provider_indicated_time [type=median allowedFaults=1 lax=true];
data_received_time [type=median allowedFaults=1 lax=true];
provider_indicated_time -> benchmark_price_timestamp;
data_received_time -> benchmark_price_timestamp;

benchmark_price_timestamp [type=coalesce streamID=%d index=1];
`, bridgeName, bridgeName, timestampedStreamValueValueStreamID, timestampedStreamValueTimestampStreamID)

		fundingRatePipeline := fmt.Sprintf(`
dp          [type=bridge name="%s" requestData="{\\"data\\":{\\"data\\":\\"foo\\"}}"];

binance_funding_rate_parse   [type=jsonparse path="result,binanceFundingRate"];
binance_funding_rate_decimal [type=multiply times=1 streamID=%d];

binance_funding_time_parse   [type=jsonparse path="result,binanceFundingTime"];
binance_funding_time_decimal [type=multiply times=1 streamID=%d];

binance_funding_interval_hours_parse   [type=jsonparse path="result,binanceFundingIntervalHours"];
binance_funding_interval_hours_decimal [type=multiply times=1 streamID=%d];

deribit_funding_rate_parse   [type=jsonparse path="result,deribitFundingRate"];
deribit_funding_rate_decimal [type=multiply times=1 streamID=%d];

deribit_funding_time_parse   [type=jsonparse path="result,deribitFundingTime"];
deribit_funding_time_decimal [type=multiply times=1 streamID=%d];

deribit_funding_interval_hours_parse   [type=jsonparse path="result,deribitFundingIntervalHours"];
deribit_funding_interval_hours_decimal [type=multiply times=1 streamID=%d];

dp -> binance_funding_rate_parse -> binance_funding_rate_decimal;
dp -> binance_funding_time_parse -> binance_funding_time_decimal;
dp -> binance_funding_interval_hours_parse -> binance_funding_interval_hours_decimal;
dp -> deribit_funding_rate_parse -> deribit_funding_rate_decimal;
dp -> deribit_funding_time_parse -> deribit_funding_time_decimal;
dp -> deribit_funding_interval_hours_parse -> deribit_funding_interval_hours_decimal;

`, bridgeName, binanceFundingRateStreamID, binanceFundingTimeStreamID, binanceFundingIntervalHoursStreamID, deribitFundingRateStreamID, deribitFundingTimeStreamID, deribitFundingIntervalHoursStreamID)

		for i, node := range nodes {
			// superBridge returns a JSON with everything you want in it,
			// stream specs can just pick the individual fields they need
			createBridge(t, bridgeName, responseJSON, node.App.BridgeORM())
			addStreamSpec(t, node, "pricePipeline", nil, pricePipeline)
			addStreamSpec(t, node, "dexBasedAssetPipeline", nil, dexBasedAssetPipeline)
			addStreamSpec(t, node, "rwaPipeline", nil, rwaPipeline)
			addStreamSpec(t, node, "benchmarkPricePipeline", nil, benchmarkPricePipeline)
			addStreamSpec(t, node, "timestampedStreamValuePipeline", nil, timestampedStreamValuePipeline)
			addStreamSpec(t, node, "fundingRatePipeline", nil, fundingRatePipeline)

			addLLOJob(
				t,
				node,
				configuratorAddress,
				bootstrapPeerID,
				bootstrapNodePort,
				clientPubKeys[i],
				"llo-evm-abi-encode-unpacked-test",
				pluginConfig,
				relayType,
				relayConfig,
			)
		}

		// Set config on configurator
		digest := setProductionConfig(
			t, donID, steve, backend, configurator, configuratorAddress, nodes, WithOracles(oracles), WithOffchainConfig(offchainConfig),
		)

		// NOTE: Wait for one of each type of report
		feedIDs := map[[32]byte]struct{}{
			dexBasedAssetFeedID:              {},
			rwaFeedID:                        {},
			benchmarkPriceFeedID:             {},
			timestampedStreamValueFeedID:     {},
			fundingRateFeedID:                {},
			simpleStreamlinedFeedID:          {},
			complexStreamlinedFeedID:         {},
			sampleTimestampsStockPriceFeedID: {},
		}

		for {
			pckt, err := receiveWithTimeout(t, packetCh, reportTimeout)
			require.NoError(t, err)
			req := pckt.req
			switch req.ReportFormat {
			case uint32(llotypes.ReportFormatEVMABIEncodeUnpacked):
				v := make(map[string]any)
				err = mercury.PayloadTypes.UnpackIntoMap(v, req.Payload)
				require.NoError(t, err)
				report, exists := v["report"]
				if !exists {
					t.Fatalf("expected payload %#v to contain 'report'", v)
				}
				reportCtx, exists := v["reportContext"]
				if !exists {
					t.Fatalf("expected payload %#v to contain 'reportContext'", v)
				}

				// Check the report context
				assert.Equal(t, [32]byte(digest), reportCtx.([3][32]uint8)[0])                                                                      // config digest
				assert.Equal(t, "000000000000000000000000000000000000000000000000000d8e0d00000001", fmt.Sprintf("%x", reportCtx.([3][32]uint8)[2])) // extra hash

				reportElems := make(map[string]any)
				err = lloevm.BaseSchemaUint32.UnpackIntoMap(reportElems, report.([]byte))
				require.NoError(t, err)

				feedID := reportElems["feedId"].([32]uint8)
				delete(feedIDs, feedID)

				// Check headers
				assert.GreaterOrEqual(t, reportElems["validFromTimestamp"].(uint32), uint32(testStartTimeStamp.Unix())) //nolint:gosec // G115
				assert.GreaterOrEqual(t, int(reportElems["observationsTimestamp"].(uint32)), int(testStartTimeStamp.Unix()))
				// Zero fees since both eth/link stream specs are missing, don't
				// care about billing for purposes of this test
				assert.Equal(t, "25148438659186", reportElems["nativeFee"].(*big.Int).String())
				assert.Equal(t, "4264392324093817", reportElems["linkFee"].(*big.Int).String())
				assert.Equal(t, reportElems["observationsTimestamp"].(uint32)+expirationWindow, reportElems["expiresAt"].(uint32))

				// Check payload values
				payload := report.([]byte)[192:]
				switch hex.EncodeToString(feedID[:]) {
				case hex.EncodeToString(dexBasedAssetFeedID[:]):
					require.Len(t, payload, 96)
					args := abi.Arguments([]abi.Argument{
						{Name: "benchmarkPrice", Type: mustNewType("int192")},
						{Name: "baseMarketDepth", Type: mustNewType("int192")},
						{Name: "quoteMarketDepth", Type: mustNewType("int192")},
					})
					v := make(map[string]any)
					err := args.UnpackIntoMap(v, payload)
					require.NoError(t, err)

					assert.Equal(t, "2976390000000000000000", v["benchmarkPrice"].(*big.Int).String())
					assert.Equal(t, "1000", v["baseMarketDepth"].(*big.Int).String())
					assert.Equal(t, "998", v["quoteMarketDepth"].(*big.Int).String())
				case hex.EncodeToString(rwaFeedID[:]):
					require.Len(t, payload, 32)
					args := abi.Arguments([]abi.Argument{
						{Name: "marketStatus", Type: mustNewType("uint32")},
					})
					v := make(map[string]any)
					err := args.UnpackIntoMap(v, payload)
					require.NoError(t, err)

					assert.Equal(t, uint32(1), v["marketStatus"].(uint32))
				case hex.EncodeToString(benchmarkPriceFeedID[:]):
					require.Len(t, payload, 32)
					args := abi.Arguments([]abi.Argument{
						{Name: "benchmarkPrice", Type: mustNewType("int192")},
					})
					v := make(map[string]any)
					err := args.UnpackIntoMap(v, payload)
					require.NoError(t, err)

					assert.Equal(t, "2976390000000000000000", v["benchmarkPrice"].(*big.Int).String())
				case hex.EncodeToString(timestampedStreamValueFeedID[:]):
					require.Len(t, payload, 64)
					args := abi.Arguments([]abi.Argument{
						{Name: "benchmarkPrice", Type: mustNewType("int192")},
						{Name: "benchmarkPriceTimestamp", Type: mustNewType("uint64")},
					})
					v := make(map[string]any)
					err := args.UnpackIntoMap(v, payload)
					require.NoError(t, err)

					assert.Equal(t, "3066590000000000000000", v["benchmarkPrice"].(*big.Int).String())
					assert.Equal(t, uint64(1742314715000_000_000), v["benchmarkPriceTimestamp"].(uint64))
				case hex.EncodeToString(fundingRateFeedID[:]):
					require.Len(t, payload, 192)
					args := abi.Arguments([]abi.Argument{
						{Name: "binanceFundingRate", Type: mustNewType("int192")},
						{Name: "binanceFundingTime", Type: mustNewType("int192")},
						{Name: "binanceFundingIntervalHours", Type: mustNewType("int192")},
						{Name: "deribitFundingRate", Type: mustNewType("int192")},
						{Name: "deribitFundingTime", Type: mustNewType("int192")},
						{Name: "deribitFundingIntervalHours", Type: mustNewType("int192")},
					})
					v := make(map[string]any)
					err := args.UnpackIntoMap(v, payload)
					require.NoError(t, err)

					assert.Equal(t, "1234", v["binanceFundingRate"].(*big.Int).String())
					assert.Equal(t, "1630000000", v["binanceFundingTime"].(*big.Int).String())
					assert.Equal(t, "8", v["binanceFundingIntervalHours"].(*big.Int).String())
					assert.Equal(t, "5432", v["deribitFundingRate"].(*big.Int).String())
					assert.Equal(t, "1630000000", v["deribitFundingTime"].(*big.Int).String())
					assert.Equal(t, "8", v["deribitFundingIntervalHours"].(*big.Int).String())
				default:
					t.Fatalf("unexpected feedID: %x", feedID)
				}
			case uint32(llotypes.ReportFormatEVMStreamlined):
				p := &lloevm.LLOEVMStreamlinedReportWithContext{}
				require.NoError(t, proto.Unmarshal(req.Payload, p))
				// proto auxiliary fields
				assert.Equal(t, digest[:], p.ConfigDigest)
				assert.Greater(t, p.SeqNr, uint64(1))

				// payload check
				payload := p.PackedPayload
				assert.Equal(t, digest[:], payload[:32])
				lenReport := int(binary.BigEndian.Uint16(payload[32:34]))
				report := make([]byte, lenReport)
				copy(report, payload[34:])
				numSigs := payload[34+lenReport]
				assert.Equal(t, int(fNodes+1), int(numSigs))
				assert.Len(t, payload, 32+2+lenReport+1+int(numSigs)*65)

				// report contents check
				// uint32 report format
				// uint32 channel ID
				rfBytes := report[:4]
				rf := binary.BigEndian.Uint32(rfBytes)
				assert.Equal(t, uint32(llotypes.ReportFormatEVMStreamlined), rf)
				cidBytes := report[4:8]
				cid := binary.BigEndian.Uint32(cidBytes)
				switch cid {
				case simpleStreamlinedChannelID:
					assert.Len(t, report, 32)
					tsbytes := report[8:16]
					ts := binary.BigEndian.Uint64(tsbytes)
					assert.GreaterOrEqual(t, ts, uint64(testStartTimeStamp.Unix())) //nolint:gosec // g115
					// int128
					assert.Equal(t, "00000000000000d78f7f252ecf870000", hex.EncodeToString(report[16:]))
				case complexStreamlinedChannelID:
					assert.Len(t, report, 49)
					tsbytes := report[8:16]
					ts := binary.BigEndian.Uint64(tsbytes)
					assert.GreaterOrEqual(t, ts, uint64(testStartTimeStamp.Unix())) //nolint:gosec // g115
					// int192, int8, uint64 stream values
					assert.Equal(t, "000000000000000000000000000000d78f7f252ecf870000", hex.EncodeToString(report[16:40]))
					assert.Equal(t, "17", hex.EncodeToString(report[40:41]))
					assert.Equal(t, "0000000000048aa7", hex.EncodeToString(report[41:]))
				default:
					t.Fatalf("unexpected channel: %d", cid)
				}
				delete(feedIDs, pad32bytes(cid))
			case uint32(llotypes.ReportFormatJSON):
				v := make(map[string]any)
				err := json.Unmarshal(req.Payload, &v)
				require.NoError(t, err)
				report := v["report"].(map[string]any)
				cid := report["ChannelID"].(float64)
				delete(feedIDs, pad32bytes(uint32(cid)))
				assert.Len(t, report["Values"].([]any), 3)
				// default uses provider indicated time
				tsv1 := report["Values"].([]any)[0].(map[string]any)
				assert.Equal(t, 2, int(tsv1["t"].(float64)))
				assert.Equal(t, `TSV{ObservedAtNanoseconds: 1742314713000000000, StreamValue: {"t":0,"v":"111.22"}}`, tsv1["v"].(string))
				// null provider indicated time - uses data received time fallback
				tsv2 := report["Values"].([]any)[1].(map[string]any)
				assert.Equal(t, 2, int(tsv2["t"].(float64)))
				assert.Equal(t, `TSV{ObservedAtNanoseconds: 1742314713050000000, StreamValue: {"t":0,"v":"111.22"}}`, tsv2["v"].(string))
				// missing provider indicated time - uses data received time fallback
				tsv3 := report["Values"].([]any)[2].(map[string]any)
				assert.Equal(t, 2, int(tsv3["t"].(float64)))
				assert.Equal(t, `TSV{ObservedAtNanoseconds: 1742314713050000000, StreamValue: {"t":0,"v":"111.22"}}`, tsv3["v"].(string))
			default:
				t.Fatalf("unexpected report format: %d", req.ReportFormat)
			}

			if len(feedIDs) == 0 {
				break
			}
		}

		if len(feedIDs) > 0 {
			t.Fatalf("expected all feedIDs to be processed, got remaining: %d", len(feedIDs))
		}
	})
}

func TestIntegration_LLO_stress_test_V1(t *testing.T) {
	t.Parallel()

	// logLevel: the log level to use for the nodes
	// setting a more verbose log level increases cpu usage significantly
	// const logLevel = toml.LogLevel(zapcore.DebugLevel)
	const logLevel = toml.LogLevel(zapcore.ErrorLevel)

	// NOTE: Tweak these values to increase or decrease the intensity of the
	// stress test
	//
	// nChannels: the total number of channels
	// nReports: the number of reports to expect per node
	// defaultMinReportInterval: minimum time between report emission (set to 1ns to produce as fast as possible)

	// STRESS TEST PARAMETERS

	// LOW STRESS
	const nChannels = 100
	const nReports = 250
	const defaultMinReportInterval = 5 * time.Millisecond

	// HIGHER STRESS
	// const nChannels = 2000
	// const nReports = 50_000
	// const defaultMinReportInterval = 1 * time.Nanosecond

	// PROTOCOL CONFIGURATION
	ocrConfigOpts := []OCRConfigOption{
		WithOffchainConfig(datastreamsllo.OffchainConfig{
			ProtocolVersion:                     1,
			DefaultMinReportIntervalNanoseconds: uint64(defaultMinReportInterval),
			EnableObservationCompression:        true,
		}),
		func(cfg *OCRConfig) {
			// cfg.DeltaRound = 0 // Go as fast as possible
			cfg.DeltaRound = 50 * time.Millisecond
			cfg.DeltaGrace = 5 * time.Millisecond
			cfg.DeltaCertifiedCommitRequest = 5 * time.Millisecond
		},
	}

	// SETUP

	clientCSAKeys := make([]csakey.KeyV2, nNodes)
	clientPubKeys := make([]ed25519.PublicKey, nNodes)

	const salt = 302

	for i := range nNodes {
		k := big.NewInt(int64(salt + i))
		key := csakey.MustNewV2XXXTestingOnly(k)
		clientCSAKeys[i] = key
		clientPubKeys[i] = key.PublicKey
	}

	steve, backend, configurator, configuratorAddress, _, _, _, _, configStore, configStoreAddress, _, _, _, _ := setupBlockchain(t)
	fromBlock := 1

	// Setup bootstrap
	bootstrapCSAKey := csakey.MustNewV2XXXTestingOnly(big.NewInt(salt - 1))
	bootstrapNodePort := freeport.GetOne(t)
	appBootstrap, bootstrapPeerID, _, bootstrapKb, _ := setupNode(t, bootstrapNodePort, "bootstrap_llo", backend, bootstrapCSAKey, func(c *chainlink.Config) {
		c.Log.Level = ptr(logLevel)
	})
	bootstrapNode := Node{App: appBootstrap, KeyBundle: bootstrapKb}

	t.Run("produces reports properly", func(t *testing.T) {
		packets := make(chan *packet, nReports*nNodes)
		serverKey := csakey.MustNewV2XXXTestingOnly(big.NewInt(salt - 2))
		serverPubKey := serverKey.PublicKey
		srv := NewMercuryServer(t, serverKey, packets)

		serverURL := startMercuryServer(t, srv, clientPubKeys)

		donID := uint32(888333)
		streams := []Stream{ethStream}
		streamMap := make(map[uint32]Stream)
		for _, strm := range streams {
			streamMap[strm.id] = strm
		}

		// Setup oracle nodes
		oracles, nodes := setupNodes(t, nNodes, backend, clientCSAKeys, func(c *chainlink.Config) {
			c.Mercury.Transmitter.Protocol = ptr(config.MercuryTransmitterProtocolGRPC)
			c.Log.Level = ptr(logLevel)
		})

		chainID := testutils.SimulatedChainID
		relayType := "evm"
		relayConfig := fmt.Sprintf(`
chainID = "%s"
fromBlock = %d
lloDonID = %d
lloConfigMode = "bluegreen"
`, chainID, fromBlock, donID)
		addBootstrapJob(t, bootstrapNode, configuratorAddress, "job-3", relayType, relayConfig)

		// Channel definitions
		channelDefinitions := llotypes.ChannelDefinitions{}
		for i := range uint32(nChannels) {
			channelDefinitions[i] = llotypes.ChannelDefinition{
				ReportFormat: llotypes.ReportFormatJSON,
				Streams: []llotypes.Stream{
					{
						StreamID:   ethStreamID,
						Aggregator: llotypes.AggregatorMedian,
					},
				},
			}
		}
		url, sha := newChannelDefinitionsServer(t, channelDefinitions)

		// Set channel definitions
		_, err := configStore.SetChannelDefinitions(steve, donID, url, sha)
		require.NoError(t, err)
		backend.Commit()

		// one working and one broken transmission server
		pluginConfig := fmt.Sprintf(`servers = { "%s" = "%x" }
donID = %d
channelDefinitionsContractAddress = "0x%x"
channelDefinitionsContractFromBlock = %d`, serverURL, serverPubKey, donID, configStoreAddress, fromBlock)
		for i, node := range nodes {
			addLLOJob(
				t,
				node,
				configuratorAddress,
				bootstrapPeerID,
				bootstrapNodePort,
				clientPubKeys[i],
				"feed-1",
				pluginConfig,
				relayType,
				relayConfig,
			)
			addMemoStreamSpecs(t, node, streams)
		}

		// Set config on configurator
		opts := []OCRConfigOption{WithOracles(oracles)}
		opts = append(opts, ocrConfigOpts...)
		blueDigest := setProductionConfig(
			t, donID, steve, backend, configurator, configuratorAddress, nodes, opts...,
		)

		// NOTE: Wait for nReports reports per node
		// transmitter addr => count of reports
		cnts := map[string]int{}
		// transmitter addr => channel ID => reports
		m := map[string]map[uint32][]datastreamsllo.Report{}

		for {
			pckt, err := receiveWithTimeout(t, packets, reportTimeout)
			require.NoError(t, err)
			pr, ok := peer.FromContext(pckt.ctx)
			require.True(t, ok)
			addr := pr.Addr
			req := pckt.req

			assert.Equal(t, uint32(llotypes.ReportFormatJSON), req.ReportFormat)
			_, _, r, _, err := (datastreamsllo.JSONReportCodec{}).UnpackDecode(req.Payload)
			require.NoError(t, err)

			cm, exists := m[addr.String()]
			if !exists {
				cm = make(map[uint32][]datastreamsllo.Report)
				m[addr.String()] = cm
			}
			cm[r.ChannelID] = append(cm[r.ChannelID], r)

			cnts[addr.String()]++
			finished := 0
			for _, cnt := range cnts {
				if cnt >= nReports {
					finished++
				}
			}
			if finished >= nNodes {
				for _, node := range nodes {
					//nolint:testifylint // need assert to ensure we don't leak nodes
					assert.NoError(t, node.App.Stop())
				}
				defer close(packets)
				break
			}
		}

		// Transmissions can occur out of order when we go very fast, so sort by seqNr
		for _, cm := range m {
			for _, rs := range cm {
				sort.Slice(rs, func(i, j int) bool {
					return rs[i].SeqNr < rs[j].SeqNr
				})
			}
		}

		// Check reports
		for addr, cm := range m {
			spacings := []uint64{}
			for _, rs := range cm {
				var prevObsTsNanos uint64
				for i, r := range rs {
					assert.Equal(t, blueDigest, r.ConfigDigest)
					assert.False(t, r.Specimen)
					assert.Len(t, r.Values, 1)
					assert.Equal(t, "2976.39", r.Values[0].(*datastreamsllo.Decimal).String())

					if i > 0 {
						if rs[i-1].SeqNr+1 != r.SeqNr {
							// t.Logf("gap in SeqNr at index %d; %d!=%d: len(rs)=%d", i, rs[i-1].SeqNr, r.SeqNr, len(rs))
							// We actually expect a transmission every round; if there's a gap in seqNr it means that the transmissions were likely cut off due to the app being shut down. We are probably at the end of the usable reports list so just assume completion here.
							break
						}

						// No gaps
						require.Equal(t, prevObsTsNanos, r.ValidAfterNanoseconds, "gap in reports for transmitter %s at index %d; %d!=%d: prevReport=%s, thisReport=%s", addr, i, prevObsTsNanos, r.ValidAfterNanoseconds, mustMarshalJSON(rs[i-1]), mustMarshalJSON(r))
						// Timestamps are sane
						require.GreaterOrEqual(t, r.ObservationTimestampNanoseconds, r.ValidAfterNanoseconds, "observation timestamp is before valid after timestamp for transmitter %s at index %d: report=%s", addr, i, mustMarshalJSON(r))
						// Reports are separated by at least the minimum interval
						require.GreaterOrEqual(t, r.ObservationTimestampNanoseconds-uint64(defaultMinReportInterval), prevObsTsNanos, "reports are too close together for transmitter %s at index %d: prevReport=%s, thisReport=%s; expected at least %d nanoseconds of distance", addr, i, mustMarshalJSON(rs[i-1]), mustMarshalJSON(r), defaultMinReportInterval)

						spacings = append(spacings, r.ObservationTimestampNanoseconds-prevObsTsNanos)
					}
					prevObsTsNanos = r.ObservationTimestampNanoseconds
				}
			}
			avgSpacing := uint64(0)
			for _, spacing := range spacings {
				avgSpacing += spacing
			}
			avgSpacing /= uint64(len(spacings))
			t.Logf("transmitter %s: average spacing between reports: %d nanoseconds (%f seconds)", addr, avgSpacing, float64(avgSpacing)/1e9)
		}
	})
}

func TestIntegration_LLO_transmit_errors(t *testing.T) {
	t.Parallel()

	// logLevel: the log level to use for the nodes
	// setting a more verbose log level increases cpu usage significantly
	const logLevel = toml.LogLevel(zapcore.ErrorLevel)
	// const logLevel = toml.LogLevel(zapcore.ErrorLevel)

	// NOTE: Tweak these values to increase or decrease the intensity of the
	// stress test
	//
	// nChannels: the total number of channels
	// maxQueueSize: the maximum size of the transmit queue
	// nReports: the number of reports to expect per node

	// LESS STRESSFUL
	const nChannels = 200
	const maxQueueSize = 10
	const nReports = 1_000

	// MORE STRESSFUL
	// const nChannels = 2000
	// const maxQueueSize = 4_000
	// const nReports = 10_000

	// PROTOCOL CONFIGURATION
	// TODO: test both
	offchainConfig := datastreamsllo.OffchainConfig{
		ProtocolVersion:                     1,
		DefaultMinReportIntervalNanoseconds: uint64(50 * time.Millisecond),
	}

	clientCSAKeys := make([]csakey.KeyV2, nNodes)
	clientPubKeys := make([]ed25519.PublicKey, nNodes)

	const salt = 301

	for i := range nNodes {
		k := big.NewInt(int64(salt + i))
		key := csakey.MustNewV2XXXTestingOnly(k)
		clientCSAKeys[i] = key
		clientPubKeys[i] = key.PublicKey
	}

	steve, backend, configurator, configuratorAddress, _, _, _, _, configStore, configStoreAddress, _, _, _, _ := setupBlockchain(t)
	fromBlock := 1

	// Setup bootstrap
	bootstrapCSAKey := csakey.MustNewV2XXXTestingOnly(big.NewInt(salt - 1))
	bootstrapNodePort := freeport.GetOne(t)
	appBootstrap, bootstrapPeerID, _, bootstrapKb, _ := setupNode(t, bootstrapNodePort, "bootstrap_llo", backend, bootstrapCSAKey, func(c *chainlink.Config) {
		c.Log.Level = ptr(logLevel)
	})
	bootstrapNode := Node{App: appBootstrap, KeyBundle: bootstrapKb}

	t.Run("transmit queue does not grow unbounded", func(t *testing.T) {
		packets := make(chan *packet, 100000)
		serverKey := csakey.MustNewV2XXXTestingOnly(big.NewInt(salt - 2))
		serverPubKey := serverKey.PublicKey
		srv := NewMercuryServer(t, serverKey, packets)

		serverURL := startMercuryServer(t, srv, clientPubKeys)

		donID := uint32(888333)
		streams := []Stream{ethStream, linkStream}
		streamMap := make(map[uint32]Stream)
		for _, strm := range streams {
			streamMap[strm.id] = strm
		}

		// Setup oracle nodes
		oracles, nodes := setupNodes(t, nNodes, backend, clientCSAKeys, func(c *chainlink.Config) {
			c.Mercury.Transmitter.Protocol = ptr(config.MercuryTransmitterProtocolGRPC)
			c.Mercury.Transmitter.TransmitQueueMaxSize = ptr(uint32(maxQueueSize)) // Test queue overflow
			c.Log.Level = ptr(logLevel)
		})

		chainID := testutils.SimulatedChainID
		relayType := "evm"
		relayConfig := fmt.Sprintf(`
chainID = "%s"
fromBlock = %d
lloDonID = %d
lloConfigMode = "bluegreen"
`, chainID, fromBlock, donID)
		addBootstrapJob(t, bootstrapNode, configuratorAddress, "job-3", relayType, relayConfig)

		// Channel definitions
		channelDefinitions := llotypes.ChannelDefinitions{}
		for i := range uint32(nChannels) {
			channelDefinitions[i] = llotypes.ChannelDefinition{
				ReportFormat: llotypes.ReportFormatJSON,
				Streams: []llotypes.Stream{
					{
						StreamID:   ethStreamID,
						Aggregator: llotypes.AggregatorMedian,
					},
				},
			}
		}
		url, sha := newChannelDefinitionsServer(t, channelDefinitions)

		// Set channel definitions
		_, err := configStore.SetChannelDefinitions(steve, donID, url, sha)
		require.NoError(t, err)
		backend.Commit()

		// one working and one broken transmission server
		pluginConfig := fmt.Sprintf(`servers = { "%s" = "%x", "example.invalid" = "%x" }
donID = %d
channelDefinitionsContractAddress = "0x%x"
channelDefinitionsContractFromBlock = %d`, serverURL, serverPubKey, serverPubKey, donID, configStoreAddress, fromBlock)
		addOCRJobsEVMPremiumLegacy(t, streams, serverPubKey, serverURL, configuratorAddress, bootstrapPeerID, bootstrapNodePort, nodes, configStoreAddress, clientPubKeys, pluginConfig, relayType, relayConfig)

		var blueDigest ocr2types.ConfigDigest

		{
			// Set config on configurator
			blueDigest = setProductionConfig(
				t, donID, steve, backend, configurator, configuratorAddress, nodes, WithOracles(oracles), WithOffchainConfig(offchainConfig),
			)

			// NOTE: Wait for nReports reports
			// count of packets received keyed by transmitter IP
			m := map[string]int{}
			for {
				pckt, err := receiveWithTimeout(t, packets, reportTimeout)
				require.NoError(t, err)
				pr, ok := peer.FromContext(pckt.ctx)
				require.True(t, ok)
				addr := pr.Addr
				req := pckt.req

				assert.Equal(t, uint32(llotypes.ReportFormatJSON), req.ReportFormat)
				_, _, r, _, err := (datastreamsllo.JSONReportCodec{}).UnpackDecode(req.Payload)
				require.NoError(t, err)

				assert.Equal(t, blueDigest, r.ConfigDigest)
				assert.False(t, r.Specimen)
				assert.Len(t, r.Values, 1)
				assert.Equal(t, "2976.39", r.Values[0].(*datastreamsllo.Decimal).String())

				m[addr.String()]++
				finished := 0
				for _, cnt := range m {
					if cnt >= nReports {
						finished++
					}
				}
				if finished == 4 {
					break
				}
			}
		}

		// Shut all nodes down
		for i, node := range nodes {
			require.NoError(t, node.App.Stop())
			// Ensure that the transmit queue was limited
			db := node.App.GetDB()
			cnt := 0

			// The failing server
			err := db.GetContext(t.Context(), &cnt, "SELECT count(*) FROM llo_mercury_transmit_queue WHERE server_url = 'example.invalid'")
			require.NoError(t, err)
			assert.LessOrEqual(t, cnt, maxQueueSize, "persisted transmit queue size too large for node %d for failing server", i)

			// The succeeding server
			err = db.GetContext(t.Context(), &cnt, "SELECT count(*) FROM llo_mercury_transmit_queue WHERE server_url = $1", serverURL)
			require.NoError(t, err)
			assert.LessOrEqual(t, cnt, maxQueueSize, "persisted transmit queue size too large for node %d for succeeding server", i)
		}
	})
}

func TestIntegration_LLO_blue_green_lifecycle(t *testing.T) {
	t.Parallel()

	// starting offchainConfig, the test will handle
	// blue green for ProtocolVersion and EnableObservationCompression changes
	offchainConfig := datastreamsllo.OffchainConfig{
		ProtocolVersion:                     0,
		DefaultMinReportIntervalNanoseconds: 0,
		EnableObservationCompression:        false}
	testIntegrationLLOBlueGreenLifecycle(t, offchainConfig)
}

func testIntegrationLLOBlueGreenLifecycle(t *testing.T, offchainConfig datastreamsllo.OffchainConfig) {
	clientCSAKeys := make([]csakey.KeyV2, nNodes)
	clientPubKeys := make([]ed25519.PublicKey, nNodes)

	const salt = 300

	for i := range nNodes {
		k := big.NewInt(int64(salt + i))
		key := csakey.MustNewV2XXXTestingOnly(k)
		clientCSAKeys[i] = key
		clientPubKeys[i] = key.PublicKey
	}

	steve, backend, configurator, configuratorAddress, _, _, _, _, configStore, configStoreAddress, _, _, _, _ := setupBlockchain(t)
	fromBlock := 1

	// Setup bootstrap
	bootstrapCSAKey := csakey.MustNewV2XXXTestingOnly(big.NewInt(salt - 1))
	bootstrapNodePort := freeport.GetOne(t)
	appBootstrap, bootstrapPeerID, _, bootstrapKb, _ := setupNode(t, bootstrapNodePort, "bootstrap_llo", backend, bootstrapCSAKey, nil)
	bootstrapNode := Node{App: appBootstrap, KeyBundle: bootstrapKb}

	t.Run("Blue/Green lifecycle (using JSON report format)", func(t *testing.T) {
		packetCh := make(chan *packet, 100000)
		serverKey := csakey.MustNewV2XXXTestingOnly(big.NewInt(salt - 2))
		serverPubKey := serverKey.PublicKey
		srv := NewMercuryServer(t, serverKey, packetCh)

		serverURL := startMercuryServer(t, srv, clientPubKeys)

		donID := uint32(888333)
		streams := []Stream{ethStream, linkStream}
		streamMap := make(map[uint32]Stream)
		for _, strm := range streams {
			streamMap[strm.id] = strm
		}

		// Setup oracle nodes
		oracles, nodes := setupNodes(t, nNodes, backend, clientCSAKeys, func(c *chainlink.Config) {
			c.Mercury.Transmitter.Protocol = ptr(config.MercuryTransmitterProtocolGRPC)
		})

		chainID := testutils.SimulatedChainID
		relayType := "evm"
		relayConfig := fmt.Sprintf(`
chainID = "%s"
fromBlock = %d
lloDonID = %d
lloConfigMode = "bluegreen"
`, chainID, fromBlock, donID)
		addBootstrapJob(t, bootstrapNode, configuratorAddress, "job-3", relayType, relayConfig)

		// Channel definitions
		channelDefinitions := llotypes.ChannelDefinitions{
			1: {
				ReportFormat: llotypes.ReportFormatJSON,
				Streams: []llotypes.Stream{
					{
						StreamID:   ethStreamID,
						Aggregator: llotypes.AggregatorMedian,
					},
				},
			},
		}
		url, sha := newChannelDefinitionsServer(t, channelDefinitions)

		// Set channel definitions
		_, err := configStore.SetChannelDefinitions(steve, donID, url, sha)
		require.NoError(t, err)
		backend.Commit()

		pluginConfig := fmt.Sprintf(`servers = { "%s" = "%x" }
donID = %d
channelDefinitionsContractAddress = "0x%x"
channelDefinitionsContractFromBlock = %d`, serverURL, serverPubKey, donID, configStoreAddress, fromBlock)
		addOCRJobsEVMPremiumLegacy(t, streams, serverPubKey, serverURL, configuratorAddress, bootstrapPeerID, bootstrapNodePort, nodes, configStoreAddress, clientPubKeys, pluginConfig, relayType, relayConfig)

		var blueDigest ocr2types.ConfigDigest
		var greenDigest ocr2types.ConfigDigest

		allReports := make(map[types.ConfigDigest][]datastreamsllo.Report)
		// start off with blue=production, green=staging (specimen reports)
		{
			// Set config on configurator
			blueDigest = setProductionConfig(
				t, donID, steve, backend, configurator, configuratorAddress, nodes, WithOracles(oracles), WithOffchainConfig(offchainConfig),
			)

			// NOTE: Wait until blue produces a report

			for {
				pckt, err := receiveWithTimeout(t, packetCh, reportTimeout)
				require.NoError(t, err)
				req := pckt.req
				assert.Equal(t, uint32(llotypes.ReportFormatJSON), req.ReportFormat)
				_, _, r, _, err := (datastreamsllo.JSONReportCodec{}).UnpackDecode(req.Payload)
				require.NoError(t, err)

				allReports[r.ConfigDigest] = append(allReports[r.ConfigDigest], r)

				assert.Equal(t, blueDigest, r.ConfigDigest)
				assert.False(t, r.Specimen)
				assert.Len(t, r.Values, 1)
				assert.Equal(t, "2976.39", r.Values[0].(*datastreamsllo.Decimal).String())
				break
			}
		}
		// setStagingConfig does not affect production
		{
			offchainConfig.EnableObservationCompression = true
			greenDigest = setStagingConfig(
				t, donID, steve, backend, configurator, configuratorAddress, nodes, WithPredecessorConfigDigest(blueDigest), WithOracles(oracles), WithOffchainConfig(offchainConfig),
			)

			// NOTE: Wait until green produces the first "specimen" report

			for {
				pckt, err := receiveWithTimeout(t, packetCh, reportTimeout)
				require.NoError(t, err)
				req := pckt.req
				assert.Equal(t, uint32(llotypes.ReportFormatJSON), req.ReportFormat)
				_, _, r, _, err := (datastreamsllo.JSONReportCodec{}).UnpackDecode(req.Payload)
				require.NoError(t, err)

				allReports[r.ConfigDigest] = append(allReports[r.ConfigDigest], r)
				if r.Specimen {
					assert.Len(t, r.Values, 1)
					assert.Equal(t, "2976.39", r.Values[0].(*datastreamsllo.Decimal).String())

					assert.Equal(t, greenDigest, r.ConfigDigest)
					break
				}
				assert.Equal(t, blueDigest, r.ConfigDigest)
			}
		}
		// promoteStagingConfig flow has clean and gapless hand off from old production to newly promoted staging instance, leaving old production instance in 'retired' state
		{
			promoteStagingConfig(t, donID, steve, backend, configurator, configuratorAddress, false)

			// NOTE: Wait for first non-specimen report for the newly promoted (green) instance

			for {
				pckt, err := receiveWithTimeout(t, packetCh, reportTimeout)
				require.NoError(t, err)
				req := pckt.req
				assert.Equal(t, uint32(llotypes.ReportFormatJSON), req.ReportFormat)
				_, _, r, _, err := (datastreamsllo.JSONReportCodec{}).UnpackDecode(req.Payload)
				require.NoError(t, err)

				allReports[r.ConfigDigest] = append(allReports[r.ConfigDigest], r)

				if !r.Specimen && r.ConfigDigest == greenDigest {
					break
				}
			}

			initialPromotedGreenReport := allReports[greenDigest][len(allReports[greenDigest])-1]
			finalBlueReport := allReports[blueDigest][len(allReports[blueDigest])-1]

			for _, digest := range []ocr2types.ConfigDigest{blueDigest, greenDigest} {
				// Transmissions are not guaranteed to be in order
				sort.Slice(allReports[digest], func(i, j int) bool {
					return allReports[digest][i].SeqNr < allReports[digest][j].SeqNr
				})
				seenSeqNr := uint64(0)
				highestObsTsNanos := uint64(0)
				highestValidAfterNanos := uint64(0)
				for i := 0; i < len(allReports[digest]); i++ {
					r := allReports[digest][i]
					switch digest {
					case greenDigest:
						if i == len(allReports[digest])-1 {
							assert.False(t, r.Specimen)
						} else {
							assert.True(t, r.Specimen)
						}
					case blueDigest:
						assert.False(t, r.Specimen)
					}
					if r.SeqNr > seenSeqNr {
						// skip first one
						if highestObsTsNanos > 0 {
							if digest == greenDigest && i == len(allReports[digest])-1 {
								// NOTE: This actually CHANGES on the staging
								// handover and can go backwards - the gapless
								// handover test is handled below
								break
							}
							if offchainConfig.ProtocolVersion == 0 {
								// validAfter is always truncated to 1s in v0
								// IMPORTANT: gapless handovers in v0 ONLY supported at 1s resolution!!
								assert.Equal(t, highestObsTsNanos/1e9*1e9, r.ValidAfterNanoseconds/1e9*1e9, "%d: (n-1)ObservationsTimestampSeconds->(n)ValidAfterNanoseconds should be gapless to within 1s resolution, got: %d vs %d", i, highestObsTsNanos, r.ValidAfterNanoseconds)
							} else {
								assert.Equal(t, highestObsTsNanos, r.ValidAfterNanoseconds, "%d: (n-1)ObservationsTimestampSeconds->(n)ValidAfterNanoseconds should be gapless, got: %d vs %d", i, highestObsTsNanos, r.ValidAfterNanoseconds)
							}
							assert.Greater(t, r.ObservationTimestampNanoseconds, highestObsTsNanos, "%d: overlapping/duplicate report ObservationTimestampNanoseconds, got: %d vs %d", i, r.ObservationTimestampNanoseconds, highestObsTsNanos)
							assert.Greater(t, r.ValidAfterNanoseconds, highestValidAfterNanos, "%d: overlapping/duplicate report ValidAfterNanoseconds, got: %d vs %d", i, r.ValidAfterNanoseconds, highestValidAfterNanos)
							assert.Less(t, r.ValidAfterNanoseconds, r.ObservationTimestampNanoseconds)
						}
						seenSeqNr = r.SeqNr
						highestObsTsNanos = r.ObservationTimestampNanoseconds
						highestValidAfterNanos = r.ValidAfterNanoseconds
					}
				}
			}

			// Gapless handover
			assert.Less(t, finalBlueReport.ValidAfterNanoseconds, finalBlueReport.ObservationTimestampNanoseconds)

			if offchainConfig.ProtocolVersion == 0 {
				// validAfter is always truncated to 1s in v0
				// IMPORTANT: gapless handovers in v0 ONLY supported at 1s resolution!!
				assert.Equal(t, finalBlueReport.ObservationTimestampNanoseconds/1e9*1e9, initialPromotedGreenReport.ValidAfterNanoseconds/1e9*1e9)
			} else {
				assert.Equal(t, finalBlueReport.ObservationTimestampNanoseconds, initialPromotedGreenReport.ValidAfterNanoseconds)
			}

			assert.Less(t, initialPromotedGreenReport.ValidAfterNanoseconds, initialPromotedGreenReport.ObservationTimestampNanoseconds)
		}
		// retired instance does not produce reports
		{
			// NOTE: Wait for five "green" reports to be produced and assert no "blue" reports

			i := 0
			for {
				pckt, err := receiveWithTimeout(t, packetCh, reportTimeout)
				require.NoError(t, err)
				req := pckt.req
				i++
				if i == 5 {
					break
				}
				assert.Equal(t, uint32(llotypes.ReportFormatJSON), req.ReportFormat)
				_, _, r, _, err := (datastreamsllo.JSONReportCodec{}).UnpackDecode(req.Payload)
				require.NoError(t, err)

				allReports[r.ConfigDigest] = append(allReports[r.ConfigDigest], r)
				assert.False(t, r.Specimen)
				assert.Equal(t, greenDigest, r.ConfigDigest)
			}
		}
		// setStagingConfig replaces 'retired' instance with new config and starts producing specimen reports again
		{
			offchainConfig.ProtocolVersion = 1
			offchainConfig.DefaultMinReportIntervalNanoseconds = 1
			blueDigest = setStagingConfig(
				t, donID, steve, backend, configurator, configuratorAddress, nodes, WithPredecessorConfigDigest(greenDigest), WithOracles(oracles), WithOffchainConfig(offchainConfig),
			)

			// NOTE: Wait until blue produces the first "specimen" report

			for {
				pckt, err := receiveWithTimeout(t, packetCh, reportTimeout)
				require.NoError(t, err)
				req := pckt.req
				assert.Equal(t, uint32(llotypes.ReportFormatJSON), req.ReportFormat)
				_, _, r, _, err := (datastreamsllo.JSONReportCodec{}).UnpackDecode(req.Payload)
				require.NoError(t, err)

				allReports[r.ConfigDigest] = append(allReports[r.ConfigDigest], r)
				if r.Specimen {
					assert.Equal(t, blueDigest, r.ConfigDigest)
					break
				}
				assert.Equal(t, greenDigest, r.ConfigDigest)
			}
		}
		// promoteStagingConfig swaps the instances again
		{
			// TODO: Check that once an instance enters 'retired' state, it
			// doesn't produce reports or bother making observations
			promoteStagingConfig(t, donID, steve, backend, configurator, configuratorAddress, true)

			// NOTE: Wait for first non-specimen report for the newly promoted (blue) instance

			for {
				pckt, err := receiveWithTimeout(t, packetCh, reportTimeout)
				require.NoError(t, err)
				req := pckt.req
				assert.Equal(t, uint32(llotypes.ReportFormatJSON), req.ReportFormat)
				_, _, r, _, err := (datastreamsllo.JSONReportCodec{}).UnpackDecode(req.Payload)
				require.NoError(t, err)

				allReports[r.ConfigDigest] = append(allReports[r.ConfigDigest], r)

				if !r.Specimen && r.ConfigDigest == blueDigest {
					break
				}
			}

			initialPromotedBlueReport := allReports[blueDigest][len(allReports[blueDigest])-1]
			finalGreenReport := allReports[greenDigest][len(allReports[greenDigest])-1]
			assert.Less(t, finalGreenReport.ValidAfterNanoseconds, finalGreenReport.ObservationTimestampNanoseconds)
			assert.Less(t, initialPromotedBlueReport.ValidAfterNanoseconds, initialPromotedBlueReport.ObservationTimestampNanoseconds)
		}
		// adding a new channel definition is picked up on the fly
		{
			channelDefinitions[2] = llotypes.ChannelDefinition{
				ReportFormat: llotypes.ReportFormatJSON,
				Streams: []llotypes.Stream{
					{
						StreamID:   linkStreamID,
						Aggregator: llotypes.AggregatorMedian,
					},
				},
			}

			url, sha := newChannelDefinitionsServer(t, channelDefinitions)

			// Set channel definitions
			_, err := configStore.SetChannelDefinitions(steve, donID, url, sha)
			require.NoError(t, err)
			backend.Commit()

			// NOTE: Wait until the first report for the new channel definition is produced

			for {
				pckt, err := receiveWithTimeout(t, packetCh, reportTimeout)
				require.NoError(t, err)
				req := pckt.req
				assert.Equal(t, uint32(llotypes.ReportFormatJSON), req.ReportFormat)
				_, _, r, _, err := (datastreamsllo.JSONReportCodec{}).UnpackDecode(req.Payload)
				require.NoError(t, err)

				allReports[r.ConfigDigest] = append(allReports[r.ConfigDigest], r)

				// Green is retired, it shouldn't be producing anything
				assert.Equal(t, blueDigest, r.ConfigDigest)
				assert.False(t, r.Specimen)

				if r.ChannelID == 2 {
					assert.Len(t, r.Values, 1)
					assert.Equal(t, "13.25", r.Values[0].(*datastreamsllo.Decimal).String())
					break
				}
				assert.Len(t, r.Values, 1)
				assert.Equal(t, "2976.39", r.Values[0].(*datastreamsllo.Decimal).String())
			}
		}
		t.Run("deleting the jobs turns off oracles and cleans up resources", func(t *testing.T) {
			t.Skip("TODO - MERC-3524")
		})
		t.Run("adding new jobs again picks up the correct configs", func(t *testing.T) {
			t.Skip("TODO - MERC-3524")
		})
	})
}

func TestIntegration_LLO_channel_merging_owners_adders(t *testing.T) {
	t.Parallel()

	offchainConfig := datastreamsllo.OffchainConfig{
		ProtocolVersion:                     1,
		DefaultMinReportIntervalNanoseconds: uint64(1 * time.Second),
		EnableObservationCompression:        true,
	}

	clientCSAKeys := make([]csakey.KeyV2, nNodes)
	clientPubKeys := make([]ed25519.PublicKey, nNodes)

	const salt = 400

	for i := range nNodes {
		k := big.NewInt(int64(salt + i))
		key := csakey.MustNewV2XXXTestingOnly(k)
		clientCSAKeys[i] = key
		clientPubKeys[i] = key.PublicKey
	}

	// Create adder accounts before creating backend
	adder1 := evmtestutils.MustNewSimTransactor(t)
	adder2 := evmtestutils.MustNewSimTransactor(t)

	steve, backend, configurator, configuratorAddress, _, _, _, _, configStore, configStoreAddress, _, _, _, _ := setupBlockchain(t, adder1, adder2)
	fromBlock := 1

	// Setup bootstrap
	bootstrapCSAKey := csakey.MustNewV2XXXTestingOnly(big.NewInt(salt - 1))
	bootstrapNodePort := freeport.GetOne(t)
	appBootstrap, bootstrapPeerID, _, bootstrapKb, _ := setupNode(t, bootstrapNodePort, "bootstrap_llo", backend, bootstrapCSAKey, nil)
	bootstrapNode := Node{App: appBootstrap, KeyBundle: bootstrapKb}

	t.Run("Channel merging lifecycle with owners and adders", func(t *testing.T) {
		packetCh := make(chan *packet, 100000)
		serverKey := csakey.MustNewV2XXXTestingOnly(big.NewInt(salt - 2))
		serverPubKey := serverKey.PublicKey
		srv := NewMercuryServer(t, serverKey, packetCh)

		serverURL := startMercuryServer(t, srv, clientPubKeys)

		donID := uint32(999888)
		streams := []Stream{ethStream, linkStream}
		streamMap := make(map[uint32]Stream)
		for _, strm := range streams {
			streamMap[strm.id] = strm
		}

		// Setup oracle nodes
		oracles, nodes := setupNodes(t, nNodes, backend, clientCSAKeys, func(c *chainlink.Config) {
			c.Mercury.Transmitter.Protocol = ptr(config.MercuryTransmitterProtocolGRPC)
		})

		chainID := testutils.SimulatedChainID
		relayType := "evm"
		relayConfig := fmt.Sprintf(`
chainID = "%s"
fromBlock = %d
lloDonID = %d
lloConfigMode = "bluegreen"
`, chainID, fromBlock, donID)
		addBootstrapJob(t, bootstrapNode, configuratorAddress, "job-channel-merge", relayType, relayConfig)

		// Configure adders on the contract
		// Adder IDs start from 1000
		adder1ID := uint32(1001)
		adder2ID := uint32(1002)

		require.NoError(t, utils.JustError(configStore.SetChannelAdderAddress(steve, adder1ID, adder1.From)))
		backend.Commit()
		require.NoError(t, utils.JustError(configStore.SetChannelAdderAddress(steve, adder2ID, adder2.From)))
		backend.Commit()

		// Enable adders
		require.NoError(t, utils.JustError(configStore.SetChannelAdder(steve, donID, adder1ID, true)))
		backend.Commit()
		require.NoError(t, utils.JustError(configStore.SetChannelAdder(steve, donID, adder2ID, true)))
		backend.Commit()

		pluginConfig := fmt.Sprintf(`servers = { "%s" = "%x" }
donID = %d
channelDefinitionsContractAddress = "0x%x"
channelDefinitionsContractFromBlock = %d`, serverURL, serverPubKey, donID, configStoreAddress, fromBlock)

		// Add stream specs and LLO jobs to all nodes
		for i, node := range nodes {
			addMemoStreamSpecs(t, node, streams)
			addLLOJob(
				t,
				node,
				configuratorAddress,
				bootstrapPeerID,
				bootstrapNodePort,
				clientPubKeys[i],
				"channel-merge-test",
				pluginConfig,
				relayType,
				relayConfig,
			)
		}

		// Set initial OCR config
		digest := setProductionConfig(
			t, donID, steve, backend, configurator, configuratorAddress, nodes, WithOracles(oracles), WithOffchainConfig(offchainConfig),
		)

		// Track reports by channel ID
		reportsByChannel := make(map[uint32][]datastreamsllo.Report)
		lastReportTimeByChannel := make(map[uint32]time.Time)

		// Helper function to wait for reports from specific channels
		waitForReportsFromChannels := func(t *testing.T, expectedChannels map[uint32]bool, timeout time.Duration) {
			seenChannels := make(map[uint32]bool)
			deadline := time.Now().Add(timeout)
			for time.Now().Before(deadline) && len(seenChannels) < len(expectedChannels) {
				pckt, err := receiveWithTimeout(t, packetCh, 2*time.Second)
				if err != nil {
					// If we're getting timeouts and haven't seen any channels yet, continue waiting
					if len(seenChannels) == 0 {
						continue
					}
					// If we've seen some channels but not all, continue waiting
					continue
				}
				req := pckt.req
				if req.ReportFormat != uint32(llotypes.ReportFormatJSON) {
					continue
				}
				_, _, r, _, err := (datastreamsllo.JSONReportCodec{}).UnpackDecode(req.Payload)
				if err != nil {
					continue
				}

				if expectedChannels[r.ChannelID] {
					reportsByChannel[r.ChannelID] = append(reportsByChannel[r.ChannelID], r)
					lastReportTimeByChannel[r.ChannelID] = time.Now()
					seenChannels[r.ChannelID] = true
				}
			}
			require.Len(t, seenChannels, len(expectedChannels), "expected reports from all channels: got %v, expected %v", seenChannels, expectedChannels)
		}

		// Scenario 1: Owner adds initial channels
		t.Run("Owner adds initial channels", func(t *testing.T) {
			channelDefinitions := llotypes.ChannelDefinitions{
				1: {
					ReportFormat: llotypes.ReportFormatJSON,
					Streams: []llotypes.Stream{
						{
							StreamID:   ethStreamID,
							Aggregator: llotypes.AggregatorMedian,
						},
					},
				},
				2: {
					ReportFormat: llotypes.ReportFormatJSON,
					Streams: []llotypes.Stream{
						{
							StreamID:   ethStreamID,
							Aggregator: llotypes.AggregatorMedian,
						},
					},
				},
				3: {
					ReportFormat: llotypes.ReportFormatJSON,
					Streams: []llotypes.Stream{
						{
							StreamID:   linkStreamID,
							Aggregator: llotypes.AggregatorMedian,
						},
					},
				},
			}
			url, sha := newChannelDefinitionsServer(t, channelDefinitions)

			// Set channel definitions
			_, err := configStore.SetChannelDefinitions(steve, donID, url, sha)
			require.NoError(t, err)
			backend.Commit()

			// Wait for channel definitions to be processed (give time for log polling and fetching)
			time.Sleep(3 * time.Second)

			// Wait for reports from all owner channels
			expectedChannels := map[uint32]bool{1: true, 2: true, 3: true}
			waitForReportsFromChannels(t, expectedChannels, reportTimeout)

			// Verify reports were generated
			require.NotEmpty(t, reportsByChannel[1], "channel 1 should have reports")
			require.NotEmpty(t, reportsByChannel[2], "channel 2 should have reports")
			require.NotEmpty(t, reportsByChannel[3], "channel 3 should have reports")

			// Verify report content
			for channelID := range expectedChannels {
				report := reportsByChannel[channelID][0]
				assert.Equal(t, digest, report.ConfigDigest)
				assert.False(t, report.Specimen)
				if channelID == 3 {
					assert.Equal(t, "13.25", report.Values[0].(*datastreamsllo.Decimal).String())
				} else {
					assert.Equal(t, "2976.39", report.Values[0].(*datastreamsllo.Decimal).String())
				}
			}
		})

		// Scenario 2: Adders add new channels
		t.Run("Adders add new channels", func(t *testing.T) {
			// Adder1 adds channels
			adder1Definitions := llotypes.ChannelDefinitions{
				10: {
					ReportFormat: llotypes.ReportFormatJSON,
					Streams: []llotypes.Stream{
						{
							StreamID:   ethStreamID,
							Aggregator: llotypes.AggregatorMedian,
						},
					},
					Source:    adder1ID,
					Tombstone: false,
				},
				11: {
					ReportFormat: llotypes.ReportFormatJSON,
					Streams: []llotypes.Stream{
						{
							StreamID:   linkStreamID,
							Aggregator: llotypes.AggregatorMedian,
						},
					},
					Source:    adder1ID,
					Tombstone: false,
				},
			}

			adder1DefinitionsJSON, err := json.MarshalIndent(adder1Definitions, "", "  ")
			require.NoError(t, err)
			adder1DefinitionsSHA := sha3.Sum256(adder1DefinitionsJSON)

			adder1Server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
				_, errWrite := w.Write(adder1DefinitionsJSON)
				if errWrite != nil {
					w.WriteHeader(http.StatusInternalServerError)
					return
				}
				w.Header().Set("Content-Type", "application/json")
				w.WriteHeader(http.StatusOK)
			}))
			t.Cleanup(adder1Server.Close)

			_, err = configStore.AddChannelDefinitions(adder1, donID, adder1ID, adder1Server.URL, adder1DefinitionsSHA)
			require.NoError(t, err)
			backend.Commit()

			// Adder2 adds channels
			adder2Definitions := llotypes.ChannelDefinitions{
				20: {
					ReportFormat: llotypes.ReportFormatJSON,
					Streams: []llotypes.Stream{
						{
							StreamID:   ethStreamID,
							Aggregator: llotypes.AggregatorMedian,
						},
					},
				},
				21: {
					ReportFormat: llotypes.ReportFormatJSON,
					Streams: []llotypes.Stream{
						{
							StreamID:   linkStreamID,
							Aggregator: llotypes.AggregatorMedian,
						},
					},
				},
			}

			adder2DefinitionsJSON, err := json.MarshalIndent(adder2Definitions, "", "  ")
			require.NoError(t, err)
			adder2DefinitionsSHA := sha3.Sum256(adder2DefinitionsJSON)

			adder2Server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
				_, errWrite := w.Write(adder2DefinitionsJSON)
				if errWrite != nil {
					w.WriteHeader(http.StatusInternalServerError)
					return
				}
				w.Header().Set("Content-Type", "application/json")
				w.WriteHeader(http.StatusOK)
			}))
			t.Cleanup(adder2Server.Close)

			_, err = configStore.AddChannelDefinitions(adder2, donID, adder2ID, adder2Server.URL, adder2DefinitionsSHA)
			require.NoError(t, err)
			backend.Commit()

			// Wait for channel definitions to be processed (give time for log polling and fetching)
			time.Sleep(3 * time.Second)

			// Wait for reports from all channels (owner + adders)
			expectedChannels := map[uint32]bool{1: true, 2: true, 3: true, 10: true, 11: true, 20: true, 21: true}
			waitForReportsFromChannels(t, expectedChannels, reportTimeout)

			// Verify all channels have reports
			for channelID := range expectedChannels {
				require.NotEmpty(t, reportsByChannel[channelID], "channel %d should have reports", channelID)
			}
		})

		// Scenario 3: Owner tombstone some channels
		t.Run("Owner tombstone channels", func(t *testing.T) {
			// Owner updates definitions, add tombstone to channel 2 and 21
			channelDefinitions := llotypes.ChannelDefinitions{
				1: {
					ReportFormat: llotypes.ReportFormatJSON,
					Streams: []llotypes.Stream{
						{
							StreamID:   ethStreamID,
							Aggregator: llotypes.AggregatorMedian,
						},
					},
				},
				2: {
					ReportFormat: llotypes.ReportFormatJSON,
					Tombstone:    true,
					Streams: []llotypes.Stream{
						{
							StreamID:   ethStreamID,
							Aggregator: llotypes.AggregatorMedian,
						},
					},
				},
				3: {
					ReportFormat: llotypes.ReportFormatJSON,
					Streams: []llotypes.Stream{
						{
							StreamID:   linkStreamID,
							Aggregator: llotypes.AggregatorMedian,
						},
					},
				},
				21: {
					ReportFormat: llotypes.ReportFormatJSON,
					Tombstone:    true,
					Streams: []llotypes.Stream{
						{
							StreamID:   linkStreamID,
							Aggregator: llotypes.AggregatorMedian,
						},
					},
				},
			}
			url, sha := newChannelDefinitionsServer(t, channelDefinitions)

			// Set channel definitions with tombstoned channels 2 and 21
			_, err := configStore.SetChannelDefinitions(steve, donID, url, sha)
			require.NoError(t, err)
			backend.Commit()

			// Verify that channels 2 and 21 stop producing reports after tombstoning
			// We wait for a period where we don't see reports from these channels
			tombstonedChannels := map[uint32]bool{2: true, 21: true}
			checkPeriod := 5 * time.Second

			require.Eventually(t, func() bool {
				// Collect reports for a period and verify tombstoned channels don't appear
				startTime := time.Now()
				seenTombstonedChannels := make(map[uint32]bool)

				for time.Since(startTime) < checkPeriod {
					pckt, err := receiveWithTimeout(t, packetCh, 1*time.Second)
					if err != nil {
						// Timeout is okay, continue checking
						continue
					}
					req := pckt.req
					if req.ReportFormat == uint32(llotypes.ReportFormatJSON) {
						_, _, r, _, err := (datastreamsllo.JSONReportCodec{}).UnpackDecode(req.Payload)
						if err == nil && tombstonedChannels[r.ChannelID] {
							seenTombstonedChannels[r.ChannelID] = true
						}
					}
				}

				// Success if we didn't see any reports from tombstoned channels
				return len(seenTombstonedChannels) == 0
			}, 30*time.Second, 100*time.Millisecond, "channels 2 and 21 should stop producing reports after tombstoning")
		})

		// Scenario 4: Owner overwrites adder channel
		t.Run("Owner overwrites adder channel", func(t *testing.T) {
			// Owner sets a channel definition with same ID as adder1's channel 10
			channelDefinitions := llotypes.ChannelDefinitions{
				1: {
					ReportFormat: llotypes.ReportFormatJSON,
					Streams: []llotypes.Stream{
						{
							StreamID:   ethStreamID,
							Aggregator: llotypes.AggregatorMedian,
						},
					},
				},
				3: {
					ReportFormat: llotypes.ReportFormatJSON,
					Streams: []llotypes.Stream{
						{
							StreamID:   linkStreamID,
							Aggregator: llotypes.AggregatorMedian,
						},
					},
				},
				10: {
					ReportFormat: llotypes.ReportFormatJSON,
					Streams: []llotypes.Stream{
						{
							StreamID:   linkStreamID, // Changed from ethStreamID to linkStreamID
							Aggregator: llotypes.AggregatorMedian,
						},
					},
				},
			}
			url, sha := newChannelDefinitionsServer(t, channelDefinitions)

			// Set channel definitions
			_, err := configStore.SetChannelDefinitions(steve, donID, url, sha)
			require.NoError(t, err)
			backend.Commit()

			// Wait for channel definitions to be processed
			time.Sleep(10 * time.Second)

			// Wait for reports from channel 10 and verify it eventually uses owner's configuration (linkStreamID)
			// The owner's definition should take precedence over the adder's definition
			foundOwnerReport := false
			deadline := time.Now().Add(reportTimeout)
			for time.Now().Before(deadline) && !foundOwnerReport {
				pckt, err := receiveWithTimeout(t, packetCh, 2*time.Second)
				if err != nil {
					continue
				}
				req := pckt.req
				if req.ReportFormat == uint32(llotypes.ReportFormatJSON) {
					_, _, r, _, err := (datastreamsllo.JSONReportCodec{}).UnpackDecode(req.Payload)
					if err == nil && r.ChannelID == 10 {
						// Check if it has linkStream value (13.25) - owner's configuration
						// It might still have ethStream value (2976.39) initially, but should eventually switch
						value := r.Values[0].(*datastreamsllo.Decimal).String()
						if value == "13.25" {
							foundOwnerReport = true
						}
					}
				}
			}
			assert.True(t, foundOwnerReport, "should eventually receive report from channel 10 with owner's configuration (linkStream=13.25) after overwrite")
		})

		// Scenario 5: Verify adder cannot remove channels
		t.Run("Adder cannot remove channels", func(t *testing.T) {
			// Adder1 tries to set definitions that exclude channel 11 (which they previously added)
			adder1NewDefinitions := llotypes.ChannelDefinitions{
				10: {
					ReportFormat: llotypes.ReportFormatJSON,
					Streams: []llotypes.Stream{
						{
							StreamID:   ethStreamID,
							Aggregator: llotypes.AggregatorMedian,
						},
					},
					Source:    adder1ID,
					Tombstone: false,
				},
				// Channel 11 is intentionally omitted
			}

			adder1NewDefinitionsJSON, err := json.MarshalIndent(adder1NewDefinitions, "", "  ")
			require.NoError(t, err)
			adder1NewDefinitionsSHA := sha3.Sum256(adder1NewDefinitionsJSON)

			adder1NewServer := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
				_, errWrite := w.Write(adder1NewDefinitionsJSON)
				if errWrite != nil {
					w.WriteHeader(http.StatusInternalServerError)
					return
				}
				w.Header().Set("Content-Type", "application/json")
				w.WriteHeader(http.StatusOK)
			}))
			t.Cleanup(adder1NewServer.Close)

			_, err = configStore.AddChannelDefinitions(adder1, donID, adder1ID, adder1NewServer.URL, adder1NewDefinitionsSHA)
			require.NoError(t, err)
			backend.Commit()

			// Wait for processing
			time.Sleep(3 * time.Second)

			// Verify channel 11 still produces reports (adder cannot remove it)
			foundChannel11Report := false
			deadline := time.Now().Add(5 * time.Second)
			for time.Now().Before(deadline) && !foundChannel11Report {
				pckt, err := receiveWithTimeout(t, packetCh, 1*time.Second)
				if err != nil {
					continue
				}
				req := pckt.req
				if req.ReportFormat == uint32(llotypes.ReportFormatJSON) {
					_, _, r, _, err := (datastreamsllo.JSONReportCodec{}).UnpackDecode(req.Payload)
					if err == nil && r.ChannelID == 11 {
						foundChannel11Report = true
					}
				}
			}
			assert.True(t, foundChannel11Report, "channel 11 should still produce reports (adder cannot remove)")
		})
	})
}

func setupNodes(t *testing.T, nNodes int, backend evmtypes.Backend, clientCSAKeys []csakey.KeyV2, f func(*chainlink.Config)) (oracles []confighelper.OracleIdentityExtra, nodes []Node) {
	ports := freeport.GetN(t, nNodes)
	for i := range nNodes {
		app, peerID, transmitter, kb, observedLogs := setupNode(t, ports[i], fmt.Sprintf("oracle_streams_%d", i), backend, clientCSAKeys[i], f)

		nodes = append(nodes, Node{
			app, transmitter, kb, observedLogs,
		})
		offchainPublicKey, err := hex.DecodeString(strings.TrimPrefix(kb.OnChainPublicKey(), "0x"))
		require.NoError(t, err)
		oracles = append(oracles, confighelper.OracleIdentityExtra{
			OracleIdentity: confighelper.OracleIdentity{
				OnchainPublicKey:  offchainPublicKey,
				TransmitAccount:   ocr2types.Account(hex.EncodeToString(transmitter[:])),
				OffchainPublicKey: kb.OffchainPublicKey(),
				PeerID:            peerID,
			},
			ConfigEncryptionPublicKey: kb.ConfigEncryptionPublicKey(),
		})
	}
	return
}

func newChannelDefinitionsServer(t *testing.T, channelDefinitions llotypes.ChannelDefinitions) (url string, sha [32]byte) {
	channelDefinitionsJSON, err := json.MarshalIndent(channelDefinitions, "", "  ")
	require.NoError(t, err)
	channelDefinitionsSHA := sha3.Sum256(channelDefinitionsJSON)

	// Set up channel definitions server
	channelDefinitionsServer := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		assert.Equal(t, "GET", r.Method)
		assert.Equal(t, "application/json", r.Header.Get("Content-Type"))
		_, errWrite := w.Write(channelDefinitionsJSON)
		if errWrite != nil {
			w.WriteHeader(http.StatusInternalServerError)
			return
		}
		w.Header().Set("Content-Type", "application/json")
		w.WriteHeader(http.StatusOK)
	}))
	t.Cleanup(channelDefinitionsServer.Close)
	return channelDefinitionsServer.URL, channelDefinitionsSHA
}

func mustNewType(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
}

func mustMarshalJSON(v any) string {
	b, err := json.Marshal(v)
	if err != nil {
		panic(err)
	}
	return string(b)
}

func pad32bytes(d uint32) [32]byte {
	var result [32]byte
	binary.BigEndian.PutUint32(result[28:], d)
	return result
}

func newSingleABIEncoder(typ string, multiplier *sqlutil.Big) (enc lloevm.ABIEncoder) {
	if multiplier == nil {
		err := json.Unmarshal(fmt.Appendf(nil, `{"type":"%s"}`, typ), &enc)
		if err != nil {
			panic(err)
		}
		return
	}
	err := json.Unmarshal(fmt.Appendf(nil, `{"type":"%s","multiplier":"%s"}`, typ, multiplier.String()), &enc)
	if err != nil {
		panic(err)
	}
	return
}