// Solar Data Ingestor - Multi-Format Parser
// Supports NOAA DSD (flux), SIDC SSN (sunspots), and NOAA Kp/Ap (geomagnetic indices)
//
// Format Detection:
//   1. NOAA DSD Format: Space-delimited, starts with date columns (YYYY MM DD)
//   2. SIDC SSN Format: Semicolon-delimited, sunspot number data
//   3. NOAA Kp/Ap Format: Fixed-width fields, geomagnetic indices
//
// Database:
//   - Table: wspr.solar_indices (ReplacingMergeTree on date field)
//   - Automatic deduplication by date (latest value wins)
//
// Usage:
//   solar-ingest --file /path/to/data.txt --format auto
//   solar-ingest --data /solar-data/raw --format noaa
package main

import (
	"bufio"
	"context"
	"flag"
	"fmt"
	"log"
	"os"
	"path/filepath"
	"strconv"
	"strings"
	"time"

	"github.com/ClickHouse/clickhouse-go/v2"
	"github.com/ClickHouse/clickhouse-go/v2/lib/driver"
)

const (
	// Default ClickHouse connection
	defaultHost     = "localhost"
	defaultPort     = 9000
	defaultDatabase = "wspr"
	defaultTable    = "solar_indices"
)

// DataFormat represents the type of solar data file
type DataFormat int

const (
	FormatAuto DataFormat = iota // Auto-detect from filename/content
	FormatNOAA                   // NOAA DSD space-delimited format (flux)
	FormatSIDC                   // SIDC semicolon-delimited format (SSN)
	FormatKpAp                   // NOAA fixed-width format (Kp/Ap)
)

// SolarRecord represents a complete solar indices record
// Matches the expanded ClickHouse schema with SSN, Kp, Ap, and source tracking
type SolarRecord struct {
	Date          time.Time // Primary key for ReplacingMergeTree
	Timestamp     time.Time
	ObservedFlux  float32
	AdjustedFlux  float32
	SSN           float32 // Sunspot Number
	KpIndex       float32 // Planetary K-index (0-9)
	ApIndex       float32 // Planetary A-index (0-400)
	Year          uint16
	Month         uint8
	Day           uint8
	OriginalValue float32
	SourceFile    string  // Raw vs. Curated: track source file for lineage
}

var (
	// Command-line flags
	dataFile  = flag.String("file", "", "Path to solar data file (overrides default)")
	dataDir   = flag.String("data", "/solar-data/raw", "Path to solar data directory")
	format    = flag.String("format", "auto", "Data format: auto, noaa, sidc, kpap")
	all       = flag.Bool("all", false, "Process all .txt and .csv files in data directory")
	host      = flag.String("host", defaultHost, "ClickHouse host")
	port      = flag.Int("port", defaultPort, "ClickHouse native port")
	database  = flag.String("database", defaultDatabase, "ClickHouse database")
	table     = flag.String("table", defaultTable, "ClickHouse table name")
	createTbl = flag.Bool("create-table", true, "Create table if not exists")
	batchSize = flag.Int("batch-size", 10000, "Batch size for inserts")
	verbose   = flag.Bool("verbose", false, "Verbose output")
)

