feat(configdb): ISiteCallAuditRepository + EF impl, monotonic upsert (#22, #23 M3)

Bundle B3 of Audit Log #23 M3: data-access layer for the central SiteCalls table introduced in B1+B2. UpsertAsync is insert-if-not-exists then monotonic-status update so out-of-order telemetry, duplicate gRPC packets, and reconciliation pulls all converge on the same row without rolling state backward. - src/ScadaLink.Commons/Interfaces/Repositories/ISiteCallAuditRepository.cs: UpsertAsync (monotonic), GetAsync, QueryAsync, PurgeTerminalAsync. - src/ScadaLink.Commons/Types/Audit/SiteCallQueryFilter.cs + SiteCallPaging.cs: filter (Channel/SourceSite/Status/Target/time range) and keyset paging cursor on (CreatedAtUtc DESC, TrackedOperationId DESC), mirrored on M1's AuditLog* equivalents. - src/ScadaLink.ConfigurationDatabase/Repositories/SiteCallAuditRepository.cs: raw-SQL InsertIfNotExists + conditional UPDATE with inline CASE rank compare (Submitted=0, Forwarded=1, Attempted/Skipped=2, terminal=3 — terminal statuses are mutually exclusive so e.g. Delivered cannot overwrite Parked). Duplicate-key violations (SQL 2601/2627) are swallowed at Debug, identical to AuditLogRepository's race-fix. QueryAsync uses FromSqlInterpolated because EF Core 10 cannot translate string.Compare against the value-converted TrackedOperationId column inside an expression tree. - ServiceCollectionExtensions wires the repository (scoped, after IAuditLogRepository). - 12 integration tests in tests/ScadaLink.ConfigurationDatabase.Tests/ Repositories/ (MsSqlMigrationFixture + [SkippableFact]): fresh insert, monotonic advance, older-status no-op, same-status no-op, terminal-over-terminal no-op, 50-way concurrent-insert race produces exactly one row, Get known/unknown, filter by site, keyset paging no overlap, purge terminal-and-old, purge keeps non-terminal-and-recent.
2026-05-20 14:10:24 -04:00
parent 6667f345fa
commit bedfa6b8f3
6 changed files with 705 additions and 0 deletions
--- a/src/ScadaLink.ConfigurationDatabase/Repositories/SiteCallAuditRepository.cs
+++ b/src/ScadaLink.ConfigurationDatabase/Repositories/SiteCallAuditRepository.cs
@@ -0,0 +1,214 @@
+using Microsoft.Data.SqlClient;
+using Microsoft.EntityFrameworkCore;
+using Microsoft.Extensions.Logging;
+using Microsoft.Extensions.Logging.Abstractions;
+using ScadaLink.Commons.Entities.Audit;
+using ScadaLink.Commons.Interfaces.Repositories;
+using ScadaLink.Commons.Types;
+using ScadaLink.Commons.Types.Audit;
+
+namespace ScadaLink.ConfigurationDatabase.Repositories;
+
+/// <summary>
+/// EF Core implementation of <see cref="ISiteCallAuditRepository"/>. See the
+/// interface for the monotonic-upsert contract; this class adds notes on the
+/// data-access strategy used by each method.
+/// </summary>
+public class SiteCallAuditRepository : ISiteCallAuditRepository
+{
+    // SQL Server duplicate-key error numbers, identical to the AuditLogRepository
+    // race-fix: 2601 = unique-index violation, 2627 = PK/unique-constraint
+    // violation. The IF NOT EXISTS … INSERT pattern has a check-then-act window
+    // and the loser surfaces as one of these; monotonic-upsert semantics demand
+    // we swallow them.
+    private const int SqlErrorUniqueIndexViolation = 2601;
+    private const int SqlErrorPrimaryKeyViolation = 2627;
+
+    // Monotonic status ordering. Lower rank wins on tie (same-rank upserts are
+    // no-ops, including terminal-over-terminal). Spec from Bundle B3 plan:
+    // Submitted < Forwarded < Attempted == Skipped < Delivered == Failed == Parked == Discarded.
+    private static readonly Dictionary<string, int> StatusRank = new(StringComparer.Ordinal)
+    {
+        ["Submitted"] = 0,
+        ["Forwarded"] = 1,
+        ["Attempted"] = 2,
+        ["Skipped"] = 2,
+        ["Delivered"] = 3,
+        ["Failed"] = 3,
+        ["Parked"] = 3,
+        ["Discarded"] = 3,
+    };
+
+    private readonly ScadaLinkDbContext _context;
+    private readonly ILogger<SiteCallAuditRepository> _logger;
+
+    public SiteCallAuditRepository(ScadaLinkDbContext context, ILogger<SiteCallAuditRepository>? logger = null)
+    {
+        _context = context ?? throw new ArgumentNullException(nameof(context));
+        _logger = logger ?? NullLogger<SiteCallAuditRepository>.Instance;
+    }
+
+    /// <summary>
+    /// Two-step: <c>IF NOT EXISTS INSERT</c> then conditional <c>UPDATE</c> with
+    /// an inline <c>CASE</c> rank comparison. Both go through
+    /// <see cref="Microsoft.EntityFrameworkCore.RelationalDatabaseFacadeExtensions.ExecuteSqlInterpolatedAsync"/>
+    /// so the change tracker is bypassed and the value-converted PK column is
+    /// written as the canonical "D"-format GUID string. Duplicate-key violations
+    /// from the insert race are swallowed.
+    /// </summary>
+    public async Task UpsertAsync(SiteCall siteCall, CancellationToken ct = default)
+    {
+        if (siteCall is null)
+        {
+            throw new ArgumentNullException(nameof(siteCall));
+        }
+
+        var idText = siteCall.TrackedOperationId.Value.ToString("D");
+        var incomingRank = GetRankOrThrow(siteCall.Status);
+
+        // Step 1: insert-if-not-exists. Like AuditLogRepository.InsertIfNotExistsAsync
+        // this is check-then-act so a duplicate-key violation may surface under
+        // concurrent inserts on the same id — caught + logged at Debug.
+        try
+        {
+            await _context.Database.ExecuteSqlInterpolatedAsync(
+                $@"IF NOT EXISTS (SELECT 1 FROM dbo.SiteCalls WHERE TrackedOperationId = {idText})
+INSERT INTO dbo.SiteCalls
+    (TrackedOperationId, Channel, Target, SourceSite, Status, RetryCount,
+     LastError, HttpStatus, CreatedAtUtc, UpdatedAtUtc, TerminalAtUtc, IngestedAtUtc)
+VALUES
+    ({idText}, {siteCall.Channel}, {siteCall.Target}, {siteCall.SourceSite}, {siteCall.Status}, {siteCall.RetryCount},
+     {siteCall.LastError}, {siteCall.HttpStatus}, {siteCall.CreatedAtUtc}, {siteCall.UpdatedAtUtc}, {siteCall.TerminalAtUtc}, {siteCall.IngestedAtUtc});",
+                ct);
+        }
+        catch (SqlException ex) when (
+            ex.Number == SqlErrorUniqueIndexViolation
+            || ex.Number == SqlErrorPrimaryKeyViolation)
+        {
+            _logger.LogDebug(
+                ex,
+                "SiteCallAuditRepository.UpsertAsync swallowed duplicate-key violation (error {SqlErrorNumber}) for TrackedOperationId {TrackedOperationId}; falling through to monotonic update.",
+                ex.Number,
+                idText);
+        }
+
+        // Step 2: monotonic update. The CASE expression maps the stored Status
+        // string to the same rank table the caller uses; we only mutate if the
+        // incoming rank is strictly greater. Same-rank (including
+        // terminal-over-terminal) is a no-op — first-write-wins at each rank.
+        await _context.Database.ExecuteSqlInterpolatedAsync(
+            $@"UPDATE dbo.SiteCalls
+SET Status = {siteCall.Status},
+    RetryCount = {siteCall.RetryCount},
+    LastError = {siteCall.LastError},
+    HttpStatus = {siteCall.HttpStatus},
+    UpdatedAtUtc = {siteCall.UpdatedAtUtc},
+    TerminalAtUtc = {siteCall.TerminalAtUtc},
+    IngestedAtUtc = {siteCall.IngestedAtUtc}
+WHERE TrackedOperationId = {idText}
+  AND {incomingRank} > (CASE Status
+      WHEN 'Submitted' THEN 0
+      WHEN 'Forwarded' THEN 1
+      WHEN 'Attempted' THEN 2
+      WHEN 'Skipped'   THEN 2
+      WHEN 'Delivered' THEN 3
+      WHEN 'Failed'    THEN 3
+      WHEN 'Parked'    THEN 3
+      WHEN 'Discarded' THEN 3
+      ELSE -1
+  END);",
+            ct);
+    }
+
+    /// <summary>
+    /// Single <c>FindAsync</c> against the PK. Returns <c>null</c> for unknown ids.
+    /// </summary>
+    public async Task<SiteCall?> GetAsync(TrackedOperationId id, CancellationToken ct = default)
+    {
+        return await _context.Set<SiteCall>().FindAsync(new object?[] { id }, ct);
+    }
+
+    /// <summary>
+    /// Builds a parameterised SQL query against <c>dbo.SiteCalls</c> ordered by
+    /// <c>(CreatedAtUtc DESC, TrackedOperationId DESC)</c>, with keyset paging.
+    /// Raw SQL is used here (rather than LINQ) because EF Core 10 cannot
+    /// translate the lexicographic string comparison against the value-converted
+    /// <see cref="TrackedOperationId"/> column inside an expression tree — the
+    /// converter is applied to equality but not to inequality comparisons
+    /// against the underlying Guid. The keyset tiebreaker is varchar lex order,
+    /// which is deterministic and gives "no overlap, every row exactly once"
+    /// paging without depending on Guid byte ordering.
+    /// </summary>
+    public async Task<IReadOnlyList<SiteCall>> QueryAsync(
+        SiteCallQueryFilter filter, SiteCallPaging paging, CancellationToken ct = default)
+    {
+        if (filter is null)
+        {
+            throw new ArgumentNullException(nameof(filter));
+        }
+
+        if (paging is null)
+        {
+            throw new ArgumentNullException(nameof(paging));
+        }
+
+        // FormattableString interpolation parameterises every value (no concatenation)
+        // so this is injection-safe. EF Core resolves the parameter values, the
+        // composed sql is shaped to SQL Server's grammar and projected into the
+        // SiteCall entity via FromSqlInterpolated. The CASE expressions wrap each
+        // optional predicate so a null filter field degrades to a no-op (matches
+        // every row) instead of branching at C# level into N variants.
+        var afterCreated = paging.AfterCreatedAtUtc;
+        var afterIdString = paging.AfterId?.Value.ToString("D");
+        var hasCursor = afterCreated is not null && afterIdString is not null;
+
+        var fromUtc = filter.FromUtc;
+        var toUtc = filter.ToUtc;
+
+        FormattableString sql = $@"
+SELECT TOP ({paging.PageSize})
+    TrackedOperationId, Channel, Target, SourceSite, Status, RetryCount,
+    LastError, HttpStatus, CreatedAtUtc, UpdatedAtUtc, TerminalAtUtc, IngestedAtUtc
+FROM dbo.SiteCalls
+WHERE ({filter.Channel} IS NULL OR Channel = {filter.Channel})
+  AND ({filter.SourceSite} IS NULL OR SourceSite = {filter.SourceSite})
+  AND ({filter.Status} IS NULL OR Status = {filter.Status})
+  AND ({filter.Target} IS NULL OR Target = {filter.Target})
+  AND ({fromUtc} IS NULL OR CreatedAtUtc >= {fromUtc})
+  AND ({toUtc} IS NULL OR CreatedAtUtc <= {toUtc})
+  AND ({(hasCursor ? 1 : 0)} = 0
+       OR CreatedAtUtc < {afterCreated}
+       OR (CreatedAtUtc = {afterCreated} AND TrackedOperationId < {afterIdString}))
+ORDER BY CreatedAtUtc DESC, TrackedOperationId DESC;";
+
+        var rows = await _context.Set<SiteCall>()
+            .FromSqlInterpolated(sql)
+            .AsNoTracking()
+            .ToListAsync(ct);
+
+        return rows;
+    }
+
+    /// <summary>
+    /// Deletes rows whose <see cref="SiteCall.TerminalAtUtc"/> is non-null AND
+    /// strictly less than <paramref name="olderThanUtc"/>. Non-terminal rows are
+    /// never touched. Returns the number of rows removed.
+    /// </summary>
+    public async Task<int> PurgeTerminalAsync(DateTime olderThanUtc, CancellationToken ct = default)
+    {
+        return await _context.Database.ExecuteSqlInterpolatedAsync(
+            $"DELETE FROM dbo.SiteCalls WHERE TerminalAtUtc IS NOT NULL AND TerminalAtUtc < {olderThanUtc};",
+            ct);
+    }
+
+    private static int GetRankOrThrow(string status)
+    {
+        if (!StatusRank.TryGetValue(status, out var rank))
+        {
+            throw new ArgumentException(
+                $"Unknown SiteCall status '{status}'. Expected one of: {string.Join(", ", StatusRank.Keys)}.",
+                nameof(status));
+        }
+        return rank;
+    }
+}
--- a/src/ScadaLink.ConfigurationDatabase/ServiceCollectionExtensions.cs
+++ b/src/ScadaLink.ConfigurationDatabase/ServiceCollectionExtensions.cs
@@ -47,6 +47,7 @@ public static class ServiceCollectionExtensions
        services.AddScoped<INotificationRepository, NotificationRepository>();
        services.AddScoped<INotificationOutboxRepository, NotificationOutboxRepository>();
        services.AddScoped<IAuditLogRepository, AuditLogRepository>();
+        services.AddScoped<ISiteCallAuditRepository, SiteCallAuditRepository>();
        services.AddScoped<IInboundApiRepository, InboundApiRepository>();
        services.AddScoped<IAuditService, AuditService>();
        services.AddScoped<IInstanceLocator, InstanceLocator>();