<commit_before>﻿using System;
using System.Collections.Generic;
using System.Linq;
using NEventStore.Domain;
using NEventStore.Domain.Persistence;
using NEventStore;
using NEventStore.Persistence;
using System.Collections.Concurrent;
using System.Threading;
using System.Diagnostics;
using Metrics;
using Jarvis.NEventStoreEx.Support;

namespace Jarvis.NEventStoreEx.CommonDomainEx.Persistence.EventStore
{
    public class AbstractRepository : IRepositoryEx
    {
        private const string AggregateTypeHeader = "AggregateType";

        private static readonly Counter RepositoryLoadCounter = Metric.Counter("RepositoryGetById", Unit.Custom("ticks"));
        private static readonly Counter FailedLockCounter = Metric.Counter("AbstractRepositoryFailedLocks", Unit.Calls);

        private readonly IDetectConflicts _conflictDetector;

        private readonly IStoreEvents _eventStore;

        private readonly IConstructAggregatesEx _factory;
        private readonly IIdentityConverter _identityConverter;

        private readonly IDictionary<string, IEventStream> _streams = new Dictionary<string, IEventStream>();

        /// <summary>
        /// This is used to keep a better track of locking, because each unique instance of repository
        /// will have its unique index, and each lock is marked with that index.
        /// </summary>
        private readonly Int64 _identity;

        /// <summary>
        /// This is the last identity assigned.
        /// </summary>
        private static Int64 _lastAssignedIdentity;

        /// <summary>
        /// This hashset contains the identities this instance of repository had obtained an exclusive lock to. 
        /// The real instance that holds all locked entities is <see cref="IdSerializerDictionary"/> property.
        /// It is necessary to remove all the lock after this instance is disposed.
        /// </summary>
        private readonly HashSet<IIdentity> _identitiesLocked = new HashSet<IIdentity>();

        /// <summary>
        /// This dictionary is used to hold a list of the aggregate ids that are acutally locked.
        /// It is used to avoid two threads to access at the very same moment the same entity
        /// thus avoiding unnecessary ConcurrencyException.
        /// </summary>
        private static readonly ConcurrentDictionary<IIdentity, Int64> IdSerializerDictionary;

        public NEventStore.Logging.ILog _logger;

        static AbstractRepository()
        {
            IdSerializerDictionary = new ConcurrentDictionary<IIdentity, Int64>();
            _lastAssignedIdentity = 0;
        }

        protected AbstractRepository(
            IStoreEvents eventStore, 
            IConstructAggregatesEx factory, 
            IDetectConflicts conflictDetector, 
            IIdentityConverter identityConverter,
            NEventStore.Logging.ILog logger)
        {
            this._eventStore = eventStore;
            this._factory = factory;
            this._conflictDetector = conflictDetector;
            this._identityConverter = identityConverter;
            this._identity = Interlocked.Increment(ref _lastAssignedIdentity);
            _logger = logger;
        }

        public void Dispose()
        {
            this.Dispose(true);
            GC.SuppressFinalize(this);
        }

        public virtual TAggregate GetById<TAggregate>(IIdentity id) where TAggregate : class, IAggregateEx
        {
            return this.GetById<TAggregate>(Bucket.Default, id);
        }

        public virtual TAggregate GetById<TAggregate>(IIdentity id, int versionToLoad) where TAggregate : class, IAggregateEx
        {
            return this.GetById<TAggregate>(Bucket.Default, id, versionToLoad);
        }

        public TAggregate GetById<TAggregate>(string bucketId, IIdentity id) where TAggregate : class, IAggregateEx
        {
            return this.GetById<TAggregate>(bucketId, id, int.MaxValue);
        }

