fix(ui): auto-scroll on optimistic local messages

Subscribe to `channel.state.messagesStream` (or `threadsStream[parentId]`
in thread mode) instead of `channel.on(EventType.messageNew)` so the list
follows to the new bottom message the moment it lands in state. The event
path only fires on server-confirmed messages, which meant the user's own
send wasn't auto-scrolled until the server round-trip completed.

The data-source-driven pattern matches what the Android, iOS, and React
Native SDKs do. New-message detection uses a `lengthGrew && lastChanged`
check between emissions; the bottom-most snapshot is seeded from current
state on subscribe so we don't auto-scroll on the BehaviorSubject replay.
The synchronous `controller.scrollTo(index: 0)` call still clears SPL's
anchor key before `didUpdateWidget` (no race).

Adds an `auto_scroll_test.dart` covering: other-user-at-bottom,
other-user-scrolled-up, own-message-scrolled-up, optimistic local send,
and rapid burst.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

Author: xsahil03x

packages/stream_chat_flutter/CHANGELOG.md

@@ -1,3 +1,10 @@
+## Upcoming Changes
+
+🐞 Fixed
+
+- Fixed `StreamMessageListView` not auto-scrolling to the bottom on the user's own outgoing message
+  until the server confirmed it.
+
 ## 9.24.0
 
 ✅ Added

packages/stream_chat_flutter/lib/src/message_list_view/message_list_view.dart

@@ -416,7 +416,7 @@ class _StreamMessageListViewState extends State<StreamMessageListView> {
   MessageListController get _messageListController =>
       widget.messageListController ?? _defaultController;
 
-  StreamSubscription? _messageNewListener;
+  StreamSubscription<Message>? _messageNewListener;
   StreamSubscription? _userReadListener;
 
   @override
@@ -444,17 +444,16 @@ class _StreamMessageListViewState extends State<StreamMessageListView> {
       debouncedMarkRead.cancel();
       debouncedMarkThreadRead.cancel();
 
-      _messageNewListener?.cancel();
-      _userReadListener?.cancel();
-
       _unreadState.value = _readUnreadSnapshot();
 
-      _messageNewListener =
-          streamChannel!.channel.on(EventType.messageNew).listen((event) {
-        final message = event.message;
-        if (message == null) return;
-        if (message.parentId != widget.parentMessage?.id) return;
+      final state = streamChannel?.channel.state;
+      final newMessageStream = switch (widget.parentMessage?.id) {
+        final parentId? => state?.newThreadMessageStream(parentId),
+        _ => state?.newMessageStream,
+      };
 
+      _messageNewListener?.cancel();
+      _messageNewListener = newMessageStream?.listen((message) {
         // Don't fight a scroll already in motion (drag, fling, or
         // still-running animated scrollTo).
         if (_scrollController?.isScrolling == true) return;
@@ -479,8 +478,8 @@ class _StreamMessageListViewState extends State<StreamMessageListView> {
         }
       });
 
-      _userReadListener =
-          streamChannel!.channel.state?.currentUserReadStream.listen((_) {
+      _userReadListener?.cancel();
+      _userReadListener = state?.currentUserReadStream.listen((_) {
         _unreadState.value = _readUnreadSnapshot();
       });
     }

packages/stream_chat_flutter/lib/src/message_list_view/mlv_utils.dart

@@ -1,4 +1,5 @@
 import 'package:collection/collection.dart';
+import 'package:rxdart/rxdart.dart';
 import 'package:stream_chat_flutter/scrollable_positioned_list/scrollable_positioned_list.dart';
 import 'package:stream_chat_flutter/stream_chat_flutter.dart';
 
@@ -101,3 +102,100 @@ bool isElementAtIndexVisible(
 bool isInitialMessage(String id, StreamChannelState? channelState) {
   return channelState!.initialMessageId == id;
 }
+
+/// Stream helpers for observing newly arrived messages on a channel,
+/// either in the main message list or scoped to a thread.
+extension ChannelClientStateMessageStreamX on ChannelClientState {
+  // True when [candidate] represents a newer tail message than
+  // [previous].
+  //
+  // A new arrival requires both a different id *and* a strictly later
+  // [Message.createdAt], so edits, reactions, and reorderings are
+  // ignored.
+  bool _isNewTailArrival(Message candidate, Message? previous) {
+    if (previous == null) return true;
+    return candidate.id != previous.id &&
+        candidate.createdAt.isAfter(previous.createdAt);
+  }
+
+  /// A stream that emits each newly arrived bottom message in
+  /// [messages].
+  ///
+  /// Fires for every upstream that grows the list, including
+  /// server-confirmed `message.new` events, optimistic local sends,
+  /// and any other update that appends to the tail.
+  ///
+  /// A new arrival is detected when the bottom message's id changes
+  /// **and** its [Message.createdAt] is strictly after the previously
+  /// observed tail. Edits, reactions, tail deletions, and pruning are
+  /// therefore ignored.
+  ///
+  /// Gated on [isUpToDate]: while the channel is loaded around a
+  /// historic message the stream stays silent, and the first emission
+  /// after the gate re-opens re-seeds the baseline without yielding.
+  Stream<Message> get newMessageStream async* {
+    var wasUpToDate = isUpToDate;
+    var lastSeen = wasUpToDate ? messages.lastOrNull : null;
+
+    await for (final updated in messagesStream) {
+      if (!isUpToDate) {
+        wasUpToDate = false;
+        lastSeen = null;
+        continue;
+      }
+
+      // Re-seed without yielding: the gate just re-opened, the next
+      // emission is a wholesale window replacement, not an arrival.
+      if (!wasUpToDate) {
+        wasUpToDate = true;
+        lastSeen = updated.lastOrNull;
+        continue;
+      }
+
+      final newLast = updated.lastOrNull;
+      if (newLast == null) {
+        lastSeen = null;
+        continue;
+      }
+
+      final isNewArrival = _isNewTailArrival(newLast, lastSeen);
+      lastSeen = newLast;
+      if (!isNewArrival) continue;
+      yield newLast;
+    }
+  }
+
+  /// A stream that emits each newly arrived reply at the bottom of
+  /// the thread identified by [parentMessageId].
+  ///
+  /// Fires for every upstream that grows the thread, including
+  /// server-confirmed replies, optimistic local sends, and any other
+  /// update that appends to the tail of [threads].
+  ///
+  /// A new arrival is detected when the bottom reply's id changes
+  /// **and** its [Message.createdAt] is strictly after the previously
+  /// observed tail. Edits, reactions, tail deletions, and pruning are
+  /// therefore ignored.
+  ///
+  /// Threads load lazily, so the stream stays silent until [threads]
+  /// carries replies for [parentMessageId]; that first snapshot seeds
+  /// the baseline without yielding.
+  Stream<Message> newThreadMessageStream(String parentMessageId) async* {
+    final threadMessages =
+        threadsStream.mapNotNull((it) => it[parentMessageId]);
+
+    var lastSeen = threads[parentMessageId]?.lastOrNull;
+    await for (final updated in threadMessages) {
+      final newLast = updated.lastOrNull;
+      if (newLast == null) {
+        lastSeen = null;
+        continue;
+      }
+
+      final isNewArrival = _isNewTailArrival(newLast, lastSeen);
+      lastSeen = newLast;
+      if (!isNewArrival) continue;
+      yield newLast;
+    }
+  }
+}