package onboarding import ( "context" "errors" "github.com/jackc/pgx/v5" ) // Join the direct-call transaction fence before taking the same twin/device // lock order as commands and OTA. A new prepare cannot race the final idle check. func lockBLEControl(ctx context.Context, tx pgx.Tx, deviceID, factoryID string) error { if deviceID == "" { err := tx.QueryRow(ctx, `SELECT id::text FROM device WHERE factory_device_id=$1`, factoryID).Scan(&deviceID) if errors.Is(err, pgx.ErrNoRows) { return nil } // Normal access validation supplies the precise error. if err != nil { return err } } if _, err := tx.Exec(ctx, `SELECT pg_advisory_xact_lock(hashtextextended($1,11))`, deviceID); err != nil { return err } _, err := tx.Exec(ctx, `SELECT device_id FROM device_twin WHERE device_id=$1::uuid FOR UPDATE`, deviceID) return err } func (s *Service) checkBLETransfer(ctx context.Context, tx pgx.Tx, d Device, target claimRecord, sourceID string) error { source := claimRecord{TenantID: d.TenantID, UserID: target.UserID} err := tx.QueryRow(ctx, `SELECT household_id::text,subject_id::text FROM device_binding WHERE id=$1::uuid AND device_id=$2::uuid AND tenant_id=$3::uuid AND unbound_at IS NULL AND generation=$4 FOR UPDATE`, sourceID, d.ID, d.TenantID, d.Generation).Scan(&source.HouseholdID, &source.SubjectID) if errors.Is(err, pgx.ErrNoRows) { return fail(409, "BINDING_CHANGED", "设备归属已变化,请刷新后重新选择") } if err != nil { return err } if err = claimPermission(ctx, tx, source); err != nil { return err } if source.SubjectID == target.SubjectID { return fail(400, "ELDER_UNCHANGED", "请选择另一位使用老人") } if err = s.expireClaims(ctx, tx, d.ID); err != nil { return err } var busy bool err = tx.QueryRow(ctx, `SELECT EXISTS(SELECT 1 FROM call_session WHERE device_id=$1::uuid AND call_state IN ('PREPARING','STARTING','TALKING','STOPPING')) OR EXISTS(SELECT 1 FROM conversation WHERE device_id=$1::uuid AND status NOT IN ('COMPLETED','FAILED','EXPIRED')) OR EXISTS(SELECT 1 FROM device_command WHERE device_id=$1::uuid AND status IN ('QUEUED','DELIVERED','ACKNOWLEDGED') AND expires_at>now()) OR EXISTS(SELECT 1 FROM firmware_assignment WHERE device_id=$1::uuid AND status IN ('PENDING','ASSIGNED','COMMAND_QUEUED','DOWNLOADING','INSTALLING')) OR EXISTS(SELECT 1 FROM fleet_ota WHERE device_id=$1::uuid AND status IN ('pending','accepted')) OR EXISTS(SELECT 1 FROM device_binding_claim WHERE device_id=$1::uuid AND status='PENDING') OR EXISTS(SELECT 1 FROM device_account_binding WHERE device_id=$1::uuid AND released_at IS NULL)`, d.ID).Scan(&busy) if err != nil { return err } if busy { return fail(409, "DEVICE_BUSY", "设备有通话、命令、升级或待确认操作,请完成后重试") } return nil } // Only commitBLEBinding, after DEVREQ1 device authentication, can reach this. // The old assignment stays active while a token waits for the physical tap. func (s *Service) commitBLETransfer(ctx context.Context, tx pgx.Tx, d Device, c bleClaim, requestID, token string) (map[string]any, error) { if err := s.checkBLETransfer(ctx, tx, d, c.claimRecord, c.BindingID); err != nil { return nil, err } sourceID := c.BindingID if _, err := tx.Exec(ctx, `UPDATE device_binding SET unbound_at=now(),unbound_by=$2::uuid,reason='ble_physical_transfer' WHERE id=$1::uuid`, sourceID, c.UserID); err != nil { return nil, err } // Closing the old binding advances the generation. The new claim must be // fenced against that value, and its receipt must survive response retries. d, err := readDevice(ctx, tx, d.ID, true) if err != nil { return nil, err } c.Generation = d.Generation if _, err = tx.Exec(ctx, `UPDATE device_ble_binding SET expected_generation=$2 WHERE id=$1::uuid`, c.ID, d.Generation); err != nil { return nil, err } result, err := s.commitBLESubject(ctx, tx, d, c, requestID, token) if err != nil { return nil, err } if err = s.auditEvent(ctx, tx, d, "device", d.PublicID, "device.binding.ble_transfer", requestID, map[string]any{"previous_binding_id": sourceID, "binding_generation": d.Generation + 1}); err != nil { return nil, err } return result, tx.Commit(ctx) }