kstate

type BatchKeyValueStore[K comparable, V any] interface {
	KeyValueStore[K, V]

	// SetBatch stores multiple key-value pairs atomically
	SetBatch(ctx context.Context, entries []kprocessor.KV[K, V]) error

	// GetBatch retrieves multiple values by keys
	GetBatch(ctx context.Context, keys []K) ([]kprocessor.KV[K, V], error)

	// DeleteBatch removes multiple keys atomically
	DeleteBatch(ctx context.Context, keys []K) error
}

type BatchKeyValueStoreRef[K comparable, V any] struct {
	// contains filtered or unexported fields
}

type BatchingKeyValueStoreRestoreCallback[K comparable, V any] struct {
	// contains filtered or unexported fields
}

type ChangeloggingBatchKeyValueStore[K comparable, V any] struct {
	*ChangeloggingKeyValueStore[K, V]
	// contains filtered or unexported fields
}

type ChangeloggingKeyValueStore[K comparable, V any] struct {
	// contains filtered or unexported fields
}

type KeyValueStore[K comparable, V any] interface {
	StateStore

	// Get retrieves a value by key
	// Returns (value, true, nil) if found
	// Returns (zero, false, nil) if not found
	// Returns (zero, false, err) on error
	Get(ctx context.Context, key K) (V, bool, error)

	// Set stores a key-value pair
	Set(ctx context.Context, key K, value V) error

	// Delete removes a key
	Delete(ctx context.Context, key K) error

	// Range returns an iterator over a range of keys [from, to)
	Range(ctx context.Context, from, to K) iter.Seq2[K, V]

	// All returns an iterator over all keys
	All(ctx context.Context) iter.Seq2[K, V]
}

type KeyValueStoreRef[K comparable, V any] struct {
	// contains filtered or unexported fields
}

type KeyValueStoreRestoreCallback[K comparable, V any] struct {
	// contains filtered or unexported fields
}

type NoOpRestoreListener struct{}

type StateRestoreCallback interface {
	// Restore a single record (called if RestoreBatch not implemented)
	// value = nil indicates tombstone (deletion)
	Restore(key, value []byte) error

	// RestoreBatch restores a batch of records for efficiency
	// Implementations should check for tombstones (record.Value == nil)
	RestoreBatch(records []*kgo.Record) error
}

type StateRestoreListener interface {
	// OnRestoreStart is called when restoration begins
	// startingOffset: First offset to restore (usually checkpoint offset + 1)
	// endingOffset: EXCLUSIVE ending offset (high water mark, one past the last offset to restore)
	//               Matches Kafka Streams semantics: restoration completes when reaching endingOffset - 1
	OnRestoreStart(topicPartition checkpoint.TopicPartition, storeName string, startingOffset, endingOffset int64)

	// OnBatchRestored is called after each batch is applied to the store
	// batchEndOffset: Last offset in the batch
	// numRestored: Number of records in this batch
	OnBatchRestored(topicPartition checkpoint.TopicPartition, storeName string, batchEndOffset int64, numRestored int64)

	// OnRestoreEnd is called when restoration completes
	// totalRestored: Total number of records restored
	OnRestoreEnd(topicPartition checkpoint.TopicPartition, storeName string, totalRestored int64)
}

type StateStore interface {
	// Name returns the store name
	Name() string

	// Init initializes the store with processor context
	Init(ctx kprocessor.ProcessorContextInternal) error

	// Flush persists any cached data
	Flush(ctx context.Context) error

	// Close closes the store
	Close() error

	// Persistent returns true if the store persists data to disk
	// Returns false for in-memory stores
	// CRITICAL: Only persistent stores should be checkpointed
	// Matches Kafka Streams' StateStore.persistent()
	Persistent() bool
}

type StateStoreGetter interface {
	GetStateStore() StateStore
}

type StoreBackend interface {
	StateStore
	Set(k, v []byte) error
	Get(k []byte) ([]byte, error)
	Delete(k []byte) error
	Range(lower, upper []byte) iter.Seq2[[]byte, []byte]
	All() iter.Seq2[[]byte, []byte]
}

type StoreBuilder[T StateStore] interface {
	// WithChangelogEnabled enables changelog topic for this store
	// config contains Kafka topic-level configuration (optional, can be nil)
	//
	// Common config keys:
	//   - "retention.ms": Changelog retention time in milliseconds
	//   - "cleanup.policy": "compact" or "delete"
	//   - "segment.ms": Segment size for time-based retention
	//
	// Default (if config is nil): Uses application-level defaults
	WithChangelogEnabled(config map[string]string) StoreBuilder[T]

	// WithChangelogDisabled disables changelog topic for this store
	// Use for stores that don't need durability (e.g., remote stores like S3, BigTable)
	WithChangelogDisabled() StoreBuilder[T]

	// Build constructs the state store with configured options
	// If changelog is enabled, wraps store in ChangeloggingKeyValueStore
	// Returns error if configuration is invalid or store initialization fails
	Build() (T, error)

	// Name returns the store name
	Name() string

	// LogConfig returns the changelog topic configuration
	// Returns nil if changelog is disabled
	LogConfig() map[string]string

	// ChangelogEnabled returns whether changelog is enabled for this store
	ChangelogEnabled() bool

	// RestoreCallback returns the callback for restoring this store from changelog
	// Returns nil if changelog is disabled
	RestoreCallback() StateRestoreCallback
}

type TypeErasedStoreBuilder interface {
	// BuildStateStore constructs the state store (type-erased version)
	BuildStateStore() (StateStore, error)

	// Name returns the store name
	Name() string

	// LogConfig returns the changelog topic configuration
	LogConfig() map[string]string

	// ChangelogEnabled returns whether changelog is enabled
	ChangelogEnabled() bool

	// RestoreCallback returns the restore callback
	RestoreCallback() StateRestoreCallback
}

type WindowedStore[K comparable, V any] interface {
	StateStore

	// Set stores a value for a key at a specific timestamp
	Set(ctx context.Context, key K, value V, timestamp int64) error

	// Get retrieves a value by key at a specific timestamp
	Get(ctx context.Context, key K, timestamp int64) (V, bool, error)

	// Fetch retrieves values for a key within a time range
	Fetch(ctx context.Context, key K, timeFrom, timeTo int64) iter.Seq2[K, V]

	// Delete removes a value for a key at a specific timestamp
	Delete(ctx context.Context, key K, timestamp int64) error
}

type WindowedStoreRef[K comparable, V any] struct {
	// contains filtered or unexported fields
}