1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
695
696
697
698
699
700
701
702
703
704
705
706
707
708
709
710
711
712
713
714
715
716
717
718
719
720
721
722
723
724
725
726
727
728
729
730
731
732
733
734
735
736
737
738
739
740
741
742
743
744
745
746
747
748
749
750
751
752
753
754
755
756
757
758
759
760
761
762
763
764
765
766
767
768
769
770
771
772
773
774
775
776
777
778
779
780
781
782
783
784
785
786
787
788
789
790
791
792
793
794
795
796
797
798
799
800
801
802
803
804
805
806
807
808
809
810
811
812
813
814
815
816
817
818
819
820
821
822
823
824
825
826
827
828
829
830
831
832
833
834
835
836
837
838
839
840
841
842
843
844
845
846
847
848
849
850
851
852
853
854
855
856
857
858
859
860
861
862
863
864
865
866
867
868
869
870
871
872
873
874
875
876
877
878
879
880
881
882
883
884
885
886
887
888
889
890
891
892
893
894
895
896
897
898
899
900
901
902
903
904
905
906
907
908
909
910
911
912
913
914
915
916
917
918
919
920
921
922
923
924
925
926
927
928
929
930
931
932
933
934
935
936
937
938
939
940
941
942
943
944
945
946
947
948
949
950
951
952
953
954
955
956
957
958
959
960
961
962
963
964
965
966
967
968
969
970
971
972
973
974
975
976
977
978
979
980
981
982
983
984
985
986
987
988
989
990
991
992
993
994
995
996
997
998
999
1000
1001
1002
1003
1004
1005
1006
1007
1008
1009
1010
1011
1012
1013
1014
1015
1016
1017
1018
1019
1020
1021
1022
1023
1024
1025
1026
1027
1028
1029
1030
1031
1032
1033
1034
1035
1036
1037
1038
1039
1040
1041
1042
1043
1044
1045
1046
1047
1048
1049
1050
1051
1052
1053
1054
1055
1056
1057
1058
1059
1060
1061
1062
1063
1064
1065
1066
1067
1068
1069
1070
1071
1072
1073
1074
1075
1076
1077
1078
1079
1080
1081
1082
1083
1084
1085
1086
1087
1088
1089
1090
1091
1092
1093
1094
1095
1096
1097
1098
1099
1100
1101
1102
1103
1104
1105
1106
1107
1108
1109
1110
1111
1112
1113
1114
1115
1116
1117
1118
1119
1120
1121
1122
1123
1124
1125
1126
1127
1128
1129
1130
1131
1132
1133
1134
1135
1136
1137
1138
1139
1140
1141
1142
1143
1144
1145
1146
1147
1148
1149
1150
1151
1152
1153
1154
1155
1156
1157
1158
1159
1160
1161
1162
1163
1164
1165
1166
1167
1168
1169
1170
1171
1172
1173
1174
1175
1176
1177
1178
1179
1180
1181
1182
1183
1184
1185
1186
1187
1188
1189
1190
1191
1192
1193
1194
1195
1196
1197
1198
1199
1200
1201
1202
1203
1204
1205
1206
1207
1208
1209
1210
1211
1212
1213
1214
1215
1216
1217
1218
1219
1220
1221
1222
1223
1224
1225
1226
1227
1228
1229
1230
1231
1232
1233
1234
1235
1236
1237
1238
1239
1240
1241
1242
1243
1244
1245
1246
1247
1248
1249
1250
1251
1252
1253
1254
1255
1256
1257
1258
1259
1260
1261
1262
1263
1264
1265
1266
1267
1268
1269
1270
1271
1272
1273
1274
1275
1276
1277
1278
1279
1280
1281
1282
1283
1284
1285
1286
1287
1288
1289
1290
1291
1292
1293
1294
1295
1296
1297
1298
1299
1300
1301
1302
1303
1304
1305
1306
1307
1308
1309
1310
1311
1312
1313
1314
1315
1316
1317
1318
1319
1320
1321
1322
1323
1324
1325
1326
1327
1328
1329
1330
1331
1332
1333
1334
1335
1336
1337
1338
1339
1340
1341
1342
1343
1344
1345
1346
1347
1348
1349
1350
1351
1352
1353
1354
1355
1356
1357
1358
1359
1360
1361
1362
1363
1364
1365
1366
1367
1368
1369
1370
1371
1372
1373
1374
1375
1376
1377
1378
1379
1380
1381
1382
1383
1384
1385
1386
1387
1388
1389
1390
1391
1392
1393
1394
1395
1396
1397
1398
1399
1400
1401
1402
1403
1404
1405
1406
1407
1408
1409
1410
1411
1412
1413
1414
1415
1416
1417
1418
1419
1420
1421
1422
1423
1424
1425
1426
1427
1428
1429
1430
1431
1432
1433
1434
1435
1436
1437
1438
1439
1440
1441
1442
1443
1444
1445
1446
1447
1448
1449
1450
1451
1452
1453
1454
1455
1456
1457
1458
1459
1460
1461
1462
1463
1464
1465
1466
1467
1468
1469
1470
1471
1472
1473
1474
1475
1476
1477
1478
1479
1480
1481
1482
1483
1484
1485
1486
1487
1488
1489
1490
1491
1492
1493
1494
1495
1496
1497
1498
1499
1500
1501
1502
1503
1504
1505
1506
1507
1508
1509
1510
1511
1512
1513
1514
1515
1516
1517
1518
1519
1520
1521
1522
1523
1524
1525
1526
1527
1528
1529
1530
1531
1532
1533
1534
1535
1536
1537
1538
1539
1540
1541
1542
1543
1544
1545
1546
1547
1548
1549
1550
1551
1552
1553
1554
1555
1556
1557
1558
1559
1560
1561
1562
1563
1564
1565
1566
1567
1568
1569
1570
1571
1572
1573
1574
1575
1576
1577
1578
1579
1580
1581
1582
1583
1584
1585
1586
1587
1588
1589
1590
1591
1592
1593
1594
1595
1596
1597
1598
1599
1600
1601
1602
1603
1604
1605
1606
1607
1608
1609
1610
1611
1612
1613
1614
1615
1616
1617
1618
1619
1620
1621
1622
1623
1624
1625
1626
1627
1628
1629
1630
1631
1632
1633
1634
1635
1636
1637
1638
1639
1640
1641
1642
1643
1644
1645
1646
1647
1648
1649
1650
1651
1652
1653
1654
1655
1656
1657
1658
1659
1660
1661
1662
1663
1664
1665
1666
1667
1668
1669
1670
1671
1672
1673
1674
1675
1676
1677
1678
1679
1680
1681
1682
1683
1684
1685
1686
1687
1688
1689
1690
1691
1692
1693
1694
1695
1696
1697
1698
1699
1700
1701
1702
1703
1704
1705
1706
1707
1708
1709
1710
1711
1712
1713
1714
1715
1716
1717
1718
1719
1720
1721
1722
1723
1724
1725
1726
1727
1728
1729
1730
1731
1732
1733
1734
1735
1736
1737
1738
1739
1740
1741
1742
1743
1744
1745
1746
1747
1748
1749
1750
1751
1752
1753
1754
1755
1756
1757
1758
1759
1760
1761
1762
1763
1764
1765
1766
1767
1768
// Copyright 2024 The Matrix.org Foundation C.I.C.
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
//     http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.

