package taskqueue

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

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

var (
	TypeStr = component.MustNewType("taskqueue")
)

func createDefaultConfig() component.Config {
	return &Config{
		Interval:  "15s",
		Timeout:   "60s",
		BatchMode: false,
		BatchSize: "15m",
		Persist: persistent.Config{
			PersistType:   "configmap",
			ConfigMapName: "downstream-ts",
		},
	}
}

func createQueue(ctx context.Context, params extension.Settings, baseCfg component.Config) (extension.Extension, error) {
	cfg := baseCfg.(*Config)
	interval, _ := time.ParseDuration(cfg.Interval)
	timeout, _ := time.ParseDuration(cfg.Timeout)
	batch, _ := time.ParseDuration(cfg.BatchSize)

	if cfg.BatchMode && batch <= 0 {
		return nil, fmt.Errorf("batch size should greater than zero when batch mode is enabled")
	}

	rdr := &taskQueueExtension{
		logger: params.Logger,

		config:   cfg,
		stopChan: make(chan struct{}),

		consumer: &consumerGroup{
			queue:    make(chan Task),
			ackChan:  make(chan Task),
			nackChan: make(chan nackRequest),
			stopChan: make(chan struct{}),
		},

		queue: list.New(),

		interval: interval,
		timeout:  timeout,
		startTs:  cfg.StartTs,
		endTs:    cfg.EndTs,
		batch:    batch,
	}

	persistentCli, err := persistent.NewPersistCli(cfg.Persist)
	if err != nil {
		return nil, fmt.Errorf("create persist client failed: %w", err)
	}

	err = persistentCli.Init()
	if err != nil {
		return nil, fmt.Errorf("init persist client failed: %w", err)
	}

	rdr.persistCli = persistentCli

	rdr.queueCond = sync.NewCond(&rdr.queueMutex)

	return rdr, nil
}

// NewFactory creates a factory for tailtracer receiver.
func NewFactory() extension.Factory {
	return extension.NewFactory(
		TypeStr,
		createDefaultConfig,
		createQueue,
		component.StabilityLevelAlpha,
	)
}