package minio

import (
	"context"
	"fmt"
	"log"
	"math"
	"net/url"
	"path"
	"reflect"
	"regexp"
	"strconv"
	"strings"
	"time"

	"github.com/dustin/go-humanize"
	"github.com/hashicorp/go-cty/cty"
	"github.com/hashicorp/terraform-plugin-sdk/v2/diag"
	"github.com/hashicorp/terraform-plugin-sdk/v2/helper/schema"
	"github.com/hashicorp/terraform-plugin-sdk/v2/helper/validation"
	"github.com/minio/madmin-go/v3"
	"github.com/minio/minio-go/v7/pkg/replication"
	"github.com/minio/minio-go/v7/pkg/s3utils"
	"github.com/rs/xid"
	"golang.org/x/exp/slices"
)

func resourceMinioBucketReplication() *schema.Resource {
	return &schema.Resource{
		CreateContext: minioPutBucketReplication,
		ReadContext:   minioReadBucketReplication,
		UpdateContext: minioPutBucketReplication,
		DeleteContext: minioDeleteBucketReplication,
		Importer: &schema.ResourceImporter{
			StateContext: schema.ImportStatePassthroughContext,
		},
		Schema: map[string]*schema.Schema{
			"bucket": {
				Type:        schema.TypeString,
				Description: "Name of the bucket on which to setup replication rules",
				Required:    true,
				ForceNew:    true,
			},
			"rule": {
				Type:        schema.TypeList,
				Description: "Rule definitions",
				Optional:    true,
				Elem: &schema.Resource{
					Schema: map[string]*schema.Schema{
						"id": {
							Type:        schema.TypeString,
							Description: "Rule ID generated by MinIO",
							Computed:    true,
						},
						"arn": {
							Type:        schema.TypeString,
							Description: "Rule ARN genrated by MinIO",
							Computed:    true,
						},
						"enabled": {
							Type:        schema.TypeBool,
							Description: "Whether or not this rule is enabled",
							Optional:    true,
							Default:     true,
						},
						"priority": {
							Type:         schema.TypeInt,
							Description:  "Rule priority. If omitted, the inverted index will be used as priority. This means that the first rule definition will have the higher priority",
							Optional:     true,
							ValidateFunc: validation.IntAtLeast(1),
							DiffSuppressFunc: func(k, oldValue, newValue string, d *schema.ResourceData) bool {
								oldVal, _ := strconv.Atoi(oldValue)
								newVal, _ := strconv.Atoi(newValue)
								return oldVal == 0 && newVal == 0 || oldVal == newVal
							},
						},
						"prefix": {
							Type:        schema.TypeString,
							Description: "Bucket prefix object must be in to be syncronised",
							Optional:    true,
							Default:     "",
						},
						"tags": {
							Type:        schema.TypeMap,
							Description: "Tags which objects must have to be syncronised",
							Optional:    true,
							ValidateDiagFunc: validation.AllDiag(
								validation.MapValueMatch(regexp.MustCompile(`^[a-zA-Z0-9-+\-._:/@ ]+$`), ""),
								validation.MapKeyMatch(regexp.MustCompile(`^[a-zA-Z0-9-+\-._:/@ ]+$`), ""),
								validation.MapValueLenBetween(1, 256),
								validation.MapKeyLenBetween(1, 128),
							),
						},
						"delete_replication": {
							Type:        schema.TypeBool,
							Description: "Whether or not to propagate deletion",
							Optional:    true,
						},
						"delete_marker_replication": {
							Type:        schema.TypeBool,
							Description: "Whether or not to synchronise marker deletion",
							Optional:    true,
						},
						"existing_object_replication": {
							Type:        schema.TypeBool,
							Description: "Whether or not to synchronise object created prior the replication configuration",
							Optional:    true,
						},
						"metadata_sync": {
							Type:        schema.TypeBool,
							Description: "Whether or not to synchonise buckets and objects metadata (such as locks). This must be enabled to achieve a two-way replication",
							Optional:    true,
						},
						"target": {
							Type:        schema.TypeList,
							Description: "Bucket prefix",
							MinItems:    1,
							MaxItems:    1,
							Required:    true,
							Elem: &schema.Resource{
								Schema: map[string]*schema.Schema{
									"bucket": {
										Type:        schema.TypeString,
										Description: "The name of the existing target bucket to replicate into",
										Required:    true,
									},
									"storage_class": {
										Type:        schema.TypeString,
										Description: "The storage class to use for the object on this target",
										Optional:    true,
									},
									"host": {
										Type:        schema.TypeString,
										Description: "The target host (pair IP/port or domain port). If port is omitted, HTTPS port (or HTTP if unsecure) will be used. This host must be reachable by the MinIO instance itself",
										Required:    true,
									},
									"secure": {
										Type:        schema.TypeBool,
										Description: "Whether to use HTTPS with this target (Recommended). Note that disabling HTTPS will yield Terraform warning for security reason`",
										Optional:    true,
										Default:     true,
									},
									"path_style": {
										Type:         schema.TypeString,
										Description:  "Whether to use path-style or virtual-hosted-syle request to this target (https://docs.aws.amazon.com/AmazonS3/latest/userguide/VirtualHosting.html#path-style-access). `auto` allows MinIO to chose automatically the appropriate option (Recommened)`",
										Optional:     true,
										Default:      "auto",
										ValidateFunc: validation.StringInSlice([]string{"on", "off", "auto"}, true),
									},
									"path": {
										Type:        schema.TypeString,
										Description: "Path of the Minio endpoint. This is usefull if MinIO API isn't served on at the root, e.g for `example.com/minio/`, the path would be `/minio/`",
										Optional:    true,
									},
									"syncronous": {
										Type:        schema.TypeBool,
										Description: "Use synchronous replication.",
										Optional:    true,
										Default:     false,
									},
									"disable_proxy": {
										Type:        schema.TypeBool,
										Description: "Disable proxy for this target",
										Optional:    true,
										Default:     false,
									},
									"health_check_period": {
										Type:        schema.TypeString,
										Description: "Period where the health of this target will be checked. This must be a valid duration, such as `5s` or `2m`",
										Optional:    true,
										Default:     "30s",
										DiffSuppressFunc: func(k, oldValue, newValue string, d *schema.ResourceData) bool {
											newVal, err := time.ParseDuration(newValue)
											return err == nil && shortDur(newVal) == oldValue
										},
										ValidateFunc: validation.StringMatch(regexp.MustCompile(`^[0-9]+\s?[s|m|h]$`), "must be a valid golang duration"),
									},
									"bandwidth_limit": {
										Type:        schema.TypeString,
										Description: "Maximum bandwidth in byte per second that MinIO can used when syncronysing this target. Minimum is 100MB",
										Optional:    true,
										Default:     "0",
										DiffSuppressFunc: func(k, oldValue, newValue string, d *schema.ResourceData) bool {
											newVal, err := humanize.ParseBytes(newValue)
											return err == nil && humanize.Bytes(newVal) == oldValue
										},
										ValidateDiagFunc: func(i interface{}, _ cty.Path) (diags diag.Diagnostics) {
											v, ok := i.(string)
											if !ok {
												diags = append(diags, diag.Diagnostic{
													Severity: diag.Error,
													Summary:  "expected type of bandwidth_limit to be string",
												})
												return
											}

											if v == "" {
												return
											}

											bandwidth, err := humanize.ParseBytes(v)
											if err != nil {
												diags = append(diags, diag.Diagnostic{
													Severity: diag.Error,
													Summary:  fmt.Sprintf("bandwidth_limit must be a positive value. It may use suffixes (k, m, g, ..) '%s'", v),
												})
												return
											}
											// 100 MB (SI) = 100 * 1,000,000 = 100,000,000 bytes
											const minBandwidthLimit = uint64(100000000)
											if bandwidth < minBandwidthLimit {
												diags = append(diags, diag.Diagnostic{
													Severity: diag.Error,
													Summary:  fmt.Sprintf("When set, bandwidth_limit must be at least 100MBps '%s'", v),
												})

											}
											return
										},
									},
									"region": {
										Type:        schema.TypeString,
										Description: "Region of the target MinIO. This will be used to generate the target ARN",
										Optional:    true,
									},
									"access_key": {
										Type:         schema.TypeString,
										Description:  "Access key for the replication service account in the target MinIO",
										Required:     true,
										ValidateFunc: validation.StringIsNotEmpty,
									},
									"secret_key": {
										Type:         schema.TypeString,
										Description:  "Secret key for the replication service account in the target MinIO. This is optional so it can be imported but prevent secret update",
										Optional:     true,
										Sensitive:    true,
										ValidateFunc: validation.StringIsNotEmpty,
									},
								},
							},
						},
					},
				},
			},
		},
	}
}