func main() {
	flag.Parse()

	log.SetFlags(log.LstdFlags | log.Lshortfile)
	log.Printf("WSPR Solar Data Ingestor (Multi-Format) starting...")

	// Connect to ClickHouse
	conn, err := connectClickHouse()
	if err != nil {
		log.Fatalf("Failed to connect to ClickHouse: %v", err)
	}
	defer conn.Close()

	ctx := context.Background()

	// Create table if requested
	if *createTbl {
		if err := createTable(ctx, conn); err != nil {
			log.Fatalf("Failed to create table: %v", err)
		}
		log.Printf("✓ Table %s.%s ready (ReplacingMergeTree)", *database, *table)
	}

	// Bulk mode: process all files in directory using filepath.Walk
	if *all {
		log.Printf("Bulk mode: walking %s for solar data files...", *dataDir)

		// TRUNCATE TABLE for clean slate (prevents ReplacingMergeTree conflicts)
		log.Printf("Truncating %s.%s for clean bulk ingestion...", *database, *table)
		truncateQuery := fmt.Sprintf("TRUNCATE TABLE %s.%s", *database, *table)
		if err := conn.Exec(ctx, truncateQuery); err != nil {
			log.Fatalf("Failed to truncate table: %v", err)
		}
		log.Printf("✓ Table truncated successfully")

		// Use filepath.Walk to recursively scan directory
		files, err := walkDataDirectory(*dataDir)
		if err != nil {
			log.Fatalf("Failed to walk directory: %v", err)
		}

		if len(files) == 0 {
			log.Fatalf("No .txt or .csv files found in %s", *dataDir)
		}

		log.Printf("Found %d files to process", len(files))
		totalRecords := processBulkFiles(ctx, conn, files)
		log.Printf("✓ Bulk ingestion complete: %d total records from %d files", totalRecords, len(files))

		// Validate Trinity (SFI, SSN, Kp) on known active date
		if err := validateTrinity(ctx, conn); err != nil {
			log.Printf("⚠ Trinity validation warning: %v", err)
		}

		return
	}

	// Single-file mode
	var inputFile string
	if *dataFile != "" {
		inputFile = *dataFile
	} else {
		inputFile = filepath.Join(*dataDir, "sfi_daily_flux.txt")
	}

	log.Printf("Single-file mode: %s", inputFile)
	log.Printf("ClickHouse: %s:%d/%s.%s", *host, *port, *database, *table)

	recordCount, err := processFile(ctx, conn, inputFile)
	if err != nil {
		log.Fatalf("Failed to ingest data: %v", err)
	}

	log.Printf("✓ Total records ingested: %d", recordCount)
	log.Printf("✓ Solar data ingestion complete")
}

// connectClickHouse establishes connection using native protocol with LZ4 compression
func connectClickHouse() (driver.Conn, error) {
	conn, err := clickhouse.Open(&clickhouse.Options{
		Addr: []string{fmt.Sprintf("%s:%d", *host, *port)},
		Auth: clickhouse.Auth{
			Database: *database,
		},
		Compression: &clickhouse.Compression{
			Method: clickhouse.CompressionLZ4,
		},
		Settings: clickhouse.Settings{
			"max_execution_time": 60,
		},
		DialTimeout:  10 * time.Second,
		MaxOpenConns: 5,
		MaxIdleConns: 2,
	})

	if err != nil {
		return nil, err
	}

	// Test connection
	if err := conn.Ping(context.Background()); err != nil {
		return nil, fmt.Errorf("ping failed: %w", err)
	}

	return conn, nil
}

// walkDataDirectory recursively walks the directory tree for .txt and .csv files
// Uses filepath.Walk to handle nested directories and large file sets
func walkDataDirectory(rootPath string) ([]string, error) {
	var files []string

	err := filepath.Walk(rootPath, func(path string, info os.FileInfo, err error) error {
		if err != nil {
			// Log walk error but continue (fault tolerance)
			log.Printf("⚠ Walk error for %s: %v (continuing)", path, err)
			return nil
		}

		// Skip directories
		if info.IsDir() {
			return nil
		}

		// Only process .txt and .csv files
		ext := strings.ToLower(filepath.Ext(info.Name()))
		if ext == ".txt" || ext == ".csv" {
			files = append(files, path)
		}

		return nil
	})

	if err != nil {
		return nil, fmt.Errorf("walk directory tree: %w", err)
	}

	// Sort files by priority (historical first, then daily updates)
	sortFilesByPriority(files)

	return files, nil
}

// sortFilesByPriority orders files to ensure optimal ReplacingMergeTree behavior
// Priority: historical/complete datasets first, then daily updates
func sortFilesByPriority(files []string) {
	// Priority order (higher number = process first):
	// 1. Historical sunspot data (SN_d_tot_V2.0.csv) - oldest data (1818+)
	// 2. Historical Kp/Ap data (kp_ap_historical.txt) - old geomagnetic (1932+)
	// 3. Historical flux data (sfi_daily_flux.txt) - old solar flux (1947+)
	// 4. Recent/daily updates (everything else)

	priority := func(filename string) int {
		base := strings.ToLower(filepath.Base(filename))

		// Historical datasets (process first for best ReplacingMergeTree performance)
		if strings.Contains(base, "sn_d_tot") || strings.Contains(base, "sn_y_tot") {
			return 100 // SIDC sunspot data (1818+)
		}
		if strings.Contains(base, "kp_ap_historical") || strings.Contains(base, "kp_ap.txt") {
			return 90 // Kp/Ap historical (1932+)
		}
		if strings.Contains(base, "sfi_daily_flux") || strings.Contains(base, "dsd.txt") {
			return 80 // Solar flux historical (1947+)
		}

		// Recent/daily updates (process after historical)
		if strings.Contains(base, "recent") || strings.Contains(base, "daily") {
			return 50
		}
		if strings.Contains(base, "forecast") || strings.Contains(base, "predicted") {
			return 40
		}

		return 10 // Default priority for unknown files
	}

	// Sort by priority (descending - highest first)
	for i := 0; i < len(files)-1; i++ {
		for j := i + 1; j < len(files); j++ {
			if priority(files[i]) < priority(files[j]) {
				files[i], files[j] = files[j], files[i]
			}
		}
	}
}

