package regionalapi

import (
	"context"
	"fmt"
	"net/http"

	infraapi "github.com/tidbcloud/dataflow-service/pkg/infra-api"
	"github.com/tidbcloud/dataflow-service/pkg/log"
	"github.com/tidbcloud/pkgv2/bizerr"
	"go.uber.org/zap"
	corev1 "k8s.io/api/core/v1"
)

const (
	createCompactBackupPath = "/cluster-server/api/v1/clusters/{cluster_id}/compactbackups"
	getCompactBackupPath    = "/cluster-server/api/v1/clusters/{cluster_id}/compactbackups/{compact_backup_id}"
	deleteCompactBackupPath = "/cluster-server/api/v1/clusters/{cluster_id}/compactbackups/{compact_backup_id}"

	compactBackupIDPathParam = "compact_backup_id"
)

// CompactBackupResponse represents the response from downstream compact backup API
type CompactBackupResponse struct {
	CompactBackupId string `json:"id"`
	ClusterId       uint64 `json:"cluster_id"`
	Status          string `json:"status"` // pending/running/success/failed/retry
	CreatedAt       string `json:"created_at"`
	FinishedAt      string `json:"finished_at,omitempty"`
	ErrorMessage    string `json:"error_message,omitempty"`

	// Extended fields to match new protobuf structure
	Progress      string                      `json:"progress,omitempty"`
	BackupPath    string                      `json:"backup_path,omitempty"`
	Configuration *CompactBackupConfiguration `json:"configuration,omitempty"`

	// Optional request fields
	StartTs              *string                      `json:"start_ts,omitempty"`
	EndTs                *string                      `json:"end_ts,omitempty"`
	Concurrency          *int32                       `json:"concurrency,omitempty"`
	InstanceType         *string                      `json:"instance_type,omitempty"`
	CleanPolicy          *string                      `json:"clean_policy,omitempty"`
	ResourceRequirements *corev1.ResourceRequirements `json:"resource_requirements,omitempty"`
}

// CompactBackupConfiguration represents the configuration for compact backup
type CompactBackupConfiguration struct {
	BaseImage   string `json:"baseImage,omitempty"`
	Tag         string `json:"tag,omitempty"`
	Config      string `json:"config,omitempty"`
	TiKVImage   string `json:"tikvImage,omitempty"`   // 新增：自定义 TiKV 镜像
	TiKVVersion string `json:"tikvVersion,omitempty"` // 新增：自定义 TiKV 版本
}

// CreateCompactBackupRequest represents the request to create a compact backup
type CreateCompactBackupRequest struct {
	ClusterId     string         `json:"cluster_id"`
	CompactBackup *CompactBackup `json:"compact_backup"`
}

// CompactBackup represents the compact backup object
type CompactBackup struct {
	ID                   string                       `json:"id,omitempty"`
	ClusterID            string                       `json:"cluster_id,omitempty"`
	StartTS              *string                      `json:"start_ts,omitempty"`
	EndTS                *string                      `json:"end_ts,omitempty"`
	Concurrency          *int32                       `json:"concurrency,omitempty"`
	InstanceType         *string                      `json:"instance_type,omitempty"`
	Configuration        *CompactBackupConfiguration  `json:"configuration,omitempty"`
	Storage              *string                      `json:"storage,omitempty"`
	Status               string                       `json:"status,omitempty"`
	CreatedAt            string                       `json:"created_at,omitempty"`
	FinishedAt           *string                      `json:"finished_at,omitempty"`
	ErrorMessage         *string                      `json:"error_message,omitempty"`
	Progress             *string                      `json:"progress,omitempty"`
	BackupPath           *string                      `json:"backup_path,omitempty"`
	CleanPolicy          *string                      `json:"clean_policy,omitempty"`
	ResourceRequirements *corev1.ResourceRequirements `json:"resource_requirements,omitempty"`
	// ShardCount enables sharded compaction when > 0.
	ShardCount int `json:"shard_count,omitempty"`
}

type CompactBackupClient interface {
	CreateCompactBackup(ctx context.Context, region *Region, clusterID uint64, startTS *uint64, endTS *uint64, concurrency *uint32, instanceType *string, cleanPolicy *string, storage *string, config *CompactBackupConfiguration, resourceRequirements *corev1.ResourceRequirements, shardCount int, compactBackupName string) (*CompactBackupResponse, bizerr.BizError)
	GetCompactBackup(ctx context.Context, region *Region, clusterID uint64, compactBackupID string) (*CompactBackupResponse, bizerr.BizError)
	DeleteCompactBackup(ctx context.Context, region *Region, clusterID uint64, compactBackupID string) bizerr.BizError
}

