Cappl 3 ph

.goreleaser.develop.yaml

@@ -70,6 +70,7 @@ dockers:
       extra_files:
         - tmp/libs
         - tmp/plugins
+        - tmp/workflows
       build_flag_templates:
         - --platform=linux/amd64
         - --pull
@@ -121,6 +122,7 @@ dockers:
       extra_files:
         - tmp/libs
         - tmp/plugins
+        - tmp/workflows
       build_flag_templates:
         - --platform=linux/arm64
         - --pull
@@ -149,6 +151,7 @@ dockers:
         - '{{ .Env.IMG_PRE }}/ccip:sha-{{ .ShortCommit }}-amd64'
       extra_files:
         - tmp/libs
+        - tmp/workflows
         - ccip/config
       build_flag_templates:
         - --platform=linux/amd64
@@ -176,6 +179,7 @@ dockers:
       extra_files:
         - tmp/libs
         - tmp/plugins
+        - tmp/workflows
         - ccip/config
       build_flag_templates:
         - --platform=linux/amd64
@@ -205,6 +209,7 @@ dockers:
         - '{{ .Env.IMG_PRE }}/ccip:sha-{{ .ShortCommit }}-arm64'
       extra_files:
         - tmp/libs
+        - tmp/workflows
         - ccip/config
       build_flag_templates:
         - --platform=linux/arm64
@@ -231,6 +236,7 @@ dockers:
       extra_files:
         - tmp/libs
         - tmp/plugins
+        - tmp/workflows
         - ccip/config
       build_flag_templates:
         - --platform=linux/arm64

.mockery.yaml

@@ -579,21 +579,6 @@ packages:
   github.com/smartcontractkit/chainlink/v2/core/services/registrysyncer:
     interfaces:
       ORM:
-  github.com/smartcontractkit/chainlink/v2/core/services/workflows/syncer:
-    interfaces:
-      ORM:
-      ContractReader:
-        config:
-          mockname: "Mock{{ .InterfaceName }}"
-          filename: contract_reader_mock.go
-          inpackage: true
-          dir: "{{ .InterfaceDir }}"
-      Handler:
-        config:
-          mockname: "Mock{{ .InterfaceName }}"
-          filename: handler_mock.go
-          inpackage: true
-          dir: "{{ .InterfaceDir }}"
   github.com/smartcontractkit/chainlink/v2/core/capabilities/targets:
     interfaces:
       ContractValueGetter:

contracts/gas-snapshots/ccip.gas-snapshot

@@ -765,4 +765,4 @@ USDCTokenPool_releaseOrMint:test_TokenMaxCapacityExceeded_Revert() (gas: 47231)
 USDCTokenPool_releaseOrMint:test_UnlockingUSDCFailed_Revert() (gas: 95315)
 USDCTokenPool_setDomains:test_InvalidDomain_Revert() (gas: 66437)
 USDCTokenPool_setDomains:test_OnlyOwner_Revert() (gas: 11314)
-USDCTokenPool_supportsInterface:test_SupportsInterface_Success() (gas: 10107)
+USDCTokenPool_supportsInterface:test_SupportsInterface_Success() (gas: 10107)

core/capabilities/compute/compute.go

@@ -308,6 +308,7 @@ func (c *Compute) createFetcher() func(ctx context.Context, req *wasmpb.FetchReq
 			headersReq[k] = v.String()
 		}
 
