story/cmd/sim-device/main.go

package main

import (
	"bytes"
	"context"
	"crypto/rand"
	"crypto/sha256"
	"encoding/binary"
	"encoding/hex"
	"encoding/json"
	"errors"
	"fmt"
	"io"
	"net/http"
	"net/url"
	"os"
	"strconv"
	"strings"
	"time"

	"data-terminal-cloud/internal/deviceauth"
)

type simulator struct {
	baseURL, deviceID string
	secret            []byte
	client            *http.Client
}

type apiResult struct {
	Status int
	Header http.Header
	Body   []byte
}

type envelope struct {
	Data json.RawMessage `json:"data"`
}

type createdConversation struct {
	ID  string `json:"conversation_id"`
	RTC struct {
		RoomID    string    `json:"room_id"`
		UserID    string    `json:"user_id"`
		ExpiresAt time.Time `json:"expires_at"`
	} `json:"rtc"`
}

type deviceConversationState struct {
	Status      string `json:"status"`
	Action      string `json:"action"`
	FailureCode string `json:"failure_code"`
}

type uploadBatch struct {
	ID       string `json:"id"`
	Status   string `json:"status"`
	Segments []struct {
		ID               string     `json:"id"`
		Status           string     `json:"status"`
		UploadURL        string     `json:"upload_url"`
		ContentType      string     `json:"content_type"`
		LastErrorCode    string     `json:"last_error_code"`
		LocalDeleteAfter *time.Time `json:"local_delete_after"`
	} `json:"segments"`
}

func main() {
	if err := run(context.Background()); err != nil {
		fmt.Fprintln(os.Stderr, "模拟设备失败：", err)
		os.Exit(1)
	}
}

func run(ctx context.Context) error {
	secret := []byte(env("SIM_DEVICE_SECRET", "local-simulated-device-secret-32-bytes-minimum"))
	if len(secret) < 32 {
		return errors.New("SIM_DEVICE_SECRET must contain at least 32 characters")
	}
	sim := &simulator{
		baseURL:  strings.TrimRight(env("SIM_API_URL", "http://localhost:58080"), "/"),
		deviceID: env("SIM_DEVICE_ID", "10000000-0000-0000-0000-000000000511"),
		secret:   secret, client: &http.Client{Timeout: 15 * time.Second},
	}
	scenarios := parseScenarios(env("SIM_SCENARIO", "bootstrap"))
	if scenarios["offline"] {
		if err := sim.offlineRetry(ctx); err != nil {
			return err
		}
	}
	if scenarios["bootstrap"] || scenarios["conversation"] || scenarios["all"] {
		if err := sim.bootstrap(ctx); err != nil {
			return err
		}
	}
	conversationID := strings.TrimSpace(os.Getenv("SIM_CONVERSATION_ID"))
	if scenarios["conversation"] || scenarios["all"] {
		id, err := sim.conversation(ctx)
		if err != nil {
			return err
		}
		conversationID = id
	}
	if scenarios["transcript"] || scenarios["all"] {
		if conversationID == "" {
			return errors.New("transcript scenario requires conversation or SIM_CONVERSATION_ID")
		}
		if err := sim.transcript(ctx, conversationID); err != nil {
			return err
		}
	}
	if scenarios["telemetry"] || scenarios["all"] {
		if err := sim.telemetry(ctx, conversationID); err != nil {
			return err
		}
	}
	if scenarios["commands"] || scenarios["all"] {
		if err := sim.commands(ctx); err != nil {
			return err
		}
	}
	if scenarios["media"] {
		if conversationID == "" {
			return errors.New("media scenario requires SIM_CONVERSATION_ID")
		}
		if err := sim.media(ctx, conversationID); err != nil {
			return err
		}
	}
	return nil
}

