package sfcrmappservice

import (
	"bytes"
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"io"
	"net/http"
	"net/textproto"
	"net/url"
	"regexp"
	"strings"

	crmwire "github.com/pingcap-inc/nexuscrm-app-packs/sfcrm/wire"
	mutationsemantic "github.com/pingcap-inc/nexuscrm-contract/pkg/runtime/recordmutationv1"
)

const compositeRequestLimit = 25

var compositeReferencePattern = regexp.MustCompile(`@\{([^}]+)\}`)

type genericCompositeSubrequest struct {
	Body        json.RawMessage   `json:"body,omitempty"`
	Method      string            `json:"method"`
	URL         string            `json:"url"`
	ReferenceID string            `json:"referenceId"`
	HTTPHeaders map[string]string `json:"httpHeaders,omitempty"`
}

type genericCompositeRequest struct {
	AllOrNone        bool                         `json:"allOrNone"`
	CompositeRequest []genericCompositeSubrequest `json:"compositeRequest"`
}

type genericCompositeSubresponse struct {
	Body           any               `json:"body"`
	HTTPHeaders    map[string]string `json:"httpHeaders,omitempty"`
	HTTPStatusCode int               `json:"httpStatusCode"`
	ReferenceID    string            `json:"referenceId"`
}

type genericCompositeResponse struct {
	CompositeResponse []genericCompositeSubresponse `json:"compositeResponse"`
}

type genericCompositeBatchRequest struct {
	BatchRequests []genericCompositeSubrequest `json:"batchRequests"`
	HaltOnError   bool                         `json:"haltOnError,omitempty"`
}

type genericCompositeBatchSubresponse struct {
	StatusCode  int               `json:"statusCode"`
	ReferenceID string            `json:"referenceId,omitempty"`
	Result      any               `json:"result"`
	HTTPHeaders map[string]string `json:"httpHeaders,omitempty"`
}

type genericCompositeBatchResponse struct {
	HasErrors bool                               `json:"hasErrors"`
	Results   []genericCompositeBatchSubresponse `json:"results"`
}

type compositeRoute int

const (
	compositeRouteNone compositeRoute = iota
	compositeRouteDependent
	compositeRouteBatch
)

func matchCompositeRoute(path string) (compositeRoute, string) {
	parts := strings.Split(strings.Trim(path, "/"), "/")
	if len(parts) != 5 || parts[0] != "services" || parts[1] != "data" ||
		!supportedAPIVersion(parts[2]) || parts[3] != "composite" {
		return compositeRouteNone, ""
	}
	switch parts[4] {
	case "batch":
		return compositeRouteBatch, parts[2]
	case "":
		return compositeRouteNone, ""
	default:
		return compositeRouteNone, ""
	}
}

func matchCompositeRouteWithRoot(path string) (compositeRoute, string) {
	parts := strings.Split(strings.Trim(path, "/"), "/")
	if len(parts) == 4 && parts[0] == "services" && parts[1] == "data" &&
		supportedAPIVersion(parts[2]) && parts[3] == "composite" {
		return compositeRouteDependent, parts[2]
	}
	return matchCompositeRoute(path)
}

