unify errors, refactor retry_queue

Signed-off-by: Sahil Kumar <[email protected]>

packages/stream_chat/lib/src/api/channel.dart

@@ -7,6 +7,8 @@ import 'package:dio/dio.dart';
 import 'package:logging/logging.dart';
 import 'package:rxdart/rxdart.dart';
 import 'package:stream_chat/src/api/retry_queue.dart';
+import 'package:stream_chat/src/core/utils.dart';
+import 'package:stream_chat/src/errors/stream_chat_error.dart';
 import 'package:stream_chat/src/event_type.dart';
 import 'package:stream_chat/src/extensions/rate_limit.dart';
 import 'package:stream_chat/src/core/models/attachment_file.dart';
@@ -233,12 +235,14 @@ class Channel {
   }) {
     final cancelToken = _cancelableAttachmentUploadRequest[attachmentId];
     if (cancelToken == null) {
-      throw Exception(
-        "Upload request for this Attachment hasn't started yet or else "
+      throw const StreamChatError(
+        "Upload request for this Attachment hasn't started yet or maybe "
         'Already completed',
       );
     }
-    if (cancelToken.isCancelled) throw Exception('Already cancelled');
+    if (cancelToken.isCancelled) {
+      throw const StreamChatError('Upload request already cancelled');
+    }
     cancelToken.cancel(reason);
   }
 
@@ -255,7 +259,7 @@ class Channel {
     );
 
     if (message == null) {
-      throw Exception('Error, Message not found');
+      throw const StreamChatError('Error, Message not found');
     }
 
     final attachments = message.attachments.where((it) {
@@ -396,9 +400,9 @@ class Channel {
       final response = await _client.sendMessage(message, id!, type);
       state!.addMessage(response.message);
       return response;
-    } catch (error) {
-      if (error is DioError && error.type != DioErrorType.response) {
-        state!.retryQueue?.add([message]);
+    } catch (e) {
+      if (e is StreamChatNetworkError && e.isRetriable) {
+        state!._retryQueue.add([message]);
       }
       rethrow;
     }
@@ -453,9 +457,9 @@ class Channel {
       state?.addMessage(m);
 
       return response;
-    } catch (error) {
-      if (error is DioError && error.type != DioErrorType.response) {
-        state?.retryQueue?.add([message]);
+    } catch (e) {
+      if (e is StreamChatNetworkError && e.isRetriable) {
+        state!._retryQueue.add([message]);
       }
       rethrow;
     }
@@ -494,9 +498,9 @@ class Channel {
       state?.addMessage(message.copyWith(status: MessageSendingStatus.sent));
 
       return response;
-    } catch (error) {
-      if (error is DioError && error.type != DioErrorType.response) {
-        state?.retryQueue?.add([message]);
+    } catch (e) {
+      if (e is StreamChatNetworkError && e.isRetriable) {
+        state!._retryQueue.add([message]);
       }
       rethrow;
     }
@@ -1188,9 +1192,11 @@ class ChannelClientState {
                 _channel._client.chatPersistenceClient
                     ?.updateChannelState(state))
             .debounced(const Duration(seconds: 1)) {
-    retryQueue = RetryQueue(
+    _retryQueue = RetryQueue(
       channel: _channel,
-      logger: Logger('RETRY QUEUE ${_channel.cid}'),
+      logger: _channel.client.detachedLogger(
+        '⟳ (${generateHash([_channel.cid])})',
+      ),
     );
 
     _checkExpiredAttachmentMessages(channelState);
@@ -1338,7 +1344,7 @@ class ChannelClientState {
       BehaviorSubject.seeded(true);
 
   /// The retry queue associated to this channel
-  RetryQueue? retryQueue;
+  late final RetryQueue _retryQueue;
 
   /// Retry failed message
   Future<void> retryFailedMessages() async {
@@ -1357,7 +1363,7 @@ class ChannelClientState {
             )
             .toList();
 
-    retryQueue!.add(failedMessages);
+    _retryQueue.add(failedMessages);
   }
 
   void _listenReactionDeleted() {
@@ -1832,7 +1838,7 @@ class ChannelClientState {
   void dispose() {
     _debouncedUpdatePersistenceChannelState.cancel();
     _unreadCountController.close();
-    retryQueue!.dispose();
+    _retryQueue.dispose();
     _subscriptions.forEach((s) => s.cancel());
     _channelStateController.close();
     _isUpToDateController.close();

packages/stream_chat/lib/src/api/retry_policy.dart

@@ -1,5 +1,5 @@
 import 'package:stream_chat/src/client.dart';
-import 'package:stream_chat/src/exceptions.dart';
+import 'package:stream_chat/src/errors/stream_chat_error.dart';
 
 /// The retry options
 class RetryPolicy {
@@ -7,32 +7,25 @@ class RetryPolicy {
   RetryPolicy({
     required this.shouldRetry,
     required this.retryTimeout,
-    this.attempt = 0,
+    this.maxRetryAttempts = 6,
   });
 
-  /// The number of attempts tried so far
-  int attempt = 0;
+  /// Hard limit on maximum retry attempts before giving up, defaults to 6
+  /// Resets once connection recovers.
+  final int maxRetryAttempts;
 
   /// This function evaluates if we should retry the failure
-  final bool Function(StreamChatClient client, int attempt, ApiError? apiError)
-      shouldRetry;
+  final bool Function(
+    StreamChatClient client,
+    int attempt,
+    StreamChatError? error,
+  ) shouldRetry;
 
   /// In the case that we want to retry a failed request the retryTimeout
   /// method is called to determine the timeout
   final Duration Function(
-      StreamChatClient client, int attempt, ApiError? apiError) retryTimeout;
-
-  /// Creates a copy of [RetryPolicy] with specified attributes overridden.
-  RetryPolicy copyWith({
-    bool Function(StreamChatClient client, int attempt, ApiError? apiError)?
-        shouldRetry,
-    Duration Function(StreamChatClient client, int attempt, ApiError? apiError)?
-        retryTimeout,
-    int? attempt,
-  }) =>
-      RetryPolicy(
-        retryTimeout: retryTimeout ?? this.retryTimeout,
-        shouldRetry: shouldRetry ?? this.shouldRetry,
-        attempt: attempt ?? this.attempt,
-      );
+    StreamChatClient client,
+    int attempt,
+    StreamChatError? error,
+  ) retryTimeout;
 }

packages/stream_chat/lib/src/api/retry_queue.dart

@@ -2,10 +2,11 @@ import 'dart:async';
 
 import 'package:collection/collection.dart';
 import 'package:logging/logging.dart';
+import 'package:rxdart/rxdart.dart';
 import 'package:stream_chat/src/api/channel.dart';
 import 'package:stream_chat/src/api/retry_policy.dart';
+import 'package:stream_chat/src/errors/stream_chat_error.dart';
 import 'package:stream_chat/src/event_type.dart';
-import 'package:stream_chat/src/exceptions.dart';
 import 'package:stream_chat/src/core/models/message.dart';
 import 'package:stream_chat/stream_chat.dart';
 
@@ -15,116 +16,155 @@ class RetryQueue {
   RetryQueue({
     required this.channel,
     this.logger,
-  }) {
-    _retryPolicy = channel.client.retryPolicy;
-
+  }) : client = channel.client {
+    _retryPolicy = client.retryPolicy;
     _listenConnectionRecovered();
-
     _listenFailedEvents();
   }
 
   /// The channel of this queue
   final Channel channel;
 
+  /// The client associated with this [channel]
+  final StreamChatClient client;
+
   /// The logger associated to this queue
   final Logger? logger;
 
-  final _subscriptions = <StreamSubscription>[];
+  late final RetryPolicy _retryPolicy;
+
+  final _compositeSubscription = CompositeSubscription();
+
+  final _messageQueue = HeapPriorityQueue(_byDate);
+  bool _isRetrying = false;
 
   void _listenConnectionRecovered() {
-    _subscriptions
-        .add(channel.client.on(EventType.connectionRecovered).listen((event) {
-      if (!_isRetrying && event.online!) {
+    client.on(EventType.connectionRecovered).listen((event) {
+      if (event.online == true) {
         _startRetrying();
       }
-    }));
+    }).addTo(_compositeSubscription);
   }
 
-  final HeapPriorityQueue<Message> _messageQueue = HeapPriorityQueue(_byDate);
-  bool _isRetrying = false;
-  RetryPolicy? _retryPolicy;
+  void _listenFailedEvents() {
+    channel.on().where((event) => event.message != null).listen((event) {
+      final message = event.message!;
+      final containsMessage = _messageQueue.containsMessage(message);
+      if (!containsMessage) return;
+      if (message.status == MessageSendingStatus.sent) {
+        logger?.info('Removing sent message from queue : ${message.id}');
+        _messageQueue.removeMessage(message);
+        return;
+      } else {
+        if ([
+          MessageSendingStatus.failed_update,
+          MessageSendingStatus.failed,
+          MessageSendingStatus.failed_delete,
+        ].contains(message.status)) {
+          logger?.info('Adding failed message from event : ${event.type}');
+          add([message]);
+        }
+      }
+    }).addTo(_compositeSubscription);
+  }
 
   /// Add a list of messages
   void add(List<Message> messages) {
-    logger?.info('added ${messages.length} messages');
+    if (messages.isEmpty) return;
+    if (_messageQueue.containsAllMessage(messages)) return;
+
+    logger?.info('Adding ${messages.length} messages');
     final messageList = _messageQueue.toList();
-
-    _messageQueue.addAll(messages
-        .where((element) => !messageList.any((m) => m.id == element.id)));
-
-    if (_messageQueue.isNotEmpty && !_isRetrying) {
-      _startRetrying();
-    }
+    // we should not add message if already available in the queue
+    _messageQueue.addAll(messages.where(
+      (it) => !messageList.any((m) => m.id == it.id),
+    ));
+    _startRetrying();
   }
 
   Future<void> _startRetrying() async {
-    logger?.info('start retrying');
+    if (_isRetrying) return;
     _isRetrying = true;
-    final retryPolicy = _retryPolicy!.copyWith(attempt: 0);
 
+    logger?.info('Started retrying failed messages');
     while (_messageQueue.isNotEmpty) {
+      logger?.info('${_messageQueue.length} messages remaining in the queue');
       final message = _messageQueue.first;
-      try {
-        logger?.info('retry attempt ${retryPolicy.attempt}');
-        await _sendMessage(message);
-        logger?.info('message sent - removing it from the queue');
-        _messageQueue.remove(message);
-        logger?.info('now ${_messageQueue.length} messages in the queue');
-        retryPolicy.attempt = 0;
-      } catch (error) {
-        ApiError? apiError;
-        if (error is DioError) {
-          if (error.type == DioErrorType.response) {
-            _messageQueue.remove(message);
-            return;
-          }
-          apiError = ApiError(
-            error.response?.data,
-            error.response?.statusCode,
-          );
-        } else if (error is ApiError) {
-          apiError = error;
-          if (apiError.status?.toString().startsWith('4') == true) {
-            _messageQueue.remove(message);
-            return;
-          }
-        }
-
-        if (!retryPolicy.shouldRetry(
-          channel.client,
-          retryPolicy.attempt,
-          apiError,
-        )) {
-          _messageQueue.toList().forEach(_sendFailedEvent);
-          _isRetrying = false;
-          return;
-        }
-
-        retryPolicy.attempt++;
-
-        final timeout = retryPolicy.retryTimeout(
-          channel.client,
-          retryPolicy.attempt,
-          apiError,
-        );
-        await Future.delayed(timeout);
-      }
+      await _runAndRetry(message);
     }
     _isRetrying = false;
   }
 
-  void _sendFailedEvent(Message? message) {
-    final newStatus = message!.status == MessageSendingStatus.sending
-        ? MessageSendingStatus.failed
-        : (message.status == MessageSendingStatus.updating
-            ? MessageSendingStatus.failed_update
-            : MessageSendingStatus.failed_delete);
-    channel.state!.addMessage(message.copyWith(
-      status: newStatus,
-    ));
+  Future<void> _runAndRetry(Message message) async {
+    var attempt = 1;
+
+    final maxAttempt = _retryPolicy.maxRetryAttempts;
+    // early return in case maxAttempt is less than 0
+    if (attempt > maxAttempt) return;
+
+    // ignore: literal_only_boolean_expressions
+    while (true) {
+      try {
+        logger?.info('Message (${message.id}) retry attempt $attempt');
+        await _retryMessage(message);
+        logger?.info('Message (${message.id}) sent successfully');
+        _messageQueue.removeMessage(message);
+        break;
+      } on StreamChatError catch (e) {
+        // retry logic
+        final maxAttempt = _retryPolicy.maxRetryAttempts;
+        if (attempt < maxAttempt) {
+          final shouldRetry = _retryPolicy.shouldRetry(client, attempt, e);
+          if (shouldRetry) {
+            final timeout = _retryPolicy.retryTimeout(client, attempt, e);
+            // temporary failure, continue
+            logger?.info(
+              'API call failed (attempt $attempt), '
+              'retrying in ${timeout.inSeconds} seconds. Error was $e',
+            );
+            await Future.delayed(timeout);
+            attempt += 1;
+          } else {
+            logger?.info(
+              'API call failed (attempt $attempt). '
+              'Giving up for now, will retry when connection recovers. '
+              'Error was $e',
+            );
+            _sendFailedEvent(message);
+            break;
+          }
+        } else {
+          logger?.info(
+            'API call failed (attempt $attempt). '
+            'Exceeds maxRetryAttempt : $maxAttempt '
+            'Giving up for now, will retry when connection recovers. '
+            'Error was $e',
+          );
+          _sendFailedEvent(message);
+          break;
+        }
+      } catch (e) {
+        logger?.info(
+          'API call failed due to unknown error (attempt $attempt). '
+          'Giving up for now, will retry when connection recovers. '
+          'Error was $e',
+        );
+        _sendFailedEvent(message);
+        break;
+      }
+    }
   }
 
-  Future<void> _sendMessage(Message message) async {
+  void _sendFailedEvent(Message message) {
+    final newStatus = message.status == MessageSendingStatus.sending
+        ? MessageSendingStatus.failed
+        : message.status == MessageSendingStatus.updating
+            ? MessageSendingStatus.failed_update
+            : MessageSendingStatus.failed_delete;
+    channel.state?.addMessage(message.copyWith(status: newStatus));
+  }
+
+  Future<void> _retryMessage(Message message) async {
     if (message.status == MessageSendingStatus.failed_update ||
         message.status == MessageSendingStatus.updating) {
       await channel.updateMessage(message);
@@ -137,39 +177,10 @@ class RetryQueue {
     }
   }
 
-  void _listenFailedEvents() {
-    _subscriptions.add(channel.on().listen((event) {
-      final messageList = _messageQueue.toList();
-      if (event.message != null) {
-        final messageIndex =
-            messageList.indexWhere((m) => m.id == event.message!.id);
-        if (messageIndex == -1 &&
-            [
-              MessageSendingStatus.failed_update,
-              MessageSendingStatus.failed,
-              MessageSendingStatus.failed_delete,
-            ].contains(event.message!.status)) {
-          logger?.info('add message from events');
-          final m = event.message;
-
-          if (m != null) {
-            add([m]);
-          }
-        } else if (messageIndex != -1 &&
-            [
-              MessageSendingStatus.sent,
-              null,
-            ].contains(event.message!.status)) {
-          _messageQueue.remove(messageList[messageIndex]);
-        }
-      }
-    }));
-  }
-
   /// Call this method to dispose this object
   void dispose() {
     _messageQueue.clear();
-    _subscriptions.forEach((s) => s.cancel());
+    _compositeSubscription.dispose();
   }
 
   static int _byDate(Message m1, Message m2) {
@@ -201,3 +212,27 @@ class RetryQueue {
     }
   }
 }
+
+extension _MessageHeapPriorityQueue on HeapPriorityQueue<Message> {
+  void removeMessage(Message message) {
+    final list = toUnorderedList();
+    final index = list.indexWhere((it) => it.id == message.id);
+    if (index == -1) return;
+    final element = list[index];
+    remove(element);
+  }
+
+  bool containsMessage(Message message) {
+    final list = toUnorderedList();
+    final index = list.indexWhere((it) => it.id == message.id);
+    if (index == -1) return false;
+    return true;
+  }
+
+  bool containsAllMessage(List<Message> messages) {
+    if (isEmpty) return false;
+    final list = toUnorderedList();
+    final messageIds = messages.map((it) => it.id);
+    return list.every((it) => messageIds.contains(it.id));
+  }
+}

packages/stream_chat/lib/src/client.dart

@@ -7,6 +7,8 @@ import 'package:dio/dio.dart';
 import 'package:logging/logging.dart';
 import 'package:rxdart/rxdart.dart';
 import 'package:stream_chat/src/api/channel.dart';
+import 'package:stream_chat/src/core/utils.dart';
+import 'package:stream_chat/src/errors/stream_chat_error.dart';
 import 'package:stream_chat/src/location.dart';
 import 'package:stream_chat/src/ws/connection_status.dart';
 import 'package:stream_chat/src/core/api/attachment_file_uploader.dart';
@@ -92,7 +94,7 @@ class StreamChatClient {
           tokenManager: _tokenManager,
           connectionIdManager: _connectionIdManager,
           attachmentFileUploader: attachmentFileUploader,
-          logger: _detachedLogger('🕸️'),
+          logger: detachedLogger('🕸️'),
         );
 
     _ws = WebSocket(
@@ -100,13 +102,13 @@ class StreamChatClient {
       baseUrl: options.baseUrl,
       tokenManager: _tokenManager,
       handler: handleEvent,
-      logger: _detachedLogger('🔌'),
+      logger: detachedLogger('🔌'),
     );
 
     _retryPolicy = retryPolicy ??
         RetryPolicy(
-          retryTimeout: (_, int attempt, __) => Duration(seconds: 1 * attempt),
-          shouldRetry: (_, int attempt, __) => attempt < 5,
+          shouldRetry: (_, attempt, __) => attempt < 5,
+          retryTimeout: (_, attempt, __) => Duration(seconds: attempt),
         );
 
     state = ClientState(this);
@@ -137,7 +139,7 @@ class StreamChatClient {
   /// Whether the chat persistence is available or not
   bool get persistenceEnabled => _chatPersistenceClient != null;
 
-  RetryPolicy? _retryPolicy;
+  late final RetryPolicy _retryPolicy;
 
   // sync state of the channels present inside state, defaults to false
   bool _synced = false;
@@ -146,7 +148,7 @@ class StreamChatClient {
   DateTime? _lastSyncedAt;
 
   /// The retry policy options getter
-  RetryPolicy? get retryPolicy => _retryPolicy;
+  RetryPolicy get retryPolicy => _retryPolicy;
 
   /// By default the Chat client will write all messages with level Warn or
   /// Error to stdout.
@@ -217,7 +219,8 @@ class StreamChatClient {
         if (record.stackTrace != null) print(record.stackTrace);
       };
 
-  Logger _detachedLogger(
+  ///
+  Logger detachedLogger(
     String name,
   ) =>
       Logger.detached(name)
@@ -303,7 +306,7 @@ class StreamChatClient {
     TokenProvider? provider,
   }) async {
     if (_ws.connectionCompleter?.isCompleted == false) {
-      throw Exception(
+      throw const StreamChatError(
         'User already getting connected, try calling `disconnectUser` '
         'before trying to connect again',
       );
@@ -346,11 +349,11 @@ class StreamChatClient {
     logger.info('Opening web-socket connection for ${user.id}');
 
     if (wsConnectionStatus == ConnectionStatus.connecting) {
-      throw Exception('Connection already in progress for ${user.id}');
+      throw StreamChatError('Connection already in progress for ${user.id}');
     }
 
     if (wsConnectionStatus == ConnectionStatus.connected) {
-      throw Exception('Connection already connected for ${user.id}');
+      throw StreamChatError('Connection already connected for ${user.id}');
     }
 
     _wsConnectionStatus = ConnectionStatus.connecting;
@@ -486,13 +489,6 @@ class StreamChatClient {
     }
   }
 
-  String _generateHash(List<Object?> objects) {
-    final payload = json.encode(objects);
-    final payloadBytes = utf8.encode(payload);
-    final payloadB64 = base64.encode(payloadBytes);
-    return payloadB64;
-  }
-
   final _queryChannelsStreams = <String, Future<List<Channel>>>{};
 
   /// Requests channels with a given query.
@@ -512,7 +508,7 @@ class StreamChatClient {
       watch = false;
     }
 
-    final hash = _generateHash([
+    final hash = generateHash([
       filter,
       sort,
       state,
@@ -575,7 +571,7 @@ class StreamChatClient {
         await _ws.connectionCompleter?.future;
       }
       if (wsConnectionStatus != ConnectionStatus.connected) {
-        throw Exception(
+        throw const StreamChatError(
           'You cannot use queryChannels without an active connection. '
           'Please call `connectUser` to connect the client.',
         );

packages/stream_chat/lib/src/core/http/interceptor/auth_interceptor.dart

@@ -27,12 +27,12 @@ class AuthInterceptor extends Interceptor {
     try {
       token = await _tokenManager.loadToken();
     } catch (_) {
-      final error = StreamChatError(ChatErrorCode.undefinedToken);
+      final error = StreamChatNetworkError(ChatErrorCode.undefinedToken);
       final dioError = StreamChatDioError(
         error: error,
         requestOptions: options,
       );
-      return handler.reject(dioError);
+      return handler.reject(dioError, true);
     }
     final params = {'user_id': token.userId};
     final headers = {

packages/stream_chat/lib/src/core/http/stream_chat_dio_error.dart

@@ -17,5 +17,5 @@ class StreamChatDioError extends DioError {
         );
 
   @override
-  final StreamChatError error;
+  final StreamChatNetworkError error;
 }

packages/stream_chat/lib/src/core/http/stream_http_client.dart

@@ -100,14 +100,15 @@ class StreamHttpClient {
   void close({bool force = false}) => httpClient.close(force: force);
 
   StreamChatNetworkError _parseError(DioError err) {
+    StreamChatNetworkError error;
     // locally thrown dio error
     if (err is StreamChatDioError) {
-      final code = err.error.code;
-      final message = err.error.message;
-      return StreamChatNetworkError.raw(code: code, message: message);
+      error = err.error;
+    } else {
+      // real network request dio error
+      error = StreamChatNetworkError.fromDioError(err);
     }
-    // real network request dio error
-    return StreamChatNetworkError.fromDioError(err);
+    return error..stackTrace = err.stackTrace;
   }
 
   /// Handy method to make http GET request with error parsing.

packages/stream_chat/lib/src/core/models/event.dart

@@ -29,7 +29,7 @@ class Event {
     this.parentId,
     this.extraData = const {},
     this.isLocal = true,
-  }) : createdAt = createdAt ?? DateTime.now();
+  }) : createdAt = createdAt?.toUtc() ?? DateTime.now().toUtc();
 
   /// Create a new instance from a json
   factory Event.fromJson(Map<String, dynamic> json) =>

packages/stream_chat/lib/src/core/utils.dart

@@ -1,3 +1,4 @@
+import 'dart:convert';
 import 'dart:math' as math;
 
 // This alphabet uses `A-Za-z0-9_-` symbols. The genetic algorithm helped
@@ -14,3 +15,11 @@ String randomId({int size = 21}) {
   }
   return id;
 }
+
+/// Creates a hash string from the passed [objects]
+String generateHash(List<Object?> objects) {
+  final payload = json.encode(objects);
+  final payloadBytes = utf8.encode(payload);
+  final payloadB64 = base64.encode(payloadBytes);
+  return payloadB64;
+}

packages/stream_chat/lib/src/errors/chat_error_code.dart

@@ -1,91 +1,166 @@
-///
-enum ChatErrorCode {
-  // client error codes
-  networkFailed,
-  parserError,
-  socketClosed,
-  socketFailure,
-  cantParseConnectionEvent,
-  cantParseEvent,
-  invalidToken,
-  undefinedToken,
-  unableToParseSocketEvent,
-  noErrorBody,
+// ignore_for_file: lines_longer_than_80_chars
 
-  // server error codes
+import 'package:collection/collection.dart';
+
+/// Complete list of errors that are returned by the API
+/// together with the description and API code.
+enum ChatErrorCode {
+  // Client errors
+
+  /// Unauthenticated, token not defined
+  undefinedToken,
+
+  // Bad Request
+
+  /// Wrong data/parameter is sent to the API
+  inputError,
+
+  /// Duplicate username is sent while enforce_unique_usernames is enabled
+  duplicateUsername,
+
+  /// Message is too long
+  messageTooLong,
+
+  /// Event is not supported
+  eventNotSupported,
+
+  /// The feature is currently disabled
+  /// on the dashboard (i.e. Reactions & Replies)
+  channelFeatureNotSupported,
+
+  /// Multiple Levels Reply is not supported
+  /// the API only supports 1 level deep reply threads
+  multipleNestling,
+
+  /// Custom Command handler returned an error
+  customCommandEndpointCall,
+
+  /// App config does not have custom_action_handler_url
+  customCommandEndpointMissing,
+
+  // Unauthorised
+
+  /// Unauthenticated, problem with authentication
   authenticationError,
+
+  /// Unauthenticated, token expired
   tokenExpired,
+
+  /// Unauthenticated, token date incorrect
+  tokenBeforeIssuedAt,
+
+  /// Unauthenticated, token not valid yet
   tokenNotValid,
-  tokenDateIncorrect,
-  tokenSignatureIncorrect,
-  apiKeyNotFound,
+
+  /// Unauthenticated, token signature invalid
+  tokenSignatureInvalid,
+
+  /// Access Key invalid
+  accessKeyError,
+
+  // Forbidden
+
+  /// Unauthorised / forbidden to make request
+  notAllowed,
+
+  /// App suspended
+  appSuspended,
+
+  /// User tried to post a message during the cooldown period
+  cooldownError,
+
+  // Miscellaneous
+
+  /// Resource not found
+  doesNotExist,
+
+  /// Request timed out
+  requestTimeout,
+
+  /// Payload too big
+  payloadTooBig,
+
+  /// Too many requests in a certain time frame
+  rateLimitError,
+
+  /// Request headers are too large
+  maximumHeaderSizeExceeded,
+
+  /// Something goes wrong in the system
+  internalSystemError,
+
+  /// No access to requested channels
+  noAccessToChannels
 }
 
+const _errorCodeWithDescription = {
+  ChatErrorCode.undefinedToken:
+      MapEntry(1000, 'Unauthorised, token not defined'),
+  ChatErrorCode.inputError:
+      MapEntry(4, 'Wrong data/parameter is sent to the API'),
+  ChatErrorCode.duplicateUsername: MapEntry(6,
+      'Duplicate username is sent while enforce_unique_usernames is enabled'),
+  ChatErrorCode.messageTooLong: MapEntry(20, 'Message is too long'),
+  ChatErrorCode.eventNotSupported: MapEntry(18, 'Event is not supported'),
+  ChatErrorCode.channelFeatureNotSupported: MapEntry(19,
+      'The feature is currently disabled on the dashboard (i.e. Reactions & Replies)'),
+  ChatErrorCode.multipleNestling: MapEntry(21,
+      'Multiple Levels Reply is not supported - the API only supports 1 level deep reply threads'),
+  ChatErrorCode.customCommandEndpointCall:
+      MapEntry(45, 'Custom Command handler returned an error'),
+  ChatErrorCode.customCommandEndpointMissing:
+      MapEntry(44, 'App config does not have custom_action_handler_url'),
+  ChatErrorCode.authenticationError:
+      MapEntry(5, 'Unauthenticated, problem with authentication'),
+  ChatErrorCode.tokenExpired: MapEntry(40, 'Unauthenticated, token expired'),
+  ChatErrorCode.tokenBeforeIssuedAt:
+      MapEntry(42, 'Unauthenticated, token date incorrect'),
+  ChatErrorCode.tokenNotValid:
+      MapEntry(41, 'Unauthenticated, token not valid yet'),
+  ChatErrorCode.tokenSignatureInvalid:
+      MapEntry(43, 'Unauthenticated, token signature invalid'),
+  ChatErrorCode.accessKeyError: MapEntry(2, 'Access Key invalid'),
+  ChatErrorCode.notAllowed:
+      MapEntry(17, 'Unauthorised / forbidden to make request'),
+  ChatErrorCode.appSuspended: MapEntry(99, 'App suspended'),
+  ChatErrorCode.cooldownError:
+      MapEntry(60, 'User tried to post a message during the cooldown period'),
+  ChatErrorCode.doesNotExist: MapEntry(16, 'Resource not found'),
+  ChatErrorCode.requestTimeout: MapEntry(23, 'Request timed out'),
+  ChatErrorCode.payloadTooBig: MapEntry(22, 'Payload too big'),
+  ChatErrorCode.rateLimitError:
+      MapEntry(9, 'Too many requests in a certain time frame'),
+  ChatErrorCode.maximumHeaderSizeExceeded:
+      MapEntry(24, 'Request headers are too large'),
+  ChatErrorCode.internalSystemError:
+      MapEntry(-1, 'Something goes wrong in the system'),
+  ChatErrorCode.noAccessToChannels:
+      MapEntry(70, 'No access to requested channels'),
+};
+
+const _authenticationErrors = [
+  ChatErrorCode.undefinedToken,
+  ChatErrorCode.authenticationError,
+  ChatErrorCode.tokenExpired,
+  ChatErrorCode.tokenBeforeIssuedAt,
+  ChatErrorCode.tokenNotValid,
+  ChatErrorCode.tokenSignatureInvalid,
+  ChatErrorCode.accessKeyError,
+  ChatErrorCode.noAccessToChannels,
+];
+
+///
+ChatErrorCode? chatErrorCodeFromCode(int code) => _errorCodeWithDescription.keys
+    .firstWhereOrNull((key) => _errorCodeWithDescription[key]!.key == code);
+
 ///
 extension ChatErrorCodeX on ChatErrorCode {
   ///
-  String get message => {
-        // client error message
-        ChatErrorCode.networkFailed: 'Response is failed. See cause',
-        ChatErrorCode.parserError: 'Unable to parse error',
-        ChatErrorCode.socketClosed: 'Server closed connection',
-        ChatErrorCode.socketFailure: 'See stack trace in logs. '
-            'Intercept error in error handler of setUser',
-        ChatErrorCode.cantParseConnectionEvent:
-            'Unable to parse connection event',
-        ChatErrorCode.cantParseEvent: 'Unable to parse event',
-        ChatErrorCode.invalidToken: 'Invalid token',
-        ChatErrorCode.undefinedToken:
-            'No defined token. Check if client.setUser was called and finished',
-        ChatErrorCode.unableToParseSocketEvent:
-            'Socket event payload either invalid or null',
-        ChatErrorCode.noErrorBody: 'No error body. See http status code',
-
-        // server error message
-        ChatErrorCode.authenticationError:
-            'Unauthenticated, problem with authentication',
-        ChatErrorCode.tokenExpired: 'Token expired, new one must be requested.',
-        ChatErrorCode.tokenNotValid: 'Unauthenticated, token not valid yet',
-        ChatErrorCode.tokenDateIncorrect:
-            'Unauthenticated, token date incorrect',
-        ChatErrorCode.tokenSignatureIncorrect:
-            'Unauthenticated, token signature invalid',
-        ChatErrorCode.apiKeyNotFound:
-            "Api key is not found, verify it if it's correct or was created.",
-      }[this]!;
+  String get message => _errorCodeWithDescription[this]!.value;
 
   ///
-  int get code => {
-        // client error codes
-        ChatErrorCode.networkFailed: 1000,
-        ChatErrorCode.parserError: 1001,
-        ChatErrorCode.socketClosed: 1002,
-        ChatErrorCode.socketFailure: 1003,
-        ChatErrorCode.cantParseConnectionEvent: 1004,
-        ChatErrorCode.cantParseEvent: 1005,
-        ChatErrorCode.invalidToken: 1006,
-        ChatErrorCode.undefinedToken: 1007,
-        ChatErrorCode.unableToParseSocketEvent: 1008,
-        ChatErrorCode.noErrorBody: 1009,
-
-        // server error codes
-        ChatErrorCode.authenticationError: 5,
-        ChatErrorCode.tokenExpired: 40,
-        ChatErrorCode.tokenNotValid: 41,
-        ChatErrorCode.tokenDateIncorrect: 42,
-        ChatErrorCode.tokenSignatureIncorrect: 43,
-        ChatErrorCode.apiKeyNotFound: 2,
-      }[this]!;
+  int get code => _errorCodeWithDescription[this]!.key;
 
   ///
-  Set<int> get authenticationErrors => {
-        ChatErrorCode.authenticationError,
-        ChatErrorCode.tokenExpired,
-        ChatErrorCode.tokenNotValid,
-        ChatErrorCode.tokenDateIncorrect,
-        ChatErrorCode.tokenSignatureIncorrect,
-      }.map((it) => it.code).toSet();
-
-  ///
-  bool isAuthenticationError(int code) => authenticationErrors.contains(code);
+  bool get isAuthenticationError => _authenticationErrors.contains(this);
 }

packages/stream_chat/lib/src/errors/stream_chat_error.dart

@@ -3,45 +3,74 @@ import 'package:stream_chat/src/errors/chat_error_code.dart';
 import 'package:stream_chat/stream_chat.dart';
 
 ///
-class StreamChatError extends Equatable implements Exception {
+class StreamChatError with EquatableMixin implements Exception {
   ///
-  StreamChatError(ChatErrorCode errorCode)
-      : code = errorCode.code,
-        message = errorCode.message;
-
-  ///
-  const StreamChatError.raw(this.code, this.message);
-
-  /// Error code
-  final int code;
+  const StreamChatError(this.message);
 
   /// Error message
   final String message;
 
   @override
-  List<Object?> get props => [code, message];
+  List<Object?> get props => [message];
 
   @override
-  String toString() => 'StreamChatError(code: $code, message: $message)';
+  String toString() => 'StreamChatError(message: $message)';
+}
+
+///
+class StreamWebSocketError extends StreamChatError {
+  ///
+  const StreamWebSocketError(
+    String message, {
+    this.data,
+  }) : super(message);
+
+  ///
+  factory StreamWebSocketError.fromStreamError(Map<String, Object?> error) {
+    final data = ErrorResponse.fromJson(error);
+    final message = data.message ?? '';
+    return StreamWebSocketError(message, data: data);
+  }
+
+  ///
+  int? get code => data?.code;
+
+  ///
+  ChatErrorCode? get errorCode {
+    final code = this.code;
+    if (code == null) return null;
+    return chatErrorCodeFromCode(code);
+  }
+
+  /// Response body. please refer to [ErrorResponse].
+  final ErrorResponse? data;
+
+  @override
+  String toString() {
+    var params = 'message: $message';
+    if (data != null) params += ', data: $data';
+    return 'WebSocketError($params)';
+  }
 }
 
 ///
 class StreamChatNetworkError extends StreamChatError {
   ///
-  StreamChatNetworkError({
-    required ChatErrorCode errorCode,
+  StreamChatNetworkError(
+    ChatErrorCode errorCode, {
     int? statusCode,
     this.data,
-  })  : statusCode = statusCode ?? data?.statusCode,
-        super(errorCode);
+  })  : code = errorCode.code,
+        statusCode = statusCode ?? data?.statusCode,
+        super(errorCode.message);
 
   ///
-  const StreamChatNetworkError.raw({
-    required int code,
+  StreamChatNetworkError.raw({
+    required this.code,
     required String message,
     this.statusCode,
     this.data,
-  }) : super.raw(code, message);
+  }) : super(message);
 
   ///
   factory StreamChatNetworkError.fromDioError(DioError error) {
@@ -53,25 +82,46 @@ class StreamChatNetworkError extends StreamChatError {
     }
     return StreamChatNetworkError.raw(
       code: errorResponse?.code ?? -1,
-      message: errorResponse?.message ?? response?.statusMessage ?? '',
+      message:
+          errorResponse?.message ?? response?.statusMessage ?? error.message,
       statusCode: errorResponse?.statusCode ?? response?.statusCode,
       data: errorResponse,
-    );
+    )..stackTrace = error.stackTrace;
   }
 
+  /// Error code
+  final int code;
+
   /// HTTP status code
   final int? statusCode;
 
   /// Response body. please refer to [ErrorResponse].
   final ErrorResponse? data;
 
-  @override
-  List<Object?> get props => [...super.props, statusCode];
+  StackTrace? _stackTrace;
+
+  ///
+  set stackTrace(StackTrace? stack) => _stackTrace = stack;
+
+  ///
+  ChatErrorCode? get errorCode => chatErrorCodeFromCode(code);
+
+  ///
+  bool get isRetriable => data == null;
 
   @override
-  String toString() => 'StreamChatNetworkError('
-      'code: $code, '
-      'message: $message, '
-      'statusCode: $statusCode, '
-      'data: $data)';
+  List<Object?> get props => [...super.props, code, statusCode];
+
+  @override
+  String toString({bool printStackTrace = false}) {
+    var params = 'code: $code, message: $message';
+    if (statusCode != null) params += ', statusCode: $statusCode';
+    if (data != null) params += ', data: $data';
+    var msg = 'StreamChatNetworkError($params)';
+
+    if (printStackTrace && _stackTrace != null) {
+      msg += '\n$_stackTrace';
+    }
+    return msg;
+  }
 }

packages/stream_chat/lib/src/exceptions.dart

@@ -1,53 +0,0 @@
-import 'dart:convert';
-
-/// Exception related to api calls
-class ApiError extends Error {
-  /// Creates a new ApiError instance using the response body and status code
-  ApiError(this.body, this.status) : jsonData = _decode(body) {
-    if (jsonData != null && jsonData!.containsKey('code')) {
-      _code = jsonData!['code'];
-    }
-  }
-
-  /// Raw body of the response
-  final String? body;
-
-  /// Json parsed body
-  final Map<String, dynamic>? jsonData;
-
-  /// Http status code of the response
-  final int? status;
-
-  /// Stream specific error code
-  int? get code => _code;
-  int? _code;
-
-  static Map<String, dynamic>? _decode(String? body) {
-    try {
-      if (body == null) {
-        return null;
-      }
-      return json.decode(body);
-    } on FormatException {
-      return null;
-    }
-  }
-
-  @override
-  bool operator ==(Object other) =>
-      identical(this, other) ||
-      other is ApiError &&
-          runtimeType == other.runtimeType &&
-          body == other.body &&
-          jsonData == other.jsonData &&
-          status == other.status &&
-          _code == other._code;
-
-  @override
-  int get hashCode =>
-      body.hashCode ^ jsonData.hashCode ^ status.hashCode ^ _code.hashCode;
-
-  @override
-  String toString() => 'ApiError{body: $body, jsonData: $jsonData, '
-      'status: $status, code: $_code}';
-}

packages/stream_chat/lib/src/ws/socket_error.dart

@@ -1,25 +0,0 @@
-///
-class SocketError {
-  ///
-  const SocketError({
-    this.code = -1,
-    this.statusCode = -1,
-    this.message = '',
-  });
-
-  ///
-  factory SocketError.fromJson(Map<String, dynamic> json) => SocketError(
-        code: json['code'],
-        statusCode: json['StatusCode'],
-        message: json['message'],
-      );
-
-  ///
-  final int code;
-
-  ///
-  final int statusCode;
-
-  ///
-  final String message;
-}

packages/stream_chat/lib/src/ws/websocket.dart

@@ -5,13 +5,14 @@ import 'dart:math' as math;
 import 'package:logging/logging.dart';
 import 'package:meta/meta.dart';
 import 'package:rxdart/rxdart.dart';
+import 'package:stream_chat/src/errors/chat_error_code.dart';
+import 'package:stream_chat/src/errors/stream_chat_error.dart';
 import 'package:stream_chat/src/ws/connection_status.dart';
 import 'package:stream_chat/src/core/http/token.dart';
 import 'package:stream_chat/src/core/http/token_manager.dart';
 import 'package:stream_chat/src/core/models/event.dart';
 import 'package:stream_chat/src/core/models/user.dart';
 import 'package:stream_chat/src/event_type.dart';
-import 'package:stream_chat/src/ws/socket_error.dart';
 import 'package:stream_chat/src/ws/timer_helper.dart';
 import 'package:web_socket_channel/web_socket_channel.dart';
 import 'package:web_socket_channel/status.dart' as status;
@@ -27,8 +28,6 @@ typedef WebSocketChannelProvider = WebSocketChannel Function(
   Iterable<String>? protocols,
 });
 
-const _tokenExpiredErrorCode = 40;
-
 /// A WebSocket connection that reconnects upon failure.
 class WebSocket with TimerHelper {
   /// Creates a new websocket
@@ -174,7 +173,7 @@ class WebSocket with TimerHelper {
   /// Connect the WS using the parameters passed in the constructor
   Future<Event> connect(User user) async {
     if (_connectRequestInProgress) {
-      throw Exception('''
+      throw const StreamWebSocketError('''
         You've called connect twice,
         can only attempt 1 connection at the time,
         ''');
@@ -195,7 +194,7 @@ class WebSocket with TimerHelper {
   int _reconnectAttempt = 0;
   bool _reconnectRequestInProgress = false;
 
-  Future<void> _reconnect({bool refreshToken = false}) async {
+  void _reconnect({bool refreshToken = false}) async {
     _logger?.info('Retrying connection : $_reconnectAttempt');
     if (_reconnectRequestInProgress) return;
     _reconnectRequestInProgress = true;
@@ -270,10 +269,35 @@ class WebSocket with TimerHelper {
     _connectionStatus = ConnectionStatus.connected;
   }
 
+  void _handleStreamError(Map<String, Object?> errorResponse) {
+    // resetting connect, reconnect request flag
+    _resetRequestFlags();
+
+    final error = StreamWebSocketError.fromStreamError(errorResponse);
+    final isTokenExpired = error.errorCode == ChatErrorCode.tokenExpired;
+    if (isTokenExpired && !tokenManager.isStatic) {
+      _logger?.warning('Connection failed, token expired');
+      return _reconnect(refreshToken: true);
+    }
+
+    _logger?.severe('Connection failed', error);
+
+    final completer = connectionCompleter;
+    // complete with error if not yet completed
+    if (completer != null && !completer.isCompleted) {
+      // complete the connection with error
+      completer.completeError(error);
+      // disconnect the web-socket connection
+      return disconnect();
+    }
+
+    return _reconnect();
+  }
+
   void _onDataReceived(dynamic data) {
-    final jsonData = json.decode(data);
-    final error = jsonData['error'];
-    if (error != null) return _onConnectionError(error);
+    final jsonData = json.decode(data) as Map<String, Object?>;
+    final error = jsonData['error'] as Map<String, Object?>?;
+    if (error != null) return _handleStreamError(error);
 
     // resetting connect, reconnect request flag
     _resetRequestFlags(resetAttempts: true);
@@ -300,25 +324,18 @@ class WebSocket with TimerHelper {
   }
 
   void _onConnectionError(error, [stacktrace]) {
-    _logger?.severe('Error occurred', error, stacktrace);
+    _logger?.warning('Error occurred', error, stacktrace);
 
     // resetting connect, reconnect request flag
     _resetRequestFlags();
 
-    var refreshToken = false;
-    try {
-      final socketError = SocketError.fromJson(json.decode(error));
-      refreshToken = socketError.code == _tokenExpiredErrorCode;
-    } catch (_) {}
-
-    // refresh token in case it is expired
-    _reconnect(refreshToken: refreshToken);
+    _reconnect();
   }
 
   bool _manuallyClosed = false;
 
   void _onConnectionClosed() {
-    _logger?.info('Connection closed : $connectionId');
+    _logger?.warning('Connection closed : $connectionId');
 
     // resetting connect, reconnect request flag
     _resetRequestFlags();
@@ -381,7 +398,7 @@ class WebSocket with TimerHelper {
     if (connectionStatus == ConnectionStatus.disconnected) return;
     _connectionStatus = ConnectionStatus.disconnected;
 
-    _logger?.info('Disconnecting $connectionId');
+    _logger?.info('Disconnecting web-socket connection');
 
     // resetting user
     _user = null;

packages/stream_chat/lib/stream_chat.dart

@@ -32,8 +32,9 @@ export './src/core/models/reaction.dart';
 export './src/core/models/read.dart';
 export './src/core/models/user.dart';
 export './src/db/chat_persistence_client.dart';
+export './src/errors/chat_error_code.dart';
+export './src/errors/stream_chat_error.dart';
 export './src/event_type.dart';
-export './src/exceptions.dart';
 export './src/extensions/rate_limit.dart';
 export './src/extensions/string_extension.dart';
 export './src/location.dart';

packages/stream_chat/test/src/client_test.dart

@@ -8,7 +8,6 @@ import 'package:logging/logging.dart';
 import 'package:mocktail/mocktail.dart';
 import 'package:stream_chat/src/core/api/requests.dart';
 import 'package:stream_chat/src/client.dart';
-import 'package:stream_chat/src/exceptions.dart';
 import 'package:stream_chat/src/core/models/channel_model.dart';
 import 'package:stream_chat/src/core/models/filter.dart';
 import 'package:stream_chat/src/core/models/message.dart';

packages/stream_chat_flutter/lib/src/message_actions_modal.dart

@@ -341,7 +341,8 @@ class _MessageActionsModalState extends State<MessageActionsModal> {
           okText: 'OK',
         );
       } catch (err) {
-        if (err is ApiError && json.decode(err.body ?? '{}')['code'] == 4) {
+        if (err is StreamChatNetworkError &&
+            err.errorCode == ChatErrorCode.inputError) {
           await showInfoDialog(
             context,
             icon: StreamSvgIcon.flag(

packages/stream_chat_persistence/lib/src/db/moor_chat_database.dart

@@ -51,7 +51,7 @@ class MoorChatDatabase extends _$MoorChatDatabase {
 
   // you should bump this number whenever you change or add a table definition.
   @override
-  int get schemaVersion => 3;
+  int get schemaVersion => 4;
 
   @override
   MigrationStrategy get migration => MigrationStrategy(