i2pd/libi2pd_client/TorrentsTunnel.cpp

/*
* Copyright (c) 2026, The PurpleI2P Project
*
* This file is part of Purple i2pd project and licensed under BSD3
*
* See full license text in LICENSE file at top of project tree
*/

#include <fstream>
#include <boost/algorithm/string.hpp>
#include <boost/asio.hpp>
#include <boost/beast.hpp>
#include "Log.h"
#include "Timestamp.h"
#include "I2PEndian.h"
#include "HTTP.h"
#include "AddressBook.h"
#include "ClientContext.h"
#include "TorrentsTunnel.h"

namespace i2p
{
namespace torrents
{
	TorrentsTunnel::TorrentsTunnel (std::string_view name, std::shared_ptr<i2p::client::ClientDestination> localDestination,
		std::string_view torrentsDir, std::string_view trackers):
		i2p::client::I2PService (localDestination), m_Name (name), m_PeerID ("-I2PD-"),
		m_TorrentsDir (torrentsDir), m_TrackerRequestsCheckTimer (GetService ()),
		m_KeepAliveCheckTimer (GetService ()), m_ReconnectCheckTimer (GetService ()),
		m_TorrentsStatusUpdateTimer (GetService ())
	{
		if (localDestination)
			m_PeerID += localDestination->GetIdentHash ().ToBase64 ();
		m_PeerID.resize (20, '0');
		if (!trackers.empty ())
			boost::split(m_Trackers, trackers, boost::is_any_of(","), boost::token_compress_on);
	}

	void TorrentsTunnel::Start ()
	{
		i2p::client::I2PService::Start ();
		m_DiskIOService.Start ();

		auto dgramDest = GetLocalDestination ()->CreateDatagramDestination (false, i2p::datagram::eDatagramV3);
		if (dgramDest)
			dgramDest->SetRawReceiver (std::bind (&TorrentsTunnel::HandleRecvFromI2PRaw,
				std::static_pointer_cast<TorrentsTunnel>(shared_from_this ()),
				std::placeholders::_1, std::placeholders::_2, std::placeholders::_3, std::placeholders::_4));

		Accept ();

		if (!m_TorrentsDir.empty() && std::filesystem::exists (m_TorrentsDir) &&
			std::filesystem::is_directory (m_TorrentsDir))
		{
			for (const auto& it: std::filesystem::directory_iterator (m_TorrentsDir))
				if (std::filesystem::is_regular_file (it.status()) && it.path ().extension () == ".torrent")
					ReadTorrentFile (it.path ());
        }

		ScheduleTrackerRequestsCheck ();
		ScheduleKeepAliveCheck ();
		ScheduleStatusUpdate ();
	}

	void TorrentsTunnel::Stop ()
	{
		auto localDestination = GetLocalDestination ();
		if (localDestination)
		{
			localDestination->StopAcceptingStreams ();
			auto dgramDest = localDestination->GetDatagramDestination ();
			if (dgramDest)
				dgramDest->ResetRawReceiver ();
		}
		m_TrackerRequestsCheckTimer.cancel ();
		m_KeepAliveCheckTimer.cancel ();
		m_ReconnectCheckTimer.cancel ();
		m_TorrentsStatusUpdateTimer.cancel ();
		m_Torrents.clear ();
		for (auto it: m_Torrents)
		{
			auto fullPath = it.second->GetFullPath (); fullPath += ".resume";
			boost::asio::post (m_DiskIOService.GetService (),  [torrent = it.second, fullPath]()
				{
					torrent->SaveTorrentResumeFile (fullPath);
				});
		}
		m_DiskIOService.Stop ();
		i2p::client::I2PService::Stop ();
	}

	void TorrentsTunnel::ReadTorrentFile (const std::filesystem::path& torrentFilePath)
	{
		std::shared_ptr<Torrent> torrent;
 		std::ifstream s(torrentFilePath, std::ifstream::binary);
		if (s)
		{
			s.seekg (0,std::ios::end);
			size_t len = s.tellg ();
			if (len > 0)
			{
				s.seekg(0, std::ios::beg);
				char * buf = new char[len];
				s.read(buf, len);
				torrent = std::make_shared<Torrent>(std::string_view{buf, len});
				delete[] buf;
			}
			else
				LogPrint (eLogError, "TorrentsTunnel: Empty file ", torrentFilePath);
		}
		else
			LogPrint (eLogError, "TorrentsTunnel: Can't open file ", torrentFilePath);

		if (torrent && !torrent->IsValid ())
		{
			// a torrent whose parsing stopped, an unsafe name among the rest, must
			// not be used even in part: it would leave stray files behind
			LogPrint (eLogError, "TorrentsTunnel: Invalid torrent file ", torrentFilePath, ". Skipped");
			torrent = nullptr;
		}
		if (torrent)
		{
			torrent->SetFullPath (m_TorrentsDir/std::filesystem::path (torrent->GetName ()));
			InitTorrentFiles (torrent);
			InsertTorrent (torrent);
		}
	}

