story/internal/devicereset/service.go

package devicereset

import (
	"context"
	"crypto/rand"
	"crypto/sha256"
	"encoding/base32"
	"encoding/hex"
	"encoding/json"
	"errors"
	"fmt"
	"net/http"
	"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/idempotency"

	"github.com/google/uuid"
	"github.com/jackc/pgx/v5"
	"github.com/jackc/pgx/v5/pgxpool"
)

var (
	ErrInvalid              = errors.New("device factory reset input is invalid")
	ErrStateConflict        = errors.New("device state does not allow factory reset")
	ErrResetCounterConflict = errors.New("device factory reset counter is stale")
	ErrFirmwareInProgress   = errors.New("device firmware update is in progress")
	ErrPendingTranscripts   = errors.New("device has pending transcripts")
	ErrCallInProgress       = errors.New("device has an active call")
)

type Input struct {
	ResetID              string    `json:"reset_id"`
	ResetCounter         int64     `json:"reset_counter"`
	OccurredAt           time.Time `json:"occurred_at"`
	Trigger              string    `json:"trigger"`
	FirmwareVersion      string    `json:"firmware_version"`
	PendingTurnCount     int       `json:"pending_turn_count"`
	ActiveConversationID string    `json:"active_conversation_id,omitempty"`
}

type Claim struct {
	Code      string    `json:"code"`
	IssuedAt  time.Time `json:"issued_at"`
	ExpiresAt time.Time `json:"expires_at"`
}

type Response struct {
	BindingGeneration        int64     `json:"binding_generation,omitempty"`
	ResetID                  string    `json:"reset_id"`
	DeviceID                 string    `json:"device_id"`
	ActivationState          string    `json:"activation_state"`
	ResetCounter             int64     `json:"reset_counter"`
	BindingRemoved           bool      `json:"binding_removed"`
	PreviousBindingID        string    `json:"previous_binding_id,omitempty"`
	ProviderCredentialStatus string    `json:"provider_credential_status"`
	LicenseAction            string    `json:"license_action"`
	EndedConversationCount   int       `json:"ended_conversation_count"`
	CancelledCommandCount    int64     `json:"cancelled_command_count"`
	ConfigRevision           int64     `json:"config_revision"`
	Claim                    *Claim    `json:"claim,omitempty"`
	LocalActions             []string  `json:"local_actions"`
	ResetAt                  time.Time `json:"reset_at"`
}

type Service struct {
	pool     *pgxpool.Pool
	executor *idempotency.Executor
	audit    *audit.Repository
}

func NewService(pool *pgxpool.Pool, executor *idempotency.Executor, auditRepository *audit.Repository) (*Service, error) {
	if pool == nil || executor == nil || auditRepository == nil {
		return nil, errors.New("device factory reset dependencies are required")
	}
	return &Service{pool: pool, executor: executor, audit: auditRepository}, nil
}

func Validate(input Input) error {
	if _, err := uuid.Parse(strings.TrimSpace(input.ResetID)); err != nil || input.ResetCounter < 1 || input.OccurredAt.IsZero() ||
		input.OccurredAt.After(time.Now().UTC().Add(5*time.Minute)) || len(strings.TrimSpace(input.FirmwareVersion)) < 1 ||
		len(strings.TrimSpace(input.FirmwareVersion)) > 80 || input.PendingTurnCount < 0 {
		return ErrInvalid
	}
	if input.Trigger != "physical_long_press" && input.Trigger != "device_menu" && input.Trigger != "service_tool" {
		return ErrInvalid
	}
	if input.ActiveConversationID != "" {
		if _, err := uuid.Parse(input.ActiveConversationID); err != nil {
			return ErrInvalid
		}
	}
	return nil
}