type compactBackupClient struct {
	regionalCli RegionalClient
}

func NewCompactBackupClient(regionalCli RegionalClient) CompactBackupClient {
	return &compactBackupClient{
		regionalCli: regionalCli,
	}
}

func (c *compactBackupClient) CreateCompactBackup(ctx context.Context, region *Region, clusterID uint64, startTS *uint64, endTS *uint64, concurrency *uint32, instanceType *string, cleanPolicy *string, storage *string, config *CompactBackupConfiguration, resourceRequirements *corev1.ResourceRequirements, shardCount int, compactBackupName string) (*CompactBackupResponse, bizerr.BizError) {
	logger := log.GetLogger(ctx).With(
		zap.Uint64("cluster_id", clusterID),
		zap.String("region", region.Name),
	)
	logger.Info("[REGIONAL_API] creating compact backup with ResourceRequirements",
		zap.Bool("has_resource_requirements", resourceRequirements != nil),
		zap.Any("resource_requirements", resourceRequirements))

	pathParams := map[string]string{
		clusterIDPathParam: fmt.Sprintf("%d", clusterID),
	}

	// Convert parameters
	var startTSStr, endTSStr *string
	var concurrencyInt32 *int32

	if startTS != nil {
		s := fmt.Sprintf("%d", *startTS)
		startTSStr = &s
	}
	if endTS != nil {
		s := fmt.Sprintf("%d", *endTS)
		endTSStr = &s
	}
	if concurrency != nil {
		c := int32(*concurrency)
		concurrencyInt32 = &c
	}

	// Log ResourceRequirements details before building request
	if resourceRequirements != nil {
		logger.Info("[REGIONAL_API] ResourceRequirements details",
			zap.Any("requests", resourceRequirements.Requests),
			zap.Any("limits", resourceRequirements.Limits))
	}

	// Build request with nested CompactBackup object
	compactBackup := &CompactBackup{
		ID:                   compactBackupName,
		ClusterID:            fmt.Sprintf("%d", clusterID),
		StartTS:              startTSStr,
		EndTS:                endTSStr,
		Concurrency:          concurrencyInt32,
		InstanceType:         instanceType,
		Configuration:        config,
		CleanPolicy:          cleanPolicy,
		Storage:              storage,
		ResourceRequirements: resourceRequirements,
		ShardCount:           shardCount,
	}

	logger.Info("[REGIONAL_API] Built CompactBackup request object",
		zap.String("compact_backup_id", compactBackupName),
		zap.String("instance_type", func() string {
			if compactBackup.InstanceType != nil {
				return *compactBackup.InstanceType
			}
			return "nil"
		}()),
		zap.Bool("has_resource_requirements_in_request", compactBackup.ResourceRequirements != nil))

	req := &CreateCompactBackupRequest{
		ClusterId:     fmt.Sprintf("%d", clusterID),
		CompactBackup: compactBackup,
	}

	logger.Info("[REGIONAL_API] Sending request to downstream API",
		zap.String("path", createCompactBackupPath),
		zap.Any("compact_backup_resource_requirements", req.CompactBackup.ResourceRequirements))

	var resp CompactBackupResponse
	bizErr := c.regionalCli.Post(ctx, region, createCompactBackupPath, pathParams, nil, req, &resp)
	if bizErr != nil {
		logger.Error("failed to create compact backup", zap.Error(bizErr))
		return nil, bizErr
	}

	logger.Info("compact backup created successfully",
		zap.String("compact_backup_id", resp.CompactBackupId))
	return &resp, nil
}

