package connector_inbound_routing

import (
	"encoding/json"
	"fmt"
	"io"
	"strings"
	"time"
)

const (
	InvocationModePerEvent                                       = "per_event"
	InvocationModeBatchCollection                                = "batch_collection"
	InvocationModeLatestBySource                                 = "latest_by_source"
	InvocationModeLatestBySourceFullRecord                       = "latest_by_source_full_record"
	InvocationModeLatestBySourceFullRecordPreservePayload        = "latest_by_source_full_record_preserve_payload"
	InvocationModeLatestBySourceFullRecordPreservePayloadSettled = "latest_by_source_full_record_preserve_payload_settled"

	QueueClassInteractive = "interactive"
	QueueClassDefault     = "default"
	QueueClassBulk        = "bulk"

	DefaultBatchInputName        = "events"
	DefaultBatchResultOutputName = "eventResults"
	DefaultRouteMaxBatchSize     = 100
	DefaultQueueClass            = QueueClassDefault
	DefaultSourceSettleDelay     = 5 * time.Second
)

// RouteExecutionSpec is the immutable execution contract snapshotted by a
// versioned connector inbound route row. L0 only interprets this generic shape;
// the pack-owned Flow decides what each event means.
type RouteExecutionSpec struct {
	FlowDeveloperName     string
	InvocationMode        string
	BatchInputName        string
	BatchResultOutputName string
	MaxBatchSize          int
	QueueClass            string
	FlowExecutionFilter   *PayloadValueFilter
}

// PayloadValueFilter is a generic event-payload predicate owned by route
// metadata. It intentionally knows only JSON paths and scalar values; connector
// packs decide which provider properties are worth full Flow execution.
type PayloadValueFilter struct {
	Path      string   `json:"path"`
	EqualsAny []string `json:"equalsAny"`
}

// NormalizeRouteExecutionSpec canonicalizes optional execution settings and
// rejects unknown invocation modes.
func NormalizeRouteExecutionSpec(spec RouteExecutionSpec) (RouteExecutionSpec, error) {
	spec.FlowDeveloperName = strings.TrimSpace(spec.FlowDeveloperName)
	spec.InvocationMode = strings.ToLower(strings.TrimSpace(spec.InvocationMode))
	spec.BatchInputName = strings.TrimSpace(spec.BatchInputName)
	spec.BatchResultOutputName = strings.TrimSpace(spec.BatchResultOutputName)
	spec.QueueClass = strings.ToLower(strings.TrimSpace(spec.QueueClass))
	if spec.InvocationMode == "" {
		spec.InvocationMode = InvocationModePerEvent
	}
	if spec.QueueClass == "" {
		spec.QueueClass = DefaultQueueClass
	}
	if spec.BatchInputName == "" {
		spec.BatchInputName = DefaultBatchInputName
	}
	if spec.BatchResultOutputName == "" {
		spec.BatchResultOutputName = DefaultBatchResultOutputName
	}
	if spec.MaxBatchSize <= 0 {
		spec.MaxBatchSize = DefaultRouteMaxBatchSize
	}
	switch spec.InvocationMode {
	case InvocationModePerEvent, InvocationModeBatchCollection, InvocationModeLatestBySource, InvocationModeLatestBySourceFullRecord, InvocationModeLatestBySourceFullRecordPreservePayload, InvocationModeLatestBySourceFullRecordPreservePayloadSettled:
	default:
		return RouteExecutionSpec{}, fmt.Errorf("unknown connector inbound invocationMode %q", spec.InvocationMode)
	}
	switch spec.QueueClass {
	case QueueClassInteractive, QueueClassDefault, QueueClassBulk:
	default:
		return RouteExecutionSpec{}, fmt.Errorf("unsupported connector inbound queueClass %q", spec.QueueClass)
	}
	filter, err := NormalizePayloadValueFilter(spec.FlowExecutionFilter)
	if err != nil {
		return RouteExecutionSpec{}, err
	}
	spec.FlowExecutionFilter = filter
	if (spec.InvocationMode == InvocationModeLatestBySourceFullRecord ||
		spec.InvocationMode == InvocationModeLatestBySourceFullRecordPreservePayload ||
		spec.InvocationMode == InvocationModeLatestBySourceFullRecordPreservePayloadSettled) &&
		spec.FlowExecutionFilter != nil {
		return RouteExecutionSpec{}, fmt.Errorf("connector inbound invocationMode %q cannot be combined with flowExecutionFilter", spec.InvocationMode)
	}
	return spec, nil
}

// SourceSettleDelay returns the durable not-before delay for invocation modes
// that deliberately collect a short same-source burst before Flow execution.
func SourceSettleDelay(invocationMode string) time.Duration {
	if strings.EqualFold(strings.TrimSpace(invocationMode), InvocationModeLatestBySourceFullRecordPreservePayloadSettled) {
		return DefaultSourceSettleDelay
	}
	return 0
}

func RouteExecutionSpecEqual(a, b RouteExecutionSpec) bool {
	return a.FlowDeveloperName == b.FlowDeveloperName &&
		a.InvocationMode == b.InvocationMode &&
		a.BatchInputName == b.BatchInputName &&
		a.BatchResultOutputName == b.BatchResultOutputName &&
		a.MaxBatchSize == b.MaxBatchSize &&
		a.QueueClass == b.QueueClass &&
		payloadValueFilterEqual(a.FlowExecutionFilter, b.FlowExecutionFilter)
}