//! A send queue facility to serializing queuing and sending of messages.
//!
//! # [`Room`] send queue
//!
//! Each room gets its own [`RoomSendQueue`], that's available by calling
//! [`Room::send_queue()`]. The first time this method is called, it will spawn
//! a background task that's used to actually send events, in the order they
//! were passed from calls to [`RoomSendQueue::send()`].
//!
//! This queue tries to simplify error management around sending events, using
//! [`RoomSendQueue::send`] or [`RoomSendQueue::send_raw`]: by default, it will retry to send the
//! same event a few times, before automatically disabling itself, and emitting
//! a notification that can be listened to with the global send queue (see
//! paragraph below) or using [`RoomSendQueue::subscribe()`].
//!
//! It is possible to control whether a single room is enabled using
//! [`RoomSendQueue::set_enabled()`].
//!
//! # Global [`SendQueue`] object
//!
//! The [`Client::send_queue()`] method returns an API object allowing to
//! control all the room send queues:
//!
//! - enable/disable them all at once with [`SendQueue::set_enabled()`].
//! - get notifications about send errors with [`SendQueue::subscribe_errors`].
//! - reload all unsent events that had been persisted in storage using
//!   [`SendQueue::respawn_tasks_for_rooms_with_unsent_requests()`]. It is
//!   recommended to call this method during initialization of a client,
//!   otherwise persisted unsent events will only be re-sent after the send
//!   queue for the given room has been reopened for the first time.

use std::{
    collections::{BTreeMap, BTreeSet, HashMap},
    sync::{
        atomic::{AtomicBool, Ordering},
        Arc, RwLock as SyncRwLock,
    },
};

use matrix_sdk_base::{
    store::{
        ChildTransactionId, DependentQueuedRequest, DependentQueuedRequestKind, QueueWedgeError,
        QueuedRequest, QueuedRequestKind, SerializableEventContent,
    },
    RoomState, StoreError,
};
use matrix_sdk_common::executor::{spawn, JoinHandle};
use ruma::{
    events::{
        reaction::ReactionEventContent, relation::Annotation, AnyMessageLikeEventContent,
        EventContent as _,
    },
    serde::Raw,
    EventId, OwnedEventId, OwnedRoomId, OwnedTransactionId, TransactionId,
};
use tokio::sync::{broadcast, Notify, RwLock};
use tracing::{debug, error, info, instrument, trace, warn};

#[cfg(feature = "e2e-encryption")]
use crate::crypto::{OlmError, SessionRecipientCollectionError};
use crate::{
    client::WeakClient,
    config::RequestConfig,
    error::RetryKind,
    room::{edit::EditedContent, WeakRoom},
    Client, Room,
};

/// A client-wide send queue, for all the rooms known by a client.
pub struct SendQueue {
    client: Client,
}

#[cfg(not(tarpaulin_include))]
impl std::fmt::Debug for SendQueue {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("SendQueue").finish_non_exhaustive()
    }
}

impl SendQueue {
    pub(super) fn new(client: Client) -> Self {
        Self { client }
    }

    /// Reload all the rooms which had unsent requests, and respawn tasks for
    /// those rooms.
    pub async fn respawn_tasks_for_rooms_with_unsent_requests(&self) {
        if !self.is_enabled() {
            return;
        }

        let room_ids =
            self.client.store().load_rooms_with_unsent_requests().await.unwrap_or_else(|err| {
                warn!("error when loading rooms with unsent requests: {err}");
                Vec::new()
            });

        // Getting the [`RoomSendQueue`] is sufficient to spawn the task if needs be.
        for room_id in room_ids {
            if let Some(room) = self.client.get_room(&room_id) {
                let _ = self.for_room(room);
            }
        }
    }

    /// Tiny helper to get the send queue's global context from the [`Client`].
    #[inline(always)]
    fn data(&self) -> &SendQueueData {
        &self.client.inner.send_queue_data
    }

    /// Get or create a new send queue for a given room, and insert it into our
    /// memoized rooms mapping.
    fn for_room(&self, room: Room) -> RoomSendQueue {
        let data = self.data();

        let mut map = data.rooms.write().unwrap();

        let room_id = room.room_id();
        if let Some(room_q) = map.get(room_id).cloned() {
            return room_q;
        }

        let owned_room_id = room_id.to_owned();
        let room_q = RoomSendQueue::new(
            self.is_enabled(),
            data.error_reporter.clone(),
            data.is_dropping.clone(),
            &self.client,
            owned_room_id.clone(),
        );

        map.insert(owned_room_id, room_q.clone());

        room_q
    }

    /// Enable or disable the send queue for the entire client, i.e. all rooms.
    ///
    /// If we're disabling the queue, and requests were being sent, they're not
    /// aborted, and will continue until a status resolves (error responses
    /// will keep the events in the buffer of events to send later). The
    /// disablement will happen before the next request is sent.
    ///
    /// This may wake up background tasks and resume sending of requests in the
    /// background.
    pub async fn set_enabled(&self, enabled: bool) {
        debug!(?enabled, "setting global send queue enablement");

        self.data().globally_enabled.store(enabled, Ordering::SeqCst);

        // Wake up individual rooms we already know about.
        for room in self.data().rooms.read().unwrap().values() {
            room.set_enabled(enabled);
        }

        // Reload some extra rooms that might not have been awaken yet, but could have
        // requests from previous sessions.
        self.respawn_tasks_for_rooms_with_unsent_requests().await;
    }

    /// Returns whether the send queue is enabled, at a client-wide
    /// granularity.
    pub fn is_enabled(&self) -> bool {
        self.data().globally_enabled.load(Ordering::SeqCst)
    }

    /// A subscriber to the enablement status (enabled or disabled) of the
    /// send queue, along with useful errors.
    pub fn subscribe_errors(&self) -> broadcast::Receiver<SendQueueRoomError> {
        self.data().error_reporter.subscribe()
    }
}

/// A specific room's send queue ran into an error, and it has disabled itself.
#[derive(Clone, Debug)]
pub struct SendQueueRoomError {
    /// For which room is the send queue failing?
    pub room_id: OwnedRoomId,

    /// The error the room has ran into, when trying to send a request.
    pub error: Arc<crate::Error>,

    /// Whether the error is considered recoverable or not.
    ///
    /// An error that's recoverable will disable the room's send queue, while an
    /// unrecoverable error will be parked, until the user decides to do
    /// something about it.
    pub is_recoverable: bool,
}

impl Client {
    /// Returns a [`SendQueue`] that handles sending, retrying and not
    /// forgetting about requests that are to be sent.
    pub fn send_queue(&self) -> SendQueue {
        SendQueue::new(self.clone())
    }
}

pub(super) struct SendQueueData {
    /// Mapping of room to their unique send queue.
    rooms: SyncRwLock<BTreeMap<OwnedRoomId, RoomSendQueue>>,

    /// Is the whole mechanism enabled or disabled?
    ///
    /// This is only kept in memory to initialize new room queues with an
    /// initial enablement state.
    globally_enabled: AtomicBool,

    /// Global error updates for the send queue.
    error_reporter: broadcast::Sender<SendQueueRoomError>,

    /// Are we currently dropping the Client?
    is_dropping: Arc<AtomicBool>,
}

impl SendQueueData {
    /// Create the data for a send queue, in the given enabled state.
    pub fn new(globally_enabled: bool) -> Self {
        let (sender, _) = broadcast::channel(32);

        Self {
            rooms: Default::default(),
            globally_enabled: AtomicBool::new(globally_enabled),
            error_reporter: sender,
            is_dropping: Arc::new(false.into()),
        }
    }
}

impl Drop for SendQueueData {
    fn drop(&mut self) {
        // Mark the whole send queue as shutting down, then wake up all the room
        // queues so they're stopped too.
        debug!("globally dropping the send queue");
        self.is_dropping.store(true, Ordering::SeqCst);

        let rooms = self.rooms.read().unwrap();
        for room in rooms.values() {
            room.inner.notifier.notify_one();
        }
    }
}

impl Room {
    /// Returns the [`RoomSendQueue`] for this specific room.
    pub fn send_queue(&self) -> RoomSendQueue {
        self.client.send_queue().for_room(self.clone())
    }
}

/// A per-room send queue.
///
/// This is cheap to clone.
#[derive(Clone)]
pub struct RoomSendQueue {
    inner: Arc<RoomSendQueueInner>,
}

