package service

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

	. "github.com/onsi/ginkgo/v2"
	. "github.com/onsi/gomega"
	"github.com/tidbcloud/devops/pkg/infra/errs"
	metrics "github.com/tidbcloud/devops/pkg/infra/metrics/devops-server"

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

func TestService(t *testing.T) {
	RegisterFailHandler(Fail)
	RunSpecs(t, "EventJob Suite")
}

var _ = Describe("EventJob", func() {
	var (
		ctx       context.Context
		cancel    context.CancelFunc
		eventCh   chan int
		processed *sync.Map // Use sync.Map for concurrent-safe access to processed events.
	)

	BeforeEach(func() {
		ctx, cancel = context.WithCancel(context.Background())
		eventCh = make(chan int, 10) // Use a buffered channel to avoid blocking the sender in tests.
		processed = &sync.Map{}
	})

	AfterEach(func() {
		cancel() // Ensure the context is canceled after each test.
	})

	doFunc := func(ctx context.Context, ev int) error {
		time.Sleep(10 * time.Millisecond)
		processed.Store(ev, true)
		return nil
	}

	It("should process events from the channel", func() {
		job := NewEventJob("test-basic", eventCh, doFunc)
		go job.Run(ctx)

		eventCh <- 123
		Eventually(func() bool {
			_, ok := processed.Load(123)
			return ok
		}).Should(BeTrue())
	})

	It("should stop all workers when the context is canceled", func() {
		// Use a 'do' function that blocks to test the cancellation logic.
		blocker := make(chan struct{})
		blockingDoFunc := func(ctx context.Context, ev int) error {
			processed.Store(ev, true)
			<-blocker // Wait until unblocked.
			return nil
		}

		job := NewEventJob("test-cancel", eventCh, blockingDoFunc).With(WithConcurrency(2))
		go job.Run(ctx)

		// Send an event and confirm it's received by a worker.
		eventCh <- 1
		Eventually(func() bool {
			_, ok := processed.Load(1)
			return ok
		}).Should(BeTrue())

		// Cancel the context.
		cancel()

		// Try to send another event.
		eventCh <- 2

		// Consistently check that the second event is not processed, as workers should have exited.
		Consistently(func() bool {
			_, ok := processed.Load(2)
			return ok
		}, "100ms", "10ms").Should(BeFalse())

		// Unblock the goroutine to allow it to exit cleanly.
		close(blocker)
	})

	It("should exit cleanly when the event channel is closed", func() {
		localEventCh := make(chan int, 1)
		var processedCount int64

		job := NewEventJob("test-eventch-closed", localEventCh, 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(localEventCh)

		Eventually(runDone).Should(BeClosed())
		Consistently(func() int64 {
			return atomic.LoadInt64(&processedCount)
		}, "100ms", "10ms").Should(Equal(int64(0)))
	})

	Context("when configured with WithConcurrency option", func() {
		It("should process events concurrently with the specified number of workers", func() {
			concurrency := 5
			// A channel for synchronization, workers will signal when they start.
			workerStarted := make(chan struct{}, concurrency)
			// A channel to block workers until we are ready.
			releaseWorkers := make(chan struct{})

			concurrentDoFunc := func(ctx context.Context, ev int) error {
				workerStarted <- struct{}{} // Signal that processing has started.
				processed.Store(ev, true)
				<-releaseWorkers // Wait for the release signal.
				return nil
			}

			job := NewEventJob("test-concurrency", eventCh, concurrentDoFunc).With(WithConcurrency(concurrency))
			go job.Run(ctx)

			// Send enough events to start all workers.
			for i := 0; i < concurrency; i++ {
				eventCh <- i
			}

			// Verify that all workers have started and are busy.
			Eventually(func() int {
				return len(workerStarted)
			}).Should(Equal(concurrency))

			// Release all workers.
			close(releaseWorkers)

			// Verify that all events have been processed.
			Eventually(func() int {
				count := 0
				processed.Range(func(key, value any) bool {
					count++
					return true
				})
				return count
			}).Should(Equal(concurrency))
		})
	})

	Context("when using middleware", func() {
		Describe("StatMiddleware", func() {
			It("should correctly count successful, failed, and retriable events", func() {
				stats := &JobStat{}
				statDoFunc := func(ctx context.Context, ev int) error {
					switch ev {
					case 1: // Success
						return nil
					case 2: // Retriable error
						return errs.RetriableErrorf("retriable error")
					default: // Failure
						return errors.New("fatal error")
					}
				}

				job := NewEventJob("test-stats", eventCh, statDoFunc)
				job.Use(StatMiddleware[int](stats)) // Apply the middleware.
				go job.Run(ctx)

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

				Eventually(func() int64 { return atomic.LoadInt64(&stats.TotalCount) }).Should(Equal(int64(3)))
				Eventually(func() int64 { return atomic.LoadInt64(&stats.SuccessCount) }).Should(Equal(int64(1)))
				Eventually(func() int64 { return atomic.LoadInt64(&stats.RetriableCount) }).Should(Equal(int64(1)))
				Eventually(func() int64 { return atomic.LoadInt64(&stats.FailedCount) }).Should(Equal(int64(1)))
			})
		})

		Describe("DedupeMiddleware", func() {
			It("should prevent processing of an in-flight duplicate event, but allow it after completion", func() {
				processingCh := make(chan int, 1) // Signal: an event is being processed.
				finishCh := make(chan int, 1)     // Signal: finish processing an event.

				dedupeDoFunc := func(ctx context.Context, ev int) error {
					processed.Store(ev, true)
					processingCh <- ev // Signal that processing has started.
					<-finishCh         // Wait for a signal to finish processing.
					return nil
				}

				job := NewEventJob("test-dedupe", eventCh, dedupeDoFunc).
					With(WithConcurrency(2)). // Allow two workers to process events.
					Use(DedupeMiddleware(func(k int) int { return k }))
				go job.Run(ctx)

				// 1. Send the first event.
				eventCh <- 101

				// Wait for event 101 to start processing.
				Eventually(processingCh).Should(Receive(Equal(101)))

				// 2. While event 101 is processing, send a duplicate and a new event.
				eventCh <- 101 // This one should be ignored by the dedupe logic.
				eventCh <- 202 // This one should be processed by the second worker.

				// 3. Verify that the new event 202 starts processing.
				Eventually(processingCh).Should(Receive(Equal(202)))

				// 4. Finish processing both events.
				finishCh <- 101
				finishCh <- 202

				// 5. Verify the results.
				// Wait for tasks to complete; a short sleep is simple enough for this test.
				time.Sleep(100 * time.Millisecond)

				processedCount := 0
				processedValues := []int{}
				processed.Range(func(key, value any) bool {
					processedCount++
					processedValues = append(processedValues, key.(int))
					return true
				})
				Expect(processedCount).To(Equal(2))
				Expect(processedValues).To(ConsistOf(101, 202))

				// 6. Now that the previous event is done, sending it again should succeed.
				eventCh <- 101
				Eventually(processingCh).Should(Receive(Equal(101)))
				finishCh <- 101
			})
		})
	})

	Context("metrics", func() {
		It("should emit sync count and duration metrics with result labels", func() {
			serviceName := "EventJobMetricService"
			jobName := fmt.Sprintf("event-metric-%d", time.Now().UnixNano())
			ctx = context.WithValue(ctx, serviceNameContextKey{}, serviceName)
			eventCh = make(chan int, 10)

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

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

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

			Eventually(func() float64 {
				return testutil.ToFloat64(
					metrics.EventJob.SyncCount.WithLabelValues(serviceName, jobName, "success"),
				)
			}).Should(Equal(beforeSuccessCount + 1))
			Eventually(func() float64 {
				return testutil.ToFloat64(
					metrics.EventJob.SyncCount.WithLabelValues(serviceName, jobName, "retriable"),
				)
			}).Should(Equal(beforeRetriableCount + 1))
			Eventually(func() float64 {
				return testutil.ToFloat64(
					metrics.EventJob.SyncCount.WithLabelValues(serviceName, jobName, "failed"),
				)
			}).Should(Equal(beforeFailedCount + 1))
			Eventually(func() uint64 {
				return summarySampleCount(metrics.EventJob.SyncDur, serviceName, jobName, "success")
			}).Should(Equal(beforeSuccessDurCount + 1))
			Eventually(func() uint64 {
				return summarySampleCount(metrics.EventJob.SyncDur, serviceName, jobName, "retriable")
			}).Should(Equal(beforeRetriableDurCount + 1))
			Eventually(func() uint64 {
				return summarySampleCount(metrics.EventJob.SyncDur, serviceName, jobName, "failed")
			}).Should(Equal(beforeFailedDurCount + 1))
		})
	})
})

func summarySampleCount(summaryVec *prometheus.SummaryVec, labels ...string) uint64 {
	observer, err := summaryVec.GetMetricWithLabelValues(labels...)
	Expect(err).NotTo(HaveOccurred())

	metric, ok := observer.(prometheus.Metric)
	Expect(ok).To(BeTrue())

	dtoMetric := &dto.Metric{}
	err = metric.Write(dtoMetric)
	Expect(err).NotTo(HaveOccurred())

	return dtoMetric.GetSummary().GetSampleCount()
}