speedtest/internal/web/server.go

package web

import (
	"context"
	"crypto/rand"
	"encoding/hex"
	"encoding/json"
	"errors"
	"fmt"
	"io"
	"math"
	"net"
	"net/http"
	"net/url"
	"os"
	"os/exec"
	"path/filepath"
	"runtime"
	"sort"
	"strconv"
	"strings"
	"sync"
	"time"

	"speedtest/internal/limit"
	"speedtest/internal/probe"
	recordstore "speedtest/internal/store"
)

type RecordStore interface {
	Append(ctx context.Context, record recordstore.Record) error
	List(ctx context.Context, limit int) ([]recordstore.Record, error)
}

type Runner func(ctx context.Context, request TestRequest, emit func(Event) error) (probe.Summary, error)
type ClientResolver func(ctx context.Context, ip string) (recordstore.ClientInfo, error)

type Options struct {
	Store              RecordStore
	Runner             Runner
	ClientResolver     ClientResolver
	TrustProxyHeaders  bool
	BandwidthLimitMbps float64
	Iperf3Path         string
	DownloadsDir       string
	StaticFS           http.FileSystem
	StaticDir          string
}

type Server struct {
	store             RecordStore
	runner            Runner
	clientResolver    ClientResolver
	trustProxyHeaders bool
	bandwidthLimiter  limit.Limiter
	iperf3Path        string
	downloadsDir      string
	staticFS          http.FileSystem
	staticDir         string

	mu            sync.Mutex
	sessions      map[string]TestRequest
	iperfSessions map[string]*iperfSession
	queue         []string
	activeID      string
}

type TestRequest struct {
	ID              string
	Target          probe.Target
	HTTPURL         string
	Duration        time.Duration
	Interval        time.Duration
	Timeout         time.Duration
	CreatedAt       time.Time
	Client          recordstore.ClientInfo
	DurationSeconds int
	IntervalMillis  int
	TimeoutMillis   int
}

type Event struct {
	Type    string         `json:"type"`
	Sample  *probe.Sample  `json:"sample,omitempty"`
	Summary *probe.Summary `json:"summary,omitempty"`
	Error   string         `json:"error,omitempty"`
}

type recordsResponse struct {
	Records []recordstore.Record `json:"records"`
}

type configResponse struct {
	BandwidthLimitMbps  float64 `json:"bandwidthLimitMbps"`
	BandwidthLimitLabel string  `json:"bandwidthLimitLabel"`
}

type heartbeatResponse struct {
	At                  time.Time `json:"at"`
	BandwidthLimitMbps  float64   `json:"bandwidthLimitMbps"`
	BandwidthLimitLabel string    `json:"bandwidthLimitLabel"`
	ActiveTest          bool      `json:"activeTest"`
}

type createIperf3Request struct {
	DurationSeconds int     `json:"durationSeconds"`
	Protocol        string  `json:"protocol"`
	Reverse         bool    `json:"reverse"`
	BandwidthMbps   float64 `json:"bandwidthMbps"`
}

type createIperf3Response struct {
	ID              string  `json:"id"`
	Host            string  `json:"host"`
	Port            int     `json:"port"`
	Protocol        string  `json:"protocol"`
	Reverse         bool    `json:"reverse"`
	DurationSeconds int     `json:"durationSeconds"`
	BandwidthMbps   float64 `json:"bandwidthMbps"`
	CompleteURL     string  `json:"completeUrl"`
}

type completeIperf3Request struct {
	StartedAt  time.Time      `json:"startedAt"`
	FinishedAt time.Time      `json:"finishedAt"`
	Samples    []probe.Sample `json:"samples"`
}

type createTestResponse struct {
	ID            string `json:"id"`
	Status        string `json:"status"`
	QueuePosition int    `json:"queuePosition"`
	QueueURL      string `json:"queueUrl"`
	EventsURL     string `json:"eventsUrl"`
	DownloadURL   string `json:"downloadUrl"`
	UploadURL     string `json:"uploadUrl"`
	CompleteURL   string `json:"completeUrl"`
	CancelURL     string `json:"cancelUrl"`
}

type queueStatusResponse struct {
	ID            string `json:"id"`
	Status        string `json:"status"`
	QueuePosition int    `json:"queuePosition"`
}

type transferResponse struct {
	Bytes      int64 `json:"bytes"`
	DurationMS int64 `json:"durationMs"`
}

type completeTestRequest struct {
	StartedAt  time.Time      `json:"startedAt"`
	FinishedAt time.Time      `json:"finishedAt"`
	Samples    []probe.Sample `json:"samples"`
}

type completeTestResponse struct {
	Summary probe.Summary `json:"summary"`
}

type errorResponse struct {
	Error string `json:"error"`
}

type busyResponse struct {
	Error        string `json:"error"`
	ActiveTestID string `json:"activeTestId,omitempty"`
}

type rawTestRequest struct {
	Target          string `json:"target"`
	HTTPURL         string `json:"httpUrl"`
	DurationSeconds int    `json:"durationSeconds"`
	IntervalMillis  int    `json:"intervalMillis"`
	TimeoutMillis   int    `json:"timeoutMillis"`
}

type iperfSession struct {
	ID            string
	Request       TestRequest
	Process       *exec.Cmd
	Protocol      string
	Reverse       bool
	Port          int
	BandwidthMbps float64
}

const maxCompleteBodyBytes = 64 * 1024 * 1024

func New(options Options) *Server {
	iperf3Path := strings.TrimSpace(options.Iperf3Path)
	if iperf3Path == "" {
		iperf3Path = "iperf3"
	}
	return &Server{
		store:             options.Store,
		runner:            options.Runner,
		clientResolver:    options.ClientResolver,
		trustProxyHeaders: options.TrustProxyHeaders,
		bandwidthLimiter:  limit.New(options.BandwidthLimitMbps),
		iperf3Path:        iperf3Path,
		downloadsDir:      strings.TrimSpace(options.DownloadsDir),
		staticFS:          options.StaticFS,
		staticDir:         options.StaticDir,
		sessions:          make(map[string]TestRequest),
		iperfSessions:     make(map[string]*iperfSession),
	}
}

