package connector_inbound

import (
	"context"
	"database/sql"
	"errors"
	"fmt"
	"reflect"
	"strings"
	"sync"
	"time"

	connector_events "github.com/pingcap-inc/nexuscrm-connectors/features/connector/events"
	"github.com/pingcap-inc/nexuscrm-connectors/features/connector/events/claimpolicy"
	connector_workqueue "github.com/pingcap-inc/nexuscrm-connectors/features/connector/workqueue"
	"github.com/pingcap-inc/nexuscrm-connectors/internal/id"
)

const (
	defaultWorkItemLeaseDuration = 5 * time.Minute
)

func normalizeClaimPath(path ClaimPath) ClaimPath {
	switch strings.TrimSpace(string(path)) {
	case "", string(ClaimPathLegacyEvent):
		return ClaimPathLegacyEvent
	case string(ClaimPathWorkItem):
		return ClaimPathWorkItem
	default:
		return path
	}
}

func validateClaimPathConfig(cfg WorkerConfig, path ClaimPath) error {
	switch path {
	case ClaimPathLegacyEvent:
		if cfg.WorkItemClaims != nil ||
			cfg.WorkItemEventLoader != nil ||
			cfg.WorkItemEventBridge != nil ||
			strings.TrimSpace(cfg.WorkItemSchedulerOwnerID) != "" ||
			strings.TrimSpace(cfg.WorkItemSchedulerFencingToken) != "" ||
			cfg.WorkItemLeaseDuration != 0 {
			return fmt.Errorf("%w: WorkItem claim config set while ClaimPath is legacy_event", ErrInvalidWorkerConfig)
		}
		return nil
	case ClaimPathWorkItem:
		if cfg.WorkItemClaims == nil {
			return fmt.Errorf("%w: WorkItemClaims is required for work_item claim path", ErrInvalidWorkerConfig)
		}
		if strings.TrimSpace(cfg.WorkItemSchedulerOwnerID) == "" {
			return fmt.Errorf("%w: WorkItemSchedulerOwnerID is required for work_item claim path", ErrInvalidWorkerConfig)
		}
		if strings.TrimSpace(cfg.WorkItemSchedulerFencingToken) == "" {
			return fmt.Errorf("%w: WorkItemSchedulerFencingToken is required for work_item claim path", ErrInvalidWorkerConfig)
		}
		if cfg.WorkItemLeaseDuration < 0 {
			return fmt.Errorf("%w: WorkItemLeaseDuration is negative", ErrInvalidWorkerConfig)
		}
		return nil
	default:
		return fmt.Errorf("%w: unknown ClaimPath %q", ErrInvalidWorkerConfig, path)
	}
}

func validateMandatoryWorkItemConfig(cfg WorkerConfig) error {
	path := ClaimPath(strings.TrimSpace(string(cfg.ClaimPath)))
	if path != "" && path != ClaimPathWorkItem {
		return fmt.Errorf("%w: ClaimPath %q is not accepted by NewWorkItemWorker", ErrInvalidWorkerConfig, path)
	}
	if cfg.WorkItemClaims != nil || cfg.WorkItemEventLoader != nil || cfg.WorkItemEventBridge != nil {
		return fmt.Errorf("%w: split WorkItem dependencies are not accepted by NewWorkItemWorker", ErrInvalidWorkerConfig)
	}
	if strings.TrimSpace(cfg.WorkItemSchedulerOwnerID) == "" {
		return fmt.Errorf("%w: WorkItemSchedulerOwnerID is required", ErrInvalidWorkerConfig)
	}
	if strings.TrimSpace(cfg.WorkItemSchedulerFencingToken) == "" {
		return fmt.Errorf("%w: WorkItemSchedulerFencingToken is required", ErrInvalidWorkerConfig)
	}
	if cfg.WorkItemLeaseDuration < 0 {
		return fmt.Errorf("%w: WorkItemLeaseDuration is negative", ErrInvalidWorkerConfig)
	}
	return nil
}

func workItemClaimInputServiceIsNil(service WorkItemClaimInputService) bool {
	if service == nil {
		return true
	}
	value := reflect.ValueOf(service)
	switch value.Kind() {
	case reflect.Chan, reflect.Func, reflect.Interface, reflect.Map, reflect.Pointer, reflect.Slice:
		return value.IsNil()
	default:
		return false
	}
}

// validateConsolidatedWorkItemClaimInputs admits the consolidated selection
// path only when the EventsService and the configured mutation surface share
// the exact in-tree WorkItem repository. A configured legacy loader keeps the
// published split surfaces authoritative and intentionally skips this path.
func validateConsolidatedWorkItemClaimInputs(events EventsService, cfg WorkerConfig) error {
	if cfg.WorkItemEventLoader != nil {
		return nil
	}
	claimInputs, ok := events.(WorkItemClaimInputService)
	if !ok || workItemClaimInputServiceIsNil(claimInputs) {
		return fmt.Errorf("%w: WorkItemEventLoader is required for the split work_item compatibility path", ErrInvalidWorkerConfig)
	}
	if !claimInputs.HonorsClaimWorkTypes() {
		return fmt.Errorf("%w: WorkItem claim/input service must honor ClaimWorkTypes", ErrInvalidWorkerConfig)
	}
	repositoryOwner, ok := events.(interface {
		WorkItemRepository() *connector_workqueue.Repository
	})
	configuredRepository, configured := cfg.WorkItemClaims.(*connector_workqueue.Repository)
	if !ok || repositoryOwner.WorkItemRepository() == nil || !configured || configuredRepository == nil || repositoryOwner.WorkItemRepository() != configuredRepository {
		return fmt.Errorf("%w: consolidated WorkItem claim/input service must share the configured WorkItemClaims repository", ErrInvalidWorkerConfig)
	}
	return nil
}

type claimedWorkItemRef struct {
	workID     string
	leaseToken string
	retryCount int
	generation int64
	attempt    int
}

type workItemEventsService struct {
	claims                WorkItemClaimService
	claimInputs           WorkItemClaimInputService
	workItems             WorkItemService
	loader                WorkItemEventLoader
	bridge                WorkItemEventBridge
	cfg                   WorkerConfig
	preferredQueueClasses []string
	distributedClaims     bool

	mu      sync.Mutex
	claimed map[string]claimedWorkItemRef
	peeked  map[string]string
}

type workItemClaimLookupService interface {
	// PeekClaimableWorkItems must apply ClaimWorkTypes before its candidate
	// limit. ClaimWorkItemByEvent must honor the requested WorkType.
	PeekClaimableWorkItems(ctx context.Context, in connector_workqueue.ClaimWorkItemInput, limit int) ([]*connector_workqueue.WorkItem, error)
	ClaimWorkItemByEvent(ctx context.Context, in connector_workqueue.ClaimWorkItemByEventInput) (*connector_workqueue.WorkItem, bool, error)
}

type staleWorkItemClaimLookupService interface {
	ClaimStaleWorkItemByEvent(ctx context.Context, in connector_workqueue.ClaimWorkItemByEventInput) (*connector_workqueue.WorkItem, bool, error)
}

type workItemEventBatchLoader interface {
	LoadEventsForWorkItems(ctx context.Context, eventIDs []string) (map[string]*connector_events.ConnectorEvent, error)
}

type workItemFailureService interface {
	FailWorkItem(ctx context.Context, in connector_workqueue.FailWorkItemInput) (*connector_workqueue.FailWorkItemResult, bool, error)
}

type workItemDependencyService interface {
	BlockWorkItem(ctx context.Context, in connector_workqueue.BlockWorkItemInput) (bool, error)
}

func (s *workItemEventsService) dependencyRepository() *connector_workqueue.Repository {
	for _, candidate := range []any{s.workItems, s.claims, s.claimInputs} {
		if owner, ok := candidate.(interface {
			WorkItemRepository() *connector_workqueue.Repository
		}); ok && owner.WorkItemRepository() != nil {
			return owner.WorkItemRepository()
		}
		if repository, ok := candidate.(*connector_workqueue.Repository); ok && repository != nil {
			return repository
		}
	}
	return nil
}

// legacyWorkItemClaimInputs keeps the published split constructor working
// until the compatibility-removal phase. The supervisor-managed path never
// uses this shim: it supplies one admitted WorkItemClaimInputService directly.
type legacyWorkItemClaimInputs struct {
	claims WorkItemClaimService
	loader WorkItemEventLoader
}

func (s legacyWorkItemClaimInputs) HonorsClaimWorkTypes() bool { return false }

func (s legacyWorkItemClaimInputs) ClaimNextWorkItem(ctx context.Context, in connector_workqueue.ClaimWorkItemInput) (*connector_workqueue.WorkItem, bool, error) {
	return s.claims.ClaimNextWorkItem(ctx, in)
}

func (s legacyWorkItemClaimInputs) PeekClaimableWorkItems(ctx context.Context, in connector_workqueue.ClaimWorkItemInput, limit int) ([]*connector_workqueue.WorkItem, error) {
	claims, ok := s.claims.(workItemClaimLookupService)
	if !ok {
		return nil, fmt.Errorf("connector_inbound: WorkItem claim service does not support event-backed lookup")
	}
	return claims.PeekClaimableWorkItems(ctx, in, limit)
}

