﻿using EventCentric.Database;
using EventCentric.EventSourcing;
using EventCentric.Handling;
using EventCentric.Log;
using EventCentric.Messaging;
using EventCentric.Polling;
using EventCentric.Serialization;
using EventCentric.Utils;
using System;
using System.Collections.Generic;
using System.Data;
using System.Data.SqlClient;
using System.Linq;
using System.Runtime.Caching;

namespace EventCentric.Persistence
{
    public class AdoDotNetEventStore<T> : EventStoreBase, ISubscriptionRepository, IEventStore<T> where T : class, IEventSourced
    {
        private long eventCollectionVersion;
        private readonly ILogger log;
        private readonly ITextSerializer serializer;
        private readonly IUtcTimeProvider time;
        private readonly IGuidProvider guid;
        private readonly ObjectCache cache;

        private readonly Func<Guid, IEnumerable<IEvent>, T> aggregateFactory;
        private readonly Func<Guid, ISnapshot, T> originatorAggregateFactory;
        private readonly Action<SqlConnection, SqlTransaction, DateTime, IEvent> addToInboxFactory;
        private readonly Func<string, ITextSerializer, string, bool> consumerFilter;

        private readonly string connectionString;
        private readonly int saveTimeoutInSeconds = 600;
        private readonly SqlClientLite sql;
        private readonly object dbLock = new object();

        private readonly bool persistSnapshots;

        public AdoDotNetEventStore(string streamType, ITextSerializer serializer, string connectionString, IUtcTimeProvider time, IGuidProvider guid, ILogger log, bool persistIncomingPayloads, bool persistSnapshots, Func<string, ITextSerializer, string, bool> consumerFilter)
            : base(streamType)
        {
            Ensure.NotNull(serializer, nameof(serializer));
            Ensure.NotNullNeitherEmtpyNorWhiteSpace(connectionString, nameof(connectionString));
            Ensure.NotNull(time, nameof(time));
            Ensure.NotNull(guid, nameof(guid));
            Ensure.NotNull(log, nameof(log));

            this.serializer = serializer;
            this.connectionString = connectionString;
            this.time = time;
            this.guid = guid;
            this.log = log;
            this.consumerFilter = consumerFilter != null ? consumerFilter : EventStoreFuncs.DefaultFilter;
            this.cache = new MemoryCache(streamType);

            this.sql = new SqlClientLite(this.connectionString, timeoutInSeconds: 120);

            /// TODO: could be replaced with a compiled lambda to make it more performant.
            var fromMementoConstructor = typeof(T).GetConstructor(new[] { typeof(Guid), typeof(ISnapshot) });
            Ensure.CastIsValid(fromMementoConstructor, "Type T must have a constructor with the following signature: .ctor(Guid, IMemento)");
            this.originatorAggregateFactory = (id, memento) => (T)fromMementoConstructor.Invoke(new object[] { id, memento });

            var fromStreamConstructor = typeof(T).GetConstructor(new[] { typeof(Guid), typeof(IEnumerable<IEvent>) });
            Ensure.CastIsValid(fromStreamConstructor, "Type T must have a constructor with the following signature: .ctor(Guid, IEnumerable<IEvent>)");
            this.aggregateFactory = (id, streamOfEvents) => (T)fromStreamConstructor.Invoke(new object[] { id, streamOfEvents });

            this.eventCollectionVersion = this.GetLatestEventCollectionVersionFromDb();
            this.CurrentEventCollectionVersion = this.eventCollectionVersion;

            if (persistIncomingPayloads)
                this.addToInboxFactory = this.AddToInboxWithPayload;
            else
                this.addToInboxFactory = this.AddToInboxWithoutPayload;

            // adding app subscription if missing
            var appSubCount = this.sql.ExecuteReaderFirstOrDefault(this.tryFindAppSubscription, r => r.GetInt32(0),
                                new SqlParameter("@SubscriberStreamType", this.streamType),
                                new SqlParameter("@StreamType", this.streamType + Constants.AppEventStreamNameSufix));

            if (appSubCount == 0)
            {
                var now = DateTime.Now;
                this.sql.ExecuteNonQuery(this.createAppSubscription,
                    new SqlParameter("@SubscriberStreamType", this.streamType),
                    new SqlParameter("@StreamType", this.streamType + Constants.AppEventStreamNameSufix),
                    new SqlParameter("@CreationLocalTime", now),
                    new SqlParameter("@UpdateLocalTime", now));
            }

            this.persistSnapshots = persistSnapshots;
        }

