package process

import (
	"bytes"
	"context"
	"errors"
	"fmt"
	"io"
	"path"
	"strconv"
	"strings"
	"sync"
	"time"

	gcs "cloud.google.com/go/storage"
	"github.com/Azure/azure-sdk-for-go/sdk/azidentity"
	"github.com/Azure/azure-sdk-for-go/sdk/storage/azblob"
	"github.com/aliyun/alibabacloud-oss-go-sdk-v2/oss"
	"github.com/aliyun/alibabacloud-oss-go-sdk-v2/oss/credentials"
	openapicred "github.com/aliyun/credentials-go/credentials"
	awsConfig "github.com/aws/aws-sdk-go-v2/config"
	"github.com/aws/aws-sdk-go-v2/service/s3"
	"github.com/pingcap/log"
	"github.com/tidbcloud/observability/cmd/diagnosis-compute/pkg/config"
	"github.com/tidbcloud/observability/cmd/diagnosis-compute/pkg/types"
	"github.com/tidbcloud/observability/o11ydiagnosis/pkg/storage"
	"go.uber.org/zap"
	"gopkg.in/yaml.v2"
)

func LoadClusterInfo(ctx context.Context, store storage.Storage, bucket, key string) (*types.ClusterInfo, error) {
	obj, err := store.Object(ctx, bucket, key)
	if err != nil {
		return nil, err
	}
	defer func() {
		if err := obj.Reader.Close(); err != nil {
			log.Error("failed to close object", zap.Error(err))
		}
	}()
	b, err := io.ReadAll(obj.Reader)
	if err != nil {
		return nil, err
	}
	var info types.S3ClusterInfo
	if err := yaml.Unmarshal(b, &info); err != nil {
		return nil, err
	}
	if info.Meta.ClusterType != types.ClusterTypeDedicated && info.Meta.ClusterType != types.ClusterTypeNextgen {
		return nil, fmt.Errorf("invalid cluster type: %s", info.Meta.ClusterType)
	}
	clusterInfo := &types.ClusterInfo{
		Type:        0,
		ClusterType: info.Meta.ClusterType,
		Env:         info.Meta.Env,
		TenantID:    info.Meta.TenantID,
		ProjectID:   info.Meta.ProjectID,
		ClusterID:   info.Meta.ClusterID,
		StorageID:   info.Meta.ClusterID,
	}
	if clusterInfo.Env == "" {
		clusterInfo.Env = config.Env
	}
	if info.Meta.ClusterType == types.ClusterTypeNextgen {
		subClusterType, err := getSubClusterType(ctx, store, info.Meta.ClusterID)
		if err != nil {
			return nil, err
		}
		if subClusterType == "" {
			return nil, fmt.Errorf("non supported cluster: %s", info.Meta.ClusterID)
		}
		clusterInfo.SubClusterType = subClusterType
	}
	return clusterInfo, nil
}

func LoadStorage(ctx context.Context) (storage.Storage, error) {
	switch config.Vendor {
	case "aws":
		cfg, err := awsConfig.LoadDefaultConfig(ctx, awsConfig.WithRegion(config.Region))
		if err != nil {
			return nil, err
		}
		return storage.NewAwsStorage(s3.NewFromConfig(cfg)), nil
	case "gcp":
		client, err := gcs.NewClient(ctx)
		if err != nil {
			return nil, err
		}
		return storage.NewGcpStorage(client), nil
	case "azure":
		cred, err := azidentity.NewDefaultAzureCredential(nil)
		if err != nil {
			return nil, fmt.Errorf("failed to create azure credential: %w", err)
		}
		client, err := azblob.NewClient(fmt.Sprintf("https://o11yprod%s.blob.core.windows.net", config.Region), cred, nil)
		if err != nil {
			return nil, fmt.Errorf("failed to create azure blob client: %w", err)
		}
		return storage.NewAzureStorage(client), nil
	case "alicloud", "aliyun":
		cred, err := openapicred.NewCredential(nil)
		if err != nil {
			return nil, fmt.Errorf("failed to create default AliCloud credentials: %w", err)
		}
		provider := credentials.CredentialsProviderFunc(func(ctx context.Context) (credentials.Credentials, error) {
			cred, err := cred.GetCredential()
			if err != nil {
				return credentials.Credentials{}, err
			}
			return credentials.Credentials{
				AccessKeyID:     *cred.AccessKeyId,
				AccessKeySecret: *cred.AccessKeySecret,
				SecurityToken:   *cred.SecurityToken,
			}, nil
		})
		cfg := oss.LoadDefaultConfig().
			WithRegion(config.Region).
			WithCredentialsProvider(provider).
			WithEndpoint(fmt.Sprintf("http://oss-%s-internal.aliyuncs.com", config.Region))
		client := oss.NewClient(cfg)
		return storage.NewAlicloudStorage(client), nil
	default:
		return nil, fmt.Errorf("invalid vendor: %s", config.Vendor)
	}
}

