package taskqueue

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

	"go.uber.org/zap"
)

const (
	requeueRetryDelay   = 500 * time.Millisecond
	retryBypassDuration = 5 * time.Minute
)

func (tqe *taskQueueExtension) runTaskGenerator(ctx context.Context, initialDelay time.Duration) {
	defer tqe.wg.Done()

	// 初始延迟
	if initialDelay > 0 {
		select {
		case <-time.After(initialDelay):
		case <-ctx.Done():
			return
		}
	}

	ticker := time.NewTicker(tqe.interval)
	defer ticker.Stop()

	for {
		select {
		case <-ticker.C:
			// 生成任务并加入队列
			task := Task{
				Timestamp: tqe.GetCurrentTimestampAndProgress(),
			}
			_ = tqe.Enqueue(task) // 忽略错误或记录日志
		case <-ctx.Done():
			return
		case <-tqe.stopChan:
			return
		}
	}
}

// this function is only applicable to single consumer scheme.
func (tqe *taskQueueExtension) runDispatchConsumer(ctx context.Context) {
	defer tqe.wg.Done()

	consumer := tqe.consumer
	if consumer == nil || tqe.queue == nil {
		// tqe isn't correct initialized, should panic immediately.
		tqe.logger.Fatal("tqe isn't correct initialized, should panic immediately.")
	}

	for {
		task, ok := tqe.dequeue()
		if !ok {
			tqe.logger.Info("consumer loop closed")
			return // task queue closed
		}

		// currentTime serve as an identifier to see if we received a staled task or not.
		currentTime := time.Now()
		task.createAt = currentTime

		// todo: add timeout
		select {
		case consumer.queue <- task:

		case <-consumer.stopChan:
			tqe.logger.Info("consumer loop closed")
			return
		}

		err := tqe.processAck(consumer, currentTime, task)
		if err != nil {
			tqe.logger.Info("proccesing ack failed, exiting ......", zap.Error(err))
			return
		}
	}
}

// return until ack, nack, timeout or ctx canceled
func (tqe *taskQueueExtension) processAck(consumer *consumerGroup, currentTaskCreateTime time.Time, task Task) error {
	timeout := task.timeout
	now := time.Now()

	// deal with the issue when timestamp is after current time
	if task.EndTimestamp.After(now) {
		timeout = timeout + task.EndTimestamp.Sub(now)
	}

	tm := time.NewTimer(timeout)

	for {
		select {
		case ackTask := <-consumer.ackChan:
			tm.Stop()
			if ackTask.createAt != currentTaskCreateTime {
				tqe.logger.Error("receive stale task ack", zap.Time("current_time", currentTaskCreateTime), zap.Time("task_time", ackTask.createAt))
				continue
			}

			atomic.AddInt64(&consumer.stats.processedTasks, 1)
			atomic.AddInt32(&consumer.stats.activeTasks, -1)
			tqe.logger.Info("task finished", zap.Time("task_timestamp", task.Timestamp), zap.Time("task_end_timestamp", task.EndTimestamp))

			err := tqe.GenerateRangedTaskWithAck(ackTask.AckTimestamp, false)
			if err != nil {
				tqe.logger.Error("generate new task failed", zap.Error(err))
			}

			return nil

		case nack := <-consumer.nackChan:
			tm.Stop()
			if nack.task.createAt != currentTaskCreateTime {
				tqe.logger.Error("receive stale task nack", zap.Time("current_time", currentTaskCreateTime), zap.Time("task_time", nack.task.createAt), zap.Error(nack.err))
				continue
			}

			atomic.AddInt64(&consumer.stats.failedTasks, 1)
			atomic.AddInt32(&consumer.stats.activeTasks, -1)
			tqe.logger.Info("task failed, should be requeue and re-execute", zap.Error(nack.err))

			if nack.requeue {
				now := time.Now()
				nack.task.MarkFailedAt(now)
				switch nack.task.retryStrategy {
				case RETRYSTRATEGY_REQUEUE:
					if shouldBypassRetry(nack.task, now) {
						nack.task.AckTimestamp = nack.task.EndTimestamp.Truncate(nack.task.Step)
						tqe.logger.Warn("task retry exceeded bypass duration, ack and continue",
							zap.Time("task_timestamp", nack.task.Timestamp),
							zap.Time("task_end_timestamp", nack.task.EndTimestamp),
							zap.Time("first_failed_at", nack.task.FirstFailedAt()),
							zap.Duration("retry_elapsed", now.Sub(nack.task.FirstFailedAt())),
							zap.Duration("bypass_duration", retryBypassDuration),
							zap.Int("retried_count", nack.task.GetRetriedCount()),
						)
						err := tqe.GenerateRangedTaskWithAck(nack.task.AckTimestamp, false)
						if err != nil {
							tqe.logger.Error("generate new task failed", zap.Error(err))
						}
						return nil
					}
					nack.task.AddRetriedCount()
					select {
					case <-time.After(requeueRetryDelay):
					case <-consumer.stopChan:
						return fmt.Errorf("consumer group closed: %w", ErrorConsumerClosed)
					}
					err := tqe.EnqueueToFront(nack.task)
					if err != nil {
						tqe.logger.Error("fail to requeue task, may cause data point lost")
					}
				case RETRYSTRATEGY_REGENERATE:
					err := tqe.GenerateRangedTask(nack.task.Timestamp, true)
					if err != nil {
						tqe.logger.Error("generate new task failed", zap.Error(err))
					}
				default:
					tqe.logger.Warn("unsupported requeue strategy", zap.Any("task", nack.task))
				}

			}
			return nil

		case <-consumer.stopChan:
			tm.Stop()
			tqe.logger.Info("consumer loop closed")
			return fmt.Errorf("consumer group closed: %w", ErrorConsumerClosed)

		case <-tm.C:
			tm.Stop()
			tqe.logger.Info("task timeout", zap.Time("task_timestamp", task.Timestamp))
			now := time.Now()
			task.MarkFailedAt(now)
			if shouldBypassRetry(task, now) {
				task.AckTimestamp = task.EndTimestamp.Truncate(task.Step)
				tqe.logger.Warn("task timeout exceeded bypass duration, ack and continue",
					zap.Time("task_timestamp", task.Timestamp),
					zap.Time("task_end_timestamp", task.EndTimestamp),
					zap.Time("first_failed_at", task.FirstFailedAt()),
					zap.Duration("retry_elapsed", now.Sub(task.FirstFailedAt())),
					zap.Duration("bypass_duration", retryBypassDuration),
					zap.Int("retried_count", task.GetRetriedCount()),
				)
				err := tqe.GenerateRangedTaskWithAck(task.AckTimestamp, false)
				if err != nil {
					tqe.logger.Error("generate new task failed", zap.Error(err))
				}
				return nil
			}
			task.AddRetriedCount()
			select {
			case <-time.After(requeueRetryDelay):
			case <-consumer.stopChan:
				return fmt.Errorf("consumer group closed: %w", ErrorConsumerClosed)
			}
			err := tqe.EnqueueToFront(task)
			if err != nil {
				tqe.logger.Error("fail to requeue task, may cause data point lost")
			}
			return nil

		}
	}

}

func shouldBypassRetry(task Task, now time.Time) bool {
	if task.FirstFailedAt().IsZero() {
		return false
	}

	return now.Sub(task.FirstFailedAt()) >= retryBypassDuration
}