// processFile ingests a single file with auto-detection
func processFile(ctx context.Context, conn driver.Conn, filePath string) (int, error) {
	// Detect format
	detectedFormat, err := detectFormat(filePath, *format)
	if err != nil {
		return 0, fmt.Errorf("detect format: %w", err)
	}

	if *verbose {
		log.Printf("Detected format: %s", formatName(detectedFormat))
	}

	// Ingest data based on format
	var recordCount int
	switch detectedFormat {
	case FormatNOAA:
		recordCount, err = ingestNOAA(ctx, conn, filePath)
	case FormatSIDC:
		recordCount, err = ingestSIDC(ctx, conn, filePath)
	case FormatKpAp:
		recordCount, err = ingestKpAp(ctx, conn, filePath)
	default:
		return 0, fmt.Errorf("unknown format: %d", detectedFormat)
	}

	if err != nil {
		return 0, fmt.Errorf("ingest: %w", err)
	}

	return recordCount, nil
}

// processBulkFiles processes multiple files with progress reporting
func processBulkFiles(ctx context.Context, conn driver.Conn, files []string) int {
	totalRecords := 0
	successCount := 0
	failCount := 0

	log.Printf("━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━")
	log.Printf("Starting bulk ingestion of %d files", len(files))
	log.Printf("━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━")

	for i, filePath := range files {
		filename := filepath.Base(filePath)
		progress := float64(i+1) / float64(len(files)) * 100

		log.Printf("[%d/%d] (%.1f%%) Processing: %s", i+1, len(files), progress, filename)

		recordCount, err := processFile(ctx, conn, filePath)
		if err != nil {
			log.Printf("  ✗ FAILED: %v", err)
			failCount++
			continue
		}

		totalRecords += recordCount
		successCount++
		log.Printf("  ✓ SUCCESS: %d records ingested", recordCount)
	}

	log.Printf("━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━")
	log.Printf("Bulk ingestion complete:")
	log.Printf("  Files processed: %d/%d", successCount, len(files))
	log.Printf("  Files failed:    %d", failCount)
	log.Printf("  Total records:   %d", totalRecords)
	log.Printf("━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━")

	return totalRecords
}

// createTable creates the solar_indices table with expanded schema (Raw Layer)
// Uses Date32 to support historical data from 1900-2283
// For dates before 1900, date field is set to 1900-01-01 and actual date is stored in year/month/day
// Includes source_file for lineage tracking (Raw vs. Curated architecture)
func createTable(ctx context.Context, conn driver.Conn) error {
	query := fmt.Sprintf(`
		CREATE TABLE IF NOT EXISTS %s.%s (
			date Date32,
			time DateTime,
			observed_flux Float32,
			adjusted_flux Float32,
			ssn Float32,
			kp_index Float32,
			ap_index Float32,
			year UInt16,
			month UInt8,
			day UInt8,
			original_value Float32,
			source_file LowCardinality(String)
		) ENGINE = ReplacingMergeTree()
		ORDER BY date
		SETTINGS index_granularity = 8192
	`, *database, *table)

	return conn.Exec(ctx, query)
}

