package load

import (
	"context"
	"fmt"
	"math"
	"path"
	"strconv"
	"strings"
	"sync"

	"github.com/pingcap-inc/clinic/pkg/ngm/storage"

	lru "github.com/hashicorp/golang-lru/v2"
	"github.com/pingcap/log"
	"go.uber.org/zap"
)

type LoadParam struct {
	Env           string
	Provider      string
	Region        string
	OrgID         uint64
	ProjectID     uint64
	ClusterID     string
	SharedTier    bool
	StorageBucket string
	StoragePrefix string
	StartUnixSec  int64
	EndUnixSec    int64
}

const (
	allowTimeRangeDrift = 240 // 4 minutes

	globalDownloadConcurrency     = 200
	perRequestDownloadConcurrency = 40
)

var globalDownloadSem = make(chan struct{}, globalDownloadConcurrency)

type CacheKey struct {
	bucket string
	key    string
}

func Load(
	ctx context.Context,
	param *LoadParam,
	sf storage.StorageFactory,
	cache *lru.Cache[CacheKey, TimeRangeRecords],
) (<-chan TimeRangeRecords, error) {
	store, err := sf.Storage(ctx, param.Provider, param.Region)
	if err != nil {
		return nil, err
	}

	objs, err := listObjects(ctx, param, store)
	if err != nil {
		return nil, err
	}

	ch := make(chan TimeRangeRecords, len(objs))
	go func() {
		var wg sync.WaitGroup
		defer func() {
			wg.Wait()
			close(ch)
		}()

		localSem := make(chan struct{}, perRequestDownloadConcurrency)
		for _, obj := range objs {
			bkt := bucket(param)
			key := obj.Key
			cacheK := CacheKey{bkt, key}

			if records, ok := cache.Get(cacheK); ok {
				ch <- records.Clone()
				continue
			}

			exit := acquireSem(ctx, localSem)
			if exit {
				return
			}

			wg.Add(1)
			go func() {
				defer func() {
					releaseSem(localSem)
					wg.Done()
				}()

				if records, ok := cache.Get(cacheK); ok {
					ch <- records.Clone()
					return
				}

				obj, err := store.Object(ctx, bkt, key)
				if err != nil {
					log.Warn("download statement object failed", zap.String("key", key), zap.Error(err))
					return
				}
				defer func() {
					_ = obj.Reader.Close()
				}()

				records := Parse(obj.Reader)
				cache.Add(cacheK, records)
				ch <- records.Clone()
			}()
		}
	}()

	return ch, nil
}

func listObjects(ctx context.Context, param *LoadParam, store storage.Storage) ([]storage.ObjectMeta, error) {
	// if !param.SharedTier {
	// 	// dedicated tier not support for now
	// 	return nil, nil
	// }

	bkt := bucket(param)
	prefix := keyPrefix(param) + rangeTsPrefix(int(param.StartUnixSec), int(param.EndUnixSec))
	objs, err := store.Objects(ctx, bkt, prefix)
	if err != nil {
		return nil, err
	}

	tr := driftableTimeRange(param)
	metas := filterMetasByTimeRange(objs, tr)
	return metas, nil
}

func bucket(param *LoadParam) string {
	if param.StorageBucket != "" {
		return param.StorageBucket
	}
	if param.SharedTier {
		if (param.Env == "dev" || param.Env == "staging") && param.Provider == "aws" && param.Region == "us-east-1" {
			return fmt.Sprintf("o11y-prod-shared-%s-serverless", param.Region)
		}
		if (param.Env == "dev") && param.Provider == "alicloud" {
			return fmt.Sprintf("o11y-dev-shared-%s", param.Region)
		}
		return fmt.Sprintf("o11y-prod-shared-%s", param.Region)
	}
	switch param.Env {
	case "prod":
		return fmt.Sprintf("o11y-prod-shared-%s", param.Region)
	default:
		if param.Provider == "aws" && param.Region == "us-west-2" {
			return fmt.Sprintf("o11y-prod-shared-%s-staging", param.Region)
		} else {
			return fmt.Sprintf("o11y-test-shared-%s", param.Region)
		}
	}
}

