operations

var OpJDDeleteJob = fwops.NewOperation(
	"jd-delete-job",
	semver.MustParse("1.0.0"),
	"Delete a job",
	func(b fwops.Bundle, deps JDOpDeps, in DeleteJobInput) (DeleteJobOutput, error) {
		res, err := deps.Offchain.DeleteJob(b.GetContext(), &jobv1.DeleteJobRequest{
			IdOneof: &jobv1.DeleteJobRequest_Id{Id: in.JobID},
		})
		if err != nil {
			return DeleteJobOutput{}, fmt.Errorf("failed to delete job %q: %w", in.JobID, err)
		}
		if res.GetJob().GetDeletedAt() == nil {
			return DeleteJobOutput{}, fmt.Errorf("delete job %q response did not confirm deletion", in.JobID)
		}
		b.Logger.Infow("deleted job", "job_id", in.JobID)

		return DeleteJobOutput{}, nil
	},
)

var OpJDDisableNode = fwops.NewOperation(
	"jd-disable-node",
	semver.MustParse("1.0.0"),
	"Disable a node by CSA key",
	func(b fwops.Bundle, deps JDOpDeps, in DisableNodeInput) (DisableNodeOutput, error) {
		ctx := b.GetContext()
		nodeInfo, err := ListNodeByPublicKey(ctx, deps.Offchain, in.CSAKey)
		if err != nil {
			return DisableNodeOutput{}, fmt.Errorf("failed to look up node with csa_key %s: %w", in.CSAKey, err)
		}
		if nodeInfo == nil {
			b.Logger.Infow("node not found, skipping disable", "csa_key", in.CSAKey)

			return DisableNodeOutput{Skipped: true}, nil
		}
		if !nodeInfo.GetIsEnabled() {
			b.Logger.Infow("node already disabled, skipping", "node_id", nodeInfo.GetId())

			return DisableNodeOutput{NodeID: nodeInfo.GetId(), Skipped: true}, nil
		}
		if _, err = deps.Offchain.DisableNode(ctx, &nodev1.DisableNodeRequest{Id: nodeInfo.GetId()}); err != nil {
			return DisableNodeOutput{}, fmt.Errorf("failed to disable node %q (%s): %w", nodeInfo.GetId(), nodeInfo.GetName(), err)
		}
		b.Logger.Infow("disabled node", "node_id", nodeInfo.GetId(), "name", nodeInfo.GetName())

		return DisableNodeOutput{NodeID: nodeInfo.GetId()}, nil
	},
)

var OpJDProposeJob = fwops.NewOperation(
	"jd-propose-job",
	semver.MustParse("1.0.0"),
	"Propose a job spec to nodes matched by label",
	func(b fwops.Bundle, deps JDOpDeps, in ProposeJobInput) (ProposeJobOutput, error) {
		ctx := b.GetContext()
		d := domain.NewDomain("", in.Domain)
		domainKey := d.Key()
		envName := deps.EnvName

		selectors := make([]*ptypes.Selector, 0, 2+len(in.NodeLabels))
		selectors = append(selectors,
			&ptypes.Selector{Key: "product", Op: ptypes.SelectorOp_EQ, Value: &domainKey},
			&ptypes.Selector{Key: "environment", Op: ptypes.SelectorOp_EQ, Value: &envName},
		)
		for k, v := range in.NodeLabels {
			selectors = append(selectors, &ptypes.Selector{Key: k, Op: ptypes.SelectorOp_EQ, Value: &v})
		}

		resp, err := deps.Offchain.ListNodes(ctx, &nodev1.ListNodesRequest{
			Filter: &nodev1.ListNodesRequest_Filter{Enabled: 1, Selectors: selectors},
		})
		if err != nil {
			return ProposeJobOutput{}, fmt.Errorf("failed to list nodes: %w", err)
		}
		nodes := resp.GetNodes()
		if len(nodes) == 0 {
			return ProposeJobOutput{}, fmt.Errorf("no nodes matched domain=%q env=%q labels=%v", in.Domain, deps.EnvName, in.NodeLabels)
		}

		var merr error
		for _, node := range nodes {
			if _, err = deps.Offchain.ProposeJob(ctx, &jobv1.ProposeJobRequest{
				NodeId: node.GetId(),
				Spec:   in.JobspecTOML,
				Labels: labelsFromMap(in.JobLabels),
			}); err != nil {
				merr = errors.Join(merr, fmt.Errorf("failed to propose job to node %q: %w", node.GetId(), err))
			}
		}
		if merr != nil {
			return ProposeJobOutput{}, merr
		}
		b.Logger.Infow("proposed job", "node_labels", in.NodeLabels, "node_count", len(nodes))

		return ProposeJobOutput{}, nil
	},
)