func (s *Server) Handler() http.Handler {
	mux := http.NewServeMux()
	mux.HandleFunc("/api/config", s.handleConfig)
	mux.HandleFunc("/api/health/events", s.handleHealthEvents)
	mux.HandleFunc("/api/download", s.handleDownload)
	mux.HandleFunc("/api/upload", s.handleUpload)
	mux.HandleFunc("/api/iperf3/sessions", s.handleCreateIperf3Session)
	mux.HandleFunc("/api/iperf3/sessions/", s.handleIperf3Route)
	mux.HandleFunc("/api/records", s.handleRecords)
	mux.HandleFunc("/api/tests", s.handleCreateTest)
	mux.HandleFunc("/api/tests/", s.handleTestRoute)
	mux.HandleFunc("/downloads/", s.handleBinaryDownload)
	if s.staticFS != nil {
		mux.Handle("/", http.FileServer(s.staticFS))
	} else if s.staticDir != "" {
		mux.Handle("/", http.FileServer(http.Dir(s.staticDir)))
	}
	return mux
}

func (s *Server) handleUpload(w http.ResponseWriter, r *http.Request) {
	if r.Method != http.MethodPost {
		writeJSON(w, http.StatusMethodNotAllowed, errorResponse{Error: "method not allowed"})
		return
	}
	testID := strings.TrimSpace(r.URL.Query().Get("testId"))
	if !s.isActive(testID) {
		writeJSON(w, http.StatusConflict, errorResponse{Error: "没有可用的活动测速会话，请先开始测试。"})
		return
	}

	startedAt := time.Now()
	reader := http.MaxBytesReader(w, r.Body, 1024*1024*1024)
	defer reader.Close()

	var received int64
	buffer := make([]byte, 32*1024)
	for {
		if err := r.Context().Err(); err != nil {
			return
		}
		n, err := reader.Read(buffer)
		if n > 0 {
			received += int64(n)
			if delay := s.bandwidthLimiter.DelayFor(received, time.Since(startedAt)); delay > 0 {
				timer := time.NewTimer(delay)
				select {
				case <-r.Context().Done():
					timer.Stop()
					return
				case <-timer.C:
				}
			}
		}
		if errors.Is(err, io.EOF) {
			break
		}
		if err != nil {
			writeJSON(w, http.StatusBadRequest, errorResponse{Error: "failed to read upload body"})
			return
		}
	}

	writeJSON(w, http.StatusOK, transferResponse{
		Bytes:      received,
		DurationMS: time.Since(startedAt).Milliseconds(),
	})
}

func (s *Server) handleConfig(w http.ResponseWriter, r *http.Request) {
	if r.Method != http.MethodGet {
		writeJSON(w, http.StatusMethodNotAllowed, errorResponse{Error: "method not allowed"})
		return
	}
	writeJSON(w, http.StatusOK, configResponse{
		BandwidthLimitMbps:  s.bandwidthLimiter.Mbps(),
		BandwidthLimitLabel: bandwidthLimitLabel(s.bandwidthLimiter),
	})
}

func (s *Server) handleDownload(w http.ResponseWriter, r *http.Request) {
	if r.Method != http.MethodGet {
		writeJSON(w, http.StatusMethodNotAllowed, errorResponse{Error: "method not allowed"})
		return
	}

	size, err := parseDownloadSize(r.URL.Query().Get("bytes"))
	if err != nil {
		writeJSON(w, http.StatusBadRequest, errorResponse{Error: err.Error()})
		return
	}
	testID := strings.TrimSpace(r.URL.Query().Get("testId"))
	if !s.isActive(testID) {
		writeJSON(w, http.StatusConflict, errorResponse{Error: "没有可用的活动测速会话，请先开始测试。"})
		return
	}

	w.Header().Set("Content-Type", "application/octet-stream")
	w.Header().Set("Content-Length", strconv.FormatInt(size, 10))
	w.Header().Set("X-NetStable-Bandwidth-Limit-Mbps", formatMbps(s.bandwidthLimiter.Mbps()))

	flusher, _ := w.(http.Flusher)
	startedAt := time.Now()
	var sent int64
	buffer := make([]byte, 32*1024)
	for sent < size {
		if err := r.Context().Err(); err != nil {
			return
		}

		remaining := size - sent
		chunk := int64(len(buffer))
		if remaining < chunk {
			chunk = remaining
		}
		written, err := w.Write(buffer[:chunk])
		if err != nil {
			return
		}
		sent += int64(written)
		if flusher != nil {
			flusher.Flush()
		}

		if delay := s.bandwidthLimiter.DelayFor(sent, time.Since(startedAt)); delay > 0 {
			timer := time.NewTimer(delay)
			select {
			case <-r.Context().Done():
				timer.Stop()
				return
			case <-timer.C:
			}
		}
	}
}

func (s *Server) handleRecords(w http.ResponseWriter, r *http.Request) {
	if r.Method != http.MethodGet {
		writeJSON(w, http.StatusMethodNotAllowed, errorResponse{Error: "method not allowed"})
		return
	}
	limit := 100
	if value := r.URL.Query().Get("limit"); value != "" {
		parsed, err := strconv.Atoi(value)
		if err != nil || parsed < 1 || parsed > 1000 {
			writeJSON(w, http.StatusBadRequest, errorResponse{Error: "limit must be between 1 and 1000"})
			return
		}
		limit = parsed
	}
	records, err := s.store.List(r.Context(), limit)
	if err != nil {
		writeJSON(w, http.StatusInternalServerError, errorResponse{Error: "failed to load records"})
		return
	}
	records = sanitizeRecords(records)
	writeJSON(w, http.StatusOK, recordsResponse{Records: records})
}

func (s *Server) handleHealthEvents(w http.ResponseWriter, r *http.Request) {
	if r.Method != http.MethodGet {
		writeJSON(w, http.StatusMethodNotAllowed, errorResponse{Error: "method not allowed"})
		return
	}
	limit := 0
	if value := r.URL.Query().Get("limit"); value != "" {
		parsed, err := strconv.Atoi(value)
		if err != nil || parsed < 1 || parsed > 100 {
			writeJSON(w, http.StatusBadRequest, errorResponse{Error: "limit must be between 1 and 100"})
			return
		}
		limit = parsed
	}

	w.Header().Set("Content-Type", "text/event-stream; charset=utf-8")
	w.Header().Set("Cache-Control", "no-cache")
	w.Header().Set("X-Accel-Buffering", "no")
	flusher, _ := w.(http.Flusher)

	ticker := time.NewTicker(15 * time.Second)
	defer ticker.Stop()
	for sent := 0; ; sent++ {
		if err := writeSSEData(w, "heartbeat", heartbeatResponse{
			At:                  time.Now().UTC(),
			BandwidthLimitMbps:  s.bandwidthLimiter.Mbps(),
			BandwidthLimitLabel: bandwidthLimitLabel(s.bandwidthLimiter),
			ActiveTest:          s.hasActiveTest(),
		}); err != nil {
			return
		}
		if flusher != nil {
			flusher.Flush()
		}
		if limit > 0 && sent+1 >= limit {
			return
		}
		select {
		case <-r.Context().Done():
			return
		case <-ticker.C:
		}
	}
}

