story/internal/onboarding/web.go

package onboarding

import (
	"context"
	"crypto/hmac"
	"crypto/sha256"
	"errors"
	"time"

	"data-terminal-cloud/internal/identity"
	"data-terminal-cloud/internal/saas"
	"github.com/jackc/pgx/v5"
)

type WebSession struct {
	Hash, SetupID, UserID, Issuer, Subject string
	Device                                 Device
	Expires                                time.Time
}
type BrowserCredentials struct {
	TrustedDeviceToken string         `json:"-"`
	Token              string         `json:"-"`
	CSRF               string         `json:"csrf_token"`
	AccountToken       string         `json:"-"`
	Data               map[string]any `json:"data"`
}

func (s *Service) webSession(ctx context.Context, q queryer, hash string) (WebSession, error) {
	var w WebSession
	var id string
	err := q.QueryRow(ctx, `SELECT ws.token_hash,ws.setup_id::text,COALESCE(ws.user_id::text,''),COALESCE(ws.issuer,''),COALESCE(ws.subject,''),ws.expires_at,ss.device_id::text FROM onboarding_web_session ws JOIN device_setup_session ss ON ss.id=ws.setup_id WHERE ws.token_hash=$1 AND ws.revoked_at IS NULL AND ws.expires_at>now()`, hash).Scan(&w.Hash, &w.SetupID, &w.UserID, &w.Issuer, &w.Subject, &w.Expires, &id)
	if errors.Is(err, pgx.ErrNoRows) {
		return w, fail(401, "WEB_SESSION_EXPIRED", "请从设备重新打开设置入口")
	}
	if err != nil {
		return w, err
	}
	w.Device, err = readDevice(ctx, q, id, false)
	return w, err
}
func (s *Service) checkWebActor(ctx context.Context, tx pgx.Tx, hash string, d Device) error {
	w, err := s.webSession(ctx, tx, hash)
	if err != nil {
		return err
	}
	if w.Device.ID != d.ID {
		return errAccess
	}
	setup, err := readSetup(ctx, tx, w.SetupID, d, false)
	if err != nil {
		return err
	}
	// A closed setup can report a completed binding, but cannot authorize writes.
	if setup.Closed != nil || !setup.Expires.After(s.now()) {
		return fail(410, "SETUP_EXPIRED", "设置入口已关闭，请从设备列表查看结果")
	}
	if w.UserID != "" {
		var active bool
		err = tx.QueryRow(ctx, `SELECT EXISTS(SELECT 1 FROM user_account WHERE id=$1::uuid AND tenant_id=$2::uuid AND status='active')`, w.UserID, d.TenantID).Scan(&active)
		if err != nil {
			return err
		}
		if !active {
			return errAccess
		}
	}
	return nil
}
func (s *Service) checkCSRF(ctx context.Context, hash, token string) error {
	if !validB64(token, 32) {
		return errAccess
	}
	var expected string
	err := s.pool.QueryRow(ctx, `SELECT csrf_hash FROM onboarding_web_session WHERE token_hash=$1 AND revoked_at IS NULL AND expires_at>now()`, hash).Scan(&expected)
	if err != nil || !hmac.Equal([]byte(expected), []byte(digest([]byte(token)))) {
		return errAccess
	}
	return nil
}

type ExchangeInput struct {
	Token    string `json:"token"`
	Verifier string `json:"verifier"`
}