#[cfg(not(tarpaulin_include))]
impl std::fmt::Debug for RoomSendQueue {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("RoomSendQueue").finish_non_exhaustive()
    }
}

impl RoomSendQueue {
    fn new(
        globally_enabled: bool,
        global_error_reporter: broadcast::Sender<SendQueueRoomError>,
        is_dropping: Arc<AtomicBool>,
        client: &Client,
        room_id: OwnedRoomId,
    ) -> Self {
        let (updates_sender, _) = broadcast::channel(32);

        let queue = QueueStorage::new(WeakClient::from_client(client), room_id.clone());
        let notifier = Arc::new(Notify::new());

        let weak_room = WeakRoom::new(WeakClient::from_client(client), room_id);
        let locally_enabled = Arc::new(AtomicBool::new(globally_enabled));

        let task = spawn(Self::sending_task(
            weak_room.clone(),
            queue.clone(),
            notifier.clone(),
            updates_sender.clone(),
            locally_enabled.clone(),
            global_error_reporter,
            is_dropping,
        ));

        Self {
            inner: Arc::new(RoomSendQueueInner {
                room: weak_room,
                updates: updates_sender,
                _task: task,
                queue,
                notifier,
                locally_enabled,
            }),
        }
    }

    /// Queues a raw event for sending it to this room.
    ///
    /// This immediately returns, and will push the event to be sent into a
    /// queue, handled in the background.
    ///
    /// Callers are expected to consume [`RoomSendQueueUpdate`] via calling
    /// the [`Self::subscribe()`] method to get updates about the sending of
    /// that event.
    ///
    /// By default, if sending failed on the first attempt, it will be retried a
    /// few times. If sending failed after those retries, the entire
    /// client's sending queue will be disabled, and it will need to be
    /// manually re-enabled by the caller (e.g. after network is back, or when
    /// something has been done about the faulty requests).
    pub async fn send_raw(
        &self,
        content: Raw<AnyMessageLikeEventContent>,
        event_type: String,
    ) -> Result<SendHandle, RoomSendQueueError> {
        let Some(room) = self.inner.room.get() else {
            return Err(RoomSendQueueError::RoomDisappeared);
        };
        if room.state() != RoomState::Joined {
            return Err(RoomSendQueueError::RoomNotJoined);
        }

        let content = SerializableEventContent::from_raw(content, event_type);

        let transaction_id = self.inner.queue.push(content.clone()).await?;
        trace!(%transaction_id, "manager sends a raw event to the background task");

        self.inner.notifier.notify_one();

        let _ = self.inner.updates.send(RoomSendQueueUpdate::NewLocalEvent(LocalEcho {
            transaction_id: transaction_id.clone(),
            content: LocalEchoContent::Event {
                serialized_event: content,
                send_handle: SendHandle {
                    room: self.clone(),
                    transaction_id: transaction_id.clone(),
                },
                send_error: None,
            },
        }));

        Ok(SendHandle { transaction_id, room: self.clone() })
    }

    /// Queues an event for sending it to this room.
    ///
    /// This immediately returns, and will push the event to be sent into a
    /// queue, handled in the background.
    ///
    /// Callers are expected to consume [`RoomSendQueueUpdate`] via calling
    /// the [`Self::subscribe()`] method to get updates about the sending of
    /// that event.
    ///
    /// By default, if sending failed on the first attempt, it will be retried a
    /// few times. If sending failed after those retries, the entire
    /// client's sending queue will be disabled, and it will need to be
    /// manually re-enabled by the caller (e.g. after network is back, or when
    /// something has been done about the faulty requests).
    pub async fn send(
        &self,
        content: AnyMessageLikeEventContent,
    ) -> Result<SendHandle, RoomSendQueueError> {
        self.send_raw(
            Raw::new(&content).map_err(RoomSendQueueStorageError::JsonSerialization)?,
            content.event_type().to_string(),
        )
        .await
    }

    /// Returns the current local requests as well as a receiver to listen to
    /// the send queue updates, as defined in [`RoomSendQueueUpdate`].
    pub async fn subscribe(
        &self,
    ) -> Result<(Vec<LocalEcho>, broadcast::Receiver<RoomSendQueueUpdate>), RoomSendQueueError>
    {
        let local_echoes = self.inner.queue.local_echoes(self).await?;

        Ok((local_echoes, self.inner.updates.subscribe()))
    }

    /// A task that must be spawned in the async runtime, running in the
    /// background for each room that has a send queue.
    ///
    /// It only progresses forward: nothing can be cancelled at any point, which
    /// makes the implementation not overly complicated to follow.
    #[instrument(skip_all, fields(room_id = %room.room_id()))]
    async fn sending_task(
        room: WeakRoom,
        queue: QueueStorage,
        notifier: Arc<Notify>,
        updates: broadcast::Sender<RoomSendQueueUpdate>,
        locally_enabled: Arc<AtomicBool>,
        global_error_reporter: broadcast::Sender<SendQueueRoomError>,
        is_dropping: Arc<AtomicBool>,
    ) {
        info!("spawned the sending task");

        loop {
            // A request to shut down should be preferred above everything else.
            if is_dropping.load(Ordering::SeqCst) {
                trace!("shutting down!");
                break;
            }

            // Try to apply dependent requests now; those applying to previously failed
            // attempts (local echoes) would succeed now.
            if let Err(err) = queue.apply_dependent_requests().await {
                warn!("errors when applying dependent requests: {err}");
            }

            if !locally_enabled.load(Ordering::SeqCst) {
                trace!("not enabled, sleeping");
                // Wait for an explicit wakeup.
                notifier.notified().await;
                continue;
            }

            let queued_request = match queue.peek_next_to_send().await {
                Ok(Some(request)) => request,

                Ok(None) => {
                    trace!("queue is empty, sleeping");
                    // Wait for an explicit wakeup.
                    notifier.notified().await;
                    continue;
                }

                Err(err) => {
                    warn!("error when loading next request to send: {err}");
                    continue;
                }
            };

            trace!(txn_id = %queued_request.transaction_id, "received a request to send!");

            let Some(room) = room.get() else {
                if is_dropping.load(Ordering::SeqCst) {
                    break;
                }
                error!("the weak room couldn't be upgraded but we're not shutting down?");
                continue;
            };

            let (event, event_type) = match &queued_request.kind {
                QueuedRequestKind::Event { content } => content.raw(),
            };

            match room
                .send_raw(event_type, event)
                .with_transaction_id(&queued_request.transaction_id)
                .with_request_config(RequestConfig::short_retry())
                .await
            {
                Ok(res) => {
                    trace!(txn_id = %queued_request.transaction_id, event_id = %res.event_id, "successfully sent");

                    match queue.mark_as_sent(&queued_request.transaction_id, &res.event_id).await {
                        Ok(()) => {
                            let _ = updates.send(RoomSendQueueUpdate::SentEvent {
                                transaction_id: queued_request.transaction_id,
                                event_id: res.event_id,
                            });
                        }

                        Err(err) => {
                            warn!("unable to mark queued event as sent: {err}");
                        }
                    }
                }

                Err(err) => {
                    let is_recoverable = match err {
                        crate::Error::Http(ref http_err) => {
                            // All transient errors are recoverable.
                            matches!(
                                http_err.retry_kind(),
                                RetryKind::Transient { .. } | RetryKind::NetworkFailure
                            )
                        }

                        // `ConcurrentRequestFailed` typically happens because of an HTTP failure;
                        // since we don't get the underlying error, be lax and consider it
                        // recoverable, and let observers decide to retry it or not. At some point
                        // we'll get the actual underlying error.
                        crate::Error::ConcurrentRequestFailed => true,

                        // As of 2024-06-27, all other error types are considered unrecoverable.
                        _ => false,
                    };

                    if is_recoverable {
                        warn!(txn_id = %queued_request.transaction_id, error = ?err, "Recoverable error when sending request: {err}, disabling send queue");

                        // In this case, we intentionally keep the request in the queue, but mark it
                        // as not being sent anymore.
                        queue.mark_as_not_being_sent(&queued_request.transaction_id).await;

                        // Let observers know about a failure *after* we've marked the item as not
                        // being sent anymore. Otherwise, there's a possible race where a caller
                        // might try to remove an item, while it's still marked as being sent,
                        // resulting in a cancellation failure.

                        // Disable the queue for this room after a recoverable error happened. This
                        // should be the sign that this error is temporary (maybe network
                        // disconnected, maybe the server had a hiccup).
                        locally_enabled.store(false, Ordering::SeqCst);
                    } else {
                        warn!(txn_id = %queued_request.transaction_id, error = ?err, "Unrecoverable error when sending request: {err}");

                        // Mark the request as wedged, so it's not picked at any future point.

                        if let Err(storage_error) = queue
                            .mark_as_wedged(
                                &queued_request.transaction_id,
                                QueueWedgeError::from(&err),
                            )
                            .await
                        {
                            warn!("unable to mark request as wedged: {storage_error}");
                        }
                    }

                    let error = Arc::new(err);

                    let _ = global_error_reporter.send(SendQueueRoomError {
                        room_id: room.room_id().to_owned(),
                        error: error.clone(),
                        is_recoverable,
                    });

                    let _ = updates.send(RoomSendQueueUpdate::SendError {
                        transaction_id: queued_request.transaction_id,
                        error,
                        is_recoverable,
                    });
                }
            }
        }

        info!("exited sending task");
    }

