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

namespace Aiwend\Knowledge;

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

/**
 * Knowledge sources (B05). Visibility and status are read from this table at
 * retrieval time, so privatizing or deleting a source is enforced on the very
 * next query (no cache to invalidate). Physical purge of derivatives is a
 * separate job.
 *
 * Status: queued → processing → ready | failed | ocr_required; deleting → deleted.
 */
final class SourceRepository
{
    public const VISIBILITY = ['public', 'private'];
    public const ORIGINS = ['post', 'upload'];
    /** How long a waiting or running re-index is still reported as "updating". */
    private const UPDATE_WINDOW = 900;

    private ?\Closure $listener = null;

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

    /** Operator-only caller; returns current published chunks, never partially built replacements. */
    public function inspect(int $siteId, int $sourceId, int $page = 1): ?array
    {
        $source = $this->find($siteId, $sourceId);
        if ($source === null || in_array($source['status'], ['deleting', 'deleted'], true)) { return null; }
        $version = $source['current_version_id'] === null ? null : $this->db->selectOne("SELECT id, version_no, embedding_model, dimension, chunk_count, updated_at FROM {knowledge_versions} WHERE id = ? AND site_id = ? AND source_id = ? AND state = 'ready'", [$source['current_version_id'], $siteId, $sourceId]);
        $chunks = $version === null ? [] : $this->db->select('SELECT seq, anchor, heading, text FROM {knowledge_chunks} WHERE site_id = ? AND source_id = ? AND version_id = ? ORDER BY seq LIMIT 51 OFFSET ?', [$siteId, $sourceId, $version['id'], (max(1, min(10000, $page)) - 1) * 50]);
        return ['source' => $source, 'version' => $version, 'chunks' => array_slice($chunks, 0, 50), 'has_more' => count($chunks) > 50];
    }

    /** Creates the source, or returns the existing one for the same origin (re-ingest = new version). */
    public function upsert(int $siteId, string $origin, string $originRef, string $title, ?string $url, string $visibility, ?int $ownerUserId): int
    {
        if (!in_array($origin, self::ORIGINS, true) || !in_array($visibility, self::VISIBILITY, true)) {
            throw new \InvalidArgumentException('Invalid origin or visibility.');
        }
        $existing = $this->db->selectOne('SELECT id FROM {knowledge_sources} WHERE site_id = ? AND origin = ? AND origin_ref = ?', [$siteId, $origin, $originRef]);
        $now = $this->clock->now();
        if ($existing !== null) {
            $this->db->execute(
                'UPDATE {knowledge_sources} SET title = ?, url = ?, updated_at = ? WHERE id = ? AND site_id = ?',
                [mb_substr($title, 0, 191), $url, $now, (int) $existing['id'], $siteId]
            );
            return (int) $existing['id'];
        }
        return $this->db->insert('knowledge_sources', [
            'site_id' => $siteId,
            'owner_user_id' => $ownerUserId,
            'origin' => $origin,
            'origin_ref' => mb_substr($originRef, 0, 191),
            'title' => mb_substr($title, 0, 191),
            'url' => $url === null ? null : mb_substr($url, 0, 500),
            'visibility' => $visibility,
            'status' => 'queued',
            'current_version_id' => null,
            'language' => 'en',
            'error' => null,
            'created_at' => $now,
            'updated_at' => $now,
        ]);
    }

    /** True when a source for this origin already exists (upsert would return it instead of creating one). */
    public function originExists(int $siteId, string $origin, string $originRef): bool
    {
        return $this->db->selectOne('SELECT id FROM {knowledge_sources} WHERE site_id = ? AND origin = ? AND origin_ref = ?', [$siteId, $origin, $originRef]) !== null;
    }

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

    /** @return list<array<string, mixed>> */
    public function list(int $siteId): array
    {
        $updating = $this->updatingIds($siteId);
        return array_map(
            fn(array $r): array => $this->shape($r) + ['updating' => isset($updating[(int) $r['id']]) && $r['status'] === 'ready' && $r['current_version_id'] !== null],
            $this->db->select("SELECT * FROM {knowledge_sources} WHERE site_id = ? AND status <> 'deleted' ORDER BY id", [$siteId])
        );
    }

