Line data Source code
1 : /*
2 : * Copyright (C) 2004-2026 Savoir-faire Linux Inc.
3 : *
4 : * This program is free software: you can redistribute it and/or modify
5 : * it under the terms of the GNU General Public License as published by
6 : * the Free Software Foundation, either version 3 of the License, or
7 : * (at your option) any later version.
8 : *
9 : * This program is distributed in the hope that it will be useful,
10 : * but WITHOUT ANY WARRANTY; without even the implied warranty of
11 : * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
12 : * GNU General Public License for more details.
13 : *
14 : * You should have received a copy of the GNU General Public License
15 : * along with this program. If not, see <https://www.gnu.org/licenses/>.
16 : */
17 :
18 : #include "message_engine.h"
19 : #include "sip/sipaccountbase.h"
20 : #include "manager.h"
21 : #include "fileutils.h"
22 :
23 : #include "client/jami_signal.h"
24 : #include "jami/account_const.h"
25 :
26 : #include <dhtnet/fileutils.h>
27 : #include <opendht/thread_pool.h>
28 : #include <fmt/std.h>
29 :
30 : #include <cstring>
31 : #include <fstream>
32 : #include <vector>
33 :
34 : namespace jami {
35 : namespace im {
36 :
37 948 : MessageEngine::MessageEngine(SIPAccountBase& acc, const std::filesystem::path& path)
38 948 : : account_(acc)
39 948 : , savePath_(path)
40 948 : , ioContext_(Manager::instance().ioContext())
41 948 : , saveTimer_(*ioContext_)
42 : {
43 948 : dhtnet::fileutils::check_dir(savePath_.parent_path());
44 948 : load();
45 948 : }
46 :
47 : MessageToken
48 13132 : MessageEngine::sendMessage(const std::string& to,
49 : const std::string& deviceId,
50 : const std::map<std::string, std::string>& payloads,
51 : uint64_t refreshToken)
52 : {
53 13132 : if (payloads.empty() or to.empty())
54 0 : return 0;
55 13126 : MessageToken token = 0;
56 : {
57 13126 : std::lock_guard lock(messagesMutex_);
58 13151 : auto& peerMessages = deviceId.empty() ? messages_[to] : messagesDevices_[deviceId];
59 13130 : if (refreshToken != 0) {
60 10841 : for (auto& m : peerMessages) {
61 641 : if (m.token == refreshToken) {
62 641 : token = refreshToken;
63 641 : m.to = to;
64 641 : m.payloads = payloads;
65 641 : m.status = MessageStatus::IDLE;
66 641 : break;
67 : }
68 : }
69 : }
70 13146 : if (token == 0) {
71 12512 : token = std::uniform_int_distribution<MessageToken> {1, JAMI_ID_MAX_VAL}(account_.rand);
72 12511 : auto& m = peerMessages.emplace_back(Message {token});
73 12458 : m.to = to;
74 12454 : m.payloads = payloads;
75 : }
76 13125 : scheduleSave();
77 13163 : }
78 13164 : asio::post(*ioContext_, [this, w = account_.weak_from_this(), to, deviceId]() {
79 13166 : if (w.lock())
80 13166 : retrySend(to, deviceId, true);
81 13166 : });
82 13122 : return token;
83 : }
84 :
85 : void
86 3682 : MessageEngine::onPeerOnline(const std::string& peer, const std::string& deviceId, bool retryOnTimeout)
87 : {
88 3682 : retrySend(peer, deviceId, retryOnTimeout);
89 3686 : }
90 :
91 : void
92 842 : MessageEngine::onRegistrationResumed()
93 : {
94 842 : std::vector<std::string> peers;
95 842 : std::vector<std::string> devices;
96 : {
97 842 : std::lock_guard lock(messagesMutex_);
98 842 : peers.reserve(messages_.size());
99 857 : for (const auto& [peer, _] : messages_)
100 15 : peers.emplace_back(peer);
101 842 : devices.reserve(messagesDevices_.size());
102 850 : for (const auto& [device, _] : messagesDevices_)
103 8 : devices.emplace_back(device);
104 842 : }
105 :
106 842 : auto w = account_.weak_from_this();
107 857 : for (auto& peer : peers) {
108 15 : asio::post(*ioContext_, [this, w, peer = std::move(peer)] {
109 15 : if (w.lock())
110 15 : retrySend(peer, {}, true);
111 15 : });
112 : }
113 850 : for (auto& device : devices) {
114 8 : asio::post(*ioContext_, [this, w, device = std::move(device)] {
115 8 : if (w.lock())
116 8 : retrySend({}, device, true);
117 8 : });
118 : }
119 842 : }
120 :
121 : void
122 16874 : MessageEngine::retrySend(const std::string& peer, const std::string& deviceId, bool retryOnTimeout)
123 : {
124 : struct PendingMsg
125 : {
126 : MessageToken token;
127 : std::string to;
128 : std::map<std::string, std::string> payloads;
129 : };
130 16874 : std::vector<PendingMsg> pending {};
131 16874 : auto now = clock::now();
132 : {
133 16877 : std::lock_guard lock(messagesMutex_);
134 16880 : auto& m = deviceId.empty() ? messages_ : messagesDevices_;
135 16879 : auto p = m.find(deviceId.empty() ? peer : deviceId);
136 16877 : if (p == m.end())
137 2918 : return;
138 13959 : auto& messages = p->second;
139 :
140 27951 : for (auto& m : messages) {
141 13993 : if (m.status == MessageStatus::IDLE) {
142 13991 : m.status = MessageStatus::SENDING;
143 13991 : m.retried++;
144 13991 : m.last_op = now;
145 13991 : pending.emplace_back(PendingMsg {m.token, m.to, m.payloads});
146 : }
147 : }
148 16876 : }
149 : // avoid locking while calling callback
150 27949 : for (const auto& p : pending) {
151 13991 : JAMI_DEBUG("[Account {:s}] [message {:d}] Reattempt sending", account_.getAccountID(), p.token);
152 41973 : if (p.payloads.find("application/im-gitmessage-id") == p.payloads.end())
153 653 : emitSignal<libjami::ConfigurationSignal::AccountMessageStatusChanged>(
154 653 : account_.getAccountID(),
155 : "",
156 653 : p.to,
157 1306 : std::to_string(p.token),
158 : (int) libjami::Account::MessageStates::SENDING);
159 13991 : account_.sendMessage(p.to, deviceId, p.payloads, p.token, retryOnTimeout, false);
160 : }
161 16877 : }
162 :
163 : MessageStatus
164 0 : MessageEngine::getStatus(MessageToken t) const
165 : {
166 0 : std::lock_guard lock(messagesMutex_);
167 0 : for (const auto& p : messages_) {
168 0 : for (const auto& m : p.second) {
169 0 : if (m.token == t)
170 0 : return m.status;
171 : }
172 : }
173 0 : return MessageStatus::UNKNOWN;
174 0 : }
175 :
176 : void
177 13991 : MessageEngine::onMessageSent(const std::string& peer, MessageToken token, bool ok, const std::string& deviceId)
178 : {
179 13991 : JAMI_DEBUG("[Account {:s}] [message {:d}] Message sent: {:s}",
180 : account_.getAccountID(),
181 : token,
182 : ok ? "success"sv : "failure"sv);
183 13991 : std::lock_guard lock(messagesMutex_);
184 13991 : auto& m = deviceId.empty() ? messages_ : messagesDevices_;
185 :
186 13991 : auto p = m.find(deviceId.empty() ? peer : deviceId);
187 13991 : if (p == m.end()) {
188 0 : JAMI_WARNING("[Account {:s}] onMessageSent: Peer not found: id:{} device:{}",
189 : account_.getAccountID(),
190 : peer,
191 : deviceId);
192 0 : return;
193 : }
194 :
195 28158 : auto f = std::find_if(p->second.begin(), p->second.end(), [&](const Message& m) { return m.token == token; });
196 13991 : if (f != p->second.end()) {
197 27981 : auto emit = f->payloads.find("application/im-gitmessage-id") == f->payloads.end();
198 13991 : if (f->status == MessageStatus::SENDING) {
199 13990 : if (ok) {
200 12179 : f->status = MessageStatus::SENT;
201 12179 : JAMI_LOG("[Account {:s}] [message {:d}] Status changed to SENT", account_.getAccountID(), token);
202 12179 : if (emit)
203 465 : emitSignal<libjami::ConfigurationSignal::AccountMessageStatusChanged>(
204 465 : account_.getAccountID(),
205 : "",
206 465 : f->to,
207 930 : std::to_string(token),
208 : static_cast<int>(libjami::Account::MessageStates::SENT));
209 12179 : p->second.erase(f);
210 12179 : scheduleSave();
211 1811 : } else if (f->retried >= MAX_RETRIES) {
212 1 : f->status = MessageStatus::FAILURE;
213 1 : JAMI_WARNING("[Account {:s}] [message {:d}] Status changed to FAILURE", account_.getAccountID(), token);
214 1 : if (emit)
215 0 : emitSignal<libjami::ConfigurationSignal::AccountMessageStatusChanged>(
216 0 : account_.getAccountID(),
217 : "",
218 0 : f->to,
219 0 : std::to_string(token),
220 : static_cast<int>(libjami::Account::MessageStates::FAILURE));
221 1 : p->second.erase(f);
222 1 : scheduleSave();
223 : } else {
224 1810 : f->status = MessageStatus::IDLE;
225 1810 : JAMI_DEBUG("[Account {:s}] [message {:d}] Status changed to IDLE", account_.getAccountID(), token);
226 : }
227 : } else {
228 1 : JAMI_DEBUG("[Account {:s}] [message {:d}] State is not SENDING", account_.getAccountID(), token);
229 : }
230 : } else {
231 0 : JAMI_DEBUG("[Account {:s}] [message {:d}] Unable to find message", account_.getAccountID(), token);
232 : }
233 13991 : }
234 :
235 : void
236 948 : MessageEngine::load()
237 : {
238 : try {
239 948 : decltype(messages_) messages;
240 948 : decltype(messagesDevices_) deviceMessages;
241 : {
242 948 : std::lock_guard lock(dhtnet::fileutils::getFileLock(savePath_));
243 1896 : const auto data = fileutils::loadFile(savePath_);
244 0 : msgpack::unpacker unpacker;
245 0 : unpacker.reserve_buffer(data.size());
246 0 : std::memcpy(unpacker.buffer(), data.data(), data.size());
247 0 : unpacker.buffer_consumed(data.size());
248 :
249 0 : msgpack::object_handle object;
250 0 : if (!unpacker.next(object))
251 0 : throw std::runtime_error("Invalid message queue");
252 0 : object.get().convert(messages);
253 0 : if (unpacker.next(object))
254 0 : object.get().convert(deviceMessages);
255 0 : }
256 0 : std::lock_guard lock(messagesMutex_);
257 0 : messages_ = std::move(messages);
258 0 : messagesDevices_ = std::move(deviceMessages);
259 0 : normalizeLoadedMessages();
260 0 : if (not messages_.empty() || not messagesDevices_.empty())
261 0 : JAMI_LOG("[Account {}] Loaded {} peer and {} device message queues from {}",
262 : account_.getAccountID(),
263 : messages_.size(),
264 : messagesDevices_.size(),
265 : savePath_);
266 2844 : } catch (const std::exception& e) {
267 948 : JAMI_LOG("[Account {}] Unable to load messages from {}: {}", account_.getAccountID(), savePath_, e.what());
268 948 : }
269 948 : }
270 :
271 : void
272 0 : MessageEngine::normalizeLoadedMessages()
273 : {
274 0 : auto normalize = [](auto& queues) {
275 0 : for (auto& [_, messages] : queues)
276 0 : for (auto& message : messages)
277 0 : if (message.status == MessageStatus::SENDING)
278 0 : message.status = MessageStatus::IDLE;
279 0 : };
280 0 : normalize(messages_);
281 0 : normalize(messagesDevices_);
282 0 : }
283 :
284 : void
285 968 : MessageEngine::save() const
286 : {
287 968 : std::lock_guard lock(messagesMutex_);
288 968 : save_();
289 968 : }
290 :
291 : void
292 25312 : MessageEngine::scheduleSave()
293 : {
294 25312 : saveTimer_.expires_after(std::chrono::seconds(5));
295 25344 : saveTimer_.async_wait([this, w = account_.weak_from_this()](const std::error_code& ec) {
296 25346 : if (!ec)
297 150 : if (auto acc = w.lock())
298 150 : save();
299 25346 : });
300 25341 : }
301 :
302 : void
303 968 : MessageEngine::save_() const
304 : {
305 : try {
306 968 : std::ofstream file;
307 968 : file.exceptions(std::ifstream::failbit | std::ifstream::badbit);
308 968 : file.open(savePath_, std::ios::trunc | std::ios::binary);
309 967 : if (file.is_open()) {
310 967 : msgpack::pack(file, messages_);
311 967 : msgpack::pack(file, messagesDevices_);
312 : }
313 969 : } catch (const std::exception& e) {
314 1 : JAMI_ERROR("[Account {}] Unable to serialize pending messages: {}", account_.getAccountID(), e.what());
315 1 : }
316 968 : }
317 :
318 : } // namespace im
319 : } // namespace jami
|
