tests/MaxSystems.IntegrationTests/Infrastructure/IntegrationTestDatabase.cs - master - afanasevn/MaxSystems

tests/MaxSystems.IntegrationTests/Infrastructure/IntegrationTestDatabase.cs
380 строк15 KB
IBS\NAfanasev
Update solution references
23 июн 2026, 16:24
23 июн 2026, 16:2402b1783
using MaxSystems.Domain.Enums;
using MaxSystems.Infrastructure.Integration;
using Npgsql;

namespace MaxSystems.IntegrationTests.Infrastructure;

/// <summary>
/// Прямые SQL-запросы к PostgreSQL для assert'ов без EF в тестах.
/// </summary>
internal static class IntegrationTestDatabase
{
    /// <summary>
    /// Wolverine PostgreSQL message storage (см. <see cref="IntegrationWolverinePublishConfiguration.MessageStorageSchema"/>).
    /// Проверить имя таблицы после major-апгрейда Wolverine.
    /// </summary>
    public const string OutgoingEnvelopesTable = "wolverine_outgoing_envelopes";

    /// <summary>
    /// Все строки журнала доставки для указанной работы.
    /// </summary>
    public static async Task<IReadOnlyList<DeliveryLogRow>> GetDeliveryLogsForWorkAsync(
        string postgresConnectionString,
        Guid workId,
        CancellationToken ct = default)
    {
        await using var connection = new NpgsqlConnection(postgresConnectionString);
        await connection.OpenAsync(ct);

        await using var command = new NpgsqlCommand(
            """
            SELECT "EventType", "RoutingKey", "QueueName", "Status", "WorkId", "MessageId"
            FROM integration_delivery_logs
            WHERE "WorkId" = @workId
            ORDER BY "EventType", "QueueName"
            """,
            connection);

        command.Parameters.AddWithValue("workId", workId);

        var rows = new List<DeliveryLogRow>();
        await using var reader = await command.ExecuteReaderAsync(ct);
        while (await reader.ReadAsync(ct))
        {
            rows.Add(new DeliveryLogRow(
                reader.GetString(0),
                reader.GetString(1),
                reader.GetString(2),
                (IntegrationDeliveryStatus)reader.GetInt32(3),
                reader.IsDBNull(4) ? null : reader.GetGuid(4),
                reader.IsDBNull(5) ? null : reader.GetGuid(5)));
        }

        return rows;
    }

    /// <summary>Общее число записей в integration_delivery_logs (для rollback-тестов).</summary>
    public static async Task<int> CountDeliveryLogsAsync(
        string postgresConnectionString,
        CancellationToken ct = default)
    {
        await using var connection = new NpgsqlConnection(postgresConnectionString);
        await connection.OpenAsync(ct);

        await using var command = new NpgsqlCommand(
            "SELECT COUNT(*) FROM integration_delivery_logs",
            connection);

        return Convert.ToInt32(await command.ExecuteScalarAsync(ct), System.Globalization.CultureInfo.InvariantCulture);
    }

    /// <summary>Число исходящих envelope в transactional outbox Wolverine.</summary>
    public static async Task<int> CountOutgoingEnvelopesAsync(
        string postgresConnectionString,
        CancellationToken ct = default)
    {
        await using var connection = new NpgsqlConnection(postgresConnectionString);
        await connection.OpenAsync(ct);

        var sql =
            $"""
             SELECT COUNT(*)
             FROM {IntegrationWolverinePublishConfiguration.MessageStorageSchema}.{OutgoingEnvelopesTable}
             """;

        await using var command = new NpgsqlCommand(sql, connection);
        return Convert.ToInt32(await command.ExecuteScalarAsync(ct), System.Globalization.CultureInfo.InvariantCulture);
    }

    /// <summary>Ждёт строку журнала с ожидаемым статусом для работы и очереди.</summary>
    public static async Task<DeliveryLogDetailRow> WaitForDeliveryLogStatusAsync(
        string postgresConnectionString,
        Guid workId,
        string queueName,
        IntegrationDeliveryStatus expectedStatus,
        TimeSpan timeout,
        CancellationToken ct = default)
    {
        var deadline = DateTime.UtcNow + timeout;

        while (DateTime.UtcNow < deadline)
        {
            await using var connection = new NpgsqlConnection(postgresConnectionString);
            await connection.OpenAsync(ct);

            await using var command = new NpgsqlCommand(
                """
                SELECT "RoutingKey", "PayloadSummary", "AttemptCount", "ErrorMessage"
                FROM integration_delivery_logs
                WHERE "WorkId" = @workId
                  AND "QueueName" = @queueName
                  AND "Status" = @status
                ORDER BY "UpdatedAt" DESC
                LIMIT 1
                """,
                connection);

            command.Parameters.AddWithValue("workId", workId);
            command.Parameters.AddWithValue("queueName", queueName);
            command.Parameters.AddWithValue("status", (int)expectedStatus);

            await using var reader = await command.ExecuteReaderAsync(ct);
            if (await reader.ReadAsync(ct))
            {
                return new DeliveryLogDetailRow(
                    reader.GetString(0),
                    reader.IsDBNull(1) ? null : reader.GetString(1),
                    reader.GetInt32(2),
                    reader.IsDBNull(3) ? null : reader.GetString(3));
            }

            await Task.Delay(TimeSpan.FromMilliseconds(250), ct);
        }

        throw new InvalidOperationException(
            $"Delivery log for work {workId}, queue {queueName} did not reach status {expectedStatus} within {timeout}.");
    }

