SolarNetwork/Swarm
2
3
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

⚗️ Testing out the File Storing System v2

Browse Source
This commit is contained in:
LittleSheep
2026-01-10 16:54:22 +08:00
parent 8177bda232
commit 98c100c864
19 changed files with 1353 additions and 1035 deletions
Download Patch File Download Diff File Expand all files Collapse all files

15
DysonNetwork.Drive/Storage/CloudFileUnusedRecyclingJob.cs
View File

@@ -26,7 +26,7 @@ public class CloudFileUnusedRecyclingJob(
File.Delete(file);
}
}
logger.LogInformation("Marking unused cloud files...");
var recyclablePools = await db.Pools
@@ -47,7 +47,7 @@ public class CloudFileUnusedRecyclingJob(
logger.LogInformation("Found {TotalFiles} files to check for unused status", totalFiles);
// Define a timestamp to limit the age of files we're processing in this run
// This spreads the processing across multiple job runs for very large databases
// This spreads processing across multiple job runs for very large databases
var ageThreshold = now - Duration.FromDays(30); // Process files up to 90 days old in this run
// Instead of loading all files at once, use pagination
@@ -80,13 +80,12 @@ public class CloudFileUnusedRecyclingJob(
processedCount += fileBatch.Count;
lastProcessedId = fileBatch.Last();
// Optimized query: Find files that have no references OR all references are expired
// This replaces the memory-intensive approach of loading all references
// Optimized query: Find files that have no other cloud files sharing the same object
// A file is considered "unused" if no other SnCloudFile shares its ObjectId
var filesToMark = await db.Files
.Where(f => fileBatch.Contains(f.Id))
.Where(f => !db.FileReferences.Any(r => r.FileId == f.Id) || // No references at all
!db.FileReferences.Any(r => r.FileId == f.Id && // OR has references but all are expired
(r.ExpiredAt == null || r.ExpiredAt > now)))
.Where(f => f.ObjectId == null || // No file object at all
!db.Files.Any(cf => cf.ObjectId == f.ObjectId && cf.Id != f.Id)) // Or no other files share this object
.Select(f => f.Id)
.ToListAsync();
@@ -112,7 +111,7 @@ public class CloudFileUnusedRecyclingJob(
);
}
}
var expiredCount = await db.Files
.Where(f => f.ExpiredAt.HasValue && f.ExpiredAt.Value <= now)
.ExecuteUpdateAsync(s => s.SetProperty(f => f.IsMarkedRecycle, true));

13
DysonNetwork.Drive/Storage/FileController.cs
View File

