package runtimeworker

import (
	"context"
	"database/sql"
	"errors"
	"fmt"
	"strconv"
	"sync"
	"time"

	"github.com/nexuscrm/automation"
	database "github.com/nexuscrm/db"
	"github.com/nexuscrm/persistence"
	contractmodels "github.com/pingcap-inc/nexuscrm-contract/pkg/models"
	flowcommandsemantic "github.com/pingcap-inc/nexuscrm-contract/pkg/runtime/flowcommandv1"
	flowcommandsemanticv2 "github.com/pingcap-inc/nexuscrm-contract/pkg/runtime/flowcommandv2"
	flowcommandsemanticv3 "github.com/pingcap-inc/nexuscrm-contract/pkg/runtime/flowcommandv3"
)

const (
	RuntimeFlowCommandKind                   = flowcommandsemantic.CommandKind
	RuntimeFlowCommandPayloadFormatVersion   = flowcommandsemantic.PayloadFormatVersion
	RuntimeFlowCommandPayloadFormatVersionV2 = flowcommandsemanticv2.PayloadFormatVersion
	RuntimeFlowCommandPayloadFormatVersionV3 = flowcommandsemanticv3.PayloadFormatVersion
)

type PlatformCommandExecution struct {
	CommandID              string
	CommandKind            string
	PartitionKey           string
	SerializationKeyDigest string
	PayloadFormatVersion   uint32
	Payload                []byte
	CallerWorkloadID       string
	DefinitionAuthority    PlatformCommandDefinitionAuthority
}

// PlatformCommandDefinitionAuthority binds payload-selected executable
// identity to the same immutable command view that admitted the durable row.
type PlatformCommandDefinitionAuthority interface {
	CommandExecutionContext(
		kind string,
		version uint32,
		authorizedWorkloadID string,
		definitionRef string,
		definitionVersion string,
	) (string, bool)
}

// PlatformCommandExecutor is a release-owned app-blind mechanism. It may
// perform only bounded Runtime effects through the transaction carried by ctx;
// provider/external effects remain Integration-owned.
type PlatformCommandExecutor interface {
	ExecutePlatformCommand(context.Context, PlatformCommandExecution) ([]byte, error)
}

// platformCommandApplicationExecutor is the V3-only terminal seam. Its typed
// completion is consumed exclusively by the persistence ApplicationResult
// owner; ordinary errors remain nonterminal queue dispositions.
type platformCommandApplicationExecutor interface {
	ExecutePlatformApplicationCommand(
		context.Context,
		PlatformCommandExecution,
	) (persistence.RuntimePlatformCommandApplicationCompletion, error)
}

// platformCommandPublicationExecutor commits only a pending semantic
// postcondition. Publication and terminal verification are a separate phase;
// once this callback commits it must never be invoked again for that command.
type platformCommandPublicationExecutor interface {
	ExecutePlatformPublicationCommand(context.Context, PlatformCommandExecution) ([]byte, error)
}

// PlatformCommandBusinessFailure is a deterministic caller-safe rejection
// produced inside the effect transaction. Persistence may write Encoded only
// after PreparePublication proves that the attempted mutation rolled back.
type PlatformCommandBusinessFailure struct {
	Cause   error
	Encoded []byte
}

func (e *PlatformCommandBusinessFailure) Error() string {
	if e == nil || e.Cause == nil {
		return "platform command business operation was rejected"
	}
	return e.Cause.Error()
}

func (e *PlatformCommandBusinessFailure) Unwrap() error {
	if e == nil {
		return nil
	}
	return e.Cause
}

type PlatformCommandExecutionError struct {
	Cause     error
	Permanent bool
}

// PlatformCommandDependencyError is returned only after a V2 Flow publishes
// one fully bound durable dependency from inside the owned effect transaction.
// The processor may persist it only after the transaction owner proves that
// every attempted business effect rolled back.
type PlatformCommandDependencyError struct {
	Dependency contractmodels.FlowExecutionDependency
}

