package runtimehandoff

import (
	"context"
	"crypto/sha256"
	"encoding/hex"
	"encoding/json"
	"errors"
	"fmt"
	"strings"

	connector_inbound "github.com/pingcap-inc/nexuscrm-connectors/features/connector/inbound"
	catalogv1 "github.com/pingcap-inc/nexuscrm-contract/pkg/catalog/consumerv1"
	commandv1 "github.com/pingcap-inc/nexuscrm-contract/pkg/runtime/commandv1"
	commonv1 "github.com/pingcap-inc/nexuscrm-contract/pkg/runtime/commonv1"
	flowcommandv1 "github.com/pingcap-inc/nexuscrm-contract/pkg/runtime/flowcommandv1"
	flowcommandv2 "github.com/pingcap-inc/nexuscrm-contract/pkg/runtime/flowcommandv2"
	recordmutationv1 "github.com/pingcap-inc/nexuscrm-contract/pkg/runtime/recordmutationv1"
)

// PlatformClient is the complete neutral L0 surface needed by the Integration
// handoff. The production api-host RuntimePlatformClient satisfies it directly.
type PlatformClient interface {
	GetIntegrationBindingView(
		context.Context,
		catalogv1.GetRequest,
	) (catalogv1.IntegrationBindingView, error)
	AcceptCommand(
		context.Context,
		commandv1.Request,
	) (commandv1.Response, error)
}

type ReceiptStore interface {
	VerifySchema(context.Context) error
	LoadForScope(
		context.Context,
		connector_inbound.RuntimeCommandHandoffScope,
	) (PreparedCommand, bool, error)
	Prepare(
		context.Context,
		connector_inbound.RuntimeCommandHandoffScope,
		PreparedCommand,
	) (PreparedCommand, bool, error)
}

func (h *Handoff) VerifyReady(ctx context.Context) error {
	if h == nil || h.store == nil {
		return errors.New("connector Runtime command handoff is unavailable")
	}
	return h.store.VerifySchema(ctx)
}

type Config struct {
	OrgID                string
	AppInstallationID    string
	AuthorizedWorkloadID string
}

type Handoff struct {
	client PlatformClient
	store  ReceiptStore
	config Config
}

func NewHandoff(
	client PlatformClient,
	store ReceiptStore,
	config Config,
) (*Handoff, error) {
	config.OrgID = strings.TrimSpace(config.OrgID)
	config.AppInstallationID = strings.TrimSpace(config.AppInstallationID)
	config.AuthorizedWorkloadID = strings.TrimSpace(config.AuthorizedWorkloadID)
	if client == nil || store == nil || config.OrgID == "" ||
		config.AppInstallationID == "" || config.AuthorizedWorkloadID == "" {
		return nil, errors.New(
			"connector Runtime command handoff requires client, store, and exact consumer scope",
		)
	}
	return &Handoff{client: client, store: store, config: config}, nil
}

func (h *Handoff) ResumePreparedRuntimeCommand(
	ctx context.Context,
	scope connector_inbound.RuntimeCommandHandoffScope,
) (bool, error) {
	scope, err := connector_inbound.CanonicalRuntimeCommandHandoffScope(scope)
	if err != nil {
		return true, permanentFailure(err)
	}
	prepared, found, err := h.store.LoadForScope(ctx, scope)
	if err != nil {
		return true, classifyFailure(err)
	}
	if !found {
		return false, nil
	}
	return true, h.acceptPrepared(ctx, scope, prepared)
}

