<?php

if (! defined('ABSPATH')) {
    exit;
}

/**
 * The training pipeline: decides what can be trained, turns posts and PDFs
 * into embedded chunks, runs training in background batches under a lock,
 * and keeps the training data in step with post saves, publishes and deletes.
 *
 * What's stored is RSVL_Index; a run's progress is RSVL_Indexing_Status.
 */
class RSVL_Indexer
{
    /**
     * Option row used as a server-wide mutex around training runs. Written
     * with INSERT IGNORE so acquisition is atomic at the database level —
     * two admins clicking Train at the same moment can't both win. Holds
     * "time|batch ID": the run that holds it, and when it last made progress.
     */
    private const INDEXING_LOCK_OPTION = 'resolver_indexing_lock';

    /**
     * Seconds a run can go without progress before it's taken to have
     * stopped. Longer than a batch's time limit, since time spent waiting on
     * the embedding API doesn't count towards PHP's limit on most servers.
     */
    private const STALLED_AFTER = 15 * MINUTE_IN_SECONDS;

    /** Option naming the batch a run does next, as "batch ID:offset"; see claim_batch(). */
    private const NEXT_BATCH_OPTION = 'resolver_indexing_next';

    /** Posts a background batch trains at once. */
    private const BATCH_SIZE = 50;

    /* ---------------------------------------------------------------------
     * Starting training
     * ------------------------------------------------------------------- */

    /**
     * Queue posts for training in background batches.
     *
     * @param  int[] $post_ids
     * @return int|false Number of posts queued, or false when another batch
     *                   run holds the indexing lock.
     */
    public function start_selective_index(array $post_ids)
    {
        $post_ids = array_values(array_filter(array_map('absint', $post_ids)));
        if (empty($post_ids)) {
            return 0;
        }

        // One run at a time, server-wide: concurrent runs clobber each
        // other's status/queue options, and overlapping post sets can corrupt
        // stored embeddings. The JS blocks double-starts within one tab; this
        // is the backstop for two admins (or tabs) starting at the same moment.
        // A run that has stopped responding doesn't count.
        self::check_stalled_run();

        $batch_id = 'resolver_bulk_' . time() . '_' . wp_rand(1000, 9999);
        if (! self::acquire_indexing_lock($batch_id)) {
            return false;
        }

        RSVL_Indexing_Status::start(count($post_ids));
        update_option('resolver_training_queue', $post_ids);

        self::queue_batch($batch_id, 0);

        return count($post_ids);
    }

    /**
     * Retrain everything, with the current embedding model.
     *
     * @return int|false Number of posts queued, or false when another batch
     *                   run holds the indexing lock.
     */
    public function start_full_reindex()
    {
        $post_ids = RSVL_Index::get_post_ids();

        /**
         * Fires when all training data is being re-embedded (after an
         * embedding model upgrade or a switch of training platform), for
         * training data that isn't a post, such as Pro's notes.
         */
        do_action('resolver_full_reindex');

        return $post_ids ? $this->start_selective_index($post_ids) : 0;
    }

