package serving

import (
	"context"
	"errors"
	"fmt"
	"time"

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

// ExecutionRuntimeClient is the complete neutral Runtime surface used by the
// conformance-only V2 reconciler.
type ExecutionRuntimeClient interface {
	AcceptCommand(context.Context, commandv1.Request) (commandv1.Response, error)
	GetApplicationResult(
		context.Context,
		applicationresultv1.GetResultRequest,
	) (applicationresultv1.GetResultResponse, error)
}

// ConformanceExecutionReconciler intentionally has no production composition
// in this phase. It proves only the dependency-free ready-to-completed path.
type ConformanceExecutionReconciler struct {
	repository *ExecutionRepository
	runtime    ExecutionRuntimeClient
}

func NewConformanceExecutionReconciler(
	repository *ExecutionRepository,
	runtime ExecutionRuntimeClient,
) (*ConformanceExecutionReconciler, error) {
	if repository == nil || repository.db == nil || runtime == nil {
		return nil, fmt.Errorf("%w: execution repository or Runtime client is unavailable", ErrDependencyMissing)
	}
	return &ConformanceExecutionReconciler{repository: repository, runtime: runtime}, nil
}

// Reconcile reads Runtime's immutable ApplicationResult before acquiring a new
// submission claim. An uncertain command can therefore complete locally while
// its original epoch is halting, without re-execution. Only a missing or pending
// result may proceed to the exact frozen FlowCommand v3 submission path.
func (reconciler *ConformanceExecutionReconciler) Reconcile(
	ctx context.Context,
	executionID string,
	holderID string,
	lease time.Duration,
) (Execution, error) {
	if reconciler == nil || reconciler.repository == nil || reconciler.runtime == nil {
		return Execution{}, fmt.Errorf("%w: execution reconciler is unavailable", ErrDependencyMissing)
	}
	if ctx == nil {
		return Execution{}, fmt.Errorf("%w: context is nil", ErrDependencyMissing)
	}
	if _, ok := ctx.Deadline(); !ok {
		return Execution{}, invalidExecutionCandidate("reconciliation context requires a deadline")
	}
	if err := ctx.Err(); err != nil {
		return Execution{}, err
	}

	stored, found, err := reconciler.repository.LoadExecution(ctx, executionID)
	if err != nil {
		return Execution{}, err
	}
	if !found {
		return Execution{}, fmt.Errorf("%w: %s", ErrExecutionNotFound, executionID)
	}
	if stored.state == executionStateCompleted {
		return stored.copy(), nil
	}
	if stored.state != executionStateReady {
		return Execution{}, fmt.Errorf("%w: state=%s", ErrExecutionNotClaimable, stored.state)
	}

	result, err := reconciler.readApplicationResult(ctx, stored)
	if err != nil {
		return Execution{}, err
	}
	if result != nil {
		return reconciler.repository.completeFromApplicationResult(ctx, stored, *result)
	}

	claim, err := reconciler.repository.ClaimConformance(ctx, executionID, holderID, lease)
	if err != nil {
		// A halt can win after the first result observation. Query once more so a
		// concurrently committed Runtime result still converges without a claim.
		if errors.Is(err, ErrExecutionAuthorityConflict) {
			result, resultErr := reconciler.readApplicationResult(ctx, stored)
			if resultErr != nil {
				return Execution{}, resultErr
			}
			if result != nil {
				return reconciler.repository.completeFromApplicationResult(ctx, stored, *result)
			}
		}
		return Execution{}, err
	}
	return reconciler.submitClaimedExecution(ctx, claim)
}

// submitClaimedExecution consumes the exact submission capability committed by
// ClaimConformance. A later halt rejects new claims but cannot retroactively
// revoke this attempt; halt completion must wait for its terminal result.
func (reconciler *ConformanceExecutionReconciler) submitClaimedExecution(
	ctx context.Context,
	claim ExecutionClaim,
) (Execution, error) {
	payload, err := flowcommandv3.Encode(flowcommandv3.Payload{
		ExecutionRef:        claim.execution.executionRef(),
		CanonicalInputs:     claim.execution.canonicalInputs,
		ExternalEffectsMode: flowcommandv3.SuppressExternalEffects,
	})
	if err != nil {
		return Execution{}, fmt.Errorf("encode exact Runtime FlowCommand v3: %w", err)
	}
	catalogRevision := claim.catalogRevision
	request := commandv1.Request{
		CommandID:            claim.execution.runtimeCommandID,
		IdempotencyKey:       claim.execution.runtimeCommandID,
		CommandKind:          flowcommandv3.CommandKind,
		PartitionKey:         stableExecutionIdentity("connector-v2-account", claim.execution.connectorAccountID),
		PayloadFormatVersion: flowcommandv3.PayloadFormatVersion,
		Payload:              payload,
		PayloadDigest:        commandv1.ComputePayloadDigest(payload),
		MinCatalogRevision:   &catalogRevision,
	}
	if err := commandv1.ValidateRequest(ctx, request); err != nil {
		return Execution{}, fmt.Errorf("validate exact Runtime command request: %w", err)
	}
	response, err := reconciler.runtime.AcceptCommand(ctx, request)
	if err != nil {
		return Execution{}, fmt.Errorf("accept exact Runtime command: %w", err)
	}
	if err := commandv1.ValidateResponse(request, response); err != nil {
		return Execution{}, fmt.Errorf("validate Runtime command acceptance: %w", err)
	}
	if response.Receipts.CatalogRevision < catalogRevision {
		return Execution{}, errors.New("Runtime command acceptance used a stale Catalog revision")
	}

	result, err := reconciler.readApplicationResult(ctx, claim.execution)
	if err != nil {
		return Execution{}, err
	}
	if result == nil {
		return Execution{}, fmt.Errorf("%w: command=%s", ErrExecutionResultPending, request.CommandID)
	}
	return reconciler.repository.completeFromApplicationResult(ctx, claim.execution, *result)
}

func (reconciler *ConformanceExecutionReconciler) readApplicationResult(
	ctx context.Context,
	execution Execution,
) (*applicationresultv1.ApplicationResult, error) {
	request := applicationresultv1.GetResultRequest{
		CommandID:          execution.runtimeCommandID,
		ExecutionRefDigest: execution.executionRefDigest,
	}
	if err := applicationresultv1.ValidateGetResultRequest(ctx, request); err != nil {
		return nil, fmt.Errorf("validate Runtime ApplicationResult request: %w", err)
	}
	response, err := reconciler.runtime.GetApplicationResult(ctx, request)
	if err != nil {
		var failure *commonv1.Failure
		if errors.As(err, &failure) && failure.Category == commonv1.FailureMissing {
			return nil, nil
		}
		return nil, fmt.Errorf("read Runtime ApplicationResult: %w", err)
	}
	if err := applicationresultv1.ValidateGetResultResponse(request, response); err != nil {
		return nil, fmt.Errorf("validate Runtime ApplicationResult response: %w", err)
	}
	if response.State == applicationresultv1.ResultPending {
		return nil, nil
	}
	result := *response.Result
	return &result, nil
}