story/internal/onboarding/inventory.go

package onboarding

import (
	"context"
	"crypto/rand"
	"crypto/rsa"
	"crypto/sha256"
	"crypto/x509"
	"encoding/base64"
	"encoding/json"
	"encoding/pem"
	"errors"
	"regexp"
	"strings"
	"time"

	"data-terminal-cloud/internal/deviceactivation"
	"data-terminal-cloud/internal/identity"
	"github.com/google/uuid"
	"github.com/jackc/pgx/v5"
)

func encode64(b []byte) string { return base64.RawURLEncoding.EncodeToString(b) }

type operator struct{ ID, PublicID, TenantID, Role string }

func operatorAccess(ctx context.Context, q queryer, principal identity.Principal, tenantPublicID string, write bool) (operator, error) {
	if !uuidValid(tenantPublicID) {
		return operator{}, errAccess
	}
	var a operator
	err := q.QueryRow(ctx, `SELECT ua.id::text,ua.public_id::text,t.id::text,ua.platform_role FROM user_account ua JOIN tenant t ON t.id=ua.tenant_id AND t.status='active' WHERE t.public_id=$1::uuid AND ua.oidc_issuer=$2 AND ua.oidc_subject=$3 AND ua.status='active' AND ua.account_kind='operator'`, tenantPublicID, principal.Issuer, principal.Subject).Scan(&a.ID, &a.PublicID, &a.TenantID, &a.Role)
	if errors.Is(err, pgx.ErrNoRows) {
		return a, errAccess
	}
	if err != nil {
		return a, err
	}
	if write && a.Role != "ops_admin" && a.Role != "ops_operator" {
		return a, errAccess
	}
	return a, nil
}

type InventoryInput struct {
	FactoryMAC       string `json:"factory_mac"`
	Model            string `json:"hardware_model"`
	HardwareRevision string `json:"hardware_revision"`
	BatchNo          string `json:"batch_no"`
}