func (s *simulator) bootstrap(ctx context.Context) error {
	query := url.Values{
		"firmware_version": {env("SIM_FIRMWARE_VERSION", "0.2.0-sim")}, "board_rev": {"sim-rev-a"}, "rtc_sdk_version": {"mock-rtc-1"},
		"codec": {"opus"}, "sample_rate": {"16000"}, "network": {"simulated"},
		"device_time": {time.Now().UTC().Format(time.RFC3339)},
	}
	first, err := s.request(ctx, http.MethodGet, "/device/v2/bootstrap?"+query.Encode(), nil, "")
	if err != nil {
		return err
	}
	var bootstrap struct {
		Device struct {
			FirmwareCompatible     bool    `json:"firmware_compatible"`
			MinimumFirmwareVersion *string `json:"minimum_firmware_version"`
		} `json:"device"`
	}
	if err := decodeData(first.Body, &bootstrap); err != nil {
		return err
	}
	if !bootstrap.Device.FirmwareCompatible {
		minimum := "当前策略"
		if bootstrap.Device.MinimumFirmwareVersion != nil {
			minimum = *bootstrap.Device.MinimumFirmwareVersion
		}
		return fmt.Errorf("firmware update required before conversation; minimum=%s", minimum)
	}
	second, err := s.request(ctx, http.MethodGet, "/device/v2/config", nil, "")
	if err != nil {
		return err
	}
	fmt.Printf("bootstrap HTTP %d; config HTTP %d\n", first.Status, second.Status)
	return nil
}

func (s *simulator) offlineRetry(ctx context.Context) error {
	offline := *s
	offline.baseURL = "http://127.0.0.1:1"
	offline.client = &http.Client{Timeout: 250 * time.Millisecond}
	if _, err := offline.request(ctx, http.MethodGet, "/device/v2/config", nil, ""); err == nil {
		return errors.New("offline simulation unexpectedly reached a server")
	}
	fmt.Println("offline request retained locally; reconnecting with a new nonce")
	_, err := s.request(ctx, http.MethodGet, "/device/v2/config", nil, "")
	return err
}

func (s *simulator) conversation(ctx context.Context) (string, error) {
	createBody := jsonBody(map[string]any{
		"device_time": time.Now().UTC(), "monotonic_ms": 1000,
		"capabilities": map[string]any{
			"rtc": true, "audio_tracks": []string{"near", "far"}, "upload_resume": true,
			"rtc_credential_renewal": true,
		},
		"network": map[string]any{"type": "simulated", "rssi": -48}, "requested_goal": "open_conversation",
	})
	createKey := "sim-create-" + nonce()
	createdResponse, err := s.request(ctx, http.MethodPost, "/device/v2/conversations", createBody, createKey)
	if err != nil {
		return "", err
	}
	var created createdConversation
	if err := decodeData(createdResponse.Body, &created); err != nil {
		return "", err
	}
	replay, err := s.request(ctx, http.MethodPost, "/device/v2/conversations", createBody, createKey)
	if err != nil || replay.Header.Get("Idempotency-Replayed") != "true" {
		return "", fmt.Errorf("create replay failed: %w", err)
	}
	activate := jsonBody(map[string]any{
		"joined_at": time.Now().UTC(), "room_id": created.RTC.RoomID, "rtc_user_id": created.RTC.UserID,
		"rtc_join_result": "success", "config_revision": 1, "connection_reference": "sim-device",
	})
	activationPath := "/device/v2/conversations/" + created.ID + "/activate"
	if _, err := s.request(ctx, http.MethodPost, activationPath, activate, "sim-activate-"+nonce()); err != nil {
		return "", err
	}
	if err := s.waitForConversationActive(ctx, created.ID, 15*time.Second); err != nil {
		return "", err
	}
	heartbeat := jsonBody(map[string]any{
		"sequence_no": 1, "device_time": time.Now().UTC(), "monotonic_ms": 2000, "conversation_status": "ACTIVE",
		"rtc": map[string]any{"joined": true, "uplink": true, "downlink": true}, "network": map[string]any{"type": "simulated", "rssi": -50},
		"firmware_version": "0.2.0-sim", "reported_config_revision": 1,
		"rtc_credential_expires_at": created.RTC.ExpiresAt,
		"recording_queue":           map[string]any{"pending_segments": 2, "failed_segments": 0}, "command_results": []any{},
	})
	if _, err := s.request(ctx, http.MethodPost, "/device/v2/conversations/"+created.ID+"/heartbeat", heartbeat, ""); err != nil {
		return "", err
	}
	stop := jsonBody(map[string]any{
		"reason": "simulated_hangup", "device_time": time.Now().UTC(), "last_heartbeat_sequence": 1,
		"rtc_left": true, "recording": map[string]any{"near_segments": 1, "far_segments": 1},
	})
	stopKey := "sim-stop-" + nonce()
	if _, err := s.request(ctx, http.MethodPost, "/device/v2/conversations/"+created.ID+"/stop", stop, stopKey); err != nil {
		return "", err
	}
	replay, err = s.request(ctx, http.MethodPost, "/device/v2/conversations/"+created.ID+"/stop", stop, stopKey)
	if err != nil || replay.Header.Get("Idempotency-Replayed") != "true" {
		return "", fmt.Errorf("stop replay failed: %w", err)
	}
	time.Sleep(2 * time.Second)
	fmt.Println("conversation and repeated requests completed:", created.ID)
	return created.ID, nil
}