func (s legacyWorkItemClaimInputs) ClaimWorkItemByEvent(ctx context.Context, in connector_workqueue.ClaimWorkItemByEventInput) (*connector_workqueue.WorkItem, bool, error) {
	claims, ok := s.claims.(workItemClaimLookupService)
	if !ok {
		return nil, false, fmt.Errorf("connector_inbound: WorkItem claim service does not support event-backed lookup")
	}
	return claims.ClaimWorkItemByEvent(ctx, in)
}

func (s legacyWorkItemClaimInputs) ClaimStaleWorkItemByEvent(ctx context.Context, in connector_workqueue.ClaimWorkItemByEventInput) (*connector_workqueue.WorkItem, bool, error) {
	claims, ok := s.claims.(staleWorkItemClaimLookupService)
	if !ok {
		return nil, false, fmt.Errorf("connector_inbound: WorkItem claim service does not support stale event-backed lookup")
	}
	return claims.ClaimStaleWorkItemByEvent(ctx, in)
}

func (s legacyWorkItemClaimInputs) ResolveClaimableWorkItemGeneration(ctx context.Context, in connector_workqueue.ResolveClaimableGenerationInput) (int64, bool, error) {
	resolver, ok := s.claims.(WorkItemClaimGenerationResolver)
	if !ok {
		return in.Generation, false, nil
	}
	return resolver.ResolveClaimableWorkItemGeneration(ctx, in)
}

func (s legacyWorkItemClaimInputs) LoadEventForWorkItem(ctx context.Context, eventID string) (*connector_events.ConnectorEvent, error) {
	if s.loader == nil {
		return nil, fmt.Errorf("connector_inbound: WorkItem event loader is unavailable")
	}
	return s.loader.LoadEventForWorkItem(ctx, eventID)
}

func (s legacyWorkItemClaimInputs) LoadEventsForWorkItems(ctx context.Context, eventIDs []string) (map[string]*connector_events.ConnectorEvent, error) {
	if loader, ok := s.loader.(workItemEventBatchLoader); ok {
		return loader.LoadEventsForWorkItems(ctx, eventIDs)
	}
	result := make(map[string]*connector_events.ConnectorEvent, len(eventIDs))
	for _, eventID := range eventIDs {
		if _, ok := result[eventID]; ok {
			continue
		}
		event, err := s.LoadEventForWorkItem(ctx, eventID)
		if err != nil {
			return nil, err
		}
		result[eventID] = event
	}
	return result, nil
}

func (s legacyWorkItemClaimInputs) LoadSourceProjectionPayloads(ctx context.Context, workID string) ([]*connector_events.ConnectorEvent, error) {
	if s.loader == nil {
		return nil, fmt.Errorf("connector_inbound: WorkItem event loader is unavailable")
	}
	return s.loader.LoadSourceProjectionPayloads(ctx, workID)
}

func newWorkItemEventsService(events EventsService, cfg WorkerConfig) *workItemEventsService {
	var claimInputs WorkItemClaimInputService
	// A non-nil configured loader selects the published split compatibility
	// surfaces. Only the loader-free, constructor-validated path may take claim
	// selection and immutable input from the consolidated EventsService.
	if cfg.WorkItemEventLoader == nil {
		claimInputs, _ = events.(WorkItemClaimInputService)
	}
	return &workItemEventsService{
		claims:      cfg.WorkItemClaims,
		claimInputs: claimInputs,
		loader:      cfg.WorkItemEventLoader,
		bridge:      cfg.WorkItemEventBridge,
		cfg:         cfg,
		claimed:     make(map[string]claimedWorkItemRef),
		peeked:      make(map[string]string),
	}
}

func newMandatoryWorkItemEventsService(workItems WorkItemService, cfg WorkerConfig) *workItemEventsService {
	return &workItemEventsService{
		claimInputs: workItems,
		workItems:   workItems,
		cfg:         cfg,
		claimed:     make(map[string]claimedWorkItemRef),
		peeked:      make(map[string]string),
	}
}

func (s *workItemEventsService) claimInputService() WorkItemClaimInputService {
	if s.claimInputs != nil {
		return s.claimInputs
	}
	return legacyWorkItemClaimInputs{claims: s.claims, loader: s.loader}
}

func (s *workItemEventsService) completeWorkItem(ctx context.Context, workID, leaseToken, terminalState string, now time.Time) (bool, error) {
	if s.workItems != nil {
		return s.workItems.CompleteWorkItem(ctx, workID, leaseToken, terminalState, now)
	}
	return s.claims.CompleteWorkItem(ctx, workID, leaseToken, terminalState, now)
}

func (s *workItemEventsService) retryWorkItem(ctx context.Context, workID, leaseToken string, runAfter, now time.Time) (bool, error) {
	if s.workItems != nil {
		return s.workItems.RetryWorkItem(ctx, workID, leaseToken, runAfter, now)
	}
	return s.claims.RetryWorkItem(ctx, workID, leaseToken, runAfter, now)
}

func (s *workItemEventsService) renewWorkItemLease(ctx context.Context, workID, leaseToken string, leaseUntil, now time.Time) (bool, error) {
	if s.workItems != nil {
		return s.workItems.RenewWorkItemLease(ctx, workID, leaseToken, leaseUntil, now)
	}
	return s.claims.RenewWorkItemLease(ctx, workID, leaseToken, leaseUntil, now)
}

// LoadSettledProjectionPayloads exposes the mandatory WorkItem loader
// capability after the claim adapter replaces the base event service facade.
func (s *workItemEventsService) LoadSettledProjectionPayloads(ctx context.Context, eventID string) ([]*connector_events.ConnectorEvent, error) {
	workID, ok := s.WorkItemIDForEvent(eventID)
	if !ok {
		return nil, connector_events.ErrLeaseLost
	}
	return s.claimInputService().LoadSourceProjectionPayloads(ctx, workID)
}

func (s *workItemEventsService) setPreferredQueueClasses(queueClasses []string) {
	s.preferredQueueClasses = append([]string(nil), queueClasses...)
}

func (s *workItemEventsService) distributedClaimsEnabled() bool {
	if s.workItems != nil {
		return true
	}
	if s.distributedClaims {
		return true
	}
	authority, ok := s.claims.(WorkItemRowCASClaimAuthority)
	return ok && authority.UsesWorkItemRowCASClaimAuthority()
}

func (s *workItemEventsService) workItemExecutorStateEnabled() bool {
	if s.workItems != nil {
		return true
	}
	authority, ok := s.claims.(WorkItemExecutorStateAuthority)
	return ok && authority.UsesWorkItemExecutorStateAuthority()
}

func (s *workItemEventsService) claimContext(ctx context.Context, queueClasses, preferredQueueClasses []string) context.Context {
	if preferredQueueClasses == nil {
		preferredQueueClasses = s.preferredQueueClasses
	}
	ctx = connector_workqueue.WithClaimQueueClassPreference(ctx, preferredQueueClassesForClaim(queueClasses, preferredQueueClasses))
	ctx = connector_workqueue.WithClaimWorkTypes(ctx, eventBackedWorkItemTypes())
	if s.distributedClaimsEnabled() {
		ctx = connector_workqueue.WithDistributedClaim(ctx)
	}
	return ctx
}

func (s *workItemEventsService) ClaimNext(ctx context.Context, installedPackID, workerID string) (*connector_events.ConnectorEvent, bool, error) {
	return s.claimNext(ctx, installedPackID, workerID, nil, false)
}

func (s *workItemEventsService) ClaimNextForClasses(ctx context.Context, installedPackID, workerID string, queueClasses []string) (*connector_events.ConnectorEvent, bool, error) {
	return s.claimNext(ctx, installedPackID, workerID, queueClasses, false)
}

func (s *workItemEventsService) ReclaimStaleNext(ctx context.Context, installedPackID, workerID string) (*connector_events.ConnectorEvent, bool, error) {
	return s.claimNext(ctx, installedPackID, workerID, nil, true)
}

func (s *workItemEventsService) ReclaimStaleNextForClasses(ctx context.Context, installedPackID, workerID string, queueClasses []string) (*connector_events.ConnectorEvent, bool, error) {
	return s.claimNext(ctx, installedPackID, workerID, queueClasses, true)
}

func (s *workItemEventsService) PeekStaleBatch(context.Context, string, int) ([]*connector_events.ConnectorEvent, error) {
	return nil, nil
}

func (s *workItemEventsService) PeekStaleBatchForClasses(context.Context, string, []string, int) ([]*connector_events.ConnectorEvent, error) {
	return nil, nil
}

