package automation

import (
	"context"
	"encoding/json"
	"fmt"
	"strings"
	"time"

	apperrors "github.com/nexuscrm/kernel/pkg/errors"
	"github.com/nexuscrm/shared/pkg/models"
)

// directoryPullTimeout bounds a single scheduled connector-directory pull.
// Scheduled ticks run on context.Background() and the outbound connector HTTP
// client has no client-level timeout, so without this a stalled remote endpoint
// would permanently consume scheduled-flow worker capacity. Mirrors the 30s
// webhook client timeout (infrastructure/http_client/webhook_client.go).
const directoryPullTimeout = 30 * time.Second

// ConnectorDirectoryPullRequest carries all parameters for a single directory
// pull. Using a struct avoids positional-arg confusion and lets callers add
// fields (e.g. MaxItems) without breaking existing call sites.
type ConnectorDirectoryPullRequest struct {
	// RequesterOwnerPack is the pack that owns the scheduled flow issuing this
	// pull.  Used to enforce same-pack authority: a flow may only pull profiles
	// from its own pack.
	RequesterOwnerPack string
	// ProfileOwnerPack is the pack that owns the callout profile.
	ProfileOwnerPack string
	ProfileName      string
	ObjectKey        string
	ItemsPath        string
	// MaxItems is a hard upper bound on accumulated items.  The adapter fails
	// closed as soon as the running total exceeds this value, without reading
	// further pages.  Zero means no adapter-level cap (the caller is expected
	// to enforce its own post-pull cap instead).
	MaxItems int
}

// ConnectorDirectoryReader pulls a connector directory (list endpoint) and
// returns its items as raw maps. The concrete implementation composes
// connector_callout (application layer); automation depends only on this
// interface.
type ConnectorDirectoryReader interface {
	// PullDirectory returns every item in the named directory. All-or-nothing:
	// transport/callout/pagination errors return a non-nil error (NOT a partial
	// slice); an empty-but-successful directory returns an empty slice + nil error.
	PullDirectory(ctx context.Context, req ConnectorDirectoryPullRequest) ([]map[string]interface{}, error)
}

