ScadaBridge/tests/ZB.MOM.WW.ScadaBridge.IntegrationTests/BulkDeployPoisonedContextTests.cs

using Akka.Actor;
using Akka.TestKit.Xunit2;
using Microsoft.AspNetCore.DataProtection;
using Microsoft.Data.Sqlite;
using Microsoft.EntityFrameworkCore;
using Microsoft.EntityFrameworkCore.Diagnostics;
using Microsoft.EntityFrameworkCore.Storage.ValueConversion;
using Microsoft.Extensions.Logging.Abstractions;
using Microsoft.Extensions.Options;
using NSubstitute;
using ZB.MOM.WW.ScadaBridge.Commons.Entities.Deployment;
using ZB.MOM.WW.ScadaBridge.Commons.Entities.Instances;
using ZB.MOM.WW.ScadaBridge.Commons.Entities.Sites;
using ZB.MOM.WW.ScadaBridge.Commons.Entities.Templates;
using ZB.MOM.WW.ScadaBridge.Commons.Interfaces.Services;
using ZB.MOM.WW.ScadaBridge.Commons.Interfaces.Transport;
using ZB.MOM.WW.ScadaBridge.Commons.Messages.Deployment;
using ZB.MOM.WW.ScadaBridge.Commons.Types;
using ZB.MOM.WW.ScadaBridge.Commons.Types.Enums;
using ZB.MOM.WW.ScadaBridge.Commons.Types.Flattening;
using ZB.MOM.WW.ScadaBridge.Commons.Types.Templates;
using ZB.MOM.WW.ScadaBridge.Communication;
using ZB.MOM.WW.ScadaBridge.ConfigurationDatabase;
using ZB.MOM.WW.ScadaBridge.ConfigurationDatabase.Repositories;
using ZB.MOM.WW.ScadaBridge.DeploymentManager;
using ZB.MOM.WW.ScadaBridge.TemplateEngine.Flattening;

namespace ZB.MOM.WW.ScadaBridge.IntegrationTests;

/// <summary>
/// P7-5: a bulk site deploy runs all three phases over ONE scoped
/// <c>DbContext</c>, so a persistent <c>SaveChanges</c> failure on a single
/// instance used to fail every instance that came after it.
///
/// <para>
/// <b>Why this needs a real EF context.</b> The mechanism is not in the service's
/// control flow, it is in EF's: a failed <c>SaveChanges</c> does NOT untrack the
/// entity that caused it. The entity stays in the change set and is re-attempted
/// by the very next save made through the same context — including the cleanup
/// save inside <c>MarkDeploymentFailedAsync</c>, whose whole job is to stop a
/// record being stranded <c>InProgress</c>. A mocked repository could not prove
/// that; a real <see cref="ScadaBridgeDbContext"/> over SQLite does.
/// </para>
///
/// <para>
/// <b>Honesty note on the injected failure.</b> The persistent failure itself is
/// injected by a save interceptor rather than provoked by a real constraint
/// violation, because the entity that must fail (the <see cref="Instance"/> row
/// the deploy path mutates) is created and mutated inside the service, where a
/// test cannot reach in and corrupt it mid-batch. The interceptor throws
/// <see cref="DbUpdateException"/> at exactly the point a constraint violation,
/// a concurrency conflict or a dropped connection would — everything after that,
/// including the cascade and the fix, is real EF behaviour.
/// </para>
/// </summary>
public sealed class BulkDeployPoisonedContextTests : TestKit, IDisposable
{
    private const int InstanceCount = 4;

    private readonly SqliteConnection _connection;
    private readonly ScadaBridgeDbContext _context;
    private readonly PoisonedEntityInterceptor _poison = new();
    private int _siteId;

    public BulkDeployPoisonedContextTests()
    {
        _connection = new SqliteConnection("DataSource=:memory:");
        _connection.Open();

        var options = new DbContextOptionsBuilder<ScadaBridgeDbContext>()
            .UseSqlite(_connection)
            .ConfigureWarnings(w => w.Ignore(RelationalEventId.PendingModelChangesWarning))
            .AddInterceptors(_poison)
            .Options;

        _context = new SqliteContext(options);
        _context.Database.EnsureCreated();
        Seed();
    }

