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

namespace Aiwend\Knowledge;

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

/**
 * Index generations: complete, versioned indexes of one site's knowledge, built with one provider, model and dimension.
 *
 * State machine (per generation):
 *
 *   building ──finish()──▶ ready ──activate()──▶ active ──(another activate/rollback/deactivate)──▶ superseded ──expiry──▶ (deleted)
 *       │                    │                                                     ▲
 *       └────── fail() ──────┴──▶ failed ──expiry──▶ (deleted)          rollback() reactivates a retained superseded one
 *
 * Guarantees:
 *  - The active generation is a single pointer (knowledge_site_index.active_generation_id) guarded by a revision. activate(),
 *    rollback() and deactivate() each change it in ONE transaction with a compare-and-swap on that revision, so readers see the old
 *    or the new generation, never a mix, and two concurrent changes cannot both win (the loser gets GenerationConflictException).
 *  - Only a "ready" generation can be activated, and only after validate(): right site, supported provider/model/dimension, a
 *    fingerprint that matches them, every version built with exactly that model id and dimension, no unfinished versions, chunk
 *    counts that match, and an up-to-date ready version for every ready source (nothing partial is ever activated).
 *  - Building never touches the active generation or the lexical index: generation versions carry generation_id, are never the
 *    source's current_version_id, and the existing single-index path ignores them.
 *  - A superseded generation is retained (RETAIN_SECONDS) as the rollback candidate; failed builds drop their chunks immediately;
 *    cleanup() deletes expired generations with all their versions and chunks.
 *  - Erasure: Purger removes every version and chunk of a source by source_id, which covers every generation.
 *
 * Nothing here contacts a provider or decides pricing. Activation is internal (no endpoint, no automatic call).
 */
final class IndexGenerations
{
    public const BUILDING = 'building';
    public const READY = 'ready';
    public const ACTIVE = 'active';
    public const SUPERSEDED = 'superseded';
    public const FAILED = 'failed';

    /** How long a superseded generation stays available for rollback (proposal for owner decision D3). */
    public const RETAIN_SECONDS = 7 * 86400;
    /** How long a failed generation's row (without vectors) is kept for diagnostics. */
    public const FAILED_RETAIN_SECONDS = 86400;

    public function __construct(private Connection $db, private Clock $clock)
    {
    }

    /** Starts a build for a semantic configuration. @throws GenerationException */
    public function begin(EmbeddingConfig $config): int
    {
        if (!$config->isSemantic()) {
            throw new GenerationException('not_semantic');
        }
        $site = $config->siteId;
        return $this->db->transaction(function () use ($config, $site): int {
            $this->lockSite($site);
            if ($this->db->selectOne("SELECT id FROM {knowledge_generations} WHERE site_id = ? AND state IN ('building', 'ready')", [$site]) !== null) {
                throw new GenerationException('build_in_progress');
            }
            $active = $this->activeRow($site);
            if ($active !== null && $active['fingerprint'] === $config->fingerprint()) {
                throw new GenerationException('already_active');
            }
            $now = $this->clock->now();
            return $this->db->insert('knowledge_generations', [
                'site_id' => $site, 'state' => self::BUILDING, 'provider' => (string) $config->provider, 'model' => (string) $config->model,
                'dimension' => (int) $config->dimension, 'fingerprint' => (string) $config->fingerprint(), 'config_version' => $config->version,
                'sources_total' => 0, 'sources_built' => 0, 'sources_failed' => 0, 'error' => null,
                'created_at' => $now, 'activated_at' => null, 'superseded_at' => null, 'retain_until' => null, 'updated_at' => $now,
            ]);
        });
    }

    /** @return array<string, mixed>|null */
    public function find(int $siteId, int $generationId): ?array
    {
        return $this->db->selectOne('SELECT * FROM {knowledge_generations} WHERE id = ? AND site_id = ?', [$generationId, $siteId]);
    }

