yovision/Sense/server/app/sense/device/service/device.go

package service

import (
	"encoding/json"
	"errors"
	"fmt"
	"strings"
	"time"

	coreService "github.com/go-admin-team/go-admin-core/sdk/service"
	"github.com/google/uuid"
	"gorm.io/gorm"
	"gorm.io/gorm/clause"

	"git.ilapage.cn/ila/yovision/Sense/server/app/sense/credential"
	"git.ilapage.cn/ila/yovision/Sense/server/app/sense/device/models"
	"git.ilapage.cn/ila/yovision/Sense/server/app/sense/device/service/dto"
	"git.ilapage.cn/ila/yovision/Sense/server/app/sense/quota"
)

var (
	ErrDeviceNotFound  = errors.New("设备不存在")
	ErrVersionConflict = errors.New("设备已被其他用户更新，请刷新后重试")
	ErrInvalidDevice   = errors.New("设备信息不符合要求")
)

var supportedValues = map[string]struct{}{
	"video": {}, "radar": {}, "contact": {}, "button": {}, "wearable": {}, "other": {},
}

type Device struct {
	coreService.Service
	VaultFactory func() (*credential.Vault, error)
}

type credentialPresence struct {
	DeviceID string
	Purpose  string
}

func (e *Device) GetPage(req *dto.PageReq, list *[]dto.DeviceResponse, count *int64) error {
	query := e.Orm.Model(&models.Device{})
	if keyword := strings.TrimSpace(req.Keyword); keyword != "" {
		pattern := "%" + strings.ToLower(keyword) + "%"
		query = query.Where("LOWER(name) LIKE ? OR LOWER(location) LIKE ?", pattern, pattern)
	}
	if req.Modality != "" {
		query = query.Where("modality = ?", req.Modality)
	}
	if req.Status != "" {
		query = query.Where("status = ?", req.Status)
	}
	if err := query.Count(count).Error; err != nil {
		return fmt.Errorf("count devices: %w", err)
	}
	pageSize := req.GetPageSize()
	if pageSize > 100 {
		pageSize = 100
	}
	req.PageSize = pageSize
	var devices []models.Device
	if err := query.Order("created_at DESC").Limit(pageSize).Offset((req.GetPageIndex() - 1) * pageSize).Find(&devices).Error; err != nil {
		return fmt.Errorf("list devices: %w", err)
	}
	responses, err := e.responses(devices)
	if err != nil {
		return err
	}
	*list = responses
	return nil
}

func (e *Device) Get(id string, response *dto.DeviceResponse) error {
	var model models.Device
	if err := e.Orm.First(&model, "id = ?", id).Error; err != nil {
		if errors.Is(err, gorm.ErrRecordNotFound) {
			return ErrDeviceNotFound
		}
		return fmt.Errorf("get device: %w", err)
	}
	responses, err := e.responses([]models.Device{model})
	if err != nil {
		return err
	}
	*response = responses[0]
	return nil
}

func (e *Device) Insert(req *dto.CreateReq, response *dto.DeviceResponse) error {
	name, location, modality, capabilities, err := normalizeDevice(req.Name, req.Location, req.Modality, req.Capabilities)
	if err != nil {
		return err
	}
	encoded, _ := json.Marshal(capabilities)
	adapterStatus := models.AdapterNotReady
	if modality == models.ModalityVideo {
		adapterStatus = models.AdapterReady
	}
	model := models.Device{
		ID:               uuid.NewString(),
		Name:             name,
		Location:         location,
		Modality:         modality,
		CapabilitiesJSON: string(encoded),
		Status:           models.StatusPending,
		AdapterStatus:    adapterStatus,
		RTSPSameAsONVIF:  true,
		Version:          1,
	}
	model.CreateBy = req.CreateBy
	model.UpdateBy = req.CreateBy
	if err = quota.WithAvailableSlot(e.Orm, func(tx *gorm.DB) error {
		return tx.Create(&model).Error
	}); err != nil {
		if errors.Is(err, quota.ErrUnavailable) || errors.Is(err, quota.ErrExceeded) {
			return err
		}
		return fmt.Errorf("create device: %w", err)
	}
	return e.Get(model.ID, response)
}

