go-p2p/node/worker.go

package node

import (
	"encoding/base64"
	"time"

	core "dappco.re/go/core"

	"dappco.re/go/core/p2p/logging"
	"github.com/adrg/xdg"
)

// MinerManager interface for the mining package integration.
// This allows the node package to interact with mining.Manager without import cycles.
//
//	var minerManager MinerManager
type MinerManager interface {
	StartMiner(minerType string, config any) (MinerInstance, error)
	StopMiner(name string) error
	ListMiners() []MinerInstance
	GetMiner(name string) (MinerInstance, error)
}

// MinerInstance represents a running miner for stats collection.
//
//	var miner MinerInstance
type MinerInstance interface {
	GetName() string
	GetType() string
	GetStats() (any, error)
	GetConsoleHistory(lines int) []string
}

// ProfileManager interface for profile operations.
//
//	var profileManager ProfileManager
type ProfileManager interface {
	GetProfile(id string) (any, error)
	SaveProfile(profile any) error
}

// Worker handles incoming messages on a worker node.
//
//	worker := NewWorker(nodeManager, transport)
type Worker struct {
	nodeManager    *NodeManager
	transport      *Transport
	minerManager   MinerManager
	profileManager ProfileManager
	startedAt      time.Time
	DataDir        string // Base directory for deployments (defaults to xdg.DataHome)
	DataDirectory  string // Deprecated compatibility alias for DataDir
}

// NewWorker creates a new Worker instance.
//
//	worker := NewWorker(nodeManager, transport)
func NewWorker(nodeManager *NodeManager, transport *Transport) *Worker {
	return &Worker{
		nodeManager:   nodeManager,
		transport:     transport,
		startedAt:     time.Now(),
		DataDir:       xdg.DataHome,
		DataDirectory: xdg.DataHome,
	}
}

// SetMinerManager attaches the miner manager used for miner operations.
//
//	worker.SetMinerManager(minerManager)
func (w *Worker) SetMinerManager(manager MinerManager) {
	w.minerManager = manager
}

// SetProfileManager attaches the profile manager used for profile operations.
//
//	worker.SetProfileManager(profileManager)
func (w *Worker) SetProfileManager(manager ProfileManager) {
	w.profileManager = manager
}

// HandleMessage routes an incoming message to the correct worker handler.
//
//	worker.HandleMessage(conn, msg)
func (w *Worker) HandleMessage(conn *PeerConnection, msg *Message) {
	var response *Message
	var err error

	switch msg.Type {
	case MessagePing:
		response, err = w.handlePing(msg)
	case MessageGetStats:
		response, err = w.handleStats(msg)
	case MessageStartMiner:
		response, err = w.handleStartMiner(msg)
	case MessageStopMiner:
		response, err = w.handleStopMiner(msg)
	case MessageGetLogs:
		response, err = w.handleLogs(msg)
	case MessageDeploy:
		response, err = w.handleDeploy(conn, msg)
	default:
		// Unknown message type - ignore or send error
		return
	}

	if err != nil {
		// Send error response
		identity := w.nodeManager.GetIdentity()
		if identity != nil {
			errMsg, _ := NewErrorMessage(
				identity.ID,
				msg.From,
				ErrorCodeOperationFailed,
				err.Error(),
				msg.ID,
			)
			conn.Send(errMsg)
		}
		return
	}

	if response != nil {
		logging.Debug("sending response", logging.Fields{"type": response.Type, "to": msg.From})
		if err := conn.Send(response); err != nil {
			logging.Error("failed to send response", logging.Fields{"error": err})
		} else {
			logging.Debug("response sent successfully")
		}
	}
}

// handlePing responds to ping requests.
func (w *Worker) handlePing(msg *Message) (*Message, error) {
	var ping PingPayload
	if err := msg.ParsePayload(&ping); err != nil {
		return nil, core.E("Worker.handlePing", "invalid ping payload", err)
	}

	pong := PongPayload{
		SentAt:     ping.SentAt,
		ReceivedAt: time.Now().UnixMilli(),
	}

	return msg.Reply(MessagePong, pong)
}

