package br

import (
	"context"
	"crypto/sha256"
	"errors"
	"fmt"
	"net/http"
	"strings"
	"time"

	v1alpha1br "github.com/pingcap/tidb-operator/api/v2/br/v1alpha1"
	"github.com/tidbcloud/cluster-service-ng/domain/entity"
	"github.com/tidbcloud/cluster-service-ng/domain/workflow"
	brutil "github.com/tidbcloud/cluster-service-ng/pkg/br"
	"github.com/tidbcloud/cluster-service-ng/pkg/clustergateway"
	"github.com/tidbcloud/cluster-service-ng/pkg/dataplane"
	"github.com/tidbcloud/cluster-service-ng/pkg/runtime/actions"
	"github.com/tidbcloud/cluster-service-ng/pkg/storage/repo"
	"github.com/tidbcloud/cluster-service-ng/pkg/tioperator"
	"github.com/tidbcloud/cluster-service-ng/pkg/utils"
	"google.golang.org/grpc/codes"
	grpcstatus "google.golang.org/grpc/status"
	batchv1 "k8s.io/api/batch/v1"
	corev1 "k8s.io/api/core/v1"
	apierrors "k8s.io/apimachinery/pkg/api/errors"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"k8s.io/apimachinery/pkg/types"
	"sigs.k8s.io/controller-runtime/pkg/client"
)

const (
	TriggerArchiveAfterResumeActionName = "trigger-archive-after-resume"
	archiveBaselineAnnotation           = "tidbcloud.pingcap.com/pause-archive-baseline"
	archiveRequiredDateAnnotation       = "tidbcloud.pingcap.com/pause-archive-required-date"
	archiveNoBaseline                   = "none"
	archiveCoverageJobSuffix            = "-coverage"
	archiveExpirationDateOption         = "--expiration-date"
	archiveExpirationDateLayout         = "20060102"
)

// triggerArchiveAfterResumeAction creates one Job from TiBRGC's existing T3
// CronJob template. Keeping the template intact is important: it preserves the
// operator-managed archive command, credentials, volumes, and resource config.
type triggerArchiveAfterResumeAction struct {
	sharedPoolRepo      repo.SharedPoolRepo
	backupGcSettingRepo repo.BackupGcSettingRepo
	dataplaneManager    dataplane.Manager
	clusterGateway      clustergateway.Client
	guard               actions.ActionGuard
}

func NewTriggerArchiveAfterResumeAction(sharedPoolRepo repo.SharedPoolRepo, backupGcSettingRepo repo.BackupGcSettingRepo, dataplaneManager dataplane.Manager, clusterGateway clustergateway.Client, guard actions.ActionGuard) actions.Action {
	asyncAction := &triggerArchiveAfterResumeAction{
		sharedPoolRepo: sharedPoolRepo, backupGcSettingRepo: backupGcSettingRepo, dataplaneManager: dataplaneManager, clusterGateway: clusterGateway, guard: guard,
	}
	return actions.CreateGuardedActionFromAsync(asyncAction, 10*time.Second, 60*time.Minute, guard)
}

func (a *triggerArchiveAfterResumeAction) Name() string { return TriggerArchiveAfterResumeActionName }

func (a *triggerArchiveAfterResumeAction) Start(args actions.ActionArgs) error {
	_, skip, err := a.loadGCSetting(utils.GenerateContext(args), args)
	if err != nil || skip {
		return a.ignoreRetryableStartError(err)
	}
	return a.ignoreRetryableStartError(a.ensureArchiveJob(args))
}

