executor.go

package batchflow

import (
	"context"
	"database/sql"
	"errors"
	"fmt"
	"strconv"
	"sync"
	"time"

	"github.com/redis/go-redis/v9"
)

// BatchExecutor 批量执行器接口 - 所有数据库驱动的统一入口
type BatchExecutor interface {
	// ExecuteBatch 执行批量操作
	ExecuteBatch(ctx context.Context, schema SchemaInterface, data []map[string]any) error
}

/*
Metrics 相关接口设计说明

- BatchExecutor：仅负责“执行”职责，保持最小接口。
- MetricsCapable[T]（泛型）：提供“读 + 写”的度量能力，方法返回自类型 T，便于在具体类型或已实例化接口上进行链式配置。
  注意：泛型接口在运行时类型断言时必须使用具体的类型实参（如 MetricsCapable[*ThrottledBatchExecutor]）。
- 兼容性与运行时探测：
  在 BatchFlow 等仅持有 BatchExecutor 的位置，无法统一断言到 MetricsCapable[T]，
  因此使用一个非泛型只读探测接口（即仅调用 MetricsReporter()）来判断是否已有 Reporter；
  若返回 nil，则在本地使用 Noop 兜底，不强制写回（写回需要具体类型 T）。

这允许：
- 在需要链式的调用点：以具体类型或已实例化能力接口使用 Set/With 风格链式；
- 在框架内部通用路径：通过只读探测保证安全、零开销的观测兜底。
*/
// MetricsCapable 扩展接口：支持性能监控报告器（自类型泛型）
type MetricsCapable[T any] interface {
	// WithMetricsReporter 设置性能监控报告器（返回实现者类型以支持链式）
	WithMetricsReporter(MetricsReporter) T
	// MetricsReporter 返回当前性能监控报告器（可能为 nil）
	MetricsReporter() MetricsReporter
}

type ConcurrencyCapable[T any] interface {
	WithConcurrencyLimit(int) T
}

// ThrottledBatchExecutor 通用批量执行器
// 架构：ThrottledBatchExecutor -> BatchProcessor -> backend driver/client
//
// 设计优势：
// - 代码复用：SQL、Redis、自定义后端共享执行、限流、重试和观测逻辑
// - 职责分离：执行控制与具体处理逻辑分离
// - 易于扩展：新增后端只需实现 BatchProcessor，可选实现 OperationPreviewer
type ThrottledBatchExecutor struct {
	processor       BatchProcessor  // 具体的批量处理逻辑
	metricsReporter MetricsReporter // 性能指标报告器
	observer        Observer
	coalescer       Coalescer
	semaphore       chan struct{} // 可选信号量，用于限制 ExecuteBatch 并发

	// 重试配置（默认关闭）
	retryEnabled     bool
	retryMaxAttempts int
	retryBackoffBase time.Duration
	retryMaxBackoff  time.Duration
	retryClassifier  func(error) (retryable bool, reason string)
}

var _ MetricsCapable[*ThrottledBatchExecutor] = (*ThrottledBatchExecutor)(nil)

var _ ConcurrencyCapable[*ThrottledBatchExecutor] = (*ThrottledBatchExecutor)(nil)

var _ BatchExecutor = (*ThrottledBatchExecutor)(nil)

// NewThrottledBatchExecutor 创建通用执行器（使用自定义BatchProcessor）
func NewThrottledBatchExecutor(processor BatchProcessor) *ThrottledBatchExecutor {
	return &ThrottledBatchExecutor{
		processor: processor,
	}
}

// NewThrottledBatchExecutorWithDriver 创建SQL数据库执行器（推荐方式）
// 内部使用 SQLBatchProcessor + SQLDriver 组合
func NewSQLThrottledBatchExecutorWithDriver(db *sql.DB, driver SQLDriver) *ThrottledBatchExecutor {
	return NewThrottledBatchExecutor(NewSQLBatchProcessor(db, driver))
}

func NewRedisThrottledBatchExecutor(client *redis.Client) *ThrottledBatchExecutor {
	return NewThrottledBatchExecutor(NewRedisBatchProcessor(client, DefaultRedisPipelineDriver))
}

func NewRedisThrottledBatchExecutorWithDriver(client *redis.Client, driver RedisDriver) *ThrottledBatchExecutor {
	return NewThrottledBatchExecutor(NewRedisBatchProcessor(client, driver))
}

// RetryConfig 可选重试配置（零值关闭）
type RetryConfig struct {
	Enabled     bool
	MaxAttempts int           // 总尝试次数（含首轮），建议 2~3
	BackoffBase time.Duration // 退避基值（指数退避起点）
	MaxBackoff  time.Duration // 最大退避时长（上限）
	// 自定义错误分类（可选）；返回是否可重试与原因标签
	Classifier func(error) (retryable bool, reason string)
}