	void TorrentsTunnel::InitTorrentFiles (std::shared_ptr<Torrent> torrent)
	{
		if (!torrent) return;
		if (torrent->GetFiles ().empty ())
		{
			if (std::filesystem::exists (torrent->GetFullPath ()))
				torrent->SetComplete ();
			else
			{
				auto partFilePath = torrent->GetFullPath (); partFilePath += ".part";
				if (!std::filesystem::exists (partFilePath))
					CreateAndReserveFile (partFilePath, torrent->GetLength ());
			}
		}
		else
		{
			bool completed = true;
			for (auto& [filePath, fileLength]: torrent->GetFiles ())
			{
				filePath = torrent->GetFullPath ()/filePath;
				if (!std::filesystem::exists (filePath))
				{
					auto partFilePath = filePath; partFilePath += ".part";
					if (!std::filesystem::exists (partFilePath))
						CreateAndReserveFile (partFilePath, fileLength);
					completed = false;
				}
			}
			if (completed) torrent->SetComplete ();
		}
		auto resumeFilePath = torrent->GetFullPath (); resumeFilePath += ".resume";
		if (std::filesystem::exists (resumeFilePath))
		{
			if (!torrent->IsComplete ())
			{
				std::ifstream rs(resumeFilePath, std::ifstream::binary);
				if (rs)
				{
					rs.seekg (0,std::ios::end);
					size_t l = rs.tellg ();
					if (l > 0)
					{
						rs.seekg(0, std::ios::beg);
						std::vector<uint8_t> bitfield(l);
						rs.read((char *)bitfield.data (), l);
						if (torrent->ApplyBitfield (bitfield))
							CompleteTorrent (torrent);
					}
				}
			}
			else
				std::filesystem::remove (resumeFilePath);
		}
	}

	bool TorrentsTunnel::CreateAndReserveFile (const std::filesystem::path& filePath, size_t reserve)
	{
		if (std::filesystem::exists (filePath)) return false;
		auto subdirs = filePath.parent_path ();
		if (!subdirs.empty ())
		{
			// try to create all subdirs
			try
			{
				std::filesystem::create_directories (subdirs);
			}
			catch (std::exception& ex)
			{
				LogPrint (eLogError, "TorrentsTunnel: Can't create subdirs ", subdirs, " : ", ex.what());
				return false;
			}
		}
		// create file
		std::ofstream f(filePath, std::ios::binary);
		if (!f) return false;
		f.close ();
		if (reserve > 0)
		{
			// resize
			try
			{
				std::filesystem::resize_file (filePath, reserve);
			}
			catch (std::exception& ex)
			{
				LogPrint (eLogError, "TorrentsTunnel: Can't resize file ", filePath, " to ", reserve, " : ", ex.what());
				return false;
			}
		}
		return true;
	}

	void TorrentsTunnel::CompleteTorrent (std::shared_ptr<Torrent> torrent)
	{
		boost::asio::post (GetDiskIOService (), [this, torrent]()
		{
			bool completed = false;
			if (torrent->GetFiles ().empty ())
			{
				auto partFilePath = torrent->GetFullPath ();  partFilePath += ".part";
				std::error_code ec;
				std::filesystem::rename (partFilePath, torrent->GetFullPath (), ec);
				if (!ec)
					completed = true;
				else
					LogPrint (eLogError, "TorrentsTunnel: Can't rename ", partFilePath);
			}
			else
			{
				completed = true;
				for (const auto& [filePath, fileSize]: torrent->GetFiles ())
				{
					auto partFilePath = filePath; partFilePath += ".part";
					std::error_code ec;
					std::filesystem::rename (partFilePath, filePath, ec);
					if (ec)
					{
						completed = false;
						LogPrint (eLogError, "TorrentsTunnel: Can't rename ", partFilePath);
					}
				}
			}
			if (completed)
			{
				torrent->SetComplete ();
				auto resumeFilePath = torrent->GetFullPath (); resumeFilePath += ".resume";
				if (!std::filesystem::remove (resumeFilePath))
					LogPrint (eLogError, "TorrentsTunnel: Can't delete resume file ", resumeFilePath);
				LogPrint (eLogInfo, "TorrentsTunnel: Download complete ", torrent->GetFullPath ());

				boost::asio::post (GetService (), [this, torrent]()
					{
						// inform tracker that we are done
						RequestTorrentTrackers (torrent, "completed");
						// close connections with seeds and reset stats for remaining
						auto conns = GetTorrentConnections (torrent);
						for (auto it: conns)
						{
							if (it->GetRemoteBitfield ().all ()) // seed
								it->Close ();
							else
								it->ResetStats ();
						}
					});
			}
		});
	}

