<?php
declare(strict_types=1);

namespace app\common\service;

/** Request-local numeric telemetry. Raw events, prompts, titles, URLs and IDs never leave it. */
final class DifyStreamTiming
{
    private const MAX_MILLISECONDS = 3600000;
    private const MAX_COUNT = 1000000000;
    private const EVENTS = ['workflow_started', 'workflow_finished', 'node_started', 'node_finished', 'node_retry',
        'message', 'agent_message', 'agent_thought', 'message_end', 'error', 'ping', 'choices'];
    private const TYPES = ['start', 'llm', 'knowledge-retrieval', 'tool', 'code', 'template-transform',
        'if-else', 'answer', 'end', 'agent', 'http-request', 'iteration', 'question-classifier',
        'variable-aggregator', 'parameter-extractor', 'list-operator', 'document-extractor'];
    private \Closure $clock;
    private float $started;
    private int $lastElapsed = 0;
    private array $metrics = [];
    private array $events = [];
    private array $nodes = [];
    private array $nodePositions = [];

    public function __construct(?callable $clock = null)
    {
        $this->clock = $clock === null ? static fn (): float => hrtime(true) / 1e6 : \Closure::fromCallable($clock);
        $this->started = ($this->clock)();
    }

    private function elapsed(): int
    {
        $elapsed = ($this->clock)() - $this->started;
        if (!is_finite($elapsed)) return $this->lastElapsed;
        return $this->lastElapsed = max($this->lastElapsed,
            (int) min(self::MAX_MILLISECONDS, max(0, floor($elapsed))));
    }

    private function increment(string $key): void
    {
        $this->metrics[$key] = min(self::MAX_COUNT, (int) ($this->metrics[$key] ?? 0) + 1);
    }

    /** Abort signals also include validation rejection; they do not identify user intent. */
    public function interrupted(string $trigger): void
    {
        if (isset($this->metrics['interruption_detected_ms'])) return;
        $this->metrics['interruption_detected_ms'] = $this->elapsed();
        $this->metrics['interruption_trigger'] = in_array($trigger, ['abort_signal', 'callback_error', 'timeout'], true)
            ? $trigger : 'other';
    }

    /** Call only inside the callable that actually performs stop HTTP, never on lease polls. */
    public function stopIoBegin(): void
    {
        $now = $this->elapsed();
        $this->metrics['stop_io_begin_ms'] ??= $now;
        $this->metrics['last_stop_io_begin_ms'] = $now;
        $this->increment('stop_io_attempt_count');
    }

    public function stopIoEnd(bool $accepted): void
    {
        $now = $this->elapsed();
        $this->metrics['stop_io_end_ms'] ??= $now;
        $this->metrics['last_stop_io_end_ms'] = $now;
        $this->increment('stop_io_completed_count');
        if ($accepted) {
            $this->increment('stop_io_accepted_count');
            $this->stopAccepted('local_io');
        }
    }

    /** A peer can confirm acceptance without this transport performing stop I/O. */
    public function stopAccepted(string $source): void
    {
        if (isset($this->metrics['stop_accepted_ms'])) return;
        $this->metrics['stop_accepted_ms'] = $this->elapsed();
        $this->metrics['stop_accepted_source'] = in_array($source, ['local_io', 'coordinator'], true) ? $source : 'other';
    }

    /** The parser calls this only after its bound-task and allowed-status terminal gate. */
    public function workflowTerminal(string $status): void
    {
        if (isset($this->metrics['workflow_terminal_ms']) || !in_array($status, ['stopped', 'succeeded', 'failed'], true)) return;
        $this->metrics['workflow_terminal_ms'] = $this->elapsed();
        $this->metrics['workflow_terminal_status'] = $status;
    }

    /** Natural EOF means the HTTP response body completed; it is not provider/GPU proof. */
    public function transportReturned(string $outcome): void
    {
        if (isset($this->metrics['transport_return_ms'])) return;
        $this->metrics['transport_return_ms'] = $this->elapsed();
        $outcome = in_array($outcome, ['natural_eof', 'local_abort', 'timeout', 'transport_error', 'exception'], true)
            ? $outcome : 'other';
        $this->metrics['transport_exit'] = $outcome;
        if ($outcome === 'natural_eof') $this->metrics['natural_eof_ms'] = $this->metrics['transport_return_ms'];
    }

    public function transportClosed(): void { $this->metrics['transport_closed_ms'] ??= $this->elapsed(); }

    public function headers(int $status): void
    {
        $this->metrics['first_header_ms'] ??= $this->elapsed();
        if ($status >= 200 && $status <= 599) $this->metrics['final_header_ms'] ??= $this->elapsed();
    }

    public function body(int $bytes): void
    {
        if ($bytes <= 0) return;
        $this->metrics['first_body_ms'] ??= $this->elapsed();
    }

