package connector_inbound

import (
	"bytes"
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"net/url"
	"strings"
	"sync"
	"time"

	"github.com/pingcap-inc/nexuscrm-connectors/features/connector/callout"
	zlog "github.com/rs/zerolog/log"
)

type calloutBatchReadAdapter struct {
	namespace   string
	engine      *connector_callout.Engine
	byObjectKey map[string]batchProfileBinding
	coalescer   *batchReadCoalescer
}

type batchProfileBinding struct {
	profile                 *connector_callout.ConnectorCalloutProfile
	relativePath            string
	legacyRequestOrderDemux bool
}

type batchResponseRecord struct {
	sourceID string
	body     json.RawMessage
}

type canonicalAliasDemuxCountError struct {
	namespace  string
	objectKind string
	got        int
	want       int
}

func (e *canonicalAliasDemuxCountError) Error() string {
	return fmt.Sprintf(
		"connector_inbound: batch response for %s/%s returned %d records for %d requested IDs; canonical alias demux requires one response record per requested ID in request order",
		e.namespace,
		e.objectKind,
		e.got,
		e.want,
	)
}

const defaultBatchReadCoalesceWindow = 25 * time.Millisecond

var ErrNoEligibleBatchProfiles = errors.New("connector_inbound: no eligible batch profiles")
var ErrMultipleEligibleBatchProfiles = errors.New("connector_inbound: multiple eligible batch profiles")

// NewCalloutBatchReadAdapter adapts connector_callout.Engine plus the
// pack-owned connector_callout profiles for one connector namespace into the
// worker-facing BatchReadClient contract.
func NewCalloutBatchReadAdapter(namespace string, engine *connector_callout.Engine, profiles []*connector_callout.ConnectorCalloutProfile) (BatchReadClient, error) {
	namespace = strings.TrimSpace(namespace)
	if namespace == "" {
		return nil, fmt.Errorf("connector_inbound: batch adapter namespace is required")
	}
	if engine == nil {
		return nil, fmt.Errorf("connector_inbound: batch adapter engine is required")
	}

	byObjectKey := make(map[string]batchProfileBinding)
	for _, profile := range profiles {
		if profile == nil {
			return nil, fmt.Errorf("connector_inbound: batch adapter profile is nil")
		}
		if strings.TrimSpace(profile.OwnerPack) != namespace {
			return nil, fmt.Errorf("connector_inbound: profile %s ownerPack %q does not match namespace %q", profile.ProfileName, profile.OwnerPack, namespace)
		}
		if !batchSpecConfiguredForProfile(profile) {
			continue
		}
		for objectKind, objectPath := range profile.ObjectPathMap {
			objectKind = strings.TrimSpace(objectKind)
			objectPath = strings.TrimSpace(objectPath)
			if objectKind == "" || objectPath == "" {
				// Blank object coverage is not executable batch coverage.
				// Treat it as absent so payload-only routes fall back cleanly and
				// required-enrichment routes fail closed in the supervisor wrapper.
				continue
			}
			if _, exists := byObjectKey[objectKind]; exists {
				return nil, fmt.Errorf("%w for %s/%s", ErrMultipleEligibleBatchProfiles, namespace, objectKind)
			}
			byObjectKey[objectKind] = batchProfileBinding{
				profile:                 profile,
				relativePath:            joinRelativePath(objectPath, profile.BatchSpec.RelativePath),
				legacyRequestOrderDemux: connector_callout.UsesLegacyRequestOrderBatchDemux(profile),
			}
		}
	}
	if len(byObjectKey) == 0 {
		return nil, fmt.Errorf("%w for namespace %s", ErrNoEligibleBatchProfiles, namespace)
	}

	return &calloutBatchReadAdapter{
		namespace:   namespace,
		engine:      engine,
		byObjectKey: byObjectKey,
		coalescer:   newBatchReadCoalescer(defaultBatchReadCoalesceWindow),
	}, nil
}

func (a *calloutBatchReadAdapter) BatchRead(ctx context.Context, namespace, objectKind string, sourceIDs []string) (map[string]json.RawMessage, error) {
	outcome, err := a.BatchReadOutcome(ctx, namespace, objectKind, sourceIDs)
	if err != nil {
		return nil, err
	}
	return CollapseBatchReadOutcome(outcome, sourceIDs)
}

