<commit_before>﻿using System;
using System.Collections.Generic;
using System.Linq;
using System.Transactions;

using NuClear.AdvancedSearch.Common.Metadata.Identities;
using NuClear.AdvancedSearch.Common.Metadata.Model.Operations;
using NuClear.Metamodeling.Elements;
using NuClear.Metamodeling.Elements.Identities.Builder;
using NuClear.Metamodeling.Provider;
using NuClear.Replication.Core.API.Aggregates;
using NuClear.Telemetry.Probing;

namespace NuClear.Replication.Core.Aggregates
{
    public sealed class AggregatesConstructor : IAggregatesConstructor
    {
        private readonly IMetadataProvider _metadataProvider;
        private readonly IAggregateProcessorFactory _aggregateProcessorFactory;

        public AggregatesConstructor(IMetadataProvider metadataProvider, IAggregateProcessorFactory aggregateProcessorFactory)
        {
            _metadataProvider = metadataProvider;
            _aggregateProcessorFactory = aggregateProcessorFactory;
        }

        public void Construct(IEnumerable<AggregateOperation> operations)
        {
            using (Probe.Create("ETL2 Transforming"))
            {
                var slices = operations.GroupBy(x => new { Operation = x.GetType(), x.AggregateType })
                                       .OrderByDescending(x => x.Key.Operation, new AggregateOperationPriorityComparer());

                foreach (var slice in slices)
                {
                    var operation = slice.Key.Operation;
                    var aggregateType = slice.Key.AggregateType;

                    IMetadataElement aggregateMetadata;
                    var metadataId = ReplicationMetadataIdentity.Instance.Id.WithRelative(new Uri(string.Format("Aggregates/{0}", aggregateType.Name), UriKind.Relative));
                    if (!_metadataProvider.TryGetMetadata(metadataId, out aggregateMetadata))
                    {
                        throw new NotSupportedException(string.Format("The aggregate of type '{0}' is not supported.", aggregateType));
                    }

                    var aggregateIds = slice.Select(x => x.AggregateId).Distinct().ToArray();
                    using (var transaction = new TransactionScope(TransactionScopeOption.Required,
                                                                  new TransactionOptions { IsolationLevel = IsolationLevel.ReadCommitted, Timeout = TimeSpan.Zero }))
                    {
                        using (Probe.Create("ETL2 Transforming", aggregateType.Name))
                        {
                            var processor = _aggregateProcessorFactory.Create(aggregateMetadata);

                            if (operation == typeof(InitializeAggregate))
                            {
                                processor.Initialize(aggregateIds);
                                continue;
                            }

                            if (operation == typeof(RecalculateAggregate))
                            {
                                processor.Recalculate(aggregateIds);
                                continue;
                            }

                            if (operation == typeof(DestroyAggregate))
                            {
                                processor.Destroy(aggregateIds);
                                continue;
                            }
                        }

                        transaction.Complete();
                    }
                }
            }
        }
    }
}

<commit_msg>Исправил баг с незакоммиченной транзакцией<commit_after>﻿using System;
using System.Collections.Generic;
using System.Linq;
using System.Transactions;

using NuClear.AdvancedSearch.Common.Metadata.Identities;
using NuClear.AdvancedSearch.Common.Metadata.Model.Operations;
using NuClear.Metamodeling.Elements;
using NuClear.Metamodeling.Elements.Identities.Builder;
using NuClear.Metamodeling.Provider;
using NuClear.Replication.Core.API.Aggregates;
using NuClear.Telemetry.Probing;

namespace NuClear.Replication.Core.Aggregates
{
    public sealed class AggregatesConstructor : IAggregatesConstructor
    {
        private readonly IMetadataProvider _metadataProvider;
        private readonly IAggregateProcessorFactory _aggregateProcessorFactory;

        public AggregatesConstructor(IMetadataProvider metadataProvider, IAggregateProcessorFactory aggregateProcessorFactory)
        {
            _metadataProvider = metadataProvider;
            _aggregateProcessorFactory = aggregateProcessorFactory;
        }

        public void Construct(IEnumerable<AggregateOperation> operations)
        {
            using (Probe.Create("ETL2 Transforming"))
            {
                var slices = operations.GroupBy(x => new { Operation = x.GetType(), x.AggregateType })
                                       .OrderByDescending(x => x.Key.Operation, new AggregateOperationPriorityComparer());

                foreach (var slice in slices)
                {
                    var operation = slice.Key.Operation;
                    var aggregateType = slice.Key.AggregateType;

                    IMetadataElement aggregateMetadata;
                    var metadataId = ReplicationMetadataIdentity.Instance.Id.WithRelative(new Uri(string.Format("Aggregates/{0}", aggregateType.Name), UriKind.Relative));
                    if (!_metadataProvider.TryGetMetadata(metadataId, out aggregateMetadata))
                    {
                        throw new NotSupportedException(string.Format("The aggregate of type '{0}' is not supported.", aggregateType));
                    }

                    var aggregateIds = slice.Select(x => x.AggregateId).Distinct().ToArray();
                    using (var transaction = new TransactionScope(TransactionScopeOption.Required,
                                                                  new TransactionOptions { IsolationLevel = IsolationLevel.ReadCommitted, Timeout = TimeSpan.Zero }))
                    {
                        using (Probe.Create("ETL2 Transforming", aggregateType.Name))
                        {
                            var processor = _aggregateProcessorFactory.Create(aggregateMetadata);

                            if (operation == typeof(InitializeAggregate))
                            {
                                processor.Initialize(aggregateIds);
                            }

                            if (operation == typeof(RecalculateAggregate))
                            {
                                processor.Recalculate(aggregateIds);
                            }

                            if (operation == typeof(DestroyAggregate))
                            {
                                processor.Destroy(aggregateIds);
                            }
                        }

                        transaction.Complete();
                    }
                }
            }
        }
    }
}


{'fraction_non_alphanumeric': 0.07059669302659957, 'fraction_numerical': 0.0011502516175413372, 'mean_word_length': 1.6150375939849624, 'pattern_counts': {'":': 0, '<': 5, '<?xml version=': 0, '>': 11, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 8, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '4043679', 'n_tokens_mistral': 1537, 'n_tokens_neox': 1482, 'n_words': 273}