    /** A new version is being built (or an ingest job is waiting) for these sources: two queries per listing, never one per source. */
    private function updatingIds(int $siteId): array
    {
        $ids = [];
        $since = $this->clock->now() - self::UPDATE_WINDOW;
        foreach ($this->db->select("SELECT DISTINCT source_id FROM {knowledge_versions} WHERE site_id = ? AND state = 'processing' AND generation_id IS NULL AND created_at > ?", [$siteId, $since]) as $r) {
            $ids[(int) $r['source_id']] = true;
        }
        foreach ($this->db->select("SELECT payload FROM {jobs} WHERE site_id = ? AND type = 'knowledge.ingest' AND state IN ('queued', 'running') AND created_at > ?", [$siteId, $since]) as $r) {
            $payload = json_decode((string) $r['payload'], true);
            if (is_array($payload) && isset($payload['source_id'])) {
                $ids[(int) $payload['source_id']] = true;
            }
        }
        return $ids;
    }

    /**
     * Replaces the uploaded file of an existing upload source: same source (id, assistants, visibility, owner), new stored
     * text; indexing then builds a new version while the current one keeps answering. Returns the content hash, or null when
     * the source is not an upload or is being deleted (nothing is left behind in that case).
     */
    public function replaceUpload(int $siteId, int $id, string $name, string $type, string $content, ?string $title): ?string
    {
        $s = $this->find($siteId, $id);
        if ($s === null || $s['origin'] !== 'upload' || in_array($s['status'], ['deleting', 'deleted'], true)) {
            return null;
        }
        $hash = $this->putBlob($siteId, $id, $type, $content);
        $pos = strrpos((string) $s['origin_ref'], ':');
        $suffix = $pos !== false ? substr((string) $s['origin_ref'], $pos + 1) : bin2hex(random_bytes(6));
        $this->db->execute(
            'UPDATE {knowledge_sources} SET origin_ref = ?, title = ?, updated_at = ? WHERE id = ? AND site_id = ?',
            [mb_substr($name, 0, 190 - strlen($suffix)) . ':' . $suffix, $title !== null && $title !== '' ? mb_substr($title, 0, 191) : $s['title'], $this->clock->now(), $id, $siteId]
        );
        $after = $this->find($siteId, $id);
        if ($after === null || in_array($after['status'], ['deleting', 'deleted'], true)) {
            // Deleted while the file was being stored: do not leave the new text behind.
            $this->db->execute('DELETE FROM {knowledge_blobs} WHERE source_id = ? AND site_id = ?', [$id, $siteId]);
            return null;
        }
        return $hash;
    }

    public function resource(int $siteId, int $id): ?Resource
    {
        $s = $this->find($siteId, $id);
        return $s === null ? null : new Resource('knowledge_source', $s['id'], $siteId, $s['owner_user_id'], null, ['visibility' => $s['visibility'], 'status' => $s['status']]);
    }

    /**
     * Updates processing status. Never moves a source out of deleting/deleted,
     * so a deletion that races with indexing always wins. Returns false then.
     */
    /**
     * Sources that wait for indexing ("queued" or "processing") since before $olderThan although no indexing job exists that will still run
     * (queued, running or suspended with its module; finished, dead and cancelled jobs do not count). Such a source is the trace of a request
     * that died between registering the source and queuing its job. Bounded by $limit; oldest numbers first.
     *
     * @return list<int>
     */
    public function withoutIndexingJob(int $siteId, int $olderThan, int $limit): array
    {
        $out = [];
        $rows = $this->db->select("SELECT id FROM {knowledge_sources} WHERE site_id = ? AND status IN ('queued', 'processing') AND updated_at < ? ORDER BY id LIMIT ?", [$siteId, $olderThan, max(200, $limit * 4)]);
        foreach ($rows as $r) {
            $id = (int) $r['id'];
            // phpcs:ignore WordPress.WP.AlternativeFunctions.json_encode_json_encode -- must equal the payload the queue stored with json_encode() byte for byte; this class runs without WordPress in its unit tests
            $live = $this->db->selectOne("SELECT id FROM {jobs} WHERE site_id = ? AND type = 'knowledge.ingest' AND state IN ('queued', 'running', 'suspended') AND payload = ? LIMIT 1", [$siteId, json_encode(['source_id' => $id])]);
            if ($live === null) {
                $out[] = $id;
                if (count($out) >= $limit) {
                    break;
                }
            }
        }
        return $out;
    }