    /// <summary>Все строки journal maintenance.overdue (для assert'ов cron/dedup).</summary>
    public static async Task<IReadOnlyList<MaintenanceDeliveryLogRow>> GetMaintenanceOverdueDeliveryLogsAsync(
        string postgresConnectionString,
        CancellationToken ct = default)
    {
        await using var connection = new NpgsqlConnection(postgresConnectionString);
        await connection.OpenAsync(ct);

        await using var command = new NpgsqlCommand(
            """
            SELECT "EventType", "RoutingKey", "QueueName", "Status", "WorkId",
                   "EquipmentId", "MaintenanceRegulationId", "MessageId"
            FROM integration_delivery_logs
            WHERE "EventType" = @eventType
            ORDER BY "EquipmentId", "QueueName"
            """,
            connection);

        command.Parameters.AddWithValue("eventType", MaxSystems.Application.Integration.IntegrationEventTypes.MaintenanceOverdue);

        var rows = new List<MaintenanceDeliveryLogRow>();
        await using var reader = await command.ExecuteReaderAsync(ct);
        while (await reader.ReadAsync(ct))
        {
            rows.Add(new MaintenanceDeliveryLogRow(
                reader.GetString(0),
                reader.GetString(1),
                reader.GetString(2),
                (IntegrationDeliveryStatus)reader.GetInt32(3),
                reader.IsDBNull(4) ? null : reader.GetGuid(4),
                reader.IsDBNull(5) ? null : reader.GetGuid(5),
                reader.IsDBNull(6) ? null : reader.GetGuid(6),
                reader.IsDBNull(7) ? null : reader.GetGuid(7)));
        }

        return rows;
    }

    /// <summary>Ждёт Delivered для maintenance.overdue по оборудованию и очереди.</summary>
    public static async Task<DeliveryLogDetailRow> WaitForMaintenanceOverdueDeliveryLogStatusAsync(
        string postgresConnectionString,
        Guid equipmentId,
        Guid maintenanceRegulationId,
        string queueName,
        IntegrationDeliveryStatus expectedStatus,
        TimeSpan timeout,
        CancellationToken ct = default)
    {
        var deadline = DateTime.UtcNow + timeout;

        while (DateTime.UtcNow < deadline)
        {
            await using var connection = new NpgsqlConnection(postgresConnectionString);
            await connection.OpenAsync(ct);

            await using var command = new NpgsqlCommand(
                """
                SELECT "RoutingKey", "PayloadSummary", "AttemptCount", "ErrorMessage"
                FROM integration_delivery_logs
                WHERE "EquipmentId" = @equipmentId
                  AND "MaintenanceRegulationId" = @regulationId
                  AND "EventType" = @eventType
                  AND "QueueName" = @queueName
                  AND "Status" = @status
                ORDER BY "UpdatedAt" DESC
                LIMIT 1
                """,
                connection);

            command.Parameters.AddWithValue("equipmentId", equipmentId);
            command.Parameters.AddWithValue("regulationId", maintenanceRegulationId);
            command.Parameters.AddWithValue("eventType", MaxSystems.Application.Integration.IntegrationEventTypes.MaintenanceOverdue);
            command.Parameters.AddWithValue("queueName", queueName);
            command.Parameters.AddWithValue("status", (int)expectedStatus);

            await using var reader = await command.ExecuteReaderAsync(ct);
            if (await reader.ReadAsync(ct))
            {
                return new DeliveryLogDetailRow(
                    reader.GetString(0),
                    reader.IsDBNull(1) ? null : reader.GetString(1),
                    reader.GetInt32(2),
                    reader.IsDBNull(3) ? null : reader.GetString(3));
            }

            await Task.Delay(TimeSpan.FromMilliseconds(250), ct);
        }

        throw new InvalidOperationException(
            $"Maintenance overdue delivery log for equipment {equipmentId}, queue {queueName} did not reach status {expectedStatus} within {timeout}.");
    }