func (s *Service) Reset(ctx context.Context, identity deviceauth.Identity, idempotencyKey, requestID string, rawRequest []byte, input Input) (Response, bool, error) {
	if err := Validate(input); err != nil {
		return Response{}, false, err
	}
	if input.PendingTurnCount > 0 {
		return Response{}, false, ErrPendingTranscripts
	}
	tenantID := identity.TenantID
	if err := devicebinding.RequireUnboundForReset(ctx, s.pool, identity.ID); err != nil {
		return Response{}, false, err
	}
	guard := func(ctx context.Context, tx pgx.Tx, cached *idempotency.Result) error {
		// Match device control, OTA and ownership writers: twin before device.
		var lockedDevice string
		if err := tx.QueryRow(ctx, `SELECT device_id::text FROM device_twin WHERE device_id=$1 AND tenant_id=$2 FOR UPDATE`, identity.ID, identity.TenantID).Scan(&lockedDevice); err != nil {
			return err
		}
		var mode string
		var generation, counter int64
		if err := tx.QueryRow(ctx, `SELECT enrollment_mode,binding_generation,factory_reset_counter FROM device WHERE id=$1::uuid FOR UPDATE`, identity.ID).Scan(&mode, &generation, &counter); err != nil {
			return err
		}
		if mode != "physical_confirm" {
			return nil
		}
		if generation != identity.BindingGeneration {
			return ErrStateConflict
		}
		if err := devicebinding.RequireUnboundForReset(ctx, tx, identity.ID); err != nil {
			return err
		}
		if cached != nil {
			var previous Response
			if err := json.Unmarshal(cached.Body, &previous); err != nil {
				return err
			}
			if counter != input.ResetCounter || generation != previous.BindingGeneration {
				return ErrResetCounterConflict
			}
		}
		return nil
	}
	result, err := s.executor.ExecuteGuarded(ctx, idempotency.Request{
		TenantID: &tenantID, ActorKey: "device:" + identity.PublicID,
		Scope: "POST /device/v3/factory-reset", Key: idempotencyKey,
		RequestHash: idempotency.Hash(rawRequest), ExpiresAfter: 7 * 24 * time.Hour,
	}, guard, func(operationContext context.Context, tx pgx.Tx) (idempotency.Result, error) {
		response, resetErr := s.reset(operationContext, tx, identity, requestID, input)
		if resetErr != nil {
			return idempotency.Result{}, resetErr
		}
		body, marshalErr := json.Marshal(response)
		if marshalErr != nil {
			return idempotency.Result{}, marshalErr
		}
		resourceID := identity.PublicID
		return idempotency.Result{StatusCode: http.StatusOK, Body: body, ResourceID: &resourceID}, nil
	})
	if err != nil {
		return Response{}, false, err
	}
	var response Response
	if err := json.Unmarshal(result.Body, &response); err != nil {
		return Response{}, result.Replayed, fmt.Errorf("decode device factory reset response: %w", err)
	}
	return response, result.Replayed, nil
}

