361 lines
14 KiB
Go
361 lines
14 KiB
Go
package provisioning
|
|
|
|
import (
|
|
"context"
|
|
"errors"
|
|
"fmt"
|
|
"net/url"
|
|
"sort"
|
|
"strings"
|
|
"time"
|
|
|
|
coreService "github.com/go-admin-team/go-admin-core/sdk/service"
|
|
"github.com/google/uuid"
|
|
"github.com/jackc/pgx/v5/pgconn"
|
|
"gorm.io/gorm"
|
|
|
|
"git.ilapage.cn/ila/yovision/Sense/server/app/sense/admission"
|
|
"git.ilapage.cn/ila/yovision/Sense/server/app/sense/credential"
|
|
deviceService "git.ilapage.cn/ila/yovision/Sense/server/app/sense/device/service"
|
|
deviceDTO "git.ilapage.cn/ila/yovision/Sense/server/app/sense/device/service/dto"
|
|
"git.ilapage.cn/ila/yovision/Sense/server/app/sense/quota"
|
|
)
|
|
|
|
var (
|
|
ErrInvalidRequest = errors.New("批量开通请求不符合要求")
|
|
ErrBatchNotFound = errors.New("批量开通批次不存在")
|
|
ErrItemNotFound = errors.New("批量开通条目不存在")
|
|
)
|
|
|
|
type Activator func(context.Context, *Service, *Item, CredentialInput, int) (string, string, string, error)
|
|
|
|
type Service struct {
|
|
coreService.Service
|
|
Quota int
|
|
Activator Activator
|
|
}
|
|
|
|
func (s *Service) quotaLimit() (int, error) {
|
|
if s.Quota > 0 {
|
|
return s.Quota, nil
|
|
}
|
|
return quota.ReadLimit(s.Orm)
|
|
}
|
|
|
|
func (s *Service) CreateBatch(request CreateBatchRequest) (BatchResponse, error) {
|
|
request.IdempotencyKey = strings.TrimSpace(request.IdempotencyKey)
|
|
if request.IdempotencyKey == "" || len(request.IdempotencyKey) > 128 || len(request.Rows) == 0 || len(request.Rows) > 1000 {
|
|
return BatchResponse{}, ErrInvalidRequest
|
|
}
|
|
var existing Batch
|
|
if err := s.Orm.Preload("Items", func(db *gorm.DB) *gorm.DB { return db.Order("line_number") }).First(&existing, "idempotency_key = ?", request.IdempotencyKey).Error; err == nil {
|
|
return batchResponse(existing), nil
|
|
} else if !errors.Is(err, gorm.ErrRecordNotFound) {
|
|
return BatchResponse{}, fmt.Errorf("read provisioning idempotency key: %w", err)
|
|
}
|
|
|
|
var existingDevices int64
|
|
if err := s.Orm.Table("sense_devices").Where("status <> ?", "disabled").Count(&existingDevices).Error; err != nil {
|
|
return BatchResponse{}, fmt.Errorf("count provisioned devices: %w", err)
|
|
}
|
|
quotaLimit, err := s.quotaLimit()
|
|
if err != nil {
|
|
return BatchResponse{}, err
|
|
}
|
|
available := quotaLimit - int(existingDevices)
|
|
if available < 0 {
|
|
available = 0
|
|
}
|
|
batch := Batch{ID: uuid.NewString(), IdempotencyKey: request.IdempotencyKey, Status: BatchReady, QuotaLimit: quotaLimit, ExistingCount: int(existingDevices), TotalCount: len(request.Rows)}
|
|
batch.CreateBy, batch.UpdateBy = request.CreateBy, request.CreateBy
|
|
seenLines := map[int]bool{}
|
|
seenAddresses := map[string]bool{}
|
|
ready := 0
|
|
for _, row := range request.Rows {
|
|
item := Item{ID: uuid.NewString(), BatchID: batch.ID, LineNumber: row.LineNumber, Name: strings.TrimSpace(row.Name), Location: strings.TrimSpace(row.Location), Address: strings.TrimSpace(row.Address), Status: ItemReady}
|
|
item.CreateBy, item.UpdateBy = request.CreateBy, request.CreateBy
|
|
code, detail := validateImportRow(item, seenLines, seenAddresses)
|
|
if code != "" {
|
|
item.Status, item.FailureCode, item.Detail = ItemInvalid, code, detail
|
|
} else if ready >= available {
|
|
item.Status, item.FailureCode, item.Detail = ItemQuotaExceeded, "quota_exceeded", "超出当前可用配额，请调整配额或批次后重试"
|
|
} else {
|
|
ready++
|
|
}
|
|
batch.Items = append(batch.Items, item)
|
|
}
|
|
batch.ReadyCount = ready
|
|
if ready == 0 {
|
|
batch.Status = BatchFailed
|
|
}
|
|
if err := s.Orm.Transaction(func(tx *gorm.DB) error {
|
|
if err := tx.Omit("Items").Create(&batch).Error; err != nil {
|
|
return err
|
|
}
|
|
return tx.Create(&batch.Items).Error
|
|
}); err != nil {
|
|
if isDuplicateKey(err) {
|
|
return s.GetBatchByKey(request.IdempotencyKey)
|
|
}
|
|
return BatchResponse{}, fmt.Errorf("create provisioning batch: %w", err)
|
|
}
|
|
return s.GetBatch(batch.ID)
|
|
}
|
|
|
|
func validateImportRow(item Item, seenLines map[int]bool, seenAddresses map[string]bool) (string, string) {
|
|
if item.LineNumber < 1 || seenLines[item.LineNumber] {
|
|
return "invalid_line_number", "行号必须为正整数且批次内唯一"
|
|
}
|
|
seenLines[item.LineNumber] = true
|
|
if item.Name == "" || len([]rune(item.Name)) > 128 || len([]rune(item.Location)) > 255 {
|
|
return "invalid_device", "设备名称不能为空，名称或安装位置长度不能超过限制"
|
|
}
|
|
parsed, err := url.Parse(item.Address)
|
|
if err != nil || (parsed.Scheme != "http" && parsed.Scheme != "https") || parsed.Hostname() == "" || parsed.User != nil || parsed.RawQuery != "" || parsed.Fragment != "" {
|
|
return "invalid_address", "设备地址必须是无账号、查询参数和片段的 HTTP(S) 地址"
|
|
}
|
|
key := strings.ToLower(parsed.String())
|
|
if seenAddresses[key] {
|
|
return "duplicate_address", "同一批次中设备地址不能重复"
|
|
}
|
|
seenAddresses[key] = true
|
|
return "", ""
|
|
}
|
|
|
|
func (s *Service) ListBatches(request *BatchPageRequest) ([]BatchResponse, int64, error) {
|
|
query := s.Orm.Model(&Batch{})
|
|
if request.Status != "" {
|
|
query = query.Where("status = ?", request.Status)
|
|
}
|
|
var count int64
|
|
if err := query.Count(&count).Error; err != nil {
|
|
return nil, 0, err
|
|
}
|
|
pageSize := request.GetPageSize()
|
|
if pageSize > 100 {
|
|
pageSize = 100
|
|
}
|
|
var batches []Batch
|
|
if err := query.Order("created_at DESC").Limit(pageSize).Offset((request.GetPageIndex() - 1) * pageSize).Find(&batches).Error; err != nil {
|
|
return nil, 0, err
|
|
}
|
|
result := make([]BatchResponse, 0, len(batches))
|
|
for _, batch := range batches {
|
|
result = append(result, batchResponse(batch))
|
|
}
|
|
return result, count, nil
|
|
}
|
|
|
|
func (s *Service) GetBatch(id string) (BatchResponse, error) {
|
|
var batch Batch
|
|
if err := s.Orm.Preload("Items", func(db *gorm.DB) *gorm.DB { return db.Order("line_number") }).First(&batch, "id = ?", id).Error; err != nil {
|
|
if errors.Is(err, gorm.ErrRecordNotFound) {
|
|
return BatchResponse{}, ErrBatchNotFound
|
|
}
|
|
return BatchResponse{}, err
|
|
}
|
|
return batchResponse(batch), nil
|
|
}
|
|
|
|
func (s *Service) GetBatchByKey(key string) (BatchResponse, error) {
|
|
var batch Batch
|
|
if err := s.Orm.Preload("Items", func(db *gorm.DB) *gorm.DB { return db.Order("line_number") }).First(&batch, "idempotency_key = ?", key).Error; err != nil {
|
|
return BatchResponse{}, err
|
|
}
|
|
return batchResponse(batch), nil
|
|
}
|
|
|
|
func (s *Service) Execute(ctx context.Context, batchID string, itemID string, retryFailed bool, request ExecuteRequest) (BatchResponse, error) {
|
|
credentialByItem := make(map[string]CredentialInput, len(request.Credentials))
|
|
for _, input := range request.Credentials {
|
|
if input.ItemID == "" || credentialByItem[input.ItemID].ItemID != "" {
|
|
return BatchResponse{}, ErrInvalidRequest
|
|
}
|
|
credentialByItem[input.ItemID] = input
|
|
}
|
|
var batch Batch
|
|
if err := s.Orm.First(&batch, "id = ?", batchID).Error; err != nil {
|
|
if errors.Is(err, gorm.ErrRecordNotFound) {
|
|
return BatchResponse{}, ErrBatchNotFound
|
|
}
|
|
return BatchResponse{}, err
|
|
}
|
|
query := s.Orm.Where("batch_id = ?", batchID)
|
|
if itemID != "" {
|
|
query = query.Where("id = ?", itemID)
|
|
}
|
|
if retryFailed {
|
|
query = query.Where("status = ?", ItemFailed)
|
|
} else {
|
|
query = query.Where("status = ?", ItemReady)
|
|
}
|
|
var items []Item
|
|
if err := query.Order("line_number").Find(&items).Error; err != nil {
|
|
return BatchResponse{}, err
|
|
}
|
|
if itemID != "" && len(items) == 0 {
|
|
return BatchResponse{}, ErrItemNotFound
|
|
}
|
|
if len(items) == 0 {
|
|
return s.GetBatch(batchID)
|
|
}
|
|
if err := s.Orm.Model(&Batch{}).Where("id = ?", batchID).Updates(map[string]any{"status": BatchRunning, "update_by": request.UpdateBy}).Error; err != nil {
|
|
return BatchResponse{}, fmt.Errorf("mark provisioning batch running: %w", err)
|
|
}
|
|
activate := s.Activator
|
|
if activate == nil {
|
|
activate = defaultActivate
|
|
}
|
|
for index := range items {
|
|
item := &items[index]
|
|
now := time.Now().UTC()
|
|
expectedStatus := ItemReady
|
|
if retryFailed {
|
|
expectedStatus = ItemFailed
|
|
}
|
|
claim := s.Orm.Model(&Item{}).Where("id = ? AND status = ?", item.ID, expectedStatus).Updates(map[string]any{"status": ItemRunning, "attempts": gorm.Expr("attempts + 1"), "last_tried_at": now, "update_by": request.UpdateBy})
|
|
if claim.Error != nil {
|
|
return BatchResponse{}, fmt.Errorf("mark provisioning item running: %w", claim.Error)
|
|
}
|
|
if claim.RowsAffected == 0 {
|
|
continue
|
|
}
|
|
input, ok := credentialByItem[item.ID]
|
|
if !ok || strings.TrimSpace(input.ONVIFUsername) == "" || input.ONVIFPassword == "" {
|
|
if err := s.failItem(item.ID, request.UpdateBy, "credentials_required", "请安全填写该设备的账号密码后重试"); err != nil {
|
|
return BatchResponse{}, err
|
|
}
|
|
continue
|
|
}
|
|
deviceID, status, detail, err := activate(ctx, s, item, input, request.UpdateBy)
|
|
if deviceID != "" && deviceID != item.DeviceID {
|
|
item.DeviceID = deviceID
|
|
if err := s.Orm.Model(&Item{}).Where("id = ?", item.ID).Update("device_id", deviceID).Error; err != nil {
|
|
return BatchResponse{}, fmt.Errorf("link provisioning item to device: %w", err)
|
|
}
|
|
}
|
|
if err != nil {
|
|
code, safeDetail := safeFailure(err)
|
|
if updateErr := s.failItem(item.ID, request.UpdateBy, code, safeDetail); updateErr != nil {
|
|
return BatchResponse{}, updateErr
|
|
}
|
|
continue
|
|
}
|
|
if status != "ready" {
|
|
if detail == "" {
|
|
detail = "设备或视频流尚未通过验证"
|
|
}
|
|
if err := s.failItem(item.ID, request.UpdateBy, status, detail); err != nil {
|
|
return BatchResponse{}, err
|
|
}
|
|
continue
|
|
}
|
|
if err := s.Orm.Model(&Item{}).Where("id = ?", item.ID).Updates(map[string]any{"status": ItemSucceeded, "failure_code": "", "detail": "设备已开通并通过视频验证", "update_by": request.UpdateBy}).Error; err != nil {
|
|
return BatchResponse{}, fmt.Errorf("mark provisioning item succeeded: %w", err)
|
|
}
|
|
}
|
|
if err := s.refreshBatch(batchID, request.UpdateBy); err != nil {
|
|
return BatchResponse{}, err
|
|
}
|
|
return s.GetBatch(batchID)
|
|
}
|
|
|
|
func isDuplicateKey(err error) bool {
|
|
if errors.Is(err, gorm.ErrDuplicatedKey) {
|
|
return true
|
|
}
|
|
var postgresError *pgconn.PgError
|
|
return errors.As(err, &postgresError) && postgresError.Code == "23505"
|
|
}
|
|
|
|
func (s *Service) failItem(id string, updateBy int, code, detail string) error {
|
|
if err := s.Orm.Model(&Item{}).Where("id = ?", id).Updates(map[string]any{"status": ItemFailed, "failure_code": code, "detail": detail, "update_by": updateBy}).Error; err != nil {
|
|
return fmt.Errorf("mark provisioning item failed: %w", err)
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func (s *Service) refreshBatch(batchID string, updateBy int) error {
|
|
var items []Item
|
|
if err := s.Orm.Where("batch_id = ?", batchID).Find(&items).Error; err != nil {
|
|
return err
|
|
}
|
|
ready, success, failure := 0, 0, 0
|
|
for _, item := range items {
|
|
switch item.Status {
|
|
case ItemReady, ItemRunning:
|
|
ready++
|
|
case ItemSucceeded:
|
|
success++
|
|
case ItemFailed:
|
|
failure++
|
|
}
|
|
}
|
|
status := BatchFailed
|
|
if success == len(items) {
|
|
status = BatchSucceeded
|
|
} else if success > 0 {
|
|
status = BatchPartial
|
|
} else if ready > 0 {
|
|
status = BatchReady
|
|
}
|
|
return s.Orm.Model(&Batch{}).Where("id = ?", batchID).Updates(map[string]any{"status": status, "ready_count": ready, "success_count": success, "failure_count": failure, "update_by": updateBy}).Error
|
|
}
|
|
|
|
func defaultActivate(ctx context.Context, service *Service, item *Item, input CredentialInput, updateBy int) (string, string, string, error) {
|
|
device := deviceService.Device{Service: service.Service}
|
|
deviceID := item.DeviceID
|
|
var response deviceDTO.DeviceResponse
|
|
if deviceID == "" {
|
|
if err := device.Insert(&deviceDTO.CreateReq{Name: item.Name, Location: item.Location, Modality: "video", Capabilities: []string{"video"}, CreateBy: updateBy}, &response); err != nil {
|
|
return "", "", "", err
|
|
}
|
|
deviceID = response.ID
|
|
if err := service.Orm.Model(&Item{}).Where("id = ?", item.ID).Update("device_id", deviceID).Error; err != nil {
|
|
return deviceID, "", "", err
|
|
}
|
|
} else if err := device.Get(deviceID, &response); err != nil {
|
|
return deviceID, "", "", err
|
|
}
|
|
if err := device.UpdateCredentials(&deviceDTO.CredentialUpdateReq{ID: deviceID, ONVIFUsername: input.ONVIFUsername, ONVIFPassword: input.ONVIFPassword, RTSPSameAsONVIF: input.RTSPSameAsONVIF, RTSPUsername: input.RTSPUsername, RTSPPassword: input.RTSPPassword, Version: response.Version, UpdateBy: updateBy}, &response); err != nil {
|
|
return deviceID, "", "", err
|
|
}
|
|
probeService, err := admission.NewRuntime(service.Service)
|
|
if err != nil {
|
|
return deviceID, "", "", err
|
|
}
|
|
result, err := probeService.Probe(ctx, admission.ProbeRequest{DeviceID: deviceID, Address: item.Address, Version: response.Version, UpdateBy: updateBy})
|
|
if err != nil {
|
|
return deviceID, "", "", err
|
|
}
|
|
return deviceID, result.Status, result.Detail, nil
|
|
}
|
|
|
|
func safeFailure(err error) (string, string) {
|
|
switch {
|
|
case errors.Is(err, credential.ErrKeyUnavailable):
|
|
return "credential_key_unavailable", "摄像头凭据安全配置不可用"
|
|
case errors.Is(err, credential.ErrCredentialNotConfigured):
|
|
return "credentials_required", "请安全填写摄像头账号密码后重试"
|
|
case errors.Is(err, deviceService.ErrInvalidDevice), errors.Is(err, admission.ErrInvalid):
|
|
return "invalid_device", "设备信息或凭据不符合要求"
|
|
case errors.Is(err, deviceService.ErrVersionConflict), errors.Is(err, admission.ErrConflict):
|
|
return "version_conflict", "设备已被其他操作更新，请重试"
|
|
case errors.Is(err, quota.ErrExceeded):
|
|
return "quota_exceeded", "当前配额已满，请调整配额或停用其他设备后重试"
|
|
case errors.Is(err, quota.ErrUnavailable):
|
|
return "quota_unavailable", "配额配置不可读取，已拒绝新增或启用设备"
|
|
default:
|
|
return "activation_failed", "设备开通失败，请检查网络、地址和凭据后重试"
|
|
}
|
|
}
|
|
|
|
func batchResponse(batch Batch) BatchResponse {
|
|
response := BatchResponse{ID: batch.ID, IdempotencyKey: batch.IdempotencyKey, Status: batch.Status, QuotaLimit: batch.QuotaLimit, ExistingCount: batch.ExistingCount, TotalCount: batch.TotalCount, ReadyCount: batch.ReadyCount, SuccessCount: batch.SuccessCount, FailureCount: batch.FailureCount, CreatedAt: batch.CreatedAt, UpdatedAt: batch.UpdatedAt, Items: make([]ItemResponse, 0, len(batch.Items))}
|
|
sort.Slice(batch.Items, func(i, j int) bool { return batch.Items[i].LineNumber < batch.Items[j].LineNumber })
|
|
for _, item := range batch.Items {
|
|
response.Items = append(response.Items, ItemResponse{ID: item.ID, LineNumber: item.LineNumber, Name: item.Name, Location: item.Location, Address: item.Address, Status: item.Status, FailureCode: item.FailureCode, Detail: item.Detail, DeviceID: item.DeviceID, Attempts: item.Attempts, LastTriedAt: item.LastTriedAt})
|
|
}
|
|
return response
|
|
}
|