KnowledgeBase/files/custom/Espo/Modules/KnowledgeBase/EntryPoints/KnowledgeBaseAskStream.php

<?php

namespace Espo\Modules\KnowledgeBase\EntryPoints;

use Espo\Core\Api\Request;
use Espo\Core\Api\Response;
use Espo\Core\EntryPoint\EntryPoint;
use Espo\Core\Exceptions\BadRequest;
use Espo\Core\Exceptions\Error;
use Espo\Core\Exceptions\Forbidden;
use Espo\Core\InjectableFactory;
use Espo\Entities\User;
use Espo\Modules\KnowledgeBase\Services\KnowledgeBaseService;

/**
 * Streams Server-Sent Events from shira-hermes /kb/ask/stream to the
 * browser's EventSource. The browser opens this with a GET (EventSource
 * has no POST), passes the question via ?message=, and we forward it
 * to shira-hermes with the firm-wide API key — the credential never
 * leaves the EspoCRM server.
 *
 * Response is text/event-stream with no buffering. We bypass EspoCRM's
 * Response wrapper entirely (echo + flush + exit) because the framework
 * is wired for buffered JSON responses, not infinite streams.
 *
 * Access: `?entryPoint=KnowledgeBaseAskStream&message=<urlencoded>`
 */
class KnowledgeBaseAskStream implements EntryPoint
{
    public function __construct(
        private InjectableFactory $injectableFactory,
        private User $user,
    ) {}

    public function run(Request $request, Response $response): void
    {
        if ($this->user->isPortal()) {
            throw new Forbidden('Portal users have no KB access.');
        }

        $message = $request->getQueryParam('message');
        if (!is_string($message) || trim($message) === '') {
            throw new BadRequest('message (string) is required.');
        }
        if (strlen($message) > 2000) {
            throw new BadRequest('message is too long.');
        }

        // Optional topic scope. EventSource is GET-only so the client passes
        // it via ?topicId=N; we forward it as topic_id in the JSON body to
        // /kb/ask/stream. Bad ints just become null and the upstream falls
        // back to the lowest active topic.
        $topicIdRaw = $request->getQueryParam('topicId');
        $topicId = (is_string($topicIdRaw) && is_numeric($topicIdRaw))
            ? (int) $topicIdRaw : null;

        $service = $this->injectableFactory->create(KnowledgeBaseService::class);
        try {
            $upstreamUrl = $service->getStreamingUrl('/kb/ask/stream');
            $apiKey = $service->getStreamingApiKey();
        } catch (\Throwable $e) {
            throw new Error('SmartAssistant integration is not configured: ' . $e->getMessage());
        }

        // Disable PHP/Apache output buffering and compression so each SSE
        // event flushes to the wire as soon as it arrives from upstream.
        // Order matters: clean any open buffers BEFORE setting headers.
        while (ob_get_level() > 0) {
            @ob_end_clean();
        }
        @ini_set('zlib.output_compression', 'Off');
        @ini_set('output_buffering', 'Off');
        @ini_set('implicit_flush', '1');
        if (function_exists('apache_setenv')) {
            @apache_setenv('no-gzip', '1');
        }

        ignore_user_abort(false);
        @set_time_limit(300);

        header('Content-Type: text/event-stream; charset=utf-8');
        header('Cache-Control: no-cache, no-transform');
        header('X-Accel-Buffering: no');
        header('Connection: keep-alive');
        // Defense in depth: same locked-down CSP we use for the PDF entry.
        header("Content-Security-Policy: default-src 'self'");

        $payloadArr = ['message' => $message];
        if ($topicId !== null) {
            $payloadArr['topic_id'] = $topicId;
        }
        $payload = json_encode($payloadArr, JSON_UNESCAPED_UNICODE);

        $ch = curl_init();
        curl_setopt_array($ch, [
            CURLOPT_URL => $upstreamUrl,
            CURLOPT_POST => true,
            CURLOPT_POSTFIELDS => $payload,
            CURLOPT_HTTPHEADER => [
                'Content-Type: application/json',
                'Accept: text/event-stream',
                'X-Api-Key: ' . $apiKey,
            ],
            CURLOPT_RETURNTRANSFER => false,
            CURLOPT_BUFFERSIZE => 1024,
            CURLOPT_TIMEOUT => 300,
            CURLOPT_CONNECTTIMEOUT => 10,
            CURLOPT_WRITEFUNCTION => function ($ch, $data) {
                if (connection_aborted()) {
                    // Browser closed the EventSource — bail so cURL stops.
                    return -1;
                }
                echo $data;
                @ob_flush();
                @flush();
                return strlen($data);
            },
        ]);
        $ok = curl_exec($ch);
        $errno = curl_errno($ch);
        $errmsg = curl_error($ch);
        curl_close($ch);

        if (!$ok && $errno !== 23 /* aborted by callback */) {
            // Best effort: if upstream errored before any data was
            // flushed, surface a final SSE error event so the client
            // doesn't silently spin.
            echo "data: " . json_encode(
                ['type' => 'error', 'message' => 'upstream: ' . $errmsg],
                JSON_UNESCAPED_UNICODE,
            ) . "\n\n";
            @ob_flush();
            @flush();
        }

        // We've fully written the response. Bypass EspoCRM's Response
        // emit() to avoid double-headers / extra buffering.
        exit;
    }
}