    /**
     * The same figures as state() without ever writing: a site that has no index row yet is represented in memory (revision 0, nothing
     * active), exactly what the row would hold if created. For status screens and background reads.
     *
     * @return array{revision:int, active:?int, previous:?int}
     */
    public function peekState(int $siteId): array
    {
        $row = $this->db->selectOne('SELECT revision, active_generation_id, previous_generation_id FROM {knowledge_site_index} WHERE site_id = ?', [$siteId]);
        return $row === null
            ? ['revision' => 0, 'active' => null, 'previous' => null]
            : ['revision' => (int) $row['revision'], 'active' => $row['active_generation_id'] === null ? null : (int) $row['active_generation_id'], 'previous' => $row['previous_generation_id'] === null ? null : (int) $row['previous_generation_id']];
    }

    /** Creates the site's index row if needed (explicit actions: it is the revision that guards activation). @return array{revision:int, active:?int, previous:?int} */
    public function state(int $siteId): array
    {
        $row = $this->indexRow($siteId);
        return ['revision' => (int) $row['revision'], 'active' => $row['active_generation_id'] === null ? null : (int) $row['active_generation_id'], 'previous' => $row['previous_generation_id'] === null ? null : (int) $row['previous_generation_id']];
    }

    /**
     * What the Retriever uses: the active generation's id if there is one and it was built for exactly this embedder (model id and
     * dimension), otherwise null (the existing single-index path). One small query, scoped to the site.
     */
    public function activeFor(int $siteId, Embedder $embedder): ?int
    {
        $row = $this->activeRow($siteId);
        if ($row === null || $row['model'] === null) {
            return null;
        }
        $modelId = $row['provider'] . ':' . $row['model'] . ':' . $row['dimension'];
        return $modelId === $embedder->model() && (int) $row['dimension'] === $embedder->dimension() ? (int) $row['id'] : null;
    }

    /**
     * Recomputes the build progress from the data: ready sources with an up-to-date ready version in this generation count as built.
     *
     * @return array{total:int, built:int, failed:int}
     */
    public function refreshProgress(int $siteId, int $generationId): array
    {
        $g = $this->find($siteId, $generationId);
        if ($g === null) {
            throw new GenerationException('not_found');
        }
        $total = (int) $this->db->selectOne("SELECT COUNT(*) AS n FROM {knowledge_sources} s JOIN {knowledge_versions} cv ON cv.id = s.current_version_id WHERE s.site_id = ? AND s.status = 'ready' AND s.visibility = 'public'", [$siteId])['n'];
        $missing = $this->countMissing($siteId, $generationId, 1_000_000);
        $failed = (int) $this->db->selectOne(
            "SELECT COUNT(DISTINCT v.source_id) AS n FROM {knowledge_versions} v WHERE v.site_id = ? AND v.generation_id = ? AND v.state = 'failed'
               AND NOT EXISTS (SELECT 1 FROM {knowledge_versions} r WHERE r.site_id = v.site_id AND r.source_id = v.source_id AND r.generation_id = v.generation_id AND r.state = 'ready')",
            [$siteId, $generationId]
        )['n'];
        $this->db->execute('UPDATE {knowledge_generations} SET sources_total = ?, sources_built = ?, sources_failed = ?, updated_at = ? WHERE id = ? AND site_id = ?', [$total, $total - $missing, $failed, $this->clock->now(), $generationId, $siteId]);
        return ['total' => $total, 'built' => $total - $missing, 'failed' => $failed];
    }

    /**
     * Ready, public sources of the site that have no ready, up-to-date version in the generation: never built, failed, or built from
     * content that has since been replaced (the content hash of the generation's version must equal that of the source's current
     * version). Private sources are never answered from (the pilot disables private-knowledge answering) and are therefore neither
     * required nor sent to an embedding provider; a source made public later simply appears here and is built.
     *
     * @return list<int>
     */
    public function missingSources(int $siteId, int $generationId, int $afterId = 0, int $limit = 0): array
    {
        $rows = $this->db->select(
            "SELECT s.id FROM {knowledge_sources} s
             JOIN {knowledge_versions} cv ON cv.id = s.current_version_id
             WHERE s.site_id = ? AND s.id > ? AND s.status = 'ready' AND s.visibility = 'public'
               AND NOT EXISTS (SELECT 1 FROM {knowledge_versions} v WHERE v.site_id = s.site_id AND v.source_id = s.id AND v.generation_id = ? AND v.state = 'ready' AND v.content_hash = cv.content_hash)
             ORDER BY s.id" . ($limit > 0 ? ' LIMIT ' . $limit : ''),
            [$siteId, $afterId, $generationId]
        );
        return array_map(static fn(array $r): int => (int) $r['id'], $rows);
    }

