package k8stool

import (
	"context"

	"k8s.io/client-go/util/workqueue"
)

type WorkqueuePuller[T comparable] struct {
	workqueue workqueue.TypedRateLimitingInterface[T]
	work      chan T
}

func NewWorkqueuePuller[T comparable](workqueue workqueue.TypedRateLimitingInterface[T]) *WorkqueuePuller[T] {
	return &WorkqueuePuller[T]{
		workqueue: workqueue,
		work:      make(chan T),
	}
}

func (p *WorkqueuePuller[T]) Run(ctx context.Context) {
	done := ctx.Done()
	for p.pullNextItemFromWorkqueue(done) {
	}
}

func (p *WorkqueuePuller[T]) Work() <-chan T {
	return p.work
}

func (p *WorkqueuePuller[T]) pullNextItemFromWorkqueue(done <-chan struct{}) bool {
	key, shutdown := p.workqueue.Get()
	if shutdown {
		return false
	}

	select {
	case <-done:
		p.workqueue.Add(key)  // put it back
		p.workqueue.Done(key) // mark as done with it
		return false
	case p.work <- key:
		return true
	}
}
