package media import ( "bytes" "context" "encoding/json" "errors" "fmt" "io" "net/http" "net/url" "regexp" "strings" "time" ) var validPath = regexp.MustCompile(`^[A-Za-z0-9_-]{1,96}$`) type Source struct { Path, URI, Username, Password string } type PathStatus struct { Exists, Ready bool Readers int } type Controller interface { Health(context.Context) error Apply(context.Context, Source) error Delete(context.Context, string) error Status(context.Context, string) (PathStatus, error) } type HTTPController struct { base string client *http.Client } func NewHTTPController(base string) (*HTTPController, error) { if err := validateControlAPI(base); err != nil { return nil, err } transport := http.DefaultTransport.(*http.Transport).Clone() transport.Proxy = nil return &HTTPController{base: strings.TrimRight(base, "/"), client: &http.Client{ Timeout: 5 * time.Second, Transport: transport, CheckRedirect: func(*http.Request, []*http.Request) error { return errors.New("MediaMTX Control API redirect rejected") }, }}, nil } func (c *HTTPController) Health(ctx context.Context) error { return c.request(ctx, http.MethodGet, "/v3/config/global/get", nil, nil) } func (c *HTTPController) Apply(ctx context.Context, source Source) error { if !validPath.MatchString(source.Path) { return errors.New("invalid MediaMTX path") } u, err := url.Parse(source.URI) if err != nil || u.User != nil || u.Host == "" || (u.Scheme != "rtsp" && u.Scheme != "rtsps") { return errors.New("invalid credential-free RTSP source") } payload := map[string]any{"source": u.String(), "sourceOnDemand": true, "rtspTransport": "tcp"} if source.Username != "" { // MediaMTX v1.19.3 has no sourceUser/sourcePass fields. Credentials // are assembled only for this loopback request and are never stored, // logged, or returned by a Sense endpoint. u.User = url.UserPassword(source.Username, source.Password) payload["source"] = u.String() } data, err := json.Marshal(payload) if err != nil { return err } configured, err := c.configured(ctx, source.Path) if err != nil { return err } method, action := http.MethodPost, "add" if configured { method, action = http.MethodPatch, "patch" } return c.request(ctx, method, "/v3/config/paths/"+action+"/"+url.PathEscape(source.Path), bytes.NewReader(data), nil) } func (c *HTTPController) configured(ctx context.Context, path string) (bool, error) { status := 0 err := c.request(ctx, http.MethodGet, "/v3/config/paths/get/"+url.PathEscape(path), nil, &status) if status == http.StatusNotFound { return false, nil } return err == nil, err } func (c *HTTPController) Delete(ctx context.Context, path string) error { if !validPath.MatchString(path) { return errors.New("invalid MediaMTX path") } status := 0 err := c.request(ctx, http.MethodDelete, "/v3/config/paths/delete/"+url.PathEscape(path), nil, &status) if status == http.StatusNotFound { return nil } return err } func (c *HTTPController) Status(ctx context.Context, path string) (PathStatus, error) { statusCode := 0 var raw struct { Ready bool `json:"ready"` Readers []any `json:"readers"` } err := c.requestJSON(ctx, http.MethodGet, "/v3/paths/get/"+url.PathEscape(path), &statusCode, &raw) if statusCode == http.StatusNotFound { return PathStatus{}, nil } if err != nil { return PathStatus{}, err } return PathStatus{Exists: true, Ready: raw.Ready, Readers: len(raw.Readers)}, nil } func (c *HTTPController) request(ctx context.Context, method, path string, body io.Reader, statusOut *int) error { return c.requestJSON(ctx, method, path, body, statusOut, nil) } func (c *HTTPController) requestJSON(ctx context.Context, method, path string, args ...any) error { var body io.Reader var statusOut *int var target any for _, arg := range args { switch value := arg.(type) { case io.Reader: body = value case *int: statusOut = value default: target = value } } req, err := http.NewRequestWithContext(ctx, method, c.base+path, body) if err != nil { return err } if body != nil { req.Header.Set("Content-Type", "application/json") } res, err := c.client.Do(req) if err != nil { return fmt.Errorf("MediaMTX Control API unavailable: %w", err) } defer res.Body.Close() if statusOut != nil { *statusOut = res.StatusCode } if res.StatusCode < 200 || res.StatusCode >= 300 { _, _ = io.Copy(io.Discard, io.LimitReader(res.Body, 1<<20)) return fmt.Errorf("MediaMTX Control API returned %d", res.StatusCode) } if target == nil { _, _ = io.Copy(io.Discard, io.LimitReader(res.Body, 1<<20)) return nil } return json.NewDecoder(io.LimitReader(res.Body, 1<<20)).Decode(target) }