package connector_webhooks

import (
	"context"
	"errors"
	"fmt"
	"hash/fnv"
	"log"
	"os"
	"strconv"
	"strings"
	"sync"
	"time"

	connector_events "github.com/pingcap-inc/nexuscrm-connectors/features/connector/events"
	"github.com/pingcap-inc/nexuscrm-connectors/features/connector/ingressreceipts"
	"github.com/pingcap-inc/nexuscrm-connectors/internal/dependencyretry"
	"github.com/pingcap-inc/nexuscrm-connectors/internal/id"
)

const (
	defaultReceiptRouterLease     = 2 * time.Minute
	defaultReceiptRouterRetryBase = 30 * time.Second
	defaultReceiptRouterRetryMax  = 15 * time.Minute
	defaultReceiptRouterRecovery  = 5 * time.Minute

	// envReceiptRouterPollIntervalMs controls the legacy-store and schedule-read
	// error fallback. Production stores use post-commit wake, persisted retry
	// deadlines, and a low-frequency recovery scan instead of this interval for
	// steady-state idle polling. An explicit cfg.PollInterval still wins.
	envReceiptRouterPollIntervalMs         = "NEXUSCRM_CONNECTOR_RECEIPT_POLL_INTERVAL_MS"
	defaultReceiptRouterPollIntervalMs     = 5000
	minReceiptRouterPollIntervalMs         = 50
	maxReceiptRouterPollIntervalMs         = 5000
	envReceiptRouterBatchSize              = "NEXUSCRM_CONNECTOR_RECEIPT_ROUTER_BATCH_SIZE"
	defaultReceiptRouterBatchSize          = 50
	minReceiptRouterBatchSize              = 1
	maxReceiptRouterBatchSize              = 100
	envReceiptRouterPrepareConcurrency     = "NEXUSCRM_CONNECTOR_RECEIPT_ROUTER_PREPARE_CONCURRENCY"
	defaultReceiptRouterPrepareConcurrency = 16
	minReceiptRouterPrepareConcurrency     = 1
	maxReceiptRouterPrepareConcurrency     = 64
	envReceiptRouterPrepareTimeoutMs       = "NEXUSCRM_CONNECTOR_RECEIPT_ROUTER_PREPARE_TIMEOUT_MS"
	defaultReceiptRouterPrepareTimeoutMs   = 10000
	minReceiptRouterPrepareTimeoutMs       = 1000
	maxReceiptRouterPrepareTimeoutMs       = 60000
)

type ReceiptStore interface {
	ClaimNext(ctx context.Context, workerID, claimToken string, leaseExpiresAt time.Time) (*connector_ingress_receipts.Receipt, bool, error)
	MarkMaterializing(ctx context.Context, receiptID, claimToken string, metadata map[string]string) error
	MarkEnqueued(ctx context.Context, receiptID, claimToken string, eventIDs []string, metadata map[string]string) error
	MarkRetry(ctx context.Context, receiptID, claimToken string, nextAttemptAt time.Time, reason error) error
	MarkPendingConfig(ctx context.Context, receiptID, claimToken string, nextAttemptAt time.Time, reason error) error
	MarkDead(ctx context.Context, receiptID, claimToken string, reason error) error
	MarkConfigTerminal(ctx context.Context, receiptID, claimToken string, reason error) error
}

type ReceiptBatchStore interface {
	ClaimBatch(ctx context.Context, workerID, claimToken string, leaseExpiresAt time.Time, limit int) ([]*connector_ingress_receipts.Receipt, bool, error)
	MarkMaterializingBatch(ctx context.Context, updates []connector_ingress_receipts.MaterializingReceiptUpdate) error
	MarkEnqueuedBatch(ctx context.Context, updates []connector_ingress_receipts.EnqueuedReceiptUpdate) error
}

// receiptScheduleStore is an additive optimization implemented by the
// production receipt repository. The returned deadline is only a wake hint;
// ClaimBatch remains the sole authority for ownership, ordering, and leases.
type receiptScheduleStore interface {
	NextClaimableAt(context.Context) (time.Time, bool, error)
}

type ReceiptRouterConfig struct {
	WorkerID           string
	PollInterval       time.Duration
	LeaseDuration      time.Duration
	RetryBase          time.Duration
	RetryMax           time.Duration
	MaxClaims          int
	BatchSize          int
	PrepareConcurrency int
	PrepareTimeout     time.Duration
}

type ReceiptRouter struct {
	cfg              ReceiptRouterConfig
	store            ReceiptStore
	dispatch         *Dispatcher
	wakeCh           chan struct{}
	recoveryInterval time.Duration
	now              func() time.Time
	idgen            func() string
}

type preparedReceiptDelivery struct {
	receipt  *connector_ingress_receipts.Receipt
	prepared preparedVerifiedDelivery
}

type preparedReceiptGroup struct {
	namespace     string
	packInstallID string
	items         []preparedReceiptDelivery
}

type preparedReceiptResult struct {
	receipt               *connector_ingress_receipts.Receipt
	prepared              preparedVerifiedDelivery
	pinnedMaterialization bool
	err                   error
	outcome               string
	duration              time.Duration
}