    /** Whether this one source still needs a ready, up-to-date version in the generation (public, ready, no matching content hash yet). One indexed lookup. */
    public function sourceMissing(int $siteId, int $generationId, int $sourceId): bool
    {
        return $this->db->selectOne(
            "SELECT s.id FROM {knowledge_sources} s JOIN {knowledge_versions} cv ON cv.id = s.current_version_id
             WHERE s.site_id = ? AND s.id = ? AND s.status = 'ready' AND s.visibility = 'public'
               AND NOT EXISTS (SELECT 1 FROM {knowledge_versions} v WHERE v.site_id = s.site_id AND v.source_id = s.id AND v.generation_id = ? AND v.state = 'ready' AND v.content_hash = cv.content_hash)",
            [$siteId, $sourceId, $generationId]
        ) !== null;
    }

    /** How many sources are missing, counted up to $cap (a bounded query: an administrator-facing figure, not a loop condition). */
    public function countMissing(int $siteId, int $generationId, int $cap = 5001): int
    {
        return (int) $this->db->selectOne(
            "SELECT COUNT(*) AS n FROM (SELECT s.id FROM {knowledge_sources} s
               JOIN {knowledge_versions} cv ON cv.id = s.current_version_id
               WHERE s.site_id = ? AND s.status = 'ready' AND s.visibility = 'public'
                 AND NOT EXISTS (SELECT 1 FROM {knowledge_versions} v WHERE v.site_id = s.site_id AND v.source_id = s.id AND v.generation_id = ? AND v.state = 'ready' AND v.content_hash = cv.content_hash)
               LIMIT " . max(1, $cap) . ') t',
            [$siteId, $generationId]
        )['n'];
    }

    /** The scheduling cursor: the builder resumes after this source id (0 = from the start). */
    public function cursor(int $siteId, int $generationId): int
    {
        return (int) ($this->db->selectOne('SELECT cursor_source_id AS c FROM {knowledge_generations} WHERE id = ? AND site_id = ?', [$generationId, $siteId])['c'] ?? 0);
    }

    public function setCursor(int $siteId, int $generationId, int $sourceId): void
    {
        $this->db->execute('UPDATE {knowledge_generations} SET cursor_source_id = ? WHERE id = ? AND site_id = ?', [$sourceId, $generationId, $siteId]);
    }

    /**
     * Privacy: removes every generation version (any generation, including the active and the retained one) whose content no longer
     * equals the source's current version, or whose source is private or erased, with its chunks. Replaced, private or erased text must not
     * linger in a generation just because it is no longer served; the lexical index drops it at publish time and the source events
     * already remove it synchronously, so this is the safety net for anything that raced them (a worker that published just after an
     * event). Versions being built (processing) are not touched.
     *
     * @return array{versions:int, chunks:int}
     */
    public function dropStaleVersions(int $siteId, ?int $sourceId = null): array
    {
        $sql = "SELECT v.id FROM {knowledge_versions} v
                JOIN {knowledge_sources} s ON s.id = v.source_id AND s.site_id = v.site_id
                LEFT JOIN {knowledge_versions} cv ON cv.id = s.current_version_id
                WHERE v.site_id = ? AND v.generation_id IS NOT NULL AND v.state = 'ready'
                  AND (cv.id IS NULL OR v.content_hash <> cv.content_hash OR s.visibility <> 'public' OR s.status IN ('deleting', 'deleted'))";
        $args = [$siteId];
        if ($sourceId !== null) {
            $sql .= ' AND v.source_id = ?';
            $args[] = $sourceId;
        }
        return $this->deleteVersions($siteId, array_map(static fn(array $r): int => (int) $r['id'], $this->db->select($sql, $args)));
    }

