package upstream

import (
	"cmpp-platform/gateway/internal/queue"
	"context"
	"net/http"
	"sync"
	"time"
)

type connectionPool struct {
	channelID    string
	connectionID string
	config       queue.UpstreamConfig
	apiBaseURL   string
	httpClient   *http.Client
	reporter     func(context.Context, ConnectionState) error

	mu                     sync.Mutex
	connectMu              sync.Mutex
	conns                  []*connection
	next                   int
	reconnectSignal        chan struct{}
	stopCh                 chan struct{}
	stopOnce               sync.Once
	supervisorOnce         sync.Once
	reconnectCount         int
	lastReconnectAttemptAt time.Time
	nextReconnectAt        time.Time
	lastErrorCategory      string
}

func (p *connectionPool) matches(config queue.UpstreamConfig) bool {
	return p.config == normalizeUpstreamConfig(config)
}

func (p *connectionPool) ensureConnected() error {
	p.connectMu.Lock()
	defer p.connectMu.Unlock()
	desired := p.config.DesiredConnections
	if desired <= 0 {
		desired = 1
	}
	connectedAny := false
	for {
		p.mu.Lock()
		active := p.conns[:0]
		for _, existing := range p.conns {
			existing.mu.Lock()
			usable := existing.client != nil && !existing.closed
			existing.mu.Unlock()
			if usable {
				active = append(active, existing)
			}
		}
		p.conns = active
		if len(p.conns) >= desired {
			p.mu.Unlock()
			break
		}
		index := len(p.conns)
		conn := &connection{
			channelID:        p.channelID,
			config:           p.config,
			index:            index,
			pool:             p,
			apiBaseURL:       p.apiBaseURL,
			httpClient:       p.httpClient,
			window:           make(chan struct{}, p.config.WindowSize),
			pending:          make(map[uint32]chan submitPartResponse),
			tracker:          make(map[uint64]queue.SubmitCommand),
			longUplink:       make(map[string]*longUplinkAssembly),
			heartbeatPending: make(map[uint32]time.Time),
		}
		p.mu.Unlock()

		connected, err := conn.ensureConnected()
		if err != nil {
			return err
		}

		p.mu.Lock()
		p.conns = append(p.conns, conn)
		p.mu.Unlock()
		connectedAny = connectedAny || connected
	}
	if connectedAny {
		_ = p.reportState(context.Background(), "connected", nil)
	}
	return nil
}

func (p *connectionPool) close() {
	p.connectMu.Lock()
	defer p.connectMu.Unlock()
	p.stopOnce.Do(func() {
		close(p.stopCh)
	})
	p.mu.Lock()
	conns := p.conns
	p.conns = nil
	p.mu.Unlock()
	for _, conn := range conns {
		conn.close()
	}
}

func (p *connectionPool) countActiveConnections() int {
	p.mu.Lock()
	defer p.mu.Unlock()
	count := 0
	for _, conn := range p.conns {
		conn.mu.Lock()
		active := conn.client != nil && !conn.closed
		conn.mu.Unlock()
		if active {
			count += 1
		}
	}
	return count
}