package process

import (
	"context"
	"path"
	"strings"

	"github.com/pingcap/log"
	"github.com/tidbcloud/observability/cmd/diagnosis-compute/pkg/config"
	"github.com/tidbcloud/observability/cmd/diagnosis-compute/pkg/metrics"
	"github.com/tidbcloud/observability/cmd/diagnosis-compute/pkg/types"
	"github.com/tidbcloud/observability/o11ydiagnosis/pkg/storage"
	"go.uber.org/zap"
)

// runConprofRetentionCleanup applies tiered conprof retention for the hour aligned
// with the current log-stats checkpoint:
//   - list the hour from N days ago and delete profiles where (ts/60) % 2 != 0
//   - list the hour from M days ago and delete profiles where (ts/60) % 4 != 0
//
// N/M are configured via -conprof-first-halving-days / -conprof-second-halving-days (default 7/15).
// Data older than 32 days is removed by S3 lifecycle rules managed by task collector.
func runConprofRetentionCleanup(
	ctx context.Context,
	store storage.Storage,
	bucket string,
	info types.ClusterInfo,
	hourBegin int64,
) {
	firstHourBegin := hourBegin - config.ConprofFirstHalvingAgeSeconds()
	if firstHourBegin > 0 {
		cleanupConprofHour(ctx, store, bucket, info, firstHourBegin, 2, "first_halving")
	}

	secondHourBegin := hourBegin - config.ConprofSecondHalvingAgeSeconds()
	if secondHourBegin > 0 {
		cleanupConprofHour(ctx, store, bucket, info, secondHourBegin, 4, "second_halving")
	}
}

func cleanupConprofHour(
	ctx context.Context,
	store storage.Storage,
	bucket string,
	info types.ClusterInfo,
	hourBegin int64,
	modulus int64,
	phase string,
) {
	l := log.L().With(
		zap.String("cluster", info.StorageID),
		zap.String("phase", phase),
		zap.Int64("hour_begin", hourBegin),
	)
	hourEnd := hourBegin + hourSeconds - 1
	profilesPrefix := strings.TrimSuffix(getProfilesPrefix(info), "/")
	metas := listUnixTsPrefixedObjects(ctx, store, bucket, profilesPrefix, hourBegin, hourEnd)

	keysToDelete := make([]string, 0, len(metas)/2)
	var deletedBytes int64
	for _, meta := range metas {
		filename := path.Base(meta.Key)
		ts, ok := unixTsFromFilename(filename)
		if !ok {
			continue
		}
		if !shouldDeleteConprofProfile(ts, modulus) {
			continue
		}
		keysToDelete = append(keysToDelete, meta.Key)
		deletedBytes += meta.Size
	}

	if len(keysToDelete) == 0 {
		l.Info("conprof retention cleanup completed",
			zap.Int("listed", len(metas)),
			zap.Int("deleted", 0),
			zap.Int64("deleted_bytes", 0),
		)
		return
	}

	if err := deleteStorageObjects(ctx, store, bucket, keysToDelete); err != nil {
		l.Error("failed to batch delete conprof profiles",
			zap.Int("delete_count", len(keysToDelete)),
			zap.Error(err),
		)
		metrics.RecordConprofCleanupError(info, phase)
		return
	}

	metrics.RecordConprofCleanupDeleted(info, phase, len(keysToDelete), deletedBytes)
	l.Info("conprof retention cleanup completed",
		zap.Int("listed", len(metas)),
		zap.Int("deleted", len(keysToDelete)),
		zap.Int64("deleted_bytes", deletedBytes),
	)
}

func shouldDeleteConprofProfile(ts, modulus int64) bool {
	return (ts/60)%modulus != 0
}

func deleteStorageObjects(ctx context.Context, store storage.Storage, bucket string, keys []string) error {
	if len(keys) == 0 {
		return nil
	}
	if err := config.S3sem.Acquire(ctx, 1); err != nil {
		return err
	}
	defer config.S3sem.Release(1)
	return store.DeleteObjects(ctx, bucket, keys)
}