package tasks

import (
	"context"
	"fmt"
	"os"
	"strconv"
	"strings"

	v1 "github.com/tidbcloud/observability/api/regionalserver/resourceworker/v1"
	"github.com/tidbcloud/observability/o11ydiagnosis/pkg/utils"
	"github.com/tidbcloud/observability/pkg/common/message"
	"github.com/tidbcloud/observability/pkg/resourceworker/providers"
)

const (
	DefaultZeroPrefixExpirationDays    = int32(180)
	DefaultDeltalakeExpirationDays     = int32(180)
	DefaultAthenaResultsExpirationDays = int32(1)
	DefaultProfileExpirationDays       = int32(32)
	DefaultOOMRecordExpirationDays     = int32(32)
	DefaultKeyvizRecordExpirationDays  = int32(32)
	DefaultHistoryTaskExpirationDays   = int32(32)
	deltalakePrefix                    = "deltalake/"
	athenaResultsPrefix                = "athena-results/logs-query/"
	deltalakeExpirationRuleName        = "deltalake_expiration"
	athenaResultsExpirationRuleName    = "athena_results_expiration"
	zeroPrefixFallbackRuleName         = "zero_prefix_fallback"
	diagnosisDataFallbackRuleName      = "diagnosis_data_fallback"
	onePrefixFallbackRuleName          = "one_prefix_fallback"
	metricsGlacierRuleName             = "metrics_glacier"
	historyTaskExpirationRuleName      = "diagnosis_history_task_expiration"
)

type LifecycleExpirationConfig struct {
	ZeroPrefixExpirationDays    int32
	DeltalakeExpirationDays     int32
	AthenaResultsExpirationDays int32
	ProfileExpirationDays       int32
	OOMRecordExpirationDays     int32
	KeyvizRecordExpirationDays  int32
}

func NewLifecycleConfiguration(clientGroup utils.ClientGroup, s3Client providers.S3Client, region string, env string, vendor string, expiration LifecycleExpirationConfig) {
	bucket := fmt.Sprintf("o11y-%s-shared-%s", env, region)
	if os.Getenv("BUCKET_NAME") != "" {
		bucket = os.Getenv("BUCKET_NAME")
	}
	rules := buildLifecycleRules(clientGroup, s3Client, bucket, vendor, expiration)

	fmt.Printf("add %d lifecycles\n", len(rules))
	err := s3Client.AddLifecycle(context.TODO(), bucket, rules, buildUnusedLifecycleRulePrefixes(vendor))
	if err != nil {
		panic(err)
	}
}

func ResolveZeroPrefixExpirationDays(flagValue int32) int32 {
	return resolveExpirationDays("ZERO_PREFIX_EXPIRATION_DAYS", flagValue, DefaultZeroPrefixExpirationDays)
}

func ResolveProfileExpirationDays(flagValue int32) int32 {
	return resolveExpirationDays("PROFILE_EXPIRATION_DAYS", flagValue, DefaultProfileExpirationDays)
}

func ResolveOOMRecordExpirationDays(flagValue int32) int32 {
	return resolveExpirationDays("OOM_RECORD_EXPIRATION_DAYS", flagValue, DefaultOOMRecordExpirationDays)
}

func ResolveKeyvizRecordExpirationDays(flagValue int32) int32 {
	return resolveExpirationDays("KEYVIZ_RECORD_EXPIRATION_DAYS", flagValue, DefaultKeyvizRecordExpirationDays)
}

func ResolveDeltalakeExpirationDays(flagValue int32) int32 {
	return resolveExpirationDays("DELTALAKE_EXPIRATION_DAYS", flagValue, DefaultDeltalakeExpirationDays)
}

func ResolveAthenaResultsExpirationDays(flagValue int32) int32 {
	return resolveExpirationDays("ATHENA_RESULTS_EXPIRATION_DAYS", flagValue, DefaultAthenaResultsExpirationDays)
}

func resolveExpirationDays(envKey string, flagValue int32, defaultValue int32) int32 {
	if envValue := os.Getenv(envKey); envValue != "" {
		parsed, err := strconv.ParseInt(envValue, 10, 32)
		if err == nil && parsed >= 0 {
			return int32(parsed)
		}
	}
	if flagValue >= 0 {
		return flagValue
	}
	return defaultValue
}

func buildLifecycleRules(clientGroup utils.ClientGroup, s3Client providers.S3Client, bucket string, vendor string, expiration LifecycleExpirationConfig) []message.LifecycleRuleParam {
	rules := []message.LifecycleRuleParam{
		BuildDiagnosisTaskRule(),
		BuildHistoryTaskRule(),
	}
	if shouldEnableMetricsGlacierRule(vendor) {
		rules = append(rules, BuildMetricsGlacierRule(bucket))
	}
	historyTasks, err := loadHistoryDiagnosisTasks(context.TODO(), s3Client, bucket)
	if err != nil {
		panic(err)
	}
	fmt.Printf("loaded %d history diagnosis tasks for lifecycle rules\n", len(historyTasks))
	rules = append(rules, BuildProfileAndOomRecordRule(
		clientGroup,
		historyTasks,
		expiration.ProfileExpirationDays,
		expiration.OOMRecordExpirationDays,
		expiration.KeyvizRecordExpirationDays,
	)...)
	rules = dedupeLifecycleRuleParams(rules)
	if expiration.DeltalakeExpirationDays > 0 {
		rules = append(rules, BuildDeltalakeRule(expiration.DeltalakeExpirationDays))
	}
	if vendor == "aws" && expiration.AthenaResultsExpirationDays > 0 {
		rules = append(rules, BuildAthenaResultsRule(expiration.AthenaResultsExpirationDays))
	}
	if expiration.ZeroPrefixExpirationDays > 0 {
		rules = append(rules, BuildPrefixExpirationFallbackRules(expiration.ZeroPrefixExpirationDays)...)
	}
	return rules
}

