feat(workflows): adds registry syncer

.mockery.yaml

@@ -579,6 +579,21 @@ 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:

core/services/chainlink/application.go

@@ -23,6 +23,7 @@ import (
 	"github.com/smartcontractkit/chainlink-common/pkg/loop"
 	commonservices "github.com/smartcontractkit/chainlink-common/pkg/services"
 	"github.com/smartcontractkit/chainlink-common/pkg/sqlutil"
+	commontypes "github.com/smartcontractkit/chainlink-common/pkg/types"
 	"github.com/smartcontractkit/chainlink-common/pkg/utils"
 	"github.com/smartcontractkit/chainlink-common/pkg/utils/jsonserializable"
 	"github.com/smartcontractkit/chainlink-common/pkg/utils/mailbox"
@@ -215,7 +216,7 @@ 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.NewWorkflowRegistry()
+	workflowRegistrySyncer := syncer.NewWorkflowRegistry[commontypes.ContractReader](nil, nil, nil, nil, "")
 	srvcs = append(srvcs, workflowRegistrySyncer)
 
 	var externalPeerWrapper p2ptypes.PeerWrapper

core/services/workflows/engine.go

@@ -92,7 +92,7 @@ func (sucm *stepUpdateManager) len() int64 {
 }
 
 type secretsFetcher interface {
-	SecretsFor(workflowOwner, workflowName string) (map[string]string, error)
+	SecretsFor(ctx context.Context, workflowOwner, workflowName string) (map[string]string, error)
 }
 
 // Engine handles the lifecycle of a single workflow and its executions.
@@ -849,7 +849,7 @@ func (e *Engine) interpolateEnvVars(config map[string]any, env exec.Env) (*value
 // registry (for capability-level configuration). It doesn't perform any caching of the config values, since
 // the two registries perform their own caching.
 func (e *Engine) configForStep(ctx context.Context, lggr logger.Logger, step *step) (*values.Map, error) {
-	secrets, err := e.secretsFetcher.SecretsFor(e.workflow.owner, e.workflow.name)
+	secrets, err := e.secretsFetcher.SecretsFor(ctx, e.workflow.owner, e.workflow.name)
 	if err != nil {
 		return nil, fmt.Errorf("failed to fetch secrets: %w", err)
 	}

core/services/workflows/engine_test.go

@@ -152,7 +152,7 @@ func newTestEngineWithYAMLSpec(t *testing.T, reg *coreCap.Registry, spec string,
 
 type mockSecretsFetcher struct{}
 
-func (s mockSecretsFetcher) SecretsFor(workflowOwner, workflowName string) (map[string]string, error) {
+func (s mockSecretsFetcher) SecretsFor(ctx context.Context, workflowOwner, workflowName string) (map[string]string, error) {
 	return map[string]string{}, nil
 }
 
@@ -1605,7 +1605,7 @@ type mockFetcher struct {
 	retval map[string]string
 }
 
-func (m *mockFetcher) SecretsFor(workflowOwner, workflowName string) (map[string]string, error) {
+func (m *mockFetcher) SecretsFor(ctx context.Context, workflowOwner, workflowName string) (map[string]string, error) {
 	return m.retval, nil
 }
 

core/services/workflows/syncer/handler.go

@@ -0,0 +1,72 @@
+package syncer
+
+import (
+	"context"
+	"encoding/hex"
+	"fmt"
+
+	"github.com/smartcontractkit/chainlink/v2/core/logger"
+)
+
+// eventHandler is a handler for WorkflowRegistryEvent events.  Each event type has a corresponding
+// method that handles the event.
+type eventHandler struct {
+	lggr    logger.Logger
+	orm     ORM
+	fetcher FetcherFunc
+}
+
+// newEventHandler returns a new eventHandler instance.
+func newEventHandler(
+	lggr logger.Logger,
+	orm ORM,
+	gateway FetcherFunc,
+) *eventHandler {
+	return &eventHandler{
+		lggr:    lggr,
+		orm:     orm,
+		fetcher: gateway,
+	}
+}
+
+func (h *eventHandler) Handle(ctx context.Context, event WorkflowRegistryEvent) error {
+	switch event.EventType {
+	case ForceUpdateSecretsEvent:
+		return h.forceUpdateSecretsEvent(ctx, event)
+	default:
+		return fmt.Errorf("event type unsupported: %v", event.EventType)
+	}
+}
+
+// forceUpdateSecretsEvent handles the ForceUpdateSecretsEvent event type.
+func (h *eventHandler) forceUpdateSecretsEvent(
+	ctx context.Context,
+	event WorkflowRegistryEvent,
+) error {
+	// Get the URL of the secrets file from the event data
+	data, ok := event.Data.(WorkflowRegistryForceUpdateSecretsRequestedV1)
+	if !ok {
+		return fmt.Errorf("invalid data type %T for event", event.Data)
+	}
+
+	hash := hex.EncodeToString(data.SecretsURLHash)
+
+	url, err := h.orm.GetSecretsURLByHash(ctx, hash)
+	if err != nil {
+		h.lggr.Errorf("failed to get URL by hash %s : %s", hash, err)
+		return err
+	}
+
+	// Fetch the contents of the secrets file from the url via the fetcher
+	secrets, err := h.fetcher(ctx, url)
+	if err != nil {
+		return err
+	}
+
+	// Update the secrets in the ORM
+	if _, err := h.orm.Update(ctx, hash, string(secrets)); err != nil {
+		return err
+	}
+
+	return nil
+}