// WithRetryConfig 启用/配置重试（仅对 ThrottledBatchExecutor 可用）
func (e *ThrottledBatchExecutor) WithRetryConfig(cfg RetryConfig) *ThrottledBatchExecutor {
	if cfg.MaxAttempts <= 0 {
		cfg.MaxAttempts = 1
	}
	if cfg.BackoffBase <= 0 {
		cfg.BackoffBase = 20 * time.Millisecond
	}
	if cfg.MaxBackoff <= 0 {
		cfg.MaxBackoff = 2 * time.Second
	}
	e.retryEnabled = cfg.Enabled
	e.retryMaxAttempts = cfg.MaxAttempts
	e.retryBackoffBase = cfg.BackoffBase
	e.retryMaxBackoff = cfg.MaxBackoff
	if cfg.Classifier != nil {
		e.retryClassifier = cfg.Classifier
	} else {
		e.retryClassifier = defaultRetryClassifier
	}
	return e
}

/*
默认重试分类器策略说明：
- 对调用方外层 ctx 的取消/超时（context.Canceled/context.DeadlineExceeded）判为不可重试（final:context）。
- 可通过 RetryConfig.Classifier 自定义策略，例如将“处理器内部超时（由处理器用 WithTimeoutCause 附带 cause）”判为可重试。
- 建议对可重试错误使用指数退避与抖动，避免热点重试风暴。
*/
func defaultRetryClassifier(err error) (bool, string) {
	return ClassifyError(err)
}

// ExecuteBatch 执行批量操作
func (e *ThrottledBatchExecutor) ExecuteBatch(ctx context.Context, schema SchemaInterface, data []map[string]any) error {
	if len(data) == 0 {
		return nil
	}
	if e.coalescer != nil {
		result, err := e.coalescer.Coalesce(ctx, schema, data)
		if err != nil {
			return batchErrorFromError(BatchStageValidate, OperationPreview{
				Backend:    BackendCustom,
				Operation:  OperationCustom,
				Schema:     schema.Name(),
				InputItems: len(data),
			}, len(data), err)
		}
		data = result.Batch
		if len(data) == 0 {
			return nil
		}
	}

	// 可选并发限流：当设置了信号量时，进入前需占用一个令牌
	if e.semaphore != nil {
		select {
		case e.semaphore <- struct{}{}:
			defer func() { <-e.semaphore }()
		case <-ctx.Done():
			return ctx.Err()
		}
	}

	startTime := time.Now()
	status := "success"
	// 在途批次 +1（整个批次生命周期内有效）
	if e.metricsReporter != nil {
		e.metricsReporter.IncInflight()
		defer e.metricsReporter.DecInflight()
	}

	var err error
	attempts := 1
	if e.retryEnabled && e.retryMaxAttempts > 1 {
		attempts = e.retryMaxAttempts
	}

RETRY:
	for attempt := 1; attempt <= attempts; attempt++ {
		result := e.executeAttempt(ctx, schema, data, attempt)
		err = result.err
		if err == nil {
			status = "success"
			break
		}

		shouldRetry, retryErr := e.handleRetry(ctx, schema, data, result, attempt, attempts, startTime)
		if retryErr != nil {
			status = "fail"
			err = retryErr
			break RETRY
		}
		if !shouldRetry {
			status = "fail"
			break
		}
	}

	if e.metricsReporter != nil {
		e.metricsReporter.ObserveExecuteDuration(schema.Name(), len(data), time.Since(startTime), status)
	}
	return err
}

// WithMetricsReporter 设置指标报告器
func (e *ThrottledBatchExecutor) WithMetricsReporter(metricsReporter MetricsReporter) *ThrottledBatchExecutor {
	e.metricsReporter = metricsReporter
	// 注入 reporter 后，立即上报一次当前并发度（如已配置）
	if e.metricsReporter != nil {
		if e.semaphore != nil {
			e.metricsReporter.SetConcurrency(cap(e.semaphore))
		} else {
			e.metricsReporter.SetConcurrency(0)
		}
	}
	return e
}

// MetricsReporter 获取指标报告器
func (e *ThrottledBatchExecutor) MetricsReporter() MetricsReporter { return e.metricsReporter }

func (e *ThrottledBatchExecutor) WithObserver(observer Observer) *ThrottledBatchExecutor {
	e.observer = observer
	return e
}

func (e *ThrottledBatchExecutor) Observer() Observer { return e.observer }

