04 Streaming

import type { FastifyInstance } from "fastify";
import { z } from "zod";
import { supportAgent } from "../ai/support-agent";

const SupportStreamRequest = z.object({
  message: z.string().trim().min(1, "message is required"),
});

export async function supportRoutes(app: FastifyInstance) {
  app.post("/support/stream", async (request, reply) => {
    const parsed = SupportStreamRequest.safeParse(request.body);

    if (!parsed.success) {
      return reply.status(400).send({
        error: { code: "bad_request", message: parsed.error.issues[0]?.message },
      });
    }

    return reply
      .header("Content-Type", "application/x-ndjson")
      .header("Cache-Control", "no-cache")
      .send(supportAgent.prompt(parsed.data.message).readableStream());
  });
}

const response = await fetch("http://localhost:3000/api/support/stream", {
  method: "POST",
  headers: { "Content-Type": "application/json" },
  body: JSON.stringify({ message: "Draft a support reply." }),
});

const reader = response.body?.getReader();
const decoder = new TextDecoder();

while (reader) {
  const next = await reader.read();
  if (next.done) break;

  for (const line of decoder.decode(next.value).split("\n")) {
    if (line.trim()) console.log(JSON.parse(line));
  }
}