// PlatformCommandRecoveryError is emitted by a durable Flow command only after
// its atomic attempt has selected an exact Catalog-bound recovery candidate.
// The processor handles it only after the effect transaction rolls back; V3
// recovery remains nonterminal and replays the same immutable command.
type PlatformCommandRecoveryError struct {
	Request automation.ConnectorInboundRemoteIdentityRecoveryRequest
}

func (e *PlatformCommandRecoveryError) Error() string {
	return "platform command execution requires connector identity recovery"
}

func (e *PlatformCommandDependencyError) Error() string {
	return "platform command execution is blocked on a durable dependency"
}

func (e *PlatformCommandExecutionError) Error() string {
	if e == nil || e.Cause == nil {
		return "platform command execution failed"
	}
	return e.Cause.Error()
}

func (e *PlatformCommandExecutionError) Unwrap() error {
	if e == nil {
		return nil
	}
	return e.Cause
}

type platformCommandExecutorRegistry struct {
	byKindVersion map[string]platformCommandExecutorRegistration
}

type platformCommandExecutorRegistration struct {
	legacy      PlatformCommandExecutor
	application platformCommandApplicationExecutor
	publication platformCommandPublicationExecutor
}

func newPlatformCommandExecutorRegistry(
	values map[string]map[uint32]PlatformCommandExecutor,
	applicationValues map[string]map[uint32]platformCommandApplicationExecutor,
	publicationValueSets ...map[string]map[uint32]platformCommandPublicationExecutor,
) (*platformCommandExecutorRegistry, error) {
	registry := &platformCommandExecutorRegistry{
		byKindVersion: make(map[string]platformCommandExecutorRegistration),
	}
	for kind, versions := range values {
		if kind == "" || len(versions) == 0 {
			return nil, errors.New("runtime platform command registry: kind and versions are required")
		}
		for version, executor := range versions {
			if version == 0 || executor == nil {
				return nil, errors.New("runtime platform command registry: versioned executor is required")
			}
			if isRuntimeFlowApplicationCommand(kind, version) {
				return nil, errors.New("runtime platform command registry: V3 Flow requires an application executor")
			}
			key := platformCommandExecutorKey(kind, version)
			if _, duplicate := registry.byKindVersion[key]; duplicate {
				return nil, fmt.Errorf("runtime platform command registry: duplicate %s version %d", kind, version)
			}
			registry.byKindVersion[key] = platformCommandExecutorRegistration{legacy: executor}
		}
	}
	for kind, versions := range applicationValues {
		if kind == "" || len(versions) == 0 {
			return nil, errors.New("runtime platform command registry: application kind and versions are required")
		}
		for version, executor := range versions {
			if !isRuntimeFlowApplicationCommand(kind, version) || executor == nil {
				return nil, errors.New("runtime platform command registry: only the V3 Flow application executor is permitted")
			}
			key := platformCommandExecutorKey(kind, version)
			if _, duplicate := registry.byKindVersion[key]; duplicate {
				return nil, fmt.Errorf("runtime platform command registry: duplicate %s version %d", kind, version)
			}
			registry.byKindVersion[key] = platformCommandExecutorRegistration{application: executor}
		}
	}
	if len(publicationValueSets) > 1 {
		return nil, errors.New("runtime platform command registry: publication registry is ambiguous")
	}
	if len(publicationValueSets) == 1 {
		for kind, versions := range publicationValueSets[0] {
			if kind == "" || len(versions) == 0 {
				return nil, errors.New("runtime platform command registry: publication kind and versions are required")
			}
			for version, executor := range versions {
				if version == 0 || executor == nil {
					return nil, errors.New("runtime platform command registry: versioned publication executor is required")
				}
				key := platformCommandExecutorKey(kind, version)
				if _, duplicate := registry.byKindVersion[key]; duplicate {
					return nil, fmt.Errorf("runtime platform command registry: duplicate %s version %d", kind, version)
				}
				registry.byKindVersion[key] = platformCommandExecutorRegistration{publication: executor}
			}
		}
	}
	if len(registry.byKindVersion) == 0 {
		return nil, errors.New("runtime platform command registry: at least one neutral executor is required")
	}
	return registry, nil
}