// detectFormat auto-detects the data format based on filename and content
func detectFormat(filePath string, formatStr string) (DataFormat, error) {
	// Manual override
	switch strings.ToLower(formatStr) {
	case "noaa":
		return FormatNOAA, nil
	case "sidc":
		return FormatSIDC, nil
	case "kpap", "kp":
		return FormatKpAp, nil
	}

	// Auto-detect from filename
	basename := strings.ToLower(filepath.Base(filePath))

	if strings.Contains(basename, "sfi") || strings.Contains(basename, "flux") {
		return FormatNOAA, nil
	}

	if strings.Contains(basename, "ssn") || strings.Contains(basename, "sunspot") {
		return FormatSIDC, nil
	}

	if strings.Contains(basename, "kp") || strings.Contains(basename, "ap") {
		return FormatKpAp, nil
	}

	// Peek at file content
	file, err := os.Open(filePath)
	if err != nil {
		return FormatAuto, fmt.Errorf("open file: %w", err)
	}
	defer file.Close()

	scanner := bufio.NewScanner(file)
	for scanner.Scan() {
		line := strings.TrimSpace(scanner.Text())
		if line == "" || strings.HasPrefix(line, "#") {
			continue
		}

		// SIDC format has semicolons
		if strings.Contains(line, ";") {
			return FormatSIDC, nil
		}

		// NOAA format has space-delimited date (YYYY MM DD)
		fields := strings.Fields(line)
		if len(fields) >= 4 {
			if _, err := strconv.Atoi(fields[0]); err == nil {
				if _, err := strconv.Atoi(fields[1]); err == nil {
					return FormatNOAA, nil
				}
			}
		}

		break
	}

	// Default to NOAA if uncertain
	return FormatNOAA, nil
}

// formatName returns a human-readable format name
func formatName(format DataFormat) string {
	switch format {
	case FormatNOAA:
		return "NOAA DSD (Space-delimited)"
	case FormatSIDC:
		return "SIDC SSN (Semicolon-delimited)"
	case FormatKpAp:
		return "NOAA Kp/Ap (Fixed-width)"
	default:
		return "Unknown"
	}
}

// ingestNOAA parses and ingests NOAA sfi_daily_flux.txt data
// Format: YYYY MM DD Radio_Flux SESC_Sunspot_Number ...
// Fault-tolerant: uses continue instead of return for malformed lines
func ingestNOAA(ctx context.Context, conn driver.Conn, filePath string) (int, error) {
	file, err := os.Open(filePath)
	if err != nil {
		return 0, fmt.Errorf("open file: %w", err)
	}
	defer file.Close()

	// Extract filename for source tracking
	sourceFile := filepath.Base(filePath)

	var records []SolarRecord
	scanner := bufio.NewScanner(file)
	lineNum := 0
	recordCount := 0
	errorCount := 0

	for scanner.Scan() {
		lineNum++
		line := strings.TrimSpace(scanner.Text())

		// Skip header lines and empty lines
		if line == "" || strings.HasPrefix(line, "#") || strings.HasPrefix(line, ":") {
			continue
		}

		// Skip column headers and separators
		if strings.Contains(line, "Date") || strings.HasPrefix(line, "---") {
			continue
		}

		// Parse space-delimited: YYYY MM DD Radio_Flux ...
		fields := strings.Fields(line)
		if len(fields) < 4 {
			if *verbose {
				log.Printf("[%s:%d] Insufficient fields (%d) - skipping", sourceFile, lineNum, len(fields))
			}
			errorCount++
			continue // Fault tolerance
		}

		year, month, day, err := parseDate(fields[0], fields[1], fields[2], lineNum)
		if err != nil {
			if *verbose {
				log.Printf("[%s:%d] %v - skipping", sourceFile, lineNum, err)
			}
			errorCount++
			continue // Fault tolerance
		}

		// Parse radio flux (10.7cm Solar Flux Index)
		flux, err := strconv.ParseFloat(fields[3], 32)
		if err != nil {
			if *verbose {
				log.Printf("[%s:%d] Invalid flux '%s' - skipping", sourceFile, lineNum, fields[3])
			}
			errorCount++
			continue // Fault tolerance
		}

		// Normalize large values (data quality check)
		originalValue := float32(flux)
		if flux > 500 {
			flux /= 10
		}

		// Create date and timestamp (handles pre-1900 dates)
		date, clamped := createDate(year, month, day)
		timestamp := time.Date(year, time.Month(month), day, 20, 0, 0, 0, time.UTC)

		if clamped && *verbose {
			log.Printf("[%s:%d] Pre-1900 date (%d-%02d-%02d) clamped to 1900-01-01 in date field (preserved in year/month/day)",
				sourceFile, lineNum, year, month, day)
		}

		record := SolarRecord{
			Date:          date,
			Timestamp:     timestamp,
			ObservedFlux:  float32(flux),
			AdjustedFlux:  float32(flux),
			SSN:           0, // Not available in this format
			KpIndex:       0,
			ApIndex:       0,
			Year:          uint16(year),
			Month:         uint8(month),
			Day:           uint8(day),
			OriginalValue: originalValue,
			SourceFile:    sourceFile,
		}

		records = append(records, record)
		recordCount++

		// Batch insert
		if len(records) >= *batchSize {
			if err := insertBatch(ctx, conn, records); err != nil {
				log.Printf("[%s:%d] Batch insert error: %v - continuing", sourceFile, lineNum, err)
				errorCount++
				// Clear batch and continue (fault tolerance)
				records = records[:0]
				continue
			}
			if *verbose {
				log.Printf("[%s] Inserted batch of %d records", sourceFile, len(records))
			}
			records = records[:0]
		}
	}

	// Insert remaining records
	if len(records) > 0 {
		if err := insertBatch(ctx, conn, records); err != nil {
			log.Printf("[%s] Final batch insert error: %v", sourceFile, err)
			errorCount++
		}
	}

	if errorCount > 0 {
		log.Printf("[%s] Completed with %d errors (fault tolerance enabled)", sourceFile, errorCount)
	}

	return recordCount, scanner.Err()
}

