<?php

declare(strict_types=1);

namespace Apexl\Io\Jobs\Services;

use Apexl\Io\Collections\LoaderSortCollection;
use Apexl\Io\Dtos\LoaderCondition;
use Apexl\Io\Dtos\LoaderSort;
use Apexl\Io\Jobs\Entities\QueueJob;
use Apexl\Io\Jobs\Helpers\QueueJobTask;
use Apexl\Io\Jobs\Interfaces\QueuedTaskFinderInterface;
use Apexl\Vault\Enums\SortDirection;
use Generator;

class QueuedTaskFinderDb implements QueuedTaskFinderInterface
{
    private const int BATCH = 1000;

    public function tasks(?int $limit = 10): Generator
    {
        $batch = min($limit, self::BATCH);
        $remaining = $limit;

        // Keyset (seek) pagination cursor: the [sequence, id] of the last row
        // we handed out. Null on the first pass so nothing is excluded.
        //
        // We order by sequence (jobs MUST run in sequence order within a
        // handler) with id as a unique tiebreaker — sequence is only unique
        // per handler, and external services supply their own sequences, so id
        // and sequence are not guaranteed to share an order. The cursor is
        // therefore the full tuple, seeked past with the standard row-wise
        // comparison expanded into an OR (MariaDB range-optimises this form;
        // it does not optimise the (a, b) > (x, y) row-constructor form).
        $cursor = null;

        while ($remaining > 0) {
            $take = min($batch, $remaining);

            $conditions = [
                LoaderCondition::isNull('processed'),
                LoaderCondition::isNull('failed'),
                LoaderCondition::equals('locked', 0),
                // Deferred / scheduled-retry jobs (run_after in the future) are
                // skipped until due. Legacy rows default to 0 = due now.
                LoaderCondition::lessThanOrEqual('run_after', time()),
            ];

            if ($cursor !== null) {
                [$afterSequence, $afterId] = $cursor;

                // Seek past the last (sequence, id) we returned instead of using
                // OFFSET, so each batch is an index range scan whose cost is
                // independent of how many processed rows sit behind the cursor.
                $conditions[] = LoaderCondition::raw(
                    '(sequence > :cursor_sequence_gt'
                    . ' OR (sequence = :cursor_sequence_eq AND id > :cursor_id_gt))',
                    [
                        'cursor_sequence_gt' => $afterSequence,
                        'cursor_sequence_eq' => $afterSequence,
                        'cursor_id_gt' => $afterId,
                    ],
                );
            }

            $jobs = QueueJob::findMultiple(
                conditions: $conditions,
                limit: $take,
                sorts: LoaderSortCollection::createFromArray([
                    new LoaderSort('sequence', SortDirection::ASC),
                    new LoaderSort('id', SortDirection::ASC),
                ]),
            );

            if ($jobs->isEmpty()) {
                break;
            }

            // Advance the cursor to the last row of this batch.
            $last = $jobs->last();
            $cursor = [$last->sequence, $last->id];
            $remaining -= $jobs->count();

            yield array_map(
                fn(QueueJob $queueJob): QueueJobTask => new QueueJobTask($queueJob),
                $jobs->getArray(),
            );
        }
    }
}
