subscription

type AuditResult struct {
	Partition       string
	LostMessages    int
	UnresolvedHoles int
	Duration        time.Duration
	TimedOut        bool
	LateMessages    int
}

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

type BroadcastConsumerConfig struct {
	// StreamName is the JetStream stream name where subjects live. Required.
	StreamName string `validate:"required"`

	// ConsumerPrefix is the prefix for the durable consumer name.
	// It must contain only alphanumeric characters (a-z, A-Z, 0-9), dashes (-), or underscores (_).
	// Final name format: <ConsumerPrefix>_broadcast_<consumerID>
	// Required.
	ConsumerPrefix string `validate:"required"`

	// ConsumerID controls the identity used in the broadcast durable name.
	//
	// 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.
	ConsumerID string

	// WildcardFilter is the JetStream FilterSubject for the consumer.
	// Use ">" for multi-token partition IDs (recommended).
	// Use "*" only for single-token partition IDs.
	//
	// Examples:
	//   - "orders.>" for partition IDs like "region.us-east"
	//   - "orders.*.events" for single-token partition IDs like "p1"
	//
	// Required.
	WildcardFilter string `validate:"required"`

	// Logger provides structured logging. Defaults to a no-op logger when nil.
	Logger types.Logger

	// Metrics is the global metrics collector used across the library.
	// If nil, no metrics are emitted from the consumer.
	Metrics types.MetricsCollector

	// ManualAck, when true, disables the helper's automatic Ack/Nak behavior.
	// Handlers are responsible for calling msg.Ack/Nak/Term and optionally
	// msg.InProgress() to extend AckWait. This enables handler-controlled
	// backpressure. Defaults to false.
	ManualAck bool

	// AckWait is the time allowed for processing before redelivery.
	// Default: 30s.
	AckWait time.Duration `default:"30s" validate:"gt=0"`

	// MaxDeliver is the maximum redelivery attempts before moving to DLQ (if configured).
	// Default: -1 (unlimited attempts).
	MaxDeliver int `default:"-1" validate:"gte=-1"`

	// BatchSize is the max number of messages to pull per iterator request.
	// Default: 1.
	BatchSize int `default:"1" validate:"gt=0"`

	// FetchTimeout is the max time to wait when pulling a batch.
	// Default: 5s.
	FetchTimeout time.Duration `default:"5s" validate:"gt=0"`

	// MaxWaiting caps outstanding pull requests for the durable consumer.
	// Default: 2.
	MaxWaiting int `default:"2" validate:"gt=0"`

	// MaxAckPending limits the number of in-flight unacknowledged messages the server
	// will allow. If zero, the server default is used.
	MaxAckPending int `validate:"gte=0"`

	// InactiveThreshold is how long an idle consumer is kept by the server before cleanup.
	// Default: 24h.
	InactiveThreshold time.Duration `default:"24h" validate:"gt=0"`

	// AckPolicy controls JetStream ack policy. Defaults to AckExplicitPolicy.
	AckPolicy jetstream.AckPolicy

	// Retry configures backoff behavior for control-plane operations.
	Retry RetryConfig

	// IteratorFactory optionally overrides the iterator creation logic for testing or
	// advanced customization. When nil, a default factory is used.
	IteratorFactory func(cons jetstream.Consumer, batch int, expiry time.Duration) (jetstream.MessagesContext, error)
}

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

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 single message.
	//
	// In default mode, return nil for the helper to ACK, or an error for the helper to NAK.
	// In ManualAck mode, the handler must call msg.Ack/Nak/Term exactly once and may
	// periodically call msg.InProgress() to extend AckWait while long-running work completes.
	Handle(ctx context.Context, msg jetstream.Msg) error
}

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

type MessageTracker interface {
	// OnDelivery records arrival of a message sequence for a partition.
	// publishTime is the server publish timestamp or derived header time.
	OnDelivery(partition string, sequence uint64, publishTime time.Time)

	// OnNAK records a negative acknowledgement and its reason.
	OnNAK(partition string, sequence uint64, reason NAKReason)

	// OnAck records successful final processing of a message sequence.
	OnAck(partition string, sequence uint64)

	// DrainAudit performs a bounded audit to classify unresolved holes as lost.
	DrainAudit(ctx context.Context, partition string) (AuditResult, error)
}