    /// Returns whether the room is enabled, at the room level.
    pub fn is_enabled(&self) -> bool {
        self.inner.locally_enabled.load(Ordering::SeqCst)
    }

    /// Set the locally enabled flag for this room queue.
    pub fn set_enabled(&self, enabled: bool) {
        self.inner.locally_enabled.store(enabled, Ordering::SeqCst);

        // No need to wake a task to tell it it's been disabled, so only notify if we're
        // re-enabling the queue.
        if enabled {
            self.inner.notifier.notify_one();
        }
    }

    /// Unwedge a local echo identified by its transaction identifier and try to
    /// resend it.
    pub async fn unwedge(&self, transaction_id: &TransactionId) -> Result<(), RoomSendQueueError> {
        self.inner
            .queue
            .mark_as_unwedged(transaction_id)
            .await
            .map_err(RoomSendQueueError::StorageError)?;

        // Wake up the queue, in case the room was asleep before unwedging the request.
        self.inner.notifier.notify_one();

        let _ = self
            .inner
            .updates
            .send(RoomSendQueueUpdate::RetryEvent { transaction_id: transaction_id.to_owned() });

        Ok(())
    }
}

impl From<&crate::Error> for QueueWedgeError {
    fn from(value: &crate::Error) -> Self {
        match value {
            #[cfg(feature = "e2e-encryption")]
            crate::Error::OlmError(OlmError::SessionRecipientCollectionError(error)) => match error
            {
                SessionRecipientCollectionError::VerifiedUserHasUnsignedDevice(user_map) => {
                    QueueWedgeError::InsecureDevices { user_device_map: user_map.clone() }
                }

                SessionRecipientCollectionError::VerifiedUserChangedIdentity(users) => {
                    QueueWedgeError::IdentityViolations { users: users.clone() }
                }

                SessionRecipientCollectionError::CrossSigningNotSetup
                | SessionRecipientCollectionError::SendingFromUnverifiedDevice => {
                    QueueWedgeError::CrossVerificationRequired
                }
            },

            _ => QueueWedgeError::GenericApiError { msg: value.to_string() },
        }
    }
}

struct RoomSendQueueInner {
    /// The room which this send queue relates to.
    room: WeakRoom,

    /// Broadcaster for notifications about the statuses of requests to be sent.
    ///
    /// Can be subscribed to from the outside.
    updates: broadcast::Sender<RoomSendQueueUpdate>,

    /// Queue of requests that are either to be sent, or being sent.
    ///
    /// When a request has been sent to the server, it is removed from that
    /// queue *after* being sent. That way, we will retry sending upon
    /// failure, in the same order requests have been inserted in the first
    /// place.
    queue: QueueStorage,

    /// A notifier that's updated any time common data is touched (stopped or
    /// enabled statuses), or the associated room [`QueueStorage`].
    notifier: Arc<Notify>,

    /// Should the room process new requests or not (because e.g. it might be
    /// running off the network)?
    locally_enabled: Arc<AtomicBool>,

    /// Handle to the actual sending task. Unused, but kept alive along this
    /// data structure.
    _task: JoinHandle<()>,
}

#[derive(Clone)]
struct QueueStorage {
    /// Reference to the client, to get access to the underlying store.
    client: WeakClient,

    /// To which room is this storage related.
    room_id: OwnedRoomId,

    /// All the queued requests that are being sent at the moment.
    ///
    /// It also serves as an internal lock on the storage backend.
    being_sent: Arc<RwLock<BTreeSet<OwnedTransactionId>>>,
}

impl QueueStorage {
    /// Create a new queue for queuing requests to be sent later.
    fn new(client: WeakClient, room: OwnedRoomId) -> Self {
        Self { room_id: room, being_sent: Default::default(), client }
    }

    /// Small helper to get a strong Client from the weak one.
    fn client(&self) -> Result<Client, RoomSendQueueStorageError> {
        self.client.get().ok_or(RoomSendQueueStorageError::ClientShuttingDown)
    }

    /// Push a new event to be sent in the queue.
    ///
    /// Returns the transaction id chosen to identify the request.
    async fn push(
        &self,
        serializable: SerializableEventContent,
    ) -> Result<OwnedTransactionId, RoomSendQueueStorageError> {
        let transaction_id = TransactionId::new();

        self.client()?
            .store()
            .save_send_queue_request(&self.room_id, transaction_id.clone(), serializable)
            .await?;

        Ok(transaction_id)
    }

    /// Peeks the next request to be sent, marking it as being sent.
    ///
    /// It is required to call [`Self::mark_as_sent`] after it's been
    /// effectively sent.
    async fn peek_next_to_send(&self) -> Result<Option<QueuedRequest>, RoomSendQueueStorageError> {
        // Keep the lock until we're done touching the storage.
        let mut being_sent = self.being_sent.write().await;

        let queued_requests =
            self.client()?.store().load_send_queue_requests(&self.room_id).await?;

        if let Some(request) = queued_requests.iter().find(|queued| !queued.is_wedged()) {
            being_sent.insert(request.transaction_id.clone());

            Ok(Some(request.clone()))
        } else {
            Ok(None)
        }
    }

    /// Marks a request popped with [`Self::peek_next_to_send`] and identified
    /// with the given transaction id as not being sent anymore, so it can
    /// be removed from the queue later.
    async fn mark_as_not_being_sent(&self, transaction_id: &TransactionId) {
        self.being_sent.write().await.remove(transaction_id);
    }

    /// Marks a request popped with [`Self::peek_next_to_send`] and identified
    /// with the given transaction id as being wedged (and not being sent
    /// anymore), so it can be removed from the queue later.
    async fn mark_as_wedged(
        &self,
        transaction_id: &TransactionId,
        reason: QueueWedgeError,
    ) -> Result<(), RoomSendQueueStorageError> {
        // Keep the lock until we're done touching the storage.
        let mut being_sent = self.being_sent.write().await;
        being_sent.remove(transaction_id);

        Ok(self
            .client()?
            .store()
            .update_send_queue_request_status(&self.room_id, transaction_id, Some(reason))
            .await?)
    }

    /// Marks a request identified with the given transaction id as being now
    /// unwedged and adds it back to the queue.
    async fn mark_as_unwedged(
        &self,
        transaction_id: &TransactionId,
    ) -> Result<(), RoomSendQueueStorageError> {
        Ok(self
            .client()?
            .store()
            .update_send_queue_request_status(&self.room_id, transaction_id, None)
            .await?)
    }

