lmxopcua/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Historian/ContinuousHistorizationOptions.cs

using System.Collections.Generic;
using ZB.MOM.WW.OtOpcUa.Core.Abstractions;

namespace ZB.MOM.WW.OtOpcUa.Runtime.Historian;

/// <summary>
///     Binds the <c>ContinuousHistorization</c> configuration section that gates the continuous
///     historization of driver (non-Galaxy) tag values. When <see cref="Enabled"/> is <c>true</c>
///     <em>and</em> the <c>ServerHistorian</c> gateway is configured, the Host builds a durable,
///     crash-safe <c>FasterLogHistorizationOutbox</c> + a gateway-backed <c>IHistorianValueWriter</c>
///     and <c>WithOtOpcUaRuntimeActors</c> spawns the <see cref="ContinuousHistorizationRecorder"/>;
///     otherwise no recorder is spawned and driver tag values are not historized.
///     <para>
///         The recorder taps the per-node dependency-mux value fan-out, appends each numeric value to
///         the outbox (the crash boundary), and drains the outbox to the historian's SQL live-value
///         write path (<c>WriteLiveValues</c>) through the single <c>ServerHistorian</c> gateway. The
///         gateway connection (endpoint / key / TLS) is sourced from <c>ServerHistorianOptions</c> — this
///         section carries only the recorder + outbox knobs, never a gateway address or credential.
///     </para>
/// </summary>
public sealed class ContinuousHistorizationOptions
{
    /// <summary>The configuration section name this options class binds.</summary>
    public const string SectionName = "ContinuousHistorization";

    /// <summary>
    ///     When <c>true</c> (and the <c>ServerHistorian</c> gateway is configured), the
    ///     continuous-historization recorder + its durable outbox are wired and spawned; when
    ///     <c>false</c> (the default) no recorder is spawned and driver tag values are not historized.
    /// </summary>
    public bool Enabled { get; init; }

    /// <summary>
    ///     Directory holding the FasterLog outbox segment + commit files. Required when
    ///     <see cref="Enabled"/> is <c>true</c>. In production set an <b>absolute</b> path on durable
    ///     storage — a relative path resolves against the host's working directory, which may change
    ///     across deployments.
    /// </summary>
    public string OutboxPath { get; init; } = "";

    /// <summary>
    ///     Outbox commit cadence: <c>PerEntry</c> (the default, and what the shipped <c>appsettings.json</c>, docker-dev
    ///     and the unparseable-value fallback all use) fsyncs before each append returns — no loss window; <c>Periodic</c>
    ///     is the opt-in throughput mode that commits on a background timer every <see cref="CommitIntervalMs"/> ms
    ///     (100 ms by default), so a crash can lose at most the last interval of captured values. Parsed
    ///     case-insensitively against <c>HistorizationCommitMode</c>; see <see cref="ResolveCommitMode"/>.
    ///     <para>
    ///         <b>Trade-off.</b> Under <c>PerEntry</c> every ingest append waits for an fsync, so ingestion is capped at
    ///         <c>1 / fsync_latency</c> appends per second — but the recorder appends off its mailbox and coalesces
    ///         every burst waiting behind a slow append into one group commit, so it is one fsync per group, not per
    ///         value. Set <c>Periodic</c> explicitly where higher ingest throughput matters more than a bounded
    ///         (<c>&lt;= CommitIntervalMs</c>) crash-loss window.
    ///     </para>
    /// </summary>
    public string CommitMode { get; init; } = DefaultCommitMode.ToString();

    /// <summary>The commit mode used when <see cref="CommitMode"/> is not set or cannot be parsed.</summary>
    public const HistorizationCommitMode DefaultCommitMode = HistorizationCommitMode.PerEntry;

    /// <summary>Parses <see cref="CommitMode"/> case-insensitively; an unparseable value falls back to
    /// <see cref="DefaultCommitMode"/> so the fallback always agrees with the options default.</summary>
    public HistorizationCommitMode ResolveCommitMode() =>
        Enum.TryParse<HistorizationCommitMode>(CommitMode, ignoreCase: true, out var mode) ? mode : DefaultCommitMode;

    /// <summary>Periodic-mode commit cadence in milliseconds; must be positive when
    /// <see cref="CommitMode"/> is <c>Periodic</c>. Ignored under <c>PerEntry</c>.</summary>
    public int CommitIntervalMs { get; init; } = 100;

    /// <summary>
    ///     Maximum outbox entries peeked + written per drain pass; clamped to a positive value by the
    ///     recorder (a non-positive value falls back to the default).
    ///     <para>
    ///         Defaults to 1024. It was 64 when a pass wrote its per-tag groups <em>sequentially</em>, so
    ///         a wide batch cost <c>tags × latency</c> on the drain thread. Both halves of that reasoning
    ///         are gone: the peek is an in-memory scan of the outbox's live index (widening it costs
    ///         allocation, not I/O), and the pass is now bounded-parallel over
    ///         <see cref="DrainConcurrency"/>, so its cost is <c>⌈tags / concurrency⌉ × latency</c>
    ///         regardless of batch width. At 64, a drain draws down at most 64 entries every
    ///         <see cref="DrainIntervalSeconds"/> — below the arrival rate of even a modest historized
    ///         tag set — so any gateway hiccup left the outbox permanently behind, growing toward
    ///         <see cref="Capacity"/> and shedding oldest samples, rather than catching up once the
    ///         gateway recovered.
    ///     </para>
    /// </summary>
    public int DrainBatchSize { get; init; } = 1024;

