package collector

import (
	"context"
	"encoding/json"
	"fmt"
	"log/slog"
	"math"
	"os"
	"path/filepath"
	"strings"
	"time"

	"bufio"
	"runtime"
	"sync"

	"github.com/turbot/tailpipe-plugin-sdk/events"
	"github.com/turbot/tailpipe-plugin-sdk/row_source"
	"github.com/turbot/tailpipe-plugin-sdk/schema"
	"github.com/turbot/tailpipe-plugin-sdk/table"
	"github.com/turbot/tailpipe/internal/config"
	"github.com/turbot/tailpipe/internal/plugin"
)

// doCollectSynthetic initiates synthetic data collection for testing and performance benchmarking.
// This function simulates the data collection process by generating dummy data instead of collecting from real sources.
//
// The function:
// 1. Creates an execution context to track the synthetic collection process
// 2. Builds a synthetic schema based on the number of columns specified in the partition metadata
// 3. Starts a background goroutine to generate and write synthetic data in chunks
// 4. Returns a CollectResponse that mimics what a real plugin would return
//
// This enables testing of the entire data collection pipeline without requiring actual data sources,
// making it useful for performance testing, load testing, and development/debugging scenarios.
//
// Parameters:
//   - ctx: Context for cancellation and timeout handling
//   - fromTime: Start time for the synthetic data (timestamps will be distributed across this range)
//   - toTime: End time for the synthetic data
//   - overwrite: Whether to overwrite existing data (not used in synthetic collection)
//
// Returns:
//   - *plugin.CollectResponse: Response containing execution ID and schema information
//   - error: Any error that occurred during initialization
func (c *Collector) doCollectSynthetic(ctx context.Context, fromTime time.Time, toTime time.Time, overwrite bool) (*plugin.CollectResponse, error) {
	// Create the execution context to track the synthetic collection process
	// This must be created before starting the collection goroutine to ensure proper event handling
	c.execution = &execution{
		id:             "synthetic",                 // Use "synthetic" as the execution ID
		partition:      c.partition.UnqualifiedName, // Full partition name for identification
		table:          c.partition.TableName,       // Table name (always "synthetic" for synthetic partitions)
		plugin:         "synthetic",                 // Plugin name for logging and identification
		state:          ExecutionState_PENDING,      // Initial state before collection starts
		completionChan: make(chan error, 1),         // Channel to signal completion or errors
	}

	// Build the synthetic schema based on the number of columns specified in the partition metadata
	// This creates a table schema with the specified number of columns of various types
	schema := buildsyntheticchema(c.partition.SyntheticMetadata.Columns)

	// Start a background goroutine to perform the actual synthetic data generation
	// This simulates the asynchronous nature of real data collection
	go c.collectSynthetic(ctx, schema, fromTime, toTime)

	// Build a collect response that mimics what a real plugin would return
	// This allows the synthetic collection to integrate seamlessly with the existing collection pipeline
	collectResponse := &plugin.CollectResponse{
		ExecutionId: c.execution.id, // Use the execution ID for tracking
		Schema:      schema,         // The generated synthetic schema
		FromTime: &row_source.ResolvedFromTime{
			Time:   fromTime,    // Start time for the data collection
			Source: "synthetic", // Source identifier for synthetic data
		},
	}

	// Update the execution ID to match the response (in case it was modified)
	c.execution.id = collectResponse.ExecutionId
	return collectResponse, nil
}

