package connector_inbound

import (
	"context"
	"errors"
	"testing"
	"time"

	connector_events "github.com/pingcap-inc/nexuscrm-connectors/features/connector/events"
	connector_inbound_routing "github.com/pingcap-inc/nexuscrm-connectors/features/connector/inboundrouting"
	connector_workqueue "github.com/pingcap-inc/nexuscrm-connectors/features/connector/workqueue"
)

var _ WorkItemService = (*connector_events.Service)(nil)
var _ WorkItemService = (*completeWorkItemServiceFake)(nil)

type completeWorkItemServiceFake struct {
	claims                *fakeWorkItemClaims
	inputs                testWorkItemClaimInputService
	bridge                *fakeWorkItemEventBridge
	coalescedClaimMutator func(*connector_events.WorkItemCoalescedSiblingClaim)
	cutoverErr            error
}

type workItemServiceWithoutCutover struct{ WorkItemService }

type testWorkItemClaimInputService interface {
	HonorsClaimWorkTypes() bool
	ClaimNextWorkItem(context.Context, connector_workqueue.ClaimWorkItemInput) (*connector_workqueue.WorkItem, bool, error)
	PeekClaimableWorkItems(context.Context, connector_workqueue.ClaimWorkItemInput, int) ([]*connector_workqueue.WorkItem, error)
	ClaimWorkItemByEvent(context.Context, connector_workqueue.ClaimWorkItemByEventInput) (*connector_workqueue.WorkItem, bool, error)
	ClaimStaleWorkItemByEvent(context.Context, connector_workqueue.ClaimWorkItemByEventInput) (*connector_workqueue.WorkItem, bool, error)
	ResolveClaimableWorkItemGeneration(context.Context, connector_workqueue.ResolveClaimableGenerationInput) (int64, bool, error)
	LoadEventForWorkItem(context.Context, string) (*connector_events.ConnectorEvent, error)
	LoadEventsForWorkItems(context.Context, []string) (map[string]*connector_events.ConnectorEvent, error)
	LoadSourceProjectionPayloads(context.Context, string) ([]*connector_events.ConnectorEvent, error)
}

func newCompleteWorkItemServiceFake(claims *fakeWorkItemClaims, loader testWorkItemEventLoader, bridge *fakeWorkItemEventBridge) *completeWorkItemServiceFake {
	return &completeWorkItemServiceFake{
		claims: claims,
		inputs: newFakeWorkItemClaimInputs(claims, loader),
		bridge: bridge,
	}
}

func (f *completeWorkItemServiceFake) HonorsClaimWorkTypes() bool {
	return f != nil && f.inputs != nil && f.inputs.HonorsClaimWorkTypes()
}

func (f *completeWorkItemServiceFake) RequireWorkItemOnlyExecutionCutover(context.Context) error {
	return f.cutoverErr
}

func (f *completeWorkItemServiceFake) ClaimNextWorkItem(ctx context.Context, in connector_workqueue.ClaimWorkItemInput) (*connector_workqueue.WorkItem, bool, error) {
	return f.inputs.ClaimNextWorkItem(ctx, in)
}

func (f *completeWorkItemServiceFake) PeekClaimableWorkItems(ctx context.Context, in connector_workqueue.ClaimWorkItemInput, limit int) ([]*connector_workqueue.WorkItem, error) {
	return f.inputs.PeekClaimableWorkItems(ctx, in, limit)
}

func (f *completeWorkItemServiceFake) ClaimWorkItemByEvent(ctx context.Context, in connector_workqueue.ClaimWorkItemByEventInput) (*connector_workqueue.WorkItem, bool, error) {
	return f.inputs.ClaimWorkItemByEvent(ctx, in)
}

func (f *completeWorkItemServiceFake) ClaimStaleWorkItemByEvent(ctx context.Context, in connector_workqueue.ClaimWorkItemByEventInput) (*connector_workqueue.WorkItem, bool, error) {
	return f.inputs.ClaimStaleWorkItemByEvent(ctx, in)
}

func (f *completeWorkItemServiceFake) ResolveClaimableWorkItemGeneration(ctx context.Context, in connector_workqueue.ResolveClaimableGenerationInput) (int64, bool, error) {
	return f.inputs.ResolveClaimableWorkItemGeneration(ctx, in)
}

func (f *completeWorkItemServiceFake) LoadEventForWorkItem(ctx context.Context, eventID string) (*connector_events.ConnectorEvent, error) {
	return f.inputs.LoadEventForWorkItem(ctx, eventID)
}

