//go:build ignore

package main

import (
	"bufio"
	"encoding/json"
	"flag"
	"fmt"
	"os"
	"path/filepath"
	"sort"
	"strings"
	"time"

	"github.com/pingcap/log"
	v1 "github.com/tidbcloud/observability/api/regionalserver/resourceworker/v1"
	"go.uber.org/zap"
)

// Config holds the application configuration
type Config struct {
	InputDir      string
	OutputDir     string
	FilterLabels  map[string]string
	FilterBefore  time.Time
	BatchSize     int
	Workers       int
	Execute       bool
	MySQLHost     string
	MySQLPort     string
	MySQLUser     string
	MySQLPassword string
	MySQLDatabase string
}

func main() {
	config := parseFlags()

	if err := validateConfig(config); err != nil {
		log.Error("Configuration error", zap.Error(err))
		os.Exit(1)
	}

	startTime := time.Now()
	log.Info("=== Generate SQL from Cluster CRs (Go Version) ===")
	log.Info("Start time", zap.Time("start_time", startTime))

	if err := run(config); err != nil {
		log.Error("Execution failed", zap.Error(err))
		os.Exit(1)
	}

	duration := time.Since(startTime)
	log.Info("=== SQL Generation Complete ===")
	log.Info("End time", zap.Time("end_time", time.Now()))
	log.Info("Total duration", zap.Duration("duration", duration))
}

func parseFlags() *Config {
	config := &Config{
		FilterLabels: make(map[string]string),
	}

	var filterLabelsStr string
	var filterBeforeStr string

	flag.StringVar(&config.InputDir, "input-dir", "", "Input directory containing JSON files (required)")
	flag.StringVar(&config.InputDir, "i", "", "Input directory (short)")
	flag.StringVar(&config.OutputDir, "output-dir", "", "Output directory for SQL files")
	flag.StringVar(&config.OutputDir, "o", "", "Output directory (short)")
	flag.StringVar(&filterLabelsStr, "labels", "", "Filter by labels (e.g., env=prod,tier=premium)")
	flag.StringVar(&filterLabelsStr, "l", "", "Filter by labels (short)")
	flag.StringVar(&filterBeforeStr, "before-time", "", "Only convert CRs created before this time (RFC3339)")
	flag.StringVar(&filterBeforeStr, "t", "", "Before time (short)")
	flag.IntVar(&config.BatchSize, "batch-size", 100, "Batch size for processing")
	flag.IntVar(&config.BatchSize, "b", 100, "Batch size (short)")
	flag.IntVar(&config.Workers, "workers", 4, "Number of concurrent workers")
	flag.IntVar(&config.Workers, "w", 4, "Workers (short)")
	flag.BoolVar(&config.Execute, "execute", false, "Execute SQL directly in MySQL")
	flag.BoolVar(&config.Execute, "e", false, "Execute (short)")
	flag.StringVar(&config.MySQLHost, "mysql-host", "", "MySQL host")
	flag.StringVar(&config.MySQLPort, "mysql-port", "3306", "MySQL port")
	flag.StringVar(&config.MySQLUser, "mysql-user", "", "MySQL user")
	flag.StringVar(&config.MySQLPassword, "mysql-password", "", "MySQL password")
	flag.StringVar(&config.MySQLDatabase, "mysql-database", "", "MySQL database")

	flag.Parse()

	// Parse filter labels
	if filterLabelsStr != "" {
		for _, pair := range strings.Split(filterLabelsStr, ",") {
			parts := strings.SplitN(pair, "=", 2)
			if len(parts) == 2 {
				config.FilterLabels[parts[0]] = parts[1]
			}
		}
	}

	// Parse filter time
	if filterBeforeStr != "" {
		t, err := time.Parse(time.RFC3339, filterBeforeStr)
		if err != nil {
			log.Error("Invalid time format", zap.Error(err))
			os.Exit(1)
		}
		config.FilterBefore = t
	}

	// Set defaults
	if config.OutputDir == "" {
		config.OutputDir = filepath.Join(config.InputDir, "sql")
	}

	return config
}

func validateConfig(config *Config) error {
	if config.InputDir == "" {
		return fmt.Errorf("input directory is required")
	}

	if _, err := os.Stat(config.InputDir); os.IsNotExist(err) {
		return fmt.Errorf("input directory not found: %s", config.InputDir)
	}

	if config.Execute {
		if config.MySQLHost == "" || config.MySQLUser == "" ||
			config.MySQLPassword == "" || config.MySQLDatabase == "" {
			return fmt.Errorf("MySQL connection parameters required for --execute")
		}
	}

	if err := os.MkdirAll(config.OutputDir, 0755); err != nil {
		return fmt.Errorf("failed to create output directory: %w", err)
	}

	return nil
}

