<?php
declare(strict_types=1);

namespace FiloBlu\Refilo\Console\Command;

use Exception;
use Magento\Eav\Model\Config as EavConfig;
use Magento\Framework\App\ResourceConnection;
use Magento\Framework\Console\Cli;
use Symfony\Component\Console\Command\Command;
use Symfony\Component\Console\Input\InputInterface;
use Symfony\Component\Console\Input\InputOption;
use Symfony\Component\Console\Output\OutputInterface;

/**
 * Removes the autogenerated url_rewrite rows of configurable children (size variants).
 *
 * Those products are enabled (they must be, to be purchasable through the parent) but
 * not visible, so they have no page of their own. Magento core never generates rewrites
 * for them: ProductProcessUrlRewriteSavingObserver is guarded by isVisibleInSiteVisibility().
 * The rows exist as a leftover and are never exported to Beehive/Mongo, since every url
 * provider in this module filters them out. On a catalog built out of configurables with many
 * size variants they can dominate the table, and they slow the export down even though they are
 * never exported, because AbstractFromQueryProvider paginates with LIMIT/OFFSET and every page
 * rescans the whole per-store slice. Run the command without --execute to see the proportion
 * on the install at hand.
 *
 * Disabled products are deliberately NOT touched: re-enabling a product does not regenerate
 * its rewrite, because status is not one of the events the core observer reacts to.
 *
 * Usage:
 *   bin/magento refilo:urlrewrite:cleanup                       # dry run, reports and changes nothing
 *   bin/magento refilo:urlrewrite:cleanup --execute             # backs up, then deletes
 *   bin/magento refilo:urlrewrite:cleanup --restore             # dry run of the rollback
 *   bin/magento refilo:urlrewrite:cleanup --restore --execute   # puts the backed up rows back
 *
 * Nothing is ever modified without --execute.
 *
 * The backup is not a separate step: --execute always copies the rows it is about to delete into
 * the table given by --backup-table, defaulting to url_rewrite_bkp_cleanup, and refuses to delete
 * if that copy came out short. The table is never dropped automatically, so the rollback stays
 * available until someone decides otherwise.
 *
 * Both --execute and --restore --execute need the beehive_urls materialized view frozen first,
 * because its triggers write into beehive_urls_cl on both insert and delete. Freezing removes
 * them, so a million row change generates no changelog at all:
 *   bin/magento indexer:set-mode realtime beehive_urls
 *   bin/magento refilo:urlrewrite:cleanup --execute
 *   bin/magento indexer:set-mode schedule beehive_urls
 *
 * The command refuses to touch anything while the triggers are in place, so the order cannot be
 * got wrong by accident.
 *
 * Interrupting a run mid deletion is safe, but the restart has to be explicit:
 *   bin/magento refilo:urlrewrite:cleanup --execute --resume
 *
 * The backup is written in full before the first row is removed, and the work table is recomputed
 * from url_rewrite on every run, so it only ever holds rows that are still there. Appending that
 * subset to the existing backup adds nothing new and the deletion carries on from where it stopped.
 * Without --resume an existing backup table is refused on purpose: the command cannot tell an
 * interrupted run apart from a brand new cleanup months later, and appending in the second case
 * would mix two operations into one backup, so that a later --restore would resurrect rows that
 * were deliberately removed the first time round.
 *
 * After the deletion run OPTIMIZE TABLE url_rewrite. A DELETE leaves the InnoDB pages sparse, so
 * a scan keeps reading roughly as many pages as before: the bulk of the gain is there without the
 * rebuild, but the rebuild is what makes the smaller table actually behave like a smaller table.
 */
class CleanupProductUrlRewriteCommand extends Command
{
    const OPT_EXECUTE = 'execute';
    const OPT_BATCH_SIZE = 'batch-size';
    const OPT_BACKUP_TABLE = 'backup-table';
    const OPT_RESTORE = 'restore';
    const OPT_RESUME = 'resume';

    const DEFAULT_BATCH_SIZE = 50000;
    const DEFAULT_BACKUP_TABLE = 'url_rewrite_bkp_cleanup';
    const WORK_TABLE = 'url_rewrite_cleanup_work';

    /**
     * @var ResourceConnection
     */
    private $resourceConnection;

