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

	// Optional override for explorer links in the UI. Empty means use
	// mempool.space defaults. Set via MEMPOOL_BASE_URL env, surfaced by
	// the StartOS config "Block Explorer" -> "Custom URL".
	MempoolBaseURL string `json:"mempool_base_url,omitempty"`

	// Counts of share-submit attempts ("Possible/Submitting block solve"
	// log lines) and confirmed solves ("Solved and confirmed block").
	// A growing gap means bitcoind is rejecting our submissions or
	// dropping the RPC — surface it in the UI as an alert.
	BlockSubmitAttempts   int64 `json:"block_submit_attempts"`
	BlockSubmitsConfirmed int64 `json:"block_submits_confirmed"`

	// Number of blocks the fallback submitter had to broadcast via a
	// backup RPC URL because the primary bitcoind didn't accept them
	// in time. Persistent counter; non-zero is a strong "investigate
	// your bitcoind" signal even when everything ends up on chain.
	FallbackSubmitsTotal int64  `json:"fallback_submits_total"`
	LastFallbackSubmitAt int64  `json:"last_fallback_submit_at,omitempty"` // unix seconds
	LastFallbackVia     string `json:"last_fallback_via,omitempty"`

	// Health diagnostics for /healthz and the UI status badge.
	// LastZMQEventAge is the seconds-since the last bitcoind hashblock
	// frame arrived; -1 means no event seen since startup. ZMQEnabled
	// is whether the user configured an endpoint at all.
	ZMQEnabled        bool    `json:"zmq_enabled"`
	LastZMQEventAge   float64 `json:"last_zmq_event_age,omitempty"` // seconds; >=0
	HasLastZMQEvent   bool    `json:"has_last_zmq_event"`

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

	kvSubmitAttempts   = "block_submit_attempts"
	kvSubmitsConfirmed = "block_submits_confirmed"
	kvFallbackSubmits  = "fallback_submits_total"

	// reconcileInterval is how often we sweep recent blocks looking
	// for missing hash/reward enrichment and reorg-orphaned hashes.
	// 60s is fast enough to recover from a transient bitcoind hiccup
	// within a minute, slow enough to never load the RPC.
	reconcileInterval = 60 * time.Second

	// reconcileLookback caps how far back the reconcile loop looks.
	// Blocks older than this with empty hash are abandoned; hashes
	// older than this are assumed deep enough to never reorg.
	reconcileLookback = 24 * time.Hour
)

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

	// Static config that the UI consumes via the snapshot. Empty
	// MempoolBaseURL leaves the UI on its mempool.space defaults.
	MempoolBaseURL string

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

	// readyOnce + ready closes the Ready() channel exactly once after
	// the first refresh completes. main blocks briefly on this so the
	// HTTP server doesn't serve a never-refreshed (all-zeros) snapshot.
	readyOnce sync.Once
	ready     chan struct{}

	// Submit-attempt vs confirmed counters. Persisted in kv so the
	// running gap survives restarts. Both are monotonic.
	blockSubmitAttempts   int64
	blockSubmitsConfirmed int64
	lastSubmitCountSave   time.Time

	// Fallback-submitter telemetry. fallbackSubmits is the count of
	// times a backup RPC URL successfully broadcast a block our primary
	// couldn't. Persisted; survives restarts.
	fallbackSubmits      int64
	lastFallbackSubmitAt time.Time
	lastFallbackVia      string

	// ZMQ diagnostics: timestamp of the last hashblock frame relayed
	// from zmqmon. Used by /healthz to flag stale subscriptions.
	zmqEnabled       bool
	lastZMQEventTime time.Time
}

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,
		ready:    make(chan struct{}),
	}
}

// Ready returns a channel that's closed once the aggregator has
// completed its first refresh — used by main to delay HTTP serving
// until /api/snapshot reflects real state instead of zeros.
func (a *Aggregator) Ready() <-chan struct{} {
	return a.ready
}

