package nextgenrepoter

import (
	"context"
	"database/sql"
	"fmt"
	"strconv"
	"time"

	_ "github.com/duckdb/duckdb-go/v2"
	deltalake "github.com/tidbcloud/metering/delta-rs-golang"
	"github.com/tidbcloud/metering/pkg/common/checkpoint"
	"github.com/tidbcloud/metering/pkg/common/config"
	"github.com/tidbcloud/metering/pkg/common/models"
	"github.com/tidbcloud/metering/pkg/common/utils"
	"github.com/tidbcloud/metering/pkg/common/utils/deltautils"
	"github.com/tidbcloud/metering/pkg/common/utils/duckdb"
	"go.uber.org/zap"
)

const (
	nextgenReporterTaskName = "nextgen_reporter"
)

type duckDBClient interface {
	QueryData(tablePath, customSQL string) (*sql.Rows, error)
	Close() error
}

type duckDBClientFactory interface {
	NewDuckDBClientWithConfig(conf *config.ServiceConfig, log *zap.Logger) (duckDBClient, error)
}

type defaultDuckDBClientFactory struct{}

func (f *defaultDuckDBClientFactory) NewDuckDBClientWithConfig(conf *config.ServiceConfig, log *zap.Logger) (duckDBClient, error) {
	return duckdb.NewDuckDBClientWithConfig(conf, log)
}

var defaultDuckDBFactory duckDBClientFactory = &defaultDuckDBClientFactory{}

type checkpointFactory interface {
	CreateCheckpointManager() (checkpoint.NextGenCheckpoint, error)
}

var newCheckpointFactory = func(conf *config.ServiceConfig, log *zap.Logger) checkpointFactory {
	return checkpoint.NewFactory(conf, log)
}

// tableManagerInterface abstracts the TableManager for testing
type tableManagerInterface interface {
	OpenTable(ctx context.Context) error
	Query(ctx context.Context, sql string, options *deltalake.QueryOptions) ([]map[string]interface{}, error)
}

// NextGenRepoterLoaderImpl NextGen reporter data loader implementation
type NextGenRepoterLoaderImpl struct {
	conf          *config.ServiceConfig
	log           *zap.Logger
	checkpointMgr checkpoint.NextGenCheckpoint
	duckFactory   duckDBClientFactory
	tableManager  tableManagerInterface
}

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

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

	// Get or initialize global table manager
	tableManager := deltautils.GetGlobalTableManager()
	if tableManager == nil {
		tableManager, err = deltautils.InitGlobalTableManager(conf, log)
		if err != nil {
			return nil, fmt.Errorf("failed to initialize global table manager: %w", err)
		}
	}

	return &NextGenRepoterLoaderImpl{
		conf:          conf,
		log:           log,
		checkpointMgr: checkpointMgr,
		duckFactory:   defaultDuckDBFactory,
		tableManager:  tableManager,
	}, nil
}

// LoadCheckpoint loads checkpoint - using the same method as nextGenTask
func (l *NextGenRepoterLoaderImpl) LoadCheckpoint(ctx context.Context) (int64, error) {
	checkpoint, err := l.checkpointMgr.GetCheckpoint(ctx, nextgenReporterTaskName)
	if err != nil {
		l.log.Error("failed to get checkpoint", zap.Error(err))
		return 0, err
	}

	l.log.Info("loaded checkpoint", zap.Int64("timestamp", checkpoint))
	return checkpoint, nil
}

// SaveCheckpoint saves checkpoint - using the same method as nextGenTask
func (l *NextGenRepoterLoaderImpl) SaveCheckpoint(ctx context.Context, checkpoint int64) error {
	err := l.checkpointMgr.SaveCheckpoint(ctx, nextgenReporterTaskName, checkpoint)
	if err != nil {
		l.log.Error("failed to save checkpoint", zap.Error(err))
		return err
	}

	l.log.Info("saved checkpoint", zap.Int64("timestamp", checkpoint))
	return nil
}