    /**
     * @var EavConfig
     */
    private $eavConfig;

    /**
     * @param ResourceConnection $resourceConnection
     * @param EavConfig          $eavConfig
     * @param string|null        $name
     */
    public function __construct(
        ResourceConnection $resourceConnection,
        EavConfig          $eavConfig,
        string             $name = null
    )
    {
        $this->resourceConnection = $resourceConnection;
        $this->eavConfig = $eavConfig;
        parent::__construct($name);
    }

    /**
     * @return void
     */
    protected function configure()
    {
        $this->setName('refilo:urlrewrite:cleanup')
            ->setDescription('Remove autogenerated url_rewrite rows of non visible configurable children')
            ->setHelp($this->buildHelp())
            ->addOption(
                self::OPT_EXECUTE,
                null,
                InputOption::VALUE_NONE,
                'Actually delete. Without this flag the command only reports what it would do'
            )
            ->addOption(
                self::OPT_BATCH_SIZE,
                null,
                InputOption::VALUE_REQUIRED,
                'Size of the url_rewrite_id window deleted per statement',
                (string)self::DEFAULT_BATCH_SIZE
            )
            ->addOption(
                self::OPT_BACKUP_TABLE,
                null,
                InputOption::VALUE_REQUIRED,
                'Table receiving a copy of the deleted rows',
                self::DEFAULT_BACKUP_TABLE
            )
            ->addOption(
                self::OPT_RESTORE,
                null,
                InputOption::VALUE_NONE,
                'Put the rows of the backup table back into url_rewrite instead of deleting'
            )
            ->addOption(
                self::OPT_RESUME,
                null,
                InputOption::VALUE_NONE,
                'Carry on an interrupted run, appending to the existing backup table instead of refusing'
            );
    }

    /**
     * Extended guide printed by --help. Kept here rather than in a README so that whoever is about
     * to delete a million rows at eleven at night finds it in the terminal they are already in.
     *
     * @return string
     */
    private function buildHelp(): string
    {
        return <<<'HELP'
<info>What it removes</info>

  The autogenerated url_rewrite rows of the size variants of configurable products: items that
  are enabled, because they must be to be purchasable through the parent, but not visible, so
  they have no page of their own on the site.

  A row is removed only when all five of these hold:
    1. it belongs to a product, not a category, CMS page or content
    2. it is an autogenerated URL, redirect_type = 0, so never a redirect
    3. the product is enabled
    4. the product is not visible in catalog nor in search
    5. the product is a child of a configurable

  Those URLs are never sent to Beehive: every url provider in this module filters them out, so
  the storefront has never known them. On a catalog made of configurables with many size
  variants they can end up dominating the table. Run the command with no options to see how
  many there are on this install before deciding anything.

<info>What it never touches</info>

  Redirects. Category, CMS and content rows. Products that are visible.
  <comment>Disabled products, on purpose</comment>: re-enabling a product does not regenerate its rewrite,
  because status is not one of the events the core observer reacts to, so a URL removed here
  would stay dead. Variants are safe instead, because a change of visibility does trigger the
  observer and Magento regenerates the URL on its own.

  The four invariants are re-checked at every run and nothing is deleted if any of them fails.

<info>Normal sequence</info>

  <comment>1.</comment> See what would happen. Changes nothing:
       bin/magento refilo:urlrewrite:cleanup

  <comment>2.</comment> Freeze the materialized view. Its triggers write into beehive_urls_cl on both insert
     and delete, so without this a million row change becomes a million changelog entries and
     the indexer chews on them for hours while real URL updates queue up behind:
       bin/magento indexer:set-mode realtime beehive_urls

  <comment>3.</comment> Back up and delete. Run it under screen or nohup, it takes minutes:
       bin/magento refilo:urlrewrite:cleanup --execute

  <comment>4.</comment> Rebuild the table, otherwise the InnoDB pages stay sparse:
       OPTIMIZE TABLE url_rewrite;

  <comment>5.</comment> Put the view back:
       bin/magento indexer:set-mode schedule beehive_urls

  The command refuses to touch anything while the triggers are in place, so step 2 cannot be
  skipped by accident.

<info>Backup</info>

  There is no separate backup step. --execute always copies the rows it is about to delete into
  the table named by --backup-table, default url_rewrite_bkp_cleanup, before removing a single
  one, and it aborts if that copy comes out short. The table is never dropped automatically:
  drop it yourself once the result has been validated.

<info>If a run is interrupted</info>

  Deploy, pod restart, closed terminal. Nothing is corrupted: the backup was already complete
  and part of the rows are gone. Carry on with:
       bin/magento refilo:urlrewrite:cleanup --execute --resume

  The remaining set is recomputed from scratch, so no row is deleted twice or skipped.
  <comment>--resume is mandatory on purpose.</comment> An existing backup table can mean an interrupted run,
  where appending is correct, or a new cleanup months later reusing the same name, where
  appending would mix two operations into one backup and a later --restore would resurrect
  rows that were deliberately removed the first time. The command cannot tell those apart,
  so it asks you to say which. For a new cleanup, drop the old table or pass --backup-table.

<info>Rollback</info>

       bin/magento refilo:urlrewrite:cleanup --restore              # shows what it would restore
       bin/magento refilo:urlrewrite:cleanup --restore --execute    # does it

  The view has to be frozen for this too, for the same reason. Rows are put back with their
  original ids. INSERT IGNORE is used because url_rewrite has a unique key on
  (request_path, store_id): if something regenerated one of those URLs in the meantime a plain
  insert would fail and roll back everything. So the restored count can legitimately be lower
  than the size of the backup, and the command tells you when that happens.

<info>Worth knowing</info>

  The cleanup decays. Every new variant added to the catalog gets its own rewrite per store view,
  so the rows come back at the rate the catalog grows. This is meant to be re-run periodically
  rather than once.

  Anything that regenerates product URLs in bulk can undo it. Before scheduling such a job, check
  that it filters on status and visibility: a regeneration that does not will recreate exactly the
  rows this command removes.

  Measure rather than assume. Timings depend on catalog size, number of store views and how far
  the table has drifted, so take the counts from the dry run on the install at hand.
HELP;
    }