    /**
     * Posts queued in the current run but not yet trained, for the Pages
     * table's Training rows.
     *
     * A queued post counts as still training until its indexed_at moves past
     * the run's start, so a retrain shows as Training until it's actually
     * reprocessed, rather than as Trained from its old row.
     *
     * @return array[] Each: ['post_id', 'title', 'url', 'type_slug', 'type_label'].
     */
    public function get_training_queue_posts(): array
    {
        if (! RSVL_Indexing_Status::is_running()) {
            return [];
        }

        $queue = get_option('resolver_training_queue', []);
        if (empty($queue)) {
            return [];
        }

        $started_at   = RSVL_Indexing_Status::started_at();
        $indexed_data = RSVL_Index::get_post_data($queue);
        $posts        = [];

        foreach ($queue as $post_id) {
            $post_id = (int) $post_id;
            $info    = $indexed_data[ $post_id ] ?? ['indexed' => false, 'indexed_at' => null];

            if ($info['indexed'] && $started_at && $info['indexed_at'] >= $started_at) {
                continue;
            }

            $post = get_post($post_id);
            if (! $post) {
                continue;
            }
            if ($post->post_type === 'attachment') {
                $filename   = basename(get_attached_file($post->ID) ?: '');
                $title      = $post->post_title ?: $filename;
                $type_slug  = 'rsvl-pdf';
                $type_label = __('PDF', 'resolver-ai-chatbot');
                $url        = wp_get_attachment_url($post->ID) ?: '';
            } else {
                $pt_obj     = get_post_type_object($post->post_type);
                $title      = $post->post_title;
                $type_slug  = $post->post_type;
                $type_label = $pt_obj ? $pt_obj->labels->singular_name : ucfirst($post->post_type);
                $url        = get_permalink($post->ID) ?: '';
            }
            $posts[] = [
                'post_id'    => $post->ID,
                'title'      => $title,
                'url'        => $url,
                'type_slug'  => $type_slug,
                'type_label' => $type_label,
            ];
        }

        return $posts;
    }

    /* ---------------------------------------------------------------------
     * Keeping in step with post changes
     * ------------------------------------------------------------------- */

    /**
     * save_post callback: retrain a trained post when it's saved, a minute
     * later, along with anything else saved meanwhile.
     */
    public function handle_post_save(int $post_id, WP_Post $_post)
    {
        if (defined('DOING_AUTOSAVE') && DOING_AUTOSAVE) {
            return;
        }
        if (wp_is_post_revision($post_id)) {
            return;
        }
        if (resolver_get_setting('disable_auto_retrain')) {
            return;
        }

        // Only re-index pages the user has already trained on. New pages are
        // added by handle_post_publish(), and only when first published.
        if (! RSVL_Index::is_indexed($post_id)) {
            return;
        }

        $this->enqueue_auto_retrain($post_id);
    }

    /**
     * Queue a newly published post for training when its type is one the user
     * chose to train automatically (the auto_train_post_types setting).
     *
     * Only the move into a trainable status counts. Edits to a published post
     * never qualify, so a page the user removed from training stays removed.
     */
    public function handle_post_publish(string $new_status, string $old_status, WP_Post $post): void
    {
        if (! in_array($post->post_type, (array) resolver_get_setting('auto_train_post_types'), true)) {
            return;
        }

        $trainable = self::trainable_statuses();
        if (! in_array($new_status, $trainable, true) || in_array($old_status, $trainable, true)) {
            return;
        }

        // Already trained (added by hand while still a draft, for instance):
        // handle_post_save() retrains it.
        if (RSVL_Index::is_indexed($post->ID)) {
            return;
        }

        $this->enqueue_auto_retrain($post->ID);
    }

    /** Post delete and trash callback. */
    public function handle_post_delete(int $post_id)
    {
        RSVL_Index::delete_post($post_id);
    }

    /* ---------------------------------------------------------------------
     * Training
     * ------------------------------------------------------------------- */

