package source

import (
	"fmt"
	"sync"
	"time"

	"github.com/pingcap/log"

	"github.com/tidbcloud/observability/pkg/mysql/client"
	"k8s.io/apimachinery/pkg/types"
	"k8s.io/client-go/util/workqueue"
	"k8s.io/utils/clock"
	"sigs.k8s.io/controller-runtime/pkg/reconcile"
)

// NewMysqlRateLimitingQueueWithConfig constructs a new workqueue with rateLimited queuing ability
// with options to customize different properties.
// Remember to call Forget!  If you don't, you may end up tracking failures forever.
func NewMysqlRateLimitingQueueWithConfig(rateLimiter workqueue.TypedRateLimiter[any], config workqueue.TypedRateLimitingQueueConfig[any], ttl time.Duration) workqueue.TypedRateLimitingInterface[any] {
	if config.Clock == nil {
		config.Clock = clock.RealClock{}
	}

	if config.DelayingQueue == nil {
		config.DelayingQueue = workqueue.NewTypedDelayingQueueWithConfig(workqueue.TypedDelayingQueueConfig[any]{
			Name:            config.Name,
			MetricsProvider: config.MetricsProvider,
			Clock:           config.Clock,
		})
	}

	return &mysqlRateLimitingType{
		DelayingInterface: config.DelayingQueue,
		rateLimiter:       rateLimiter,
		processing:        make(map[any]*FixedQueue[itemMySQLExt]),
		metrics:           newMetricsProvider(config.Name),
	}
}

type itemMySQLExt struct {
	MySQLItem
	forgot bool
}

// There are multiple sources will add item to the queue, such k8s informer's event handler.
// Let's trace the scenarios with the FixedQueue (size=2), and check the correctness of current implementation
// the definition of correct is
// - make sure the object status is eventually consistency, the latest event will never lose.
// - make sure the mysql event is cleaned up if it's popped out and not leaked.
// 1. FixedQueue is Empty
// Action: Add(K8s_Event) is called.
// Logic: An empty itemMySQLExt{} (zero value) is created and enqueued.
// Result: Queue: [EmptyItem].
// Status: Safe
// 2. FixedQueue has One Item: [MySQL_Event_A]
// Action: Add(K8s_Event) is called.
// Logic: An empty itemMySQLExt{} is enqueued.
// Result: Queue: [MySQL_Event_A, EmptyItem].
// Status: Safe
// 3. FixedQueue has Two Items: [MySQL_Event_A, MySQL_Event_B]
// Action: Add(K8s_Event) is called (e.g., a VMRule update related to this cluster).
// Logic:
// The K8s_Event is treated as a new empty item.
// FixedQueue.Enqueue(EmptyItem) detects the queue is full.
// Eviction: It pops the oldest item, MySQL_Event_A.
// Auto-Ack: It calls deleteMessage(&MySQL_Event_A).
// deleteMessage runs UPDATE ... SET status='done' in the database.
// Result: Queue: [MySQL_Event_B, EmptyItem].
// Status: Safe, MySQL_Event_A is marked Done in the DB before the Reconciler has ever seen it.
// But there are items which is newer in the queue, so reconciler will see the latest event and reconcile based on it.
// Even if the regional server got restarted, the k8s re-list mechanism at start phase will enqueue the Item again.
// We still maintain the eventually consistency.
// mysqlRateLimitingType wraps an Interface and provides rateLimited re-enquing
type mysqlRateLimitingType struct {
	workqueue.DelayingInterface

	rateLimiter workqueue.TypedRateLimiter[any]

	// metrics counts the number of deletes
	metrics metricsProvider

	mu         sync.Mutex
	processing map[any]*FixedQueue[itemMySQLExt] // fixed 2 items
}