    /**
     * @param InputInterface  $input
     * @param OutputInterface $output
     *
     * @return int
     */
    protected function execute(InputInterface $input, OutputInterface $output)
    {
        $isExecute = (bool)$input->getOption(self::OPT_EXECUTE);
        $batchSize = (int)$input->getOption(self::OPT_BATCH_SIZE);
        $backupTable = (string)$input->getOption(self::OPT_BACKUP_TABLE);

        if ($batchSize < 1) {
            $output->writeln('<error>batch-size must be a positive integer</error>');
            return Cli::RETURN_FAILURE;
        }

        if ($input->getOption(self::OPT_RESTORE)) {
            return $this->restore($output, $backupTable, $isExecute);
        }

        try {
            $this->buildWorkTable($output);

            $candidates = $this->countWorkTable();
            $this->report($output, $candidates);

            if ($candidates === 0) {
                $output->writeln('<info>Nothing to do.</info>');
                return Cli::RETURN_SUCCESS;
            }

            $failed = $this->runSafetyChecks($output);
            if ($failed) {
                $output->writeln('<error>Safety checks failed, aborting without touching anything.</error>');
                return Cli::RETURN_FAILURE;
            }

            if (!$isExecute) {
                $output->writeln('');
                $output->writeln('<comment>Dry run, nothing was modified. Re-run with --execute to apply.</comment>');
                return Cli::RETURN_SUCCESS;
            }

            if (!$this->assertMviewFrozen($output)) {
                return Cli::RETURN_FAILURE;
            }

            if (!$this->backup($output, $backupTable, (bool)$input->getOption(self::OPT_RESUME))) {
                return Cli::RETURN_FAILURE;
            }

            $deleted = $this->deleteInBatches($output, $batchSize);

            $output->writeln('');
            $output->writeln(sprintf('<info>Deleted %s rows. Backup kept in %s.</info>', number_format($deleted), $backupTable));
            $output->writeln('<comment>Now run: OPTIMIZE TABLE url_rewrite; then re-enable the view with</comment>');
            $output->writeln('<comment>  bin/magento indexer:set-mode schedule beehive_urls</comment>');

            return Cli::RETURN_SUCCESS;
        } catch (Exception $e) {
            $output->writeln('<error>Error: ' . $e->getMessage() . '</error>');
            return Cli::RETURN_FAILURE;
        } finally {
            $this->dropWorkTable();
        }
    }

