feat: add llo stream to data streams deployment module

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,118 @@
+package jobs
+
+import (
+	"testing"
+	"time"
+
+	"github.com/google/uuid"
+	"github.com/stretchr/testify/require"
+)
+
+func TestDonJobSpec_MarshalTOML(t *testing.T) {
+	trueVal := true
+	ocrKeyBundle := "ocr-bundle-123"
+
+	testCases := []struct {
+		name       string
+		spec       DonJobSpec
+		wantSubstr []string
+	}{
+		{
+			name: "with fields populated",
+			spec: DonJobSpec{
+				Base: Base{
+					Name:          "Test-DON",
+					Type:          "don",
+					SchemaVersion: 1,
+					ExternalJobID: uuid.New(),
+				},
+				ContractID:                        "contract-123",
+				TransmitterID:                     "tx-123",
+				ForwardingAllowed:                 &trueVal,
+				P2PV2Bootstrappers:                []string{"bootstrap1", "bootstrap2"},
+				OCRKeyBundleID:                    &ocrKeyBundle,
+				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"},
+				},
+			},
+			wantSubstr: []string{
+				"contractID",
+				"contract-123",
+				"transmitterID",
+				"tx-123",
+				"p2pv2Bootstrappers",
+				"bootstrap1",
+				"ocrKeyBundleID",
+				"ocr-bundle-123",
+				"maxTaskDuration",
+				"contractConfigTrackerPollInterval",
+				"relay",
+				"testrelay",
+				"pluginType",
+				"testplugin",
+				"chainID",
+				"chain",
+				"fromBlock",
+				"100",
+				"lloConfigMode",
+				"mode",
+				"lloDonID",
+				"200",
+				"channelDefinitionsContractAddress",
+				"0xabc",
+				"channelDefinitionsContractFromBlock",
+				"50",
+				"donID",
+				"300",
+				"servers",
+				"server1",
+				"http://localhost",
+			},
+		},
+		{
+			name: "empty minimal fields",
+			spec: DonJobSpec{
+				Base: Base{
+					Name:          "Empty-DON-Test",
+					Type:          "don",
+					SchemaVersion: 1,
+					ExternalJobID: uuid.New(),
+				},
+				ContractID:  "contract-empty",
+				RelayConfig: RelayConfigDon{},
+				PluginConfig: PluginConfigDon{
+					Servers: map[string]string{},
+				},
+			},
+			wantSubstr: []string{
+				"contractID",
+				"contract-empty",
+			},
+		},
+	}
+
+	for _, tc := range testCases {
+		t.Run(tc.name, func(t *testing.T) {
+			tomlBytes, err := tc.spec.MarshalTOML()
+			require.NoError(t, err)
+
+			result := string(tomlBytes)
+			for _, substr := range tc.wantSubstr {
+				require.Contains(t, result, substr, "result %q does not contain expected substring %q", result, substr)
+			}
+		})
+	}
+}

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)
+}