func (s *workItemEventsService) claimNext(ctx context.Context, installedPackID, workerID string, queueClasses []string, staleOnly bool) (*connector_events.ConnectorEvent, bool, error) {
	now := time.Now().UTC()
	leaseDuration := s.cfg.WorkItemLeaseDuration
	if leaseDuration <= 0 {
		leaseDuration = defaultWorkItemLeaseDuration
	}
	generation, err := s.resolveClaimGeneration(ctx, installedPackID, queueClasses, staleOnly, now)
	if err != nil {
		return nil, false, err
	}
	if err := s.ensureSchedulerLease(ctx, installedPackID, generation, now, leaseDuration); err != nil {
		return nil, false, err
	}
	for {
		leaseToken := id.Generate()
		work, ok, err := s.claimInputService().ClaimNextWorkItem(s.claimContext(ctx, queueClasses, nil), connector_workqueue.ClaimWorkItemInput{
			InstalledPackID: strings.TrimSpace(installedPackID),
			Generation:      generation,
			OwnerID:         strings.TrimSpace(s.cfg.WorkItemSchedulerOwnerID),
			FencingToken:    strings.TrimSpace(s.cfg.WorkItemSchedulerFencingToken),
			LeaseToken:      leaseToken,
			LeaseUntil:      now.Add(leaseDuration),
			QueueClasses:    append([]string(nil), queueClasses...),
			StaleOnly:       staleOnly,
			Now:             now,
		})
		if err != nil || !ok {
			return nil, ok, err
		}
		if work == nil {
			return nil, false, fmt.Errorf("connector_inbound: WorkItem claim service returned a nil WorkItem")
		}
		if err := validateEventBackedWorkItemClaim(work, ""); err != nil {
			_, _ = s.retryWorkItem(ctx, work.WorkID, work.LeaseToken, now.Add(time.Minute), now)
			return nil, false, err
		}
		ev, err := s.claimInputService().LoadEventForWorkItem(ctx, work.EventID)
		if err != nil || ev == nil {
			_, _ = s.retryWorkItem(ctx, work.WorkID, work.LeaseToken, now.Add(time.Minute), now)
			if err == nil {
				err = fmt.Errorf("%w: immutable Event %q for WorkItem %q", connector_events.ErrEventNotFound, work.EventID, work.WorkID)
			}
			return nil, false, err
		}
		if s.workItems != nil || s.workItemExecutorStateEnabled() {
			s.prepareClaimedEvent(ev, work, workerID, now)
			s.remember(ev.ID, claimedWorkItemRef{
				workID: work.WorkID, leaseToken: work.LeaseToken, generation: work.Generation, attempt: work.Attempt,
			})
			return ev, true, nil
		}
		if terminalState, terminal := workItemTerminalStateForEventStatus(ev.Status); terminal {
			ok, completeErr := s.completeWorkItem(ctx, work.WorkID, work.LeaseToken, terminalState, now)
			if completeErr != nil {
				return nil, false, completeErr
			}
			if !ok {
				return nil, false, connector_events.ErrLeaseLost
			}
			continue
		}
		if runAfter, backingOff := workItemEventBackoffRunAfter(ev, now); backingOff {
			retried, retryErr := s.retryWorkItem(ctx, work.WorkID, work.LeaseToken, runAfter, now)
			if retryErr != nil {
				return nil, false, retryErr
			}
			if !retried {
				return nil, false, connector_events.ErrLeaseLost
			}
			continue
		}
		if strings.TrimSpace(ev.Status) == connector_workqueue.WorkStateProcessing && !workItemProcessingStale(ev.LastModifiedDate, now) {
			if connector_workqueue.IsSourceProjectionWorkID(work.WorkID) {
				if retryErr := s.retryWorkItemAfterEventLeaseLost(ctx, work, ev, now); retryErr != nil {
					return nil, false, retryErr
				}
			} else {
				retried, retryErr := s.retryWorkItem(ctx, work.WorkID, work.LeaseToken, now, now)
				if retryErr != nil {
					return nil, false, retryErr
				}
				if !retried {
					return nil, false, connector_events.ErrLeaseLost
				}
			}
			return nil, false, connector_events.ErrLeaseLost
		}
		claimedAt := now
		if !work.LastModifiedDate.IsZero() {
			claimedAt = work.LastModifiedDate
		}
		if err := s.bridge.MarkEventProcessingForWorkItem(ctx, ev.ID, workerID, work.LeaseToken, claimedAt, now); err != nil {
			if errors.Is(err, connector_events.ErrEventAlreadyTerminal) {
				if connector_workqueue.IsSourceProjectionWorkID(work.WorkID) {
					if convergeErr := s.retryWorkItemAfterEventLeaseLost(ctx, work, ev, now); convergeErr != nil {
						return nil, false, convergeErr
					}
					continue
				}
				ok, completeErr := s.completeWorkItem(ctx, work.WorkID, work.LeaseToken, connector_workqueue.WorkStateSkipped, now)
				if completeErr != nil {
					return nil, false, completeErr
				}
				if !ok {
					return nil, false, connector_events.ErrLeaseLost
				}
				continue
			}
			if errors.Is(err, connector_events.ErrLeaseLost) {
				if retryErr := s.retryWorkItemAfterEventLeaseLost(ctx, work, ev, now); retryErr != nil {
					return nil, false, retryErr
				}
				return nil, false, err
			}
			_, _ = s.retryWorkItem(ctx, work.WorkID, work.LeaseToken, now.Add(time.Minute), now)
			return nil, false, err
		}
		ev.Status = "processing"
		ev.ClaimedBy = workerID
		ev.ClaimToken = work.LeaseToken
		ev.ClaimedAt = &claimedAt
		s.remember(ev.ID, claimedWorkItemRef{workID: work.WorkID, leaseToken: work.LeaseToken, retryCount: ev.RetryCount, generation: work.Generation, attempt: work.Attempt})
		return ev, true, nil
	}
}

func isEventBackedWorkItemType(workType string) bool {
	switch strings.TrimSpace(workType) {
	case connector_workqueue.WorkTypeConnectorEvent, connector_workqueue.WorkTypeWorkflowActionApply:
		return true
	default:
		return false
	}
}

func validateEventBackedWorkItemClaim(work *connector_workqueue.WorkItem, expectedWorkType string) error {
	if work == nil {
		return fmt.Errorf("connector_inbound: WorkItem claim service returned a nil WorkItem")
	}
	if !isEventBackedWorkItemType(work.WorkType) {
		return fmt.Errorf("connector_inbound: WorkItem claim service returned out-of-scope unsupported WorkItem type %q", work.WorkType)
	}
	expectedWorkType = strings.TrimSpace(expectedWorkType)
	if expectedWorkType != "" && strings.TrimSpace(work.WorkType) != expectedWorkType {
		return fmt.Errorf("connector_inbound: WorkItem claim service returned type %q outside requested type %q", work.WorkType, expectedWorkType)
	}
	return nil
}

func validateEventBackedWorkItemResults(works []*connector_workqueue.WorkItem) error {
	for _, work := range works {
		if err := validateEventBackedWorkItemClaim(work, ""); err != nil {
			return err
		}
	}
	return nil
}

func validateCoalescedWorkItemClaim(
	claim *connector_events.WorkItemCoalescedSiblingClaim,
	currentEventID, currentWorkID, expectedLeaseToken string,
	expectedGeneration int64,
) error {
	if claim == nil || claim.Event == nil || claim.Work == nil {
		return fmt.Errorf("connector_inbound: WorkItem coalesced claim returned an empty sibling")
	}
	if err := validateEventBackedWorkItemClaim(claim.Work, ""); err != nil {
		return err
	}
	eventID := strings.TrimSpace(claim.Event.ID)
	workID := strings.TrimSpace(claim.Work.WorkID)
	if eventID == "" || workID == "" {
		return fmt.Errorf("connector_inbound: WorkItem coalesced claim returned an unidentified sibling")
	}
	if eventID == strings.TrimSpace(currentEventID) {
		return fmt.Errorf("connector_inbound: WorkItem coalesced claim returned the current Event as its successor")
	}
	if workID == strings.TrimSpace(currentWorkID) {
		return fmt.Errorf("connector_inbound: WorkItem coalesced claim returned the current WorkItem as its successor")
	}
	if connector_workqueue.IsSourceProjectionWorkID(workID) {
		return fmt.Errorf("connector_inbound: WorkItem coalesced claim returned a source-projection WorkItem")
	}
	if strings.TrimSpace(claim.Work.EventID) != eventID {
		return fmt.Errorf("connector_inbound: WorkItem coalesced claim returned mismatched Event and WorkItem identities")
	}
	if strings.TrimSpace(claim.Work.LeaseToken) != strings.TrimSpace(expectedLeaseToken) {
		return fmt.Errorf("connector_inbound: WorkItem coalesced claim returned an unexpected lease token")
	}
	if claim.Work.Generation != expectedGeneration {
		return fmt.Errorf("connector_inbound: WorkItem coalesced claim returned generation %d, want %d", claim.Work.Generation, expectedGeneration)
	}
	if strings.TrimSpace(claim.Work.State) != connector_workqueue.WorkStateProcessing {
		return fmt.Errorf("connector_inbound: WorkItem coalesced claim returned state %q, want %q", claim.Work.State, connector_workqueue.WorkStateProcessing)
	}
	return nil
}

func eventBackedWorkItemTypes() []string {
	return []string{
		connector_workqueue.WorkTypeConnectorEvent,
		connector_workqueue.WorkTypeWorkflowActionApply,
	}
}

func (s *workItemEventsService) resolveClaimGeneration(ctx context.Context, installedPackID string, queueClasses []string, staleOnly bool, now time.Time) (int64, error) {
	generation := s.cfg.WorkItemGeneration
	resolved, found, err := s.claimInputService().ResolveClaimableWorkItemGeneration(connector_workqueue.WithClaimWorkTypes(ctx, eventBackedWorkItemTypes()), connector_workqueue.ResolveClaimableGenerationInput{
		InstalledPackID: strings.TrimSpace(installedPackID),
		Generation:      generation,
		QueueClasses:    append([]string(nil), queueClasses...),
		StaleOnly:       staleOnly,
		Now:             now,
	})
	if err != nil || !found {
		return generation, err
	}
	return resolved, nil
}