var OpJDRegisterNode = fwops.NewOperation(
	"jd-register-node",
	semver.MustParse("1.0.0"),
	"Register a node",
	func(b fwops.Bundle, deps JDOpDeps, in RegisterNodeInput) (RegisterNodeOutput, error) {
		ctx := b.GetContext()

		existing, err := ListNodeByPublicKey(ctx, deps.Offchain, in.CSAKey)
		if err != nil {
			return RegisterNodeOutput{}, fmt.Errorf("failed to check if node is already registered: %w", err)
		}
		if existing != nil {
			b.Logger.Infow("node already registered, skipping", "name", in.Name, "csa_key", in.CSAKey)

			return RegisterNodeOutput{NodeID: existing.GetId(), Skipped: true}, nil
		}
		labels := in.Labels
		if labels == nil {
			labels = make(map[string]string)
		}
		d := domain.NewDomain("", in.Domain)
		nodeID, err := cldf_engine_offchain.RegisterNode(ctx, deps.Offchain, in.Name, in.CSAKey, in.IsBootstrap, d, deps.EnvName, labels)
		if err != nil {
			return RegisterNodeOutput{}, fmt.Errorf("failed to register node %q (csa_key: %s): %w", in.Name, in.CSAKey, err)
		}
		b.Logger.Infow("registered node", "name", in.Name, "id", nodeID)

		return RegisterNodeOutput{NodeID: nodeID}, nil
	},
)

var OpJDRevokeJob = fwops.NewOperation(
	"jd-revoke-job",
	semver.MustParse("1.0.0"),
	"Revoke a job",
	func(b fwops.Bundle, deps JDOpDeps, in RevokeJobInput) (RevokeJobOutput, error) {
		_, err := deps.Offchain.RevokeJob(b.GetContext(), &jobv1.RevokeJobRequest{
			IdOneof: &jobv1.RevokeJobRequest_Id{Id: in.JobID},
		})
		if err != nil {
			if isJDNotFound(err) {
				b.Logger.Infow("job already absent or revoked", "job_id", in.JobID)

				return RevokeJobOutput{AlreadyAbsent: true}, nil
			}

			return RevokeJobOutput{}, fmt.Errorf("failed to revoke job %q: %w", in.JobID, err)
		}
		b.Logger.Infow("revoked job", "job_id", in.JobID)

		return RevokeJobOutput{}, nil
	},
)

var OpJDUpdateNode = fwops.NewOperation(
	"jd-update-node",
	semver.MustParse("1.0.0"),
	"Update a node's name and/or labels",
	func(b fwops.Bundle, deps JDOpDeps, in UpdateNodeInput) (UpdateNodeOutput, error) {
		ctx := b.GetContext()
		resp, err := deps.Offchain.GetNode(ctx, &nodev1.GetNodeRequest{Id: in.ID})
		if err != nil {
			return UpdateNodeOutput{}, fmt.Errorf("failed to get node %q: %w", in.ID, err)
		}
		nodeInfo := resp.GetNode()

		existing, lookupErr := ListNodeByPublicKey(ctx, deps.Offchain, in.CSAKey)
		if lookupErr != nil {
			b.Logger.Warnw("could not verify CSA key conflict, skipping check", "csa_key", in.CSAKey, "error", lookupErr)
		} else if existing != nil && existing.GetId() != in.ID {
			return UpdateNodeOutput{}, fmt.Errorf(
				"CSA key %s is already registered to node %q (%s), cannot reassign to %q",
				in.CSAKey, existing.GetId(), existing.GetName(), in.ID,
			)
		}

		name := nodeInfo.GetName()
		if in.Name != "" {
			name = in.Name
		}
		labels := nodeInfo.GetLabels()
		if in.Labels != nil {
			labels = labelsFromMap(in.Labels)
		}

		if _, err = deps.Offchain.UpdateNode(ctx, &nodev1.UpdateNodeRequest{
			Id:        in.ID,
			Name:      name,
			PublicKey: in.CSAKey,
			Labels:    labels,
		}); err != nil {
			return UpdateNodeOutput{}, fmt.Errorf("failed to update node %q: %w", in.ID, err)
		}
		b.Logger.Infow("updated node", "id", in.ID, "name", name,
			"old_csa_key", nodeInfo.GetPublicKey(), "new_csa_key", in.CSAKey)

		return UpdateNodeOutput{NodeID: in.ID}, nil
	},
)

