package source

import (
	"context"
	"errors"
	"fmt"
	"time"

	"github.com/go-logr/logr"
	"github.com/tidbcloud/observability/pkg/mysql/client"
	"github.com/tidbcloud/pkgv2/gormutil"
	"gorm.io/gorm"
)

const (
	defaultPollingInterval = 1 * time.Second
	defaultCleanupInterval = 10 * time.Second
)

type Adder interface {
	Add(item any)
}

type Consumer struct {
	log         logr.Logger
	lastEventID uint64

	ConnCfg *gormutil.Config

	// PollingInterval is the interval for polling new events from client.
	PollingInterval time.Duration
	// CleanupInterval is the interval for cleaning up expired events.
	CleanupInterval time.Duration
	TTL             time.Duration

	WorkQueue Adder
	DB        *gorm.DB
}

func (c *Consumer) SetLogger(l logr.Logger) {
	c.log = l.WithName("MySQLConsumer")
}

func (c *Consumer) Init(ctx context.Context) error {
	db, err := client.NewGormDB(c.ConnCfg)
	if err != nil {
		return fmt.Errorf("failed to create gorm DB: %w", err)
	}
	c.DB = db
	c.log.Info("mysql consumer initialized")
	return nil
}

func (c *Consumer) Start(ctx context.Context) error {
	pollingInterval := defaultPollingInterval
	if c.PollingInterval != 0 {
		pollingInterval = c.PollingInterval
	}
	cleanupInterval := defaultCleanupInterval
	if c.CleanupInterval != 0 {
		cleanupInterval = c.CleanupInterval
	}

	pollingTicker := time.NewTicker(pollingInterval)
	defer pollingTicker.Stop()
	cleanupTicker := time.NewTicker(cleanupInterval)
	defer cleanupTicker.Stop()

	for {
		select {
		case <-ctx.Done():
			c.log.Info("mysql consumer stopped")
			return nil
		case <-pollingTicker.C:
			if err := c.fetchNewEvents(ctx); err != nil {
				if !errors.Is(err, context.Canceled) {
					c.log.Error(err, "failed to fetch new events", "lastEventID", c.lastEventID)
				}
			}
		case <-cleanupTicker.C:
			if err := c.cleanupExpiredEvents(ctx); err != nil {
				if !errors.Is(err, context.Canceled) {
					c.log.Error(err, "failed to cleanup old events", "lastEventID", c.lastEventID)
				}
			}
		}
	}
}

func (c *Consumer) cleanupExpiredEvents(ctx context.Context) error {
	tx := c.DB.WithContext(ctx).Begin()
	if tx.Error != nil {
		return fmt.Errorf("failed to begin transaction: %w", tx.Error)
	}
	defer func() {
		if r := recover(); r != nil {
			tx.Rollback()
			panic(r)
		}
	}()

	query := tx.Where("status = ? AND ttl_expiration < ?", client.EventStatusDone, time.Now())

	if err := query.Delete(&client.ClusterEvent{}).Error; err != nil {
		tx.Rollback()
		return fmt.Errorf("failed to cleanup expired events: %w", err)
	}
	return tx.Commit().Error
}

func (c *Consumer) fetchNewEvents(ctx context.Context) error {
	tx := c.DB.WithContext(ctx).Begin()
	if tx.Error != nil {
		return fmt.Errorf("failed to begin transaction: %w", tx.Error)
	}
	defer func() {
		if r := recover(); r != nil {
			tx.Rollback()
			panic(r)
		}
	}()

	var events []*client.ClusterEvent
	query := tx.Where("event_id > ? AND status != ?", c.lastEventID, client.EventStatusDone)
	if err := query.Order("event_id ASC").Find(&events).Error; err != nil {
		tx.Rollback()
		return fmt.Errorf("failed to query events: %w", err)
	}

	if len(events) == 0 {
		return tx.Commit().Error
	}

	c.lastEventID = events[len(events)-1].EventID

	mergedEvents := make(map[string]*client.ClusterEvent) // key is namespace/name
	for _, event := range events {
		key := fmt.Sprintf("%s/%s", event.Namespace, event.Name)
		mergedEvents[key] = event
	}

	c.log.Info("fetched new events", "count", len(mergedEvents), "lastEventID", c.lastEventID)
	for _, event := range mergedEvents {
		c.WorkQueue.Add(MySQLItem{
			Ctx:       ctx,
			DB:        c.DB,
			TTL:       c.TTL,
			EventID:   event.EventID,
			Namespace: event.Namespace,
			Name:      event.Name,
		})
	}
	return tx.Commit().Error
}