// syntheticColumnTypes defines the available column types for synthetic data generation
var syntheticColumnTypes = []struct {
	Name         string
	SQLType      string
	StructFields []*schema.ColumnSchema
}{
	{"string_col", "VARCHAR", nil},
	{"int_col", "INTEGER", nil},
	{"float_col", "DOUBLE", nil},
	{"bool_col", "BOOLEAN", nil},
	{"json_col", "JSON", nil},
	{"timestamp_col", "TIMESTAMP", nil},
	{"array_col", "JSON", nil},
	{"nested_json_col", "JSON", nil},
	{"uuid_col", "VARCHAR", nil},
	{"simple_struct_col", "STRUCT", []*schema.ColumnSchema{
		{
			SourceName:  "id",
			ColumnName:  "id",
			Type:        "INTEGER",
			Description: "Simple struct ID field",
		},
		{
			SourceName:  "name",
			ColumnName:  "name",
			Type:        "VARCHAR",
			Description: "Simple struct name field",
		},
		{
			SourceName:  "active",
			ColumnName:  "active",
			Type:        "BOOLEAN",
			Description: "Simple struct active field",
		},
	}},
	{"nested_struct_col", "STRUCT", []*schema.ColumnSchema{
		{
			SourceName: "metadata",
			ColumnName: "metadata",
			Type:       "STRUCT",
			StructFields: []*schema.ColumnSchema{
				{
					SourceName:  "created_at",
					ColumnName:  "created_at",
					Type:        "VARCHAR",
					Description: "Creation timestamp",
				},
				{
					SourceName:  "version",
					ColumnName:  "version",
					Type:        "VARCHAR",
					Description: "Version string",
				},
			},
			Description: "Metadata information",
		},
		{
			SourceName: "data",
			ColumnName: "data",
			Type:       "STRUCT",
			StructFields: []*schema.ColumnSchema{
				{
					SourceName:  "field1",
					ColumnName:  "field1",
					Type:        "INTEGER",
					Description: "Numeric field 1",
				},
				{
					SourceName:  "field2",
					ColumnName:  "field2",
					Type:        "VARCHAR",
					Description: "String field 2",
				},
				{
					SourceName:  "field3",
					ColumnName:  "field3",
					Type:        "BOOLEAN",
					Description: "Boolean field 3",
				},
			},
			Description: "Data fields",
		},
	}},
	{"complex_struct_col", "STRUCT", []*schema.ColumnSchema{
		{
			SourceName: "user",
			ColumnName: "user",
			Type:       "STRUCT",
			StructFields: []*schema.ColumnSchema{
				{
					SourceName:  "id",
					ColumnName:  "id",
					Type:        "INTEGER",
					Description: "User ID",
				},
				{
					SourceName:  "name",
					ColumnName:  "name",
					Type:        "VARCHAR",
					Description: "User name",
				},
				{
					SourceName: "profile",
					ColumnName: "profile",
					Type:       "STRUCT",
					StructFields: []*schema.ColumnSchema{
						{
							SourceName:  "age",
							ColumnName:  "age",
							Type:        "INTEGER",
							Description: "User age",
						},
						{
							SourceName:  "email",
							ColumnName:  "email",
							Type:        "VARCHAR",
							Description: "User email",
						},
						{
							SourceName:  "verified",
							ColumnName:  "verified",
							Type:        "BOOLEAN",
							Description: "Email verified",
						},
					},
					Description: "User profile information",
				},
			},
			Description: "User information",
		},
		{
			SourceName: "settings",
			ColumnName: "settings",
			Type:       "STRUCT",
			StructFields: []*schema.ColumnSchema{
				{
					SourceName:  "theme",
					ColumnName:  "theme",
					Type:        "VARCHAR",
					Description: "UI theme",
				},
				{
					SourceName:  "notifications",
					ColumnName:  "notifications",
					Type:        "BOOLEAN",
					Description: "Notifications enabled",
				},
			},
			Description: "User settings",
		},
	}},
}

// ConcurrentDataGenerator handles concurrent data generation and marshaling
type ConcurrentDataGenerator struct {
	numWorkers int
	rowChan    chan []byte
	errorChan  chan error
	doneChan   chan bool
}

// NewConcurrentDataGenerator creates a new concurrent data generator
func NewConcurrentDataGenerator(numWorkers int) *ConcurrentDataGenerator {
	return &ConcurrentDataGenerator{
		numWorkers: numWorkers,
		rowChan:    make(chan []byte, numWorkers*100), // Buffer for generated rows
		errorChan:  make(chan error, 1),
		doneChan:   make(chan bool, 1),
	}
}

// generateRowData generates a single row's JSON data
func generateRowData(rowIndex int, partition *config.Partition, tableSchema *schema.TableSchema, fromTime time.Time, timestampInterval time.Duration) ([]byte, error) {
	// Create row map
	rowMap := make(map[string]any, len(tableSchema.Columns))
	timestamp := fromTime.Add(time.Duration(rowIndex) * timestampInterval).Format("2006-01-02 15:04:05")

	// Populate row map (skip tp_index and tp_date)
	for _, column := range tableSchema.Columns {
		if column.ColumnName == "tp_index" || column.ColumnName == "tp_date" {
			continue
		}

		switch column.ColumnName {
		case "tp_timestamp":
			rowMap[column.ColumnName] = timestamp
		case "tp_partition":
			rowMap[column.ColumnName] = partition.ShortName
		case "tp_table":
			rowMap[column.ColumnName] = partition.TableName
		default:
			// Generate synthetic data for other columns
			rowMap[column.ColumnName] = generateSyntheticValue(column, rowIndex)
		}
	}

	// Marshal to JSON
	data, err := json.Marshal(rowMap)
	if err != nil {
		return nil, fmt.Errorf("failed to marshal row %d: %w", rowIndex, err)
	}

	// Add newline
	data = append(data, '\n')
	return data, nil
}

