235 lines
7.3 KiB
Go
235 lines
7.3 KiB
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.ReadRTSPCredential, now: time.Now}
|
|
}
|
|
|
|
// Restore recreates desired routes after MediaMTX starts with its base config.
|
|
// Individual route failures are persisted as safe business states and do not
|
|
// prevent the Sense management plane from starting.
|
|
func (s *Service) Restore(ctx context.Context) error {
|
|
items, err := s.store.List(ctx)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
for _, route := range items {
|
|
if route.Desired != "running" {
|
|
continue
|
|
}
|
|
if _, reconcileErr := s.Reconcile(ctx, identity.Principal{}, route.ID); reconcileErr != nil {
|
|
route.Actual = "apply_failed"
|
|
route.Detail = "媒体路径恢复失败，请检查视频服务"
|
|
route.Readers = 0
|
|
route.Version++
|
|
route.UpdatedAt = s.now().UTC()
|
|
if saveErr := s.store.Save(ctx, route); saveErr != nil {
|
|
return saveErr
|
|
}
|
|
}
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func (s *Service) ConfigureProfiles(ctx context.Context, actor identity.Principal, result admission.Result) admission.MediaOutcome {
|
|
configured := 0
|
|
ready := 0
|
|
for _, profile := range result.Profiles {
|
|
if profile.Verification.Status != "ready" {
|
|
continue
|
|
}
|
|
configured++
|
|
route, err := s.Configure(ctx, actor, result.DeviceID, profile.Token)
|
|
if err != nil {
|
|
continue
|
|
}
|
|
route, err = s.Reconcile(ctx, actor, route.ID)
|
|
if err == nil && route.Actual == "ready" {
|
|
ready++
|
|
}
|
|
}
|
|
if configured == 0 {
|
|
return admission.MediaOutcome{Status: "needs_attention", Detail: "未找到可用码流，请检查接入结果"}
|
|
}
|
|
if ready == configured {
|
|
return admission.MediaOutcome{Status: "ready", Detail: "视频已进入实时监看"}
|
|
}
|
|
return admission.MediaOutcome{Status: "needs_attention", Detail: "摄像机已接入，媒体服务未就绪，请到视频服务查看并重试"}
|
|
}
|
|
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 = "媒体进程未启动，请检查视频服务配置"
|
|
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 = "媒体路径配置失败，请检查视频服务"
|
|
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.Exists {
|
|
route.Actual = "apply_failed"
|
|
route.Detail = "媒体路径不存在，请重新对账"
|
|
route.Readers = 0
|
|
} 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
|
|
}
|
|
|
|
// Refresh reads the MediaMTX runtime state without reapplying configuration or
|
|
// exposing the source URI. It is safe to call while a playback session polls.
|
|
func (s *Service) Refresh(ctx context.Context, id string) (Route, error) {
|
|
route, err := s.store.Get(ctx, id)
|
|
if err != nil {
|
|
return Route{}, err
|
|
}
|
|
if route.Desired != "running" {
|
|
return route, nil
|
|
}
|
|
status, err := s.controller.Status(ctx, route.Path)
|
|
if err != nil {
|
|
return route, nil
|
|
}
|
|
actual := "waiting"
|
|
detail := "等待播放器连接并按需拉流"
|
|
if !status.Exists {
|
|
actual = "apply_failed"
|
|
detail = "媒体路径不存在，请重新对账"
|
|
} else if status.Ready {
|
|
actual = "ready"
|
|
detail = "上游拉流正常"
|
|
}
|
|
if route.Actual == actual && route.Detail == detail && route.Readers == status.Readers {
|
|
return route, nil
|
|
}
|
|
route.Actual = actual
|
|
route.Detail = detail
|
|
route.Readers = status.Readers
|
|
route.Version++
|
|
route.UpdatedAt = s.now().UTC()
|
|
if err := s.store.Save(ctx, route); err != nil {
|
|
return Route{}, err
|
|
}
|
|
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)
|
|
}
|
|
|