func run(config *Config) error {
	// Find batch files
	batchFiles, err := findBatchFiles(config.InputDir)
	if err != nil {
		return fmt.Errorf("failed to find batch files: %w", err)
	}

	if len(batchFiles) == 0 {
		return fmt.Errorf("no batch files found in %s/json/", config.InputDir)
	}

	log.Info("Found batch files", zap.Int("count", len(batchFiles)))

	// Process batches
	processedTotal := 0
	skippedTotal := 0
	failedTotal := 0

	for i := 0; i < len(batchFiles); i++ {
		batchNum := fmt.Sprintf("%04d", i)
		log.Info("Processing batch", zap.String("batch", batchNum), zap.Int("current", i+1), zap.Int("total", len(batchFiles)))

		processed, skipped, failed, err := processBatch(config, batchFiles[i], batchNum)
		if err != nil {
			log.Error("Failed to process batch", zap.String("batch", batchNum), zap.Error(err))
			failedTotal++
		}
		processedTotal += processed
		skippedTotal += skipped
		failedTotal += failed

		log.Info("Progress", zap.Int("processed", processedTotal), zap.Int("skipped", skippedTotal), zap.Int("failed", failedTotal))
	}

	// Generate summary
	if err := generateSummary(config, processedTotal, skippedTotal, failedTotal); err != nil {
		return fmt.Errorf("failed to generate summary: %w", err)
	}

	// Execute SQL if requested
	if config.Execute {
		if err := executeSQL(config); err != nil {
			return fmt.Errorf("failed to execute SQL: %w", err)
		}
	}

	return nil
}

func findBatchFiles(inputDir string) ([]string, error) {
	pattern := filepath.Join(inputDir, "json", "batch_*")
	files, err := filepath.Glob(pattern)
	if err != nil {
		return nil, err
	}
	sort.Strings(files)
	return files, nil
}

func processBatch(config *Config, batchFile, batchNum string) (int, int, int, error) {
	sqlFile := filepath.Join(config.OutputDir, fmt.Sprintf("batch_%s.sql", batchNum))

	f, err := os.Create(sqlFile)
	if err != nil {
		return 0, 0, 0, err
	}
	defer f.Close()

	writer := bufio.NewWriter(f)
	defer writer.Flush()

	// Write SQL header
	fmt.Fprintf(writer, "-- Migration Batch %s\n", batchNum)
	fmt.Fprintf(writer, "-- Generated at: %s\n", time.Now().Format(time.RFC3339))
	fmt.Fprintf(writer, "SET NAMES utf8mb4;\n")
	fmt.Fprintf(writer, "SET FOREIGN_KEY_CHECKS=0;\n\n")
	fmt.Fprintf(writer, "START TRANSACTION;\n\n")

	// Read and process CRs
	processed, skipped, failed, err := processFile(config, batchFile, writer)
	if err != nil {
		return processed, skipped, failed, err
	}

	// Write SQL footer
	fmt.Fprintf(writer, "COMMIT;\n")
	fmt.Fprintf(writer, "SET FOREIGN_KEY_CHECKS=1;\n")
	fmt.Fprintf(writer, "-- Batch %s complete: %d converted, %d skipped, %d failed\n",
		batchNum, processed, skipped, failed)

	return processed, skipped, failed, nil
}

func processFile(config *Config, filename string, writer *bufio.Writer) (int, int, int, error) {
	file, err := os.Open(filename)
	if err != nil {
		return 0, 0, 0, err
	}
	defer file.Close()

	scanner := bufio.NewScanner(file)
	// Increase buffer size for large JSON lines
	buf := make([]byte, 0, 1024*1024)
	scanner.Buffer(buf, 10*1024*1024)

	processed := 0
	skipped := 0
	failed := 0

	for scanner.Scan() {
		line := scanner.Bytes()
		if len(line) == 0 {
			continue
		}

		var cr v1.Cluster
		if err := json.Unmarshal(line, &cr); err != nil {
			failed++
			continue
		}

		// Apply filters
		if !matchesFilters(config, &cr) {
			skipped++
			continue
		}

		// Convert to SQL
		if err := convertToSQL(&cr, writer); err != nil {
			failed++
			continue
		}

		processed++
	}

	return processed, skipped, failed, scanner.Err()
}

func matchesFilters(config *Config, cr *v1.Cluster) bool {
	// Check label filters
	for key, value := range config.FilterLabels {
		if cr.Labels[key] != value {
			return false
		}
	}

	// Check time filter
	if !config.FilterBefore.IsZero() {
		if cr.CreationTimestamp.Time.After(config.FilterBefore) {
			return false
		}
	}

	return true
}