func (s *Server) handleBinaryDownload(w http.ResponseWriter, r *http.Request) {
	if r.Method != http.MethodGet {
		writeJSON(w, http.StatusMethodNotAllowed, errorResponse{Error: "method not allowed"})
		return
	}
	name := strings.Trim(strings.TrimPrefix(r.URL.Path, "/downloads/"), "/")
	if !isDownloadBinaryName(name) {
		writeJSON(w, http.StatusNotFound, errorResponse{Error: "download not found"})
		return
	}

	if s.downloadsDir != "" {
		path := filepath.Join(s.downloadsDir, name)
		if info, err := os.Stat(path); err == nil && info.Mode().IsRegular() {
			serveBinaryFile(w, r, name, path)
			return
		}
	}

	if name == currentLinuxBinaryName() {
		path, err := os.Executable()
		if err == nil {
			if info, statErr := os.Stat(path); statErr == nil && info.Mode().IsRegular() {
				serveBinaryFile(w, r, name, path)
				return
			}
		}
	}

	writeJSON(w, http.StatusNotFound, errorResponse{Error: "download not found"})
}

func (s *Server) handleCreateIperf3Session(w http.ResponseWriter, r *http.Request) {
	if r.Method != http.MethodPost {
		writeJSON(w, http.StatusMethodNotAllowed, errorResponse{Error: "method not allowed"})
		return
	}

	var raw createIperf3Request
	decoder := json.NewDecoder(http.MaxBytesReader(w, r.Body, 64*1024))
	decoder.DisallowUnknownFields()
	if err := decoder.Decode(&raw); err != nil {
		writeJSON(w, http.StatusBadRequest, errorResponse{Error: "invalid JSON request"})
		return
	}

	request, protocol, reverse, bandwidthMbps, err := s.normalizeIperf3Request(raw, r)
	if err != nil {
		writeJSON(w, http.StatusBadRequest, errorResponse{Error: err.Error()})
		return
	}
	request.ID = newID()
	request.CreatedAt = time.Now().UTC()
	request.Client = s.resolveClient(r.Context(), clientIPFromRequest(r, s.trustProxyHeaders))

	s.mu.Lock()
	s.expireActiveLocked(time.Now().UTC())
	if s.activeID != "" {
		activeID := s.activeID
		s.mu.Unlock()
		writeJSON(w, http.StatusConflict, busyResponse{
			Error:        "有其他用户正在测速，请稍等。",
			ActiveTestID: activeID,
		})
		return
	}
	s.sessions[request.ID] = request
	s.activeID = request.ID
	s.mu.Unlock()

	port, err := allocateHighPort()
	if err != nil {
		s.takeSession(request.ID)
		writeJSON(w, http.StatusInternalServerError, errorResponse{Error: "failed to allocate iperf3 port"})
		return
	}
	cmd := s.iperf3ServerCommand(port, protocol, bandwidthMbps)
	if err := cmd.Start(); err != nil {
		s.takeSession(request.ID)
		writeJSON(w, http.StatusServiceUnavailable, errorResponse{Error: "failed to start iperf3 server: " + err.Error()})
		return
	}
	go func() { _ = cmd.Wait() }()

	session := &iperfSession{
		ID:            request.ID,
		Request:       request,
		Process:       cmd,
		Protocol:      protocol,
		Reverse:       reverse,
		Port:          port,
		BandwidthMbps: bandwidthMbps,
	}
	s.mu.Lock()
	s.iperfSessions[request.ID] = session
	s.mu.Unlock()
	go s.expireIperf3Session(request.ID, request.Duration+60*time.Second)

	writeJSON(w, http.StatusAccepted, createIperf3Response{
		ID:              request.ID,
		Host:            iperfHostFromRequest(r),
		Port:            port,
		Protocol:        protocol,
		Reverse:         reverse,
		DurationSeconds: request.DurationSeconds,
		BandwidthMbps:   bandwidthMbps,
		CompleteURL:     "/api/iperf3/sessions/" + request.ID + "/complete",
	})
}

func (s *Server) normalizeIperf3Request(raw createIperf3Request, r *http.Request) (TestRequest, string, bool, float64, error) {
	durationSeconds := raw.DurationSeconds
	if durationSeconds == 0 {
		durationSeconds = 30
	}
	if durationSeconds < 1 || durationSeconds > 600 {
		return TestRequest{}, "", false, 0, errors.New("durationSeconds must be between 1 and 600")
	}

	protocol := strings.ToLower(strings.TrimSpace(raw.Protocol))
	if protocol == "" {
		protocol = "tcp"
	}
	if protocol != "tcp" && protocol != "udp" {
		return TestRequest{}, "", false, 0, errors.New("protocol must be tcp or udp")
	}

	bandwidthMbps := raw.BandwidthMbps
	if bandwidthMbps == 0 {
		bandwidthMbps = s.bandwidthLimiter.Mbps()
	}
	if bandwidthMbps < 0 || bandwidthMbps > 100000 {
		return TestRequest{}, "", false, 0, errors.New("bandwidthMbps must be between 0 and 100000")
	}

	host := iperfHostFromRequest(r)
	target, err := probe.ParseTarget(net.JoinHostPort(host, "1"))
	if err != nil {
		return TestRequest{}, "", false, 0, err
	}
	target.Port = 0
	target.Address = fmt.Sprintf("iperf3/%s/%s", protocol, host)
	target.Raw = target.Address

	return TestRequest{
		Target:          target,
		Duration:        time.Duration(durationSeconds) * time.Second,
		Timeout:         3 * time.Second,
		DurationSeconds: durationSeconds,
		TimeoutMillis:   3000,
	}, protocol, raw.Reverse, bandwidthMbps, nil
}

func (s *Server) iperf3ServerCommand(port int, protocol string, bandwidthMbps float64) *exec.Cmd {
	args := []string{"-s", "-1", "-p", strconv.Itoa(port)}
	if bandwidthMbps > 0 {
		args = append(args, "--server-bitrate-limit", formatMbps(bandwidthMbps)+"M")
	}
	return exec.Command(s.iperf3Path, args...)
}

