Recurring Cleanup Task with Symfony Scheduler and Messenger Handler

<?php

namespace App\Scheduler;

use App\Message\PurgeExpiredTokensMessage;
use Symfony\Component\Scheduler\Attribute\AsSchedule;
use Symfony\Component\Scheduler\RecurringMessage;
use Symfony\Component\Scheduler\Schedule;
use Symfony\Component\Scheduler\ScheduleProviderInterface;
use Symfony\Component\Lock\LockFactory;
use Symfony\Contracts\Cache\CacheInterface;

#[AsSchedule('default')]
final class CleanupSchedule implements ScheduleProviderInterface
{
    private ?Schedule $schedule = null;

    public function __construct(
        private readonly CacheInterface $cache,
        private readonly LockFactory $lockFactory,
    ) {
    }

    public function getSchedule(): Schedule
    {
        return $this->schedule ??= (new Schedule())
            ->add(
                RecurringMessage::cron('0 3 * * *', new PurgeExpiredTokensMessage(1000)),
                RecurringMessage::every('6 hours', new PurgeExpiredTokensMessage(500)),
            )
            ->stateful($this->cache)
            ->lock($this->lockFactory->createLock('cleanup-schedule'));
    }
}

<?php

namespace App\MessageHandler;

use App\Message\PurgeExpiredTokensMessage;
use App\Repository\ApiTokenRepository;
use Psr\Log\LoggerInterface;
use Symfony\Component\Messenger\Attribute\AsMessageHandler;

#[AsMessageHandler]
final class PurgeExpiredTokensHandler
{
    public function __construct(
        private readonly ApiTokenRepository $tokens,
        private readonly LoggerInterface $logger,
    ) {
    }

    public function __invoke(PurgeExpiredTokensMessage $message): void
    {
        $total = 0;

        do {
            $deleted = $this->tokens->deleteExpiredBatch($message->batchSize);
            $total += $deleted;
        } while ($deleted === $message->batchSize);

        $this->logger->info('Purged expired API tokens', ['count' => $total]);
    }
}

<?php

namespace App\Repository;

use App\Entity\ApiToken;
use Doctrine\Bundle\DoctrineBundle\Repository\ServiceEntityRepository;
use Doctrine\Persistence\ManagerRegistry;

class ApiTokenRepository extends ServiceEntityRepository
{
    public function __construct(ManagerRegistry $registry)
    {
        parent::__construct($registry, ApiToken::class);
    }

    public function deleteExpiredBatch(int $limit): int
    {
        $conn = $this->getEntityManager()->getConnection();

        $sql = <<<SQL
            DELETE FROM api_token
            WHERE id IN (
                SELECT id FROM api_token
                WHERE expires_at < :now
                ORDER BY expires_at ASC
                LIMIT :limit
            )
        SQL;

        return (int) $conn->executeStatement($sql, [
            'now' => (new \DateTimeImmutable())->format('Y-m-d H:i:s'),
            'limit' => $limit,
        ], [
            'limit' => \PDO::PARAM_INT,
        ]);
    }
}