func (s *simulator) waitForConversationActive(ctx context.Context, conversationID string, timeout time.Duration) error {
	deadline := time.Now().Add(timeout)
	for {
		statusResponse, err := s.request(ctx, http.MethodGet, "/device/v2/conversations/"+conversationID, nil, "")
		if err != nil {
			return err
		}
		var state deviceConversationState
		if err := decodeData(statusResponse.Body, &state); err != nil {
			return err
		}
		switch state.Action {
		case "continue":
			if state.Status != "ACTIVE" {
				return fmt.Errorf("conversation returned continue in unexpected state %s", state.Status)
			}
			return nil
		case "leave_room":
			code := state.FailureCode
			if code == "" {
				code = strings.ToLower(state.Status)
			}
			return fmt.Errorf("conversation cannot start; leave RTC room: %s", code)
		case "wait":
		default:
			return fmt.Errorf("conversation returned unsupported device action %q", state.Action)
		}
		if time.Now().After(deadline) {
			return errors.New("provider did not activate the simulated conversation within 15 seconds")
		}
		if err := waitForContext(ctx, 500*time.Millisecond); err != nil {
			return err
		}
	}
}

func (s *simulator) transcript(ctx context.Context, conversationID string) error {
	now := time.Now().UTC()
	body := jsonBody(map[string]any{"turns": []any{map[string]any{
		"turn_key": "sim-final-" + nonce(), "sequence_no": 100, "speaker": "subject",
		"start_at": now.Add(-2 * time.Second), "end_at": now,
		"final_text": "这是模拟设备离线缓存后补交的最终字幕。", "interrupted": false,
	}}})
	key := "sim-transcript-" + nonce()
	if _, err := s.request(ctx, http.MethodPost, "/device/v2/conversations/"+conversationID+"/turns:batch", body, key); err != nil {
		return err
	}
	replay, err := s.request(ctx, http.MethodPost, "/device/v2/conversations/"+conversationID+"/turns:batch", body, key)
	if err != nil || replay.Header.Get("Idempotency-Replayed") != "true" {
		return fmt.Errorf("transcript replay failed: %w", err)
	}
	fmt.Println("transcript supplement and replay completed")
	return nil
}

func (s *simulator) telemetry(ctx context.Context, conversationID string) error {
	sequence := time.Now().UnixNano()
	event := map[string]any{
		"event_key": "sim-status-" + strconv.FormatInt(sequence, 10), "sequence_no": sequence,
		"occurred_at": time.Now().UTC(), "event_type": "status",
		"status": map[string]any{
			"firmware_version": "0.2.0-sim", "reported_config_revision": 1,
			"network":         map[string]any{"type": "simulated", "rssi": -52},
			"rtc":             map[string]any{"joined": false, "uplink": false, "downlink": false},
			"recording_queue": map[string]any{"pending_segments": 0, "failed_segments": 0}, "conversation_status": "COMPLETED",
		},
	}
	if conversationID != "" {
		event["conversation_id"] = conversationID
	}
	body := jsonBody(map[string]any{"events": []any{event}})
	key := "sim-telemetry-" + nonce()
	if _, err := s.request(ctx, http.MethodPost, "/device/v2/telemetry:batch", body, key); err != nil {
		return err
	}
	replay, err := s.request(ctx, http.MethodPost, "/device/v2/telemetry:batch", body, key)
	if err != nil || replay.Header.Get("Idempotency-Replayed") != "true" {
		return fmt.Errorf("telemetry replay failed: %w", err)
	}
	fmt.Println("telemetry batch and replay completed")
	return nil
}

