Fix button to retry failed downloads & show error if downloads fail to start

app/Desktop/Common/DatabaseGui.cs

@@ -9,7 +9,7 @@ using DHT.Desktop.Dialogs.Message;
 using DHT.Server.Database;
 using DHT.Server.Database.Exceptions;
 using DHT.Server.Database.Sqlite;
-using DHT.Server.Database.Sqlite.Utils;
+using DHT.Server.Database.Sqlite.Schema;
 using DHT.Utils.Logging;
 
 namespace DHT.Desktop.Common;

app/Desktop/Main/Pages/AttachmentsPageModel.cs

@@ -3,9 +3,11 @@ using System.Collections.Generic;
 using System.Collections.ObjectModel;
 using System.Reactive.Linq;
 using System.Threading.Tasks;
+using Avalonia.Controls;
 using Avalonia.ReactiveUI;
 using CommunityToolkit.Mvvm.ComponentModel;
 using DHT.Desktop.Common;
+using DHT.Desktop.Dialogs.Message;
 using DHT.Desktop.Main.Controls;
 using DHT.Server;
 using DHT.Server.Data;
@@ -39,7 +41,7 @@ sealed partial class AttachmentsPageModel : ObservableObject, IDisposable {
 	[NotifyPropertyChangedFor(nameof(IsRetryFailedOnDownloadsButtonEnabled))]
 	private bool hasFailedDownloads;
 
-	public bool IsRetryFailedOnDownloadsButtonEnabled => !IsRetryingFailedDownloads && hasFailedDownloads;
+	public bool IsRetryFailedOnDownloadsButtonEnabled => !IsRetryingFailedDownloads && HasFailedDownloads;
 
 	[ObservableProperty(Setter = Access.Private)]
 	private string downloadMessage = "";
@@ -57,6 +59,7 @@ sealed partial class AttachmentsPageModel : ObservableObject, IDisposable {
 
 	public bool IsDownloading => state.Downloader.IsDownloading;
 
+	private readonly Window window;
 	private readonly State state;
 	private readonly ThrottledTask<int> enqueueDownloadItemsTask;
 	private readonly ThrottledTask<DownloadStatusStatistics> downloadStatisticsTask;
@@ -69,9 +72,10 @@ sealed partial class AttachmentsPageModel : ObservableObject, IDisposable {
 	private int totalEnqueuedItemCount;
 	private int? totalItemsToDownloadCount;
 
-	public AttachmentsPageModel() : this(State.Dummy) {}
+	public AttachmentsPageModel() : this(null!, State.Dummy) {}
 
-	public AttachmentsPageModel(State state) {
+	public AttachmentsPageModel(Window window, State state) {
+		this.window = window;
 		this.state = state;
 
 		FilterModel = new AttachmentFilterPanelModel(state);
@@ -117,7 +121,12 @@ sealed partial class AttachmentsPageModel : ObservableObject, IDisposable {
 	}
 
 	private async Task EnqueueDownloadItems() {
-		OnItemsEnqueued(await state.Db.Downloads.EnqueueDownloadItems(CreateAttachmentFilter()));
+		try {
+			OnItemsEnqueued(await state.Db.Downloads.EnqueueDownloadItems(CreateAttachmentFilter()));
+		} catch (Exception e) {
+			Log.Error(e);
+			await Dialog.ShowOk(window, "Download Error", "Failed to enqueue items for download.");
+		}
 	}
 
 	private void EnqueueDownloadItemsLater() {
@@ -220,7 +229,7 @@ sealed partial class AttachmentsPageModel : ObservableObject, IDisposable {
 		statisticsSkipped.Items = statusStatistics.SkippedCount;
 		statisticsSkipped.Size = statusStatistics.SkippedSize;
 
-		hasFailedDownloads = statusStatistics.FailedCount > 0;
+		HasFailedDownloads = statusStatistics.FailedCount > 0;
 
 		UpdateDownloadMessage();
 	}

app/Desktop/Main/Pages/DatabasePageModel.cs

@@ -18,7 +18,7 @@ using DHT.Server;
 using DHT.Server.Data;
 using DHT.Server.Database;
 using DHT.Server.Database.Import;
-using DHT.Server.Database.Sqlite.Utils;
+using DHT.Server.Database.Sqlite.Schema;
 using DHT.Utils.Logging;
 
 namespace DHT.Desktop.Main.Pages;

app/Desktop/Main/Screens/MainContentScreenModel.cs

@@ -52,7 +52,7 @@ sealed class MainContentScreenModel : IDisposable {
 		TrackingPageModel = new TrackingPageModel(window);
 		TrackingPage = new TrackingPage { DataContext = TrackingPageModel };
 
-		AttachmentsPageModel = new AttachmentsPageModel(state);
+		AttachmentsPageModel = new AttachmentsPageModel(window, state);
 		AttachmentsPage = new AttachmentsPage { DataContext = AttachmentsPageModel };
 
 		ViewerPageModel = new ViewerPageModel(window, state);

app/Desktop/Main/Screens/WelcomeScreenModel.cs

@@ -8,7 +8,7 @@ using DHT.Desktop.Common;
 using DHT.Desktop.Dialogs.Message;
 using DHT.Desktop.Dialogs.Progress;
 using DHT.Server.Database;
-using DHT.Server.Database.Sqlite.Utils;
+using DHT.Server.Database.Sqlite.Schema;
 
 namespace DHT.Desktop.Main.Screens;
 

app/Server/Database/Exceptions/DatabaseTooNewException.cs

@@ -5,9 +5,9 @@ namespace DHT.Server.Database.Exceptions;
 
 public sealed class DatabaseTooNewException : Exception {
 	public int DatabaseVersion { get; }
-	public int CurrentVersion => Schema.Version;
+	public int CurrentVersion => SqliteSchema.Version;
 
-	internal DatabaseTooNewException(int databaseVersion) : base("Database is too new: " + databaseVersion + " > " + Schema.Version) {
+	internal DatabaseTooNewException(int databaseVersion) : base("Database is too new: " + databaseVersion + " > " + SqliteSchema.Version) {
 		this.DatabaseVersion = databaseVersion;
 	}
 }

app/Server/Database/Sqlite/Repositories/SqliteChannelRepository.cs

@@ -57,7 +57,7 @@ sealed class SqliteChannelRepository : BaseSqliteRepository, IChannelRepository
 		await using var cmd = conn.Command("SELECT id, server, name, parent_id, position, topic, nsfw FROM channels");
 		await using var reader = await cmd.ExecuteReaderAsync();
 
-		while (reader.Read()) {
+		while (await reader.ReadAsync()) {
 			yield return new Channel {
 				Id = reader.GetUint64(0),
 				Server = reader.GetUint64(1),

app/Server/Database/Sqlite/Repositories/SqliteDownloadRepository.cs

@@ -59,7 +59,7 @@ sealed class SqliteDownloadRepository : BaseSqliteRepository, IDownloadRepositor
 
 			await using var reader = await cmd.ExecuteReaderAsync(cancellationToken);
 
-			if (reader.Read()) {
+			if (await reader.ReadAsync(cancellationToken)) {
 				result.SkippedCount = reader.GetInt32(0);
 				result.SkippedSize = reader.GetUint64(1);
 			}
@@ -85,7 +85,7 @@ sealed class SqliteDownloadRepository : BaseSqliteRepository, IDownloadRepositor
 
 			await using var reader = await cmd.ExecuteReaderAsync(cancellationToken);
 
-			if (reader.Read()) {
+			if (await reader.ReadAsync(cancellationToken)) {
 				result.EnqueuedCount = reader.GetInt32(0);
 				result.EnqueuedSize = reader.GetUint64(1);
 				result.SuccessfulCount = reader.GetInt32(2);
@@ -109,7 +109,7 @@ sealed class SqliteDownloadRepository : BaseSqliteRepository, IDownloadRepositor
 		await using var cmd = conn.Command("SELECT normalized_url, download_url, status, size FROM downloads");
 		await using var reader = await cmd.ExecuteReaderAsync();
 
-		while (reader.Read()) {
+		while (await reader.ReadAsync()) {
 			string normalizedUrl = reader.GetString(0);
 			string downloadUrl = reader.GetString(1);
 			var status = (DownloadStatus) reader.GetInt32(2);
@@ -127,7 +127,7 @@ sealed class SqliteDownloadRepository : BaseSqliteRepository, IDownloadRepositor
 
 		await using var reader = await cmd.ExecuteReaderAsync();
 
-		if (reader.Read() && !reader.IsDBNull(0)) {
+		if (await reader.ReadAsync() && !reader.IsDBNull(0)) {
 			return download.WithData((byte[]) reader["blob"]);
 		}
 		else {
@@ -151,7 +151,7 @@ sealed class SqliteDownloadRepository : BaseSqliteRepository, IDownloadRepositor
 
 		await using var reader = await cmd.ExecuteReaderAsync();
 
-		if (!reader.Read()) {
+		if (!await reader.ReadAsync()) {
 			return null;
 		}
 
@@ -189,7 +189,7 @@ sealed class SqliteDownloadRepository : BaseSqliteRepository, IDownloadRepositor
 
 			await using var reader = await cmd.ExecuteReaderAsync(cancellationToken);
 
-			while (reader.Read()) {
+			while (await reader.ReadAsync(cancellationToken)) {
 				found.Add(new DownloadItem {
 					NormalizedUrl = reader.GetString(0),
 					DownloadUrl = reader.GetString(1),

app/Server/Database/Sqlite/Repositories/SqliteServerRepository.cs

@@ -49,7 +49,7 @@ sealed class SqliteServerRepository : BaseSqliteRepository, IServerRepository {
 		await using var cmd = conn.Command("SELECT id, name, type FROM servers");
 		await using var reader = await cmd.ExecuteReaderAsync();
 
-		while (reader.Read()) {
+		while (await reader.ReadAsync()) {
 			yield return new Data.Server {
 				Id = reader.GetUint64(0),
 				Name = reader.GetString(1),

app/Server/Database/Sqlite/Repositories/SqliteUserRepository.cs

@@ -51,7 +51,7 @@ sealed class SqliteUserRepository : BaseSqliteRepository, IUserRepository {
 		await using var cmd = conn.Command("SELECT id, name, avatar_url, discriminator FROM users");
 		await using var reader = await cmd.ExecuteReaderAsync();
 
-		while (reader.Read()) {
+		while (await reader.ReadAsync()) {
 			yield return new User {
 				Id = reader.GetUint64(0),
 				Name = reader.GetString(1),

app/Server/Database/Sqlite/Schema.cs

@@ -1,358 +0,0 @@
-using System.Collections.Generic;
-using System.Data.Common;
-using System.Threading.Tasks;
-using DHT.Server.Database.Exceptions;
-using DHT.Server.Database.Sqlite.Utils;
-using DHT.Server.Download;
-using DHT.Utils.Logging;
-using Microsoft.Data.Sqlite;
-
-namespace DHT.Server.Database.Sqlite;
-
-sealed class Schema {
-	internal const int Version = 6;
-
-	private static readonly Log Log = Log.ForType<Schema>();
-
-	private readonly ISqliteConnection conn;
-
-	public Schema(ISqliteConnection conn) {
-		this.conn = conn;
-	}
-
-	public async Task<bool> Setup(ISchemaUpgradeCallbacks callbacks) {
-		await conn.ExecuteAsync("CREATE TABLE IF NOT EXISTS metadata (key TEXT PRIMARY KEY, value TEXT)");
-
-		var dbVersionStr = await conn.ExecuteReaderAsync("SELECT value FROM metadata WHERE key = 'version'", static reader => reader?.GetString(0));
-		if (dbVersionStr == null) {
-			await InitializeSchemas();
-		}
-		else if (!int.TryParse(dbVersionStr, out int dbVersion) || dbVersion < 1) {
-			throw new InvalidDatabaseVersionException(dbVersionStr);
-		}
-		else if (dbVersion > Version) {
-			throw new DatabaseTooNewException(dbVersion);
-		}
-		else if (dbVersion < Version) {
-			var proceed = await callbacks.CanUpgrade();
-			if (!proceed) {
-				return false;
-			}
-
-			await callbacks.Start(Version - dbVersion, async reporter => await UpgradeSchemas(dbVersion, reporter));
-		}
-
-		return true;
-	}
-
-	private async Task InitializeSchemas() {
-		await conn.ExecuteAsync("""
-		                        CREATE TABLE users (
-		                        	id            INTEGER PRIMARY KEY NOT NULL,
-		                        	name          TEXT NOT NULL,
-		                        	avatar_url    TEXT,
-		                        	discriminator TEXT
-		                        )
-		                        """);
-
-		await conn.ExecuteAsync("""
-		                        CREATE TABLE servers (
-		                        	id   INTEGER PRIMARY KEY NOT NULL,
-		                        	name TEXT NOT NULL,
-		                        	type TEXT NOT NULL
-		                        )
-		                        """);
-
-		await conn.ExecuteAsync("""
-		                        CREATE TABLE channels (
-		                        	id        INTEGER PRIMARY KEY NOT NULL,
-		                        	server    INTEGER NOT NULL,
-		                        	name      TEXT NOT NULL,
-		                        	parent_id INTEGER,
-		                        	position  INTEGER,
-		                        	topic     TEXT,
-		                        	nsfw      INTEGER
-		                        )
-		                        """);
-
-		await conn.ExecuteAsync("""
-		                        CREATE TABLE messages (
-		                        	message_id INTEGER PRIMARY KEY NOT NULL,
-		                        	sender_id  INTEGER NOT NULL,
-		                        	channel_id INTEGER NOT NULL,
-		                        	text       TEXT NOT NULL,
-		                        	timestamp  INTEGER NOT NULL
-		                        )
-		                        """);
-
-		await conn.ExecuteAsync("""
-		                        CREATE TABLE attachments (
-		                        	message_id     INTEGER NOT NULL,
-		                        	attachment_id  INTEGER NOT NULL PRIMARY KEY NOT NULL,
-		                        	name           TEXT NOT NULL,
-		                        	type           TEXT,
-		                        	normalized_url TEXT NOT NULL,
-		                        	download_url   TEXT,
-		                        	size           INTEGER NOT NULL,
-		                        	width          INTEGER,
-		                        	height         INTEGER
-		                        )
-		                        """);
-
-		await conn.ExecuteAsync("""
-		                        CREATE TABLE embeds (
-		                        	message_id INTEGER NOT NULL,
-		                        	json       TEXT NOT NULL
-		                        )
-		                        """);
-
-		await conn.ExecuteAsync("""
-		                        CREATE TABLE downloads (
-		                        	normalized_url TEXT NOT NULL PRIMARY KEY,
-		                        	download_url   TEXT,
-		                        	status         INTEGER NOT NULL,
-		                        	size           INTEGER NOT NULL,
-		                        	blob           BLOB
-		                        )
-		                        """);
-
-		await conn.ExecuteAsync("""
-		                        CREATE TABLE reactions (
-		                        	message_id  INTEGER NOT NULL,
-		                        	emoji_id    INTEGER,
-		                        	emoji_name  TEXT,
-		                        	emoji_flags INTEGER NOT NULL,
-		                        	count       INTEGER NOT NULL
-		                        )
-		                        """);
-
-		await CreateMessageEditTimestampTable();
-		await CreateMessageRepliedToTable();
-
-		await conn.ExecuteAsync("CREATE INDEX attachments_message_ix ON attachments(message_id)");
-		await conn.ExecuteAsync("CREATE INDEX embeds_message_ix ON embeds(message_id)");
-		await conn.ExecuteAsync("CREATE INDEX reactions_message_ix ON reactions(message_id)");
-
-		await conn.ExecuteAsync("INSERT INTO metadata (key, value) VALUES ('version', " + Version + ")");
-	}
-
-	private async Task CreateMessageEditTimestampTable() {
-		await conn.ExecuteAsync("""
-		                        CREATE TABLE edit_timestamps (
-		                        	message_id     INTEGER PRIMARY KEY NOT NULL,
-		                        	edit_timestamp INTEGER NOT NULL
-		                        )
-		                        """);
-	}
-
-	private async Task CreateMessageRepliedToTable() {
-		await conn.ExecuteAsync("""
-		                        CREATE TABLE replied_to (
-		                        	message_id    INTEGER PRIMARY KEY NOT NULL,
-		                        	replied_to_id INTEGER NOT NULL
-		                        )
-		                        """);
-	}
-
-	private async Task NormalizeAttachmentUrls(ISchemaUpgradeCallbacks.IProgressReporter reporter) {
-		await reporter.SubWork("Preparing attachments...", 0, 0);
-
-		var normalizedUrls = new Dictionary<long, string>();
-
-		await using (var selectCmd = conn.Command("SELECT attachment_id, url FROM attachments")) {
-			await using var reader = await selectCmd.ExecuteReaderAsync();
-
-			while (reader.Read()) {
-				var attachmentId = reader.GetInt64(0);
-				var originalUrl = reader.GetString(1);
-				normalizedUrls[attachmentId] = DiscordCdn.NormalizeUrl(originalUrl);
-			}
-		}
-
-		await using var tx = await conn.BeginTransactionAsync();
-
-		int totalUrls = normalizedUrls.Count;
-		int processedUrls = -1;
-
-		await using (var updateCmd = conn.Command("UPDATE attachments SET download_url = url, url = :normalized_url WHERE attachment_id = :attachment_id")) {
-			updateCmd.Add(":attachment_id", SqliteType.Integer);
-			updateCmd.Add(":normalized_url", SqliteType.Text);
-
-			foreach (var (attachmentId, normalizedUrl) in normalizedUrls) {
-				if (++processedUrls % 1000 == 0) {
-					await reporter.SubWork("Updating URLs...", processedUrls, totalUrls);
-				}
-
-				updateCmd.Set(":attachment_id", attachmentId);
-				updateCmd.Set(":normalized_url", normalizedUrl);
-				updateCmd.ExecuteNonQuery();
-			}
-		}
-
-		await reporter.SubWork("Updating URLs...", totalUrls, totalUrls);
-
-		await tx.CommitAsync();
-	}
-
-	private async Task NormalizeDownloadUrls(ISchemaUpgradeCallbacks.IProgressReporter reporter) {
-		await reporter.SubWork("Preparing downloads...", 0, 0);
-
-		var normalizedUrlsToOriginalUrls = new Dictionary<string, string>();
-		var duplicateUrlsToDelete = new HashSet<string>();
-
-		await using (var selectCmd = conn.Command("SELECT url FROM downloads ORDER BY CASE WHEN status = 200 THEN 0 ELSE 1 END")) {
-			await using var reader = await selectCmd.ExecuteReaderAsync();
-
-			while (reader.Read()) {
-				var originalUrl = reader.GetString(0);
-				var normalizedUrl = DiscordCdn.NormalizeUrl(originalUrl);
-
-				if (!normalizedUrlsToOriginalUrls.TryAdd(normalizedUrl, originalUrl)) {
-					duplicateUrlsToDelete.Add(originalUrl);
-				}
-			}
-		}
-
-		await conn.ExecuteAsync("PRAGMA cache_size = -20000");
-
-		DbTransaction tx;
-
-		await using (tx = await conn.BeginTransactionAsync()) {
-			await reporter.SubWork("Deleting duplicates...", 0, 0);
-
-			await using (var deleteCmd = conn.Delete("downloads", ("url", SqliteType.Text))) {
-				foreach (var duplicateUrl in duplicateUrlsToDelete) {
-					deleteCmd.Set(":url", duplicateUrl);
-					deleteCmd.ExecuteNonQuery();
-				}
-			}
-
-			await tx.CommitAsync();
-		}
-
-		int totalUrls = normalizedUrlsToOriginalUrls.Count;
-		int processedUrls = -1;
-
-		tx = await conn.BeginTransactionAsync();
-
-		await using (var updateCmd = conn.Command("UPDATE downloads SET download_url = :download_url, url = :normalized_url WHERE url = :download_url")) {
-			updateCmd.Add(":normalized_url", SqliteType.Text);
-			updateCmd.Add(":download_url", SqliteType.Text);
-
-			foreach (var (normalizedUrl, downloadUrl) in normalizedUrlsToOriginalUrls) {
-				if (++processedUrls % 100 == 0) {
-					await reporter.SubWork("Updating URLs...", processedUrls, totalUrls);
-
-					// Not proper way of dealing with transactions, but it avoids a long commit at the end.
-					// Schema upgrades are already non-atomic anyways, so this doesn't make it worse.
-					await tx.CommitAsync();
-					await tx.DisposeAsync();
-
-					tx = await conn.BeginTransactionAsync();
-					updateCmd.Transaction = (SqliteTransaction) tx;
-				}
-
-				updateCmd.Set(":normalized_url", normalizedUrl);
-				updateCmd.Set(":download_url", downloadUrl);
-				updateCmd.ExecuteNonQuery();
-			}
-		}
-
-		await reporter.SubWork("Updating URLs...", totalUrls, totalUrls);
-
-		await tx.CommitAsync();
-		await tx.DisposeAsync();
-
-		await conn.ExecuteAsync("PRAGMA cache_size = -2000");
-	}
-
-	private async Task UpgradeSchemas(int dbVersion, ISchemaUpgradeCallbacks.IProgressReporter reporter) {
-		var perf = Log.Start("from version " + dbVersion);
-
-		await conn.ExecuteAsync("UPDATE metadata SET value = " + Version + " WHERE key = 'version'");
-
-		if (dbVersion <= 1) {
-			await reporter.MainWork("Applying schema changes...", 0, 1);
-			await conn.ExecuteAsync("ALTER TABLE channels ADD parent_id INTEGER");
-
-			perf.Step("Upgrade to version 2");
-			await reporter.NextVersion();
-		}
-
-		if (dbVersion <= 2) {
-			await reporter.MainWork("Applying schema changes...", 0, 1);
-
-			await CreateMessageEditTimestampTable();
-			await CreateMessageRepliedToTable();
-
-			await conn.ExecuteAsync("""
-			                        INSERT INTO edit_timestamps (message_id, edit_timestamp)
-			                        SELECT message_id, edit_timestamp
-			                        FROM messages
-			                        WHERE edit_timestamp IS NOT NULL
-			                        """);
-
-			await conn.ExecuteAsync("""
-			                        INSERT INTO replied_to (message_id, replied_to_id)
-			                        SELECT message_id, replied_to_id
-			                        FROM messages
-			                        WHERE replied_to_id IS NOT NULL
-			                        """);
-
-			await conn.ExecuteAsync("ALTER TABLE messages DROP COLUMN replied_to_id");
-			await conn.ExecuteAsync("ALTER TABLE messages DROP COLUMN edit_timestamp");
-
-			perf.Step("Upgrade to version 3");
-
-			await reporter.MainWork("Vacuuming the database...", 1, 1);
-			await conn.ExecuteAsync("VACUUM");
-			perf.Step("Vacuum");
-
-			await reporter.NextVersion();
-		}
-
-		if (dbVersion <= 3) {
-			await conn.ExecuteAsync("""
-			                        CREATE TABLE downloads (
-			                        	url    TEXT NOT NULL PRIMARY KEY,
-			                        	status INTEGER NOT NULL,
-			                        	size   INTEGER NOT NULL,
-			                        	blob   BLOB
-			                        )
-			                        """);
-
-			perf.Step("Upgrade to version 4");
-			await reporter.NextVersion();
-		}
-
-		if (dbVersion <= 4) {
-			await reporter.MainWork("Applying schema changes...", 0, 1);
-			await conn.ExecuteAsync("ALTER TABLE attachments ADD width INTEGER");
-			await conn.ExecuteAsync("ALTER TABLE attachments ADD height INTEGER");
-
-			perf.Step("Upgrade to version 5");
-			await reporter.NextVersion();
-		}
-
-		if (dbVersion <= 5) {
-			await reporter.MainWork("Applying schema changes...", 0, 3);
-			await conn.ExecuteAsync("ALTER TABLE attachments ADD download_url TEXT");
-			await conn.ExecuteAsync("ALTER TABLE downloads ADD download_url TEXT");
-
-			await reporter.MainWork("Updating attachments...", 1, 3);
-			await NormalizeAttachmentUrls(reporter);
-
-			await reporter.MainWork("Updating downloads...", 2, 3);
-			await NormalizeDownloadUrls(reporter);
-
-			await reporter.MainWork("Applying schema changes...", 3, 3);
-			await conn.ExecuteAsync("ALTER TABLE attachments RENAME COLUMN url TO normalized_url");
-			await conn.ExecuteAsync("ALTER TABLE downloads RENAME COLUMN url TO normalized_url");
-
-			perf.Step("Upgrade to version 6");
-			await reporter.NextVersion();
-		}
-
-		perf.End();
-	}
-}

app/Server/Database/Sqlite/Schema/ISchemaUpgrade.cs

@@ -0,0 +1,8 @@
+using System.Threading.Tasks;
+using DHT.Server.Database.Sqlite.Utils;
+
+namespace DHT.Server.Database.Sqlite.Schema;
+
+interface ISchemaUpgrade {
+	Task Run(ISqliteConnection conn, ISchemaUpgradeCallbacks.IProgressReporter reporter);
+}

app/Server/Database/Sqlite/Schema/ISchemaUpgradeCallbacks.cs

@@ -1,7 +1,7 @@
 using System;
 using System.Threading.Tasks;
 
-namespace DHT.Server.Database.Sqlite.Utils;
+namespace DHT.Server.Database.Sqlite.Schema;
 
 public interface ISchemaUpgradeCallbacks {
 	Task<bool> CanUpgrade();

app/Server/Database/Sqlite/Schema/SqliteSchemaUpgradeTo2.cs

@@ -0,0 +1,11 @@
+using System.Threading.Tasks;
+using DHT.Server.Database.Sqlite.Utils;
+
+namespace DHT.Server.Database.Sqlite.Schema;
+
+sealed class SqliteSchemaUpgradeTo2 : ISchemaUpgrade {
+	async Task ISchemaUpgrade.Run(ISqliteConnection conn, ISchemaUpgradeCallbacks.IProgressReporter reporter) {
+		await reporter.MainWork("Applying schema changes...", 0, 1);
+		await conn.ExecuteAsync("ALTER TABLE channels ADD parent_id INTEGER");
+	}
+}

app/Server/Database/Sqlite/Schema/SqliteSchemaUpgradeTo3.cs

@@ -0,0 +1,33 @@
+using System.Threading.Tasks;
+using DHT.Server.Database.Sqlite.Utils;
+
+namespace DHT.Server.Database.Sqlite.Schema;
+
+sealed class SqliteSchemaUpgradeTo3 : ISchemaUpgrade {
+	async Task ISchemaUpgrade.Run(ISqliteConnection conn, ISchemaUpgradeCallbacks.IProgressReporter reporter) {
+		await reporter.MainWork("Applying schema changes...", 0, 1);
+
+		await SqliteSchema.CreateMessageEditTimestampTable(conn);
+		await SqliteSchema.CreateMessageRepliedToTable(conn);
+
+		await conn.ExecuteAsync("""
+		                        INSERT INTO edit_timestamps (message_id, edit_timestamp)
+		                        SELECT message_id, edit_timestamp
+		                        FROM messages
+		                        WHERE edit_timestamp IS NOT NULL
+		                        """);
+
+		await conn.ExecuteAsync("""
+		                        INSERT INTO replied_to (message_id, replied_to_id)
+		                        SELECT message_id, replied_to_id
+		                        FROM messages
+		                        WHERE replied_to_id IS NOT NULL
+		                        """);
+
+		await conn.ExecuteAsync("ALTER TABLE messages DROP COLUMN replied_to_id");
+		await conn.ExecuteAsync("ALTER TABLE messages DROP COLUMN edit_timestamp");
+
+		await reporter.MainWork("Vacuuming the database...", 1, 1);
+		await conn.ExecuteAsync("VACUUM");
+	}
+}

app/Server/Database/Sqlite/Schema/SqliteSchemaUpgradeTo4.cs

@@ -0,0 +1,19 @@
+using System.Threading.Tasks;
+using DHT.Server.Database.Sqlite.Utils;
+
+namespace DHT.Server.Database.Sqlite.Schema;
+
+sealed class SqliteSchemaUpgradeTo4 : ISchemaUpgrade {
+	async Task ISchemaUpgrade.Run(ISqliteConnection conn, ISchemaUpgradeCallbacks.IProgressReporter reporter) {
+		await reporter.MainWork("Applying schema changes...", 0, 1);
+
+		await conn.ExecuteAsync("""
+		                        CREATE TABLE downloads (
+		                        	url    TEXT NOT NULL PRIMARY KEY,
+		                        	status INTEGER NOT NULL,
+		                        	size   INTEGER NOT NULL,
+		                        	blob   BLOB
+		                        )
+		                        """);
+	}
+}

app/Server/Database/Sqlite/Schema/SqliteSchemaUpgradeTo5.cs

@@ -0,0 +1,12 @@
+using System.Threading.Tasks;
+using DHT.Server.Database.Sqlite.Utils;
+
+namespace DHT.Server.Database.Sqlite.Schema;
+
+sealed class SqliteSchemaUpgradeTo5 : ISchemaUpgrade {
+	async Task ISchemaUpgrade.Run(ISqliteConnection conn, ISchemaUpgradeCallbacks.IProgressReporter reporter) {
+		await reporter.MainWork("Applying schema changes...", 0, 1);
+		await conn.ExecuteAsync("ALTER TABLE attachments ADD width INTEGER");
+		await conn.ExecuteAsync("ALTER TABLE attachments ADD height INTEGER");
+	}
+}

app/Server/Database/Sqlite/Schema/SqliteSchemaUpgradeTo6.cs

@@ -0,0 +1,138 @@
+using System.Collections.Generic;
+using System.Data.Common;
+using System.Threading.Tasks;
+using DHT.Server.Database.Sqlite.Utils;
+using DHT.Server.Download;
+using Microsoft.Data.Sqlite;
+
+namespace DHT.Server.Database.Sqlite.Schema;
+
+sealed class SqliteSchemaUpgradeTo6 : ISchemaUpgrade {
+	async Task ISchemaUpgrade.Run(ISqliteConnection conn, ISchemaUpgradeCallbacks.IProgressReporter reporter) {
+		await reporter.MainWork("Applying schema changes...", 0, 3);
+		await conn.ExecuteAsync("ALTER TABLE attachments ADD download_url TEXT");
+		await conn.ExecuteAsync("ALTER TABLE downloads ADD download_url TEXT");
+
+		await reporter.MainWork("Updating attachments...", 1, 3);
+		await NormalizeAttachmentUrls(conn, reporter);
+
+		await reporter.MainWork("Updating downloads...", 2, 3);
+		await NormalizeDownloadUrls(conn, reporter);
+
+		await reporter.MainWork("Applying schema changes...", 3, 3);
+		await conn.ExecuteAsync("ALTER TABLE attachments RENAME COLUMN url TO normalized_url");
+		await conn.ExecuteAsync("ALTER TABLE downloads RENAME COLUMN url TO normalized_url");
+	}
+	
+	private async Task NormalizeAttachmentUrls(ISqliteConnection conn, ISchemaUpgradeCallbacks.IProgressReporter reporter) {
+		await reporter.SubWork("Preparing attachments...", 0, 0);
+
+		var normalizedUrls = new Dictionary<long, string>();
+
+		await using (var selectCmd = conn.Command("SELECT attachment_id, url FROM attachments")) {
+			await using var reader = await selectCmd.ExecuteReaderAsync();
+
+			while (await reader.ReadAsync()) {
+				var attachmentId = reader.GetInt64(0);
+				var originalUrl = reader.GetString(1);
+				normalizedUrls[attachmentId] = DiscordCdn.NormalizeUrl(originalUrl);
+			}
+		}
+
+		await using var tx = await conn.BeginTransactionAsync();
+
+		int totalUrls = normalizedUrls.Count;
+		int processedUrls = -1;
+
+		await using (var updateCmd = conn.Command("UPDATE attachments SET download_url = url, url = :normalized_url WHERE attachment_id = :attachment_id")) {
+			updateCmd.Add(":attachment_id", SqliteType.Integer);
+			updateCmd.Add(":normalized_url", SqliteType.Text);
+
+			foreach (var (attachmentId, normalizedUrl) in normalizedUrls) {
+				if (++processedUrls % 1000 == 0) {
+					await reporter.SubWork("Updating URLs...", processedUrls, totalUrls);
+				}
+
+				updateCmd.Set(":attachment_id", attachmentId);
+				updateCmd.Set(":normalized_url", normalizedUrl);
+				await updateCmd.ExecuteNonQueryAsync();
+			}
+		}
+
+		await reporter.SubWork("Updating URLs...", totalUrls, totalUrls);
+
+		await tx.CommitAsync();
+	}
+
+	private async Task NormalizeDownloadUrls(ISqliteConnection conn, ISchemaUpgradeCallbacks.IProgressReporter reporter) {
+		await reporter.SubWork("Preparing downloads...", 0, 0);
+
+		var normalizedUrlsToOriginalUrls = new Dictionary<string, string>();
+		var duplicateUrlsToDelete = new HashSet<string>();
+
+		await using (var selectCmd = conn.Command("SELECT url FROM downloads ORDER BY CASE WHEN status = 200 THEN 0 ELSE 1 END")) {
+			await using var reader = await selectCmd.ExecuteReaderAsync();
+
+			while (await reader.ReadAsync()) {
+				var originalUrl = reader.GetString(0);
+				var normalizedUrl = DiscordCdn.NormalizeUrl(originalUrl);
+
+				if (!normalizedUrlsToOriginalUrls.TryAdd(normalizedUrl, originalUrl)) {
+					duplicateUrlsToDelete.Add(originalUrl);
+				}
+			}
+		}
+
+		await conn.ExecuteAsync("PRAGMA cache_size = -20000");
+
+		DbTransaction tx;
+
+		await using (tx = await conn.BeginTransactionAsync()) {
+			await reporter.SubWork("Deleting duplicates...", 0, 0);
+
+			await using (var deleteCmd = conn.Delete("downloads", ("url", SqliteType.Text))) {
+				foreach (var duplicateUrl in duplicateUrlsToDelete) {
+					deleteCmd.Set(":url", duplicateUrl);
+					await deleteCmd.ExecuteNonQueryAsync();
+				}
+			}
+
+			await tx.CommitAsync();
+		}
+
+		int totalUrls = normalizedUrlsToOriginalUrls.Count;
+		int processedUrls = -1;
+
+		tx = await conn.BeginTransactionAsync();
+
+		await using (var updateCmd = conn.Command("UPDATE downloads SET download_url = :download_url, url = :normalized_url WHERE url = :download_url")) {
+			updateCmd.Add(":normalized_url", SqliteType.Text);
+			updateCmd.Add(":download_url", SqliteType.Text);
+
+			foreach (var (normalizedUrl, downloadUrl) in normalizedUrlsToOriginalUrls) {
+				if (++processedUrls % 100 == 0) {
+					await reporter.SubWork("Updating URLs...", processedUrls, totalUrls);
+
+					// Not proper way of dealing with transactions, but it avoids a long commit at the end.
+					// Schema upgrades are already non-atomic anyways, so this doesn't make it worse.
+					await tx.CommitAsync();
+					await tx.DisposeAsync();
+
+					tx = await conn.BeginTransactionAsync();
+					updateCmd.Transaction = (SqliteTransaction) tx;
+				}
+
+				updateCmd.Set(":normalized_url", normalizedUrl);
+				updateCmd.Set(":download_url", downloadUrl);
+				await updateCmd.ExecuteNonQueryAsync();
+			}
+		}
+
+		await reporter.SubWork("Updating URLs...", totalUrls, totalUrls);
+
+		await tx.CommitAsync();
+		await tx.DisposeAsync();
+
+		await conn.ExecuteAsync("PRAGMA cache_size = -2000");
+	}
+}

app/Server/Database/Sqlite/SqliteDatabaseFile.cs

@@ -2,6 +2,7 @@ using System;
 using System.Threading.Tasks;
 using DHT.Server.Database.Repositories;
 using DHT.Server.Database.Sqlite.Repositories;
+using DHT.Server.Database.Sqlite.Schema;
 using DHT.Server.Database.Sqlite.Utils;
 using Microsoft.Data.Sqlite;
 
@@ -21,7 +22,7 @@ public sealed class SqliteDatabaseFile : IDatabaseFile {
 
 		try {
 			await using var conn = await pool.Take();
-			wasOpened = await new Schema(conn).Setup(schemaUpgradeCallbacks);
+			wasOpened = await new SqliteSchema(conn).Setup(schemaUpgradeCallbacks);
 		} catch (Exception) {
 			await pool.DisposeAsync();
 			throw;

app/Server/Database/Sqlite/SqliteSchema.cs

@@ -0,0 +1,191 @@
+using System.Collections.Generic;
+using System.Threading.Tasks;
+using DHT.Server.Database.Exceptions;
+using DHT.Server.Database.Sqlite.Schema;
+using DHT.Server.Database.Sqlite.Utils;
+using DHT.Utils.Logging;
+
+namespace DHT.Server.Database.Sqlite;
+
+sealed class SqliteSchema {
+	internal const int Version = 6;
+
+	private static readonly Log Log = Log.ForType<SqliteSchema>();
+
+	private readonly ISqliteConnection conn;
+
+	public SqliteSchema(ISqliteConnection conn) {
+		this.conn = conn;
+	}
+
+	public async Task<bool> Setup(ISchemaUpgradeCallbacks callbacks) {
+		await conn.ExecuteAsync("CREATE TABLE IF NOT EXISTS metadata (key TEXT PRIMARY KEY, value TEXT)");
+
+		var dbVersionStr = await conn.ExecuteReaderAsync("SELECT value FROM metadata WHERE key = 'version'", static reader => reader?.GetString(0));
+		if (dbVersionStr == null) {
+			await InitializeSchemas();
+		}
+		else if (!int.TryParse(dbVersionStr, out int dbVersion) || dbVersion < 1) {
+			throw new InvalidDatabaseVersionException(dbVersionStr);
+		}
+		else if (dbVersion > Version) {
+			throw new DatabaseTooNewException(dbVersion);
+		}
+		else if (dbVersion < Version) {
+			var proceed = await callbacks.CanUpgrade();
+			if (!proceed) {
+				return false;
+			}
+
+			await callbacks.Start(Version - dbVersion, async reporter => await UpgradeSchemas(dbVersion, reporter));
+		}
+
+		return true;
+	}
+
+	private async Task InitializeSchemas() {
+		await conn.ExecuteAsync("""
+		                        CREATE TABLE users (
+		                        	id            INTEGER PRIMARY KEY NOT NULL,
+		                        	name          TEXT NOT NULL,
+		                        	avatar_url    TEXT,
+		                        	discriminator TEXT
+		                        )
+		                        """);
+
+		await conn.ExecuteAsync("""
+		                        CREATE TABLE servers (
+		                        	id   INTEGER PRIMARY KEY NOT NULL,
+		                        	name TEXT NOT NULL,
+		                        	type TEXT NOT NULL
+		                        )
+		                        """);
+
+		await conn.ExecuteAsync("""
+		                        CREATE TABLE channels (
+		                        	id        INTEGER PRIMARY KEY NOT NULL,
+		                        	server    INTEGER NOT NULL,
+		                        	name      TEXT NOT NULL,
+		                        	parent_id INTEGER,
+		                        	position  INTEGER,
+		                        	topic     TEXT,
+		                        	nsfw      INTEGER
+		                        )
+		                        """);
+
+		await conn.ExecuteAsync("""
+		                        CREATE TABLE messages (
+		                        	message_id INTEGER PRIMARY KEY NOT NULL,
+		                        	sender_id  INTEGER NOT NULL,
+		                        	channel_id INTEGER NOT NULL,
+		                        	text       TEXT NOT NULL,
+		                        	timestamp  INTEGER NOT NULL
+		                        )
+		                        """);
+
+		await conn.ExecuteAsync("""
+		                        CREATE TABLE attachments (
+		                        	message_id     INTEGER NOT NULL,
+		                        	attachment_id  INTEGER NOT NULL PRIMARY KEY NOT NULL,
+		                        	name           TEXT NOT NULL,
+		                        	type           TEXT,
+		                        	normalized_url TEXT NOT NULL,
+		                        	download_url   TEXT,
+		                        	size           INTEGER NOT NULL,
+		                        	width          INTEGER,
+		                        	height         INTEGER
+		                        )
+		                        """);
+
+		await conn.ExecuteAsync("""
+		                        CREATE TABLE embeds (
+		                        	message_id INTEGER NOT NULL,
+		                        	json       TEXT NOT NULL
+		                        )
+		                        """);
+
+		await conn.ExecuteAsync("""
+		                        CREATE TABLE reactions (
+		                        	message_id  INTEGER NOT NULL,
+		                        	emoji_id    INTEGER,
+		                        	emoji_name  TEXT,
+		                        	emoji_flags INTEGER NOT NULL,
+		                        	count       INTEGER NOT NULL
+		                        )
+		                        """);
+
+		await CreateMessageEditTimestampTable(conn);
+		await CreateMessageRepliedToTable(conn);
+
+		await conn.ExecuteAsync("""
+		                        CREATE TABLE downloads (
+		                        	normalized_url TEXT NOT NULL PRIMARY KEY,
+		                        	download_url   TEXT,
+		                        	status         INTEGER NOT NULL,
+		                        	size           INTEGER NOT NULL,
+		                        	blob           BLOB
+		                        )
+		                        """);
+
+		await conn.ExecuteAsync("""
+		                        CREATE TABLE reactions (
+		                        	message_id  INTEGER NOT NULL,
+		                        	emoji_id    INTEGER,
+		                        	emoji_name  TEXT,
+		                        	emoji_flags INTEGER NOT NULL,
+		                        	count       INTEGER NOT NULL
+		                        )
+		                        """);
+
+		await conn.ExecuteAsync("CREATE INDEX attachments_message_ix ON attachments(message_id)");
+		await conn.ExecuteAsync("CREATE INDEX embeds_message_ix ON embeds(message_id)");
+		await conn.ExecuteAsync("CREATE INDEX reactions_message_ix ON reactions(message_id)");
+
+		await conn.ExecuteAsync("INSERT INTO metadata (key, value) VALUES ('version', " + Version + ")");
+	}
+
+	internal static async Task CreateMessageEditTimestampTable(ISqliteConnection conn) {
+		await conn.ExecuteAsync("""
+		                        CREATE TABLE edit_timestamps (
+		                        	message_id     INTEGER PRIMARY KEY NOT NULL,
+		                        	edit_timestamp INTEGER NOT NULL
+		                        )
+		                        """);
+	}
+
+	internal static async Task CreateMessageRepliedToTable(ISqliteConnection conn) {
+		await conn.ExecuteAsync("""
+		                        CREATE TABLE replied_to (
+		                        	message_id    INTEGER PRIMARY KEY NOT NULL,
+		                        	replied_to_id INTEGER NOT NULL
+		                        )
+		                        """);
+	}
+
+	private async Task UpgradeSchemas(int dbVersion, ISchemaUpgradeCallbacks.IProgressReporter reporter) {
+		var upgrades = new Dictionary<int, ISchemaUpgrade> {
+			{ 1, new SqliteSchemaUpgradeTo2() },
+			{ 2, new SqliteSchemaUpgradeTo3() },
+			{ 3, new SqliteSchemaUpgradeTo4() },
+			{ 4, new SqliteSchemaUpgradeTo5() },
+			{ 5, new SqliteSchemaUpgradeTo6() },
+		};
+
+		var perf = Log.Start("from version " + dbVersion);
+
+		for (int fromVersion = dbVersion; fromVersion < Version; fromVersion++) {
+			var toVersion = fromVersion + 1;
+			
+			if (upgrades.TryGetValue(fromVersion, out var upgrade)) {
+				await upgrade.Run(conn, reporter);
+			}
+
+			await conn.ExecuteAsync("UPDATE metadata SET value = " + toVersion + " WHERE key = 'version'");
+
+			perf.Step("Upgrade to version " + toVersion);
+			await reporter.NextVersion();
+		}
+
+		perf.End();
+	}
+}

app/Server/Database/Sqlite/Utils/SqliteExtensions.cs

@@ -28,7 +28,7 @@ static class SqliteExtensions {
 		await using var cmd = conn.Command(sql);
 		await using var reader = await cmd.ExecuteReaderAsync(cancellationToken);
 
-		return reader.Read() ? readFunction(reader) : readFunction(null);
+		return await reader.ReadAsync(cancellationToken) ? readFunction(reader) : readFunction(null);
 	}
 
 	public static SqliteCommand Insert(this ISqliteConnection conn, string tableName, (string Name, SqliteType Type)[] columns) {