func (s *Service) serveComposite(writer http.ResponseWriter, request *http.Request) bool {
	route, version := matchCompositeRouteWithRoot(request.URL.EscapedPath())
	if route == compositeRouteNone {
		return false
	}
	if request.Method != http.MethodPost {
		writer.Header().Set("Allow", http.MethodPost)
		writeCRMError(writer, http.StatusMethodNotAllowed, crmwire.SFCodeInvalidRequest, "Method not allowed")
		return true
	}
	s.requests.Add(1)
	if s.draining.Load() {
		s.drainRejected.Add(1)
		s.failures.Add(1)
		writeCRMError(writer, http.StatusServiceUnavailable, crmwire.SFCodeUnknownException, "Service is draining")
		return true
	}
	if _, ok := endUserBearer(request.Header.Values("Authorization")); !ok {
		s.localRejected.Add(1)
		s.failures.Add(1)
		writeCRMError(writer, http.StatusUnauthorized, crmwire.SFCodeInvalidSessionID, "Session expired or invalid")
		return true
	}
	s.inFlight.Add(1)
	defer s.inFlight.Add(-1)
	ctx, cancel := context.WithTimeout(request.Context(), s.requestTimeout)
	defer cancel()
	request = request.WithContext(ctx)

	request.Body = http.MaxBytesReader(writer, request.Body, s.maxRequestBytes)
	decoder := json.NewDecoder(request.Body)
	switch route {
	case compositeRouteDependent:
		var body genericCompositeRequest
		if err := decoder.Decode(&body); err != nil || requireJSONEOF(decoder) != nil {
			s.rejectComposite(writer, "Invalid request body")
			return true
		}
		if len(body.CompositeRequest) > compositeRequestLimit {
			s.rejectCompositeLimit(writer)
			return true
		}
		response := s.executeComposite(request, version, body)
		s.successes.Add(1)
		writeJSON(writer, http.StatusOK, response)
	case compositeRouteBatch:
		var body genericCompositeBatchRequest
		if err := decoder.Decode(&body); err != nil || requireJSONEOF(decoder) != nil {
			s.rejectComposite(writer, "Invalid request body")
			return true
		}
		if len(body.BatchRequests) > compositeRequestLimit {
			s.rejectCompositeLimit(writer)
			return true
		}
		response := s.executeCompositeBatch(request, version, body)
		s.successes.Add(1)
		writeJSON(writer, http.StatusOK, response)
	}
	return true
}

func (s *Service) rejectComposite(writer http.ResponseWriter, message string) {
	s.localRejected.Add(1)
	s.failures.Add(1)
	writeCRMError(writer, http.StatusBadRequest, crmwire.SFCodeInvalidRequest, message)
}

func (s *Service) rejectCompositeLimit(writer http.ResponseWriter) {
	s.localRejected.Add(1)
	s.failures.Add(1)
	writeCRMError(writer, http.StatusBadRequest, crmwire.SFCodeLimitExceeded, "Maximum 25 subrequests allowed")
}

func (s *Service) executeComposite(request *http.Request, version string, body genericCompositeRequest) genericCompositeResponse {
	if body.AllOrNone {
		if response, handled := s.executeAtomicMutationComposite(request, version, body); handled {
			return response
		}
	}
	resultsByReference := make(map[string]any)
	response := genericCompositeResponse{CompositeResponse: make([]genericCompositeSubresponse, 0, len(body.CompositeRequest))}
	for _, subrequest := range body.CompositeRequest {
		if isCompositeDispatchURL(subrequest.URL) {
			response.CompositeResponse = append(response.CompositeResponse, genericCompositeSubresponse{
				HTTPStatusCode: http.StatusBadRequest,
				ReferenceID:    subrequest.ReferenceID,
				Body:           crmwire.SFErrorResponse(crmwire.SFCodeInvalidRequest, "Composite subrequest cannot target composite endpoints"),
				HTTPHeaders:    map[string]string{},
			})
			if body.AllOrNone {
				break
			}
			continue
		}
		resolvedURL := resolveCompositeReferences(subrequest.URL, resultsByReference)
		resolvedBody := subrequest.Body
		if len(resolvedBody) != 0 {
			resolvedBody = json.RawMessage(resolveCompositeBodyReferences(string(resolvedBody), resultsByReference))
		}
		result, err := s.dispatchCompositeSubrequest(request, version, subrequest.Method, resolvedURL, resolvedBody, subrequest.HTTPHeaders, true)
		if err != nil {
			result = genericCompositeSubresponse{
				HTTPStatusCode: http.StatusBadRequest,
				ReferenceID:    subrequest.ReferenceID,
				Body:           crmwire.SFErrorResponse(crmwire.SFCodeInvalidRequest, err.Error()),
				HTTPHeaders:    map[string]string{},
			}
		} else {
			result.ReferenceID = subrequest.ReferenceID
		}
		response.CompositeResponse = append(response.CompositeResponse, result)
		if subrequest.ReferenceID != "" {
			resultsByReference[subrequest.ReferenceID] = result.Body
		}
		if result.HTTPStatusCode >= http.StatusBadRequest && body.AllOrNone {
			break
		}
	}
	return response
}

