package packs

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

	kernelconnector "github.com/nexuscrm/kernel/pkg/connector"
	connector "github.com/pingcap-inc/nexuscrm-contract/pkg/connector"
)

const componentKindConnectorPushRecoverySourceEligibility ComponentKind = "connector_push_recovery_source_eligibility"

const persistedConnectorPushContractV2FormatVersion = uint32(2)

// persistedConnectorPushContractV2Envelope is a read-only rollback bridge for
// generations written during the halted V2 cutover. Stable code continues to
// author and return the V1 contract; the unfinished match-before-create policy
// is deliberately opaque and cannot become executable through this reader.
type persistedConnectorPushContractV2Envelope struct {
	FormatVersion     uint32
	Contract          connector.ConnectorPushDurableMetadataContract
	MatchBeforeCreate json.RawMessage `json:",omitempty"`
}

// LoadPersistedConnectorPushContract selects one exact connector contract from
// the immutable Runtime declaration stored with a pack generation. It never
// resolves the pack source or consults live metadata.
func LoadPersistedConnectorPushContract(
	raw string,
	businessKey string,
) (connector.ConnectorPushDurableMetadataContract, error) {
	businessKey = strings.TrimSpace(businessKey)
	if businessKey == "" {
		return connector.ConnectorPushDurableMetadataContract{}, fmt.Errorf("connector push contract business key is required")
	}
	rows, err := persistedRuntimeDeclarationRows(raw)
	if err != nil {
		return connector.ConnectorPushDurableMetadataContract{}, err
	}
	contracts := make([]connector.ConnectorPushDurableMetadataContract, 0)
	selected := -1
	seenBusinessKeys := make(map[string]struct{})
	mappings := make([]connector.ConnectorFieldMappingInput, 0)
	for index := range rows {
		row := rows[index]
		switch row.Kind {
		case ComponentKindConnectorPushContract:
			if _, duplicate := seenBusinessKeys[row.BusinessKey]; duplicate {
				return connector.ConnectorPushDurableMetadataContract{}, fmt.Errorf("connector push contract business key %q is ambiguous", row.BusinessKey)
			}
			seenBusinessKeys[row.BusinessKey] = struct{}{}
			contract, err := decodePersistedConnectorPushContractProjection(row)
			if err != nil {
				return connector.ConnectorPushDurableMetadataContract{}, fmt.Errorf("connector push contract %q projection: %w", row.BusinessKey, err)
			}
			if row.BusinessKey == businessKey {
				selected = len(contracts)
			}
			contracts = append(contracts, contract)
		case ComponentKindConnectorFieldMapping:
			owner, err := persistedConnectorFieldMappingProjectionOwner(row)
			if err != nil {
				return connector.ConnectorPushDurableMetadataContract{}, fmt.Errorf("connector field mapping %q projection: %w", row.BusinessKey, err)
			}
			values, err := decodePersistedConnectorFieldMappings(row, owner)
			if err != nil {
				return connector.ConnectorPushDurableMetadataContract{}, fmt.Errorf("connector field mapping %q projection: %w", row.BusinessKey, err)
			}
			mappings = append(mappings, values...)
		}
	}
	if selected < 0 {
		return connector.ConnectorPushDurableMetadataContract{}, fmt.Errorf("connector push contract business key %q is absent from persisted generation", businessKey)
	}
	if err := kernelconnector.ValidateConnectorPushDurableMetadataContracts(contracts, mappings); err != nil {
		return connector.ConnectorPushDurableMetadataContract{}, fmt.Errorf("connector push contract immutable closure: %w", err)
	}
	return contracts[selected], nil
}

// LoadPersistedConnectorPushRecoverySourceEligibility selects the optional
// recovery sidecar from an immutable pack generation. Generations published
// before this sidecar remain readable and return nil.
func LoadPersistedConnectorPushRecoverySourceEligibility(
	raw string,
	businessKey string,
) (*connector.ConnectorPushRecoverySourceEligibilityContract, error) {
	businessKey = strings.TrimSpace(businessKey)
	if businessKey == "" {
		return nil, fmt.Errorf("connector push recovery source eligibility business key is required")
	}
	rows, err := persistedRuntimeDeclarationRows(raw)
	if err != nil {
		return nil, err
	}
	var selected *runtimeStartupShallowDeclaredRow
	for index := range rows {
		row := rows[index]
		if row.Kind != componentKindConnectorPushRecoverySourceEligibility || row.BusinessKey != businessKey {
			continue
		}
		if selected != nil {
			return nil, fmt.Errorf("connector push recovery source eligibility business key %q is ambiguous", businessKey)
		}
		copy := row
		selected = &copy
	}
	if selected == nil {
		return nil, nil
	}
	eligibility, err := decodePersistedConnectorPushRecoverySourceEligibilityProjection(*selected)
	if err != nil {
		return nil, fmt.Errorf("connector push recovery source eligibility %q projection: %w", businessKey, err)
	}
	return eligibility, nil
}

