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 373 : MSGPACK_DEFINE_MAP(version, synced)
74 : };
75 : } // namespace
76 :
77 708 : SyncModule::Impl::Impl(const std::shared_ptr<JamiAccount>& account)
78 708 : : account_(account)
79 1416 : , accountId_ {account->getAccountID()}
80 : {
81 708 : versionPath_ = account->getPath() / "syncVersions";
82 708 : loadVersions();
83 708 : }
84 :
85 : void
86 708 : SyncModule::Impl::loadVersions()
87 : {
88 : try {
89 1416 : 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 708 : } 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 708 : }
100 708 : }
101 :
102 : void
103 373 : SyncModule::Impl::saveVersions()
104 : {
105 : // versionMtx_ must be held
106 : try {
107 373 : std::ofstream file(versionPath_, std::ios::trunc | std::ios::binary);
108 373 : SyncVersionData data;
109 373 : data.version = localVersion_;
110 373 : data.synced = lastSynced_;
111 373 : msgpack::pack(file, data);
112 373 : } catch (const std::exception& e) {
113 0 : JAMI_WARNING("[Account {}] Unable to save sync versions: {:s}", accountId_, e.what());
114 0 : }
115 373 : }
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 138 : SyncModule::Impl::currentVersion() const
128 : {
129 138 : std::lock_guard lk(versionMtx_);
130 138 : return localVersion_;
131 138 : }
132 :
133 : bool
134 1121 : SyncModule::Impl::needsSync(const DeviceId& deviceId) const
135 : {
136 1121 : std::lock_guard lk(versionMtx_);
137 1121 : auto it = lastSynced_.find(deviceId);
138 : // Never synced, or synced at an older version than the current one.
139 2242 : return it == lastSynced_.end() || it->second < localVersion_;
140 1121 : }
141 :
142 : void
143 138 : SyncModule::Impl::markSynced(const DeviceId& deviceId, uint64_t version)
144 : {
145 138 : std::lock_guard lk(versionMtx_);
146 138 : auto& synced = lastSynced_[deviceId];
147 138 : if (synced < version) {
148 20 : synced = version;
149 20 : saveVersions();
150 : }
151 138 : }
152 :
153 : bool
154 225 : SyncModule::Impl::syncInfos(const std::shared_ptr<dhtnet::ChannelSocket>& socket,
155 : const std::shared_ptr<SyncMsg>& syncMsg)
156 : {
157 225 : auto acc = account_.lock();
158 225 : if (!acc)
159 0 : return false;
160 225 : msgpack::sbuffer buffer(UINT16_MAX); // Use max pkt size
161 225 : std::error_code ec;
162 225 : 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 184 : if (auto info = acc->accountManager()->getInfo()) {
167 184 : if (info->contacts) {
168 184 : SyncMsg msg;
169 184 : msg.ds = info->contacts->getSyncData();
170 184 : msgpack::pack(buffer, msg);
171 184 : socket->write(reinterpret_cast<const unsigned char*>(buffer.data()), buffer.size(), ec);
172 184 : if (ec) {
173 0 : JAMI_ERROR("[Account {}] [device {}] {:s}", accountId_, socket->deviceId(), ec.message());
174 0 : return false;
175 : }
176 184 : }
177 : }
178 184 : buffer.clear();
179 : // Sync conversations. Collaborative documents are excluded: replication
180 : // is a per-device choice, each device joins one by opening it.
181 184 : auto c = ConversationModule::convInfos(acc->getAccountID());
182 304 : for (auto it = c.begin(); it != c.end();) {
183 120 : if (it->second.mode == ConversationMode::DOCUMENT)
184 8 : it = c.erase(it);
185 : else
186 112 : ++it;
187 : }
188 184 : if (!c.empty()) {
189 96 : SyncMsg msg;
190 96 : msg.c = std::move(c);
191 96 : msgpack::pack(buffer, msg);
192 96 : socket->write(reinterpret_cast<const unsigned char*>(buffer.data()), buffer.size(), ec);
193 96 : if (ec) {
194 0 : JAMI_ERROR("[Account {}] [device {}] {:s}", accountId_, socket->deviceId(), ec.message());
195 0 : return false;
196 : }
197 96 : }
198 184 : buffer.clear();
199 : // Sync requests
200 184 : auto cr = ConversationModule::convRequests(acc->getAccountID());
201 184 : if (!cr.empty()) {
202 9 : SyncMsg msg;
203 9 : msg.cr = std::move(cr);
204 9 : msgpack::pack(buffer, msg);
205 9 : socket->write(reinterpret_cast<const unsigned char*>(buffer.data()), buffer.size(), ec);
206 9 : if (ec) {
207 0 : JAMI_ERROR("[Account {}] [device {}] {:s}", accountId_, socket->deviceId(), ec.message());
208 0 : return false;
209 : }
210 9 : }
211 184 : buffer.clear();
212 184 : auto convModule = acc->convModule(true);
213 184 : if (!convModule)
214 0 : return false;
215 : // Sync conversation's preferences
216 184 : auto p = convModule->convPreferences();
217 184 : if (!p.empty()) {
218 3 : SyncMsg msg;
219 3 : msg.p = std::move(p);
220 3 : msgpack::pack(buffer, msg);
221 3 : socket->write(reinterpret_cast<const unsigned char*>(buffer.data()), buffer.size(), ec);
222 3 : if (ec) {
223 0 : JAMI_ERROR("[Account {}] [device {}] {:s}", accountId_, socket->deviceId(), ec.message());
224 0 : return false;
225 : }
226 3 : }
227 184 : buffer.clear();
228 : // Sync read's status
229 183 : auto ms = convModule->convMessageStatus();
230 184 : if (!ms.empty()) {
231 38 : SyncMsg msg;
232 38 : msg.ms = std::move(ms);
233 38 : msgpack::pack(buffer, msg);
234 38 : socket->write(reinterpret_cast<const unsigned char*>(buffer.data()), buffer.size(), ec);
235 38 : if (ec) {
236 0 : JAMI_ERROR("[Account {}] [device {}] {:s}", accountId_, socket->deviceId(), ec.message());
237 0 : return false;
238 : }
239 38 : }
240 184 : buffer.clear();
241 :
242 184 : } else {
243 41 : msgpack::pack(buffer, *syncMsg);
244 41 : socket->write(reinterpret_cast<const unsigned char*>(buffer.data()), buffer.size(), ec);
245 41 : if (ec) {
246 0 : JAMI_ERROR("[Account {}] [device {}] {:s}", accountId_, socket->deviceId(), ec.message());
247 0 : return false;
248 : }
249 : }
250 224 : return true;
251 224 : }
252 :
253 : ////////////////////////////////////////////////////////////////
254 :
255 708 : SyncModule::SyncModule(const std::shared_ptr<JamiAccount>& account)
256 708 : : pimpl_ {std::make_shared<Impl>(account)}
257 708 : {}
258 :
259 : void
260 138 : SyncModule::Impl::onChannelShutdown(const std::shared_ptr<dhtnet::ChannelSocket>& socket, const DeviceId& device)
261 : {
262 138 : std::lock_guard lk(syncConnectionsMtx_);
263 138 : auto connectionsIt = syncConnections_.find(device);
264 138 : if (connectionsIt == syncConnections_.end()) {
265 0 : JAMI_WARNING("[Account {}] [device {}] onChannelShutdown: no connection found.", accountId_, device.to_view());
266 0 : return;
267 : }
268 138 : auto& connections = connectionsIt->second;
269 138 : auto conn = std::find(connections.begin(), connections.end(), socket);
270 138 : if (conn != connections.end())
271 138 : connections.erase(conn);
272 138 : JAMI_LOG("[Account {}] [device {}] removed connection, remaining: {:d}",
273 : accountId_,
274 : device.to_view(),
275 : connections.size());
276 138 : if (connections.empty())
277 70 : syncConnections_.erase(connectionsIt);
278 138 : }
279 :
280 : void
281 138 : SyncModule::cacheSyncConnection(std::shared_ptr<dhtnet::ChannelSocket>&& socket,
282 : const std::string& peerId,
283 : const DeviceId& device)
284 : {
285 138 : std::lock_guard lk(pimpl_->syncConnectionsMtx_);
286 138 : pimpl_->syncConnections_[device].emplace_back(socket);
287 :
288 138 : socket->setOnRecv(dhtnet::buildMsgpackReader<SyncMsg>([acc = pimpl_->account_, device, peerId](SyncMsg&& msg) {
289 371 : auto account = acc.lock();
290 370 : if (!account)
291 0 : return std::make_error_code(std::errc::operation_canceled);
292 :
293 : try {
294 370 : if (auto manager = account->accountManager())
295 371 : manager->onSyncData(std::move(msg.ds), false);
296 :
297 371 : if (!msg.c.empty() || !msg.cr.empty() || !msg.p.empty() || !msg.ld.empty() || !msg.ms.empty())
298 172 : if (auto cm = account->convModule(true))
299 172 : cm->onSyncData(msg, peerId, device.toString());
300 0 : } catch (const std::exception& e) {
301 0 : JAMI_WARNING("[Account {}] [device {}] [convInfo] error on sync: {:s}",
302 : account->getAccountID(),
303 : device.to_view(),
304 : e.what());
305 0 : }
306 371 : return std::error_code();
307 371 : }));
308 138 : socket->onShutdown([w = pimpl_->weak_from_this(), device, s = std::weak_ptr(socket)](const std::error_code&) {
309 138 : if (auto shared = w.lock())
310 138 : shared->onChannelShutdown(s.lock(), device);
311 138 : });
312 :
313 : // Capture the version we are about to deliver before sending the full
314 : // state. On success, record that this device is synced up to that version
315 : // so we don't reconnect to it until something changes again. Captured
316 : // before the send so a concurrent change is never considered delivered.
317 138 : auto version = pimpl_->currentVersion();
318 138 : dht::ThreadPool::io().run([w = pimpl_->weak_from_this(), socket = std::move(socket), device, version]() {
319 138 : if (auto s = w.lock()) {
320 138 : if (s->syncInfos(socket, nullptr))
321 138 : s->markSynced(device, version);
322 138 : }
323 138 : });
324 138 : }
325 :
326 : bool
327 70 : SyncModule::isConnected(const DeviceId& deviceId) const
328 : {
329 70 : std::lock_guard lk(pimpl_->syncConnectionsMtx_);
330 70 : auto it = pimpl_->syncConnections_.find(deviceId);
331 70 : if (it == pimpl_->syncConnections_.end())
332 69 : return false;
333 1 : return !it->second.empty();
334 70 : }
335 :
336 : void
337 2831 : SyncModule::syncWithConnected(const std::shared_ptr<SyncMsg>& syncMsg, const DeviceId& deviceId)
338 : {
339 2831 : std::lock_guard lk(pimpl_->syncConnectionsMtx_);
340 2831 : size_t count = 0;
341 2918 : for (const auto& [did, sockets] : pimpl_->syncConnections_) {
342 87 : if (not sockets.empty() and (!deviceId || deviceId == did)) {
343 87 : count++;
344 87 : dht::ThreadPool::io().run([w = pimpl_->weak_from_this(), s = sockets.back(), syncMsg] {
345 87 : if (auto sthis = w.lock())
346 87 : sthis->syncInfos(s, syncMsg);
347 87 : });
348 : }
349 : }
350 2831 : if (count == 0) {
351 2744 : JAMI_WARNING("[Account {}] [device {}] no sync connection.", pimpl_->accountId_, deviceId.toString());
352 : } else {
353 87 : JAMI_DEBUG("[Account {}] [device {}] syncing with {:d} devices", pimpl_->accountId_, deviceId.to_view(), count);
354 : }
355 2831 : }
356 :
357 : uint64_t
358 353 : SyncModule::bumpVersion()
359 : {
360 353 : return pimpl_->bumpVersion();
361 : }
362 :
363 : uint64_t
364 0 : SyncModule::currentVersion() const
365 : {
366 0 : return pimpl_->currentVersion();
367 : }
368 :
369 : bool
370 1121 : SyncModule::needsSync(const DeviceId& deviceId) const
371 : {
372 1121 : return pimpl_->needsSync(deviceId);
373 : }
374 :
375 : void
376 0 : SyncModule::markSynced(const DeviceId& deviceId, uint64_t version)
377 : {
378 0 : pimpl_->markSynced(deviceId, version);
379 0 : }
380 :
381 : } // namespace jami
|