func minioPutBucketReplication(ctx context.Context, d *schema.ResourceData, meta interface{}) diag.Diagnostics {
	bucketReplicationConfig, diags := BucketReplicationConfig(d, meta)
	replicationConfig := bucketReplicationConfig.ReplicationRules

	if replicationConfig == nil || diags.HasError() {
		return diags
	}

	log.Printf("[DEBUG] S3 bucket: %s, put replication configuration: %v", bucketReplicationConfig.MinioBucket, replicationConfig)

	cfg, err := convertBucketReplicationConfig(bucketReplicationConfig, replicationConfig)

	if err != nil {
		return NewResourceError(fmt.Sprintf("error generating bucket replication configuration for %q", bucketReplicationConfig.MinioBucket), d.Id(), err)
	}

	err = bucketReplicationConfig.MinioClient.SetBucketReplication(
		ctx,
		bucketReplicationConfig.MinioBucket,
		cfg,
	)

	if err != nil {
		return NewResourceError(fmt.Sprintf("error putting bucket replication configuration for %q", bucketReplicationConfig.MinioBucket), d.Id(), err)
	}

	d.SetId(bucketReplicationConfig.MinioBucket)

	return nil
}

func minioReadBucketReplication(ctx context.Context, d *schema.ResourceData, meta interface{}) diag.Diagnostics {
	bucketReplicationConfig, diags := BucketReplicationConfig(d, meta)

	if diags.HasError() {
		return diags
	}

	client := bucketReplicationConfig.MinioClient
	admclient := bucketReplicationConfig.MinioAdmin
	bucketName := d.Id()

	// Reverse index to store rule definition read from Minio to macth the order they have in the IaC. This prevent Terrfaform from try to re-order rule each time
	rulePriorityMap := map[int]int{}
	// Reverse index to store arn and index in the rule set. This is used to match bucket config and remote target order
	ruleArnMap := map[string]int{}

	if bucketReplicationConfig.ReplicationRules != nil {
		for idx, rule := range bucketReplicationConfig.ReplicationRules {
			priority := rule.Priority
			if priority == 0 {
				priority = -len(bucketReplicationConfig.ReplicationRules) + idx
			}
			rulePriorityMap[priority] = idx
		}
	}

	log.Printf("[DEBUG] S3 bucket replication, read for bucket: %s", bucketName)

	// First, gather the bucket replication config
	rcfg, err := client.GetBucketReplication(ctx, bucketName)
	if err != nil {
		log.Printf("[WARN] Unable to fetch bucket replication config for %q: %v", bucketName, err)
		return diag.FromErr(fmt.Errorf("error reading bucket replication configuration: %s", err))
	}

	rules := make([]map[string]interface{}, len(rcfg.Rules))

	for idx, rule := range rcfg.Rules {
		var ruleIdx int
		var ok bool
		if ruleIdx, ok = rulePriorityMap[rule.Priority]; !ok {
			ruleIdx = idx
		}
		if _, ok = ruleArnMap[rule.Destination.Bucket]; ok {
			log.Printf("[WARN] Conflict detetcted between two rules containing the same ARN for %q: %q", bucketName, rule.Destination.Bucket)
			return diag.FromErr(fmt.Errorf("conflict detetcted between two rules containing the same ARN for %q: %q", bucketName, rule.Destination.Bucket))
		}
		ruleArnMap[rule.Destination.Bucket] = ruleIdx
		target := map[string]interface{}{
			"storage_class": rule.Destination.StorageClass,
		}
		var priority interface{} = rule.Priority
		if len(bucketReplicationConfig.ReplicationRules) > ruleIdx && rule.Priority == -bucketReplicationConfig.ReplicationRules[ruleIdx].Priority {
			priority = nil
		}
		rules[ruleIdx] = map[string]interface{}{
			"id":                          rule.ID,
			"arn":                         rule.Destination.Bucket,
			"enabled":                     rule.Status == replication.Enabled,
			"priority":                    priority,
			"prefix":                      rule.Prefix(),
			"delete_replication":          rule.DeleteReplication.Status == replication.Enabled,
			"delete_marker_replication":   rule.DeleteMarkerReplication.Status == replication.Enabled,
			"existing_object_replication": rule.ExistingObjectReplication.Status == replication.Enabled,
			"metadata_sync":               rule.SourceSelectionCriteria.ReplicaModifications.Status == replication.Enabled,
		}

		log.Printf("[DEBUG] Rule data for rule#%d is: %q", ruleIdx, rule)

		if len(rule.Filter.And.Tags) != 0 || rule.Filter.And.Prefix != "" {
			tags := map[string]string{}
			for _, tag := range rule.Filter.And.Tags {
				if tag.IsEmpty() {
					continue
				}
				tags[tag.Key] = tag.Value
			}
			rules[ruleIdx]["tags"] = tags
		} else if rule.Filter.Tag.Key != "" {
			rules[ruleIdx]["tags"] = map[string]string{
				rule.Filter.Tag.Key: rule.Filter.Tag.Value,
			}
		} else {
			rules[ruleIdx]["tags"] = nil
		}

		// During import, there is no rules defined. Furthermore, since it is impossible to read the secret from the API, we
		// default it to an empty string, allowing user to prevent remote changes by also using an empty string or omiting the secret_key
		if len(bucketReplicationConfig.ReplicationRules) > ruleIdx {
			target["secret_key"] = bucketReplicationConfig.ReplicationRules[ruleIdx].Target.SecretKey
		}

		rules[ruleIdx]["target"] = []interface{}{target}
	}

	// Second, we read the remote bucket config
	existingRemoteTargets, err := admclient.ListRemoteTargets(ctx, bucketName, "")
	if err != nil {
		log.Printf("[WARN] Unable to fetch existing remote target config for %q: %v", bucketName, err)
		return diag.FromErr(fmt.Errorf("error reading replication remote target configuration: %s", err))
	}

	if len(existingRemoteTargets) != len(rules) {
		return diag.FromErr(fmt.Errorf("inconsistent number of remote target and bucket replication rules (%d != %d)", len(existingRemoteTargets), len(rules)))
	}

	for _, remoteTarget := range existingRemoteTargets {
		var ruleIdx int
		var ok bool
		var target map[string]interface{}
		if ruleIdx, ok = ruleArnMap[remoteTarget.Arn]; !ok {
			return diag.FromErr(fmt.Errorf("unable to find the remote target configuration for ARN %q on %s", remoteTarget.Arn, bucketName))
		}
		var targets []interface{}
		if targets, ok = rules[ruleIdx]["target"].([]interface{}); !ok || len(targets) != 1 {
			return diag.FromErr(fmt.Errorf("unable to find the bucket replication configuration associated to ARN %q (rule#%d) on %s", remoteTarget.Arn, ruleIdx, bucketName))
		}
		if target, ok = targets[0].(map[string]interface{}); !ok || len(target) == 0 {
			return diag.FromErr(fmt.Errorf("unable to extract the target information for the this remote target configuration associated on %s", bucketName))
		}

		pathComponent := strings.Split(remoteTarget.TargetBucket, "/")

		log.Printf("[DEBUG] absolute remote target path is %s", remoteTarget.TargetBucket)

		target["bucket"] = pathComponent[len(pathComponent)-1]
		target["host"] = remoteTarget.Endpoint
		target["secure"] = remoteTarget.Secure
		target["path_style"] = remoteTarget.Path
		target["path"] = strings.Join(pathComponent[:len(pathComponent)-1], "/")
		target["syncronous"] = remoteTarget.ReplicationSync
		target["disable_proxy"] = remoteTarget.DisableProxy
		target["health_check_period"] = shortDur(remoteTarget.HealthCheckDuration)
		var bwUint64 uint64
		if remoteTarget.BandwidthLimit < 0 {
			// Bandwidth limit shouldn't be negative, but handle defensively.
			// Setting to 0 as a safe default.
			log.Printf("[WARN] Received negative bandwidth limit (%d) from MinIO, treating as 0", remoteTarget.BandwidthLimit)
			bwUint64 = 0
		} else {
			bwUint64 = uint64(remoteTarget.BandwidthLimit)
		}
		target["bandwidth_limit"] = humanize.Bytes(bwUint64)
		target["region"] = remoteTarget.Region
		target["access_key"] = remoteTarget.Credentials.AccessKey

		log.Printf("[DEBUG] serialise remote target data is %v", target)

		rules[ruleIdx]["target"] = []interface{}{target}
	}

	if err := d.Set("bucket", d.Id()); err != nil {
		return diag.FromErr(fmt.Errorf("error setting replication configuration: %w", err))
	}

	if err := d.Set("rule", rules); err != nil {
		return diag.FromErr(fmt.Errorf("error setting replication configuration: %w", err))
	}

	return diags
}