func (f *completeWorkItemServiceFake) LoadEventsForWorkItems(ctx context.Context, eventIDs []string) (map[string]*connector_events.ConnectorEvent, error) {
	return f.inputs.LoadEventsForWorkItems(ctx, eventIDs)
}

func (f *completeWorkItemServiceFake) LoadSourceProjectionPayloads(ctx context.Context, eventID string) ([]*connector_events.ConnectorEvent, error) {
	return f.inputs.LoadSourceProjectionPayloads(ctx, eventID)
}

func (f *completeWorkItemServiceFake) FailWorkItem(ctx context.Context, in connector_workqueue.FailWorkItemInput) (*connector_workqueue.FailWorkItemResult, bool, error) {
	return f.claims.FailWorkItem(ctx, in)
}

func (f *completeWorkItemServiceFake) CompleteWorkItem(ctx context.Context, workID, leaseToken, terminalState string, now time.Time) (bool, error) {
	return f.claims.CompleteWorkItem(ctx, workID, leaseToken, terminalState, now)
}

func (f *completeWorkItemServiceFake) RetryWorkItem(ctx context.Context, workID, leaseToken string, runAfter, now time.Time) (bool, error) {
	return f.claims.RetryWorkItem(ctx, workID, leaseToken, runAfter, now)
}

func (f *completeWorkItemServiceFake) RenewWorkItemLease(ctx context.Context, workID, leaseToken string, leaseUntil, now time.Time) (bool, error) {
	return f.claims.RenewWorkItemLease(ctx, workID, leaseToken, leaseUntil, now)
}

func (f *completeWorkItemServiceFake) ClaimNextCoalescedExecutorWorkItemSibling(ctx context.Context, currentEventID, currentWorkID, currentLeaseToken, workerID, ownerID, fencingToken, nextLeaseToken string, generation int64, leaseUntil, now time.Time) (*connector_events.WorkItemCoalescedSiblingClaim, bool, error) {
	if f.bridge == nil {
		return nil, false, nil
	}
	claim, ok, err := f.bridge.ClaimNextCoalescedExecutorWorkItemSibling(ctx, currentEventID, currentWorkID, currentLeaseToken, workerID, ownerID, fencingToken, nextLeaseToken, generation, leaseUntil, now)
	if claim != nil && f.coalescedClaimMutator != nil {
		f.coalescedClaimMutator(claim)
	}
	return claim, ok, err
}

func (f *completeWorkItemServiceFake) ClaimNextCoalescedExecutorWorkItemSiblingSourceScoped(ctx context.Context, currentEventID, currentWorkID, currentLeaseToken, workerID, ownerID, fencingToken, nextLeaseToken string, generation int64, leaseUntil, now time.Time) (*connector_events.WorkItemCoalescedSiblingClaim, bool, error) {
	if f.bridge == nil {
		return nil, false, nil
	}
	claim, ok, err := f.bridge.ClaimNextCoalescedExecutorWorkItemSiblingSourceScoped(ctx, currentEventID, currentWorkID, currentLeaseToken, workerID, ownerID, fencingToken, nextLeaseToken, generation, leaseUntil, now)
	if claim != nil && f.coalescedClaimMutator != nil {
		f.coalescedClaimMutator(claim)
	}
	return claim, ok, err
}

func mandatoryWorkerTestConfig() WorkerConfig {
	return WorkerConfig{
		Namespace:                     workerTestNS,
		PackInstallID:                 "pack-test",
		WorkerID:                      "worker-test",
		WorkItemSchedulerOwnerID:      "scheduler-owner",
		WorkItemSchedulerFencingToken: "scheduler-fence",
		WorkItemLeaseDuration:         time.Minute,
		FinalizeTimeout:               time.Second,
	}
}

func newMandatoryWorkerForTest(t *testing.T, cfg WorkerConfig, service WorkItemService) *Worker {
	t.Helper()
	worker, err := NewWorkItemWorker(context.Background(),
		cfg,
		service,
		&fakeSyncState{},
		fakeBatchRead{},
		&fakeFlowInvoker{},
		&stubRouteResolver{},
	)
	if err != nil {
		t.Fatalf("NewWorkItemWorker: %v", err)
	}
	return worker
}

func newMandatoryWorkerWithDependenciesForTest(
	t *testing.T,
	cfg WorkerConfig,
	service WorkItemService,
	syncstate SyncStateService,
	batchRead BatchReadClient,
	flow FlowInvoker,
	resolver connector_inbound_routing.RouteResolver,
) *Worker {
	t.Helper()
	worker, err := NewWorkItemWorkerForTest(context.Background(), cfg, service, syncstate, batchRead, flow, resolver)
	if err != nil {
		t.Fatalf("NewWorkItemWorkerForTest: %v", err)
	}
	return worker
}