        private long GetLatestEventCollectionVersionFromDb()
        {
            using (var connection = new SqlConnection(this.connectionString))
            {
                connection.Open();

                using (var command = connection.CreateCommand())
                {
                    command.CommandType = CommandType.Text;

                    command.CommandText = @"select top 1 EventCollectionVErsion from EventStore.Events 
                                            where StreamType = @StreamType
                                            order by EventCollectionVersion desc";

                    command.Parameters.Add(new SqlParameter("@StreamType", this.streamType));

                    using (var reader = command.ExecuteReader())
                    {
                        if (!reader.Read())
                            return 0;

                        return (long)reader["EventCollectionVersion"];
                    }
                }
            }
        }

        public T Find(Guid id)
        {
            // get memento from cache
            var cachedMemento = (Tuple<ISnapshot, DateTime?>)this.cache.Get(id.ToString());
            if (cachedMemento == null || !cachedMemento.Item2.HasValue)
            {
                if (!this.persistSnapshots)
                    return this.GetFromFullStreamOfEvents(id);

                // try return memento from SQL Server;
                var snapshotEntity = this.sql
                                         .ExecuteReaderFirstOrDefault(this.findSnapshotQuery,
                                         r => new SnapshotEntity
                                         {
                                             Payload = r.GetString("Payload"),
                                         },
                                         new SqlParameter("@StreamType", this.streamType),
                                         new SqlParameter("@StreamId", id));

                if (snapshotEntity != null)
                    cachedMemento = new Tuple<ISnapshot, DateTime?>(this.serializer.Deserialize<ISnapshot>(snapshotEntity.Payload), null);
                else
                    return this.GetFromFullStreamOfEvents(id);
            }

            try
            {
                return this.originatorAggregateFactory.Invoke(id, cachedMemento.Item1);
            }
            catch (Exception ex)
            {
                this.log.Error(ex, $"An error ocurred while hydrating aggregate from snapshot. Now we will try to recover state from full stream of events for stream id {id.ToString()}");
                return this.GetFromFullStreamOfEvents(id);
            }
        }

        private T GetFromFullStreamOfEvents(Guid id)
        {
            var streamOfEvents =
                this.sql.ExecuteReader(600, this.getEventsQuery, r =>
                this.serializer.Deserialize<IEvent>(r.GetString("Payload")),
                new SqlParameter("@StreamType", this.streamType),
                new SqlParameter("@StreamId", id))
                .AsCachedAnyEnumerable();

            if (streamOfEvents.Any())
                return this.aggregateFactory.Invoke(id, streamOfEvents);

            return null;
        }

        public T Get(Guid id)
        {
            var aggregate = this.Find(id);
            if (aggregate == null)
            {
                var ex = new StreamNotFoundException(id, streamType);
                this.log.Error(ex, string.Format("Stream not found exception for stream {0} with id of {1}", streamType, id));
                throw ex;
            }

            return aggregate;
        }

