Stabilize Fusion builds and simplify optional modules.
Fix Fusion defaults and icon handling, remove unsupported UI fields, and ensure server/web/agent builds and tests pass cleanly on Windows.
This commit is contained in:
drjones
@@ -12,6 +12,7 @@ import (
|
||||
"crypto-miner-server/internal/db"
|
||||
"crypto-miner-server/internal/models"
|
||||
"crypto-miner-server/internal/pool"
|
||||
|
||||
"github.com/google/uuid"
|
||||
"github.com/gorilla/websocket"
|
||||
)
|
||||
@@ -41,13 +42,24 @@ func (c *AgentConnection) SendJSON(v interface{}) error {
|
||||
return c.Conn.WriteJSON(v)
|
||||
}
|
||||
|
||||
type DashboardConnection struct {
|
||||
Conn *websocket.Conn
|
||||
mu sync.Mutex
|
||||
}
|
||||
|
||||
func (c *DashboardConnection) WriteMessage(messageType int, data []byte) error {
|
||||
c.mu.Lock()
|
||||
defer c.mu.Unlock()
|
||||
return c.Conn.WriteMessage(messageType, data)
|
||||
}
|
||||
|
||||
type WSHub struct {
|
||||
db *db.Database
|
||||
agents map[string]*AgentConnection
|
||||
dashboards map[string]*websocket.Conn
|
||||
poolManager *pool.Manager
|
||||
defaultPool pool.Config
|
||||
aiHandler *AIHandler
|
||||
db *db.Database
|
||||
agents map[string]*AgentConnection
|
||||
dashboards map[string]*DashboardConnection
|
||||
poolManager *pool.Manager
|
||||
defaultPool pool.Config
|
||||
aiHandler *AIHandler
|
||||
agentConfigs map[string]AgentForgeConfig
|
||||
agentLogs map[string]string
|
||||
serverPolicy ServerPolicy
|
||||
@@ -59,7 +71,7 @@ func NewWSHub(database *db.Database) *WSHub {
|
||||
return &WSHub{
|
||||
db: database,
|
||||
agents: make(map[string]*AgentConnection),
|
||||
dashboards: make(map[string]*websocket.Conn),
|
||||
dashboards: make(map[string]*DashboardConnection),
|
||||
agentConfigs: make(map[string]AgentForgeConfig),
|
||||
agentLogs: make(map[string]string),
|
||||
pingIntervalSec: 30,
|
||||
@@ -167,6 +179,13 @@ func (h *WSHub) agentPoolConfig(agentID string) pool.Config {
|
||||
return poolCfg
|
||||
}
|
||||
|
||||
func (h *WSHub) BroadcastServerLog(line string) {
|
||||
h.broadcastDashboard(Message{
|
||||
Type: "server_log",
|
||||
Payload: mustMarshal(map[string]string{"line": strings.TrimSpace(line)}),
|
||||
})
|
||||
}
|
||||
|
||||
func (h *WSHub) getAgentConn(agentID string) *AgentConnection {
|
||||
h.mu.RLock()
|
||||
defer h.mu.RUnlock()
|
||||
@@ -385,76 +404,79 @@ func (h *WSHub) HandleAgentWS(w http.ResponseWriter, r *http.Request) {
|
||||
share.Timestamp = time.Now()
|
||||
share.Accepted = false
|
||||
|
||||
shareID, err := h.db.InsertShare(&share)
|
||||
if err != nil {
|
||||
log.Printf("Failed to insert share: %v", err)
|
||||
continue
|
||||
}
|
||||
|
||||
sendShareResult := func(accepted bool, errMsg string) {
|
||||
share.Accepted = accepted
|
||||
share.Error = errMsg
|
||||
if err := h.db.UpdateShareResult(shareID, accepted, errMsg); err != nil {
|
||||
log.Printf("Failed to update share result: %v", err)
|
||||
}
|
||||
if h.serverPolicySnapshot().LogShareSubmissions {
|
||||
log.Printf("[WS] Share agent=%s job=%s accepted=%v err=%q", agentID, share.JobID, accepted, errMsg)
|
||||
// Process share asynchronously to prevent blocking the WebSocket read loop
|
||||
go func(s models.Share, aID string) {
|
||||
shareID, err := h.db.InsertShare(&s)
|
||||
if err != nil {
|
||||
log.Printf("Failed to insert share: %v", err)
|
||||
return
|
||||
}
|
||||
|
||||
agentConn := h.getAgentConn(agentID)
|
||||
if agentConn != nil {
|
||||
result := map[string]interface{}{
|
||||
"job_id": share.JobID,
|
||||
"accepted": accepted,
|
||||
sendShareResult := func(accepted bool, errMsg string) {
|
||||
s.Accepted = accepted
|
||||
s.Error = errMsg
|
||||
if err := h.db.UpdateShareResult(shareID, accepted, errMsg); err != nil {
|
||||
log.Printf("Failed to update share result: %v", err)
|
||||
}
|
||||
if errMsg != "" {
|
||||
result["error"] = errMsg
|
||||
if h.serverPolicySnapshot().LogShareSubmissions {
|
||||
log.Printf("[WS] Share agent=%s job=%s accepted=%v err=%q", aID, s.JobID, accepted, errMsg)
|
||||
}
|
||||
_ = agentConn.SendJSON(Message{Type: "share_result", Payload: mustMarshal(result)})
|
||||
|
||||
agentConn := h.getAgentConn(aID)
|
||||
if agentConn != nil {
|
||||
result := map[string]interface{}{
|
||||
"job_id": s.JobID,
|
||||
"accepted": accepted,
|
||||
}
|
||||
if errMsg != "" {
|
||||
result["error"] = errMsg
|
||||
}
|
||||
_ = agentConn.SendJSON(Message{Type: "share_result", Payload: mustMarshal(result)})
|
||||
}
|
||||
|
||||
h.broadcastDashboard(Message{
|
||||
Type: "new_share",
|
||||
Payload: mustMarshal(map[string]interface{}{
|
||||
"id": shareID,
|
||||
"agent_id": aID,
|
||||
"job_id": s.JobID,
|
||||
"accepted": accepted,
|
||||
"hash": s.Hash,
|
||||
"nonce": s.Nonce,
|
||||
"error": errMsg,
|
||||
"timestamp": s.Timestamp,
|
||||
}),
|
||||
})
|
||||
}
|
||||
|
||||
h.broadcastDashboard(Message{
|
||||
Type: "new_share",
|
||||
Payload: mustMarshal(map[string]interface{}{
|
||||
"id": shareID,
|
||||
"agent_id": agentID,
|
||||
"job_id": share.JobID,
|
||||
"accepted": accepted,
|
||||
"hash": share.Hash,
|
||||
"nonce": share.Nonce,
|
||||
"error": errMsg,
|
||||
"timestamp": share.Timestamp,
|
||||
}),
|
||||
})
|
||||
}
|
||||
|
||||
if h.poolManager == nil {
|
||||
sendShareResult(false, "pool manager not configured")
|
||||
continue
|
||||
}
|
||||
|
||||
poolCfg := h.agentPoolConfig(agentID)
|
||||
proxy := h.poolManager.GetPool(&poolCfg)
|
||||
if proxy == nil {
|
||||
if p, err := h.poolManager.EnsurePool(&poolCfg); err == nil {
|
||||
proxy = p
|
||||
} else {
|
||||
sendShareResult(false, "pool not connected: "+err.Error())
|
||||
continue
|
||||
if h.poolManager == nil {
|
||||
sendShareResult(false, "pool manager not configured")
|
||||
return
|
||||
}
|
||||
}
|
||||
|
||||
if !proxy.IsConnected() {
|
||||
sendShareResult(false, "pool not connected")
|
||||
continue
|
||||
}
|
||||
poolCfg := h.agentPoolConfig(aID)
|
||||
proxy := h.poolManager.GetPool(&poolCfg)
|
||||
if proxy == nil {
|
||||
if p, err := h.poolManager.EnsurePool(&poolCfg); err == nil {
|
||||
proxy = p
|
||||
} else {
|
||||
sendShareResult(false, "pool not connected: "+err.Error())
|
||||
return
|
||||
}
|
||||
}
|
||||
|
||||
wallet := poolCfg.Wallet
|
||||
if wallet == "" {
|
||||
wallet = h.defaultPool.Wallet
|
||||
}
|
||||
if !proxy.IsConnected() {
|
||||
sendShareResult(false, "pool not connected")
|
||||
return
|
||||
}
|
||||
|
||||
proxy.SubmitShare(agentID, wallet, share.JobID, share.Nonce, share.Hash, sendShareResult)
|
||||
wallet := poolCfg.Wallet
|
||||
if wallet == "" {
|
||||
wallet = h.defaultPool.Wallet
|
||||
}
|
||||
|
||||
proxy.SubmitShare(aID, wallet, s.JobID, s.Nonce, s.Hash, sendShareResult)
|
||||
}(share, agentID)
|
||||
|
||||
case "get_job":
|
||||
var proxy *pool.Proxy
|
||||
@@ -513,8 +535,9 @@ func (h *WSHub) HandleDashboardWS(w http.ResponseWriter, r *http.Request) {
|
||||
}
|
||||
|
||||
dashID := uuid.New().String()
|
||||
dashConn := &DashboardConnection{Conn: conn}
|
||||
h.mu.Lock()
|
||||
h.dashboards[dashID] = conn
|
||||
h.dashboards[dashID] = dashConn
|
||||
h.mu.Unlock()
|
||||
|
||||
defer func() {
|
||||
@@ -553,17 +576,19 @@ func (h *WSHub) broadcastDashboard(msg Message) {
|
||||
return
|
||||
}
|
||||
|
||||
for id, conn := range h.dashboards {
|
||||
if err := conn.WriteMessage(websocket.TextMessage, data); err != nil {
|
||||
log.Printf("Failed to send to dashboard %s: %v", id, err)
|
||||
conn.Close()
|
||||
id := id
|
||||
go func() {
|
||||
h.mu.Lock()
|
||||
delete(h.dashboards, id)
|
||||
h.mu.Unlock()
|
||||
}()
|
||||
}
|
||||
for id, dashConn := range h.dashboards {
|
||||
go func(dashID string, dc *DashboardConnection) {
|
||||
if err := dc.WriteMessage(websocket.TextMessage, data); err != nil {
|
||||
// Use fmt.Printf to avoid infinite loop with the global log interceptor
|
||||
fmt.Printf("Failed to send to dashboard %s: %v\n", dashID, err)
|
||||
dc.Conn.Close()
|
||||
go func() {
|
||||
h.mu.Lock()
|
||||
delete(h.dashboards, dashID)
|
||||
h.mu.Unlock()
|
||||
}()
|
||||
}
|
||||
}(id, dashConn)
|
||||
}
|
||||
}
|
||||
|
||||
@@ -578,9 +603,11 @@ func (h *WSHub) BroadcastToAgents(msg Message) {
|
||||
defer h.mu.RUnlock()
|
||||
|
||||
for id, agent := range h.agents {
|
||||
if err := agent.SendJSON(msg); err != nil {
|
||||
log.Printf("Failed to send to agent %s: %v", id, err)
|
||||
}
|
||||
go func(a *AgentConnection, agentID string) {
|
||||
if err := a.SendJSON(msg); err != nil {
|
||||
fmt.Printf("Failed to send to agent %s: %v\n", agentID, err)
|
||||
}
|
||||
}(agent, id)
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
Reference in New Issue
Block a user