nshop/packages/Webkul/BagistoApi/src/Console/Commands/ReconcileSweepCommand.php

<?php

namespace Webkul\BagistoApi\Console\Commands;

use Illuminate\Console\Command;
use Webkul\BagistoApi\Jobs\ReconcilePendingPaymentJob;
use Webkul\BagistoApi\Models\PaymentAttempt;
use Webkul\Sales\Models\Order;
use Webkul\Sales\Models\OrderProxy;

/**
 * Safety net for orders that fell out of the reconciliation queue.
 *
 * ReconcilePendingPaymentJob normally keeps itself alive by re-dispatching,
 * but that chain can break: the initial enqueue failed, no worker consumed
 * the queue for a while, or Redis dropped the delayed key. Without a sweep
 * such an order would sit in PENDING forever.
 *
 * Re-dispatching is free of side effects - ShouldBeUniqueUntilProcessing
 * drops the duplicate whenever a tick is already queued.
 */
class ReconcileSweepCommand extends Command
{
    protected $signature = 'bagistoapi:payment:reconcile-sweep
                            {--dry-run : Report what would be queued without dispatching}';

    protected $description = 'Re-queue reconciliation for pending orders whose polling chain was lost';

    public function handle(): int
    {
        if (! config('bagistoapi.reconcile.enabled', true)) {
            $this->info('Payment reconciliation is disabled; nothing to sweep.');

            return self::SUCCESS;
        }

        $cutoff = now()->subSeconds((int) config('bagistoapi.reconcile.max_seconds', 10800));
        $queue = (string) config('bagistoapi.reconcile.queue', 'payment-reconcile');
        $dryRun = (bool) $this->option('dry-run');

        /*
         * A replayed order can be older than the cutoff and still be inside
         * its polling window, because the window is measured from the last
         * redirect rather than from order creation.
         */
        $replayedOrderIds = PaymentAttempt::query()
            ->where('status', PaymentAttempt::STATUS_REDIRECTED)
            ->where('created_at', '>=', $cutoff)
            ->distinct()
            ->pluck('order_id')
            ->filter()
            ->all();

        $orderModelClass = OrderProxy::modelClass();

        $queued = 0;

        $orderModelClass::query()
            ->whereIn('status', [Order::STATUS_PENDING, Order::STATUS_PENDING_PAYMENT])
            ->where(function ($query) use ($cutoff, $replayedOrderIds) {
                $query->where('created_at', '>=', $cutoff);

                if ($replayedOrderIds) {
                    $query->orWhereIn('id', $replayedOrderIds);
                }
            })
            ->orderBy('id')
            ->chunkById(200, function ($orders) use (&$queued, $queue, $dryRun) {
                foreach ($orders as $order) {
                    $queued++;

                    if ($dryRun) {
                        $this->line("Would queue reconciliation for order {$order->id}");

                        continue;
                    }

                    dispatch((new ReconcilePendingPaymentJob($order->id))->onQueue($queue));
                }
            });

        $this->info($dryRun
            ? "Sweep found {$queued} pending order(s)."
            : "Sweep queued reconciliation for {$queued} pending order(s).");

        return self::SUCCESS;
    }
}