type NAKReason string

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 [StateStable].
	//
	// Common configurations:
	//  - [StateStable]: Strict consistency (default). Messages only processed when ownership is stable.
	//  - [StateStable, StateHandoff]: Higher availability. Allows processing during handoff.
	AllowedStates []types.HandoffState `validate:"omitempty,min=1"` //nolint:revive

	// 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,
	// unknown ownership). Only effective when a logger is configured.
	Debug bool

	// Metrics optionally records processing gate metrics (NAK reasons, delays).
	// When nil, no gate metrics are emitted.
	Metrics GateMetrics
}

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/Prefix. 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, WorkerConsumer
	// will automatically get/create the KV bucket and start a claim resolver.
	// Defaults to "parti-handoff" if empty and gate is enabled.
	// Should match the HandoffBucket in parti.Config.KVBuckets for consistency.
	HandoffBucketName string `default:"parti-handoff"`

	// HandoffClaimsPrefix is the key prefix for handoff claims in the KV bucket.
	// Defaults to "claims/" if empty and gate is enabled.
	HandoffClaimsPrefix string `default:"claims/"`

	// BatchWindow sets the resolver KV watch coalescing window for 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"`
}

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)
}

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

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

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

	// Base is the base backoff used for decorrelated jitter retries for control-plane ops.
	// Default: DefaultRetryBase (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 TrackerPolicy interface {
	// ConfigurePolicy sets latency classification policy.
	ConfigurePolicy(latencyThreshold time.Duration, strict bool, allowLateTail int)
}

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