func NewReceiptRouter(cfg ReceiptRouterConfig, store ReceiptStore, dispatch *Dispatcher) (*ReceiptRouter, error) {
	cfg.WorkerID = strings.TrimSpace(cfg.WorkerID)
	if cfg.WorkerID == "" {
		cfg.WorkerID = "connector-ingress-receipt-router"
	}
	if cfg.PollInterval <= 0 {
		cfg.PollInterval = receiptRouterPollIntervalFromEnv(os.Getenv)
	}
	if cfg.LeaseDuration <= 0 {
		cfg.LeaseDuration = defaultReceiptRouterLease
	}
	if cfg.RetryBase <= 0 {
		cfg.RetryBase = defaultReceiptRouterRetryBase
	}
	if cfg.RetryMax <= 0 {
		cfg.RetryMax = defaultReceiptRouterRetryMax
	}
	if cfg.BatchSize <= 0 {
		cfg.BatchSize = receiptRouterBatchSizeFromEnv(os.Getenv)
	}
	if cfg.PrepareConcurrency <= 0 {
		cfg.PrepareConcurrency = receiptRouterPrepareConcurrencyFromEnv(os.Getenv)
	}
	if cfg.PrepareTimeout <= 0 {
		cfg.PrepareTimeout = receiptRouterPrepareTimeoutFromEnv(os.Getenv)
	}
	if store == nil {
		return nil, fmt.Errorf("connector_webhooks: receipt router store is required")
	}
	if dispatch == nil {
		return nil, fmt.Errorf("connector_webhooks: receipt router dispatcher materializer is required")
	}
	router := &ReceiptRouter{
		cfg:              cfg,
		store:            store,
		dispatch:         dispatch,
		wakeCh:           make(chan struct{}, 1),
		recoveryInterval: defaultReceiptRouterRecovery,
		now:              time.Now,
		idgen:            id.Generate,
	}
	// Receipt creation and routing live in the same serving process. Bind a
	// best-effort wake after the durable insert so low ingress latency no longer
	// requires a permanent high-frequency empty database poll. The database row
	// and bounded safety poll remain the cross-process/restart authority.
	dispatch.receiptRouterWake = router.Wake
	return router, nil
}

// Wake asks the router to re-check durable receipt work immediately. It is
// intentionally edge-coalescing: the database remains the source of truth and
// one signal is sufficient to drain every currently eligible row.
func (r *ReceiptRouter) Wake() {
	if r == nil || r.wakeCh == nil {
		return
	}
	select {
	case r.wakeCh <- struct{}{}:
	default:
	}
}

func (r *ReceiptRouter) Run(ctx context.Context) error {
	return r.run(ctx, nil)
}

// RunReady reports the lane ready after its own durable store shape is verified.
// A slow receipt batch must not consume the process's startup budget. Run keeps
// its established contract for callers that do not participate in readiness.
func (r *ReceiptRouter) RunReady(ctx context.Context, started func()) error {
	if started == nil {
		return fmt.Errorf("connector_webhooks: receipt router startup callback is required")
	}
	return r.run(ctx, started)
}

func (r *ReceiptRouter) run(ctx context.Context, started func()) error {
	if ctx == nil {
		return fmt.Errorf("connector_webhooks: receipt router requires context")
	}
	if r == nil || r.store == nil || r.dispatch == nil {
		return fmt.Errorf("connector_webhooks: receipt router is not wired")
	}
	if started != nil {
		store, ok := r.store.(interface{ VerifyReady(context.Context) error })
		if !ok {
			return fmt.Errorf("connector_webhooks: receipt store readiness is unavailable")
		}
		if err := dependencyretry.Do(ctx, "ingress receipt schema readiness", func() error { return store.VerifyReady(ctx) }); err != nil {
			return err
		}
		started()
	}
	claims := 0
	idlePollInterval := r.cfg.PollInterval
	for {
		if ctx.Err() != nil {
			return ctx.Err()
		}
		if r.cfg.MaxClaims > 0 && claims >= r.cfg.MaxClaims {
			return nil
		}
		limit := r.cfg.BatchSize
		if r.cfg.MaxClaims > 0 {
			remaining := r.cfg.MaxClaims - claims
			if remaining < limit {
				limit = remaining
			}
		}
		claimed, err := r.processBatch(ctx, limit)
		if err != nil {
			log.Printf("connector ingress receipt router: %v", err)
		}
		if claimed > 0 {
			claims += claimed
			idlePollInterval = r.cfg.PollInterval
			continue
		}
		wait := idlePollInterval
		if scheduled, ok := r.store.(receiptScheduleStore); ok {
			next, found, scheduleErr := scheduled.NextClaimableAt(ctx)
			wait = receiptRouterScheduledWait(r.now(), next, found, scheduleErr, idlePollInterval, r.recoveryInterval)
		}
		timer := time.NewTimer(wait)
		select {
		case <-ctx.Done():
			stopTimer(timer)
			return ctx.Err()
		case <-r.wakeCh:
			stopTimer(timer)
			idlePollInterval = r.cfg.PollInterval
		case <-timer.C:
			idlePollInterval = nextReceiptRouterIdlePollInterval(idlePollInterval, r.cfg.PollInterval)
		}
	}
}

