using System.Text;
using Npgsql;
using NpgsqlTypes;
using ZB.MOM.WW.OtOpcUa.Core.Abstractions;
using ZB.MOM.WW.OtOpcUa.Driver.Historian.Conformance;
using ZB.MOM.WW.OtOpcUa.Driver.Historian.Sql.Common;
namespace ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb.IntegrationTests;
///
/// The QuestDB historian as a conformance backend: a fresh unique table prefix per instance (dropped on dispose),
/// seeded by plain multi-row INSERTs (never through the writer or updater) followed by the WAL visibility wait.
/// Storage resolution is 1 µs (10 ticks); is , and
/// rounds the times it is given the same way before storing them.
///
internal sealed class QuestDbConformanceBackend : IConformanceBackend
{
private const int SeedBatchRows = 1000;
private readonly QuestDbFixture _fx;
private readonly string _prefix;
private QuestDbConformanceBackend(QuestDbFixture fx, string prefix, QuestDbHistorianOptions options,
QuestDbConnection conn, QuestDbSchemaBootstrapper boot)
{
_fx = fx;
_prefix = prefix;
Options = options;
Conn = conn;
Boot = boot;
Wal = new QuestDbWalVisibility(conn, options.VisibilityTimeout);
}
/// The options of this instance's table set.
public QuestDbHistorianOptions Options { get; }
/// The connection helper.
public QuestDbConnection Conn { get; }
/// The schema bootstrapper (already run).
public QuestDbSchemaBootstrapper Boot { get; }
/// The WAL visibility wait.
public QuestDbWalVisibility Wal { get; }
/// Creates a backend over a fresh table prefix; skips the calling test when the gate is closed.
/// The shared fixture.
/// The backend.
public static async Task CreateAsync(QuestDbFixture fx) => await CreateConcreteAsync(fx);
/// As , typed (for QuestDB-specific tests that need the connection).
/// The shared fixture.
/// The backend.
public static async Task CreateConcreteAsync(QuestDbFixture fx)
{
fx.SkipIfUnavailable();
var prefix = fx.UniquePrefix();
var o = fx.Options(prefix);
var conn = new QuestDbConnection(o);
var boot = new QuestDbSchemaBootstrapper(conn, o);
await boot.EnsureAsync(CancellationToken.None);
return new QuestDbConformanceBackend(fx, prefix, o, conn, boot);
}
///
public long ResolutionTicks => 10;
///
public DateTime Round(DateTime t) => QuestDbConnection.Round(t);
/// A fresh tag catalog (fresh positive cache) over this table set.
public QuestDbTagCatalog NewCatalog() => new(Conn, Boot, Wal);
///
public IHistorianDataSource CreateDataSource(TimeProvider? clock = null) =>
new QuestDbHistorianDataSource(Conn, Boot, NewCatalog(), timeProvider: clock);
///
public IHistorianDataSource CreateDataSource(int maxTimeWeightedRawRows) =>
new QuestDbHistorianDataSource(Conn, Boot, NewCatalog(), maxTimeWeightedRawRows: maxTimeWeightedRawRows);
///
/// The T4 live-value writer over a fresh catalog (fresh positive cache) and this instance's WAL wait.
public IHistorianValueWriter CreateValueWriter(Func? utcNow = null) =>
new QuestDbHistorianValueWriter(Conn, NewCatalog(), Wal, utcNow: utcNow);
///
/// The T6 soft-delete updater over a fresh catalog and this instance's WAL wait.
public IHistorianDataUpdater CreateUpdater() => new QuestDbHistorianDataUpdater(Conn, NewCatalog(), Wal);
///
///
/// Times are rounded to the microsecond () before they are stored; the tag row is registered
/// through the catalog (data type 0, as the SQL Server seed does); every row is live (deleted = false);
/// server_ts = rounded time + 1 s. Waits for WAL visibility before returning.
///
public async Task SeedAsync(string tag, params (DateTime T, HistorianValue V, uint Status)[] samples)
{
await NewCatalog().EnsureAsync(tag, 0, CancellationToken.None);
await InsertRowsAsync(tag, samples.Select(s => (Round(s.T), s.V, s.Status, Deleted: false)).ToArray());
}
///
/// Inserts raw rows (the tag must already be registered when reads should find it), including tombstones, then
/// waits for WAL visibility. A tombstone re-inserted at an existing (ts, tag) replaces that row (DEDUP).
///
/// The tag.
/// Rows: time (already rounded by the caller or not — it is bound through BindTimestamp), value, status, deleted.
public async Task InsertRowsAsync(string tag, (DateTime T, HistorianValue V, uint Status, bool Deleted)[] rows)
{
if (rows.Length == 0) return;
await InsertRowsNoWaitAsync(tag, rows);
await Wal.WaitAsync(Conn.SampleTable, CancellationToken.None);
}
///
/// without the WAL visibility wait: the rows become visible whenever QuestDB applies
/// them (the live write/read race stress test commits one row per call and never waits).
///
/// The tag.
/// Rows: time, value, status, deleted.
public async Task InsertRowsNoWaitAsync(string tag, (DateTime T, HistorianValue V, uint Status, bool Deleted)[] rows)
{
if (rows.Length == 0) return;
await using (var c = await Conn.OpenAsync(CancellationToken.None))
{
for (var offset = 0; offset < rows.Length; offset += SeedBatchRows)
{
var batch = rows.Skip(offset).Take(SeedBatchRows).ToArray();
var sql = new StringBuilder(
$"INSERT INTO {Conn.SampleTable} (ts, tag, kind, num_value, int_value, str_value, status, deleted, server_ts) VALUES ");
await using var cmd = Conn.CreateCommand(c, "");
var n = 0;
for (var i = 0; i < batch.Length; i++)
{
var (t, v, status, deleted) = batch[i];
var col = SampleColumnMapper.ToColumns(v);
if (i > 0) sql.Append(", ");
sql.Append('(');
for (var j = 0; j < 9; j++) sql.Append(j == 0 ? "" : ", ").Append('$').Append(++n);
sql.Append(')');
QuestDbConnection.BindTimestamp(cmd, t);
QuestDbConnection.BindText(cmd, tag);
QuestDbConnection.BindByte(cmd, col.ValueKind);
cmd.Parameters.Add(new NpgsqlParameter { NpgsqlDbType = NpgsqlDbType.Double, Value = (object?)col.NumValue ?? DBNull.Value });
cmd.Parameters.Add(new NpgsqlParameter { NpgsqlDbType = NpgsqlDbType.Bigint, Value = (object?)col.IntValue ?? DBNull.Value });
QuestDbConnection.BindText(cmd, col.StrValue);
QuestDbConnection.BindLong(cmd, SampleColumnMapper.ToLongStatus(status));
cmd.Parameters.Add(new NpgsqlParameter { NpgsqlDbType = NpgsqlDbType.Boolean, Value = deleted });
QuestDbConnection.BindTimestamp(cmd, Round(t).AddSeconds(1));
}
cmd.CommandText = sql.ToString();
await cmd.ExecuteNonQueryAsync();
}
}
}
///
public async Task SeedEventsAsync(params (DateTime T, string Id, string Source)[] events)
{
await using (var c = await Conn.OpenAsync(CancellationToken.None))
{
foreach (var (t, id, source) in events)
{
await using var cmd = Conn.CreateCommand(c,
$"INSERT INTO {Conn.EventTable} (ts, event_id, source, severity, message, received_utc) VALUES ($1, $2, $3, $4, $5, $6)");
QuestDbConnection.BindTimestamp(cmd, t);
QuestDbConnection.BindText(cmd, id);
QuestDbConnection.BindText(cmd, source);
cmd.Parameters.Add(new NpgsqlParameter { NpgsqlDbType = NpgsqlDbType.Integer, Value = 500 });
QuestDbConnection.BindText(cmd, null);
QuestDbConnection.BindTimestamp(cmd, t);
await cmd.ExecuteNonQueryAsync();
}
}
await Wal.WaitAsync(Conn.EventTable, CancellationToken.None);
}
/// Runs a statement against this table set (test-only escape hatch for hand-made rows), then waits for WAL visibility of the sample table.
/// The statement; parameters are bound positionally by .
/// Binds the parameters.
public async Task ExecuteAsync(string sql, Action bind)
{
await using (var c = await Conn.OpenAsync(CancellationToken.None))
{
await using var cmd = Conn.CreateCommand(c, sql);
bind(cmd);
await cmd.ExecuteNonQueryAsync();
}
await Wal.WaitAsync(Conn.SampleTable, CancellationToken.None);
}
///
public async Task> ReadBackAsync(string tag)
{
await using var c = await Conn.OpenAsync(CancellationToken.None);
await using var cmd = Conn.CreateCommand(c,
$"SELECT ts, kind, num_value, int_value, str_value, status FROM {Conn.SampleTable} " +
"WHERE tag = $1 AND NOT deleted ORDER BY ts");
QuestDbConnection.BindText(cmd, tag);
var list = new List<(DateTime, HistorianValue, uint)>();
await using var r = await cmd.ExecuteReaderAsync();
while (await r.ReadAsync())
{
var v = QuestDbHistorianDataSource.ToValue(QuestDbConnection.ReadKind(r, 1),
r.IsDBNull(2) ? null : r.GetDouble(2), r.IsDBNull(3) ? null : r.GetInt64(3), r.IsDBNull(4) ? null : r.GetString(4));
// Verbatim: no read-side clamp. A NULL status cannot be written by any code path; it reads as the raw 0 bits.
var status = r.IsDBNull(5) ? 0u : SampleColumnMapper.FromLongStatus(r.GetInt64(5));
list.Add((QuestDbConnection.ReadUtc(r, 0), v, status));
}
return list;
}
///
public async Task TagExistsAsync(string tag)
{
await using var c = await Conn.OpenAsync(CancellationToken.None);
await using var cmd = Conn.CreateCommand(c, $"SELECT count() FROM {Conn.TagTable} WHERE name = $1");
QuestDbConnection.BindText(cmd, tag);
return Convert.ToInt64(await cmd.ExecuteScalarAsync()) > 0;
}
///
public async Task TagDataTypeAsync(string tag)
{
await using var c = await Conn.OpenAsync(CancellationToken.None);
await using var cmd = Conn.CreateCommand(c, $"SELECT data_type FROM {Conn.TagTable} WHERE name = $1 LIMIT 1");
QuestDbConnection.BindText(cmd, tag);
await using var r = await cmd.ExecuteReaderAsync();
return await r.ReadAsync() ? (DriverDataType)QuestDbConnection.ReadKind(r, 0) : null;
}
///
public async ValueTask DisposeAsync()
{
await Conn.DisposeAsync();
await _fx.DropPrefixAsync(_prefix); // early; the fixture also drops every prefix it handed out
}
}