<?php
declare(strict_types=1);

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

use PpPhp\Core\Platform\Event\Streaming\Publisher\PublishRequest;
use PpPhp\Core\Platform\Event\Streaming\Publisher\PublisherManager;
use PpPhp\Core\Platform\Event\Streaming\Subscriber\StreamSubscriberRuntime;

final class StreamBroadcastRuntime
{
    private static ?self $instance = null;
    /** @var StreamBroadcastRequest[] */ private array $requests = [];
    /** @var StreamBroadcastResult[] */ private array $results = [];
    private StreamBroadcastStore $store;
    private function __construct(private readonly string $basePath) { $this->store = new StreamBroadcastStore($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 broadcast(StreamBroadcastRequest $request): StreamBroadcastResult
    {
        $started = microtime(true);
        $this->requests[] = $request;
        $subscribers = $this->matchingSubscribers($request);
        $deliveries = [];
        $publisher = PublisherManager::instance($this->basePath);
        $subscriberRuntime = StreamSubscriberRuntime::instance($this->basePath);
        foreach ($subscribers as $subscriber) {
            $subscriberId = (string) ($subscriber['subscriberId'] ?? '');
            if ($subscriberId === '') { continue; }
            $result = $publisher->publish(PublishRequest::create($request->publisherId(), $request->topic(), $request->type(), $request->payload(), $request->headers() + ['broadcastId' => $request->broadcastId(), 'subscriberId' => $subscriberId]));
            if (!$result->published()) { $deliveries[] = StreamBroadcastDelivery::skipped($request->broadcastId(), $subscriberId, $request->topic(), 'publish failed'); continue; }
            $subscriberRuntime->consume($subscriberId, 1, true);
            $deliveries[] = StreamBroadcastDelivery::delivered($request->broadcastId(), $subscriberId, $request->topic(), $result->messageId());
        }
        if ($deliveries === []) {
            foreach ($request->targetSubscribers() as $subscriberId) { $deliveries[] = StreamBroadcastDelivery::skipped($request->broadcastId(), (string) $subscriberId, $request->topic(), 'subscriber not found'); }
        }
        $result = StreamBroadcastResult::create($request->broadcastId(), $request->topic(), $deliveries, (microtime(true) - $started) * 1000);
        $this->results[] = $result;
        $this->save();
        return $result;
    }

    public function fanOut(string $publisherId, string $topic, string $type, array $payload = [], array $headers = []): StreamBroadcastResult
    { return $this->broadcast(StreamBroadcastRequest::create($publisherId, $topic, $type, $payload, $headers)); }

    public function hydrate(): void { $snapshot = $this->store->load(); $this->requests = []; foreach ($snapshot->requests() as $request) { if (is_array($request)) { $this->requests[] = StreamBroadcastRequest::fromArray($request); } } $this->results = []; foreach ($snapshot->results() as $result) { if (is_array($result)) { $this->results[] = StreamBroadcastResult::fromArray($result); } } }
    public function snapshot(): StreamBroadcastSnapshot { $requests = array_map(static fn (StreamBroadcastRequest $request): array => $request->toArray(), $this->requests); $results = array_map(static fn (StreamBroadcastResult $result): array => $result->toArray(), $this->results); return new StreamBroadcastSnapshot($requests, $results, StreamBroadcastStatistics::from($requests, $results), date('c')); }
    public function save(): StreamBroadcastSnapshot { return $this->store->save($this->snapshot()); }
    public function clear(): void { $this->requests = []; $this->results = []; $this->store->reset(); }
    public function store(): StreamBroadcastStore { return $this->store; }
    public function summary(): array { $snapshot = $this->snapshot(); return ['requests' => $snapshot->requests(), 'results' => $snapshot->results(), 'statistics' => $snapshot->statistics(), 'persisted' => $this->store->exists(), 'stateFile' => $this->store->stateFile()]; }

    private function matchingSubscribers(StreamBroadcastRequest $request): array
    {
        $subscribers = StreamSubscriberRuntime::instance($this->basePath)->summary()['subscribers'] ?? [];
        $targets = $request->targetSubscribers();
        return array_values(array_filter($subscribers, static function (array $subscriber) use ($request, $targets): bool {
            if (($subscriber['topic'] ?? '') !== $request->topic()) { return false; }
            if ($targets !== [] && !in_array((string) ($subscriber['subscriberId'] ?? ''), $targets, true)) { return false; }
            return ($subscriber['active'] ?? false) === true;
        }));
    }
    private static function detectBasePath(): string { $candidate = dirname(__DIR__, 5); $real = realpath($candidate); return $real !== false ? $real : $candidate; }
}
