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

namespace Aiwend\Knowledge;

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

/**
 * Database-backed ownership of one (site, generation, source) build task, so two workers cannot legitimately build the same source
 * at once, a crashed worker cannot block anything for longer than the lease, and a worker that lost its claim cannot publish.
 *
 *  - claim(): one INSERT wins the task; an existing row is taken over only if its lease expired, in a single UPDATE conditioned on the
 *    previous owner, so exactly one of any number of racing workers succeeds. Each statement is short; no lock is held while a
 *    worker embeds (the lease is a column, not a database lock).
 *  - renew(): extends the lease and proves ownership (owner token + not expired). It is called between embedding batches and, as the
 *    fence, INSIDE the publishing transaction, where its row lock serializes it against a takeover: either the publish commits under a
 *    valid lease, or the takeover wins and the stale worker's publish aborts and discards its version.
 *  - A claim taken over from an expired one carries what the dead worker had announced ($previous: operation prefix and batch count),
 *    so the successor can check the ledger for dispatched work before it spends anything again.
 *  - 'blocked' is the explicit state of a task whose earlier dispatch is uncertain: it never expires and only unblock() (an explicit
 *    administrator action) removes it. Nothing here makes provider billing exactly-once: it prevents duplicate dispatches by workers
 *    that coordinate through the database; a request whose network outcome is unknown stays unknown (see the ledger's uncertain holds).
 */
final class GenerationClaims
{
    public const TTL = 300;
    public const ACTIVE = 'active';
    public const BLOCKED = 'blocked';

    public function __construct(private Connection $db, private Clock $clock, private int $ttl = self::TTL)
    {
    }

    /** @return GenerationClaim|'busy'|'blocked' */
    public function claim(int $siteId, int $generationId, int $sourceId, string $contentHash): GenerationClaim|string
    {
        for ($i = 0; $i < 3; $i++) {
            $now = $this->clock->now();
            $token = bin2hex(random_bytes(12));
            // Losing this race is normal, not an error: the insert absorbs only the task's unique key (nothing is raised or logged for it),
            // while any other failure still throws.
            $inserted = $this->db->insertIfAbsent('knowledge_claims', [
                'site_id' => $siteId, 'generation_id' => $generationId, 'source_id' => $sourceId, 'owner' => $token, 'state' => self::ACTIVE,
                'content_hash' => $contentHash, 'operation' => null, 'batches' => 0, 'beats' => 0, 'expires_at' => $now + $this->ttl, 'created_at' => $now, 'updated_at' => $now,
            ], ['site_id', 'generation_id', 'source_id']);
            // The affected-row count behind $inserted depends on client flags (MySQL's CLIENT_FOUND_ROWS reports a taken key as 1), and one
            // wrong "won" would mean two owners: the owner token written by THIS insert is the proof, one primary-key-free indexed read.
            if ($inserted && ($this->db->selectOne('SELECT owner FROM {knowledge_claims} WHERE site_id = ? AND generation_id = ? AND source_id = ?', [$siteId, $generationId, $sourceId])['owner'] ?? null) === $token) {
                return new GenerationClaim($siteId, $generationId, $sourceId, $token, null);
            }
            $row = $this->db->selectOne('SELECT owner, state, operation, batches, expires_at FROM {knowledge_claims} WHERE site_id = ? AND generation_id = ? AND source_id = ?', [$siteId, $generationId, $sourceId]);
            if ($row === null) {
                continue; // released between our insert and our read: try again
            }
            if ($row['state'] === self::BLOCKED) {
                return self::BLOCKED;
            }
            if ((int) $row['expires_at'] >= $now) {
                return 'busy';
            }
            $won = $this->db->execute(
                "UPDATE {knowledge_claims} SET owner = ?, content_hash = ?, operation = NULL, batches = 0, beats = beats + 1, expires_at = ?, updated_at = ?
                 WHERE site_id = ? AND generation_id = ? AND source_id = ? AND owner = ? AND state = 'active' AND expires_at < ?",
                [$token, $contentHash, $now + $this->ttl, $now, $siteId, $generationId, $sourceId, $row['owner'], $now]
            );
            if ($won === 1) {
                $previous = $row['operation'] === null ? null : ['operation' => (string) $row['operation'], 'batches' => (int) $row['batches']];
                return new GenerationClaim($siteId, $generationId, $sourceId, $token, $previous);
            }
            return 'busy'; // another worker took it over first
        }
        return 'busy';
    }

    /** Extends the lease. False when the claim was lost (expired and taken over, blocked, or released): the caller must stop. */
    public function renew(GenerationClaim $c): bool
    {
        $now = $this->clock->now();
        return $this->db->execute(
            "UPDATE {knowledge_claims} SET expires_at = ?, beats = beats + 1, updated_at = ?
             WHERE site_id = ? AND generation_id = ? AND source_id = ? AND owner = ? AND state = 'active' AND expires_at >= ?",
            [$now + $this->ttl, $now, $c->siteId, $c->generationId, $c->sourceId, $c->owner, $now]
        ) === 1;
    }

    /** Announces the embedding operation before anything is dispatched, so a successor can find out what may have been sent. */
    public function announce(GenerationClaim $c, string $operation, int $batches): bool
    {
        return $this->db->execute(
            "UPDATE {knowledge_claims} SET operation = ?, batches = ?, updated_at = ? WHERE site_id = ? AND generation_id = ? AND source_id = ? AND owner = ? AND state = 'active'",
            [$operation, $batches, $this->clock->now(), $c->siteId, $c->generationId, $c->sourceId, $c->owner]
        ) === 1;
    }

    public function release(GenerationClaim $c): void
    {
        $this->db->execute("DELETE FROM {knowledge_claims} WHERE site_id = ? AND generation_id = ? AND source_id = ? AND owner = ? AND state = 'active'", [$c->siteId, $c->generationId, $c->sourceId, $c->owner]);
    }

    /** Parks a task whose earlier dispatch is uncertain, until unblock(). */
    public function block(GenerationClaim $c): void
    {
        $this->db->execute("UPDATE {knowledge_claims} SET state = 'blocked', updated_at = ? WHERE site_id = ? AND generation_id = ? AND source_id = ? AND owner = ?", [$this->clock->now(), $c->siteId, $c->generationId, $c->sourceId, $c->owner]);
    }

    /** The explicit administrator decision to build a blocked source again (with a new operation id). */
    public function unblock(int $siteId, int $generationId, int $sourceId): bool
    {
        return $this->db->execute("DELETE FROM {knowledge_claims} WHERE site_id = ? AND generation_id = ? AND source_id = ? AND state = 'blocked'", [$siteId, $generationId, $sourceId]) === 1;
    }

    /** What was in doubt on a blocked task, as announced before it was parked, or null. @return array{operation:string, batches:int}|null */
    public function blockedOperation(int $siteId, int $generationId, int $sourceId): ?array
    {
        $r = $this->db->selectOne("SELECT operation, batches FROM {knowledge_claims} WHERE site_id = ? AND generation_id = ? AND source_id = ? AND state = 'blocked'", [$siteId, $generationId, $sourceId]);
        return $r === null || $r['operation'] === null ? null : ['operation' => (string) $r['operation'], 'batches' => (int) $r['batches']];
    }

    /** @return list<int> */
    public function blocked(int $siteId, int $generationId): array
    {
        return array_map(static fn(array $r): int => (int) $r['source_id'], $this->db->select("SELECT source_id FROM {knowledge_claims} WHERE site_id = ? AND generation_id = ? AND state = 'blocked' ORDER BY source_id", [$siteId, $generationId]));
    }
}