func loadCheckpoint(ctx context.Context, store storage.Storage, storageID string) (int64, error) {
	bucket := config.BucketName()
	obj, err := store.Object(ctx, bucket, checkpointKey(storageID))
	if err != nil {
		if err == storage.ErrNotFound {
			return 0, nil
		}
		return 0, err
	}
	content, err := io.ReadAll(obj.Reader)
	if err != nil {
		return 0, err
	}
	checkpointBody := map[string]int64{}
	if err := yaml.Unmarshal(content, &checkpointBody); err != nil {
		return 0, err
	}
	if err := obj.Reader.Close(); err != nil {
		return 0, err
	}
	checkpointTs, ok := checkpointBody["checkpoint_time"]
	if !ok {
		return 0, errors.New("invalid checkpoint content")
	}
	return checkpointTs, nil
}

func saveCheckpoint(ctx context.Context, store storage.Storage, storageID string, checkpointTs int64) error {
	bucket := config.BucketName()
	b, err := yaml.Marshal(map[string]int64{"checkpoint_time": checkpointTs})
	if err != nil {
		return err
	}
	return store.SaveObject(ctx, bucket, checkpointKey(storageID), bytes.NewReader(b))
}

func scanComponents(ctx context.Context, store storage.Storage, info types.ClusterInfo) ([]string, error) {
	bucket := config.BucketName()
	dir := getLogPrefix(info)
	prefixes, err := store.ObjectsDelimiter(ctx, bucket, dir)
	if err != nil {
		return nil, err
	}
	components := []string{}
	for n := range prefixes {
		prefix := strings.TrimSuffix(strings.TrimPrefix(prefixes[n], dir), "/")
		if config.UseComponentWhitelist {
			if _, ok := config.ComponentWhitelist[prefix]; ok {
				components = append(components, prefix)
			}
		} else {
			if prefix == "slowlog" || prefix == "statementlog" {
				continue
			}
			components = append(components, prefix)
		}
	}
	return components, nil
}

func findMatchedKeys(ctx context.Context,
	store storage.Storage,
	bucket string,
	info types.ClusterInfo,
	component string,
	begin, end int64) []string {
	dir := getLogPrefix(info) + component
	metas := listUnixTsPrefixedObjects(ctx, store, bucket, dir, begin, end)
	keys := make([]string, 0, len(metas))
	for _, meta := range metas {
		keys = append(keys, meta.Key)
	}
	return keys
}

// listUnixTsPrefixedObjects lists objects under dir whose filename starts with a
// 10-digit unix timestamp within [begin, end]. Used by logs/ and profiles/.
func listUnixTsPrefixedObjects(
	ctx context.Context,
	store storage.Storage,
	bucket, dir string,
	begin, end int64,
) []storage.ObjectMeta {
	prefixes := []string{}
	for i := begin / 1000; i <= end/1000; i++ {
		prefixes = append(prefixes, fmt.Sprintf("%s/%d", dir, i))
	}
	out := []storage.ObjectMeta{}
	lock := sync.Mutex{}
	wg := sync.WaitGroup{}
	for _, prefix := range prefixes {
		wg.Add(1)
		go func(prefix string) {
			defer wg.Done()
			metas, err := store.Objects(ctx, bucket, prefix)
			if err != nil {
				log.Error("failed to list objects", zap.Error(err), zap.String("prefix", prefix))
				return
			}
			lock.Lock()
			for _, meta := range metas {
				filename := path.Base(meta.Key)
				uploadTs, ok := unixTsFromFilename(filename)
				if !ok {
					continue
				}
				if uploadTs < begin || uploadTs > end {
					continue
				}
				out = append(out, meta)
			}
			lock.Unlock()
		}(prefix)
	}
	wg.Wait()
	return out
}

