package httpapi import ( "encoding/json" "net/http" "os" "strings" "webui4frpc/internal/cluster" "webui4frpc/internal/store" ) // canvasData is the full drawing canvas exchanged with the frontend. type canvasData struct { Locals []store.Local `json:"locals"` Remotes []store.Remote `json:"remotes"` Links []store.Link `json:"links"` } func (h *Handler) handleCanvasGet(w http.ResponseWriter, _ *http.Request) { locals, _ := h.Store.ListLocals() remotes, _ := h.Store.ListRemotes() links, _ := h.Store.ListLinks() // Merge ring topology entries not in the local store so every node's // canvas shows the FULL cluster picture. This is a read-time merge — // nothing is written back to SQLite (avoids the per-forward stop/start // issue that broke the old topology-derived-canvas model). if h.Ring != nil { snap := h.Ring.Snapshot() localNames := make(map[string]bool, len(locals)) for _, l := range locals { localNames[l.Name] = true } remoteNames := make(map[string]bool, len(remotes)) for _, r := range remotes { remoteNames[r.Name] = true } type linkKey struct{ local, remote string; port int } linkSeen := make(map[linkKey]bool, len(links)) for _, l := range links { linkSeen[linkKey{l.Local, l.Remote, l.RemotePort}] = true } for _, t := range snap.Topology { if !localNames[t.Local.Name] { locals = append(locals, t.Local) localNames[t.Local.Name] = true } if !remoteNames[t.Remote.Name] { remotes = append(remotes, t.Remote) remoteNames[t.Remote.Name] = true } k := linkKey{t.Link.Local, t.Link.Remote, t.Link.RemotePort} if !linkSeen[k] { links = append(links, t.Link) linkSeen[k] = true } } } writeJSON(w, http.StatusOK, canvasData{Locals: locals, Remotes: remotes, Links: links}) } func (h *Handler) handleCanvasSave(w http.ResponseWriter, r *http.Request) { switch r.Method { case http.MethodGet: h.handleCanvasGet(w, r) case http.MethodPut: // Route is registered at read so viewer GETs work; PUT needs write. if !hasLevel(r, "write") { forbidden(w) return } h.saveCanvas(w, r) default: methodNotAllowed(w) } } func (h *Handler) saveCanvas(w http.ResponseWriter, r *http.Request) { var canvas canvasData if err := json.NewDecoder(r.Body).Decode(&canvas); err != nil { http.Error(w, "parse json: "+err.Error(), http.StatusBadRequest) return } if !h.applyCanvas(w, r, &canvas) { return } h.handleCanvasGet(w, r) } // applyCanvas performs the full-replace semantics shared by PUT /canvas and // POST /canvas/import: loopback rewrite, upsert locals + delete-missing (with // localOnly stop / cluster revoke), upsert remotes + delete-missing, wholesale // link replace, cluster task submit for non-localOnly forwards, and a // SyncWorkers pass. It writes errors to w and returns false on failure so the // caller knows not to write a success response. func (h *Handler) applyCanvas(w http.ResponseWriter, r *http.Request, canvas *canvasData) bool { s := h.Store // Rewrite loopback backend addresses for cluster-distributed forwards. // Non-localOnly locals targeting 127.0.0.1/0.0.0.0/localhost must be // reachable from whichever node claims them, so swap in this node LAN addr. // Local-only forwards keep the loopback as-is (they never leave this node). for i := range canvas.Locals { if !canvas.Locals[i].LocalOnly && cluster.IsLoopbackIP(canvas.Locals[i].IP) { canvas.Locals[i].IP = cluster.RewriteForCluster(canvas.Locals[i].IP) } } // Upsert locals. for _, l := range canvas.Locals { if err := s.UpsertLocal(l); err != nil { http.Error(w, err.Error(), http.StatusBadRequest) return false } } // Delete locals not present. A removed localOnly forward is cancelled // locally (stop worker). A removed cluster forward publishes a REVOKE // task through the token so its owning node cancels it everywhere. if existing, err := s.ListLocals(); err == nil { keep := map[string]bool{} for _, l := range canvas.Locals { keep[l.Name] = true } for _, old := range existing { if !keep[old.Name] { if old.LocalOnly { if h.Process != nil { if fwd, _ := s.LinksForLocal(old.Name); len(fwd) > 0 { _ = h.Process.Stop(fwd[0].Remote) } } } else if h.Ring != nil { // Publish a REVOKE task for EVERY link of the removed local — // a local may fan out to several remotes, and revoking only the // first (fwd[0]) left the rest as orphan workers running on // their owning cluster nodes. fwd, _ := s.LinksForLocal(old.Name) for _, ln := range fwd { rem, ok := s.GetRemote(ln.Remote) if !ok { continue } h.Ring.RevokeTask(old, rem, store.Link{ Local: old.Name, Remote: rem.Name, RemotePort: ln.RemotePort, }) } } _ = s.DeleteLocal(old.Name) } } } // Upsert remotes. for _, rem := range canvas.Remotes { if err := s.UpsertRemote(rem); err != nil { http.Error(w, err.Error(), http.StatusBadRequest) return false } } if existing, err := s.ListRemotes(); err == nil { keep := map[string]bool{} for _, rem := range canvas.Remotes { keep[rem.Name] = true } for _, old := range existing { if !keep[old.Name] { _ = s.DeleteRemote(old.Name) } } } // Replace links wholesale. if err := s.ReplaceLinks(canvas.Links); err != nil { http.Error(w, err.Error(), http.StatusBadRequest) return false } // Cluster distribution: reconcile non-localOnly forwards against the ring // topology, respecting each link's Disabled flag. A non-disabled forward // not yet in topology is submitted (lowest-load member claims it); a // disabled forward present in topology is revoked. plan §画布差异判断, // extended so a per-forward stop made on the forwards page (disabled=true) // is not re-activated by a later canvas save. Local-only forwards are NOT // submitted — they stay on this node. localByName := map[string]store.Local{} for _, l := range canvas.Locals { localByName[l.Name] = l } remoteByName := map[string]store.Remote{} for _, r := range canvas.Remotes { remoteByName[r.Name] = r } if h.Ring != nil { for _, ln := range canvas.Links { loc, ok := localByName[ln.Local] if !ok || loc.LocalOnly { continue } rem, ok := remoteByName[ln.Remote] if !ok { continue } if ln.Disabled { // Stopped on the forwards page: make sure it leaves the topology. if h.Ring.HasTask(ln.Local, ln.Remote, ln.RemotePort) { h.Ring.RevokeTask(loc, rem, ln) } continue } // SubmitTask is idempotent (HasTask guard), so re-saving an active // canvas is a no-op for forwards already in the topology. h.Ring.SubmitTask(loc, rem, ln) } } // Restart affected running workers so changes take effect immediately // (local-only forwards get their worker started right here). if h.SyncWorkers != nil { h.SyncWorkers() } return true } // ---- Settings ---- func (h *Handler) handleSettingsGet(w http.ResponseWriter, r *http.Request) { settings, _ := h.Store.Settings() writeJSON(w, http.StatusOK, settings) } func (h *Handler) handleSettingsPut(w http.ResponseWriter, r *http.Request) { var st store.Settings if err := json.NewDecoder(r.Body).Decode(&st); err != nil { http.Error(w, err.Error(), http.StatusBadRequest) return } if err := h.Store.UpdateSettings(st); err != nil { http.Error(w, err.Error(), http.StatusInternalServerError) return } h.handleSettingsGet(w, r) } // ---- Binary ---- func (h *Handler) handleBinaryStatus(w http.ResponseWriter, _ *http.Request) { resp := map[string]any{"binaryPath": ""} if h.BinaryPath != nil { resp["binaryPath"] = h.BinaryPath() } writeJSON(w, http.StatusOK, resp) } func (h *Handler) handleBinaryInstall(w http.ResponseWriter, r *http.Request) { if r.Method != http.MethodPost { methodNotAllowed(w) return } var req struct { Version string `json:"version"` } _ = json.NewDecoder(r.Body).Decode(&req) if h.InstallBinary == nil { http.Error(w, "install not configured", http.StatusInternalServerError) return } path, version, err := h.InstallBinary(req.Version) if err != nil { http.Error(w, "install failed: "+err.Error(), http.StatusBadGateway) return } writeJSON(w, http.StatusOK, map[string]any{"path": path, "version": version}) } // ---- Single remote upsert/delete (used by the status page) ---- func (h *Handler) handleRemoteUpsert(w http.ResponseWriter, r *http.Request) { if r.Method != http.MethodPut { methodNotAllowed(w) return } var rem store.Remote if err := json.NewDecoder(r.Body).Decode(&rem); err != nil { http.Error(w, "parse json: "+err.Error(), http.StatusBadRequest) return } if rem.Name == "" || rem.IP == "" || rem.Port <= 0 || rem.Port > 65535 { http.Error(w, "name/ip/port required, port in [1,65535]", http.StatusBadRequest) return } if err := h.Store.UpsertRemote(rem); err != nil { http.Error(w, err.Error(), http.StatusInternalServerError) return } // Start the worker if enabled. if rem.Enabled { _ = h.Process.Start(rem.Name) } writeJSON(w, http.StatusOK, rem) } func (h *Handler) handleRemoteDelete(w http.ResponseWriter, r *http.Request) { if r.Method != http.MethodDelete { methodNotAllowed(w) return } name := strings.TrimPrefix(r.URL.Path, apiPrefix+"/remotes/") if name == "" { http.Error(w, "remote name required", http.StatusBadRequest) return } _ = h.Process.Stop(name) if err := h.Store.DeleteRemote(name); err != nil { http.Error(w, err.Error(), http.StatusInternalServerError) return } w.WriteHeader(http.StatusOK) } // ---- Profile lifecycle ---- func (h *Handler) handleProfile(w http.ResponseWriter, r *http.Request) { // Path: /api/manager/profiles/{name}/{action?} rel := strings.TrimPrefix(r.URL.Path, apiPrefix+"/profiles/") parts := strings.Split(rel, "/") if len(parts) == 0 || parts[0] == "" { http.Error(w, "profile name required", http.StatusBadRequest) return } name := parts[0] action := "" if len(parts) > 1 { action = parts[1] } switch action { case "": // GET profile status st, has := h.Process.Status(name) forwards, _ := h.Store.LinksForRemote(name) writeJSON(w, http.StatusOK, map[string]any{ "name": name, "process": st, "hasProcess": has, "forwards": forwards, }) case "start", "stop", "restart": switch r.Method { case http.MethodPost: // Route registered at read so status/config/logs GETs work; worker // lifecycle mutations need write. if !hasLevel(r, "write") { forbidden(w) return } var err error if action == "start" { err = h.Process.Start(name) } else if action == "stop" { err = h.Process.Stop(name) } else { err = h.Process.Restart(name) } if err != nil { http.Error(w, err.Error(), http.StatusInternalServerError) return } w.WriteHeader(http.StatusOK) default: methodNotAllowed(w) } case "config": data, err := os.ReadFile(h.Process.ConfigPath(name)) if err != nil { http.NotFound(w, r) return } w.Header().Set("Content-Type", "application/json") _, _ = w.Write(data) case "logs": path := h.Process.LogPath(name) data, err := tailFile(path, 64*1024) if err != nil { http.Error(w, err.Error(), http.StatusInternalServerError) return } _, _ = w.Write([]byte(data)) default: http.NotFound(w, r) } } // ---- helpers ---- // tailFile returns the last maxBytes bytes of a file. func tailFile(path string, maxBytes int64) (string, error) { info, err := os.Stat(path) if err != nil { if os.IsNotExist(err) { return "", nil } return "", err } size := min(info.Size(), maxBytes) f, err := os.Open(path) if err != nil { return "", err } defer f.Close() buf := make([]byte, size) if _, err := f.ReadAt(buf, info.Size()-size); err != nil { return "", err } return string(buf), nil }