[KS-87] Generic workflow engine (#12461)

* [KS-86] Add handler to execute single step (#12394) * Add hardcoded YAML workflow and parsing * Execute an individual step * Execute an individual step * Execute an individual step * [KS-87] Add scheduler to workflow engine * Replace graph with graph lib * Add comments and renames * Replace graph with graph lib --------- Co-authored-by: HenryNguyen5 <[email protected]>
smartcontractkit · Mar 22, 2024 · 4fe2e77 · 4fe2e77
1 parent 84913bf
commit 4fe2e77
Show file tree

Hide file tree

Showing 16 changed files with 1,391 additions and 267 deletions.
diff --git a/core/scripts/go.mod b/core/scripts/go.mod
@@ -104,6 +104,7 @@ require (
 	github.com/dgryski/go-farm v0.0.0-20200201041132-a6ae2369ad13 // indirect
 	github.com/docker/distribution v2.8.2+incompatible // indirect
 	github.com/docker/go-units v0.5.0 // indirect
+	github.com/dominikbraun/graph v0.23.0 // indirect
 	github.com/dustin/go-humanize v1.0.1 // indirect
 	github.com/dvsekhvalnov/jose2go v1.5.0 // indirect
 	github.com/esote/minmaxheap v1.0.0 // indirect

diff --git a/core/scripts/go.sum b/core/scripts/go.sum
@@ -363,6 +363,8 @@ github.com/docker/go-connections v0.4.0 h1:El9xVISelRB7BuFusrZozjnkIM5YnzCViNKoh
 github.com/docker/go-connections v0.4.0/go.mod h1:Gbd7IOopHjR8Iph03tsViu4nIes5XhDvyHbTtUxmeec=
 github.com/docker/go-units v0.5.0 h1:69rxXcBk27SvSaaxTtLh/8llcHD8vYHT7WSdRZ/jvr4=
 github.com/docker/go-units v0.5.0/go.mod h1:fgPhTUdO+D/Jk86RDLlptpiXQzgHJF7gydDDbaIK4Dk=
+github.com/dominikbraun/graph v0.23.0 h1:TdZB4pPqCLFxYhdyMFb1TBdFxp8XLcJfTTBQucVPgCo=
+github.com/dominikbraun/graph v0.23.0/go.mod h1:yOjYyogZLY1LSG9E33JWZJiq5k83Qy2C6POAuiViluc=
 github.com/dustin/go-humanize v1.0.0/go.mod h1:HtrtbFcZ19U5GC7JDqmcUSB87Iq5E25KnS6fMYU6eOk=
 github.com/dustin/go-humanize v1.0.1 h1:GzkhY7T5VNhEkwH0PVJgjz+fX1rhBrR7pRT3mDkpeCY=
 github.com/dustin/go-humanize v1.0.1/go.mod h1:Mu1zIs6XwVuF/gI1OepvI0qD18qycQx+mFykh5fBlto=

diff --git a/core/services/workflows/delegate.go b/core/services/workflows/delegate.go
@@ -15,6 +15,56 @@ import (
 	"github.com/smartcontractkit/chainlink/v2/core/services/pg"
 )
 
+const hardcodedWorkflow = `
+triggers:
+  - type: "on_mercury_report"
+    config:
+      feedlist:
+        - "0x1111111111111111111100000000000000000000000000000000000000000000" # ETHUSD
+        - "0x2222222222222222222200000000000000000000000000000000000000000000" # LINKUSD
+        - "0x3333333333333333333300000000000000000000000000000000000000000000" # BTCUSD
+        
+consensus:
+  - type: "offchain_reporting"
+    ref: "evm_median"
+    inputs:
+      observations:
+        - "$(trigger.outputs)"
+    config:
+      aggregation_method: "data_feeds_2_0"
+      aggregation_config:
+        "0x1111111111111111111100000000000000000000000000000000000000000000":
+          deviation: "0.001"
+          heartbeat: "30m"
+        "0x2222222222222222222200000000000000000000000000000000000000000000":
+          deviation: "0.001"
+          heartbeat: "30m"
+        "0x3333333333333333333300000000000000000000000000000000000000000000":
+          deviation: "0.001"
+          heartbeat: "30m"
+      encoder: "EVM"
+      encoder_config:
+        abi: "mercury_reports bytes[]"
+
+targets:
+  - type: "write_polygon-testnet-mumbai"
+    inputs:
+      report:
+        - "$(evm_median.outputs.reports)"
+    config:
+      address: "0x3F3554832c636721F1fD1822Ccca0354576741Ef"
+      params: ["$(inputs.report)"]
+      abi: "receive(report bytes)"
+  - type: "write_ethereum-testnet-sepolia"
+    inputs:
+      report:
+        - "$(evm_median.outputs.reports)"
+    config:
+      address: "0x54e220867af6683aE6DcBF535B4f952cB5116510"
+      params: ["$(inputs.report)"]
+      abi: "receive(report bytes)"
+`
+
 type Delegate struct {
 	registry types.CapabilitiesRegistry
 	logger   logger.Logger
@@ -36,7 +86,12 @@ func (d *Delegate) OnDeleteJob(ctx context.Context, jb job.Job, q pg.Queryer) er
 
 // ServicesForSpec satisfies the job.Delegate interface.
 func (d *Delegate) ServicesForSpec(ctx context.Context, spec job.Job) ([]job.ServiceCtx, error) {
-	engine, err := NewEngine(d.logger, d.registry)
+	cfg := Config{
+		Lggr:     d.logger,
+		Spec:     hardcodedWorkflow,
+		Registry: d.registry,
+	}
+	engine, err := NewEngine(cfg)
 	if err != nil {
 		return nil, err
 	}