resource.go

package job

import (
	"context"
	"encoding/json"
	"fmt"
	"regexp"
	"strconv"
	"strings"

	"github.com/dbt-labs/terraform-provider-dbtcloud/pkg/dbt_cloud"
	"github.com/dbt-labs/terraform-provider-dbtcloud/pkg/helper"
	"github.com/dbt-labs/terraform-provider-dbtcloud/pkg/utils"
	"github.com/hashicorp/terraform-plugin-framework/path"
	"github.com/hashicorp/terraform-plugin-framework/resource"
	"github.com/hashicorp/terraform-plugin-framework/types"
)

var (
	_ resource.Resource                = &jobResource{}
	_ resource.ResourceWithConfigure   = &jobResource{}
	_ resource.ResourceWithImportState = &jobResource{}
	_ resource.ResourceWithModifyPlan  = &jobResource{}
)

// Job type constants matching the server-side JobType enum
const (
	JobTypeCI        = "ci"
	JobTypeMerge     = "merge"
	JobTypeScheduled = "scheduled"
	JobTypeOther     = "other"
	JobTypeAdaptive  = "adaptive"
)

type jobResource struct {
	client *dbt_cloud.Client
}

// validateJobTypeChange validates if a job type transition is allowed.
// This mirrors the server-side validation in _validate_job_type_change.
func validateJobTypeChange(prevJobType, newJobType string) error {
	// If no change, always allowed
	if prevJobType == newJobType {
		return nil
	}

	// If previous type is empty (not set), any new type is allowed
	if prevJobType == "" {
		return nil
	}

	// CI jobs can only stay CI
	if prevJobType == JobTypeCI && newJobType != JobTypeCI {
		return fmt.Errorf("the job type field for this job can only be set to 'ci'")
	}

	// Adaptive jobs can only stay adaptive
	if prevJobType == JobTypeAdaptive && newJobType != JobTypeAdaptive {
		return fmt.Errorf("the job type field for this job can only be set to 'adaptive'")
	}

	// Scheduled jobs can only change to scheduled or other
	if prevJobType == JobTypeScheduled && (newJobType == JobTypeCI || newJobType == JobTypeAdaptive) {
		return fmt.Errorf("the job type field for this job can only be set to 'scheduled' or 'other'")
	}

	// Other jobs can only change to scheduled or other
	if prevJobType == JobTypeOther && (newJobType == JobTypeCI || newJobType == JobTypeAdaptive) {
		return fmt.Errorf("the job type field for this job can only be set to 'scheduled' or 'other'")
	}

	// Merge jobs - treating similar to CI (cannot change away from merge)
	if prevJobType == JobTypeMerge && newJobType != JobTypeMerge {
		return fmt.Errorf("the job type field for this job can only be set to 'merge'")
	}

	return nil
}

func (j *jobResource) ModifyPlan(ctx context.Context, req resource.ModifyPlanRequest, resp *resource.ModifyPlanResponse) {
	if !req.Plan.Raw.IsNull() {
		var plan JobResourceModel
		resp.Diagnostics.Append(req.Plan.Get(ctx, &plan)...)

		if plan.ValidateExecuteSteps.ValueBool() {
			executeSteps := make([]string, len(plan.ExecuteSteps))
			for i, step := range plan.ExecuteSteps {
				executeSteps[i] = step.ValueString()
			}

			if err := j.validateExecuteSteps(executeSteps); err != nil {
				resp.Diagnostics.AddError("Error validating execute steps", err.Error())
				return
			}
		}
	}

	// Don't do anything on resource creation or deletion
	if req.State.Raw.IsNull() || req.Plan.Raw.IsNull() {
		return
	}

	var plan, state JobResourceModel
	resp.Diagnostics.Append(req.Plan.Get(ctx, &plan)...)
	resp.Diagnostics.Append(req.State.Get(ctx, &state)...)

	if resp.Diagnostics.HasError() {
		return
	}

	// Skip checks if necessary fields are null
	if plan.Triggers == nil || state.Triggers == nil {
		return
	}

	// if we change the job type (CI, merge or "empty"), we need to recreate the job as dbt Cloud doesn't allow updating them
	// the job type is determined by the triggers
	if plan.Triggers != nil && state.Triggers != nil {
		oldCI := state.Triggers.GithubWebhook.ValueBool() || state.Triggers.GitProviderWebhook.ValueBool()
		oldOnMerge := state.Triggers.OnMerge.ValueBool()

		oldType := ""
		if oldCI {
			oldType = "ci"
		} else if oldOnMerge {
			oldType = "merge"
		}

		newCI := plan.Triggers.GithubWebhook.ValueBool() || plan.Triggers.GitProviderWebhook.ValueBool()
		newOnMerge := plan.Triggers.OnMerge.ValueBool()

		newType := ""
		if newCI {
			newType = "ci"
		} else if newOnMerge {
			newType = "merge"
		}

		if oldType != newType {
			resp.RequiresReplace = append(resp.RequiresReplace, path.Root("triggers"))
		}
	}

	// Validate job_type field changes if the field is being explicitly set
	// Note: If plan.JobType is set but state.JobType is null (first time setting it),
	// the validation will happen in Update against the actual server value
	// Skip validation if either value is empty (empty means "not explicitly set")
	if !plan.JobType.IsNull() && !state.JobType.IsNull() {
		prevJobType := state.JobType.ValueString()
		newJobType := plan.JobType.ValueString()

		// Only validate if both values are non-empty (explicitly set)
		if prevJobType != "" && newJobType != "" {
			if err := validateJobTypeChange(prevJobType, newJobType); err != nil {
				resp.Diagnostics.AddError(
					"Invalid job_type change",
					fmt.Sprintf("Cannot change job_type from '%s' to '%s': %s", prevJobType, newJobType, err.Error()),
				)
				return
			}
		}
	}
}

