package service

import (
	"context"
	"errors"
	"fmt"
	"sync"
	"sync/atomic"
	"testing"
	"time"

	"github.com/prometheus/client_golang/prometheus"
	"github.com/prometheus/client_golang/prometheus/testutil"
	dto "github.com/prometheus/client_model/go"

	"github.com/tidbcloud/aisupport/packages/sdk-go/infra/metrics"
)

var registerTestServiceMetricsOnce sync.Once

func TestEventJobProcessEventsFromChannel(t *testing.T) {
	t.Parallel()
	registerTestServiceMetrics()

	ctx, cancel := context.WithCancel(context.Background())
	defer cancel()
	eventCh := make(chan int, 10)
	processed := &sync.Map{}

	job := NewEventJob("test-basic", eventCh, func(ctx context.Context, ev int) error {
		time.Sleep(10 * time.Millisecond)
		processed.Store(ev, true)
		return nil
	})
	go job.Run(ctx)

	eventCh <- 123
	eventually(t, time.Second, 10*time.Millisecond, func() bool {
		_, ok := processed.Load(123)
		return ok
	}, "event was not processed")
}

func TestEventJobStopsWorkersWhenContextIsCanceled(t *testing.T) {
	t.Parallel()
	registerTestServiceMetrics()

	ctx, cancel := context.WithCancel(context.Background())
	defer cancel()
	eventCh := make(chan int, 10)
	processed := &sync.Map{}
	blocker := make(chan struct{})

	job := NewEventJob("test-cancel", eventCh, func(ctx context.Context, ev int) error {
		processed.Store(ev, true)
		<-blocker
		return nil
	}).With(WithConcurrency(2))
	runDone := make(chan struct{})
	go func() {
		defer close(runDone)
		job.Run(ctx)
	}()

	eventCh <- 1
	eventually(t, time.Second, 10*time.Millisecond, func() bool {
		_, ok := processed.Load(1)
		return ok
	}, "first event was not processed")

	cancel()
	close(blocker)

	select {
	case <-runDone:
	case <-time.After(time.Second):
		t.Fatal("event job did not exit after context was canceled")
	}
}

func TestEventJobExitsCleanlyWhenEventChannelIsClosed(t *testing.T) {
	t.Parallel()
	registerTestServiceMetrics()

	ctx, cancel := context.WithCancel(context.Background())
	defer cancel()
	eventCh := make(chan int, 1)
	var processedCount int64

	job := NewEventJob("test-eventch-closed", eventCh, func(ctx context.Context, ev int) error {
		atomic.AddInt64(&processedCount, 1)
		return nil
	}).With(WithConcurrency(2))

	runDone := make(chan struct{})
	go func() {
		defer close(runDone)
		job.Run(ctx)
	}()

	close(eventCh)

	select {
	case <-runDone:
	case <-time.After(time.Second):
		t.Fatal("event job did not exit after event channel was closed")
	}
	consistently(t, 100*time.Millisecond, 10*time.Millisecond, func() bool {
		return atomic.LoadInt64(&processedCount) == 0
	}, "closed event channel caused a zero-value event to be processed")
}

func TestEventJobProcessesEventsConcurrently(t *testing.T) {
	t.Parallel()
	registerTestServiceMetrics()

	ctx, cancel := context.WithCancel(context.Background())
	defer cancel()
	eventCh := make(chan int, 10)
	processed := &sync.Map{}
	concurrency := 5
	workerStarted := make(chan struct{}, concurrency)
	releaseWorkers := make(chan struct{})
	defer close(releaseWorkers)

	job := NewEventJob("test-concurrency", eventCh, func(ctx context.Context, ev int) error {
		workerStarted <- struct{}{}
		processed.Store(ev, true)
		<-releaseWorkers
		return nil
	}).With(WithConcurrency(concurrency))
	go job.Run(ctx)

	for i := 0; i < concurrency; i++ {
		eventCh <- i
	}

	eventually(t, time.Second, 10*time.Millisecond, func() bool {
		return len(workerStarted) == concurrency
	}, "not all workers started")

	eventually(t, time.Second, 10*time.Millisecond, func() bool {
		return syncMapLen(processed) == concurrency
	}, "not all events were processed")
}

