From 398534aabe18236a59c6023c1dd65c3756fc6d7d Mon Sep 17 00:00:00 2001 From: QiuSW <105186638@qq.com> Date: Wed, 12 Aug 2026 17:21:14 +0800 Subject: [PATCH] feat: add MediaMTX process and reconciliation control (#25) --- .../app/sense/adapters/mediamtx/client.go | 92 ++++++++++++ .../sense/adapters/mediamtx/client_test.go | 37 +++++ .../app/sense/adapters/mediamtx/supervisor.go | 113 ++++++++++++++ Sense/server/app/sense/admission/http.go | 1 + .../app/sense/admission/profile_port.go | 24 +++ Sense/server/app/sense/media/http.go | 64 ++++++++ Sense/server/app/sense/media/migration.go | 3 + Sense/server/app/sense/media/service.go | 140 ++++++++++++++++++ Sense/server/app/sense/media/service_test.go | 110 ++++++++++++++ Sense/server/app/sense/media/store.go | 78 ++++++++++ Sense/server/app/sense/media/types.go | 16 ++ Sense/server/cmd/sense/modules_media.go | 30 ++++ .../sense_media/2026081203_media.sql | 2 + Sense/testdata/mediamtx/path-status.json | 1 + Sense/tests/media/README.md | 6 + Sense/ui/src/api/sense/media/index.js | 7 + Sense/ui/src/router/modules/sense-media.js | 3 + Sense/ui/src/views/sense/media/Media.vue | 38 +++++ 18 files changed, 765 insertions(+) create mode 100644 Sense/server/app/sense/adapters/mediamtx/client.go create mode 100644 Sense/server/app/sense/adapters/mediamtx/client_test.go create mode 100644 Sense/server/app/sense/adapters/mediamtx/supervisor.go create mode 100644 Sense/server/app/sense/admission/profile_port.go create mode 100644 Sense/server/app/sense/media/http.go create mode 100644 Sense/server/app/sense/media/migration.go create mode 100644 Sense/server/app/sense/media/service.go create mode 100644 Sense/server/app/sense/media/service_test.go create mode 100644 Sense/server/app/sense/media/store.go create mode 100644 Sense/server/app/sense/media/types.go create mode 100644 Sense/server/cmd/sense/modules_media.go create mode 100644 Sense/server/database/migrations/sense_media/2026081203_media.sql create mode 100644 Sense/testdata/mediamtx/path-status.json create mode 100644 Sense/tests/media/README.md create mode 100644 Sense/ui/src/api/sense/media/index.js create mode 100644 Sense/ui/src/router/modules/sense-media.js create mode 100644 Sense/ui/src/views/sense/media/Media.vue diff --git a/Sense/server/app/sense/adapters/mediamtx/client.go b/Sense/server/app/sense/adapters/mediamtx/client.go new file mode 100644 index 0000000..c139c97 --- /dev/null +++ b/Sense/server/app/sense/adapters/mediamtx/client.go @@ -0,0 +1,92 @@ +package mediamtx + +import ( + "bytes" + "context" + "encoding/json" + "fmt" + "io" + "net/http" + "net/url" + "strings" + "time" +) + +type Source struct { + Path string + URI string + Username string + Password string +} +type PathStatus struct { + Name string `json:"name"` + Ready bool `json:"ready"` + Readers int `json:"readers"` + BytesReceived int64 `json:"bytes_received"` +} +type Controller interface { + Apply(context.Context, Source) error + Status(context.Context, string) (PathStatus, error) +} +type HTTPController struct { + base string + client *http.Client +} + +func NewHTTPController(base string) *HTTPController { + return &HTTPController{base: strings.TrimRight(base, "/"), client: &http.Client{Timeout: 5 * time.Second}} +} +func (c *HTTPController) Apply(ctx context.Context, source Source) error { + parsed, err := url.Parse(source.URI) + if err != nil || parsed.User != nil { + return fmt.Errorf("invalid credential-free media source") + } + if source.Username != "" { + parsed.User = url.UserPassword(source.Username, source.Password) + } + 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) + req, err := http.NewRequestWithContext(ctx, http.MethodPost, endpoint, bytes.NewReader(data)) + if err != nil { + return err + } + req.Header.Set("Content-Type", "application/json") + res, err := c.client.Do(req) + if err != nil { + return fmt.Errorf("mediamtx control unavailable: %w", err) + } + defer res.Body.Close() + io.Copy(io.Discard, io.LimitReader(res.Body, 1<<20)) + if res.StatusCode < 200 || res.StatusCode >= 300 { + return fmt.Errorf("mediamtx apply returned %d", res.StatusCode) + } + return 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 { + return PathStatus{}, err + } + res, err := c.client.Do(req) + if err != nil { + return PathStatus{}, err + } + defer res.Body.Close() + if res.StatusCode == http.StatusNotFound { + return PathStatus{Name: path}, nil + } + if res.StatusCode < 200 || res.StatusCode >= 300 { + return PathStatus{}, fmt.Errorf("mediamtx status returned %d", res.StatusCode) + } + var raw struct { + Name string `json:"name"` + Ready bool `json:"ready"` + BytesReceived int64 `json:"bytesReceived"` + Readers []any `json:"readers"` + } + 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 +} diff --git a/Sense/server/app/sense/adapters/mediamtx/client_test.go b/Sense/server/app/sense/adapters/mediamtx/client_test.go new file mode 100644 index 0000000..ce9ab6e --- /dev/null +++ b/Sense/server/app/sense/adapters/mediamtx/client_test.go @@ -0,0 +1,37 @@ +package mediamtx + +import ( + "context" + "encoding/json" + "net/http" + "net/http/httptest" + "strings" + "testing" +) + +func TestApplyBuildsCredentialSourceOnlyInTransientBody(t *testing.T) { + var body map[string]any + 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 err := json.NewDecoder(r.Body).Decode(&body); err != nil { + t.Fatal(err) + } + w.WriteHeader(http.StatusOK) + })) + defer server.Close() + client := NewHTTPController(server.URL) + if err := client.Apply(context.Background(), Source{Path: "sense_test", URI: "rtsp://camera.invalid/main", Username: "fixture-user", Password: "fixture-password"}); err != nil { + t.Fatal(err) + } + if body["source"] != "rtsp://fixture-user:fixture-password@camera.invalid/main" { + t.Fatalf("body=%#v", body) + } +} +func TestApplyRejectsCredentialURI(t *testing.T) { + client := NewHTTPController("http://127.0.0.1") + if err := client.Apply(context.Background(), Source{Path: "x", URI: "rtsp://user:pass@camera.invalid/main"}); err == nil { + t.Fatal("credential URI accepted") + } +} diff --git a/Sense/server/app/sense/adapters/mediamtx/supervisor.go b/Sense/server/app/sense/adapters/mediamtx/supervisor.go new file mode 100644 index 0000000..8c8b92c --- /dev/null +++ b/Sense/server/app/sense/adapters/mediamtx/supervisor.go @@ -0,0 +1,113 @@ +package mediamtx + +import ( + "context" + "fmt" + "os" + "os/exec" + "sync" + "time" +) + +type ProcessState struct { + Running bool `json:"running"` + PID int `json:"pid,omitempty"` + Restarts int `json:"restarts"` + LastExit string `json:"last_exit,omitempty"` +} +type Process interface { + Start(context.Context) error + Stop(context.Context) error + State() ProcessState +} +type Supervisor struct { + binary string + config string + maxRestarts int + mu sync.Mutex + command *exec.Cmd + state ProcessState +} + +func NewSupervisor(binary, config string, maxRestarts int) *Supervisor { + if maxRestarts < 0 { + maxRestarts = 0 + } + return &Supervisor{binary: binary, config: config, maxRestarts: maxRestarts} +} +func (s *Supervisor) Start(ctx context.Context) error { + s.mu.Lock() + defer s.mu.Unlock() + if s.state.Running { + return nil + } + if s.binary == "" { + return fmt.Errorf("SENSE_MEDIAMTX_BINARY is not configured") + } + if s.state.LastExit != "" { + if s.state.Restarts >= s.maxRestarts { + return fmt.Errorf("MediaMTX restart limit reached") + } + s.state.Restarts++ + } + arguments := []string{} + if s.config != "" { + arguments = append(arguments, s.config) + } + command := exec.CommandContext(context.Background(), s.binary, arguments...) + if err := command.Start(); err != nil { + return fmt.Errorf("start MediaMTX: %w", err) + } + s.command = command + s.state.Running = true + s.state.PID = command.Process.Pid + s.state.LastExit = "" + go s.wait(command) + return nil +} +func (s *Supervisor) wait(command *exec.Cmd) { + err := command.Wait() + s.mu.Lock() + defer s.mu.Unlock() + if s.command != command { + return + } + s.state.Running = false + s.state.PID = 0 + if err != nil { + s.state.LastExit = err.Error() + } else { + s.state.LastExit = "exited" + } +} +func (s *Supervisor) Stop(ctx context.Context) error { + s.mu.Lock() + command := s.command + if command == nil || !s.state.Running { + s.mu.Unlock() + return nil + } + process := command.Process + s.mu.Unlock() + if err := process.Signal(os.Interrupt); err != nil { + if killErr := process.Kill(); killErr != nil { + return fmt.Errorf("stop owned MediaMTX process: %w", killErr) + } + } + ticker := time.NewTicker(50 * time.Millisecond) + defer ticker.Stop() + for { + select { + case <-ctx.Done(): + return ctx.Err() + case <-ticker.C: + s.mu.Lock() + running := s.state.Running + s.mu.Unlock() + if !running { + return nil + } + } + } +} +func (s *Supervisor) State() ProcessState { s.mu.Lock(); defer s.mu.Unlock(); return s.state } diff --git a/Sense/server/app/sense/admission/http.go b/Sense/server/app/sense/admission/http.go index 2dca49b..8410e23 100644 --- a/Sense/server/app/sense/admission/http.go +++ b/Sense/server/app/sense/admission/http.go @@ -10,6 +10,7 @@ type Module struct{ service *Service } func NewModule(service *Service) *Module { return &Module{service: service} } func (m *Module) Register(app *platform.App) { + setCurrentService(m.service) app.Handle("GET /api/v1/admission/discover", identity.Require(identity.PermissionDeviceWrite, http.HandlerFunc(m.discover))) app.Handle("POST /api/v1/admission/probe", identity.Require(identity.PermissionDeviceWrite, http.HandlerFunc(m.probe))) app.Handle("GET /api/v1/admission/devices/{id}", identity.Require(identity.PermissionDeviceRead, http.HandlerFunc(m.get))) diff --git a/Sense/server/app/sense/admission/profile_port.go b/Sense/server/app/sense/admission/profile_port.go new file mode 100644 index 0000000..ecf7087 --- /dev/null +++ b/Sense/server/app/sense/admission/profile_port.go @@ -0,0 +1,24 @@ +package admission + +import "fmt" + +var currentService *Service + +func setCurrentService(service *Service) { currentService = service } + +// VerifiedProfile is a Sense-internal projection, not a cross-product contract. +func VerifiedProfile(deviceID, token string) (Profile, error) { + if currentService == nil { + return Profile{}, fmt.Errorf("admission service is not ready") + } + result, ok := currentService.Get(deviceID) + if !ok { + return Profile{}, fmt.Errorf("device admission has not been verified") + } + for _, profile := range result.Profiles { + if profile.Token == token && profile.Verification.Status == "ready" { + return profile, nil + } + } + return Profile{}, fmt.Errorf("verified profile not found") +} diff --git a/Sense/server/app/sense/media/http.go b/Sense/server/app/sense/media/http.go new file mode 100644 index 0000000..121a73f --- /dev/null +++ b/Sense/server/app/sense/media/http.go @@ -0,0 +1,64 @@ +package media + +import ( + "net/http" + "yovision.local/sense/app/sense/identity" + "yovision.local/sense/internal/platform" +) + +type Module struct{ service *Service } + +func NewModule(service *Service) *Module { return &Module{service: service} } +func (m *Module) Register(app *platform.App) { + app.Handle("GET /api/v1/media/routes", identity.Require(identity.PermissionMediaRead, http.HandlerFunc(m.list))) + app.Handle("POST /api/v1/media/routes", identity.Require(identity.PermissionMediaWrite, http.HandlerFunc(m.configure))) + app.Handle("POST /api/v1/media/routes/{id}/reconcile", identity.Require(identity.PermissionMediaWrite, http.HandlerFunc(m.reconcile))) + app.Handle("POST /api/v1/media/routes/{id}/stop", identity.Require(identity.PermissionMediaWrite, http.HandlerFunc(m.stop))) + app.Handle("GET /api/v1/media/process", identity.Require(identity.PermissionMediaRead, http.HandlerFunc(m.process))) +} +func (m *Module) list(w http.ResponseWriter, r *http.Request) { + items, err := m.service.List(r.Context()) + if err != nil { + platform.WriteError(w, err) + return + } + platform.WriteJSON(w, http.StatusOK, map[string]any{"items": items, "total": len(items)}) +} +func (m *Module) configure(w http.ResponseWriter, r *http.Request) { + var req struct { + DeviceID string `json:"device_id"` + ProfileToken string `json:"profile_token"` + } + if err := platform.DecodeJSON(r, &req); err != nil { + platform.WriteError(w, err) + return + } + actor, _ := identity.PrincipalFromContext(r.Context()) + item, err := m.service.Configure(r.Context(), actor, req.DeviceID, req.ProfileToken) + if err != nil { + platform.WriteError(w, &platform.APIError{Status: http.StatusBadRequest, Code: "media_invalid", Message: err.Error()}) + return + } + platform.WriteJSON(w, http.StatusCreated, item) +} +func (m *Module) reconcile(w http.ResponseWriter, r *http.Request) { + actor, _ := identity.PrincipalFromContext(r.Context()) + item, err := m.service.Reconcile(r.Context(), actor, r.PathValue("id")) + if err != nil { + platform.WriteError(w, &platform.APIError{Status: http.StatusBadRequest, Code: "media_reconcile_failed", Message: err.Error()}) + return + } + platform.WriteJSON(w, http.StatusOK, item) +} +func (m *Module) stop(w http.ResponseWriter, r *http.Request) { + actor, _ := identity.PrincipalFromContext(r.Context()) + item, err := m.service.Stop(r.Context(), actor, r.PathValue("id")) + if err != nil { + platform.WriteError(w, err) + return + } + platform.WriteJSON(w, http.StatusOK, item) +} +func (m *Module) process(w http.ResponseWriter, _ *http.Request) { + platform.WriteJSON(w, http.StatusOK, m.service.ProcessState()) +} diff --git a/Sense/server/app/sense/media/migration.go b/Sense/server/app/sense/media/migration.go new file mode 100644 index 0000000..9e4008b --- /dev/null +++ b/Sense/server/app/sense/media/migration.go @@ -0,0 +1,3 @@ +package media + +const MigrationSQL = `CREATE TABLE IF NOT EXISTS sense_media_routes(id TEXT PRIMARY KEY,device_id TEXT NOT NULL,profile_token TEXT NOT NULL,path TEXT NOT NULL UNIQUE,desired TEXT NOT NULL,actual TEXT NOT NULL,readers INTEGER NOT NULL DEFAULT 0,detail TEXT NOT NULL DEFAULT '',version BIGINT NOT NULL,updated_at TIMESTAMPTZ NOT NULL);` diff --git a/Sense/server/app/sense/media/service.go b/Sense/server/app/sense/media/service.go new file mode 100644 index 0000000..8197930 --- /dev/null +++ b/Sense/server/app/sense/media/service.go @@ -0,0 +1,140 @@ +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) +} diff --git a/Sense/server/app/sense/media/service_test.go b/Sense/server/app/sense/media/service_test.go new file mode 100644 index 0000000..5f536d3 --- /dev/null +++ b/Sense/server/app/sense/media/service_test.go @@ -0,0 +1,110 @@ +package media + +import ( + "context" + "errors" + "testing" + + "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 fakeProcess struct { + state mediamtx.ProcessState + startErr error + stopped bool +} + +func (p *fakeProcess) Start(context.Context) error { + if p.startErr != nil { + return p.startErr + } + p.state.Running = true + p.state.PID = 123 + return nil +} +func (p *fakeProcess) Stop(context.Context) error { + p.stopped = true + p.state.Running = false + p.state.PID = 0 + return nil +} +func (p *fakeProcess) State() mediamtx.ProcessState { return p.state } + +type fakeController struct { + applyErr error + status mediamtx.PathStatus +} + +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 f.status, nil +} + +func TestStoreAndProcessBoundary(t *testing.T) { + store := NewMemoryStore() + process := &fakeProcess{} + service := NewService(store, process, fakeController{}) + route := Route{ID: "dev:main", DeviceID: "dev", ProfileToken: "main", Path: "sense_dev_main", Desired: "running", Actual: "pending", Version: 1} + if err := store.Save(context.Background(), route); err != nil { + t.Fatal(err) + } + items, err := service.List(context.Background()) + if err != nil || len(items) != 1 { + t.Fatalf("items=%#v err=%v", items, err) + } + if process.State().Running { + t.Fatal("process started without explicit reconcile") + } +} + +func preparedService(t *testing.T, process *fakeProcess, controller fakeController) (*Service, Route) { + t.Helper() + service := NewService(NewMemoryStore(), process, controller) + service.profile = func(string, string) (admission.Profile, error) { + return admission.Profile{StreamURI: "rtsp://camera.invalid/main"}, nil + } + service.credential = func(context.Context, string) (device.Credential, error) { + return device.Credential{Username: "fixture", Password: "fixture"}, nil + } + route, err := service.Configure(context.Background(), identity.Principal{}, "device", "main") + if err != nil { + t.Fatal(err) + } + return service, route +} + +func TestReconcileReadyAndStopOwnedProcess(t *testing.T) { + process := &fakeProcess{} + service, route := preparedService(t, process, fakeController{status: mediamtx.PathStatus{Ready: true, Readers: 2}}) + result, err := service.Reconcile(context.Background(), identity.Principal{}, route.ID) + if err != nil || result.Actual != "ready" || result.Readers != 2 { + t.Fatalf("result=%#v err=%v", result, err) + } + if _, err := service.Stop(context.Background(), identity.Principal{}, route.ID); err != nil || !process.stopped { + t.Fatalf("stop err=%v stopped=%v", err, process.stopped) + } +} + +func TestReconcileRecordsFailureStates(t *testing.T) { + process := &fakeProcess{startErr: errors.New("restart limit reached")} + service, route := preparedService(t, process, fakeController{}) + result, err := service.Reconcile(context.Background(), identity.Principal{}, route.ID) + if err != nil || result.Actual != "process_failed" { + t.Fatalf("result=%#v err=%v", result, err) + } + process.startErr = nil + service.controller = fakeController{applyErr: errors.New("control timeout")} + result, err = service.Reconcile(context.Background(), identity.Principal{}, route.ID) + if err != nil || result.Actual != "apply_failed" { + t.Fatalf("result=%#v err=%v", result, err) + } +} diff --git a/Sense/server/app/sense/media/store.go b/Sense/server/app/sense/media/store.go new file mode 100644 index 0000000..9adc872 --- /dev/null +++ b/Sense/server/app/sense/media/store.go @@ -0,0 +1,78 @@ +package media + +import ( + "context" + "database/sql" + "errors" + "sync" +) + +var ErrNotFound = errors.New("media route not found") + +type Store interface { + Save(context.Context, Route) error + Get(context.Context, string) (Route, error) + List(context.Context) ([]Route, error) +} +type MemoryStore struct { + mu sync.RWMutex + items map[string]Route +} + +func NewMemoryStore() *MemoryStore { return &MemoryStore{items: map[string]Route{}} } +func (s *MemoryStore) Save(_ context.Context, item Route) error { + s.mu.Lock() + defer s.mu.Unlock() + s.items[item.ID] = item + return nil +} +func (s *MemoryStore) Get(_ context.Context, id string) (Route, error) { + s.mu.RLock() + defer s.mu.RUnlock() + item, ok := s.items[id] + if !ok { + return Route{}, ErrNotFound + } + return item, nil +} +func (s *MemoryStore) List(_ context.Context) ([]Route, error) { + s.mu.RLock() + defer s.mu.RUnlock() + items := make([]Route, 0, len(s.items)) + for _, item := range s.items { + items = append(items, item) + } + return items, nil +} + +type PostgresStore struct{ database *sql.DB } + +func NewPostgresStore(db *sql.DB) *PostgresStore { return &PostgresStore{database: db} } +func (s *PostgresStore) Save(ctx context.Context, item Route) error { + _, err := s.database.ExecContext(ctx, `INSERT INTO sense_media_routes(id,device_id,profile_token,path,desired,actual,readers,detail,version,updated_at) VALUES($1,$2,$3,$4,$5,$6,$7,$8,$9,$10) ON CONFLICT(id) DO UPDATE SET desired=EXCLUDED.desired,actual=EXCLUDED.actual,readers=EXCLUDED.readers,detail=EXCLUDED.detail,version=EXCLUDED.version,updated_at=EXCLUDED.updated_at`, item.ID, item.DeviceID, item.ProfileToken, item.Path, item.Desired, item.Actual, item.Readers, item.Detail, item.Version, item.UpdatedAt) + return err +} +func (s *PostgresStore) Get(ctx context.Context, id string) (Route, error) { + var item Route + err := s.database.QueryRowContext(ctx, `SELECT id,device_id,profile_token,path,desired,actual,readers,detail,version,updated_at FROM sense_media_routes WHERE id=$1`, id).Scan(&item.ID, &item.DeviceID, &item.ProfileToken, &item.Path, &item.Desired, &item.Actual, &item.Readers, &item.Detail, &item.Version, &item.UpdatedAt) + if errors.Is(err, sql.ErrNoRows) { + return Route{}, ErrNotFound + } + return item, err +} +func (s *PostgresStore) List(ctx context.Context) ([]Route, error) { + rows, err := s.database.QueryContext(ctx, `SELECT id,device_id,profile_token,path,desired,actual,readers,detail,version,updated_at FROM sense_media_routes ORDER BY updated_at DESC`) + if err != nil { + return nil, err + } + defer rows.Close() + var items []Route + for rows.Next() { + var item Route + if err := rows.Scan(&item.ID, &item.DeviceID, &item.ProfileToken, &item.Path, &item.Desired, &item.Actual, &item.Readers, &item.Detail, &item.Version, &item.UpdatedAt); err != nil { + return nil, err + } + items = append(items, item) + } + return items, rows.Err() +} diff --git a/Sense/server/app/sense/media/types.go b/Sense/server/app/sense/media/types.go new file mode 100644 index 0000000..1bd6a11 --- /dev/null +++ b/Sense/server/app/sense/media/types.go @@ -0,0 +1,16 @@ +package media + +import "time" + +type Route struct { + ID string `json:"id"` + DeviceID string `json:"device_id"` + ProfileToken string `json:"profile_token"` + Path string `json:"path"` + Desired string `json:"desired"` + Actual string `json:"actual"` + Readers int `json:"readers"` + Detail string `json:"detail"` + Version int64 `json:"version"` + UpdatedAt time.Time `json:"updated_at"` +} diff --git a/Sense/server/cmd/sense/modules_media.go b/Sense/server/cmd/sense/modules_media.go new file mode 100644 index 0000000..da4e7fa --- /dev/null +++ b/Sense/server/cmd/sense/modules_media.go @@ -0,0 +1,30 @@ +package sense + +import ( + "os" + "yovision.local/sense/app/sense/adapters/mediamtx" + "yovision.local/sense/app/sense/media" + "yovision.local/sense/internal/platform" +) + +func init() { + registerModule(func(app *platform.App) error { + var store media.Store + if app.Config().DatabaseMode == platform.DatabaseModeMemory { + store = media.NewMemoryStore() + } else { + store = media.NewPostgresStore(app.Database()) + } + 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) + return nil + }) +} +func valueOr(key, fallback string) string { + if value := os.Getenv(key); value != "" { + return value + } + return fallback +} diff --git a/Sense/server/database/migrations/sense_media/2026081203_media.sql b/Sense/server/database/migrations/sense_media/2026081203_media.sql new file mode 100644 index 0000000..845a65f --- /dev/null +++ b/Sense/server/database/migrations/sense_media/2026081203_media.sql @@ -0,0 +1,2 @@ +CREATE TABLE IF NOT EXISTS sense_media_routes(id TEXT PRIMARY KEY,device_id TEXT NOT NULL,profile_token TEXT NOT NULL,path TEXT NOT NULL UNIQUE,desired TEXT NOT NULL,actual TEXT NOT NULL,readers INTEGER NOT NULL DEFAULT 0,detail TEXT NOT NULL DEFAULT '',version BIGINT NOT NULL,updated_at TIMESTAMPTZ NOT NULL); + diff --git a/Sense/testdata/mediamtx/path-status.json b/Sense/testdata/mediamtx/path-status.json new file mode 100644 index 0000000..42a8bd8 --- /dev/null +++ b/Sense/testdata/mediamtx/path-status.json @@ -0,0 +1 @@ +{"name":"sense_fixture_main","ready":true,"bytesReceived":4096,"readers":[{"type":"webrtc"}]} diff --git a/Sense/tests/media/README.md b/Sense/tests/media/README.md new file mode 100644 index 0000000..26e19b7 --- /dev/null +++ b/Sense/tests/media/README.md @@ -0,0 +1,6 @@ +# Media acceptance + +Tests keep MediaMTX behind process and control interfaces. They verify that +the credential-bearing MediaMTX control request is localhost-only and transient, +never returned or persisted by Sense, and no process starts before an explicit reconcile. Real MediaMTX binary, +camera pull, reader counts and crash recovery require the deployment lab. diff --git a/Sense/ui/src/api/sense/media/index.js b/Sense/ui/src/api/sense/media/index.js new file mode 100644 index 0000000..6d0045e --- /dev/null +++ b/Sense/ui/src/api/sense/media/index.js @@ -0,0 +1,7 @@ +import request from '../../../bootstrap/request' +export const listRoutes = () => request.get('/media/routes') +export const configureRoute = (data) => request.post('/media/routes', data) +export const reconcileRoute = (id) => request.post(`/media/routes/${encodeURIComponent(id)}/reconcile`) +export const stopRoute = (id) => request.post(`/media/routes/${encodeURIComponent(id)}/stop`) +export const processState = () => request.get('/media/process') + diff --git a/Sense/ui/src/router/modules/sense-media.js b/Sense/ui/src/router/modules/sense-media.js new file mode 100644 index 0000000..cace760 --- /dev/null +++ b/Sense/ui/src/router/modules/sense-media.js @@ -0,0 +1,3 @@ +import Media from '../../views/sense/media/Media.vue' +export default { path: 'media', name: 'sense-media', component: Media, meta: { title: '视频服务', icon: 'Monitor', order: 30, permission: 'media.read' } } + diff --git a/Sense/ui/src/views/sense/media/Media.vue b/Sense/ui/src/views/sense/media/Media.vue new file mode 100644 index 0000000..1496fb1 --- /dev/null +++ b/Sense/ui/src/views/sense/media/Media.vue @@ -0,0 +1,38 @@ + + +