func (s *Service) reset(ctx context.Context, tx pgx.Tx, identity deviceauth.Identity, requestID string, input Input) (Response, error) {
	var activationState string
	var currentCounter int64
	if err := tx.QueryRow(ctx, `
		SELECT activation_state,factory_reset_counter
		FROM device
		WHERE id=$1::uuid AND tenant_id=$2::uuid AND status='registered'
		FOR UPDATE`, identity.ID, identity.TenantID).Scan(&activationState, &currentCounter); err != nil {
		return Response{}, fmt.Errorf("lock device for factory reset: %w", err)
	}
	if activationState != string(deviceactivation.Claimable) && activationState != string(deviceactivation.Bound) &&
		activationState != string(deviceactivation.Provisioning) && activationState != string(deviceactivation.Active) {
		return Response{}, ErrStateConflict
	}
	if input.ResetCounter <= currentCounter {
		return Response{}, ErrResetCounterConflict
	}
	if err := devicebinding.RequireUnboundForReset(ctx, tx, identity.ID); err != nil {
		return Response{}, err
	}

	var activeFirmware int
	if err := tx.QueryRow(ctx, `
		SELECT (SELECT count(*) FROM firmware_assignment
		WHERE device_id=$1::uuid AND tenant_id=$2::uuid AND status IN ('PENDING','ASSIGNED','COMMAND_QUEUED','DOWNLOADING','INSTALLING'))
		+ (SELECT count(*) FROM fleet_ota WHERE device_id=$1::uuid AND status IN ('pending','accepted'))`,
		identity.ID, identity.TenantID).Scan(&activeFirmware); err != nil {
		return Response{}, fmt.Errorf("check active firmware update: %w", err)
	}
	if activeFirmware > 0 {
		return Response{}, ErrFirmwareInProgress
	}
	// A physical-confirm device must not clear a live direct-call session.
	if identity.EnrollmentMode == "physical_confirm" {
		var busy bool
		if err := tx.QueryRow(ctx, `SELECT EXISTS(SELECT 1 FROM call_session WHERE device_id=$1 AND call_state IN ('PREPARING','STARTING','TALKING','STOPPING'))
			OR EXISTS(SELECT 1 FROM conversation WHERE device_id=$1 AND status NOT IN ('COMPLETED','FAILED','EXPIRED'))`, identity.ID).Scan(&busy); err != nil {
			return Response{}, err
		}
		if busy {
			return Response{}, ErrCallInProgress
		}
	}

	now := time.Now().UTC()
	endedConversations, err := endActiveConversations(ctx, tx, identity, requestID, now)
	if err != nil {
		return Response{}, err
	}

	var previousBindingID, previousHouseholdID, previousSubjectID string
	err = tx.QueryRow(ctx, `
		UPDATE device_binding
		SET unbound_at=$3,unbound_by=NULL,reason='device_factory_reset'
		WHERE device_id=$1::uuid AND tenant_id=$2::uuid AND unbound_at IS NULL
		RETURNING public_id::text,household_id::text,subject_id::text`, identity.ID, identity.TenantID, now).
		Scan(&previousBindingID, &previousHouseholdID, &previousSubjectID)
	bindingRemoved := err == nil
	if err != nil && !errors.Is(err, pgx.ErrNoRows) {
		return Response{}, fmt.Errorf("close device binding for factory reset: %w", err)
	}
	if bindingRemoved {
		if err := notifyPreviousOwner(ctx, tx, identity, previousHouseholdID, previousSubjectID, now); err != nil {
			return Response{}, err
		}
	}

	if _, err := tx.Exec(ctx, `
		UPDATE device_claim SET revoked_at=$3
		WHERE device_id=$1::uuid AND tenant_id=$2::uuid AND used_at IS NULL AND revoked_at IS NULL`,
		identity.ID, identity.TenantID, now); err != nil {
		return Response{}, fmt.Errorf("revoke old device claims: %w", err)
	}

	providerStatus := "NOT_PROVISIONED"
	if tag, err := tx.Exec(ctx, `
		UPDATE hardware_agent_provider_credential
		SET status='SUSPENDED',updated_at=$3
		WHERE device_id=$1::uuid AND tenant_id=$2::uuid AND status='ACTIVE'`, identity.ID, identity.TenantID, now); err != nil {
		return Response{}, fmt.Errorf("suspend provider credential: %w", err)
	} else if tag.RowsAffected() > 0 {
		providerStatus = "SUSPENDED"
	} else if err := tx.QueryRow(ctx, `SELECT status FROM hardware_agent_provider_credential WHERE device_id=$1::uuid`, identity.ID).Scan(&providerStatus); err != nil && !errors.Is(err, pgx.ErrNoRows) {
		return Response{}, fmt.Errorf("read provider credential status: %w", err)
	}
	if _, err := tx.Exec(ctx, `UPDATE hardware_agent_device_binding SET status='DISABLED',updated_at=$2 WHERE device_id=$1::uuid AND status='ACTIVE'`, identity.ID, now); err != nil {
		return Response{}, fmt.Errorf("disable hardware agent binding: %w", err)
	}

	var cancelledCommands int64
	if tag, err := tx.Exec(ctx, `
		UPDATE device_command
		SET status='CANCELLED',result_summary='{"error_code":"DEVICE_FACTORY_RESET"}'::jsonb,
			completed_at=$3,version=version+1,updated_at=$3
		WHERE device_id=$1::uuid AND tenant_id=$2::uuid AND status IN ('QUEUED','DELIVERED','ACKNOWLEDGED')`,
		identity.ID, identity.TenantID, now); err != nil {
		return Response{}, fmt.Errorf("cancel device commands: %w", err)
	} else {
		cancelledCommands = tag.RowsAffected()
	}

	if err := transitionToClaimable(ctx, tx, identity, requestID); err != nil {
		return Response{}, err
	}

	var configRevision int64
	if err := tx.QueryRow(ctx, `
		UPDATE device_twin
		SET desired_revision=desired_revision+1,desired_config='{}'::jsonb,reported_revision=NULL,reported_config='{}'::jsonb,
			capabilities='{}'::jsonb,updated_at=$3
		WHERE device_id=$1::uuid AND tenant_id=$2::uuid
		RETURNING desired_revision`, identity.ID, identity.TenantID, now).Scan(&configRevision); err != nil {
		return Response{}, fmt.Errorf("reset device twin: %w", err)
	}
	if _, err := tx.Exec(ctx, `
		UPDATE device SET factory_reset_counter=$3,version=version+1,updated_at=$4
		WHERE id=$1::uuid AND tenant_id=$2::uuid`, identity.ID, identity.TenantID, input.ResetCounter, now); err != nil {
		return Response{}, fmt.Errorf("record factory reset counter: %w", err)
	}

	var claim *Claim
	var mode string
	if err = tx.QueryRow(ctx, `SELECT enrollment_mode FROM device WHERE id=$1::uuid`, identity.ID).Scan(&mode); err != nil {
		return Response{}, err
	}
	if mode == "legacy" {
		value, e := issueClaim(ctx, tx, identity, now)
		if e != nil {
			return Response{}, e
		}
		claim = &value
	} else {
		_, err = tx.Exec(ctx, `UPDATE device SET binding_generation=binding_generation+1,onboarding_revision=onboarding_revision+1,onboarding_boot_id=NULL WHERE id=$1::uuid`, identity.ID)
		if err != nil {
			return Response{}, err
		}
		_, err = tx.Exec(ctx, `UPDATE device_setup_session SET closed_at=COALESCE(closed_at,now()),close_reason='FACTORY_RESET' WHERE device_id=$1::uuid`, identity.ID)
		if err != nil {
			return Response{}, err
		}
		_, err = tx.Exec(ctx, `UPDATE device_binding_claim SET status='CANCELLED',completed_at=now() WHERE device_id=$1::uuid AND status='PENDING'`, identity.ID)
		if err != nil {
			return Response{}, err
		}
		_, err = tx.Exec(ctx, `UPDATE onboarding_web_session SET revoked_at=COALESCE(revoked_at,now()) WHERE setup_id IN (SELECT id FROM device_setup_session WHERE device_id=$1::uuid)`, identity.ID)
		if err != nil {
			return Response{}, err
		}
	}
	response := Response{
		ResetID: input.ResetID, DeviceID: identity.PublicID, ActivationState: string(deviceactivation.Claimable),
		ResetCounter: input.ResetCounter, BindingRemoved: bindingRemoved, PreviousBindingID: previousBindingID,
		ProviderCredentialStatus: providerStatus, LicenseAction: "RETAINED",
		EndedConversationCount: endedConversations, CancelledCommandCount: cancelledCommands,
		ConfigRevision: configRevision, Claim: claim, ResetAt: now,
		LocalActions: []string{"clear_volc_credential", "clear_user_context", "clear_session_state", "clear_network_credentials", "reboot_to_claimable"},
	}
	if mode == "physical_confirm" {
		if err = tx.QueryRow(ctx, `SELECT binding_generation FROM device WHERE id=$1::uuid`, identity.ID).Scan(&response.BindingGeneration); err != nil {
			return Response{}, err
		}
	}
	actorID := identity.PublicID
	if err := s.audit.Record(ctx, tx, audit.Event{
		TenantID: &identity.TenantID, ActorType: "device", ActorPublicID: &actorID,
		Action: "device.factory_reset", ResourceType: "device", ResourceID: &actorID,
		RequestID: requestID, Result: "success", Detail: map[string]any{
			"reset_id": input.ResetID, "reset_counter": input.ResetCounter, "trigger": input.Trigger,
			"firmware_version": input.FirmwareVersion, "previous_binding_id": previousBindingID,
			"device_reported_conversation_id": input.ActiveConversationID,
			"ended_conversation_count":        endedConversations, "cancelled_command_count": cancelledCommands,
			"provider_credential_status": providerStatus, "license_action": "RETAINED",
		},
	}); err != nil {
		return Response{}, err
	}
	return response, nil
}