type WorkerConsumerConfig struct {
	// StreamName is the JetStream stream name where subjects live. Required.
	StreamName string `validate:"required"`

	// ConsumerPrefix is the prefix for the durable consumer name.
	// It must contain only alphanumeric characters (a-z, A-Z, 0-9), dashes (-), or underscores (_).
	// Required.
	ConsumerPrefix string `validate:"required"`

	// SubjectTemplate is a text/template used to build subjects from a partition.
	// Uses Go's text/template syntax.
	//
	// Available template fields:
	//   - {{.PartitionID}}: Partition keys joined with "." (equals partition.SubjectKey())
	//
	// Note: Only PartitionID is available. Other Partition fields (Keys, Weight) are not
	// directly accessible in the template.
	//
	// Example: "orders.{{.PartitionID}}.events" with Keys=["region", "us-east"] produces
	// "orders.region.us-east.events"
	//
	// Required.
	SubjectTemplate string `validate:"required"`

	// Logger provides structured logging. Defaults to a no-op logger when nil.
	Logger types.Logger

	// Metrics is the global metrics collector used across the library.
	// If nil, no metrics are emitted from the worker consumer helper.
	Metrics types.MetricsCollector

	// ManualAck, when true, disables the helper's automatic Ack/Nak behavior.
	// Handlers are responsible for calling msg.Ack/Nak/Term and optionally
	// msg.InProgress() to extend AckWait. This enables handler-controlled
	// backpressure via an internal work queue. Defaults to false.
	//
	// Recommendation: Enable this for any asynchronous processing (e.g. spawning goroutines).
	// Keep false (default) for simple synchronous handlers to ensure safety.
	ManualAck bool

	// ProcessingGate configures optional exclusive processing enforcement.
	// When enabled, WorkerConsumer automatically creates and manages a claim-based
	// ownership resolver backed by the handoff KV bucket. The resolver lifecycle
	// (start/stop) is handled internally - no manual management needed.
	//
	// Recommendation: Enable this for stateful workloads or when strict partition
	// exclusivity is required. Without this, ownership transitions may be "loose",
	// resulting in brief periods where a partition is processed by a worker that
	// is no longer the assigned owner (though duplicates are rare).
	ProcessingGate *ProcessingGateConfig `validate:"omitempty"` //nolint:revive

	// Resolver configures the ownership resolver when ProcessingGate is enabled.
	Resolver ResolverConfig

	// PullGatingEnabled enables pre-pull ownership/state gating for per-subject consumers.
	// When true, pulls are suppressed (not issued) if the current worker is not the owner or
	// the handoff state is not Commit/Stable. Reduces NAK churn during handoffs.
	PullGatingEnabled bool

	// DrainOnRemove enables a graceful per-subject drain when a subject is removed
	// from the worker's assignment. When true, the consumer helper will stop issuing
	// new pulls for the subject and wait for pending acknowledgements to reach zero
	// (or until DrainOnRemoveTimeout elapses) before cancelling the loop. This reduces
	// NAK churn and minimizes gaps during scale-down and rebalancing.
	// Default: false (immediate cancellation as before).
	DrainOnRemove bool

	// DrainOnRemoveTimeout caps the drain wait per subject when DrainOnRemove is enabled.
	// Default: 10s when zero.
	DrainOnRemoveTimeout time.Duration `default:"10s" validate:"gte=0"`

	// AckWait is the time allowed for processing before redelivery.
	// Default: DefaultAckWait (30s).
	AckWait time.Duration `default:"30s" validate:"gt=0"`

	// MaxDeliver is the maximum redelivery attempts before moving to DLQ (if configured).
	// Default: DefaultMaxDeliver (-1 = unlimited attempts; relies on server default/unlimited behavior).
	MaxDeliver int `default:"-1" validate:"gte=-1"`

	// BatchSize is the max number of messages to pull per iterator request.
	// Default: DefaultBatchSize (1).
	BatchSize int `default:"1" validate:"gt=0"`

	// FetchTimeout is the max time to wait when pulling a batch.
	// Default: DefaultFetchTimeout (5s).
	FetchTimeout time.Duration `default:"5s" validate:"gt=0"`

	// MaxWaiting caps outstanding pull requests for each per-subject durable.
	// Default: DefaultMaxWaiting (2).
	MaxWaiting int `default:"2" validate:"gt=0"`

	// MaxAckPending limits the number of in-flight unacknowledged messages the server will allow
	// for each per-subject durable. If zero, the server default is used. This is most useful when using
	// ManualAck and background processing to cap concurrent in-flight work at the server layer.
	MaxAckPending int `validate:"gte=0"`

	// MaxConcurrentSubjects caps the total number of per-subject consumers/loops.
	// When exceeded, additional subjects are skipped with a warning and metric increment.
	MaxConcurrentSubjects int `validate:"gte=0"`

	// AckPolicy controls JetStream ack policy. Defaults to AckExplicitPolicy.
	AckPolicy jetstream.AckPolicy

	// InactiveThreshold is how long an idle consumer is kept by the server before cleanup.
	// Default: DefaultInactiveThreshold (24h).
	InactiveThreshold time.Duration `default:"24h" validate:"gt=0"`

	// PartitionRefreshMinInterval sets the minimum interval between forced claim refreshes
	// per partition when pull gating is enabled. This throttles KV hits when many
	// subjects are suppressed. Default: 500ms when zero.
	PartitionRefreshMinInterval time.Duration `default:"500ms" validate:"gt=0"`

	// Retry configures backoff behavior for control-plane operations.
	Retry RetryConfig

	// IteratorEscalationWindow defines the sliding time window used to aggregate
	// iterator failures for escalation detection. If zero, defaults to
	// DefaultIteratorEscalationWindow.
	IteratorEscalationWindow time.Duration `default:"60s" validate:"gt=0"`

	// IteratorEscalationThreshold is the number of iterator failures within the
	// escalation window that triggers a single escalation (consumer refresh).
	// If zero, defaults to DefaultIteratorEscalationThreshold.
	IteratorEscalationThreshold int `default:"3" validate:"gt=0"`

	// AllowWorkerIDChange controls whether workerID changes are allowed after initialization.
	// Default: false (immutable once set). Intended for controlled migrations only.
	AllowWorkerIDChange bool

	// IteratorFactory optionally overrides the iterator creation logic for testing or
	// advanced customization. When nil, a default factory is used that configures
	// heartbeat and expiry based on BatchSize and FetchTimeout.
	IteratorFactory func(cons jetstream.Consumer, batch int, expiry time.Duration) (jetstream.MessagesContext, error)
}