story/internal/httpapi/provider_callbacks.go

package httpapi

import (
	"bytes"
	"context"
	"encoding/json"
	"errors"
	"io"
	"log/slog"
	"net/http"
	"regexp"
	"time"

	"data-terminal-cloud/internal/orchestration"
	"data-terminal-cloud/internal/voiceprovider"
)

type ProviderCallbackProcessor interface {
	Process(context.Context, voiceprovider.CallbackEvent, []byte) (orchestration.ProviderEventResult, error)
}

func VoiceProviderCallbackHandler(secret []byte, maxSkew time.Duration, processor ProviderCallbackProcessor) http.Handler {
	return http.HandlerFunc(func(writer http.ResponseWriter, request *http.Request) {
		rawBody, err := io.ReadAll(request.Body)
		if err != nil {
			WriteError(writer, request, http.StatusBadRequest, "INVALID_CALLBACK", "供应商通知无法读取")
			return
		}
		if err := voiceprovider.VerifyCallback(
			secret, request.Header.Get(voiceprovider.HeaderCallbackTimestamp),
			request.Header.Get(voiceprovider.HeaderCallbackSignature), rawBody, time.Now(), maxSkew,
		); err != nil {
			WriteError(writer, request, http.StatusUnauthorized, "PROVIDER_CALLBACK_AUTHENTICATION_FAILED", "供应商通知身份验证失败")
			return
		}
		decoder := json.NewDecoder(bytes.NewReader(rawBody))
		decoder.DisallowUnknownFields()
		var event voiceprovider.CallbackEvent
		if err := decoder.Decode(&event); err != nil || ensureJSONEOF(decoder) != nil ||
			event.Provider == "" || event.EventID == "" || event.TaskRef == "" || event.ConversationID == "" ||
			!validProviderEventType(event.EventType) || !validProviderStatus(event.Status) || event.OccurredAt.IsZero() ||
			len(event.EventID) > 255 || len(event.TaskRef) > 200 || !validProviderTurns(event) || !validProviderUsage(event) {
			WriteError(writer, request, http.StatusBadRequest, "INVALID_CALLBACK", "供应商通知字段不正确")
			return
		}
		result, err := processor.Process(request.Context(), event, rawBody)
		if err != nil {
			slog.ErrorContext(request.Context(), "provider callback processing failed", "event_type", event.EventType, "provider", event.Provider, "error", err)
			switch {
			case errors.Is(err, orchestration.ErrProviderTaskNotFound):
				WriteError(writer, request, http.StatusNotFound, "PROVIDER_TASK_NOT_FOUND", "供应商任务不存在")
			case errors.Is(err, orchestration.ErrProviderEventConflict):
				WriteError(writer, request, http.StatusConflict, "PROVIDER_EVENT_CONFLICT", "供应商事件编号对应的内容不一致")
			default:
				WriteAPIError(writer, request, http.StatusServiceUnavailable, Error{Code: "PROVIDER_CALLBACK_UNAVAILABLE", Message: "供应商通知暂时无法处理", Retryable: true})
			}
			return
		}
		WriteJSON(writer, request, http.StatusOK, result)
	})
}

func validProviderEventType(value string) bool {
	return value == "voice_chat.started" || value == "voice_chat.stopped" || value == "voice_chat.failed" || value == "voice_chat.transcript.final" || value == "voice_chat.usage"
}

func validProviderUsage(event voiceprovider.CallbackEvent) bool {
	if event.EventType != "voice_chat.usage" {
		return len(event.Usage) == 0
	}
	if event.Status != "FINAL" || len(event.Usage) == 0 || len(event.Usage) > 100 {
		return false
	}
	categories := map[string]bool{"RTC": true, "END_TO_END_VOICE": true, "ASR": true, "LLM": true, "TTS": true, "STORAGE": true, "POST_PROCESSING": true}
	code := regexp.MustCompile(`^[a-z][a-z0-9_.-]{1,79}$`)
	unit := regexp.MustCompile(`^[a-z][a-z0-9_.-]{1,39}$`)
	quantity := regexp.MustCompile(`^(0|[1-9][0-9]{0,17})(\.[0-9]{1,6})?$`)
	seen := map[string]struct{}{}
	for _, metric := range event.Usage {
		key := metric.UsageCategory + ":" + metric.MetricCode + ":" + metric.Unit
		if !categories[metric.UsageCategory] || !code.MatchString(metric.MetricCode) || !unit.MatchString(metric.Unit) || !quantity.MatchString(metric.Quantity) {
			return false
		}
		if _, exists := seen[key]; exists {
			return false
		}
		seen[key] = struct{}{}
	}
	return true
}

func validProviderStatus(value string) bool {
	return value == "ACTIVE" || value == "COMPLETED" || value == "FAILED" || value == "FINAL"
}

func validProviderTurns(event voiceprovider.CallbackEvent) bool {
	if event.EventType != "voice_chat.transcript.final" {
		return len(event.Turns) == 0
	}
	if event.Status != "FINAL" || len(event.Turns) == 0 || len(event.Turns) > 500 {
		return false
	}
	seenKeys := make(map[string]struct{}, len(event.Turns))
	seenSequences := make(map[int64]struct{}, len(event.Turns))
	for _, turn := range event.Turns {
		if turn.TurnKey == "" || len(turn.TurnKey) > 255 || turn.SequenceNo <= 0 ||
			(turn.Speaker != "subject" && turn.Speaker != "agent") || turn.StartAt.IsZero() ||
			turn.EndAt.Before(turn.StartAt) || turn.FinalText == "" || len([]rune(turn.FinalText)) > 20000 || len(turn.ProviderRef) > 255 {
			return false
		}
		if _, exists := seenKeys[turn.TurnKey]; exists {
			return false
		}
		if _, exists := seenSequences[turn.SequenceNo]; exists {
			return false
		}
		seenKeys[turn.TurnKey] = struct{}{}
		seenSequences[turn.SequenceNo] = struct{}{}
	}
	return true
}