consumer

type Broadcast struct {
	// contains filtered or unexported fields
}

type BroadcastConfig struct {
	CommonConfig

	// StreamName is the JetStream stream to consume from.
	// Required.
	StreamName string `validate:"required"`

	// InstanceID identifies the unique instance of the application.
	//
	// This ID allows each instance to have its own durable consumer, ensuring
	// that every instance receives a copy of every message (fan-out).
	//
	// Accepted formats:
	//   - "fixed-string": Uses the literal value as the identity
	//   - "env:ENV_NAME": Uses the value of the specified environment variable
	//
	// If empty, the consumer will attempt to derive an identity from
	// HOSTNAME, then POD_NAME, and finally fall back to a generated short ID.
	InstanceID string

	// FilterSubject is the subject filter to consume from.
	// Supports wildcards (e.g., "orders.*", "events.>").
	// Messages matching this filter will be broadcast to all instances.
	FilterSubject string `validate:"required"`

	// ConsumerPrefix is the prefix for the durable consumer name.
	//
	// The final durable name is constructed as "<ConsumerPrefix>_broadcast_<InstanceID>".
	// This ensures that each instance gets a unique durable name, achieving fan-out.
	ConsumerPrefix string `validate:"required"`

	// Retry configures the backoff behavior for the consume loop.
	//
	// On each iterator failure the loop sleeps for a jittered delay that grows via
	// decorrelated jitter from Base toward Max using the Multiplier factor. Seed
	// makes the sequence deterministic (useful in tests); zero uses the package-level
	// PRNG. The delay resets to zero after every successful iteration cycle.
	Retry RetryConfig

	// IteratorFactory optionally overrides the internal iterator creation logic.
	// This is primarily used for testing to inject mock iterators.
	IteratorFactory func(cons jetstream.Consumer, batch int, expiry time.Duration) (jetstream.MessagesContext, error)

	// RecoveryStrategy defines how a recreated consumer resumes after an unexpected deletion.
	//
	// All strategies are supported. [RecoverFromLastProcessed] works with both
	// ManualAck=false (checkpoint advances automatically) and ManualAck=true
	// (checkpoint advances when the handler calls msg.Ack() or msg.DoubleAck()).
	RecoveryStrategy RecoveryStrategy
}

type BroadcastOption interface {
	// contains filtered or unexported methods
}