    /**
     * Collects the url_rewrite_id of every row matching the five conditions of the safe set.
     *
     * The created_in/updated_in bounds pin the attribute lookup to the staging version currently
     * in force. Without them a product with a scheduled update can be read from the wrong version
     * and slip into the set with the wrong status.
     *
     * @param OutputInterface $output
     *
     * @return void
     */
    private function buildWorkTable(OutputInterface $output)
    {
        $connection = $this->resourceConnection->getConnection();
        $work = $this->resourceConnection->getTableName(self::WORK_TABLE);

        $connection->query(sprintf('DROP TABLE IF EXISTS %s', $connection->quoteIdentifier($work)));
        $connection->query(sprintf(
            'CREATE TABLE %s (url_rewrite_id INT UNSIGNED NOT NULL, PRIMARY KEY (url_rewrite_id))',
            $connection->quoteIdentifier($work)
        ));

        $output->writeln('Collecting candidate rows...');

        $connection->query(
            sprintf(
                'INSERT INTO %s (url_rewrite_id)
                 SELECT DISTINCT u.url_rewrite_id
                   FROM %s u
                   INNER JOIN %s p  ON p.entity_id = u.entity_id
                                   AND p.created_in <= :now AND p.updated_in > :now
                   INNER JOIN %s s  ON s.row_id = p.row_id AND s.attribute_id = :status
                                   AND s.store_id = 0 AND s.value = 1
                   INNER JOIN %s v  ON v.row_id = p.row_id AND v.attribute_id = :visibility
                                   AND v.store_id = 0 AND v.value = 1
                   INNER JOIN %s l  ON l.product_id = p.entity_id
                  WHERE u.entity_type = :entityType
                    AND u.redirect_type = 0',
                $connection->quoteIdentifier($work),
                $connection->quoteIdentifier($this->resourceConnection->getTableName('url_rewrite')),
                $connection->quoteIdentifier($this->resourceConnection->getTableName('catalog_product_entity')),
                $connection->quoteIdentifier($this->resourceConnection->getTableName('catalog_product_entity_int')),
                $connection->quoteIdentifier($this->resourceConnection->getTableName('catalog_product_entity_int')),
                $connection->quoteIdentifier($this->resourceConnection->getTableName('catalog_product_super_link'))
            ),
            [
                'now' => time(),
                'status' => $this->getAttributeId('status'),
                'visibility' => $this->getAttributeId('visibility'),
                'entityType' => 'product',
            ]
        );
    }

    /**
     * @param OutputInterface $output
     * @param int             $candidates
     *
     * @return void
     */
    private function report(OutputInterface $output, int $candidates)
    {
        $connection = $this->resourceConnection->getConnection();
        $urlRewrite = $this->resourceConnection->getTableName('url_rewrite');
        $work = $this->resourceConnection->getTableName(self::WORK_TABLE);

        $total = (int)$connection->fetchOne(sprintf('SELECT COUNT(*) FROM %s', $connection->quoteIdentifier($urlRewrite)));
        $products = (int)$connection->fetchOne(sprintf(
            'SELECT COUNT(DISTINCT u.entity_id) FROM %s u INNER JOIN %s t ON t.url_rewrite_id = u.url_rewrite_id',
            $connection->quoteIdentifier($urlRewrite),
            $connection->quoteIdentifier($work)
        ));

        $output->writeln('');
        $output->writeln(sprintf('  url_rewrite rows now      : %s', number_format($total)));
        $output->writeln(sprintf('  candidates for deletion   : %s (%s%%)', number_format($candidates), $total > 0 ? round(100 * $candidates / $total, 1) : 0));
        $output->writeln(sprintf('  distinct products         : %s', number_format($products)));
        $output->writeln(sprintf('  rows remaining after      : %s', number_format($total - $candidates)));

        $sample = $connection->fetchCol(sprintf(
            'SELECT u.request_path FROM %s u INNER JOIN %s t ON t.url_rewrite_id = u.url_rewrite_id LIMIT 5',
            $connection->quoteIdentifier($urlRewrite),
            $connection->quoteIdentifier($work)
        ));

        if ($sample) {
            $output->writeln('  sample:');
            foreach ($sample as $path) {
                $output->writeln('    ' . $path);
            }
        }
    }

