時間のかかる処理の進捗と AI 回答のストリーミング

エクスポート、インポート、エンコード、AI タスク：実際の処理に合わせて進むプログレスバー、完成した瞬間に届く結果、書かれるそばから表示される回答。

エクスポート、動画のエンコード、レポート作成、インポート、AI タスクには 10 秒から 10 分かかることがあります。何の手がかりもないスピナーは、再読み込みや二度押し、離脱を招きます。本当の進捗を見せ、結果ができた瞬間に表示すれば、ユーザーはページにとどまります。AI 機能にも使えます。処理がバックグラウンドジョブで動いていても、アシスタントの回答を書かれるそばからストリーミングできます。

ユーザーにとっての価値

実際の処理に合わせて進むプログレスバー：「10,000 行中 3,200 行を処理中」。
別のタブにいても、ジョブが終わった瞬間にダウンロードリンクや結果が表示されます。
AI の回答は、長く待たされずに少しずつ表示されます。
ページを離れても大丈夫。終わったら知らせます。

使う機能

ユーザー認証と sendToUser。ジョブの更新は、それを始めた本人のすべてのタブにだけ届きます。
複数の人が同じジョブを追うときは、プライベートチャンネル（private-job-<id>）。
どの言語のワーカーでも：Node.js、PHP、Python、Go、Ruby、Java、.NET の公式 Pusher サーバーライブラリは、キューのワーカーから配信できます。
アプリを閉じたユーザーに長いジョブの完了を知らせるプッシュ通知。

ステップ 1：ジョブを始めて ID を受け取る

app.post('/exports', requireLogin, async (req, res) => {
  const job = await queue.add('export', { userId: req.user.id, filters: req.body });
  res.status(202).json({ jobId: job.id });
});

ステップ 2：ワーカーから進捗を報告する

1 行ごとではなく、意味のある区切りで進捗を報告します：

worker.process('export', async (job) => {
  const user = String(job.data.userId);
  const rows = await db.orders.query(job.data.filters);
  for (let i = 0; i < rows.length; i += 1000) {
    await writeChunk(rows.slice(i, i + 1000));
    await pusher.sendToUser(user, 'job-progress', { jobId: job.id, done: Math.min(i + 1000, rows.length), total: rows.length });
  }
  const url = await uploadAndSign();
  await pusher.sendToUser(user, 'job-done', { jobId: job.id, url });
});

pusher パッケージを使う Python のワーカーでも同じです：

pusher_client.send_to_user(str(user_id), "job-progress", {"jobId": job_id, "done": done, "total": total})

ステップ 3：ブラウザに表示する

ユーザーがソケット上でサインインしていれば（アプリ内通知を参照）：

const { jobId } = await fetch('/exports', { method: 'POST', body: form }).then((r) => r.json());

pusher.user.bind('job-progress', (p) => {
  if (p.jobId === jobId) progress.set(p.done / p.total);
});
pusher.user.bind('job-done', (p) => {
  if (p.jobId === jobId) showDownload(p.url);
});

AI の回答をストリーミングする

AI モデルがバックグラウンドジョブで回答を書くときは、出力を小さなまとまりでユーザーに転送します。そうすれば長い回答でも数千ではなく数十メッセージで済みます：

let buffer = '';
let last = Date.now();
for await (const chunk of llm.stream(prompt)) {
  buffer += chunk.text;
  if (Date.now() - last > 150 || buffer.length > 2000) {
    await pusher.sendToUser(user, 'ai-delta', { id: answerId, text: buffer });
    buffer = '';
    last = Date.now();
  }
}
await pusher.sendToUser(user, 'ai-done', { id: answerId, text: buffer });

pusher.user.bind('ai-delta', ({ id, text }) => answers.get(id).append(text));
pusher.user.bind('ai-done', ({ id, text }) => answers.get(id).append(text).finish());

再読み込み後も残るよう、回答全体はサーバーにも保存しておきましょう。

ヒント

更新 1 回が 1 メッセージです。1〜5% ごと、または 150 ミリ秒ごとの更新で十分滑らかに見え、1 つのジョブは数十メッセージで済みます。
ジョブを始める前に購読するか、ページを開いたときにジョブの状態を 1 回取得すれば、ページが接続する前に送られた更新を見逃しません。
ユーザーがタブを閉じていたら、ジョブの完了時にプッシュ通知ガイドのとおりにプッシュを送りましょう。