type CommonConfig struct {
	// Logger provides structured logging for the consumer.
	// If nil, a no-op logger is used.
	Logger types.Logger

	// Metrics is the metrics collector for worker consumer operations.
	// If nil, a no-op collector is used.
	Metrics types.WorkerConsumerMetrics

	// ManualAck disables automatic acknowledgement of messages.
	//
	// When false (default):
	//  - If the handler returns nil, the message is automatically acknowledged.
	//  - If the handler returns an error, the message is negatively acknowledged (Nak).
	//
	// When true:
	//  - The handler MUST explicitly call msg.Ack(), msg.Nak(), or msg.Term().
	//  - A returned handler error is discarded (neither logged nor acted on);
	//    surface failures from inside the handler itself.
	ManualAck bool

	// AckWait is the time allowed for processing a message before it is considered lost
	// and re-delivered by the server.
	//
	// This should be longer than the expected maximum processing time of a single message.
	//
	// Default: 30s.
	AckWait time.Duration `default:"30s" validate:"gt=0"`

	// MaxDeliver is the maximum number of times a message will be delivered.
	//
	// If a message fails processing (Nak) or times out (AckWait) this many times,
	// it will be terminated or moved to a Dead Letter Queue (if configured on the stream).
	//
	// Default: -1 (unlimited).
	MaxDeliver int `default:"-1" validate:"gte=-1"`

	// BatchSize is the maximum number of messages to pull from the server in a single request.
	//
	// A higher batch size can improve throughput but typically increases memory usage
	// and potentially latency for individual messages if processing is slow.
	//
	// Default: 1.
	BatchSize int `default:"1" validate:"gt=0"`

	// FetchTimeout is the maximum duration to wait for a batch of messages to arrive
	// when pulling from the server.
	//
	// If no messages are available within this timeout, the pull request expires.
	// The consumer loop manages this automatically.
	//
	// Minimum: 1s. NATS rejects a PullExpiry below 1s at iterator-creation
	// time; a sub-second value previously caused Start to return success and
	// then fail every iterator creation forever — a permanently dead consumer
	// with no terminal signal. Construction now fails fast instead.
	//
	// Default: 5s.
	FetchTimeout time.Duration `default:"5s" validate:"gt=0"`

	// PullHeartbeatCap optionally bounds the derived nats.go PullHeartbeat
	// value used by the consumer's pull loop. The heartbeat is normally
	// FetchTimeout/2, clamped to nats.go's PullHeartbeat validity range
	// [500ms, 30s]; when PullHeartbeatCap > 0 the derived heartbeat is
	// further capped to this value.
	//
	// The heartbeat drives missed-heartbeat detection: nats.go's
	// ErrNoHeartbeat fires at roughly 2x the heartbeat, which is how a
	// deleted durable consumer is detected. Raising FetchTimeout to reduce
	// idle pull-request churn also raises the derived heartbeat and
	// therefore that detection latency; this cap bounds it independent of
	// FetchTimeout.
	//
	// 0 (default) disables the cap: the heartbeat stays FetchTimeout/2,
	// capped only at 30s. A nonzero value outside [500ms, 30s] (nats.go's
	// PullHeartbeat bounds) is rejected at construction with
	// ErrInvalidConfig. See [WithPullHeartbeatCap].
	PullHeartbeatCap time.Duration `validate:"gte=0"`

	// MaxWaiting is the maximum number of outstanding pull requests allowed.
	//
	// This controls the pre-fetch buffer. A value of 2 with BatchSize of 1 means
	// there can be 2 messages buffered locally (1 being processed, 1 ready).
	//
	// Default: 2.
	MaxWaiting int `default:"2" validate:"gt=0"`

	// MaxAckPending limits the number of messages that can be in-flight (unacknowledged)
	// at any given time.
	//
	// If the limit is reached, the server will pause delivery until some messages are acknowledged.
	// If zero, the server's consumer default is used.
	MaxAckPending int `validate:"gte=0"`

	// InactiveThreshold is the duration after which an idle consumer (with no active subscriptions)
	// will be automatically deleted by the server.
	//
	// For durable consumers, this should be set high enough to survive application restarts.
	//
	// Default: 24h.
	InactiveThreshold time.Duration `default:"24h" validate:"gt=0"`

	// AckPolicy controls the JetStream acknowledgement policy.
	//
	// Typically set to AckExplicitPolicy for reliable processing.
	// Defaults to AckExplicitPolicy if usually not set manually.
	AckPolicy jetstream.AckPolicy

	// ConsumerMemoryStorage, when true, sets the underlying
	// jetstream.ConsumerConfig.MemoryStorage flag on consumer create,
	// keeping the consumer's delivery/ack state in memory rather than
	// inheriting the stream's storage type.
	//
	// Default: false (inherit stream storage).
	//
	// Trade-off: the consumer's delivery/ack offsets are NOT durable
	// across coordinated cluster restart. With ConsumerReplicas ≥ 2,
	// single-node failure is still survivable via raft peers. With
	// ConsumerReplicas = 1, any failure of the consumer-state holder
	// loses ack state and triggers redelivery from DeliverPolicy.
	//
	// IMPORTANT: this field is NOT live-editable on the NATS server.
	// Changing it after the consumer exists requires delete + recreate,
	// which drops ack/delivery offsets. Pick the value at construction
	// time.
	//
	// For at-least-once work-queue patterns with idempotent handlers
	// this is typically safe and yields a large IOPS reduction. See
	// docs/plans/iops-investigation/findings.md §2 for measurements
	// and §4 for the operator decision tree.
	ConsumerMemoryStorage bool

	// ConsumerReplicas overrides the underlying
	// jetstream.ConsumerConfig.Replicas value at consumer create time.
	//
	// Default: 0 (inherit the stream's replica count). Set to 1 to
	// disable consumer-state raft replication (lowest IOPS, no
	// consumer-state HA). Values between 1 and the stream's replica
	// count give intermediate IOPS/HA trade-offs.
	//
	// Constraint: must be ≤ the parent stream's Replicas. NATS rejects
	// invalid values at consumer create with error code 10126
	// ("consumer config replica count exceeds parent stream"). parti
	// does not pre-validate; the JetStream error is surfaced verbatim
	// when the underlying consumer is created or updated
	// (Queue/Static/Broadcast at Start, Dynamic at Update).
	//
	// Unlike ConsumerMemoryStorage, this field IS live-editable on
	// the NATS server via `nats consumer edit --replicas=N`; the raft
	// group expands/shrinks in place.
	ConsumerReplicas int `validate:"gte=0"`
}