func NormalizePayloadValueFilter(filter *PayloadValueFilter) (*PayloadValueFilter, error) {
	if filter == nil {
		return nil, nil
	}
	out := &PayloadValueFilter{
		Path:      strings.TrimSpace(filter.Path),
		EqualsAny: make([]string, 0, len(filter.EqualsAny)),
	}
	if out.Path == "" {
		return nil, fmt.Errorf("flowExecutionFilter.path is required")
	}
	if !strings.HasPrefix(out.Path, ".") || strings.Contains(out.Path, "..") || out.Path == "." {
		return nil, fmt.Errorf("flowExecutionFilter.path must be a simple dotted JSON path, got %q", filter.Path)
	}
	for _, segment := range strings.Split(strings.TrimPrefix(out.Path, "."), ".") {
		if strings.TrimSpace(segment) == "" {
			return nil, fmt.Errorf("flowExecutionFilter.path contains an empty segment: %q", filter.Path)
		}
	}
	for _, value := range filter.EqualsAny {
		value = strings.TrimSpace(value)
		if value != "" {
			out.EqualsAny = append(out.EqualsAny, value)
		}
	}
	if len(out.EqualsAny) == 0 {
		return nil, fmt.Errorf("flowExecutionFilter.equalsAny must contain at least one non-empty value")
	}
	return out, nil
}

func ParsePayloadValueFilterJSON(raw string) (*PayloadValueFilter, error) {
	raw = strings.TrimSpace(raw)
	if raw == "" || raw == "null" {
		return nil, nil
	}
	var filter PayloadValueFilter
	dec := json.NewDecoder(strings.NewReader(raw))
	dec.DisallowUnknownFields()
	if err := dec.Decode(&filter); err != nil {
		return nil, fmt.Errorf("parse flowExecutionFilter: %w", err)
	}
	if err := dec.Decode(&struct{}{}); err != io.EOF {
		return nil, fmt.Errorf("parse flowExecutionFilter: trailing data")
	}
	return NormalizePayloadValueFilter(&filter)
}

func PayloadValueFilterJSON(filter *PayloadValueFilter) (string, error) {
	filter, err := NormalizePayloadValueFilter(filter)
	if err != nil {
		return "", err
	}
	if filter == nil {
		return "", nil
	}
	data, err := json.Marshal(filter)
	if err != nil {
		return "", err
	}
	return string(data), nil
}

func (f *PayloadValueFilter) Matches(payload json.RawMessage) (bool, error) {
	filter, err := NormalizePayloadValueFilter(f)
	if err != nil {
		return false, err
	}
	if filter == nil {
		return true, nil
	}
	var doc any
	dec := json.NewDecoder(strings.NewReader(string(payload)))
	dec.UseNumber()
	if err := dec.Decode(&doc); err != nil {
		return false, fmt.Errorf("parse filter payload: %w", err)
	}
	if err := dec.Decode(&struct{}{}); err != io.EOF {
		return false, fmt.Errorf("parse filter payload: trailing data")
	}
	value, ok := valueAtJSONPath(doc, filter.Path)
	if !ok {
		// Fail closed for data retention: indeterminate payload shape executes the
		// Flow instead of being terminally skipped.
		return true, nil
	}
	actual, ok := scalarString(value)
	if !ok {
		// A filter can only skip after resolving a concrete scalar not in
		// equalsAny. Null/object/array values execute the Flow.
		return true, nil
	}
	for _, allowed := range filter.EqualsAny {
		if actual == allowed {
			return true, nil
		}
	}
	return false, nil
}

func payloadValueFilterEqual(a, b *PayloadValueFilter) bool {
	a, aerr := NormalizePayloadValueFilter(a)
	b, berr := NormalizePayloadValueFilter(b)
	if aerr != nil || berr != nil {
		return false
	}
	if a == nil || b == nil {
		return a == nil && b == nil
	}
	if a.Path != b.Path || len(a.EqualsAny) != len(b.EqualsAny) {
		return false
	}
	for i := range a.EqualsAny {
		if a.EqualsAny[i] != b.EqualsAny[i] {
			return false
		}
	}
	return true
}

func valueAtJSONPath(doc any, path string) (any, bool) {
	current := doc
	for _, segment := range strings.Split(strings.TrimPrefix(path, "."), ".") {
		obj, ok := current.(map[string]any)
		if !ok {
			return nil, false
		}
		current, ok = obj[segment]
		if !ok {
			return nil, false
		}
	}
	return current, true
}

func scalarString(value any) (string, bool) {
	switch v := value.(type) {
	case string:
		return v, true
	case json.Number:
		return v.String(), true
	case bool:
		if v {
			return "true", true
		}
		return "false", true
	default:
		return "", false
	}
}

func NormalizeQueueClass(queueClass string) (string, error) {
	queueClass = strings.ToLower(strings.TrimSpace(queueClass))
	if queueClass == "" {
		return DefaultQueueClass, nil
	}
	switch queueClass {
	case QueueClassInteractive, QueueClassDefault, QueueClassBulk:
		return queueClass, nil
	default:
		return "", fmt.Errorf("unsupported connector inbound queueClass %q", queueClass)
	}
}