func (s *workItemEventsService) ensureSchedulerLease(ctx context.Context, installedPackID string, generation int64, now time.Time, leaseDuration time.Duration) error {
	if s.distributedClaimsEnabled() {
		return nil
	}
	in := connector_workqueue.SchedulerLeaseInput{
		InstalledPackID: strings.TrimSpace(installedPackID),
		Generation:      generation,
		OwnerID:         strings.TrimSpace(s.cfg.WorkItemSchedulerOwnerID),
		FencingToken:    strings.TrimSpace(s.cfg.WorkItemSchedulerFencingToken),
		LeaseUntil:      now.Add(leaseDuration),
		Now:             now,
	}
	renewed, err := s.claims.RenewSchedulerLease(ctx, in)
	if err != nil {
		return err
	}
	if renewed {
		return nil
	}
	acquired, err := s.claims.AcquireSchedulerLease(ctx, in)
	if err != nil {
		return err
	}
	if !acquired {
		return connector_events.ErrLeaseLost
	}
	return nil
}

func (s *workItemEventsService) PeekBatch(ctx context.Context, installedPackID string, n int) ([]*connector_events.ConnectorEvent, error) {
	return s.peekBatch(ctx, installedPackID, nil, n, false)
}

func (s *workItemEventsService) PeekBatchForClasses(ctx context.Context, installedPackID string, queueClasses []string, n int) ([]*connector_events.ConnectorEvent, error) {
	return s.peekBatch(ctx, installedPackID, queueClasses, n, false)
}

type workItemClaimByIDInput struct {
	eventID            string
	workerID           string
	installedPackID    string
	queueClasses       []string
	generation         int64
	generationResolved bool
	staleOnly          bool
}

// workItemProjectedEventClaimService is the internal handoff used by worker
// paths that have already peeked an immutable Event. handled is false only for
// retained split adapters that do not make WorkItem executor state authoritative;
// their public token/time ClaimByID behavior remains unchanged.
type workItemProjectedEventClaimService interface {
	claimProjectedWorkItemEventByID(context.Context, workItemClaimByIDInput) (*connector_events.ConnectorEvent, bool, bool, error)
}

func (s *workItemEventsService) ClaimByID(ctx context.Context, eventID, workerID, installedPackID string, _ sql.NullString, _ int) (string, time.Time, bool, error) {
	return s.claimByID(ctx, workItemClaimByIDInput{
		eventID: eventID, workerID: workerID, installedPackID: installedPackID,
	})
}

func (s *workItemEventsService) ClaimByIDForClasses(ctx context.Context, eventID, workerID, installedPackID string, _ sql.NullString, _ int, queueClasses []string, generation int64, generationResolved bool) (string, time.Time, bool, error) {
	return s.claimByID(ctx, workItemClaimByIDInput{
		eventID: eventID, workerID: workerID, installedPackID: installedPackID,
		queueClasses: queueClasses, generation: generation, generationResolved: generationResolved,
	})
}

func (s *workItemEventsService) ClaimByIDForClassesWithMode(ctx context.Context, eventID, workerID, installedPackID string, _ sql.NullString, _ int, queueClasses []string, generation int64, generationResolved, staleOnly bool) (string, time.Time, bool, error) {
	return s.claimByID(ctx, workItemClaimByIDInput{
		eventID: eventID, workerID: workerID, installedPackID: installedPackID,
		queueClasses: queueClasses, generation: generation, generationResolved: generationResolved, staleOnly: staleOnly,
	})
}

func (s *workItemEventsService) claimProjectedWorkItemEventByID(ctx context.Context, in workItemClaimByIDInput) (*connector_events.ConnectorEvent, bool, bool, error) {
	if s.workItems == nil && !s.workItemExecutorStateEnabled() {
		return nil, false, false, nil
	}
	ev, claimed, err := s.claimWorkItemEventByID(ctx, in)
	return ev, true, claimed, err
}

func (s *workItemEventsService) claimByID(ctx context.Context, in workItemClaimByIDInput) (string, time.Time, bool, error) {
	ev, claimed, err := s.claimWorkItemEventByID(ctx, in)
	if err != nil || !claimed {
		return "", time.Time{}, claimed, err
	}
	if ev == nil || ev.ClaimedAt == nil {
		return "", time.Time{}, false, fmt.Errorf("connector_inbound: WorkItem claim returned no execution projection for Event %q", in.eventID)
	}
	return ev.ClaimToken, *ev.ClaimedAt, true, nil
}

func (s *workItemEventsService) claimWorkItemEventByID(ctx context.Context, in workItemClaimByIDInput) (*connector_events.ConnectorEvent, bool, error) {
	now := time.Now().UTC()
	leaseDuration := s.cfg.WorkItemLeaseDuration
	if leaseDuration <= 0 {
		leaseDuration = defaultWorkItemLeaseDuration
	}
	if !in.generationResolved {
		resolved, err := s.resolveClaimGeneration(ctx, in.installedPackID, in.queueClasses, in.staleOnly, now)
		if err != nil {
			return nil, false, err
		}
		in.generation = resolved
	}
	if err := s.ensureSchedulerLease(ctx, in.installedPackID, in.generation, now, leaseDuration); err != nil {
		return nil, false, err
	}
	work, ok, err := s.claimEventBackedWorkItemByEvent(ctx, in.eventID, in.installedPackID, in.workerID, in.generation, now, leaseDuration, s.claimWorkTypesForEvent(in.eventID), in.staleOnly)
	if err != nil || !ok {
		return nil, ok, err
	}
	if work == nil {
		return nil, false, fmt.Errorf("connector_inbound: WorkItem claim service returned a nil WorkItem for event %q", in.eventID)
	}
	ev, err := s.claimInputService().LoadEventForWorkItem(ctx, work.EventID)
	if err != nil || ev == nil {
		_, _ = s.retryWorkItem(ctx, work.WorkID, work.LeaseToken, now.Add(time.Minute), now)
		if err == nil {
			err = fmt.Errorf("%w: immutable Event %q for WorkItem %q", connector_events.ErrEventNotFound, work.EventID, work.WorkID)
		}
		return nil, false, err
	}
	if s.workItems != nil || s.workItemExecutorStateEnabled() {
		s.prepareClaimedEvent(ev, work, in.workerID, now)
		s.remember(ev.ID, claimedWorkItemRef{
			workID: work.WorkID, leaseToken: work.LeaseToken, generation: work.Generation, attempt: work.Attempt,
		})
		return ev, true, nil
	}
	if terminalState, terminal := workItemTerminalStateForEventStatus(ev.Status); terminal {
		completeOK, completeErr := s.completeWorkItem(ctx, work.WorkID, work.LeaseToken, terminalState, now)
		if completeErr != nil {
			return nil, false, completeErr
		}
		if !completeOK {
			return nil, false, connector_events.ErrLeaseLost
		}
		return nil, false, nil
	}
	if runAfter, backingOff := workItemEventBackoffRunAfter(ev, now); backingOff {
		retried, retryErr := s.retryWorkItem(ctx, work.WorkID, work.LeaseToken, runAfter, now)
		if retryErr != nil {
			return nil, false, retryErr
		}
		if !retried {
			return nil, false, connector_events.ErrLeaseLost
		}
		return nil, false, nil
	}
	claimedAt := now
	if !work.LastModifiedDate.IsZero() {
		claimedAt = work.LastModifiedDate
	}
	if err := s.bridge.MarkEventProcessingForWorkItem(ctx, ev.ID, in.workerID, work.LeaseToken, claimedAt, now); err != nil {
		if errors.Is(err, connector_events.ErrEventAlreadyTerminal) {
			if connector_workqueue.IsSourceProjectionWorkID(work.WorkID) {
				if convergeErr := s.retryWorkItemAfterEventLeaseLost(ctx, work, ev, now); convergeErr != nil {
					return nil, false, convergeErr
				}
				return nil, false, nil
			}
			completeOK, completeErr := s.completeWorkItem(ctx, work.WorkID, work.LeaseToken, connector_workqueue.WorkStateSkipped, now)
			if completeErr != nil {
				return nil, false, completeErr
			}
			if !completeOK {
				return nil, false, connector_events.ErrLeaseLost
			}
			return nil, false, nil
		}
		if errors.Is(err, connector_events.ErrLeaseLost) {
			if retryErr := s.retryWorkItemAfterEventLeaseLost(ctx, work, ev, now); retryErr != nil {
				return nil, false, errors.Join(err, retryErr)
			}
			return nil, false, err
		}
		_, _ = s.retryWorkItem(ctx, work.WorkID, work.LeaseToken, now.Add(time.Minute), now)
		return nil, false, err
	}
	ev.ClaimToken = work.LeaseToken
	ev.ClaimedAt = &claimedAt
	s.remember(ev.ID, claimedWorkItemRef{workID: work.WorkID, leaseToken: work.LeaseToken, retryCount: ev.RetryCount, generation: work.Generation, attempt: work.Attempt})
	return ev, true, nil
}

