292 lines
8.7 KiB
Go
292 lines
8.7 KiB
Go
package liveview
|
|
|
|
import (
|
|
"context"
|
|
"crypto/rand"
|
|
"encoding/hex"
|
|
"errors"
|
|
"fmt"
|
|
"net"
|
|
"net/http"
|
|
"net/url"
|
|
"os"
|
|
"regexp"
|
|
"strings"
|
|
"sync"
|
|
"time"
|
|
|
|
"gorm.io/gorm"
|
|
)
|
|
|
|
var (
|
|
ErrInvalidRequest = errors.New("实时监看请求不符合要求")
|
|
ErrRouteNotFound = errors.New("可监看的视频不存在")
|
|
ErrSessionExpired = errors.New("播放会话已过期")
|
|
validMediaPath = regexp.MustCompile(`^[A-Za-z0-9_-]{1,96}$`)
|
|
validDNSHost = regexp.MustCompile(`^[A-Za-z0-9.-]+$`)
|
|
)
|
|
|
|
const sessionTTL = 2 * time.Minute
|
|
|
|
type PageRequest struct {
|
|
Keyword string
|
|
PageIndex int
|
|
PageSize int
|
|
}
|
|
|
|
type RouteResponse struct {
|
|
ID string `json:"id"`
|
|
DeviceID string `json:"deviceId"`
|
|
DeviceName string `json:"deviceName"`
|
|
DeviceLocation string `json:"deviceLocation"`
|
|
ProfileToken string `json:"profileToken"`
|
|
ProfileName string `json:"profileName"`
|
|
ProfileKind string `json:"profileKind"`
|
|
Width int `json:"width"`
|
|
Height int `json:"height"`
|
|
Encoding string `json:"encoding"`
|
|
Actual string `json:"actual"`
|
|
Detail string `json:"detail"`
|
|
Readers int `json:"readers"`
|
|
}
|
|
|
|
type SessionResponse struct {
|
|
ID string `json:"id"`
|
|
RouteID string `json:"routeId"`
|
|
PlayerURL string `json:"playerUrl"`
|
|
Status string `json:"status"`
|
|
Detail string `json:"detail"`
|
|
ExpiresAt time.Time `json:"expiresAt"`
|
|
DeviceName string `json:"deviceName"`
|
|
ProfileName string `json:"profileName"`
|
|
}
|
|
|
|
type routeRecord struct {
|
|
RouteResponse
|
|
Path string `gorm:"column:path"`
|
|
Desired string `gorm:"column:desired"`
|
|
}
|
|
|
|
type sessionRecord struct {
|
|
ID string
|
|
OwnerID int
|
|
RouteID string
|
|
TargetURL string
|
|
ExpiresAt time.Time
|
|
DeviceName string
|
|
ProfileName string
|
|
}
|
|
|
|
type Store struct {
|
|
mu sync.RWMutex
|
|
sessions map[string]sessionRecord
|
|
now func() time.Time
|
|
}
|
|
|
|
func NewStore() *Store {
|
|
return &Store{sessions: make(map[string]sessionRecord), now: time.Now}
|
|
}
|
|
|
|
var defaultStore = NewStore()
|
|
|
|
type Service struct {
|
|
db *gorm.DB
|
|
store *Store
|
|
}
|
|
|
|
func NewService(db *gorm.DB, store *Store) *Service {
|
|
if store == nil {
|
|
store = defaultStore
|
|
}
|
|
return &Service{db: db, store: store}
|
|
}
|
|
|
|
func (s *Service) List(ctx context.Context, request PageRequest) ([]RouteResponse, int64, error) {
|
|
if request.PageIndex < 1 {
|
|
request.PageIndex = 1
|
|
}
|
|
if request.PageSize < 1 {
|
|
request.PageSize = 10
|
|
}
|
|
if request.PageSize > 50 {
|
|
request.PageSize = 50
|
|
}
|
|
query := s.routeQuery(ctx).Where("r.desired = ?", "running")
|
|
keyword := strings.TrimSpace(request.Keyword)
|
|
if len([]rune(keyword)) > 128 {
|
|
return nil, 0, ErrInvalidRequest
|
|
}
|
|
if keyword != "" {
|
|
pattern := "%" + escapeLike(keyword) + "%"
|
|
query = query.Where("LOWER(d.name) LIKE LOWER(?) ESCAPE '\\' OR LOWER(d.location) LIKE LOWER(?) ESCAPE '\\' OR LOWER(p.name) LIKE LOWER(?) ESCAPE '\\'", pattern, pattern, pattern)
|
|
}
|
|
var total int64
|
|
if err := query.Count(&total).Error; err != nil {
|
|
return nil, 0, err
|
|
}
|
|
var records []routeRecord
|
|
if err := query.Order("d.name ASC, p.kind ASC, p.width DESC").Offset((request.PageIndex - 1) * request.PageSize).Limit(request.PageSize).Scan(&records).Error; err != nil {
|
|
return nil, 0, err
|
|
}
|
|
items := make([]RouteResponse, 0, len(records))
|
|
for _, record := range records {
|
|
items = append(items, record.RouteResponse)
|
|
}
|
|
return items, total, nil
|
|
}
|
|
|
|
func (s *Service) Create(ctx context.Context, ownerID int, routeID string, request *http.Request) (SessionResponse, error) {
|
|
if ownerID < 1 || strings.TrimSpace(routeID) == "" {
|
|
return SessionResponse{}, ErrInvalidRequest
|
|
}
|
|
record, err := s.route(ctx, routeID)
|
|
if err != nil {
|
|
return SessionResponse{}, err
|
|
}
|
|
if record.Desired != "running" || !validMediaPath.MatchString(record.Path) {
|
|
return SessionResponse{}, ErrRouteNotFound
|
|
}
|
|
base, err := playbackBase(request)
|
|
if err != nil {
|
|
return SessionResponse{}, err
|
|
}
|
|
target := *base
|
|
target.Path = strings.TrimRight(target.Path, "/") + "/" + record.Path
|
|
target.RawQuery = "controls=true&muted=true&autoplay=true"
|
|
id, err := randomID()
|
|
if err != nil {
|
|
return SessionResponse{}, err
|
|
}
|
|
now := s.store.now().UTC()
|
|
session := sessionRecord{ID: id, OwnerID: ownerID, RouteID: record.ID, TargetURL: target.String(), ExpiresAt: now.Add(sessionTTL), DeviceName: record.DeviceName, ProfileName: record.ProfileName}
|
|
s.store.put(session)
|
|
return responseFrom(session, record), nil
|
|
}
|
|
|
|
func (s *Service) Get(ctx context.Context, ownerID int, id string) (SessionResponse, error) {
|
|
session, err := s.store.get(id, ownerID, true)
|
|
if err != nil {
|
|
return SessionResponse{}, err
|
|
}
|
|
record, err := s.route(ctx, session.RouteID)
|
|
if err != nil || record.Desired != "running" {
|
|
return SessionResponse{}, ErrRouteNotFound
|
|
}
|
|
session.ExpiresAt = s.store.now().UTC().Add(sessionTTL)
|
|
s.store.put(session)
|
|
return responseFrom(session, record), nil
|
|
}
|
|
|
|
func (s *Service) PlayerTarget(id string) (string, error) {
|
|
session, err := s.store.get(id, 0, false)
|
|
if err != nil {
|
|
return "", err
|
|
}
|
|
return session.TargetURL, nil
|
|
}
|
|
|
|
func (s *Service) route(ctx context.Context, id string) (routeRecord, error) {
|
|
var record routeRecord
|
|
if err := s.routeQuery(ctx).Where("r.id = ?", id).Limit(1).Scan(&record).Error; err != nil {
|
|
return routeRecord{}, err
|
|
}
|
|
if record.ID == "" {
|
|
return routeRecord{}, ErrRouteNotFound
|
|
}
|
|
return record, nil
|
|
}
|
|
|
|
func (s *Service) routeQuery(ctx context.Context) *gorm.DB {
|
|
return s.db.WithContext(ctx).Table("sense_media_routes AS r").
|
|
Select("r.id, r.device_id, d.name AS device_name, d.location AS device_location, r.profile_token, p.name AS profile_name, p.kind AS profile_kind, p.width, p.height, p.encoding, r.actual, r.detail, r.readers, r.path, r.desired").
|
|
Joins("JOIN sense_devices AS d ON d.id = r.device_id").
|
|
Joins("JOIN sense_admission_profiles AS p ON p.device_id = r.device_id AND p.token = r.profile_token").
|
|
Where("d.status <> ? AND p.verification_status = ?", "disabled", "ready")
|
|
}
|
|
|
|
func responseFrom(session sessionRecord, route routeRecord) SessionResponse {
|
|
return SessionResponse{ID: session.ID, RouteID: session.RouteID, PlayerURL: "/api/v1/liveview/player/" + session.ID, Status: playbackStatus(route.Actual), Detail: route.Detail, ExpiresAt: session.ExpiresAt, DeviceName: session.DeviceName, ProfileName: session.ProfileName}
|
|
}
|
|
|
|
func playbackStatus(actual string) string {
|
|
switch actual {
|
|
case "ready", "waiting", "stopped":
|
|
return actual
|
|
case "credential_unavailable", "profile_unavailable":
|
|
return "authentication_failed"
|
|
case "path_missing", "apply_failed", "status_unavailable":
|
|
return "stream_not_found"
|
|
case "process_unavailable":
|
|
return "service_unavailable"
|
|
default:
|
|
return "offline"
|
|
}
|
|
}
|
|
|
|
func (s *Store) put(session sessionRecord) {
|
|
s.mu.Lock()
|
|
defer s.mu.Unlock()
|
|
now := s.now()
|
|
for id, item := range s.sessions {
|
|
if !now.Before(item.ExpiresAt) || item.OwnerID == session.OwnerID {
|
|
delete(s.sessions, id)
|
|
}
|
|
}
|
|
s.sessions[session.ID] = session
|
|
}
|
|
|
|
func (s *Store) get(id string, ownerID int, checkOwner bool) (sessionRecord, error) {
|
|
s.mu.RLock()
|
|
session, ok := s.sessions[id]
|
|
s.mu.RUnlock()
|
|
if !ok || !s.now().Before(session.ExpiresAt) || (checkOwner && session.OwnerID != ownerID) {
|
|
return sessionRecord{}, ErrSessionExpired
|
|
}
|
|
return session, nil
|
|
}
|
|
|
|
func playbackBase(request *http.Request) (*url.URL, error) {
|
|
configured := strings.TrimSpace(os.Getenv("SENSE_MEDIAMTX_WEBRTC_PUBLIC_BASE"))
|
|
if configured != "" {
|
|
return validatePlaybackBase(configured)
|
|
}
|
|
if request == nil || request.Host == "" {
|
|
return nil, errors.New("无法确定浏览器可访问的视频服务地址")
|
|
}
|
|
host := request.Host
|
|
if parsedHost, _, err := net.SplitHostPort(request.Host); err == nil {
|
|
host = parsedHost
|
|
}
|
|
host = strings.Trim(host, "[]")
|
|
if host == "" || (net.ParseIP(host) == nil && host != "localhost" && !validDNSHost.MatchString(host)) {
|
|
return nil, errors.New("无效的请求主机")
|
|
}
|
|
scheme := "http"
|
|
if request.TLS != nil {
|
|
scheme = "https"
|
|
}
|
|
return validatePlaybackBase(fmt.Sprintf("%s://%s", scheme, net.JoinHostPort(host, "8889")))
|
|
}
|
|
|
|
func validatePlaybackBase(value string) (*url.URL, error) {
|
|
parsed, err := url.Parse(strings.TrimRight(value, "/"))
|
|
if err != nil || (parsed.Scheme != "http" && parsed.Scheme != "https") || parsed.Hostname() == "" || parsed.User != nil || parsed.RawQuery != "" || parsed.Fragment != "" || (parsed.Path != "" && parsed.Path != "/") {
|
|
return nil, errors.New("SENSE_MEDIAMTX_WEBRTC_PUBLIC_BASE 配置不安全")
|
|
}
|
|
return parsed, nil
|
|
}
|
|
|
|
func randomID() (string, error) {
|
|
value := make([]byte, 24)
|
|
if _, err := rand.Read(value); err != nil {
|
|
return "", err
|
|
}
|
|
return "view_" + hex.EncodeToString(value), nil
|
|
}
|
|
|
|
func escapeLike(value string) string {
|
|
value = strings.ReplaceAll(value, `\`, `\\`)
|
|
value = strings.ReplaceAll(value, `%`, `\%`)
|
|
return strings.ReplaceAll(value, `_`, `\_`)
|
|
}
|