// handleStats responds with current miner statistics.
func (w *Worker) handleStats(msg *Message) (*Message, error) {
	identity := w.nodeManager.GetIdentity()
	if identity == nil {
		return nil, ErrorIdentityNotInitialized
	}

	stats := StatsPayload{
		NodeID:   identity.ID,
		NodeName: identity.Name,
		Miners:   []MinerStatsItem{},
		Uptime:   int64(time.Since(w.startedAt).Seconds()),
	}

	if w.minerManager != nil {
		miners := w.minerManager.ListMiners()
		for _, miner := range miners {
			minerStats, err := miner.GetStats()
			if err != nil {
				continue
			}

			// Convert to MinerStatsItem - this is a simplified conversion
			// The actual implementation would need to match the mining package's stats structure
			item := convertMinerStats(miner, minerStats)
			stats.Miners = append(stats.Miners, item)
		}
	}

	return msg.Reply(MessageStats, stats)
}

// convertMinerStats converts miner stats to the protocol format.
func convertMinerStats(miner MinerInstance, rawStats any) MinerStatsItem {
	item := MinerStatsItem{
		Name: miner.GetName(),
		Type: miner.GetType(),
	}

	// Try to extract common fields from the stats
	if statsMap, ok := rawStats.(map[string]any); ok {
		if hashrate, ok := statsMap["hashrate"].(float64); ok {
			item.Hashrate = hashrate
		}
		if shares, ok := statsMap["shares"].(int); ok {
			item.Shares = shares
		}
		if rejected, ok := statsMap["rejected"].(int); ok {
			item.Rejected = rejected
		}
		if uptime, ok := statsMap["uptime"].(int); ok {
			item.Uptime = uptime
		}
		if pool, ok := statsMap["pool"].(string); ok {
			item.Pool = pool
		}
		if algorithm, ok := statsMap["algorithm"].(string); ok {
			item.Algorithm = algorithm
		}
	}

	return item
}

// handleStartMiner starts a miner with the given profile.
func (w *Worker) handleStartMiner(msg *Message) (*Message, error) {
	if w.minerManager == nil {
		return nil, ErrorMinerManagerNotConfigured
	}

	var payload StartMinerPayload
	if err := msg.ParsePayload(&payload); err != nil {
		return nil, core.E("Worker.handleStartMiner", "invalid start miner payload", err)
	}

	// Validate miner type is provided
	if payload.MinerType == "" {
		return nil, core.E("Worker.handleStartMiner", "miner type is required", nil)
	}

	// Get the config from the profile or use the override
	var config any
	if payload.Config != nil {
		config = payload.Config
	} else if w.profileManager != nil {
		profile, err := w.profileManager.GetProfile(payload.ProfileID)
		if err != nil {
			return nil, core.E("Worker.handleStartMiner", "profile not found: "+payload.ProfileID, nil)
		}
		config = profile
	} else {
		return nil, core.E("Worker.handleStartMiner", "no config provided and no profile manager configured", nil)
	}

	// Start the miner
	miner, err := w.minerManager.StartMiner(payload.MinerType, config)
	if err != nil {
		ack := MinerAckPayload{
			Success: false,
			Error:   err.Error(),
		}
		return msg.Reply(MessageMinerAck, ack)
	}

	ack := MinerAckPayload{
		Success:   true,
		MinerName: miner.GetName(),
	}
	return msg.Reply(MessageMinerAck, ack)
}

// handleStopMiner stops a running miner.
func (w *Worker) handleStopMiner(msg *Message) (*Message, error) {
	if w.minerManager == nil {
		return nil, ErrorMinerManagerNotConfigured
	}

	var payload StopMinerPayload
	if err := msg.ParsePayload(&payload); err != nil {
		return nil, core.E("Worker.handleStopMiner", "invalid stop miner payload", err)
	}

	err := w.minerManager.StopMiner(payload.MinerName)
	ack := MinerAckPayload{
		Success:   err == nil,
		MinerName: payload.MinerName,
	}
	if err != nil {
		ack.Error = err.Error()
	}

	return msg.Reply(MessageMinerAck, ack)
}

// handleLogs returns console logs from a miner.
func (w *Worker) handleLogs(msg *Message) (*Message, error) {
	if w.minerManager == nil {
		return nil, ErrorMinerManagerNotConfigured
	}

	var payload LogsRequestPayload
	if err := msg.ParsePayload(&payload); err != nil {
		return nil, core.E("Worker.handleLogs", "invalid logs payload", err)
	}

	// Validate and limit the Lines parameter to prevent resource exhaustion
	const maxLogLines = 10000
	if payload.Lines <= 0 || payload.Lines > maxLogLines {
		payload.Lines = maxLogLines
	}

	miner, err := w.minerManager.GetMiner(payload.MinerName)
	if err != nil {
		return nil, core.E("Worker.handleLogs", "miner not found: "+payload.MinerName, nil)
	}

	lines := miner.GetConsoleHistory(payload.Lines)

	logs := LogsPayload{
		MinerName: payload.MinerName,
		Lines:     lines,
		HasMore:   len(lines) >= payload.Lines,
	}

	return msg.Reply(MessageLogs, logs)
}