    /**
     * Cron callback: retrain every post accumulated in the auto-retrain queue,
     * embedding all of them together, so the whole queue costs one (or a few)
     * API requests however many posts are in it.
     */
    public function process_auto_retrain(): void
    {
        // Leave the queue in place when no provider is connected; the next
        // save schedules another run, which picks these posts up.
        $embedder = new RSVL_Embedder();
        if (! $embedder->is_available()) {
            return;
        }

        // A training run may be training these very posts: try again once
        // it's over, rather than pay to embed them twice.
        if (self::is_indexing_locked()) {
            if (! wp_next_scheduled('resolver_auto_retrain')) {
                wp_schedule_single_event(time() + MINUTE_IN_SECONDS, 'resolver_auto_retrain');
            }

            return;
        }

        $queue = get_option('resolver_auto_retrain_queue', []);
        delete_option('resolver_auto_retrain_queue');

        if (! is_array($queue) || empty($queue)) {
            return;
        }

        $items = [];
        foreach (array_unique(array_map('intval', $queue)) as $post_id) {
            $item = $this->prepare_item($post_id);
            if (null !== $item) {
                $items[] = $item;
            }
        }

        if (empty($items)) {
            return;
        }

        $result = $this->train_items($items, $embedder);
        if (! is_wp_error($result)) {
            return;
        }

        resolver_log('Auto-retrain embed failed: ' . $result->get_error_message());
        $error_string = $embedder->get_provider_name() . ': ' . $result->get_error_message();

        // A post that was never trained has no row to hold the error, so
        // it goes back in the queue for the next run rather than being
        // dropped without a trace.
        $untrained = [];
        foreach ($items as $item) {
            if (RSVL_Index::is_indexed($item['post_id'])) {
                RSVL_Index::set_training_error($item['post_id'], $error_string);
            } else {
                $untrained[] = $item['post_id'];
            }
        }
        if ($untrained) {
            $this->add_to_auto_retrain_queue($untrained);
        }
    }

    /**
     * Train one batch of a run, then queue the next, until the run's posts
     * are all done. Each batch embeds all its posts together.
     */
    public function process_bulk_index_batch(string $batch_id, int $offset = 0)
    {
        // Already started by the other trigger, or not this run's next batch.
        if (! self::claim_batch($batch_id, $offset)) {
            return;
        }

        // The loopback that starts a batch hangs up at once, and PHP may stop
        // a script whose client is gone as soon as it prints anything (a PHP
        // notice, say). wp-cron.php does the same, for the same reason.
        ignore_user_abort(true);

        // Up to 50 posts, embedded in requests of up to 90 seconds each.
        if (function_exists('set_time_limit')) {
            set_time_limit(5 * MINUTE_IN_SECONDS);
        }

        self::record_progress($batch_id);

        // If the batch dies (a fatal error, an uncaught exception, running out
        // of time), end the run as failed with the reason, rather than leave
        // it showing as running with nothing left to continue it.
        $finished = false;
        register_shutdown_function(static function () use (&$finished, $batch_id) {
            if (! $finished && self::is_current_run($batch_id)) {
                self::fail_stopped_batch($batch_id);
            }
        });

        $this->run_batch($batch_id, $offset);
        $finished = true;
    }