func (e *ThrottledBatchExecutor) WithObservability(config ObservabilityConfig) *ThrottledBatchExecutor {
	e.observer = config.observer()
	return e
}

func (e *ThrottledBatchExecutor) WithCoalescer(coalescer Coalescer) *ThrottledBatchExecutor {
	e.coalescer = coalescer
	return e
}

type attemptResult struct {
	preview  OperationPreview
	err      error
	duration time.Duration
}

func (e *ThrottledBatchExecutor) executeAttempt(ctx context.Context, schema SchemaInterface, data []map[string]any, attempt int) attemptResult {
	attemptStart := time.Now()
	operations, preview, hasPreview, err := e.generateOperations(ctx, schema, data)
	if err != nil {
		err = batchErrorFromError(BatchStageGenerate, preview, len(data), err)
		e.reportOperationError(schema.Name(), BatchStageGenerate, err)
		duration := time.Since(attemptStart)
		e.observeBatchEvent(ctx, newBatchEvent(BatchStageGenerate, "fail", attempt, len(data), duration, schema.Name(), preview, err, defaultOperationErrorReason(err)))
		return attemptResult{preview: preview, err: err, duration: duration}
	}

	e.reportOperationGenerated(schema.Name(), operations, data, preview, hasPreview)
	err = e.processor.ExecuteOperations(ctx, operations)
	if err != nil {
		err = batchErrorFromError(BatchStageExecute, preview, len(data), err)
		e.reportOperationError(schema.Name(), BatchStageExecute, err)
		duration := time.Since(attemptStart)
		e.observeBatchEvent(ctx, newBatchEvent(BatchStageExecute, "fail", attempt, len(data), duration, schema.Name(), preview, err, defaultOperationErrorReason(err)))
		return attemptResult{preview: preview, err: err, duration: duration}
	}

	duration := time.Since(attemptStart)
	e.observeBatchEvent(ctx, newBatchEvent(BatchStageExecute, "success", attempt, len(data), duration, schema.Name(), preview, nil, ""))
	return attemptResult{preview: preview, duration: duration}
}

func (e *ThrottledBatchExecutor) handleRetry(ctx context.Context, schema SchemaInterface, data []map[string]any, result attemptResult, attempt, attempts int, startTime time.Time) (bool, error) {
	retryable, reason := false, "unknown"
	if e.retryClassifier != nil {
		retryable, reason = e.retryClassifier(result.err)
	}
	if !e.retryEnabled || attempt == attempts || !retryable {
		if e.metricsReporter != nil {
			e.metricsReporter.IncError(schema.Name(), "final:"+reason)
		}
		e.observeBatchEvent(ctx, newBatchEvent(BatchStageFinal, "fail", attempt, len(data), time.Since(startTime), schema.Name(), result.preview, result.err, reason))
		return false, nil
	}

	if e.metricsReporter != nil {
		e.metricsReporter.IncError(schema.Name(), "retry:"+reason)
	}
	e.observeBatchEvent(ctx, newBatchEvent(BatchStageRetry, "retry", attempt, len(data), result.duration, schema.Name(), result.preview, result.err, reason))

	timer := time.NewTimer(e.retryBackoff(attempt))
	select {
	case <-ctx.Done():
		if !timer.Stop() {
			select {
			case <-timer.C:
			default:
			}
		}
		return false, ctx.Err()
	case <-timer.C:
		return true, nil
	}
}

func (e *ThrottledBatchExecutor) retryBackoff(attempt int) time.Duration {
	backoff := e.retryBackoffBase
	for i := 1; i < attempt; i++ {
		backoff *= 2
		if backoff > e.retryMaxBackoff {
			backoff = e.retryMaxBackoff
			break
		}
	}
	jitter := time.Duration(int64(float64(backoff) * 0.2))
	return backoff - jitter + time.Duration(randInt63n(int64(2*jitter+1)))
}

func (e *ThrottledBatchExecutor) generateOperations(ctx context.Context, schema SchemaInterface, data []map[string]any) (Operations, OperationPreview, bool, error) {
	if p, ok := e.processor.(OperationPreviewer); ok {
		ops, preview, err := p.GenerateOperationPreview(ctx, schema, data)
		if preview.Schema == "" {
			preview.Schema = schema.Name()
		}
		if preview.InputItems == 0 {
			preview.InputItems = len(data)
		}
		return ops, preview, true, err
	}
	ops, err := e.processor.GenerateOperations(ctx, schema, data)
	preview := OperationPreview{
		Backend:     BackendCustom,
		Operation:   OperationCustom,
		Schema:      schema.Name(),
		InputItems:  len(data),
		OutputItems: len(ops),
		ArgCount:    len(ops),
		Fingerprint: OperationFingerprint(BackendCustom, schema.Name(), strconv.Itoa(len(ops))),
	}
	return ops, preview, false, err
}

