package openai

import (
	"bytes"
	"context"
	"encoding/json"
	"errors"
	"go.uber.org/zap"
	edgeservice "iop/apps/edge/internal/service"
	"iop/packages/go/config"
	"net/http"
	"strconv"
	"strings"
)

// chatHotPathDispositionPolicy is the OpenAI Chat projection of the common
// Hot Path disposition. Commit state is intentionally absent: the Chat codec
// selects either the normal JSON error/status contract or the committed SSE
// error + [DONE] sequence without changing these semantics.
type chatHotPathDispositionPolicy struct {
	status        int
	errorType     string
	finishReason  string
	silent        bool
	errorTerminal bool
}

func chatHotPathPolicy(disposition hotPathTerminalDisposition) chatHotPathDispositionPolicy {
	switch disposition.Kind {
	case hotPathDispositionSuccess:
		return chatHotPathDispositionPolicy{status: http.StatusOK, finishReason: "stop"}
	case hotPathDispositionToolTurn:
		return chatHotPathDispositionPolicy{status: http.StatusOK, finishReason: "tool_calls"}
	case hotPathDispositionLength:
		return chatHotPathDispositionPolicy{status: http.StatusOK, finishReason: "length"}
	case hotPathDispositionValidationError:
		return chatHotPathDispositionPolicy{
			status: http.StatusBadRequest, errorType: "invalid_request_error", errorTerminal: true,
		}
	case hotPathDispositionProviderError, hotPathDispositionTimeout:
		return chatHotPathDispositionPolicy{
			status: http.StatusBadGateway, errorType: "run_error", errorTerminal: true,
		}
	case hotPathDispositionCallerCancel:
		return chatHotPathDispositionPolicy{silent: true}
	default:
		return chatHotPathDispositionPolicy{
			status: http.StatusBadGateway, errorType: "run_error", errorTerminal: true,
		}
	}
}

