// Copyright 2025 The etcd 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 cache

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

const (
	pollPeriod      = 1 * time.Millisecond
	minimalNoChange = 20 * time.Millisecond
	pollTimeout     = 5 * time.Second
)

// Copied from https://github.com/kubernetes/kubernetes/blob/4116c15/staging/src/k8s.io/apiserver/pkg/storage/cacher/progress/watch_progress_test.go
func TestConditionalProgressRequestor(t *testing.T) {
	ctx := context.Background()

	fc := newFakeClock()
	pr := newTestConditionalProgressRequestor(fc, 100*time.Millisecond)
	stopCtx, cancel := context.WithCancel(ctx)
	defer cancel()
	go pr.run(stopCtx)

	var wantRequestsSent int32
	var requestsSent int32

	t.Log("Wait for goroutine to start")
	time.Sleep(10 * time.Millisecond)

	t.Log("No progress requests if no-one is waiting")
	fc.Advance(250 * time.Millisecond)

	if err := pollConditionNoChange(func() bool {
		requestsSent = pr.progressRequestsSentCount.Load()
		return requestsSent == wantRequestsSent
	}); err != nil {
		t.Fatalf("Failed to wait progress requests, err: %s, want: %d, got %d", err, wantRequestsSent, requestsSent)
	}

	t.Log("Adding waiters allows progress request to be sent")
	pr.add()
	fc.Advance(150 * time.Millisecond)
	wantRequestsSent++
	if err := pollConditionNoChange(func() bool {
		requestsSent = pr.progressRequestsSentCount.Load()
		return requestsSent == wantRequestsSent
	}); err != nil {
		t.Fatalf("Failed to wait progress requests, err: %s, want: %d, got %d", err, wantRequestsSent, requestsSent)
	}

	t.Log("Periodically request progress to be sent every period")
	for wantRequestsSent < 5 {
		fc.Advance(100 * time.Millisecond)
		wantRequestsSent++

		if err := pollConditionNoChange(func() bool {
			requestsSent = pr.progressRequestsSentCount.Load()
			return requestsSent == wantRequestsSent
		}); err != nil {
			t.Fatalf("Failed to wait progress requests, err: %s, want: %d, got %d", err, wantRequestsSent, requestsSent)
		}
	}
	pr.remove()

	t.Log("No progress requests if no-one is waiting")
	fc.Advance(250 * time.Millisecond)
	if err := pollConditionNoChange(func() bool {
		requestsSent = pr.progressRequestsSentCount.Load()
		return requestsSent == wantRequestsSent
	}); err != nil {
		t.Fatalf("Failed to wait progress requests, err: %s, want: %d, got %d", err, wantRequestsSent, requestsSent)
	}

	t.Log("No progress after stopping")
	cancel()
	fc.Advance(50 * time.Millisecond)
	if err := pollConditionNoChange(func() bool {
		requestsSent = pr.progressRequestsSentCount.Load()
		return requestsSent == wantRequestsSent
	}); err != nil {
		t.Fatalf("Failed to wait progress requests, err: %s, want: %d, got %d", err, wantRequestsSent, requestsSent)
	}
	pr.add()
	fc.Advance(250 * time.Millisecond)
	if err := pollConditionNoChange(func() bool {
		requestsSent = pr.progressRequestsSentCount.Load()
		return requestsSent == wantRequestsSent
	}); err != nil {
		t.Fatalf("Failed to wait progress requests, err: %s, want: %d, got %d", err, wantRequestsSent, requestsSent)
	}
}

func newTestConditionalProgressRequestor(clock Clock, requestInterval time.Duration) *testConditionalProgressRequestor {
	pr := &testConditionalProgressRequestor{}
	pr.conditionalProgressRequestor = newConditionalProgressRequestor(pr, clock, requestInterval)
	return pr
}

type testConditionalProgressRequestor struct {
	*conditionalProgressRequestor
	progressRequestsSentCount atomic.Int32
}

func (pr *testConditionalProgressRequestor) RequestProgress(ctx context.Context) error {
	pr.progressRequestsSentCount.Add(1)
	return nil
}

func pollConditionNoChange(condition func() bool) error {
	passCounter := 0
	requiredNumberOfPasses := int(minimalNoChange/pollPeriod) + 1
	deadline := time.Now().Add(pollTimeout)

	ticker := time.NewTicker(pollPeriod)
	defer ticker.Stop()

	for {
		select {
		case <-ticker.C:
			if condition() {
				passCounter++
			} else {
				passCounter = 0
			}
			if passCounter >= requiredNumberOfPasses {
				return nil
			}
			if time.Now().After(deadline) {
				return &timeoutError{}
			}
		}
	}
}

type timeoutError struct{}

func (e *timeoutError) Error() string {
	return "timed out waiting for condition to stabilize"
}