feat: 实现 MediaMTX 分片分配与故障范围可观察 (#77)

Sense/server/app/admin/router/sense_media_shard.go

@@ -0,0 +1,20 @@
+package router
+
+import (
+	"github.com/gin-gonic/gin"
+	jwt "github.com/go-admin-team/go-admin-core/sdk/pkg/jwtauth"
+
+	"git.ilapage.cn/ila/yovision/Sense/server/app/sense/media_shard"
+	"git.ilapage.cn/ila/yovision/Sense/server/common/actions"
+	"git.ilapage.cn/ila/yovision/Sense/server/common/middleware"
+)
+
+func init() { routerCheckRole = append(routerCheckRole, registerSenseMediaShardRouter) }
+
+func registerSenseMediaShardRouter(v1 *gin.RouterGroup, auth *jwt.GinJWTMiddleware) {
+	api := &media_shard.API{}
+	r := v1.Group("/media-shards").Use(auth.MiddlewareFunc()).Use(middleware.AuthCheckRole()).Use(actions.PermissionAction())
+	r.GET("", api.List)
+	r.GET("/:id", api.Get)
+	r.GET("/:id/migration-preflight", api.Preflight)
+}

Sense/server/app/sense/media/runtime.go

@@ -9,6 +9,8 @@ import (
 	"time"
 
 	"gorm.io/gorm"
+
+	"git.ilapage.cn/ila/yovision/Sense/server/app/sense/media_shard"
 )
 
 var runtimeState struct {
@@ -41,6 +43,27 @@ func StartRuntime(parent context.Context, db *gorm.DB) error {
 			return err
 		}
 	}
+	shards := media_shard.NewService(db, func(probeCtx context.Context, endpoint string) error {
+		probe, probeErr := NewHTTPController(endpoint)
+		if probeErr != nil {
+			return probeErr
+		}
+		return probe.Health(probeCtx)
+	})
+	specs, err := media_shard.SpecsFromEnvironment(db, mode, config.APIBase)
+	if err != nil {
+		cancel()
+		return err
+	}
+	if err = shards.SyncSpecs(ctx, specs); err != nil {
+		cancel()
+		return err
+	}
+	if err = shards.RefreshAll(ctx); err != nil {
+		cancel()
+		return err
+	}
+	service.WithShardService(shards)
 	runtimeState.Lock()
 	if runtimeState.cancel != nil {
 		runtimeState.cancel()

Sense/server/app/sense/media/service.go

@@ -13,6 +13,7 @@ import (
 	"gorm.io/gorm/clause"
 
 	"git.ilapage.cn/ila/yovision/Sense/server/app/sense/credential"
+	"git.ilapage.cn/ila/yovision/Sense/server/app/sense/media_shard"
 	"git.ilapage.cn/ila/yovision/Sense/server/app/sense/reconcile"
 )
 
@@ -26,10 +27,16 @@ type Service struct {
 	controller Controller
 	process    *Supervisor
 	config     RuntimeConfig
+	shards     *media_shard.Service
 	now        func() time.Time
 	mu         sync.Mutex
 }
 
+func (s *Service) WithShardService(shards *media_shard.Service) *Service {
+	s.shards = shards
+	return s
+}
+
 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}
 }
@@ -76,11 +83,19 @@ func (s *Service) ensureDevice(ctx context.Context, deviceID string, reactivate
 				return err
 			}
 		}
+		if s.shards != nil {
+			if _, err := s.shards.EnsureAssignment(ctx, id); err != nil {
+				return fmt.Errorf("assign media route to shard: %w", err)
+			}
+		}
 	}
 	return nil
 }
 
 func (s *Service) Run(ctx context.Context) {
+	if s.shards != nil {
+		_ = s.shards.RefreshAll(ctx)
+	}
 	_ = s.EnsureAllVerified(ctx)
 	_ = s.ReconcileDue(ctx)
 	interval := s.config.PollInterval
@@ -94,6 +109,9 @@ func (s *Service) Run(ctx context.Context) {
 		case <-ctx.Done():
 			return
 		case <-ticker.C:
+			if s.shards != nil {
+				_ = s.shards.RefreshAll(ctx)
+			}
 			_ = s.ReconcileDue(ctx)
 		}
 	}
@@ -127,10 +145,11 @@ func (s *Service) Refresh(ctx context.Context, id string) (RouteResponse, error)
 	if err := s.db.WithContext(ctx).First(&route, "id = ?", id).Error; err != nil {
 		return RouteResponse{}, ErrNotFound
 	}