func minioDeleteBucketReplication(ctx context.Context, d *schema.ResourceData, meta interface{}) diag.Diagnostics {
	bucketReplicationConfig, diags := BucketReplicationConfig(d, meta)

	if len(bucketReplicationConfig.ReplicationRules) == 0 && !diags.HasError() {
		log.Printf("[DEBUG] Removing bucket replication for unversioned bucket (%s) from state", d.Id())
	} else if diags.HasError() {
		return diags
	}

	client := bucketReplicationConfig.MinioClient
	admclient := bucketReplicationConfig.MinioAdmin

	rcfg, err := client.GetBucketReplication(ctx, bucketReplicationConfig.MinioBucket)
	if err != nil {
		log.Printf("[WARN] Unable to fetch bucket replication config for %q: %v", bucketReplicationConfig.MinioBucket, err)
		return diag.FromErr(fmt.Errorf("error reading bucket replication configuration: %s", err))
	}

	log.Printf("[DEBUG] S3 bucket: %s, disabling replication", bucketReplicationConfig.MinioBucket)

	rcfg.Rules = []replication.Rule{}
	err = client.SetBucketReplication(ctx, bucketReplicationConfig.MinioBucket, rcfg)
	if err != nil {
		log.Printf("[WARN] Unable to set an empty replication config for %q: %v", bucketReplicationConfig.MinioBucket, err)
		return diag.FromErr(fmt.Errorf("error writing bucket replication configuration: %s", err))
	}

	existingRemoteTargets, err := admclient.ListRemoteTargets(ctx, bucketReplicationConfig.MinioBucket, "")
	if err != nil {
		log.Printf("[WARN] Unable to fetch existing remote target config for %q: %v", bucketReplicationConfig.MinioBucket, err)
		return diag.FromErr(fmt.Errorf("error reading replication remote target configuration: %s", err))
	}
	if len(existingRemoteTargets) != 0 {
		return diag.FromErr(fmt.Errorf("%d remote targets are still present on the bukcet while none are expected", len(existingRemoteTargets)))
	}

	return diags
}
func toEnableFlag(b bool) string {
	if b {
		return "enable"
	}
	return "disable"
}