func keyPrefix(param *LoadParam) string {
	if param.StoragePrefix != "" {
		return ensureTrailingSlash(param.StoragePrefix)
	}
	var r string
	if param.SharedTier {
		projectID := param.ProjectID
		// todo: remove this special case when o11y api is fixed
		if param.OrgID == 1372813089209292046 && param.ProjectID == 1372813089454610005 && param.ClusterID == "10661816924665497020" {
			projectID = 1372813089454738610
			log.Info("project changed for this shared tier",
				zap.Uint64("project_id", param.ProjectID),
				zap.String("original_project_id", "1372813089454738610"),
				zap.String("cluster", param.ClusterID))
		}
		r = fmt.Sprintf("0/%s-devtier/serverless/%d/%d/%s/logs/statement/", param.Env, param.OrgID, projectID, param.ClusterID)
	} else {
		r = fmt.Sprintf("0/%v/%v/%v/logs/statementlog/", param.OrgID, param.ProjectID, param.ClusterID)
	}
	switch param.Provider {
	case "alicloud", "aliyun":
		r = fmt.Sprintf("%s/%s", bucket(param), r)
	}
	return r
}

func ensureTrailingSlash(prefix string) string {
	if prefix == "" || strings.HasSuffix(prefix, "/") {
		return prefix
	}
	return prefix + "/"
}

func filterMetasByTimeRange(metas []storage.ObjectMeta, tr TimeRange) []storage.ObjectMeta {
	var filtered []storage.ObjectMeta
	for _, meta := range metas {
		_, name := path.Split(meta.Key)
		if objectNameInTimeRange(name, tr) {
			filtered = append(filtered, meta)
		}
	}
	return filtered
}

func objectNameInTimeRange(name string, tr TimeRange) bool {
	ts := 0
	_, err := fmt.Sscanf(name, "%d-", &ts)
	if err != nil {
		return false
	}

	return inTimeRange(ts, tr)
}

func inTimeRange(timestamp int, tr TimeRange) bool {
	return int64(timestamp) > tr.Begin && int64(timestamp) < tr.End
}

func rangeTsPrefix(beginTime, endTime int) string {
	if beginTime <= 0 || endTime <= 0 {
		return ""
	}
	for n := 2; n <= 9; n++ {
		base := beginTime - beginTime%int(math.Pow10(n))
		if base <= beginTime && (base+int(math.Pow10(n)-1)) >= endTime {
			baseStr := strconv.Itoa(base)
			return baseStr[:len(baseStr)-n]
		}
	}
	return ""
}

func driftableTimeRange(param *LoadParam) TimeRange {
	var begin int64
	if param.StartUnixSec-allowTimeRangeDrift < 0 {
		begin = 0
	} else {
		begin = param.StartUnixSec - allowTimeRangeDrift
	}
	var end int64
	if param.EndUnixSec > math.MaxInt64-allowTimeRangeDrift {
		end = math.MaxInt64
	} else {
		end = param.EndUnixSec + allowTimeRangeDrift
	}
	return TimeRange{
		Begin: begin,
		End:   end,
	}
}

// acquireSem tries to acquireSem the global download semaphore and the local semaphore.
func acquireSem(ctx context.Context, localSem chan struct{}) (exit bool) {
	select {
	case localSem <- struct{}{}:
	case <-ctx.Done():
		return true
	}

	select {
	case globalDownloadSem <- struct{}{}:
	case <-ctx.Done():
		<-localSem
		return true
	}

	return
}

// releaseSem releases the global download semaphore and the local semaphore.
func releaseSem(localSem chan struct{}) {
	<-globalDownloadSem
	<-localSem
}