260 lines
12 KiB
C#
260 lines
12 KiB
C#
using System.Security.Cryptography;
|
|
using ItWorx.KnowledgeHub.Application.Attachments;
|
|
using ItWorx.KnowledgeHub.Application.Attachments.UseCases;
|
|
using ItWorx.KnowledgeHub.Application.Audit;
|
|
using ItWorx.KnowledgeHub.Application.Ingestion;
|
|
using ItWorx.KnowledgeHub.Application.Ingestion.UseCases;
|
|
using ItWorx.KnowledgeHub.Domain.Attachments;
|
|
using ItWorx.KnowledgeHub.Domain.Ingestion;
|
|
|
|
namespace ItWorx.KnowledgeHub.IntegrationTests.Ingestion;
|
|
|
|
/// <summary>
|
|
/// Does not use the PostgreSQL fixture: the repository port deliberately provides a
|
|
/// controllable save-failure seam around the real upload pipeline. It models the
|
|
/// database rollback that happens when the shared DbContext save fails after the
|
|
/// physical blob commit.
|
|
/// </summary>
|
|
public sealed class IngestionProposalSourceCompensationTests
|
|
{
|
|
private static readonly Guid WorkspaceId = Guid.Parse("11111111-1111-1111-1111-111111111111");
|
|
private static readonly Guid ActorId = Guid.Parse("22222222-2222-2222-2222-222222222222");
|
|
|
|
[Fact]
|
|
public async Task Failed_shared_database_save_after_a_new_blob_commit_compensates_every_staged_artifact()
|
|
{
|
|
var proposal = IngestionProposal.Create(WorkspaceId, ActorId, "compensation", "safe manifest");
|
|
var attachments = new TransactionalAttachmentRepository();
|
|
var audits = new TransactionalAuditRecorder();
|
|
var proposals = new FailingSaveIngestionProposalRepository(proposal, attachments, audits);
|
|
var blobs = new MemoryBlobStore();
|
|
var uploads = new UploadAttachmentUseCase(
|
|
pages: null!,
|
|
stagingSessions: null!,
|
|
ingestionProposals: proposals,
|
|
projects: null!,
|
|
attachments,
|
|
blobs,
|
|
new PlainTextSignatureDetector(),
|
|
audits,
|
|
new AttachmentContentLock(),
|
|
TimeProvider.System);
|
|
var useCase = new StageIngestionProposalSourceUseCase(
|
|
proposals,
|
|
attachments,
|
|
blobs,
|
|
uploads,
|
|
audits,
|
|
new AttachmentContentLock(),
|
|
TimeProvider.System);
|
|
var content = "post-blob save failure"u8.ToArray();
|
|
var digest = Convert.ToHexString(SHA256.HashData(content)).ToLowerInvariant();
|
|
|
|
await using var stream = new MemoryStream(content);
|
|
await Assert.ThrowsAsync<InvalidOperationException>(() => useCase.ExecuteAsync(
|
|
new StageIngestionProposalSourceCommand(
|
|
WorkspaceId,
|
|
proposal.Id,
|
|
ActorId,
|
|
"evidence.txt",
|
|
"text/plain",
|
|
stream,
|
|
"compensation-test",
|
|
MaximumSourceCount: 5,
|
|
ExpectedETag: "1",
|
|
AuditActorKind.ServiceAccount,
|
|
AuditSource: "api"),
|
|
CancellationToken.None));
|
|
|
|
Assert.Equal(1, proposals.SaveAttempts);
|
|
Assert.Equal(1, proposals.PendingSourceCountAtFailedSave);
|
|
Assert.Equal(0, proposals.PersistedSourceCount);
|
|
Assert.Empty(attachments.Blobs);
|
|
Assert.Empty(attachments.References);
|
|
Assert.Empty(attachments.Usages);
|
|
Assert.Empty(audits.Events);
|
|
Assert.False(blobs.Exists(digest));
|
|
Assert.Equal([digest], blobs.DeletedHashes);
|
|
}
|
|
|
|
private sealed class FailingSaveIngestionProposalRepository(
|
|
IngestionProposal proposal,
|
|
TransactionalAttachmentRepository attachments,
|
|
TransactionalAuditRecorder audits) : IIngestionProposalRepository
|
|
{
|
|
public int SaveAttempts { get; private set; }
|
|
|
|
public int PendingSourceCountAtFailedSave { get; private set; }
|
|
|
|
// This is the durable snapshot; it changes only after a successful shared
|
|
// save, exactly like a fresh DbContext query after the database rollback.
|
|
public int PersistedSourceCount { get; private set; }
|
|
|
|
public Task<IngestionProposalPage> ListAsync(IngestionProposalQuery query, CancellationToken cancellationToken) =>
|
|
Task.FromResult(new IngestionProposalPage([], null));
|
|
|
|
public Task<IngestionProposal?> FindAsync(Guid workspaceId, Guid proposalId, CancellationToken cancellationToken) =>
|
|
Task.FromResult<IngestionProposal?>(workspaceId == proposal.WorkspaceId && proposalId == proposal.Id ? proposal : null);
|
|
|
|
public Task<IngestionProposal?> FindByIdempotencyKeyAsync(Guid workspaceId, string idempotencyKey, CancellationToken cancellationToken) =>
|
|
Task.FromResult<IngestionProposal?>(null);
|
|
|
|
public Task<string?> GetETagAsync(Guid workspaceId, Guid proposalId, CancellationToken cancellationToken) =>
|
|
Task.FromResult<string?>(workspaceId == proposal.WorkspaceId && proposalId == proposal.Id ? "1" : null);
|
|
|
|
public Task<IngestionProposal?> FindForUpdateAsync(Guid workspaceId, Guid proposalId, string expectedETag, CancellationToken cancellationToken) =>
|
|
Task.FromResult<IngestionProposal?>(workspaceId == proposal.WorkspaceId && proposalId == proposal.Id && expectedETag == "1" ? proposal : null);
|
|
|
|
public void Add(IngestionProposal item) => throw new NotSupportedException();
|
|
|
|
public Task<int> SaveChangesAsync(CancellationToken cancellationToken)
|
|
{
|
|
SaveAttempts += 1;
|
|
PendingSourceCountAtFailedSave = proposal.Sources.Count;
|
|
attachments.RollBack();
|
|
audits.RollBack();
|
|
return Task.FromException<int>(new InvalidOperationException("Forced shared DbContext save failure."));
|
|
}
|
|
}
|
|
|
|
private sealed class TransactionalAttachmentRepository : IAttachmentRepository
|
|
{
|
|
private readonly List<AttachmentBlob> blobs = [];
|
|
private readonly List<AttachmentReference> references = [];
|
|
private readonly List<AttachmentUsage> usages = [];
|
|
|
|
public IReadOnlyList<AttachmentBlob> Blobs => blobs;
|
|
public IReadOnlyList<AttachmentReference> References => references;
|
|
public IReadOnlyList<AttachmentUsage> Usages => usages;
|
|
|
|
public Task<AttachmentBlob?> FindBlobAsync(string sha256, CancellationToken cancellationToken) =>
|
|
Task.FromResult(blobs.SingleOrDefault(blob => blob.Sha256 == sha256));
|
|
|
|
public void AddBlob(AttachmentBlob blob) => blobs.Add(blob);
|
|
|
|
public Task<AttachmentReference?> FindReferenceAsync(Guid referenceId, CancellationToken cancellationToken) =>
|
|
Task.FromResult(references.SingleOrDefault(reference => reference.Id == referenceId));
|
|
|
|
public void AddReference(AttachmentReference reference) => references.Add(reference);
|
|
|
|
public void AddUsage(AttachmentUsage usage) => usages.Add(usage);
|
|
|
|
public Task<IReadOnlyList<AttachmentReference>> GetReferencesForPageAsync(Guid pageId, CancellationToken cancellationToken) =>
|
|
Task.FromResult<IReadOnlyList<AttachmentReference>>([]);
|
|
|
|
public Task<IReadOnlyDictionary<string, AttachmentBlob>> FindBlobsAsync(IReadOnlyCollection<string> sha256Hashes, CancellationToken cancellationToken) =>
|
|
Task.FromResult<IReadOnlyDictionary<string, AttachmentBlob>>(new Dictionary<string, AttachmentBlob>());
|
|
|
|
public Task<AttachmentUsage?> FindUsageForReferenceAsync(Guid attachmentReferenceId, CancellationToken cancellationToken) =>
|
|
Task.FromResult(usages.SingleOrDefault(usage => usage.AttachmentReferenceId == attachmentReferenceId));
|
|
|
|
public Task<IReadOnlyList<string>> FindUnreferencedBlobHashesAsync(IReadOnlyCollection<string> candidateHashes, CancellationToken cancellationToken) =>
|
|
Task.FromResult<IReadOnlyList<string>>(candidateHashes.Where(hash => references.All(reference => reference.BlobSha256 != hash)).ToArray());
|
|
|
|
public Task<int> RemoveBlobRecordsAsync(IReadOnlyCollection<string> sha256Hashes, CancellationToken cancellationToken) =>
|
|
Task.FromResult(0);
|
|
|
|
public Task<IReadOnlyList<AttachmentUsage>> GetUsagesForStagingSessionAsync(Guid stagingSessionId, CancellationToken cancellationToken) =>
|
|
Task.FromResult<IReadOnlyList<AttachmentUsage>>([]);
|
|
|
|
public Task TransferProposalUsagesToPageAsync(Guid ingestionProposalId, Guid pageId, CancellationToken cancellationToken)
|
|
{
|
|
foreach (var usage in usages.Where(usage => usage.IngestionProposalId == ingestionProposalId))
|
|
{
|
|
usage.TransferProposalToPage(pageId);
|
|
}
|
|
return Task.CompletedTask;
|
|
}
|
|
|
|
public Task<StagingAttachmentCleanupResult> RemoveOrphanedStagingAttachmentsAsync(Guid stagingSessionId, CancellationToken cancellationToken) =>
|
|
Task.FromResult(StagingAttachmentCleanupResult.None);
|
|
|
|
public Task SaveChangesAsync(CancellationToken cancellationToken) => Task.CompletedTask;
|
|
|
|
public void RollBack()
|
|
{
|
|
blobs.Clear();
|
|
references.Clear();
|
|
usages.Clear();
|
|
}
|
|
}
|
|
|
|
private sealed class TransactionalAuditRecorder : IAuditEventRecorder
|
|
{
|
|
private readonly List<AuditEventRecord> events = [];
|
|
|
|
public IReadOnlyList<AuditEventRecord> Events => events;
|
|
|
|
public void Add(AuditEventRecord auditEvent) => events.Add(auditEvent);
|
|
|
|
public Task RecordAsync(AuditEventRecord auditEvent, CancellationToken cancellationToken)
|
|
{
|
|
events.Add(auditEvent);
|
|
return Task.CompletedTask;
|
|
}
|
|
|
|
public void RollBack() => events.Clear();
|
|
}
|
|
|
|
private sealed class MemoryBlobStore : IAttachmentBlobStore
|
|
{
|
|
private readonly Dictionary<string, byte[]> blobs = new(StringComparer.Ordinal);
|
|
private readonly Dictionary<string, byte[]> quarantined = new(StringComparer.Ordinal);
|
|
|
|
public IReadOnlyList<string> DeletedHashes => deletedHashes;
|
|
|
|
private readonly List<string> deletedHashes = [];
|
|
|
|
public bool Exists(string sha256) => blobs.ContainsKey(sha256);
|
|
|
|
public async Task<QuarantinedUpload> WriteToQuarantineAsync(Stream content, CancellationToken cancellationToken)
|
|
{
|
|
using var bytes = new MemoryStream();
|
|
await content.CopyToAsync(bytes, cancellationToken);
|
|
var value = bytes.ToArray();
|
|
var sha256 = Convert.ToHexString(SHA256.HashData(value)).ToLowerInvariant();
|
|
var token = Guid.NewGuid().ToString("N");
|
|
quarantined.Add(token, value);
|
|
return new QuarantinedUpload(token, sha256, value.LongLength);
|
|
}
|
|
|
|
public Task<byte[]> ReadQuarantinePrefixAsync(string quarantineToken, int maxBytes, CancellationToken cancellationToken) =>
|
|
Task.FromResult(quarantined[quarantineToken].Take(maxBytes).ToArray());
|
|
|
|
public Task<bool> BlobExistsAsync(string sha256, CancellationToken cancellationToken) => Task.FromResult(Exists(sha256));
|
|
|
|
public Task CommitAsync(string quarantineToken, string sha256, CancellationToken cancellationToken)
|
|
{
|
|
blobs.Add(sha256, quarantined[quarantineToken]);
|
|
quarantined.Remove(quarantineToken);
|
|
return Task.CompletedTask;
|
|
}
|
|
|
|
public Task DiscardQuarantineAsync(string quarantineToken, CancellationToken cancellationToken)
|
|
{
|
|
quarantined.Remove(quarantineToken);
|
|
return Task.CompletedTask;
|
|
}
|
|
|
|
public Task<Stream> OpenReadAsync(string sha256, CancellationToken cancellationToken) =>
|
|
Task.FromResult<Stream>(new MemoryStream(blobs[sha256], writable: false));
|
|
|
|
public Task<bool> DeleteVerifiedAsync(string sha256, CancellationToken cancellationToken)
|
|
{
|
|
var removed = blobs.Remove(sha256);
|
|
if (removed)
|
|
{
|
|
deletedHashes.Add(sha256);
|
|
}
|
|
|
|
return Task.FromResult(removed);
|
|
}
|
|
}
|
|
|
|
private sealed class PlainTextSignatureDetector : IContentSignatureDetector
|
|
{
|
|
public SignatureDetectionResult Detect(ReadOnlySpan<byte> prefix, string? declaredContentType, string? extension) =>
|
|
new("text/plain", AttachmentDeliveryPolicy.DownloadOnly, DeclaredTypeMismatch: false);
|
|
}
|
|
}
|