Skip to content

feat(core): Show sent chat messages right away, with new status attribute and resend function #1497

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

Draft
jeffdaley wants to merge 1 commit into
base: main
Choose a base branch
from
Draft

feat(core): Show sent chat messages right away, with new status attribute and resend function #1497

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
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
9 changes: 9 additions & 0 deletions .changeset/chat-message-status.md
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,9 @@
---
'@livekit/components-core': minor
'@livekit/components-react': minor
---

Show sent chat messages right away. `send` now adds the message as soon as it's called, with a
`status` of `sending`, then `sent` or `failed`, and keeps a failed message in place. Use the new
`resend` from `useChat` or `useSessionMessages` to send it again. Messages sent from this client now
keep a local id for their whole lifetime instead of taking the text stream id once sent.
138 changes: 134 additions & 4 deletions packages/core/src/components/chat.test.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,8 +1,8 @@
import { Encryption_Type } from '@livekit/protocol';
import { DataPacket_Kind, Encryption_Type } from '@livekit/protocol';
import { Room, RoomEvent, type TextStreamInfo } from 'livekit-client';
import { afterEach, describe, expect, it, vi } from 'vitest';
import { Future } from '../helper/future';
import { setupChat } from './chat';
import { setupChat, type ReceivedChatMessage } from './chat';