        public void Save(T eventSourced, IEvent incomingEvent)
        {
            string key = null;
            try
            {
                using (var connection = new SqlConnection(this.connectionString))
                {
                    connection.Open();

                    using (var command = connection.CreateCommand())
                    {
                        command.CommandType = CommandType.Text;
                        command.CommandTimeout = saveTimeoutInSeconds;
                        command.CommandText = this.isDuplicateQuery;
                        command.Parameters.Add(new SqlParameter("@EventId", SqlDbType.UniqueIdentifier) { Value = incomingEvent.EventId });

                        using (var reader = command.ExecuteReader())
                        {
                            if (reader.Read())
                                return;
                        }
                    }

                    using (var transaction = connection.BeginTransaction(IsolationLevel.ReadUncommitted))
                    {
                        try
                        {
                            var now = this.time.Now;
                            var localNow = this.time.Now.ToLocalTime();

                            if (eventSourced == null)
                            {
                                this.addToInboxFactory.Invoke(connection, transaction, localNow, incomingEvent);
                                transaction.Commit();
                                return;
                            }

                            if (eventSourced.Id == default(Guid))
                                throw new ArgumentOutOfRangeException("StreamId", $"The eventsourced of type {typeof(T).FullName} has a default GUID value for its stream id, which is not valid");

                            // get current version
                            var pendingEvents = eventSourced.PendingEvents;
                            if (pendingEvents.Count > 0)
                            {
                                long currentVersion = 0;
                                using (var command = connection.CreateCommand())
                                {
                                    command.CommandType = CommandType.Text;
                                    command.Transaction = transaction;
                                    command.CommandTimeout = this.saveTimeoutInSeconds;
                                    command.CommandText = this.getStreamVersion;
                                    command.Parameters.Add(new SqlParameter("@StreamType", SqlDbType.NVarChar, 40) { Value = this.streamType });
                                    command.Parameters.Add(new SqlParameter("@StreamId", SqlDbType.UniqueIdentifier) { Value = eventSourced.Id });

                                    using (var reader = command.ExecuteReader())
                                    {
                                        if (reader.Read())
                                        {
                                            currentVersion = reader.GetInt64("Version");
                                        }
                                    }
                                }


                                if (currentVersion + 1 != pendingEvents.First().Version)
                                    throw new EventStoreConcurrencyException();

                                // Cache Memento And Publish Stream
                                var snapshot = ((ISnapshotOriginator)eventSourced).SaveToSnapshot();
                                var serializedMemento = this.serializer.Serialize(snapshot);

                                // Cache in memory
                                key = eventSourced.Id.ToString();
                                this.cache.Set(
                                    key: key,
                                    value: new Tuple<ISnapshot, DateTime?>(snapshot, now),
                                    policy: new CacheItemPolicy { AbsoluteExpiration = this.time.OffSetNow.AddMinutes(30) });

                                // Insert or Update snapshot
                                if (this.persistSnapshots)
                                {
                                    using (var command = connection.CreateCommand())
                                    {
                                        command.Transaction = transaction;
                                        command.CommandTimeout = this.saveTimeoutInSeconds;
                                        command.CommandType = CommandType.Text;
                                        command.CommandText = this.insertOrUpdateSnapshot;
                                        command.Parameters.Add(new SqlParameter("@StreamType", SqlDbType.NVarChar, 40) { Value = this.streamType });
                                        command.Parameters.Add(new SqlParameter("@StreamId", SqlDbType.UniqueIdentifier) { Value = eventSourced.Id });
                                        command.Parameters.Add(new SqlParameter("@Version", SqlDbType.BigInt) { Value = eventSourced.Version });
                                        command.Parameters.Add(new SqlParameter("@Payload", SqlDbType.NVarChar) { Value = serializedMemento });
                                        command.Parameters.Add(new SqlParameter("@CreationLocalTime", SqlDbType.DateTime) { Value = localNow });
                                        command.Parameters.Add(new SqlParameter("@UpdateLocalTime", SqlDbType.DateTime) { Value = localNow });

                                        command.ExecuteNonQuery();
                                    }
                                }
                            }

                            // Log the incoming message in the inbox
                            this.addToInboxFactory.Invoke(connection, transaction, localNow, incomingEvent);

                            var eventEntities = new EventEntity[pendingEvents.Count];
                            for (int i = 0; i < pendingEvents.Count; i++)
                            {
                                var @event = pendingEvents[i];
                                var e = (Message)@event;
                                e.TransactionId = incomingEvent.TransactionId;
                                e.EventId = this.guid.NewGuid();
                                e.StreamType = this.streamType;

                                eventEntities[i] = new EventEntity
                                {
                                    StreamType = this.streamType,
                                    StreamId = @event.StreamId,
                                    Version = @event.Version,
                                    EventId = @event.EventId,
                                    TransactionId = @event.TransactionId,
                                    EventType = @event.GetType().Name,
                                    CorrelationId = incomingEvent.EventId,
                                    LocalTime = localNow,
                                    UtcTime = now
                                };
                            }

                            lock (this.dbLock)
                            {
                                var eventCollectionBeforeCrash = this.eventCollectionVersion;
                                try
                                {
                                    for (int i = 0; i < pendingEvents.Count; i++)
                                    {
                                        this.eventCollectionVersion += 1;
                                        var @event = pendingEvents[i];
                                        ((Message)@event).EventCollectionVersion = this.eventCollectionVersion;
                                        var entity = eventEntities[i];
                                        entity.EventCollectionVersion = this.eventCollectionVersion;
                                        entity.Payload = this.serializer.Serialize(@event);

                                        // Insert each event
                                        using (var command = connection.CreateCommand())
                                        {
                                            command.Transaction = transaction;
                                            command.CommandType = CommandType.Text;
                                            command.CommandTimeout = this.saveTimeoutInSeconds;
                                            command.CommandText = this.appendEventCommand;
                                            command.Parameters.Add(new SqlParameter("@StreamType", SqlDbType.NVarChar, 40) { Value = entity.StreamType });
                                            command.Parameters.Add(new SqlParameter("@StreamId", SqlDbType.UniqueIdentifier) { Value = entity.StreamId });
                                            command.Parameters.Add(new SqlParameter("@Version", SqlDbType.BigInt) { Value = entity.Version });
                                            command.Parameters.Add(new SqlParameter("@TransactionId", SqlDbType.UniqueIdentifier) { Value = entity.TransactionId });
                                            command.Parameters.Add(new SqlParameter("@EventId", SqlDbType.UniqueIdentifier) { Value = entity.EventId });
                                            command.Parameters.Add(new SqlParameter("@EventType", SqlDbType.NVarChar, 255) { Value = entity.EventType });
                                            command.Parameters.Add(new SqlParameter("@CorrelationId", SqlDbType.UniqueIdentifier) { Value = entity.CorrelationId });
                                            command.Parameters.Add(new SqlParameter("@EventCollectionVersion", SqlDbType.BigInt) { Value = entity.EventCollectionVersion });
                                            command.Parameters.Add(new SqlParameter("@LocalTime", SqlDbType.DateTime) { Value = entity.LocalTime });
                                            command.Parameters.Add(new SqlParameter("@UtcTime", SqlDbType.DateTime) { Value = entity.UtcTime });
                                            command.Parameters.Add(new SqlParameter("@Payload", SqlDbType.NVarChar) { Value = entity.Payload });

                                            command.ExecuteNonQuery();
                                        }
                                    }
                                    transaction.Commit();

                                    this.CurrentEventCollectionVersion = this.eventCollectionVersion;
                                }
                                catch (Exception ex)
                                {
                                    this.eventCollectionVersion = eventCollectionBeforeCrash;
                                    throw ex;
                                }
                            }
                        }
                        catch (Exception)
                        {
                            try
                            {
                                transaction.Rollback();
                            }
                            catch
                            {
                            }
                            throw;
                        }
                    }
                }
            }
            catch (Exception ex)
            {
                this.log.Error(ex, $"An error ocurred while storing events while processing incoming event of type '{incomingEvent.GetType().Name}'");

                // Mark cache as stale
                if (key != null)
                {
                    var item = (Tuple<ISnapshot, DateTime?>)this.cache.Get(key);
                    if (item != null && item.Item2.HasValue)
                    {
                        item = new Tuple<ISnapshot, DateTime?>(item.Item1, null);
                        this.cache.Set(
                            key,
                            item,
                            new CacheItemPolicy { AbsoluteExpiration = this.time.OffSetNow.AddMinutes(30) });
                    }
                }

                throw;
            }
        }