func JobResource() resource.Resource {
	return &jobResource{}
}

func (j *jobResource) ImportState(ctx context.Context, req resource.ImportStateRequest, resp *resource.ImportStateResponse) {
	jobID, err := strconv.Atoi(req.ID)
	if err != nil {
		resp.Diagnostics.AddError(
			"Error Parsing Job ID",
			fmt.Sprintf("Could not parse job_id from import ID %q: %v", req.ID, err),
		)
		return
	}

	resp.Diagnostics.Append(resp.State.SetAttribute(ctx, path.Root("id"), jobID)...)
	resp.Diagnostics.Append(resp.State.SetAttribute(ctx, path.Root("job_id"), jobID)...)
}

func (j *jobResource) Configure(_ context.Context, req resource.ConfigureRequest, _ *resource.ConfigureResponse) {
	if req.ProviderData == nil {
		return
	}

	j.client = req.ProviderData.(*dbt_cloud.Client)
}

func (j *jobResource) Create(ctx context.Context, req resource.CreateRequest, resp *resource.CreateResponse) {
	var plan JobResourceModel

	resp.Diagnostics.Append(req.Plan.Get(ctx, &plan)...)

	if resp.Diagnostics.HasError() {
		return
	}

	projectId := plan.ProjectID
	environmentId := plan.EnvironmentID
	name := plan.Name.ValueString()
	description := plan.Description.ValueString()

	executeSteps := make([]string, len(plan.ExecuteSteps))
	for i, step := range plan.ExecuteSteps {
		executeSteps[i] = step.ValueString()
	}

	var dbtVersion *string
	if !plan.DbtVersion.IsNull() {
		dbtVersionValue := plan.DbtVersion.ValueString()
		dbtVersion = &dbtVersionValue
	}

	isActive := true // when being created, the job should be active by default
	triggers := map[string]any{
		"github_webhook":       plan.Triggers.GithubWebhook.ValueBool(),
		"git_provider_webhook": plan.Triggers.GitProviderWebhook.ValueBool(),
		"schedule":             plan.Triggers.Schedule.ValueBool(),
		"on_merge":             plan.Triggers.OnMerge.ValueBool(),
	}
	numThreads := int(plan.NumThreads.ValueInt64())
	targetName := plan.TargetName.ValueString()
	generateDocs := plan.GenerateDocs.ValueBool()
	runGenerateSources := plan.RunGenerateSources.ValueBool()
	scheduleType := plan.ScheduleType.ValueString()
	scheduleInterval := int(plan.ScheduleInterval.ValueInt64())

	scheduleHours := make([]int, len(plan.ScheduleHours))
	for i, hour := range plan.ScheduleHours {
		scheduleHours[i] = int(hour.ValueInt64())
	}

	scheduleDays := make([]int, len(plan.ScheduleDays))
	for i, day := range plan.ScheduleDays {
		scheduleDays[i] = int(day.ValueInt64())
	}

	scheduleCron := plan.ScheduleCron.ValueString()

	var deferringJobId *int
	if !plan.DeferringJobId.IsNull() {
		deferringJobId = helper.Int64ToIntPointer(plan.DeferringJobId.ValueInt64())
	}

	var deferringEnvironmentID *int
	if !plan.DeferringEnvironmentID.IsNull() {
		deferringEnvironmentID = helper.Int64ToIntPointer(plan.DeferringEnvironmentID.ValueInt64())
	}

	selfDeferring := plan.SelfDeferring.ValueBool()
	// Handle timeout_seconds from either execution block (preferred) or top-level (deprecated)
	var timeoutSeconds int
	if plan.Execution != nil && !plan.Execution.TimeoutSeconds.IsNull() {
		timeoutSeconds = int(plan.Execution.TimeoutSeconds.ValueInt64())
	} else {
		if !plan.TimeoutSeconds.IsNull() {
			timeoutSeconds = int(plan.TimeoutSeconds.ValueInt64())
		}
	}
	triggersOnDraftPR := plan.TriggersOnDraftPr.ValueBool()
	runCompareChanges := plan.RunCompareChanges.ValueBool()
	runLint := plan.RunLint.ValueBool()
	errorsOnLintFailure := plan.ErrorsOnLintFailure.ValueBool()

	var jobType string
	if !plan.JobType.IsNull() {
		jobType = plan.JobType.ValueString()
	}

	compareChangesFlags := plan.CompareChangesFlags.ValueString()

	var forceNodeSelection *bool
	if !plan.ForceNodeSelection.IsNull() {
		fns := plan.ForceNodeSelection.ValueBool()
		forceNodeSelection = &fns
	}

	var jobCompletionTriggerCondition map[string]any

	if len(plan.JobCompletionTriggerCondition) != 0 {
		condition := plan.JobCompletionTriggerCondition[0]
		statuses := make([]int, len(condition.Statuses))
		for i, status := range condition.Statuses {
			statuses[i] = utils.JobCompletionTriggerConditionsMappingHumanCode[status.ValueString()]
		}
		jobCompletionTriggerCondition = map[string]any{
			"job_id":     int(condition.JobID.ValueInt64()),
			"project_id": int(condition.ProjectID.ValueInt64()),
			"statuses":   statuses,
		}
	}

	createDbtVersion := ""
	if dbtVersion != nil {
		createDbtVersion = *dbtVersion
	}
	createDeferringJobID := 0
	if deferringJobId != nil {
		createDeferringJobID = *deferringJobId
	}
	createDeferringEnvironmentID := 0
	if deferringEnvironmentID != nil {
		createDeferringEnvironmentID = *deferringEnvironmentID
	}

	createdJob, err := j.client.CreateJob(int(projectId.ValueInt64()),
		int(environmentId.ValueInt64()),
		name,
		description,
		executeSteps,
		createDbtVersion,
		isActive,
		triggers,
		numThreads,
		targetName,
		generateDocs,
		runGenerateSources,
		scheduleType,
		scheduleInterval,
		scheduleHours,
		scheduleDays,
		scheduleCron,
		createDeferringJobID,
		createDeferringEnvironmentID,
		selfDeferring,
		timeoutSeconds,
		triggersOnDraftPR,
		jobCompletionTriggerCondition,
		runCompareChanges,
		runLint,
		errorsOnLintFailure,
		jobType,
		compareChangesFlags,
		forceNodeSelection,
	)
	if err != nil {
		resp.Diagnostics.AddError(
			"Error creating job",
			"Could not create job, unexpected error: "+err.Error(),
		)
		return
	}

	// Defensive check: ensure createdJob and its ID are not nil before dereferencing
	if createdJob == nil {
		resp.Diagnostics.AddError(
			"Error creating job",
			"Job creation returned nil response without an error. This may indicate a permissions issue or an API problem.",
		)
		return
	}

	if createdJob.ID == nil {
		resp.Diagnostics.AddError(
			"Error creating job",
			"Job creation returned a response without a job ID. This may indicate a permissions issue or an API problem.",
		)
		return
	}

	plan.ID = types.Int64Value(int64(*createdJob.ID))
	plan.JobId = types.Int64Value(int64(*createdJob.ID))

	// Populate execution block only if user configured it, otherwise keep it nil to avoid state drift
	if plan.Execution != nil {
		plan.Execution = &JobExecution{
			TimeoutSeconds: types.Int64Value(int64(createdJob.Execution.TimeoutSeconds)),
		}
		// Don't update deprecated timeout_seconds when user is using execution block
		// to avoid "inconsistent result after apply" error
	} else {
		// Only update deprecated timeout_seconds when user is NOT using execution block
		plan.TimeoutSeconds = types.Int64Value(int64(createdJob.Execution.TimeoutSeconds))
	}

	if createdJob.JobType != "" {
		plan.JobType = types.StringValue(createdJob.JobType)
	} else {
		plan.JobType = types.StringNull()
	}

	// Populate force_node_selection from API response
	if createdJob.ForceNodeSelection != nil {
		plan.ForceNodeSelection = types.BoolValue(*createdJob.ForceNodeSelection)
	} else {
		// If not set in config and API doesn't return it, keep it null
		if plan.ForceNodeSelection.IsNull() {
			plan.ForceNodeSelection = types.BoolNull()
		}
	}

	jobIDStr := strconv.FormatInt(int64(*createdJob.ID), 10)

	// Check if DeferringJobId is set and matches this job's ID for self-deferring
	createdSelfDeferring := false
	if createdJob.DeferringJobId != nil {
		createdSelfDeferring = strconv.Itoa(*createdJob.DeferringJobId) == jobIDStr
	}
	plan.SelfDeferring = types.BoolValue(createdSelfDeferring)

	diags := resp.State.Set(ctx, plan)
	resp.Diagnostics.Append(diags...)
	if resp.Diagnostics.HasError() {
		return
	}
}

