edubox/agent/websocket.go

package main

import (
	"fmt"
	"log"
	"sync"
	"time"

	"github.com/gorilla/websocket"
)

type WSMessage struct {
	Action        string `json:"action"`
	NodeID        string `json:"nodeId,omitempty"`
	Code          string `json:"code,omitempty"`
	InstanceID    string `json:"instanceId,omitempty"`
	Type          string `json:"type,omitempty"`
	Port          int    `json:"port,omitempty"`
	ComposeConfig string `json:"composeConfig,omitempty"`
	StudentId     string `json:"studentId,omitempty"`
	StudentName   string `json:"studentName,omitempty"`
	Error         string `json:"error,omitempty"`
}

var (
	mainConn   *websocket.Conn
	mainConnMu sync.Mutex
)

func sendMessage(msg WSMessage) error {
	mainConnMu.Lock()
	defer mainConnMu.Unlock()
	if mainConn == nil {
		return fmt.Errorf("not connected to server")
	}
	if msg.Action != "heartbeat" {
		log.Printf("sendMessage: sending %+v", msg)
	}
	return mainConn.WriteJSON(msg)
}

// UI notifier system: broadcast activation results to all connected UI clients
type uiNotifier func(msg map[string]interface{})

var (
	uiNotifiers   = make(map[int]uiNotifier)
	uiNotifiersMu sync.Mutex
	uiNotifierID  int
)

func registerUINotifier(fn uiNotifier) int {
	uiNotifiersMu.Lock()
	defer uiNotifiersMu.Unlock()
	id := uiNotifierID
	uiNotifierID++
	uiNotifiers[id] = fn
	log.Printf("registerUINotifier: registered ID %d (total: %d)", id, len(uiNotifiers))
	return id
}

func unregisterUINotifier(id int) {
	uiNotifiersMu.Lock()
	defer uiNotifiersMu.Unlock()
	delete(uiNotifiers, id)
	log.Printf("unregisterUINotifier: removed ID %d (total: %d)", id, len(uiNotifiers))
}

func notifyUI(msg map[string]interface{}) {
	uiNotifiersMu.Lock()
	notifiers := make([]uiNotifier, 0, len(uiNotifiers))
	for _, fn := range uiNotifiers {
		notifiers = append(notifiers, fn)
	}
	uiNotifiersMu.Unlock()

	log.Printf("notifyUI: broadcasting to %d UI clients", len(notifiers))
	for _, fn := range notifiers {
		go fn(msg)
	}
}

func startWebSocket(serverAddr, nodeID, dataDir string) {
	for {
		conn, _, err := websocket.DefaultDialer.Dial(serverAddr, nil)
		if err != nil {
			log.Printf("WS connect error: %v, retrying in 5s...", err)
			time.Sleep(5 * time.Second)
			continue
		}

		log.Printf("WS connected to %s", serverAddr)

		mainConnMu.Lock()
		mainConn = conn
		mainConnMu.Unlock()

		// Register
		if err := conn.WriteJSON(WSMessage{Action: "register", NodeID: nodeID}); err != nil {
			log.Printf("WS register error: %v", err)
			conn.Close()
			mainConnMu.Lock()
			mainConn = nil
			mainConnMu.Unlock()
			continue
		}

		// Activation flow
		act, err := loadActivation(dataDir)
		if err != nil || !act.Activated {
			log.Println("Waiting for activation...")
		} else {
			log.Printf("Already activated as %s", act.StudentName)
		}

		// Heartbeat goroutine
		done := make(chan struct{})
		go func() {
			ticker := time.NewTicker(10 * time.Second)
			defer ticker.Stop()
			for {
				select {
				case <-ticker.C:
					if err := sendMessage(WSMessage{Action: "heartbeat", NodeID: nodeID}); err != nil {
						return
					}
				case <-done:
					return
				}
			}
		}()

		// Read loop
		for {
			var msg WSMessage
			if err := conn.ReadJSON(&msg); err != nil {
				log.Printf("WS read error: %v", err)
				break
			}
			log.Printf("WS received from server: action=%s", msg.Action)
			handleMessage(conn, msg, dataDir, nodeID)
		}

		close(done)
		conn.Close()
		mainConnMu.Lock()
		mainConn = nil
		mainConnMu.Unlock()
		log.Println("WS disconnected, reconnecting in 5s...")
		time.Sleep(5 * time.Second)
	}
}