        private void AddToInboxWithPayload(SqlConnection connection, SqlTransaction transaction, DateTime localNow, IEvent incomingEvent)
        {
            using (var command = connection.CreateCommand())
            {
                command.Transaction = transaction;
                command.CommandTimeout = this.saveTimeoutInSeconds;
                command.CommandType = CommandType.Text;
                command.CommandText = this.addToInboxWithPayloadCommand;
                command.Parameters.Add(new SqlParameter("@InboxStreamType", SqlDbType.NVarChar, 40) { Value = this.streamType });
                command.Parameters.Add(new SqlParameter("@EventId", SqlDbType.UniqueIdentifier) { Value = incomingEvent.EventId });
                command.Parameters.Add(new SqlParameter("@TransactionId", SqlDbType.UniqueIdentifier) { Value = incomingEvent.TransactionId });
                command.Parameters.Add(new SqlParameter("@StreamType", SqlDbType.NVarChar, 40) { Value = incomingEvent.StreamType });
                command.Parameters.Add(new SqlParameter("@StreamId", SqlDbType.UniqueIdentifier) { Value = incomingEvent.StreamId });
                command.Parameters.Add(new SqlParameter("@Version", SqlDbType.BigInt) { Value = incomingEvent.Version });
                command.Parameters.Add(new SqlParameter("@EventType", SqlDbType.NVarChar, 255) { Value = incomingEvent.GetType().Name });
                command.Parameters.Add(new SqlParameter("@EventCollectionVersion", SqlDbType.BigInt) { Value = incomingEvent.EventCollectionVersion });
                command.Parameters.Add(new SqlParameter("@CreationLocalTime", SqlDbType.DateTime) { Value = localNow });
                command.Parameters.Add(new SqlParameter("@Payload", SqlDbType.NVarChar) { Value = this.serializer.Serialize(incomingEvent) });

                command.ExecuteNonQuery();
            }
        }

