#include "EventHandler.h" #include #include #include #include #include #include #include #include #include #include "dao/MaterialDAO.h" #include "dao/BankProfileDAO.h" #include "dao/DeviceDAO.h" #include "dao/EventDAO.h" #include "dao/SettingsDAO.h" #include "dao/TransactionDAO.h" #include "db/BankProfileInfo.h" #include "adb/AdbUtils.h" #include "update/UpdatePause.h" #include "dao/GeneralLogDAO.h" #include "net/NetworkService.h" #include "black/GetLastTransactionsScript.h" #include "black/LoginAndCheckAccountsScript.h" #include "ozon/GetLastTransactionsScript.h" #include "ozon/LoginAndCheckAccountsScript.h" #include "black/PayByCardScript.h" #include "ozon/PayByCardScript.h" #include "ozon/PayByPhoneScript.h" #include "dushanbe/LoginAndCheckAccountsScript.h" #include "dushanbe/GetLastTransactionsScript.h" #include "dushanbe/PayByCardScript.h" #include "dushanbe/PayByPhoneScript.h" static void sendTaskResult(const QString &type, const QString &materialId, const QString &bankName, const QJsonValue &data, int eventId = -1, bool updateStatus = true); static void sendTaskError(const QString &type, const QString &materialId, const QString &bankName, const QString &description, const QByteArray &screenshot = {}, int eventId = -1, const QString &transactionId = {}, int accountId = -1); static void sendMonitoringLog(const QString &type, const QString &event, const QString &message, const QByteArray &image = {}, int eventId = -1, bool toApiBase = false); EventHandler::EventHandler(QObject *parent) : QObject(parent) {} EventHandler::~EventHandler() = default; void EventHandler::start() { qDebug() << "[EventHandler] start() called, thread:" << QThread::currentThread(); m_network = new NetworkService(this); connect(m_network, &NetworkService::tasksReceived, this, &EventHandler::onTasksReceived); m_pollTimer = new QTimer(this); m_pollTimer->setInterval(5000); connect(m_pollTimer, &QTimer::timeout, this, [this]() { // На время установки обновления не трогаем устройства и не запрашиваем задачи. if (UpdatePause::isPaused()) return; // Пробуем запустить задачи из очереди для свободных устройств const QStringList onlineDevices = DeviceDAO::getOnlineDeviceCodes(); for (const QString &deviceId : onlineDevices) { processNextTask(deviceId); } // Запрашиваем новые задачи только для устройств без pending events const QSet busyDevices = EventDAO::getDevicesWithPendingEvents(); const QStringList materialIds = MaterialDAO::getActiveMaterialIdsExcludingDevices(busyDevices); if (!materialIds.isEmpty()) { m_network->getTasks(materialIds); } else { // qDebug() << "[EventHandler] No free materials to poll." // << "busyDevices:" << busyDevices // << "onlineDevices:" << onlineDevices; } }); m_pollTimer->start(); // Закрываем зависшие события с прошлого запуска (только не отправленные на сервер) for (const EventStatus st : {EventStatus::Wait, EventStatus::InProgress}) { const QList stale = EventDAO::getAllEvents(st); for (const EventInfo &e : stale) { if (e.sentToServer) continue; EventDAO::updateEvent(e.id, EventStatus::Error, "App restarted"); EventDAO::markSentToServer(e.id); sendTaskError(eventTypeToString(e.type), e.externalId, e.bankName, "App restarted, event was pending", {}, e.id, e.transactionId, e.accountId); qDebug() << "[EventHandler] Closed stale event id:" << e.id; } } // Первый запрос сразу при старте qDebug() << "[EventHandler] starting first getTasks()"; m_network->getTasks(); } void EventHandler::onTasksReceived(const QJsonArray &tasks) { if (!m_running) return; if (!tasks.isEmpty()) qDebug() << "[EventHandler] onTasksReceived, count:" << tasks.size(); // Получаем список онлайн устройств один раз const QStringList onlineDevices = DeviceDAO::getOnlineDeviceCodes(); // 1. Сохраняем новые задачи в БД for (const auto &taskVal : tasks) { const QJsonObject task = taskVal.toObject(); const QString bankName = task["bank_name"].toString(); const QString materialId = task["material_id"].isString() ? task["material_id"].toString() : QString::number(task["material_id"].toInt()); const QString typeStr = task["type"].toString(); const EventType eventType = eventTypeFromString(typeStr); if (eventType == EventType::Unknown) { qWarning() << "[EventHandler] Unknown task type:" << typeStr; continue; } // transaction_id присутствует только в CREATE_TRANSACTION; для других типов это пусто. const QString incomingTransactionId = task["body"].toObject()["transaction_id"].toString(); // Ищем аккаунт по material_id напрямую const MaterialInfo account = MaterialDAO::findByMaterialId(materialId); if (account.id == -1) { qWarning() << "[EventHandler] Account not found for bank:" << bankName << "material_id:" << materialId; sendTaskError(typeStr, materialId, bankName, "Account not found for material_id: " + materialId, {}, -1, incomingTransactionId); continue; } // FETCH_PROFILE: отменяем старые ожидающие задачи такого же типа для этого материала if (eventType == EventType::FetchProfile) { const QList cancelled = EventDAO::cancelWaitingByMaterial(materialId, EventType::FetchProfile); for (const EventInfo &c : cancelled) { sendTaskError(eventTypeToString(c.type), c.externalId, c.bankName, "Cancelled: replaced by new task", {}, c.id, c.transactionId, c.accountId); EventDAO::markSentToServer(c.id); } } EventInfo event; event.status = EventStatus::Wait; event.type = eventType; event.externalId = materialId; event.accountId = account.id; event.deviceId = account.deviceId; event.bankName = bankName; event.timestamp = QDateTime::currentDateTimeUtc(); event.json = QJsonDocument(task).toJson(QJsonDocument::Compact); const QJsonObject body = task["body"].toObject(); if (eventType == EventType::FetchTransactions) { event.amount = body["amount"].toDouble() / 100.0; } if (eventType == EventType::CreateTransaction) { event.amount = body["amount"].toDouble() / 100.0; event.phone = body["bank_participator_id"].toString(); event.transactionId = body["transaction_id"].toString(); // Название банка-получателя из extra const QString bankNameRu = body["extra"].toObject()["bank_participator_name_ru"].toString(); if (!bankNameRu.isEmpty()) { event.comment = bankNameRu; } } const int eventId = EventDAO::insertEvent(event); if (eventId == -1) { qCritical() << "[EventHandler] Failed to insert event, external_id:" << materialId; continue; } // Если устройство офлайн — сразу отправляем ошибку на сервер if (!onlineDevices.contains(account.deviceId)) { qWarning() << "[EventHandler] Device offline for material_id:" << materialId << "device:" << account.deviceId; EventDAO::updateEvent(eventId, EventStatus::Error, "Device is offline"); sendTaskError(typeStr, materialId, bankName, "Device is offline", {}, eventId, incomingTransactionId, account.id); } } // 2. Для каждого онлайн устройства без активного потока — запускаем следующую задачу for (const QString &deviceId : onlineDevices) { processNextTask(deviceId); } } static void sendTaskResult(const QString &type, const QString &materialId, const QString &bankName, const QJsonValue &data, int eventId, bool updateStatus) { QJsonObject obj; obj["type"] = type; obj["material_id"] = materialId; // UUID string obj["bank_name"] = bankName; obj["data"] = data; if (eventId > 0) obj["internal_task_id"] = eventId; auto *thread = new QThread; auto *networkService = new NetworkService; networkService->moveToThread(thread); networkService->setPayload(QJsonDocument(obj).toJson()); QObject::connect(thread, &QThread::started, networkService, &NetworkService::postTaskResult); QObject::connect(networkService, &NetworkService::finishedWithResult, networkService, [eventId, updateStatus](int result) { // updateStatus=false для потоковых per-tx результатов FETCH_TRANSACTIONS: // статус события финализируется один раз в updateEventThread, а не на // каждой отдельной отправке найденной транзакции. if (eventId != -1 && updateStatus) { if (result == 0) { EventDAO::markSentToServer(eventId); } else { EventDAO::updateEvent(eventId, EventStatus::Error, "Failed to send result to server"); qWarning() << "[EventHandler] Task result not sent, marking event" << eventId << "as error"; } } }); QObject::connect(networkService, &NetworkService::finished, thread, &QThread::quit); QObject::connect(networkService, &NetworkService::finished, networkService, &QObject::deleteLater); QObject::connect(thread, &QThread::finished, thread, &QThread::deleteLater); thread->start(); } static void sendTaskError(const QString &type, const QString &materialId, const QString &bankName, const QString &description, const QByteArray &screenshot, int eventId, const QString &transactionId, int accountId) { QJsonObject errorData; errorData["status"] = QStringLiteral("error"); errorData["event_type"] = type; errorData["description"] = description; if (!transactionId.isEmpty()) errorData["transaction_id"] = transactionId; sendTaskResult(QStringLiteral("ERROR"), materialId, bankName, errorData, eventId); // Шлём в мониторинг всегда — даже без скриншота (offline / expired / // disabled / account not found и т.п.), чтобы такие ошибки не «молчали» на // релее. Скриншот прикладываем, если он есть. // api-токен общий на нескольких операторов — по нему не отличить, чей // профиль упал. Резолвим банковский профиль по accountId и пишем его // серверный UUID + материал + телефон профиля. QString profileUuid, profilePhone; if (accountId > 0) { const MaterialInfo acc = MaterialDAO::getAccountById(accountId); if (acc.id != -1) { const BankProfileInfo app = BankProfileDAO::getApplication(acc.deviceId, acc.appCode); profileUuid = app.bankProfileId; profilePhone = app.phone; } } const QString desktopId = SettingsDAO::get("desktop_id"); QStringList lines; lines << QStringLiteral("[TASK_ERROR] [%1] [%2]") .arg(type.toHtmlEscaped(), bankName.toUpper().toHtmlEscaped()); if (!desktopId.isEmpty()) lines << QStringLiteral("Desktop: %1").arg(desktopId.toHtmlEscaped()); if (!profileUuid.isEmpty()) lines << QStringLiteral("Profile: %1").arg(profileUuid.toHtmlEscaped()); if (!materialId.isEmpty()) lines << QStringLiteral("Material: %1").arg(materialId.toHtmlEscaped()); if (!profilePhone.isEmpty()) lines << QStringLiteral("Phone: %1").arg(profilePhone.toHtmlEscaped()); lines << QStringLiteral("Error: %1").arg(description.toHtmlEscaped()); const QString message = lines.join(QStringLiteral("\n")); sendMonitoringLog("error", "task_error", message, screenshot, eventId); // application.log + БД отправляются из TaskDumpRecorder::finishInternal, // чтобы покрывать также non-event-driven задачи (login flow и т.п.). } static void sendMonitoringLog(const QString &type, const QString &event, const QString &message, const QByteArray &image, int eventId, bool toApiBase) { auto *thread = new QThread; auto *networkService = new NetworkService; networkService->moveToThread(thread); if (eventId > 0) networkService->addExtra("internal_task_id", eventId); QObject::connect(thread, &QThread::started, networkService, [networkService, type, event, message, image, toApiBase]() { networkService->postMonitoringLog(type, event, message, image, toApiBase); emit networkService->finished(); }); QObject::connect(networkService, &NetworkService::finished, thread, &QThread::quit); QObject::connect(networkService, &NetworkService::finished, networkService, &QObject::deleteLater); QObject::connect(thread, &QThread::finished, thread, &QThread::deleteLater); thread->start(); } static void sendEventStatus(const QString &deviceId, const QString &externalId, const int eventId, const QString &status, const QString &error) { QJsonObject obj; obj["id"] = externalId; obj["externalId"] = eventId; obj["status"] = status; obj["error"] = error; auto *paymentThread = new QThread; auto *networkService = new NetworkService; networkService->moveToThread(paymentThread); networkService->setDeviceId(deviceId); networkService->setPayload(QJsonDocument(obj).toJson()); QObject::connect(paymentThread, &QThread::started, networkService, &NetworkService::postEventStatus); QObject::connect(networkService, &NetworkService::finished, paymentThread, &QThread::quit); QObject::connect(networkService, &NetworkService::finished, networkService, &QObject::deleteLater); QObject::connect(paymentThread, &QThread::finished, paymentThread, &QThread::deleteLater); paymentThread->start(); } bool EventHandler::tryRetry(const QString &key, const EventInfo &event, const QString &reason) { constexpr int kMaxRetries = 2; // ещё 2 попытки сверх первой → всего 3 запуска const int used = m_retryCount.value(event.id, 0); if (used >= kMaxRetries) { m_retryCount.remove(event.id); // попытки исчерпаны — пусть caller обработает как ошибку return false; } const int attempt = used + 1; m_retryCount[event.id] = attempt; qWarning() << "[EventHandler] FETCH_TRANSACTIONS retry" << attempt << "/" << kMaxRetries << "for event" << event.id << "reason:" << reason; GeneralLogDAO::insert("WARNING", "EventHandler", QString("[FETCH_TRANSACTIONS] retry %1/%2 device=%3 material=%4: %5") .arg(QString::number(attempt), QString::number(kMaxRetries), event.deviceId, event.externalId, reason)); // Чистим артефакты прошлой попытки. Ретрай запускается только когда ещё // ничего не отправлено (см. guard в updateEventThread), так что сбрасываем // счётчик отправленных в 0. m_screenshots.remove(key); m_parsedTxCount.remove(key); // Снимаем текущий поток/таймер (в timeout-ветке они уже сняты — take() вернёт null) if (const auto tm = m_timers.take(key)) { tm->stop(); tm->deleteLater(); } m_threads.remove(key); // Сбрасываем банковское приложение, чтобы попытка стартовала с чистого экрана const MaterialInfo account = MaterialDAO::getAccountById(event.accountId); const BankProfileInfo app = BankProfileDAO::getApplication(event.deviceId, account.appCode); if (!app.package.isEmpty()) { const DeviceInfo device = DeviceDAO::getDeviceById(event.deviceId); const QString adbSerial = device.adbSerial.isEmpty() ? event.deviceId : device.adbSerial; AdbUtils::tryToKillApplication(adbSerial, app.package); } // Держим событие InProgress и перезапускаем скрипт EventDAO::updateEvent(event.id, EventStatus::InProgress, ""); startThreadForTask(account, event); return true; } void EventHandler::updateEventThread(const QString &key, const EventInfo &event, const QString &result) { // Поиск транзакции: при ошибке скрипта перезапускаем (ещё до 2 раз) вместо // немедленной отправки ошибки на сервер. Но только если НИ ОДНОЙ транзакции // ещё не отправлено: при частичном успехе (часть списка найдена и уже // улетела на сервер) ретрай бессмыслен — переотправит дубли, поэтому // ненайденные просто репортим ошибкой ниже. // Ретраи отключены: при ошибке сразу репортим как Error без повторов. // if (!result.isEmpty() && event.type == EventType::FetchTransactions // && m_parsedTxCount.value(key, 0) == 0 // && tryRetry(key, event, result)) { // return; // } const EventStatus newStatus = result.isEmpty() ? EventStatus::Complete : EventStatus::Error; if (!EventDAO::updateEvent(event.id, newStatus, result)) { qCritical() << "[EventHandler] Failed to update event:" << event.id; } else { if (newStatus == EventStatus::Error) { // Используем скриншот из скрипта если есть, иначе снимаем новый QByteArray screenshot = m_screenshots.take(key); // "Interrupted" — не самостоятельная ошибка, а следствие внешнего // requestInterruption(): его выставляет либо watchdog (он уже шлёт // "Script timeout (N min)" в startThreadForTask), либо clearAll() при // остановке/выходе (намеренная остановка). Реальную причину репортит // инициатор прерывания, поэтому здесь Interrupted на сервер НЕ // дублируем — иначе на один таймаут выходит три записи в мониторинге // (timeout-log + interrupted-log + interrupted-dump). Логику declined // для CREATE_TRANSACTION ниже это не затрагивает. if (result != "Interrupted") { if (screenshot.isEmpty() && !event.deviceId.isEmpty()) { const DeviceInfo device = DeviceDAO::getDeviceById(event.deviceId); const QString adbSerial = device.adbSerial.isEmpty() ? event.deviceId : device.adbSerial; screenshot = AdbUtils::captureScreenshotBytes(adbSerial); } if (!screenshot.isEmpty()) { EventDAO::updateScreenshot(event.id, screenshot); } // Дублируем ошибку в general_logs со скриншотом const DeviceInfo logDevice = DeviceDAO::getDeviceById(event.deviceId); const QString deviceLabel = logDevice.name.isEmpty() ? event.deviceId : logDevice.name; const QString logMsg = QString("[%1] device=%2 bank=%3: %4") .arg(eventTypeToString(event.type), deviceLabel, event.bankName, result); GeneralLogDAO::insert("CRITICAL", "EventHandler", logMsg, screenshot); sendTaskError(eventTypeToString(event.type), event.externalId, event.bankName, result, screenshot, event.id, event.transactionId, event.accountId); } else { qDebug() << "[EventHandler] Interrupted — server error report skipped (cause already reported by interrupter), event id:" << event.id; } if (event.type == EventType::CreateTransaction) { const MaterialInfo declAcc = MaterialDAO::getAccountById(event.accountId); const QMap declCurrencyCodes = { {"RUB", 643}, {"USD", 840}, {"KRW", 410}, {"AZN", 944}, {"ARS", 32}, {"TJS", 972} }; const int declCurrencyCode = declCurrencyCodes.value(declAcc.currency.toUpper(), 643); const QString declTxType = QJsonDocument::fromJson(event.json.toUtf8()) .object()["body"].toObject()["type"].toString(); QString declTxId; const QList declTxList = TransactionDAO::getTransactionsWithinHours(event.accountId, 1); for (const auto &tx : declTxList) { if (!tx.bankTrExternalId.isEmpty() && qAbs(tx.amount) == event.amount && tx.phone == event.phone) { declTxId = tx.bankTrExternalId; break; } } // Скрипт мог упасть до записи транзакции в БД — сервер всё равно // ждёт уникальный bank_transaction_id. Генерируем по той же схеме, // что и в success-ветке скриптов: __. if (declTxId.isEmpty()) { const BankProfileInfo declApp = BankProfileDAO::getApplication(declAcc.deviceId, event.bankName); declTxId = event.bankName + "_" + declApp.phone + "_" + QString::number(QDateTime::currentSecsSinceEpoch()); } QJsonObject declined; if (!event.transactionId.isEmpty()) declined["transaction_id"] = event.transactionId; declined["bank_transaction_id"] = declTxId; declined["amount"] = -static_cast(event.amount * 100); declined["currency"] = declCurrencyCode; declined["transaction_type"] = declTxType.isEmpty() ? QStringLiteral("top-up") : declTxType; declined["bank_participator_id"] = event.phone; declined["status"] = QStringLiteral("declined"); declined["extra"] = QJsonObject(); declined["created_at"] = QDateTime::currentDateTimeUtc().toString(Qt::ISODate); sendTaskResult(QStringLiteral("CREATE_TRANSACTION"), event.externalId, event.bankName, declined, event.id); } EventDAO::markSentToServer(event.id); } if (newStatus != EventStatus::Error) { const MaterialInfo account = MaterialDAO::getAccountById(event.accountId); const QMap currencyCodes = { {"RUB", 643}, {"USD", 840}, {"KRW", 410}, {"AZN", 944}, {"ARS", 32}, {"TJS", 972} }; const int currencyCode = currencyCodes.value(account.currency.toUpper(), 643); QJsonValue taskData; if (event.type == EventType::FetchProfile) { // FETCH_PROFILE: баланс из bank_profiles (скрипт обновляет через // BankProfileDAO::updateAmount), fallback на materials. const BankProfileInfo freshApp = BankProfileDAO::getApplication( account.deviceId, event.bankName); double freshAmount = (freshApp.id != -1 && freshApp.amount > 0.0) ? freshApp.amount : account.amount; EventDAO::updateEventAmount(event.id, freshAmount); QJsonObject obj; obj["bank_account_id"] = account.materialId; obj["amount"] = static_cast(freshAmount * 100); obj["currency"] = currencyCode; taskData = obj; } else if (event.type == EventType::FetchTransactions) { // FETCH_TRANSACTIONS шлёт результаты потоково: каждая найденная // транзакция уже отправлена отдельным task_result по сигналу // transactionParsed (см. startThreadForTask). Здесь, при полном // успехе (m_error пуст → найдены все цели списка), отдельный // финальный task_result НЕ отправляем — только финализируем // событие, чтобы оно не считалось «зависшим» при перезапуске. EventDAO::markSentToServer(event.id); } else if (event.type == EventType::CreateTransaction) { // Определяем тип транзакции из JSON задачи const QJsonObject taskJson = QJsonDocument::fromJson(event.json.toUtf8()).object(); const QString txType = taskJson["body"].toObject()["type"].toString(); QJsonObject obj; if (!event.transactionId.isEmpty()) obj["transaction_id"] = event.transactionId; obj["amount"] = -static_cast(event.amount * 100); obj["currency"] = currencyCode; obj["transaction_type"] = txType.isEmpty() ? "top-up" : txType; obj["bank_participator_id"]= event.phone; obj["status"] = "completed"; obj["created_at"] = QDateTime::currentDateTimeUtc().toString(Qt::ISODate); // Берём bank_transaction_id и fee из последней сохранённой транзакции QList txList = TransactionDAO::getTransactionsWithinHours(event.accountId, 1); int feeInCents = 0; for (const auto &tx : txList) { if (!tx.bankTrExternalId.isEmpty() && qAbs(tx.amount) == event.amount && tx.phone == event.phone) { obj["bank_transaction_id"] = tx.bankTrExternalId; feeInCents = static_cast(tx.fee * 100); break; } } QJsonObject extraObj; extraObj["fee"] = feeInCents; obj["extra"] = extraObj; taskData = obj; } else { QJsonObject obj; obj["status"] = "completed"; taskData = obj; } // FETCH_TRANSACTIONS уже отправил данные потоково (per-tx) — // финальный агрегирующий результат не нужен. if (event.type != EventType::FetchTransactions) { sendTaskResult( eventTypeToString(event.type), event.externalId, event.bankName, taskData, event.id ); } } if (newStatus == EventStatus::Complete) { const DeviceInfo device = DeviceDAO::getDeviceById(event.deviceId); const QString adbSerial = device.adbSerial.isEmpty() ? event.deviceId : device.adbSerial; if (event.type == EventType::CreateTransaction) { const QString msg = QStringLiteral("[CREATE_TRANSACTION] bank=%1 material=%2 amount=%3 phone=%4") .arg(event.bankName, event.externalId, QString::number(event.amount, 'f', 2), event.phone); QByteArray screenshot; const QString tmpPath = QDir::tempPath() + "/arc_tx_" + QString::number(event.id); if (AdbUtils::takeScreenshot(event.deviceId, tmpPath)) { QFile f(tmpPath + ".png"); if (f.open(QIODevice::ReadOnly)) { screenshot = f.readAll(); f.close(); } f.remove(); } sendMonitoringLog("info", "task_success", msg, screenshot, event.id, /*toApiBase=*/true); // Декрементируем материальный/профильный баланс на сумму расхода + комиссию. // Между запусками FETCH_PROFILE это держит UI/локальный баланс // в актуальном виде; drift по входящим всё равно сбросит следующий re-parse. const MaterialInfo acc = MaterialDAO::getAccountById(event.accountId); if (acc.id != -1) { double fee = 0.0; const QList txList = TransactionDAO::getTransactionsWithinHours(event.accountId, 1); for (const auto &tx : txList) { if (!tx.bankTrExternalId.isEmpty() && qAbs(tx.amount) == event.amount && tx.phone == event.phone) { fee = tx.fee; break; } } const double delta = event.amount + fee; // Для Dushanbe все карты сидят на одном счёте — двигаем // все материалы (см. memory: project_dushanbe_single_account). if (acc.appCode == "dushanbe_city_bank") { if (!MaterialDAO::adjustAmountByApp(acc.deviceId, acc.appCode, delta)) { qWarning() << "[EventHandler] adjustAmountByApp failed for" << acc.deviceId << acc.appCode; } } else { if (!MaterialDAO::adjustAmount(acc.id, delta)) { qWarning() << "[EventHandler] adjustAmount failed for material" << acc.id; } } const BankProfileInfo app = BankProfileDAO::getApplication(acc.deviceId, acc.appCode); if (app.id != -1) { if (!BankProfileDAO::adjustAmount(app.id, delta)) { qWarning() << "[EventHandler] BankProfileDAO::adjustAmount failed for app" << app.id; } } qDebug() << "[EventHandler] balance decremented: material" << acc.id << "app" << acc.appCode << "delta=" << delta << "(amount=" << event.amount << "fee=" << fee << ")"; } } if (event.type == EventType::FetchTransactions) { const QString msg = QStringLiteral("[FETCH_TRANSACTIONS] bank=%1 material=%2") .arg(event.bankName, event.externalId); const QByteArray scr = m_screenshots.take(key); sendMonitoringLog("info", "task_success", msg, scr, event.id, /*toApiBase=*/true); } if (event.type == EventType::FetchProfile) { const QString msg = QStringLiteral("[FETCH_PROFILE] bank=%1 material=%2") .arg(event.bankName, event.externalId); const QByteArray scr = m_screenshots.take(key); sendMonitoringLog("info", "task_success", msg, scr, event.id, /*toApiBase=*/true); } } } if (const auto tm = m_timers.take(key)) { tm->stop(); tm->deleteLater(); } m_threads.remove(key); m_retryCount.remove(event.id); m_parsedTxCount.remove(key); // Убиваем банковское приложение на устройстве после завершения задачи { const MaterialInfo acc = MaterialDAO::getAccountById(event.accountId); const BankProfileInfo app = BankProfileDAO::getApplication(event.deviceId, acc.appCode); if (!app.package.isEmpty()) { const DeviceInfo device = DeviceDAO::getDeviceById(event.deviceId); const QString adbSerial = device.adbSerial.isEmpty() ? event.deviceId : device.adbSerial; AdbUtils::tryToKillApplication(adbSerial, app.package); qDebug() << "[EventHandler] Killed app" << app.package << "on device" << adbSerial; } } // Сразу пробуем запустить следующую задачу для этого устройства processNextTask(key); } void EventHandler::startThreadForTask(const MaterialInfo &account, const EventInfo &event) { const auto thr = new QThread(this); const QString key = account.deviceId; // Generic helper: подключает сигналы и переносит worker в поток auto setup = [&](auto *worker) { using W = std::remove_pointer_t; worker->moveToThread(thr); connect(thr, &QThread::started, worker, &W::start); connect(worker, &W::finished, thr, &QThread::quit); connect(worker, &W::finished, worker, &QObject::deleteLater); connect(thr, &QThread::finished, thr, &QObject::deleteLater); connect(worker, &W::finishedWithResult, this, [this, key, event](const QString &result) { updateEventThread(key, event, result); } ); }; const QString &appCode = account.appCode; // Передаём в скрипт канонический device_id из материала — скрипт сам // разрезолвит его в adb serial через DeviceDAO и использует ТОТ ЖЕ ключ // для поиска BankProfileInfo (чтобы PIN поднимался из правильной записи). const QString adbSerial = account.deviceId; // AutoConnection: worker emits из своего потока, лямбда выполняется в потоке // EventHandler (где live `this`), там же где take()/clear() — m_screenshots // не нужен mutex. auto connectScreenshot = [this, key](Ozon::CommonScript *w) { connect(w, &Ozon::CommonScript::screenshotReady, this, [this, key](const QByteArray &s) { m_screenshots[key] = s; }); }; auto connectScreenshotDushanbe = [this, key](Dushanbe::CommonScript *w) { connect(w, &Dushanbe::CommonScript::screenshotReady, this, [this, key](const QByteArray &s) { m_screenshots[key] = s; }); }; if (event.type == EventType::FetchProfile) { if (appCode == "ozon") { auto *w = new Ozon::LoginAndCheckAccountsScript(adbSerial, appCode, nullptr, true); setup(w); connectScreenshot(w); } else if (appCode == "black") setup(new Black::LoginAndCheckAccountsScript(adbSerial, appCode)); else if (appCode == "dushanbe_city_bank") { auto *w = new Dushanbe::LoginAndCheckAccountsScript(adbSerial, appCode); setup(w); connectScreenshotDushanbe(w); } else qWarning() << "[EventHandler] FETCH_PROFILE: unknown appCode:" << appCode; } if (event.type == EventType::FetchTransactions) { // body теперь — массив транзакций для поиска: [{amount, created_at}, ...]. // Для обратной совместимости поддерживаем и старый формат — одиночный // объект {amount, created_at} (оборачиваем в массив из одного элемента). const QJsonObject taskJson = QJsonDocument::fromJson(event.json.toUtf8()).object(); QJsonArray bodyArr = taskJson["body"].toArray(); if (bodyArr.isEmpty() && taskJson["body"].isObject()) bodyArr.append(taskJson["body"]); QList> targets; for (const auto &v : bodyArr) { const QJsonObject item = v.toObject(); const double searchAmount = item["amount"].toDouble() / 100.0; // копейки → рубли (с учётом знака) QDateTime searchTime = QDateTime::fromString(item["created_at"].toString(), Qt::ISODate); // Backend контракт — UTC ISO с суффиксом 'Z'. Если 'Z'/смещение // отсутствует — Qt парсит как LocalTime, что даёт off-by-one для // вечерних транзакций (см. task_dump (26)). Жёстко интерпретируем как UTC. if (searchTime.isValid() && searchTime.timeSpec() == Qt::LocalTime) { qWarning() << "[EventHandler] FETCH_TRANSACTIONS: created_at без TZ-маркера" << item["created_at"].toString() << "— интерпретирую как UTC"; searchTime.setTimeZone(QTimeZone::utc()); } targets.append({searchAmount, searchTime}); } qDebug() << "[EventHandler] FETCH_TRANSACTIONS: searching" << targets.size() << "transaction(s)"; // Потоковая отправка: каждую найденную транзакцию шлём на сервер сразу // отдельным task_result (updateStatus=false — статус события финализирует // updateEventThread). m_parsedTxCount считает отправленные, чтобы не // ретраить уже частично отработанную задачу. auto streamParsedTx = [this, key, event](const QJsonObject &tx) { m_parsedTxCount[key] = m_parsedTxCount.value(key, 0) + 1; QJsonArray arr; arr.append(tx); sendTaskResult(eventTypeToString(EventType::FetchTransactions), event.externalId, event.bankName, arr, event.id, /*updateStatus=*/false); }; if (appCode == "ozon") { auto *w = new Ozon::GetLastTransactionsScript(adbSerial, appCode, 0, targets, account.materialId); setup(w); connectScreenshot(w); connect(w, &Ozon::GetLastTransactionsScript::transactionParsed, this, streamParsedTx); } else if (appCode == "black") setup(new Black::GetLastTransactionsScript(adbSerial, appCode)); else if (appCode == "dushanbe_city_bank") { auto *w = new Dushanbe::GetLastTransactionsScript(adbSerial, appCode, targets, account.materialId); setup(w); connectScreenshotDushanbe(w); connect(w, &Dushanbe::GetLastTransactionsScript::transactionParsed, this, streamParsedTx); } else qWarning() << "[EventHandler] FETCH_TRANSACTIONS: unknown appCode:" << appCode; } if (event.type == EventType::CreateTransaction) { const QString recipientBank = event.comment; // резолвленное имя банка-получателя // Определяем тип транзакции из JSON задачи const QJsonObject taskJson = QJsonDocument::fromJson(event.json.toUtf8()).object(); const QString txType = taskJson["body"].toObject()["type"].toString(); if (txType == "bank") { // Перевод по номеру карты if (appCode == "ozon") { auto *w = new Ozon::PayByCardScript(account, event.phone, event.amount); setup(w); connectScreenshot(w); } else if (appCode == "black") setup(new Black::PayByCardScript(account, event.phone, event.amount)); else if (appCode == "dushanbe_city_bank") { auto *w = new Dushanbe::PayByCardScript(account, event.phone, event.amount); setup(w); connectScreenshotDushanbe(w); } else qWarning() << "[EventHandler] CREATE_TRANSACTION bank: unknown appCode:" << appCode; } else { // Перевод по номеру телефона (top-up) if (appCode == "ozon") { auto *w = new Ozon::PayByPhoneScript(account, event.phone, recipientBank, event.amount); setup(w); connectScreenshot(w); } else if (appCode == "dushanbe_city_bank") { auto *w = new Dushanbe::PayByPhoneScript(account, event.phone, event.amount); setup(w); connectScreenshotDushanbe(w); } else qWarning() << "[EventHandler] CREATE_TRANSACTION: unknown appCode:" << appCode; } } // Watchdog-таймер — убиваем зависший поток. // FETCH_TRANSACTIONS листает историю транзакций и не успевает за 5 минут // на длинных списках — даём ему 10 минут. Остальным задачам — 5 минут. const int timeoutMin = (event.type == EventType::FetchTransactions) ? 10 : 5; const QString timeoutMsg = QString("Script timeout (%1 min)").arg(timeoutMin); const auto timer = new QTimer(this); timer->setSingleShot(true); connect(timer, &QTimer::timeout, this, [this, key, event, timeoutMsg]() { qWarning() << "[EventHandler] Timeout for key" << key << "- killing stuck thread."; if (QThread *thread = m_threads.value(key); thread && thread->isRunning()) { thread->requestInterruption(); if (!thread->wait(15000)) { qWarning() << "[EventHandler] Thread still running after 15s for key" << key << "— abandoning (no terminate to avoid heap corruption)"; } } if (const auto tm = m_timers.take(key)) { tm->deleteLater(); } m_threads.remove(key); // Поиск транзакции: таймаут тоже ретраим (ещё до 2 раз), но только если // ещё ни одной транзакции не отправлено (иначе ретрай переотправит дубли). // Ретраи отключены: при таймауте сразу репортим как Error без повторов. // if (event.type == EventType::FetchTransactions // && m_parsedTxCount.value(key, 0) == 0 // && tryRetry(key, event, "Script timeout (5 min)")) { // return; // } // Помечаем событие как ошибку EventDAO::updateEvent(event.id, EventStatus::Error, timeoutMsg); const QString logMsg = QString("[%1] bank=%2 material=%3: Script timeout") .arg(eventTypeToString(event.type), event.bankName, event.externalId); GeneralLogDAO::insert("CRITICAL", "EventHandler", logMsg); sendTaskError(eventTypeToString(event.type), event.externalId, event.bankName, timeoutMsg, {}, event.id, event.transactionId, event.accountId); EventDAO::markSentToServer(event.id); // Убиваем банковское приложение на устройстве после таймаута { const MaterialInfo acc = MaterialDAO::getAccountById(event.accountId); const BankProfileInfo app = BankProfileDAO::getApplication(event.deviceId, acc.appCode); if (!app.package.isEmpty()) { const DeviceInfo device = DeviceDAO::getDeviceById(event.deviceId); const QString adbSerial = device.adbSerial.isEmpty() ? event.deviceId : device.adbSerial; AdbUtils::tryToKillApplication(adbSerial, app.package); qDebug() << "[EventHandler] Killed app" << app.package << "on device" << adbSerial << "(timeout)"; } } processNextTask(key); }); m_threads[key] = thr; m_timers[key] = timer; timer->start(timeoutMin * 60 * 1000); thr->start(); } void EventHandler::processNextTask(const QString &deviceId) { if (!m_running) return; if (const QThread *thr = m_threads.value(deviceId, nullptr); thr && thr->isRunning()) return; while (true) { const EventInfo event = EventDAO::getFirstWaitEventByDeviceId(deviceId); if (event.id == -1) break; const qint64 waitSecs = event.timestamp.secsTo(QDateTime::currentDateTimeUtc()); if (waitSecs > 5 * 60) { qWarning() << "[EventHandler] Task expired after" << waitSecs << "s, event id:" << event.id; EventDAO::updateEvent(event.id, EventStatus::Error, "Task expired: waited " + QString::number(waitSecs) + "s"); sendTaskError(eventTypeToString(event.type), event.externalId, event.bankName, "Task expired: not started within 5 minutes", {}, event.id, event.transactionId, event.accountId); continue; } const MaterialInfo account = MaterialDAO::getAccountById(event.accountId); const BankProfileInfo app = BankProfileDAO::getApplication(deviceId, account.appCode); if (app.status != "active") { qWarning() << "[EventHandler] Bank profile disabled for" << account.appCode << "device:" << deviceId; EventDAO::updateEvent(event.id, EventStatus::Error, "Bank profile is disabled"); sendTaskError(eventTypeToString(event.type), event.externalId, event.bankName, "Bank profile is disabled: " + account.appCode, {}, event.id, event.transactionId, event.accountId); continue; } if (!EventDAO::updateEvent(event.id, EventStatus::InProgress, "")) { qCritical() << "[EventHandler] Failed to set InProgress, event id:" << event.id; break; } startThreadForTask(account, event); break; } } void EventHandler::clearAll() { // Параллельная остановка: сначала всем requestInterruption + quit, // потом ждём каждый. terminate() намеренно НЕ используем — он опасен // (heap corruption, утечки DB-локов); если поток не успел остановиться, // abandon'им — ОС зачистит ресурсы при выходе процесса. Этот же подход // применяет watchdog в startThreadForTask. for (QThread *thread : m_threads) { if (thread->isRunning()) { thread->requestInterruption(); thread->quit(); } } for (QThread *thread : m_threads) { if (thread->isRunning() && !thread->wait(5000)) { qWarning() << "[EventHandler] Thread did not stop in 5s on shutdown," << "abandoning (no terminate to avoid heap corruption)"; continue; // не deleteLater для всё ещё работающего QThread } thread->deleteLater(); } m_threads.clear(); for (QTimer *timer : m_timers) { timer->stop(); timer->deleteLater(); } m_timers.clear(); m_screenshots.clear(); m_parsedTxCount.clear(); } void EventHandler::stop() { m_running = false; if (m_pollTimer) m_pollTimer->stop(); clearAll(); }