wip; message: Create an outbox message on send; manage its states

TODO fix broken tests; more tests, documentation.

This does not debounce new outbox messages. That will be handled in a
later commit.

Author: PIG208

lib/model/message.dart

@@ -1,6 +1,7 @@
 import 'dart:convert';
 
 import '../api/core.dart';
+import '../api/exception.dart';
 import '../api/model/events.dart';
 import '../api/model/model.dart';
 import '../api/route/messages.dart';
@@ -145,14 +146,23 @@ mixin MessageStore {
 }
 
 class MessageStoreImpl with MessageStore {
-  MessageStoreImpl({required this.connection})
+  MessageStoreImpl({
+    required this.connection,
+    required this.queueId,
+    required this.selfUserId,
+  })
     // There are no messages in InitialSnapshot, so we don't have
     // a use case for initializing MessageStore with nonempty [messages].
     : messages = {},
       outboxMessages = {};
 
   final ApiConnection connection;
 
+  /// The event queue ID, for local-echoing, as in [InitialSnapshot.queueId].
+  final String queueId;
+
+  final int selfUserId;
+
   @override
   final Map<int, Message> messages;
 
@@ -200,14 +210,46 @@ class MessageStoreImpl with MessageStore {
   }
 
   @override
-  Future<void> sendMessage({required MessageDestination destination, required String content}) {
-    // TODO implement outbox; see design at
-    //   https://chat.zulip.org/#narrow/stream/243-mobile-team/topic/.23M3881.20Sending.20outbox.20messages.20is.20fraught.20with.20issues/near/1405739
-    return _apiSendMessage(connection,
-      destination: destination,
-      content: content,
-      readBySender: true,
-    );
+  Future<void> sendMessage({required MessageDestination destination, required String content}) async {
+    final outboxMessage = OutboxMessage.fromDestination(
+      destination, selfUserId: selfUserId, content: content);
+    final localMessageId = outboxMessage.localMessageId;
+    assert(!outboxMessages.containsKey(localMessageId));
+    // TODO debounce new outbox messages
+    outboxMessages[localMessageId] = outboxMessage;
+    for (final view in _messageListViews) {
+      view.handleOutboxMessage(outboxMessage);
+    }
+
+    try {
+      await _apiSendMessage(connection,
+        destination: destination,
+        content: content,
+        readBySender: true,
+        queueId: queueId,
+        localId: localMessageId);
+      _updateOutboxMessage(
+        localMessageId: localMessageId, newState: OutboxMessageLifecycle.sent);
+    } on ApiRequestException {
+      _updateOutboxMessage(
+        localMessageId: localMessageId, newState: OutboxMessageLifecycle.failed);
+      rethrow;
+    }
+  }
+
+  void _updateOutboxMessage({
+    required int localMessageId,
+    required OutboxMessageLifecycle newState,
+  }) {
+    final outboxMessage = outboxMessages[localMessageId];
+    assert(outboxMessage != null);
+    if (outboxMessage == null) {
+      return;
+    }
+    outboxMessage.state = newState;
+    for (final view in _messageListViews) {
+      view.handleUpdateOutboxMessage(localMessageId);
+    }
   }
 
   @override
@@ -245,6 +287,8 @@ class MessageStoreImpl with MessageStore {
     // See [fetchedMessages] for reasoning.
     messages[event.message.id] = event.message;
 
+    outboxMessages.remove(event.localMessageId);
+
     for (final view in _messageListViews) {
       view.handleMessageEvent(event);
     }

lib/model/store.dart

@@ -374,7 +374,11 @@ class PerAccountStore extends ChangeNotifier with EmojiStore, UserStore, Channel
         typingStartedExpiryPeriod: Duration(milliseconds: initialSnapshot.serverTypingStartedExpiryPeriodMilliseconds),
       ),
       channels: channels,
-      messages: MessageStoreImpl(connection: connection),
+      messages: MessageStoreImpl(
+        connection: connection,
+        queueId: queueId,
+        selfUserId: account.userId,
+      ),
       unreads: Unreads(
         initial: initialSnapshot.unreadMsgs,
         selfUserId: account.userId,

test/example_data.dart

@@ -620,8 +620,8 @@ UserTopicEvent userTopicEvent(
   );
 }
 
-MessageEvent messageEvent(Message message) =>
-  MessageEvent(id: 0, message: message, localMessageId: null);
+MessageEvent messageEvent(Message message, {int? localMessageId}) =>
+  MessageEvent(id: 0, message: message, localMessageId: localMessageId);
 
 DeleteMessageEvent deleteMessageEvent(List<StreamMessage> messages) {
   assert(messages.isNotEmpty);

test/model/message_checks.dart

@@ -0,0 +1,6 @@
+import 'package:checks/checks.dart';
+import 'package:zulip/model/message.dart';
+
+extension OutboxMessageChecks on Subject<OutboxMessage> {
+  Subject<OutboxMessageLifecycle> get state => has((x) => x.state, 'state');
+}

test/model/message_test.dart

@@ -1,10 +1,13 @@
 import 'dart:convert';
 
 import 'package:checks/checks.dart';
+import 'package:http/http.dart' as http;
 import 'package:test/scaffolding.dart';
 import 'package:zulip/api/model/events.dart';
 import 'package:zulip/api/model/model.dart';
 import 'package:zulip/api/model/submessage.dart';
+import 'package:zulip/api/route/messages.dart';
+import 'package:zulip/model/message.dart';
 import 'package:zulip/model/message_list.dart';
 import 'package:zulip/model/narrow.dart';
 import 'package:zulip/model/store.dart';
@@ -13,7 +16,9 @@ import '../api/fake_api.dart';
 import '../api/model/model_checks.dart';
 import '../api/model/submessage_checks.dart';
 import '../example_data.dart' as eg;
+import '../fake_async.dart';
 import '../stdlib_checks.dart';
+import 'message_checks.dart';
 import 'message_list_test.dart';
 import 'store_checks.dart';
 import 'test_store.dart';
@@ -77,6 +82,69 @@ void main() {
     checkNotified(count: messageList.fetched ? messages.length : 0);
   }
 
+  group('sendMessage', () {
+    const destination =
+      StreamDestination(eg.defaultStreamMessageStreamId, TopicName('some topic'));
+
+    test('message sent successfully, message event arrives on time', () async {
+      await prepare();
+
+      connection.prepare(json: SendMessageResult(id: 1).toJson(),
+        delay: Duration.zero);
+      final future = store.sendMessage(destination: destination, content: 'content');
+      final outboxMessage = store.outboxMessages.values.single;
+      check(outboxMessage).state.equals(OutboxMessageLifecycle.sending);
+      checkNotNotified();
+      check(connection.lastRequest).isA<http.Request>()
+        ..bodyFields['queue_id'].equals(store.queueId)
+        ..bodyFields['local_id'].equals('${outboxMessage.localMessageId}');
+
+      await future;
+      check(outboxMessage).state.equals(OutboxMessageLifecycle.sent);
+      checkNotifiedOnce();
+
+      await store.handleEvent(eg.messageEvent(
+        eg.streamMessage(), localMessageId: outboxMessage.localMessageId));
+      check(store.outboxMessages).isEmpty();
+      // TODO uncomment once message list support is added
+      // checkNotifiedOnce();
+    });
+
+    test('message sent successfully, message event arrives after debounce timeout', () => awaitFakeAsync((async) async {
+      await prepare();
+
+      connection.prepare(json: SendMessageResult(id: 1).toJson());
+      await store.sendMessage(destination: destination, content: 'content');
+      final outboxMessage = store.outboxMessages.values.single;
+      check(outboxMessage).state.equals(OutboxMessageLifecycle.sent);
+      checkNotifiedOnce();
+
+      async.elapse(kLocalEchoDebounceDuration);
+      check(outboxMessage).state.equals(OutboxMessageLifecycle.sent);
+
+      await store.handleEvent(eg.messageEvent(
+        eg.streamMessage(), localMessageId: outboxMessage.localMessageId));
+      check(store.outboxMessages).isEmpty();
+      // TODO uncomment once message list support is added
+      // checkNotifiedOnce();
+    }));
+
+    test('message failed to send', () async {
+      await prepare();
+
+      connection.prepare(apiException: eg.apiBadRequest(),
+        delay: Duration.zero);
+      final future = store.sendMessage(destination: destination, content: 'content');
+      final outboxMessage = store.outboxMessages.values.single;
+      check(outboxMessage).state.equals(OutboxMessageLifecycle.sending);
+      checkNotNotified();
+
+      await check(future).throws();
+      check(outboxMessage).state.equals(OutboxMessageLifecycle.failed);
+      checkNotifiedOnce();
+    });
+  });
+
   group('reconcileMessages', () {
     test('from empty', () async {
       await prepare();