package nextgenrepoter

import (
	"context"
	"fmt"
	"sort"
	"strings"
	"time"

	"github.com/tidbcloud/metering/pkg/common/config"
	"github.com/tidbcloud/metering/pkg/common/constants"
	"github.com/tidbcloud/metering/pkg/common/models"
	"github.com/tidbcloud/metering/pkg/repoter"
	"go.uber.org/zap"
)

// NextGenBillingRepoter NextGen version metering reporter
type NextGenBillingRepoter struct {
	conf   *config.ServiceConfig
	loader NextGenRepoterLoader
	sender measureSender
	log    *zap.Logger
}

type measureSender interface {
	repoter.MeasureSender
}

var (
	newNextGenRepoterLoaderFunc = NewNextGenRepoterLoader
	measureSenderFactory        = func() measureSender {
		return new(repoter.MeasureAPIImpl)
	}
)

// NewNextGenBillingRepoter creates a new NextGen metering reporter instance
func NewNextGenBillingRepoter(conf *config.ServiceConfig, log *zap.Logger) (*NextGenBillingRepoter, error) {
	if conf.NextGenMeteringConfig == nil {
		return nil, fmt.Errorf("NextGenMeteringConfig is required")
	}

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

	// Create sender (reuse existing MeasureSender)
	var sender measureSender = measureSenderFactory()
	err = sender.InitMeasureHandler(conf.Env, conf.Vendor, false, true)
	if err != nil {
		return nil, fmt.Errorf("failed to init measure sender: %w", err)
	}

	return &NextGenBillingRepoter{
		conf:   conf,
		loader: loader,
		sender: sender,
		log:    log,
	}, nil
}

// SetSender sets the sender (mainly used for testing)
func (r *NextGenBillingRepoter) SetSender(sender measureSender) {
	r.sender = sender
}

// Report executes the reporting task
func (r *NextGenBillingRepoter) Report(ctx context.Context) error {
	r.log.Info("starting NextGen billing report task")

	// Get checkpoint
	checkpoint, err := r.loader.LoadCheckpoint(ctx)
	if err != nil {
		r.log.Error("failed to load checkpoint", zap.Error(err))
		return err
	}

	// Use current time as end time (no delay applied)
	currentTime := time.Now()
	endTime := currentTime.Unix()

	// If checkpoint is 0, start from timestamp 0 (beginning of time)
	if checkpoint == 0 {
		r.log.Info("no checkpoint found, starting from beginning", zap.Int64("start_time", checkpoint))
	}

	// If checkpoint is already greater than or equal to endTime, no processing needed
	if checkpoint >= endTime {
		r.log.Info("checkpoint is up to date, no new data to report",
			zap.Int64("checkpoint", checkpoint),
			zap.Int64("end_time", endTime))
		return nil
	}

	r.log.Info("loading metering data",
		zap.Int64("start_time", checkpoint),
		zap.Int64("end_time", endTime))

	// Load data from DeltaLake - choose appropriate method based on configuration
	reports, err := r.loadMeteringData(ctx, checkpoint, endTime)
	if err != nil {
		r.log.Error("failed to load metering data", zap.Error(err))
		return err
	}

	if len(reports) == 0 {
		r.log.Info("no new metering data found")
		// When no data found, keep checkpoint unchanged
		return nil
	}

	r.log.Info("loaded metering data", zap.Int("count", len(reports)))

	// Send data directly without grouping (SendMeasure only supports single item sending)
	var newCheckpoint int64 = checkpoint
	totalReports := 0
	for _, report := range reports {
		// Convert BillingReportV2Deltalake to BillingReport before sending
		billingReport, err := report.Convert2BillingReport()
		if err != nil {
			r.log.Error("failed to convert report",
				zap.String("cost_id", report.CostID),
				zap.Error(err))
			return err
		}

		r.log.Debug("billing records for analysis", zap.Any("billing_record", billingReport))

		err = r.sender.SendMeasure(ctx, billingReport)
		if err != nil {
			r.log.Error("failed to send report",
				zap.String("cost_id", report.CostID),
				zap.Error(err))
			return err
		}
		// r.log.Debug("send measure success", zap.Uint64("tenantid", report.TenantID), zap.String("costid", report.CostID), zap.Int64("costtimestamp", report.CostTimestamp))
		totalReports++

		// Update checkpoint to the latest created_at time
		if report.CreatedAt > newCheckpoint {
			newCheckpoint = report.CreatedAt
		}
	}

	r.log.Info("successfully sent all reports", zap.Int("total_reports", totalReports))

	// Save the latest checkpoint
	err = r.loader.SaveCheckpoint(ctx, newCheckpoint)
	if err != nil {
		r.log.Error("failed to save checkpoint", zap.Error(err))
		return err
	}

	r.log.Info("NextGen billing report task completed successfully",
		zap.Int64("new_checkpoint", newCheckpoint),
		zap.Int("total_reports", totalReports))

	return nil
}

