340 lines
9.1 KiB
Go
340 lines
9.1 KiB
Go
package api
|
|
|
|
import (
|
|
"encoding/json"
|
|
"fmt"
|
|
"net/http"
|
|
"net/url"
|
|
"strings"
|
|
"time"
|
|
|
|
"3588AdminBackend/internal/models"
|
|
"3588AdminBackend/internal/service"
|
|
"github.com/go-chi/chi/v5"
|
|
)
|
|
|
|
type Handler struct {
|
|
discovery *service.DiscoveryService
|
|
registry *service.RegistryService
|
|
agent *service.AgentClient
|
|
tasks *service.TaskService
|
|
templates *service.TemplateService
|
|
}
|
|
|
|
func NewHandler(discovery *service.DiscoveryService, registry *service.RegistryService, agent *service.AgentClient, tasks *service.TaskService, templates *service.TemplateService) *Handler {
|
|
return &Handler{
|
|
discovery: discovery,
|
|
registry: registry,
|
|
agent: agent,
|
|
tasks: tasks,
|
|
templates: templates,
|
|
}
|
|
}
|
|
|
|
func (h *Handler) Search(w http.ResponseWriter, r *http.Request) {
|
|
var req struct {
|
|
TimeoutMs int `json:"timeout_ms"`
|
|
}
|
|
if err := json.NewDecoder(r.Body).Decode(&req); err != nil {
|
|
req.TimeoutMs = 1200 // Default
|
|
}
|
|
|
|
items, err := h.discovery.Search(req.TimeoutMs)
|
|
if err != nil {
|
|
http.Error(w, err.Error(), http.StatusInternalServerError)
|
|
return
|
|
}
|
|
|
|
json.NewEncoder(w).Encode(map[string]interface{}{
|
|
"items": items,
|
|
})
|
|
}
|
|
|
|
func (h *Handler) ListDevices(w http.ResponseWriter, r *http.Request) {
|
|
items := h.registry.GetDevices()
|
|
json.NewEncoder(w).Encode(map[string]interface{}{
|
|
"items": items,
|
|
})
|
|
}
|
|
|
|
func (h *Handler) CreateDevice(w http.ResponseWriter, r *http.Request) {
|
|
var req struct {
|
|
DeviceID string `json:"device_id"`
|
|
DeviceName string `json:"device_name"`
|
|
IP string `json:"ip"`
|
|
AgentPort int `json:"agent_port"`
|
|
MediaPort int `json:"media_port"`
|
|
}
|
|
if err := json.NewDecoder(r.Body).Decode(&req); err != nil {
|
|
http.Error(w, "invalid request body", http.StatusBadRequest)
|
|
return
|
|
}
|
|
if req.DeviceID == "" || req.IP == "" {
|
|
http.Error(w, "device_id and ip are required", http.StatusBadRequest)
|
|
return
|
|
}
|
|
if req.AgentPort == 0 {
|
|
req.AgentPort = 9100
|
|
}
|
|
if req.MediaPort == 0 {
|
|
req.MediaPort = 9000
|
|
}
|
|
|
|
dev := &models.Device{
|
|
DeviceID: req.DeviceID,
|
|
DeviceName: req.DeviceName,
|
|
IP: req.IP,
|
|
AgentPort: req.AgentPort,
|
|
MediaPort: req.MediaPort,
|
|
Online: true,
|
|
LastSeenMs: time.Now().UnixMilli(),
|
|
}
|
|
h.registry.UpdateDevice(dev)
|
|
json.NewEncoder(w).Encode(map[string]interface{}{"ok": true})
|
|
}
|
|
|
|
func (h *Handler) GetDevice(w http.ResponseWriter, r *http.Request) {
|
|
id := chi.URLParam(r, "id")
|
|
dev, ok := h.findDevice(id)
|
|
if !ok {
|
|
http.Error(w, "device not found", http.StatusNotFound)
|
|
return
|
|
}
|
|
json.NewEncoder(w).Encode(dev)
|
|
}
|
|
|
|
func (h *Handler) findDevice(id string) (*models.Device, bool) {
|
|
devices := h.registry.GetDevices()
|
|
for _, d := range devices {
|
|
if d.DeviceID == id {
|
|
return d, true
|
|
}
|
|
}
|
|
return nil, false
|
|
}
|
|
|
|
func (h *Handler) ProxyAgent(w http.ResponseWriter, r *http.Request) {
|
|
id := chi.URLParam(r, "id")
|
|
name := chi.URLParam(r, "name")
|
|
dev, ok := h.findDevice(id)
|
|
if !ok {
|
|
http.Error(w, "device not found", http.StatusNotFound)
|
|
return
|
|
}
|
|
var agentPath string
|
|
method := r.Method
|
|
|
|
switch {
|
|
case r.URL.Path == fmt.Sprintf("/api/devices/%s/info", id):
|
|
agentPath = "/v1/info"
|
|
case r.URL.Path == fmt.Sprintf("/api/devices/%s/config/status", id):
|
|
agentPath = "/v1/config/status"
|
|
case r.URL.Path == fmt.Sprintf("/api/devices/%s/config/candidate", id):
|
|
agentPath = "/v1/config/candidate"
|
|
method = "PUT"
|
|
case r.URL.Path == fmt.Sprintf("/api/devices/%s/config/candidate/apply", id):
|
|
agentPath = "/v1/config/candidate/apply"
|
|
method = "POST"
|
|
case r.URL.Path == fmt.Sprintf("/api/devices/%s/reload", id):
|
|
agentPath = "/v1/media-server/reload"
|
|
method = "POST"
|
|
case r.URL.Path == fmt.Sprintf("/api/devices/%s/rollback", id):
|
|
agentPath = "/v1/media-server/rollback"
|
|
method = "POST"
|
|
case r.URL.Path == fmt.Sprintf("/api/devices/%s/media-server/start", id):
|
|
agentPath = "/v1/media-server/start"
|
|
method = "POST"
|
|
case r.URL.Path == fmt.Sprintf("/api/devices/%s/media-server/restart", id):
|
|
agentPath = "/v1/media-server/restart"
|
|
method = "POST"
|
|
case r.URL.Path == fmt.Sprintf("/api/devices/%s/media-server/stop", id):
|
|
agentPath = "/v1/media-server/stop"
|
|
method = "POST"
|
|
case r.URL.Path == fmt.Sprintf("/api/devices/%s/media-server/status", id):
|
|
agentPath = "/v1/media-server/status"
|
|
case r.URL.Path == fmt.Sprintf("/api/devices/%s/graphs", id):
|
|
agentPath = "/v1/graphs"
|
|
case name != "" && r.URL.Path == fmt.Sprintf("/api/devices/%s/graphs/%s", id, name):
|
|
agentPath = "/v1/graphs/" + url.PathEscape(name)
|
|
case r.URL.Path == fmt.Sprintf("/api/devices/%s/logs", id):
|
|
agentPath = "/v1/logs/recent"
|
|
if q := r.URL.RawQuery; q != "" {
|
|
agentPath += "?" + q
|
|
}
|
|
case r.URL.Path == fmt.Sprintf("/api/devices/%s/config/apply", id):
|
|
agentPath = "/v1/config"
|
|
method = "PUT"
|
|
case r.URL.Path == fmt.Sprintf("/api/devices/%s/models", id):
|
|
agentPath = "/v1/models"
|
|
default:
|
|
http.Error(w, "path not mapped", http.StatusNotFound)
|
|
return
|
|
}
|
|
|
|
contentType := r.Header.Get("Content-Type")
|
|
resp, code, err := h.agent.DoStream(method, dev.IP, dev.AgentPort, agentPath, r.Body, contentType, r.ContentLength)
|
|
if err != nil {
|
|
http.Error(w, err.Error(), http.StatusInternalServerError)
|
|
return
|
|
}
|
|
|
|
// Update online status on successful response
|
|
if code < 500 && h.registry != nil {
|
|
h.registry.TouchDevice(id)
|
|
}
|
|
|
|
w.WriteHeader(code)
|
|
w.Write(resp)
|
|
}
|
|
|
|
// ProxyAgentV1 proxies any request under /api/devices/{id}/v1/* to the device agent as-is (path preserved).
|
|
func (h *Handler) ProxyAgentV1(w http.ResponseWriter, r *http.Request) {
|
|
id := chi.URLParam(r, "id")
|
|
dev, ok := h.findDevice(id)
|
|
if !ok {
|
|
http.Error(w, "device not found", http.StatusNotFound)
|
|
return
|
|
}
|
|
prefix := fmt.Sprintf("/api/devices/%s", id)
|
|
agentPath := strings.TrimPrefix(r.URL.Path, prefix)
|
|
if !strings.HasPrefix(agentPath, "/v1/") && agentPath != "/v1" {
|
|
http.Error(w, "path not mapped", http.StatusNotFound)
|
|
return
|
|
}
|
|
if q := r.URL.RawQuery; q != "" {
|
|
agentPath += "?" + q
|
|
}
|
|
|
|
contentType := r.Header.Get("Content-Type")
|
|
resp, code, err := h.agent.DoStream(r.Method, dev.IP, dev.AgentPort, agentPath, r.Body, contentType, r.ContentLength)
|
|
if err != nil {
|
|
http.Error(w, err.Error(), http.StatusInternalServerError)
|
|
return
|
|
}
|
|
|
|
// Update online status on successful response
|
|
if code < 500 && h.registry != nil {
|
|
h.registry.TouchDevice(id)
|
|
}
|
|
|
|
w.WriteHeader(code)
|
|
_, _ = w.Write(resp)
|
|
}
|
|
|
|
func (h *Handler) CreateTask(w http.ResponseWriter, r *http.Request) {
|
|
var req struct {
|
|
Type string `json:"type"`
|
|
DeviceIDs []string `json:"device_ids"`
|
|
Payload interface{} `json:"payload"`
|
|
}
|
|
if err := json.NewDecoder(r.Body).Decode(&req); err != nil {
|
|
http.Error(w, "invalid request body", http.StatusBadRequest)
|
|
return
|
|
}
|
|
|
|
task, err := h.tasks.CreateTask(req.Type, req.DeviceIDs, req.Payload)
|
|
if err != nil {
|
|
http.Error(w, err.Error(), http.StatusInternalServerError)
|
|
return
|
|
}
|
|
|
|
json.NewEncoder(w).Encode(map[string]string{"task_id": task.ID})
|
|
}
|
|
|
|
func (h *Handler) ListTasks(w http.ResponseWriter, r *http.Request) {
|
|
if h.tasks == nil {
|
|
http.Error(w, "task service not initialized", http.StatusInternalServerError)
|
|
return
|
|
}
|
|
|
|
items := h.tasks.ListTasks()
|
|
json.NewEncoder(w).Encode(map[string]interface{}{
|
|
"items": items,
|
|
})
|
|
}
|
|
|
|
func (h *Handler) TaskEvents(w http.ResponseWriter, r *http.Request) {
|
|
taskID := chi.URLParam(r, "id")
|
|
ch, cleanup := h.tasks.Subscribe(taskID)
|
|
defer cleanup()
|
|
|
|
w.Header().Set("Content-Type", "text/event-stream")
|
|
w.Header().Set("Cache-Control", "no-cache")
|
|
w.Header().Set("Connection", "keep-alive")
|
|
|
|
flusher, ok := w.(http.Flusher)
|
|
if !ok {
|
|
http.Error(w, "SSE not supported", http.StatusInternalServerError)
|
|
return
|
|
}
|
|
|
|
for {
|
|
select {
|
|
case <-r.Context().Done():
|
|
return
|
|
case update := <-ch:
|
|
data, _ := json.Marshal(update)
|
|
fmt.Fprintf(w, "event: device_update\ndata: %s\n\n", data)
|
|
flusher.Flush()
|
|
}
|
|
}
|
|
}
|
|
|
|
func (h *Handler) ListTemplates(w http.ResponseWriter, r *http.Request) {
|
|
list, err := h.templates.ListTemplates()
|
|
if err != nil {
|
|
http.Error(w, err.Error(), http.StatusInternalServerError)
|
|
return
|
|
}
|
|
json.NewEncoder(w).Encode(list)
|
|
}
|
|
|
|
func (h *Handler) GetTemplate(w http.ResponseWriter, r *http.Request) {
|
|
name := chi.URLParam(r, "name")
|
|
t, err := h.templates.GetTemplate(name)
|
|
if err != nil {
|
|
http.Error(w, err.Error(), http.StatusNotFound)
|
|
return
|
|
}
|
|
json.NewEncoder(w).Encode(t)
|
|
}
|
|
|
|
func (h *Handler) UploadModel(w http.ResponseWriter, r *http.Request) {
|
|
id := chi.URLParam(r, "id")
|
|
dev, ok := h.findDevice(id)
|
|
if !ok {
|
|
http.Error(w, "device not found", http.StatusNotFound)
|
|
return
|
|
}
|
|
|
|
// Parse multipart
|
|
if err := r.ParseMultipartForm(100 << 20); err != nil { // 100MB
|
|
http.Error(w, err.Error(), http.StatusBadRequest)
|
|
return
|
|
}
|
|
|
|
name := r.FormValue("name")
|
|
file, hdr, err := r.FormFile("file")
|
|
if err != nil {
|
|
http.Error(w, err.Error(), http.StatusBadRequest)
|
|
return
|
|
}
|
|
defer file.Close()
|
|
|
|
if name == "" {
|
|
http.Error(w, "name is required", http.StatusBadRequest)
|
|
return
|
|
}
|
|
|
|
// Forward to agent
|
|
agentPath := fmt.Sprintf("/v1/models/%s", name)
|
|
resp, code, err := h.agent.DoStream("PUT", dev.IP, dev.AgentPort, agentPath, file, "application/octet-stream", hdr.Size)
|
|
if err != nil {
|
|
http.Error(w, err.Error(), http.StatusInternalServerError)
|
|
return
|
|
}
|
|
|
|
w.WriteHeader(code)
|
|
w.Write(resp)
|
|
}
|