// worker generates data for a range of rows
func (cdg *ConcurrentDataGenerator) worker(startRow, endRow int, partition *config.Partition, tableSchema *schema.TableSchema, fromTime time.Time, timestampInterval time.Duration) {
	for rowIndex := startRow; rowIndex < endRow; rowIndex++ {
		data, err := generateRowData(rowIndex, partition, tableSchema, fromTime, timestampInterval)
		if err != nil {
			select {
			case cdg.errorChan <- err:
			default:
			}
			return
		}

		select {
		case cdg.rowChan <- data:
		case <-cdg.doneChan:
			return
		}
	}
}

// writeOptimizedChunkToJSONLConcurrent uses multiple goroutines for data generation
func writeOptimizedChunkToJSONLConcurrent(filepath string, tableSchema *schema.TableSchema, rows int, startRowIndex int, partition *config.Partition, fromTime time.Time, timestampInterval time.Duration) error {
	file, err := os.Create(filepath)
	if err != nil {
		return fmt.Errorf("failed to create file %s: %w", filepath, err)
	}
	defer file.Close()

	// Use buffered writer for better I/O performance
	bufWriter := bufio.NewWriter(file)
	defer bufWriter.Flush()

	// Determine number of workers (use CPU cores, but cap at reasonable number)
	numWorkers := runtime.NumCPU()
	if numWorkers > 8 {
		numWorkers = 8 // Cap at 8 to avoid too much overhead
	}
	if numWorkers > rows {
		numWorkers = rows // Don't create more workers than rows
	}

	// Create concurrent data generator
	cdg := NewConcurrentDataGenerator(numWorkers)

	// Calculate rows per worker
	rowsPerWorker := rows / numWorkers
	remainder := rows % numWorkers

	// Start workers
	var wg sync.WaitGroup
	startRow := startRowIndex
	for i := 0; i < numWorkers; i++ {
		endRow := startRow + rowsPerWorker
		if i < remainder {
			endRow++ // Distribute remainder rows
		}

		wg.Add(1)
		go func(start, end int) {
			defer wg.Done()
			cdg.worker(start, end, partition, tableSchema, fromTime, timestampInterval)
		}(startRow, endRow)

		startRow = endRow
	}

	// Start a goroutine to close the row channel when all workers are done
	go func() {
		wg.Wait()
		close(cdg.rowChan)
	}()

	// Write rows from channel to file
	rowsWritten := 0
	for data := range cdg.rowChan {
		if _, err := bufWriter.Write(data); err != nil {
			close(cdg.doneChan) // Signal workers to stop
			return fmt.Errorf("failed to write row %d: %w", rowsWritten, err)
		}
		rowsWritten++
	}

	// Check for errors
	select {
	case err := <-cdg.errorChan:
		return fmt.Errorf("worker error: %w", err)
	default:
	}

	if rowsWritten != rows {
		return fmt.Errorf("expected %d rows, but wrote %d", rows, rowsWritten)
	}

	return nil
}

func buildsyntheticchema(columns int) *schema.TableSchema {
	// Create a basic schema with the required number of columns
	// Start with required tp_ fields
	s := &schema.TableSchema{
		Columns: make([]*schema.ColumnSchema, 0, columns+5), // +5 for tp_ fields (including tp_index and tp_date)
	}

	// Add required tp_ fields first
	tpFields := []struct {
		name        string
		columnType  string
		description string
	}{
		{"tp_timestamp", "TIMESTAMP", "Timestamp when the record was collected"},
		{"tp_partition", "VARCHAR", "Partition identifier"},
		{"tp_table", "VARCHAR", "Table identifier"},
		{"tp_index", "VARCHAR", "Index identifier"},
		{"tp_date", "VARCHAR", "Date identifier"},
	}

	for _, tpField := range tpFields {
		column := &schema.ColumnSchema{
			SourceName:   tpField.name,
			ColumnName:   tpField.name,
			Type:         tpField.columnType,
			StructFields: nil,
			Description:  tpField.description,
			Required:     true, // tp_ fields are always required
			NullIf:       "",
			Transform:    "",
		}
		s.Columns = append(s.Columns, column)
	}

	// Add the specified number of synthetic columns by cycling through the column types
	for i := 0; i < columns; i++ {
		// Cycle through the column types
		typeIndex := i % len(syntheticColumnTypes)
		baseType := syntheticColumnTypes[typeIndex]

		// Create a unique column name
		columnName := fmt.Sprintf("%s_%d", baseType.Name, i)

		column := &schema.ColumnSchema{
			SourceName:   columnName,
			ColumnName:   columnName,
			Type:         baseType.SQLType,
			StructFields: baseType.StructFields,
			Description:  fmt.Sprintf("Synthetic column of type %s", baseType.SQLType),
			Required:     false,
			NullIf:       "",
			Transform:    "",
		}

		s.Columns = append(s.Columns, column)
	}

	return s
}

