package healthcheck

import (
	"bytes"
	"encoding/json"
	"errors"
	"fmt"
	"io"
	"net/http"
	"os"
	"strings"
	"time"

	"github.com/aws/aws-sdk-go/aws"
	"github.com/aws/aws-sdk-go/aws/awserr"
	"github.com/aws/aws-sdk-go/aws/credentials/stscreds"
	"github.com/aws/aws-sdk-go/aws/session"
	"github.com/aws/aws-sdk-go/service/s3"
	"github.com/aws/aws-sdk-go/service/sts"
	"github.com/pingcap/log"
	"go.uber.org/zap"

	"github.com/pingcap-inc/clinic/pkg/api/common"
	"github.com/pingcap-inc/clinic/pkg/api/healthcheck"
	"github.com/pingcap-inc/clinic/pkg/config"
	"github.com/pingcap-inc/clinic/pkg/db"
	"github.com/pingcap-inc/clinic/pkg/db/model"
	"github.com/pingcap-inc/clinic/pkg/o11y"
	"github.com/pingcap-inc/clinic/pkg/task_manager/task"
	"github.com/pingcap-inc/clinic/pkg/task_manager/task/healthcheck/jobs/configcheck"
	"github.com/pingcap-inc/clinic/pkg/task_manager/task/healthcheck/jobs/diskcheck"
)

const (
	healthCheckEndpointSlowQueryAnalyze = "http://clinic-data-service:8087/clinic-internal/healthcheck/sql/slowquery-analyze"
	healthCheckEndpointMetricsAnalyze   = "http://clinic-data-service:8087/clinic-internal/healthcheck/metrics/analyze"
	healthCheckEndpointTableFullScan    = "http://sql-health-check:8080/clinic-internal/healthcheck/sql/tablefullscan-slowquery"
	healthCheckEndpointSlowLogTopDigest = "http://sql-health-check:8080/clinic-internal/healthcheck/sql/slowlog-top-digest"
	healthCheckEndpointTiKVHighDuration = "http://sql-health-check:8080/clinic-internal/healthcheck/sql/tikv-high-duration"
	healthCheckEndpointTiKVHighCPU      = "http://sql-health-check:8080/clinic-internal/healthcheck/sql/tikv-high-cpu"
)

// Job represents a health check job
type Job struct {
	cfg    *config.Config
	result *db.HealthCheckResult
}

// NewJob creates a new health check job
func NewJob(cfg *config.Config, result *db.HealthCheckResult) *Job {
	return &Job{
		cfg:    cfg,
		result: result,
	}
}

// Run executes the health check job
func (j *Job) Run(kc task.K8sClient) error {
	log.Info("start health check job",
		zap.String("uuid", j.result.UUID),
		zap.String("cluster_id", j.result.ClusterID),
	)

	// Double-check the status
	result, err := repository.GetHealthCheckResultByUUID(j.result.UUID)
	if err != nil {
		log.Error("get health check result failed", zap.Error(err))
		return err
	}
	j.result = result

	if db.StatusType(result.Status) == db.StatusSuccess {
		log.Info("health check job already succeeded", zap.String("uuid", j.result.UUID))
		return nil
	}

	if db.StatusType(result.Status) == db.StatusFailed {
		log.Info("health check job already failed", zap.String("uuid", j.result.UUID))
		return nil
	}

	// Handle based on current status
	switch db.StatusType(result.Status) {
	case db.StatusPending:
		// First time: update to running, call endpoint, then check S3
		return j.handlePending()
	case db.StatusRunning, db.StatusPartial:
		// Already called endpoint or partial results, check S3 for updates
		return j.handleRunning()
	default:
		log.Warn("unknown status, skipping",
			zap.String("uuid", result.UUID),
			zap.String("status", result.Status),
		)
		return nil
	}
}