// ensureArchiveJob is called both at Start and when polling observes that a
// transient creation failure left no Job. Retryable failures are kept inside
// the async action so its 60-minute deadline remains effective.
func (a *triggerArchiveAfterResumeAction) ensureArchiveJob(args actions.ActionArgs) error {
	ctx := utils.GenerateContext(args)
	ctx, cancel := context.WithTimeout(ctx, actions.GuardedWriteTimeout)
	defer cancel()
	sharedPool, remoteClient, jobName, err := a.loadContext(ctx, args)
	if err != nil {
		return err
	}
	baseline := ""
	coverageBaseline := ""
	existing := &batchv1.Job{}
	key := types.NamespacedName{Namespace: sharedPool.GetDataPlaneNamespace(), Name: jobName}
	switch err := remoteClient.Get(ctx, key, existing); {
	case err == nil:
		baseline, err = archiveJobBaseline(existing)
		if err != nil {
			return err
		}
		if !archiveJobNeedsCoverageUpgrade(existing, baseline) {
			return nil
		}
		jobName = archiveCoverageJobName(jobName)
		status, statusErr := a.clusterGateway.GetBackupStatus(ctx, sharedPool, tioperator.DefaultTiBRName)
		if statusErr != nil {
			return classifyBackupStatusError("before upgrading resume archive coverage", statusErr)
		}
		if status == nil {
			return actions.NewTerminalError(errors.New("backup status is empty before upgrading resume archive coverage"))
		}
		coverageBaseline, err = legacyArchiveCoverageBaseline(status, baseline)
		if err != nil {
			return err
		}
	case apierrors.IsNotFound(err):
		status, statusErr := a.clusterGateway.GetBackupStatus(ctx, sharedPool, tioperator.DefaultTiBRName)
		if statusErr != nil {
			return classifyBackupStatusError("before triggering archive", statusErr)
		}
		if status == nil {
			return actions.NewTerminalError(errors.New("backup status is empty before triggering archive"))
		}
		baseline = archiveBaseline(status.LastAutoBackupTime)
		coverageBaseline = archiveCoverageBaseline(status)
	case err != nil:
		return classifyKubernetesError(fmt.Errorf("get one-off T3 archive job %s: %w", jobName, err))
	}
	cronJob := &batchv1.CronJob{}
	cronJobName := t3ArchiveCronJobName()
	if err := remoteClient.Get(ctx, types.NamespacedName{Namespace: sharedPool.GetDataPlaneNamespace(), Name: cronJobName}, cronJob); err != nil {
		if apierrors.IsNotFound(err) {
			return actions.NewTerminalError(fmt.Errorf("T3 archive cronjob %s is missing: %w", cronJobName, err))
		}
		return fmt.Errorf("get T3 archive cronjob %s: %w", cronJobName, err)
	}
	job, err := newArchiveJob(cronJob, sharedPool.GetDataPlaneNamespace(), jobName, baseline, coverageBaseline)
	if err != nil {
		return err
	}
	if err := remoteClient.Create(ctx, job); err != nil && !apierrors.IsAlreadyExists(err) {
		return classifyKubernetesError(fmt.Errorf("create one-off T3 archive job %s: %w", jobName, err))
	}
	return nil
}