func receiptRouterScheduledWait(now, next time.Time, found bool, scheduleErr error, fallback, recovery time.Duration) time.Duration {
	if fallback <= 0 {
		fallback = time.Duration(defaultReceiptRouterPollIntervalMs) * time.Millisecond
	}
	if scheduleErr != nil {
		return fallback
	}
	if recovery <= 0 {
		recovery = defaultReceiptRouterRecovery
	}
	if !found {
		return recovery
	}
	wait := next.Sub(now)
	minimum := time.Duration(minReceiptRouterPollIntervalMs) * time.Millisecond
	if wait < minimum {
		wait = minimum
	}
	if wait > recovery {
		return recovery
	}
	return wait
}

func nextReceiptRouterIdlePollInterval(current, initial time.Duration) time.Duration {
	if initial <= 0 {
		initial = time.Duration(defaultReceiptRouterPollIntervalMs) * time.Millisecond
	}
	if current < initial {
		current = initial
	}
	maximum := time.Duration(maxReceiptRouterPollIntervalMs) * time.Millisecond
	if initial > maximum {
		// Explicit in-process configs predate the env clamp. Backoff must never
		// shorten a caller's deliberately longer safety interval.
		maximum = initial
	}
	if current >= maximum {
		return maximum
	}
	next := current * 2
	if next < current || next > maximum {
		return maximum
	}
	return next
}

func (r *ReceiptRouter) ProcessBatch(ctx context.Context) (int, error) {
	return r.processBatch(ctx, r.cfg.BatchSize)
}

func (r *ReceiptRouter) processBatch(ctx context.Context, limit int) (int, error) {
	if r == nil || r.store == nil || r.dispatch == nil {
		return 0, fmt.Errorf("connector_webhooks: receipt router is not wired")
	}
	batchStore, ok := r.store.(ReceiptBatchStore)
	if !ok || limit <= 1 {
		claimed, err := r.ProcessOne(ctx)
		if claimed {
			return 1, err
		}
		return 0, err
	}

	timer := newReceiptRouterStageTimer()
	releaseAdmission, admitted := r.acquireAdmission(ctx, timer)
	if !admitted {
		timer.finish("admission_rejected")
		return 0, nil
	}
	defer releaseAdmission()

	now := r.now()
	claimToken := r.idgen()
	receipts, ok, err := batchStore.ClaimBatch(ctx, r.cfg.WorkerID, claimToken, now.Add(r.cfg.LeaseDuration), limit)
	claimOutcome := "ok"
	if err != nil {
		claimOutcome = "error"
	} else if !ok || len(receipts) == 0 {
		claimOutcome = "none"
	}
	timer.mark("claim_batch", claimOutcome)
	if err != nil || !ok || len(receipts) == 0 {
		if err != nil {
			timer.finish("error")
		} else {
			timer.finish("no_claim")
		}
		return 0, err
	}

	prepareResults := r.prepareClaimedReceipts(ctx, receipts)
	timer.mark("materialize_prepare_batch", "ok")

	enqueued := make([]connector_ingress_receipts.EnqueuedReceiptUpdate, 0, len(receipts))
	materializing := make([]connector_ingress_receipts.MaterializingReceiptUpdate, 0, len(receipts))
	groups := make([]preparedReceiptGroup, 0, 1)
	groupIndex := map[string]int{}
	for _, result := range prepareResults {
		receipt := result.receipt
		if receipt == nil {
			continue
		}
		if result.err == nil {
			if len(result.prepared.events) == 0 {
				materialized := result.prepared.result([]string{}, []bool{})
				enqueued = append(enqueued, enqueuedReceiptUpdate(receipt, claimToken, materialized))
				continue
			}
			if !result.pinnedMaterialization {
				materializing = append(materializing, connector_ingress_receipts.MaterializingReceiptUpdate{
					ReceiptID:  receipt.ID,
					ClaimToken: claimToken,
					Metadata:   result.prepared.metadata,
				})
			}
			groups = appendPreparedReceiptGroup(groups, groupIndex, preparedReceiptDelivery{
				receipt:  receipt,
				prepared: result.prepared,
			})
			continue
		}
		if markErr := r.markMaterializeError(ctx, receipt, claimToken, result.err, timer.mark); markErr != nil {
			timer.finish("error")
			return len(receipts), markErr
		}
	}
	if len(materializing) > 0 {
		if err := batchStore.MarkMaterializingBatch(ctx, materializing); err != nil {
			timer.mark("mark_materializing_batch", "error")
			timer.finish("error")
			return len(receipts), err
		}
		timer.mark("mark_materializing_batch", "ok")
	}
	for _, group := range groups {
		groupEnqueued, err := r.enqueuePreparedGroup(ctx, group, claimToken, timer.mark)
		if err != nil {
			timer.finish("error")
			return len(receipts), err
		}
		enqueued = append(enqueued, groupEnqueued...)
	}
	if len(enqueued) > 0 {
		if err := batchStore.MarkEnqueuedBatch(ctx, enqueued); err != nil {
			timer.mark("mark_enqueued_batch", "error")
			timer.finish("error")
			return len(receipts), err
		}
		timer.mark("mark_enqueued_batch", "ok")
	}
	timer.finish("enqueued")
	return len(receipts), nil
}

