582 lines
19 KiB
Dart
582 lines
19 KiB
Dart
import 'dart:async';
|
|
|
|
import 'package:flutter_riverpod/flutter_riverpod.dart';
|
|
import 'package:logger/logger.dart';
|
|
import 'package:real_estate_mobile/features/auth/presentation/providers/auth_provider.dart';
|
|
import 'package:real_estate_mobile/features/messaging/data/models/messaging_models.dart';
|
|
import 'package:real_estate_mobile/features/messaging/data/messaging_repository.dart';
|
|
import 'package:real_estate_mobile/features/messaging/data/socket_service.dart';
|
|
|
|
final _log = Logger(printer: PrettyPrinter(methodCount: 0));
|
|
|
|
class MessagingState {
|
|
final List<Conversation> conversations;
|
|
final Map<String, List<ChatMessage>> messages;
|
|
final Map<String, PaginationInfo> pagination;
|
|
final Set<String> typingUsers;
|
|
final bool isLoadingConversations;
|
|
final bool isLoadingMessages;
|
|
final bool isSending;
|
|
final String? error;
|
|
final String? activeConversationId;
|
|
final int unreadCount;
|
|
|
|
const MessagingState({
|
|
this.conversations = const [],
|
|
this.messages = const {},
|
|
this.pagination = const {},
|
|
this.typingUsers = const {},
|
|
this.isLoadingConversations = false,
|
|
this.isLoadingMessages = false,
|
|
this.isSending = false,
|
|
this.error,
|
|
this.activeConversationId,
|
|
this.unreadCount = 0,
|
|
});
|
|
|
|
MessagingState copyWith({
|
|
List<Conversation>? conversations,
|
|
Map<String, List<ChatMessage>>? messages,
|
|
Map<String, PaginationInfo>? pagination,
|
|
Set<String>? typingUsers,
|
|
bool? isLoadingConversations,
|
|
bool? isLoadingMessages,
|
|
bool? isSending,
|
|
String? error,
|
|
String? activeConversationId,
|
|
int? unreadCount,
|
|
bool clearError = false,
|
|
bool clearActiveConversation = false,
|
|
}) {
|
|
return MessagingState(
|
|
conversations: conversations ?? this.conversations,
|
|
messages: messages ?? this.messages,
|
|
pagination: pagination ?? this.pagination,
|
|
typingUsers: typingUsers ?? this.typingUsers,
|
|
isLoadingConversations:
|
|
isLoadingConversations ?? this.isLoadingConversations,
|
|
isLoadingMessages: isLoadingMessages ?? this.isLoadingMessages,
|
|
isSending: isSending ?? this.isSending,
|
|
error: clearError ? null : (error ?? this.error),
|
|
activeConversationId: clearActiveConversation
|
|
? null
|
|
: (activeConversationId ?? this.activeConversationId),
|
|
unreadCount: unreadCount ?? this.unreadCount,
|
|
);
|
|
}
|
|
}
|
|
|
|
class MessagingNotifier extends StateNotifier<MessagingState> {
|
|
final MessagingRepository _repository;
|
|
final Ref _ref;
|
|
final SocketService _socket = SocketService();
|
|
final List<StreamSubscription> _subscriptions = [];
|
|
|
|
MessagingNotifier(this._repository, this._ref)
|
|
: super(const MessagingState()) {
|
|
_initSocketListeners();
|
|
}
|
|
|
|
String? get _currentUserId => _ref.read(authProvider).user?.id;
|
|
|
|
void _initSocketListeners() {
|
|
// Online/offline status changes
|
|
_subscriptions.add(_socket.onStatusChange.listen((data) {
|
|
final userId = data['userId'] as String?;
|
|
final isOnline = data['isOnline'] as bool? ?? false;
|
|
final lastSeenAt = data['lastSeenAt'] as String?;
|
|
if (userId != null) {
|
|
updateUserStatus(userId, isOnline, lastSeenAt: lastSeenAt);
|
|
}
|
|
}));
|
|
|
|
// Incoming messages from other users
|
|
_subscriptions.add(_socket.onNewMessage.listen((message) {
|
|
if (!mounted) return;
|
|
final convId = message.conversationId;
|
|
final currentMessages =
|
|
List<ChatMessage>.from(state.messages[convId] ?? []);
|
|
|
|
// Avoid duplicates (broadcast may arrive after ack for sender's own message)
|
|
if (currentMessages.any((m) => m.id == message.id)) return;
|
|
|
|
// Ignore own messages — the sender path handles these via sendMessage()
|
|
final isMine = message.senderId == _currentUserId;
|
|
if (isMine) return;
|
|
|
|
currentMessages.insert(0, message);
|
|
final updatedMessages =
|
|
Map<String, List<ChatMessage>>.from(state.messages)
|
|
..[convId] = currentMessages;
|
|
|
|
// Show friendly preview for non-text messages (matching web)
|
|
String previewText = message.content;
|
|
if (message.messageType == MessageType.image) {
|
|
previewText = 'Sent an image';
|
|
} else if (message.messageType == MessageType.file) {
|
|
previewText = 'Sent a file';
|
|
} else if (_isGifUrl(message.content)) {
|
|
previewText = 'GIF';
|
|
}
|
|
|
|
// Update conversation's last message and unread count
|
|
final updatedConversations = state.conversations.map((c) {
|
|
if (c.id == convId) {
|
|
final isActive = state.activeConversationId == convId;
|
|
return c.copyWith(
|
|
lastMessageText: previewText,
|
|
lastMessageAt: message.createdAt,
|
|
unreadCount: isActive ? c.unreadCount : c.unreadCount + 1,
|
|
);
|
|
}
|
|
return c;
|
|
}).toList();
|
|
|
|
state = state.copyWith(
|
|
messages: updatedMessages,
|
|
conversations: updatedConversations,
|
|
);
|
|
}));
|
|
|
|
// Typing indicators
|
|
_subscriptions.add(_socket.onTypingStart.listen((data) {
|
|
if (!mounted) return;
|
|
final userId = data['userId'] as String?;
|
|
if (userId != null && userId != _currentUserId) {
|
|
state = state.copyWith(
|
|
typingUsers: {...state.typingUsers, userId},
|
|
);
|
|
}
|
|
}));
|
|
|
|
_subscriptions.add(_socket.onTypingStop.listen((data) {
|
|
if (!mounted) return;
|
|
final userId = data['userId'] as String?;
|
|
if (userId != null) {
|
|
final updated = Set<String>.from(state.typingUsers)..remove(userId);
|
|
state = state.copyWith(typingUsers: updated);
|
|
}
|
|
}));
|
|
|
|
// Read receipts
|
|
_subscriptions.add(_socket.onMessagesRead.listen((data) {
|
|
if (!mounted) return;
|
|
final convId = data['conversationId'] as String?;
|
|
if (convId == null) return;
|
|
final updatedConversations = state.conversations.map((c) {
|
|
if (c.id == convId) return c.copyWith(unreadCount: 0);
|
|
return c;
|
|
}).toList();
|
|
state = state.copyWith(conversations: updatedConversations);
|
|
}));
|
|
|
|
// Reconnection: re-join active conversation room and catch up missed messages
|
|
_subscriptions.add(_socket.onConnectionChange.listen((connected) {
|
|
if (!mounted || !connected) return;
|
|
final activeConvId = state.activeConversationId;
|
|
if (activeConvId != null) {
|
|
_rejoinAndCatchUp(activeConvId);
|
|
}
|
|
// Refresh conversation list to catch any missed updates
|
|
loadConversations();
|
|
}));
|
|
}
|
|
|
|
/// Re-join socket room and fetch missed messages after reconnection (matching web)
|
|
Future<void> _rejoinAndCatchUp(String conversationId, {int attempt = 1}) async {
|
|
try {
|
|
await _socket.joinConversation(conversationId);
|
|
} catch (_) {
|
|
if (attempt < 3 && mounted && state.activeConversationId == conversationId) {
|
|
await Future.delayed(Duration(seconds: attempt));
|
|
return _rejoinAndCatchUp(conversationId, attempt: attempt + 1);
|
|
}
|
|
return;
|
|
}
|
|
|
|
// Fetch recent messages to catch any missed during disconnect
|
|
if (!mounted || state.activeConversationId != conversationId) return;
|
|
try {
|
|
final result = await _repository.getMessages(conversationId, page: 1);
|
|
final recentMessages = result.messages.reversed.toList();
|
|
final currentMessages = List<ChatMessage>.from(state.messages[conversationId] ?? []);
|
|
final existingIds = currentMessages.map((m) => m.id).toSet();
|
|
final newMessages = recentMessages.where((m) => !existingIds.contains(m.id)).toList();
|
|
|
|
if (newMessages.isNotEmpty && mounted) {
|
|
currentMessages.insertAll(0, newMessages);
|
|
// Sort newest first
|
|
currentMessages.sort((a, b) => b.createdAt.compareTo(a.createdAt));
|
|
final updatedMessages = Map<String, List<ChatMessage>>.from(state.messages)
|
|
..[conversationId] = currentMessages;
|
|
state = state.copyWith(messages: updatedMessages);
|
|
}
|
|
} catch (e) {
|
|
_log.e('Failed to catch up messages after reconnect', error: e);
|
|
}
|
|
}
|
|
|
|
static bool _isGifUrl(String text) {
|
|
final lower = text.trim().toLowerCase();
|
|
return lower.endsWith('.gif') ||
|
|
lower.contains('giphy.com/') ||
|
|
lower.contains('giphy.gif') ||
|
|
lower.contains('tenor.com/');
|
|
}
|
|
|
|
/// Update online status for a user across all conversations (matching web)
|
|
void updateUserStatus(String userId, bool isOnline, {String? lastSeenAt}) {
|
|
if (!mounted) return;
|
|
final updatedConversations = state.conversations.map((c) {
|
|
if (c.otherParty.userId == userId || c.otherParty.id == userId) {
|
|
return c.copyWith(
|
|
otherParty: c.otherParty.copyWith(
|
|
isOnline: isOnline,
|
|
lastSeenAt: lastSeenAt,
|
|
),
|
|
);
|
|
}
|
|
return c;
|
|
}).toList();
|
|
|
|
state = state.copyWith(conversations: updatedConversations);
|
|
}
|
|
|
|
@override
|
|
void dispose() {
|
|
for (final sub in _subscriptions) {
|
|
sub.cancel();
|
|
}
|
|
_subscriptions.clear();
|
|
super.dispose();
|
|
}
|
|
|
|
Future<void> loadConversations() async {
|
|
state = state.copyWith(isLoadingConversations: true, clearError: true);
|
|
try {
|
|
final result = await _repository.getConversations();
|
|
state = state.copyWith(
|
|
conversations: result,
|
|
isLoadingConversations: false,
|
|
);
|
|
} catch (e, stack) {
|
|
_log.e('Failed to load conversations', error: e, stackTrace: stack);
|
|
state = state.copyWith(
|
|
isLoadingConversations: false,
|
|
error: 'Failed to load conversations',
|
|
);
|
|
}
|
|
}
|
|
|
|
Future<void> loadMessages(
|
|
String conversationId, {
|
|
bool loadMore = false,
|
|
}) async {
|
|
if (loadMore) {
|
|
final existing = state.pagination[conversationId];
|
|
if (existing != null && !existing.hasMore) return;
|
|
}
|
|
|
|
state = state.copyWith(isLoadingMessages: true, clearError: true);
|
|
|
|
try {
|
|
final nextPage =
|
|
loadMore ? ((state.pagination[conversationId]?.page ?? 0) + 1) : 1;
|
|
|
|
final result = await _repository.getMessages(
|
|
conversationId,
|
|
page: nextPage,
|
|
);
|
|
|
|
// API returns chronological order (oldest first), but our ListView is
|
|
// reverse: true so we need newest-first order (index 0 = newest).
|
|
final newMessages = result.messages.reversed.toList();
|
|
final paginationInfo = result.pagination;
|
|
|
|
final existingMessages =
|
|
loadMore ? (state.messages[conversationId] ?? []) : <ChatMessage>[];
|
|
// When loading more (older messages), append to end (older = higher index)
|
|
final mergedMessages =
|
|
loadMore ? [...existingMessages, ...newMessages] : newMessages;
|
|
|
|
final updatedMessages =
|
|
Map<String, List<ChatMessage>>.from(state.messages)
|
|
..[conversationId] = mergedMessages;
|
|
|
|
final updatedPagination =
|
|
Map<String, PaginationInfo>.from(state.pagination)
|
|
..[conversationId] = paginationInfo;
|
|
|
|
state = state.copyWith(
|
|
messages: updatedMessages,
|
|
pagination: updatedPagination,
|
|
isLoadingMessages: false,
|
|
);
|
|
} catch (_) {
|
|
state = state.copyWith(
|
|
isLoadingMessages: false,
|
|
error: 'Failed to load messages',
|
|
);
|
|
}
|
|
}
|
|
|
|
Future<void> sendMessage(String conversationId, String content) async {
|
|
final currentUserId = _currentUserId;
|
|
if (currentUserId == null) return;
|
|
|
|
final tempId = 'temp_${DateTime.now().millisecondsSinceEpoch}';
|
|
final now = DateTime.now().toIso8601String();
|
|
|
|
final tempMessage = ChatMessage(
|
|
id: tempId,
|
|
conversationId: conversationId,
|
|
senderId: currentUserId,
|
|
content: content,
|
|
messageType: MessageType.text,
|
|
status: MessageStatus.sent,
|
|
createdAt: now,
|
|
updatedAt: now,
|
|
sender: MessageSender(
|
|
id: currentUserId,
|
|
role: 'user',
|
|
),
|
|
);
|
|
|
|
// Add optimistic message at the beginning (newest-first)
|
|
final currentMessages =
|
|
List<ChatMessage>.from(state.messages[conversationId] ?? []);
|
|
currentMessages.insert(0, tempMessage);
|
|
|
|
final updatedMessages =
|
|
Map<String, List<ChatMessage>>.from(state.messages)
|
|
..[conversationId] = currentMessages;
|
|
|
|
state = state.copyWith(
|
|
messages: updatedMessages,
|
|
isSending: true,
|
|
clearError: true,
|
|
);
|
|
|
|
try {
|
|
// Send via socket so the backend broadcasts new_message to the other user.
|
|
// The REST endpoint does NOT emit socket events, so using REST here
|
|
// would prevent real-time delivery to the other participant.
|
|
ChatMessage? realMessage;
|
|
|
|
if (_socket.isConnected) {
|
|
realMessage = await _socket.sendMessage(conversationId, {
|
|
'content': content,
|
|
'messageType': 'TEXT',
|
|
});
|
|
}
|
|
|
|
// Fallback to REST if socket send failed or not connected
|
|
realMessage ??= await _repository.sendMessage(
|
|
conversationId,
|
|
content: content,
|
|
);
|
|
|
|
final messagesAfterSend =
|
|
List<ChatMessage>.from(state.messages[conversationId] ?? []);
|
|
final tempIndex = messagesAfterSend.indexWhere((m) => m.id == tempId);
|
|
if (tempIndex != -1) {
|
|
messagesAfterSend[tempIndex] = realMessage;
|
|
}
|
|
|
|
final finalMessages =
|
|
Map<String, List<ChatMessage>>.from(state.messages)
|
|
..[conversationId] = messagesAfterSend;
|
|
|
|
// Update conversation's last message with friendly preview
|
|
final sent = realMessage;
|
|
String sentPreview = sent.content;
|
|
if (sent.messageType == MessageType.image) {
|
|
sentPreview = 'Sent an image';
|
|
} else if (sent.messageType == MessageType.file) {
|
|
sentPreview = 'Sent a file';
|
|
} else if (_isGifUrl(sent.content)) {
|
|
sentPreview = 'GIF';
|
|
}
|
|
|
|
final updatedConversations = state.conversations.map((c) {
|
|
if (c.id == conversationId) {
|
|
return c.copyWith(
|
|
lastMessageText: sentPreview,
|
|
lastMessageAt: sent.createdAt,
|
|
);
|
|
}
|
|
return c;
|
|
}).toList();
|
|
|
|
state = state.copyWith(
|
|
messages: finalMessages,
|
|
conversations: updatedConversations,
|
|
isSending: false,
|
|
);
|
|
} catch (_) {
|
|
final messagesAfterError =
|
|
List<ChatMessage>.from(state.messages[conversationId] ?? []);
|
|
messagesAfterError.removeWhere((m) => m.id == tempId);
|
|
|
|
final errorMessages =
|
|
Map<String, List<ChatMessage>>.from(state.messages)
|
|
..[conversationId] = messagesAfterError;
|
|
|
|
state = state.copyWith(
|
|
messages: errorMessages,
|
|
isSending: false,
|
|
error: 'Failed to send message',
|
|
);
|
|
}
|
|
}
|
|
|
|
Future<void> markAsRead(String conversationId) async {
|
|
try {
|
|
await _repository.markAsRead(conversationId);
|
|
|
|
final updatedConversations = state.conversations.map((c) {
|
|
if (c.id == conversationId) {
|
|
return c.copyWith(unreadCount: 0);
|
|
}
|
|
return c;
|
|
}).toList();
|
|
|
|
final totalUnread = updatedConversations.fold<int>(
|
|
0,
|
|
(sum, c) => sum + c.unreadCount,
|
|
);
|
|
|
|
state = state.copyWith(
|
|
conversations: updatedConversations,
|
|
unreadCount: totalUnread,
|
|
);
|
|
} catch (_) {}
|
|
}
|
|
|
|
void setActiveConversation(String? conversationId) {
|
|
if (conversationId == null) {
|
|
state = state.copyWith(clearActiveConversation: true);
|
|
} else {
|
|
state = state.copyWith(activeConversationId: conversationId);
|
|
}
|
|
}
|
|
|
|
Future<void> loadUnreadCount() async {
|
|
try {
|
|
final count = await _repository.getUnreadCount();
|
|
state = state.copyWith(unreadCount: count);
|
|
} catch (_) {}
|
|
}
|
|
|
|
Future<String?> startConversation(String agentProfileId) async {
|
|
if (!mounted) return null;
|
|
state = state.copyWith(isLoadingConversations: true, clearError: true);
|
|
try {
|
|
final conversation =
|
|
await _repository.startConversation(agentProfileId);
|
|
|
|
if (!mounted) return conversation.id;
|
|
|
|
final exists = state.conversations.any((c) => c.id == conversation.id);
|
|
final updatedConversations = exists
|
|
? state.conversations
|
|
: [conversation, ...state.conversations];
|
|
|
|
state = state.copyWith(
|
|
conversations: updatedConversations,
|
|
isLoadingConversations: false,
|
|
);
|
|
return conversation.id;
|
|
} catch (_) {
|
|
if (mounted) {
|
|
state = state.copyWith(
|
|
isLoadingConversations: false,
|
|
error: 'Failed to start conversation',
|
|
);
|
|
}
|
|
return null;
|
|
}
|
|
}
|
|
|
|
Future<String?> startConversationWithUser(String userId) async {
|
|
if (!mounted) return null;
|
|
state = state.copyWith(isLoadingConversations: true, clearError: true);
|
|
try {
|
|
final conversation =
|
|
await _repository.startConversationWithUser(userId);
|
|
|
|
if (!mounted) return conversation.id;
|
|
|
|
final exists = state.conversations.any((c) => c.id == conversation.id);
|
|
final updatedConversations = exists
|
|
? state.conversations
|
|
: [conversation, ...state.conversations];
|
|
|
|
state = state.copyWith(
|
|
conversations: updatedConversations,
|
|
isLoadingConversations: false,
|
|
);
|
|
return conversation.id;
|
|
} catch (_) {
|
|
if (mounted) {
|
|
state = state.copyWith(
|
|
isLoadingConversations: false,
|
|
error: 'Failed to start conversation',
|
|
);
|
|
}
|
|
return null;
|
|
}
|
|
}
|
|
|
|
Future<void> clearChat(String conversationId) async {
|
|
try {
|
|
await _repository.clearChat(conversationId);
|
|
// Clear local messages
|
|
final updatedMessages =
|
|
Map<String, List<ChatMessage>>.from(state.messages)
|
|
..[conversationId] = [];
|
|
// Update conversation preview
|
|
final updatedConversations = state.conversations.map((c) {
|
|
if (c.id == conversationId) {
|
|
return c.copyWith(lastMessageText: '', unreadCount: 0);
|
|
}
|
|
return c;
|
|
}).toList();
|
|
state = state.copyWith(
|
|
messages: updatedMessages,
|
|
conversations: updatedConversations,
|
|
);
|
|
} catch (_) {
|
|
state = state.copyWith(error: 'Failed to clear chat');
|
|
}
|
|
}
|
|
|
|
Future<void> deleteConversation(String conversationId) async {
|
|
try {
|
|
await _repository.deleteConversation(conversationId);
|
|
// Remove from local state
|
|
final updatedMessages =
|
|
Map<String, List<ChatMessage>>.from(state.messages)
|
|
..remove(conversationId);
|
|
final updatedConversations =
|
|
state.conversations.where((c) => c.id != conversationId).toList();
|
|
state = state.copyWith(
|
|
messages: updatedMessages,
|
|
conversations: updatedConversations,
|
|
clearActiveConversation: true,
|
|
);
|
|
} catch (_) {
|
|
state = state.copyWith(error: 'Failed to delete conversation');
|
|
}
|
|
}
|
|
}
|
|
|
|
final messagingProvider =
|
|
StateNotifierProvider<MessagingNotifier, MessagingState>((ref) {
|
|
return MessagingNotifier(ref.watch(messagingRepositoryProvider), ref);
|
|
});
|
|
|
|
final currentUserIdProvider = Provider<String?>((ref) {
|
|
return ref.watch(authProvider).user?.id;
|
|
});
|