Persistent Text Streaming

Copy codenpm install @convex-dev/persistent-text-streaming

Copy code// convex/convex.config.ts
import { defineApp } from "convex/server";
import persistentTextStreaming from "@convex-dev/persistent-text-streaming/convex.config.js";

const app = defineApp();
app.use(persistentTextStreaming);
export default app;

Copy codeconst persistentTextStreaming = new PersistentTextStreaming(
  components.persistentTextStreaming,
);

// Create a stream using the component and store the id in the database with
// our chat message.
export const createChat = mutation({
  args: {
    prompt: v.string(),
  },
  handler: async (ctx, args) => {
    const streamId = await persistentTextStreaming.createStream(ctx);
    const chatId = await ctx.db.insert("chats", {
      title: "...",
      prompt: args.prompt,
      stream: streamId,
    });
    return chatId;
  },
});

// Create a query that returns the chat body.
export const getChatBody = query({
  args: {
    streamId: StreamIdValidator,
  },
  handler: async (ctx, args) => {
    return await persistentTextStreaming.getStreamBody(
      ctx,
      args.streamId as StreamId,
    );
  },
});

// Create an HTTP action that generates chunks of the chat body
// and uses the component to stream them to the client and save them to the database.
export const streamChat = httpAction(async (ctx, request) => {
  const body = (await request.json()) as { streamId: string };
  const generateChat = async (ctx, request, streamId, chunkAppender) => {
    await chunkAppender("Hi there!");
    await chunkAppender("How are you?");
    await chunkAppender("Pretend I'm an AI or something!");
  };

  const response = await persistentTextStreaming.stream(
    ctx,
    request,
    body.streamId as StreamId,
    generateChat,
  );

  // Set CORS headers appropriately.
  response.headers.set("Access-Control-Allow-Origin", "*");
  response.headers.set("Vary", "Origin");
  return response;
});

Copy codehttp.route({
  path: "/chat-stream",
  method: "POST",
  handler: streamChat,
});

// Handle CORS preflight requests so browsers will allow the POST above when
// your app is served from a different origin than your Convex deployment.
http.route({
  path: "/chat-stream",
  method: "OPTIONS",
  handler: httpAction(async (_, request) => {
    const headers = request.headers;
    if (
      headers.get("Origin") !== null &&
      headers.get("Access-Control-Request-Method") !== null &&
      headers.get("Access-Control-Request-Headers") !== null
    ) {
      return new Response(null, {
        headers: new Headers({
          "Access-Control-Allow-Origin": "*",
          "Access-Control-Allow-Methods": "POST",
          "Access-Control-Allow-Headers": "Content-Type, Digest, Authorization",
          "Access-Control-Max-Age": "86400",
        }),
      });
    } else {
      return new Response();
    }
  }),
});

Copy code// chat-input.tsx, maybe?
const createChat = useMutation(api.chat.createChat);
const formSubmit = async (e: React.FormEvent) => {
  e.preventDefault();
  const chatId = await createChat({
    prompt: inputValue,
  });
};

// chat-message.tsx, maybe?
import { useStream } from "@convex-dev/persistent-text-streaming/react";

// ...

// In our component:
const { text, status } = useStream(
  api.chat.getChatBody, // The query to call for the full stream body
  new URL(`${convexSiteUrl}/chat-stream`), // The HTTP endpoint for streaming
  driven, // True if this browser session created this chat and should generate the stream
  chat.streamId as StreamId, // The streamId from the chat database record
);

Copy code<script lang="ts">
  import { useStream } from "@convex-dev/persistent-text-streaming/svelte";

  let { chat, driven }: { chat: Doc<"chats">; driven: boolean } = $props();

  const stream = useStream(
    api.chat.getChatBody,
    new URL(`${convexSiteUrl}/chat-stream`),
    () => driven,
    () => chat.streamId as StreamId,
    // Optional, read when a stream starts:
    // () => ({ authToken, headers }),
  );
</script>

<p>{stream.text}</p>