<?php

declare(strict_types=1);

namespace WordPress\CloudflareWorkersAiProvider\Embeddings;

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

/**
 * {prefix}aipcf_index: one row per embedded object (not per chunk). The chunk
 * count is enough to know which vector IDs an object owns.
 *
 * Statuses: pending (to embed), indexed, deleting (vectors to remove),
 * failed (gave up; reason kept), skipped (eligible but has no text).
 * A row whose signature differs from Settings::signature() is stale: it was
 * written into an index/model that is no longer current.
 */
final class IndexTable
{
    /** 2: revision column (1.9.0). */
    public const DB_VERSION = 5; // 3: related_at (related posts); 4: linked_at (link assistant); 5: word_count (1.10.1)

    public const VERSION_OPTION = 'aipcf_index_db';

    public const PENDING  = 'pending';
    public const INDEXED  = 'indexed';
    public const DELETING = 'deleting';
    public const FAILED   = 'failed';
    public const SKIPPED  = 'skipped';

    private const MAX_ERROR = 250;

    public static function name(): string
    {
        global $wpdb;
        return $wpdb->prefix . 'aipcf_index';
    }

    public static function install(): void
    {
        $was = (int) get_option(self::VERSION_OPTION);
        if ($was === self::DB_VERSION && self::exists()) {
            return;
        }
        global $wpdb;
        require_once ABSPATH . 'wp-admin/includes/upgrade.php';
        $table   = self::name();
        $charset = $wpdb->get_charset_collate();
        dbDelta("CREATE TABLE {$table} (
  id bigint(20) unsigned NOT NULL AUTO_INCREMENT,
  object_type varchar(20) NOT NULL,
  object_id bigint(20) unsigned NOT NULL,
  status varchar(12) NOT NULL,
  content_hash char(40) NOT NULL DEFAULT '',
  chunk_count smallint(5) unsigned NOT NULL DEFAULT 0,
  word_count int(10) unsigned NOT NULL DEFAULT 0,
  signature char(12) NOT NULL DEFAULT '',
  attempts tinyint(3) unsigned NOT NULL DEFAULT 0,
  revision int(10) unsigned NOT NULL DEFAULT 0,
  last_error varchar(255) NOT NULL DEFAULT '',
  queued_at datetime NOT NULL,
  indexed_at datetime NULL,
  related_at datetime NULL,
  linked_at datetime NULL,
  PRIMARY KEY  (id),
  UNIQUE KEY object (object_type,object_id),
  KEY status (status,queued_at)
) {$charset};");
        update_option(self::VERSION_OPTION, self::DB_VERSION, false);
        if ($was > 0 && $was < 5) {
            // Rows embedded before word counts: check them again. Unchanged text is not sent
            // to the AI again, so this only reads the posts and fills in their word counts.
            self::requeueIndexed();
        }
    }

    public static function exists(): bool
    {
        global $wpdb;
        // Probed rather than SHOW TABLES, which cannot see temporary tables (the test suite uses them).
        $suppress = $wpdb->suppress_errors(true);
        $wpdb->get_var($wpdb->prepare('SELECT 1 FROM %i LIMIT 1', self::name())); // phpcs:ignore WordPress.DB.DirectDatabaseQuery
        $ok = $wpdb->last_error === '';
        $wpdb->suppress_errors($suppress);
        return $ok;
    }

    public static function drop(): void
    {
        global $wpdb;
        $wpdb->query($wpdb->prepare('DROP TABLE IF EXISTS %i', self::name())); // phpcs:ignore WordPress.DB.DirectDatabaseQuery, WordPress.DB.DirectDatabaseQuery.SchemaChange
        delete_option(self::VERSION_OPTION);
    }

    public static function objectType(\WP_Post $post): string
    {
        return $post->post_type === 'attachment' ? 'attachment' : 'post';
    }

    /**
     * @return array<string, mixed>|null
     */
    public static function get(string $type, int $id): ?array
    {
        global $wpdb;
        $row = $wpdb->get_row($wpdb->prepare('SELECT * FROM %i WHERE object_type = %s AND object_id = %d', self::name(), $type, $id), ARRAY_A); // phpcs:ignore WordPress.DB.DirectDatabaseQuery
        return is_array($row) ? self::cast($row) : null;
    }