// ingestSIDC parses and ingests SIDC sunspot number data
// Format: YYYY;MM;DD;Decimal_Year;SSN;StdDev;Observations;Definitive/Provisional
// Fault-tolerant: uses continue instead of return for malformed lines
func ingestSIDC(ctx context.Context, conn driver.Conn, filePath string) (int, error) {
	file, err := os.Open(filePath)
	if err != nil {
		return 0, fmt.Errorf("open file: %w", err)
	}
	defer file.Close()

	// Extract filename for source tracking
	sourceFile := filepath.Base(filePath)

	var records []SolarRecord
	scanner := bufio.NewScanner(file)
	lineNum := 0
	recordCount := 0
	errorCount := 0

	for scanner.Scan() {
		lineNum++
		line := strings.TrimSpace(scanner.Text())

		// Skip header lines and empty lines
		if line == "" || strings.HasPrefix(line, "#") {
			continue
		}

		// Parse semicolon-delimited format
		fields := strings.Split(line, ";")
		if len(fields) < 5 {
			if *verbose {
				log.Printf("[%s:%d] Insufficient fields (%d) - skipping", sourceFile, lineNum, len(fields))
			}
			errorCount++
			continue // Fault tolerance
		}

		year, month, day, err := parseDate(
			strings.TrimSpace(fields[0]),
			strings.TrimSpace(fields[1]),
			strings.TrimSpace(fields[2]),
			lineNum,
		)
		if err != nil {
			if *verbose {
				log.Printf("[%s:%d] %v - skipping", sourceFile, lineNum, err)
			}
			errorCount++
			continue // Fault tolerance
		}

		// Parse sunspot number (column 4)
		ssn, err := strconv.ParseFloat(strings.TrimSpace(fields[4]), 32)
		if err != nil {
			if *verbose {
				log.Printf("[%s:%d] Invalid SSN '%s' - skipping", sourceFile, lineNum, fields[4])
			}
			errorCount++
			continue // Fault tolerance
		}

		// Handle missing data (-1)
		if ssn < 0 {
			ssn = 0
		}

		// Create date and timestamp (handles pre-1900 dates)
		date, clamped := createDate(year, month, day)
		timestamp := time.Date(year, time.Month(month), day, 12, 0, 0, 0, time.UTC)

		if clamped && *verbose {
			log.Printf("[%s:%d] Pre-1900 date (%d-%02d-%02d) clamped to 1900-01-01 in date field (preserved in year/month/day)",
				sourceFile, lineNum, year, month, day)
		}

		record := SolarRecord{
			Date:          date,
			Timestamp:     timestamp,
			ObservedFlux:  0, // Not available in this format
			AdjustedFlux:  0,
			SSN:           float32(ssn),
			KpIndex:       0,
			ApIndex:       0,
			Year:          uint16(year),
			Month:         uint8(month),
			Day:           uint8(day),
			OriginalValue: float32(ssn),
			SourceFile:    sourceFile,
		}

		records = append(records, record)
		recordCount++

		// Batch insert
		if len(records) >= *batchSize {
			if err := insertBatch(ctx, conn, records); err != nil {
				log.Printf("[%s:%d] Batch insert error: %v - continuing", sourceFile, lineNum, err)
				errorCount++
				records = records[:0]
				continue // Fault tolerance
			}
			if *verbose {
				log.Printf("[%s] Inserted batch of %d records", sourceFile, len(records))
			}
			records = records[:0]
		}
	}

	// Insert remaining records
	if len(records) > 0 {
		if err := insertBatch(ctx, conn, records); err != nil {
			log.Printf("[%s] Final batch insert error: %v", sourceFile, err)
			errorCount++
		}
	}

	if errorCount > 0 {
		log.Printf("[%s] Completed with %d errors (fault tolerance enabled)", sourceFile, errorCount)
	}

	return recordCount, scanner.Err()
}