func (s *Service) Exchange(ctx context.Context, in ExchangeInput) (BrowserCredentials, error) {
	var out BrowserCredentials
	if !validB64(in.Token, 32) || !validB64(in.Verifier, 32) {
		return out, errAccess
	}
	var deviceID string
	err := s.pool.QueryRow(ctx, `SELECT ss.device_id::text FROM device_handoff h JOIN device_setup_session ss ON ss.id=h.setup_id WHERE h.token_hash=$1`, digest([]byte(in.Token))).Scan(&deviceID)
	if errors.Is(err, pgx.ErrNoRows) {
		return out, errAccess
	}
	if err != nil {
		return out, err
	}
	tx, err := s.pool.Begin(ctx)
	if err != nil {
		return out, err
	}
	defer tx.Rollback(ctx)
	d, err := readDevice(ctx, tx, deviceID, true)
	if err != nil {
		return out, err
	}
	if !deviceAllowed(d) {
		return out, errAccess
	}
	var setupID, challenge string
	var expires time.Time
	var consumed *time.Time
	err = tx.QueryRow(ctx, `SELECT setup_id::text,code_challenge,expires_at,consumed_at FROM device_handoff WHERE token_hash=$1 FOR UPDATE`, digest([]byte(in.Token))).Scan(&setupID, &challenge, &expires, &consumed)
	if err != nil {
		return out, err
	}
	sum := sha256.Sum256([]byte(in.Verifier))
	if consumed != nil || !expires.After(s.now()) || !hmac.Equal([]byte(challenge), []byte(encode64(sum[:]))) {
		return out, fail(410, "HANDOFF_INVALID", "入口已失效，请从设备获取新入口")
	}
	setup, err := readSetup(ctx, tx, setupID, d, true)
	if err != nil {
		return out, err
	}
	if setup.Mode != "ENROLL" {
		return out, errAccess
	}
	st, err := s.state(ctx, tx, d)
	if err != nil {
		return out, err
	}
	if st.BindingState == "BOUND" {
		return out, errAccess
	}
	out.Token = randomToken(32)
	out.CSRF = randomToken(32)
	_, err = tx.Exec(ctx, `INSERT INTO onboarding_web_session(token_hash,setup_id,csrf_hash,expires_at) VALUES($1,$2::uuid,$3,$4)`, digest([]byte(out.Token)), setupID, digest([]byte(out.CSRF)), setup.Expires)
	if err != nil {
		return out, err
	}
	_, err = tx.Exec(ctx, `UPDATE device_handoff SET consumed_at=now() WHERE token_hash=$1`, digest([]byte(in.Token)))
	if err != nil {
		return out, err
	}
	out.Data = map[string]any{"setup_id": setupID, "device_id": d.FactoryID, "state": st, "expires_at_ms": setup.Expires.UnixMilli(), "sms_enabled": s.options.SMSLoginEnabled}
	return out, tx.Commit(ctx)
}
func (s *Service) attachPrincipal(ctx context.Context, tx pgx.Tx, w WebSession, d Device, p identity.Principal, allowSignup bool) (BrowserCredentials, error) {
	var out BrowserCredentials
	var user string
	if err := s.checkWebActor(ctx, tx, w.Hash, d); err != nil {
		return out, err
	}
	_, err := tx.Exec(ctx, `SELECT pg_advisory_xact_lock(hashtextextended($1,0))`, "account:"+d.TenantID+":"+p.Issuer+":"+p.Subject)
	if err != nil {
		return out, err
	}
	err = tx.QueryRow(ctx, `SELECT id::text FROM user_account WHERE tenant_id=$1::uuid AND oidc_issuer=$2 AND oidc_subject=$3 AND status='active' FOR SHARE`, d.TenantID, p.Issuer, p.Subject).Scan(&user)
	if errors.Is(err, pgx.ErrNoRows) {
		var tenantPublic string
		if err = tx.QueryRow(ctx, `SELECT public_id::text FROM tenant WHERE id=$1::uuid`, d.TenantID).Scan(&tenantPublic); err != nil {
			return out, err
		}
		allowed := false
		for _, id := range s.options.SelfSignupTenantIDs {
			if id == tenantPublic {
				allowed = true
			}
		}
		if !allowSignup || !allowed {
			return out, fail(403, "TENANT_MEMBERSHIP_REQUIRED", "请先由该租户管理员授权账号")
		}
		err = tx.QueryRow(ctx, `INSERT INTO user_account(tenant_id,oidc_issuer,oidc_subject,display_name,platform_role,account_kind) VALUES($1::uuid,$2,$3,$4,'customer','customer') RETURNING id::text`, d.TenantID, p.Issuer, p.Subject, p.DisplayName).Scan(&user)
	}
	if err != nil {
		return out, err
	}
	out.Token = randomToken(32)
	out.CSRF = randomToken(32)
	out.AccountToken = randomToken(32)
	_, err = tx.Exec(ctx, `INSERT INTO onboarding_web_session(token_hash,setup_id,csrf_hash,user_id,issuer,subject,expires_at) VALUES($1,$2::uuid,$3,$4::uuid,$5,$6,$7)`, digest([]byte(out.Token)), w.SetupID, digest([]byte(out.CSRF)), user, p.Issuer, p.Subject, w.Expires)
	if err != nil {
		return out, err
	}
	_, err = tx.Exec(ctx, `UPDATE onboarding_web_session SET revoked_at=now() WHERE token_hash=$1`, w.Hash)
	if err != nil {
		return out, err
	}
	_, err = tx.Exec(ctx, `INSERT INTO onboarding_account_session(token_hash,csrf_hash,user_id,expires_at) VALUES($1,$2,$3::uuid,$4)`, digest([]byte(out.AccountToken)), digest([]byte(out.CSRF)), user, s.now().Add(24*time.Hour))
	if err != nil {
		return out, err
	}
	out.Data = map[string]any{"authenticated": true, "setup_id": w.SetupID}
	return out, nil
}
func (s *Service) AuthLink(ctx context.Context, w WebSession, p identity.Principal) (BrowserCredentials, error) {
	tx, err := s.pool.Begin(ctx)
	if err != nil {
		return BrowserCredentials{}, err
	}
	defer tx.Rollback(ctx)
	d, err := readDevice(ctx, tx, w.Device.ID, true)
	if err != nil {
		return BrowserCredentials{}, err
	}
	out, err := s.attachPrincipal(ctx, tx, w, d, p, true)
	if err != nil {
		return out, err
	}
	return out, tx.Commit(ctx)
}

