package node

import (
	"context"
	"errors"
	"sync"
	"sync/atomic"
)

// ErrConcurrencyLimitExceeded is returned when a run is rejected because the
// active run count has reached the effective concurrency limit.
var ErrConcurrencyLimitExceeded = errors.New("concurrency limit exceeded")

type runHandle struct {
	runID     string
	adapter   string
	target    string
	sessionID string
	cancel    context.CancelFunc
	done      chan struct{}
}

type runManager struct {
	mu   sync.RWMutex
	runs map[string]*runHandle
}

func newRunManager() *runManager {
	return &runManager{runs: make(map[string]*runHandle)}
}

func (m *runManager) register(h *runHandle) {
	m.mu.Lock()
	defer m.mu.Unlock()
	m.runs[h.runID] = h
}

func (m *runManager) deregister(runID string) {
	m.mu.Lock()
	defer m.mu.Unlock()
	delete(m.runs, runID)
}

func (m *runManager) cancelRun(runID string) bool {
	m.mu.RLock()
	h, ok := m.runs[runID]
	m.mu.RUnlock()
	if !ok {
		return false
	}
	h.cancel()
	return true
}

// permitManager enforces a concurrency limit using an atomic counter.
// limit <= 0 means unlimited. Policy is reject-on-exceed (no queue, no wait).
type permitManager struct {
	limit  int32
	active atomic.Int32
}

func newPermitManager(limit int) *permitManager {
	return &permitManager{limit: int32(limit)}
}

// tryAcquire attempts to take a permit. Returns false immediately if the limit
// is already reached. limit <= 0 always succeeds.
func (p *permitManager) tryAcquire() bool {
	if p.limit <= 0 {
		p.active.Add(1)
		return true
	}
	for {
		cur := p.active.Load()
		if cur >= p.limit {
			return false
		}
		if p.active.CompareAndSwap(cur, cur+1) {
			return true
		}
	}
}

// release returns a permit. Must be called exactly once per successful tryAcquire.
func (p *permitManager) release() {
	p.active.Add(-1)
}

// activeCount returns the number of currently held permits (for observability).
func (p *permitManager) activeCount() int {
	return int(p.active.Load())
}