func TestNewWorkItemWorkerRejectsIncompleteServiceAndSplitConfig(t *testing.T) {
	valid := newCompleteWorkItemServiceFake(&fakeWorkItemClaims{}, &fakeWorkItemEventLoader{}, nil)

	t.Run("typed nil", func(t *testing.T) {
		var service *completeWorkItemServiceFake
		_, err := NewWorkItemWorker(context.Background(), mandatoryWorkerTestConfig(), service, &fakeSyncState{}, fakeBatchRead{}, &fakeFlowInvoker{}, &stubRouteResolver{})
		if !errors.Is(err, ErrInvalidWorkerConfig) {
			t.Fatalf("error = %v, want ErrInvalidWorkerConfig", err)
		}
	})

	t.Run("nil activation context", func(t *testing.T) {
		_, err := NewWorkItemWorker(nil, mandatoryWorkerTestConfig(), valid, &fakeSyncState{}, fakeBatchRead{}, &fakeFlowInvoker{}, &stubRouteResolver{})
		if !errors.Is(err, ErrInvalidWorkerConfig) {
			t.Fatalf("error = %v, want ErrInvalidWorkerConfig", err)
		}
	})

	t.Run("claim scope not honored", func(t *testing.T) {
		service := newCompleteWorkItemServiceFake(&fakeWorkItemClaims{}, &fakeWorkItemEventLoader{}, nil)
		service.inputs.(*fakeWorkItemClaimInputs).honors = false
		_, err := NewWorkItemWorker(context.Background(), mandatoryWorkerTestConfig(), service, &fakeSyncState{}, fakeBatchRead{}, &fakeFlowInvoker{}, &stubRouteResolver{})
		if !errors.Is(err, ErrInvalidWorkerConfig) {
			t.Fatalf("error = %v, want ErrInvalidWorkerConfig", err)
		}
	})

	t.Run("missing cutover verifier", func(t *testing.T) {
		service := workItemServiceWithoutCutover{WorkItemService: valid}
		if _, err := NewWorkItemWorker(context.Background(), mandatoryWorkerTestConfig(), service, &fakeSyncState{}, fakeBatchRead{}, &fakeFlowInvoker{}, &stubRouteResolver{}); !errors.Is(err, ErrInvalidWorkerConfig) {
			t.Fatalf("NewWorkItemWorker error = %v, want ErrInvalidWorkerConfig", err)
		}
	})

	t.Run("cancelled activation context reaches verifier", func(t *testing.T) {
		ctx, cancel := context.WithCancel(context.Background())
		cancel()
		service := newCompleteWorkItemServiceFake(&fakeWorkItemClaims{}, &fakeWorkItemEventLoader{}, nil)
		service.cutoverErr = context.Canceled
		if _, err := NewWorkItemWorker(ctx, mandatoryWorkerTestConfig(), service, &fakeSyncState{}, fakeBatchRead{}, &fakeFlowInvoker{}, &stubRouteResolver{}); !errors.Is(err, context.Canceled) {
			t.Fatalf("NewWorkItemWorker error = %v, want context.Canceled", err)
		}
	})

	t.Run("split or legacy dependency", func(t *testing.T) {
		for _, tc := range []struct {
			name   string
			mutate func(*WorkerConfig)
		}{
			{
				name: "legacy custom claim authority",
				mutate: func(cfg *WorkerConfig) {
					cfg.WorkItemClaims = &fakeWorkItemClaims{legacyLeaseAuthority: true}
				},
			},
			{
				name: "legacy Event loader",
				mutate: func(cfg *WorkerConfig) {
					cfg.WorkItemEventLoader = &fakeWorkItemEventLoader{}
				},
			},
			{
				name: "legacy Event bridge",
				mutate: func(cfg *WorkerConfig) {
					cfg.WorkItemEventBridge = &fakeWorkItemEventBridge{}
				},
			},
		} {
			t.Run(tc.name, func(t *testing.T) {
				cfg := mandatoryWorkerTestConfig()
				tc.mutate(&cfg)
				_, err := NewWorkItemWorker(context.Background(), cfg, valid, &fakeSyncState{}, fakeBatchRead{}, &fakeFlowInvoker{}, &stubRouteResolver{})
				if !errors.Is(err, ErrInvalidWorkerConfig) {
					t.Fatalf("error = %v, want ErrInvalidWorkerConfig", err)
				}
			})
		}
	})

	t.Run("legacy claim path", func(t *testing.T) {
		cfg := mandatoryWorkerTestConfig()
		cfg.ClaimPath = ClaimPathLegacyEvent
		_, err := NewWorkItemWorker(context.Background(), cfg, valid, &fakeSyncState{}, fakeBatchRead{}, &fakeFlowInvoker{}, &stubRouteResolver{})
		if !errors.Is(err, ErrInvalidWorkerConfig) {
			t.Fatalf("error = %v, want ErrInvalidWorkerConfig", err)
		}
	})

	t.Run("missing scheduler owner", func(t *testing.T) {
		cfg := mandatoryWorkerTestConfig()
		cfg.WorkItemSchedulerOwnerID = ""
		_, err := NewWorkItemWorker(context.Background(), cfg, valid, &fakeSyncState{}, fakeBatchRead{}, &fakeFlowInvoker{}, &stubRouteResolver{})
		if !errors.Is(err, ErrInvalidWorkerConfig) {
			t.Fatalf("error = %v, want ErrInvalidWorkerConfig", err)
		}
	})

	t.Run("missing scheduler fence", func(t *testing.T) {
		cfg := mandatoryWorkerTestConfig()
		cfg.WorkItemSchedulerFencingToken = ""
		_, err := NewWorkItemWorker(context.Background(), cfg, valid, &fakeSyncState{}, fakeBatchRead{}, &fakeFlowInvoker{}, &stubRouteResolver{})
		if !errors.Is(err, ErrInvalidWorkerConfig) {
			t.Fatalf("error = %v, want ErrInvalidWorkerConfig", err)
		}
	})

	t.Run("negative WorkItem lease", func(t *testing.T) {
		cfg := mandatoryWorkerTestConfig()
		cfg.WorkItemLeaseDuration = -time.Second
		_, err := NewWorkItemWorker(context.Background(), cfg, valid, &fakeSyncState{}, fakeBatchRead{}, &fakeFlowInvoker{}, &stubRouteResolver{})
		if !errors.Is(err, ErrInvalidWorkerConfig) {
			t.Fatalf("error = %v, want ErrInvalidWorkerConfig", err)
		}
	})

	for _, tc := range []struct {
		name   string
		mutate func(*WorkerConfig)
	}{
		{name: "empty namespace", mutate: func(cfg *WorkerConfig) { cfg.Namespace = "" }},
		{name: "empty pack install", mutate: func(cfg *WorkerConfig) { cfg.PackInstallID = "" }},
		{name: "empty worker", mutate: func(cfg *WorkerConfig) { cfg.WorkerID = "" }},
		{name: "negative finalize timeout", mutate: func(cfg *WorkerConfig) { cfg.FinalizeTimeout = -time.Second }},
	} {
		t.Run(tc.name, func(t *testing.T) {
			cfg := mandatoryWorkerTestConfig()
			tc.mutate(&cfg)
			_, err := NewWorkItemWorker(context.Background(), cfg, valid, &fakeSyncState{}, fakeBatchRead{}, &fakeFlowInvoker{}, &stubRouteResolver{})
			if !errors.Is(err, ErrInvalidWorkerConfig) {
				t.Fatalf("error = %v, want ErrInvalidWorkerConfig", err)
			}
		})
	}
}

