package upstream

import (
	"context"
	"errors"
	"net"
	"strings"
	"time"
)

// Authentication failures intentionally use the slow retry class while
// transient network failures use capped backoff; manual disconnect closes stopCh.

func (p *connectionPool) startSupervisor() {
	p.supervisorOnce.Do(func() {
		go p.superviseReconnects()
	})
}

func (p *connectionPool) stopped() bool {
	select {
	case <-p.stopCh:
		return true
	default:
		return false
	}
}

func (p *connectionPool) signalReconnect() {
	select {
	case <-p.stopCh:
		return
	default:
	}
	select {
	case p.reconnectSignal <- struct{}{}:
	default:
	}
}

func (p *connectionPool) scheduleReconnect(stateErr error) {
	select {
	case <-p.stopCh:
		return
	default:
	}
	p.mu.Lock()
	now := time.Now().UTC()
	if !p.nextReconnectAt.IsZero() && p.nextReconnectAt.After(now) {
		p.mu.Unlock()
		return
	}
	p.reconnectCount++
	p.lastReconnectAttemptAt = now
	p.lastErrorCategory = connectionErrorCategory(stateErr)
	delay := reconnectDelay(p.reconnectCount, p.lastErrorCategory)
	p.nextReconnectAt = p.lastReconnectAttemptAt.Add(delay)
	p.mu.Unlock()
	p.signalReconnect()
}

func (p *connectionPool) resetReconnectState() {
	p.mu.Lock()
	p.reconnectCount = 0
	p.lastReconnectAttemptAt = time.Time{}
	p.nextReconnectAt = time.Time{}
	p.lastErrorCategory = ""
	p.mu.Unlock()
	p.signalReconnect()
}

func (p *connectionPool) superviseReconnects() {
	for {
		select {
		case <-p.stopCh:
			return
		case <-p.reconnectSignal:
		}
		for {
			p.mu.Lock()
			next := p.nextReconnectAt
			p.mu.Unlock()
			if next.IsZero() {
				break
			}
			timer := time.NewTimer(time.Until(next))
			select {
			case <-p.stopCh:
				if !timer.Stop() {
					<-timer.C
				}
				return
			case <-p.reconnectSignal:
				if !timer.Stop() {
					<-timer.C
				}
				continue
			case <-timer.C:
			}
			_ = p.reportState(context.Background(), "reconnecting", nil)
			p.mu.Lock()
			p.lastReconnectAttemptAt = time.Now().UTC()
			p.mu.Unlock()
			if err := p.ensureConnected(); err != nil {
				select {
				case <-p.stopCh:
					return
				default:
				}
				p.scheduleReconnect(err)
				_ = p.reportState(context.Background(), "failed", err)
				continue
			}
			select {
			case <-p.stopCh:
				return
			default:
			}
			p.resetReconnectState()
			_ = p.reportState(context.Background(), "connected", nil)
			break
		}
	}
}

func connectionErrorCategory(err error) string {
	if err == nil {
		return ""
	}
	message := strings.ToLower(err.Error())
	switch {
	case strings.Contains(message, "auth"), strings.Contains(message, "password"), strings.Contains(message, "credential"):
		return "authentication"
	case strings.Contains(message, "heartbeat"):
		return "heartbeat_timeout"
	case strings.Contains(message, "timeout"):
		return "timeout"
	default:
		return "network"
	}
}

func reconnectDelay(attempt int, category string) time.Duration {
	if category == "authentication" {
		return defaultAuthReconnectDelay
	}
	if attempt < 1 {
		attempt = 1
	}
	delays := []time.Duration{
		defaultReconnectInitialDelay,
		15 * time.Second,
		30 * time.Second,
		time.Minute,
		2 * time.Minute,
		defaultReconnectMaximumDelay,
	}
	delay := delays[min(attempt-1, len(delays)-1)]
	// Deterministic ±10% jitter prevents a large set of channels from retrying together.
	offsetPercent := (attempt*37)%21 - 10
	delay += time.Duration(int64(delay) * int64(offsetPercent) / 100)
	if delay < time.Second {
		return time.Second
	}
	return delay
}

func isTemporaryReadTimeout(err error) bool {
	var netErr net.Error
	return errors.As(err, &netErr) && netErr.Timeout()
}