func (s *workItemEventsService) claimEventBackedWorkItemByEvent(ctx context.Context, eventID, installedPackID, workerID string, generation int64, now time.Time, leaseDuration time.Duration, workTypes []string, staleOnly bool) (*connector_workqueue.WorkItem, bool, error) {
	claimInputs := s.claimInputService()
	for _, workType := range workTypes {
		in := connector_workqueue.ClaimWorkItemByEventInput{
			EventID:         eventID,
			WorkType:        workType,
			InstalledPackID: strings.TrimSpace(installedPackID),
			Generation:      generation,
			OwnerID:         strings.TrimSpace(s.cfg.WorkItemSchedulerOwnerID),
			FencingToken:    strings.TrimSpace(s.cfg.WorkItemSchedulerFencingToken),
			LeaseToken:      id.Generate(),
			LeaseUntil:      now.Add(leaseDuration),
			Now:             now,
		}
		var work *connector_workqueue.WorkItem
		var ok bool
		var err error
		if staleOnly {
			work, ok, err = claimInputs.ClaimStaleWorkItemByEvent(s.claimContext(ctx, nil, nil), in)
		} else {
			work, ok, err = claimInputs.ClaimWorkItemByEvent(s.claimContext(ctx, nil, nil), in)
		}
		if err != nil {
			return nil, false, err
		}
		if !ok {
			continue
		}
		if work == nil {
			return nil, false, fmt.Errorf("connector_inbound: WorkItem claim service returned a nil WorkItem for event %q", eventID)
		}
		if validateErr := validateEventBackedWorkItemClaim(work, workType); validateErr != nil {
			_, _ = s.retryWorkItem(ctx, work.WorkID, work.LeaseToken, now.Add(time.Minute), now)
			return nil, false, validateErr
		}
		if strings.TrimSpace(work.EventID) != strings.TrimSpace(eventID) {
			_, _ = s.retryWorkItem(ctx, work.WorkID, work.LeaseToken, now.Add(time.Minute), now)
			return nil, false, fmt.Errorf("connector_inbound: WorkItem claim service returned event %q outside requested event %q", work.EventID, eventID)
		}
		return work, true, nil
	}
	return nil, false, nil
}

func (s *workItemEventsService) retryWorkItemAfterEventLeaseLost(ctx context.Context, work *connector_workqueue.WorkItem, ev *connector_events.ConnectorEvent, now time.Time) error {
	runAfter := now
	if ev != nil {
		latest, loadErr := s.claimInputService().LoadEventForWorkItem(ctx, ev.ID)
		if loadErr != nil && connector_workqueue.IsSourceProjectionWorkID(work.WorkID) {
			return loadErr
		}
		if loadErr == nil && latest != nil {
			ev = latest
		}
		if retryAt, backingOff := workItemEventBackoffRunAfter(ev, now); backingOff {
			runAfter = retryAt
		}
	}
	if connector_workqueue.IsSourceProjectionWorkID(work.WorkID) && ev != nil {
		if terminalState, terminal := workItemTerminalStateForEventStatus(ev.Status); terminal {
			completed, err := s.completeWorkItem(ctx, work.WorkID, work.LeaseToken, terminalState, now)
			if err != nil {
				return err
			}
			if !completed {
				return connector_events.ErrLeaseLost
			}
			return nil
		}
		if strings.TrimSpace(ev.Status) == connector_workqueue.WorkStateProcessing && !workItemProcessingStale(ev.LastModifiedDate, now) {
			if ev.LastModifiedDate.IsZero() {
				runAfter = now.Add((connector_events.StaleProcessingSeconds + 1) * time.Second)
			} else {
				runAfter = ev.LastModifiedDate.Add((connector_events.StaleProcessingSeconds + 1) * time.Second)
			}
		}
	}
	retried, err := s.retryWorkItem(ctx, work.WorkID, work.LeaseToken, runAfter, now)
	if err != nil {
		return err
	}
	if !retried {
		return connector_events.ErrLeaseLost
	}
	return nil
}

func (s *workItemEventsService) prepareClaimedEvent(ev *connector_events.ConnectorEvent, work *connector_workqueue.WorkItem, workerID string, now time.Time) {
	if ev == nil || work == nil {
		return
	}
	claimedAt := work.LastModifiedDate
	if claimedAt.IsZero() {
		claimedAt = now
	}
	ev.Status = connector_workqueue.WorkStateProcessing
	ev.ClaimedBy = workerID
	ev.ClaimToken = work.LeaseToken
	ev.ClaimedAt = &claimedAt
	ev.NextAttemptAt = nil
	ev.ProcessedAt = nil
	ev.RetryCount = 0
	ev.LastError = ""
	ev.LastErrorAt = nil
	ev.LastModifiedDate = claimedAt
}

func (s *workItemEventsService) peekBatch(ctx context.Context, installedPackID string, queueClasses []string, n int, staleOnly bool) ([]*connector_events.ConnectorEvent, error) {
	if n <= 0 {
		return nil, nil
	}
	now := time.Now().UTC()
	leaseDuration := s.cfg.WorkItemLeaseDuration
	if leaseDuration <= 0 {
		leaseDuration = defaultWorkItemLeaseDuration
	}
	generation, err := s.resolveClaimGeneration(ctx, installedPackID, queueClasses, staleOnly, now)
	if err != nil {
		return nil, err
	}
	if err := s.ensureSchedulerLease(ctx, installedPackID, generation, now, leaseDuration); err != nil {
		return nil, err
	}
	works, err := s.claimInputService().PeekClaimableWorkItems(s.claimContext(ctx, queueClasses, nil), connector_workqueue.ClaimWorkItemInput{
		InstalledPackID: strings.TrimSpace(installedPackID),
		Generation:      generation,
		OwnerID:         strings.TrimSpace(s.cfg.WorkItemSchedulerOwnerID),
		FencingToken:    strings.TrimSpace(s.cfg.WorkItemSchedulerFencingToken),
		LeaseToken:      id.Generate(),
		LeaseUntil:      now.Add(leaseDuration),
		QueueClasses:    append([]string(nil), queueClasses...),
		StaleOnly:       staleOnly,
		Now:             now,
	}, n)
	if err != nil || len(works) == 0 {
		return nil, err
	}
	if err := validateEventBackedWorkItemResults(works); err != nil {
		return nil, err
	}
	events := make([]*connector_events.ConnectorEvent, 0, len(works))
	for _, work := range works {
		ev, loadErr := s.claimInputService().LoadEventForWorkItem(ctx, work.EventID)
		if loadErr != nil {
			return nil, loadErr
		}
		if ev == nil {
			return nil, fmt.Errorf("%w: immutable Event %q for WorkItem %q", connector_events.ErrEventNotFound, work.EventID, work.WorkID)
		}
		s.rememberPeekedWorkType(ev.ID, work.WorkType)
		events = append(events, ev)
	}
	return events, nil
}

func (s *workItemEventsService) peekClaimCandidatesForScheduler(ctx context.Context, installedPackID string, queueClasses, preferredQueueClasses []string, n int, staleOnly bool) ([]schedulerClaimCandidate, error) {
	if n <= 0 {
		return nil, nil
	}
	now := time.Now().UTC()
	leaseDuration := s.cfg.WorkItemLeaseDuration
	if leaseDuration <= 0 {
		leaseDuration = defaultWorkItemLeaseDuration
	}
	generation, err := s.resolveClaimGeneration(ctx, installedPackID, queueClasses, staleOnly, now)
	if err != nil {
		return nil, err
	}
	if err := s.ensureSchedulerLease(ctx, installedPackID, generation, now, leaseDuration); err != nil {
		return nil, err
	}
	works, err := s.claimInputService().PeekClaimableWorkItems(s.claimContext(ctx, queueClasses, preferredQueueClasses), connector_workqueue.ClaimWorkItemInput{
		InstalledPackID: strings.TrimSpace(installedPackID),
		Generation:      generation,
		OwnerID:         strings.TrimSpace(s.cfg.WorkItemSchedulerOwnerID),
		FencingToken:    strings.TrimSpace(s.cfg.WorkItemSchedulerFencingToken),
		LeaseToken:      id.Generate(),
		LeaseUntil:      now.Add(leaseDuration),
		QueueClasses:    append([]string(nil), queueClasses...),
		StaleOnly:       staleOnly,
		Now:             now,
	}, n)
	if err != nil || len(works) == 0 {
		return nil, err
	}
	if err := validateEventBackedWorkItemResults(works); err != nil {
		return nil, err
	}
	eventsByID, err := s.loadCandidateEvents(ctx, works)
	if err != nil {
		return nil, err
	}
	candidates := make([]schedulerClaimCandidate, 0, len(works))
	for _, work := range works {
		ev := eventsByID[work.EventID]
		if ev == nil {
			return nil, fmt.Errorf("%w: immutable Event %q for WorkItem %q", connector_events.ErrEventNotFound, work.EventID, work.WorkID)
		}
		eligibleAt := work.RunAfter
		if eligibleAt.IsZero() || (!work.CreatedDate.IsZero() && work.CreatedDate.After(eligibleAt)) {
			eligibleAt = work.CreatedDate
		}
		candidates = append(candidates, schedulerClaimCandidate{
			event:              ev,
			workID:             work.WorkID,
			workType:           work.WorkType,
			queueClass:         work.QueueClass,
			generation:         generation,
			generationResolved: true,
			staleOnly:          staleOnly,
			eligibleAt:         eligibleAt,
		})
	}
	return candidates, nil
}

func (s *workItemEventsService) loadCandidateEvents(ctx context.Context, works []*connector_workqueue.WorkItem) (map[string]*connector_events.ConnectorEvent, error) {
	eventIDs := make([]string, 0, len(works))
	for _, work := range works {
		if work != nil && isEventBackedWorkItemType(work.WorkType) {
			eventIDs = append(eventIDs, work.EventID)
		}
	}
	return s.claimInputService().LoadEventsForWorkItems(ctx, eventIDs)
}

