diff --git a/.env.example b/.env.example index 70be479..f3a1ad2 100644 --- a/.env.example +++ b/.env.example @@ -4,6 +4,14 @@ APP_ENV=prod # The root directory of the application APP_ROOT=/app/data +# Database connection information, default is sqlite on $APP_ROOT/data.db, you can use mysql or postgresql. +#DB_TYPE=postgres/mysql +#DB_HOST=localhost +#DB_PORT=5432 +#DB_USER=postgres +#DB_PASSWORD=postgres +#DB_NAME=telegram-files + # PUID and PGID are the user id and group id of the user who owns the files in the mounted volume. # PUID: 1000 # PGID: 1000 diff --git a/api/build.gradle b/api/build.gradle index a5f725a..1df3d1c 100644 --- a/api/build.gradle +++ b/api/build.gradle @@ -20,6 +20,9 @@ dependencies { implementation 'io.vertx:vertx-sql-client-templates:4.5.11' implementation 'io.vertx:vertx-health-check:4.5.11' implementation 'org.xerial:sqlite-jdbc:3.47.1.0' + implementation 'io.vertx:vertx-pg-client:4.5.11' + implementation 'io.vertx:vertx-mysql-client:4.5.11' + implementation 'com.ongres.scram:client:2.1' implementation 'io.agroal:agroal-api:1.16' implementation 'io.agroal:agroal-pool:1.16' implementation 'cn.hutool:hutool-core:5.8.34' diff --git a/api/src/main/java/telegram/files/Config.java b/api/src/main/java/telegram/files/Config.java index 4c9f52d..29118ac 100644 --- a/api/src/main/java/telegram/files/Config.java +++ b/api/src/main/java/telegram/files/Config.java @@ -12,6 +12,7 @@ import io.vertx.core.ThreadingModel; import java.io.File; import java.io.IOException; +import java.util.Objects; import java.util.logging.*; public class Config { @@ -21,6 +22,18 @@ public class Config { public static final String APP_ROOT = System.getenv("APP_ROOT"); + public static final String DB_TYPE = StrUtil.blankToDefault(System.getenv("DB_TYPE"), "sqlite"); + + public static final String DB_HOST = System.getenv("DB_HOST"); + + public static final int DB_PORT = Convert.toInt(System.getenv("DB_PORT")); + + public static final String DB_USER = System.getenv("DB_USER"); + + public static final String DB_PASSWORD = System.getenv("DB_PASSWORD"); + + public static final String DB_NAME = System.getenv("DB_NAME"); + public static final String LOG_PATH = APP_ROOT + File.separator + "logs"; public static final String TELEGRAM_ROOT = APP_ROOT + File.separator + "account"; @@ -109,6 +122,18 @@ public class Config { Logger.getLogger("telegram.files").setLevel(logLevel); } + public static boolean isSqlite() { + return Objects.equals(DB_TYPE, "sqlite"); + } + + public static boolean isPostgres() { + return Objects.equals(DB_TYPE, "postgres"); + } + + public static boolean isMysql() { + return Objects.equals(DB_TYPE, "mysql"); + } + public static class JDKLogFactory extends LogFactory { public JDKLogFactory() { diff --git a/api/src/main/java/telegram/files/DataVerticle.java b/api/src/main/java/telegram/files/DataVerticle.java index f391bb0..5bfcede 100644 --- a/api/src/main/java/telegram/files/DataVerticle.java +++ b/api/src/main/java/telegram/files/DataVerticle.java @@ -7,10 +7,18 @@ import cn.hutool.log.LogFactory; import io.vertx.core.AbstractVerticle; import io.vertx.core.Future; import io.vertx.core.Promise; +import io.vertx.core.Vertx; +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.SqlConnection; +import io.vertx.sqlclient.SqlClient; +import io.vertx.sqlclient.SqlConnectOptions; import org.jooq.lambda.tuple.Tuple; import telegram.files.repository.*; import telegram.files.repository.impl.FileRepositoryImpl; @@ -25,7 +33,7 @@ public class DataVerticle extends AbstractVerticle { private static final Log log = LogFactory.get(); - public static JDBCPool pool; + public static Pool pool; public static FileRepository fileRepository; @@ -35,16 +43,28 @@ public class DataVerticle extends AbstractVerticle { public static StatisticRepository statisticRepository; + private static SqlConnectOptions sqlConnectOptions; + + 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() + .setPort(Config.DB_PORT) + .setHost(Config.DB_HOST) + .setDatabase(Config.DB_NAME) + .setUser(Config.DB_USER) + .setPassword(Config.DB_PASSWORD); + } + } + public void start(Promise stopPromise) { - pool = JDBCPool.pool(vertx, - new JDBCConnectOptions() - .setJdbcUrl("jdbc:sqlite:%s?journal_mode=WAL&busy_timeout=30000&synchronous=NORMAL&cache_size=-2000".formatted(getDataPath())), - new PoolOptions() - .setMaxSize(8) - .setName("pool-tf") - .setIdleTimeout(300000) - .setPoolCleanerPeriod(300000) - ); + pool = buildSqlClient(); settingRepository = new SettingRepositoryImpl(pool); telegramRepository = new TelegramRepositoryImpl(pool); fileRepository = new FileRepositoryImpl(pool); @@ -55,21 +75,19 @@ public class DataVerticle extends AbstractVerticle { new FileRecord.FileRecordDefinition(), new StatisticRecord.StatisticRecordDefinition() ); - pool.getConnection() - .compose(conn -> isCompletelyNewInitialization(conn).map(isNew -> Tuple.tuple(conn, isNew))) - .compose(tuple -> Future.all(definitions.stream().map(d -> d.createTable(tuple.v1)).toList()).map(tuple)) - .compose(tuple -> settingRepository.getByKey(SettingKey.version).map(tuple::concat)) + 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))) .compose(tuple -> { - if (tuple.v2) return Future.succeededFuture(); + if (tuple.v1) return Future.succeededFuture(); - SqlConnection conn = tuple.v1; - Version version = tuple.v3 == null ? new Version("0.0.0") : tuple.v3; - return Future.all(definitions.stream().map(d -> d.migrate(conn, version, new Version(Start.VERSION))).toList()); + Version version = tuple.v2 == null ? new Version("0.0.0") : tuple.v2; + return Future.all(definitions.stream().map(d -> d.migrate(pool, version, new Version(Start.VERSION))).toList()); }) .compose(r -> settingRepository.createOrUpdate(SettingKey.version.name(), Start.VERSION)) .onSuccess(r -> { - log.info("Database initialized"); + log.info("Database {} initialized.", Config.DB_TYPE); stopPromise.complete(); }) .onFailure(err -> { @@ -99,14 +117,148 @@ public class DataVerticle extends AbstractVerticle { return dataPath; } - private Future isCompletelyNewInitialization(SqlConnection conn) { - return conn.query(""" - SELECT name - FROM sqlite_master - WHERE type = 'table' - ORDER BY name; - """) + private Pool buildSqlClient() { + PoolOptions poolOptions = new PoolOptions() + .setShared(true) + .setMaxSize(8) + .setName("pool-tf") + .setIdleTimeout(300000) + .setPoolCleanerPeriod(300000); + + return createPool(vertx, + Config.isSqlite() ? new JDBCConnectOptions() + .setJdbcUrl("jdbc:sqlite:%s?journal_mode=WAL&busy_timeout=30000&synchronous=NORMAL&cache_size=-2000".formatted(getDataPath())) : + sqlConnectOptions, + poolOptions); + } + + private Future isCompletelyNewInitialization() { + if (Config.isSqlite()) { + return pool.query(""" + SELECT name + FROM sqlite_master + WHERE type = 'table' + ORDER BY name; + """) + .execute() + .map(rs -> rs.size() == 0); + } else if (Config.isPostgres()) { + return isCompletelyNewInitializationForPostgres(); + } else if (Config.isMysql()) { + return isCompletelyNewInitializationForMySQL(); + } else { + return Future.failedFuture("Unsupported database type"); + } + } + + private Future isCompletelyNewInitializationForPostgres() { + SqlClient sqlClient = createSqlClient(vertx, createDefaultOptions()); + + return sqlClient.query(""" + SELECT 1 FROM pg_database WHERE datname = '%s' + """.formatted(Config.DB_NAME)) .execute() - .map(rs -> rs.size() == 0); + .map(rs -> rs.size() == 0) + .compose(isNew -> { + if (isNew) { + return sqlClient.query(""" + CREATE DATABASE "%s" + """.formatted(Config.DB_NAME)) + .execute() + .map(true); + } else { + return Future.succeededFuture(false); + } + }) + .eventually(() -> sqlClient.close()) + .onFailure(err -> log.error("Failed to check database initialization: %s".formatted(err.getMessage()))); + } + + private Future isCompletelyNewInitializationForMySQL() { + SqlClient sqlClient = createSqlClient(vertx, createDefaultOptions()); + + return sqlClient.query(""" + SELECT SCHEMA_NAME + FROM INFORMATION_SCHEMA.SCHEMATA + WHERE SCHEMA_NAME = '%s' + """.formatted(Config.DB_NAME)) + .execute() + .map(rs -> rs.size() == 0) + .compose(isNew -> { + if (isNew) { + return sqlClient.query(""" + CREATE DATABASE `%s` collate utf8mb4_bin; + """.formatted(Config.DB_NAME)) + .execute() + .map(true); + } else { + return Future.succeededFuture(false); + } + }) + .eventually(() -> sqlClient.close()) + .onFailure(err -> log.error("Failed to check database initialization: %s".formatted(err.getMessage()))); + } + + public static SqlConnectOptions getSqlConnectOptions() { + return sqlConnectOptions; + } + + public static SqlConnectOptions createDefaultOptions() { + if (Config.isPostgres()) { + SqlConnectOptions options = new SqlConnectOptions(sqlConnectOptions.toJson()); + options.setDatabase("postgres"); + return options; + } else if (Config.isMysql()) { + SqlConnectOptions options = new SqlConnectOptions(sqlConnectOptions.toJson()); + options.setDatabase("mysql"); + return options; + } else { + throw new NoStackTraceException("Unsupported database type"); + } + } + + public static Pool createPool(Vertx vertx, Object connectOptions, PoolOptions poolOptions) { + if (Config.isSqlite()) { + return JDBCPool.pool(vertx, + (JDBCConnectOptions) connectOptions, + poolOptions + ); + } else if (Config.isPostgres()) { + return PgBuilder + .pool() + .using(vertx) + .with(poolOptions) + .connectingTo((SqlConnectOptions) connectOptions) + .build(); + } else if (Config.isMysql()) { + return MySQLBuilder + .pool() + .using(vertx) + .with(poolOptions) + .connectingTo((SqlConnectOptions) connectOptions) + .build(); + } else { + throw new NoStackTraceException("Unsupported database type"); + } + } + + public static SqlClient createSqlClient(Vertx vertx, Object connectOptions) { + if (Config.isSqlite()) { + return JDBCPool.pool(vertx, (JDBCConnectOptions) connectOptions, new PoolOptions().setMaxSize(1)); + } else if (Config.isPostgres()) { + return PgBuilder + .client() + .using(vertx) + .connectingTo((SqlConnectOptions) connectOptions) + .build(); + } else if (Config.isMysql()) { + return MySQLBuilder + .client() + .using(vertx) + .connectingTo((SqlConnectOptions) connectOptions) + .build(); + } else { + throw new NoStackTraceException("Unsupported database type"); + } } } diff --git a/api/src/main/java/telegram/files/repository/Definition.java b/api/src/main/java/telegram/files/repository/Definition.java index 30ef4ba..0ab38c5 100644 --- a/api/src/main/java/telegram/files/repository/Definition.java +++ b/api/src/main/java/telegram/files/repository/Definition.java @@ -4,7 +4,7 @@ import cn.hutool.core.lang.Version; import cn.hutool.log.Log; import cn.hutool.log.LogFactory; import io.vertx.core.Future; -import io.vertx.sqlclient.SqlConnection; +import io.vertx.sqlclient.SqlClient; import java.util.TreeMap; import java.util.stream.Stream; @@ -19,15 +19,15 @@ public interface Definition { return new TreeMap<>(); } - default Future createTable(SqlConnection conn) { - return conn + default Future createTable(SqlClient sqlClient) { + return sqlClient .query(getScheme()) .execute() .onFailure(err -> log.error("Failed to create table: %s".formatted(err.getMessage()))) .mapEmpty(); } - default Future migrate(SqlConnection conn, Version lastVersion, Version currentVersion) { + default Future migrate(SqlClient sqlClient, Version lastVersion, Version currentVersion) { TreeMap migrations = getMigrations(); if (migrations.isEmpty()) { return Future.succeededFuture(); @@ -35,7 +35,7 @@ public interface Definition { return Future.all(migrations.subMap(lastVersion, false, currentVersion, true).values() .stream() .flatMap(arr -> Stream.of(arr) - .map(sql -> conn.query(sql) + .map(sql -> sqlClient.query(sql) .execute() .onFailure(e -> log.error("Failed to apply migration: %s".formatted(sql), e))) ) diff --git a/api/src/main/java/telegram/files/repository/FileRecord.java b/api/src/main/java/telegram/files/repository/FileRecord.java index cc4df6f..3517b95 100644 --- a/api/src/main/java/telegram/files/repository/FileRecord.java +++ b/api/src/main/java/telegram/files/repository/FileRecord.java @@ -5,6 +5,7 @@ import cn.hutool.core.lang.Version; import cn.hutool.core.map.MapUtil; import io.vertx.sqlclient.templates.RowMapper; import io.vertx.sqlclient.templates.TupleMapper; +import telegram.files.Config; import java.util.Objects; import java.util.TreeMap; @@ -55,7 +56,7 @@ public record FileRecord(int id, //file id will change type VARCHAR(255), mime_type VARCHAR(255), file_name VARCHAR(255), - thumbnail VARCHAR(255), + thumbnail VARCHAR(2056), caption VARCHAR(255), local_path VARCHAR(255), download_status VARCHAR(255), @@ -99,7 +100,7 @@ public record FileRecord(int id, //file id will change row.getLong("message_id"), Objects.requireNonNullElse(row.getLong("media_album_id"), 0L), row.getInteger("date"), - Convert.toBool(row.getInteger("has_sensitive_content")), + Config.isPostgres() ? row.getBoolean("has_sensitive_content") : Convert.toBool(row.getInteger("has_sensitive_content")), row.getLong("size"), row.getLong("downloaded_size"), row.getString("type"), diff --git a/api/src/main/java/telegram/files/repository/SettingRecord.java b/api/src/main/java/telegram/files/repository/SettingRecord.java index 149d367..8dd40ff 100644 --- a/api/src/main/java/telegram/files/repository/SettingRecord.java +++ b/api/src/main/java/telegram/files/repository/SettingRecord.java @@ -2,18 +2,21 @@ package telegram.files.repository; import io.vertx.sqlclient.templates.RowMapper; import io.vertx.sqlclient.templates.TupleMapper; +import telegram.files.Config; import java.util.Map; public record SettingRecord(String key, String value) { + public static final String KEY_FIELD = Config.isMysql() ? "`key`" : "key"; + public static final String SCHEME = """ CREATE TABLE IF NOT EXISTS setting_record ( - key VARCHAR(255) PRIMARY KEY, - value TEXT + %s VARCHAR(255) PRIMARY KEY, + value TEXT ) - """; + """.formatted(KEY_FIELD); public static class SettingRecordDefinition implements Definition { @Override diff --git a/api/src/main/java/telegram/files/repository/impl/AbstractSqlRepository.java b/api/src/main/java/telegram/files/repository/impl/AbstractSqlRepository.java new file mode 100644 index 0000000..a7837c3 --- /dev/null +++ b/api/src/main/java/telegram/files/repository/impl/AbstractSqlRepository.java @@ -0,0 +1,14 @@ +package telegram.files.repository.impl; + + +import io.vertx.sqlclient.SqlClient; + +public abstract class AbstractSqlRepository { + + protected final SqlClient sqlClient; + + public AbstractSqlRepository(SqlClient sqlClient) { + this.sqlClient = sqlClient; + } + +} 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 5ac5537..ed240ae 100644 --- a/api/src/main/java/telegram/files/repository/impl/FileRepositoryImpl.java +++ b/api/src/main/java/telegram/files/repository/impl/FileRepositoryImpl.java @@ -12,7 +12,7 @@ import cn.hutool.log.LogFactory; import io.vertx.core.Future; import io.vertx.core.json.JsonArray; import io.vertx.core.json.JsonObject; -import io.vertx.jdbcclient.JDBCPool; +import io.vertx.sqlclient.SqlClient; import io.vertx.sqlclient.SqlResult; import io.vertx.sqlclient.templates.SqlTemplate; import org.jooq.lambda.tuple.Tuple; @@ -29,20 +29,18 @@ import java.util.*; import java.util.stream.Collectors; import java.util.stream.IntStream; -public class FileRepositoryImpl implements FileRepository { +public class FileRepositoryImpl extends AbstractSqlRepository implements FileRepository { private static final Log log = LogFactory.get(); - private final JDBCPool pool; - - public FileRepositoryImpl(JDBCPool pool) { - this.pool = pool; + public FileRepositoryImpl(SqlClient sqlClient) { + super(sqlClient); } @Override public Future create(FileRecord fileRecord) { return SqlTemplate - .forUpdate(pool, """ + .forUpdate(sqlClient, """ INSERT INTO file_record(id, unique_id, telegram_id, chat_id, message_id, media_album_id, date, has_sensitive_content, size, downloaded_size, type, mime_type, @@ -78,7 +76,7 @@ public class FileRepositoryImpl implements FileRepository { return Future.succeededFuture(new HashMap<>()); } return SqlTemplate - .forQuery(pool, """ + .forQuery(sqlClient, """ SELECT * FROM file_record WHERE chat_id = #{chatId} AND id IN (#{fileIds}) """) .mapTo(FileRecord.ROW_MAPPER) @@ -184,7 +182,7 @@ public class FileRepositoryImpl implements FileRepository { log.trace("Get files with where: %s params: %s".formatted(whereClause, params)); return Future.all( SqlTemplate - .forQuery(pool, """ + .forQuery(sqlClient, """ SELECT * FROM file_record WHERE %s ORDER BY %s LIMIT #{limit} """.formatted(whereClause, orderBy)) .mapTo(FileRecord.ROW_MAPPER) @@ -193,7 +191,7 @@ public class FileRepositoryImpl implements FileRepository { .map(IterUtil::toList) , SqlTemplate - .forQuery(pool, """ + .forQuery(sqlClient, """ SELECT COUNT(*) FROM file_record WHERE %s """.formatted(countClause)) .mapTo(rs -> rs.getLong(0)) @@ -223,7 +221,7 @@ public class FileRepositoryImpl implements FileRepository { params.put("uniqueId" + i, uniqueIds.get(i)); } return SqlTemplate - .forQuery(pool, """ + .forQuery(sqlClient, """ SELECT * FROM file_record WHERE unique_id IN (%s) """.formatted(uniqueIdPlaceholders)) .mapTo(FileRecord.ROW_MAPPER) @@ -241,7 +239,7 @@ public class FileRepositoryImpl implements FileRepository { @Override public Future getByPrimaryKey(int fileId, String uniqueId) { return SqlTemplate - .forQuery(pool, """ + .forQuery(sqlClient, """ SELECT * FROM file_record WHERE id = #{fileId} AND unique_id = #{uniqueId} """) .mapTo(FileRecord.ROW_MAPPER) @@ -254,7 +252,7 @@ public class FileRepositoryImpl implements FileRepository { @Override public Future getByUniqueId(String uniqueId) { return SqlTemplate - .forQuery(pool, """ + .forQuery(sqlClient, """ SELECT * FROM file_record WHERE unique_id = #{uniqueId} LIMIT 1 """) .mapTo(FileRecord.ROW_MAPPER) @@ -270,7 +268,7 @@ public class FileRepositoryImpl implements FileRepository { return Future.succeededFuture(null); } return SqlTemplate - .forQuery(pool, """ + .forQuery(sqlClient, """ SELECT caption FROM file_record WHERE media_album_id = #{mediaAlbumId} LIMIT 1 """) .mapTo(row -> row.getString("caption")) @@ -282,7 +280,7 @@ public class FileRepositoryImpl implements FileRepository { @Override public Future getDownloadStatistics(long telegramId) { return SqlTemplate - .forQuery(pool, """ + .forQuery(sqlClient, """ SELECT COUNT(*) AS total, COUNT(CASE WHEN download_status = 'downloading' THEN 1 END) AS downloading, COUNT(CASE WHEN download_status = 'paused' THEN 1 END) AS paused, @@ -316,7 +314,7 @@ public class FileRepositoryImpl implements FileRepository { @Override public Future getCompletedRangeStatistics(long telegramId, long startTime, long endTime, int timeRange) { return SqlTemplate - .forQuery(pool, """ + .forQuery(sqlClient, """ SELECT strftime( CASE WHEN #{timeRange} = 1 THEN '%Y-%m-%d %H:%M' @@ -373,7 +371,7 @@ public class FileRepositoryImpl implements FileRepository { @Override public Future countByStatus(long telegramId, FileRecord.DownloadStatus downloadStatus) { return SqlTemplate - .forQuery(pool, """ + .forQuery(sqlClient, """ SELECT COUNT(*) FROM file_record WHERE telegram_id = #{telegramId} AND download_status = #{downloadStatus} """) .mapTo(rs -> rs.getInteger(0)) @@ -408,7 +406,7 @@ public class FileRepositoryImpl implements FileRepository { } return SqlTemplate - .forUpdate(pool, """ + .forUpdate(sqlClient, """ UPDATE file_record SET id = #{fileId}, local_path = #{localPath}, download_status = #{downloadStatus}, @@ -459,7 +457,7 @@ public class FileRepositoryImpl implements FileRepository { } return SqlTemplate - .forUpdate(pool, """ + .forUpdate(sqlClient, """ UPDATE file_record SET transfer_status = #{transferStatus}, local_path = #{localPath} @@ -498,7 +496,7 @@ public class FileRepositoryImpl implements FileRepository { return Future.succeededFuture(); } return SqlTemplate - .forUpdate(pool, """ + .forUpdate(sqlClient, """ UPDATE file_record SET id = #{fileId} WHERE unique_id = #{uniqueId} """) .execute(Map.of("fileId", fileId, "uniqueId", uniqueId)) @@ -527,7 +525,7 @@ public class FileRepositoryImpl implements FileRepository { return Future.succeededFuture(0); } return SqlTemplate - .forUpdate(pool, """ + .forUpdate(sqlClient, """ UPDATE file_record SET caption = #{caption} WHERE media_album_id = #{mediaAlbumId} """) .execute(Map.of("mediaAlbumId", mediaAlbumId, "caption", theCaption)) @@ -543,7 +541,7 @@ public class FileRepositoryImpl implements FileRepository { return Future.succeededFuture(); } return SqlTemplate - .forUpdate(pool, """ + .forUpdate(sqlClient, """ DELETE FROM file_record WHERE unique_id = #{uniqueId} """) .execute(Map.of("uniqueId", uniqueId)) 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 4e431f5..16fd3f0 100644 --- a/api/src/main/java/telegram/files/repository/impl/SettingRepositoryImpl.java +++ b/api/src/main/java/telegram/files/repository/impl/SettingRepositoryImpl.java @@ -6,8 +6,9 @@ import cn.hutool.core.util.StrUtil; import cn.hutool.log.Log; import cn.hutool.log.LogFactory; import io.vertx.core.Future; -import io.vertx.jdbcclient.JDBCPool; +import io.vertx.sqlclient.SqlClient; import io.vertx.sqlclient.templates.SqlTemplate; +import telegram.files.Config; import telegram.files.repository.SettingKey; import telegram.files.repository.SettingRecord; import telegram.files.repository.SettingRepository; @@ -17,23 +18,24 @@ import java.util.List; import java.util.Map; import java.util.stream.Collectors; -public class SettingRepositoryImpl implements SettingRepository { +public class SettingRepositoryImpl extends AbstractSqlRepository implements SettingRepository { private static final Log log = LogFactory.get(); - private final JDBCPool pool; - - public SettingRepositoryImpl(JDBCPool pool) { - this.pool = pool; + public SettingRepositoryImpl(SqlClient sqlClient) { + super(sqlClient); } @Override public Future createOrUpdate(String key, String value) { return SqlTemplate - .forUpdate(pool, """ - INSERT INTO setting_record(key, value) VALUES (#{key}, #{value}) - ON CONFLICT (key) DO UPDATE SET value = #{value} - """) + .forUpdate(sqlClient, Config.isMysql() ? + """ + INSERT INTO setting_record(`key`, value) VALUES (#{key}, #{value}) + ON DUPLICATE KEY UPDATE value = VALUES(value)""" : + """ + INSERT INTO setting_record(key, value) VALUES (#{key}, #{value}) + ON CONFLICT (key) DO UPDATE SET value = #{value}""") .mapFrom(SettingRecord.PARAM_MAPPER) .execute(new SettingRecord(key, value)) .map(r -> new SettingRecord(key, value)) @@ -55,9 +57,9 @@ public class SettingRepositoryImpl implements SettingRepository { .collect(Collectors.joining(",")); return SqlTemplate - .forQuery(pool, """ - SELECT * FROM setting_record WHERE key IN (%s) - """.formatted(keyStr)) + .forQuery(sqlClient, """ + SELECT %s, value FROM setting_record WHERE %s IN (%s) + """.formatted(SettingRecord.KEY_FIELD, SettingRecord.KEY_FIELD, keyStr)) .mapTo(SettingRecord.ROW_MAPPER) .execute(Collections.emptyMap()) .map(IterUtil::toList) @@ -71,9 +73,9 @@ public class SettingRepositoryImpl implements SettingRepository { @SuppressWarnings("unchecked") public Future getByKey(SettingKey key) { return SqlTemplate - .forQuery(pool, """ - SELECT value FROM setting_record WHERE key = #{key} - """) + .forQuery(sqlClient, """ + SELECT value FROM setting_record WHERE %s = #{key} + """.formatted(SettingRecord.KEY_FIELD)) .mapTo(row -> row.getString("value")) .execute(Map.of("key", key.name())) .map(rs -> { 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 55c6842..3b2e149 100644 --- a/api/src/main/java/telegram/files/repository/impl/StatisticRepositoryImpl.java +++ b/api/src/main/java/telegram/files/repository/impl/StatisticRepositoryImpl.java @@ -4,7 +4,7 @@ import cn.hutool.core.collection.IterUtil; import cn.hutool.log.Log; import cn.hutool.log.LogFactory; import io.vertx.core.Future; -import io.vertx.jdbcclient.JDBCPool; +import io.vertx.sqlclient.SqlClient; import io.vertx.sqlclient.templates.SqlTemplate; import telegram.files.repository.StatisticRecord; import telegram.files.repository.StatisticRepository; @@ -12,20 +12,18 @@ import telegram.files.repository.StatisticRepository; import java.util.List; import java.util.Map; -public class StatisticRepositoryImpl implements StatisticRepository { +public class StatisticRepositoryImpl extends AbstractSqlRepository implements StatisticRepository { private static final Log log = LogFactory.get(); - private final JDBCPool pool; - - public StatisticRepositoryImpl(JDBCPool pool) { - this.pool = pool; + public StatisticRepositoryImpl(SqlClient sqlClient) { + super(sqlClient); } @Override public Future create(StatisticRecord record) { return SqlTemplate - .forUpdate(pool, """ + .forUpdate(sqlClient, """ INSERT INTO statistic_record(related_id, type, timestamp, data) VALUES (#{related_id}, #{type}, #{timestamp}, #{data}) """) @@ -44,7 +42,7 @@ public class StatisticRepositoryImpl implements StatisticRepository { long startTime, long endTime) { return SqlTemplate - .forQuery(pool, """ + .forQuery(sqlClient, """ SELECT * FROM statistic_record WHERE type = #{type} 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 1e0bf39..6da44b1 100644 --- a/api/src/main/java/telegram/files/repository/impl/TelegramRepositoryImpl.java +++ b/api/src/main/java/telegram/files/repository/impl/TelegramRepositoryImpl.java @@ -5,7 +5,7 @@ import cn.hutool.core.map.MapUtil; import cn.hutool.log.Log; import cn.hutool.log.LogFactory; import io.vertx.core.Future; -import io.vertx.jdbcclient.JDBCPool; +import io.vertx.sqlclient.SqlClient; import io.vertx.sqlclient.templates.SqlTemplate; import telegram.files.Config; import telegram.files.repository.TelegramRecord; @@ -16,14 +16,12 @@ import java.util.Collections; import java.util.List; import java.util.UUID; -public class TelegramRepositoryImpl implements TelegramRepository { +public class TelegramRepositoryImpl extends AbstractSqlRepository implements TelegramRepository { private static final Log log = LogFactory.get(); - private final JDBCPool pool; - - public TelegramRepositoryImpl(JDBCPool pool) { - this.pool = pool; + public TelegramRepositoryImpl(SqlClient sqlClient) { + super(sqlClient); } @Override @@ -34,7 +32,7 @@ public class TelegramRepositoryImpl implements TelegramRepository { @Override public Future create(TelegramRecord telegramRecord) { return SqlTemplate - .forUpdate(pool, "INSERT INTO telegram_record(id, first_name, root_path, proxy) VALUES (#{id}, #{first_name}, #{root_path}, #{proxy})") + .forUpdate(sqlClient, "INSERT INTO telegram_record(id, first_name, root_path, proxy) VALUES (#{id}, #{first_name}, #{root_path}, #{proxy})") .mapFrom(TelegramRecord.PARAM_MAPPER) .execute(telegramRecord) .map(r -> telegramRecord) @@ -47,7 +45,7 @@ public class TelegramRepositoryImpl implements TelegramRepository { @Override public Future update(TelegramRecord telegramRecord) { return SqlTemplate - .forUpdate(pool, "UPDATE telegram_record SET first_name = #{first_name}, root_path = #{root_path}, proxy = #{proxy} WHERE id = #{id}") + .forUpdate(sqlClient, "UPDATE telegram_record SET first_name = #{first_name}, root_path = #{root_path}, proxy = #{proxy} WHERE id = #{id}") .mapFrom(TelegramRecord.PARAM_MAPPER) .execute(telegramRecord) .map(r -> telegramRecord) @@ -60,7 +58,7 @@ public class TelegramRepositoryImpl implements TelegramRepository { @Override public Future getById(long id) { return SqlTemplate - .forQuery(pool, "SELECT * FROM telegram_record WHERE id = #{id} limit 1") + .forQuery(sqlClient, "SELECT * FROM telegram_record WHERE id = #{id} limit 1") .mapTo(TelegramRecord.ROW_MAPPER) .execute(MapUtil.of("id", id)) .map(rs -> rs.size() == 0 ? null : rs.iterator().next()); @@ -69,7 +67,7 @@ public class TelegramRepositoryImpl implements TelegramRepository { @Override public Future> getAll() { return SqlTemplate - .forQuery(pool, "SELECT * FROM telegram_record ORDER BY id") + .forQuery(sqlClient, "SELECT * FROM telegram_record ORDER BY id") .mapTo(TelegramRecord.ROW_MAPPER) .execute(Collections.emptyMap()) .map(CollUtil::newArrayList); diff --git a/api/src/test/java/telegram/files/DataVerticleMigrationTest.java b/api/src/test/java/telegram/files/DataVerticleMigrationTest.java index ad862db..c7bb5c8 100644 --- a/api/src/test/java/telegram/files/DataVerticleMigrationTest.java +++ b/api/src/test/java/telegram/files/DataVerticleMigrationTest.java @@ -2,20 +2,22 @@ package telegram.files; import cn.hutool.core.collection.IterUtil; -import cn.hutool.core.io.FileUtil; import cn.hutool.core.lang.Version; import cn.hutool.log.Log; import cn.hutool.log.LogFactory; import io.vertx.core.Future; import io.vertx.core.Vertx; +import io.vertx.core.impl.NoStackTraceException; import io.vertx.jdbcclient.JDBCConnectOptions; import io.vertx.jdbcclient.JDBCPool; import io.vertx.junit5.VertxExtension; import io.vertx.junit5.VertxTestContext; import io.vertx.sqlclient.PoolOptions; +import io.vertx.sqlclient.SqlClient; import org.junit.jupiter.api.*; import org.junit.jupiter.api.extension.ExtendWith; import telegram.files.repository.SettingKey; +import telegram.files.repository.SettingRecord; import java.util.Set; import java.util.stream.Collectors; @@ -27,16 +29,19 @@ public class DataVerticleMigrationTest { @BeforeAll static void setUp() { - log.debug("APP_ROOT: " + Config.APP_ROOT); - log.debug("DATA_PATH:" + DataVerticle.getDataPath()); + DataVerticleTest.printDBInfo(); } @AfterEach - void tearDown() { - String dataPath = DataVerticle.getDataPath(); - if (FileUtil.file(dataPath).exists()) { - FileUtil.del(dataPath); - } + void tearDown(Vertx vertx, VertxTestContext testContext) { + DataVerticleTest.clear(vertx) + .onComplete(v -> { + if (v.failed()) { + testContext.failNow(v.cause()); + } else { + testContext.completeNow(); + } + }); } @Test @@ -56,7 +61,7 @@ public class DataVerticleMigrationTest { initializeLegacyDatabase(vertx) .compose(v -> vertx.deployVerticle(new DataVerticle())) .compose(v -> DataVerticle.pool.getConnection()) - .compose(conn -> conn.query("PRAGMA table_info(file_record)").execute() + .compose(conn -> conn.query(getColumnsQuery()).execute() .compose(result -> { testContext.verify(() -> { // Verify whether there are new columns @@ -81,7 +86,7 @@ public class DataVerticleMigrationTest { initializeOldVersionDatabase(vertx) .compose(v -> vertx.deployVerticle(new DataVerticle())) .compose(v -> DataVerticle.pool.getConnection()) - .compose(conn -> conn.query("PRAGMA table_info(file_record)").execute() + .compose(conn -> conn.query(getColumnsQuery()).execute() .compose(result -> { testContext.verify(() -> { // Verify whether there are new columns @@ -105,10 +110,7 @@ public class DataVerticleMigrationTest { .compose(id -> { // Verify that the database has been fully initialized return DataVerticle.pool.getConnection() - .compose(conn -> conn.query(""" - SELECT name FROM sqlite_master - WHERE type='table' AND name IN ('setting_record', 'telegram_record', 'file_record', 'statistic_record') - """).execute() + .compose(conn -> conn.query(getTablesQuery()).execute() .compose(result -> { testContext.verify(() -> { Assertions.assertEquals(4, result.size()); @@ -122,58 +124,117 @@ public class DataVerticleMigrationTest { private Future initializeLegacyDatabase(Vertx vertx) { // Create an old version database without a version field return Future.succeededFuture() - .compose(v -> { - JDBCPool tempPool = JDBCPool.pool(vertx, - new JDBCConnectOptions() - .setJdbcUrl("jdbc:sqlite:%s".formatted(DataVerticle.getDataPath())), - new PoolOptions().setMaxSize(1).setName("temp-pool") - ); - return tempPool.getConnection() - .compose(conn -> conn.query(""" - CREATE TABLE file_record ( - id INTEGER PRIMARY KEY AUTOINCREMENT, - file_id TEXT NOT NULL, - file_name TEXT NOT NULL - ) - """).execute() - .compose(v2 -> conn.query(""" - INSERT INTO file_record (file_id, file_name) - VALUES ('123', 'test.txt') - """).execute()) - .compose(v3 -> conn.close()) - .eventually(() -> tempPool.close())); - }); + .compose(v -> createTempSqlClient(vertx)) + .compose(sqlClient -> sqlClient.query(""" + CREATE TABLE file_record ( + id INTEGER PRIMARY KEY, + file_id TEXT NOT NULL, + file_name TEXT NOT NULL + ) + """).execute() + .compose(v2 -> sqlClient.query(""" + INSERT INTO file_record (id, file_id, file_name) + VALUES (1, '123', 'test.txt') + """).execute()) + .eventually(() -> sqlClient.close()) + ) + .mapEmpty(); } private Future initializeOldVersionDatabase(Vertx vertx) { // Create a database with a version field but a lower version return Future.succeededFuture() - .compose(v -> { - JDBCPool tempPool = JDBCPool.pool(vertx, - new JDBCConnectOptions() - .setJdbcUrl("jdbc:sqlite:%s".formatted(DataVerticle.getDataPath())), - new PoolOptions().setMaxSize(1).setName("temp-pool") - ); - return tempPool.getConnection() - .compose(conn -> conn.query(""" - CREATE TABLE setting_record ( - key TEXT PRIMARY KEY, - value TEXT NOT NULL - ) - """).execute() - .compose(v2 -> conn.query(""" - INSERT INTO setting_record (key, value) - VALUES ('version', '0.1.6') - """).execute()) - .compose(v3 -> conn.query(""" - CREATE TABLE file_record ( - id INTEGER PRIMARY KEY AUTOINCREMENT, - file_id TEXT NOT NULL, - file_name TEXT NOT NULL - ) - """).execute()) - .compose(v4 -> conn.close()) - .eventually(() -> tempPool.close())); - }); + .compose(v -> createTempSqlClient(vertx)) + .compose(sqlClient -> sqlClient.query(""" + CREATE TABLE setting_record ( + %s VARCHAR(255) PRIMARY KEY, + value TEXT NOT NULL + ) + """.formatted(SettingRecord.KEY_FIELD)).execute() + .compose(v2 -> sqlClient.query(""" + INSERT INTO setting_record (%s, value) + VALUES ('version', '0.1.6') + """.formatted(SettingRecord.KEY_FIELD)).execute()) + .compose(v3 -> sqlClient.query(""" + CREATE TABLE file_record ( + id INTEGER PRIMARY KEY, + file_id TEXT NOT NULL, + file_name TEXT NOT NULL + ) + """).execute()) + .eventually(() -> sqlClient.close()) + ) + .mapEmpty(); + } + + private Future createTempSqlClient(Vertx vertx) { + if (Config.isSqlite()) { + return Future.succeededFuture(JDBCPool.pool(vertx, + new JDBCConnectOptions() + .setJdbcUrl("jdbc:sqlite:%s".formatted(DataVerticle.getDataPath())), + new PoolOptions().setMaxSize(1).setName("temp-pool") + )); + } else if (Config.isPostgres() || Config.isMysql()) { + SqlClient sqlClient = DataVerticle.createSqlClient(vertx, DataVerticle.createDefaultOptions()); + String createDatabaseQuery; + if (Config.isPostgres()) { + createDatabaseQuery = """ + CREATE DATABASE "%s"; + """.formatted(Config.DB_NAME); + } else { + createDatabaseQuery = """ + CREATE DATABASE `%s` collate utf8mb4_bin; + """.formatted(Config.DB_NAME); + } + + return sqlClient.query(createDatabaseQuery) + .execute() + .onSuccess(r -> log.info("Database created")) + .onFailure(err -> log.error("Failed to create database: %s".formatted(err.getMessage()))) + .map(DataVerticle.createSqlClient(vertx, DataVerticle.getSqlConnectOptions())); + } else { + throw new NoStackTraceException("Unsupported database type"); + } + } + + private String getColumnsQuery() { + String getColumnsQuery; + if (Config.isPostgres()) { + getColumnsQuery = """ + SELECT column_name as name FROM information_schema.columns + WHERE table_name = 'file_record' + """; + } else if (Config.isMysql()) { + getColumnsQuery = """ + SELECT column_name as name FROM information_schema.columns + WHERE table_name = 'file_record' + """; + } else { + getColumnsQuery = """ + PRAGMA table_info(file_record) + """; + } + return getColumnsQuery; + } + + private String getTablesQuery() { + String getTablesQuery; + if (Config.isPostgres()) { + getTablesQuery = """ + SELECT table_name FROM information_schema.tables + WHERE table_schema = 'public' AND table_name IN ('setting_record', 'telegram_record', 'file_record', 'statistic_record') + """; + } else if (Config.isMysql()) { + getTablesQuery = """ + SELECT table_name FROM information_schema.tables + WHERE table_schema = DATABASE() AND table_name IN ('setting_record', 'telegram_record', 'file_record', 'statistic_record') + """; + } else { + getTablesQuery = """ + SELECT name FROM sqlite_master + WHERE type='table' AND name IN ('setting_record', 'telegram_record', 'file_record', 'statistic_record') + """; + } + return getTablesQuery; } } diff --git a/api/src/test/java/telegram/files/DataVerticleTest.java b/api/src/test/java/telegram/files/DataVerticleTest.java index eb8efd9..2154060 100644 --- a/api/src/test/java/telegram/files/DataVerticleTest.java +++ b/api/src/test/java/telegram/files/DataVerticleTest.java @@ -4,9 +4,11 @@ import cn.hutool.core.io.FileUtil; import cn.hutool.core.lang.Version; import cn.hutool.log.Log; import cn.hutool.log.LogFactory; +import io.vertx.core.Future; import io.vertx.core.Vertx; import io.vertx.junit5.VertxExtension; import io.vertx.junit5.VertxTestContext; +import io.vertx.sqlclient.SqlClient; import org.junit.jupiter.api.*; import org.junit.jupiter.api.extension.ExtendWith; import telegram.files.repository.FileRecord; @@ -20,15 +22,69 @@ public class DataVerticleTest { @BeforeAll static void setUp() { - log.debug("APP_ROOT: " + Config.APP_ROOT); - log.debug("DATA_PATH:" + DataVerticle.getDataPath()); + printDBInfo(); } @AfterEach - void tearDown() { - String dataPath = DataVerticle.getDataPath(); - if (FileUtil.file(dataPath).exists()) { - FileUtil.del(dataPath); + void tearDown(Vertx vertx, VertxTestContext testContext) { + clear(vertx).onComplete(testContext.succeedingThenComplete()); + } + + public static void printDBInfo() { + if (Config.isSqlite()) { + log.debug("DB_PATH: " + DataVerticle.getDataPath()); + } else if (Config.isPostgres() || Config.isMysql()) { + log.debug("DB_HOST: " + Config.DB_HOST); + log.debug("DB_PORT: " + Config.DB_PORT); + log.debug("DB_NAME: " + Config.DB_NAME); + log.debug("DB_USER: " + Config.DB_USER); + } else { + log.error("Unknown database type"); + } + } + + public static Future clear(Vertx vertx) { + if (Config.isSqlite()) { + String dataPath = DataVerticle.getDataPath(); + if (FileUtil.file(dataPath).exists()) { + FileUtil.del(dataPath); + } + return Future.succeededFuture(); + } else if (Config.isPostgres()) { + SqlClient sqlClient = DataVerticle.createSqlClient(vertx, DataVerticle.createDefaultOptions()); + + return sqlClient + .query("ALTER DATABASE \"%s\" ALLOW_CONNECTIONS false;".formatted(Config.DB_NAME)).execute() + .compose(r -> sqlClient + .query("SELECT pg_terminate_backend(pid) FROM pg_stat_activity WHERE datname = '%s';".formatted(Config.DB_NAME)).execute() + ) + .compose(r -> sqlClient + .query("DROP DATABASE \"%s\";".formatted(Config.DB_NAME)).execute() + ) + .onFailure(t -> log.error("Failed to clear database", t)) + .onComplete(r -> { + if (r.succeeded()) { + log.debug("Database cleared"); + } + sqlClient.close(); + }) + .mapEmpty(); + } else if (Config.isMysql()) { + SqlClient sqlClient = DataVerticle.createSqlClient(vertx, DataVerticle.createDefaultOptions()); + + return sqlClient + .query("DROP SCHEMA IF EXISTS `%s`".formatted(Config.DB_NAME)) + .execute() + .onFailure(t -> log.error("Failed to clear database", t)) + .onComplete(r -> { + if (r.succeeded()) { + log.debug("Database cleared"); + } + sqlClient.close(); + }) + .mapEmpty(); + } else { + return Future.failedFuture("Unknown database type"); } } diff --git a/api/src/test/java/telegram/files/FileDownloadStatusConcurrentTest.java b/api/src/test/java/telegram/files/FileDownloadStatusConcurrentTest.java index eb412a3..9d9089e 100644 --- a/api/src/test/java/telegram/files/FileDownloadStatusConcurrentTest.java +++ b/api/src/test/java/telegram/files/FileDownloadStatusConcurrentTest.java @@ -1,6 +1,5 @@ package telegram.files; -import cn.hutool.core.io.FileUtil; import io.vertx.core.Future; import io.vertx.core.Vertx; import io.vertx.core.json.JsonObject; @@ -23,6 +22,8 @@ public class FileDownloadStatusConcurrentTest { private static ExecutorService executor; + static Vertx vertx = Vertx.vertx(); + static FileRecord fileRecord = new FileRecord( 1, "unique_id", 1, 1, 1, 1, 1, false, 1, 0, "type", "mime_type", "file_name", "thumbnail", "caption", null, FileRecord.DownloadStatus.idle.name(), FileRecord.TransferStatus.idle.name(), 0, null @@ -32,7 +33,6 @@ public class FileDownloadStatusConcurrentTest { static void setUpAll() { executor = Executors.newFixedThreadPool(THREAD_COUNT); - Vertx vertx = Vertx.vertx(); Future future = vertx.deployVerticle(new DataVerticle()); MessyUtils.await(future); @@ -43,10 +43,7 @@ public class FileDownloadStatusConcurrentTest { @AfterAll static void tearDownAll() { executor.shutdown(); - String dataPath = DataVerticle.getDataPath(); - if (FileUtil.file(dataPath).exists()) { - FileUtil.del(dataPath); - } + DataVerticleTest.clear(vertx); } @RepeatedTest(THREAD_COUNT) @@ -85,10 +82,10 @@ public class FileDownloadStatusConcurrentTest { try { DataVerticle.fileRepository.updateDownloadStatus(newFileId, - fileRecord.uniqueId(), - updateLocalPath, - FileRecord.DownloadStatus.downloading, - completionDate) + fileRecord.uniqueId(), + updateLocalPath, + FileRecord.DownloadStatus.downloading, + completionDate) .onComplete(result -> completedCount.incrementAndGet()); } catch (Exception e) { throw new RuntimeException(e);