func (r *ReceiptRouter) acquireAdmission(ctx context.Context, timer *receiptRouterStageTimer) (func(), bool) {
	releaseAdmission, admission := r.dispatch.tryAcquireAdmission(ctx)
	if releaseAdmission == nil {
		releaseAdmission = func() {}
	}
	timer.mark("admission", webhookStageOutcomeOK(admission.Allow))
	r.dispatch.logAdmissionDecision("connector-ingress-receipt-router", admission)
	return releaseAdmission, admission.Allow
}

func (r *ReceiptRouter) enqueuePreparedGroup(
	ctx context.Context,
	group preparedReceiptGroup,
	claimToken string,
	mark func(stage, outcome string),
) ([]connector_ingress_receipts.EnqueuedReceiptUpdate, error) {
	out, err := r.enqueuePreparedGroupOnce(ctx, group, claimToken)
	if err == nil {
		mark("enqueue_batch", "ok")
		return out, nil
	}
	mark("enqueue_batch", "error")
	if ctx.Err() != nil {
		return nil, ctx.Err()
	}
	// These sentinels can describe one stale route or companion inside a
	// pack-level group. Probe source-lane components so healthy independent
	// components advance without allowing related receipts to overtake.
	if len(group.items) > 1 && r.shouldIsolatePreparedGroupError(ctx, group, err) {
		components := preparedReceiptGroupComponents(group)
		if len(components) > 1 {
			return r.enqueuePreparedGroupFallback(ctx, components, claimToken, mark)
		}
	}
	// EnqueueBatch is atomic. An unknown database error is therefore a
	// group-scoped retry, not evidence that one receipt is malformed. Probing
	// every receipt here would multiply pressure on the same pack lock.
	mark("enqueue_group_deferred", preparedGroupEnqueueOutcome(err))
	retryDelays := r.preparedGroupRetryDelays(group)
	return r.deferPreparedGroupEnqueue(ctx, group, claimToken, err, retryDelays, mark)
}

func (r *ReceiptRouter) enqueuePreparedGroupOnce(
	ctx context.Context,
	group preparedReceiptGroup,
	claimToken string,
) ([]connector_ingress_receipts.EnqueuedReceiptUpdate, error) {
	out := make([]connector_ingress_receipts.EnqueuedReceiptUpdate, 0, len(group.items))
	totalEvents := 0
	for _, item := range group.items {
		totalEvents += len(item.prepared.events)
	}
	if totalEvents > 0 {
		events := make([]connector_events.BatchEvent, 0, totalEvents)
		for _, item := range group.items {
			events = append(events, item.prepared.events...)
		}
		eventIDs, dups, err := r.dispatch.enqueueBatch(ctx, group.namespace, group.packInstallID, "", events)
		if err == nil && (len(eventIDs) != totalEvents || len(dups) != totalEvents) {
			err = fmt.Errorf("connector_webhooks: batch enqueue returned mismatched result length ids=%d dups=%d events=%d", len(eventIDs), len(dups), totalEvents)
		}
		if err != nil {
			return nil, err
		}
		offset := 0
		for _, item := range group.items {
			n := len(item.prepared.events)
			if n == 0 {
				continue
			}
			result := item.prepared.result(eventIDs[offset:offset+n], dups[offset:offset+n])
			out = append(out, enqueuedReceiptUpdate(item.receipt, claimToken, result))
			offset += n
		}
	}
	return out, nil
}

func (r *ReceiptRouter) shouldIsolatePreparedGroupError(ctx context.Context, group preparedReceiptGroup, err error) bool {
	if errors.Is(err, connector_events.ErrCompanionConsentMissing) ||
		errors.Is(err, connector_events.ErrCompanionVersionUnsatisfied) {
		return true
	}
	if !errors.Is(err, connector_events.ErrPackIncarnationChanged) || r.dispatch == nil || r.dispatch.resolvePack == nil {
		return false
	}
	// ErrPackIncarnationChanged also covers a swept route. Only isolate receipts
	// when the provider pack is still the same active incarnation; a real pack
	// rotation remains one group-scoped retry.
	row, ok, resolveErr := r.dispatch.resolvePack(ctx, group.namespace)
	return resolveErr == nil && ok && row != nil && strings.TrimSpace(row.InstallID) == group.packInstallID
}

func preparedGroupEnqueueOutcome(err error) string {
	if errors.Is(err, connector_events.ErrPackIncarnationChanged) ||
		errors.Is(err, connector_events.ErrCompanionConsentMissing) ||
		errors.Is(err, connector_events.ErrCompanionVersionUnsatisfied) {
		return string(materializerPendingConfig)
	}
	return string(materializerRetry)
}

