package impl

import (
	"context"
	"errors"
	"fmt"
	"strings"

	"cloud.google.com/go/iam"
	"cloud.google.com/go/longrunning/autogen/longrunningpb"
	"cloud.google.com/go/storage"
	storagetransfer "cloud.google.com/go/storagetransfer/apiv1"
	storagetransferpb "cloud.google.com/go/storagetransfer/apiv1/storagetransferpb"
	"github.com/elliotchance/pie/v2"
	"github.com/googleapis/gax-go/v2/apierror"
	"github.com/tidbcloud/dataflow-service/app/internal/gcp_facade"
	"github.com/tidbcloud/dataflow-service/pkg/log"
	"go.uber.org/zap"
	date "google.golang.org/genproto/googleapis/type/date"
	"google.golang.org/grpc/codes"
	"google.golang.org/protobuf/types/known/durationpb"
)

var _ gcp_facade.GCPStorageTransferClient = &GCPStorageTransferClientImpl{}

// GCPStorageTransferClientImpl implements the GCPStorageTransferClient interface
type GCPStorageTransferClientImpl struct {
	storageTransferCli *storagetransfer.Client
	storageClient      *storage.Client
}

// NewGCPStorageTransferClient creates a new GCPStorageTransferClientImpl
func NewGCPStorageTransferClient() *GCPStorageTransferClientImpl {
	ctx := context.Background()
	cli, err := storagetransfer.NewClient(ctx)
	if err != nil {
		panic(fmt.Sprintf("failed to create GCP Storage Transfer Client: %s", err))
	}
	// Create storage client
	storageClient, err := storage.NewClient(ctx)
	if err != nil {
		panic(fmt.Sprintf("failed to create storage client: %s", err))
	}
	return &GCPStorageTransferClientImpl{
		storageTransferCli: cli,
		storageClient:      storageClient,
	}
}

// GetServiceAccount gets the service account of GCP Storage Transfer Service in the specified GCP project
// The SA running dataflow-svc is used to perform this API, so make sure the SA has the permission `storagetransfer.projects.getServiceAccount`.
func (c *GCPStorageTransferClientImpl) GetServiceAccount(ctx context.Context, gcpProjectID string) (string, error) {
	req := &storagetransferpb.GetGoogleServiceAccountRequest{
		ProjectId: gcpProjectID,
	}
	resp, err := c.storageTransferCli.GetGoogleServiceAccount(ctx, req)
	if err != nil {
		return "", fmt.Errorf("failed to get GCP Storage Transfer Service account: %w", err)
	}
	return resp.GetAccountEmail(), nil
}

// GrantPermissionToSAOnBucket grants permission to a service account for a bucket
func (c *GCPStorageTransferClientImpl) GrantPermissionToSAOnBucket(
	ctx context.Context, bucketName, serviceAccount string, roles []iam.RoleName,
) error {
	// TODO(ideascf): we should use app/models/bucket/interface.go:XXXBucketManagerModel to grant permission
	// Currently, the XXXBucketManagerModel is not support to grant permission with specific roles,
	// so we need to do it manually here.

	bucket := c.storageClient.Bucket(bucketName)
	iam := bucket.IAM()
	policy, err := iam.Policy(ctx)
	if err != nil {
		return fmt.Errorf("failed to get bucket IAM policy: %w", err)
	}

	memberName := fmt.Sprintf("serviceAccount:%s", serviceAccount)
	for _, role := range roles {
		policy.Add(memberName, role)
	}

	// Update IAM policy
	if err := iam.SetPolicy(ctx, policy); err != nil {
		return fmt.Errorf("failed to set bucket IAM policy: %w", err)
	}

	return nil
}

//nolint:funlen
func (c *GCPStorageTransferClientImpl) CreateOneTimeTransferJob(
	ctx context.Context, params *gcp_facade.TransferJobParams,
) (string, error) {
	nonRetryableCodes := []codes.Code{
		codes.FailedPrecondition,
	}
	logger := log.GetLogger(ctx)

	// Create transfer job request
	transferJob := &storagetransferpb.TransferJob{
		ProjectId:   params.ProjectID,
		Name:        params.JobName,
		Description: params.JobDescription,
		Status:      storagetransferpb.TransferJob_ENABLED,
		Schedule: &storagetransferpb.Schedule{
			ScheduleStartDate: &date.Date{
				Year:  int32(params.RunAt.Year()),
				Month: int32(params.RunAt.Month()),
				Day:   int32(params.RunAt.Day()),
			},
			ScheduleEndDate: &date.Date{
				Year:  int32(params.RunAt.Year()),
				Month: int32(params.RunAt.Month()),
				Day:   int32(params.RunAt.Day()),
			},
		},
		TransferSpec: &storagetransferpb.TransferSpec{
			ObjectConditions: &storagetransferpb.ObjectConditions{
				MinTimeElapsedSinceLastModification: &durationpb.Duration{Seconds: 0},
			},
			DataSource: &storagetransferpb.TransferSpec_GcsDataSource{
				GcsDataSource: &storagetransferpb.GcsData{
					BucketName: params.SourceBucket,
					Path:       params.SourcePath,
				},
			},
			DataSink: &storagetransferpb.TransferSpec_GcsDataSink{
				GcsDataSink: &storagetransferpb.GcsData{
					BucketName: params.TargetBucket,
					Path:       params.TargetPath,
				},
			},
			TransferOptions: &storagetransferpb.TransferOptions{
				OverwriteWhen: storagetransferpb.TransferOptions_DIFFERENT,
			},
		},
	}

	// Create the transfer job
	req := &storagetransferpb.CreateTransferJobRequest{
		TransferJob: transferJob,
	}

	resp, err := c.storageTransferCli.CreateTransferJob(ctx, req)
	if err != nil {
		// Refer to https://pkg.go.dev/cloud.google.com/go#hdr-Inspecting_errors
		var ae *apierror.APIError
		if errors.As(err, &ae) {
			logger.Error("failed to create transfer job",
				zap.String("reason", ae.Reason()), zap.Any("details", ae.Details().Help.GetLinks()),
			)

			// If a gRPC transport was used you can extract the
			// google.golang.org/grpc/status.Status from the error
			s := ae.GRPCStatus()
			logger.Error("failed to create transfer job",
				zap.String("code", s.Code().String()),
			)
			if pie.Contains(nonRetryableCodes, s.Code()) {
				return "", &gcp_facade.NonRetryableError{Err: err}
			}
		}

		return "", fmt.Errorf("failed to create transfer job: %w", err)
	}

	return resp.GetName(), nil
}