    /**
     * Queues an object to be (re)embedded. A failed or skipped row is retried, since the content changed.
     */
    public static function markPending(string $type, int $id): void
    {
        global $wpdb;
        $wpdb->query($wpdb->prepare( // phpcs:ignore WordPress.DB.DirectDatabaseQuery
            "INSERT INTO %i (object_type, object_id, status, queued_at) VALUES (%s, %d, %s, %s)
             ON DUPLICATE KEY UPDATE status = %s, attempts = 0, last_error = '', queued_at = VALUES(queued_at), revision = revision + 1",
            self::name(),
            $type,
            $id,
            self::PENDING,
            current_time('mysql', true),
            self::PENDING
        ));
    }

    /**
     * Queues many objects of one type at once (the reconcile's add phase): one
     * INSERT per call instead of one per object.
     *
     * @param list<int> $ids
     */
    public static function markPendingMany(string $type, array $ids): void
    {
        global $wpdb;
        $ids = array_values(array_unique(array_map('intval', $ids)));
        if ($ids === []) {
            return;
        }
        $now    = current_time('mysql', true);
        $values = [];
        $args   = [self::name()];
        foreach ($ids as $id) {
            $values[] = '(%s, %d, %s, %s)';
            array_push($args, $type, $id, self::PENDING, $now);
        }
        $args[] = self::PENDING;
        // phpcs:ignore WordPress.DB.DirectDatabaseQuery, WordPress.DB.PreparedSQL.NotPrepared, WordPress.DB.PreparedSQL.InterpolatedNotPrepared, WordPress.DB.PreparedSQLPlaceholders.ReplacementsWrongNumber, PluginCheck.Security.DirectDB.UnescapedDBParameter -- a literal "(%s, %d, %s, %s)" group per ID; every value goes through prepare()
        $wpdb->query($wpdb->prepare('INSERT INTO %i (object_type, object_id, status, queued_at) VALUES ' . implode(',', $values) . " ON DUPLICATE KEY UPDATE status = %s, attempts = 0, last_error = '', queued_at = VALUES(queued_at), revision = revision + 1", ...$args));
    }

    /**
     * Marks an object's vectors for removal. A row that never reached the store is simply dropped.
     */
    public static function markDeleting(string $type, int $id): void
    {
        $row = self::get($type, $id);
        if ($row === null) {
            return;
        }
        if ((int) $row['chunk_count'] === 0) {
            self::delete($type, $id);
            return;
        }
        global $wpdb;
        $wpdb->query($wpdb->prepare('UPDATE %i SET status = %s, queued_at = %s, revision = revision + 1 WHERE object_type = %s AND object_id = %d', self::name(), self::DELETING, current_time('mysql', true), $type, $id)); // phpcs:ignore WordPress.DB.DirectDatabaseQuery
    }

    /**
     * @param array<string, mixed> $fields
     */
    public static function update(string $type, int $id, array $fields): void
    {
        global $wpdb;
        if (isset($fields['last_error'])) {
            $fields['last_error'] = mb_substr((string) $fields['last_error'], 0, self::MAX_ERROR);
        }
        $wpdb->update(self::name(), $fields, ['object_type' => $type, 'object_id' => $id]); // phpcs:ignore WordPress.DB.DirectDatabaseQuery
    }

    /**
     * Updates the row only if no hook queued or deleted it since it was read at
     * $revision (a save, trash or delete while the worker was embedding it; the
     * revision counts those, where queued_at would miss one within the same second).
     * Returns false when it had changed.
     *
     * @param array<string, mixed> $fields
     */
    public static function updateIfUnchanged(string $type, int $id, array $fields, int $revision): bool
    {
        global $wpdb;
        $n = $wpdb->update(self::name(), $fields, ['object_type' => $type, 'object_id' => $id, 'revision' => $revision]); // phpcs:ignore WordPress.DB.DirectDatabaseQuery
        if ((int) $n > 0) {
            return true;
        }
        // 0 rows can also mean "matched, nothing to change": look again.
        $row = self::get($type, $id);
        return $row !== null && (int) $row['revision'] === $revision;
    }

    public static function delete(string $type, int $id): void
    {
        global $wpdb;
        $wpdb->delete(self::name(), ['object_type' => $type, 'object_id' => $id]); // phpcs:ignore WordPress.DB.DirectDatabaseQuery
    }

    /**
     * Rows to remove from the store, oldest first.
     *
     * @return list<array<string, mixed>>
     */
    public static function deleting(int $limit): array
    {
        global $wpdb;
        $rows = $wpdb->get_results($wpdb->prepare('SELECT * FROM %i WHERE status = %s ORDER BY queued_at, id LIMIT %d', self::name(), self::DELETING, $limit), ARRAY_A); // phpcs:ignore WordPress.DB.DirectDatabaseQuery
        return array_map([self::class, 'cast'], (array) $rows);
    }

