package service

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

	"github.com/tidbcloud/devops/pkg/infra/errs"
	"github.com/tidbcloud/devops/pkg/infra/log"
	metrics "github.com/tidbcloud/devops/pkg/infra/metrics/devops-server"

	"go.opentelemetry.io/otel"
)

type (
	EventJobDo[T any]         func(ctx context.Context, ev T) error
	EventJobOption            func(*eventJobOptions)
	EventJobMiddleware[T any] func(do EventJobDo[T]) EventJobDo[T]
)

// WithConcurrency make job handle events concurrently.
func WithConcurrency(concurrency int) EventJobOption {
	return func(o *eventJobOptions) {
		o.concurrency = concurrency
	}
}

// DedupeMiddleware deduplicate events by a hash function.
// If an event is processing, it will be ignored and return nil directly.
func DedupeMiddleware[T any, K comparable](hash func(ev T) K) EventJobMiddleware[T] {
	return func(do EventJobDo[T]) EventJobDo[T] {
		mu := &sync.Mutex{}
		dedupeMaps := make(map[K]struct{})

		return func(ctx context.Context, ev T) error {
			key := hash(ev)

			mu.Lock()

			if _, exists := dedupeMaps[key]; exists {
				mu.Unlock()
				log.FromContext(ctx).Info(fmt.Sprintf("ignore an event because it is duplicated; key: %v", key))
				return nil
			}

			dedupeMaps[key] = struct{}{}
			mu.Unlock()

			err := do(ctx, ev)

			mu.Lock()
			delete(dedupeMaps, key)
			mu.Unlock()
			return err
		}
	}
}

// TraceMiddleware adds tracing to the event job.
func TraceMiddleware[T any](spanName string) EventJobMiddleware[T] {
	tracer := otel.Tracer(TraceScopeName)

	return func(do EventJobDo[T]) EventJobDo[T] {
		return func(ctx context.Context, ev T) error {
			subCtx, span := tracer.Start(ctx, spanName)
			defer span.End()

			return do(subCtx, ev)
		}
	}
}

// StatMiddleware collects statistics for the event job.
func StatMiddleware[T any](stat *JobStat) EventJobMiddleware[T] {
	return func(do EventJobDo[T]) EventJobDo[T] {
		return func(ctx context.Context, ev T) error {
			atomic.AddInt64(&stat.TotalCount, 1)

			err := do(ctx, ev)
			if err != nil {
				if _, is := errs.AsRetriableError(err); is {
					atomic.AddInt64(&stat.RetriableCount, 1)
				} else {
					atomic.AddInt64(&stat.FailedCount, 1)
				}
			} else {
				atomic.AddInt64(&stat.SuccessCount, 1)
			}

			return err
		}
	}
}

// RecoveryMiddleware recovers from panics in the event job.
func RecoveryMiddleware[T any](do EventJobDo[T]) EventJobDo[T] {
	return func(ctx context.Context, ev T) (err error) {
		defer func() {
			if r := recover(); r != nil {
				log.FromContext(ctx).Error(err, "recovered a panic")
				err = fmt.Errorf("recovered from panic: %v", r)
			}
		}()

		return do(ctx, ev)
	}
}

type eventJobOptions struct {
	concurrency int // default 1
}

type eventJob[T any] struct {
	name    string
	eventCh <-chan T
	do      EventJobDo[T]

	opts   eventJobOptions
	workCh chan T
	workWg sync.WaitGroup
}

// NewEventJob creates a new job to handle events from channel.
func NewEventJob[T any](name string, ch <-chan T, do EventJobDo[T]) *eventJob[T] {
	job := &eventJob[T]{
		name:    name,
		eventCh: ch,
		do:      do,

		opts: eventJobOptions{
			concurrency: 1,
		},
	}

	return job
}

func (j *eventJob[T]) Name() string {
	return j.name
}

func (j *eventJob[T]) With(opt EventJobOption) *eventJob[T] {
	opt(&j.opts)
	return j
}

func (j *eventJob[T]) Use(middleware EventJobMiddleware[T]) *eventJob[T] {
	j.do = middleware(j.do)
	return j
}

func (j *eventJob[T]) Run(ctx context.Context) {
	j.Use(TraceMiddleware[T](j.name)).Use(RecoveryMiddleware) // add trace and recovery middleware by default

	// builder workers
	j.workCh = make(chan T, j.opts.concurrency)
	j.workWg = sync.WaitGroup{}
	for index := range j.opts.concurrency {
		j.workWg.Add(1)
		go j.run(ctx, index)
	}

	stopWorkers := func() {
		close(j.workCh)
		j.workWg.Wait()
	}

	for {
		select {
		case <-ctx.Done():
			// wait all worker finished
			stopWorkers()
			return
		case obj, ok := <-j.eventCh:
			if !ok {
				stopWorkers()
				return
			}

			j.workCh <- obj
		}
	}
}

func (j *eventJob[T]) run(ctx context.Context, workerNr int) {
	defer j.workWg.Done()

	for obj := range j.workCh {
		subLogger := log.FromContext(ctx, "worker", workerNr, "object", obj).WithName(j.name)
		jobCtx := log.NewContext(ctx, subLogger)
		beginAt := time.Now()

		err := j.do(jobCtx, obj)

		// record metrics
		result := metrics.ResultFromErr(err)
		serviceName := GetServiceNameFromContext(jobCtx)
		metrics.EventJob.SyncCount.WithLabelValues(serviceName, j.name, result).Inc()
		metrics.EventJob.SyncDur.WithLabelValues(serviceName, j.name, result).Observe(time.Since(beginAt).Seconds())

		if err != nil {
			if rerr, is := errs.AsRetriableError(err); is {
				subLogger.Info("the result of a event job run is retriable", "reason", rerr.Error())
				continue
			}

			subLogger.Error(err, "the result of a event job run is failed")
			continue
		}
	}
}