func (s *simulator) commands(ctx context.Context) error {
	response, err := s.request(ctx, http.MethodGet, "/device/v2/commands?limit=20", nil, "")
	if err != nil {
		return err
	}
	var data struct {
		Items []simulatedDeviceCommand `json:"items"`
	}
	if err := decodeData(response.Body, &data); err != nil {
		return err
	}
	completed, failed := 0, 0
	for _, command := range data.Items {
		if command.Type == "firmware_update" {
			outcome, err := s.executeFirmwareCommand(ctx, command)
			if err != nil {
				return err
			}
			switch outcome {
			case firmwareCompleted:
				completed++
			case firmwareFailed:
				failed++
			case firmwareInterrupted:
				fmt.Printf("firmware command %s intentionally interrupted; server state retained for recovery\n", command.ID)
				return nil
			case firmwareRebootRequired:
				fmt.Printf("firmware command %s armed the inactive slot; restart the simulator to perform boot health confirmation\n", command.ID)
				return nil
			}
			continue
		}
		if err := s.ackDeviceCommand(ctx, command.ID, "SUCCEEDED", map[string]any{"result_code": "SIMULATED_OK"}); err != nil {
			return err
		}
		completed++
	}
	fmt.Printf("processed %d pending commands: %d completed, %d simulated failures\n", len(data.Items), completed, failed)
	return nil
}

func (s *simulator) media(ctx context.Context, conversationID string) error {
	durationMS, err := strconv.Atoi(env("SIM_MEDIA_DURATION_MS", "100"))
	if err != nil || durationMS < 1 || durationMS > 600000 {
		return errors.New("SIM_MEDIA_DURATION_MS must be between 1 and 600000")
	}
	mediaPath := strings.TrimSpace(os.Getenv("SIM_MEDIA_FILE"))
	content := syntheticWAV(durationMS)
	if mediaPath != "" {
		content, err = os.ReadFile(mediaPath)
		if err != nil {
			return fmt.Errorf("read simulated media file: %w", err)
		}
	}
	digest := sha256.Sum256(content)
	manifest := jsonBody(map[string]any{"segments": []any{map[string]any{
		"track": env("SIM_MEDIA_TRACK", "near"), "sequence_no": int(time.Now().Unix() % 1000000000),
		"codec": "wav_pcm_s16le", "content_type": "audio/wav", "duration_ms": durationMS,
		"expected_size": len(content), "expected_digest": hex.EncodeToString(digest[:]),
	}}})
	prepared, err := s.request(ctx, http.MethodPost, "/device/v2/conversations/"+conversationID+"/upload-batches", manifest, "sim-media-"+nonce())
	if err != nil {
		return err
	}
	var batch uploadBatch
	if err := decodeData(prepared.Body, &batch); err != nil {
		return err
	}
	if len(batch.Segments) != 1 || batch.Segments[0].UploadURL == "" {
		return errors.New("media preparation did not return one upload URL")
	}
	uploadRequest, err := http.NewRequestWithContext(ctx, http.MethodPut, batch.Segments[0].UploadURL, bytes.NewReader(content))
	if err != nil {
		return err
	}
	uploadRequest.Header.Set("Content-Type", "audio/wav")
	uploadResponse, err := s.client.Do(uploadRequest)
	if err != nil {
		return fmt.Errorf("upload simulated media: %w", err)
	}
	uploadResponse.Body.Close()
	if uploadResponse.StatusCode < 200 || uploadResponse.StatusCode >= 300 {
		return fmt.Errorf("upload simulated media returned HTTP %d", uploadResponse.StatusCode)
	}
	if _, err := s.request(ctx, http.MethodPost, "/device/v2/upload-batches/"+batch.ID+"/complete", []byte(`{}`), ""); err != nil {
		return err
	}
	verifyTimeout, err := simulatorDuration("SIM_MEDIA_VERIFY_TIMEOUT", 20*time.Second, 5*time.Minute)
	if err != nil {
		return err
	}
	verified, err := s.waitForVerifiedMedia(ctx, batch.ID, verifyTimeout)
	if err != nil {
		return err
	}
	if verified.LocalDeleteAfter == nil {
		return errors.New("verified media segment did not return local_delete_after")
	}
	deleteWaitMax, err := simulatorDuration("SIM_MEDIA_DELETE_WAIT_MAX", 30*time.Second, 10*time.Minute)
	if err != nil {
		return err
	}
	wait := time.Until(*verified.LocalDeleteAfter) + 100*time.Millisecond
	if wait > deleteWaitMax {
		fmt.Printf("media verified; local encrypted copy retained until %s (outside simulator wait window)\n", verified.LocalDeleteAfter.UTC().Format(time.RFC3339))
		return nil
	}
	if wait > 0 {
		if err := waitForContext(ctx, wait); err != nil {
			return err
		}
	}
	// The simulator only releases its in-memory copy. SIM_MEDIA_FILE is an operator-owned
	// fixture and is intentionally never removed from disk.
	content = nil
	ack := jsonBody(map[string]any{"status": "DELETED"})
	if _, err := s.request(ctx, http.MethodPost, "/device/v2/media-segments/"+verified.ID+"/local-delete-ack", ack, "sim-local-delete-"+nonce()); err != nil {
		return err
	}
	fmt.Println("media verified and local encrypted copy deletion acknowledged:", batch.ID)
	return nil
}