    /**
     * Four invariants that must hold for the set to be safe. Any non zero value means the
     * criterion picked up something it should not have, and nothing gets deleted.
     *
     * @param OutputInterface $output
     *
     * @return bool true when at least one check failed
     */
    private function runSafetyChecks(OutputInterface $output): bool
    {
        $connection = $this->resourceConnection->getConnection();
        $urlRewrite = $connection->quoteIdentifier($this->resourceConnection->getTableName('url_rewrite'));
        $productEntity = $connection->quoteIdentifier($this->resourceConnection->getTableName('catalog_product_entity'));
        $productInt = $connection->quoteIdentifier($this->resourceConnection->getTableName('catalog_product_entity_int'));
        $work = $connection->quoteIdentifier($this->resourceConnection->getTableName(self::WORK_TABLE));

        $join = sprintf('%s u INNER JOIN %s t ON t.url_rewrite_id = u.url_rewrite_id', $urlRewrite, $work);
        $version = sprintf(
            'INNER JOIN %s p ON p.entity_id = u.entity_id AND p.created_in <= %d AND p.updated_in > %d',
            $productEntity,
            time(),
            time()
        );

        $checks = [
            'no redirect is touched' => sprintf('SELECT COUNT(*) FROM %s WHERE u.redirect_type <> 0', $join),
            'no non-product row is touched' => sprintf('SELECT COUNT(*) FROM %s WHERE u.entity_type <> \'product\'', $join),
            'no visible product in the set' => sprintf(
                'SELECT COUNT(*) FROM %s %s INNER JOIN %s v ON v.row_id = p.row_id AND v.attribute_id = %d AND v.store_id = 0 AND v.value <> 1',
                $join,
                $version,
                $productInt,
                $this->getAttributeId('visibility')
            ),
            'no disabled product in the set' => sprintf(
                'SELECT COUNT(*) FROM %s %s INNER JOIN %s s ON s.row_id = p.row_id AND s.attribute_id = %d AND s.store_id = 0 AND s.value <> 1',
                $join,
                $version,
                $productInt,
                $this->getAttributeId('status')
            ),
        ];

        $output->writeln('');
        $output->writeln('Safety checks:');

        $failed = false;
        foreach ($checks as $label => $sql) {
            $value = (int)$connection->fetchOne($sql);
            if ($value === 0) {
                $output->writeln(sprintf('  <info>ok</info>    %s', $label));
                continue;
            }
            $output->writeln(sprintf('  <error>FAIL</error>  %s (found %d)', $label, $value));
            $failed = true;
        }

        return $failed;
    }

    /**
     * The beehive_urls delete trigger writes OLD.url_rewrite_id into beehive_urls_cl. Deleting
     * with the triggers in place would push over a million entries into the changelog and wake
     * the indexer on every one of them, which is the very problem this cleanup exists to reduce.
     *
     * @param OutputInterface $output
     *
     * @return bool
     */
    private function assertMviewFrozen(OutputInterface $output): bool
    {
        $connection = $this->resourceConnection->getConnection();

        $triggers = (int)$connection->fetchOne(
            'SELECT COUNT(*) FROM information_schema.TRIGGERS
              WHERE EVENT_OBJECT_TABLE = :table AND TRIGGER_SCHEMA = DATABASE()',
            ['table' => $this->resourceConnection->getTableName('url_rewrite')]
        );

        if ($triggers === 0) {
            return true;
        }

        $output->writeln('');
        $output->writeln(sprintf('<error>%d mview triggers are still active on url_rewrite.</error>', $triggers));
        $output->writeln('<comment>Freeze the view first, otherwise every deleted row lands in beehive_urls_cl:</comment>');
        $output->writeln('<comment>  bin/magento indexer:set-mode realtime beehive_urls</comment>');

        return false;
    }