    public function setStatus(int $siteId, int $id, string $status, ?string $error = null): bool
    {
        if (!in_array($status, ['queued', 'processing', 'ready', 'failed', 'ocr_required'], true)) {
            throw new \InvalidArgumentException('Use markDeleting() or the purger for deletion states.');
        }
        return $this->db->execute(
            "UPDATE {knowledge_sources} SET status = ?, error = ?, updated_at = ? WHERE id = ? AND site_id = ? AND status NOT IN ('deleting', 'deleted')",
            [$status, $error === null ? null : mb_substr($error, 0, 500), $this->clock->now(), $id, $siteId]
        ) === 1;
    }

    /** Takes effect for retrieval immediately. */
    /**
     * The queue stopped retrying: replace the interim "will be retried" text with the final state. A source that
     * already has a published version keeps serving it (still "ready"), with the failure recorded.
     */
    public function setIndexingGaveUp(int $siteId, int $id): bool
    {
        $s = $this->find($siteId, $id);
        if ($s === null) {
            return false;
        }
        return $s['current_version_id'] !== null
            ? $this->setStatus($siteId, $id, 'ready', 'Latest update failed after several attempts.')
            : $this->setStatus($siteId, $id, 'failed', 'Indexing failed after several attempts. Try Re-index.');
    }

    /**
     * Takes effect for retrieval on the very next query (every generation read joins this row). Owner policy: making a source private
     * deletes its vectors from every generation in the same transaction (GenerationCopies::removeAll); the source content and the
     * lexical index stay. Nothing is reused later: when the source is public again it is embedded afresh, which the listener (called
     * after commit) lets the generations pick up. A build that was running for it is fenced twice (its claim is gone, and the publish
     * re-checks visibility inside its transaction).
     */
    public function setVisibility(int $siteId, int $id, string $visibility): bool
    {
        if (!in_array($visibility, self::VISIBILITY, true)) {
            throw new \InvalidArgumentException('Invalid visibility.');
        }
        $before = $this->db->selectOne('SELECT visibility FROM {knowledge_sources} WHERE id = ? AND site_id = ?', [$id, $siteId]);
        $done = $this->db->transaction(function () use ($siteId, $id, $visibility): bool {
            // updated_at only ever moves forward, even inside one clock second: publication tokens are bound to it (see PublishToken).
            $now = $this->clock->now();
            $changed = $this->db->execute('UPDATE {knowledge_sources} SET visibility = ?, updated_at = CASE WHEN ? > updated_at THEN ? ELSE updated_at + 1 END WHERE id = ? AND site_id = ?', [$visibility, $now, $now, $id, $siteId]) === 1;
            if ($changed && $visibility === 'private') {
                // Every generation (active and retained) loses its vectors of this source in the same transaction, and the claims that
                // could let a running worker republish it go too. The lexical index and the source content are not touched. Idempotent.
                GenerationCopies::removeAll($this->db, $siteId, $id);
            }
            return $changed;
        });
        if ($done && $visibility === 'public' && $before !== null && $before['visibility'] !== 'public') {
            $this->notify($siteId, $id);
        }
        return $done;
    }