    /**
     * A crashed worker leaves 'processing' versions behind (they would block finish()). Removes those of this generation that no live
     * claim covers: a version is somebody's work in progress exactly while its (generation, source) task has an unexpired claim.
     */
    public function discardUnclaimed(int $siteId, int $generationId): array
    {
        $ids = $this->db->select(
            "SELECT v.id FROM {knowledge_versions} v WHERE v.site_id = ? AND v.generation_id = ? AND v.state = 'processing'
               AND NOT EXISTS (SELECT 1 FROM {knowledge_claims} c WHERE c.site_id = v.site_id AND c.generation_id = v.generation_id AND c.source_id = v.source_id AND c.state = 'active' AND c.expires_at >= ?)",
            [$siteId, $generationId, $this->clock->now()]
        );
        return $this->deleteVersions($siteId, array_map(static fn(array $r): int => (int) $r['id'], $ids));
    }

    /** The claim holder discards what an earlier (dead or fenced) owner left unfinished for this source in this generation. */
    public function discardProcessing(int $siteId, int $generationId, int $sourceId): array
    {
        $ids = $this->db->select("SELECT id FROM {knowledge_versions} WHERE site_id = ? AND generation_id = ? AND source_id = ? AND state = 'processing'", [$siteId, $generationId, $sourceId]);
        return $this->deleteVersions($siteId, array_map(static fn(array $r): int => (int) $r['id'], $ids));
    }

    /** Removes every generation version and chunk of one source (the generation part of erasure; the lexical index is Purger's). */
    public function eraseSource(int $siteId, int $sourceId): array
    {
        return GenerationCopies::removeAll($this->db, $siteId, $sourceId);
    }

    /** Records a short diagnostic code on a building or active generation (null clears it); it never changes the state. */
    public function note(int $siteId, int $generationId, ?string $code): void
    {
        $code = $code !== null && preg_match('/^[a-z_]{1,40}$/', $code) === 1 ? $code : null;
        $this->db->execute("UPDATE {knowledge_generations} SET error = ?, updated_at = ? WHERE id = ? AND site_id = ? AND state IN ('building', 'active')", [$code, $this->clock->now(), $generationId, $siteId]);
    }

    /**
     * @param list<int> $versionIds
     * @return array{versions:int, chunks:int}
     */
    private function deleteVersions(int $siteId, array $versionIds): array
    {
        $out = ['versions' => 0, 'chunks' => 0];
        foreach ($versionIds as $id) {
            // The counts are taken from the attempt that committed: a transaction re-run after a deadlock must not be counted twice.
            $done = $this->db->transaction(function () use ($siteId, $id): array {
                return [
                    $this->db->execute('DELETE FROM {knowledge_chunks} WHERE site_id = ? AND version_id = ?', [$siteId, $id]),
                    $this->db->execute('DELETE FROM {knowledge_versions} WHERE site_id = ? AND id = ? AND generation_id IS NOT NULL', [$siteId, $id]),
                ];
            });
            $out['chunks'] += $done[0];
            $out['versions'] += $done[1];
        }
        return $out;
    }

    /** building -> ready, only when nothing is missing. @throws GenerationException */
    public function finish(int $siteId, int $generationId): void
    {
        $g = $this->find($siteId, $generationId);
        if ($g === null) {
            throw new GenerationException('not_found');
        }
        if ($g['state'] !== self::BUILDING) {
            throw new GenerationException('not_building');
        }
        $this->refreshProgress($siteId, $generationId);
        $problems = $this->problems($siteId, $g);
        if ($problems !== []) {
            // phpcs:ignore WordPress.Security.EscapeOutput.ExceptionNotEscaped -- exception messages go to logs and callers, never echoed; public errors are fixed codes
            throw new GenerationException($problems[0]);
        }
        $this->db->execute("UPDATE {knowledge_generations} SET state = 'ready', updated_at = ? WHERE id = ? AND site_id = ? AND state = 'building'", [$this->clock->now(), $generationId, $siteId]);
    }

