<?php
// SPDX-License-Identifier: GPL-3.0-or-later
declare(strict_types=1);

namespace Aiwend\Knowledge;

use Aiwend\Db\Connection;
use Aiwend\Support\Clock;

/**
 * Staged, versioned indexing (B05):
 *   extract → chunk → embed → write chunks under a new version (state processing)
 *   → in one transaction: publish the version, supersede the old one, drop its chunks.
 *
 * A failed new version never replaces or mixes with the published one. Its
 * partial chunks are removed and the previous version keeps serving.
 * Re-ingesting identical content with the same embedding model is a no-op.
 */
final class Ingestor
{
    public const BATCH = 32;
    /** Indexing is accounted on its own bucket, never on the chat buckets (owner decision D1). */
    public const EMBEDDING_BUCKETS = ['embedding'];

    public function __construct(
        private Connection $db,
        private SourceRepository $sources,
        private Extractor $extractor,
        private Chunker $chunker,
        private Embedder $embedder,
        private Clock $clock,
        private int $timeBudgetSeconds = 120,
        private ?\Closure $afterBatch = null,
        private ?string $embeddingOperation = null,
        private array $embeddingBuckets = self::EMBEDDING_BUCKETS,
        private ?\Closure $beforeBatch = null,
        private ?\Closure $onIndexed = null,
    ) {
    }

    /**
     * The most searchable passages (chunks) one site can hold. Retrieval scans at most Retriever::MAX_SCAN of them, so anything
     * beyond that would silently never be found; a source that would push the library over the limit is refused with a clear
     * message instead (Phase 86). Measured cost of a question grows linearly with this number (see the Phase 86 review).
     */
    public const SITE_CHUNK_LIMIT = Retriever::MAX_SCAN;

    /** Above this many searchable passages the Knowledge page warns that questions are getting slow (measured: about 0.4 s per 1,000 on a slow host). */
    public const SITE_CHUNK_WARNING = 5000;

    private int $siteChunkLimit = self::SITE_CHUNK_LIMIT;

    /** Lowers the site limit (tests only; the production limit is the constant). */
    public function withSiteChunkLimit(int $limit): self
    {
        $this->siteChunkLimit = max(1, $limit);
        return $this;
    }
    /**
     * @return array{outcome:string, version_id:?int, chunks:int, reason:?string}
     *   outcome: indexed | unchanged | rejected | skipped
     * @throws \Throwable for retryable failures (after cleanup)
     */
    public function ingest(int $siteId, int $sourceId, string $raw, string $type): array
    {
        $source = $this->sources->find($siteId, $sourceId);
        if ($source === null || in_array($source['status'], ['deleting', 'deleted'], true)) {
            // A retry or recovery run for an erased source must not leave (or revive) derivatives,
            // e.g. from an earlier run that crashed after the purge.
            $this->removeDerivatives($siteId, $sourceId);
            return ['outcome' => 'skipped', 'version_id' => null, 'chunks' => 0, 'reason' => 'source_unavailable'];
        }
        $hash = self::contentHash($type, $raw);
        if ($source['current_version_id'] !== null) {
            $current = $this->db->selectOne('SELECT content_hash, embedding_model FROM {knowledge_versions} WHERE id = ?', [$source['current_version_id']]);
            if ($current !== null && $current['content_hash'] === $hash && $current['embedding_model'] === $this->embedder->model()) {
                $this->sources->setStatus($siteId, $sourceId, 'ready');
                $this->indexed($siteId, $sourceId);
                return ['outcome' => 'unchanged', 'version_id' => $source['current_version_id'], 'chunks' => 0, 'reason' => null];
            }
        }

        // A ready source that already serves a published version keeps that status while the new version is built, so a
        // re-index, a post update or a file replacement never takes its current text out of answers; only a first index
        // (or a source that is not ready, e.g. an unpublished post being restored) is "processing". A concurrent delete still wins (see publish()).
        if ($source['current_version_id'] === null || $source['status'] !== 'ready') {
            if (!$this->sources->setStatus($siteId, $sourceId, 'processing')) {
                return ['outcome' => 'skipped', 'version_id' => null, 'chunks' => 0, 'reason' => 'source_unavailable'];
            }
        } elseif ($this->sourceErased($siteId, $sourceId)) {
            return ['outcome' => 'skipped', 'version_id' => null, 'chunks' => 0, 'reason' => 'source_unavailable'];
        }
        $now = $this->clock->now();
        $next = (int) ($this->db->selectOne('SELECT COALESCE(MAX(version_no), 0) AS v FROM {knowledge_versions} WHERE source_id = ?', [$sourceId])['v'] ?? 0) + 1;
        $versionId = $this->db->insert('knowledge_versions', [
            'site_id' => $siteId, 'source_id' => $sourceId, 'version_no' => $next, 'content_hash' => $hash,
            'state' => 'processing', 'embedding_model' => $this->embedder->model(), 'dimension' => $this->embedder->dimension(),
            'chunk_count' => 0, 'error' => null, 'created_at' => $now, 'updated_at' => $now,
        ]);

        $started = $this->clock->now();
        try {
            $count = $this->indexChunks($siteId, $sourceId, $versionId, (string) $source['title'], $raw, $type, $now, $started);
            $this->publish($siteId, $sourceId, $versionId, $count, $hash);
            $this->indexed($siteId, $sourceId);
            return ['outcome' => 'indexed', 'version_id' => $versionId, 'chunks' => $count, 'reason' => null];
        } catch (ExtractionException $e) {
            $this->discard($siteId, $sourceId, $versionId, $e->getMessage(), $e->reason);
            return ['outcome' => 'rejected', 'version_id' => null, 'chunks' => 0, 'reason' => $e->reason];
        } catch (\Throwable $e) {
            $this->discard($siteId, $sourceId, $versionId, 'Indexing failed temporarily; it will be retried.', null);
            if ($this->sourceErased($siteId, $sourceId)) {
                // Erased during indexing: nothing to retry, and its derivatives are already gone.
                return ['outcome' => 'skipped', 'version_id' => null, 'chunks' => 0, 'reason' => 'source_unavailable'];
            }
            throw $e;
        }
    }

