<?php
declare(strict_types=1);

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

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

final class StreamPersistenceRuntime
{
    private static ?self $instance = null;
    private StreamEventLog $log;
    private array $replays = [];
    private StreamPersistenceStore $store;
    private function __construct(private readonly string $basePath) { $this->log = new StreamEventLog(); $this->store = new StreamPersistenceStore($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 persistMessage(array $message): StreamPersistedEvent { $event=$this->log->append($message); $this->save(); return $event; }
    public function persistBus(): int { $messages = MessageBus::instance($this->basePath)->summary()['messages'] ?? []; $existing=[]; foreach($this->log->toArray() as $event){$existing[(string)($event['messageId']??'')]=true;} $persisted=0; foreach($messages as $envelope){$message=(array)($envelope['message']??[]); $messageId=(string)($message['messageId']??''); if($messageId===''||isset($existing[$messageId])) continue; $this->log->append($message); $existing[$messageId]=true; $persisted++;} $this->save(); return $persisted; }
    public function replay(StreamReplayRequest $request): StreamReplayResult { $started=microtime(true); $events=$this->log->replay($request->topic(), $request->fromSequence(), $request->limit()); $result=StreamReplayResult::create($request,$events,(microtime(true)-$started)*1000); $this->replays[]=$result; $this->save(); return $result; }
    public function hydrate(): void { $snapshot=$this->store->load(); if($snapshot->events()!==[]) $this->log=StreamEventLog::fromArray($snapshot->events()); $this->replays=[]; foreach($snapshot->replays() as $replay){ if(is_array($replay)) $this->replays[]=StreamReplayResult::fromArray($replay); } }
    public function snapshot(): StreamPersistenceSnapshot { $events=$this->log->toArray(); $replays=array_map(static fn(StreamReplayResult $replay): array => $replay->toArray(), $this->replays); return new StreamPersistenceSnapshot($events,$replays,StreamPersistenceStatistics::from($events,$replays),date('c')); }
    public function save(): StreamPersistenceSnapshot { return $this->store->save($this->snapshot()); }
    public function clear(): void { $this->log=new StreamEventLog(); $this->replays=[]; $this->store->reset(); }
    public function store(): StreamPersistenceStore { return $this->store; }
    public function summary(): array { $snapshot=$this->snapshot(); return ['events'=>$snapshot->events(),'replays'=>$snapshot->replays(),'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; }
}
