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

namespace Aiwend\Knowledge;

use Aiwend\Queue\Job;
use Aiwend\Queue\Queue;

/**
 * Queue plumbing for GenerationBuilder: a build chain (knowledge.generation_build) and the dual-write that keeps the active
 * generation current (knowledge.generation_sync). Each job runs ONE bounded step and, if there is more to do, enqueues its
 * successor; the successor's idempotency key derives from the parent job id, so a duplicate run of the same job cannot fork the chain,
 * and a second start request for the same generation returns the job that already exists.
 *
 * Stale or duplicate jobs are harmless: a step on a generation that is no longer building (finished, failed, superseded) is
 * 'abandoned' and ends the chain. A job lost with its worker is recovered by the queue's lease expiry and simply re-runs the step.
 * A build is never activated here (activation is an explicit administrator action) and the handlers are inert unless the owner's
 * semantic gate is switched on by the caller that registers them.
 */
final class GenerationJobs
{
    public const BUILD = 'knowledge.generation_build';
    public const SYNC = 'knowledge.generation_sync';
    public const ATTEMPTS = 3;
    /** Consecutive steps that could make no progress (sources not settled) before the build gives up. */
    public const MAX_IDLE = 10;
    private const IDLE_DELAY = 60;

    public function __construct(
        private Queue $queue,
        private GenerationBuilder $builder,
        private IndexGenerations $generations,
        private EmbeddingConfigStore $configs,
        private ?GenerationControl $control = null,
    ) {
    }

    /**
     * Starts (or resumes, with a different $token) the build chain of a generation. The same token returns the existing job.
     *
     * @param ?int $tokenCeiling estimate units the whole build may spend; the units already spent are carried from step to step
     */
    public function enqueueBuild(int $siteId, int $generationId, string $token = 'start', ?int $tokenCeiling = null): int
    {
        return $this->queue->enqueue($siteId, self::BUILD, ['generation_id' => $generationId, 'idle' => 0, 'units' => 0, 'ceiling' => $tokenCeiling], "genbuild:$generationId:$token", 0, self::ATTEMPTS);
    }

    /** Asks for the active generation to be brought up to date. The same key (for example a source id and its content hash) is one job. */
    public function enqueueSync(int $siteId, string $key): int
    {
        return $this->queue->enqueue($siteId, self::SYNC, ['idle' => 0], "gensync:$key", 0, self::ATTEMPTS);
    }

    /**
     * Asks for the active generation to be brought up to date, coalesced: while a sync job of the site is still waiting to start, any
     * number of source events share it (its step looks at every source, not at the one that caused it). A job that is already running
     * may have scanned before this event, so it does not count; a new one is queued behind it. Returns the job id.
     */
    public function requestSync(int $siteId): int
    {
        return $this->queue->queued($siteId, self::SYNC) ?? $this->queue->enqueue($siteId, self::SYNC, ['idle' => 0], 'gensync:req:' . bin2hex(random_bytes(6)), 0, self::ATTEMPTS);
    }

    /** The administrator's own "update now": runs whether or not automatic updates are approved. */
    public function requestManualSync(int $siteId): int
    {
        return $this->queue->enqueue($siteId, self::SYNC, ['idle' => 0, 'manual' => true], 'gensync:manual:' . bin2hex(random_bytes(6)), 0, self::ATTEMPTS);
    }

    public function runBuild(Job $job): void
    {
        $generationId = (int) ($job->payload['generation_id'] ?? 0);
        if ($this->control?->isPaused($job->siteId, $generationId)) {
            return; // an administrator paused this build: no step, no successor; Resume starts a fresh job
        }
        $idle = (int) ($job->payload['idle'] ?? 0);
        $units = (int) ($job->payload['units'] ?? 0);
        $ceiling = isset($job->payload['ceiling']) ? (int) $job->payload['ceiling'] : null;
        $r = $this->builder->step($job->siteId, $generationId, max(1, $job->attempts) + 100 * $job->id, $this->configs->get($job->siteId), $ceiling, $units);
        $next = ['generation_id' => $generationId, 'idle' => 0, 'units' => $units + $r['units'], 'ceiling' => $ceiling];
        switch ($r['state']) {
            case 'continue':
                $this->successor($job, self::BUILD, $next, 0);
                return;
            case 'waiting':
                if ($idle + 1 >= self::MAX_IDLE) {
                    $this->giveUp($job->siteId, $generationId, 'source_not_settling');
                    return;
                }
                $this->successor($job, self::BUILD, ['idle' => $idle + 1] + $next, self::IDLE_DELAY);
                return;
            case 'retry':
                if ($job->attempts >= self::ATTEMPTS) {
                    $this->giveUp($job->siteId, $generationId, 'build_retries_exhausted');
                    return;
                }
                throw new \RuntimeException('Generation build step failed; it will be retried.');
            default:
                return; // complete (ready, awaiting explicit activation), paused (explicit resume), failed, abandoned
        }
    }

    public function runSync(Job $job): void
    {
        $active = $this->generations->peekState($job->siteId)['active'];
        if ($active === null) {
            return; // nothing active (deactivated or rolled away): nothing to keep current
        }
        if (!($job->payload['manual'] ?? false) && $this->control !== null && !$this->control->autoSync($job->siteId)) {
            return; // approval withdrawn since the job was queued: nothing is spent
        }
        $manual = ($job->payload['manual'] ?? false) === true;
        $idle = (int) ($job->payload['idle'] ?? 0);
        $r = $this->builder->step($job->siteId, $active, max(1, $job->attempts) + 100 * $job->id);
        switch ($r['state']) {
            case 'continue':
                $this->successor($job, self::SYNC, ['idle' => 0, 'manual' => $manual], 0);
                return;
            case 'waiting':
                if ($idle + 1 < self::MAX_IDLE) {
                    $this->successor($job, self::SYNC, ['idle' => $idle + 1, 'manual' => $manual], self::IDLE_DELAY);
                } else {
                    $this->generations->note($job->siteId, $active, 'sync_not_settling');
                }
                return;
            case 'retry':
                if ($job->attempts >= self::ATTEMPTS) {
                    $this->generations->note($job->siteId, $active, 'sync_failed'); // lexical keeps serving; the source is not answered semantically
                    return;
                }
                throw new \RuntimeException('Generation sync step failed; it will be retried.');
            default:
                return; // synced, paused (noted), abandoned
        }
    }

    /** @param array<string, mixed> $payload */
    private function successor(Job $parent, string $type, array $payload, int $delay): void
    {
        $this->queue->enqueue($parent->siteId, $type, $payload, "{$type}:after:{$parent->id}", $delay, self::ATTEMPTS);
    }

    private function giveUp(int $siteId, int $generationId, string $code): void
    {
        try {
            $this->generations->fail($siteId, $generationId, $code);
        } catch (GenerationException) {
            // already ready, active or ended: nothing to give up
        }
    }
}