        private void AddToInboxWithoutPayload(SqlConnection connection, SqlTransaction transaction, DateTime localNow, IEvent incomingEvent)
        {
            using (var command = connection.CreateCommand())
            {
                command.Transaction = transaction;
                command.CommandTimeout = this.saveTimeoutInSeconds;
                command.CommandType = CommandType.Text;
                command.CommandText = this.addToInboxWithoutPayloadCommand;
                command.Parameters.Add(new SqlParameter("@InboxStreamType", SqlDbType.NVarChar, 40) { Value = this.streamType });
                command.Parameters.Add(new SqlParameter("@EventId", SqlDbType.UniqueIdentifier) { Value = incomingEvent.EventId });
                command.Parameters.Add(new SqlParameter("@CreationLocalTime", SqlDbType.DateTime) { Value = localNow });
                command.Parameters.Add(new SqlParameter("@TransactionId", SqlDbType.UniqueIdentifier) { Value = incomingEvent.TransactionId });

                command.ExecuteNonQuery();
            }
        }

        public bool IsDuplicate(Guid eventId, out Guid transactionId)
        {
            transactionId = Guid.Empty;
            using (var connection = new SqlConnection(this.connectionString))
            {
                connection.Open();

                using (var command = connection.CreateCommand())
                {
                    command.CommandType = CommandType.Text;
                    command.CommandText = this.isDuplicateQuery;
                    command.Parameters.Add(new SqlParameter("@EventId", SqlDbType.UniqueIdentifier) { Value = eventId });

                    using (var reader = command.ExecuteReader())
                    {
                        if (reader.Read())
                        {
                            transactionId = reader.GetGuid("TransactionId");
                            return true;
                        }
                        else
                        {
                            return false;
                        }
                    }
                }
            }
        }

        public void DeleteSnapshot(Guid streamId)
        {
            cache.Remove(streamId.ToString());
            this.sql.ExecuteNonQuery(this.removeSnapshotCommand,
                new SqlParameter("@StreamId", SqlDbType.UniqueIdentifier) { Value = streamId },
                new SqlParameter(@"StreamType", SqlDbType.NVarChar, 40) { Value = this.streamType });
        }

        public long CurrentEventCollectionVersion { get; private set; }

        public string StreamName => this.streamType;