func (s *Server) handleIperf3Route(w http.ResponseWriter, r *http.Request) {
	switch {
	case strings.HasSuffix(r.URL.Path, "/complete"):
		s.handleCompleteIperf3Session(w, r)
	default:
		writeJSON(w, http.StatusNotFound, errorResponse{Error: "not found"})
	}
}

func (s *Server) handleCompleteIperf3Session(w http.ResponseWriter, r *http.Request) {
	if r.Method != http.MethodPost {
		writeJSON(w, http.StatusMethodNotAllowed, errorResponse{Error: "method not allowed"})
		return
	}
	id, ok := completeIperf3SessionID(r.URL.Path)
	if !ok {
		writeJSON(w, http.StatusNotFound, errorResponse{Error: "not found"})
		return
	}

	var raw completeIperf3Request
	decoder := json.NewDecoder(http.MaxBytesReader(w, r.Body, maxCompleteBodyBytes))
	decoder.DisallowUnknownFields()
	if err := decoder.Decode(&raw); err != nil {
		writeJSON(w, http.StatusBadRequest, errorResponse{Error: "invalid JSON request"})
		return
	}

	session, exists := s.activeIperf3Session(id)
	if !exists {
		writeJSON(w, http.StatusNotFound, errorResponse{Error: "iperf3 session not found"})
		return
	}

	summary := summarizeBrowserSamples(session.Request, completeTestRequest{
		StartedAt:  raw.StartedAt,
		FinishedAt: raw.FinishedAt,
		Samples:    raw.Samples,
	})
	if s.store != nil {
		if err := s.store.Append(r.Context(), recordstore.Record{
			ID:        session.Request.ID,
			Client:    session.Request.Client,
			Summary:   summary,
			Samples:   raw.Samples,
			CreatedAt: session.Request.CreatedAt,
		}); err != nil {
			writeJSON(w, http.StatusInternalServerError, errorResponse{Error: "failed to persist iperf3 record"})
			return
		}
	}
	session, _ = s.takeIperf3Session(id)
	stopIperf3Process(session)

	writeJSON(w, http.StatusOK, completeTestResponse{Summary: summary})
}

func parseDownloadSize(value string) (int64, error) {
	if value == "" {
		return 10 * 1024 * 1024, nil
	}
	size, err := strconv.ParseInt(value, 10, 64)
	if err != nil || size < 1 || size > 1024*1024*1024 {
		return 0, errors.New("bytes must be between 1 and 1073741824")
	}
	return size, nil
}

func bandwidthLimitLabel(limiter limit.Limiter) string {
	if !limiter.Enabled() {
		return "不限速"
	}
	return formatMbps(limiter.Mbps()) + " Mbps"
}

func formatMbps(mbps float64) string {
	if mbps == math.Trunc(mbps) {
		return strconv.FormatInt(int64(mbps), 10)
	}
	return strings.TrimRight(strings.TrimRight(fmt.Sprintf("%.2f", mbps), "0"), ".")
}

func (s *Server) handleCreateTest(w http.ResponseWriter, r *http.Request) {
	if r.Method != http.MethodPost {
		writeJSON(w, http.StatusMethodNotAllowed, errorResponse{Error: "method not allowed"})
		return
	}

	var raw rawTestRequest
	decoder := json.NewDecoder(http.MaxBytesReader(w, r.Body, 64*1024))
	decoder.DisallowUnknownFields()
	if err := decoder.Decode(&raw); err != nil {
		writeJSON(w, http.StatusBadRequest, errorResponse{Error: "invalid JSON request"})
		return
	}

	request, err := normalizeTestRequest(raw)
	if err != nil {
		writeJSON(w, http.StatusBadRequest, errorResponse{Error: err.Error()})
		return
	}
	request.ID = newID()
	request.CreatedAt = time.Now().UTC()
	request.Client = s.resolveClient(r.Context(), clientIPFromRequest(r, s.trustProxyHeaders))

	s.mu.Lock()
	s.expireActiveLocked(time.Now().UTC())
	status := "ready"
	position := 0
	if s.activeID != "" {
		status = "queued"
		s.queue = append(s.queue, request.ID)
		position = queuePositionLocked(s.queue, request.ID)
	}
	s.sessions[request.ID] = request
	if status == "ready" {
		s.activeID = request.ID
	}
	s.mu.Unlock()

	writeJSON(w, http.StatusAccepted, createTestResponse{
		ID:            request.ID,
		Status:        status,
		QueuePosition: position,
		QueueURL:      "/api/tests/" + request.ID + "/queue",
		EventsURL:     "/api/tests/" + request.ID + "/events",
		DownloadURL:   "/api/download?testId=" + request.ID,
		UploadURL:     "/api/upload?testId=" + request.ID,
		CompleteURL:   "/api/tests/" + request.ID + "/complete",
		CancelURL:     "/api/tests/" + request.ID + "/cancel",
	})
}

func (s *Server) handleTestRoute(w http.ResponseWriter, r *http.Request) {
	switch {
	case strings.HasSuffix(r.URL.Path, "/queue"):
		s.handleQueueStatus(w, r)
	case strings.HasSuffix(r.URL.Path, "/events"):
		s.handleTestEvents(w, r)
	case strings.HasSuffix(r.URL.Path, "/complete"):
		s.handleCompleteTest(w, r)
	case strings.HasSuffix(r.URL.Path, "/cancel"):
		s.handleCancelTest(w, r)
	default:
		writeJSON(w, http.StatusNotFound, errorResponse{Error: "not found"})
	}
}

func (s *Server) handleQueueStatus(w http.ResponseWriter, r *http.Request) {
	if r.Method != http.MethodGet {
		writeJSON(w, http.StatusMethodNotAllowed, errorResponse{Error: "method not allowed"})
		return
	}
	id, ok := queueSessionID(r.URL.Path)
	if !ok {
		writeJSON(w, http.StatusNotFound, errorResponse{Error: "not found"})
		return
	}

	s.mu.Lock()
	s.expireActiveLocked(time.Now().UTC())
	response, exists := s.queueStatusLocked(id)
	s.mu.Unlock()
	if !exists {
		writeJSON(w, http.StatusNotFound, errorResponse{Error: "test session not found"})
		return
	}
	writeJSON(w, http.StatusOK, response)
}