func (r *platformCommandExecutorRegistry) resolve(
	kind string,
	version uint32,
) (platformCommandExecutorRegistration, bool) {
	if r == nil {
		return platformCommandExecutorRegistration{}, false
	}
	registration, ok := r.byKindVersion[platformCommandExecutorKey(kind, version)]
	return registration, ok
}

func isRuntimeFlowApplicationCommand(kind string, version uint32) bool {
	return kind == RuntimeFlowCommandKind &&
		version == RuntimeFlowCommandPayloadFormatVersionV3
}

func platformCommandExecutorKey(kind string, version uint32) string {
	return kind + "\x00" + strconv.FormatUint(uint64(version), 10)
}

type platformCommandExecutionStore interface {
	CheckReady(context.Context) error
	HasWork(context.Context) (bool, error)
	Claim(context.Context) (persistence.RuntimePlatformCommandLease, bool, error)
	Execute(
		context.Context,
		persistence.RuntimePlatformCommandLease,
		func(context.Context) ([]byte, error),
	) error
	ExecuteApplication(
		context.Context,
		persistence.RuntimePlatformCommandLease,
		func(context.Context) (persistence.RuntimePlatformCommandApplicationCompletion, error),
	) error
	PreparePublication(
		context.Context,
		persistence.RuntimePlatformCommandLease,
		func(context.Context) ([]byte, error),
	) error
	Block(
		context.Context,
		persistence.RuntimePlatformCommandLease,
		contractmodels.FlowExecutionDependency,
		time.Duration,
	) error
	Retry(context.Context, persistence.RuntimePlatformCommandLease, error, time.Duration) error
	Replay(context.Context, persistence.RuntimePlatformCommandLease, error) error
	Fail(context.Context, persistence.RuntimePlatformCommandLease, error) error
	FailWithResult(context.Context, persistence.RuntimePlatformCommandLease, []byte) error
	Poison(context.Context, persistence.RuntimePlatformCommandLease, error) error
}

type platformCommandPublicationStore interface {
	HasPublicationWork(context.Context) (bool, error)
	ClaimPublication(context.Context) (persistence.RuntimePlatformCommandPublicationLease, bool, error)
	RetryPublication(context.Context, persistence.RuntimePlatformCommandPublicationLease, error, time.Duration) error
	CompletePublication(context.Context, persistence.RuntimePlatformCommandPublicationLease, []byte, []byte) error
}

type platformCommandObservationSource interface {
	Observe(context.Context) (persistence.RuntimePlatformCommandExecutionState, error)
}

type platformCommandObservationFactory func() (platformCommandObservationSource, func() error, error)

const platformCommandDependencyRetryDelay = time.Minute

type platformCommandViewBinder interface {
	OpenCommand(
		context.Context,
		string,
		string,
		uint32,
	) (platformCommandExecutionAuthority, error)
}

type platformCommandExecutionAuthority interface {
	PlatformCommandDefinitionAuthority
	AllowsCommand(string, uint32) bool
	Bind(context.Context) (context.Context, error)
}

type platformCommandRecoveryOwner interface {
	Recover(
		context.Context,
		automation.ConnectorInboundRemoteIdentityRecoveryRequest,
	) (*automation.ConnectorInboundRemoteIdentityRecoveryResult, error)
}

type platformCommandPublicationCompletion struct {
	ResultPayload []byte
	Receipts      []byte
}

type platformCommandPublicationOwner interface {
	Publish(
		context.Context,
		persistence.RuntimePlatformCommandPublicationLease,
	) (platformCommandPublicationCompletion, error)
}

type platformCommandProcessor struct {
	store              platformCommandExecutionStore
	observationFactory platformCommandObservationFactory
	binder             platformCommandViewBinder
	executors          *platformCommandExecutorRegistry
	retryBackoff       time.Duration
	recovery           platformCommandRecoveryOwner
	publication        platformCommandPublicationOwner
	publicationStore   platformCommandPublicationStore

	observationMu      sync.Mutex
	observationSource  platformCommandObservationSource
	releaseObservation func() error
}

type platformCommandProcessorOption func(*platformCommandProcessor) error