func (a *calloutBatchReadAdapter) BatchReadOutcome(ctx context.Context, namespace, objectKind string, sourceIDs []string) (BatchReadOutcome, error) {
	namespace = strings.TrimSpace(namespace)
	objectKind = strings.TrimSpace(objectKind)
	if namespace != a.namespace {
		return BatchReadOutcome{}, fmt.Errorf("connector_inbound: batch adapter namespace mismatch: got %q want %q", namespace, a.namespace)
	}
	binding, ok := a.byObjectKey[objectKind]
	if !ok {
		// Some packs only enrich a subset of object kinds. The wrapper in
		// application/connectors enforces any required-enrichment contract, so the
		// adapter should fall back to payload-only flow execution here.
		return BatchReadOutcome{Details: map[string]json.RawMessage{}}, nil
	}

	if len(sourceIDs) == 1 {
		var details map[string]json.RawMessage
		var err error
		if binding.profile.BatchSpec.MaxBatchSize > 1 {
			details, err = a.coalescer.read(ctx, a, namespace, objectKind, binding, sourceIDs[0])
		} else {
			details, err = a.batchReadDirect(ctx, namespace, objectKind, binding, sourceIDs)
		}
		outcome := BatchReadOutcome{Details: details}
		if err != nil {
			outcome.SourceErrors = map[string]error{sourceIDs[0]: err}
		}
		return outcome, nil
	}

	details, err := a.batchReadDirectWithInactiveProbe(ctx, namespace, objectKind, binding, sourceIDs, false)
	if err != nil {
		if len(binding.profile.BatchSpec.InactiveProbeQueryParams) > 0 && shouldFallbackCanonicalAliasCoalescedRead(binding, err) {
			return a.batchReadIndividualOutcomes(ctx, namespace, objectKind, binding, sourceIDs), nil
		}
		return BatchReadOutcome{}, err
	}
	outcome := BatchReadOutcome{Details: details}
	if len(binding.profile.BatchSpec.InactiveProbeQueryParams) == 0 {
		return outcome, nil
	}
	probedSourceIDs := make(map[string]struct{}, len(sourceIDs))
	for _, sourceID := range sourceIDs {
		if sourceID == "" {
			continue
		}
		if _, alreadyProbed := probedSourceIDs[sourceID]; alreadyProbed {
			continue
		}
		probedSourceIDs[sourceID] = struct{}{}
		if _, found := details[sourceID]; found {
			continue
		}
		confirmed, probeErr := a.confirmSourceInactive(ctx, namespace, objectKind, binding, sourceID)
		if probeErr != nil {
			if outcome.SourceErrors == nil {
				outcome.SourceErrors = make(map[string]error)
			}
			outcome.SourceErrors[sourceID] = fmt.Errorf(
				"connector_inbound: inactive source confirmation probe for %s/%s source %q: %w",
				namespace, objectKind, sourceID, probeErr,
			)
			continue
		}
		if confirmed {
			if outcome.SourceErrors == nil {
				outcome.SourceErrors = make(map[string]error)
			}
			outcome.SourceErrors[sourceID] = fmt.Errorf(
				"%w for %s/%s source %q", ErrSourceRecordInactive,
				namespace, objectKind, sourceID,
			)
		}
	}
	return outcome, nil
}

func (a *calloutBatchReadAdapter) batchReadIndividualOutcomes(
	ctx context.Context,
	namespace string,
	objectKind string,
	binding batchProfileBinding,
	sourceIDs []string,
) BatchReadOutcome {
	outcome := BatchReadOutcome{Details: make(map[string]json.RawMessage)}
	seen := make(map[string]struct{}, len(sourceIDs))
	for _, sourceID := range sourceIDs {
		if sourceID == "" {
			continue
		}
		if _, duplicate := seen[sourceID]; duplicate {
			continue
		}
		seen[sourceID] = struct{}{}
		details, err := a.batchReadDirect(ctx, namespace, objectKind, binding, []string{sourceID})
		if err != nil {
			if outcome.SourceErrors == nil {
				outcome.SourceErrors = make(map[string]error)
			}
			outcome.SourceErrors[sourceID] = err
			continue
		}
		if body := details[sourceID]; body != nil {
			outcome.Details[sourceID] = body
		}
	}
	return outcome
}

