ScadaBridge/src/ZB.MOM.WW.ScadaBridge.DeploymentManager/FlatteningPipeline.cs

using ZB.MOM.WW.ScadaBridge.Commons.Entities.Sites;
using ZB.MOM.WW.ScadaBridge.Commons.Interfaces.Protocol;
using ZB.MOM.WW.ScadaBridge.Commons.Interfaces.Repositories;
using ZB.MOM.WW.ScadaBridge.Commons.Interfaces.Services;
using ZB.MOM.WW.ScadaBridge.Commons.Types;
using ZB.MOM.WW.ScadaBridge.Commons.Types.Flattening;
using ZB.MOM.WW.ScadaBridge.TemplateEngine.Flattening;
using ZB.MOM.WW.ScadaBridge.TemplateEngine.Validation;

namespace ZB.MOM.WW.ScadaBridge.DeploymentManager;

/// <summary>
/// Orchestrates the TemplateEngine services (FlatteningService, ValidationService, RevisionHashService)
/// into a single pipeline for deployment use.
///
/// This captures template state at the time of flatten, ensuring that concurrent template edits
/// (last-write-wins) do not conflict with in-progress deployments.
/// </summary>
public class FlatteningPipeline : IFlatteningPipeline
{
    private readonly ITemplateEngineRepository _templateRepo;
    private readonly ISiteRepository _siteRepo;
    private readonly FlatteningService _flatteningService;
    private readonly ValidationService _validationService;
    private readonly RevisionHashService _revisionHashService;
    private readonly ISharedSchemaRepository _sharedSchemaRepo;
    private readonly ITemplateGraphWatermark _watermark;

    /// <summary>Initializes a new <see cref="FlatteningPipeline"/> with the required template engine and site repositories and services.</summary>
    /// <param name="templateRepo">Repository for loading templates and instance data.</param>
    /// <param name="siteRepo">Repository for loading site data used during validation.</param>
    /// <param name="flatteningService">Service that flattens the template inheritance chain into a resolved config.</param>
    /// <param name="validationService">Service that performs semantic validation on the flattened config.</param>
    /// <param name="revisionHashService">Service that computes the revision hash for staleness detection.</param>
    /// <param name="sharedSchemaRepo">
    /// Repository backing the JSON-Schema <c>$ref</c> resolution seam. Used to
    /// look up <c>lib:Name</c> library references so a dangling reference in any validated
    /// script schema becomes a deploy-blocking error.
    /// </param>
    /// <param name="watermark">
    /// Process-wide template version watermark. Supplies the per-template version
    /// used as the <see cref="FlattenSession"/> cache-key discriminator, so a
    /// memoised chain is dropped the moment any of its templates is edited.
    /// </param>
    public FlatteningPipeline(
        ITemplateEngineRepository templateRepo,
        ISiteRepository siteRepo,
        FlatteningService flatteningService,
        ValidationService validationService,
        RevisionHashService revisionHashService,
        ISharedSchemaRepository sharedSchemaRepo,
        ITemplateGraphWatermark watermark)
    {
        _templateRepo = templateRepo;
        _siteRepo = siteRepo;
        _flatteningService = flatteningService;
        _validationService = validationService;
        _revisionHashService = revisionHashService;
        _sharedSchemaRepo = sharedSchemaRepo;
        _watermark = watermark;
    }

    /// <inheritdoc />
    public FlattenSession CreateSession() => new(_watermark);