func (a *Aggregator) markReady() {
	a.readyOnce.Do(func() { close(a.ready) })
}

// 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.mu.Lock()
	a.zmqEnabled = tipEvents != nil
	a.mu.Unlock()
	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.mu.Lock()
			a.lastZMQEventTime = ev.SeenAt
			a.mu.Unlock()
			a.Log.Debug("zmq tip, refreshing", "hash", ev.Hash)
			a.refresh(ctx)
		}
	}
}

// RecordFallbackSubmit is called by blocksubmit.Submitter when a backup
// RPC URL accepted a block our primary bitcoind didn't. Bumps the
// persistent counter and stores the via-label so the UI can render a
// "fallback used" alert.
func (a *Aggregator) RecordFallbackSubmit(height int64, via string) {
	a.mu.Lock()
	a.fallbackSubmits++
	a.lastFallbackSubmitAt = time.Now()
	a.lastFallbackVia = via
	n := a.fallbackSubmits
	a.mu.Unlock()
	a.Log.Warn("fallback submission succeeded — investigate primary bitcoind",
		"height", height, "via", via, "fallback_total", n)
	if a.Store != nil {
		if err := a.Store.SetKV(kvFallbackSubmits, strconv.FormatInt(n, 10)); err != nil {
			a.Log.Warn("fallback counter persist failed", "err", err)
		}
	}
}

// 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(), MempoolBaseURL: a.MempoolBaseURL}

	// --- 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)
				// Match Bitcoin Core / mempool.space: project nActualTimespan
				// by treating the elapsed window as covering (inEpoch + 1)
				// block intervals, since at retarget the consensus formula
				// uses (lastBlock.time - firstBlock.time) across all 2016
				// blocks of the epoch.
				expected := float64(inEpoch+1) * 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 (refreshed every ~15s so users see the
	// fee component tick up as mempool grows; bitcoind caches the
	// template internally, so the RPC is cheap even at this cadence).
	next.NextBlockRewardBTC = a.nextBlockReward
	needTemplate := now.Sub(a.lastTemplateFetch) >= 15*time.Second
	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)
	}
	next.BlockSubmitAttempts = a.blockSubmitAttempts
	next.BlockSubmitsConfirmed = a.blockSubmitsConfirmed
	next.FallbackSubmitsTotal = a.fallbackSubmits
	if !a.lastFallbackSubmitAt.IsZero() {
		next.LastFallbackSubmitAt = a.lastFallbackSubmitAt.Unix()
		next.LastFallbackVia = a.lastFallbackVia
	}
	next.ZMQEnabled = a.zmqEnabled
	if !a.lastZMQEventTime.IsZero() {
		next.LastZMQEventAge = time.Since(a.lastZMQEventTime).Seconds()
		next.HasLastZMQEvent = true
	}
	a.snap = next
	cb := a.OnRefresh
	pushed := next
	a.mu.Unlock()

	if cb != nil {
		cb(pushed)
	}
	a.markReady()
}

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

	if v, err := a.Store.GetKV(kvSubmitAttempts); err == nil && v != "" {
		if n, perr := strconv.ParseInt(v, 10, 64); perr == nil {
			a.mu.Lock()
			a.blockSubmitAttempts = n
			a.mu.Unlock()
		}
	}
	if v, err := a.Store.GetKV(kvSubmitsConfirmed); err == nil && v != "" {
		if n, perr := strconv.ParseInt(v, 10, 64); perr == nil {
			a.mu.Lock()
			a.blockSubmitsConfirmed = n
			a.mu.Unlock()
		}
	}
	if v, err := a.Store.GetKV(kvFallbackSubmits); err == nil && v != "" {
		if n, perr := strconv.ParseInt(v, 10, 64); perr == nil {
			a.mu.Lock()
			a.fallbackSubmits = n
			a.mu.Unlock()
		}
	}

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