yovision/Sense/internal/mtx/client.go

package mtx

import (
	"context"
	"errors"
	"fmt"
	"net/http"
	"sort"
	"strings"

	mediamtxapi "yovision/sense/internal/mtx/generated"
)

var ErrPathNotFound = errors.New("MediaMTX path not found")

type APIError struct {
	Operation  string
	StatusCode int
}

func (e *APIError) Error() string {
	return fmt.Sprintf("MediaMTX %s failed with HTTP status %d", e.Operation, e.StatusCode)
}

type PathConfig struct {
	Name   string
	Source string
}

type pathAPI interface {
	ConfigPathsAddWithResponse(context.Context, string, mediamtxapi.ConfigPathsAddJSONRequestBody, ...mediamtxapi.RequestEditorFn) (*mediamtxapi.ConfigPathsAddResponse, error)
	ConfigPathsGetWithResponse(context.Context, string, ...mediamtxapi.RequestEditorFn) (*mediamtxapi.ConfigPathsGetResponse, error)
	ConfigPathsPatchWithResponse(context.Context, string, mediamtxapi.ConfigPathsPatchJSONRequestBody, ...mediamtxapi.RequestEditorFn) (*mediamtxapi.ConfigPathsPatchResponse, error)
	ConfigPathsDeleteWithResponse(context.Context, string, ...mediamtxapi.RequestEditorFn) (*mediamtxapi.ConfigPathsDeleteResponse, error)
	ConfigPathsListWithResponse(context.Context, *mediamtxapi.ConfigPathsListParams, ...mediamtxapi.RequestEditorFn) (*mediamtxapi.ConfigPathsListResponse, error)
	PathsGetWithResponse(context.Context, string, ...mediamtxapi.RequestEditorFn) (*mediamtxapi.PathsGetResponse, error)
}

// ListPathNames enumerates only configuration names. Sources are deliberately
// discarded so inventory and orphan reports cannot expose stream URIs.
func (c *Client) ListPathNames(ctx context.Context) ([]string, error) {
	const (
		itemsPerPage = 100
		maxPages     = 1000
	)
	result := make(map[string]struct{})
	seenPages := make(map[string]struct{})
	for page := 0; page < maxPages; page++ {
		pageValue, limitValue := page, itemsPerPage
		response, err := c.api.ConfigPathsListWithResponse(ctx, &mediamtxapi.ConfigPathsListParams{
			Page: &pageValue, ItemsPerPage: &limitValue,
		})
		if err != nil {
			return nil, fmt.Errorf("MediaMTX list paths transport: %w", err)
		}
		if response.StatusCode() != http.StatusOK || response.JSON200 == nil ||
			response.JSON200.PageCount == nil || response.JSON200.Items == nil {
			return nil, &APIError{Operation: "list paths", StatusCode: response.StatusCode()}
		}
		pageCount := *response.JSON200.PageCount
		if pageCount < 0 || pageCount > maxPages {
			return nil, &APIError{Operation: "list paths pagination", StatusCode: response.StatusCode()}
		}
		pageNames := make([]string, 0, len(*response.JSON200.Items))
		for _, item := range *response.JSON200.Items {
			if item.Name == nil || strings.TrimSpace(*item.Name) == "" {
				return nil, &APIError{Operation: "list paths response", StatusCode: response.StatusCode()}
			}
			pageNames = append(pageNames, *item.Name)
			result[*item.Name] = struct{}{}
		}
		sort.Strings(pageNames)
		signature := strings.Join(pageNames, "\x00")
		if page > 0 && signature != "" {
			if _, duplicate := seenPages[signature]; duplicate {
				return nil, &APIError{Operation: "list paths repeated page", StatusCode: response.StatusCode()}
			}
		}
		seenPages[signature] = struct{}{}
		if int64(page+1) >= pageCount {
			values := make([]string, 0, len(result))
			for name := range result {
				values = append(values, name)
			}
			sort.Strings(values)
			return values, nil
		}
	}
	return nil, &APIError{Operation: "list paths page limit", StatusCode: http.StatusOK}
}

type Client struct {
	api pathAPI
}

