306 lines
11 KiB
Go
306 lines
11 KiB
Go
package media
|
|
|
|
import (
|
|
"context"
|
|
"crypto/sha256"
|
|
"errors"
|
|
"fmt"
|
|
"strings"
|
|
"sync"
|
|
"time"
|
|
|
|
"gorm.io/gorm"
|
|
"gorm.io/gorm/clause"
|
|
|
|
"git.ilapage.cn/ila/yovision/Sense/server/app/sense/credential"
|
|
"git.ilapage.cn/ila/yovision/Sense/server/app/sense/reconcile"
|
|
)
|
|
|
|
var (
|
|
ErrNotFound = errors.New("媒体路由不存在")
|
|
ErrRuntimeUnavailable = errors.New("视频服务运行配置不可用")
|
|
)
|
|
|
|
type Service struct {
|
|
db *gorm.DB
|
|
controller Controller
|
|
process *Supervisor
|
|
config RuntimeConfig
|
|
now func() time.Time
|
|
mu sync.Mutex
|
|
}
|
|
|
|
func NewService(db *gorm.DB, controller Controller, process *Supervisor, config RuntimeConfig) *Service {
|
|
return &Service{db: db, controller: controller, process: process, config: config, now: time.Now}
|
|
}
|
|
|
|
func (s *Service) EnsureAllVerified(ctx context.Context) error {
|
|
var ids []string
|
|
if err := s.db.WithContext(ctx).Model(&admissionProfile{}).Where("verification_status = ?", "ready").Distinct().Pluck("device_id", &ids).Error; err != nil {
|
|
return err
|
|
}
|
|
for _, id := range ids {
|
|
if err := s.ensureDevice(ctx, id, false); err != nil {
|
|
return err
|
|
}
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func (s *Service) EnsureDevice(ctx context.Context, deviceID string) error {
|
|
return s.ensureDevice(ctx, deviceID, true)
|
|
}
|
|
|
|
func (s *Service) ensureDevice(ctx context.Context, deviceID string, reactivate bool) error {
|
|
if strings.TrimSpace(deviceID) == "" {
|
|
return errors.New("device id is required")
|
|
}
|
|
var profiles []admissionProfile
|
|
if err := s.db.WithContext(ctx).Where("device_id = ? AND verification_status = ?", deviceID, "ready").Find(&profiles).Error; err != nil {
|
|
return err
|
|
}
|
|
now := s.now().UTC()
|
|
for _, profile := range profiles {
|
|
id := profile.DeviceID + ":" + profile.Token
|
|
digest := sha256.Sum256([]byte(id))
|
|
route := Route{ID: id, DeviceID: profile.DeviceID, ProfileToken: profile.Token, Path: fmt.Sprintf("sense_%x", digest[:12]), Desired: DesiredRunning, Actual: "pending", Detail: "等待视频服务对账", Version: 1, UpdatedAt: now}
|
|
if err := s.db.WithContext(ctx).Clauses(clause.OnConflict{Columns: []clause.Column{{Name: "id"}}, DoNothing: true}).Create(&route).Error; err != nil {
|
|
return err
|
|
}
|
|
var existing Route
|
|
if err := s.db.WithContext(ctx).First(&existing, "id = ?", id).Error; err != nil {
|
|
return err
|
|
}
|
|
if reactivate && existing.Desired != DesiredRunning {
|
|
if err := s.db.WithContext(ctx).Model(&existing).Updates(map[string]any{"desired": DesiredRunning, "actual": "pending", "detail": "等待视频服务对账", "next_retry_at": nil, "version": existing.Version + 1, "updated_at": now}).Error; err != nil {
|
|
return err
|
|
}
|
|
}
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func (s *Service) Run(ctx context.Context) {
|
|
_ = s.EnsureAllVerified(ctx)
|
|
_ = s.ReconcileDue(ctx)
|
|
interval := s.config.PollInterval
|
|
if interval <= 0 {
|
|
interval = 5 * time.Second
|
|
}
|
|
ticker := time.NewTicker(interval)
|
|
defer ticker.Stop()
|
|
for {
|
|
select {
|
|
case <-ctx.Done():
|
|
return
|
|
case <-ticker.C:
|
|
_ = s.ReconcileDue(ctx)
|
|
}
|
|
}
|
|
}
|
|
|
|
func (s *Service) ReconcileDue(ctx context.Context) error {
|
|
now := s.now().UTC()
|
|
var routes []Route
|
|
if err := s.db.WithContext(ctx).Where("desired = ? AND (next_retry_at IS NULL OR next_retry_at <= ?)", DesiredRunning, now).Order("updated_at ASC").Limit(128).Find(&routes).Error; err != nil {
|
|
return err
|
|
}
|
|
var result error
|
|
for _, route := range routes {
|
|
var err error
|
|
if route.FailureCount == 0 && (route.Actual == "waiting" || route.Actual == "ready") {
|
|
_, err = s.Refresh(ctx, route.ID)
|
|
} else {
|
|
_, err = s.Reconcile(ctx, route.ID)
|
|
}
|
|
if err != nil {
|
|
result = errors.Join(result, err)
|
|
}
|
|
}
|
|
return result
|
|
}
|
|
|
|
func (s *Service) Refresh(ctx context.Context, id string) (RouteResponse, error) {
|
|
s.mu.Lock()
|
|
defer s.mu.Unlock()
|
|
var route Route
|
|
if err := s.db.WithContext(ctx).First(&route, "id = ?", id).Error; err != nil {
|
|
return RouteResponse{}, ErrNotFound
|
|
}
|
|
if err := s.ensureControl(ctx); err != nil {
|
|
return s.saveFailure(ctx, route, "process_unavailable", "MediaMTX 未启动或 Control API 未就绪", err)
|
|
}
|
|
status, err := s.controller.Status(ctx, route.Path)
|
|
if err != nil {
|
|
return s.saveFailure(ctx, route, "status_unavailable", "尚未取得媒体路径状态", err)
|
|
}
|
|
if !status.Exists {
|
|
// A cold MediaMTX start begins with paths: {}; re-apply the route
|
|
// after releasing the service lock.
|
|
s.mu.Unlock()
|
|
response, reconcileErr := s.Reconcile(ctx, id)
|
|
s.mu.Lock()
|
|
return response, reconcileErr
|
|
}
|
|
if status.Ready {
|
|
return s.saveSuccess(ctx, route, "ready", true, status.Readers, "上游拉流正常")
|
|
}
|
|
return s.saveSuccess(ctx, route, "waiting", false, status.Readers, "等待播放器连接并按需拉流")
|
|
}
|
|
|
|
func (s *Service) Reconcile(ctx context.Context, id string) (RouteResponse, error) {
|
|
s.mu.Lock()
|
|
defer s.mu.Unlock()
|
|
var route Route
|
|
if err := s.db.WithContext(ctx).First(&route, "id = ?", id).Error; err != nil {
|
|
if errors.Is(err, gorm.ErrRecordNotFound) {
|
|
return RouteResponse{}, ErrNotFound
|
|
}
|
|
return RouteResponse{}, err
|
|
}
|
|
if route.Desired == DesiredStopped {
|
|
if s.controller != nil {
|
|
_ = s.controller.Delete(ctx, route.Path)
|
|
}
|
|
return s.saveSuccess(ctx, route, "stopped", false, 0, "媒体路径已停止")
|
|
}
|
|
if err := s.ensureControl(ctx); err != nil {
|
|
return s.saveFailure(ctx, route, "process_unavailable", "MediaMTX 未启动或 Control API 未就绪", err)
|
|
}
|
|
var profile admissionProfile
|
|
if err := s.db.WithContext(ctx).Where("device_id = ? AND token = ? AND verification_status = ?", route.DeviceID, route.ProfileToken, "ready").First(&profile).Error; err != nil {
|
|
return s.saveFailure(ctx, route, "profile_unavailable", "已验证 Profile 不可用", err)
|
|
}
|
|
value, err := credential.Read(s.db.WithContext(ctx), route.DeviceID, credential.PurposeRTSP)
|
|
if err != nil {
|
|
return s.saveFailure(ctx, route, "credential_unavailable", "RTSP 凭据不可用", err)
|
|
}
|
|
if err = s.controller.Apply(ctx, Source{Path: route.Path, URI: profile.StreamURI, Username: value.Username, Password: value.Password}); err != nil {
|
|
return s.saveFailure(ctx, route, "apply_failed", "媒体路径配置失败", err)
|
|
}
|
|
status, err := s.controller.Status(ctx, route.Path)
|
|
if err != nil {
|
|
return s.saveFailure(ctx, route, "status_unavailable", "尚未取得媒体路径状态", err)
|
|
}
|
|
if !status.Exists {
|
|
return s.saveFailure(ctx, route, "path_missing", "媒体路径不存在", errors.New("MediaMTX path missing after apply"))
|
|
}
|
|
if status.Ready {
|
|
return s.saveSuccess(ctx, route, "ready", true, status.Readers, "上游拉流正常")
|
|
}
|
|
return s.saveSuccess(ctx, route, "waiting", false, status.Readers, "等待播放器连接并按需拉流")
|
|
}
|
|
|
|
func (s *Service) ensureControl(ctx context.Context) error {
|
|
if s.controller == nil {
|
|
return ErrRuntimeUnavailable
|
|
}
|
|
if err := s.controller.Health(ctx); err == nil {
|
|
if s.process != nil && !s.process.State().Owned {
|
|
s.process.MarkExternal()
|
|
}
|
|
return nil
|
|
}
|
|
if s.process == nil {
|
|
return ErrRuntimeUnavailable
|
|
}
|
|
if err := EnsureBaseConfig(s.config.ConfigPath, s.config.APIBase); err != nil {
|
|
s.process.MarkFailed("MediaMTX 基础配置不可用")
|
|
return err
|
|
}
|
|
if err := s.process.Start(ctx); err != nil {
|
|
return err
|
|
}
|
|
deadline := s.now().Add(s.config.StartTimeout)
|
|
for s.now().Before(deadline) {
|
|
if err := s.controller.Health(ctx); err == nil {
|
|
s.process.MarkReady()
|
|
return nil
|
|
}
|
|
select {
|
|
case <-ctx.Done():
|
|
return ctx.Err()
|
|
case <-time.After(100 * time.Millisecond):
|
|
}
|
|
}
|
|
s.process.MarkFailed("MediaMTX Control API 就绪超时")
|
|
return errors.New("MediaMTX readiness timeout")
|
|
}
|
|
|
|
func (s *Service) saveSuccess(ctx context.Context, route Route, actual string, sourceReady bool, readers int, detail string) (RouteResponse, error) {
|
|
if route.Actual == actual && route.SourceReady == sourceReady && route.Readers == readers && route.FailureCount == 0 && route.NextRetryAt == nil && route.LastErrorCode == "" && route.Detail == detail {
|
|
return routeResponse(route), nil
|
|
}
|
|
now := s.now().UTC()
|
|
updates := map[string]any{"actual": actual, "source_ready": sourceReady, "readers": readers, "failure_count": 0, "next_retry_at": nil, "last_error_code": "", "detail": detail, "version": route.Version + 1, "updated_at": now}
|
|
if err := s.db.WithContext(ctx).Model(&route).Updates(updates).Error; err != nil {
|
|
return RouteResponse{}, err
|
|
}
|
|
for key, value := range updates {
|
|
switch key {
|
|
case "actual":
|
|
route.Actual = value.(string)
|
|
case "source_ready":
|
|
route.SourceReady = value.(bool)
|
|
case "readers":
|
|
route.Readers = value.(int)
|
|
case "failure_count":
|
|
route.FailureCount = value.(int)
|
|
case "last_error_code":
|
|
route.LastErrorCode = value.(string)
|
|
case "detail":
|
|
route.Detail = value.(string)
|
|
case "version":
|
|
route.Version = value.(int64)
|
|
case "updated_at":
|
|
route.UpdatedAt = value.(time.Time)
|
|
}
|
|
}
|
|
route.NextRetryAt = nil
|
|
return routeResponse(route), nil
|
|
}
|
|
|
|
func (s *Service) saveFailure(ctx context.Context, route Route, code, detail string, cause error) (RouteResponse, error) {
|
|
now := s.now().UTC()
|
|
failures := route.FailureCount + 1
|
|
next := now.Add(reconcile.Backoff(failures))
|
|
updates := map[string]any{"actual": code, "source_ready": false, "readers": 0, "failure_count": failures, "next_retry_at": &next, "last_error_code": code, "detail": detail, "version": route.Version + 1, "updated_at": now}
|
|
if err := s.db.WithContext(ctx).Model(&route).Updates(updates).Error; err != nil {
|
|
return RouteResponse{}, errors.Join(cause, err)
|
|
}
|
|
route.Actual, route.SourceReady, route.Readers, route.FailureCount, route.NextRetryAt, route.LastErrorCode, route.Detail = code, false, 0, failures, &next, code, detail
|
|
route.Version, route.UpdatedAt = route.Version+1, now
|
|
return routeResponse(route), cause
|
|
}
|
|
|
|
func (s *Service) StopRoute(ctx context.Context, id string) (RouteResponse, error) {
|
|
var route Route
|
|
if err := s.db.WithContext(ctx).First(&route, "id = ?", id).Error; err != nil {
|
|
return RouteResponse{}, ErrNotFound
|
|
}
|
|
route.Desired = DesiredStopped
|
|
if err := s.db.WithContext(ctx).Model(&route).Updates(map[string]any{"desired": DesiredStopped, "next_retry_at": nil, "version": route.Version + 1, "updated_at": s.now().UTC()}).Error; err != nil {
|
|
return RouteResponse{}, err
|
|
}
|
|
return s.Reconcile(ctx, id)
|
|
}
|
|
|
|
func (s *Service) List(ctx context.Context) ([]RouteResponse, error) {
|
|
var routes []Route
|
|
if err := s.db.WithContext(ctx).Order("updated_at DESC").Find(&routes).Error; err != nil {
|
|
return nil, err
|
|
}
|
|
items := make([]RouteResponse, 0, len(routes))
|
|
for _, route := range routes {
|
|
items = append(items, routeResponse(route))
|
|
}
|
|
return items, nil
|
|
}
|
|
|
|
func (s *Service) ProcessState() ProcessState {
|
|
if s.process == nil {
|
|
return ProcessState{Phase: "configuration_failed", Detail: "视频服务运行配置不可用", LastChanged: s.now().UTC()}
|
|
}
|
|
return s.process.State()
|
|
}
|