func unixTsFromFilename(filename string) (int64, bool) {
	if len(filename) < 10 {
		return 0, false
	}
	uploadTs, err := strconv.ParseInt(filename[:10], 10, 64)
	if err != nil {
		return 0, false
	}
	return uploadTs, true
}

// listRegionObjectsInHour lists regions/ objects for one UTC hour.
// Filenames look like: 202602040000.json.log.gz (yyyyMMddHHmm).
func listRegionObjectsInHour(
	ctx context.Context,
	store storage.Storage,
	bucket, dir string,
	hourBegin, hourEnd int64,
) []storage.ObjectMeta {
	hourPrefix := time.Unix(hourBegin, 0).UTC().Format("2006010215")
	metas, err := store.Objects(ctx, bucket, dir+"/"+hourPrefix)
	if err != nil {
		log.Error("failed to list region objects", zap.Error(err), zap.String("dir", dir), zap.String("hour", hourPrefix))
		return nil
	}
	out := make([]storage.ObjectMeta, 0, len(metas))
	for _, meta := range metas {
		ts, ok := regionFileTime(path.Base(meta.Key))
		if !ok {
			continue
		}
		if ts < hourBegin || ts > hourEnd {
			continue
		}
		out = append(out, meta)
	}
	return out
}

func regionFileTime(filename string) (int64, bool) {
	base := filename
	for {
		ext := path.Ext(base)
		if ext == "" {
			break
		}
		base = strings.TrimSuffix(base, ext)
	}
	if len(base) < 12 {
		return 0, false
	}
	tm, err := time.Parse("200601021504", base[:12])
	if err != nil {
		return 0, false
	}
	return tm.UTC().Unix(), true
}

func checkpointKey(clusterID string) string {
	return fmt.Sprintf("%s/checkpoints-logstats/%s/checkpoint-logstats.yaml", config.S3Prefix(), clusterID)
}

func getSubClusterType(ctx context.Context, store storage.Storage, clusterID string) (types.SubClusterType, error) {
	bucket := config.BucketName()
	dir := fmt.Sprintf("0/shared_pool/%s/logs/", clusterID)
	prefixes, err := store.ObjectsDelimiter(ctx, bucket, dir)
	if err != nil {
		return "", err
	}
	if len(prefixes) > 0 {
		return types.SubClusterTypePhysical, nil
	}

	dir = fmt.Sprintf("0/next-gen/%s/logs/", clusterID)
	prefixes, err = store.ObjectsDelimiter(ctx, bucket, dir)
	if err != nil {
		return "", err
	}
	if len(prefixes) > 0 {
		return types.SubClusterTypeLogical, nil
	}
	return "", nil
}

func getLogPrefix(info types.ClusterInfo) string {
	return getClusterDataPrefix(info, "logs")
}

func getProfilesPrefix(info types.ClusterInfo) string {
	return getClusterDataPrefix(info, "profiles")
}

func getRegionsPrefix(info types.ClusterInfo) string {
	return getClusterDataPrefix(info, "regions")
}

func getClusterDataPrefix(info types.ClusterInfo, dataType string) string {
	switch info.ClusterType {
	case types.ClusterTypeDedicated:
		return fmt.Sprintf("%d/%s/%s/%s/%s/", info.Type, info.TenantID, info.ProjectID, info.ClusterID, dataType)
	case types.ClusterTypeNextgen:
		switch info.SubClusterType {
		case types.SubClusterTypePhysical:
			return fmt.Sprintf("0/shared_pool/%s/%s/", info.ClusterID, dataType)
		case types.SubClusterTypeLogical:
			return fmt.Sprintf("0/next-gen/%s/%s/", info.ClusterID, dataType)
		default:
			panic("Shouldn't be here")
		}
	default:
		return fmt.Sprintf("%d/%s/%s/%s/%s/", info.Type, info.TenantID, info.ProjectID, info.ClusterID, dataType)
	}
}