    /**
     * Assistants that have this source attached. `active` are the ones that would start quoting it to anonymous visitors the moment it is
     * public (status active and public); `other` are attached drafts or private assistants (informational: they quote it once published).
     *
     * @param bool $lock true inside the publishing transaction: a locking read (MySQL/MariaDB), so a concurrent change of an assistant waits for it
     * @return array{active: list<array{id:int, name:string}>, other: list<array{id:int, name:string, status:string, is_public:bool}>}
     */
    public function affectedAssistants(int $siteId, int $id, bool $lock = false): array
    {
        $rows = $this->db->select(
            'SELECT b.id, b.name, b.status, b.is_public FROM {bots} b JOIN {bot_sources} s ON s.bot_id = b.id AND s.site_id = b.site_id WHERE b.site_id = ? AND s.source_id = ? ORDER BY b.id'
            . ($lock && $this->db->driver() === 'mysql' ? ' FOR UPDATE' : ''),
            [$siteId, $id]
        );
        $active = [];
        $other = [];
        foreach ($rows as $r) {
            if ($r['status'] === 'active' && (int) $r['is_public'] === 1) {
                $active[] = ['id' => (int) $r['id'], 'name' => (string) $r['name']];
            } else {
                $other[] = ['id' => (int) $r['id'], 'name' => (string) $r['name'], 'status' => (string) $r['status'], 'is_public' => (int) $r['is_public'] === 1];
            }
        }
        return ['active' => $active, 'other' => $other];
    }

    /**
     * The ONLY way a private source becomes public (Phase 96). Inside one transaction: the source row is locked, its state and the set of
     * affected assistants are read (locking read), `$approve` must accept them (the confirmation token check), and the change is a
     * compare-and-swap on visibility and updated_at. Two administrators confirming at once cannot both win; a source or an assistant that
     * changed since the confirmation screen makes it refuse.
     *
     * @param callable(array<string, mixed> $row, list<int> $activeAssistantIds): bool $approve
     * @return 'published'|'stale'|'already_public'|'not_found'
     */
    public function publish(int $siteId, int $id, callable $approve): string
    {
        $lock = $this->db->driver() === 'mysql' ? ' FOR UPDATE' : '';
        $result = $this->db->transaction(function () use ($siteId, $id, $approve, $lock): string {
            $row = $this->db->selectOne('SELECT id, visibility, status, updated_at FROM {knowledge_sources} WHERE id = ? AND site_id = ?' . $lock, [$id, $siteId]);
            if ($row === null || in_array($row['status'], ['deleting', 'deleted'], true)) {
                return 'not_found';
            }
            if ($row['visibility'] === 'public') {
                return 'already_public';
            }
            $ids = array_column($this->affectedAssistants($siteId, $id, true)['active'], 'id');
            if (!$approve($row, $ids)) {
                return 'stale';
            }
            $now = $this->clock->now();
            $changed = $this->db->execute(
                "UPDATE {knowledge_sources} SET visibility = 'public', updated_at = CASE WHEN ? > updated_at THEN ? ELSE updated_at + 1 END WHERE id = ? AND site_id = ? AND visibility = 'private' AND updated_at = ?",
                [$now, $now, $id, $siteId, (int) $row['updated_at']]
            );
            return $changed === 1 ? 'published' : 'stale';
        });
        if ($result === 'published') {
            $this->notify($siteId, $id); // same listener the plain visibility change used (generations pick the source up afresh)
        }
        return $result;
    }

    /** Fixed-format audit detail for a visibility change: ids and numbers only, never a title, text or request body. @param list<int> $assistantIds */
    public static function visibilityDetail(string $from, string $to, array $assistantIds): string
    {
        sort($assistantIds);
        return mb_substr($from . '->' . $to . '; active assistants: ' . count($assistantIds) . ($assistantIds === [] ? '' : ' (' . implode(',', array_slice($assistantIds, 0, 20)) . ')'), 0, 191);
    }

    /** Immediately excludes the source from retrieval and from every generation; the purge job removes the rest. */
    public function markDeleting(int $siteId, int $id): bool
    {
        return $this->db->transaction(function () use ($siteId, $id): bool {
            $marked = $this->db->execute(
                "UPDATE {knowledge_sources} SET status = 'deleting', updated_at = ? WHERE id = ? AND site_id = ? AND status <> 'deleted'",
                [$this->clock->now(), $id, $siteId]
            ) === 1;
            if ($marked) {
                GenerationCopies::removeAll($this->db, $siteId, $id);
            }
            return $marked;
        });
    }

    /** Registers who is told when a source may have become eligible for the generations (made public). */
    public function onEligible(?\Closure $listener): void
    {
        $this->listener = $listener;
    }

    private function notify(int $siteId, int $id): void
    {
        if ($this->listener !== null) {
            ($this->listener)($siteId, $id);
        }
    }

