AetherForge/server/internal/pool/manager.go

package pool

import (
	"fmt"
	"log"
	"sync"
	"time"
)


// Manager maintains Stratum connections keyed by forged pool + wallet settings.
type Manager struct {
	mu             sync.RWMutex
	pools          map[string]*Proxy
	onJob          func(job *Job)
	onErr          func(err error)
	reconnectDelay time.Duration
	verboseTraffic bool
}

func NewManager(onJob func(job *Job), onErr func(err error)) *Manager {
	return &Manager{
		pools: make(map[string]*Proxy),
		onJob: onJob,
		onErr: onErr,
	}
}

func (m *Manager) SetReconnectDelay(seconds int) {
	if seconds > 0 {
		m.mu.Lock()
		m.reconnectDelay = time.Duration(seconds) * time.Second
		m.mu.Unlock()
	}
}

func (m *Manager) SetVerboseTraffic(enabled bool) {
	m.mu.Lock()
	m.verboseTraffic = enabled
	for _, p := range m.pools {
		p.SetVerboseTraffic(enabled)
	}
	m.mu.Unlock()
}

func poolKey(cfg *Config) string {
	return fmt.Sprintf("%s:%d:tls=%v:wallet=%s", cfg.Host, cfg.Port, cfg.UseTLS, cfg.Wallet)
}

// EnsurePoolWithBackups connects to cfg and registers backup pool configs for
// automatic failover on reconnect.  If the primary cfg is unreachable it tries
// each backup in order at initial connect time as well — restoring the behaviour
// that existed before the reconnect-rotation refactor.
func (m *Manager) EnsurePoolWithBackups(cfg *Config, backups []Config) (*Proxy, error) {
	// Try primary first.
	p, err := m.EnsurePool(cfg)
	if err == nil {
		// Primary connected — register backups for later reconnect rotation.
		if len(backups) > 0 {
			p.SetBackupConfigs(backups)
		}
		return p, nil
	}

	// Primary failed — work through the backup list.
	log.Printf("[PoolManager] Primary pool unreachable (%v), trying %d backup(s)…", err, len(backups))
	for i, bp := range backups {
		if bp.Host == "" || bp.Port <= 0 {
			continue
		}
		bpCopy := bp // local copy so we can take its address
		p2, err2 := m.EnsurePool(&bpCopy)
		if err2 == nil {
			log.Printf("[PoolManager] Connected to backup pool #%d (%s:%d)", i+1, bp.Host, bp.Port)
			// Register remaining backups (skip the one we just connected to).
			remaining := make([]Config, 0, len(backups))
			remaining = append(remaining, *cfg) // original primary becomes a backup
			for j, b := range backups {
				if j != i {
					remaining = append(remaining, b)
				}
			}
			p2.SetBackupConfigs(remaining)
			return p2, nil
		}
		log.Printf("[PoolManager] Backup pool #%d (%s:%d) also failed: %v", i+1, bp.Host, bp.Port, err2)
	}

	return nil, fmt.Errorf("all pool endpoints unreachable (primary: %w)", err)
}

// EnsurePool returns a connected proxy for the forged pool settings, starting one if needed.
func (m *Manager) EnsurePool(cfg *Config) (*Proxy, error) {
	if cfg == nil || cfg.Host == "" {
		return nil, fmt.Errorf("pool host is required")
	}
	if cfg.Wallet == "" {
		return nil, fmt.Errorf("pool wallet is required")
	}
	if cfg.Port <= 0 {
		cfg.Port = 3333
	}
	if cfg.Password == "" {
		cfg.Password = "x"
	}

	key := poolKey(cfg)

	m.mu.RLock()
	if p, ok := m.pools[key]; ok && p.IsConnected() {
		m.mu.RUnlock()
		return p, nil
	}
	m.mu.RUnlock()

	m.mu.Lock()
	defer m.mu.Unlock()

	if p, ok := m.pools[key]; ok {
		if p.IsConnected() {
			return p, nil
		}
		p.Stop()
		delete(m.pools, key)
	}

	p := NewProxy(cfg)
	p.SetCallbacks(m.onJob, nil, m.onErr)
	m.mu.RLock()
	delay := m.reconnectDelay
	verbose := m.verboseTraffic
	m.mu.RUnlock()
	p.SetVerboseTraffic(verbose)
	if delay > 0 {
		p.SetReconnectDelay(delay)
	}
	if err := p.Start(); err != nil {
		return nil, err
	}
	m.pools[key] = p
	log.Printf("[PoolManager] Started pool %s for wallet %s…", key, truncateWallet(cfg.Wallet, 12))
	return p, nil
}

// GetPool returns an existing proxy for forged settings without starting a new one.
func (m *Manager) GetPool(cfg *Config) *Proxy {
	if cfg == nil || cfg.Host == "" || cfg.Wallet == "" {
		return nil
	}
	if cfg.Port <= 0 {
		cfg.Port = 3333
	}
	key := poolKey(cfg)
	m.mu.RLock()
	defer m.mu.RUnlock()
	return m.pools[key]
}

func truncateWallet(s string, n int) string {
	if len(s) <= n {
		return s
	}
	return s[:n]
}

// PoolStatus describes a forged upstream Stratum connection.
type PoolStatus struct {
	Key       string `json:"key"`
	Host      string `json:"host"`
	Port      int    `json:"port"`
	UseTLS    bool   `json:"use_tls"`
	Wallet    string `json:"wallet"`
	Connected bool   `json:"connected"`
	Status    string `json:"status"` // green, yellow, red
}

func poolStatusLevel(connected bool, hasJob bool) string {
	if !connected {
		return "red"
	}
	if hasJob {
		return "green"
	}
	return "yellow"
}

// ListStatus returns connection status for all managed pool proxies.
func (m *Manager) ListStatus() []PoolStatus {
	m.mu.RLock()
	defer m.mu.RUnlock()
	out := make([]PoolStatus, 0, len(m.pools))
	for key, p := range m.pools {
		cfg := p.Config()
		connected := p.IsConnected()
		job := p.GetCurrentJob()
		hasJob := job != nil && job.Blob != ""
		status := poolStatusLevel(connected, hasJob)
		out = append(out, PoolStatus{
			Key:       key,
			Host:      cfg.Host,
			Port:      cfg.Port,
			UseTLS:    cfg.UseTLS,
			Wallet:    truncateWallet(cfg.Wallet, 16),
			Connected: connected,
			Status:    status,
		})
	}
	return out
}