type ConsumerCreateLimiter interface {
	// Wait blocks until the limiter grants one token or ctx is cancelled.
	Wait(ctx context.Context) error
}

type Dynamic struct {
	// contains filtered or unexported fields
}

type DynamicConfig struct {
	CommonConfig

	// StreamName is the JetStream stream to consume from.
	// Required.
	StreamName string `validate:"required"`

	// SubjectTemplate is a text/template for building subjects from partitions.
	//
	// It relies on the standard Go text/template package.
	// The template context provides a {{.PartitionID}} variable.
	//
	// Example: "orders.{{.PartitionID}}.events"
	SubjectTemplate string `validate:"required"`

	// ConsumerPrefix is the prefix for the durable consumer name.
	//
	// The final durable name is constructed dynamically for each assigned partition:
	// "<ConsumerPrefix>_<partitionID>_<hash>".
	//
	// This ensures unique, stable durability for each partition assignment.
	ConsumerPrefix string `validate:"required"`

	// ProcessingGate configures optional per-message admission control.
	//
	// When enabled, the consumer checks KV-backed ownership claims before each
	// handler invocation and NAKs deliveries for partitions it does not own.
	// This bounds — but does not eliminate — cross-worker overlap during
	// rebalances: an already-in-flight handler invocation cannot be revoked,
	// and AckWait-expiry redelivery through the shared per-partition durable
	// can hand the same message to the new owner while the old owner's handler
	// is still running. Wrap long-running handlers with [NewWIPHandler] to
	// prevent that expiry. See docs/LIFECYCLE.md "Two-Phase Handoff" for the
	// per-tier overlap contract.
	ProcessingGate *ProcessingGateConfig

	// Resolver configures the ownership resolver used when ProcessingGate is enabled.
	//
	// It defines how ownership is claimed, refreshed, and verified.
	Resolver ResolverConfig

	// PullGatingEnabled enables pre-pull ownership/state gating for consumers.
	//
	// When true, the consumer will check if it still owns the partition before
	// issuing a pull request to JetStream. This reduces "ghost" processing of
	// messages after assignment revocation.
	PullGatingEnabled bool

	// DrainOnRemove enables graceful draining when a partition assignment is revoked.
	//
	// When true, removal first polls the partition's durable until the server
	// reports no pending acknowledgements — the pull loop keeps running during
	// this wait — and only then stops the loop. Both the drain wait and the
	// loop-stop wait are bounded by DrainOnRemoveTimeout.
	DrainOnRemove bool

	// DrainOnRemoveTimeout caps the time spent draining a revoked partition
	// and bounds the wait for its pull loop to stop.
	//
	// The wait is best-effort: if loops have not stopped within this bound,
	// [Dynamic.Update] returns an error (the manager retries the apply with
	// backoff) while an already-in-flight handler invocation may still run
	// to completion. Default: 10s.
	DrainOnRemoveTimeout time.Duration `default:"10s" validate:"gte=0"`

	// MaxConcurrentSubjects limits the number of partitions (subjects) processed concurrently.
	//
	// If an assignment's deduped subject count exceeds this limit,
	// [Dynamic.Update] rejects the whole update with [ErrMaxSubjectsExceeded]
	// before making any changes; the manager retries the apply with backoff
	// and the previous owners keep consuming. Size this above the worst-case
	// per-worker partition count (e.g. after scale-down) or leave it 0
	// (unlimited).
	MaxConcurrentSubjects int `validate:"gte=0"`

	// AllowWorkerIDChange controls whether the worker's identity can change during runtime.
	//
	// Default: false (immutable once set). Changing WorkerID usually requires a restart.
	AllowWorkerIDChange bool

	// Retry configures the backoff behavior for the consume loop.
	//
	// On each iterator failure the loop sleeps for a jittered delay that grows via
	// decorrelated jitter from Base toward Max using the Multiplier factor. Seed
	// makes the sequence deterministic (useful in tests); zero uses the package-level
	// PRNG. The delay resets to zero after every successful iteration cycle.
	Retry RetryConfig

	// IteratorEscalationWindow defines the sliding time window used to aggregate
	// iterator failures for escalation detection.
	//
	// If too many iterator errors occur within this window, the consumer will
	// attempt to escalate recovery.
	//
	// Default: 60s.
	IteratorEscalationWindow time.Duration `default:"60s" validate:"gt=0"`

	// IteratorEscalationThreshold is the number of iterator failures within the
	// escalation window that triggers consumer refresh/escalation.
	//
	// Default: 3.
	IteratorEscalationThreshold int `default:"3" validate:"gt=0"`

	// PartitionRefreshMinInterval sets the minimum interval between forced claim refreshes
	// per partition when pull gating is enabled.
	//
	// This prevents excessive load on the coordination backend (KV) during high-throughput pulling.
	//
	// Default: 500ms.
	PartitionRefreshMinInterval time.Duration `default:"500ms" validate:"gt=0"`

	// IteratorFactory optionally overrides the internal iterator creation logic.
	// This is primarily used for testing to inject mock iterators.
	IteratorFactory func(cons jetstream.Consumer, batch int, expiry time.Duration) (jetstream.MessagesContext, error)

	// RecoveryStrategy defines how a recreated consumer resumes after an unexpected deletion.
	//
	// All strategies are supported. [RecoverFromLastProcessed] works with both
	// ManualAck=false (checkpoint advances automatically) and ManualAck=true
	// (checkpoint advances when the handler calls msg.Ack() or msg.DoubleAck()).
	//
	// Note: Dynamic consumers have a separate iterator-escalation detector (see
	// [DynamicConfig.IteratorEscalationWindow]) that rebinds to the existing durable
	// on repeated failures. RecoveryStrategy controls a distinct, complementary mechanism
	// that recreates the durable itself when it has been deleted.
	//
	// # WorkQueuePolicy streams
	//
	// NATS only permits [jetstream.DeliverAllPolicy] on WorkQueuePolicy streams.
	// [RecoverFromNew] and [RecoverFromLastProcessed] are incompatible and will
	// cause the first [Dynamic.Update] call to return [ErrInvalidConfig]. Use
	// [RecoverFromBeginning] or [RecoveryDisabled] for WorkQueuePolicy streams.
	RecoveryStrategy RecoveryStrategy

	// StreamMissingHook fires when the dynamic partition consumer cannot
	// create a consumer because the underlying JetStream stream is absent.
	// The hook is the operator-driven recovery escalation path; the
	// library does not recreate streams itself.
	//
	// Returning nil indicates the operator has recreated the stream. The
	// library then bumps the recovery controller's stream-epoch, resets
	// its checkpoint, re-seeds from the new consumer info, and rebuilds
	// the consumer. Returning a non-nil error (or omitting the hook
	// entirely) surfaces the loss via the F2 envelope's exhaustion path:
	// OnPermanentFailure fires with the error wrapped in
	// [types.ErrStreamMissing] so the manager routes it to enterDegraded.
	//
	// Requires [RecoveryStrategy] in {[RecoverFromLastProcessed],
	// [RecoverFromBeginning]}; [RecoveryDisabled] (default) and
	// [RecoverFromNew] are rejected at [DynamicConfig.Validate] time.
	// See [types.StreamMissingHook] godoc for the full operator
	// contract (same-durable-name preservation, compatible-config
	// restore, fresh-stream replay semantics).
	StreamMissingHook types.StreamMissingHook

	// OnPermanentFailure is the application-facing observability seam for
	// permanent partition-consumer failure. It fires exactly once per
	// partition consumer when the iterator-creation envelope or the
	// stream-missing Site B detour exhausts its attempt budget; the
	// consumption loop exits immediately afterwards.
	//
	// The callback runs synchronously on the consumption loop's goroutine
	// and MUST be non-blocking. The error preserves the wrap chain:
	// errors.Is(err, [types.ErrStreamMissing]) is true when stream-missing
	// exhaustion drove the failure, so applications can route those
	// distinctly from generic envelope exhaustion.
	//
	// Registering this callback does NOT disable the Parti manager's
	// auto-degraded route: for stream-missing exhaustion the manager
	// observer is notified after this callback returns. Use
	// [DynamicConfig.SuppressManagerDegradeOnStreamMissing] (option form:
	// [WithSuppressManagerDegradeOnStreamMissing]) to opt out explicitly.
	// See [WithOnPermanentFailure] for the full contract.
	OnPermanentFailure func(subject string, err error)

	// SuppressManagerDegradeOnStreamMissing disables the Parti manager's
	// auto-degraded route for stream-missing recovery exhaustion. By default
	// the manager observer is notified even when OnPermanentFailure is set
	// (both fire; application callback first). Set this only when the
	// application deliberately owns degrade/rotation signaling itself.
	SuppressManagerDegradeOnStreamMissing bool

	// RecoveryRetry tunes the bounded-retry envelope wrapped around
	// the iterator-creation path and the stream-missing Site B
	// detour. Zero-valued fields fall back to the durable layer's
	// defaults (MaxAttempts=8, BaseBackoff=500ms, MaxBackoff=30s).
	// See [WithRecoveryRetry] for the option form.
	RecoveryRetry RecoveryRetryConfig
}