func (c *compactBackupClient) GetCompactBackup(ctx context.Context, region *Region, clusterID uint64, compactBackupID string) (*CompactBackupResponse, bizerr.BizError) {
	logger := log.GetLogger(ctx).With(
		zap.Uint64("cluster_id", clusterID),
		zap.String("compact_backup_id", compactBackupID),
		zap.String("cloudProvider", region.CloudProvider),
		zap.String("region", region.Name),
	)
	logger.Info("[REGIONAL_API] getting compact backup status")

	pathParams := map[string]string{
		clusterIDPathParam:       fmt.Sprintf("%d", clusterID),
		compactBackupIDPathParam: compactBackupID,
	}

	var compactBackup CompactBackupResponse
	logger.Info("[REGIONAL_API] calling downstream API",
		zap.String("path", getCompactBackupPath),
		zap.Any("path_params", pathParams))

	bizErr := c.regionalCli.Get(ctx, region, getCompactBackupPath, pathParams, nil, &compactBackup)
	if bizErr != nil {
		logger.Error("[REGIONAL_API] downstream API call failed",
			zap.Error(bizErr),
			zap.Int("http_status", bizErr.HttpStatus()),
			zap.String("error_type", fmt.Sprintf("%T", bizErr)))

		if bizErr.HttpStatus() == http.StatusNotFound {
			logger.Warn("[REGIONAL_API] compact backup not found in downstream")
			wrappedErr := bizerr.New(bizerr.ErrCodeRecordNotFound, "compact backup not found")
			logger.Info("[REGIONAL_API] created wrapped error for not found",
				zap.Int64("wrapped_error_code", int64(wrappedErr.ErrCode())),
				zap.String("wrapped_error_type", fmt.Sprintf("%T", wrappedErr)))
			return nil, wrappedErr
		}
		logger.Error("[REGIONAL_API] other downstream error")
		return nil, bizErr
	}

	logger.Info("[REGIONAL_API] successfully got compact backup status from downstream",
		zap.String("status", compactBackup.Status))
	return &compactBackup, nil
}

func (c *compactBackupClient) DeleteCompactBackup(ctx context.Context, region *Region, clusterID uint64, compactBackupID string) bizerr.BizError {
	logger := log.GetLogger(ctx).With(
		zap.Uint64("cluster_id", clusterID),
		zap.String("compact_backup_id", compactBackupID),
		zap.String("region", region.Name),
	)
	logger.Info("deleting compact backup")

	pathParams := map[string]string{
		clusterIDPathParam:       fmt.Sprintf("%d", clusterID),
		compactBackupIDPathParam: compactBackupID,
	}

	bizErr := c.regionalCli.Delete(ctx, region, deleteCompactBackupPath, pathParams, nil)
	if bizErr != nil {
		if bizErr.HttpStatus() == http.StatusNotFound {
			logger.Warn("compact backup not found for deletion")
			// Consider not found as success for deletion
			return nil
		}
		logger.Error("failed to delete compact backup", zap.Error(bizErr))
		return bizErr
	}

	logger.Info("compact backup deleted successfully")
	return nil
}

// CompactRegionalFacade implementation
type compactRegionalFacade struct {
	compactClient CompactBackupClient
}

// NewCompactFacade creates a new CompactFacade implementation using regional API
func NewCompactFacade(regionalCli RegionalClient) infraapi.CompactFacade {
	return &compactRegionalFacade{
		compactClient: NewCompactBackupClient(regionalCli),
	}
}

func (f *compactRegionalFacade) CreateCompactBackupIgnoreExists(
	ctx context.Context, cloudProvider, region string, param *infraapi.CreateCompactBackupParam,
) (*infraapi.CompactBackupResponse, error) {
	logger := log.GetLogger(ctx).Named("CompactRegionalFacade")

	logger.Info("[FACADE] CreateCompactBackupIgnoreExists called with ResourceRequirements",
		zap.Uint64("cluster_id", param.ClusterID),
		zap.Uint64("tenant_id", param.TenantID),
		zap.Uint64("project_id", param.ProjectID),
		zap.String("cloudProvider", cloudProvider),
		zap.String("region", region),
		zap.Bool("has_resource_requirements", param.ResourceRequirements != nil))

	if param.ResourceRequirements != nil {
		logger.Info("[FACADE] ResourceRequirements details from infraapi param",
			zap.Any("requests", param.ResourceRequirements.Requests),
			zap.Any("limits", param.ResourceRequirements.Limits))
	}

	// Convert infraapi param to regional API config
	var config *CompactBackupConfiguration
	if param.Configuration != nil {
		config = &CompactBackupConfiguration{
			BaseImage:   param.Configuration.BaseImage,
			Tag:         param.Configuration.Tag,
			Config:      param.Configuration.Config,
			TiKVImage:   param.Configuration.TiKVImage,
			TiKVVersion: param.Configuration.TiKVVersion,
		}
	}

	// Prepare region for regional API call
	logicalRegion := &Region{
		TenantID:      param.TenantID,
		Name:          region,
		CloudProvider: cloudProvider,
	}

	logger.Info("[FACADE] Passing ResourceRequirements to regional API client",
		zap.Any("resource_requirements", param.ResourceRequirements))

	// Call regional API
	resp, bizErr := f.compactClient.CreateCompactBackup(ctx, logicalRegion, param.ClusterID, param.StartTS, param.EndTS, param.Concurrency, param.InstanceType, param.CleanPolicy, &param.Storage, config, param.ResourceRequirements, param.ShardCount, param.CompactBackupName)
	if bizErr != nil {
		// Handle conflict (already exists) - ignore for IgnoreExists semantics
		if bizErr.HttpStatus() == http.StatusConflict {
			return f.GetCompactBackup(ctx, cloudProvider, region, param.TenantID, param.ProjectID, param.ClusterID, param.CompactBackupName)
		}
		return nil, bizErr
	}

	// Convert regional API response to facade response
	return f.convertToFacadeResponse(resp), nil
}