func (a *triggerArchiveAfterResumeAction) CheckFinish(args actions.ActionArgs) (bool, error) {
	ctx := utils.GenerateContext(args)
	gcSetting, skip, err := a.loadGCSetting(ctx, args)
	if err != nil {
		return false, a.ignoreRetryableStartError(err)
	}
	if skip {
		return true, nil
	}
	sharedPool, remoteClient, jobName, err := a.loadContext(ctx, args)
	if err != nil {
		return false, a.ignoreRetryableStartError(err)
	}
	job := &batchv1.Job{}
	if err := remoteClient.Get(ctx, types.NamespacedName{Namespace: sharedPool.GetDataPlaneNamespace(), Name: jobName}, job); err != nil {
		if apierrors.IsNotFound(err) {
			// Recreate through the owner guard without leaving this async
			// invocation, so repeated Job loss cannot refresh the 60-minute
			// deadline indefinitely.
			err = a.guard(args, func() error { return a.ensureArchiveJob(args) })
			return false, a.ignoreRetryableStartError(err)
		}
		return false, a.ignoreRetryableStartError(classifyKubernetesError(fmt.Errorf("get one-off T3 archive job %s: %w", jobName, err)))
	}
	baseline, err := archiveJobBaseline(job)
	if err != nil {
		return false, err
	}
	if archiveJobNeedsCoverageUpgrade(job, baseline) {
		jobName = archiveCoverageJobName(jobName)
		job = &batchv1.Job{}
		if err := remoteClient.Get(ctx, types.NamespacedName{Namespace: sharedPool.GetDataPlaneNamespace(), Name: jobName}, job); err != nil {
			if apierrors.IsNotFound(err) {
				err = a.guard(args, func() error { return a.ensureArchiveJob(args) })
				return false, a.ignoreRetryableStartError(err)
			}
			return false, a.ignoreRetryableStartError(classifyKubernetesError(fmt.Errorf("get one-off T3 archive coverage job %s: %w", jobName, err)))
		}
	}
	if job.Spec.Suspend != nil && *job.Spec.Suspend {
		ready, err := a.hasNewAutoBackup(ctx, sharedPool, job)
		if err != nil || !ready {
			return false, err
		}
		err = a.guard(args, func() error {
			writeCtx, cancel := context.WithTimeout(utils.GenerateContext(args), actions.GuardedWriteTimeout)
			defer cancel()
			latest := &batchv1.Job{}
			key := types.NamespacedName{Namespace: job.Namespace, Name: job.Name}
			if err := remoteClient.Get(writeCtx, key, latest); err != nil {
				return classifyKubernetesError(fmt.Errorf("get suspended one-off T3 archive job %s: %w", jobName, err))
			}
			if latest.Spec.Suspend == nil || !*latest.Spec.Suspend {
				return nil
			}
			patch := client.MergeFrom(latest.DeepCopy())
			suspend := false
			latest.Spec.Suspend = &suspend
			if err := remoteClient.Patch(writeCtx, latest, patch); err != nil {
				return classifyKubernetesError(fmt.Errorf("start one-off T3 archive job %s: %w", jobName, err))
			}
			return nil
		})
		return false, a.ignoreRetryableStartError(err)
	}
	for _, condition := range job.Status.Conditions {
		switch condition.Type {
		case batchv1.JobComplete:
			if condition.Status == corev1.ConditionTrue {
				return a.archiveAllowsLifecycleRestore(ctx, sharedPool, gcSetting, remoteClient, job)
			}
		case batchv1.JobFailed:
			if condition.Status == corev1.ConditionTrue {
				return false, actions.NewTerminalError(fmt.Errorf("one-off T3 archive job %s failed: %s", jobName, condition.Message))
			}
		}
	}
	return false, nil
}

func (a *triggerArchiveAfterResumeAction) loadGCSetting(ctx context.Context, args actions.ActionArgs) (*entity.BackupGcSetting, bool, error) {
	sharedPoolID, err := args.GetString(workflow.SharedPoolIdKey)
	if err != nil {
		return nil, false, actions.NewTerminalError(err)
	}
	gcSetting, err := a.backupGcSettingRepo.Load(ctx, sharedPoolID)
	if err == nil {
		return gcSetting, false, nil
	}
	if grpcstatus.Code(err) == codes.NotFound {
		// Early-created pools without a GC row also skip lifecycle-rule
		// application, so there is no physical expiration to gate here.
		return nil, true, nil
	}
	return nil, false, fmt.Errorf("load backup GC setting before one-off T3 archive: %w", err)
}

// hasNewAutoBackup keeps the one-off archive Job suspended until TiBR has
// produced a backup after the watermark captured when this action started.
// The archive result, rather than the watermark itself, remains the authority
// for deciding whether lifecycle expiration can be restored.
func (a *triggerArchiveAfterResumeAction) hasNewAutoBackup(ctx context.Context, sharedPool *entity.SharedPool, job *batchv1.Job) (bool, error) {
	baseline, ok := job.Annotations[archiveBaselineAnnotation]
	if !ok {
		return false, actions.NewTerminalError(fmt.Errorf("one-off T3 archive job %s is missing its backup baseline", job.Name))
	}
	status, err := a.clusterGateway.GetBackupStatus(ctx, sharedPool, tioperator.DefaultTiBRName)
	if err != nil {
		return false, a.ignoreRetryableStartError(classifyBackupStatusError("while waiting for a new auto backup", err))
	}
	if status == nil {
		return false, actions.NewTerminalError(errors.New("backup status is empty while waiting for a new auto backup"))
	}
	if baseline == archiveNoBaseline {
		return status.LastAutoBackupTime != nil, nil
	}
	baselineTime, err := time.Parse(time.RFC3339Nano, baseline)
	if err != nil {
		return false, actions.NewTerminalError(fmt.Errorf("parse one-off T3 archive job baseline %q: %w", baseline, err))
	}
	return status.LastAutoBackupTime != nil && status.LastAutoBackupTime.After(baselineTime), nil
}

