story/internal/onboarding/service.go

package onboarding

import (
	"context"
	"data-terminal-cloud/internal/devicefleet"
	"encoding/json"
	"errors"
	"fmt"
	"log/slog"
	"net/http"
	"net/url"
	"strings"
	"time"

	"data-terminal-cloud/internal/audit"
	"data-terminal-cloud/internal/deviceauth"
	"data-terminal-cloud/internal/hardwareprovision"
	"data-terminal-cloud/internal/identity"
	"data-terminal-cloud/internal/outbox"

	"github.com/jackc/pgx/v5"
	"github.com/jackc/pgx/v5/pgxpool"
	"github.com/redis/go-redis/v9"
)

type Error struct {
	Status                  int
	Code, Message, Recovery string
}

func (e *Error) Error() string   { return e.Code }
func (e *Error) HTTPStatus() int { return e.Status }
func fail(status int, code, message string) *Error {
	return &Error{Status: status, Code: code, Message: message, Recovery: "NONE"}
}

var errAuth = fail(401, "DEVICE_AUTHENTICATION_FAILED", "设备身份验证失败")
var errAccess = fail(403, "RESOURCE_ACCESS_DENIED", "当前账号或设备无权执行此操作")
var errState = fail(409, "BINDING_GENERATION_CONFLICT", "设备状态已变化，请重新同步")

type LegacyVerifier interface {
	Verify(*http.Request) (deviceauth.Identity, error)
}
type Options struct {
	SaaSCaptchaDisabled                    bool
	Captcha                                CaptchaProvider
	PauseClaims                            bool
	TrustedProxyCIDRs                      []string
	Enabled                                bool
	MasterKey, EntryURL, ProviderMasterKey string
	AllowLocalHTTP                         bool
	TokenVerifier                          identity.TokenVerifier
	LegacyVerifier                         LegacyVerifier
	Registrar                              hardwareprovision.Registrar
	// Compatibility harness only; production defaults to server registration.
	LegacyDeviceRegistration                                                   bool
	ActivationEnabled                                                          bool
	InstanceID, ProductKey, ProductSecret, BotID, Transport, MinimumSDKVersion string
	SMS                                                                        SMSProvider
	WeChat                                                                     WeChatProvider
	WeChatAppID, WeChatAppSecret                                               string
	SMSLoginEnabled                                                            bool
	SelfSignupTenantIDs                                                        []string
	Logger                                                                     *slog.Logger
}
type Service struct {
	Fleet         *devicefleet.Service
	operatorLogin bool
	pool          *pgxpool.Pool
	redis         *redis.Client
	crypt         cryptor
	providerCrypt cryptor
	options       Options
	now           func() time.Time
	audit         *audit.Repository
	outbox        *outbox.Store
}

func NewService(pool *pgxpool.Pool, redisClient *redis.Client, options Options) (*Service, error) {
	if pool == nil || redisClient == nil {
		return nil, errors.New("onboarding requires PostgreSQL and Redis")
	}
	crypt, err := newCryptor(options.MasterKey)
	if err != nil {
		return nil, err
	}
	entry, err := url.Parse(options.EntryURL)
	if err != nil || entry.Host == "" || entry.User != nil || entry.RawQuery != "" || entry.Fragment != "" || (entry.Scheme != "https" && !(options.AllowLocalHTTP && entry.Scheme == "http" && (entry.Hostname() == "localhost" || entry.Hostname() == "127.0.0.1"))) {
		return nil, errors.New("onboarding entry must be a fixed HTTPS URL")
	}
	if options.Logger == nil {
		options.Logger = slog.Default()
	}
	if options.WeChat == nil && options.WeChatAppID != "" && options.WeChatAppSecret != "" {
		options.WeChat = newWeChatClient(options.WeChatAppID, options.WeChatAppSecret)
	}
	if options.ActivationEnabled && (len(options.ProductSecret) < 16 || options.InstanceID == "" || options.ProductKey == "" || options.BotID == "") {
		return nil, errors.New("onboarding activation requires configured provider")
	}
	if options.SMSLoginEnabled && options.SMS == nil {
		return nil, errors.New("SMS login requires a configured sender")
	}
	var providerCrypt cryptor
	if options.ProviderMasterKey != "" {
		providerCrypt, err = newCryptor(options.ProviderMasterKey)
		if err != nil {
			return nil, err
		}
	}
	if options.ActivationEnabled && providerCrypt.aead == nil {
		return nil, errors.New("onboarding activation requires provider credential master key")
	}
	if options.ActivationEnabled && options.Registrar == nil && !options.LegacyDeviceRegistration {
		options.Registrar, err = hardwareprovision.NewHTTPRegistrar("", options.ProductSecret, 20*time.Second)
		if err != nil {
			return nil, err
		}
	}
	return &Service{pool: pool, redis: redisClient, crypt: crypt, providerCrypt: providerCrypt, options: options, now: time.Now, audit: audit.NewRepository(), outbox: outbox.NewStore(pool)}, nil
}
func (s *Service) Enabled() bool { return s != nil && s.options.Enabled }