var SeqJDDeleteJobs = fwops.NewSequence(
	"seq-jd-delete-jobs",
	semver.MustParse("1.0.0"),
	"Delete multiple jobs",
	func(b fwops.Bundle, deps JDOpDeps, in DeleteJobsInput) (DeleteJobsOutput, error) {
		resp, err := deps.Offchain.ListProposals(b.GetContext(), &jobv1.ListProposalsRequest{
			Filter: &jobv1.ListProposalsRequest_Filter{JobIds: in.JobIDs},
		})
		if err != nil {
			return DeleteJobsOutput{}, fmt.Errorf("failed to list proposals: %w", err)
		}
		if len(resp.Proposals) == 0 {
			return DeleteJobsOutput{}, fmt.Errorf("no proposals found for job ids %v", in.JobIDs)
		}

		eligible := eligibleJobIDsForDeletion(resp.Proposals)
		if len(eligible) == 0 {
			return DeleteJobsOutput{}, errors.New("no jobs eligible for deletion: no proposals in PROPOSED, APPROVED, or PENDING state")
		}

		var deletedIDs []string
		var failed int
		var merr error
		for _, jobID := range eligible {
			_, err := fwops.ExecuteOperation(b, OpJDDeleteJob, deps, DeleteJobInput{JobID: jobID})
			if err != nil {
				failed++
				merr = errors.Join(merr, err)
				b.Logger.Errorw("failed to delete job", "job_id", jobID, "error", err)

				continue
			}
			deletedIDs = append(deletedIDs, jobID)
		}
		b.Logger.Infow("delete_jobs complete",
			"total", len(eligible),
			"deleted", len(deletedIDs),
			"failed", failed)

		return DeleteJobsOutput{DeletedJobIDs: deletedIDs}, merr
	},
)

var SeqJDDisableNodes = fwops.NewSequence(
	"seq-jd-disable-nodes",
	semver.MustParse("1.0.0"),
	"Disable multiple nodes by CSA key",
	func(b fwops.Bundle, deps JDOpDeps, in DisableNodesInput) (DisableNodesOutput, error) {
		var disabledIDs, skippedCSAKeys []string
		var failed int
		var merr error
		for _, csaKey := range in.CSAKeys {
			report, err := fwops.ExecuteOperation(b, OpJDDisableNode, deps, DisableNodeInput{CSAKey: csaKey})
			if err != nil {
				failed++
				merr = errors.Join(merr, err)
				b.Logger.Errorw("failed to disable node", "csa_key", csaKey, "error", err)

				continue
			}
			if report.Output.Skipped {
				skippedCSAKeys = append(skippedCSAKeys, csaKey)
			} else {
				disabledIDs = append(disabledIDs, report.Output.NodeID)
			}
		}
		b.Logger.Infow("disable_nodes complete",
			"total", len(in.CSAKeys),
			"disabled", len(disabledIDs),
			"skipped", len(skippedCSAKeys),
			"failed", failed)

		return DisableNodesOutput{DisabledNodeIDs: disabledIDs, SkippedCSAKeys: skippedCSAKeys}, merr
	},
)

var SeqJDProposeJobs = fwops.NewSequence(
	"seq-jd-propose-jobs",
	semver.MustParse("1.0.0"),
	"Propose multiple job specs",
	func(b fwops.Bundle, deps JDOpDeps, in ProposeJobsInput) (ProposeJobsOutput, error) {
		var failed int
		var merr error
		for i, j := range in.Jobs {
			_, err := fwops.ExecuteOperation(b, OpJDProposeJob, deps, ProposeJobInput{
				Domain:      in.Domain,
				NodeLabels:  j.NodeLabels,
				JobLabels:   j.JobLabels,
				JobspecTOML: j.JobspecTOML,
			})
			if err != nil {
				failed++
				merr = errors.Join(merr, fmt.Errorf("job[%d]: propose failed: %w", i, err))
				b.Logger.Errorw("failed to propose job", "index", i, "node_labels", j.NodeLabels, "error", err)
			}
		}
		b.Logger.Infow("propose_jobs complete",
			"total", len(in.Jobs),
			"succeeded", len(in.Jobs)-failed,
			"failed", failed)

		return ProposeJobsOutput{}, merr
	},
)