func (r *ReceiptRouter) deferPreparedGroupEnqueue(
	ctx context.Context,
	group preparedReceiptGroup,
	claimToken string,
	groupErr error,
	retryDelays map[string]time.Duration,
	mark func(stage, outcome string),
) ([]connector_ingress_receipts.EnqueuedReceiptUpdate, error) {
	for _, item := range group.items {
		if len(item.prepared.events) == 0 {
			continue
		}
		classifiedErr := r.dispatch.classifyPreparedEnqueueError(item.prepared, groupErr)
		if markErr := r.markMaterializeErrorAfter(ctx, item.receipt, claimToken, classifiedErr, retryDelays[item.receipt.ID], mark); markErr != nil {
			return nil, markErr
		}
	}
	return nil, nil
}

func (r *ReceiptRouter) enqueuePreparedGroupFallback(
	ctx context.Context,
	components []preparedReceiptGroup,
	claimToken string,
	mark func(stage, outcome string),
) ([]connector_ingress_receipts.EnqueuedReceiptUpdate, error) {
	out := make([]connector_ingress_receipts.EnqueuedReceiptUpdate, 0)
	type componentFailure struct {
		component preparedReceiptGroup
		err       error
	}
	failures := make([]componentFailure, 0)
	for _, component := range components {
		componentOut, err := r.enqueuePreparedGroupOnce(ctx, component, claimToken)
		if err == nil {
			mark("enqueue_fallback", "ok")
			out = append(out, componentOut...)
			continue
		}
		mark("enqueue_fallback", preparedGroupEnqueueOutcome(err))
		if ctx.Err() != nil {
			return nil, ctx.Err()
		}
		failures = append(failures, componentFailure{component: component, err: err})
	}

	// No failed component clears its lease while later probes are still running.
	// Compute each absolute deadline only when that receipt leaves processing so
	// slow serial state writes cannot consume the next retry window.
	for _, failure := range failures {
		retryDelays := r.preparedGroupRetryDelays(failure.component)
		for _, item := range failure.component.items {
			classifiedErr := r.dispatch.classifyPreparedEnqueueError(item.prepared, failure.err)
			if markErr := r.markMaterializeErrorAfter(ctx, item.receipt, claimToken, classifiedErr, retryDelays[item.receipt.ID], mark); markErr != nil {
				return nil, markErr
			}
		}
	}
	return out, nil
}

func appendPreparedReceiptGroup(groups []preparedReceiptGroup, groupIndex map[string]int, item preparedReceiptDelivery) []preparedReceiptGroup {
	key := preparedReceiptGroupKey(item.prepared)
	idx, found := groupIndex[key]
	if !found {
		idx = len(groups)
		groupIndex[key] = idx
		groups = append(groups, preparedReceiptGroup{
			namespace:     item.prepared.namespace,
			packInstallID: item.prepared.packInstallID,
		})
	}
	groups[idx].items = append(groups[idx].items, item)
	return groups
}

func preparedReceiptGroupKey(prepared preparedVerifiedDelivery) string {
	return prepared.namespace + "\x00" + prepared.packInstallID
}

func enqueuedReceiptUpdate(receipt *connector_ingress_receipts.Receipt, claimToken string, result MaterializeResult) connector_ingress_receipts.EnqueuedReceiptUpdate {
	return connector_ingress_receipts.EnqueuedReceiptUpdate{
		ReceiptID:  receipt.ID,
		ClaimToken: claimToken,
		EventIDs:   result.EventIDs,
		Metadata:   result.Metadata,
	}
}

func (r *ReceiptRouter) ProcessOne(ctx context.Context) (bool, error) {
	if r == nil || r.store == nil || r.dispatch == nil {
		return false, fmt.Errorf("connector_webhooks: receipt router is not wired")
	}

	timer := newReceiptRouterStageTimer()
	releaseAdmission, admitted := r.acquireAdmission(ctx, timer)
	if !admitted {
		timer.finish("admission_rejected")
		return false, nil
	}
	defer releaseAdmission()

	now := r.now()
	claimToken := r.idgen()
	receipt, ok, err := r.store.ClaimNext(ctx, r.cfg.WorkerID, claimToken, now.Add(r.cfg.LeaseDuration))
	claimOutcome := "ok"
	if err != nil {
		claimOutcome = "error"
	} else if !ok || receipt == nil {
		claimOutcome = "none"
	}
	timer.mark("claim", claimOutcome)
	if err != nil || !ok || receipt == nil {
		if err != nil {
			timer.finish("error")
		} else {
			timer.finish("no_claim")
		}
		return ok, err
	}

	result, err := r.materializeClaimedReceipt(ctx, receipt, claimToken)
	timer.mark("materialize", materializerOutcome(err, "ok"))
	if err == nil {
		markErr := r.store.MarkEnqueued(ctx, receipt.ID, claimToken, result.EventIDs, result.Metadata)
		if markErr != nil {
			timer.mark("mark_enqueued", "error")
			timer.finish("error")
			return true, markErr
		}
		timer.mark("mark_enqueued", "ok")
		timer.finish("enqueued")
		return true, nil
	}

	if markErr := r.markMaterializeError(ctx, receipt, claimToken, err, timer.mark); markErr != nil {
		timer.finish("error")
		return true, markErr
	}
	timer.finish(materializerOutcome(err, "ok"))
	return true, nil
}