func (j *jobResource) Delete(ctx context.Context, req resource.DeleteRequest, resp *resource.DeleteResponse) {
	var state JobResourceModel
	resp.Diagnostics.Append(req.State.Get(ctx, &state)...)
	if resp.Diagnostics.HasError() {
		return
	}

	// Ensure ID is not null before accessing
	if state.ID.IsNull() {
		resp.Diagnostics.AddError("Client Error", "Job ID is null")
		return
	}

	jobID := state.ID.ValueInt64()
	jobIDStr := strconv.FormatInt(jobID, 10)

	job, err := j.client.GetJob(jobIDStr)
	if err != nil {
		if helper.HandleResourceNotFound(ctx, err, &resp.Diagnostics, &resp.State, "job") {
			return
		}
		resp.Diagnostics.AddError("Client Error", "Unable to retrieve job before deletion: "+err.Error())
		return
	}

	job.State = dbt_cloud.STATE_DELETED
	_, err = j.client.UpdateJob(jobIDStr, *job)
	if err != nil {
		resp.Diagnostics.AddError("Client Error", "Unable to delete job: "+err.Error())
		return
	}
}

func (j *jobResource) Metadata(_ context.Context, req resource.MetadataRequest, resp *resource.MetadataResponse) {
	resp.TypeName = req.ProviderTypeName + "_job"
}

