story/internal/onboarding/replay.go

package onboarding

import (
	"context"
	"encoding/json"
	"github.com/jackc/pgx/v5"
	"strings"
	"time"
)

func (s *Service) replayGuard(ctx context.Context, tx pgx.Tx, d Device, actor, scope string, raw []byte) error {
	if strings.HasPrefix(actor, "web:") {
		if err := s.checkWebActor(ctx, tx, strings.TrimPrefix(actor, "web:"), d); err != nil {
			return err
		}
	}
	switch scope {
	case "register":
		var in RegisterInput
		_ = json.Unmarshal(raw, &in)
		if d.BootID != in.BootID {
			return errState
		}
	case "setup":
		var in SetupInput
		_ = json.Unmarshal(raw, &in)
		var valid bool
		err := tx.QueryRow(ctx, `SELECT EXISTS(SELECT 1 FROM device_setup_session WHERE device_id=$1::uuid AND boot_id=$2::uuid AND local_setup_id=$3::uuid AND binding_generation=$4 AND closed_at IS NULL AND expires_at>now())`, d.ID, in.BootID, in.LocalSetupID, d.Generation).Scan(&valid)
		if err != nil {
			return err
		}
		if !valid {
			return fail(410, "SETUP_EXPIRED", "请创建新的恢复会话")
		}
	case "handoff":
		var in HandoffInput
		_ = json.Unmarshal(raw, &in)
		if _, err := readSetup(ctx, tx, in.SetupID, d, true); err != nil {
			return err
		}
		var valid bool
		err := tx.QueryRow(ctx, `SELECT EXISTS(SELECT 1 FROM device_handoff WHERE setup_id=$1::uuid AND code_challenge=$2 AND consumed_at IS NULL AND expires_at>now())`, in.SetupID, in.Challenge).Scan(&valid)
		if err != nil {
			return err
		}
		if !valid {
			return fail(410, "HANDOFF_INVALID", "入口已消费或失效，请重新获取")
		}
	case "confirm":
		var in ConfirmationInput
		_ = json.Unmarshal(raw, &in)
		c, err := readClaim(ctx, tx, in.ClaimID, d.ID)
		if err != nil {
			return err
		}
		if err = claimPermission(ctx, tx, c); err != nil {
			return err
		}
		if c.Status != "CONFIRMED" || c.PressID != in.PressEventID || d.Generation != c.Generation+1 {
			return errState
		}
		var valid bool
		err = tx.QueryRow(ctx, `SELECT EXISTS(SELECT 1 FROM device_binding WHERE id=$1::uuid AND generation=$2 AND unbound_at IS NULL)`, c.BindingID, d.Generation).Scan(&valid)
		if err != nil {
			return err
		}
		if !valid {
			return errState
		}
	case "claim":
		var in ClaimInput
		_ = json.Unmarshal(raw, &in)
		w, err := s.webSession(ctx, tx, strings.TrimPrefix(actor, "web:"))
		if err != nil {
			return err
		}
		var c claimRecord
		c.UserID = w.UserID
		c.TenantID = d.TenantID
		err = tx.QueryRow(ctx, `SELECT id::text,household_id::text FROM subject_profile WHERE public_id=$1::uuid AND tenant_id=$2::uuid`, in.ElderID, d.TenantID).Scan(&c.SubjectID, &c.HouseholdID)
		if err != nil {
			return errAccess
		}
		if err = claimPermission(ctx, tx, c); err != nil {
			return err
		}
		var valid bool
		err = tx.QueryRow(ctx, `SELECT EXISTS(SELECT 1 FROM device_binding_claim WHERE setup_id=$1::uuid AND device_id=$2::uuid AND user_id=$3::uuid AND subject_id=$4::uuid AND status='PENDING' AND expires_at>now())`, w.SetupID, d.ID, w.UserID, c.SubjectID).Scan(&valid)
		if err != nil {
			return err
		}
		if !valid {
			return fail(410, "CLAIM_EXPIRED", "请查询当前绑定进度")
		}
	case "attempt":
		var in AttemptInput
		_ = json.Unmarshal(raw, &in)
		j, err := readJob(ctx, tx, d.ID)
		if err != nil {
			return err
		}
		if err = validConfig(s, d, j, in.ActivationID, in.ConfigVersion); err != nil {
			return err
		}
		var status string
		var expiry time.Time
		var conversation *string
		err = tx.QueryRow(ctx, `SELECT status,permit_expires_at,conversation_id::text FROM device_activation_attempt WHERE id=$1::uuid AND activation_id=$2::uuid`, in.AttemptID, j.ID).Scan(&status, &expiry, &conversation)
		if err != nil {
			return err
		}
		if status != "CONTACT_POSSIBLE" || conversation != nil || !expiry.After(s.now()) {
			return fail(409, "ATTEMPT_ALREADY_ISSUED", "请恢复原尝试的结果")
		}
	case "activation-ack":
		var in ActivationAckInput
		_ = json.Unmarshal(raw, &in)
		j, err := readJob(ctx, tx, d.ID)
		if err != nil {
			return err
		}
		return validConfig(s, d, j, in.ActivationID, in.ConfigVersion)
	case "activation-result":
		var in ResultInput
		_ = json.Unmarshal(raw, &in)
		j, err := readJob(ctx, tx, d.ID)
		if err != nil {
			return err
		}
		var version int64
		err = tx.QueryRow(ctx, `SELECT config_version FROM device_activation_attempt WHERE id=$1::uuid AND activation_id=$2::uuid`, in.AttemptID, j.ID).Scan(&version)
		if err != nil {
			return err
		}
		if j.ID != in.ActivationID || j.Generation != d.Generation || j.ConfigVersion != version || j.Status == "BLOCKED" {
			return errState
		}
		return nil
	case "profile":
		var in ProfileInput
		_ = json.Unmarshal(raw, &in)
		if in.HouseholdID != "" {
			w, err := s.webSession(ctx, tx, strings.TrimPrefix(actor, "web:"))
			if err != nil {
				return err
			}
			var allowed bool
			err = tx.QueryRow(ctx, `SELECT EXISTS(SELECT 1 FROM household h JOIN household_member hm ON hm.household_id=h.id WHERE h.public_id=$1::uuid AND h.status='active' AND hm.user_id=$2::uuid AND hm.status='active' AND hm.role IN ('owner','admin','editor'))`, in.HouseholdID, w.UserID).Scan(&allowed)
			if err != nil {
				return err
			}
			if !allowed {
				return errAccess
			}
		}
	}
	return nil
}