func (s *Server) handleCompleteTest(w http.ResponseWriter, r *http.Request) {
	if r.Method != http.MethodPost {
		writeJSON(w, http.StatusMethodNotAllowed, errorResponse{Error: "method not allowed"})
		return
	}
	id, ok := completeSessionID(r.URL.Path)
	if !ok {
		writeJSON(w, http.StatusNotFound, errorResponse{Error: "not found"})
		return
	}

	var raw completeTestRequest
	decoder := json.NewDecoder(http.MaxBytesReader(w, r.Body, maxCompleteBodyBytes))
	decoder.DisallowUnknownFields()
	if err := decoder.Decode(&raw); err != nil {
		writeJSON(w, http.StatusBadRequest, errorResponse{Error: "invalid JSON request"})
		return
	}

	request, exists := s.activeSession(id)
	if !exists {
		writeJSON(w, http.StatusNotFound, errorResponse{Error: "test session not found"})
		return
	}

	summary := summarizeBrowserSamples(request, raw)
	if s.store != nil {
		if err := s.store.Append(r.Context(), recordstore.Record{
			ID:        request.ID,
			Client:    request.Client,
			Summary:   summary,
			Samples:   raw.Samples,
			CreatedAt: request.CreatedAt,
		}); err != nil {
			writeJSON(w, http.StatusInternalServerError, errorResponse{Error: "failed to persist test record"})
			return
		}
	}
	s.finishSession(id)

	writeJSON(w, http.StatusOK, completeTestResponse{Summary: summary})
}

func (s *Server) handleCancelTest(w http.ResponseWriter, r *http.Request) {
	if r.Method != http.MethodPost {
		writeJSON(w, http.StatusMethodNotAllowed, errorResponse{Error: "method not allowed"})
		return
	}
	id, ok := cancelSessionID(r.URL.Path)
	if !ok {
		writeJSON(w, http.StatusNotFound, errorResponse{Error: "not found"})
		return
	}
	if _, exists := s.takeSession(id); !exists {
		writeJSON(w, http.StatusNotFound, errorResponse{Error: "test session not found"})
		return
	}
	writeJSON(w, http.StatusOK, map[string]string{"status": "canceled"})
}

func (s *Server) handleTestEvents(w http.ResponseWriter, r *http.Request) {
	if r.Method != http.MethodGet {
		writeJSON(w, http.StatusMethodNotAllowed, errorResponse{Error: "method not allowed"})
		return
	}
	id, ok := eventSessionID(r.URL.Path)
	if !ok {
		writeJSON(w, http.StatusNotFound, errorResponse{Error: "not found"})
		return
	}

	s.mu.Lock()
	_, exists := s.sessions[id]
	active := s.activeID == id
	s.mu.Unlock()
	if !exists {
		writeJSON(w, http.StatusNotFound, errorResponse{Error: "test session not found"})
		return
	}
	if !active {
		writeJSON(w, http.StatusConflict, errorResponse{Error: "该测速会话正在排队，请等待轮到后再开始。"})
		return
	}

	s.mu.Lock()
	request, exists := s.sessions[id]
	delete(s.sessions, id)
	s.mu.Unlock()
	if !exists {
		writeJSON(w, http.StatusNotFound, errorResponse{Error: "test session not found"})
		return
	}

	w.Header().Set("Content-Type", "text/event-stream; charset=utf-8")
	w.Header().Set("Cache-Control", "no-cache")
	w.Header().Set("X-Accel-Buffering", "no")

	flusher, _ := w.(http.Flusher)
	var samples []probe.Sample
	emit := func(event Event) error {
		if event.Sample != nil {
			sample := *event.Sample
			if sample.LatencyMS == 0 && sample.Latency > 0 {
				sample.LatencyMS = float64(sample.Latency.Microseconds()) / 1000
			}
			samples = append(samples, sample)
			event.Sample = &sample
		}
		if err := writeSSE(w, event); err != nil {
			return err
		}
		if flusher != nil {
			flusher.Flush()
		}
		return nil
	}

	if s.runner == nil {
		_ = emit(Event{Type: "error", Error: "test runner is not configured"})
		s.releaseActive(request.ID)
		return
	}
	defer s.releaseActive(request.ID)

	summary, err := s.runner(r.Context(), request, emit)
	if err != nil {
		_ = emit(Event{Type: "error", Error: err.Error()})
		return
	}
	summary.ID = request.ID
	summary.HTTPURL = request.HTTPURL

	if s.store != nil {
		if err := s.store.Append(r.Context(), recordstore.Record{
			ID:        request.ID,
			Client:    request.Client,
			Summary:   summary,
			Samples:   samples,
			CreatedAt: request.CreatedAt,
		}); err != nil {
			_ = emit(Event{Type: "error", Error: "failed to persist test record"})
			return
		}
	}

	_ = emit(Event{Type: "summary", Summary: &summary})
}

func normalizeTestRequest(raw rawTestRequest) (TestRequest, error) {
	target, err := probe.ParseTarget(raw.Target)
	if err != nil {
		return TestRequest{}, fmt.Errorf("invalid target: %w", err)
	}

	durationSeconds := raw.DurationSeconds
	if durationSeconds == 0 {
		durationSeconds = 30
	}
	if durationSeconds < 1 || durationSeconds > 600 {
		return TestRequest{}, errors.New("durationSeconds must be between 1 and 600")
	}

	intervalMillis := raw.IntervalMillis
	if intervalMillis == 0 {
		intervalMillis = 1000
	}
	if intervalMillis < 100 || intervalMillis > 60000 {
		return TestRequest{}, errors.New("intervalMillis must be between 100 and 60000")
	}

	timeoutMillis := raw.TimeoutMillis
	if timeoutMillis == 0 {
		timeoutMillis = 3000
	}
	if timeoutMillis < 100 || timeoutMillis > 30000 {
		return TestRequest{}, errors.New("timeoutMillis must be between 100 and 30000")
	}

	httpURL := strings.TrimSpace(raw.HTTPURL)
	if httpURL != "" {
		parsed, err := url.Parse(httpURL)
		if err != nil || parsed.Host == "" || (parsed.Scheme != "http" && parsed.Scheme != "https") {
			return TestRequest{}, errors.New("httpUrl must be a valid http or https URL")
		}
	}

	return TestRequest{
		Target:          target,
		HTTPURL:         httpURL,
		Duration:        time.Duration(durationSeconds) * time.Second,
		Interval:        time.Duration(intervalMillis) * time.Millisecond,
		Timeout:         time.Duration(timeoutMillis) * time.Millisecond,
		DurationSeconds: durationSeconds,
		IntervalMillis:  intervalMillis,
		TimeoutMillis:   timeoutMillis,
	}, nil
}