type DynamicOption interface {
	// contains filtered or unexported methods
}

type GateMetrics interface {
	// IncGateNAK increments the NAK counter by reason
	// (e.g., "unknown_ownership", "non_owner", "owner_disallowed_state").
	IncGateNAK(reason string)

	// ObserveGateNakDelay records the NAK delay applied by the gate.
	ObserveGateNakDelay(d time.Duration)
}

type MessageHandler interface {
	// Handle processes a JetStream message.
	// Return nil to indicate successful processing; return an error for failures.
	// Ack/Nak behavior depends on the consumer's ManualAck setting:
	//   - ManualAck=false: nil → Ack, error → Nak (default behavior)
	//   - ManualAck=true: handler must call msg.Ack/Nak/Term explicitly
	Handle(ctx context.Context, msg jetstream.Msg) error
}

type MessageHandlerFunc func(ctx context.Context, msg jetstream.Msg) error

type Option interface {
	// contains filtered or unexported methods
}

type ProcessingGateConfig struct {
	// Enabled toggles the gate.
	// Default: false (gate disabled).
	Enabled bool

	// AllowedStates defines which handoff states permit message processing.
	// If empty, defaults to [types.StateStable].
	//
	// Common configurations:
	//  - [types.StateStable]: Strict consistency (default). Messages only processed when ownership is stable.
	//  - [types.StateStable, types.StateHandoff]: Higher availability. Allows processing during handoff.
	AllowedStates []types.HandoffState `validate:"omitempty,min=1"`

	// WarmupDuration, when >0, enables a warm-up phase during which only
	// WarmupAllowedStates are permitted. After the duration elapses, AllowedStates
	// take effect for steady-state processing.
	WarmupDuration time.Duration `validate:"gte=0"`

	// WarmupAllowedStates defines which states are permitted during the warm-up phase.
	// If empty and WarmupDuration>0, defaults to Stable-only.
	WarmupAllowedStates []types.HandoffState

	// NakDelay is the base delay for NAK when the worker is not the owner or
	// the state is disallowed.
	// Default: 100ms.
	NakDelay time.Duration `default:"100ms" validate:"gte=0"`

	// NakJitter is a fractional jitter in [0.0, 1.0] applied to NakDelay.
	// Default: 0.0 (no jitter).
	NakJitter float64 `validate:"gte=0,lte=1"`

	// Debug enables verbose logging for NAK decisions (non-owner, disallowed state, etc).
	Debug bool

	// Metrics optionally records processing gate metrics (NAK reasons, delays).
	Metrics GateMetrics
}