func buildUnusedLifecycleRulePrefixes(vendor string) []string {
	prefixes := []string{
		"profile_deletion_",
		"oom_record_deletion_",
		"keyviz_record_deletion_",
	}
	if !shouldEnableMetricsGlacierRule(vendor) {
		prefixes = append(prefixes, metricsGlacierRuleName)
	}
	return prefixes
}

func shouldEnableMetricsGlacierRule(vendor string) bool {
	return vendor != "alicloud"
}

func BuildDiagnosisTaskRule() message.LifecycleRuleParam {
	return message.LifecycleRuleParam{
		RuleName:       "diagnosis_task_expiration",
		Prefix:         diagnosisTaskPrefix,
		ColdDays:       0,
		ExpirationDays: 2,
	}
}

func BuildHistoryTaskRule() message.LifecycleRuleParam {
	return message.LifecycleRuleParam{
		RuleName:       historyTaskExpirationRuleName,
		Prefix:         diagnosisHistoryTaskPrefix,
		ColdDays:       0,
		ExpirationDays: DefaultHistoryTaskExpirationDays,
	}
}

func BuildMetricsGlacierRule(bucket string) message.LifecycleRuleParam {
	return message.LifecycleRuleParam{
		RuleName:       metricsGlacierRuleName,
		Prefix:         fmt.Sprintf("%s/", bucket),
		ColdDays:       32,
		ExpirationDays: 0,
	}
}

func BuildZeroPrefixFallbackRule(expirationDays int32) message.LifecycleRuleParam {
	return buildPrefixExpirationRule(zeroPrefixFallbackRuleName, "0/", expirationDays)
}

func BuildDiagnosisDataFallbackRule(expirationDays int32) message.LifecycleRuleParam {
	return buildPrefixExpirationRule(diagnosisDataFallbackRuleName, "diagnosis/data/", expirationDays)
}

func BuildOnePrefixFallbackRule(expirationDays int32) message.LifecycleRuleParam {
	return buildPrefixExpirationRule(onePrefixFallbackRuleName, "1/", expirationDays)
}

func BuildDeltalakeRule(expirationDays int32) message.LifecycleRuleParam {
	return buildPrefixExpirationRule(deltalakeExpirationRuleName, deltalakePrefix, expirationDays)
}

func BuildAthenaResultsRule(expirationDays int32) message.LifecycleRuleParam {
	return buildPrefixExpirationRule(athenaResultsExpirationRuleName, athenaResultsPrefix, expirationDays)
}

func BuildPrefixExpirationFallbackRules(expirationDays int32) []message.LifecycleRuleParam {
	return []message.LifecycleRuleParam{
		BuildZeroPrefixFallbackRule(expirationDays),
		BuildDiagnosisDataFallbackRule(expirationDays),
		BuildOnePrefixFallbackRule(expirationDays),
	}
}

func buildPrefixExpirationRule(ruleName, prefix string, expirationDays int32) message.LifecycleRuleParam {
	return message.LifecycleRuleParam{
		RuleName:       ruleName,
		Prefix:         prefix,
		ColdDays:       0,
		ExpirationDays: expirationDays,
	}
}

func BuildProfileAndOomRecordRule(clientGroup utils.ClientGroup, historyTasks []O11yDiagnosisTask, profileExpirationDays, oomRecordExpirationDays, keyvizRecordExpirationDays int32) []message.LifecycleRuleParam {
	rules := []message.LifecycleRuleParam{}
	seenRuleNames := map[string]struct{}{}
	seenPrefixes := map[string]struct{}{}
	appendUniqueRule := func(rule message.LifecycleRuleParam) {
		if _, exists := seenRuleNames[rule.RuleName]; exists {
			return
		}
		if _, exists := seenPrefixes[rule.Prefix]; exists {
			return
		}
		seenRuleNames[rule.RuleName] = struct{}{}
		seenPrefixes[rule.Prefix] = struct{}{}
		rules = append(rules, rule)
	}

	clusterList := new(v1.ClusterList)
	err := clientGroup.Client.List(context.TODO(), clusterList)
	if err != nil {
		panic(err)
	}
	fmt.Printf("build profile and oom record rule for %d clusters\n", len(clusterList.Items))
	for _, cluster := range clusterList.Items {
		identity, ok := lifecycleIdentityFromCluster(cluster)
		if !ok {
			continue
		}
		appendClusterDataRetentionRules(appendUniqueRule, identity, profileExpirationDays, oomRecordExpirationDays, keyvizRecordExpirationDays)
	}

	fmt.Printf("build profile and keyviz record rule for %d history tasks\n", len(historyTasks))
	for _, task := range historyTasks {
		identity, ok := lifecycleIdentityFromHistoryTask(task)
		if !ok {
			continue
		}
		appendClusterDataRetentionRules(appendUniqueRule, identity, profileExpirationDays, oomRecordExpirationDays, keyvizRecordExpirationDays)
	}

	return rules
}

