package pipeline

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

// TestRateGuardLimitsConcurrency proves the semaphore REALLY bounds concurrency (WS6 precondition,
// ревью-2 F4): with max_concurrency=2, no more than 2 goroutines are ever inside the guarded region
// at once, even under 20 racing callers. A broken guard (missing/over-sized semaphore) trips the
// observed-max assertion.
func TestRateGuardLimitsConcurrency(t *testing.T) {
	const cap, workers = 2, 20
	g := newRateGuard(cap, 0)
	var inFlight, maxSeen int64
	var wg sync.WaitGroup
	for i := 0; i < workers; i++ {
		wg.Add(1)
		go func() {
			defer wg.Done()
			rel, err := g.acquire(context.Background())
			if err != nil {
				t.Errorf("acquire: %v", err)
				return
			}
			cur := atomic.AddInt64(&inFlight, 1)
			for {
				m := atomic.LoadInt64(&maxSeen)
				if cur <= m || atomic.CompareAndSwapInt64(&maxSeen, m, cur) {
					break
				}
			}
			time.Sleep(time.Millisecond) // widen the window a concurrency bug would exploit
			atomic.AddInt64(&inFlight, -1)
			rel()
		}()
	}
	wg.Wait()
	if maxSeen > cap {
		t.Fatalf("rate-guard admitted %d concurrent callers, cap is %d", maxSeen, cap)
	}
	if maxSeen == 0 {
		t.Fatalf("no caller was observed in-flight — the guard never admitted anyone")
	}
}

// TestRateGuardNilAndUnlimitedAreNoOps: a nil guard and a zero-config guard both admit instantly and
// never block (the sequential path / an unconfigured model must cost nothing).
func TestRateGuardNilAndUnlimitedAreNoOps(t *testing.T) {
	var nilGuard *rateGuard
	rel, err := nilGuard.acquire(context.Background())
	if err != nil {
		t.Fatalf("nil guard acquire: %v", err)
	}
	rel() // must not panic
	unlimited := newRateGuard(0, 0)
	for i := 0; i < 100; i++ {
		rel, err := unlimited.acquire(context.Background())
		if err != nil {
			t.Fatalf("unlimited acquire %d: %v", i, err)
		}
		rel()
	}
}

// TestRateGuardHonoursContextCancel: a caller blocked on a full semaphore unblocks on ctx cancel and
// does NOT leak a slot (the wave dispatcher must be able to abort a run cleanly).
func TestRateGuardHonoursContextCancel(t *testing.T) {
	g := newRateGuard(1, 0)
	rel, err := g.acquire(context.Background())
	if err != nil {
		t.Fatalf("first acquire: %v", err)
	}
	ctx, cancel := context.WithCancel(context.Background())
	done := make(chan error, 1)
	go func() {
		r, err := g.acquire(ctx) // blocks: the one slot is held
		r()
		done <- err
	}()
	cancel()
	select {
	case err := <-done:
		if err != context.Canceled {
			t.Fatalf("blocked acquire on cancel = %v, want context.Canceled", err)
		}
	case <-time.After(time.Second):
		t.Fatal("acquire did not unblock on ctx cancel")
	}
	rel() // release the first; a fresh acquire must now succeed (no leaked slot)
	r2, err := g.acquire(context.Background())
	if err != nil {
		t.Fatalf("acquire after release: %v", err)
	}
	r2()
}

// TestRateGuardPacingSpacesStarts: with min_interval set, two serial acquires are spaced by at least
// the interval (best-effort pacing of call STARTS).
func TestRateGuardPacingSpacesStarts(t *testing.T) {
	g := newRateGuard(0, 30*time.Millisecond)
	start := time.Now()
	r1, _ := g.acquire(context.Background())
	r1()
	r2, _ := g.acquire(context.Background())
	r2()
	if elapsed := time.Since(start); elapsed < 30*time.Millisecond {
		t.Fatalf("paced acquires took %v, want ≥30ms between starts", elapsed)
	}
}