func TestNewWorkItemWorkerUsesOneLeaseFreeService(t *testing.T) {
	now := time.Now().UTC()
	claims := &fakeWorkItemClaims{work: &connector_workqueue.WorkItem{
		WorkID: "work-1", EventID: "event-1", WorkType: connector_workqueue.WorkTypeConnectorEvent,
		State: connector_workqueue.WorkStateProcessing, LeaseToken: "lease-1", Generation: 4, LastModifiedDate: now,
	}}
	loader := &fakeWorkItemEventLoader{event: &connector_events.ConnectorEvent{ID: "event-1", Status: connector_workqueue.WorkStateDead}}
	service := newCompleteWorkItemServiceFake(claims, loader, nil)
	worker := newMandatoryWorkerForTest(t, mandatoryWorkerTestConfig(), service)

	adapter, ok := worker.events.(*workItemEventsService)
	if !ok || adapter.workItems != service || adapter.claims != nil || adapter.bridge != nil {
		t.Fatalf("mandatory adapter = %+v, want one complete service and no split dependencies", adapter)
	}
	event, claimed, err := worker.events.ClaimNext(context.Background(), "pack-test", "worker-test")
	if err != nil || !claimed {
		t.Fatalf("ClaimNext = (%+v, %v, %v), want WorkItem claim", event, claimed, err)
	}
	if event.Status != connector_workqueue.WorkStateProcessing || event.ClaimToken != "lease-1" {
		t.Fatalf("claimed Event = %+v, want in-memory WorkItem execution fields", event)
	}
	if claims.acquireCalls != 0 || claims.renewCalls != 0 {
		t.Fatalf("SchedulerLease acquire/renew calls = %d/%d, want none", claims.acquireCalls, claims.renewCalls)
	}
	if !claims.lastDistributed {
		t.Fatal("mandatory claim context did not carry distributed WorkItem authority")
	}
}