type lifecycleClusterIdentity struct {
	AppID     string
	TenantID  string
	ProjectID string
	NextGen   bool
}

func lifecycleIdentityFromCluster(cluster v1.Cluster) (lifecycleClusterIdentity, bool) {
	bizType := cluster.Labels["observability.pingcap.com/bizType"]
	if !strings.Contains(bizType, "tidbcloud") {
		return lifecycleClusterIdentity{}, false
	}
	appID := cluster.Labels["observability.pingcap.com/applicationID"]
	if skipLifecycleAppID(appID) {
		return lifecycleClusterIdentity{}, false
	}
	return lifecycleClusterIdentity{
		AppID:     appID,
		TenantID:  cluster.Labels["observability.pingcap.com/tenantID"],
		ProjectID: cluster.Labels["observability.pingcap.com/projectID"],
		NextGen:   strings.HasSuffix(bizType, "tidbcloud-nextgen"),
	}, true
}

func lifecycleIdentityFromHistoryTask(task O11yDiagnosisTask) (lifecycleClusterIdentity, bool) {
	if skipLifecycleAppID(task.Meta.ClusterID) {
		return lifecycleClusterIdentity{}, false
	}
	return lifecycleClusterIdentity{
		AppID:     task.Meta.ClusterID,
		TenantID:  task.Meta.TenantID,
		ProjectID: task.Meta.ProjectID,
		NextGen:   task.Meta.ClusterType == "nextgen",
	}, true
}

func skipLifecycleAppID(appID string) bool {
	if appID == "dataapp" || appID == "k8s-infra" || appID == "seed" || appID == "base" {
		return true
	}
	return strings.HasPrefix(appID, "dp")
}

func appendClusterDataRetentionRules(
	appendUniqueRule func(message.LifecycleRuleParam),
	identity lifecycleClusterIdentity,
	profileExpirationDays, oomRecordExpirationDays, keyvizRecordExpirationDays int32,
) {
	var profileRulePrefix string
	if identity.NextGen {
		profileRulePrefix = fmt.Sprintf("0/shared_pool/%s/profiles/", identity.AppID)
	} else {
		profileRulePrefix = fmt.Sprintf("0/%s/%s/%s/profiles/", identity.TenantID, identity.ProjectID, identity.AppID)
	}
	if profileExpirationDays > 0 {
		appendUniqueRule(message.LifecycleRuleParam{
			RuleName:       fmt.Sprintf("profile_deletion_%s", identity.AppID),
			Prefix:         profileRulePrefix,
			ColdDays:       0,
			ExpirationDays: profileExpirationDays,
		})
	}

	if identity.NextGen {
		return
	}

	if oomRecordExpirationDays > 0 {
		appendUniqueRule(message.LifecycleRuleParam{
			RuleName:       fmt.Sprintf("oom_record_deletion_%s", identity.AppID),
			Prefix:         fmt.Sprintf("0/%s/%s/%s/oom_record/", identity.TenantID, identity.ProjectID, identity.AppID),
			ColdDays:       0,
			ExpirationDays: oomRecordExpirationDays,
		})
	}

	if keyvizRecordExpirationDays > 0 {
		appendUniqueRule(message.LifecycleRuleParam{
			RuleName:       fmt.Sprintf("keyviz_record_deletion_%s", identity.AppID),
			Prefix:         fmt.Sprintf("0/%s/%s/%s/regions/", identity.TenantID, identity.ProjectID, identity.AppID),
			ColdDays:       0,
			ExpirationDays: keyvizRecordExpirationDays,
		})
	}
}

func lifecyclePrefixesConflict(a, b string) bool {
	if a == b {
		return true
	}
	return strings.HasPrefix(a, b) || strings.HasPrefix(b, a)
}

func dedupeLifecycleRuleParams(rules []message.LifecycleRuleParam) []message.LifecycleRuleParam {
	out := make([]message.LifecycleRuleParam, 0, len(rules))
	seenRuleNames := map[string]struct{}{}
	seenPrefixes := make([]string, 0, len(rules))
	for _, rule := range rules {
		if _, exists := seenRuleNames[rule.RuleName]; exists {
			continue
		}
		conflict := false
		for _, prefix := range seenPrefixes {
			if lifecyclePrefixesConflict(rule.Prefix, prefix) {
				conflict = true
				break
			}
		}
		if conflict {
			continue
		}
		seenRuleNames[rule.RuleName] = struct{}{}
		seenPrefixes = append(seenPrefixes, rule.Prefix)
		out = append(out, rule)
	}
	return out
}