yovision/Sense/server/app/sense/media/service.go

package media

import (
	"context"
	"strings"
	"time"

	"yovision.local/sense/app/sense/adapters/mediamtx"
	"yovision.local/sense/app/sense/admission"
	"yovision.local/sense/app/sense/device"
	"yovision.local/sense/app/sense/identity"
)

type Service struct {
	store      Store
	process    mediamtx.Process
	controller mediamtx.Controller
	profile    func(string, string) (admission.Profile, error)
	credential func(context.Context, string) (device.Credential, error)
	now        func() time.Time
}

func NewService(store Store, process mediamtx.Process, controller mediamtx.Controller) *Service {
	return &Service{store: store, process: process, controller: controller, profile: admission.VerifiedProfile, credential: device.ReadCredential, now: time.Now}
}
func (s *Service) Configure(ctx context.Context, actor identity.Principal, deviceID, profileToken string) (Route, error) {
	if _, err := s.profile(deviceID, profileToken); err != nil {
		return Route{}, err
	}
	id := deviceID + ":" + profileToken
	path := "sense_" + safe(deviceID) + "_" + safe(profileToken)
	now := s.now().UTC()
	route := Route{ID: id, DeviceID: deviceID, ProfileToken: profileToken, Path: path, Desired: "running", Actual: "pending", Detail: "等待状态对账", Version: 1, UpdatedAt: now}
	if previous, err := s.store.Get(ctx, id); err == nil {
		route.Version = previous.Version + 1
	}
	if err := s.store.Save(ctx, route); err != nil {
		return Route{}, err
	}
	identity.RecordAudit(ctx, actor.UserID, "media.configure", id, "success", map[string]any{"path": path})
	return route, nil
}
func (s *Service) Reconcile(ctx context.Context, actor identity.Principal, id string) (Route, error) {
	route, err := s.store.Get(ctx, id)
	if err != nil {
		return Route{}, err
	}
	if route.Desired == "stopped" {
		route.Actual = "stopped"
		route.Detail = "已停止"
		route.Readers = 0
		route.Version++
		route.UpdatedAt = s.now().UTC()
		return route, s.store.Save(ctx, route)
	}
	if err := s.process.Start(ctx); err != nil {
		route.Actual = "process_failed"
		route.Detail = err.Error()
		route.Version++
		route.UpdatedAt = s.now().UTC()
		_ = s.store.Save(ctx, route)
		return route, nil
	}
	profile, err := s.profile(route.DeviceID, route.ProfileToken)
	if err != nil {
		return Route{}, err
	}
	credential, err := s.credential(ctx, route.DeviceID)
	if err != nil {
		return Route{}, err
	}
	source := mediamtx.Source{Path: route.Path, URI: profile.StreamURI, Username: credential.Username, Password: credential.Password}
	if err := s.controller.Apply(ctx, source); err != nil {
		route.Actual = "apply_failed"
		route.Detail = err.Error()
		route.Version++
		route.UpdatedAt = s.now().UTC()
		_ = s.store.Save(ctx, route)
		return route, nil
	}
	status, err := s.controller.Status(ctx, route.Path)
	if err != nil {
		route.Actual = "unconverged"
		route.Detail = "尚未取得媒体状态"
	} else if status.Ready {
		route.Actual = "ready"
		route.Detail = "上游拉流正常"
		route.Readers = status.Readers
	} else {
		route.Actual = "waiting"
		route.Detail = "等待上游拉流"
		route.Readers = status.Readers
	}
	route.Version++
	route.UpdatedAt = s.now().UTC()
	if err := s.store.Save(ctx, route); err != nil {
		return Route{}, err
	}
	identity.RecordAudit(ctx, actor.UserID, "media.reconcile", id, "success", map[string]any{"actual": route.Actual})
	return route, nil
}
func (s *Service) Stop(ctx context.Context, actor identity.Principal, id string) (Route, error) {
	route, err := s.store.Get(ctx, id)
	if err != nil {
		return Route{}, err
	}
	route.Desired = "stopped"
	route.Actual = "stopped"
	route.Readers = 0
	route.Detail = "已停止"
	route.Version++
	route.UpdatedAt = s.now().UTC()
	if err := s.store.Save(ctx, route); err != nil {
		return Route{}, err
	}
	items, _ := s.store.List(ctx)
	allStopped := true
	for _, item := range items {
		if item.ID != id && item.Desired == "running" {
			allStopped = false
		}
	}
	if allStopped {
		stopCtx, cancel := context.WithTimeout(ctx, 5*time.Second)
		defer cancel()
		_ = s.process.Stop(stopCtx)
	}
	identity.RecordAudit(ctx, actor.UserID, "media.stop", id, "success", nil)
	return route, nil
}
func (s *Service) List(ctx context.Context) ([]Route, error) { return s.store.List(ctx) }
func (s *Service) ProcessState() mediamtx.ProcessState       { return s.process.State() }
func safe(value string) string {
	return strings.Map(func(r rune) rune {
		if r >= 'a' && r <= 'z' || r >= 'A' && r <= 'Z' || r >= '0' && r <= '9' || r == '_' || r == '-' {
			return r
		}
		return '_'
	}, value)
}