package state import ( "context" "time" "github.com/kamadopool/kamado-api/internal/logmon" "github.com/kamadopool/kamado-api/internal/store" ) // BlockRecord is a found block, merged from a logmon event with bitcoind // data if available. Hash and Reward are populated best-effort via the // RPC lookup scheduled right after the log line is seen. type BlockRecord struct { Height int64 `json:"height"` Hash string `json:"hash,omitempty"` RewardBT float64 `json:"reward_btc,omitempty"` FoundAt time.Time `json:"found_at"` Source string `json:"source"` // "logmon" for now; "zmq" later } // IngestBlockEvents reads block events from the tailer and appends them // to the snapshot's block history. Runs until ctx is cancelled. func (a *Aggregator) IngestBlockEvents(ctx context.Context, events <-chan logmon.BlockEvent) { for { select { case <-ctx.Done(): return case ev, ok := <-events: if !ok { return } rec := BlockRecord{ Height: ev.Height, FoundAt: ev.SeenAt, Source: "logmon", } // Best-effort enrich with hash via bitcoind. if a.RPC != nil { lookupCtx, cancel := context.WithTimeout(ctx, 5*time.Second) if hash, err := a.RPC.GetBlockHash(lookupCtx, ev.Height); err == nil { rec.Hash = hash } else { a.Log.Warn("bitcoind getblockhash failed", "height", ev.Height, "err", err) } cancel() } if a.Store != nil { if err := a.Store.InsertBlock(store.Block{ Height: rec.Height, Hash: rec.Hash, RewardBT: rec.RewardBT, FoundAt: rec.FoundAt, Source: rec.Source, }); err != nil { a.Log.Warn("block persist failed", "height", rec.Height, "err", err) } } pushed := a.appendBlock(rec) a.Log.Info("block recorded", "height", rec.Height, "hash", rec.Hash) // Push immediately so WebSocket clients see the solve // without waiting for the next poll tick. if a.OnRefresh != nil { a.OnRefresh(pushed) } } } } // maxBlockHistory caps in-memory block history. Persistence comes in // Phase 2b.5 via SQLite; for now recent blocks survive only this // process's lifetime. const maxBlockHistory = 256 // appendBlock records a new block and returns a copy of the current // snapshot with the updated history attached, suitable for an immediate // WebSocket broadcast. func (a *Aggregator) appendBlock(rec BlockRecord) Snapshot { a.mu.Lock() defer a.mu.Unlock() a.blocks = append(a.blocks, rec) if len(a.blocks) > maxBlockHistory { a.blocks = a.blocks[len(a.blocks)-maxBlockHistory:] } snap := a.snap snap.RecentBlocks = make([]BlockRecord, len(a.blocks)) copy(snap.RecentBlocks, a.blocks) a.snap = snap return snap } // loadPersistedBlocks seeds a.blocks from the store so history survives // restarts. Safe to call with a nil Store — becomes a no-op. func (a *Aggregator) loadPersistedBlocks() { if a.Store == nil { return } rows, err := a.Store.Recent(maxBlockHistory) if err != nil { a.Log.Warn("block history load failed", "err", err) return } // Store returns newest-first; a.blocks is newest-last. recs := make([]BlockRecord, 0, len(rows)) for i := len(rows) - 1; i >= 0; i-- { r := rows[i] recs = append(recs, BlockRecord{ Height: r.Height, Hash: r.Hash, RewardBT: r.RewardBT, FoundAt: r.FoundAt, Source: r.Source, }) } a.mu.Lock() a.blocks = recs a.mu.Unlock() a.Log.Info("block history loaded", "count", len(recs)) } // Blocks returns a copy of the recent block history, newest last. func (a *Aggregator) Blocks() []BlockRecord { a.mu.RLock() defer a.mu.RUnlock() out := make([]BlockRecord, len(a.blocks)) copy(out, a.blocks) return out }