<?php

declare(strict_types=1);

namespace App\EventSubscriber\Messenger;

use Symfony\Component\EventDispatcher\EventSubscriberInterface;
use Symfony\Component\Messenger\Event\WorkerMessageReceivedEvent;
use Symfony\Component\Messenger\Event\WorkerRunningEvent;
use Symfony\Component\Messenger\Event\WorkerStoppedEvent;
use Symfony\Component\Messenger\Handler\BatchHandlerInterface;
use Symfony\Component\Messenger\Handler\HandlersLocatorInterface;
use Symfony\Component\Messenger\Worker;

class FlushBatchesOnWorkerEventsListener implements EventSubscriberInterface
{
    /** @var BatchHandlerInterface[][] */
    private array $batchHandlersToFlush = [];

    private \Closure $ackCall;

    public function __construct(private readonly HandlersLocatorInterface $handlersLocator)
    {
    }

    public function onWorkerRunning(WorkerRunningEvent $event): void
    {
        if ($event->isWorkerIdle()) {
            $this->flush(false);
            $this->ack($event->getWorker());
        }
    }

    public function onWorkerStopped(WorkerStoppedEvent $event): void
    {
        $this->flush(true);
        $this->ack($event->getWorker());
    }

    public function onWorkerMessageReceived(WorkerMessageReceivedEvent $event): void
    {
        $messageClass = $event->getEnvelope()->getMessage()::class;

        if (isset($this->batchHandlersToFlush[$messageClass])) {
            return;
        }

        $this->batchHandlersToFlush[$messageClass] = [];

        foreach ($this->handlersLocator->getHandlers($event->getEnvelope()) as $handlerDescriptor) {
            if ($batchHandler = $handlerDescriptor->getBatchHandler()) {
                $this->batchHandlersToFlush[$messageClass][] = $batchHandler;
            }
        }
    }

    private function ack(Worker $worker): void
    {
        if (!isset($this->ackCall)) {
            $this->ackCall = \Closure::bind(static fn() => $worker->ack(), null, $worker);
        }

        ($this->ackCall)();
    }

    private function flush(bool $force): void
    {
        foreach ($this->batchHandlersToFlush as $handlers) {
            foreach ($handlers as $handler) {
                $handler->flush($force);
            }
        }
    }

    public static function getSubscribedEvents(): array
    {
        return [
            WorkerRunningEvent::class         => ['onWorkerRunning'],
            WorkerStoppedEvent::class         => ['onWorkerStopped'],
            WorkerMessageReceivedEvent::class => ['onWorkerMessageReceived']
        ];
    }
}

    App\EventSubscriber\Messenger\FlushBatchesOnWorkerEventsListener:
        arguments:
            $handlersLocator: '@command_bus.messenger.handlers_locator'