    /// <summary>Последняя созданная работа ТО для оборудования (после generate).</summary>
    public static async Task<Guid?> GetLatestMaintenanceWorkIdForEquipmentAsync(
        string postgresConnectionString,
        Guid equipmentId,
        CancellationToken ct = default)
    {
        await using var connection = new NpgsqlConnection(postgresConnectionString);
        await connection.OpenAsync(ct);

        await using var command = new NpgsqlCommand(
            """
            SELECT "Id"
            FROM works
            WHERE "EquipmentId" = @equipmentId
              AND "Type" = @maintenanceType
            ORDER BY "CreatedAt" DESC
            LIMIT 1
            """,
            connection);

        command.Parameters.AddWithValue("equipmentId", equipmentId);
        command.Parameters.AddWithValue("maintenanceType", WorkType.Maintenance.ToString());

        var result = await command.ExecuteScalarAsync(ct);
        return result is null or DBNull ? null : (Guid)result;
    }

    /// <summary>Строка журнала maintenance-события для assert'ов.</summary>
    internal sealed record MaintenanceDeliveryLogRow(
        string EventType,
        string RoutingKey,
        string QueueName,
        IntegrationDeliveryStatus Status,
        Guid? WorkId,
        Guid? EquipmentId,
        Guid? MaintenanceRegulationId,
        Guid? MessageId);

    /// <summary>Строка журнала доставки для assert'ов.</summary>
    internal sealed record DeliveryLogRow(
        string EventType,
        string RoutingKey,
        string QueueName,
        IntegrationDeliveryStatus Status,
        Guid? WorkId,
        Guid? MessageId);

    /// <summary>Детали строки журнала для E2E assert'ов.</summary>
    internal sealed record DeliveryLogDetailRow(
        string RoutingKey,
        string? PayloadSummary,
        int AttemptCount,
        string? ErrorMessage);

    /// <summary>Идентификатор последней записи журнала по работе, очереди и статусу.</summary>
    public static async Task<Guid?> GetLatestDeliveryLogIdAsync(
        string postgresConnectionString,
        Guid workId,
        string queueName,
        IntegrationDeliveryStatus status,
        CancellationToken ct = default)
    {
        await using var connection = new NpgsqlConnection(postgresConnectionString);
        await connection.OpenAsync(ct);

        await using var command = new NpgsqlCommand(
            """
            SELECT "Id"
            FROM integration_delivery_logs
            WHERE "WorkId" = @workId
              AND "QueueName" = @queueName
              AND "Status" = @status
            ORDER BY "UpdatedAt" DESC
            LIMIT 1
            """,
            connection);

        command.Parameters.AddWithValue("workId", workId);
        command.Parameters.AddWithValue("queueName", queueName);
        command.Parameters.AddWithValue("status", (int)status);

        var result = await command.ExecuteScalarAsync(ct);
        return result is null or DBNull ? null : (Guid)result;
    }

    /// <summary>Обновляет статус записи журнала (для симуляции Failed перед replay-тестом).</summary>
    public static async Task SetDeliveryLogStatusAsync(
        string postgresConnectionString,
        Guid id,
        IntegrationDeliveryStatus status,
        CancellationToken ct = default)
    {
        await using var connection = new NpgsqlConnection(postgresConnectionString);
        await connection.OpenAsync(ct);

        await using var command = new NpgsqlCommand(
            """
            UPDATE integration_delivery_logs
            SET "Status" = @status, "UpdatedAt" = NOW()
            WHERE "Id" = @id
            """,
            connection);

        command.Parameters.AddWithValue("id", id);
        command.Parameters.AddWithValue("status", (int)status);

        await command.ExecuteNonQueryAsync(ct);
    }

    /// <summary>Ждёт запись журнала с указанным MessageId и статусом.</summary>
    public static async Task WaitForDeliveryLogStatusByMessageIdAsync(
        string postgresConnectionString,
        Guid messageId,
        string queueName,
        IntegrationDeliveryStatus expectedStatus,
        TimeSpan timeout,
        CancellationToken ct = default)
    {
        var deadline = DateTime.UtcNow + timeout;

        while (DateTime.UtcNow < deadline)
        {
            await using var connection = new NpgsqlConnection(postgresConnectionString);
            await connection.OpenAsync(ct);

            await using var command = new NpgsqlCommand(
                """
                SELECT 1
                FROM integration_delivery_logs
                WHERE "MessageId" = @messageId
                  AND "QueueName" = @queueName
                  AND "Status" = @status
                LIMIT 1
                """,
                connection);

            command.Parameters.AddWithValue("messageId", messageId);
            command.Parameters.AddWithValue("queueName", queueName);
            command.Parameters.AddWithValue("status", (int)expectedStatus);

            var exists = await command.ExecuteScalarAsync(ct);
            if (exists is not null and not DBNull)
                return;

            await Task.Delay(TimeSpan.FromMilliseconds(250), ct);
        }

        throw new InvalidOperationException(
            $"Delivery log for message {messageId}, queue {queueName} did not reach status {expectedStatus} within {timeout}.");
    }
}