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