	std::shared_ptr<Torrent> TorrentsTunnel::FindTorrent (const Torrent::InfoHash& infoHash) const
	{
		std::lock_guard<std::mutex> l(m_TorrentsMutex);
		auto it = m_Torrents.find (infoHash);
		if (it != m_Torrents.end ())
			return it->second;
		return nullptr;
	}

	std::shared_ptr<Torrent> TorrentsTunnel::FindTorrentByID (int id) const
	{
		std::lock_guard<std::mutex> l(m_TorrentsMutex);
		auto it = m_TorrentsByID.find (id);
		if (it != m_TorrentsByID.end ())
			return it->second.lock ();
		return nullptr;
	}

	std::vector<int> TorrentsTunnel::GetTorrentIDs () const
	{
		std::vector<int> ids;
		std::lock_guard<std::mutex> l(m_TorrentsMutex);
		for (const auto& it: m_TorrentsByID)
			if (!it.second.expired ()) ids.push_back (it.first);
		return ids;
	}

	std::pair<std::shared_ptr<Torrent>, int> TorrentsTunnel::AddTorrent (std::string_view torrentFileContent)
	{
		auto torrent = std::make_shared<Torrent> (torrentFileContent);
		if (m_Torrents.find (torrent->GetInfoHash ()) == m_Torrents.end ())
		{
			torrent->SetFullPath (m_TorrentsDir/std::filesystem::path (torrent->GetName ()));
			{
				auto torrentFilePath = torrent->GetFullPath ();  torrentFilePath += ".torrent";
				std::ofstream f(torrentFilePath, std::ofstream::binary);
				if (f)
					f.write (torrentFileContent.data (), torrentFileContent.size ());
				else
					return { torrent, 0 };
			}
			InitTorrentFiles (torrent);
			return { torrent, InsertTorrent (torrent) };
		}
		return { torrent, 0 };
	}

	int TorrentsTunnel::InsertTorrent (std::shared_ptr<Torrent> torrent)
	{
		if (!torrent) return 0;
		std::lock_guard<std::mutex> l(m_TorrentsMutex);
		if (m_Torrents.emplace (torrent->GetInfoHash (), torrent).second)
		{
			int id = 1;
			if (!m_TorrentsByID.empty ())
				id = m_TorrentsByID.rbegin ()->first + 1;
			m_TorrentsByID.emplace (id, torrent);
			return id;
 		}
		return 0;
	}

	bool TorrentsTunnel::RemoveTorrent (int id, bool deleteFiles)
	{
		std::shared_ptr<Torrent> torrent;
		{
			std::lock_guard<std::mutex> l(m_TorrentsMutex);
			auto it = m_TorrentsByID.find (id);
			if (it == m_TorrentsByID.end ()) return false;
			torrent = it->second.lock ();
			m_TorrentsByID.erase (it);
			if (!torrent) return false;
			m_Torrents.erase (torrent->GetInfoHash ());
		}
		boost::asio::post (GetService (), [this, torrent, deleteFiles]()
			{
				RemoveTorrent (torrent, deleteFiles);
			});
		return true;
	}

