66 lines
2.8 KiB
TypeScript
66 lines
2.8 KiB
TypeScript
import { appendTextDelta } from '../../utils/topic-sync';
|
|
|
|
interface LivePart {
|
|
messageId: string;
|
|
part: Record<string, any> & { id: string; content: string };
|
|
}
|
|
|
|
/** Covers the interval between an emitted delta and its throttled DB write. */
|
|
export const createLiveTopicText = () => {
|
|
const topics = new Map<string, Map<string, LivePart>>();
|
|
|
|
const removeMessage = (topicId: string, messageId: string) => {
|
|
const parts = topics.get(topicId);
|
|
if (!parts) return;
|
|
for (const [id, entry] of parts) {
|
|
if (entry.messageId === messageId) parts.delete(id);
|
|
}
|
|
if (parts.size === 0) topics.delete(topicId);
|
|
};
|
|
|
|
return {
|
|
update(topicId: string, event: { type: string; payload: any }) {
|
|
const { type, payload } = event;
|
|
if (type === 'text-start') {
|
|
let parts = topics.get(topicId);
|
|
if (!parts) topics.set(topicId, parts = new Map());
|
|
parts.set(payload.part.id, {
|
|
messageId: payload.messageId,
|
|
part: { ...payload.part, toolCall: null, content: payload.part.content ?? '' },
|
|
});
|
|
} else if (type === 'text-delta') {
|
|
const entry = topics.get(topicId)?.get(payload.partId);
|
|
if (entry) {
|
|
entry.part.content = appendTextDelta(entry.part.content, payload.content, payload.offset);
|
|
entry.part.lastUpdatedAt = payload.lastUpdatedAt;
|
|
}
|
|
} else if (type === 'text-end' || type === 'text-delete') {
|
|
const parts = topics.get(topicId);
|
|
parts?.delete(payload.partId);
|
|
if (parts?.size === 0) topics.delete(topicId);
|
|
} else if (type === 'MESSAGE_DELETED') {
|
|
removeMessage(topicId, payload.id);
|
|
}
|
|
},
|
|
apply(topicId: string, messages: { id: string; parts: any[] }[]) {
|
|
const parts = topics.get(topicId);
|
|
if (!parts) return;
|
|
const byId = new Map(messages.map(message => [message.id, message]));
|
|
for (const entry of parts.values()) {
|
|
const message = byId.get(entry.messageId);
|
|
if (!message) continue;
|
|
const existing = message.parts.find(part => part.id === entry.part.id);
|
|
if (existing) {
|
|
if (!existing.finished && (existing.content?.length ?? 0) <= entry.part.content.length) {
|
|
existing.content = entry.part.content;
|
|
existing.lastUpdatedAt = entry.part.lastUpdatedAt;
|
|
}
|
|
} else {
|
|
message.parts.push({ ...entry.part });
|
|
}
|
|
}
|
|
},
|
|
removeMessage,
|
|
};
|
|
};
|