kafka

type Broker interface {
	// Init initializes broker
	Init(ctx context.Context, cfg *BrokerConfig) error
	// AddProducer adds a producer with configuration
	AddProducer(ctx context.Context, topic *TopicConfig, cfg *ProducerConfig) (Producer, error)
	// AddSubscriber adds a subscriber with configuration
	AddSubscriber(ctx context.Context, topic *TopicConfig, cfg *SubscriberConfig, handlers ...HandlerFn) error
	// DeclareTopics declares topics in kafka broker
	// must be executed after all producer and subscribers added
	DeclareTopics(ctx context.Context) error
	// Start starts listening
	Start(ctx context.Context) error
	// Close closes broker
	Close(ctx context.Context)
}

type BrokerConfig struct {
	ClientId          string `mapstructure:"client_id"`           // ClientId identifies client
	TopicAutoCreation bool   `mapstructure:"topic_auto_creation"` // TopicAutoCreation if true topics are created by code (otherwise must be preliminary declared)
	Url               string // Url comma separated host-port pairs ("localhost:9092,localhost:9093")
	Sasl              Sasl   // Sasl configuration
}

type HandlerFn func(payload []byte) error

type Message struct {
	// Ctx allows passing context from producer to subscriber
	Ctx *jet.RequestContext `json:"ctx"`
	// Key is a kafka message key which allow to specify target kafka partition
	Key string `json:"key"`
	// Payload arbitrary data
	Payload any `json:"payload"`
}

type MessageT[T any] struct {
	// Ctx allows passing context from producer to subscriber
	Ctx *jet.RequestContext `json:"ctx"`
	// Key is a kafka message key which allow to specify target kafka partition
	Key string `json:"key"`
	// Payload arbitrary data
	Payload T `json:"payload"`
}

type Producer interface {
	// Send sends a message to broker
	Send(ctx context.Context, key string, payload interface{}) error
	// SendMany sends bulk of messages to broker
	SendMany(ctx context.Context, messages ...*Message) error
}

type ProducerConfig struct {
	BatchSize    *int
	BatchTimeout *time.Duration
	RequiredAcks *int
	MaxAttempts  *int
	Async        bool
	RetryTimes   *int
	RetryTimeout *time.Duration
}

type ProducerConfigBuilder interface {
	// BatchTimeout sets batch timeout value (default: 1s)
	BatchTimeout(to time.Duration) ProducerConfigBuilder
	// BatchSize sets batch size (default: 100)
	BatchSize(size int) ProducerConfigBuilder
	// Async if true, WriteMessages call will never block but errors aren't returned (default: false)
	Async(v bool) ProducerConfigBuilder
	// Retry sets retry params (default: time=3, timeout = 1s)
	Retry(time int, timeout time.Duration) ProducerConfigBuilder
	// Build builds config
	Build() *ProducerConfig
}

type Sasl struct {
	Enabled  bool   // Enabled true if Sasl must be applied
	Type     string // Type of mechanism (plain, sha256, sha512)
	Username string // Username
	Password string // Password
}

type SubscriberConfig struct {
	GroupId          string
	BatchTimeout     *time.Duration
	MaxWait          *time.Duration
	CommitInterval   *time.Duration
	Workers          *int
	MaxAttempts      *int
	StartOffset      *int64
	JoinGroupBackoff *time.Duration
	Logging          bool
}

type SubscriberConfigBuilder interface {
	// GroupId allows load balancing messages within the same group
	GroupId(groupId string) SubscriberConfigBuilder
	// BatchTimeout sets timeout of batch fetching from kafka (default: 10s)
	BatchTimeout(to time.Duration) SubscriberConfigBuilder
	// MaxWait sets maximum amount of time to wait for new data to come when fetching batches (default: 10s)
	MaxWait(to time.Duration) SubscriberConfigBuilder
	// CommitInterval sets interval to commit to kafka (default: sync)
	CommitInterval(to time.Duration) SubscriberConfigBuilder
	// Workers sets number of workers (default: 4)
	Workers(num int) SubscriberConfigBuilder
	// StartOffset determines from which offset a new group starts to consume. it must be set to one of FirstOffset = -2 or LastOffset = -1 (Default: FirstOffset)
	// Only used when GroupID is set
	StartOffset(v int64) SubscriberConfigBuilder
	// JoinGroupBackoff optionally sets the length of time to wait between re-joining
	JoinGroupBackoff(t time.Duration) SubscriberConfigBuilder
	// Logging if true subscriber logging enabled
	Logging(v bool) SubscriberConfigBuilder
	// Build builds config
	Build() *SubscriberConfig
}

type TopicBuilder interface {
	// WithPartitionNum setting num of partitions
	WithPartitionNum(num int) TopicBuilder
	// WithParams setting additional params
	WithParams(params ...TopicParam) TopicBuilder
	// Build builds config
	Build() *TopicConfig
}

type TopicConfig struct {
	// Topic name
	Topic string
	// Partitions number of partitions
	Partitions *int
	// Configs config params
	Configs map[string]string
}

type TopicParam struct {
	Name  string
	Value string
}