package checker

import (
	"context"
	"crypto/tls"
	"database/sql"
	"fmt"
	"log/slog"
	"strconv"
	"strings"
	"time"

	"github.com/go-sql-driver/mysql"
	"github.com/tidbcloud/observability/test/ft/common/objstore"
	"k8s.io/apimachinery/pkg/util/wait"
)

const MAX_UPLOAD_LATENCY_SECS = 300

type DataType string

const (
	DataTypeMetrics           DataType = "metrics"
	DataTypeTopsql            DataType = "topsql"
	DataTypeLokiLogs          DataType = "loki_logs"
	DataTypeS3Logs            DataType = "s3_logs"
	DataTypeContinuousProfile DataType = "continuous_profile"
	DataTypeKeyVisualizer     DataType = "key_visualizer"
	DataTypeSlowLogs          DataType = "slow_logs"
)

func CheckAll(ctx context.Context, cluster ClusterInfo, tr TimeRange, enableSlowLog bool, whenNotFound func(DataType) error) (bool, error) {
	if !cluster.Valid() {
		return false, fmt.Errorf("invalid cluster info: %v", cluster)
	}
	if !tr.Valid() {
		return false, fmt.Errorf("invalid time range: %v", tr)
	}
	if whenNotFound == nil {
		whenNotFound = func(DataType) error { return nil }
	}
	//check metrics and loki
	slog.Info("check metrics", "startTs", tr.StartTs, "endTs", tr.EndTs)
	ok, err := QueryMetricsCase(cluster, tr.StartTs, tr.EndTs)
	if err != nil {
		return false, err
	}
	if !ok {
		return false, whenNotFound(DataTypeMetrics)
	}

	slog.Info("check topsql", "startTs", tr.StartTs, "endTs", tr.EndTs)
	ok, err = QueryTopsqlCase(cluster, tr.StartTs, tr.EndTs)
	if err != nil {
		return false, err
	}
	if !ok {
		return false, whenNotFound(DataTypeTopsql)
	}

	slog.Info("check loki logs", "startTs", tr.StartTs, "endTs", tr.EndTs)
	ok, err = QueryLogsCase(cluster, tr.StartTs, tr.EndTs)
	if err != nil {
		return false, err
	}
	if !ok {
		return false, whenNotFound(DataTypeLokiLogs)
	}
	// check logs, continuous profiles, key visualizer files
	store, err := NewStorageFromCluster(ctx, cluster)
	if err != nil {
		return false, err
	}
	slog.Info("check s3 logs", "startTs", tr.StartTs, "endTs", tr.EndTs)
	ok, err = HasS3Logs(ctx, store, cluster, tr)
	if err != nil {
		return false, err
	}
	if !ok {
		return false, whenNotFound(DataTypeS3Logs)
	}
	slog.Info("check continuous profiles", "startTs", tr.StartTs, "endTs", tr.EndTs)
	ok, err = HasContinuousProfiles(ctx, store, cluster, tr)
	if err != nil {
		return false, err
	}
	if !ok {
		return false, whenNotFound(DataTypeContinuousProfile)
	}
	slog.Info("check key visualizer files", "startTs", tr.StartTs, "endTs", tr.EndTs)
	ok, err = HasKeyVisualizerFiles(ctx, store, cluster, tr)
	if err != nil {
		return false, err
	}
	if !ok {
		return false, whenNotFound(DataTypeKeyVisualizer)
	}
	if enableSlowLog {
		slog.Info("check slow logs", "startTs", tr.StartTs, "endTs", tr.EndTs)
		if err := triggerSlowLog(cluster.TiDBConnString); err != nil {
			return false, err
		}
		err := wait.PollUntilContextTimeout(ctx, 15*time.Second, 3*time.Minute, true, func(ctx context.Context) (bool, error) {
			ok, err = HasSlowLogs(ctx, store, cluster, tr)
			if err != nil || !ok {
				slog.Info("error checking slow logs", "error", err, "ok", ok)
				return false, nil
			}
			return ok, nil
		})
		if err != nil {
			return false, fmt.Errorf("error checking slow logs: %v", err)
		}
	}
	return true, nil
}