var SeqJDRegisterNodes = fwops.NewSequence(
	"seq-jd-register-nodes",
	semver.MustParse("1.0.0"),
	"Register multiple nodes",
	func(b fwops.Bundle, deps JDOpDeps, in RegisterNodesInput) (RegisterNodesOutput, error) {
		var registeredIDs, skippedIDs []string
		var failed int
		var merr error
		for _, n := range in.Nodes {
			report, err := fwops.ExecuteOperation(b, OpJDRegisterNode, deps, RegisterNodeInput{
				Domain:      in.Domain,
				Name:        n.Name,
				CSAKey:      n.CSAKey,
				IsBootstrap: n.IsBootstrap,
				Labels:      n.Labels,
			})
			if err != nil {
				failed++
				merr = errors.Join(merr, err)
				b.Logger.Errorw("failed to register node", "name", n.Name, "error", err)

				continue
			}
			if report.Output.Skipped {
				skippedIDs = append(skippedIDs, report.Output.NodeID)
			} else {
				registeredIDs = append(registeredIDs, report.Output.NodeID)
			}
		}
		b.Logger.Infow("register_nodes complete",
			"total", len(in.Nodes),
			"registered", len(registeredIDs),
			"skipped", len(skippedIDs),
			"failed", failed)

		return RegisterNodesOutput{RegisteredNodeIDs: registeredIDs, SkippedNodeIDs: skippedIDs}, merr
	},
)

var SeqJDRevokeJobs = fwops.NewSequence(
	"seq-jd-revoke-jobs",
	semver.MustParse("1.0.0"),
	"Revoke multiple jobs",
	func(b fwops.Bundle, deps JDOpDeps, in RevokeJobsInput) (RevokeJobsOutput, error) {
		var revokedIDs, alreadyAbsentIDs []string
		var failed int
		var merr error
		for _, jobID := range in.JobIDs {
			report, err := fwops.ExecuteOperation(b, OpJDRevokeJob, deps, RevokeJobInput{JobID: jobID})
			if err != nil {
				failed++
				merr = errors.Join(merr, err)
				b.Logger.Errorw("failed to revoke job", "job_id", jobID, "error", err)

				continue
			}
			if report.Output.AlreadyAbsent {
				alreadyAbsentIDs = append(alreadyAbsentIDs, jobID)
			} else {
				revokedIDs = append(revokedIDs, jobID)
			}
		}
		b.Logger.Infow("revoke_jobs complete",
			"total", len(in.JobIDs),
			"revoked", len(revokedIDs),
			"already_absent", len(alreadyAbsentIDs),
			"failed", failed)

		return RevokeJobsOutput{RevokedJobIDs: revokedIDs, AlreadyAbsentJobIDs: alreadyAbsentIDs}, merr
	},
)

var SeqJDUpdateNodes = fwops.NewSequence(
	"seq-jd-update-nodes",
	semver.MustParse("1.0.0"),
	"Update multiple nodes",
	func(b fwops.Bundle, deps JDOpDeps, in UpdateNodesInput) (UpdateNodesOutput, error) {
		var updatedIDs []string
		var failed int
		var merr error
		for _, n := range in.Nodes {
			_, err := fwops.ExecuteOperation(b, OpJDUpdateNode, deps, UpdateNodeInput(n))
			if err != nil {
				failed++
				merr = errors.Join(merr, err)
				b.Logger.Errorw("failed to update node", "id", n.ID, "error", err)

				continue
			}
			updatedIDs = append(updatedIDs, n.ID)
		}
		b.Logger.Infow("update_nodes complete",
			"total", len(in.Nodes),
			"updated", len(updatedIDs),
			"failed", failed)

		return UpdateNodesOutput{UpdatedNodeIDs: updatedIDs}, merr
	},
)