func (j *jobResource) Read(ctx context.Context, req resource.ReadRequest, resp *resource.ReadResponse) {
	var state JobResourceModel

	resp.Diagnostics.Append(req.State.Get(ctx, &state)...)

	jobID := state.ID.ValueInt64()
	jobIDStr := strconv.FormatInt(jobID, 10)

	retrievedJob, err := j.client.GetJob(jobIDStr)

	if err != nil {
		if helper.HandleResourceNotFound(ctx, err, &resp.Diagnostics, &resp.State, "job") {
			return
		}
		resp.Diagnostics.AddError("Error getting the job", err.Error())
		return
	}

	state.ID = types.Int64Value(int64(*retrievedJob.ID))
	state.JobId = types.Int64Value(int64(*retrievedJob.ID))
	state.ProjectID = types.Int64Value(int64(retrievedJob.ProjectId))
	state.EnvironmentID = types.Int64Value(int64(retrievedJob.EnvironmentId))
	state.Name = types.StringValue(retrievedJob.Name)
	state.Description = types.StringValue(retrievedJob.Description)
	state.ExecuteSteps = helper.SliceStringToSliceTypesString(retrievedJob.ExecuteSteps)

	if retrievedJob.DbtVersion != nil {
		state.DbtVersion = types.StringValue(*retrievedJob.DbtVersion)
	} else {
		state.DbtVersion = types.StringNull()
	}

	state.IsActive = types.BoolValue(retrievedJob.State == 1)
	state.NumThreads = types.Int64Value(int64(retrievedJob.Settings.Threads))
	state.TargetName = types.StringValue(retrievedJob.Settings.TargetName)
	state.GenerateDocs = types.BoolValue(retrievedJob.GenerateDocs)
	state.RunGenerateSources = types.BoolValue(retrievedJob.RunGenerateSources)
	state.ScheduleType = types.StringValue(retrievedJob.Schedule.Date.Type)

	schedule := 1
	if retrievedJob.Schedule.Date.Type == "interval_cron" && retrievedJob.Schedule.Date.Cron != nil {
		// For interval_cron, parse the interval from the cron expression (e.g., "4 */5 * * 0,1,2,3,4,5,6")
		cronParts := strings.Split(*retrievedJob.Schedule.Date.Cron, " ")
		if len(cronParts) >= 2 && strings.HasPrefix(cronParts[1], "*/") {
			if intervalVal, err := strconv.Atoi(strings.TrimPrefix(cronParts[1], "*/")); err == nil {
				schedule = intervalVal
			}
		}
	} else if retrievedJob.Schedule.Time.Interval > 0 {
		schedule = retrievedJob.Schedule.Time.Interval
	}
	state.ScheduleInterval = types.Int64Value(int64(schedule))

	if retrievedJob.Schedule.Time.Hours != nil {
		state.ScheduleHours = helper.SliceIntToSliceTypesInt64(*retrievedJob.Schedule.Time.Hours)
	} else {
		var scheduleHoursNull []types.Int64
		state.ScheduleHours = scheduleHoursNull
	}

	if retrievedJob.Schedule.Date.Days != nil {
		state.ScheduleDays = helper.SliceIntToSliceTypesInt64(*retrievedJob.Schedule.Date.Days)
	} else {
		var scheduleDaysNull []types.Int64
		state.ScheduleDays = scheduleDaysNull
	}

	if retrievedJob.Schedule.Date.Cron != nil &&
		retrievedJob.Schedule.Date.Type != "interval_cron" { // for interval_cron, the cron expression is auto generated in the code
		state.ScheduleCron = types.StringValue(*retrievedJob.Schedule.Date.Cron)
	} else {
		state.ScheduleCron = types.StringNull()
	}

	selfDeferring := retrievedJob.DeferringJobId != nil && strconv.Itoa(*retrievedJob.DeferringJobId) == jobIDStr

	if retrievedJob.DeferringJobId != nil && !selfDeferring {
		state.DeferringJobId = types.Int64Value(int64(*retrievedJob.DeferringJobId))
	} else {
		state.DeferringJobId = types.Int64Null()
	}

	if retrievedJob.DeferringEnvironmentId != nil {
		state.DeferringEnvironmentID = types.Int64Value(int64(*retrievedJob.DeferringEnvironmentId))
	} else {
		state.DeferringEnvironmentID = types.Int64Null()
	}

	state.SelfDeferring = types.BoolValue(selfDeferring)

	// Populate execution block only if it was already in state, otherwise keep it nil to avoid state drift
	if state.Execution != nil {
		state.Execution = &JobExecution{
			TimeoutSeconds: types.Int64Value(int64(retrievedJob.Execution.TimeoutSeconds)),
		}
		// Don't update deprecated timeout_seconds when user is using execution block
		// to avoid "inconsistent result after apply" error
	} else {
		// Only update deprecated timeout_seconds when user is NOT using execution block
		state.TimeoutSeconds = types.Int64Value(int64(retrievedJob.Execution.TimeoutSeconds))
	}

	var triggers map[string]interface{}
	triggersInput, _ := json.Marshal(retrievedJob.Triggers)
	json.Unmarshal(triggersInput, &triggers)

	// for now, we allow people to keep the triggers.custom_branch_only config even if the parameter was deprecated in the API
	// we set the state to the current config value, so it doesn't do anything
	var customBranchValue types.Bool
	diags := req.State.GetAttribute(ctx, path.Root("triggers").AtMapKey("custom_branch_only"), &customBranchValue)

	if !diags.HasError() && !customBranchValue.IsNull() {
		triggers["custom_branch_only"] = customBranchValue.ValueBool()
	}

	// we remove triggers.on_merge if it is not set in the config and it is set to false in the remote
	// that way it works if people don't define it, but also works to import jobs that have it set to true
	// TODO: remove this when we make on_merge mandatory
	var onMergeValue types.Bool
	hasOnMergeAttr := !req.State.GetAttribute(ctx, path.Root("triggers").AtMapKey("on_merge"), &onMergeValue).HasError()
	noOnMergeConfig := !hasOnMergeAttr || onMergeValue.IsNull()

	onMergeRemoteVal, _ := triggers["on_merge"].(bool)
	onMergeRemoteFalse := !onMergeRemoteVal

	if noOnMergeConfig && onMergeRemoteFalse {
		delete(triggers, "on_merge")
	}

	state.Triggers = &JobTriggers{
		GithubWebhook:      types.BoolValue(retrievedJob.Triggers.GithubWebhook),
		GitProviderWebhook: types.BoolValue(retrievedJob.Triggers.GitProviderWebhook),
		Schedule:           types.BoolValue(retrievedJob.Triggers.Schedule),
		OnMerge:            types.BoolValue(retrievedJob.Triggers.OnMerge),
	}

	state.TriggersOnDraftPr = types.BoolValue(retrievedJob.TriggersOnDraftPR)
	if retrievedJob.JobCompletionTrigger != nil {
		// Only sync from API if this resource is managing job_completion_trigger_condition.
		// If prior state had it nil, a separate dbtcloud_job_completion_trigger resource
		// may own the trigger — leave state untouched to avoid spurious drift.
		if len(state.JobCompletionTriggerCondition) > 0 {
			statusesStr := make([]types.String, 0)
			for _, status := range retrievedJob.JobCompletionTrigger.Condition.Statuses {
				statusStr := utils.JobCompletionTriggerConditionsMappingCodeHuman[status].(string)
				statusesStr = append(statusesStr, types.StringValue(statusStr))
			}

			state.JobCompletionTriggerCondition = []*JobCompletionTriggerCondition{
				{
					JobID:     types.Int64Value(int64(retrievedJob.JobCompletionTrigger.Condition.JobID)),
					ProjectID: types.Int64Value(int64(retrievedJob.JobCompletionTrigger.Condition.ProjectID)),
					Statuses:  statusesStr,
				},
			}
		}
	} else {
		state.JobCompletionTriggerCondition = nil
	}

	state.RunCompareChanges = types.BoolValue(retrievedJob.RunCompareChanges)
	state.CompareChangesFlags = types.StringValue(retrievedJob.CompareChangesFlags)
	state.RunLint = types.BoolValue(retrievedJob.RunLint)
	state.ErrorsOnLintFailure = types.BoolValue(retrievedJob.ErrorsOnLintFailure)

	if retrievedJob.ForceNodeSelection != nil {
		state.ForceNodeSelection = types.BoolValue(*retrievedJob.ForceNodeSelection)
	} else {
		state.ForceNodeSelection = types.BoolNull()
	}

	if retrievedJob.JobType != "" {
		state.JobType = types.StringValue(retrievedJob.JobType)
	} else {
		state.JobType = types.StringNull()
	}

	if state.SelfDeferring.IsNull() {
		state.SelfDeferring = types.BoolValue(selfDeferring)
	}

	resp.Diagnostics.Append(resp.State.Set(ctx, &state)...)
}