	void TorrentsTunnel::RemoveTorrent (std::shared_ptr<Torrent> torrent, bool deleteFiles)
	{
		if (!torrent) return;
		auto connections = GetTorrentConnections (torrent);
		// close connections
		for (auto it: connections)
			it->Close ();
		if (deleteFiles)
			boost::asio::post (GetDiskIOService (), [torrent]()
				{
					auto fullPath = torrent->GetFullPath ();
					auto torrentFilePath = fullPath; torrentFilePath += ".torrent";
					std::error_code ec;
					std::filesystem::remove (torrentFilePath, ec);
					if (ec)
						LogPrint (eLogError, "TorrentsTunnel: Can't delete ", torrentFilePath);
					auto resumeFilePath = fullPath; resumeFilePath += ".resume";
					if (std::filesystem::exists (resumeFilePath))
					{
						std::filesystem::remove (resumeFilePath, ec);
						if (ec)
							LogPrint (eLogError, "TorrentsTunnel: Can't delete ", resumeFilePath);
					}
					if (torrent->IsComplete () || !torrent->GetFiles ().empty ())
					{
						std::filesystem::remove_all (fullPath, ec);
						if (ec)
							LogPrint (eLogError, "TorrentsTunnel: Can't delete ", fullPath);
					}
					else
					{
						auto partFilePath = fullPath; partFilePath += ".part";
						std::filesystem::remove (partFilePath, ec);
						if (ec)
							LogPrint (eLogError, "TorrentsTunnel: Can't delete ", partFilePath);
					}
				});
	}

	void TorrentsTunnel::Accept ()
	{
		auto localDestination = GetLocalDestination ();
		if (localDestination)
		{
			if (!localDestination->IsAcceptingStreams ()) // set it as default if not set yet
				localDestination->AcceptStreams ([this](std::shared_ptr<i2p::stream::Stream> stream)
					{
						if (stream)
						{
							auto conn = std::make_shared<PeerConnection> (shared_from_this (), stream);
							AddHandler (conn);
							conn->ReceiveHandshake ();
						}
					});
		}
		else
			LogPrint (eLogError, "TorrentsTunnel: Local destination not set");
	}

	void TorrentsTunnel::RequestTorrentTrackers (std::shared_ptr<Torrent> torrent, std::string_view event)
	{
		if (!m_Trackers.empty ())
			for (size_t i = 0; i < m_Trackers.size (); i++)
				RequestTracker (i, torrent, event);
		else
			RequestTracker (0, torrent, event); // from announce
	}

	void TorrentsTunnel::RequestTracker (size_t trackerID, std::shared_ptr<Torrent> torrent, std::string_view event)
	{
		if (!torrent) return;
		i2p::http::URL reqURL;
		if (trackerID < m_Trackers.size())
			reqURL.parse (m_Trackers[trackerID]);
		else
			reqURL.parse (torrent->GetAnnounce ());
#if __cplusplus >= 202002L // C++20
		if (!reqURL.host.ends_with (".i2p"))
#else
		if (reqURL.host.find(".i2p") == reqURL.host.npos)
#endif
		{
			LogPrint (eLogWarning, "TorrentsTunnel: Non-I2P address ", reqURL.host, " for torrent ", torrent->GetName ());
			return;
		}
		if (reqURL.schema == "udp")
		{
			ConnectToDatagramTracker (torrent, trackerID, reqURL.host, reqURL.port,
				(event == "completed") ? eTrackerAnnounceEventCompleted : eTrackerAnnounceEventNone);
			return;
		}
		std::map<std::string, std::string> params;
		params.emplace ("info_hash", torrent->GetHexStringInfoHash ());
		params.emplace ("peer_id", m_PeerID);
		params.emplace ("ip", GetLocalDestination ()->GetIdentity ()->ToBase64 () + ".i2p");
		params.emplace ("port", std::to_string (TORRENT_PORT)); // 6881
		params.emplace ("compact", "1");
		params.emplace ("uploaded", std::to_string (torrent->GetUploaded ()));
		params.emplace ("downloaded", std::to_string (torrent->GetLength () - torrent->GetLeft ()));
		params.emplace ("left", std::to_string (torrent->GetLeft ()));
		params.emplace ("numwant", torrent->IsComplete () ? "0" : "25"); // max num of peers, 0 if seeding
		if (!event.empty ())
			params.emplace ("event", event);
		reqURL.create_query (params);

		auto req = std::make_shared<boost::beast::http::request<boost::beast::http::string_body> >(boost::beast::http::verb::get, reqURL.to_string (true), 11); // HTTP 1.1
		req->set (boost::beast::http::field::host, reqURL.host);
		req->set (boost::beast::http::field::user_agent, "I2PSocketEepGet");
		req->keep_alive (false); // Connection: close
		CreateStream ([this, req, torrent, trackerID](std::shared_ptr<i2p::stream::Stream> stream)
			{
				if (stream)
				{
					auto httpStream = std::make_shared<i2p::client::BoostAsyncStream>(stream);
					boost::beast::http::async_write (*httpStream, *req,
						std::bind (&TorrentsTunnel::TrackerRequestSent, this, std::placeholders::_1,
							std::placeholders::_2, httpStream, torrent, req, trackerID));
				}
			}, reqURL.host, reqURL.port);
	}

