lmxopcua/src/ZB.MOM.WW.OtOpcUa.Host/GalaxyRepository/GalaxyRepositoryService.cs

using System;
using System.Collections.Generic;
using System.Data.SqlClient;
using System.Linq;
using System.Threading;
using System.Threading.Tasks;
using Serilog;
using ZB.MOM.WW.OtOpcUa.Host.Configuration;
using ZB.MOM.WW.OtOpcUa.Host.Domain;

namespace ZB.MOM.WW.OtOpcUa.Host.GalaxyRepository
{
    /// <summary>
    ///     Implements IGalaxyRepository using SQL queries against the Galaxy ZB database. (GR-001 through GR-007)
    /// </summary>
    public class GalaxyRepositoryService : IGalaxyRepository
    {
        private static readonly ILogger Log = Serilog.Log.ForContext<GalaxyRepositoryService>();

        private readonly GalaxyRepositoryConfiguration _config;

        /// <summary>
        ///     When <see cref="Configuration.GalaxyScope.LocalPlatform" /> filtering is active, caches the set of
        ///     gobject_ids that passed the hierarchy filter so <see cref="GetAttributesAsync" /> can apply the same scope.
        ///     Populated by <see cref="GetHierarchyAsync" /> and consumed by <see cref="GetAttributesAsync" />.
        /// </summary>
        private HashSet<int>? _scopeFilteredGobjectIds;

        /// <summary>
        ///     Initializes a new repository service that reads Galaxy metadata from the configured SQL database.
        /// </summary>
        /// <param name="config">The repository connection, timeout, and attribute-selection settings.</param>
        public GalaxyRepositoryService(GalaxyRepositoryConfiguration config)
        {
            _config = config;
        }

        /// <summary>
        ///     Occurs when the repository detects a Galaxy deploy change that should trigger an address-space rebuild.
        /// </summary>
        public event Action? OnGalaxyChanged;

        /// <summary>
        ///     Queries the Galaxy repository for the deployed object hierarchy that becomes the OPC UA browse tree.
        /// </summary>
        /// <param name="ct">A token that cancels the database query.</param>
        /// <returns>The deployed Galaxy objects that should appear in the namespace.</returns>
        public async Task<List<GalaxyObjectInfo>> GetHierarchyAsync(CancellationToken ct = default)
        {
            var results = new List<GalaxyObjectInfo>();

            using var conn = new SqlConnection(_config.ConnectionString);
            await conn.OpenAsync(ct);

            using var cmd = new SqlCommand(HierarchySql, conn) { CommandTimeout = _config.CommandTimeoutSeconds };
            using var reader = await cmd.ExecuteReaderAsync(ct);

            while (await reader.ReadAsync(ct))
            {
                var templateChainRaw = reader.IsDBNull(8) ? "" : reader.GetString(8);
                var templateChain = string.IsNullOrEmpty(templateChainRaw)
                    ? new List<string>()
                    : templateChainRaw.Split(new[] { '|' }, StringSplitOptions.RemoveEmptyEntries)
                        .Select(s => s.Trim())
                        .Where(s => s.Length > 0)
                        .ToList();

                results.Add(new GalaxyObjectInfo
                {
                    GobjectId = Convert.ToInt32(reader.GetValue(0)),
                    TagName = reader.GetString(1),
                    ContainedName = reader.IsDBNull(2) ? "" : reader.GetString(2),
                    BrowseName = reader.GetString(3),
                    ParentGobjectId = Convert.ToInt32(reader.GetValue(4)),
                    IsArea = Convert.ToInt32(reader.GetValue(5)) == 1,
                    CategoryId = Convert.ToInt32(reader.GetValue(6)),
                    HostedByGobjectId = Convert.ToInt32(reader.GetValue(7)),
                    TemplateChain = templateChain
                });
            }

            if (results.Count == 0)
                Log.Warning("GetHierarchyAsync returned zero rows");
            else
                Log.Information("GetHierarchyAsync returned {Count} objects", results.Count);

            if (_config.Scope == GalaxyScope.LocalPlatform)
            {
                var platforms = await GetPlatformsAsync(ct);
                var platformName = string.IsNullOrWhiteSpace(_config.PlatformName)
                    ? Environment.MachineName
                    : _config.PlatformName;
                var (filtered, gobjectIds) = PlatformScopeFilter.Filter(results, platforms, platformName);
                _scopeFilteredGobjectIds = gobjectIds;
                return filtered;
            }

            _scopeFilteredGobjectIds = null;
            return results;
        }