    /// Marks a request pushed with [`Self::push`] and identified with the given
    /// transaction id as sent, by removing it from the local queue.
    async fn mark_as_sent(
        &self,
        transaction_id: &TransactionId,
        event_id: &EventId,
    ) -> Result<(), RoomSendQueueStorageError> {
        // Keep the lock until we're done touching the storage.
        let mut being_sent = self.being_sent.write().await;
        being_sent.remove(transaction_id);

        let client = self.client()?;
        let store = client.store();

        // Update all dependent requests.
        store
            .update_dependent_queued_request(&self.room_id, transaction_id, event_id.to_owned())
            .await?;

        let removed = store.remove_send_queue_request(&self.room_id, transaction_id).await?;

        if !removed {
            warn!(txn_id = %transaction_id, "request marked as sent was missing from storage");
        }

        Ok(())
    }

    /// Cancel a sending command for an event that has been sent with
    /// [`Self::push`] with the given transaction id.
    ///
    /// Returns whether the given transaction has been effectively removed. If
    /// false, this either means that the transaction id was unrelated to
    /// this queue, or that the request was sent before we cancelled it.
    async fn cancel_event(
        &self,
        transaction_id: &TransactionId,
    ) -> Result<bool, RoomSendQueueStorageError> {
        // Keep the lock until we're done touching the storage.
        let being_sent = self.being_sent.read().await;

        if being_sent.contains(transaction_id) {
            // Save the intent to redact the event.
            self.client()?
                .store()
                .save_dependent_queued_request(
                    &self.room_id,
                    transaction_id,
                    ChildTransactionId::new(),
                    DependentQueuedRequestKind::RedactEvent,
                )
                .await?;

            return Ok(true);
        }

        let removed =
            self.client()?.store().remove_send_queue_request(&self.room_id, transaction_id).await?;

        Ok(removed)
    }

    /// Replace an event that has been sent with [`Self::push`] with the given
    /// transaction id, before it's been actually sent.
    ///
    /// Returns whether the given transaction has been effectively edited. If
    /// false, this either means that the transaction id was unrelated to
    /// this queue, or that the request was sent before we edited it.
    async fn replace_event(
        &self,
        transaction_id: &TransactionId,
        serializable: SerializableEventContent,
    ) -> Result<bool, RoomSendQueueStorageError> {
        // Keep the lock until we're done touching the storage.
        let being_sent = self.being_sent.read().await;

        if being_sent.contains(transaction_id) {
            // Save the intent to edit the associated event.
            self.client()?
                .store()
                .save_dependent_queued_request(
                    &self.room_id,
                    transaction_id,
                    ChildTransactionId::new(),
                    DependentQueuedRequestKind::EditEvent { new_content: serializable },
                )
                .await?;

            return Ok(true);
        }

        let edited = self
            .client()?
            .store()
            .update_send_queue_request(&self.room_id, transaction_id, serializable)
            .await?;

        Ok(edited)
    }

    /// Reacts to the given local echo of an event.
    #[instrument(skip(self))]
    async fn react(
        &self,
        transaction_id: &TransactionId,
        key: String,
    ) -> Result<Option<ChildTransactionId>, RoomSendQueueStorageError> {
        let client = self.client()?;
        let store = client.store();

        let requests = store.load_send_queue_requests(&self.room_id).await?;

        // If the target event has been already sent, abort immediately.
        if !requests.iter().any(|item| item.transaction_id == transaction_id) {
            return Ok(None);
        }

        // Record the dependent request.
        let reaction_txn_id = ChildTransactionId::new();
        store
            .save_dependent_queued_request(
                &self.room_id,
                transaction_id,
                reaction_txn_id.clone(),
                DependentQueuedRequestKind::ReactEvent { key },
            )
            .await?;

        Ok(Some(reaction_txn_id))
    }

    /// Returns a list of the local echoes, that is, all the requests that we're
    /// about to send but that haven't been sent yet (or are being sent).
    async fn local_echoes(
        &self,
        room: &RoomSendQueue,
    ) -> Result<Vec<LocalEcho>, RoomSendQueueStorageError> {
        let client = self.client()?;
        let store = client.store();

        let local_requests =
            store.load_send_queue_requests(&self.room_id).await?.into_iter().map(|queued| {
                LocalEcho {
                    transaction_id: queued.transaction_id.clone(),
                    content: match queued.kind {
                        QueuedRequestKind::Event { content } => LocalEchoContent::Event {
                            serialized_event: content,
                            send_handle: SendHandle {
                                room: room.clone(),
                                transaction_id: queued.transaction_id,
                            },
                            send_error: queued.error,
                        },
                    },
                }
            });

        let local_reactions =
            store.load_dependent_queued_requests(&self.room_id).await?.into_iter().filter_map(
                |dep| match dep.kind {
                    DependentQueuedRequestKind::EditEvent { .. }
                    | DependentQueuedRequestKind::RedactEvent => {
                        // TODO: reflect local edits/redacts too?
                        None
                    }
                    DependentQueuedRequestKind::ReactEvent { key } => Some(LocalEcho {
                        transaction_id: dep.own_transaction_id.clone().into(),
                        content: LocalEchoContent::React {
                            key,
                            send_handle: SendReactionHandle {
                                room: room.clone(),
                                transaction_id: dep.own_transaction_id,
                            },
                            applies_to: dep.parent_transaction_id,
                        },
                    }),
                },
            );

        Ok(local_requests.chain(local_reactions).collect())
    }

    /// Try to apply a single dependent request, whether it's local or remote.
    ///
    /// This swallows errors that would retrigger every time if we retried
    /// applying the dependent request: invalid edit content, etc.
    ///
    /// Returns true if the dependent request has been sent (or should not be
    /// retried later).
    #[instrument(skip_all)]
    async fn try_apply_single_dependent_request(
        &self,
        client: &Client,
        de: DependentQueuedRequest,
    ) -> Result<bool, RoomSendQueueError> {
        let store = client.store();

        match de.kind {
            DependentQueuedRequestKind::EditEvent { new_content } => {
                if let Some(event_id) = de.event_id {
                    // The parent event has been sent, so send an edit event.
                    let room = client
                        .get_room(&self.room_id)
                        .ok_or(RoomSendQueueError::RoomDisappeared)?;

                    // Check the event is one we know how to edit with an edit event.

                    // It must be deserializable…
                    let edited_content = match new_content.deserialize() {
                        Ok(AnyMessageLikeEventContent::RoomMessage(c)) => {
                            // Assume no relationships.
                            EditedContent::RoomMessage(c.into())
                        }

                        Ok(AnyMessageLikeEventContent::UnstablePollStart(c)) => {
                            let poll_start = c.poll_start().clone();
                            EditedContent::PollStart {
                                fallback_text: poll_start.question.text.clone(),
                                new_content: poll_start,
                            }
                        }

                        Ok(c) => {
                            warn!("Unsupported edit content type: {:?}", c.event_type());
                            return Ok(true);
                        }

                        Err(err) => {
                            warn!("Unable to deserialize: {err}");
                            return Ok(true);
                        }
                    };

                    let edit_event = match room.make_edit_event(&event_id, edited_content).await {
                        Ok(e) => e,
                        Err(err) => {
                            warn!("couldn't create edited event: {err}");
                            return Ok(true);
                        }
                    };

                    // Queue the edit event in the send queue 🧠.
                    let serializable = SerializableEventContent::from_raw(
                        Raw::new(&edit_event)
                            .map_err(RoomSendQueueStorageError::JsonSerialization)?,
                        edit_event.event_type().to_string(),
                    );

                    store
                        .save_send_queue_request(
                            &self.room_id,
                            de.own_transaction_id.into(),
                            serializable,
                        )
                        .await
                        .map_err(RoomSendQueueStorageError::StorageError)?;
                } else {
                    // The parent event is still local; update the local echo.
                    let edited = store
                        .update_send_queue_request(
                            &self.room_id,
                            &de.parent_transaction_id,
                            new_content,
                        )
                        .await
                        .map_err(RoomSendQueueStorageError::StorageError)?;

                    if !edited {
                        warn!("missing local echo upon dependent edit");
                    }
                }
            }

            DependentQueuedRequestKind::RedactEvent => {
                if let Some(event_id) = de.event_id {
                    // The parent event has been sent; send a redaction.
                    let room = client
                        .get_room(&self.room_id)
                        .ok_or(RoomSendQueueError::RoomDisappeared)?;

                    // Ideally we'd use the send queue to send the redaction, but the protocol has
                    // changed the shape of a room.redaction after v11, so keep it simple and try
                    // once here.

                    // Note: no reason is provided because we materialize the intent of "cancel
                    // sending the parent event".

                    if let Err(err) =
                        room.redact(&event_id, None, Some(de.own_transaction_id.into())).await
                    {
                        warn!("error when sending a redact for {event_id}: {err}");
                        return Ok(false);
                    }
                } else {
                    // The parent event is still local (sending must have failed); redact the local
                    // echo.
                    let removed = store
                        .remove_send_queue_request(&self.room_id, &de.parent_transaction_id)
                        .await
                        .map_err(RoomSendQueueStorageError::StorageError)?;

                    if !removed {
                        warn!("missing local echo upon dependent redact");
                    }
                }
            }

            DependentQueuedRequestKind::ReactEvent { key } => {
                if let Some(event_id) = de.event_id {
                    // Queue the reaction event in the send queue 🧠.
                    let react_event =
                        ReactionEventContent::new(Annotation::new(event_id, key)).into();
                    let serializable = SerializableEventContent::from_raw(
                        Raw::new(&react_event)
                            .map_err(RoomSendQueueStorageError::JsonSerialization)?,
                        react_event.event_type().to_string(),
                    );

                    store
                        .save_send_queue_request(
                            &self.room_id,
                            de.own_transaction_id.into(),
                            serializable,
                        )
                        .await
                        .map_err(RoomSendQueueStorageError::StorageError)?;
                } else {
                    // Not applied yet, we should retry later => false.
                    return Ok(false);
                }
            }
        }

        Ok(true)
    }

