diff --git a/api/src/main/java/telegram/files/Config.java b/api/src/main/java/telegram/files/Config.java index eca8b80..7938289 100644 --- a/api/src/main/java/telegram/files/Config.java +++ b/api/src/main/java/telegram/files/Config.java @@ -34,6 +34,8 @@ public class Config { public static final String DB_NAME = System.getenv("DB_NAME"); + public static final boolean DB_NEED_CREATE = Convert.toBool(System.getenv("DB_NEED_CREATE"), false); + public static final String LOG_PATH = APP_ROOT + File.separator + "logs"; public static final String TELEGRAM_ROOT = APP_ROOT + File.separator + "account"; diff --git a/api/src/main/java/telegram/files/DataVerticle.java b/api/src/main/java/telegram/files/DataVerticle.java index 5bfcede..1318467 100644 --- a/api/src/main/java/telegram/files/DataVerticle.java +++ b/api/src/main/java/telegram/files/DataVerticle.java @@ -12,9 +12,7 @@ import io.vertx.core.impl.NoStackTraceException; import io.vertx.jdbcclient.JDBCConnectOptions; import io.vertx.jdbcclient.JDBCPool; import io.vertx.mysqlclient.MySQLBuilder; -import io.vertx.mysqlclient.MySQLConnectOptions; import io.vertx.pgclient.PgBuilder; -import io.vertx.pgclient.PgConnectOptions; import io.vertx.sqlclient.Pool; import io.vertx.sqlclient.PoolOptions; import io.vertx.sqlclient.SqlClient; @@ -45,22 +43,24 @@ public class DataVerticle extends AbstractVerticle { private static SqlConnectOptions sqlConnectOptions; + public static final List definitions; + static { - if (Config.isPostgres()) { - sqlConnectOptions = new PgConnectOptions() - .setPort(Config.DB_PORT) - .setHost(Config.DB_HOST) - .setDatabase(Config.DB_NAME) - .setUser(Config.DB_USER) - .setPassword(Config.DB_PASSWORD); - } else if (Config.isMysql()) { - sqlConnectOptions = new MySQLConnectOptions() + if (Config.isPostgres() || Config.isMysql()) { + sqlConnectOptions = new SqlConnectOptions() .setPort(Config.DB_PORT) .setHost(Config.DB_HOST) .setDatabase(Config.DB_NAME) .setUser(Config.DB_USER) .setPassword(Config.DB_PASSWORD); } + + definitions = List.of( + new SettingRecord.SettingRecordDefinition(), + new TelegramRecord.TelegramRecordDefinition(), + new FileRecord.FileRecordDefinition(), + new StatisticRecord.StatisticRecordDefinition() + ); } public void start(Promise stopPromise) { @@ -69,12 +69,6 @@ public class DataVerticle extends AbstractVerticle { telegramRepository = new TelegramRepositoryImpl(pool); fileRepository = new FileRepositoryImpl(pool); statisticRepository = new StatisticRepositoryImpl(pool); - List definitions = List.of( - new SettingRecord.SettingRecordDefinition(), - new TelegramRecord.TelegramRecordDefinition(), - new FileRecord.FileRecordDefinition(), - new StatisticRecord.StatisticRecordDefinition() - ); isCompletelyNewInitialization() .compose(isNew -> Future.all(definitions.stream().map(d -> d.createTable(pool)).toList()).map(isNew)) .compose(isNew -> settingRepository.getByKey(SettingKey.version).map(version -> Tuple.tuple(isNew, version))) @@ -152,50 +146,85 @@ public class DataVerticle extends AbstractVerticle { } private Future isCompletelyNewInitializationForPostgres() { - SqlClient sqlClient = createSqlClient(vertx, createDefaultOptions()); + SqlClient sqlClient; + String query; + if (Config.DB_NEED_CREATE) { + query = """ + SELECT 1 FROM pg_database WHERE datname = '%s' + """.formatted(Config.DB_NAME); + sqlClient = createSqlClient(vertx, createDefaultOptions()); + } else { + query = """ + SELECT table_name + FROM information_schema.tables + WHERE table_schema = 'public' + ORDER BY table_name; + """; + sqlClient = pool; + } - return sqlClient.query(""" - SELECT 1 FROM pg_database WHERE datname = '%s' - """.formatted(Config.DB_NAME)) + return sqlClient.query(query) .execute() .map(rs -> rs.size() == 0) .compose(isNew -> { if (isNew) { - return sqlClient.query(""" + return Config.DB_NEED_CREATE ? sqlClient.query(""" CREATE DATABASE "%s" """.formatted(Config.DB_NAME)) .execute() - .map(true); + .map(true) : Future.succeededFuture(true); } else { return Future.succeededFuture(false); } }) - .eventually(() -> sqlClient.close()) + .eventually(() -> { + if (Config.DB_NEED_CREATE) + return sqlClient.close(); + return Future.succeededFuture(); + }) .onFailure(err -> log.error("Failed to check database initialization: %s".formatted(err.getMessage()))); } private Future isCompletelyNewInitializationForMySQL() { - SqlClient sqlClient = createSqlClient(vertx, createDefaultOptions()); + SqlClient sqlClient; + String query; + if (Config.DB_NEED_CREATE) { + query = """ + SELECT SCHEMA_NAME + FROM INFORMATION_SCHEMA.SCHEMATA + WHERE SCHEMA_NAME = '%s' + """.formatted(Config.DB_NAME); + sqlClient = createSqlClient(vertx, createDefaultOptions()); + } else { + query = """ + SELECT table_name + FROM information_schema.tables + WHERE table_schema = '%s' + ORDER BY table_name; + """.formatted(Config.DB_NAME); + sqlClient = pool; + } - return sqlClient.query(""" - SELECT SCHEMA_NAME - FROM INFORMATION_SCHEMA.SCHEMATA - WHERE SCHEMA_NAME = '%s' - """.formatted(Config.DB_NAME)) + return sqlClient.query(query) .execute() .map(rs -> rs.size() == 0) .compose(isNew -> { if (isNew) { - return sqlClient.query(""" + return Config.DB_NEED_CREATE ? sqlClient.query(""" CREATE DATABASE `%s` collate utf8mb4_bin; """.formatted(Config.DB_NAME)) .execute() - .map(true); + .map(true) : + Future.succeededFuture(true); } else { return Future.succeededFuture(false); } }) - .eventually(() -> sqlClient.close()) + .eventually(() -> { + if (Config.DB_NEED_CREATE) + return sqlClient.close(); + return Future.succeededFuture(); + }) .onFailure(err -> log.error("Failed to check database initialization: %s".formatted(err.getMessage()))); } diff --git a/api/src/main/java/telegram/files/TelegramVerticle.java b/api/src/main/java/telegram/files/TelegramVerticle.java index 27d87a3..a3509eb 100644 --- a/api/src/main/java/telegram/files/TelegramVerticle.java +++ b/api/src/main/java/telegram/files/TelegramVerticle.java @@ -810,6 +810,10 @@ public class TelegramVerticle extends AbstractVerticle { private void onMessageReceived(TdApi.Message message) { log.trace("[%s] Receive message: %s".formatted(getRootId(), message)); + if (this.telegramRecord == null) { + log.trace("[%s] Telegram record is null, can't handle message".formatted(getRootId())); + return; + } 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/impl/FileRepositoryImpl.java b/api/src/main/java/telegram/files/repository/impl/FileRepositoryImpl.java index ed240ae..7c5f6df 100644 --- a/api/src/main/java/telegram/files/repository/impl/FileRepositoryImpl.java +++ b/api/src/main/java/telegram/files/repository/impl/FileRepositoryImpl.java @@ -17,6 +17,7 @@ import io.vertx.sqlclient.SqlResult; import io.vertx.sqlclient.templates.SqlTemplate; import org.jooq.lambda.tuple.Tuple; import org.jooq.lambda.tuple.Tuple3; +import telegram.files.Config; import telegram.files.MessyUtils; import telegram.files.repository.FileRecord; import telegram.files.repository.FileRepository; @@ -313,9 +314,10 @@ public class FileRepositoryImpl extends AbstractSqlRepository implements FileRep @Override public Future getCompletedRangeStatistics(long telegramId, long startTime, long endTime, int timeRange) { - return SqlTemplate - .forQuery(sqlClient, """ - SELECT strftime( + String query; + if (Config.isSqlite()) { + query = """ + SELECT strftime( CASE WHEN #{timeRange} = 1 THEN '%Y-%m-%d %H:%M' WHEN #{timeRange} = 2 THEN '%Y-%m-%d %H:00' @@ -332,7 +334,48 @@ public class FileRepositoryImpl extends AbstractSqlRepository implements FileRep AND completion_date <= #{endTime} GROUP BY time ORDER BY time; - """) + """; + } else if (Config.isPostgres()) { + query = """ + SELECT TO_CHAR( + TO_TIMESTAMP(completion_date / 1000), + CASE + WHEN #{timeRange} = 1 THEN 'YYYY-MM-DD HH24:MI' + WHEN #{timeRange} = 2 THEN 'YYYY-MM-DD HH24:00' + WHEN #{timeRange} IN (3, 4) THEN 'YYYY-MM-DD' + END + ) AS time, + COUNT(*) AS total + FROM file_record + WHERE telegram_id = #{telegramId} + AND completion_date IS NOT NULL + AND completion_date >= #{startTime} + AND completion_date <= #{endTime} + GROUP BY time + ORDER BY time; + """; + } else { + query = """ + SELECT DATE_FORMAT( + FROM_UNIXTIME(completion_date / 1000), + CASE + WHEN #{timeRange} = 1 THEN '%Y-%m-%d %H:%i' + WHEN #{timeRange} = 2 THEN '%Y-%m-%d %H:00' + WHEN #{timeRange} IN (3, 4) THEN '%Y-%m-%d' + END + ) AS time, + COUNT(*) AS total + FROM file_record + WHERE telegram_id = #{telegramId} + AND completion_date IS NOT NULL + AND completion_date >= #{startTime} + AND completion_date <= #{endTime} + GROUP BY time + ORDER BY time; + """; + } + return SqlTemplate + .forQuery(sqlClient, query) .mapTo(row -> new JsonObject() .put("time", row.getString("time")) .put("total", row.getInteger("total")) diff --git a/api/src/main/java/telegram/files/repository/impl/StatisticRepositoryImpl.java b/api/src/main/java/telegram/files/repository/impl/StatisticRepositoryImpl.java index 3b2e149..878e944 100644 --- a/api/src/main/java/telegram/files/repository/impl/StatisticRepositoryImpl.java +++ b/api/src/main/java/telegram/files/repository/impl/StatisticRepositoryImpl.java @@ -1,6 +1,7 @@ package telegram.files.repository.impl; import cn.hutool.core.collection.IterUtil; +import cn.hutool.core.convert.Convert; import cn.hutool.log.Log; import cn.hutool.log.LogFactory; import io.vertx.core.Future; @@ -54,7 +55,7 @@ public class StatisticRepositoryImpl extends AbstractSqlRepository implements St .mapTo(StatisticRecord.ROW_MAPPER) .execute(Map.of( "type", type.name(), - "relatedId", relatedId, + "relatedId", Convert.toStr(relatedId), "startTime", startTime, "endTime", endTime )) diff --git a/api/src/test/java/telegram/files/DataVerticleMigrationTest.java b/api/src/test/java/telegram/files/DataVerticleMigrationTest.java index c7bb5c8..c1fc136 100644 --- a/api/src/test/java/telegram/files/DataVerticleMigrationTest.java +++ b/api/src/test/java/telegram/files/DataVerticleMigrationTest.java @@ -113,7 +113,7 @@ public class DataVerticleMigrationTest { .compose(conn -> conn.query(getTablesQuery()).execute() .compose(result -> { testContext.verify(() -> { - Assertions.assertEquals(4, result.size()); + Assertions.assertEquals(DataVerticle.definitions.size(), result.size()); }); return conn.close(); }));