func TestNewWorkItemWorkerIgnoresStaleEventLifecycleForWorkItemExecution(t *testing.T) {
	now := time.Now().UTC()
	staleAt := now.Add(-time.Hour)
	futureRetryAt := now.Add(time.Hour)
	newHostileEvent := func(id string) *connector_events.ConnectorEvent {
		return &connector_events.ConnectorEvent{
			ID:               id,
			Status:           connector_workqueue.WorkStateDead,
			RetryCount:       connector_events.MaxRetryAttempts,
			LastError:        "stale legacy failure",
			LastErrorAt:      &staleAt,
			NextAttemptAt:    &futureRetryAt,
			ClaimedBy:        "legacy-worker",
			ClaimedAt:        &staleAt,
			ClaimToken:       "legacy-lease",
			ProcessedAt:      &staleAt,
			LastModifiedDate: staleAt,
		}
	}
	claims := &fakeWorkItemClaims{works: []*connector_workqueue.WorkItem{
		{WorkID: "work-fail", EventID: "event-fail", WorkType: connector_workqueue.WorkTypeConnectorEvent, State: connector_workqueue.WorkStateProcessing, LeaseToken: "lease-fail", Generation: 4, LastModifiedDate: now},
		{WorkID: "work-retry", EventID: "event-retry", WorkType: connector_workqueue.WorkTypeConnectorEvent, State: connector_workqueue.WorkStateProcessing, LeaseToken: "lease-retry", Generation: 4, LastModifiedDate: now},
		{WorkID: "work-terminal", EventID: "event-terminal", WorkType: connector_workqueue.WorkTypeConnectorEvent, State: connector_workqueue.WorkStateProcessing, LeaseToken: "lease-terminal", Generation: 4, LastModifiedDate: now},
		{WorkID: "work-renew", EventID: "event-renew", WorkType: connector_workqueue.WorkTypeConnectorEvent, State: connector_workqueue.WorkStateProcessing, LeaseToken: "lease-renew", Generation: 4, LastModifiedDate: now},
	}}
	loader := &fakeWorkItemEventLoader{events: map[string]*connector_events.ConnectorEvent{
		"event-fail":     newHostileEvent("event-fail"),
		"event-retry":    newHostileEvent("event-retry"),
		"event-terminal": newHostileEvent("event-terminal"),
		"event-renew":    newHostileEvent("event-renew"),
	}}
	service := newCompleteWorkItemServiceFake(claims, loader, nil)
	worker := newMandatoryWorkerForTest(t, mandatoryWorkerTestConfig(), service)

	adapter, ok := worker.events.(*workItemEventsService)
	if !ok || adapter.workItems != service || adapter.claims != nil || adapter.loader != nil || adapter.bridge != nil {
		t.Fatalf("mandatory adapter = %+v, want only complete WorkItem service", adapter)
	}
	claimHostileEvent := func(wantEventID, wantLeaseToken string) *connector_events.ConnectorEvent {
		t.Helper()
		event, claimed, err := worker.events.ClaimNext(context.Background(), "pack-test", "worker-test")
		if err != nil || !claimed {
			t.Fatalf("ClaimNext = (%+v, %v, %v), want WorkItem claim", event, claimed, err)
		}
		if event.ID != wantEventID || event.Status != connector_workqueue.WorkStateProcessing || event.ClaimedBy != "worker-test" || event.ClaimToken != wantLeaseToken {
			t.Fatalf("claimed Event = %+v, want WorkItem execution projection", event)
		}
		if event.RetryCount != 0 || event.LastError != "" || event.LastErrorAt != nil || event.NextAttemptAt != nil || event.ProcessedAt != nil {
			t.Fatalf("claimed Event retained stale lifecycle fields: %+v", event)
		}
		if event.ClaimedAt == nil || !event.ClaimedAt.Equal(now) || !event.LastModifiedDate.Equal(now) {
			t.Fatalf("claimed Event timestamps = %+v, want WorkItem claim time", event)
		}
		return event
	}

	failure := claimHostileEvent("event-fail", "lease-fail")
	if err := worker.events.MarkFailed(context.Background(), failure.ID, failure.ClaimToken, "provider failure"); err != nil {
		t.Fatalf("MarkFailed: %v", err)
	}
	retry := claimHostileEvent("event-retry", "lease-retry")
	if err := adapter.ReleaseClaimForRetry(context.Background(), retry.ID, retry.ClaimToken, "transient"); err != nil {
		t.Fatalf("ReleaseClaimForRetry: %v", err)
	}
	terminal := claimHostileEvent("event-terminal", "lease-terminal")
	if err := worker.events.MarkProcessed(context.Background(), terminal.ID, terminal.ClaimToken); err != nil {
		t.Fatalf("MarkProcessed: %v", err)
	}
	renew := claimHostileEvent("event-renew", "lease-renew")
	if err := adapter.RenewLease(context.Background(), renew.ID, renew.ClaimToken); err != nil {
		t.Fatalf("RenewLease: %v", err)
	}

	if len(claims.failInputs) != 1 || claims.failInputs[0].WorkID != "work-fail" || claims.failInputs[0].EventID != "event-fail" || claims.failInputs[0].LeaseToken != "lease-fail" {
		t.Fatalf("FailWorkItem inputs = %+v, want WorkItem failure CAS", claims.failInputs)
	}
	if claims.retryCalls != 1 || len(claims.retryWorkIDs) != 1 || claims.retryWorkIDs[0] != "work-retry" || claims.retryWorkTokens[0] != "lease-retry" {
		t.Fatalf("RetryWorkItem calls = %d %v %v, want one WorkItem retry CAS", claims.retryCalls, claims.retryWorkIDs, claims.retryWorkTokens)
	}
	if len(claims.completeCalls) != 1 || claims.completeCalls[0] != connector_workqueue.WorkStateProcessed || len(claims.completeWorkIDs) != 1 || claims.completeWorkIDs[0] != "work-terminal" {
		t.Fatalf("CompleteWorkItem calls = %v %v, want one WorkItem terminal CAS", claims.completeCalls, claims.completeWorkIDs)
	}
	if claims.renewWorkCalls != 1 || len(claims.renewWorkIDs) != 1 || claims.renewWorkIDs[0] != "work-renew" || claims.renewWorkTokens[0] != "lease-renew" {
		t.Fatalf("RenewWorkItemLease calls = %d %v %v, want one WorkItem lease CAS", claims.renewWorkCalls, claims.renewWorkIDs, claims.renewWorkTokens)
	}
	if claims.acquireCalls != 0 || claims.renewCalls != 0 {
		t.Fatalf("SchedulerLease calls = %d/%d, want none", claims.acquireCalls, claims.renewCalls)
	}
}