        /// <summary>
        ///     Queries the Galaxy repository for attribute metadata that becomes OPC UA variable nodes.
        /// </summary>
        /// <param name="ct">A token that cancels the database query.</param>
        /// <returns>The attribute rows required to build runtime tag mappings and variable metadata.</returns>
        public async Task<List<GalaxyAttributeInfo>> GetAttributesAsync(CancellationToken ct = default)
        {
            var results = new List<GalaxyAttributeInfo>();
            var extended = _config.ExtendedAttributes;
            var sql = extended ? ExtendedAttributesSql : AttributesSql;

            using var conn = new SqlConnection(_config.ConnectionString);
            await conn.OpenAsync(ct);

            using var cmd = new SqlCommand(sql, conn) { CommandTimeout = _config.CommandTimeoutSeconds };
            using var reader = await cmd.ExecuteReaderAsync(ct);

            while (await reader.ReadAsync(ct))
                results.Add(extended ? ReadExtendedAttribute(reader) : ReadStandardAttribute(reader));

            Log.Information("GetAttributesAsync returned {Count} attributes (extended={Extended})", results.Count,
                extended);

            if (_config.Scope == GalaxyScope.LocalPlatform && _scopeFilteredGobjectIds != null)
                return PlatformScopeFilter.FilterAttributes(results, _scopeFilteredGobjectIds);

            return results;
        }

        /// <summary>
        ///     Reads the latest Galaxy deploy timestamp so change detection can decide whether the address space is stale.
        /// </summary>
        /// <param name="ct">A token that cancels the database query.</param>
        /// <returns>The most recent deploy timestamp, or <see langword="null" /> when none is available.</returns>
        public async Task<DateTime?> GetLastDeployTimeAsync(CancellationToken ct = default)
        {
            using var conn = new SqlConnection(_config.ConnectionString);
            await conn.OpenAsync(ct);

            using var cmd = new SqlCommand(ChangeDetectionSql, conn) { CommandTimeout = _config.CommandTimeoutSeconds };
            var result = await cmd.ExecuteScalarAsync(ct);

            return result is DateTime dt ? dt : null;
        }

        /// <summary>
        ///     Executes a lightweight query to confirm that the repository database is reachable.
        /// </summary>
        /// <param name="ct">A token that cancels the connectivity check.</param>
        /// <returns><see langword="true" /> when the query succeeds; otherwise, <see langword="false" />.</returns>
        public async Task<bool> TestConnectionAsync(CancellationToken ct = default)
        {
            try
            {
                using var conn = new SqlConnection(_config.ConnectionString);
                await conn.OpenAsync(ct);

                using var cmd = new SqlCommand(TestConnectionSql, conn)
                    { CommandTimeout = _config.CommandTimeoutSeconds };
                await cmd.ExecuteScalarAsync(ct);

                Log.Information("Galaxy repository database connection successful");
                return true;
            }
            catch (Exception ex)
            {
                Log.Warning(ex, "Galaxy repository database connection failed");
                return false;
            }
        }

