itworx-knowledge-hub-Public/tests/backend/ItWorx.KnowledgeHub.IntegrationTests/Ingestion/IngestionProposalPersistenceTests.cs

using ItWorx.KnowledgeHub.Application.Common;
using ItWorx.KnowledgeHub.Application.Ingestion;
using ItWorx.KnowledgeHub.Domain.Ingestion;
using ItWorx.KnowledgeHub.Infrastructure.Ingestion;
using ItWorx.KnowledgeHub.Infrastructure.Persistence;
using ItWorx.KnowledgeHub.IntegrationTests.Infrastructure;
using Microsoft.EntityFrameworkCore;

namespace ItWorx.KnowledgeHub.IntegrationTests.Operations;

[Collection(PostgresTestGroup.Name)]
public sealed class IngestionProposalPersistenceTests(PostgresContainerFixture postgres)
{
    private static readonly Guid WorkspaceId = Guid.Parse("11111111-1111-1111-1111-111111111111");
    private static readonly Guid ActorId = Guid.Parse("22222222-2222-2222-2222-222222222222");
    private const string Manifest = "# Proposed knowledge\n\nSafe manifest metadata.";
    private const string Digest = "c8c86d360bfb7bb0d036b7352bf89714946fb527c4074340752259109371f429";

    [Fact]
    public async Task Proposal_preserves_digest_and_unique_key()
    {
        await using var target = await RestoreTargetDatabase.CreateAsync(postgres);
        var options = new DbContextOptionsBuilder<KnowledgeHubDbContext>()
            .UseNpgsql(target.ConnectionString, npgsql => npgsql.MigrationsAssembly(typeof(KnowledgeHubDbContext).Assembly.GetName().Name))
            .UseSnakeCaseNamingConvention()
            .Options;

        await using (var db = new KnowledgeHubDbContext(options))
        {
            var repository = new IngestionProposalRepository(db);
            var proposal = IngestionProposal.Create(WorkspaceId, ActorId, "correlation-2", Manifest);
            proposal.AddSource("proposal.pdf", "application/pdf", 42, Digest, "clean", Guid.NewGuid());
            proposal.AssignIdempotencyKey("key-2");
            repository.Add(proposal);
            await repository.SaveChangesAsync(CancellationToken.None);
        }

        await using (var db = new KnowledgeHubDbContext(options))
        {
            var repository = new IngestionProposalRepository(db);
            var roundTripped = await repository.FindByIdempotencyKeyAsync(WorkspaceId, "key-2", CancellationToken.None);

            Assert.NotNull(roundTripped);
            var source = Assert.Single(roundTripped.Sources);
            Assert.Equal(Digest, source.DigestSha256);
            Assert.Equal("proposal.pdf", source.FileName);
            Assert.Equal("application/pdf", source.MimeType);
            Assert.Equal(42, source.ByteSize);
            Assert.Equal("clean", source.ScanResult);
            Assert.Single(roundTripped.Revisions);
        }

        await Assert.ThrowsAsync<DbUpdateException>(async () =>
        {
            await using var db = new KnowledgeHubDbContext(options);
            var repository = new IngestionProposalRepository(db);
            var duplicate = IngestionProposal.Create(WorkspaceId, ActorId, "correlation-duplicate", Manifest);
            duplicate.AssignIdempotencyKey("key-2");
            repository.Add(duplicate);
            await repository.SaveChangesAsync(CancellationToken.None);
        });
    }

