package s3clean

import (
	"context"
	"fmt"

	"github.com/aws/aws-sdk-go-v2/service/s3"
	"github.com/aws/aws-sdk-go-v2/service/s3/types"
	cloudv1 "github.com/tidbcloud/infra-api/api/v1"
)

func (r *Reconciler) Clean(ctx context.Context, backup *cloudv1.Backup) error {
	// no s3 data
	if backup.Status.BucketName == "" {
		return nil
	}
	var ct *string

	for {
		var err error
		ct, err = r.batchDelete(ctx, backup.Status.BucketName, backup.Name, ct)
		if err != nil {
			return err
		}
		if ct == nil {
			return nil
		}
	}
}

func (r *Reconciler) batchDelete(ctx context.Context, bucket, prefix string, ct *string) (*string, error) {
	s3Client := r.AwsClient.S3()

	r.Logger.Info("list s3 objects", "bucket", bucket, "prefix", prefix, "continue token", ct)

	res, err := s3Client.ListObjectsV2(ctx, &s3.ListObjectsV2Input{
		Bucket:            &bucket,
		Prefix:            &prefix,
		ContinuationToken: ct,
	})
	if err != nil {
		return nil, err
	}

	if res.KeyCount == 0 {
		return nil, nil
	}

	r.Logger.Info("delete s3 objects", "bucket", bucket, "prefix", prefix, "count", res.KeyCount)

	delRes, err := s3Client.DeleteObjects(ctx, &s3.DeleteObjectsInput{
		Bucket: &bucket,
		Delete: &types.Delete{
			Objects: ObjectsToIdentifiers(res.Contents),
		},
	})
	if err != nil {
		return nil, err
	}
	if len(delRes.Errors) > 0 {
		return nil, fmt.Errorf("can't delete some objects: %v", delRes.Errors)
	}
	if res.IsTruncated {
		return res.NextContinuationToken, nil
	}

	return nil, nil
}

func ObjectsToIdentifiers(objs []types.Object) []types.ObjectIdentifier {
	ids := []types.ObjectIdentifier{}
	for _, obj := range objs {
		ids = append(ids, types.ObjectIdentifier{
			Key: obj.Key,
		})
	}

	return ids
}