package service

import (
	"context"
	"fmt"
	"sync"
	"sync/atomic"
	"time"
)

// BatchResult 批量任务结果
type BatchResult struct {
	Success []any
	Fail    []BatchError
}

// BatchError 批量任务失败记录
type BatchError struct {
	Index int
	Item  any
	Err   error
}

// BatchWorker 批量处理器，适合文件处理、数据转换等高并发场景
type BatchWorker struct {
	mu          sync.Mutex
	running     atomic.Int64
	total       atomic.Int64
	success     atomic.Int64
	failed      atomic.Int64
	maxConcurrency int
}

// NewBatchWorker 创建批量处理器
func NewBatchWorker(maxConcurrency int) *BatchWorker {
	if maxConcurrency <= 0 {
		maxConcurrency = 10
	}
	return &BatchWorker{
		maxConcurrency: maxConcurrency,
	}
}

// ProcessConcurrent 并发处理一批数据
func (bw *BatchWorker) ProcessConcurrent(
	ctx context.Context,
	items []any,
	handler func(ctx context.Context, item any) (any, error),
) BatchResult {
	bw.total.Add(int64(len(items)))
	results := BatchResult{}
	sem := make(chan struct{}, bw.maxConcurrency)
	var wg sync.WaitGroup
	var mu sync.Mutex

	for i, item := range items {
		select {
		case <-ctx.Done():
			break
		default:
		}
		wg.Add(1)
		sem <- struct{}{}
		go func(index int, data any) {
			defer wg.Done()
			defer func() { <-sem }()
			bw.running.Add(1)
			defer bw.running.Add(-1)

			result, err := handler(ctx, data)
			bw.success.Add(1)

			mu.Lock()
			defer mu.Unlock()
			if err != nil {
				bw.failed.Add(1)
				results.Fail = append(results.Fail, BatchError{
					Index: index,
					Item:  data,
					Err:   err,
				})
			} else {
				results.Success = append(results.Success, result)
			}
		}(i, item)
	}

	wg.Wait()
	return results
}

// ProcessSequential 顺序处理（用于需要严格顺序的场景）
func (bw *BatchWorker) ProcessSequential(
	ctx context.Context,
	items []any,
	handler func(ctx context.Context, item any) (any, error),
) BatchResult {
	results := BatchResult{}
	for i, item := range items {
		select {
		case <-ctx.Done():
			return results
		default:
		}
		result, err := handler(ctx, item)
		if err != nil {
			results.Fail = append(results.Fail, BatchError{Index: i, Item: item, Err: err})
		} else {
			results.Success = append(results.Success, result)
		}
	}
	return results
}

// Stats 当前处理统计
func (bw *BatchWorker) Stats() (running, total, success, failed int64) {
	return bw.running.Load(), bw.total.Load(), bw.success.Load(), bw.failed.Load()
}

// RingBuffer 无锁环形缓冲区，适合高频消息传递
type RingBuffer struct {
	data   []any
	head   int64
	tail   int64
	capacity int
	size   atomic.Int64
}

// NewRingBuffer 创建环形缓冲区
func NewRingBuffer(capacity int) *RingBuffer {
	if capacity <= 0 {
		capacity = 1024
	}
	rb := &RingBuffer{
		data:     make([]any, capacity),
		capacity: capacity,
	}
	return rb
}

// Enqueue 入队（阻塞直到有空位）
func (rb *RingBuffer) Enqueue(item any) {
	for {
		tail := atomic.LoadInt64(&rb.tail)
		head := atomic.LoadInt64(&rb.head)
		if tail-head >= int64(rb.capacity) {
			time.Sleep(time.Microsecond)
			continue
		}
		if atomic.CompareAndSwapInt64(&rb.tail, tail, tail+1) {
			rb.data[tail%int64(rb.capacity)] = item
			rb.size.Add(1)
			return
		}
	}
}

// Dequeue 出队（阻塞直到有数据）
func (rb *RingBuffer) Dequeue() (any, bool) {
	for {
		head := atomic.LoadInt64(&rb.head)
		tail := atomic.LoadInt64(&rb.tail)
		if head >= tail {
			return nil, false
		}
		if atomic.CompareAndSwapInt64(&rb.head, head, head+1) {
			item := rb.data[head%int64(rb.capacity)]
			rb.data[head%int64(rb.capacity)] = nil
			rb.size.Add(-1)
			return item, true
		}
	}
}

// Size 当前元素数量
func (rb *RingBuffer) Size() int64 {
	return rb.size.Load()
}

// RateLimiter 令牌桶限流器
type RateLimiter struct {
	tokens     int64
	maxTokens  int64
	refillRate int64 // 每秒补充数
	lastRefill time.Time
	mu         sync.Mutex
}

