<?php
declare(strict_types=1);

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

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

final class PublisherPipeline
{
    public function __construct(
        private readonly MessageBus $bus,
        private readonly PublisherValidator $validator = new PublisherValidator()
    ) {}

    public function publish(PublishRequest $request): PublishResult
    {
        $started = microtime(true);
        $validation = $this->validator->validate($request);

        if (!$validation['ok']) {
            return PublishResult::failure(
                $request->topic(),
                implode('; ', $validation['issues']),
                (microtime(true) - $started) * 1000
            );
        }

        if ($request->delaySeconds() > 0) {
            $headers = $request->headers() + [
                'delaySeconds' => $request->delaySeconds(),
                'availableAt' => date('c', time() + $request->delaySeconds()),
            ];
        } else {
            $headers = $request->headers();
        }

        $headers += [
            'publisherId' => $request->publisherId(),
            'priority' => $request->priority(),
        ];

        $envelope = $this->bus->publish($request->topic(), $request->type(), $request->payload(), $headers);
        $bytes = strlen(json_encode($request->payload()) ?: '');

        return PublishResult::success(
            $envelope->message()->messageId(),
            $request->topic(),
            (microtime(true) - $started) * 1000,
            $bytes
        );
    }
}