    /**
     * ready -> building: a generation that was complete but is not any more (a source was added, replaced or made public while it
     * waited for activation) takes the missing sources instead of being thrown away and re-embedded from scratch. Refused unless the
     * generation is 'ready'; an activation that wins the race makes this fail with not_ready. Never touches the active generation.
     *
     * @throws GenerationException
     */
    public function reopen(int $siteId, int $generationId): void
    {
        $this->db->transaction(function () use ($siteId, $generationId): void {
            $this->lockSite($siteId);
            $g = $this->find($siteId, $generationId);
            if ($g === null) {
                throw new GenerationException('not_found');
            }
            $changed = $this->db->execute("UPDATE {knowledge_generations} SET state = 'building', updated_at = ? WHERE id = ? AND site_id = ? AND state = 'ready'", [$this->clock->now(), $generationId, $siteId]);
            if ($changed !== 1) {
                throw new GenerationException('not_ready');
            }
        });
    }

    /**
     * Gives up a build: its vectors are deleted at once (a failed build must not keep growing the database), the row stays with a
     * reason code for diagnostics until it expires. The active generation can never be failed.
     */
    public function fail(int $siteId, int $generationId, string $code): void
    {
        $code = preg_match('/^[a-z_]{1,40}$/', $code) === 1 ? $code : 'build_failed';
        $this->db->transaction(function () use ($siteId, $generationId, $code): void {
            $g = $this->find($siteId, $generationId);
            if ($g === null) {
                throw new GenerationException('not_found');
            }
            if ($g['state'] === self::ACTIVE) {
                throw new GenerationException('cannot_fail_active');
            }
            if (!in_array($g['state'], [self::BUILDING, self::READY], true)) {
                throw new GenerationException('not_building');
            }
            $this->deleteVectors($siteId, $generationId);
            $now = $this->clock->now();
            $this->db->execute("UPDATE {knowledge_generations} SET state = 'failed', error = ?, retain_until = ?, updated_at = ? WHERE id = ? AND site_id = ?", [$code, $now + self::FAILED_RETAIN_SECONDS, $now, $generationId, $siteId]);
        });
    }

    /**
     * Everything that would stop this generation from being served: empty list = valid. Does not check the state; callers do.
     *
     * @param array<string, mixed> $g
     * @return list<string> stable codes
     */
    private function problems(int $siteId, array $g): array
    {
        $out = [];
        $provider = (string) $g['provider'];
        $model = (string) $g['model'];
        $dimension = (int) $g['dimension'];
        if ((EmbeddingConfig::supported()[$provider][$model] ?? null) !== $dimension) {
            $out[] = 'unsupported_combination';
        }
        if (!hash_equals((string) $g['fingerprint'], EmbeddingConfig::fingerprintOf($provider, $model, $dimension))) {
            $out[] = 'fingerprint_mismatch';
        }
        $id = (int) $g['id'];
        $modelId = $provider . ':' . $model . ':' . $dimension;
        if ($this->db->selectOne('SELECT id FROM {knowledge_versions} WHERE site_id = ? AND generation_id = ? AND (embedding_model <> ? OR dimension <> ?) LIMIT 1', [$siteId, $id, $modelId, $dimension]) !== null) {
            $out[] = 'incompatible_vectors';
        }
        if ($this->db->selectOne("SELECT id FROM {knowledge_versions} WHERE site_id = ? AND generation_id = ? AND state = 'processing' LIMIT 1", [$siteId, $id]) !== null) {
            $out[] = 'unfinished_versions';
        }
        if ($this->db->selectOne(
            "SELECT v.id FROM {knowledge_versions} v WHERE v.site_id = ? AND v.generation_id = ? AND v.state = 'ready'
               AND v.chunk_count <> (SELECT COUNT(*) FROM {knowledge_chunks} c WHERE c.version_id = v.id) LIMIT 1",
            [$siteId, $id]
        ) !== null) {
            $out[] = 'chunk_mismatch';
        }
        if ($this->missingSources($siteId, $id, 0, 1) !== []) {
            $out[] = 'incomplete';
        }
        return $out;
    }