        public TAggregate GetById<TAggregate>(string bucketId, IIdentity id, int versionToLoad) where TAggregate : class, IAggregateEx
        {
            Stopwatch sw = null;
            if (NeventStoreExGlobalConfiguration.MetricsEnabled) sw = Stopwatch.StartNew();

            if (NeventStoreExGlobalConfiguration.RepositoryLockOnAggregateId) 
            {
                //try to acquire a lock on the identity, to minimize risk of ConcurrencyException
                //do not sleep more than a certain amount of time (avoid some missing dispose).
                bool lockAcquired = TryAcquireLock(id);
                if (lockAcquired == false)
                {
                    //There is nothing to do if a lock failed to be aquired, but it worth to be signaled.
                    FailedLockCounter.Increment();
                }
            }

            ISnapshot snapshot = this.GetSnapshot<TAggregate>(bucketId, id, versionToLoad);
            IEventStream stream = this.OpenStream(bucketId, id, versionToLoad, snapshot);
            IAggregateEx aggregate = this.GetAggregate<TAggregate>(snapshot, stream);

            ApplyEventsToAggregate(versionToLoad, stream, aggregate);
            if (NeventStoreExGlobalConfiguration.MetricsEnabled)
            {
                sw.Stop();
                RepositoryLoadCounter.Increment(typeof(TAggregate).Name, sw.ElapsedTicks);
            }
            return aggregate as TAggregate;
        }

        /// <summary>
        /// Try to acquire the lock, if more than 5 second passed the lock is not acquired and the 
        /// method returns false.
        /// </summary>
        /// <param name="id"></param>
        /// <returns></returns>
        private bool TryAcquireLock(IIdentity id)
        {
            Int32 sleepCount = 0;
            Boolean lockAcquired = false;
            var stopCycle = 100 + NeventStoreExGlobalConfiguration.LockThreadSleepCount;
            while (!(lockAcquired = IdSerializerDictionary.TryAdd(id, _identity)) && sleepCount < stopCycle)
            {
                //some other thread is accessing that entity. Do a very fast spinning for a while, then sleep for a little.
                if (sleepCount < 100)
                    Thread.SpinWait(100 * sleepCount);
                else
                    Thread.Sleep(50);
                sleepCount++;
            }
            if (lockAcquired) _identitiesLocked.Add(id);
            return lockAcquired;
        }

        public virtual Int32 Save(IAggregateEx aggregate, Guid commitId, Action<IDictionary<string, object>> updateHeaders)
        {
            return Save(Bucket.Default, aggregate, commitId, updateHeaders);
        }