func (s *Service) CreateInventory(ctx context.Context, principal identity.Principal, tenant, requestID, idempotencyKey string, raw []byte, input InventoryInput) (map[string]any, error) {
	if len(idempotencyKey) < 16 || len(idempotencyKey) > 160 || !regexp.MustCompile(`^[0-9A-F]{12}$`).MatchString(input.FactoryMAC) || len(input.Model) < 1 || len(input.Model) > 32 || len(input.HardwareRevision) < 1 || len(input.HardwareRevision) > 16 || len(input.BatchNo) < 1 || len(input.BatchNo) > 120 {
		return nil, fail(400, "VALIDATION_FAILED", "生产登记字段不正确")
	}
	tx, err := s.pool.Begin(ctx)
	if err != nil {
		return nil, err
	}
	defer tx.Rollback(ctx)
	a, err := operatorAccess(ctx, tx, principal, tenant, true)
	if err != nil {
		return nil, err
	}
	if _, err = tx.Exec(ctx, `SELECT pg_advisory_xact_lock(hashtextextended($1,0))`, "inventory-request:"+a.TenantID+":"+a.ID+":"+idempotencyKey); err != nil {
		return nil, err
	}
	if _, err = tx.Exec(ctx, `SELECT pg_advisory_xact_lock(hashtextextended($1,0))`, "inventory:"+input.FactoryMAC); err != nil {
		return nil, err
	}
	var jobID, deviceID, storedDigest string
	err = tx.QueryRow(ctx, `SELECT id::text,device_id::text,request_digest FROM device_enrollment_job WHERE tenant_id=$1::uuid AND actor_id=$2::uuid AND request_key=$3`, a.TenantID, a.ID, idempotencyKey).Scan(&jobID, &deviceID, &storedDigest)
	if err == nil {
		if storedDigest != digest(raw) {
			return nil, fail(409, "REQUEST_BODY_CONFLICT", "生产请求号已用于其他内容")
		}
		d, e := readDevice(ctx, tx, deviceID, false)
		if e != nil {
			return nil, e
		}
		d, e = s.makeInventoryClaimable(ctx, tx, d, jobID, requestID)
		if e != nil {
			return nil, e
		}
		return inventoryResult(d, jobID), tx.Commit(ctx)
	}
	if !errors.Is(err, pgx.ErrNoRows) {
		return nil, err
	}
	var existingTenant, model, revision string
	err = tx.QueryRow(ctx, `SELECT id::text,tenant_id::text,model,COALESCE(hardware_revision,'') FROM device WHERE factory_mac=$1 FOR UPDATE`, input.FactoryMAC).Scan(&deviceID, &existingTenant, &model, &revision)
	var d Device
	if err == nil {
		if existingTenant != a.TenantID || model != input.Model || revision != input.HardwareRevision {
			return nil, fail(409, "INVENTORY_CONFLICT", "设备已有不同的生产或租户记录")
		}
		d, err = readDevice(ctx, tx, deviceID, false)
		if err != nil {
			return nil, err
		}
		// Re-submitting an existing device also upgrades records created by the
		// former delivery/release flow into the direct waiting-for-binding flow.
		err = tx.QueryRow(ctx, `SELECT id::text FROM device_enrollment_job WHERE device_id=$1::uuid ORDER BY created_at DESC,id DESC LIMIT 1`, deviceID).Scan(&jobID)
		if err != nil {
			return nil, err
		}
		d, err = s.makeInventoryClaimable(ctx, tx, d, jobID, requestID)
		if err != nil {
			return nil, err
		}
		return inventoryResult(d, jobID), tx.Commit(ctx)
	}
	if !errors.Is(err, pgx.ErrNoRows) {
		return nil, err
	}
	d = Device{ID: uuid.NewString(), PublicID: uuid.NewString(), TenantID: a.TenantID, FactoryID: "KP01-" + input.FactoryMAC, Model: input.Model, HardwareRevision: input.HardwareRevision, Mode: "physical_confirm", Lifecycle: string(deviceactivation.Factory)}
	_, err = tx.Exec(ctx, `INSERT INTO device(id,public_id,tenant_id,serial_digest,model,hardware_revision,credential_ref,activation_state,activation_reason,enrollment_mode,factory_device_id,factory_mac) VALUES($1::uuid,$2::uuid,$3::uuid,$4,$5,$6,$7,'FACTORY','controlled-production','physical_confirm',$8,$9)`, d.ID, d.PublicID, d.TenantID, digest([]byte(d.FactoryID)), d.Model, d.HardwareRevision, "onboarding://"+d.FactoryID, d.FactoryID, input.FactoryMAC)
	if err != nil {
		return nil, err
	}
	_, err = tx.Exec(ctx, `INSERT INTO device_twin(device_id,tenant_id,desired_config) VALUES($1::uuid,$2::uuid,'{"voice_runtime":"volc_hardware_agent"}'::jsonb)`, d.ID, d.TenantID)
	if err != nil {
		return nil, err
	}
	_, err = tx.Exec(ctx, `INSERT INTO device_onboarding_key(device_id,tenant_id,key_version,ciphertext,status,activated_at) VALUES($1::uuid,$2::uuid,1,$3,'ACTIVE',now())`, d.ID, d.TenantID, s.crypt.seal(randomBytes(32), keyAAD(d.ID, 1)))
	if err != nil {
		return nil, err
	}
	err = tx.QueryRow(ctx, `INSERT INTO device_enrollment_job(device_id,tenant_id,actor_id,request_key,request_digest,batch_no,status,evidence_ref) VALUES($1::uuid,$2::uuid,$3::uuid,$4,$5,$6,'RELEASED','direct-create') RETURNING id::text`, d.ID, d.TenantID, a.ID, idempotencyKey, digest(raw), input.BatchNo).Scan(&jobID)
	if err != nil {
		return nil, err
	}
	d, err = s.makeInventoryClaimable(ctx, tx, d, jobID, requestID)
	if err != nil {
		return nil, err
	}
	if err = s.auditEvent(ctx, tx, d, "user", a.PublicID, "device.inventory.create", requestID, map[string]any{"factory_device_id": d.FactoryID, "enrollment_job_id": jobID}); err != nil {
		return nil, err
	}
	return inventoryResult(d, jobID), tx.Commit(ctx)
}