    /**
     * Makes a ready generation the active one, atomically. $expectedRevision is the revision the caller read from state(); a
     * concurrent change makes it stale and the call fails with GenerationConflictException without changing anything. The
     * previous active generation becomes the retained rollback candidate.
     *
     * @throws GenerationException
     */
    public function activate(int $siteId, int $generationId, int $expectedRevision): void
    {
        $this->db->transaction(function () use ($siteId, $generationId, $expectedRevision): void {
            $idx = $this->indexRow($siteId);
            $this->assertRevision($idx, $expectedRevision);
            $g = $this->find($siteId, $generationId);
            if ($g === null) {
                throw new GenerationException('not_found');
            }
            if ($g['state'] !== self::READY) {
                throw new GenerationException('not_ready');
            }
            $problems = $this->problems($siteId, $g);
            if ($problems !== []) {
                // phpcs:ignore WordPress.Security.EscapeOutput.ExceptionNotEscaped -- exception messages go to logs and callers, never echoed; public errors are fixed codes
                throw new GenerationException($problems[0]);
            }
            $this->swap($siteId, $idx, $generationId);
        });
    }

    /**
     * Goes back to the retained previous generation if it is still valid and complete for the sources that exist now. Refused when
     * there is none, when it expired, or when sources were added or changed since (rolling back would silently drop them).
     *
     * @throws GenerationException
     */
    public function rollback(int $siteId, int $expectedRevision): void
    {
        $this->db->transaction(function () use ($siteId, $expectedRevision): void {
            $idx = $this->indexRow($siteId);
            $this->assertRevision($idx, $expectedRevision);
            if ($idx['previous_generation_id'] === null) {
                throw new GenerationException('no_rollback_target');
            }
            $g = $this->find($siteId, (int) $idx['previous_generation_id']);
            if ($g === null || $g['state'] !== self::SUPERSEDED) {
                throw new GenerationException('not_retained');
            }
            if ($g['retain_until'] !== null && (int) $g['retain_until'] <= $this->clock->now()) {
                throw new GenerationException('rollback_expired');
            }
            $problems = $this->problems($siteId, $g);
            if ($problems !== []) {
                // phpcs:ignore WordPress.Security.EscapeOutput.ExceptionNotEscaped -- exception messages go to logs and callers, never echoed; public errors are fixed codes
                throw new GenerationException($problems[0]);
            }
            $this->swap($siteId, $idx, (int) $g['id']);
        });
    }

    /** Back to the existing single-index (lexical) path; the generation stays as the rollback candidate. @throws GenerationException */
    public function deactivate(int $siteId, int $expectedRevision): void
    {
        $this->db->transaction(function () use ($siteId, $expectedRevision): void {
            $idx = $this->indexRow($siteId);
            $this->assertRevision($idx, $expectedRevision);
            if ($idx['active_generation_id'] === null) {
                throw new GenerationException('nothing_active');
            }
            $this->swap($siteId, $idx, null);
        });
    }

    /**
     * Deletes expired superseded generations and failed rows with all their versions and chunks. Never touches the active or a
     * building/ready generation, and clears the rollback pointer if it pointed at a deleted generation.
     *
     * @return array{generations:int, versions:int, chunks:int}
     */
    public function cleanup(int $siteId): array
    {
        return $this->db->transaction(function () use ($siteId): array {
            $now = $this->clock->now();
            $totals = ['generations' => 0, 'versions' => 0, 'chunks' => 0];
            $expired = $this->db->select(
                "SELECT id FROM {knowledge_generations} WHERE site_id = ? AND state IN ('superseded', 'failed') AND retain_until IS NOT NULL AND retain_until <= ?",
                [$siteId, $now]
            );
            foreach ($expired as $row) {
                $id = (int) $row['id'];
                $counts = $this->deleteVectors($siteId, $id);
                $this->db->execute('DELETE FROM {knowledge_generations} WHERE id = ? AND site_id = ?', [$id, $siteId]);
                $this->db->execute('UPDATE {knowledge_site_index} SET previous_generation_id = NULL, updated_at = ? WHERE site_id = ? AND previous_generation_id = ?', [$now, $siteId, $id]);
                $totals['generations']++;
                $totals['versions'] += $counts['versions'];
                $totals['chunks'] += $counts['chunks'];
            }
            return $totals;
        });
    }