func (r *ReceiptRouter) materializeClaimedReceipt(ctx context.Context, receipt *connector_ingress_receipts.Receipt, claimToken string) (MaterializeResult, error) {
	delivery, pinnedMaterialization := receiptRouterVerifiedDelivery(receipt)
	if !pinnedMaterialization {
		delivery.BeforeEnqueue = func(ctx context.Context, snapshot MaterializationSnapshot) error {
			return r.store.MarkMaterializing(ctx, receipt.ID, claimToken, snapshot.Metadata)
		}
	}
	return r.dispatch.MaterializeVerifiedDelivery(ctx, delivery)
}

func (r *ReceiptRouter) prepareClaimedReceipt(ctx context.Context, receipt *connector_ingress_receipts.Receipt) (preparedVerifiedDelivery, bool, error) {
	if r != nil && r.cfg.PrepareTimeout > 0 {
		var cancel context.CancelFunc
		ctx, cancel = context.WithTimeout(ctx, r.cfg.PrepareTimeout)
		defer cancel()
	}
	delivery, pinnedMaterialization := receiptRouterVerifiedDelivery(receipt)
	timer := newMaterializerStageTimer()
	prepared, err := r.dispatch.prepareVerifiedDelivery(ctx, delivery, timer)
	timer.finish(materializerOutcome(err, "prepared"))
	return prepared, pinnedMaterialization, err
}

func (r *ReceiptRouter) prepareClaimedReceipts(ctx context.Context, receipts []*connector_ingress_receipts.Receipt) []preparedReceiptResult {
	results := make([]preparedReceiptResult, len(receipts))
	if len(receipts) == 0 {
		return results
	}
	workers := receiptRouterPrepareWorkerCount(r.cfg.PrepareConcurrency, len(receipts))
	jobs := make(chan int)
	var wg sync.WaitGroup
	wg.Add(workers)
	for i := 0; i < workers; i++ {
		go func() {
			defer wg.Done()
			for idx := range jobs {
				recordPreparedReceiptResult(results, idx, r.prepareReceiptResult(ctx, receipts[idx]))
			}
		}()
	}
	for idx := range receipts {
		if ctx.Err() != nil {
			break
		}
		jobs <- idx
	}
	close(jobs)
	wg.Wait()
	if ctx.Err() == nil {
		return results
	}
	recordCanceledPrepareResults(ctx.Err(), receipts, results)
	return results
}

func receiptRouterPrepareWorkerCount(configured int, receiptCount int) int {
	if receiptCount <= 0 {
		return 0
	}
	workers := configured
	if workers < minReceiptRouterPrepareConcurrency {
		workers = minReceiptRouterPrepareConcurrency
	}
	if workers > maxReceiptRouterPrepareConcurrency {
		workers = maxReceiptRouterPrepareConcurrency
	}
	if workers > receiptCount {
		workers = receiptCount
	}
	return workers
}

func (r *ReceiptRouter) prepareReceiptResult(ctx context.Context, receipt *connector_ingress_receipts.Receipt) preparedReceiptResult {
	started := time.Now()
	result := preparedReceiptResult{receipt: receipt}
	if receipt == nil {
		result.outcome = "nil"
		result.duration = time.Since(started)
		return result
	}
	result.prepared, result.pinnedMaterialization, result.err = r.prepareClaimedReceipt(ctx, receipt)
	result.outcome = materializerOutcome(result.err, "ok")
	result.duration = time.Since(started)
	return result
}

func recordPreparedReceiptResult(results []preparedReceiptResult, idx int, result preparedReceiptResult) {
	results[idx] = result
	recordConnectorReceiptRouterStageDuration("materialize_prepare", result.outcome, result.duration.Seconds())
}

func recordCanceledPrepareResults(ctxErr error, receipts []*connector_ingress_receipts.Receipt, results []preparedReceiptResult) {
	if ctxErr == nil {
		ctxErr = context.Canceled
	}
	for idx := range results {
		if results[idx].receipt != nil || receipts[idx] == nil {
			continue
		}
		recordPreparedReceiptResult(results, idx, preparedReceiptResult{
			receipt:  receipts[idx],
			err:      materializerRetryError(ctxErr),
			outcome:  string(materializerRetry),
			duration: 0,
		})
	}
}

func (r *ReceiptRouter) markMaterializeError(
	ctx context.Context,
	receipt *connector_ingress_receipts.Receipt,
	claimToken string,
	err error,
	mark func(stage, outcome string),
) error {
	return r.markMaterializeErrorAt(ctx, receipt, claimToken, err, r.nextRetryAt(receipt.AttemptCount), mark)
}