func (a *calloutBatchReadAdapter) batchReadDirect(ctx context.Context, namespace, objectKind string, binding batchProfileBinding, sourceIDs []string) (map[string]json.RawMessage, error) {
	return a.batchReadDirectWithInactiveProbe(ctx, namespace, objectKind, binding, sourceIDs, true)
}

func (a *calloutBatchReadAdapter) batchReadDirectWithInactiveProbe(ctx context.Context, namespace, objectKind string, binding batchProfileBinding, sourceIDs []string, allowInactiveProbe bool) (map[string]json.RawMessage, error) {
	spec := binding.profile.BatchSpec
	spec.RelativePath = binding.relativePath
	var responses []json.RawMessage
	var err error
	if binding.legacyRequestOrderDemux {
		responses, err = a.engine.BatchReadLegacyRequestOrderDemux(ctx, spec, sourceIDs, binding.profile.RetryPolicy)
	} else {
		responses, err = a.engine.BatchRead(ctx, spec, sourceIDs, binding.profile.RetryPolicy)
	}
	if err != nil {
		return nil, err
	}

	out := make(map[string]json.RawMessage, len(sourceIDs))
	for chunkIdx, body := range responses {
		records, err := extractBatchResponseRecords(body, spec.RecordsResponsePath)
		if err != nil {
			return nil, fmt.Errorf("connector_inbound: parse batch response chunk %d for %s/%s: %w", chunkIdx, namespace, objectKind, err)
		}
		if binding.legacyRequestOrderDemux {
			start := chunkIdx * spec.MaxBatchSize
			end := start + spec.MaxBatchSize
			if end > len(sourceIDs) {
				end = len(sourceIDs)
			}
			if start >= len(sourceIDs) {
				return nil, fmt.Errorf("connector_inbound: legacy request-order batch response chunk %d exceeded request chunks for %s/%s", chunkIdx, namespace, objectKind)
			}
			if err := appendLegacyRequestOrderBatchDetails(out, sourceIDs[start:end], records); err != nil {
				return nil, fmt.Errorf("connector_inbound: parse legacy request-order batch response chunk %d for %s/%s: %w", chunkIdx, namespace, objectKind, err)
			}
			continue
		}
		prepared := make([]batchResponseRecord, 0, len(records))
		for recordIdx, record := range records {
			sourceID, err := extractRecordSourceID(record, spec.IDResponseField)
			if err != nil {
				return nil, fmt.Errorf("connector_inbound: parse batch response chunk %d record %d for %s/%s: %w", chunkIdx, recordIdx, namespace, objectKind, err)
			}
			// Flatten a pack-declared nested object so field-mapping sourceField
			// lookups (e.g. "email") resolve without dotted-path support. The key
			// to flatten is declared by the connector pack's BatchSpec, keeping L0
			// blind to which vendor nests fields under which key. Existing top-level
			// keys win; the nested key is removed after promotion.
			if spec.RecordFlattenKey != "" {
				nested, ok := record[spec.RecordFlattenKey].(map[string]any)
				if !ok {
					return nil, fmt.Errorf("connector_inbound: batch response record %q for %s/%s has no object at RecordFlattenKey %q (response-shape mismatch)", sourceID, namespace, objectKind, spec.RecordFlattenKey)
				}
				for k, v := range nested {
					if _, exists := record[k]; !exists {
						record[k] = v
					}
				}
				delete(record, spec.RecordFlattenKey)
			}
			recordBytes, err := json.Marshal(record)
			if err != nil {
				return nil, fmt.Errorf("connector_inbound: re-marshal batch response record %q for %s/%s: %w", sourceID, namespace, objectKind, err)
			}
			prepared = append(prepared, batchResponseRecord{
				sourceID: sourceID,
				body:     json.RawMessage(recordBytes),
			})
		}
		for _, record := range prepared {
			if _, exists := out[record.sourceID]; exists {
				if !spec.AllowCanonicalIDAliases {
					return nil, fmt.Errorf("connector_inbound: duplicate source ID %q in batch response for %s/%s", record.sourceID, namespace, objectKind)
				}
				continue
			}
			out[record.sourceID] = record.body
		}
		if spec.AllowCanonicalIDAliases {
			chunkStart := chunkIdx * spec.MaxBatchSize
			chunkEnd := chunkStart + spec.MaxBatchSize
			if chunkEnd > len(sourceIDs) {
				chunkEnd = len(sourceIDs)
			}
			requestedCount := chunkEnd - chunkStart
			if len(prepared) != requestedCount {
				countErr := &canonicalAliasDemuxCountError{
					namespace:  namespace,
					objectKind: objectKind,
					got:        len(prepared),
					want:       requestedCount,
				}
				if allowInactiveProbe && len(sourceIDs) == 1 && len(prepared) == 0 && len(spec.InactiveProbeQueryParams) > 0 {
					confirmed, probeErr := a.confirmSourceInactive(ctx, namespace, objectKind, binding, sourceIDs[0])
					if probeErr != nil {
						return nil, fmt.Errorf("connector_inbound: inactive source confirmation probe for %s/%s: %w", namespace, objectKind, probeErr)
					}
					if confirmed {
						return nil, fmt.Errorf("%w for %s/%s", ErrSourceRecordInactive, namespace, objectKind)
					}
				}
				return nil, countErr
			}
			for recordIdx, record := range prepared {
				requestIdx := chunkStart + recordIdx
				if requestIdx >= len(sourceIDs) {
					continue
				}
				requestedID := sourceIDs[requestIdx]
				if requestedID == "" {
					continue
				}
				if _, exists := out[requestedID]; !exists {
					out[requestedID] = record.body
				}
			}
		}
	}
	if allowInactiveProbe && len(sourceIDs) == 1 && len(spec.InactiveProbeQueryParams) > 0 {
		requestedID := sourceIDs[0]
		if requestedID != "" {
			if _, found := out[requestedID]; !found {
				confirmed, probeErr := a.confirmSourceInactive(ctx, namespace, objectKind, binding, requestedID)
				if probeErr != nil {
					return nil, fmt.Errorf("connector_inbound: inactive source confirmation probe for %s/%s: %w", namespace, objectKind, probeErr)
				}
				if confirmed {
					return nil, fmt.Errorf("%w for %s/%s", ErrSourceRecordInactive, namespace, objectKind)
				}
			}
		}
	}
	return out, nil
}

