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) }