// collectSynthetic generates synthetic data in chunks and writes it to JSONL files.
// This function runs in a background goroutine and simulates the data collection process
// by generating dummy data according to the synthetic partition specifications.
//
// The function:
// 1. Notifies that collection has started
// 2. Calculates timestamp intervals to distribute timestamps across the time range
// 3. Generates data in chunks according to the specified chunk size
// 4. Writes each chunk to a JSONL file using optimized concurrent writing
// 5. Respects the delivery interval to simulate real-time data flow
// 6. Sends progress events (chunk and status) to maintain the collection UI
// 7. Handles cancellation and error conditions gracefully
// 8. Notifies completion when all data has been generated
//
// Parameters:
//   - ctx: Context for cancellation and timeout handling
//   - tableSchema: The schema defining the structure of the synthetic data
//   - fromTime: Start time for timestamp generation
//   - toTime: End time for timestamp generation
func (c *Collector) collectSynthetic(ctx context.Context, tableSchema *schema.TableSchema, fromTime time.Time, toTime time.Time) {
	metadata := c.partition.SyntheticMetadata

	// Set the execution state to started to indicate collection is in progress
	c.execution.state = ExecutionState_STARTED

	// Notify that collection has started - this triggers the collection UI to show progress
	if err := c.Notify(ctx, &events.Started{ExecutionId: c.execution.id}); err != nil {
		slog.Error("failed to notify started event", "error", err)
		c.execution.completionChan <- fmt.Errorf("failed to notify started event: %w", err)
		return
	}

	var chunkIdx int32 = 0           // Track the current chunk number
	var totalRowsProcessed int64 = 0 // Track total rows processed for progress reporting

	// Calculate timestamp interval to distribute timestamps evenly across the time range
	// This ensures synthetic data has realistic timestamp progression
	var timestampInterval time.Duration
	if metadata.Rows > 1 {
		// Distribute timestamps evenly between fromTime and toTime
		timestampInterval = toTime.Sub(fromTime) / time.Duration(metadata.Rows-1)
	} else {
		// Single row case - no interval needed
		timestampInterval = 0
	}

	// Generate data in chunks according to the specified chunk size
	// This allows for memory-efficient processing of large datasets
	for rowCount := 0; rowCount < metadata.Rows; rowCount += metadata.ChunkSize {
		t := time.Now() // Track chunk processing time for delivery interval calculation

		// Check if context is cancelled - allows for graceful shutdown
		select {
		case <-ctx.Done():
			c.execution.completionChan <- ctx.Err()
			return
		default:
		}

		// Calculate the number of rows for this chunk (may be less than chunk size for the last chunk)
		rows := int(math.Min(float64(metadata.Rows-rowCount), float64(metadata.ChunkSize)))

		// Generate filename for this chunk's JSONL file
		filename := table.ExecutionIdToJsonlFileName(c.execution.id, chunkIdx)
		filepath := filepath.Join(c.sourcePath, filename)

		// Write the chunk to JSONL file using optimized concurrent approach
		// This generates synthetic data and writes it efficiently to disk
		if err := writeOptimizedChunkToJSONLConcurrent(filepath, tableSchema, rows, rowCount, c.partition, fromTime, timestampInterval); err != nil {
			c.execution.completionChan <- fmt.Errorf("error writing chunk to JSONL file: %w", err)
			return
		}

		dur := time.Since(t) // Calculate how long this chunk took to process

		// Respect the delivery interval to simulate real-time data flow
		// If processing was faster than the interval, wait for the remaining time
		if metadata.DeliveryIntervalMs > 0 && dur < time.Duration(metadata.DeliveryIntervalMs)*time.Millisecond {
			slog.Debug("Waiting for delivery interval", "duration", dur, "expected", time.Duration(metadata.DeliveryIntervalMs)*time.Millisecond)
			select {
			case <-time.After(time.Duration(metadata.DeliveryIntervalMs)*time.Millisecond - dur):
				// Wait for the remaining time
			case <-ctx.Done():
				// Context was cancelled during wait
				c.execution.completionChan <- ctx.Err()
				return
			}
		}

		// Send chunk event to notify that a chunk has been completed
		// This updates the collection UI and allows other components to process the chunk
		chunkEvent := &events.Chunk{ExecutionId: c.execution.id, ChunkNumber: chunkIdx}
		if err := c.Notify(ctx, chunkEvent); err != nil {
			slog.Error("failed to notify chunk event", "error", err)
			c.execution.completionChan <- fmt.Errorf("failed to notify chunk event: %w", err)
			return
		}

		// Update total rows processed and send status event
		// This provides progress information to the collection UI
		totalRowsProcessed += int64(rows)
		statusEvent := &events.Status{ExecutionId: c.execution.id, RowsReceived: totalRowsProcessed, RowsEnriched: totalRowsProcessed}
		if err := c.Notify(ctx, statusEvent); err != nil {
			slog.Error("failed to notify status event", "error", err)
			c.execution.completionChan <- fmt.Errorf("failed to notify status event: %w", err)
			return
		}

		chunkIdx++ // Move to next chunk
	}

	// Send completion event to indicate all data has been generated
	// This triggers final processing and updates the collection UI
	if err := c.Notify(ctx, events.NewCompletedEvent(c.execution.id, int64(metadata.Rows), chunkIdx, nil)); err != nil {
		slog.Error("failed to notify completed event", "error", err)
		c.execution.completionChan <- fmt.Errorf("failed to notify completed event: %w", err)
		return
	}

	// Signal completion by sending nil to the completion channel
	// This allows the main collection process to know that synthetic data generation is complete
	c.execution.completionChan <- nil
}