func withPlatformCommandRecoveryOwner(owner platformCommandRecoveryOwner) platformCommandProcessorOption {
	return func(processor *platformCommandProcessor) error {
		if owner == nil {
			return errors.New("runtime platform command processor: recovery owner is unavailable")
		}
		processor.recovery = owner
		return nil
	}
}

func withPlatformCommandPublicationOwner(owner platformCommandPublicationOwner) platformCommandProcessorOption {
	return func(processor *platformCommandProcessor) error {
		if owner == nil {
			return errors.New("runtime platform command processor: publication owner is unavailable")
		}
		processor.publication = owner
		return nil
	}
}

type platformCommandExecutorFailure struct {
	cause error
}

func (e *platformCommandExecutorFailure) Error() string { return e.cause.Error() }
func (e *platformCommandExecutorFailure) Unwrap() error { return e.cause }

func newPlatformCommandProcessor(
	store platformCommandExecutionStore,
	observationFactory platformCommandObservationFactory,
	binder platformCommandViewBinder,
	executors *platformCommandExecutorRegistry,
	retryBackoff time.Duration,
	options ...platformCommandProcessorOption,
) (*platformCommandProcessor, error) {
	if store == nil || observationFactory == nil || binder == nil || executors == nil ||
		retryBackoff < 0 || retryBackoff > 5*time.Minute {
		return nil, errors.New("runtime platform command processor: configuration is incomplete")
	}
	processor := &platformCommandProcessor{
		store: store, observationFactory: observationFactory, binder: binder, executors: executors,
		retryBackoff: retryBackoff,
	}
	for _, option := range options {
		if option == nil {
			return nil, errors.New("runtime platform command processor: nil option")
		}
		if err := option(processor); err != nil {
			return nil, err
		}
	}
	if processor.publication != nil {
		publicationStore, ok := store.(platformCommandPublicationStore)
		if !ok {
			return nil, errors.New("runtime platform command processor: publication store is unavailable")
		}
		processor.publicationStore = publicationStore
	}
	return processor, nil
}

func (p *platformCommandProcessor) CheckReady(ctx context.Context) error {
	if p == nil || p.store == nil || p.binder == nil || p.executors == nil {
		return errors.New("runtime platform command processor: processor is unavailable")
	}
	return p.store.CheckReady(ctx)
}

func (p *platformCommandProcessor) Observe(ctx context.Context) (metadataWorkState, error) {
	if p == nil || p.observationFactory == nil {
		return metadataWorkState{}, errors.New("runtime platform command processor: observation source is unavailable")
	}
	p.observationMu.Lock()
	defer p.observationMu.Unlock()
	if p.observationSource == nil {
		source, release, err := p.observationFactory()
		if err != nil {
			return metadataWorkState{}, fmt.Errorf("runtime platform command processor: open observation source: %w", err)
		}
		if source == nil || release == nil {
			if release != nil {
				_ = release()
			}
			return metadataWorkState{}, errors.New("runtime platform command processor: observation source is incomplete")
		}
		p.observationSource = source
		p.releaseObservation = release
	}
	state, err := p.observationSource.Observe(ctx)
	if err != nil {
		return metadataWorkState{}, err
	}
	return metadataWorkState{
		queueDepth: state.QueueDepth, poison: state.PoisonDepth, oldestAge: state.OldestAge,
	}, nil
}

func (p *platformCommandProcessor) closeObservation() error {
	if p == nil {
		return nil
	}
	p.observationMu.Lock()
	defer p.observationMu.Unlock()
	release := p.releaseObservation
	p.observationSource = nil
	p.releaseObservation = nil
	if release == nil {
		return nil
	}
	return release()
}

func newPlatformCommandObservationFactory(
	openIndependentDB func(database.PoolOptions) (*sql.DB, error),
	expected persistence.RuntimeWorkOwnership,
	scope persistence.RuntimeWorkScope,
	trackers ...*workerAuxiliaryPoolTracker,
) platformCommandObservationFactory {
	return func() (platformCommandObservationSource, func() error, error) {
		if openIndependentDB == nil {
			return nil, nil, errors.New("independent database opener is unavailable")
		}
		observationDB, release, err := openWorkerObservationPool(
			openIndependentDB, optionalWorkerAuxiliaryPoolTracker(trackers),
		)
		if err != nil {
			return nil, nil, err
		}
		store, err := persistence.NewRuntimePlatformCommandObservationStore(observationDB, expected, scope)
		if err != nil {
			_ = release()
			return nil, nil, err
		}
		return store, release, nil
	}
}