type Queue struct {
	// contains filtered or unexported fields
}

type QueueConfig struct {
	CommonConfig

	// StreamName is the name of the JetStream stream to consume from.
	// This field is required and must match an existing stream.
	StreamName string `validate:"required"`

	// FilterSubject is the subject filter to consume from.
	// Supports wildcards (e.g., "orders.*", "events.>").
	// Only messages matching this filter will be delivered to the consumer.
	FilterSubject string `validate:"required"`

	// ConsumerName is the durable consumer name.
	// This name must be unique within the stream.
	// For Queue consumers, this name identifies the shared consumer group;
	// multiple instances using the same ConsumerName will share the load.
	ConsumerName string `validate:"required"`

	// Retry configures the backoff behavior for control-plane operations
	// (e.g., initial connection, creating the consumer).
	Retry RetryConfig

	// IteratorFactory optionally overrides the internal iterator creation logic.
	// This is primarily used for testing to inject mock iterators.
	IteratorFactory func(cons jetstream.Consumer, batch int, expiry time.Duration) (jetstream.MessagesContext, error)

	// RecoveryStrategy defines how a recreated consumer resumes after an unexpected deletion.
	//
	// Supported values for Queue consumers:
	//   - [RecoverFromNew]: skip messages published during the outage.
	//   - [RecoverFromBeginning]: replay all messages from the start of the stream.
	//
	// [RecoverFromLastProcessed] is rejected at construction time because Queue shares
	// one durable across all replicas — per-process checkpointing is nondeterministic.
	//
	// # WorkQueuePolicy streams
	//
	// NATS only permits [jetstream.DeliverAllPolicy] on WorkQueuePolicy streams.
	// [RecoverFromNew] maps to [jetstream.DeliverNewPolicy] and is therefore incompatible:
	// [Queue.Start] rejects the combination with [ErrInvalidConfig]. The pre-flight
	// check is best-effort — transient failures to fetch stream info are ignored
	// (see [CheckWorkQueueRecoveryCompat]) — so an incompatible combination can
	// slip past Start and only surface when recovery first misbehaves.
	// Use [RecoverFromBeginning] or [RecoveryDisabled] for WorkQueuePolicy streams.
	RecoveryStrategy RecoveryStrategy
}

