feat(realtime): sync message edits and deletes instantly
This commit is contained in:
@@ -77,7 +77,9 @@ async def edit_message(
|
|||||||
db: AsyncSession = Depends(get_db),
|
db: AsyncSession = Depends(get_db),
|
||||||
current_user: User = Depends(get_current_user),
|
current_user: User = Depends(get_current_user),
|
||||||
) -> MessageRead:
|
) -> MessageRead:
|
||||||
return await update_message(db, message_id=message_id, user_id=current_user.id, payload=payload)
|
message = await update_message(db, message_id=message_id, user_id=current_user.id, payload=payload)
|
||||||
|
await realtime_gateway.publish_message_updated(message=message)
|
||||||
|
return message
|
||||||
|
|
||||||
|
|
||||||
@router.delete("/{message_id}", status_code=status.HTTP_204_NO_CONTENT)
|
@router.delete("/{message_id}", status_code=status.HTTP_204_NO_CONTENT)
|
||||||
@@ -91,6 +93,7 @@ async def remove_message(
|
|||||||
message = await get_message_by_id(db, message_id)
|
message = await get_message_by_id(db, message_id)
|
||||||
await delete_message_for_all(db, message_id=message_id, user_id=current_user.id)
|
await delete_message_for_all(db, message_id=message_id, user_id=current_user.id)
|
||||||
if message:
|
if message:
|
||||||
|
await realtime_gateway.publish_message_deleted(chat_id=message.chat_id, message_id=message_id)
|
||||||
await realtime_gateway.publish_chat_updated(chat_id=message.chat_id)
|
await realtime_gateway.publish_chat_updated(chat_id=message.chat_id)
|
||||||
return
|
return
|
||||||
await delete_message(db, message_id=message_id, user_id=current_user.id)
|
await delete_message(db, message_id=message_id, user_id=current_user.id)
|
||||||
|
|||||||
@@ -11,6 +11,8 @@ RealtimeEventName = Literal[
|
|||||||
"disconnect",
|
"disconnect",
|
||||||
"send_message",
|
"send_message",
|
||||||
"receive_message",
|
"receive_message",
|
||||||
|
"message_updated",
|
||||||
|
"message_deleted",
|
||||||
"typing_start",
|
"typing_start",
|
||||||
"typing_stop",
|
"typing_stop",
|
||||||
"message_read",
|
"message_read",
|
||||||
|
|||||||
@@ -208,6 +208,27 @@ class RealtimeGateway:
|
|||||||
},
|
},
|
||||||
)
|
)
|
||||||
|
|
||||||
|
async def publish_message_updated(self, *, message) -> None:
|
||||||
|
message_data = MessageRead.model_validate(message).model_dump(mode="json")
|
||||||
|
await self._publish_chat_event(
|
||||||
|
message.chat_id,
|
||||||
|
event="message_updated",
|
||||||
|
payload={
|
||||||
|
"chat_id": message.chat_id,
|
||||||
|
"message": message_data,
|
||||||
|
},
|
||||||
|
)
|
||||||
|
|
||||||
|
async def publish_message_deleted(self, *, chat_id: int, message_id: int) -> None:
|
||||||
|
await self._publish_chat_event(
|
||||||
|
chat_id,
|
||||||
|
event="message_deleted",
|
||||||
|
payload={
|
||||||
|
"chat_id": chat_id,
|
||||||
|
"message_id": message_id,
|
||||||
|
},
|
||||||
|
)
|
||||||
|
|
||||||
async def _send_user_event(self, user_id: int, event: OutgoingRealtimeEvent) -> None:
|
async def _send_user_event(self, user_id: int, event: OutgoingRealtimeEvent) -> None:
|
||||||
user_connections = self._connections.get(user_id, {})
|
user_connections = self._connections.get(user_id, {})
|
||||||
if not user_connections:
|
if not user_connections:
|
||||||
|
|||||||
@@ -112,6 +112,24 @@ export function useRealtime() {
|
|||||||
scheduleReloadChats();
|
scheduleReloadChats();
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
if (event.event === "message_updated") {
|
||||||
|
const chatId = Number(event.payload.chat_id);
|
||||||
|
const message = event.payload.message as Message;
|
||||||
|
if (!Number.isFinite(chatId) || !message?.id) {
|
||||||
|
return;
|
||||||
|
}
|
||||||
|
chatStore.upsertMessage(chatId, message);
|
||||||
|
scheduleReloadChats();
|
||||||
|
}
|
||||||
|
if (event.event === "message_deleted") {
|
||||||
|
const chatId = Number(event.payload.chat_id);
|
||||||
|
const messageId = Number(event.payload.message_id);
|
||||||
|
if (!Number.isFinite(chatId) || !Number.isFinite(messageId)) {
|
||||||
|
return;
|
||||||
|
}
|
||||||
|
chatStore.removeMessage(chatId, messageId);
|
||||||
|
scheduleReloadChats();
|
||||||
|
}
|
||||||
if (event.event === "chat_updated") {
|
if (event.event === "chat_updated") {
|
||||||
const chatId = Number(event.payload.chat_id);
|
const chatId = Number(event.payload.chat_id);
|
||||||
if (Number.isFinite(chatId)) {
|
if (Number.isFinite(chatId)) {
|
||||||
|
|||||||
@@ -85,6 +85,7 @@ interface ChatState {
|
|||||||
status: DeliveryStatus,
|
status: DeliveryStatus,
|
||||||
senderId: number
|
senderId: number
|
||||||
) => void;
|
) => void;
|
||||||
|
upsertMessage: (chatId: number, message: Message) => void;
|
||||||
removeMessage: (chatId: number, messageId: number) => void;
|
removeMessage: (chatId: number, messageId: number) => void;
|
||||||
restoreMessages: (chatId: number, messages: Message[]) => void;
|
restoreMessages: (chatId: number, messages: Message[]) => void;
|
||||||
clearChatMessages: (chatId: number) => void;
|
clearChatMessages: (chatId: number) => void;
|
||||||
@@ -299,6 +300,30 @@ export const useChatStore = create<ChatState>((set, get) => ({
|
|||||||
messagesByChat: { ...state.messagesByChat, [chatId]: next }
|
messagesByChat: { ...state.messagesByChat, [chatId]: next }
|
||||||
}));
|
}));
|
||||||
},
|
},
|
||||||
|
upsertMessage: (chatId, message) => {
|
||||||
|
const old = get().messagesByChat[chatId] ?? [];
|
||||||
|
if (!old.length) {
|
||||||
|
set((state) => ({
|
||||||
|
messagesByChat: { ...state.messagesByChat, [chatId]: [message] }
|
||||||
|
}));
|
||||||
|
return;
|
||||||
|
}
|
||||||
|
const idx = old.findIndex((m) => m.id === message.id);
|
||||||
|
if (idx === -1) {
|
||||||
|
const next = [...old, message].sort((a, b) => a.id - b.id);
|
||||||
|
set((state) => ({
|
||||||
|
messagesByChat: { ...state.messagesByChat, [chatId]: next }
|
||||||
|
}));
|
||||||
|
return;
|
||||||
|
}
|
||||||
|
const next = [...old];
|
||||||
|
const existing = next[idx];
|
||||||
|
const deliveryStatus = mergeDeliveryStatus(message.delivery_status, existing.delivery_status);
|
||||||
|
next[idx] = deliveryStatus ? { ...message, delivery_status: deliveryStatus } : message;
|
||||||
|
set((state) => ({
|
||||||
|
messagesByChat: { ...state.messagesByChat, [chatId]: next }
|
||||||
|
}));
|
||||||
|
},
|
||||||
removeMessage: (chatId, messageId) => {
|
removeMessage: (chatId, messageId) => {
|
||||||
const old = get().messagesByChat[chatId] ?? [];
|
const old = get().messagesByChat[chatId] ?? [];
|
||||||
set((state) => ({
|
set((state) => ({
|
||||||
|
|||||||
Reference in New Issue
Block a user