    #[instrument(skip(self))]
    async fn apply_dependent_requests(&self) -> Result<(), RoomSendQueueError> {
        // Keep the lock until we're done touching the storage.
        let _being_sent = self.being_sent.read().await;

        let client = self.client()?;
        let store = client.store();

        let dependent_requests = store
            .load_dependent_queued_requests(&self.room_id)
            .await
            .map_err(RoomSendQueueStorageError::StorageError)?;

        let num_initial_dependent_requests = dependent_requests.len();
        if num_initial_dependent_requests == 0 {
            // Returning early here avoids a bit of useless logging.
            return Ok(());
        }

        let canonicalized_dependent_requests = canonicalize_dependent_requests(&dependent_requests);

        // Get rid of the all non-canonical dependent events.
        for original in &dependent_requests {
            if !canonicalized_dependent_requests
                .iter()
                .any(|canonical| canonical.own_transaction_id == original.own_transaction_id)
            {
                store
                    .remove_dependent_queued_request(&self.room_id, &original.own_transaction_id)
                    .await
                    .map_err(RoomSendQueueStorageError::StorageError)?;
            }
        }

        let mut num_dependent_requests = canonicalized_dependent_requests.len();

        debug!(
            num_dependent_requests,
            num_initial_dependent_requests, "starting handling of dependent requests"
        );

        for dependent in canonicalized_dependent_requests {
            let dependent_id = dependent.own_transaction_id.clone();

            match self.try_apply_single_dependent_request(&client, dependent).await {
                Ok(should_remove) => {
                    if should_remove {
                        // The dependent request has been successfully applied, forget about it.
                        store
                            .remove_dependent_queued_request(&self.room_id, &dependent_id)
                            .await
                            .map_err(RoomSendQueueStorageError::StorageError)?;

                        num_dependent_requests -= 1;
                    }
                }

                Err(err) => {
                    warn!("error when applying single dependent request: {err}");
                }
            }
        }

        debug!(
            leftover_dependent_requests = num_dependent_requests,
            "stopped handling dependent request"
        );

        Ok(())
    }

    /// Remove a single dependent request from storage.
    async fn remove_dependent_send_queue_request(
        &self,
        dependent_event_id: &ChildTransactionId,
    ) -> Result<bool, RoomSendQueueStorageError> {
        // Keep the lock until we're done touching the storage.
        let _being_sent = self.being_sent.read().await;

        Ok(self
            .client()?
            .store()
            .remove_dependent_queued_request(&self.room_id, dependent_event_id)
            .await?)
    }
}

/// The content of a local echo.
#[derive(Clone, Debug)]
pub enum LocalEchoContent {
    /// The local echo contains an actual event ready to display.
    Event {
        /// Content of the event itself (along with its type) that we are about
        /// to send.
        serialized_event: SerializableEventContent,
        /// A handle to manipulate the sending of the associated event.
        send_handle: SendHandle,
        /// Whether trying to send this local echo failed in the past with an
        /// unrecoverable error (see [`SendQueueRoomError::is_recoverable`]).
        send_error: Option<QueueWedgeError>,
    },

    /// A local echo has been reacted to.
    React {
        /// The key with which the local echo has been reacted to.
        key: String,
        /// A handle to manipulate the sending of the reaction.
        send_handle: SendReactionHandle,
        /// The local echo which has been reacted to.
        applies_to: OwnedTransactionId,
    },
}

/// A local representation for a request that hasn't been sent yet to the user's
/// homeserver.
#[derive(Clone, Debug)]
pub struct LocalEcho {
    /// Transaction id used to identify the associated request.
    pub transaction_id: OwnedTransactionId,
    /// The content for the local echo.
    pub content: LocalEchoContent,
}

/// An update to a room send queue, observable with
/// [`RoomSendQueue::subscribe`].
#[derive(Clone, Debug)]
pub enum RoomSendQueueUpdate {
    /// A new local event is being sent.
    ///
    /// There's been a user query to create this event. It is being sent to the
    /// server.
    NewLocalEvent(LocalEcho),

    /// A local event that hadn't been sent to the server yet has been cancelled
    /// before sending.
    CancelledLocalEvent {
        /// Transaction id used to identify this event.
        transaction_id: OwnedTransactionId,
    },

    /// A local event's content has been replaced with something else.
    ReplacedLocalEvent {
        /// Transaction id used to identify this event.
        transaction_id: OwnedTransactionId,

        /// The new content replacing the previous one.
        new_content: SerializableEventContent,
    },

    /// An error happened when an event was being sent.
    ///
    /// The event has not been removed from the queue. All the send queues
    /// will be disabled after this happens, and must be manually re-enabled.
    SendError {
        /// Transaction id used to identify this event.
        transaction_id: OwnedTransactionId,
        /// Error received while sending the event.
        error: Arc<crate::Error>,
        /// Whether the error is considered recoverable or not.
        ///
        /// An error that's recoverable will disable the room's send queue,
        /// while an unrecoverable error will be parked, until the user
        /// decides to cancel sending it.
        is_recoverable: bool,
    },

    /// The event has been unwedged and sending is now being retried.
    RetryEvent {
        /// Transaction id used to identify this event.
        transaction_id: OwnedTransactionId,
    },

    /// The event has been sent to the server, and the query returned
    /// successfully.
    SentEvent {
        /// Transaction id used to identify this event.
        transaction_id: OwnedTransactionId,
        /// Received event id from the send response.
        event_id: OwnedEventId,
    },
}

/// An error triggered by the send queue module.
#[derive(Debug, thiserror::Error)]
pub enum RoomSendQueueError {
    /// The room isn't in the joined state.
    #[error("the room isn't in the joined state")]
    RoomNotJoined,