func (e *ThrottledBatchExecutor) reportOperationGenerated(table string, operations Operations, data []map[string]any, preview OperationPreview, hasPreview bool) {
	if reporter, ok := e.metricsReporter.(OperationMetricsReporter); ok {
		if hasPreview {
			reporter.ObserveOperationGenerated(preview)
		} else {
			reporter.ObserveOperationGenerated(OperationPreview{
				Backend:     BackendCustom,
				Operation:   OperationCustom,
				Schema:      table,
				InputItems:  len(data),
				OutputItems: len(operations),
				ArgCount:    len(operations),
				Fingerprint: OperationFingerprint(BackendCustom, table, strconv.Itoa(len(operations))),
			})
		}
	}
	reporter, ok := e.metricsReporter.(SQLMetricsReporter)
	if !ok {
		return
	}
	if sqlPreview, ok := sqlPreviewFromOperation(preview); ok {
		reporter.ObserveSQLGenerated(table, sqlPreview.DedupStats.InputRows, sqlPreview.DedupStats.OutputRows, sqlPreview.ArgsCount)
		reporter.ObserveSQLDeduplicated(table, sqlPreview.ConflictStrategy, sqlPreview.DedupStats.DeduplicatedRows, sqlPreview.DedupStats.MergedRows)
	}
}

func (e *ThrottledBatchExecutor) reportOperationError(table, stage string, err error) {
	if err == nil {
		return
	}
	reason := defaultOperationErrorReason(err)
	if reporter, ok := e.metricsReporter.(OperationMetricsReporter); ok {
		backend := ""
		var batchErr *BatchError
		if errors.As(err, &batchErr) {
			backend = batchErr.Backend
		}
		reporter.IncOperationError(table, backend, stage, reason)
	}
	if reporter, ok := e.metricsReporter.(SQLMetricsReporter); ok && isSQLError(err) {
		sqlStage := SQLStage(stage)
		var sqlErr *SQLError
		if errors.As(err, &sqlErr) {
			sqlStage = sqlErr.Stage
			reason = defaultOperationErrorReason(sqlErr.Cause)
		}
		reporter.IncSQLError(table, sqlStage, reason)
	}
}

func isSQLError(err error) bool {
	var sqlErr *SQLError
	if errors.As(err, &sqlErr) {
		return true
	}
	var batchErr *BatchError
	return errors.As(err, &batchErr) && batchErr.Backend == BackendSQL
}

func (e *ThrottledBatchExecutor) observeBatchEvent(ctx context.Context, event BatchEvent) {
	if e.observer != nil {
		e.observer.OnBatchEvent(ctx, event)
	}
}

func newBatchEvent(stage, status string, attempt, batchSize int, duration time.Duration, schema string, preview OperationPreview, err error, reason string) BatchEvent {
	return BatchEvent{
		Stage:       stage,
		Status:      status,
		Attempt:     attempt,
		BatchSize:   batchSize,
		Duration:    duration,
		Err:         err,
		Reason:      reason,
		Backend:     preview.Backend,
		Operation:   preview.Operation,
		Schema:      schema,
		InputItems:  preview.InputItems,
		OutputItems: preview.OutputItems,
		ArgCount:    preview.ArgCount,
		Fingerprint: preview.Fingerprint,
		Attributes:  cloneAttributes(preview.Attributes),
	}
}

func defaultOperationErrorReason(err error) string {
	if err == nil {
		return ErrorReasonUnknown
	}
	_, reason := ClassifyError(err)
	return reason
}

func sqlPreviewFromOperation(preview OperationPreview) (SQLPreview, bool) {
	if preview.Backend != BackendSQL {
		return SQLPreview{}, false
	}
	stats := SQLDedupStats{
		InputRows:        preview.InputItems,
		OutputRows:       preview.OutputItems,
		DeduplicatedRows: intAttribute(preview.Attributes, "deduplicated_rows"),
		MergedRows:       intAttribute(preview.Attributes, "merged_rows"),
	}
	return SQLPreview{
		Table:            preview.Schema,
		ArgsCount:        preview.ArgCount,
		ConflictStrategy: conflictStrategyFromName(fmt.Sprint(preview.Attributes["conflict_strategy"])),
		DedupStats:       stats,
		Fingerprint:      preview.Fingerprint,
	}, true
}

