🔊 Add more logging and optimzation

♻️ Refactored the message repository logic
2025-08-16 23:39:41 +08:00 · 2025-08-16 23:07:21 +08:00
4 changed files with 845 additions and 1064 deletions
--- a/lib/database/message_repository.dart
+++ b/lib/database/message_repository.dart
@@ -1,484 +0,0 @@
 import 'package:dio/dio.dart';
 import 'package:island/database/drift_db.dart';
 import 'package:island/database/message.dart';
 import 'package:island/models/chat.dart';
 import 'package:island/models/file.dart';
 import 'package:island/services/file.dart';
 import 'package:island/widgets/alert.dart';
 import 'package:uuid/uuid.dart';
 class MessageRepository {
  final SnChatRoom room;
  final SnChatMember identity;
  final Dio _apiClient;
  final AppDatabase _database;
  final Map<String, LocalChatMessage> pendingMessages = {};
  final Map<String, Map<int, double>> fileUploadProgress = {};
  int? _totalCount;
  MessageRepository(this.room, this.identity, this._apiClient, this._database);
  Future<LocalChatMessage?> getLastMessages() async {
    final dbMessages = await _database.getMessagesForRoom(
      room.id,
      offset: 0,
      limit: 1,
    );
    if (dbMessages.isEmpty) {
      return null;
    }
    return _database.companionToMessage(dbMessages.first);
  }
  Future<bool> syncMessages() async {
    final lastMessage = await getLastMessages();
    if (lastMessage == null) return false;
    try {
      final resp = await _apiClient.post(
        '/sphere/chat/${room.id}/sync',
        data: {
          'last_sync_timestamp':
              lastMessage.toRemoteMessage().updatedAt.millisecondsSinceEpoch,
        },
      );
      final response = MessageSyncResponse.fromJson(resp.data);
      for (final change in response.changes) {
        switch (change.action) {
          case MessageChangeAction.create:
            await receiveMessage(change.message!);
            break;
          case MessageChangeAction.update:
            await receiveMessageUpdate(change.message!);
            break;
          case MessageChangeAction.delete:
            await receiveMessageDeletion(change.messageId.toString());
            break;
        }
      }
    } catch (err) {
      showErrorAlert(err);
    }
    return true;
  }
  Future<List<LocalChatMessage>> listMessages({
    int offset = 0,
    int take = 20,
    bool synced = false,
  }) async {
    try {
      // For initial load, fetch latest messages in the background to sync.
      if (offset == 0 && !synced) {
        // Not awaiting this is intentional, for a quicker UI response.
        // The UI should rely on a stream from the database to get updates.
        _fetchAndCacheMessages(room.id, offset: 0, take: take).catchError((_) {
          // Best effort, errors will be handled by later fetches.
          return <LocalChatMessage>[];
        });
      }
      final localMessages = await _getCachedMessages(
        room.id,
        offset: offset,
        take: take,
      );
      // If local cache has messages, return them. This is the common case for scrolling up.
      if (localMessages.isNotEmpty) {
        return localMessages;
      }
      // If local cache is empty, we've probably reached the end of cached history.
      // Fetch from remote. This will also be hit on first load if cache is empty.
      return await _fetchAndCacheMessages(room.id, offset: offset, take: take);
    } catch (e) {
      // Final fallback to cache in case of network errors during fetch.
      final localMessages = await _getCachedMessages(
        room.id,
        offset: offset,
        take: take,
      );
      if (localMessages.isNotEmpty) {
        return localMessages;
      }
      rethrow;
    }
  }
  Future<List<LocalChatMessage>> _getCachedMessages(
    String roomId, {
    int offset = 0,
    int take = 20,
  }) async {
    // Get messages from local database
    final dbMessages = await _database.getMessagesForRoom(
      roomId,
      offset: offset,
      limit: take,
    );
    final dbLocalMessages =
        dbMessages.map(_database.companionToMessage).toList();
    // Combine with pending messages for the first page
    if (offset == 0) {
      final pendingForRoom =
          pendingMessages.values.where((msg) => msg.roomId == roomId).toList();
      final allMessages = [...pendingForRoom, ...dbLocalMessages];
      allMessages.sort((a, b) => b.createdAt.compareTo(a.createdAt));
      // Remove duplicates by ID, preserving the order
      final uniqueMessages = <LocalChatMessage>[];
      final seenIds = <String>{};
      for (final message in allMessages) {
        if (seenIds.add(message.id)) {
          uniqueMessages.add(message);
        }
      }
      return uniqueMessages;
    }
    return dbLocalMessages;
  }
  Future<List<LocalChatMessage>> _fetchAndCacheMessages(
    String roomId, {
    int offset = 0,
    int take = 20,
  }) async {
    // Use cached total count if available, otherwise fetch it
    if (_totalCount == null) {
      final response = await _apiClient.get(
        '/sphere/chat/$roomId/messages',
        queryParameters: {'offset': 0, 'take': 1},
      );
      _totalCount = int.parse(response.headers['x-total']?.firstOrNull ?? '0');
    }
    if (offset >= _totalCount!) {
      return [];
    }
    final response = await _apiClient.get(
      '/sphere/chat/$roomId/messages',
      queryParameters: {'offset': offset, 'take': take},
    );
    final List<dynamic> data = response.data;
    // Update total count from response headers
    _totalCount = int.parse(response.headers['x-total']?.firstOrNull ?? '0');
    final messages =
        data.map((json) {
          final remoteMessage = SnChatMessage.fromJson(json);
          return LocalChatMessage.fromRemoteMessage(
            remoteMessage,
            MessageStatus.sent,
          );
        }).toList();
    for (final message in messages) {
      await _database.saveMessage(_database.messageToCompanion(message));
      if (message.nonce != null) {
        pendingMessages.removeWhere(
          (_, pendingMsg) => pendingMsg.nonce == message.nonce,
        );
      }
    }
    return messages;
  }
  Future<LocalChatMessage> sendMessage(
    String token,
    String baseUrl,
    String roomId,
    String content,
    String nonce, {
    required List<UniversalFile> attachments,
    Map<String, dynamic>? meta,
    SnChatMessage? replyingTo,
    SnChatMessage? forwardingTo,
    SnChatMessage? editingTo,
    Function(LocalChatMessage)? onPending,
    Function(String, Map<int, double>)? onProgress,
  }) async {
    // Generate a unique nonce for this message
    final nonce = const Uuid().v4();
    // Create a local message with pending status
    final mockMessage = SnChatMessage(
      id: 'pending_$nonce',
      chatRoomId: roomId,
      senderId: identity.id,
      content: content,
      createdAt: DateTime.now(),
      updatedAt: DateTime.now(),
      nonce: nonce,
      sender: identity,
    );
    final localMessage = LocalChatMessage.fromRemoteMessage(
      mockMessage,
      MessageStatus.pending,
    );
    // Store in memory and database
    pendingMessages[localMessage.id] = localMessage;
    fileUploadProgress[localMessage.id] = {};
    await _database.saveMessage(_database.messageToCompanion(localMessage));
    onPending?.call(localMessage);
    try {
      var cloudAttachments = List.empty(growable: true);
      // Upload files
      for (var idx = 0; idx < attachments.length; idx++) {
        final cloudFile =
            await putMediaToCloud(
              fileData: attachments[idx],
              atk: token,
              baseUrl: baseUrl,
              filename: attachments[idx].data.name ?? 'Post media',
              mimetype:
                  attachments[idx].data.mimeType ??
                  switch (attachments[idx].type) {
                    UniversalFileType.image => 'image/unknown',
                    UniversalFileType.video => 'video/unknown',
                    UniversalFileType.audio => 'audio/unknown',
                    UniversalFileType.file => 'application/octet-stream',
                  },
              onProgress: (progress, _) {
                fileUploadProgress[localMessage.id]?[idx] = progress;
                onProgress?.call(
                  localMessage.id,
                  fileUploadProgress[localMessage.id] ?? {},
                );
              },
            ).future;
        if (cloudFile == null) {
          throw ArgumentError('Failed to upload the file...');
        }
        cloudAttachments.add(cloudFile);
      }
      // Send to server
      final response = await _apiClient.request(
        editingTo == null
            ? '/sphere/chat/$roomId/messages'
            : '/sphere/chat/$roomId/messages/${editingTo.id}',
        data: {
          'content': content,
          'attachments_id': cloudAttachments.map((e) => e.id).toList(),
          'replied_message_id': replyingTo?.id,
          'forwarded_message_id': forwardingTo?.id,
          'meta': meta,
          'nonce': nonce,
        },
        options: Options(method: editingTo == null ? 'POST' : 'PATCH'),
      );
      // Update with server response
      final remoteMessage = SnChatMessage.fromJson(response.data);
      final updatedMessage = LocalChatMessage.fromRemoteMessage(
        remoteMessage,
        MessageStatus.sent,
      );
      // Remove from pending and update in database
      pendingMessages.remove(localMessage.id);
      await _database.deleteMessage(localMessage.id);
      await _database.saveMessage(_database.messageToCompanion(updatedMessage));
      return updatedMessage;
    } catch (e) {
      // Update status to failed
      localMessage.status = MessageStatus.failed;
      pendingMessages[localMessage.id] = localMessage;
      await _database.updateMessageStatus(
        localMessage.id,
        MessageStatus.failed,
      );
      rethrow;
    }
  }
  Future<LocalChatMessage> retryMessage(String pendingMessageId) async {
    final message = await getMessageById(pendingMessageId);
    if (message == null) {
      throw Exception('Message not found');
    }
    // Update status back to pending
    message.status = MessageStatus.pending;
    pendingMessages[pendingMessageId] = message;
    await _database.updateMessageStatus(
      pendingMessageId,
      MessageStatus.pending,
    );
    try {
      // Send to server
      var remoteMessage = message.toRemoteMessage();
      final response = await _apiClient.post(
        '/sphere/chat/${message.roomId}/messages',
        data: {
          'content': remoteMessage.content,
          'attachments_id': remoteMessage.attachments,
          'meta': remoteMessage.meta,
          'nonce': message.nonce,
        },
      );
      // Update with server response
      remoteMessage = SnChatMessage.fromJson(response.data);
      final updatedMessage = LocalChatMessage.fromRemoteMessage(
        remoteMessage,
        MessageStatus.sent,
      );
      // Remove from pending and update in database
      pendingMessages.remove(pendingMessageId);
      await _database.deleteMessage(pendingMessageId);
      await _database.saveMessage(_database.messageToCompanion(updatedMessage));
      return updatedMessage;
    } catch (e) {
      // Update status to failed
      message.status = MessageStatus.failed;
      pendingMessages[pendingMessageId] = message;
      await _database.updateMessageStatus(
        pendingMessageId,
        MessageStatus.failed,
      );
      rethrow;
    }
  }
  Future<LocalChatMessage> receiveMessage(SnChatMessage remoteMessage) async {
    final localMessage = LocalChatMessage.fromRemoteMessage(
      remoteMessage,
      MessageStatus.sent,
    );
    if (remoteMessage.nonce != null) {
      pendingMessages.removeWhere(
        (_, pendingMsg) => pendingMsg.nonce == remoteMessage.nonce,
      );
    }
    await _database.saveMessage(_database.messageToCompanion(localMessage));
    return localMessage;
  }
  Future<LocalChatMessage> receiveMessageUpdate(
    SnChatMessage remoteMessage,
  ) async {
    final localMessage = LocalChatMessage.fromRemoteMessage(
      remoteMessage,
      MessageStatus.sent,
    );
    await _database.updateMessage(_database.messageToCompanion(localMessage));
    return localMessage;
  }
  Future<void> receiveMessageDeletion(String messageId) async {
    // Remove from pending messages if exists
    pendingMessages.remove(messageId);
    // Delete from local database
    await _database.deleteMessage(messageId);
  }
  Future<LocalChatMessage> updateMessage(
    String messageId,
    String content, {
    List<SnCloudFile>? attachments,
    Map<String, dynamic>? meta,
  }) async {
    final message = pendingMessages[messageId];
    if (message != null) {
      // Update pending message
      final rmMessage = message.toRemoteMessage();
      final updatedRemoteMessage = rmMessage.copyWith(
        content: content,
        meta: meta ?? rmMessage.meta,
      );
      final updatedLocalMessage = LocalChatMessage.fromRemoteMessage(
        updatedRemoteMessage,
        MessageStatus.pending,
      );
      pendingMessages[messageId] = updatedLocalMessage;
      await _database.updateMessage(
        _database.messageToCompanion(updatedLocalMessage),
      );
      return message;
    }
    try {
      // Update on server
      final response = await _apiClient.put(
        '/sphere/chat/${room.id}/messages/$messageId',
        data: {'content': content, 'attachments': attachments, 'meta': meta},
      );
      // Update local copy
      final remoteMessage = SnChatMessage.fromJson(response.data);
      final updatedMessage = LocalChatMessage.fromRemoteMessage(
        remoteMessage,
        MessageStatus.sent,
      );
      await _database.updateMessage(
        _database.messageToCompanion(updatedMessage),
      );
      return updatedMessage;
    } catch (e) {
      rethrow;
    }
  }
  Future<void> deleteMessage(String messageId) async {
    try {
      await _apiClient.delete('/sphere/chat/${room.id}/messages/$messageId');
      pendingMessages.remove(messageId);
      await _database.deleteMessage(messageId);
    } catch (e) {
      rethrow;
    }
  }
  Future<LocalChatMessage?> getMessageById(String messageId) async {
    try {
      // Attempt to get the message from the local database
      final localMessage =
          await (_database.select(_database.chatMessages)
            ..where((tbl) => tbl.id.equals(messageId))).getSingleOrNull();
      if (localMessage != null) {
        return _database.companionToMessage(localMessage);
      }
      // If not found locally, fetch from the server
      final response = await _apiClient.get(
        '/sphere/chat/${room.id}/messages/$messageId',
      );
      final remoteMessage = SnChatMessage.fromJson(response.data);
      final message = LocalChatMessage.fromRemoteMessage(
        remoteMessage,
        MessageStatus.sent,
      );
      // Save the fetched message to the local database
      await _database.saveMessage(_database.messageToCompanion(message));
      return message;
    } catch (e) {
      if (e is DioException) return null;
      // Handle errors
      rethrow;
    }
  }
 }
