package nextgenmaintenance

import (
	"context"
	"fmt"
	"time"

	"github.com/tidbcloud/metering/pkg/common/config"
	"github.com/tidbcloud/metering/pkg/common/utils"
	"github.com/tidbcloud/metering/pkg/common/utils/deltautils"
	"go.uber.org/zap"
)

// NextGenMaintenance orchestrates maintenance tasks for the NextGen Delta Lake table.
type NextGenMaintenance struct {
	conf         *config.ServiceConfig
	log          *zap.Logger
	now          func() time.Time
	tableManager *deltautils.TableManager
}

// NewNextGenMaintenance creates a new NextGenMaintenance instance when maintenance is enabled.
func NewNextGenMaintenance(conf *config.ServiceConfig, log *zap.Logger) (*NextGenMaintenance, error) {
	if conf == nil {
		return nil, fmt.Errorf("service config is required")
	}
	if conf.NextGenMeteringConfig == nil {
		return nil, fmt.Errorf("NextGenMeteringConfig is required")
	}
	if conf.NextGenMeteringConfig.MaintenanceConfig == nil {
		return nil, fmt.Errorf("NextGen maintenance config is required")
	}
	if !conf.NextGenMeteringConfig.MaintenanceConfig.Enable {
		return nil, fmt.Errorf("NextGen maintenance task is disabled in configuration")
	}
	if conf.NextGenMeteringConfig.DeltaTablePath == "" {
		return nil, fmt.Errorf("DeltaTablePath is required for maintenance task")
	}

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

	return &NextGenMaintenance{
		conf:         conf,
		log:          log.With(zap.String("task", "nextgen-maintenance")),
		now:          time.Now,
		tableManager: tableManager,
	}, nil
}

// Run executes the maintenance workflow: delete old data, compact files, and vacuum storage.
func (m *NextGenMaintenance) Run(ctx context.Context) error {
	cfg := m.conf.NextGenMeteringConfig
	maint := cfg.MaintenanceConfig

	m.log.Info("starting NextGen maintenance run",
		zap.String("delta_table_path", cfg.DeltaTablePath))

	// Ensure table is open before maintenance operations
	if err := m.tableManager.OpenTable(ctx); err != nil {
		m.log.Error("failed to open Delta Lake table", zap.Error(err))
		return fmt.Errorf("failed to open Delta Lake table: %w", err)
	}

	// Execute delete operation if enabled
	if maint.EnableDelete {
		cutoff, err := m.resolveDeleteCutoff()
		if err != nil {
			m.log.Error("invalid delete configuration", zap.Error(err))
			return err
		}

		predicate := buildDeletePredicate(cutoff)
		m.log.Info("deleting outdated records",
			zap.Time("cutoff", cutoff),
			zap.String("predicate", predicate))

		if _, err := m.tableManager.Delete(ctx, predicate); err != nil {
			m.log.Error("delete operation failed", zap.Error(err))
			return fmt.Errorf("failed to delete outdated records: %w", err)
		}
	} else {
		m.log.Info("delete step disabled by configuration")
	}

	// Execute compact operation
	m.log.Info("running compact operation")
	if _, err := m.tableManager.Compact(ctx); err != nil {
		m.log.Error("compact operation failed", zap.Error(err))
		return fmt.Errorf("failed to compact Delta table: %w", err)
	}

	// Execute vacuum operation if enabled
	if maint.EnableVacuum {
		retention := maint.VacuumRetentionHours
		if retention <= 0 {
			retention = 168 // Delta Lake default of 7 days
		}

		vacuumOpts := &deltautils.VacuumOptions{
			RetentionHours: retention,
			DryRun:         maint.VacuumDryRun,
		}

		if _, err := m.tableManager.Vacuum(ctx, vacuumOpts); err != nil {
			m.log.Error("vacuum operation failed", zap.Error(err))
			return fmt.Errorf("failed to vacuum Delta table: %w", err)
		}
	} else {
		m.log.Info("vacuum step disabled by configuration")
	}

	m.log.Info("NextGen maintenance run completed successfully")
	return nil
}

// resolveDeleteCutoff determines the cutoff timestamp for deleting historical data.
func (m *NextGenMaintenance) resolveDeleteCutoff() (time.Time, error) {
	maint := m.conf.NextGenMeteringConfig.MaintenanceConfig
	now := m.now().UTC()

	if maint.DeleteBeforeTimestamp > 0 {
		cutoff := time.Unix(maint.DeleteBeforeTimestamp, 0).UTC()
		if cutoff.After(now) {
			return time.Time{}, fmt.Errorf("deleteBeforeTimestamp %d is in the future", maint.DeleteBeforeTimestamp)
		}
		return cutoff, nil
	}

	if maint.DeleteRetentionDays > 0 {
		cutoff := now.Add(-time.Duration(maint.DeleteRetentionDays) * 24 * time.Hour)
		return cutoff, nil
	}

	return time.Time{}, fmt.Errorf("delete configuration requires deleteBeforeTimestamp or deleteRetentionDays")
}

// buildDeletePredicate creates the Delta Lake predicate for removing rows older than the cutoff hour partition.
func buildDeletePredicate(cutoff time.Time) string {
	cutoffHour := utils.GenerateHourDimension(cutoff.UTC().Unix())
	return fmt.Sprintf("hour_dimension < '%s'", cutoffHour)
}

// GetTableVersion returns the current version of the Delta Lake table
func (m *NextGenMaintenance) GetTableVersion(ctx context.Context) (int64, error) {
	// Ensure table is open before getting version
	if err := m.tableManager.OpenTable(ctx); err != nil {
		return 0, fmt.Errorf("failed to open Delta Lake table: %w", err)
	}
	return m.tableManager.GetTableVersion(ctx)
}