    /// <inheritdoc />
    public async Task<Result<FlatteningPipelineResult>> FlattenAndValidateAsync(
        int instanceId,
        CancellationToken cancellationToken = default,
        bool validateScripts = true,
        FlattenSession? session = null)
    {
        // A caller that supplies no session gets a private one. That still pays
        // off inside a single flatten: the composed-chain walk below reaches the
        // same template through several composing parents, and the memo collapses
        // those to one load each.
        session ??= CreateSession();

        // Load instance with full graph
        var instance = await _templateRepo.GetInstanceByIdAsync(instanceId, cancellationToken);
        if (instance == null)
            return Result<FlatteningPipelineResult>.Failure($"Instance with ID {instanceId} not found.");

        var graph = await LoadTemplateGraphAsync(instance.TemplateId, session, cancellationToken);
        if (graph is null)
            return Result<FlatteningPipelineResult>.Failure("Template chain is empty.");

        var (templateChain, compositionMap, composedChains) = graph.Value;

        // Load data connections for the site
        var dataConnections = await session.GetDataConnectionsAsync(
            instance.SiteId,
            siteId => LoadDataConnectionsAsync(siteId, cancellationToken));

        // Flatten
        var flattenResult = _flatteningService.Flatten(
            instance, templateChain, compositionMap, composedChains, dataConnections);

        if (flattenResult.IsFailure)
            return Result<FlatteningPipelineResult>.Failure(flattenResult.Error);

        var config = flattenResult.Value;

        // Load shared scripts for semantic validation. Session-global: identical
        // for every instance in a batch, so hoisted to one query per session
        // (the ArtifactDeploymentService global-hoist shape).
        var resolvedSharedScripts = await session.GetSharedScriptsAsync(async () =>
        {
            var sharedScriptEntities = await _templateRepo.GetAllSharedScriptsAsync(cancellationToken);
            return (IReadOnlyList<ResolvedScript>)sharedScriptEntities.Select(s => new ResolvedScript
            {
                CanonicalName = s.Name,
                Code = s.Code,
                ParameterDefinitions = s.ParameterDefinitions,
                ReturnDefinition = s.ReturnDefinition
            }).ToList();
        });

        // Compute the alarm-capable connection-name set so the semantic validator
        // can gate native-alarm-source bindings. "Alarm-capable" matches the DCL
        // runtime decision (DataConnectionActor: _adapter is IAlarmSubscribableConnection);
        // here we filter connections by alarm-capable protocol, then collect their names.
        //
        // StringComparer.Ordinal is intentional: connection names are stored and
        // matched as authored throughout the pipeline (all other name-keyed
        // dictionaries in FlatteningService and SemanticValidator use the same
        // case-sensitive semantics). OrdinalIgnoreCase would be inconsistent with
        // the rest of the binding-resolution path.
        var alarmCapableConnectionNames = dataConnections.Values
            .Where(c => AlarmCapableProtocols.IsAlarmCapable(c.Protocol))
            .Select(c => c.Name)
            .ToHashSet(StringComparer.Ordinal);

        // The set of data-connection names that actually exist on the
        // target site, used to verify each bound connection resolves to a real site
        // connection. Same StringComparer.Ordinal as the rest of the binding-resolution
        // path (connection names are matched as-authored throughout the pipeline).
        var siteConnectionNames = dataConnections.Values
            .Select(c => c.Name)
            .ToHashSet(StringComparer.Ordinal);

        // Build the JSON-Schema $ref resolution seam from the shared-schema
        // library. The seam ValidationService consumes is synchronous, so the library is
        // pre-loaded once into a name→JSON map here (avoiding sync-over-async) and the
        // seam is a pure in-memory lookup. An unresolved {"$ref":"lib:Name"} in any
        // validated script schema then becomes a deploy-blocking SchemaReference error.
        // Session-global, same as the shared scripts above.
        var schemaLibrary = await session.GetSchemaLibraryAsync(async () =>
        {
            var sharedSchemas = await _sharedSchemaRepo.ListAsync(cancellationToken);
            return (IReadOnlyDictionary<string, string>)sharedSchemas
                .ToDictionary(s => s.Name, s => s.SchemaJson, StringComparer.Ordinal);
        });
        Func<string, string?> resolveSchemaRef = name => schemaLibrary.GetValueOrDefault(name);

        // Validate. This is the deploy-gating path, so connection-binding completeness
        // is enforced as an Error (enforceConnectionBindings: true): a data-sourced
        // attribute with no binding — or one bound to a connection that no longer exists
        // on the site — blocks the deployment. (The template DESIGN-TIME validate path in
        // ManagementActor leaves this non-blocking by NOT enforcing, since bindings are
        // set later at instance/deploy time.)
        // validateScripts == false is the read-only staleness/comparison path: skip
        // only the expensive Roslyn ValidateScriptCompilation (which loads a
        // non-collectible InteractiveAssemblyLoader assembly per script). Structural
        // and semantic validation still run so a genuinely broken flatten is surfaced;
        // the deploy path (validateScripts: true) remains the authoritative compile gate.
        var validation = _validationService.Validate(
            config,
            resolvedSharedScripts,
            alarmCapableConnectionNames,
            enforceConnectionBindings: true,
            siteConnectionNames: siteConnectionNames,
            resolveSchemaRef: resolveSchemaRef,
            validateScriptCompilation: validateScripts);

        // Compute revision hash
        var hash = _revisionHashService.ComputeHash(config);

        return Result<FlatteningPipelineResult>.Success(
            new FlatteningPipelineResult(config, hash, validation));
    }

