<?php

use PHPUnit\Framework\TestCase;

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

    public function testEmptyGroupIdThrows(): void
    {
        $this->expectException(\InvalidArgumentException::class);
        new Consumer(new KafkaConfig('127.0.0.1:9092'), '', ['orders']);
    }

    public function testEmptyTopicsThrows(): void
    {
        $this->expectException(\InvalidArgumentException::class);
        new Consumer(new KafkaConfig('127.0.0.1:9092'), 'grp', []);
    }

    public function testGetGroupId(): void
    {
        $consumer = new Consumer(new KafkaConfig('127.0.0.1:9092'), 'email-service', ['orders']);
        $this->assertSame('email-service', $consumer->getGroupId());
    }

    /**
     * normalise() is private; verify via reflection that an \RdKafka\Message
     * is converted to the flat array shape handlers receive.
     */
    public function testNormaliseShapesMessage(): void
    {
        $consumer = new Consumer(new KafkaConfig('127.0.0.1:9092'), 'grp', ['orders']);

        $message            = new \RdKafka\Message();
        $message->topic_name = 'orders';
        $message->partition  = 2;
        $message->offset     = 57;
        $message->key        = 'order-1001';
        $message->payload    = '{"order_id":1001}';
        $message->headers    = ['source' => 'checkout'];

        $ref = new \ReflectionMethod(Consumer::class, 'normalise');
        $ref->setAccessible(true);

        $out = $ref->invoke($consumer, $message);

        $this->assertSame('orders', $out['topic']);
        $this->assertSame(2, $out['partition']);
        $this->assertSame(57, $out['offset']);
        $this->assertSame('order-1001', $out['key']);
        $this->assertSame('{"order_id":1001}', $out['payload']);
        $this->assertSame(['source' => 'checkout'], $out['headers']);
    }
}