func (j *jobResource) Update(ctx context.Context, req resource.UpdateRequest, resp *resource.UpdateResponse) {
	var plan, state JobResourceModel

	resp.Diagnostics.Append(req.Plan.Get(ctx, &plan)...)
	resp.Diagnostics.Append(req.State.Get(ctx, &state)...)

	if resp.Diagnostics.HasError() {
		return
	}

	jobID := state.ID.ValueInt64()
	jobIDStr := strconv.FormatInt(jobID, 10)

	job, err := j.client.GetJob(jobIDStr)
	if err != nil {
		resp.Diagnostics.AddError(
			"Error retrieving job",
			"Could not retrieve job with ID "+jobIDStr+": "+err.Error(),
		)
		return
	}

	job.ProjectId = int(plan.ProjectID.ValueInt64())
	job.EnvironmentId = int(plan.EnvironmentID.ValueInt64())
	job.Name = plan.Name.ValueString()
	job.Description = plan.Description.ValueString()

	if plan.DbtVersion.IsNull() {
		job.DbtVersion = nil
	} else {
		dbtVersion := plan.DbtVersion.ValueString()
		job.DbtVersion = &dbtVersion
	}

	job.Settings.Threads = int(plan.NumThreads.ValueInt64())
	job.Settings.TargetName = plan.TargetName.ValueString()
	job.RunGenerateSources = plan.RunGenerateSources.ValueBool()
	job.GenerateDocs = plan.GenerateDocs.ValueBool()

	executeSteps := make([]string, len(plan.ExecuteSteps))
	for i, step := range plan.ExecuteSteps {
		executeSteps[i] = step.ValueString()
	}
	job.ExecuteSteps = executeSteps

	// todo check if trigger handling is ok
	if plan.Triggers != nil {
		job.Triggers.GithubWebhook = plan.Triggers.GithubWebhook.ValueBool()
		job.Triggers.GitProviderWebhook = plan.Triggers.GitProviderWebhook.ValueBool()
		job.Triggers.Schedule = plan.Triggers.Schedule.ValueBool()
		job.Triggers.OnMerge = plan.Triggers.OnMerge.ValueBool()
	}

	scheduleInterval := int(plan.ScheduleInterval.ValueInt64())
	job.Schedule.Time.Interval = scheduleInterval

	if len(plan.ScheduleHours) > 0 {
		scheduleHours := make([]int, len(plan.ScheduleHours))
		for i, hour := range plan.ScheduleHours {
			scheduleHours[i] = int(hour.ValueInt64())
		}
		job.Schedule.Time.Hours = &scheduleHours
		job.Schedule.Time.Type = "at_exact_hours"
		job.Schedule.Time.Interval = 0
	} else {
		job.Schedule.Time.Hours = nil
		job.Schedule.Time.Type = "every_hour"
		job.Schedule.Time.Interval = scheduleInterval
	}

	if len(plan.ScheduleDays) > 0 {
		scheduleDays := make([]int, len(plan.ScheduleDays))
		for i, day := range plan.ScheduleDays {
			scheduleDays[i] = int(day.ValueInt64())
		}
		job.Schedule.Date.Days = &scheduleDays
	} else {
		job.Schedule.Date.Days = nil
	}

	if plan.ScheduleCron.IsNull() || plan.ScheduleCron.ValueString() == "" {
		job.Schedule.Date.Cron = nil
	} else {
		scheduleCron := plan.ScheduleCron.ValueString()
		job.Schedule.Date.Cron = &scheduleCron
	}

	// we set this after the subfields to remove the fields not matching the schedule type
	// if it was before, some of those fields would be set again
	scheduleType := plan.ScheduleType.ValueString()
	job.Schedule.Date.Type = scheduleType

	if scheduleType == "days_of_week" || scheduleType == "every_day" {
		job.Schedule.Date.Cron = nil
	}
	if scheduleType == "custom_cron" || scheduleType == "every_day" {
		job.Schedule.Date.Days = nil
	}
	if scheduleType == "interval_cron" {
		// For interval_cron, build the cron expression like CreateJob does
		daysStr := make([]string, len(plan.ScheduleDays))
		for i, day := range plan.ScheduleDays {
			daysStr[i] = strconv.Itoa(int(day.ValueInt64()))
		}
		cronExpr := fmt.Sprintf("4 */%d * * %s", scheduleInterval, strings.Join(daysStr, ","))
		job.Schedule.Date.Cron = &cronExpr
	}

	if plan.DeferringEnvironmentID.IsNull() || plan.DeferringEnvironmentID.ValueInt64() == 0 {
		job.DeferringEnvironmentId = nil
	} else {
		deferringEnvId := int(plan.DeferringEnvironmentID.ValueInt64())
		job.DeferringEnvironmentId = &deferringEnvId
	}

	// If self_deferring has been toggled to true, set deferring_job_id as own ID
	// Otherwise, set it back to what deferring_job_id specifies it to be
	selfDeferring := plan.SelfDeferring.ValueBool()
	if selfDeferring {
		deferringJobID := int(jobID)
		job.DeferringJobId = &deferringJobID
	} else {
		if plan.DeferringJobId.IsNull() || plan.DeferringJobId.ValueInt64() == 0 {
			job.DeferringJobId = nil
		} else {
			deferringJobId := int(plan.DeferringJobId.ValueInt64())
			job.DeferringJobId = &deferringJobId
		}
	}

	// Handle timeout_seconds from either execution block (preferred) or top-level (deprecated)
	if plan.Execution != nil && !plan.Execution.TimeoutSeconds.IsNull() {
		job.Execution.TimeoutSeconds = int(plan.Execution.TimeoutSeconds.ValueInt64())
	} else {
		job.Execution.TimeoutSeconds = int(plan.TimeoutSeconds.ValueInt64())
	}
	job.TriggersOnDraftPR = plan.TriggersOnDraftPr.ValueBool()

	if len(plan.JobCompletionTriggerCondition) == 0 {
		job.JobCompletionTrigger = nil
	} else {
		condition := plan.JobCompletionTriggerCondition[0]
		statuses := make([]int, len(condition.Statuses))
		for i, status := range condition.Statuses {
			statuses[i] = utils.JobCompletionTriggerConditionsMappingHumanCode[status.ValueString()]
		}
		jobCondTrigger := dbt_cloud.JobCompletionTrigger{
			Condition: dbt_cloud.JobCompletionTriggerCondition{
				JobID:     int(condition.JobID.ValueInt64()),
				ProjectID: int(condition.ProjectID.ValueInt64()),
				Statuses:  statuses,
			},
		}
		job.JobCompletionTrigger = &jobCondTrigger
	}

	job.RunCompareChanges = plan.RunCompareChanges.ValueBool()
	job.RunLint = plan.RunLint.ValueBool()
	job.ErrorsOnLintFailure = plan.ErrorsOnLintFailure.ValueBool()
	job.CompareChangesFlags = plan.CompareChangesFlags.ValueString()

	if plan.ForceNodeSelection.IsNull() {
		job.ForceNodeSelection = nil
	} else {
		fns := plan.ForceNodeSelection.ValueBool()
		job.ForceNodeSelection = &fns
	}

	// Handle job_type updates with validation
	// Only validate and set if the plan has an explicit non-empty job_type value
	if !plan.JobType.IsNull() && plan.JobType.ValueString() != "" {
		newJobType := plan.JobType.ValueString()
		prevJobType := job.JobType // This is the current value from the API

		// Validate the job type change
		if err := validateJobTypeChange(prevJobType, newJobType); err != nil {
			resp.Diagnostics.AddError(
				"Invalid job_type change",
				fmt.Sprintf("Cannot change job_type from '%s' to '%s': %s", prevJobType, newJobType, err.Error()),
			)
			return
		}

		job.JobType = newJobType
	}

	// Capture what's changing for better error messages
	oldEnvID := state.EnvironmentID.ValueInt64()
	newEnvID := plan.EnvironmentID.ValueInt64()
	oldName := state.Name.ValueString()
	newName := plan.Name.ValueString()

	updatedJob, err := j.client.UpdateJob(jobIDStr, *job)
	if err != nil {
		// Build a well-formatted, context-aware error message
		var errorMsg strings.Builder
		errorMsg.WriteString(fmt.Sprintf("Could not update job with ID %s\n\n", jobIDStr))
		errorMsg.WriteString(fmt.Sprintf("Error: %s\n", err.Error()))

		// Add context about what was being changed
		var changes []string
		if oldEnvID != newEnvID {
			changes = append(changes, fmt.Sprintf("  • environment_id: %d → %d", oldEnvID, newEnvID))
		}
		if oldName != newName {
			changes = append(changes, fmt.Sprintf("  • name: '%s' → '%s'", oldName, newName))
		}

		if len(changes) > 0 {
			errorMsg.WriteString("\nAttempted changes:\n")
			errorMsg.WriteString(strings.Join(changes, "\n"))

			// If environment is changing and it's a permission error, add extra context
			if oldEnvID != newEnvID && (strings.Contains(err.Error(), "permission") || strings.Contains(err.Error(), "forbidden") || strings.Contains(err.Error(), "resource-not-found-permissions")) {
				errorMsg.WriteString(fmt.Sprintf("\n\nℹ️  Note: The API token may not have write access to environment %d.\nEnvironment-level permissions are required to move jobs between environments.", newEnvID))
			}
		}

		resp.Diagnostics.AddError(
			"Error updating job",
			errorMsg.String(),
		)
		return
	}

	if updatedJob.JobType != "" {
		plan.JobType = types.StringValue(updatedJob.JobType)
	} else {
		plan.JobType = types.StringNull()
	}

	// Populate execution block only if user configured it, otherwise keep it nil to avoid state drift
	if plan.Execution != nil {
		plan.Execution = &JobExecution{
			TimeoutSeconds: types.Int64Value(int64(updatedJob.Execution.TimeoutSeconds)),
		}
		// Don't update deprecated timeout_seconds when user is using execution block
		// to avoid "inconsistent result after apply" error
	} else {
		// Only update deprecated timeout_seconds when user is NOT using execution block
		plan.TimeoutSeconds = types.Int64Value(int64(updatedJob.Execution.TimeoutSeconds))
	}

	// Populate force_node_selection from API response
	if updatedJob.ForceNodeSelection != nil {
		plan.ForceNodeSelection = types.BoolValue(*updatedJob.ForceNodeSelection)
	} else {
		plan.ForceNodeSelection = types.BoolNull()
	}

	updatedJobIDStr := strconv.FormatInt(jobID, 10)
	updatedSelfDeferring := updatedJob.DeferringJobId != nil && strconv.Itoa(*updatedJob.DeferringJobId) == updatedJobIDStr
	plan.SelfDeferring = types.BoolValue(updatedSelfDeferring)

	diags := resp.State.Set(ctx, plan)
	resp.Diagnostics.Append(diags...)
	if resp.Diagnostics.HasError() {
		return
	}
}