// handlePending handles first-time execution (status = pending)
func (j *Job) handlePending() error {
	log.Info("job is pending, calling endpoint",
		zap.String("uuid", j.result.UUID),
		zap.String("cluster_id", j.result.ClusterID),
		zap.String("dbaas_org_id", j.result.DbaasOrgID),
	)

	// Update status to running
	if err := repository.UpdateHealthCheckResultStatus(j.result.UUID, string(db.StatusRunning), ""); err != nil {
		log.Error("update health check result status to running failed", zap.Error(err))
		return err
	}

	// Get cluster info
	cluster, err := model.GetAnyClusterWithOrg(j.result.ClusterID)
	if err != nil {
		log.Error("get cluster failed", zap.Error(err))
		j.failJob(fmt.Sprintf("get cluster failed: %v", err))
		return err
	}

	// Get items to run from result
	itemsToRun := j.getItemsToRun()
	if len(itemsToRun) == 0 {
		log.Error("no items to run")
		j.failJob("no items to run")
		return fmt.Errorf("no items to run")
	}

	// Track execution results
	successCount := 0
	failedCount := 0

	// Call endpoint for each item
	for i := range itemsToRun {
		item := &itemsToRun[i]

		// Build the health check request
		req, err := j.buildHealthCheckRequest(cluster, *item)
		if err != nil {
			log.Error("build health check request failed", zap.Error(err), zap.String("item", item.Name))
			item.Failed = true
			item.ErrorMsg = fmt.Sprintf("build request failed: %v", err)
			failedCount++
			continue
		}

		var callErr error

		// Embedded job
		switch item.Name {
		case "config_check":
			log.Info("executing embedded job", zap.String("item", item.Name), zap.String("uuid", j.result.UUID), zap.String("cluster_id", j.result.ClusterID))
			callErr = configcheck.AsyncRun(req, &j.cfg.OpsDBConfig)
		case "disk_check":
			log.Info("executing embedded job", zap.String("item", item.Name), zap.String("uuid", j.result.UUID), zap.String("cluster_id", j.result.ClusterID))
			callErr = diskcheck.AsyncRun(req)
		default:
			// Remote job, call the health check endpoint
			callErr = j.callHealthCheckEndpoint(req)
		}

		if callErr != nil {
			log.Error("call health check endpoint/job failed",
				zap.Error(callErr),
				zap.String("item", item.Name),
			)
			item.Failed = true
			item.ErrorMsg = fmt.Sprintf("execution failed: %v", callErr)
			failedCount++
		} else {
			successCount++
		}
	}

	// Save items with failure status back to result
	if err := j.saveItemsToRun(itemsToRun); err != nil {
		log.Error("save items to run failed", zap.Error(err))
		return err
	}

	// Determine final status based on success/failure counts
	if failedCount == len(itemsToRun) {
		// All items failed
		j.failJob(fmt.Sprintf("all %d items failed", failedCount))
		return fmt.Errorf("all items failed")
	} else if failedCount > 0 {
		// Partial failure: some items failed, some succeeded
		// Status remains "running" and handleRunning will check S3 and determine final status
		log.Info("partial failure: some items failed, waiting for S3 results",
			zap.String("uuid", j.result.UUID),
			zap.Int("success_count", successCount),
			zap.Int("failed_count", failedCount),
		)
		// Save items with failure info, but keep running status
		// Error message will be set when S3 check completes
	}
	// If all succeeded, status remains "running" and handleRunning will check S3

	return nil
}

