package datasource

import (
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"io"
	"net/http"
	"sort"
	"strings"
	"sync"
	"time"

	gcs "cloud.google.com/go/storage"
	"github.com/Azure/azure-sdk-for-go/sdk/azcore"
	"github.com/aws/aws-sdk-go-v2/service/s3/types"
	"github.com/pingcap/log"
	"go.uber.org/zap"
	"golang.org/x/sync/semaphore"

	"github.com/tidbcloud/observability/o11ydiagnosis/pkg/keyviz/keyvisual/input"
	"github.com/tidbcloud/observability/o11ydiagnosis/pkg/keyviz/model"
	"github.com/tidbcloud/observability/o11ydiagnosis/pkg/keyviz/utils/stepper"
	"github.com/tidbcloud/observability/o11ydiagnosis/pkg/storage"
)

const (
	minutelyTimeFormat = "200601021504"
)

var s3sem = semaphore.NewWeighted(500)

func FetchDBTablesInfo(ctx context.Context,
	store storage.Storage,
	env, provider, region, tenant, project, cluster string,
) ([]model.DBTablesInfo, error) {
	bucket := bucketName(env, region)
	key := fmt.Sprintf("0/%s/%s/%s/regions/tidb-schema.log.gz", tenant, project, cluster)
	if provider == "gcp" {
		key = strings.TrimSuffix(key, ".gz")
	}
	object, err := store.Object(ctx, bucket, key)
	if err != nil {
		if err == storage.ErrNotFound {
			return nil, nil
		}
		var noSuchKey *types.NoSuchKey
		if errors.As(err, &noSuchKey) {
			return nil, nil
		}
		if err == gcs.ErrObjectNotExist {
			return nil, nil
		}
		var azerr *azcore.ResponseError
		if errors.As(err, &azerr) && azerr.StatusCode == http.StatusNotFound {
			return nil, nil
		}
		return nil, err
	}
	defer object.Reader.Close()
	content, err := io.ReadAll(object.Reader)
	if err != nil {
		return nil, err
	}

	var dbtables []model.DBTablesInfo
	if err := json.Unmarshal(content, &dbtables); err != nil {
		return nil, err
	}
	return dbtables, nil
}

func FetchRegionsInfo(ctx context.Context,
	store storage.Storage,
	env, provider, region, tenant, project, cluster string,
	start, end time.Time,
) ([]TimedRegionsInfo, error) {
	start = start.Truncate(time.Minute)
	end = end.Truncate(time.Minute)
	times := stepper.GenerateTimes(start, end)
	if len(times) == 0 {
		return nil, nil
	}
	return FetchRegionsInfoByTimes(ctx, store, env, provider, region, tenant, project, cluster, times)
}

var (
	fetchMutex sync.Mutex
	memoryMutx sync.Mutex
)

func FetchRegionsInfoByTimes(ctx context.Context,
	store storage.Storage,
	env, provider, region, tenant, project, cluster string,
	times []time.Time,
) ([]TimedRegionsInfo, error) {
	fetchMutex.Lock()
	defer fetchMutex.Unlock()
	bucket := bucketName(env, region)
	data := sync.Map{}
	wg := sync.WaitGroup{}
	wg.Add(len(times))
	memoryLimit := 13 * 1024 * 1024 * 1024 // 13GB would used at most 40GB memory
	usedMemory := 0
	for _, t := range times {
		go func(t time.Time) {
			defer wg.Done()
			utcMinute := t.UTC().Format(minutelyTimeFormat)
			key := fmt.Sprintf("0/%s/%s/%s/regions/%s.json.log.gz", tenant, project, cluster, utcMinute)
			if provider == "gcp" {
				key = strings.TrimSuffix(key, ".gz")
			}
			if err := s3sem.Acquire(ctx, 1); err != nil {
				log.Error("failed to acquire semaphore", zap.Error(err))
				return
			}
			defer s3sem.Release(1)
			object, err := store.Object(ctx, bucket, key)
			if err != nil {
				if err == storage.ErrNotFound {
					return
				}
				var noSuchKey *types.NoSuchKey
				if errors.As(err, &noSuchKey) {
					return
				}
				if err == gcs.ErrObjectNotExist {
					return
				}
				var azerr *azcore.ResponseError
				if errors.As(err, &azerr) && azerr.StatusCode == http.StatusNotFound {
					return
				}
				log.Error("failed to get object", zap.String("key", key), zap.Error(err))
				return
			}
			defer object.Reader.Close()
			b, err := io.ReadAll(object.Reader)
			if err != nil {
				log.Error("failed to read object", zap.String("key", key), zap.Error(err))
				return
			}
			memoryMutx.Lock()
			usedMemory += len(b)
			memoryMutx.Unlock()
			if usedMemory > memoryLimit {
				log.Warn("used memory exceeds limit", zap.Int("used_memory", usedMemory), zap.Int("memory_limit", memoryLimit))
				return
			}
			var info input.RegionsInfo
			if err := json.Unmarshal(b, &info); err != nil {
				log.Error("failed to unmarshal object", zap.String("key", key), zap.Error(err))
				return
			}
			data.Store(t, info)
		}(t)
	}
	wg.Wait()
	var rs []TimedRegionsInfo
	data.Range(func(key, value any) bool {
		t := key.(time.Time)
		info := value.(input.RegionsInfo)
		rs = append(rs, TimedRegionsInfo{
			Time: t,
			Info: info,
		})
		return true
	})
	sort.Sort(TimeRegionsInfoList(rs))
	return rs, nil
}

type TimedRegionsInfo struct {
	Time time.Time
	Info input.RegionsInfo
}

type TimeRegionsInfoList []TimedRegionsInfo

func (s TimeRegionsInfoList) Len() int {
	return len(s)
}

func (s TimeRegionsInfoList) Less(i, j int) bool {
	return s[i].Time.Before(s[j].Time)
}

func (s TimeRegionsInfoList) Swap(i, j int) {
	s[i], s[j] = s[j], s[i]
}

func bucketName(env, region string) string {
	switch env {
	case "prod":
		return fmt.Sprintf("o11y-prod-shared-%s", region)
	default:
		return fmt.Sprintf("o11y-test-shared-%s", region)
	}
}