func (a *triggerArchiveAfterResumeAction) archiveAllowsLifecycleRestore(ctx context.Context, sharedPool *entity.SharedPool, gcSetting *entity.BackupGcSetting, remoteClient client.Client, job *batchv1.Job) (bool, error) {
	coverageRequiredDate, err := archiveJobRequiredDate(job)
	if err != nil {
		return false, err
	}
	var coverageBaselineTime *time.Time
	if coverageRequiredDate != archiveNoBaseline {
		requiredDate, err := time.Parse(time.DateOnly, coverageRequiredDate)
		if err != nil {
			return false, actions.NewTerminalError(fmt.Errorf("parse one-off T3 archive required date %q: %w", coverageRequiredDate, err))
		}
		parsed := requiredDate.AddDate(0, 0, 1)
		coverageBaselineTime = &parsed
	}
	status, err := a.clusterGateway.GetBackupStatus(ctx, sharedPool, tioperator.DefaultTiBRName)
	if err != nil {
		return false, a.ignoreRetryableStartError(classifyBackupStatusError("after triggering archive", err))
	}
	archiveDelay, err := getT3ArchiveDelay(ctx, remoteClient, sharedPool.GetDataPlaneNamespace())
	if err != nil {
		return false, a.ignoreRetryableStartError(err)
	}
	return archiveSafetyWindowSatisfied(status, gcSetting, archiveDelay, coverageBaselineTime, time.Now().UTC())
}

func getT3ArchiveDelay(ctx context.Context, remoteClient client.Client, namespace string) (time.Duration, error) {
	tibrgc := &v1alpha1br.TiBRGC{}
	if err := remoteClient.Get(ctx, types.NamespacedName{Name: tioperator.DefaultTiBRGCRName, Namespace: namespace}, tibrgc); err != nil {
		return 0, classifyKubernetesError(fmt.Errorf("get TiBRGC for T3 archive delay: %w", err))
	}
	for _, strategy := range tibrgc.Spec.GCStrategy.TieredStrategies {
		if strategy.Name == v1alpha1br.TieredStorageStrategyNameToT3Storage {
			if strategy.TimeThresholdDays == 0 {
				return 0, actions.NewTerminalError(errors.New("TiBRGC T3 archive delay is zero"))
			}
			return time.Duration(strategy.TimeThresholdDays) * 24 * time.Hour, nil
		}
	}
	return 0, actions.NewTerminalError(errors.New("TiBRGC has no T3 archive strategy"))
}

// archiveSafetyWindowSatisfied decides whether normal object expiration can be
// restored after the one-off archive Job completes. A recent archive or one
// that covers the pause baseline is safe. An absent archive index is safe only
// while the oldest backup is still inside the physical expiration window:
// archive may legitimately have nothing old enough to index yet.
func archiveSafetyWindowSatisfied(status *clustergateway.BackupStatus, gcSetting *entity.BackupGcSetting, archiveDelay time.Duration, baseline *time.Time, now time.Time) (bool, error) {
	if status == nil {
		return false, actions.NewTerminalError(errors.New("backup status is empty"))
	}
	if status.LastAutoBackupTime == nil {
		return false, actions.NewTerminalError(errors.New("backup status lost last auto backup time after one-off archive job"))
	}
	if gcSetting == nil {
		return false, actions.NewTerminalError(errors.New("backup GC setting is empty"))
	}
	effectiveGcSetting := brutil.ArchiveSafeBackupGcSetting(gcSetting)
	ttl, enabled := earliestBRObjectExpirationTTL(effectiveGcSetting)
	if !enabled {
		return true, nil
	}
	if archiveDelay >= ttl {
		return false, actions.NewTerminalError(fmt.Errorf("TiBRGC archive delay %s must be shorter than BR object expiration TTL %s", archiveDelay, ttl))
	}
	if status.LastArchivedDate == nil {
		if status.FirstAutoBackupTime == nil {
			return false, actions.NewTerminalError(errors.New("backup status has last auto backup time but no first auto backup time"))
		}
		return now.Sub(status.FirstAutoBackupTime.UTC()) < ttl, nil
	}
	lastArchivedDate, err := time.Parse(time.DateOnly, *status.LastArchivedDate)
	if err != nil {
		return false, actions.NewTerminalError(fmt.Errorf("parse last archived date %q: %w", *status.LastArchivedDate, err))
	}
	fresh := now.Sub(lastArchivedDate.UTC()) < ttl
	if baseline == nil {
		return fresh, nil
	}
	baselineDate := time.Date(baseline.UTC().Year(), baseline.UTC().Month(), baseline.UTC().Day(), 0, 0, 0, 0, time.UTC)
	return fresh || !lastArchivedDate.UTC().Before(baselineDate), nil
}