    /// <inheritdoc />
    public async Task<Result<FlattenedConfiguration>> PreviewAsync(
        int templateId,
        int siteId,
        CancellationToken cancellationToken = default)
    {
        var session = CreateSession();

        var graph = await LoadTemplateGraphAsync(templateId, session, cancellationToken);
        if (graph is null)
            return Result<FlattenedConfiguration>.Failure($"Template {templateId} not found or its chain is empty.");

        var dataConnections = await session.GetDataConnectionsAsync(
            siteId,
            id => LoadDataConnectionsAsync(id, cancellationToken));

        // A stand-in instance with no overrides and no bindings: the flattener sees
        // exactly what a freshly created instance would, and nothing is persisted.
        var preview = new Commons.Entities.Instances.Instance("(preview)")
        {
            TemplateId = templateId,
            SiteId = siteId
        };

        return _flatteningService.Flatten(
            preview, graph.Value.Chain, graph.Value.CompositionMap, graph.Value.ComposedChains, dataConnections);
    }

    /// <summary>
    /// Loads the template inheritance chain for <paramref name="templateId"/> plus the
    /// composition maps, walking nested compositions so the flattener can resolve
    /// composed-of-composed members (e.g. a parent composing Pump where Pump itself
    /// composes AlarmSensor produces "Pump.AlarmSensor.SensorReading").
    /// Returns <see langword="null"/> when the chain is empty (unknown template).
    /// </summary>
    private async Task<(IReadOnlyList<Commons.Entities.Templates.Template> Chain,
                        Dictionary<int, IReadOnlyList<Commons.Entities.Templates.TemplateComposition>> CompositionMap,
                        Dictionary<int, IReadOnlyList<Commons.Entities.Templates.Template>> ComposedChains)?>
        LoadTemplateGraphAsync(int templateId, FlattenSession session, CancellationToken cancellationToken)
    {
        Task<Commons.Entities.Templates.Template?> LoadTemplate(int id) =>
            _templateRepo.GetTemplateWithChildrenAsync(id, cancellationToken);

        var templateChain = await session.GetChainAsync(templateId, LoadTemplate);
        if (templateChain.Count == 0)
            return null;

        var compositionMap = new Dictionary<int, IReadOnlyList<Commons.Entities.Templates.TemplateComposition>>();
        var composedChains = new Dictionary<int, IReadOnlyList<Commons.Entities.Templates.Template>>();
        var processedTemplateIds = new HashSet<int>();
        var pendingChains = new Queue<IReadOnlyList<Commons.Entities.Templates.Template>>();

        pendingChains.Enqueue(templateChain);
        while (pendingChains.Count > 0)
        {
            var chain = pendingChains.Dequeue();
            foreach (var template in chain)
            {
                if (!processedTemplateIds.Add(template.Id)) continue;

                var compositions = await session.GetCompositionsAsync(
                    template.Id,
                    id => _templateRepo.GetCompositionsByTemplateIdAsync(id, cancellationToken));
                if (compositions.Count == 0) continue;

                compositionMap[template.Id] = compositions;
                foreach (var comp in compositions)
                {
                    if (composedChains.ContainsKey(comp.ComposedTemplateId)) continue;

                    var composedChain = await session.GetChainAsync(comp.ComposedTemplateId, LoadTemplate);
                    composedChains[comp.ComposedTemplateId] = composedChain;
                    pendingChains.Enqueue(composedChain);
                }
            }
        }

        return (templateChain, compositionMap, composedChains);
    }

    private async Task<IReadOnlyDictionary<int, DataConnection>> LoadDataConnectionsAsync(
        int siteId,
        CancellationToken cancellationToken)
    {
        var connections = await _siteRepo.GetDataConnectionsBySiteIdAsync(siteId, cancellationToken);
        return connections.ToDictionary(c => c.Id);
    }
}