feat: add llo stream to data streams deployment module (#16459)

* feat: add llo stream to data streams deployment module

* feat: add don

* feat: update tests

* feat: add pointer lib to deployment module

* feat: change pointer util

* fix: goimport

deployment/data-streams/jobs/don.go

@@ -0,0 +1,42 @@
+package jobs
+
+import (
+	"time"
+
+	"github.com/pelletier/go-toml/v2"
+)
+
+type DonJobSpec struct {
+	Base
+
+	ContractID                        string          `toml:"contractID"`
+	TransmitterID                     string          `toml:"transmitterID,omitempty"`
+	ForwardingAllowed                 *bool           `toml:"forwardingAllowed,omitempty"`
+	P2PV2Bootstrappers                []string        `toml:"p2pv2Bootstrappers,omitempty"`
+	OCRKeyBundleID                    *string         `toml:"ocrKeyBundleID,omitempty"`
+	MaxTaskDuration                   time.Duration   `toml:"maxTaskDuration,omitempty"`
+	ContractConfigTrackerPollInterval time.Duration   `toml:"contractConfigTrackerPollInterval,omitempty"`
+	Relay                             RelayType       `toml:"relay,omitempty"`
+	PluginType                        string          `toml:"pluginType,omitempty"`
+	RelayConfig                       RelayConfigDon  `toml:"relayConfig"`
+	PluginConfig                      PluginConfigDon `toml:"pluginConfig"`
+}
+
+// RelayConfig is the configuration for the relay. This could change depending on the relay type.
+type RelayConfigDon struct {
+	ChainID       string `toml:"chainID"`
+	FromBlock     uint64 `toml:"fromBlock,omitempty"`
+	LLOConfigMode string `toml:"lloConfigMode,omitempty"`
+	LLODonID      int64  `toml:"lloDonID,omitempty"`
+}
+
+type PluginConfigDon struct {
+	ChannelDefinitionsContractAddress   string            `toml:"channelDefinitionsContractAddress"`
+	ChannelDefinitionsContractFromBlock uint64            `toml:"channelDefinitionsContractFromBlock"`
+	DonID                               int64             `toml:"donID"`
+	Servers                             map[string]string `toml:"servers,inline"`
+}
+
+func (s *DonJobSpec) MarshalTOML() ([]byte, error) {
+	return toml.Marshal(s)
+}

deployment/data-streams/jobs/don_test.go

@@ -0,0 +1,122 @@
+package jobs
+
+import (
+	"testing"
+	"time"
+
+	"github.com/google/uuid"
+	"github.com/stretchr/testify/require"
+
+	"github.com/smartcontractkit/chainlink/deployment/data-streams/utils/pointer"
+)
+
+const donSpecTOML1 = `name = 'Test-DON'
+type = 'don'
+schemaVersion = 1
+externalJobID = '00000000-0000-0000-0000-000000000000'
+contractID = 'contract-123'
+transmitterID = 'tx-123'
+forwardingAllowed = true
+p2pv2Bootstrappers = ['bootstrap1', 'bootstrap2']
+ocrKeyBundleID = 'ocr-bundle-123'
+maxTaskDuration = 10000000000
+contractConfigTrackerPollInterval = 60000000000
+relay = 'testrelay'
+pluginType = 'testplugin'
+
+[relayConfig]
+chainID = 'chain'
+fromBlock = 100
+lloConfigMode = 'mode'
+lloDonID = 200
+
+[pluginConfig]
+channelDefinitionsContractAddress = '0xabc'
+channelDefinitionsContractFromBlock = 50
+donID = 300
+servers = {server1 = 'http://localhost'}
+`
+
+const donSpecTOML2 = `name = 'Empty-DON-Test'
+type = 'don'
+schemaVersion = 1
+externalJobID = '00000000-0000-0000-0000-000000000000'
+contractID = 'contract-empty'
+
+[relayConfig]
+chainID = ''
+
+[pluginConfig]
+channelDefinitionsContractAddress = ''
+channelDefinitionsContractFromBlock = 0
+donID = 0
+servers = {}
+`
+
+func TestDonJobSpec_MarshalTOML(t *testing.T) {
+	testCases := []struct {
+		name string
+		spec DonJobSpec
+		want string
+	}{
+		{
+			name: "with fields populated",
+			spec: DonJobSpec{
+				Base: Base{
+					Name:          "Test-DON",
+					Type:          "don",
+					SchemaVersion: 1,
+					ExternalJobID: uuid.MustParse("00000000-0000-0000-0000-000000000000"),
+				},
+				ContractID:                        "contract-123",
+				TransmitterID:                     "tx-123",
+				ForwardingAllowed:                 pointer.To(true),
+				P2PV2Bootstrappers:                []string{"bootstrap1", "bootstrap2"},
+				OCRKeyBundleID:                    pointer.To("ocr-bundle-123"),
+				MaxTaskDuration:                   10 * time.Second,
+				ContractConfigTrackerPollInterval: 1 * time.Minute,
+				Relay:                             "testrelay",
+				PluginType:                        "testplugin",
+				RelayConfig: RelayConfigDon{
+					ChainID:       "chain",
+					FromBlock:     100,
+					LLOConfigMode: "mode",
+					LLODonID:      200,
+				},
+				PluginConfig: PluginConfigDon{
+					ChannelDefinitionsContractAddress:   "0xabc",
+					ChannelDefinitionsContractFromBlock: 50,
+					DonID:                               300,
+					Servers:                             map[string]string{"server1": "http://localhost"},
+				},
+			},
+			want: donSpecTOML1,
+		},
+		{
+			name: "empty minimal fields",
+			spec: DonJobSpec{
+				Base: Base{
+					Name:          "Empty-DON-Test",
+					Type:          "don",
+					SchemaVersion: 1,
+					ExternalJobID: uuid.MustParse("00000000-0000-0000-0000-000000000000"),
+				},
+				ContractID:  "contract-empty",
+				RelayConfig: RelayConfigDon{},
+				PluginConfig: PluginConfigDon{
+					Servers: map[string]string{},
+				},
+			},
+			want: donSpecTOML2,
+		},
+	}
+
+	for _, tc := range testCases {
+		t.Run(tc.name, func(t *testing.T) {
+			tomlBytes, err := tc.spec.MarshalTOML()
+			require.NoError(t, err)
+			got := string(tomlBytes)
+			require.Equal(t, tc.want, got)
+		})
+	}
+}

deployment/data-streams/jobs/stream.go

@@ -1 +1,66 @@
 package jobs
+
+import (
+	"github.com/pelletier/go-toml/v2"
+)
+
+type Datasource struct {
+	BridgeName string
+	ReqData    string
+}
+
+type ReportFieldLLO struct {
+	ResultPath string
+}
+
+type Pipeline interface {
+	Render() (string, error)
+}
+
+type BaseObservationSource struct {
+	Datasources   []Datasource
+	AllowedFaults int
+	Benchmark     ReportFieldLLO
+}
+
+type QuoteObservationSource struct {
+	BaseObservationSource
+	Bid ReportFieldLLO
+	Ask ReportFieldLLO
+}
+
+type MedianObservationSource struct {
+	BaseObservationSource
+}
+
+func renderObservationTemplate(fname string, obs any) (string, error) {
+	return renderTemplate(fname, obs)
+}
+
+func (src QuoteObservationSource) Render() (string, error) {
+	return renderObservationTemplate("osrc_mercury_v1_quote.go.tmpl", src)
+}
+
+func (src MedianObservationSource) Render() (string, error) {
+	return renderObservationTemplate("osrc_mercury_v1_median.go.tmpl", src)
+}
+
+type StreamJobSpec struct {
+	Base
+
+	StreamID          string `toml:"streamID"`
+	ObservationSource string `toml:"observationSource,multiline,omitempty"`
+}
+
+func (s *StreamJobSpec) SetObservationSource(obs Pipeline) error {
+	rendered, err := obs.Render()
+	if err != nil {
+		return err
+	}
+	s.ObservationSource = rendered
+	return nil
+}
+
+func (s *StreamJobSpec) MarshalTOML() ([]byte, error) {
+	return toml.Marshal(s)
+}