    /**
     * The work of one batch: see process_bulk_index_batch().
     */
    private function run_batch(string $batch_id, int $offset): void
    {
        $post_ids = (array) get_option('resolver_training_queue', []);
        $batch    = array_slice($post_ids, $offset, self::BATCH_SIZE);
        $total    = count($post_ids);

        if (empty($batch)) {
            RSVL_Indexing_Status::finish($total);
            self::end_run($batch_id);

            return;
        }

        $embedder = new RSVL_Embedder();
        if (! $embedder->is_available()) {
            RSVL_Indexing_Status::fail($total, $offset, __('No embedding provider configured.', 'resolver-ai-chatbot'));
            self::end_run($batch_id);

            return;
        }

        $items = [];
        foreach ($batch as $post_id) {
            // Attachments (PDFs) can take a while to read, so their step is shown.
            if (get_post_type($post_id) === 'attachment') {
                /* translators: %s: file name of the PDF being read. */
                RSVL_Indexing_Status::log_step('extract', sprintf(__('Extracting text: %s', 'resolver-ai-chatbot'), basename(get_attached_file($post_id) ?: 'PDF')));
                self::record_progress($batch_id);
            }

            $item = $this->prepare_item((int) $post_id);
            if (null !== $item) {
                $items[] = $item;
            }
        }

        if ($items) {
            $provider_name = $embedder->get_provider_name();
            $range_end     = min($offset + count($batch), $total);
            if ($total > self::BATCH_SIZE) {
                $range_start = $offset + 1;
                $range_label = $range_start === $range_end ? $range_end : $range_start . '–' . $range_end;
                RSVL_Indexing_Status::log_step(
                    'embed',
                    /* translators: 1: embedding provider, e.g. OpenAI. 2: pages in this batch, e.g. 51–100. 3: total pages. */
                    sprintf(__('Making API calls to %1$s (%2$s of %3$s)', 'resolver-ai-chatbot'), $provider_name, $range_label, $total),
                    ['done' => $range_end, 'total' => $total]
                );
            } else {
                /* translators: %s: embedding provider, e.g. OpenAI. */
                RSVL_Indexing_Status::log_step('embed', sprintf(__('Making API calls to %s', 'resolver-ai-chatbot'), $provider_name));
            }

            self::record_progress($batch_id);
            $result = $this->train_items($items, $embedder);
            if (is_wp_error($result)) {
                resolver_log('Batch embed failed: ' . $result->get_error_message());

                $message = $result->get_error_message();
                if ($result->get_error_code() === 'http_request_failed') {
                    /* translators: %s: the connection error. */
                    $message = sprintf(__('Could not reach the embedding API (%s). Check your server\'s outbound connection and try again.', 'resolver-ai-chatbot'), $message);
                }
                $error_string = $provider_name . ': ' . $message;

                foreach ($items as $item) {
                    RSVL_Index::set_training_error($item['post_id'], $error_string);
                }

                RSVL_Indexing_Status::fail($total, $offset, $error_string);
                self::end_run($batch_id);

                return;
            }

            RSVL_Indexing_Status::log_trained(array_column($items, 'post_id'));
        }

        // A batch that took long enough for its run to be ended as stalled
        // leaves what it trained, but mustn't carry on the run.
        if (! self::is_current_run($batch_id)) {
            return;
        }

        // Progress counts every post in the batch, not only those trained:
        // posts prepare_item() skipped (emptied, unpublished) are done too, or
        // the run could never reach its total.
        $next_offset = $offset + self::BATCH_SIZE;
        $has_more    = $next_offset < $total;
        RSVL_Indexing_Status::advance($has_more ? min($offset + count($batch), $total) : $total, $total);

        if (! $has_more) {
            self::end_run($batch_id);

            return;
        }

        self::record_progress($batch_id);
        self::queue_batch($batch_id, $next_offset);
    }

    /**
     * Embed prepared items together and store each one.
     *
     * @param array[] $items Prepared items (see prepare_post()).
     * @return true|WP_Error
     */
    private function train_items(array $items, RSVL_Embedder $embedder)
    {
        $items   = array_values($items);
        $vectors = $embedder->embed_chunk_lists(array_column($items, 'chunks'), array_map(fn (array $item) => (string) ($item['title'] ?? ''), $items));
        if (is_wp_error($vectors)) {
            return $vectors;
        }

        foreach ($items as $index => $item) {
            RSVL_Index::replace_post($item, $vectors[ $index ] ?? [], $embedder->get_model());
        }

        return true;
    }

    /**
     * Get a post ready to embed: check it can be trained, extract its text,
     * chunk it. A post that can't be trained, or has no text, is removed from
     * the training data and skipped.
     *
     * Its current chunks stay until new ones are stored (RSVL_Index::
     * replace_post()), so a failed embed never wipes good data.
     *
     * @return array|null ['post_id', 'source_url', 'post_type', 'title', 'chunks', 'total', 'found_fields'], or null to skip.
     */
    private function prepare_post(int $post_id): ?array
    {
        $post = get_post($post_id);
        if (! $post) {
            return null;
        }

        $content = self::is_trainable($post) ? RSVL_Chunker::extract_post_content($post) : ['title' => '', 'text' => ''];
        $chunks  = (new RSVL_Chunker())->chunk($content['text'], $content['title']);
        if (empty($chunks)) {
            RSVL_Index::delete_post($post_id);

            return null;
        }

        return [
            'post_id'      => $post_id,
            'source_url'   => get_permalink($post_id),
            'post_type'    => $post->post_type,
            'title'        => $content['title'],
            'chunks'       => array_values($chunks),
            'total'        => count($chunks),
            'found_fields' => RSVL_Chunker::$last_found_fields,
        ];
    }