func (h *Handoff) AcceptRuntimeCommand(
	ctx context.Context,
	request connector_inbound.RuntimeCommandHandoffRequest,
) error {
	scope, err := connector_inbound.CanonicalRuntimeCommandHandoffScope(
		request.Scope,
	)
	if err != nil {
		return permanentFailure(err)
	}
	if prepared, found, err := h.store.LoadForScope(ctx, scope); err != nil {
		return classifyFailure(err)
	} else if found {
		return h.acceptPrepared(ctx, scope, prepared)
	}

	view, err := h.client.GetIntegrationBindingView(
		ctx,
		catalogv1.GetRequest{},
	)
	if err != nil {
		return classifyFailure(err)
	}
	if err := catalogv1.ValidateIntegrationBindingView(view); err != nil {
		return permanentFailure(
			fmt.Errorf("validate Integration binding view: %w", err),
		)
	}
	if view.OrgID != h.config.OrgID ||
		view.AppInstallationID != h.config.AppInstallationID ||
		view.AuthorizedWorkloadID != h.config.AuthorizedWorkloadID {
		return permanentFailure(
			errors.New("Integration binding view is outside configured consumer scope"),
		)
	}
	binding, err := selectBinding(view, scope, request)
	if err != nil {
		return permanentFailure(err)
	}
	canonicalInputs, err := canonicalInputs(request.Inputs)
	if err != nil {
		return permanentFailure(err)
	}
	payload, err := encodeRuntimeFlowCommand(binding, request, canonicalInputs)
	if err != nil {
		return permanentFailure(err)
	}
	revision := view.Lineage.CatalogRevision
	commandID := stableIdentity("integration-workitem", scope.WorkItemID)
	prepared, _, err := h.store.Prepare(ctx, scope, PreparedCommand{
		OrgID:                  view.OrgID,
		AppInstallationID:      view.AppInstallationID,
		AuthorizedWorkloadID:   view.AuthorizedWorkloadID,
		ScopeDigest:            consumerScopeDigest(view.OrgID, view.AppInstallationID, view.AuthorizedWorkloadID),
		MinimumCatalogRevision: revision,
		BindingViewDigest:      view.Lineage.ViewDigest,
		Request: commandv1.Request{
			CommandID:            commandID,
			IdempotencyKey:       commandID,
			CommandKind:          binding.CommandKind,
			PartitionKey:         stableOptionalIdentity("integration-source", scope.SourcePartition),
			PayloadFormatVersion: binding.PayloadFormatVersion,
			Payload:              payload,
			PayloadDigest:        commandv1.ComputePayloadDigest(payload),
			MinCatalogRevision:   &revision,
		},
	})
	if err != nil {
		return classifyFailure(err)
	}
	return h.acceptPrepared(ctx, scope, prepared)
}

func (h *Handoff) acceptPrepared(
	ctx context.Context,
	scope connector_inbound.RuntimeCommandHandoffScope,
	prepared PreparedCommand,
) error {
	if prepared.OrgID != h.config.OrgID ||
		prepared.AppInstallationID != h.config.AppInstallationID ||
		prepared.AuthorizedWorkloadID != h.config.AuthorizedWorkloadID ||
		prepared.ScopeDigest != consumerScopeDigest(h.config.OrgID, h.config.AppInstallationID, h.config.AuthorizedWorkloadID) ||
		prepared.Request.CommandID != stableIdentity("integration-workitem", scope.WorkItemID) ||
		prepared.Request.IdempotencyKey != prepared.Request.CommandID ||
		prepared.Request.PartitionKey != stableOptionalIdentity("integration-source", scope.SourcePartition) ||
		prepared.Request.CommandKind != flowcommandv1.CommandKind ||
		!supportedRuntimeFlowCommandVersion(prepared.Request.PayloadFormatVersion) ||
		prepared.Request.MinCatalogRevision == nil ||
		*prepared.Request.MinCatalogRevision != prepared.MinimumCatalogRevision {
		return permanentFailure(
			errors.New("stored Runtime command receipt is outside exact WorkItem consumer scope"),
		)
	}
	response, err := h.client.AcceptCommand(ctx, prepared.Request)
	if err != nil {
		return classifyFailure(err)
	}
	if err := commandv1.ValidateResponse(prepared.Request, response); err != nil {
		// The Runtime may already have durably accepted the stable CommandID even
		// when its response is malformed or lost in translation. Keep the
		// Integration WorkItem non-terminal and retry the same frozen bytes; a
		// permanent disposition here would create two conflicting outcome facts.
		return retriableFailure(
			fmt.Errorf("validate Runtime command acceptance: %w", err),
		)
	}
	if response.Receipts.CatalogRevision < prepared.MinimumCatalogRevision {
		// Acceptance has already crossed the owner boundary. Treat an invalid
		// revision receipt as an uncertain outcome, not as proof that Runtime did
		// not accept the command.
		return retriableFailure(
			errors.New("Runtime command acceptance used a stale Catalog revision"),
		)
	}
	return nil
}

