package nextgentasks

import (
	"context"
	"fmt"
	"time"

	"github.com/tidbcloud/metering/pkg/common/checkpoint"
	"github.com/tidbcloud/metering/pkg/common/config"
	"github.com/tidbcloud/metering/pkg/common/constants"
	"github.com/tidbcloud/metering/pkg/common/metrics"
	"github.com/tidbcloud/metering/pkg/nextgenrepoter"
	"go.uber.org/zap"
)

// NextGenRepoterTask NextGen reporting task
type NextGenRepoterTask struct {
	conf          *config.ServiceConfig
	reporter      nextgenrepoter.NextGenRepoter
	checkpointMgr checkpoint.NextGenCheckpoint
	log           *zap.Logger
}

var newNextGenReporter = func(conf *config.ServiceConfig, log *zap.Logger) (nextgenrepoter.NextGenRepoter, error) {
	return nextgenrepoter.NewNextGenBillingRepoter(conf, log)
}

// NewNextGenRepoterTask creates a new NextGen reporting task
func NewNextGenRepoterTask(conf *config.ServiceConfig, log *zap.Logger) (*NextGenRepoterTask, error) {
	if conf.NextGenMeteringConfig == nil {
		return nil, fmt.Errorf("NextGenMeteringConfig is required")
	}

	if !conf.NextGenMeteringConfig.EnableReporter {
		return nil, fmt.Errorf("NextGen reporter is disabled in configuration")
	}

	// Create NextGen reporter
	reporter, err := newNextGenReporter(conf, log)
	if err != nil {
		return nil, fmt.Errorf("failed to create NextGen billing reporter: %w", err)
	}

	// Create checkpoint manager for loading initial checkpoint metric
	factory := newCheckpointFactory(conf, log)
	checkpointMgr, err := factory.CreateCheckpointManager()
	if err != nil {
		return nil, fmt.Errorf("failed to create checkpoint manager: %w", err)
	}

	// Load checkpoint and set metric at startup to avoid metric gaps
	taskName := "nextgen_reporter"
	initCheckpoint, err := checkpointMgr.GetCheckpoint(context.Background(), taskName)
	if err != nil {
		log.Warn("failed to load initial checkpoint for metric", zap.Error(err))
	} else if initCheckpoint > 0 {
		metrics.MetricsCheckpointTimestamp.WithLabelValues(taskName).Set(float64(initCheckpoint))
		log.Info("loaded initial checkpoint metric", zap.String("task", taskName), zap.Int64("checkpoint", initCheckpoint))
	}

	return &NextGenRepoterTask{
		conf:          conf,
		reporter:      reporter,
		checkpointMgr: checkpointMgr,
		log:           log,
	}, nil
}

// Run executes the NextGen reporting task
func (t *NextGenRepoterTask) Run(ctx context.Context) (err error) {
	// Acquire Delta Lake lock to prevent concurrent access with metering task
	lock := GetGlobalDeltaLakeLock()
	lock.Lock(t.GetTaskName())
	defer lock.Unlock(t.GetTaskName())

	runStart := time.Now()
	defer func() {
		recordNextGenTaskDurationMetric(t.GetTaskName(), t.conf.Vendor, "report", time.Since(runStart))
		if err != nil {
			recordNextGenTaskResultMetric(t.GetTaskName(), t.conf.Vendor, constants.MetricsResultFailed)
		} else {
			recordNextGenTaskResultMetric(t.GetTaskName(), t.conf.Vendor, constants.MetricsResultSuccess)
		}
	}()

	t.log.Info("starting NextGen reporter task")
	err = t.reporter.Report(ctx)
	if err != nil {
		t.log.Error("NextGen reporter task failed", zap.Error(err))
		return err
	}

	// Update checkpoint metric after successful report
	updatedCheckpoint, err := t.checkpointMgr.GetCheckpoint(ctx, t.GetTaskName())
	if err == nil && updatedCheckpoint > 0 {
		metrics.MetricsCheckpointTimestamp.WithLabelValues(t.GetTaskName()).Set(float64(updatedCheckpoint))
	}

	t.log.Info("NextGen reporter task completed successfully")
	return nil
}

// GetTaskName gets the task name
func (t *NextGenRepoterTask) GetTaskName() string {
	return "nextgen_reporter"
}