diff --git a/.env.example b/.env.example index 1776b75..02224e6 100644 --- a/.env.example +++ b/.env.example @@ -1,4 +1,5 @@ - +# Telegram files log level, ALL < FINEST < FINER < FINE < CONFIG < INFO < WARNING < SEVERE < OFF +LOG_LEVEL=INFO # The root directory of the application APP_ENV=dev APP_ROOT= @@ -6,3 +7,5 @@ TDLIB_PATH= TELEGRAM_API_ID= TELEGRAM_API_HASH= +# Value 0 corresponds to fatal errors, value 1 corresponds to errors, value 2 corresponds to warnings and debug warnings, value 3 corresponds to informational, value 4 corresponds to debug, value 5 corresponds to verbose debug, value greater than 5 and up to 1023 can be used to enable even more logging. +TELEGRAM_LOG_LEVEL=0 diff --git a/api/src/main/java/telegram/files/AutoDownloadVerticle.java b/api/src/main/java/telegram/files/AutoDownloadVerticle.java index 01385ef..dc43307 100644 --- a/api/src/main/java/telegram/files/AutoDownloadVerticle.java +++ b/api/src/main/java/telegram/files/AutoDownloadVerticle.java @@ -11,6 +11,7 @@ import io.vertx.core.Promise; import io.vertx.core.json.Json; import io.vertx.core.json.JsonObject; import org.drinkless.tdlib.TdApi; +import telegram.files.repository.FileRecord; import telegram.files.repository.SettingAutoRecords; import telegram.files.repository.SettingKey; @@ -27,7 +28,7 @@ public class AutoDownloadVerticle extends AbstractVerticle { private static final int HISTORY_SCAN_INTERVAL = 2 * 60 * 1000; - private static final int MAX_HISTORY_SCAN_TIME = 20 * 1000; + private static final int MAX_HISTORY_SCAN_TIME = 10 * 1000; private static final int MAX_WAITING_LENGTH = 30; @@ -107,7 +108,7 @@ public class AutoDownloadVerticle extends AbstractVerticle { this.limit = Convert.toInt(message.body()); }); vertx.eventBus().consumer(EventEnum.MESSAGE_RECEIVED.address(), message -> { - log.debug("Auto download message received: %s".formatted(message.body())); + log.trace("Auto download message received: %s".formatted(message.body())); this.onNewMessage((JsonObject) message.body()); }); return Future.succeededFuture(); @@ -128,7 +129,9 @@ public class AutoDownloadVerticle extends AbstractVerticle { } private void addHistoryMessage(SettingAutoRecords.Item auto, long currentTimeMillis) { + log.debug("Start scan history! TelegramId: %d ChatId: %d FileType: %s".formatted(auto.telegramId, auto.chatId, auto.nextFileType)); if (System.currentTimeMillis() - currentTimeMillis > MAX_HISTORY_SCAN_TIME || isExceedLimit(auto.telegramId)) { + log.debug("Scan history end! TelegramId: %d ChatId: %d".formatted(auto.telegramId, auto.chatId)); return; } if (StrUtil.isBlank(auto.nextFileType)) { @@ -140,36 +143,39 @@ public class AutoDownloadVerticle extends AbstractVerticle { searchChatMessages.fromMessageId = auto.nextFromMessageId; searchChatMessages.limit = Math.min(MAX_WAITING_LENGTH, 100); searchChatMessages.filter = TdApiHelp.getSearchMessagesFilter(auto.nextFileType); - telegramVerticle.execute(searchChatMessages) - .onSuccess(foundChatMessages -> { - if (foundChatMessages.messages.length == 0) { - int nextTypeIndex = FILE_TYPE_ORDER.indexOf(auto.nextFileType) + 1; - if (nextTypeIndex < FILE_TYPE_ORDER.size()) { - String originalType = auto.nextFileType; - auto.nextFileType = FILE_TYPE_ORDER.get(nextTypeIndex); - auto.nextFromMessageId = 0; - log.debug("%s No more %s files found! Switch to %s".formatted(auto.uniqueKey(), originalType, auto.nextFileType)); + TdApi.FoundChatMessages foundChatMessages = Future.await(telegramVerticle.execute(searchChatMessages) + .onFailure(r -> log.error("Search chat messages failed! TelegramId: %d ChatId: %d".formatted(auto.telegramId, auto.chatId), r)) + ); + if (foundChatMessages == null) { + return; + } + if (foundChatMessages.messages.length == 0) { + int nextTypeIndex = FILE_TYPE_ORDER.indexOf(auto.nextFileType) + 1; + if (nextTypeIndex < FILE_TYPE_ORDER.size()) { + String originalType = auto.nextFileType; + auto.nextFileType = FILE_TYPE_ORDER.get(nextTypeIndex); + auto.nextFromMessageId = 0; + log.debug("%s No more %s files found! Switch to %s".formatted(auto.uniqueKey(), originalType, auto.nextFileType)); + addHistoryMessage(auto, currentTimeMillis); + } else { + log.debug("%s No more history files found! TelegramId: %d ChatId: %d".formatted(auto.uniqueKey(), auto.telegramId, auto.chatId)); + } + } else { + DataVerticle.fileRepository.getFilesByUniqueId(TdApiHelp.getFileUniqueIds(Arrays.asList(foundChatMessages.messages))) + .onSuccess(existFiles -> { + List messages = Stream.of(foundChatMessages.messages) + .filter(message -> !existFiles.containsKey(TdApiHelp.getFileUniqueId(message))) + .toList(); + if (CollUtil.isEmpty(messages)) { + auto.nextFromMessageId = foundChatMessages.nextFromMessageId; addHistoryMessage(auto, currentTimeMillis); + } else if (addWaitingDownloadMessages(auto.telegramId, messages, false)) { + auto.nextFromMessageId = foundChatMessages.nextFromMessageId; } else { - log.debug("%s No more history files found! TelegramId: %d ChatId: %d".formatted(auto.uniqueKey(), auto.telegramId, auto.chatId)); + addHistoryMessage(auto, currentTimeMillis); } - } else { - DataVerticle.fileRepository.getFilesByUniqueId(TdApiHelp.getFileUniqueIds(Arrays.asList(foundChatMessages.messages))) - .onSuccess(existFiles -> { - List messages = Stream.of(foundChatMessages.messages) - .filter(message -> !existFiles.containsKey(TdApiHelp.getFileUniqueId(message))) - .toList(); - if (CollUtil.isEmpty(messages)) { - auto.nextFromMessageId = foundChatMessages.nextFromMessageId; - addHistoryMessage(auto, currentTimeMillis); - } else if (addWaitingDownloadMessages(auto.telegramId, messages, false)) { - auto.nextFromMessageId = foundChatMessages.nextFromMessageId; - } else { - addHistoryMessage(auto, currentTimeMillis); - } - }); - } - }); + }); + } } private boolean isExceedLimit(long telegramId) { @@ -178,11 +184,8 @@ public class AutoDownloadVerticle extends AbstractVerticle { } private int getSurplusSize(long telegramId) { - TelegramVerticle telegramVerticle = this.getTelegramVerticle(telegramId); - Integer downloading = telegramVerticle.getDownloadStatistics() - .map(statistics -> statistics.getInteger("downloading")) - .result(); - return downloading == null ? limit : Math.min(0, limit - downloading); + Integer downloading = Future.await(DataVerticle.fileRepository.countByStatus(telegramId, FileRecord.DownloadStatus.downloading)); + return downloading == null ? limit : Math.max(0, limit - downloading); } private boolean addWaitingDownloadMessages(long telegramId, List messages, boolean force) { diff --git a/api/src/main/java/telegram/files/Config.java b/api/src/main/java/telegram/files/Config.java index f7b26d4..6966631 100644 --- a/api/src/main/java/telegram/files/Config.java +++ b/api/src/main/java/telegram/files/Config.java @@ -5,8 +5,12 @@ import cn.hutool.core.io.FileUtil; import cn.hutool.core.util.StrUtil; import java.io.File; +import java.util.logging.Level; +import java.util.logging.Logger; public class Config { + public static final String LOG_LEVEL = StrUtil.blankToDefault(System.getenv("LOG_LEVEL"), "INFO"); + public static final String APP_ENV = StrUtil.blankToDefault(System.getenv("APP_ENV"), "prod"); public static final String APP_ROOT = System.getenv("APP_ROOT"); @@ -17,6 +21,10 @@ public class Config { public static final String TELEGRAM_API_HASH = System.getenv("TELEGRAM_API_HASH"); + public static final int TELEGRAM_LOG_LEVEL = Convert.toInt(System.getenv("TELEGRAM_LOG_LEVEL"), 0); + + private static Level logLevel; + static { if (APP_ENV == null) { throw new RuntimeException("APP_ENV is not set"); @@ -31,6 +39,15 @@ public class Config { throw new RuntimeException("TELEGRAM_API_HASH is not set"); } + try { + logLevel = Level.parse(Config.LOG_LEVEL); + System.out.println("Setting telegram.files log level to " + logLevel); + } catch (IllegalArgumentException e) { + System.err.println("Invalid log level [" + Config.LOG_LEVEL + "], using default INFO."); + } + + Logger.getLogger("telegram.files").setLevel(logLevel); + if (!FileUtil.exist(APP_ROOT)) { FileUtil.mkdir(APP_ROOT); } diff --git a/api/src/main/java/telegram/files/HttpVerticle.java b/api/src/main/java/telegram/files/HttpVerticle.java index fbca5f2..2c9f5ac 100644 --- a/api/src/main/java/telegram/files/HttpVerticle.java +++ b/api/src/main/java/telegram/files/HttpVerticle.java @@ -196,7 +196,9 @@ public class HttpVerticle extends AbstractVerticle { } public Future initAutoDownloadVerticle() { - return vertx.deployVerticle(new AutoDownloadVerticle(), new DeploymentOptions().setThreadingModel(ThreadingModel.WORKER)) + return vertx.deployVerticle(new AutoDownloadVerticle(), new DeploymentOptions() + .setThreadingModel(ThreadingModel.VIRTUAL_THREAD) + ) .mapEmpty(); } diff --git a/api/src/main/java/telegram/files/TelegramUpdateHandler.java b/api/src/main/java/telegram/files/TelegramUpdateHandler.java index 1d25146..5306e98 100644 --- a/api/src/main/java/telegram/files/TelegramUpdateHandler.java +++ b/api/src/main/java/telegram/files/TelegramUpdateHandler.java @@ -30,7 +30,7 @@ public class TelegramUpdateHandler implements Client.ResultHandler { if (onFileUpdated != null) onFileUpdated.accept((TdApi.UpdateFile) object); case TdApi.UpdateFileDownload.CONSTRUCTOR: - log.debug("File download update: %s".formatted(object)); + log.trace("File download update: %s".formatted(object)); break; case TdApi.UpdateFileDownloads.CONSTRUCTOR: if (onFileDownloadsUpdated != null) diff --git a/api/src/main/java/telegram/files/TelegramVerticle.java b/api/src/main/java/telegram/files/TelegramVerticle.java index e3a7b5d..7c189c3 100644 --- a/api/src/main/java/telegram/files/TelegramVerticle.java +++ b/api/src/main/java/telegram/files/TelegramVerticle.java @@ -56,7 +56,7 @@ public class TelegramVerticle extends AbstractVerticle { Client.setLogMessageHandler(0, new LogMessageHandler()); try { - Client.execute(new TdApi.SetLogVerbosityLevel(0)); + Client.execute(new TdApi.SetLogVerbosityLevel(Config.TELEGRAM_LOG_LEVEL)); Client.execute(new TdApi.SetLogStream(new TdApi.LogStreamFile("tdlib.log", 1 << 27, false))); } catch (Client.ExecutionException error) { throw new IOError(new IOException("Write access to the current directory is required")); @@ -600,7 +600,7 @@ public class TelegramVerticle extends AbstractVerticle { request.systemLanguageCode = "en"; request.deviceModel = "Telegram Files"; request.applicationVersion = Start.VERSION; - log.debug("[%s] Send SetTdlibParameters: %s".formatted(getRootId(), request)); + log.trace("[%s] Send SetTdlibParameters: %s".formatted(getRootId(), request)); client.send(request, this::handleAuthorizationResult); break; @@ -650,7 +650,7 @@ public class TelegramVerticle extends AbstractVerticle { } private void onFileUpdated(TdApi.UpdateFile updateFile) { - log.debug("[%s] Receive file update: %s".formatted(getRootId(), updateFile)); + log.trace("[%s] Receive file update: %s".formatted(getRootId(), updateFile)); TdApi.File file = updateFile.file; if (file != null) { String localPath = null; @@ -675,12 +675,12 @@ public class TelegramVerticle extends AbstractVerticle { } private void onFileDownloadsUpdated(TdApi.UpdateFileDownloads updateFileDownloads) { - log.debug("[%s] Receive file downloads update: %s".formatted(getRootId(), updateFileDownloads)); + log.trace("[%s] Receive file downloads update: %s".formatted(getRootId(), updateFileDownloads)); sendHttpEvent(EventPayload.build(EventPayload.TYPE_FILE_DOWNLOAD, updateFileDownloads)); } private void onMessageReceived(TdApi.Message message) { - log.debug("[%s] Receive message: %s".formatted(getRootId(), message)); + log.trace("[%s] Receive message: %s".formatted(getRootId(), message)); vertx.eventBus().publish(EventEnum.MESSAGE_RECEIVED.address(), JsonObject.of() .put("telegramId", telegramRecord.id()) .put("chatId", message.chatId) diff --git a/api/src/main/java/telegram/files/repository/FileRepository.java b/api/src/main/java/telegram/files/repository/FileRepository.java index d686133..c86aec7 100644 --- a/api/src/main/java/telegram/files/repository/FileRepository.java +++ b/api/src/main/java/telegram/files/repository/FileRepository.java @@ -25,6 +25,8 @@ public interface FileRepository { Future getDownloadStatistics(long telegramId); + Future countByStatus(long telegramId, FileRecord.DownloadStatus downloadStatus); + Future updateStatus(int fileId, String uniqueId, String localPath, diff --git a/api/src/main/java/telegram/files/repository/impl/FileRepositoryImpl.java b/api/src/main/java/telegram/files/repository/impl/FileRepositoryImpl.java index 2b0d180..774e146 100644 --- a/api/src/main/java/telegram/files/repository/impl/FileRepositoryImpl.java +++ b/api/src/main/java/telegram/files/repository/impl/FileRepositoryImpl.java @@ -42,7 +42,7 @@ public class FileRepositoryImpl implements FileRepository { .execute() .onComplete(r -> conn.close()) .onFailure(err -> log.error("Failed to create table file_record: %s".formatted(err.getMessage()))) - .onSuccess(ps -> log.debug("Successfully created table: file_record")) + .onSuccess(ps -> log.trace("Successfully created table: file_record")) ) .mapEmpty(); } @@ -61,7 +61,7 @@ public class FileRepositoryImpl implements FileRepository { .mapFrom(FileRecord.PARAM_MAPPER) .execute(fileRecord) .map(r -> fileRecord) - .onSuccess(r -> log.debug("Successfully created file record: %s".formatted(fileRecord.id())) + .onSuccess(r -> log.trace("Successfully created file record: %s".formatted(fileRecord.id())) ) .onFailure( err -> log.error("Failed to create file record: %s".formatted(err.getMessage())) @@ -234,6 +234,18 @@ public class FileRepositoryImpl implements FileRepository { .onFailure(err -> log.error("Failed to get download statistics: %s".formatted(err.getMessage()))); } + @Override + public Future countByStatus(long telegramId, FileRecord.DownloadStatus downloadStatus) { + return SqlTemplate + .forQuery(pool, """ + SELECT COUNT(*) FROM file_record WHERE telegram_id = #{telegramId} AND download_status = #{downloadStatus} + """) + .mapTo(rs -> rs.getInteger(0)) + .execute(Map.of("telegramId", telegramId, "downloadStatus", downloadStatus.name())) + .map(rs -> rs.size() > 0 ? rs.iterator().next() : 0) + .onFailure(err -> log.error("Failed to count file record: %s".formatted(err.getMessage()))); + } + @Override public Future updateStatus(int fileId, String uniqueId, String localPath, FileRecord.DownloadStatus downloadStatus) { if (StrUtil.isBlank(localPath) && downloadStatus == null) { diff --git a/api/src/main/java/telegram/files/repository/impl/SettingRepositoryImpl.java b/api/src/main/java/telegram/files/repository/impl/SettingRepositoryImpl.java index f30b9f7..055d5be 100644 --- a/api/src/main/java/telegram/files/repository/impl/SettingRepositoryImpl.java +++ b/api/src/main/java/telegram/files/repository/impl/SettingRepositoryImpl.java @@ -35,7 +35,7 @@ public class SettingRepositoryImpl implements SettingRepository { .execute() .onComplete(r -> conn.close()) .onFailure(err -> log.error("Failed to create table setting_record: %s".formatted(err.getMessage()))) - .onSuccess(ps -> log.debug("Successfully created table: setting_record")) + .onSuccess(ps -> log.trace("Successfully created table: setting_record")) ) .mapEmpty(); } @@ -50,7 +50,7 @@ public class SettingRepositoryImpl implements SettingRepository { .mapFrom(SettingRecord.PARAM_MAPPER) .execute(new SettingRecord(key, value)) .map(r -> new SettingRecord(key, value)) - .onSuccess(r -> log.debug("Successfully created or updated setting record: %s".formatted(key))) + .onSuccess(r -> log.trace("Successfully created or updated setting record: %s".formatted(key))) .onFailure( err -> log.error("Failed to create or update setting record: %s".formatted(err.getMessage())) ); @@ -74,7 +74,7 @@ public class SettingRepositoryImpl implements SettingRepository { .mapTo(SettingRecord.ROW_MAPPER) .execute(Collections.emptyMap()) .map(IterUtil::toList) - .onSuccess(r -> log.debug("Successfully fetched setting record for keys: " + keyStr)) + .onSuccess(r -> log.trace("Successfully fetched setting record for keys: " + keyStr)) .onFailure( err -> log.error("Failed to fetch setting record: %s".formatted(err.getMessage())) ); @@ -95,7 +95,7 @@ public class SettingRepositoryImpl implements SettingRepository { } return key.defaultValue == null ? null : (T) key.defaultValue; }) - .onSuccess(r -> log.debug("Successfully fetched setting record for key: " + key)) + .onSuccess(r -> log.trace("Successfully fetched setting record for key: " + key)) .onFailure( err -> log.error("Failed to fetch setting record: %s".formatted(err.getMessage())) ); diff --git a/api/src/main/java/telegram/files/repository/impl/TelegramRepositoryImpl.java b/api/src/main/java/telegram/files/repository/impl/TelegramRepositoryImpl.java index 897f3de..88c2477 100644 --- a/api/src/main/java/telegram/files/repository/impl/TelegramRepositoryImpl.java +++ b/api/src/main/java/telegram/files/repository/impl/TelegramRepositoryImpl.java @@ -34,7 +34,7 @@ public class TelegramRepositoryImpl implements TelegramRepository { .execute() .onComplete(r -> conn.close()) .onFailure(err -> log.error("Failed to create table telegram_record: %s".formatted(err.getMessage()))) - .onSuccess(ps -> log.debug("Successfully created table: telegram_record")) + .onSuccess(ps -> log.trace("Successfully created table: telegram_record")) ) .mapEmpty(); } @@ -51,7 +51,7 @@ public class TelegramRepositoryImpl implements TelegramRepository { .mapFrom(TelegramRecord.PARAM_MAPPER) .execute(telegramRecord) .map(r -> telegramRecord) - .onSuccess(r -> log.debug("Successfully created telegram record: %s".formatted(telegramRecord.id()))) + .onSuccess(r -> log.trace("Successfully created telegram record: %s".formatted(telegramRecord.id()))) .onFailure( err -> log.error("Failed to create telegram record: %s".formatted(err.getMessage())) ); diff --git a/api/src/main/resources/logging.properties b/api/src/main/resources/logging.properties index 6c7fc07..1aa9a69 100644 --- a/api/src/main/resources/logging.properties +++ b/api/src/main/resources/logging.properties @@ -5,7 +5,7 @@ handlers = java.util.logging.ConsoleHandler, java.util.logging.FileHandler # \u63A7\u5236\u53F0 Handler \u7684\u65E5\u5FD7\u7EA7\u522B -java.util.logging.ConsoleHandler.level = INFO +java.util.logging.ConsoleHandler.level = FINE # \u63A7\u5236\u53F0 Handler \u7684\u65E5\u5FD7\u683C\u5F0F\u5316\u5668 java.util.logging.ConsoleHandler.formatter = java.util.logging.SimpleFormatter diff --git a/api/src/test/java/telegram/files/DataVerticleTest.java b/api/src/test/java/telegram/files/DataVerticleTest.java index f76b2a4..d67a3f9 100644 --- a/api/src/test/java/telegram/files/DataVerticleTest.java +++ b/api/src/test/java/telegram/files/DataVerticleTest.java @@ -40,7 +40,7 @@ public class DataVerticleTest { @Test @DisplayName("Test Create telegram record") void createTelegramRecordTest(Vertx vertx, VertxTestContext testContext) { - TelegramRecord telegramRecord = new TelegramRecord(1, "test", "test"); + TelegramRecord telegramRecord = new TelegramRecord(1, "test", "test", null); DataVerticle.telegramRepository.create(telegramRecord) .compose(r -> DataVerticle.telegramRepository.getById(r.id())) .onComplete(testContext.succeeding(r -> testContext.verify(() -> { @@ -52,8 +52,8 @@ public class DataVerticleTest { @Test @DisplayName("Test Get all telegram record") void getAllTelegramRecordTest(Vertx vertx, VertxTestContext testContext) { - TelegramRecord telegramRecord = new TelegramRecord(1, "test", "test"); - TelegramRecord telegramRecord2 = new TelegramRecord(2, "test2", "test2"); + TelegramRecord telegramRecord = new TelegramRecord(1, "test", "test", null); + TelegramRecord telegramRecord2 = new TelegramRecord(2, "test2", "test2", null); DataVerticle.telegramRepository.create(telegramRecord) .compose(r -> DataVerticle.telegramRepository.create(telegramRecord2)) .compose(r -> DataVerticle.telegramRepository.getAll()) diff --git a/web/src/components/file-statistics.tsx b/web/src/components/file-statistics.tsx index 9b828fa..a560d86 100644 --- a/web/src/components/file-statistics.tsx +++ b/web/src/components/file-statistics.tsx @@ -8,6 +8,7 @@ import { File, FileText, Image, + LoaderPinwheel, Music, Network, PauseCircle, @@ -76,7 +77,10 @@ const FileStatistics: React.FC = ({ telegramId }) => { if (!data) { return (
- + Loading...
); @@ -121,7 +125,7 @@ const FileStatistics: React.FC = ({ telegramId }) => { ]; return ( -
+