package serving

import (
	"bytes"
	"crypto/sha256"
	"encoding/hex"
	"errors"
	"fmt"
	"time"

	applicationresultv1 "github.com/pingcap-inc/nexuscrm-contract/pkg/runtime/applicationresultv1"
	flowcommandv3 "github.com/pingcap-inc/nexuscrm-contract/pkg/runtime/flowcommandv3"
)

var (
	// ErrExecutionAuthorityConflict reports that an execution no longer names
	// the exact active gate, epoch, binding, and PackGeneration authority.
	ErrExecutionAuthorityConflict = errors.New("connector serving: execution authority conflict")
	// ErrExecutionIdentityConflict reports a durable identity collision whose
	// immutable execution bytes differ.
	ErrExecutionIdentityConflict = errors.New("connector serving: execution identity conflict")
	// ErrExecutionNotFound reports that the exact execution ID is absent.
	ErrExecutionNotFound = errors.New("connector serving: execution not found")
	// ErrExecutionNotClaimable reports a terminal execution or a live lease.
	ErrExecutionNotClaimable = errors.New("connector serving: execution not claimable")
	// ErrExecutionResultPending reports that Runtime has not committed a terminal
	// ApplicationResult. It is safe to retry only with the same execution.
	ErrExecutionResultPending = errors.New("connector serving: execution result pending")
	// ErrStoredExecutionInvalid reports a persisted row outside the closed V2
	// execution contract.
	ErrStoredExecutionInvalid = errors.New("connector serving: stored execution invalid")
	// ErrExecutionSchemaUnavailable identifies the Core-owned execution schema
	// within the package-wide schema-unavailable class.
	ErrExecutionSchemaUnavailable = errors.New("connector serving: execution schema unavailable")
)

const (
	executionProtocolVersion  = "nexuscrm.connector-v2-execution.v1"
	executionStateReady       = "ready"
	executionStateCompleted   = "completed"
	executionStateQuarantined = "quarantined"

	maxExecutionLease = 15 * time.Minute
)

// ConformanceExecutionInput names every caller-owned coordinate needed to
// mint one dependency-free Connector V2 execution. The constructor derives all
// Runtime digests and identities; callers cannot supply those derived facts.
// This type grants no production admission or claimant reachability.
type ConformanceExecutionInput struct {
	ExecutionID              string
	ServingScopeDigest       string
	ConnectorAccountID       string
	AccountServingGateID     string
	AccountServingFenceToken int64
	ServingEpochID           string
	ServingEpochDigest       string
	ServingEpochBindingID    string
	PackGenerationID         string
	DefinitionRef            string
	DefinitionVersion        string
	CanonicalInputs          []byte
}

// ConformanceExecutionCandidate is a detached, immutable admission candidate.
type ConformanceExecutionCandidate struct {
	execution Execution
}

// NewConformanceExecutionCandidate derives the byte-stable Runtime execution
// identity. The first conformance slice always suppresses external effects.
func NewConformanceExecutionCandidate(
	input ConformanceExecutionInput,
) (ConformanceExecutionCandidate, error) {
	execution := Execution{
		id:                       input.ExecutionID,
		executionProtocolVersion: executionProtocolVersion,
		servingScopeDigest:       input.ServingScopeDigest,
		connectorAccountID:       input.ConnectorAccountID,
		accountServingGateID:     input.AccountServingGateID,
		accountServingFenceToken: input.AccountServingFenceToken,
		servingEpochID:           input.ServingEpochID,
		servingEpochDigest:       input.ServingEpochDigest,
		servingEpochBindingID:    input.ServingEpochBindingID,
		packGenerationID:         input.PackGenerationID,
		definitionRef:            input.DefinitionRef,
		definitionVersion:        input.DefinitionVersion,
		canonicalInputs:          append([]byte(nil), input.CanonicalInputs...),
		externalEffectsMode:      string(flowcommandv3.SuppressExternalEffects),
		state:                    executionStateReady,
	}

	inputDigest, err := flowcommandv3.ComputeInputDigest(
		execution.canonicalInputs,
		flowcommandv3.SuppressExternalEffects,
	)
	if err != nil {
		return ConformanceExecutionCandidate{}, invalidExecutionCandidate(
			"canonical inputs are not an exact FlowCommand v3 value",
		)
	}
	execution.inputDigest = inputDigest
	ref := execution.executionRef()
	if err := applicationresultv1.ValidateExecutionRef(ref); err != nil {
		return ConformanceExecutionCandidate{}, invalidExecutionCandidate(
			"Runtime ExecutionRef is invalid: %v",
			err,
		)
	}
	execution.executionRefDigest, err = applicationresultv1.ExecutionRefDigest(ref)
	if err != nil {
		return ConformanceExecutionCandidate{}, invalidExecutionCandidate(
			"derive Runtime ExecutionRef digest: %v",
			err,
		)
	}
	execution.runtimeCommandID = stableExecutionIdentity(
		"connector-v2-execution",
		execution.id,
	)
	if err := execution.validateStored(); err != nil {
		return ConformanceExecutionCandidate{}, fmt.Errorf("%w: %v", ErrInvalidCandidate, err)
	}
	return ConformanceExecutionCandidate{execution: execution.copy()}, nil
}

