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:payment=%s", cfg.Host, cfg.Port, cfg.UseTLS, cfg.Wallet, cfg.PaymentID) } // 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 }