func NewStorageFromCluster(ctx context.Context, cluster ClusterInfo) (objstore.Storage, error) {
	switch strings.ToLower(cluster.Provider) {
	case "aws":
		return objstore.NewAwsStorage(ctx, cluster.Region, "")
	case "gcp":
		return objstore.NewGcpStorage(ctx)
	case "azure":
		return objstore.NewAzureStorage(ctx, cluster.AzureStorageAccountName())
	default:
		return nil, fmt.Errorf("unsupported provider: %s", cluster.Provider)
	}
}

func HasS3Logs(ctx context.Context, store objstore.Storage, cluster ClusterInfo, tr TimeRange) (bool, error) {
	bucket := cluster.BucketName()
	keyPath := cluster.KeyPrefixS3Logs()
	for chunkTs := tr.StartTs - tr.StartTs%1000; chunkTs < tr.EndTs; chunkTs += 1000 {
		prefix := fmt.Sprintf("%s%d", keyPath, chunkTs/1000)
		objs, err := store.Objects(ctx, bucket, prefix)
		if err != nil {
			return false, err
		}
		for _, obj := range objs {
			filename := strings.TrimPrefix(obj.Key, keyPath)
			var ts int
			if n, err := fmt.Sscanf(filename, "%d-", &ts); n != 1 || err != nil {
				continue // invalid filename
			}
			if ts < tr.StartTs {
				continue // old file
			}
			if ts >= tr.EndTs+MAX_UPLOAD_LATENCY_SECS { // Log files have batch duration.
				continue // too new file
			}
			slog.Info("s3 logs found", "filename", filename)
			return true, nil
		}
	}
	return false, nil
}

func HasContinuousProfiles(ctx context.Context, store objstore.Storage, cluster ClusterInfo, tr TimeRange) (bool, error) {
	bucket := cluster.BucketName()
	keyPath := cluster.KeyPrefixContinuousProfiles()
	for chunkTs := tr.StartTs - tr.StartTs%1000; chunkTs < tr.EndTs; chunkTs += 1000 {
		prefix := fmt.Sprintf("%s%d", keyPath, chunkTs/1000)
		objs, err := store.Objects(ctx, bucket, prefix)
		if err != nil {
			return false, err
		}
		for _, obj := range objs {
			filename := strings.TrimPrefix(obj.Key, keyPath)
			var ts int
			if n, err := fmt.Sscanf(filename, "%d-", &ts); n != 1 || err != nil {
				continue // invalid filename
			}
			if ts < tr.StartTs {
				continue // old file
			}
			if ts > tr.EndTs {
				continue // new file
			}
			slog.Info("continuous profiles found", "filename", filename)
			return true, nil
		}
	}
	return false, nil
}

func HasKeyVisualizerFiles(ctx context.Context, store objstore.Storage, cluster ClusterInfo, tr TimeRange) (bool, error) {
	bucket := cluster.BucketName()
	keyPath := cluster.KeyPrefixKeyVisualizerFiles()
	prefixes, err := tr.FormatedHourlyPrefixes("2006010215")
	if err != nil {
		return false, err
	}
	for _, prefix := range prefixes {
		keyPrefix := keyPath + prefix
		objs, err := store.Objects(ctx, bucket, keyPrefix)
		if err != nil {
			return false, err
		}
		for _, obj := range objs {
			filename := strings.TrimPrefix(obj.Key, keyPath)
			// aws: 202410290904.json.slog.gz
			// gcp: 202410290904.json.log
			filename = strings.TrimSuffix(filename, ".gz")
			filename = strings.TrimSuffix(filename, ".json.log")
			minutelyTm, err := time.Parse("200601021504", filename)
			if err != nil {
				continue // invalid filename
			}
			ts := int(minutelyTm.Unix())
			if ts < tr.StartTs {
				continue // old file
			}
			if ts > tr.EndTs {
				continue // new file
			}
			slog.Info("key visualizer files found", "filename", filename)
			return true, nil
		}
	}
	return false, nil
}

