package connector_inbound

import (
	"context"
	"database/sql"
	"encoding/json"
	"sync"
	"testing"
	"time"

	connector_inbound_routing "github.com/pingcap-inc/nexuscrm-connectors/features/connector/inboundrouting"

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

// renewRecordingEvents wraps a fakeEventsService and records every RenewLease
// call (eventID + token). It exists so the heartbeat tests can assert the worker
// renews an in-flight lease while a slow flow runs, without a live DB. Because it
// embeds *fakeEventsService it satisfies the full EventsService surface AND the
// optional claim-releasing / stale-reclaim capabilities, plus the new
// leaseRenewingEventsService capability the worker type-asserts for.
type renewRecordingEvents struct {
	*fakeEventsService

	mu         sync.Mutex
	renewCalls []renewCall
	renewErr   error // returned by RenewLease when set (e.g. ErrLeaseLost)
}

type renewCall struct {
	eventID    string
	claimToken string
}

func (r *renewRecordingEvents) RenewLease(_ context.Context, eventID, claimToken string) error {
	r.mu.Lock()
	defer r.mu.Unlock()
	r.renewCalls = append(r.renewCalls, renewCall{eventID: eventID, claimToken: claimToken})
	return r.renewErr
}

func (r *renewRecordingEvents) renewCount() int {
	r.mu.Lock()
	defer r.mu.Unlock()
	return len(r.renewCalls)
}

func (r *renewRecordingEvents) renewedTokens() []renewCall {
	r.mu.Lock()
	defer r.mu.Unlock()
	out := make([]renewCall, len(r.renewCalls))
	copy(out, r.renewCalls)
	return out
}

// renewRecordingWorkItemService records the mandatory WorkItem renewal path.
// It keeps the two live-executor heartbeat tests independent from the retained
// legacy Event compatibility fixture below.
type renewRecordingWorkItemService struct {
	*completeWorkItemServiceFake

	mu              sync.Mutex
	renewCalls      []renewWorkItemCall
	completeCalls   []completeWorkItemCall
	bindClaimToRow  bool
	claimInput      connector_workqueue.ClaimWorkItemInput
	claimedWorkItem *connector_workqueue.WorkItem
}

type renewWorkItemCall struct {
	workID     string
	leaseToken string
}

type completeWorkItemCall struct {
	workID        string
	leaseToken    string
	terminalState string
}

// panicHeartbeatWorkItemService adds renewal recording to the stateful
// WorkItem row-CAS fake used by the panic-recovery tests. The embedded service
// materializes each claim from the worker input and performs the failure
// transition against that active lease; this wrapper lets the heartbeat test
// prove its read-window renewer has stopped after the recovered panic.
type panicHeartbeatWorkItemService struct {
	*panicDeadLetterWorkItemService

	renewCalls []panicHeartbeatRenewal
	renewed    chan struct{}
	renewOnce  sync.Once
}

var _ WorkItemService = (*panicHeartbeatWorkItemService)(nil)

type panicHeartbeatRenewal struct {
	workID          string
	leaseToken      string
	rowLeaseToken   string
	claimLeaseToken string
	state           string
	leaseUntil      time.Time
	now             time.Time
	accepted        bool
}

func (s *panicHeartbeatWorkItemService) RenewWorkItemLease(_ context.Context, workID, leaseToken string, leaseUntil, now time.Time) (bool, error) {
	s.panicDeadLetterWorkItemService.mu.Lock()
	defer s.panicDeadLetterWorkItemService.mu.Unlock()

	renewal := panicHeartbeatRenewal{
		workID: workID, leaseToken: leaseToken, leaseUntil: leaseUntil, now: now,
	}
	for _, row := range s.rows {
		if row.work.WorkID != workID {
			continue
		}
		renewal.state = row.work.State
		renewal.rowLeaseToken = row.work.LeaseToken
		if len(row.claims) > 0 {
			renewal.claimLeaseToken = row.claims[len(row.claims)-1].work.LeaseToken
		}
		renewal.accepted = row.work.State == connector_workqueue.WorkStateProcessing &&
			renewal.rowLeaseToken != "" &&
			renewal.rowLeaseToken == renewal.claimLeaseToken &&
			renewal.rowLeaseToken == leaseToken &&
			!leaseUntil.IsZero() &&
			leaseUntil.After(now)
		if renewal.accepted {
			row.work.LeaseUntil = sql.NullTime{Time: leaseUntil, Valid: true}
			row.work.LastModifiedDate = now
		}
		break
	}
	s.renewCalls = append(s.renewCalls, renewal)
	s.renewOnce.Do(func() { close(s.renewed) })
	return renewal.accepted, nil
}

func (s *panicHeartbeatWorkItemService) renewCount() int {
	s.panicDeadLetterWorkItemService.mu.Lock()
	defer s.panicDeadLetterWorkItemService.mu.Unlock()
	return len(s.renewCalls)
}

func (s *panicHeartbeatWorkItemService) renewals() []panicHeartbeatRenewal {
	s.panicDeadLetterWorkItemService.mu.Lock()
	defer s.panicDeadLetterWorkItemService.mu.Unlock()
	out := make([]panicHeartbeatRenewal, len(s.renewCalls))
	copy(out, s.renewCalls)
	return out
}

// panicAfterWorkItemRenewalBatchRead deliberately delegates the induced panic
// to the native panickingBatchRead fixture only after the mandatory WorkItem
// heartbeat has completed one row-CAS renewal. This makes the no-leak assertion
// exercise a live renewer rather than a heartbeat goroutine that never ticked.
type panicAfterWorkItemRenewalBatchRead struct {
	renewed <-chan struct{}
}

func (b panicAfterWorkItemRenewalBatchRead) BatchRead(ctx context.Context, namespace, objectKind string, sourceIDs []string) (map[string]json.RawMessage, error) {
	select {
	case <-b.renewed:
		return panickingBatchRead{}.BatchRead(ctx, namespace, objectKind, sourceIDs)
	case <-ctx.Done():
		return nil, ctx.Err()
	}
}

// bindClaimedWorkItem makes the test service materialize the row returned from
// ClaimNextWorkItem with the worker-generated row-CAS claim input. This keeps
// deadline-finalization tests honest about the lease they must return.
func (r *renewRecordingWorkItemService) bindClaimedWorkItem() {
	r.mu.Lock()
	defer r.mu.Unlock()
	r.bindClaimToRow = true
}

func (r *renewRecordingWorkItemService) ClaimNextWorkItem(ctx context.Context, in connector_workqueue.ClaimWorkItemInput) (*connector_workqueue.WorkItem, bool, error) {
	work, claimed, err := r.completeWorkItemServiceFake.ClaimNextWorkItem(ctx, in)
	if err != nil || !claimed || work == nil {
		return work, claimed, err
	}
	r.mu.Lock()
	bind := r.bindClaimToRow
	r.mu.Unlock()
	if !bind {
		return work, true, nil
	}
	claimedWorkItem := *work
	claimedWorkItem.InstalledPackID = in.InstalledPackID
	claimedWorkItem.Generation = in.Generation
	claimedWorkItem.State = connector_workqueue.WorkStateProcessing
	claimedWorkItem.LeaseOwner = in.OwnerID
	claimedWorkItem.LeaseToken = in.LeaseToken
	claimedWorkItem.LeaseUntil = sql.NullTime{Time: in.LeaseUntil, Valid: !in.LeaseUntil.IsZero()}
	r.mu.Lock()
	r.claimInput = in
	r.claimedWorkItem = &claimedWorkItem
	r.mu.Unlock()
	return &claimedWorkItem, true, nil
}

func (r *renewRecordingWorkItemService) claimedWork() (connector_workqueue.ClaimWorkItemInput, connector_workqueue.WorkItem, bool) {
	r.mu.Lock()
	defer r.mu.Unlock()
	if r.claimedWorkItem == nil {
		return connector_workqueue.ClaimWorkItemInput{}, connector_workqueue.WorkItem{}, false
	}
	return r.claimInput, *r.claimedWorkItem, true
}

func (r *renewRecordingWorkItemService) RenewWorkItemLease(_ context.Context, workID, leaseToken string, _, _ time.Time) (bool, error) {
	r.mu.Lock()
	defer r.mu.Unlock()
	r.renewCalls = append(r.renewCalls, renewWorkItemCall{workID: workID, leaseToken: leaseToken})
	return true, nil
}

func (r *renewRecordingWorkItemService) CompleteWorkItem(ctx context.Context, workID, leaseToken, terminalState string, now time.Time) (bool, error) {
	r.mu.Lock()
	r.completeCalls = append(r.completeCalls, completeWorkItemCall{workID: workID, leaseToken: leaseToken, terminalState: terminalState})
	r.mu.Unlock()
	return r.completeWorkItemServiceFake.CompleteWorkItem(ctx, workID, leaseToken, terminalState, now)
}

func (r *renewRecordingWorkItemService) renewCount() int {
	r.mu.Lock()
	defer r.mu.Unlock()
	return len(r.renewCalls)
}

func (r *renewRecordingWorkItemService) renewedClaims() []renewWorkItemCall {
	r.mu.Lock()
	defer r.mu.Unlock()
	out := make([]renewWorkItemCall, len(r.renewCalls))
	copy(out, r.renewCalls)
	return out
}

func (r *renewRecordingWorkItemService) completedClaims() []completeWorkItemCall {
	r.mu.Lock()
	defer r.mu.Unlock()
	out := make([]completeWorkItemCall, len(r.completeCalls))
	copy(out, r.completeCalls)
	return out
}

func bindHeartbeatClaim(service *renewRecordingWorkItemService, seededWorkItem *connector_workqueue.WorkItem, generation int64) {
	seededWorkItem.LeaseToken = "seed-heartbeat-lease"
	seededWorkItem.LeaseOwner = "seed-heartbeat-owner"
	seededWorkItem.Generation = -1
	service.claims.resolveGeneration = &generation
	service.bindClaimedWorkItem()
}

func assertHeartbeatClaim(t *testing.T, service *renewRecordingWorkItemService, seededWorkItem *connector_workqueue.WorkItem, wantGeneration int64) (connector_workqueue.ClaimWorkItemInput, connector_workqueue.WorkItem) {
	t.Helper()
	claim, claimedWorkItem, ok := service.claimedWork()
	if !ok {
		t.Fatal("heartbeat path did not materialize a claimed WorkItem")
	}
	if claim.InstalledPackID != "pack-test" || claim.Generation != wantGeneration || claim.OwnerID != "scheduler-owner" || claim.FencingToken != "scheduler-fence" {
		t.Fatalf("claim input = %+v, want pack-test/generation %d/scheduler owner and fence", claim, wantGeneration)
	}
	if claim.LeaseToken == "" || claim.LeaseToken == seededWorkItem.LeaseToken || claim.LeaseUntil.IsZero() || !claim.LeaseUntil.After(claim.Now) {
		t.Fatalf("claim lease = %q until %v, want generated token and deadline", claim.LeaseToken, claim.LeaseUntil)
	}
	if claimedWorkItem.State != connector_workqueue.WorkStateProcessing || claimedWorkItem.WorkID != seededWorkItem.WorkID || claimedWorkItem.EventID != seededWorkItem.EventID || claimedWorkItem.LeaseToken != claim.LeaseToken || claimedWorkItem.LeaseOwner != claim.OwnerID || claimedWorkItem.Generation != claim.Generation || !claimedWorkItem.LeaseUntil.Valid || !claimedWorkItem.LeaseUntil.Time.Equal(claim.LeaseUntil) {
		t.Fatalf("claimed WorkItem = %+v, want processing row materialized from claim input %+v", claimedWorkItem, claim)
	}
	return claim, claimedWorkItem
}

func newWorkItemHeartbeatTestWorker(
	t *testing.T,
	ev *connector_events.ConnectorEvent,
	flow FlowInvoker,
	cfg WorkerConfig,
) (*Worker, *renewRecordingWorkItemService, *connector_workqueue.WorkItem) {
	t.Helper()
	cfg.Namespace = workerTestNS
	cfg.PackInstallID = "pack-test"
	if cfg.WorkerID == "" {
		cfg.WorkerID = "test-worker"
	}
	if cfg.PollInterval == 0 {
		cfg.PollInterval = time.Millisecond
	}
	cfg.WorkItemSchedulerOwnerID = "scheduler-owner"
	cfg.WorkItemSchedulerFencingToken = "scheduler-fence"
	cfg.WorkItemLeaseDuration = time.Minute
	work := liveWorkItem("work-"+ev.ID, ev.ID, connector_workqueue.WorkTypeConnectorEvent, 0, "lease-"+ev.ID)
	work.InstalledPackID = cfg.PackInstallID
	work.State = connector_workqueue.WorkStatePending
	claims := &fakeWorkItemClaims{work: work}
	service := &renewRecordingWorkItemService{
		completeWorkItemServiceFake: newCompleteWorkItemServiceFake(claims, &fakeWorkItemEventLoader{event: ev}, nil),
	}
	resolver := &stubRouteResolver{routes: map[string]string{
		workerTestNS + "/noop.created": "NoopFlow__c",
	}}
	w, err := NewWorkItemWorkerForTest(context.Background(), cfg, service, &fakeSyncState{}, nil, flow, resolver)
	if err != nil {
		t.Fatalf("NewWorkItemWorkerForTest: %v", err)
	}
	return w, service, work
}

// newHeartbeatTestWorker builds a Worker wired to the noop.created route (same
// route map buildTestWorker uses) with a caller-supplied WorkerConfig so each
// test can set LeaseHeartbeatInterval / FlowDeadline. Namespace/PackInstallID/
// WorkerID are forced to the workerTestNS values noopEvent() carries.
func newHeartbeatTestWorker(t *testing.T, eventsSvc EventsService, flow FlowInvoker, cfg WorkerConfig) *Worker {
	t.Helper()
	resolver := &stubRouteResolver{routes: map[string]string{
		workerTestNS + "/noop.created": "NoopFlow__c",
	}}
	cfg.Namespace = workerTestNS
	cfg.PackInstallID = "pack-test"
	if cfg.WorkerID == "" {
		cfg.WorkerID = "test-worker"
	}
	if cfg.PollInterval == 0 {
		cfg.PollInterval = 1 * time.Millisecond
	}
	w, err := newWorkerForTest(cfg, eventsSvc, &fakeSyncState{}, fakeBatchRead{}, flow, resolver)
	if err != nil {
		t.Fatalf("NewWorker: %v", err)
	}
	return w
}

var _ connector_inbound_routing.RouteResolver = (*stubRouteResolver)(nil)

// TestWorker_Heartbeat_RenewsInFlightLeaseWhileFlowRuns proves the mandatory
// WorkItem executor renews a slow-but-alive flow's row-CAS lease periodically.
func TestWorker_Heartbeat_RenewsInFlightLeaseWhileFlowRuns(t *testing.T) {
	ev := noopEvent("E-slow", "noop.created")

	blocking := &blockingOnceFlowInvoker{started: make(chan struct{}), release: make(chan struct{})}

	w, service, seededWorkItem := newWorkItemHeartbeatTestWorker(t, ev, blocking, WorkerConfig{
		MaxClaims:              1,
		LeaseHeartbeatInterval: 20 * time.Millisecond,
	})
	generation := int64(17)
	bindHeartbeatClaim(service, seededWorkItem, generation)

	done := make(chan error, 1)
	go func() { done <- w.Run(context.Background()) }()

	<-blocking.started
	claim, claimedWorkItem := assertHeartbeatClaim(t, service, seededWorkItem, generation)

	// While the flow blocks, the heartbeat ticker must fire repeatedly. Wait until
	// we observe at least 2 renewals (proves it is periodic, not one-shot).
	deadline := time.After(2 * time.Second)
	for service.renewCount() < 2 {
		select {
		case <-deadline:
			t.Fatalf("heartbeat did not renew the in-flight WorkItem lease at least twice (got %d)", service.renewCount())
		case <-time.After(5 * time.Millisecond):
		}
	}

	for _, c := range service.renewedClaims() {
		if c.workID != claimedWorkItem.WorkID {
			t.Fatalf("RenewWorkItemLease targeted %q, want %q", c.workID, claimedWorkItem.WorkID)
		}
		if c.leaseToken != claim.LeaseToken {
			t.Fatalf("RenewWorkItemLease lease token = %q, want generated claim token %q", c.leaseToken, claim.LeaseToken)
		}
	}

	close(blocking.release)
	if err := <-done; err != nil {
		t.Fatalf("worker Run returned error: %v", err)
	}
	if got := service.claims.completeCalls; len(got) != 1 || got[0] != connector_workqueue.WorkStateProcessed {
		t.Fatalf("WorkItem terminal states = %v, want [processed]", got)
	}
	if got := service.completedClaims(); len(got) != 1 || got[0].workID != claimedWorkItem.WorkID || got[0].leaseToken != claim.LeaseToken || got[0].terminalState != connector_workqueue.WorkStateProcessed {
		t.Fatalf("CompleteWorkItem calls = %+v, want claimed WorkItem processed with generated lease", got)
	}
}

// TestWorker_Heartbeat_StopsAfterFinalize proves the heartbeat goroutine is
// stopped deterministically once the WorkItem finalizes: no renewal fires after
// flow completion and its processed CAS (no leaked goroutine or post-finalize
// renewal that could touch a reclaimed row).
func TestWorker_Heartbeat_StopsAfterFinalize(t *testing.T) {
	ev := noopEvent("E-fast", "noop.created")

	flow := &fakeFlowInvoker{successResult: &FlowResult{IsSuccess: true}}

	w, service, seededWorkItem := newWorkItemHeartbeatTestWorker(t, ev, flow, WorkerConfig{
		MaxClaims:              1,
		LeaseHeartbeatInterval: 10 * time.Millisecond,
	})
	generation := int64(23)
	bindHeartbeatClaim(service, seededWorkItem, generation)

	if err := w.Run(context.Background()); err != nil {
		t.Fatalf("worker Run: %v", err)
	}

	settled := service.renewCount()
	time.Sleep(120 * time.Millisecond)
	if after := service.renewCount(); after != settled {
		t.Fatalf("heartbeat kept renewing after finalize (%d -> %d) — goroutine leaked / could touch a reclaimed row", settled, after)
	}

	claim, claimedWorkItem := assertHeartbeatClaim(t, service, seededWorkItem, generation)
	if got := service.claims.completeWorkIDs; len(got) != 1 || got[0] != claimedWorkItem.WorkID {
		t.Fatalf("completed WorkItems = %v, want [%s]", got, claimedWorkItem.WorkID)
	}
	if got := service.claims.completeCalls; len(got) != 1 || got[0] != connector_workqueue.WorkStateProcessed {
		t.Fatalf("WorkItem terminal states = %v, want [processed]", got)
	}
	if got := service.completedClaims(); len(got) != 1 || got[0].workID != claimedWorkItem.WorkID || got[0].leaseToken != claim.LeaseToken || got[0].terminalState != connector_workqueue.WorkStateProcessed {
		t.Fatalf("CompleteWorkItem calls = %+v, want claimed WorkItem processed with generated lease", got)
	}
}

// TestWorker_Heartbeat_ErrLeaseLostIsSwallowed proves a heartbeat renewal that
// returns ErrLeaseLost (the row was finalized or reclaimed mid-flight) does not
// crash the worker or fail the event — it is a quiet no-op.
func TestWorker_Heartbeat_ErrLeaseLostIsSwallowed(t *testing.T) {
	ev := noopEvent("E-lost", "noop.created")
	fake := newFakeEvents(ev)
	rec := &renewRecordingEvents{fakeEventsService: fake, renewErr: connector_events.ErrLeaseLost}

	blocking := &blockingOnceFlowInvoker{started: make(chan struct{}), release: make(chan struct{})}

	w := newHeartbeatTestWorker(t, rec, blocking, WorkerConfig{
		MaxClaims:              1,
		LeaseHeartbeatInterval: 10 * time.Millisecond,
	})

	done := make(chan error, 1)
	go func() { done <- w.Run(context.Background()) }()

	<-blocking.started
	deadline := time.After(1 * time.Second)
	for rec.renewCount() < 1 {
		select {
		case <-deadline:
			t.Fatalf("heartbeat never fired")
		case <-time.After(5 * time.Millisecond):
		}
	}
	close(blocking.release)

	if err := <-done; err != nil {
		t.Fatalf("worker Run returned error despite swallowed ErrLeaseLost: %v", err)
	}
	if len(fake.processed) != 1 {
		t.Fatalf("event not processed despite benign heartbeat ErrLeaseLost: processed=%v", fake.processed)
	}
}

// TestWorker_Heartbeat_DisabledWhenServiceLacksRenewLease proves the heartbeat is
// fail-soft: a bare fakeEventsService does NOT implement RenewLease, so the worker
// must run the event to completion without the heartbeat, never panicking.
func TestWorker_Heartbeat_DisabledWhenServiceLacksRenewLease(t *testing.T) {
	ev := noopEvent("E-norenew", "noop.created")
	fake := newFakeEvents(ev)
	flow := &fakeFlowInvoker{successResult: &FlowResult{IsSuccess: true}}

	w := newHeartbeatTestWorker(t, fake, flow, WorkerConfig{
		MaxClaims:              1,
		LeaseHeartbeatInterval: 5 * time.Millisecond,
	})
	if err := w.Run(context.Background()); err != nil {
		t.Fatalf("worker Run with no-RenewLease service: %v", err)
	}
	if len(fake.processed) != 1 {
		t.Fatalf("event not processed when heartbeat capability absent: %v", fake.processed)
	}
}

// TestWorker_FlowDeadline_ReleasesForRetryWithoutBudgetBurn proves the opt-in
// per-event flow deadline retries the mandatory WorkItem without creating a
// failure-budget transition.
func TestWorker_FlowDeadline_ReleasesForRetryWithoutBudgetBurn(t *testing.T) {
	ev := noopEvent("E-hung", "noop.created")

	// A flow that blocks until its ctx is cancelled (never released) — a hung flow.
	blocking := &blockingOnceFlowInvoker{started: make(chan struct{}), release: make(chan struct{})}

	w, service, seededWorkItem := newWorkItemHeartbeatTestWorker(t, ev, blocking, WorkerConfig{
		MaxClaims:    1,
		FlowDeadline: 40 * time.Millisecond, // opt-in, short for the test
	})
	seededWorkItem.LeaseToken = "seed-lease"
	seededWorkItem.LeaseOwner = "seed-owner"
	seededWorkItem.Generation = -1
	generation := int64(17)
	service.claims.resolveGeneration = &generation
	service.bindClaimedWorkItem()

	// Run swallows per-event errors (RunClaimFirstUntilDrained continues after a
	// finalized event), so the assertion is on the finalize path, not Run's return:
	// the deadline must route through RetryWorkItem, not FailWorkItem.
	if err := w.Run(context.Background()); err != nil {
		t.Fatalf("worker Run returned an unexpected top-level error: %v", err)
	}

	if service.claims.retryCalls != 1 {
		t.Fatalf("deadline RetryWorkItem calls = %d, want 1", service.claims.retryCalls)
	}
	claim, claimedWorkItem, ok := service.claimedWork()
	if !ok {
		t.Fatal("deadline path did not materialize a claimed WorkItem")
	}
	if claim.InstalledPackID != "pack-test" || claim.Generation != generation || claim.OwnerID != "scheduler-owner" || claim.FencingToken != "scheduler-fence" {
		t.Fatalf("claim input = %+v, want pack-test/generation %d/scheduler owner and fence", claim, generation)
	}
	if claim.LeaseToken == "" || claim.LeaseToken == seededWorkItem.LeaseToken || claim.LeaseUntil.IsZero() || !claim.LeaseUntil.After(claim.Now) {
		t.Fatalf("claim lease = %q until %v, want generated token and deadline", claim.LeaseToken, claim.LeaseUntil)
	}
	if claimedWorkItem.State != connector_workqueue.WorkStateProcessing || claimedWorkItem.WorkID == "" || claimedWorkItem.LeaseToken != claim.LeaseToken || claimedWorkItem.LeaseOwner != claim.OwnerID || claimedWorkItem.Generation != claim.Generation || !claimedWorkItem.LeaseUntil.Valid || !claimedWorkItem.LeaseUntil.Time.Equal(claim.LeaseUntil) {
		t.Fatalf("claimed WorkItem = %+v, want processing row materialized from claim input %+v", claimedWorkItem, claim)
	}
	if got := service.claims.retryWorkIDs; len(got) != 1 || got[0] != claimedWorkItem.WorkID {
		t.Fatalf("retried WorkItems = %v, want [%s]", got, claimedWorkItem.WorkID)
	}
	if got := service.claims.retryWorkTokens; len(got) != 1 || got[0] != claim.LeaseToken {
		t.Fatalf("retry lease tokens = %v, want generated claim token [%s]", got, claim.LeaseToken)
	}
	if len(service.claims.failInputs) != 0 || len(service.claims.completeCalls) != 0 {
		t.Fatalf("WorkItem failure/complete calls = %v/%v, want none", service.claims.failInputs, service.claims.completeCalls)
	}
}

// TestWorker_FlowDeadline_ZeroDisablesNoBehaviorChange proves FlowDeadline==0
// disables the per-event deadline at the worker layer: a slow-but-completing
// flow finishes normally (MarkProcessed), never released-for-retry. The
// supervisor injects a generous non-zero default (defaultConnectorFlowDeadline);
// 0 is the explicit "off" escape hatch operators set via the env knob.
func TestWorker_FlowDeadline_ZeroDisablesNoBehaviorChange(t *testing.T) {
	ev := noopEvent("E-slow-ok", "noop.created")
	fake := newFakeEvents(ev)

	flow := &fakeFlowInvoker{
		successResult: &FlowResult{IsSuccess: true},
		onExecute:     func() { time.Sleep(60 * time.Millisecond) },
	}

	w := newHeartbeatTestWorker(t, fake, flow, WorkerConfig{
		MaxClaims:    1,
		FlowDeadline: 0, // OFF
	})
	if err := w.Run(context.Background()); err != nil {
		t.Fatalf("worker Run with deadline OFF: %v", err)
	}
	if len(fake.processed) != 1 || fake.processed[0] != "E-slow-ok" {
		t.Fatalf("slow-but-completing flow not processed with deadline OFF: processed=%v released=%v", fake.processed, fake.releasedForRetry)
	}
	if len(fake.releasedForRetry) != 0 {
		t.Fatalf("FlowDeadline==0 still released-for-retry: %v (disabled deadline must not regress a slow flow)", fake.releasedForRetry)
	}
}

// TestWorker_FlowDeadline_BatchDeadlineReleasesWholeBatchForRetry proves the
// batch-collection path honours FlowDeadline too: a hung batch flow that exceeds
// the deadline re-drives the WHOLE batch via ReleaseClaimForRetry (no RetryCount
// burn), never MarkFailed — the batch mirror of the single-event deadline path.
// Without this a batch flow kept alive by the lease heartbeat would pin its lane
// forever (the heartbeat renews the lease but nothing bounds the hung flow).
func TestWorker_FlowDeadline_BatchDeadlineReleasesWholeBatchForRetry(t *testing.T) {
	ev := noopEvent("E-batch-hung", "noop.created")
	fake := newFakeEvents(ev)

	// A batch flow that blocks until its ctx is cancelled (never released) — hung.
	blocking := &blockingOnceFlowInvoker{started: make(chan struct{}), release: make(chan struct{})}

	resolver := &stubRouteResolver{
		specs: map[string]connector_inbound_routing.RouteExecutionSpec{
			workerTestNS + "/noop.created": {
				FlowDeveloperName: "NoopBatchFlow",
				InvocationMode:    connector_inbound_routing.InvocationModeBatchCollection,
				MaxBatchSize:      10,
			},
		},
	}
	w, err := newWorkerForTest(WorkerConfig{
		Namespace:     workerTestNS,
		PackInstallID: "pack-test",
		WorkerID:      "test-worker",
		PollInterval:  time.Millisecond,
		MaxClaims:     1,
		BatchSize:     1,
		FlowDeadline:  40 * time.Millisecond, // bounds the hung batch flow
	}, fake, &fakeSyncState{}, fakeBatchRead{}, blocking, resolver)
	if err != nil {
		t.Fatalf("NewWorker: %v", err)
	}

	if err := w.Run(context.Background()); err != nil {
		t.Fatalf("worker Run returned an unexpected top-level error: %v", err)
	}

	if len(fake.releasedForRetry) != 1 || fake.releasedForRetry[0].eventID != "E-batch-hung" {
		t.Fatalf("batch deadline did not ReleaseClaimForRetry E-batch-hung: releasedForRetry=%v failed=%v", fake.releasedForRetry, fake.failed)
	}
	if len(fake.failed) != 0 {
		t.Fatalf("batch deadline burned the dead-letter budget via MarkFailed: %v (must use ReleaseClaimForRetry)", fake.failed)
	}
	if len(fake.processed) != 0 {
		t.Fatalf("hung batch flow was marked processed: %v", fake.processed)
	}
}

// slowBatchRead blocks BatchRead until release is closed (signaling started), so
// a test can hold the worker in the post-claim batchRead window while the lease
// heartbeat ticker fires. Mirrors blockingOnceFlowInvoker but for the vendor read.
type slowBatchRead struct {
	started chan struct{}
	release chan struct{}
	once    sync.Once
}

func (s *slowBatchRead) BatchRead(ctx context.Context, _, _ string, _ []string) (map[string]json.RawMessage, error) {
	s.once.Do(func() { close(s.started) })
	select {
	case <-s.release:
		return map[string]json.RawMessage{}, nil
	case <-ctx.Done():
		return nil, ctx.Err()
	}
}

// TestWorker_Heartbeat_RenewsLeaseDuringSlowBatchRead proves the lease heartbeat
// covers the POST-CLAIM vendor read, not only flow execution. The claim-first
// path claims the row, then does a retry-backed BatchRead (which can exceed the
// 300s stale cutoff) BEFORE executeEvent starts the flow heartbeat. Without a
// read-window heartbeat the claimed row goes stale during a slow read and is
// reclaimed + double-executed. The assertion (>=2 renewals while blocked in the
// read) fails on the unguarded version and passes once the read is heartbeated.
func TestWorker_Heartbeat_RenewsLeaseDuringSlowBatchRead(t *testing.T) {
	ev := noopEvent("E-slow-read", "noop.created")
	fake := newFakeEvents(ev)
	rec := &renewRecordingEvents{fakeEventsService: fake}

	slow := &slowBatchRead{started: make(chan struct{}), release: make(chan struct{})}
	flow := &fakeFlowInvoker{successResult: &FlowResult{IsSuccess: true}}

	resolver := &stubRouteResolver{routes: map[string]string{
		workerTestNS + "/noop.created": "NoopFlow__c",
	}}
	w, err := newWorkerForTest(WorkerConfig{
		Namespace:              workerTestNS,
		PackInstallID:          "pack-test",
		WorkerID:               "test-worker",
		PollInterval:           time.Millisecond,
		MaxClaims:              1,
		LeaseHeartbeatInterval: 20 * time.Millisecond,
	}, rec, &fakeSyncState{}, slow, flow, resolver)
	if err != nil {
		t.Fatalf("NewWorker: %v", err)
	}

	done := make(chan error, 1)
	go func() { done <- w.RunClaimFirstUntilDrained(context.Background(), nil) }()

	<-slow.started // worker is now blocked in the post-claim batchRead

	deadline := time.After(2 * time.Second)
	for rec.renewCount() < 2 {
		select {
		case <-deadline:
			t.Fatalf("heartbeat did not renew during the post-claim batchRead (got %d) — a slow vendor read would be stale-reclaimed and double-executed", rec.renewCount())
		case <-time.After(5 * time.Millisecond):
		}
	}
	for _, c := range rec.renewedTokens() {
		if c.eventID != "E-slow-read" {
			t.Fatalf("RenewLease targeted %q during the read, want E-slow-read", c.eventID)
		}
		if c.claimToken == "" {
			t.Fatalf("RenewLease carried an empty claim token — the CAS gate would never match")
		}
	}

	close(slow.release)
	if err := <-done; err != nil {
		t.Fatalf("worker RunClaimFirstUntilDrained returned error: %v", err)
	}
	if len(fake.processed) != 1 || fake.processed[0] != "E-slow-read" {
		t.Fatalf("event not processed after slow read completed: %v", fake.processed)
	}
}

// TestWorker_Heartbeat_BatchReadPanic_NoGoroutineLeak proves the mandatory
// WorkItem read heartbeat is panic-safe. A BatchRead panic is a supported
// recovered path (panickingBatchRead, worker_panic_recovery_test.go); the read
// heartbeat's stop is deferred inside batchReadHeartbeated, so the panic unwinds
// through it and the WorkItem renewer goroutine is halted rather than leaking
// recurring RenewWorkItemLease calls.
func TestWorker_Heartbeat_BatchReadPanic_NoGoroutineLeak(t *testing.T) {
	ev := noopEvent("E-panic-read", "noop.created")
	const (
		workID         = "work-panic-read"
		seedLeaseToken = "seed-panic-read-lease"
	)
	service := &panicHeartbeatWorkItemService{
		panicDeadLetterWorkItemService: newPanicDeadLetterWorkItemService(
			map[string]*connector_events.ConnectorEvent{ev.ID: ev},
			&connector_workqueue.WorkItem{
				WorkID: workID, EventID: ev.ID, InstalledPackID: "seed-pack",
				WorkType: connector_workqueue.WorkTypeConnectorEvent, QueueClass: connector_workqueue.DefaultQueueClass,
				State: connector_workqueue.WorkStatePending, Generation: -1,
				LeaseOwner: "seed-panic-read-owner", LeaseToken: seedLeaseToken,
			},
		),
		renewed: make(chan struct{}),
	}
	generation := int64(29)
	service.completeWorkItemServiceFake.claims.resolveGeneration = &generation
	flow := &fakeFlowInvoker{successResult: &FlowResult{IsSuccess: true}}

	resolver := &stubRouteResolver{routes: map[string]string{
		workerTestNS + "/noop.created": "NoopFlow__c",
	}}
	cfg := liveWorkItemConfig()
	cfg.MaxClaims = 1
	cfg.BatchSize = 1
	cfg.LeaseHeartbeatInterval = 5 * time.Millisecond
	cfg.FlowDeadline = time.Second
	w, err := NewWorkItemWorker(context.Background(),
		cfg,
		service,
		&fakeSyncState{},
		panicAfterWorkItemRenewalBatchRead{renewed: service.renewed},
		flow,
		resolver,
	)
	if err != nil {
		t.Fatalf("NewWorkItemWorker: %v", err)
	}

	// The recovered batchRead panic must finalize exactly the WorkItem row that
	// was claimed under the generated lease. It occurs before flow execution.
	if err := runUntilReturn(t, func() error {
		return w.RunUntilDrained(context.Background(), nil)
	}, 5*time.Second); err != nil {
		t.Fatalf("RunUntilDrained: %v", err)
	}
	assertPanicDeadLetterWorkItemHistory(
		t,
		service.panicDeadLetterWorkItemService,
		workID,
		ev.ID,
		generation,
		seedLeaseToken,
		connector_workqueue.WorkStateRetry,
	)
	renewals := service.renewals()
	if len(renewals) == 0 {
		t.Fatal("BatchRead panicked before the WorkItem heartbeat renewed the claimed row")
	}
	for i, renewal := range renewals {
		if !renewal.accepted || renewal.workID != workID || renewal.state != connector_workqueue.WorkStateProcessing || renewal.rowLeaseToken == "" || renewal.rowLeaseToken != renewal.claimLeaseToken || renewal.rowLeaseToken != renewal.leaseToken || renewal.leaseToken == seedLeaseToken || renewal.leaseUntil.IsZero() || !renewal.leaseUntil.After(renewal.now) {
			t.Fatalf("RenewWorkItemLease[%d] = %+v, want the active generated WorkItem row-CAS lease", i, renewal)
		}
	}
	if flow.calls != 0 {
		t.Fatalf("flow calls = %d, want 0: batch-read panic must finalize before flow execution", flow.calls)
	}

	base := service.renewCount()
	time.Sleep(60 * time.Millisecond) // 12 heartbeat intervals; a leaked goroutine would renew repeatedly
	if after := service.renewCount(); after != base {
		t.Fatalf("WorkItem renewer goroutine leaked after a recovered batchRead panic (%d -> %d renewals) — the read heartbeat stop was skipped by the panic", base, after)
	}
}

// TestWorker_BatchRead_DeadlineBoundsWedgedRead proves a WEDGED post-claim read is
// bounded by FlowDeadline. The read heartbeat keeps an alive read's lease fresh,
// but a hung vendor read would otherwise be renewed forever (stale-reclaim can no
// longer recover a heartbeated row) and pin the lane indefinitely. FlowDeadline
// cancels the wedged read's ctx so the read errors and the row is RELEASED FOR
// RETRY (NOT MarkFailed — a hung-but-not-dead read self-heals on backoff, exactly
// like a hung flow) — a bounded recovery, not an indefinite pin. Without the
// deadline the worker would never return (the read never completes), so the test's
// own timeout is the regression guard.
func TestWorker_BatchRead_DeadlineBoundsWedgedRead(t *testing.T) {
	ev := noopEvent("E-wedged-read", "noop.created")
	fake := newFakeEvents(ev)

	// A read that never returns on its own; only ctx cancellation frees it.
	wedged := &slowBatchRead{started: make(chan struct{}), release: make(chan struct{})}
	flow := &fakeFlowInvoker{successResult: &FlowResult{IsSuccess: true}}

	resolver := &stubRouteResolver{routes: map[string]string{
		workerTestNS + "/noop.created": "NoopFlow__c",
	}}
	w, err := newWorkerForTest(WorkerConfig{
		Namespace:              workerTestNS,
		PackInstallID:          "pack-test",
		WorkerID:               "test-worker",
		PollInterval:           time.Millisecond,
		MaxClaims:              1,
		LeaseHeartbeatInterval: 10 * time.Millisecond,
		FlowDeadline:           50 * time.Millisecond, // bounds the wedged read
	}, fake, &fakeSyncState{}, wedged, flow, resolver)
	if err != nil {
		t.Fatalf("NewWorker: %v", err)
	}

	done := make(chan error, 1)
	go func() { done <- w.RunClaimFirstUntilDrained(context.Background(), nil) }()

	select {
	case <-done:
	case <-time.After(3 * time.Second):
		t.Fatal("worker did not return — the wedged read was not bounded by FlowDeadline (lane pinned indefinitely)")
	}

	// The deadline-cancelled read surfaces as a batch-read error; like a hung flow,
	// a wedged read is RELEASED FOR RETRY (NOT MarkFailed) so it self-heals on
	// backoff rather than burning the dead-letter budget — and the row leaves
	// 'processing', so the lane is not pinned.
	if len(fake.releasedForRetry) != 1 || fake.releasedForRetry[0].eventID != "E-wedged-read" {
		t.Fatalf("wedged read did not ReleaseClaimForRetry: released=%v failed=%v processed=%v", fake.releasedForRetry, fake.failed, fake.processed)
	}
	if len(fake.failed) != 0 {
		t.Fatalf("wedged read burned the dead-letter budget via MarkFailed: %v (must release for retry)", fake.failed)
	}
	if len(fake.processed) != 0 {
		t.Fatalf("wedged read was processed despite never returning: %v", fake.processed)
	}
}