	void TorrentsTunnel::TrackerRequestSent (const boost::beast::error_code& ecode, size_t bytes_transferred,
		std::shared_ptr<i2p::client::BoostAsyncStream> httpStream, std::shared_ptr<Torrent> torrent,
		std::shared_ptr<boost::beast::http::request<boost::beast::http::string_body> > req, size_t trackerID)
	{
		if (!ecode)
		{
			// receive
			auto buf = std::make_shared<boost::beast::flat_buffer> ();
			auto res = std::make_shared<boost::beast::http::response<boost::beast::http::string_body> >();
			boost::beast::http::async_read (*httpStream, *buf, *res,
				[this, httpStream, torrent, buf, res, trackerID](const boost::beast::error_code& ecode, size_t bytes_transferred)
				{
					httpStream->GetStream ()->AsyncClose ();
					if (!ecode)
					{
						if (res->result () == boost::beast::http::status::ok)
						{
							torrent->ParseTrackerResponse (trackerID, res->body ());
							ConnectToPeers (torrent);
							ScheduleReconnectCheck ();
						}
						else
							LogPrint (eLogWarning, "TorrentsTunnel: Tracker ", trackerID, " response code ", res->result_int());
					}
				});
		}
	}

	void TorrentsTunnel::ConnectToPeer (std::shared_ptr<Torrent> torrent, const i2p::data::IdentHash& peer)
	{
		if (!torrent) return;
		LogPrint (eLogDebug, "TorrentsTunnel: Connecting to peer ", peer.ToBase32 () + ".b32.i2p");
		if (peer == GetLocalDestination ()->GetIdentHash ())
		{
			LogPrint (eLogInfo, "TorrentsTunnel: Can't connect to self");
			return;
		}
		CreateStream ([this, torrent, peer](std::shared_ptr<i2p::stream::Stream> stream)
			{
				if (stream)
				{
					LogPrint (eLogDebug, "TorrentsTunnel: Connected to peer ", peer.ToBase32 () + ".b32.i2p");
					auto connection = std::make_shared<PeerConnection>(shared_from_this (), stream, torrent);
					AddHandler (connection);
					connection->Connect ();
				}
				else
					LogPrint (eLogInfo, "TorrentsTunnel: Can't connect to peer ", peer.ToBase32 () + ".b32.i2p");
			}, std::make_shared<i2p::client::Address>(peer), TORRENT_PORT);
	}

	size_t TorrentsTunnel::ConnectToPeers (std::shared_ptr<Torrent> torrent)
	{
		if (!torrent) return 0;
		auto peersToConnect = GetNonConnectedPeers (torrent);
		if (!peersToConnect.empty ())
		{
			for (const auto& it: peersToConnect)
				ConnectToPeer (torrent, it);
		}
		return peersToConnect.size ();
	}

	void TorrentsTunnel::ScheduleTrackerRequestsCheck ()
	{
		m_TrackerRequestsCheckTimer.expires_after (std::chrono::milliseconds(TRACKER_REQUESTS_CHECK_TIMEOUT));
		m_TrackerRequestsCheckTimer.async_wait (std::bind (&TorrentsTunnel::HandleTrackerRequestsCheckTimer,
			this, std::placeholders::_1));
	}

	void TorrentsTunnel::HandleTrackerRequestsCheckTimer (const boost::system::error_code& ecode)
	{
		if (ecode != boost::asio::error::operation_aborted)
		{
			auto ts = i2p::util::GetMonotonicMilliseconds ();
			if (!m_DatragramTrackerTransactions.empty ())
			{
				// cleanup  expired transactions
				auto it = m_DatragramTrackerTransactions.begin ();
				while (it != m_DatragramTrackerTransactions.end ())
				{
					if (ts > (std::get<5>(it->second) + DATAGRAM_TRACKER_TRANSACTION_TIMEOUT)*1000LL)
						it = m_DatragramTrackerTransactions.erase (it);
					else
						it++;
				}
			}
			for (auto it: m_Torrents)
				for (size_t i = 0; i < m_Trackers.size (); i++)
				{
					if (!it.second->GetNextTrackerRequestTime (i)) // first time
					{
						auto initialInterval = GetLocalDestination ()->GetRng()() % TRACKER_INITIAL_REQUEST_INTERVAL_VARIANCE;
						if (initialInterval <= TRACKER_REQUESTS_CHECK_TIMEOUT) initialInterval = 0; // request immeditely
						it.second->SetNextTrackerRequestTime (i, ts + initialInterval);
					}
					if (ts >= it.second->GetNextTrackerRequestTime (i))
					{
						auto nextInterval = it.second->GetInterval (i) + GetLocalDestination ()->GetRng()() % TRACKER_REQUESTS_INTERVAL_VARIANCE;
						it.second->SetNextTrackerRequestTime (i, ts + nextInterval);
						RequestTracker (i, it.second);
					}
				}
			ScheduleTrackerRequestsCheck ();
		}
	}