type verifiedMediaSegment struct {
	ID               string
	LocalDeleteAfter *time.Time
}

func (s *simulator) waitForVerifiedMedia(ctx context.Context, batchID string, timeout time.Duration) (verifiedMediaSegment, error) {
	deadline := time.Now().Add(timeout)
	for {
		response, err := s.request(ctx, http.MethodGet, "/device/v2/upload-batches/"+batchID, nil, "")
		if err != nil {
			return verifiedMediaSegment{}, err
		}
		var batch uploadBatch
		if err := decodeData(response.Body, &batch); err != nil {
			return verifiedMediaSegment{}, err
		}
		if batch.Status == "FAILED" {
			code := "SEGMENT_REJECTED"
			if len(batch.Segments) > 0 && batch.Segments[0].LastErrorCode != "" {
				code = batch.Segments[0].LastErrorCode
			}
			return verifiedMediaSegment{}, fmt.Errorf("media verification failed: %s", code)
		}
		if batch.Status == "COMPLETED" {
			for _, segment := range batch.Segments {
				if segment.Status == "VERIFIED" && segment.ID != "" {
					return verifiedMediaSegment{ID: segment.ID, LocalDeleteAfter: segment.LocalDeleteAfter}, nil
				}
			}
			return verifiedMediaSegment{}, errors.New("completed media batch has no verified segment")
		}
		if time.Now().After(deadline) {
			return verifiedMediaSegment{}, fmt.Errorf("media verification did not complete within %s", timeout)
		}
		if err := waitForContext(ctx, 300*time.Millisecond); err != nil {
			return verifiedMediaSegment{}, err
		}
	}
}

func simulatorDuration(key string, fallback, maximum time.Duration) (time.Duration, error) {
	value := strings.TrimSpace(os.Getenv(key))
	if value == "" {
		return fallback, nil
	}
	parsed, err := time.ParseDuration(value)
	if err != nil || parsed <= 0 || parsed > maximum {
		return 0, fmt.Errorf("%s must be a positive duration no greater than %s", key, maximum)
	}
	return parsed, nil
}