func TestEventJobStatMiddlewareCountsResults(t *testing.T) {
	t.Parallel()
	registerTestServiceMetrics()

	ctx, cancel := context.WithCancel(context.Background())
	defer cancel()
	eventCh := make(chan int, 10)
	stats := &JobStat{}

	job := NewEventJob("test-stats", eventCh, func(ctx context.Context, ev int) error {
		switch ev {
		case 1:
			return nil
		case 2:
			return RetriableErrorf("retriable error")
		default:
			return errors.New("fatal error")
		}
	})
	job.Use(StatMiddleware[int](stats))
	go job.Run(ctx)

	eventCh <- 1
	eventCh <- 2
	eventCh <- 3

	eventually(t, time.Second, 10*time.Millisecond, func() bool {
		return atomic.LoadInt64(&stats.TotalCount) == 3
	}, "total count was not updated")
	if got := atomic.LoadInt64(&stats.SuccessCount); got != 1 {
		t.Fatalf("success count = %d, want 1", got)
	}
	if got := atomic.LoadInt64(&stats.RetriableCount); got != 1 {
		t.Fatalf("retriable count = %d, want 1", got)
	}
	if got := atomic.LoadInt64(&stats.FailedCount); got != 1 {
		t.Fatalf("failed count = %d, want 1", got)
	}
}

func TestEventJobDedupeMiddlewareIgnoresInFlightDuplicates(t *testing.T) {
	t.Parallel()
	registerTestServiceMetrics()

	ctx, cancel := context.WithCancel(context.Background())
	defer cancel()
	eventCh := make(chan int, 10)
	processed := &sync.Map{}
	processingCh := make(chan int, 2)
	finishCh := make(chan int, 2)
	completedCh := make(chan int, 2)

	job := NewEventJob("test-dedupe", eventCh, func(ctx context.Context, ev int) error {
		processed.Store(ev, true)
		processingCh <- ev
		<-finishCh
		completedCh <- ev
		return nil
	}).With(WithConcurrency(2)).Use(DedupeMiddleware(func(k int) int { return k }))
	go job.Run(ctx)

	eventCh <- 101
	assertReceiveInt(t, processingCh, 101)

	eventCh <- 101
	eventCh <- 202
	assertReceiveInt(t, processingCh, 202)

	finishCh <- 101
	finishCh <- 202
	assertReceiveAnyInt(t, completedCh, 101, 202)
	assertReceiveAnyInt(t, completedCh, 101, 202)

	eventually(t, time.Second, 10*time.Millisecond, func() bool {
		values := syncMapIntKeys(processed)
		return len(values) == 2 && containsInt(values, 101) && containsInt(values, 202)
	}, "dedupe middleware processed unexpected values")

	eventCh <- 101
	assertReceiveInt(t, processingCh, 101)
	finishCh <- 101
}