// LoadMeteringDataInTimeRange uses DuckDB to read metering data from startTime to latest from DeltaLake
func (l *NextGenRepoterLoaderImpl) LoadMeteringDataInTimeRange(ctx context.Context, startTime, endTime int64) ([]*models.BillingReportV2Deltalake, error) {
	l.log.Info("loading metering data using DuckDB",
		zap.Int64("start_time", startTime),
		zap.String("delta_table_path", l.conf.NextGenMeteringConfig.DeltaTablePath))

	// Check if DeltaTablePath is configured
	if l.conf.NextGenMeteringConfig.DeltaTablePath == "" {
		l.log.Warn("DeltaTablePath not configured, returning empty result")
		return []*models.BillingReportV2Deltalake{}, nil
	}

	// Create DuckDB client with configuration
	factory := l.duckFactory
	if factory == nil {
		factory = defaultDuckDBFactory
	}

	duckClient, err := factory.NewDuckDBClientWithConfig(l.conf, l.log)
	if err != nil {
		return nil, fmt.Errorf("failed to create DuckDB client: %w", err)
	}
	defer duckClient.Close()

	// Build query SQL (no endTime constraint, get all data from startTime to latest)
	query := l.buildQuery(startTime)
	l.log.Debug("executing query", zap.String("sql", query))

	// Execute query using DuckDB client
	rows, err := duckClient.QueryData(l.conf.NextGenMeteringConfig.DeltaTablePath, query)
	if err != nil {
		return nil, fmt.Errorf("failed to execute query: %w", err)
	}
	defer rows.Close()

	// Parse results
	reports, err := l.parseQueryResults(rows)
	if err != nil {
		return nil, fmt.Errorf("failed to parse query results: %w", err)
	}

	l.log.Info("successfully loaded metering data using DuckDB",
		zap.Int("record_count", len(reports)))

	return reports, nil
}

// buildQuery builds query SQL for DuckDB
func (l *NextGenRepoterLoaderImpl) buildQuery(startTime int64) string {
	// Calculate hour_dimension for partition pruning (only start time for filtering)
	startHour := time.Unix(startTime, 0).UTC()

	// Round start time down to the beginning of the hour
	startHourRounded := time.Date(startHour.Year(), startHour.Month(), startHour.Day(), startHour.Hour(), 0, 0, 0, time.UTC)

	startHourDimension := utils.GenerateHourDimension(startHourRounded.Unix())

	// Build Delta Lake query SQL using placeholder that will be replaced by DuckDB client
	// Use both hour_dimension (for partition pruning) and created_at (for precise filtering)
	// to handle recalculations correctly and improve query performance
	// Only query public SKU types, exclude private SKUs
	// Use created_at > startTime (not >=) to avoid reprocessing already sent data
	query := fmt.Sprintf(`
		SELECT 
			scheme_version,
			tenant_id,
			cost_id,
			cost_time_stamp,
			unit,
			unit_value,
			cost,
			labels,
			created_at,
			sku_type,
			hour_dimension
		FROM {table_path}
		WHERE hour_dimension >= '%s' 
		  AND created_at > %d 
		  AND sku_type = '%s'
		ORDER BY created_at
	`, startHourDimension, startTime, models.NextGenSKUTypePublic)

	return query
}