    /** The one place the pointer moves: CAS on the revision, then the generation rows. */
    private function swap(int $siteId, array $idx, ?int $newActive): void
    {
        $now = $this->clock->now();
        $old = $idx['active_generation_id'] === null ? null : (int) $idx['active_generation_id'];
        // The rollback candidate is whatever was active before; with nothing active the earlier candidate stays, unless it is the one being activated.
        $previous = $idx['previous_generation_id'] === null ? null : (int) $idx['previous_generation_id'];
        $newPrevious = $old ?? ($newActive !== null && $newActive !== $previous ? $previous : null);
        $changed = $this->db->execute(
            'UPDATE {knowledge_site_index} SET active_generation_id = ?, previous_generation_id = ?, revision = revision + 1, updated_at = ? WHERE site_id = ? AND revision = ?',
            [$newActive, $newPrevious, $now, $siteId, (int) $idx['revision']]
        );
        if ($changed !== 1) {
            throw new GenerationConflictException();
        }
        if ($newActive !== null) {
            $up = $this->db->execute("UPDATE {knowledge_generations} SET state = 'active', activated_at = ?, retain_until = NULL, superseded_at = NULL, error = NULL, updated_at = ? WHERE id = ? AND site_id = ? AND state IN ('ready', 'superseded')", [$now, $now, $newActive, $siteId]);
            if ($up !== 1) {
                throw new GenerationException('not_ready'); // rolls the whole transaction back
            }
        }
        if ($old !== null) {
            $this->db->execute("UPDATE {knowledge_generations} SET state = 'superseded', superseded_at = ?, retain_until = ?, updated_at = ? WHERE id = ? AND site_id = ? AND state = 'active'", [$now, $now + self::RETAIN_SECONDS, $now, $old, $siteId]);
        }
    }

    private function assertRevision(array $idx, int $expected): void
    {
        if ((int) $idx['revision'] !== $expected) {
            throw new GenerationConflictException();
        }
    }

    /** @return array{chunks:int, versions:int} */
    private function deleteVectors(int $siteId, int $generationId): array
    {
        $chunks = $this->db->execute('DELETE FROM {knowledge_chunks} WHERE site_id = ? AND version_id IN (SELECT id FROM {knowledge_versions} WHERE site_id = ? AND generation_id = ?)', [$siteId, $siteId, $generationId]);
        $versions = $this->db->execute('DELETE FROM {knowledge_versions} WHERE site_id = ? AND generation_id = ?', [$siteId, $generationId]);
        $this->db->execute('DELETE FROM {knowledge_claims} WHERE site_id = ? AND generation_id = ?', [$siteId, $generationId]);
        return ['chunks' => $chunks, 'versions' => $versions];
    }

    /** @return array<string, mixed>|null the active generation's row, or null */
    private function activeRow(int $siteId): ?array
    {
        return $this->db->selectOne(
            "SELECT g.* FROM {knowledge_site_index} i JOIN {knowledge_generations} g ON g.id = i.active_generation_id AND g.site_id = i.site_id AND g.state = 'active' WHERE i.site_id = ?",
            [$siteId]
        );
    }

    /** @return array<string, mixed> */
    private function indexRow(int $siteId): array
    {
        $row = $this->db->selectOne('SELECT * FROM {knowledge_site_index} WHERE site_id = ?', [$siteId]);
        if ($row !== null) {
            return $row;
        }
        // Creating the row is for explicit actions only (state(), locking); a concurrent creator simply wins (peekState() never gets here).
        $this->db->insertIfAbsent('knowledge_site_index', ['site_id' => $siteId, 'active_generation_id' => null, 'previous_generation_id' => null, 'revision' => 0, 'updated_at' => $this->clock->now()], ['site_id']);
        return $this->db->selectOne('SELECT * FROM {knowledge_site_index} WHERE site_id = ?', [$siteId]) ?? throw new GenerationException('not_found');
    }

    /** Serializes builds per site: takes the row lock of the site's index row until the surrounding transaction ends. */
    private function lockSite(int $siteId): void
    {
        $this->indexRow($siteId);
        $this->db->execute('UPDATE {knowledge_site_index} SET updated_at = updated_at WHERE site_id = ?', [$siteId]);
    }
}