vibeproxy/internal/api/streams.go

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
}