    /**
     * Prepare anything that can be trained for embedding: a post, or an
     * attachment when something handles its file type (Pro's PDFs).
     *
     * @return array|null Same shape as prepare_post(), or null to skip.
     */
    private function prepare_item(int $post_id): ?array
    {
        if (get_post_type($post_id) !== 'attachment') {
            return $this->prepare_post($post_id);
        }

        /**
         * Filter the prepared data for an attachment being trained. Nothing
         * trains attachments by default, so it's skipped unless a callback
         * returns data in the shape of RSVL_Indexer::prepare_post()'s.
         *
         * @param array|null $post_data     Default null (skip).
         * @param int        $attachment_id The attachment.
         */
        $post_data = apply_filters('resolver_prepare_attachment', null, $post_id);

        return is_array($post_data) ? $post_data : null;
    }

    /* ---------------------------------------------------------------------
     * Auto-retrain queue
     * ------------------------------------------------------------------- */

    /**
     * Add a post to the auto-retrain queue and schedule a run to process it.
     */
    private function enqueue_auto_retrain(int $post_id): void
    {
        $this->add_to_auto_retrain_queue([$post_id]);

        // Schedule the cron only if one isn't already pending.
        if (! wp_next_scheduled('resolver_auto_retrain')) {
            wp_schedule_single_event(time() + MINUTE_IN_SECONDS, 'resolver_auto_retrain');
        }
    }

    /**
     * Accumulate post IDs in an option so bulk updates and duplicate
     * save_post firings collapse into a single reindex run. An option rather
     * than a transient: WP-Cron can fire long after the save on a quiet site,
     * and the queue must still be there when it does.
     */
    private function add_to_auto_retrain_queue(array $post_ids): void
    {
        // Read the stored list, not this request's cached copy. A long request
        // (an import, say) can outlive a cron run that processed and deleted
        // the queue; writing back its stale copy would update a row that no
        // longer exists, and these IDs would be silently lost.
        wp_cache_delete('resolver_auto_retrain_queue', 'options');
        $queue = get_option('resolver_auto_retrain_queue', []);
        $queue = is_array($queue) ? $queue : [];
        $queue = array_values(array_unique(array_merge($queue, array_map('intval', $post_ids))));
        update_option('resolver_auto_retrain_queue', $queue, false);
    }

    /**
     * Retrain everything when the training platform is switched: search only
     * uses content embedded by the current model, so until it's retrained,
     * the chatbot has nothing to answer from. Pages, PDFs and notes alike.
     * The switch is only saved when this can start (see RSVL_Settings::sanitize()).
     *
     * @param mixed $old_value Previous settings.
     * @param mixed $new_value New settings.
     */
    public static function handle_settings_update($old_value, $new_value): void
    {
        if (RSVL_Settings::is_embedding_provider_switch((string) ($old_value['embedding_provider'] ?? ''), (string) ($new_value['embedding_provider'] ?? ''))) {
            (new self())->start_full_reindex();
        }
    }

    /* ---------------------------------------------------------------------
     * What can be trained
     * ------------------------------------------------------------------- */

    /** The post statuses that can be trained: published, and private when that setting is on. */
    public static function trainable_statuses(): array
    {
        $statuses = ['publish'];
        if (resolver_get_setting('include_private_pages')) {
            $statuses[] = 'private';
        }

        return $statuses;
    }

    /**
     * Whether a post can be trained: its status, and no password unless that
     * setting allows it. Its post type isn't checked here.
     */
    public static function is_trainable(WP_Post $post): bool
    {
        return in_array($post->post_status, self::trainable_statuses(), true)
            && ('' === $post->post_password || resolver_get_setting('include_password_pages'));
    }