func generateSyntheticValue(column *schema.ColumnSchema, rowIndex int) any {
	// Use the column's Type field directly instead of fuzzy matching on name
	columnType := column.Type

	// Generate value based on exact type match (case-insensitive)
	switch strings.ToUpper(columnType) {
	case "VARCHAR":
		return fmt.Sprintf("%s_val%d", column.ColumnName, rowIndex%100000)
	case "INTEGER":
		return (rowIndex % 100000) + 1
	case "DOUBLE":
		return float64(rowIndex%100000) * 0.1
	case "BOOLEAN":
		return rowIndex%2 == 0
	case "JSON":
		return generateJSONValue(column, rowIndex)
	case "TIMESTAMP":
		return time.Now().AddDate(0, 0, -rowIndex%30).Format("2006-01-02 15:04:05")
	default:
		// Handle struct types and complex types
		if strings.Contains(strings.ToUpper(columnType), "STRUCT") {
			return generateStructValue(column, rowIndex)
		}
		// For any other unrecognized type, throw an error
		panic(fmt.Sprintf("Unsupported column type '%s' for column '%s'", columnType, column.ColumnName))
	}
}

func generateJSONValue(column *schema.ColumnSchema, rowIndex int) any {
	// Generate different JSON structures based on column name
	if strings.Contains(column.ColumnName, "nested_json") {
		return map[string]any{
			"metadata": map[string]any{
				"created_at": time.Now().AddDate(0, 0, -rowIndex%30).Format("2006-01-02"),
				"version":    fmt.Sprintf("v%d.%d", rowIndex%10, rowIndex%5),
			},
			"data": map[string]any{
				"field1": rowIndex % 100000,
				"field2": fmt.Sprintf("field_%d", rowIndex%100000),
				"field3": rowIndex%2 == 0,
			},
		}
	} else if strings.Contains(column.ColumnName, "array") {
		return []any{
			fmt.Sprintf("item_%d", rowIndex%100000),
			rowIndex % 100000,
			rowIndex%2 == 0,
			float64(rowIndex%100000) * 0.1,
		}
	} else {
		// Default JSON object
		return map[string]any{
			"id":    rowIndex % 100000,
			"name":  fmt.Sprintf("item_%d", rowIndex%100000),
			"value": (rowIndex % 100000) + 1,
			"tags":  []string{"tag1", "tag2", "tag3"},
		}
	}
}

func generateStructValue(column *schema.ColumnSchema, rowIndex int) any {
	if column.StructFields == nil {
		return map[string]any{
			"id":   rowIndex % 100000,
			"name": fmt.Sprintf("struct_%d", rowIndex%100000),
		}
	}

	result := make(map[string]any)
	for _, field := range column.StructFields {
		if field.StructFields != nil {
			// Nested struct
			result[field.ColumnName] = generateStructValue(field, rowIndex)
		} else {
			// Simple field
			result[field.ColumnName] = generateSyntheticValue(field, rowIndex)
		}
	}
	return result
}
