danvics/telegram-files
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

🐛 fix: Adapt database differences.

Browse source
This commit is contained in:
jarvis2f 2025-03-06 23:00:00 +08:00
parent 0ed21b69f8
commit c6a4938edf
6 changed files with 118 additions and 39 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

2
api/src/main/java/telegram/files/Config.java
View file

@ -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";

95
api/src/main/java/telegram/files/DataVerticle.java
View file

@ -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<Definition> 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<Void> stopPromise) {
@ -69,12 +69,6 @@ public class DataVerticle extends AbstractVerticle {
telegramRepository = new TelegramRepositoryImpl(pool);
fileRepository = new FileRepositoryImpl(pool);
statisticRepository = new StatisticRepositoryImpl(pool);
List<Definition> 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.<Version>getByKey(SettingKey.version).map(version -> Tuple.tuple(isNew, version)))
@ -152,50 +146,85 @@ public class DataVerticle extends AbstractVerticle {
}
private Future<Boolean> 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<Boolean> 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())));
}

4
api/src/main/java/telegram/files/TelegramVerticle.java
View file

@ -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)

51
api/src/main/java/telegram/files/repository/impl/FileRepositoryImpl.java
View file

@ -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<JsonArray> 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"))

3
api/src/main/java/telegram/files/repository/impl/StatisticRepositoryImpl.java
View file

@ -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
))

2
api/src/test/java/telegram/files/DataVerticleMigrationTest.java
View file

@ -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();
}));