Fix merging databases causing random files to be re-downloaded, if automatic downloads are enabled

Closes #295

app/Desktop/Main/Pages/DatabasePageModel.cs

@@ -55,11 +55,11 @@ sealed class DatabasePageModel {
 				break;
 			
 			case PlatformID.Unix:
-				Process.Start("xdg-open", [ folder ]);
+				Process.Start("xdg-open", [folder]);
 				break;
 			
 			case PlatformID.MacOSX:
-				Process.Start("open", [ folder ]);
+				Process.Start("open", [folder]);
 				break;
 			
 			default:
@@ -80,22 +80,25 @@ sealed class DatabasePageModel {
 		
 		const string Title = "Database Merge";
 		
-		ImportResult? result;
+		var result = new TaskCompletionSource<ImportResult?>();
 		try {
-			result = await ProgressDialog.Show(window, Title, async (dialog, callback) => await MergeWithDatabaseFromPaths(Db, paths, dialog, callback));
+			var dialog = new ProgressDialog();
+			dialog.DataContext = new ProgressDialogModel(Title, async callbacks => result.SetResult(await MergeWithDatabaseFromPaths(Db, paths, dialog, callbacks)), progressItems: 2);
+			await dialog.ShowProgressDialog(window);
 		} catch (Exception e) {
 			Log.Error("Could not merge databases.", e);
 			await Dialog.ShowOk(window, Title, "Could not merge databases: " + e.Message);
 			return;
 		}
 		
-		await Dialog.ShowOk(window, Title, GetImportDialogMessage(result, "database file"));
+		await Dialog.ShowOk(window, Title, GetImportDialogMessage(result.Task.Result, "database file"));
 	}
 	
-	private static async Task<ImportResult?> MergeWithDatabaseFromPaths(IDatabaseFile target, string[] paths, ProgressDialog dialog, IProgressCallback callback) {
+	private static async Task<ImportResult?> MergeWithDatabaseFromPaths(IDatabaseFile target, string[] paths, ProgressDialog dialog, IReadOnlyList<IProgressCallback> callbacks) {
 		var schemaUpgradeCallbacks = new SchemaUpgradeCallbacks(dialog, paths.Length);
+		var databaseMergeProgressCallback = new DatabaseMergeProgressCallback(callbacks[1]);
 		
-		return await PerformImport(target, paths, dialog, callback, "Database Merge", async path => {
+		return await PerformImport(target, paths, dialog, callbacks[0], "Database Merge", async path => {
 			IDatabaseFile? db = await DatabaseGui.TryOpenOrCreateDatabaseFromPath(path, dialog, schemaUpgradeCallbacks);
 			
 			if (db == null) {
@@ -103,7 +106,7 @@ sealed class DatabasePageModel {
 			}
 			
 			try {
-				await target.AddFrom(db);
+				await target.Merge(db, databaseMergeProgressCallback);
 				return true;
 			} finally {
 				await db.DisposeAsync();
@@ -143,6 +146,20 @@ sealed class DatabasePageModel {
 		}
 	}
 	
+	private sealed class DatabaseMergeProgressCallback(IProgressCallback callback) : DatabaseMerging.IProgressCallback {
+		public void OnImportingMetadata() {
+			callback.UpdateIndeterminate("Importing metadata...");
+		}
+		
+		public void OnMessagesImported(long finished, long total) {
+			callback.Update("Importing messages...", finished, total);
+		}
+		
+		public void OnDownloadsImported(long finished, long total) {
+			callback.Update("Importing downloaded files...", finished, total);
+		}
+	}
+	
 	public async Task ImportLegacyArchive() {
 		string[] paths = await window.StorageProvider.OpenFiles(new FilePickerOpenOptions {
 			Title = "Open Legacy DHT Archive",
@@ -223,7 +240,7 @@ sealed class DatabasePageModel {
 		int finished = 0;
 		
 		foreach (string path in paths) {
-			await callback.Update(Path.GetFileName(path), finished, total);
+			await callback.Update("File: " + Path.GetFileName(path), finished, total);
 			++finished;
 			
 			if (!File.Exists(path)) {

app/Server/Database/DatabaseExtensions.cs

@@ -1,34 +0,0 @@
-using System.Collections.Generic;
-using System.Linq;
-using System.Threading.Tasks;
-using DHT.Server.Data;
-
-namespace DHT.Server.Database;
-
-public static class DatabaseExtensions {
-	public static async Task AddFrom(this IDatabaseFile target, IDatabaseFile source) {
-		await target.Users.Add(await source.Users.Get().ToListAsync());
-		await target.Servers.Add(await source.Servers.Get().ToListAsync());
-		await target.Channels.Add(await source.Channels.Get().ToListAsync());
-		
-		const int MessageBatchSize = 100;
-		List<Message> batchedMessages = new (MessageBatchSize);
-		
-		await foreach (Message message in source.Messages.Get()) {
-			batchedMessages.Add(message);
-			
-			if (batchedMessages.Count >= MessageBatchSize) {
-				await target.Messages.Add(batchedMessages);
-				batchedMessages.Clear();
-			}
-		}
-		
-		await target.Messages.Add(batchedMessages);
-		
-		await foreach (Data.Download download in source.Downloads.Get()) {
-			if (download.Status != DownloadStatus.Success || !await source.Downloads.GetDownloadData(download.NormalizedUrl, stream => target.Downloads.AddDownload(download, stream))) {
-				await target.Downloads.AddDownload(download, stream: null);
-			}
-		}
-	}
-}

app/Server/Database/Import/DatabaseMerging.cs

@@ -0,0 +1,77 @@
+using System.Collections.Generic;
+using System.Linq;
+using System.Threading.Tasks;
+using DHT.Server.Data;
+
+namespace DHT.Server.Database.Import;
+
+public static class DatabaseMerging {
+	public static async Task Merge(this IDatabaseFile target, IDatabaseFile source, IProgressCallback callback) {
+		// Import downloads first, otherwise automatic downloads would try to re-download files from other imported data.
+		await MergeDownloads(target, source, callback);
+		
+		callback.OnImportingMetadata();
+		await target.Users.Add(await source.Users.Get().ToListAsync());
+		await target.Servers.Add(await source.Servers.Get().ToListAsync());
+		await target.Channels.Add(await source.Channels.Get().ToListAsync());
+		
+		await MergeMessages(target, source, callback);
+	}
+	
+	private static async Task MergeDownloads(IDatabaseFile target, IDatabaseFile source, IProgressCallback callback) {
+		const int ReportBatchSize = 100;
+		
+		long totalDownloads = await source.Downloads.Count();
+		long importedDownloads = 0;
+		
+		callback.OnDownloadsImported(importedDownloads, totalDownloads);
+		
+		await foreach (Data.Download download in source.Downloads.Get()) {
+			if (download.Status != DownloadStatus.Success || !await source.Downloads.GetDownloadData(download.NormalizedUrl, stream => target.Downloads.AddDownload(download, stream))) {
+				await target.Downloads.AddDownload(download, stream: null);
+			}
+			
+			if (++importedDownloads % ReportBatchSize == 0) {
+				callback.OnDownloadsImported(importedDownloads, totalDownloads);
+			}
+		}
+		
+		callback.OnDownloadsImported(totalDownloads, totalDownloads);
+	}
+	
+	private static async Task MergeMessages(IDatabaseFile target, IDatabaseFile source, IProgressCallback callback) {
+		const int MessageBatchSize = 100;
+		const int ReportEveryBatches = 10;
+		List<Message> batchedMessages = new (MessageBatchSize);
+		
+		long totalMessages = await source.Messages.Count();
+		long importedMessages = 0;
+		
+		callback.OnMessagesImported(importedMessages, totalMessages);
+		
+		await foreach (Message message in source.Messages.Get()) {
+			batchedMessages.Add(message);
+			
+			if (batchedMessages.Count >= MessageBatchSize) {
+				await target.Messages.Add(batchedMessages);
+				
+				importedMessages += batchedMessages.Count;
+				
+				if (importedMessages % (MessageBatchSize * ReportEveryBatches) == 0) {
+					callback.OnMessagesImported(importedMessages, totalMessages);
+				}
+				
+				batchedMessages.Clear();
+			}
+		}
+		
+		await target.Messages.Add(batchedMessages);
+		callback.OnMessagesImported(totalMessages, totalMessages);
+	}
+	
+	public interface IProgressCallback {
+		void OnImportingMetadata();
+		void OnMessagesImported(long finished, long total);
+		void OnDownloadsImported(long finished, long total);
+	}
+}

app/Server/Database/Repositories/IDownloadRepository.cs

@@ -17,7 +17,7 @@ public interface IDownloadRepository {
 	
 	Task AddDownload(Data.Download item, Stream? stream);
 	
-	Task<long> Count(DownloadItemFilter filter, CancellationToken cancellationToken = default);
+	Task<long> Count(DownloadItemFilter? filter = null, CancellationToken cancellationToken = default);
 	
 	Task<DownloadStatusStatistics> GetStatistics(DownloadItemFilter nonSkippedFilter, CancellationToken cancellationToken = default);
 	
@@ -44,7 +44,7 @@ public interface IDownloadRepository {
 			return Task.CompletedTask;
 		}
 		
-		public Task<long> Count(DownloadItemFilter filter, CancellationToken cancellationToken) {
+		public Task<long> Count(DownloadItemFilter? filter, CancellationToken cancellationToken) {
 			return Task.FromResult(0L);
 		}