    /// The room is missing from the client.
    ///
    /// This happens only whenever the client is shutting down.
    #[error("the room is now missing from the client")]
    RoomDisappeared,

    /// Error coming from storage.
    #[error(transparent)]
    StorageError(#[from] RoomSendQueueStorageError),
}

/// An error triggered by the send queue storage.
#[derive(Debug, thiserror::Error)]
pub enum RoomSendQueueStorageError {
    /// Error caused by the state store.
    #[error(transparent)]
    StorageError(#[from] StoreError),

    /// Error caused when (de)serializing into/from json.
    #[error(transparent)]
    JsonSerialization(#[from] serde_json::Error),

    /// The client is shutting down.
    #[error("The client is shutting down.")]
    ClientShuttingDown,
}

/// A handle to manipulate an event that was scheduled to be sent to a room.
// TODO (bnjbvr): consider renaming `SendEventHandle`, unless we can reuse it for medias too.
#[derive(Clone, Debug)]
pub struct SendHandle {
    room: RoomSendQueue,
    transaction_id: OwnedTransactionId,
}

impl SendHandle {
    /// Aborts the sending of the event, if it wasn't sent yet.
    ///
    /// Returns true if the sending could be aborted, false if not (i.e. the
    /// event had already been sent).
    #[instrument(skip(self), fields(room_id = %self.room.inner.room.room_id(), txn_id = %self.transaction_id))]
    pub async fn abort(&self) -> Result<bool, RoomSendQueueStorageError> {
        trace!("received an abort request");

        if self.room.inner.queue.cancel_event(&self.transaction_id).await? {
            trace!("successful abort");

            // Propagate a cancelled update too.
            let _ = self.room.inner.updates.send(RoomSendQueueUpdate::CancelledLocalEvent {
                transaction_id: self.transaction_id.clone(),
            });

            Ok(true)
        } else {
            debug!("local echo didn't exist anymore, can't abort");
            Ok(false)
        }
    }

    /// Edits the content of a local echo with a raw event content.
    ///
    /// Returns true if the event to be sent was replaced, false if not (i.e.
    /// the event had already been sent).
    #[instrument(skip(self, new_content), fields(room_id = %self.room.inner.room.room_id(), txn_id = %self.transaction_id))]
    pub async fn edit_raw(
        &self,
        new_content: Raw<AnyMessageLikeEventContent>,
        event_type: String,
    ) -> Result<bool, RoomSendQueueStorageError> {
        trace!("received an edit request");

        let serializable = SerializableEventContent::from_raw(new_content, event_type);

        if self.room.inner.queue.replace_event(&self.transaction_id, serializable.clone()).await? {
            trace!("successful edit");

            // Wake up the queue, in case the room was asleep before the edit.
            self.room.inner.notifier.notify_one();

            // Propagate a replaced update too.
            let _ = self.room.inner.updates.send(RoomSendQueueUpdate::ReplacedLocalEvent {
                transaction_id: self.transaction_id.clone(),
                new_content: serializable,
            });

            Ok(true)
        } else {
            debug!("local echo doesn't exist anymore, can't edit");
            Ok(false)
        }
    }

    /// Edits the content of a local echo with an event content.
    ///
    /// Returns true if the event to be sent was replaced, false if not (i.e.
    /// the event had already been sent).
    pub async fn edit(
        &self,
        new_content: AnyMessageLikeEventContent,
    ) -> Result<bool, RoomSendQueueStorageError> {
        self.edit_raw(
            Raw::new(&new_content).map_err(RoomSendQueueStorageError::JsonSerialization)?,
            new_content.event_type().to_string(),
        )
        .await
    }

    /// Send a reaction to the event as soon as it's sent.
    ///
    /// If returning `Ok(None)`; this means the reaction couldn't be sent
    /// because the event is already a remote one.
    #[instrument(skip(self), fields(room_id = %self.room.inner.room.room_id(), txn_id = %self.transaction_id))]
    pub async fn react(
        &self,
        key: String,
    ) -> Result<Option<SendReactionHandle>, RoomSendQueueStorageError> {
        trace!("received an intent to react");

        if let Some(reaction_txn_id) =
            self.room.inner.queue.react(&self.transaction_id, key.clone()).await?
        {
            trace!("successfully queued react");

            // Wake up the queue, in case the room was asleep before the sending.
            self.room.inner.notifier.notify_one();

            // Propagate a new local event.
            let send_handle = SendReactionHandle {
                room: self.room.clone(),
                transaction_id: reaction_txn_id.clone(),
            };

            let _ = self.room.inner.updates.send(RoomSendQueueUpdate::NewLocalEvent(LocalEcho {
                // Note: we do want to use the txn_id we're going to use for the reaction, not the
                // one for the event we're reacting to.
                transaction_id: reaction_txn_id.into(),
                content: LocalEchoContent::React {
                    key,
                    send_handle: send_handle.clone(),
                    applies_to: self.transaction_id.clone(),
                },
            }));

            Ok(Some(send_handle))
        } else {
            debug!("local echo doesn't exist anymore, can't react");
            Ok(None)
        }
    }
}

/// A handle to execute actions on the sending of a reaction.
#[derive(Clone, Debug)]
pub struct SendReactionHandle {
    /// Reference to the send queue for the room where this reaction was sent.
    room: RoomSendQueue,
    /// The own transaction id for the reaction.
    transaction_id: ChildTransactionId,
}

impl SendReactionHandle {
    /// Abort the sending of the reaction.
    ///
    /// Will return true if the reaction could be aborted, false if it's been
    /// sent (and there's no matching local echo anymore).
    pub async fn abort(&self) -> Result<bool, RoomSendQueueStorageError> {
        if self.room.inner.queue.remove_dependent_send_queue_request(&self.transaction_id).await? {
            // Simple case: the reaction was found in the dependent event list.

            // Propagate a cancelled update too.
            let _ = self.room.inner.updates.send(RoomSendQueueUpdate::CancelledLocalEvent {
                transaction_id: self.transaction_id.clone().into(),
            });

            return Ok(true);
        }

        // The reaction has already been queued for sending, try to abort it using a
        // regular abort.
        let handle = SendHandle {
            room: self.room.clone(),
            transaction_id: self.transaction_id.clone().into(),
        };

        handle.abort().await
    }

    /// The transaction id that will be used to send this reaction later.
    pub fn transaction_id(&self) -> &TransactionId {
        &self.transaction_id
    }
}

/// From a given source of [`DependentQueuedRequest`], return only the most
/// meaningful, i.e. the ones that wouldn't be overridden after applying the
/// others.
fn canonicalize_dependent_requests(
    dependent: &[DependentQueuedRequest],
) -> Vec<DependentQueuedRequest> {
    let mut by_event_id = HashMap::<OwnedTransactionId, Vec<&DependentQueuedRequest>>::new();

    for d in dependent {
        let prevs = by_event_id.entry(d.parent_transaction_id.clone()).or_default();

        if prevs.iter().any(|prev| matches!(prev.kind, DependentQueuedRequestKind::RedactEvent)) {
            // The parent event has already been flagged for redaction, don't consider the
            // other dependent events.
            continue;
        }

        match &d.kind {
            DependentQueuedRequestKind::EditEvent { .. } => {
                // Replace any previous edit with this one.
                if let Some(prev_edit) = prevs
                    .iter_mut()
                    .find(|prev| matches!(prev.kind, DependentQueuedRequestKind::EditEvent { .. }))
                {
                    *prev_edit = d;
                } else {
                    prevs.insert(0, d);
                }
            }

            DependentQueuedRequestKind::ReactEvent { .. } => {
                prevs.push(d);
            }

            DependentQueuedRequestKind::RedactEvent => {
                // Remove every other dependent action.
                prevs.clear();
                prevs.push(d);
            }
        }
    }

    by_event_id
        .into_iter()
        .flat_map(|(_parent_txn_id, entries)| entries.into_iter().cloned())
        .collect()
}

#[cfg(all(test, not(target_arch = "wasm32")))]
mod tests {
    use std::{sync::Arc, time::Duration};

    use assert_matches2::{assert_let, assert_matches};
    use matrix_sdk_base::store::{
        ChildTransactionId, DependentQueuedRequest, DependentQueuedRequestKind,
        SerializableEventContent,
    };
    use matrix_sdk_test::{async_test, JoinedRoomBuilder, SyncResponseBuilder};
    use ruma::{
        events::{room::message::RoomMessageEventContent, AnyMessageLikeEventContent},
        room_id, TransactionId,
    };

    use super::canonicalize_dependent_requests;
    use crate::{client::WeakClient, test_utils::logged_in_client};

    #[async_test]
    async fn test_client_no_cycle_with_send_queue() {
        for enabled in [true, false] {
            let client = logged_in_client(None).await;
            let weak_client = WeakClient::from_client(&client);

            {
                let mut sync_response_builder = SyncResponseBuilder::new();

                let room_id = room_id!("!a:b.c");

                // Make sure the client knows about the room.
                client
                    .base_client()
                    .receive_sync_response(
                        sync_response_builder
                            .add_joined_room(JoinedRoomBuilder::new(room_id))
                            .build_sync_response(),
                    )
                    .await
                    .unwrap();

                let room = client.get_room(room_id).unwrap();
                let q = room.send_queue();

                let _watcher = q.subscribe().await;

                client.send_queue().set_enabled(enabled).await;
            }

            drop(client);

            // Give a bit of time for background tasks to die.
            tokio::time::sleep(Duration::from_millis(500)).await;

            // The weak client must be the last reference to the client now.
            let client = weak_client.get();
            assert!(
                client.is_none(),
                "too many strong references to the client: {}",
                Arc::strong_count(&client.unwrap().inner)
            );
        }
    }

    #[test]
    fn test_canonicalize_dependent_events_smoke_test() {
        // Smoke test: canonicalizing a single dependent event returns it.
        let txn = TransactionId::new();

        let edit = DependentQueuedRequest {
            own_transaction_id: ChildTransactionId::new(),
            parent_transaction_id: txn.clone(),
            kind: DependentQueuedRequestKind::EditEvent {
                new_content: SerializableEventContent::new(
                    &RoomMessageEventContent::text_plain("edit").into(),
                )
                .unwrap(),
            },
            event_id: None,
        };
        let res = canonicalize_dependent_requests(&[edit]);

        assert_eq!(res.len(), 1);
        assert_matches!(&res[0].kind, DependentQueuedRequestKind::EditEvent { .. });
        assert_eq!(res[0].parent_transaction_id, txn);
        assert!(res[0].event_id.is_none());
    }

    #[test]
    fn test_canonicalize_dependent_events_redaction_preferred() {
        // A redaction is preferred over any other kind of dependent event.
        let txn = TransactionId::new();

        let mut inputs = Vec::with_capacity(100);
        let redact = DependentQueuedRequest {
            own_transaction_id: ChildTransactionId::new(),
            parent_transaction_id: txn.clone(),
            kind: DependentQueuedRequestKind::RedactEvent,
            event_id: None,
        };

        let edit = DependentQueuedRequest {
            own_transaction_id: ChildTransactionId::new(),
            parent_transaction_id: txn.clone(),
            kind: DependentQueuedRequestKind::EditEvent {
                new_content: SerializableEventContent::new(
                    &RoomMessageEventContent::text_plain("edit").into(),
                )
                .unwrap(),
            },
            event_id: None,
        };

        inputs.push({
            let mut edit = edit.clone();
            edit.own_transaction_id = ChildTransactionId::new();
            edit
        });

        inputs.push(redact);

        for _ in 0..98 {
            let mut edit = edit.clone();
            edit.own_transaction_id = ChildTransactionId::new();
            inputs.push(edit);
        }

        let res = canonicalize_dependent_requests(&inputs);

        assert_eq!(res.len(), 1);
        assert_matches!(&res[0].kind, DependentQueuedRequestKind::RedactEvent);
        assert_eq!(res[0].parent_transaction_id, txn);
    }

    #[test]
    fn test_canonicalize_dependent_events_last_edit_preferred() {
        let parent_txn = TransactionId::new();

        // The latest edit of a list is always preferred.
        let inputs = (0..10)
            .map(|i| DependentQueuedRequest {
                own_transaction_id: ChildTransactionId::new(),
                parent_transaction_id: parent_txn.clone(),
                kind: DependentQueuedRequestKind::EditEvent {
                    new_content: SerializableEventContent::new(
                        &RoomMessageEventContent::text_plain(format!("edit{i}")).into(),
                    )
                    .unwrap(),
                },
                event_id: None,
            })
            .collect::<Vec<_>>();

        let txn = inputs[9].parent_transaction_id.clone();

        let res = canonicalize_dependent_requests(&inputs);

        assert_eq!(res.len(), 1);
        assert_let!(DependentQueuedRequestKind::EditEvent { new_content } = &res[0].kind);
        assert_let!(
            AnyMessageLikeEventContent::RoomMessage(msg) = new_content.deserialize().unwrap()
        );
        assert_eq!(msg.body(), "edit9");
        assert_eq!(res[0].parent_transaction_id, txn);
    }

    #[test]
    fn test_canonicalize_multiple_local_echoes() {
        let txn1 = TransactionId::new();
        let txn2 = TransactionId::new();

        let child1 = ChildTransactionId::new();
        let child2 = ChildTransactionId::new();

        let inputs = vec![
            // This one pertains to txn1.
            DependentQueuedRequest {
                own_transaction_id: child1.clone(),
                kind: DependentQueuedRequestKind::RedactEvent,
                parent_transaction_id: txn1.clone(),
                event_id: None,
            },
            // This one pertains to txn2.
            DependentQueuedRequest {
                own_transaction_id: child2,
                kind: DependentQueuedRequestKind::EditEvent {
                    new_content: SerializableEventContent::new(
                        &RoomMessageEventContent::text_plain("edit").into(),
                    )
                    .unwrap(),
                },
                parent_transaction_id: txn2.clone(),
                event_id: None,
            },
        ];

        let res = canonicalize_dependent_requests(&inputs);

        // The canonicalization shouldn't depend per event id.
        assert_eq!(res.len(), 2);

        for dependent in res {
            if dependent.own_transaction_id == child1 {
                assert_eq!(dependent.parent_transaction_id, txn1);
                assert_matches!(dependent.kind, DependentQueuedRequestKind::RedactEvent);
            } else {
                assert_eq!(dependent.parent_transaction_id, txn2);
                assert_matches!(dependent.kind, DependentQueuedRequestKind::EditEvent { .. });
            }
        }
    }

    #[test]
    fn test_canonicalize_reactions_after_edits() {
        // Sending reactions should happen after edits to a given event.
        let txn = TransactionId::new();

        let react_id = ChildTransactionId::new();
        let react = DependentQueuedRequest {
            own_transaction_id: react_id.clone(),
            kind: DependentQueuedRequestKind::ReactEvent { key: "🧠".to_owned() },
            parent_transaction_id: txn.clone(),
            event_id: None,
        };

        let edit_id = ChildTransactionId::new();
        let edit = DependentQueuedRequest {
            own_transaction_id: edit_id.clone(),
            kind: DependentQueuedRequestKind::EditEvent {
                new_content: SerializableEventContent::new(
                    &RoomMessageEventContent::text_plain("edit").into(),
                )
                .unwrap(),
            },
            parent_transaction_id: txn,
            event_id: None,
        };

        let res = canonicalize_dependent_requests(&[react, edit]);

        assert_eq!(res.len(), 2);
        assert_eq!(res[0].own_transaction_id, edit_id);
        assert_eq!(res[1].own_transaction_id, react_id);
    }
}