func (m *mysqlRateLimitingType) Add(item any) {
	// the item may be added from non-mysql source, e.g. k8s informer's event handler.
	req := item
	if isItemMySQL(item) {
		req = reconcile.Request{NamespacedName: types.NamespacedName{
			Namespace: item.(MySQLItem).Namespace,
			Name:      item.(MySQLItem).Name,
		}}
	}

	m.mu.Lock()
	fixedQueue := m.processing[req]
	if fixedQueue == nil {
		fixedQueue = NewFixedQueue[itemMySQLExt](2)
		m.processing[req] = fixedQueue
	}

	// enqueue the empty item if it's not added from mysql source.
	itemExt := itemMySQLExt{}
	if isItemMySQL(item) {
		itemExt.MySQLItem = item.(MySQLItem)
	}
	popped, ok := fixedQueue.Enqueue(itemExt)
	m.mu.Unlock()

	if ok {
		m.deleteMessage(&popped)
		m.metrics.ackMsgInAdd()
	}

	m.DelayingInterface.Add(req) // always need to add even if has been DeleteMessage
}

func (m *mysqlRateLimitingType) AddAfter(item any, duration time.Duration) {
	m.mu.Lock()
	if fixedQueue := m.processing[item]; fixedQueue != nil {
		if front := fixedQueue.Peek(); front != nil {
			front.forgot = false // can't forgot because it adds after
		}
	}
	m.mu.Unlock()

	m.DelayingInterface.AddAfter(item, duration)
}

// AddRateLimited AddAfter's the item based on the time when the rate limiter says it's ok
func (m *mysqlRateLimitingType) AddRateLimited(item any) {
	m.DelayingInterface.AddAfter(item, m.rateLimiter.When(item))
}

func (m *mysqlRateLimitingType) NumRequeues(item any) int {
	return m.rateLimiter.NumRequeues(item)
}

func (m *mysqlRateLimitingType) Forget(item any) {
	m.mu.Lock()
	if fixedQueue := m.processing[item]; fixedQueue != nil {
		if front := fixedQueue.Peek(); front != nil {
			front.forgot = true
		}
	}
	m.mu.Unlock()

	m.rateLimiter.Forget(item)
}

func (m *mysqlRateLimitingType) Done(item any) {
	// q.AddAfter, q.Forget, and q.Done can be considered as a single operation, which has multi entry but end with q.Done
	m.mu.Lock()
	var itemToDeleteMessage *itemMySQLExt
	if fixedQueue := m.processing[item]; fixedQueue != nil {
		if front, ok := fixedQueue.Dequeue(); ok {
			fixedQueueLen := fixedQueue.Len()
			if fixedQueueLen == 0 {
				delete(m.processing, item)
			}
			// DeleteMessage the front item in non-empty fixedQueue(after Dequeue) regardless or the item is forgotten
			if fixedQueueLen > 0 || front.forgot {
				itemToDeleteMessage = &front
			}
		}
	}
	m.mu.Unlock()

	if itemToDeleteMessage != nil {
		m.deleteMessage(itemToDeleteMessage)
		m.metrics.ackMsgInDone()
	}

	m.DelayingInterface.Done(item)
}

func (m *mysqlRateLimitingType) deleteMessage(item *itemMySQLExt) {
	// if item.DB is nil, it means the item is added from non-mysql source, e.g. k8s informer's event handler.
	// in this case, we don't need to update the event status in the database.
	if item == nil || item.DB == nil {
		return
	}
	// Update all pending events for the same cluster_id with event_id <= current item.EventID to StatusDone
	// and set a TTL for the event.
	updates := map[string]any{
		"status":         client.EventStatusDone,
		"ttl_expiration": time.Now().Add(item.TTL),
	}
	if err := item.DB.WithContext(item.Ctx).Model(&client.ClusterEvent{}).
		Where("namespace = ? AND name = ? AND event_id <= ? AND status = ?", item.Namespace, item.Name, item.EventID, client.EventStatusPending).
		Updates(updates).Error; err != nil {
		log.Info(fmt.Sprintf("[mysqlRateLimitingType.deleteMessage] failed to update events for cluster %s/%s (up to event %d) status to done: %v", item.Namespace, item.Name, item.EventID, err))
	}
}

func isItemMySQL(item any) bool {
	_, ok := item.(MySQLItem)
	return ok
}