@@ -14,8 +14,7 @@ public class FileController(
AppDatabase db,
FileService fs,
IConfiguration configuration,
IWebHostEnvironment env,
FileReferenceService fileReferenceService
IWebHostEnvironment env
) : ControllerBase
{
[HttpGet("{id}")]
@@ -232,17 +231,19 @@ public class FileController(
}
[HttpGet("{id}/references")]
public async Task<ActionResult<List<Shared.Models.SnCloudFileReference>>> GetFileReferences(string id)
public async Task<ActionResult<List<SnCloudFile>>> GetFileReferences(string id)
{
var file = await fs.GetFileAsync(id);
if (file is null) return NotFound("File not found.");
// Check if user has access to the file
// Check if user has access to
var accessResult = await ValidateFileAccess(file, null);
if (accessResult is not null) return accessResult;
// Get references using the injected FileReferenceService
var references = await fileReferenceService.GetReferencesAsync(id);
// Get other cloud files sharing the same object
var references = await db.Files
.Where(f => f.ObjectId == file.ObjectId && f.Id != file.Id)
.ToListAsync();
return Ok(references);
}

70
DysonNetwork.Drive/Storage/FileExpirationJob.cs
View File

@@ -1,70 +0,0 @@
using Microsoft.EntityFrameworkCore;
using NodaTime;
using Quartz;
namespace DysonNetwork.Drive.Storage;
/// <summary>
/// Job responsible for cleaning up expired file references
/// </summary>
public class FileExpirationJob(AppDatabase db, FileService fileService, ILogger<FileExpirationJob> logger) : IJob
{
public async Task Execute(IJobExecutionContext context)
{
var now = SystemClock.Instance.GetCurrentInstant();
logger.LogInformation("Running file reference expiration job at {now}", now);
// Delete expired references in bulk and get affected file IDs
var affectedFileIds = await db.FileReferences
.Where(r => r.ExpiredAt < now && r.ExpiredAt != null)
.Select(r => r.FileId)
.Distinct()
.ToListAsync();
if (!affectedFileIds.Any())
{
logger.LogInformation("No expired file references found");
return;
}
logger.LogInformation("Found expired references for {count} files", affectedFileIds.Count);
// Delete expired references in bulk
var deletedReferencesCount = await db.FileReferences
.Where(r => r.ExpiredAt < now && r.ExpiredAt != null)
.ExecuteDeleteAsync();
logger.LogInformation("Deleted {count} expired file references", deletedReferencesCount);
// Find files that now have no remaining references (bulk operation)
var filesToDelete = await db.Files
.Where(f => affectedFileIds.Contains(f.Id))
.Where(f => !db.FileReferences.Any(r => r.FileId == f.Id))
.Select(f => f.Id)
.ToListAsync();
if (filesToDelete.Any())
{
logger.LogInformation("Deleting {count} files that have no remaining references", filesToDelete.Count);
// Get files for deletion
var files = await db.Files
.Where(f => filesToDelete.Contains(f.Id))
.ToListAsync();
// Delete files and their data in parallel
var deleteTasks = files.Select(f => fileService.DeleteFileAsync(f));
await Task.WhenAll(deleteTasks);
}
// Purge cache for files that still have references
var filesWithRemainingRefs = affectedFileIds.Except(filesToDelete).ToList();
if (filesWithRemainingRefs.Any())
{
var cachePurgeTasks = filesWithRemainingRefs.Select(fileService._PurgeCacheAsync);
await Task.WhenAll(cachePurgeTasks);
}
logger.LogInformation("Completed file reference expiration job");
}
}

101
DysonNetwork.Drive/Storage/FileObjectCleanupJob.cs Normal file
View File

@@ -0,0 +1,101 @@
using Microsoft.EntityFrameworkCore;
using Minio.DataModel.Args;
using NodaTime;
using Quartz;
namespace DysonNetwork.Drive.Storage;
/// <summary>
/// Job responsible for cleaning up orphaned file objects
/// When no SnCloudFile references a SnFileObject, the file object is considered orphaned
/// and should be deleted from disk and database
/// </summary>
public class FileObjectCleanupJob(AppDatabase db, FileService fileService, ILogger<FileObjectCleanupJob> logger) : IJob
{
public async Task Execute(IJobExecutionContext context)
{
var now = SystemClock.Instance.GetCurrentInstant();
logger.LogInformation("Running file object cleanup job at {now}", now);
// Find orphaned file objects (objects with no cloud files referencing them)
var referencedObjectIds = await db.Files
.Where(f => f.ObjectId != null)
.Select(f => f.ObjectId)
.Distinct()
.ToListAsync();
var orphanedObjects = await db.FileObjects
.Where(fo => !referencedObjectIds.Contains(fo.Id))
.ToListAsync();
if (!orphanedObjects.Any())
{
logger.LogInformation("No orphaned file objects found");
return;
}
logger.LogInformation("Found {count} orphaned file objects", orphanedObjects.Count);
// Delete orphaned objects and their data
foreach (var fileObject in orphanedObjects)
{
try
{
var replicas = await db.FileReplicas
.Where(r => r.ObjectId == fileObject.Id)
.ToListAsync();
foreach (var replica in replicas)
{
var dest = await fileService.GetRemoteStorageConfig(replica.PoolId);
if (dest != null)
{
var client = fileService.CreateMinioClient(dest);
if (client != null)
{
try
{
await client.RemoveObjectAsync(
new RemoveObjectArgs()
.WithBucket(dest.Bucket)
.WithObject(replica.StorageId)
);
if (fileObject.HasCompression)
{
await client.RemoveObjectAsync(
new RemoveObjectArgs()
.WithBucket(dest.Bucket)
.WithObject(replica.StorageId + ".compressed")
);
}
if (fileObject.HasThumbnail)
{
await client.RemoveObjectAsync(
new RemoveObjectArgs()
.WithBucket(dest.Bucket)
.WithObject(replica.StorageId + ".thumbnail")
);
}
}
catch (Exception ex)
{
logger.LogError(ex, "Failed to delete orphaned file object {ObjectId} from remote storage", fileObject.Id);
}
}
}
}
db.FileReplicas.RemoveRange(replicas);
db.FileObjects.Remove(fileObject);
await db.SaveChangesAsync();
logger.LogInformation("Deleted orphaned file object {ObjectId}", fileObject.Id);
}
catch (Exception ex)
{
logger.LogError(ex, "Failed to clean up orphaned file object {ObjectId}", fileObject.Id);
}
}
logger.LogInformation("Completed file object cleanup job");
}
}

532
DysonNetwork.Drive/Storage/FileReferenceService.cs
View File

@@ -1,532 +0,0 @@
using DysonNetwork.Shared.Cache;
using DysonNetwork.Shared.Data;
using DysonNetwork.Shared.Models;
using Microsoft.EntityFrameworkCore;
using NodaTime;
namespace DysonNetwork.Drive.Storage;
public class FileReferenceService(AppDatabase db, FileService fileService, ICacheService cache)
{
private const string CacheKeyPrefix = "file:ref:";
private static readonly TimeSpan CacheDuration = TimeSpan.FromMinutes(15);
/// <summary>
/// Creates a new reference to a file for a specific resource
/// </summary>
/// <param name="fileId">The ID of the file to reference</param>
/// <param name="usage">The usage context (e.g., "avatar", "post-attachment")</param>
/// <param name="resourceId">The ID of the resource using the file</param>
/// <param name="expiredAt">Optional expiration time for the file</param>
/// <param name="duration">Optional duration after which the file expires (alternative to expiredAt)</param>
/// <returns>The created file reference</returns>
public async Task<SnCloudFileReference> CreateReferenceAsync(
string fileId,
string usage,
string resourceId,
Instant? expiredAt = null,
Duration? duration = null
)
{
// Calculate expiration time if needed
var finalExpiration = expiredAt;
if (duration.HasValue)
finalExpiration = SystemClock.Instance.GetCurrentInstant() + duration.Value;
var reference = new SnCloudFileReference
{
FileId = fileId,
Usage = usage,
ResourceId = resourceId,
ExpiredAt = finalExpiration
};
db.FileReferences.Add(reference);
await db.SaveChangesAsync();
await fileService._PurgeCacheAsync(fileId);
return reference;
}
public async Task<List<SnCloudFileReference>> CreateReferencesAsync(
List<string> fileId,
string usage,
string resourceId,
Instant? expiredAt = null,
Duration? duration = null
)
{
var now = SystemClock.Instance.GetCurrentInstant();
var finalExpiredAt = expiredAt;
if (finalExpiredAt == null && duration.HasValue)
{
finalExpiredAt = now + duration.Value;
}
var data = fileId.Select(id => new SnCloudFileReference
{
FileId = id,
Usage = usage,
ResourceId = resourceId,
ExpiredAt = finalExpiredAt,
CreatedAt = now,
UpdatedAt = now
})
.ToList();
db.FileReferences.AddRange(data);
await db.SaveChangesAsync();
return data;
}
/// <summary>
/// Gets all references to a file
/// </summary>
/// <param name="fileId">The ID of the file</param>
/// <returns>A list of all references to the file</returns>
public async Task<List<SnCloudFileReference>> GetReferencesAsync(string fileId)
{
var cacheKey = $"{CacheKeyPrefix}list:{fileId}";
var cachedReferences = await cache.GetAsync<List<SnCloudFileReference>>(cacheKey);
if (cachedReferences is not null)
return cachedReferences;
var references = await db.FileReferences
.Where(r => r.FileId == fileId)
.ToListAsync();
await cache.SetAsync(cacheKey, references, CacheDuration);
return references;
}
public async Task<Dictionary<string, List<SnCloudFileReference>>> GetReferencesAsync(IEnumerable<string> fileIds)
{
var fileIdList = fileIds.ToList();
var result = new Dictionary<string, List<SnCloudFileReference>>();
// Check cache for each file ID
var uncachedFileIds = new List<string>();
foreach (var fileId in fileIdList)
{
var cacheKey = $"{CacheKeyPrefix}list:{fileId}";
var cachedReferences = await cache.GetAsync<List<SnCloudFileReference>>(cacheKey);
if (cachedReferences is not null)
{
result[fileId] = cachedReferences;
}
else
{
uncachedFileIds.Add(fileId);
}
}
// Fetch uncached references from database
if (uncachedFileIds.Any())
{
var dbReferences = await db.FileReferences
.Where(r => uncachedFileIds.Contains(r.FileId))
.GroupBy(r => r.FileId)
.ToDictionaryAsync(r => r.Key, r => r.ToList());
// Cache the results
foreach (var kvp in dbReferences)
{
var cacheKey = $"{CacheKeyPrefix}list:{kvp.Key}";
await cache.SetAsync(cacheKey, kvp.Value, CacheDuration);
result[kvp.Key] = kvp.Value;
}
}
return result;
}
/// <summary>
/// Gets the number of references to a file
/// </summary>
/// <param name="fileId">The ID of the file</param>
/// <returns>The number of references to the file</returns>
public async Task<int> GetReferenceCountAsync(string fileId)
{
var cacheKey = $"{CacheKeyPrefix}count:{fileId}";
var cachedCount = await cache.GetAsync<int?>(cacheKey);
if (cachedCount.HasValue)
return cachedCount.Value;
var count = await db.FileReferences
.Where(r => r.FileId == fileId)
.CountAsync();
await cache.SetAsync(cacheKey, count, CacheDuration);
return count;
}
/// <summary>
/// Gets all references for a specific resource
/// </summary>
/// <param name="resourceId">The ID of the resource</param>
/// <returns>A list of file references associated with the resource</returns>
public async Task<List<SnCloudFileReference>> GetResourceReferencesAsync(string resourceId)
{
var cacheKey = $"{CacheKeyPrefix}resource:{resourceId}";
var cachedReferences = await cache.GetAsync<List<SnCloudFileReference>>(cacheKey);
if (cachedReferences is not null)
return cachedReferences;
var references = await db.FileReferences
.Where(r => r.ResourceId == resourceId)
.ToListAsync();
await cache.SetAsync(cacheKey, references, CacheDuration);
return references;
}
/// <summary>
/// Gets all file references for a specific usage context
/// </summary>
/// <param name="usage">The usage context</param>
/// <returns>A list of file references with the specified usage</returns>
public async Task<List<SnCloudFileReference>> GetUsageReferencesAsync(string usage)
{
var cacheKey = $"{CacheKeyPrefix}usage:{usage}";
var cachedReferences = await cache.GetAsync<List<SnCloudFileReference>>(cacheKey);
if (cachedReferences is not null)
return cachedReferences;
var references = await db.FileReferences
.Where(r => r.Usage == usage)
.ToListAsync();
await cache.SetAsync(cacheKey, references, CacheDuration);
return references;
}
/// <summary>
/// Deletes references for a specific resource
/// </summary>
/// <param name="resourceId">The ID of the resource</param>
/// <returns>The number of deleted references</returns>
public async Task<int> DeleteResourceReferencesAsync(string resourceId)
{
var references = await db.FileReferences
.Where(r => r.ResourceId == resourceId)
.ToListAsync();
var fileIds = references.Select(r => r.FileId).Distinct().ToList();
db.FileReferences.RemoveRange(references);
var deletedCount = await db.SaveChangesAsync();
// Purge caches
var tasks = fileIds.Select(fileService._PurgeCacheAsync).ToList();
tasks.Add(PurgeCacheForResourceAsync(resourceId));
await Task.WhenAll(tasks);
return deletedCount;
}
/// <summary>
/// Deletes references for a specific resource and usage
/// </summary>
/// <param name="resourceId">The ID of the resource</param>
/// <param name="usage">The usage context</param>
/// <returns>The number of deleted references</returns>
public async Task<int> DeleteResourceReferencesAsync(string resourceId, string usage)
{
var references = await db.FileReferences
.Where(r => r.ResourceId == resourceId && r.Usage == usage)
.ToListAsync();
if (references.Count == 0)
return 0;
var fileIds = references.Select(r => r.FileId).Distinct().ToList();
db.FileReferences.RemoveRange(references);
var deletedCount = await db.SaveChangesAsync();
// Purge caches
var tasks = fileIds.Select(fileService._PurgeCacheAsync).ToList();
tasks.Add(PurgeCacheForResourceAsync(resourceId));
await Task.WhenAll(tasks);
return deletedCount;
}
public async Task<int> DeleteResourceReferencesBatchAsync(IEnumerable<string> resourceIds, string? usage = null)
{
var resourceIdList = resourceIds.ToList();
var references = await db.FileReferences
.Where(r => resourceIdList.Contains(r.ResourceId))
.If(usage != null, q => q.Where(q => q.Usage == usage))
.ToListAsync();
if (references.Count == 0)
return 0;
var fileIds = references.Select(r => r.FileId).Distinct().ToList();
db.FileReferences.RemoveRange(references);
var deletedCount = await db.SaveChangesAsync();
// Purge caches for files and resources
var tasks = fileIds.Select(fileService._PurgeCacheAsync).ToList();
tasks.AddRange(resourceIdList.Select(PurgeCacheForResourceAsync));
await Task.WhenAll(tasks);
return deletedCount;
}
/// <summary>
/// Deletes a specific file reference
/// </summary>
/// <param name="referenceId">The ID of the reference to delete</param>
/// <returns>True if the reference was deleted, false otherwise</returns>
public async Task<bool> DeleteReferenceAsync(Guid referenceId)
{
var reference = await db.FileReferences
.FirstOrDefaultAsync(r => r.Id == referenceId);
if (reference == null)
return false;
db.FileReferences.Remove(reference);
await db.SaveChangesAsync();
// Purge caches
await fileService._PurgeCacheAsync(reference.FileId);
await PurgeCacheForResourceAsync(reference.ResourceId);
await PurgeCacheForFileAsync(reference.FileId);
return true;
}
/// <summary>
/// Updates the files referenced by a resource
/// </summary>
/// <param name="resourceId">The ID of the resource</param>
/// <param name="newFileIds">The new list of file IDs</param>
/// <param name="usage">The usage context</param>
/// <param name="expiredAt">Optional expiration time for newly added files</param>
/// <param name="duration">Optional duration after which newly added files expire</param>
/// <returns>A list of the updated file references</returns>
public async Task<List<SnCloudFileReference>> UpdateResourceFilesAsync(
string resourceId,
IEnumerable<string>? newFileIds,
string usage,
Instant? expiredAt = null,
Duration? duration = null)
{
if (newFileIds == null)
return new List<SnCloudFileReference>();
var existingReferences = await db.FileReferences
.Where(r => r.ResourceId == resourceId && r.Usage == usage)
.ToListAsync();
var existingFileIds = existingReferences.Select(r => r.FileId).ToHashSet();
var newFileIdsList = newFileIds.ToList();
var newFileIdsSet = newFileIdsList.ToHashSet();
// Files to remove
var toRemove = existingReferences
.Where(r => !newFileIdsSet.Contains(r.FileId))
.ToList();
// Files to add
var toAdd = newFileIdsList
.Where(id => !existingFileIds.Contains(id))
.Select(id => new SnCloudFileReference
{
FileId = id,
Usage = usage,
ResourceId = resourceId
})
.ToList();
// Apply changes
if (toRemove.Any())
db.FileReferences.RemoveRange(toRemove);
if (toAdd.Any())
db.FileReferences.AddRange(toAdd);
await db.SaveChangesAsync();
// Update expiration for newly added references if specified
if ((expiredAt.HasValue || duration.HasValue) && toAdd.Any())
{
var finalExpiration = expiredAt;
if (duration.HasValue)
{
finalExpiration = SystemClock.Instance.GetCurrentInstant() + duration.Value;
}
// Update newly added references with the expiration time
var referenceIds = await db.FileReferences
.Where(r => toAdd.Select(a => a.FileId).Contains(r.FileId) &&
r.ResourceId == resourceId &&
r.Usage == usage)
.Select(r => r.Id)
.ToListAsync();
await db.FileReferences
.Where(r => referenceIds.Contains(r.Id))
.ExecuteUpdateAsync(setter => setter.SetProperty(
r => r.ExpiredAt,
_ => finalExpiration
));
}
// Purge caches
var allFileIds = existingFileIds.Union(newFileIdsSet).ToList();
var tasks = allFileIds.Select(fileService._PurgeCacheAsync).ToList();
tasks.Add(PurgeCacheForResourceAsync(resourceId));
await Task.WhenAll(tasks);
// Return updated references
return await db.FileReferences
.Where(r => r.ResourceId == resourceId && r.Usage == usage)
.ToListAsync();
}
/// <summary>
/// Gets all files referenced by a resource
/// </summary>
/// <param name="resourceId">The ID of the resource</param>
/// <param name="usage">Optional filter by usage context</param>
/// <returns>A list of files referenced by the resource</returns>
public async Task<List<SnCloudFile>> GetResourceFilesAsync(string resourceId, string? usage = null)
{
var query = db.FileReferences.Where(r => r.ResourceId == resourceId);
if (usage != null)
query = query.Where(r => r.Usage == usage);
var references = await query.ToListAsync();
var fileIds = references.Select(r => r.FileId).ToList();
return await db.Files
.Where(f => fileIds.Contains(f.Id))
.ToListAsync();
}
/// <summary>
/// Purges all caches related to a resource
/// </summary>
private async Task PurgeCacheForResourceAsync(string resourceId)
{
var cacheKey = $"{CacheKeyPrefix}resource:{resourceId}";
await cache.RemoveAsync(cacheKey);
}
/// <summary>
/// Purges all caches related to a file
/// </summary>
private async Task PurgeCacheForFileAsync(string fileId)
{
var cacheKeys = new[]
{
$"{CacheKeyPrefix}list:{fileId}",
$"{CacheKeyPrefix}count:{fileId}"
};
var tasks = cacheKeys.Select(cache.RemoveAsync);
await Task.WhenAll(tasks);
}
/// <summary>
/// Updates the expiration time for a file reference
/// </summary>
/// <param name="referenceId">The ID of the reference</param>
/// <param name="expiredAt">The new expiration time, or null to remove expiration</param>
/// <returns>True if the reference was found and updated, false otherwise</returns>
public async Task<bool> SetReferenceExpirationAsync(Guid referenceId, Instant? expiredAt)
{
var reference = await db.FileReferences
.FirstOrDefaultAsync(r => r.Id == referenceId);
if (reference == null)
return false;
reference.ExpiredAt = expiredAt;
await db.SaveChangesAsync();
await PurgeCacheForFileAsync(reference.FileId);
await PurgeCacheForResourceAsync(reference.ResourceId);
return true;
}
/// <summary>
/// Updates the expiration time for all references to a file
/// </summary>
/// <param name="fileId">The ID of the file</param>
/// <param name="expiredAt">The new expiration time, or null to remove expiration</param>
/// <returns>The number of references updated</returns>
public async Task<int> SetFileReferencesExpirationAsync(string fileId, Instant? expiredAt)
{
var rowsAffected = await db.FileReferences
.Where(r => r.FileId == fileId)
.ExecuteUpdateAsync(setter => setter.SetProperty(
r => r.ExpiredAt,
_ => expiredAt
));
if (rowsAffected > 0)
{
await fileService._PurgeCacheAsync(fileId);
await PurgeCacheForFileAsync(fileId);
}
return rowsAffected;
}
/// <summary>
/// Get all file references for a specific resource and usage type
/// </summary>
/// <param name="resourceId">The resource ID</param>
/// <param name="usageType">The usage type</param>
/// <returns>List of file references</returns>
public async Task<List<SnCloudFileReference>> GetResourceReferencesAsync(string resourceId, string usageType)
{
return await db.FileReferences
.Where(r => r.ResourceId == resourceId && r.Usage == usageType)
.ToListAsync();
}
/// <summary>
/// Check if a file has any references
/// </summary>
/// <param name="fileId">The file ID to check</param>
/// <returns>True if the file has references, false otherwise</returns>
public async Task<bool> HasFileReferencesAsync(string fileId)
{
return await db.FileReferences.AnyAsync(r => r.FileId == fileId);
}
/// <summary>
/// Updates the expiration time for a file reference using a duration from now
/// </summary>
/// <param name="referenceId">The ID of the reference</param>
/// <param name="duration">The duration after which the reference expires, or null to remove expiration</param>
/// <returns>True if the reference was found and updated, false otherwise</returns>
public async Task<bool> SetReferenceExpirationDurationAsync(Guid referenceId, Duration? duration)
{
Instant? expiredAt = null;
if (duration.HasValue)
{
expiredAt = SystemClock.Instance.GetCurrentInstant() + duration.Value;
}
return await SetReferenceExpirationAsync(referenceId, expiredAt);
}
}

174
DysonNetwork.Drive/Storage/FileReferenceServiceGrpc.cs
View File

@@ -1,174 +0,0 @@
using DysonNetwork.Shared.Proto;
using Grpc.Core;
using NodaTime;
using Duration = NodaTime.Duration;
namespace DysonNetwork.Drive.Storage;
public class FileReferenceServiceGrpc(FileReferenceService fileReferenceService)
: Shared.Proto.FileReferenceService.FileReferenceServiceBase
{
public override async Task<Shared.Proto.CloudFileReference> CreateReference(CreateReferenceRequest request,
ServerCallContext context)
{
Instant? expiredAt = null;
if (request.ExpiredAt != null)
expiredAt = Instant.FromUnixTimeSeconds(request.ExpiredAt.Seconds);
else if (request.Duration != null)
expiredAt = SystemClock.Instance.GetCurrentInstant() +
Duration.FromTimeSpan(request.Duration.ToTimeSpan());
var reference = await fileReferenceService.CreateReferenceAsync(
request.FileId,
request.Usage,
request.ResourceId,
expiredAt
);
return reference.ToProtoValue();
}
public override async Task<CreateReferenceBatchResponse> CreateReferenceBatch(CreateReferenceBatchRequest request,
ServerCallContext context)
{
Instant? expiredAt = null;
if (request.ExpiredAt != null)
expiredAt = Instant.FromUnixTimeSeconds(request.ExpiredAt.Seconds);
else if (request.Duration != null)
expiredAt = SystemClock.Instance.GetCurrentInstant() +
Duration.FromTimeSpan(request.Duration.ToTimeSpan());
var references = await fileReferenceService.CreateReferencesAsync(
request.FilesId.ToList(),
request.Usage,
request.ResourceId,
expiredAt
);
var response = new CreateReferenceBatchResponse();
response.References.AddRange(references.Select(r => r.ToProtoValue()));
return response;
}
public override async Task<GetReferencesResponse> GetReferences(GetReferencesRequest request,
ServerCallContext context)
{
var references = await fileReferenceService.GetReferencesAsync(request.FileId);
var response = new GetReferencesResponse();
response.References.AddRange(references.Select(r => r.ToProtoValue()));
return response;
}
public override async Task<GetReferenceCountResponse> GetReferenceCount(GetReferenceCountRequest request,
ServerCallContext context)
{
var count = await fileReferenceService.GetReferenceCountAsync(request.FileId);
return new GetReferenceCountResponse { Count = count };
}
public override async Task<GetReferencesResponse> GetResourceReferences(GetResourceReferencesRequest request,
ServerCallContext context)
{
var references = await fileReferenceService.GetResourceReferencesAsync(request.ResourceId, request.Usage);
var response = new GetReferencesResponse();
response.References.AddRange(references.Select(r => r.ToProtoValue()));
return response;
}
public override async Task<GetResourceFilesResponse> GetResourceFiles(GetResourceFilesRequest request,
ServerCallContext context)
{
var files = await fileReferenceService.GetResourceFilesAsync(request.ResourceId, request.Usage);
var response = new GetResourceFilesResponse();
response.Files.AddRange(files.Select(f => f.ToProtoValue()));
return response;
}
public override async Task<DeleteResourceReferencesResponse> DeleteResourceReferences(
DeleteResourceReferencesRequest request, ServerCallContext context)
{
int deletedCount;
if (request.Usage is null)
deletedCount = await fileReferenceService.DeleteResourceReferencesAsync(request.ResourceId);
else
deletedCount =
await fileReferenceService.DeleteResourceReferencesAsync(request.ResourceId, request.Usage!);
return new DeleteResourceReferencesResponse { DeletedCount = deletedCount };
}
public override async Task<DeleteResourceReferencesResponse> DeleteResourceReferencesBatch(DeleteResourceReferencesBatchRequest request, ServerCallContext context)
{
var resourceIds = request.ResourceIds.ToList();
int deletedCount;
if (request.Usage is null)
deletedCount = await fileReferenceService.DeleteResourceReferencesBatchAsync(resourceIds);
else
deletedCount =
await fileReferenceService.DeleteResourceReferencesBatchAsync(resourceIds, request.Usage!);
return new DeleteResourceReferencesResponse { DeletedCount = deletedCount };
}
public override async Task<DeleteReferenceResponse> DeleteReference(DeleteReferenceRequest request,
ServerCallContext context)
{
var success = await fileReferenceService.DeleteReferenceAsync(Guid.Parse(request.ReferenceId));
return new DeleteReferenceResponse { Success = success };
}
public override async Task<UpdateResourceFilesResponse> UpdateResourceFiles(UpdateResourceFilesRequest request,
ServerCallContext context)
{
Instant? expiredAt = null;
if (request.ExpiredAt != null)
{
expiredAt = Instant.FromUnixTimeSeconds(request.ExpiredAt.Seconds);
}
else if (request.Duration != null)
{
expiredAt = SystemClock.Instance.GetCurrentInstant() +
Duration.FromTimeSpan(request.Duration.ToTimeSpan());
}
var references = await fileReferenceService.UpdateResourceFilesAsync(
request.ResourceId,
request.FileIds,
request.Usage,
expiredAt
);
var response = new UpdateResourceFilesResponse();
response.References.AddRange(references.Select(r => r.ToProtoValue()));
return response;
}
public override async Task<SetReferenceExpirationResponse> SetReferenceExpiration(
SetReferenceExpirationRequest request, ServerCallContext context)
{
Instant? expiredAt = null;
if (request.ExpiredAt != null)
{
expiredAt = Instant.FromUnixTimeSeconds(request.ExpiredAt.Seconds);
}
else if (request.Duration != null)
{
expiredAt = SystemClock.Instance.GetCurrentInstant() +
Duration.FromTimeSpan(request.Duration.ToTimeSpan());
}
var success =
await fileReferenceService.SetReferenceExpirationAsync(Guid.Parse(request.ReferenceId), expiredAt);
return new SetReferenceExpirationResponse { Success = success };
}
public override async Task<SetFileReferencesExpirationResponse> SetFileReferencesExpiration(
SetFileReferencesExpirationRequest request, ServerCallContext context)
{
var expiredAt = Instant.FromUnixTimeSeconds(request.ExpiredAt.Seconds);
var updatedCount = await fileReferenceService.SetFileReferencesExpirationAsync(request.FileId, expiredAt);
return new SetFileReferencesExpirationResponse { UpdatedCount = updatedCount };
}
public override async Task<HasFileReferencesResponse> HasFileReferences(HasFileReferencesRequest request,
ServerCallContext context)
{
var hasReferences = await fileReferenceService.HasFileReferencesAsync(request.FileId);
return new HasFileReferencesResponse { HasReferences = hasReferences };
}
}

130
DysonNetwork.Drive/Storage/FileService.cs
View File

@@ -40,6 +40,8 @@ public class FileService(
.Where(f => f.Id == fileId)
.Include(f => f.Pool)
.Include(f => f.Bundle)
.Include(f => f.Object)
.ThenInclude(o => o.FileReplicas)
.FirstOrDefaultAsync();
if (file != null)
@@ -69,6 +71,8 @@ public class FileService(
var dbFiles = await db.Files
.Where(f => uncachedIds.Contains(f.Id))
.Include(f => f.Pool)
.Include(f => f.Object)
.ThenInclude(o => o.FileReplicas)
.ToListAsync();
foreach (var file in dbFiles)
@@ -228,8 +232,34 @@ public class FileService(
private async Task SaveFileToDatabaseAsync(SnCloudFile file)
{
var fileObject = new SnFileObject
{
Id = file.Id,
AccountId = file.AccountId,
Size = file.Size,
Meta = file.FileMeta,
MimeType = file.MimeType,
Hash = file.Hash,
HasCompression = file.HasCompression,
HasThumbnail = file.HasThumbnail
};
var replica = new SnFileReplica
{
Id = Guid.NewGuid(),
ObjectId = file.Id,
PoolId = file.PoolId!.Value,
StorageId = file.StorageId ?? file.Id,
Status = SnFileReplicaStatus.Available,
IsPrimary = true
};
db.Files.Add(file);
db.FileObjects.Add(fileObject);
db.FileReplicas.Add(replica);
await db.SaveChangesAsync();
file.ObjectId = file.Id;
file.StorageId ??= file.Id;
}
@@ -470,8 +500,20 @@ public class FileService(
await db.Files.Where(f => f.Id == file.Id).ExecuteUpdateAsync(updatable.ToSetPropertyCalls());
if (updateMask.Paths.Contains("file_meta"))
{
await db.FileObjects
.Where(fo => fo.Id == file.ObjectId)
.ExecuteUpdateAsync(setter => setter
.SetProperty(fo => fo.Meta, file.FileMeta));
}
await _PurgeCacheAsync(file.Id);
return await db.Files.AsNoTracking().FirstAsync(f => f.Id == file.Id);
return await db.Files
.AsNoTracking()
.Include(f => f.Object)
.ThenInclude(o => o.FileReplicas)
.FirstAsync(f => f.Id == file.Id);
}
public async Task DeleteFileAsync(SnCloudFile file, bool skipData = false)
@@ -481,17 +523,23 @@ public class FileService(
await _PurgeCacheAsync(file.Id);
if (!skipData)
await DeleteFileDataAsync(file);
{
var hasOtherReferences = await db.Files
.AnyAsync(f => f.ObjectId == file.ObjectId && f.Id != file.Id);
if (!hasOtherReferences)
await DeleteFileDataAsync(file);
}
}
public async Task DeleteFileDataAsync(SnCloudFile file, bool force = false)
{
if (!file.PoolId.HasValue) return;
if (!file.PoolId.HasValue || file.ObjectId == null) return;
if (!force)
{
var sameOriginFiles = await db.Files
.Where(f => f.StorageId == file.StorageId && f.Id != file.Id)
.Where(f => f.ObjectId == file.ObjectId && f.Id != file.Id)
.Select(f => f.Id)
.ToListAsync();
@@ -499,6 +547,17 @@ public class FileService(
return;
}
var replicas = await db.FileReplicas
.Where(r => r.ObjectId == file.ObjectId)
.ToListAsync();
if (replicas.Count == 0)
{
logger.LogWarning("No replicas found for file object {ObjectId}", file.ObjectId);
return;
}
var primaryReplica = replicas.First(r => r.IsPrimary);
var dest = await GetRemoteStorageConfig(file.PoolId.Value);
if (dest is null) throw new InvalidOperationException($"No remote storage configured for pool {file.PoolId}");
var client = CreateMinioClient(dest);
@@ -508,7 +567,7 @@ public class FileService(
);
var bucket = dest.Bucket;
var objectId = file.StorageId ?? file.Id;
var objectId = primaryReplica.StorageId;
await client.RemoveObjectAsync(
new RemoveObjectArgs().WithBucket(bucket).WithObject(objectId)
@@ -541,36 +600,55 @@ public class FileService(
logger.LogWarning("Failed to delete thumbnail of file {fileId}", file.Id);
}
}
db.FileReplicas.RemoveRange(replicas);
var fileObject = await db.FileObjects.FindAsync(file.ObjectId);
if (fileObject != null) db.FileObjects.Remove(fileObject);
await db.SaveChangesAsync();
}
public async Task DeleteFileDataBatchAsync(List<SnCloudFile> files)
{
files = files.Where(f => f.PoolId.HasValue).ToList();
files = files.Where(f => f.PoolId.HasValue && f.ObjectId != null).ToList();
foreach (var fileGroup in files.GroupBy(f => f.PoolId!.Value))
var objectIds = files.Select(f => f.ObjectId).Distinct().ToList();
var replicas = await db.FileReplicas
.Where(r => objectIds.Contains(r.ObjectId))
.ToListAsync();
foreach (var poolGroup in replicas.GroupBy(r => r.PoolId))
{
var dest = await GetRemoteStorageConfig(fileGroup.Key);
var dest = await GetRemoteStorageConfig(poolGroup.Key);
if (dest is null)
throw new InvalidOperationException($"No remote storage configured for pool {fileGroup.Key}");
throw new InvalidOperationException($"No remote storage configured for pool {poolGroup.Key}");
var client = CreateMinioClient(dest);
if (client is null)
throw new InvalidOperationException(
$"Failed to configure client for remote destination '{fileGroup.Key}'"
$"Failed to configure client for remote destination '{poolGroup.Key}'"
);
List<string> objectsToDelete = [];
foreach (var file in fileGroup)
foreach (var replica in poolGroup)
{
objectsToDelete.Add(file.StorageId ?? file.Id);
if (file.HasCompression) objectsToDelete.Add(file.StorageId ?? file.Id + ".compressed");
if (file.HasThumbnail) objectsToDelete.Add(file.StorageId ?? file.Id + ".thumbnail");
var file = files.First(f => f.ObjectId == replica.ObjectId);
objectsToDelete.Add(replica.StorageId);
if (file.HasCompression) objectsToDelete.Add(replica.StorageId + ".compressed");
if (file.HasThumbnail) objectsToDelete.Add(replica.StorageId + ".thumbnail");
}
await client.RemoveObjectsAsync(
new RemoveObjectsArgs().WithBucket(dest.Bucket).WithObjects(objectsToDelete)
);
db.FileReplicas.RemoveRange(poolGroup);
}
var fileObjects = await db.FileObjects
.Where(fo => objectIds.Contains(fo.Id))
.ToListAsync();
db.FileObjects.RemoveRange(fileObjects);
await db.SaveChangesAsync();
}
private async Task<SnFileBundle?> GetBundleAsync(Guid id, Guid accountId)
@@ -654,6 +732,8 @@ public class FileService(
{
var dbFiles = await db.Files
.Where(f => uncachedIds.Contains(f.Id))
.Include(f => f.Object)
.ThenInclude(o => o.FileReplicas)
.ToListAsync();
foreach (var file in dbFiles)
@@ -674,15 +754,21 @@ public class FileService(
public async Task<int> GetReferenceCountAsync(string fileId)
{
return await db.FileReferences
.Where(r => r.FileId == fileId)
var file = await db.Files.FirstOrDefaultAsync(f => f.Id == fileId);
if (file == null || file.ObjectId == null) return 0;
return await db.Files
.Where(f => f.ObjectId == file.ObjectId && f.Id != fileId)
.CountAsync();
}
public async Task<bool> IsReferencedAsync(string fileId)
{
return await db.FileReferences
.Where(r => r.FileId == fileId)
var file = await db.Files.FirstOrDefaultAsync(f => f.Id == fileId);
if (file == null || file.ObjectId == null) return false;
return await db.Files
.Where(f => f.ObjectId == file.ObjectId && f.Id != fileId)
.AnyAsync();
}
@@ -709,8 +795,6 @@ public class FileService(
.Where(f => f.AccountId == accountId && f.IsMarkedRecycle)
.ToListAsync();
var count = files.Count;
var tasks = files.Select(f => DeleteFileDataAsync(f, true));
await Task.WhenAll(tasks);
var fileIds = files.Select(f => f.Id).ToList();
await _PurgeCacheRangeAsync(fileIds);
db.RemoveRange(files);
@@ -724,8 +808,6 @@ public class FileService(
.Where(f => f.AccountId == accountId && fileIds.Contains(f.Id))
.ToListAsync();
var count = files.Count;
var tasks = files.Select(f => DeleteFileDataAsync(f, true));
await Task.WhenAll(tasks);
var fileIdsList = files.Select(f => f.Id).ToList();
await _PurgeCacheRangeAsync(fileIdsList);
db.RemoveRange(files);
@@ -739,8 +821,6 @@ public class FileService(
.Where(f => f.PoolId == poolId && f.IsMarkedRecycle)
.ToListAsync();
var count = files.Count;
var tasks = files.Select(f => DeleteFileDataAsync(f, true));
await Task.WhenAll(tasks);
var fileIds = files.Select(f => f.Id).ToList();
await _PurgeCacheRangeAsync(fileIds);
db.RemoveRange(files);
@@ -754,8 +834,6 @@ public class FileService(
.Where(f => f.IsMarkedRecycle)
.ToListAsync();
var count = files.Count;
var tasks = files.Select(f => DeleteFileDataAsync(f, true));
await Task.WhenAll(tasks);
var fileIds = files.Select(f => f.Id).ToList();
await _PurgeCacheRangeAsync(fileIds);
db.RemoveRange(files);