package connector_workqueue

import (
	"database/sql"
	"errors"
	"fmt"
	"strings"
	"time"

	contractmodels "github.com/pingcap-inc/nexuscrm-contract/pkg/models"
)

const (
	WorkStatePending      = "pending"
	WorkStateProcessing   = "processing"
	WorkStateRetry        = "retry"
	WorkStateProcessed    = "processed"
	WorkStateSkipped      = "skipped"
	WorkStateDead         = "dead"
	WorkStateBlocked      = "blocked"
	WorkStateBlockedRetry = "blocked_retry"
	// WorkStateDependencyBlocked is a source-compatibility alias only. A durable
	// dependency wait is represented by lease-free processing plus non-null
	// dependency coordinates, never by a distinct persisted State value.
	WorkStateDependencyBlocked = WorkStateProcessing

	// NonTerminalWorkStateSQLList is the canonical static SQL literal list for
	// lifecycle/backlog queries that must keep workers alive until every durable
	// WorkItem reaches a terminal state. Runtime transition queries should still
	// bind individual states as parameters.
	NonTerminalWorkStateSQLList = "'" + WorkStatePending + "', '" + WorkStateRetry + "', '" + WorkStateBlocked + "', '" + WorkStateBlockedRetry + "', '" + WorkStateProcessing + "'"

	DefaultQueueClass = "default"
	// WorkTypeConnectorEvent is the first executable work type: run the
	// materialized connector event through the existing connector worker flow.
	WorkTypeConnectorEvent       = "connector_event"
	WorkTypeWorkflowActionApply  = "workflow_action_apply"
	SourceProjectionWorkIDPrefix = "csp:"

	// EventExecutionOwnerEvent keeps execution on the bounded legacy
	// _System_ConnectorEvent claimant. EventExecutionOwnerWorkItem marks the
	// atomic handoff to WorkItem row authority; the Event Status is frozen as
	// EventWorkItemCompatibilityFenceStatus so older binaries lose their
	// existing claim/finalizer CAS without needing to understand this column.
	EventExecutionOwnerEvent              = "event"
	EventExecutionOwnerWorkItem           = "work_item"
	EventWorkItemCompatibilityFenceStatus = "skipped"

	FailureClassInboundExecution             = "inbound_execution_failed"
	FailureSummaryInbound                    = "Inbound connector execution failed; inspect correlated logs and SyncState."
	FailureClassIdentityConvergenceExpired   = "identity_convergence_expired"
	FailureSummaryIdentityConvergenceExpired = "Identity convergence prerequisite did not become authoritative before the bounded wait expired."
	FailureClassIdentityConvergenceInvalid   = "identity_convergence_invalid"
	FailureSummaryIdentityConvergenceInvalid = "Identity convergence prerequisite state was invalid and was terminally isolated."
)

var (
	ErrInvalidWorkItem          = errors.New("connector work item is invalid")
	ErrWorkItemIdentityConflict = errors.New("connector work item immutable identity conflicts with the persisted row")
	ErrInvalidSchedulerLease    = errors.New("connector scheduler lease is invalid")
	ErrInvalidTerminalState     = errors.New("connector work item terminal state is invalid")
	ErrReplayBlocked            = errors.New("connector work item replay is blocked by an active projection generation")
)

// WorkItemIdentityConflictError reports the two durable identities that did
// not converge to one ordinary WorkItem. Mutable execution state is excluded:
// materialization owns identity, while the executor owns lifecycle state.
type WorkItemIdentityConflictError struct {
	RequestedWorkID string
	EventID         string
	WorkType        string
	ExistingWorkID  string
	Reason          string
}

func (e *WorkItemIdentityConflictError) Error() string {
	if e == nil {
		return ErrWorkItemIdentityConflict.Error()
	}
	return fmt.Sprintf(
		"%s: requested_work_id=%q event_id=%q work_type=%q existing_work_id=%q reason=%q",
		ErrWorkItemIdentityConflict.Error(),
		strings.TrimSpace(e.RequestedWorkID),
		strings.TrimSpace(e.EventID),
		strings.TrimSpace(e.WorkType),
		strings.TrimSpace(e.ExistingWorkID),
		strings.TrimSpace(e.Reason),
	)
}

func (e *WorkItemIdentityConflictError) Unwrap() error {
	return ErrWorkItemIdentityConflict
}

// WorkItem is the executable connector queue row. The immutable event fact
// remains identified by EventID; execution state, retry, lanes, and lease
// ownership live here.
type WorkItem struct {
	WorkID                string
	EventID               string
	InstalledPackID       string
	WorkType              string
	QueueClass            string
	State                 string
	RunAfter              time.Time
	Attempt               int
	LeaseOwner            string
	LeaseToken            string
	LeaseUntil            sql.NullTime
	SourceLaneKey         string
	TargetMutationLaneKey string
	LaneOrdinal           int64
	Generation            int64
	CreatedDate           time.Time
	LastModifiedDate      time.Time
}

type FailWorkItemInput struct {
	WorkID         string
	EventID        string
	LeaseToken     string
	FailureClass   string
	FailureSummary string
	RetryThreshold int
	RetryDelays    []time.Duration
	Now            time.Time
}

type FailWorkItemResult struct {
	State        string
	FailureCount int
	RunAfter     time.Time
}

