sqlinsight-pro/src/SQLInsight.Collectors/DynamicsAx/AxBatchCollector.cs

using SQLInsight.Core.DynamicsAx;
using SQLInsight.Core.Instances;

namespace SQLInsight.Collectors.DynamicsAx;

/// <summary>
/// Batch jobs in every monitored AX database where the agent may read the batch tables (optional AX read access),
/// every 5 minutes: status, planned and actual start, and completed runs (from BATCHJOBHISTORY in AX 2009 and later,
/// from what SQLInsight itself sees in AX 3.0/4.0). Each job is compared with its own history, kept by SQLInsight in
/// its repository (insight.AxBatchRuns). Only standard AX tables are read; bounded with TOP; never written to.
/// </summary>
public sealed class AxBatchCollector(AxDatabaseDetector detector, AxThresholds? thresholds = null) : Collector<AxBatchSnapshot>
{
    /// <summary>Run keys remembered to avoid storing a run twice; more than this and the set starts over (the repository also ignores duplicates).</summary>
    private const int RecentRunKeysKept = 20_000;

    /// <summary>Finished rows older than this are not kept in the snapshot (AX 3.0 keeps one BATCH row per recurrence).</summary>
    private static readonly TimeSpan FinishedShown = TimeSpan.FromDays(7);

    private const int JobsShown = 500;

    private readonly AxThresholds _thresholds = thresholds ?? new AxThresholds();

    public override string Name => "dynamics-ax-batch";
    public override TimeSpan DefaultInterval => TimeSpan.FromMinutes(5);
    public override IReadOnlyCollection<Capability> RequiredCapabilities => [Capability.DynamicManagementViews];

    protected override async Task<AxBatchSnapshot> CollectAsync(CollectorContext context, AxBatchSnapshot? previous)
    {
        var ax = await detector.GetAsync(context).ConfigureAwait(false);
        var history = previous?.History.Clone() ?? new AxBatchHistory();
        var firstSeen = new Dictionary<string, DateTime>(StringComparer.OrdinalIgnoreCase);
        var recentKeys = previous?.RecentRunKeys is { Count: < RecentRunKeysKept } kept ? new HashSet<string>(kept) : [];
        var databases = new List<AxBatchDatabase>();
        foreach (var database in ax.Databases)
        {
            databases.Add(await CollectDatabaseAsync(context, database, previous, history, firstSeen, recentKeys).ConfigureAwait(false));
        }

        return new AxBatchSnapshot
        {
            CapturedAt = context.Now,
            Thresholds = _thresholds,
            Databases = databases,
            History = history,
            FirstSeenExecuting = firstSeen,
            RecentRunKeys = recentKeys,
        };
    }