func (e *Device) Update(req *dto.UpdateReq, response *dto.DeviceResponse) error {
	name := strings.TrimSpace(req.Name)
	if name == "" || len([]rune(name)) > 128 || len([]rune(req.Location)) > 255 || req.Version < 1 {
		return ErrInvalidDevice
	}
	capabilities, err := normalizeCapabilities(req.Capabilities)
	if err != nil {
		return err
	}
	encoded, _ := json.Marshal(capabilities)
	updates := map[string]any{
		"name": name, "location": strings.TrimSpace(req.Location), "capabilities": string(encoded),
		"version": req.Version + 1, "update_by": req.UpdateBy, "updated_at": time.Now().UTC(),
	}
	result := e.Orm.Model(&models.Device{}).Where("id = ? AND version = ?", req.ID, req.Version).Updates(updates)
	if result.Error != nil {
		return fmt.Errorf("update device: %w", result.Error)
	}
	if result.RowsAffected == 0 {
		return e.notFoundOrConflict(req.ID)
	}
	return e.Get(req.ID, response)
}

func (e *Device) Disable(req *dto.DisableReq, response *dto.DeviceResponse) error {
	if req.Version < 1 {
		return ErrInvalidDevice
	}
	result := e.Orm.Model(&models.Device{}).Where("id = ? AND version = ?", req.ID, req.Version).Updates(map[string]any{
		"status": models.StatusDisabled, "version": req.Version + 1,
		"update_by": req.UpdateBy, "updated_at": time.Now().UTC(),
	})
	if result.Error != nil {
		return fmt.Errorf("disable device: %w", result.Error)
	}
	if result.RowsAffected == 0 {
		return e.notFoundOrConflict(req.ID)
	}
	return e.Get(req.ID, response)
}

func (e *Device) Enable(req *dto.EnableReq, response *dto.DeviceResponse) error {
	if req.Version < 1 {
		return ErrInvalidDevice
	}
	err := quota.WithAvailableSlot(e.Orm, func(tx *gorm.DB) error {
		result := tx.Model(&models.Device{}).
			Where("id = ? AND version = ? AND status = ?", req.ID, req.Version, models.StatusDisabled).
			Updates(map[string]any{
				"status":  models.StatusPending,
				"version": req.Version + 1, "update_by": req.UpdateBy, "updated_at": time.Now().UTC(),
			})
		if result.Error != nil {
			return result.Error
		}
		if result.RowsAffected == 0 {
			return e.notFoundOrConflictWith(tx, req.ID)
		}
		return nil
	})
	if err != nil {
		if errors.Is(err, quota.ErrUnavailable) || errors.Is(err, quota.ErrExceeded) ||
			errors.Is(err, ErrDeviceNotFound) || errors.Is(err, ErrVersionConflict) {
			return err
		}
		return fmt.Errorf("enable device: %w", err)
	}
	return e.Get(req.ID, response)
}

func (e *Device) UpdateCredentials(req *dto.CredentialUpdateReq, response *dto.DeviceResponse) error {
	if req.Version < 1 || strings.TrimSpace(req.ONVIFUsername) == "" || req.ONVIFPassword == "" {
		return ErrInvalidDevice
	}
	var target models.Device
	if err := e.Orm.Select("id", "modality").First(&target, "id = ?", req.ID).Error; err != nil {
		if errors.Is(err, gorm.ErrRecordNotFound) {
			return ErrDeviceNotFound
		}
		return fmt.Errorf("read credential target: %w", err)
	}
	if target.Modality != models.ModalityVideo {
		return ErrInvalidDevice
	}
	if len(req.ONVIFUsername) > 255 || len(req.ONVIFPassword) > 1024 {
		return ErrInvalidDevice
	}
	rtspUsername, rtspPassword := req.RTSPUsername, req.RTSPPassword
	if req.RTSPSameAsONVIF {
		rtspUsername, rtspPassword = req.ONVIFUsername, req.ONVIFPassword
	} else if strings.TrimSpace(rtspUsername) == "" || rtspPassword == "" || len(rtspUsername) > 255 || len(rtspPassword) > 1024 {
		return ErrInvalidDevice
	}
	factory := e.VaultFactory
	if factory == nil {
		factory = credential.NewVaultFromEnvironment
	}
	vault, err := factory()
	if err != nil {
		return err
	}
	onvifCiphertext, err := vault.Encrypt(req.ID, credential.PurposeONVIF, strings.TrimSpace(req.ONVIFUsername), req.ONVIFPassword)
	if err != nil {
		return err
	}
	rtspCiphertext, err := vault.Encrypt(req.ID, credential.PurposeRTSP, strings.TrimSpace(rtspUsername), rtspPassword)
	if err != nil {
		return err
	}
	now := time.Now().UTC()
	err = e.Orm.Transaction(func(tx *gorm.DB) error {
		result := tx.Model(&models.Device{}).Where("id = ? AND version = ?", req.ID, req.Version).Updates(map[string]any{
			"rtsp_same_as_onvif":    req.RTSPSameAsONVIF,
			"credential_updated_at": now, "retry_requested_at": now,
			"version": req.Version + 1, "update_by": req.UpdateBy, "updated_at": now,
		})
		if result.Error != nil {
			return result.Error
		}
		if result.RowsAffected == 0 {
			return e.notFoundOrConflictWith(tx, req.ID)
		}
		rows := []credential.DeviceCredential{
			{DeviceID: req.ID, Purpose: credential.PurposeONVIF, Ciphertext: onvifCiphertext, KeyVersion: credential.Version()},
			{DeviceID: req.ID, Purpose: credential.PurposeRTSP, Ciphertext: rtspCiphertext, KeyVersion: credential.Version()},
		}
		return tx.Clauses(clause.OnConflict{
			Columns:   []clause.Column{{Name: "device_id"}, {Name: "purpose"}},
			DoUpdates: clause.AssignmentColumns([]string{"ciphertext", "key_version", "updated_at"}),
		}).Create(&rows).Error
	})
	if err != nil {
		if errors.Is(err, ErrDeviceNotFound) || errors.Is(err, ErrVersionConflict) {
			return err
		}
		return fmt.Errorf("update device credential: %w", err)
	}
	return e.Get(req.ID, response)
}