    /**
     * Extracts, chunks, embeds and stores the chunks of one version (shared by the single-index path and generation builds).
     * Returns the number of chunks. Throws on any failure; the caller discards the version.
     */
    private function indexChunks(int $siteId, int $sourceId, int $versionId, string $title, string $raw, string $type, int $now, int $started): int
    {
        $doc = $this->extractor->extract($raw, $type, $title);
        $chunks = $this->chunker->chunk($doc);
        $existing = (int) ($this->db->selectOne(
            "SELECT COUNT(*) AS n FROM {knowledge_chunks} c JOIN {knowledge_sources} s ON s.id = c.source_id AND s.current_version_id = c.version_id
             WHERE s.site_id = ? AND s.id <> ? AND s.status NOT IN ('deleting', 'deleted')",
            [$siteId, $sourceId]
        )['n'] ?? 0);
        if ($existing + count($chunks) > $this->siteChunkLimit) {
            // phpcs:ignore WordPress.Security.EscapeOutput.ExceptionNotEscaped -- exception messages go to logs and callers, never echoed; public errors are fixed codes
            throw new ExtractionException('site_chunk_limit', sprintf('The knowledge library is full: it holds %d searchable passages and this source adds %d, but the limit is %d. Delete sources you no longer need, or shorten this one.', $existing, count($chunks), $this->siteChunkLimit));
        }
        foreach (array_chunk($chunks, self::BATCH) as $batchNo => $batch) {
            if ($batchNo > 0 && $this->sourceErased($siteId, $sourceId)) {
                throw new SourceErasedException();
            }
            if ($this->clock->now() - $started > $this->timeBudgetSeconds) {
                throw new \RuntimeException('Indexing exceeded its time budget; it will be retried.');
            }
            $texts = array_map(static fn(array $c): string => ($c['heading'] !== '' ? $c['heading'] . "\n" : '') . $c['text'], $batch);
            if ($this->beforeBatch !== null) {
                // Re-proves ownership immediately before every request: a worker that lost its claim never begins another one.
                ($this->beforeBatch)($sourceId, $batchNo);
            }
            if ($this->embedder instanceof RemoteEmbedder && $this->embeddingOperation === null) { throw new ExtractionException('embedding_context', 'Explicit indexing operation required.'); }
            $vectors = $this->embedder instanceof RemoteEmbedder
                ? $this->embedder->embedFor($texts, self::batchOperation((string) $this->embeddingOperation, $batchNo), $this->embeddingBuckets)
                : $this->embedder->embed($texts);
            foreach ($batch as $i => $c) {
                if (count($vectors[$i]) !== $this->embedder->dimension()) {
                    throw new \UnexpectedValueException('Embedding dimension mismatch.');
                }
                $this->db->insert('knowledge_chunks', [
                    'site_id' => $siteId, 'source_id' => $sourceId, 'version_id' => $versionId, 'seq' => $c['seq'],
                    'anchor' => mb_substr($c['anchor'], 0, 191), 'heading' => mb_substr($c['heading'], 0, 191),
                    'text' => $c['text'], 'embedding' => VectorCodec::encode($vectors[$i]), 'created_at' => $now,
                ]);
            }
            if ($this->afterBatch !== null) {
                ($this->afterBatch)($sourceId, $batchNo);
            }
        }
        return count($chunks);
    }

