pubsub

type Backend string

type Config struct {
	// Backend specifies which backend to use
	Backend Backend

	// BufferSize is the channel buffer size for subscribers (default: 100)
	BufferSize int

	// OnFull specifies behavior when subscriber buffer is full
	OnFull OverflowPolicy

	// BatchSizeMax is the maximum number of messages per pipeline batch (default: 1000)
	// Only applicable to backends that support pipelining (e.g., Redis)
	BatchSizeMax int

	// Options contains backend-specific configuration
	Options map[string]any
}

type Factory func(cfg Config) (PubSub, error)

type Handler func(ctx context.Context, msg message.Message[any]) error

type HandlerMiddleware func(Handler) Handler

type LogFunc func(format string, args ...any)

type Metrics struct {
	// PublishTotal counts total publish operations per topic
	PublishTotal *prometheus.CounterVec

	// DeliveredTotal counts total messages delivered to subscribers per topic
	DeliveredTotal *prometheus.CounterVec

	// DroppedTotal counts messages dropped due to full buffers per topic
	DroppedTotal *prometheus.CounterVec

	// SubscribeTotal counts total subscribe operations per topic
	SubscribeTotal *prometheus.CounterVec

	// UnsubscribeTotal counts total unsubscribe operations per topic
	UnsubscribeTotal *prometheus.CounterVec

	// SubscribersGauge tracks current number of subscribers per topic
	SubscribersGauge *prometheus.GaugeVec

	// HandlerErrorTotal counts handler errors per topic
	HandlerErrorTotal *prometheus.CounterVec

	// PublishLatency tracks publish operation latency per topic
	PublishLatency *prometheus.HistogramVec

	// BatchSize tracks batch sizes for PublishBatch operations
	BatchSize *prometheus.HistogramVec

	// PipelineLatency tracks pipeline execution latency
	PipelineLatency *prometheus.HistogramVec

	// PipelineErrorTotal counts pipeline execution errors
	PipelineErrorTotal *prometheus.CounterVec
}

type OverflowPolicy int

type PubSub interface {
	Publisher
	Subscriber

	// Topics returns a list of all topics with active subscriptions
	Topics() []string

	// SubscriberCount returns the number of active subscribers for a topic
	SubscriberCount(topic string) int

	// Close gracefully shuts down the PubSub instance.
	// All subscriptions are closed and no more messages can be published.
	Close() error
}

type Publisher interface {
	// Publish sends a message to all subscribers of the topic.
	// Returns nil if there are no subscribers (message is silently dropped).
	Publish(ctx context.Context, topic string, msg message.Message[any]) error

	// PublishBatch sends multiple messages to all subscribers of the topic.
	// All messages are sent atomically if the backend supports it.
	PublishBatch(ctx context.Context, topic string, msgs []message.Message[any]) error
}

type PublisherMiddleware func(Publisher) Publisher

type RedisClient interface {
	redis.Cmdable
	Subscribe(ctx context.Context, channels ...string) *redis.PubSub
	PSubscribe(ctx context.Context, channels ...string) *redis.PubSub
}

type RedisOptions struct {
	// Addr is the Redis server address (default: "localhost:6379")
	Addr string

	// Password is the Redis password (optional)
	Password string

	// DB is the Redis database number (default: 0)
	DB int

	// Prefix is the key prefix for pubsub channels (default: "pubsub:")
	Prefix string

	// PoolSize is the maximum number of socket connections (default: 10)
	PoolSize int

	// ReadTimeout is the timeout for reading from Redis (default: 3s)
	ReadTimeout time.Duration

	// WriteTimeout is the timeout for writing to Redis (default: 3s)
	WriteTimeout time.Duration
}

type Subscriber interface {
	// Subscribe returns a channel that receives all messages published to the topic.
	// The channel is closed when:
	// - The context is cancelled
	// - Unsubscribe is called on the returned Subscription
	// - The PubSub instance is closed
	//
	// Each call to Subscribe creates a new independent subscription.
	// All subscriptions receive all messages (fan-out).
	Subscribe(ctx context.Context, topic string) (Subscription, error)

	// SubscribeWithHandler registers a handler for the topic.
	// The handler is called for each message received.
	// This is a blocking call that returns when:
	// - The context is cancelled
	// - The PubSub instance is closed
	// - An unrecoverable error occurs
	//
	// Handler errors are logged but do not stop the subscription.
	SubscribeWithHandler(ctx context.Context, topic string, handler Handler) error
}

type SubscriberMiddleware func(Subscriber) Subscriber

type Subscription interface {
	// ID returns the unique identifier of this subscription
	ID() string

	// Topic returns the topic this subscription is listening to
	Topic() string

	// Messages returns the channel that receives messages
	// The channel is closed when the subscription ends
	Messages() <-chan message.Message[any]

	// Unsubscribe cancels this subscription and closes the message channel
	Unsubscribe() error
}

type TypedHandler[T any] func(ctx context.Context, msg message.Message[T]) error

type TypedPubSub[T any] interface {
	TypedPublisher[T]
	TypedSubscriber[T]
	Topics() []string
	SubscriberCount(topic string) int
	Close() error
}

type TypedPublisher[T any] interface {
	// Publish sends a typed message to all subscribers
	Publish(ctx context.Context, topic string, data T) error

	// PublishBatch sends multiple typed messages
	PublishBatch(ctx context.Context, topic string, data []T) error
}

type TypedSubscriber[T any] interface {
	// Subscribe returns a channel that receives typed messages
	Subscribe(ctx context.Context, topic string) (TypedSubscription[T], error)

	// SubscribeWithHandler registers a typed handler
	SubscribeWithHandler(ctx context.Context, topic string, handler TypedHandler[T]) error
}

type TypedSubscription[T any] interface {
	ID() string
	Topic() string
	Messages() <-chan message.Message[T]
	Unsubscribe() error
}