    /// <summary>
    /// The whole finding in one run: instance 1's post-success persistence fails
    /// permanently (best-effort, so the deploy still reports Success), which leaves
    /// its <see cref="Instance"/> row poisoning the shared change tracker. Every
    /// later instance's status write then rides on that same change set.
    ///
    /// <para>
    /// Pre-fix this cascaded: instances 2, 3 and 4 all failed, and their records
    /// stayed <c>InProgress</c> because the cleanup save re-attempted — and
    /// re-failed on — the same poisoned entity. Post-fix the cleanup detaches the
    /// pending work, so instance 2 is the only casualty (correctly recorded
    /// <c>Failed</c>, not stranded), and 3 and 4 save cleanly.
    /// </para>
    /// </summary>
    [Fact]
    public async Task BulkDeploy_PersistentSaveFailureOnOneInstance_DoesNotCascadeToTheRest()
    {
        _poison.PoisonInstanceNamed("Inst-01");

        var service = CreateService();

        var result = await service.DeploySiteAsync(_siteId, "admin");

        Assert.True(result.IsSuccess);
        var summary = result.Value;
        Assert.Equal(InstanceCount, summary.InstanceResults.Count);

        // Instances 3 and 4 are downstream of the poison and must be unaffected:
        // this is the assertion that fails outright without the detach.
        foreach (var name in new[] { "Inst-03", "Inst-04" })
        {
            var row = Assert.Single(summary.InstanceResults, r => r.UniqueName == name);
            Assert.True(row.Success, $"{name} failed with: {row.ErrorMessage}");
        }

        // Instance 2 is the one collateral casualty — its own status write shared
        // the poisoned change set — but it must be recorded Failed, never left
        // InProgress, which is what the retry on the cleaned tracker buys.
        var stranded = await NoLongerInProgressAsync();
        Assert.Empty(stranded);

        // ...and the fix is not "swallow everything": the poisoned instance's own
        // side effects genuinely did not persist, so its Instance row is still
        // NotDeployed even though the site confirmed the apply. That divergence is
        // the pre-existing best-effort contract, logged for operator reconciliation.
        await using var verify = NewContext();
        Assert.Equal(
            InstanceState.NotDeployed,
            verify.Instances.Single(i => i.UniqueName == "Inst-01").State);
    }

    /// <summary>
    /// The narrow EF fact the fix rests on, pinned on its own so a future reader
    /// does not have to re-derive it: a failed save leaves its entity tracked and
    /// the NEXT save re-attempts it, and
    /// <c>IDeploymentManagerRepository.DetachPendingChanges</c> is what clears it.
    /// </summary>
    [Fact]
    public async Task FailedSave_KeepsTheEntityTracked_UntilPendingChangesAreDetached()
    {
        var repository = new DeploymentManagerRepository(_context);
        var instance = _context.Instances.Single(i => i.UniqueName == "Inst-01");

        _poison.PoisonInstanceNamed("Inst-01");
        instance.State = InstanceState.Enabled;
        await repository.UpdateInstanceAsync(instance);
        await Assert.ThrowsAsync<DbUpdateException>(() => repository.SaveChangesAsync());

        // An unrelated, perfectly valid write now fails too — the poisoned entity
        // is still in the change set.
        var record = new DeploymentRecord("dep-unrelated", "admin")
        {
            InstanceId = _context.Instances.Single(i => i.UniqueName == "Inst-02").Id,
            Status = DeploymentStatus.Failed,
            RevisionHash = "sha256:x",
            DeployedAt = DateTimeOffset.UtcNow,
        };
        await repository.AddDeploymentRecordAsync(record);
        await Assert.ThrowsAsync<DbUpdateException>(() => repository.SaveChangesAsync());

        // Detaching the pending work un-poisons the unit of work; the same write
        // then succeeds.
        Assert.True(repository.DetachPendingChanges() > 0);
        await repository.AddDeploymentRecordAsync(record);
        Assert.Equal(1, await repository.SaveChangesAsync());
    }