func (a *calloutBatchReadAdapter) confirmSourceInactive(ctx context.Context, namespace, objectKind string, binding batchProfileBinding, sourceID string) (bool, error) {
	if binding.profile == nil || len(binding.profile.BatchSpec.InactiveProbeQueryParams) == 0 {
		return false, nil
	}

	profile := *binding.profile
	probeSpec := profile.BatchSpec
	probeSpec.QueryParams = overlayQueryParams(probeSpec.QueryParams, probeSpec.InactiveProbeQueryParams)
	probeSpec.InactiveProbeQueryParams = nil
	profile.BatchSpec = probeSpec
	probeBinding := binding
	probeBinding.profile = &profile

	details, err := a.batchReadDirectWithInactiveProbe(ctx, namespace, objectKind, probeBinding, []string{sourceID}, false)
	if err != nil {
		var countErr *canonicalAliasDemuxCountError
		if errors.As(err, &countErr) && countErr.got == 0 && countErr.want == 1 {
			// A successful, well-shaped probe that still returned no record does
			// not confirm inactivity. Preserve the ordinary read's prior outcome.
			return false, nil
		}
		return false, err
	}
	_, found := details[sourceID]
	return found, nil
}

func overlayQueryParams(base, overlay url.Values) url.Values {
	merged := make(url.Values, len(base)+len(overlay))
	for key, values := range base {
		merged[key] = append([]string(nil), values...)
	}
	for key, values := range overlay {
		merged[key] = append([]string(nil), values...)
	}
	return merged
}

type batchReadCoalescer struct {
	window  time.Duration
	mu      sync.Mutex
	pending map[batchReadCoalesceKey]*batchReadCoalesceBatch
}

