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 "sync_module.h"
19 :
20 : #include "jamidht/conversation_module.h"
21 : #include "jamidht/archive_account_manager.h"
22 : #include "fileutils.h"
23 :
24 : #include <dhtnet/multiplexed_socket.h>
25 : #include <dhtnet/channel_utils.h>
26 : #include <opendht/thread_pool.h>
27 :
28 : #include <fstream>
29 :
30 : namespace jami {
31 :
32 : class SyncModule::Impl : public std::enable_shared_from_this<Impl>
33 : {
34 : public:
35 : Impl(const std::shared_ptr<JamiAccount>& account);
36 :
37 : std::weak_ptr<JamiAccount> account_;
38 : const std::string accountId_;
39 :
40 : // Sync connections
41 : std::recursive_mutex syncConnectionsMtx_;
42 : std::map<DeviceId /* deviceId */, std::vector<std::shared_ptr<dhtnet::ChannelSocket>>> syncConnections_;
43 :
44 : // Local sync-version tracking (never transmitted, see header). Used to
45 : // decide whether a sync connection must be (re)established with a device.
46 : std::filesystem::path versionPath_;
47 : mutable std::mutex versionMtx_;
48 : uint64_t localVersion_ {0};
49 : std::map<DeviceId, uint64_t> lastSynced_;
50 :
51 : void loadVersions();
52 : void saveVersions(); // versionMtx_ must be held
53 : uint64_t bumpVersion();
54 : uint64_t currentVersion() const;
55 : bool needsSync(const DeviceId& deviceId) const;
56 : void markSynced(const DeviceId& deviceId, uint64_t version);
57 :
58 : /**
59 : * Build SyncMsg and send it on socket
60 : * @param socket
61 : * @return true if the whole state was written without error
62 : */
63 : bool syncInfos(const std::shared_ptr<dhtnet::ChannelSocket>& socket, const std::shared_ptr<SyncMsg>& syncMsg);
64 : void onChannelShutdown(const std::shared_ptr<dhtnet::ChannelSocket>& socket, const DeviceId& device);
65 : };
66 :
67 : namespace {
68 : // On-disk representation of the local sync-version state.
69 : struct SyncVersionData
70 : {
71 : uint64_t version {0};
72 : std::map<DeviceId, uint64_t> synced;
73 370 : MSGPACK_DEFINE_MAP(version, synced)
74 : };
75 : } // namespace
76 :
77 681 : SyncModule::Impl::Impl(const std::shared_ptr<JamiAccount>& account)
78 681 : : account_(account)
79 1362 : , accountId_ {account->getAccountID()}
80 : {
81 681 : versionPath_ = account->getPath() / "syncVersions";
82 681 : loadVersions();
83 681 : }
84 :
85 : void
86 681 : SyncModule::Impl::loadVersions()
87 : {
88 : try {
89 1362 : auto file = fileutils::loadFile(versionPath_);
90 0 : msgpack::object_handle oh = msgpack::unpack((const char*) file.data(), file.size());
91 0 : SyncVersionData data;
92 0 : oh.get().convert(data);
93 0 : std::lock_guard lk(versionMtx_);
94 0 : localVersion_ = data.version;
95 0 : lastSynced_ = std::move(data.synced);
96 681 : } catch (const std::exception&) {
97 : // No (or unreadable) file yet: start fresh. Every known device will be
98 : // considered out-of-date and synced once on first contact.
99 681 : }
100 681 : }
101 :
102 : void
103 370 : SyncModule::Impl::saveVersions()
104 : {
105 : // versionMtx_ must be held
106 : try {
107 370 : std::ofstream file(versionPath_, std::ios::trunc | std::ios::binary);
108 370 : SyncVersionData data;
109 370 : data.version = localVersion_;
110 370 : data.synced = lastSynced_;
111 370 : msgpack::pack(file, data);
112 370 : } catch (const std::exception& e) {
113 0 : JAMI_WARNING("[Account {}] Unable to save sync versions: {:s}", accountId_, e.what());
114 0 : }
115 370 : }
116 :
117 : uint64_t
118 353 : SyncModule::Impl::bumpVersion()
119 : {
120 353 : std::lock_guard lk(versionMtx_);
121 353 : ++localVersion_;
122 353 : saveVersions();
123 353 : return localVersion_;
124 353 : }
125 :
126 : uint64_t
127 135 : SyncModule::Impl::currentVersion() const
128 : {
129 135 : std::lock_guard lk(versionMtx_);
130 136 : return localVersion_;
131 135 : }
132 :
133 : bool
134 1116 : SyncModule::Impl::needsSync(const DeviceId& deviceId) const
135 : {
136 1116 : std::lock_guard lk(versionMtx_);
137 1116 : auto it = lastSynced_.find(deviceId);
138 : // Never synced, or synced at an older version than the current one.
139 2232 : return it == lastSynced_.end() || it->second < localVersion_;
140 1116 : }
141 :
142 : void
143 135 : SyncModule::Impl::markSynced(const DeviceId& deviceId, uint64_t version)
144 : {
145 135 : std::lock_guard lk(versionMtx_);
146 135 : auto& synced = lastSynced_[deviceId];
147 134 : if (synced < version) {
148 17 : synced = version;
149 17 : saveVersions();
150 : }
151 134 : }
152 :
153 : bool
154 223 : SyncModule::Impl::syncInfos(const std::shared_ptr<dhtnet::ChannelSocket>& socket,
155 : const std::shared_ptr<SyncMsg>& syncMsg)
156 : {
157 223 : auto acc = account_.lock();
158 223 : if (!acc)
159 0 : return false;
160 223 : msgpack::sbuffer buffer(UINT16_MAX); // Use max pkt size
161 223 : std::error_code ec;
162 223 : if (!syncMsg) {
163 : // Send contacts infos
164 : // This message can be big. TODO rewrite to only take UINT16_MAX bytes max or split it multiple
165 : // messages. For now, write 3 messages (UINT16_MAX*3 should be enough for all information).
166 182 : if (auto info = acc->accountManager()->getInfo()) {
167 182 : if (info->contacts) {
168 182 : SyncMsg msg;
169 182 : msg.ds = info->contacts->getSyncData();
170 182 : msgpack::pack(buffer, msg);
171 182 : socket->write(reinterpret_cast<const unsigned char*>(buffer.data()), buffer.size(), ec);
172 182 : if (ec) {
173 4 : JAMI_ERROR("[Account {}] [device {}] {:s}", accountId_, socket->deviceId(), ec.message());
174 1 : return false;
175 : }
176 182 : }
177 : }
178 181 : buffer.clear();
179 : // Sync conversations
180 181 : auto c = ConversationModule::convInfos(acc->getAccountID());
181 181 : if (!c.empty()) {
182 86 : SyncMsg msg;
183 86 : msg.c = std::move(c);
184 86 : msgpack::pack(buffer, msg);
185 86 : socket->write(reinterpret_cast<const unsigned char*>(buffer.data()), buffer.size(), ec);
186 86 : if (ec) {
187 0 : JAMI_ERROR("[Account {}] [device {}] {:s}", accountId_, socket->deviceId(), ec.message());
188 0 : return false;
189 : }
190 86 : }
191 181 : buffer.clear();
192 : // Sync requests
193 181 : auto cr = ConversationModule::convRequests(acc->getAccountID());
194 180 : if (!cr.empty()) {
195 12 : SyncMsg msg;
196 12 : msg.cr = std::move(cr);
197 12 : msgpack::pack(buffer, msg);
198 12 : socket->write(reinterpret_cast<const unsigned char*>(buffer.data()), buffer.size(), ec);
199 12 : if (ec) {
200 0 : JAMI_ERROR("[Account {}] [device {}] {:s}", accountId_, socket->deviceId(), ec.message());
201 0 : return false;
202 : }
203 12 : }
204 180 : buffer.clear();
205 180 : auto convModule = acc->convModule(true);
206 181 : if (!convModule)
207 0 : return false;
208 : // Sync conversation's preferences
209 181 : auto p = convModule->convPreferences();
210 181 : if (!p.empty()) {
211 3 : SyncMsg msg;
212 3 : msg.p = std::move(p);
213 3 : msgpack::pack(buffer, msg);
214 3 : socket->write(reinterpret_cast<const unsigned char*>(buffer.data()), buffer.size(), ec);
215 3 : if (ec) {
216 0 : JAMI_ERROR("[Account {}] [device {}] {:s}", accountId_, socket->deviceId(), ec.message());
217 0 : return false;
218 : }
219 3 : }
220 180 : buffer.clear();
221 : // Sync read's status
222 180 : auto ms = convModule->convMessageStatus();
223 181 : if (!ms.empty()) {
224 30 : SyncMsg msg;
225 30 : msg.ms = std::move(ms);
226 30 : msgpack::pack(buffer, msg);
227 30 : socket->write(reinterpret_cast<const unsigned char*>(buffer.data()), buffer.size(), ec);
228 30 : if (ec) {
229 0 : JAMI_ERROR("[Account {}] [device {}] {:s}", accountId_, socket->deviceId(), ec.message());
230 0 : return false;
231 : }
232 30 : }
233 181 : buffer.clear();
234 :
235 181 : } else {
236 41 : msgpack::pack(buffer, *syncMsg);
237 41 : socket->write(reinterpret_cast<const unsigned char*>(buffer.data()), buffer.size(), ec);
238 41 : if (ec) {
239 0 : JAMI_ERROR("[Account {}] [device {}] {:s}", accountId_, socket->deviceId(), ec.message());
240 0 : return false;
241 : }
242 : }
243 222 : return true;
244 223 : }
245 :
246 : ////////////////////////////////////////////////////////////////
247 :
248 681 : SyncModule::SyncModule(const std::shared_ptr<JamiAccount>& account)
249 681 : : pimpl_ {std::make_shared<Impl>(account)}
250 681 : {}
251 :
252 : void
253 136 : SyncModule::Impl::onChannelShutdown(const std::shared_ptr<dhtnet::ChannelSocket>& socket, const DeviceId& device)
254 : {
255 136 : std::lock_guard lk(syncConnectionsMtx_);
256 136 : auto connectionsIt = syncConnections_.find(device);
257 136 : if (connectionsIt == syncConnections_.end()) {
258 0 : JAMI_WARNING("[Account {}] [device {}] onChannelShutdown: no connection found.", accountId_, device.to_view());
259 0 : return;
260 : }
261 136 : auto& connections = connectionsIt->second;
262 136 : auto conn = std::find(connections.begin(), connections.end(), socket);
263 136 : if (conn != connections.end())
264 136 : connections.erase(conn);
265 544 : JAMI_LOG("[Account {}] [device {}] removed connection, remaining: {:d}",
266 : accountId_,
267 : device.to_view(),
268 : connections.size());
269 136 : if (connections.empty())
270 68 : syncConnections_.erase(connectionsIt);
271 136 : }
272 :
273 : void
274 136 : SyncModule::cacheSyncConnection(std::shared_ptr<dhtnet::ChannelSocket>&& socket,
275 : const std::string& peerId,
276 : const DeviceId& device)
277 : {
278 136 : std::lock_guard lk(pimpl_->syncConnectionsMtx_);
279 135 : pimpl_->syncConnections_[device].emplace_back(socket);
280 :
281 135 : socket->setOnRecv(dhtnet::buildMsgpackReader<SyncMsg>([acc = pimpl_->account_, device, peerId](SyncMsg&& msg) {
282 352 : auto account = acc.lock();
283 352 : if (!account)
284 0 : return std::make_error_code(std::errc::operation_canceled);
285 :
286 : try {
287 351 : if (auto manager = account->accountManager())
288 352 : manager->onSyncData(std::move(msg.ds), false);
289 :
290 353 : if (!msg.c.empty() || !msg.cr.empty() || !msg.p.empty() || !msg.ld.empty() || !msg.ms.empty())
291 155 : if (auto cm = account->convModule(true))
292 155 : cm->onSyncData(msg, peerId, device.toString());
293 0 : } catch (const std::exception& e) {
294 0 : JAMI_WARNING("[Account {}] [device {}] [convInfo] error on sync: {:s}",
295 : account->getAccountID(),
296 : device.to_view(),
297 : e.what());
298 0 : }
299 353 : return std::error_code();
300 353 : }));
301 136 : socket->onShutdown([w = pimpl_->weak_from_this(), device, s = std::weak_ptr(socket)](const std::error_code&) {
302 136 : if (auto shared = w.lock())
303 136 : shared->onChannelShutdown(s.lock(), device);
304 136 : });
305 :
306 : // Capture the version we are about to deliver before sending the full
307 : // state. On success, record that this device is synced up to that version
308 : // so we don't reconnect to it until something changes again. Captured
309 : // before the send so a concurrent change is never considered delivered.
310 135 : auto version = pimpl_->currentVersion();
311 136 : dht::ThreadPool::io().run([w = pimpl_->weak_from_this(), socket = std::move(socket), device, version]() {
312 136 : if (auto s = w.lock()) {
313 136 : if (s->syncInfos(socket, nullptr))
314 135 : s->markSynced(device, version);
315 136 : }
316 136 : });
317 136 : }
318 :
319 : bool
320 68 : SyncModule::isConnected(const DeviceId& deviceId) const
321 : {
322 68 : std::lock_guard lk(pimpl_->syncConnectionsMtx_);
323 68 : auto it = pimpl_->syncConnections_.find(deviceId);
324 68 : if (it == pimpl_->syncConnections_.end())
325 68 : return false;
326 0 : return !it->second.empty();
327 68 : }
328 :
329 : void
330 2268 : SyncModule::syncWithConnected(const std::shared_ptr<SyncMsg>& syncMsg, const DeviceId& deviceId)
331 : {
332 2268 : std::lock_guard lk(pimpl_->syncConnectionsMtx_);
333 2268 : size_t count = 0;
334 2355 : for (const auto& [did, sockets] : pimpl_->syncConnections_) {
335 87 : if (not sockets.empty() and (!deviceId || deviceId == did)) {
336 87 : count++;
337 87 : dht::ThreadPool::io().run([w = pimpl_->weak_from_this(), s = sockets.back(), syncMsg] {
338 87 : if (auto sthis = w.lock())
339 87 : sthis->syncInfos(s, syncMsg);
340 87 : });
341 : }
342 : }
343 2268 : if (count == 0) {
344 8724 : JAMI_WARNING("[Account {}] [device {}] no sync connection.", pimpl_->accountId_, deviceId.toString());
345 : } else {
346 348 : JAMI_DEBUG("[Account {}] [device {}] syncing with {:d} devices", pimpl_->accountId_, deviceId.to_view(), count);
347 : }
348 2268 : }
349 :
350 : uint64_t
351 353 : SyncModule::bumpVersion()
352 : {
353 353 : return pimpl_->bumpVersion();
354 : }
355 :
356 : uint64_t
357 0 : SyncModule::currentVersion() const
358 : {
359 0 : return pimpl_->currentVersion();
360 : }
361 :
362 : bool
363 1116 : SyncModule::needsSync(const DeviceId& deviceId) const
364 : {
365 1116 : return pimpl_->needsSync(deviceId);
366 : }
367 :
368 : void
369 0 : SyncModule::markSynced(const DeviceId& deviceId, uint64_t version)
370 : {
371 0 : pimpl_->markSynced(deviceId, version);
372 0 : }
373 :
374 : } // namespace jami
|
