<?php

use PHPUnit\Framework\TestCase;

/**
 * End-to-end tests that talk to a REAL Kafka broker.
 *
 * They only run when a broker is provided:
 *     KAFKA_BROKERS=127.0.0.1:9092 vendor/bin/phpunit
 *
 * Otherwise every test is skipped, so unit runs stay green without Kafka.
 */
class IntegrationRoundTripTest extends TestCase
{
    private ?string $brokers = null;

    protected function setUp(): void
    {
        if (!extension_loaded('rdkafka')) {
            $this->markTestSkipped('rdkafka extension not installed.');
        }

        $this->brokers = getenv('KAFKA_BROKERS') ?: null;
        if ($this->brokers === null) {
            $this->markTestSkipped('Set KAFKA_BROKERS to run integration tests.');
        }
    }

    public function testProduceThenConsume(): void
    {
        // Unique topic + payload per run so we never read stale messages.
        $topic   = 'kafkalib_it_' . bin2hex(random_bytes(4));
        $payload = ['ping' => bin2hex(random_bytes(4))];

        $config   = new KafkaConfig($this->brokers);
        $producer = new Producer($config);
        $producer->produce($topic, $payload, key: 'k1');

        $consumer = new Consumer($config, 'it-group-' . bin2hex(random_bytes(4)), [$topic]);

        $received = null;
        $consumer->consume(function (array $msg) use (&$received) {
            $received = $msg;
        }, maxMessages: 1, timeoutMs: 15000);
        $consumer->close();

        $this->assertNotNull($received, 'No message received from broker.');
        $this->assertSame($payload, json_decode($received['payload'], true));
        $this->assertSame('k1', $received['key']);
    }

    /**
     * The headline requirement: one message, two groups, two actions.
     * Both groups must receive the SAME message independently.
     */
    public function testSameMessageDeliveredToMultipleGroups(): void
    {
        $topic   = 'kafkalib_multi_' . bin2hex(random_bytes(4));
        $payload = ['id' => bin2hex(random_bytes(4))];

        $config = new KafkaConfig($this->brokers);

        // Create both groups BEFORE producing so their offsets start at 0.
        $groupA = new Consumer($config, 'group-a-' . bin2hex(random_bytes(4)), [$topic]);
        $groupB = new Consumer($config, 'group-b-' . bin2hex(random_bytes(4)), [$topic]);

        (new Producer($config))->produce($topic, $payload);

        $gotA = null;
        $groupA->consume(fn(array $m) => $gotA = json_decode($m['payload'], true), maxMessages: 1, timeoutMs: 15000);
        $groupA->close();

        $gotB = null;
        $groupB->consume(fn(array $m) => $gotB = json_decode($m['payload'], true), maxMessages: 1, timeoutMs: 15000);
        $groupB->close();

        $this->assertSame($payload, $gotA, 'group-a did not receive the message.');
        $this->assertSame($payload, $gotB, 'group-b did not receive the same message.');
    }
}