package source

import (
	"context"
	"time"

	"github.com/go-logr/logr"
	"github.com/tidbcloud/observability/pkg/mysql/client"
	"github.com/tidbcloud/pkgv2/gormutil"
	"k8s.io/client-go/util/workqueue"
	"sigs.k8s.io/controller-runtime/pkg/source"
)

var (
	_ source.TypedSource[any] = (*MysqlSource)(nil)
)

type MysqlSource struct {
	Logger  logr.Logger
	ConnCfg *gormutil.Config

	PollingInterval time.Duration
	CleanupInterval time.Duration
	TTL             time.Duration
}

func (ms *MysqlSource) Start(ctx context.Context, queue workqueue.TypedRateLimitingInterface[any]) error {
	log := ms.Logger.WithName("MysqlSource")

	ttl := client.DefaultEventTTL
	if ms.TTL != 0 {
		ttl = ms.TTL
	}
	mysqlConsumer := &Consumer{
		WorkQueue: queue,
		ConnCfg:   ms.ConnCfg,

		PollingInterval: ms.PollingInterval,
		CleanupInterval: ms.CleanupInterval,
		TTL:             ttl,
	}
	mysqlConsumer.SetLogger(log)

	if err := mysqlConsumer.Init(ctx); err != nil {
		return err
	}
	log.Info("start mysql consumer")
	go mysqlConsumer.Start(ctx) //nolint:errcheck // ignore

	return nil
}