96 lines
2.9 KiB
TypeScript
96 lines
2.9 KiB
TypeScript
import { db } from "~~/server/lib/db";
|
|
import { topicEvents } from "~~/server/utils/events";
|
|
|
|
export default defineEventHandler(async (event) => {
|
|
await protectRoute(event);
|
|
|
|
const topicId = getRouterParam(event, 'topicId')!;
|
|
const isSSE = getHeader(event, 'accept')?.includes('text/event-stream') ?? false;
|
|
|
|
const userId = event.context.user!.id as string;
|
|
|
|
const topic = await db.query.topics.findFirst({
|
|
where: {
|
|
id: topicId,
|
|
userId,
|
|
},
|
|
with: {
|
|
messages: {
|
|
orderBy: {
|
|
createdAt: 'asc',
|
|
},
|
|
with: {
|
|
parts: {
|
|
orderBy: {
|
|
createdAt: 'asc',
|
|
},
|
|
with: {
|
|
toolCall: true,
|
|
}
|
|
},
|
|
attachments: {
|
|
with: {
|
|
file: true,
|
|
}
|
|
},
|
|
generation: true,
|
|
}
|
|
},
|
|
},
|
|
});
|
|
|
|
if (!topic) {
|
|
console.log("topic not found");
|
|
throw createError({
|
|
statusCode: 404,
|
|
statusMessage: 'Not Found',
|
|
});
|
|
}
|
|
|
|
if (isSSE === false) {
|
|
return topic;
|
|
}
|
|
|
|
setHeader(event, 'Content-Type', 'text/event-stream');
|
|
setHeader(event, 'Cache-Control', 'no-cache');
|
|
setHeader(event, 'Connection', 'keep-alive');
|
|
|
|
const { lastUpdate, count } = getQuery(event);
|
|
const serverLastUpdate = topic.messages.at(-1)?.updatedAt;
|
|
const serverCount = topic.messages.length;
|
|
|
|
// TODO: there is potentially a race condition here where the client could
|
|
// connect at milisecond 87, and the last database write was at 0 (there
|
|
// is a write every 100ms for some operations), if there were tokens sent
|
|
// at 20 40 and 60, those tokens are lost to the client.
|
|
let streamController: ReadableStreamDefaultController;
|
|
let interval: NodeJS.Timeout;
|
|
const stream = new ReadableStream({
|
|
start(controller) {
|
|
streamController = controller;
|
|
controller.enqueue(`:connected\n\n`);
|
|
|
|
if ((lastUpdate && count) && (lastUpdate === serverLastUpdate?.toISOString() && count === serverCount)) {
|
|
controller.enqueue(':already_synced\n\n')
|
|
} else {
|
|
controller.enqueue(`data: ${JSON.stringify({
|
|
type: 'initial_state',
|
|
payload: topic
|
|
})}\n\n`);
|
|
}
|
|
|
|
interval = setInterval(() => {
|
|
controller.enqueue(`:heartbeat\n\n`);
|
|
}, 15000);
|
|
|
|
topicEvents.subscribe(topicId, streamController);
|
|
},
|
|
cancel() {
|
|
topicEvents.unsubscribe(topicId, streamController);
|
|
clearInterval(interval);
|
|
}
|
|
});
|
|
|
|
return sendStream(event, stream);
|
|
});
|