func (s *Server) releaseActive(id string) {
	s.mu.Lock()
	defer s.mu.Unlock()
	if s.activeID == id {
		s.activeID = ""
		s.promoteNextLocked(time.Now().UTC())
	}
}

func (s *Server) takeSession(id string) (TestRequest, bool) {
	s.mu.Lock()
	defer s.mu.Unlock()
	request, exists := s.sessions[id]
	if exists {
		delete(s.sessions, id)
	}
	s.removeQueuedLocked(id)
	if session, exists := s.iperfSessions[id]; exists {
		delete(s.iperfSessions, id)
		stopIperf3Process(session)
	}
	if s.activeID == id {
		s.activeID = ""
		s.promoteNextLocked(time.Now().UTC())
	}
	return request, exists
}

func (s *Server) activeSession(id string) (TestRequest, bool) {
	s.mu.Lock()
	defer s.mu.Unlock()
	if s.activeID != id {
		return TestRequest{}, false
	}
	request, exists := s.sessions[id]
	return request, exists
}

func (s *Server) finishSession(id string) {
	s.mu.Lock()
	defer s.mu.Unlock()
	delete(s.sessions, id)
	s.removeQueuedLocked(id)
	if s.activeID == id {
		s.activeID = ""
		s.promoteNextLocked(time.Now().UTC())
	}
}

func (s *Server) takeIperf3Session(id string) (*iperfSession, bool) {
	s.mu.Lock()
	defer s.mu.Unlock()
	session, exists := s.iperfSessions[id]
	if exists {
		delete(s.iperfSessions, id)
		delete(s.sessions, id)
	}
	if s.activeID == id {
		s.activeID = ""
		s.promoteNextLocked(time.Now().UTC())
	}
	return session, exists
}

func (s *Server) activeIperf3Session(id string) (*iperfSession, bool) {
	s.mu.Lock()
	defer s.mu.Unlock()
	if s.activeID != id {
		return nil, false
	}
	session, exists := s.iperfSessions[id]
	return session, exists
}

func (s *Server) expireIperf3Session(id string, ttl time.Duration) {
	timer := time.NewTimer(ttl)
	defer timer.Stop()
	<-timer.C

	s.mu.Lock()
	session, exists := s.iperfSessions[id]
	if exists {
		delete(s.iperfSessions, id)
		delete(s.sessions, id)
	}
	if s.activeID == id {
		s.activeID = ""
		s.promoteNextLocked(time.Now().UTC())
	}
	s.mu.Unlock()
	if exists {
		stopIperf3Process(session)
	}
}

func (s *Server) queueStatusLocked(id string) (queueStatusResponse, bool) {
	if s.activeID == id {
		return queueStatusResponse{ID: id, Status: "ready", QueuePosition: 0}, true
	}
	if _, exists := s.sessions[id]; !exists {
		return queueStatusResponse{}, false
	}
	position := queuePositionLocked(s.queue, id)
	if position > 0 {
		return queueStatusResponse{ID: id, Status: "queued", QueuePosition: position}, true
	}
	return queueStatusResponse{ID: id, Status: "ready", QueuePosition: 0}, true
}

func (s *Server) promoteNextLocked(now time.Time) {
	for len(s.queue) > 0 {
		id := s.queue[0]
		s.queue = s.queue[1:]
		request, exists := s.sessions[id]
		if !exists {
			continue
		}
		request.CreatedAt = now
		s.sessions[id] = request
		s.activeID = id
		return
	}
}

func (s *Server) removeQueuedLocked(id string) {
	for index, queuedID := range s.queue {
		if queuedID == id {
			s.queue = append(s.queue[:index], s.queue[index+1:]...)
			return
		}
	}
}

func queuePositionLocked(queue []string, id string) int {
	for index, queuedID := range queue {
		if queuedID == id {
			return index + 1
		}
	}
	return 0
}

func (s *Server) hasActiveTest() bool {
	s.mu.Lock()
	defer s.mu.Unlock()
	s.expireActiveLocked(time.Now().UTC())
	return s.activeID != ""
}

func (s *Server) expireActiveLocked(now time.Time) {
	if s.activeID == "" {
		s.promoteNextLocked(now)
		return
	}
	request, exists := s.sessions[s.activeID]
	if !exists {
		s.activeID = ""
		s.promoteNextLocked(now)
		return
	}
	if now.Sub(request.CreatedAt) > request.Duration*2+60*time.Second {
		if session, exists := s.iperfSessions[s.activeID]; exists {
			stopIperf3Process(session)
			delete(s.iperfSessions, s.activeID)
		}
		delete(s.sessions, s.activeID)
		s.activeID = ""
		s.promoteNextLocked(now)
	}
}

func (s *Server) isActive(id string) bool {
	if id == "" {
		return false
	}
	s.mu.Lock()
	defer s.mu.Unlock()
	return s.activeID == id
}

func (s *Server) resolveClient(ctx context.Context, ip string) recordstore.ClientInfo {
	if ip == "" {
		return normalizeClientInfo(recordstore.ClientInfo{IP: "unknown", Source: "request"}, "")
	}
	if s.clientResolver == nil {
		return normalizeClientInfo(recordstore.ClientInfo{IP: ip, Source: "request"}, ip)
	}
	info, err := s.clientResolver(ctx, ip)
	if err != nil {
		return normalizeClientInfo(recordstore.ClientInfo{IP: ip, Source: "request"}, ip)
	}
	if info.IP == "" {
		info.IP = ip
	}
	return normalizeClientInfo(info, ip)
}

func sanitizeRecords(records []recordstore.Record) []recordstore.Record {
	out := make([]recordstore.Record, len(records))
	for i, record := range records {
		record.Client = normalizeClientInfo(record.Client, record.Client.IP)
		out[i] = record
	}
	return out
}

func normalizeClientInfo(info recordstore.ClientInfo, fallbackIP string) recordstore.ClientInfo {
	if strings.TrimSpace(info.IP) == "" {
		info.IP = fallbackIP
	}
	info.IP = anonymizeIP(info.IP)
	if strings.TrimSpace(info.ISP) == "" {
		info.ISP = "未知运营商"
	}
	if strings.TrimSpace(info.Source) == "" {
		info.Source = "request"
	}
	return info
}