type Device struct {
	ID, PublicID, TenantID, FactoryID, Model, HardwareRevision, Status, Lifecycle, Mode string
	Generation, Revision                                                                int64
	BootID                                                                              string
	Released                                                                            bool
	KeyVersion                                                                          int64
	Key                                                                                 []byte
}
type queryer interface {
	QueryRow(context.Context, string, ...any) pgx.Row
}

func readDevice(ctx context.Context, q queryer, id string, lock bool) (Device, error) {
	var d Device
	suffix := ""
	if lock {
		suffix = " FOR UPDATE OF d"
	}
	err := q.QueryRow(ctx, `SELECT d.id::text,d.public_id::text,d.tenant_id::text,COALESCE(d.factory_device_id,''),d.model,COALESCE(d.hardware_revision,''),d.status,d.activation_state,d.enrollment_mode,d.binding_generation,d.onboarding_revision,COALESCE(d.onboarding_boot_id::text,''),d.onboarding_released_at IS NOT NULL FROM device d JOIN tenant t ON t.id=d.tenant_id AND t.status='active' WHERE d.id=$1::uuid`+suffix, id).Scan(&d.ID, &d.PublicID, &d.TenantID, &d.FactoryID, &d.Model, &d.HardwareRevision, &d.Status, &d.Lifecycle, &d.Mode, &d.Generation, &d.Revision, &d.BootID, &d.Released)
	if errors.Is(err, pgx.ErrNoRows) {
		return d, errAccess
	}
	return d, err
}
func deviceAllowed(d Device) bool {
	return d.Mode == "physical_confirm" && d.Status == "registered" && d.Released && d.Lifecycle != "REVOKED" && d.Lifecycle != "LOST"
}

type State struct {
	DeviceID         string `json:"device_id"`
	PlatformDeviceID string `json:"-"`
	Revision         int64  `json:"device_revision"`
	Generation       int64  `json:"binding_generation"`
	BindingState     string `json:"binding_state"`
	ActivationState  string `json:"activation_state"`
	NextAction       string `json:"-"`
	Hash             string `json:"state_hash"`
}

func stateHash(st State) string {
	return digest([]byte(fmt.Sprintf("STATE1\n%s\n%d\n%d\n%s\n%s", st.DeviceID, st.Revision, st.Generation, st.BindingState, st.ActivationState)))
}
func (s *Service) state(ctx context.Context, tx pgx.Tx, d Device) (State, error) {
	if err := s.expireClaims(ctx, tx, d.ID); err != nil {
		return State{}, err
	}
	d, err := readDevice(ctx, tx, d.ID, true)
	if err != nil {
		return State{}, err
	}
	st := State{DeviceID: d.FactoryID, PlatformDeviceID: d.PublicID, Revision: d.Revision, Generation: d.Generation, BindingState: "UNBOUND", ActivationState: "NOT_STARTED", NextAction: "WAIT_BIND"}
	var bound, pending bool
	err = tx.QueryRow(ctx, `SELECT (EXISTS(SELECT 1 FROM device_binding WHERE device_id=$1::uuid AND unbound_at IS NULL) OR EXISTS(SELECT 1 FROM device_account_binding WHERE device_id=$1::uuid AND released_at IS NULL)),EXISTS(SELECT 1 FROM device_binding_claim WHERE device_id=$1::uuid AND status='PENDING' AND expires_at>now())`, d.ID).Scan(&bound, &pending)
	if err != nil {
		return st, err
	}
	err = tx.QueryRow(ctx, `SELECT status FROM device_activation_job WHERE device_id=$1::uuid`, d.ID).Scan(&st.ActivationState)
	if err != nil && !errors.Is(err, pgx.ErrNoRows) {
		return st, err
	}
	if pending {
		st.BindingState = "CLAIM_PENDING"
	}
	if bound {
		st.BindingState = "BOUND"
		st.NextAction = "RESUME_ACTIVATION"
	}
	if st.ActivationState == "READY" && bound {
		st.NextAction = "ENTER_READY"
	}
	if !deviceAllowed(d) {
		st.NextAction = "CONTACT_SUPPORT"
	}
	st.Hash = stateHash(st)
	raw, _ := json.Marshal(st)
	_, err = tx.Exec(ctx, `INSERT INTO device_state_snapshot(device_id,revision,binding_generation,state_hash,state) VALUES($1::uuid,$2,$3,$4,$5::jsonb) ON CONFLICT(device_id,revision) DO NOTHING`, d.ID, st.Revision, st.Generation, st.Hash, raw)
	if err != nil {
		return st, err
	}
	var stored string
	err = tx.QueryRow(ctx, `SELECT state_hash FROM device_state_snapshot WHERE device_id=$1::uuid AND revision=$2`, d.ID, st.Revision).Scan(&stored)
	if err == nil && stored != st.Hash {
		return st, errors.New("onboarding state changed without revision")
	}
	return st, err
}
func bump(ctx context.Context, tx pgx.Tx, deviceID string) error {
	_, err := tx.Exec(ctx, `UPDATE device SET onboarding_revision=onboarding_revision+1 WHERE id=$1::uuid`, deviceID)
	return err
}
func (s *Service) auditEvent(ctx context.Context, tx pgx.Tx, d Device, actorType, actorID, action, requestID string, detail map[string]any) error {
	return s.audit.Record(ctx, tx, audit.Event{TenantID: &d.TenantID, ActorType: actorType, ActorPublicID: &actorID, Action: action, ResourceType: "device", ResourceID: &d.PublicID, RequestID: requestID, Result: "success", Detail: detail})
}