func (s *Server) handleChatCompletions(w http.ResponseWriter, r *http.Request) {
	if r.Method != http.MethodPost {
		writeError(w, http.StatusMethodNotAllowed, "method_not_allowed", "method not allowed")
		return
	}
	defer r.Body.Close()

	// Apply the request-start policy before reading. MaxBytesReader performs a
	// limit+1 probe, so no unbounded body exists before the Core snapshot.
	rawBody, err := readOpenAIIngressBody(w, r, s.maxIngressSnapshotBytes())
	if err != nil {
		writeOpenAIIngressError(w, err)
		return
	}

	// The route decision runs before strict normalization: only the
	// routing-relevant envelope fields are decoded leniently so provider
	// passthrough can forward Codex/provider-specific payloads verbatim.
	// Strict field validation stays on the non-provider path.
	env, err := decodeChatCompletionEnvelope(rawBody)
	if err != nil {
		writeError(w, http.StatusBadRequest, "invalid_request_error", err.Error())
		return
	}
	dispatch, err := s.resolveRouteDispatchForPrincipal(r.Context(), env.Model)
	if err != nil {
		writeError(w, http.StatusBadRequest, "invalid_request_error", "model is required")
		return
	}

	providerNativeThinking := chatRequestHasProviderNativeThinking(rawBody)
	var req chatCompletionRequest
	if dispatch.ProviderPool {
		// Provider-pool path: lenient decode so caller unknown/Codex/
		// provider-specific fields survive into the tunnel body.
		if err := decodeChatCompletionRequestLenient(json.NewDecoder(bytes.NewReader(rawBody)), &req); err != nil {
			writeError(w, http.StatusBadRequest, "invalid_request_error", err.Error())
			return
		}
	} else {
		// Non-provider path keeps the strict allowlist-based decode.
		if err := decodeChatCompletionRequest(json.NewDecoder(bytes.NewReader(rawBody)), &req); err != nil {
			writeError(w, http.StatusBadRequest, "invalid_request_error", err.Error())
			return
		}
	}
	providerTunnelRoute := routeUsesProviderTunnel(dispatch)
	basePrompt := promptFromMessages(req.Messages)
	if strings.TrimSpace(basePrompt) == "" {
		writeError(w, http.StatusBadRequest, "invalid_request_error", "messages are required")
		return
	}
	if dispatch.SingleRequest != nil {
		capability, ok := s.service.(singleRequestService)
		if !ok {
			writeError(w, http.StatusServiceUnavailable, "run_error", "single-request execution is unavailable")
			return
		}
		recordSingleRequestIngress()
		if req.Stream {
			s.handleChatSingleRequestStream(w, r, capability, dispatch, rawBody)
		} else {
			s.handleChatSingleRequest(w, r, capability, dispatch, rawBody)
		}
		return
	}
	outputPolicy := s.resolveOutputPolicy(basePrompt)
	// The mutable model-catalog generation policy is applied here, at the single
	// ingress point, before req is frozen into the dispatch context. Every stage
	// downstream reads the post-policy request.
	if catalogEntry := s.findProviderPoolEntry(dispatch.effectiveModelGroupKey(req.Model)); catalogEntry != nil {
		applyModelCatalogGenerationPolicyToChat(&req, *catalogEntry, outputPolicy.Strict, providerNativeThinking)
	}

	// Freeze canonical bytes plus the effective typed request before identity
	// derivation or any provider admission. A typed-view overflow is therefore
	// a zero-dispatch request-too-large failure.
	ingress, err := buildOpenAIIngressSnapshot(s.maxIngressSnapshotBytes(), rawBody, req)
	if err != nil {
		writeOpenAIIngressError(w, err)
		return
	}
	defer ingress.Close()
	rawBody = nil

	runMeta, err := resolveCallerIdentity(r, dispatch, req.Metadata)
	if err != nil {
		writeError(w, http.StatusBadRequest, "invalid_request_error", err.Error())
		return
	}
	if dispatch.IsPreset {
		applyHotPathOutputTokenCap(runMeta, req.MaxTokens, req.MaxCompletionTokens)
		presetChatCodec := newHotPathChatOuterCodec(req.Stream, req.Model, hotPathOutputTokenCap(runMeta))
		r = withHotPathChatOuterCodec(r, presetChatCodec)
	}

	var presetIngress presetIngressResult
	if dispatch.IsPreset {
		rawBytes, err := ingress.canonicalBody()
		if err != nil {
			writeError(w, http.StatusBadRequest, "invalid_request_error", err.Error())
			return
		}
		presetIngress, err = s.joinPresetChatIngress(r, dispatch, rawBytes, runMeta)
		if err != nil {
			writeError(w, http.StatusBadRequest, "invalid_request_error", err.Error())
			return
		}
		if presetIngress.localStageEligible() {
			_ = s.runHotPathLocalEligible(w, r, dispatch, "openai", req.Stream, runMeta)
			return
		}
		if presetIngress.lightStageContinuation() {
			_ = s.runHotPathLightContinuation(w, r, dispatch, "openai", req.Stream, runMeta)
			return
		}
		if presetIngress.cleanupIssued() {
			_ = s.writeHotPathStageResponse(w, r, dispatch, "openai", req.Stream, presetIngress.Cleanup.RequestID, presetIngress.Cleanup.Output)
			return
		}
		if presetIngress.terminalReady() {
			_ = s.writeHotPathTerminal(w, r, dispatch, "openai", req.Stream, runMeta["iop_logical_request_id"], *presetIngress.Terminal)
			return
		}
	}

	// The response path is decided by the resolved route, never by caller
	// metadata: provider routes relay pure passthrough over the raw tunnel;
	// every other route uses the normalized RunEvent path. Caller metadata is
	// arbitrary context and does not select route or response shape (SDD S01).
	// Pre-compute estimate/contextClass early so provider-pool path can use them.
	estimate := s.estimateChatInputTokens(basePrompt, runMeta, req.Tools, req.ToolChoice)
	contextClass := classifyContext(estimate, s.longContextThreshold())

	requestCtx := openAIRequestContext{
		r:              r,
		route:          dispatch,
		ingress:        ingress,
		callerMetadata: runMeta,
		estimate:       estimate,
		contextClass:   contextClass,
		endpoint:       usageEndpointChatCompletions,
		usage:          s.newOpenAIUsageRecorder(r.Context(), req.Model, usageEndpointChatCompletions),
	}

	// Provider-pool: use one-shot SubmitProviderPool which dispatches either
	// tunnel or normalized based on the selected provider's executionPath.
	dc := s.newChatDispatchContext(requestCtx, req, basePrompt, outputPolicy)
	if dispatch.ProviderPool {
		s.handleChatCompletionsProviderPool(w, dc)
		return
	}

	if providerTunnelRoute {
		s.tunnelChatCompletionPassthrough(w, dc)
		return
	}

	handle, err := s.service.SubmitRun(r.Context(), dc.submitReq)
	if err != nil {
		dc.finishUsageRequest(usageStatusForError(err), responseModeNormalized)
		writeError(w, http.StatusBadGateway, "node_dispatch_error", err.Error())
		return
	}
	s.logChatDispatch("openai chat completion dispatch", handle.Dispatch())

	dc = dc.withRetrySubmit(func(ctx context.Context, req edgeservice.SubmitRunRequest) (any, error) {
		return s.service.SubmitRun(ctx, req)
	})
	if req.Stream {
		s.streamChatCompletion(w, dc, handle)
		return
	}
	s.completeChatCompletion(w, dc, handle)
}