    /**
     * Rows to embed: pending ones, then indexed ones written under another signature (unless
     * a move is copying those: $stale false).
     *
     * @return list<array<string, mixed>>
     */
    public static function work(string $signature, int $limit, bool $stale = true): array
    {
        global $wpdb;
        if (!$stale) {
            $rows = $wpdb->get_results($wpdb->prepare('SELECT * FROM %i WHERE status = %s ORDER BY queued_at, id LIMIT %d', self::name(), self::PENDING, $limit), ARRAY_A); // phpcs:ignore WordPress.DB.DirectDatabaseQuery
            return array_map([self::class, 'cast'], (array) $rows);
        }
        $rows = $wpdb->get_results($wpdb->prepare( // phpcs:ignore WordPress.DB.DirectDatabaseQuery
            'SELECT * FROM %i WHERE status = %s OR (status = %s AND signature <> %s)
             ORDER BY status = %s DESC, queued_at, id LIMIT %d',
            self::name(),
            self::PENDING,
            self::INDEXED,
            $signature,
            self::PENDING,
            $limit
        ), ARRAY_A);
        return array_map([self::class, 'cast'], (array) $rows);
    }

    /**
     * Indexed rows written under one signature (a move copies them), oldest first.
     *
     * @return list<array<string, mixed>>
     */
    public static function indexedWithSignature(string $signature, int $limit): array
    {
        global $wpdb;
        if (!self::exists()) {
            return [];
        }
        $rows = $wpdb->get_results($wpdb->prepare('SELECT * FROM %i WHERE status = %s AND signature = %s ORDER BY id LIMIT %d', self::name(), self::INDEXED, $signature, $limit), ARRAY_A); // phpcs:ignore WordPress.DB.DirectDatabaseQuery
        return array_map([self::class, 'cast'], (array) $rows);
    }

    /**
     * Indexed rows and their vectors under one signature, for a move's progress.
     *
     * @return array{rows: int, vectors: int}
     */
    public static function signatureCounts(string $signature): array
    {
        global $wpdb;
        if (!self::exists()) {
            return ['rows' => 0, 'vectors' => 0];
        }
        $row = $wpdb->get_row($wpdb->prepare('SELECT COUNT(*) AS n, SUM(chunk_count) AS c FROM %i WHERE status = %s AND signature = %s', self::name(), self::INDEXED, $signature), ARRAY_A); // phpcs:ignore WordPress.DB.DirectDatabaseQuery
        return ['rows' => (int) ($row['n'] ?? 0), 'vectors' => (int) ($row['c'] ?? 0)];
    }

    /**
     * @return array{pending: int, indexed: int, stale: int, deleting: int, failed: int, skipped: int, chunks: int, words: int}
     */
    public static function counts(string $signature): array
    {
        global $wpdb;
        $out = ['pending' => 0, 'indexed' => 0, 'stale' => 0, 'deleting' => 0, 'failed' => 0, 'skipped' => 0, 'chunks' => 0, 'words' => 0];
        if (!self::exists()) {
            return $out;
        }
        $rows = $wpdb->get_results($wpdb->prepare( // phpcs:ignore WordPress.DB.DirectDatabaseQuery
            'SELECT status, signature = %s AS current, COUNT(*) AS n, SUM(chunk_count) AS chunks, SUM(word_count) AS words FROM %i GROUP BY status, current',
            $signature,
            self::name()
        ), ARRAY_A);
        foreach ((array) $rows as $r) {
            $status = (string) $r['status'];
            $n      = (int) $r['n'];
            if ($status === self::INDEXED && !(int) $r['current']) {
                $out['stale'] += $n;
                continue;
            }
            if (isset($out[$status])) {
                $out[$status] += $n;
            }
            if ($status === self::INDEXED) {
                $out['chunks'] += (int) $r['chunks'];
                $out['words']  += (int) $r['words'];
            }
        }
        return $out;
    }

    /**
     * When the longest-waiting row was queued (GMT), or '' when nothing waits.
     */
    public static function oldestWaiting(): string
    {
        global $wpdb;
        if (!self::exists()) {
            return '';
        }
        return (string) $wpdb->get_var($wpdb->prepare('SELECT MIN(queued_at) FROM %i WHERE status IN (%s, %s)', self::name(), self::PENDING, self::DELETING)); // phpcs:ignore WordPress.DB.DirectDatabaseQuery
    }

