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

✨ feat: Add database support postgresql and mysql.

Browse source
This commit is contained in:
jarvis2f 2025-03-06 11:19:22 +08:00
parent 6431ce7846
commit 86a462a636
15 changed files with 488 additions and 172 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

8
.env.example
View file

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

3
api/build.gradle
View file

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

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

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

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

@ -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<Void> 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.<Version>getByKey(SettingKey.version).map(tuple::concat))
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)))
.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<Boolean> 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<Boolean> 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<Boolean> 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<Boolean> 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");
}
}
}

10
api/src/main/java/telegram/files/repository/Definition.java
View file

@ -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<Void> createTable(SqlConnection conn) {
return conn
default Future<Void> createTable(SqlClient sqlClient) {
return sqlClient
.query(getScheme())
.execute()
.onFailure(err -> log.error("Failed to create table: %s".formatted(err.getMessage())))
.mapEmpty();
}
default Future<Void> migrate(SqlConnection conn, Version lastVersion, Version currentVersion) {
default Future<Void> migrate(SqlClient sqlClient, Version lastVersion, Version currentVersion) {
TreeMap<Version, String[]> 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)))
)

5
api/src/main/java/telegram/files/repository/FileRecord.java
View file

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

9
api/src/main/java/telegram/files/repository/SettingRecord.java
View file

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

14
api/src/main/java/telegram/files/repository/impl/AbstractSqlRepository.java Normal file
View file

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

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

@ -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<FileRecord> 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<FileRecord> 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<FileRecord> 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<JsonObject> 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<JsonArray> 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<Integer> 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))

34
api/src/main/java/telegram/files/repository/impl/SettingRepositoryImpl.java
View file

@ -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<SettingRecord> 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 <T> Future<T> 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 -> {

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

@ -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<Void> 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}

18
api/src/main/java/telegram/files/repository/impl/TelegramRepositoryImpl.java
View file

@ -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<TelegramRecord> 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<TelegramRecord> 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<TelegramRecord> 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<List<TelegramRecord>> 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);

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

@ -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<Void> 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<Void> 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<SqlClient> 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;
}
}

68
api/src/test/java/telegram/files/DataVerticleTest.java
View file

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

17
api/src/test/java/telegram/files/FileDownloadStatusConcurrentTest.java
View file

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