    public function frame(array $decoded, string $protocol): void
    {
        $this->metrics['first_frame_ms'] ??= $this->elapsed();
        $name = $protocol === 'dify' ? ($decoded['event'] ?? 'other') : (isset($decoded['choices']) ? 'choices' : 'other');
        if (!is_string($name) || !in_array($name, self::EVENTS, true)) $name = 'other';
        $now = $this->elapsed();
        $this->events[$name] ??= ['first_ms' => $now, 'last_ms' => $now, 'count' => 0];
        ++$this->events[$name]['count']; $this->events[$name]['last_ms'] = $now;
        if ($name === 'node_retry') {
            $retry = $decoded['data']['retry_index'] ?? null;
            if (is_numeric($retry) && (float) $retry >= 0 && (float) $retry <= 1000) {
                $this->metrics['max_node_retry_index'] = max((int) ($this->metrics['max_node_retry_index'] ?? 0), (int) $retry);
            }
        }
        $reasoning = $protocol === 'dify' ? ($decoded['thought'] ?? null)
            : ($decoded['choices'][0]['delta']['reasoning_content'] ?? null);
        if (is_string($reasoning) && $reasoning !== '') {
            $this->metrics['first_reasoning_frame_ms'] ??= $now;
            $this->metrics['reasoning_frame_count'] = (int) ($this->metrics['reasoning_frame_count'] ?? 0) + 1;
            $this->metrics['reasoning_frame_bytes'] = min(1000000000,
                (int) ($this->metrics['reasoning_frame_bytes'] ?? 0) + strlen($reasoning));
        }
        if (in_array($name, ['message', 'agent_message', 'choices'], true)) {
            $answer = $protocol === 'dify' ? ($decoded['answer'] ?? '') : ($decoded['choices'][0]['delta']['content'] ?? '');
            $visible = is_string($answer) && $answer !== '';
            // Chat Completions also supports text-part arrays. Match the parser's
            // accepted shape without retaining or concatenating clinical content.
            if ($protocol !== 'dify' && is_array($answer)) {
                foreach ($answer as $part) {
                    if (is_array($part) && ($part['type'] ?? '') === 'text'
                        && is_string($part['text'] ?? null) && $part['text'] !== '') { $visible = true; break; }
                }
            }
            if ($visible) $this->metrics['first_answer_ms'] ??= $now;
        }
        if (in_array($name, ['node_started', 'node_finished'], true)) {
            $data = is_array($decoded['data'] ?? null) ? $decoded['data'] : [];
            $id = $data['id'] ?? $data['node_id'] ?? '';
            if (is_string($id) && $id !== '' && strlen($id) <= 512) {
                $key = hash('sha256', $id);
                if (!isset($this->nodePositions[$key]) && count($this->nodes) >= 128) return;
                if (!isset($this->nodePositions[$key])) {
                    $this->nodePositions[$key] = count($this->nodes);
                    $type = $data['node_type'] ?? 'other';
                    $this->nodes[] = ['type' => is_string($type) && in_array($type, self::TYPES, true) ? $type : 'other'];
                }
                $position = $this->nodePositions[$key];
                $field = $name === 'node_started' ? 'start_ms' : 'finish_ms';
                $this->nodes[$position][$field] ??= $now;
                $elapsed = $data['elapsed_time'] ?? null;
                if (is_numeric($elapsed) && is_finite((float) $elapsed) && (float) $elapsed >= 0) {
                    $this->nodes[$position]['provider_elapsed_ms'] = (int) min(3600000, (float) $elapsed * 1000);
                }
                if ($this->nodes[$position]['type'] === 'llm' && $name === 'node_finished') {
                    $this->imageCounts($position, $data);
                }
            }
        }
        $usage = $name === 'message_end' ? ($decoded['metadata']['usage'] ?? []) : ($decoded['usage'] ?? []);
        if (is_array($usage) && !isset($usage['reasoning_tokens'])
            && isset($usage['completion_tokens_details']['reasoning_tokens'])) {
            $usage['reasoning_tokens'] = $usage['completion_tokens_details']['reasoning_tokens'];
        }
        foreach (['prompt_tokens', 'completion_tokens', 'total_tokens', 'reasoning_tokens'] as $key) {
            $value = is_array($usage) ? ($usage[$key] ?? null) : null;
            if (is_numeric($value) && is_finite((float) $value) && (float) $value >= 0) {
                $this->metrics[$key] = (int) min(1000000000, (float) $value);
            }
        }
    }

    public function delivered(): void { $this->metrics['first_delivered_ms'] ??= $this->elapsed(); }

    /** Count the actual LLM image parts when the provider exposes them, never their contents. */
    private function imageCounts(int $position, array $data): void
    {
        $files = $data['inputs']['#files#'] ?? null;
        if (is_array($files) && count($files) <= 1000) {
            $this->nodes[$position]['input_image_count'] = count(array_filter($files,
                static fn ($file): bool => is_array($file) && ($file['type'] ?? '') === 'image'));
        }
        $prompts = $data['process_data']['prompts'] ?? null;
        if (!is_array($prompts) || count($prompts) > 512) return;
        $parts = 0;
        $high = 0;
        foreach ($prompts as $prompt) {
            if (!is_array($prompt) || !is_array($prompt['files'] ?? null) || count($prompt['files']) > 1000) return;
            foreach ($prompt['files'] as $file) {
                if (!is_array($file) || ($file['type'] ?? '') !== 'image') continue;
                ++$parts;
                if (($file['detail'] ?? '') === 'high') ++$high;
            }
        }
        $this->nodes[$position]['model_image_part_count'] = $parts;
        $this->nodes[$position]['model_high_detail_image_part_count'] = $high;
    }

    public function transport(array $values): void
    {
        foreach (['connect_ms', 'tls_ms', 'header_ttfb_ms', 'transfer_ms', 'http_status', 'curl_errno'] as $key) {
            $value = $values[$key] ?? null;
            if ((is_float($value) || is_int($value)) && is_finite((float) $value) && $value >= 0) {
                $this->metrics[$key] = (int) min(3600000, $value);
            }
        }
    }

    public function snapshot(): array
    {
        return ['duration_ms' => $this->elapsed()] + $this->metrics + ['events' => $this->events, 'nodes' => $this->nodes];
    }
}