// handleRunning handles subsequent checks (status = running, partial, or partial_failed)
func (j *Job) handleRunning() error {
	log.Info("job is running, checking S3 result",
		zap.String("uuid", j.result.UUID),
		zap.String("cluster_id", j.result.ClusterID),
		zap.String("dbaas_org_id", j.result.DbaasOrgID),
		zap.String("status", j.result.Status),
	)

	// Get items to run from result
	itemsToRun := j.getItemsToRun()
	if len(itemsToRun) == 0 {
		log.Error("no items to run")
		j.failJob("no items to run")
		return fmt.Errorf("no items to run")
	}

	// Filter out failed items - only check S3 for successful items
	var resultPaths []string
	failedCount := 0
	for _, item := range itemsToRun {
		if item.Failed {
			// Skip S3 check for failed items
			failedCount++
			log.Info("skipping S3 check for failed item",
				zap.String("uuid", j.result.UUID),
				zap.String("cluster_id", j.result.ClusterID),
				zap.String("item", item.Name),
			)
			continue
		}
		resultPath := fmt.Sprintf("healthcheck-report/%s_%s_%s.json", j.result.ClusterID, j.result.UUID, item.Name)
		resultPaths = append(resultPaths, resultPath)
	}

	log.Info("checking S3 for successful items",
		zap.String("uuid", j.result.UUID),
		zap.String("cluster_id", j.result.ClusterID),
		zap.String("dbaas_org_id", j.result.DbaasOrgID),
		zap.Int("total_items", len(itemsToRun)),
		zap.Int("failed_items", failedCount),
		zap.Int("items_to_check", len(resultPaths)),
	)

	return j.checkS3AndComplete(resultPaths, itemsToRun)
}

// checkS3AndComplete checks S3 for all result paths and updates status based on results:
// - All paths exist: status = success (if no failed items) or partial_failed (if some items failed)
// - Some paths exist: status = partial (still waiting for some S3 results)
// - No paths exist: keep running status for retry
func (j *Job) checkS3AndComplete(resultPaths []string, itemsToRun []healthcheck.CheckItem) error {
	// Count failed items from itemsToRun
	failedItemCount := 0
	failedItemErrors := []string{}
	for _, item := range itemsToRun {
		if item.Failed {
			failedItemCount++
			if item.ErrorMsg != "" {
				failedItemErrors = append(failedItemErrors, fmt.Sprintf("%s: %s", item.Name, item.ErrorMsg))
			}
		}
	}

	// Handle case where all items failed during execution
	if len(resultPaths) == 0 && failedItemCount > 0 {
		errorMsg := "all items failed during execution"
		if len(failedItemErrors) > 0 {
			errorMsg = strings.Join(failedItemErrors, "; ")
		}
		j.failJob(errorMsg)
		return nil
	}

	existingCount := 0
	errorCount := 0
	missingResultPaths := make([]string, 0)

	for _, resultPath := range resultPaths {
		exists, err := j.checkS3ObjectExists(resultPath)
		if err != nil {
			log.Error("check s3 object failed",
				zap.Error(err),
				zap.String("uuid", j.result.UUID),
				zap.String("cluster_id", j.result.ClusterID),
				zap.String("result_path", resultPath),
			)
			errorCount++
			continue
		}
		if exists {
			existingCount++
		} else {
			missingResultPaths = append(missingResultPaths, resultPath)
		}
	}

	// If no results are ready yet, keep running status for retry
	if existingCount == 0 {
		log.Info("health check results not ready in s3, will retry",
			zap.String("uuid", j.result.UUID),
			zap.String("cluster_id", j.result.ClusterID),
			zap.Int("total_paths", len(resultPaths)),
			zap.Strings("missing_result_paths", missingResultPaths),
		)
		return nil
	}

	// Determine status based on how many results are ready
	var newStatus db.StatusType
	var errorMsg string
	hasFailedItems := failedItemCount > 0
	currentStatus := db.StatusType(j.result.Status)

	if existingCount == len(resultPaths) {
		// All remaining items (non-failed) have results in S3
		if hasFailedItems {
			// Some items failed initially, but remaining items are all ready
			newStatus = db.StatusPartialFailed
			errorMsg = fmt.Sprintf("%d of %d items failed", failedItemCount, len(itemsToRun))
			if len(failedItemErrors) > 0 {
				errorMsg = strings.Join(failedItemErrors, "; ")
			}
		} else {
			// All items succeeded initially and all results ready
			newStatus = db.StatusSuccess
		}
	} else {
		// Some results are ready, some are still pending
		newStatus = db.StatusPartial
	}

	// Only update status if it changed
	if newStatus != currentStatus {
		if err := repository.UpdateHealthCheckResultStatus(j.result.UUID, string(newStatus), errorMsg); err != nil {
			log.Error("update health check result status failed",
				zap.Error(err),
				zap.String("uuid", j.result.UUID),
				zap.String("status", string(newStatus)),
			)
			return err
		}

		log.Info("health check job status updated",
			zap.String("uuid", j.result.UUID),
			zap.String("cluster_id", j.result.ClusterID),
			zap.String("old_status", string(currentStatus)),
			zap.String("new_status", string(newStatus)),
			zap.Int("existing_count", existingCount),
			zap.Int("total_paths", len(resultPaths)),
			zap.Int("error_count", errorCount),
			zap.Int("failed_items", failedItemCount),
			zap.Strings("missing_result_paths", missingResultPaths),
		)
	} else if newStatus == db.StatusPartial {
		// Still waiting, log progress
		log.Info("health check job still waiting for s3 results",
			zap.String("uuid", j.result.UUID),
			zap.String("cluster_id", j.result.ClusterID),
			zap.String("status", string(newStatus)),
			zap.Int("existing_count", existingCount),
			zap.Int("total_paths", len(resultPaths)),
			zap.Int("error_count", errorCount),
			zap.Int("failed_items", failedItemCount),
			zap.Strings("missing_result_paths", missingResultPaths),
		)
	}

	return nil
}