--- a/lib/screens/chat/room.dart
+++ b/lib/screens/chat/room.dart
--- a/lib/screens/chat/room.g.dart
+++ b/lib/screens/chat/room.g.dart
@@ -6,7 +6,7 @@ part of 'room.dart';
 // RiverpodGenerator
 // **************************************************************************
-String _$messagesNotifierHash() => r'afc4d43f4948ec571118cef0321838a6cefc89c0';
+String _$messagesNotifierHash() => r'3b10c3101404f6528c7a83baa0d39cba1a30f579';
 /// Copied from Dart SDK
 class _SystemHash {
--- a/lib/screens/notification.g.dart
+++ b/lib/screens/notification.g.dart
@@ -7,7 +7,7 @@ part of 'notification.dart';
 // **************************************************************************
 String _$notificationUnreadCountNotifierHash() =>
-    r'd199abf0d16944587e747798399a267a790341f3';
+    r'0763b66bd64e5a9b7c317887e109ab367515dfa4';
 /// See also [NotificationUnreadCountNotifier].
@ProviderFor(NotificationUnreadCountNotifier)
Author	SHA1	Message	Date
LittleSheep	6892afb974	🔊 Add more logging and optimzation	2025-08-16 23:39:41 +08:00
LittleSheep	007b46b080	♻️ Refactored the message repository logic	2025-08-16 23:07:21 +08:00