type atomicCompositeMutation struct {
	route       mutationRoute
	operation   mutationsemantic.Operation
	referenceID string
}

// executeAtomicMutationComposite recognizes the launch-required allOrNone
// profile made only of independent single-record mutations. Those operations
// must cross L0 as one existing neutral atomic Mutate command; recursive HTTP
// dispatch would create one transaction per subrequest and could not roll back
// an earlier success. Read-only composite requests continue through the normal
// dispatcher. Mixed or reference-dependent mutation shapes fail before any
// effect because the current neutral mutation contract cannot prove them
// atomic.
func (s *Service) executeAtomicMutationComposite(
	request *http.Request,
	version string,
	body genericCompositeRequest,
) (genericCompositeResponse, bool) {
	mutations := make([]atomicCompositeMutation, 0, len(body.CompositeRequest))
	firstNonMutation := -1
	for index, subrequest := range body.CompositeRequest {
		mutation, recognized, err := decodeAtomicCompositeMutation(version, subrequest)
		if err != nil {
			return atomicCompositePreflightFailure(body.CompositeRequest, index, err), true
		}
		if !recognized {
			if firstNonMutation < 0 {
				firstNonMutation = index
			}
			continue
		}
		mutations = append(mutations, mutation)
	}
	if len(mutations) == 0 {
		return genericCompositeResponse{}, false
	}
	if len(mutations) != len(body.CompositeRequest) {
		return atomicCompositePreflightFailure(
			body.CompositeRequest,
			firstNonMutation,
			errors.New("allOrNone composite cannot mix record mutations with other subrequests"),
		), true
	}
	if s.mutation == nil {
		return atomicCompositeRuntimeFailure(
			mutations,
			&mutationsemantic.Failure{Category: mutationsemantic.FailureUnavailable},
		), true
	}
	bearer, ok := endUserBearer(request.Header.Values("Authorization"))
	if !ok {
		return atomicCompositeRuntimeFailure(
			mutations,
			&mutationsemantic.Failure{Category: mutationsemantic.FailureUnauthenticated},
		), true
	}
	commandID, err := mutationCommandID(request.Header.Values(crmMutationIdempotencyHeader))
	if err != nil {
		return atomicCompositePreflightFailure(body.CompositeRequest, 0, err), true
	}
	operations := make([]mutationsemantic.Operation, len(mutations))
	for index := range mutations {
		operations[index] = mutations[index].operation
	}
	result, err := s.mutation.Mutate(
		request.Context(), bearer, commandID,
		mutationsemantic.MutationUseCaseCollection,
		mutationsemantic.BatchAtomic,
		operations,
	)
	if err != nil {
		return atomicCompositeRuntimeFailure(mutations, err), true
	}
	response := genericCompositeResponse{
		CompositeResponse: make([]genericCompositeSubresponse, len(mutations)),
	}
	for index := range mutations {
		response.CompositeResponse[index] = projectAtomicCompositeMutation(
			mutations[index], result.Results[index],
		)
	}
	return response, true
}

