Skip to content

Fix/sse streaming events #203

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
Antherix wants to merge 5 commits into
base: main
Choose a base branch
from
Open

Fix/sse streaming events #203

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
5 commits
Select commit Hold shift + click to select a range
0594fc4
fix: prevent NaN token totals by coalescing null tokens to 0
Antherix Jun 4, 2026
d6e9631
feat: add delete-my-data endpoint and profile UI
Antherix Jun 5, 2026
45b9f02
fix: convert streaming to real SSE events (chunk/usage/done/error)
Antherix Jun 12, 2026
33b5827
Merge branch 'main' into fix/sse-streaming-events
Antherix Jun 12, 2026
03b5151
Merge branch 'main' into fix/sse-streaming-events
Antherix Jun 17, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
47 changes: 25 additions & 22 deletions backend/controllers/chatMessage.controller.js
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -28,16 +28,10 @@ if (MEM0_ENABLED) {
}
}

// Daily token budget tracked per user per UTC day.
const dailyBudgetKey = (userId) => `tokenBudget:${userId}:${new Date().toISOString().slice(0, 10)}`;

const startOfUtcDay = () => {
const now = new Date();
return new Date(Date.UTC(now.getUTCFullYear(), now.getUTCMonth(), now.getUTCDate()));
};

const secondsUntilUtcMidnight = () =>
Math.ceil((startOfUtcDay().getTime() + 86400000 - Date.now()) / 1000);
// SSE helper — writes a properly-framed SSE event
function writeSSE(res, event, data) {
res.write(`event: ${event}\ndata: ${JSON.stringify(data)}\n\n`);
}

