<?php
// Licensed to the Apache Software Foundation (ASF) under one
// or more contributor license agreements.  See the NOTICE file
// distributed with this work for additional information
// regarding copyright ownership.  The ASF licenses this file
// to you under the Apache License, Version 2.0 (the
// "License"); you may not use this file except in compliance
// with the License.  You may obtain a copy of the License at
//
//   http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing,
// software distributed under the License is distributed on an
// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
// KIND, either express or implied.  See the License for the
// specific language governing permissions and limitations
// under the License.

declare(strict_types=1);

use Iggy\Client as IggyClient;
use Iggy\ReceiveMessage;
use PHPUnit\Framework\Assert;

function assert_true(bool $condition, string $message = 'expected condition to be true'): void
{
    Assert::assertTrue($condition, $message);
}

function assert_same(mixed $expected, mixed $actual, string $message = ''): void
{
    Assert::assertSame($expected, $actual, $message);
}

function assert_not_null(mixed $value, string $message = 'expected value not to be null'): void
{
    Assert::assertNotNull($value, $message);
}

function assert_instance_of(string $expected, mixed $actual, string $message = ''): void
{
    Assert::assertInstanceOf($expected, $actual, $message);
}

function assert_null(mixed $value, string $message = 'expected value to be null'): void
{
    Assert::assertNull($value, $message);
}

function assert_count(int $expected, array $actual, string $message = ''): void
{
    Assert::assertCount($expected, $actual, $message);
}

function assert_throws(callable $callback, ?string $messageContains = null): Throwable
{
    try {
        $callback();
    } catch (Throwable $throwable) {
        if ($messageContains !== null && !str_contains($throwable->getMessage(), $messageContains)) {
            Assert::fail(
                'expected exception message to contain ' . var_export($messageContains, true)
                . ', got ' . var_export($throwable->getMessage(), true)
            );
        }

        Assert::assertInstanceOf(Throwable::class, $throwable);

        return $throwable;
    }

    Assert::fail('expected callable to throw');
}

function unique_name(string $prefix): string
{
    return $prefix . '-' . bin2hex(random_bytes(4));
}

function env_or_default(string $name, string $default): string
{
    $value = getenv($name);

    return $value === false || $value === '' ? $default : $value;
}

function server_host(): string
{
    return env_or_default('IGGY_HOST', '127.0.0.1');
}

function server_port(): int
{
    return (int) env_or_default('IGGY_PORT', '8090');
}

function wait_for_server(string $host, int $port, int $timeoutSeconds = 30): void
{
    $deadline = microtime(true) + $timeoutSeconds;
    $lastError = null;

    while (microtime(true) < $deadline) {
        $socket = @fsockopen($host, $port, $errno, $errstr, 1.0);
        if (is_resource($socket)) {
            fclose($socket);

            return;
        }

        $lastError = trim($errstr !== '' ? $errstr : (string) $errno);
        usleep(250_000);
    }

    Assert::fail("Iggy server was not reachable at {$host}:{$port}" . ($lastError !== null ? " ({$lastError})" : ''));
}

function new_client(): IggyClient
{
    $client = new IggyClient(server_host() . ':' . server_port());
    $client->connect();
    $client->loginUser(env_or_default('IGGY_USERNAME', 'iggy'), env_or_default('IGGY_PASSWORD', 'iggy'));

    return $client;
}

function new_connection_string_client(): IggyClient
{
    $host = server_host();
    $port = server_port();
    $username = rawurlencode(env_or_default('IGGY_USERNAME', 'iggy'));
    $password = rawurlencode(env_or_default('IGGY_PASSWORD', 'iggy'));

    $client = IggyClient::fromConnectionString("iggy+tcp://{$username}:{$password}@{$host}:{$port}");
    $client->connect();

    return $client;
}

function create_stream_and_topic(object $client, string $stream, string $topic, int $partitions = 1): void
{
    $client->createStream($stream);
    $client->createTopic($stream, $topic, $partitions, null, null, null, null);
}

function cleanup_topic(object $client, string|int $stream, string|int $topic): void
{
    try {
        $client->deleteTopic($stream, $topic);
    } catch (Throwable) {
    }
}

function cleanup_stream(object $client, string|int $stream): void
{
    try {
        $client->deleteStream($stream);
    } catch (Throwable) {
    }
}

function cleanup_stream_with_topics(object $client, string|int $stream, array $topics): void
{
    foreach ($topics as $topic) {
        cleanup_topic($client, $stream, $topic);
    }

    cleanup_stream($client, $stream);
}

function collect_payloads(array $messages): array
{
    return array_map(static fn (ReceiveMessage $message): string => $message->payload(), $messages);
}

function collect_offsets(array $messages): array
{
    return array_map(static fn (ReceiveMessage $message): int => $message->offset(), $messages);
}

function micros(int $seconds): int
{
    return $seconds * 1_000_000;
}

if (!extension_loaded('iggy-php')) {
    Assert::fail('The iggy-php extension is not loaded.');
}

wait_for_server(server_host(), server_port());