    public function attachToBot(int $siteId, int $botId, int $sourceId): void
    {
        if ($this->db->selectOne('SELECT id FROM {bot_sources} WHERE bot_id = ? AND source_id = ?', [$botId, $sourceId]) === null) {
            $this->db->insert('bot_sources', ['site_id' => $siteId, 'bot_id' => $botId, 'source_id' => $sourceId, 'created_at' => $this->clock->now()]);
        }
    }

    public function detachFromBot(int $siteId, int $botId, int $sourceId): void
    {
        $this->db->execute('DELETE FROM {bot_sources} WHERE site_id = ? AND bot_id = ? AND source_id = ?', [$siteId, $botId, $sourceId]);
    }

    /** How many sources of this site could not be processed (failed, or a scanned document without text): one indexed COUNT, no rows loaded. */
    public function countFailed(int $siteId): int
    {
        return (int) ($this->db->selectOne("SELECT COUNT(*) AS n FROM {knowledge_sources} WHERE site_id = ? AND status IN ('failed', 'ocr_required')", [$siteId])['n'] ?? 0);
    }

    public function botHasSources(int $siteId, int $botId): bool
    {
        return $this->db->selectOne('SELECT id FROM {bot_sources} WHERE site_id = ? AND bot_id = ? LIMIT 1', [$siteId, $botId]) !== null;
    }

    /** Searchable passages (chunks of the current version of every live source) of a site: what each question has to scan. */
    public function searchablePassages(int $siteId): int
    {
        return (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.status NOT IN ('deleting', 'deleted')",
            [$siteId]
        )['n'] ?? 0);
    }
    /** Stores (replaces) the original uploaded text for a source. */
    public function putBlob(int $siteId, int $sourceId, string $type, string $content): string
    {
        $hash = hash('sha256', $type . "\0" . $content);
        if ($type === 'pdf') {
            $content = base64_encode($content);
        }
        $this->db->transaction(function () use ($siteId, $sourceId, $type, $content, $hash): void {
            $this->db->execute('DELETE FROM {knowledge_blobs} WHERE source_id = ? AND site_id = ?', [$sourceId, $siteId]);
            $this->db->insert('knowledge_blobs', ['site_id' => $siteId, 'source_id' => $sourceId, 'content_type' => $type, 'content' => $content, 'content_hash' => $hash, 'created_at' => $this->clock->now()]);
        });
        return $hash;
    }

    /** @return array{type:string, content:string, hash:string}|null */
    public function blob(int $siteId, int $sourceId): ?array
    {
        $r = $this->db->selectOne('SELECT content_type, content, content_hash FROM {knowledge_blobs} WHERE source_id = ? AND site_id = ?', [$sourceId, $siteId]);
        return $r === null ? null : ['type' => (string) $r['content_type'], 'content' => $r['content_type'] === 'pdf' ? (string) base64_decode((string) $r['content'], true) : (string) $r['content'], 'hash' => (string) $r['content_hash']];
    }

    /** @return array<string, mixed>|null */
    public function findByOrigin(int $siteId, string $origin, string $originRef): ?array
    {
        $row = $this->db->selectOne('SELECT * FROM {knowledge_sources} WHERE site_id = ? AND origin = ? AND origin_ref = ?', [$siteId, $origin, $originRef]);
        return $row === null ? null : $this->shape($row);
    }

    /** @param array<string, mixed> $r @return array<string, mixed> */
    private function shape(array $r): array
    {
        return [
            'id' => (int) $r['id'],
            'site_id' => (int) $r['site_id'],
            'owner_user_id' => $r['owner_user_id'] === null ? null : (int) $r['owner_user_id'],
            'origin' => (string) $r['origin'],
            'origin_ref' => (string) $r['origin_ref'],
            'title' => (string) $r['title'],
            'url' => $r['url'] === null ? null : (string) $r['url'],
            'visibility' => (string) $r['visibility'],
            'status' => (string) $r['status'],
            'current_version_id' => $r['current_version_id'] === null ? null : (int) $r['current_version_id'],
            'error' => $r['error'] === null ? null : (string) $r['error'],
            'updated_at' => (int) $r['updated_at'],
        ];
    }
}