// buildConnectorDirectoryInputs builds scheduled-flow input sets by pulling a
// connector directory via ConnectorDirectoryReader. One pulled item becomes one
// input set; the item's stable id (at ItemIDPath) is lifted to the top-level
// "Id" key so scheduledInputRecordID deduplicates correctly.
//
// External items carry no NexusCRM object type or schema.
// Start.Object / Start.Filters / Start.FilterLogic / Start.FilterFormula are
// rejected: entry-condition filtering cannot work against external items
// (no field metadata). Filter inside the flow body (a Decision element) instead.
// $Record is the raw external item, accessible via {!$Record.<field>} references.
func (e *FlowExecutor) buildConnectorDirectoryInputs(ctx context.Context, flowVersion *models.FlowVersion, _ *models.UserSession) ([]map[string]interface{}, error) {
	start := flowVersion.Metadata.Start
	if err := models.ValidateConnectorSource(start); err != nil {
		return nil, apperrors.NewInternalError(err.Error(), nil)
	}
	src := start.ConnectorSource

	if e.connectorDirectoryReader == nil {
		return nil, apperrors.NewInternalError("connector directory reader unavailable", nil)
	}

	// Only pack-owned scheduled flows may pull a connector directory.
	// DefinitionOwnerPack is populated by applyFlowDefinitionRuntimeFields
	// (scheduler_service.go) before buildConnectorDirectoryInputs is called.
	requesterPack := strings.TrimSpace(flowVersion.DefinitionOwnerPack)
	if requesterPack == "" {
		return nil, apperrors.NewInternalError("connector directory source requires a pack-owned scheduled flow", nil)
	}
	// Same pack-owned-STANDARD authority contract connector push enforces
	// (flowIdentityIsPackOwnedStandard): a tenant-customized copy of a shipped
	// flow keeps OwnerPack but flips DefinitionIsCustomized / Source=custom, and
	// must NOT inherit the pack's connector-profile/credential authority.
	if !flowVersionIsPackOwnedStandard(flowVersion) {
		return nil, apperrors.NewInternalError("connector directory source requires an uncustomized standard pack-owned flow version; customized or admin-authored versions cannot use pack-owned connector profile authority", nil)
	}

	pullCtx, cancel := context.WithTimeout(ctx, directoryPullTimeout)
	defer cancel()
	items, err := e.connectorDirectoryReader.PullDirectory(pullCtx, ConnectorDirectoryPullRequest{
		RequesterOwnerPack: requesterPack,
		ProfileOwnerPack:   src.OwnerPack,
		ProfileName:        src.ProfileName,
		ObjectKey:          src.ObjectKey,
		ItemsPath:          src.ItemsPath,
		MaxItems:           scheduledFlowMaxMatched,
	})
	if err != nil {
		return nil, err
	}

	// Backstop: fail closed when the directory is unexpectedly large — a
	// partial mirror would cause wrong dedup/fallback behavior downstream.
	// The adapter enforces the same cap mid-loop; this guards against adapters
	// that do not honour MaxItems.
	if len(items) > scheduledFlowMaxMatched {
		return nil, apperrors.NewInternalError(
			fmt.Sprintf("connector directory source: item count %d exceeds scheduledFlowMaxMatched cap %d", len(items), scheduledFlowMaxMatched),
			nil,
		)
	}

	inputSets := make([]map[string]interface{}, 0, len(items))
	seen := make(map[string]struct{}, len(items))
	for i, item := range items {
		// Extract the item's stable unique id at ItemIDPath (dotted path relative
		// to the item). Empty or missing id is a hard error — mirrors the local
		// path's "record without Id" fail-closed invariant.
		id, err := extractItemID(item, src.ItemIDPath)
		if err != nil {
			return nil, fmt.Errorf("connector directory source: item %d id extraction: %w", i, err)
		}
		if id == "" {
			return nil, apperrors.NewInternalError(
				fmt.Sprintf("connector directory source: item %d has empty id at path %q", i, src.ItemIDPath),
				nil,
			)
		}
		if _, dup := seen[id]; dup {
			return nil, apperrors.NewInternalError(
				fmt.Sprintf("connector directory source: duplicate item id %q (items %d) — ids must be unique within the directory", id, i),
				nil,
			)
		}
		seen[id] = struct{}{}

		inputSets = append(inputSets, map[string]interface{}{
			"Id":      id,
			"$Record": models.SObject(item),
		})
	}
	return inputSets, nil
}

// extractItemID walks a dotted path within an item map and returns the string
// representation of the leaf value. Returns ("", nil) when the path is missing
// or the value is nil/empty-string; returns an error on type mismatch or
// non-object mid-path segment.
func extractItemID(item map[string]interface{}, path string) (string, error) {
	if path == "" {
		return "", nil
	}
	var current interface{} = item
	parts := splitDotPath(path)
	for i, seg := range parts {
		m, ok := current.(map[string]interface{})
		if !ok {
			return "", fmt.Errorf("path segment %d (%q) descends into non-object %T", i+1, seg, current)
		}
		val, exists := m[seg]
		if !exists {
			return "", nil
		}
		current = val
	}
	if current == nil {
		return "", nil
	}
	switch v := current.(type) {
	case string:
		// Trim so the emitted top-level Id matches the scheduler's trimmed
		// reservation key (scheduledRecordIDValue uses TrimSpace). A
		// whitespace-only id thus becomes "" and is rejected by the caller's
		// empty-id hard error instead of silently collapsing to the global
		// reservation sentinel.
		return strings.TrimSpace(v), nil
	case json.Number:
		return v.String(), nil
	default:
		return "", fmt.Errorf("connector directory source: itemIdPath %q resolved to non-scalar %T; expected string or number", path, current)
	}
}

// splitDotPath splits a dotted path into segments. It is intentionally minimal:
// empty segments (consecutive dots) are passed through; callers that need strict
// validation should invoke connector_callout.ValidateDottedJSONPath first.
func splitDotPath(path string) []string {
	if path == "" {
		return nil
	}
	var parts []string
	start := 0
	for i := 0; i < len(path); i++ {
		if path[i] == '.' {
			parts = append(parts, path[start:i])
			start = i + 1
		}
	}
	parts = append(parts, path[start:])
	return parts
}