// parseQueryResults parses query results
func (l *NextGenRepoterLoaderImpl) parseQueryResults(rows *sql.Rows) ([]*models.BillingReportV2Deltalake, error) {
	var reports []*models.BillingReportV2Deltalake

	for rows.Next() {
		var report models.BillingReportV2Deltalake
		var schemeVersionStr, tenantIDStr, costIDStr, costTimestampStr, unit, unitValue, costStr, labels, createdAtStr, skuType, hourDimension sql.NullString

		err := rows.Scan(
			&schemeVersionStr,
			&tenantIDStr,
			&costIDStr,
			&costTimestampStr,
			&unit,
			&unitValue,
			&costStr,
			&labels,
			&createdAtStr,
			&skuType,
			&hourDimension,
		)
		if err != nil {
			l.log.Warn("failed to scan row", zap.Error(err))
			continue
		}

		// Convert scheme_version
		if schemeVersionStr.Valid {
			report.SchemeVersion = schemeVersionStr.String
		}

		// Convert tenant_id
		if tenantIDStr.Valid {
			tenantID, err := strconv.ParseUint(tenantIDStr.String, 10, 64)
			if err != nil {
				l.log.Warn("invalid tenant_id", zap.String("value", tenantIDStr.String), zap.Error(err))
				continue
			}
			report.TenantID = tenantID
		}

		// Convert cost_id
		if costIDStr.Valid {
			report.CostID = costIDStr.String
		}

		// Convert cost_time_stamp
		if costTimestampStr.Valid {
			timestamp, err := strconv.ParseInt(costTimestampStr.String, 10, 64)
			if err != nil {
				l.log.Warn("invalid cost_time_stamp", zap.String("value", costTimestampStr.String), zap.Error(err))
				continue
			}
			report.CostTimestamp = timestamp
		}

		// Convert unit
		if unit.Valid {
			report.Unit = unit.String
		}

		// Convert unit_value
		if unitValue.Valid {
			report.UnitValue = unitValue.String
		}

		// Convert cost
		if costStr.Valid {
			cost, err := strconv.ParseInt(costStr.String, 10, 64)
			if err != nil {
				l.log.Warn("invalid cost", zap.String("value", costStr.String), zap.Error(err))
				continue
			}
			report.Cost = cost
		}

		// Convert labels
		if labels.Valid {
			report.Labels = labels.String
		}

		// Convert created_at
		if createdAtStr.Valid {
			createdAt, err := strconv.ParseInt(createdAtStr.String, 10, 64)
			if err != nil {
				l.log.Warn("invalid created_at", zap.String("value", createdAtStr.String), zap.Error(err))
				continue
			}
			report.CreatedAt = createdAt
		}

		// Convert sku_type
		if skuType.Valid {
			report.SKUType = skuType.String
		}

		// Convert hour_dimension
		if hourDimension.Valid {
			report.HourDimension = hourDimension.String
		}

		// Validate required fields
		if report.TenantID == 0 || report.CostID == "" {
			l.log.Warn("skipping record with missing required fields",
				zap.Uint64("tenant_id", report.TenantID),
				zap.String("cost_id", report.CostID),
				zap.Int64("cost_timestemp", report.CostTimestamp),
				zap.String("labels", report.Labels),
				zap.String("sku_type", report.SKUType),
				zap.String("hour_dimension", report.HourDimension),
				zap.String("unit", report.Unit),
				zap.String("unit_value", report.UnitValue),
				zap.String("cost", costStr.String),
			)
			continue
		}

		reports = append(reports, &report)
	}

	if err := rows.Err(); err != nil {
		return nil, fmt.Errorf("error iterating over rows: %w", err)
	}

	return reports, nil
}

