package taskqueue

import (
	"container/list"
	"context"
	"errors"
	"fmt"
	"strconv"
	"sync"
	"time"

	"github.com/tidbcloud/metrics-integration/pipeline/internal/otel_collector/extension/taskqueue/persistent"
	"go.opentelemetry.io/collector/component"
	"go.uber.org/zap"
)

var (
	ErrorConsumerClosed = errors.New("consumer_closed")
)

type taskQueueExtension struct {
	logger *zap.Logger

	config   *Config
	wg       sync.WaitGroup
	stopChan chan struct{}

	// consumer related fields
	consumer *consumerGroup

	// core data structure related
	queue      *list.List
	queueCond  *sync.Cond
	queueMutex sync.Mutex

	// task generator setting
	timerMu  sync.Mutex
	interval time.Duration
	timeout  time.Duration
	batch    time.Duration
	startTs  int64
	endTs    int64
	ackedTs  time.Time
	lastTs   time.Time

	// persistent
	persistCli persistent.PersistClient
}

func (tQueue *taskQueueExtension) Start(ctx context.Context, host component.Host) error {
	// init start ts
	if tQueue.startTs != 0 {
		// use startTs specified in config
		tQueue.logger.Info("use startTs specified in the config as startTime", zap.Int64("start_ts", tQueue.startTs))
	} else {
		recorded, err := tQueue.persistCli.Get(DOWNSTREAM_TIMESTAMP_KEY)
		if persistent.IsNotFound(err) {
			// should generate startTs from now()
			tQueue.logger.Info("recorded timestamp not found, use time.now() as starttime", zap.Time("startTime", time.Now().Truncate(tQueue.interval)))
		} else if err != nil {
			return fmt.Errorf("error start taskqueue extension component: %w", err)
		} else {
			recordedTs, err := strconv.ParseInt(recorded, 10, 64)
			if err != nil {
				return fmt.Errorf("failed to parse recorded timestamp (%s)", recorded)
			}

			tQueue.logger.Info("startTs found in persister", zap.Int64("recorded_ts", recordedTs))
			tQueue.ackedTs = time.Unix(recordedTs, 0).Truncate(tQueue.interval)
		}
	}

	if tQueue.endTs != 0 {
		tQueue.logger.Info("the endTs is configured", zap.Int64("end_ts", tQueue.endTs))
	}

	// tQueue.currentTs = time.Now().Truncate(tQueue.interval)

	err := tQueue.GenerateInitialTask()
	if err != nil {
		return fmt.Errorf("generate initial task failed:%w", err)
	}

	tQueue.wg.Add(1)
	go tQueue.runDispatchConsumer(ctx)
	return nil
}

func (tQueue *taskQueueExtension) Shutdown(ctx context.Context) error {
	close(tQueue.stopChan)
	tQueue.consumer.stopChan <- consumerStop{}
	tQueue.wg.Wait()
	tQueue.consumer.Close()
	return nil
}

func (tQueue *taskQueueExtension) Enqueue(task Task) error {
	return tQueue.enqueue(task, false)
}

func (tQueue *taskQueueExtension) EnqueueToFront(task Task) error {
	return tQueue.enqueue(task, true)

}

// QueueSize 获取当前队列大小
func (tQueue *taskQueueExtension) QueueSize() int {
	return tQueue.queue.Len()
}

func (tqe *taskQueueExtension) enqueue(task Task, toFront bool) error {
	tqe.queueMutex.Lock()
	defer tqe.queueMutex.Unlock()

	// current no size limit for the queue
	// if tqe.queue.Len() >= tqe.config.QueueSize {
	// 	return fmt.Errorf("task queue is full")
	// }

	if toFront {
		tqe.queue.PushFront(task)
	} else {
		tqe.queue.PushBack(task)
	}

	// 通知等待的消费者
	tqe.queueCond.Signal()

	return nil
}

func (tqe *taskQueueExtension) dequeue() (Task, bool) {
	tqe.queueMutex.Lock()
	defer tqe.queueMutex.Unlock()

	for tqe.queue.Len() == 0 {
		select {
		case <-tqe.stopChan:
			return Task{}, false
		default:
			tqe.queueCond.Wait()
		}
	}

	element := tqe.queue.Front()
	return tqe.queue.Remove(element).(Task), true
}