package opsdevice import ( "context" "crypto/rand" "crypto/sha256" "encoding/base32" "encoding/hex" "encoding/json" "errors" "fmt" "net/url" "strings" "time" "data-terminal-cloud/internal/audit" "data-terminal-cloud/internal/deviceactivation" "data-terminal-cloud/internal/deviceauth" "data-terminal-cloud/internal/devicebinding" "data-terminal-cloud/internal/identity" "data-terminal-cloud/internal/onboarding" "github.com/google/uuid" "github.com/jackc/pgx/v5" "github.com/jackc/pgx/v5/pgxpool" ) var ( ErrAccessDenied = errors.New("operator access denied") ErrInvalid = errors.New("invalid device management input") ErrNotFound = errors.New("device not found") ErrVersionConflict = errors.New("device version conflict") ErrClaimUnavailable = errors.New("device claim is unavailable") ErrActivationConflict = errors.New("device activation transition conflict") ErrActivationNotReady = errors.New("device activation prerequisites are not ready") ErrProvisioningUnavailable = errors.New("device provisioning is unavailable") ) type Service struct { pool *pgxpool.Pool credential *deviceauth.DatabaseSecretResolver audit *audit.Repository onboardingMasterKey string deviceOrigin string } type Options struct { OnboardingMasterKey string DeviceOrigin string } type Device struct { ID string `json:"id"` TenantID string `json:"tenant_id"` FactoryID string `json:"factory_id"` EnrollmentMode string `json:"enrollment_mode"` Bound bool `json:"bound"` Model string `json:"model"` HardwareRevision *string `json:"hardware_revision,omitempty"` Status string `json:"status"` ActivationState string `json:"activation_state"` ActivationRevision int64 `json:"activation_revision"` ActivationReason *string `json:"activation_reason,omitempty"` ActivationUpdatedAt time.Time `json:"activation_updated_at"` MinimumSecurityVersion *string `json:"minimum_security_version,omitempty"` MinimumFirmwareVersion *string `json:"minimum_firmware_version,omitempty"` HouseholdID *string `json:"household_id,omitempty"` HouseholdName *string `json:"household_name,omitempty"` SubjectID *string `json:"subject_id,omitempty"` SubjectName *string `json:"subject_name,omitempty"` CredentialStatus string `json:"credential_status"` KeyVersion int64 `json:"key_version"` FirmwareVersion *string `json:"firmware_version,omitempty"` LastReportedAt *time.Time `json:"last_reported_at,omitempty"` RegisteredAt time.Time `json:"registered_at"` UpdatedAt time.Time `json:"updated_at"` Version int64 `json:"version"` VoiceRuntime string `json:"voice_runtime"` HardwareAgentBotID string `json:"hardware_agent_bot_id,omitempty"` HardwareAgentTransport string `json:"hardware_agent_transport,omitempty"` HardwareAgentMinimumSDKVersion string `json:"hardware_agent_minimum_sdk_version,omitempty"` HardwareAgentStatus string `json:"hardware_agent_status,omitempty"` } type AssignmentOption struct { HouseholdID string `json:"household_id"` HouseholdName string `json:"household_name"` SubjectID string `json:"subject_id"` SubjectName string `json:"subject_name"` } type CreateInput struct{ Model string } type ActivationTransitionInput struct { TargetState string Reason string ExpectedRevision int64 } type ActivationTransition struct { ID string `json:"id"` FromState *string `json:"from_state,omitempty"` ToState string `json:"to_state"` Reason string `json:"reason"` Source string `json:"source"` ActorType string `json:"actor_type"` ActorPublicID *string `json:"actor_public_id,omitempty"` RequestID *string `json:"request_id,omitempty"` ActivationRevision int64 `json:"activation_revision"` CreatedAt time.Time `json:"created_at"` } type UpdateInput struct { Model *string HardwareRevision *string MinimumSecurityVersion *string MinimumFirmwareVersion *string Status *string HouseholdID *string SubjectID *string VoiceRuntime *string HardwareAgentBotID *string HardwareAgentTransport *string HardwareAgentMinimumSDKVersion *string } type CredentialIssue struct { FactoryID string `json:"factory_id"` Model string `json:"model"` Origin string `json:"origin"` DeviceKeyHex string `json:"device_key_hex"` KeyVersion int64 `json:"key_version"` } type CreateResult struct { Device Device `json:"device"` Credential CredentialIssue `json:"credential"` } type DeviceClaimIssue struct { DeviceID string `json:"device_id"` ClaimCode string `json:"claim_code"` IssuedAt time.Time `json:"issued_at"` ExpiresAt time.Time `json:"expires_at"` } type operatorAccess struct { tenantID, tenantPublicID, userID, userPublicID, role string } type queryRow interface { QueryRow(context.Context, string, ...any) pgx.Row } func NewService(pool *pgxpool.Pool, credential *deviceauth.DatabaseSecretResolver, auditRepository *audit.Repository, option ...Options) (*Service, error) { if pool == nil || credential == nil || auditRepository == nil { return nil, errors.New("operations device service dependencies are required") } if len(option) > 1 { return nil, errors.New("only one operations device service option is allowed") } var configured Options if len(option) == 1 { configured = option[0] } origin := strings.TrimSpace(configured.DeviceOrigin) if origin != "" { parsed, err := url.Parse(origin) if err != nil || parsed.Host == "" || (parsed.Scheme != "https" && parsed.Scheme != "http") { return nil, errors.New("device origin must be an absolute HTTP URL") } origin = parsed.Scheme + "://" + parsed.Host } return &Service{pool: pool, credential: credential, audit: auditRepository, onboardingMasterKey: strings.TrimSpace(configured.OnboardingMasterKey), deviceOrigin: origin}, nil } func (s *Service) List(ctx context.Context, principal identity.Principal, tenantPublicID string) ([]Device, error) { access, err := requireAccess(ctx, s.pool, principal, tenantPublicID, false) if err != nil { return nil, err } rows, err := s.pool.Query(ctx, deviceSelectSQL+` WHERE d.tenant_id=$1::uuid ORDER BY d.registered_at DESC,d.public_id`, access.tenantID) if err != nil { return nil, fmt.Errorf("list operations devices: %w", err) } defer rows.Close() items := make([]Device, 0) for rows.Next() { item, scanErr := scanDevice(rows) if scanErr != nil { return nil, scanErr } items = append(items, item) } return items, rows.Err() } func (s *Service) Options(ctx context.Context, principal identity.Principal, tenantPublicID string) ([]AssignmentOption, error) { access, err := requireAccess(ctx, s.pool, principal, tenantPublicID, false) if err != nil { return nil, err } rows, err := s.pool.Query(ctx, ` SELECT h.public_id::text,h.name,sp.public_id::text,sp.preferred_name FROM household h JOIN subject_profile sp ON sp.household_id=h.id AND sp.tenant_id=h.tenant_id WHERE h.tenant_id=$1::uuid AND h.status='active' AND sp.status IN ('active','suspended') ORDER BY h.name,sp.preferred_name`, access.tenantID) if err != nil { return nil, fmt.Errorf("list device assignment options: %w", err) } defer rows.Close() items := make([]AssignmentOption, 0) for rows.Next() { var item AssignmentOption if err := rows.Scan(&item.HouseholdID, &item.HouseholdName, &item.SubjectID, &item.SubjectName); err != nil { return nil, fmt.Errorf("scan device assignment option: %w", err) } items = append(items, item) } return items, rows.Err() } func (s *Service) ListActivationTransitions(ctx context.Context, principal identity.Principal, tenantPublicID, devicePublicID string) ([]ActivationTransition, error) { access, err := requireAccess(ctx, s.pool, principal, tenantPublicID, false) if err != nil { return nil, err } rows, err := s.pool.Query(ctx, ` SELECT at.public_id::text,at.from_state,at.to_state,at.reason,at.source,at.actor_type,at.actor_public_id::text, at.request_id,at.activation_revision,at.created_at FROM device_activation_transition at JOIN device d ON d.id=at.device_id AND d.tenant_id=at.tenant_id WHERE at.tenant_id=$1::uuid AND d.public_id=$2::uuid ORDER BY at.activation_revision DESC`, access.tenantID, devicePublicID) if err != nil { return nil, fmt.Errorf("list device activation transitions: %w", err) } defer rows.Close() items := make([]ActivationTransition, 0) for rows.Next() { var item ActivationTransition if err := rows.Scan(&item.ID, &item.FromState, &item.ToState, &item.Reason, &item.Source, &item.ActorType, &item.ActorPublicID, &item.RequestID, &item.ActivationRevision, &item.CreatedAt); err != nil { return nil, fmt.Errorf("scan device activation transition: %w", err) } items = append(items, item) } if err := rows.Err(); err != nil { return nil, err } if len(items) == 0 { var exists bool if err := s.pool.QueryRow(ctx, `SELECT true FROM device WHERE tenant_id=$1::uuid AND public_id=$2::uuid`, access.tenantID, devicePublicID).Scan(&exists); errors.Is(err, pgx.ErrNoRows) { return nil, ErrNotFound } else if err != nil { return nil, fmt.Errorf("resolve activation transition device: %w", err) } } return items, nil } func (s *Service) Create(ctx context.Context, principal identity.Principal, tenantPublicID, requestID string, input CreateInput) (CreateResult, error) { input.Model = strings.TrimSpace(input.Model) if input.Model == "" || len([]rune(input.Model)) > 120 { return CreateResult{}, ErrInvalid } if s.onboardingMasterKey == "" || s.deviceOrigin == "" { return CreateResult{}, ErrProvisioningUnavailable } secretBytes := make([]byte, 32) if _, err := rand.Read(secretBytes); err != nil { return CreateResult{}, fmt.Errorf("generate device secret: %w", err) } factoryBytes := make([]byte, 6) if _, err := rand.Read(factoryBytes); err != nil { return CreateResult{}, fmt.Errorf("generate factory device ID: %w", err) } secret := hex.EncodeToString(secretBytes) factoryMAC := strings.ToUpper(hex.EncodeToString(factoryBytes)) factoryID := "KP01-" + factoryMAC serialDigest := sha256.Sum256([]byte(factoryID)) deviceID := uuid.NewString() publicID := uuid.NewString() credentialRef := "onboarding://" + factoryID const keyVersion int64 = 1 ciphertext, err := onboarding.SealDeviceKey(s.onboardingMasterKey, deviceID, keyVersion, secretBytes) if err != nil { return CreateResult{}, fmt.Errorf("seal onboarding device key: %w", err) } tx, err := s.pool.Begin(ctx) if err != nil { return CreateResult{}, fmt.Errorf("begin device registration: %w", err) } defer tx.Rollback(ctx) access, err := requireAccess(ctx, tx, principal, tenantPublicID, true) if err != nil { return CreateResult{}, err } _, err = tx.Exec(ctx, ` INSERT INTO device(id,public_id,tenant_id,serial_digest,model,status,credential_ref,activation_state,activation_reason,enrollment_mode,factory_device_id,factory_mac,onboarding_released_at) VALUES($1::uuid,$2::uuid,$3::uuid,$4,$5,'registered',$6,'CLAIMABLE','ops-device-create','physical_confirm',$7,$8,now())`, deviceID, publicID, access.tenantID, hex.EncodeToString(serialDigest[:]), input.Model, credentialRef, factoryID, factoryMAC) if err != nil { return CreateResult{}, fmt.Errorf("insert device: %w", err) } if _, err := tx.Exec(ctx, `INSERT INTO device_activation_transition(tenant_id,device_id,from_state,to_state,reason,source,actor_type,actor_public_id,request_id,activation_revision) VALUES($1::uuid,$2::uuid,NULL,'CLAIMABLE','ops-device-create','operator','user',$3::uuid,NULLIF($4,''),1)`, access.tenantID, deviceID, access.userPublicID, requestID); err != nil { return CreateResult{}, fmt.Errorf("record initial device activation: %w", err) } if _, err := tx.Exec(ctx, `INSERT INTO device_twin(device_id,tenant_id,desired_config) VALUES($1::uuid,$2::uuid,'{}'::jsonb)`, deviceID, access.tenantID); err != nil { return CreateResult{}, fmt.Errorf("insert device twin: %w", err) } if _, err := tx.Exec(ctx, `INSERT INTO device_onboarding_key(device_id,tenant_id,key_version,ciphertext,status,activated_at) VALUES($1::uuid,$2::uuid,$3,$4,'ACTIVE',now())`, deviceID, access.tenantID, keyVersion, ciphertext); err != nil { return CreateResult{}, fmt.Errorf("insert onboarding device key: %w", err) } if err := s.audit.Record(ctx, tx, audit.Event{TenantID: &access.tenantID, ActorType: "user", ActorPublicID: &access.userPublicID, Action: "ops.device.create", ResourceType: "device", ResourceID: &publicID, RequestID: requestID, Result: "success", Detail: map[string]any{"factory_id": factoryID, "model": input.Model, "key_version": keyVersion, "enrollment_mode": "physical_confirm"}}); err != nil { return CreateResult{}, err } item, err := readDevice(ctx, tx, access.tenantID, publicID) if err != nil { return CreateResult{}, err } if err := tx.Commit(ctx); err != nil { return CreateResult{}, fmt.Errorf("commit device registration: %w", err) } return CreateResult{Device: item, Credential: CredentialIssue{FactoryID: factoryID, Model: input.Model, Origin: s.deviceOrigin, DeviceKeyHex: secret, KeyVersion: keyVersion}}, nil } func (s *Service) IssueDeviceClaim(ctx context.Context, principal identity.Principal, tenantPublicID, devicePublicID, requestID string) (DeviceClaimIssue, error) { tx, err := s.pool.Begin(ctx) if err != nil { return DeviceClaimIssue{}, fmt.Errorf("begin device claim issue: %w", err) } defer tx.Rollback(ctx) access, err := requireAccess(ctx, tx, principal, tenantPublicID, true) if err != nil { return DeviceClaimIssue{}, err } var deviceID, status, activationState string var bound bool if err := tx.QueryRow(ctx, ` SELECT d.id::text,d.status,d.activation_state,EXISTS(SELECT 1 FROM device_binding b WHERE b.device_id=d.id AND b.unbound_at IS NULL) FROM device d WHERE d.public_id=$1::uuid AND d.tenant_id=$2::uuid FOR UPDATE`, devicePublicID, access.tenantID).Scan(&deviceID, &status, &activationState, &bound); errors.Is(err, pgx.ErrNoRows) { return DeviceClaimIssue{}, ErrNotFound } else if err != nil { return DeviceClaimIssue{}, fmt.Errorf("lock device for claim issue: %w", err) } if status != "registered" || activationState != string(deviceactivation.Claimable) || bound { return DeviceClaimIssue{}, ErrClaimUnavailable } issued, err := s.issueDeviceClaimTx(ctx, tx, access, devicePublicID, deviceID, requestID) if err != nil { return DeviceClaimIssue{}, err } if err := tx.Commit(ctx); err != nil { return DeviceClaimIssue{}, fmt.Errorf("commit device claim issue: %w", err) } return issued, nil } func (s *Service) issueDeviceClaimTx(ctx context.Context, tx pgx.Tx, access operatorAccess, devicePublicID, deviceID, requestID string) (DeviceClaimIssue, error) { if err := devicebinding.RequireLegacy(ctx, tx, deviceID); err != nil { return DeviceClaimIssue{}, err } randomBytes := make([]byte, 10) if _, err := rand.Read(randomBytes); err != nil { return DeviceClaimIssue{}, fmt.Errorf("generate device claim: %w", err) } rawCode := base32.StdEncoding.WithPadding(base32.NoPadding).EncodeToString(randomBytes) claimCode := strings.Join([]string{rawCode[0:4], rawCode[4:8], rawCode[8:12], rawCode[12:16]}, "-") digest := sha256.Sum256([]byte(claimCode)) issuedAt := time.Now().UTC() expiresAt := issuedAt.Add(24 * time.Hour) if _, err := tx.Exec(ctx, `UPDATE device_claim SET revoked_at=$1 WHERE device_id=$2::uuid AND used_at IS NULL AND revoked_at IS NULL`, issuedAt, deviceID); err != nil { return DeviceClaimIssue{}, fmt.Errorf("revoke previous device claim: %w", err) } if _, err := tx.Exec(ctx, `INSERT INTO device_claim(tenant_id,device_id,token_digest,expires_at,created_at) VALUES($1::uuid,$2::uuid,$3,$4,$5)`, access.tenantID, deviceID, hex.EncodeToString(digest[:]), expiresAt, issuedAt); err != nil { return DeviceClaimIssue{}, fmt.Errorf("insert device claim: %w", err) } if err := s.audit.Record(ctx, tx, audit.Event{TenantID: &access.tenantID, ActorType: "user", ActorPublicID: &access.userPublicID, Action: "ops.device.claim.issue", ResourceType: "device", ResourceID: &devicePublicID, RequestID: requestID, Result: "success", Detail: map[string]any{"expires_at": expiresAt}}); err != nil { return DeviceClaimIssue{}, err } return DeviceClaimIssue{DeviceID: devicePublicID, ClaimCode: claimCode, IssuedAt: issuedAt, ExpiresAt: expiresAt}, nil } func (s *Service) Update(ctx context.Context, principal identity.Principal, tenantPublicID, devicePublicID, requestID string, expectedVersion int64, input UpdateInput) (Device, error) { if input.HardwareAgentBotID != nil || input.HardwareAgentTransport != nil || input.HardwareAgentMinimumSDKVersion != nil { // Bot, transport and SDK policy are product defaults resolved during // provider provisioning, never per-device operator input. return Device{}, ErrInvalid } if expectedVersion < 1 || (input.Model == nil && input.HardwareRevision == nil && input.MinimumSecurityVersion == nil && input.MinimumFirmwareVersion == nil && input.Status == nil && input.HouseholdID == nil && input.SubjectID == nil && input.VoiceRuntime == nil) { return Device{}, ErrInvalid } if (input.HouseholdID == nil) != (input.SubjectID == nil) { return Device{}, ErrInvalid } tx, err := s.pool.Begin(ctx) if err != nil { return Device{}, fmt.Errorf("begin device update: %w", err) } defer tx.Rollback(ctx) access, err := requireAccess(ctx, tx, principal, tenantPublicID, true) if err != nil { return Device{}, err } var deviceID, currentActivationState string var currentVersion int64 if err := tx.QueryRow(ctx, `SELECT id::text,version,activation_state FROM device WHERE public_id=$1::uuid AND tenant_id=$2::uuid FOR UPDATE`, devicePublicID, access.tenantID).Scan(&deviceID, ¤tVersion, ¤tActivationState); errors.Is(err, pgx.ErrNoRows) { return Device{}, ErrNotFound } else if err != nil { return Device{}, fmt.Errorf("lock device: %w", err) } if currentVersion != expectedVersion { return Device{}, ErrVersionConflict } model, modelSet := pointerValue(input.Model) if input.Model != nil || input.HardwareRevision != nil || input.VoiceRuntime != nil || (input.HouseholdID != nil && strings.TrimSpace(*input.HouseholdID) != "") { if err := devicebinding.RequireLegacy(ctx, tx, deviceID); err != nil { return Device{}, err } } hardware, hardwareSet := pointerValue(input.HardwareRevision) security, securitySet := pointerValue(input.MinimumSecurityVersion) firmware, firmwareSet := pointerValue(input.MinimumFirmwareVersion) status, statusSet := pointerValue(input.Status) voiceRuntime, voiceRuntimeSet := pointerValue(input.VoiceRuntime) if (modelSet && (model == "" || len([]rune(model)) > 120)) || len([]rune(hardware)) > 80 || len([]rune(security)) > 80 || len([]rune(firmware)) > 80 || (statusSet && status != "registered" && status != "quarantined") || (voiceRuntimeSet && !validVoiceRuntime(voiceRuntime)) { return Device{}, ErrInvalid } if _, err := tx.Exec(ctx, `UPDATE device SET model=CASE WHEN $3 THEN $4 ELSE model END, hardware_revision=CASE WHEN $5 THEN NULLIF($6,'') ELSE hardware_revision END, minimum_security_version=CASE WHEN $7 THEN NULLIF($8,'') ELSE minimum_security_version END, minimum_firmware_version=CASE WHEN $9 THEN NULLIF($10,'') ELSE minimum_firmware_version END, status=CASE WHEN $11 THEN $12 ELSE status END,version=version+1,updated_at=now() WHERE id=$1::uuid AND tenant_id=$2::uuid`, deviceID, access.tenantID, modelSet, model, hardwareSet, hardware, securitySet, security, firmwareSet, firmware, statusSet, status); err != nil { return Device{}, fmt.Errorf("update device: %w", err) } if input.HouseholdID != nil { householdID := strings.TrimSpace(*input.HouseholdID) subjectID := strings.TrimSpace(*input.SubjectID) if (householdID == "") != (subjectID == "") { return Device{}, ErrInvalid } if householdID != "" { if err := validateAssignment(ctx, tx, access.tenantID, householdID, subjectID); err != nil { return Device{}, err } } if _, err := tx.Exec(ctx, `UPDATE device_binding SET unbound_at=now(),unbound_by=$2::uuid,reason='运维端变更归属' WHERE device_id=$1::uuid AND unbound_at IS NULL`, deviceID, access.userID); err != nil { return Device{}, fmt.Errorf("close device binding: %w", err) } if householdID != "" { if err := insertBinding(ctx, tx, access, deviceID, householdID, subjectID, "运维端变更归属"); err != nil { return Device{}, err } if _, err := tx.Exec(ctx, `UPDATE device_claim SET revoked_at=now() WHERE device_id=$1::uuid AND used_at IS NULL AND revoked_at IS NULL`, deviceID); err != nil { return Device{}, fmt.Errorf("revoke device claims after assignment: %w", err) } if currentActivationState == string(deviceactivation.Claimable) { transition, err := deviceactivation.Apply(ctx, tx, deviceactivation.Input{DeviceID: deviceID, TenantID: access.tenantID, Target: deviceactivation.Bound, Reason: "运维端绑定设备", Source: "operator-assignment", ActorType: "user", ActorPublicID: &access.userPublicID, RequestID: requestID}) if err != nil { return Device{}, mapActivationError(err) } currentActivationState = string(transition.To) } } else if currentActivationState == string(deviceactivation.Bound) || currentActivationState == string(deviceactivation.Provisioning) || currentActivationState == string(deviceactivation.Active) || currentActivationState == string(deviceactivation.Lost) { transition, err := deviceactivation.Apply(ctx, tx, deviceactivation.Input{DeviceID: deviceID, TenantID: access.tenantID, Target: deviceactivation.Claimable, Reason: "运维端解绑设备", Source: "operator-unbind", ActorType: "user", ActorPublicID: &access.userPublicID, RequestID: requestID}) if err != nil { return Device{}, mapActivationError(err) } currentActivationState = string(transition.To) if _, err := tx.Exec(ctx, `UPDATE hardware_agent_provider_credential SET status='SUSPENDED',updated_at=now() WHERE device_id=$1::uuid AND status='ACTIVE'`, deviceID); err != nil { return Device{}, fmt.Errorf("suspend provider credential after unbind: %w", err) } } } if voiceRuntimeSet { desiredConfig, _ := json.Marshal(map[string]string{"voice_runtime": voiceRuntime}) if voiceRuntime == "volc_hardware_agent" { if _, err := tx.Exec(ctx, `UPDATE device_twin SET desired_revision=desired_revision+1,desired_config=desired_config||$2::jsonb,updated_at=now() WHERE device_id=$1::uuid`, deviceID, desiredConfig); err != nil { return Device{}, err } } else { if _, err := tx.Exec(ctx, `UPDATE device_twin SET desired_revision=desired_revision+1,desired_config=(desired_config||$2::jsonb)-'hardware_agent_transport'-'minimum_sdk_version',updated_at=now() WHERE device_id=$1::uuid`, deviceID, desiredConfig); err != nil { return Device{}, err } if _, err := tx.Exec(ctx, `UPDATE hardware_agent_device_binding SET status='DISABLED',updated_at=now() WHERE device_id=$1::uuid`, deviceID); err != nil { return Device{}, err } } } if currentActivationState == string(deviceactivation.Bound) { var runtime string var hasBinding bool if err := tx.QueryRow(ctx, `SELECT COALESCE(desired_config->>'voice_runtime','legacy_voicechat'),EXISTS(SELECT 1 FROM device_binding WHERE device_id=$1::uuid AND unbound_at IS NULL) FROM device_twin WHERE device_id=$1::uuid`, deviceID).Scan(&runtime, &hasBinding); err != nil { return Device{}, fmt.Errorf("read activation runtime: %w", err) } if runtime == "legacy_voicechat" && hasBinding { if _, err := deviceactivation.Apply(ctx, tx, deviceactivation.Input{DeviceID: deviceID, TenantID: access.tenantID, Target: deviceactivation.Provisioning, Reason: "旧版语音运行时平台凭据预配", Source: "platform-provisioning", ActorType: "user", ActorPublicID: &access.userPublicID, RequestID: requestID}); err != nil { return Device{}, mapActivationError(err) } if _, err := deviceactivation.Apply(ctx, tx, deviceactivation.Input{DeviceID: deviceID, TenantID: access.tenantID, Target: deviceactivation.Active, Reason: "旧版语音运行时已具备绑定与平台凭据", Source: "platform-provisioning", ActorType: "user", ActorPublicID: &access.userPublicID, RequestID: requestID}); err != nil { return Device{}, mapActivationError(err) } } } if currentActivationState == string(deviceactivation.Active) { var runtime string var providerReady bool if err := tx.QueryRow(ctx, `SELECT COALESCE(desired_config->>'voice_runtime','legacy_voicechat'),EXISTS(SELECT 1 FROM hardware_agent_provider_credential WHERE device_id=$1::uuid AND status='ACTIVE') FROM device_twin WHERE device_id=$1::uuid`, deviceID).Scan(&runtime, &providerReady); err != nil { return Device{}, fmt.Errorf("read active runtime readiness: %w", err) } if runtime == "volc_hardware_agent" && !providerReady { if _, err := deviceactivation.Apply(ctx, tx, deviceactivation.Input{DeviceID: deviceID, TenantID: access.tenantID, Target: deviceactivation.Provisioning, Reason: "语音运行时切换为火山硬件智能体,等待一机一密", Source: "runtime-change", ActorType: "user", ActorPublicID: &access.userPublicID, RequestID: requestID}); err != nil { return Device{}, mapActivationError(err) } } } var finalVersion int64 if err := tx.QueryRow(ctx, `SELECT version FROM device WHERE id=$1::uuid`, deviceID).Scan(&finalVersion); err != nil { return Device{}, fmt.Errorf("read updated device version: %w", err) } if err := s.audit.Record(ctx, tx, audit.Event{TenantID: &access.tenantID, ActorType: "user", ActorPublicID: &access.userPublicID, Action: "ops.device.update", ResourceType: "device", ResourceID: &devicePublicID, RequestID: requestID, Result: "success", Detail: map[string]any{"from_version": currentVersion, "to_version": finalVersion}}); err != nil { return Device{}, err } item, err := readDevice(ctx, tx, access.tenantID, devicePublicID) if err != nil { return Device{}, err } if err := tx.Commit(ctx); err != nil { return Device{}, fmt.Errorf("commit device update: %w", err) } return item, nil } func (s *Service) Delete(ctx context.Context, principal identity.Principal, tenantPublicID, devicePublicID, requestID string, expectedVersion int64) (Device, error) { tx, err := s.pool.Begin(ctx) if err != nil { return Device{}, fmt.Errorf("begin device retirement: %w", err) } defer tx.Rollback(ctx) access, err := requireAccess(ctx, tx, principal, tenantPublicID, true) if err != nil { return Device{}, err } var deviceID string var version int64 if err := tx.QueryRow(ctx, `SELECT id::text,version FROM device WHERE public_id=$1::uuid AND tenant_id=$2::uuid FOR UPDATE`, devicePublicID, access.tenantID).Scan(&deviceID, &version); errors.Is(err, pgx.ErrNoRows) { return Device{}, ErrNotFound } else if err != nil { return Device{}, fmt.Errorf("lock device for retirement: %w", err) } if version != expectedVersion { return Device{}, ErrVersionConflict } if _, err := deviceactivation.Apply(ctx, tx, deviceactivation.Input{DeviceID: deviceID, TenantID: access.tenantID, Target: deviceactivation.Revoked, Reason: "运维端注销设备", Source: "operator-retirement", ActorType: "user", ActorPublicID: &access.userPublicID, RequestID: requestID}); err != nil { return Device{}, mapActivationError(err) } if _, err := tx.Exec(ctx, `UPDATE device SET status='revoked',revoked_at=COALESCE(revoked_at,now()),updated_at=now() WHERE id=$1::uuid`, deviceID); err != nil { return Device{}, fmt.Errorf("retire device: %w", err) } if _, err := tx.Exec(ctx, `UPDATE device_credential SET status='revoked',revoked_at=COALESCE(revoked_at,now()) WHERE device_id=$1::uuid`, deviceID); err != nil { return Device{}, fmt.Errorf("revoke device credential: %w", err) } if _, err := tx.Exec(ctx, `UPDATE device_onboarding_key SET status='REVOKED' WHERE device_id=$1::uuid AND status IN ('ACTIVE','PENDING')`, deviceID); err != nil { return Device{}, fmt.Errorf("revoke onboarding device key: %w", err) } if _, err := tx.Exec(ctx, `UPDATE device_binding SET unbound_at=now(),unbound_by=$2::uuid,reason='运维端删除设备' WHERE device_id=$1::uuid AND unbound_at IS NULL`, deviceID, access.userID); err != nil { return Device{}, fmt.Errorf("close retired device binding: %w", err) } if _, err := tx.Exec(ctx, `UPDATE device_account_binding SET released_at=now() WHERE device_id=$1::uuid AND released_at IS NULL`, deviceID); err != nil { return Device{}, fmt.Errorf("close retired device account binding: %w", err) } if _, err := tx.Exec(ctx, `UPDATE device_claim SET revoked_at=now() WHERE device_id=$1::uuid AND used_at IS NULL AND revoked_at IS NULL`, deviceID); err != nil { return Device{}, fmt.Errorf("revoke retired device claims: %w", err) } if _, err := tx.Exec(ctx, `UPDATE hardware_agent_device_binding SET status='REVOKED',updated_at=now() WHERE device_id=$1::uuid`, deviceID); err != nil { return Device{}, fmt.Errorf("revoke hardware agent binding: %w", err) } if _, err := tx.Exec(ctx, `UPDATE hardware_agent_provider_credential SET status='REVOKED',revoked_at=COALESCE(revoked_at,now()),updated_at=now() WHERE device_id=$1::uuid`, deviceID); err != nil { return Device{}, fmt.Errorf("revoke hardware provider credential: %w", err) } if err := s.audit.Record(ctx, tx, audit.Event{TenantID: &access.tenantID, ActorType: "user", ActorPublicID: &access.userPublicID, Action: "ops.device.delete", ResourceType: "device", ResourceID: &devicePublicID, RequestID: requestID, Result: "success", Detail: map[string]any{"soft_delete": true, "credential_revoked": true}}); err != nil { return Device{}, err } item, err := readDevice(ctx, tx, access.tenantID, devicePublicID) if err != nil { return Device{}, err } if err := tx.Commit(ctx); err != nil { return Device{}, fmt.Errorf("commit device retirement: %w", err) } return item, nil } func (s *Service) TransitionActivation(ctx context.Context, principal identity.Principal, tenantPublicID, devicePublicID, requestID string, input ActivationTransitionInput) (Device, error) { input.Reason = strings.TrimSpace(input.Reason) target, err := deviceactivation.Parse(input.TargetState) if err != nil || input.Reason == "" || len([]rune(input.Reason)) > 500 || input.ExpectedRevision < 1 || target == deviceactivation.Revoked { return Device{}, ErrInvalid } tx, err := s.pool.Begin(ctx) if err != nil { return Device{}, fmt.Errorf("begin device activation transition: %w", err) } defer tx.Rollback(ctx) access, err := requireAccess(ctx, tx, principal, tenantPublicID, true) if err != nil { return Device{}, err } var deviceID, currentState, runtime string if err := tx.QueryRow(ctx, ` SELECT d.id::text,d.activation_state,COALESCE(dt.desired_config->>'voice_runtime','legacy_voicechat') FROM device d LEFT JOIN device_twin dt ON dt.device_id=d.id WHERE d.public_id=$1::uuid AND d.tenant_id=$2::uuid FOR UPDATE OF d`, devicePublicID, access.tenantID). Scan(&deviceID, ¤tState, &runtime); errors.Is(err, pgx.ErrNoRows) { return Device{}, ErrNotFound } else if err != nil { return Device{}, fmt.Errorf("lock device activation target: %w", err) } // Binding is a business operation, retirement is DELETE, and provisioning // completion normally comes from the device provisioning endpoint. if target == deviceactivation.Bound || (target == deviceactivation.Claimable && currentState != string(deviceactivation.Factory)) { return Device{}, ErrInvalid } if target == deviceactivation.Active { var bindingReady, platformCredentialReady, providerCredentialReady bool if err := tx.QueryRow(ctx, `SELECT EXISTS(SELECT 1 FROM device_binding WHERE device_id=$1::uuid AND unbound_at IS NULL), EXISTS(SELECT 1 FROM device_credential WHERE device_id=$1::uuid AND status='active'), EXISTS(SELECT 1 FROM hardware_agent_provider_credential WHERE device_id=$1::uuid AND status IN ('ACTIVE','SUSPENDED'))`, deviceID). Scan(&bindingReady, &platformCredentialReady, &providerCredentialReady); err != nil { return Device{}, fmt.Errorf("check activation readiness: %w", err) } if !bindingReady || !platformCredentialReady || (runtime == "volc_hardware_agent" && !providerCredentialReady) { return Device{}, ErrActivationNotReady } } expected := input.ExpectedRevision transition, err := deviceactivation.Apply(ctx, tx, deviceactivation.Input{DeviceID: deviceID, TenantID: access.tenantID, Target: target, Reason: input.Reason, Source: "operator-transition", ActorType: "user", ActorPublicID: &access.userPublicID, RequestID: requestID, ExpectedRevision: &expected}) if err != nil { return Device{}, mapActivationError(err) } if target == deviceactivation.Lost { if _, err := tx.Exec(ctx, `UPDATE device SET status='quarantined',updated_at=now() WHERE id=$1::uuid`, deviceID); err != nil { return Device{}, fmt.Errorf("quarantine lost device: %w", err) } if _, err := tx.Exec(ctx, `UPDATE hardware_agent_provider_credential SET status='SUSPENDED',updated_at=now() WHERE device_id=$1::uuid AND status='ACTIVE'`, deviceID); err != nil { return Device{}, fmt.Errorf("suspend lost device provider credential: %w", err) } } else if target == deviceactivation.Active { if _, err := tx.Exec(ctx, `UPDATE device SET status='registered',updated_at=now() WHERE id=$1::uuid`, deviceID); err != nil { return Device{}, fmt.Errorf("restore active device: %w", err) } if _, err := tx.Exec(ctx, `UPDATE hardware_agent_provider_credential SET status='ACTIVE',revoked_at=NULL,updated_at=now() WHERE device_id=$1::uuid AND status='SUSPENDED'`, deviceID); err != nil { return Device{}, fmt.Errorf("restore provider credential: %w", err) } } if err := s.audit.Record(ctx, tx, audit.Event{TenantID: &access.tenantID, ActorType: "user", ActorPublicID: &access.userPublicID, Action: "ops.device.activation.transition", ResourceType: "device", ResourceID: &devicePublicID, RequestID: requestID, Result: "success", Detail: map[string]any{"from_state": transition.From, "to_state": transition.To, "activation_revision": transition.Revision, "reason": input.Reason}}); err != nil { return Device{}, err } item, err := readDevice(ctx, tx, access.tenantID, devicePublicID) if err != nil { return Device{}, err } if err := tx.Commit(ctx); err != nil { return Device{}, fmt.Errorf("commit device activation transition: %w", err) } return item, nil } func (s *Service) RotateCredential(ctx context.Context, principal identity.Principal, tenantPublicID, devicePublicID, requestID string) (CredentialIssue, error) { secretBytes := make([]byte, 32) if _, err := rand.Read(secretBytes); err != nil { return CredentialIssue{}, fmt.Errorf("generate device credential: %w", err) } tx, err := s.pool.Begin(ctx) if err != nil { return CredentialIssue{}, fmt.Errorf("begin credential rotation: %w", err) } defer tx.Rollback(ctx) access, err := requireAccess(ctx, tx, principal, tenantPublicID, true) if err != nil { return CredentialIssue{}, err } var rotationDeviceID, enrollmentMode, factoryID, model string if err := tx.QueryRow(ctx, `SELECT id::text,enrollment_mode,COALESCE(factory_device_id,''),model FROM device WHERE public_id=$1::uuid AND tenant_id=$2::uuid AND status='registered' AND activation_state NOT IN ('LOST','REVOKED') FOR UPDATE`, devicePublicID, access.tenantID).Scan(&rotationDeviceID, &enrollmentMode, &factoryID, &model); err != nil { return CredentialIssue{}, ErrNotFound } if enrollmentMode != "physical_confirm" || factoryID == "" || s.onboardingMasterKey == "" || s.deviceOrigin == "" { return CredentialIssue{}, ErrProvisioningUnavailable } var version int64 if err := tx.QueryRow(ctx, `SELECT COALESCE(max(key_version),0)+1 FROM device_onboarding_key WHERE device_id=$1::uuid`, rotationDeviceID).Scan(&version); err != nil { return CredentialIssue{}, fmt.Errorf("read onboarding key version: %w", err) } ciphertext, err := onboarding.SealDeviceKey(s.onboardingMasterKey, rotationDeviceID, version, secretBytes) if err != nil { return CredentialIssue{}, fmt.Errorf("seal rotated onboarding device key: %w", err) } if _, err := tx.Exec(ctx, `UPDATE device_onboarding_key SET status='REVOKED' WHERE device_id=$1::uuid AND status='ACTIVE'`, rotationDeviceID); err != nil { return CredentialIssue{}, fmt.Errorf("revoke previous onboarding device key: %w", err) } if _, err := tx.Exec(ctx, `INSERT INTO device_onboarding_key(device_id,tenant_id,key_version,ciphertext,status,activated_at) VALUES($1::uuid,$2::uuid,$3,$4,'ACTIVE',now())`, rotationDeviceID, access.tenantID, version, ciphertext); err != nil { return CredentialIssue{}, fmt.Errorf("insert rotated onboarding device key: %w", err) } if _, err := tx.Exec(ctx, `UPDATE device SET onboarding_revision=onboarding_revision+1,version=version+1,updated_at=now() WHERE id=$1::uuid`, rotationDeviceID); err != nil { return CredentialIssue{}, fmt.Errorf("update onboarding key revision: %w", err) } if err := s.audit.Record(ctx, tx, audit.Event{TenantID: &access.tenantID, ActorType: "user", ActorPublicID: &access.userPublicID, Action: "ops.device.credential.rotate", ResourceType: "device", ResourceID: &devicePublicID, RequestID: requestID, Result: "success", Detail: map[string]any{"key_version": version}}); err != nil { return CredentialIssue{}, err } if err := tx.Commit(ctx); err != nil { return CredentialIssue{}, fmt.Errorf("commit credential rotation: %w", err) } return CredentialIssue{FactoryID: factoryID, Model: model, Origin: s.deviceOrigin, DeviceKeyHex: hex.EncodeToString(secretBytes), KeyVersion: version}, nil } func requireAccess(ctx context.Context, q queryRow, principal identity.Principal, tenantPublicID string, mutate bool) (operatorAccess, error) { var access operatorAccess err := q.QueryRow(ctx, `SELECT t.id::text,t.public_id::text,ua.id::text,ua.public_id::text,ua.platform_role FROM tenant t JOIN user_account ua ON ua.tenant_id=t.id AND ua.status='active' AND ua.account_kind='operator' WHERE t.public_id=$1::uuid AND t.status='active' AND ua.oidc_issuer=$2 AND ua.oidc_subject=$3`, tenantPublicID, principal.Issuer, principal.Subject). Scan(&access.tenantID, &access.tenantPublicID, &access.userID, &access.userPublicID, &access.role) if errors.Is(err, pgx.ErrNoRows) { return operatorAccess{}, ErrAccessDenied } if err != nil { return operatorAccess{}, fmt.Errorf("resolve operations device access: %w", err) } if mutate && access.role != "ops_operator" && access.role != "ops_admin" { return operatorAccess{}, ErrAccessDenied } return access, nil } func validateAssignment(ctx context.Context, q queryRow, tenantID, householdPublicID, subjectPublicID string) error { var exists bool err := q.QueryRow(ctx, `SELECT true FROM household h JOIN subject_profile sp ON sp.household_id=h.id AND sp.tenant_id=h.tenant_id WHERE h.public_id=$1::uuid AND sp.public_id=$2::uuid AND h.tenant_id=$3::uuid AND h.status='active' AND sp.status IN ('active','suspended')`, householdPublicID, subjectPublicID, tenantID).Scan(&exists) if errors.Is(err, pgx.ErrNoRows) { return ErrInvalid } if err != nil { return fmt.Errorf("validate device assignment: %w", err) } return nil } func insertBinding(ctx context.Context, tx pgx.Tx, access operatorAccess, deviceID, householdPublicID, subjectPublicID, reason string) error { if err := devicebinding.RequireLegacy(ctx, tx, deviceID); err != nil { return err } _, err := tx.Exec(ctx, `INSERT INTO device_binding(tenant_id,device_id,household_id,subject_id,bound_by,reason) SELECT $1::uuid,$2::uuid,h.id,sp.id,$5::uuid,$6 FROM household h JOIN subject_profile sp ON sp.household_id=h.id AND sp.tenant_id=h.tenant_id WHERE h.public_id=$3::uuid AND sp.public_id=$4::uuid AND h.tenant_id=$1::uuid`, access.tenantID, deviceID, householdPublicID, subjectPublicID, access.userID, reason) if err != nil { return fmt.Errorf("insert device binding: %w", err) } return nil } func pointerValue(value *string) (string, bool) { if value == nil { return "", false } return strings.TrimSpace(*value), true } func validVoiceRuntime(value string) bool { return value == "legacy_voicechat" || value == "volc_hardware_agent" } func validHardwareTransport(value string) bool { return value == "high_quality" || value == "low_load" } func mapActivationError(err error) error { if errors.Is(err, deviceactivation.ErrRevisionConflict) || errors.Is(err, deviceactivation.ErrTransitionRejected) { return ErrActivationConflict } if errors.Is(err, deviceactivation.ErrInvalidState) { return ErrInvalid } return err } const deviceSelectSQL = ` SELECT d.public_id::text,t.public_id::text,COALESCE(d.factory_device_id,''),d.enrollment_mode, (EXISTS(SELECT 1 FROM device_binding current_binding WHERE current_binding.device_id=d.id AND current_binding.unbound_at IS NULL) OR EXISTS(SELECT 1 FROM device_account_binding account_binding WHERE account_binding.device_id=d.id AND account_binding.released_at IS NULL AND account_binding.generation=d.binding_generation)), d.model,d.hardware_revision,d.status,d.activation_state,d.activation_revision,d.activation_reason,d.activation_updated_at,d.minimum_security_version,d.minimum_firmware_version, h.public_id::text,h.name,sp.public_id::text,sp.preferred_name,COALESCE(lower(ok.status),dc.status,'missing'),COALESCE(ok.key_version,dc.key_version,0), dt.firmware_version,dt.last_reported_at,d.registered_at,d.updated_at,d.version, COALESCE(dt.desired_config->>'voice_runtime',CASE WHEN d.enrollment_mode='physical_confirm' THEN 'volc_hardware_agent' ELSE 'legacy_voicechat' END),COALESCE(hb.bot_id,''),COALESCE(hb.transport,''),COALESCE(hb.minimum_sdk_version,''),COALESCE(hb.status,'') FROM device d JOIN tenant t ON t.id=d.tenant_id LEFT JOIN device_twin dt ON dt.device_id=d.id AND dt.tenant_id=d.tenant_id LEFT JOIN device_binding b ON b.device_id=d.id AND b.tenant_id=d.tenant_id AND b.unbound_at IS NULL LEFT JOIN household h ON h.id=b.household_id AND h.tenant_id=b.tenant_id LEFT JOIN subject_profile sp ON sp.id=b.subject_id AND sp.tenant_id=b.tenant_id LEFT JOIN device_credential dc ON dc.device_id=d.id AND dc.tenant_id=d.tenant_id LEFT JOIN device_onboarding_key ok ON ok.device_id=d.id AND ok.tenant_id=d.tenant_id AND ok.status='ACTIVE' LEFT JOIN hardware_agent_device_binding hb ON hb.device_id=d.id AND hb.tenant_id=d.tenant_id` type scanner interface{ Scan(...any) error } func scanDevice(row scanner) (Device, error) { var item Device if err := row.Scan(&item.ID, &item.TenantID, &item.FactoryID, &item.EnrollmentMode, &item.Bound, &item.Model, &item.HardwareRevision, &item.Status, &item.ActivationState, &item.ActivationRevision, &item.ActivationReason, &item.ActivationUpdatedAt, &item.MinimumSecurityVersion, &item.MinimumFirmwareVersion, &item.HouseholdID, &item.HouseholdName, &item.SubjectID, &item.SubjectName, &item.CredentialStatus, &item.KeyVersion, &item.FirmwareVersion, &item.LastReportedAt, &item.RegisteredAt, &item.UpdatedAt, &item.Version, &item.VoiceRuntime, &item.HardwareAgentBotID, &item.HardwareAgentTransport, &item.HardwareAgentMinimumSDKVersion, &item.HardwareAgentStatus); err != nil { return Device{}, fmt.Errorf("scan operations device: %w", err) } return item, nil } func readDevice(ctx context.Context, q queryRow, tenantID, publicID string) (Device, error) { item, err := scanDevice(q.QueryRow(ctx, deviceSelectSQL+` WHERE d.tenant_id=$1::uuid AND d.public_id=$2::uuid`, tenantID, publicID)) if errors.Is(err, pgx.ErrNoRows) { return Device{}, ErrNotFound } return item, err }