func TestEventJobMetrics(t *testing.T) {
	t.Parallel()
	registerTestServiceMetrics()

	serviceName := "EventJobMetricService"
	jobName := fmt.Sprintf("event-metric-%d", time.Now().UnixNano())
	ctx, cancel := context.WithCancel(context.Background())
	defer cancel()
	ctx = context.WithValue(ctx, serviceNameContextKey{}, serviceName)
	eventCh := make(chan int, 10)

	beforeSuccessCount := testutil.ToFloat64(
		metrics.ServiceEventJob.SyncCount.WithLabelValues(serviceName, jobName, "success"),
	)
	beforeRetriableCount := testutil.ToFloat64(
		metrics.ServiceEventJob.SyncCount.WithLabelValues(serviceName, jobName, "retriable"),
	)
	beforeFailedCount := testutil.ToFloat64(
		metrics.ServiceEventJob.SyncCount.WithLabelValues(serviceName, jobName, "failed"),
	)
	beforeSuccessDurCount := summarySampleCount(t, metrics.ServiceEventJob.SyncDur, serviceName, jobName, "success")
	beforeRetriableDurCount := summarySampleCount(t, metrics.ServiceEventJob.SyncDur, serviceName, jobName, "retriable")
	beforeFailedDurCount := summarySampleCount(t, metrics.ServiceEventJob.SyncDur, serviceName, jobName, "failed")

	job := NewEventJob(jobName, eventCh, func(ctx context.Context, ev int) error {
		switch ev {
		case 1:
			return nil
		case 2:
			return RetriableErrorf("retriable")
		default:
			return errors.New("failed")
		}
	})
	go job.Run(ctx)

	eventCh <- 1
	eventCh <- 2
	eventCh <- 3

	eventually(t, time.Second, 10*time.Millisecond, func() bool {
		return testutil.ToFloat64(
			metrics.ServiceEventJob.SyncCount.WithLabelValues(serviceName, jobName, "success"),
		) == beforeSuccessCount+1
	}, "success count was not incremented")
	eventually(t, time.Second, 10*time.Millisecond, func() bool {
		return testutil.ToFloat64(
			metrics.ServiceEventJob.SyncCount.WithLabelValues(serviceName, jobName, "retriable"),
		) == beforeRetriableCount+1
	}, "retriable count was not incremented")
	eventually(t, time.Second, 10*time.Millisecond, func() bool {
		return testutil.ToFloat64(
			metrics.ServiceEventJob.SyncCount.WithLabelValues(serviceName, jobName, "failed"),
		) == beforeFailedCount+1
	}, "failed count was not incremented")
	eventually(t, time.Second, 10*time.Millisecond, func() bool {
		return summarySampleCount(t, metrics.ServiceEventJob.SyncDur, serviceName, jobName, "success") == beforeSuccessDurCount+1
	}, "success duration count was not incremented")
	eventually(t, time.Second, 10*time.Millisecond, func() bool {
		return summarySampleCount(t, metrics.ServiceEventJob.SyncDur, serviceName, jobName, "retriable") == beforeRetriableDurCount+1
	}, "retriable duration count was not incremented")
	eventually(t, time.Second, 10*time.Millisecond, func() bool {
		return summarySampleCount(t, metrics.ServiceEventJob.SyncDur, serviceName, jobName, "failed") == beforeFailedDurCount+1
	}, "failed duration count was not incremented")
}

func registerTestServiceMetrics() {
	registerTestServiceMetricsOnce.Do(func() {
		metrics.RegisterServiceMetrics("aisupport_test")
	})
}

func eventually(t *testing.T, timeout, interval time.Duration, condition func() bool, message string) {
	t.Helper()

	deadline := time.Now().Add(timeout)
	for {
		if condition() {
			return
		}
		if time.Now().After(deadline) {
			t.Fatal(message)
		}
		time.Sleep(interval)
	}
}

func consistently(t *testing.T, duration, interval time.Duration, condition func() bool, message string) {
	t.Helper()

	deadline := time.Now().Add(duration)
	for time.Now().Before(deadline) {
		if !condition() {
			t.Fatal(message)
		}
		time.Sleep(interval)
	}
}

func assertReceiveInt(t *testing.T, ch <-chan int, want int) {
	t.Helper()

	select {
	case got := <-ch:
		if got != want {
			t.Fatalf("received %d, want %d", got, want)
		}
	case <-time.After(time.Second):
		t.Fatalf("timed out waiting to receive %d", want)
	}
}

func assertReceiveAnyInt(t *testing.T, ch <-chan int, wants ...int) {
	t.Helper()

	select {
	case got := <-ch:
		if !containsInt(wants, got) {
			t.Fatalf("received %d, want one of %v", got, wants)
		}
	case <-time.After(time.Second):
		t.Fatalf("timed out waiting to receive one of %v", wants)
	}
}

func syncMapLen(m *sync.Map) int {
	count := 0
	m.Range(func(key, value any) bool {
		count++
		return true
	})
	return count
}

func syncMapIntKeys(m *sync.Map) []int {
	values := []int{}
	m.Range(func(key, value any) bool {
		values = append(values, key.(int))
		return true
	})
	return values
}

func containsInt(values []int, want int) bool {
	for _, value := range values {
		if value == want {
			return true
		}
	}
	return false
}

func summarySampleCount(t *testing.T, summaryVec *prometheus.SummaryVec, labels ...string) uint64 {
	t.Helper()

	observer, err := summaryVec.GetMetricWithLabelValues(labels...)
	if err != nil {
		t.Fatalf("get summary metric: %v", err)
	}

	metric, ok := observer.(prometheus.Metric)
	if !ok {
		t.Fatalf("summary observer has type %T, want prometheus.Metric", observer)
	}

	dtoMetric := &dto.Metric{}
	if err := metric.Write(dtoMetric); err != nil {
		t.Fatalf("write summary metric: %v", err)
	}

	return dtoMetric.GetSummary().GetSampleCount()
}