Streams can now have up to 2500 entries

Author: ericallam

apps/webapp/app/services/realtimeStreams.server.ts

@@ -134,7 +134,7 @@ export class RealtimeStreams {
           if (line.trim()) {
             // Avoid unnecessary parsing; assume 'line' is already a JSON string
             // Add XADD command with MAXLEN option to limit stream size
-            batchCommands.push([streamKey, "MAXLEN", "~", "1000", "*", "data", line]);
+            batchCommands.push([streamKey, "MAXLEN", "~", "2500", "*", "data", line]);
 
             if (batchCommands.length >= batchSize) {
               // Send batch using a pipeline

pnpm-lock.yaml

@@ -10,7 +10,7 @@
     "dev:trigger": "trigger dev"
   },
   "dependencies": {
-    "@ai-sdk/openai": "^0.0.72",
+    "@ai-sdk/openai": "^1.0.1",
     "@fal-ai/serverless-client": "^0.15.0",
     "@radix-ui/react-dialog": "^1.0.3",
     "@radix-ui/react-icons": "^1.3.0",
@@ -19,7 +19,7 @@
     "@trigger.dev/react-hooks": "workspace:^3",
     "@trigger.dev/sdk": "workspace:^3",
     "@uploadthing/react": "^7.0.3",
-    "ai": "^3.4.33",
+    "ai": "^4.0.0",
     "class-variance-authority": "^0.7.0",
     "clsx": "^2.1.1",
     "lucide-react": "^0.451.0",

references/nextjs-realtime/package.json

@@ -114,3 +114,33 @@ export const openaiStreaming = schemaTask({
     return { text };
   },
 });
+
+export const openaiO1Model = schemaTask({
+  id: "openai-o1-model",
+  description: "Stream data from OpenAI to get the weather",
+  schema: z.object({
+    model: z.string().default("o1-preview"),
+    prompt: z.string().default("Hello, how are you?"),
+  }),
+  run: async ({ model, prompt }) => {
+    logger.info("Running OpenAI model", { model, prompt });
+
+    const result = await streamText({
+      model: openai(model),
+      prompt,
+      experimental_continueSteps: true,
+    });
+
+    const stream = await metadata.stream("openai", result.textStream);
+
+    let text = "";
+
+    for await (const chunk of stream) {
+      logger.log("Received chunk", { chunk });
+
+      text += chunk;
+    }
+
+    return { text };
+  },
+});