    /**
     * The most recent failures, with their reasons.
     *
     * @return list<array<string, mixed>>
     */
    public static function failures(int $limit = 10): array
    {
        global $wpdb;
        $rows = $wpdb->get_results($wpdb->prepare('SELECT * FROM %i WHERE status = %s ORDER BY queued_at DESC LIMIT %d', self::name(), self::FAILED, $limit), ARRAY_A); // phpcs:ignore WordPress.DB.DirectDatabaseQuery
        return array_map([self::class, 'cast'], (array) $rows);
    }

    /**
     * Queues every indexed row again (what is embedded per item changed); the
     * worker re-embeds only those whose text actually differs.
     */
    public static function requeueIndexed(): int
    {
        global $wpdb;
        if (!self::exists()) {
            return 0;
        }
        return (int) $wpdb->query($wpdb->prepare('UPDATE %i SET status = %s WHERE status = %s', self::name(), self::PENDING, self::INDEXED)); // phpcs:ignore WordPress.DB.DirectDatabaseQuery
    }

    /**
     * Makes every indexed row look changed, so the worker embeds and writes it
     * again under the same IDs (an in-place rebuild).
     */
    public static function forceReembed(): int
    {
        global $wpdb;
        if (!self::exists()) {
            return 0;
        }
        return (int) $wpdb->query($wpdb->prepare("UPDATE %i SET status = %s, content_hash = '' WHERE status = %s", self::name(), self::PENDING, self::INDEXED)); // phpcs:ignore WordPress.DB.DirectDatabaseQuery
    }

    /**
     * Queues skipped rows again (the minimum length changed, so some may now qualify).
     */
    public static function requeueSkipped(): int
    {
        global $wpdb;
        if (!self::exists()) {
            return 0;
        }
        return (int) $wpdb->query($wpdb->prepare("UPDATE %i SET status = %s, attempts = 0, last_error = '' WHERE status = %s", self::name(), self::PENDING, self::SKIPPED)); // phpcs:ignore WordPress.DB.DirectDatabaseQuery
    }

    public static function retryFailed(): int
    {
        global $wpdb;
        return (int) $wpdb->query($wpdb->prepare("UPDATE %i SET status = %s, attempts = 0, last_error = '' WHERE status IN (%s, %s)", self::name(), self::PENDING, self::FAILED, self::SKIPPED)); // phpcs:ignore WordPress.DB.DirectDatabaseQuery
    }

    /**
     * One page of rows by id, for the reconcile sweep.
     *
     * @return list<array<string, mixed>>
     */
    /**
     * Indexed posts whose related list is missing or older than their vectors, and whose
     * vectors are old enough to be visible to queries (Vectorize takes about a minute).
     *
     * @return list<array<string, mixed>>
     */
    public static function relatedDue(int $limit, string $indexedBefore): array
    {
        global $wpdb;
        if (!self::exists()) {
            return [];
        }
        $rows = $wpdb->get_results($wpdb->prepare( // phpcs:ignore WordPress.DB.DirectDatabaseQuery
            "SELECT * FROM %i WHERE status = %s AND object_type = 'post' AND indexed_at <= %s AND (related_at IS NULL OR related_at < indexed_at) ORDER BY indexed_at LIMIT %d",
            self::name(),
            self::INDEXED,
            $indexedBefore,
            $limit
        ), ARRAY_A);
        return array_map([self::class, 'cast'], (array) $rows);
    }

    /**
     * Indexed posts whose terms and links are missing or older than their last embedding.
     *
     * @return list<int> post IDs
     */
    public static function linksDue(int $limit): array
    {
        global $wpdb;
        if (!self::exists()) {
            return [];
        }
        return array_map('intval', (array) $wpdb->get_col($wpdb->prepare( // phpcs:ignore WordPress.DB.DirectDatabaseQuery
            "SELECT object_id FROM %i WHERE status = %s AND object_type = 'post' AND (linked_at IS NULL OR linked_at < indexed_at) ORDER BY id LIMIT %d",
            self::name(),
            self::INDEXED,
            $limit
        )));
    }

    public static function markLinked(int $id): void
    {
        global $wpdb;
        $wpdb->query($wpdb->prepare("UPDATE %i SET linked_at = %s WHERE object_type = 'post' AND object_id = %d", self::name(), current_time('mysql', true), $id)); // phpcs:ignore WordPress.DB.DirectDatabaseQuery
    }