    /**
     * WP_Query arguments for the trainable posts of a type, alphabetical.
     */
    public static function trainable_query_args(string $post_type): array
    {
        $args = [
            'post_type'      => $post_type,
            'post_status'    => self::trainable_statuses(),
            'posts_per_page' => -1,
            'orderby'        => 'title',
            'order'          => 'ASC',
        ];
        if (! resolver_get_setting('include_password_pages')) {
            $args['has_password'] = false;
        }

        return $args;
    }

    /**
     * Whether a run currently holds the indexing lock. Advisory only —
     * callers that need a guarantee must go through acquire (via
     * start_selective_index) and handle its false return.
     */
    public static function is_indexing_locked(): bool
    {
        self::check_stalled_run();

        return null !== self::read_lock();
    }

    /**
     * Deal with a run that has stopped making progress. One whose next batch
     * was queued but never started (its loopback was lost, and WP-Cron hasn't
     * run it yet) is queued again. One that stopped partway through a batch,
     * killed without a chance to report it (a server restart, say), is ended
     * as failed, so training can start again. Called wherever a run's state
     * is read: the Training screen, its status polling, and starting a run.
     */
    public static function check_stalled_run(): void
    {
        $lock = self::read_lock();
        if ($lock && time() - $lock['time'] <= self::STALLED_AFTER) {
            return;
        }
        if (! $lock && ! RSVL_Indexing_Status::is_running()) {
            return;
        }

        wp_cache_delete(self::NEXT_BATCH_OPTION, 'options');
        $next = (string) get_option(self::NEXT_BATCH_OPTION, '');
        if ($lock && RSVL_Indexing_Status::is_running() && preg_match('/^(.+):(\d+)$/', $next, $match) && $match[1] === $lock['run']) {
            self::record_progress($lock['run']);
            self::queue_batch($match[1], (int) $match[2]);

            return;
        }

        if (RSVL_Indexing_Status::is_running()) {
            $status = RSVL_Indexing_Status::get();
            RSVL_Indexing_Status::fail((int) $status['total'], (int) $status['done'], __('Training stopped responding, so it was ended. Pages it didn\'t reach keep their earlier training: train them again to finish.', 'resolver-ai-chatbot'));
        }
        self::end_run($lock['run'] ?? '');
    }

    /* ---------------------------------------------------------------------
     * Background runs and their lock
     * ------------------------------------------------------------------- */

    /**
     * A run is over: its queue, any batch still scheduled, and its lock go.
     * The status is written first, so a run is never seen running without
     * its lock.
     */
    private static function end_run(string $batch_id): void
    {
        global $wpdb;

        delete_option(self::NEXT_BATCH_OPTION);
        delete_option('resolver_training_queue');
        wp_unschedule_hook('resolver_bulk_index_batch');

        // Only this run's lock: another run may already hold a new one.
        $wpdb->query($wpdb->prepare(
            "DELETE FROM {$wpdb->options} WHERE option_name = %s AND option_value LIKE %s",
            self::INDEXING_LOCK_OPTION,
            '%|' . $wpdb->esc_like($batch_id)
        ));
    }

    /**
     * End the run of a batch that died, as failed, with the PHP error that
     * stopped it when there is one. Called on shutdown.
     */
    private static function fail_stopped_batch(string $batch_id): void
    {
        $error  = error_get_last();
        $fatal  = $error && ($error['type'] & (E_ERROR | E_PARSE | E_CORE_ERROR | E_COMPILE_ERROR | E_USER_ERROR | E_RECOVERABLE_ERROR));
        $reason = $fatal ? mb_substr((string) strtok($error['message'], "\n"), 0, 300) : __('The batch ended unexpectedly.', 'resolver-ai-chatbot');

        resolver_log('Training batch stopped: ' . $reason);

        $status = RSVL_Indexing_Status::get();
        /* translators: %s: why training stopped, e.g. a PHP error message. */
        RSVL_Indexing_Status::fail((int) $status['total'], (int) $status['done'], sprintf(__('Training stopped: %s', 'resolver-ai-chatbot'), $reason));
        self::end_run($batch_id);
    }