// LoadMeteringDataInTimeRangeWithDeltaRS uses delta-rs-golang SDK to read metering data from startTime to latest from DeltaLake
// This method is designed for Alicloud OSS compatibility where DuckDB may not work properly
func (l *NextGenRepoterLoaderImpl) LoadMeteringDataInTimeRangeWithDeltaRS(ctx context.Context, startTime, endTime int64) ([]*models.BillingReportV2Deltalake, error) {
	l.log.Info("loading metering data using delta-rs-golang SDK",
		zap.Int64("start_time", startTime),
		zap.String("delta_table_path", l.conf.NextGenMeteringConfig.DeltaTablePath))

	// Check if DeltaTablePath is configured
	if l.conf.NextGenMeteringConfig.DeltaTablePath == "" {
		l.log.Warn("DeltaTablePath not configured, returning empty result")
		return []*models.BillingReportV2Deltalake{}, nil
	}

	// Check if table manager is initialized
	if l.tableManager == nil {
		return nil, fmt.Errorf("table manager not initialized")
	}

	// Ensure table is opened before query (if already open, this returns immediately)
	if err := l.tableManager.OpenTable(ctx); err != nil {
		return nil, fmt.Errorf("failed to open delta table: %w", err)
	}

	// Build query SQL
	query := l.buildQueryForDeltaRS(startTime)
	l.log.Debug("executing delta-rs query", zap.String("sql", query))

	// Execute query using global table manager
	queryOptions := &deltalake.QueryOptions{
		TableAlias: "metering_table",
	}

	rows, err := l.tableManager.Query(ctx, query, queryOptions)
	if err != nil {
		return nil, fmt.Errorf("failed to execute delta-rs query: %w", err)
	}

	// Parse results from delta-rs format
	reports, err := l.parseQueryResultsFromDeltaRS(rows)
	if err != nil {
		return nil, fmt.Errorf("failed to parse delta-rs query results: %w", err)
	}

	l.log.Info("successfully loaded metering data using delta-rs-golang SDK",
		zap.Int("record_count", len(reports)))

	return reports, nil
}

// buildQueryForDeltaRS builds query SQL for delta-rs-golang Query method
func (l *NextGenRepoterLoaderImpl) buildQueryForDeltaRS(startTime int64) string {
	// Calculate hour_dimension for partition pruning (only start time for filtering)
	startHour := time.Unix(startTime, 0).UTC()

	// Round start time down to the beginning of the hour
	startHourRounded := time.Date(startHour.Year(), startHour.Month(), startHour.Day(), startHour.Hour(), 0, 0, 0, time.UTC)

	startHourDimension := utils.GenerateHourDimension(startHourRounded.Unix())

	// Build Delta Lake query SQL for delta-rs-golang
	// Use both hour_dimension (for partition pruning) and created_at (for precise filtering)
	// to handle recalculations correctly and improve query performance
	// Only query public SKU types, exclude private SKUs
	// Use created_at > startTime (not >=) to avoid reprocessing already sent data
	query := fmt.Sprintf(`
		SELECT 
			scheme_version,
			tenant_id,
			cost_id,
			cost_time_stamp,
			unit,
			unit_value,
			cost,
			labels,
			created_at,
			sku_type,
			hour_dimension
		FROM metering_table
		WHERE hour_dimension >= '%s' 
		  AND created_at > %d 
		  AND sku_type = '%s'
		ORDER BY created_at
	`, startHourDimension, startTime, models.NextGenSKUTypePublic)

	return query
}