        /// <summary>
        ///     Queries the platform table for deployed platform-to-hostname mappings used by
        ///     <see cref="Configuration.GalaxyScope.LocalPlatform" /> filtering.
        /// </summary>
        private async Task<List<PlatformInfo>> GetPlatformsAsync(CancellationToken ct = default)
        {
            var results = new List<PlatformInfo>();

            using var conn = new SqlConnection(_config.ConnectionString);
            await conn.OpenAsync(ct);

            using var cmd = new SqlCommand(PlatformLookupSql, conn) { CommandTimeout = _config.CommandTimeoutSeconds };
            using var reader = await cmd.ExecuteReaderAsync(ct);

            while (await reader.ReadAsync(ct))
            {
                results.Add(new PlatformInfo
                {
                    GobjectId = Convert.ToInt32(reader.GetValue(0)),
                    NodeName = reader.IsDBNull(1) ? "" : reader.GetString(1)
                });
            }

            Log.Information("GetPlatformsAsync returned {Count} platform(s)", results.Count);
            return results;
        }

        /// <summary>
        ///     Reads a row from the standard attributes query (12 columns).
        ///     Columns: gobject_id, tag_name, attribute_name, full_tag_reference, mx_data_type,
        ///     data_type_name, is_array, array_dimension, mx_attribute_category,
        ///     security_classification, is_historized, is_alarm
        /// </summary>
        private static GalaxyAttributeInfo ReadStandardAttribute(SqlDataReader reader)
        {
            return new GalaxyAttributeInfo
            {
                GobjectId = Convert.ToInt32(reader.GetValue(0)),
                TagName = reader.GetString(1),
                AttributeName = reader.GetString(2),
                FullTagReference = reader.GetString(3),
                MxDataType = Convert.ToInt32(reader.GetValue(4)),
                DataTypeName = reader.IsDBNull(5) ? "" : reader.GetString(5),
                IsArray = Convert.ToBoolean(reader.GetValue(6)),
                ArrayDimension = reader.IsDBNull(7) ? null : Convert.ToInt32(reader.GetValue(7)),
                SecurityClassification = Convert.ToInt32(reader.GetValue(9)),
                IsHistorized = Convert.ToInt32(reader.GetValue(10)) == 1,
                IsAlarm = Convert.ToInt32(reader.GetValue(11)) == 1
            };
        }

        /// <summary>
        ///     Reads a row from the extended attributes query (14 columns).
        ///     Columns: gobject_id, tag_name, primitive_name, attribute_name, full_tag_reference,
        ///     mx_data_type, data_type_name, is_array, array_dimension,
        ///     mx_attribute_category, security_classification, is_historized, is_alarm, attribute_source
        /// </summary>
        private static GalaxyAttributeInfo ReadExtendedAttribute(SqlDataReader reader)
        {
            return new GalaxyAttributeInfo
            {
                GobjectId = Convert.ToInt32(reader.GetValue(0)),
                TagName = reader.GetString(1),
                PrimitiveName = reader.IsDBNull(2) ? "" : reader.GetString(2),
                AttributeName = reader.GetString(3),
                FullTagReference = reader.GetString(4),
                MxDataType = Convert.ToInt32(reader.GetValue(5)),
                DataTypeName = reader.IsDBNull(6) ? "" : reader.GetString(6),
                IsArray = Convert.ToBoolean(reader.GetValue(7)),
                ArrayDimension = reader.IsDBNull(8) ? null : Convert.ToInt32(reader.GetValue(8)),
                SecurityClassification = Convert.ToInt32(reader.GetValue(10)),
                IsHistorized = Convert.ToInt32(reader.GetValue(11)) == 1,
                IsAlarm = Convert.ToInt32(reader.GetValue(12)) == 1,
                AttributeSource = reader.IsDBNull(13) ? "" : reader.GetString(13)
            };
        }

        /// <summary>
        ///     Raises the change event used by tests and monitoring components to simulate or announce a Galaxy deploy.
        /// </summary>
        public void RaiseGalaxyChanged()
        {
            OnGalaxyChanged?.Invoke();
        }

        #region SQL Queries (GR-006: const string, no dynamic SQL)