type QueueOption interface {
	// contains filtered or unexported methods
}

type RecoveryRetryConfig = durable.RecoveryRetryConfig

type RecoveryStrategy = recovery.Strategy

type ResolverConfig struct {
	// OwnershipResolver (advanced) supplies custom ownership lookups.
	// When nil and ProcessingGate.Enabled is true, a claim-based resolver
	// is auto-created using HandoffBucketName/HandoffClaimsPrefix. When
	// non-nil, it overrides the automatic resolver creation.
	OwnershipResolver types.OwnershipResolver

	// HandoffBucketName specifies the KV bucket name for handoff claims.
	// When ProcessingGate.Enabled is true and this is set, the Dynamic consumer
	// will automatically get/create the KV bucket and start a claim resolver.
	// Should match the HandoffBucket in parti.Config.KVBuckets for consistency.
	// Default: "parti-handoff".
	HandoffBucketName string `default:"parti-handoff"`

	// HandoffClaimsPrefix is the key prefix for handoff claims in the KV bucket.
	// Default: "claims/".
	HandoffClaimsPrefix string `default:"claims/"`

	// BatchWindow is the coalescing window for batching claim updates into the
	// auto-created claim-based resolver when ProcessingGate is enabled.
	// If zero, a default (5ms) is used. Ignored when a custom OwnershipResolver
	// is provided.
	BatchWindow time.Duration `default:"5ms" validate:"gte=0"`

	// BatchMaxItems caps the number of unique partition updates coalesced into a
	// single apply. If zero, a default (1024) is used. Ignored when a custom
	// OwnershipResolver is provided.
	BatchMaxItems int `default:"1024" validate:"gt=0"`

	// ReconcileInterval is the cadence at which the auto-created claim-based
	// resolver re-lists the handoff bucket and reconciles its cache against
	// KV. This is the recovery mechanism for silent watcher stalls (the
	// nats.go KV watcher does NOT surface NATS server restarts as Updates()
	// channel close; only an explicit Stop / connection close / subscription
	// teardown does). After such a stall the cache stays stale for at most
	// one reconcile period.
	//
	// Choose a value shorter than parti.Config.ExtendedApplyGracePeriod
	// (default 5 × HeartbeatTTL) so the leader-side audit cannot escalate
	// audit_repair while the worker's resolver cache is still stale. With
	// the default HeartbeatTTL=15s, this gives an audit grace of 75s and
	// the default 30s reconcile is comfortably inside it. If you have tuned
	// HeartbeatTTL below ~6s, set ReconcileInterval to HeartbeatTTL or
	// HeartbeatTTL/2.
	//
	// Idle-bucket cost note: reconcile passes are scan-gated — when a
	// cheap stream-position probe proves the bucket unchanged since the
	// last full pass, the pass skips its key walk entirely (no ephemeral
	// consumer creation). Shortening this interval therefore no longer
	// multiplies steady-state consumer churn on an idle handoff bucket;
	// the interval still bounds worst-case staleness recovery as described
	// above.
	//
	// Zero uses the default (30s). Negative values are rejected at startup.
	// Ignored when a custom OwnershipResolver is provided.
	ReconcileInterval time.Duration `default:"30s" validate:"gte=0"`
}