    /**
     * @param OutputInterface $output
     * @param string          $backupTable
     *
     * @return bool
     */
    private function backup(OutputInterface $output, string $backupTable, bool $isResume): bool
    {
        $connection = $this->resourceConnection->getConnection();
        $target = $this->resourceConnection->getTableName($backupTable);
        $urlRewrite = $this->resourceConnection->getTableName('url_rewrite');
        $work = $this->resourceConnection->getTableName(self::WORK_TABLE);

        if ($connection->isTableExists($target)) {
            // An existing backup means one of two things, and the command cannot tell them apart:
            // either a previous run of THIS cleanup was interrupted, in which case appending is
            // correct and harmless, or this is a NEW cleanup months later, in which case appending
            // would mix two operations into one backup and a later --restore would resurrect rows
            // that were deliberately removed the first time. So the operator has to say which.
            if (!$isResume) {
                $output->writeln(sprintf('<error>Backup table %s already exists with %s rows.</error>',
                    $target,
                    number_format((int)$connection->fetchOne(sprintf('SELECT COUNT(*) FROM %s', $connection->quoteIdentifier($target))))
                ));
                $output->writeln('<comment>If a previous run of this same cleanup was interrupted, add --resume to carry on and top up that backup.</comment>');
                $output->writeln('<comment>If this is a new cleanup, drop the old table or pass a different --backup-table, so the two backups stay separate.</comment>');
                return false;
            }

            $output->writeln(sprintf('<comment>Resuming, appending to the existing backup %s.</comment>', $target));
        } elseif ($isResume) {
            $output->writeln(sprintf('<error>--resume was given but %s does not exist.</error>', $target));
            return false;
        } else {
            $connection->query(sprintf(
                'CREATE TABLE %s LIKE %s',
                $connection->quoteIdentifier($target),
                $connection->quoteIdentifier($urlRewrite)
            ));
        }

        $output->writeln(sprintf('Copying rows into %s...', $target));

        $connection->query(sprintf(
            'INSERT IGNORE INTO %s SELECT u.* FROM %s u INNER JOIN %s t ON t.url_rewrite_id = u.url_rewrite_id',
            $connection->quoteIdentifier($target),
            $connection->quoteIdentifier($urlRewrite),
            $connection->quoteIdentifier($work)
        ));

        $backedUp = (int)$connection->fetchOne(sprintf('SELECT COUNT(*) FROM %s', $connection->quoteIdentifier($target)));
        $pending = $this->countWorkTable();

        if ($backedUp < $pending) {
            $output->writeln(sprintf(
                '<error>Backup holds %s rows but %s are about to be deleted. Aborting.</error>',
                number_format($backedUp),
                number_format($pending)
            ));
            return false;
        }

        $output->writeln(sprintf('  backup now holds %s rows', number_format($backedUp)));

        return true;
    }

    /**
     * Puts the backed up rows back into url_rewrite.
     *
     * Uses INSERT IGNORE on purpose: url_rewrite has a unique key on (request_path, store_id), so
     * if anything regenerated one of those rewrites in the meantime a plain INSERT would fail on
     * the duplicate and roll back the whole restore. With IGNORE the rows that came back on their
     * own are skipped and everything else is restored, which is why the final count can legitimately
     * be lower than the size of the backup.
     *
     * The backup table is never dropped here. Deciding that it is no longer needed is the operator's
     * call, after the restore has been validated.
     *
     * @param OutputInterface $output
     * @param string          $backupTable
     * @param bool            $isExecute
     *
     * @return int
     */
    private function restore(OutputInterface $output, string $backupTable, bool $isExecute): int
    {
        $connection = $this->resourceConnection->getConnection();
        $target = $this->resourceConnection->getTableName($backupTable);
        $urlRewrite = $this->resourceConnection->getTableName('url_rewrite');

        if (!$connection->isTableExists($target)) {
            $output->writeln(sprintf('<error>Backup table %s does not exist, nothing to restore.</error>', $target));
            return Cli::RETURN_FAILURE;
        }

        $inBackup = (int)$connection->fetchOne(sprintf('SELECT COUNT(*) FROM %s', $connection->quoteIdentifier($target)));
        $missing = (int)$connection->fetchOne(sprintf(
            'SELECT COUNT(*) FROM %s b LEFT JOIN %s u ON u.url_rewrite_id = b.url_rewrite_id
              WHERE u.url_rewrite_id IS NULL',
            $connection->quoteIdentifier($target),
            $connection->quoteIdentifier($urlRewrite)
        ));
        $before = (int)$connection->fetchOne(sprintf('SELECT COUNT(*) FROM %s', $connection->quoteIdentifier($urlRewrite)));

        $output->writeln('');
        $output->writeln(sprintf('  url_rewrite rows now      : %s', number_format($before)));
        $output->writeln(sprintf('  rows in %-18s: %s', $backupTable, number_format($inBackup)));
        $output->writeln(sprintf('  actually missing          : %s', number_format($missing)));

        if ($missing === 0) {
            $output->writeln('<info>Every backed up row is already in place, nothing to restore.</info>');
            return Cli::RETURN_SUCCESS;
        }

        if (!$isExecute) {
            $output->writeln('');
            $output->writeln('<comment>Dry run, nothing was modified. Add --execute to apply the restore.</comment>');
            return Cli::RETURN_SUCCESS;
        }

        if (!$this->assertMviewFrozen($output)) {
            return Cli::RETURN_FAILURE;
        }

        $connection->query(sprintf(
            'INSERT IGNORE INTO %s SELECT * FROM %s',
            $connection->quoteIdentifier($urlRewrite),
            $connection->quoteIdentifier($target)
        ));

        $after = (int)$connection->fetchOne(sprintf('SELECT COUNT(*) FROM %s', $connection->quoteIdentifier($urlRewrite)));

        $output->writeln('');
        $output->writeln(sprintf('<info>Restored %s rows, url_rewrite now holds %s.</info>', number_format($after - $before), number_format($after)));

        if ($after - $before < $missing) {
            $output->writeln(sprintf(
                '<comment>%s rows were skipped on the unique key (request_path, store_id), meaning something had already recreated them.</comment>',
                number_format($missing - ($after - $before))
            ));
        }

        $output->writeln(sprintf('<comment>%s was left in place. Drop it only once the restore is validated.</comment>', $target));
        $output->writeln('<comment>Re-enable the view with: bin/magento indexer:set-mode schedule beehive_urls</comment>');

        return Cli::RETURN_SUCCESS;
    }

