feat: Implement real-time messaging functionality with dedicated services, types, a custom hook, and UI integration.
This commit is contained in:
pradeepkumar
462
src/hooks/useMessaging.ts
Normal file
462
src/hooks/useMessaging.ts
Normal file
@@ -0,0 +1,462 @@
|
||||
'use client';
|
||||
|
||||
import { useState, useEffect, useCallback, useRef } from 'react';
|
||||
import { useSession } from 'next-auth/react';
|
||||
import { socketService, messagesService } from '@/services';
|
||||
import type {
|
||||
Conversation,
|
||||
Message,
|
||||
CreateMessageDto,
|
||||
TypingIndicator,
|
||||
UserStatusChange,
|
||||
MessagesReadEvent,
|
||||
MessageType,
|
||||
MessageStatus,
|
||||
} from '@/types/messaging';
|
||||
|
||||
interface UseMessagingOptions {
|
||||
autoConnect?: boolean;
|
||||
}
|
||||
|
||||
interface UseMessagingReturn {
|
||||
// State
|
||||
conversations: Conversation[];
|
||||
currentConversation: Conversation | null;
|
||||
messages: Message[];
|
||||
isConnected: boolean;
|
||||
isLoading: boolean;
|
||||
isLoadingMessages: boolean;
|
||||
typingUsers: Set<string>;
|
||||
unreadCount: number;
|
||||
|
||||
// Actions
|
||||
connect: () => Promise<void>;
|
||||
disconnect: () => void;
|
||||
loadConversations: () => Promise<void>;
|
||||
selectConversation: (conversationId: string) => Promise<void>;
|
||||
startConversation: (agentProfileId: string) => Promise<Conversation>;
|
||||
sendMessage: (content: string, messageType?: MessageType) => Promise<void>;
|
||||
loadMoreMessages: () => Promise<void>;
|
||||
startTyping: () => void;
|
||||
stopTyping: () => void;
|
||||
markAsRead: () => Promise<void>;
|
||||
updateUserStatus: (userId: string, isOnline: boolean) => void;
|
||||
hasMoreMessages: boolean;
|
||||
}
|
||||
|
||||
export function useMessaging(options: UseMessagingOptions = {}): UseMessagingReturn {
|
||||
const { autoConnect = true } = options;
|
||||
const { data: session } = useSession();
|
||||
|
||||
// State
|
||||
const [conversations, setConversations] = useState<Conversation[]>([]);
|
||||
const [currentConversation, setCurrentConversation] = useState<Conversation | null>(null);
|
||||
const [messages, setMessages] = useState<Message[]>([]);
|
||||
const [isConnected, setIsConnected] = useState(false);
|
||||
const [isLoading, setIsLoading] = useState(false);
|
||||
const [isLoadingMessages, setIsLoadingMessages] = useState(false);
|
||||
const [typingUsers, setTypingUsers] = useState<Set<string>>(new Set());
|
||||
const [unreadCount, setUnreadCount] = useState(0);
|
||||
const [currentPage, setCurrentPage] = useState(1);
|
||||
const [hasMoreMessages, setHasMoreMessages] = useState(true);
|
||||
|
||||
// Refs
|
||||
const typingTimeoutRef = useRef<NodeJS.Timeout | null>(null);
|
||||
const currentConversationIdRef = useRef<string | null>(null);
|
||||
const isInitializedRef = useRef(false);
|
||||
|
||||
// Get access token from session
|
||||
const accessToken = session?.user?.accessToken;
|
||||
|
||||
// Connect to WebSocket
|
||||
const connect = useCallback(async () => {
|
||||
if (!accessToken) {
|
||||
console.warn('No access token available for socket connection');
|
||||
return;
|
||||
}
|
||||
|
||||
try {
|
||||
await socketService.connect(accessToken);
|
||||
} catch (error) {
|
||||
console.error('Failed to connect to socket:', error);
|
||||
}
|
||||
}, [accessToken]);
|
||||
|
||||
// Disconnect from WebSocket
|
||||
const disconnect = useCallback(() => {
|
||||
socketService.disconnect();
|
||||
}, []);
|
||||
|
||||
// Load conversations
|
||||
const loadConversations = useCallback(async () => {
|
||||
if (!accessToken) return;
|
||||
|
||||
setIsLoading(true);
|
||||
try {
|
||||
const data = await messagesService.getConversations();
|
||||
setConversations(data);
|
||||
} catch (error) {
|
||||
console.error('Failed to load conversations:', error);
|
||||
} finally {
|
||||
setIsLoading(false);
|
||||
}
|
||||
}, [accessToken]);
|
||||
|
||||
// Select a conversation and load its messages
|
||||
const selectConversation = useCallback(
|
||||
async (conversationId: string) => {
|
||||
if (currentConversationIdRef.current === conversationId) return;
|
||||
|
||||
// Leave previous conversation room
|
||||
if (currentConversationIdRef.current) {
|
||||
socketService.leaveConversation(currentConversationIdRef.current);
|
||||
}
|
||||
|
||||
currentConversationIdRef.current = conversationId;
|
||||
setCurrentPage(1);
|
||||
setHasMoreMessages(true);
|
||||
setMessages([]);
|
||||
setTypingUsers(new Set());
|
||||
setIsLoadingMessages(true);
|
||||
|
||||
try {
|
||||
// Get conversation details
|
||||
const conversation = await messagesService.getConversation(conversationId);
|
||||
setCurrentConversation(conversation);
|
||||
|
||||
// Join WebSocket room
|
||||
if (isConnected) {
|
||||
await socketService.joinConversation(conversationId);
|
||||
}
|
||||
|
||||
// Load messages
|
||||
const { messages: loadedMessages, pagination } =
|
||||
await messagesService.getMessages(conversationId, 1, 50);
|
||||
setMessages(loadedMessages);
|
||||
setHasMoreMessages(pagination.page < pagination.pages);
|
||||
|
||||
// Mark as read
|
||||
await messagesService.markAsRead(conversationId);
|
||||
|
||||
// Update unread count in conversations list
|
||||
setConversations((prev) =>
|
||||
prev.map((c) => (c.id === conversationId ? { ...c, unreadCount: 0 } : c))
|
||||
);
|
||||
} catch (error) {
|
||||
console.error('Failed to load conversation:', error);
|
||||
} finally {
|
||||
setIsLoadingMessages(false);
|
||||
}
|
||||
},
|
||||
[isConnected]
|
||||
);
|
||||
|
||||
// Start a new conversation with an agent
|
||||
const startConversation = useCallback(
|
||||
async (agentProfileId: string): Promise<Conversation> => {
|
||||
const conversation = await messagesService.startConversation(agentProfileId);
|
||||
|
||||
// Add to conversations list if not exists
|
||||
setConversations((prev) => {
|
||||
const exists = prev.some((c) => c.id === conversation.id);
|
||||
if (exists) return prev;
|
||||
return [conversation, ...prev];
|
||||
});
|
||||
|
||||
return conversation;
|
||||
},
|
||||
[]
|
||||
);
|
||||
|
||||
// Send a message
|
||||
const sendMessage = useCallback(
|
||||
async (content: string, messageType: MessageType = 'TEXT' as MessageType) => {
|
||||
if (!currentConversationIdRef.current) return;
|
||||
|
||||
const dto: CreateMessageDto = {
|
||||
content,
|
||||
messageType,
|
||||
};
|
||||
|
||||
// Stop typing indicator
|
||||
if (typingTimeoutRef.current) {
|
||||
clearTimeout(typingTimeoutRef.current);
|
||||
typingTimeoutRef.current = null;
|
||||
}
|
||||
socketService.stopTyping(currentConversationIdRef.current);
|
||||
|
||||
// Try WebSocket first, fallback to REST
|
||||
if (isConnected) {
|
||||
const result = await socketService.sendMessage(
|
||||
currentConversationIdRef.current,
|
||||
dto
|
||||
);
|
||||
if (result.success && result.message) {
|
||||
// Immediately update UI with the sent message (optimistic update)
|
||||
setMessages((prev) => {
|
||||
// Avoid duplicates (in case broadcast also arrives)
|
||||
if (prev.some((m) => m.id === result.message!.id)) return prev;
|
||||
return [...prev, result.message!];
|
||||
});
|
||||
} else if (!result.success) {
|
||||
// Fallback to REST
|
||||
const message = await messagesService.sendMessage(
|
||||
currentConversationIdRef.current,
|
||||
dto
|
||||
);
|
||||
setMessages((prev) => [...prev, message]);
|
||||
}
|
||||
} else {
|
||||
// REST fallback
|
||||
const message = await messagesService.sendMessage(
|
||||
currentConversationIdRef.current,
|
||||
dto
|
||||
);
|
||||
setMessages((prev) => [...prev, message]);
|
||||
}
|
||||
|
||||
// Update conversation in list
|
||||
const conversationId = currentConversationIdRef.current;
|
||||
setConversations((prev) =>
|
||||
prev.map((c) =>
|
||||
c.id === conversationId
|
||||
? {
|
||||
...c,
|
||||
lastMessageAt: new Date().toISOString(),
|
||||
lastMessageText:
|
||||
content.length > 255 ? content.substring(0, 252) + '...' : content,
|
||||
}
|
||||
: c
|
||||
)
|
||||
);
|
||||
},
|
||||
[isConnected]
|
||||
);
|
||||
|
||||
// Load more messages (pagination) with debouncing for smooth scrolling
|
||||
const loadMoreMessages = useCallback(async () => {
|
||||
if (!currentConversationIdRef.current || !hasMoreMessages || isLoadingMessages)
|
||||
return;
|
||||
|
||||
setIsLoadingMessages(true);
|
||||
try {
|
||||
const nextPage = currentPage + 1;
|
||||
const { messages: moreMessages, pagination } = await messagesService.getMessages(
|
||||
currentConversationIdRef.current,
|
||||
nextPage,
|
||||
50
|
||||
);
|
||||
|
||||
// Prepend older messages smoothly
|
||||
setMessages((prev) => [...moreMessages, ...prev]);
|
||||
setCurrentPage(nextPage);
|
||||
setHasMoreMessages(pagination.page < pagination.pages);
|
||||
} catch (error) {
|
||||
console.error('Failed to load more messages:', error);
|
||||
} finally {
|
||||
setIsLoadingMessages(false);
|
||||
}
|
||||
}, [currentPage, hasMoreMessages, isLoadingMessages]);
|
||||
|
||||
// Start typing indicator
|
||||
const startTyping = useCallback(() => {
|
||||
if (!currentConversationIdRef.current) return;
|
||||
|
||||
socketService.startTyping(currentConversationIdRef.current);
|
||||
|
||||
// Clear previous timeout
|
||||
if (typingTimeoutRef.current) {
|
||||
clearTimeout(typingTimeoutRef.current);
|
||||
}
|
||||
|
||||
// Auto-stop typing after 3 seconds
|
||||
typingTimeoutRef.current = setTimeout(() => {
|
||||
if (currentConversationIdRef.current) {
|
||||
socketService.stopTyping(currentConversationIdRef.current);
|
||||
}
|
||||
}, 3000);
|
||||
}, []);
|
||||
|
||||
// Stop typing indicator
|
||||
const stopTyping = useCallback(() => {
|
||||
if (!currentConversationIdRef.current) return;
|
||||
|
||||
if (typingTimeoutRef.current) {
|
||||
clearTimeout(typingTimeoutRef.current);
|
||||
typingTimeoutRef.current = null;
|
||||
}
|
||||
socketService.stopTyping(currentConversationIdRef.current);
|
||||
}, []);
|
||||
|
||||
// Mark messages as read
|
||||
const markAsRead = useCallback(async () => {
|
||||
if (!currentConversationIdRef.current) return;
|
||||
|
||||
if (isConnected) {
|
||||
await socketService.markAsRead(currentConversationIdRef.current);
|
||||
} else {
|
||||
await messagesService.markAsRead(currentConversationIdRef.current);
|
||||
}
|
||||
}, [isConnected]);
|
||||
|
||||
// Update user online status (called from status change events)
|
||||
const updateUserStatus = useCallback((userId: string, isOnline: boolean) => {
|
||||
setConversations((prev) =>
|
||||
prev.map((c) => {
|
||||
if (c.otherParty.userId === userId || c.otherParty.id === userId) {
|
||||
return {
|
||||
...c,
|
||||
otherParty: { ...c.otherParty, isOnline },
|
||||
};
|
||||
}
|
||||
return c;
|
||||
})
|
||||
);
|
||||
|
||||
// Update current conversation if applicable
|
||||
setCurrentConversation((prev) => {
|
||||
if (!prev) return prev;
|
||||
if (
|
||||
prev.otherParty.userId === userId ||
|
||||
prev.otherParty.id === userId
|
||||
) {
|
||||
return {
|
||||
...prev,
|
||||
otherParty: { ...prev.otherParty, isOnline },
|
||||
};
|
||||
}
|
||||
return prev;
|
||||
});
|
||||
}, []);
|
||||
|
||||
// Set up socket event listeners
|
||||
useEffect(() => {
|
||||
const unsubscribeConnection = socketService.onConnectionChange((connected) => {
|
||||
setIsConnected(connected);
|
||||
});
|
||||
|
||||
const unsubscribeMessage = socketService.onNewMessage((message) => {
|
||||
// Only add if it's for the current conversation
|
||||
if (message.conversationId === currentConversationIdRef.current) {
|
||||
setMessages((prev) => {
|
||||
// Avoid duplicates
|
||||
if (prev.some((m) => m.id === message.id)) return prev;
|
||||
return [...prev, message];
|
||||
});
|
||||
}
|
||||
|
||||
// Update conversation list
|
||||
setConversations((prev) =>
|
||||
prev.map((c) => {
|
||||
if (c.id === message.conversationId) {
|
||||
const isCurrentConversation =
|
||||
c.id === currentConversationIdRef.current;
|
||||
return {
|
||||
...c,
|
||||
lastMessageAt: message.createdAt,
|
||||
lastMessageText: message.content.substring(0, 255),
|
||||
unreadCount: isCurrentConversation ? 0 : c.unreadCount + 1,
|
||||
};
|
||||
}
|
||||
return c;
|
||||
})
|
||||
);
|
||||
});
|
||||
|
||||
const unsubscribeTypingStart = socketService.onTypingStart(
|
||||
(data: TypingIndicator) => {
|
||||
if (data.conversationId === currentConversationIdRef.current) {
|
||||
setTypingUsers((prev) => new Set(prev).add(data.userId));
|
||||
}
|
||||
}
|
||||
);
|
||||
|
||||
const unsubscribeTypingStop = socketService.onTypingStop(
|
||||
(data: TypingIndicator) => {
|
||||
if (data.conversationId === currentConversationIdRef.current) {
|
||||
setTypingUsers((prev) => {
|
||||
const next = new Set(prev);
|
||||
next.delete(data.userId);
|
||||
return next;
|
||||
});
|
||||
}
|
||||
}
|
||||
);
|
||||
|
||||
const unsubscribeStatus = socketService.onUserStatusChange(
|
||||
(data: UserStatusChange) => {
|
||||
updateUserStatus(data.userId, data.isOnline);
|
||||
}
|
||||
);
|
||||
|
||||
const unsubscribeRead = socketService.onMessagesRead(
|
||||
(data: MessagesReadEvent) => {
|
||||
if (data.conversationId === currentConversationIdRef.current) {
|
||||
// Update message statuses
|
||||
setMessages((prev) =>
|
||||
prev.map((m) => ({
|
||||
...m,
|
||||
status: 'READ' as MessageStatus,
|
||||
readAt: data.readAt,
|
||||
}))
|
||||
);
|
||||
}
|
||||
}
|
||||
);
|
||||
|
||||
return () => {
|
||||
unsubscribeConnection();
|
||||
unsubscribeMessage();
|
||||
unsubscribeTypingStart();
|
||||
unsubscribeTypingStop();
|
||||
unsubscribeStatus();
|
||||
unsubscribeRead();
|
||||
};
|
||||
}, [updateUserStatus]);
|
||||
|
||||
// Auto-connect on mount
|
||||
useEffect(() => {
|
||||
if (autoConnect && accessToken && !isInitializedRef.current) {
|
||||
isInitializedRef.current = true;
|
||||
connect();
|
||||
loadConversations();
|
||||
|
||||
// Get initial unread count
|
||||
messagesService.getUnreadCount().then(setUnreadCount).catch(console.error);
|
||||
}
|
||||
|
||||
// Don't disconnect on cleanup - socket is singleton and should stay connected
|
||||
// Disconnect only happens when user logs out (handled elsewhere)
|
||||
// eslint-disable-next-line react-hooks/exhaustive-deps
|
||||
}, [autoConnect, accessToken]);
|
||||
|
||||
// Re-join conversation room when reconnecting
|
||||
useEffect(() => {
|
||||
if (isConnected && currentConversationIdRef.current) {
|
||||
socketService.joinConversation(currentConversationIdRef.current);
|
||||
}
|
||||
}, [isConnected]);
|
||||
|
||||
return {
|
||||
conversations,
|
||||
currentConversation,
|
||||
messages,
|
||||
isConnected,
|
||||
isLoading,
|
||||
isLoadingMessages,
|
||||
typingUsers,
|
||||
unreadCount,
|
||||
connect,
|
||||
disconnect,
|
||||
loadConversations,
|
||||
selectConversation,
|
||||
startConversation,
|
||||
sendMessage,
|
||||
loadMoreMessages,
|
||||
startTyping,
|
||||
stopTyping,
|
||||
markAsRead,
|
||||
updateUserStatus,
|
||||
hasMoreMessages,
|
||||
};
|
||||
}
|
||||
Reference in New Issue
Block a user