// steamcache/coalescing.go
// Request coalescing (de-duplicating concurrent identical upstream fetches for the same
// cache key). Includes the coalescedRequest state machine + waiter/leader coordination,
// response buffering for thundering herd avoidance, and the coalescer wrapper that
// owns the in-flight map + mutex (SteamCache methods delegate; no direct map access
// in core or handler).
package steamcache

import (
	"net/http"
	"sync"
	"sync/atomic"
)

type coalescedRequest struct {
	waitingCount atomic.Int32
	done         bool
	mu           sync.Mutex
	// Buffered response data for coalesced clients
	responseData    []byte
	responseHeaders http.Header
	statusCode      int
	// Broadcast signal for all waiters (closed by leader in complete)
	doneCh        chan struct{}
	completionErr error
	// Active protocol (post-legacy cleanup): waiters wake on doneCh, then read completionErr/response* under mu (or pre-unlock copies in waiter).
}

func newCoalescedRequest() *coalescedRequest {
	cr := &coalescedRequest{
		done:            false,
		responseHeaders: make(http.Header),
		doneCh:          make(chan struct{}),
	}
	cr.waitingCount.Store(1)
	return cr
}

func (cr *coalescedRequest) addWaiter() {
	cr.waitingCount.Add(1)
}

func (cr *coalescedRequest) complete(resp *http.Response, err error) {
	cr.mu.Lock()
	defer cr.mu.Unlock()
	if cr.done {
		return
	}
	cr.done = true

	if err != nil {
		cr.completionErr = err
	} else {
		// Store response data for coalesced clients
		if resp != nil {
			cr.statusCode = resp.StatusCode
			// Copy headers (excluding hop-by-hop headers via filter)
			cr.responseHeaders = filterHopByHopHeaders(resp.Header)
		}
	}
	// Broadcast to *all* waiters (thundering herd fix). Close is safe here because of the done guard above.
	close(cr.doneCh)
}

// setResponseData stores the buffered response data for coalesced clients
func (cr *coalescedRequest) setResponseData(data []byte) {
	cr.mu.Lock()
	defer cr.mu.Unlock()
	cr.responseData = make([]byte, len(data))
	copy(cr.responseData, data)
}

// coalescer owns the coalesced requests map and mutex. It encapsulates the
// in-flight request dedup state so SteamCache no longer directly manipulates
// the raw map (Phase 2 extraction). Unexported; same-package access for tests.
type coalescer struct {
	mu       sync.Mutex
	requests map[string]*coalescedRequest
}

// newCoalescer constructs an empty coalescer (called from SteamCache.New).
func newCoalescer() *coalescer {
	return &coalescer{
		requests: make(map[string]*coalescedRequest),
	}
}

func (c *coalescer) getOrCreate(cacheKey string) (*coalescedRequest, bool) {
	c.mu.Lock()
	defer c.mu.Unlock()

	if cr, exists := c.requests[cacheKey]; exists {
		cr.addWaiter()
		return cr, false
	}

	cr := newCoalescedRequest()
	c.requests[cacheKey] = cr
	return cr, true
}

func (c *coalescer) remove(cacheKey string) {
	c.mu.Lock()
	defer c.mu.Unlock()
	delete(c.requests, cacheKey)
}

// getOrCreateCoalescedRequest delegates to the owned coalescer (preserves
// existing call sites in handler.go and any white-box tests unchanged).
func (sc *SteamCache) getOrCreateCoalescedRequest(cacheKey string) (*coalescedRequest, bool) {
	return sc.coalescer.getOrCreate(cacheKey)
}

// removeCoalescedRequest delegates to the owned coalescer.
func (sc *SteamCache) removeCoalescedRequest(cacheKey string) {
	sc.coalescer.remove(cacheKey)
}