    /// <summary>Deployment records left non-terminal after the batch — the stranded-InProgress check.</summary>
    private async Task<List<string>> NoLongerInProgressAsync()
    {
        await using var verify = NewContext();
        return await verify.DeploymentRecords
            .Where(r => r.Status == DeploymentStatus.InProgress || r.Status == DeploymentStatus.Pending)
            .Select(r => r.DeploymentId)
            .ToListAsync();
    }

    private DeploymentService CreateService()
    {
        var repository = new DeploymentManagerRepository(_context);
        var siteRepository = new SiteRepository(_context);

        var pipeline = Substitute.For<IFlatteningPipeline>();
        pipeline.CreateSession().Returns(_ => new FlattenSession(new TemplateGraphWatermark()));
        pipeline
            .FlattenAndValidateAsync(
                Arg.Any<int>(), Arg.Any<CancellationToken>(), Arg.Any<bool>(), Arg.Any<FlattenSession?>())
            .Returns(ci => Result<FlatteningPipelineResult>.Success(
                new FlatteningPipelineResult(
                    new FlattenedConfiguration { InstanceUniqueName = $"Inst-{ci.ArgAt<int>(0):00}" },
                    $"sha256:{ci.ArgAt<int>(0)}",
                    ValidationResult.Success())));

        var comms = new CommunicationService(
            Options.Create(new CommunicationOptions { DeploymentTimeout = TimeSpan.FromSeconds(10) }),
            NullLogger<CommunicationService>.Instance);
        comms.SetCommunicationActor(Sys.ActorOf(Props.Create(() => new AcceptingSiteActor())));

        // Null probe result = "staleness indeterminate", so the up-to-date
        // pre-filter never skips: this test is about the persistence cascade and
        // needs every instance to actually run.
        var probe = Substitute.For<IStaleInstanceProbe>();
        probe.GetCurrentRevisionHashAsync(Arg.Any<int>(), Arg.Any<CancellationToken>()).Returns((string?)null);

        return new DeploymentService(
            repository, siteRepository, pipeline, comms, new OperationLockManager(),
            Substitute.For<IAuditService>(),
            new DiffService(),
            new RevisionHashService(),
            new DeploymentStatusNotifier(NullLogger<DeploymentStatusNotifier>.Instance),
            probe,
            Options.Create(new DeploymentManagerOptions
            {
                OperationLockTimeout = TimeSpan.FromSeconds(5),
                SiteDeploymentMaxParallelism = 1,
                SiteDeploymentTimeoutPerInstance = TimeSpan.FromSeconds(10),
                SiteDeploymentReconcileTimeout = TimeSpan.FromSeconds(2),
            }),
            Options.Create(new CommunicationOptions
            {
                CentralFetchBaseUrl = "https://central.test:9000",
                PendingDeploymentTtl = TimeSpan.FromMinutes(5),
            }),
            NullLogger<DeploymentService>.Instance);
    }

    private void Seed()
    {
        using var ctx = NewContext();
        var site = new Site("Poison Site", "site-poison");
        var template = new Template("T-Poison");
        ctx.Sites.Add(site);
        ctx.Templates.Add(template);
        ctx.SaveChanges();
        _siteId = site.Id;

        for (var i = 1; i <= InstanceCount; i++)
        {
            ctx.Instances.Add(new Instance($"Inst-{i:00}")
            {
                SiteId = site.Id,
                TemplateId = template.Id,
                State = InstanceState.NotDeployed,
            });
        }

        ctx.SaveChanges();
    }

    private ScadaBridgeDbContext NewContext() =>
        new SqliteContext(new DbContextOptionsBuilder<ScadaBridgeDbContext>()
            .UseSqlite(_connection)
            .ConfigureWarnings(w => w.Ignore(RelationalEventId.PendingModelChangesWarning))
            .Options);