func handleMessage(conn *websocket.Conn, msg WSMessage, dataDir, nodeID string) {
	switch msg.Action {
	case "activated":
		log.Printf("handleMessage: activated received, student=%s", msg.StudentName)
		if msg.StudentName != "" {
			act := &Activation{Activated: true, StudentId: msg.StudentId, StudentName: msg.StudentName, Code: msg.Code}
			if err := saveActivation(dataDir, act); err != nil {
				log.Printf("saveActivation error: %v", err)
			} else {
				log.Printf("Activated as %s", msg.StudentName)
			}
		}
		notifyUI(map[string]interface{}{
			"action":      "activated",
			"studentName": msg.StudentName,
		})
    case "registered":
        // Server acknowledged our register message; nothing to do.
        return
	case "activation_failed":
		log.Printf("handleMessage: activation_failed received, error=%s", msg.Error)
		notifyUI(map[string]interface{}{
			"action": "activation_failed",
			"error":  msg.Error,
		})
	case "start":
		log.Printf("Start instance %s on port %d", msg.InstanceID, msg.Port)
		if err := upsertInstance(dataDir, &InstanceInfo{
			ID:           msg.InstanceID,
			TemplateName: msg.Type,
			Port:         msg.Port,
			Status:       "starting",
		}); err != nil {
			log.Printf("upsertInstance error: %v", err)
		}
		if err := writeCompose(dataDir, msg.InstanceID, msg.ComposeConfig); err != nil {
			log.Printf("writeCompose error: %v", err)
			_ = upsertInstance(dataDir, &InstanceInfo{ID: msg.InstanceID, TemplateName: msg.Type, Port: msg.Port, Status: "error"})
			sendMessage(WSMessage{Action: "instance_error", InstanceID: msg.InstanceID, Error: err.Error()})
			return
		}
		if err := dockerComposeUp(dataDir, msg.InstanceID); err != nil {
			log.Printf("dockerComposeUp error: %v", err)
			_ = upsertInstance(dataDir, &InstanceInfo{ID: msg.InstanceID, TemplateName: msg.Type, Port: msg.Port, Status: "error"})
			sendMessage(WSMessage{Action: "instance_error", InstanceID: msg.InstanceID, Error: err.Error()})
			return
		}
		status := getInstanceStatus(dataDir, msg.InstanceID)
		_ = upsertInstance(dataDir, &InstanceInfo{ID: msg.InstanceID, TemplateName: msg.Type, Port: msg.Port, Status: status})
		sendMessage(WSMessage{Action: "instance_started", InstanceID: msg.InstanceID, Port: msg.Port})
		notifyUI(map[string]interface{}{"action": "instances_updated"})
	case "stop":
		log.Printf("Stop instance %s", msg.InstanceID)
		if err := dockerComposeDown(dataDir, msg.InstanceID); err != nil {
			log.Printf("dockerComposeDown error: %v", err)
		}
		if inst, _ := loadInstances(dataDir); inst[msg.InstanceID] != nil {
			inst[msg.InstanceID].Status = "stopped"
			_ = saveInstances(dataDir, inst)
		}
		notifyUI(map[string]interface{}{"action": "instances_updated"})
	case "reset":
		log.Printf("Reset instance %s", msg.InstanceID)
		dockerComposeRm(dataDir, msg.InstanceID)
		if err := writeCompose(dataDir, msg.InstanceID, msg.ComposeConfig); err != nil {
			log.Printf("writeCompose error: %v", err)
			_ = upsertInstance(dataDir, &InstanceInfo{ID: msg.InstanceID, TemplateName: msg.Type, Port: msg.Port, Status: "error"})
			sendMessage(WSMessage{Action: "instance_error", InstanceID: msg.InstanceID, Error: err.Error()})
			return
		}
		if err := dockerComposeUp(dataDir, msg.InstanceID); err != nil {
			log.Printf("dockerComposeUp error: %v", err)
			_ = upsertInstance(dataDir, &InstanceInfo{ID: msg.InstanceID, TemplateName: msg.Type, Port: msg.Port, Status: "error"})
			sendMessage(WSMessage{Action: "instance_error", InstanceID: msg.InstanceID, Error: err.Error()})
			return
		}
		status := getInstanceStatus(dataDir, msg.InstanceID)
		_ = upsertInstance(dataDir, &InstanceInfo{ID: msg.InstanceID, TemplateName: msg.Type, Port: msg.Port, Status: status})
		sendMessage(WSMessage{Action: "instance_started", InstanceID: msg.InstanceID, Port: msg.Port})
		notifyUI(map[string]interface{}{"action": "instances_updated"})
	default:
		log.Printf("Unknown action: %s", msg.Action)
	}
}