    /**
     * Deletes over windows of the primary key instead of one large statement, so each step is
     * bounded, interruptible and resumable. MySQL does not accept LIMIT on a multi table delete,
     * which is why the batching is expressed as an id range.
     *
     * @param OutputInterface $output
     * @param int             $batchSize
     *
     * @return int
     */
    private function deleteInBatches(OutputInterface $output, int $batchSize): int
    {
        $connection = $this->resourceConnection->getConnection();
        $urlRewrite = $connection->quoteIdentifier($this->resourceConnection->getTableName('url_rewrite'));
        $work = $connection->quoteIdentifier($this->resourceConnection->getTableName(self::WORK_TABLE));

        $max = (int)$connection->fetchOne(sprintf('SELECT MAX(url_rewrite_id) FROM %s', $work));
        $deleted = 0;

        for ($low = 0; $low <= $max; $low += $batchSize) {
            $affected = $connection->query(sprintf(
                'DELETE u FROM %s u INNER JOIN %s t ON t.url_rewrite_id = u.url_rewrite_id
                  WHERE u.url_rewrite_id >= %d AND u.url_rewrite_id < %d',
                $urlRewrite,
                $work,
                $low,
                $low + $batchSize
            ))->rowCount();

            $deleted += $affected;

            if ($affected > 0) {
                $output->writeln(sprintf('  ids %d-%d: %s rows', $low, $low + $batchSize, number_format($affected)));
            }
        }

        return $deleted;
    }

    /**
     * @return int
     */
    private function countWorkTable(): int
    {
        $connection = $this->resourceConnection->getConnection();

        return (int)$connection->fetchOne(sprintf(
            'SELECT COUNT(*) FROM %s',
            $connection->quoteIdentifier($this->resourceConnection->getTableName(self::WORK_TABLE))
        ));
    }

    /**
     * @return void
     */
    private function dropWorkTable()
    {
        $connection = $this->resourceConnection->getConnection();
        $connection->query(sprintf(
            'DROP TABLE IF EXISTS %s',
            $connection->quoteIdentifier($this->resourceConnection->getTableName(self::WORK_TABLE))
        ));
    }

    /**
     * @param string $code
     *
     * @return int
     * @throws \Magento\Framework\Exception\LocalizedException
     */
    private function getAttributeId(string $code): int
    {
        return (int)$this->eavConfig->getAttribute('catalog_product', $code)->getId();
    }
}
