package connector_callout

import (
	"context"
	"encoding/json"
	"fmt"
	"net/http"
	"strings"
)

// BatchRead sends requests in chunks of BatchSpec.MaxBatchSize, posts each
// chunk's IDs in the request body under IDsRequestField, and aggregates the
// raw JSON response bodies.
//
// BatchRead always sets Content-Type: application/json and Accept: application/json
// on the outgoing CalloutRequest when the caller has not already set them. The
// body is always JSON-marshaled, so these headers are never wrong for a
// BatchRead call. Callers that need to override them can pre-populate
// CalloutRequest.Headers before calling Engine.Do directly.
//
// Note: Engine.Do intentionally does NOT auto-set Content-Type because
// arbitrary Do callers may send non-JSON bodies (form, binary, etc.). Callers
// that use Do directly must set req.Headers["Content-Type"] themselves.
//
// Returns ErrEmptyBatch when ids is empty, ErrInvalidBatchSpec when any
// required BatchSpec field is missing or invalid. Per-chunk errors include
// the id-index range.
func (e *Engine) BatchRead(ctx context.Context, spec BatchSpec, ids []string, policy RetryPolicy) ([]json.RawMessage, error) {
	return e.batchRead(ctx, spec, ids, policy, validateBatchSpec)
}

// BatchReadLegacyRequestOrderDemux sends batch requests for historical drain-only
// profiles authored before BatchSpec.IDResponseField existed. It intentionally
// validates every executable request field except IDResponseField; callers must
// demux the returned records by request order.
func (e *Engine) BatchReadLegacyRequestOrderDemux(ctx context.Context, spec BatchSpec, ids []string, policy RetryPolicy) ([]json.RawMessage, error) {
	return e.batchRead(ctx, spec, ids, policy, validateBatchSpecForLegacyRequestOrderDemux)
}

func (e *Engine) batchRead(ctx context.Context, spec BatchSpec, ids []string, policy RetryPolicy, validate func(BatchSpec) error) ([]json.RawMessage, error) {
	if len(ids) == 0 {
		return nil, ErrEmptyBatch
	}
	// Normalize leading/trailing whitespace before validation so that padded
	// values (" POST ", " /batch ") are accepted and used trimmed downstream,
	// rather than passing the non-empty check with padding that then causes
	// http.NewRequestWithContext to reject the padded method or path.
	spec.Method = strings.TrimSpace(spec.Method)
	spec.RelativePath = strings.TrimSpace(spec.RelativePath)
	spec.IDsRequestField = strings.TrimSpace(spec.IDsRequestField)
	spec.CredentialRef = strings.TrimSpace(spec.CredentialRef)
	spec.EndpointClass = strings.TrimSpace(spec.EndpointClass)
	spec.RecordsResponsePath = strings.TrimSpace(spec.RecordsResponsePath)
	spec.IDResponseField = strings.TrimSpace(spec.IDResponseField)
	if err := validate(spec); err != nil {
		return nil, err
	}

	var results []json.RawMessage
	for start := 0; start < len(ids); start += spec.MaxBatchSize {
		end := start + spec.MaxBatchSize
		if end > len(ids) {
			end = len(ids)
		}
		chunk := ids[start:end]

		// Build the request body so ExtraBodyFields are preserved byte-for-byte
		// (no float64 coercion of JSON numbers) alongside the IDs array.
		// IDsRequestField always wins; an ExtraBodyFields entry that is not valid
		// JSON fails closed rather than being silently dropped.
		bodyFields := make(map[string]any, 1+len(spec.ExtraBodyFields))
		for k, v := range spec.ExtraBodyFields {
			if k == spec.IDsRequestField {
				continue
			}
			if !json.Valid(v) {
				return nil, fmt.Errorf("connector_callout: ExtraBodyFields[%q] is not valid JSON", k)
			}
			bodyFields[k] = json.RawMessage(v)
		}
		bodyFields[spec.IDsRequestField] = chunk
		bodyBytes, err := json.Marshal(bodyFields)
		if err != nil {
			return nil, fmt.Errorf("chunk [%d, %d): marshal body: %w", start, end, err)
		}

		// Set Content-Type and Accept for JSON batch requests. BatchRead always
		// marshals a JSON body; vendors that require these headers (and most do)
		// will reject the request without them. Don't overwrite caller-set values.
		//
		// Seed headers from BatchSpec.ExtraHeaders first so that caller-supplied
		// values (e.g. a custom Authorization header for AuthTypeCustom callers)
		// are present before the Content-Type / Accept defaulting runs. The
		// defaulting only writes headers that are still absent.
		headers := make(http.Header)
		for k, vv := range spec.ExtraHeaders {
			for _, v := range vv {
				headers.Add(k, v)
			}
		}
		if headers.Get("Content-Type") == "" {
			headers.Set("Content-Type", "application/json")
		}
		if headers.Get("Accept") == "" {
			headers.Set("Accept", "application/json")
		}

		req := CalloutRequest{
			Method:        spec.Method,
			RelativePath:  spec.RelativePath,
			CredentialRef: spec.CredentialRef,
			EndpointClass: spec.EndpointClass,
			Headers:       headers,
			QueryParams:   spec.QueryParams,
			Body:          bodyBytes,
		}
		resp, err := e.Do(ctx, req, policy)
		if err != nil {
			return nil, &BatchChunkError{Start: start, End: end, Response: resp, Err: err}
		}
		results = append(results, json.RawMessage(resp.Body))
	}
	return results, nil
}

