package actor

import (
	"runtime"
	"sync/atomic"
)

// mailbox owns the single-consumer handoff and scheduling policy. running also
// covers the interval after scheduling and before the task starts executing.
type mailbox[T any] struct {
	queue      *queue[T]
	invoker    invoker[T]
	dispatcher Dispatcher
	task       func()
	batchSize  int
	yield      func()
	running    atomic.Bool
}

func newMailbox[T any](q *queue[T], invoke invoker[T], batchSize int, dispatcher Dispatcher) *mailbox[T] {
	m := &mailbox[T]{
		queue:      q,
		invoker:    invoke,
		dispatcher: dispatcher,
		batchSize:  batchSize,
		yield:      runtime.Gosched,
	}
	// Reuse this method value for every scheduling request, without allocating
	// a closure on each Send or ownership handoff.
	m.task = m.drain
	return m
}

func (m *mailbox[T]) send(message T) {
	m.queue.push(message)
	m.tryRun()
}

func (m *mailbox[T]) tryRun() {
	if m.running.CompareAndSwap(false, true) {
		m.dispatcher.Schedule(m.task)
	}
}

func (m *mailbox[T]) drain() {
	processed := 0
	for {
		batch := m.queue.takeBatch(m.batchSize, m.yield)
		if batch == nil {
			m.running.Store(false)
			// A producer may have published while running was still true.
			// Either we reclaim ownership or that producer schedules a task.
			if m.queue.empty() || !m.running.CompareAndSwap(false, true) {
				return
			}
			continue
		}
		for n := batch; n != nil; {
			message, next := m.queue.pop(n)
			n = next
			m.invoker.receive(message)
			processed++
			if processed == m.batchSize {
				processed = 0
				m.yield()
			}
		}
	}
}