// newChatDispatchContext freezes the ingress-resolved values into the immutable
// dispatch context for the resolved route. The provider-pool route leaves
// adapter/target unset: the service layer resolves them per candidate at
// admission and rewrites the target to the winning provider's served model.
func (s *Server) newChatDispatchContext(requestCtx openAIRequestContext, req chatCompletionRequest, basePrompt string, outputPolicy strictOutputPolicy) *chatDispatchContext {
	dc := &chatDispatchContext{
		openAIRequestContext: requestCtx,
		req:                  req,
		basePrompt:           basePrompt,
		outputPolicy:         outputPolicy,
		validation:           toolValidationContractFromRequest(req),
	}

	messages := req.Messages
	if !requestCtx.route.ProviderPool {
		// Strict output is a normalized-dispatch contract only; the pool route
		// decides the path after admission, so the instruction is not baked in
		// here (SDD D02).
		if instruction := strictOutputContractInstruction(outputPolicy); instruction != "" {
			messages = prependSystemMessage(messages, instruction)
		}
	}
	dc.prompt = promptFromMessages(messages)
	dc.input = req.runInput(dc.prompt, messages, outputPolicy.Strict, routeSupportsNativeToolCalls(requestCtx.route))

	if !requestCtx.route.ProviderPool {
		// Re-estimate against the instruction-augmented prompt for the
		// normalized path.
		dc.estimate = s.estimateChatInputTokens(dc.prompt, requestCtx.callerMetadata, req.Tools, req.ToolChoice)
		dc.contextClass = classifyContext(dc.estimate, s.longContextThreshold())
	}

	dc.runMetadata = chatRunMetadata(cloneMetadata(requestCtx.callerMetadata), req, outputPolicy)
	dc.runMetadata["estimated_input_tokens"] = strconv.Itoa(dc.estimate)
	dc.runMetadata["context_class"] = dc.contextClass

	if requestCtx.route.ProviderPool {
		modelGroupKey := requestCtx.route.effectiveModelGroupKey(req.Model)
		if requestCtx.route.IsPreset && strings.TrimSpace(requestCtx.route.Preset.Selector.Model) != "" {
			modelGroupKey = presetSelectorModelGroupKey(requestCtx.route, req.Model)
		}
		dc.submitReq = edgeservice.SubmitRunRequest{
			NodeRef:              requestCtx.route.NodeRef,
			ModelGroupKey:        modelGroupKey,
			ProviderID:           requestCtx.route.ProviderID,
			UsageAttribution:     requestCtx.route.UsageAttribution,
			SessionID:            requestCtx.route.SessionID,
			Prompt:               dc.prompt,
			Input:                dc.input,
			TimeoutSec:           requestCtx.route.TimeoutSec,
			MaxQueue:             requestCtx.route.MaxQueue,
			QueueTimeoutMS:       requestCtx.route.QueueTimeoutMS,
			Metadata:             dc.runMetadata,
			EstimatedInputTokens: dc.estimate,
			ContextClass:         dc.contextClass,
			ProviderPool:         true,
		}
		return dc
	}

	if dc.validation.enabled {
		dc.runMetadata = toolValidationAttemptMetadata(dc.runMetadata, 1, "", "")
	}
	dc.submitReq = chatSubmitRunRequest(requestCtx.route, req, dc.prompt, dc.input, dc.runMetadata)
	dc.submitReq.EstimatedInputTokens = dc.estimate
	dc.submitReq.ContextClass = dc.contextClass
	return dc
}