func (r *ReceiptRouter) markMaterializeErrorAt(
	ctx context.Context,
	receipt *connector_ingress_receipts.Receipt,
	claimToken string,
	err error,
	retryAt time.Time,
	mark func(stage, outcome string),
) error {
	switch materializerErrorKind(err) {
	case materializerDead:
		markErr := r.store.MarkDead(ctx, receipt.ID, claimToken, err)
		if markErr != nil {
			mark("mark_dead", "error")
			return markErr
		}
		mark("mark_dead", "ok")
		return nil
	case materializerPendingConfig:
		markErr := r.store.MarkPendingConfig(ctx, receipt.ID, claimToken, retryAt, err)
		if markErr != nil {
			mark("mark_pending_config", "error")
			return markErr
		}
		mark("mark_pending_config", "ok")
		return nil
	case materializerConfigTerminal:
		markErr := r.store.MarkConfigTerminal(ctx, receipt.ID, claimToken, err)
		if markErr != nil {
			mark("mark_config_terminal", "error")
			return markErr
		}
		mark("mark_config_terminal", "ok")
		return nil
	default:
		markErr := r.store.MarkRetry(ctx, receipt.ID, claimToken, retryAt, err)
		if markErr != nil {
			mark("mark_retry", "error")
			return markErr
		}
		mark("mark_retry", "ok")
		return nil
	}
}

func (r *ReceiptRouter) markMaterializeErrorAfter(
	ctx context.Context,
	receipt *connector_ingress_receipts.Receipt,
	claimToken string,
	err error,
	retryAfter time.Duration,
	mark func(stage, outcome string),
) error {
	return r.markMaterializeErrorAt(ctx, receipt, claimToken, err, r.now().Add(retryAfter), mark)
}

func receiptRouterVerifiedDelivery(receipt *connector_ingress_receipts.Receipt) (VerifiedDelivery, bool) {
	metadata := receiptRouterMetadata(receipt)
	delivery := VerifiedDelivery{
		Namespace:       receipt.Namespace,
		VerifiedPackID:  receipt.InstalledPackID,
		VerifiedVersion: receipt.PackVersion,
		SecretName:      receipt.SecretName,
		RawBody:         receipt.RawBody,
		EnqueueMetadata: metadata,
	}
	pinnedPackID := strings.TrimSpace(metadata["materializedPackId"])
	if pinnedPackID == "" {
		return delivery, false
	}
	delivery.MaterializePackID = pinnedPackID
	delivery.MaterializePackVersion = strings.TrimSpace(metadata["materializedPackVersion"])
	return delivery, true
}

func receiptRouterMetadata(receipt *connector_ingress_receipts.Receipt) map[string]string {
	out := map[string]string{}
	if receipt != nil {
		for k, v := range receipt.EnqueueMetadata {
			if strings.TrimSpace(k) != "" {
				out[k] = v
			}
		}
		out["receiptId"] = receipt.ID
		out["receiptKey"] = receipt.ReceiptKey
		out["bodyHash"] = receipt.BodyHash
		out["verifiedPackId"] = receipt.InstalledPackID
		out["verifiedPackVersion"] = receipt.PackVersion
	}
	return out
}

func (r *ReceiptRouter) nextRetryAt(attemptCount int) time.Time {
	return r.now().Add(r.retryDelay(attemptCount))
}

func (r *ReceiptRouter) preparedGroupRetryDelays(group preparedReceiptGroup) map[string]time.Duration {
	retryAttempts := preparedGroupRetryAttempts(group)
	delaysByAttempt := make(map[int]time.Duration)
	delays := make(map[string]time.Duration, len(group.items))
	for i, item := range group.items {
		if item.receipt == nil {
			continue
		}
		attemptCount := retryAttempts[i]
		delay, ok := delaysByAttempt[attemptCount]
		if !ok {
			delay = r.preparedGroupRetryDelay(group, attemptCount)
			delaysByAttempt[attemptCount] = delay
		}
		delays[item.receipt.ID] = delay
	}
	return delays
}

func preparedReceiptGroupComponents(group preparedReceiptGroup) []preparedReceiptGroup {
	roots := preparedReceiptGroupComponentRoots(group)
	componentIndex := make(map[int]int)
	components := make([]preparedReceiptGroup, 0, len(group.items))
	for i, item := range group.items {
		root := roots[i]
		idx, ok := componentIndex[root]
		if !ok {
			idx = len(components)
			componentIndex[root] = idx
			components = append(components, preparedReceiptGroup{
				namespace:     group.namespace,
				packInstallID: group.packInstallID,
			})
		}
		components[idx].items = append(components[idx].items, item)
	}
	return components
}