+		fmt.Printf("LEN: %s, %d", req.Body, len(req.Body))
 		payloadBytes, err := json.Marshal(ghcapabilities.Request{
 			URL:       req.Url,
 			Method:    req.Method,

core/chainlink.goreleaser.Dockerfile

@@ -20,6 +20,9 @@ COPY ./chainlink /usr/local/bin/
 # Copy native libs if cgo is enabled
 COPY ./tmp/libs /usr/local/bin/libs
 
+# Copy workflows
+COPY ./tm[p]/workflow[s] /usr/local/workflows
+
 # Copy plugins if exist and enable them
 # https://stackoverflow.com/questions/70096208/dockerfile-copy-folder-if-it-exists-conditional-copy/70096420#70096420
 COPY ./tm[p]/plugin[s]/ /usr/local/bin/

core/cmd/shell_local.go

@@ -474,6 +474,12 @@ func (s *Shell) runNode(c *cli.Context) error {
 		if err2 != nil {
 			return errors.Wrap(err2, "failed to ensure workflow key")
 		}
+		keys, err := app.GetKeyStore().Workflow().GetAll()
+		if err != nil {
+			lggr.Errorf("WorkflowPublicKey: failed to get keys")
+		} else {
+			lggr.Infof("WorkflowPublicKey: %s", keys[0].PublicKeyString())
+		}
 	}
 
 	err2 := app.GetKeyStore().CSA().EnsureKey(rootCtx)

core/services/chainlink/application.go

@@ -215,9 +215,6 @@ func NewApplication(opts ApplicationOpts) (Application, error) {
 
 	// TODO: wire this up to config so we only instantiate it
 	// if a workflow registry address is provided.
-	workflowRegistrySyncer := syncer.NewNullWorkflowRegistrySyncer()
-	srvcs = append(srvcs, workflowRegistrySyncer)
-
 	var externalPeerWrapper p2ptypes.PeerWrapper
 	if cfg.Capabilities().Peering().Enabled() {
 		var dispatcher remotetypes.Dispatcher
@@ -475,6 +472,14 @@ func NewApplication(opts ApplicationOpts) (Application, error) {
 		webhookJobRunner = delegates[job.Webhook].(*webhook.Delegate).WebhookJobRunner()
 	)
 
+	workflowRegistrySyncer := &syncer.WorkflowRegistry{
+		DS:       opts.DS,
+		Logger:   globalLogger.Named("WorkflowRegistrySyncer"),
+		Store:    workflowORM,
+		Registry: opts.CapabilitiesRegistry,
+	}
+	srvcs = append(srvcs, workflowRegistrySyncer)
+
 	delegates[job.Workflow] = workflows.NewDelegate(
 		globalLogger,
 		opts.CapabilitiesRegistry,

core/services/gateway/handlers/capabilities/handler.go

@@ -20,10 +20,9 @@
 
 const (
 	// NOTE: more methods will go here. HTTP trigger/action/target; etc.
-	MethodWebAPITarget   = "web_api_target"
-	MethodWebAPITrigger  = "web_api_trigger"
-	MethodComputeAction  = "compute_action"
-	MethodWorkflowSyncer = "workflow_syncer"
+	MethodWebAPITarget  = "web_api_target"
+	MethodWebAPITrigger = "web_api_trigger"
+	MethodComputeAction = "compute_action"
 )
 
 type handler struct {
@@ -77,6 +76,7 @@
 // returns message to be sent back to the capability node
 func (h *handler) sendHTTPMessageToClient(ctx context.Context, req network.HTTPRequest, msg *api.Message) (*api.Message, error) {
 	var payload Response
+	fmt.Printf("req: %v, %d", req)
 	resp, err := h.httpClient.Send(ctx, req)
 	if err != nil {
 		return nil, err

core/services/gateway/network/httpclient.go

@@ -3,6 +3,7 @@ package network
 import (
 	"bytes"
 	"context"
+	"fmt"
 	"io"
 	"net/http"
 	"strings"
@@ -67,6 +68,7 @@ func (c *httpClient) Send(ctx context.Context, req HTTPRequest) (*HTTPResponse,
 	}
 	defer resp.Body.Close()
 
+	fmt.Printf("bytes: %d", c.config.MaxResponseBytes)
 	reader := http.MaxBytesReader(nil, resp.Body, int64(c.config.MaxResponseBytes))
 	body, err := io.ReadAll(reader)
 	if err != nil {

core/services/job/job_orm_test.go

@@ -45,7 +45,6 @@ import (
 	"github.com/smartcontractkit/chainlink/v2/core/services/relay"
 	"github.com/smartcontractkit/chainlink/v2/core/services/vrf/vrfcommon"
 	"github.com/smartcontractkit/chainlink/v2/core/services/webhook"
-	"github.com/smartcontractkit/chainlink/v2/core/services/workflows/syncer"
 	"github.com/smartcontractkit/chainlink/v2/core/testdata/testspecs"
 	"github.com/smartcontractkit/chainlink/v2/core/utils/testutils/heavyweight"
 )
@@ -1874,7 +1873,6 @@ func Test_ORM_FindJobByWorkflow(t *testing.T) {
 					c.ID = s.ID
 					c.Workflow = pkgworkflows.WFYamlSpec(t, "workflow99", addr1) // insert with mismatched name
 					c.SpecType = job.YamlSpec
-					c.SecretsID = s.SecretsID
 					return mustInsertWFJob(t, o, &c)
 				},
 			},
@@ -1894,40 +1892,29 @@ func Test_ORM_FindJobByWorkflow(t *testing.T) {
 					var c job.WorkflowSpec
 					c.ID = s.ID
 					c.Workflow = pkgworkflows.WFYamlSpec(t, "workflow03", addr2) // insert with mismatched owner
-					c.SecretsID = s.SecretsID
 					return mustInsertWFJob(t, o, &c)
 				},
 			},
 			wantErr: true,
 		},
 	}
 
-	for i, tt := range tests {
+	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
-			ctx := testutils.Context(t)
 			ks := cltest.NewKeyStore(t, tt.fields.ds)
-
-			secretsORM := syncer.NewWorkflowRegistryDS(tt.fields.ds, logger.TestLogger(t))
-
-			sid, err := secretsORM.Create(ctx, "some-url.com", fmt.Sprintf("some-hash-%d", i), "some-contentz")
-			require.NoError(t, err)
-			tt.args.spec.SecretsID = sql.NullInt64{Int64: sid, Valid: true}
-
 			pipelineORM := pipeline.NewORM(tt.fields.ds, logger.TestLogger(t), configtest.NewTestGeneralConfig(t).JobPipeline().MaxSuccessfulRuns())
 			bridgesORM := bridges.NewORM(tt.fields.ds)
 			o := NewTestORM(t, tt.fields.ds, pipelineORM, bridgesORM, ks)
-
 			var wantJobID int32
 			if tt.args.before != nil {
 				wantJobID = tt.args.before(t, o, tt.args.spec)
 			}
-
+			ctx := testutils.Context(t)
 			gotJ, err := o.FindJobIDByWorkflow(ctx, *tt.args.spec)
 			if (err != nil) != tt.wantErr {
 				t.Errorf("orm.FindJobByWorkflow() error = %v, wantErr %v", err, tt.wantErr)
 				return
 			}
-
 			if err == nil {
 				assert.Equal(t, wantJobID, gotJ, "mismatch job id")
 			}
@@ -1949,36 +1936,25 @@ func Test_ORM_FindJobByWorkflow_Multiple(t *testing.T) {
 			bridges.NewORM(db),
 			cltest.NewKeyStore(t, db))
 		ctx := testutils.Context(t)
-		secretsORM := syncer.NewWorkflowRegistryDS(db, logger.TestLogger(t))
-
-		var sids []int64
-		for i := 0; i < 3; i++ {
-			sid, err := secretsORM.Create(ctx, "some-url.com", fmt.Sprintf("some-hash-%d", i), "some-contentz")
-			require.NoError(t, err)
-			sids = append(sids, sid)
-		}
 
 		wfYaml1 := pkgworkflows.WFYamlSpec(t, "workflow00", addr1)
 		s1 := job.WorkflowSpec{
-			Workflow:  wfYaml1,
-			SpecType:  job.YamlSpec,
-			SecretsID: sql.NullInt64{Int64: sids[0], Valid: true},
+			Workflow: wfYaml1,
+			SpecType: job.YamlSpec,
 		}
 		wantJobID1 := mustInsertWFJob(t, o, &s1)
 
 		wfYaml2 := pkgworkflows.WFYamlSpec(t, "workflow01", addr1)
 		s2 := job.WorkflowSpec{
-			Workflow:  wfYaml2,
-			SpecType:  job.YamlSpec,
-			SecretsID: sql.NullInt64{Int64: sids[1], Valid: true},
+			Workflow: wfYaml2,
+			SpecType: job.YamlSpec,
 		}
 		wantJobID2 := mustInsertWFJob(t, o, &s2)
 
 		wfYaml3 := pkgworkflows.WFYamlSpec(t, "workflow00", addr2)
 		s3 := job.WorkflowSpec{
-			Workflow:  wfYaml3,
-			SpecType:  job.YamlSpec,
-			SecretsID: sql.NullInt64{Int64: sids[2], Valid: true},
+			Workflow: wfYaml3,
+			SpecType: job.YamlSpec,
 		}
 		wantJobID3 := mustInsertWFJob(t, o, &s3)
 
@@ -2016,7 +1992,7 @@ func mustInsertWFJob(t *testing.T, orm job.ORM, s *job.WorkflowSpec) int32 {
 	}
 
 	err = orm.CreateJob(ctx, &j)
-	require.NoError(t, err, "failed to insert job with wf spec %+v %s", s, err)
+	require.NoError(t, err, "failed to insert job with wf spec %v %s", s, s.Workflow)
 	return j.ID
 }
 

core/services/job/models.go

@@ -2,7 +2,6 @@ package job
 
 import (
 	"context"
-	"database/sql"
 	"database/sql/driver"
 	"encoding/json"
 	"fmt"
@@ -869,30 +868,18 @@ const (
 	DefaultSpecType                  = ""
 )
 
-type WorkflowSpecStatus string
-
-const (
-	WorkflowSpecStatusActive  WorkflowSpecStatus = "active"
-	WorkflowSpecStatusPaused  WorkflowSpecStatus = "paused"
-	WorkflowSpecStatusDefault WorkflowSpecStatus = ""
-)
-
 type WorkflowSpec struct {
 	ID       int32  `toml:"-"`
 	Workflow string `toml:"workflow"`           // the raw representation of the workflow
 	Config   string `toml:"config" db:"config"` // the raw representation of the config
 	// fields derived from the yaml spec, used for indexing the database
 	// note: i tried to make these private, but translating them to the database seems to require them to be public
-	WorkflowID    string             `toml:"-" db:"workflow_id"`    // Derived. Do not modify. the CID of the workflow.
-	WorkflowOwner string             `toml:"-" db:"workflow_owner"` // Derived. Do not modify. the owner of the workflow.
-	WorkflowName  string             `toml:"-" db:"workflow_name"`  // Derived. Do not modify. the name of the workflow.
-	Status        WorkflowSpecStatus `db:"status"`
-	BinaryURL     string             `db:"binary_url"`
-	ConfigURL     string             `db:"config_url"`
-	SecretsID     sql.NullInt64      `db:"secrets_id"`
-	CreatedAt     time.Time          `toml:"-"`
-	UpdatedAt     time.Time          `toml:"-"`
-	SpecType      WorkflowSpecType   `toml:"spec_type" db:"spec_type"`
+	WorkflowID    string           `toml:"-" db:"workflow_id"`    // Derived. Do not modify. the CID of the workflow.
+	WorkflowOwner string           `toml:"-" db:"workflow_owner"` // Derived. Do not modify. the owner of the workflow.
+	WorkflowName  string           `toml:"-" db:"workflow_name"`  // Derived. Do not modify. the name of the workflow.
+	CreatedAt     time.Time        `toml:"-"`
+	UpdatedAt     time.Time        `toml:"-"`
+	SpecType      WorkflowSpecType `toml:"spec_type" db:"spec_type"`
 	sdkWorkflow   *sdk.WorkflowSpec
 	rawSpec       []byte
 	config        []byte

core/services/job/orm.go

@@ -433,8 +433,8 @@ func (o *orm) CreateJob(ctx context.Context, jb *Job) error {
 		case Stream:
 			// 'stream' type has no associated spec, nothing to do here
 		case Workflow:
-			sql := `INSERT INTO workflow_specs (workflow, workflow_id, workflow_owner, workflow_name, binary_url, config_url, secrets_id, created_at, updated_at, spec_type, config)
-			VALUES (:workflow, :workflow_id, :workflow_owner, :workflow_name, :binary_url, :config_url, :secrets_id, NOW(), NOW(), :spec_type, :config)
+			sql := `INSERT INTO workflow_specs (workflow, workflow_id, workflow_owner, workflow_name, created_at, updated_at, spec_type, config)
+			VALUES (:workflow, :workflow_id, :workflow_owner, :workflow_name, NOW(), NOW(), :spec_type, :config)
 			RETURNING id;`
 			specID, err := tx.prepareQuerySpecID(ctx, sql, jb.WorkflowSpec)
 			if err != nil {