1
0
mirror of https://github.com/chylex/Discord-History-Tracker.git synced 2024-11-25 14:42:44 +01:00

Compare commits

...

3 Commits

16 changed files with 440 additions and 230 deletions

View File

@ -1,9 +1,11 @@
using System; using System;
using System.Collections.Generic; using System.Collections.Generic;
using System.IO; using System.IO;
using System.Threading;
using System.Threading.Tasks; using System.Threading.Tasks;
using Avalonia.Controls; using Avalonia.Controls;
using Avalonia.Platform.Storage; using Avalonia.Platform.Storage;
using Avalonia.Threading;
using DHT.Desktop.Dialogs.File; using DHT.Desktop.Dialogs.File;
using DHT.Desktop.Dialogs.Message; using DHT.Desktop.Dialogs.Message;
using DHT.Server.Database; using DHT.Server.Database;
@ -41,11 +43,16 @@ static class DatabaseGui {
}); });
} }
public static async Task<IDatabaseFile?> TryOpenOrCreateDatabaseFromPath(string path, Window window, Func<Task<bool>> checkCanUpgradeDatabase) { public static async Task<IDatabaseFile?> TryOpenOrCreateDatabaseFromPath(string path, Window window, ISchemaUpgradeCallbacks schemaUpgradeCallbacks) {
var prevSynchronizationContext = SynchronizationContext.Current;
SynchronizationContext.SetSynchronizationContext(new AvaloniaSynchronizationContext());
var taskScheduler = TaskScheduler.FromCurrentSynchronizationContext();
SynchronizationContext.SetSynchronizationContext(prevSynchronizationContext);
IDatabaseFile? file = null; IDatabaseFile? file = null;
try { try {
file = await SqliteDatabaseFile.OpenOrCreate(path, checkCanUpgradeDatabase); file = await SqliteDatabaseFile.OpenOrCreate(path, schemaUpgradeCallbacks, taskScheduler);
} catch (InvalidDatabaseVersionException ex) { } catch (InvalidDatabaseVersionException ex) {
await Dialog.ShowOk(window, "Database Error", "Database '" + Path.GetFileName(path) + "' appears to be corrupted (invalid version: " + ex.Version + ")."); await Dialog.ShowOk(window, "Database Error", "Database '" + Path.GetFileName(path) + "' appears to be corrupted (invalid version: " + ex.Version + ").");
} catch (DatabaseTooNewException ex) { } catch (DatabaseTooNewException ex) {

View File

@ -4,4 +4,5 @@ namespace DHT.Desktop.Dialogs.Progress;
interface IProgressCallback { interface IProgressCallback {
Task Update(string message, int finishedItems, int totalItems); Task Update(string message, int finishedItems, int totalItems);
Task Hide();
} }

View File

@ -32,12 +32,18 @@
</Style> </Style>
</Window.Styles> </Window.Styles>
<StackPanel Margin="20"> <ItemsRepeater ItemsSource="{Binding Items}" Margin="0 10">
<ItemsRepeater.ItemTemplate>
<DataTemplate>
<StackPanel Margin="20 10" IsHitTestVisible="{Binding IsVisible}" Opacity="{Binding Opacity}">
<DockPanel> <DockPanel>
<TextBlock DockPanel.Dock="Right" Text="{Binding Items}" Classes="items" /> <TextBlock DockPanel.Dock="Right" Text="{Binding Items}" Classes="items" />
<TextBlock DockPanel.Dock="Left" Text="{Binding Message}" /> <TextBlock DockPanel.Dock="Left" Text="{Binding Message}" />
</DockPanel> </DockPanel>
<ProgressBar Value="{Binding Progress}" /> <ProgressBar Value="{Binding Progress}" />
</StackPanel> </StackPanel>
</DataTemplate>
</ItemsRepeater.ItemTemplate>
</ItemsRepeater>
</Window> </Window>

View File

@ -1,4 +1,6 @@
using System; using System;
using System.Collections.Generic;
using System.Linq;
using System.Threading.Tasks; using System.Threading.Tasks;
using Avalonia.Threading; using Avalonia.Threading;
using DHT.Desktop.Common; using DHT.Desktop.Common;
@ -9,57 +11,43 @@ namespace DHT.Desktop.Dialogs.Progress;
sealed class ProgressDialogModel : BaseModel { sealed class ProgressDialogModel : BaseModel {
public string Title { get; init; } = ""; public string Title { get; init; } = "";
private string message = ""; public IReadOnlyList<ProgressItem> Items { get; } = Array.Empty<ProgressItem>();
public string Message {
get => message;
private set => Change(ref message, value);
}
private string items = "";
public string Items {
get => items;
private set => Change(ref items, value);
}
private int progress = 0;
public int Progress {
get => progress;
private set => Change(ref progress, value);
}
private readonly TaskRunner? task; private readonly TaskRunner? task;
[Obsolete("Designer")] [Obsolete("Designer")]
public ProgressDialogModel() {} public ProgressDialogModel() {}
public ProgressDialogModel(TaskRunner task) { public ProgressDialogModel(TaskRunner task, int progressItems = 1) {
this.Items = Enumerable.Range(0, progressItems).Select(static _ => new ProgressItem()).ToArray();
this.task = task; this.task = task;
} }
internal async Task StartTask() { internal async Task StartTask() {
if (task != null) { if (task != null) {
await task(new Callback(this)); await task(Items.Select(static item => new Callback(item)).ToArray());
} }
} }
public delegate Task TaskRunner(IProgressCallback callback); public delegate Task TaskRunner(IReadOnlyList<IProgressCallback> callbacks);
private sealed class Callback : IProgressCallback { private sealed class Callback : IProgressCallback {
private readonly ProgressDialogModel model; private readonly ProgressItem item;
public Callback(ProgressDialogModel model) { public Callback(ProgressItem item) {
this.model = model; this.item = item;
} }
async Task IProgressCallback.Update(string message, int finishedItems, int totalItems) { public async Task Update(string message, int finishedItems, int totalItems) {
await Dispatcher.UIThread.InvokeAsync(() => { await Dispatcher.UIThread.InvokeAsync(() => {
model.Message = message; item.Message = message;
model.Items = finishedItems.Format() + " / " + totalItems.Format(); item.Items = totalItems == 0 ? string.Empty : finishedItems.Format() + " / " + totalItems.Format();
model.Progress = 100 * finishedItems / totalItems; item.Progress = totalItems == 0 ? 0 : 100 * finishedItems / totalItems;
}); });
} }
public Task Hide() {
return Update(string.Empty, 0, 0);
}
} }
} }

View File

@ -0,0 +1,41 @@
using DHT.Utils.Models;
namespace DHT.Desktop.Dialogs.Progress;
sealed class ProgressItem : BaseModel {
private bool isVisible = false;
public bool IsVisible {
get => isVisible;
private set {
Change(ref isVisible, value);
OnPropertyChanged(nameof(Opacity));
}
}
public double Opacity => IsVisible ? 1.0 : 0.0;
private string message = "";
public string Message {
get => message;
set {
Change(ref message, value);
IsVisible = !string.IsNullOrEmpty(value);
}
}
private string items = "";
public string Items {
get => items;
set => Change(ref items, value);
}
private int progress = 0;
public int Progress {
get => progress;
set => Change(ref progress, value);
}
}

View File

@ -17,6 +17,7 @@ using DHT.Desktop.Dialogs.TextBox;
using DHT.Server.Data; using DHT.Server.Data;
using DHT.Server.Database; using DHT.Server.Database;
using DHT.Server.Database.Import; using DHT.Server.Database.Import;
using DHT.Server.Database.Sqlite;
using DHT.Utils.Logging; using DHT.Utils.Logging;
using DHT.Utils.Models; using DHT.Utils.Models;
@ -77,7 +78,7 @@ sealed class DatabasePageModel : BaseModel {
} }
ProgressDialog progressDialog = new ProgressDialog(); ProgressDialog progressDialog = new ProgressDialog();
progressDialog.DataContext = new ProgressDialogModel(async callback => await MergeWithDatabaseFromPaths(Db, paths, progressDialog, callback)) { progressDialog.DataContext = new ProgressDialogModel(async callbacks => await MergeWithDatabaseFromPaths(Db, paths, progressDialog, callbacks[0])) {
Title = "Database Merge" Title = "Database Merge"
}; };
@ -85,23 +86,10 @@ sealed class DatabasePageModel : BaseModel {
} }
private static async Task MergeWithDatabaseFromPaths(IDatabaseFile target, string[] paths, ProgressDialog dialog, IProgressCallback callback) { private static async Task MergeWithDatabaseFromPaths(IDatabaseFile target, string[] paths, ProgressDialog dialog, IProgressCallback callback) {
int total = paths.Length; var schemaUpgradeCallbacks = new SchemaUpgradeCallbacks(dialog, paths.Length);
DialogResult.YesNo? upgradeResult = null;
async Task<bool> CheckCanUpgradeDatabase() {
upgradeResult ??= total > 1
? await DatabaseGui.ShowCanUpgradeMultipleDatabaseDialog(dialog)
: await DatabaseGui.ShowCanUpgradeDatabaseDialog(dialog);
return DialogResult.YesNo.Yes == upgradeResult;
}
await PerformImport(target, paths, dialog, callback, "Database Merge", "Database Error", "database file", async path => { await PerformImport(target, paths, dialog, callback, "Database Merge", "Database Error", "database file", async path => {
SynchronizationContext? prevSyncContext = SynchronizationContext.Current; IDatabaseFile? db = await DatabaseGui.TryOpenOrCreateDatabaseFromPath(path, dialog, schemaUpgradeCallbacks);
SynchronizationContext.SetSynchronizationContext(new AvaloniaSynchronizationContext());
IDatabaseFile? db = await DatabaseGui.TryOpenOrCreateDatabaseFromPath(path, dialog, CheckCanUpgradeDatabase);
SynchronizationContext.SetSynchronizationContext(prevSyncContext);
if (db == null) { if (db == null) {
return false; return false;
@ -116,6 +104,41 @@ sealed class DatabasePageModel : BaseModel {
}); });
} }
private sealed class SchemaUpgradeCallbacks : ISchemaUpgradeCallbacks {
private readonly ProgressDialog dialog;
private readonly int total;
private bool? decision;
public SchemaUpgradeCallbacks(ProgressDialog dialog, int total) {
this.total = total;
this.dialog = dialog;
}
public async Task<bool> CanUpgrade() {
return decision ??= (total > 1
? await DatabaseGui.ShowCanUpgradeMultipleDatabaseDialog(dialog)
: await DatabaseGui.ShowCanUpgradeDatabaseDialog(dialog)) == DialogResult.YesNo.Yes;
}
public Task Start(int versionSteps, Func<ISchemaUpgradeCallbacks.IProgressReporter, Task> doUpgrade) {
return doUpgrade(new NullReporter());
}
private sealed class NullReporter : ISchemaUpgradeCallbacks.IProgressReporter {
public Task NextVersion() {
return Task.CompletedTask;
}
public Task MainWork(string message, int finishedItems, int totalItems) {
return Task.CompletedTask;
}
public Task SubWork(string message, int finishedItems, int totalItems) {
return Task.CompletedTask;
}
}
}
public async void ImportLegacyArchive() { public async void ImportLegacyArchive() {
var paths = await window.StorageProvider.OpenFiles(new FilePickerOpenOptions { var paths = await window.StorageProvider.OpenFiles(new FilePickerOpenOptions {
Title = "Open Legacy DHT Archive", Title = "Open Legacy DHT Archive",
@ -128,11 +151,11 @@ sealed class DatabasePageModel : BaseModel {
} }
ProgressDialog progressDialog = new ProgressDialog(); ProgressDialog progressDialog = new ProgressDialog();
progressDialog.DataContext = new ProgressDialogModel(async callback => await ImportLegacyArchiveFromPaths(Db, paths, progressDialog, callback)) { progressDialog.DataContext = new ProgressDialogModel(async callbacks => await ImportLegacyArchiveFromPaths(Db, paths, progressDialog, callbacks[0])) {
Title = "Legacy Archive Import" Title = "Legacy Archive Import"
}; };
await progressDialog.ShowDialog(window); await progressDialog.ShowProgressDialog(window);
} }
private static async Task ImportLegacyArchiveFromPaths(IDatabaseFile target, string[] paths, ProgressDialog dialog, IProgressCallback callback) { private static async Task ImportLegacyArchiveFromPaths(IDatabaseFile target, string[] paths, ProgressDialog dialog, IProgressCallback callback) {

View File

@ -45,7 +45,7 @@ namespace DHT.Desktop.Main.Pages {
} }
ProgressDialog progressDialog = new ProgressDialog { ProgressDialog progressDialog = new ProgressDialog {
DataContext = new ProgressDialogModel(async callback => await GenerateRandomData(channels, users, messages, callback)) { DataContext = new ProgressDialogModel(async callbacks => await GenerateRandomData(channels, users, messages, callbacks[0])) {
Title = "Generating Random Data" Title = "Generating Random Data"
} }
}; };

View File

@ -1,10 +1,13 @@
using System; using System;
using System.Collections.Generic;
using System.IO; using System.IO;
using System.Threading.Tasks; using System.Threading.Tasks;
using Avalonia.Controls; using Avalonia.Controls;
using DHT.Desktop.Common; using DHT.Desktop.Common;
using DHT.Desktop.Dialogs.Message; using DHT.Desktop.Dialogs.Message;
using DHT.Desktop.Dialogs.Progress;
using DHT.Server.Database; using DHT.Server.Database;
using DHT.Server.Database.Sqlite;
using DHT.Utils.Models; using DHT.Utils.Models;
namespace DHT.Desktop.Main.Screens; namespace DHT.Desktop.Main.Screens;
@ -39,16 +42,73 @@ sealed class WelcomeScreenModel : BaseModel, IDisposable {
} }
dbFilePath = path; dbFilePath = path;
Db = await DatabaseGui.TryOpenOrCreateDatabaseFromPath(path, window, CheckCanUpgradeDatabase); Db = await DatabaseGui.TryOpenOrCreateDatabaseFromPath(path, window, new SchemaUpgradeCallbacks(window));
OnPropertyChanged(nameof(Db)); OnPropertyChanged(nameof(Db));
OnPropertyChanged(nameof(HasDatabase)); OnPropertyChanged(nameof(HasDatabase));
} }
private async Task<bool> CheckCanUpgradeDatabase() { private sealed class SchemaUpgradeCallbacks : ISchemaUpgradeCallbacks {
private readonly Window window;
public SchemaUpgradeCallbacks(Window window) {
this.window = window;
}
public async Task<bool> CanUpgrade() {
return DialogResult.YesNo.Yes == await DatabaseGui.ShowCanUpgradeDatabaseDialog(window); return DialogResult.YesNo.Yes == await DatabaseGui.ShowCanUpgradeDatabaseDialog(window);
} }
public async Task Start(int versionSteps, Func<ISchemaUpgradeCallbacks.IProgressReporter, Task> doUpgrade) {
async Task StartUpgrade(IReadOnlyList<IProgressCallback> callbacks) {
var reporter = new ProgressReporter(versionSteps, callbacks);
await reporter.NextVersion();
await Task.Delay(TimeSpan.FromMilliseconds(800));
await doUpgrade(reporter);
await Task.Delay(TimeSpan.FromMilliseconds(600));
}
await new ProgressDialog {
DataContext = new ProgressDialogModel(StartUpgrade, progressItems: 3) {
Title = "Upgrading Database"
}
}.ShowProgressDialog(window);
}
private sealed class ProgressReporter : ISchemaUpgradeCallbacks.IProgressReporter {
private readonly IReadOnlyList<IProgressCallback> callbacks;
private readonly int versionSteps;
private int versionProgress = 0;
public ProgressReporter(int versionSteps, IReadOnlyList<IProgressCallback> callbacks) {
this.callbacks = callbacks;
this.versionSteps = versionSteps;
}
public async Task NextVersion() {
await callbacks[0].Update("Upgrading schema version...", versionProgress++, versionSteps);
await HideChildren(0);
}
public async Task MainWork(string message, int finishedItems, int totalItems) {
await callbacks[1].Update(message, finishedItems, totalItems);
await HideChildren(1);
}
public async Task SubWork(string message, int finishedItems, int totalItems) {
await callbacks[2].Update(message, finishedItems, totalItems);
await HideChildren(2);
}
private async Task HideChildren(int parentIndex) {
for (int i = parentIndex + 1; i < callbacks.Count; i++) {
await callbacks[i].Hide();
}
}
}
}
public void CloseDatabase() { public void CloseDatabase() {
Dispose(); Dispose();
OnPropertyChanged(nameof(Db)); OnPropertyChanged(nameof(Db));

View File

@ -21,7 +21,7 @@ public static class LegacyArchiveImport {
public static async Task<bool> Read(Stream stream, IDatabaseFile db, FakeSnowflake fakeSnowflake, Func<Data.Server[], Task<Dictionary<Data.Server, ulong>?>> askForServerIds) { public static async Task<bool> Read(Stream stream, IDatabaseFile db, FakeSnowflake fakeSnowflake, Func<Data.Server[], Task<Dictionary<Data.Server, ulong>?>> askForServerIds) {
var perf = Log.Start(); var perf = Log.Start();
var root = await JsonSerializer.DeserializeAsync(stream, LegacyArchiveJsonContext.Default.JsonElement); var root = await JsonSerializer.DeserializeAsync(stream, JsonElementContext.Default.JsonElement);
try { try {
var meta = root.RequireObject("meta"); var meta = root.RequireObject("meta");

View File

@ -0,0 +1,15 @@
using System;
using System.Threading.Tasks;
namespace DHT.Server.Database.Sqlite;
public interface ISchemaUpgradeCallbacks {
Task<bool> CanUpgrade();
Task Start(int versionSteps, Func<IProgressReporter, Task> doUpgrade);
public interface IProgressReporter {
Task NextVersion();
Task MainWork(string message, int finishedItems, int totalItems);
Task SubWork(string message, int finishedItems, int totalItems);
}
}

View File

@ -1,4 +1,3 @@
using System;
using System.Collections.Generic; using System.Collections.Generic;
using System.Threading.Tasks; using System.Threading.Tasks;
using DHT.Server.Database.Exceptions; using DHT.Server.Database.Exceptions;
@ -20,12 +19,8 @@ sealed class Schema {
this.conn = conn; this.conn = conn;
} }
private void Execute(string sql) { public async Task<bool> Setup(ISchemaUpgradeCallbacks callbacks) {
conn.Command(sql).ExecuteNonQuery(); conn.Execute(@"CREATE TABLE IF NOT EXISTS metadata (key TEXT PRIMARY KEY, value TEXT)");
}
public async Task<bool> Setup(Func<Task<bool>> checkCanUpgradeSchemas) {
Execute(@"CREATE TABLE IF NOT EXISTS metadata (key TEXT PRIMARY KEY, value TEXT)");
var dbVersionStr = conn.SelectScalar("SELECT value FROM metadata WHERE key = 'version'"); var dbVersionStr = conn.SelectScalar("SELECT value FROM metadata WHERE key = 'version'");
if (dbVersionStr == null) { if (dbVersionStr == null) {
@ -38,19 +33,19 @@ sealed class Schema {
throw new DatabaseTooNewException(dbVersion); throw new DatabaseTooNewException(dbVersion);
} }
else if (dbVersion < Version) { else if (dbVersion < Version) {
var proceed = await checkCanUpgradeSchemas(); var proceed = await callbacks.CanUpgrade();
if (!proceed) { if (!proceed) {
return false; return false;
} }
UpgradeSchemas(dbVersion); await callbacks.Start(Version - dbVersion, async reporter => await UpgradeSchemas(dbVersion, reporter));
} }
return true; return true;
} }
private void InitializeSchemas() { private void InitializeSchemas() {
Execute(""" conn.Execute("""
CREATE TABLE users ( CREATE TABLE users (
id INTEGER PRIMARY KEY NOT NULL, id INTEGER PRIMARY KEY NOT NULL,
name TEXT NOT NULL, name TEXT NOT NULL,
@ -59,7 +54,7 @@ sealed class Schema {
) )
"""); """);
Execute(""" conn.Execute("""
CREATE TABLE servers ( CREATE TABLE servers (
id INTEGER PRIMARY KEY NOT NULL, id INTEGER PRIMARY KEY NOT NULL,
name TEXT NOT NULL, name TEXT NOT NULL,
@ -67,7 +62,7 @@ sealed class Schema {
) )
"""); """);
Execute(""" conn.Execute("""
CREATE TABLE channels ( CREATE TABLE channels (
id INTEGER PRIMARY KEY NOT NULL, id INTEGER PRIMARY KEY NOT NULL,
server INTEGER NOT NULL, server INTEGER NOT NULL,
@ -79,7 +74,7 @@ sealed class Schema {
) )
"""); """);
Execute(""" conn.Execute("""
CREATE TABLE messages ( CREATE TABLE messages (
message_id INTEGER PRIMARY KEY NOT NULL, message_id INTEGER PRIMARY KEY NOT NULL,
sender_id INTEGER NOT NULL, sender_id INTEGER NOT NULL,
@ -89,7 +84,7 @@ sealed class Schema {
) )
"""); """);
Execute(""" conn.Execute("""
CREATE TABLE attachments ( CREATE TABLE attachments (
message_id INTEGER NOT NULL, message_id INTEGER NOT NULL,
attachment_id INTEGER NOT NULL PRIMARY KEY NOT NULL, attachment_id INTEGER NOT NULL PRIMARY KEY NOT NULL,
@ -103,14 +98,14 @@ sealed class Schema {
) )
"""); """);
Execute(""" conn.Execute("""
CREATE TABLE embeds ( CREATE TABLE embeds (
message_id INTEGER NOT NULL, message_id INTEGER NOT NULL,
json TEXT NOT NULL json TEXT NOT NULL
) )
"""); """);
Execute(""" conn.Execute("""
CREATE TABLE downloads ( CREATE TABLE downloads (
normalized_url TEXT NOT NULL PRIMARY KEY, normalized_url TEXT NOT NULL PRIMARY KEY,
download_url TEXT, download_url TEXT,
@ -120,7 +115,7 @@ sealed class Schema {
) )
"""); """);
Execute(""" conn.Execute("""
CREATE TABLE reactions ( CREATE TABLE reactions (
message_id INTEGER NOT NULL, message_id INTEGER NOT NULL,
emoji_id INTEGER, emoji_id INTEGER,
@ -133,15 +128,15 @@ sealed class Schema {
CreateMessageEditTimestampTable(); CreateMessageEditTimestampTable();
CreateMessageRepliedToTable(); CreateMessageRepliedToTable();
Execute("CREATE INDEX attachments_message_ix ON attachments(message_id)"); conn.Execute("CREATE INDEX attachments_message_ix ON attachments(message_id)");
Execute("CREATE INDEX embeds_message_ix ON embeds(message_id)"); conn.Execute("CREATE INDEX embeds_message_ix ON embeds(message_id)");
Execute("CREATE INDEX reactions_message_ix ON reactions(message_id)"); conn.Execute("CREATE INDEX reactions_message_ix ON reactions(message_id)");
Execute("INSERT INTO metadata (key, value) VALUES ('version', " + Version + ")"); conn.Execute("INSERT INTO metadata (key, value) VALUES ('version', " + Version + ")");
} }
private void CreateMessageEditTimestampTable() { private void CreateMessageEditTimestampTable() {
Execute(""" conn.Execute("""
CREATE TABLE edit_timestamps ( CREATE TABLE edit_timestamps (
message_id INTEGER PRIMARY KEY NOT NULL, message_id INTEGER PRIMARY KEY NOT NULL,
edit_timestamp INTEGER NOT NULL edit_timestamp INTEGER NOT NULL
@ -150,7 +145,7 @@ sealed class Schema {
} }
private void CreateMessageRepliedToTable() { private void CreateMessageRepliedToTable() {
Execute(""" conn.Execute("""
CREATE TABLE replied_to ( CREATE TABLE replied_to (
message_id INTEGER PRIMARY KEY NOT NULL, message_id INTEGER PRIMARY KEY NOT NULL,
replied_to_id INTEGER NOT NULL replied_to_id INTEGER NOT NULL
@ -158,11 +153,13 @@ sealed class Schema {
"""); """);
} }
private void NormalizeAttachmentUrls() { private async Task NormalizeAttachmentUrls(ISchemaUpgradeCallbacks.IProgressReporter reporter) {
await reporter.SubWork("Preparing attachments...", 0, 0);
var normalizedUrls = new Dictionary<long, string>(); var normalizedUrls = new Dictionary<long, string>();
using (var selectCmd = conn.Command("SELECT attachment_id, url FROM attachments")) { await using (var selectCmd = conn.Command("SELECT attachment_id, url FROM attachments")) {
using var reader = selectCmd.ExecuteReader(); await using var reader = await selectCmd.ExecuteReaderAsync();
while (reader.Read()) { while (reader.Read()) {
var attachmentId = reader.GetInt64(0); var attachmentId = reader.GetInt64(0);
@ -171,28 +168,39 @@ sealed class Schema {
} }
} }
using var tx = conn.BeginTransaction(); await using var tx = conn.BeginTransaction();
using (var updateCmd = conn.Command("UPDATE attachments SET download_url = url, url = :normalized_url WHERE attachment_id = :attachment_id")) { 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.Parameters.Add(":attachment_id", SqliteType.Integer); updateCmd.Parameters.Add(":attachment_id", SqliteType.Integer);
updateCmd.Parameters.Add(":normalized_url", SqliteType.Text); updateCmd.Parameters.Add(":normalized_url", SqliteType.Text);
foreach (var (attachmentId, normalizedUrl) in normalizedUrls) { foreach (var (attachmentId, normalizedUrl) in normalizedUrls) {
if (++processedUrls % 1000 == 0) {
await reporter.SubWork("Updating URLs...", processedUrls, totalUrls);
}
updateCmd.Set(":attachment_id", attachmentId); updateCmd.Set(":attachment_id", attachmentId);
updateCmd.Set(":normalized_url", normalizedUrl); updateCmd.Set(":normalized_url", normalizedUrl);
updateCmd.ExecuteNonQuery(); updateCmd.ExecuteNonQuery();
} }
} }
tx.Commit(); await reporter.SubWork("Updating URLs...", totalUrls, totalUrls);
await tx.CommitAsync();
} }
private void NormalizeDownloadUrls() { private async Task NormalizeDownloadUrls(ISchemaUpgradeCallbacks.IProgressReporter reporter) {
await reporter.SubWork("Preparing downloads...", 0, 0);
var normalizedUrlsToOriginalUrls = new Dictionary<string, string>(); var normalizedUrlsToOriginalUrls = new Dictionary<string, string>();
var duplicateUrlsToDelete = new HashSet<string>(); var duplicateUrlsToDelete = new HashSet<string>();
using (var selectCmd = conn.Command("SELECT url FROM downloads ORDER BY CASE WHEN status = 200 THEN 0 ELSE 1 END")) { await using (var selectCmd = conn.Command("SELECT url FROM downloads ORDER BY CASE WHEN status = 200 THEN 0 ELSE 1 END")) {
using var reader = selectCmd.ExecuteReader(); await using var reader = await selectCmd.ExecuteReaderAsync();
while (reader.Read()) { while (reader.Read()) {
var originalUrl = reader.GetString(0); var originalUrl = reader.GetString(0);
@ -204,68 +212,106 @@ sealed class Schema {
} }
} }
using var tx = conn.BeginTransaction(); conn.Execute("PRAGMA cache_size = -20000");
using (var deleteCmd = conn.Delete("downloads", ("url", SqliteType.Text))) { SqliteTransaction tx;
await using (tx = conn.BeginTransaction()) {
await reporter.SubWork("Deleting duplicates...", 0, 0);
await using (var deleteCmd = conn.Delete("downloads", ("url", SqliteType.Text))) {
foreach (var duplicateUrl in duplicateUrlsToDelete) { foreach (var duplicateUrl in duplicateUrlsToDelete) {
deleteCmd.Set(":url", duplicateUrl); deleteCmd.Set(":url", duplicateUrl);
deleteCmd.ExecuteNonQuery(); deleteCmd.ExecuteNonQuery();
} }
} }
using (var updateCmd = conn.Command("UPDATE downloads SET download_url = :download_url, url = :normalized_url WHERE url = :download_url")) { await tx.CommitAsync();
}
int totalUrls = normalizedUrlsToOriginalUrls.Count;
int processedUrls = -1;
tx = conn.BeginTransaction();
await using (var updateCmd = conn.Command("UPDATE downloads SET download_url = :download_url, url = :normalized_url WHERE url = :download_url")) {
updateCmd.Parameters.Add(":normalized_url", SqliteType.Text); updateCmd.Parameters.Add(":normalized_url", SqliteType.Text);
updateCmd.Parameters.Add(":download_url", SqliteType.Text); updateCmd.Parameters.Add(":download_url", SqliteType.Text);
foreach (var (normalizedUrl, downloadUrl) in normalizedUrlsToOriginalUrls) { 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 = conn.BeginTransaction();
updateCmd.Transaction = tx;
}
updateCmd.Set(":normalized_url", normalizedUrl); updateCmd.Set(":normalized_url", normalizedUrl);
updateCmd.Set(":download_url", downloadUrl); updateCmd.Set(":download_url", downloadUrl);
updateCmd.ExecuteNonQuery(); updateCmd.ExecuteNonQuery();
} }
} }
tx.Commit(); await reporter.SubWork("Updating URLs...", totalUrls, totalUrls);
await tx.CommitAsync();
await tx.DisposeAsync();
conn.Execute("PRAGMA cache_size = -2000");
} }
private void UpgradeSchemas(int dbVersion) { private async Task UpgradeSchemas(int dbVersion, ISchemaUpgradeCallbacks.IProgressReporter reporter) {
var perf = Log.Start("from version " + dbVersion); var perf = Log.Start("from version " + dbVersion);
Execute("UPDATE metadata SET value = " + Version + " WHERE key = 'version'"); conn.Execute("UPDATE metadata SET value = " + Version + " WHERE key = 'version'");
if (dbVersion <= 1) { if (dbVersion <= 1) {
Execute("ALTER TABLE channels ADD parent_id INTEGER"); await reporter.MainWork("Applying schema changes...", 0, 1);
conn.Execute("ALTER TABLE channels ADD parent_id INTEGER");
perf.Step("Upgrade to version 2"); perf.Step("Upgrade to version 2");
await reporter.NextVersion();
} }
if (dbVersion <= 2) { if (dbVersion <= 2) {
await reporter.MainWork("Applying schema changes...", 0, 1);
CreateMessageEditTimestampTable(); CreateMessageEditTimestampTable();
CreateMessageRepliedToTable(); CreateMessageRepliedToTable();
Execute(""" conn.Execute("""
INSERT INTO edit_timestamps (message_id, edit_timestamp) INSERT INTO edit_timestamps (message_id, edit_timestamp)
SELECT message_id, edit_timestamp SELECT message_id, edit_timestamp
FROM messages FROM messages
WHERE edit_timestamp IS NOT NULL WHERE edit_timestamp IS NOT NULL
"""); """);
Execute(""" conn.Execute("""
INSERT INTO replied_to (message_id, replied_to_id) INSERT INTO replied_to (message_id, replied_to_id)
SELECT message_id, replied_to_id SELECT message_id, replied_to_id
FROM messages FROM messages
WHERE replied_to_id IS NOT NULL WHERE replied_to_id IS NOT NULL
"""); """);
Execute("ALTER TABLE messages DROP COLUMN replied_to_id"); conn.Execute("ALTER TABLE messages DROP COLUMN replied_to_id");
Execute("ALTER TABLE messages DROP COLUMN edit_timestamp"); conn.Execute("ALTER TABLE messages DROP COLUMN edit_timestamp");
perf.Step("Upgrade to version 3"); perf.Step("Upgrade to version 3");
Execute("VACUUM"); await reporter.MainWork("Vacuuming the database...", 1, 1);
conn.Execute("VACUUM");
perf.Step("Vacuum"); perf.Step("Vacuum");
await reporter.NextVersion();
} }
if (dbVersion <= 3) { if (dbVersion <= 3) {
Execute(""" conn.Execute("""
CREATE TABLE downloads ( CREATE TABLE downloads (
url TEXT NOT NULL PRIMARY KEY, url TEXT NOT NULL PRIMARY KEY,
status INTEGER NOT NULL, status INTEGER NOT NULL,
@ -275,25 +321,35 @@ sealed class Schema {
"""); """);
perf.Step("Upgrade to version 4"); perf.Step("Upgrade to version 4");
await reporter.NextVersion();
} }
if (dbVersion <= 4) { if (dbVersion <= 4) {
Execute("ALTER TABLE attachments ADD width INTEGER"); await reporter.MainWork("Applying schema changes...", 0, 1);
Execute("ALTER TABLE attachments ADD height INTEGER"); conn.Execute("ALTER TABLE attachments ADD width INTEGER");
conn.Execute("ALTER TABLE attachments ADD height INTEGER");
perf.Step("Upgrade to version 5"); perf.Step("Upgrade to version 5");
await reporter.NextVersion();
} }
if (dbVersion <= 5) { if (dbVersion <= 5) {
Execute("ALTER TABLE attachments ADD download_url TEXT"); await reporter.MainWork("Applying schema changes...", 0, 3);
Execute("ALTER TABLE downloads ADD download_url TEXT"); conn.Execute("ALTER TABLE attachments ADD download_url TEXT");
conn.Execute("ALTER TABLE downloads ADD download_url TEXT");
NormalizeAttachmentUrls(); await reporter.MainWork("Updating attachments...", 1, 3);
NormalizeDownloadUrls(); await NormalizeAttachmentUrls(reporter);
Execute("ALTER TABLE attachments RENAME COLUMN url TO normalized_url"); await reporter.MainWork("Updating downloads...", 2, 3);
Execute("ALTER TABLE downloads RENAME COLUMN url TO normalized_url"); await NormalizeDownloadUrls(reporter);
await reporter.MainWork("Applying schema changes...", 3, 3);
conn.Execute("ALTER TABLE attachments RENAME COLUMN url TO normalized_url");
conn.Execute("ALTER TABLE downloads RENAME COLUMN url TO normalized_url");
perf.Step("Upgrade to version 6"); perf.Step("Upgrade to version 6");
await reporter.NextVersion();
} }
perf.End(); perf.End();

View File

@ -18,7 +18,7 @@ namespace DHT.Server.Database.Sqlite;
public sealed class SqliteDatabaseFile : IDatabaseFile { public sealed class SqliteDatabaseFile : IDatabaseFile {
private const int DefaultPoolSize = 5; private const int DefaultPoolSize = 5;
public static async Task<SqliteDatabaseFile?> OpenOrCreate(string path, Func<Task<bool>> checkCanUpgradeSchemas) { public static async Task<SqliteDatabaseFile?> OpenOrCreate(string path, ISchemaUpgradeCallbacks schemaUpgradeCallbacks, TaskScheduler computeTaskResultScheduler) {
var connectionString = new SqliteConnectionStringBuilder { var connectionString = new SqliteConnectionStringBuilder {
DataSource = path, DataSource = path,
Mode = SqliteOpenMode.ReadWriteCreate, Mode = SqliteOpenMode.ReadWriteCreate,
@ -27,12 +27,16 @@ public sealed class SqliteDatabaseFile : IDatabaseFile {
var pool = new SqliteConnectionPool(connectionString, DefaultPoolSize); var pool = new SqliteConnectionPool(connectionString, DefaultPoolSize);
bool wasOpened; bool wasOpened;
using (var conn = pool.Take()) { try {
wasOpened = await new Schema(conn).Setup(checkCanUpgradeSchemas); using var conn = pool.Take();
wasOpened = await new Schema(conn).Setup(schemaUpgradeCallbacks);
} catch (Exception) {
pool.Dispose();
throw;
} }
if (wasOpened) { if (wasOpened) {
return new SqliteDatabaseFile(path, pool); return new SqliteDatabaseFile(path, pool, computeTaskResultScheduler);
} }
else { else {
pool.Dispose(); pool.Dispose();
@ -49,13 +53,13 @@ public sealed class SqliteDatabaseFile : IDatabaseFile {
private readonly AsyncValueComputer<long>.Single totalAttachmentsComputer; private readonly AsyncValueComputer<long>.Single totalAttachmentsComputer;
private readonly AsyncValueComputer<long>.Single totalDownloadsComputer; private readonly AsyncValueComputer<long>.Single totalDownloadsComputer;
private SqliteDatabaseFile(string path, SqliteConnectionPool pool) { private SqliteDatabaseFile(string path, SqliteConnectionPool pool, TaskScheduler computeTaskResultScheduler) {
this.log = Log.ForType(typeof(SqliteDatabaseFile), System.IO.Path.GetFileName(path)); this.log = Log.ForType(typeof(SqliteDatabaseFile), System.IO.Path.GetFileName(path));
this.pool = pool; this.pool = pool;
this.totalMessagesComputer = AsyncValueComputer<long>.WithResultProcessor(UpdateMessageStatistics).WithOutdatedResults().BuildWithComputer(ComputeMessageStatistics); this.totalMessagesComputer = AsyncValueComputer<long>.WithResultProcessor(UpdateMessageStatistics, computeTaskResultScheduler).WithOutdatedResults().BuildWithComputer(ComputeMessageStatistics);
this.totalAttachmentsComputer = AsyncValueComputer<long>.WithResultProcessor(UpdateAttachmentStatistics).WithOutdatedResults().BuildWithComputer(ComputeAttachmentStatistics); this.totalAttachmentsComputer = AsyncValueComputer<long>.WithResultProcessor(UpdateAttachmentStatistics, computeTaskResultScheduler).WithOutdatedResults().BuildWithComputer(ComputeAttachmentStatistics);
this.totalDownloadsComputer = AsyncValueComputer<long>.WithResultProcessor(UpdateDownloadStatistics).WithOutdatedResults().BuildWithComputer(ComputeDownloadStatistics); this.totalDownloadsComputer = AsyncValueComputer<long>.WithResultProcessor(UpdateDownloadStatistics, computeTaskResultScheduler).WithOutdatedResults().BuildWithComputer(ComputeDownloadStatistics);
this.Path = path; this.Path = path;
this.Statistics = new DatabaseStatistics(); this.Statistics = new DatabaseStatistics();

View File

@ -5,14 +5,19 @@ using Microsoft.Data.Sqlite;
namespace DHT.Server.Database.Sqlite.Utils; namespace DHT.Server.Database.Sqlite.Utils;
static class SqliteExtensions { static class SqliteExtensions {
public static SqliteTransaction BeginTransaction(this ISqliteConnection conn) {
return conn.InnerConnection.BeginTransaction();
}
public static SqliteCommand Command(this ISqliteConnection conn, string sql) { public static SqliteCommand Command(this ISqliteConnection conn, string sql) {
var cmd = conn.InnerConnection.CreateCommand(); var cmd = conn.InnerConnection.CreateCommand();
cmd.CommandText = sql; cmd.CommandText = sql;
return cmd; return cmd;
} }
public static SqliteTransaction BeginTransaction(this ISqliteConnection conn) { public static void Execute(this ISqliteConnection conn, string sql) {
return conn.InnerConnection.BeginTransaction(); using var cmd = conn.Command(sql);
cmd.ExecuteNonQuery();
} }
public static object? SelectScalar(this ISqliteConnection conn, string sql) { public static object? SelectScalar(this ISqliteConnection conn, string sql) {

View File

@ -60,6 +60,10 @@ abstract class BaseEndpoint {
protected abstract Task<IHttpOutput> Respond(HttpContext ctx); protected abstract Task<IHttpOutput> Respond(HttpContext ctx);
protected static async Task<JsonElement> ReadJson(HttpContext ctx) { protected static async Task<JsonElement> ReadJson(HttpContext ctx) {
return await ctx.Request.ReadFromJsonAsync<JsonElement?>() ?? throw new HttpException(HttpStatusCode.UnsupportedMediaType, "This endpoint only accepts JSON."); try {
return await ctx.Request.ReadFromJsonAsync(JsonElementContext.Default.JsonElement);
} catch (JsonException) {
throw new HttpException(HttpStatusCode.UnsupportedMediaType, "This endpoint only accepts JSON.");
}
} }
} }

View File

@ -1,8 +1,8 @@
using System.Text.Json; using System.Text.Json;
using System.Text.Json.Serialization; using System.Text.Json.Serialization;
namespace DHT.Server.Database.Import; namespace DHT.Utils.Http;
[JsonSourceGenerationOptions(PropertyNamingPolicy = JsonKnownNamingPolicy.CamelCase, GenerationMode = JsonSourceGenerationMode.Default)] [JsonSourceGenerationOptions(PropertyNamingPolicy = JsonKnownNamingPolicy.CamelCase, GenerationMode = JsonSourceGenerationMode.Default)]
[JsonSerializable(typeof(JsonElement))] [JsonSerializable(typeof(JsonElement))]
sealed partial class LegacyArchiveJsonContext : JsonSerializerContext {} public sealed partial class JsonElementContext : JsonSerializerContext {}

View File

@ -8,5 +8,5 @@ using DHT.Utils;
namespace DHT.Utils; namespace DHT.Utils;
static class Version { static class Version {
public const string Tag = "39.0.0.0"; public const string Tag = "39.1.0.0";
} }