package actor

import "sync/atomic"

// queue separates message storage and FIFO publication from consumer ownership.
// Only its current consumer may call takeBatch and pop.
type queue[T any] struct {
	head atomic.Pointer[node[T]]

	fixed    []node[T]
	reserved atomic.Uint64
	consumed int // Accessed only by the current consumer.
}

type node[T any] struct {
	value T
	next  *node[T]
	fixed bool // Set at construction and never changed.
}

func newQueue[T any](capacity int) *queue[T] {
	q := &queue[T]{fixed: make([]node[T], capacity)}
	for i := range q.fixed {
		q.fixed[i].fixed = true
	}
	return q
}

func (q *queue[T]) push(message T) {
	n := q.reserveNode()
	n.value = message
	q.publish(n)
}

func (q *queue[T]) reserveNode() *node[T] {
	capacity := uint64(len(q.fixed))
	for {
		index := q.reserved.Load()
		if index == capacity {
			return new(node[T])
		}
		if q.reserved.CompareAndSwap(index, index+1) {
			// A slot belongs to this producer only after the reservation CAS.
			return &q.fixed[index]
		}
	}
}

func (q *queue[T]) publish(n *node[T]) {
	for {
		head := q.head.Load()
		n.next = head
		if q.head.CompareAndSwap(head, n) {
			// The consumer may immediately clear and reuse n. Do not access
			// any of its fields after this successful publication.
			return
		}
	}
}

func (q *queue[T]) empty() bool {
	return q.head.Load() == nil
}

func (q *queue[T]) takeBatch(batchSize int, yield func()) *node[T] {
	n := q.head.Swap(nil)
	var reversed *node[T]
	steps := 0
	for n != nil {
		next := n.next
		n.next = reversed
		reversed = n
		n = next
		steps++
		if steps == batchSize {
			steps = 0
			yield()
		}
	}
	return reversed
}

func (q *queue[T]) pop(n *node[T]) (message T, next *node[T]) {
	next, message, fixed := n.next, n.value, n.fixed
	var zero T
	n.value = zero
	n.next = nil
	// Save everything first: reopening the fixed storage allows a producer to
	// reuse n immediately. Neither this method nor its caller may read it again.
	if fixed {
		q.consumed++
		if q.consumed == len(q.fixed) {
			q.consumed = 0
			q.reserved.Store(0)
		}
	}
	return message, next
}