lmxopcua/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb/QuestDbSchemaBootstrapper.cs

using System.Globalization;
using Microsoft.Extensions.Logging;
using Microsoft.Extensions.Logging.Abstractions;
using Npgsql;
using ZB.MOM.WW.OtOpcUa.Driver.Historian.Sql.Common;

namespace ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb;

/// <summary>
///     Lazily creates the four tables once per instance (<c>CREATE TABLE IF NOT EXISTS</c>), then reconciles the TTL
///     of the sample and event tables with <see cref="QuestDbHistorianOptions.RetentionDays"/>.
///     <list type="bullet">
///     <item><b>Concurrency:</b> two nodes starting together are safe. If a CREATE still fails with a server error
///     whose message says <c>already exists</c>, <c>tables()</c> is re-checked and the error is swallowed only when the
///     table is really there; anything else is rethrown (SQLSTATE is always <c>00000</c>, so it cannot classify).</item>
///     <item><b>TTL (RC3):</b> <c>tables()</c> normalises units (7 days reads back as <c>1 WEEK</c>) so values are compared
///     in hours, and an ALTER becomes visible only after the WAL applies it, so the bootstrapper never re-reads to verify
///     its own ALTER. A boot whose TTL already matches issues no ALTER.</item>
///     <item><b>Tag index (#683):</b> a sample table created before the <c>tag</c> symbol index existed gains it with
///     <see cref="QuestDbTables.AddTagIndex"/> (idempotent; a boot that finds it indexed issues no ALTER, and like the TTL
///     the ALTER is a WAL transaction the bootstrapper does not re-read). Building it is a one-off WAL apply — ~3 s for
///     68M rows / 63 daily partitions measured — during which writes queue and reads keep scanning. A refused ALTER is
///     logged as a Warning, never fails the bootstrap (the index is optional); a failure at WAL-apply time would suspend
///     the table unseen, the same blind spot as the TTL ALTER.</item>
///     <item><b>Failure:</b> not latched. Within <see cref="BootstrapFailureCache.DefaultDuration"/> the last failure is
///     re-thrown without touching QuestDB.</item>
///     <item><c>AutoCreateSchema=false</c> only verifies the tables exist (no TTL changes) and throws
///     <see cref="HistorianSchemaMissingException"/> otherwise; a missing tag index is only logged (a Warning naming
///     the ALTER to run), since reads stay correct without it.</item>
///     </list>
/// </summary>
internal sealed class QuestDbSchemaBootstrapper
{
    private readonly QuestDbConnection _conn;
    private readonly QuestDbHistorianOptions _options;
    private readonly ILogger _log;
    private readonly SemaphoreSlim _gate = new(1, 1);
    private readonly BootstrapFailureCache _failure; // guarded by _gate
    private volatile bool _ready;

    /// <summary>Creates the bootstrapper.</summary>
    /// <param name="conn">Connection helper.</param>
    /// <param name="options">Validated options.</param>
    /// <param name="log">Optional logger.</param>
    /// <param name="timeProvider">Clock for the failure cache; system by default.</param>
    public QuestDbSchemaBootstrapper(QuestDbConnection conn, QuestDbHistorianOptions options, ILogger? log = null,
        TimeProvider? timeProvider = null)
    {
        _conn = conn;
        _options = options;
        _log = log ?? NullLogger.Instance;
        _failure = new BootstrapFailureCache(timeProvider ?? TimeProvider.System, BootstrapFailureCache.DefaultDuration);
    }

    /// <summary>Ensures the schema is ready; no-op after the first success.</summary>
    /// <param name="ct">Cancellation.</param>
    /// <returns>A task completing when the tables are usable.</returns>
    /// <exception cref="HistorianSchemaMissingException">AutoCreateSchema is false and a table is absent.</exception>
    public async Task EnsureAsync(CancellationToken ct)
    {
        if (_ready) return;
        await _gate.WaitAsync(ct).ConfigureAwait(false);
        try
        {
            if (_ready) return;
            _failure.ThrowIfCached();
            try
            {
                await using var conn = await _conn.OpenAsync(ct).ConfigureAwait(false);
                if (_options.AutoCreateSchema)
                {
                    await CreateTablesAsync(conn, ct).ConfigureAwait(false);
                    await ReconcileTtlAsync(conn, ct).ConfigureAwait(false);
                    await EnsureTagIndexAsync(conn, apply: true, ct).ConfigureAwait(false);
                }
                else
                {
                    await VerifyAsync(conn, ct).ConfigureAwait(false);
                    await EnsureTagIndexAsync(conn, apply: false, ct).ConfigureAwait(false);
                }
                _ready = true;
                _failure.Clear();
            }
            catch (Exception ex) when (ex is not OperationCanceledException || !ct.IsCancellationRequested)
            {
                _failure.Record(ex);
                _log.LogWarning("QuestDB historian schema bootstrap (prefix '{Prefix}') failed ({Exception}: {Message}); " +
                    "historian calls fail fast for {Seconds} s before retrying.",
                    _conn.TablePrefix, ex.GetType().Name, ex.Message, BootstrapFailureCache.DefaultDuration.TotalSeconds);
                throw;
            }
        }
        finally
        {
            _gate.Release();
        }
    }