// BlockWorkItemInput binds a live WorkItem lease to one neutral cross-feed
// identity prerequisite. Dependency selectors are internal executor state and
// must not be copied into sanitized failure summaries or logs.
type BlockWorkItemInput struct {
	WorkID     string
	EventID    string
	LeaseToken string
	Dependency contractmodels.FlowExecutionDependency
	Now        time.Time
}

// IdentityConvergenceWait is the restart-safe resolver input assembled from
// WorkItem executor state plus immutable Event provenance.
type IdentityConvergenceWait struct {
	WorkID               string
	InstalledPackID      string
	DependencyBlockedAt  time.Time
	DependencyProbeAfter *time.Time
	Namespace            string
	ObjectKind           string
	SourceID             string
	Dependency           contractmodels.FlowExecutionDependency
}

// IdentityConvergenceRemoteSignal is the committed producer fact that a
// finalized remote identity now exists for the same authority tuple used by
// RemoteLocatorActiveKey. It deliberately excludes the waiting Flow's local
// record candidate, Event source-object provenance, and dependency-key hash:
// producers do not own those facts, and every matching waiter must revalidate
// its complete persisted Event scope and authority after wake.
type IdentityConvergenceRemoteSignal struct {
	ConnectorOwnerKey  string
	ConnectorNamespace string
	DestinationSlotID  string
	ConnectorRemoteID  string
}

// IdentityConvergenceCursor is a stable keyset cursor over dependency waits.
// Persisted ordering prevents unresolved rows at the front of the queue from
// starving later waits across restart-recovery pages.
type IdentityConvergenceCursor struct {
	DependencyBlockedAt time.Time
	WorkID              string
}

type CreateWorkItemInput struct {
	WorkID                string
	EventID               string
	InstalledPackID       string
	WorkType              string
	QueueClass            string
	State                 string
	RunAfter              time.Time
	LaneOrdinal           int64
	SourceLaneKey         string
	TargetMutationLaneKey string
	Generation            int64
	Now                   time.Time
}

// ReplayWorkItemInput extends the stable CreateWorkItemInput contract with
// immutable replay lineage. Keeping replay lineage on a dedicated input avoids
// changing the positional shape consumed by pinned callers.
type ReplayWorkItemInput struct {
	WorkItem       CreateWorkItemInput
	ReplayOfWorkID string
}

// SchedulerLease is retained for pinned and custom callers that still use the
// former pack-wide coordination protocol during the compatibility window.
//
// Deprecated: in-tree WorkItem workers use row CAS and LeaseToken.
type SchedulerLease struct {
	InstalledPackID  string
	Generation       int64
	OwnerID          string
	FencingToken     string
	LeaseUntil       time.Time
	RenewedAt        time.Time
	CreatedDate      time.Time
	LastModifiedDate time.Time
}

// SchedulerLeaseInput is retained for source compatibility.
//
// Deprecated: in-tree WorkItem workers do not acquire or renew SchedulerLease.
type SchedulerLeaseInput struct {
	InstalledPackID string
	Generation      int64
	OwnerID         string
	FencingToken    string
	LeaseUntil      time.Time
	Now             time.Time
}

// SchedulerLeaseReleaseInput is retained for source compatibility.
//
// Deprecated: in-tree WorkItem workers do not release SchedulerLease.
type SchedulerLeaseReleaseInput struct {
	InstalledPackID string
	Generation      int64
	OwnerID         string
	FencingToken    string
	Now             time.Time
}

type ClaimWorkItemInput struct {
	InstalledPackID string
	Generation      int64
	OwnerID         string
	// FencingToken retains the published/custom claim contract. The in-tree
	// distributed path validates and forwards it without consulting
	// SchedulerLease; unmarked legacy calls still use it for compatibility
	// gating. WorkItem row CAS and LeaseToken are the active execution fences.
	FencingToken string
	LeaseToken   string
	LeaseUntil   time.Time
	// QueueClass is the legacy single-class filter. QueueClasses is the full
	// profile filter used by multi-class workers; when both are set the filters
	// are unioned after trimming and de-duplicating.
	QueueClass   string
	QueueClasses []string
	StaleOnly    bool
	Now          time.Time
}

type ClaimWorkItemByEventInput struct {
	EventID         string
	WorkType        string
	InstalledPackID string
	Generation      int64
	OwnerID         string
	// FencingToken retains the published/custom claim contract. The in-tree
	// distributed path validates and forwards it without consulting
	// SchedulerLease; unmarked legacy calls still use it for compatibility
	// gating.
	FencingToken string
	LeaseToken   string
	LeaseUntil   time.Time
	Now          time.Time
}

type ResolveClaimableGenerationInput struct {
	InstalledPackID string
	Generation      int64
	QueueClasses    []string
	StaleOnly       bool
	Now             time.Time
}

type LegacyEventMirrorReadinessInput struct {
	InstalledPackID string
	Generation      int64
}

func isTerminalWorkState(state string) bool {
	return state == WorkStateProcessed || state == WorkStateSkipped || state == WorkStateDead
}

func IsSourceProjectionWorkID(workID string) bool {
	return strings.HasPrefix(strings.TrimSpace(workID), SourceProjectionWorkIDPrefix)
}