diff --git a/Parallel.Cli/Commands/CleanCommand.cs b/Parallel.Cli/Commands/CleanCommand.cs index b05a096..57bb19d 100644 --- a/Parallel.Cli/Commands/CleanCommand.cs +++ b/Parallel.Cli/Commands/CleanCommand.cs @@ -1,6 +1,7 @@ // Copyright 2025 Kyle Ebbinga using System.CommandLine; +using System.Runtime.InteropServices; using Parallel.Cli.Utils; using Parallel.Core.IO.Scanning; using Parallel.Core.IO.Syncing; diff --git a/Parallel.Cli/Commands/DiskCommand.cs b/Parallel.Cli/Commands/DiskCommand.cs index ba713cd..0a62b6d 100644 --- a/Parallel.Cli/Commands/DiskCommand.cs +++ b/Parallel.Cli/Commands/DiskCommand.cs @@ -36,8 +36,8 @@ public DiskCommand() : base("disk", "Shows the current disk usage.") private async Task DisplayDiskInformationAsync(LocalVaultConfig vault) { CommandLine.WriteLine($"Retrieving vault information...", ConsoleColor.DarkGray); - ISyncManager syncManager = SyncManager.CreateNew(vault); - if (!await syncManager.ConnectAsync()) + ISyncManager? syncManager = SyncManager.CreateNew(vault); + if (syncManager == null || !await syncManager.ConnectAsync()) { CommandLine.WriteLine(vault, $"Failed to connect to vault '{vault.Name}'!", ConsoleColor.Red); return; @@ -50,17 +50,17 @@ private async Task DisplayDiskInformationAsync(LocalVaultConfig vault) long totalObjects = await db.GetTotalObjectsAsync(); CommandLine.WriteLine($"Using vault '{vault.Name}' ({vault.Id}):"); - CommandLine.WriteLine($"Service Type: {vault.FileSystem.Service}"); - CommandLine.WriteLine($"Root Directory: {vault.FileSystem.RootDirectory}"); + CommandLine.WriteLine($"Service Type: {vault.Credentials.Service}"); + CommandLine.WriteLine($"Root Directory: {vault.Credentials.RootDirectory}"); CommandLine.WriteLine($"Managed Files: {(totalLocalFiles + totalDeletedFiles):N0}"); CommandLine.WriteLine($"Local Files: {totalLocalFiles:N0}"); CommandLine.WriteLine($"Deleted Files: {totalDeletedFiles:N0}"); CommandLine.WriteLine($"Total Objects: {totalObjects:N0}"); CommandLine.WriteLine($"Local Size: {Formatter.FromBytes(localSize)}"); - if (vault.FileSystem.Service.Equals(FileService.Local)) + if (vault.Credentials.Service.Equals(FileService.Local)) { - DriveInfo drive = new(vault.FileSystem.RootDirectory); + DriveInfo drive = new(vault.Credentials.RootDirectory); long diskUsage = drive.TotalSize - drive.TotalFreeSpace; CommandLine.WriteLine($"Total Usage: {Formatter.FromBytes(diskUsage)} ({Math.Round(diskUsage / (double)drive.TotalSize * 100, 1)}%)"); CommandLine.WriteLine($"Disk Free: {Formatter.FromBytes(drive.TotalFreeSpace)} ({Math.Round(drive.TotalFreeSpace / (double)drive.TotalSize * 100, 1)}%)"); diff --git a/Parallel.Cli/Commands/DuplicatesCommand.cs b/Parallel.Cli/Commands/DuplicatesCommand.cs index d5f1393..05136b9 100644 --- a/Parallel.Cli/Commands/DuplicatesCommand.cs +++ b/Parallel.Cli/Commands/DuplicatesCommand.cs @@ -12,7 +12,7 @@ namespace Parallel.Cli.Commands { public class DuplicatesCommand : Command { - private Argument sourceArg = new("path", "The directory to scan."); + private readonly Argument sourceArg = new("path", "The directory to scan."); public DuplicatesCommand() : base("duplicates", "Scans a directory for duplicate files.") { diff --git a/Parallel.Cli/Commands/PullCommand.cs b/Parallel.Cli/Commands/PullCommand.cs index 6f6c1c9..17a9ad2 100644 --- a/Parallel.Cli/Commands/PullCommand.cs +++ b/Parallel.Cli/Commands/PullCommand.cs @@ -40,8 +40,8 @@ public PullCommand() : base("pull", "Pulls changes from a vault.") private async Task PullPathAsync(LocalVaultConfig vault, string path, bool force) { CommandLine.WriteLine($"Retrieving vault information...", ConsoleColor.DarkGray); - ISyncManager syncManager = SyncManager.CreateNew(vault); - if (!await syncManager.ConnectAsync()) + ISyncManager? syncManager = SyncManager.CreateNew(vault); + if (syncManager == null || !await syncManager.ConnectAsync()) { CommandLine.WriteLine(vault, $"Failed to connect to vault '{vault.Name}'!", ConsoleColor.Red); return; @@ -71,7 +71,7 @@ private async Task PullPathAsync(LocalVaultConfig vault, string path, bool force Log.Debug($"Pulling {pullFiles.Count} files..."); await syncManager.PullFilesAsync(pullFiles.ToArray(), new ProgressReport(vault, files.Count())); - CommandLine.WriteLine(vault, $"Successfully pulled {pullFiles.Count:N0} files from '{vault.FileSystem.RootDirectory}'.", ConsoleColor.Green); + CommandLine.WriteLine(vault, $"Successfully pulled {pullFiles.Count:N0} files from '{vault.Credentials.RootDirectory}'.", ConsoleColor.Green); } private async Task PullFileAsync(ISyncManager syncManager, string fullPath, bool force) @@ -95,7 +95,7 @@ private async Task PullFileAsync(ISyncManager syncManager, string fullPath, bool await syncManager.PullFilesAsync([remoteFile], new ProgressLogger()); await syncManager.DisconnectAsync(); - CommandLine.WriteLine(syncManager.RemoteVault, $"Successfully pulled file from '{syncManager.RemoteVault.FileSystem.RootDirectory}'.", ConsoleColor.Green); + CommandLine.WriteLine(syncManager.RemoteVault, $"Successfully pulled file from '{syncManager.RemoteVault.Credentials.RootDirectory}'.", ConsoleColor.Green); } } } \ No newline at end of file diff --git a/Parallel.Cli/Commands/PushCommand.cs b/Parallel.Cli/Commands/PushCommand.cs index 6668fcb..a3a85fd 100644 --- a/Parallel.Cli/Commands/PushCommand.cs +++ b/Parallel.Cli/Commands/PushCommand.cs @@ -1,6 +1,7 @@ // Copyright 2025 Kyle Ebbinga using System.CommandLine; +using System.Diagnostics; using Parallel.Cli.Utils; using Parallel.Core.Diagnostics; using Parallel.Core.IO; @@ -13,31 +14,34 @@ namespace Parallel.Cli.Commands { public class PushCommand : Command { + private Stopwatch _sw = new Stopwatch(); + private Command addCmd = new("add", "Adds a new directory to the sync list."); private Command listCmd = new("list", "Shows all directories in the sync list."); private Command removeCmd = new("remove", "Removes a directory from the sync list."); private readonly Option _sourceArg = new(["--path", "-p"], "The source path to sync."); private readonly Option _configOpt = new(["--config", "-c"], "The vault configuration to use."); - private readonly Option _verboseOpt = new(["--verbose", "-v"], "Shows verbose output."); + private readonly Option _forceOpt = new(["--force", "-f"], "Forces the pull overwriting any files."); public PushCommand() : base("push", "Pushes changed files to vaults.") { this.AddOption(_sourceArg); this.AddOption(_configOpt); - this.AddOption(_verboseOpt); - this.SetHandler(async (path, config, verbose) => + this.AddOption(_forceOpt); + this.SetHandler(async (path, config, force) => { + _sw = Stopwatch.StartNew(); if (string.IsNullOrEmpty(path)) { await SyncSystemAsync(); } else { - await SyncPathAsync(path); + await SyncPathAsync(path, force); } - }, _sourceArg, _configOpt, _verboseOpt); + }, _sourceArg, _configOpt, _forceOpt); } private Task SyncSystemAsync() @@ -45,13 +49,13 @@ private Task SyncSystemAsync() throw new NotImplementedException(); } - private async Task SyncPathAsync(string path) + private async Task SyncPathAsync(string path, bool force) { CommandLine.WriteLine($"Retrieving vault information...", ConsoleColor.DarkGray); await Program.Settings.ForEachVaultAsync(async vault => { - ISyncManager syncManager = SyncManager.CreateNew(vault); - if (!await syncManager.ConnectAsync()) + ISyncManager? syncManager = SyncManager.CreateNew(vault); + if (syncManager == null || !await syncManager.ConnectAsync()) { CommandLine.WriteLine(vault, $"Failed to connect to vault '{vault.Name}'!", ConsoleColor.Red); return; @@ -79,7 +83,7 @@ await Program.Settings.ForEachVaultAsync(async vault => CommandLine.WriteLine(vault, $"Scanning for file changes in {path}...", ConsoleColor.DarkGray); FileScanner scanner = new FileScanner(syncManager); - SystemFile[] files = await scanner.GetFileChangesAsync(path, ignoredFolders); + SystemFile[] files = await scanner.GetFileChangesAsync(path, ignoredFolders, force); int successFiles = files.Length; if (successFiles == 0) { @@ -89,10 +93,11 @@ await Program.Settings.ForEachVaultAsync(async vault => } CommandLine.WriteLine(vault, $"Backing up {files.Length.ToString("N0")} files...", ConsoleColor.DarkGray); - await syncManager.PushFilesAsync(files, new ProgressReport(vault, successFiles)); + await syncManager.PushFilesAsync(files, force, new ProgressReport(vault, successFiles)); + //await syncManager.PushFilesAsync(files, new ProgressBarReporter()); await syncManager.DisconnectAsync(); - CommandLine.WriteLine(vault, $"Successfully pushed {successFiles.ToString("N0")} files to '{vault.FileSystem.RootDirectory}'.", ConsoleColor.Green); + CommandLine.WriteLine(vault, $"Successfully pushed {successFiles.ToString("N0")} files in {_sw.Elapsed}.", ConsoleColor.Green); }); } } diff --git a/Parallel.Cli/Commands/RemapCommand.cs b/Parallel.Cli/Commands/RemapCommand.cs index 8641ae5..8d0c68e 100644 --- a/Parallel.Cli/Commands/RemapCommand.cs +++ b/Parallel.Cli/Commands/RemapCommand.cs @@ -48,8 +48,8 @@ public RemapCommand() : base("remap", "Remaps paths in the vault.") private async Task RemapPathAsync(LocalVaultConfig vault, string source, string target) { - ISyncManager syncManager = SyncManager.CreateNew(vault); - if (!await syncManager.ConnectAsync()) + ISyncManager? syncManager = SyncManager.CreateNew(vault); + if (syncManager == null || !await syncManager.ConnectAsync()) { CommandLine.WriteLine(vault, $"Failed to connect to vault '{vault.Name}'!", ConsoleColor.Red); return; diff --git a/Parallel.Cli/Commands/VaultsCommand.cs b/Parallel.Cli/Commands/VaultsCommand.cs index de0d753..c524f4d 100644 --- a/Parallel.Cli/Commands/VaultsCommand.cs +++ b/Parallel.Cli/Commands/VaultsCommand.cs @@ -16,12 +16,12 @@ public class VaultsCommand : Command private readonly Argument configArg = new("config", "The vault configuration to use."); private readonly Option configOpt = new(["--config", "-c"], "The vault configuration to use."); - private Command addCmd = new("add", "Adds a new vault configuration."); - private Command editCmd = new("edit", "Edits a vault configuration."); + private readonly Command addCmd = new("add", "Adds a new vault configuration."); + private readonly Command editCmd = new("edit", "Edits a vault configuration."); private readonly Command findCmd = new("find", "Finds vault configurations in a location."); - private Command viewCmd = new("view", "Shows the vault configuration."); - private Command setCmd = new("set", "Sets a new vault configuration."); - private Command delCmd = new("delete", "Deletes a vault configuration."); + private readonly Command viewCmd = new("view", "Shows the vault configuration."); + private readonly Command setCmd = new("set", "Sets a new vault configuration."); + private readonly Command delCmd = new("delete", "Deletes a vault configuration."); public VaultsCommand() : base("vaults", "View or edit the vaults.") { @@ -39,31 +39,37 @@ public VaultsCommand() : base("vaults", "View or edit the vaults.") addCmd.SetHandler(() => { CommandLine.WriteLine("Creating new storage vault...", ConsoleColor.DarkGray); - FileSystemCredentials fsc = new FileSystemCredentials(); - fsc.Service = Enum.Parse(CommandLine.ReadString($"Service ({string.Join(", ", Enum.GetNames(typeof(FileService)))})"), true); - if (fsc.Service == FileService.Local) + StorageCredentials spc = new StorageCredentials { - fsc.RootDirectory = CommandLine.ReadString("Root"); + Service = Enum.Parse(CommandLine.ReadString($"Service ({string.Join(", ", Enum.GetNames(typeof(FileService)))})") ?? string.Empty, true) + }; + + if (spc.Service == FileService.Local) + { + CommandLine.WriteLine("It is NOT RECOMMENDED to use a network drive!", ConsoleColor.Yellow); + spc.RootDirectory = CommandLine.ReadString("Root") ?? string.Empty; } - else if (fsc.Service == FileService.Cloud) + else if (spc.Service == FileService.Cloud) { - fsc.Address = CommandLine.ReadString("Bucket Name"); - fsc.Username = CommandLine.ReadString("Access Key"); - fsc.Password = CommandLine.ReadPassword("Secret Key"); + spc.Address = CommandLine.ReadString("Bucket Name"); + spc.Username = CommandLine.ReadString("Access Key"); + spc.Password = CommandLine.ReadPassword("Secret Key"); } else { - fsc.RootDirectory = CommandLine.ReadString("Root"); - fsc.Address = CommandLine.ReadString("Address"); - fsc.Username = CommandLine.ReadString("Username"); - fsc.Password = CommandLine.ReadPassword("Password"); + spc.RootDirectory = CommandLine.ReadString("Root") ?? string.Empty; + spc.Address = CommandLine.ReadString("Address"); + spc.Username = CommandLine.ReadString("Username"); + spc.Password = CommandLine.ReadPassword("Password"); } - //fsc.Encrypt = CommandLine.ReadBool("Encrypt files? (y/n)", false); - //fsc.EncryptionKey = HashGenerator.GenerateHash(32, true); + string profileId = CommandLine.ReadString("Id") ?? HashGenerator.GenerateHash(8, true); + string profileName = CommandLine.ReadString("Name") ?? "Default"; + + spc.Encrypt = CommandLine.ReadBool("Encrypt files? (y/n)", false); + spc.EncryptionKey = spc.Encrypt ? HashGenerator.GenerateHash(32, true) : null; - string? profileName = CommandLine.ReadString("Profile Name"); - LocalVaultConfig localVault = new LocalVaultConfig(profileName, fsc); + LocalVaultConfig localVault = new(profileId, profileName, spc); Program.Settings.Vaults.Add(localVault); Program.Settings.Save(); @@ -78,28 +84,28 @@ public VaultsCommand() : base("vaults", "View or edit the vaults.") this.AddCommand(viewCmd); viewCmd.AddArgument(configArg); - viewCmd.SetHandler(async (vault) => + viewCmd.SetHandler(async (config) => { CommandLine.WriteLine($"Retrieving vault information...", ConsoleColor.DarkGray); - LocalVaultConfig? config = ParallelConfig.GetVault(vault); - if (config == null) + LocalVaultConfig? vault = ParallelConfig.GetVault(config); + if (vault == null) { CommandLine.WriteLine($"Unable to find vault with name: '{vault}'", ConsoleColor.Yellow); return; } - ISyncManager syncManager = SyncManager.CreateNew(config); - if (!await syncManager.ConnectAsync()) + ISyncManager? syncManager = SyncManager.CreateNew(vault); + if (syncManager == null || !await syncManager.ConnectAsync()) { - CommandLine.WriteLine(config, $"Failed to connect to vault '{config.Name}'!", ConsoleColor.Red); + CommandLine.WriteLine(vault, $"Failed to connect to vault '{vault.Name}'!", ConsoleColor.Red); return; } RemoteVaultConfig remoteVault = syncManager.RemoteVault; CommandLine.WriteLine($"'{remoteVault.Name}' ({remoteVault.Id}):"); - CommandLine.WriteArray($"Backup Directories", remoteVault.BackupDirectories); - CommandLine.WriteArray($"Ignore Directories", remoteVault.IgnoreDirectories); - CommandLine.WriteArray($"Prune Directories", remoteVault.PruneDirectories); + CommandLine.WriteArray("Backup Directories", remoteVault.BackupDirectories); + CommandLine.WriteArray("Ignore Directories", remoteVault.IgnoreDirectories); + CommandLine.WriteArray("Prune Directories", remoteVault.PruneDirectories); CommandLine.WriteLine($"Prune Period: {remoteVault.PrunePeriod} days"); await syncManager.DisconnectAsync(); diff --git a/Parallel.Cli/Parallel.Cli.csproj b/Parallel.Cli/Parallel.Cli.csproj index fb20e4e..b655695 100644 --- a/Parallel.Cli/Parallel.Cli.csproj +++ b/Parallel.Cli/Parallel.Cli.csproj @@ -8,7 +8,7 @@ enable enable Parallel - 1.0.0 + 0.0.0.0 Kyle Ebbinga Copyright $(Company). All Rights Reserved. $(AssemblyVersion) @@ -35,23 +35,19 @@ - - - - + + + + - - + + - - - - - + diff --git a/Parallel.Cli/Program.cs b/Parallel.Cli/Program.cs index 6aa1da6..24b7e4a 100644 --- a/Parallel.Cli/Program.cs +++ b/Parallel.Cli/Program.cs @@ -16,11 +16,11 @@ public static async Task Main(string[] args) Settings = ParallelConfig.Load(); string logFile = Path.Combine(PathBuilder.ProgramData, "Logs", "latest.txt"); if (File.Exists(logFile)) File.Delete(logFile); - #if DEBUG +#if DEBUG Log.Logger = new LoggerConfiguration().MinimumLevel.Debug().WriteTo.Console().CreateLogger(); - #else +#else Log.Logger = new LoggerConfiguration().MinimumLevel.Warning().WriteTo.File(logFile).CreateLogger(); - #endif +#endif AssemblyName assembly = Assembly.GetExecutingAssembly().GetName(); Log.Information($"{assembly.Name} [Version {assembly.Version}]"); diff --git a/Parallel.Cli/Utils/CommandLine.cs b/Parallel.Cli/Utils/CommandLine.cs index 0e01b11..def0982 100644 --- a/Parallel.Cli/Utils/CommandLine.cs +++ b/Parallel.Cli/Utils/CommandLine.cs @@ -7,16 +7,19 @@ namespace Parallel.Cli.Utils { - public class CommandLine + /// + /// Represents a series of methods for printing to the console. This class can not be inherited. + /// + public abstract class CommandLine { private static readonly object _consoleLock = new(); - public static string ReadString(object value, ConsoleColor color = ConsoleColor.Gray) + public static string? ReadString(object value, ConsoleColor color = ConsoleColor.Gray) { Console.ForegroundColor = color; Console.Write($"> {value}: "); Console.ResetColor(); - return Console.ReadLine() ?? string.Empty; + return Console.ReadLine(); } public static bool ReadBool(object value, bool defaultValue, ConsoleColor color = ConsoleColor.Gray) @@ -78,7 +81,7 @@ public static void Write(object value, ConsoleColor color = ConsoleColor.Gray) public static void WriteLine(LocalVaultConfig localVault, object value, ConsoleColor color = ConsoleColor.Gray) { string baseLog = $"[{localVault.Id}] {value}"; - switch(color) + switch (color) { default: Log.Information(baseLog); @@ -104,7 +107,7 @@ public static void WriteLine(LocalVaultConfig localVault, object value, ConsoleC public static void WriteLine(object value, ConsoleColor color = ConsoleColor.Gray) { string baseLog = $"{value}"; - switch(color) + /*switch(color) { default: Log.Information(baseLog); @@ -117,7 +120,7 @@ public static void WriteLine(object value, ConsoleColor color = ConsoleColor.Gra case ConsoleColor.Red: Log.Error(baseLog); break; - } + }*/ lock (_consoleLock) { diff --git a/Parallel.Cli/Utils/ProgressBarReporter.cs b/Parallel.Cli/Utils/ProgressBarReporter.cs new file mode 100644 index 0000000..7c15f60 --- /dev/null +++ b/Parallel.Cli/Utils/ProgressBarReporter.cs @@ -0,0 +1,68 @@ +// Copyright 2025 Kyle Ebbinga + +using System.Collections.Concurrent; +using Parallel.Core.Diagnostics; +using Parallel.Core.Models; + +namespace Parallel.Cli.Utils +{ + public class ProgressBarReporter : IProgressReporter + { + private readonly ConcurrentDictionary _lines = new(); + private readonly object _consoleLock = new(); + private int _nextLine = 0; + + // Assign a line for a file + private int GetLine(string fileId) + { + if (!_lines.ContainsKey(fileId)) + { + _lines[fileId] = (_nextLine++).ToString(); + } + + return int.Parse(_lines[fileId]); + } + + public void Report(ProgressOperation operation, SystemFile file) + { + int line = GetLine(file.Id); + string msg = $"{operation,-10} {file.Name,-30} "; + + // Optionally show progress, e.g., percentage + if (file.LocalSize > 0 && file.RemoteSize > 0) + { + double percent = (double)file.RemoteSize / file.LocalSize * 100; + msg += $"{percent:0.0}%"; + } + + lock (_consoleLock) + { + Console.SetCursorPosition(0, line); + Console.Write(msg.PadRight(Console.WindowWidth)); + } + } + + public void Reset() + { + lock (_consoleLock) + { + Console.Clear(); + _lines.Clear(); + _nextLine = 0; + } + } + + + public void Failed(Exception exception, SystemFile file) + { + int line = GetLine(file.Id); + lock (_consoleLock) + { + Console.SetCursorPosition(0, line); + Console.ForegroundColor = ConsoleColor.Red; + Console.Write($"FAILED {file.Name}: {exception.Message}".PadRight(Console.WindowWidth)); + Console.ResetColor(); + } + } + } +} \ No newline at end of file diff --git a/Parallel.Cli/Utils/ProgressReport.cs b/Parallel.Cli/Utils/ProgressReport.cs index 5fe0409..623c28e 100644 --- a/Parallel.Cli/Utils/ProgressReport.cs +++ b/Parallel.Cli/Utils/ProgressReport.cs @@ -10,7 +10,7 @@ public class ProgressReport : IProgressReporter { private readonly LocalVaultConfig _localVault; private int _current; - private int _total; + private readonly int _total; public ProgressReport(LocalVaultConfig localVault, int totalFiles) { diff --git a/Parallel.Core/Database/Contexts/SqliteContext.cs b/Parallel.Core/Database/Contexts/SqliteContext.cs index f3292cc..d6cbb05 100644 --- a/Parallel.Core/Database/Contexts/SqliteContext.cs +++ b/Parallel.Core/Database/Contexts/SqliteContext.cs @@ -6,7 +6,6 @@ using Dapper; using Newtonsoft.Json.Linq; using Parallel.Core.IO; -using Parallel.Core.IO.Blobs; using Parallel.Core.Models; using Parallel.Core.Settings; using Parallel.Core.Utils; @@ -43,7 +42,9 @@ public async Task InitializeAsync() using IDbConnection connection = CreateConnection(); await connection.ExecuteAsync("CREATE TABLE IF NOT EXISTS `objects` (`id` TEXT NOT NULL, `hash` TEXT NOT NULL, orderIndex INTEGER NOT NULL, UNIQUE (id, orderIndex));"); - await connection.ExecuteAsync("CREATE TABLE IF NOT EXISTS `files` (`id` TEXT NOT NULL, `name` TEXT NOT NULL, `localpath` TEXT NOT NULL, `remotepath` TEXT NOT NULL, `lastwrite` LONG INTEGER NOT NULL, `lastupdate` LONG INTEGER NOT NULL, `localsize` LONG INTEGER NOT NULL, `remotesize` LONG INTEGER NOT NULL, `type` TEXT NOT NULL DEFAULT Other CHECK(`type` IN ('Document', 'Photo', 'Music', 'Video', 'Other')), `hidden` INTEGER NOT NULL DEFAULT 0, `readonly` INTEGER NOT NULL DEFAULT 0, `deleted` INTEGER NOT NULL DEFAULT 0, `checksum` BLOB, PRIMARY KEY(`id`));"); + await connection.ExecuteAsync( + "CREATE TABLE IF NOT EXISTS `files` (`id` TEXT NOT NULL, `name` TEXT NOT NULL, `localpath` TEXT NOT NULL, `remotepath` TEXT NOT NULL, `lastwrite` LONG INTEGER NOT NULL, `lastupdate` LONG INTEGER NOT NULL, `localsize` LONG INTEGER NOT NULL, `remotesize` LONG INTEGER NOT NULL, `type` TEXT NOT NULL DEFAULT Other CHECK(`type` IN ('Document', 'Photo', 'Music', 'Video', 'Other')), `hidden` INTEGER NOT NULL DEFAULT 0, `readonly` INTEGER NOT NULL DEFAULT 0, `deleted` INTEGER NOT NULL DEFAULT 0, `checksum` BLOB, PRIMARY KEY(`id`));"); + await connection.ExecuteAsync("CREATE TABLE IF NOT EXISTS `history` (`timestamp` LONG INTEGER NOT NULL, `path` TEXT NOT NULL, `name` TEXT NOT NULL, `type` TEXT NOT NULL, PRIMARY KEY(`timestamp`));"); } @@ -67,6 +68,7 @@ public async Task RemoveFileAsync(SystemFile file) await connection.ExecuteAsync(sql, new { file.Id }); } + /// public async Task GetLocalSizeAsync() { using IDbConnection connection = CreateConnection(); @@ -81,6 +83,7 @@ public async Task GetRemoteSizeAsync() return await connection.QuerySingleOrDefaultAsync(sql); } + /// public async Task GetTotalFilesAsync(bool deleted) { using IDbConnection connection = CreateConnection(); @@ -100,8 +103,8 @@ public async Task> GetFilesAsync(string path) public async Task> GetFilesAsync(string path, bool deleted) { using IDbConnection connection = CreateConnection(); - string sql = $"SELECT * FROM files WHERE deleted = @deleted ORDER BY lastupdate DESC"; - return await connection.QueryAsync(sql,new { deleted }); + string sql = $"SELECT * FROM files WHERE localpath LIKE @Path AND deleted = @deleted ORDER BY lastupdate DESC"; + return await connection.QueryAsync(sql, new { Path = $"%{path}%", deleted }); } /// diff --git a/Parallel.Core/Database/IDatabase.cs b/Parallel.Core/Database/IDatabase.cs index 2ef3b69..061cfb9 100644 --- a/Parallel.Core/Database/IDatabase.cs +++ b/Parallel.Core/Database/IDatabase.cs @@ -3,7 +3,6 @@ using Parallel.Core.IO; using System; using System.Data; -using Parallel.Core.IO.Blobs; using Parallel.Core.IO.FileSystem; using Parallel.Core.Models; diff --git a/Parallel.Core/Diagnostics/IProgressReporter.cs b/Parallel.Core/Diagnostics/IProgressReporter.cs index be7934f..ec41121 100644 --- a/Parallel.Core/Diagnostics/IProgressReporter.cs +++ b/Parallel.Core/Diagnostics/IProgressReporter.cs @@ -6,12 +6,10 @@ namespace Parallel.Core.Diagnostics { public enum ProgressOperation { - Archiving, - Pulling, - Pushing, - Compressing, - Decompressing, - Syncing + Archived, + Pulled, + Pushed, + Synced } /// diff --git a/Parallel.Core/IO/Blobs/BlobStorage.cs b/Parallel.Core/IO/Blobs/BlobStorage.cs deleted file mode 100644 index f34413f..0000000 --- a/Parallel.Core/IO/Blobs/BlobStorage.cs +++ /dev/null @@ -1,71 +0,0 @@ -// Copyright 2025 Kyle Ebbinga - -using Parallel.Core.Diagnostics; -using Parallel.Core.IO.FileSystem; -using Parallel.Core.Security; - -namespace Parallel.Core.IO.Blobs -{ - /// - /// Represents the way to chunk files into blobs for syncing. - /// - public abstract class BlobStorage - { - /// - /// The size, in bytes, to use for chunks of a file. - /// - private static readonly int ChunkSize = 4194304; - - /// - /// Chunks a file into hashes for blob storage. - /// - /// The source path of the file. - /// The temp directory to send chunked objects to. - /// - /// - public static async Task CreateManifestAsync(IStorageProvider fileSystem, string sourcePath, string tempObjDir, IProgressReporter progress) - { - List chunkHashes = new List(); - await using FileStream fs = File.OpenRead(sourcePath); - byte[] buffer = new byte[ChunkSize]; - int bytesRead = 0; - - while ((bytesRead = await fs.ReadAsync(buffer)) > 0) - { - byte[] chunkData = new byte[bytesRead]; - Buffer.BlockCopy(buffer, 0, chunkData, 0, bytesRead); - - string hash = HashGenerator.CreateSHA256(chunkData); - string chunkPath = PathBuilder.GetObjectPath(tempObjDir, hash); - - if(!File.Exists(chunkPath)) await File.WriteAllBytesAsync(chunkPath, chunkData); - chunkHashes.Add(hash); - } - - Log.Debug($"Wrote {chunkHashes.Count} hashes to {tempObjDir}"); - return new FileManifest(sourcePath, chunkHashes, new FileInfo(sourcePath).Length); - } - - /// - /// Assembles a file from the chunked hashes. - /// - /// - /// The path to the chunked objects' folder. - /// - public async Task AssembleFileAsync(IEnumerable chunkHashes, string sourcePath, string createFilePath) - { - Log.Debug($"Assembling '{createFilePath}' from {chunkHashes.Count()} hashes."); - await using FileStream createStream = File.Create(createFilePath); - foreach (string hash in chunkHashes) - { - string chunkPath = PathBuilder.GetObjectPath(sourcePath, hash); - if(!File.Exists(chunkPath)) throw new FileNotFoundException($"Missing chunk for hash: {hash}"); - - await using FileStream chunkStream = File.OpenRead(chunkPath); - await chunkStream.CopyToAsync(createStream); - } - - await createStream.FlushAsync(); - } - } -} \ No newline at end of file diff --git a/Parallel.Core/IO/Blobs/FileManifest.cs b/Parallel.Core/IO/Blobs/FileManifest.cs deleted file mode 100644 index d2afcec..0000000 --- a/Parallel.Core/IO/Blobs/FileManifest.cs +++ /dev/null @@ -1,22 +0,0 @@ -// Copyright 2025 Kyle Ebbinga - -using Parallel.Core.Security; - -namespace Parallel.Core.IO.Blobs -{ - public class FileManifest - { - public string Id { get; } - public string Fullname { get; set; } - public List Hashes { get; } - public long Length { get; set; } - - public FileManifest(string path, IEnumerable hashes, long length) - { - Id = HashGenerator.CreateSHA1(path); - Fullname = path; - Hashes = hashes.ToList(); - Length = length; - } - } -} \ No newline at end of file diff --git a/Parallel.Core/IO/PathBuilder.cs b/Parallel.Core/IO/PathBuilder.cs index fff8c25..4c17568 100644 --- a/Parallel.Core/IO/PathBuilder.cs +++ b/Parallel.Core/IO/PathBuilder.cs @@ -6,6 +6,7 @@ using Parallel.Core.IO.FileSystem; using Parallel.Core.Models; using Parallel.Core.Settings; +using Parallel.Core.Storage; using Parallel.Core.Utils; namespace Parallel.Core.IO @@ -98,7 +99,7 @@ public static string Combine(params string[] paths) /// public static string GetRootDirectory(LocalVaultConfig localVault) { - return Combine(localVault.FileSystem.RootDirectory, "Parallel", localVault.Id); + return Combine(localVault.Credentials.RootDirectory, "Parallel", localVault.Id); } /// @@ -149,8 +150,8 @@ public static string GetDatabaseFile(LocalVaultConfig localVault) /// public static string Remote(string path, RemoteVaultConfig remoteVaultConfig) { - string root = Path.Combine(remoteVaultConfig.FileSystem.RootDirectory, "Parallel", remoteVaultConfig.Id, "Files", path.Replace(":", string.Empty)) + ".gz"; - return remoteVaultConfig.FileSystem.Service switch + string root = Path.Combine(remoteVaultConfig.Credentials.RootDirectory, "Parallel", remoteVaultConfig.Id, "Files", path.Replace(":", string.Empty)) + ".gz"; + return remoteVaultConfig.Credentials.Service switch { FileService.Local => root, FileService.Remote => root.Replace('\\', '/'), @@ -171,7 +172,7 @@ public static bool IsFile(string path) public static string GetObjectPath(string basePath, string hash) { string parentDir = Path.Combine(basePath, hash.Substring(0, 2), hash.Substring(2, 2)); - if (!Directory.Exists(parentDir)) Directory.CreateDirectory(parentDir); + if (!Directory.Exists(parentDir)) Directory.CreateDirectory(parentDir); return Path.Combine(parentDir, hash[4..]); } } diff --git a/Parallel.Core/IO/Scanning/FileScanner.cs b/Parallel.Core/IO/Scanning/FileScanner.cs index cdd0ac2..0bc6445 100644 --- a/Parallel.Core/IO/Scanning/FileScanner.cs +++ b/Parallel.Core/IO/Scanning/FileScanner.cs @@ -52,15 +52,19 @@ public SystemFile[] GetFileChanges() /// /// /// + /// /// A list of files that have changed since the last backup. - public async Task GetFileChangesAsync(string path, string[] ignoreFolders) + public async Task GetFileChangesAsync(string path, string[] ignoreFolders, bool force) { if (!Directory.Exists(path)) return Array.Empty(); ConcurrentBag scannedFiles = new(); ConcurrentBag changedFiles = new(); + Log.Debug("Getting system files..."); HashSet localFiles = FileScanner.GetFiles(path, ignoreFolders, ".").ToHashSet(); + + Log.Debug("Getting database files..."); IEnumerable remoteFiles = await _db.GetFilesAsync(path, false); System.Threading.Tasks.Parallel.ForEach(remoteFiles, ParallelConfig.Options, (remoteFile, ct) => { @@ -69,14 +73,14 @@ public async Task GetFileChangesAsync(string path, string[] ignore SystemFile localFile = new SystemFile(remoteFile.LocalPath); if (IsIgnored(localFile.LocalPath, ignoreFolders)) { - //Log.Debug($"Ignored -> {localFile.LocalPath}"); + Log.Debug($"Ignored -> {localFile.LocalPath}"); localFile.RemotePath = remoteFile.RemotePath; localFile.Deleted = true; changedFiles.Add(localFile); } - else if (HasChanged(localFile, remoteFile)) + else if (HasChanged(localFile, remoteFile) || force) { - //Log.Debug($"Changed -> {localFile.LocalPath}"); + Log.Debug($"Changed -> {localFile.LocalPath}"); localFile.RemotePath = remoteFile.RemotePath; changedFiles.Add(localFile); } @@ -85,7 +89,7 @@ public async Task GetFileChangesAsync(string path, string[] ignore } else { - //Log.Debug($"Deleted -> {remoteFile.LocalPath}"); + Log.Debug($"Deleted -> {remoteFile.LocalPath}"); remoteFile.Deleted = true; changedFiles.Add(remoteFile); } @@ -93,16 +97,24 @@ public async Task GetFileChangesAsync(string path, string[] ignore HashSet remainingFiles = localFiles.Except(new HashSet(scannedFiles)).ToHashSet(); Log.Debug($"{remainingFiles.Count} files are untracked! Adding..."); - System.Threading.Tasks.Parallel.ForEach(remainingFiles, ParallelConfig.Options, (file, ct) => + + remainingFiles.AsParallel().WithDegreeOfParallelism(Environment.ProcessorCount).Where(f => !IsIgnored(f, ignoreFolders)).ForAll(file => { - if (!IsIgnored(file, ignoreFolders)) - { - //Log.Debug($"Created -> {file}"); - changedFiles.Add(new SystemFile(file) { RemotePath = PathBuilder.Remote(file, _config) }); - } + Log.Debug($"Created -> {file}"); + changedFiles.Add(new SystemFile(file)); }); - Log.Debug($"{localFiles.Count} files remaining."); + // System.Threading.Tasks.Parallel.ForEach(remainingFiles, ParallelConfig.Options, (file, ct) => + // { + // if (!IsIgnored(file, ignoreFolders)) + // { + // Log.Debug($"Created -> {file}"); + // changedFiles.Add(new SystemFile(file) { RemotePath = PathBuilder.Remote(file, _config) }); + // } + // + // Log.Debug($"Done"); + // }); + Log.Information($"Found {changedFiles.Count:N0} changes in '{path}'"); return changedFiles.ToArray(); } @@ -322,44 +334,33 @@ public static bool IsIgnored(string path, string[] exempt) { foreach (string entry in exempt) { - if (!_cache.TryGetValue(entry, out Regex? regex)) + if (path.StartsWith(entry)) { - regex = BuildRegexCache(entry); - _cache[entry] = regex; - } - - if (regex.IsMatch(path)) return true; - } - - return false; - } - - private static Regex BuildRegexCache(string entry) - { - string pattern; - if (!entry.Contains('*') && !entry.EndsWith("/") && !entry.EndsWith("\\")) - { - pattern = "^" + Regex.Escape(entry); - return new Regex(pattern, RegexOptions.IgnoreCase | RegexOptions.Compiled); - } + } - if (entry.EndsWith("/") || entry.EndsWith("\\")) - { - string folder = entry.TrimEnd('/', '\\'); - pattern = @"(^|\\)" + Regex.Escape(folder) + @"(\\|$)"; - return new Regex(pattern, RegexOptions.IgnoreCase | RegexOptions.Compiled); - } + if (entry.EndsWith('/')) + { + string[] folders = path.Split('\\'); + foreach (string dir in folders) + { + if (dir.ToLower() == entry.Remove(entry.Length - 1, 1).ToLower()) + { + return true; + } + } + } - if (entry.StartsWith("*")) - { - string ext = Regex.Escape(entry.TrimStart('*')); - pattern = ".*" + ext + "$"; - return new Regex(pattern, RegexOptions.IgnoreCase | RegexOptions.Compiled); + if (entry.StartsWith('*')) + { + if (path.EndsWith(entry.Replace("*", string.Empty))) + { + return true; + } + } } - pattern = "^" + Regex.Escape(entry) + "$"; - return new Regex(pattern, RegexOptions.IgnoreCase | RegexOptions.Compiled); + return false; } } } \ No newline at end of file diff --git a/Parallel.Core/IO/Syncing/BaseSyncManager.cs b/Parallel.Core/IO/Syncing/BaseSyncManager.cs index f368ab2..0bb3e9c 100644 --- a/Parallel.Core/IO/Syncing/BaseSyncManager.cs +++ b/Parallel.Core/IO/Syncing/BaseSyncManager.cs @@ -5,6 +5,7 @@ using Parallel.Core.IO.FileSystem; using Parallel.Core.Models; using Parallel.Core.Settings; +using Parallel.Core.Storage; namespace Parallel.Core.IO.Syncing { @@ -27,7 +28,7 @@ public abstract class BaseSyncManager : ISyncManager public IDatabase Database { get; set; } /// - public IStorageProvider Storage { get; set; } + public IStorageProvider StorageProvider { get; set; } /// /// @@ -35,7 +36,7 @@ public abstract class BaseSyncManager : ISyncManager /// public BaseSyncManager(LocalVaultConfig localVault) { - Storage = StorageProvider.CreateNew(localVault); + StorageProvider = StorageConnection.CreateNew(localVault); LocalVault = localVault; } @@ -43,13 +44,13 @@ public BaseSyncManager(LocalVaultConfig localVault) public async Task ConnectAsync() { string root = PathBuilder.GetRootDirectory(LocalVault); - if (!await Storage.ExistsAsync(root)) + if (!await StorageProvider.ExistsAsync(root)) { - await Storage.CreateDirectoryAsync(root); + await StorageProvider.CreateDirectoryAsync(root); Log.Debug($"Created root directory: {root}"); } - if (!await Storage.ExistsAsync(PathBuilder.GetConfigurationFile(LocalVault))) + if (!await StorageProvider.ExistsAsync(PathBuilder.GetConfigurationFile(LocalVault))) { RemoteVault = new RemoteVaultConfig(LocalVault); RemoteVault.IgnoreDirectories.Add(PathBuilder.GetRootDirectory(LocalVault)); @@ -59,15 +60,15 @@ public async Task ConnectAsync() } else { - await Storage.DownloadFilesAsync([new SystemFile(TempConfigFile, PathBuilder.GetConfigurationFile(LocalVault))], new NullProgressReporter()); + await StorageProvider.DownloadFilesAsync([new SystemFile(TempConfigFile, PathBuilder.GetConfigurationFile(LocalVault))], new NullProgressReporter()); RemoteVaultConfig? config = RemoteVaultConfig.Load(TempConfigFile); - if(config == null) return false; + if (config == null) return false; RemoteVault = config; Log.Debug($"Downloaded config file: {TempConfigFile}"); } - if (!await Storage.ExistsAsync(PathBuilder.GetDatabaseFile(LocalVault))) + if (!await StorageProvider.ExistsAsync(PathBuilder.GetDatabaseFile(LocalVault))) { Database = new SqliteContext(TempDbFile); await Database.InitializeAsync(); @@ -76,7 +77,7 @@ public async Task ConnectAsync() } else { - await Storage.DownloadFilesAsync([new SystemFile(TempDbFile, PathBuilder.GetDatabaseFile(LocalVault))], new NullProgressReporter()); + await StorageProvider.DownloadFilesAsync([new SystemFile(TempDbFile, PathBuilder.GetDatabaseFile(LocalVault))], new NullProgressReporter()); Database = new SqliteContext(TempDbFile); Log.Debug($"Downloaded db file: {TempDbFile}"); @@ -92,12 +93,12 @@ public async Task DisconnectAsync() Log.Debug($"Uploaded db file: {TempDbFile}"); SystemFile[] tempFiles = [new SystemFile(TempConfigFile, PathBuilder.GetConfigurationFile(LocalVault)), new SystemFile(TempDbFile, PathBuilder.GetDatabaseFile(LocalVault))]; - await Storage.UploadFilesAsync(tempFiles, new NullProgressReporter()); - Storage.Dispose(); + await StorageProvider.UploadFilesAsync(tempFiles, new NullProgressReporter()); + StorageProvider.Dispose(); } /// - public abstract Task PushFilesAsync(SystemFile[] files, IProgressReporter progress); + public abstract Task PushFilesAsync(SystemFile[] files, bool force, IProgressReporter progress); /// public abstract Task PullFilesAsync(SystemFile[] files, IProgressReporter progress); diff --git a/Parallel.Core/IO/Syncing/DeltaSyncManager.cs b/Parallel.Core/IO/Syncing/DeltaSyncManager.cs index be0b55c..56551f2 100644 --- a/Parallel.Core/IO/Syncing/DeltaSyncManager.cs +++ b/Parallel.Core/IO/Syncing/DeltaSyncManager.cs @@ -18,7 +18,7 @@ public class DeltaSyncManager : BaseSyncManager public DeltaSyncManager(RemoteVaultConfig remoteVaultConfig) : base(remoteVaultConfig) { } /// - public override Task PushFilesAsync(SystemFile[] files, IProgressReporter progress) + public override Task PushFilesAsync(SystemFile[] files, bool force, IProgressReporter progress) { throw new NotImplementedException(); } diff --git a/Parallel.Core/IO/Syncing/FileSyncManager.cs b/Parallel.Core/IO/Syncing/FileSyncManager.cs index 7cfc03f..cc76941 100644 --- a/Parallel.Core/IO/Syncing/FileSyncManager.cs +++ b/Parallel.Core/IO/Syncing/FileSyncManager.cs @@ -19,32 +19,33 @@ public class FileSyncManager : BaseSyncManager public FileSyncManager(LocalVaultConfig localVault) : base(localVault) { } /// - public override async Task PushFilesAsync(SystemFile[] files, IProgressReporter progress) + public override async Task PushFilesAsync(SystemFile[] files, bool force, IProgressReporter progress) { if (!files.Any()) return; SystemFile[] backupFiles = files.Where(f => !f.Deleted).ToArray(); Log.Information($"Backing up {backupFiles.Length} files..."); - await Storage.UploadFilesAsync(backupFiles, progress); + await StorageProvider.UploadFilesAsync(backupFiles, progress); progress.Reset(); await System.Threading.Tasks.Parallel.ForEachAsync(files, ParallelConfig.Options, async (file, ct) => { if (file.Deleted) { - progress.Report(ProgressOperation.Archiving, file); await Database.AddHistoryAsync(file.LocalPath, HistoryType.Archived); await Database.AddFileAsync(file); + progress.Report(ProgressOperation.Archived, file); } else { - progress.Report(ProgressOperation.Syncing, file); - SystemFile? remote = await Storage.GetFileAsync(file.RemotePath); + SystemFile? remote = await StorageProvider.GetFileAsync(file.RemotePath); if (remote is not null) { file.RemoteSize = remote.RemoteSize; await Database.AddHistoryAsync(file.LocalPath, HistoryType.Pushed); await Database.AddFileAsync(file); } + + progress.Report(ProgressOperation.Synced, file); } }); } @@ -52,7 +53,7 @@ await System.Threading.Tasks.Parallel.ForEachAsync(files, ParallelConfig.Options /// public override async Task PullFilesAsync(SystemFile[] files, IProgressReporter progress) { - await Storage.DownloadFilesAsync(files, progress); + await StorageProvider.DownloadFilesAsync(files, progress); } } } \ No newline at end of file diff --git a/Parallel.Core/IO/Syncing/ISyncManager.cs b/Parallel.Core/IO/Syncing/ISyncManager.cs index 80f93f6..c77d455 100644 --- a/Parallel.Core/IO/Syncing/ISyncManager.cs +++ b/Parallel.Core/IO/Syncing/ISyncManager.cs @@ -5,6 +5,7 @@ using Parallel.Core.IO.FileSystem; using Parallel.Core.Models; using Parallel.Core.Settings; +using Parallel.Core.Storage; namespace Parallel.Core.IO.Syncing { @@ -31,7 +32,7 @@ public interface ISyncManager /// /// The associated file system connection. /// - public IStorageProvider Storage { get; set; } + public IStorageProvider StorageProvider { get; set; } /// /// Establishes a connection to the associated and downloads the needed files. @@ -47,8 +48,9 @@ public interface ISyncManager /// Pushes an array of files to a vault. /// /// + /// /// - Task PushFilesAsync(SystemFile[] files, IProgressReporter progress); + Task PushFilesAsync(SystemFile[] files, bool force, IProgressReporter progress); /// /// Pulls an array of files from a vault. diff --git a/Parallel.Core/IO/Syncing/ObjectSyncManager.cs b/Parallel.Core/IO/Syncing/ObjectSyncManager.cs index 7d464cb..a6b2276 100644 --- a/Parallel.Core/IO/Syncing/ObjectSyncManager.cs +++ b/Parallel.Core/IO/Syncing/ObjectSyncManager.cs @@ -1,14 +1,16 @@ // Copyright 2025 Kyle Ebbinga +using System.Buffers; using System.Collections.Concurrent; -using System.Reflection.Metadata; +using System.Diagnostics; +using System.Threading.Channels; using Newtonsoft.Json.Linq; using Parallel.Core.Database; using Parallel.Core.Diagnostics; -using Parallel.Core.IO.Blobs; using Parallel.Core.Models; using Parallel.Core.Security; using Parallel.Core.Settings; +using Parallel.Core.Workers; namespace Parallel.Core.IO.Syncing { @@ -31,87 +33,176 @@ public class ObjectSyncManager : BaseSyncManager public ObjectSyncManager(LocalVaultConfig localVault) : base(localVault) { } /// - public override async Task PushFilesAsync(SystemFile[] files, IProgressReporter progress) + public override async Task PushFilesAsync(SystemFile[] files, bool force, IProgressReporter progress) { - await System.Threading.Tasks.Parallel.ForEachAsync(files, ParallelConfig.Options, async (file, ct) => + long queued = 0, completed = 0, total = 0; + TimeSpan uploadTimeout = TimeSpan.FromSeconds(30); + TimeSpan writeBlockWarn = TimeSpan.FromSeconds(5); + + Channel channel = Channel.CreateBounded(new BoundedChannelOptions(256) { - try + FullMode = BoundedChannelFullMode.Wait, + SingleReader = false, + SingleWriter = false + }); + + Task[] workerTasks = Enumerable.Range(0, ParallelConfig.MaxStaticTransfers).Select(workerId => Task.Run(async () => + { + await foreach (UploadWorker job in channel.Reader.ReadAllAsync()) { - if (file.Deleted) + Interlocked.Decrement(ref queued); + + try { - progress.Report(ProgressOperation.Archiving, file); - await Database.AddHistoryAsync(file.LocalPath, HistoryType.Archived); - await Database.AddFileAsync(file); + Interlocked.Increment(ref total); + string fullPath = PathBuilder.Combine(job.RemotePath, job.Filename); + + if (await StorageProvider.ExistsAsync(fullPath) && !force) + { + Log.Debug($"[WORKER {workerId}] UPLOAD SKIPPED: {fullPath}"); + Interlocked.Increment(ref completed); + continue; + } + + if (!await StorageProvider.ExistsAsync(job.RemotePath)) await StorageProvider.CreateDirectoryAsync(job.RemotePath); + await using MemoryStream ms = new MemoryStream(job.Data, false); + + using CancellationTokenSource cts = new CancellationTokenSource(uploadTimeout); + Task uploadTask = StorageProvider.UploadStreamAsync(ms, fullPath); + + Task timeoutTask = await Task.WhenAny(uploadTask, Task.Delay(uploadTimeout, cts.Token)); + if (timeoutTask != uploadTask) + { + Log.Error($"[WORKER {workerId}] UPLOAD TIMEOUT: {fullPath}"); + job.OnException?.Invoke(new TimeoutException($"Upload timed out after {uploadTimeout}")); + continue; + } + + await uploadTask; + Interlocked.Increment(ref completed); + Log.Debug($"[WORKER {workerId}] UPLOAD COMPLETE: {fullPath} complete={completed}"); } - else + catch (Exception ex) { - int bytesRead = 0; - byte[] buffer = new byte[ChunkSize]; - await using FileStream fs = File.OpenRead(file.LocalPath); - - int index = 0; - progress.Report(ProgressOperation.Pushing, file); - await Database.AddFileAsync(file); + Log.Error($"[WORKER {workerId}] UPLOAD ERROR: {job.Filename} : {ex}"); + job.OnException?.Invoke(ex); + } + } + })).ToArray(); - while ((bytesRead = await fs.ReadAsync(buffer, ct)) > 0) + Task producer = Task.Run(async () => + { + try + { + await System.Threading.Tasks.Parallel.ForEachAsync(files, ParallelConfig.Options, async (file, ct) => + { + try { - await using MemoryStream ms = new MemoryStream(buffer, 0, bytesRead); - string hash = HashGenerator.CreateSHA256(buffer.AsSpan(0, bytesRead)); - await Database.AddObjectAsync(file.Id, hash, index); - index++; - - string basePath = PathBuilder.Combine(RemoteVault.FileSystem.RootDirectory, "Parallel", RemoteVault.Id, "objects"); - string parentDir = PathBuilder.Combine(basePath, hash.Substring(0, 2), hash.Substring(2, 2)); - string remotePath = PathBuilder.Combine(parentDir, hash[4..]); - if (!await Storage.ExistsAsync(remotePath)) + if (file.Deleted) { - if (!await Storage.ExistsAsync(parentDir)) await Storage.CreateDirectoryAsync(parentDir); - await Storage.UploadStreamAsync(ms, remotePath); + await Database.AddHistoryAsync(file.LocalPath, HistoryType.Archived); + await Database.AddFileAsync(file); + progress.Report(ProgressOperation.Archived, file); + return; } - } - await Database.AddHistoryAsync(file.LocalPath, HistoryType.Pushed); - } + await using FileStream fs = new FileStream(file.LocalPath, FileMode.Open, FileAccess.Read, FileShare.Read, bufferSize: ChunkSize, useAsync: true); + byte[] buffer = ArrayPool.Shared.Rent(ChunkSize); + + try + { + int index = 0, bytesRead; + while ((bytesRead = await fs.ReadAsync(buffer.AsMemory(0, ChunkSize), ct)) > 0) + { + byte[] chunk = new byte[bytesRead]; + Buffer.BlockCopy(buffer, 0, chunk, 0, bytesRead); + + string hash = HashGenerator.CreateSHA256(chunk); + await Database.AddObjectAsync(file.Id, hash, index++); + + string basePath = PathBuilder.Combine(RemoteVault.Credentials.RootDirectory, "Parallel", RemoteVault.Id, "objects"); + string parentDir = PathBuilder.Combine(basePath, hash.Substring(0, 2), hash.Substring(2, 2)); + string remotePath = PathBuilder.Combine(parentDir, hash); + + UploadWorker worker = new(chunk, hash, parentDir, ex => progress.Failed(ex, file)); + Task writeTask = channel.Writer.WriteAsync(worker, ct).AsTask(); + Task timeoutTask = await Task.WhenAny(writeTask, Task.Delay(writeBlockWarn, ct)); + if (timeoutTask != writeTask) + { + Log.Warning($"[PRODUCER] channel.Writer.WriteAsync is blocked > {writeBlockWarn} (remote={remotePath})"); + await writeTask; + } + + Interlocked.Increment(ref queued); + } + } + finally + { + ArrayPool.Shared.Return(buffer); + } + + progress.Report(ProgressOperation.Pushed, file); + await Database.AddHistoryAsync(file.LocalPath, HistoryType.Pushed); + await Database.AddFileAsync(file); + } + catch (Exception ex) + { + Log.Error($"[PRODUCER] ERROR: {file.LocalPath}: {ex}"); + progress.Failed(ex, file); + } + }); } - catch (Exception ex) + finally + { + Log.Debug("[PRODUCER] COMPLETE"); + channel.Writer.Complete(); + } + }); + + Task monitor = Task.Run(async () => + { + Stopwatch sw = Stopwatch.StartNew(); + while (!Task.WhenAll(workerTasks).IsCompleted) { - Log.Error(ex.GetBaseException().ToString()); - progress.Failed(ex, file); + Log.Debug($"PIPELINE STATS @ {sw.Elapsed}: queued={queued}, completed={completed}, total={total}"); + await Task.Delay(1000); } }); + + Log.Debug($"PIPELINE COMPLETED: queued={queued}, completed={completed}, total={total}"); + await Task.WhenAll(producer, monitor).ConfigureAwait(false); + await Task.WhenAll(workerTasks).ConfigureAwait(false); } /// public override async Task PullFilesAsync(SystemFile[] files, IProgressReporter progress) { - await System.Threading.Tasks.Parallel.ForEachAsync(files, ParallelConfig.Options, async (file, ct) => + long queued = 0, completed = 0, total = 0; + TimeSpan downloadTimeout = TimeSpan.FromSeconds(30); + TimeSpan writeBlockWarn = TimeSpan.FromSeconds(5); + + Channel channel = Channel.CreateBounded(new BoundedChannelOptions(256) + { + FullMode = BoundedChannelFullMode.Wait, + SingleReader = false, + SingleWriter = false + }); + + /*Task[] workerTasks = Enumerable.Range(0, ParallelConfig.MaxStaticTransfers).Select(workerId -> Task.Run(async () => { try { - progress.Report(ProgressOperation.Pulling, file); - string? parentDir = Path.GetDirectoryName(file.LocalPath); - if (!Directory.Exists(parentDir)) Directory.CreateDirectory(parentDir); - - await using FileStream fs = File.Create(file.LocalPath); - foreach (string hash in await Database.GetObjectsAsync(file.Id)) + await foreach (DownloadWorker job in channel.Reader.ReadAllAsync()) { - string basePath = PathBuilder.Combine(RemoteVault.FileSystem.RootDirectory, "Parallel", RemoteVault.Id, "objects"); - string remotePath = PathBuilder.Combine(basePath, hash.Substring(0, 2), hash.Substring(2, 2), hash[4..]); - if (await Storage.ExistsAsync(remotePath)) + Interlocked.Decrement(ref queued); + + try { - await Storage.DownloadStreamAsync(fs, remotePath); + Interlocked.Increment(ref total); } } - - await Database.AddHistoryAsync(file.LocalPath, HistoryType.Pulled); - await fs.FlushAsync(ct); } - catch (Exception ex) - { - Log.Error(ex.GetBaseException().ToString()); - progress.Failed(ex, file); - } - }); + }));*/ } } } \ No newline at end of file diff --git a/Parallel.Core/IO/Syncing/SyncManager.cs b/Parallel.Core/IO/Syncing/SyncManager.cs index 842fe2d..1d91dc4 100644 --- a/Parallel.Core/IO/Syncing/SyncManager.cs +++ b/Parallel.Core/IO/Syncing/SyncManager.cs @@ -14,9 +14,9 @@ public static class SyncManager /// /// /// - public static ISyncManager CreateNew(LocalVaultConfig localVault) + public static ISyncManager? CreateNew(LocalVaultConfig? localVault) { - return new ObjectSyncManager(localVault); + return localVault?.Credentials is null ? null : new ObjectSyncManager(localVault); } } } \ No newline at end of file diff --git a/Parallel.Core/Parallel.Core.csproj b/Parallel.Core/Parallel.Core.csproj index fdd0ec4..81f4173 100644 --- a/Parallel.Core/Parallel.Core.csproj +++ b/Parallel.Core/Parallel.Core.csproj @@ -8,7 +8,7 @@ AnyCPU enable Parallel.Core - 1.0.0.0 + 0.0.0.0 Kyle Ebbinga Copyright $(Company). All Rights Reserved. $(AssemblyVersion) @@ -17,12 +17,12 @@ - - - - - - + + + + + + @@ -35,8 +35,8 @@ - - + + diff --git a/Parallel.Core/Settings/LocalVaultConfig.cs b/Parallel.Core/Settings/LocalVaultConfig.cs index 82d8203..8565a50 100644 --- a/Parallel.Core/Settings/LocalVaultConfig.cs +++ b/Parallel.Core/Settings/LocalVaultConfig.cs @@ -2,6 +2,7 @@ using Parallel.Core.IO.FileSystem; using Parallel.Core.Security; +using Parallel.Core.Storage; namespace Parallel.Core.Settings { @@ -28,27 +29,20 @@ public class LocalVaultConfig /// /// The credentials needed to log in to the associated . /// - public FileSystemCredentials FileSystem { get; } + public StorageCredentials Credentials { get; } /// /// Initializes a new instance of the class. /// /// /// - /// + /// [JsonConstructor] - public LocalVaultConfig(string id, string name, FileSystemCredentials fileSystem) + public LocalVaultConfig(string id, string name, StorageCredentials credentials) { Id = id; Name = name; - FileSystem = fileSystem; - } - - public LocalVaultConfig(string name, FileSystemCredentials fileSystem) - { - Id = HashGenerator.GenerateHash(8, true); - Name = name; - FileSystem = fileSystem; + Credentials = credentials; } /// diff --git a/Parallel.Core/Settings/ParallelConfig.cs b/Parallel.Core/Settings/ParallelConfig.cs index dc95650..0f89d30 100644 --- a/Parallel.Core/Settings/ParallelConfig.cs +++ b/Parallel.Core/Settings/ParallelConfig.cs @@ -21,12 +21,18 @@ public class ParallelConfig /// public static string VaultsDir { get; } = Path.Combine(PathBuilder.ProgramData, "Vaults"); + /// + /// Gets a set of static options for . + /// public static ParallelOptions Options { get; } = new ParallelOptions { - MaxDegreeOfParallelism = Load().MaxConcurrentProcesses + MaxDegreeOfParallelism = Math.Max(1, Load().MaxConcurrentProcesses) }; - public static int MaxUploads { get; } = Load().MaxConcurrentUploads; + /// + /// Gets the static value for . + /// + public static int MaxStaticTransfers { get; } = Math.Max(1, Load().MaxConcurrentTransfers); // /// // /// The address that will accept incoming commands. @@ -42,9 +48,9 @@ public class ParallelConfig /// /// Gets or sets the maximum number of concurrent vaults that can run. - /// Default: 2 + /// Default: 1 /// - public int MaxConcurrentVaults { get; set; } = 2; + public int MaxConcurrentVaults { get; set; } = 1; /// /// Gets or sets the maximum number of concurrent processes that can run. @@ -53,10 +59,10 @@ public class ParallelConfig public int MaxConcurrentProcesses { get; set; } = Environment.ProcessorCount; /// - /// Gets or sets the maximum number of concurrent processes that can run. + /// Gets or sets the maximum number of concurrent transfers that can run. /// Default: 4 /// - public int MaxConcurrentUploads { get; set; } = 4; + public int MaxConcurrentTransfers { get; set; } = 4; /// /// The amount of time, in days, to hold a file before it can be cleaned. @@ -75,7 +81,7 @@ public class ParallelConfig /// The profiles to use. /// When pulling, the CLI defaults to the first in the list. /// - public HashSet Vaults { get; } = []; + public HashSet Vaults { get; set; } = []; /// diff --git a/Parallel.Core/Settings/RemoteVaultConfig.cs b/Parallel.Core/Settings/RemoteVaultConfig.cs index d31cba8..a47406f 100644 --- a/Parallel.Core/Settings/RemoteVaultConfig.cs +++ b/Parallel.Core/Settings/RemoteVaultConfig.cs @@ -48,12 +48,10 @@ public class RemoteVaultConfig : LocalVaultConfig public HashSet PruneDirectories { get; } = []; - public RemoteVaultConfig(LocalVaultConfig localVault) : base(localVault.Id, localVault.Name, localVault.FileSystem) { } - - public RemoteVaultConfig(string profileName, FileSystemCredentials fsc) : base(profileName, fsc) { } + public RemoteVaultConfig(LocalVaultConfig localVault) : base(localVault.Id, localVault.Name, localVault.Credentials) { } [JsonConstructor] - public RemoteVaultConfig(string id, string name, FileSystemCredentials fileSystem, int backupInterval, int prunePeriod, IEnumerable backupDirectories, IEnumerable ignoreDirectories, IEnumerable pruneDirectories) : base(id, name, fileSystem) + public RemoteVaultConfig(string id, string name, StorageCredentials credentials, int backupInterval, int prunePeriod, IEnumerable backupDirectories, IEnumerable ignoreDirectories, IEnumerable pruneDirectories) : base(id, name, credentials) { BackupInterval = backupInterval; PrunePeriod = prunePeriod; diff --git a/Parallel.Core/Settings/FileSystemCredentials.cs b/Parallel.Core/Settings/StorageCredentials.cs similarity index 85% rename from Parallel.Core/Settings/FileSystemCredentials.cs rename to Parallel.Core/Settings/StorageCredentials.cs index ac7d4f9..4b0209c 100644 --- a/Parallel.Core/Settings/FileSystemCredentials.cs +++ b/Parallel.Core/Settings/StorageCredentials.cs @@ -3,13 +3,14 @@ using Newtonsoft.Json; using Newtonsoft.Json.Linq; using Parallel.Core.IO.FileSystem; +using Parallel.Core.Storage; namespace Parallel.Core.Settings { /// /// Represents credentials used to gain access to various s. /// - public class FileSystemCredentials + public class StorageCredentials { public FileService Service { get; set; } = FileService.Local; public string RootDirectory { get; set; } = string.Empty; @@ -27,9 +28,9 @@ public class FileSystemCredentials /// public string? EncryptionKey { get; set; } = null; - public FileSystemCredentials() { } + public StorageCredentials() { } - public FileSystemCredentials(string root) + public StorageCredentials(string root) { RootDirectory = root; } diff --git a/Parallel.Core/Storage/IStorageProvider.cs b/Parallel.Core/Storage/IStorageProvider.cs index 7c6ce3f..345f94a 100644 --- a/Parallel.Core/Storage/IStorageProvider.cs +++ b/Parallel.Core/Storage/IStorageProvider.cs @@ -1,15 +1,9 @@ // Copyright 2025 Kyle Ebbinga -using System; -using System.Collections.Generic; -using System.Linq; -using System.Text; -using System.Threading.Tasks; using Parallel.Core.Diagnostics; -using Parallel.Core.IO.Blobs; using Parallel.Core.Models; -namespace Parallel.Core.IO.FileSystem +namespace Parallel.Core.Storage { /// /// Defines the way for communicating with a file system. diff --git a/Parallel.Core/Storage/LocalStorageProvider.cs b/Parallel.Core/Storage/LocalStorageProvider.cs index 15f87e1..948611b 100644 --- a/Parallel.Core/Storage/LocalStorageProvider.cs +++ b/Parallel.Core/Storage/LocalStorageProvider.cs @@ -1,17 +1,11 @@ // Copyright 2025 Kyle Ebbinga -using System.Diagnostics; using System.IO.Compression; -using Microsoft.VisualBasic.FileIO; -using Newtonsoft.Json.Linq; using Parallel.Core.Diagnostics; -using Parallel.Core.IO.Blobs; using Parallel.Core.Models; using Parallel.Core.Settings; -using Parallel.Core.Utils; -using SearchOption = System.IO.SearchOption; -namespace Parallel.Core.IO.FileSystem +namespace Parallel.Core.Storage { /// /// Represents the wrapper for a default dotnet file system. @@ -84,7 +78,7 @@ public Task ExistsAsync(string path) /// public Task GetFileAsync(string path) { - if(!File.Exists(path)) return Task.FromResult(null); + if (!File.Exists(path)) return Task.FromResult(null); FileInfo fi = new(path); SystemFile file = new SystemFile(path) @@ -93,6 +87,7 @@ public Task ExistsAsync(string path) RemotePath = fi.FullName, RemoteSize = fi.Length }; + return Task.FromResult(file); } @@ -124,7 +119,7 @@ await System.Threading.Tasks.Parallel.ForEachAsync(files, ParallelConfig.Options /// public async Task UploadStreamAsync(Stream input, string remotePath) { - await using FileStream createStream = File.Create(remotePath); + await using FileStream createStream = File.OpenWrite(remotePath); await using GZipStream gzipStream = new GZipStream(createStream, CompressionLevel.SmallestSize); await input.CopyToAsync(gzipStream); diff --git a/Parallel.Core/Storage/SshStorageProvider.cs b/Parallel.Core/Storage/SshStorageProvider.cs index 28847c2..411e026 100644 --- a/Parallel.Core/Storage/SshStorageProvider.cs +++ b/Parallel.Core/Storage/SshStorageProvider.cs @@ -1,18 +1,15 @@ // Copyright 2025 Kyle Ebbinga using System.Diagnostics; -using Parallel.Core.Settings; -using Renci.SshNet; -using Renci.SshNet.Sftp; using System.IO.Compression; -using Newtonsoft.Json.Linq; using Parallel.Core.Diagnostics; -using Parallel.Core.IO.Blobs; using Parallel.Core.Models; using Parallel.Core.Security; -using Parallel.Core.Utils; +using Parallel.Core.Settings; +using Renci.SshNet; +using Renci.SshNet.Sftp; -namespace Parallel.Core.IO.FileSystem +namespace Parallel.Core.Storage { /// /// Represents the wrapper for an SFTP file system through SSH. @@ -28,7 +25,7 @@ public class SshStorageProvider : IStorageProvider /// The credentials to log in with. public SshStorageProvider(LocalVaultConfig localVault) { - _connectionInfo = new ConnectionInfo(localVault.FileSystem.Address, localVault.FileSystem.Username, new PasswordAuthenticationMethod(localVault.FileSystem.Username, Encryption.Decode(localVault.FileSystem.Password))); + _connectionInfo = new ConnectionInfo(localVault.Credentials.Address, localVault.Credentials.Username, new PasswordAuthenticationMethod(localVault.Credentials.Username, Encryption.Decode(localVault.Credentials.Password))); _client = new SftpClient(_connectionInfo); _client.Connect(); } @@ -123,7 +120,7 @@ public async Task UploadFilesAsync(SystemFile[] files, IProgressReporter progres string[] subDirs = file.RemotePath.Split('/'); string parentDir = string.Join("/", subDirs.Take(subDirs.Length - 1)); - if(!await _client.ExistsAsync(parentDir)) await CreateDirectoryAsync(parentDir); + if (!await _client.ExistsAsync(parentDir)) await CreateDirectoryAsync(parentDir); await using SftpFileStream createStream = _client.Create(file.RemotePath); await using FileStream openStream = File.OpenRead(file.LocalPath); diff --git a/Parallel.Core/Storage/StorageConnection.cs b/Parallel.Core/Storage/StorageConnection.cs new file mode 100644 index 0000000..6798286 --- /dev/null +++ b/Parallel.Core/Storage/StorageConnection.cs @@ -0,0 +1,49 @@ +// Copyright 2025 Kyle Ebbinga + +using Parallel.Core.Settings; +using Parallel.Core.Storage; + +namespace Parallel.Core.IO.FileSystem +{ + /// + /// The supported file service types. + /// + public enum FileService + { + /// + /// A local file system either through external or networked drives. + /// + Local, + + /// + /// A remote file storage server through secure shell. + /// + Remote, + + /// + /// A cloud storage server hosted through Amazon simple storage service. + /// + Cloud, + }; + + /// + /// Represents the way to connect to different file system associations. This class cannot be inherited. + /// + public static class StorageConnection + { + /// + /// Creates a new file system association. + /// + /// The vault needed for the associated file system. + public static IStorageProvider CreateNew(LocalVaultConfig vaultConfig) + { + return vaultConfig?.Credentials.Service switch + { + FileService.Local => new LocalStorageProvider(vaultConfig), + FileService.Remote => new SshStorageProvider(vaultConfig), + //FileService.Cloud => new AmazonS3FileSystem(credentials), + _ => throw new ArgumentOutOfRangeException() + }; + } + } +} \ No newline at end of file diff --git a/Parallel.Core/Storage/StorageProvider.cs b/Parallel.Core/Storage/StorageProvider.cs index 9ad1f64..5f28270 100644 --- a/Parallel.Core/Storage/StorageProvider.cs +++ b/Parallel.Core/Storage/StorageProvider.cs @@ -1,48 +1 @@ -// Copyright 2025 Kyle Ebbinga - -using Parallel.Core.Settings; - -namespace Parallel.Core.IO.FileSystem -{ - /// - /// The supported file service types. - /// - public enum FileService - { - /// - /// A local file system either through external or networked drives. - /// - Local, - - /// - /// A remote file storage server through secure shell. - /// - Remote, - - /// - /// A cloud storage server hosted through Amazon simple storage service. - /// - Cloud, - }; - - /// - /// Represents the way to connect to different file system associations. This class cannot be inherited. - /// - public static class StorageProvider - { - /// - /// Creates a new file system association. - /// - /// The vault needed for the associated file system. - public static IStorageProvider CreateNew(LocalVaultConfig vaultConfig) - { - return vaultConfig.FileSystem.Service switch - { - FileService.Local => new LocalStorageProvider(vaultConfig), - FileService.Remote => new SshStorageProvider(vaultConfig), - //FileService.Cloud => new AmazonS3FileSystem(credentials), - _ => null - }; - } - } -} \ No newline at end of file + \ No newline at end of file diff --git a/Parallel.Core/Workers/BaseWorker.cs b/Parallel.Core/Workers/BaseWorker.cs new file mode 100644 index 0000000..054833b --- /dev/null +++ b/Parallel.Core/Workers/BaseWorker.cs @@ -0,0 +1,33 @@ +// Copyright 2025 Kyle Ebbinga + +using Parallel.Core.Storage; + +namespace Parallel.Core.Workers +{ + /// + /// Represents + /// + public abstract class BaseWorker + { + /// + /// Gets the remote path in the . + /// + public string RemotePath { get; } + + /// + /// The to run when an is thrown. + /// + public Action? OnException { get; } + + /// + /// Initializes a new instance of the class with a remote path. + /// + /// + /// + protected BaseWorker(string remotePath, Action? onException) + { + RemotePath = remotePath; + OnException = onException; + } + } +} \ No newline at end of file diff --git a/Parallel.Core/Workers/DownloadWorker.cs b/Parallel.Core/Workers/DownloadWorker.cs new file mode 100644 index 0000000..b3fb2c8 --- /dev/null +++ b/Parallel.Core/Workers/DownloadWorker.cs @@ -0,0 +1,14 @@ +// Copyright 2025 Kyle Ebbinga + +namespace Parallel.Core.Workers +{ + public class DownloadWorker : BaseWorker + { + public FileAggregator Aggregator { get; } + + public DownloadWorker(string remotePath, Action? onException = null) : base(remotePath, onException) + { + + } + } +} \ No newline at end of file diff --git a/Parallel.Core/Workers/FileAggregator.cs b/Parallel.Core/Workers/FileAggregator.cs new file mode 100644 index 0000000..3410274 --- /dev/null +++ b/Parallel.Core/Workers/FileAggregator.cs @@ -0,0 +1,13 @@ +// Copyright 2025 Kyle Ebbinga + +namespace Parallel.Core.Workers +{ + public class FileAggregator + { + private readonly string _localPath; + private readonly byte[][] _chunks; + private readonly Action? _onComplete; + private readonly Action? _onError; + private int _count; + } +} \ No newline at end of file diff --git a/Parallel.Core/Workers/UploadWorker.cs b/Parallel.Core/Workers/UploadWorker.cs new file mode 100644 index 0000000..6ee3918 --- /dev/null +++ b/Parallel.Core/Workers/UploadWorker.cs @@ -0,0 +1,22 @@ +// Copyright 2025 Kyle Ebbinga + +namespace Parallel.Core.Workers +{ + /// + /// Represents a worker responsible for uploading files. + /// + public class UploadWorker : BaseWorker + { + public byte[] Data { get; } + + public string Filename { get; } + + + public UploadWorker(byte[] data, string filename, string remotePath, Action? onError = null) : base(remotePath, onError) + { + Data = data; + Filename = filename; + + } + } +} \ No newline at end of file