func notifyPreviousOwner(ctx context.Context, tx pgx.Tx, identity deviceauth.Identity, householdID, subjectID string, now time.Time) error {
	var wasOnline bool
	err := tx.QueryRow(ctx, `
		UPDATE device_presence_state SET online=false,changed_at=$4
		WHERE tenant_id=$1::uuid AND device_id=$2::uuid AND household_id=$3::uuid AND online
		RETURNING true`, identity.TenantID, identity.ID, householdID, now).Scan(&wasOnline)
	if err != nil && !errors.Is(err, pgx.ErrNoRows) {
		return fmt.Errorf("mark factory-reset device offline: %w", err)
	}
	if wasOnline {
		if _, err := tx.Exec(ctx, `SELECT append_web_event($1::uuid,$2::uuid,$3::uuid,'device.presence.changed','device',$4::uuid,'offline')`,
			identity.TenantID, householdID, subjectID, identity.PublicID); err != nil {
			return fmt.Errorf("publish factory-reset presence event: %w", err)
		}
	}
	if _, err := tx.Exec(ctx, `SELECT append_web_event($1::uuid,$2::uuid,$3::uuid,'device.binding.reset','device',$4::uuid,'CLAIMABLE')`,
		identity.TenantID, householdID, subjectID, identity.PublicID); err != nil {
		return fmt.Errorf("publish factory-reset binding event: %w", err)
	}
	return nil
}

