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