    public new void Dispose()
    {
        _context.Dispose();
        _connection.Dispose();
        base.Dispose();
    }

    /// <summary>
    /// Fails every <c>SaveChanges</c> whose change set contains the named
    /// <see cref="Instance"/> — a stand-in for a permanent constraint violation on
    /// one tracked entity, raised at the point EF raises a real one.
    /// </summary>
    private sealed class PoisonedEntityInterceptor : SaveChangesInterceptor
    {
        private string? _poisonedUniqueName;

        public void PoisonInstanceNamed(string uniqueName) => _poisonedUniqueName = uniqueName;

        public override InterceptionResult<int> SavingChanges(
            DbContextEventData eventData, InterceptionResult<int> result)
        {
            ThrowIfPoisoned(eventData);
            return base.SavingChanges(eventData, result);
        }

        public override ValueTask<InterceptionResult<int>> SavingChangesAsync(
            DbContextEventData eventData,
            InterceptionResult<int> result,
            CancellationToken cancellationToken = default)
        {
            ThrowIfPoisoned(eventData);
            return base.SavingChangesAsync(eventData, result, cancellationToken);
        }

        private void ThrowIfPoisoned(DbContextEventData eventData)
        {
            if (_poisonedUniqueName is null || eventData.Context is null)
                return;

            var poisoned = eventData.Context.ChangeTracker.Entries<Instance>()
                .Any(e => e.State != EntityState.Unchanged
                          && e.State != EntityState.Detached
                          && e.Entity.UniqueName == _poisonedUniqueName);

            if (poisoned)
                throw new DbUpdateException($"simulated persistent failure on {_poisonedUniqueName}");
        }
    }

    /// <summary>Answers every deployment with Success, so only persistence can fail here.</summary>
    private sealed class AcceptingSiteActor : ReceiveActor
    {
        public AcceptingSiteActor()
        {
            Receive<SiteEnvelope>(env =>
            {
                if (env.Message is not RefreshDeploymentCommand cmd)
                    return;

                Sender.Tell(new DeploymentStatusResponse(
                    cmd.DeploymentId, cmd.InstanceUniqueName,
                    DeploymentStatus.Success, null, DateTimeOffset.UtcNow));
            });
        }
    }

    /// <summary>
    /// SQLite-adapted context: SQLite has no rowversion and no native
    /// DateTimeOffset. Mirrors the adaptation <c>DeployNotifyFetchTests</c> uses.
    /// </summary>
    private sealed class SqliteContext : ScadaBridgeDbContext
    {
        public SqliteContext(DbContextOptions<ScadaBridgeDbContext> options)
            : base(options, new EphemeralDataProtectionProvider())
        {
        }

        protected override void OnModelCreating(ModelBuilder modelBuilder)
        {
            base.OnModelCreating(modelBuilder);

            modelBuilder.Entity<DeploymentRecord>(builder =>
            {
                builder.Property(d => d.RowVersion)
                    .IsRequired(false)
                    .IsConcurrencyToken(false)
                    .ValueGeneratedNever();
            });

            var converter = new ValueConverter<DateTimeOffset, string>(
                v => v.UtcDateTime.ToString("o"),
                v => DateTimeOffset.Parse(v));
            var nullableConverter = new ValueConverter<DateTimeOffset?, string?>(
                v => v.HasValue ? v.Value.UtcDateTime.ToString("o") : null,
                v => v != null ? DateTimeOffset.Parse(v) : null);

            foreach (var entityType in modelBuilder.Model.GetEntityTypes())
            {
                foreach (var property in entityType.GetProperties())
                {
                    if (property.ClrType == typeof(DateTimeOffset))
                    {
                        property.SetValueConverter(converter);
                        property.SetColumnType("TEXT");
                    }
                    else if (property.ClrType == typeof(DateTimeOffset?))
                    {
                        property.SetValueConverter(nullableConverter);
                        property.SetColumnType("TEXT");
                    }
                }
            }
        }
    }
}