        /// <summary>
        /// FindEvents
        /// </summary>
        /// <returns>Events if found, otherwise return empty list.</returns>
        public SerializedEvent[] FindEventsForConsumer(long from, long to, int quantity, string consumer)
        {
            return this.sql.ExecuteReader(this.findEventsQuery,
                r => new SerializedEvent(r.GetInt64("EventCollectionVersion"), r.GetString("Payload")),
                new SqlParameter("@Quantity", SqlDbType.Int) { Value = quantity },
                new SqlParameter("@StreamType", SqlDbType.NVarChar, 40) { Value = this.streamType },
                new SqlParameter("@LastReceivedVersion", SqlDbType.BigInt) { Value = from },
                new SqlParameter("@MaxVersion", SqlDbType.BigInt) { Value = to })
                .Select(e =>
                            EventStoreFuncs.ApplyConsumerFilter(
                                new SerializedEvent(e.EventCollectionVersion, e.Payload),
                                consumer,
                                this.serializer,
                                this.consumerFilter))
                .ToArray();
        }
        public SerializedEvent[] FindEventsForConsumer(long from, long to, Guid streamId, int quantity, string consumer)
        {
            var events = this.sql.ExecuteReader(this.findEventsWithStreamIdQuery,
                r => new SerializedEvent(r.GetInt64("EventCollectionVersion"), r.GetString("Payload")),
                new SqlParameter("@Quantity", SqlDbType.Int) { Value = quantity },
                new SqlParameter("@StreamType", SqlDbType.NVarChar, 40) { Value = this.streamType },
                new SqlParameter("@LastReceivedVersion", SqlDbType.BigInt) { Value = from },
                new SqlParameter("@MaxVersion", SqlDbType.BigInt) { Value = to },
                new SqlParameter("@StreamId", SqlDbType.UniqueIdentifier) { Value = streamId })
                .Select(e =>
                            EventStoreFuncs.ApplyConsumerFilter(
                                new SerializedEvent(e.EventCollectionVersion, e.Payload),
                                consumer,
                                this.serializer,
                                this.consumerFilter))
                .ToArray();

            return events.Length > 0
                ? events
                : new SerializedEvent[] { new SerializedEvent(to, this.serializer.Serialize(new CloakedEvent(to, this.streamType))) };
        }

        public SubscriptionBuffer[] GetSubscriptions()
        {
            return this.sql.ExecuteReader(this.getSubscriptionsQuery,
                r => new SubscriptionBuffer(
                        r.GetString("StreamType"),
                        r.GetString("Url"),
                        r.SafeGetString("Token"),
                        r.GetInt64("ProcessorBufferVersion") - 1, // We substract one version in order to set the current version bellow the last one, in case that first event was not yet processed.
                        false),
                        new SqlParameter("@SubscriberStreamType", this.streamType))
                        .ToArray();
        }

        public void FlagSubscriptionAsPoisoned(IEvent poisonedEvent, PoisonMessageException exception)
        {
            string exceptionMessage;

            try
            {
                exceptionMessage = this.serializer.Serialize(exception);
            }
            catch (Exception)
            {
                exceptionMessage = string.Format("Exception type: {0}. Exception message: {1}. Inner exception: {2}", exception.GetType().Name, exception.Message, exception.InnerException.Message != null ? exception.InnerException.Message : "null");
            }

            string deadLetterPayload;
            try
            {
                deadLetterPayload = this.serializer.Serialize(poisonedEvent);
            }
            catch (Exception)
            {
                deadLetterPayload = string.Format("EventType: {0}", poisonedEvent.GetType().Name);
            }

            this.sql.ExecuteNonQuery(this.markAsPoisoned,
                new SqlParameter("@UpdateLocalTime", this.time.Now.ToLocalTime()),
                new SqlParameter("@PoisonEventCollectionVersion", poisonedEvent.EventCollectionVersion),
                new SqlParameter("@ExceptionMessage", SqlDbType.NVarChar) { Value = exceptionMessage },
                new SqlParameter("@DeadLetterPayload", SqlDbType.NVarChar) { Value = deadLetterPayload },
                new SqlParameter("@SubscriberStreamType", this.streamType),
                new SqlParameter("@StreamType", poisonedEvent.StreamType));

        }

        public bool TryAddNewSubscriptionOnTheFly(string streamType, string url, string token)
        {
            var subCount = this.sql.ExecuteReaderFirstOrDefault(this.subscriptionAlreadyExistsScript,
                                    r => r.GetInt32("SubCount"),
                                    new SqlParameter("@SubscriberStreamType", this.streamType),
                                    new SqlParameter("@StreamType", streamType));

            if (subCount > 0)
                return false;

            var now = this.time.Now.ToLocalTime();
            this.sql.ExecuteNonQuery(this.addSubOnTheFlyScript,
                new SqlParameter("@SubscriberStreamType", this.StreamName),
                new SqlParameter("@StreamType", streamType),
                new SqlParameter("@Url", url),
                new SqlParameter("@Token", token),
                new SqlParameter("@CreationLocalTime", now),
                new SqlParameter("@UpdateLocalTime", now));
            return true;
        }