func (c *GCPStorageTransferClientImpl) GetOneTimeTransferJobStatus(
	ctx context.Context, gcpProjectID, jobName string,
) (*gcp_facade.TransferJobStatus, error) {
	logger := log.GetLogger(ctx)
	req := &storagetransferpb.GetTransferJobRequest{
		JobName:   jobName,
		ProjectId: gcpProjectID,
	}
	resp, err := c.storageTransferCli.GetTransferJob(ctx, req)
	if err != nil {
		return nil, fmt.Errorf("failed to get transfer job status: %w", err)
	}

	// The TransferJob we used for backup export is one-time job, so that there is only one operation
	latestOperationName := resp.GetLatestOperationName()
	if latestOperationName == "" {
		return nil, fmt.Errorf("no latest operation name found, retry it later")
	}

	opResp, err := c.storageTransferCli.GetOperation(ctx, &longrunningpb.GetOperationRequest{
		Name: latestOperationName,
	})
	if err != nil {
		return nil, fmt.Errorf("failed to get transfer operation status: %w", err)
	}
	logger.Debug("transfer operation", zap.Any("operation", opResp))
	status := &gcp_facade.TransferJobStatus{}
	if !opResp.GetDone() {
		return status, nil
	}
	return c.processTransferOperationResponse(ctx, opResp)
}

func (c *GCPStorageTransferClientImpl) processTransferOperationResponse(
	ctx context.Context, opResp *longrunningpb.Operation,
) (*gcp_facade.TransferJobStatus, error) {
	logger := log.GetLogger(ctx).With(zap.String("operation_name", opResp.GetName()))

	status := &gcp_facade.TransferJobStatus{}
	if err := opResp.GetError(); err != nil {
		status.Failed = true
		status.Message = err.GetMessage()
		return status, nil
	}

	if opResp.Metadata.MessageIs(&storagetransferpb.TransferOperation{}) {
		to := &storagetransferpb.TransferOperation{}
		if err := opResp.Metadata.UnmarshalTo(to); err != nil {
			return nil, fmt.Errorf("failed to unmarshal transfer operation metadata: %w", err)
		}
		logger.Info("transfer operation status", zap.String("status", to.Status.String()))
		switch to.Status {
		case storagetransferpb.TransferOperation_SUCCESS:
			status.Completed = true
			status.Message = to.GetCounters().String()
		case storagetransferpb.TransferOperation_FAILED, storagetransferpb.TransferOperation_ABORTED:
			var errMsgs []string
			for _, breakdown := range to.GetErrorBreakdowns() {
				errMsgs = append(errMsgs, breakdown.String())
			}
			logger.Error("transfer job failed", zap.String("message", strings.Join(errMsgs, "\n")))
			status.Failed = true
			status.Message = fmt.Sprintf("transfer job status: %s", to.Status.String())
		default:
			logger.Info("waiting for transfer operation to complete", zap.String("status", to.Status.String()))
		}
	} else {
		logger.Error("unknown transfer operation metadata", zap.Any("metadata", opResp.Metadata))
		return nil, fmt.Errorf("unknown transfer operation metadata")
	}

	return status, nil
}

func (c *GCPStorageTransferClientImpl) CancelOneTimeTransferJob(
	ctx context.Context, gcpProjectID, jobName string,
) error {
	logger := log.GetLogger(ctx)
	req := &storagetransferpb.GetTransferJobRequest{
		JobName:   jobName,
		ProjectId: gcpProjectID,
	}
	resp, err := c.storageTransferCli.GetTransferJob(ctx, req)
	if err != nil {
		return fmt.Errorf("failed to get transfer job status: %w", err)
	}

	// The TransferJob we used for backup export is one-time job, so that there is only one operation
	latestOperationName := resp.GetLatestOperationName()
	if latestOperationName == "" {
		return fmt.Errorf("no latest operation name found, retry it later")
	}
	// get the operation
	opResp, err := c.storageTransferCli.GetOperation(ctx, &longrunningpb.GetOperationRequest{
		Name: latestOperationName,
	})
	if err != nil {
		return fmt.Errorf("failed to get transfer operation status: %w", err)
	}

	// check if the operation is already done
	if opResp.GetDone() {
		logger.Info("transfer operation is already done, ignore cancel",
			zap.String("job_name", jobName),
			zap.String("operation_name", latestOperationName),
		)
		return nil
	}

	err = c.storageTransferCli.CancelOperation(ctx, &longrunningpb.CancelOperationRequest{
		Name: latestOperationName,
	})
	if err != nil {
		return fmt.Errorf("failed to cancel transfer operation(%s): %w", latestOperationName, err)
	}

	return nil
}