func convertBucketReplicationConfig(bucketReplicationConfig *S3MinioBucketReplication, c []S3MinioBucketReplicationRule) (rcfg replication.Config, err error) {
	client := bucketReplicationConfig.MinioClient
	admclient := bucketReplicationConfig.MinioAdmin

	ctx := context.Background()

	rcfg, err = client.GetBucketReplication(ctx, bucketReplicationConfig.MinioBucket)
	if err != nil {
		log.Printf("[WARN] Unable to fetch bucket replication config for %q: %v", bucketReplicationConfig.MinioBucket, err)
		return
	}

	usedARNs := make([]string, len(c))
	existingRemoteTargets, err := admclient.ListRemoteTargets(ctx, bucketReplicationConfig.MinioBucket, "")
	if err != nil {
		log.Printf("[WARN] Unable to fetch existing remote target config for %q: %v", bucketReplicationConfig.MinioBucket, err)
		return
	}

	for i, rule := range c {
		err = s3utils.CheckValidBucketName(rule.Target.Bucket)
		if err != nil {
			log.Printf("[WARN] Invalid bucket name for %q: %v", rule.Target.Bucket, err)
			return
		}

		tgtBucket := rule.Target.Bucket
		if rule.Target.Path != "" {
			tgtBucket = path.Clean("./" + rule.Target.Path + "/" + tgtBucket)
		}
		log.Printf("[DEBUG] Full path to target bucket is %s", tgtBucket)

		creds := &madmin.Credentials{AccessKey: rule.Target.AccessKey, SecretKey: rule.Target.SecretKey}
		bktTarget := &madmin.BucketTarget{
			TargetBucket:        tgtBucket,
			Secure:              rule.Target.Secure,
			Credentials:         creds,
			Endpoint:            rule.Target.Host,
			Path:                rule.Target.PathStyle.String(),
			API:                 "s3v4",
			Type:                madmin.ReplicationService,
			Region:              rule.Target.Region,
			BandwidthLimit:      rule.Target.BandwidthLimit,
			ReplicationSync:     rule.Target.Syncronous,
			DisableProxy:        rule.Target.DisableProxy,
			HealthCheckDuration: rule.Target.HealthCheckPeriod,
		}
		targets, _ := admclient.ListRemoteTargets(ctx, bucketReplicationConfig.MinioBucket, string(madmin.ReplicationService))
		log.Printf("[DEBUG] Existing remote targets %q: %v", bucketReplicationConfig.MinioBucket, targets)
		var arn string

		var existingRemoteTarget *madmin.BucketTarget
		if rule.Arn != "" {
			for i, target := range targets {
				if target.Arn == rule.Arn {
					existingRemoteTarget = &targets[i]
					break
				}
				// At this stage, we could also anticipate already existing remote target failure since endpoint is unique
				// per bucket (https://github.com/minio/minio/blob/master/cmd/bucket-targets.go#L356) but this behavior could change in the future
			}
		}

		if existingRemoteTarget == nil {
			log.Printf("[DEBUG] Adding new remote target %v for %q", *bktTarget, bucketReplicationConfig.MinioBucket)
			arn, err = admclient.SetRemoteTarget(ctx, bucketReplicationConfig.MinioBucket, bktTarget)
			if err != nil {
				log.Printf("[WARN] Unable to configure remote target %v for %q: %v", *bktTarget, bucketReplicationConfig.MinioBucket, err)
				return
			}
		} else {
			var remoteTargetUpdate []madmin.TargetUpdateType

			if *existingRemoteTarget.Credentials != *bktTarget.Credentials {
				existingRemoteTarget.Credentials = bktTarget.Credentials
				remoteTargetUpdate = append(remoteTargetUpdate, madmin.CredentialsUpdateType)
			}
			if existingRemoteTarget.ReplicationSync != bktTarget.ReplicationSync {
				existingRemoteTarget.ReplicationSync = bktTarget.ReplicationSync
				remoteTargetUpdate = append(remoteTargetUpdate, madmin.SyncUpdateType)
			}
			if existingRemoteTarget.DisableProxy != bktTarget.DisableProxy {
				existingRemoteTarget.DisableProxy = bktTarget.DisableProxy
				remoteTargetUpdate = append(remoteTargetUpdate, madmin.ProxyUpdateType)
			}
			if existingRemoteTarget.BandwidthLimit != bktTarget.BandwidthLimit {
				existingRemoteTarget.BandwidthLimit = bktTarget.BandwidthLimit
				remoteTargetUpdate = append(remoteTargetUpdate, madmin.BandwidthLimitUpdateType)
			}
			if existingRemoteTarget.HealthCheckDuration != bktTarget.HealthCheckDuration {
				existingRemoteTarget.HealthCheckDuration = bktTarget.HealthCheckDuration
				remoteTargetUpdate = append(remoteTargetUpdate, madmin.HealthCheckDurationUpdateType)
			}
			if existingRemoteTarget.Path != bktTarget.Path {
				existingRemoteTarget.Path = bktTarget.Path
				remoteTargetUpdate = append(remoteTargetUpdate, madmin.PathUpdateType)
			}
			log.Printf("[DEBUG] Editing remote target %v for %q", *bktTarget, bucketReplicationConfig.MinioBucket)
			arn, err = admclient.UpdateRemoteTarget(ctx, existingRemoteTarget, remoteTargetUpdate...)
			if err != nil {
				log.Printf("[WARN] Unable to update the remote target %v for %q: %v", *bktTarget, bucketReplicationConfig.MinioBucket, err)
				return
			}
		}

		tagList := []string{}
		for k, v := range rule.Tags {
			var escapedValue *url.URL
			escapedValue, err = url.Parse(v)
			if err != nil {
				return
			}

			tagList = append(tagList, fmt.Sprintf("%s=%s", k, escapedValue.String()))
		}

		opts := replication.Options{
			TagString:               strings.Join(tagList, "&"),
			IsTagSet:                len(tagList) != 0,
			StorageClass:            rule.Target.StorageClass,
			Priority:                strconv.Itoa(int(math.Abs(float64(rule.Priority)))),
			Prefix:                  rule.Prefix,
			RuleStatus:              toEnableFlag(rule.Enabled),
			ID:                      rule.Id,
			DestBucket:              arn,
			ReplicateDeleteMarkers:  toEnableFlag(rule.DeleteMarkerReplication),
			ReplicateDeletes:        toEnableFlag(rule.DeleteReplication),
			ReplicaSync:             toEnableFlag(rule.MetadataSync),
			ExistingObjectReplicate: toEnableFlag(rule.ExistingObjectReplication),
		}
		if strings.TrimSpace(opts.ID) == "" {
			rule.Id = xid.New().String()
			opts.ID = rule.Id
			opts.Op = replication.AddOption
			log.Printf("[DEBUG] Adding replication option for rule#%d: %v", i, opts)
			err = rcfg.AddRule(opts)
		} else {
			opts.Op = replication.SetOption
			log.Printf("[DEBUG] Editing replication option for rule#%d: %v", i, opts)
			err = rcfg.EditRule(opts)
		}

		if err != nil {
			return
		}
		usedARNs[i] = arn
	}

	for _, existingRemoteTarget := range existingRemoteTargets {
		if !slices.Contains(usedARNs, existingRemoteTarget.Arn) {
			err = admclient.RemoveRemoteTarget(ctx, bucketReplicationConfig.MinioBucket, existingRemoteTarget.Arn)
		}

		if err != nil {
			return
		}
	}

	return
}

