math_cms/app/Http/Controllers/ImportStreamController.php

<?php

namespace App\Http\Controllers;

use Filament\Facades\Filament;
use Illuminate\Http\Request;
use Illuminate\Support\Facades\Redis;

class ImportStreamController extends Controller
{
    public function stream(Request $request): \Symfony\Component\HttpFoundation\StreamedResponse
    {
        if (!Filament::auth()->check()) {
            abort(403);
        }

        $type = $request->query('type', 'markdown-imports');
        $importId = (int) $request->query('import_id', 0);
        $channel = $type === 'pre-question-candidates' ? 'pre-question-candidates' : 'markdown-imports';

        return response()->stream(function () use ($channel, $importId) {
            echo "retry: 2000\n\n";
            @ob_flush();
            flush();

            Redis::connection()->subscribe([$channel], function ($message) use ($importId) {
                $payload = is_string($message) ? $message : json_encode($message, JSON_UNESCAPED_UNICODE);
                $decoded = json_decode($payload, true);

                if ($importId > 0 && is_array($decoded)) {
                    $messageImportId = (int) ($decoded['import_id'] ?? 0);
                    if ($messageImportId !== $importId) {
                        return;
                    }
                }

                echo "event: update\n";
                echo 'data: ' . $payload . "\n\n";
                @ob_flush();
                flush();
            });
        }, 200, [
            'Content-Type' => 'text/event-stream',
            'Cache-Control' => 'no-cache',
            'X-Accel-Buffering' => 'no',
            'Connection' => 'keep-alive',
        ]);
    }
}