package connector_outboundintent

import (
	"bytes"
	"context"
	"errors"
	"strings"

	"github.com/pingcap-inc/nexuscrm-connectors/internal/sqltx"
	outboundintentv1 "github.com/pingcap-inc/nexuscrm-contract/pkg/runtime/outboundintentv1"
)

// writebackCompletion is minted only after the worker validates Runtime's
// exact idempotent acknowledgement. A restart reacquires it through Runtime;
// neither provider effects nor custom public stores can manufacture it.
type writebackCompletion struct {
	request  outboundintentv1.CommitResultRequest
	response outboundintentv1.CommitResultResponse
}

// recordNoEffectResult checkpoints affirmative terminal no-effect evidence on
// the same command that owns provider dispatch. This transition cannot erase a
// prior provider identity or convert an expired/uncertain claim into completion.
func (r *Repository) recordNoEffectResult(
	ctx context.Context, claim Claim, status int, code string, reason outboundintentv1.ResourceNoEffectReason,
) (Command, error) {
	if r == nil || r.db == nil || ctx == nil || validateClaim(claim) != nil || !safeCode(code) ||
		(status != 0 && (status < 100 || status > 599)) {
		return Command{}, ErrInvalidInput
	}
	available, err := r.typedResultsAvailable(ctx)
	if err != nil {
		return Command{}, err
	}
	if !available {
		return Command{}, ErrSchemaUnverified
	}
	canonical, digest, err := outboundintentv1.CanonicalResourceNoEffectResultV1(outboundintentv1.ResourceNoEffectResultV1{
		Version: 1, Outcome: outboundintentv1.ResourceOutcomeNoEffect, Reason: reason,
	})
	if err != nil {
		return Command{}, ErrInvalidInput
	}
	tx, cleanup, err := sqltx.BeginPessimisticTx(ctx, r.db)
	if err != nil {
		return Command{}, err
	}
	defer func() { _ = cleanup() }()
	rollback := func(cause error) (Command, error) { return Command{}, errors.Join(cause, tx.Rollback()) }
	locked, err := scanCommand(tx.QueryRowContext(ctx, commandSelect+` WHERE Id = ? LIMIT 1 FOR UPDATE`, claim.Command.CommandID))
	if err != nil {
		return rollback(err)
	}
	now := r.now().UTC()
	if locked.ClaimToken != claim.Token || locked.LeaseExpiresAt == nil || !locked.LeaseExpiresAt.After(now) ||
		!sameCommand(locked, claim.Command.Scope, claim.Command.Intent) {
		return rollback(ErrLeaseLost)
	}
	if locked.RemoteID != "" ||
		(reason == outboundintentv1.ResourceNoEffectRejectedBeforeDispatch &&
			(locked.State != StateClaimed || locked.DispatchStarted != nil || status != 0)) ||
		(reason == outboundintentv1.ResourceNoEffectDefinitiveNoEffect &&
			(locked.State != StateDispatching || locked.DispatchStarted == nil)) {
		return rollback(ErrIdentityConflict)
	}
	request := outboundintentv1.CommitResultRequest{
		IntentID: locked.Intent.IntentID, PayloadDigest: locked.Intent.PayloadDigest,
		CatalogRevision: locked.Scope.CatalogRevision, IntegrationCommandID: locked.CommandID,
		ResultFormatVersion: outboundintentv1.ResourceNoEffectResultFormatV1,
		ResultDigest:        digest, CanonicalResult: canonical,
	}
	if outboundintentv1.ValidateCommitResultRequestForIntent(locked.Intent, request) != nil {
		return rollback(ErrInvalidInput)
	}
	result, err := tx.ExecContext(ctx, `
UPDATE _System_IntegrationOutboundCommand
   SET State = ?, ClaimedBy = NULL, ClaimToken = NULL, ClaimedAt = NULL,
       LeaseExpiresAt = NULL, DispatchStartedAt = NULL, NextAttemptAt = NULL,
       FailureCode = ?, RetryCount = 0, ResponseStatus = NULLIF(?, 0),
       ResultFormatVersion = ?, ResultDigest = ?, CanonicalResult = ?, LastModifiedDate = ?
 WHERE Id = ? AND State = ? AND BINARY ClaimToken = BINARY ? AND LeaseExpiresAt > ?
   AND RemoteId IS NULL AND ResultFormatVersion IS NULL AND ResultDigest IS NULL AND CanonicalResult IS NULL`,
		StateWriteback, code, status, request.ResultFormatVersion, digest, canonical, now,
		locked.CommandID, locked.State, claim.Token, now)
	if err != nil {
		return rollback(err)
	}
	if affected, err := result.RowsAffected(); err != nil || affected != 1 {
		return rollback(ErrIdentityConflict)
	}
	if err := tx.Commit(); err != nil {
		return Command{}, err
	}
	return r.loadByID(ctx, locked.CommandID)
}

func validateWritebackCompletion(command Command, typed storedTypedTerminalResult, completion *writebackCompletion) error {
	if completion == nil {
		return ErrIdentityConflict
	}
	request := completion.request
	if request.IntegrationCommandID != command.CommandID || request.CatalogRevision != command.Scope.CatalogRevision ||
		request.RemoteID != command.RemoteID ||
		request.ResultFormatVersion != typed.formatVersion || request.ResultDigest != typed.digest ||
		!bytes.Equal(request.CanonicalResult, typed.canonical) ||
		outboundintentv1.ValidateCommitResultRequestForIntent(command.Intent, request) != nil ||
		outboundintentv1.ValidateCommitResultResponse(request, completion.response) != nil {
		return ErrIdentityConflict
	}
	return nil
}

func typedResultTerminalState(typed storedTypedTerminalResult) (state, failureCode string) {
	if typed.formatVersion == outboundintentv1.ResourceNoEffectResultFormatV1 {
		// The result has already been decoded and checked against the immutable
		// intent. Keep the provider rejection distinct from a successful effect.
		result, err := outboundintentv1.DecodeResourceNoEffectResultV1(typed.canonical)
		if err == nil {
			return StateRejected, "no_effect_" + strings.ToLower(string(result.Reason))
		}
	}
	return StateSucceeded, ""
}