// getItemsToRun returns the list of items to run for this job.
func (j *Job) getItemsToRun() []healthcheck.CheckItem {
	// Try to parse ItemsToRun as []CheckItem first (new format)
	if j.result.ItemsToRun != "" && j.result.ItemsToRun != "null" {
		var items []healthcheck.CheckItem
		if err := json.Unmarshal([]byte(j.result.ItemsToRun), &items); err == nil && len(items) > 0 {
			// Verify it's the new format by checking if first item has a name
			if items[0].Name != "" {
				return items
			}
		}
	}
	return nil
}

// saveItemsToRun saves the items with their execution status back to the result
func (j *Job) saveItemsToRun(items []healthcheck.CheckItem) error {
	itemsJSON, err := json.Marshal(items)
	if err != nil {
		return fmt.Errorf("marshal items to run failed: %w", err)
	}

	result, err := repository.GetHealthCheckResultByUUID(j.result.UUID)
	if err != nil {
		return fmt.Errorf("get health check result failed: %w", err)
	}

	result.ItemsToRun = string(itemsJSON)
	if err := repository.UpdateHealthCheckResult(result); err != nil {
		return fmt.Errorf("update health check result items failed: %w", err)
	}

	return nil
}

func (j *Job) buildHealthCheckRequest(cluster *db.Cluster, item healthcheck.CheckItem) (*healthcheck.HealthCheckJobRequest, error) {
	// Format time ranges
	formatTime := func(t *time.Time) string {
		if t == nil {
			return time.Now().UTC().Format(time.RFC3339)
		}
		return t.UTC().Format(time.RFC3339)
	}

	// Get time ranges
	startA := formatTime(j.result.TimeRangeAStart)
	endA := formatTime(j.result.TimeRangeAEnd)
	// startB := formatTime(result.TimeRangeBStart)
	// endB := formatTime(result.TimeRangeBEnd)

	// Parse cluster URLs from result (stored by clinic server at trigger time)
	var clusterURLs healthcheck.HealthCheckClusterUrls
	if j.result.ClusterURLs != "" && j.result.ClusterURLs != "null" {
		if err := json.Unmarshal([]byte(j.result.ClusterURLs), &clusterURLs); err != nil {
			return nil, fmt.Errorf("failed to parse cluster URLs: %w", err)
		}
	} else {
		return nil, fmt.Errorf("cluster URLs not found in result")
	}

	log.Info("using pre-resolved cluster URLs",
		zap.String("cluster_id", j.result.ClusterID),
		zap.String("cluster_metrics_url", clusterURLs.ClusterMetricsUrl),
		zap.String("dataplane_metrics_url", clusterURLs.DataPlaneMetricsUrl),
	)

	var matchVersion string
	if cluster.DeployType == o11y.NgHostType || cluster.DeployType == o11y.PremiumType || cluster.DeployType == o11y.ByocType || cluster.DeployType == o11y.ByocHostType {
		matchVersion = common.GetVersion(cluster.Version, j.cfg.SinkConfig.MetricConfig.SortedNextGenDashboardVersion)
	} else {
		matchVersion = common.GetVersion(cluster.Version, j.cfg.SinkConfig.MetricConfig.SortedDashboardVersion)
	}

	req := &healthcheck.HealthCheckJobRequest{
		Name:              item.Name,
		ClusterId:         j.result.ClusterID,
		PhysicalClusterId: cluster.ParentID,
		ReportId:          j.result.UUID,
		DeployType:        cluster.DeployType,
		ProjectId:         cluster.ProjectID,
		Provider:          cluster.Provider,
		Region:            cluster.Region,
		ClinicOrgId:       fmt.Sprintf("%d", cluster.OrgID),
		TenantId:          fmt.Sprintf("%d", cluster.TenantID),
		Version:           cluster.Version,
		DashboardVersion:  matchVersion,
		TimeRange: healthcheck.HealthCheckTimeRange{
			Start: startA,
			End:   endA,
		},
		// TimeRangeB: healthcheck.HealthCheckTimeRange{
		// 	Start: startB,
		// 	End:   endB,
		// },
		ClusterUrls: clusterURLs,
		ClusterMeta: buildClusterMeta(cluster, j.result),
		DataSource:  clusterURLs.SlowQueryS3Url, // TODO: remove
		Options: map[string]string{
			"region": cluster.Region,
		},
	}

	paramsMap := map[string]uint64{}
	_ = json.Unmarshal([]byte(item.Params), &paramsMap)
	if len(paramsMap) > 0 {
		if memMaxThreshold, ok := paramsMap["memMaxThreshold"]; ok {
			req.MemMaxThreshold = memMaxThreshold
		}
		if queryTimeThreshold, ok := paramsMap["queryTimeThreshold"]; ok {
			req.QueryTimeThreshold = queryTimeThreshold
		}
	}
	itemParams, err := parseItemParams(item.Params)
	if err != nil {
		return nil, err
	}
	req.Params = itemParams

	return req, nil
}