    [Fact]
    public async Task Repository_scopes_reads_etags_updates_and_bounded_lists_to_workspace()
    {
        await using var target = await RestoreTargetDatabase.CreateAsync(postgres);
        var options = new DbContextOptionsBuilder<KnowledgeHubDbContext>()
            .UseNpgsql(target.ConnectionString, npgsql => npgsql.MigrationsAssembly(typeof(KnowledgeHubDbContext).Assembly.GetName().Name))
            .UseSnakeCaseNamingConvention()
            .Options;
        var otherWorkspaceId = Guid.NewGuid();
        Guid proposalId;

        await using (var db = new KnowledgeHubDbContext(options))
        {
            var repository = new IngestionProposalRepository(db);
            var first = IngestionProposal.Create(WorkspaceId, ActorId, "first", Manifest);
            first.AssignIdempotencyKey("queue-1");
            repository.Add(first);
            var second = IngestionProposal.Create(WorkspaceId, ActorId, "second", Manifest);
            second.AssignIdempotencyKey("queue-2");
            repository.Add(second);
            var other = IngestionProposal.Create(otherWorkspaceId, ActorId, "other", Manifest);
            other.AssignIdempotencyKey("other-1");
            repository.Add(other);
            await repository.SaveChangesAsync(CancellationToken.None);
            proposalId = first.Id;
        }

        await using (var db = new KnowledgeHubDbContext(options))
        {
            var repository = new IngestionProposalRepository(db);
            var page = await repository.ListAsync(new IngestionProposalQuery(WorkspaceId, Limit: 1), CancellationToken.None);

            Assert.Single(page.Items);
            Assert.NotNull(page.NextCursor);
            Assert.Null(await repository.FindAsync(otherWorkspaceId, proposalId, CancellationToken.None));
            Assert.Null(await repository.GetETagAsync(otherWorkspaceId, proposalId, CancellationToken.None));
            Assert.Null(await repository.FindForUpdateAsync(otherWorkspaceId, proposalId, "0", CancellationToken.None));
            Assert.DoesNotContain((await repository.ListAsync(new IngestionProposalQuery(otherWorkspaceId, Limit: 100), CancellationToken.None)).Items,
                proposal => proposal.WorkspaceId == WorkspaceId);
        }
    }

    [Fact]
    public async Task Repository_translates_a_stale_xmin_to_an_application_concurrency_conflict()
    {
        await using var target = await RestoreTargetDatabase.CreateAsync(postgres);
        var options = new DbContextOptionsBuilder<KnowledgeHubDbContext>()
            .UseNpgsql(target.ConnectionString, npgsql => npgsql.MigrationsAssembly(typeof(KnowledgeHubDbContext).Assembly.GetName().Name))
            .UseSnakeCaseNamingConvention()
            .Options;
        Guid proposalId;

        await using (var db = new KnowledgeHubDbContext(options))
        {
            var repository = new IngestionProposalRepository(db);
            var proposal = IngestionProposal.Create(WorkspaceId, ActorId, "stale-xmin", Manifest);
            proposal.AssignIdempotencyKey("stale-xmin-key");
            repository.Add(proposal);
            await repository.SaveChangesAsync(CancellationToken.None);
            proposalId = proposal.Id;
        }

        await using var staleDb = new KnowledgeHubDbContext(options);
        var staleRepository = new IngestionProposalRepository(staleDb);
        var staleEtag = await staleRepository.GetETagAsync(WorkspaceId, proposalId, CancellationToken.None);

        await using (var currentDb = new KnowledgeHubDbContext(options))
        {
            var currentRepository = new IngestionProposalRepository(currentDb);
            var currentEtag = await currentRepository.GetETagAsync(WorkspaceId, proposalId, CancellationToken.None);
            var current = await currentRepository.FindForUpdateAsync(WorkspaceId, proposalId, currentEtag!, CancellationToken.None);
            Assert.NotNull(current);
            current.AddSource("current-source.pdf", "application/pdf", 42, Digest, "clean", Guid.NewGuid());
            current.BeginAnalysis(ActorId, DateTimeOffset.UtcNow);
            await currentRepository.SaveChangesAsync(CancellationToken.None);
        }

        await using (var verificationDb = new KnowledgeHubDbContext(options))
        {
            var verificationRepository = new IngestionProposalRepository(verificationDb);
            var updated = await verificationRepository.FindAsync(WorkspaceId, proposalId, CancellationToken.None);
            Assert.NotNull(updated);
            Assert.Equal(IngestionProposalStatus.Analyzing, updated.Status);
            Assert.Contains(updated.Revisions, revision => revision.Operation == "analysis-started");
            Assert.Contains(updated.Sources, source => source.FileName == "current-source.pdf");
        }

        var stale = await staleRepository.FindForUpdateAsync(WorkspaceId, proposalId, staleEtag!, CancellationToken.None);
        Assert.NotNull(stale);
        stale.ApplyAnalysis("stale analysis", ActorId, DateTimeOffset.UtcNow);

        var conflict = await Assert.ThrowsAsync<ConcurrencyConflictException>(
            () => staleRepository.SaveChangesAsync(CancellationToken.None));
        Assert.Equal(proposalId, conflict.EntityId);
        Assert.Empty(staleDb.ChangeTracker.Entries());
    }
}