func NewClient(baseURL string, httpClient *http.Client) (*Client, error) {
	options := make([]mediamtxapi.ClientOption, 0, 1)
	if httpClient != nil {
		options = append(options, mediamtxapi.WithHTTPClient(httpClient))
	}
	generated, err := mediamtxapi.NewClientWithResponses(strings.TrimRight(baseURL, "/"), options...)
	if err != nil {
		return nil, fmt.Errorf("create MediaMTX client: %w", err)
	}
	return &Client{api: generated}, nil
}

func newClientWithAPI(api pathAPI) *Client {
	return &Client{api: api}
}

func (c *Client) CreatePath(ctx context.Context, name, source string) error {
	response, err := c.api.ConfigPathsAddWithResponse(ctx, name, mediamtxapi.PathConf{Source: &source})
	if err != nil {
		return fmt.Errorf("MediaMTX create path transport: %w", err)
	}
	if response.StatusCode() != http.StatusOK {
		return &APIError{Operation: "create path", StatusCode: response.StatusCode()}
	}
	return nil
}

func (c *Client) GetPath(ctx context.Context, name string) (PathConfig, error) {
	response, err := c.api.ConfigPathsGetWithResponse(ctx, name)
	if err != nil {
		return PathConfig{}, fmt.Errorf("MediaMTX read path transport: %w", err)
	}
	if response.StatusCode() == http.StatusNotFound {
		return PathConfig{}, ErrPathNotFound
	}
	if response.StatusCode() != http.StatusOK || response.JSON200 == nil {
		return PathConfig{}, &APIError{Operation: "read path", StatusCode: response.StatusCode()}
	}
	result := PathConfig{Name: name}
	if response.JSON200.Name != nil {
		result.Name = *response.JSON200.Name
	}
	if response.JSON200.Source != nil {
		result.Source = *response.JSON200.Source
	}
	return result, nil
}

func (c *Client) DeletePath(ctx context.Context, name string) error {
	response, err := c.api.ConfigPathsDeleteWithResponse(ctx, name)
	if err != nil {
		return fmt.Errorf("MediaMTX delete path transport: %w", err)
	}
	if response.StatusCode() == http.StatusNotFound {
		// Deletion is an idempotent convergence operation. A missing exact path
		// already satisfies the disabled desired state.
		return nil
	}
	if response.StatusCode() != http.StatusOK {
		return &APIError{Operation: "delete path", StatusCode: response.StatusCode()}
	}
	return nil
}

// EnsurePath converges one desired path. It never enumerates or deletes orphans.
func (c *Client) EnsurePath(ctx context.Context, name, source string) (bool, error) {
	current, err := c.GetPath(ctx, name)
	if errors.Is(err, ErrPathNotFound) {
		if err := c.CreatePath(ctx, name, source); err != nil {
			return false, err
		}
		return true, nil
	}
	if err != nil {
		return false, err
	}
	if current.Source == source {
		return false, nil
	}
	response, err := c.api.ConfigPathsPatchWithResponse(ctx, name, mediamtxapi.PathConf{Source: &source})
	if err != nil {
		return false, fmt.Errorf("MediaMTX patch path transport: %w", err)
	}
	if response.StatusCode() != http.StatusOK {
		return false, &APIError{Operation: "patch path", StatusCode: response.StatusCode()}
	}
	return true, nil
}

func (c *Client) PathReady(ctx context.Context, name string) (bool, error) {
	response, err := c.api.PathsGetWithResponse(ctx, name)
	if err != nil {
		return false, fmt.Errorf("MediaMTX probe path transport: %w", err)
	}
	if response.StatusCode() == http.StatusNotFound {
		return false, ErrPathNotFound
	}
	if response.StatusCode() != http.StatusOK || response.JSON200 == nil {
		return false, &APIError{Operation: "probe path", StatusCode: response.StatusCode()}
	}
	if response.JSON200.Online == nil || response.JSON200.Available == nil {
		return false, &APIError{Operation: "probe path response", StatusCode: response.StatusCode()}
	}
	return *response.JSON200.Online && *response.JSON200.Available, nil
}