// ValidateBatchSpec reports whether spec is executable by Engine.BatchRead
// after the same whitespace normalization BatchRead applies at runtime.
func ValidateBatchSpec(spec BatchSpec) error {
	spec.Method = strings.TrimSpace(spec.Method)
	spec.RelativePath = strings.TrimSpace(spec.RelativePath)
	spec.IDsRequestField = strings.TrimSpace(spec.IDsRequestField)
	spec.CredentialRef = strings.TrimSpace(spec.CredentialRef)
	spec.EndpointClass = strings.TrimSpace(spec.EndpointClass)
	spec.RecordsResponsePath = strings.TrimSpace(spec.RecordsResponsePath)
	spec.IDResponseField = strings.TrimSpace(spec.IDResponseField)
	return validateBatchSpec(spec)
}

// ValidateBatchSpecForLegacyRequestOrderDemux validates the historical
// drain-only batch shape that predates IDResponseField. New authoring and active
// runtime paths must use ValidateBatchSpec instead.
func ValidateBatchSpecForLegacyRequestOrderDemux(spec BatchSpec) error {
	spec.Method = strings.TrimSpace(spec.Method)
	spec.RelativePath = strings.TrimSpace(spec.RelativePath)
	spec.IDsRequestField = strings.TrimSpace(spec.IDsRequestField)
	spec.CredentialRef = strings.TrimSpace(spec.CredentialRef)
	spec.EndpointClass = strings.TrimSpace(spec.EndpointClass)
	spec.RecordsResponsePath = strings.TrimSpace(spec.RecordsResponsePath)
	spec.IDResponseField = strings.TrimSpace(spec.IDResponseField)
	return validateBatchSpecForLegacyRequestOrderDemux(spec)
}

// validateBatchSpec enforces the full BatchSpec invariant set at engine entry.
// Defense-in-depth: install-time validation in
// controlplane/packs/connector_callout_profile_component.go also enforces
// this, but a caller invoking BatchRead directly with a hand-built BatchSpec
// must still fail closed for blank required fields.
func validateBatchSpec(spec BatchSpec) error {
	return validateBatchSpecWithOptions(spec, true)
}

func validateBatchSpecForLegacyRequestOrderDemux(spec BatchSpec) error {
	return validateBatchSpecWithOptions(spec, false)
}

func validateBatchSpecWithOptions(spec BatchSpec, requireIDResponseField bool) error {
	if spec.MaxBatchSize < 1 {
		return fmt.Errorf("%w: MaxBatchSize must be >= 1, got %d", ErrInvalidBatchSpec, spec.MaxBatchSize)
	}
	if strings.TrimSpace(spec.Method) == "" {
		return fmt.Errorf("%w: Method required", ErrInvalidBatchSpec)
	}
	// Reject methods that Engine.Do drops bodies for. Engine.Do only discards
	// req.Body for GET and HEAD (HTTP semantics: some proxies/servers reject
	// GET-with-body, and HEAD responses carry no body). DELETE, PATCH, POST,
	// PUT, TRACE, and OPTIONS all pass through with a body intact.
	// Aligning with Engine.Do's actual body semantics: reject ONLY GET/HEAD.
	method := strings.ToUpper(strings.TrimSpace(spec.Method))
	switch method {
	case http.MethodGet, http.MethodHead:
		return fmt.Errorf("%w: method %q does not carry a request body; BatchRead requires a body-bearing method (DELETE, PATCH, POST, PUT are all valid)", ErrInvalidBatchSpec, method)
	}
	// Empty RelativePath is permitted: Engine.Do treats it as the credential
	// base URL root, which is a valid batch endpoint shape (e.g. POST against
	// the NamedCredential base URL directly).
	if strings.TrimSpace(spec.IDsRequestField) == "" {
		return fmt.Errorf("%w: IDsRequestField required", ErrInvalidBatchSpec)
	}
	if strings.TrimSpace(spec.CredentialRef) == "" {
		return fmt.Errorf("%w: CredentialRef required", ErrInvalidBatchSpec)
	}
	if strings.TrimSpace(spec.EndpointClass) == "" {
		return fmt.Errorf("%w: EndpointClass required", ErrInvalidBatchSpec)
	}
	if requireIDResponseField && strings.TrimSpace(spec.IDResponseField) == "" {
		return fmt.Errorf("%w: IDResponseField required", ErrInvalidBatchSpec)
	}
	if strings.TrimSpace(spec.IDResponseField) != "" {
		if err := validateDottedJSONPath(strings.TrimSpace(spec.IDResponseField)); err != nil {
			return fmt.Errorf("%w: invalid IDResponseField: %v", ErrInvalidBatchSpec, err)
		}
	}
	if trimmed := strings.TrimSpace(spec.RecordsResponsePath); trimmed != "" {
		if err := validateDottedJSONPath(trimmed); err != nil {
			return fmt.Errorf("%w: invalid RecordsResponsePath: %v", ErrInvalidBatchSpec, err)
		}
	}
	for key, values := range spec.InactiveProbeQueryParams {
		if strings.TrimSpace(key) == "" {
			return fmt.Errorf("%w: InactiveProbeQueryParams contains a blank key", ErrInvalidBatchSpec)
		}
		if len(values) == 0 {
			return fmt.Errorf("%w: InactiveProbeQueryParams[%q] requires at least one value", ErrInvalidBatchSpec, key)
		}
	}
	return nil
}