Stream several messages together.

Telegram/SourceFiles/data/data_session.cpp

@@ -30,6 +30,7 @@ https://github.com/telegramdesktop/tdesktop/blob/master/LEGAL
 #include "history/history.h"
 #include "history/history_item.h"
 #include "history/history_item_components.h"
+#include "history/history_streamed_drafts.h"
 #include "history/view/media/history_view_media.h"
 #include "history/view/history_view_element.h"
 #include "inline_bots/inline_bot_layout_item.h"
@@ -3107,6 +3108,20 @@ HistoryItem *Session::addNewMessage(
 		return nullptr;
 	}
 
+	if (data.type() == mtpc_message) {
+		if (const auto h = historyLoaded(peerId)) {
+			if (const auto streamed = h->streamedDraftsIfExists()) {
+				if (const auto adopted = streamed->adoptIncoming(
+						data.c_message())) {
+					if (type == NewMessageType::Unread) {
+						CheckForSwitchInlineButton(adopted);
+					}
+					return adopted;
+				}
+			}
+		}
+	}
+
 	const auto result = history(peerId)->addNewMessage(
 		id,
 		data,

Telegram/SourceFiles/history/history.cpp

@@ -1594,9 +1594,6 @@ void History::newItemAdded(not_null<HistoryItem*> item) {
 	if (const auto sublist = item->savedSublist()) {
 		sublist->applyItemAdded(item);
 	}
-	if (const auto streamed = _streamedDrafts.get()) {
-		streamed->applyItemAdded(item);
-	}
 	if (const auto media = item->media()) {
 		if (const auto gift = media->gift()) {
 			if (const auto unique = gift->unique.get()) {
@@ -3958,6 +3955,10 @@ HistoryStreamedDrafts &History::streamedDrafts() {
 	return *_streamedDrafts;
 }
 
+HistoryStreamedDrafts *History::streamedDraftsIfExists() const {
+	return _streamedDrafts.get();
+}
+
 HistoryItem *History::joinedMessageInstance() const {
 	return _joinedMessage;
 }

Telegram/SourceFiles/history/history.h

@@ -93,6 +93,7 @@ public:
 	[[nodiscard]] Data::HistoryMessages *maybeMessages();
 
 	[[nodiscard]] HistoryStreamedDrafts &streamedDrafts();
+	[[nodiscard]] HistoryStreamedDrafts *streamedDraftsIfExists() const;
 
 	[[nodiscard]] HistoryItem *joinedMessageInstance() const;
 	void checkLocalMessages();

Telegram/SourceFiles/history/history_item.cpp

@@ -2711,6 +2711,10 @@ QString HistoryItem::notificationHeader() const {
 	return QString();
 }
 
+void HistoryItem::markBeingSentForAdoption() {
+	_flags |= MessageFlag::BeingSent;
+}
+
 void HistoryItem::setRealId(MsgId newId) {
 	Expects(_flags & MessageFlag::BeingSent);
 	Expects(IsClientMsgId(id));

Telegram/SourceFiles/history/history_item.h

@@ -440,6 +440,7 @@ public:
 		bool isForumPost);
 	void setPostAuthor(const QString &author);
 	void setRealId(MsgId newId);
+	void markBeingSentForAdoption();
 	void incrementReplyToTopCounter();
 	void applyEffectWatchedOnUnreadKnown();
 

Telegram/SourceFiles/history/history_streamed_drafts.cpp

@@ -9,6 +9,7 @@ https://github.com/telegramdesktop/tdesktop/blob/master/LEGAL
 
 #include "api/api_text_entities.h"
 #include "data/data_forum_topic.h"
+#include "data/data_peer_id.h"
 #include "data/data_session.h"
 #include "history/history.h"
 #include "history/history_item.h"
@@ -17,6 +18,15 @@ namespace {
 
 constexpr auto kClearTimeout = 30 * crl::time(1000);
 
+[[nodiscard]] int CommonPrefixLength(const QString &a, const QString &b) {
+	const auto count = std::min(a.size(), b.size());
+	auto i = 0;
+	while (i < count && a[i] == b[i]) {
+		++i;
+	}
+	return i;
+}
+
 } // namespace
 
 HistoryStreamedDrafts::HistoryStreamedDrafts(not_null<History*> history)
@@ -25,7 +35,7 @@ HistoryStreamedDrafts::HistoryStreamedDrafts(not_null<History*> history)
 }
 
 HistoryStreamedDrafts::~HistoryStreamedDrafts() {
-	for (const auto &[rootId, draft] : base::take(_drafts)) {
+	for (const auto &[randomId, draft] : base::take(_drafts)) {
 		draft.message->destroy();
 	}
 }
@@ -41,19 +51,18 @@ void HistoryStreamedDrafts::apply(
 	if (!rootId) {
 		rootId = Data::ForumTopic::kGeneralId;
 	}
+	const auto randomId = data.vrandom_id().v;
 	if (!when) {
-		clear(rootId);
+		clearByRandomId(randomId);
 		return;
 	}
 	const auto text = Api::ParseTextWithEntities(
 		&_history->session(),
 		data.vtext());
-	const auto randomId = data.vrandom_id().v;
-	if (update(rootId, randomId, text)) {
+	if (update(randomId, text)) {
 		return;
 	}
-	clear(rootId);
-	_drafts.emplace(rootId, Draft{
+	_drafts.emplace(randomId, Draft{
 		.message = _history->addNewLocalMessage({
 			.id = _history->owner().nextLocalMessageId(),
 			.flags = (MessageFlag::Local
@@ -66,7 +75,8 @@ void HistoryStreamedDrafts::apply(
 			},
 			.date = when,
 		}, text, MTP_messageMediaEmpty()),
-		.randomId = randomId,
+		.rootId = rootId,
+		.fromId = fromId,
 		.updated = crl::now(),
 	});
 	if (!_checkTimer.isActive()) {
@@ -75,11 +85,10 @@ void HistoryStreamedDrafts::apply(
 }
 
 bool HistoryStreamedDrafts::update(
-		MsgId rootId,
 		uint64 randomId,
 		const TextWithEntities &text) {
-	const auto i = _drafts.find(rootId);
-	if (i == end(_drafts) || i->second.randomId != randomId) {
+	const auto i = _drafts.find(randomId);
+	if (i == end(_drafts)) {
 		return false;
 	}
 	i->second.message->setText(text);
@@ -87,8 +96,8 @@ bool HistoryStreamedDrafts::update(
 	return true;
 }
 
-void HistoryStreamedDrafts::clear(MsgId rootId) {
-	if (const auto draft = _drafts.take(rootId)) {
+void HistoryStreamedDrafts::clearByRandomId(uint64 randomId) {
+	if (const auto draft = _drafts.take(randomId)) {
 		draft->message->destroy();
 	}
 	if (_drafts.empty()) {
@@ -98,18 +107,13 @@ void HistoryStreamedDrafts::clear(MsgId rootId) {
 
 bool HistoryStreamedDrafts::hasFor(not_null<HistoryItem*> item) const {
 	const auto rootId = item->topicRootId();
-	const auto i = _drafts.find(rootId);
-	return (i != end(_drafts))
-		&& (i->second.message->from() == item->from());
-}
-
-void HistoryStreamedDrafts::applyItemAdded(not_null<HistoryItem*> item) {
-	const auto rootId = item->topicRootId();
-	const auto i = _drafts.find(rootId);
-	if (i == end(_drafts) || i->second.message->from() != item->from()) {
-		return;
+	const auto from = item->from();
+	for (const auto &[randomId, draft] : _drafts) {
+		if (draft.rootId == rootId && draft.message->from() == from) {
+			return true;
+		}
 	}
-	clear(rootId);
+	return false;
 }
 
 void HistoryStreamedDrafts::applyItemRemoved(not_null<HistoryItem*> item) {
@@ -124,6 +128,63 @@ void HistoryStreamedDrafts::applyItemRemoved(not_null<HistoryItem*> item) {
 	}
 }
 
+HistoryItem *HistoryStreamedDrafts::adoptIncoming(
+		const MTPDmessage &data) {
+	if (_drafts.empty()) {
+		return nullptr;
+	}
+	const auto fromId = data.vfrom_id()
+		? peerFromMTP(*data.vfrom_id())
+		: _history->peer->id;
+	auto rootId = MsgId(0);
+	if (const auto reply = data.vreply_to()) {
+		reply->match([&](const MTPDmessageReplyHeader &d) {
+			if (d.is_forum_topic()) {
+				rootId = d.vreply_to_top_id().value_or_empty();
+				if (!rootId) {
+					rootId = d.vreply_to_msg_id().value_or_empty();
+				}
+			}
+		}, [](const MTPDmessageReplyStoryHeader &) {});
+	}
+	if (!rootId) {
+		rootId = Data::ForumTopic::kGeneralId;
+	}
+	const auto incomingText = qs(data.vmessage());
+	auto best = end(_drafts);
+	auto bestPrefix = 0;
+	for (auto i = begin(_drafts); i != end(_drafts); ++i) {
+		const auto &draft = i->second;
+		if (draft.rootId != rootId) {
+			continue;
+		}
+		if (draft.message->from()->id != fromId) {
+			continue;
+		}
+		const auto prefix = CommonPrefixLength(
+			draft.message->originalText().text,
+			incomingText);
+		if (prefix > bestPrefix) {
+			bestPrefix = prefix;
+			best = i;
+		}
+	}
+	if (best == end(_drafts) || bestPrefix <= 0) {
+		return nullptr;
+	}
+	const auto item = best->second.message.get();
+	_drafts.erase(best);
+
+	item->markBeingSentForAdoption();
+	item->setRealId(data.vid().v);
+	item->applySentMessage(data);
+
+	if (_drafts.empty()) {
+		scheduleDestroy();
+	}
+	return item;
+}
+
 void HistoryStreamedDrafts::check() {
 	auto closest = crl::time();
 	const auto now = crl::now();

Telegram/SourceFiles/history/history_streamed_drafts.h

@@ -11,6 +11,7 @@ https://github.com/telegramdesktop/tdesktop/blob/master/LEGAL
 #include "base/weak_ptr.h"
 
 class History;
+class MTPDmessage;
 
 class HistoryStreamedDrafts final : public base::has_weak_ptr {
 public:
@@ -26,24 +27,25 @@ public:
 		const MTPDsendMessageTextDraftAction &data);
 
 	[[nodiscard]] bool hasFor(not_null<HistoryItem*> item) const;
-	void applyItemAdded(not_null<HistoryItem*> item);
 	void applyItemRemoved(not_null<HistoryItem*> item);
+	HistoryItem *adoptIncoming(const MTPDmessage &data);
 
 private:
 	struct Draft {
 		not_null<HistoryItem*> message;
-		uint64 randomId = 0;
+		MsgId rootId = 0;
+		PeerId fromId = 0;
 		crl::time updated = 0;
 	};
 
-	bool update(MsgId rootId, uint64 randomId, const TextWithEntities &text);
-	void clear(MsgId rootId);
+	bool update(uint64 randomId, const TextWithEntities &text);
+	void clearByRandomId(uint64 randomId);
 
 	void check();
 	void scheduleDestroy();
 
 	const not_null<History*> _history;
-	base::flat_map<MsgId, Draft> _drafts;
+	base::flat_map<uint64, Draft> _drafts;
 
 	base::Timer _checkTimer;