Реализация прогресс-бара для долгих фоновых задач на сайте

Реализация прогресс-бара для долгих фоновых задач на сайте

Когда пользователь запускает задачу, которая занимает больше 2–3 секунд, интерфейс обязан показывать прогресс. Иначе — повторные клики, закрытие страницы, звонки в поддержку. Задача технически нетривиальная: прогресс формируется на сервере, нужно передавать его в браузер без polling каждые 500ms.

Архитектура

Схема работает так: клиент запускает задачу → получает job_id → подписывается на обновления через SSE или WebSocket → бэкенд обрабатывает задачу в очереди → воркер периодически публикует прогресс в Redis → SSE/WebSocket сервер доставляет обновления клиенту.

Polling (запросы каждые N секунд) работает, но создаёт ненужную нагрузку и даёт дёрганый UX. SSE — правильный выбор для однонаправленного потока прогресса.

Бэкенд: Laravel + Redis

// app/Jobs/ProcessImportJob.php
class ProcessImportJob implements ShouldQueue
{
    use Dispatchable, InteractsWithQueue, Queueable, SerializesModels;

    public function __construct(
        private readonly string $jobId,
        private readonly int $userId,
        private readonly string $filePath
    ) {}

    public function handle(): void
    {
        $rows = $this->parseFile($this->filePath);
        $total = count($rows);

        foreach ($rows as $index => $row) {
            $this->processRow($row);

            // Публикуем прогресс каждые 10 записей
            if ($index % 10 === 0 || $index === $total - 1) {
                $progress = (int)(($index + 1) / $total * 100);
                $this->publishProgress($progress, $index + 1, $total);
            }
        }

        $this->publishProgress(100, $total, $total, 'completed');
    }

    private function publishProgress(
        int $percent,
        int $processed,
        int $total,
        string $status = 'running'
    ): void {
        $payload = json_encode([
            'jobId'     => $this->jobId,
            'percent'   => $percent,
            'processed' => $processed,
            'total'     => $total,
            'status'    => $status,
            'ts'        => microtime(true),
        ]);

        // Публикуем в Redis Pub/Sub
        Redis::publish("job-progress:{$this->userId}", $payload);

        // Сохраняем последнее состояние для переподключений
        Redis::setex("job-progress-state:{$this->jobId}", 3600, $payload);
    }
}

// routes/api.php
Route::middleware('auth:sanctum')->group(function () {
    Route::post('/jobs/import', function (Request $request) {
        $jobId = Str::uuid()->toString();
        ProcessImportJob::dispatch($jobId, Auth::id(), $request->file('csv')->store('imports'));
        return response()->json(['jobId' => $jobId]);
    });

    Route::get('/jobs/{jobId}/progress-stream', function (string $jobId) {
        // Проверяем принадлежность задачи пользователю
        abort_unless(JobOwnership::check($jobId, Auth::id()), 403);

        return response()->stream(function () use ($jobId) {
            // Возвращаем последнее известное состояние сразу
            $lastState = Redis::get("job-progress-state:{$jobId}");
            if ($lastState) {
                echo "data: {$lastState}\n\n";
                ob_flush();
                flush();
            }

            $redis = new \Redis();
            $redis->connect(config('database.redis.default.host'));
            $redis->subscribe(["job-progress:" . Auth::id()], function ($redis, $channel, $message) use ($jobId) {
                $data = json_decode($message, true);
                if ($data['jobId'] !== $jobId) return; // фильтр по jobId

                echo "data: {$message}\n\n";
                ob_flush();
                flush();

                if ($data['status'] === 'completed' || $data['status'] === 'failed') {
                    $redis->unsubscribe();
                }
            });
        }, 200, [
            'Content-Type'      => 'text/event-stream',
            'Cache-Control'     => 'no-cache',
            'X-Accel-Buffering' => 'no', // отключает буферизацию Nginx
            'Connection'        => 'keep-alive',
        ]);
    });
});

Фронтенд: React компонент