-	if err := s.ensureControl(ctx); err != nil {
+	controller, err := s.controllerForRoute(ctx, route)
+	if err != nil {
 		return s.saveFailure(ctx, route, "process_unavailable", "MediaMTX 未启动或 Control API 未就绪", err)
 	}
-	status, err := s.controller.Status(ctx, route.Path)
+	status, err := controller.Status(ctx, route.Path)
 	if err != nil {
 		return s.saveFailure(ctx, route, "status_unavailable", "尚未取得媒体路径状态", err)
 	}
@@ -159,12 +178,13 @@ func (s *Service) Reconcile(ctx context.Context, id string) (RouteResponse, erro
 		return RouteResponse{}, err
 	}
 	if route.Desired == DesiredStopped {
-		if s.controller != nil {
-			_ = s.controller.Delete(ctx, route.Path)
+		if controller, err := s.controllerForRoute(ctx, route); err == nil {
+			_ = controller.Delete(ctx, route.Path)
 		}
 		return s.saveSuccess(ctx, route, "stopped", false, 0, "媒体路径已停止")
 	}
-	if err := s.ensureControl(ctx); err != nil {
+	controller, err := s.controllerForRoute(ctx, route)
+	if err != nil {
 		return s.saveFailure(ctx, route, "process_unavailable", "MediaMTX 未启动或 Control API 未就绪", err)
 	}
 	var profile admissionProfile
@@ -175,10 +195,10 @@ func (s *Service) Reconcile(ctx context.Context, id string) (RouteResponse, erro
 	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 {
+	if err = 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)
+	status, err := controller.Status(ctx, route.Path)
 	if err != nil {
 		return s.saveFailure(ctx, route, "status_unavailable", "尚未取得媒体路径状态", err)
 	}
@@ -191,6 +211,43 @@ func (s *Service) Reconcile(ctx context.Context, id string) (RouteResponse, erro
 	return s.saveSuccess(ctx, route, "waiting", false, status.Readers, "等待播放器连接并按需拉流")
 }
 
+func (s *Service) controllerForRoute(ctx context.Context, route Route) (Controller, error) {
+	if s.shards == nil {
+		if err := s.ensureControl(ctx); err != nil {
+			return nil, err
+		}
+		return s.controller, nil
+	}
+	shard, err := s.shards.ResolveRoute(ctx, route.ID)
+	if errors.Is(err, media_shard.ErrNotFound) {
+		shard, err = s.shards.EnsureAssignment(ctx, route.ID)
+	}
+	if err != nil {
+		return nil, err
+	}
+	if shard.Status != media_shard.StatusRunning {
+		return nil, ErrRuntimeUnavailable
+	}
+	if shard.ID == "primary" {
+		if err = s.ensureControl(ctx); err != nil {
+			return nil, err
+		}
+		return s.controller, nil
+	}
+	endpoint, err := s.shards.ControlAPI(ctx, shard.ID)
+	if err != nil {
+		return nil, err
+	}
+	controller, err := NewHTTPController(endpoint)
+	if err != nil {
+		return nil, err
+	}
+	if err = controller.Health(ctx); err != nil {
+		return nil, err
+	}
+	return controller, nil
+}
+
 func (s *Service) ensureControl(ctx context.Context) error {
 	if s.controller == nil {
 		return ErrRuntimeUnavailable

Sense/server/app/sense/media_shard/apis.go

@@ -0,0 +1,91 @@
+package media_shard
+
+import (
+	"errors"
+	"net/http"
+	"time"
+
+	"github.com/gin-gonic/gin"
+	"github.com/go-admin-team/go-admin-core/sdk/api"
+	"github.com/go-admin-team/go-admin-core/sdk/pkg/jwtauth/user"
+	coreService "github.com/go-admin-team/go-admin-core/sdk/service"
+
+	"git.ilapage.cn/ila/yovision/Sense/server/common"
+)
+
+const auditSuccess, auditFailure = "1", "2"
+
+type API struct{ api.Api }
+
+func (e *API) service(c *gin.Context) (*Service, error) {
+	base := coreService.Service{}
+	if err := e.MakeContext(c).MakeOrm().MakeService(&base).Errors; err != nil {
+		return nil, err
+	}
+	return NewService(base.Orm, nil), nil
+}
+
+func (e *API) List(c *gin.Context) {
+	service, err := e.service(c)
+	if err != nil {
+		e.writeError(err)
+		return
+	}
+	result, err := service.List(c.Request.Context())
+	if err != nil {
+		e.audit(c, service, "List", auditFailure, "媒体分片列表查询失败")
+		e.writeError(err)
+		return
+	}
+	e.audit(c, service, "List", auditSuccess, "读取媒体分片列表")
+	e.OK(result, "查询成功")
+}
+
+func (e *API) Get(c *gin.Context) {
+	service, err := e.service(c)
+	if err != nil {
+		e.writeError(err)
+		return
+	}
+	result, err := service.Get(c.Request.Context(), c.Param("id"))
+	if err != nil {
+		e.audit(c, service, "Get", auditFailure, "媒体分片详情查询失败")
+		e.writeError(err)
+		return
+	}
+	e.audit(c, service, "Get", auditSuccess, "读取媒体分片详情 "+result.ID)
+	e.OK(result, "查询成功")
+}
+
+func (e *API) Preflight(c *gin.Context) {
+	service, err := e.service(c)
+	if err != nil {
+		e.writeError(err)
+		return
+	}
+	result, err := service.Preflight(c.Request.Context(), c.Param("id"))
+	if err != nil {
+		e.audit(c, service, "Preflight", auditFailure, "媒体分片迁移预检失败")
+		e.writeError(err)
+		return
+	}
+	e.audit(c, service, "Preflight", auditSuccess, "执行媒体分片只读迁移预检 "+result.SourceShardID)
+	e.OK(result, "预检完成")
+}
+
+func (e *API) audit(c *gin.Context, service *Service, action, status, remark string) {
+	if err := WriteAudit(service.DB, Audit{Action: action, Method: c.Request.Method, Status: status, Username: user.GetUserName(c), UserID: user.GetUserId(c), ClientIP: common.GetClientIP(c), Route: c.FullPath(), Remark: remark, At: time.Now()}); err != nil {
+		api.GetRequestLogger(c).Errorf("media shard audit failed: %s", err.Error())
+	}
+}
+
+func (e *API) writeError(err error) {
+	switch {
+	case errors.Is(err, ErrInvalidID), errors.Is(err, ErrInvalidConfig):
+		e.Error(http.StatusBadRequest, err, err.Error())
+	case errors.Is(err, ErrNotFound):
+		e.Error(http.StatusNotFound, err, err.Error())
+	default:
+		e.Error(http.StatusInternalServerError, err, "媒体分片查询失败")
+	}
+}

Sense/server/app/sense/media_shard/audit.go

@@ -0,0 +1,19 @@
+package media_shard
+
+import (
+	"time"
+
+	adminModels "git.ilapage.cn/ila/yovision/Sense/server/app/admin/models"
+	"gorm.io/gorm"
+)
+
+type Audit struct {
+	Action, Method, Status, Username, ClientIP, Route, Remark string
+	UserID                                                    int
+	At                                                        time.Time
+}
+
+func WriteAudit(db *gorm.DB, input Audit) error {
+	model := adminModels.SysOperaLog{Title: "媒体分片", BusinessType: "other", Method: "media_shard.API." + input.Action, RequestMethod: input.Method, OperatorType: "1", OperName: input.Username, OperUrl: input.Route, OperIp: input.ClientIP, Status: input.Status, OperTime: input.At.UTC(), Remark: input.Remark, CreatedAt: input.At.UTC(), UpdatedAt: input.At.UTC()}
+	return db.Create(&model).Error
+}

Sense/server/app/sense/media_shard/config.go

@@ -0,0 +1,80 @@
+package media_shard
+
+import (
+	"encoding/json"
+	"errors"
+	"fmt"
+	"net"
+	"net/url"
+	"os"
+	"strconv"
+	"strings"
+
+	"gorm.io/gorm"
+
+	"git.ilapage.cn/ila/yovision/Sense/server/app/sense/quota"
+)
+
+var ErrInvalidConfig = errors.New("MediaMTX 分片配置不符合要求")
+
+func SpecsFromEnvironment(db *gorm.DB, mode, primaryAPI string) ([]Spec, error) {
+	capacity, err := quota.ReadLimit(db)
+	if raw := strings.TrimSpace(os.Getenv("SENSE_MEDIAMTX_CAPACITY")); raw != "" {
+		capacity, err = strconv.Atoi(raw)
+	}
+	if err != nil || capacity < 1 {
+		return nil, fmt.Errorf("%w: primary capacity", ErrInvalidConfig)
+	}
+	primary := Spec{ID: "primary", Name: "主媒体分片", Mode: mode, ControlAPI: strings.TrimSpace(primaryAPI), Capacity: capacity}
+	if err = validateSpec(primary, true); err != nil {
+		return nil, err
+	}
+	result := []Spec{primary}
+	path := strings.TrimSpace(os.Getenv("SENSE_MEDIAMTX_SHARDS_FILE"))
+	if path == "" {
+		return result, nil
+	}
+	data, err := os.ReadFile(path)
+	if err != nil {
+		return nil, fmt.Errorf("read MediaMTX shards file: %w", err)
+	}
+	var extra []Spec
+	if err = json.Unmarshal(data, &extra); err != nil {
+		return nil, fmt.Errorf("%w: shards JSON", ErrInvalidConfig)
+	}
+	seen := map[string]bool{"primary": true}
+	for _, item := range extra {
+		item.ID, item.Name, item.Mode, item.ControlAPI = strings.TrimSpace(item.ID), strings.TrimSpace(item.Name), strings.ToLower(strings.TrimSpace(item.Mode)), strings.TrimSpace(item.ControlAPI)
+		if item.Mode == "" {
+			item.Mode = "external"
+		}
+		if seen[item.ID] || item.Mode != "external" {
+			return nil, fmt.Errorf("%w: duplicate id or non-external extra shard", ErrInvalidConfig)
+		}
+		if err = validateSpec(item, false); err != nil {
+			return nil, err
+		}
+		seen[item.ID] = true
+		result = append(result, item)
+	}
+	return result, nil
+}
+
+func validateSpec(spec Spec, primary bool) error {
+	if spec.ID == "" || len(spec.ID) > 64 || spec.Name == "" || len([]rune(spec.Name)) > 128 || spec.Capacity < 1 || spec.Capacity > 100000 {
+		return fmt.Errorf("%w: shard identity or capacity", ErrInvalidConfig)
+	}
+	if primary && spec.Mode != "managed" && spec.Mode != "external" {
+		return fmt.Errorf("%w: primary mode", ErrInvalidConfig)
+	}
+	parsed, err := url.Parse(spec.ControlAPI)
+	if err != nil || parsed.Scheme != "http" || parsed.User != nil || parsed.RawQuery != "" || parsed.Fragment != "" || parsed.Path != "" {
+		return fmt.Errorf("%w: control API", ErrInvalidConfig)
+	}
+	host := parsed.Hostname()
+	ip := net.ParseIP(host)
+	if !strings.EqualFold(host, "localhost") && (ip == nil || !ip.IsLoopback()) {
+		return fmt.Errorf("%w: control API must be loopback", ErrInvalidConfig)
+	}
+	return nil
+}

Sense/server/app/sense/media_shard/config_test.go

@@ -0,0 +1,46 @@
+package media_shard
+
+import (
+	"os"
+	"path/filepath"
+	"testing"
+
+	"gorm.io/driver/sqlite"
+	"gorm.io/gorm"
+
+	"git.ilapage.cn/ila/yovision/Sense/server/app/sense/quota"
+)
+
+func TestSpecsFromEnvironmentUsesDatabaseQuotaAndExternalFile(t *testing.T) {
+	db, err := gorm.Open(sqlite.Open(":memory:"), &gorm.Config{})
+	if err != nil {
+		t.Fatal(err)
+	}
+	if err = db.AutoMigrate(&quota.Setting{}); err != nil {
+		t.Fatal(err)
+	}
+	if err = db.Create(&quota.Setting{ID: quota.SettingID, Limit: 23, Source: "database", Version: 1}).Error; err != nil {
+		t.Fatal(err)
+	}
+	path := filepath.Join(t.TempDir(), "shards.json")
+	if err = os.WriteFile(path, []byte(`[{"id":"secondary","name":"备用分片","controlAPI":"http://127.0.0.1:19997","capacity":11}]`), 0600); err != nil {
+		t.Fatal(err)
+	}
+	t.Setenv("SENSE_MEDIAMTX_CAPACITY", "")
+	t.Setenv("SENSE_MEDIAMTX_SHARDS_FILE", path)
+	specs, err := SpecsFromEnvironment(db, "managed", "http://127.0.0.1:9997")
+	if err != nil {
+		t.Fatal(err)
+	}
+	if len(specs) != 2 || specs[0].Capacity != 23 || specs[1].Capacity != 11 || specs[1].Mode != "external" {
+		t.Fatalf("unexpected specs: %+v", specs)
+	}
+}
+
+func TestSpecsRejectsNonLoopbackControlAPI(t *testing.T) {
+	t.Setenv("SENSE_MEDIAMTX_CAPACITY", "5")
+	t.Setenv("SENSE_MEDIAMTX_SHARDS_FILE", "")
+	if _, err := SpecsFromEnvironment(nil, "external", "http://192.0.2.1:9997"); err == nil {
+		t.Fatal("expected unsafe control API to be rejected")
+	}
+}

Sense/server/app/sense/media_shard/dto.go

@@ -0,0 +1,72 @@
+package media_shard
+
+import "time"
+
+type Spec struct {
+	ID         string `json:"id"`
+	Name       string `json:"name"`
+	Mode       string `json:"mode"`
+	ControlAPI string `json:"controlAPI"`
+	Capacity   int    `json:"capacity"`
+}
+
+type Summary struct {
+	Total      int `json:"total"`
+	Available  int `json:"available"`
+	Failed     int `json:"failed"`
+	Configured int `json:"configuredCapacity"`
+	Assigned   int `json:"assignedPaths"`
+	Impacted   int `json:"impactedPaths"`
+}
+
+type Response struct {
+	ID            string     `json:"id"`
+	Name          string     `json:"name"`
+	Mode          string     `json:"mode"`
+	Status        string     `json:"status"`
+	Detail        string     `json:"detail"`
+	Capacity      int        `json:"capacity"`
+	AssignedPaths int        `json:"assignedPaths"`
+	Remaining     int        `json:"remaining"`
+	LastProbeAt   *time.Time `json:"lastProbeAt,omitempty"`
+	Stale         bool       `json:"stale"`
+}
+
+type Impact struct {
+	RouteID      string `json:"routeId"`
+	DeviceID     string `json:"deviceId"`
+	DeviceName   string `json:"deviceName"`
+	Location     string `json:"location"`
+	ProfileToken string `json:"profileToken"`
+	Path         string `json:"path"`
+	Desired      string `json:"desired"`
+	Actual       string `json:"actual"`
+}
+
+type DetailResponse struct {
+	Response
+	Impact []Impact `json:"impact"`
+}
+
+type PageResponse struct {
+	Summary Summary    `json:"summary"`
+	List    []Response `json:"list"`
+	Count   int64      `json:"count"`
+}
+
+type Check struct {
+	Name   string `json:"name"`
+	Passed bool   `json:"passed"`
+	Detail string `json:"detail"`
+}
+
+type PreflightResponse struct {
+	SourceShardID       string  `json:"sourceShardId"`
+	TargetShardID       string  `json:"targetShardId,omitempty"`
+	TargetShardName     string  `json:"targetShardName,omitempty"`
+	ImpactedPaths       int     `json:"impactedPaths"`
+	Ready               bool    `json:"ready"`
+	ExecutionAuthorized bool    `json:"executionAuthorized"`
+	RequiresIssue       bool    `json:"requiresSeparateHighRiskIssue"`
+	Checks              []Check `json:"checks"`
+}

Sense/server/app/sense/media_shard/models.go

@@ -0,0 +1,59 @@
+package media_shard
+
+import "time"
+
+const (
+	StatusUnknown  = "unknown"
+	StatusRunning  = "running"
+	StatusFailed   = "failed"
+	StatusDisabled = "disabled"
+)
+
+// Shard stores only a loopback control endpoint. It is deliberately omitted
+// from API response DTOs so credentials and control-plane topology cannot leak.
+type Shard struct {
+	ID            string     `gorm:"size:64;primaryKey"`
+	Name          string     `gorm:"size:128;not null"`
+	Mode          string     `gorm:"size:16;not null"`
+	ControlAPI    string     `gorm:"size:512;not null"`
+	Capacity      int        `gorm:"not null"`
+	Status        string     `gorm:"size:16;not null;index"`
+	Detail        string     `gorm:"size:256;not null"`
+	LastProbeAt   *time.Time `gorm:"index"`
+	ConfigVersion int64      `gorm:"not null;default:1"`
+	CreatedAt     time.Time
+	UpdatedAt     time.Time
+}
+
+func (Shard) TableName() string { return "sense_media_shards" }
+
+// Assignment is the stable, Sense-owned mapping between an existing media
+// route and a MediaMTX shard. Failures never rewrite this record automatically.
+type Assignment struct {
+	RouteID   string    `gorm:"size:96;primaryKey"`
+	ShardID   string    `gorm:"size:64;not null;index"`
+	Algorithm string    `gorm:"size:32;not null"`
+	CreatedAt time.Time `gorm:"not null"`
+	UpdatedAt time.Time `gorm:"not null"`
+}
+
+func (Assignment) TableName() string { return "sense_media_shard_assignments" }
+
+type routeProjection struct {
+	ID           string `gorm:"column:id"`
+	DeviceID     string `gorm:"column:device_id"`
+	ProfileToken string `gorm:"column:profile_token"`
+	Path         string `gorm:"column:path"`
+	Desired      string `gorm:"column:desired"`
+	Actual       string `gorm:"column:actual"`
+}
+
+func (routeProjection) TableName() string { return "sense_media_routes" }
+
+type deviceProjection struct {
+	ID       string `gorm:"column:id"`
+	Name     string `gorm:"column:name"`
+	Location string `gorm:"column:location"`
+}
+
+func (deviceProjection) TableName() string { return "sense_devices" }

Sense/server/app/sense/media_shard/service.go

@@ -0,0 +1,259 @@
+package media_shard
+
+import (
+	"context"
+	"crypto/sha256"
+	"encoding/binary"
+	"errors"
+	"fmt"
+	"sort"
+	"strings"
+	"time"
+
+	"gorm.io/gorm"
+	"gorm.io/gorm/clause"
+)
+
+var (
+	ErrNotFound   = errors.New("媒体分片不存在")
+	ErrNoCapacity = errors.New("没有可用的媒体分片容量")
+	ErrInvalidID  = errors.New("媒体分片标识不符合要求")
+)
+
+const staleAfter = 30 * time.Second
+
+type Probe func(context.Context, string) error
+
+type Service struct {
+	DB    *gorm.DB
+	Probe Probe
+	Now   func() time.Time
+}
+
+func NewService(db *gorm.DB, probe Probe) *Service {
+	return &Service{DB: db, Probe: probe, Now: time.Now}
+}
+
+func (s *Service) SyncSpecs(ctx context.Context, specs []Spec) error {
+	return s.DB.WithContext(ctx).Transaction(func(tx *gorm.DB) error {
+		for _, spec := range specs {
+			var current Shard
+			err := tx.First(&current, "id = ?", spec.ID).Error
+			if err != nil && !errors.Is(err, gorm.ErrRecordNotFound) {
+				return err
+			}
+			if errors.Is(err, gorm.ErrRecordNotFound) {
+				current = Shard{ID: spec.ID, Status: StatusUnknown, Detail: "等待健康探测", ConfigVersion: 1}
+			} else if current.Name != spec.Name || current.Mode != spec.Mode || current.ControlAPI != spec.ControlAPI || current.Capacity != spec.Capacity {
+				current.ConfigVersion++
+			}
+			current.Name, current.Mode, current.ControlAPI, current.Capacity = spec.Name, spec.Mode, spec.ControlAPI, spec.Capacity
+			if err = tx.Save(&current).Error; err != nil {
+				return err
+			}
+		}
+		ids := make([]string, 0, len(specs))
+		for _, spec := range specs {
+			ids = append(ids, spec.ID)
+		}
+		return tx.Model(&Shard{}).Where("id NOT IN ?", ids).Updates(map[string]any{"status": StatusDisabled, "detail": "分片已从当前运行配置移除", "updated_at": s.now()}).Error
+	})
+}
+
+func (s *Service) RefreshAll(ctx context.Context) error {
+	var shards []Shard
+	if err := s.DB.WithContext(ctx).Where("status <> ?", StatusDisabled).Find(&shards).Error; err != nil {
+		return err
+	}
+	var result error
+	for _, shard := range shards {
+		status, detail := StatusRunning, "Control API 正常"
+		if s.Probe == nil || s.Probe(ctx, shard.ControlAPI) != nil {
+			status, detail = StatusFailed, "Control API 不可用"
+		}
+		now := s.now()
+		if err := s.DB.WithContext(ctx).Model(&Shard{}).Where("id = ?", shard.ID).Updates(map[string]any{"status": status, "detail": detail, "last_probe_at": &now, "updated_at": now}).Error; err != nil {
+			result = errors.Join(result, err)
+		}
+	}
+	return result
+}
+
+func (s *Service) EnsureAssignment(ctx context.Context, routeID string) (Shard, error) {
+	routeID = strings.TrimSpace(routeID)
+	if routeID == "" {
+		return Shard{}, ErrInvalidID
+	}
+	var selected Shard
+	err := s.DB.WithContext(ctx).Transaction(func(tx *gorm.DB) error {
+		var existing Assignment
+		if err := tx.Clauses(clause.Locking{Strength: "UPDATE"}).First(&existing, "route_id = ?", routeID).Error; err == nil {
+			return tx.First(&selected, "id = ?", existing.ShardID).Error
+		} else if !errors.Is(err, gorm.ErrRecordNotFound) {
+			return err
+		}
+		var candidates []Shard
+		if err := tx.Clauses(clause.Locking{Strength: "UPDATE"}).Where("status IN ?", []string{StatusRunning, StatusUnknown}).Find(&candidates).Error; err != nil {
+			return err
+		}
+		counts, err := assignmentCounts(tx)
+		if err != nil {
+			return err
+		}
+		eligible := candidates[:0]
+		for _, item := range candidates {
+			if counts[item.ID] < int64(item.Capacity) {
+				eligible = append(eligible, item)
+			}
+		}
+		if len(eligible) == 0 {
+			return ErrNoCapacity
+		}
+		sort.SliceStable(eligible, func(i, j int) bool { return rendezvous(routeID, eligible[i].ID) > rendezvous(routeID, eligible[j].ID) })
+		selected = eligible[0]
+		now := s.now()
+		assignment := Assignment{RouteID: routeID, ShardID: selected.ID, Algorithm: "rendezvous-v1", CreatedAt: now, UpdatedAt: now}
+		if err := tx.Clauses(clause.OnConflict{DoNothing: true}).Create(&assignment).Error; err != nil {
+			return err
+		}
+		if assignment.RouteID != "" { // reload also handles a concurrent winner on PostgreSQL.
+			if err := tx.First(&existing, "route_id = ?", routeID).Error; err != nil {
+				return err
+			}
+			return tx.First(&selected, "id = ?", existing.ShardID).Error
+		}
+		return nil
+	})
+	return selected, err
+}
+
+func (s *Service) ResolveRoute(ctx context.Context, routeID string) (Shard, error) {
+	var shard Shard
+	err := s.DB.WithContext(ctx).Table("sense_media_shards AS s").Select("s.*").Joins("JOIN sense_media_shard_assignments a ON a.shard_id = s.id").Where("a.route_id = ?", routeID).Scan(&shard).Error
+	if err != nil {
+		return Shard{}, err
+	}
+	if shard.ID == "" {
+		return Shard{}, ErrNotFound
+	}
+	return shard, nil
+}
+
+func (s *Service) List(ctx context.Context) (PageResponse, error) {
+	var shards []Shard
+	if err := s.DB.WithContext(ctx).Order("name ASC, id ASC").Find(&shards).Error; err != nil {
+		return PageResponse{}, err
+	}
+	counts, err := assignmentCounts(s.DB.WithContext(ctx))
+	if err != nil {
+		return PageResponse{}, err
+	}
+	result := PageResponse{List: make([]Response, 0, len(shards)), Count: int64(len(shards))}
+	for _, shard := range shards {
+		item := s.response(shard, int(counts[shard.ID]))
+		result.List = append(result.List, item)
+		result.Summary.Total++
+		result.Summary.Configured += item.Capacity
+		result.Summary.Assigned += item.AssignedPaths
+		if item.Status == StatusRunning {
+			result.Summary.Available++
+		}
+		if item.Status == StatusFailed {
+			result.Summary.Failed++
+			result.Summary.Impacted += item.AssignedPaths
+		}
+	}
+	return result, nil
+}
+
+func (s *Service) Get(ctx context.Context, id string) (DetailResponse, error) {
+	id = strings.TrimSpace(id)
+	if id == "" || len(id) > 64 {
+		return DetailResponse{}, ErrInvalidID
+	}
+	var shard Shard
+	if err := s.DB.WithContext(ctx).First(&shard, "id = ?", id).Error; err != nil {
+		if errors.Is(err, gorm.ErrRecordNotFound) {
+			return DetailResponse{}, ErrNotFound
+		}
+		return DetailResponse{}, err
+	}
+	impact, err := s.impact(ctx, id)
+	if err != nil {
+		return DetailResponse{}, err
+	}
+	return DetailResponse{Response: s.response(shard, len(impact)), Impact: impact}, nil
+}
+
+func (s *Service) Preflight(ctx context.Context, id string) (PreflightResponse, error) {
+	detail, err := s.Get(ctx, id)
+	if err != nil {
+		return PreflightResponse{}, err
+	}
+	result := PreflightResponse{SourceShardID: id, ImpactedPaths: len(detail.Impact), ExecutionAuthorized: false, RequiresIssue: true}
+	result.Checks = append(result.Checks, Check{Name: "源分片状态", Passed: detail.Status == StatusRunning && !detail.Stale, Detail: detail.Detail})
+	var candidates []Shard
+	if err = s.DB.WithContext(ctx).Where("id <> ? AND status = ?", id, StatusRunning).Find(&candidates).Error; err != nil {
+		return PreflightResponse{}, err
+	}
+	counts, err := assignmentCounts(s.DB.WithContext(ctx))
+	if err != nil {
+		return PreflightResponse{}, err
+	}
+	for _, item := range candidates {
+		if item.Capacity-int(counts[item.ID]) >= len(detail.Impact) && (result.TargetShardID == "" || rendezvous(id, item.ID) > rendezvous(id, result.TargetShardID)) {
+			result.TargetShardID, result.TargetShardName = item.ID, item.Name
+		}
+	}
+	result.Checks = append(result.Checks, Check{Name: "目标容量", Passed: result.TargetShardID != "", Detail: map[bool]string{true: "存在可容纳全部受影响路径的健康目标分片", false: "没有可容纳全部受影响路径的健康目标分片"}[result.TargetShardID != ""]})
+	result.Checks = append(result.Checks, Check{Name: "实施授权", Passed: false, Detail: "本工单只提供只读预检；实际迁移必须另建高风险工单并人工确认"})
+	result.Ready = result.Checks[0].Passed && result.Checks[1].Passed
+	return result, nil
+}
+
+func (s *Service) impact(ctx context.Context, id string) ([]Impact, error) {
+	var rows []Impact
+	err := s.DB.WithContext(ctx).Table("sense_media_shard_assignments AS a").Select("r.id AS route_id, r.device_id, COALESCE(d.name, '') AS device_name, COALESCE(d.location, '') AS location, r.profile_token, r.path, r.desired, r.actual").Joins("JOIN sense_media_routes r ON r.id = a.route_id").Joins("LEFT JOIN sense_devices d ON d.id = r.device_id").Where("a.shard_id = ?", id).Order("d.name ASC, r.profile_token ASC").Scan(&rows).Error
+	return rows, err
+}
+
+func (s *Service) response(shard Shard, assigned int) Response {
+	remaining := shard.Capacity - assigned
+	if remaining < 0 {
+		remaining = 0
+	}
+	stale := shard.LastProbeAt == nil || s.now().Sub(shard.LastProbeAt.UTC()) > staleAfter
+	return Response{ID: shard.ID, Name: shard.Name, Mode: shard.Mode, Status: shard.Status, Detail: shard.Detail, Capacity: shard.Capacity, AssignedPaths: assigned, Remaining: remaining, LastProbeAt: shard.LastProbeAt, Stale: stale}
+}
+
+func assignmentCounts(db *gorm.DB) (map[string]int64, error) {
+	type row struct {
+		ShardID string
+		Count   int64
+	}
+	var rows []row
+	err := db.Model(&Assignment{}).Select("shard_id, COUNT(*) AS count").Group("shard_id").Scan(&rows).Error
+	out := map[string]int64{}
+	for _, r := range rows {
+		out[r.ShardID] = r.Count
+	}
+	return out, err
+}
+func rendezvous(routeID, shardID string) uint64 {
+	sum := sha256.Sum256([]byte(routeID + "\x00" + shardID))
+	return binary.BigEndian.Uint64(sum[:8])
+}
+func (s *Service) now() time.Time {
+	if s.Now != nil {
+		return s.Now().UTC()
+	}
+	return time.Now().UTC()
+}
+
+func (s *Service) ControlAPI(ctx context.Context, id string) (string, error) {
+	var shard Shard
+	if err := s.DB.WithContext(ctx).First(&shard, "id = ?", id).Error; err != nil {
+		return "", fmt.Errorf("get media shard control endpoint: %w", err)
+	}
+	return shard.ControlAPI, nil
+}

Sense/server/app/sense/media_shard/service_test.go

@@ -0,0 +1,130 @@
+package media_shard
+
+import (
+	"context"
+	"errors"
+	"testing"
+	"time"
+
+	"gorm.io/driver/sqlite"
+	"gorm.io/gorm"
+)
+
+type testRoute struct{ ID, DeviceID, ProfileToken, Path, Desired, Actual string }
+
+func (testRoute) TableName() string { return "sense_media_routes" }
+
+type testDevice struct{ ID, Name, Location string }
+
+func (testDevice) TableName() string { return "sense_devices" }
+
+func testService(t *testing.T) (*Service, *gorm.DB) {
+	t.Helper()
+	db, err := gorm.Open(sqlite.Open("file:"+t.Name()+"?mode=memory&cache=shared"), &gorm.Config{})
+	if err != nil {
+		t.Fatal(err)
+	}
+	if err = db.AutoMigrate(&Shard{}, &Assignment{}, &testRoute{}, &testDevice{}); err != nil {
+		t.Fatal(err)
+	}
+	now := time.Date(2026, 8, 28, 8, 0, 0, 0, time.UTC)
+	service := NewService(db, func(context.Context, string) error { return nil })
+	service.Now = func() time.Time { return now }
+	return service, db
+}
+
+func TestStableAssignmentUsesConfiguredCapacity(t *testing.T) {
+	service, _ := testService(t)
+	ctx := context.Background()
+	if err := service.SyncSpecs(ctx, []Spec{{ID: "alpha", Name: "A", Mode: "external", ControlAPI: "http://127.0.0.1:9997", Capacity: 3}, {ID: "beta", Name: "B", Mode: "external", ControlAPI: "http://127.0.0.1:19997", Capacity: 2}}); err != nil {
+		t.Fatal(err)
+	}
+	if err := service.RefreshAll(ctx); err != nil {
+		t.Fatal(err)
+	}
+	first, err := service.EnsureAssignment(ctx, "route-1")
+	if err != nil {
+		t.Fatal(err)
+	}
+	second, err := service.EnsureAssignment(ctx, "route-1")
+	if err != nil {
+		t.Fatal(err)
+	}
+	if first.ID != second.ID {
+		t.Fatalf("assignment changed from %s to %s", first.ID, second.ID)
+	}
+	for _, id := range []string{"route-2", "route-3", "route-4", "route-5"} {
+		if _, err = service.EnsureAssignment(ctx, id); err != nil {
+			t.Fatalf("assign %s: %v", id, err)
+		}
+	}
+	if _, err = service.EnsureAssignment(ctx, "route-6"); !errors.Is(err, ErrNoCapacity) {
+		t.Fatalf("expected configured capacity error, got %v", err)
+	}
+}
+
+func TestFailureKeepsOwnershipAndLocatesImpact(t *testing.T) {
+	service, db := testService(t)
+	ctx := context.Background()
+	if err := service.SyncSpecs(ctx, []Spec{{ID: "failed", Name: "故障分片", Mode: "external", ControlAPI: "http://127.0.0.1:9997", Capacity: 7}}); err != nil {
+		t.Fatal(err)
+	}
+	if err := service.RefreshAll(ctx); err != nil {
+		t.Fatal(err)
+	}
+	if err := db.Create(&testDevice{ID: "device-1", Name: "东门摄像机", Location: "东门"}).Error; err != nil {
+		t.Fatal(err)
+	}
+	if err := db.Create(&testRoute{ID: "device-1:main", DeviceID: "device-1", ProfileToken: "main", Path: "sense_abc", Desired: "running", Actual: "ready"}).Error; err != nil {
+		t.Fatal(err)
+	}
+	if _, err := service.EnsureAssignment(ctx, "device-1:main"); err != nil {
+		t.Fatal(err)
+	}
+	service.Probe = func(context.Context, string) error { return errors.New("offline") }
+	if err := service.RefreshAll(ctx); err != nil {
+		t.Fatal(err)
+	}
+	detail, err := service.Get(ctx, "failed")
+	if err != nil {
+		t.Fatal(err)
+	}
+	if detail.Status != StatusFailed || len(detail.Impact) != 1 || detail.Impact[0].DeviceName != "东门摄像机" || detail.Impact[0].ProfileToken != "main" {
+		t.Fatalf("unexpected detail: %+v", detail)
+	}
+	resolved, err := service.ResolveRoute(ctx, "device-1:main")
+	if err != nil || resolved.ID != "failed" {
+		t.Fatalf("failure rewrote ownership: %+v %v", resolved, err)
+	}
+}
+
+func TestMigrationPreflightIsReadOnly(t *testing.T) {
+	service, db := testService(t)
+	ctx := context.Background()
+	if err := service.SyncSpecs(ctx, []Spec{{ID: "source", Name: "源", Mode: "external", ControlAPI: "http://127.0.0.1:9997", Capacity: 1}, {ID: "target", Name: "目标", Mode: "external", ControlAPI: "http://127.0.0.1:19997", Capacity: 4}}); err != nil {
+		t.Fatal(err)
+	}
+	if err := service.RefreshAll(ctx); err != nil {
+		t.Fatal(err)
+	}
+	if err := db.Create(&testRoute{ID: "route", DeviceID: "d", ProfileToken: "p", Path: "path", Desired: "running", Actual: "ready"}).Error; err != nil {
+		t.Fatal(err)
+	}
+	if err := db.Create(&Assignment{RouteID: "route", ShardID: "source", Algorithm: "rendezvous-v1", CreatedAt: service.now(), UpdatedAt: service.now()}).Error; err != nil {
+		t.Fatal(err)
+	}
+	before := Assignment{}
+	db.First(&before, "route_id = ?", "route")
+	result, err := service.Preflight(ctx, "source")
+	if err != nil {
+		t.Fatal(err)
+	}
+	after := Assignment{}
+	db.First(&after, "route_id = ?", "route")
+	if !result.Ready || result.ExecutionAuthorized || !result.RequiresIssue || result.TargetShardID != "target" {
+		t.Fatalf("unexpected preflight: %+v", result)
+	}
+	if before.ShardID != after.ShardID || !before.UpdatedAt.Equal(after.UpdatedAt) {
+		t.Fatalf("preflight mutated assignment: before=%+v after=%+v", before, after)
+	}
+}

Sense/server/cmd/migrate/migration/version/2026082814000_media_shard.go

@@ -0,0 +1,57 @@
+package version
+
+import (
+	"fmt"
+	"runtime"
+
+	"gorm.io/gorm"
+	"gorm.io/gorm/clause"
+
+	"git.ilapage.cn/ila/yovision/Sense/server/app/sense/media_shard"
+	"git.ilapage.cn/ila/yovision/Sense/server/cmd/migrate/migration"
+	migrationModels "git.ilapage.cn/ila/yovision/Sense/server/cmd/migrate/migration/models"
+	common "git.ilapage.cn/ila/yovision/Sense/server/common/models"
+)
+
+func init() {
+	_, fileName, _, _ := runtime.Caller(0)
+	migration.Migrate.SetVersion(migration.GetFilename(fileName), migrateSenseMediaShard)
+}
+
+func migrateSenseMediaShard(db *gorm.DB, version string) error {
+	return db.Transaction(func(tx *gorm.DB) error {
+		if err := tx.AutoMigrate(&media_shard.Shard{}, &media_shard.Assignment{}); err != nil {
+			return err
+		}
+		root, err := ensureDeviceMenu(tx, migrationModels.SysMenu{MenuName: senseLayoutMenuName, Title: "视频感知", Icon: "video-camera", Path: "/sense", MenuType: "M", Component: "Layout", Sort: 5, Visible: "0", IsFrame: "1"})
+		if err != nil {
+			return err
+		}
+		page, err := ensureDeviceMenu(tx, migrationModels.SysMenu{MenuName: "SenseMediaShard", Title: "媒体分片", Icon: "connection", Path: "media-shard", Paths: fmt.Sprintf("/0/%d", root.MenuId), MenuType: "C", Permission: "sense:media-shard:list", ParentId: root.MenuId, Component: "/sense/media-shard/index", Sort: 10, Visible: "0", IsFrame: "1"})
+		if err != nil {
+			return err
+		}
+		detail, err := ensureDeviceMenu(tx, migrationModels.SysMenu{MenuName: "SenseMediaShardDetail", Title: "查看分片详情", MenuType: "F", Action: "GET", Permission: "sense:media-shard:detail", ParentId: page.MenuId, Paths: fmt.Sprintf("/0/%d/%d", root.MenuId, page.MenuId), Sort: 1, Visible: "1", IsFrame: "1"})
+		if err != nil {
+			return err
+		}
+		preflight, err := ensureDeviceMenu(tx, migrationModels.SysMenu{MenuName: "SenseMediaShardPreflight", Title: "迁移预检", MenuType: "F", Action: "GET", Permission: "sense:media-shard:preflight", ParentId: page.MenuId, Paths: fmt.Sprintf("/0/%d/%d", root.MenuId, page.MenuId), Sort: 2, Visible: "1", IsFrame: "1"})
+		if err != nil {
+			return err
+		}
+		for _, role := range []string{"implementation_operator", "site_admin", "viewer"} {
+			if err = attachDeviceRole(tx, role, []migrationModels.SysMenu{page, detail, preflight}); err != nil {
+				return err
+			}
+			for _, policy := range [][2]string{{"/api/v1/media-shards", "GET"}, {"/api/v1/media-shards/:id", "GET"}, {"/api/v1/media-shards/:id/migration-preflight", "GET"}} {
+				if err = tx.Clauses(clause.OnConflict{DoNothing: true}).Create(&deviceCasbinRule{Ptype: "p", V0: role, V1: policy[0], V2: policy[1]}).Error; err != nil {
+					return err
+				}
+			}
+		}
+		if err = rebuildSenseMenuPaths(tx, root.MenuId, "/0"); err != nil {
+			return err
+		}
+		return tx.Create(&common.Migration{Version: version}).Error
+	})
+}

Sense/server/cmd/migrate/migration/version/2026082814000_media_shard_test.go

@@ -0,0 +1,86 @@
+package version
+
+import (
+	"os"
+	"testing"
+
+	"gorm.io/driver/postgres"
+	"gorm.io/driver/sqlite"
+	"gorm.io/gorm"
+
+	"git.ilapage.cn/ila/yovision/Sense/server/app/sense/media_shard"
+	migrationModels "git.ilapage.cn/ila/yovision/Sense/server/cmd/migrate/migration/models"
+	common "git.ilapage.cn/ila/yovision/Sense/server/common/models"
+)
+
+func TestMediaShardMigrationAddsReadOnlySurfaceWithoutFixtures(t *testing.T) {
+	db, err := gorm.Open(sqlite.Open(":memory:"), &gorm.Config{})
+	if err != nil {
+		t.Fatal(err)
+	}
+	if err = db.AutoMigrate(&migrationModels.SysRole{}, &migrationModels.SysMenu{}, &deviceCasbinRule{}, &common.Migration{}); err != nil {
+		t.Fatal(err)
+	}
+	for _, role := range []string{"implementation_operator", "site_admin", "viewer"} {
+		if err = db.Create(&migrationModels.SysRole{RoleName: role, RoleKey: role, Status: "2"}).Error; err != nil {
+			t.Fatal(err)
+		}
+	}
+	const version = "2026082814000_media_shard.go"
+	if err = migrateSenseMediaShard(db, version); err != nil {
+		t.Fatal(err)
+	}
+	if !db.Migrator().HasTable(&media_shard.Shard{}) || !db.Migrator().HasTable(&media_shard.Assignment{}) {
+		t.Fatal("media shard tables missing")
+	}
+	var menus, reads, writes, fixtures, applied int64
+	db.Model(&migrationModels.SysMenu{}).Where("menu_name LIKE ?", "SenseMediaShard%").Count(&menus)
+	db.Model(&deviceCasbinRule{}).Where("v1 LIKE ? AND v2 = ?", "/api/v1/media-shards%", "GET").Count(&reads)
+	db.Model(&deviceCasbinRule{}).Where("v1 LIKE ? AND v2 <> ?", "/api/v1/media-shards%", "GET").Count(&writes)
+	db.Model(&media_shard.Shard{}).Count(&fixtures)
+	db.Model(&common.Migration{}).Where("version = ?", version).Count(&applied)
+	if menus != 3 || reads != 9 || writes != 0 || fixtures != 0 || applied != 1 {
+		t.Fatalf("menus=%d reads=%d writes=%d fixtures=%d applied=%d", menus, reads, writes, fixtures, applied)
+	}
+}
+
+func TestMediaShardMigrationOnPostgres(t *testing.T) {
+	dsn := os.Getenv("SENSE_MEDIA_SHARD_MIGRATION_TEST_DATABASE_URL")
+	if dsn == "" {
+		t.Skip("set SENSE_MEDIA_SHARD_MIGRATION_TEST_DATABASE_URL to run the PostgreSQL migration test")
+	}
+	db, err := gorm.Open(postgres.Open(dsn), &gorm.Config{})
+	if err != nil {
+		t.Fatal(err)
+	}
+	const schema = "sense_media_shard_77_test"
+	if err = db.Exec("DROP SCHEMA IF EXISTS " + schema + " CASCADE").Error; err != nil {
+		t.Fatal(err)
+	}
+	if err = db.Exec("CREATE SCHEMA " + schema).Error; err != nil {
+		t.Fatal(err)
+	}
+	t.Cleanup(func() { _ = db.Exec("DROP SCHEMA IF EXISTS " + schema + " CASCADE").Error })
+	sqlDB, err := db.DB()
+	if err != nil {
+		t.Fatal(err)
+	}
+	sqlDB.SetMaxOpenConns(1)
+	if err = db.Exec("SET search_path TO " + schema).Error; err != nil {
+		t.Fatal(err)
+	}
+	if err = db.AutoMigrate(&migrationModels.SysRole{}, &migrationModels.SysMenu{}, &deviceCasbinRule{}, &common.Migration{}); err != nil {
+		t.Fatal(err)
+	}
+	for _, role := range []string{"implementation_operator", "site_admin", "viewer"} {
+		if err = db.Create(&migrationModels.SysRole{RoleName: role, RoleKey: role, Status: "2"}).Error; err != nil {
+			t.Fatal(err)
+		}
+	}
+	if err = migrateSenseMediaShard(db, "2026082814000_media_shard.go"); err != nil {
+		t.Fatal(err)
+	}
+	if !db.Migrator().HasTable(&media_shard.Shard{}) || !db.Migrator().HasTable(&media_shard.Assignment{}) {
+		t.Fatal("media shard tables missing")
+	}
+}

Sense/server/config/credential.env.example

@@ -9,3 +9,5 @@ SENSE_ONVIF_ALLOWED_CIDRS=
 SENSE_MEDIAMTX_BINARY=
 SENSE_MEDIAMTX_CONFIG=
 SENSE_MEDIAMTX_API=http://127.0.0.1:9997
+SENSE_MEDIAMTX_CAPACITY=
+SENSE_MEDIAMTX_SHARDS_FILE=

Sense/server/tests/media/integration_test.go

@@ -10,6 +10,9 @@ import (
 	"time"
 
 	"git.ilapage.cn/ila/yovision/Sense/server/app/sense/media"
+	"git.ilapage.cn/ila/yovision/Sense/server/app/sense/media_shard"
+	"gorm.io/driver/sqlite"
+	"gorm.io/gorm"
 )
 
 func freeAddress(t *testing.T) string {
@@ -25,6 +28,81 @@ func freeAddress(t *testing.T) string {
 	return address
 }
 
+func startTestMediaMTX(t *testing.T, binary string) string {
+	t.Helper()
+	apiAddress, rtspAddress := freeAddress(t), freeAddress(t)
+	configPath := filepath.Join(t.TempDir(), "mediamtx.yml")
+	config := fmt.Sprintf("logLevel: warn\napi: true\napiAddress: %s\nrtspAddress: %s\nrtspTransports: [tcp]\nrtmp: false\nhls: false\nwebrtc: false\nsrt: false\nmoq: false\nplayback: false\npaths: {}\n", apiAddress, rtspAddress)
+	if err := os.WriteFile(configPath, []byte(config), 0o600); err != nil {
+		t.Fatal(err)
+	}
+	supervisor := media.NewSupervisor(binary, configPath)
+	if err := supervisor.Start(context.Background()); err != nil {
+		t.Fatal(err)
+	}
+	t.Cleanup(func() {
+		ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
+		defer cancel()
+		_ = supervisor.Stop(ctx)
+	})
+	controller, err := media.NewHTTPController("http://" + apiAddress)
+	if err != nil {
+		t.Fatal(err)
+	}
+	deadline := time.Now().Add(8 * time.Second)
+	for {
+		if err = controller.Health(context.Background()); err == nil {
+			break
+		}
+		if time.Now().After(deadline) {
+			t.Fatalf("MediaMTX did not become ready: %v", err)
+		}
+		time.Sleep(100 * time.Millisecond)
+	}
+	return "http://" + apiAddress
+}
+
+func TestTwoRealMediaMTXShardsAreIndependentlyUsable(t *testing.T) {
+	binary := os.Getenv("SENSE_MEDIAMTX_TEST_BINARY")
+	if binary == "" {
+		t.Skip("set SENSE_MEDIAMTX_TEST_BINARY to run the real MediaMTX integration")
+	}
+	first, second := startTestMediaMTX(t, binary), startTestMediaMTX(t, binary)
+	db, err := gorm.Open(sqlite.Open("file:real_media_shards?mode=memory&cache=shared"), &gorm.Config{})
+	if err != nil {
+		t.Fatal(err)
+	}
+	if err = db.AutoMigrate(&media_shard.Shard{}, &media_shard.Assignment{}); err != nil {
+		t.Fatal(err)
+	}
+	service := media_shard.NewService(db, func(ctx context.Context, endpoint string) error {
+		controller, buildErr := media.NewHTTPController(endpoint)
+		if buildErr != nil {
+			return buildErr
+		}
+		return controller.Health(ctx)
+	})
+	specs := []media_shard.Spec{{ID: "one", Name: "测试分片一", Mode: "external", ControlAPI: first, Capacity: 2}, {ID: "two", Name: "测试分片二", Mode: "external", ControlAPI: second, Capacity: 2}}
+	if err = service.SyncSpecs(context.Background(), specs); err != nil {
+		t.Fatal(err)
+	}
+	if err = service.RefreshAll(context.Background()); err != nil {
+		t.Fatal(err)
+	}
+	for _, routeID := range []string{"route-a", "route-b", "route-c", "route-d"} {
+		if _, err = service.EnsureAssignment(context.Background(), routeID); err != nil {
+			t.Fatalf("assign %s: %v", routeID, err)
+		}
+	}
+	result, err := service.List(context.Background())
+	if err != nil {
+		t.Fatal(err)
+	}
+	if result.Summary.Available != 2 || result.Summary.Assigned != 4 || result.Summary.Configured != 4 {
+		t.Fatalf("unexpected real shard summary: %+v", result.Summary)
+	}
+}
+
 func TestRealMediaMTXControlLifecycle(t *testing.T) {
 	binary := os.Getenv("SENSE_MEDIAMTX_TEST_BINARY")
 	if binary == "" {
@@ -32,7 +110,7 @@ func TestRealMediaMTXControlLifecycle(t *testing.T) {
 	}
 	apiAddress, rtspAddress := freeAddress(t), freeAddress(t)
 	configPath := filepath.Join(t.TempDir(), "mediamtx.yml")
-	config := fmt.Sprintf("logLevel: warn\napi: true\napiAddress: %s\nrtspAddress: %s\nrtmp: false\nhls: false\nwebrtc: false\nsrt: false\nplayback: false\npaths: {}\n", apiAddress, rtspAddress)
+	config := fmt.Sprintf("logLevel: warn\napi: true\napiAddress: %s\nrtspAddress: %s\nrtspTransports: [tcp]\nrtmp: false\nhls: false\nwebrtc: false\nsrt: false\nmoq: false\nplayback: false\npaths: {}\n", apiAddress, rtspAddress)
 	if err := os.WriteFile(configPath, []byte(config), 0o600); err != nil {
 		t.Fatal(err)
 	}