feat: jd changesets

Author: ChrisAmora

go.mod

@@ -31,6 +31,7 @@ require (
 	golang.org/x/crypto v0.53.0
 	golang.org/x/mod v0.37.0
 	golang.org/x/sync v0.21.0
+	google.golang.org/grpc v1.81.1
 	gopkg.in/yaml.v3 v3.0.1
 )
 
@@ -320,7 +321,6 @@ require (
 	golang.org/x/tools v0.45.0 // indirect
 	google.golang.org/genproto/googleapis/api v0.0.0-20260526163538-3dc84a4a5aaa // indirect
 	google.golang.org/genproto/googleapis/rpc v0.0.0-20260526163538-3dc84a4a5aaa // indirect
-	google.golang.org/grpc v1.81.1 // indirect
 	google.golang.org/protobuf v1.36.11 // indirect
 	gopkg.in/evanphx/json-patch.v4 v4.12.0 // indirect
 	gopkg.in/inf.v0 v0.9.1 // indirect

jd/changesets/jobs.go

@@ -0,0 +1,137 @@
+package changesets
+
+import (
+	"errors"
+	"fmt"
+
+	jobv1 "github.com/smartcontractkit/chainlink-protos/job-distributor/v1/job"
+
+	cldf "github.com/smartcontractkit/chainlink-deployments-framework/deployment"
+	fwops "github.com/smartcontractkit/chainlink-deployments-framework/operations"
+
+	jdops "github.com/smartcontractkit/cld-changesets/jd/operations"
+)
+
+var (
+	_ cldf.ChangeSetV2[ProposeJobsInput] = ProposeJobsChangeset{}
+	_ cldf.ChangeSetV2[RevokeJobsInput]  = RevokeJobsChangeset{}
+	_ cldf.ChangeSetV2[DeleteJobsInput]  = DeleteJobsChangeset{}
+)
+
+type JobSpec = jdops.JobSpec
+type ProposeJobsInput = jdops.ProposeJobsInput
+type RevokeJobsInput = jdops.RevokeJobsInput
+type DeleteJobsInput = jdops.DeleteJobsInput
+
+// ProposeJobsChangeset proposes a batch of job specs to nodes matched by label.
+type ProposeJobsChangeset struct{}
+
+func (ProposeJobsChangeset) VerifyPreconditions(_ cldf.Environment, cfg ProposeJobsInput) error {
+	if cfg.Domain == "" {
+		return errors.New("domain is required")
+	}
+	if len(cfg.Jobs) == 0 {
+		return errors.New("no jobs provided")
+	}
+	for i, j := range cfg.Jobs {
+		if j.JobspecTOML == "" {
+			return fmt.Errorf("job[%d]: jobspec_toml is required", i)
+		}
+		if len(j.NodeLabels) == 0 {
+			return fmt.Errorf("job[%d]: node_labels is required", i)
+		}
+		for k, v := range j.NodeLabels {
+			if k == "" || v == "" {
+				return fmt.Errorf("job[%d]: node_labels key and value must be non-empty (key=%q value=%q)", i, k, v)
+			}
+		}
+	}
+
+	return nil
+}
+
+func (ProposeJobsChangeset) Apply(e cldf.Environment, input ProposeJobsInput) (cldf.ChangesetOutput, error) {
+	deps := jdops.JDOpDeps{Offchain: e.Offchain, EnvName: e.Name}
+	seqReport, err := fwops.ExecuteSequence(e.OperationsBundle, jdops.SeqJDProposeJobs, deps, input)
+	reports := append(seqReport.ExecutionReports, seqReport.ToGenericReport())
+
+	return cldf.ChangesetOutput{Reports: reports}, err
+}
+
+// RevokeJobsChangeset revokes jobs.
+type RevokeJobsChangeset struct{}
+
+func (RevokeJobsChangeset) VerifyPreconditions(_ cldf.Environment, cfg RevokeJobsInput) error {
+	if len(cfg.JobIDs) == 0 {
+		return errors.New("no job_ids provided")
+	}
+	seen := make(map[string]struct{}, len(cfg.JobIDs))
+	for _, id := range cfg.JobIDs {
+		if id == "" {
+			return errors.New("job id cannot be empty")
+		}
+		if _, ok := seen[id]; ok {
+			return fmt.Errorf("duplicate job id %q", id)
+		}
+		seen[id] = struct{}{}
+	}
+
+	return nil
+}
+
+func (RevokeJobsChangeset) Apply(e cldf.Environment, input RevokeJobsInput) (cldf.ChangesetOutput, error) {
+	deps := jdops.JDOpDeps{Offchain: e.Offchain, EnvName: e.Name}
+	seqReport, err := fwops.ExecuteSequence(e.OperationsBundle, jdops.SeqJDRevokeJobs, deps, input)
+	reports := append(seqReport.ExecutionReports, seqReport.ToGenericReport())
+
+	return cldf.ChangesetOutput{Reports: reports}, err
+}
+
+// DeleteJobsChangeset deletes jobs.
+type DeleteJobsChangeset struct{}
+
+func (DeleteJobsChangeset) VerifyPreconditions(env cldf.Environment, cfg DeleteJobsInput) error {
+	if len(cfg.JobIDs) == 0 {
+		return errors.New("no job_ids provided")
+	}
+	seen := make(map[string]struct{}, len(cfg.JobIDs))
+	for _, id := range cfg.JobIDs {
+		if id == "" {
+			return errors.New("job id cannot be empty")
+		}
+		if _, ok := seen[id]; ok {
+			return fmt.Errorf("duplicate job id %q", id)
+		}
+		seen[id] = struct{}{}
+	}
+
+	jobs, err := env.Offchain.ListJobs(env.GetContext(), &jobv1.ListJobsRequest{
+		Filter: &jobv1.ListJobsRequest_Filter{Ids: cfg.JobIDs, IncludeDeleted: true},
+	})
+	if err != nil {
+		return fmt.Errorf("failed to list jobs: %w", err)
+	}
+	for _, j := range jobs.Jobs {
+		if j.DeletedAt != nil {
+			return fmt.Errorf("job %q is already deleted", j.Id)
+		}
+	}
+	if len(jobs.Jobs) != len(cfg.JobIDs) {
+		found := make([]string, 0, len(jobs.Jobs))
+		for _, j := range jobs.Jobs {
+			found = append(found, j.Id)
+		}
+
+		return fmt.Errorf("not all jobs found: requested %v, found %v", cfg.JobIDs, found)
+	}
+
+	return nil
+}
+
+func (DeleteJobsChangeset) Apply(e cldf.Environment, input DeleteJobsInput) (cldf.ChangesetOutput, error) {
+	deps := jdops.JDOpDeps{Offchain: e.Offchain, EnvName: e.Name}
+	seqReport, err := fwops.ExecuteSequence(e.OperationsBundle, jdops.SeqJDDeleteJobs, deps, input)
+	reports := append(seqReport.ExecutionReports, seqReport.ToGenericReport())
+
+	return cldf.ChangesetOutput{Reports: reports}, err
+}