Joseph Doherty
655 lines
33 KiB
C#
655 lines
33 KiB
C#
using System.Data;
|
|
using Microsoft.Data.SqlClient;
|
|
using Microsoft.EntityFrameworkCore;
|
|
using Microsoft.Extensions.Logging;
|
|
using Microsoft.Extensions.Logging.Abstractions;
|
|
using ZB.MOM.WW.ScadaBridge.Commons.Entities.Audit;
|
|
using ZB.MOM.WW.ScadaBridge.Commons.Interfaces.Repositories;
|
|
using ZB.MOM.WW.ScadaBridge.Commons.Types;
|
|
using ZB.MOM.WW.ScadaBridge.Commons.Types.Audit;
|
|
|
|
namespace ZB.MOM.WW.ScadaBridge.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:
|
|
// Submitted < Forwarded < Attempted == Skipped < Delivered == Failed == Parked == Discarded.
|
|
// A higher incoming rank always wins. WITHIN an equal NON-terminal rank
|
|
// (Attempted/Skipped, rank 2 — and the transient Submitted/Forwarded ranks),
|
|
// the newest UpdatedAtUtc wins so a retrying call's live RetryCount/LastError
|
|
// no longer freezes at first-write. Equal terminal ranks (rank 3)
|
|
// stay immutable — the freshness tiebreaker is deliberately scoped to
|
|
// rank < 3, so a later terminal NEVER flips an earlier one (Delivered cannot
|
|
// overwrite Parked). Still idempotent (equal stamps are inert) and still
|
|
// regression-proof (a lower rank is always a no-op).
|
|
private const int TerminalRank = 3;
|
|
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 ScadaBridgeDbContext _context;
|
|
private readonly ILogger<SiteCallAuditRepository> _logger;
|
|
|
|
/// <summary>
|
|
/// Initializes a new instance of the <see cref="SiteCallAuditRepository"/> class.
|
|
/// </summary>
|
|
/// <param name="context">The EF Core database context.</param>
|
|
/// <param name="logger">Optional logger for diagnostic information.</param>
|
|
public SiteCallAuditRepository(ScadaBridgeDbContext context, ILogger<SiteCallAuditRepository>? logger = null)
|
|
{
|
|
_context = context ?? throw new ArgumentNullException(nameof(context));
|
|
_logger = logger ?? NullLogger<SiteCallAuditRepository>.Instance;
|
|
}
|
|
|
|
/// <inheritdoc />
|
|
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);
|
|
|
|
// ONE round trip, INSERT-first. Both statements ship in a single command
|
|
// text, so the round-trip saving of the single-command rewrite is preserved — but
|
|
// the ORDER is back to insert-then-update, because UPDATE-first LOSES
|
|
// DATA under a concurrent first-write.
|
|
//
|
|
// The UPDATE-first shape was: UPDATE; SET @updated = @@ROWCOUNT;
|
|
// IF @updated = 0 AND NOT EXISTS(…) INSERT. Two writers racing the FIRST
|
|
// packet of one TrackedOperationId — the cached dual-write and the
|
|
// reconciliation pull, which routinely carry DIFFERENT lifecycle states —
|
|
// both find no row, so both UPDATEs match nothing. The loser then either
|
|
// fails its own NOT EXISTS re-check (READ COMMITTED, after the winner
|
|
// committed) and skips the INSERT, or attempts it and eats a 2627 in the
|
|
// catch below. EITHER WAY the loser's Status/RetryCount/HttpStatus/
|
|
// TerminalAtUtc are silently dropped: it never ran an UPDATE against the
|
|
// winner's row.
|
|
//
|
|
// INSERT-first has no such hole. The loser's INSERT is skipped or faults,
|
|
// and the monotonic UPDATE that FOLLOWS it applies its state to whichever
|
|
// row won — so the newer lifecycle state survives regardless of
|
|
// interleaving, and a stale one is still rejected by the rank guard.
|
|
//
|
|
// Monotonic update semantics are unchanged: mutate when EITHER the
|
|
// incoming rank is strictly greater, OR the incoming rank equals the
|
|
// stored rank AND that rank is non-terminal (< TerminalRank) AND the
|
|
// incoming UpdatedAtUtc is strictly newer than the stored one — so a
|
|
// retrying call's Attempted-phase RetryCount/LastError/HttpStatus stay
|
|
// live instead of freezing at the first Attempted packet. Terminal ranks
|
|
// are excluded from the tiebreaker, so a later terminal NEVER overwrites
|
|
// an earlier one; equal stamps are inert (idempotent replay) and a lower
|
|
// rank is always a no-op. That inertness is what makes the UPDATE
|
|
// harmless immediately after this same call's own INSERT: equal rank,
|
|
// equal UpdatedAtUtc, zero rows changed.
|
|
//
|
|
// Raw SQL with explicitly-typed parameters (rather than
|
|
// ExecuteSqlInterpolated) so the monotonic UPDATE exists as ONE statement
|
|
// text shared by the combined batch and the duplicate-key retry below —
|
|
// the predicate is far too intricate to keep in two copies. Explicit
|
|
// SqlDbType is load-bearing: an untyped DateTime parameter binds as
|
|
// `datetime` (3.33 ms rounding), which would corrupt both the stored
|
|
// datetime2 stamps and the `UpdatedAtUtc <` freshness tiebreaker. Sizes
|
|
// are deliberately left at the value's own length so the server enforces
|
|
// the column widths (see AuditLogRepository.AddParameter).
|
|
//
|
|
// SourceNode-stamping: the column is included in the INSERT column list /
|
|
// VALUES so a fresh row carries the originating node name (node-a/node-b
|
|
// for site rows). A null SourceNode (legacy hosts / unstamped reconciled
|
|
// rows) writes NULL straight through. On the UPDATE leg SourceNode is
|
|
// written via
|
|
// COALESCE(@SourceNode, SourceNode). The operator returns @SourceNode
|
|
// when it is non-null, otherwise the stored value — so the column
|
|
// behaves protectively: a later packet that carries a null
|
|
// SourceNode (e.g. a reconciliation pull from an unstamped node)
|
|
// NEVER blanks out a value the first stamping packet set. A later
|
|
// packet that DOES carry a non-null SourceNode replaces the previous
|
|
// value — combined with the monotonic-rank guard this is
|
|
// "last-non-null-wins on rank advance", which lets a missing
|
|
// SourceNode be filled in later if Submit happened to be unstamped
|
|
// and an Attempt/Resolve carries the node identity. Within one
|
|
// lifecycle every packet should carry the same SourceNode value (one
|
|
// execution, one node) so the "overwrite" path is in practice
|
|
// idempotent.
|
|
try
|
|
{
|
|
await _context.Database.ExecuteSqlRawAsync(
|
|
InsertIfAbsentSql + "\n\n" + MonotonicUpdateSql,
|
|
BuildUpsertParameters(siteCall, idText, incomingRank),
|
|
ct)
|
|
.ConfigureAwait(false);
|
|
}
|
|
catch (SqlException ex) when (
|
|
ex.Number == SqlErrorUniqueIndexViolation
|
|
|| ex.Number == SqlErrorPrimaryKeyViolation)
|
|
{
|
|
// Two concurrent sessions both found the row absent and both raced to
|
|
// INSERT; the loser raises 2601/2627 against the TrackedOperationId
|
|
// primary key. The winner's row IS the first-write, but it is NOT
|
|
// necessarily this packet's lifecycle state — the reconciliation pull
|
|
// and the cached dual-write feed this method with different states —
|
|
// so the loser MUST still apply its monotonic UPDATE against the
|
|
// winner's row. The batch aborted at the faulting INSERT, so re-run
|
|
// the UPDATE alone here; it is idempotent and rank-guarded, so
|
|
// re-running it is safe even if the batch did reach it.
|
|
_logger.LogDebug(
|
|
ex,
|
|
"SiteCallAuditRepository.UpsertAsync swallowed duplicate-key violation (error {SqlErrorNumber}) for TrackedOperationId {TrackedOperationId}; re-running the monotonic update against the winning row.",
|
|
ex.Number,
|
|
idText);
|
|
|
|
await _context.Database.ExecuteSqlRawAsync(
|
|
MonotonicUpdateSql,
|
|
BuildUpsertParameters(siteCall, idText, incomingRank),
|
|
ct)
|
|
.ConfigureAwait(false);
|
|
}
|
|
}
|
|
|
|
/// <summary>
|
|
/// Rows per batched upsert command. Each row binds 15 parameters, so 128 rows stay
|
|
/// under SQL Server's 2 100-parameter limit (1 920) with headroom.
|
|
/// </summary>
|
|
internal const int UpsertManyChunkSize = 128;
|
|
|
|
/// <inheritdoc />
|
|
public async Task UpsertManyAsync(IReadOnlyList<SiteCall> siteCalls, CancellationToken ct = default)
|
|
{
|
|
ArgumentNullException.ThrowIfNull(siteCalls);
|
|
|
|
// The reconciliation pull's write path (review 2026-10-21 06/PERF-2): one command
|
|
// per chunk of UpsertManyChunkSize rows instead of one round trip per row — a
|
|
// 500-row page is 4 round trips, not 500. Each row runs the SAME two statements
|
|
// UpsertAsync runs (InsertIfAbsentSql then MonotonicUpdateSql), with its parameters
|
|
// suffixed per row, so the insert-first / monotonic / ownership semantics are
|
|
// byte-identical rather than re-derived set-based.
|
|
//
|
|
// Any SqlException from a chunk — an insert race's 2601/2627 (statement-level,
|
|
// the batch runs on) or an ownership THROW 51001 (batch-terminating) — re-runs that
|
|
// chunk row by row through UpsertAsync. Re-applying rows the batch already applied
|
|
// is inert (equal rank + equal UpdatedAtUtc changes nothing) and UpsertAsync handles
|
|
// the race exactly as before. A genuinely bad row does not stop the rest of the
|
|
// page: the remaining rows are still applied and the FIRST such failure is
|
|
// rethrown at the end, so the caller sees the same exception the old per-row
|
|
// loop surfaced (and keeps its cursor), minus the collateral of the rows behind it.
|
|
System.Runtime.ExceptionServices.ExceptionDispatchInfo? firstFailure = null;
|
|
// Rows UpsertAsync would reject before touching the database (a null row, an
|
|
// unknown status) are validated up front and SKIPPED with their failure captured,
|
|
// so one malformed row cannot stop the rest of the page from applying.
|
|
var valid = new List<(SiteCall Row, int Rank)>(siteCalls.Count);
|
|
foreach (var siteCall in siteCalls)
|
|
{
|
|
try
|
|
{
|
|
if (siteCall is null)
|
|
{
|
|
throw new ArgumentException("A null SiteCall row was supplied.", nameof(siteCalls));
|
|
}
|
|
|
|
valid.Add((siteCall, GetRankOrThrow(siteCall.Status)));
|
|
}
|
|
catch (ArgumentException invalid)
|
|
{
|
|
firstFailure ??= System.Runtime.ExceptionServices.ExceptionDispatchInfo.Capture(invalid);
|
|
}
|
|
}
|
|
|
|
for (var offset = 0; offset < valid.Count; offset += UpsertManyChunkSize)
|
|
{
|
|
var chunk = valid.Skip(offset).Take(UpsertManyChunkSize).Select(v => v.Row).ToList();
|
|
var ranks = valid.Skip(offset).Take(UpsertManyChunkSize).Select(v => v.Rank).ToList();
|
|
var sql = new System.Text.StringBuilder();
|
|
var parameters = new List<object>(chunk.Count * 15);
|
|
for (var i = 0; i < chunk.Count; i++)
|
|
{
|
|
var siteCall = chunk[i];
|
|
var suffix = "_" + i.ToString(System.Globalization.CultureInfo.InvariantCulture);
|
|
sql.AppendLine(SuffixParameters(InsertIfAbsentSql, suffix));
|
|
sql.AppendLine(SuffixParameters(MonotonicUpdateSql, suffix));
|
|
foreach (SqlParameter p in BuildUpsertParameters(siteCall, siteCall.TrackedOperationId.Value.ToString("D"), ranks[i]))
|
|
{
|
|
p.ParameterName += suffix;
|
|
parameters.Add(p);
|
|
}
|
|
}
|
|
|
|
try
|
|
{
|
|
await _context.Database.ExecuteSqlRawAsync(sql.ToString(), parameters, ct).ConfigureAwait(false);
|
|
}
|
|
catch (SqlException ex)
|
|
{
|
|
_logger.LogDebug(
|
|
ex,
|
|
"SiteCallAuditRepository.UpsertManyAsync chunk of {Count} rows faulted (error {SqlErrorNumber}); re-applying it row by row.",
|
|
chunk.Count,
|
|
ex.Number);
|
|
foreach (var siteCall in chunk)
|
|
{
|
|
try
|
|
{
|
|
await UpsertAsync(siteCall, ct).ConfigureAwait(false);
|
|
}
|
|
catch (Exception rowFailure) when (rowFailure is not OperationCanceledException)
|
|
{
|
|
firstFailure ??= System.Runtime.ExceptionServices.ExceptionDispatchInfo.Capture(rowFailure);
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
firstFailure?.Throw();
|
|
}
|
|
|
|
/// <summary>Renames every <c>@Name</c> parameter reference in <paramref name="sql"/> to <c>@Name{suffix}</c>.</summary>
|
|
private static string SuffixParameters(string sql, string suffix) =>
|
|
System.Text.RegularExpressions.Regex.Replace(sql, @"@(\w+)", "@$1" + suffix);
|
|
|
|
// Leg 1 of the upsert: create the row when it does not exist yet. Runs FIRST
|
|
// so a concurrent first-write loser still has a row to update (see UpsertAsync).
|
|
private const string InsertIfAbsentSql = @"
|
|
IF NOT EXISTS (SELECT 1 FROM dbo.SiteCalls WHERE TrackedOperationId = @Id)
|
|
INSERT INTO dbo.SiteCalls
|
|
(TrackedOperationId, Channel, Target, SourceSite, SourceNode, Status, RetryCount,
|
|
LastError, HttpStatus, CreatedAtUtc, UpdatedAtUtc, TerminalAtUtc, IngestedAtUtc)
|
|
VALUES
|
|
(@Id, @Channel, @Target, @SourceSite, @SourceNode, @Status, @RetryCount,
|
|
@LastError, @HttpStatus, @CreatedAtUtc, @UpdatedAtUtc, @TerminalAtUtc, @IngestedAtUtc);";
|
|
|
|
// Leg 2 of the upsert: the monotonic guard. Also re-run standalone from the
|
|
// duplicate-key catch, which is why it lives in its own constant. Ownership
|
|
// is checked here as well as in the UPDATE predicate: an insert-race loser
|
|
// must not advance another site's operation. Throwing rolls back the paired
|
|
// audit/operational ingest transaction instead of acknowledging forged state.
|
|
private const string MonotonicUpdateSql = @"
|
|
IF EXISTS (SELECT 1 FROM dbo.SiteCalls
|
|
WHERE TrackedOperationId = @Id
|
|
AND SourceSite COLLATE Latin1_General_100_BIN2 <> @SourceSite COLLATE Latin1_General_100_BIN2)
|
|
THROW 51001, 'Site-call operation ownership denied.', 1;
|
|
|
|
UPDATE dbo.SiteCalls
|
|
SET Status = @Status,
|
|
RetryCount = @RetryCount,
|
|
LastError = @LastError,
|
|
HttpStatus = @HttpStatus,
|
|
UpdatedAtUtc = @UpdatedAtUtc,
|
|
TerminalAtUtc = @TerminalAtUtc,
|
|
IngestedAtUtc = @IngestedAtUtc,
|
|
SourceNode = COALESCE(@SourceNode, SourceNode)
|
|
WHERE TrackedOperationId = @Id
|
|
AND SourceSite COLLATE Latin1_General_100_BIN2 = @SourceSite COLLATE Latin1_General_100_BIN2
|
|
AND ( @Rank > (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)
|
|
OR ( @Rank = (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)
|
|
AND @Rank < @TerminalRank
|
|
AND UpdatedAtUtc < @UpdatedAtUtc ) );";
|
|
|
|
/// <summary>
|
|
/// Builds one FRESH parameter set for the upsert statements. Fresh per call
|
|
/// because a <see cref="SqlParameter"/> instance cannot be attached to two
|
|
/// commands, and the duplicate-key retry issues a second command.
|
|
/// </summary>
|
|
private static object[] BuildUpsertParameters(SiteCall siteCall, string idText, int incomingRank)
|
|
{
|
|
return
|
|
[
|
|
Param("@Id", SqlDbType.VarChar, idText),
|
|
Param("@Channel", SqlDbType.VarChar, siteCall.Channel),
|
|
Param("@Target", SqlDbType.VarChar, siteCall.Target),
|
|
Param("@SourceSite", SqlDbType.VarChar, siteCall.SourceSite),
|
|
Param("@SourceNode", SqlDbType.VarChar, siteCall.SourceNode),
|
|
Param("@Status", SqlDbType.VarChar, siteCall.Status),
|
|
Param("@RetryCount", SqlDbType.Int, siteCall.RetryCount),
|
|
Param("@LastError", SqlDbType.NVarChar, siteCall.LastError),
|
|
Param("@HttpStatus", SqlDbType.Int, siteCall.HttpStatus),
|
|
Param("@CreatedAtUtc", SqlDbType.DateTime2, siteCall.CreatedAtUtc),
|
|
Param("@UpdatedAtUtc", SqlDbType.DateTime2, siteCall.UpdatedAtUtc),
|
|
Param("@TerminalAtUtc", SqlDbType.DateTime2, siteCall.TerminalAtUtc),
|
|
Param("@IngestedAtUtc", SqlDbType.DateTime2, siteCall.IngestedAtUtc),
|
|
Param("@Rank", SqlDbType.Int, incomingRank),
|
|
Param("@TerminalRank", SqlDbType.Int, TerminalRank),
|
|
];
|
|
}
|
|
|
|
/// <summary>
|
|
/// Binds one explicitly-typed parameter, mapping a null CLR value to
|
|
/// <see cref="DBNull"/> while KEEPING the declared type. Size is never set —
|
|
/// SqlClient sizes from the value, so an over-long string is rejected by the
|
|
/// server rather than truncated client-side.
|
|
/// </summary>
|
|
private static SqlParameter Param(string name, SqlDbType type, object? value) =>
|
|
new(name, type) { Value = value ?? DBNull.Value };
|
|
|
|
/// <inheritdoc />
|
|
public async Task<SiteCall?> GetAsync(TrackedOperationId id, CancellationToken ct = default)
|
|
{
|
|
return await _context.Set<SiteCall>().FindAsync(new object?[] { id }, ct);
|
|
}
|
|
|
|
/// <inheritdoc />
|
|
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));
|
|
}
|
|
|
|
// Raw SQL with NAMED parameters (not FromSqlInterpolated). FromSqlInterpolated binds every interpolated string
|
|
// value as an nvarchar(4000) parameter by default; every SiteCalls string
|
|
// column is IsUnicode(false) (varchar), so those parameters forced an
|
|
// implicit CONVERT on the column side of each comparison and blocked
|
|
// index seeks — including the keyset predicate below, which is exactly
|
|
// what IX_SiteCalls_Created_TrackedOperationId exists to serve. Each
|
|
// string filter now gets an explicitly-typed SqlDbType.VarChar parameter,
|
|
// mirroring the write path's Param(...) helper above. A named ADO.NET
|
|
// parameter can appear more than once in the SQL text (unlike a
|
|
// positional FromSqlInterpolated hole) while still being added to the
|
|
// command's Parameters collection exactly once, so the "IS NULL OR
|
|
// col = @p" shape needs no duplicate parameter objects.
|
|
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;
|
|
var stuckCutoff = filter.StuckCutoffUtc;
|
|
// TrackedOperationId is a varchar(36) column holding the "D"-format GUID
|
|
// string (see SiteCallEntityTypeConfiguration), so the exact-match filter
|
|
// is compared as a string — same shape as the @AfterId cursor below. A
|
|
// uniqueidentifier parameter would make SQL Server convert the COLUMN side
|
|
// (higher type precedence) and lose the PK seek.
|
|
var trackedOperationIdString = filter.TrackedOperationId?.ToString("D");
|
|
|
|
// The stuck predicate (TerminalAtUtc IS NULL AND CreatedAtUtc < cutoff)
|
|
// is pushed into SQL here — both columns are plain (no value converter)
|
|
// and compose with the keyset cursor, so a StuckOnly page is honest:
|
|
// never under-filled with a non-null next cursor. Mirrors how
|
|
// NotificationOutboxRepository.QueryAsync applies NotificationOutboxFilter.StuckCutoff.
|
|
//
|
|
// SELECT-list maintenance: EF Core's FromSqlRaw requires every
|
|
// entity-tracked column to appear in the result set. Adding a new column
|
|
// to the SiteCall entity means extending the list below too — otherwise
|
|
// every read trips "The required column 'X' was not present" at runtime.
|
|
const string sql = @"
|
|
SELECT TOP (@PageSize)
|
|
TrackedOperationId, Channel, Target, SourceSite, SourceNode, Status, RetryCount,
|
|
LastError, HttpStatus, CreatedAtUtc, UpdatedAtUtc, TerminalAtUtc, IngestedAtUtc
|
|
FROM dbo.SiteCalls
|
|
WHERE (@Channel IS NULL OR Channel = @Channel)
|
|
AND (@SourceSite IS NULL OR SourceSite = @SourceSite)
|
|
AND (@SourceNode IS NULL OR SourceNode = @SourceNode)
|
|
AND (@Status IS NULL OR Status = @Status)
|
|
AND (@Target IS NULL OR Target = @Target)
|
|
AND (@TrackedOperationId IS NULL OR TrackedOperationId = @TrackedOperationId)
|
|
AND (@FromUtc IS NULL OR CreatedAtUtc >= @FromUtc)
|
|
AND (@ToUtc IS NULL OR CreatedAtUtc <= @ToUtc)
|
|
AND (@StuckCutoff IS NULL OR (TerminalAtUtc IS NULL AND CreatedAtUtc < @StuckCutoff))
|
|
AND (@HasCursor = 0
|
|
OR CreatedAtUtc < @AfterCreated
|
|
OR (CreatedAtUtc = @AfterCreated AND TrackedOperationId < @AfterId))
|
|
ORDER BY CreatedAtUtc DESC, TrackedOperationId DESC
|
|
-- Every filter above is the (@p IS NULL OR col = @p) optional-parameter shape, so a
|
|
-- single cached plan would be parameter-sniffed for whichever filters happened to be
|
|
-- non-null on first compile. RECOMPILE lets the optimizer prune the dead (@p IS NULL)
|
|
-- predicates per invocation and pick IX_SiteCalls_Status_Updated / IX_SiteCalls_NonTerminal /
|
|
-- IX_SiteCalls_Created_TrackedOperationId for the filters actually supplied (arch-review 04,
|
|
-- P5). Per-invocation compile cost is negligible at UI-page cadence.
|
|
OPTION (RECOMPILE);";
|
|
|
|
var parameters = new object[]
|
|
{
|
|
Param("@PageSize", SqlDbType.Int, paging.PageSize),
|
|
Param("@Channel", SqlDbType.VarChar, filter.Channel),
|
|
Param("@SourceSite", SqlDbType.VarChar, filter.SourceSite),
|
|
Param("@SourceNode", SqlDbType.VarChar, filter.SourceNode),
|
|
Param("@Status", SqlDbType.VarChar, filter.Status),
|
|
Param("@Target", SqlDbType.VarChar, filter.Target),
|
|
Param("@TrackedOperationId", SqlDbType.VarChar, trackedOperationIdString),
|
|
Param("@FromUtc", SqlDbType.DateTime2, fromUtc),
|
|
Param("@ToUtc", SqlDbType.DateTime2, toUtc),
|
|
Param("@StuckCutoff", SqlDbType.DateTime2, stuckCutoff),
|
|
Param("@HasCursor", SqlDbType.Int, hasCursor ? 1 : 0),
|
|
Param("@AfterCreated", SqlDbType.DateTime2, afterCreated),
|
|
Param("@AfterId", SqlDbType.VarChar, afterIdString),
|
|
};
|
|
|
|
var rows = await _context.Set<SiteCall>()
|
|
.FromSqlRaw(sql, parameters)
|
|
.AsNoTracking()
|
|
.ToListAsync(ct);
|
|
|
|
return rows;
|
|
}
|
|
|
|
/// <inheritdoc />
|
|
public async Task<int> PurgeTerminalAsync(DateTime olderThanUtc, CancellationToken ct = default)
|
|
{
|
|
// Time-sliced batches (arch-review 04 round 2, R6 — the one maintenance DELETE
|
|
// that missed round 1's batching pass): each DELETE covers at most one DAY of
|
|
// terminal rows, capping the lock/log footprint per statement. Steady state
|
|
// (daily purge, 365-day retention) is a single slice; only catch-up after an
|
|
// outage runs several. One-day (not one-hour) slices are proportionate to
|
|
// SiteCalls volume, which is far below KpiSample's. The MIN() anchor and the
|
|
// DELETE predicate both seek IX_SiteCalls_Terminal (filtered IS NOT NULL).
|
|
var total = 0;
|
|
var floor = await _context.SiteCalls
|
|
.Where(s => s.TerminalAtUtc != null && s.TerminalAtUtc < olderThanUtc)
|
|
.MinAsync(s => s.TerminalAtUtc, ct);
|
|
while (floor is not null && floor < olderThanUtc)
|
|
{
|
|
var ceiling = floor.Value.AddDays(1) < olderThanUtc ? floor.Value.AddDays(1) : olderThanUtc;
|
|
total += await _context.Database.ExecuteSqlInterpolatedAsync(
|
|
$"DELETE FROM dbo.SiteCalls WHERE TerminalAtUtc IS NOT NULL AND TerminalAtUtc < {ceiling};",
|
|
ct);
|
|
floor = await _context.SiteCalls
|
|
.Where(s => s.TerminalAtUtc != null && s.TerminalAtUtc < olderThanUtc)
|
|
.MinAsync(s => s.TerminalAtUtc, ct);
|
|
}
|
|
|
|
return total;
|
|
}
|
|
|
|
// Terminal status string literals for the interval-throughput KPIs. The
|
|
// Status column is a plain varchar (no value converter), so these compare
|
|
// directly in translated SQL.
|
|
//
|
|
// NOTE on the "buffered/non-terminal" definition: the SiteCalls operational
|
|
// mirror stores AuditStatus-derived strings (Attempted/Delivered/Parked/
|
|
// Failed/...), NOT the tracking-lifecycle Pending/Retrying names the spec's
|
|
// KPI section uses. There is therefore no Status string that means
|
|
// "buffered". The schema-honest predicate for "non-terminal / buffered" is
|
|
// TerminalAtUtc IS NULL — consistent with PurgeTerminalAsync's terminal
|
|
// predicate and with the SiteCall entity's own contract ("TerminalAtUtc ...
|
|
// null while still active"). All buffered / stuck / oldest-pending counts
|
|
// below key off TerminalAtUtc, not Status.
|
|
private const string StatusParked = "Parked";
|
|
private const string StatusDelivered = "Delivered";
|
|
private const string StatusFailed = "Failed";
|
|
|
|
/// <summary>
|
|
/// Restricts a KPI aggregation to the rows that contribute to at least one metric:
|
|
/// every non-terminal row (buffered / stuck / oldest-pending), every Parked row, and
|
|
/// Delivered/Failed rows that went terminal inside the interval. The historical
|
|
/// terminal bulk of the table contributes to none of them, so the aggregation's cost
|
|
/// tracks the working set rather than retention, and a site/node with only uncounted
|
|
/// rows does not surface as an all-zero per-scope row (preserving the old
|
|
/// union-of-metric-keys semantics). The filtered <c>IX_SiteCalls_NonTerminal</c> /
|
|
/// <c>IX_SiteCalls_Terminal</c> indexes cover the legs.
|
|
/// </summary>
|
|
private IQueryable<SiteCall> KpiContributingRows(DateTime intervalSince) =>
|
|
_context.SiteCalls.Where(s => s.TerminalAtUtc == null
|
|
|| s.Status == StatusParked
|
|
|| ((s.Status == StatusDelivered || s.Status == StatusFailed)
|
|
&& s.TerminalAtUtc >= intervalSince));
|
|
|
|
/// <inheritdoc />
|
|
public async Task<SiteCallKpiSnapshot> ComputeKpisAsync(
|
|
DateTime stuckCutoff, DateTime intervalSince, CancellationToken ct = default)
|
|
{
|
|
var now = DateTime.UtcNow;
|
|
|
|
// One conditional-aggregation pass replaces the five sequential COUNTs + Any +
|
|
// Min round trips (review 2026-10-21 06/PERF-3, 08/P3), mirroring
|
|
// NotificationOutboxRepository.ComputeKpisAsync. GroupBy(_ => 1) yields a single
|
|
// group (no contributing rows → no group → all zeros). The oldest non-terminal
|
|
// CreatedAtUtc is a MIN over a CASE in the same scan.
|
|
var counts = await KpiContributingRows(intervalSince)
|
|
.GroupBy(_ => 1)
|
|
.Select(g => new
|
|
{
|
|
Buffered = g.Count(s => s.TerminalAtUtc == null),
|
|
Parked = g.Count(s => s.Status == StatusParked),
|
|
Failed = g.Count(s => s.Status == StatusFailed
|
|
&& s.TerminalAtUtc != null && s.TerminalAtUtc >= intervalSince),
|
|
Delivered = g.Count(s => s.Status == StatusDelivered
|
|
&& s.TerminalAtUtc != null && s.TerminalAtUtc >= intervalSince),
|
|
Stuck = g.Count(s => s.TerminalAtUtc == null && s.CreatedAtUtc < stuckCutoff),
|
|
OldestCreatedAt = g.Min(s => s.TerminalAtUtc == null ? (DateTime?)s.CreatedAtUtc : null),
|
|
})
|
|
.FirstOrDefaultAsync(ct);
|
|
|
|
return new SiteCallKpiSnapshot(
|
|
BufferedCount: counts?.Buffered ?? 0,
|
|
ParkedCount: counts?.Parked ?? 0,
|
|
FailedLastInterval: counts?.Failed ?? 0,
|
|
DeliveredLastInterval: counts?.Delivered ?? 0,
|
|
OldestPendingAge: counts?.OldestCreatedAt is { } oldest ? now - oldest : null,
|
|
StuckCount: counts?.Stuck ?? 0);
|
|
}
|
|
|
|
/// <inheritdoc />
|
|
public async Task<IReadOnlyList<SiteCallSiteKpiSnapshot>> ComputePerSiteKpisAsync(
|
|
DateTime stuckCutoff, DateTime intervalSince, CancellationToken ct = default)
|
|
{
|
|
var now = DateTime.UtcNow;
|
|
|
|
// One grouped conditional-aggregation query (counts + MIN) instead of five
|
|
// grouped COUNTs + a grouped MIN (review 2026-10-21 06/PERF-3, 08/P3).
|
|
var rows = await KpiContributingRows(intervalSince)
|
|
.GroupBy(s => s.SourceSite)
|
|
.Select(g => new
|
|
{
|
|
Site = g.Key,
|
|
Buffered = g.Count(s => s.TerminalAtUtc == null),
|
|
Parked = g.Count(s => s.Status == StatusParked),
|
|
Failed = g.Count(s => s.Status == StatusFailed
|
|
&& s.TerminalAtUtc != null && s.TerminalAtUtc >= intervalSince),
|
|
Delivered = g.Count(s => s.Status == StatusDelivered
|
|
&& s.TerminalAtUtc != null && s.TerminalAtUtc >= intervalSince),
|
|
Stuck = g.Count(s => s.TerminalAtUtc == null && s.CreatedAtUtc < stuckCutoff),
|
|
OldestCreatedAt = g.Min(s => s.TerminalAtUtc == null ? (DateTime?)s.CreatedAtUtc : null),
|
|
})
|
|
.ToListAsync(ct);
|
|
|
|
return rows
|
|
.OrderBy(r => r.Site, StringComparer.Ordinal)
|
|
.Select(r => new SiteCallSiteKpiSnapshot(
|
|
SourceSite: r.Site,
|
|
BufferedCount: r.Buffered,
|
|
ParkedCount: r.Parked,
|
|
FailedLastInterval: r.Failed,
|
|
DeliveredLastInterval: r.Delivered,
|
|
OldestPendingAge: r.OldestCreatedAt is { } oldest ? now - oldest : null,
|
|
StuckCount: r.Stuck))
|
|
.ToList();
|
|
}
|
|
|
|
/// <inheritdoc />
|
|
public async Task<IReadOnlyList<SiteCallNodeKpiSnapshot>> ComputePerNodeKpisAsync(
|
|
DateTime stuckCutoff, DateTime intervalSince, CancellationToken ct = default)
|
|
{
|
|
var now = DateTime.UtcNow;
|
|
|
|
// Same single grouped aggregation as ComputePerSiteKpisAsync, keyed by
|
|
// SourceNode. Rows with a NULL SourceNode are excluded — per-node KPIs are only
|
|
// meaningful when the node identity is known.
|
|
var rows = await KpiContributingRows(intervalSince)
|
|
.Where(s => s.SourceNode != null)
|
|
.GroupBy(s => s.SourceNode!)
|
|
.Select(g => new
|
|
{
|
|
Node = g.Key,
|
|
Buffered = g.Count(s => s.TerminalAtUtc == null),
|
|
Parked = g.Count(s => s.Status == StatusParked),
|
|
Failed = g.Count(s => s.Status == StatusFailed
|
|
&& s.TerminalAtUtc != null && s.TerminalAtUtc >= intervalSince),
|
|
Delivered = g.Count(s => s.Status == StatusDelivered
|
|
&& s.TerminalAtUtc != null && s.TerminalAtUtc >= intervalSince),
|
|
Stuck = g.Count(s => s.TerminalAtUtc == null && s.CreatedAtUtc < stuckCutoff),
|
|
OldestCreatedAt = g.Min(s => s.TerminalAtUtc == null ? (DateTime?)s.CreatedAtUtc : null),
|
|
})
|
|
.ToListAsync(ct);
|
|
|
|
return rows
|
|
.OrderBy(r => r.Node, StringComparer.Ordinal)
|
|
.Select(r => new SiteCallNodeKpiSnapshot(
|
|
SourceNode: r.Node,
|
|
BufferedCount: r.Buffered,
|
|
ParkedCount: r.Parked,
|
|
FailedLastInterval: r.Failed,
|
|
DeliveredLastInterval: r.Delivered,
|
|
OldestPendingAge: r.OldestCreatedAt is { } oldest ? now - oldest : null,
|
|
StuckCount: r.Stuck))
|
|
.ToList();
|
|
}
|
|
|
|
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;
|
|
}
|
|
}
|