// Use the earliest enabled expiration among the objects consumed by T3
// archive. Packed backups are restore artifacts and do not participate in
// archive progression.
func earliestBRObjectExpirationTTL(setting *entity.BackupGcSetting) (time.Duration, bool) {
	values := []*uint32{
		setting.MaxT1MetaLifetimeDays,
		setting.MaxT1WalLifetimeDays,
		setting.MaxT2SstLifetimeDays,
	}
	var days uint32
	for _, value := range values {
		if value == nil || *value == 0 {
			continue
		}
		if days == 0 || *value < days {
			days = *value
		}
	}
	if days == 0 {
		return 0, false
	}
	return time.Duration(days) * 24 * time.Hour, true
}

func (a *triggerArchiveAfterResumeAction) ignoreRetryableStartError(err error) error {
	if err == nil || actions.IsTerminalError(err) {
		return err
	}
	return nil
}

func classifyKubernetesError(err error) error {
	if apierrors.IsNotFound(err) || apierrors.IsInvalid(err) || apierrors.IsBadRequest(err) ||
		apierrors.IsForbidden(err) || apierrors.IsUnauthorized(err) || apierrors.IsMethodNotSupported(err) {
		return actions.NewTerminalError(err)
	}
	return err
}

func classifyBackupStatusError(when string, err error) error {
	var decodeErr *clustergateway.BackupStatusDecodeError
	var httpErr *clustergateway.BackupStatusHTTPError
	if errors.As(err, &decodeErr) ||
		(errors.As(err, &httpErr) && httpErr.StatusCode >= 400 && httpErr.StatusCode < 500 &&
			httpErr.StatusCode != http.StatusRequestTimeout && httpErr.StatusCode != http.StatusTooManyRequests) {
		return actions.NewTerminalError(fmt.Errorf("get backup status %s: %w", when, err))
	}
	return fmt.Errorf("get backup status %s: %w", when, err)
}

func (a *triggerArchiveAfterResumeAction) loadContext(ctx context.Context, args actions.ActionArgs) (*entity.SharedPool, client.Client, string, error) {
	sharedPoolID, err := args.GetString(workflow.SharedPoolIdKey)
	if err != nil {
		return nil, nil, "", actions.NewTerminalError(err)
	}
	operationID, err := args.GetString(workflow.PauseResumeOperationIdKey)
	if err != nil {
		return nil, nil, "", actions.NewTerminalError(err)
	}
	sharedPool, err := a.sharedPoolRepo.Load(ctx, sharedPoolID, false)
	if err != nil {
		return nil, nil, "", fmt.Errorf("load shared pool: %w", err)
	}
	remoteClient, err := a.dataplaneManager.GetClient(sharedPool.GetDataplaneName())
	if err != nil {
		return nil, nil, "", fmt.Errorf("get dataplane client: %w", err)
	}
	return sharedPool, remoteClient, t3ArchiveJobName(operationID), nil
}

func t3ArchiveCronJobName() string {
	return fmt.Sprintf("%s-to-t3-storage", tioperator.DefaultTiBRGCRName)
}

func t3ArchiveJobPrefix() string {
	return t3ArchiveCronJobName() + "-pause-"
}

func t3ArchiveJobName(operationID string) string {
	digest := sha256.Sum256([]byte(operationID))
	return fmt.Sprintf("%s%x", t3ArchiveJobPrefix(), digest[:6])
}