func (e *Device) responses(devices []models.Device) ([]dto.DeviceResponse, error) {
	result := make([]dto.DeviceResponse, 0, len(devices))
	if len(devices) == 0 {
		return result, nil
	}
	ids := make([]string, 0, len(devices))
	for _, device := range devices {
		ids = append(ids, device.ID)
	}
	var presence []credentialPresence
	if err := e.Orm.Model(&credential.DeviceCredential{}).Select("device_id", "purpose").Where("device_id IN ?", ids).Find(&presence).Error; err != nil {
		return nil, fmt.Errorf("read credential status: %w", err)
	}
	configured := make(map[string]map[string]bool, len(devices))
	for _, row := range presence {
		if configured[row.DeviceID] == nil {
			configured[row.DeviceID] = map[string]bool{}
		}
		configured[row.DeviceID][row.Purpose] = true
	}
	for _, device := range devices {
		var capabilities []string
		if err := json.Unmarshal([]byte(device.CapabilitiesJSON), &capabilities); err != nil {
			return nil, fmt.Errorf("decode device capabilities: %w", err)
		}
		result = append(result, dto.DeviceResponse{
			ID: device.ID, Name: device.Name, Location: device.Location, Modality: device.Modality,
			Capabilities: capabilities, Status: device.Status, AdapterStatus: device.AdapterStatus,
			ONVIFCredentialConfigured: configured[device.ID][credential.PurposeONVIF],
			RTSPCredentialConfigured:  configured[device.ID][credential.PurposeRTSP],
			RTSPCredentialSameAsONVIF: device.RTSPSameAsONVIF,
			RetryPending:              device.RetryRequestedAt != nil, Version: device.Version,
			CreatedAt: device.CreatedAt, UpdatedAt: device.UpdatedAt,
		})
	}
	return result, nil
}

func (e *Device) notFoundOrConflict(id string) error { return e.notFoundOrConflictWith(e.Orm, id) }

func (e *Device) notFoundOrConflictWith(db *gorm.DB, id string) error {
	var count int64
	if err := db.Model(&models.Device{}).Where("id = ?", id).Count(&count).Error; err != nil {
		return err
	}
	if count == 0 {
		return ErrDeviceNotFound
	}
	return ErrVersionConflict
}

func normalizeDevice(name, location, modality string, capabilities []string) (string, string, string, []string, error) {
	name, location, modality = strings.TrimSpace(name), strings.TrimSpace(location), strings.TrimSpace(modality)
	if modality == "" {
		modality = models.ModalityVideo
	}
	if name == "" || len([]rune(name)) > 128 || len([]rune(location)) > 255 {
		return "", "", "", nil, ErrInvalidDevice
	}
	if _, ok := supportedValues[modality]; !ok {
		return "", "", "", nil, ErrInvalidDevice
	}
	if len(capabilities) == 0 {
		capabilities = []string{modality}
	}
	normalized, err := normalizeCapabilities(capabilities)
	return name, location, modality, normalized, err
}

func normalizeCapabilities(values []string) ([]string, error) {
	if len(values) == 0 || len(values) > 16 {
		return nil, ErrInvalidDevice
	}
	seen := map[string]bool{}
	result := make([]string, 0, len(values))
	for _, value := range values {
		value = strings.TrimSpace(value)
		if _, ok := supportedValues[value]; !ok {
			return nil, ErrInvalidDevice
		}
		if !seen[value] {
			seen[value] = true
			result = append(result, value)
		}
	}
	return result, nil
}