// NewRateLimiter 创建限流器
func NewRateLimiter(maxTokens, refillRate int64) *RateLimiter {
	rl := &RateLimiter{
		tokens:     maxTokens,
		maxTokens:  maxTokens,
		refillRate: refillRate,
		lastRefill: time.Now(),
	}
	go rl.refillLoop()
	return rl
}

func (rl *RateLimiter) refillLoop() {
	ticker := time.NewTicker(time.Second)
	defer ticker.Stop()
	for range ticker.C {
		rl.mu.Lock()
		elapsed := time.Since(rl.lastRefill).Seconds()
		add := int64(elapsed * float64(rl.refillRate))
		current := rl.tokens
		if current+add > rl.maxTokens {
			add = rl.maxTokens - current
		}
		rl.tokens += add
		rl.lastRefill = time.Now()
		rl.mu.Unlock()
	}
}

// Acquire 获取一个令牌，返回是否成功
func (rl *RateLimiter) Acquire() bool {
	rl.mu.Lock()
	defer rl.mu.Unlock()
	if rl.tokens <= 0 {
		return false
	}
	rl.tokens--
	return true
}

// Limit 带限制的包装函数
func (rl *RateLimiter) Limit(fn func() error) error {
	for !rl.Acquire() {
		time.Sleep(time.Millisecond * 10)
	}
	return fn()
}

// ConcurrentMap 线程安全的 map，支持并发读写
type ConcurrentMap struct {
	mu    sync.RWMutex
	items map[any]any
}

// NewConcurrentMap 创建线程安全 map
func NewConcurrentMap() *ConcurrentMap {
	return &ConcurrentMap{items: make(map[any]any)}
}

func (cm *ConcurrentMap) Set(k, v any) {
	cm.mu.Lock()
	defer cm.mu.Unlock()
	cm.items[k] = v
}

func (cm *ConcurrentMap) Get(k any) (any, bool) {
	cm.mu.RLock()
	defer cm.mu.RUnlock()
	v, ok := cm.items[k]
	return v, ok
}

func (cm *ConcurrentMap) Delete(k any) {
	cm.mu.Lock()
	defer cm.mu.Unlock()
	delete(cm.items, k)
}

func (cm *ConcurrentMap) Len() int {
	cm.mu.RLock()
	defer cm.mu.RUnlock()
	return len(cm.items)
}

// Pipeline 函数式管道，链式处理数据流
type Pipeline struct {
	stages []func(any) any
	input  chan any
	output chan any
}

// NewPipeline 创建管道
func NewPipeline() *Pipeline {
	return &Pipeline{
		stages: make([]func(any) any, 0),
		input:  make(chan any, 256),
		output: make(chan any, 256),
	}
}

// AddStage 添加处理阶段
func (p *Pipeline) AddStage(fn func(any) any) *Pipeline {
	p.stages = append(p.stages, fn)
	return p
}

// Run 启动管道（阻塞，直到 input 关闭）
func (p *Pipeline) Run(items []any) []any {
	go func() {
		for _, item := range items {
			p.input <- item
		}
		close(p.input)
	}()

	go p.process()
	results := make([]any, 0)
	for v := range p.output {
		results = append(results, v)
	}
	return results
}

func (p *Pipeline) process() {
	defer close(p.output)
	for item := range p.input {
		for _, stage := range p.stages {
			item = stage(item)
		}
		p.output <- item
	}
}

// GoCtx 便捷并发模式
type GoCtx struct {
	wg      sync.WaitGroup
	errs    sync.Map
	results sync.Map
}

// Go 启动一个并发任务
func (gc *GoCtx) Go(id string, fn func() error) {
	gc.wg.Add(1)
	go func() {
		defer gc.wg.Done()
		if err := fn(); err != nil {
			gc.errs.Store(id, err)
		}
	}()
}

// Wait 等待所有任务完成，返回错误列表
func (gc *GoCtx) Wait() []error {
	gc.wg.Wait()
	var errs []error
	gc.errs.Range(func(_, v any) bool {
		errs = append(errs, v.(error))
		return true
	})
	return errs
}

// WaitWithTimeout 带超时的等待
func (gc *GoCtx) WaitWithTimeout(timeout time.Duration) []error {
	done := make(chan struct{})
	go func() {
		gc.wg.Wait()
		close(done)
	}()
	select {
	case <-done:
		var errs []error
		gc.errs.Range(func(_, v any) bool {
			errs = append(errs, v.(error))
			return true
		})
		return errs
	case <-time.After(timeout):
		return []error{fmt.Errorf("并发任务超时: %v", timeout)}
	}
}