func newArchiveJob(cronJob *batchv1.CronJob, namespace, jobName, baseline, coverageBaseline string) (*batchv1.Job, error) {
	labels := copyStringMap(cronJob.Spec.JobTemplate.Labels)
	annotations := copyStringMap(cronJob.Spec.JobTemplate.Annotations)
	if annotations == nil {
		annotations = map[string]string{}
	}
	annotations[archiveBaselineAnnotation] = baseline
	suspend := true
	job := &batchv1.Job{
		ObjectMeta: metav1.ObjectMeta{
			Name:        jobName,
			Namespace:   namespace,
			Labels:      labels,
			Annotations: annotations,
		},
		Spec: *cronJob.Spec.JobTemplate.Spec.DeepCopy(),
	}
	requiredDate, err := configureArchiveCoverage(&job.Spec, coverageBaseline)
	if err != nil {
		return nil, err
	}
	job.Annotations[archiveRequiredDateAnnotation] = requiredDate
	job.Spec.Suspend = &suspend
	return job, nil
}

func archiveCoverageJobName(jobName string) string { return jobName + archiveCoverageJobSuffix }

func archiveJobBaseline(job *batchv1.Job) (string, error) {
	baseline, ok := job.Annotations[archiveBaselineAnnotation]
	if !ok {
		return "", actions.NewTerminalError(fmt.Errorf("one-off T3 archive job %s is missing its backup baseline", job.Name))
	}
	if baseline != archiveNoBaseline {
		if _, err := time.Parse(time.RFC3339Nano, baseline); err != nil {
			return "", actions.NewTerminalError(fmt.Errorf("parse one-off T3 archive job baseline %q: %w", baseline, err))
		}
	}
	return baseline, nil
}

func archiveJobRequiredDate(job *batchv1.Job) (string, error) {
	if requiredDate, ok := job.Annotations[archiveRequiredDateAnnotation]; ok {
		return requiredDate, nil
	}
	// Jobs created by the previous implementation have no required-date
	// annotation. A "none" baseline has no pause range to cover, so retaining
	// the TTL-based safety check is backward compatible and sufficient.
	baseline, err := archiveJobBaseline(job)
	if err != nil {
		return "", err
	}
	if baseline == archiveNoBaseline {
		return archiveNoBaseline, nil
	}
	return "", actions.NewTerminalError(fmt.Errorf("one-off T3 archive job %s is missing its required archive date", job.Name))
}

func archiveJobNeedsCoverageUpgrade(job *batchv1.Job, baseline string) bool {
	_, configured := job.Annotations[archiveRequiredDateAnnotation]
	return baseline != archiveNoBaseline && !configured
}

func configureArchiveCoverage(spec *batchv1.JobSpec, baseline string) (string, error) {
	if baseline == archiveNoBaseline {
		return archiveNoBaseline, nil
	}
	baselineTime, err := time.Parse(time.RFC3339Nano, baseline)
	if err != nil {
		return "", actions.NewTerminalError(fmt.Errorf("parse one-off T3 archive job baseline %q: %w", baseline, err))
	}
	baselineUTC := baselineTime.UTC()
	requiredExpiration := time.Date(baselineUTC.Year(), baselineUTC.Month(), baselineUTC.Day(), 0, 0, 0, 0, time.UTC).AddDate(0, 0, -1)
	formattedRequiredExpiration := requiredExpiration.Format(archiveExpirationDateLayout)
	archiveContainerIndex := -1
	for containerIndex := range spec.Template.Spec.Containers {
		container := &spec.Template.Spec.Containers[containerIndex]
		if !isT3ArchiveContainer(container) {
			continue
		}
		archiveContainerIndex = containerIndex
		for _, argv := range []*[]string{&container.Command, &container.Args} {
			found, err := configureArchiveExpirationDate(*argv, requiredExpiration)
			if err != nil {
				return "", err
			}
			if found {
				return requiredExpiration.Format(time.DateOnly), nil
			}
		}
	}
	if archiveContainerIndex < 0 {
		return "", actions.NewTerminalError(errors.New("T3 archive job template has no archive container"))
	}
	container := &spec.Template.Spec.Containers[archiveContainerIndex]
	option := archiveExpirationDateOption + "=" + formattedRequiredExpiration
	// TiDB Operator puts the complete cse-ctl invocation in Command. Preserve
	// that representation when present; Args is used by overlay/custom images.
	if len(container.Command) > 0 {
		container.Command = append(container.Command, option)
	} else {
		container.Args = append(container.Args, option)
	}
	return requiredExpiration.Format(time.DateOnly), nil
}

