feat: implement background streaming with Convex actions - #520
Conversation
- Add streamJobs table to Convex schema for tracking stream jobs - Create backgroundStream.ts with Convex action that calls OpenRouter independently - Refactor use-persistent-chat hook to use Convex-based streaming - Remove skeleton loader for faster perceived loading - Add Redis client and typing endpoint for future features - Streaming now survives page reloads via Convex subscriptions
|
🚅 Deployed to the openchat-pr-520 environment in OpenChat
|
🚀 Preview Deployment Ready
Convex Preview Backend
🤖 Deployed automatically by GitHub Actions |
Greptile SummaryThis PR successfully migrates chat streaming from HTTP endpoints to Convex background actions, enabling streams to survive page reloads. The core implementation is solid with reactive subscriptions replacing polling. Key Changes
Issues Found
Recommendations
Confidence Score: 3/5
Important Files Changed
Sequence DiagramsequenceDiagram
participant User
participant Frontend as Frontend<br/>(use-persistent-chat)
participant Convex as Convex DB
participant Action as Convex Action<br/>(executeStream)
participant OpenRouter as OpenRouter API
User->>Frontend: Send message
Frontend->>Convex: chats.create (if new chat)
Convex-->>Frontend: chatId
Frontend->>Convex: messages.send (user message)
Frontend->>Convex: backgroundStream.startStream
Note over Convex: Creates streamJob<br/>status: "pending"
Convex->>Convex: Update chat.activeStreamId
Convex->>Action: Schedule executeStream action
Convex-->>Frontend: jobId
Frontend->>Convex: Subscribe to getActiveStreamJob query
Action->>Convex: Update job status: "running"
Action->>OpenRouter: POST /chat/completions (stream: true)
loop Stream tokens
OpenRouter-->>Action: SSE chunk (delta.content)
Action->>Action: Accumulate content (batch 5 tokens)
Action->>Convex: updateStreamContent mutation
Convex-->>Frontend: Reactive update (via subscription)
Frontend->>User: Display streaming text
end
OpenRouter-->>Action: [DONE]
Action->>Convex: completeStream mutation
Note over Convex: Insert/update message<br/>Clear activeStreamId<br/>Set job: "completed"
Convex-->>Frontend: Final reactive update
Frontend->>User: Show completed message
alt On page reload during stream
User->>Frontend: Reload page
Frontend->>Convex: getActiveStreamJob query
Convex-->>Frontend: Current job with partial content
Frontend->>User: Resume displaying stream
end
alt On error
OpenRouter-->>Action: Error response
Action->>Convex: failStream mutation
Convex-->>Frontend: Error status update
Frontend->>User: Display error
end
|
![greptile-apps[bot]](https://avatars.githubusercontent.com/in/867647?s=60&v=4){kind=small}
| }); | ||
| updateCounter = 0; | ||
| } | ||
| } catch {} |
There was a problem hiding this comment.
style: silent error swallowing - parsing errors are completely ignored
| } catch {} | |
| } catch (parseError) { | |
| console.error("[BackgroundStream] Failed to parse SSE chunk:", data, parseError); | |
| } |
Confidence: 4/5 - This is a clear code smell where parsing errors are silently ignored. While it may not cause immediate issues, it makes debugging extremely difficult when malformed SSE data arrives.
| await ctx.scheduler.runAfter(0, internal.backgroundStream.executeStream, { | ||
| jobId, | ||
| apiKey: args.apiKey, | ||
| }); |
There was a problem hiding this comment.
logic: race condition - if another stream starts immediately after this one (within milliseconds), both could have activeStreamId set simultaneously
The second startStream call should check if there's already an active stream:
| await ctx.scheduler.runAfter(0, internal.backgroundStream.executeStream, { | |
| jobId, | |
| apiKey: args.apiKey, | |
| }); | |
| const existingActiveStream = await ctx.db | |
| .query("streamJobs") | |
| .withIndex("by_chat", (q) => | |
| q.eq("chatId", args.chatId).eq("status", "running") | |
| ) | |
| .first(); | |
| if (existingActiveStream) { | |
| throw new Error("Stream already in progress for this chat"); | |
| } | |
| await ctx.scheduler.runAfter(0, internal.backgroundStream.executeStream, { |
Confidence: 2/5 - This is a potential edge case, but concurrent stream starts could cause issues.
| const allMsgs = messages.map((m) => { | ||
| const textPart = m.parts?.find((p): p is { type: "text"; text: string } => p.type === "text"); | ||
| return { role: m.role, content: textPart?.text || "" }; | ||
| }); | ||
| allMsgs.push({ role: "user", content: message.text }); |
There was a problem hiding this comment.
logic: message history excludes reasoning parts - when building the conversation history for the AI, only text parts are extracted, losing reasoning context
| const allMsgs = messages.map((m) => { | |
| const textPart = m.parts?.find((p): p is { type: "text"; text: string } => p.type === "text"); | |
| return { role: m.role, content: textPart?.text || "" }; | |
| }); | |
| allMsgs.push({ role: "user", content: message.text }); | |
| const allMsgs = messages.map((m) => { | |
| const textPart = m.parts?.find((p): p is { type: "text"; text: string } => p.type === "text"); | |
| const reasoningPart = m.parts?.find((p): p is { type: "reasoning"; text: string } => p.type === "reasoning"); | |
| const content = [reasoningPart?.text, textPart?.text].filter(Boolean).join("\n\n"); | |
| return { role: m.role, content: content || "" }; | |
| }); |
Confidence: 3/5 - Depending on the AI model, excluding reasoning from context might reduce response quality for follow-up messages.
Note: If this suggestion doesn't match your team's coding style, reply to this and let me know. I'll remember it for next time!
| let fullReasoning = ""; | ||
| let buffer = ""; | ||
| let updateCounter = 0; | ||
| const UPDATE_INTERVAL = 5; |
There was a problem hiding this comment.
style: hardcoded batch size may cause UI lag - UPDATE_INTERVAL = 5 means updates every 5 tokens, which could feel laggy for fast models
Consider making this configurable or reducing to 3:
| const UPDATE_INTERVAL = 5; | |
| const UPDATE_INTERVAL = 3; |
Confidence: 2/5 - This is a UX trade-off. Lower values increase DB writes but improve perceived responsiveness.
Note: If this suggestion doesn't match your team's coding style, reply to this and let me know. I'll remember it for next time!
| const pattern = `chat:${chatId}:typing:*`; | ||
| const foundKeys = await client.keys(pattern); | ||
|
|
||
| return foundKeys.map((key) => key.split(":").pop() || ""); |
There was a problem hiding this comment.
logic: inefficient key scanning - client.keys(pattern) scans all keys and blocks Redis, which can cause performance issues at scale
| const pattern = `chat:${chatId}:typing:*`; | |
| const foundKeys = await client.keys(pattern); | |
| return foundKeys.map((key) => key.split(":").pop() || ""); | |
| const pattern = `chat:${chatId}:typing:*`; | |
| const foundKeys: string[] = []; | |
| for await (const key of client.scanIterator({ MATCH: pattern, COUNT: 100 })) { | |
| foundKeys.push(key); | |
| } | |
| return foundKeys.map((key) => key.split(":").pop() || ""); |
Confidence: 4/5 - KEYS command is documented as dangerous in production and should be replaced with SCAN.
| try { | ||
| const response = await fetch("https://openrouter.ai/api/v1/chat/completions", { | ||
| method: "POST", | ||
| headers: { | ||
| "Content-Type": "application/json", | ||
| "Authorization": `Bearer ${apiKey}`, | ||
| "HTTP-Referer": process.env.CONVEX_SITE_URL || "https://osschat.io", | ||
| "X-Title": "OSSChat", | ||
| }, | ||
| body: JSON.stringify({ | ||
| model: job.model, | ||
| messages: job.messages.map((m: { role: string; content: string }) => ({ | ||
| role: m.role, | ||
| content: m.content, | ||
| })), | ||
| stream: true, | ||
| }), | ||
| }); |
There was a problem hiding this comment.
style: no timeout for OpenRouter API call - if OpenRouter hangs, the Convex action could run indefinitely, consuming resources
Convex actions have a maximum execution time, but it's better to explicitly handle timeouts:
| try { | |
| const response = await fetch("https://openrouter.ai/api/v1/chat/completions", { | |
| method: "POST", | |
| headers: { | |
| "Content-Type": "application/json", | |
| "Authorization": `Bearer ${apiKey}`, | |
| "HTTP-Referer": process.env.CONVEX_SITE_URL || "https://osschat.io", | |
| "X-Title": "OSSChat", | |
| }, | |
| body: JSON.stringify({ | |
| model: job.model, | |
| messages: job.messages.map((m: { role: string; content: string }) => ({ | |
| role: m.role, | |
| content: m.content, | |
| })), | |
| stream: true, | |
| }), | |
| }); | |
| const timeoutMs = 5 * 60 * 1000; // 5 minutes | |
| const controller = new AbortController(); | |
| const timeoutId = setTimeout(() => controller.abort(), timeoutMs); | |
| const response = await fetch("https://openrouter.ai/api/v1/chat/completions", { | |
| method: "POST", | |
| headers: { | |
| "Content-Type": "application/json", | |
| "Authorization": `Bearer ${apiKey}`, | |
| "HTTP-Referer": process.env.CONVEX_SITE_URL || "https://osschat.io", | |
| "X-Title": "OSSChat", | |
| }, | |
| body: JSON.stringify({ | |
| model: job.model, | |
| messages: job.messages.map((m: { role: string; content: string }) => ({ | |
| role: m.role, | |
| content: m.content, | |
| })), | |
| stream: true, | |
| }), | |
| signal: controller.signal, | |
| }); | |
| clearTimeout(timeoutId); |
Confidence: 3/5 - Adds defensive error handling for long-running requests.
Note: If this suggestion doesn't match your team's coding style, reply to this and let me know. I'll remember it for next time!
| const existingMessage = await ctx.db | ||
| .query("messages") | ||
| .withIndex("by_client_id", (q) => | ||
| q.eq("chatId", job.chatId).eq("clientMessageId", job.messageId) | ||
| ) | ||
| .first(); | ||
|
|
||
| if (!existingMessage) { | ||
| await ctx.db.insert("messages", { | ||
| chatId: job.chatId, | ||
| clientMessageId: job.messageId, | ||
| role: "assistant", | ||
| content: args.content, | ||
| reasoning: args.reasoning, | ||
| createdAt: Date.now(), | ||
| }); | ||
| } else { | ||
| await ctx.db.patch(existingMessage._id, { | ||
| content: args.content, | ||
| reasoning: args.reasoning, | ||
| }); | ||
| } |
There was a problem hiding this comment.
style: duplicate message insertion logic - the completion handler either inserts a new message or updates an existing one, but this duplicate check is unnecessary if clientMessageId is unique
The frontend already creates assistantMsgId with crypto.randomUUID() which should be unique. Consider removing the query and always inserting:
| const existingMessage = await ctx.db | |
| .query("messages") | |
| .withIndex("by_client_id", (q) => | |
| q.eq("chatId", job.chatId).eq("clientMessageId", job.messageId) | |
| ) | |
| .first(); | |
| if (!existingMessage) { | |
| await ctx.db.insert("messages", { | |
| chatId: job.chatId, | |
| clientMessageId: job.messageId, | |
| role: "assistant", | |
| content: args.content, | |
| reasoning: args.reasoning, | |
| createdAt: Date.now(), | |
| }); | |
| } else { | |
| await ctx.db.patch(existingMessage._id, { | |
| content: args.content, | |
| reasoning: args.reasoning, | |
| }); | |
| } | |
| await ctx.db.insert("messages", { | |
| chatId: job.chatId, | |
| clientMessageId: job.messageId, | |
| role: "assistant", | |
| content: args.content, | |
| reasoning: args.reasoning, | |
| createdAt: Date.now(), | |
| }); |
Confidence: 2/5 - This might be defensive programming for edge cases, but adds unnecessary DB queries.
Note: If this suggestion doesn't match your team's coding style, reply to this and let me know. I'll remember it for next time!
- Add error logging for SSE parse failures - Add race condition check for concurrent streams - Add 5-minute timeout for OpenRouter API calls - Replace Redis KEYS with SCAN for better performance
1 participant
Summary
streamJobstable to track streaming jobs in Convexuse-persistent-chathook to use Convex-based streaming instead of HTTPChanges
Backend (Convex)
streamJobstable with indexes for chat, user, and statusstartStreammutation - creates job and schedules actionexecuteStreamaction - calls OpenRouter API independently of client connectionupdateStreamContent- batches content updates every 5 tokenscompleteStream/failStream- handles completion and errorsgetActiveStreamJobquery - enables resuming on page reloadFrontend
How it works
startBackgroundStreammutation calledgetActiveStreamJobqueryTesting