// makeInventoryClaimable finishes the internal credential setup in the same
// transaction as device creation. Operators no longer need to exchange an RSA
// envelope and perform a separate release before a device can wait for binding.
func (s *Service) makeInventoryClaimable(ctx context.Context, tx pgx.Tx, d Device, jobID, requestID string) (Device, error) {
	if d.Released {
		return d, nil
	}
	var version int64
	var keyStatus string
	if err := tx.QueryRow(ctx, `SELECT ej.key_version,k.status FROM device_enrollment_job ej JOIN device_onboarding_key k ON k.device_id=ej.device_id AND k.key_version=ej.key_version WHERE ej.id=$1::uuid AND ej.device_id=$2::uuid FOR UPDATE OF ej,k`, jobID, d.ID).Scan(&version, &keyStatus); err != nil {
		return d, err
	}
	if keyStatus == "PENDING" {
		if _, err := tx.Exec(ctx, `UPDATE device_onboarding_key SET status='REVOKED' WHERE device_id=$1::uuid AND status='ACTIVE' AND key_version<>$2`, d.ID, version); err != nil {
			return d, err
		}
		if _, err := tx.Exec(ctx, `UPDATE device_onboarding_key SET status='ACTIVE',activated_at=COALESCE(activated_at,now()) WHERE device_id=$1::uuid AND key_version=$2 AND status='PENDING'`, d.ID, version); err != nil {
			return d, err
		}
	} else if keyStatus != "ACTIVE" {
		return d, fail(409, "INVENTORY_CREDENTIAL_CLOSED", "设备凭据已经关闭，不能直接创建")
	}
	if d.Lifecycle == string(deviceactivation.Factory) {
		if err := requireLifecycle(ctx, tx, d, deviceactivation.Claimable, requestID); err != nil {
			return d, err
		}
	}
	if _, err := tx.Exec(ctx, `UPDATE device SET onboarding_released_at=COALESCE(onboarding_released_at,now()),onboarding_revision=onboarding_revision+1,version=version+1,updated_at=now() WHERE id=$1::uuid`, d.ID); err != nil {
		return d, err
	}
	if _, err := tx.Exec(ctx, `UPDATE device_enrollment_job SET status='RELEASED',evidence_ref=COALESCE(evidence_ref,'direct-create'),delivery_ciphertext=NULL,delivery_expires_at=NULL WHERE id=$1::uuid`, jobID); err != nil {
		return d, err
	}
	d.Released = true
	d.Lifecycle = string(deviceactivation.Claimable)
	d.KeyVersion = version
	return d, nil
}
func inventoryResult(d Device, job string) map[string]any {
	return map[string]any{"device_id": d.FactoryID, "platform_device_id": d.PublicID, "enrollment_job_id": job, "contract_version": ContractVersion, "released": d.Released}
}

type DeliveryInput struct {
	JobID       string `json:"enrollment_job_id"`
	ReceiverKey string `json:"receiver_public_key_pem"`
}

