Refactor downloads

This commit is contained in:
chylex committed 2025-04-25 13:07:35 +02:00
1 parent 0d3600492e
commit 44b42657ef
14 files changed
+81 -76

No files matched your search

+1 -1
View File
@@ -74,7 +74,7 @@ sealed class DebugPageModel {
Id = RandomId(rand),
Name = RandomName("u"),
DisplayName = RandomName("u"),
AvatarUrl = null,
AvatarHash = null,
Discriminator = rand.Next(minValue: 0, maxValue: 9999).ToString(),
}).ToArray();
+2 -2
View File
@@ -180,8 +180,8 @@ sealed partial class DownloadsPageModel : IAsyncDisposable {
HashSet<string> reachableNormalizedUrls = [];
HashSet<string> orphanedNormalizedUrls = [];
await foreach (Download download in state.Db.Downloads.FindAllDownloadableUrls()) {
reachableNormalizedUrls.Add(download.NormalizedUrl);
await foreach (FileUrl fileUrl in state.Db.Downloads.FindReachableFiles()) {
reachableNormalizedUrls.Add(fileUrl.NormalizedUrl);
}
await foreach (Download download in state.Db.Downloads.Get()) {
+9
View File
@@ -0,0 +1,9 @@
namespace DHT.Server.Data;
public readonly record struct FileUrl(string NormalizedUrl, string DownloadUrl, string? Type) {
public FileUrl(string url, string? type) : this(url, url, type) {}
public Download ToPendingDownload() {
return new Download(NormalizedUrl, DownloadUrl, DownloadStatus.Pending, Type, size: null);
}
}
+5 -1
View File
@@ -1,9 +1,13 @@
using DHT.Server.Download;
namespace DHT.Server.Data;
public readonly struct User {
public ulong Id { get; init; }
public string Name { get; init; }
public string? DisplayName { get; init; }
public string? AvatarUrl { get; init; }
public string? AvatarHash { get; init; }
public string? Discriminator { get; init; }
internal FileUrl? AvatarUrl => AvatarHash == null ? null : DownloadLinkExtractor.UserAvatar(Id, AvatarHash);
}
@@ -90,7 +90,7 @@ static class ViewerJsonExport {
users[user.Id] = new ViewerJson.JsonUser {
Name = user.Name,
DisplayName = user.DisplayName,
Avatar = user.AvatarUrl,
Avatar = user.AvatarHash,
};
}
@@ -106,7 +106,7 @@ public static class LegacyArchiveImport {
users[userindex[userId]] = new User {
Id = userId,
Name = userObj.RequireString("name", path),
AvatarUrl = userObj.HasKey("avatar") ? userObj.RequireString("avatar", path) : null,
AvatarHash = userObj.HasKey("avatar") ? userObj.RequireString("avatar", path) : null,
Discriminator = userObj.HasKey("tag") ? userObj.RequireString("tag", path) : null,
};
}
@@ -5,6 +5,7 @@ using System.Linq;
using System.Reactive.Linq;
using System.Threading;
using System.Threading.Tasks;
using DHT.Server.Data;
using DHT.Server.Data.Aggregations;
using DHT.Server.Data.Filters;
using DHT.Server.Download;
@@ -34,7 +35,7 @@ public interface IDownloadRepository {
Task Remove(ICollection<string> normalizedUrls);
IAsyncEnumerable<Data.Download> FindAllDownloadableUrls(CancellationToken cancellationToken = default);
IAsyncEnumerable<FileUrl> FindReachableFiles(CancellationToken cancellationToken = default);
internal sealed class Dummy : IDownloadRepository {
public IObservable<long> TotalCount { get; } = Observable.Return(0L);
@@ -79,8 +80,8 @@ public interface IDownloadRepository {
return Task.CompletedTask;
}
public IAsyncEnumerable<Data.Download> FindAllDownloadableUrls(CancellationToken cancellationToken) {
return AsyncEnumerable.Empty<Data.Download>();
public IAsyncEnumerable<FileUrl> FindReachableFiles(CancellationToken cancellationToken) {
return AsyncEnumerable.Empty<FileUrl>();
}
}
}
@@ -62,6 +62,15 @@ sealed class SqliteDownloadRepository(SqliteConnectionPool pool) : BaseSqliteRep
hasChanged |= await metadataCmd.ExecuteNonQueryAsync() > 0;
}
public Task AddIfNotNull(Data.Download? download) {
if (download != null) {
return Add(download);
}
else {
return Task.CompletedTask;
}
}
public void OnCommitted() {
if (hasChanged) {
repository.UpdateTotalCount();
@@ -361,14 +370,17 @@ sealed class SqliteDownloadRepository(SqliteConnectionPool pool) : BaseSqliteRep
UpdateTotalCount();
}
public async IAsyncEnumerable<Data.Download> FindAllDownloadableUrls([EnumeratorCancellation] CancellationToken cancellationToken = default) {
public async IAsyncEnumerable<FileUrl> FindReachableFiles([EnumeratorCancellation] CancellationToken cancellationToken) {
await using var conn = await pool.Take();
await using (var cmd = conn.Command("SELECT normalized_url, download_url, type, size FROM attachments")) {
await using (var cmd = conn.Command("SELECT type, normalized_url, download_url FROM attachments")) {
await using var reader = await cmd.ExecuteReaderAsync(cancellationToken);
while (await reader.ReadAsync(cancellationToken)) {
yield return DownloadLinkExtractor.FromAttachment(reader.GetString(0), reader.GetString(1), reader.IsDBNull(2) ? null : reader.GetString(2), reader.GetUint64(3));
string? type = reader.IsDBNull(0) ? null : reader.GetString(0);
string normalizedUrl = reader.GetString(1);
string downloadUrl = reader.GetString(2);
yield return new FileUrl(normalizedUrl, downloadUrl, type);
}
}
@@ -376,8 +388,7 @@ sealed class SqliteDownloadRepository(SqliteConnectionPool pool) : BaseSqliteRep
await using var reader = await cmd.ExecuteReaderAsync(cancellationToken);
while (await reader.ReadAsync(cancellationToken)) {
var result = await DownloadLinkExtractor.TryFromEmbedJson(reader.GetStream(0));
if (result is not null) {
if (await DownloadLinkExtractor.TryFromEmbedJson(reader.GetStream(0)) is {} result) {
yield return result;
}
}
@@ -387,7 +398,9 @@ sealed class SqliteDownloadRepository(SqliteConnectionPool pool) : BaseSqliteRep
await using var reader = await cmd.ExecuteReaderAsync(cancellationToken);
while (await reader.ReadAsync(cancellationToken)) {
yield return DownloadLinkExtractor.FromUserAvatar(reader.GetUint64(0), reader.GetString(1));
ulong id = reader.GetUint64(0);
string avatarHash = reader.GetString(1);
yield return DownloadLinkExtractor.UserAvatar(id, avatarHash);
}
}
@@ -395,7 +408,9 @@ sealed class SqliteDownloadRepository(SqliteConnectionPool pool) : BaseSqliteRep
await using var reader = await cmd.ExecuteReaderAsync(cancellationToken);
while (await reader.ReadAsync(cancellationToken)) {
yield return DownloadLinkExtractor.FromEmoji(reader.GetUint64(0), (EmojiFlags) reader.GetInt16(1));
ulong emojiId = reader.GetUint64(0);
EmojiFlags emojiFlags = (EmojiFlags) reader.GetInt16(1);
yield return DownloadLinkExtractor.Emoji(emojiId, emojiFlags);
}
}
}
@@ -10,6 +10,7 @@ using DHT.Server.Database.Repositories;
using DHT.Server.Database.Sqlite.Utils;
using DHT.Server.Download;
using DHT.Utils.Logging;
using JetBrains.Annotations;
using Microsoft.Data.Sqlite;
namespace DHT.Server.Database.Sqlite.Repositories;
@@ -17,15 +18,16 @@ namespace DHT.Server.Database.Sqlite.Repositories;
sealed class SqliteMessageRepository(SqliteConnectionPool pool, SqliteDownloadRepository downloads) : BaseSqliteRepository(Log), IMessageRepository {
private static readonly Log Log = Log.ForType<SqliteMessageRepository>();
// Moved outside the Add method due to language injections not working in local methods.
private static SqliteCommand DeleteByMessageId(ISqliteConnection conn, [LanguageInjection("sql", Prefix = "SELECT * FROM ")] string tableName) {
return conn.Delete(tableName, ("message_id", SqliteType.Integer));
}
public async Task Add(IReadOnlyList<Message> messages) {
if (messages.Count == 0) {
return;
}
static SqliteCommand DeleteByMessageId(ISqliteConnection conn, string tableName) {
return conn.Delete(tableName, ("message_id", SqliteType.Integer));
}
static async Task ExecuteDeleteByMessageId(SqliteCommand cmd, object id) {
cmd.Set(":message_id", id);
await cmd.ExecuteNonQueryAsync();
@@ -137,7 +139,7 @@ sealed class SqliteMessageRepository(SqliteConnectionPool pool, SqliteDownloadRe
messageAttachmentCmd.Set(":attachment_id", attachmentId);
await messageAttachmentCmd.ExecuteNonQueryAsync();
await downloadCollector.Add(DownloadLinkExtractor.FromAttachment(attachment));
await downloadCollector.Add(new Data.Download(attachment.NormalizedUrl, attachment.DownloadUrl, DownloadStatus.Pending, attachment.Type, attachment.Size));
}
}
@@ -148,7 +150,7 @@ sealed class SqliteMessageRepository(SqliteConnectionPool pool, SqliteDownloadRe
await messageEmbedCmd.ExecuteNonQueryAsync();
if (DownloadLinkExtractor.TryFromEmbedJson(embed.Json) is {} download) {
await downloadCollector.Add(download);
await downloadCollector.Add(download.ToPendingDownload());
}
}
}
@@ -163,7 +165,7 @@ sealed class SqliteMessageRepository(SqliteConnectionPool pool, SqliteDownloadRe
await messageReactionCmd.ExecuteNonQueryAsync();
if (reaction.EmojiId is {} emojiId) {
await downloadCollector.Add(DownloadLinkExtractor.FromEmoji(emojiId, reaction.EmojiFlags));
await downloadCollector.Add(DownloadLinkExtractor.Emoji(emojiId, reaction.EmojiFlags).ToPendingDownload());
}
}
}
@@ -5,23 +5,14 @@ using System.Threading.Tasks;
using DHT.Server.Data;
using DHT.Server.Database.Repositories;
using DHT.Server.Database.Sqlite.Utils;
using DHT.Server.Download;
using DHT.Utils.Logging;
using Microsoft.Data.Sqlite;
namespace DHT.Server.Database.Sqlite.Repositories;
sealed class SqliteUserRepository : BaseSqliteRepository, IUserRepository {
sealed class SqliteUserRepository(SqliteConnectionPool pool, SqliteDownloadRepository downloads) : BaseSqliteRepository(Log), IUserRepository {
private static readonly Log Log = Log.ForType<SqliteUserRepository>();
private readonly SqliteConnectionPool pool;
private readonly SqliteDownloadRepository downloads;
public SqliteUserRepository(SqliteConnectionPool pool, SqliteDownloadRepository downloads) : base(Log) {
this.pool = pool;
this.downloads = downloads;
}
public async Task Add(IReadOnlyList<User> users) {
await using (var conn = await pool.Take()) {
await conn.BeginTransactionAsync();
@@ -40,13 +31,10 @@ sealed class SqliteUserRepository : BaseSqliteRepository, IUserRepository {
cmd.Set(":id", user.Id);
cmd.Set(":name", user.Name);
cmd.Set(":display_name", user.DisplayName);
cmd.Set(":avatar_url", user.AvatarUrl);
cmd.Set(":avatar_url", user.AvatarHash);
cmd.Set(":discriminator", user.Discriminator);
await cmd.ExecuteNonQueryAsync();
if (user.AvatarUrl is {} avatarUrl) {
await downloadCollector.Add(DownloadLinkExtractor.FromUserAvatar(user.Id, avatarUrl));
}
await downloadCollector.AddIfNotNull(user.AvatarUrl?.ToPendingDownload());
}
await conn.CommitTransactionAsync();
@@ -72,7 +60,7 @@ sealed class SqliteUserRepository : BaseSqliteRepository, IUserRepository {
Id = reader.GetUint64(0),
Name = reader.GetString(1),
DisplayName = reader.IsDBNull(2) ? null : reader.GetString(2),
AvatarUrl = reader.IsDBNull(3) ? null : reader.GetString(3),
AvatarHash = reader.IsDBNull(3) ? null : reader.GetString(3),
Discriminator = reader.IsDBNull(4) ? null : reader.GetString(4),
};
}
@@ -124,7 +124,7 @@ sealed class SqliteSchemaUpgradeTo7 : ISchemaUpgrade {
await using var reader = await embedCmd.ExecuteReaderAsync();
while (await reader.ReadAsync()) {
await InsertDownload(insertCmd, await DownloadLinkExtractor.TryFromEmbedJson(reader.GetStream(0)));
await InsertDownload(insertCmd, (await DownloadLinkExtractor.TryFromEmbedJson(reader.GetStream(0)))?.ToPendingDownload());
}
}
@@ -134,7 +134,7 @@ sealed class SqliteSchemaUpgradeTo7 : ISchemaUpgrade {
await using var reader = await avatarCmd.ExecuteReaderAsync();
while (await reader.ReadAsync()) {
await InsertDownload(insertCmd, DownloadLinkExtractor.FromUserAvatar(reader.GetUint64(0), reader.GetString(1)));
await InsertDownload(insertCmd, DownloadLinkExtractor.UserAvatar(reader.GetUint64(0), reader.GetString(1)).ToPendingDownload());
}
}
@@ -144,7 +144,7 @@ sealed class SqliteSchemaUpgradeTo7 : ISchemaUpgrade {
await using var reader = await avatarCmd.ExecuteReaderAsync();
while (await reader.ReadAsync()) {
await InsertDownload(insertCmd, DownloadLinkExtractor.FromEmoji(reader.GetUint64(0), (EmojiFlags) reader.GetInt16(1)));
await InsertDownload(insertCmd, DownloadLinkExtractor.Emoji(reader.GetUint64(0), (EmojiFlags) reader.GetInt16(1)).ToPendingDownload());
}
}
@@ -30,7 +30,7 @@ static class SqliteExtensions {
return (long) (await command.ExecuteScalarAsync())!;
}
public static SqliteCommand Insert(this ISqliteConnection conn, string tableName, (string Name, SqliteType Type)[] columns) {
public static SqliteCommand Insert(this ISqliteConnection conn, [LanguageInjection("sql", Prefix = "SELECT * FROM ")] string tableName, (string Name, SqliteType Type)[] columns) {
string columnNames = string.Join(separator: ',', columns.Select(static c => c.Name));
string columnParams = string.Join(separator: ',', columns.Select(static c => ':' + c.Name));
@@ -41,7 +41,7 @@ static class SqliteExtensions {
return cmd;
}
public static SqliteCommand Upsert(this ISqliteConnection conn, string tableName, (string Name, SqliteType Type)[] columns) {
public static SqliteCommand Upsert(this ISqliteConnection conn, [LanguageInjection("sql", Prefix = "SELECT * FROM ")] string tableName, (string Name, SqliteType Type)[] columns) {
string columnNames = string.Join(separator: ',', columns.Select(static c => c.Name));
string columnParams = string.Join(separator: ',', columns.Select(static c => ':' + c.Name));
string columnUpdates = string.Join(separator: ',', columns.Skip(1).Select(static c => c.Name + " = excluded." + c.Name));
@@ -55,7 +55,7 @@ static class SqliteExtensions {
return cmd;
}
public static SqliteCommand Delete(this ISqliteConnection conn, string tableName, (string Name, SqliteType Type) column) {
public static SqliteCommand Delete(this ISqliteConnection conn, [LanguageInjection("sql", Prefix = "SELECT * FROM ")] string tableName, (string Name, SqliteType Type) column) {
var cmd = conn.Command("DELETE FROM " + tableName + " WHERE " + column.Name + " = :" + column.Name);
CreateParameters(cmd, [column]);
return cmd;
+16 -30
View File
@@ -12,30 +12,20 @@ namespace DHT.Server.Download;
static class DownloadLinkExtractor {
private static readonly Log Log = Log.ForType(typeof(DownloadLinkExtractor));
public static Data.Download FromUserAvatar(ulong userId, string avatarPath) {
string url = $"https://cdn.discordapp.com/avatars/{userId}/{avatarPath}.webp";
return new Data.Download(url, url, DownloadStatus.Pending, MediaTypeNames.Image.Webp, size: null);
public static FileUrl UserAvatar(ulong id, string avatarHash) {
return new FileUrl($"https://cdn.discordapp.com/avatars/{id}/{avatarHash}.webp", MediaTypeNames.Image.Webp);
}
public static Data.Download FromEmoji(ulong emojiId, EmojiFlags flags) {
public static FileUrl Emoji(ulong emojiId, EmojiFlags flags) {
bool isAnimated = flags.HasFlag(EmojiFlags.Animated);
string ext = isAnimated ? "gif" : "webp";
string type = isAnimated ? MediaTypeNames.Image.Gif : MediaTypeNames.Image.Webp;
string url = $"https://cdn.discordapp.com/emojis/{emojiId}.{ext}";
return new Data.Download(url, url, DownloadStatus.Pending, type, size: null);
return new FileUrl($"https://cdn.discordapp.com/emojis/{emojiId}.{ext}", type);
}
public static Data.Download FromAttachment(Attachment attachment) {
return FromAttachment(attachment.NormalizedUrl, attachment.DownloadUrl, attachment.Type, attachment.Size);
}
public static Data.Download FromAttachment(string normalizedUrl, string downloadUrl, string? type, ulong size) {
return new Data.Download(normalizedUrl, downloadUrl, DownloadStatus.Pending, type, size);
}
public static async Task<Data.Download?> TryFromEmbedJson(Stream jsonStream) {
public static async Task<FileUrl?> TryFromEmbedJson(Stream jsonStream) {
try {
return FromEmbed(await JsonSerializer.DeserializeAsync(jsonStream, DiscordEmbedJsonContext.Default.DiscordEmbedJson));
} catch (Exception e) {
@@ -44,7 +34,7 @@ static class DownloadLinkExtractor {
}
}
public static Data.Download? TryFromEmbedJson(string json) {
public static FileUrl? TryFromEmbedJson(string json) {
try {
return FromEmbed(JsonSerializer.Deserialize(json, DiscordEmbedJsonContext.Default.DiscordEmbedJson));
} catch (Exception e) {
@@ -53,21 +43,17 @@ static class DownloadLinkExtractor {
}
}
private static Data.Download? FromEmbed(DiscordEmbedJson? embed) {
if (embed is { Type: "image", Image.Url: {} imageUrl }) {
return FromEmbedImage(imageUrl);
}
else if (embed is { Type: "video", Video.Url: {} videoUrl }) {
return FromEmbedVideo(videoUrl);
}
else {
return null;
}
private static FileUrl? FromEmbed(DiscordEmbedJson? embed) {
return embed switch {
{ Type: "image", Image.Url: {} imageUrl } => FromEmbedImage(imageUrl),
{ Type: "video", Video.Url: {} videoUrl } => FromEmbedVideo(videoUrl),
_ => null,
};
}
private static Data.Download? FromEmbedImage(string url) {
private static FileUrl? FromEmbedImage(string url) {
if (DiscordCdn.NormalizeUrlAndReturnIfCdn(url, out string normalizedUrl)) {
return new Data.Download(normalizedUrl, url, DownloadStatus.Pending, GuessImageType(normalizedUrl), size: null);
return new FileUrl(normalizedUrl, url, GuessImageType(normalizedUrl));
}
else {
Log.Debug("Skipping non-CDN image url: " + url);
@@ -75,9 +61,9 @@ static class DownloadLinkExtractor {
}
}
private static Data.Download? FromEmbedVideo(string url) {
private static FileUrl? FromEmbedVideo(string url) {
if (DiscordCdn.NormalizeUrlAndReturnIfCdn(url, out string normalizedUrl)) {
return new Data.Download(normalizedUrl, url, DownloadStatus.Pending, GuessVideoType(normalizedUrl), size: null);
return new FileUrl(normalizedUrl, url, GuessVideoType(normalizedUrl));
}
else {
Log.Debug("Skipping non-CDN video url: " + url);
+1 -1
View File
@@ -32,7 +32,7 @@ sealed class TrackUsersEndpoint(IDatabaseFile db) : BaseEndpoint {
Id = json.RequireSnowflake("id", path),
Name = json.RequireString("name", path),
DisplayName = json.HasKey("displayName") ? json.RequireString("displayName", path) : null,
AvatarUrl = json.HasKey("avatar") ? json.RequireString("avatar", path) : null,
AvatarHash = json.HasKey("avatar") ? json.RequireString("avatar", path) : null,
Discriminator = json.HasKey("discriminator") ? json.RequireString("discriminator", path) : null,
};
}