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"
	"strconv"
	"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"
)

// HashratePoint is a single timestamped hashrate sample for the 24h chart.
type HashratePoint struct {
	T int64   `json:"t"` // unix seconds
	V float64 `json:"v"` // H/s
}

// 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"`

	// Best share difficulty ever seen across all workers.
	BestDiff float64 `json:"best_diff"`

	// Cumulative work done by the pool across its entire lifetime, in
	// diff-1-normalized shares (multiply by 2^32 for total hashes).
	// Survives ckpool restarts via kv-store persistence.
	CumulativeShares float64 `json:"cumulative_shares"`

	// Next-block reward (subsidy + fees) from bitcoind getblocktemplate,
	// in BTC. Refreshed at most once per minute.
	NextBlockRewardBTC float64 `json:"next_block_reward_btc"`

	// Predicted percent change at the next difficulty retarget, clamped
	// to Bitcoin's consensus range of [-75, +300]. Derived from the
	// observed block interval since the current epoch started.
	NextDifficultyPercent float64 `json:"next_difficulty_percent"`

	// 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"`

	// 24-hour hashrate history, sampled once per minute (max 1440 points).
	HashrateHistory []HashratePoint `json:"hashrate_history,omitempty"`

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

const (
	maxHistoryPoints = 1440 // 24h at 1 sample/min

	// kvCumulativeWork is versioned: "cumulative_shares" (v1) accidentally
	// summed pool.shares (raw per-share count), which is meaningless for
	// hashrate math. "cumulative_work" (v2) sums pool.accepted — the
	// diff-1-normalized work, so cumulative_work * 2^32 is real hashes.
	// The old key is orphaned in the kv table on upgrade; harmless.
	kvCumulativeWork = "cumulative_work"
)