func normalizePersistedConnectorPushRecoverySourceEligibilityProjection(
	row runtimeStartupShallowDeclaredRow,
) (string, error) {
	eligibility, err := decodePersistedConnectorPushRecoverySourceEligibilityProjection(row)
	if err != nil {
		return "", err
	}
	return marshalRuntimeStartupShallowProjection(eligibility)
}

func decodePersistedConnectorPushRecoverySourceEligibilityProjection(
	row runtimeStartupShallowDeclaredRow,
) (*connector.ConnectorPushRecoverySourceEligibilityContract, error) {
	if strings.TrimSpace(row.Projection) == "" {
		return nil, fmt.Errorf("projection is required")
	}
	var eligibility connector.ConnectorPushRecoverySourceEligibilityContract
	if err := decodeStrictJSON([]byte(row.Projection), &eligibility); err != nil {
		return nil, err
	}
	sourcePayload, err := json.Marshal(struct {
		RecoverySourceEligibility connector.ConnectorPushRecoverySourceEligibilityContract `json:"recoverySourceEligibility"`
	}{RecoverySourceEligibility: eligibility})
	if err != nil {
		return nil, err
	}
	return kernelconnector.ParseConnectorPushRecoverySourceEligibilityFromPayload(sourcePayload)
}

func normalizePersistedConnectorPushContractProjection(
	row runtimeStartupShallowDeclaredRow,
) (string, error) {
	contract, err := decodePersistedConnectorPushContractProjection(row)
	if err != nil {
		return "", err
	}
	return marshalRuntimeStartupShallowProjection(contract)
}

func decodePersistedConnectorPushContractProjection(
	row runtimeStartupShallowDeclaredRow,
) (connector.ConnectorPushDurableMetadataContract, error) {
	if strings.TrimSpace(row.Projection) == "" {
		return connector.ConnectorPushDurableMetadataContract{}, fmt.Errorf("projection is required")
	}
	var contract connector.ConnectorPushDurableMetadataContract
	projection := []byte(row.Projection)
	if err := decodeStrictJSON(projection, &contract); err != nil {
		var envelope persistedConnectorPushContractV2Envelope
		if envelopeErr := decodeStrictJSON(projection, &envelope); envelopeErr != nil {
			return connector.ConnectorPushDurableMetadataContract{}, envelopeErr
		}
		if envelope.FormatVersion != persistedConnectorPushContractV2FormatVersion {
			return connector.ConnectorPushDurableMetadataContract{}, fmt.Errorf(
				"unsupported connector push contract format version %d",
				envelope.FormatVersion,
			)
		}
		contract = envelope.Contract
	}
	if err := validatePersistedConnectorPushContractIdentity(row, contract); err != nil {
		return connector.ConnectorPushDurableMetadataContract{}, err
	}
	return contract, nil
}

func validatePersistedConnectorPushContractIdentity(
	row runtimeStartupShallowDeclaredRow,
	contract connector.ConnectorPushDurableMetadataContract,
) error {
	expectedBusinessKey := JoinComponentBusinessKey(
		strings.TrimSpace(contract.ConnectorOwnerKey),
		strings.TrimSpace(contract.FlowDeveloperName),
		strings.TrimSpace(contract.FlowActionName),
		strings.TrimSpace(contract.SourceObject),
		strings.TrimSpace(contract.DestinationSlotID),
	)
	if expectedBusinessKey != strings.TrimSpace(row.BusinessKey) {
		return fmt.Errorf(
			"assembled durable identity %q does not match business key %q",
			expectedBusinessKey,
			strings.TrimSpace(row.BusinessKey),
		)
	}
	for _, field := range []struct{ name, value string }{
		{name: "connectorOwnerKey", value: contract.ConnectorOwnerKey},
		{name: "flowDeveloperName", value: contract.FlowDeveloperName},
		{name: "flowActionName", value: contract.FlowActionName},
		{name: "actionSlotId", value: contract.ActionSlotID},
		{name: "actionType", value: contract.ActionType},
		{name: "sourceObject", value: contract.SourceObject},
		{name: "destinationSlotId", value: contract.DestinationSlotID},
		{name: "targetObject", value: contract.TargetObject},
		{name: "profileName", value: contract.ProfileName},
		{name: "remoteIdField", value: contract.RemoteIDField},
	} {
		if strings.TrimSpace(field.value) == "" || strings.TrimSpace(field.value) != field.value {
			return fmt.Errorf("%s is missing or non-canonical", field.name)
		}
	}
	return nil
}