func convertToSQL(cr *v1.Cluster, writer *bufio.Writer) error {
	clusterID := cr.Name
	namespace := cr.Namespace

	// Marshal JSON fields
	labelsJSON, _ := json.Marshal(cr.Labels)
	annotationsJSON, _ := json.Marshal(cr.Annotations)
	ruleSelectorJSON, _ := json.Marshal(cr.Spec.VictoriaAccount.RuleSelector)
	alertManagersJSON, _ := json.Marshal(cr.Spec.VictoriaAccount.Alerting.Alertmanagers)

	// Extract label values
	region := cr.GetLabel("observability.pingcap.com/region")
	tenant := cr.GetLabel("observability.pingcap.com/tenantID")
	projectID := cr.GetLabel("observability.pingcap.com/projectID")
	applicationID := cr.GetLabel("observability.pingcap.com/applicationID")
	bizType := cr.GetLabel("observability.pingcap.com/bizType")
	ruleTemplateID := cr.GetLabel("observability.pingcap.com/rule-templateID")

	// Get timestamps
	creationTimestamp := cr.CreationTimestamp.Time.Format(time.RFC3339)
	var deletionTimestamp string
	if cr.DeletionTimestamp != nil {
		deletionTimestamp = cr.DeletionTimestamp.Time.Format(time.RFC3339)
	}

	// Get VictoriaAccount fields
	vmAccountID := fmt.Sprintf("%d", cr.Spec.VictoriaAccount.AccountID)
	vmProjectID := fmt.Sprintf("%d", cr.Spec.VictoriaAccount.ProjectID)
	vmMaxQueueSize := 0
	if cr.Spec.VictoriaAccount.MaxQueueSize != nil {
		vmMaxQueueSize = int(*cr.Spec.VictoriaAccount.MaxQueueSize)
	}

	// Get resource limits
	vmAlertCPURequest := cr.Spec.VictoriaAccount.AlertResources.Requests.Cpu().String()
	vmAlertCPULimit := cr.Spec.VictoriaAccount.AlertResources.Limits.Cpu().String()
	vmAlertMemoryRequest := cr.Spec.VictoriaAccount.AlertResources.Requests.Memory().String()
	vmAlertMemoryLimit := cr.Spec.VictoriaAccount.AlertResources.Limits.Memory().String()

	fmt.Fprintf(writer, "-- Cluster: %s/%s\n", namespace, clusterID)

	// INSERT into clusters table
	fmt.Fprintf(writer, "INSERT IGNORE INTO clusters (cluster_id, namespace, region, biz_type, tenant, project_id, application_id, rule_template_id, creation_timestamp")
	if deletionTimestamp != "" {
		fmt.Fprintf(writer, ", deletion_timestamp")
	}
	fmt.Fprintf(writer, ") VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s",
		sqlString(clusterID),
		sqlString(namespace),
		sqlString(region),
		sqlString(bizType),
		sqlString(tenant),
		sqlString(projectID),
		sqlString(applicationID),
		sqlString(ruleTemplateID),
		sqlString(creationTimestamp))
	if deletionTimestamp != "" {
		fmt.Fprintf(writer, ", %s", sqlString(deletionTimestamp))
	}
	fmt.Fprintf(writer, ");\n\n")

	// INSERT into cluster_specs table
	fmt.Fprintf(writer, "INSERT IGNORE INTO cluster_specs (cluster_id, labels, annotations, endpoint_service_name, "+
		"endpoint_service_user_arn, endpoint_service_shared, mute_alert, vm_account_id, vm_project_id, "+
		"vm_support_serverless, vm_cluster_name, vm_availability_zone, vm_alert_max_queue_size, vm_rule_selector, "+
		"vm_alert_cpu_request, vm_alert_cpu_limit, vm_alert_memory_request, vm_alert_memory_limit, vm_alert_managers) "+
		"VALUES (%s, %s, %s, %s, %s, %t, %t, %s, %s, %t, %s, %s, %d, %s, %s, %s, %s, %s, %s);\n\n",
		sqlString(clusterID),
		sqlString(string(labelsJSON)),
		sqlString(string(annotationsJSON)),
		sqlString(cr.Spec.Endpoint.Name),
		sqlString(cr.Spec.Endpoint.UserARN),
		cr.Spec.Endpoint.Shared,
		cr.Spec.MuteAlert,
		sqlString(vmAccountID),
		sqlString(vmProjectID),
		cr.Spec.VictoriaAccount.SupportServerless,
		sqlString(cr.Spec.VictoriaAccount.VMClusterName),
		sqlString(cr.Spec.VictoriaAccount.AvailabilityZone),
		vmMaxQueueSize,
		sqlString(string(ruleSelectorJSON)),
		sqlString(vmAlertCPURequest),
		sqlString(vmAlertCPULimit),
		sqlString(vmAlertMemoryRequest),
		sqlString(vmAlertMemoryLimit),
		sqlString(string(alertManagersJSON)))

	// INSERT into cluster_statuses table
	updateAt := ""
	if cr.Status.UpdateAt != "" {
		updateAt = cr.Status.UpdateAt
	}

	fmt.Fprintf(writer, "INSERT IGNORE INTO cluster_statuses (cluster_id, phase, dns_name, service_name, resource_version")
	if updateAt != "" {
		fmt.Fprintf(writer, ", update_at")
	}
	fmt.Fprintf(writer, ") VALUES (%s, %s, %s, %s, 1",
		sqlString(clusterID),
		sqlString(string(cr.Status.Phase)),
		sqlString(cr.Status.DnsName),
		sqlString(cr.Status.ServiceName))
	if updateAt != "" {
		fmt.Fprintf(writer, ", %s", sqlString(updateAt))
	}
	fmt.Fprintf(writer, ");\n\n")

	return nil
}