    /// <summary>Steady drain cadence in seconds (also the post-success reschedule). Defaults to 2.</summary>
    public double DrainIntervalSeconds { get; init; } = 2;

    /// <summary>
    ///     Maximum historian tags a single drain pass writes concurrently. Defaults to 8; clamped to at
    ///     least 1 by the recorder (1 restores the fully sequential drain).
    ///     <para>
    ///         One peeked batch is grouped per historian tag and each group is an independent gateway
    ///         round-trip, so a sequential pass costs <c>tags × latency</c> — with a wide batch that is the
    ///         dominant term in how fast the outbox can drain. The window bounds how many of those
    ///         round-trips are outstanding at once, so a large tag count cannot flood the gateway. Nothing
    ///         about the ack changes: the pass is still acked all-or-nothing after every group has
    ///         reported, and the per-tag quarantine still keys on "another group wrote this
    ///         pass".
    ///     </para>
    /// </summary>
    public int DrainConcurrency { get; init; } = 8;

    /// <summary>
    ///     Maximum un-acked outbox entries before the drop-oldest capacity policy kicks in; <c>0</c>
    ///     means unbounded.
    ///     <para>
    ///         Defaults to 500,000 — roughly 50 MB of FasterLog, and hours-to-days of
    ///         backlog depending on tag count and update rate. It is a disk backstop for the one case
    ///         the per-tag quarantine cannot resolve: a historian that is unreachable or refusing
    ///         <em>every</em> write, where nothing proves any particular tag is at fault, so the
    ///         recorder correctly retains rather than discards. Set <c>0</c> to restore unbounded
    ///         retention where the outbox sits on storage that can absorb an arbitrarily long outage.
    ///     </para>
    /// </summary>
    public int Capacity { get; init; } = 500_000;

    /// <summary>
    ///     consecutive failed drain passes a single historian tag may accumulate — counted
    ///     only across passes where <em>other</em> tags wrote successfully, so a backend outage never
    ///     advances it — before that tag's queued entries are discarded and the FIFO head is allowed to
    ///     advance. Defaults to 5; clamped to at least 1 by the recorder.
    ///     <para>
    ///         Without this, one permanently-failing tag (typically one whose historian tag was never
    ///         provisioned) pins the head forever and every <em>healthy</em> tag in the batch is
    ///         re-written to the historian on every pass, at <see cref="DrainIntervalSeconds"/> cadence,
    ///         indefinitely. Each abandonment logs a Warning naming the tag.
    ///     </para>
    /// </summary>
    public int MaxWriteAttempts { get; init; } = 5;

    /// <summary>Initial retry backoff (seconds) after a failed drain pass. Defaults to 1.</summary>
    public double MinBackoffSeconds { get; init; } = 1;

    /// <summary>Cap (seconds) on the exponential retry backoff after repeated drain failures. Defaults to 30.</summary>
    public double MaxBackoffSeconds { get; init; } = 30;

    /// <summary>
    ///     when <c>true</c> (the default), a redundant node whose capture is stood down for its
    ///     mesh-partner Primary resumes capturing as soon as the cluster failure detector reports
    ///     that Primary <b>unreachable</b>, instead of waiting for auto-down and a fresh redundancy snapshot.
    ///     <para>
    ///         A crashed Primary otherwise leaves a capture hole of failure-detection + auto-down (~25 s on the
    ///         shipped <c>akka.conf</c>): the redundancy singleton usually died with it, so the standby's last
    ///         snapshot keeps naming the dead node Primary. With this on, the hole shrinks to the failure-detector
    ///         window (~10–12 s). The cost is duplicates: during a genuine partition both nodes capture from
    ///         detection onward, and a briefly-unreachable-but-alive Primary overlaps the standby. SqlServer and
    ///         QuestDB absorb a duplicate of a source-timestamped sample idempotently; a driver with no source
    ///         timestamp (e.g. Modbus, stamped with each node's own poll time) writes a duplicate row on every
    ///         backend, and the Gateway backend stores duplicates. <c>false</c> restores the previous behaviour.
    ///         Only the historization capture verdict is affected — never the alarm drain or the Primary data-plane
    ///         gate.
    ///     </para>
    /// </summary>
    public bool CaptureWhilePrimaryUnreachable { get; init; } = true;

    /// <summary>Returns operator-facing misconfiguration warnings for an <c>Enabled</c> recorder
    /// (empty when disabled or correctly configured). Pure — the registration logs each entry.</summary>
    /// <returns>Zero or more human-readable warning messages (never carrying secret values).</returns>
    public IReadOnlyList<string> Validate()
    {
        var warnings = new List<string>();
        if (!Enabled) return warnings;
        if (string.IsNullOrWhiteSpace(OutboxPath))
            warnings.Add("ContinuousHistorization:OutboxPath is empty while historization is enabled — the durable outbox has no directory to persist to; host startup will fail when the outbox is constructed. Set an absolute path on durable storage.");
        if (string.Equals(CommitMode, "Periodic", StringComparison.OrdinalIgnoreCase) && CommitIntervalMs <= 0)
            warnings.Add($"ContinuousHistorization:CommitIntervalMs is {CommitIntervalMs} — must be > 0 in Periodic commit mode; the periodic-commit loop cannot run.");
        if (DrainConcurrency < 1)
            warnings.Add($"ContinuousHistorization:DrainConcurrency is {DrainConcurrency} — must be >= 1; the recorder clamps it to 1 (a fully sequential drain).");
        return warnings;
    }
}