func selectBinding(
	view catalogv1.IntegrationBindingView,
	scope connector_inbound.RuntimeCommandHandoffScope,
	request connector_inbound.RuntimeCommandHandoffRequest,
) (catalogv1.RuntimeCommandBinding, error) {
	if request.DefinitionRef == "" ||
		strings.TrimSpace(request.DefinitionRef) != request.DefinitionRef ||
		strings.TrimSpace(request.DefinitionVersion) != request.DefinitionVersion {
		return catalogv1.RuntimeCommandBinding{}, errors.New(
			"Runtime command definition identity is invalid",
		)
	}
	wantBusinessKey := scope.Namespace + "/" + scope.EventKind
	var selected *catalogv1.RuntimeCommandBinding
	for i := range view.Bindings {
		binding := view.Bindings[i]
		if binding.Kind != catalogv1.BindingConnectorInboundRoute ||
			binding.OwnerPack != scope.Namespace ||
			binding.PackInstallationID != scope.InstalledPackID ||
			binding.BusinessKey != wantBusinessKey ||
			binding.DefinitionVersion != scope.RouteVersion ||
			binding.RuntimeCommand == nil ||
			binding.RuntimeCommand.CommandKind != flowcommandv1.CommandKind ||
			!supportedRuntimeFlowCommandVersion(binding.RuntimeCommand.PayloadFormatVersion) ||
			binding.RuntimeCommand.DefinitionRef != request.DefinitionRef ||
			(request.DefinitionVersion != "" &&
				binding.RuntimeCommand.DefinitionVersion != request.DefinitionVersion) {
			continue
		}
		if selected != nil {
			return catalogv1.RuntimeCommandBinding{}, errors.New(
				"multiple Integration bindings match the exact Runtime command route",
			)
		}
		value := *binding.RuntimeCommand
		selected = &value
	}
	if selected == nil {
		return catalogv1.RuntimeCommandBinding{}, errors.New(
			"exact Integration Runtime command binding is unavailable",
		)
	}
	return *selected, nil
}

func encodeRuntimeFlowCommand(
	binding catalogv1.RuntimeCommandBinding,
	request connector_inbound.RuntimeCommandHandoffRequest,
	canonicalInputs []byte,
) ([]byte, error) {
	switch binding.PayloadFormatVersion {
	case flowcommandv1.PayloadFormatVersion:
		return flowcommandv1.Encode(flowcommandv1.Payload{
			DefinitionRef:     binding.DefinitionRef,
			DefinitionVersion: binding.DefinitionVersion,
			CanonicalInputs:   canonicalInputs,
		})
	case flowcommandv2.PayloadFormatVersion:
		return flowcommandv2.Encode(flowcommandv2.Payload{
			DefinitionRef:     binding.DefinitionRef,
			DefinitionVersion: binding.DefinitionVersion,
			CanonicalInputs:   canonicalInputs,
			DependencySource:  request.DependencySource,
		})
	default:
		return nil, errors.New("Integration binding selects an unsupported Runtime Flow command version")
	}
}

func supportedRuntimeFlowCommandVersion(version uint32) bool {
	return version == flowcommandv1.PayloadFormatVersion ||
		version == flowcommandv2.PayloadFormatVersion
}

func canonicalInputs(inputs map[string]any) ([]byte, error) {
	if inputs == nil {
		return nil, errors.New("Runtime command inputs are required")
	}
	raw, err := json.Marshal(inputs)
	if err != nil {
		return nil, errors.New("Runtime command inputs are not JSON")
	}
	canonical, err := recordmutationv1.CanonicalSourceJSON(string(raw))
	if err != nil {
		return nil, errors.New("Runtime command inputs are not canonical JSON")
	}
	return []byte(canonical), nil
}

func stableIdentity(domain, value string) string {
	sum := sha256.Sum256([]byte(domain + "\x00" + value))
	return domain + ":" + hex.EncodeToString(sum[:])
}

func consumerScopeDigest(orgID, appInstallationID, workloadID string) string {
	sum := sha256.Sum256([]byte(
		"nexuscrm.integration.runtime-handoff.scope.v1\x00" +
			orgID + "\x00" + appInstallationID + "\x00" + workloadID + "\x00" +
			commandv1.CapabilityName,
	))
	return hex.EncodeToString(sum[:])
}

func stableOptionalIdentity(domain, value string) string {
	if strings.TrimSpace(value) == "" {
		return ""
	}
	return stableIdentity(domain, value)
}

func classifyFailure(err error) error {
	if err == nil {
		return nil
	}
	if errors.Is(err, context.Canceled) || errors.Is(err, context.DeadlineExceeded) ||
		errors.Is(err, ErrLeaseLost) {
		return retriableFailure(err)
	}
	var storeUnavailable *storeUnavailableError
	if errors.As(err, &storeUnavailable) {
		return retriableFailure(err)
	}
	var failure *commonv1.Failure
	if errors.As(err, &failure) {
		switch failure.Category {
		case commonv1.FailureCanceled,
			commonv1.FailureDeadlineExceeded,
			commonv1.FailureResourceExhausted,
			commonv1.FailureStaleState,
			commonv1.FailureUnavailable,
			commonv1.FailureInternal:
			return retriableFailure(err)
		}
	}
	return permanentFailure(err)
}

func retriableFailure(err error) error {
	return &connector_inbound.RuntimeCommandHandoffFailure{
		Cause: err, Retriable: true,
	}
}

func permanentFailure(err error) error {
	return &connector_inbound.RuntimeCommandHandoffFailure{Cause: err}
}

var _ connector_inbound.RuntimeCommandHandoff = (*Handoff)(nil)