func waitForContext(ctx context.Context, duration time.Duration) error {
	timer := time.NewTimer(duration)
	defer timer.Stop()
	select {
	case <-ctx.Done():
		return ctx.Err()
	case <-timer.C:
		return nil
	}
}

func (s *simulator) request(ctx context.Context, method, path string, body []byte, key string) (apiResult, error) {
	request, err := http.NewRequestWithContext(ctx, method, s.baseURL+path, bytes.NewReader(body))
	if err != nil {
		return apiResult{}, err
	}
	timestamp := time.Now().UTC().Format(time.RFC3339)
	requestNonce, err := newNonce()
	if err != nil {
		return apiResult{}, err
	}
	hash := sha256.Sum256(body)
	digest := hex.EncodeToString(hash[:])
	request.Header.Set(deviceauth.HeaderDeviceID, s.deviceID)
	request.Header.Set(deviceauth.HeaderTimestamp, timestamp)
	request.Header.Set(deviceauth.HeaderNonce, requestNonce)
	request.Header.Set(deviceauth.HeaderContentSHA256, digest)
	request.Header.Set(deviceauth.HeaderSignature, deviceauth.SignatureHeader(s.secret, method, request.URL.RequestURI(), timestamp, requestNonce, digest))
	if len(body) > 0 {
		request.Header.Set("Content-Type", "application/json")
	}
	if key != "" {
		request.Header.Set("Idempotency-Key", key)
	}
	response, err := s.client.Do(request)
	if err != nil {
		return apiResult{}, err
	}
	defer response.Body.Close()
	responseBody, err := io.ReadAll(io.LimitReader(response.Body, 1<<20))
	if err != nil {
		return apiResult{}, err
	}
	result := apiResult{Status: response.StatusCode, Header: response.Header.Clone(), Body: responseBody}
	if response.StatusCode < 200 || response.StatusCode >= 300 {
		return result, fmt.Errorf("%s returned HTTP %d: %s", request.URL.Path, response.StatusCode, responseBody)
	}
	return result, nil
}

func decodeData(body []byte, target any) error {
	var result envelope
	if err := json.Unmarshal(body, &result); err != nil {
		return err
	}
	return json.Unmarshal(result.Data, target)
}

func parseScenarios(value string) map[string]bool {
	result := map[string]bool{}
	for _, item := range strings.Split(strings.ToLower(value), ",") {
		if item = strings.TrimSpace(item); item != "" {
			result[item] = true
		}
	}
	return result
}

func syntheticWAV(durationMS int) []byte {
	samples := 16000 * durationMS / 1000
	dataSize := samples * 2
	buffer := bytes.NewBuffer(make([]byte, 0, 44+dataSize))
	buffer.WriteString("RIFF")
	_ = binary.Write(buffer, binary.LittleEndian, uint32(36+dataSize))
	buffer.WriteString("WAVEfmt ")
	_ = binary.Write(buffer, binary.LittleEndian, uint32(16))
	_ = binary.Write(buffer, binary.LittleEndian, uint16(1))
	_ = binary.Write(buffer, binary.LittleEndian, uint16(1))
	_ = binary.Write(buffer, binary.LittleEndian, uint32(16000))
	_ = binary.Write(buffer, binary.LittleEndian, uint32(32000))
	_ = binary.Write(buffer, binary.LittleEndian, uint16(2))
	_ = binary.Write(buffer, binary.LittleEndian, uint16(16))
	buffer.WriteString("data")
	_ = binary.Write(buffer, binary.LittleEndian, uint32(dataSize))
	buffer.Write(make([]byte, dataSize))
	return buffer.Bytes()
}

func jsonBody(value any) []byte {
	body, err := json.Marshal(value)
	if err != nil {
		panic(err)
	}
	return body
}
func nonce() string {
	value, err := newNonce()
	if err != nil {
		panic(err)
	}
	return value
}
func newNonce() (string, error) {
	value := make([]byte, 16)
	if _, err := rand.Read(value); err != nil {
		return "", err
	}
	return hex.EncodeToString(value), nil
}
func env(key, fallback string) string {
	if value := strings.TrimSpace(os.Getenv(key)); value != "" {
		return value
	}
	return fallback
}