        public void PersistSubscriptionVersion(string subscription, long consumerVersion, long producerVersion)
        {
            this.sql.ExecuteNonQuery(this.updateSubscriptionScript,
                new SqlParameter("@SubscriberStreamType", SqlDbType.NVarChar, 40) { Value = this.streamType },
                new SqlParameter("@StreamType", SqlDbType.NVarChar, 40) { Value = subscription },
                new SqlParameter("@ConsumerVersion", SqlDbType.BigInt) { Value = consumerVersion },
                new SqlParameter("@ProducerVersion", SqlDbType.BigInt) { Value = producerVersion },
                new SqlParameter("@ConsistencyPercentage", SqlDbType.NVarChar, 10) { Value = base.GetConsistencyPercentage(consumerVersion, producerVersion) },
                new SqlParameter("@UpdateLocalTime", SqlDbType.DateTime) { Value = this.time.Now.ToLocalTime() });
        }


        #region Scripts
        private readonly string findSnapshotQuery =
        @"select Payload from EventStore.Snapshots
where StreamType = @StreamType 
and StreamId = @StreamId";

        private readonly string getEventsQuery =
        @"select Payload from EventStore.Events 
where StreamType = @StreamType
and StreamId = @StreamId
order by [Version]";

        private readonly string isDuplicateQuery =
        "select TransactionId from eventstore.inbox where EventId = @EventId";

        private readonly string tryFindAppSubscription =
        @"select count(*) from EventStore.Subscriptions
where SubscriberStreamType = @SubscriberStreamType
and StreamType = @StreamType";

        private readonly string removeSnapshotCommand =
        @"delete from EventStore.Snapshots
where StreamType = @StreamType 
and StreamId = @StreamId";

        private readonly string getStreamVersion =
        @"select top 1 [Version] from EventStore.Events 
where StreamType = @StreamType
and StreamId = @StreamId
order by [Version] desc";

        private readonly string addToInboxWithPayloadCommand =
        @"INSERT INTO [EventStore].[Inbox]
    ([InboxStreamType]
    ,[EventId]
    ,[TransactionId]
    ,[StreamType]
    ,[StreamId]
    ,[Version]
    ,[EventType]
    ,[EventCollectionVersion]
    ,[CreationLocalTime]
    ,[Payload])
VALUES
    (@InboxStreamType, 
    @EventId, 
    @TransactionId, 
    @StreamType, 
    @StreamId, 
    @Version, 
    @EventType, 
    @EventCollectionVersion, 
    @CreationLocalTime,
    @Payload)";

        private readonly string addToInboxWithoutPayloadCommand =
        @"INSERT INTO [EventStore].[Inbox]
    ([InboxStreamType]
    ,[EventId]
    ,[CreationLocalTime]
    ,[TransactionId])
VALUES
    (@InboxStreamType, 
    @EventId,
    @CreationLocalTime,
    @TransactionId)";

        private readonly string insertOrUpdateSnapshot =
        @"if exists (select StreamType from EventStore.Snapshots where StreamId = @StreamId AND StreamType = @StreamType)
BEGIN
UPDATE [EventStore].[Snapshots]
   SET [Version] = @Version
      ,[Payload] = @Payload
      ,[UpdateLocalTime] = @UpdateLocalTime
 WHERE StreamId = @StreamId AND StreamType = @StreamType
END
ELSE
BEGIN
INSERT INTO [EventStore].[Snapshots]
           ([StreamType]
           ,[StreamId]
           ,[Version]
           ,[Payload]
           ,[CreationLocalTime]
           ,[UpdateLocalTime])
     VALUES
           (@StreamType
           ,@StreamId
           ,@Version
           ,@Payload
           ,@CreationLocalTime
           ,@UpdateLocalTime)
END";