	void TorrentsTunnel::ScheduleKeepAliveCheck ()
	{
		m_KeepAliveCheckTimer.expires_after (std::chrono::seconds(PEER_KEEP_ALIVE_CHECK_INTERVAL));
		m_KeepAliveCheckTimer.async_wait (std::bind (&TorrentsTunnel::HandleKeepAliveCheckTimer,
			this, std::placeholders::_1));
	}

	void TorrentsTunnel::HandleKeepAliveCheckTimer (const boost::system::error_code& ecode)
	{
		if (ecode != boost::asio::error::operation_aborted)
		{
			auto ts = i2p::util::GetMonotonicSeconds ();
			IterateHandlers ([ts](std::shared_ptr<i2p::client::I2PServiceHandler> handler)
				{
					if (handler)
						std::static_pointer_cast<PeerConnection>(handler)->CheckKeepAlive (ts);
				});
			ScheduleKeepAliveCheck ();
		}
	}

	void TorrentsTunnel::ScheduleReconnectCheck ()
	{
		m_ReconnectCheckTimer.cancel ();
		m_ReconnectCheckTimer.expires_after (std::chrono::seconds(RECONNECT_CHECK_INTERVAL));
		m_ReconnectCheckTimer.async_wait (std::bind (&TorrentsTunnel::HandleReconnectCheckTimer,
			this, std::placeholders::_1));
	}

	void TorrentsTunnel::HandleReconnectCheckTimer (const boost::system::error_code& ecode)
	{
		if (ecode != boost::asio::error::operation_aborted)
		{
			for (auto it: m_Torrents)
			{
				if (!it.second->IsComplete ())
				{
					auto numPeers = ConnectToPeers (it.second);
					if (numPeers)
						LogPrint (eLogDebug, "TorrentsTunnel: Reconnecting to ", numPeers, " peers");
				}
			}
			ScheduleReconnectCheck ();
		}
	}

	void TorrentsTunnel::ScheduleStatusUpdate ()
	{
		m_TorrentsStatusUpdateTimer.cancel ();
		m_TorrentsStatusUpdateTimer.expires_after (std::chrono::seconds(TORRENTS_STATUS_UPDATE_INTERVAL));
		m_TorrentsStatusUpdateTimer.async_wait (std::bind (&TorrentsTunnel::HandleTorrentsStatusUpdateTimer,
			this, std::placeholders::_1));
	}

	void TorrentsTunnel::HandleTorrentsStatusUpdateTimer (const boost::system::error_code& ecode)
	{
		if (ecode != boost::asio::error::operation_aborted)
		{
			auto ts = i2p::util::GetMonotonicSeconds ();
			for (auto it: m_Torrents)
			{
				if (!it.second->IsComplete ())
				{
					if (it.second->UpdateStatus (ts))
						CompleteTorrent (it.second);
					else
						UpdatePeersPerPiece (it.second);
				}
			}
			UpdateStats ();
			ScheduleStatusUpdate ();
		}
	}

	std::list<std::shared_ptr<PeerConnection> > TorrentsTunnel::GetTorrentConnections (std::shared_ptr<Torrent> torrent)
	{
		std::list<std::shared_ptr<PeerConnection> > ret;
		if (torrent)
		{
			IterateHandlers ([&ret, torrent](std::shared_ptr<i2p::client::I2PServiceHandler> handler)
				{
					if (handler)
					{
						auto conn = std::static_pointer_cast<PeerConnection>(handler);
						if (conn->GetTorrent () == torrent && conn->GetStream ())
							ret.emplace_back (conn);
					}
				});
		}
		return ret;
	}