// ingestKpAp parses and ingests NOAA Kp/Ap geomagnetic index data
// Format: Fixed-width fields with date and 8 3-hour Kp values
// Uses strict character offsets with TrimSpace for precision
// Fault-tolerant: uses continue instead of return for malformed lines
func ingestKpAp(ctx context.Context, conn driver.Conn, filePath string) (int, error) {
	file, err := os.Open(filePath)
	if err != nil {
		return 0, fmt.Errorf("open file: %w", err)
	}
	defer file.Close()

	// Extract filename for source tracking
	sourceFile := filepath.Base(filePath)

	var records []SolarRecord
	scanner := bufio.NewScanner(file)
	lineNum := 0
	recordCount := 0
	errorCount := 0

	for scanner.Scan() {
		lineNum++
		line := scanner.Text() // Don't trim - fixed width format needs exact positions

		// Skip empty lines
		if len(strings.TrimSpace(line)) == 0 {
			continue
		}

		// Skip header/comment lines (lines starting with #)
		if strings.HasPrefix(line, "#") {
			continue
		}

		// GFZ Kp_ap format is 3-hourly data (8 records/day), minimum 59 chars
		// We only process 00.0 hour records for daily values
		if len(line) < 59 {
			if *verbose {
				log.Printf("[%s:%d] Insufficient length (%d chars, need 59) - skipping", sourceFile, lineNum, len(line))
			}
			errorCount++
			continue // Fault tolerance
		}

		// Fixed-width parsing with strict character offsets and TrimSpace:
		// Year[0:4] Month[5:7] Day[8:10] Hour[11:15] ... Kp[47:52] ap[55:57]
		yearStr := strings.TrimSpace(line[0:4])
		monthStr := strings.TrimSpace(line[5:7])
		dayStr := strings.TrimSpace(line[8:10])
		hourStr := strings.TrimSpace(line[11:15])
		kpStr := strings.TrimSpace(line[47:52])
		apStr := strings.TrimSpace(line[55:57])

		// Only process 00.0 hour records (first record of each day for daily values)
		if hourStr != "00.0" {
			continue
		}

		// Parse date fields (continue on error, never return)
		year, month, day, err := parseDate(yearStr, monthStr, dayStr, lineNum)
		if err != nil {
			if *verbose {
				log.Printf("[%s:%d] %v - skipping", sourceFile, lineNum, err)
			}
			errorCount++
			continue // Fault tolerance
		}

		// Parse Kp index (0-9 scale, continue on error)
		kp, err := strconv.ParseFloat(kpStr, 32)
		if err != nil {
			if *verbose {
				log.Printf("[%s:%d] Invalid Kp value '%s' - skipping", sourceFile, lineNum, kpStr)
			}
			errorCount++
			continue // Fault tolerance
		}

		// Parse ap index (0-400 scale, continue on error)
		ap, err := strconv.ParseFloat(apStr, 32)
		if err != nil {
			if *verbose {
				log.Printf("[%s:%d] Invalid ap value '%s' - skipping", sourceFile, lineNum, apStr)
			}
			errorCount++
			continue // Fault tolerance
		}

		// Create date and timestamp (handles pre-1900 dates)
		date, clamped := createDate(year, month, day)
		timestamp := time.Date(year, time.Month(month), day, 12, 0, 0, 0, time.UTC)

		if clamped && *verbose {
			log.Printf("[%s:%d] Pre-1900 date (%d-%02d-%02d) clamped to 1900-01-01 in date field (preserved in year/month/day)",
				sourceFile, lineNum, year, month, day)
		}

		record := SolarRecord{
			Date:          date,
			Timestamp:     timestamp,
			ObservedFlux:  0, // Not available in this format
			AdjustedFlux:  0,
			SSN:           0,
			KpIndex:       float32(kp),
			ApIndex:       float32(ap),
			Year:          uint16(year),
			Month:         uint8(month),
			Day:           uint8(day),
			OriginalValue: float32(ap),
			SourceFile:    sourceFile,
		}

		records = append(records, record)
		recordCount++

		// Batch insert
		if len(records) >= *batchSize {
			if err := insertBatch(ctx, conn, records); err != nil {
				log.Printf("[%s:%d] Batch insert error: %v - continuing", sourceFile, lineNum, err)
				errorCount++
				records = records[:0]
				continue // Fault tolerance
			}
			if *verbose {
				log.Printf("[%s] Inserted batch of %d records", sourceFile, len(records))
			}
			records = records[:0]
		}
	}

	// Insert remaining records
	if len(records) > 0 {
		if err := insertBatch(ctx, conn, records); err != nil {
			log.Printf("[%s] Final batch insert error: %v", sourceFile, err)
			errorCount++
		}
	}

	if errorCount > 0 {
		log.Printf("[%s] Completed with %d errors (fault tolerance enabled)", sourceFile, errorCount)
	}

	return recordCount, scanner.Err()
}