// HasWork uses only the bounded primary-store hint. Hint errors fail open in
// metadataWorker, and Claim remains the fenced business authority.
func (p *platformCommandProcessor) HasWork(ctx context.Context) (bool, error) {
	if p == nil || p.store == nil {
		return false, errors.New("runtime platform command processor: processor is unavailable")
	}
	if p.publicationStore != nil {
		publication, err := p.publicationStore.HasPublicationWork(ctx)
		if err != nil || publication {
			return publication, err
		}
	}
	return p.store.HasWork(ctx)
}

func (*platformCommandProcessor) Maintain(context.Context, int) error { return nil }

func (p *platformCommandProcessor) ProcessOne(
	ctx context.Context,
	_ int,
) (int, bool, error) {
	if p.publication != nil {
		lease, found, err := p.publicationStore.ClaimPublication(ctx)
		if err != nil {
			return 0, false, err
		}
		if found {
			return p.processPublication(ctx, lease)
		}
	}
	lease, found, err := p.store.Claim(ctx)
	if err != nil || !found {
		return 0, found, err
	}
	authority, err := p.binder.OpenCommand(
		ctx,
		lease.ImmutableBinding,
		lease.Row.CommandKind,
		lease.Row.PayloadFormatVersion,
	)
	if err != nil {
		var dispositionErr error
		if isPermanentMetadataBindingError(err) {
			dispositionErr = p.store.Poison(ctx, lease, err)
		} else {
			dispositionErr = p.store.Retry(ctx, lease, err, p.retryBackoff)
		}
		if dispositionErr != nil {
			return 0, true, dispositionErr
		}
		if isPermanentMetadataBindingError(err) {
			return 1, true, nil
		}
		return 0, true, nil
	}
	if !authority.AllowsCommand(lease.Row.CommandKind, lease.Row.PayloadFormatVersion) {
		err := errors.New("runtime platform command: immutable view does not authorize accepted kind/version")
		if dispositionErr := p.store.Poison(ctx, lease, err); dispositionErr != nil {
			return 0, true, dispositionErr
		}
		return 1, true, nil
	}
	registration, ok := p.executors.resolve(
		lease.Row.CommandKind,
		lease.Row.PayloadFormatVersion,
	)
	if !ok {
		err := errors.New("runtime platform command: accepted kind/version has no release-owned neutral executor")
		if dispositionErr := p.store.Poison(ctx, lease, err); dispositionErr != nil {
			return 0, true, dispositionErr
		}
		return 1, true, nil
	}
	application := registration.application != nil
	publication := registration.publication != nil
	if application {
		err = p.store.ExecuteApplication(
			ctx,
			lease,
			func(txCtx context.Context) (persistence.RuntimePlatformCommandApplicationCompletion, error) {
				txCtx, bindErr := authority.Bind(txCtx)
				if bindErr != nil {
					return persistence.RuntimePlatformCommandApplicationCompletion{},
						&platformCommandExecutorFailure{cause: bindErr}
				}
				completion, executeErr := registration.application.ExecutePlatformApplicationCommand(
					txCtx,
					platformCommandExecutionFromLease(lease, authority),
				)
				if executeErr != nil {
					return persistence.RuntimePlatformCommandApplicationCompletion{},
						&platformCommandExecutorFailure{cause: executeErr}
				}
				return completion, nil
			},
		)
	} else if publication {
		err = p.store.PreparePublication(ctx, lease, func(txCtx context.Context) ([]byte, error) {
			txCtx, bindErr := authority.Bind(txCtx)
			if bindErr != nil {
				return nil, &platformCommandExecutorFailure{cause: bindErr}
			}
			result, executeErr := registration.publication.ExecutePlatformPublicationCommand(
				txCtx,
				platformCommandExecutionFromLease(lease, authority),
			)
			if executeErr != nil {
				return nil, &platformCommandExecutorFailure{cause: executeErr}
			}
			return result, nil
		})
	} else {
		err = p.store.Execute(ctx, lease, func(txCtx context.Context) ([]byte, error) {
			txCtx, bindErr := authority.Bind(txCtx)
			if bindErr != nil {
				return nil, bindErr
			}
			result, executeErr := registration.legacy.ExecutePlatformCommand(
				txCtx,
				platformCommandExecutionFromLease(lease, authority),
			)
			if executeErr != nil {
				return nil, &platformCommandExecutorFailure{cause: executeErr}
			}
			return result, nil
		})
	}
	if err == nil {
		return 1, true, nil
	}
	if ctx.Err() != nil ||
		errors.Is(err, persistence.ErrRuntimeWorkOwnershipLost) ||
		errors.Is(err, persistence.ErrRuntimePlatformCommandLeaseLost) {
		// Cancellation/ownership loss and an unknown commit outcome leave the
		// lease for state-based restart reconciliation. Do not author a second
		// disposition from an uncertain caller.
		return 0, true, err
	}
	if errors.Is(err, persistence.ErrRuntimePlatformCommandResultInvalid) {
		// The effect transaction definitely rolled back because its terminal
		// result could not satisfy the store contract. This is a deterministic
		// command failure, not an unknown commit outcome and not retryable work.
		// A V3 FAILED row must carry its typed ApplicationResult, so an invalid
		// completion is quarantined rather than sent through the legacy Fail path.
		disposition := p.store.Fail
		if application || publication {
			disposition = p.store.Poison
		}
		if dispositionErr := disposition(ctx, lease, err); dispositionErr != nil {
			return 0, true, dispositionErr
		}
		return 1, true, nil
	}
	var executorFailure *platformCommandExecutorFailure
	if !errors.As(err, &executorFailure) {
		// TransactionManager may be reporting an unknown commit outcome. Since
		// the business effect and terminal row share that commit, only the
		// durable state after lease expiry may decide whether replay is safe.
		return 0, true, err
	}
	var businessFailure *PlatformCommandBusinessFailure
	if errors.As(executorFailure.cause, &businessFailure) {
		if !publication || len(businessFailure.Encoded) == 0 {
			if dispositionErr := p.store.Poison(ctx, lease, errors.New("runtime platform command: typed business failure belongs only to a publication executor")); dispositionErr != nil {
				return 0, true, dispositionErr
			}
			return 1, true, nil
		}
		if dispositionErr := p.store.FailWithResult(ctx, lease, businessFailure.Encoded); dispositionErr != nil {
			return 0, true, dispositionErr
		}
		return 1, true, nil
	}
	var dependencyError *PlatformCommandDependencyError
	if errors.As(executorFailure.cause, &dependencyError) {
		if application || publication {
			dispositionErr := p.store.Poison(
				ctx,
				lease,
				errors.New("runtime platform command: V3 Flow has no Runtime dependency lifecycle"),
			)
			if dispositionErr != nil {
				return 0, true, dispositionErr
			}
			return 1, true, nil
		}
		dispositionErr := p.store.Block(
			ctx, lease, dependencyError.Dependency, platformCommandDependencyRetryDelay,
		)
		if errors.Is(dispositionErr, persistence.ErrRuntimePlatformCommandDependencyChanged) {
			if failErr := p.store.Fail(ctx, lease, dispositionErr); failErr != nil {
				return 0, true, failErr
			}
			return 1, true, nil
		}
		if dispositionErr != nil {
			return 0, true, dispositionErr
		}
		return 0, true, nil
	}
	var recoveryError *PlatformCommandRecoveryError
	if errors.As(executorFailure.cause, &recoveryError) {
		if p.recovery == nil {
			disposition := p.store.Fail
			if application || publication {
				disposition = p.store.Poison
			}
			if dispositionErr := disposition(ctx, lease, errors.New("runtime platform command: recovery owner is not configured")); dispositionErr != nil {
				return 0, true, dispositionErr
			}
			return 1, true, nil
		}
		if lease.AttemptsRemaining == 0 {
			replayErr := errors.New("runtime platform command: connector identity recovery requires one remaining replay attempt")
			if dispositionErr := p.store.Retry(ctx, lease, replayErr, 0); dispositionErr != nil {
				return 0, true, dispositionErr
			}
			return 0, true, nil
		}
		recoveryCtx, bindErr := authority.Bind(ctx)
		if bindErr != nil {
			if dispositionErr := p.store.Retry(ctx, lease, bindErr, p.retryBackoff); dispositionErr != nil {
				return 0, true, dispositionErr
			}
			return 0, true, nil
		}
		recoveryResult, recoverErr := p.recovery.Recover(recoveryCtx, recoveryError.Request)
		if contextErr := recoveryCtx.Err(); contextErr != nil {
			return 0, true, contextErr
		}
		if recoverErr != nil {
			if dispositionErr := p.store.Retry(ctx, lease, recoverErr, p.retryBackoff); dispositionErr != nil {
				return 0, true, dispositionErr
			}
			return 0, true, nil
		}
		replayReason := errors.New("connector identity recovery candidate disappeared; replay exact command")
		if recoveryResult != nil {
			replayReason = errors.New("connector identity recovery transition completed; replay exact command")
		}
		if dispositionErr := p.store.Replay(ctx, lease, replayReason); dispositionErr != nil {
			return 0, true, dispositionErr
		}
		return 0, true, nil
	}
	var executionError *PlatformCommandExecutionError
	if errors.As(executorFailure.cause, &executionError) && executionError.Permanent {
		disposition := p.store.Fail
		if application || publication {
			disposition = p.store.Poison
		}
		if dispositionErr := disposition(ctx, lease, err); dispositionErr != nil {
			return 0, true, dispositionErr
		}
		return 1, true, nil
	}
	if dispositionErr := p.store.Retry(ctx, lease, err, p.retryBackoff); dispositionErr != nil {
		return 0, true, dispositionErr
	}
	return 0, true, nil
}