func (s *Service) DeliverCredential(ctx context.Context, p identity.Principal, tenant, publicID, requestID string, input DeliveryInput) (map[string]any, error) {
	if !uuidValid(publicID) || !uuidV4(input.JobID) {
		return nil, fail(400, "VALIDATION_FAILED", "交付任务编号不正确")
	}
	block, _ := pem.Decode([]byte(input.ReceiverKey))
	if block == nil {
		return nil, fail(400, "INVALID_RECEIVER_KEY", "需要RSA公钥")
	}
	parsed, err := x509.ParsePKIXPublicKey(block.Bytes)
	if err != nil {
		return nil, fail(400, "INVALID_RECEIVER_KEY", "需要PKIX格式的RSA公钥")
	}
	key, ok := parsed.(*rsa.PublicKey)
	if !ok || key.N.BitLen() < 2048 || key.N.BitLen() > 4096 || key.E != 65537 {
		return nil, fail(400, "INVALID_RECEIVER_KEY", "需要2048至4096位RSA公钥")
	}
	tx, err := s.pool.Begin(ctx)
	if err != nil {
		return nil, err
	}
	defer tx.Rollback(ctx)
	a, err := operatorAccess(ctx, tx, p, tenant, true)
	if err != nil {
		return nil, err
	}
	var id string
	err = tx.QueryRow(ctx, `SELECT id::text FROM device WHERE public_id=$1::uuid AND tenant_id=$2::uuid FOR UPDATE`, publicID, a.TenantID).Scan(&id)
	if err != nil {
		return nil, errAccess
	}
	d, err := readDevice(ctx, tx, id, false)
	if err != nil {
		return nil, err
	}
	var version int64
	var storedKey, status string
	var delivered []byte
	var expiry *time.Time
	err = tx.QueryRow(ctx, `SELECT key_version,COALESCE(receiver_key,''),delivery_ciphertext,delivery_expires_at,status FROM device_enrollment_job WHERE id=$1::uuid AND device_id=$2::uuid AND tenant_id=$3::uuid FOR UPDATE`, input.JobID, id, a.TenantID).Scan(&version, &storedKey, &delivered, &expiry, &status)
	if err != nil {
		return nil, errAccess
	}
	if storedKey != "" && storedKey != input.ReceiverKey {
		return nil, fail(409, "RECEIVER_CONFLICT", "同一交付任务不能更换接收方")
	}
	if status == "RELEASED" || status == "CANCELLED" {
		return nil, fail(409, "DELIVERY_CLOSED", "该交付任务已经关闭")
	}
	if len(delivered) > 0 && expiry != nil && expiry.After(s.now()) {
		var result map[string]any
		if err = json.Unmarshal(delivered, &result); err != nil {
			return nil, err
		}
		return result, tx.Commit(ctx)
	}
	var encrypted []byte
	err = tx.QueryRow(ctx, `SELECT ciphertext FROM device_onboarding_key WHERE device_id=$1::uuid AND key_version=$2 AND status='PENDING'`, id, version).Scan(&encrypted)
	if err != nil {
		return nil, err
	}
	secret, err := s.crypt.open(encrypted, keyAAD(id, version))
	if err != nil {
		return nil, err
	}
	// Envelope encryption supports a full manufacturing payload independently
	// of the RSA OAEP plaintext size. The receiver unwraps a random AES key.
	envelopeKey := randomBytes(32)
	ec, _ := newCryptor(fmtHex(envelopeKey))
	until := s.now().Add(15 * time.Minute)
	plain, _ := json.Marshal(map[string]any{"device_id": d.FactoryID, "platform_device_id": d.PublicID, "device_key_b64": encode64(secret), "key_version": version, "contract_version": ContractVersion, "expires_at_ms": until.UnixMilli()})
	wrapped, err := rsa.EncryptOAEP(sha256.New(), rand.Reader, key, envelopeKey, []byte(input.JobID))
	if err != nil {
		return nil, err
	}
	result := map[string]any{"algorithm": "RSA-OAEP-SHA256+A256GCM", "enrollment_job_id": input.JobID, "wrapped_key_b64": encode64(wrapped), "ciphertext_b64": encode64(ec.seal(plain, input.JobID)), "expires_at_ms": until.UnixMilli()}
	saved, _ := json.Marshal(result)
	_, err = tx.Exec(ctx, `UPDATE device_enrollment_job SET receiver_key=$2,delivery_ciphertext=$3,delivery_expires_at=$4,status='DELIVERED' WHERE id=$1::uuid`, input.JobID, input.ReceiverKey, saved, until)
	if err != nil {
		return nil, err
	}
	if err = s.auditEvent(ctx, tx, d, "user", a.PublicID, "device.credential.deliver", requestID, map[string]any{"job_id": input.JobID, "key_version": version}); err != nil {
		return nil, err
	}
	return result, tx.Commit(ctx)
}
func fmtHex(b []byte) string {
	const alphabet = "0123456789abcdef"
	out := make([]byte, len(b)*2)
	for i, v := range b {
		out[i*2] = alphabet[v>>4]
		out[i*2+1] = alphabet[v&15]
	}
	return string(out)
}