func decodeAtomicCompositeMutation(
	version string,
	subrequest genericCompositeSubrequest,
) (atomicCompositeMutation, bool, error) {
	path := strings.TrimSpace(subrequest.URL)
	if path == "" {
		return atomicCompositeMutation{}, false, nil
	}
	if !strings.HasPrefix(path, "/") {
		path = "/services/data/" + version + "/" + path
	}
	parsed, err := url.ParseRequestURI(path)
	if err != nil || parsed.IsAbs() || parsed.Host != "" {
		return atomicCompositeMutation{}, false, nil
	}
	method := strings.TrimSpace(subrequest.Method)
	if fileRoute, _, filesOwned := matchRecordFilesRoute(parsed.EscapedPath(), method); filesOwned && fileRoute.kind == recordFilesRouteDelete {
		return atomicCompositeMutation{}, true, errors.New(
			"allOrNone Files delete is not supported",
		)
	}
	route, _, recognized := matchMutationRoute(parsed.Path, method)
	if !recognized || route.kind == "" {
		return atomicCompositeMutation{}, false, nil
	}
	if route.useCase != mutationsemantic.MutationUseCaseSingle {
		return atomicCompositeMutation{}, true, errors.New("allOrNone nested collection mutation is not supported")
	}
	if parsed.RawQuery != "" || len(subrequest.HTTPHeaders) != 0 {
		return atomicCompositeMutation{}, true, errors.New("allOrNone record mutation has unsupported query parameters or subrequest headers")
	}
	if compositeReferencePattern.MatchString(subrequest.URL) ||
		compositeReferencePattern.Match(subrequest.Body) {
		return atomicCompositeMutation{}, true, errors.New("allOrNone reference-dependent record mutation is not supported")
	}
	operation := mutationsemantic.Operation{
		Kind: route.kind, ObjectID: route.objectID, RecordID: route.recordID,
	}
	if route.kind != mutationsemantic.OperationDelete {
		var source map[string]any
		decoder := json.NewDecoder(bytes.NewReader(subrequest.Body))
		decoder.UseNumber()
		if err := decoder.Decode(&source); err != nil {
			return atomicCompositeMutation{}, true, errors.New("invalid allOrNone record mutation body")
		}
		fields, err := mutationFields(source, false)
		if err != nil {
			return atomicCompositeMutation{}, true, err
		}
		operation.Fields = fields
	}
	return atomicCompositeMutation{
		route: route, operation: operation, referenceID: subrequest.ReferenceID,
	}, true, nil
}

func atomicCompositePreflightFailure(
	subrequests []genericCompositeSubrequest,
	index int,
	cause error,
) genericCompositeResponse {
	if index < 0 || index >= len(subrequests) {
		index = 0
	}
	referenceID := ""
	if len(subrequests) != 0 {
		referenceID = subrequests[index].ReferenceID
	}
	return genericCompositeResponse{CompositeResponse: []genericCompositeSubresponse{{
		HTTPStatusCode: http.StatusBadRequest,
		ReferenceID:    referenceID,
		Body: crmwire.SFErrorResponse(
			crmwire.SFCodeInvalidRequest,
			cause.Error(),
		),
		HTTPHeaders: map[string]string{},
	}}}
}

func atomicCompositeRuntimeFailure(
	mutations []atomicCompositeMutation,
	cause error,
) genericCompositeResponse {
	recorder := newCompositeResponseRecorder()
	writeRuntimeFailure(recorder, cause)
	referenceID := ""
	if len(mutations) != 0 {
		referenceID = mutations[0].referenceID
	}
	return genericCompositeResponse{CompositeResponse: []genericCompositeSubresponse{{
		HTTPStatusCode: recorder.statusCode(),
		ReferenceID:    referenceID,
		Body:           parseCompositeDispatchBody(recorder.body.Bytes()),
		HTTPHeaders:    flattenCompositeHeaders(recorder.header),
	}}}
}

func projectAtomicCompositeMutation(
	mutation atomicCompositeMutation,
	result mutationsemantic.OperationResult,
) genericCompositeSubresponse {
	if result.Success {
		response := genericCompositeSubresponse{
			HTTPStatusCode: http.StatusNoContent,
			ReferenceID:    mutation.referenceID,
			HTTPHeaders:    map[string]string{},
		}
		if result.Kind == mutationsemantic.OperationCreate {
			response.HTTPStatusCode = http.StatusCreated
			response.Body = map[string]any{
				"id": result.RecordID, "success": true, "errors": []any{},
			}
		}
		return response
	}
	recorder := newCompositeResponseRecorder()
	writeSingleMutationFailures(recorder, mutation.route, result.Errors)
	return genericCompositeSubresponse{
		HTTPStatusCode: recorder.statusCode(),
		ReferenceID:    mutation.referenceID,
		Body:           parseCompositeDispatchBody(recorder.body.Bytes()),
		HTTPHeaders:    flattenCompositeHeaders(recorder.header),
	}
}