// 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

	// 24h hashrate history ring buffer, sampled once per minute.
	hrHistory      []HashratePoint
	lastHRSampleAt time.Time

	// Cumulative pool work in diff-1-normalized shares, surviving
	// ckpool restarts. Maintained by re-reading pool.Shares each
	// refresh and integrating only the positive delta — a decrease
	// means ckpool restarted and its counter reset to 0, so we reset
	// our delta baseline without losing the accumulated total.
	//
	// hasPoolSharesBaseline guards against double-counting on api
	// restart: cumulativeShares is loaded from disk, and the first
	// post-load observation of pool.Shares only establishes the
	// baseline — the current ckpool counter was already accounted for
	// before we crashed.
	cumulativeShares       float64
	lastPoolShares         float64
	hasPoolSharesBaseline  bool
	lastCumulativeSave     time.Time

	// Next-block reward cache; refreshed on its own cadence so we don't
	// hammer bitcoind with getblocktemplate on every poll tick.
	nextBlockReward   float64
	lastTemplateFetch time.Time

	// Retarget epoch-start timestamp cache. Updated when we cross a
	// new retarget boundary (every 2016 blocks); within an epoch we
	// just re-use the cached value and re-extrapolate each refresh.
	retargetEpoch     int64
	retargetStartUnix int64

	// 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.loadPersistedState()
	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()
		}
	}

	// --- predicted difficulty adjustment ---
	// Fetch the timestamp of the first block in the current retarget
	// epoch (height - height%2016) once per epoch and cache it, then
	// compute (expected / elapsed - 1) * 100 where expected = blocks_in *
	// 600 seconds. Bitcoin clamps the consensus adjustment factor to
	// [1/4, 4x], i.e. [-75%, +300%].
	if next.Chain != nil && next.Chain.Blocks > 0 {
		height := next.Chain.Blocks
		epoch := height / 2016
		inEpoch := height % 2016
		if inEpoch > 0 {
			if a.retargetEpoch != epoch || a.retargetStartUnix == 0 {
				startHeight := epoch * 2016
				lookupCtx, cancelLookup := context.WithTimeout(ctx, 3*time.Second)
				if hash, err := a.RPC.GetBlockHash(lookupCtx, startHeight); err == nil {
					if blk, err := a.RPC.GetBlock(lookupCtx, hash); err == nil {
						a.retargetEpoch = epoch
						a.retargetStartUnix = blk.Time
					}
				}
				cancelLookup()
			}
			if a.retargetStartUnix > 0 {
				elapsed := float64(time.Now().Unix() - a.retargetStartUnix)
				expected := float64(inEpoch) * 600
				if elapsed > 0 {
					factor := expected / elapsed
					if factor > 4 {
						factor = 4
					}
					if factor < 0.25 {
						factor = 0.25
					}
					next.NextDifficultyPercent = (factor - 1) * 100
				}
			}
		}
	}

	// Compute pool-wide best-ever share diff from workers.
	for _, w := range next.Workers {
		d := w.BestEver
		if d == 0 {
			d = w.BestDiff
		}
		if d > next.BestDiff {
			next.BestDiff = d
		}
	}

	a.mu.Lock()
	now := time.Now()

	// --- cumulative work tracking ---
	// Integrate only positive deltas on pool.Accepted, which is ckpool's
	// accounted_diff_shares — the sum of each accepted share's
	// difficulty, in diff-1-normalized units. pool.Shares is the raw
	// per-share count and is useless for hashrate math. A decrease
	// means ckpool restarted (or its state reset); we reset the
	// baseline without touching the accumulated total. The first
	// observation after load only establishes the baseline — the
	// current counter was already integrated before we shut down.
	if next.Pool != nil {
		cur := float64(next.Pool.Accepted)
		if a.hasPoolSharesBaseline && cur >= a.lastPoolShares {
			a.cumulativeShares += cur - a.lastPoolShares
		}
		a.lastPoolShares = cur
		a.hasPoolSharesBaseline = true
	}
	next.CumulativeShares = a.cumulativeShares

	// Persist cumulative_shares at most once per minute.
	if a.Store != nil && now.Sub(a.lastCumulativeSave) >= time.Minute {
		val := strconv.FormatFloat(a.cumulativeShares, 'f', -1, 64)
		if err := a.Store.SetKV(kvCumulativeWork, val); err != nil {
			a.Log.Warn("cumulative_shares persist failed", "err", err)
		}
		a.lastCumulativeSave = now
	}

	// --- hashrate history sample (once per minute) ---
	if now.Sub(a.lastHRSampleAt) >= time.Minute {
		p := HashratePoint{T: now.Unix(), V: next.HashrateHs}
		a.hrHistory = append(a.hrHistory, p)
		if len(a.hrHistory) > maxHistoryPoints {
			a.hrHistory = a.hrHistory[len(a.hrHistory)-maxHistoryPoints:]
		}
		a.lastHRSampleAt = now
		if a.Store != nil {
			if err := a.Store.InsertHashrateSample(p.T, p.V); err != nil {
				a.Log.Warn("hashrate persist failed", "err", err)
			}
			// Keep the persisted window bounded to 24h + a small slack.
			cutoff := now.Add(-25 * time.Hour).Unix()
			_ = a.Store.PruneHashrateBefore(cutoff)
		}
	}
	if len(a.hrHistory) > 0 {
		next.HashrateHistory = make([]HashratePoint, len(a.hrHistory))
		copy(next.HashrateHistory, a.hrHistory)
	}

	// --- next-block reward (at most once per minute) ---
	next.NextBlockRewardBTC = a.nextBlockReward
	needTemplate := now.Sub(a.lastTemplateFetch) >= time.Minute
	a.mu.Unlock()

	if needTemplate && a.RPC != nil && next.BitcoinOK {
		tplCtx, cancelTpl := context.WithTimeout(ctx, 5*time.Second)
		tpl, err := a.RPC.GetBlockTemplate(tplCtx)
		cancelTpl()
		if err == nil && tpl != nil {
			reward := float64(tpl.CoinbaseValue) / 1e8
			a.mu.Lock()
			a.nextBlockReward = reward
			a.lastTemplateFetch = now
			next.NextBlockRewardBTC = reward
			a.mu.Unlock()
		} else if err != nil {
			a.Log.Debug("getblocktemplate failed", "err", err)
		}
	}

	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)
	}
}

// loadPersistedState restores cumulative work and hashrate history from
// the store so they survive process restarts. Safe to call with a nil
// Store — becomes a no-op.
func (a *Aggregator) loadPersistedState() {
	if a.Store == nil {
		return
	}

	if v, err := a.Store.GetKV(kvCumulativeWork); err != nil {
		a.Log.Warn("cumulative_shares load failed", "err", err)
	} else if v != "" {
		if f, perr := strconv.ParseFloat(v, 64); perr == nil {
			a.mu.Lock()
			a.cumulativeShares = f
			a.mu.Unlock()
			a.Log.Info("cumulative_shares loaded", "value", f)
		}
	}

	cutoff := time.Now().Add(-24 * time.Hour).Unix()
	if samples, err := a.Store.HashrateSince(cutoff); err != nil {
		a.Log.Warn("hashrate history load failed", "err", err)
	} else if len(samples) > 0 {
		hist := make([]HashratePoint, 0, len(samples))
		for _, s := range samples {
			hist = append(hist, HashratePoint{T: s.T, V: s.V})
		}
		a.mu.Lock()
		a.hrHistory = hist
		a.lastHRSampleAt = time.Unix(hist[len(hist)-1].T, 0)
		a.mu.Unlock()
		a.Log.Info("hashrate history loaded", "count", len(hist))
	}
}