    /**
     * Makes every post's related list due again.
     */
    public static function resetRelated(): int
    {
        global $wpdb;
        if (!self::exists()) {
            return 0;
        }
        return (int) $wpdb->query($wpdb->prepare('UPDATE %i SET related_at = NULL WHERE related_at IS NOT NULL', self::name())); // phpcs:ignore WordPress.DB.DirectDatabaseQuery
    }

    public static function markRelated(int $id): void
    {
        global $wpdb;
        $wpdb->query($wpdb->prepare("UPDATE %i SET related_at = %s WHERE object_type = 'post' AND object_id = %d", self::name(), current_time('mysql', true), $id)); // phpcs:ignore WordPress.DB.DirectDatabaseQuery
    }

    /**
     * Indexed posts with a current related list, and those still waiting for one.
     *
     * @return array{done: int, waiting: int}
     */
    public static function relatedCounts(): array
    {
        global $wpdb;
        if (!self::exists()) {
            return ['done' => 0, 'waiting' => 0];
        }
        $row = $wpdb->get_row($wpdb->prepare( // phpcs:ignore WordPress.DB.DirectDatabaseQuery
            "SELECT SUM(related_at IS NOT NULL AND related_at >= indexed_at) AS done, SUM(related_at IS NULL OR related_at < indexed_at) AS waiting FROM %i WHERE status = %s AND object_type = 'post'",
            self::name(),
            self::INDEXED
        ), ARRAY_A);
        return ['done' => (int) ($row['done'] ?? 0), 'waiting' => (int) ($row['waiting'] ?? 0)];
    }

    public static function page(int $afterId, int $limit): array
    {
        global $wpdb;
        $rows = $wpdb->get_results($wpdb->prepare('SELECT * FROM %i WHERE id > %d ORDER BY id LIMIT %d', self::name(), $afterId, $limit), ARRAY_A); // phpcs:ignore WordPress.DB.DirectDatabaseQuery
        return array_map([self::class, 'cast'], (array) $rows);
    }

    /**
     * Which of these object IDs already have a row.
     *
     * @param list<int> $ids
     * @return list<int>
     */
    public static function known(string $type, array $ids): array
    {
        global $wpdb;
        if ($ids === []) {
            return [];
        }
        $ids          = array_values(array_map('intval', $ids));
        $placeholders = implode(',', array_fill(0, count($ids), '%d'));
        // phpcs:ignore WordPress.DB.DirectDatabaseQuery, WordPress.DB.PreparedSQL.InterpolatedNotPrepared, WordPress.DB.PreparedSQLPlaceholders.ReplacementsWrongNumber -- $placeholders is only "%d,%d,...", one per spread ID
        return array_map('intval', (array) $wpdb->get_col($wpdb->prepare("SELECT object_id FROM %i WHERE object_type = %s AND object_id IN ({$placeholders})", self::name(), $type, ...$ids)));
    }

    /**
     * Which of these object IDs the index skipped (no text, or under the minimum length).
     *
     * @param list<int> $ids
     * @return list<int>
     */
    public static function skippedAmong(string $type, array $ids): array
    {
        global $wpdb;
        $ids = array_values(array_filter(array_map('intval', $ids)));
        if ($ids === [] || !self::exists()) {
            return [];
        }
        $placeholders = implode(',', array_fill(0, count($ids), '%d'));
        // phpcs:ignore WordPress.DB.DirectDatabaseQuery, WordPress.DB.PreparedSQL.InterpolatedNotPrepared, WordPress.DB.PreparedSQLPlaceholders.ReplacementsWrongNumber -- $placeholders is only "%d,%d,...", one per spread ID
        return array_map('intval', (array) $wpdb->get_col($wpdb->prepare("SELECT object_id FROM %i WHERE object_type = %s AND status = %s AND object_id IN ({$placeholders})", self::name(), $type, self::SKIPPED, ...$ids)));
    }

    public static function truncate(): void
    {
        global $wpdb;
        if (self::exists()) {
            // DELETE, not TRUNCATE: TRUNCATE commits any open transaction in MySQL.
            $wpdb->query($wpdb->prepare('DELETE FROM %i', self::name())); // phpcs:ignore WordPress.DB.DirectDatabaseQuery
        }
    }

    /**
     * @param array<string, mixed> $row
     * @return array<string, mixed>
     */
    private static function cast(array $row): array
    {
        foreach (['id', 'object_id', 'chunk_count', 'word_count', 'attempts', 'revision'] as $k) {
            $row[$k] = (int) ($row[$k] ?? 0);
        }
        return $row;
    }
}