        private const string HierarchySql = @"
;WITH template_chain AS (
    SELECT g.gobject_id AS instance_gobject_id, t.gobject_id AS template_gobject_id,
           t.tag_name AS template_tag_name, t.derived_from_gobject_id, 0 AS depth
    FROM gobject g
    INNER JOIN gobject t ON t.gobject_id = g.derived_from_gobject_id
    WHERE g.is_template = 0 AND g.deployed_package_id <> 0 AND g.derived_from_gobject_id <> 0
    UNION ALL
    SELECT tc.instance_gobject_id, t.gobject_id, t.tag_name, t.derived_from_gobject_id, tc.depth + 1
    FROM template_chain tc
    INNER JOIN gobject t ON t.gobject_id = tc.derived_from_gobject_id
    WHERE tc.derived_from_gobject_id <> 0 AND tc.depth < 10
)
SELECT DISTINCT
    g.gobject_id,
    g.tag_name,
    g.contained_name,
    CASE WHEN g.contained_name IS NULL OR g.contained_name = ''
         THEN g.tag_name
         ELSE g.contained_name
    END AS browse_name,
    CASE WHEN g.contained_by_gobject_id = 0
         THEN g.area_gobject_id
         ELSE g.contained_by_gobject_id
    END AS parent_gobject_id,
    CASE WHEN td.category_id = 13
         THEN 1
         ELSE 0
    END AS is_area,
    td.category_id AS category_id,
    g.hosted_by_gobject_id AS hosted_by_gobject_id,
    ISNULL(
        STUFF((
            SELECT '|' + tc.template_tag_name
            FROM template_chain tc
            WHERE tc.instance_gobject_id = g.gobject_id
            ORDER BY tc.depth
            FOR XML PATH('')
        ), 1, 1, ''),
        ''
    ) AS template_chain
FROM gobject g
INNER JOIN template_definition td
    ON g.template_definition_id = td.template_definition_id
WHERE td.category_id IN (1, 3, 4, 10, 11, 13, 17, 24, 26)
    AND g.is_template = 0
    AND g.deployed_package_id <> 0
ORDER BY parent_gobject_id, g.tag_name";

        private const string AttributesSql = @"
;WITH deployed_package_chain AS (
    SELECT g.gobject_id, p.package_id, p.derived_from_package_id, 0 AS depth
    FROM gobject g
    INNER JOIN package p ON p.package_id = g.deployed_package_id
    WHERE g.is_template = 0 AND g.deployed_package_id <> 0
    UNION ALL
    SELECT dpc.gobject_id, p.package_id, p.derived_from_package_id, dpc.depth + 1
    FROM deployed_package_chain dpc
    INNER JOIN package p ON p.package_id = dpc.derived_from_package_id
    WHERE dpc.derived_from_package_id <> 0 AND dpc.depth < 10
)
SELECT gobject_id, tag_name, attribute_name, full_tag_reference,
    mx_data_type, data_type_name, is_array, array_dimension,
    mx_attribute_category, security_classification, is_historized, is_alarm
FROM (
    SELECT
        dpc.gobject_id,
        g.tag_name,
        da.attribute_name,
        g.tag_name + '.' + da.attribute_name
            + CASE WHEN da.is_array = 1 THEN '[]' ELSE '' END
            AS full_tag_reference,
        da.mx_data_type,
        dt.description AS data_type_name,
        da.is_array,
        CASE WHEN da.is_array = 1
             THEN CONVERT(int, CONVERT(varbinary(2),
                  SUBSTRING(da.mx_value, 15, 2) + SUBSTRING(da.mx_value, 13, 2), 2))
             ELSE NULL
        END AS array_dimension,
        da.mx_attribute_category,
        da.security_classification,
        CASE WHEN EXISTS (
            SELECT 1 FROM deployed_package_chain dpc2
            INNER JOIN primitive_instance pi ON pi.package_id = dpc2.package_id AND pi.primitive_name = da.attribute_name
            INNER JOIN primitive_definition pd ON pd.primitive_definition_id = pi.primitive_definition_id AND pd.primitive_name = 'HistoryExtension'
            WHERE dpc2.gobject_id = dpc.gobject_id
        ) THEN 1 ELSE 0 END AS is_historized,
        CASE WHEN EXISTS (
            SELECT 1 FROM deployed_package_chain dpc2
            INNER JOIN primitive_instance pi ON pi.package_id = dpc2.package_id AND pi.primitive_name = da.attribute_name
            INNER JOIN primitive_definition pd ON pd.primitive_definition_id = pi.primitive_definition_id AND pd.primitive_name = 'AlarmExtension'
            WHERE dpc2.gobject_id = dpc.gobject_id
        ) THEN 1 ELSE 0 END AS is_alarm,
        ROW_NUMBER() OVER (
            PARTITION BY dpc.gobject_id, da.attribute_name
            ORDER BY dpc.depth
        ) AS rn
    FROM deployed_package_chain dpc
    INNER JOIN dynamic_attribute da
        ON da.package_id = dpc.package_id
    INNER JOIN gobject g
        ON g.gobject_id = dpc.gobject_id
    INNER JOIN template_definition td
        ON td.template_definition_id = g.template_definition_id
    LEFT JOIN data_type dt
        ON dt.mx_data_type = da.mx_data_type
    WHERE td.category_id IN (1, 3, 4, 10, 11, 13, 17, 24, 26)
        AND da.attribute_name NOT LIKE '[_]%'
        AND da.attribute_name NOT LIKE '%.Description'
        AND da.mx_attribute_category IN (2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 24)
) ranked
WHERE rn = 1
ORDER BY tag_name, attribute_name";