type ResolverMetrics interface {
	// ObserveVisibilityLag records the lag between claim LastUpdated and
	// when the resolver observes the update (watcher delivery time).
	ObserveVisibilityLag(d time.Duration)

	// SetCacheSize sets the current resolver cache size (number of partitions
	// with visible claims).
	SetCacheSize(n int)

	// IncUpdate increments update counters by operation type
	// (e.g., "upsert", "delete").
	IncUpdate(op string)

	// ObserveBatch records properties of an applied batch: the number of
	// coalesced items and how long it took to apply to the cache.
	ObserveBatch(size int, applyDuration time.Duration)

	// IncBatchFlush increments a counter for batch flush events by reason
	// (e.g., "timer", "maxitems").
	IncBatchFlush(reason string)

	// IncWatcherRestart increments a counter when the resolver's KV watcher
	// is re-established after closure or establishment failure.
	// Reasons: "channel_closed", "drift_detected", "establish_failed".
	IncWatcherRestart(reason string)

	// IncReconcileRescue increments when reconcileScan applies any
	// change to the cache — i.e., the reconciler observed drift between
	// KV and the in-memory cache and rescued the cache. Persistent
	// non-zero values are strong evidence the KV watcher is silently
	// stalled (the nats.go KV watcher does not surface NATS server
	// restarts as Updates() channel close, so the supervisor's
	// channel_closed restart path will not fire for that trigger).
	// Operators should alert on this metric being non-zero across
	// multiple consecutive reconcile cycles.
	IncReconcileRescue()
}

type RetryConfig struct {
	// Backoff is the delay between retries for control-plane operations.
	// Default: 100ms.
	Backoff time.Duration `default:"100ms" validate:"gte=0"`

	// Max caps the jittered backoff.
	// Default: 5s.
	Max time.Duration `default:"5s" validate:"gte=0,gtefield=Backoff"`

	// Multiplier grows the backoff window for decorrelated jitter.
	// Default: 1.6.
	Multiplier float64 `default:"1.6" validate:"gte=1"`

	// Base is the base backoff used for decorrelated jitter retries.
	// Default: 200ms. If zero and Backoff is set, Base falls back to Backoff.
	Base time.Duration `default:"200ms" validate:"gte=0"`

	// Seed optionally seeds the jitter RNG for deterministic tests.
	// When zero, a random seed is used.
	Seed int64
}