type batchReadCoalesceKey struct {
	namespace  string
	objectKind string
}

type batchReadCoalesceBatch struct {
	key       batchReadCoalesceKey
	ctx       context.Context
	binding   batchProfileBinding
	maxSize   int
	timer     *time.Timer
	ids       []string
	waiters   map[string][]*batchReadCoalesceWaiter
	waiterCnt int
}

type batchReadCoalesceWaiter struct {
	id   string
	done chan struct{}
	body json.RawMessage
	err  error
}

func newBatchReadCoalescer(window time.Duration) *batchReadCoalescer {
	if window <= 0 {
		window = defaultBatchReadCoalesceWindow
	}
	return &batchReadCoalescer{
		window:  window,
		pending: make(map[batchReadCoalesceKey]*batchReadCoalesceBatch),
	}
}

func (c *batchReadCoalescer) read(ctx context.Context, adapter *calloutBatchReadAdapter, namespace, objectKind string, binding batchProfileBinding, sourceID string) (map[string]json.RawMessage, error) {
	if ctx == nil {
		ctx = context.Background()
	}
	if sourceID == "" {
		return adapter.batchReadDirect(ctx, namespace, objectKind, binding, []string{sourceID})
	}

	waiter := &batchReadCoalesceWaiter{
		id:   sourceID,
		done: make(chan struct{}),
	}
	key := batchReadCoalesceKey{namespace: namespace, objectKind: objectKind}
	var flushNow *batchReadCoalesceBatch

	c.mu.Lock()
	pending := c.pending[key]
	if pending == nil {
		pending = &batchReadCoalesceBatch{
			key:     key,
			ctx:     ctx,
			binding: binding,
			maxSize: binding.profile.BatchSpec.MaxBatchSize,
			waiters: make(map[string][]*batchReadCoalesceWaiter),
		}
		if pending.maxSize < 1 {
			pending.maxSize = 1
		}
		c.pending[key] = pending
		pending.timer = time.AfterFunc(c.window, func() {
			c.flush(adapter, key)
		})
	}
	if _, exists := pending.waiters[sourceID]; !exists {
		pending.ids = append(pending.ids, sourceID)
	}
	pending.waiters[sourceID] = append(pending.waiters[sourceID], waiter)
	pending.waiterCnt++
	if len(pending.ids) >= pending.maxSize {
		flushNow = pending
		delete(c.pending, key)
		if pending.timer != nil {
			pending.timer.Stop()
		}
	}
	c.mu.Unlock()

	if flushNow != nil {
		go c.execute(adapter, flushNow)
	}

	select {
	case <-waiter.done:
		if waiter.err != nil {
			return nil, waiter.err
		}
		if waiter.body == nil {
			return map[string]json.RawMessage{}, nil
		}
		return map[string]json.RawMessage{sourceID: waiter.body}, nil
	case <-ctx.Done():
		return nil, ctx.Err()
	}
}

func (c *batchReadCoalescer) flush(adapter *calloutBatchReadAdapter, key batchReadCoalesceKey) {
	c.mu.Lock()
	batch := c.pending[key]
	if batch != nil {
		delete(c.pending, key)
	}
	c.mu.Unlock()
	if batch == nil {
		return
	}
	c.execute(adapter, batch)
}

