<?php

declare(strict_types=1);

namespace app\adminapi\service;

use app\common\service\DifyChatService;

/**
 * Shared between FPM workers on one host. Contains no credentials or clinical text.
 * The caller MUST authorize the diagnosis before constructing or reading a control.
 */
final class AssistantRequestControl
{
    private const STOP_LEASE_SECONDS = 1.25;
    private const STOP_RETRY_DELAY_SECONDS = 0.25;
    private const STOP_RETENTION_SECONDS = 30;
    private const MAX_STOP_ATTEMPTS = 2;

    private string $path;
    private int $ttl;
    private string $directory;

    public function __construct(int $adminId, int $diagnosisId, string $requestId, ?string $directory = null, int $ttl = 1200)
    {
        if ($adminId <= 0 || $diagnosisId <= 0 || !self::validRequestId($requestId)) {
            throw new \InvalidArgumentException('Invalid assistant request identity');
        }
        $directory = $directory ?? app()->getRuntimePath() . 'assistant-requests';
        if (!is_dir($directory) && !@mkdir($directory, 0700, true) && !is_dir($directory)) {
            throw new \RuntimeException('Assistant request control unavailable');
        }
        $this->path = rtrim($directory, '/\\') . DIRECTORY_SEPARATOR
            . hash('sha256', $adminId . ':' . $diagnosisId . ':' . strtolower($requestId)) . '.json';
        $this->ttl = max(600, $ttl);
        $this->directory = $directory;
        $this->collectExpired();
    }

    public static function validRequestId(string $id): bool
    {
        return preg_match('/^[0-9a-f]{8}-[0-9a-f]{4}-[1-8][0-9a-f]{3}-[89ab][0-9a-f]{3}-[0-9a-f]{12}$/i', $id) === 1;
    }

    /** A cancel tombstone created before prepare is retained; a replay never restarts work. */
    public function begin(): bool
    {
        return $this->update(static function (array &$state): bool {
            if (!empty($state['begun'])) {
                return false;
            }
            $state['begun'] = true;
            $state['status'] = !empty($state['cancel_requested']) ? 'cancel_requested' : 'running';
            return true;
        });
    }

    public function shouldAbort(): bool
    {
        return $this->update(static fn (array &$state): bool => !empty($state['cancel_requested']), false);
    }

    public function bindTask(string $profile, string $user, string $taskId): void
    {
        if (!in_array($profile, ['qwen', 'openai'], true)
            || preg_match('/^[a-zA-Z0-9_-]{1,128}$/', $taskId) !== 1 || strlen($user) > 128) {
            throw new \RuntimeException('Invalid upstream task identity');
        }
        $this->update(static function (array &$state) use ($profile, $user, $taskId): void {
            if (in_array($state['status'] ?? '', ['done', 'error', 'cancelled'], true)) return;
            $previous = $state['upstream'] ?? [];
            if (($previous['task_id'] ?? '') === $taskId && ($previous['user'] ?? '') === $user
                && ($previous['profile'] ?? '') === $profile) return;
            $state['task_generation'] = (int) ($state['task_generation'] ?? 0) + 1;
            $state['upstream'] = ['profile' => $profile, 'user' => $user, 'task_id' => $taskId,
                'generation' => $state['task_generation']];
            $state['upstream_stop_confirmed'] = false;
            $state['upstream_terminal_confirmed'] = false;
            $state['upstream_terminal_status'] = '';
            $state['stop_attempts'] = 0;
            unset($state['stop_claim'], $state['stop_retry_after']);
        });
    }

    public function clearTask(?string $taskId = null): void
    {
        $this->update(static function (array &$state) use ($taskId): void {
            if ($taskId !== null && ($state['upstream']['task_id'] ?? '') !== $taskId) return;
            self::releaseTask($state);
        });
    }

    public function markStop(bool $confirmed, ?string $taskId = null): void
    {
        $this->update(static function (array &$state) use ($confirmed, $taskId): void {
            if ($taskId !== null && ($state['upstream']['task_id'] ?? '') !== $taskId) return;
            $state['upstream_stop_confirmed'] = !empty($state['upstream_stop_confirmed']) || $confirmed;
        });
    }