func (p *platformCommandProcessor) processPublication(
	ctx context.Context,
	lease persistence.RuntimePlatformCommandPublicationLease,
) (int, bool, error) {
	completion, err := p.publication.Publish(ctx, lease)
	if err == nil {
		err = p.publicationStore.CompletePublication(
			ctx, lease, completion.ResultPayload, completion.Receipts,
		)
		if err == nil {
			return 1, true, nil
		}
	}
	if ctx.Err() != nil || errors.Is(err, persistence.ErrRuntimeWorkOwnershipLost) ||
		errors.Is(err, persistence.ErrRuntimePlatformCommandLeaseLost) ||
		errors.Is(err, persistence.ErrRuntimePlatformCommandOutcomeUncertain) {
		// The Catalog operation and terminal acknowledgement are both recovered
		// from durable identity/state. Do not author a competing disposition when
		// either external publication or the completion commit may have succeeded.
		return 0, true, err
	}
	if retryErr := p.publicationStore.RetryPublication(ctx, lease, err, p.retryBackoff); retryErr != nil {
		return 0, true, retryErr
	}
	return 0, true, nil
}

func platformCommandExecutionFromLease(
	lease persistence.RuntimePlatformCommandLease,
	authority PlatformCommandDefinitionAuthority,
) PlatformCommandExecution {
	return PlatformCommandExecution{
		CommandID: lease.Row.CommandID, CommandKind: lease.Row.CommandKind,
		PartitionKey: lease.Row.PartitionKey, PayloadFormatVersion: lease.Row.PayloadFormatVersion,
		SerializationKeyDigest: lease.Row.SerializationKeyDigest,
		Payload:                append([]byte(nil), lease.Row.Payload...), CallerWorkloadID: lease.CallerWorkloadID,
		DefinitionAuthority: authority,
	}
}

var _ metadataWorkProcessor = (*platformCommandProcessor)(nil)
var _ metadataWorkHinter = (*platformCommandProcessor)(nil)
var _ metadataObservationCloser = (*platformCommandProcessor)(nil)