type ClaimInput struct {
	ElderID    string `json:"elder_id"`
	Generation int64  `json:"expected_binding_generation"`
}

func (s *Service) Claim(ctx context.Context, w WebSession, requestID string, raw []byte, in ClaimInput) (map[string]any, error) {
	if w.UserID == "" {
		return nil, fail(401, "LOGIN_REQUIRED", "请先登录")
	}
	if !uuidValid(in.ElderID) || in.Generation < 0 {
		return nil, fail(400, "VALIDATION_FAILED", "请选择使用者")
	}
	return s.mutate(ctx, w.Device, "web:"+w.Hash, "claim", requestID, raw, 120*time.Second, func(tx pgx.Tx, d Device) (map[string]any, error) {
		if err := s.checkWebActor(ctx, tx, w.Hash, d); err != nil {
			return nil, err
		}
		if d.Generation != in.Generation {
			return nil, errState
		}
		setup, err := readSetup(ctx, tx, w.SetupID, d, true)
		if err != nil {
			return nil, err
		}
		if setup.Mode != "ENROLL" {
			return nil, errAccess
		}
		var household, subject string
		err = tx.QueryRow(ctx, `SELECT id::text,household_id::text FROM subject_profile WHERE public_id=$1::uuid AND tenant_id=$2::uuid AND status='active'`, in.ElderID, d.TenantID).Scan(&subject, &household)
		if errors.Is(err, pgx.ErrNoRows) {
			return nil, errAccess
		}
		if err != nil {
			return nil, err
		}
		if err = claimPermission(ctx, tx, claimRecord{UserID: w.UserID, HouseholdID: household, SubjectID: subject, TenantID: d.TenantID}); err != nil {
			return nil, err
		}
		st, err := s.state(ctx, tx, d)
		if err != nil {
			return nil, err
		}
		if st.BindingState == "BOUND" {
			return nil, errState
		}
		return newClaim(ctx, s, tx, d, setup.ID, w.UserID, household, subject)
	})
}

type CancelInput struct {
	ClaimID    string `json:"claim_id"`
	Generation int64  `json:"expected_binding_generation"`
}