// groupReportsByHour groups reports by hour
// groupReportsByHour groups reports by hour for potential future batch processing
// NOTE: Currently unused as SendMeasure only supports single item sending
// Kept for potential future use when batch sending is supported
func (r *NextGenBillingRepoter) groupReportsByHour(reports []*models.BillingReportV2Deltalake) ([]*HourlyBatch, error) {
	// Group by hour
	hourlyGroups := make(map[time.Time][]*models.BillingReportV2Deltalake)

	for _, report := range reports {
		// Convert timestamp to rounded hour
		reportTime := time.Unix(report.CostTimestamp, 0).UTC()
		hourTime := time.Date(reportTime.Year(), reportTime.Month(), reportTime.Day(),
			reportTime.Hour(), 0, 0, 0, time.UTC)

		hourlyGroups[hourTime] = append(hourlyGroups[hourTime], report)
	}

	// Convert to HourlyBatch array and sort
	batches := make([]*HourlyBatch, 0, len(hourlyGroups))
	for hourTime, reports := range hourlyGroups {
		batch := &HourlyBatch{
			Hour:      hourTime,
			StartTime: hourTime.Unix(),
			EndTime:   hourTime.Add(time.Hour).Unix(),
			Reports:   reports,
		}
		batches = append(batches, batch)
	}

	// Sort by time
	sort.Slice(batches, func(i, j int) bool {
		return batches[i].StartTime < batches[j].StartTime
	})

	return batches, nil
}

// sendBatch sends one hour batch of data to SQS
// sendBatch sends a batch of reports for a specific hour
// NOTE: Currently unused as SendMeasure only supports single item sending
// Kept for potential future use when batch sending is supported
func (r *NextGenBillingRepoter) sendBatch(ctx context.Context, batch *HourlyBatch) error {
	for _, report := range batch.Reports {
		// Convert BillingReportV2Deltalake to BillingReport before sending
		billingReport, err := report.Convert2BillingReport()
		if err != nil {
			r.log.Error("failed to convert to billing report",
				zap.String("cost_id", report.CostID),
				zap.Time("hour", batch.Hour),
				zap.Error(err))
			return err
		}

		err = r.sender.SendMeasure(ctx, billingReport)
		if err != nil {
			r.log.Error("failed to send measure",
				zap.String("cost_id", report.CostID),
				zap.Time("hour", batch.Hour),
				zap.Error(err))
			return err
		}
	}
	return nil
}

// loadMeteringData chooses the appropriate data loading method based on vendor configuration
func (r *NextGenBillingRepoter) loadMeteringData(ctx context.Context, startTime, endTime int64) ([]*models.BillingReportV2Deltalake, error) {
	// For Alicloud vendor, use delta-rs-golang SDK to avoid DuckDB compatibility issues with OSS
	if strings.EqualFold(r.conf.Vendor, constants.VendorAlicloud) {
		r.log.Info("using delta-rs-golang SDK for Alicloud OSS compatibility")
		return r.loader.LoadMeteringDataInTimeRangeWithDeltaRS(ctx, startTime, endTime)
	}

	// For other vendors (AWS, GCP, Azure), use DuckDB method
	r.log.Info("using DuckDB method for data loading")
	return r.loader.LoadMeteringDataInTimeRange(ctx, startTime, endTime)
}