func isT3ArchiveContainer(container *corev1.Container) bool {
	if container.Name == string(v1alpha1br.TieredStorageStrategyNameToT3Storage) {
		return true
	}
	for _, argv := range [][]string{container.Command, container.Args} {
		for _, arg := range argv {
			if arg == "archive" {
				return true
			}
		}
	}
	return false
}

func configureArchiveExpirationDate(argv []string, requiredExpiration time.Time) (bool, error) {
	prefix := archiveExpirationDateOption + "="
	for argIndex := 0; argIndex < len(argv); argIndex++ {
		valueIndex := argIndex
		var value string
		switch {
		case strings.HasPrefix(argv[argIndex], prefix):
			value = strings.TrimPrefix(argv[argIndex], prefix)
		case argv[argIndex] == archiveExpirationDateOption && argIndex+1 < len(argv):
			valueIndex = argIndex + 1
			value = argv[valueIndex]
		default:
			continue
		}
		existingExpiration, err := time.Parse(archiveExpirationDateLayout, value)
		if err != nil {
			return false, actions.NewTerminalError(fmt.Errorf("parse T3 archive expiration date %q: %w", value, err))
		}
		effectiveExpiration := existingExpiration.UTC()
		if requiredExpiration.Before(effectiveExpiration) {
			effectiveExpiration = requiredExpiration
		}
		formatted := effectiveExpiration.Format(archiveExpirationDateLayout)
		if valueIndex == argIndex {
			argv[argIndex] = prefix + formatted
		} else {
			argv[valueIndex] = formatted
		}
		return true, nil
	}
	return false, nil
}

func archiveBaseline(lastAutoBackupTime *time.Time) string {
	if lastAutoBackupTime == nil {
		return archiveNoBaseline
	}
	return lastAutoBackupTime.UTC().Format(time.RFC3339Nano)
}

// archiveCoverageBaseline is deliberately independent of the watermark used
// to wait for a post-resume auto backup. On a retry, LastAutoBackupTime may
// already point at a backup created after compute resumed and would lose the
// still-referenced pre-pause range. Starting from FirstAutoBackupTime is
// conservative; cse-ctl discovers existing archive progress and skips data it
// has already archived.
func archiveCoverageBaseline(status *clustergateway.BackupStatus) string {
	if status == nil {
		return archiveNoBaseline
	}
	if status.FirstAutoBackupTime != nil {
		return archiveBaseline(status.FirstAutoBackupTime)
	}
	return archiveBaseline(status.LastAutoBackupTime)
}

// legacyArchiveCoverageBaseline preserves the pre-resume watermark recorded
// on a legacy Job. LastAutoBackupTime is not a safe fallback here because it
// may already refer to a backup created after resume. FirstAutoBackupTime can
// only extend the coverage range further into the past.
func legacyArchiveCoverageBaseline(status *clustergateway.BackupStatus, legacyBaseline string) (string, error) {
	if status == nil || status.FirstAutoBackupTime == nil {
		return legacyBaseline, nil
	}
	legacyBaselineTime, err := time.Parse(time.RFC3339Nano, legacyBaseline)
	if err != nil {
		return "", actions.NewTerminalError(fmt.Errorf("parse legacy archive job baseline %q: %w", legacyBaseline, err))
	}
	if status.FirstAutoBackupTime.UTC().Before(legacyBaselineTime) {
		return archiveBaseline(status.FirstAutoBackupTime), nil
	}
	return legacyBaseline, nil
}

func copyStringMap(values map[string]string) map[string]string {
	if len(values) == 0 {
		return nil
	}
	result := make(map[string]string, len(values))
	for key, value := range values {
		result[key] = value
	}
	return result
}