import { useState, useEffect, useRef } from 'react';

interface JobProgress {
  jobId: string;
  percent: number;
  processed: number;
  total: number;
  status: 'running' | 'completed' | 'failed';
}

function useJobProgress(jobId: string | null) {
  const [progress, setProgress] = useState<JobProgress | null>(null);
  const eventSourceRef = useRef<EventSource | null>(null);

  useEffect(() => {
    if (!jobId) return;

    const es = new EventSource(`/api/jobs/${jobId}/progress-stream`, {
      withCredentials: true,
    });

    es.onmessage = (event) => {
      const data: JobProgress = JSON.parse(event.data);
      setProgress(data);

      if (data.status === 'completed' || data.status === 'failed') {
        es.close();
      }
    };

    es.onerror = () => {
      // SSE автоматически переподключается при ошибке
      // но при 403/404 — нет, поэтому проверяем статус
      es.close();
    };

    eventSourceRef.current = es;
    return () => es.close();
  }, [jobId]);

  return progress;
}

interface ProgressBarProps {
  percent: number;
  status: JobProgress['status'];
  processed: number;
  total: number;
}

function ProgressBar({ percent, status, processed, total }: ProgressBarProps) {
  return (
    <div className="w-full">
      <div className="flex justify-between text-sm text-gray-600 mb-1">
        <span>
          {status === 'completed' ? 'Готово' : `Обработано ${processed} из ${total}`}
        </span>
        <span>{percent}%</span>
      </div>
      <div className="w-full bg-gray-200 rounded-full h-2.5">
        <div
          className={`h-2.5 rounded-full transition-all duration-300 ${
            status === 'failed' ? 'bg-red-500' :
            status === 'completed' ? 'bg-green-500' :
            'bg-blue-600'
          }`}
          style={{ width: `${percent}%` }}
        />
      </div>
    </div>
  );
}

export function ImportWithProgress() {
  const [jobId, setJobId] = useState<string | null>(null);
  const progress = useJobProgress(jobId);

  async function handleFileUpload(file: File) {
    const formData = new FormData();
    formData.append('csv', file);

    const res = await fetch('/api/jobs/import', {
      method: 'POST',
      body: formData,
    });
    const { jobId } = await res.json();
    setJobId(jobId);
  }

  return (
    <div>
      {!jobId && (
        <input
          type="file"
          accept=".csv"
          onChange={(e) => e.target.files?.[0] && handleFileUpload(e.target.files[0])}
        />
      )}
      {progress && (
        <ProgressBar
          percent={progress.percent}
          status={progress.status}
          processed={progress.processed}
          total={progress.total}
        />
      )}
      {progress?.status === 'completed' && (
        <p className="text-green-600 mt-2">Импорт завершён успешно</p>
      )}
    </div>
  );
}

Nginx конфигурация

SSE требует отключения буферизации на уровне прокси:

location /api/jobs/ {
    proxy_pass http://backend;
    proxy_http_version 1.1;
    proxy_set_header Connection '';
    proxy_buffering off;
    proxy_cache off;
    proxy_read_timeout 3600s;  # долгие задачи
    chunked_transfer_encoding on;
}

Обработка падения воркера

Если воркер упал в середине задачи, прогресс зависнет. Нужен таймаут:

// В SSE-контроллере после подписки
$timeout = 0;
$maxWait = 300; // 5 минут

// Периодически проверяем, не умерла ли задача
while ($timeout < $maxWait) {
    sleep(5);
    $timeout += 5;
    $state = Redis::get("job-progress-state:{$jobId}");
    if (!$state) {
        echo "data: " . json_encode(['status' => 'failed', 'error' => 'timeout']) . "\n\n";
        flush();
        break;
    }
}

Сроки

Прогресс-бар для одного типа задачи с SSE — 1–2 дня. Универсальная система с несколькими типами задач, обработкой ошибок, историей задач пользователя и мониторингом зависших джобов — 4–5 дней.