kafka

type Consumer interface {
	// Subscribe subscribes the consumer to the configured topics.
	Subscribe() error
	// Poll polls for a single message. Returns nil if timeout is reached without a message.
	Poll(timeoutMs int) (*Message, error)
	// Consume starts consuming messages and invokes the handler for each message.
	// This is a blocking call that runs until the context is cancelled.
	Consume(ctx context.Context, handler MessageHandler) error
	// CommitMessage commits the offset of the given message.
	CommitMessage(msg *Message) error
	// Close gracefully shuts down the consumer.
	Close() error
}

type ConsumerConfig struct {
	Brokers          string
	Username         string
	Password         string
	SecurityProtocol string
	SASLMechanism    string
	CompressionCodec string
	MessageTimeoutMs int
	SocketTimeoutMs  int
	ClientID         string
	GroupID          string
	Topics           []string
	AutoOffsetReset  string
	PollTimeout      time.Duration
	AdditionalConfig map[string]string
}

type Header struct {
	Key   string
	Value []byte
}

type KafkaConsumerClient interface {
	SubscribeTopics(topics []string, rebalanceCb kafka.RebalanceCb) error
	Poll(timeoutMs int) kafka.Event
	CommitMessage(m *kafka.Message) ([]kafka.TopicPartition, error)
	Close() error
}

type KafkaProducerClient interface {
	Produce(msg *kafka.Message, deliveryChan chan kafka.Event) error
	Flush(timeoutMs int) int
	Close()
	Events() chan kafka.Event
}

type Logger interface {
	// OnDeliverySuccess is called when a message is successfully delivered to Kafka.
	OnDeliverySuccess(topic string, partition int32, offset kafka.Offset, key []byte)
	// OnDeliveryFailed is called when a message delivery fails.
	OnDeliveryFailed(topic string, key []byte, err error)
	// OnProduceError is called when the producer fails to enqueue a message.
	OnProduceError(topic string, key []byte, err error)
	// OnError is called for general Kafka errors (producer or consumer).
	OnError(component string, err kafka.Error)
	// OnRebalance is called when consumer group rebalancing occurs.
	OnRebalance(partitions []kafka.TopicPartition, assigned bool)
	// OnConsume is called when a message is consumed.
	OnConsume(topic string, partition int32, offset int64, key []byte)
	// OnConsumeError is called when the message handler returns an error.
	OnConsumeError(topic string, key []byte, err error)
	// OnCommitSuccess is called when an offset commit succeeds.
	OnCommitSuccess(topic string, partition int32, offset int64)
	// OnCommitError is called when an offset commit fails.
	OnCommitError(topic string, partition int32, offset int64, err error)
	// OnLog is called for general Kafka log messages.
	OnLog(level int, facility string, message string)
}

type Message struct {
	Topic     string
	Partition int32
	Offset    int64
	Key       []byte
	Value     []byte
	Headers   []Header
	Timestamp time.Time
}

type MessageHandler func(ctx context.Context, msg *Message) error

type NopLogger struct{}

type Producer interface {
	// Publish sends a message to the specified topic.
	Publish(ctx context.Context, topic string, key []byte, value []byte, headers []Header) error
	// PublishWithPartition sends a message to a specific partition of the topic.
	PublishWithPartition(ctx context.Context, topic string, partition int32, key []byte, value []byte, headers []Header) error
	// Flush waits for all outstanding produce requests to complete.
	Flush(timeoutMs int) int
	// Close gracefully shuts down the producer.
	Close()
}

type ProducerConfig struct {
	Brokers          string
	Username         string
	Password         string
	SecurityProtocol string
	SASLMechanism    string
	CompressionCodec string
	MessageTimeoutMs int
	SocketTimeoutMs  int
	AdditionalConfig map[string]string
}