streamsql.go

/*
 * Copyright 2025 The RuleGo Authors.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package streamsql

import (
	"fmt"
	"sync/atomic"

	"github.com/rulego/streamsql/aggregator"
	"github.com/rulego/streamsql/logger"
	"github.com/rulego/streamsql/metrics"
	"github.com/rulego/streamsql/rsql"
	"github.com/rulego/streamsql/schema"
	"github.com/rulego/streamsql/stream"
	"github.com/rulego/streamsql/types"
	"github.com/rulego/streamsql/utils/table"
)

// Streamsql is the main interface for the StreamSQL streaming engine.
// It encapsulates core functionality including SQL parsing, stream processing, and window management.
//
// Usage example:
//
//	ssql := streamsql.New()
//	err := ssql.Execute("SELECT AVG(temperature) FROM stream GROUP BY TumblingWindow('5s')")
//	ssql.Emit(map[string]interface{}{"temperature": 25.5})
type Streamsql struct {
	stream *stream.Stream

	// Performance configuration mode
	performanceMode string // "default", "high_performance", "low_latency", "custom"
	customConfig    *types.PerformanceConfig

	// Save original SELECT field order to maintain field order for table output
	fieldOrder []string

	// Flag to track if Execute has been called
	executed int32

	// Per-instance logger (set via WithLogger, defaults to logger.GetDefault());
	// immutable after New, injected into the stream pipeline at Execute.
	log logger.Logger

	// Opt-in input validation. schemaValidator is non-nil only when WithSchema
	// is set; nil means Emit/EmitSync skip validation entirely (zero overhead).
	schemaValidator *schema.Schema
	schemaDropped   int64
}

// New creates a new StreamSQL instance.
// Supports configuration through optional Option parameters.
//
// Parameters:
//   - options: Variable configuration options for customizing StreamSQL behavior
//
// Returns:
//   - *Streamsql: Newly created StreamSQL instance
//
// Examples:
//
//	// Create default instance
//	ssql := streamsql.New()
//
//	// Create high performance instance
//	ssql := streamsql.New(streamsql.WithHighPerformance())
func New(options ...Option) *Streamsql {
	s := &Streamsql{
		performanceMode: "default", // Default to standard performance configuration
		log:             logger.GetDefault(),
	}

	// Apply all configuration options
	for _, option := range options {
		option(s)
	}

	return s
}

// Execute parses and executes SQL queries, creating corresponding stream processing pipelines.
// This is the core method of StreamSQL, responsible for converting SQL into actual stream processing logic.
//
// Supported SQL syntax:
//   - SELECT clause: Select fields and aggregate functions
//   - FROM clause: Specify data source (usually 'stream')
//   - WHERE clause: Data filtering conditions
//   - GROUP BY clause: Grouping fields and window functions
//   - HAVING clause: Aggregate result filtering
//   - LIMIT clause: Limit result count
//   - DISTINCT: Result deduplication
//
// Window functions:
//   - TumblingWindow('5s'): Tumbling window
//   - SlidingWindow('30s', '10s'): Sliding window
//   - CountingWindow(100): Counting window
//   - SessionWindow('5m'): Session window
//
// Parameters:
//   - sql: SQL query statement to execute
//
// Returns:
//   - error: Returns error if SQL parsing or execution fails
//
// Examples:
//
//	// Basic aggregation query
//	err := ssql.Execute("SELECT deviceId, AVG(temperature) FROM stream GROUP BY deviceId, TumblingWindow('5s')")
//
//	// Query with filtering conditions
//	err := ssql.Execute("SELECT * FROM stream WHERE temperature > 30")
//
//	// Complex window aggregation
//	err := ssql.Execute(`
//	    SELECT deviceId,
//	           AVG(temperature) as avg_temp,
//	           MAX(humidity) as max_humidity
//	    FROM stream
//	    WHERE deviceId != 'test'
//	    GROUP BY deviceId, SlidingWindow('1m', '30s')
//	    HAVING avg_temp > 25
//	    LIMIT 100
//	`)
func (s *Streamsql) Execute(sql string) error {
	// Try to acquire execution lock using CAS operation
	if !atomic.CompareAndSwapInt32(&s.executed, 0, 1) {
		return fmt.Errorf("Execute() has already been called, create a new Streamsql instance for different queries")
	}

	// Parse SQL statement
	config, condition, err := rsql.Parse(sql)
	if err != nil {
		// Reset executed flag on error
		atomic.StoreInt32(&s.executed, 0)
		return fmt.Errorf("SQL parsing failed: %w", err)
	}

	// Validate aggregation functions up front: reject per-row window functions
	// (row_number/lead) the per-group model cannot evaluate, so they fail here
	// with a clear error instead of crashing on the data path or returning nil.
	for _, aggType := range config.SelectFields {
		if err := aggregator.ValidateAggregateType(aggType); err != nil {
			atomic.StoreInt32(&s.executed, 0)
			return fmt.Errorf("unsupported aggregation: %w", err)
		}
	}

	// Get field order information from parsing result
	s.fieldOrder = config.FieldOrder

	// Inject the per-instance logger into the stream pipeline.
	config.Logger = s.log

	// Create stream processor based on performance mode
	var streamInstance *stream.Stream

	switch s.performanceMode {
	case "high_performance":
		streamInstance, err = stream.NewStreamWithHighPerformance(*config)
	case "low_latency":
		streamInstance, err = stream.NewStreamWithLowLatency(*config)
	case "custom":
		if s.customConfig != nil {
			streamInstance, err = stream.NewStreamWithCustomPerformance(*config, *s.customConfig)
		} else {
			streamInstance, err = stream.NewStream(*config)
		}
	default: // "default"
		streamInstance, err = stream.NewStream(*config)
	}

	if err != nil {
		// Reset executed flag on error
		atomic.StoreInt32(&s.executed, 0)
		return fmt.Errorf("failed to create stream processor: %w", err)
	}

	s.stream = streamInstance

	// Register filter condition
	if err = s.stream.RegisterFilter(condition); err != nil {
		// Reset executed flag on error
		atomic.StoreInt32(&s.executed, 0)
		return fmt.Errorf("failed to register filter condition: %w", err)
	}

	// Start stream processing
	s.stream.Start()

	return nil
}

// Emit adds data to the stream processing pipeline.
// Accepts type-safe map[string]interface{} format data.
//
// Parameters:
//   - data: Data to add, must be map[string]interface{} type
//
// Examples:
//
//	// Add device data
//	ssql.Emit(map[string]interface{}{
//	    "deviceId": "sensor001",
//	    "temperature": 25.5,
//	    "humidity": 60.0,
//	    "timestamp": time.Now(),
//	})
//
//	// Add user behavior data
//	ssql.Emit(map[string]interface{}{
//	    "userId": "user123",
//	    "action": "click",
//	    "page": "/home",
//	})
func (s *Streamsql) Emit(data map[string]interface{}) {
	if s.stream == nil {
		return
	}
	if s.schemaValidator != nil {
		if err := s.schemaValidator.Validate(data); err != nil {
			n := atomic.AddInt64(&s.schemaDropped, 1)
			if n == 1 || n%1000 == 0 {
				s.log.Warn("schema validation failed, dropping row (total %d): %v", n, err)
			}
			return
		}
	}
	s.stream.Emit(data)
}

// EmitSync processes data synchronously, returning results immediately.
// Only applicable for non-aggregation queries, aggregation queries will return an error.
// Accepts type-safe map[string]interface{} format data.
//
// Parameters:
//   - data: Data to process, must be map[string]interface{} type
//
// Returns:
//   - map[string]interface{}: Processed result data, returns nil if filter conditions don't match
//   - error: Processing error
//
// Examples:
//
//	result, err := ssql.EmitSync(map[string]interface{}{
//	    "deviceId": "sensor001",
//	    "temperature": 25.5,
//	})
//	if err != nil {
//	    log.Printf("processing error: %v", err)
//	} else if result != nil {
//	    // Use processed result immediately (result is map[string]interface{} type)
//	    fmt.Printf("Processing result: %v\n", result)
//	}
func (s *Streamsql) EmitSync(data map[string]interface{}) (map[string]interface{}, error) {
	if s.stream == nil {
		return nil, fmt.Errorf("stream not initialized")
	}

	// Check if it's a non-aggregation query
	if s.stream.IsAggregationQuery() {
		return nil, fmt.Errorf("synchronous mode only supports non-aggregation queries, use Emit() method for aggregation queries")
	}

	if s.schemaValidator != nil {
		if err := s.schemaValidator.Validate(data); err != nil {
			atomic.AddInt64(&s.schemaDropped, 1)
			return nil, fmt.Errorf("schema validation failed: %w", err)
		}
	}
	return s.stream.ProcessSync(data)
}

// SchemaDropped returns the count of rows dropped by schema validation.
func (s *Streamsql) SchemaDropped() int64 {
	return atomic.LoadInt64(&s.schemaDropped)
}

// IsAggregationQuery checks if the current query is an aggregation query
func (s *Streamsql) IsAggregationQuery() bool {
	if s.stream == nil {
		return false
	}
	return s.stream.IsAggregationQuery()
}

// Stream returns the underlying stream processor instance.
// Provides access to lower-level stream processing functionality.
//
// Returns:
//   - *stream.Stream: Underlying stream processor instance, returns nil if SQL not executed
//
// Common use cases:
//   - Add result processing callbacks
//   - Get result channel
//   - Manually control stream processing lifecycle
//
// Examples:
//
//	// Add result processing callback
//	ssql.Stream().AddSink(func(results []map[string]interface{}) {
//	    fmt.Printf("Processing results: %v\n", results)
//	})
//
//	// Get result channel
//	resultChan := ssql.Stream().GetResultsChan()
//	go func() {
//	    for result := range resultChan {
//	        // Process result
//	    }
//	}()
func (s *Streamsql) Stream() *stream.Stream {
	return s.stream
}

// TriggerWindow manually triggers the current window to emit immediately,
// bypassing its normal time/count trigger. Intended for tests that need a
// window to fire deterministically, and as an explicit flush hook.
func (s *Streamsql) TriggerWindow() {
	if s.stream != nil && s.stream.Window != nil {
		s.stream.Window.Trigger()
	}
}

// GetStats returns stream processing statistics
func (s *Streamsql) GetStats() map[string]int64 {
	if s.stream != nil {
		return s.stream.GetStats()
	}
	return make(map[string]int64)
}

// GetDetailedStats returns detailed performance statistics
func (s *Streamsql) GetDetailedStats() map[string]interface{} {
	if s.stream != nil {
		return s.stream.GetDetailedStats()
	}
	return make(map[string]interface{})
}

// Metrics returns the underlying metrics registry, or nil before Execute.
// Callers can inspect named counters/gauges/histograms directly via the registry.
func (s *Streamsql) Metrics() *metrics.Registry {
	if s.stream != nil {
		return s.stream.MetricsRegistry()
	}
	return nil
}

// Stop stops the stream processor and releases related resources.
// After calling this method, the stream processor will stop receiving and processing new data.
//
// Recommended to call this method for cleanup before application exit:
//
//	defer ssql.Stop()
//
// Note: StreamSQL instance cannot be restarted after stopping, create a new instance.
func (s *Streamsql) Stop() {
	if s.stream != nil {
		s.stream.Stop()
	}
}

// AddSink directly adds result processing callback functions.
// Convenience wrapper for Stream().AddSink() for cleaner API calls.
//
// Parameters:
//   - sink: Result processing function, receives []map[string]interface{} type result data
//
// Examples:
//
//	// Directly add result processing
//	ssql.AddSink(func(results []map[string]interface{}) {
//	    fmt.Printf("Processing results: %v\n", results)
//	})
//
//	// Add multiple processors
//	ssql.AddSink(func(results []map[string]interface{}) {
//	    // Save to database
//	    saveToDatabase(results)
//	})
//	ssql.AddSink(func(results []map[string]interface{}) {
//	    // Send to message queue
//	    sendToQueue(results)
//	})
func (s *Streamsql) AddSink(sink func([]map[string]interface{})) {
	if s.stream != nil {
		s.stream.AddSink(sink)
	}
}

// AddSyncSink directly adds synchronous result processing callback functions.
// Convenience wrapper for Stream().AddSyncSink() for cleaner API calls.
//
// Parameters:
//   - sink: Result processing function, receives []map[string]interface{} type result data
//
// Note: Sync sinks are executed sequentially in the result processing goroutine.
// Use this when order of execution matters.
func (s *Streamsql) AddSyncSink(sink func([]map[string]interface{})) {
	if s.stream != nil {
		s.stream.AddSyncSink(sink)
	}
}

// PrintTable prints results to console in table format, similar to database output.
// Displays column names first, then data rows.
//
// Supported data formats:
//   - []map[string]interface{}: Multiple rows
//   - map[string]interface{}: Single row
//   - Other types: Direct print
//
// Example:
//
//	// Print results in table format
//	ssql.PrintTable()
//
//	// Output format:
//	// +--------+----------+
//	// | device | max_temp |
//	// +--------+----------+
//	// | aa     | 30.0     |
//	// | bb     | 22.0     |
//	// +--------+----------+
func (s *Streamsql) PrintTable() {
	s.AddSink(func(results []map[string]interface{}) {
		s.printTableFormat(results)
	})
}

// printTableFormat formats and prints table data
// Parameters:
//   - results: Result data of type []map[string]interface{}
func (s *Streamsql) printTableFormat(results []map[string]interface{}) {
	table.FormatTableData(results, s.fieldOrder)
}

// ToChannel returns result channel for asynchronous result retrieval.
// Provides non-blocking access to stream processing results.
//
// Returns:
//   - <-chan interface{}: Read-only result channel, returns nil if SQL not executed
//
// Example:
//
//	// Get result channel
//	resultChan := ssql.ToChannel()
//	if resultChan != nil {
//	    go func() {
//	        for result := range resultChan {
//	            fmt.Printf("Async result: %v\n", result)
//	        }
//	    }()
//	}

// ToChannel converts query results to channel output
// Returns a read-only channel for receiving query results
//
// Notes:
//   - Consumer must continuously read from channel to prevent stream processing blocking
//   - Channel transmits batch result data
func (s *Streamsql) ToChannel() <-chan []map[string]interface{} {
	if s.stream != nil {
		return s.stream.GetResultsChan()
	}
	return nil
}

// RegisterTable registers an in-memory metadata table for stream-table JOIN.
//
// The index key is auto-derived from the JOIN ON clause's table-side field(s)
// when keyFields is omitted, so callers do not redeclare what ON already states
// (single or composite key both auto-derived). Pass keyFields explicitly to
// override. Returns the source so callers can Upsert/Delete rows incrementally.
//
// Must be called after Execute. Examples:
//
//	// Auto-derive key from ON (recommended): ON deviceId = m.deviceId
//	ssql.RegisterTable("meta", rows)
//	// Explicit composite key:
//	ssql.RegisterTable("meta", rows, "deviceId", "tenant")
func (s *Streamsql) RegisterTable(name string, rows []map[string]interface{}, keyFields ...string) (*stream.MemoryTableSource, error) {
	if s.stream == nil {
		return nil, fmt.Errorf("Execute must be called before RegisterTable")
	}
	if len(keyFields) == 0 {
		derived, err := s.stream.JoinKeyFields(name)
		if err != nil {
			return nil, err
		}
		keyFields = derived
	}
	return s.stream.RegisterMemoryTable(name, keyFields, rows)
}

// RegisterTableSource registers a custom table source (file/DB/Redis/HTTP). The
// implementation owns data loading, refresh, and cleanup; Lookup must be
// concurrency-safe. Must be called after Execute.
func (s *Streamsql) RegisterTableSource(src stream.TableSource) error {
	if s.stream == nil {
		return fmt.Errorf("Execute must be called before RegisterTableSource")
	}
	return s.stream.RegisterTableSource(src)
}

// UpsertTable adds or replaces a row in a previously registered in-memory table.
// Only affects rows emitted after the call (tables are snapshots).
func (s *Streamsql) UpsertTable(name string, row map[string]interface{}) error {
	if s.stream == nil {
		return fmt.Errorf("Execute must be called before UpsertTable")
	}
	return s.stream.UpsertTableRow(name, row)
}