type Static struct {
	// contains filtered or unexported fields
}

type StaticConfig struct {
	CommonConfig

	// StreamName is the JetStream stream to consume from.
	// Required.
	StreamName string `validate:"required"`

	// NumPartitions is the total number of partitions for the stream.
	//
	// This defines the sharding factor. Messages are consistently hashed
	// to a partition index in the range [0, NumPartitions-1].
	// Must be > 0.
	//
	// WARNING: Changing NumPartitions changes the hash mapping.
	NumPartitions int `validate:"required,gt=0"`

	// Partition is the specific partition index this consumer should process.
	//
	// Must be in the range [0, NumPartitions-1].
	// Typically assigned based on the application instance's ordinal (e.g., StatefulSet index).
	Partition int `validate:"gte=0"`

	// SubjectPattern is the subject template with placeholders.
	//
	// Placeholders:
	//   - {{partition}} - Replaced with partition index (0 to N-1). Required.
	//   - {{key}}       - Replaced with the partition key. Optional.
	//
	// Placeholders must occupy a full token between dots. Embedded placeholders
	// like "events.{{partition}}-v1" are invalid.
	//
	// Examples:
	//   - "events.completed.{{partition}}"       → "events.completed.0"
	//   - "events.{{key}}.{{partition}}"         → "events.tool-abc.3"
	//   - "orders.{{partition}}.{{key}}.created" → "orders.2.customer-xyz.created"
	//
	// Validation:
	//   - Must contain {{partition}} placeholder
	//   - Must not produce empty NATS subject tokens (e.g., "events..{{partition}}" is invalid)
	SubjectPattern string `validate:"required"`

	// ConsumerName is the durable consumer name.
	//
	// Must be unique per partition.
	// The final durable name on the server often incorporates the partition index
	// to avoid collisions between partitions, or the user must ensure uniqueness.
	ConsumerName string `validate:"required"`

	// HashSeed is an optional seed for the consistent hashing algorithm.
	//
	// Using a consistent seed ensures that the same message key always maps to
	// the same partition index across restarts/redeployments.
	HashSeed uint64

	// RecoveryStrategy defines how a recreated consumer resumes after an unexpected deletion.
	//
	// All strategies are supported. [RecoverFromLastProcessed] works with both
	// ManualAck=false (checkpoint advances automatically) and ManualAck=true
	// (checkpoint advances when the handler calls msg.Ack() or msg.DoubleAck()).
	//
	// # WorkQueuePolicy streams
	//
	// NATS only permits [jetstream.DeliverAllPolicy] on WorkQueuePolicy streams.
	// [RecoverFromNew] and [RecoverFromLastProcessed] are incompatible and will
	// cause [Static.Start] to return [ErrInvalidConfig]. Use [RecoverFromBeginning]
	// or [RecoveryDisabled] for WorkQueuePolicy streams.
	RecoveryStrategy RecoveryStrategy
}

type StaticOption interface {
	// contains filtered or unexported methods
}

type WIPConfig struct {
	// Interval is the heartbeat interval for msg.InProgress() calls.
	//
	// Recommended: set Interval to AckWait/3 for safety margin.
	// Values <= 0 disable heartbeats (returns unwrapped handler).
	// Values < DefaultWIPMinInterval are clamped to prevent overhead.
	Interval time.Duration

	// MinInterval overrides the minimum interval threshold.
	// If zero, DefaultWIPMinInterval (100ms) is used.
	// Set to a negative value to disable clamping (not recommended).
	MinInterval time.Duration

	// Logger receives heartbeat errors. If nil, errors are silently ignored.
	Logger types.Logger
}

type WIPHandler struct {
	// contains filtered or unexported fields
}