feat: progress AyuSync implementation

Telegram/SourceFiles/ayu/libs/pipe.hpp

@@ -85,7 +85,7 @@ public:
     using off_type     = typename Traits::off_type;
 
 public:
-    static constexpr std::size_t buf_size{1024};
+    static constexpr std::size_t buf_size{4096};
 
 public:
     basic_pipe_streambuf() = default;
@@ -138,7 +138,7 @@ public:
             {
                 native_mode = mode & std::ios_base::in ? PIPE_ACCESS_INBOUND : PIPE_ACCESS_OUTBOUND;
 
-                handle = CreateNamedPipeW(std::data(native_name), native_mode, PIPE_READMODE_BYTE | PIPE_WAIT, PIPE_UNLIMITED_INSTANCES, buf_size, buf_size, 0, nullptr);
+                handle = CreateNamedPipeW(std::data(native_name), native_mode, PIPE_TYPE_BYTE | PIPE_WAIT, PIPE_UNLIMITED_INSTANCES, buf_size, buf_size, 0, nullptr);
                 if(handle == INVALID_HANDLE_VALUE)
                     return false;
 

Telegram/SourceFiles/ayu/sync/ayu_sync_controller.cpp

@@ -17,7 +17,10 @@
 #include <QString>
 #include <thread>
 
-#include "core/sandbox.h"
+#include "data/data_histories.h"
+
+#include "history/history_item.h"
+#include "history/view/history_view_element.h"
 
 namespace AyuSync
 {
@@ -30,7 +33,7 @@ namespace AyuSync
 
 	bool isAgentRunning()
 	{
-		return is_process_running(AgentFilename);
+		return isProcessRunning(AgentFilename);
 	}
 
 	void initialize()
@@ -43,7 +46,7 @@ namespace AyuSync
 		controller = ayu_sync_controller();
 	}
 
-	ayu_sync_controller& getControllerInstance()
+	ayu_sync_controller& getInstance()
 	{
 		initialize();
 		return controller.value();
@@ -56,20 +59,44 @@ namespace AyuSync
 			return;
 		}
 
-		if (!isAgentRunning())
+		if (isAgentRunning())
 		{
-			auto configPath = std::filesystem::absolute("./tdata/sync_preferences.json");
+			killProcess(AgentFilename);
-			auto process = nes::process{AgentPath, {configPath.string(), ""}, nes::process_options::none};
-			process.detach();
 		}
 
+		auto configPath = std::filesystem::absolute("./tdata/sync_preferences.json");
+		auto process = nes::process{AgentPath, {configPath.string(), ""}, nes::process_options::none};
+		process.detach();
+
 		std::thread receiverThread(&ayu_sync_controller::receiver, this);
 		receiverThread.detach();
+
+		initialized = true;
+	}
+
+	void ayu_sync_controller::syncRead(not_null<History*> history, MsgId untilId)
+	{
+		if (!initialized)
+		{
+			return;
+		}
+
+		SyncRead ev;
+		ev.userId = history->owner().session().userId().bare;
+
+		ev.args.dialogId = getDialogIdFromPeer(history->peer);
+		ev.args.untilId = untilId.bare;
+		ev.args.unread = history->unreadCount();
+
+		pipe->send(ev);
 	}
 
 	void ayu_sync_controller::receiver()
 	{
 		pipe = std::make_unique<ayu_pipe_wrapper>();
+		pipe->connect();
+
+		LOG(("Pipe created"));
 
 		while (true)
 		{
@@ -123,6 +150,47 @@ namespace AyuSync
 
 	void ayu_sync_controller::onSyncForce(SyncForce ev)
 	{
+		auto session = getSession(ev.userId);
+		auto histories = session->data().chatsList();
+
+		SyncBatch readsBatchEvent;
+		readsBatchEvent.userId = ev.userId;
+
+		for (const auto& row : histories->indexed()->all())
+		{
+			if (const auto history = row->history())
+			{
+				auto dialogId = getDialogIdFromPeer(history->peer);
+
+				SyncRead readEv;
+				readEv.userId = ev.userId;
+
+				history->calculateFirstUnreadMessage();
+				auto unreadElement = history->firstUnreadMessage();
+
+				if (!unreadElement && history->unreadCount())
+				{
+					LOG(("No unread can be calculated for %1").arg(dialogId));
+					continue;
+				}
+
+				auto untilId = unreadElement ? unreadElement->data()->id.bare : history->lastMessage()->id.bare;
+
+				readEv.args.dialogId = dialogId;
+				readEv.args.untilId = untilId;
+				readEv.args.unread = history->unreadCount();
+
+				readsBatchEvent.args.events.emplace_back(readEv);
+			}
+		}
+
+		pipe->send(readsBatchEvent);
+
+		// send finish event
+		SyncForceFinish newEv;
+		newEv.userId = ev.userId;
+
+		pipe->send(newEv);
 	}
 
 	void ayu_sync_controller::onSyncBatch(json ev)

Telegram/SourceFiles/ayu/sync/ayu_sync_controller.h

@@ -9,6 +9,9 @@
 
 #include "models.h"
 #include "ayu/libs/json.hpp"
+
+#include "history/history.h"
+
 #include "utils/ayu_pipe_wrapper.h"
 
 using json = nlohmann::json;
@@ -29,6 +32,8 @@ namespace AyuSync
 	public:
 		void initializeAgent();
 
+		void syncRead(not_null<History*> history, MsgId untilId);
+
 		void onSyncForce(SyncForce ev);
 		void onSyncBatch(json ev);
 		void onSyncRead(SyncRead ev);
@@ -39,9 +44,10 @@ namespace AyuSync
 		void receiver();
 
 		std::unique_ptr<ayu_pipe_wrapper> pipe;
+		bool initialized;
 	};
 
-	ayu_sync_controller& getControllerInstance();
+	ayu_sync_controller& getInstance();
 
 	bool isAgentDownloaded();
 	bool isAgentRunning();

Telegram/SourceFiles/ayu/sync/models.h

@@ -7,6 +7,8 @@
 
 #define ID long long
 
+using json = nlohmann::json;
+
 class SyncEvent
 {
 public:
@@ -17,13 +19,15 @@ public:
 class SyncBatch : public SyncEvent
 {
 public:
-	std::string type = "sync_batch";
+	explicit SyncBatch()
-	ID userId;
+	{
+		type = "sync_batch";
+	}
 
 	class SyncBatchArgs
 	{
 	public:
-		std::vector<SyncEvent> events;
+		std::vector<json> events;
 	};
 
 	SyncBatchArgs args;
@@ -32,8 +36,10 @@ public:
 class SyncRead : public SyncEvent
 {
 public:
-	std::string type = "sync_read";
+	explicit SyncRead()
-	ID userId;
+	{
+		type = "sync_read";
+	}
 
 	class SyncReadArgs
 	{
@@ -49,8 +55,10 @@ public:
 class SyncForce : public SyncEvent
 {
 public:
-	std::string type = "sync_force";
+	explicit SyncForce()
-	ID userId;
+	{
+		type = "sync_force";
+	}
 
 	class SyncForceArgs
 	{
@@ -64,8 +72,10 @@ public:
 class SyncForceFinish : public SyncEvent
 {
 public:
-	std::string type = "sync_force_finish";
+	explicit SyncForceFinish()
-	ID userId;
+	{
+		type = "sync_force_finish";
+	}
 
 	class SyncForceFinishArgs
 	{

Telegram/SourceFiles/ayu/sync/utils/ayu_pipe_wrapper.cpp

@@ -11,29 +11,34 @@
 
 using stringbuf = std::basic_stringbuf<unsigned char, std::char_traits<unsigned char>, std::allocator<unsigned char>>;
 
-template <class T>
+void ayu_pipe_wrapper::connect()
-void ayu_pipe_wrapper::send(T obj)
 {
-	//    auto s = json(obj).dump();
+	is = std::make_unique<pipein>("AyuSync");
-	//    auto length = s.length();
+	receive();
-	//    char lengthBuff[4];
+	os = std::make_unique<pipeout>("AyuSync1338");
-	//    bit_converter::i32_to_bytes(length, false, lengthBuff);
+}
-	//
+
-	//    os.write(lengthBuff, 4);
+void ayu_pipe_wrapper::send(json p)
-	//    os.write(s.c_str(), length);
+{
-	//    os.flush();
+	auto s = p.dump();
-	throw std::logic_error("not implemented");
+	auto length = s.length();
+	unsigned char lengthBuff[4];
+	bit_converter::i32_to_bytes(length, false, lengthBuff);
+
+	os->write(lengthBuff, 4);
+	os->write(reinterpret_cast<const unsigned char*>(s.c_str()), length);
+	os->flush();
 }
 
 std::optional<json> ayu_pipe_wrapper::receive()
 {
-	if (!is.is_open())
+	if (!is->is_open())
 	{
 		return std::nullopt;
 	}
 
 	unsigned char lengthBuff[4];
-	is.read(lengthBuff, 4);
+	is->read(lengthBuff, 4);
 
 	auto length = bit_converter::bytes_to_i32(lengthBuff, false);
 
@@ -50,9 +55,9 @@ std::optional<json> ayu_pipe_wrapper::receive()
 	while (length > 0)
 	{
 		auto readSize = std::min(length, static_cast<int>(sizeof(buff)));
-		is.read(buff, readSize);
+		is->read(buff, readSize);
 
-		auto reallyRead = is.gcount();
+		auto reallyRead = is->gcount();
 
 		sb.sputn(buff, reallyRead);
 		length -= reallyRead;

Telegram/SourceFiles/ayu/sync/utils/ayu_pipe_wrapper.h

@@ -12,15 +12,18 @@
 #include "ayu/sync/models.h"
 
 using json = nlohmann::json;
+using pipein = nes::basic_pipe_istream<unsigned char>;
+using pipeout = nes::basic_pipe_ostream<unsigned char>;
 
 class ayu_pipe_wrapper
 {
 public:
-	template <class T>
+	void connect();
-	void send(T obj);
 
+	void send(json p);
 	std::optional<json> receive();
 
 private:
-	nes::basic_pipe_istream<unsigned char> is{"AyuSync"};
+	std::unique_ptr<pipein> is;
+	std::unique_ptr<pipeout> os;
 };

Telegram/SourceFiles/ayu/sync/utils/process_utils.hpp

@@ -1,23 +1,18 @@
-#include <algorithm>
 #include <codecvt>
 #include <iostream>
 #include <locale>
 #include <string>
-#include <vector>
 
 #ifdef _WIN32
 
 #include <tlhelp32.h>
 #include <windows.h>
 
-#else
-#include <dirent.h>
-#include <unistd.h>
 #endif
 
 // A function to check if a process is running by its name
 // Bing AI generated
-bool is_process_running(const std::string& name)
+inline bool isProcessRunning(const std::string& name)
 {
 #ifdef _WIN32
 	// Create a snapshot of all processes
@@ -52,41 +47,50 @@ bool is_process_running(const std::string& name)
 	CloseHandle(snapshot);
 	return false;
 #else
-    // Open the /proc directory
+	throw std::logic_error("not implemented");
-    DIR* dir = opendir("/proc");
+#endif
-    if (dir == nullptr) {
+}
-        std::cerr << "Failed to open /proc\n";
+
-        return false;
+// Copilot generated
-    }
+void killProcess(const std::string& name)
-    // Read the subdirectories
+{
-    struct dirent* entry;
+#ifdef _WIN32
-    while ((entry = readdir(dir)) != nullptr) {
+	HANDLE snapshot = CreateToolhelp32Snapshot(TH32CS_SNAPPROCESS, 0);
-        // Check if the subdirectory is a number (pid)
+	if (snapshot == INVALID_HANDLE_VALUE)
-        std::string pid = entry->d_name;
+	{
-        if (std::all_of(pid.begin(), pid.end(), isdigit)) {
+		std::cerr << "Failed to create snapshot\n";
-            // Read the /proc/pid/cmdline file
+		return;
-            std::string cmdline_file = "/proc/" + pid + "/cmdline";
+	}
-            std::ifstream file(cmdline_file);
+	// Iterate over the processes and compare the names
-            if (file.is_open()) {
+	PROCESSENTRY32 entry;
-                // Get the first word of the file (process name)
+	entry.dwSize = sizeof(entry);
-                std::string process_name;
+	if (!Process32First(snapshot, &entry))
-                std::getline(file, process_name, '\0');
+	{
-                // Remove the path if present
+		std::cerr << "Failed to get first process\n";
-                size_t pos = process_name.rfind('/');
+		CloseHandle(snapshot);
-                if (pos != std::string::npos) {
+		return;
-                    process_name = process_name.substr(pos + 1);
+	}
-                }
+	do
-                // Compare the names
+	{
-                if (name == process_name) {
+		std::wstring_convert<std::codecvt_utf8_utf16<wchar_t>> converter;
-                    // Found a match
+		std::string entry_name = converter.to_bytes(entry.szExeFile);
-                    closedir(dir);
+		if (name == entry_name)
-                    return true;
+		{
-                }
+			// Found a match
-            }
+			HANDLE hProcess = OpenProcess(PROCESS_TERMINATE, FALSE, entry.th32ProcessID);
-        }
+			if (hProcess != nullptr)
-    }
+			{
-    // No match found
+				TerminateProcess(hProcess, 9);
-    closedir(dir);
+				CloseHandle(hProcess);
-    return false;
+			}
+			CloseHandle(snapshot);
+			return;
+		}
+	}
+	while (Process32Next(snapshot, &entry));
+	// No match found
+	CloseHandle(snapshot);
+#else
+	throw std::logic_error("not implemented");
 #endif
 }

Telegram/SourceFiles/ayu/sync/utils/telegram_helpers.cpp

@@ -65,3 +65,21 @@ not_null<History*> getHistoryFromDialogId(ID dialogId, Main::Session* session)
 
 	return session->data().history(peerFromChat(abs(dialogId)));
 }
+
+ID getDialogIdFromPeer(not_null<PeerData*> peer)
+{
+	auto peerId = peerIsUser(peer->id)
+		              ? peerToUser(peer->id).bare
+		              : peerIsChat(peer->id)
+		              ? peerToChat(peer->id).bare
+		              : peerIsChannel(peer->id)
+		              ? peerToChannel(peer->id).bare
+		              : peer->id.value;
+
+	if (peer->isChannel() || peer->isChat())
+	{
+		peerId = -peerId;
+	}
+
+	return peerId;
+}

Telegram/SourceFiles/ayu/sync/utils/telegram_helpers.h

@@ -18,3 +18,4 @@ Main::Session* getSession(ID userId);
 bool accountExists(ID userId);
 void dispatchToMainThread(std::function<void()> callback);
 not_null<History*> getHistoryFromDialogId(ID dialogId, Main::Session* session);
+ID getDialogIdFromPeer(not_null<PeerData*> peer);

Telegram/SourceFiles/ayu/ui/settings/settings_ayu.cpp

@@ -376,7 +376,7 @@ namespace Settings
 			st::settingsButtonNoIcon
 		)->addClickHandler([=]
 		{
-			auto controller = &AyuSync::getControllerInstance();
+			auto controller = &AyuSync::getInstance();
 			controller->initializeAgent();
 		});
 	}

Telegram/SourceFiles/data/data_histories.cpp

@@ -24,8 +24,10 @@ https://github.com/telegramdesktop/tdesktop/blob/master/LEGAL
 #include "core/application.h"
 #include "apiwrap.h"
 
+// AyuGram includes
 #include "ayu/ayu_settings.h"
 #include "ayu/ayu_state.h"
+#include "ayu/sync/ayu_sync_controller.h"
 
 namespace Data {
 namespace {
@@ -176,6 +178,8 @@ void Histories::readInboxTill(
 
 	Core::App().notifications().clearIncomingFromHistory(history);
 
+	AyuSync::getInstance().syncRead(history, tillId);
+
     // AyuGram sendReadPackets
     const auto settings = &AyuSettings::getInstance();
     auto allow = settings->sendReadPackets;