        private const string ExtendedAttributesSql = @"
;WITH deployed_package_chain AS (
    SELECT g.gobject_id, p.package_id, p.derived_from_package_id, 0 AS depth
    FROM gobject g
    INNER JOIN package p ON p.package_id = g.deployed_package_id
    WHERE g.is_template = 0 AND g.deployed_package_id <> 0
    UNION ALL
    SELECT dpc.gobject_id, p.package_id, p.derived_from_package_id, dpc.depth + 1
    FROM deployed_package_chain dpc
    INNER JOIN package p ON p.package_id = dpc.derived_from_package_id
    WHERE dpc.derived_from_package_id <> 0 AND dpc.depth < 10
),
ranked_dynamic AS (
    SELECT
        dpc.gobject_id,
        g.tag_name,
        da.attribute_name,
        g.tag_name + '.' + da.attribute_name
            + CASE WHEN da.is_array = 1 THEN '[]' ELSE '' END
            AS full_tag_reference,
        da.mx_data_type,
        dt.description AS data_type_name,
        da.is_array,
        CASE WHEN da.is_array = 1
             THEN CONVERT(int, CONVERT(varbinary(2),
                  SUBSTRING(da.mx_value, 15, 2) + SUBSTRING(da.mx_value, 13, 2), 2))
             ELSE NULL
        END AS array_dimension,
        da.mx_attribute_category,
        da.security_classification,
        CASE WHEN EXISTS (
            SELECT 1 FROM deployed_package_chain dpc2
            INNER JOIN primitive_instance pi ON pi.package_id = dpc2.package_id AND pi.primitive_name = da.attribute_name
            INNER JOIN primitive_definition pd ON pd.primitive_definition_id = pi.primitive_definition_id AND pd.primitive_name = 'HistoryExtension'
            WHERE dpc2.gobject_id = dpc.gobject_id
        ) THEN 1 ELSE 0 END AS is_historized,
        CASE WHEN EXISTS (
            SELECT 1 FROM deployed_package_chain dpc2
            INNER JOIN primitive_instance pi ON pi.package_id = dpc2.package_id AND pi.primitive_name = da.attribute_name
            INNER JOIN primitive_definition pd ON pd.primitive_definition_id = pi.primitive_definition_id AND pd.primitive_name = 'AlarmExtension'
            WHERE dpc2.gobject_id = dpc.gobject_id
        ) THEN 1 ELSE 0 END AS is_alarm,
        ROW_NUMBER() OVER (
            PARTITION BY dpc.gobject_id, da.attribute_name
            ORDER BY dpc.depth
        ) AS rn
    FROM deployed_package_chain dpc
    INNER JOIN dynamic_attribute da
        ON da.package_id = dpc.package_id
    INNER JOIN gobject g
        ON g.gobject_id = dpc.gobject_id
    INNER JOIN template_definition td
        ON td.template_definition_id = g.template_definition_id
    LEFT JOIN data_type dt
        ON dt.mx_data_type = da.mx_data_type
    WHERE td.category_id IN (1, 3, 4, 10, 11, 13, 17, 24, 26)
        AND da.attribute_name NOT LIKE '[_]%'
        AND da.attribute_name NOT LIKE '%.Description'
        AND da.mx_attribute_category IN (2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 24)
)
SELECT
    gobject_id,
    tag_name,
    primitive_name,
    attribute_name,
    full_tag_reference,
    mx_data_type,
    data_type_name,
    is_array,
    array_dimension,
    mx_attribute_category,
    security_classification,
    is_historized,
    is_alarm,
    attribute_source
FROM (
    SELECT
        g.gobject_id,
        g.tag_name,
        pi.primitive_name,
        ad.attribute_name,
        CASE WHEN pi.primitive_name = ''
             THEN g.tag_name + '.' + ad.attribute_name
             ELSE g.tag_name + '.' + pi.primitive_name + '.' + ad.attribute_name
        END + CASE WHEN ad.is_array = 1 THEN '[]' ELSE '' END
            AS full_tag_reference,
        ad.mx_data_type,
        dt.description AS data_type_name,
        ad.is_array,
        CASE WHEN ad.is_array = 1
             THEN CONVERT(int, CONVERT(varbinary(2),
                  SUBSTRING(ad.mx_value, 15, 2) + SUBSTRING(ad.mx_value, 13, 2), 2))
             ELSE NULL
        END AS array_dimension,
        ad.mx_attribute_category,
        ad.security_classification,
        CAST(0 AS int) AS is_historized,
        CAST(0 AS int) AS is_alarm,
        'primitive' AS attribute_source
    FROM gobject g
    INNER JOIN instance i
        ON i.gobject_id = g.gobject_id
    INNER JOIN template_definition td
        ON td.template_definition_id = g.template_definition_id
        AND td.runtime_clsid <> '{00000000-0000-0000-0000-000000000000}'
    INNER JOIN package p
        ON p.package_id = g.deployed_package_id
    INNER JOIN primitive_instance pi
        ON pi.package_id = p.package_id
        AND pi.property_bitmask & 0x10 <> 0x10
    INNER JOIN attribute_definition ad
        ON ad.primitive_definition_id = pi.primitive_definition_id
        AND ad.attribute_name NOT LIKE '[_]%'
        AND ad.mx_attribute_category IN (2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 24)
    LEFT JOIN data_type dt
        ON dt.mx_data_type = ad.mx_data_type
    WHERE td.category_id IN (1, 3, 4, 10, 11, 13, 17, 24, 26)
        AND g.is_template = 0
        AND g.deployed_package_id <> 0

    UNION ALL

    SELECT
        gobject_id,
        tag_name,
        '' AS primitive_name,
        attribute_name,
        full_tag_reference,
        mx_data_type,
        data_type_name,
        is_array,
        array_dimension,
        mx_attribute_category,
        security_classification,
        is_historized,
        is_alarm,
        'dynamic' AS attribute_source
    FROM ranked_dynamic
    WHERE rn = 1
) all_attributes
ORDER BY tag_name, primitive_name, attribute_name";

        private const string PlatformLookupSql = @"
SELECT p.platform_gobject_id, p.node_name
FROM platform p
INNER JOIN gobject g ON g.gobject_id = p.platform_gobject_id
WHERE g.is_template = 0 AND g.deployed_package_id <> 0";

        private const string ChangeDetectionSql = "SELECT time_of_last_deploy FROM galaxy";

        private const string TestConnectionSql = "SELECT 1";

        #endregion
    }
}