// Execution is the Connector-owned durable execution identity and lifecycle.
// All fields remain private so callers cannot mutate a validated stored value.
type Execution struct {
	id                       string
	executionProtocolVersion string
	servingScopeDigest       string
	connectorAccountID       string
	accountServingGateID     string
	accountServingFenceToken int64
	servingEpochID           string
	servingEpochDigest       string
	servingEpochBindingID    string
	packGenerationID         string
	definitionRef            string
	definitionVersion        string
	canonicalInputs          []byte
	externalEffectsMode      string
	inputDigest              string
	executionRefDigest       string
	runtimeCommandID         string
	state                    string
	claimFenceToken          int64
	claimHolderID            *string
	claimLeaseUntil          *time.Time
	resultDigest             *string
	runtimeCommitReceipt     *string
	quarantineReasonCode     *string
}

func (execution Execution) ID() string                 { return execution.id }
func (execution Execution) State() string              { return execution.state }
func (execution Execution) RuntimeCommandID() string   { return execution.runtimeCommandID }
func (execution Execution) ExecutionRefDigest() string { return execution.executionRefDigest }
func (execution Execution) ClaimFenceToken() int64     { return execution.claimFenceToken }
func (execution Execution) ConnectorAccountID() string { return execution.connectorAccountID }
func (execution Execution) ServingEpochID() string     { return execution.servingEpochID }
func (execution Execution) ServingEpochDigest() string { return execution.servingEpochDigest }
func (execution Execution) PackGenerationID() string   { return execution.packGenerationID }
func (execution Execution) DefinitionRef() string      { return execution.definitionRef }
func (execution Execution) DefinitionVersion() string  { return execution.definitionVersion }
func (execution Execution) CanonicalInputs() []byte {
	return append([]byte(nil), execution.canonicalInputs...)
}

func (execution Execution) ResultCorrelation() (string, string, bool) {
	if execution.resultDigest == nil || execution.runtimeCommitReceipt == nil {
		return "", "", false
	}
	return *execution.resultDigest, *execution.runtimeCommitReceipt, true
}

func (execution Execution) executionRef() applicationresultv1.ExecutionRef {
	return applicationresultv1.ExecutionRef{
		ExecutionID:        execution.id,
		PackGenerationID:   execution.packGenerationID,
		ServingEpochDigest: execution.servingEpochDigest,
		DefinitionRef:      execution.definitionRef,
		DefinitionVersion:  execution.definitionVersion,
		InputDigest:        execution.inputDigest,
	}
}

func (execution Execution) copy() Execution {
	execution.canonicalInputs = append([]byte(nil), execution.canonicalInputs...)
	execution.claimHolderID = copyExecutionString(execution.claimHolderID)
	execution.claimLeaseUntil = copyExecutionTime(execution.claimLeaseUntil)
	execution.resultDigest = copyExecutionString(execution.resultDigest)
	execution.runtimeCommitReceipt = copyExecutionString(execution.runtimeCommitReceipt)
	execution.quarantineReasonCode = copyExecutionString(execution.quarantineReasonCode)
	return execution
}

func (execution Execution) equalImmutable(other Execution) bool {
	return execution.id == other.id &&
		execution.executionProtocolVersion == other.executionProtocolVersion &&
		execution.servingScopeDigest == other.servingScopeDigest &&
		execution.connectorAccountID == other.connectorAccountID &&
		execution.accountServingGateID == other.accountServingGateID &&
		execution.accountServingFenceToken == other.accountServingFenceToken &&
		execution.servingEpochID == other.servingEpochID &&
		execution.servingEpochDigest == other.servingEpochDigest &&
		execution.servingEpochBindingID == other.servingEpochBindingID &&
		execution.packGenerationID == other.packGenerationID &&
		execution.definitionRef == other.definitionRef &&
		execution.definitionVersion == other.definitionVersion &&
		bytes.Equal(execution.canonicalInputs, other.canonicalInputs) &&
		execution.externalEffectsMode == other.externalEffectsMode &&
		execution.inputDigest == other.inputDigest &&
		execution.executionRefDigest == other.executionRefDigest &&
		execution.runtimeCommandID == other.runtimeCommandID
}