const getAvailableModels = asyncHandler(async (req, res) => {
const apikeys = await prisma.apiKey.findMany({
Expand Down Expand Up @@ -210,11 +204,18 @@ if (!chat.chatSources[0].isVectorLess) {
limit: 20,
with_payload: true,
});

const [denseResults, keywordResults] = await Promise.all([denseTask, keywordTask]);

if (denseResults?.points?.length) {
allDensePoints.push(...denseResults.points);
treeindex.loadData(docTree.sourceData);
treeindex.loadTree(docTree.treeData);

relevantNodeIds = await treeindex.retrieveRelevantNodes(userPrompt);
if(relevantNodeIds.length == 0) {
res.setHeader("Content-Type", "text/event-stream");
res.setHeader("Cache-Control", "no-cache, no-transform");
res.setHeader("Connection", "keep-alive");
res.setHeader("X-Accel-Buffering", "no");
writeSSE(res, "error", { message: "No relevant sources found for this query" });
res.end();
return;
}

if (keywordResults?.points?.length) {
Expand Down Expand Up @@ -305,7 +306,7 @@ if (!chat.chatSources[0].isVectorLess) {
res.setHeader("Cache-Control", "no-cache, no-transform");
res.setHeader("Connection", "keep-alive");
res.setHeader("X-Accel-Buffering", "no");

let llmResponse = "";
let inputTokens = 0;
let outputTokens = 0;
Expand All @@ -317,17 +318,19 @@ if (!chat.chatSources[0].isVectorLess) {
if (chunk.usage) {
inputTokens = chunk.usage.prompt_tokens;
outputTokens = chunk.usage.completion_tokens;
writeSSE(res, "usage", { inputTokens, outputTokens });
}
if (content) {
llmResponse += content;
res.write(content);
writeSSE(res, "chunk", { content });
}
}
} catch (error) {
res.write(`\n\ndata: {"error": "Stream ended with error: ${error.message.replace(/\n/g, ' ')}"}\n\n`);
} finally {
res.end();
}
writeSSE(res, "done", {});
} catch (error) {
writeSSE(res, "error", { message: error.message || "Stream ended with error" });
} finally {
res.end();
}

if (llmResponse.trim()) {
if (MEM0_ENABLED && memory) {
Expand Down
163 changes: 72 additions & 91 deletions backend/controllers/user.controller.js
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -361,101 +361,82 @@ const resetPassword = asyncHandler(async (req, res) => {
});

const deleteMyData = asyncHandler(async (req, res) => {
const userId = req.user.id; // set by your auth middleware
const { confirm } = req.query;

// Acceptance criteria: require explicit confirmation flag
if (confirm !== "true") {
return res.status(400).json({
success: false,
message: "Pass confirm=true as a query param to confirm deletion.",
const userId = req.user.id; // set by your auth middleware
const { confirm } = req.query;

// Acceptance criteria: require explicit confirmation flag
if (confirm !== "true") {
return res.status(400).json({
success: false,
message: 'Pass confirm=true as a query param to confirm deletion.',
});
}

try {
await prisma.$transaction(async (tx) => {
// 1. Delete ChatMessageSources (deepest dependency first)
await tx.chatMessageSource.deleteMany({
where: { chatMessage: { chat: { userId } } },
});

// 2. Delete ChatMessages
await tx.chatMessage.deleteMany({
where: { chat: { userId } },
});

// 3. Delete UsageEvents (onDelete: SetNull means we must handle these)
await tx.usageEvent.deleteMany({ where: { userId } });

// 4. Delete ApiKeys
await tx.apiKey.deleteMany({ where: { userId } });

// 5. Delete ChatSources that belong ONLY to this user's chats
// Safe detach: only remove if no other user's chat references them
const userChatIds = (
await tx.chat.findMany({
where: { userId },
select: { id: true },
})
).map((c) => c.id);

// Find ChatSource IDs used by this user's chats
const userChatSources = await tx.chatSource.findMany({
where: { chatId: { in: userChatIds } },
select: { id: true, docsUrl: true },
});

// Only delete ChatSources not referenced by any other chat
for (const cs of userChatSources) {
const otherChatCount = await tx.chat.count({
where: {
chatSources: { some: { docsUrl: cs.docsUrl } },
userId: { not: userId },
},
});
}

try {
await prisma.$transaction(async (tx) => {
// 1. Delete ChatMessageSources (deepest dependency first)
await tx.chatMessageSource.deleteMany({
where: { chatMessage: { chat: { userId } } },
});

// 2. Delete ChatMessages
await tx.chatMessage.deleteMany({
where: { chat: { userId } },
});

// 3. Delete UsageEvents (onDelete: SetNull means we must handle these)
await tx.usageEvents.deleteMany({ where: { userId } });

// 4. Delete ApiKeys
await tx.apiKey.deleteMany({ where: { userId } });

// 5. Delete ChatSources that belong ONLY to this user's chats
// Safe detach: only remove if no other user's chat references them
const userChatIds = (
await tx.chat.findMany({
where: { userId },
select: { id: true },
})
).map((c) => c.id);

// Find ChatSources used by this user's chats
const userChatSources = await tx.chatSource.findMany({
where: {
chats: {
some: {
id: {
in: userChatIds,
},
},
},
},
select: {
id: true,
documentationUrl: true,
lastIndexedAt: true,
},
});

// Only delete ChatSources not referenced by any other user's chats
for (const cs of userChatSources) {
const otherChatCount = await tx.chat.count({
where: {
userId: { not: userId },
chatSources: {
some: {
documentationUrl: cs.documentationUrl,
},
},
},
});

if (otherChatCount === 0) {
await tx.chatSource.delete({
where: { id: cs.id },
});
}
}
if (otherChatCount === 0) {
await tx.chatSource.delete({ where: { id: cs.id } });
}
}

// 6. Delete Chats
await tx.chat.deleteMany({ where: { userId } });
});
// 6. Delete Chats
await tx.chat.deleteMany({ where: { userId } });
});

return res.status(200).json({
success: true,
message: "All your data has been deleted.",
});
} catch (error) {
// Idempotent: if already deleted, return success
if (error.code === "P2025") {
return res.status(200).json({
success: true,
message: "Data already deleted or not found.",
});
}
console.error("deleteMyData error:", error);
return res.status(500).json({ success: false, message: "Server error." });
return res.status(200).json({
success: true,
message: 'All your data has been deleted.',
});
} catch (error) {
// Idempotent: if already deleted, return success
if (error.code === 'P2025') {
return res.status(200).json({
success: true,
message: 'Data already deleted or not found.',
});
}
console.error('deleteMyData error:', error);
return res.status(500).json({ success: false, message: 'Server error.' });
}
});

export {
Expand Down
80 changes: 51 additions & 29 deletions src/lib/api.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -388,13 +388,15 @@ export const getMessageSources = (messageId: string) =>
}),
);


export const sendMessageStream = async (payload: {
userPrompt: string;
model: string;
provider: string;
chatId: string;
onChunk?: (chunk: string) => void;
signal?: AbortSignal;
onError?: (message: string) => void;
signal?: AbortSignal;
}) => {
const token = getAccessToken();
const headers = new Headers({ "Content-Type": "application/json" });
Expand All @@ -406,8 +408,13 @@ export const sendMessageStream = async (payload: {
method: "POST",
headers,
credentials: "include",
body: JSON.stringify(payload),
signal: payload.signal,
body: JSON.stringify({
userPrompt: payload.userPrompt,
model: payload.model,
provider: payload.provider,
chatId: payload.chatId,
}),
signal: payload.signal,
});

if (!response.ok || !response.body) {
Expand All @@ -418,25 +425,54 @@ export const sendMessageStream = async (payload: {
throw new Error(payload?.message || "Unable to send message");
}

// AFTER:
const reader = response.body.getReader();
const decoder = new TextDecoder();
let text = "";

let buffer = "";

while (true) {
const { done, value } = await reader.read();
if (done) break;
const chunk = decoder.decode(value, { stream: true });
if (!chunk) continue;
text += chunk;
payload.onChunk?.(chunk);
}

const tail = decoder.decode();
if (tail) {
text += tail;
payload.onChunk?.(tail);
buffer += decoder.decode(value, { stream: true });

// SSE frames are separated by double newlines
const frames = buffer.split("\n\n");
buffer = frames.pop() ?? ""; // last partial frame stays in buffer

for (const frame of frames) {
if (!frame.trim()) continue;

let eventName = "message";
let dataLine = "";

for (const line of frame.split("\n")) {
if (line.startsWith("event: ")) eventName = line.slice(7).trim();
else if (line.startsWith("data: ")) dataLine = line.slice(6).trim();
}

if (!dataLine) continue;

let parsed: Record<string, unknown>;
try {
parsed = JSON.parse(dataLine);
} catch {
continue;
}

if (eventName === "chunk" && typeof parsed.content === "string") {
text += parsed.content;
payload.onChunk?.(parsed.content);
} else if (eventName === "error") {
const msg = typeof parsed.message === "string" ? parsed.message : "Stream error";
payload.onError?.(msg);
payload.onChunk?.(msg); // surface it in the UI

}
// "usage" and "done" events — no UI action needed, silently consumed
}
}

invalidateChatMessages(payload.chatId);
return text;
};
Expand Down Expand Up @@ -755,17 +791,3 @@ export const forkSharedChat = (shareToken: string) =>

export const deleteMyData = () =>
apiRequest<{ message: string }>("/user/delete-my-data?confirm=true", { method: "DELETE" });

export const adminImpersonate = (userId: string) =>
apiRequest<{
accessToken: string;
user: {
id: string;
fullname?: string | null;
username?: string | null;
email?: string | null;
};
}>(`/admin/impersonate/${userId}`, { method: "POST" });

export const adminStopImpersonation = () =>
apiRequest<void>("/admin/stop-impersonation", { method: "POST" });
3 changes: 3 additions & 0 deletions src/pages/Profile.tsx
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,6 +1,9 @@
import { useEffect, useState } from "react";
import { useNavigate } from "react-router-dom";
import { Sidebar } from "../components/Sidebar";



import { User, Mail, LogOut, Save, Key, CheckCircle2, Trash2, Pencil, X } from "lucide-react";
import { logoutUser, forceSignOut } from "../lib/auth";
import { getUserProfile, deleteMyData } from "../lib/api";
Expand Down