func endActiveConversations(ctx context.Context, tx pgx.Tx, identity deviceauth.Identity, requestID string, now time.Time) (int, error) {
	rows, err := tx.Query(ctx, `
		SELECT id::text,status FROM conversation
		WHERE device_id=$1::uuid AND tenant_id=$2::uuid AND status NOT IN ('COMPLETED','FAILED','EXPIRED')
		FOR UPDATE`, identity.ID, identity.TenantID)
	if err != nil {
		return 0, fmt.Errorf("lock active conversations for factory reset: %w", err)
	}
	type activeConversation struct{ id, status string }
	items := make([]activeConversation, 0)
	for rows.Next() {
		var item activeConversation
		if err := rows.Scan(&item.id, &item.status); err != nil {
			rows.Close()
			return 0, err
		}
		items = append(items, item)
	}
	rows.Close()
	if err := rows.Err(); err != nil {
		return 0, err
	}
	for _, item := range items {
		if _, err := tx.Exec(ctx, `
			UPDATE conversation SET status='FAILED',ended_at=$2,stop_reason='device_factory_reset',
				failure_code='DEVICE_FACTORY_RESET',version=version+1,updated_at=$2
			WHERE id=$1::uuid`, item.id, now); err != nil {
			return 0, fmt.Errorf("fail conversation during factory reset: %w", err)
		}
		if _, err := tx.Exec(ctx, `
			INSERT INTO conversation_state_history(conversation_id,from_status,to_status,reason,source,request_id,occurred_at)
			VALUES($1::uuid,$2,'FAILED','device factory reset','device',NULLIF($3,''),$4)`, item.id, item.status, requestID, now); err != nil {
			return 0, fmt.Errorf("record factory reset conversation transition: %w", err)
		}
		if _, err := tx.Exec(ctx, `
			UPDATE hardware_agent_session SET status='FAILED',provider_ended_at=COALESCE(provider_ended_at,$2),updated_at=$2
			WHERE conversation_id=$1::uuid AND status NOT IN ('COMPLETED','FAILED','EXPIRED')`, item.id, now); err != nil {
			return 0, fmt.Errorf("fail hardware agent session during factory reset: %w", err)
		}
		if _, err := tx.Exec(ctx, `
			UPDATE rtc_task SET status='FAILED',updated_at=$2
			WHERE conversation_id=$1::uuid AND status NOT IN ('COMPLETED','FAILED')`, item.id, now); err != nil {
			return 0, fmt.Errorf("fail rtc task during factory reset: %w", err)
		}
	}
	return len(items), nil
}