func preferredQueueClassesForClaim(queueClasses, preferredQueueClasses []string) []string {
	if len(preferredQueueClasses) == 0 {
		return nil
	}
	if len(queueClasses) == 0 {
		return append([]string(nil), preferredQueueClasses...)
	}
	allowed := make(map[string]struct{}, len(queueClasses))
	for _, class := range queueClasses {
		class = strings.TrimSpace(class)
		if class != "" {
			allowed[class] = struct{}{}
		}
	}
	out := make([]string, 0, len(preferredQueueClasses))
	for _, class := range preferredQueueClasses {
		class = strings.TrimSpace(class)
		if _, ok := allowed[class]; ok {
			out = append(out, class)
		}
	}
	return out
}

type sourceScopedWorkItemEventBridge interface {
	ClaimNextCoalescedWorkItemSiblingSourceScoped(ctx context.Context, currentEventID, currentWorkID, currentLeaseToken, workerID, ownerID, fencingToken, nextLeaseToken string, generation int64, leaseUntil, now time.Time) (*connector_events.WorkItemCoalescedSiblingClaim, bool, error)
}

type executorStateWorkItemEventBridge interface {
	ClaimNextCoalescedExecutorWorkItemSibling(ctx context.Context, currentEventID, currentWorkID, currentLeaseToken, workerID, ownerID, fencingToken, nextLeaseToken string, generation int64, leaseUntil, now time.Time) (*connector_events.WorkItemCoalescedSiblingClaim, bool, error)
}

type sourceScopedExecutorStateWorkItemEventBridge interface {
	ClaimNextCoalescedExecutorWorkItemSiblingSourceScoped(ctx context.Context, currentEventID, currentWorkID, currentLeaseToken, workerID, ownerID, fencingToken, nextLeaseToken string, generation int64, leaseUntil, now time.Time) (*connector_events.WorkItemCoalescedSiblingClaim, bool, error)
}

type sourceProjectionTerminalWorkItemBridge interface {
	CompleteSourceProjectionEventForWorkItem(ctx context.Context, eventID, workID, leaseToken, terminalStatus, errorMsg string, now time.Time) (bool, error)
}

type atomicTerminalWorkItemBridge interface {
	CompleteEventForWorkItem(ctx context.Context, eventID, workID, leaseToken, terminalStatus, errorMsg string, now time.Time) (bool, error)
}

// atomicTerminalWorkItemRepositoryBridge identifies the repository that the
// bridge will complete inside its transaction. The generic bridge is usable
// only when it owns the configured WorkItemClaims implementation; otherwise a
// custom claim implementation remains the completion authority.
type atomicTerminalWorkItemRepositoryBridge interface {
	atomicTerminalWorkItemBridge
	WorkItemRepository() *connector_workqueue.Repository
}

type atomicFailureWorkItemBridge interface {
	FailEventForWorkItem(ctx context.Context, eventID, workID, leaseToken, errorMsg string, now time.Time) (*connector_workqueue.FailWorkItemResult, bool, error)
}

type atomicFailureWorkItemRepositoryBridge interface {
	atomicFailureWorkItemBridge
	WorkItemRepository() *connector_workqueue.Repository
}

type sourceProjectionRetryWorkItemBridge interface {
	RetrySourceProjectionEventForWorkItem(ctx context.Context, eventID, workID, leaseToken, errorMsg string, runAfter, now time.Time, retryCount int) (bool, error)
}

func (s *workItemEventsService) ClaimNextCoalescedSibling(ctx context.Context, currentEventID, currentClaimToken, workerID string) (*connector_events.ConnectorEvent, bool, error) {
	return s.claimNextCoalescedSibling(ctx, currentEventID, currentClaimToken, workerID, false)
}

func (s *workItemEventsService) ClaimNextCoalescedSiblingSourceScoped(ctx context.Context, currentEventID, currentClaimToken, workerID string) (*connector_events.ConnectorEvent, bool, error) {
	return s.claimNextCoalescedSibling(ctx, currentEventID, currentClaimToken, workerID, true)
}

func (s *workItemEventsService) claimNextCoalescedSibling(ctx context.Context, currentEventID, currentClaimToken, workerID string, allowCrossSourceDelivery bool) (*connector_events.ConnectorEvent, bool, error) {
	ref, err := s.claimedRef(currentEventID, currentClaimToken)
	if err != nil {
		return nil, false, err
	}
	// A durable source-projection WorkItem chooses its successor atomically on
	// completion. Running the legacy pre-execution bridge here could race that
	// handoff and replace the claimed snapshot before it is executed.
	if connector_workqueue.IsSourceProjectionWorkID(ref.workID) {
		return nil, false, nil
	}
	now := time.Now().UTC()
	leaseDuration := s.cfg.WorkItemLeaseDuration
	if leaseDuration <= 0 {
		leaseDuration = defaultWorkItemLeaseDuration
	}
	nextLeaseToken := id.Generate()
	claimCtx := s.claimContext(ctx, nil, nil)
	var claim *connector_events.WorkItemCoalescedSiblingClaim
	var ok bool
	executorState := s.workItems != nil || s.workItemExecutorStateEnabled()
	if s.workItems != nil {
		if allowCrossSourceDelivery {
			claim, ok, err = s.workItems.ClaimNextCoalescedExecutorWorkItemSiblingSourceScoped(claimCtx, currentEventID, ref.workID, ref.leaseToken, workerID, strings.TrimSpace(s.cfg.WorkItemSchedulerOwnerID), strings.TrimSpace(s.cfg.WorkItemSchedulerFencingToken), nextLeaseToken, ref.generation, now.Add(leaseDuration), now)
		} else {
			claim, ok, err = s.workItems.ClaimNextCoalescedExecutorWorkItemSibling(claimCtx, currentEventID, ref.workID, ref.leaseToken, workerID, strings.TrimSpace(s.cfg.WorkItemSchedulerOwnerID), strings.TrimSpace(s.cfg.WorkItemSchedulerFencingToken), nextLeaseToken, ref.generation, now.Add(leaseDuration), now)
		}
	} else if executorState {
		if allowCrossSourceDelivery {
			if scoped, supported := s.bridge.(sourceScopedExecutorStateWorkItemEventBridge); supported {
				claim, ok, err = scoped.ClaimNextCoalescedExecutorWorkItemSiblingSourceScoped(claimCtx, currentEventID, ref.workID, ref.leaseToken, workerID, strings.TrimSpace(s.cfg.WorkItemSchedulerOwnerID), strings.TrimSpace(s.cfg.WorkItemSchedulerFencingToken), nextLeaseToken, ref.generation, now.Add(leaseDuration), now)
			} else {
				executorBridge, supported := s.bridge.(executorStateWorkItemEventBridge)
				if !supported {
					return nil, false, fmt.Errorf("connector_inbound: repository-backed WorkItem coalescing capability is required")
				}
				claim, ok, err = executorBridge.ClaimNextCoalescedExecutorWorkItemSibling(claimCtx, currentEventID, ref.workID, ref.leaseToken, workerID, strings.TrimSpace(s.cfg.WorkItemSchedulerOwnerID), strings.TrimSpace(s.cfg.WorkItemSchedulerFencingToken), nextLeaseToken, ref.generation, now.Add(leaseDuration), now)
			}
		} else {
			executorBridge, supported := s.bridge.(executorStateWorkItemEventBridge)
			if !supported {
				return nil, false, fmt.Errorf("connector_inbound: repository-backed WorkItem coalescing capability is required")
			}
			claim, ok, err = executorBridge.ClaimNextCoalescedExecutorWorkItemSibling(claimCtx, currentEventID, ref.workID, ref.leaseToken, workerID, strings.TrimSpace(s.cfg.WorkItemSchedulerOwnerID), strings.TrimSpace(s.cfg.WorkItemSchedulerFencingToken), nextLeaseToken, ref.generation, now.Add(leaseDuration), now)
		}
	} else if allowCrossSourceDelivery {
		if scoped, supported := s.bridge.(sourceScopedWorkItemEventBridge); supported {
			claim, ok, err = scoped.ClaimNextCoalescedWorkItemSiblingSourceScoped(claimCtx, currentEventID, ref.workID, ref.leaseToken, workerID, strings.TrimSpace(s.cfg.WorkItemSchedulerOwnerID), strings.TrimSpace(s.cfg.WorkItemSchedulerFencingToken), nextLeaseToken, ref.generation, now.Add(leaseDuration), now)
		} else {
			claim, ok, err = s.bridge.ClaimNextCoalescedWorkItemSibling(claimCtx, currentEventID, ref.workID, ref.leaseToken, workerID, strings.TrimSpace(s.cfg.WorkItemSchedulerOwnerID), strings.TrimSpace(s.cfg.WorkItemSchedulerFencingToken), nextLeaseToken, ref.generation, now.Add(leaseDuration), now)
		}
	} else {
		claim, ok, err = s.bridge.ClaimNextCoalescedWorkItemSibling(claimCtx, currentEventID, ref.workID, ref.leaseToken, workerID, strings.TrimSpace(s.cfg.WorkItemSchedulerOwnerID), strings.TrimSpace(s.cfg.WorkItemSchedulerFencingToken), nextLeaseToken, ref.generation, now.Add(leaseDuration), now)
	}
	if err != nil || !ok {
		return nil, ok, err
	}
	if s.workItems != nil {
		// A successful mandatory-service handoff has already terminalized the
		// current durable claim. Forget it even when a custom service violates
		// the successor contract, then release only the lease token generated for
		// this handoff.
		s.forget(currentEventID)
		if validationErr := validateCoalescedWorkItemClaim(claim, currentEventID, ref.workID, nextLeaseToken, ref.generation); validationErr != nil {
			if claim == nil || claim.Work == nil || strings.TrimSpace(claim.Work.WorkID) == "" {
				return nil, false, validationErr
			}
			followupCtx, cancel := s.finalizeFollowupContext(ctx)
			released, releaseErr := s.retryWorkItem(followupCtx, claim.Work.WorkID, nextLeaseToken, now.Add(time.Minute), now)
			cancel()
			if releaseErr != nil {
				return nil, false, errors.Join(validationErr, fmt.Errorf("connector_inbound: release invalid WorkItem coalesced claim: %w", releaseErr))
			}
			if !released {
				return nil, false, errors.Join(validationErr, fmt.Errorf("connector_inbound: release invalid WorkItem coalesced claim: %w", connector_events.ErrLeaseLost))
			}
			return nil, false, validationErr
		}
	} else {
		if claim == nil || claim.Event == nil || claim.Work == nil {
			return nil, false, fmt.Errorf("connector_inbound: WorkItem coalesced claim returned an empty sibling")
		}
		s.forget(currentEventID)
	}
	if executorState {
		s.prepareClaimedEvent(claim.Event, claim.Work, workerID, now)
	} else {
		claim.Event.Status = connector_workqueue.WorkStateProcessing
		claim.Event.ClaimedBy = workerID
		claim.Event.ClaimToken = claim.Work.LeaseToken
		claimedAt := claim.Work.LastModifiedDate
		if claimedAt.IsZero() {
			claimedAt = now
		}
		claim.Event.ClaimedAt = &claimedAt
	}
	s.remember(claim.Event.ID, claimedWorkItemRef{
		workID:     claim.Work.WorkID,
		leaseToken: claim.Work.LeaseToken,
		retryCount: claim.Event.RetryCount,
		generation: claim.Work.Generation,
		attempt:    claim.Work.Attempt,
	})
	return claim.Event, true, nil
}

