<?php
declare(strict_types=1);

namespace PpPhp\Core\Platform\Event\Streaming\Subscriber;

use PpPhp\Core\Platform\Event\Streaming\MessageBus;

final class StreamSubscriberRuntime
{
    private static ?self $instance = null;

    private StreamSubscriberCollection $subscribers;
    /** @var StreamSubscriberResult[] */
    private array $results = [];
    private StreamSubscriberStore $store;

    private function __construct(private readonly string $basePath)
    {
        $this->subscribers = new StreamSubscriberCollection();
        $this->store = new StreamSubscriberStore($basePath);
        $this->hydrate();
    }

    public static function instance(?string $basePath = null): self
    {
        if (self::$instance === null) {
            $basePath ??= self::detectBasePath();
            self::$instance = new self($basePath);
        }

        return self::$instance;
    }

    public static function resetInstance(?string $basePath = null): self
    {
        $basePath ??= self::detectBasePath();
        self::$instance = new self($basePath);

        return self::$instance;
    }

    public function register(string $subscriberId, string $topic): StreamSubscriber
    {
        $subscriber = $this->subscribers->register($subscriberId, $topic);
        $this->save();

        return $subscriber;
    }

    public function consume(string $subscriberId, int $limit = 10, bool $autoAck = true): StreamSubscriberResult
    {
        $started = microtime(true);
        $subscriber = $this->subscribers->get($subscriberId);

        if ($subscriber === null) {
            $result = StreamSubscriberResult::create($subscriberId, '', 0, 0, 1, [], (microtime(true) - $started) * 1000);
            $this->results[] = $result;
            $this->save();

            return $result;
        }

        $messages = MessageBus::instance($this->basePath)->messagesForTopic($subscriber->topic());
        $offset = $subscriber->offset();
        $slice = array_slice($messages, $offset, max(1, $limit));

        $consumed = 0;
        $acknowledged = 0;
        $failed = 0;
        $messageRows = [];

        foreach ($slice as $envelope) {
            $message = $envelope->message();
            $messageId = $message->messageId();

            if (!$this->subscribers->receive($subscriberId, $messageId)) {
                $failed++;
                continue;
            }

            $consumed++;
            $messageRows[] = $message->toArray();

            if ($autoAck && $this->subscribers->acknowledge($subscriberId, $messageId)) {
                $acknowledged++;
            }
        }

        $result = StreamSubscriberResult::create(
            $subscriberId,
            $subscriber->topic(),
            $consumed,
            $acknowledged,
            $failed,
            $messageRows,
            (microtime(true) - $started) * 1000
        );

        $this->results[] = $result;
        $this->save();

        return $result;
    }

    public function acknowledge(string $subscriberId, ?string $messageId = null): bool
    {
        $ok = $this->subscribers->acknowledge($subscriberId, $messageId);
        $this->save();

        return $ok;
    }

    public function pause(string $subscriberId): bool
    {
        $ok = $this->subscribers->pause($subscriberId);
        $this->save();

        return $ok;
    }

    public function resume(string $subscriberId): bool
    {
        $ok = $this->subscribers->resume($subscriberId);
        $this->save();

        return $ok;
    }

    public function hydrate(): void
    {
        $snapshot = $this->store->load();

        if ($snapshot->subscribers() !== []) {
            $this->subscribers = StreamSubscriberCollection::fromArray($snapshot->subscribers());
        }

        $this->results = [];
        foreach ($snapshot->results() as $result) {
            if (is_array($result)) {
                $this->results[] = StreamSubscriberResult::fromArray($result);
            }
        }
    }

    public function snapshot(): StreamSubscriberSnapshot
    {
        $subscribers = $this->subscribers->toArray();
        $results = array_map(static fn (StreamSubscriberResult $result): array => $result->toArray(), $this->results);

        return new StreamSubscriberSnapshot(
            $subscribers,
            $results,
            StreamSubscriberStatistics::from($subscribers, $results),
            date('c')
        );
    }

    public function save(): StreamSubscriberSnapshot
    {
        return $this->store->save($this->snapshot());
    }

    public function clear(): void
    {
        $this->subscribers = new StreamSubscriberCollection();
        $this->results = [];
        $this->store->reset();
    }

    public function store(): StreamSubscriberStore { return $this->store; }

    public function summary(): array
    {
        $snapshot = $this->snapshot();

        return [
            'subscribers' => $snapshot->subscribers(),
            'results' => $snapshot->results(),
            'statistics' => $snapshot->statistics(),
            'persisted' => $this->store->exists(),
            'stateFile' => $this->store->stateFile(),
        ];
    }

    private static function detectBasePath(): string
    {
        $candidate = dirname(__DIR__, 5);
        $real = realpath($candidate);
        return $real !== false ? $real : $candidate;
    }
}