    private static async Task<AxBatchDatabase> CollectDatabaseAsync(CollectorContext context, AxDatabase database, AxBatchSnapshot? previous, AxBatchHistory history,
        Dictionary<string, DateTime> firstSeen, HashSet<string> recentKeys)
    {
        var name = database.Database;
        var earlier = previous?.Databases.FirstOrDefault(d => string.Equals(d.Database, name, StringComparison.OrdinalIgnoreCase));
        if (AxSchemaAdapter.For(database.Version) is not { } adapter)
        {
            return new AxBatchDatabase { Database = name, Version = database.Version, NotMeasurable = "Ikke målbar: AX-versionen kunne ikke bestemmes ud fra skemaet." };
        }

        var query = adapter.BatchJobs(database);
        var (rows, problem) = await AxReader.ReadAsync(context, database, query, Map).ConfigureAwait(false);
        if (rows is null)
        {
            // A lock timeout keeps the previous picture (marked as such), so findings do not flap.
            return AxReader.IsTransient(problem) && earlier is { NotMeasurable: null }
                ? earlier with { NewRuns = [], Notes = [.. earlier.Notes.Where(n => n != AxReader.LockTimeoutText), AxReader.LockTimeoutText] }
                : new AxBatchDatabase { Database = name, Version = database.Version, NotMeasurable = problem };
        }

        var notes = new List<string>(query.Notes);
        var newRuns = new List<AxBatchRun>();
        long? watermark = earlier?.HistoryWatermark;

        // AX 2009 and later: completed runs from AX's own history, by RECID.
        var historyQuery = adapter.BatchHistory(database, watermark);
        var (historyRows, historyProblem) = await AxReader.ReadAsync(context, database, historyQuery, MapHistory).ConfigureAwait(false);
        var fromHistoryTable = historyRows is not null;
        if (historyRows is not null)
        {
            foreach (var row in historyRows)
            {
                watermark = Math.Max(watermark ?? 0, row.RecId);
                if (adapter.HistoryRun(row) is { } run && recentKeys.Add(Scoped(name, run.RunKey)))
                {
                    newRuns.Add(run);
                }
            }
        }
        else if (historyProblem is not null)
        {
            notes.Add("Kørselshistorik fra BATCHJOBHISTORY: " + historyProblem + " SQLInsight bruger i stedet de kørsler, den selv ser.");
        }

        // Pass 1: when jobs were first seen executing, and (AX 3.0/4.0, or no history access) runs SQLInsight saw finish.
        var prelim = new List<(AxBatchJobRow Row, AxBatchJob Job)>();
        foreach (var row in rows)
        {
            var seenKey = AxBatchHistory.Key(name, "#" + row.RecId);
            DateTime? seenBefore = previous?.FirstSeenExecuting.TryGetValue(seenKey, out var seen) == true ? seen : null;
            var job = adapter.Interpret(row, history, name, null);
            if (job.Status == AxBatchStatus.Executing)
            {
                firstSeen[seenKey] = seenBefore ?? context.Now.UtcDateTime;
            }
            else if (!fromHistoryTable && AxSchemaAdapter.ObservedRun(job, seenBefore) is { } run && recentKeys.Add(Scoped(name, run.RunKey)))
            {
                newRuns.Add(run);
            }

            prelim.Add((row, job));
        }

        foreach (var run in newRuns)
        {
            history.Add(name, run);
        }

        // Pass 2: the jobs with their history's statistics, the interesting ones first.
        var now = context.Now.UtcDateTime;
        var jobs = prelim
            .Select(p => adapter.Interpret(p.Row, history, name, firstSeen.TryGetValue(AxBatchHistory.Key(name, "#" + p.Row.RecId), out var s) ? s : null))
            .Where(j => j.Status is not (AxBatchStatus.Finished or AxBatchStatus.Canceled) || (j.EndedUtc ?? j.PlannedStartUtc ?? now) >= now - FinishedShown)
            .OrderBy(j => j.Status switch { AxBatchStatus.Executing => 0, AxBatchStatus.Error => 1, AxBatchStatus.Waiting => 2, _ => 3 })
            .ThenByDescending(j => j.RecId)
            .Take(JobsShown)
            .ToList();

        return new AxBatchDatabase
        {
            Database = name,
            Version = database.Version,
            Jobs = jobs,
            NewRuns = newRuns,
            StatusCounts = rows.GroupBy(r => r.Status).ToDictionary(g => g.Key, g => g.Count()),
            Notes = notes,
            HistoryWatermark = watermark,
        };
    }

    private static string Scoped(string database, string runKey) => database + "\u001f" + runKey;

    private static AxBatchJobRow Map(Row r) => new(
        r.Long(0), r.NullableInt(1), r.NullableString(2), r.NullableString(3), r.NullableString(4), r.Int(5),
        r.NullableDateTime(6), r.NullableInt(7), r.NullableDateTime(8), r.NullableInt(9), r.NullableDateTime(10), r.NullableInt(11), r.Int(12));

    private static AxBatchHistoryRow MapHistory(Row r) => new(
        r.Long(0), r.NullableLong(1), r.NullableString(2), r.NullableString(3), r.Int(4), r.NullableDateTime(5), r.NullableDateTime(6), r.Int(7));
}