// isNumeric checks if a string contains only digits (used for footer detection)
func isNumeric(s string) bool {
	if len(s) == 0 {
		return false
	}
	for _, c := range s {
		if c < '0' || c > '9' {
			return false
		}
	}
	return true
}

// parseDate validates and parses year, month, day strings
// Supports historical data from 1800 to 2100
// Note: ClickHouse Date32 only supports 1900-2283, so dates before 1900
// will have the date field set to 1900-01-01 (but accurate year/month/day preserved)
func parseDate(yearStr, monthStr, dayStr string, lineNum int) (int, int, int, error) {
	year, err := strconv.Atoi(yearStr)
	// Lower bound to 1800 for historical solar data (earliest records from 1818)
	if err != nil || year < 1800 || year > 2100 {
		return 0, 0, 0, fmt.Errorf("invalid year '%s' (must be 1800-2100)", yearStr)
	}

	month, err := strconv.Atoi(monthStr)
	if err != nil || month < 1 || month > 12 {
		return 0, 0, 0, fmt.Errorf("invalid month '%s'", monthStr)
	}

	day, err := strconv.Atoi(dayStr)
	if err != nil || day < 1 || day > 31 {
		return 0, 0, 0, fmt.Errorf("invalid day '%s'", dayStr)
	}

	return year, month, day, nil
}

// createDate creates a time.Time date, clamping to 1900-01-01 minimum for Date32 compatibility
// Returns the clamped date for ClickHouse and whether it was clamped
func createDate(year, month, day int) (time.Time, bool) {
	// ClickHouse Date32 minimum is 1900-01-01
	if year < 1900 {
		// Return 1900-01-01 but flag that it was clamped
		// Actual date is preserved in year/month/day fields
		return time.Date(1900, 1, 1, 0, 0, 0, 0, time.UTC), true
	}
	return time.Date(year, time.Month(month), day, 0, 0, 0, 0, time.UTC), false
}

// insertBatch inserts a batch of records using PrepareBatch
// ReplacingMergeTree will automatically deduplicate by date
// Includes source_file for Raw vs. Curated lineage tracking
func insertBatch(ctx context.Context, conn driver.Conn, records []SolarRecord) error {
	if len(records) == 0 {
		return nil
	}

	batch, err := conn.PrepareBatch(ctx, fmt.Sprintf("INSERT INTO %s.%s", *database, *table))
	if err != nil {
		return fmt.Errorf("prepare batch: %w", err)
	}

	for _, rec := range records {
		if err := batch.Append(
			rec.Date,
			rec.Timestamp,
			rec.ObservedFlux,
			rec.AdjustedFlux,
			rec.SSN,
			rec.KpIndex,
			rec.ApIndex,
			rec.Year,
			rec.Month,
			rec.Day,
			rec.OriginalValue,
			rec.SourceFile,
		); err != nil {
			return fmt.Errorf("append to batch: %w", err)
		}
	}

	if err := batch.Send(); err != nil {
		return fmt.Errorf("send batch: %w", err)
	}

	return nil
}

