package api import ( "fmt" "net/http" "github.com/owen/vibeproxy/internal/database" "github.com/owen/vibeproxy/internal/models" ) func (a *API) listStreams(w http.ResponseWriter, r *http.Request) { s, err := database.ListStreams(a.db) if err != nil { writeError(w, http.StatusInternalServerError, err.Error()) return } writeJSON(w, http.StatusOK, s) } func (a *API) getStream(w http.ResponseWriter, r *http.Request) { id, err := pathID(r) if err != nil { writeError(w, http.StatusBadRequest, "invalid id") return } s, err := database.GetStream(a.db, id) if err != nil { writeError(w, http.StatusNotFound, "stream not found") return } writeJSON(w, http.StatusOK, s) } func (a *API) createStream(w http.ResponseWriter, r *http.Request) { var s models.Stream if !decodeJSON(w, r, &s) { return } if err := validateStream(&s); err != nil { writeError(w, http.StatusBadRequest, err.Error()) return } s.ID = 0 id, err := database.SaveStream(a.db, &s) if err != nil { writeError(w, http.StatusInternalServerError, err.Error()) return } a.reloadStreams(r) audit(a.db, r, "stream_changed", fmt.Sprintf("id=%d", id), fmt.Sprintf("%s :%d -> %s:%d", s.Protocol, s.ListenPort, s.DestinationHost, s.DestinationPort)) writeJSON(w, http.StatusCreated, s) } func (a *API) updateStream(w http.ResponseWriter, r *http.Request) { id, err := pathID(r) if err != nil { writeError(w, http.StatusBadRequest, "invalid id") return } var s models.Stream if !decodeJSON(w, r, &s) { return } s.ID = id if err := validateStream(&s); err != nil { writeError(w, http.StatusBadRequest, err.Error()) return } if _, err := database.SaveStream(a.db, &s); err != nil { writeError(w, http.StatusInternalServerError, err.Error()) return } a.reloadStreams(r) audit(a.db, r, "stream_changed", fmt.Sprintf("id=%d", id), "") writeJSON(w, http.StatusOK, s) } func (a *API) deleteStream(w http.ResponseWriter, r *http.Request) { id, err := pathID(r) if err != nil { writeError(w, http.StatusBadRequest, "invalid id") return } if err := database.DeleteStream(a.db, id); err != nil { writeError(w, http.StatusInternalServerError, err.Error()) return } a.reloadStreams(r) audit(a.db, r, "stream_changed", fmt.Sprintf("id=%d", id), "deleted") w.WriteHeader(http.StatusNoContent) } func (a *API) enableStream(w http.ResponseWriter, r *http.Request) { a.setStreamEnabled(w, r, true) } func (a *API) disableStream(w http.ResponseWriter, r *http.Request) { a.setStreamEnabled(w, r, false) } func (a *API) setStreamEnabled(w http.ResponseWriter, r *http.Request, enabled bool) { id, err := pathID(r) if err != nil { writeError(w, http.StatusBadRequest, "invalid id") return } if err := database.SetStreamEnabled(a.db, id, enabled); err != nil { writeError(w, http.StatusInternalServerError, err.Error()) return } a.reloadStreams(r) audit(a.db, r, "stream_changed", fmt.Sprintf("id=%d", id), fmt.Sprintf("enabled=%v", enabled)) w.WriteHeader(http.StatusNoContent) } func (a *API) reloadStreams(r *http.Request) { if err := a.streams.Reload(r.Context()); err != nil { a.log.Error("stream reload failed", "err", err) } } func validateStream(s *models.Stream) error { if s.Protocol != models.StreamTCP && s.Protocol != models.StreamUDP { return fmt.Errorf("protocol must be tcp or udp") } if s.ListenPort <= 0 || s.ListenPort > 65535 { return fmt.Errorf("invalid listen port") } if s.DestinationHost == "" { return fmt.Errorf("destination host is required") } if s.DestinationPort <= 0 || s.DestinationPort > 65535 { return fmt.Errorf("invalid destination port") } if s.ConnectTimeoutMS <= 0 { s.ConnectTimeoutMS = 5000 } if s.IdleTimeoutMS <= 0 { s.IdleTimeoutMS = 300000 } return nil }