<?php

/**
 * KafkaConfig
 *
 * Central place for broker connection details and shared rdkafka
 * configuration. Keeping this in one class means the Producer and the
 * Consumer always agree on brokers/timeouts, and callers only have to
 * pass the pieces that actually change (topic, group, etc.).
 *
 * Requires the php-rdkafka extension (pecl install rdkafka).
 */
class KafkaConfig
{
    /** @var string Comma separated list of host:port brokers. */
    private string $brokers;

    /** @var array<string,string> Extra librdkafka properties to merge in. */
    private array $extra;

    /**
     * @param string               $brokers e.g. "127.0.0.1:9092" or "b1:9092,b2:9092"
     * @param array<string,string> $extra   any additional librdkafka config
     *                                       (e.g. security.protocol, sasl.*)
     */
    public function __construct(string $brokers = '127.0.0.1:9092', array $extra = [])
    {
        if ($brokers === '') {
            throw new \InvalidArgumentException('Kafka brokers list cannot be empty.');
        }
        $this->brokers = $brokers;
        $this->extra   = $extra;
    }

    public function getBrokers(): string
    {
        return $this->brokers;
    }

    /**
     * Build a base \RdKafka\Conf pre-filled with the shared settings.
     * Both producer and consumer start from this.
     */
    public function baseConf(): \RdKafka\Conf
    {
        $conf = new \RdKafka\Conf();
        $conf->set('metadata.broker.list', $this->brokers);
        // Log connection errors instead of dying silently.
        $conf->set('log_level', (string) LOG_WARNING);

        foreach ($this->extra as $key => $value) {
            $conf->set($key, (string) $value);
        }

        return $conf;
    }
}