func buildClusterMeta(cluster *db.Cluster, result *db.HealthCheckResult) map[string]string {
	meta := map[string]string{
		"clusterName": cluster.GetRawName(),
		"orgName":     cluster.Organization.GetRawName(),
		"projectId":   cluster.ProjectID,
		"provider":    cluster.Provider,
		"vendor":      cluster.Provider,
		"region":      cluster.Region,
		"deployType":  cluster.DeployType,
	}
	if cluster.Organization.DbaasOrgID != 0 {
		meta["orgId"] = fmt.Sprintf("%d", cluster.Organization.DbaasOrgID)
	} else if result != nil && result.DbaasOrgID != "" {
		meta["orgId"] = result.DbaasOrgID
	} else if cluster.OrgID != 0 {
		meta["orgId"] = fmt.Sprintf("%d", cluster.OrgID)
	}
	return meta
}

func parseItemParams(params string) (json.RawMessage, error) {
	if strings.TrimSpace(params) == "" {
		return nil, nil
	}
	var raw json.RawMessage
	if err := json.Unmarshal([]byte(params), &raw); err != nil {
		return nil, fmt.Errorf("invalid item params JSON: %w", err)
	}
	var compact bytes.Buffer
	if err := json.Compact(&compact, raw); err != nil {
		return nil, fmt.Errorf("compact item params JSON failed: %w", err)
	}
	return compact.Bytes(), nil
}