func TestMandatoryWorkItemMutationsUseCASAndFreshFollowupContexts(t *testing.T) {
	claims := &fakeWorkItemClaims{}
	service := newCompleteWorkItemServiceFake(claims, &fakeWorkItemEventLoader{}, nil)
	adapter := newMandatoryWorkItemEventsService(service, mandatoryWorkerTestConfig())
	adapter.claimed["event-fail"] = claimedWorkItemRef{workID: "work-fail", leaseToken: "lease-fail"}
	adapter.claimed["event-release"] = claimedWorkItemRef{workID: "work-release", leaseToken: "lease-release"}
	adapter.claimed["event-terminal"] = claimedWorkItemRef{workID: "csp:projection", leaseToken: "lease-terminal"}
	adapter.claimed["event-renew"] = claimedWorkItemRef{workID: "work-renew", leaseToken: "lease-renew"}

	canceled, cancel := context.WithCancel(context.Background())
	cancel()
	if err := adapter.MarkFailed(canceled, "event-fail", "lease-fail", "provider failure"); err != nil {
		t.Fatalf("MarkFailed: %v", err)
	}
	if len(claims.failInputs) != 1 || len(claims.failCtxErrs) != 1 || claims.failCtxErrs[0] != nil {
		t.Fatalf("FailWorkItem inputs/context errs = %v/%v", claims.failInputs, claims.failCtxErrs)
	}
	failure := claims.failInputs[0]
	if failure.WorkID != "work-fail" || failure.EventID != "event-fail" || failure.RetryThreshold != connector_events.MaxRetryAttempts || len(failure.RetryDelays) != connector_events.MaxRetryAttempts {
		t.Fatalf("FailWorkItem input = %+v", failure)
	}
	if claims.retryCalls != 0 || len(claims.completeCalls) != 0 {
		t.Fatalf("failure also called retry/complete = %d/%v", claims.retryCalls, claims.completeCalls)
	}

	if err := adapter.ReleaseClaimForRetry(canceled, "event-release", "lease-release", "transient"); err != nil {
		t.Fatalf("ReleaseClaimForRetry: %v", err)
	}
	if claims.retryCalls != 1 || len(claims.failInputs) != 1 || claims.retryCtxErrs[0] != nil {
		t.Fatalf("retry/failure/context = %d/%d/%v", claims.retryCalls, len(claims.failInputs), claims.retryCtxErrs)
	}

	if next, ok, err := adapter.ClaimNextCoalescedSibling(context.Background(), "event-terminal", "lease-terminal", "worker-test"); err != nil || ok || next != nil {
		t.Fatalf("source projection coalescing = (%+v, %v, %v), want dedicated terminal handoff", next, ok, err)
	}
	if err := adapter.MarkProcessed(canceled, "event-terminal", "lease-terminal"); err != nil {
		t.Fatalf("MarkProcessed source projection: %v", err)
	}
	if len(claims.completeCalls) != 1 || claims.completeCalls[0] != connector_workqueue.WorkStateProcessed || claims.completeCtxErrs[0] != nil {
		t.Fatalf("complete calls/context = %v/%v", claims.completeCalls, claims.completeCtxErrs)
	}

	if err := adapter.RenewLease(context.Background(), "event-renew", "lease-renew"); err != nil {
		t.Fatalf("RenewLease: %v", err)
	}
	if claims.renewWorkCalls != 1 || claims.renewWorkIDs[0] != "work-renew" || claims.renewWorkTokens[0] != "lease-renew" || !claims.renewWorkLeaseUntils[0].After(time.Now()) {
		t.Fatalf("renew CAS calls = %d %v %v %v", claims.renewWorkCalls, claims.renewWorkIDs, claims.renewWorkTokens, claims.renewWorkLeaseUntils)
	}
}

