package concurrency

import (
	"context"
	"log/slog"
	"path/filepath"
	"sync"
	"sync/atomic"
	"time"
)

const defaultMaxConcurrency = 5

type ConcurrencyLimit struct {
	Key            string `yaml:"key" json:"key"`
	MaxConcurrency int    `yaml:"max_concurrency" json:"max_concurrency"`
}

type Config struct {
	Default int                `yaml:"default" json:"default"`
	Limits  []ConcurrencyLimit `yaml:"limits" json:"limits"`
}

type Limiter struct {
	config atomic.Value // *Config
	slots  sync.Map     // map[string]chan struct{}
}

var (
	defaultLimiter     *Limiter
	defaultLimiterOnce sync.Once
)

func DefaultLimiter() *Limiter {
	defaultLimiterOnce.Do(func() {
		defaultLimiter = NewLimiter(Config{Default: defaultMaxConcurrency})
	})
	return defaultLimiter
}

func NewLimiter(cfg Config) *Limiter {
	if cfg.Default <= 0 {
		cfg.Default = defaultMaxConcurrency
	}
	l := &Limiter{}
	l.config.Store(&cfg)
	return l
}

func (l *Limiter) Update(cfg Config) {
	if cfg.Default <= 0 {
		cfg.Default = defaultMaxConcurrency
	}
	l.config.Store(&cfg)
	slog.Info("Concurrency limiter updated", "default", cfg.Default, "rules", len(cfg.Limits))
}

func (l *Limiter) Acquire(ctx context.Context, key string) (waitDuration time.Duration, release func(), err error) {
	limit := l.getLimit(key)
	sem := l.getOrCreateSemaphore(key, limit)

	start := time.Now()
	select {
	case sem <- struct{}{}:
		waitDuration = time.Since(start)
		release = func() {
			<-sem
		}
		return
	case <-ctx.Done():
		err = ctx.Err()
		return
	}
}

func (l *Limiter) getLimit(key string) int {
	cfg := l.config.Load().(*Config)

	for _, rule := range cfg.Limits {
		if rule.Key == key {
			return rule.MaxConcurrency
		}
	}

	for _, rule := range cfg.Limits {
		if matched, _ := filepath.Match(rule.Key, key); matched {
			return rule.MaxConcurrency
		}
	}

	return cfg.Default
}

func (l *Limiter) getOrCreateSemaphore(key string, limit int) chan struct{} {
	if v, ok := l.slots.Load(key); ok {
		if ch := v.(chan struct{}); cap(ch) == limit {
			return ch
		}
	}

	ch := make(chan struct{}, limit)
	actual, loaded := l.slots.LoadOrStore(key, ch)
	if !loaded {
		return ch
	}
	existing := actual.(chan struct{})
	if cap(existing) == limit {
		return existing
	}
	l.slots.Store(key, ch)
	return ch
}