        private readonly string createAppSubscription =
        @"INSERT INTO [EventStore].[Subscriptions]
    ([SubscriberStreamType]
    ,[StreamType]
    ,[Url]
    ,[Token]
    ,[ProcessorBufferVersion]
    ,[IsPoisoned]
    ,[WasCanceled]
    ,[PoisonEventCollectionVersion]
    ,[DeadLetterPayload]
    ,[ExceptionMessage]
    ,[CreationLocalTime]
    ,[UpdateLocalTime])
VALUES
    (@SubscriberStreamType
    ,@StreamType
    ,'none'
    ,'#token'
    ,0
    ,0
    ,1
    ,null
    ,null
    ,null
    ,@CreationLocalTime
    ,@UpdateLocalTime)";

        private readonly string appendEventCommand =
        @"INSERT INTO [EventStore].[Events]
    ([StreamType]
    ,[StreamId]
    ,[Version]
    ,[TransactionId]
    ,[EventId]
    ,[EventType]
    ,[CorrelationId]
    ,[EventCollectionVersion]
    ,[LocalTime]
    ,[UtcTime]
    ,[Payload])
VALUES
    (@StreamType
    ,@StreamId
    ,@Version
    ,@TransactionId
    ,@EventId
    ,@EventType
    ,@CorrelationId
    ,@EventCollectionVersion
    ,@LocalTime
    ,@UtcTime
    ,@Payload)";

        private readonly string findEventsQuery =
        @"select top (@Quantity)
EventCollectionVersion,
Payload
from EventStore.Events
where StreamType = @StreamType
and EventCollectionVersion > @LastReceivedVersion
and EventCollectionVersion <= @MaxVersion
order by EventCollectionVersion";

        private readonly string findEventsWithStreamIdQuery =
            @"select top (@Quantity)
EventCollectionVersion,
Payload
from EventStore.Events
where StreamType = @StreamType
and EventCollectionVersion > @LastReceivedVersion
and EventCollectionVersion <= @MaxVersion
and StreamId = @StreamId
order by EventCollectionVersion";

        private readonly string getSubscriptionsQuery =
@"SELECT [StreamType]
      ,[Url]
      ,[Token]
      ,[ProcessorBufferVersion]
  FROM [EventStore].[Subscriptions]
  WHERE SubscriberStreamType = @SubscriberStreamType
  AND IsPoisoned = 0
  AND WasCanceled = 0";

        private readonly string updateSubscriptionScript =
@"UPDATE [EventStore].[Subscriptions]
   SET [ProcessorBufferVersion] = @ConsumerVersion
      ,[ProducerVersion] = @ProducerVersion
      ,[ConsistencyPercentage] = @ConsistencyPercentage
      ,[UpdateLocalTime] = @UpdateLocalTime
 WHERE SubscriberStreamType = @SubscriberStreamType
 AND StreamType = @StreamType";

        private readonly string markAsPoisoned =
@"UPDATE EventStore.Subscriptions SET 
IsPoisoned = 1,
UpdateLocalTime = @UpdateLocalTime,
PoisonEventCollectionVersion = @PoisonEventCollectionVersion,
ExceptionMessage = @ExceptionMessage,
DeadLetterPayload = @DeadLetterPayload
WHERE SubscriberStreamType = @SubscriberStreamType
AND StreamType = @StreamType";

        private readonly string subscriptionAlreadyExistsScript =
@"SELECT count (*) as SubCount
  FROM [EventStore].[Subscriptions]
  WHERE SubscriberStreamType = @SubscriberStreamType
  AND StreamType = @StreamType";

        private readonly string addSubOnTheFlyScript =
@"INSERT INTO [EventStore].[Subscriptions]
           ([SubscriberStreamType]
           ,[StreamType]
           ,[Url]
           ,[Token]
           ,[ProcessorBufferVersion]
           ,[IsPoisoned]
           ,[WasCanceled]
           ,[CreationLocalTime]
           ,[UpdateLocalTime])
     VALUES
           (@SubscriberStreamType -- nvarchar(40)
           ,@StreamType -- nvarchar(40)
           ,@Url
           ,@Token
           ,0
           ,0
           ,0
           ,@CreationLocalTime
           ,@UpdateLocalTime)";
        #endregion
    }
}