func (s *workItemEventsService) MarkProcessed(ctx context.Context, eventID, claimToken string) error {
	return s.complete(ctx, eventID, claimToken, connector_workqueue.WorkStateProcessed)
}

func (s *workItemEventsService) MarkSkipped(ctx context.Context, eventID, claimToken string) error {
	return s.complete(ctx, eventID, claimToken, connector_workqueue.WorkStateSkipped)
}

func (s *workItemEventsService) MarkFailed(ctx context.Context, eventID, claimToken, errorMsg string) error {
	ref, err := s.claimedRef(eventID, claimToken)
	if err != nil {
		return err
	}
	now := time.Now().UTC()
	if s.workItems != nil || s.workItemExecutorStateEnabled() {
		var failures workItemFailureService
		if s.workItems != nil {
			failures = s.workItems
		} else {
			var ok bool
			failures, ok = s.claims.(workItemFailureService)
			if !ok {
				return fmt.Errorf("connector_inbound: repository-backed WorkItem failure capability is required")
			}
		}
		followupCtx, cancel := s.finalizeFollowupContext(ctx)
		defer cancel()
		delays := make([]time.Duration, connector_events.MaxRetryAttempts)
		for i := range delays {
			delays[i] = claimpolicy.FailureRetryDelay(i + 1)
		}
		_, failed, err := failures.FailWorkItem(followupCtx, connector_workqueue.FailWorkItemInput{
			WorkID:         ref.workID,
			EventID:        eventID,
			LeaseToken:     ref.leaseToken,
			FailureClass:   connector_workqueue.FailureClassInboundExecution,
			FailureSummary: connector_workqueue.FailureSummaryInbound,
			RetryThreshold: connector_events.MaxRetryAttempts,
			RetryDelays:    delays,
			Now:            now,
		})
		if err != nil {
			return err
		}
		if !failed {
			return connector_events.ErrLeaseLost
		}
		s.forget(eventID)
		return nil
	}
	if bridge, ok := s.atomicFailureBridge(); ok {
		followupCtx, cancel := s.finalizeFollowupContext(ctx)
		defer cancel()
		_, failed, err := bridge.FailEventForWorkItem(followupCtx, eventID, ref.workID, ref.leaseToken, errorMsg, now)
		if err != nil {
			return err
		}
		if !failed {
			return connector_events.ErrLeaseLost
		}
		s.forget(eventID)
		return nil
	}
	nextRetryCount := ref.retryCount + 1
	if nextRetryCount >= connector_events.MaxRetryAttempts {
		return s.completeTerminalWorkItem(ctx, eventID, ref, connector_workqueue.WorkStateDead, errorMsg, now)
	}
	runAfter := now.Add(claimpolicy.FailureRetryDelay(nextRetryCount))
	return s.retryClaimedWorkItem(ctx, eventID, ref, errorMsg, runAfter, now, nextRetryCount)
}

func (s *workItemEventsService) BlockOnIdentityConvergence(ctx context.Context, eventID, claimToken string, probe IdentityConvergenceProbe) error {
	ref, err := s.claimedRef(eventID, claimToken)
	if err != nil {
		return err
	}
	dependencies := s.identityConvergenceDependencyService()
	if dependencies == nil {
		return fmt.Errorf("connector_inbound: repository-backed WorkItem dependency capability is required")
	}
	followupCtx, cancel := s.finalizeFollowupContext()
	defer cancel()
	blocked, err := dependencies.BlockWorkItem(followupCtx, connector_workqueue.BlockWorkItemInput{
		WorkID:     ref.workID,
		EventID:    eventID,
		LeaseToken: ref.leaseToken,
		Dependency: probe.Dependency,
		Now:        time.Now().UTC(),
	})
	if err != nil {
		return err
	}
	if !blocked {
		return connector_events.ErrLeaseLost
	}
	s.forget(eventID)
	// BlockWorkItem persists an immediately-due authority recheck.
	// The supervisor owns that probe so the worker never waits on a host
	// resolver after surrendering its lease. An unresolved probe schedules the
	// next exact recheck; producer signals remain latency optimizations only.
	return nil
}

func (s *workItemEventsService) SupportsIdentityConvergenceBlocking() bool {
	return s.identityConvergenceDependencyService() != nil
}

func (s *workItemEventsService) identityConvergenceDependencyService() workItemDependencyService {
	if repository := s.dependencyRepository(); repository != nil {
		return repository
	}
	if s.workItems != nil {
		dependencies, _ := s.workItems.(workItemDependencyService)
		return dependencies
	}
	dependencies, _ := s.claims.(workItemDependencyService)
	return dependencies
}

func (s *workItemEventsService) ReleaseClaim(ctx context.Context, eventID, claimToken string) error {
	return s.retryAt(ctx, eventID, claimToken, time.Now().UTC())
}

func (s *workItemEventsService) ReleaseClaimForRetry(ctx context.Context, eventID, claimToken, _ string) error {
	return s.retryAt(ctx, eventID, claimToken, time.Now().UTC().Add(30*time.Second))
}

func (s *workItemEventsService) RenewLease(ctx context.Context, eventID, claimToken string) error {
	ref, err := s.claimedRef(eventID, claimToken)
	if err != nil {
		return err
	}
	now := time.Now().UTC()
	leaseDuration := s.cfg.WorkItemLeaseDuration
	if leaseDuration <= 0 {
		leaseDuration = defaultWorkItemLeaseDuration
	}
	if s.workItems != nil || s.workItemExecutorStateEnabled() {
		ok, err := s.renewWorkItemLease(ctx, ref.workID, ref.leaseToken, now.Add(leaseDuration), now)
		if err != nil {
			return err
		}
		if !ok {
			return connector_events.ErrLeaseLost
		}
		return nil
	}
	if err := s.bridge.RenewEventLeaseForWorkItem(ctx, eventID, ref.leaseToken, now); err != nil {
		var releaseErr error
		if connector_workqueue.IsSourceProjectionWorkID(ref.workID) {
			followupCtx, cancel := s.finalizeFollowupContext(ctx)
			releaseErr = s.retryWorkItemAfterEventLeaseLost(followupCtx, &connector_workqueue.WorkItem{
				WorkID: ref.workID, LeaseToken: ref.leaseToken,
			}, &connector_events.ConnectorEvent{ID: eventID}, now)
			cancel()
		} else {
			_, releaseErr = s.retryWorkItem(ctx, ref.workID, ref.leaseToken, now, now)
		}
		if releaseErr != nil {
			return errors.Join(err, releaseErr)
		}
		s.forget(eventID)
		return err
	}
	ok, err := s.renewWorkItemLease(ctx, ref.workID, ref.leaseToken, now.Add(leaseDuration), now)
	if err != nil {
		return err
	}
	if !ok {
		return connector_events.ErrLeaseLost
	}
	return nil
}

func (s *workItemEventsService) complete(ctx context.Context, eventID, claimToken, terminalState string) error {
	ref, err := s.claimedRef(eventID, claimToken)
	if err != nil {
		return err
	}
	return s.completeTerminalWorkItem(ctx, eventID, ref, terminalState, "", time.Now().UTC())
}