func anonymizeIP(value string) string {
	raw := strings.TrimSpace(value)
	if raw == "" {
		return "unknown"
	}
	if host, _, err := net.SplitHostPort(raw); err == nil {
		raw = host
	}
	raw = strings.Trim(raw, "[]")
	ip := net.ParseIP(raw)
	if ip == nil {
		return raw
	}
	if ipv4 := ip.To4(); ipv4 != nil {
		return fmt.Sprintf("%d.%d.%d.*", ipv4[0], ipv4[1], ipv4[2])
	}
	ipv6 := ip.To16()
	if ipv6 == nil {
		return raw
	}
	return fmt.Sprintf("%x:%x:%x:*",
		uint16(ipv6[0])<<8|uint16(ipv6[1]),
		uint16(ipv6[2])<<8|uint16(ipv6[3]),
		uint16(ipv6[4])<<8|uint16(ipv6[5]),
	)
}

func clientIPFromRequest(r *http.Request, trustProxyHeaders bool) string {
	if trustProxyHeaders {
		for _, header := range []string{"CF-Connecting-IP", "X-Real-IP"} {
			value := strings.TrimSpace(r.Header.Get(header))
			if value != "" {
				return value
			}
		}
		if forwarded := strings.TrimSpace(r.Header.Get("X-Forwarded-For")); forwarded != "" {
			ip := strings.TrimSpace(strings.Split(forwarded, ",")[0])
			if ip != "" {
				return ip
			}
		}
	}
	host, _, err := net.SplitHostPort(r.RemoteAddr)
	if err == nil {
		return host
	}
	return strings.TrimSpace(r.RemoteAddr)
}

func eventSessionID(path string) (string, bool) {
	if !strings.HasPrefix(path, "/api/tests/") || !strings.HasSuffix(path, "/events") {
		return "", false
	}
	id := strings.TrimSuffix(strings.TrimPrefix(path, "/api/tests/"), "/events")
	id = strings.Trim(id, "/")
	return id, id != ""
}

func completeSessionID(path string) (string, bool) {
	if !strings.HasPrefix(path, "/api/tests/") || !strings.HasSuffix(path, "/complete") {
		return "", false
	}
	id := strings.TrimSuffix(strings.TrimPrefix(path, "/api/tests/"), "/complete")
	id = strings.Trim(id, "/")
	return id, id != ""
}

func cancelSessionID(path string) (string, bool) {
	if !strings.HasPrefix(path, "/api/tests/") || !strings.HasSuffix(path, "/cancel") {
		return "", false
	}
	id := strings.TrimSuffix(strings.TrimPrefix(path, "/api/tests/"), "/cancel")
	id = strings.Trim(id, "/")
	return id, id != ""
}

func queueSessionID(path string) (string, bool) {
	if !strings.HasPrefix(path, "/api/tests/") || !strings.HasSuffix(path, "/queue") {
		return "", false
	}
	id := strings.TrimSuffix(strings.TrimPrefix(path, "/api/tests/"), "/queue")
	id = strings.Trim(id, "/")
	return id, id != ""
}

func completeIperf3SessionID(path string) (string, bool) {
	if !strings.HasPrefix(path, "/api/iperf3/sessions/") || !strings.HasSuffix(path, "/complete") {
		return "", false
	}
	id := strings.TrimSuffix(strings.TrimPrefix(path, "/api/iperf3/sessions/"), "/complete")
	id = strings.Trim(id, "/")
	return id, id != ""
}

func iperfHostFromRequest(r *http.Request) string {
	host := strings.TrimSpace(r.Host)
	if splitHost, _, err := net.SplitHostPort(host); err == nil {
		host = splitHost
	}
	host = strings.Trim(host, "[]")
	if host == "" {
		return "127.0.0.1"
	}
	return host
}

func isDownloadBinaryName(name string) bool {
	switch name {
	case "netstable-linux-amd64", "netstable-linux-arm64", "netstable-linux-amd64.tar.gz", "netstable-linux-arm64.tar.gz":
		return true
	default:
		return false
	}
}

func currentLinuxBinaryName() string {
	if runtime.GOOS != "linux" {
		return ""
	}
	return "netstable-linux-" + runtime.GOARCH
}

func serveBinaryFile(w http.ResponseWriter, r *http.Request, name string, path string) {
	w.Header().Set("Content-Type", "application/octet-stream")
	w.Header().Set("Content-Disposition", fmt.Sprintf(`attachment; filename="%s"`, name))
	http.ServeFile(w, r, path)
}

func allocateHighPort() (int, error) {
	listener, err := net.Listen("tcp", "0.0.0.0:0")
	if err != nil {
		return 0, err
	}
	defer listener.Close()
	_, portText, err := net.SplitHostPort(listener.Addr().String())
	if err != nil {
		return 0, err
	}
	port, err := strconv.Atoi(portText)
	if err != nil {
		return 0, err
	}
	if port < 20000 {
		return allocateHighPort()
	}
	return port, nil
}

func stopIperf3Process(session *iperfSession) {
	if session == nil || session.Process == nil || session.Process.Process == nil {
		return
	}
	_ = session.Process.Process.Kill()
}

func summarizeBrowserSamples(request TestRequest, raw completeTestRequest) probe.Summary {
	startedAt := raw.StartedAt
	if startedAt.IsZero() {
		startedAt = request.CreatedAt
	}
	finishedAt := raw.FinishedAt
	if finishedAt.IsZero() || finishedAt.Before(startedAt) {
		finishedAt = time.Now().UTC()
	}

	summary := probe.Summary{
		ID:         request.ID,
		Target:     request.Target.Address,
		HTTPURL:    request.HTTPURL,
		StartedAt:  startedAt,
		FinishedAt: finishedAt,
		DurationMS: finishedAt.Sub(startedAt).Milliseconds(),
		Samples:    len(raw.Samples),
	}

	var latencySum float64
	var latencyCount int
	var minLatency float64
	var maxLatency float64
	var lastLatency float64
	var jitterSum float64
	var jitterPairs int

	for _, sample := range raw.Samples {
		if !sample.Success {
			summary.Failures++
			continue
		}
		summary.Successes++
		switch sample.Kind {
		case "latency":
			latency := sample.LatencyMS
			latencySum += latency
			latencyCount++
			if latencyCount == 1 || latency < minLatency {
				minLatency = latency
			}
			if latency > maxLatency {
				maxLatency = latency
			}
			if latencyCount > 1 {
				jitterSum += math.Abs(latency - lastLatency)
				jitterPairs++
			}
			lastLatency = latency
		}
	}

	if summary.Samples > 0 {
		summary.PacketLossPct = round2(float64(summary.Failures) / float64(summary.Samples) * 100)
	}
	if latencyCount > 0 {
		summary.AvgLatencyMS = round2(latencySum / float64(latencyCount))
		summary.MinLatencyMS = round2(minLatency)
		summary.MaxLatencyMS = round2(maxLatency)
	}
	if jitterPairs > 0 {
		summary.JitterMS = round2(jitterSum / float64(jitterPairs))
	}
	avgDownload, maxDownload, downloadCount := transferStats(raw.Samples, "download")
	if downloadCount > 0 {
		summary.AvgDownloadMbps = round2(avgDownload)
		summary.MaxDownloadMbps = round2(maxDownload)
	}
	avgUpload, maxUpload, uploadCount := transferStats(raw.Samples, "upload")
	if uploadCount > 0 {
		summary.AvgUploadMbps = round2(avgUpload)
		summary.MaxUploadMbps = round2(maxUpload)
	}
	summary.Score = browserScore(summary)
	return summary
}