func getBucketReplicationConfig(v []interface{}) (result []S3MinioBucketReplicationRule, errs diag.Diagnostics) {
	if len(v) == 0 || v[0] == nil {
		return
	}

	result = make([]S3MinioBucketReplicationRule, len(v))
	for i, rule := range v {
		var ok bool
		tfMap, ok := rule.(map[string]interface{})
		if !ok {
			errs = append(errs, diag.Errorf("Unable to extra the rule %d", i)...)
			continue
		}
		log.Printf("[DEBUG] rule[%d] contains %v", i, tfMap)

		result[i].Arn, _ = tfMap["arn"].(string)
		result[i].Id, _ = tfMap["id"].(string)

		if result[i].Enabled, ok = tfMap["enabled"].(bool); !ok {
			log.Printf("[DEBUG] rule[%d].enabled omitted. Defaulting to true", i)
			result[i].Enabled = true
		}

		if result[i].Priority, ok = tfMap["priority"].(int); !ok || result[i].Priority == 0 {
			// Since priorities are always positive, we use a negative value to indicate they were automatically generated
			result[i].Priority = -len(v) + i
			log.Printf("[DEBUG] rule[%d].priority omitted. Defaulting to index (%d)", i, -result[i].Priority)
		}

		result[i].Prefix, _ = tfMap["prefix"].(string)

		if tags, ok := tfMap["tags"].(map[string]interface{}); ok {
			log.Printf("[DEBUG] rule[%d].tags map contains: %v", i, tags)
			tagMap := map[string]string{}
			for k, val := range tags {
				var valOk bool
				tagMap[k], valOk = val.(string)
				if !valOk {
					errs = append(errs, diag.Errorf("rule[%d].tags[%s] value must be a string, not a %s", i, k, reflect.TypeOf(val))...)
				}
			}
			result[i].Tags = tagMap
		} else {
			errs = append(errs, diag.Errorf("unable to extarct rule[%d].tags of type %s", i, reflect.TypeOf(tfMap["tags"]))...)
		}

		log.Printf("[DEBUG] rule[%d].tags are: %v", i, result[i].Tags)

		result[i].DeleteReplication, ok = tfMap["delete_replication"].(bool)
		result[i].DeleteReplication = result[i].DeleteReplication && ok
		result[i].DeleteMarkerReplication, ok = tfMap["delete_marker_replication"].(bool)
		result[i].DeleteMarkerReplication = result[i].DeleteMarkerReplication && ok
		result[i].ExistingObjectReplication, ok = tfMap["existing_object_replication"].(bool)
		result[i].ExistingObjectReplication = result[i].ExistingObjectReplication && ok
		result[i].MetadataSync, ok = tfMap["metadata_sync"].(bool)
		result[i].MetadataSync = result[i].MetadataSync && ok

		var targets []interface{}
		if targets, ok = tfMap["target"].([]interface{}); !ok || len(targets) != 1 {
			errs = append(errs, diag.Errorf("Unexpected value type for rule[%d].target. Exactly one target configuration is expected", i)...)
			continue
		}
		var target map[string]interface{}
		if target, ok = targets[0].(map[string]interface{}); !ok {
			errs = append(errs, diag.Errorf("Unexpected value type for rule[%d].target. Unable to convert to a usable type", i)...)
			continue
		}

		if result[i].Target.Bucket, ok = target["bucket"].(string); !ok {
			errs = append(errs, diag.Errorf("rule[%d].target.bucket cannot be omitted", i)...)
		}

		result[i].Target.StorageClass, _ = target["storage_class"].(string)

		if result[i].Target.Host, ok = target["host"].(string); !ok {
			errs = append(errs, diag.Errorf("rule[%d].target.host cannot be omitted", i)...)
		}

		result[i].Target.Path, _ = target["path"].(string)
		result[i].Target.Region, _ = target["region"].(string)

		if result[i].Target.AccessKey, ok = target["access_key"].(string); !ok {
			errs = append(errs, diag.Errorf("rule[%d].target.access_key cannot be omitted", i)...)
		}

		if result[i].Target.SecretKey, ok = target["secret_key"].(string); !ok {
			errs = append(errs, diag.Errorf("rule[%d].target.secret_key cannot be omitted", i)...)
		}

		if result[i].Target.Secure, ok = target["secure"].(bool); !result[i].Target.Secure || !ok {
			errs = append(errs, diag.Diagnostic{
				Severity: diag.Warning,
				Summary:  fmt.Sprintf("rule[%d].target.secure is false. It is unsafe to use bucket replication over HTTP", i),
			})
		}

		result[i].Target.Syncronous, ok = target["syncronous"].(bool)
		result[i].Target.Syncronous = result[i].Target.Syncronous && ok
		result[i].Target.DisableProxy, ok = target["disable_proxy"].(bool)
		result[i].Target.DisableProxy = result[i].Target.DisableProxy && ok

		var bandwidth uint64
		var err error
		bandwidth, ok, parseDiags := ParseBandwidthLimit(target)
		if len(parseDiags) > 0 {
			errs = append(errs, diag.Errorf("rule[%d].target.bandwidth_limit is invalid. Make sure to use k, m, g as prefix only", i)...)
		}

		if ok {
			var bwLimit int64
			if bandwidth > uint64(math.MaxInt64) {
				log.Printf("[WARN] Configured bandwidth limit (%d) exceeds maximum supported value (%d), clamping.", bandwidth, int64(math.MaxInt64))
				bwLimit = math.MaxInt64 // Clamp to max int64 if overflow would occur
			} else {
				bwLimit = int64(bandwidth)
			}
			result[i].Target.BandwidthLimit = bwLimit // Safe conversion
		}

		var healthcheckDuration string
		if healthcheckDuration, ok = target["health_check_period"].(string); ok {
			result[i].Target.HealthCheckPeriod, err = time.ParseDuration(healthcheckDuration)
			if err != nil {
				log.Printf("[WARN] invalid healthcheck value %q: %v", result[i].Target.HealthCheckPeriod, err)
				errs = append(errs, diag.Errorf("rule[%d].target.health_check_period is invalid. Make sure to use a valid golang time duration notation", i)...)
			}
		}

		var pathstyle string
		pathstyle, _ = target["path_style"].(string)
		switch strings.TrimSpace(strings.ToLower(pathstyle)) {
		case "on":
			result[i].Target.PathStyle = S3PathSyleOn
		case "off":
			result[i].Target.PathStyle = S3PathSyleOff
		default:
			if pathstyle != "auto" && pathstyle != "" {
				errs = append(errs, diag.Diagnostic{
					Severity: diag.Warning,
					Summary:  fmt.Sprintf("rule[%d].target.path_style must be \"on\", \"off\" or \"auto\". Defaulting to \"auto\"", i),
				})
			}
			result[i].Target.PathStyle = S3PathSyleAuto
		}

	}
	return
}