func (s *Server) logChatDispatch(msg string, disp edgeservice.RunDispatch, extra ...zap.Field) {
	fields := []zap.Field{
		zap.String("run_id", disp.RunID),
		zap.String("node_id", disp.NodeID),
		zap.String("provider_id", disp.ProviderID),
		zap.String("credential_slot_ref", disp.CredentialSlotRef),
		zap.Uint64("credential_revision", disp.CredentialRevision),
		zap.String("provider_type", disp.ProviderType),
		zap.String("execution_path", disp.ExecutionPath),
		zap.String("model_group", disp.ModelGroupKey),
		zap.String("adapter", disp.Adapter),
		zap.String("target", disp.Target),
		zap.Int("estimated_input_tokens", disp.EstimatedInputTokens),
		zap.String("context_class", disp.ContextClass),
		zap.String("queue_reason", disp.QueueReason),
	}
	s.logger.Info(msg, append(fields, extra...)...)
}

// handleChatCompletionsProviderPool dispatches a provider-pool chat completion
// through the one-shot SubmitProviderPool surface. After a single queue admission,
// the selected candidate's executionPath determines whether the response is
// delivered via tunnel passthrough or the normalized RunEvent path.
//
// Provider auth is validated via providerTunnelAuthHeaders which is called for
// tunnel-path only. For normalized paths, the auth check is skipped (SDD D01).
func (s *Server) handleChatCompletionsProviderPool(w http.ResponseWriter, dc *chatDispatchContext) {
	r := dc.r
	req := dc.req
	modelGroupKey := dc.route.effectiveModelGroupKey(req.Model)
	if dc.route.IsPreset && strings.TrimSpace(dc.route.Preset.Selector.Model) != "" {
		modelGroupKey = presetSelectorModelGroupKey(dc.route, req.Model)
	}
	poolReq := edgeservice.ProviderPoolDispatchRequest{
		Run: dc.submitReq,
		Tunnel: edgeservice.SubmitProviderTunnelRequest{
			CredentialBinding:    dc.route.credentialBinding(),
			ModelGroupKey:        modelGroupKey,
			ProviderID:           dc.route.ProviderID,
			UsageAttribution:     dc.route.UsageAttribution,
			SessionID:            dc.route.SessionID,
			Method:               http.MethodPost,
			Path:                 "/v1/chat/completions",
			Operation:            string(config.OperationChatCompletions),
			Stream:               req.Stream,
			TimeoutSec:           dc.route.TimeoutSec,
			MaxQueue:             dc.route.MaxQueue,
			QueueTimeoutMS:       dc.route.QueueTimeoutMS,
			Metadata:             dc.runMetadata,
			EstimatedInputTokens: dc.estimate,
			ContextClass:         dc.contextClass,
			ProviderPool:         true,
		},
	}

	if s.streamGateSemanticEnabled() {
		fctx, err := s.openAIChatOutputFilterContext(dc)
		if err != nil {
			dc.finishUsageRequest(usageStatusError, responseModePassthrough)
			writeError(w, http.StatusInternalServerError, "run_error", "stream gate runtime unavailable")
			return
		}
		predicate, err := openAIStreamGateCandidatePredicate(s.streamGateConfig(), fctx)
		if err != nil {
			dc.finishUsageRequest(usageStatusError, responseModePassthrough)
			writeError(w, http.StatusInternalServerError, "run_error", "stream gate runtime unavailable")
			return
		}
		poolReq.AcceptCandidate = predicate
	}
	if dc.route.Managed {
		poolReq.AcceptCandidate = composeCandidatePredicates(poolReq.AcceptCandidate, dc.route.CandidatePredicate())
	}

	// Pre-dispatch provider auth header injection. Runs inside SubmitProviderPool
	// BEFORE buildProviderTunnelRequest and the Node Send step, so the auth
	// header lands on the actual wire request. On failure the slot is released
	// and no tunnel request is sent (SDD S03).
	poolReq.PrepareTunnel = func(tunnelReq edgeservice.SubmitProviderTunnelRequest) (edgeservice.SubmitProviderTunnelRequest, error) {
		headers, err := s.providerTunnelAuthHeaders(r)
		if err != nil {
			return tunnelReq, err
		}
		tunnelReq.Headers = headers
		return tunnelReq, nil
	}
	baseProtocolPreparer := s.protocolTunnelPreparer(r, config.OperationChatCompletions)
	selectorInstruction, err := s.hotPathSelectorProviderInstruction(dc.runMetadata)
	if err != nil {
		dc.finishUsageRequest(usageStatusError, responseModePassthrough)
		writeError(w, http.StatusInternalServerError, "run_error", "selector provider instruction is unavailable")
		return
	}
	poolReq.PrepareProtocolTunnel = func(tunnelReq edgeservice.SubmitProviderTunnelRequest, selected edgeservice.ProviderPoolCandidate) (edgeservice.SubmitProviderTunnelRequest, error) {
		prepared, err := baseProtocolPreparer(tunnelReq, selected)
		if err != nil {
			return tunnelReq, err
		}
		if strings.TrimSpace(selectorInstruction) != "" {
			prepared, err = prepareHotPathSelectorProviderInstruction(prepared, selectorInstruction)
			if err != nil {
				return tunnelReq, err
			}
			prepared, err = prepareHotPathSelectorOutputLimit(prepared)
			if err != nil {
				return tunnelReq, err
			}
			prepared, err = prepareHotPathSelectorCanonicalTools(prepared, selectorInstruction, dc.route.Preset)
			if err != nil {
				return tunnelReq, err
			}
			requirements := chatProviderRequirements(req)
			if effort, ok := dc.route.Preset.Selector.Options["reasoning_effort"].(string); ok && strings.TrimSpace(effort) != "" {
				requirements.Effort = strings.TrimSpace(effort)
			}
			return prepareHotPathChatProviderOperation(prepared, selected, requirements)
		}
		prepared, err = prepareProviderChatRequestNormalization(prepared, selected)
		if err != nil {
			return tunnelReq, err
		}
		prepared, err = prepareProviderChatToolCallNormalization(prepared, selected)
		if err != nil {
			return tunnelReq, err
		}
		return prepared, nil
	}

	// strict-output output policy only applies to normalized dispatch,
	// not to raw tunnel passthrough (SDD D02). The caller's raw body is the
	// passthrough source; only the model field is rewritten to the served target.
	bodyBuilder := newOpenAIProviderBodyBuilder(func(target string) (*openAIRebuiltLease, error) {
		return rewriteChatCompletionModelFromIngress(dc.ingress, target, req)
	})
	poolReq.Tunnel.BuildBody = bodyBuilder.BuildBody

	// Normalized branch: chat completions use lenient decode so unknown
	// fields survive; PrepareRun passes through the already-built Run
	// without strict validation (SDD S02/S03). The tunnel branch is
	// unaffected.
	poolReq.PrepareRun = func(runReq edgeservice.SubmitRunRequest) (edgeservice.SubmitRunRequest, error) {
		return runReq, nil
	}

	result, err := s.service.SubmitProviderPool(r.Context(), poolReq)
	bodyBuilder.Close()
	if err != nil {
		dc.finishUsageRequest(usageStatusForError(err), responseModePassthrough)
		// Provider auth failure is a client request error (400), not a
		// backend dispatch error. The auth check now runs inside
		// SubmitProviderPool (PrepareTunnel) before any tunnel request is sent.
		if isProviderCredentialClientError(err) {
			writeError(w, http.StatusBadRequest, "invalid_request_error", providerCredentialClientMessage(err))
			return
		}
		if errors.Is(err, edgeservice.ErrProviderPoolCandidateRejected) {
			writeError(w, http.StatusBadRequest, "invalid_request_error", openAIStreamGateCandidateRejectedMessage)
			return
		}
		writeError(w, http.StatusBadGateway, "node_dispatch_error", err.Error())
		return
	}

	s.logChatDispatch("openai chat completion provider-pool dispatch", result.DispatchInfo,
		zap.String("path", string(result.Path)),
	)
	if presetHotPathEnabled(dc.route) {
		mode := responseModeNormalized
		if result.Path == edgeservice.ProviderPoolPathTunnel {
			mode = responseModePassthrough
		}
		presetChatCodec := hotPathChatOuterCodecFromRequest(r)
		if presetChatCodec == nil {
			s.terminalPresetRequest(dc.runMetadata["iop_logical_request_id"], s.edgeIDValue())
			dc.finishUsageRequest(usageStatusError, mode)
			writeError(w, http.StatusInternalServerError, "run_error", "Chat outer codec is unavailable")
			return
		}
		stage, collected, turnErr := presetChatCodec.runInitialPresetTurn(s, w, r, dc.route, dc.runMetadata, result)
		if !collected {
			s.terminalPresetRequest(dc.runMetadata["iop_logical_request_id"], s.edgeIDValue())
			dc.finishUsageRequest(usageStatusForError(turnErr), mode)
			disposition, ok := hotPathDispositionFromError(turnErr)
			if !ok {
				disposition = hotPathTerminalDisposition{
					Kind: hotPathDispositionForError(turnErr), Cause: turnErr.Error(), Source: "selector_collection",
				}
			}
			errorType := "run_error"
			if disposition.Kind == hotPathDispositionValidationError {
				errorType = "invalid_request_error"
			}
			_ = presetChatCodec.writeDisposition(
				w, disposition, httpStatusForRunError(turnErr), errorType, turnErr.Error(),
			)
			return
		}
		dc.recordUsageAttempt(result.DispatchInfo, mode, usageObservationFromOpenAIUsage(stage.OpenAIUsage, len(stage.Reasoning)))
		if turnErr != nil {
			dc.finishUsageRequest(usageStatusError, mode)
			return
		}
		dc.finishUsageRequest(usageStatusSuccess, mode)
		return
	}

	// The Core request runtime owns the whole response for both
	// selected paths. The initial admission result becomes the initial attempt
	// binding, and every recovery re-enters SubmitProviderPool through the same
	// runtime, so the actual provider/model/execution path may still change
	// while the transport is uncommitted.
	s.runOpenAIChatPoolStreamGate(w, dc.withPoolDispatch(poolReq), result)
}