	std::unordered_set<i2p::data::IdentHash> TorrentsTunnel::GetNonConnectedPeers (std::shared_ptr<Torrent> torrent)
	{
		std::unordered_set<i2p::data::IdentHash> ret;
		if (torrent)
		{
			ret = torrent->GetPeers ();
			if(!ret.empty ())
			{
				IterateHandlers ([&ret, torrent](std::shared_ptr<i2p::client::I2PServiceHandler> handler)
					{
						if (handler)
						{
							auto conn = std::static_pointer_cast<PeerConnection>(handler);
							if (conn->GetTorrent () == torrent && conn->GetStream ())
							{
								auto ident = conn->GetStream ()->GetRemoteIdentity ();
								if (ident)
									ret.erase (ident->GetIdentHash ());
							}
						}
					});
			}
		}
		return ret;
	}

	void TorrentsTunnel::UpdatePeersPerPiece (std::shared_ptr<Torrent> torrent)
	{
		if (!torrent) return;
		torrent->StartCountingPeers ();
		IterateHandlers ([torrent](std::shared_ptr<i2p::client::I2PServiceHandler> handler)
			{
				if (handler)
				{
					auto conn = std::static_pointer_cast<PeerConnection>(handler);
					if (conn->GetTorrent () == torrent)
						torrent->ApplyPeerRemoteBitfield (conn->GetRemoteBitfield ());
				}
			});
	}

	void TorrentsTunnel::UpdateStats ()
	{
		for (auto it: m_Torrents)
			it.second->ResetStats ();
		IterateHandlers ([](std::shared_ptr<i2p::client::I2PServiceHandler> handler) mutable
			{
				if (handler)
				{
					auto conn = std::static_pointer_cast<PeerConnection>(handler);
					auto torrent = conn->GetTorrent ();
					if (torrent)
					{
						torrent->SetDownloadRate (torrent->GetDownloadRate () + conn->GetDownloadRate ());
						torrent->SetUploadRate (torrent->GetUploadRate () + conn->GetUploadRate ());
						if (conn->IsDownloading ())
							torrent->SetNumDownloadingFromPeers (torrent->GetNumDownloadingFromPeers () + 1);
						if (conn->IsUploading ())
							torrent->SetNumUploadingToPeers (torrent->GetNumUploadingToPeers () + 1);
					}
				}
			});
	}

	void TorrentsTunnel::HandleRecvFromI2PRaw (uint16_t fromPort, uint16_t toPort, const uint8_t * buf, size_t len)
	{
		// response from tracker
		if (len < 8) return;
		uint32_t action = bufbe32toh (buf);
		switch (action)
		{
			case eDatagramTrackerActionConnect:
				HandleConnectResponse (buf + 4, len - 4);
			break;
			case eDatagramTrackerActionAnnounce:
				HandleAnnounceResponse (buf + 4, len - 4);
			break;
			case eDatagramTrackerActionError:
				HandleErrorResponse (buf + 4, len - 4);
			break;
			default:
				LogPrint (eLogInfo, "TorrentsTunnel: Unexpected action ", action, " from tracker");
		}
	}

	void TorrentsTunnel::HandleErrorResponse (const uint8_t * buf, size_t len)
	{
		uint32_t transactionID = bufbe32toh (buf);
		LogPrint (eLogDebug, "TorrentsTunnel: Datagram tracker action error response ", transactionID);
		m_DatragramTrackerTransactions.erase (transactionID);
		LogPrint (eLogInfo, "TorrentsTunnel: Datagram tracker error response: ", std::string_view ((const char *)(buf + 4), len - 4));
	}

	void TorrentsTunnel::ConnectToDatagramTracker (std::shared_ptr<Torrent> torrent,
		size_t trackerID, std::string_view dest, uint16_t port, TrackerAnnounceEvent event)
	{
		LogPrint (eLogDebug, "TorrentsTunnel: Connecting to datagram tracker ", dest, ":", port);
		auto address = i2p::client::context.GetAddressBook ().GetAddress (dest);
		if (address && address->IsIdentHash ())
		{
			auto localDestination = GetLocalDestination ();
			auto dgramDest = localDestination->GetDatagramDestination ();
			if (dgramDest)
			{
				uint8_t connectRequest[16];
				htobe64buf (connectRequest, 0x41727101980); // protocol_id
				htobe32buf (connectRequest + 8, eDatagramTrackerActionConnect); // action
				uint32_t transactionID = localDestination->GetRng()();
				htobe32buf (connectRequest + 12, transactionID); // transactionID
				uint16_t fromPort = localDestination->GetRng()() % 1000 + 6000;
				auto session = dgramDest->GetSession (address->identHash);
				if (session)
				{
					m_DatragramTrackerTransactions.emplace (transactionID, std::make_tuple (torrent,
						trackerID, address->identHash, port, fromPort, i2p::util::GetMonotonicSeconds (), event));
					session->SetVersion (i2p::datagram::eDatagramV2); // send datagram2
					dgramDest->SendDatagram (session, connectRequest, 16, fromPort, port);
				}
				else
					LogPrint (eLogInfo, "TorrentsTunnel: Can't obtain datagram session to ", dest);
			}
			else
				LogPrint (eLogError, "TorrentsTunnel: Datagram destination is not avaliable");
		}
		else
			LogPrint (eLogInfo, "TorrentsTunnel: Tracker not found: ", dest);
	}