func TestMandatoryWorkItemServicePerformsOrdinaryAtomicCoalescing(t *testing.T) {
	now := time.Now().UTC()
	claims := &fakeWorkItemClaims{}
	bridge := &fakeWorkItemEventBridge{coalescedClaim: &connector_events.WorkItemCoalescedSiblingClaim{
		Work: &connector_workqueue.WorkItem{
			WorkID: "work-next", EventID: "event-next", WorkType: connector_workqueue.WorkTypeConnectorEvent,
			State: connector_workqueue.WorkStateProcessing, Generation: 7, LastModifiedDate: now,
		},
		Event: &connector_events.ConnectorEvent{ID: "event-next", Status: connector_workqueue.WorkStateDead},
	}}
	service := newCompleteWorkItemServiceFake(claims, &fakeWorkItemEventLoader{}, bridge)
	adapter := newMandatoryWorkItemEventsService(service, mandatoryWorkerTestConfig())
	adapter.claimed["event-current"] = claimedWorkItemRef{workID: "work-current", leaseToken: "lease-current", generation: 7}

	next, ok, err := adapter.ClaimNextCoalescedSibling(context.Background(), "event-current", "lease-current", "worker-test")
	if err != nil || !ok {
		t.Fatalf("ClaimNextCoalescedSibling = (%+v, %v, %v)", next, ok, err)
	}
	if len(bridge.coalesceCalls) != 1 || bridge.coalesceCalls[0] != "event-current:work-current:lease-current" {
		t.Fatalf("atomic coalescing calls = %v", bridge.coalesceCalls)
	}
	if next.ID != "event-next" || next.Status != connector_workqueue.WorkStateProcessing || next.ClaimToken == "" {
		t.Fatalf("coalesced Event = %+v", next)
	}
	if _, err := adapter.claimedRef("event-current", "lease-current"); !errors.Is(err, connector_events.ErrLeaseLost) {
		t.Fatalf("current claim retained after handoff: %v", err)
	}
	if _, err := adapter.claimedRef("event-next", next.ClaimToken); err != nil {
		t.Fatalf("successor claim not remembered: %v", err)
	}
}