// The device lock protects both replay validation and the mutation. Cached
// responses are encrypted and may never outlive their business authorization.
func (s *Service) mutate(ctx context.Context, d Device, actor, scope, requestID string, raw []byte, ttl time.Duration, op func(pgx.Tx, Device) (map[string]any, error)) (map[string]any, error) {
	if !uuidV4(requestID) {
		return nil, fail(400, "REQUEST_ID_REQUIRED", "X-Request-ID必须是UUID v4")
	}
	tx, err := s.pool.Begin(ctx)
	if err != nil {
		return nil, err
	}
	defer tx.Rollback(ctx)
	current, err := readDevice(ctx, tx, d.ID, true)
	if err != nil {
		return nil, err
	}
	if !deviceAllowed(current) {
		return nil, errAccess
	}
	if current.Generation != d.Generation {
		return nil, errState
	}
	if strings.HasPrefix(actor, "device:") && d.KeyVersion > 0 {
		var valid bool
		if err = tx.QueryRow(ctx, `SELECT EXISTS(SELECT 1 FROM device_onboarding_key WHERE device_id=$1::uuid AND key_version=$2 AND status='ACTIVE')`, d.ID, d.KeyVersion).Scan(&valid); err != nil {
			return nil, err
		}
		if !valid {
			return nil, errAuth
		}
		if scope != "register" && current.BootID != d.BootID {
			return nil, errState
		}
	}
	if strings.HasPrefix(actor, "web:") {
		if err = s.checkWebActor(ctx, tx, strings.TrimPrefix(actor, "web:"), current); err != nil {
			return nil, err
		}
	}
	requestHash := digest(raw)
	aad := actor + "\n" + scope + "\n" + requestID
	var stored string
	var generation int64
	var encrypted []byte
	var expires time.Time
	err = tx.QueryRow(ctx, `SELECT request_digest,binding_generation,response_ciphertext,expires_at FROM device_onboarding_request WHERE actor_key=$1 AND scope=$2 AND request_id=$3::uuid`, actor, scope, requestID).Scan(&stored, &generation, &encrypted, &expires)
	if err == nil {
		if stored != requestHash {
			return nil, fail(409, "REQUEST_BODY_CONFLICT", "相同请求号不能提交不同内容")
		}
		if generation != current.Generation {
			return nil, errState
		}
		if !expires.After(s.now()) {
			return nil, fail(410, "REQUEST_RESULT_EXPIRED", "该请求的短时结果已过期，请恢复当前流程")
		}
		if err = s.replayGuard(ctx, tx, current, actor, scope, raw); err != nil {
			return nil, err
		}
		plain, err := s.crypt.open(encrypted, aad)
		if err != nil {
			return nil, err
		}
		var result map[string]any
		if err = json.Unmarshal(plain, &result); err != nil {
			return nil, err
		}
		if err = s.replayResultGuard(ctx, tx, current, actor, scope, result); err != nil {
			return nil, err
		}
		// ACK persistence is idempotent, but close readiness changes as the
		// grace period elapses. Recompute it without moving accepted_at.
		if scope == "state-ack" {
			result["setup_can_close"], err = s.stateAckCanClose(ctx, tx, current)
			if err != nil {
				return nil, err
			}
		}
		if scope == "activation-result" || scope == "activation-ack" {
			st, e := s.state(ctx, tx, current)
			if e != nil {
				return nil, e
			}
			result["state"] = st
			if scope == "activation-result" {
				result["reconcile_required"] = st.ActivationState == "RECONCILE_REQUIRED"
			}
		}
		return result, tx.Commit(ctx)
	}
	if !errors.Is(err, pgx.ErrNoRows) {
		return nil, err
	}
	result, err := op(tx, current)
	if err != nil {
		return nil, err
	}
	current, err = readDevice(ctx, tx, d.ID, false)
	if err != nil {
		return nil, err
	}
	payload, err := json.Marshal(result)
	if err != nil {
		return nil, err
	}
	_, err = tx.Exec(ctx, `INSERT INTO device_onboarding_request(actor_key,scope,request_id,request_digest,device_id,binding_generation,response_ciphertext,expires_at) VALUES($1,$2,$3::uuid,$4,$5::uuid,$6,$7,$8)`, actor, scope, requestID, requestHash, d.ID, current.Generation, s.crypt.seal(payload, aad), s.now().Add(ttl))
	if err != nil {
		return nil, err
	}
	return result, tx.Commit(ctx)
}