    /**
     * Builds one source into a BUILDING index generation (see IndexGenerations) without touching the source's status, its current
     * version, the lexical index or any other generation: the new version carries the generation id, is never the source's
     * current_version_id and is invisible to the single-index path. An older ready version of the same source in the same
     * generation is replaced atomically when the new one is published. Safe to repeat (failed versions of this source and
     * generation are removed first). Remote embedders still need an explicit operation id, as everywhere.
     *
     * @return array{outcome:string, version_id:?int, chunks:int, reason:?string}  indexed | unchanged | rejected | skipped
     * @throws GenerationException when the generation is not a building generation of this site built for this embedder
     * @throws \Throwable for retryable failures (after cleanup)
     */
    public function ingestForGeneration(int $siteId, int $sourceId, string $raw, string $type, int $generationId, bool $allowActive = false, ?\Closure $fence = null): array
    {
        $gen = $this->db->selectOne('SELECT state, provider, model, dimension FROM {knowledge_generations} WHERE id = ? AND site_id = ?', [$generationId, $siteId]);
        if ($gen === null || !self::writable((string) $gen['state'], $allowActive)) {
            throw new GenerationException('not_building');
        }
        if ($gen['provider'] . ':' . $gen['model'] . ':' . $gen['dimension'] !== $this->embedder->model() || (int) $gen['dimension'] !== $this->embedder->dimension()) {
            throw new GenerationException('incompatible_vectors');
        }
        $source = $this->sources->find($siteId, $sourceId);
        if ($source === null || in_array($source['status'], ['deleting', 'deleted'], true)) {
            return ['outcome' => 'skipped', 'version_id' => null, 'chunks' => 0, 'reason' => 'source_unavailable'];
        }
        $hash = self::contentHash($type, $raw);
        $ready =$this->db->selectOne("SELECT id FROM {knowledge_versions} WHERE site_id = ? AND source_id = ? AND generation_id = ? AND state = 'ready' AND content_hash = ?", [$siteId, $sourceId, $generationId, $hash]);
        if ($ready !== null) {
            return ['outcome' => 'unchanged', 'version_id' => (int) $ready['id'], 'chunks' => 0, 'reason' => null];
        }
        $this->db->transaction(function () use ($siteId, $sourceId, $generationId): void {
            $this->db->execute("DELETE FROM {knowledge_chunks} WHERE site_id = ? AND version_id IN (SELECT id FROM {knowledge_versions} WHERE site_id = ? AND source_id = ? AND generation_id = ? AND state = 'failed')", [$siteId, $siteId, $sourceId, $generationId]);
            $this->db->execute("DELETE FROM {knowledge_versions} WHERE site_id = ? AND source_id = ? AND generation_id = ? AND state = 'failed'", [$siteId, $sourceId, $generationId]);
        });
        $now = $this->clock->now();
        $next = (int) ($this->db->selectOne('SELECT COALESCE(MAX(version_no), 0) AS v FROM {knowledge_versions} WHERE source_id = ?', [$sourceId])['v'] ?? 0) + 1;
        $versionId = $this->db->insert('knowledge_versions', [
            'site_id' => $siteId, 'source_id' => $sourceId, 'version_no' => $next, 'content_hash' => $hash,
            'state' => 'processing', 'embedding_model' => $this->embedder->model(), 'dimension' => $this->embedder->dimension(),
            'chunk_count' => 0, 'error' => null, 'created_at' => $now, 'updated_at' => $now, 'generation_id' => $generationId,
        ]);
        try {
            $count = $this->indexChunks($siteId, $sourceId, $versionId, (string) $source['title'], $raw, $type, $now, $this->clock->now());
            $this->publishGenerationVersion($siteId, $sourceId, $generationId, $versionId, $count, $allowActive, $fence);
            return ['outcome' => 'indexed', 'version_id' => $versionId, 'chunks' => $count, 'reason' => null];
        } catch (ExtractionException $e) {
            $this->discardGenerationVersion($versionId, $e->getMessage());
            return ['outcome' => 'rejected', 'version_id' => null, 'chunks' => 0, 'reason' => $e->reason];
        } catch (\Throwable $e) {
            $this->discardGenerationVersion($versionId, 'Indexing failed temporarily; it will be retried.');
            if ($this->sourceErased($siteId, $sourceId)) {
                return ['outcome' => 'skipped', 'version_id' => null, 'chunks' => 0, 'reason' => 'source_unavailable'];
            }
            throw $e;
        }
    }