    /**
     * Queue a batch of a run: start it now, through a loopback request, with a
     * WP-Cron event a minute later in case the loopback never arrives (some
     * hosts block them). Whichever starts it first claims it.
     */
    private static function queue_batch(string $batch_id, int $offset): void
    {
        update_option(self::NEXT_BATCH_OPTION, $batch_id . ':' . $offset, false);
        wp_schedule_single_event(time() + MINUTE_IN_SECONDS, 'resolver_bulk_index_batch', [$batch_id, $offset]);
        self::loopback('resolver_run_index_batch', ['batch_id' => $batch_id, 'offset' => $offset]);
    }

    /**
     * Claim a queued batch, so it runs once although two triggers can start
     * it. A single UPDATE that only matches while the batch is still the run's
     * next makes this atomic: the first trigger clears the option, and the
     * second then finds nothing to claim.
     */
    private static function claim_batch(string $batch_id, int $offset): bool
    {
        global $wpdb;

        $claimed = (bool) $wpdb->update(
            $wpdb->options,
            ['option_value' => ''],
            ['option_name'  => self::NEXT_BATCH_OPTION, 'option_value' => $batch_id . ':' . $offset]
        );
        wp_cache_delete(self::NEXT_BATCH_OPTION, 'options');

        // The WP-Cron event is only needed while the batch is unclaimed.
        if ($claimed) {
            wp_clear_scheduled_hook('resolver_bulk_index_batch', [$batch_id, $offset]);
        }

        return $claimed;
    }

    /**
     * Fire a non-blocking loopback request to one of the indexing AJAX
     * endpoints, so work starts now instead of waiting for WP-Cron.
     */
    private static function loopback(string $action, array $args = []): void
    {
        wp_remote_post(admin_url('admin-ajax.php'), [
            'timeout'   => 0.01,
            'blocking'  => false,
            'sslverify' => false,
            'body'      => array_merge([
                'action' => $action,
                'nonce'  => wp_create_nonce('resolver_index_batch'),
            ], $args),
            'cookies' => $_COOKIE,
        ]);
    }

    /**
     * Take the indexing lock for a new run. Returns false if another run
     * holds it; a stalled one has already been ended by check_stalled_run().
     */
    private static function acquire_indexing_lock(string $batch_id): bool
    {
        global $wpdb;

        return (bool) $wpdb->query($wpdb->prepare(
            "INSERT IGNORE INTO {$wpdb->options} ( option_name, option_value, autoload ) VALUES ( %s, %s, 'no' )",
            self::INDEXING_LOCK_OPTION,
            time() . '|' . $batch_id
        ));
    }

    /**
     * The run holding the lock, and when it last made progress:
     * ['run' => batch ID, 'time' => timestamp], or null when there's none.
     */
    private static function read_lock(): ?array
    {
        global $wpdb;

        $value = (string) $wpdb->get_var($wpdb->prepare(
            "SELECT option_value FROM {$wpdb->options} WHERE option_name = %s",
            self::INDEXING_LOCK_OPTION
        ));
        if ('' === $value) {
            return null;
        }

        [$time, $run] = array_pad(explode('|', $value, 2), 2, '');

        return ['run' => $run, 'time' => (int) $time];
    }

    /** Whether a batch's run is still the one holding the lock. */
    private static function is_current_run(string $batch_id): bool
    {
        return (self::read_lock()['run'] ?? null) === $batch_id;
    }

    /** Record that a run is making progress, so it isn't taken for stalled. */
    private static function record_progress(string $batch_id): void
    {
        global $wpdb;

        $wpdb->query($wpdb->prepare(
            "UPDATE {$wpdb->options} SET option_value = %s WHERE option_name = %s AND option_value LIKE %s",
            time() . '|' . $batch_id,
            self::INDEXING_LOCK_OPTION,
            '%|' . $wpdb->esc_like($batch_id)
        ));
    }
}