KamadoPool/api/internal/state/aggregator.go

// Package state merges data from CKPool (via Unix socket), Bitcoin Core
// (via JSON-RPC), and future sources (ZMQ, log tailer) into a single
// thread-safe snapshot the HTTP layer can serve. The snapshot is refreshed
// on a ticker; readers get a copy without blocking the refresh.
package state

import (
	"context"
	"log/slog"
	"sync"
	"time"

	"github.com/kamadopool/kamado-api/internal/bitcoind"
	"github.com/kamadopool/kamado-api/internal/ckpool"
	"github.com/kamadopool/kamado-api/internal/store"
	"github.com/kamadopool/kamado-api/internal/zmqmon"
)

// Snapshot is the merged view served to the UI. All fields are safe to
// JSON-serialize directly.
type Snapshot struct {
	GeneratedAt time.Time `json:"generated_at"`

	// CKPool-derived fields
	Pool    *ckpool.PoolStats      `json:"pool"`
	Uptime  int64                  `json:"uptime_seconds"`
	Users   []ckpool.User          `json:"users"`
	Workers []ckpool.Worker        `json:"workers"`
	Clients []ckpool.StratumClient `json:"clients"`

	// Derived/enriched fields
	HashrateHs    float64 `json:"hashrate_hs_1m"`    // from PoolStats.DSPS1
	HashrateHs5m  float64 `json:"hashrate_hs_5m"`
	HashrateHs1h  float64 `json:"hashrate_hs_1h"`
	HashrateHs24h float64 `json:"hashrate_hs_24h"`

	// Bitcoin Core fields
	Chain                *bitcoind.BlockchainInfo `json:"chain"`
	NetworkHashrateHs    float64                  `json:"network_hashrate_hs"`

	// Recent found blocks (in-memory history; persisted in Phase 2b.5).
	RecentBlocks []BlockRecord `json:"recent_blocks,omitempty"`

	// Health
	CKPoolOK  bool   `json:"ckpool_ok"`
	BitcoinOK bool   `json:"bitcoin_ok"`
	LastError string `json:"last_error,omitempty"`
}

// Aggregator refreshes a Snapshot on a ticker.
type Aggregator struct {
	CK       *ckpool.Client
	RPC      *bitcoind.RPC
	Store    *store.BlockStore // optional; nil disables persistence
	Interval time.Duration
	Log      *slog.Logger

	// OnRefresh, if set, is called (non-blocking) after each snapshot
	// refresh. Used by the WebSocket hub to push updates to clients.
	OnRefresh func(Snapshot)

	mu     sync.RWMutex
	snap   Snapshot
	blocks []BlockRecord

	// ckFailStreak counts consecutive refreshes where CKPool returned an
	// error. The first failure after a streak of successes is logged at
	// DEBUG (likely a transient warm-up or lock hiccup); repeated failures
	// escalate to WARN.
	ckFailStreak int
}

func New(ck *ckpool.Client, rpc *bitcoind.RPC, interval time.Duration, log *slog.Logger) *Aggregator {
	return &Aggregator{
		CK:       ck,
		RPC:      rpc,
		Interval: interval,
		Log:      log,
	}
}

// Run blocks until ctx is cancelled, refreshing the snapshot every Interval.
// It runs one immediate refresh at startup so readers don't see an empty
// snapshot after ctx launches the goroutine. Persisted block history is
// loaded from the store before the first refresh. If tipEvents is non-nil,
// each received tip triggers an immediate refresh outside the poll cadence.
func (a *Aggregator) Run(ctx context.Context, tipEvents <-chan zmqmon.TipEvent) {
	a.loadPersistedBlocks()
	a.refresh(ctx)
	t := time.NewTicker(a.Interval)
	defer t.Stop()
	for {
		select {
		case <-ctx.Done():
			return
		case <-t.C:
			a.refresh(ctx)
		case ev, ok := <-tipEvents:
			if !ok {
				tipEvents = nil
				continue
			}
			a.Log.Debug("zmq tip, refreshing", "hash", ev.Hash)
			a.refresh(ctx)
		}
	}
}

// Snapshot returns a copy of the current snapshot.
func (a *Aggregator) Snapshot() Snapshot {
	a.mu.RLock()
	defer a.mu.RUnlock()
	return a.snap
}

func (a *Aggregator) refresh(ctx context.Context) {
	ctx, cancel := context.WithTimeout(ctx, 10*time.Second)
	defer cancel()

	next := Snapshot{GeneratedAt: time.Now()}

	// --- ckpool: poolstats, users, workers, clients, uptime ---
	if ps, err := a.CK.PoolStats(ctx); err == nil {
		next.Pool = ps
		next.HashrateHs = ckpool.DSPSToHashrate(ps.DSPS1)
		next.HashrateHs5m = ckpool.DSPSToHashrate(ps.DSPS5)
		next.HashrateHs1h = ckpool.DSPSToHashrate(ps.DSPS60)
		next.HashrateHs24h = ckpool.DSPSToHashrate(ps.DSPS1440)
		next.CKPoolOK = true
		a.ckFailStreak = 0
	} else {
		a.ckFailStreak++
		// First failure or two: likely transient (stratifier warm-up or
		// a lock hiccup during a reconnect). Only escalate to WARN after
		// three consecutive failures.
		if a.ckFailStreak >= 3 {
			a.Log.Warn("ckpool poolstats failed", "err", err, "streak", a.ckFailStreak)
		} else {
			a.Log.Debug("ckpool poolstats transient error", "err", err, "streak", a.ckFailStreak)
		}
		next.LastError = err.Error()
	}
	if u, err := a.CK.Uptime(ctx); err == nil {
		next.Uptime = u
	}
	if us, err := a.CK.Users(ctx); err == nil {
		next.Users = us
	}
	if ws, err := a.CK.Workers(ctx); err == nil {
		next.Workers = ws
	}
	if cs, err := a.CK.Clients(ctx); err == nil {
		next.Clients = cs
	}

	// --- bitcoind: chain + network hashrate ---
	if bi, err := a.RPC.GetBlockchainInfo(ctx); err == nil {
		next.Chain = bi
		next.BitcoinOK = true
		if nh, err := a.RPC.GetNetworkHashPS(ctx, -1, int(bi.Blocks)); err == nil {
			next.NetworkHashrateHs = nh
		}
	} else {
		a.Log.Warn("bitcoind getblockchaininfo failed", "err", err)
		if next.LastError == "" {
			next.LastError = err.Error()
		}
	}

	a.mu.Lock()
	// Attach current block history so /api/snapshot and WebSocket
	// pushes carry the same view.
	if len(a.blocks) > 0 {
		next.RecentBlocks = make([]BlockRecord, len(a.blocks))
		copy(next.RecentBlocks, a.blocks)
	}
	a.snap = next
	cb := a.OnRefresh
	pushed := next
	a.mu.Unlock()

	if cb != nil {
		cb(pushed)
	}
}