    /** Only a real workflow terminal event confirms execution ended; stop acceptance does not. */
    public function markTerminal(string $taskId, string $status): void
    {
        if (!in_array($status, ['stopped', 'succeeded', 'failed'], true)) return;
        $this->update(static function (array &$state) use ($taskId, $status): void {
            if (($state['upstream']['task_id'] ?? '') !== $taskId || !empty($state['upstream_terminal_confirmed'])) return;
            $state['upstream_terminal_confirmed'] = true;
            $state['upstream_terminal_status'] = $status;
        });
    }

    /** Safe inside a transport callback: writes intent and never performs network I/O. */
    public function signalCancel(): void
    {
        $this->update(static function (array &$state): void {
            if (in_array($state['status'] ?? '', ['done', 'error'], true)) return;
            $state['cancel_requested'] = true;
            if (($state['status'] ?? '') !== 'cancelled') $state['status'] = 'cancel_requested';
            $state['stop_retry_until'] = $state['stop_retry_until'] ?? microtime(true) + self::STOP_RETENTION_SECONDS;
        });
    }

    /** Safe client view. A requested cancellation is never reported as confirmed stop. */
    public function requestCancel(?callable $stop = null): array
    {
        $this->signalCancel();
        $this->stopBoundTask($stop);
        return $this->publicState();
    }

    /**
     * Shared by the cancel endpoint and stream owner. The lease exceeds the 1s stop
     * transport budget; a crashed claimant can be replaced, while accepted stops
     * never repeat. Failed tasks remain eligible for one short-lived compensation.
     * @return array{accepted:bool,pending:bool}
     */
    public function stopBoundTask(?callable $stop = null, ?string $expectedTaskId = null): array
    {
        $token = bin2hex(random_bytes(12));
        $claim = $this->update(static function (array &$state) use ($token, $expectedTaskId): array {
            $now = microtime(true);
            $task = $state['upstream'] ?? null;
            $accepted = !empty($state['upstream_stop_confirmed']);
            if (!is_array($task) || ($expectedTaskId !== null && $task['task_id'] !== $expectedTaskId)) {
                return ['accepted' => false, 'pending' => false];
            }
            // A cancel worker may load a task written by the previous local
            // release. Upgrade its private identity atomically before claiming.
            if (!isset($task['generation'])) {
                $state['task_generation'] = (int) ($state['task_generation'] ?? 0) + 1;
                $task['generation'] = $state['task_generation'];
                $state['upstream'] = $task;
            }
            if ($accepted || !empty($state['upstream_terminal_confirmed']) || empty($state['cancel_requested'])
                || (float) ($state['stop_retry_until'] ?? 0) <= $now) {
                return ['accepted' => $accepted, 'pending' => false];
            }
            if ((float) ($state['stop_claim']['until'] ?? 0) > $now
                || (float) ($state['stop_retry_after'] ?? 0) > $now) {
                return ['accepted' => false, 'pending' => true];
            }
            if ((int) ($state['stop_attempts'] ?? 0) >= self::MAX_STOP_ATTEMPTS) {
                return ['accepted' => false, 'pending' => false];
            }
            $state['stop_claim'] = ['token' => $token, 'until' => $now + self::STOP_LEASE_SECONDS];
            $state['stop_attempts'] = (int) ($state['stop_attempts'] ?? 0) + 1;
            return ['task' => $task, 'accepted' => false, 'pending' => true];
        });
        if (!isset($claim['task'])) return $claim;
        $task = $claim['task'];
        $confirmed = false;
        // Never hold either file lock while waiting for the remote stop endpoint.
        try {
            $confirmed = ($stop !== null ? $stop($task) : DifyChatService::stopTask(
                $task['profile'], $task['task_id'], $task['user']
            )) === true;
        } catch (\Throwable $ignored) {}
        return $this->update(static function (array &$state) use ($task, $token, $confirmed): array {
            if (($state['upstream']['generation'] ?? 0) !== $task['generation']
                || ($state['upstream']['task_id'] ?? '') !== $task['task_id']
                || ($state['stop_claim']['token'] ?? '') !== $token) {
                return ['accepted' => false, 'pending' => false];
            }
            $state['upstream_stop_confirmed'] = !empty($state['upstream_stop_confirmed']) || $confirmed;
            unset($state['stop_claim']);
            $state['stop_retry_after'] = microtime(true) + self::STOP_RETRY_DELAY_SECONDS;
            return ['accepted' => !empty($state['upstream_stop_confirmed']), 'pending' => false];
        });
    }