func triggerSlowLog(connString string) error {
	err := mysql.RegisterTLSConfig("tidb", &tls.Config{
		//nolint:gosec
		InsecureSkipVerify: true,
	})
	if err != nil {
		return err
	}

	db, err := sql.Open("mysql", connString)
	if err != nil {
		return err
	}
	defer db.Close()
	_, err = db.Exec("SELECT sleep(3)")
	return err
}

func HasSlowLogs(ctx context.Context, store objstore.Storage, cluster ClusterInfo, tr TimeRange) (bool, error) {
	bucket := cluster.BucketName()
	keyPath := cluster.KeyPrefixSlowLogs()
	objs, err := store.Objects(ctx, bucket, keyPath)
	if err != nil {
		return false, err
	}
	for _, obj := range objs {
		filename := strings.TrimPrefix(obj.Key, keyPath)
		// filename: 1736411230-12574584-6d1c-4a86-9af2-f17e9e16821d.slog.gz
		ts := filename[:strings.Index(filename, "-")]
		tsInt, err := strconv.Atoi(ts)
		if err != nil {
			continue // invalid filename
		}
		if tsInt < tr.StartTs {
			continue // old file
		}
		if tsInt > tr.EndTs {
			continue // new file
		}
		slog.Info("slow logs found", "filename", filename)
		return true, nil
	}
	return false, nil
}

type ClusterInfo struct {
	Provider       string `yaml:"provider"`
	Env            string `yaml:"env"`
	Region         string `yaml:"region"`
	TenantID       string `yaml:"tenant"`
	ProjectID      string `yaml:"project"`
	ClusterID      string `yaml:"cluster"`
	O11yID         string `yaml:"o11y_id"`
	TiDBConnString string `yaml:"tidbConnString"`
	StorageAccount string `yaml:"storageAccount"`
}

func (c ClusterInfo) Valid() bool {
	return c.Provider != "" &&
		c.Env != "" &&
		c.Region != "" &&
		c.TenantID != "" &&
		c.ProjectID != "" &&
		c.ClusterID != ""
}

func (c ClusterInfo) BucketName() string {
	switch strings.ToLower(c.Env) {
	case "prod":
		return fmt.Sprintf("o11y-prod-shared-%s", c.Region)
	default:
		return fmt.Sprintf("o11y-test-shared-%s", c.Region)
	}
}

func (c ClusterInfo) KeyPrefixS3Logs() string {
	return fmt.Sprintf("0/%v/%v/%v/logs/tidb/", c.TenantID, c.ProjectID, c.ClusterID)
}

func (c ClusterInfo) KeyPrefixContinuousProfiles() string {
	return fmt.Sprintf("0/%v/%v/%v/profiles/", c.TenantID, c.ProjectID, c.ClusterID)
}

func (c ClusterInfo) KeyPrefixKeyVisualizerFiles() string {
	return fmt.Sprintf("0/%v/%v/%v/regions/", c.TenantID, c.ProjectID, c.ClusterID)
}

func (c ClusterInfo) KeyPrefixSlowLogs() string {
	return fmt.Sprintf("0/%v/%v/%v/logs/slowlog/", c.TenantID, c.ProjectID, c.ClusterID)
}

func (c ClusterInfo) AzureStorageAccountName() string {
	if c.StorageAccount != "" {
		return c.StorageAccount
	}
	return fmt.Sprintf("o11yprod%s", c.Region)
}

type TimeRange struct {
	StartTs int `yaml:"start"` // if 0 means now-1h
	EndTs   int `yaml:"end"`   // if 0 means now
}

func (tr *TimeRange) Valid() bool {
	if tr.StartTs > 0 && tr.EndTs > 0 {
		return true
	}
	if tr.StartTs == 0 && tr.EndTs == 0 {
		tr.EndTs = int(time.Now().Unix())
		tr.StartTs = tr.EndTs - 3600
		return true
	}
	return false
}

func (tr TimeRange) FormatedHourlyPrefixes(format string) ([]string, error) {
	start := time.Unix(int64(tr.StartTs), 0)
	end := time.Unix(int64(tr.EndTs), 0)
	var prefixes []string
	for t := start; !t.After(end); t = t.Add(time.Hour) {
		prefix := t.UTC().Format(format)
		prefixes = append(prefixes, prefix)
	}
	return prefixes, nil
}