package s3

import (
	"context"
	"fmt"
	"os"
	"path/filepath"
	"sync"

	"github.com/aws/aws-sdk-go/aws"
	"github.com/aws/aws-sdk-go/service/s3"
	"go.uber.org/zap"

	"github.com/pingcap-inc/clinic/clinic_data_service/pkg/common"
	"github.com/pingcap-inc/clinic/clinic_data_service/pkg/config"
	log2 "github.com/pingcap-inc/clinic/clinic_data_service/pkg/log"
)

// UploadFile uploads a local file to S3
func (c *S3Client) UploadFile(localFilePath string, s3Key string) error {
	// Check if local file exists
	file, err := os.Open(localFilePath)
	if err != nil {
		return fmt.Errorf("failed to open local file %s: %w", localFilePath, err)
	}
	defer file.Close()

	// Get file information
	fileInfo, err := file.Stat()
	if err != nil {
		return fmt.Errorf("failed to get file info: %w", err)
	}

	// Upload file to S3
	_, err = c.client.PutObject(&s3.PutObjectInput{
		Bucket:        aws.String(c.bucket),
		Key:           aws.String(s3Key),
		Body:          file,
		ContentLength: aws.Int64(fileInfo.Size()),
	})

	if err != nil {
		return fmt.Errorf("failed to upload file to S3: %w", err)
	}

	return nil
}

// UploadFilesToS3 uploads files to S3 based on configuration
func UploadFilesToS3(localFiles []string, s3Prefix string, orgID string, clusterID string) error {
	cfg := &config.AppConfig

	// Get S3 client
	s3Client, err := GetS3Client(cfg)
	if err != nil {
		return fmt.Errorf("failed to get S3 client: %v", err)
	}

	var wg sync.WaitGroup
	errors := make(chan error, len(localFiles))

	// Upload each file to S3
	for i, localFile := range localFiles {
		wg.Add(1)
		go func(file string, i int) {
			defer wg.Done()
			log2.PlainInfo(fmt.Sprintf("Uploading file %d/%d: %s", i+1, len(localFiles), file))

			// Generate S3 file path
			fileName := filepath.Base(file)
			s3Key := fmt.Sprintf("%s/%s/%s/%s", s3Prefix, orgID, clusterID, fileName)

			// Upload file
			if err := s3Client.UploadFile(file, s3Key); err != nil {
				log2.PlainError("Failed to upload file", zap.String("file", file), zap.Error(err))
				errors <- err
				return
			}

			log2.PlainInfo("File uploaded successfully: " + file + " -> " + s3Key)
		}(localFile, i)
	}

	// Wait for all upload tasks to complete
	wg.Wait()
	close(errors)

	// Check if there are any errors
	for err := range errors {
		if err != nil {
			return fmt.Errorf("some files failed to upload: %v", err)
		}
	}

	log2.PlainInfo("All files uploaded successfully")
	return nil
}

// UploadFilesFromConfig uploads files to S3 based on configuration
func UploadFilesFromConfig(localFiles []string, targetPrefix string) error {
	orgs := &config.OrgsConfig

	for _, org := range orgs.Orgs {
		for _, cluster := range org.Clusters {
			// cluster := config.GetClusterByID(cluster.ClusterID)
			cluster, err := common.GetCluster(context.Background(), cluster.ClusterID)
			if err != nil {
				return fmt.Errorf("failed to get cluster: %v", err)
			}
			if cluster == nil {
				log2.PlainError("Cluster not found", zap.String("cluster", cluster.ClusterID))
				continue
			}

			// Get S3 client
			s3Client, err := GetS3ClientByGithubToken(cluster.Region)
			if err != nil {
				return fmt.Errorf("failed to get S3 client: %v", err)
			}

			// Build S3 prefix
			s3Prefix := targetPrefix
			if s3Prefix == "" {
				// Use default prefix format
				date := config.AppConfig.Day
				s3Prefix = fmt.Sprintf("diagnosis/data/%s/slowlog-merged/1h/%s", cluster.ClusterID, date)
			}

			// Upload files
			var wg sync.WaitGroup
			errors := make(chan error, len(localFiles))

			for i, localFile := range localFiles {
				wg.Add(1)
				go func(file string, i int) {
					defer wg.Done()
					log2.PlainInfo(fmt.Sprintf("Uploading file %d/%d: %s to cluster %s (%s)", i+1, len(localFiles), file, cluster.ClusterID, cluster.Alias))

					// Generate S3 file path
					fileName := filepath.Base(file)
					s3Key := fmt.Sprintf("%s/%s", s3Prefix, fileName)

					// Upload file
					if err := s3Client.UploadFile(file, s3Key); err != nil {
						log2.PlainError("Failed to upload file", zap.String("file", file), zap.String("cluster", cluster.ClusterID), zap.String("alias", cluster.Alias), zap.Error(err))
						errors <- err
						return
					}

					log2.PlainInfo("File uploaded successfully: " + file + " -> " + s3Key)
				}(localFile, i)
			}

			// Wait for all upload tasks to complete
			wg.Wait()
			close(errors)

			// Check if there are any errors
			for err := range errors {
				if err != nil {
					return fmt.Errorf("some files failed to upload to cluster %s (%s): %v", cluster.ClusterID, cluster.Alias, err)
				}
			}
		}
	}

	log2.PlainInfo("All files uploaded successfully")
	return nil
}

func DownloadS3Files(ctx context.Context, clusterId, day, prefixHour string, saveDir ...string) error {

	cfg := &config.AppConfig
	// org := config.GetOrgByClusterID(clusterId)
	// cluster := config.GetClusterByID(clusterId)
	cluster, err := common.GetCluster(ctx, clusterId)
	if err != nil {
		log2.PlainError("Failed to get cluster", zap.String("cluster", clusterId), zap.Error(err))
		return err
	}

	s3Client, err := GetS3ClientByGithubToken(cluster.Region)

	if err != nil {
		log2.PlainError("Failed to get S3 client", zap.String("cluster", cluster.ClusterID), zap.String("alias", cluster.Alias), zap.Error(err))
		return err
	}

	s3Prefix := fmt.Sprintf("diagnosis/data/%s/slowlog-merged/1h/%s%s", cluster.ClusterID, day, prefixHour)
	log2.PlainInfo(fmt.Sprintf("down file: %s for cluster %s (%s)", s3Prefix, cluster.ClusterID, cluster.Alias))
	logFiles, err := s3Client.ListFiles(s3Prefix)
	if err != nil {
		log2.PlainError("Failed to list S3 log files", zap.Error(err))
		return err
	}
	if len(logFiles) == 0 {
		log2.PlainInfo(fmt.Sprintf("no log files found for cluster %s (%s)", cluster.ClusterID, cluster.Alias))
		return nil
	}
	// date := config.AppConfig.Day
	var localDir string
	if len(saveDir) > 0 {
		localDir = saveDir[0]
	} else {
		localDir = fmt.Sprintf("%s/%s/%s/%s", config.AppConfig.InputDir, day, cluster.OrgID, cluster.ClusterID)
	}
	parser := &S3LogParser{s3Client: s3Client, config: cfg}
	parser.downloadS3Files(ctx,logFiles, localDir)

	return nil
}