func sqlString(s string) string {
	if s == "" || s == "0" {
		return "NULL"
	}
	// Escape single quotes and backslashes
	s = strings.ReplaceAll(s, "\\", "\\\\")
	s = strings.ReplaceAll(s, "'", "''")
	return fmt.Sprintf("'%s'", s)
}

func generateSummary(config *Config, processed, skipped, failed int) error {
	summaryFile := filepath.Join(config.OutputDir, "migration_summary.sql")
	allSQLFile := filepath.Join(config.OutputDir, "migration_all.sql")

	// Create combined SQL file
	allSQL, err := os.Create(allSQLFile)
	if err != nil {
		return err
	}
	defer allSQL.Close()

	writer := bufio.NewWriter(allSQL)
	defer writer.Flush()

	// Write header
	fmt.Fprintf(writer, "-- Migration Summary\n")
	fmt.Fprintf(writer, "-- Generated at: %s\n", time.Now().Format(time.RFC3339))
	fmt.Fprintf(writer, "-- Input directory: %s\n", config.InputDir)
	fmt.Fprintf(writer, "-- Output directory: %s\n\n", config.OutputDir)

	// Combine all batch files
	batchFiles, _ := filepath.Glob(filepath.Join(config.OutputDir, "batch_*.sql"))
	sort.Strings(batchFiles)

	for _, batchFile := range batchFiles {
		content, err := os.ReadFile(batchFile)
		if err != nil {
			log.Error("Failed to read batch file", zap.String("file", batchFile), zap.Error(err))
			continue
		}
		if _, err := writer.Write(content); err != nil {
			log.Error("Failed to write batch file", zap.String("file", batchFile), zap.Error(err))
		}
	}

	log.Info("Combined SQL file", zap.String("file", allSQLFile))
	log.Info("Total clusters processed", zap.Int("count", processed))
	log.Info("Total skipped", zap.Int("count", skipped))
	log.Info("Total failed", zap.Int("count", failed))

	// Write summary file
	summary, err := os.Create(summaryFile)
	if err != nil {
		return err
	}
	defer summary.Close()

	fmt.Fprintf(summary, "-- Migration Summary\n")
	fmt.Fprintf(summary, "-- Total clusters: %d\n", processed)
	fmt.Fprintf(summary, "-- Total skipped: %d\n", skipped)
	fmt.Fprintf(summary, "-- Total failed: %d\n", failed)
	fmt.Fprintf(summary, "-- Batch files: %d\n\n", len(batchFiles))
	fmt.Fprintf(summary, "-- To execute:\n")
	fmt.Fprintf(summary, "-- mysql -h $MYSQL_HOST -u $MYSQL_USER -p$MYSQL_PASSWORD $MYSQL_DATABASE < %s\n", allSQLFile)

	return nil
}

func executeSQL(config *Config) error {
	log.Info("=== Executing SQL in MySQL ===")
	log.Info("MySQL Connection", zap.String("host", config.MySQLHost), zap.String("port", config.MySQLPort), zap.String("database", config.MySQLDatabase))

	// This would require MySQL driver - placeholder for now
	log.Info("SQL execution not yet implemented in Go version")
	log.Info("Please execute manually", zap.String("command", fmt.Sprintf("mysql -h %s -P %s -u %s -p %s < %s/migration_all.sql",
		config.MySQLHost, config.MySQLPort, config.MySQLUser, config.MySQLDatabase, config.OutputDir)))

	return nil
}