func preparedReceiptGroupComponentRoots(group preparedReceiptGroup) []int {
	parent := make([]int, len(group.items))
	for i := range parent {
		parent[i] = i
	}
	var find func(int) int
	find = func(i int) int {
		if parent[i] != i {
			parent[i] = find(parent[i])
		}
		return parent[i]
	}
	union := func(left, right int) {
		leftRoot := find(left)
		rightRoot := find(right)
		if leftRoot != rightRoot {
			parent[rightRoot] = leftRoot
		}
	}

	laneOwner := make(map[string]int)
	for i, item := range group.items {
		itemLanes := make(map[string]struct{})
		for _, laneID := range item.prepared.sourceLaneIDs {
			if laneID == "" {
				continue
			}
			if _, duplicate := itemLanes[laneID]; duplicate {
				continue
			}
			itemLanes[laneID] = struct{}{}
			if owner, ok := laneOwner[laneID]; ok {
				union(i, owner)
			} else {
				laneOwner[laneID] = i
			}
		}
	}
	roots := make([]int, len(group.items))
	for i := range group.items {
		roots[i] = find(i)
	}
	return roots
}

func preparedGroupRetryAttempts(group preparedReceiptGroup) []int {
	roots := preparedReceiptGroupComponentRoots(group)
	componentMax := make(map[int]int)
	for i, item := range group.items {
		if item.receipt == nil {
			continue
		}
		root := roots[i]
		if item.receipt.AttemptCount > componentMax[root] {
			componentMax[root] = item.receipt.AttemptCount
		}
	}
	attempts := make([]int, len(group.items))
	for i, item := range group.items {
		if item.receipt != nil {
			attempts[i] = componentMax[roots[i]]
		}
	}
	return attempts
}

func (r *ReceiptRouter) preparedGroupRetryDelay(group preparedReceiptGroup, attemptCount int) time.Duration {
	delay := r.retryDelay(attemptCount)

	// Jitter by pack and attempt, never by receipt or claim cohort. Receipts at
	// the same attempt therefore stay coalesced without letting an unrelated
	// high-attempt receipt impose its backoff on every lane in the pack.
	minimum := delay
	window := delay / 2
	if delay >= r.cfg.RetryMax {
		minimum = r.cfg.RetryMax - r.cfg.RetryMax/4
		window = r.cfg.RetryMax - minimum
	} else if remaining := r.cfg.RetryMax - delay; window > remaining {
		window = remaining
	}
	if window > 0 {
		h := fnv.New64a()
		_, _ = h.Write([]byte(preparedReceiptGroupRetrySeed(group)))
		_, _ = h.Write([]byte{0})
		_, _ = h.Write([]byte(strconv.Itoa(attemptCount)))
		delay = minimum + time.Duration(h.Sum64()%uint64(window))
	}
	return delay
}

func preparedReceiptGroupRetrySeed(group preparedReceiptGroup) string {
	return group.namespace + "\x00" + group.packInstallID
}

func (r *ReceiptRouter) retryDelay(attemptCount int) time.Duration {
	delay := r.cfg.RetryBase
	if delay > r.cfg.RetryMax {
		delay = r.cfg.RetryMax
	}
	for i := 0; i < attemptCount; i++ {
		if delay >= r.cfg.RetryMax/2 {
			delay = r.cfg.RetryMax
			break
		}
		delay *= 2
	}
	return delay
}

func (d *Dispatcher) verifiedIngressReceiptsEnabled() bool {
	if d != nil && d.receiptsEnabled != nil {
		return d.receiptsEnabled()
	}
	return connectorVerifiedIngressReceiptsEnabledFromEnv()
}

func VerifiedIngressReceiptsEnabledFromEnv() bool {
	return connectorVerifiedIngressReceiptsEnabledFromEnv()
}

// receiptRouterPollIntervalFromEnv resolves the idle poll interval from
// envReceiptRouterPollIntervalMs, clamped to [min,max]. With the env unset it
// returns the historical 5s default, so behavior is byte-identical when the knob
// is not configured.
func receiptRouterPollIntervalFromEnv(getenv func(string) string) time.Duration {
	ms := positiveIntFromEnv(getenv, envReceiptRouterPollIntervalMs, defaultReceiptRouterPollIntervalMs, minReceiptRouterPollIntervalMs, maxReceiptRouterPollIntervalMs)
	return time.Duration(ms) * time.Millisecond
}

func receiptRouterBatchSizeFromEnv(getenv func(string) string) int {
	return positiveIntFromEnv(getenv, envReceiptRouterBatchSize, defaultReceiptRouterBatchSize, minReceiptRouterBatchSize, maxReceiptRouterBatchSize)
}

func receiptRouterPrepareConcurrencyFromEnv(getenv func(string) string) int {
	return positiveIntFromEnv(getenv, envReceiptRouterPrepareConcurrency, defaultReceiptRouterPrepareConcurrency, minReceiptRouterPrepareConcurrency, maxReceiptRouterPrepareConcurrency)
}

func receiptRouterPrepareTimeoutFromEnv(getenv func(string) string) time.Duration {
	ms := positiveIntFromEnv(getenv, envReceiptRouterPrepareTimeoutMs, defaultReceiptRouterPrepareTimeoutMs, minReceiptRouterPrepareTimeoutMs, maxReceiptRouterPrepareTimeoutMs)
	return time.Duration(ms) * time.Millisecond
}

func stopTimer(t *time.Timer) {
	if !t.Stop() {
		select {
		case <-t.C:
		default:
		}
	}
}