func (s *Service) Cancel(ctx context.Context, w WebSession, requestID string, raw []byte, in CancelInput) (map[string]any, error) {
	if !uuidV4(in.ClaimID) || in.Generation < 0 || w.UserID == "" {
		return nil, errAccess
	}
	return s.mutate(ctx, w.Device, "web:"+w.Hash, "cancel", requestID, raw, 24*time.Hour, func(tx pgx.Tx, d Device) (map[string]any, error) {
		if err := s.checkWebActor(ctx, tx, w.Hash, d); err != nil {
			return nil, err
		}
		c, err := readClaim(ctx, tx, in.ClaimID, d.ID)
		if err != nil {
			return nil, err
		}
		if c.UserID != w.UserID || c.SetupID != w.SetupID {
			return nil, errAccess
		}
		if err = claimPermission(ctx, tx, c); err != nil {
			return nil, err
		}
		if c.Status == "CONFIRMED" && d.Generation == c.Generation+1 {
			st, e := s.state(ctx, tx, d)
			return map[string]any{"already_bound": true, "state": st}, e
		}
		if d.Generation != in.Generation {
			return nil, errState
		}
		tag, err := tx.Exec(ctx, `UPDATE device_binding_claim SET status='CANCELLED',completed_at=now() WHERE id=$1::uuid AND status='PENDING'`, c.ID)
		if err != nil {
			return nil, err
		}
		if tag.RowsAffected() > 0 {
			if err = bump(ctx, tx, d.ID); err != nil {
				return nil, err
			}
		}
		st, err := s.state(ctx, tx, d)
		return map[string]any{"state": st}, err
	})
}
func (s *Service) WebStatus(ctx context.Context, w WebSession) (map[string]any, error) {
	tx, err := s.pool.Begin(ctx)
	if err != nil {
		return nil, err
	}
	defer tx.Rollback(ctx)
	d, err := readDevice(ctx, tx, w.Device.ID, true)
	if err != nil {
		return nil, err
	}
	if !deviceAllowed(d) {
		return nil, errAccess
	}
	setup, err := readSetup(ctx, tx, w.SetupID, d, false)
	if err != nil {
		return nil, err
	}
	st, err := s.state(ctx, tx, d)
	if err != nil {
		return nil, err
	}
	var binding *string
	var deviceSaved bool
	var mine *string
	if w.UserID != "" {
		err = tx.QueryRow(ctx, `SELECT b.public_id::text,EXISTS(SELECT 1 FROM device_state_ack a WHERE a.device_id=b.device_id AND a.binding_generation=b.generation AND a.boot_id=$3::uuid) FROM device_binding b JOIN household_member hm ON hm.household_id=b.household_id AND hm.user_id=$2::uuid AND hm.status='active' 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`, d.ID, w.UserID, d.BootID).Scan(&binding, &deviceSaved)
		if err != nil && !errors.Is(err, pgx.ErrNoRows) {
			return nil, err
		}
		err = tx.QueryRow(ctx, `SELECT id::text FROM device_binding_claim WHERE setup_id=$1::uuid AND user_id=$2::uuid AND status='PENDING' AND expires_at>now()`, w.SetupID, w.UserID).Scan(&mine)
		if err != nil && !errors.Is(err, pgx.ErrNoRows) {
			return nil, err
		}
	}
	// Never expose a different household's progress through an old handoff.
	if st.BindingState == "BOUND" && binding == nil {
		return nil, errAccess
	}
	return map[string]any{"setup_id": w.SetupID, "state": st, "binding_id": binding, "claim_id": mine, "device_saved": deviceSaved, "authenticated": w.UserID != "", "closed": setup.Closed != nil || !setup.Expires.After(s.now()), "sms_enabled": s.options.SMSLoginEnabled, "expires_at_ms": setup.Expires.UnixMilli()}, tx.Commit(ctx)
}

type WebAckInput struct {
	BindingID  string `json:"binding_id"`
	Generation int64  `json:"binding_generation"`
	Revision   int64  `json:"received_revision"`
}

func (s *Service) WebAck(ctx context.Context, w WebSession, requestID string, raw []byte, in WebAckInput) (map[string]any, error) {
	if !uuidValid(in.BindingID) || in.Generation < 1 || in.Revision < 1 || w.UserID == "" {
		return nil, errAccess
	}
	return s.mutate(ctx, w.Device, "web:"+w.Hash, "web-ack", requestID, raw, 24*time.Hour, func(tx pgx.Tx, d Device) (map[string]any, error) {
		var bindingID string
		err := tx.QueryRow(ctx, `SELECT b.id::text FROM device_binding b JOIN household_member hm ON hm.household_id=b.household_id AND hm.user_id=$3::uuid AND hm.status='active' JOIN user_account ua ON ua.id=hm.user_id AND ua.status='active' JOIN device_state_snapshot ss ON ss.device_id=b.device_id AND ss.revision=$5 AND ss.binding_generation=b.generation AND ss.state->>'binding_state'='BOUND' WHERE b.device_id=$1::uuid AND b.public_id=$2::uuid AND b.unbound_at IS NULL AND b.generation=$4`, d.ID, in.BindingID, w.UserID, in.Generation, in.Revision).Scan(&bindingID)
		if errors.Is(err, pgx.ErrNoRows) {
			return nil, errState
		}
		if err != nil {
			return nil, err
		}
		if d.Generation != in.Generation {
			return nil, errState
		}
		_, err = tx.Exec(ctx, `UPDATE device_setup_session SET web_ack_binding_id=$2::uuid,web_ack_generation=$3,web_ack_revision=$4,web_ack_at=now() WHERE id=$1::uuid`, w.SetupID, bindingID, in.Generation, in.Revision)
		return map[string]any{"accepted": true}, err
	})
}
func (s *Service) Profiles(ctx context.Context, w WebSession) (map[string]any, error) {
	if w.UserID == "" {
		return nil, fail(401, "LOGIN_REQUIRED", "请先登录")
	}
	rows, err := s.pool.Query(ctx, `SELECT h.public_id::text,h.name,sp.public_id::text,sp.preferred_name FROM household_member hm JOIN user_account ua ON ua.id=hm.user_id AND ua.status='active' JOIN household h ON h.id=hm.household_id AND h.tenant_id=hm.tenant_id AND h.status='active' LEFT JOIN subject_profile sp ON sp.household_id=h.id AND sp.tenant_id=h.tenant_id AND sp.status='active' WHERE hm.user_id=$1::uuid AND hm.tenant_id=$2::uuid AND hm.status='active' AND hm.role IN ('owner','admin','editor') ORDER BY h.name,sp.preferred_name`, w.UserID, w.Device.TenantID)
	if err != nil {
		return nil, err
	}
	defer rows.Close()
	items := []map[string]any{}
	for rows.Next() {
		var h, n string
		var id, name *string
		if err = rows.Scan(&h, &n, &id, &name); err != nil {
			return nil, err
		}
		items = append(items, map[string]any{"household_id": h, "household_name": n, "elder_id": id, "preferred_name": name})
	}
	return map[string]any{"items": items}, rows.Err()
}

