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

Compare commits

..

No commits in common. "4929a19397e8893b49ee2024a481c5fa2f973546" and "ef3e34066a3bf7b5bfdc36a921e5be8410be197f" have entirely different histories.

21 changed files with 381 additions and 445 deletions

View File

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

View File

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

View File

@ -18,7 +18,7 @@ using DHT.Server;
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.Schema; using DHT.Server.Database.Sqlite.Utils;
using DHT.Utils.Logging; using DHT.Utils.Logging;
namespace DHT.Desktop.Main.Pages; namespace DHT.Desktop.Main.Pages;

View File

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

View File

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

View File

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

View File

@ -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 cmd = conn.Command("SELECT id, server, name, parent_id, position, topic, nsfw FROM channels");
await using var reader = await cmd.ExecuteReaderAsync(); await using var reader = await cmd.ExecuteReaderAsync();
while (await reader.ReadAsync()) { while (reader.Read()) {
yield return new Channel { yield return new Channel {
Id = reader.GetUint64(0), Id = reader.GetUint64(0),
Server = reader.GetUint64(1), Server = reader.GetUint64(1),

View File

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

View File

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

View File

@ -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 cmd = conn.Command("SELECT id, name, avatar_url, discriminator FROM users");
await using var reader = await cmd.ExecuteReaderAsync(); await using var reader = await cmd.ExecuteReaderAsync();
while (await reader.ReadAsync()) { while (reader.Read()) {
yield return new User { yield return new User {
Id = reader.GetUint64(0), Id = reader.GetUint64(0),
Name = reader.GetString(1), Name = reader.GetString(1),

View File

@ -0,0 +1,358 @@
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();
}
}

View File

@ -1,8 +0,0 @@
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);
}

View File

@ -1,11 +0,0 @@
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");
}
}

View File

@ -1,33 +0,0 @@
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");
}
}

View File

@ -1,19 +0,0 @@
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
)
""");
}
}

View File

@ -1,12 +0,0 @@
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");
}
}

View File

@ -1,138 +0,0 @@
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");
}
}

View File

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

View File

@ -1,191 +0,0 @@
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();
}
}

View File

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

View File

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