    private async Task VerifyAsync(NpgsqlConnection conn, CancellationToken ct)
    {
        foreach (var table in QuestDbTables.All(_conn.TablePrefix))
        {
            if (!await _conn.TableExistsAsync(conn, table, ct).ConfigureAwait(false))
                throw new HistorianSchemaMissingException(
                    $"QuestDB historian table '{table}' is missing and AutoCreateSchema is false. " +
                    "Create the tables (see docs/Historian.md) or enable AutoCreateSchema.");
        }
    }

    private async Task CreateTablesAsync(NpgsqlConnection conn, CancellationToken ct)
    {
        var statements = QuestDbTables.CreateAll(_conn.TablePrefix, _options.RetentionDays);
        var tables = QuestDbTables.All(_conn.TablePrefix);
        for (var i = 0; i < statements.Length; i++)
        {
            try
            {
                await using var cmd = _conn.CreateCommand(conn, statements[i]);
                await cmd.ExecuteNonQueryAsync(ct).ConfigureAwait(false);
            }
            catch (PostgresException ex) when (QuestDbConnection.IsAlreadyExists(ex))
            {
                // A concurrent creator won the race - swallow ONLY if the table is really there.
                if (!await _conn.TableExistsAsync(conn, tables[i], ct).ConfigureAwait(false)) throw;
                _log.LogDebug("QuestDB table {Table} was created concurrently.", tables[i]);
            }
        }
    }

    /// <summary>
    ///     #683: indexes the sample table's <c>tag</c> column when it is not (a table created before the index existed).
    ///     <paramref name="apply"/> false (AutoCreateSchema off) only logs the statement to run.
    /// </summary>
    private async Task EnsureTagIndexAsync(NpgsqlConnection conn, bool apply, CancellationToken ct)
    {
        var table = _conn.SampleTable;
        await using (var cmd = _conn.CreateCommand(conn, QuestDbTables.TagIndexedQuery(table)))
        {
            var indexed = await cmd.ExecuteScalarAsync(ct).ConfigureAwait(false);
            if (indexed is null or DBNull || Convert.ToBoolean(indexed, CultureInfo.InvariantCulture)) return;
        }
        if (!apply)
        {
            _log.LogWarning("QuestDB historian table {Table} has no index on its tag column, so the data-range lookups of " +
                "processed reads scan partitions instead of seeking (#683). AutoCreateSchema is false; run: {Statement}",
                table, QuestDbTables.AddTagIndex(table));
            return;
        }
        // A WAL transaction: table_columns() keeps reporting false until it applies, so do NOT re-read to verify.
        // The index is optional (reads are correct without it), so a refused ALTER (e.g. no DDL rights) must not fail the
        // bootstrap: it is logged and retried on the next boot. A failure at WAL-APPLY time is not seen here (QuestDB would
        // suspend the table) — the same blind spot as the TTL ALTER.
        try
        {
            await using var alter = _conn.CreateCommand(conn, QuestDbTables.AddTagIndex(table));
            await alter.ExecuteNonQueryAsync(ct).ConfigureAwait(false);
        }
        catch (Exception ex) when (ex is not OperationCanceledException || !ct.IsCancellationRequested)
        {
            _log.LogWarning("Could not add the tag index to QuestDB table {Table} ({Exception}: {Message}); data-range lookups " +
                "of processed reads scan partitions instead of seeking (#683). Run: {Statement}",
                table, ex.GetType().Name, ex.Message, QuestDbTables.AddTagIndex(table));
            return;
        }
        _log.LogInformation("Added the tag index to QuestDB table {Table} (#683).", table);
    }

    private async Task ReconcileTtlAsync(NpgsqlConnection conn, CancellationToken ct)
    {
        var wantHours = _options.RetentionDays is { } d ? d * 24L : 0L;
        foreach (var table in QuestDbTables.Expiring(_conn.TablePrefix))
        {
            long haveHours;
            await using (var cmd = _conn.CreateCommand(conn, "SELECT ttlValue, ttlUnit FROM tables() WHERE table_name = $1"))
            {
                QuestDbConnection.BindText(cmd, table);
                await using var r = await cmd.ExecuteReaderAsync(ct).ConfigureAwait(false);
                if (!await r.ReadAsync(ct).ConfigureAwait(false)) continue; // vanished (dropped between create and here)
                var value = r.IsDBNull(0) ? 0L : Convert.ToInt64(r.GetValue(0));
                var unit = r.IsDBNull(1) ? null : r.GetString(1);
                haveHours = QuestDbTables.TtlHours(value, unit);
            }
            if (haveHours == wantHours) continue;

            // The ALTER is a WAL transaction: tables() keeps the old value until it applies, so do NOT re-read to verify.
            await using (var alter = _conn.CreateCommand(conn, QuestDbTables.AlterTtl(table, _options.RetentionDays)))
                await alter.ExecuteNonQueryAsync(ct).ConfigureAwait(false);
            if (_options.RetentionDays is { } days)
                _log.LogInformation("Applied TTL {Days} day(s) to QuestDB table {Table}.", days, table);
            else
                _log.LogInformation("Cleared TTL on QuestDB table {Table}.", table);
        }
    }
}