func (s *workItemEventsService) completeTerminalWorkItem(ctx context.Context, eventID string, ref claimedWorkItemRef, terminalState, errorMsg string, now time.Time) error {
	followupCtx, cancel := s.finalizeFollowupContext(ctx)
	defer cancel()
	if s.workItems != nil || s.workItemExecutorStateEnabled() {
		completed, err := s.completeWorkItem(followupCtx, ref.workID, ref.leaseToken, terminalState, now)
		if err != nil {
			return err
		}
		if !completed {
			return connector_events.ErrLeaseLost
		}
		s.forget(eventID)
		return nil
	}
	if bridge, ok := s.atomicTerminalBridge(); ok {
		completed, err := bridge.CompleteEventForWorkItem(followupCtx, eventID, ref.workID, ref.leaseToken, terminalState, errorMsg, now)
		if err != nil {
			return err
		}
		if !completed {
			return connector_events.ErrLeaseLost
		}
		s.forget(eventID)
		return nil
	}
	if connector_workqueue.IsSourceProjectionWorkID(ref.workID) {
		bridge, ok := s.bridge.(sourceProjectionTerminalWorkItemBridge)
		if !ok {
			return fmt.Errorf("connector_inbound: atomic source-projection terminal bridge is required")
		}
		completed, err := bridge.CompleteSourceProjectionEventForWorkItem(followupCtx, eventID, ref.workID, ref.leaseToken, terminalState, errorMsg, now)
		if err != nil {
			return err
		}
		if !completed {
			return connector_events.ErrLeaseLost
		}
		s.forget(eventID)
		return nil
	}
	if err := s.bridge.MarkEventTerminalForWorkItem(ctx, eventID, ref.leaseToken, terminalState, errorMsg, now); err != nil {
		return err
	}
	ok, err := s.completeWorkItem(followupCtx, ref.workID, ref.leaseToken, terminalState, now)
	if err != nil {
		return err
	}
	if !ok {
		return connector_events.ErrLeaseLost
	}
	s.forget(eventID)
	return nil
}

func (s *workItemEventsService) atomicTerminalBridge() (atomicTerminalWorkItemBridge, bool) {
	bridge, ok := s.bridge.(atomicTerminalWorkItemRepositoryBridge)
	if !ok {
		return nil, false
	}
	claims, ok := s.claims.(*connector_workqueue.Repository)
	if !ok || claims == nil || claims != bridge.WorkItemRepository() {
		return nil, false
	}
	return bridge, true
}

func (s *workItemEventsService) atomicFailureBridge() (atomicFailureWorkItemBridge, bool) {
	bridge, ok := s.bridge.(atomicFailureWorkItemRepositoryBridge)
	if !ok {
		return nil, false
	}
	claims, ok := s.claims.(*connector_workqueue.Repository)
	if !ok || claims == nil || claims != bridge.WorkItemRepository() {
		return nil, false
	}
	return bridge, true
}

func (s *workItemEventsService) retryAt(ctx context.Context, eventID, claimToken string, runAfter time.Time) error {
	ref, err := s.claimedRef(eventID, claimToken)
	if err != nil {
		return err
	}
	now := time.Now().UTC()
	return s.retryClaimedWorkItem(ctx, eventID, ref, "work item retry", runAfter, now, ref.retryCount)
}

func (s *workItemEventsService) retryClaimedWorkItem(ctx context.Context, eventID string, ref claimedWorkItemRef, errorMsg string, runAfter, now time.Time, retryCount int) error {
	if s.workItems != nil || s.workItemExecutorStateEnabled() {
		followupCtx, cancel := s.finalizeFollowupContext(ctx)
		defer cancel()
		retried, err := s.retryWorkItem(followupCtx, ref.workID, ref.leaseToken, runAfter, now)
		if err != nil {
			return err
		}
		if !retried {
			return connector_events.ErrLeaseLost
		}
		s.forget(eventID)
		return nil
	}
	if connector_workqueue.IsSourceProjectionWorkID(ref.workID) {
		bridge, ok := s.bridge.(sourceProjectionRetryWorkItemBridge)
		if !ok {
			return fmt.Errorf("connector_inbound: atomic source-projection retry bridge is required")
		}
		followupCtx, cancel := s.finalizeFollowupContext(ctx)
		defer cancel()
		retried, err := bridge.RetrySourceProjectionEventForWorkItem(followupCtx, eventID, ref.workID, ref.leaseToken, errorMsg, runAfter, now, retryCount)
		if err != nil {
			return err
		}
		if !retried {
			return connector_events.ErrLeaseLost
		}
		s.forget(eventID)
		return nil
	}
	if err := s.bridge.ReleaseEventForWorkItemRetry(ctx, eventID, ref.leaseToken, errorMsg, runAfter, now, retryCount); err != nil {
		return err
	}
	followupCtx, cancel := s.finalizeFollowupContext(ctx)
	defer cancel()
	ok, err := s.retryWorkItem(followupCtx, ref.workID, ref.leaseToken, runAfter, now)
	if err != nil {
		return err
	}
	if !ok {
		return connector_events.ErrLeaseLost
	}
	s.forget(eventID)
	return nil
}

func (s *workItemEventsService) finalizeFollowupContext(parent ...context.Context) (context.Context, context.CancelFunc) {
	timeout := s.cfg.FinalizeTimeout
	if timeout <= 0 {
		timeout = defaultFinalizeTimeout
	}
	base := workerBackgroundContext()
	if len(parent) > 0 && parent[0] != nil {
		base = context.WithoutCancel(parent[0])
	}
	return context.WithTimeout(base, timeout)
}

func (s *workItemEventsService) remember(eventID string, ref claimedWorkItemRef) {
	s.mu.Lock()
	defer s.mu.Unlock()
	s.claimed[eventID] = ref
}

func (s *workItemEventsService) rememberPeekedWorkType(eventID, workType string) {
	if strings.TrimSpace(eventID) == "" || !isEventBackedWorkItemType(workType) {
		return
	}
	s.mu.Lock()
	defer s.mu.Unlock()
	s.peeked[eventID] = strings.TrimSpace(workType)
}

func (s *workItemEventsService) ForgetPeekedCandidate(eventID string) {
	eventID = strings.TrimSpace(eventID)
	if eventID == "" {
		return
	}
	s.mu.Lock()
	defer s.mu.Unlock()
	delete(s.peeked, eventID)
}

func (s *workItemEventsService) claimWorkTypesForEvent(eventID string) []string {
	s.mu.Lock()
	workType := s.peeked[eventID]
	delete(s.peeked, eventID)
	s.mu.Unlock()
	if isEventBackedWorkItemType(workType) {
		return []string{strings.TrimSpace(workType)}
	}
	// Direct ClaimByID callers may not have populated the peek cache, and older
	// rows may use either event-backed WorkType. Probe both durable identities;
	// each claim remains a WorkItem row CAS.
	return []string{
		connector_workqueue.WorkTypeConnectorEvent,
		connector_workqueue.WorkTypeWorkflowActionApply,
	}
}

func (s *workItemEventsService) claimedRef(eventID, claimToken string) (claimedWorkItemRef, error) {
	s.mu.Lock()
	defer s.mu.Unlock()
	ref, ok := s.claimed[eventID]
	if !ok || ref.leaseToken != claimToken {
		return claimedWorkItemRef{}, connector_events.ErrLeaseLost
	}
	return ref, nil
}

// WorkItemIDForEvent exposes only immutable execution lineage for audit
// correlation. Lease and scheduler authority remain private to this adapter.
func (s *workItemEventsService) WorkItemIDForEvent(eventID string) (string, bool) {
	eventID = strings.TrimSpace(eventID)
	if eventID == "" {
		return "", false
	}
	s.mu.Lock()
	defer s.mu.Unlock()
	ref, ok := s.claimed[eventID]
	workID := strings.TrimSpace(ref.workID)
	return workID, ok && workID != ""
}

// WorkItemAttemptForEvent exposes only the durable claim number for trace
// correlation. Lease and mutation authority remain private to this adapter.
func (s *workItemEventsService) WorkItemAttemptForEvent(eventID string) (int, bool) {
	eventID = strings.TrimSpace(eventID)
	if eventID == "" {
		return 0, false
	}
	s.mu.Lock()
	defer s.mu.Unlock()
	ref, ok := s.claimed[eventID]
	return ref.attempt, ok && ref.attempt > 0
}

func (s *workItemEventsService) forget(eventID string) {
	s.mu.Lock()
	defer s.mu.Unlock()
	delete(s.claimed, eventID)
}

func workItemProcessingStale(lastModified time.Time, now time.Time) bool {
	return !lastModified.IsZero() && lastModified.Before(now.Add(-connector_events.StaleProcessingSeconds*time.Second))
}

func workItemEventBackoffRunAfter(ev *connector_events.ConnectorEvent, now time.Time) (time.Time, bool) {
	if ev == nil || strings.TrimSpace(ev.Status) != "failed" || ev.NextAttemptAt == nil || !ev.NextAttemptAt.After(now) {
		return time.Time{}, false
	}
	return *ev.NextAttemptAt, true
}

func workItemTerminalStateForEventStatus(status string) (string, bool) {
	switch strings.TrimSpace(status) {
	case connector_workqueue.WorkStateProcessed:
		return connector_workqueue.WorkStateProcessed, true
	case connector_workqueue.WorkStateSkipped:
		return connector_workqueue.WorkStateSkipped, true
	case connector_workqueue.WorkStateDead:
		return connector_workqueue.WorkStateDead, true
	default:
		return "", false
	}
}