func (j *jobResource) validateExecuteSteps(executeSteps []string) error {
	dbt_flags := []string{
		"--warn-error",
		"--use-experimental-parser",
		"--no-partial-parse",
		"--fail-fast",
	}

	dbt_commands := []string{
		"run",
		"test",
		"archive",
		"snapshot",
		"seed",
		"source",
		"compile",
		"ls",
		"list",
		`docs\s+generate`,
		"parse",
		"build",
		"clone",
		"debug",
		"retry",
		"compare",
		"sl",
	}

	// Build regex pattern for valid commands
	flagsPattern := strings.Join(dbt_flags, "|")
	commandsPattern := strings.Join(dbt_commands, "|")
	validCommandsPattern := fmt.Sprintf(`^\s*dbt\s+((%s)\s+)*(%s)\s*.*$`, flagsPattern, commandsPattern)
	validCommandsRegex := regexp.MustCompile(validCommandsPattern)

	// Validate each execute step individually
	for _, step := range executeSteps {
		// Normalize the step by replacing newlines with spaces to support multi-line commands
		// This matches how dbt Cloud UI handles commands that span multiple lines
		normalizedStep := strings.ReplaceAll(step, "\n", " ")
		normalizedStep = strings.ReplaceAll(normalizedStep, "\r", " ")

		// Check if step matches valid dbt command pattern
		if !validCommandsRegex.MatchString(normalizedStep) {
			return fmt.Errorf("invalid command: %s. Allowed commands are: %s", step, strings.Join(dbt_commands, ", "))
		}

		// Check that each flag isn't used more than once within this step
		for _, flag := range dbt_flags {
			if strings.Count(step, flag) > 1 {
				return fmt.Errorf("flag %s can only be used once per step in: %s", flag, step)
			}
		}
	}

	return nil
}