func (s *Service) executeCompositeBatch(request *http.Request, version string, body genericCompositeBatchRequest) genericCompositeBatchResponse {
	response := genericCompositeBatchResponse{Results: make([]genericCompositeBatchSubresponse, len(body.BatchRequests))}
	for index, subrequest := range body.BatchRequests {
		if isCompositeDispatchURL(subrequest.URL) {
			response.HasErrors = true
			response.Results[index] = genericCompositeBatchSubresponse{
				StatusCode: http.StatusBadRequest, ReferenceID: subrequest.ReferenceID,
				Result: crmwire.SFErrorResponse(crmwire.SFCodeInvalidRequest, "Composite subrequest cannot target composite endpoints"),
			}
			if body.HaltOnError {
				break
			}
			continue
		}
		result, err := s.dispatchCompositeSubrequest(request, version, subrequest.Method, subrequest.URL, subrequest.Body, subrequest.HTTPHeaders, false)
		if err != nil {
			response.HasErrors = true
			response.Results[index] = genericCompositeBatchSubresponse{
				StatusCode: http.StatusBadRequest, ReferenceID: subrequest.ReferenceID,
				Result: crmwire.SFErrorResponse(crmwire.SFCodeInvalidRequest, err.Error()),
			}
		} else {
			response.Results[index] = genericCompositeBatchSubresponse{
				StatusCode: result.HTTPStatusCode, ReferenceID: subrequest.ReferenceID,
				Result: result.Body, HTTPHeaders: result.HTTPHeaders,
			}
			if result.HTTPStatusCode >= http.StatusBadRequest {
				response.HasErrors = true
			}
		}
		if response.HasErrors && body.HaltOnError {
			break
		}
	}
	return response
}

func (s *Service) dispatchCompositeSubrequest(parent *http.Request, version, method, target string, body []byte, headers map[string]string, forwardBrowserHeaders bool) (genericCompositeSubresponse, error) {
	path := strings.TrimSpace(target)
	if path == "" {
		return genericCompositeSubresponse{}, fmt.Errorf("composite subrequest URL is required")
	}
	if !strings.HasPrefix(path, "/") {
		path = "/services/data/" + version + "/" + path
	}
	parsed, err := url.ParseRequestURI(path)
	if err != nil || parsed.IsAbs() || parsed.Host != "" || !strings.HasPrefix(parsed.Path, "/services/data/") {
		return genericCompositeSubresponse{}, fmt.Errorf("invalid composite subrequest URL")
	}
	// Check the normalized path too. Relative targets such as "composite"
	// acquire the services/data prefix above and must not recurse through the
	// AppService a second time.
	if isCompositeDispatchURL(parsed.Path) {
		return genericCompositeSubresponse{}, fmt.Errorf("composite subrequest cannot target composite endpoints")
	}
	subrequest, err := http.NewRequestWithContext(parent.Context(), strings.TrimSpace(method), parsed.RequestURI(), bytes.NewReader(body))
	if err != nil {
		return genericCompositeSubresponse{}, err
	}
	if authorization := parent.Header.Get("Authorization"); authorization != "" {
		subrequest.Header.Set("Authorization", authorization)
	}
	if forwardBrowserHeaders {
		for _, key := range []string{"Origin", "Referer", "X-CSRF-Token"} {
			if value := parent.Header.Get(key); value != "" {
				subrequest.Header.Set(key, value)
			}
		}
	}
	subrequest.Header.Set("Content-Type", "application/json")
	for key, value := range headers {
		canonical := textproto.CanonicalMIMEHeaderKey(strings.TrimSpace(key))
		if canonical == "Authorization" {
			return genericCompositeSubresponse{}, fmt.Errorf("composite subrequest cannot override authorization")
		}
		if canonical != "" {
			subrequest.Header.Set(canonical, value)
		}
	}

	recorder := newCompositeResponseRecorder()
	s.ServeHTTP(recorder, subrequest)
	return genericCompositeSubresponse{
		HTTPStatusCode: recorder.statusCode(),
		HTTPHeaders:    flattenCompositeHeaders(recorder.header),
		Body:           parseCompositeDispatchBody(recorder.body.Bytes()),
	}, nil
}