describe('chat sending state', () => {
const rooms: Room[] = [];
Expand Down Expand Up @@ -42,7 +42,7 @@ describe('chat sending state', () => {
expect(states.at(-1)).toBe(true);
if (outcome === 'resolve') {
firstText.resolve?.({ ...streamInfo, id: 'first' });
await expect(first).resolves.toMatchObject({ id: 'first', message: 'first' });
await expect(first).resolves.toMatchObject({ message: 'first', status: 'sent' });
} else {
const rejected = expect(first).rejects.toBe(error);
firstText.reject?.(error);
Expand All @@ -55,7 +55,7 @@ describe('chat sending state', () => {
lateSubscription.unsubscribe();

secondText.resolve?.({ ...streamInfo, id: 'second' });
await expect(second).resolves.toMatchObject({ id: 'second', message: 'second' });
await expect(second).resolves.toMatchObject({ message: 'second', status: 'sent' });
expect(states.at(-1)).toBe(false);
} finally {
firstText.resolve?.(streamInfo);
Expand Down Expand Up @@ -106,3 +106,133 @@ describe('chat sending state', () => {
},
);
});

describe('chat message status', () => {
const rooms: Room[] = [];
const streamInfo: TextStreamInfo = {
id: 'stream',
topic: 'lk.chat',
timestamp: 1,
mimeType: 'text/plain',
encryptionType: Encryption_Type.NONE,
};

afterEach(() => {
rooms.forEach((room) => room.emit(RoomEvent.Disconnected));
rooms.length = 0;
vi.restoreAllMocks();
});

function setup() {
const room = new Room();
rooms.push(room);
vi.spyOn(room.localParticipant, 'publishData').mockResolvedValue();
const sendText = vi.spyOn(room.localParticipant, 'sendText');
const chat = setupChat(room);
let messages: ReceivedChatMessage[] = [];
const subscription = chat.messageObservable.subscribe((value) => (messages = value));
return { room, sendText, chat, getMessages: () => messages, subscription };
}

it('adds a sent message right away as sending, then marks it sent', async () => {
const { sendText, chat, getMessages, subscription } = setup();
const text = new Future<TextStreamInfo, Error>();
sendText.mockReturnValueOnce(text.promise);

const sending = chat.send('hello');
expect(getMessages()).toEqual([
expect.objectContaining({ message: 'hello', status: 'sending' }),
]);
const { id } = getMessages()[0];

text.resolve?.(streamInfo);
await expect(sending).resolves.toMatchObject({ id, status: 'sent' });
expect(getMessages()).toEqual([expect.objectContaining({ id, status: 'sent' })]);
expect(getMessages()[0].editTimestamp).toBeUndefined();
subscription.unsubscribe();
});

it('marks a message failed and keeps it when sending fails', async () => {
const { sendText, chat, getMessages, subscription } = setup();
const error = new Error('text send failed');
sendText.mockRejectedValueOnce(error);

await expect(chat.send('hello')).rejects.toBe(error);

expect(getMessages()).toEqual([
expect.objectContaining({ message: 'hello', status: 'failed' }),
]);
subscription.unsubscribe();
});

it('resends a failed message in place', async () => {
const { sendText, chat, getMessages, subscription } = setup();
sendText.mockRejectedValueOnce(new Error('text send failed'));
await chat.send('first').catch(() => {});
sendText.mockResolvedValue(streamInfo);
await chat.send('second');
const failed = getMessages()[0];

const text = new Future<TextStreamInfo, Error>();
sendText.mockReturnValueOnce(text.promise);
const resending = chat.resend(failed.id);
expect(getMessages()[0]).toMatchObject({ id: failed.id, status: 'sending' });

text.resolve?.(streamInfo);
await resending;
expect(getMessages().map(({ message, status }) => ({ message, status }))).toEqual([
{ message: 'first', status: 'sent' },
{ message: 'second', status: 'sent' },
]);
expect(getMessages()[0].timestamp).toBe(failed.timestamp);
expect(sendText).toHaveBeenLastCalledWith(
'first',
expect.objectContaining({ topic: 'lk.chat' }),
);
subscription.unsubscribe();
});

it('rejects resending a message that is sending or already sent', async () => {
const { sendText, chat, getMessages, subscription } = setup();
const text = new Future<TextStreamInfo, Error>();
sendText.mockReturnValueOnce(text.promise);

const sending = chat.send('hello');
const { id } = getMessages()[0];
await expect(chat.resend(id)).rejects.toThrow('already sending');

text.resolve?.(streamInfo);
await sending;
await expect(chat.resend(id)).rejects.toThrow('No failed message');
expect(sendText).toHaveBeenCalledTimes(1);
subscription.unsubscribe();
});

it('shares failed messages across setups for the same room, so any can resend them', async () => {
const { room, sendText, chat, getMessages, subscription } = setup();
sendText.mockRejectedValueOnce(new Error('text send failed'));
await chat.send('hello').catch(() => {});

sendText.mockResolvedValue(streamInfo);
await setupChat(room).resend(getMessages()[0].id);

expect(getMessages()).toEqual([expect.objectContaining({ message: 'hello', status: 'sent' })]);
subscription.unsubscribe();
});

it('leaves status unset on messages from other participants', async () => {
const { room, getMessages, subscription } = setup();

room.emit(
RoomEvent.DataReceived,
new TextEncoder().encode(JSON.stringify({ id: 'remote', timestamp: 1, message: 'hi' })),
undefined,
DataPacket_Kind.RELIABLE,
'lk-chat-topic',
);

expect(getMessages()).toEqual([expect.objectContaining({ id: 'remote', message: 'hi' })]);
expect(getMessages()[0].status).toBeUndefined();
subscription.unsubscribe();
});
});
111 changes: 82 additions & 29 deletions packages/core/src/components/chat.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,5 +1,5 @@
/* eslint-disable camelcase */
import type { Room, SendTextOptions } from 'livekit-client';
import type { Room, SendTextOptions, TextStreamInfo } from 'livekit-client';
import { compareVersions, RoomEvent } from 'livekit-client';
import {
BehaviorSubject,
Expand Down Expand Up @@ -69,8 +69,24 @@ export type ChatOptions = {
type ChatTopicState = {
messageSubject: Subject<ReceivedChatMessage>;
onDestroyObservable: Subject<void>;
/** Messages sent from this client that haven't been sent successfully yet, kept for `resend`. */
unsentMessages: Map<
string,
{ message: ReceivedChatMessage; options: SendTextOptions; isSending: boolean }
>;
};

let localMessageCount = 0;

/**
* `sendText` only returns the stream id once sending finishes, so a message sent from this client
* gets its own id up front and keeps it, whether it's sending, sent, or failed.
*/
function createLocalMessageId() {
localMessageCount += 1;
return `local-${Date.now().toString(36)}-${localMessageCount}`;
}
Comment on lines +85 to +88

@1egoman 1egoman Nov 4, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

suggestion: Today, stream ids are uuids: https://github.com/livekit/client-sdk-js/blob/4721d094b2062b08b12093f7eb8f5597bbff235c/src/room/data-stream/outgoing/OutgoingDataStreamManager.ts#L84. If you made this a uuid then from the perspective of a downstream user, the api interface would appear exactly the same.


const topicSubjectMap: WeakMap<Room, Map<string, ChatTopicState>> = new WeakMap();
const streamIdToAttachments = new Map<
string /* stream id */,
Expand Down Expand Up @@ -112,8 +128,9 @@ export function setupChat(room: Room, options?: ChatOptions) {
const topicState = topicMap.get(topic) ?? {
messageSubject: new Subject<ReceivedChatMessage>(),
onDestroyObservable: new Subject<void>(),
unsentMessages: new Map(),
};
const { messageSubject, onDestroyObservable } = topicState;
const { messageSubject, onDestroyObservable, unsentMessages } = topicState;
topicMap.set(topic, topicState);
topicSubjectMap.set(room, topicMap);

Expand Down Expand Up @@ -254,11 +271,14 @@ export function setupChat(room: Room, options?: ChatOptions) {
const replaceIndex = acc.findIndex((msg) => msg.id === value.id);
if (replaceIndex > -1) {
const originalMsg = acc[replaceIndex];
acc[replaceIndex] = {
...value,
timestamp: originalMsg.timestamp,
editTimestamp: value.timestamp,
};
// A status change on a message sent from this client isn't an edit
acc[replaceIndex] = value.status
? value
: {
...value,
timestamp: originalMsg.timestamp,
editTimestamp: value.timestamp,
};
}
return [...acc];
}
Expand All @@ -271,37 +291,36 @@ export function setupChat(room: Room, options?: ChatOptions) {
let pendingSends = 0;
const finalMessageEncoder = options?.messageEncoder ?? encodeLegacyMsg;

const send = async (message: string, options?: SendTextOptions) => {
if (!options) {
options = {};
}
options.topic ??= topic;
/**
* Adds the message as `sending`, then marks it `sent` or `failed`. Rejects if sending fails,
* leaving the failed message in place for `resend`.
*/
const deliver = async (unsentMessage: ReceivedChatMessage, options: SendTextOptions) => {
const { id, message } = unsentMessage;
unsentMessages.set(id, { message: unsentMessage, options, isSending: true });
pendingSends += 1;
isSending$.next(true);
messageSubject.next({ ...unsentMessage, status: 'sending' });

try {
const info = await room.localParticipant.sendText(message, options);
let info: TextStreamInfo;
try {
info = await room.localParticipant.sendText(message, options);
} catch (error) {
unsentMessages.set(id, { message: unsentMessage, options, isSending: false });
messageSubject.next({ ...unsentMessage, status: 'failed' });
throw error;
}
Comment on lines +294 to +313

@1egoman 1egoman Nov 4, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

question: Just generally, I'm curious if you have any data around how often chat message sending fails. Even anecdotally is fine - I've yet to see one fail in local development. But it's possible I just have been treating it with kid gloves, and I haven't deeply tested cases like switching wifi networks / etc

Broader context: It's definitely possible for a data stream to fail but the bar for that has gotten a lot higher now with data streams v2. I'm actually going to be doing an EPD all hands presentation about this next month (and can send you all over the slides early if you'd be interested in learning more) but the TLDR is data streams are now over 50% smaller than they used to be, which means that on phones and poor bandwidth network connections it becomes substantially harder for them to drop (though not impossible).

I bring it up because while I think this is cool and lets a user implement a pretty nice UX, I also wonder if the complexity is really warranted if a data stream delivery failure is that much harder now. Something like interface pattern that the existing send function provides (wrap in a try/catch, show a toast or alert) could be good enough if it is really an abnormal case.

@jeffdaley jeffdaley Nov 6, 2026

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

No data on my end, just seemed like something to think through, even if rare (and even less rare with data streams v2 — nice). I'm fine keeping it out for now. If it becomes an issue we can always take care of it then.


unsentMessages.delete(id);
const receivedChatMsg: ReceivedChatMessage = { ...unsentMessage, status: 'sent' };
messageSubject.next(receivedChatMsg);

const legacyChatMsg: LegacyChatMessage = {
id: info.id,
timestamp: Date.now(),
timestamp: unsentMessage.timestamp,
message,
};

const chatMsg: ChatMessage = {
...legacyChatMsg,
attachedFiles: options.attachments,
};

const receivedChatMsg: ReceivedChatMessage = {
...chatMsg,
type: 'chatMessage',
from: room.localParticipant,
attributes: options.attributes,
};

messageSubject.next(receivedChatMsg);

const encodedLegacyMsg = finalMessageEncoder({
...legacyChatMsg,
ignoreLegacy: serverSupportsDataStreams(),
Expand All @@ -323,9 +342,43 @@ export function setupChat(room: Room, options?: ChatOptions) {
}
};

/** Sends a message. It appears in the messages right away with a `sending` status. */
const send = async (message: string, options?: SendTextOptions) => {
if (!options) {
options = {};
}
options.topic ??= topic;

return deliver(
{
id: createLocalMessageId(),
timestamp: Date.now(),
message,
attachedFiles: options.attachments,
type: 'chatMessage',
from: room.localParticipant,
attributes: options.attributes,
},
options,
);
};

/** Sends a failed message again, keeping its id and place in the messages. */
const resend = async (messageId: string) => {
const unsent = unsentMessages.get(messageId);
if (!unsent) {
throw new Error(`No failed message with id ${messageId} to resend`);
}
if (unsent.isSending) {
throw new Error(`Message ${messageId} is already sending`);
}
return deliver(unsent.message, unsent.options);
};

return {
messageObservable: messagesObservable,
isSendingObservable: isSending$,
send,
resend,
};
}
9 changes: 9 additions & 0 deletions packages/core/src/messages/types.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -15,12 +15,21 @@ type ReceivedMessageWithType<Type extends string, Metadata extends object = obje
attributes?: Record<string, string>;
} & Metadata;

/**
* Where a message sent from this client is in delivery. `sent` means the LiveKit server accepted
* it, not that other participants have read it.
* @public
*/
export type ChatMessageStatus = 'sending' | 'sent' | 'failed';

/** @private */
type ReceivedChatMessageWithRequiredType = ReceivedMessageWithType<
'chatMessage',
ChatMessage & {
from?: Participant;
attributes?: Record<string, string>;
/** Set only on messages sent from this client. */
status?: ChatMessageStatus;
}
>;
Comment on lines 26 to 34

@1egoman 1egoman Nov 4, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

suggestion: I don't necessarily like some messages having a field and some not having a different field without it being clear in what cases that field can be set.

What if you were to switch this to be some sort of discriminated union based on from? Maybe not exactly this but something conceptually like this:

Suggested change
type ReceivedChatMessageWithRequiredType = ReceivedMessageWithType<
'chatMessage',
ChatMessage & {
from?: Participant;
attributes?: Record<string, string>;
/** Set only on messages sent from this client. */
status?: ChatMessageStatus;
}
>;
type ReceivedChatMessageWithRequiredType = ReceivedMessageWithType<
'chatMessage',
ChatMessage & ({
from: Participant;
attributes?: Record<string, string>;
} | {
from?: never;
attributes?: Record<string, string>;
/** Set only on messages sent from this client. */
status: ChatMessageStatus;
})
>;

@jeffdaley jeffdaley Nov 6, 2026

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Agree the type should make it clear when status exists. One catch is that local messages do have from (LocalParticipant, which extends Participant), so keying the union on from wouldn't work. Maybe this?

type Message = ChatMessage & { from?: Participant } & (
  | { isLocal: true; status: 'sending' | 'sent' }
  | { isLocal?: false }
);

Something to settle in useThreadMessages.

@1egoman 1egoman Nov 8, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

so keying the union on from wouldn't work.

This was what I was trying to accomplish with from?: never; - this does work, here is a link if you want to play around with it. It's a little confusing though and I agree maybe an isLocal / some sort of enum of options could maybe be better.


Expand Down
5 changes: 5 additions & 0 deletions packages/react/etc/components-react.api.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -10,6 +10,7 @@ import { BaseE2EEManager } from 'livekit-client';
import { BaseKeyProvider } from 'livekit-client';
import { CaptureOptionsBySource } from '@livekit/components-core';
import { ChatMessage } from '@livekit/components-core';
import { ChatMessageStatus } from '@livekit/components-core';
import { ChatOptions } from '@livekit/components-core';
import { ConnectionQuality } from 'livekit-client';
import { ConnectionState as ConnectionState_2 } from 'livekit-client';
Expand Down Expand Up @@ -227,6 +228,8 @@ export const ChatIcon: (props: SVGProps<SVGSVGElement>) => React_2.JSX.Element;

export { ChatMessage }

export { ChatMessageStatus }

// @public (undocumented)
export interface ChatProps extends React_2.HTMLAttributes<HTMLDivElement>, ChatOptions {
// (undocumented)
Expand Down Expand Up @@ -867,6 +870,7 @@ export function useChat(options?: ChatOptions & {
room?: Room;
}): {
send: (message: string, options?: SendTextOptions) => Promise<ReceivedChatMessage>;
resend: (messageId: string) => Promise<ReceivedChatMessage>;
chatMessages: ReceivedChatMessage[];
isSending: boolean;
};
Expand Down Expand Up @@ -1308,6 +1312,7 @@ export type UseSessionMessagesReturn = {
messages: Array<ReceivedMessage>;
isSending: boolean;
send: (message: string, options?: SendTextOptions) => Promise<ReceivedChatMessage>;
resend: (messageId: string) => Promise<ReceivedChatMessage>;
internal: {
emitter: default_2<MessagesCallbacks>;
};
Expand Down
Loading
Loading