func TestMandatoryWorkItemServiceRejectsMalformedCoalescedSuccessor(t *testing.T) {
	tests := []struct {
		name              string
		wantReleaseWorkID string
		mutate            func(*connector_events.WorkItemCoalescedSiblingClaim)
	}{
		{
			name:              "unsupported WorkItem type",
			wantReleaseWorkID: "work-next",
			mutate: func(claim *connector_events.WorkItemCoalescedSiblingClaim) {
				claim.Work.WorkType = "unsupported"
			},
		},
		{
			name:              "Event and WorkItem identity mismatch",
			wantReleaseWorkID: "work-next",
			mutate: func(claim *connector_events.WorkItemCoalescedSiblingClaim) {
				claim.Work.EventID = "event-other"
			},
		},
		{
			name:              "current Event replayed under successor WorkItem",
			wantReleaseWorkID: "work-next",
			mutate: func(claim *connector_events.WorkItemCoalescedSiblingClaim) {
				claim.Event.ID = "event-current"
				claim.Work.EventID = "event-current"
			},
		},
		{
			name:              "source projection returned as ordinary successor",
			wantReleaseWorkID: connector_workqueue.SourceProjectionWorkIDPrefix + "next",
			mutate: func(claim *connector_events.WorkItemCoalescedSiblingClaim) {
				claim.Work.WorkID = connector_workqueue.SourceProjectionWorkIDPrefix + "next"
			},
		},
	}

	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			now := time.Now().UTC()
			claims := &fakeWorkItemClaims{}
			bridge := &fakeWorkItemEventBridge{coalescedClaim: &connector_events.WorkItemCoalescedSiblingClaim{
				Work: &connector_workqueue.WorkItem{
					WorkID: "work-next", EventID: "event-next", WorkType: connector_workqueue.WorkTypeConnectorEvent,
					State: connector_workqueue.WorkStateProcessing, Generation: 7, LastModifiedDate: now,
				},
				Event: &connector_events.ConnectorEvent{ID: "event-next", Status: connector_workqueue.WorkStateDead},
			}}
			service := newCompleteWorkItemServiceFake(claims, &fakeWorkItemEventLoader{}, bridge)
			var generatedLeaseToken string
			var returnedEventID string
			service.coalescedClaimMutator = func(claim *connector_events.WorkItemCoalescedSiblingClaim) {
				generatedLeaseToken = claim.Work.LeaseToken
				tt.mutate(claim)
				returnedEventID = claim.Event.ID
			}
			adapter := newMandatoryWorkItemEventsService(service, mandatoryWorkerTestConfig())
			adapter.claimed["event-current"] = claimedWorkItemRef{workID: "work-current", leaseToken: "lease-current", generation: 7}

			callerCtx, cancel := context.WithCancel(context.Background())
			cancel()
			next, ok, err := adapter.ClaimNextCoalescedSibling(callerCtx, "event-current", "lease-current", "worker-test")
			if err == nil || ok || next != nil {
				t.Fatalf("ClaimNextCoalescedSibling = (%+v, %v, %v), want nil, false, validation error", next, ok, err)
			}
			if generatedLeaseToken == "" {
				t.Fatal("coalesced handoff did not generate a successor lease token")
			}
			if claims.retryCalls != 1 || len(claims.retryWorkIDs) != 1 || claims.retryWorkIDs[0] != tt.wantReleaseWorkID ||
				len(claims.retryWorkTokens) != 1 || claims.retryWorkTokens[0] != generatedLeaseToken {
				t.Fatalf("invalid successor release = calls %d ids %v tokens %v, want one CAS with generated token", claims.retryCalls, claims.retryWorkIDs, claims.retryWorkTokens)
			}
			if len(claims.retryCtxErrs) != 1 || claims.retryCtxErrs[0] != nil {
				t.Fatalf("invalid successor release context errors = %v, want fresh live context", claims.retryCtxErrs)
			}
			if len(claims.retryRunAfters) != 1 || !claims.retryRunAfters[0].After(now) {
				t.Fatalf("invalid successor retry time = %v, want bounded backoff after %v", claims.retryRunAfters, now)
			}
			if _, err := adapter.claimedRef("event-current", "lease-current"); !errors.Is(err, connector_events.ErrLeaseLost) {
				t.Fatalf("consumed current claim retained after invalid handoff: %v", err)
			}
			if _, err := adapter.claimedRef(returnedEventID, generatedLeaseToken); !errors.Is(err, connector_events.ErrLeaseLost) {
				t.Fatalf("invalid successor claim remembered: %v", err)
			}
		})
	}
}

func (*completeWorkItemServiceFake) RequireSourceProjectionProtocol(context.Context, string) error {
	return nil
}