story/internal/orchestration/onboarding.go

package orchestration

import (
	"context"
	"crypto/sha256"
	"data-terminal-cloud/internal/deviceauth"
	"data-terminal-cloud/internal/idempotency"
	"encoding/hex"
	"errors"
	"github.com/google/uuid"
	"github.com/jackc/pgx/v5"
	"time"
)

var ErrOnboardingState = errors.New("onboarding state or test permit is no longer valid")

func onboardingCreateGuard(identity deviceauth.Identity, input HardwareAgentCreateInput) idempotency.Guard {
	return func(ctx context.Context, tx pgx.Tx, cached *idempotency.Result) error {
		var epoch int64
		epochErr := tx.QueryRow(ctx, `SELECT us.privacy_epoch FROM memory_user_state us JOIN device_binding b ON b.subject_id=us.subject_id WHERE b.device_id=$1::uuid AND b.unbound_at IS NULL FOR UPDATE OF us`, identity.ID).Scan(&epoch)
		if epochErr != nil && !errors.Is(epochErr, pgx.ErrNoRows) {
			return epochErr
		}
		if epoch > 0 {
			return ErrCallContractRequired
		}
		var mode, status, lifecycle string
		var generation int64
		err := tx.QueryRow(ctx, `SELECT enrollment_mode,status,activation_state,binding_generation FROM device WHERE id=$1::uuid AND tenant_id=$2::uuid FOR UPDATE`, identity.ID, identity.TenantID).Scan(&mode, &status, &lifecycle, &generation)
		if err != nil {
			return err
		}
		if mode == "legacy" {
			if input.Intent != "" || input.ActivationID != "" || input.AttemptID != "" || input.PermitToken != "" {
				return ErrOnboardingState
			}
			return nil
		}
		if identity.EnrollmentMode != "physical_confirm" || generation != identity.BindingGeneration || status != "registered" || lifecycle == "REVOKED" || lifecycle == "LOST" {
			return ErrOnboardingState
		}
		var allowed bool
		err = tx.QueryRow(ctx, `SELECT true FROM device_binding b JOIN household h ON h.id=b.household_id AND h.status='active' JOIN subject_profile sp ON sp.id=b.subject_id AND sp.status='active' JOIN household_member hm ON hm.household_id=h.id AND hm.user_id=b.bound_by AND hm.status='active' AND hm.role IN ('owner','admin','editor') JOIN user_account ua ON ua.id=hm.user_id AND ua.status='active' WHERE b.device_id=$1::uuid AND b.unbound_at IS NULL AND b.generation=$2 FOR SHARE OF b,h,sp,hm,ua`, identity.ID, generation).Scan(&allowed)
		if errors.Is(err, pgx.ErrNoRows) {
			return ErrOnboardingState
		}
		if err != nil {
			return err
		}
		if input.Intent == "" {
			if input.ActivationID != "" || input.AttemptID != "" || input.PermitToken != "" {
				return ErrOnboardingState
			}
			err = tx.QueryRow(ctx, `SELECT EXISTS(SELECT 1 FROM device_activation_job WHERE device_id=$1::uuid AND binding_generation=$2 AND status='READY' AND config_ack_at IS NOT NULL AND config_expires_at>now())`, identity.ID, generation).Scan(&allowed)
		} else {
			if input.Intent != "activation_test" {
				return ErrOnboardingState
			}
			if _, e := uuid.Parse(input.ActivationID); e != nil {
				return ErrOnboardingState
			}
			if _, e := uuid.Parse(input.AttemptID); e != nil {
				return ErrOnboardingState
			}
			hash := sha256.Sum256([]byte(input.PermitToken))
			var attached *string
			err = tx.QueryRow(ctx, `SELECT a.conversation_id::text FROM device_activation_job j JOIN device_activation_attempt a ON a.activation_id=j.id WHERE j.id=$1::uuid AND j.device_id=$2::uuid AND a.id=$3::uuid AND j.binding_generation=$4 AND a.binding_generation=$4 AND a.config_version=j.config_version AND j.status IN ('REGISTERED_OBSERVED','CONTACT_POSSIBLE','READY') AND j.config_ack_at IS NOT NULL AND j.engine_ready_at IS NOT NULL AND j.config_expires_at>now() AND a.stage='CONVERSATION_TEST' AND a.executor='DEVICE' AND a.status='CONTACT_POSSIBLE' AND (a.permit_hash IS NULL OR a.permit_hash=$5) AND (a.permit_expires_at>now() OR a.conversation_id IS NOT NULL) FOR UPDATE OF j,a`, input.ActivationID, identity.ID, input.AttemptID, generation, hex.EncodeToString(hash[:])).Scan(&attached)
			if errors.Is(err, pgx.ErrNoRows) {
				return ErrOnboardingState
			}
			if err != nil {
				return err
			}
			allowed = true
			if cached != nil && (cached.ResourceID == nil || attached == nil || *cached.ResourceID != *attached) {
				return ErrOnboardingState
			}
		}
		if err != nil {
			return err
		}
		if !allowed {
			return ErrOnboardingState
		}
		if cached != nil {
			if cached.ResourceID == nil {
				return ErrOnboardingState
			}
			err = tx.QueryRow(ctx, `SELECT EXISTS(SELECT 1 FROM conversation WHERE id=$1::uuid AND device_id=$2::uuid AND onboarding_generation=$3 AND status IN ('ROOM_ISSUED','ACTIVE') AND (activation_test_deadline IS NULL OR activation_test_deadline>now()))`, *cached.ResourceID, identity.ID, generation).Scan(&allowed)
			if err != nil {
				return err
			}
			if !allowed {
				return ErrOnboardingState
			}
		}
		return nil
	}
}
func attachOnboardingConversation(ctx context.Context, tx pgx.Tx, identity deviceauth.Identity, input HardwareAgentCreateInput, conversation string) error {
	if identity.EnrollmentMode != "physical_confirm" {
		return nil
	}
	if input.Intent == "activation_test" {
		tag, err := tx.Exec(ctx, `UPDATE device_activation_attempt SET conversation_id=$2::uuid WHERE id=$1::uuid AND conversation_id IS NULL AND permit_expires_at>now() AND status='CONTACT_POSSIBLE'`, input.AttemptID, conversation)
		if err != nil {
			return err
		}
		if tag.RowsAffected() != 1 {
			return ErrOnboardingState
		}
		_, err = tx.Exec(ctx, `UPDATE conversation SET onboarding_generation=$2,activation_attempt_id=$3::uuid,activation_test_deadline=now()+interval '60 seconds',allocation_expires_at=LEAST(allocation_expires_at,now()+interval '60 seconds') WHERE id=$1::uuid`, conversation, identity.BindingGeneration, input.AttemptID)
		return err
	}
	_, err := tx.Exec(ctx, `UPDATE conversation SET onboarding_generation=$2 WHERE id=$1::uuid`, conversation, identity.BindingGeneration)
	return err
}
func guardOnboardingEvent(ctx context.Context, tx pgx.Tx, identity deviceauth.Identity, conversation string, event HardwareAgentEventInput) error {
	if identity.EnrollmentMode != "physical_confirm" {
		return nil
	}
	var generation int64
	var status string
	if err := tx.QueryRow(ctx, `SELECT binding_generation,status FROM device WHERE id=$1::uuid FOR UPDATE`, identity.ID).Scan(&generation, &status); err != nil {
		return err
	}
	if generation != identity.BindingGeneration || status != "registered" {
		return ErrOnboardingState
	}
	var stored *int64
	var deadline *time.Time
	err := tx.QueryRow(ctx, `SELECT onboarding_generation,activation_test_deadline FROM conversation WHERE public_id=$1::uuid AND device_id=$2::uuid`, conversation, identity.ID).Scan(&stored, &deadline)
	if err != nil {
		return err
	}
	if stored == nil || *stored != generation {
		return ErrOnboardingState
	}
	if deadline != nil && deadline.Before(time.Now()) && event.EventType != "session_ended" && event.EventType != "session_failed" {
		return ErrOnboardingState
	}
	return nil
}