func (j *Job) callHealthCheckEndpoint(req *healthcheck.HealthCheckJobRequest) error {
	jsonData, err := json.Marshal(req)
	if err != nil {
		return fmt.Errorf("marshal request failed: %w", err)
	}

	healthCheckEndpoint, ok := healthCheckEndpointForItem(req.Name)
	if !ok {
		return fmt.Errorf("unknown check item name: %s", req.Name)
	}

	httpReq, err := http.NewRequest(http.MethodPost, healthCheckEndpoint, bytes.NewBuffer(jsonData))
	if err != nil {
		return fmt.Errorf("create request failed: %w", err)
	}

	httpReq.Header.Set("Content-Type", "application/json")
	httpReq.Header.Set("Authorization", fmt.Sprintf("Bearer %s", j.cfg.DataServiceConfig.Token))

	client := &http.Client{
		Timeout: 30 * time.Second,
	}

	resp, err := client.Do(httpReq)
	if err != nil {
		return fmt.Errorf("send request failed: %w", err)
	}
	defer func() {
		_ = resp.Body.Close()
	}()

	body, err := io.ReadAll(resp.Body)
	if err != nil {
		return fmt.Errorf("read response body failed: %w", err)
	}

	if resp.StatusCode != http.StatusOK && resp.StatusCode != http.StatusAccepted {
		return fmt.Errorf("health check endpoint returned status %d: %s", resp.StatusCode, string(body))
	}

	log.Info("health check endpoint response",
		zap.String("uuid", j.result.UUID),
		zap.String("response", string(body)),
	)

	return nil
}

func healthCheckEndpointForItem(name string) (string, bool) {
	switch name {
	case "slow_query_hugeResourceUsage":
		return healthCheckEndpointSlowQueryAnalyze, true
	case "tablefullscan_slow_query":
		return healthCheckEndpointTableFullScan, true
	case "slowlog_top_digest":
		return healthCheckEndpointSlowLogTopDigest, true
	case "sql_tikv_high_duration":
		return healthCheckEndpointTiKVHighDuration, true
	case "sql_tikv_high_cpu":
		return healthCheckEndpointTiKVHighCPU, true
	case "metrics_healthcheck":
		return healthCheckEndpointMetricsAnalyze, true
	default:
		return "", false
	}
}

func (j *Job) failJob(errorMsg string) {
	if err := repository.UpdateHealthCheckResultStatus(j.result.UUID, string(db.StatusFailed), errorMsg); err != nil {
		log.Error("update health check result status to failed failed",
			zap.Error(err),
			zap.String("uuid", j.result.UUID),
		)
	}
}

// checkS3ObjectExists checks if the health check result exists in S3
func (j *Job) checkS3ObjectExists(s3Key string) (bool, error) {
	region := os.Getenv("AWS_REGION")
	if region == "" {
		region = "us-west-2"
	}
	bucket := os.Getenv("HEALTHCHECK_S3_BUCKET")
	if bucket == "" {
		bucket = "delicated-healthcheck"
	}

	// Role ARN for cross-account access
	assumeRoleARN := "arn:aws:iam::552185537300:role/clinic-job-role"

	sess, err := session.NewSession(&aws.Config{
		Region: aws.String(region),
	})
	if err != nil {
		return false, fmt.Errorf("failed to create s3 session: %w", err)
	}

	// Create STS client and assume role
	stsClient := sts.New(sess)
	creds := stscreds.NewCredentialsWithClient(stsClient, assumeRoleARN)

	// Create S3 client with assumed role credentials
	s3Client := s3.New(sess, &aws.Config{Credentials: creds})

	_, err = s3Client.HeadObject(&s3.HeadObjectInput{
		Bucket: aws.String(bucket),
		Key:    aws.String(s3Key),
	})
	if err != nil {
		if isS3ObjectNotFoundError(err) {
			return false, nil
		}
		return false, err
	}

	return true, nil
}

func isS3ObjectNotFoundError(err error) bool {
	if err == nil {
		return false
	}

	var reqErr awserr.RequestFailure
	if errors.As(err, &reqErr) && reqErr.StatusCode() == http.StatusNotFound {
		return true
	}

	var awsErr awserr.Error
	if errors.As(err, &awsErr) {
		switch awsErr.Code() {
		case s3.ErrCodeNoSuchKey, "NotFound", "404":
			return true
		}
	}

	text := strings.ToLower(err.Error())
	return strings.Contains(text, "status code: 404") ||
		strings.Contains(text, "nosuchkey") ||
		strings.Contains(text, "notfound") ||
		strings.Contains(text, "not found")
}