func (c *batchReadCoalescer) execute(adapter *calloutBatchReadAdapter, batch *batchReadCoalesceBatch) {
	started := time.Now()
	// This layer owns per-source confirmation for the combined provider read.
	// Disable the singleton helper's probe so an omitted ID is never confirmed
	// twice when the coalescer happens to flush only one waiter.
	details, err := adapter.batchReadDirectWithInactiveProbe(batch.ctx, batch.key.namespace, batch.key.objectKind, batch.binding, batch.ids, false)
	perSourceErrs := map[string]error(nil)
	fallbackSingleReads := 0
	fallbackErrorCount := 0
	var fallbackElapsed time.Duration
	if shouldFallbackCanonicalAliasCoalescedRead(batch.binding, err) {
		fallbackStarted := time.Now()
		fallbackDetails := make(map[string]json.RawMessage, len(batch.ids))
		perSourceErrs = make(map[string]error)
		for _, id := range batch.ids {
			singleDetails, singleErr := adapter.batchReadDirect(batch.ctx, batch.key.namespace, batch.key.objectKind, batch.binding, []string{id})
			fallbackSingleReads++
			if singleErr != nil {
				perSourceErrs[id] = singleErr
				fallbackErrorCount++
				continue
			}
			if body := singleDetails[id]; body != nil {
				fallbackDetails[id] = body
			}
		}
		details = fallbackDetails
		err = nil
		fallbackElapsed = time.Since(fallbackStarted)
	}

	// The public BatchRead call made by each waiter contains one source ID, but
	// the coalescer may combine those calls into one provider request. Preserve
	// the single-source inactivity contract by probing each omitted ID and
	// returning the terminal sentinel only to that ID's waiters. Canonical-alias
	// fallback already performed one fully probed direct read per source.
	inactiveProbeReads := 0
	inactiveConfirmed := 0
	inactiveProbeErrors := 0
	if err == nil && fallbackSingleReads == 0 && len(batch.binding.profile.BatchSpec.InactiveProbeQueryParams) > 0 {
		if perSourceErrs == nil {
			perSourceErrs = make(map[string]error)
		}
		for _, id := range batch.ids {
			if _, found := details[id]; found {
				continue
			}
			inactiveProbeReads++
			confirmed, probeErr := adapter.confirmSourceInactive(
				batch.ctx, batch.key.namespace, batch.key.objectKind, batch.binding, id,
			)
			if probeErr != nil {
				inactiveProbeErrors++
				perSourceErrs[id] = fmt.Errorf(
					"connector_inbound: inactive source confirmation probe for %s/%s source %q: %w",
					batch.key.namespace, batch.key.objectKind, id, probeErr,
				)
				continue
			}
			if confirmed {
				inactiveConfirmed++
				perSourceErrs[id] = fmt.Errorf(
					"%w for %s/%s source %q", ErrSourceRecordInactive,
					batch.key.namespace, batch.key.objectKind, id,
				)
			}
		}
	}

	elapsed := time.Since(started)
	zlog.Info().
		Str("event", "connector_inbound_batch_read_coalesced").
		Str("namespace", batch.key.namespace).
		Str("object_kind", batch.key.objectKind).
		Int("requested_count", len(batch.ids)).
		Int("waiter_count", batch.waiterCnt).
		Int("max_batch_size", batch.maxSize).
		Int("fallback_single_reads", fallbackSingleReads).
		Int("fallback_error_count", fallbackErrorCount).
		Int64("fallback_elapsed_ms", fallbackElapsed.Milliseconds()).
		Int("inactive_probe_reads", inactiveProbeReads).
		Int("inactive_confirmed_count", inactiveConfirmed).
		Int("inactive_probe_error_count", inactiveProbeErrors).
		Int64("elapsed_ms", elapsed.Milliseconds()).
		Bool("has_error", err != nil).
		Err(err).
		Msg("Connector inbound coalesced batch read")
	for id, waiters := range batch.waiters {
		var body json.RawMessage
		if err == nil {
			body = details[id]
		}
		for _, waiter := range waiters {
			waiter.body = body
			waiter.err = err
			if waiter.err == nil && perSourceErrs != nil {
				waiter.err = perSourceErrs[id]
			}
			close(waiter.done)
		}
	}
}

func shouldFallbackCanonicalAliasCoalescedRead(binding batchProfileBinding, err error) bool {
	if err == nil || binding.profile == nil || !binding.profile.BatchSpec.AllowCanonicalIDAliases {
		return false
	}
	var countErr *canonicalAliasDemuxCountError
	return errors.As(err, &countErr) ||
		strings.Contains(err.Error(), "canonical alias demux requires one response record per requested ID")
}

func batchSpecConfiguredForProfile(profile *connector_callout.ConnectorCalloutProfile) bool {
	if profile == nil {
		return false
	}
	if connector_callout.UsesLegacyRequestOrderBatchDemux(profile) {
		return connector_callout.ValidateBatchSpecForLegacyRequestOrderDemux(profile.BatchSpec) == nil
	}
	return connector_callout.ValidateBatchSpec(profile.BatchSpec) == nil
}

