package taskqueue

import (
	"context"
	"errors"
	"fmt"
	"sync"

	"go.uber.org/zap"
)

var ErrChanNoWaiting = errors.New("channel no waiting")

type Consumer interface {
	Subscribe(ctx context.Context) <-chan Task

	Ack(task Task) error

	Nack(task Task, requeue bool) error
}

// NackWithRequeue enforces in-place retry semantics for failure handling.
// It ensures the task keeps the same window and is requeued to retry.
func NackWithRequeue(c Consumer, task Task) error {
	task.SetRetryStrategy(RETRYSTRATEGY_REQUEUE)
	return c.Nack(task, true)
}

type consumerGroup struct {
	// name     string
	queue      chan Task
	ackChan    chan Task
	nackChan   chan nackRequest
	stopChan   chan struct{}
	consumerMu sync.RWMutex
	// wg       sync.WaitGroup
	stats consumerStats
}

type nackRequest struct {
	task    Task
	requeue bool
	err     error
}

type consumerStats struct {
	activeTasks    int32
	processedTasks int64
	failedTasks    int64
}

type consumerStop struct{}

func (cg *consumerGroup) Init() {

}

func (cg *consumerGroup) Close() {
	if cg.ackChan != nil {
		close(cg.ackChan)
	}

	if cg.nackChan != nil {
		close(cg.nackChan)
	}

	if cg.stopChan != nil {
		close(cg.stopChan)
	}
}

func (tqe *taskQueueExtension) Subscribe(ctx context.Context) <-chan Task {
	tqe.consumer.consumerMu.RLock()
	defer tqe.consumer.consumerMu.RUnlock()

	return tqe.consumer.queue
}

// Ack 确认任务完成
func (tqe *taskQueueExtension) Ack(task Task) error {

	if tqe.consumer == nil {
		return fmt.Errorf("taskqueue consumer not initiated")
	}

	go func() {
		tqe.logger.Debug("trying to put into ack channel", zap.Time("task_timestamp", task.Timestamp))
		tqe.consumer.ackChan <- task
	}()

	return nil
}

// Nack 确认任务失败
func (tqe *taskQueueExtension) Nack(task Task, requeue bool) error {

	if tqe.consumer == nil {
		return fmt.Errorf("taskqueue consumer not initiated")
	}

	tqe.logger.Info("task nacked, putting back to the request queue", zap.Time("task_ts", task.Timestamp))

	go func() {
		tqe.logger.Debug("trying to put into nack channel", zap.Time("task_timestamp", task.Timestamp))
		tqe.consumer.nackChan <- nackRequest{task: task, requeue: requeue}
	}()
	return nil
}