    /** Ready in one transaction together with the removal of the older ready version of this source in this generation. */
    private function publishGenerationVersion(int $siteId, int $sourceId, int $generationId, int $versionId, int $count, bool $allowActive, ?\Closure $fence): void
    {
        $this->db->transaction(function () use ($siteId, $sourceId, $generationId, $versionId, $count, $allowActive, $fence): void {
            $now = $this->clock->now();
            if ($fence !== null) {
                // Ownership is proved INSIDE the transaction (its row lock serializes against a takeover), and the version must still be of
                // the source's current content: a result for text that was replaced meanwhile is never published.
                $current = $this->db->selectOne('SELECT cv.content_hash AS h, v.content_hash AS vh, s.visibility AS vis, s.status AS st FROM {knowledge_sources} s JOIN {knowledge_versions} cv ON cv.id = s.current_version_id JOIN {knowledge_versions} v ON v.id = ? WHERE s.id = ? AND s.site_id = ?', [$versionId, $sourceId, $siteId]);
                if ($current === null || $current['h'] !== $current['vh'] || $current['vis'] !== 'public' || $current['st'] !== 'ready') {
                    throw new SourceChangedException(); // replaced, made private, unpublished or erased while it was embedded
                }
                if (!$fence()) {
                    throw new ClaimLostException();
                }
            }
            $gen = $this->db->selectOne('SELECT state FROM {knowledge_generations} WHERE id = ? AND site_id = ?', [$generationId, $siteId]);
            if ($gen === null || !self::writable((string) $gen['state'], $allowActive) || $this->sourceErased($siteId, $sourceId)) {
                throw new \RuntimeException('Generation or source changed during indexing.');
            }
            $old = $this->db->select("SELECT id FROM {knowledge_versions} WHERE site_id = ? AND source_id = ? AND generation_id = ? AND state = 'ready' AND id <> ?", [$siteId, $sourceId, $generationId, $versionId]);
            $this->db->execute("UPDATE {knowledge_versions} SET state = 'ready', chunk_count = ?, updated_at = ? WHERE id = ? AND site_id = ?", [$count, $now, $versionId, $siteId]);
            foreach ($old as $o) {
                $this->db->execute('DELETE FROM {knowledge_chunks} WHERE version_id = ?', [(int) $o['id']]);
                $this->db->execute('DELETE FROM {knowledge_versions} WHERE id = ?', [(int) $o['id']]);
            }
        });
    }

    /**
     * A generation takes new versions while it is being built, and (only for the dual-write that keeps the live generation current,
     * $allowActive) while it is active. A source's version is swapped in one transaction, so a reader sees the old or the new version
     * of that source, never a mix; a generation that was superseded or failed meanwhile refuses the publish and the version is discarded.
     */
    private static function writable(string $state, bool $allowActive): bool
    {
        return $state === IndexGenerations::BUILDING || ($allowActive && $state === IndexGenerations::ACTIVE);
    }

    public static function batchOperation(string $operation, int $batchNo): string
    {
        return $operation . ':batch:' . $batchNo;
    }

    /** The identity of a source's content, as stored in knowledge_versions.content_hash. */
    public static function contentHash(string $type, string $raw): string
    {
        return hash('sha256', $type . "\0" . $raw);
    }

    private function discardGenerationVersion(int $versionId, string $message): void
    {
        $this->db->execute('DELETE FROM {knowledge_chunks} WHERE version_id = ?', [$versionId]);
        $this->db->execute("UPDATE {knowledge_versions} SET state = 'failed', error = ?, updated_at = ? WHERE id = ?", [mb_substr($message, 0, 500), $this->clock->now(), $versionId]);
    }