type ProfileInput struct {
	InitialConsents saas.InitialConsentSelection `json:"initial_consents,omitempty"`
	HouseholdID     string                       `json:"household_id,omitempty"`
	HouseholdName   string                       `json:"household_name,omitempty"`
	Name            string                       `json:"preferred_name"`
}

func (s *Service) CreateProfile(ctx context.Context, w WebSession, requestID string, raw []byte, in ProfileInput) (map[string]any, error) {
	if err := saas.ValidateInitialConsents(in.InitialConsents); err != nil {
		return nil, fail(400, "VALIDATION_FAILED", "请逐项选择授权并确认用途")
	}
	if w.UserID == "" || !nonempty(in.Name, 120) || (in.HouseholdID == "" && !nonempty(in.HouseholdName, 200)) || (in.HouseholdID != "" && !uuidValid(in.HouseholdID)) {
		return nil, fail(400, "VALIDATION_FAILED", "家庭和使用者信息不正确")
	}
	return s.mutate(ctx, w.Device, "web:"+w.Hash, "profile", requestID, raw, 24*time.Hour, func(tx pgx.Tx, d Device) (map[string]any, error) {
		if err := s.checkWebActor(ctx, tx, w.Hash, d); err != nil {
			return nil, err
		}
		var household, public string
		var err error
		if in.HouseholdID == "" {
			err = tx.QueryRow(ctx, `INSERT INTO household(tenant_id,name,owner_user_id) VALUES($1::uuid,$2,$3::uuid) RETURNING id::text,public_id::text`, d.TenantID, in.HouseholdName, w.UserID).Scan(&household, &public)
			if err != nil {
				return nil, err
			}
			_, err = tx.Exec(ctx, `INSERT INTO household_member(tenant_id,household_id,user_id,role) VALUES($1::uuid,$2::uuid,$3::uuid,'owner')`, d.TenantID, household, w.UserID)
		} else {
			err = tx.QueryRow(ctx, `SELECT h.id::text,h.public_id::text FROM household h JOIN household_member hm ON hm.household_id=h.id AND hm.tenant_id=h.tenant_id WHERE h.public_id=$1::uuid AND h.tenant_id=$2::uuid AND h.status='active' AND hm.user_id=$3::uuid AND hm.status='active' AND hm.role IN ('owner','admin','editor') FOR SHARE OF h,hm`, in.HouseholdID, d.TenantID, w.UserID).Scan(&household, &public)
		}
		if errors.Is(err, pgx.ErrNoRows) {
			return nil, errAccess
		}
		if err != nil {
			return nil, err
		}
		var id string
		err = tx.QueryRow(ctx, `INSERT INTO subject_profile(tenant_id,household_id,preferred_name) VALUES($1::uuid,$2::uuid,$3) RETURNING public_id::text`, d.TenantID, household, in.Name).Scan(&id)
		if err != nil {
			return nil, err
		}
		if err = saas.InitializeSubjectMemory(ctx, tx, d.TenantID, id, w.UserID, requestID, in.InitialConsents); err != nil {
			return nil, err
		}
		if err = s.auditEvent(ctx, tx, d, "user", accountPublicID(ctx, tx, w.UserID), "onboarding.profile.create", requestID, map[string]any{"subject_id": id}); err != nil {
			return nil, err
		}
		return map[string]any{"elder_id": id, "household_id": public}, nil
	})
}
func accountPublicID(ctx context.Context, q queryer, id string) string {
	var p string
	_ = q.QueryRow(ctx, `SELECT public_id::text FROM user_account WHERE id=$1::uuid`, id).Scan(&p)
	return p
}