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 1043 : MessageEngine::MessageEngine(SIPAccountBase& acc, const std::filesystem::path& path)
38 1043 : : account_(acc)
39 1043 : , savePath_(path)
40 1043 : , ioContext_(Manager::instance().ioContext())
41 1043 : , saveTimer_(*ioContext_)
42 : {
43 1043 : dhtnet::fileutils::check_dir(savePath_.parent_path());
44 1043 : load();
45 1043 : }
46 :
47 : MessageToken
48 13981 : 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 13981 : if (payloads.empty() or to.empty())
54 0 : return 0;
55 13981 : MessageToken token = 0;
56 : {
57 13981 : std::lock_guard lock(messagesMutex_);
58 13981 : auto& peerMessages = deviceId.empty() ? messages_[to] : messagesDevices_[deviceId];
59 13981 : if (refreshToken != 0) {
60 11518 : for (auto& m : peerMessages) {
61 990 : if (m.token == refreshToken) {
62 990 : token = refreshToken;
63 990 : m.to = to;
64 990 : m.payloads = payloads;
65 990 : m.status = MessageStatus::IDLE;
66 990 : break;
67 : }
68 : }
69 : }
70 13981 : if (token == 0) {
71 12991 : token = std::uniform_int_distribution<MessageToken> {1, JAMI_ID_MAX_VAL}(account_.rand);
72 12991 : auto& m = peerMessages.emplace_back(Message {token});
73 12991 : m.to = to;
74 12991 : m.payloads = payloads;
75 : }
76 13981 : scheduleSave();
77 13981 : }
78 13981 : asio::post(*ioContext_, [this, w = account_.weak_from_this(), to, deviceId]() {
79 13981 : if (w.lock())
80 13981 : retrySend(to, deviceId, true);
81 13981 : });
82 13981 : return token;
83 : }
84 :
85 : void
86 4552 : MessageEngine::onPeerOnline(const std::string& peer, const std::string& deviceId, bool retryOnTimeout)
87 : {
88 4552 : retrySend(peer, deviceId, retryOnTimeout);
89 4552 : }
90 :
91 : void
92 938 : MessageEngine::onRegistrationResumed()
93 : {
94 938 : std::vector<std::string> peers;
95 938 : std::vector<std::string> devices;
96 : {
97 938 : std::lock_guard lock(messagesMutex_);
98 938 : peers.reserve(messages_.size());
99 959 : for (const auto& [peer, _] : messages_)
100 21 : peers.emplace_back(peer);
101 938 : devices.reserve(messagesDevices_.size());
102 950 : for (const auto& [device, _] : messagesDevices_)
103 12 : devices.emplace_back(device);
104 938 : }
105 :
106 938 : auto w = account_.weak_from_this();
107 959 : for (auto& peer : peers) {
108 21 : asio::post(*ioContext_, [this, w, peer = std::move(peer)] {
109 21 : if (w.lock())
110 21 : retrySend(peer, {}, true);
111 21 : });
112 : }
113 950 : for (auto& device : devices) {
114 12 : asio::post(*ioContext_, [this, w, device = std::move(device)] {
115 12 : if (w.lock())
116 12 : retrySend({}, device, true);
117 12 : });
118 : }
119 938 : }
120 :
121 : void
122 18566 : 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 18566 : std::vector<PendingMsg> pending {};
131 18566 : auto now = clock::now();
132 : {
133 18566 : std::lock_guard lock(messagesMutex_);
134 18566 : auto& m = deviceId.empty() ? messages_ : messagesDevices_;
135 18566 : auto p = m.find(deviceId.empty() ? peer : deviceId);
136 18566 : if (p == m.end())
137 3557 : return;
138 15009 : auto& messages = p->second;
139 :
140 29910 : for (auto& m : messages) {
141 14901 : if (m.status == MessageStatus::IDLE) {
142 14889 : m.status = MessageStatus::SENDING;
143 14889 : m.retried++;
144 14889 : m.last_op = now;
145 14889 : pending.emplace_back(PendingMsg {m.token, m.to, m.payloads});
146 : }
147 : }
148 18566 : }
149 : // avoid locking while calling callback
150 29898 : for (const auto& p : pending) {
151 14889 : JAMI_DEBUG("[Account {:s}] [message {:d}] Reattempt sending", account_.getAccountID(), p.token);
152 44667 : if (p.payloads.find("application/im-gitmessage-id") == p.payloads.end())
153 587 : emitSignal<libjami::ConfigurationSignal::AccountMessageStatusChanged>(
154 587 : account_.getAccountID(),
155 : "",
156 587 : p.to,
157 1174 : std::to_string(p.token),
158 : (int) libjami::Account::MessageStates::SENDING);
159 14889 : account_.sendMessage(p.to, deviceId, p.payloads, p.token, retryOnTimeout, false);
160 : }
161 18566 : }
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 14889 : MessageEngine::onMessageSent(const std::string& peer, MessageToken token, bool ok, const std::string& deviceId)
178 : {
179 14889 : JAMI_DEBUG("[Account {:s}] [message {:d}] Message sent: {:s}",
180 : account_.getAccountID(),
181 : token,
182 : ok ? "success"sv : "failure"sv);
183 14889 : std::lock_guard lock(messagesMutex_);
184 14889 : auto& m = deviceId.empty() ? messages_ : messagesDevices_;
185 :
186 14889 : auto p = m.find(deviceId.empty() ? peer : deviceId);
187 14889 : 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 29881 : auto f = std::find_if(p->second.begin(), p->second.end(), [&](const Message& m) { return m.token == token; });
196 14889 : if (f != p->second.end()) {
197 29778 : auto emit = f->payloads.find("application/im-gitmessage-id") == f->payloads.end();
198 14889 : if (f->status == MessageStatus::SENDING) {
199 14886 : if (ok) {
200 12556 : f->status = MessageStatus::SENT;
201 12556 : JAMI_LOG("[Account {:s}] [message {:d}] Status changed to SENT", account_.getAccountID(), token);
202 12556 : if (emit)
203 482 : emitSignal<libjami::ConfigurationSignal::AccountMessageStatusChanged>(
204 482 : account_.getAccountID(),
205 : "",
206 482 : f->to,
207 964 : std::to_string(token),
208 : static_cast<int>(libjami::Account::MessageStates::SENT));
209 12556 : p->second.erase(f);
210 12556 : scheduleSave();
211 2330 : } else if (f->retried >= MAX_RETRIES) {
212 3 : f->status = MessageStatus::FAILURE;
213 3 : JAMI_WARNING("[Account {:s}] [message {:d}] Status changed to FAILURE", account_.getAccountID(), token);
214 3 : 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 3 : p->second.erase(f);
222 3 : scheduleSave();
223 : } else {
224 2327 : f->status = MessageStatus::IDLE;
225 2327 : JAMI_DEBUG("[Account {:s}] [message {:d}] Status changed to IDLE", account_.getAccountID(), token);
226 : }
227 : } else {
228 3 : 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 14889 : }
234 :
235 : void
236 1043 : MessageEngine::load()
237 : {
238 : try {
239 1043 : decltype(messages_) messages;
240 1043 : decltype(messagesDevices_) deviceMessages;
241 : {
242 1043 : std::lock_guard lock(dhtnet::fileutils::getFileLock(savePath_));
243 2086 : 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 3129 : } catch (const std::exception& e) {
267 1043 : JAMI_LOG("[Account {}] Unable to load messages from {}: {}", account_.getAccountID(), savePath_, e.what());
268 1043 : }
269 1043 : }
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 1117 : MessageEngine::save() const
286 : {
287 1117 : std::lock_guard lock(messagesMutex_);
288 1117 : save_();
289 1117 : }
290 :
291 : void
292 26540 : MessageEngine::scheduleSave()
293 : {
294 26540 : saveTimer_.expires_after(std::chrono::seconds(5));
295 26540 : saveTimer_.async_wait([this, w = account_.weak_from_this()](const std::error_code& ec) {
296 26540 : if (!ec)
297 207 : if (auto acc = w.lock())
298 207 : save();
299 26540 : });
300 26540 : }
301 :
302 : void
303 1117 : MessageEngine::save_() const
304 : {
305 : try {
306 1117 : std::ofstream file;
307 1117 : file.exceptions(std::ifstream::failbit | std::ifstream::badbit);
308 1117 : file.open(savePath_, std::ios::trunc | std::ios::binary);
309 1114 : if (file.is_open()) {
310 1114 : msgpack::pack(file, messages_);
311 1114 : msgpack::pack(file, messagesDevices_);
312 : }
313 1120 : } catch (const std::exception& e) {
314 3 : JAMI_ERROR("[Account {}] Unable to serialize pending messages: {}", account_.getAccountID(), e.what());
315 3 : }
316 1117 : }
317 :
318 : } // namespace im
319 : } // namespace jami
|