type ReleaseInput struct {
	JobID           string `json:"enrollment_job_id"`
	Evidence        string `json:"evidence_ref"`
	ExpectedVersion int64  `json:"expected_key_version"`
}

func (s *Service) ReleaseInventory(ctx context.Context, p identity.Principal, tenant, publicID, requestID string, in ReleaseInput) (map[string]any, error) {
	if !uuidValid(publicID) || !uuidV4(in.JobID) || len(strings.TrimSpace(in.Evidence)) < 3 || len(in.Evidence) > 500 || in.ExpectedVersion < 1 {
		return nil, fail(400, "VALIDATION_FAILED", "需要烧录校验记录和凭据版本")
	}
	tx, err := s.pool.Begin(ctx)
	if err != nil {
		return nil, err
	}
	defer tx.Rollback(ctx)
	a, err := operatorAccess(ctx, tx, p, tenant, true)
	if err != nil {
		return nil, err
	}
	var id string
	err = tx.QueryRow(ctx, `SELECT id::text FROM device WHERE public_id=$1::uuid AND tenant_id=$2::uuid AND enrollment_mode='physical_confirm' AND status='registered' AND activation_state NOT IN ('LOST','REVOKED') FOR UPDATE`, publicID, a.TenantID).Scan(&id)
	if err != nil {
		return nil, errAccess
	}
	d, err := readDevice(ctx, tx, id, false)
	if err != nil {
		return nil, err
	}
	var status string
	var version int64
	err = tx.QueryRow(ctx, `SELECT status,key_version FROM device_enrollment_job WHERE id=$1::uuid AND device_id=$2::uuid FOR UPDATE`, in.JobID, id).Scan(&status, &version)
	if err != nil {
		return nil, errAccess
	}
	if version != in.ExpectedVersion || (status != "DELIVERED" && status != "RELEASED") {
		return nil, fail(409, "ENROLLMENT_NOT_DELIVERED", "请先完成对应版本凭据交付")
	}
	if status == "RELEASED" {
		return inventoryResult(d, in.JobID), tx.Commit(ctx)
	}
	_, err = tx.Exec(ctx, `UPDATE device_onboarding_key SET status='REVOKED' WHERE device_id=$1::uuid AND status='ACTIVE'`, id)
	if err != nil {
		return nil, err
	}
	_, err = tx.Exec(ctx, `UPDATE device_onboarding_key SET status='ACTIVE',activated_at=now() WHERE device_id=$1::uuid AND key_version=$2 AND status='PENDING'`, id, version)
	if err != nil {
		return nil, err
	}
	if d.Lifecycle == "FACTORY" {
		if err = requireLifecycle(ctx, tx, d, deviceactivation.Claimable, requestID); err != nil {
			return nil, err
		}
	}
	_, err = tx.Exec(ctx, `UPDATE device SET onboarding_released_at=COALESCE(onboarding_released_at,now()),onboarding_revision=onboarding_revision+1,version=version+1,updated_at=now() WHERE id=$1::uuid`, id)
	if err != nil {
		return nil, err
	}
	_, err = tx.Exec(ctx, `UPDATE device_enrollment_job SET status='RELEASED',evidence_ref=$2,delivery_ciphertext=NULL WHERE id=$1::uuid`, in.JobID, in.Evidence)
	if err != nil {
		return nil, err
	}
	if err = s.auditEvent(ctx, tx, d, "user", a.PublicID, "device.inventory.release", requestID, map[string]any{"job_id": in.JobID, "key_version": version, "evidence_ref": in.Evidence}); err != nil {
		return nil, err
	}
	d.Released = true
	return inventoryResult(d, in.JobID), tx.Commit(ctx)
}