func transitionToClaimable(ctx context.Context, tx pgx.Tx, identity deviceauth.Identity, requestID string) error {
	var stateText string
	if err := tx.QueryRow(ctx, `SELECT activation_state FROM device WHERE id=$1::uuid AND tenant_id=$2::uuid FOR UPDATE`, identity.ID, identity.TenantID).Scan(&stateText); err != nil {
		return fmt.Errorf("read device activation state after reset cleanup: %w", err)
	}
	state, err := deviceactivation.Parse(stateText)
	if err != nil {
		return ErrStateConflict
	}
	actorID := identity.PublicID
	apply := func(target deviceactivation.State, reason string) error {
		_, applyErr := deviceactivation.Apply(ctx, tx, deviceactivation.Input{
			DeviceID: identity.ID, TenantID: identity.TenantID, Target: target,
			Reason: reason, Source: "device-factory-reset", ActorType: "device",
			ActorPublicID: &actorID, RequestID: requestID,
		})
		return applyErr
	}
	if state == deviceactivation.Provisioning {
		if err := apply(deviceactivation.Bound, "设备恢复出厂，结束预配"); err != nil {
			return fmt.Errorf("return provisioning device to bound state: %w", err)
		}
		state = deviceactivation.Bound
	}
	if state != deviceactivation.Claimable {
		if state != deviceactivation.Bound && state != deviceactivation.Active && state != deviceactivation.Lost {
			return ErrStateConflict
		}
		if err := apply(deviceactivation.Claimable, "设备恢复出厂并解除用户绑定"); err != nil {
			return fmt.Errorf("make factory-reset device claimable: %w", err)
		}
	}
	return nil
}

func issueClaim(ctx context.Context, tx pgx.Tx, identity deviceauth.Identity, issuedAt time.Time) (Claim, error) {
	randomBytes := make([]byte, 10)
	if _, err := rand.Read(randomBytes); err != nil {
		return Claim{}, fmt.Errorf("generate factory reset claim: %w", err)
	}
	rawCode := base32.StdEncoding.WithPadding(base32.NoPadding).EncodeToString(randomBytes)
	code := strings.Join([]string{rawCode[0:4], rawCode[4:8], rawCode[8:12], rawCode[12:16]}, "-")
	digest := sha256.Sum256([]byte(code))
	expiresAt := issuedAt.Add(24 * time.Hour)
	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)`, identity.TenantID, identity.ID, hex.EncodeToString(digest[:]), expiresAt, issuedAt); err != nil {
		return Claim{}, fmt.Errorf("insert factory reset claim: %w", err)
	}
	return Claim{Code: code, IssuedAt: issuedAt, ExpiresAt: expiresAt}, nil
}