type compositeResponseRecorder struct {
	header http.Header
	body   bytes.Buffer
	status int
}

func newCompositeResponseRecorder() *compositeResponseRecorder {
	return &compositeResponseRecorder{header: make(http.Header)}
}

func (r *compositeResponseRecorder) Header() http.Header { return r.header }

func (r *compositeResponseRecorder) WriteHeader(status int) {
	if r.status == 0 {
		r.status = status
	}
}

func (r *compositeResponseRecorder) Write(value []byte) (int, error) {
	if r.status == 0 {
		r.status = http.StatusOK
	}
	return r.body.Write(value)
}

func (r *compositeResponseRecorder) statusCode() int {
	if r.status == 0 {
		return http.StatusOK
	}
	return r.status
}

func flattenCompositeHeaders(header http.Header) map[string]string {
	flattened := make(map[string]string, len(header))
	for key, values := range header {
		if len(values) != 0 {
			flattened[key] = values[0]
		}
	}
	return flattened
}

func parseCompositeDispatchBody(body []byte) any {
	if len(body) == 0 {
		return nil
	}
	var parsed any
	decoder := json.NewDecoder(bytes.NewReader(body))
	decoder.UseNumber()
	if err := decoder.Decode(&parsed); err != nil {
		return strings.TrimSuffix(string(body), "\n")
	}
	var trailing any
	if err := decoder.Decode(&trailing); !errors.Is(err, io.EOF) {
		return strings.TrimSuffix(string(body), "\n")
	}
	return parsed
}

func resolveCompositeReferences(input string, results map[string]any) string {
	return resolveCompositeReferencesWithString(
		input,
		results,
		func(value string) string { return value },
	)
}

// resolveCompositeBodyReferences substitutes references inside the already
// validated JSON body representation. Salesforce reference expressions in
// request bodies occupy JSON string content, so string values must be escaped
// without adding a second pair of quotes. URL substitution deliberately keeps
// its existing raw-string behavior above.
func resolveCompositeBodyReferences(input string, results map[string]any) string {
	return resolveCompositeReferencesWithString(
		input,
		results,
		func(value string) string {
			encoded, err := json.Marshal(value)
			if err != nil || len(encoded) < 2 {
				return value
			}
			return string(encoded[1 : len(encoded)-1])
		},
	)
}

func resolveCompositeReferencesWithString(
	input string,
	results map[string]any,
	stringValue func(string) string,
) string {
	if len(results) == 0 || !strings.Contains(input, "@{") {
		return input
	}
	return compositeReferencePattern.ReplaceAllStringFunc(input, func(match string) string {
		inner := match[2 : len(match)-1]
		parts := strings.SplitN(inner, ".", 2)
		if len(parts) != 2 {
			return match
		}
		value := extractCompositeField(results[parts[0]], parts[1])
		if value == nil {
			return match
		}
		switch typed := value.(type) {
		case string:
			return stringValue(typed)
		case float64:
			if typed == float64(int64(typed)) {
				return fmt.Sprintf("%d", int64(typed))
			}
			return fmt.Sprint(typed)
		case bool:
			return fmt.Sprint(typed)
		default:
			encoded, err := json.Marshal(typed)
			if err != nil {
				return match
			}
			return string(encoded)
		}
	})
}

func extractCompositeField(value any, path string) any {
	current := value
	for _, part := range strings.Split(path, ".") {
		object, ok := current.(map[string]any)
		if !ok {
			return nil
		}
		current, ok = object[part]
		if !ok {
			return nil
		}
	}
	return current
}

func isCompositeDispatchURL(raw string) bool {
	path := raw
	if index := strings.Index(path, "?"); index >= 0 {
		path = path[:index]
	}
	path = strings.TrimRight(path, "/")
	return strings.HasSuffix(path, "/composite") ||
		strings.HasSuffix(path, "/composite/batch") ||
		strings.HasSuffix(path, "/composite/graph")
}

var _ io.Writer = (*compositeResponseRecorder)(nil)