import 'dart:async'; import 'dart:convert'; import 'package:firebase_crashlytics/firebase_crashlytics.dart'; import 'package:flutter/material.dart'; import 'package:flutter/services.dart'; import 'package:frosty/apis/twitch_api.dart'; import 'package:frosty/models/badges.dart'; import 'package:frosty/models/emotes.dart'; import 'package:frosty/models/events.dart'; import 'package:frosty/models/irc.dart'; import 'package:frosty/screens/channel/chat/details/chat_details_store.dart'; import 'package:frosty/screens/channel/chat/stores/banned_user_tracker.dart'; import 'package:frosty/screens/channel/chat/stores/chat_assets_store.dart'; import 'package:frosty/screens/channel/chat/stores/chat_interaction_pause.dart'; import 'package:frosty/screens/channel/chat/stores/shared_chat_bubble_state.dart'; import 'package:frosty/screens/settings/stores/auth_store.dart'; import 'package:frosty/screens/settings/stores/settings_store.dart'; import 'package:frosty/utils.dart'; import 'package:mobx/mobx.dart'; import 'package:web_socket_channel/io.dart'; import 'package:web_socket_channel/web_socket_channel.dart'; part 'chat_store.g.dart'; /// The store and view-model for chat-related activities. class ChatStore = ChatStoreBase with _$ChatStore; abstract class ChatStoreBase with Store { /// The total maximum amount of messages in chat. static const _messageLimit = 5000; /// The maximum amount of messages to render when autoscroll is enabled. static const _renderMessageLimit = 100; /// IRC commands that should bypass chat delay. /// These are confirmations/state updates rather than chat messages to sync. static const _delayBypassCommands = { 'USERSTATE', 'GLOBALUSERSTATE', 'ROOMSTATE', 'NOTICE', // Rejection notices need immediate feedback }; /// NOTICE msg-id values indicating message was rejected (from Twitch IRC docs). /// https://dev.twitch.tv/docs/irc/msg-id/ static const _rejectionMsgIds = { // Rate limiting 'msg_slowmode', 'msg_ratelimit', 'msg_duplicate', // User restrictions 'msg_banned', 'msg_timedout', 'msg_channel_blocked', 'msg_suspended', // Room mode restrictions 'msg_emoteonly', 'msg_subsonly', 'msg_followersonly', 'msg_followersonly_followed', 'msg_followersonly_zero', 'msg_r9k', // Verification requirements 'msg_verified_email', 'msg_requires_verified_phone_number', // Moderation 'msg_rejected', 'msg_rejected_mandatory', 'msg_bad_characters', }; /// Fallback height used to pad the message list for one frame, until the /// real bottom bar reports its measured height via [setBottomBarHeight]. static const _baseBottomBarHeight = 68.0; /// Server-initiated graceful reconnect notice. Twitch sends this untagged /// before maintenance so clients can re-establish ahead of the close. static const _serverReconnectNotice = ':tmi.twitch.tv RECONNECT'; /// Muted words pre-normalized (trimmed, lowercased, empties dropped) by a /// reaction on the setting — avoids re-normalizing on every message. var _normalizedMutedWords = []; /// Whether a single IRC [line] should bypass the chat delay and be handled /// immediately. Displayable chat content (PRIVMSG/USERNOTICE and the /// moderation CLEARCHAT/CLEARMSG) is delayed, so it stays in sync with the /// delayed video. Everything else is handled at once: control/state messages /// (USERSTATE/ROOMSTATE/GLOBALUSERSTATE), NOTICE (so a send rejection surfaces /// immediately instead of one chat-delay later), the reconnect notice, and /// any untagged server line (PING keepalives, CAP, JOIN, the welcome banner) /// — delaying those would stall the send-ack, drop the PONG, or never start /// the buffer drain. bool _lineShouldBypassDelay(String line) { if (line.isEmpty || line == _serverReconnectNotice) return true; // Displayable chat content is always IRCv3-tagged; an untagged line is // never delayed. if (!line.startsWith('@')) return true; final parts = line.split(' '); return parts.length >= 3 && _delayBypassCommands.contains(parts[2]); } Duration get _chatDelay => Duration(milliseconds: (settings.effectiveChatDelay * 1000).round()); bool get _hasActiveChatDelay => settings.showVideo && _chatDelay > Duration.zero; final TwitchApi twitchApi; /// The amount of messages to free (remove) when the [_messageLimit] is reached. final _messagesToRemove = (_messageLimit * 0.2).toInt(); /// The provided auth store to determine login status, get the token, and use the headers for requests. final AuthStore auth; /// The provided setting store to account for any user-defined behaviors. final SettingsStore settings; /// The focus node for the textfield to allow for showing and hiding the keyboard/focus. final textFieldFocusNode = FocusNode(); /// The name of the channel to connect to. final String channelName; /// The channel's ID for API requests. final String channelId; /// The channel's display name to show on widgets. final String displayName; var _shouldDisconnect = false; /// The Twitch IRC WebSocket channel. WebSocketChannel? _channel; /// The subscription that handles the WebSocket connection. StreamSubscription? _channelListener; WebSocketChannel? _sevenTVChannel; /// The subscription that handles the 7TV WebSocket connection. StreamSubscription? _sevenTVChannelListener; static const _maxRetries = 5; /// Ping interval for both the IRC and 7TV WebSocket sockets. Catches /// zombie connections (silent network changes, OS suspends) without /// waking the radio every 15s. Twitch's own recommended keepalive tops /// out at 600s; community libs (e.g. dank-twitch-irc) sit around 30s. static const _pingInterval = Duration(seconds: 60); // The retry counter for exponential backoff. var _retries = 0; // The current time to wait between retries for exponential backoff. var _backoffTime = 0; // Reference to the reconnect message for in-place updates. IRCMessage? _reconnectMessage; // Timestamp when reconnection started, for calculating elapsed time. DateTime? _reconnectStartTime; /// The scroll controller that controls auto-scroll and resume-scroll behavior. final scrollController = ScrollController(); /// The text controller that handles the TextField inputs and sending of messages. final textController = TextEditingController(); /// The chat details store responsible for the chat modes and users in chat. final ChatDetailsStore chatDetailsStore; /// The assets store responsible for badges, emotes, and the emote menu. final ChatAssetsStore assetsStore; /// Requested message to be sent by the user. Will only be sent on receipt of a USERNOTICE command. IRCMessage? toSend; /// Raw text of the most recent outgoing message, retained until the send is /// confirmed. Lets the USERSTATE handler build the local echo even when the /// message was sent before this session's first USERSTATE arrived (so /// [toSend] couldn't be built up front). String? _pendingSentMessageText; /// The list of reaction disposer functions that will be used later when disposing. final reactions = []; /// The periodic timer used for batching chat message re-renders. Timer? _messageBufferTimer; /// The list of chat messages to add once autoscroll is resumed. /// This is used as an optimization to prevent the list from being updated/shifted while the user is scrolling. final messageBuffer = ObservableList(); /// Pending delayed IRC payloads waiting to be rendered. final _pendingChatCallbacks = <_DelayedCallback>[]; /// Pending delayed 7TV payloads waiting to be rendered. final _pendingSevenTVCallbacks = <_DelayedCallback>[]; /// The set of message IDs that have been revealed by the user (for deleted messages). final revealedMessageIds = ObservableSet(); /// User-driven spotlight/hide selections for shared chat participant /// channels. See [SharedChatBubbleState]. final sharedChatBubbles = SharedChatBubbleState(); @action void revealMessage(String id) { revealedMessageIds.add(id); } /// Whether a message from [sourceChannelId] should render faded. /// /// [overrideCurrentChannelId] is set only when rendering the manual /// "merged mode" view (multiple manually-added tabs interleaved), which /// always fades non-active-tab messages, gated by [focusCurrentChannel]. /// Otherwise this is the normal single-connection view, where native /// Twitch shared chat defaults to unfaded and only fades once the user /// spotlights a participant channel via a chat bubble — an explicit /// action that isn't gated by that setting. bool shouldFadeMessage({ required String? sourceChannelId, required String? overrideCurrentChannelId, }) { if (sourceChannelId == null) return false; final currentChannelId = overrideCurrentChannelId ?? channelId; if (sourceChannelId == currentChannelId) return false; final isMerged = overrideCurrentChannelId != null; if (isMerged) return settings.focusCurrentChannel; return sharedChatBubbles.focusedChannelIds.isNotEmpty && !sharedChatBubbles.isFocused(sourceChannelId); } /// Timer used for dismissing the notification. Timer? _notificationTimer; /// Timer used for resetting the sending state if no acknowledgment is received. Timer? _sendingTimeoutTimer; /// Timer used for updating the chat delay countdown message. Timer? _chatDelayCountdownTimer; /// Reference to the current countdown message for direct updates (avoids O(n) search). IRCMessage? _countdownMessage; /// Tracks whether the initial chat delay sync has completed for the current session. /// Prevents the countdown from re-appearing when latency updates arrive after initial sync. bool _chatDelaySyncCompleted = false; /// The current timer for the sleep timer if active. Timer? sleepTimer; /// The time remaining for the sleep timer. @observable var timeRemaining = const Duration(); /// A notification message to display above the chat. @readonly String? _notification; /// The list of chat messages to render and display. @readonly var _messages = ObservableList(); /// The list of chat messages that should be rendered. Used to prevent jank when resuming scroll. @computed List get renderMessages { // If autoscroll is disabled, render ALL messages in chat. // The second condition is to prevent an out-of-range index error with sublist. final base = (!_autoScroll || _messages.length < _renderMessageLimit) ? _messages : _messages.sublist(_messages.length - _renderMessageLimit); // Filter out messages from shared chat participant channels the user // has hidden. Short-circuit when nothing is hidden (the common case). if (sharedChatBubbles.hiddenChannelIds.isEmpty) return base; return base.where((message) { final sourceChannelId = message.sourceChannelId; return sourceChannelId == null || !sharedChatBubbles.hiddenChannelIds.contains(sourceChannelId); }).toList(); } /// If the chat should automatically scroll/jump to the latest message. @readonly var _autoScroll = true; /// Tracks users banned/timed-out via CLEARCHAT so the moderation menu only /// offers an unban action when it applies. final _bannedUserTracker = BannedUserTracker(); /// Whether [userId] is currently banned or within an unexpired timeout /// (as observed from CLEARCHAT in this session). bool isUserBannedOrTimedOut(String userId) => _bannedUserTracker.isBannedOrTimedOut(userId, DateTime.now()); /// Forget a user's ban state after a successful unban so the action hides. void clearUserBanState(String userId) => _bannedUserTracker.clear(userId); /// Freezes the rendered list while the user is touching it, so a long-press /// target can't scroll out from under the finger and get its gesture /// recognizer disposed mid-press. Plain field (not observable): it only gates /// the periodic [addMessages] flush, which reads it each 200ms tick. final _interactionPause = ChatInteractionPause(); /// Hooks for the message list to report touch start/move/end (see [Chat]). void onMessageListPointerDown(int pointer, Offset position) => _interactionPause.pointerDown(pointer, position); void onMessageListPointerMove(int pointer, Offset position) => _interactionPause.pointerMove(pointer, position); void onMessageListPointerUp(int pointer) => _interactionPause.pointerUp(pointer); @readonly var _inputText = ''; @readonly var _showSendButton = false; @readonly var _showEmoteAutocomplete = false; @readonly var _showMentionAutocomplete = false; /// Whether the chat is currently in shared chat mode (based on source-room-id tag presence). @readonly var _isInSharedChatMode = false; /// Whether we're waiting for server acknowledgment of a sent message. @readonly var _isWaitingForAck = false; /// Whether the chat WebSocket is currently connected. @readonly var _isConnected = false; /// Whether we've successfully connected at least once (to distinguish /// "connecting" from "disconnected" in the UI). @readonly var _hasConnected = false; /// The logged-in user's appearance in chat. @readonly var _userState = const USERSTATE(); @observable var expandChat = false; @observable IRCMessage? replyingToMessage; /// Emotes matching the current autocomplete search term. @computed List get matchingEmotes { if (!_showEmoteAutocomplete) return const []; final searchTerm = _inputText.split(' ').last.toLowerCase(); if (searchTerm.isEmpty) return const []; return [ ...assetsStore.userEmoteToObject.values, ...assetsStore.bttvEmotes, ...assetsStore.ffzEmotes, ...assetsStore.sevenTVEmotes, ].where((emote) => emote.name.toLowerCase().contains(searchTerm)).toList(); } /// Chatters matching the current mention autocomplete search term. @computed List get matchingChatters { if (!_showMentionAutocomplete) return const []; final searchTerm = _inputText .split(' ') .last .replaceFirst('@', '') .toLowerCase(); return chatDetailsStore.chatUsers .where((chatter) => chatter.contains(searchTerm)) .toList(); } /// Measured height of the rendered bottom bar (input plus any reply or /// autocomplete overlays). The bar reports this via [setBottomBarHeight] so /// the message list's bottom padding tracks the bar's *true* height — /// including multi-line input growth and OS text scaling, which fixed /// per-overlay constants silently undercount. final _measuredBottomBarHeight = Observable(null); /// Current bottom bar height, used to dynamically pad the message list so the /// bar never overlaps messages. Falls back to the single-line base height /// until the real bar reports its first measurement. @computed double get bottomBarHeight => _measuredBottomBarHeight.value ?? _baseBottomBarHeight; /// Reports the bar's real rendered height. Sub-pixel changes are ignored to /// avoid needless message-list relayouts. void setBottomBarHeight(double height) { final current = _measuredBottomBarHeight.value; if (current != null && (current - height).abs() < 0.5) return; runInAction(() => _measuredBottomBarHeight.value = height); } ChatStoreBase({ required this.twitchApi, required this.auth, required this.chatDetailsStore, required this.assetsStore, required this.settings, required this.channelName, required this.channelId, required this.displayName, }) { // Create a reaction that will reconnect to chat when logging in or out. // Closing the channel will trigger a reconnect with the new credentials. reactions.add( reaction((_) => auth.isLoggedIn, (_) => _channel?.sink.close(1000)), ); // Normalize muted words once per settings change instead of per message // — the filter runs in the IRC hot path (50+ msgs/s in busy channels). reactions.add( reaction((_) => settings.mutedWords.toList(), (List words) { _normalizedMutedWords = words .map((word) => word.trim().toLowerCase()) .where((word) => word.isNotEmpty) .toList(); }, fireImmediately: true), ); reactions.add( reaction( (_) => [ settings.showTwitchEmotes, settings.showTwitchBadges, settings.show7TVEmotes, settings.showBTTVEmotes, settings.showBTTVBadges, settings.showFFZEmotes, settings.showFFZBadges, ], (_) { if (!settings.show7TVEmotes) { _sevenTVChannel?.sink.close(1000); } getAssets().then((_) { if (settings.show7TVEmotes) { final emoteSetId = assetsStore.sevenTvEmoteSetId; if (emoteSetId != null) { listenToSevenTVEmoteSet(emoteSetId: emoteSetId); } } }); }, ), ); // Start chat delay countdown when toggling video on, cancel when off reactions.add( reaction((_) => settings.showVideo, (showVideo) { if (showVideo && settings.effectiveChatDelay > 0) { _startChatDelayCountdown(); } else if (!showVideo) { _flushPendingDelayedCallbacks(); _cancelChatDelayCountdown(); _chatDelaySyncCompleted = false; } }), ); // Start chat delay countdown when chatDelay is set (for auto sync mode) // Cancel countdown if delay becomes 0 reactions.add( reaction((_) => settings.effectiveChatDelay, (chatDelay) { if (chatDelay == 0) { _flushPendingDelayedCallbacks(); _cancelChatDelayCountdown(); _chatDelaySyncCompleted = false; } else if (settings.autoSyncChatDelay && settings.showVideo && chatDelay > 0 && !_chatDelaySyncCompleted) { _startChatDelayCountdown(); } }), ); assetsStore.init(); _messages.add(IRCMessage.createNotice(message: 'Connecting to chat...')); if (settings.showRecentMessages) { getRecentMessage().then((_) => connectToChat()); } else { connectToChat(); } // Refresh which channels the user moderates on entry so mod actions reflect // status granted/revoked since login. Fire-and-forget, like blocked users. if (auth.isLoggedIn) auth.user.refreshModeratedChannels(); // Tell the scrollController to determine when auto-scroll should be enabled or disabled. scrollController.addListener(() { // If the scroll position is at the latest message (maximum possible), enable autoscroll. // Else if the position is before the latest message (not at the edge), disable autoscroll. if (scrollController.position.pixels <= 0) { _autoScroll = true; } else if (scrollController.position.pixels > 0) { _autoScroll = false; } }); textFieldFocusNode.addListener(() { if (textFieldFocusNode.hasFocus) { // Hide the emote menu if it is currently shown. if (assetsStore.showEmoteMenu) assetsStore.showEmoteMenu = false; } // Un-expand the chat when unfocusing. if (!textFieldFocusNode.hasFocus) expandChat = false; }); // Add a listener to the textfield that will show/hide the autocomplete bar if focused. // Will also rebuild the autocomplete bar when typing, refreshing the results as the user types. textController.addListener(() { _inputText = textController.text; _showEmoteAutocomplete = !_showMentionAutocomplete && textFieldFocusNode.hasFocus && textController.text.split(' ').last.isNotEmpty; _showSendButton = textController.text.isNotEmpty; _showMentionAutocomplete = textFieldFocusNode.hasFocus && textController.text.split(' ').last.startsWith('@'); }); } void _enqueueDelayedCallback( List<_DelayedCallback> queue, void Function() callback, ) { queue.add( _DelayedCallback( releaseAt: DateTime.now().add(_chatDelay), callback: callback, ), ); } void _flushDueDelayedCallbacks() { _flushDueDelayedCallbacksForQueue(_pendingChatCallbacks); _flushDueDelayedCallbacksForQueue(_pendingSevenTVCallbacks); } void _flushDueDelayedCallbacksForQueue(List<_DelayedCallback> queue) { if (queue.isEmpty) return; final now = DateTime.now(); final readyCallbacks = <_DelayedCallback>[]; queue.removeWhere((delayedCallback) { final isReady = !delayedCallback.releaseAt.isAfter(now); if (isReady) { readyCallbacks.add(delayedCallback); } return isReady; }); for (final delayedCallback in readyCallbacks) { delayedCallback.callback(); } } void _flushPendingDelayedCallbacks() { _flushPendingDelayedCallbacksForQueue(_pendingChatCallbacks); _flushPendingDelayedCallbacksForQueue(_pendingSevenTVCallbacks); } void _flushPendingDelayedCallbacksForQueue(List<_DelayedCallback> queue) { if (queue.isEmpty) return; final pendingCallbacks = List<_DelayedCallback>.from(queue); queue.clear(); for (final delayedCallback in pendingCallbacks) { delayedCallback.callback(); } } void _clearPendingDelayedCallbacks({ bool clearChat = true, bool clearSevenTV = true, }) { if (clearChat) { _pendingChatCallbacks.clear(); } if (clearSevenTV) { _pendingSevenTVCallbacks.clear(); } } /// Handle and process the provided string-representation of the IRC data. /// /// If a message, parses the IRC data into an [IRCMessage] and handles it based on the [Command]. /// Else if a PING request, sends back the PONG to keep the connection alive. @action void _handleIRCData(String data) { // The IRC data can contain more than one message separated by CRLF. // To account for this, split by CRLF, then loop and process each message. for (final message in data.trimRight().split('\r\n')) { // debugPrint('$message\n'); if (message == _serverReconnectNotice) { // Twitch is about to close the socket for maintenance — close it // ourselves so the existing onDone path reconnects with no backoff // (first retry is immediate) instead of waiting for the server to // drop us mid-message. _channel?.sink.close(1000); continue; } if (message.startsWith('@')) { // Parse defensively: a single malformed message must never abort the // loop, or one bad frame would drop every other message batched with // it (e.g. an entire recent-messages backlog). Skip and keep going. final IRCMessage parsedIRCMessage; try { parsedIRCMessage = IRCMessage.fromString( message, userLogin: auth.user.details?.login, ); } catch (e) { debugPrint('Failed to parse IRC message: $e'); continue; } if (parsedIRCMessage.user != null) { chatDetailsStore.chatUsers.add(parsedIRCMessage.user!); } // Filter messages from any blocked users if not a moderator or not the channel owner. if (!_userState.mod && channelName != auth.user.details?.login && auth.user.blockedUsers .where( (blockedUser) => blockedUser.userLogin == parsedIRCMessage.user, ) .isNotEmpty) { continue; } // Filter messages containing any muted words. Must use `continue` // (not `return`) — a single IRC frame can batch many messages, and // returning would drop the rest of the frame with the matched one. if (parsedIRCMessage.message != null && _normalizedMutedWords.isNotEmpty) { final msg = parsedIRCMessage.message!.toLowerCase(); final words = settings.matchWholeWord ? msg.split(' ') : null; final muted = _normalizedMutedWords.any( (word) => words != null ? words.contains(word) : msg.contains(word), ); if (muted) continue; } switch (parsedIRCMessage.command) { case Command.privateMessage: case Command.notice: case Command.userNotice: // Check if this is a rejection notice for a pending message if (parsedIRCMessage.command == Command.notice && _isWaitingForAck) { final msgId = parsedIRCMessage.tags['msg-id']; if (msgId != null && _rejectionMsgIds.contains(msgId)) { // Message was rejected - clear pending but preserve text field for retry _sendingTimeoutTimer?.cancel(); _isWaitingForAck = false; toSend = null; // Drop the retained text so a later USERSTATE can't render the // rejected message as a deferred echo. _pendingSentMessageText = null; // Show notification with Twitch's rejection message if (parsedIRCMessage.message != null) { HapticFeedback.heavyImpact(); updateNotification(parsedIRCMessage.message!); } } } // Update shared chat mode based on source-room-id tag presence. // Only PRIVMSG reliably carries this tag for the session's // duration — NOTICE/USERNOTICE (subs, raids, announcements) // don't always, so gating on those too would flap this flag off // and on within a single live session (now visibly, since the // shared chat bubble bar reads it — previously this only risked // a redundant asset refetch, guarded below). if (parsedIRCMessage.command == Command.privateMessage) { final wasShared = _isInSharedChatMode; _isInSharedChatMode = parsedIRCMessage.tags.containsKey( 'source-room-id', ); // On transition into shared chat mode, fetch assets for participants. if (!wasShared && _isInSharedChatMode) { assetsStore.fetchSharedChatAssets( channelId: channelId, headers: auth.headersTwitch, onEmoteError: (error) { debugPrint(error.toString()); return []; }, onBadgeError: (error) { debugPrint(error.toString()); return {}; }, showTwitchEmotes: settings.showTwitchEmotes, showTwitchBadges: settings.showTwitchBadges, show7TVEmotes: settings.show7TVEmotes, showBTTVEmotes: settings.showBTTVEmotes, showFFZEmotes: settings.showFFZEmotes, showFFZBadges: settings.showFFZBadges, ); } else if (wasShared && !_isInSharedChatMode && assetsStore.hasLoadedSharedChatAssets) { // On leaving shared chat, drop the participants' merged emotes and // badges and restore just this channel's assets — otherwise the // other streamer's emote set and badges linger (#522). Guarded on // having actually loaded shared assets so a flapping tag can't // trigger redundant refetches. assetsStore.resetSharedChatAssets(); getAssets(); } } messageBuffer.add(parsedIRCMessage); break; case Command.clearChat: // Track the banned/timed-out user so the chat can offer an unban // action only when applicable. target-user-id is absent for a // whole-chat clear. final clearedUserId = parsedIRCMessage.tags['target-user-id']; if (clearedUserId != null) { final banDuration = parsedIRCMessage.tags['ban-duration']; final seconds = banDuration != null ? int.tryParse(banDuration) : null; if (seconds != null) { _bannedUserTracker.recordTimeout( clearedUserId, Duration(seconds: seconds), DateTime.now(), ); } else { _bannedUserTracker.recordBan(clearedUserId); } } IRCMessage.clearChat( messages: _messages, bufferedMessages: messageBuffer, ircMessage: parsedIRCMessage, ); // clearChat marks messages in place, which doesn't notify the // observable list; reassign so renderMessages recomputes and the // ban/timeout strike-through shows without waiting for the next // message. _messages = _messages.toList().asObservable(); break; case Command.clearMessage: IRCMessage.clearMessage( messages: _messages, bufferedMessages: messageBuffer, ircMessage: parsedIRCMessage, ); // Same as clearChat: force a re-render so the deleted message // updates immediately rather than on the next incoming message. _messages = _messages.toList().asObservable(); break; case Command.roomState: chatDetailsStore.roomState = chatDetailsStore.roomState .fromIRCMessage(parsedIRCMessage); continue; case Command.userState: _userState = _userState.fromIRCMessage(parsedIRCMessage); // A USERSTATE following our PRIVMSG confirms the send. Finalize // regardless of whether a local echo was pre-built — gating this on // `toSend` left the player stuck "sending" (and firing a false // "may not have been sent") whenever the message went out before // this session's first USERSTATE populated `_userState`. if (_isWaitingForAck) { _finalizeSend(parsedIRCMessage.tags['id']); } break; case Command.globalUserState: final setIds = parsedIRCMessage.tags['emote-sets']?.split(','); if (setIds != null) { assetsStore.userEmotesFuture( emoteSets: setIds, headers: auth.headersTwitch, onError: (error) { debugPrint(error.toString()); return []; }, ); } continue; case Command.none: debugPrint('Unknown command: ${parsedIRCMessage.command}'); continue; } if (!_autoScroll) { // While autoscroll is disabled, occasionally move messages from the buffer to the messages to prevent a memory leak. if (messageBuffer.length >= _messagesToRemove) { _messages.addAll(messageBuffer); messageBuffer.clear(); } } // If the message limit is reached, remove the oldest messages. if (_messages.length >= _messageLimit) { _messages = _messages.sublist(_messagesToRemove).asObservable(); } } else if (message == 'PING :tmi.twitch.tv') { _channel?.sink.add('PONG :tmi.twitch.tv'); return; } else if (message.contains('Welcome, GLHF!')) { messageBuffer.add( IRCMessage.createNotice( message: "Welcome to ${getReadableName(displayName, channelName)}'s chat!", ), ); // Activate the message buffer. // Cancel any existing timer before creating a new one to prevent duplicates on reconnect. _messageBufferTimer?.cancel(); // Create a timer that will add messages from the buffer every 200 milliseconds. _messageBufferTimer = Timer.periodic( const Duration(milliseconds: 200), (timer) => addMessages(), ); // Set up 7TV real-time listener (assets already fetched in connectToChat) if (settings.show7TVEmotes) { final emoteSetId = assetsStore.sevenTvEmoteSetId; if (emoteSetId != null && !_shouldDisconnect) { listenToSevenTVEmoteSet(emoteSetId: emoteSetId); } } // Transform reconnect message to summary on successful connection if (_reconnectMessage != null && _reconnectStartTime != null) { final elapsed = DateTime.now() .difference(_reconnectStartTime!) .inSeconds; final attempts = _retries; final index = _messages.indexOf(_reconnectMessage!); if (index != -1) { _messages[index] = IRCMessage.createNotice( message: 'Reconnected ($attempts ${attempts == 1 ? 'attempt' : 'attempts'}, ${elapsed}s)', ); } } _reconnectMessage = null; _reconnectStartTime = null; // Reset exponential backoff if successfully connected. _retries = 0; _backoffTime = 0; // Mark connection as established for UI _isConnected = true; _hasConnected = true; } } } // Fetch the assets used in chat including badges and emotes. @action Future getAssets() async { // Prepare common error handlers to avoid repetition. List onEmoteError(dynamic error) { debugPrint(error.toString()); return []; } // Returns a Map, not a List: this is attached via `.catchError` to the // `getBadgesChannel` future (a Future>), so the // handler must return that type or Dart throws at runtime. Map onBadgeError(dynamic error) { debugPrint(error.toString()); return {}; } final baseAssets = assetsStore.assetsFuture( channelId: channelId, headers: auth.headersTwitch, onEmoteError: onEmoteError, onBadgeError: onBadgeError, showTwitchEmotes: settings.showTwitchEmotes, showTwitchBadges: settings.showTwitchBadges, show7TVEmotes: settings.show7TVEmotes, showBTTVEmotes: settings.showBTTVEmotes, showBTTVBadges: settings.showBTTVBadges, showFFZEmotes: settings.showFFZEmotes, showFFZBadges: settings.showFFZBadges, ); // If shared chat mode is active, also refresh participant assets in parallel. if (_isInSharedChatMode) { final sharedAssets = assetsStore.fetchSharedChatAssets( channelId: channelId, headers: auth.headersTwitch, onEmoteError: onEmoteError, onBadgeError: onBadgeError, showTwitchEmotes: settings.showTwitchEmotes, showTwitchBadges: settings.showTwitchBadges, show7TVEmotes: settings.show7TVEmotes, showBTTVEmotes: settings.showBTTVEmotes, showFFZEmotes: settings.showFFZEmotes, showFFZBadges: settings.showFFZBadges, force: true, ); await Future.wait([baseAssets, sharedAssets]); } else { await baseAssets; } } /// Re-enables [_autoScroll] and jumps to the latest message. @action void resumeScroll() { _autoScroll = true; // Jump to the latest message (bottom of the list/chat). scrollController.jumpTo(0); // Add a post frame callback in the event a messages is added at the same time. WidgetsBinding.instance.addPostFrameCallback((_) { scrollController.jumpTo(0); }); } @action void listenToSevenTVEmoteSet({required String emoteSetId}) { final subscribePayload = SevenTVEvent( op: 35, d: SevenTVEventData( type: 'emote_set.update', condition: {'object_id': emoteSetId}, ), ); _sevenTVChannel?.sink.close(1000); _clearPendingDelayedCallbacks(clearChat: false); _sevenTVChannel = IOWebSocketChannel.connect( Uri.parse('wss://events.7tv.io/v3'), pingInterval: _pingInterval, ); void listener(dynamic data) { // debugPrint(data); final decoded = jsonDecode(data); final event = SevenTVEvent.fromJson(decoded); final body = event.d.body; if (event.d.type != 'emote_set.update' || body == null) return; if (body.pushed != null) { final pushedEmote = body.pushed?.first.value; if (pushedEmote == null) return; final emote = Emote.from7TV(pushedEmote, EmoteType.sevenTVChannel); assetsStore.addChannelEmote(emote); messageBuffer.add( IRCMessage.createNotice( message: '${getReadableName(body.actor.displayName, body.actor.username)} added 7TV emote "${emote.name}" to chat', ), ); } else if (body.pulled != null) { final pulledEmote = body.pulled?.first.oldValue; if (pulledEmote == null) return; assetsStore.removeChannelEmote(pulledEmote.name); messageBuffer.add( IRCMessage.createNotice( message: '${getReadableName(body.actor.displayName, body.actor.username)} removed 7TV emote "${pulledEmote.name}" from chat', ), ); } } // Cancel previous subscription if exists _sevenTVChannelListener?.cancel(); _sevenTVChannelListener = _sevenTVChannel?.stream.listen( (data) { if (!_hasActiveChatDelay) { listener(data); } else { _enqueueDelayedCallback( _pendingSevenTVCallbacks, () => listener(data), ); } }, onError: (error) { debugPrint('7TV events error: ${error.toString()}'); FirebaseCrashlytics.instance.recordError( error, StackTrace.current, reason: '7TV WebSocket error, channel=$channelName', ); }, onDone: () { _clearPendingDelayedCallbacks(clearChat: false); debugPrint('7TV events done'); }, ); _sevenTVChannel?.sink.add(jsonEncode(subscribePayload)); } @action Future connectToChat({bool isReconnect = false}) async { FirebaseCrashlytics.instance.setCustomKey('channel', channelName); // Fetch assets first so they're available for all messages getAssets().catchError( (e) => debugPrint('Failed to fetch chat assets: $e'), ); // Cancel existing listener to prevent duplicate message processing _channelListener?.cancel(); _clearPendingDelayedCallbacks(clearSevenTV: false); _channel?.sink.close(1000); _channel = IOWebSocketChannel.connect( Uri.parse('wss://irc-ws.chat.twitch.tv:443'), pingInterval: _pingInterval, ); // Only show chat delay countdown on initial connection or video toggle, not on reconnects if (!isReconnect && _hasActiveChatDelay) { _startChatDelayCountdown(); } // Listen for new messages and forward them to the handler. _channelListener = _channel?.stream.listen( (data) { final dataStr = data.toString(); // No active delay: process the whole frame immediately. if (!_hasActiveChatDelay) { _handleIRCData(dataStr); return; } // Dispatch each line individually so a bypass-eligible control message // isn't dragged behind the delay by chat lines batched in the same // frame — and, conversely, so a delayed chat line can't hold up a // PING/USERSTATE that shares the frame. for (final line in dataStr.split('\r\n')) { final trimmed = line.trimRight(); if (trimmed.isEmpty) continue; if (_lineShouldBypassDelay(trimmed)) { _handleIRCData(trimmed); } else { _enqueueDelayedCallback( _pendingChatCallbacks, () => _handleIRCData(trimmed), ); } } }, onError: (error) { debugPrint('Chat error: ${error.toString()}'); FirebaseCrashlytics.instance.recordError( error, StackTrace.current, reason: 'IRC WebSocket error, channel=$channelName', ); }, onDone: () async { _clearPendingDelayedCallbacks(clearSevenTV: false); // Mark connection as disconnected for UI _isConnected = false; // Cancel chat delay countdown when disconnecting _cancelChatDelayCountdown(); _chatDelaySyncCompleted = false; // Reset send state — any in-flight message is unrecoverable across // a disconnect, so don't leave the user blocked waiting for an ack // that will never arrive. _sendingTimeoutTimer?.cancel(); _isWaitingForAck = false; toSend = null; _pendingSentMessageText = null; if (_shouldDisconnect) { _sevenTVChannel?.sink.close(1000); return; } if (_retries >= _maxRetries) { FirebaseCrashlytics.instance.log( 'WebSocket reconnection failed after $_maxRetries attempts, channel=$channelName', ); // Remove the reconnect message before showing final disconnect notice if (_reconnectMessage != null) { final index = _messages.indexOf(_reconnectMessage!); if (index != -1) _messages.removeAt(index); _reconnectMessage = null; } _reconnectStartTime = null; // Add directly to messages, not buffer, so it shows immediately _messages.add( IRCMessage.createNotice( message: 'Chat disconnected. Please check your connection.', actionCallback: () { _retries = 0; _backoffTime = 0; connectToChat(isReconnect: true); }, actionLabel: 'Reconnect', ), ); return; } // Increment the retry count. _retries++; FirebaseCrashlytics.instance.log( 'WebSocket reconnecting: attempt $_retries/$_maxRetries, channel=$channelName', ); // Start exponential backoff only after first failed attempt (capped at 8s). // First retry is immediate to catch brief network hiccups. if (_retries > 1) { final newBackoff = _backoffTime == 0 ? 1 : _backoffTime * 2; _backoffTime = newBackoff > 8 ? 8 : newBackoff; } // Helper to update the single reconnect message in place (or add if first time) void updateReconnectMessage(String text) { final msg = IRCMessage.createNotice(message: text); if (_reconnectMessage == null) { _reconnectStartTime ??= DateTime.now(); // Record when reconnection started _reconnectMessage = msg; _messages.add(_reconnectMessage!); } else { final index = _messages.indexOf(_reconnectMessage!); if (index != -1) { _reconnectMessage = msg; _messages[index] = _reconnectMessage!; } else { // Reference was lost (e.g., message limit cleanup) - re-add _reconnectMessage = msg; _messages.add(_reconnectMessage!); } } } // Countdown phase (if backoff time > 0) if (_backoffTime > 0) { var remainingSeconds = _backoffTime; updateReconnectMessage( 'Reconnecting in ${remainingSeconds}s... (attempt $_retries of $_maxRetries)', ); // Update countdown every second await Future.doWhile(() async { await Future.delayed(const Duration(seconds: 1)); // Abort countdown if dispose was called if (_shouldDisconnect) { if (_reconnectMessage != null) { final index = _messages.indexOf(_reconnectMessage!); if (index != -1) _messages.removeAt(index); _reconnectMessage = null; } _reconnectStartTime = null; return false; // Exit loop } remainingSeconds--; if (remainingSeconds > 0) { updateReconnectMessage( 'Reconnecting in ${remainingSeconds}s... (attempt $_retries of $_maxRetries)', ); return true; // Continue loop } return false; // Exit loop }); // Don't proceed with reconnection if disposed if (_shouldDisconnect) return; } // Attempting phase - update same message, no add/remove updateReconnectMessage( 'Reconnecting... (attempt $_retries of $_maxRetries)', ); _channelListener?.cancel(); connectToChat(isReconnect: true); }, ); // The list of messages sent to the IRC WebSocket channel to connect and join. final commands = [ // Request the tags and commands capabilities. // This will display tags containing metadata along with each IRC message. 'CAP REQ :twitch.tv/tags twitch.tv/commands', // The OAuth token in order to connect, default or user token. 'PASS oauth:${auth.token}', // The nickname for the connecting user. 'justinfan888' is the Twitch default if not logged in. 'NICK ${auth.isLoggedIn ? auth.user.details!.login : 'justinfan888'}', // Join the desired channel's room. 'JOIN #$channelName', ]; // Send each command in order. for (final command in commands) { _channel?.sink.add(command); } } @action void addMessages() { _flushDueDelayedCallbacks(); // Hold messages back while the user is touching the list so the pressed // message stays put (and alive) for the duration of a long-press. if (!_autoScroll || _interactionPause.isPaused || messageBuffer.isEmpty) { return; } _messages.addAll(messageBuffer); messageBuffer.clear(); } /// Cancels the chat delay countdown and removes the countdown message. @action void _cancelChatDelayCountdown() { _chatDelayCountdownTimer?.cancel(); _chatDelayCountdownTimer = null; // Remove the countdown message if it exists if (_countdownMessage != null) { _messages.remove(_countdownMessage); _countdownMessage = null; } } /// Starts and manages the chat delay countdown message. @action void _startChatDelayCountdown() { // Cancel any existing countdown first (prevents duplicate messages) _cancelChatDelayCountdown(); // Flush any buffered messages first to ensure countdown is at the bottom if (messageBuffer.isNotEmpty) { _messages.addAll(messageBuffer); messageBuffer.clear(); } var remainingSeconds = _chatDelay.inSeconds; // Create and store reference to the countdown message _countdownMessage = IRCMessage.createNotice( message: 'Chat will sync in ${remainingSeconds}s...', ); _messages.add(_countdownMessage!); // Update countdown every second _chatDelayCountdownTimer = Timer.periodic(const Duration(seconds: 1), ( timer, ) { remainingSeconds--; // Use runInAction for proper MobX reactivity in async callbacks runInAction(() { // Check if countdown message still exists in the list // (could be removed by message limit cleanup) if (_countdownMessage == null || !_messages.contains(_countdownMessage)) { timer.cancel(); _countdownMessage = null; _chatDelayCountdownTimer = null; return; } final index = _messages.indexOf(_countdownMessage!); if (index != -1) { if (remainingSeconds > 0) { // Create new message and update reference _countdownMessage = IRCMessage.createNotice( message: 'Chat will sync in ${remainingSeconds}s...', ); _messages[index] = _countdownMessage!; } else { // Remove countdown message when done _messages.removeAt(index); _countdownMessage = null; _chatDelayCountdownTimer = null; _chatDelaySyncCompleted = true; timer.cancel(); } } else { _countdownMessage = null; _chatDelayCountdownTimer = null; timer.cancel(); } }); }); } /// Sends the given string message by the logged-in user and adds it to [_messages]. @action void sendMessage(String message) { // Do not send if the message is blank/empty. if (message.isEmpty) return; // Prevent double-sending while waiting for server acknowledgment if (_isWaitingForAck) { HapticFeedback.mediumImpact(); updateNotification('Please wait, sending previous message...'); return; } if (_channel == null || _channel?.closeCode != null) { messageBuffer.add( IRCMessage.createNotice( message: 'Cannot send message - chat is disconnected. Reconnecting...', ), ); return; } // Set waiting state for UI feedback _isWaitingForAck = true; // Send the message to the IRC chat room. _channel?.sink.add( '${replyingToMessage != null ? '@reply-parent-msg-id=${replyingToMessage!.tags['id']} ' : ''}PRIVMSG #$channelName :$message', ); // Text field and reply state cleared on USERSTATE confirmation (not optimistically) // to preserve message text if Twitch rejects it (e.g., slow mode) // Fallback in case the confirming USERSTATE never arrives. Twitch can omit // or drop the USERSTATE that acknowledges a send — intermittently, and more // often under a long chat delay — even though the message was delivered. // A missing ack is therefore NOT evidence of failure, so rather than cry // wolf with a "may not have been sent" error (which pushes the user to // resend and double-post), optimistically finalize the send: render the // local echo and clear the pending state. Genuine rejections almost always // arrive first — the rejection NOTICE bypasses the chat delay, so it lands // well within this window and clears the pending send before the timeout. A // rejection arriving after the timeout has finalized is treated as sent (the // echo stays), an accepted edge given how early NOTICEs arrive. Scaled by // the chat delay so a legitimately delayed USERSTATE is given time to land. _sendingTimeoutTimer?.cancel(); _sendingTimeoutTimer = Timer(const Duration(seconds: 10) + _chatDelay, () { if (_shouldDisconnect || !_isWaitingForAck) return; runInAction(() => _finalizeSend(null)); }); // Build the optimistic local echo from the current USERSTATE. If the // message is sent before this session's first USERSTATE arrives, // _buildLocalEcho returns null and the echo is built later, when the // confirming USERSTATE lands (see the Command.userState handler). The raw // text is retained so that deferred build is possible. _pendingSentMessageText = message; toSend = _buildLocalEcho(message); } /// Renders the pending outgoing message's local echo (building it lazily if /// it couldn't be built at send time) and clears all send-pending state. /// Shared by the USERSTATE confirmation and the optimistic timeout fallback. /// [serverMessageId] is the `id` tag from the confirming USERSTATE, or null /// when finalizing optimistically without one. void _finalizeSend(String? serverMessageId) { toSend ??= _pendingSentMessageText != null ? _buildLocalEcho(_pendingSentMessageText!) : null; if (toSend != null) { // Prefer the server id; fall back to any id already set, then a local one. toSend!.tags['id'] = serverMessageId ?? toSend!.tags['id'] ?? 'local-${DateTime.now().millisecondsSinceEpoch}'; messageBuffer.add(toSend!); toSend = null; } _pendingSentMessageText = null; _sendingTimeoutTimer?.cancel(); _isWaitingForAck = false; textController.clear(); replyingToMessage = null; } /// Builds the optimistic local echo of an outgoing [message] from the current /// USERSTATE, or null if no USERSTATE has been received yet this session /// (the echo can't be rendered without the user's badges/color/display-name). IRCMessage? _buildLocalEcho(String message) { var userStateString = _userState.raw; if (userStateString == null) return null; if (message == '/me' || message.startsWith('/me ')) { userStateString += ' :\x01ACTION ${message.replaceRange(0, 3, '').trim()}\x01'; } else { userStateString += ' :${replyingToMessage?.tags['display-name'] != null ? '@${replyingToMessage!.tags['display-name']} ' : ''}${message.trim()}'; } final userChatMessage = IRCMessage.fromString(userStateString); userChatMessage.localEmotes?.addAll(assetsStore.userEmoteToObject); if (auth.isLoggedIn && auth.user.details != null) { userChatMessage.tags['user-id'] = auth.user.details!.id; } if (replyingToMessage != null && replyingToMessage!.tags['id'] != null) { userChatMessage.tags['reply-parent-msg-id'] = replyingToMessage!.tags['id']!; userChatMessage.tags['reply-parent-display-name'] = replyingToMessage!.tags['display-name'] ?? ''; userChatMessage.tags['reply-parent-msg-body'] = replyingToMessage!.message ?? ''; } // Ensure a timestamp so merged-mode sorting works correctly. userChatMessage.tags['tmi-sent-ts'] ??= '${DateTime.now().millisecondsSinceEpoch}'; return userChatMessage; } /// Adds the given [emote] to the chat textfield. @action void addEmote(Emote emote, {bool autocompleteMode = false}) { if (textController.text.isEmpty || textController.text.endsWith(' ')) { textController.text += '${emote.name} '; } else if (autocompleteMode && _showEmoteAutocomplete && textController.text.endsWith('')) { final split = textController.text.split(' ') ..removeLast() ..add('${emote.name} '); textController.text = split.join(' '); } else { textController.text += ' ${emote.name} '; } assetsStore.recentEmotes ..removeWhere( (recentEmote) => recentEmote.name == emote.name && recentEmote.type == emote.type, ) ..insert(0, emote); textController.selection = TextSelection.fromPosition( TextPosition(offset: textController.text.length), ); } /// Cancels the previous notification/timer and creates a new one with the provided [notificationMessage]. @action void updateNotification(String notificationMessage) { // Cancel the previous notification to prevent the notification from phasing in and out // when copying messages repeatedly. _notificationTimer?.cancel(); // Set the new notification message and create a new timer that will dismiss it after 3 seconds. _notification = notificationMessage; _notificationTimer = Timer( const Duration(seconds: 5), () => _notification = null, ); } /// Clears the current notification immediately. @action void clearNotification() { _notificationTimer?.cancel(); _notification = null; } /// Updates the sleep timer with the given [duration]. /// Calls [onTimerFinished] when the sleep timer completes. @action void updateSleepTimer({ required Duration duration, required VoidCallback onTimerFinished, }) { // If there is an ongoing timer, cancel it since it'll be replaced. if (sleepTimer != null) cancelSleepTimer(); // Update the new time remaining timeRemaining = duration; // Set a periodic timer that will update the time remaining every second. sleepTimer = Timer.periodic(const Duration(seconds: 1), (timer) { // If the timer is up, cancel the timer and exit the app. if (timeRemaining.inSeconds == 0) { timer.cancel(); onTimerFinished(); return; } // Decrement the time remaining. timeRemaining = Duration(seconds: timeRemaining.inSeconds - 1); }); } /// Cancels the sleep timer and resets the time remaining. @action void cancelSleepTimer() { sleepTimer?.cancel(); timeRemaining = const Duration(); } @action Future getRecentMessage() async { try { final recentMessages = await twitchApi.getRecentMessages( userLogin: channelName, ); for (final message in recentMessages) { _handleIRCData(message); } } catch (e) { debugPrint('Failed to fetch recent messages: $e'); } } /// Closes and disposes all the channels and controllers used by the store. void dispose() { _shouldDisconnect = true; _interactionPause.reset(); _clearPendingDelayedCallbacks(); _messageBufferTimer?.cancel(); _notificationTimer?.cancel(); _sendingTimeoutTimer?.cancel(); _cancelChatDelayCountdown(); sleepTimer?.cancel(); // Cancel WebSocket subscriptions _channelListener?.cancel(); _sevenTVChannelListener?.cancel(); // Close WebSocket channels _channel?.sink.close(1000); _sevenTVChannel?.sink.close(1000); for (final reactionDisposer in reactions) { reactionDisposer(); } textFieldFocusNode.dispose(); textController.dispose(); scrollController.dispose(); assetsStore.dispose(); chatDetailsStore.dispose(); } /// Unfocuses the text field. @action void unfocusInput() { textFieldFocusNode.unfocus(); } /// Requests focus for the text field. @action void safeRequestFocus() { textFieldFocusNode.requestFocus(); } } class _DelayedCallback { _DelayedCallback({required this.releaseAt, required this.callback}); final DateTime releaseAt; final void Function() callback; }