type transferPoint struct {
	at        time.Time
	bytes     int64
	mbps      float64
	latencyMS float64
}

type transferSegment struct {
	bits    float64
	seconds float64
}

const sustainedPeakWindowSeconds = 5

func transferStats(samples []probe.Sample, kind string) (avgMbps float64, maxMbps float64, count int) {
	var points []transferPoint
	for _, sample := range samples {
		if !sample.Success || sample.Kind != kind {
			continue
		}
		count++
		if sample.Bytes <= 0 && sample.Mbps <= 0 {
			continue
		}
		points = append(points, transferPoint{
			at:        sample.At,
			bytes:     sample.Bytes,
			mbps:      sample.Mbps,
			latencyMS: sample.LatencyMS,
		})
	}
	if len(points) == 0 {
		return 0, 0, count
	}
	sort.Slice(points, func(i, j int) bool {
		return points[i].at.Before(points[j].at)
	})

	var totalBits float64
	var totalSeconds float64
	var fallbackSum float64
	var fallbackCount int
	var segments []transferSegment
	for index, point := range points {
		seconds := transferSampleSeconds(point)
		if seconds <= 0 {
			seconds = transferIntervalSeconds(points, index)
		}

		rate := point.mbps
		if point.bytes > 0 && seconds > 0 {
			bits := float64(point.bytes) * 8
			rate = bits / seconds / 1_000_000
			totalBits += bits
			totalSeconds += seconds
			segments = append(segments, transferSegment{bits: bits, seconds: seconds})
		} else if rate > 0 {
			fallbackSum += rate
			fallbackCount++
		}
	}

	if totalSeconds > 0 {
		avgMbps = totalBits / totalSeconds / 1_000_000
	} else if fallbackCount > 0 {
		avgMbps = fallbackSum / float64(fallbackCount)
	}
	maxMbps = sustainedPeakMbps(segments, avgMbps)
	if maxMbps == 0 && fallbackCount > 0 {
		maxMbps = avgMbps
	}
	return avgMbps, maxMbps, count
}

func sustainedPeakMbps(segments []transferSegment, fallbackAvg float64) float64 {
	var totalSeconds float64
	for _, segment := range segments {
		totalSeconds += segment.seconds
	}
	if totalSeconds <= 0 {
		return 0
	}
	if totalSeconds < sustainedPeakWindowSeconds {
		return fallbackAvg
	}

	var maxMbps float64
	for start := range segments {
		var bits float64
		var seconds float64
		for end := start; end < len(segments); end++ {
			bits += segments[end].bits
			seconds += segments[end].seconds
			if seconds >= sustainedPeakWindowSeconds {
				mbps := bits / seconds / 1_000_000
				if mbps > maxMbps {
					maxMbps = mbps
				}
			}
		}
	}
	if maxMbps == 0 {
		return fallbackAvg
	}
	return maxMbps
}

func transferIntervalSeconds(points []transferPoint, index int) float64 {
	if len(points) < 2 || points[index].at.IsZero() {
		return 0
	}
	if index == 0 {
		if points[1].at.IsZero() {
			return 0
		}
		return points[1].at.Sub(points[0].at).Seconds()
	}
	if points[index-1].at.IsZero() {
		return 0
	}
	return points[index].at.Sub(points[index-1].at).Seconds()
}

func transferSampleSeconds(point transferPoint) float64 {
	if point.latencyMS > 0 {
		return point.latencyMS / 1000
	}
	if point.bytes > 0 && point.mbps > 0 {
		return float64(point.bytes) * 8 / (point.mbps * 1_000_000)
	}
	return 0
}

func browserScore(summary probe.Summary) float64 {
	if summary.Samples == 0 {
		return 0
	}
	score := 100.0
	score -= summary.PacketLossPct * 1.5
	score -= math.Min(summary.AvgLatencyMS/20, 20)
	score -= math.Min(summary.JitterMS/5, 15)
	if summary.AvgDownloadMbps == 0 {
		score -= 10
	}
	if summary.AvgUploadMbps == 0 {
		score -= 10
	}
	if score < 0 {
		score = 0
	}
	return round2(score)
}

func round2(value float64) float64 {
	return math.Round(value*100) / 100
}

func writeSSE(w http.ResponseWriter, event Event) error {
	eventType := event.Type
	if eventType == "" {
		eventType = "message"
	}

	var data any
	switch {
	case event.Sample != nil:
		data = event.Sample
	case event.Summary != nil:
		data = event.Summary
	case event.Error != "":
		data = errorResponse{Error: event.Error}
	default:
		data = event
	}

	payload, err := json.Marshal(data)
	if err != nil {
		return err
	}
	return writeSSEPayload(w, eventType, payload)
}

func writeSSEData(w http.ResponseWriter, eventType string, data any) error {
	payload, err := json.Marshal(data)
	if err != nil {
		return err
	}
	return writeSSEPayload(w, eventType, payload)
}

func writeSSEPayload(w http.ResponseWriter, eventType string, payload []byte) error {
	if eventType == "" {
		eventType = "message"
	}
	_, err := fmt.Fprintf(w, "event: %s\ndata: %s\n\n", eventType, payload)
	return err
}

func writeJSON(w http.ResponseWriter, status int, payload any) {
	w.Header().Set("Content-Type", "application/json; charset=utf-8")
	w.WriteHeader(status)
	_ = json.NewEncoder(w).Encode(payload)
}

func newID() string {
	var bytes [8]byte
	if _, err := rand.Read(bytes[:]); err == nil {
		return hex.EncodeToString(bytes[:])
	}
	return strconv.FormatInt(time.Now().UnixNano(), 36)
}