    /** Tells the lifecycle (if anyone listens) that the source's served content is current, after the change is committed. */
    private function indexed(int $siteId, int $sourceId): void
    {
        if ($this->onIndexed !== null) {
            ($this->onIndexed)($siteId, $sourceId);
        }
    }

    private function publish(int $siteId, int $sourceId, int $versionId, int $count, string $hash): void
    {
        $this->db->transaction(function () use ($siteId, $sourceId, $versionId, $count, $hash): void {
            $now = $this->clock->now();
            // Only the single index's own versions: a generation's versions of the same source are never superseded or deleted here.
            $old = $this->db->select("SELECT id FROM {knowledge_versions} WHERE source_id = ? AND id <> ? AND state = 'ready' AND generation_id IS NULL", [$sourceId, $versionId]);
            $this->db->execute("UPDATE {knowledge_versions} SET state = 'ready', chunk_count = ?, updated_at = ? WHERE id = ?", [$count, $now, $versionId]);
            // Only a live source is published (first index: processing; update: ready); a concurrent delete wins.
            $published = $this->db->execute(
                "UPDATE {knowledge_sources} SET current_version_id = ?, status = 'ready', error = NULL, updated_at = ? WHERE id = ? AND site_id = ? AND status IN ('processing', 'ready')",
                [$versionId, $now, $sourceId, $siteId]
            );
            if ($published !== 1) {
                throw new \RuntimeException('Source changed during indexing.');
            }
            foreach ($old as $o) {
                $this->db->execute("UPDATE {knowledge_versions} SET state = 'superseded', updated_at = ? WHERE id = ?", [$now, (int) $o['id']]);
                $this->db->execute('DELETE FROM {knowledge_chunks} WHERE version_id = ?', [(int) $o['id']]);
            }
            // Replaced text must not stay in any generation for even a moment: in this same transaction every generation copy of other
            // content goes (the generations pick the new text up through a sync job; until then the lexical index answers).
            GenerationCopies::removeOtherContent($this->db, $siteId, $sourceId, $hash);
        });
    }

    private function discard(int $siteId, int $sourceId, int $versionId, string $message, ?string $reason): void
    {
        $this->db->execute('DELETE FROM {knowledge_chunks} WHERE version_id = ?', [$versionId]);
        $this->db->execute("UPDATE {knowledge_versions} SET state = 'failed', error = ?, updated_at = ? WHERE id = ?", [mb_substr($message, 0, 500), $this->clock->now(), $versionId]);
        $source = $this->sources->find($siteId, $sourceId);
        if ($source === null || in_array($source['status'], ['deleting', 'deleted'], true)) {
            // Deleted while indexing: also drop the failed version row (it holds the content hash).
            $this->removeDerivatives($siteId, $sourceId);
            return;
        }
        if ($source['current_version_id'] !== null) {
            // Keep serving the previous version; record why the update failed.
            $this->sources->setStatus($siteId, $sourceId, 'ready', 'Latest update failed: ' . $message);
        } else {
            $this->sources->setStatus($siteId, $sourceId, $reason === 'ocr_required' ? 'ocr_required' : 'failed', $message);
        }
    }

    private function sourceErased(int $siteId, int $sourceId): bool
    {
        $source = $this->sources->find($siteId, $sourceId);
        return $source === null || in_array($source['status'], ['deleting', 'deleted'], true);
    }

    /** Same local scope as Purger: chunks, versions, bot links and stored uploads of this source. */
    private function removeDerivatives(int $siteId, int $sourceId): void
    {
        $this->db->transaction(function () use ($siteId, $sourceId): void {
            $this->db->execute('DELETE FROM {knowledge_chunks} WHERE source_id = ? AND site_id = ?', [$sourceId, $siteId]);
            $this->db->execute('DELETE FROM {knowledge_versions} WHERE source_id = ? AND site_id = ?', [$sourceId, $siteId]);
            $this->db->execute('DELETE FROM {bot_sources} WHERE source_id = ? AND site_id = ?', [$sourceId, $siteId]);
            $this->db->execute('DELETE FROM {knowledge_blobs} WHERE source_id = ? AND site_id = ?', [$sourceId, $siteId]);
        });
    }
}