func conflictStrategyFromName(name string) ConflictStrategy {
	switch name {
	case "replace":
		return ConflictReplace
	case "update":
		return ConflictUpdate
	default:
		return ConflictIgnore
	}
}

func intAttribute(attrs map[string]any, key string) int {
	switch v := attrs[key].(type) {
	case int:
		return v
	case int64:
		return int(v)
	case float64:
		return int(v)
	default:
		return 0
	}
}

// WithConcurrencyLimit 设置并发上限（limit <= 0 表示不启用限流）
func (e *ThrottledBatchExecutor) WithConcurrencyLimit(limit int) *ThrottledBatchExecutor {
	if limit > 0 {
		e.semaphore = make(chan struct{}, limit)
	} else {
		e.semaphore = nil
	}
	// 配置并发上限时，上报 Gauge（0 表示不限流）
	if e.metricsReporter != nil {
		if e.semaphore != nil {
			e.metricsReporter.SetConcurrency(cap(e.semaphore))
		} else {
			e.metricsReporter.SetConcurrency(0)
		}
	}
	return e
}

// Executor 模拟批量执行器（用于测试）
type MockExecutor struct {
	ExecutedBatches [][]map[string]any
	driver          SQLDriver
	mu              sync.RWMutex

	// 并发安全的统计聚合：按表名累计批次数、行数、参数数
	statsMu sync.Mutex
	stats   map[string]*mockStats
}

var _ BatchExecutor = (*MockExecutor)(nil)

// NewMockExecutor 创建模拟批量执行器（使用默认Driver）
func NewMockExecutor() *MockExecutor {
	return &MockExecutor{
		ExecutedBatches: make([][]map[string]any, 0),
		driver:          DefaultMySQLDriver,
		stats:           make(map[string]*mockStats),
	}
}

// NewMockExecutorWithDriver 创建模拟批量执行器（使用自定义Driver）
func NewMockExecutorWithDriver(driver SQLDriver) *MockExecutor {
	if driver == nil {
		driver = DefaultMySQLDriver
	}
	return &MockExecutor{
		ExecutedBatches: make([][]map[string]any, 0),
		driver:          driver,
		stats:           make(map[string]*mockStats),
	}
}

type mockStats struct {
	Batches int64
	Rows    int64
	Args    int64
}

// addStats 并发安全地累计统计
func (e *MockExecutor) addStats(table string, rows, args int) {
	if table == "" {
		table = "_unknown_"
	}
	e.statsMu.Lock()
	s, ok := e.stats[table]
	if !ok {
		s = &mockStats{}
		e.stats[table] = s
	}
	s.Batches++
	s.Rows += int64(rows)
	s.Args += int64(args)
	e.statsMu.Unlock()
}

// SnapshotResults 返回只读快照（拷贝），用于测试收尾输出或断言
func (e *MockExecutor) SnapshotResults() map[string]map[string]int64 {
	out := make(map[string]map[string]int64)
	e.statsMu.Lock()
	for k, v := range e.stats {
		out[k] = map[string]int64{
			"batches": v.Batches,
			"rows":    v.Rows,
			"args":    v.Args,
		}
	}
	e.statsMu.Unlock()
	return out
}

// ExecuteBatch 模拟执行批量操作
func (e *MockExecutor) ExecuteBatch(ctx context.Context, schema SchemaInterface, data []map[string]any) error {
	s, ok := schema.(*SQLSchema)
	if !ok {
		return errors.New("schema is not a SQLSchema")
	}

	e.mu.Lock()
	e.ExecutedBatches = append(e.ExecutedBatches, data)
	e.mu.Unlock()

	// 生成SQL信息（不输出大参数）
	_, args, err := e.driver.GenerateInsertSQL(ctx, s, data)
	if err != nil {
		return err
	}

	// 统计聚合（避免每批次打印噪音日志）
	e.addStats(schema.Name(), len(data), len(args))

	return nil
}

// SnapshotExecutedBatches 返回一次性快照，避免并发读写竞态
func (e *MockExecutor) SnapshotExecutedBatches() [][]map[string]any {
	e.mu.RLock()
	defer e.mu.RUnlock()
	out := make([][]map[string]any, len(e.ExecutedBatches))
	copy(out, e.ExecutedBatches)
	return out
}

// randInt63n 返回 [0,n) 的随机数；避免额外依赖，用 time.Now 纳秒抖动
func randInt63n(n int64) int64 {
	if n <= 0 {
		return 0
	}
	// LCG 简易随机（不要求强随机，仅用于退避抖动）
	seed := time.Now().UnixNano()
	seed = (seed*6364136223846793005 + 1) & 0x7fffffffffffffff
	return int64(seed % n)
}