diff --git a/.changeset/chat-message-status.md b/.changeset/chat-message-status.md new file mode 100644 index 000000000..e8632d1f7 --- /dev/null +++ b/.changeset/chat-message-status.md @@ -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. diff --git a/packages/core/src/components/chat.test.ts b/packages/core/src/components/chat.test.ts index e3519dac3..ec8876694 100644 --- a/packages/core/src/components/chat.test.ts +++ b/packages/core/src/components/chat.test.ts @@ -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[] = []; @@ -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); @@ -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); @@ -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(); + 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(); + 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(); + 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(); + }); +}); diff --git a/packages/core/src/components/chat.ts b/packages/core/src/components/chat.ts index b177c9285..17aae9a19 100644 --- a/packages/core/src/components/chat.ts +++ b/packages/core/src/components/chat.ts @@ -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, @@ -69,8 +69,24 @@ export type ChatOptions = { type ChatTopicState = { messageSubject: Subject; onDestroyObservable: Subject; + /** 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}`; +} + const topicSubjectMap: WeakMap> = new WeakMap(); const streamIdToAttachments = new Map< string /* stream id */, @@ -112,8 +128,9 @@ export function setupChat(room: Room, options?: ChatOptions) { const topicState = topicMap.get(topic) ?? { messageSubject: new Subject(), onDestroyObservable: new Subject(), + unsentMessages: new Map(), }; - const { messageSubject, onDestroyObservable } = topicState; + const { messageSubject, onDestroyObservable, unsentMessages } = topicState; topicMap.set(topic, topicState); topicSubjectMap.set(room, topicMap); @@ -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]; } @@ -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; + } + + 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(), @@ -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, }; } diff --git a/packages/core/src/messages/types.ts b/packages/core/src/messages/types.ts index 3c05e872b..8c89ad3b4 100644 --- a/packages/core/src/messages/types.ts +++ b/packages/core/src/messages/types.ts @@ -15,12 +15,21 @@ type ReceivedMessageWithType; } & 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; + /** Set only on messages sent from this client. */ + status?: ChatMessageStatus; } >; diff --git a/packages/react/etc/components-react.api.md b/packages/react/etc/components-react.api.md index 406e2113b..0c02bb58b 100644 --- a/packages/react/etc/components-react.api.md +++ b/packages/react/etc/components-react.api.md @@ -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'; @@ -227,6 +228,8 @@ export const ChatIcon: (props: SVGProps) => React_2.JSX.Element; export { ChatMessage } +export { ChatMessageStatus } + // @public (undocumented) export interface ChatProps extends React_2.HTMLAttributes, ChatOptions { // (undocumented) @@ -867,6 +870,7 @@ export function useChat(options?: ChatOptions & { room?: Room; }): { send: (message: string, options?: SendTextOptions) => Promise; + resend: (messageId: string) => Promise; chatMessages: ReceivedChatMessage[]; isSending: boolean; }; @@ -1308,6 +1312,7 @@ export type UseSessionMessagesReturn = { messages: Array; isSending: boolean; send: (message: string, options?: SendTextOptions) => Promise; + resend: (messageId: string) => Promise; internal: { emitter: default_2; }; diff --git a/packages/react/src/hooks/useChat.test.ts b/packages/react/src/hooks/useChat.test.ts index 70bd1c438..953e1ab78 100644 --- a/packages/react/src/hooks/useChat.test.ts +++ b/packages/react/src/hooks/useChat.test.ts @@ -1,7 +1,7 @@ -import { DataPacket_Kind } from '@livekit/protocol'; +import { DataPacket_Kind, Encryption_Type } from '@livekit/protocol'; import { act, cleanup, renderHook } from '@testing-library/react'; import { Room, RoomEvent } from 'livekit-client'; -import { afterEach, describe, expect, it } from 'vitest'; +import { afterEach, describe, expect, it, vi } from 'vitest'; import { useChat } from './useChat'; describe('useChat options', () => { @@ -43,6 +43,39 @@ describe('useChat options', () => { }, ); + it('shows a sent message right away and resends it after a failure', async () => { + const room = new Room(); + rooms.push(room); + vi.spyOn(room.localParticipant, 'publishData').mockResolvedValue(); + const sendText = vi + .spyOn(room.localParticipant, 'sendText') + .mockRejectedValueOnce(new Error('text send failed')); + const { result } = renderHook(() => useChat({ room })); + + let sending!: Promise; + act(() => { + sending = result.current.send('hello').catch(() => {}); + }); + expect(result.current.chatMessages).toEqual([ + expect.objectContaining({ message: 'hello', status: 'sending' }), + ]); + + await act(() => sending); + expect(result.current.chatMessages[0].status).toBe('failed'); + + sendText.mockResolvedValueOnce({ + id: 'stream', + topic: 'lk.chat', + timestamp: 1, + mimeType: 'text/plain', + encryptionType: Encryption_Type.NONE, + }); + await act(() => result.current.resend(result.current.chatMessages[0].id)); + expect(result.current.chatMessages).toEqual([ + expect.objectContaining({ message: 'hello', status: 'sent' }), + ]); + }); + it('switches rooms and resets history when the room option changes', () => { const firstRoom = new Room(); const secondRoom = new Room(); diff --git a/packages/react/src/hooks/useChat.ts b/packages/react/src/hooks/useChat.ts index f706b1d40..1dd2de023 100644 --- a/packages/react/src/hooks/useChat.ts +++ b/packages/react/src/hooks/useChat.ts @@ -15,7 +15,9 @@ import { useConnectionState } from './useConnectionStatus'; * * @returns An object containing: * - `chatMessages` - Array of received chat messages - * - `send` - Function to send a new message + * - `send` - Function to send a new message. It appears in `chatMessages` right away with a + * `status` of `sending`, then `sent` or `failed`. + * - `resend` - Function to send a failed message again, by id * - `isSending` - Boolean indicating if a message is currently being sent * * @example @@ -54,5 +56,5 @@ export function useChat(options?: ChatOptions & { room?: Room }) { const isSending = useObservableState(setup.isSendingObservable, false); const chatMessages = useObservableState(setup.messageObservable, []); - return { send: setup.send, chatMessages, isSending }; + return { send: setup.send, resend: setup.resend, chatMessages, isSending }; } diff --git a/packages/react/src/hooks/useSessionMessages.ts b/packages/react/src/hooks/useSessionMessages.ts index 46d550f10..87f333d1d 100644 --- a/packages/react/src/hooks/useSessionMessages.ts +++ b/packages/react/src/hooks/useSessionMessages.ts @@ -25,6 +25,9 @@ export type UseSessionMessagesReturn = { send: (message: string, options?: SendTextOptions) => Promise; + /** Sends a failed message again, keeping its place in `messages`. */ + resend: (messageId: string) => Promise; + internal: { emitter: TypedEventEmitter; }; @@ -153,9 +156,10 @@ export function useSessionMessages(session?: UseSessionReturn): UseSessionMessag () => ({ messages: sortedReceivedMessages, send: chat.send, + resend: chat.resend, isSending: chat.isSending, internal: { emitter }, }), - [sortedReceivedMessages, chat.send, chat.isSending], + [sortedReceivedMessages, chat.send, chat.resend, chat.isSending], ); } diff --git a/packages/react/src/index.ts b/packages/react/src/index.ts index 543c94804..f06c1bd67 100644 --- a/packages/react/src/index.ts +++ b/packages/react/src/index.ts @@ -14,6 +14,7 @@ export * from './assets/images'; export { setLogLevel, setLogExtension, isTrackReference } from '@livekit/components-core'; export type { ChatMessage, + ChatMessageStatus, ReceivedMessage, ReceivedChatMessage, ReceivedUserTranscriptionMessage,