// validateTrinity verifies that the Trinity (SFI, SSN, Kp) is populated correctly
// Queries the solar_master view for a known active date (2024-05-11)
func validateTrinity(ctx context.Context, conn driver.Conn) error {
	log.Printf("━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━")
	log.Printf("Trinity Validation (using wspr.solar_master view)")
	log.Printf("━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━")

	// Test known active date: 2024-05-11 (should have all three metrics)
	testDate := "2024-05-11"
	query := fmt.Sprintf(`
		SELECT
			date,
			observed_flux,
			adjusted_flux,
			ssn,
			kp_index,
			ap_index
		FROM wspr.solar_master
		WHERE date = '%s'
	`, testDate)

	var (
		date          time.Time
		observedFlux  float32
		adjustedFlux  float32
		ssn           float32
		kpIndex       float32
		apIndex       float32
	)

	row := conn.QueryRow(ctx, query)
	err := row.Scan(&date, &observedFlux, &adjustedFlux, &ssn, &kpIndex, &apIndex)
	if err != nil {
		log.Printf("⚠ Test date %s not found in solar_master view", testDate)
		log.Printf("  This is expected if historical data hasn't been ingested yet")
	} else {
		// Display Trinity validation results
		log.Printf("Test Date: %s", date.Format("2006-01-02"))
		log.Printf("  ├─ Observed Flux (SFI): %.2f sfu", observedFlux)
		log.Printf("  ├─ Adjusted Flux (SFI): %.2f sfu", adjustedFlux)
		log.Printf("  ├─ Sunspot Number (SSN): %.1f", ssn)
		log.Printf("  ├─ Kp Index: %.2f", kpIndex)
		log.Printf("  └─ Ap Index: %.1f", apIndex)

		// Verify Trinity is complete (non-zero values)
		trinityComplete := observedFlux > 0 && ssn > 0 && kpIndex > 0
		if trinityComplete {
			log.Printf("✓ Trinity COMPLETE: All three metrics (SFI, SSN, Kp) are populated")
		} else {
			log.Printf("⚠ Trinity INCOMPLETE:")
			if observedFlux == 0 {
				log.Printf("  ✗ SFI (Solar Flux Index) is ZERO - missing flux data file?")
			}
			if ssn == 0 {
				log.Printf("  ✗ SSN (Sunspot Number) is ZERO - missing SSN data file?")
			}
			if kpIndex == 0 {
				log.Printf("  ✗ Kp (Geomagnetic Index) is ZERO - missing Kp/Ap data file?")
			}
		}
	}

	// Query total record counts by data type
	log.Printf("━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━")
	log.Printf("Record Count Summary (from solar_master view)")
	log.Printf("━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━")

	countQuery := `
		SELECT
			COUNT(*) AS total_records,
			COUNT(CASE WHEN observed_flux > 0 THEN 1 END) AS sfi_records,
			COUNT(CASE WHEN ssn > 0 THEN 1 END) AS ssn_records,
			COUNT(CASE WHEN kp_index > 0 THEN 1 END) AS kp_records,
			COUNT(CASE WHEN observed_flux > 0 AND ssn > 0 AND kp_index > 0 THEN 1 END) AS trinity_complete_records
		FROM wspr.solar_master
	`

	var (
		totalRecords           uint64
		sfiRecords             uint64
		ssnRecords             uint64
		kpRecords              uint64
		trinityCompleteRecords uint64
	)

	row = conn.QueryRow(ctx, countQuery)
	if err := row.Scan(&totalRecords, &sfiRecords, &ssnRecords, &kpRecords, &trinityCompleteRecords); err != nil {
		return fmt.Errorf("query record counts: %w", err)
	}

	log.Printf("  Total unique dates:         %d", totalRecords)
	log.Printf("  Records with SFI data:      %d (%.1f%%)", sfiRecords, float64(sfiRecords)/float64(totalRecords)*100)
	log.Printf("  Records with SSN data:      %d (%.1f%%)", ssnRecords, float64(ssnRecords)/float64(totalRecords)*100)
	log.Printf("  Records with Kp data:       %d (%.1f%%)", kpRecords, float64(kpRecords)/float64(totalRecords)*100)
	log.Printf("  Trinity Complete (all 3):   %d (%.1f%%)", trinityCompleteRecords, float64(trinityCompleteRecords)/float64(totalRecords)*100)
	log.Printf("━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━")

	return nil
}