        /// <summary>
        /// 
        /// </summary>
        /// <param name="bucketId"></param>
        /// <param name="aggregate"></param>
        /// <param name="commitId"></param>
        /// <param name="updateHeaders"></param>
        /// <returns></returns>
        public Int32 Save(string bucketId, IAggregateEx aggregate, Guid commitId, Action<IDictionary<string, object>> updateHeaders)
        {
            Dictionary<string, object> headers = PrepareHeaders(aggregate, updateHeaders);
            Int32 uncommittedEvents = aggregate.GetUncommittedEvents().Count;
            while (true)
            {
  
                IEventStream stream = this.PrepareStream(bucketId, aggregate, headers);
                int commitEventCount = stream.CommittedEvents.Count;

                try
                {
                    stream.CommitChanges(commitId);
                    aggregate.ClearUncommittedEvents();
                    return uncommittedEvents;
                }
                catch (DuplicateCommitException dex)
                {
                    stream.ClearChanges();
                    _logger.Warn(String.Format("Duplicate commit exception bucket {0} - id {1} - commitid {2}. \n{3}", bucketId, aggregate.Id, commitId, dex));
                    return 0; //no events commited
                }
                catch (ConcurrencyException e)
                {
                    _logger.Warn(String.Format("Concurrency Exception bucket {0} - id {1} - commitid {2}. \n{3}",bucketId, aggregate.Id, commitId, e));
                    if (this.ThrowOnConflict(stream, commitEventCount))
                    {
                        //@@FIX -> aggiungere prima del lancio dell'eccezione
                        stream.ClearChanges();
                        throw new ConflictingCommandException(e.Message, e);
                    }
                    stream.ClearChanges();<commit_msg>Lowered log level for DuplicateCommitException.<commit_after>﻿using System;
using System.Collections.Generic;
using System.Linq;
using NEventStore.Domain;
using NEventStore.Domain.Persistence;
using NEventStore;
using NEventStore.Persistence;
using System.Collections.Concurrent;
using System.Threading;
using System.Diagnostics;
using Metrics;
using Jarvis.NEventStoreEx.Support;

namespace Jarvis.NEventStoreEx.CommonDomainEx.Persistence.EventStore
{
    public class AbstractRepository : IRepositoryEx
    {
        private const string AggregateTypeHeader = "AggregateType";

        private static readonly Counter RepositoryLoadCounter = Metric.Counter("RepositoryGetById", Unit.Custom("ticks"));
        private static readonly Counter FailedLockCounter = Metric.Counter("AbstractRepositoryFailedLocks", Unit.Calls);

        private readonly IDetectConflicts _conflictDetector;

        private readonly IStoreEvents _eventStore;

        private readonly IConstructAggregatesEx _factory;
        private readonly IIdentityConverter _identityConverter;

        private readonly IDictionary<string, IEventStream> _streams = new Dictionary<string, IEventStream>();

        /// <summary>
        /// This is used to keep a better track of locking, because each unique instance of repository
        /// will have its unique index, and each lock is marked with that index.
        /// </summary>
        private readonly Int64 _identity;

        /// <summary>
        /// This is the last identity assigned.
        /// </summary>
        private static Int64 _lastAssignedIdentity;

        /// <summary>
        /// This hashset contains the identities this instance of repository had obtained an exclusive lock to. 
        /// The real instance that holds all locked entities is <see cref="IdSerializerDictionary"/> property.
        /// It is necessary to remove all the lock after this instance is disposed.
        /// </summary>
        private readonly HashSet<IIdentity> _identitiesLocked = new HashSet<IIdentity>();

        /// <summary>
        /// This dictionary is used to hold a list of the aggregate ids that are acutally locked.
        /// It is used to avoid two threads to access at the very same moment the same entity
        /// thus avoiding unnecessary ConcurrencyException.
        /// </summary>
        private static readonly ConcurrentDictionary<IIdentity, Int64> IdSerializerDictionary;

        public NEventStore.Logging.ILog _logger;

        static AbstractRepository()
        {
            IdSerializerDictionary = new ConcurrentDictionary<IIdentity, Int64>();
            _lastAssignedIdentity = 0;
        }

        protected AbstractRepository(
            IStoreEvents eventStore, 
            IConstructAggregatesEx factory, 
            IDetectConflicts conflictDetector, 
            IIdentityConverter identityConverter,
            NEventStore.Logging.ILog logger)
        {
            this._eventStore = eventStore;
            this._factory = factory;
            this._conflictDetector = conflictDetector;
            this._identityConverter = identityConverter;
            this._identity = Interlocked.Increment(ref _lastAssignedIdentity);
            _logger = logger;
        }

        public void Dispose()
        {
            this.Dispose(true);
            GC.SuppressFinalize(this);
        }

        public virtual TAggregate GetById<TAggregate>(IIdentity id) where TAggregate : class, IAggregateEx
        {
            return this.GetById<TAggregate>(Bucket.Default, id);
        }

        public virtual TAggregate GetById<TAggregate>(IIdentity id, int versionToLoad) where TAggregate : class, IAggregateEx
        {
            return this.GetById<TAggregate>(Bucket.Default, id, versionToLoad);
        }

        public TAggregate GetById<TAggregate>(string bucketId, IIdentity id) where TAggregate : class, IAggregateEx
        {
            return this.GetById<TAggregate>(bucketId, id, int.MaxValue);
        }

        public TAggregate GetById<TAggregate>(string bucketId, IIdentity id, int versionToLoad) where TAggregate : class, IAggregateEx
        {
            Stopwatch sw = null;
            if (NeventStoreExGlobalConfiguration.MetricsEnabled) sw = Stopwatch.StartNew();

            if (NeventStoreExGlobalConfiguration.RepositoryLockOnAggregateId) 
            {
                //try to acquire a lock on the identity, to minimize risk of ConcurrencyException
                //do not sleep more than a certain amount of time (avoid some missing dispose).
                bool lockAcquired = TryAcquireLock(id);
                if (lockAcquired == false)
                {
                    //There is nothing to do if a lock failed to be aquired, but it worth to be signaled.
                    FailedLockCounter.Increment();
                }
            }

            ISnapshot snapshot = this.GetSnapshot<TAggregate>(bucketId, id, versionToLoad);
            IEventStream stream = this.OpenStream(bucketId, id, versionToLoad, snapshot);
            IAggregateEx aggregate = this.GetAggregate<TAggregate>(snapshot, stream);

            ApplyEventsToAggregate(versionToLoad, stream, aggregate);
            if (NeventStoreExGlobalConfiguration.MetricsEnabled)
            {
                sw.Stop();
                RepositoryLoadCounter.Increment(typeof(TAggregate).Name, sw.ElapsedTicks);
            }
            return aggregate as TAggregate;
        }

        /// <summary>
        /// Try to acquire the lock, if more than 5 second passed the lock is not acquired and the 
        /// method returns false.
        /// </summary>
        /// <param name="id"></param>
        /// <returns></returns>
        private bool TryAcquireLock(IIdentity id)
        {
            Int32 sleepCount = 0;
            Boolean lockAcquired = false;
            var stopCycle = 100 + NeventStoreExGlobalConfiguration.LockThreadSleepCount;
            while (!(lockAcquired = IdSerializerDictionary.TryAdd(id, _identity)) && sleepCount < stopCycle)
            {
                //some other thread is accessing that entity. Do a very fast spinning for a while, then sleep for a little.
                if (sleepCount < 100)
                    Thread.SpinWait(100 * sleepCount);
                else
                    Thread.Sleep(50);
                sleepCount++;
            }
            if (lockAcquired) _identitiesLocked.Add(id);
            return lockAcquired;
        }

        public virtual Int32 Save(IAggregateEx aggregate, Guid commitId, Action<IDictionary<string, object>> updateHeaders)
        {
            return Save(Bucket.Default, aggregate, commitId, updateHeaders);
        }

        /// <summary>
        /// 
        /// </summary>
        /// <param name="bucketId"></param>
        /// <param name="aggregate"></param>
        /// <param name="commitId"></param>
        /// <param name="updateHeaders"></param>
        /// <returns></returns>
        public Int32 Save(string bucketId, IAggregateEx aggregate, Guid commitId, Action<IDictionary<string, object>> updateHeaders)
        {
            Dictionary<string, object> headers = PrepareHeaders(aggregate, updateHeaders);
            Int32 uncommittedEvents = aggregate.GetUncommittedEvents().Count;
            while (true)
            {
  
                IEventStream stream = this.PrepareStream(bucketId, aggregate, headers);
                int commitEventCount = stream.CommittedEvents.Count;

                try
                {
                    stream.CommitChanges(commitId);
                    aggregate.ClearUncommittedEvents();
                    return uncommittedEvents;
                }
                catch (DuplicateCommitException dex)
                {
                    stream.ClearChanges();
                    _logger.Debug(String.Format("Duplicate commit exception bucket {0} - id {1} - commitid {2}. \n{3}", bucketId, aggregate.Id, commitId, dex));
                    return 0; //no events commited
                }
                catch (ConcurrencyException e)
                {
                    _logger.Warn(String.Format("Concurrency Exception bucket {0} - id {1} - commitid {2}. \n{3}",bucketId, aggregate.Id, commitId, e));
                    if (this.ThrowOnConflict(stream, commitEventCount))
                    {
                        //@@FIX -> aggiungere prima del lancio dell'eccezione
                        stream.ClearChanges();
                        throw new ConflictingCommandException(e.Message, e);
                    }
                    stream.ClearChanges();
{'fraction_non_alphanumeric': 0.08212728857890149, 'fraction_numerical': 0.004533565823888404, 'mean_word_length': 2.0790980672870436, 'pattern_counts': {'":': 0, '<': 101, '<?xml version=': 0, '>': 99, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '682583', 'n_tokens_mistral': 4271, 'n_tokens_neox': 4126, 'n_words': 1177}