lmxopcua/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Sql.Common/AlarmBatchWrite.cs

using ZB.MOM.WW.OtOpcUa.Core.AlarmHistorian;

namespace ZB.MOM.WW.OtOpcUa.Driver.Historian.Sql.Common;

/// <summary>
///     The batched alarm-history write both SQL backends share: events go to the database in chunks of one statement each
///     (one round trip per chunk, not per event), and the per-event outcomes the store-and-forward drain needs are kept:
///     <list type="bullet">
///         <item>a chunk that succeeds is Ack for every event in it (each was inserted or already present — the writes are
///             insert-if-absent / deduplicated by event id);</item>
///         <item>a chunk failing <see cref="HistorianWriteOutcome.RetryPlease"/>-class (transient: the database is down or
///             busy) is RetryPlease for all its events — re-sending them one by one during an outage would only multiply
///             the round trips;</item>
///         <item>a chunk failing <see cref="HistorianWriteOutcome.PermanentFail"/>-class (a data error, or anything the
///             caller's classifier wants isolated) is re-sent ONE EVENT AT A TIME, so only the poison event dead-letters and
///             its siblings still Ack;</item>
///         <item>a cancellation of the caller's token makes every event not yet decided RetryPlease (never a dead letter).</item>
///     </list>
/// </summary>
public static class AlarmBatchWrite
{
    /// <summary>Writes the undecided events of a batch and returns one outcome per event.</summary>
    /// <param name="outcomes">
    ///     One slot per batch event; a non-null slot is already decided (e.g. a pre-1970 time on QuestDB) and is not sent.
    ///     Filled in place; every slot is non-null on return.
    /// </param>
    /// <param name="chunkSize">Events per statement (at least 1).</param>
    /// <param name="writeChunk">Writes the events at the given batch indexes in ONE statement; throws on failure.</param>
    /// <param name="classify">Maps a write failure to the outcome it implies (see the summary for multi-event chunks).</param>
    /// <param name="onFailure">Optional diagnostic hook per failure (the exception and its classified outcome).</param>
    /// <param name="ct">The drain's cancellation token.</param>
    /// <returns>The filled outcomes.</returns>
    public static async Task<HistorianWriteOutcome[]> RunAsync(
        HistorianWriteOutcome?[] outcomes, int chunkSize,
        Func<IReadOnlyList<int>, CancellationToken, Task> writeChunk,
        Func<Exception, HistorianWriteOutcome> classify,
        Action<Exception, HistorianWriteOutcome>? onFailure,
        CancellationToken ct)
    {
        ArgumentNullException.ThrowIfNull(outcomes);
        ArgumentNullException.ThrowIfNull(writeChunk);
        ArgumentNullException.ThrowIfNull(classify);
        chunkSize = Math.Max(1, chunkSize);

        var pending = new List<int>(outcomes.Length);
        for (var i = 0; i < outcomes.Length; i++)
            if (outcomes[i] is null) pending.Add(i);

        for (var start = 0; start < pending.Count; start += chunkSize)
        {
            var chunk = pending.GetRange(start, Math.Min(chunkSize, pending.Count - start));
            var outcome = await TryAsync(chunk).ConfigureAwait(false);
            if (outcome == HistorianWriteOutcome.PermanentFail && chunk.Count > 1)
            {
                // Isolate the poison event: one statement per event, so the siblings still land.
                foreach (var index in chunk)
                    outcomes[index] = await TryAsync([index]).ConfigureAwait(false);
                continue;
            }
            foreach (var index in chunk) outcomes[index] ??= outcome;
        }

        var result = new HistorianWriteOutcome[outcomes.Length];
        for (var i = 0; i < outcomes.Length; i++) result[i] = outcomes[i] ?? HistorianWriteOutcome.RetryPlease;
        return result;

        async Task<HistorianWriteOutcome> TryAsync(IReadOnlyList<int> indexes)
        {
            if (ct.IsCancellationRequested) return HistorianWriteOutcome.RetryPlease;
            try
            {
                await writeChunk(indexes, ct).ConfigureAwait(false);
                return HistorianWriteOutcome.Ack;
            }
            catch (OperationCanceledException) when (ct.IsCancellationRequested)
            {
                return HistorianWriteOutcome.RetryPlease; // shutdown: never a dead letter
            }
            catch (Exception ex)
            {
                var o = classify(ex);
                onFailure?.Invoke(ex, o);
                return o;
            }
        }
    }
}