// handleDeploy handles deployment of profiles or miner bundles.
func (w *Worker) handleDeploy(conn *PeerConnection, msg *Message) (*Message, error) {
	var payload DeployPayload
	if err := msg.ParsePayload(&payload); err != nil {
		return nil, core.E("Worker.handleDeploy", "invalid deploy payload", err)
	}

	// Reconstruct Bundle object from payload
	bundle := &Bundle{
		Type:     BundleType(payload.BundleType),
		Name:     payload.Name,
		Data:     payload.Data,
		Checksum: payload.Checksum,
	}

	// Use shared secret as password (base64 encoded)
	password := ""
	if conn != nil && len(conn.SharedSecret) > 0 {
		password = base64.StdEncoding.EncodeToString(conn.SharedSecret)
	}

	switch bundle.Type {
	case BundleProfile:
		if w.profileManager == nil {
			return nil, core.E("Worker.handleDeploy", "profile manager not configured", nil)
		}

		// Decrypt and extract profile data
		profileData, err := ExtractProfileBundle(bundle, password)
		if err != nil {
			return nil, core.E("Worker.handleDeploy", "failed to extract profile bundle", err)
		}

		// Unmarshal into interface{} to pass to ProfileManager
		var profile any
		if result := core.JSONUnmarshal(profileData, &profile); !result.OK {
			return nil, core.E("Worker.handleDeploy", "invalid profile data JSON", result.Value.(error))
		}

		if err := w.profileManager.SaveProfile(profile); err != nil {
			ack := DeployAckPayload{
				Success: false,
				Name:    payload.Name,
				Error:   err.Error(),
			}
			return msg.Reply(MessageDeployAck, ack)
		}

		ack := DeployAckPayload{
			Success: true,
			Name:    payload.Name,
		}
		return msg.Reply(MessageDeployAck, ack)

	case BundleMiner, BundleFull:
		// Determine installation directory
		// We use the configured deployment directory for
		// lethean-desktop/miners/<bundle_name>.
		minersDir := core.JoinPath(w.deploymentDir(), "lethean-desktop", "miners")
		installDir := core.JoinPath(minersDir, payload.Name)

		logging.Info("deploying miner bundle", logging.Fields{
			"name": payload.Name,
			"path": installDir,
			"type": payload.BundleType,
		})

		// Extract miner bundle
		minerPath, profileData, err := ExtractMinerBundle(bundle, password, installDir)
		if err != nil {
			return nil, core.E("Worker.handleDeploy", "failed to extract miner bundle", err)
		}

		// If the bundle contained a profile config, save it
		if len(profileData) > 0 && w.profileManager != nil {
			var profile any
			if result := core.JSONUnmarshal(profileData, &profile); !result.OK {
				logging.Warn("failed to parse profile from miner bundle", logging.Fields{"error": result.Value.(error)})
			} else {
				if err := w.profileManager.SaveProfile(profile); err != nil {
					logging.Warn("failed to save profile from miner bundle", logging.Fields{"error": err})
				}
			}
		}

		// Success response
		ack := DeployAckPayload{
			Success: true,
			Name:    payload.Name,
		}

		// Log the installation
		logging.Info("miner bundle installed successfully", logging.Fields{
			"name":       payload.Name,
			"miner_path": minerPath,
		})

		return msg.Reply(MessageDeployAck, ack)

	default:
		return nil, core.E("Worker.handleDeploy", "unknown bundle type: "+payload.BundleType, nil)
	}
}

// RegisterOnTransport installs the worker message handler on the transport.
//
//	worker.RegisterOnTransport()
func (w *Worker) RegisterOnTransport() {
	w.transport.OnMessage(w.HandleMessage)
}

// RegisterWithTransport is a deprecated compatibility alias for RegisterOnTransport.
func (w *Worker) RegisterWithTransport() {
	w.RegisterOnTransport()
}

// deploymentDir resolves the active deployment directory, preferring DataDir
// unless a caller has only populated the legacy DataDirectory field.
func (w *Worker) deploymentDir() string {
	switch {
	case w.DataDir != "" && w.DataDir != xdg.DataHome:
		return w.DataDir
	case w.DataDirectory != "":
		return w.DataDirectory
	case w.DataDir != "":
		return w.DataDir
	default:
		return xdg.DataHome
	}
}