diff --git a/Sense/server/app/sense/adapters/mediamtx/client.go b/Sense/server/app/sense/adapters/mediamtx/client.go index c139c97..a9e54b9 100644 --- a/Sense/server/app/sense/adapters/mediamtx/client.go +++ b/Sense/server/app/sense/adapters/mediamtx/client.go @@ -20,6 +20,7 @@ type Source struct { } type PathStatus struct { Name string `json:"name"` + Exists bool `json:"exists"` Ready bool `json:"ready"` Readers int `json:"readers"` BytesReceived int64 `json:"bytes_received"` @@ -46,7 +47,15 @@ func (c *HTTPController) Apply(ctx context.Context, source Source) error { } payload := map[string]any{"source": parsed.String(), "sourceOnDemand": true, "rtspTransport": "tcp"} data, _ := json.Marshal(payload) - endpoint := c.base + "/v3/config/paths/replace/" + url.PathEscape(source.Path) + configured, err := c.configured(ctx, source.Path) + if err != nil { + return err + } + action := "add" + if configured { + action = "replace" + } + endpoint := c.base + "/v3/config/paths/" + action + "/" + url.PathEscape(source.Path) req, err := http.NewRequestWithContext(ctx, http.MethodPost, endpoint, bytes.NewReader(data)) if err != nil { return err @@ -63,6 +72,39 @@ func (c *HTTPController) Apply(ctx context.Context, source Source) error { } return nil } + +func (c *HTTPController) configured(ctx context.Context, path string) (bool, error) { + endpoint := c.base + "/v3/config/paths/get/" + url.PathEscape(path) + deadline := time.Now().Add(5 * time.Second) + var res *http.Response + for { + req, err := http.NewRequestWithContext(ctx, http.MethodGet, endpoint, nil) + if err != nil { + return false, err + } + res, err = c.client.Do(req) + if err == nil { + break + } + if !time.Now().Before(deadline) { + return false, fmt.Errorf("mediamtx control unavailable: %w", err) + } + select { + case <-ctx.Done(): + return false, ctx.Err() + case <-time.After(100 * time.Millisecond): + } + } + defer res.Body.Close() + io.Copy(io.Discard, io.LimitReader(res.Body, 1<<20)) + if res.StatusCode == http.StatusNotFound { + return false, nil + } + if res.StatusCode < 200 || res.StatusCode >= 300 { + return false, fmt.Errorf("mediamtx config lookup returned %d", res.StatusCode) + } + return true, nil +} func (c *HTTPController) Status(ctx context.Context, path string) (PathStatus, error) { req, err := http.NewRequestWithContext(ctx, http.MethodGet, c.base+"/v3/paths/get/"+url.PathEscape(path), nil) if err != nil { @@ -74,7 +116,7 @@ func (c *HTTPController) Status(ctx context.Context, path string) (PathStatus, e } defer res.Body.Close() if res.StatusCode == http.StatusNotFound { - return PathStatus{Name: path}, nil + return PathStatus{Name: path, Exists: false}, nil } if res.StatusCode < 200 || res.StatusCode >= 300 { return PathStatus{}, fmt.Errorf("mediamtx status returned %d", res.StatusCode) @@ -88,5 +130,6 @@ func (c *HTTPController) Status(ctx context.Context, path string) (PathStatus, e if err := json.NewDecoder(io.LimitReader(res.Body, 1<<20)).Decode(&raw); err != nil { return PathStatus{}, err } - return PathStatus{Name: raw.Name, Ready: raw.Ready, Readers: len(raw.Readers), BytesReceived: raw.BytesReceived}, nil + return PathStatus{Name: raw.Name, Exists: true, Ready: raw.Ready, Readers: len(raw.Readers), BytesReceived: raw.BytesReceived}, nil } + diff --git a/Sense/server/app/sense/adapters/mediamtx/client_test.go b/Sense/server/app/sense/adapters/mediamtx/client_test.go index ce9ab6e..697b540 100644 --- a/Sense/server/app/sense/adapters/mediamtx/client_test.go +++ b/Sense/server/app/sense/adapters/mediamtx/client_test.go @@ -3,18 +3,30 @@ package mediamtx import ( "context" "encoding/json" + "errors" + "io" "net/http" "net/http/httptest" "strings" "testing" ) +type roundTripFunc func(*http.Request) (*http.Response, error) + +func (f roundTripFunc) RoundTrip(request *http.Request) (*http.Response, error) { return f(request) } + func TestApplyBuildsCredentialSourceOnlyInTransientBody(t *testing.T) { var body map[string]any + var appliedPath string server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { if strings.Contains(r.URL.String(), "password") { t.Fatal("credential leaked in control URL") } + if r.Method == http.MethodGet { + http.NotFound(w, r) + return + } + appliedPath = r.URL.Path if err := json.NewDecoder(r.Body).Decode(&body); err != nil { t.Fatal(err) } @@ -28,6 +40,50 @@ func TestApplyBuildsCredentialSourceOnlyInTransientBody(t *testing.T) { if body["source"] != "rtsp://fixture-user:fixture-password@camera.invalid/main" { t.Fatalf("body=%#v", body) } + if appliedPath != "/v3/config/paths/add/sense_test" { + t.Fatalf("applied path = %q", appliedPath) + } +} + +func TestApplyReplacesExistingPath(t *testing.T) { + var appliedPath string + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.Method == http.MethodGet { + w.WriteHeader(http.StatusOK) + return + } + appliedPath = r.URL.Path + w.WriteHeader(http.StatusOK) + })) + defer server.Close() + if err := NewHTTPController(server.URL).Apply(context.Background(), Source{Path: "sense_test", URI: "rtsp://camera.invalid/main"}); err != nil { + t.Fatal(err) + } + if appliedPath != "/v3/config/paths/replace/sense_test" { + t.Fatalf("applied path = %q", appliedPath) + } +} + +func TestApplyWaitsForNewlyStartedControlAPI(t *testing.T) { + attempts := 0 + client := NewHTTPController("http://127.0.0.1:9997") + client.client.Transport = roundTripFunc(func(request *http.Request) (*http.Response, error) { + attempts++ + if attempts < 3 { + return nil, errors.New("connection refused") + } + status := http.StatusNotFound + if request.Method == http.MethodPost { + status = http.StatusOK + } + return &http.Response{StatusCode: status, Body: io.NopCloser(strings.NewReader("")), Header: make(http.Header)}, nil + }) + if err := client.Apply(context.Background(), Source{Path: "sense_test", URI: "rtsp://camera.invalid/main"}); err != nil { + t.Fatal(err) + } + if attempts != 4 { + t.Fatalf("attempts = %d", attempts) + } } func TestApplyRejectsCredentialURI(t *testing.T) { client := NewHTTPController("http://127.0.0.1") @@ -35,3 +91,4 @@ func TestApplyRejectsCredentialURI(t *testing.T) { t.Fatal("credential URI accepted") } } + diff --git a/Sense/server/app/sense/identity/http.go b/Sense/server/app/sense/identity/http.go index 977eee0..6a3f00d 100644 --- a/Sense/server/app/sense/identity/http.go +++ b/Sense/server/app/sense/identity/http.go @@ -48,6 +48,21 @@ func Require(permission string, next http.Handler) http.Handler { }) } +// RequireSameOriginFrame authenticates browser iframe navigation that cannot +// attach the X-Product header used by API clients. Fetch Metadata keeps this +// narrow: only a same-origin iframe navigation with a valid Sense session is +// accepted. +func RequireSameOriginFrame(permission string, next http.Handler) http.Handler { + return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + module := activeModule.Load() + if module == nil { + platform.WriteError(w, &platform.APIError{Status: http.StatusServiceUnavailable, Code: "identity_not_ready", Message: "身份服务尚未就绪"}) + return + } + module.RequireSameOriginFrame(permission, next).ServeHTTP(w, r) + }) +} + func PrincipalFromContext(ctx context.Context) (Principal, bool) { principal, ok := ctx.Value(principalContextKey{}).(Principal) return principal, ok @@ -84,6 +99,28 @@ func (m *Module) Require(permission string, next http.Handler) http.Handler { })) } +func (m *Module) RequireSameOriginFrame(permission string, next http.Handler) http.Handler { + return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.Header.Get("Sec-Fetch-Site") != "same-origin" || + r.Header.Get("Sec-Fetch-Mode") != "navigate" || + r.Header.Get("Sec-Fetch-Dest") != "iframe" { + platform.WriteError(w, &platform.APIError{Status: http.StatusUnauthorized, Code: "unauthorized", Message: "登录状态无效"}) + return + } + cookie, err := r.Cookie(sessionCookieName) + if err != nil { + platform.WriteError(w, &platform.APIError{Status: http.StatusUnauthorized, Code: "unauthorized", Message: "登录状态无效"}) + return + } + principal, err := m.service.Authenticate(r.Context(), cookie.Value) + if err != nil || !HasPermission(principal.Role, permission) { + platform.WriteError(w, &platform.APIError{Status: http.StatusUnauthorized, Code: "unauthorized", Message: "登录状态无效"}) + return + } + next.ServeHTTP(w, r.WithContext(context.WithValue(r.Context(), principalContextKey{}, principal))) + }) +} + func (m *Module) bootstrap(w http.ResponseWriter, r *http.Request) { var request struct { Username string `json:"username"` @@ -220,3 +257,4 @@ func SecureCookieFromEnvironment(value string, memoryMode bool) bool { } return !strings.EqualFold(value, "false") } + diff --git a/Sense/server/app/sense/identity/http_test.go b/Sense/server/app/sense/identity/http_test.go index d96302c..b60396b 100644 --- a/Sense/server/app/sense/identity/http_test.go +++ b/Sense/server/app/sense/identity/http_test.go @@ -21,6 +21,9 @@ func TestHTTPLoginCookieAndProductBoundary(t *testing.T) { } app := platform.NewApp(platform.Config{DatabaseMode: platform.DatabaseModeMemory}, nil, slog.New(slog.NewTextHandler(io.Discard, nil))) NewModule(service, cfg).Register(app) + app.Handle("GET /same-origin-frame", RequireSameOriginFrame(PermissionMediaRead, http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + w.WriteHeader(http.StatusNoContent) + }))) bootstrap := httptest.NewRequest(http.MethodPost, "/api/v1/identity/bootstrap", bytes.NewBufferString(`{"username":"admin","display_name":"管理员","password":"StrongPass2026"}`)) bootstrap.Header.Set("X-Sense-Bootstrap-Token", "bootstrap-test") @@ -56,4 +59,41 @@ func TestHTTPLoginCookieAndProductBoundary(t *testing.T) { if senseResult.Code != http.StatusOK { t.Fatalf("Sense product header status = %d", senseResult.Code) } + + meWithoutProduct := httptest.NewRequest(http.MethodGet, "/api/v1/identity/me", nil) + meWithoutProduct.AddCookie(cookies[0]) + missingProductResult := httptest.NewRecorder() + app.Handler().ServeHTTP(missingProductResult, meWithoutProduct) + if missingProductResult.Code != http.StatusUnauthorized { + t.Fatalf("missing product header status = %d", missingProductResult.Code) + } + + frame := httptest.NewRequest(http.MethodGet, "/same-origin-frame", nil) + frame.AddCookie(cookies[0]) + frame.Header.Set("Sec-Fetch-Site", "same-origin") + frame.Header.Set("Sec-Fetch-Mode", "navigate") + frame.Header.Set("Sec-Fetch-Dest", "iframe") + frameResult := httptest.NewRecorder() + app.Handler().ServeHTTP(frameResult, frame) + if frameResult.Code != http.StatusNoContent { + t.Fatalf("same-origin frame status = %d, body = %s", frameResult.Code, frameResult.Body.String()) + } + + frame.Header.Set("Sec-Fetch-Site", "cross-site") + crossSiteResult := httptest.NewRecorder() + app.Handler().ServeHTTP(crossSiteResult, frame) + if crossSiteResult.Code != http.StatusUnauthorized { + t.Fatalf("cross-site frame status = %d", crossSiteResult.Code) + } + + missingCookie := httptest.NewRequest(http.MethodGet, "/same-origin-frame", nil) + missingCookie.Header.Set("Sec-Fetch-Site", "same-origin") + missingCookie.Header.Set("Sec-Fetch-Mode", "navigate") + missingCookie.Header.Set("Sec-Fetch-Dest", "iframe") + missingCookieResult := httptest.NewRecorder() + app.Handler().ServeHTTP(missingCookieResult, missingCookie) + if missingCookieResult.Code != http.StatusUnauthorized { + t.Fatalf("missing-cookie frame status = %d", missingCookieResult.Code) + } } + diff --git a/Sense/server/app/sense/liveview/http.go b/Sense/server/app/sense/liveview/http.go index 1e0c444..1dca78e 100644 --- a/Sense/server/app/sense/liveview/http.go +++ b/Sense/server/app/sense/liveview/http.go @@ -14,7 +14,7 @@ func (m *Module) Register(app *platform.App) { app.Handle("GET /api/v1/liveview/routes", identity.Require(identity.PermissionMediaRead, http.HandlerFunc(m.routes))) app.Handle("POST /api/v1/liveview/sessions", identity.Require(identity.PermissionMediaRead, http.HandlerFunc(m.create))) app.Handle("GET /api/v1/liveview/sessions/{id}", identity.Require(identity.PermissionMediaRead, http.HandlerFunc(m.get))) - app.Handle("GET /api/v1/liveview/sessions/{id}/player", identity.Require(identity.PermissionMediaRead, http.HandlerFunc(m.player))) + app.Handle("GET /api/v1/liveview/sessions/{id}/player", identity.RequireSameOriginFrame(identity.PermissionMediaRead, http.HandlerFunc(m.player))) } func (m *Module) routes(w http.ResponseWriter, r *http.Request) { items, err := m.service.Routes(r.Context()) @@ -61,6 +61,11 @@ func (m *Module) player(w http.ResponseWriter, r *http.Request) { } w.Header().Set("Content-Type", "text/html; charset=utf-8") w.Header().Set("Cache-Control", "no-store") - w.Header().Set("Content-Security-Policy", "default-src 'none'; frame-src http: https:; style-src 'unsafe-inline'") + // This endpoint is the authenticated, same-origin wrapper loaded by the + // live-view page. Keep the global DENY policy everywhere else, and allow + // only Sense itself to embed this wrapper. + w.Header().Set("X-Frame-Options", "SAMEORIGIN") + w.Header().Set("Content-Security-Policy", "default-src 'none'; frame-ancestors 'self'; frame-src http: https:; style-src 'unsafe-inline'") _ = playerTemplate.Execute(w, target) } + diff --git a/Sense/server/app/sense/liveview/http_test.go b/Sense/server/app/sense/liveview/http_test.go new file mode 100644 index 0000000..228ef14 --- /dev/null +++ b/Sense/server/app/sense/liveview/http_test.go @@ -0,0 +1,40 @@ +package liveview + +import ( + "context" + "net/http" + "net/http/httptest" + "strings" + "testing" + "time" + + "yovision.local/sense/app/sense/media" +) + +func TestPlayerAllowsOnlySameOriginEmbedding(t *testing.T) { + service, err := NewService("http://127.0.0.1:8889", time.Minute) + if err != nil { + t.Fatal(err) + } + route := media.Route{ID: "device:main", Path: "sense_device_main", Desired: "running", Actual: "ready"} + service.route = func(context.Context, string) (media.Route, error) { return route, nil } + service.refresh = func(context.Context, string) (media.Route, error) { return route, nil } + service.sessions["view_test"] = Session{ID: "view_test", RouteID: route.ID, ExpiresAt: time.Now().Add(time.Minute)} + + req := httptest.NewRequest(http.MethodGet, "/api/v1/liveview/sessions/view_test/player", nil) + req.SetPathValue("id", "view_test") + res := httptest.NewRecorder() + NewModule(service).player(res, req) + + if res.Code != http.StatusOK { + t.Fatalf("status = %d", res.Code) + } + if got := res.Header().Get("X-Frame-Options"); got != "SAMEORIGIN" { + t.Fatalf("X-Frame-Options = %q", got) + } + csp := res.Header().Get("Content-Security-Policy") + if !strings.Contains(csp, "frame-ancestors 'self'") { + t.Fatalf("Content-Security-Policy = %q", csp) + } +} + diff --git a/Sense/server/app/sense/liveview/service.go b/Sense/server/app/sense/liveview/service.go index dc412a4..fd9f243 100644 --- a/Sense/server/app/sense/liveview/service.go +++ b/Sense/server/app/sense/liveview/service.go @@ -47,6 +47,7 @@ type Service struct { base *url.URL ttl time.Duration route func(context.Context, string) (media.Route, error) + refresh func(context.Context, string) (media.Route, error) routes func(context.Context) ([]media.Route, error) mu sync.RWMutex sessions map[string]Session @@ -64,7 +65,7 @@ func NewService(rawBase string, ttl time.Duration) (*Service, error) { if ttl <= 0 || ttl > 10*time.Minute { ttl = 2 * time.Minute } - return &Service{base: parsed, ttl: ttl, route: media.PlaybackRoute, routes: media.PlaybackRoutes, sessions: map[string]Session{}, now: time.Now}, nil + return &Service{base: parsed, ttl: ttl, route: media.PlaybackRoute, refresh: media.RefreshPlaybackRoute, routes: media.PlaybackRoutes, sessions: map[string]Session{}, now: time.Now}, nil } func (s *Service) Routes(ctx context.Context) ([]Route, error) { items, err := s.routes(ctx) @@ -117,7 +118,7 @@ func (s *Service) Get(ctx context.Context, owner, id string) (Session, error) { if !ok || session.OwnerID != owner || !s.now().Before(session.ExpiresAt) { return Session{}, fmt.Errorf("playback session expired") } - route, err := s.route(ctx, session.RouteID) + route, err := s.refresh(ctx, session.RouteID) if err != nil { return Session{}, err } diff --git a/Sense/server/app/sense/liveview/service_test.go b/Sense/server/app/sense/liveview/service_test.go index 7bd13f9..db95b81 100644 --- a/Sense/server/app/sense/liveview/service_test.go +++ b/Sense/server/app/sense/liveview/service_test.go @@ -44,3 +44,28 @@ func TestTTLIsBounded(t *testing.T) { t.Fatalf("ttl=%v", service.ttl) } } + +func TestSessionPollRefreshesOnDemandMediaState(t *testing.T) { + service, err := NewService("http://127.0.0.1:8889", time.Minute) + if err != nil { + t.Fatal(err) + } + route := media.Route{ID: "device:main", DeviceID: "device", ProfileToken: "main", Path: "sense_device_main", Desired: "running", Actual: "waiting"} + service.route = func(context.Context, string) (media.Route, error) { return route, nil } + service.refresh = func(context.Context, string) (media.Route, error) { + updated := route + updated.Actual = "ready" + updated.Detail = "上游拉流正常" + updated.Readers = 1 + return updated, nil + } + session, err := service.Create(context.Background(), "owner", route.ID) + if err != nil || session.Status != "waiting" || session.PlayerURL == "" { + t.Fatalf("session=%#v err=%v", session, err) + } + updated, err := service.Get(context.Background(), "owner", session.ID) + if err != nil || updated.Status != "ready" || updated.Detail != "上游拉流正常" { + t.Fatalf("updated=%#v err=%v", updated, err) + } +} + diff --git a/Sense/server/app/sense/media/route_port.go b/Sense/server/app/sense/media/route_port.go index b266013..d48eaf0 100644 --- a/Sense/server/app/sense/media/route_port.go +++ b/Sense/server/app/sense/media/route_port.go @@ -19,6 +19,13 @@ func PlaybackRoute(ctx context.Context, id string) (Route, error) { } return service.store.Get(ctx, id) } +func RefreshPlaybackRoute(ctx context.Context, id string) (Route, error) { + service := activeService.Load() + if service == nil { + return Route{}, fmt.Errorf("media service is not ready") + } + return service.Refresh(ctx, id) +} func PlaybackRoutes(ctx context.Context) ([]Route, error) { service := activeService.Load() if service == nil { diff --git a/Sense/server/app/sense/media/service.go b/Sense/server/app/sense/media/service.go index b2afb49..89bd57e 100644 --- a/Sense/server/app/sense/media/service.go +++ b/Sense/server/app/sense/media/service.go @@ -24,6 +24,32 @@ func NewService(store Store, process mediamtx.Process, controller mediamtx.Contr 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 @@ -108,6 +134,10 @@ func (s *Service) Reconcile(ctx context.Context, actor identity.Principal, id st 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 = "上游拉流正常" @@ -125,6 +155,43 @@ func (s *Service) Reconcile(ctx context.Context, actor identity.Principal, id st 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 { diff --git a/Sense/server/app/sense/media/service_test.go b/Sense/server/app/sense/media/service_test.go index dc55b3b..c4c6201 100644 --- a/Sense/server/app/sense/media/service_test.go +++ b/Sense/server/app/sense/media/service_test.go @@ -40,14 +40,22 @@ type fakeController struct { status mediamtx.PathStatus } +type refreshController struct{ status mediamtx.PathStatus } + +func (f refreshController) Apply(context.Context, mediamtx.Source) error { return nil } +func (f refreshController) Status(context.Context, string) (mediamtx.PathStatus, error) { + return f.status, nil +} + func (f fakeController) Apply(context.Context, mediamtx.Source) error { return f.applyErr } func (f fakeController) Status(context.Context, string) (mediamtx.PathStatus, error) { if f.status.Name == "error" { return mediamtx.PathStatus{}, errors.New("timeout") } if !f.status.Ready { - return mediamtx.PathStatus{Ready: true, Readers: 2}, nil + return mediamtx.PathStatus{Exists: true, Ready: true, Readers: 2}, nil } + f.status.Exists = true return f.status, nil } @@ -142,5 +150,47 @@ func TestConfigureProfilesIsIdempotentAndReportsMediaState(t *testing.T) { } } +func TestRefreshTracksOnDemandReaderWithoutReapplyingRoute(t *testing.T) { + store := NewMemoryStore() + route := Route{ID: "device:main", DeviceID: "device", ProfileToken: "main", Path: "sense_device_main", Desired: "running", Actual: "waiting", Detail: "等待上游拉流", Version: 2} + if err := store.Save(context.Background(), route); err != nil { + t.Fatal(err) + } + service := NewService(store, &fakeProcess{}, refreshController{status: mediamtx.PathStatus{Name: route.Path, Exists: true, Ready: true, Readers: 1}}) + result, err := service.Refresh(context.Background(), route.ID) + if err != nil || result.Actual != "ready" || result.Readers != 1 || result.Detail != "上游拉流正常" { + t.Fatalf("result=%#v err=%v", result, err) + } + unchanged, err := service.Refresh(context.Background(), route.ID) + if err != nil || unchanged.Version != result.Version { + t.Fatalf("unchanged=%#v err=%v", unchanged, err) + } +} + +func TestRefreshReportsMissingPathInsteadOfWaiting(t *testing.T) { + store := NewMemoryStore() + route := Route{ID: "device:main", Path: "sense_device_main", Desired: "running", Actual: "waiting", Version: 1} + if err := store.Save(context.Background(), route); err != nil { + t.Fatal(err) + } + service := NewService(store, &fakeProcess{}, refreshController{status: mediamtx.PathStatus{Name: route.Path, Exists: false}}) + result, err := service.Refresh(context.Background(), route.ID) + if err != nil || result.Actual != "apply_failed" || !strings.Contains(result.Detail, "不存在") { + t.Fatalf("result=%#v err=%v", result, err) + } +} + +func TestRestoreReconcilesDesiredRoutes(t *testing.T) { + process := &fakeProcess{} + service, route := preparedService(t, process, fakeController{}) + if err := service.Restore(context.Background()); err != nil { + t.Fatal(err) + } + restored, err := service.store.Get(context.Background(), route.ID) + if err != nil || !process.State().Running || restored.Actual != "ready" { + t.Fatalf("restored=%#v process=%#v err=%v", restored, process.State(), err) + } +} + func rtspReady() rtsp.Result { return rtsp.Result{Status: "ready"} } diff --git a/Sense/server/cmd/sense/modules_media.go b/Sense/server/cmd/sense/modules_media.go index da4e7fa..2bfdbc9 100644 --- a/Sense/server/cmd/sense/modules_media.go +++ b/Sense/server/cmd/sense/modules_media.go @@ -18,7 +18,9 @@ func init() { process := mediamtx.NewSupervisor(os.Getenv("SENSE_MEDIAMTX_BINARY"), os.Getenv("SENSE_MEDIAMTX_CONFIG"), 3) controller := mediamtx.NewHTTPController(valueOr("SENSE_MEDIAMTX_API", "http://127.0.0.1:9997")) app.RegisterMigration(platform.Migration{Version: 2026081203, Name: "sense_media", SQL: media.MigrationSQL}) - media.NewModule(media.NewService(store, process, controller)).Register(app) + service := media.NewService(store, process, controller) + media.NewModule(service).Register(app) + app.RegisterStartup(service.Restore) return nil }) } @@ -28,3 +30,4 @@ func valueOr(key, fallback string) string { } return fallback } + diff --git a/Sense/server/cmd/sense/root.go b/Sense/server/cmd/sense/root.go index 235d093..cd99c56 100644 --- a/Sense/server/cmd/sense/root.go +++ b/Sense/server/cmd/sense/root.go @@ -47,6 +47,9 @@ func Run() error { if err := app.ApplyMigrations(context.Background()); err != nil { return err } + if err := app.Start(context.Background()); err != nil { + return fmt.Errorf("start Sense modules: %w", err) + } server := &http.Server{ Addr: cfg.HTTPAddress, @@ -77,3 +80,4 @@ func Run() error { return serveErr } } + diff --git a/Sense/server/internal/platform/app.go b/Sense/server/internal/platform/app.go index 397c1ed..d31068a 100644 --- a/Sense/server/internal/platform/app.go +++ b/Sense/server/internal/platform/app.go @@ -1,6 +1,7 @@ package platform import ( + "context" "database/sql" "io/fs" "log/slog" @@ -17,6 +18,20 @@ type App struct { logger *slog.Logger mux *http.ServeMux migrations []Migration + startups []func(context.Context) error +} + +func (a *App) RegisterStartup(startup func(context.Context) error) { + a.startups = append(a.startups, startup) +} + +func (a *App) Start(ctx context.Context) error { + for _, startup := range a.startups { + if err := startup(ctx); err != nil { + return err + } + } + return nil } func NewApp(cfg Config, database *sql.DB, logger *slog.Logger) *App { @@ -101,3 +116,4 @@ func requestSecurityHeaders(next http.Handler) http.Handler { next.ServeHTTP(w, r) }) } + diff --git a/Sense/ui/src/components/sense/liveview/StreamPlayer.vue b/Sense/ui/src/components/sense/liveview/StreamPlayer.vue index 3dd5e38..594ba01 100644 --- a/Sense/ui/src/components/sense/liveview/StreamPlayer.vue +++ b/Sense/ui/src/components/sense/liveview/StreamPlayer.vue @@ -1,14 +1,15 @@