func appendLegacyRequestOrderBatchDetails(out map[string]json.RawMessage, sourceIDs []string, records []map[string]any) error {
	if len(records) > len(sourceIDs) {
		return fmt.Errorf("%w: response returned %d records for %d requested IDs", connector_callout.ErrInvalidResponse, len(records), len(sourceIDs))
	}
	for idx, record := range records {
		sourceID := sourceIDs[idx]
		if sourceID == "" {
			return fmt.Errorf("%w: request source ID at index %d is blank", connector_callout.ErrInvalidResponse, idx)
		}
		if _, exists := out[sourceID]; exists {
			return fmt.Errorf("%w: duplicate source ID %q in legacy request-order batch response", connector_callout.ErrInvalidResponse, sourceID)
		}
		recordBytes, err := json.Marshal(record)
		if err != nil {
			return fmt.Errorf("re-marshal legacy request-order batch response record %q: %w", sourceID, err)
		}
		out[sourceID] = json.RawMessage(recordBytes)
	}
	return nil
}

func joinRelativePath(objectPath, batchPath string) string {
	objectPath = strings.TrimSpace(objectPath)
	batchPath = strings.TrimSpace(batchPath)

	if objectPath == "" {
		return batchPath
	}
	if batchPath == "" {
		return objectPath
	}
	return strings.TrimRight(objectPath, "/") + "/" + strings.TrimLeft(batchPath, "/")
}

func extractBatchResponseRecords(body []byte, recordsPath string) ([]map[string]any, error) {
	var root any
	dec := json.NewDecoder(bytes.NewReader(body))
	dec.UseNumber()
	if err := dec.Decode(&root); err != nil {
		return nil, fmt.Errorf("%w: decode response body: %v", connector_callout.ErrInvalidResponse, err)
	}

	current := root
	if strings.TrimSpace(recordsPath) != "" {
		var err error
		current, err = extractDottedValue(current, recordsPath)
		if err != nil {
			return nil, err
		}
	}

	items, ok := current.([]any)
	if !ok {
		return nil, fmt.Errorf("%w: records path %q resolved to %T, want array", connector_callout.ErrInvalidResponse, recordsPath, current)
	}
	records := make([]map[string]any, 0, len(items))
	for i, item := range items {
		record, ok := item.(map[string]any)
		if !ok {
			return nil, fmt.Errorf("%w: records path %q item %d resolved to %T, want object", connector_callout.ErrInvalidResponse, recordsPath, i, item)
		}
		records = append(records, record)
	}
	return records, nil
}

func extractRecordSourceID(record map[string]any, idPath string) (string, error) {
	value, err := extractDottedValue(record, idPath)
	if err != nil {
		return "", err
	}
	switch typed := value.(type) {
	case string:
		if strings.TrimSpace(typed) == "" {
			return "", fmt.Errorf("%w: IDResponseField %q resolved to empty string", connector_callout.ErrInvalidResponse, idPath)
		}
		return typed, nil
	case json.Number:
		if typed.String() == "" {
			return "", fmt.Errorf("%w: IDResponseField %q resolved to empty number", connector_callout.ErrInvalidResponse, idPath)
		}
		return typed.String(), nil
	default:
		return "", fmt.Errorf("%w: IDResponseField %q resolved to unsupported type %T", connector_callout.ErrInvalidResponse, idPath, value)
	}
}

func extractDottedValue(root any, path string) (any, error) {
	current := root
	for i, segment := range strings.Split(path, ".") {
		obj, ok := current.(map[string]any)
		if !ok {
			return nil, fmt.Errorf("%w: path %q segment %d (%q) descends into non-object %T", connector_callout.ErrInvalidResponse, path, i+1, segment, current)
		}
		next, exists := obj[segment]
		if !exists {
			return nil, fmt.Errorf("%w: path %q segment %d (%q) missing", connector_callout.ErrInvalidResponse, path, i+1, segment)
		}
		current = next
	}
	if current == nil {
		return nil, fmt.Errorf("%w: path %q resolved to null", connector_callout.ErrInvalidResponse, path)
	}
	return current, nil
}