    private static function releaseTask(array &$state): void
    {
        if (!empty($state['cancel_requested']) && empty($state['upstream_terminal_confirmed'])
            && (float) ($state['stop_retry_until'] ?? 0) > microtime(true)) {
            return;
        }
        unset($state['upstream'], $state['stop_claim']);
    }

    public function finish(string $status): void
    {
        if (!in_array($status, ['done', 'error', 'cancelled'], true)) {
            throw new \InvalidArgumentException('Invalid assistant terminal state');
        }
        $this->update(static function (array &$state) use ($status): void {
            if (in_array($state['status'] ?? '', ['done', 'error', 'cancelled'], true)) {
                return;
            }
            $state['status'] = !empty($state['cancel_requested']) ? 'cancelled' : $status;
            $state['transport_closed'] = true;
            self::releaseTask($state);
        });
    }

    public function publicState(): array
    {
        return $this->update(static fn (array &$state): array => [
            'status' => $state['status'] ?? 'pending',
            'cancel_requested' => !empty($state['cancel_requested']),
            'transport_closed' => !empty($state['transport_closed']),
            'upstream_stop_confirmed' => !empty($state['upstream_stop_confirmed']),
            'upstream_terminal_confirmed' => !empty($state['upstream_terminal_confirmed']),
            'upstream_terminal_status' => (string) ($state['upstream_terminal_status'] ?? ''),
        ], false);
    }

    private function update(callable $apply, bool $persist = true)
    {
        $guard = @fopen($this->directory . DIRECTORY_SEPARATOR . '.gc.lock', 'c+');
        if ($guard === false || !flock($guard, LOCK_SH)) {
            if (is_resource($guard)) fclose($guard);
            throw new \RuntimeException('Assistant request control unavailable');
        }
        $handle = @fopen($this->path, 'c+');
        if ($handle === false || !flock($handle, $persist ? LOCK_EX : LOCK_SH)) {
            if (is_resource($handle)) fclose($handle);
            flock($guard, LOCK_UN);
            fclose($guard);
            throw new \RuntimeException('Assistant request control unavailable');
        }
        try {
            @chmod($this->path, 0600);
            $raw = stream_get_contents($handle);
            $state = is_string($raw) ? json_decode($raw, true) : null;
            if ($raw !== '' && !is_array($state)) throw new \RuntimeException('Assistant request control unavailable');
            if (!is_array($state) || (int) ($state['expires_at'] ?? 0) <= time()) {
                $state = ['expires_at' => time() + $this->ttl];
            }
            $result = $apply($state);
            if (!$persist) return $result;
            $state['expires_at'] = time() + $this->ttl;
            $encoded = json_encode($state, JSON_THROW_ON_ERROR);
            rewind($handle);
            if (!ftruncate($handle, 0) || fwrite($handle, $encoded) !== strlen($encoded) || !fflush($handle)) {
                throw new \RuntimeException('Assistant request control unavailable');
            }
            return $result;
        } finally {
            flock($handle, LOCK_UN);
            fclose($handle);
            flock($guard, LOCK_UN);
            fclose($guard);
        }
    }

    /** The global guard prevents unlink/reopen races with any other worker. */
    private function collectExpired(): void
    {
        $guard = @fopen($this->directory . DIRECTORY_SEPARATOR . '.gc.lock', 'c+');
        if ($guard === false) return;
        try {
            if (!flock($guard, LOCK_EX | LOCK_NB)) return;
            $paths = glob($this->directory . DIRECTORY_SEPARATOR . '*.json') ?: [];
            // Bound each sweep; rotating the scan also reclaims older files in large directories.
            if (count($paths) > 256) shuffle($paths);
            foreach (array_slice($paths, 0, 256) as $path) {
                if ((int) @filemtime($path) > time() - 600) continue;
                $state = json_decode((string) @file_get_contents($path), true);
                if (!is_array($state) || (int) ($state['expires_at'] ?? 0) <= time()) @unlink($path);
            }
        } finally {
            flock($guard, LOCK_UN);
            fclose($guard);
        }
    }
}