// parseQueryResultsFromDeltaRS parses query results from delta-rs-golang Query method
func (l *NextGenRepoterLoaderImpl) parseQueryResultsFromDeltaRS(rows []map[string]interface{}) ([]*models.BillingReportV2Deltalake, error) {
	reports := make([]*models.BillingReportV2Deltalake, 0, len(rows))

	for _, row := range rows {
		var report models.BillingReportV2Deltalake

		// Convert scheme_version
		if val, ok := row["scheme_version"]; ok && val != nil {
			report.SchemeVersion = fmt.Sprintf("%v", val)
		}

		// Convert tenant_id
		if val, ok := row["tenant_id"]; ok && val != nil {
			if tenantID, err := l.convertToUint64(val); err == nil {
				report.TenantID = tenantID
			} else {
				l.log.Warn("invalid tenant_id", zap.Any("value", val), zap.Error(err))
				continue
			}
		}

		// Convert cost_id
		if val, ok := row["cost_id"]; ok && val != nil {
			report.CostID = fmt.Sprintf("%v", val)
		}

		// Convert cost_time_stamp
		if val, ok := row["cost_time_stamp"]; ok && val != nil {
			if timestamp, err := l.convertToInt64(val); err == nil {
				report.CostTimestamp = timestamp
			} else {
				l.log.Warn("invalid cost_time_stamp", zap.Any("value", val), zap.Error(err))
				continue
			}
		}

		// Convert unit
		if val, ok := row["unit"]; ok && val != nil {
			report.Unit = fmt.Sprintf("%v", val)
		}

		// Convert unit_value
		if val, ok := row["unit_value"]; ok && val != nil {
			report.UnitValue = fmt.Sprintf("%v", val)
		}

		// Convert cost
		if val, ok := row["cost"]; ok && val != nil {
			if cost, err := l.convertToInt64(val); err == nil {
				report.Cost = cost
			} else {
				l.log.Warn("invalid cost", zap.Any("value", val), zap.Error(err))
				continue
			}
		}

		// Convert labels
		if val, ok := row["labels"]; ok && val != nil {
			report.Labels = fmt.Sprintf("%v", val)
		}

		// Convert created_at
		if val, ok := row["created_at"]; ok && val != nil {
			if createdAt, err := l.convertToInt64(val); err == nil {
				report.CreatedAt = createdAt
			} else {
				l.log.Warn("invalid created_at", zap.Any("value", val), zap.Error(err))
				continue
			}
		}

		// Convert sku_type
		if val, ok := row["sku_type"]; ok && val != nil {
			report.SKUType = fmt.Sprintf("%v", val)
		}

		// Convert hour_dimension
		if val, ok := row["hour_dimension"]; ok && val != nil {
			report.HourDimension = fmt.Sprintf("%v", val)
		}

		// Validate required fields
		if report.TenantID == 0 || report.CostID == "" {
			l.log.Warn("skipping record with missing required fields",
				zap.Uint64("tenant_id", report.TenantID),
				zap.String("cost_id", report.CostID),
				zap.Int64("cost_timestamp", report.CostTimestamp),
				zap.String("labels", report.Labels),
				zap.String("sku_type", report.SKUType),
				zap.String("hour_dimension", report.HourDimension),
				zap.String("unit", report.Unit),
				zap.String("unit_value", report.UnitValue),
				zap.Int64("cost", report.Cost))
			continue
		}

		reports = append(reports, &report)
	}

	return reports, nil
}

// convertToInt64 converts various numeric types to int64
func (l *NextGenRepoterLoaderImpl) convertToInt64(value interface{}) (int64, error) {
	switch v := value.(type) {
	case int64:
		return v, nil
	case int:
		return int64(v), nil
	case int32:
		return int64(v), nil
	case float64:
		return int64(v), nil
	case float32:
		return int64(v), nil
	case string:
		return strconv.ParseInt(v, 10, 64)
	default:
		return 0, fmt.Errorf("unsupported type %T for int64 conversion", value)
	}
}

// convertToUint64 converts various numeric types to uint64
func (l *NextGenRepoterLoaderImpl) convertToUint64(value interface{}) (uint64, error) {
	switch v := value.(type) {
	case uint64:
		return v, nil
	case int64:
		if v < 0 {
			return 0, fmt.Errorf("negative value %d cannot be converted to uint64", v)
		}
		return uint64(v), nil
	case int:
		if v < 0 {
			return 0, fmt.Errorf("negative value %d cannot be converted to uint64", v)
		}
		return uint64(v), nil
	case int32:
		if v < 0 {
			return 0, fmt.Errorf("negative value %d cannot be converted to uint64", v)
		}
		return uint64(v), nil
	case uint32:
		return uint64(v), nil
	case float64:
		if v < 0 {
			return 0, fmt.Errorf("negative value %f cannot be converted to uint64", v)
		}
		return uint64(v), nil
	case float32:
		if v < 0 {
			return 0, fmt.Errorf("negative value %f cannot be converted to uint64", v)
		}
		return uint64(v), nil
	case string:
		return strconv.ParseUint(v, 10, 64)
	default:
		return 0, fmt.Errorf("unsupported type %T for uint64 conversion", value)
	}
}