	void TorrentsTunnel::HandleConnectResponse (const uint8_t * buf, size_t len)
	{
		if (len < 12)
		{
			LogPrint (eLogInfo, "TorrentsTunnel: Unexpected connect response length ", len + 4);
			return;
		}
		uint32_t transactionID = bufbe32toh (buf);
		LogPrint (eLogDebug, "TorrentsTunnel: Datagram tracker action connect response ", transactionID);
		auto it = m_DatragramTrackerTransactions.find (transactionID);
		if (it == m_DatragramTrackerTransactions.end ())
		{
			LogPrint (eLogInfo, "TorrentsTunnel: Datagram tracker transaction ", transactionID, " not found");
			return;
		}
		uint64_t connectionID = bufbe64toh (buf + 4);
		auto [torrent, trackerID, ident, port, fromPort, ts, event] = it->second;
		if (!torrent.expired ())
			SendAnnounceToDatagramTracker (transactionID, connectionID, torrent.lock (), ident, port, fromPort, event);
		else
			m_DatragramTrackerTransactions.erase (it);
	}

	void TorrentsTunnel::SendAnnounceToDatagramTracker (uint32_t transactionID,
		uint64_t connectionID, std::shared_ptr<Torrent> torrent, const i2p::data::IdentHash& ident,
		uint16_t port, uint16_t fromPort, TrackerAnnounceEvent event)
	{
		auto localDestination = GetLocalDestination ();
		auto dgramDest = localDestination->GetDatagramDestination ();
		if (dgramDest)
		{
			uint8_t announce[98];
			htobe64buf (announce, connectionID); // connection_id
			htobe32buf (announce + 8, eDatagramTrackerActionAnnounce); // action
			htobe32buf (announce + 12, transactionID); // transaction_id
			memcpy (announce + 16, torrent->GetInfoHash ().data (), 20); // info_hash
			memcpy (announce + 36, m_PeerID.data (), 20); // peer_id
			auto left = torrent->GetLeft ();
			htobe64buf (announce + 56, torrent->GetLength () - left); // downloaded
			htobe64buf (announce + 64, left); // left
			htobe64buf (announce + 72, torrent->GetUploaded ()); // uploaded
			htobe32buf (announce + 80, event); // event
			htobe32buf (announce + 84, 0); // IP address 0:not used
			htobe32buf (announce + 88, 0); // key, ignored
			htobe32buf (announce + 92, torrent->IsComplete () ? 0 : 25); // num_want
			htobe16buf (announce + 96, fromPort); // from port
			auto session = dgramDest->GetSession (ident);
			if (session)
			{
				session->SetVersion (i2p::datagram::eDatagramV3); // send datagram3
				dgramDest->SendDatagram (session, announce, 98, fromPort, port);
			}
			else
				LogPrint (eLogInfo, "TorrentsTunnel: Can't obtain datagram session");
		}
	}

	void TorrentsTunnel::HandleAnnounceResponse (const uint8_t * buf, size_t len)
	{
		if (len < 16)
		{
			LogPrint (eLogInfo, "TorrentsTunnel: Unexpected announce response length ", len + 4);
			return;
		}
		uint32_t transactionID = bufbe32toh (buf);
		LogPrint (eLogDebug, "TorrentsTunnel: Datagram tracker action announce response ", transactionID);
		auto it = m_DatragramTrackerTransactions.find (transactionID);
		if (it == m_DatragramTrackerTransactions.end ())
		{
			LogPrint (eLogInfo, "TorrentsTunnel: Datagram tracker transaction ", transactionID, " not found");
			return;
		}
		auto [torrent, trackerID, ident, port, fromPort, ts, event] = it->second;
		if (!torrent.expired ())
		{
			uint32_t interval = bufbe32toh (buf + 4);
			torrent.lock ()->HandleDatagramTrackerResponse (trackerID, interval, buf + 16, len - 16);
		}
		m_DatragramTrackerTransactions.erase (it);
	}
}
}