func (execution Execution) validateStored() error {
	for _, field := range []struct {
		name  string
		value string
	}{
		{name: "execution ID", value: execution.id},
		{name: "connector account ID", value: execution.connectorAccountID},
		{name: "account serving gate ID", value: execution.accountServingGateID},
		{name: "serving epoch ID", value: execution.servingEpochID},
		{name: "serving epoch binding ID", value: execution.servingEpochBindingID},
		{name: "PackGeneration ID", value: execution.packGenerationID},
		{name: "definition ref", value: execution.definitionRef},
		{name: "definition version", value: execution.definitionVersion},
		{name: "Runtime command ID", value: execution.runtimeCommandID},
	} {
		if !validOpaqueIdentifierWithin(field.value, connectorDigestProfileV2MaxIdentifierBytes) {
			return storedExecutionInvalid("%s is not canonical", field.name)
		}
	}
	if execution.executionProtocolVersion != executionProtocolVersion {
		return storedExecutionInvalid("execution protocol is unsupported")
	}
	if !rawSHA256Pattern.MatchString(execution.servingScopeDigest) ||
		!prefixedSHA256Pattern.MatchString(execution.servingEpochDigest) ||
		!prefixedSHA256Pattern.MatchString(execution.inputDigest) ||
		!prefixedSHA256Pattern.MatchString(execution.executionRefDigest) {
		return storedExecutionInvalid("immutable digest shape is invalid")
	}
	if execution.accountServingFenceToken <= 0 {
		return storedExecutionInvalid("account-serving fence must be positive")
	}
	if execution.externalEffectsMode != string(flowcommandv3.SuppressExternalEffects) {
		return storedExecutionInvalid("external effects are not suppressed")
	}
	expectedInputDigest, err := flowcommandv3.ComputeInputDigest(
		execution.canonicalInputs,
		flowcommandv3.SuppressExternalEffects,
	)
	if err != nil || expectedInputDigest != execution.inputDigest {
		return storedExecutionInvalid("canonical inputs disagree with InputDigest")
	}
	ref := execution.executionRef()
	expectedRefDigest, err := applicationresultv1.ExecutionRefDigest(ref)
	if err != nil || expectedRefDigest != execution.executionRefDigest {
		return storedExecutionInvalid("ExecutionRef digest disagrees")
	}
	if execution.runtimeCommandID != stableExecutionIdentity("connector-v2-execution", execution.id) {
		return storedExecutionInvalid("Runtime command identity disagrees")
	}
	if execution.claimFenceToken < 0 ||
		(execution.claimHolderID == nil) != (execution.claimLeaseUntil == nil) {
		return storedExecutionInvalid("claim fence or holder/lease pair is invalid")
	}

	switch execution.state {
	case executionStateReady:
		if execution.resultDigest != nil || execution.runtimeCommitReceipt != nil ||
			execution.quarantineReasonCode != nil {
			return storedExecutionInvalid("ready execution retains terminal fields")
		}
	case executionStateCompleted:
		if execution.claimFenceToken == 0 || execution.claimHolderID != nil ||
			execution.claimLeaseUntil != nil || execution.quarantineReasonCode != nil ||
			execution.resultDigest == nil || execution.runtimeCommitReceipt == nil ||
			!prefixedSHA256Pattern.MatchString(*execution.resultDigest) ||
			!prefixedSHA256Pattern.MatchString(*execution.runtimeCommitReceipt) {
			return storedExecutionInvalid("completed execution shape is invalid")
		}
	case executionStateQuarantined:
		if execution.claimHolderID != nil || execution.claimLeaseUntil != nil ||
			execution.resultDigest != nil || execution.runtimeCommitReceipt != nil ||
			execution.quarantineReasonCode == nil ||
			!validOpaqueIdentifierWithin(*execution.quarantineReasonCode, 64) {
			return storedExecutionInvalid("quarantined execution shape is invalid")
		}
	default:
		return storedExecutionInvalid("state %q is outside the closed lifecycle", execution.state)
	}
	return nil
}

// ExecutionClaim is a detached capability for one exact holder and fence.
type ExecutionClaim struct {
	execution       Execution
	holderID        string
	fenceToken      int64
	leaseUntil      time.Time
	catalogRevision uint64
}

func (claim ExecutionClaim) Execution() Execution    { return claim.execution.copy() }
func (claim ExecutionClaim) HolderID() string        { return claim.holderID }
func (claim ExecutionClaim) FenceToken() int64       { return claim.fenceToken }
func (claim ExecutionClaim) LeaseUntil() time.Time   { return claim.leaseUntil }
func (claim ExecutionClaim) CatalogRevision() uint64 { return claim.catalogRevision }

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

func invalidExecutionCandidate(format string, args ...any) error {
	return fmt.Errorf("%w: execution: %s", ErrInvalidCandidate, fmt.Sprintf(format, args...))
}

func storedExecutionInvalid(format string, args ...any) error {
	return fmt.Errorf("%w: %s", ErrStoredExecutionInvalid, fmt.Sprintf(format, args...))
}

func copyExecutionString(value *string) *string {
	if value == nil {
		return nil
	}
	copy := *value
	return &copy
}

func copyExecutionTime(value *time.Time) *time.Time {
	if value == nil {
		return nil
	}
	copy := *value
	return &copy
}