func (f *compactRegionalFacade) GetCompactBackup(
	ctx context.Context, cloudProvider, region string, tenantID uint64, projectID uint64, clusterID uint64, compactBackupID string,
) (*infraapi.CompactBackupResponse, error) {
	logicalRegion := &Region{
		TenantID:      tenantID,
		Name:          region,
		CloudProvider: cloudProvider,
	}

	resp, bizErr := f.compactClient.GetCompactBackup(ctx, logicalRegion, clusterID, compactBackupID)
	if bizErr != nil {
		return nil, bizErr
	}

	return f.convertToFacadeResponse(resp), nil
}

func (f *compactRegionalFacade) DeleteCompactBackupIgnoreNotFound(
	ctx context.Context, cloudProvider, region string, tenantID uint64, projectID uint64, clusterID uint64, compactBackupID string,
) error {
	logicalRegion := &Region{
		TenantID:      tenantID,
		Name:          region,
		CloudProvider: cloudProvider,
	}

	bizErr := f.compactClient.DeleteCompactBackup(ctx, logicalRegion, clusterID, compactBackupID)
	if bizErr != nil {
		// IgnoreNotFound semantics: 404 is treated as success
		if bizErr.HttpStatus() == http.StatusNotFound {
			return nil
		}
		return bizErr
	}

	return nil
}

// convertToFacadeResponse converts regional API response to facade response
func (f *compactRegionalFacade) convertToFacadeResponse(resp *CompactBackupResponse) *infraapi.CompactBackupResponse {
	var config *infraapi.CompactBackupConfiguration
	if resp.Configuration != nil {
		config = &infraapi.CompactBackupConfiguration{
			BaseImage:   resp.Configuration.BaseImage,
			Tag:         resp.Configuration.Tag,
			Config:      resp.Configuration.Config,
			TiKVImage:   resp.Configuration.TiKVImage,
			TiKVVersion: resp.Configuration.TiKVVersion,
		}
	}

	return &infraapi.CompactBackupResponse{
		CompactBackupId: resp.CompactBackupId,
		ClusterId:       resp.ClusterId,
		Status:          f.convertStatus(resp.Status),
		CreatedAt:       resp.CreatedAt,
		FinishedAt:      resp.FinishedAt,
		ErrorMessage:    resp.ErrorMessage,
		Progress:        resp.Progress,
		BackupPath:      resp.BackupPath,
		Configuration:   config,
	}
}

// convertStatus converts regional API string status to standardized status
// Since infra-provider now returns lowercase status, this mainly handles legacy compatibility
func (f *compactRegionalFacade) convertStatus(status string) infraapi.CompactBackupStatus {
	switch status {
	case "success": // Legacy compatibility: map success to complete
		return infraapi.CompactBackupStatusComplete
	case "unspecified", "": // Handle unspecified or empty status
		return infraapi.CompactBackupStatusPending // Map to pending as default initial state
	case "pending":
		return infraapi.CompactBackupStatusPending
	case "running":
		return infraapi.CompactBackupStatusRunning
	case "complete":
		return infraapi.CompactBackupStatusComplete
	case "failed":
		return infraapi.CompactBackupStatusFailed
	case "retry":
		return infraapi.CompactBackupStatusRetry
	default:
		return infraapi.CompactBackupStatusPending // Default fallback for unknown statuses
	}
}